Cancel health checks on new reconciliation request

Signed-off-by: Matheus Pimenta <[email protected]>

Author: matheuscscp

go.mod

@@ -28,7 +28,7 @@ require (
 	github.com/fluxcd/pkg/cache v0.12.0
 	github.com/fluxcd/pkg/http/fetch v0.20.0
 	github.com/fluxcd/pkg/kustomize v1.23.0
-	github.com/fluxcd/pkg/runtime v0.88.0
+	github.com/fluxcd/pkg/runtime v0.89.1-0.20251010110127-c6bc6a65ae30
 	github.com/fluxcd/pkg/ssa v0.60.0
 	github.com/fluxcd/pkg/tar v0.15.0
 	github.com/fluxcd/pkg/testserver v0.13.0

go.sum

@@ -206,8 +206,8 @@ github.com/fluxcd/pkg/http/fetch v0.20.0 h1:/Lvcu1JzABBLuQYuLKYh1K02a+RqbP4b5wIZ
 github.com/fluxcd/pkg/http/fetch v0.20.0/go.mod h1:9inwDiGOpuo14Rp06TpcgsYSkvp4YM+uWCjgDmpXMNk=
 github.com/fluxcd/pkg/kustomize v1.23.0 h1:4tNh30OsIj96YRfVP7qP0Fv3QTwdBo/udfZIcccL6NI=
 github.com/fluxcd/pkg/kustomize v1.23.0/go.mod h1:ZojUvmI4RiHk3BH3L3mBQ4ZDbNkiWfX9LvOMBjKq5Tc=
-github.com/fluxcd/pkg/runtime v0.88.0 h1:EFPJ0jnRino6yUEwiNtQTpUNyCf96N2MJb+S7LVG648=
-github.com/fluxcd/pkg/runtime v0.88.0/go.mod h1:qkmPX009tgiWufQ2Vj0QhyNgEU+0Cnz7Xy/naihLM10=
+github.com/fluxcd/pkg/runtime v0.89.1-0.20251010110127-c6bc6a65ae30 h1:iy8cdKPpD9Iaxj30jUc5A/zeO774weMnPvA725yEnq0=
+github.com/fluxcd/pkg/runtime v0.89.1-0.20251010110127-c6bc6a65ae30/go.mod h1:qkmPX009tgiWufQ2Vj0QhyNgEU+0Cnz7Xy/naihLM10=
 github.com/fluxcd/pkg/sourceignore v0.15.0 h1:tB30fuk4jlB3UGlR7ppJguZ3zaJh1iwuTCEufs91jSM=
 github.com/fluxcd/pkg/sourceignore v0.15.0/go.mod h1:mZ9X6gNtNkq9ZsD35LebEYjePc7DRvB2JdowMNoj6IU=
 github.com/fluxcd/pkg/ssa v0.60.0 h1:ikA78TWSLDmIc8I/goGAU/buYF6jto/gswE5hnOfWGk=

internal/controller/kustomization_controller.go

@@ -985,42 +985,20 @@ func (r *KustomizationReconciler) checkHealth(ctx context.Context,
 
 	// Check the health with a default timeout of 30sec shorter than the reconciliation interval.
 	healthCtx := ctx
-	if r.CancelHealthCheckOnNewRevision {
-		// Create a cancellable context for health checks that monitors for new revisions
-		var cancel context.CancelFunc
-		healthCtx, cancel = context.WithCancel(ctx)
-		defer cancel()
-
-		// Start monitoring for new revisions to allow early cancellation
-		go func() {
-			ticker := time.NewTicker(5 * time.Second)
-			defer ticker.Stop()
-
-			for {
-				select {
-				case <-healthCtx.Done():
-					return
-				case <-ticker.C:
-					// Get the latest source artifact
-					latestSrc, err := r.getSource(ctx, obj)
-					if err == nil && latestSrc.GetArtifact() != nil {
-						if newRevision := latestSrc.GetArtifact().Revision; newRevision != revision {
-							const msg = "New revision detected during health check, cancelling"
-							r.event(obj, revision, originRevision, eventv1.EventSeverityInfo, msg, nil)
-							ctrl.LoggerFrom(ctx).Info(msg, "current", revision, "new", newRevision)
-							cancel()
-							return
-						}
-					}
-				}
-			}
-		}()
+	cancelHealthCheckOnEnqueue := r.CancelHealthCheckOnNewRevision // we could OR this with a boolean in the object spec
+	if cancelHealthCheckOnEnqueue {
+		healthCtx = runtimeCtrl.GetObjectContext(ctx)
 	}
 	if err := manager.WaitForSetWithContext(healthCtx, toCheck, ssa.WaitOptions{
 		Interval: 5 * time.Second,
 		Timeout:  obj.GetTimeout(),
 		FailFast: r.FailFast,
 	}); err != nil {
+		if cancelHealthCheckOnEnqueue && runtimeCtrl.IsObjectEnqueued(ctx) {
+			const msg = "New reconciliation request detected during health check, cancelling"
+			r.event(obj, revision, originRevision, eventv1.EventSeverityInfo, msg, nil)
+			ctrl.LoggerFrom(ctx).Info(msg)
+		}
 		conditions.MarkFalse(obj, meta.ReadyCondition, meta.HealthCheckFailedReason, "%s", err)
 		conditions.MarkFalse(obj, meta.HealthyCondition, meta.HealthCheckFailedReason, "%s", err)
 		return fmt.Errorf("health check failed after %s: %w", time.Since(checkStart).String(), err)

internal/controller/kustomization_manager.go

@@ -26,10 +26,10 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/builder"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
-	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
+	runtimeCtrl "github.com/fluxcd/pkg/runtime/controller"
 	"github.com/fluxcd/pkg/runtime/predicates"
 	sourcev1 "github.com/fluxcd/source-controller/api/v1"
 
@@ -129,43 +129,45 @@ func (r *KustomizationReconciler) SetupWithManager(ctx context.Context, mgr ctrl
 		return fmt.Errorf("failed creating index %s: %w", indexSecret, err)
 	}
 
-	ctrlBuilder := ctrl.NewControllerManagedBy(mgr).
-		For(&kustomizev1.Kustomization{}, builder.WithPredicates(
+	wr := runtimeCtrl.WrapReconciler(r)
+
+	ctrlBuilder := runtimeCtrl.NewControllerManagedBy(mgr, wr).
+		For(&kustomizev1.Kustomization{},
 			predicate.Or(predicate.GenerationChangedPredicate{}, predicates.ReconcileRequestedPredicate{}),
-		)).
+		).
 		Watches(
 			&sourcev1.OCIRepository{},
-			handler.EnqueueRequestsFromMapFunc(r.requestsForRevisionChangeOf(indexOCIRepository)),
+			wr.EnqueueRequestsFromMapFunc(r.requestsForRevisionChangeOf(indexOCIRepository)),
 			builder.WithPredicates(SourceRevisionChangePredicate{}),
 		).
 		Watches(
 			&sourcev1.GitRepository{},
-			handler.EnqueueRequestsFromMapFunc(r.requestsForRevisionChangeOf(indexGitRepository)),
+			wr.EnqueueRequestsFromMapFunc(r.requestsForRevisionChangeOf(indexGitRepository)),
 			builder.WithPredicates(SourceRevisionChangePredicate{}),
 		).
 		Watches(
 			&sourcev1.Bucket{},
-			handler.EnqueueRequestsFromMapFunc(r.requestsForRevisionChangeOf(indexBucket)),
+			wr.EnqueueRequestsFromMapFunc(r.requestsForRevisionChangeOf(indexBucket)),
 			builder.WithPredicates(SourceRevisionChangePredicate{}),
 		).
 		WatchesMetadata(
 			&corev1.ConfigMap{},
-			handler.EnqueueRequestsFromMapFunc(r.requestsForConfigDependency(indexConfigMap)),
+			wr.EnqueueRequestsFromMapFunc(r.requestsForConfigDependency(indexConfigMap)),
 			builder.WithPredicates(predicate.ResourceVersionChangedPredicate{}, opts.WatchConfigsPredicate),
 		).
 		WatchesMetadata(
 			&corev1.Secret{},
-			handler.EnqueueRequestsFromMapFunc(r.requestsForConfigDependency(indexSecret)),
+			wr.EnqueueRequestsFromMapFunc(r.requestsForConfigDependency(indexSecret)),
 			builder.WithPredicates(predicate.ResourceVersionChangedPredicate{}, opts.WatchConfigsPredicate),
 		)
 
 	if opts.WatchExternalArtifacts {
 		ctrlBuilder = ctrlBuilder.Watches(
 			&sourcev1.ExternalArtifact{},
-			handler.EnqueueRequestsFromMapFunc(r.requestsForRevisionChangeOf(indexExternalArtifact)),
+			wr.EnqueueRequestsFromMapFunc(r.requestsForRevisionChangeOf(indexExternalArtifact)),
 			builder.WithPredicates(SourceRevisionChangePredicate{}),
 		)
 	}
 
-	return ctrlBuilder.WithOptions(controller.Options{RateLimiter: opts.RateLimiter}).Complete(r)
+	return ctrlBuilder.WithOptions(controller.Options{RateLimiter: opts.RateLimiter}).Complete(wr)
 }

internal/controller/kustomization_wait_test.go

@@ -621,7 +621,7 @@ spec:
 	events := getEvents(resultK.GetName(), nil)
 	var found bool
 	for _, e := range events {
-		if e.Message == "New revision detected during health check, cancelling" {
+		if e.Message == "New reconciliation request detected during health check, cancelling" {
 			found = true
 			break
 		}