WIP: Add EnqueueWatching

managedcache/enqueue_watching.go

@@ -0,0 +1,128 @@
+package managedcache
+
+import (
+	"context"
+	"fmt"
+
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/util/workqueue"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/apiutil"
+	"sigs.k8s.io/controller-runtime/pkg/event"
+	"sigs.k8s.io/controller-runtime/pkg/handler"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+)
+
+// EnqueueWatchingObjects Enqueues all objects watching the object mentioned in the event, filtered by WatcherType.
+type EnqueueWatchingObjects struct {
+	WatcherRefGetter ownerRefGetter
+	// WatcherType is the type of the Owner object to look for in OwnerReferences.  Only Group and Kind are compared.
+	WatcherType runtime.Object
+
+	scheme *runtime.Scheme
+	// groupKind is the cached Group and Kind from WatcherType
+	groupKind schema.GroupKind
+}
+
+var _ handler.EventHandler = (*EnqueueWatchingObjects)(nil)
+
+// NewEnqueueWatchingObjects returns a new mapper putting the objects registered watching into the workqueue.
+func NewEnqueueWatchingObjects(watcherRefGetter ownerRefGetter,
+	watcherType runtime.Object,
+	scheme *runtime.Scheme,
+) *EnqueueWatchingObjects {
+	e := &EnqueueWatchingObjects{
+		WatcherRefGetter: watcherRefGetter,
+		WatcherType:      watcherType,
+		scheme:           scheme,
+	}
+	if err := e.parseWatcherTypeGroupKind(scheme); err != nil {
+		// This (passing a type that is not in the scheme) HAS
+		// to be a programmer error and can't be recovered at runtime anyways.
+		panic(err)
+	}
+
+	return e
+}
+
+type ownerRefGetter interface {
+	getWatchersForGVK(gvk schema.GroupVersionKind) []accessManagerKey
+}
+
+// Create implements handler.EventHandler.
+func (e *EnqueueWatchingObjects) Create(_ context.Context, evt event.CreateEvent,
+	q workqueue.TypedRateLimitingInterface[reconcile.Request],
+) {
+	e.enqueueWatchers(evt.Object, q)
+}
+
+// Update implements handler.EventHandler.
+func (e *EnqueueWatchingObjects) Update(_ context.Context, evt event.UpdateEvent,
+	q workqueue.TypedRateLimitingInterface[reconcile.Request],
+) {
+	e.enqueueWatchers(evt.ObjectNew, q)
+	e.enqueueWatchers(evt.ObjectOld, q)
+}
+
+// Delete implements handler.EventHandler.
+func (e *EnqueueWatchingObjects) Delete(_ context.Context, evt event.DeleteEvent,
+	q workqueue.TypedRateLimitingInterface[reconcile.Request],
+) {
+	e.enqueueWatchers(evt.Object, q)
+}
+
+// Generic implements handler.EventHandler.
+func (e *EnqueueWatchingObjects) Generic(_ context.Context, evt event.GenericEvent,
+	q workqueue.TypedRateLimitingInterface[reconcile.Request],
+) {
+	e.enqueueWatchers(evt.Object, q)
+}
+
+func (e *EnqueueWatchingObjects) enqueueWatchers(obj client.Object,
+	q workqueue.TypedRateLimitingInterface[reconcile.Request],
+) {
+	if obj == nil {
+		return
+	}
+
+	gvk, err := apiutil.GVKForObject(obj, e.scheme)
+	if err != nil {
+		// TODO: error reporting?
+		panic(err)
+	}
+
+	ownerRefs := e.WatcherRefGetter.getWatchersForGVK(gvk)
+	for _, ownerRef := range ownerRefs {
+		if ownerRef.Kind != e.groupKind.Kind ||
+			ownerRef.Group != e.groupKind.Group {
+			continue
+		}
+
+		q.Add(reconcile.Request{
+			NamespacedName: types.NamespacedName{
+				Name:      ownerRef.Name,
+				Namespace: ownerRef.Namespace,
+			},
+		})
+	}
+}
+
+// parseOwnerTypeGroupKind parses the WatcherType into a Group and Kind and caches the result.  Returns false
+// if the WatcherType could not be parsed using the scheme.
+func (e *EnqueueWatchingObjects) parseWatcherTypeGroupKind(scheme *runtime.Scheme) error {
+	// Get the kinds of the type
+	kinds, _, err := scheme.ObjectKinds(e.WatcherType)
+	if err != nil {
+		return err
+	}
+	// Expect only 1 kind.  If there is more than one kind this is probably an edge case such as ListOptions.
+	if len(kinds) != 1 {
+		panic(fmt.Sprintf("Expected exactly 1 kind for WatcherType %T, but found %s kinds", e.WatcherType, kinds))
+	}
+	// Cache the Group and Kind for the WatcherType
+	e.groupKind = schema.GroupKind{Group: kinds[0].Group, Kind: kinds[0].Kind}
+
+	return nil
+}

managedcache/objectboundaccess.go

@@ -73,7 +73,7 @@
 		cacheSourcer: &cacheSource{},
 		newClient:    client.New,
 
-		accessors:         map[types.UID]accessorEntry{},
+		accessors:         map[accessManagerKey]accessorEntry{},
 		accessorRequestCh: make(chan accessorRequest[T]),
 		accessorStopCh:    make(chan accessorRequest[T]),
 	}
@@ -103,14 +103,22 @@
 	cacheSourcer cacheSourcer
 	newClient    newClientFunc
 
-	accessors         map[types.UID]accessorEntry
+	accessorsLock     sync.RWMutex
+	accessors         map[accessManagerKey]accessorEntry
 	accessorRequestCh chan accessorRequest[T]
 	accessorStopCh    chan accessorRequest[T]
 }
 
+type accessManagerKey struct {
+	// UID ensures a re-created object also gets it's own cache.
+	UID types.UID
+	schema.GroupVersionKind
+	client.ObjectKey
+}
+
 type accessorEntry struct {
 	accessor Accessor
-	users    map[types.UID]sets.Set[schema.GroupVersionKind]
+	users    map[accessManagerKey]sets.Set[schema.GroupVersionKind]
 	cancel   func()
 }
 
@@ -128,7 +136,7 @@
 
 type cacheDone struct {
 	err error
-	uid types.UID
+	key accessManagerKey
 }
 
 // implements Accessor interface.
@@ -154,11 +162,8 @@
 	for {
 		select {
 		case done := <-doneCh:
-			// Remove accessor from list.
-			delete(m.accessors, done.uid)
-
-			if done.err != nil && !errors.Is(done.err, context.Canceled) {
-				return fmt.Errorf("cache for UID %s crashed: %w", done.uid, done.err)
+			if err := m.handleCacheDone(ctx, done); err != nil {
+				return err
 			}
 
 		case req := <-m.accessorRequestCh:
@@ -191,17 +196,36 @@
 	}
 }
 
+func (m *objectBoundAccessManagerImpl[T]) handleCacheDone(
+	_ context.Context, done cacheDone,
+) error {
+	m.accessorsLock.Lock()
+	defer m.accessorsLock.Unlock()
+
+	// Remove accessor from list.
+	delete(m.accessors, done.key)
+
+	if done.err != nil && !errors.Is(done.err, context.Canceled) {
+		return fmt.Errorf("cache for Key %s crashed: %w", done.key, done.err)
+	}
+
+	return nil
+}
+
 func (m *objectBoundAccessManagerImpl[T]) handleAccessorStop(
 	ctx context.Context, req accessorRequest[T],
 ) error {
-	cache, ok := m.accessors[req.owner.GetUID()]
+	m.accessorsLock.Lock()
+	defer m.accessorsLock.Unlock()
+
+	cache, ok := m.accessors[toAccessManagerKey(req.owner)]
 	if !ok {
 		// nothing todo.
 		return nil
 	}
 
 	if req.user != nil {
-		delete(cache.users, req.owner.GetUID())
+		delete(cache.users, toAccessManagerKey(req.user))
 	}
 
 	return m.gcCache(ctx, req.owner)
@@ -210,7 +234,9 @@
 func (m *objectBoundAccessManagerImpl[T]) gcCache(ctx context.Context, owner T) error {
 	log := logr.FromContextOrDiscard(ctx)
 
-	entry, ok := m.accessors[owner.GetUID()]
+	key := toAccessManagerKey(owner)
+
+	entry, ok := m.accessors[key]
 	if !ok {
 		return nil
 	}
@@ -219,7 +245,7 @@
 		// no users left -> close
 		log.Info("no users left, closing cache")
 		entry.cancel()
-		delete(m.accessors, owner.GetUID())
+		delete(m.accessors, key)
 
 		return nil
 	}
@@ -236,18 +262,22 @@
 	ctx context.Context, req accessorRequest[T],
 	doneCh chan<- cacheDone, wg *sync.WaitGroup,
 ) (Accessor, error) {
+	m.accessorsLock.Lock()
+	defer m.accessorsLock.Unlock()
+
 	log := logr.FromContextOrDiscard(ctx)
 	log = log.WithValues(
 		"ownerUID", req.owner.GetUID(),
 	)
 	ctx = logr.NewContext(ctx, log)
+	key := toAccessManagerKey(req.owner)
 
-	entry, ok := m.accessors[req.owner.GetUID()]
+	entry, ok := m.accessors[key]
 	if ok {
 		log.V(-1).Info("reusing cache for owner")
 
 		if req.user != nil {
-			entry.users[req.user.GetUID()] = req.gvks
+			entry.users[toAccessManagerKey(req.user)] = req.gvks
 		}
 
 		return entry.accessor, m.gcCache(ctx, req.owner)
@@ -281,25 +311,25 @@
 
 	entry = accessorEntry{
 		accessor: a,
-		users:    map[types.UID]sets.Set[schema.GroupVersionKind]{},
+		users:    map[accessManagerKey]sets.Set[schema.GroupVersionKind]{},
 		cancel:   cancel,
 	}
 	if req.user != nil {
-		entry.users[req.user.GetUID()] = req.gvks
+		entry.users[toAccessManagerKey(req.user)] = req.gvks
 		log = log.WithValues(
 			"userUID", req.user.GetUID(),
 			"usedForGVKs", req.gvks.UnsortedList(),
 		)
 	}
 
-	m.accessors[req.owner.GetUID()] = entry
+	m.accessors[key] = entry
 
 	log.V(-1).Info("starting new cache")
 	wg.Add(1)
 
 	go func(ctx context.Context, doneCh chan<- cacheDone) {
 		defer wg.Done()
-		doneCh <- cacheDone{uid: req.owner.GetUID(), err: ctrlcache.Start(ctx)}
+		doneCh <- cacheDone{key: key, err: ctrlcache.Start(ctx)}
 	}(ctx, doneCh)
 
 	return a, nil
@@ -381,3 +411,29 @@
 
 	return err
 }
+
+func (m *objectBoundAccessManagerImpl[T]) getWatchersForGVK(gvk schema.GroupVersionKind) (out []accessManagerKey) {
+	m.accessorsLock.RLock()
+	defer m.accessorsLock.RUnlock()
+
+	for k, a := range m.accessors {
+		if !sets.New(a.accessor.GetGVKs()...).Has(gvk) {
+			continue
+		}
+
+		out = append(out, k)
+		for u := range a.users {
+			out = append(out, u)
+		}
+	}
+
+	return out
+}
+
+func toAccessManagerKey[T RefType](owner T) accessManagerKey {
+	return accessManagerKey{
+		UID:              owner.GetUID(),
+		ObjectKey:        client.ObjectKeyFromObject(owner),
+		GroupVersionKind: owner.GetObjectKind().GroupVersionKind(),
+	}
+}

managedcache/trackingcache.go

@@ -32,6 +32,9 @@
 
 	// RemoveOtherInformers stops all informers that are not needed to watch the given list of object types.
 	RemoveOtherInformers(ctx context.Context, gvks ...schema.GroupVersionKind) error
+
+	// GetGVKs returns a list of GVKs known by this trackingCache.
+	GetGVKs() []schema.GroupVersionKind
 }
 
 type cacheSourcer interface {
@@ -112,6 +115,14 @@
 	return wehc, nil
 }
 
+// GetGVKs returns a list of GVKs known by this trackingCache.
+func (c *trackingCache) GetGVKs() []schema.GroupVersionKind {
+	c.accessLock.RLock()
+	defer c.accessLock.RUnlock()
+
+	return c.knownInformers.UnsortedList()
+}
+
 func (c *trackingCache) Source(handler handler.EventHandler, predicates ...predicate.Predicate) source.Source {
 	return c.cacheSourcer.Source(handler, predicates...)
 }