RolloutPlugin - 11 - Analysis Fix

Author: hariharasuthan1a

cmd/combined-controller/main.go

@@ -390,6 +390,7 @@ func newCommand() *cobra.Command {
 				DynamicClientset:  dynamicClient,
 				PluginManager:     pluginManager,
 				AnalysisHelper:    analysisHelper,
+				InstanceID:        instanceID,
 			}).SetupWithManager(mgr); err != nil {
 				log.Fatalf("Failed to setup RolloutPlugin controller: %s", err.Error())
 			}

cmd/rollouts-controller/main.go

@@ -194,15 +194,40 @@ func (r *RolloutPluginReconciler) reconcileStepBasedAnalysisRun(ctx context.Cont
 	}
 
 	// Check if we need to create a new analysis run
-	if currentAr == nil || analysisutil.IsTerminating(currentAr) {
+	// Only create if:
+	// 1. No current AR exists
+	// 2. Rollout is aborted (allows retry after fixes)
+	// 3. AR is inconclusive and rollout is paused for it
+	if currentAr == nil {
 		logger.Info("Creating step-based analysis run", "step", currentStepIndex)
-		newAr, err := r.createAnalysisRun(ctx, rp, currentStep.Analysis, fmt.Sprintf("%s-step-%d", rp.Name, currentStepIndex))
+		newAr, err := r.createAnalysisRun(ctx, rp, currentStep.Analysis, fmt.Sprintf("%s-step-%d", rp.Name, currentStepIndex), v1alpha1.RolloutTypeStepLabel)
 		if err != nil {
 			return nil, err
 		}
 		return newAr, nil
 	}
 
+	// If rollout is aborted, allow creating a new analysis run (for retry scenarios)
+	if rp.Status.Aborted {
+		logger.Info("Rollout is aborted, creating new step-based analysis run", "step", currentStepIndex)
+		newAr, err := r.createAnalysisRun(ctx, rp, currentStep.Analysis, fmt.Sprintf("%s-step-%d", rp.Name, currentStepIndex), v1alpha1.RolloutTypeStepLabel)
+		if err != nil {
+			return nil, err
+		}
+		return newAr, nil
+	}
+
+	// If AR is inconclusive and rollout is paused, allow creating a new one
+	if currentAr.Status.Phase == v1alpha1.AnalysisPhaseInconclusive && rp.Status.Paused {
+		logger.Info("Analysis is inconclusive and rollout is paused, creating new step-based analysis run", "step", currentStepIndex)
+		newAr, err := r.createAnalysisRun(ctx, rp, currentStep.Analysis, fmt.Sprintf("%s-step-%d", rp.Name, currentStepIndex), v1alpha1.RolloutTypeStepLabel)
+		if err != nil {
+			return nil, err
+		}
+		return newAr, nil
+	}
+
+	// Otherwise, keep the existing AR (even if failed/error - controller needs to see it)
 	return currentAr, nil
 }
 
@@ -215,16 +240,41 @@ func (r *RolloutPluginReconciler) reconcileBackgroundAnalysisRun(ctx context.Con
 	}
 
 	// Check if we need to create a new analysis run
-	if currentAr == nil || analysisutil.IsTerminating(currentAr) {
+	// Only create if:
+	// 1. No current AR exists
+	// 2. Rollout is aborted (allows retry after fixes)
+	// 3. AR is inconclusive and rollout is paused for it
+	if currentAr == nil {
 		logger.Info("Creating background analysis run")
 		// RolloutAnalysisBackground embeds RolloutAnalysis, so we can use it directly
-		newAr, err := r.createAnalysisRun(ctx, rp, &rp.Spec.Strategy.Canary.Analysis.RolloutAnalysis, fmt.Sprintf("%s-background", rp.Name))
+		newAr, err := r.createAnalysisRun(ctx, rp, &rp.Spec.Strategy.Canary.Analysis.RolloutAnalysis, fmt.Sprintf("%s-background", rp.Name), v1alpha1.RolloutTypeBackgroundRunLabel)
+		if err != nil {
+			return nil, err
+		}
+		return newAr, nil
+	}
+
+	// If rollout is aborted, allow creating a new analysis run (for retry scenarios)
+	if rp.Status.Aborted {
+		logger.Info("Rollout is aborted, creating new background analysis run")
+		newAr, err := r.createAnalysisRun(ctx, rp, &rp.Spec.Strategy.Canary.Analysis.RolloutAnalysis, fmt.Sprintf("%s-background", rp.Name), v1alpha1.RolloutTypeBackgroundRunLabel)
 		if err != nil {
 			return nil, err
 		}
 		return newAr, nil
 	}
 
+	// If AR is inconclusive and rollout is paused, allow creating a new one
+	if currentAr.Status.Phase == v1alpha1.AnalysisPhaseInconclusive && rp.Status.Paused {
+		logger.Info("Analysis is inconclusive and rollout is paused, creating new background analysis run")
+		newAr, err := r.createAnalysisRun(ctx, rp, &rp.Spec.Strategy.Canary.Analysis.RolloutAnalysis, fmt.Sprintf("%s-background", rp.Name), v1alpha1.RolloutTypeBackgroundRunLabel)
+		if err != nil {
+			return nil, err
+		}
+		return newAr, nil
+	}
+
+	// Otherwise, keep the existing AR (even if failed/error - controller needs to see it)
 	return currentAr, nil
 }
 
@@ -246,7 +296,7 @@ func convertAnalysisRunArgsToArguments(args []v1alpha1.AnalysisRunArgument) []v1
 }
 
 // createAnalysisRun creates a new AnalysisRun from the analysis spec
-func (r *RolloutPluginReconciler) createAnalysisRun(ctx context.Context, rp *v1alpha1.RolloutPlugin, analysisSpec *v1alpha1.RolloutAnalysis, namePrefix string) (*v1alpha1.AnalysisRun, error) {
+func (r *RolloutPluginReconciler) createAnalysisRun(ctx context.Context, rp *v1alpha1.RolloutPlugin, analysisSpec *v1alpha1.RolloutAnalysis, namePrefix string, rolloutType string) (*v1alpha1.AnalysisRun, error) {
 	logger := log.FromContext(ctx)
 
 	// Get templates
@@ -277,10 +327,15 @@ func (r *RolloutPluginReconciler) createAnalysisRun(ctx context.Context, rp *v1a
 
 	// Build labels
 	labels := map[string]string{
-		v1alpha1.RolloutTypeLabel: "RolloutPlugin",
+		v1alpha1.RolloutTypeLabel: rolloutType, // "Step" or "Background"
 		"rollout-plugin-name":     rp.Name,
 	}
 
+	// Add instance ID label if set (required for controller filtering)
+	if r.InstanceID != "" {
+		labels[v1alpha1.LabelKeyControllerInstanceID] = r.InstanceID
+	}
+
 	// Build annotations
 	annotations := map[string]string{}
 

rolloutplugin/analysis.go

@@ -38,6 +38,7 @@ type RolloutPluginReconciler struct {
 	DynamicClientset  dynamic.Interface
 	PluginManager     PluginManager
 	AnalysisHelper    AnalysisHelper
+	InstanceID        string
 }
 
 // AnalysisHelper is an interface for managing AnalysisRuns
@@ -522,6 +523,41 @@ func (r *RolloutPluginReconciler) processCanaryRollout(ctx context.Context, roll
 	currentStep := canary.Steps[currentStepIndex]
 	logCtx.WithField("stepIndex", currentStepIndex).Info("Processing canary step")
 
+	// Check background analysis status (if running)
+	// Background analysis runs throughout the rollout, so check it before processing each step
+	if newStatus.Canary.CurrentBackgroundAnalysisRunStatus != nil {
+		bgAnalysisStatus := newStatus.Canary.CurrentBackgroundAnalysisRunStatus.Status
+		logCtx.WithFields(log.Fields{
+			"status":      bgAnalysisStatus,
+			"analysisRun": newStatus.Canary.CurrentBackgroundAnalysisRunStatus.Name,
+		}).Info("Background analysis status")
+
+		switch bgAnalysisStatus {
+		case v1alpha1.AnalysisPhaseFailed, v1alpha1.AnalysisPhaseError:
+			// Background analysis failed, abort the rollout
+			logCtx.Error("Background analysis failed, aborting rollout")
+			newStatus.Phase = "Failed"
+			newStatus.Message = fmt.Sprintf("Background analysis failed: %s", newStatus.Canary.CurrentBackgroundAnalysisRunStatus.Name)
+			newStatus.RolloutInProgress = false
+			newStatus.Aborted = true
+			newStatus.AbortedRevision = newStatus.UpdatedRevision
+			if err := plugin.Abort(ctx, workloadRef); err != nil {
+				logCtx.WithError(err).Error("Failed to abort rollout")
+				return ctrl.Result{}, err
+			}
+			return ctrl.Result{}, nil
+
+		case v1alpha1.AnalysisPhaseInconclusive:
+			// Background analysis is inconclusive, pause the rollout
+			logCtx.Info("Background analysis is inconclusive, pausing rollout")
+			newStatus.Paused = true
+			newStatus.Message = "Paused: Background analysis inconclusive"
+			return ctrl.Result{}, nil
+		}
+		// For Running, Pending, Successful, or unknown status, continue processing the step
+		// Successful means the analysis completed successfully
+	}
+
 	// Process setWeight step
 	if currentStep.SetWeight != nil {
 		weight := *currentStep.SetWeight
@@ -625,6 +661,8 @@ func (r *RolloutPluginReconciler) processCanaryRollout(ctx context.Context, roll
 			newStatus.Phase = "Failed"
 			newStatus.Message = fmt.Sprintf("Analysis failed: %s", newStatus.Canary.CurrentStepAnalysisRunStatus.Name)
 			newStatus.RolloutInProgress = false
+			newStatus.Aborted = true
+			newStatus.AbortedRevision = newStatus.UpdatedRevision
 			if err := plugin.Abort(ctx, workloadRef); err != nil {
 				logCtx.WithError(err).Error("Failed to abort rollout")
 				return ctrl.Result{}, err
@@ -865,9 +903,6 @@ func (r *RolloutPluginReconciler) SetupWithManager(mgr ctrl.Manager) error {
 		DeleteFunc: func(e event.DeleteEvent) bool {
 			return true
 		},
-		GenericFunc: func(e event.GenericEvent) bool {
-			return true
-		},
 	}
 
 	// Create a predicate for RolloutPlugin that watches ALL updates (like Rollouts controller)
@@ -896,13 +931,36 @@ func (r *RolloutPluginReconciler) SetupWithManager(mgr ctrl.Manager) error {
 		DeleteFunc: func(e event.DeleteEvent) bool {
 			return true
 		},
-		GenericFunc: func(e event.GenericEvent) bool {
-			return true
+	}
+
+	// Predicate to filter AnalysisRun events - only trigger on phase changes
+	analysisRunPredicate := predicate.Funcs{
+		CreateFunc: func(e event.CreateEvent) bool {
+			return true // Always trigger on creation
+		},
+		UpdateFunc: func(e event.UpdateEvent) bool {
+			oldAR, ok1 := e.ObjectOld.(*v1alpha1.AnalysisRun)
+			newAR, ok2 := e.ObjectNew.(*v1alpha1.AnalysisRun)
+			if !ok1 || !ok2 {
+				return false
+			}
+
+			// Only trigger if phase changed (matches Rollouts controller behavior)
+			if oldAR.Status.Phase != newAR.Status.Phase {
+				return true
+			}
+
+			// Skip other updates
+			return false
+		},
+		DeleteFunc: func(e event.DeleteEvent) bool {
+			return true // Always trigger on deletion
 		},
 	}
 
 	return ctrl.NewControllerManagedBy(mgr).
 		For(&v1alpha1.RolloutPlugin{}).
+		Owns(&v1alpha1.AnalysisRun{}, builder.WithPredicates(analysisRunPredicate)).
 		Watches(
 			&appsv1.StatefulSet{},
 			handler.EnqueueRequestsFromMapFunc(r.findRolloutPluginsForWorkload),

rolloutplugin/controller.go

@@ -22,7 +22,7 @@ import (
 
 const (
 	// FieldManager is the field manager name used for Server-Side Apply
-	FieldManager = "argo-rollouts"
+	FieldManager = "argo-rollouts-statefulset-plugin"
 )
 
 // Plugin implements the ResourcePlugin interface directly for StatefulSets.
@@ -423,20 +423,22 @@ func (p *Plugin) Abort(ctx context.Context, workloadRef v1alpha1.WorkloadRef) er
 		return fmt.Errorf("failed to update StatefulSet during abort using SSA: %w", err)
 	}
 
-	// STEP 2: Delete pods that were updated (ordinals < oldPartition)
+	// STEP 2: Delete pods that were updated (ordinals >= oldPartition)
 	// StatefulSet controller will recreate them using CurrentRevision (old version)
-	// because partition=replicas means all pods should be on old version
-	// We delete all pods with ordinal < oldPartition to ensure all potentially
-	// updated pods are rolled back
+	// because partition=replicas means all pods should be on old version.
+	// With partition, pods with ordinal >= partition are on the NEW (updated) version.
+	// Pods with ordinal < partition are on the OLD (current) version.
+	// So we need to delete pods with ordinal >= oldPartition to roll them back.
+	podsToDelete := replicas - oldPartition
 	p.logCtx.WithFields(log.Fields{
 		"oldPartition": oldPartition,
-		"podsToDelete": oldPartition,
+		"podsToDelete": podsToDelete,
 	}).Info("Deleting updated pods to force rollback")
 
 	deletedCount := int32(0)
 	failedDeletes := []string{}
 
-	for i := int32(0); i < oldPartition; i++ {
+	for i := oldPartition; i < replicas; i++ {
 		podName := fmt.Sprintf("%s-%d", sts.Name, i)
 		pod := &corev1.Pod{
 			ObjectMeta: metav1.ObjectMeta{

rolloutplugin/plugins/statefulset/plugin.go

@@ -0,0 +1,25 @@
+apiVersion: argoproj.io/v1alpha1
+kind: Rollout
+metadata:
+  name: basic
+spec:
+  strategy:
+    canary: 
+      steps:
+      - setWeight: 50
+      - pause: {}
+  selector:
+    matchLabels:
+      app: basic
+  template:
+    metadata:
+      labels:
+        app: basic
+    spec:
+      containers:
+      - name: basic
+        image: quay.io/nginx:1.19-alpine
+        resources:
+          requests:
+            memory: 16Mi
+            cpu: 1m

test/e2e/rolloutplugin/rollout-basic.yaml

@@ -0,0 +1,84 @@
+---
+# Background analysis with failing metrics
+# Used by: TestBackgroundAnalysisFailure
+apiVersion: argoproj.io/v1alpha1
+kind: AnalysisTemplate
+metadata:
+  name: bg-analysis-fail
+spec:
+  metrics:
+  - name: fail-metric
+    count: 1
+    interval: 3s
+    failureLimit: 0
+    provider:
+      job:
+        spec:
+          backoffLimit: 0
+          template:
+            spec:
+              restartPolicy: Never
+              containers:
+              - name: fail
+                image: quay.io/prometheus/busybox:latest
+                command: ["/bin/sh", "-c", "echo 'Analysis failed' && exit 1"]
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: bg-analysis-fail-svc
+spec:
+  clusterIP: None
+  selector:
+    app: bg-analysis-fail
+  ports:
+  - port: 80
+    name: web
+---
+apiVersion: apps/v1
+kind: StatefulSet
+metadata:
+  name: bg-analysis-fail-sts
+spec:
+  serviceName: bg-analysis-fail-svc
+  replicas: 5
+  selector:
+    matchLabels:
+      app: bg-analysis-fail
+  template:
+    metadata:
+      labels:
+        app: bg-analysis-fail
+    spec:
+      terminationGracePeriodSeconds: 1
+      containers:
+      - name: busybox
+        image: quay.io/prometheus/busybox:latest
+        command: ["sleep", "3600"]
+        ports:
+        - containerPort: 80
+  updateStrategy:
+    type: RollingUpdate
+---
+apiVersion: argoproj.io/v1alpha1
+kind: RolloutPlugin
+metadata:
+  name: bg-analysis-fail-rp
+spec:
+  workloadRef:
+    apiVersion: apps/v1
+    kind: StatefulSet
+    name: bg-analysis-fail-sts
+  plugin:
+    name: statefulset
+  strategy:
+    canary:
+      analysis:
+        templates:
+        - templateName: bg-analysis-fail
+      steps:
+      - setWeight: 60
+      - pause: {duration: 5s}
+      - setWeight: 80
+      - pause: {duration: 5s}
+      - setWeight: 100