Shift resource type filtering into watcher, instead of log query filter

Author: nick-jones

internal/auditlog/tail.go

@@ -16,13 +16,7 @@ import (
 	"google.golang.org/grpc/status"
 )
 
-func Tail(
-	ctx context.Context,
-	projectID string,
-	clusterName string,
-	resourceKinds []string,
-	cb func(*audit.AuditLog) error,
-) error {
+func Tail(ctx context.Context, projectID string, clusterName string, cb func(*audit.AuditLog) error) error {
 	client, err := logging.NewClient(ctx)
 	if err != nil {
 		return fmt.Errorf("failed to create client: %w", err)
@@ -43,7 +37,7 @@ func Tail(
 			return fmt.Errorf("limit wait failed: %w", err)
 		}
 
-		if err = tailLogs(ctx, client, projectID, clusterName, resourceKinds, cb); err != nil {
+		if err = tailLogs(ctx, client, projectID, clusterName, cb); err != nil {
 			if _, ok := status.FromError(err); ok {
 				slog.Warn("gRPC request terminated, restarting", slog.Any("error", err))
 				continue
@@ -53,14 +47,7 @@ func Tail(
 	}
 }
 
-func tailLogs(
-	ctx context.Context,
-	client *logging.Client,
-	projectID string,
-	clusterName string,
-	resourceKinds []string,
-	cb func(*audit.AuditLog) error,
-) error {
+func tailLogs(ctx context.Context, client *logging.Client, projectID, clusterName string, cb func(*audit.AuditLog) error) error {
 	stream, err := client.TailLogEntries(ctx)
 	if err != nil {
 		return fmt.Errorf("request to tail log entries failed: %w", err)
@@ -77,7 +64,7 @@ func tailLogs(
 				fmt.Sprintf(`log_name="projects/%s/logs/cloudaudit.googleapis.com%%2Factivity"`, projectID),
 				fmt.Sprintf(`resource.labels.cluster_name="%s"`, clusterName),
 				`protoPayload."@type"="type.googleapis.com/google.cloud.audit.AuditLog"`,
-				fmt.Sprintf(`protoPayload.methodName=~"io\.fluxcd\.toolkit\..*\.(%s)\.patch"`, strings.Join(resourceKinds, "|")),
+				`protoPayload.methodName=~"io\.fluxcd\.toolkit\..*\.(patch|create)"`,
 				`-protoPayload.authenticationInfo.principalEmail=~"system:.*"`,
 			},
 			" AND ",

internal/watch/watcher.go

@@ -60,39 +60,47 @@ type notifier interface {
 }
 
 func (w *Watcher) Watch(ctx context.Context) error {
+	resourceTypes, err := w.resolveFluxResourceTypes(ctx)
+	if err != nil {
+		return fmt.Errorf("could not resolve flux resource types: %w", err)
+	}
+
+	if err = w.init(ctx, resourceTypes); err != nil {
+		return fmt.Errorf("failed to initialize: %w", err)
+	}
+
+	return w.watch(ctx, resourceTypes)
+}
+
+func (w *Watcher) resolveFluxResourceTypes(ctx context.Context) ([]k8s.ResourceType, error) {
 	crds, err := w.k8sClient.GetCustomResourceDefinitions(ctx, metav1.ListOptions{
 		LabelSelector: "app.kubernetes.io/part-of=flux",
 	})
 	if err != nil {
-		return err
+		return nil, fmt.Errorf("failed to fetch crds: %w", err)
 	}
 
-	uniqueResourceGroups := make(map[string]k8s.ResourceType)
+	uniqueResourceTypes := make(map[string]k8s.ResourceType)
 	for _, crd := range crds.Items {
 		for _, version := range crd.Spec.Versions {
 			if _, exists := version.Schema.OpenAPIV3Schema.Properties["spec"].Properties["suspend"]; !exists {
 				continue
 			}
 			key := fmt.Sprintf("%s:%s", crd.Spec.Group, crd.Spec.Names.Plural)
-			uniqueResourceGroups[key] = k8s.ResourceType{
+			uniqueResourceTypes[key] = k8s.ResourceType{
 				Group:   crd.Spec.Group,
 				Version: version.Name,
 				Kind:    crd.Spec.Names.Plural,
 			}
 		}
 	}
-	resourceGroups := maps.Values(uniqueResourceGroups)
-
-	if err = w.init(ctx, resourceGroups); err != nil {
-		return fmt.Errorf("failed to initialize: %w", err)
-	}
-	return w.watch(ctx, resourceGroups)
+	return maps.Values(uniqueResourceTypes), nil
 }
 
-func (w *Watcher) init(ctx context.Context, groups []k8s.ResourceType) error {
+func (w *Watcher) init(ctx context.Context, types []k8s.ResourceType) error {
 	slog.Info("initializing")
-	for _, group := range groups {
-		res, err := w.k8sClient.GetRawResources(ctx, group)
+	for _, t := range types {
+		res, err := w.k8sClient.GetRawResources(ctx, t)
 		if err != nil {
 			return err
 		}
@@ -102,7 +110,7 @@ func (w *Watcher) init(ctx context.Context, groups []k8s.ResourceType) error {
 		}
 		for _, resource := range resourceList.Items {
 			resourceRef := k8s.ResourceReference{
-				Type:      group,
+				Type:      t,
 				Namespace: resource.Metadata.Namespace,
 				Name:      resource.Metadata.Name,
 			}
@@ -114,14 +122,10 @@ func (w *Watcher) init(ctx context.Context, groups []k8s.ResourceType) error {
 	return nil
 }
 
-func (w *Watcher) watch(ctx context.Context, groups []k8s.ResourceType) error {
+func (w *Watcher) watch(ctx context.Context, types []k8s.ResourceType) error {
 	slog.Info("watching for resource modifications")
-	resourceKinds := make([]string, 0, len(groups))
-	for _, group := range groups {
-		resourceKinds = append(resourceKinds, group.Kind)
-	}
 
-	return auditlog.Tail(ctx, w.googleCloudProjectID, w.gkeClusterName, resourceKinds, func(logEntry *audit.AuditLog) error {
+	return auditlog.Tail(ctx, w.googleCloudProjectID, w.gkeClusterName, func(logEntry *audit.AuditLog) error {
 		if code := logEntry.GetStatus().GetCode(); code != 0 {
 			slog.Warn("operation appeared to fail", slog.Int("code", int(code)))
 			return nil
@@ -135,6 +139,11 @@ func (w *Watcher) watch(ctx context.Context, groups []k8s.ResourceType) error {
 			return err
 		}
 
+		if !isWatchedResource(resourceRef, types) {
+			slog.Info("ignoring non-watched resource", slog.String("kind", resourceRef.Type.Kind))
+			return nil
+		}
+
 		res, err := w.k8sClient.GetRawResource(ctx, resourceRef)
 		if err != nil {
 			return fmt.Errorf("failed to get raw resource: %w", err)
@@ -211,3 +220,12 @@ func (w *Watcher) processResource(
 		GoogleCloudProjectID: w.googleCloudProjectID,
 	})
 }
+
+func isWatchedResource(ref k8s.ResourceReference, types []k8s.ResourceType) bool {
+	for _, t := range types {
+		if t.Group == ref.Type.Group && t.Kind == ref.Type.Kind {
+			return true
+		}
+	}
+	return false
+}