feat: support multiple sync sources

main.go

@@ -43,6 +43,7 @@ import (
 	mutationsv1beta1 "github.com/open-policy-agent/gatekeeper/v3/apis/mutations/v1beta1"
 	statusv1beta1 "github.com/open-policy-agent/gatekeeper/v3/apis/status/v1beta1"
 	"github.com/open-policy-agent/gatekeeper/v3/pkg/audit"
+	"github.com/open-policy-agent/gatekeeper/v3/pkg/cachemanager"
 	"github.com/open-policy-agent/gatekeeper/v3/pkg/controller"
 	"github.com/open-policy-agent/gatekeeper/v3/pkg/controller/config/process"
 	"github.com/open-policy-agent/gatekeeper/v3/pkg/expansion"
@@ -52,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/syncutil"
 	"github.com/open-policy-agent/gatekeeper/v3/pkg/target"
 	"github.com/open-policy-agent/gatekeeper/v3/pkg/upgrade"
 	"github.com/open-policy-agent/gatekeeper/v3/pkg/util"
@@ -69,6 +71,7 @@ import (
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/certwatcher"
 	"sigs.k8s.io/controller-runtime/pkg/client/apiutil"
+	"sigs.k8s.io/controller-runtime/pkg/event"
 	"sigs.k8s.io/controller-runtime/pkg/healthz"
 	crzap "sigs.k8s.io/controller-runtime/pkg/log/zap"
 	crWebhook "sigs.k8s.io/controller-runtime/pkg/webhook"
@@ -447,17 +450,43 @@ func setupControllers(ctx context.Context, mgr ctrl.Manager, sw *watch.Controlle
 
 	// Setup all Controllers
 	setupLog.Info("setting up controllers")
-	watchSet := watch.NewSet()
+
+	// Events ch will be used to receive events from dynamic watches registered
+	// via the registrar below.
+	events := make(chan event.GenericEvent, 1024)
+	w, err := wm.NewRegistrar(
+		cachemanager.RegistrarName,
+		events)
+	if err != nil {
+		setupLog.Error(err, "unable to set up watch registrar for cache manager")
+		return err
+	}
+
+	syncMetricsCache := syncutil.NewMetricsCache()
+	cm, err := cachemanager.NewCacheManager(&cachemanager.Config{
+		CfClient:         client,
+		SyncMetricsCache: syncMetricsCache,
+		Tracker:          tracker,
+		ProcessExcluder:  processExcluder,
+		Registrar:        w,
+		Reader:           mgr.GetCache(),
+	})
+	if err != nil {
+		setupLog.Error(err, "unable to create cache manager")
+		return err
+	}
+
 	opts := controller.Dependencies{
 		Opa:              client,
 		WatchManger:      wm,
+		SyncEventsCh:     events,
+		CacheMgr:         cm,
 		ControllerSwitch: sw,
 		Tracker:          tracker,
 		ProcessExcluder:  processExcluder,
 		MutationSystem:   mutationSystem,
 		ExpansionSystem:  expansionSystem,
 		ProviderCache:    providerCache,
-		WatchSet:         watchSet,
 		PubsubSystem:     pubsubSystem,
 	}
 
@@ -482,7 +511,7 @@ func setupControllers(ctx context.Context, mgr ctrl.Manager, sw *watch.Controlle
 
 	if operations.IsAssigned(operations.Audit) {
 		setupLog.Info("setting up audit")
-		auditCache := audit.NewAuditCacheLister(mgr.GetCache(), watchSet)
+		auditCache := audit.NewAuditCacheLister(mgr.GetCache(), cm)
 		auditDeps := audit.Dependencies{
 			Client:          client,
 			ProcessExcluder: processExcluder,

pkg/audit/audit_cache_lister.go

@@ -3,18 +3,17 @@ package audit
 import (
 	"context"
 
-	"github.com/open-policy-agent/gatekeeper/v3/pkg/watch"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
 // NewAuditCacheLister instantiates a new AuditCache which will read objects in
 // watched from auditCache.
-func NewAuditCacheLister(auditCache client.Reader, watched *watch.Set) *CacheLister {
+func NewAuditCacheLister(auditCache client.Reader, lister WatchIterator) *CacheLister {
 	return &CacheLister{
 		auditCache: auditCache,
-		watched:    watched,
+		lister:     lister,
 	}
 }
 
@@ -25,14 +24,19 @@ type CacheLister struct {
 	// Caution: only to be read from while watched is locked, such as through
 	// DoForEach.
 	auditCache client.Reader
-	// watched is the set of objects watched by the audit cache.
-	watched *watch.Set
+	// lister is a delegate like cachemanager that we can use to query a watched set of GKVs.
+	lister WatchIterator
+}
+
+// wraps DoForEach from a watch.Set.
+type WatchIterator interface {
+	DoForEach(listFunc func(gvk schema.GroupVersionKind) error) error
 }
 
 // ListObjects lists all objects from the audit cache.
 func (l *CacheLister) ListObjects(ctx context.Context) ([]unstructured.Unstructured, error) {
 	var objs []unstructured.Unstructured
-	err := l.watched.DoForEach(func(gvk schema.GroupVersionKind) error {
+	err := l.lister.DoForEach(func(gvk schema.GroupVersionKind) error {
 		gvkObjects, err := listObjects(ctx, l.auditCache, gvk)
 		if err != nil {
 			return err

pkg/cachemanager/cachemanager.go

@@ -21,6 +21,8 @@ import (
 	logf "sigs.k8s.io/controller-runtime/pkg/log"
 )
 
+const RegistrarName = "cachemanager"
+
 var (
 	log     = logf.Log.WithName("cache-manager")
 	backoff = wait.Backoff{
@@ -37,7 +39,6 @@ type Config struct {
 	Tracker          *readiness.Tracker
 	ProcessExcluder  *process.Excluder
 	Registrar        *watch.Registrar
-	WatchedSet       *watch.Set
 	GVKAggregator    *aggregator.GVKAgreggator
 	Reader           client.Reader
 }
@@ -67,9 +68,6 @@ type CFDataClient interface {
 }
 
 func NewCacheManager(config *Config) (*CacheManager, error) {
-	if config.WatchedSet == nil {
-		return nil, fmt.Errorf("watchedSet must be non-nil")
-	}
 	if config.Registrar == nil {
 		return nil, fmt.Errorf("registrar must be non-nil")
 	}
@@ -93,7 +91,7 @@ func NewCacheManager(config *Config) (*CacheManager, error) {
 		tracker:                    config.Tracker,
 		processExcluder:            config.ProcessExcluder,
 		registrar:                  config.Registrar,
-		watchedSet:                 config.WatchedSet,
+		watchedSet:                 watch.NewSet(),
 		reader:                     config.Reader,
 		gvksToSync:                 config.GVKAggregator,
 		backgroundManagementTicker: *time.NewTicker(3 * time.Second),
@@ -111,8 +109,7 @@ func (c *CacheManager) Start(ctx context.Context) error {
 }
 
 // UpsertSource adjusts the watched set of gvks according to the newGVKs passed in
-// for a given sourceKey.
-// Callers are responsible for retrying on error.
+// for a given sourceKey. Callers are responsible for retrying on error.
 func (c *CacheManager) UpsertSource(ctx context.Context, sourceKey aggregator.Key, newGVKs []schema.GroupVersionKind) error {
 	c.mu.Lock()
 	defer c.mu.Unlock()
@@ -139,8 +136,7 @@ func (c *CacheManager) UpsertSource(ctx context.Context, sourceKey aggregator.Ke
 }
 
 // replaceWatchSet looks at the gvksToSync and makes changes to the registrar's watch set.
-// assumes caller has lock.
-// On error, actual watch state may not align with intended watch state.
+// Assumes caller has lock. On error, actual watch state may not align with intended watch state.
 func (c *CacheManager) replaceWatchSet(ctx context.Context) error {
 	newWatchSet := watch.NewSet()
 	newWatchSet.Add(c.gvksToSync.GVKs()...)
@@ -159,8 +155,7 @@ func (c *CacheManager) replaceWatchSet(ctx context.Context) error {
 	return innerError
 }
 
-// RemoveSource removes the watches of the GVKs for a given aggregator.Key.
-// Callers are responsible for retrying on error.
+// RemoveSource removes the watches of the GVKs for a given aggregator.Key. Callers are responsible for retrying on error.
 func (c *CacheManager) RemoveSource(ctx context.Context, sourceKey aggregator.Key) error {
 	c.mu.Lock()
 	defer c.mu.Unlock()
@@ -193,6 +188,15 @@ func (c *CacheManager) ExcludeProcesses(newExcluder *process.Excluder) {
 	c.excluderChanged = true
 }
 
+// DoForEach runs fn function for each GVK that is being watched by the cache manager.
+func (c *CacheManager) DoForEach(fn func(gvk schema.GroupVersionKind) error) error {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	err := c.watchedSet.DoForEach(fn)
+	return err
+}
+
 func (c *CacheManager) watchesGVK(gvk schema.GroupVersionKind) bool {
 	c.mu.RLock()
 	defer c.mu.RUnlock()
@@ -241,12 +245,8 @@ func (c *CacheManager) AddObject(ctx context.Context, instance *unstructured.Uns
 }
 
 func (c *CacheManager) RemoveObject(ctx context.Context, instance *unstructured.Unstructured) error {
-	gvk := instance.GroupVersionKind()
-
-	if c.watchesGVK(gvk) {
-		if _, err := c.cfClient.RemoveData(ctx, instance); err != nil {
-			return err
-		}
+	if _, err := c.cfClient.RemoveData(ctx, instance); err != nil {
+		return err
 	}
 
 	// only delete from metrics map if the data removal was successful
@@ -383,8 +383,7 @@ func (c *CacheManager) replayGVKs(ctx context.Context, gvksToRelist []schema.Gro
 
 // wipeCacheIfNeeded performs a cache wipe if there are any gvks needing to be removed
 // from the cache or if the excluder has changed. It also marks which gvks need to be
-// re listed again in the cf data cache after the wipe.
-// assumes the caller has lock.
+// re listed again in the cf data cache after the wipe. Assumes the caller has lock.
 func (c *CacheManager) wipeCacheIfNeeded(ctx context.Context) {
 	// remove any gvks not needing to be synced anymore
 	// or re evaluate all if the excluder changed.