feat: add syncset controller, feat: syncset readiness

apis/config/v1alpha1/config_types.go

@@ -18,6 +18,7 @@ package v1alpha1
 import (
 	"github.com/open-policy-agent/gatekeeper/v3/pkg/wildcard"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 )
 
 // ConfigSpec defines the desired state of Config.
@@ -62,6 +63,14 @@ type SyncOnlyEntry struct {
 	Kind    string `json:"kind,omitempty"`
 }
 
+func (e *SyncOnlyEntry) ToGroupVersionKind() schema.GroupVersionKind {
+	return schema.GroupVersionKind{
+		Group:   e.Group,
+		Version: e.Version,
+		Kind:    e.Kind,
+	}
+}
+
 type MatchEntry struct {
 	Processes          []string            `json:"processes,omitempty"`
 	ExcludedNamespaces []wildcard.Wildcard `json:"excludedNamespaces,omitempty"`

apis/syncset/v1alpha1/syncset_types.go

@@ -2,6 +2,7 @@ package v1alpha1
 
 import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 )
 
 type SyncSetSpec struct {
@@ -14,6 +15,14 @@ type GVKEntry struct {
 	Kind    string `json:"kind,omitempty"`
 }
 
+func (e *GVKEntry) ToGroupVersionKind() schema.GroupVersionKind {
+	return schema.GroupVersionKind{
+		Group:   e.Group,
+		Version: e.Version,
+		Kind:    e.Kind,
+	}
+}
+
 // +kubebuilder:resource:scope=Cluster
 // +kubebuilder:object:root=true
 

config/crd/kustomization.yaml

@@ -3,7 +3,7 @@
 # It should be run by config/default
 resources:
 - bases/config.gatekeeper.sh_configs.yaml
-#- bases/syncset.gatekeeper.sh_syncsets.yaml
+- bases/syncset.gatekeeper.sh_syncsets.yaml
 - bases/status.gatekeeper.sh_constraintpodstatuses.yaml
 - bases/status.gatekeeper.sh_constrainttemplatepodstatuses.yaml
 - bases/status.gatekeeper.sh_mutatorpodstatuses.yaml
@@ -50,6 +50,12 @@ patchesJson6902:
     kind: CustomResourceDefinition
     name: assignimage.mutations.gatekeeper.sh
   path: patches/max_name_size.yaml
+- target:
+    group: apiextensions.k8s.io
+    version: v1
+    kind: CustomResourceDefinition
+    name: syncsets.syncset.gatekeeper.sh
+  path: patches/max_name_size.yaml
 
 patchesStrategicMerge:
 #- patches/max_name_size_for_modifyset.yaml

main.go

@@ -53,6 +53,7 @@ import (
 	"github.com/open-policy-agent/gatekeeper/v3/pkg/operations"
 	"github.com/open-policy-agent/gatekeeper/v3/pkg/pubsub"
 	"github.com/open-policy-agent/gatekeeper/v3/pkg/readiness"
+	"github.com/open-policy-agent/gatekeeper/v3/pkg/readiness/pruner"
 	"github.com/open-policy-agent/gatekeeper/v3/pkg/syncutil"
 	"github.com/open-policy-agent/gatekeeper/v3/pkg/target"
 	"github.com/open-policy-agent/gatekeeper/v3/pkg/upgrade"
@@ -486,6 +487,12 @@ func setupControllers(ctx context.Context, mgr ctrl.Manager, sw *watch.Controlle
 		return err
 	}
 
+	err = mgr.Add(pruner.NewExpectationsPruner(cm, tracker))
+	if err != nil {
+		setupLog.Error(err, "adding expectations pruner to manager")
+		return err
+	}
+
 	opts := controller.Dependencies{
 		CFClient:         client,
 		WatchManger:      wm,

manifest_staging/charts/gatekeeper/crds/syncset-customresourcedefinition.yaml

@@ -0,0 +1,52 @@
+apiVersion: apiextensions.k8s.io/v1
+kind: CustomResourceDefinition
+metadata:
+  annotations:
+    controller-gen.kubebuilder.io/version: v0.10.0
+  labels:
+    gatekeeper.sh/system: "yes"
+  name: syncsets.syncset.gatekeeper.sh
+spec:
+  group: syncset.gatekeeper.sh
+  names:
+    kind: SyncSet
+    listKind: SyncSetList
+    plural: syncsets
+    singular: syncset
+  preserveUnknownFields: false
+  scope: Cluster
+  versions:
+  - name: v1alpha1
+    schema:
+      openAPIV3Schema:
+        description: SyncSet is the Schema for the SyncSet API.
+        properties:
+          apiVersion:
+            description: 'APIVersion defines the versioned schema of this representation of an object. Servers should convert recognized schemas to the latest internal value, and may reject unrecognized values. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#resources'
+            type: string
+          kind:
+            description: 'Kind is a string value representing the REST resource this object represents. Servers may infer this from the endpoint the client submits requests to. Cannot be updated. In CamelCase. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds'
+            type: string
+          metadata:
+            properties:
+              name:
+                maxLength: 63
+                type: string
+            type: object
+          spec:
+            properties:
+              gvks:
+                items:
+                  properties:
+                    group:
+                      type: string
+                    kind:
+                      type: string
+                    version:
+                      type: string
+                  type: object
+                type: array
+            type: object
+        type: object
+    served: true
+    storage: true

manifest_staging/deploy/gatekeeper.yaml

@@ -3367,6 +3367,59 @@ spec:
     served: true
     storage: true
 ---
+apiVersion: apiextensions.k8s.io/v1
+kind: CustomResourceDefinition
+metadata:
+  annotations:
+    controller-gen.kubebuilder.io/version: v0.10.0
+  labels:
+    gatekeeper.sh/system: "yes"
+  name: syncsets.syncset.gatekeeper.sh
+spec:
+  group: syncset.gatekeeper.sh
+  names:
+    kind: SyncSet
+    listKind: SyncSetList
+    plural: syncsets
+    singular: syncset
+  preserveUnknownFields: false
+  scope: Cluster
+  versions:
+  - name: v1alpha1
+    schema:
+      openAPIV3Schema:
+        description: SyncSet is the Schema for the SyncSet API.
+        properties:
+          apiVersion:
+            description: 'APIVersion defines the versioned schema of this representation of an object. Servers should convert recognized schemas to the latest internal value, and may reject unrecognized values. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#resources'
+            type: string
+          kind:
+            description: 'Kind is a string value representing the REST resource this object represents. Servers may infer this from the endpoint the client submits requests to. Cannot be updated. In CamelCase. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds'
+            type: string
+          metadata:
+            properties:
+              name:
+                maxLength: 63
+                type: string
+            type: object
+          spec:
+            properties:
+              gvks:
+                items:
+                  properties:
+                    group:
+                      type: string
+                    kind:
+                      type: string
+                    version:
+                      type: string
+                  type: object
+                type: array
+            type: object
+        type: object
+    served: true
+    storage: true
+---
 apiVersion: v1
 kind: ServiceAccount
 metadata:

pkg/cachemanager/cachemanager.go

@@ -2,6 +2,7 @@ package cachemanager
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"sync"
 	"time"
@@ -128,8 +129,21 @@ func (c *CacheManager) UpsertSource(ctx context.Context, sourceKey aggregator.Ke
 	// may become unreferenced and need to be deleted; this will be handled async
 	// in the manageCache loop.
 
-	if err := c.replaceWatchSet(ctx); err != nil {
-		return fmt.Errorf("error watching new gvks: %w", err)
+	err := c.replaceWatchSet(ctx)
+	if general, failedGVKs := interpretErr(err, newGVKs); len(failedGVKs) > 0 {
+		var gvksToTryCancel []schema.GroupVersionKind
+		if general {
+			// if the err is general, assume all gvks need TryCancel because of some
+			// WatchManager internal error and we don't want to block readiness.
+			gvksToTryCancel = c.gvksToSync.GVKs()
+		} else {
+			gvksToTryCancel = failedGVKs
+		}
+
+		for _, g := range gvksToTryCancel {
+			c.tracker.TryCancelData(g)
+		}
+		return fmt.Errorf("error establishing watches: %w", err)
 	}
 
 	return nil
@@ -140,11 +154,7 @@ func (c *CacheManager) UpsertSource(ctx context.Context, sourceKey aggregator.Ke
 func (c *CacheManager) replaceWatchSet(ctx context.Context) error {
 	newWatchSet := watch.NewSet()
 	newWatchSet.Add(c.gvksToSync.GVKs()...)
-
-	diff := c.watchedSet.Difference(newWatchSet)
-	c.removeStaleExpectations(diff)
-
-	c.gvksToDeleteFromCache.AddSet(diff)
+	c.gvksToDeleteFromCache.AddSet(c.watchedSet.Difference(newWatchSet))
 
 	var innerError error
 	c.watchedSet.Replace(newWatchSet, func() {
@@ -158,11 +168,31 @@ func (c *CacheManager) replaceWatchSet(ctx context.Context) error {
 	return innerError
 }
 
-// removeStaleExpectations stops tracking data for any resources that are no longer watched.
-func (c *CacheManager) removeStaleExpectations(stale *watch.Set) {
-	for _, gvk := range stale.Items() {
-		c.tracker.CancelData(gvk)
+// interpret if the err received is general or whether it is specific to the provided GVKs.
+func interpretErr(e error, gvks []schema.GroupVersionKind) (bool, []schema.GroupVersionKind) {
+	if e == nil {
+		return false, nil
+	}
+
+	var f watch.WatchesError
+	if !errors.As(e, &f) || f.HasGeneralErr() {
+		return true, nil
+	}
+
+	failedGvks := watch.NewSet()
+	failedGvks.Add(f.FailingGVKs()...)
+	sourceGVKSet := watch.NewSet()
+	sourceGVKSet.Add(gvks...)
+
+	common := failedGvks.Intersection(sourceGVKSet)
+	if common.Size() > 0 {
+		return false, common.Items()
 	}
+
+	// this error is not about the gvks in this request
+	// but we still log it for visibility
+	log.Info("encountered unrelated error when replacing watch set", "error", e)
+	return false, nil
 }
 
 // RemoveSource removes the watches of the GVKs for a given aggregator.Key. Callers are responsible for retrying on error.
@@ -174,7 +204,8 @@ func (c *CacheManager) RemoveSource(ctx context.Context, sourceKey aggregator.Ke
 		return fmt.Errorf("internal error removing source: %w", err)
 	}
 
-	if err := c.replaceWatchSet(ctx); err != nil {
+	err := c.replaceWatchSet(ctx)
+	if general, _ := interpretErr(err, []schema.GroupVersionKind{}); general {
 		return fmt.Errorf("error removing watches for source %v: %w", sourceKey, err)
 	}
 
@@ -208,6 +239,13 @@ func (c *CacheManager) DoForEach(fn func(gvk schema.GroupVersionKind) error) err
 	return err
 }
 
+func (c *CacheManager) WatchedGVKs() []schema.GroupVersionKind {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+
+	return c.watchedSet.Items()
+}
+
 func (c *CacheManager) watchesGVK(gvk schema.GroupVersionKind) bool {
 	c.mu.RLock()
 	defer c.mu.RUnlock()