🐛 fix: allow reconciliation of deadline-exceeded ClusterObjectSets

internal/operator-controller/controllers/clusterobjectset_controller.go

@@ -14,6 +14,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/go-logr/logr"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/equality"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
@@ -23,6 +24,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/client-go/util/workqueue"
 	"k8s.io/utils/clock"
 	"pkg.package-operator.run/boxcutter"
 	"pkg.package-operator.run/boxcutter/machinery"
@@ -32,8 +34,8 @@ import (
 	ctrl "sigs.k8s.io/controller-runtime"
 	"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/controller/controllerutil"
-	"sigs.k8s.io/controller-runtime/pkg/event"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/log"
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
@@ -83,29 +85,6 @@ func (c *ClusterObjectSetReconciler) Reconcile(ctx context.Context, req ctrl.Req
 	reconciledRev := existingRev.DeepCopy()
 	res, reconcileErr := c.reconcile(ctx, reconciledRev)
 
-	if pd := existingRev.Spec.ProgressDeadlineMinutes; pd > 0 {
- cnd := meta.FindStatusCondition(reconciledRev.Status.Conditions, ocv1.ClusterObjectSetTypeProgressing)
- isStillProgressing := cnd != nil && cnd.Status == metav1.ConditionTrue && cnd.Reason != ocv1.ReasonSucceeded
- succeeded := meta.IsStatusConditionTrue(reconciledRev.Status.Conditions, ocv1.ClusterObjectSetTypeSucceeded)
- // check if we reached the progress deadline only if the revision is still progressing and has not succeeded yet
- if isStillProgressing && !succeeded {
- timeout := time.Duration(pd) * time.Minute
- if c.Clock.Since(existingRev.CreationTimestamp.Time) > timeout {
- // progress deadline reached, reset any errors and stop reconciling this revision
- markAsNotProgressing(reconciledRev, ocv1.ReasonProgressDeadlineExceeded, fmt.Sprintf("Revision has not rolled out for %d minute(s).", pd))
- reconcileErr = nil
- res = ctrl.Result{}
- } else if reconcileErr == nil {
- // We want to requeue so far in the future that the next reconciliation
- // can detect if the revision did not progress within the given timeout.
- // Thus, we plan the next reconcile slightly after (+2secs) the timeout is passed.
- drift := 2 * time.Second
- requeueAfter := existingRev.CreationTimestamp.Time.Add(timeout).Add(drift).Sub(c.Clock.Now()).Round(time.Second)
- l.Info(fmt.Sprintf("ProgressDeadline not exceeded, requeue after ~%v to check again.", requeueAfter))
- res = ctrl.Result{RequeueAfter: requeueAfter}
- }
- }
-	}
 	// Do checks before any Update()s, as Update() may modify the resource structure!
 	updateStatus := !equality.Semantic.DeepEqual(existingRev.Status, reconciledRev.Status)
 
@@ -146,6 +125,10 @@ func (c *ClusterObjectSetReconciler) reconcile(ctx context.Context, cos *ocv1.Cl
  return c.delete(ctx, cos)
 	}
 
+	remaining, hasDeadline := durationUntilDeadline(c.Clock, cos)
+	isDeadlineExceeded := hasDeadline && remaining <= 0
+
+	// Blocked takes precedence over ProgressDeadlineExceeded: it is more actionable for the user.
 	if err := c.verifyReferencedSecretsImmutable(ctx, cos); err != nil {
  l.Error(err, "referenced Secret verification failed, blocking reconciliation")
  markAsNotProgressing(cos, ocv1.ClusterObjectSetReasonBlocked, err.Error())
@@ -154,7 +137,7 @@ func (c *ClusterObjectSetReconciler) reconcile(ctx context.Context, cos *ocv1.Cl
 
 	phases, currentPhases, opts, err := c.buildBoxcutterPhases(ctx, cos)
 	if err != nil {
- setRetryingConditions(cos, err.Error())
+ setRetryingConditions(l, cos, err.Error(), isDeadlineExceeded)
  return ctrl.Result{}, fmt.Errorf("converting to boxcutter revision: %v", err)
 	}
 
@@ -168,7 +151,7 @@ func (c *ClusterObjectSetReconciler) reconcile(ctx context.Context, cos *ocv1.Cl
 
 	revisionEngine, err := c.RevisionEngineFactory.CreateRevisionEngine(ctx, cos)
 	if err != nil {
- setRetryingConditions(cos, err.Error())
+ setRetryingConditions(l, cos, err.Error(), isDeadlineExceeded)
  return ctrl.Result{}, fmt.Errorf("failed to create revision engine: %v", err)
 	}
 
@@ -194,7 +177,7 @@ func (c *ClusterObjectSetReconciler) reconcile(ctx context.Context, cos *ocv1.Cl
 
 	if err := c.establishWatch(ctx, cos, revision); err != nil {
  werr := fmt.Errorf("establish watch: %v", err)
- setRetryingConditions(cos, werr.Error())
+ setRetryingConditions(l, cos, werr.Error(), isDeadlineExceeded)
  return ctrl.Result{}, werr
 	}
 
@@ -204,22 +187,22 @@ func (c *ClusterObjectSetReconciler) reconcile(ctx context.Context, cos *ocv1.Cl
  // Log detailed reconcile reports only in debug mode (V(1)) to reduce verbosity.
  l.V(1).Info("reconcile report", "report", rres.String())
  }
- setRetryingConditions(cos, err.Error())
+ setRetryingConditions(l, cos, err.Error(), isDeadlineExceeded)
  return ctrl.Result{}, fmt.Errorf("revision reconcile: %v", err)
 	}
 
 	// Retry failing preflight checks with a flat 10s retry.
 	// TODO: report status, backoff?
 	if verr := rres.GetValidationError(); verr != nil {
  l.Error(fmt.Errorf("%w", verr), "preflight validation failed, retrying after 10s")
- setRetryingConditions(cos, fmt.Sprintf("revision validation error: %s", verr))
+ setRetryingConditions(l, cos, fmt.Sprintf("revision validation error: %s", verr), isDeadlineExceeded)
  return ctrl.Result{RequeueAfter: 10 * time.Second}, nil
 	}
 
 	for i, pres := range rres.GetPhases() {
  if verr := pres.GetValidationError(); verr != nil {
  l.Error(fmt.Errorf("%w", verr), "phase preflight validation failed, retrying after 10s", "phase", i)
- setRetryingConditions(cos, fmt.Sprintf("phase %d validation error: %s", i, verr))
+ setRetryingConditions(l, cos, fmt.Sprintf("phase %d validation error: %s", i, verr), isDeadlineExceeded)
  return ctrl.Result{RequeueAfter: 10 * time.Second}, nil
  }
 
@@ -232,14 +215,14 @@ func (c *ClusterObjectSetReconciler) reconcile(ctx context.Context, cos *ocv1.Cl
 
  if len(collidingObjs) > 0 {
  l.Error(fmt.Errorf("object collision detected"), "object collision, retrying after 10s", "phase", i, "collisions", collidingObjs)
- setRetryingConditions(cos, fmt.Sprintf("revision object collisions in phase %d\n%s", i, strings.Join(collidingObjs, "\n\n")))
+ setRetryingConditions(l, cos, fmt.Sprintf("revision object collisions in phase %d\n%s", i, strings.Join(collidingObjs, "\n\n")), isDeadlineExceeded)
  return ctrl.Result{RequeueAfter: 10 * time.Second}, nil
  }
 	}
 
 	revVersion := cos.GetAnnotations()[labels.BundleVersionKey]
 	if rres.InTransition() {
- markAsProgressing(cos, ocv1.ReasonRollingOut, fmt.Sprintf("Revision %s is rolling out.", revVersion))
+ markAsProgressing(l, cos, ocv1.ReasonRollingOut, fmt.Sprintf("Revision %s is rolling out.", revVersion), isDeadlineExceeded)
 	}
 
 	//nolint:nestif
@@ -259,7 +242,7 @@ func (c *ClusterObjectSetReconciler) reconcile(ctx context.Context, cos *ocv1.Cl
  }
  }
 
- markAsProgressing(cos, ocv1.ReasonSucceeded, fmt.Sprintf("Revision %s has rolled out.", revVersion))
+ markAsProgressing(l, cos, ocv1.ReasonSucceeded, fmt.Sprintf("Revision %s has rolled out.", revVersion), isDeadlineExceeded)
  markAsAvailable(cos, ocv1.ClusterObjectSetReasonProbesSucceeded, "Objects are available and pass all probes.")
 
  // We'll probably only want to remove this once we are done updating the ClusterExtension conditions
@@ -304,8 +287,9 @@ func (c *ClusterObjectSetReconciler) reconcile(ctx context.Context, cos *ocv1.Cl
  } else {
  markAsUnavailable(cos, ocv1.ReasonRollingOut, fmt.Sprintf("Revision %s is rolling out.", revVersion))
  }
- if meta.FindStatusCondition(cos.Status.Conditions, ocv1.ClusterObjectSetTypeProgressing) == nil {
- markAsProgressing(cos, ocv1.ReasonRollingOut, fmt.Sprintf("Revision %s is rolling out.", revVersion))
+ markAsProgressing(l, cos, ocv1.ReasonRollingOut, fmt.Sprintf("Revision %s is rolling out.", revVersion), isDeadlineExceeded)
+ if hasDeadline && !isDeadlineExceeded {
+ return ctrl.Result{RequeueAfter: remaining}, nil
  }
 	}
 
@@ -324,14 +308,15 @@ func (c *ClusterObjectSetReconciler) delete(ctx context.Context, cos *ocv1.Clust
 }
 
 func (c *ClusterObjectSetReconciler) archive(ctx context.Context, revisionEngine RevisionEngine, cos *ocv1.ClusterObjectSet, revision boxcutter.RevisionBuilder) (ctrl.Result, error) {
+	l := log.FromContext(ctx)
 	tdres, err := revisionEngine.Teardown(ctx, revision)
 	if err != nil {
  err = fmt.Errorf("error archiving revision: %v", err)
- setRetryingConditions(cos, err.Error())
+ setRetryingConditions(l, cos, err.Error(), false)
  return ctrl.Result{}, err
 	}
 	if tdres != nil && !tdres.IsComplete() {
- setRetryingConditions(cos, "removing revision resources that are not owned by another revision")
+ setRetryingConditions(l, cos, "removing revision resources that are not owned by another revision", false)
  return ctrl.Result{RequeueAfter: 5 * time.Second}, nil
 	}
 	// Ensure conditions are set before removing the finalizer when archiving
@@ -349,29 +334,19 @@ type Sourcoser interface {
 }
 
 func (c *ClusterObjectSetReconciler) SetupWithManager(mgr ctrl.Manager) error {
-	skipProgressDeadlineExceededPredicate := predicate.Funcs{
- UpdateFunc: func(e event.UpdateEvent) bool {
- rev, ok := e.ObjectNew.(*ocv1.ClusterObjectSet)
- if !ok {
- return true
- }
- // allow deletions to happen
- if !rev.DeletionTimestamp.IsZero() {
- return true
- }
- if cnd := meta.FindStatusCondition(rev.Status.Conditions, ocv1.ClusterObjectSetTypeProgressing); cnd != nil && cnd.Status == metav1.ConditionFalse && cnd.Reason == ocv1.ReasonProgressDeadlineExceeded {
- return false
- }
- return true
- },
-	}
 	c.Clock = clock.RealClock{}
 	return ctrl.NewControllerManagedBy(mgr).
+ WithOptions(controller.Options{
+ RateLimiter: newDeadlineAwareRateLimiter(
+ workqueue.DefaultTypedControllerRateLimiter[ctrl.Request](),
+ mgr.GetClient(),
+ c.Clock,
+ ),
+ }).
  For(
  &ocv1.ClusterObjectSet{},
  builder.WithPredicates(
  predicate.ResourceVersionChangedPredicate{},
- skipProgressDeadlineExceededPredicate,
  ),
  ).
  WatchesRawSource(
@@ -659,14 +634,32 @@ func buildProgressionProbes(progressionProbes []ocv1.ProgressionProbe) (probing.
 	return userProbes, nil
 }
 
-func setRetryingConditions(cos *ocv1.ClusterObjectSet, message string) {
-	markAsProgressing(cos, ocv1.ClusterObjectSetReasonRetrying, message)
+func setRetryingConditions(l logr.Logger, cos *ocv1.ClusterObjectSet, message string, isDeadlineExceeded bool) {
+	markAsProgressing(l, cos, ocv1.ClusterObjectSetReasonRetrying, message, isDeadlineExceeded)
 	if meta.FindStatusCondition(cos.Status.Conditions, ocv1.ClusterObjectSetTypeAvailable) != nil {
  markAsAvailableUnknown(cos, ocv1.ClusterObjectSetReasonReconciling, message)
 	}
 }
 
-func markAsProgressing(cos *ocv1.ClusterObjectSet, reason, message string) {
+var nonTerminalProgressingReasons = map[string]struct{}{
+	ocv1.ReasonRollingOut: {},
+	ocv1.ClusterObjectSetReasonRetrying: {},
+}
+
+func markAsProgressing(l logr.Logger, cos *ocv1.ClusterObjectSet, reason, message string, isDeadlineExceeded bool) {
+	switch reason {
+	case ocv1.ReasonSucceeded:
+ // Terminal — always apply.
+	default:
+ if _, known := nonTerminalProgressingReasons[reason]; !known {
+ l.Error(fmt.Errorf("unregistered progressing reason: %q", reason), "treating as non-terminal for deadline enforcement")
+ }
+ if isDeadlineExceeded {
+ markAsNotProgressing(cos, ocv1.ReasonProgressDeadlineExceeded,
+ fmt.Sprintf("Revision has not rolled out for %d minute(s). Last status: %s", cos.Spec.ProgressDeadlineMinutes, message))
+ return
+ }
+	}
 	meta.SetStatusCondition(&cos.Status.Conditions, metav1.Condition{
  Type: ocv1.ClusterObjectSetTypeProgressing,
  Status: metav1.ConditionTrue,

internal/operator-controller/controllers/clusterobjectset_controller_test.go

@@ -988,7 +988,7 @@ func Test_ClusterObjectSetReconciler_Reconcile_ProgressDeadline(t *testing.T) {
  revisionResult: &mockRevisionResult{
  inTransition: true,
  },
- reconcileResult: ctrl.Result{RequeueAfter: 62 * time.Second},
+ reconcileResult: ctrl.Result{RequeueAfter: 60 * time.Second},
  validate: func(t *testing.T, c client.Client) {
  rev := &ocv1.ClusterObjectSet{}
  err := c.Get(t.Context(), client.ObjectKey{
@@ -1000,6 +1000,39 @@ func Test_ClusterObjectSetReconciler_Reconcile_ProgressDeadline(t *testing.T) {
  require.Equal(t, ocv1.ReasonRollingOut, cnd.Reason)
  },
  },
+ {
+ name: "recovery from ProgressDeadlineExceeded to Succeeded when revision completes",
+ existingObjs: func() []client.Object {
+ ext := newTestClusterExtension()
+ rev1 := newTestClusterObjectSet(t, clusterObjectSetName, ext, testScheme)
+ rev1.Spec.ProgressDeadlineMinutes = 1
+ rev1.CreationTimestamp = metav1.NewTime(time.Date(2022, 1, 1, 0, 0, 0, 0, time.UTC))
+ meta.SetStatusCondition(&rev1.Status.Conditions, metav1.Condition{
+ Type: ocv1.ClusterObjectSetTypeProgressing,
+ Status: metav1.ConditionFalse,
+ Reason: ocv1.ReasonProgressDeadlineExceeded,
+ Message: "Revision has not rolled out for 1 minute(s). Last status: Revision 1.0.0 is rolling out.",
+ ObservedGeneration: rev1.Generation,
+ })
+ return []client.Object{rev1, ext}
+ },
+ clock: clocktesting.NewFakeClock(time.Date(2022, 1, 1, 0, 5, 0, 0, time.UTC)),
+ revisionResult: &mockRevisionResult{
+ isComplete: true,
+ },
+ validate: func(t *testing.T, c client.Client) {
+ rev := &ocv1.ClusterObjectSet{}
+ err := c.Get(t.Context(), client.ObjectKey{
+ Name: clusterObjectSetName,
+ }, rev)
+ require.NoError(t, err)
+ cnd := meta.FindStatusCondition(rev.Status.Conditions, ocv1.ClusterObjectSetTypeProgressing)
+ require.NotNil(t, cnd)
+ require.Equal(t, metav1.ConditionTrue, cnd.Status)
+ require.Equal(t, ocv1.ReasonSucceeded, cnd.Reason)
+ require.Equal(t, "Revision 1.0.0 has rolled out.", cnd.Message)
+ },
+ },
  {
  name: "no progression deadline checks on revision recovery",
  existingObjs: func() []client.Object {