queue immediate reconciliation on kustomization dependency

Signed-off-by: Daniele Fognini <[email protected]>

Author: fogninid

internal/controller/kustomization_controller.go

@@ -44,6 +44,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
+	"sigs.k8s.io/controller-runtime/pkg/event"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
@@ -113,11 +114,40 @@ type KustomizationReconcilerOptions struct {
 	RateLimiter               workqueue.TypedRateLimiter[reconcile.Request]
 }
 
+type kustomizationReadyChangePredicate struct {
+	predicate.Funcs
+}
+
+func (kustomizationReadyChangePredicate) Update(e event.UpdateEvent) bool {
+	if e.ObjectNew == nil || e.ObjectOld == nil {
+		return false
+	}
+
+	newKs, ok := e.ObjectNew.(*kustomizev1.Kustomization)
+	if !ok {
+		return false
+	}
+	oldKs, ok := e.ObjectOld.(*kustomizev1.Kustomization)
+	if !ok {
+		return false
+	}
+
+	if !conditions.IsReady(newKs) {
+		return false
+	}
+	if !conditions.IsReady(oldKs) {
+		return true
+	}
+
+	return oldKs.Status.LastAppliedRevision != newKs.Status.LastAppliedRevision
+}
+
 func (r *KustomizationReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manager, opts KustomizationReconcilerOptions) error {
 	const (
 		ociRepositoryIndexKey string = ".metadata.ociRepository"
 		gitRepositoryIndexKey string = ".metadata.gitRepository"
 		bucketIndexKey        string = ".metadata.bucket"
+		ksDependencyIndexKey  string = ".metadata.dependsOn"
 	)
 
 	// Index the Kustomizations by the OCIRepository references they (may) point at.
@@ -138,6 +168,12 @@ func (r *KustomizationReconciler) SetupWithManager(ctx context.Context, mgr ctrl
 		return fmt.Errorf("failed setting index fields: %w", err)
 	}
 
+	// Index the Kustomizations by the dependsOn references they (may) point at.
+	if err := mgr.GetCache().IndexField(ctx, &kustomizev1.Kustomization{}, ksDependencyIndexKey,
+		r.indexDependsOn()); err != nil {
+		return fmt.Errorf("failed setting index fields: %w", err)
+	}
+
 	r.requeueDependency = opts.DependencyRequeueInterval
 	r.statusManager = fmt.Sprintf("gotk-%s", r.ControllerName)
 	r.artifactFetchRetries = opts.HTTPRetry
@@ -146,6 +182,11 @@ func (r *KustomizationReconciler) SetupWithManager(ctx context.Context, mgr ctrl
 		For(&kustomizev1.Kustomization{}, builder.WithPredicates(
 			predicate.Or(predicate.GenerationChangedPredicate{}, predicates.ReconcileRequestedPredicate{}),
 		)).
+		Watches(
+			&kustomizev1.Kustomization{},
+			handler.EnqueueRequestsFromMapFunc(r.requestsForDependentsWaiting(ksDependencyIndexKey)),
+			builder.WithPredicates(kustomizationReadyChangePredicate{}),
+		).
 		Watches(
 			&sourcev1b2.OCIRepository{},
 			handler.EnqueueRequestsFromMapFunc(r.requestsForRevisionChangeOf(ociRepositoryIndexKey)),
@@ -271,7 +312,7 @@ func (r *KustomizationReconciler) Reconcile(ctx context.Context, req ctrl.Reques
 	if len(obj.Spec.DependsOn) > 0 {
 		if err := r.checkDependencies(ctx, obj, artifactSource); err != nil {
 			conditions.MarkFalse(obj, meta.ReadyCondition, meta.DependencyNotReadyReason, "%s", err)
-			msg := fmt.Sprintf("Dependencies do not meet ready condition, retrying in %s", r.requeueDependency.String())
+			msg := fmt.Sprintf("Dependencies do not meet ready condition, retrying in %s: %s", r.requeueDependency.String(), err)
 			log.Info(msg)
 			r.event(obj, revision, originRevision, eventv1.EventSeverityInfo, msg, nil)
 			return ctrl.Result{RequeueAfter: r.requeueDependency}, nil

internal/controller/kustomization_indexers.go

@@ -21,11 +21,13 @@ import (
 	"fmt"
 
 	"github.com/fluxcd/pkg/runtime/conditions"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
+	"github.com/fluxcd/pkg/apis/meta"
 	"github.com/fluxcd/pkg/runtime/dependency"
 	sourcev1 "github.com/fluxcd/source-controller/api/v1"
 
@@ -78,6 +80,49 @@ func (r *KustomizationReconciler) requestsForRevisionChangeOf(indexKey string) h
 	}
 }
 
+func isNotReadyForDependency(k *kustomizev1.Kustomization) bool {
+	c := conditions.Get(k, meta.ReadyCondition)
+	if c == nil {
+		return false
+	}
+	return c.Status == metav1.ConditionFalse && c.Reason == meta.DependencyNotReadyReason
+}
+
+func (r *KustomizationReconciler) requestsForDependentsWaiting(indexKey string) handler.MapFunc {
+	return func(ctx context.Context, obj client.Object) []reconcile.Request {
+		log := ctrl.LoggerFrom(ctx)
+
+		var list kustomizev1.KustomizationList
+		if err := r.List(ctx, &list, client.MatchingFields{
+			indexKey: client.ObjectKeyFromObject(obj).String(),
+		}); err != nil {
+			log.Error(err, "failed to list objects for dependency change")
+			return nil
+		}
+		var dd []dependency.Dependent
+		for i, d := range list.Items {
+			if isNotReadyForDependency(&list.Items[i]) {
+				dd = append(dd, d.DeepCopy())
+			}
+		}
+		sorted, err := dependency.Sort(dd)
+		if err != nil {
+			log.Error(err, "failed to sort dependents for dependency change")
+			return nil
+		}
+		reqs := make([]reconcile.Request, len(sorted))
+		source := fmt.Sprintf("%s/%s", obj.GetNamespace(), obj.GetName())
+		for i := range sorted {
+			name := sorted[i].Name
+			namespace := sorted[i].Namespace
+			log.Info(fmt.Sprintf("request reconciliation of dependent of '%s': '%s/%s'", source, namespace, name))
+			reqs[i].NamespacedName.Name = name
+			reqs[i].NamespacedName.Namespace = namespace
+		}
+		return reqs
+	}
+}
+
 func (r *KustomizationReconciler) indexBy(kind string) func(o client.Object) []string {
 	return func(o client.Object) []string {
 		k, ok := o.(*kustomizev1.Kustomization)
@@ -96,3 +141,23 @@ func (r *KustomizationReconciler) indexBy(kind string) func(o client.Object) []s
 		return nil
 	}
 }
+
+func (r *KustomizationReconciler) indexDependsOn() func(o client.Object) []string {
+	return func(o client.Object) []string {
+		k, ok := o.(*kustomizev1.Kustomization)
+		if !ok {
+			panic(fmt.Sprintf("Expected a Kustomization, got %T", o))
+		}
+
+		deps := make([]string, len(k.Spec.DependsOn))
+		for i, dep := range k.Spec.DependsOn {
+			namespace := k.GetNamespace()
+			if dep.Namespace != "" {
+				namespace = dep.Namespace
+			}
+			deps[i] = fmt.Sprintf("%s/%s", namespace, dep.Name)
+		}
+
+		return deps
+	}
+}