contentmanager: use namespace-scoped informers

Signed-off-by: Joe Lanford <[email protected]>

Author: joelanford

internal/operator-controller/contentmanager/cache/cache.go

@@ -41,15 +41,15 @@ type CloserSyncingSource interface {
 }
 
 type sourcerer interface {
-	// Source returns a CloserSyncingSource for the provided
-	// GroupVersionKind. If the CloserSyncingSource encounters an
+	// Source returns a CloserSyncingSource for the provided namespace
+	// and GroupVersionKind. If the CloserSyncingSource encounters an
 	// error after having initially synced, it should requeue the
 	// provided client.Object and call the provided callback function
-	Source(schema.GroupVersionKind, client.Object, func(context.Context)) (CloserSyncingSource, error)
+	Source(string, schema.GroupVersionKind, client.Object, func(context.Context)) (CloserSyncingSource, error)
 }
 
 type cache struct {
-	sources     map[schema.GroupVersionKind]CloserSyncingSource
+	sources     map[sourceKey]CloserSyncingSource
 	sourcerer   sourcerer
 	owner       client.Object
 	syncTimeout time.Duration
@@ -58,7 +58,7 @@ type cache struct {
 
 func NewCache(sourcerer sourcerer, owner client.Object, syncTimeout time.Duration) Cache {
 	return &cache{
-		sources:     make(map[schema.GroupVersionKind]CloserSyncingSource),
+		sources:     make(map[sourceKey]CloserSyncingSource),
 		sourcerer:   sourcerer,
 		owner:       owner,
 		syncTimeout: syncTimeout,
@@ -70,15 +70,15 @@ var _ Cache = (*cache)(nil)
 func (c *cache) Watch(ctx context.Context, watcher Watcher, objs ...client.Object) error {
 	c.mu.Lock()
 	defer c.mu.Unlock()
-	gvkSet, err := gvksForObjects(objs...)
+	sourceKeySet, err := sourceKeysForObjects(objs...)
 	if err != nil {
 		return fmt.Errorf("getting set of GVKs for managed objects: %w", err)
 	}
 
-	if err := c.removeStaleSources(gvkSet); err != nil {
+	if err := c.removeStaleSources(sourceKeySet); err != nil {
 		return fmt.Errorf("removing stale sources: %w", err)
 	}
-	return c.startNewSources(ctx, gvkSet, watcher)
+	return c.startNewSources(ctx, sourceKeySet, watcher)
 }
 
 func (c *cache) Close() error {
@@ -99,29 +99,35 @@ func (c *cache) Close() error {
 	return errors.Join(errs...)
 }
 
-func (c *cache) startNewSources(ctx context.Context, gvks sets.Set[schema.GroupVersionKind], watcher Watcher) error {
-	cacheGvks := c.getCacheGVKs()
-	gvksToCreate := gvks.Difference(cacheGvks)
+type sourceKey struct {
+	namespace string
+	gvk       schema.GroupVersionKind
+}
 
+func (c *cache) startNewSources(ctx context.Context, sources sets.Set[sourceKey], watcher Watcher) error {
 	type startResult struct {
 		source CloserSyncingSource
-		gvk    schema.GroupVersionKind
+		key    sourceKey
 		err    error
 	}
 	startResults := make(chan startResult)
 	wg := sync.WaitGroup{}
-	for _, gvk := range gvksToCreate.UnsortedList() {
+
+	existingSourceKeys := c.getCacheKeys()
+	sourcesToCreate := sources.Difference(existingSourceKeys)
+	for _, srcKey := range sourcesToCreate.UnsortedList() {
 		wg.Add(1)
 		go func() {
 			defer wg.Done()
-			source, err := c.startNewSource(ctx, gvk, watcher)
+			source, err := c.startNewSource(ctx, srcKey, watcher)
 			startResults <- startResult{
 				source: source,
-				gvk:    gvk,
+				key:    srcKey,
 				err:    err,
 			}
 		}()
 	}
+
 	go func() {
 		wg.Wait()
 		close(startResults)
@@ -134,7 +140,7 @@ func (c *cache) startNewSources(ctx context.Context, gvks sets.Set[schema.GroupV
 			continue
 		}
 
-		err := c.addSource(result.gvk, result.source)
+		err := c.addSource(result.key, result.source)
 		if err != nil {
 			// If we made it here then there is a logic error in
 			// calculating the diffs between what is currently being
@@ -146,20 +152,19 @@ func (c *cache) startNewSources(ctx context.Context, gvks sets.Set[schema.GroupV
 	slices.SortFunc(sourcesErrors, func(a, b error) int {
 		return strings.Compare(a.Error(), b.Error())
 	})
-
 	return errors.Join(sourcesErrors...)
 }
 
-func (c *cache) startNewSource(ctx context.Context, gvk schema.GroupVersionKind, watcher Watcher) (CloserSyncingSource, error) {
-	s, err := c.sourcerer.Source(gvk, c.owner, func(ctx context.Context) {
+func (c *cache) startNewSource(ctx context.Context, srcKey sourceKey, watcher Watcher) (CloserSyncingSource, error) {
+	s, err := c.sourcerer.Source(srcKey.namespace, srcKey.gvk, c.owner, func(ctx context.Context) {
 		// this callback function ensures that we remove the source from the
 		// cache if it encounters an error after it initially synced successfully
 		c.mu.Lock()
 		defer c.mu.Unlock()
-		err := c.removeSource(gvk)
+		err := c.removeSource(srcKey)
 		if err != nil {
 			logr := log.FromContext(ctx)
-			logr.Error(err, "managed content cache postSyncError removing source failed", "gvk", gvk)
+			logr.Error(err, "managed content cache postSyncError removing source failed", "namespace", srcKey.namespace, "gvk", srcKey.gvk)
 		}
 	})
 	if err != nil {
@@ -168,7 +173,7 @@ func (c *cache) startNewSource(ctx context.Context, gvk schema.GroupVersionKind,
 
 	err = watcher.Watch(s)
 	if err != nil {
-		return nil, fmt.Errorf("establishing watch for GVK %q: %w", gvk, err)
+		return nil, fmt.Errorf("establishing watch for GVK %q in namespace %q: %w", srcKey.gvk, srcKey.namespace, err)
 	}
 
 	syncCtx, syncCancel := context.WithTimeout(ctx, c.syncTimeout)
@@ -181,19 +186,19 @@ func (c *cache) startNewSource(ctx context.Context, gvk schema.GroupVersionKind,
 	return s, nil
 }
 
-func (c *cache) addSource(gvk schema.GroupVersionKind, source CloserSyncingSource) error {
-	if _, ok := c.sources[gvk]; !ok {
-		c.sources[gvk] = source
+func (c *cache) addSource(key sourceKey, source CloserSyncingSource) error {
+	if _, ok := c.sources[key]; !ok {
+		c.sources[key] = source
 		return nil
 	}
 	return errors.New("source already exists")
 }
 
-func (c *cache) removeStaleSources(gvks sets.Set[schema.GroupVersionKind]) error {
-	cacheGvks := c.getCacheGVKs()
+func (c *cache) removeStaleSources(srcKeys sets.Set[sourceKey]) error {
+	existingSrcKeys := c.getCacheKeys()
 	removeErrs := []error{}
-	gvksToRemove := cacheGvks.Difference(gvks)
-	for _, gvk := range gvksToRemove.UnsortedList() {
+	srcKeysToRemove := existingSrcKeys.Difference(srcKeys)
+	for _, gvk := range srcKeysToRemove.UnsortedList() {
 		err := c.removeSource(gvk)
 		if err != nil {
 			removeErrs = append(removeErrs, err)
@@ -207,23 +212,23 @@ func (c *cache) removeStaleSources(gvks sets.Set[schema.GroupVersionKind]) error
 	return errors.Join(removeErrs...)
 }
 
-func (c *cache) removeSource(gvk schema.GroupVersionKind) error {
-	if source, ok := c.sources[gvk]; ok {
-		err := source.Close()
+func (c *cache) removeSource(srcKey sourceKey) error {
+	if src, ok := c.sources[srcKey]; ok {
+		err := src.Close()
 		if err != nil {
-			return fmt.Errorf("closing source for GVK %q: %w", gvk, err)
+			return fmt.Errorf("closing source for GVK %q in namespace %q: %w", srcKey.gvk, srcKey.namespace, err)
 		}
 	}
-	delete(c.sources, gvk)
+	delete(c.sources, srcKey)
 	return nil
 }
 
-func (c *cache) getCacheGVKs() sets.Set[schema.GroupVersionKind] {
-	cacheGvks := sets.New[schema.GroupVersionKind]()
-	for gvk := range c.sources {
-		cacheGvks.Insert(gvk)
+func (c *cache) getCacheKeys() sets.Set[sourceKey] {
+	sourceKeys := sets.New[sourceKey]()
+	for key := range c.sources {
+		sourceKeys.Insert(key)
 	}
-	return cacheGvks
+	return sourceKeys
 }
 
 // gvksForObjects builds a sets.Set of GroupVersionKinds for
@@ -233,9 +238,10 @@ func (c *cache) getCacheGVKs() sets.Set[schema.GroupVersionKind] {
 //
 // An empty Group is assumed to be the "core" Kubernetes
 // API group.
-func gvksForObjects(objs ...client.Object) (sets.Set[schema.GroupVersionKind], error) {
-	gvkSet := sets.New[schema.GroupVersionKind]()
+func sourceKeysForObjects(objs ...client.Object) (sets.Set[sourceKey], error) {
+	sourceKeys := sets.New[sourceKey]()
 	for _, obj := range objs {
+		ns := obj.GetNamespace()
 		gvk := obj.GetObjectKind().GroupVersionKind()
 
 		// If the Kind or Version is not set in an object's GroupVersionKind
@@ -257,8 +263,8 @@ func gvksForObjects(objs ...client.Object) (sets.Set[schema.GroupVersionKind], e
 			)
 		}
 
-		gvkSet.Insert(gvk)
+		sourceKeys.Insert(sourceKey{ns, gvk})
 	}
 
-	return gvkSet, nil
+	return sourceKeys, nil
 }

internal/operator-controller/contentmanager/cache/cache_test.go

@@ -9,6 +9,7 @@ import (
 	"github.com/stretchr/testify/require"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/util/rand"
 	"k8s.io/client-go/util/workqueue"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
@@ -34,7 +35,7 @@ type mockSourcerer struct {
 
 var _ sourcerer = (*mockSourcerer)(nil)
 
-func (ms *mockSourcerer) Source(_ schema.GroupVersionKind, _ client.Object, _ func(context.Context)) (CloserSyncingSource, error) {
+func (ms *mockSourcerer) Source(_ string, _ schema.GroupVersionKind, _ client.Object, _ func(context.Context)) (CloserSyncingSource, error) {
 	if ms.err != nil {
 		return nil, ms.err
 	}
@@ -69,11 +70,11 @@ func TestCacheWatch(t *testing.T) {
 	)
 
 	pod := &corev1.Pod{}
-	podGvk := corev1.SchemeGroupVersion.WithKind("Pod")
-	pod.SetGroupVersionKind(podGvk)
+	pod.SetGroupVersionKind(corev1.SchemeGroupVersion.WithKind("Pod"))
+	pod.SetNamespace(rand.String(8))
 
 	require.NoError(t, c.Watch(context.Background(), &mockWatcher{}, pod))
-	require.Contains(t, c.(*cache).sources, podGvk, "sources", c.(*cache).sources)
+	require.Contains(t, c.(*cache).sources, sourceKey{pod.Namespace, pod.GroupVersionKind()}, "sources", c.(*cache).sources)
 }
 
 func TestCacheWatchInvalidGVK(t *testing.T) {
@@ -147,9 +148,9 @@ func TestCacheWatchExistingSourceNotPanic(t *testing.T) {
 	)
 
 	pod := &corev1.Pod{}
-	podGvk := corev1.SchemeGroupVersion.WithKind("Pod")
-	pod.SetGroupVersionKind(podGvk)
-	require.NoError(t, c.(*cache).addSource(podGvk, &mockSource{}))
+	pod.SetGroupVersionKind(corev1.SchemeGroupVersion.WithKind("Pod"))
+	pod.SetNamespace(rand.String(8))
+	require.NoError(t, c.(*cache).addSource(sourceKey{pod.Namespace, pod.GroupVersionKind()}, &mockSource{}))
 
 	// In this case, a panic means there is a logic error somewhere in the
 	// cache.Watch() method. It should never hit the condition where it panics
@@ -167,18 +168,18 @@ func TestCacheWatchRemovesStaleSources(t *testing.T) {
 	)
 
 	pod := &corev1.Pod{}
-	podGvk := corev1.SchemeGroupVersion.WithKind("Pod")
-	pod.SetGroupVersionKind(podGvk)
+	pod.SetGroupVersionKind(corev1.SchemeGroupVersion.WithKind("Pod"))
+	pod.SetNamespace(rand.String(8))
 
 	require.NoError(t, c.Watch(context.Background(), &mockWatcher{}, pod))
-	require.Contains(t, c.(*cache).sources, podGvk)
+	require.Contains(t, c.(*cache).sources, sourceKey{pod.Namespace, pod.GroupVersionKind()})
 
 	secret := &corev1.Secret{}
-	secretGvk := corev1.SchemeGroupVersion.WithKind("Secret")
-	secret.SetGroupVersionKind(secretGvk)
+	secret.SetGroupVersionKind(corev1.SchemeGroupVersion.WithKind("Secret"))
+	secret.SetNamespace(rand.String(8))
 	require.NoError(t, c.Watch(context.Background(), &mockWatcher{}, secret))
-	require.Contains(t, c.(*cache).sources, secretGvk)
-	require.NotContains(t, c.(*cache).sources, podGvk)
+	require.Contains(t, c.(*cache).sources, sourceKey{secret.Namespace, secret.GroupVersionKind()})
+	require.NotContains(t, c.(*cache).sources, sourceKey{pod.Namespace, pod.GroupVersionKind()})
 }
 
 func TestCacheWatchRemovingStaleSourcesError(t *testing.T) {
@@ -190,13 +191,16 @@ func TestCacheWatchRemovingStaleSourcesError(t *testing.T) {
 		time.Second,
 	)
 
-	podGvk := corev1.SchemeGroupVersion.WithKind("Pod")
-	require.NoError(t, c.(*cache).addSource(podGvk, &mockSource{
+	podSourceKey := sourceKey{
+		namespace: rand.String(8),
+		gvk:       corev1.SchemeGroupVersion.WithKind("Pod"),
+	}
+	require.NoError(t, c.(*cache).addSource(podSourceKey, &mockSource{
 		err: errors.New("error"),
 	}))
 
 	secret := &corev1.Secret{}
-	secretGvk := corev1.SchemeGroupVersion.WithKind("Secret")
-	secret.SetGroupVersionKind(secretGvk)
+	secret.SetGroupVersionKind(corev1.SchemeGroupVersion.WithKind("Secret"))
+	secret.SetNamespace(rand.String(8))
 	require.Error(t, c.Watch(context.Background(), &mockWatcher{}, secret))
 }

internal/operator-controller/contentmanager/contentmanager.go

@@ -116,8 +116,8 @@ func (i *managerImpl) Get(ctx context.Context, ce *ocv1.ClusterExtension) (cmcac
 		// related to reusing an informer factory, we return a new informer
 		// factory every time to ensure we are not attempting to configure or
 		// start an already started informer
-		informerFactoryCreateFunc: func() dynamicinformer.DynamicSharedInformerFactory {
-			return dynamicinformer.NewFilteredDynamicSharedInformerFactory(dynamicClient, time.Hour*10, metav1.NamespaceAll, func(lo *metav1.ListOptions) {
+		informerFactoryCreateFunc: func(namespace string) dynamicinformer.DynamicSharedInformerFactory {
+			return dynamicinformer.NewFilteredDynamicSharedInformerFactory(dynamicClient, time.Hour*10, namespace, func(lo *metav1.ListOptions) {
 				lo.LabelSelector = tgtLabels.AsSelector().String()
 			})
 		},

internal/operator-controller/contentmanager/sourcerer.go

@@ -21,11 +21,11 @@ import (
 )
 
 type dynamicSourcerer struct {
-	informerFactoryCreateFunc func() dynamicinformer.DynamicSharedInformerFactory
+	informerFactoryCreateFunc func(namespace string) dynamicinformer.DynamicSharedInformerFactory
 	mapper                    meta.RESTMapper
 }
 
-func (ds *dynamicSourcerer) Source(gvk schema.GroupVersionKind, owner client.Object, onPostSyncError func(context.Context)) (cache.CloserSyncingSource, error) {
+func (ds *dynamicSourcerer) Source(namespace string, gvk schema.GroupVersionKind, owner client.Object, onPostSyncError func(context.Context)) (cache.CloserSyncingSource, error) {
 	scheme, err := buildScheme(gvk)
 	if err != nil {
 		return nil, fmt.Errorf("building scheme: %w", err)
@@ -48,7 +48,7 @@ func (ds *dynamicSourcerer) Source(gvk schema.GroupVersionKind, owner client.Obj
 				GenericFunc: func(tge event.TypedGenericEvent[client.Object]) bool { return true },
 			},
 		},
-		DynamicInformerFactory: ds.informerFactoryCreateFunc(),
+		DynamicInformerFactory: ds.informerFactoryCreateFunc(namespace),
 		OnPostSyncError:        onPostSyncError,
 	})
 	return s, nil