WIP: Implement asynchronous watch registration

Previously, when a ResourceGraphDefinition had issues (e.g., missing RBAC
permissions), the `Register()` method would block indefinitely on
`cache.WaitForCacheSync()` while holding a global lock, causing all other RGDs
to stall. This change makes registration non-blocking by moving the cache sync
wait into a goroutine and using channel-based notification when sync completes.

Author: a-hilaly

Makefile

@@ -269,6 +269,18 @@ deploy-kind: ko ## Deploy kro to a kind cluster
 	kubectl wait --for=condition=ready --timeout=1m pod -n kro-system -l app.kubernetes.io/component=controller
 	$(KUBECTL) --context kind-${KIND_CLUSTER_NAME} get pods -A
 
+.PHONY: deploy-kind-aggregation
+deploy-kind-aggregation: export KO_DOCKER_REPO=kind.local
+deploy-kind-aggregation: ko ## Deploy kro to a kind cluster with aggregation RBAC mode
+	$(KIND) delete clusters ${KIND_CLUSTER_NAME} || true
+	$(KIND) create cluster --name ${KIND_CLUSTER_NAME}
+	$(KUBECTL) --context kind-$(KIND_CLUSTER_NAME) create namespace kro-system
+	make install
+	# Deploy with aggregation RBAC mode - requires ClusterRoles with rbac.kro.run/aggregate-to-controller label
+	${HELM} template kro ./helm --namespace kro-system --set image.pullPolicy=Never --set image.ko=true --set config.allowCRDDeletion=true --set rbac.mode=aggregation | $(KO) apply -f -
+	kubectl wait --for=condition=ready --timeout=1m pod -n kro-system -l app.kubernetes.io/component=controller
+	$(KUBECTL) --context kind-${KIND_CLUSTER_NAME} get pods -A
+
 .PHONY: ko-apply
 ko-apply: ko
 	${HELM} template kro ./helm --namespace kro-system --set image.pullPolicy=Never --set image.ko=true | $(KO) apply -f -
@@ -288,4 +300,12 @@ test-e2e: chainsaw ## Run e2e tests
 
 .PHONY: test-e2e-kind
 test-e2e-kind: deploy-kind
-	make test-e2e
+	make test-e2e
+
+.PHONY: test-e2e-kind-aggregation
+test-e2e-kind-aggregation: deploy-kind-aggregation ## Run e2e tests with aggregation RBAC mode
+	make test-e2e
+
+.PHONY: test-e2e-non-blocking
+test-e2e-non-blocking: chainsaw ## Run the non-blocking RGD test (requires aggregation RBAC mode)
+	$(CHAINSAW) test ./test/e2e/chainsaw/check-non-blocking-rgd

api/v1alpha1/resourcegraphdefinition_types.go

@@ -179,6 +179,9 @@ const (
 	ResourceGraphDefinitionStateActive ResourceGraphDefinitionState = "Active"
 	// ResourceGraphDefinitionStateInactive represents the inactive state of the resource graph definition
 	ResourceGraphDefinitionStateInactive ResourceGraphDefinitionState = "Inactive"
+	// ResourceGraphDefinitionStateInitializing represents the initializing state where the controller
+	// is registered but waiting for the informer cache to sync.
+	ResourceGraphDefinitionStateInitializing ResourceGraphDefinitionState = "Initializing"
 )
 
 // ResourceGraphDefinitionStatus defines the observed state of ResourceGraphDefinition.

go.mod

@@ -17,7 +17,6 @@ require (
 	golang.org/x/exp v0.0.0-20240719175910-8a7402abbf56
 	golang.org/x/sync v0.12.0
 	golang.org/x/time v0.9.0
-	google.golang.org/genproto/googleapis/api v0.0.0-20250303144028-a0af3efb3deb
 	k8s.io/api v0.34.1
 	k8s.io/apiextensions-apiserver v0.31.0
 	k8s.io/apimachinery v0.34.1
@@ -92,6 +91,7 @@ require (
 	golang.org/x/text v0.23.0 // indirect
 	golang.org/x/tools v0.28.0 // indirect
 	gomodules.xyz/jsonpatch/v2 v2.4.0 // indirect
+	google.golang.org/genproto/googleapis/api v0.0.0-20250303144028-a0af3efb3deb // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20250303144028-a0af3efb3deb // indirect
 	google.golang.org/protobuf v1.36.5 // indirect
 	gopkg.in/evanphx/json-patch.v4 v4.12.0 // indirect

pkg/controller/resourcegraphdefinition/controller.go

@@ -17,10 +17,12 @@ package resourcegraphdefinition
 import (
 	"context"
 	"errors"
+	"fmt"
 
 	"github.com/go-logr/logr"
 	extv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	"k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/builder"
@@ -30,6 +32,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+	"sigs.k8s.io/controller-runtime/pkg/source"
 
 	"github.com/kubernetes-sigs/kro/api/v1alpha1"
 	kroclient "github.com/kubernetes-sigs/kro/pkg/client"
@@ -95,6 +98,10 @@ func (r *ResourceGraphDefinitionReconciler) SetupWithManager(mgr ctrl.Manager) e
 		return log
 	}
 
+	// Create a channel source to watch for sync events from the dynamic controller
+	syncEventsChan := make(chan event.GenericEvent)
+	go r.watchSyncEvents(syncEventsChan)
+
 	return ctrl.NewControllerManagedBy(mgr).
 		Named("ResourceGraphDefinition").
 		For(&v1alpha1.ResourceGraphDefinition{}).
@@ -120,9 +127,39 @@ func (r *ResourceGraphDefinitionReconciler) SetupWithManager(mgr ctrl.Manager) e
 				},
 			}),
 		).
+		WatchesRawSource(
+			source.Channel(syncEventsChan, &handler.EnqueueRequestForObject{}),
+		).
 		Complete(reconcile.AsReconciler[*v1alpha1.ResourceGraphDefinition](mgr.GetClient(), r))
 }
 
+// watchSyncEvents listens for sync events from the dynamic controller and
+// converts them to generic events that trigger RGD reconciliation.
+func (r *ResourceGraphDefinitionReconciler) watchSyncEvents(eventsChan chan<- event.GenericEvent) {
+	for gvr := range r.dynamicController.SyncEvents() {
+		crdName := fmt.Sprintf("%s.%s", gvr.Resource, gvr.Group)
+
+		crd := &metav1.PartialObjectMetadata{}
+		crd.SetGroupVersionKind(extv1.SchemeGroupVersion.WithKind("CustomResourceDefinition"))
+		if err := r.Client.Get(context.Background(), client.ObjectKey{Name: crdName}, crd); err != nil {
+			continue
+		}
+
+		if !metadata.IsKROOwned(crd) {
+			continue
+		}
+		rgdName, ok := crd.GetLabels()[metadata.ResourceGraphDefinitionNameLabel]
+		if !ok {
+			continue
+		}
+
+		rgd := &v1alpha1.ResourceGraphDefinition{}
+		rgd.SetName(rgdName)
+
+		eventsChan <- event.GenericEvent{Object: rgd}
+	}
+}
+
 // findRGDsForCRD returns a list of reconcile requests for the ResourceGraphDefinition
 // that owns the given CRD. It is used to trigger reconciliation when a CRD is updated.
 func (r *ResourceGraphDefinitionReconciler) findRGDsForCRD(ctx context.Context, obj client.Object) []reconcile.Request {

pkg/controller/resourcegraphdefinition/controller_reconcile.go

@@ -76,11 +76,18 @@ func (r *ResourceGraphDefinitionReconciler) reconcileResourceGraphDefinition(
 	// TODO: the context that is passed here is tied to the reconciliation of the rgd, we might need to make
 	// a new context with our own cancel function here to allow us to cleanly term the dynamic controller
 	// rather than have it ignore this context and use the background context.
-	if err := r.reconcileResourceGraphDefinitionMicroController(ctx, processedRGD, graphExecLabeler); err != nil {
+	gvr := processedRGD.Instance.GetGroupVersionResource()
+	if err := r.reconcileResourceGraphDefinitionMicroController(ctx, gvr, processedRGD, graphExecLabeler); err != nil {
 		mark.ControllerFailedToStart(err.Error())
 		return processedRGD.TopologicalOrder, resourcesInfo, err
 	}
-	mark.ControllerRunning()
+
+	// Set condition based on sync state
+	if r.dynamicController.HasSynced(gvr) {
+		mark.ControllerRunning()
+	} else {
+		mark.ControllerSyncing()
+	}
 
 	return processedRGD.TopologicalOrder, resourcesInfo, nil
 }
@@ -168,6 +175,7 @@ func (r *ResourceGraphDefinitionReconciler) reconcileResourceGraphDefinitionCRD(
 // reconcileResourceGraphDefinitionMicroController starts the microcontroller for handling the resources
 func (r *ResourceGraphDefinitionReconciler) reconcileResourceGraphDefinitionMicroController(
 	ctx context.Context,
+	gvr schema.GroupVersionResource,
 	processedRGD *graph.Graph,
 	graphExecLabeler metadata.Labeler,
 ) error {
@@ -179,7 +187,6 @@ func (r *ResourceGraphDefinitionReconciler) reconcileResourceGraphDefinitionMicr
 	controller := r.setupMicroController(processedRGD, graphExecLabeler)
 
 	ctrl.LoggerFrom(ctx).V(1).Info("reconciling resource graph definition micro controller")
-	gvr := processedRGD.Instance.GetGroupVersionResource()
 
 	err := r.dynamicController.Register(ctx, gvr, controller.Reconcile, resourceGVRsToWatch...)
 	if err != nil {

pkg/controller/resourcegraphdefinition/controller_status.go

@@ -40,9 +40,12 @@ func (r *ResourceGraphDefinitionReconciler) updateStatus(
 	log, _ := logr.FromContext(ctx)
 	log.V(1).Info("calculating resource graph definition status and conditions")
 
-	// Set status.state.
-	if rgdConditionTypes.For(o).IsRootReady() {
+	// Set status.state based on conditions.
+	cs := rgdConditionTypes.For(o)
+	if cs.IsRootReady() {
 		o.Status.State = v1alpha1.ResourceGraphDefinitionStateActive
+	} else if isControllerSyncing(o) {
+		o.Status.State = v1alpha1.ResourceGraphDefinitionStateInitializing
 	} else {
 		o.Status.State = v1alpha1.ResourceGraphDefinitionStateInactive
 	}
@@ -164,7 +167,22 @@ func (m *ConditionsMarker) ControllerFailedToStart(msg string) {
 	m.cs.SetFalse(ControllerReady, "FailedToStart", msg)
 }
 
+// ControllerSyncing signals the microcontroller is registered but waiting for informer cache to sync.
+func (m *ConditionsMarker) ControllerSyncing() {
+	m.cs.SetFalse(ControllerReady, "Syncing", "waiting for informer cache to sync")
+}
+
 // ControllerRunning signals the microcontroller is up and running for this RGD-Kind.
 func (m *ConditionsMarker) ControllerRunning() {
 	m.cs.SetTrueWithReason(ControllerReady, "Running", "controller is running")
 }
+
+// isControllerSyncing returns true if the ControllerReady condition is False with reason "Syncing".
+func isControllerSyncing(o *v1alpha1.ResourceGraphDefinition) bool {
+	for _, c := range o.Status.Conditions {
+		if string(c.Type) == ControllerReady {
+			return c.Reason != nil && *c.Reason == "Syncing"
+		}
+	}
+	return false
+}

pkg/dynamiccontroller/dynamic_controller.go

@@ -199,6 +199,10 @@ type DynamicController struct {
 	// queue is the work queue used to process items received via watches.
 	// The queue is shared between all informers and is used to propagate events to the handlers.
 	queue workqueue.TypedRateLimitingInterface[ObjectIdentifiers]
+
+	// syncedGVRs receives notifications when parent informers finish syncing.
+	// This channel is passed to parent LazyInformers and exposed via SyncEvents().
+	syncedGVRs chan schema.GroupVersionResource
 }
 
 // NewDynamicController creates a new DynamicController.
@@ -221,7 +225,26 @@ func NewDynamicController(
 			workqueue.NewTypedItemExponentialFailureRateLimiter[ObjectIdentifiers](config.MinRetryDelay, config.MaxRetryDelay),
 			&workqueue.TypedBucketRateLimiter[ObjectIdentifiers]{Limiter: rate.NewLimiter(rate.Limit(config.RateLimit), config.BurstLimit)},
 		), workqueue.TypedRateLimitingQueueConfig[ObjectIdentifiers]{Name: "dynamic-controller-queue"}),
+		syncedGVRs: make(chan schema.GroupVersionResource, 100),
+	}
+}
+
+// SyncEvents returns a channel that receives GVRs when their informers finish syncing.
+// Consumers should use this to watch for GVR sync completion.
+func (dc *DynamicController) SyncEvents() <-chan schema.GroupVersionResource {
+	return dc.syncedGVRs
+}
+
+// HasSynced returns true if the informer for the given GVR has synced.
+func (dc *DynamicController) HasSynced(gvr schema.GroupVersionResource) bool {
+	dc.mu.Lock()
+	defer dc.mu.Unlock()
+
+	w, exists := dc.watches[gvr]
+	if !exists {
+		return false
 	}
+	return w.HasSynced()
 }
 
 // Start starts workers and blocks until ctx.Done().
@@ -434,13 +457,14 @@ func (dc *DynamicController) Deregister(_ context.Context, parent schema.GroupVe
 
 func (dc *DynamicController) ensureWatchLocked(
 	gvr schema.GroupVersionResource,
+	syncedCh chan<- schema.GroupVersionResource,
 ) *internal.LazyInformer {
 	if w, ok := dc.watches[gvr]; ok {
 		return w
 	}
 
 	// Create per-GVR watch wrapper (informer created lazily on first handler)
-	w := internal.NewLazyInformer(dc.client, gvr, dc.config.ResyncPeriod, nil, dc.log)
+	w := internal.NewLazyInformer(dc.client, gvr, dc.config.ResyncPeriod, nil, dc.log, syncedCh)
 	dc.watches[gvr] = w
 	return w
 }
@@ -471,8 +495,8 @@ func (dc *DynamicController) reconcileParentLocked(
 		return nil
 	}
 
-	// ensure watch
-	w := dc.ensureWatchLocked(parent)
+	// ensure watch - parent gets sync notifications
+	w := dc.ensureWatchLocked(parent, dc.syncedGVRs)
 
 	// create handler if missing
 	if reg.parentHandlerID == "" {
@@ -533,7 +557,7 @@ func (dc *DynamicController) reconcileChildrenLocked(
 		if _, exists := reg.childHandlerIDs[child]; exists {
 			continue
 		}
-		w := dc.ensureWatchLocked(child)
+		w := dc.ensureWatchLocked(child, nil) // children don't need sync notifications
 		childHandlerID := childHandlerID(parent, child)
 		if err := w.AddHandler(dc.ctx, childHandlerID, dc.handlerForChildGVR(parent, child)); err != nil {
 			return fmt.Errorf("add child handler %s: %w", child, err)

pkg/dynamiccontroller/dynamic_controller_test.go

@@ -349,6 +349,11 @@ func TestInstanceUpdatePolicy(t *testing.T) {
 	err := dc.Register(t.Context(), gvr, handlerFunc)
 	assert.NoError(t, err)
 
+	// Wait for cache to sync (Register is now non-blocking)
+	require.Eventually(t, func() bool {
+		return dc.HasSynced(gvr)
+	}, 5*time.Second, 10*time.Millisecond, "cache should sync")
+
 	// simulate reconciling the instances
 	for dc.queue.Len() > 0 {
 		item, _ := dc.queue.Get()
@@ -361,7 +366,10 @@ func TestInstanceUpdatePolicy(t *testing.T) {
 	assert.NoError(t, err)
 
 	// check if the expected objects are queued
-	assert.Equal(t, dc.queue.Len(), 2)
+	// Note: On re-register, the existing 2 objects should be re-queued
+	require.Eventually(t, func() bool {
+		return dc.queue.Len() == 2
+	}, 5*time.Second, 10*time.Millisecond, "expected 2 items in queue after re-register")
 	for dc.queue.Len() > 0 {
 		name, _ := dc.queue.Get()
 		_, ok := objs[name.String()]

pkg/dynamiccontroller/internal/gvr_watch.go

@@ -44,6 +44,10 @@ type LazyInformer struct {
 	done   <-chan struct{}
 	cancel context.CancelFunc
 
+	// syncedCh receives notification when the informer finishes syncing.
+	// Only the GVR is sent - the receiver is responsible for any name mapping.
+	syncedCh chan<- schema.GroupVersionResource
+
 	log logr.Logger
 }
 
@@ -53,6 +57,7 @@ func NewLazyInformer(
 	resync time.Duration,
 	tweak metadatainformer.TweakListOptionsFunc,
 	logger logr.Logger,
+	syncedCh chan<- schema.GroupVersionResource,
 ) *LazyInformer {
 	li := &LazyInformer{
 		gvr:      gvr,
@@ -61,6 +66,7 @@ func NewLazyInformer(
 		tweak:    tweak,
 		handlers: make(map[string]cache.ResourceEventHandlerRegistration),
 		log:      logger.WithValues("gvr", gvr.String()),
+		syncedCh: syncedCh,
 	}
 	return li
 }
@@ -90,13 +96,14 @@ func (w *LazyInformer) ensureInformer() {
 }
 
 // AddHandler registers a new event handler and starts the informer if needed.
+// This method is non-blocking - it starts the informer but does not wait for cache sync.
+// When the informer syncs, the GVR is sent to syncedCh (if set at construction).
 func (w *LazyInformer) AddHandler(ctx context.Context, id string, h cache.ResourceEventHandler) error {
 	w.mu.Lock()
 	defer w.mu.Unlock()
 
 	// If informer was fully stopped, reset context before re-creating it.
 	if w.cancel == nil {
-		// recreate context
 		w.resetContext(ctx)
 	}
 
@@ -110,22 +117,37 @@ func (w *LazyInformer) AddHandler(ctx context.Context, id string, h cache.Resour
 
 	// Start informer if first handler
 	if len(w.handlers) == 1 {
-		go w.run()
-
-		if !cache.WaitForCacheSync(w.done, w.informer.HasSynced) {
-			w.log.Error(fmt.Errorf("cache sync failed"), "lazy informer sync failure", "gvr", w.gvr)
-			w.cancel()
-			w.cancel = nil
-			w.informer = nil
-			w.handlers = make(map[string]cache.ResourceEventHandlerRegistration)
-			return fmt.Errorf("failed to sync informer for %s", w.gvr)
-		}
+		// Capture values before releasing the lock to avoid races
+		informer := w.informer
+		done := w.done
+		syncedCh := w.syncedCh
+		gvr := w.gvr
+		log := w.log
+		go func() {
+			// Start the informer
+			go informer.Run(done)
+
+			// Wait for cache to sync and notify via channel
+			if cache.WaitForCacheSync(done, informer.HasSynced) {
+				if syncedCh != nil {
+					select {
+					case syncedCh <- gvr:
+					case <-done:
+					}
+				}
+			} else {
+				log.Error(fmt.Errorf("cache sync failed"), "lazy informer sync failure", "gvr", gvr)
+			}
+		}()
 	}
 	return nil
 }
 
-func (w *LazyInformer) run() {
-	w.informer.Run(w.done)
+// HasSynced returns true if the informer cache has synced.
+func (w *LazyInformer) HasSynced() bool {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	return w.informer != nil && w.informer.HasSynced()
 }
 
 // RemoveHandler unregisters a handler. Returns true if informer was stopped.