Drive rollback, promote, timeout, and abort from the checkpoint contract

Rewrite processRollbackSignal, processPromoteSignal, abortExperiment, and
handleRollback to validate against the experiment's Checkpoint instead of
revision ordering or revision annotations. Add APIReader as an uncached,
direct-to-apiserver reader for ControllerRevision lookups so a stale cache
NotFound is never mistaken for a permanently-lost baseline, and enforce
ownership (namespace, label, controller-owner UID) on every read. Recover
the nil-phase rollback path so an annotation naming an unresolvable
ControllerRevision aborts as baseline_not_found instead of silently
no-op'ing, and abort in-flight experiments with no checkpoint as
baseline_missing. Short-circuit reconcileInstance after a rollback restores
the spec so stale-spec rendering doesn't run before the restore's watch
event triggers a fresh reconcile.

Environment: Datadog workspace

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

Author: levan-m

internal/controller/datadogagent/controller.go

@@ -7,6 +7,7 @@ package datadogagent
 
 import (
 	"context"
+	"fmt"
 	"time"
 
 	"github.com/go-logr/logr"
@@ -101,6 +102,10 @@ type ReconcilerOptions struct {
 	// ClusterProviderDetector supplies the detected cluster provider. Nil disables
 	// provider detection (reconcile behaves as before: empty provider).
 	ClusterProviderDetector ProviderReader
+	// APIReader is an uncached, direct-to-apiserver reader used to validate
+	// experiment rollback targets. A cached/informer-backed client's stale
+	// NotFound could be mistaken for a permanently-lost baseline revision.
+	APIReader client.Reader
 }
 
 // Reconciler is the internal reconciler for Datadog Agent
@@ -128,6 +133,10 @@ func (r *Reconciler) initializeComponentRegistry() {
 func NewReconciler(options ReconcilerOptions, client client.Client, platformInfo kubernetes.PlatformInfo,
 	scheme *runtime.Scheme, log logr.Logger, recorder record.EventRecorder, metricForwardersMgr datadog.MetricsForwardersManager,
 ) (*Reconciler, error) {
+	if options.APIReader == nil {
+		return nil, fmt.Errorf("ReconcilerOptions.APIReader must not be nil")
+	}
+
 	r := &Reconciler{
 		options:      options,
 		client:       client,

internal/controller/datadogagent/experiment.go

@@ -127,12 +127,20 @@ func (r *Reconciler) reconcileInstance(ctx context.Context, logger logr.Logger,
 		// Use user-submitted instance instead of defaulted instance
 		rawInstance := instance.DeepCopy()
 		rawInstance.Spec = rawSpec
-		experimentErr := r.manageExperiment(ctx, rawInstance, newDDAStatus, now, revList)
+		specUpdated, experimentErr := r.manageExperiment(ctx, rawInstance, newDDAStatus, now)
 		instance.ResourceVersion = rawInstance.ResourceVersion
 		if experimentErr != nil {
 			return r.updateStatusIfNeeded(logger, instance, newDDAStatus, result, experimentErr, now)
 		}
 		syncExperimentConfigStrandedCondition(newDDAStatus, now)
+		if specUpdated {
+			// A rollback restored the DDA spec via Update. instance still holds
+			// the pre-rollback defaulted spec, so rendering DDAI/dependencies
+			// from it below would apply stale config. Publish the terminal
+			// experiment status now and let the Update's watch event trigger a
+			// fresh reconcile against the restored spec.
+			return r.updateStatusIfNeeded(logger, instance, newDDAStatus, result, nil, now)
+		}
 		if err := r.manageRevision(ctx, instance, rawSpec, revList, newDDAStatus); err != nil {
 			return r.updateStatusIfNeeded(logger, instance, newDDAStatus, result, err, now)
 		}

internal/controller/datadogagent/experiment_integration_test.go

@@ -15,7 +15,6 @@ import (
 	appsv1 "k8s.io/api/apps/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/runtime"
-	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -40,19 +39,6 @@ func buildRevisionSnapshot(spec v2alpha1.DatadogAgentSpec, allAnnotations map[st
 	return json.Marshal(snap)
 }
 
-// skipRevisionBump returns true when the revision bump should be suppressed.
-// During experiment rollback the spec is restored to an older revision; bumping
-// its revision number to "latest" would make it appear newer than the experiment
-// revision, causing findRollbackTarget to return the experiment revision on the
-// next rollback attempt instead of the pre-experiment revision.
-func skipRevisionBump(newStatus *v2alpha1.DatadogAgentStatus) bool {
-	if newStatus == nil || newStatus.Experiment == nil {
-		return false
-	}
-	phase := newStatus.Experiment.Phase
-	return phase == v2alpha1.ExperimentPhaseTerminated
-}
-
 // manageRevision creates a ControllerRevision snapshot of the current spec and
 // garbage collects old revisions. Must be called after manageExperiment.
 //
@@ -61,16 +47,54 @@ func skipRevisionBump(newStatus *v2alpha1.DatadogAgentStatus) bool {
 // still used for labels, annotations, and object identity, which are
 // unaffected by defaulting.
 func (r *Reconciler) manageRevision(ctx context.Context, instance *v2alpha1.DatadogAgent, rawSpec v2alpha1.DatadogAgentSpec, revList []appsv1.ControllerRevision, newStatus *v2alpha1.DatadogAgentStatus) error {
-	revName, err := r.ensureRevision(ctx, instance, rawSpec, revList, skipRevisionBump(newStatus))
+	revName, err := r.ensureRevision(ctx, instance, rawSpec, revList)
 	if err != nil {
 		return err
 	}
-	if err := r.gcOldRevisions(ctx, map[string]bool{revName: true}, revList); err != nil {
+
+	pins := map[string]bool{revName: true}
+	var checkpoint *v2alpha1.ExperimentCheckpoint
+	var phase v2alpha1.ExperimentPhase
+	if newStatus != nil && newStatus.Experiment != nil {
+		checkpoint = newStatus.Experiment.Checkpoint
+		phase = newStatus.Experiment.Phase
+	}
+	if checkpoint != nil {
+		if !isTerminalPhase(phase) {
+			pins[checkpoint.RollbackTargetRevision] = true
+		}
+	} else if pending := instance.GetAnnotations()[v2alpha1.AnnotationExperimentRollbackTargetRevision]; pending != "" {
+		// A start or rollback signal is pending and status has not copied
+		// the checkpoint yet; keep the nominated baseline alive so it isn't
+		// GC'd out from under the in-flight signal.
+		pins[pending] = true
+	}
+
+	if err := r.gcOldRevisions(ctx, pins, revList); err != nil {
 		ctrl.LoggerFrom(ctx).Error(err, "Failed to garbage collect old ControllerRevisions, will retry on next reconcile")
 	}
 	return nil
 }
 
+// ownedByDDA reports whether rev is a ControllerRevision owned by dda: same
+// namespace, the agent-name label matches, and a controller owner reference
+// points at dda's UID. A revision left behind by a deleted-and-recreated DDA
+// (same name, new UID) or a foreign/spoofed object fails this check.
+func ownedByDDA(rev *appsv1.ControllerRevision, dda *v2alpha1.DatadogAgent) bool {
+	if rev.Namespace != dda.GetNamespace() {
+		return false
+	}
+	if rev.Labels[apicommon.DatadogAgentNameLabelKey] != dda.GetName() {
+		return false
+	}
+	for _, ref := range rev.OwnerReferences {
+		if ref.Controller != nil && *ref.Controller && ref.UID == dda.GetUID() {
+			return true
+		}
+	}
+	return false
+}
+
 // publishCurrentRevisionBarrier ensures a ControllerRevision exists for the
 // current raw spec plus Datadog-owned annotations, then durably publishes the
 // resulting pointer to status.currentRevision (plus its freshness fields) via
@@ -88,7 +112,7 @@ func (r *Reconciler) publishCurrentRevisionBarrier(
 	rawSpec v2alpha1.DatadogAgentSpec,
 	revList []appsv1.ControllerRevision,
 ) (revName string, annotationsHash string, err error) {
-	revName, err = r.ensureRevision(ctx, instance, rawSpec, revList, false)
+	revName, err = r.ensureRevision(ctx, instance, rawSpec, revList)
 	if err != nil {
 		return "", "", err
 	}
@@ -144,11 +168,8 @@ func (r *Reconciler) listRevisions(ctx context.Context, instance *v2alpha1.Datad
 	// mistaken for the current owner's history.
 	owned := revList.Items[:0]
 	for i := range revList.Items {
-		for _, ref := range revList.Items[i].OwnerReferences {
-			if ref.Controller != nil && *ref.Controller && ref.UID == instance.GetUID() {
-				owned = append(owned, revList.Items[i])
-				break
-			}
+		if ownedByDDA(&revList.Items[i], instance) {
+			owned = append(owned, revList.Items[i])
 		}
 	}
 	revList.Items = owned
@@ -161,14 +182,12 @@ func (r *Reconciler) listRevisions(ctx context.Context, instance *v2alpha1.Datad
 // rawSpec (not instance.Spec, which may carry in-memory defaults) is what
 // gets stored, so that revisions reflect only user-intended changes.
 //
-// The Revision field is a monotonic creation counter. If skipBump is true the
-// existing revision is returned as-is without bumping its Revision number.
+// The Revision field is a monotonic creation counter.
 func (r *Reconciler) ensureRevision(
 	ctx context.Context,
 	instance *v2alpha1.DatadogAgent,
 	rawSpec v2alpha1.DatadogAgentSpec,
 	revList []appsv1.ControllerRevision,
-	skipBump bool,
 ) (string, error) {
 	logger := ctrl.LoggerFrom(ctx)
 
@@ -219,17 +238,9 @@ func (r *Reconciler) ensureRevision(
 			"object.name", matchingRev.Name,
 		)
 
-		if revisionExperimentState(matchingRev) == experimentRevisionStateRolledBack && !skipBump {
-			return r.recreateRevision(ctx, matchingRev, instance, gvks[0], labels, data, maxRevision)
-		}
-
 		// Identical content already snapshotted. Bump Revision to max+1 if it
 		// has been superseded (e.g. after a revert) so ordering stays correct.
-		// Skip the bump during experiment rollback: bumping the pre-experiment
-		// revision above the experiment revision would cause findRollbackTarget
-		// to select the experiment revision as the rollback target on the next
-		// stopped signal, reversing the rollback.
-		if matchingRev.Revision < maxRevision && !skipBump {
+		if matchingRev.Revision < maxRevision {
 			objLogger.Info("Bumping ControllerRevision to latest")
 			patch := fmt.Appendf(nil, `{"revision":%d}`, maxRevision+1)
 			if err := r.client.Patch(ctx, matchingRev, client.RawPatch(types.MergePatchType, patch)); err != nil && !apierrors.IsConflict(err) {
@@ -272,46 +283,6 @@ func (r *Reconciler) ensureRevision(
 	return rev.Name, nil
 }
 
-// recreateRevision deletes a rolled-back ControllerRevision and creates a
-// fresh one with the same content but a new CreationTimestamp. This prevents
-// an immediate timeout when the same experiment spec is re-applied, since
-// CreationTimestamp is immutable in Kubernetes.
-//
-// Failure recovery:
-//   - Delete fails: error returned, next reconcile retries.
-//   - Delete succeeds, Create fails (or operator crashes): the revision is
-//     gone, so the next reconcile's ensureRevision takes the normal "no
-//     matching revision" path and creates a fresh one.
-func (r *Reconciler) recreateRevision(
-	ctx context.Context,
-	old *appsv1.ControllerRevision,
-	instance *v2alpha1.DatadogAgent,
-	gvk schema.GroupVersionKind,
-	labels map[string]string,
-	data runtime.RawExtension,
-	maxRevision int64,
-) (string, error) {
-	logger := ctrl.LoggerFrom(ctx).WithValues(
-		"object.kind", "ControllerRevision",
-		"object.namespace", old.Namespace,
-		"object.name", old.Name,
-	)
-	logger.Info("Recreating rolled-back ControllerRevision with fresh timestamp")
-
-	if err := r.client.Delete(ctx, old); err != nil && !apierrors.IsNotFound(err) {
-		return "", fmt.Errorf("failed to delete rolled-back ControllerRevision %s: %w", old.Name, err)
-	}
-
-	fresh := controllerrevisions.NewControllerRevision(instance, gvk, labels, data, maxRevision+1, nil)
-	if err := r.client.Create(ctx, fresh); err != nil {
-		if apierrors.IsAlreadyExists(err) {
-			return fresh.Name, nil
-		}
-		return "", fmt.Errorf("failed to recreate ControllerRevision %s: %w", fresh.Name, err)
-	}
-	return fresh.Name, nil
-}
-
 // datadogAnnotations returns a copy of annotations filtered to only those
 // with `.datadoghq.com/` in the key, which are used for preview features.
 // Experiment signal annotations (experiment.datadoghq.com/) are excluded
@@ -332,16 +303,13 @@ func datadogAnnotations(all map[string]string) map[string]string {
 }
 
 // gcOldRevisions deletes all but the pinned revisions and the single most
-// recent unpinned one (kept as "previous"). Stale experiment revisions
-// (marked with the rollback annotation) are kept here — they are handled by
-// ensureRevision which recreates them with a fresh timestamp when the same
-// spec is re-applied.
+// recent unpinned one (kept as "previous").
 //
-// pins is a set of revision names that must never be deleted. Today it only
-// carries status.currentRevision; later commits add the active experiment
-// checkpoint's rollback target and a pending start signal's rollback-target
-// annotation, once those states exist, without needing to reshape this
-// signature again.
+// pins is a set of revision names that must never be deleted: the caller
+// includes status.currentRevision, the active experiment checkpoint's
+// rollback target (while non-terminal), and a pending start/rollback
+// signal's rollback-target annotation before status has copied the
+// checkpoint.
 func (r *Reconciler) gcOldRevisions(
 	ctx context.Context,
 	pins map[string]bool,