feat: child object watch and resource reconciliation based on informers

cmd/controller/main.go

@@ -115,8 +115,9 @@ func main() {
 		"Burst size of events for the dynamic controller rate limiter.")
 
 	// reconciler parameters
-	flag.IntVar(&resyncPeriod, "dynamic-controller-default-resync-period", 36000,
-		"interval at which the controller will re list resources even with no changes, in seconds")
+	flag.IntVar(&resyncPeriod, "dynamic-controller-default-resync-period", 0,
+		"interval at which the controller will re list resources even with no changes, in seconds. "+
+			"By default resync is disabled.")
 	flag.IntVar(&queueMaxRetries, "dynamic-controller-default-queue-max-retries", 20,
 		"maximum number of retries for an item in the queue will be retried before being dropped")
 	// log level flags
@@ -186,7 +187,7 @@ func main() {
 		MaxRetryDelay:   maxRetryDelay,
 		RateLimit:       rateLimit,
 		BurstLimit:      burstLimit,
-	}, set.Dynamic())
+	}, set.Metadata(), set.RESTMapper())
 
 	resourceGraphDefinitionGraphBuilder, err := graph.NewBuilder(
 		restConfig, set.HTTPClient(),

config/crd/bases/kro.run_resourcegraphdefinitions.yaml

@@ -60,6 +60,25 @@ spec:
             description: ResourceGraphDefinitionSpec defines the desired state of
               ResourceGraphDefinition
             properties:
+              reconcile:
+                description: Reconcile defines the desired reconciliation settings
+                  for a ResourceGraphDefinition.
+                properties:
+                  instancePolicy:
+                    default: Reactive
+                    description: |-
+                      InstancePolicy defines the reconciliation policy for the instances
+                      created from the ResourceGraphDefinition.
+                      When set to "Periodic", or not set, the instances will be reconciled periodically.
+                      When set to "Reactive", the instances will be reconciled when any changes are detected
+                      on the resources owned by the instance. This change detection is achieved by watching
+                      the resources owned by the instance and enqueueing a reconcile request for the controller
+                      responsible for the instance.
+                    enum:
+                    - Periodic
+                    - Reactive
+                    type: string
+                type: object
               resources:
                 description: The resources that are part of the resourcegraphdefinition.
                 items:

helm/crds/kro.run_resourcegraphdefinitions.yaml

@@ -60,6 +60,25 @@ spec:
             description: ResourceGraphDefinitionSpec defines the desired state of
               ResourceGraphDefinition
             properties:
+              reconcile:
+                description: Reconcile defines the desired reconciliation settings
+                  for a ResourceGraphDefinition.
+                properties:
+                  instancePolicy:
+                    default: Reactive
+                    description: |-
+                      InstancePolicy defines the reconciliation policy for the instances
+                      created from the ResourceGraphDefinition.
+                      When set to "Periodic", or not set, the instances will be reconciled periodically.
+                      When set to "Reactive", the instances will be reconciled when any changes are detected
+                      on the resources owned by the instance. This change detection is achieved by watching
+                      the resources owned by the instance and enqueueing a reconcile request for the controller
+                      responsible for the instance.
+                    enum:
+                    - Periodic
+                    - Reactive
+                    type: string
+                type: object
               resources:
                 description: The resources that are part of the resourcegraphdefinition.
                 items:

pkg/client/set.go

@@ -22,6 +22,7 @@ import (
 	"k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/client-go/dynamic"
 	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/metadata"
 	"k8s.io/client-go/rest"
 	ctrlrtconfig "sigs.k8s.io/controller-runtime/pkg/client/config"
 	"sigs.k8s.io/release-utils/version"
@@ -58,6 +59,7 @@ type Set struct {
 	config          *rest.Config
 	kubernetes      *kubernetes.Clientset
 	dynamic         *dynamic.DynamicClient
+	metadata        metadata.Interface
 	apiExtensionsV1 *apiextensionsv1.ApiextensionsV1Client
 	// restMapper is a REST mapper for the Kubernetes API server
 	restMapper meta.RESTMapper
@@ -124,6 +126,11 @@ func (c *Set) init() error {
 		return err
 	}
 
+	c.metadata, err = metadata.NewForConfigAndClient(c.config, c.httpClient)
+	if err != nil {
+		return err
+	}
+
 	c.dynamic, err = dynamic.NewForConfigAndClient(c.config, c.httpClient)
 	if err != nil {
 		return err
@@ -146,6 +153,11 @@ func (c *Set) Kubernetes() kubernetes.Interface {
 	return c.kubernetes
 }
 
+// Metadata returns the metadata client
+func (c *Set) Metadata() metadata.Interface {
+	return c.metadata
+}
+
 // Dynamic returns the dynamic client
 func (c *Set) Dynamic() dynamic.Interface {
 	return c.dynamic

pkg/controller/instance/controller_reconcile.go

@@ -141,6 +141,25 @@ func (igr *instanceGraphReconciler) updateResourceReadiness(resourceID string) {
 	}
 }
 
+// areDependenciesReady checks if all dependencies of a resource are ready
+func (igr *instanceGraphReconciler) areDependenciesReady(resourceID string) bool {
+	dependencies := igr.runtime.ResourceDescriptor(resourceID).GetDependencies()
+
+	for _, depID := range dependencies {
+		// Check if dependency is resolved
+		if _, state := igr.runtime.GetResource(depID); state != runtime.ResourceStateResolved {
+			return false
+		}
+
+		// Check if dependency satisfies its readyWhen conditions
+		if ready, _, err := igr.runtime.IsResourceReady(depID); err != nil || !ready {
+			return false
+		}
+	}
+
+	return true
+}
+
 // reconcileInstance handles the reconciliation of an active instance
 func (igr *instanceGraphReconciler) reconcileInstance(ctx context.Context) error {
 	instance := igr.runtime.GetInstance()
@@ -195,6 +214,13 @@ func (igr *instanceGraphReconciler) reconcileInstance(ctx context.Context) error
 			break
 		}
 
+		// Check if all dependencies are ready
+		if !igr.areDependenciesReady(resourceID) {
+			unresolvedResourceID = resourceID
+			prune = false
+			break
+		}
+
 		applyable := applyset.ApplyableObject{
 			Unstructured: resource,
 			ID:           resourceID,

pkg/controller/resourcegraphdefinition/controller.go

@@ -155,7 +155,10 @@ func (r *ResourceGraphDefinitionReconciler) findRGDsForCRD(ctx context.Context,
 	}
 }
 
-func (r *ResourceGraphDefinitionReconciler) Reconcile(ctx context.Context, o *v1alpha1.ResourceGraphDefinition) (ctrl.Result, error) {
+func (r *ResourceGraphDefinitionReconciler) Reconcile(
+	ctx context.Context,
+	o *v1alpha1.ResourceGraphDefinition,
+) (ctrl.Result, error) {
 	if !o.DeletionTimestamp.IsZero() {
 		if err := r.cleanupResourceGraphDefinition(ctx, o); err != nil {
 			return ctrl.Result{}, err

pkg/controller/resourcegraphdefinition/controller_reconcile.go

@@ -17,6 +17,8 @@ package resourcegraphdefinition
 import (
 	"context"
 	"fmt"
+	"maps"
+	"slices"
 	"time"
 
 	v1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
@@ -25,7 +27,6 @@ import (
 
 	"github.com/kubernetes-sigs/kro/api/v1alpha1"
 	instancectrl "github.com/kubernetes-sigs/kro/pkg/controller/instance"
-	"github.com/kubernetes-sigs/kro/pkg/dynamiccontroller"
 	"github.com/kubernetes-sigs/kro/pkg/graph"
 	"github.com/kubernetes-sigs/kro/pkg/metadata"
 )
@@ -72,15 +73,10 @@ func (r *ResourceGraphDefinitionReconciler) reconcileResourceGraphDefinition(
 		mark.KindReady(crd.Status.AcceptedNames.Kind)
 	}
 
-	// Setup and start microcontroller
-	gvr := processedRGD.Instance.GetGroupVersionResource()
-	controller := r.setupMicroController(gvr, processedRGD, graphExecLabeler)
-
-	log.V(1).Info("reconciling resource graph definition micro controller")
 	// 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, &gvr, controller.Reconcile); err != nil {
+	if err := r.reconcileResourceGraphDefinitionMicroController(ctx, processedRGD, graphExecLabeler); err != nil {
 		mark.ControllerFailedToStart(err.Error())
 		return processedRGD.TopologicalOrder, resourcesInfo, err
 	}
@@ -89,6 +85,14 @@ func (r *ResourceGraphDefinitionReconciler) reconcileResourceGraphDefinition(
 	return processedRGD.TopologicalOrder, resourcesInfo, nil
 }
 
+func (r *ResourceGraphDefinitionReconciler) getResourceGVRsToWatchForRGD(processedRGD *graph.Graph) []schema.GroupVersionResource {
+	resourceHandlers := make(map[schema.GroupVersionResource]struct{}, len(processedRGD.Resources))
+	for _, resource := range processedRGD.Resources {
+		resourceHandlers[resource.GetGroupVersionResource()] = struct{}{}
+	}
+	return slices.Collect(maps.Keys(resourceHandlers))
+}
+
 // setupLabeler creates and merges the required labelers for the resource graph definition
 func (r *ResourceGraphDefinitionReconciler) setupLabeler(rgd *v1alpha1.ResourceGraphDefinition) (metadata.Labeler, error) {
 	rgLabeler := metadata.NewResourceGraphDefinitionLabeler(rgd)
@@ -97,10 +101,10 @@ func (r *ResourceGraphDefinitionReconciler) setupLabeler(rgd *v1alpha1.ResourceG
 
 // setupMicroController creates a new controller instance with the required configuration
 func (r *ResourceGraphDefinitionReconciler) setupMicroController(
-	gvr schema.GroupVersionResource,
 	processedRGD *graph.Graph,
 	labeler metadata.Labeler,
 ) *instancectrl.Controller {
+	gvr := processedRGD.Instance.GetGroupVersionResource()
 	instanceLogger := r.instanceLogger.WithName(fmt.Sprintf("%s-controller", gvr.Resource)).WithValues(
 		"controller", gvr.Resource,
 		"controllerGroup", processedRGD.Instance.GetCRD().Spec.Group,
@@ -162,8 +166,22 @@ func (r *ResourceGraphDefinitionReconciler) reconcileResourceGraphDefinitionCRD(
 }
 
 // reconcileResourceGraphDefinitionMicroController starts the microcontroller for handling the resources
-func (r *ResourceGraphDefinitionReconciler) reconcileResourceGraphDefinitionMicroController(ctx context.Context, gvr *schema.GroupVersionResource, handler dynamiccontroller.Handler) error {
-	err := r.dynamicController.Register(ctx, *gvr, handler)
+func (r *ResourceGraphDefinitionReconciler) reconcileResourceGraphDefinitionMicroController(
+	ctx context.Context,
+	processedRGD *graph.Graph,
+	graphExecLabeler metadata.Labeler,
+) error {
+	// If we want to react to changes to resources, we need to watch for them
+	// and trigger reconciliations of the instances whenever these resources change.
+	resourceGVRsToWatch := r.getResourceGVRsToWatchForRGD(processedRGD)
+
+	// Setup and start microcontroller
+	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 {
 		return newMicroControllerError(err)
 	}