feat: add syncset controller, feat: syncset readiness

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

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,9 @@ func setupControllers(ctx context.Context, mgr ctrl.Manager, sw *watch.Controlle
 		return err
 	}
 
+	p := pruner.NewExpecationsPruner(cm, tracker)
+	go p.Run(ctx)
+
 	opts := controller.Dependencies{
 		CFClient:         client,
 		WatchManger:      wm,

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

@@ -0,0 +1,48 @@
+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:
+            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,55 @@ 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:
+            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,10 +2,12 @@ package cachemanager
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"sync"
 	"time"
 
+	"github.com/go-logr/logr"
 	"github.com/open-policy-agent/frameworks/constraint/pkg/types"
 	"github.com/open-policy-agent/gatekeeper/v3/pkg/cachemanager/aggregator"
 	"github.com/open-policy-agent/gatekeeper/v3/pkg/controller/config/process"
@@ -128,8 +130,12 @@ 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 failedGVKs, interpreted := interpretErr(log, err, newGVKs); interpreted != nil {
+		for _, g := range failedGVKs {
+			c.tracker.TryCancelData(g)
+		}
+		return fmt.Errorf("error establishing watches: %w", interpreted)
 	}
 
 	return nil
@@ -142,8 +148,7 @@ func (c *CacheManager) replaceWatchSet(ctx context.Context) error {
 	newWatchSet.Add(c.gvksToSync.GVKs()...)
 
 	diff := c.watchedSet.Difference(newWatchSet)
-	c.removeStaleExpectations(diff)
-
+	// any resulting stale data expectations are handled async by the ExpectationsMgr.
 	c.gvksToDeleteFromCache.AddSet(diff)
 
 	var innerError error
@@ -158,11 +163,43 @@ 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)
+// interpretErr looks at the error e and unwraps it looking to find an error FailingGVKs() in the error chain.
+// If found and there is an intersection with the given gvks parameter with the failing gvks, we return the failing
+// gvks and an error that is meant to bubble up to controllers. If there is no intersection, we log the error
+// so as not to fully swallow any issues. If no error implemeting FailingGVKs() is found, we consider that
+// to be a "global error", so we return all gvks passed in and the error.
+func interpretErr(logger logr.Logger, e error, gvks []schema.GroupVersionKind) ([]schema.GroupVersionKind, error) {
+	if e == nil {
+		return nil, nil
+	}
+
+	var f interface {
+		FailingGVKs() []schema.GroupVersionKind
+		Error() string
+	}
+
+	// search for the first error that implements the FailingGVKs() interface
+	if errors.As(e, &f) {
+		// this error includes failing gvks; let's match against the original list
+		failedGvks := watch.NewSet()
+		failedGvks.Add(f.FailingGVKs()...)
+		gvksSet := watch.NewSet()
+		gvksSet.Add(gvks...)
+
+		common := failedGvks.Intersection(gvksSet)
+		if common.Size() > 0 {
+			// then this error is pertinent to the gvks and needs to be returned
+			return common.Items(), e
+		}
+
+		// if no intersection, this error is not about the gvks in this request
+		// but we still log it for visibility
+		logger.Info("encountered unrelated error when replacing watch set", "error", e)
+		return nil, nil
 	}
+
+	// otherwise, this is a "global error"
+	return gvks, e
 }
 
 // RemoveSource removes the watches of the GVKs for a given aggregator.Key. Callers are responsible for retrying on error.
@@ -174,8 +211,11 @@ 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 {
-		return fmt.Errorf("error removing watches for source %v: %w", sourceKey, err)
+	err := c.replaceWatchSet(ctx)
+	if _, interpreted := interpretErr(log, err, []schema.GroupVersionKind{}); interpreted != nil {
+		// unlike UpsertSource, we cannot TryCancel or Cancel any expectations as these
+		// GVKs may be required by other sync sources.
+		return fmt.Errorf("error removing watches for source %v: %w", sourceKey, interpreted)
 	}
 
 	return nil
@@ -208,6 +248,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()