feat: Gracefully recover when AnalysisRuns have errors (#968)

juliev0 · dpadhiar · web-flow · commit f09311845ef8 · 2026-02-23T16:18:15.000-08:00
Signed-off-by: Julie Vogelman &lt;julie_vogelman@intuit.com&gt;
Co-authored-by: Dillen Padhiar &lt;38965141+dpadhiar@users.noreply.github.com&gt;
diff --git a/internal/controller/monovertexrollout/monovertexrollout_progressive.go b/internal/controller/monovertexrollout/monovertexrollout_progressive.go
@@ -111,7 +111,8 @@ func (r *MonoVertexRolloutReconciler) AssessUpgradingChild(
 				// Success window passed, launch AnalysisRuns or declare success
 				childStatus.BasicAssessmentEndTime = &metav1.Time{Time: currentTime}
 				childStatus.BasicAssessmentResult = apiv1.AssessmentResultSuccess
-				return r.checkAnalysisTemplates(ctx, mvtxRollout, existingUpgradingChildDef)
+				assessment, err := r.checkAnalysisTemplates(ctx, mvtxRollout, existingUpgradingChildDef)
+				return assessment, "", err
 			}
 
 			numaLogger.Debugf("Assessment succeeded for upgrading child %s, but success window has not passed yet", existingUpgradingChildDef.GetName())
@@ -122,7 +123,8 @@ func (r *MonoVertexRolloutReconciler) AssessUpgradingChild(
 		if childStatus.BasicAssessmentResult == apiv1.AssessmentResultSuccess {
 			childStatus.ChildStatus.Raw = nil
 			childStatus.FailureReasons = nil
-			return r.checkAnalysisTemplates(ctx, mvtxRollout, existingUpgradingChildDef)
+			assessment, err := r.checkAnalysisTemplates(ctx, mvtxRollout, existingUpgradingChildDef)
+			return assessment, "", err
 		}
 		return childStatus.BasicAssessmentResult, "Basic assessment failed", nil
 	}
@@ -134,7 +136,7 @@ func (r *MonoVertexRolloutReconciler) AssessUpgradingChild(
 // otherwise it returns success.
 func (r *MonoVertexRolloutReconciler) checkAnalysisTemplates(ctx context.Context,
 	mvtxRollout *apiv1.MonoVertexRollout,
-	existingUpgradingChildDef *unstructured.Unstructured) (apiv1.AssessmentResult, string, error) {
+	existingUpgradingChildDef *unstructured.Unstructured) (apiv1.AssessmentResult, error) {
 
 	numaLogger := logger.FromContext(ctx)
 	analysis := mvtxRollout.GetAnalysis()
@@ -144,11 +146,15 @@ func (r *MonoVertexRolloutReconciler) checkAnalysisTemplates(ctx context.Context
 		numaLogger.Debugf("Performing analysis for upgrading child %s", existingUpgradingChildDef.GetName())
 		analysisStatus, err := progressive.PerformAnalysis(ctx, existingUpgradingChildDef, mvtxRollout, mvtxRollout.GetAnalysis(), mvtxRollout.GetAnalysisStatus(), r.client)
 		if err != nil {
-			return apiv1.AssessmentResultUnknown, "", err
+			return apiv1.AssessmentResultUnknown, err
+		}
+		assessment, err := progressive.AssessAnalysisStatus(ctx, existingUpgradingChildDef, analysisStatus)
+		if err != nil {
+			return apiv1.AssessmentResultUnknown, err
 		}
-		return progressive.AssessAnalysisStatus(ctx, existingUpgradingChildDef, analysisStatus)
+		return assessment, nil
 	}
-	return apiv1.AssessmentResultSuccess, "", nil
+	return apiv1.AssessmentResultSuccess, nil
 }
 
 // CheckForDifferences checks to see if the monovertex definition matches the spec and the required metadata
diff --git a/internal/controller/pipelinerollout/pipelinerollout_progressive.go b/internal/controller/pipelinerollout/pipelinerollout_progressive.go
@@ -166,7 +166,8 @@ func (r *PipelineRolloutReconciler) AssessUpgradingChild(
 				// Success window passed, launch AnalysisRun or declared success
 				childStatus.BasicAssessmentEndTime = &metav1.Time{Time: currentTime}
 				childStatus.BasicAssessmentResult = apiv1.AssessmentResultSuccess
-				return r.checkAnalysisTemplates(ctx, pipelineRollout, existingUpgradingChildDef)
+				assessment, err := r.checkAnalysisTemplates(ctx, pipelineRollout, existingUpgradingChildDef)
+				return assessment, "", err
 			}
 
 			numaLogger.Debugf("Assessment succeeded for upgrading child %s, but success window has not passed yet", existingUpgradingChildDef.GetName())
@@ -177,7 +178,8 @@ func (r *PipelineRolloutReconciler) AssessUpgradingChild(
 		if childStatus.BasicAssessmentResult == apiv1.AssessmentResultSuccess {
 			childStatus.ChildStatus.Raw = nil
 			childStatus.FailureReasons = nil
-			return r.checkAnalysisTemplates(ctx, pipelineRollout, existingUpgradingChildDef)
+			assessment, err := r.checkAnalysisTemplates(ctx, pipelineRollout, existingUpgradingChildDef)
+			return assessment, "", err
 		}
 		return childStatus.BasicAssessmentResult, "Basic assessment failed", nil
 	}
@@ -188,14 +190,14 @@ func (r *PipelineRolloutReconciler) AssessUpgradingChild(
 // checkAnalysisTemplates checks if there are any analysis templates to run and runs them if so.
 func (r *PipelineRolloutReconciler) checkAnalysisTemplates(ctx context.Context,
 	pipelineRollout *apiv1.PipelineRollout,
-	existingUpgradingChildDef *unstructured.Unstructured) (apiv1.AssessmentResult, string, error) {
+	existingUpgradingChildDef *unstructured.Unstructured) (apiv1.AssessmentResult, error) {
 
 	numaLogger := logger.FromContext(ctx)
 	analysis := pipelineRollout.GetAnalysis()
 
 	globalConfig, err := config.GetConfigManagerInstance().GetConfig()
 	if err != nil {
-		return apiv1.AssessmentResultUnknown, "", fmt.Errorf("error getting the global config: %v", err)
+		return apiv1.AssessmentResultUnknown, fmt.Errorf("error getting the global config: %v", err)
 	}
 
 	// only check for and create AnalysisRun if templates are specified
@@ -204,11 +206,15 @@ func (r *PipelineRolloutReconciler) checkAnalysisTemplates(ctx context.Context,
 		numaLogger.Debugf("Performing analysis for upgrading child %s", existingUpgradingChildDef.GetName())
 		analysisStatus, err := progressive.PerformAnalysis(ctx, existingUpgradingChildDef, pipelineRollout, pipelineRollout.GetAnalysis(), pipelineRollout.GetAnalysisStatus(), r.client)
 		if err != nil {
-			return apiv1.AssessmentResultUnknown, "", err
+			return apiv1.AssessmentResultUnknown, err
+		}
+		assessment, err := progressive.AssessAnalysisStatus(ctx, existingUpgradingChildDef, analysisStatus)
+		if err != nil {
+			return apiv1.AssessmentResultUnknown, err
 		}
-		return progressive.AssessAnalysisStatus(ctx, existingUpgradingChildDef, analysisStatus)
+		return assessment, nil
 	}
-	return apiv1.AssessmentResultSuccess, "", nil
+	return apiv1.AssessmentResultSuccess, nil
 }
 
 // CheckForDifferences checks to see if the pipeline definition matches the spec and the required metadata
diff --git a/internal/controller/progressive/analysis.go b/internal/controller/progressive/analysis.go
@@ -89,6 +89,44 @@ func GetAnalysisTemplatesFromRefs(ctx context.Context, templateRefs *[]argorollo
 	return uniqueTemplates, uniqueClusterTemplates, nil
 }
 
+// dedupMetrics returns a copy of the given metrics slice with duplicates removed by metric name.
+// The first occurrence of each name is kept.
+func dedupMetrics(metrics []argorolloutsv1.Metric) []argorolloutsv1.Metric {
+	seen := make(map[string]bool)
+	out := make([]argorolloutsv1.Metric, 0, len(metrics))
+	for _, m := range metrics {
+		if seen[m.Name] {
+			continue
+		}
+		seen[m.Name] = true
+		out = append(out, m)
+	}
+	return out
+}
+
+// dedupMetricsForAnalysisTemplates removes duplicate metrics from the given AnalysisTemplates and
+// ClusterAnalysisTemplates. When templates are merged (e.g. from nested refs), the same metric may
+// appear multiple times; this function ensures each metric appears at most once per template.
+// It returns the same template slices with metrics deduplicated, or an error if processing fails.
+func dedupMetricsForAnalysisTemplates(ctx context.Context, templates []*argorolloutsv1.AnalysisTemplate, clusterTemplates []*argorolloutsv1.ClusterAnalysisTemplate) ([]*argorolloutsv1.AnalysisTemplate, []*argorolloutsv1.ClusterAnalysisTemplate, error) {
+	numaLogger := logger.FromContext(ctx)
+	for index, t := range templates {
+		deduped := dedupMetrics(t.Spec.Metrics)
+		if len(deduped) < len(t.Spec.Metrics) {
+			numaLogger.Warnf("AnalysisTemplate %s had duplicate metrics; using first occurrence only", t.Name)
+		}
+		templates[index].Spec.Metrics = deduped
+	}
+	for index, t := range clusterTemplates {
+		deduped := dedupMetrics(t.Spec.Metrics)
+		if len(deduped) < len(t.Spec.Metrics) {
+			numaLogger.Warnf("ClusterAnalysisTemplate %s had duplicate metrics; using first occurrence only", t.Name)
+		}
+		clusterTemplates[index].Spec.Metrics = deduped
+	}
+	return templates, clusterTemplates, nil
+}
+
 /*
 CreateAnalysisRun finds all templates specified in the Analysis field in the spec of a rollout and creates the resulting AnalysisRun in k8s.
 
@@ -115,6 +153,12 @@ func CreateAnalysisRun(ctx context.Context, analysis apiv1.Analysis, existingUpg
 		return err
 	}
 
+	// temporary code to take care of an issue in which incoming AnalysisTemplates have a duplicate metric
+	analysisTemplates, clusterAnalysisTemplates, err = dedupMetricsForAnalysisTemplates(ctx, analysisTemplates, clusterAnalysisTemplates)
+	if err != nil {
+		return err
+	}
+
 	// set special arguments for child name and namespace
 	childName := existingUpgradingChildDef.GetName()
 	childNamespace := existingUpgradingChildDef.GetNamespace()
@@ -206,33 +250,41 @@ func PerformAnalysis(
 func AssessAnalysisStatus(
 	ctx context.Context,
 	existingUpgradingChildDef *unstructured.Unstructured,
-	analysisStatus *apiv1.AnalysisStatus) (apiv1.AssessmentResult, string, error) {
+	analysisStatus *apiv1.AnalysisStatus) (apiv1.AssessmentResult, error) {
 	numaLogger := logger.FromContext(ctx)
 
 	// make sure we haven't gone past the max time allowed for an AnalysisRun
 	analysisRunTimeout, err := getAnalysisRunTimeout(ctx)
 	if err != nil {
-		return apiv1.AssessmentResultUnknown, "", err
+		return apiv1.AssessmentResultUnknown, err
 	}
 
-	// if analysisStatus is set with an AnalysisRun's name, we must also check that it is in a Completed phase to declare success
+	// if analysisStatus is set, we must also check that it is in a Completed phase to declare success
 	if analysisStatus != nil && analysisStatus.AnalysisRunName != "" {
 		numaLogger.WithValues("namespace", existingUpgradingChildDef.GetNamespace(), "name", existingUpgradingChildDef.GetName()).
 			Debugf("AnalysisRun %s is in phase %s", analysisStatus.AnalysisRunName, analysisStatus.Phase)
 		switch analysisStatus.Phase {
 		case argorolloutsv1.AnalysisPhaseSuccessful:
-			return apiv1.AssessmentResultSuccess, "", nil
-		case argorolloutsv1.AnalysisPhaseError, argorolloutsv1.AnalysisPhaseFailed, argorolloutsv1.AnalysisPhaseInconclusive:
-			return apiv1.AssessmentResultFailure, fmt.Sprintf("AnalysisRun %s is in phase %s", analysisStatus.AnalysisRunName, analysisStatus.Phase), nil
+			return apiv1.AssessmentResultSuccess, nil
+		case argorolloutsv1.AnalysisPhaseFailed:
+			return apiv1.AssessmentResultFailure, nil
+		case argorolloutsv1.AnalysisPhaseError, argorolloutsv1.AnalysisPhaseInconclusive:
+			// Decision not to consider this a failure since it's either a misconfiguration of the AnalysisTemplate or unavailable Provider, etc
+			numaLogger.WithValues("namespace", existingUpgradingChildDef.GetNamespace(), "name", existingUpgradingChildDef.GetName()).
+				Warnf("AnalysisRun %s is in phase %s", analysisStatus.AnalysisRunName, analysisStatus.Phase)
+			return apiv1.AssessmentResultSuccess, nil
 		default:
 			// if analysisRun is not completed yet, we check if it has exceeded the analysisRunTimeout
+			// Decision not to consider this a failure since it's either a misconfiguration of the AnalysisTemplate or Argo Rollouts unavailable, etc
 			if time.Since(analysisStatus.StartTime.Time) >= analysisRunTimeout {
-				return apiv1.AssessmentResultFailure, fmt.Sprintf("AnalysisRun %s in phase %s has exceeded the analysisRunTimeout", analysisStatus.AnalysisRunName, analysisStatus.Phase), nil
+				numaLogger.WithValues("namespace", existingUpgradingChildDef.GetNamespace(), "name", existingUpgradingChildDef.GetName()).
+					Warnf("AnalysisRun %s has exceeded the analysisRunTimeout", analysisStatus.AnalysisRunName)
+				return apiv1.AssessmentResultSuccess, nil
 			}
-			return apiv1.AssessmentResultUnknown, "", nil
+			return apiv1.AssessmentResultUnknown, nil
 		}
 	}
 
 	// no AnalysisRun so by default we can mark this successful
-	return apiv1.AssessmentResultSuccess, "", nil
+	return apiv1.AssessmentResultSuccess, nil
 }
diff --git a/internal/controller/progressive/analysis_test.go b/internal/controller/progressive/analysis_test.go
@@ -0,0 +1,155 @@
+/*
+Copyright 2023.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package progressive
+
+import (
+	"context"
+	"testing"
+
+	argorolloutsv1 "github.com/argoproj/argo-rollouts/pkg/apis/rollouts/v1alpha1"
+	"github.com/stretchr/testify/assert"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func metric(name string) argorolloutsv1.Metric {
+	return argorolloutsv1.Metric{
+		Name: name,
+		Provider: argorolloutsv1.MetricProvider{
+			Prometheus: &argorolloutsv1.PrometheusMetric{
+				Query: "up",
+			},
+		},
+		SuccessCondition: "true",
+	}
+}
+
+func metricNames(metrics []argorolloutsv1.Metric) []string {
+	names := make([]string, len(metrics))
+	for i := range metrics {
+		names[i] = metrics[i].Name
+	}
+	return names
+}
+
+func TestDedupMetricsForAnalysisTemplates(t *testing.T) {
+	ctx := context.Background()
+
+	testCases := []struct {
+		name                           string
+		templates                      []*argorolloutsv1.AnalysisTemplate
+		clusterTemplates               []*argorolloutsv1.ClusterAnalysisTemplate
+		expectedTemplateMetrics        map[string][]string // template name -> expected metric names in order
+		expectedClusterTemplateMetrics map[string][]string // cluster template name -> expected metric names in order
+	}{
+		{
+			name:                           "nil and empty slices",
+			templates:                      nil,
+			clusterTemplates:               nil,
+			expectedTemplateMetrics:        nil,
+			expectedClusterTemplateMetrics: nil,
+		},
+		{
+			name:                           "empty templates no change",
+			templates:                      []*argorolloutsv1.AnalysisTemplate{},
+			clusterTemplates:               []*argorolloutsv1.ClusterAnalysisTemplate{},
+			expectedTemplateMetrics:        map[string][]string{},
+			expectedClusterTemplateMetrics: map[string][]string{},
+		},
+		{
+			name: "AnalysisTemplate with no duplicates unchanged",
+			templates: []*argorolloutsv1.AnalysisTemplate{{
+				ObjectMeta: metav1.ObjectMeta{Name: "t1", Namespace: "ns"},
+				Spec: argorolloutsv1.AnalysisTemplateSpec{
+					Metrics: []argorolloutsv1.Metric{metric("a"), metric("b")},
+				},
+			}},
+			clusterTemplates:               nil,
+			expectedTemplateMetrics:        map[string][]string{"t1": {"a", "b"}},
+			expectedClusterTemplateMetrics: nil,
+		},
+		{
+			name: "AnalysisTemplate with duplicate metric names",
+			templates: []*argorolloutsv1.AnalysisTemplate{{
+				ObjectMeta: metav1.ObjectMeta{Name: "at"},
+				Spec: argorolloutsv1.AnalysisTemplateSpec{
+					Metrics: []argorolloutsv1.Metric{metric("m1"), metric("m1"), metric("m2")},
+				},
+			}},
+			clusterTemplates:               nil,
+			expectedTemplateMetrics:        map[string][]string{"at": {"m1", "m2"}},
+			expectedClusterTemplateMetrics: nil,
+		},
+		{
+			name:      "ClusterAnalysisTemplate with duplicate metric names",
+			templates: nil,
+			clusterTemplates: []*argorolloutsv1.ClusterAnalysisTemplate{{
+				ObjectMeta: metav1.ObjectMeta{Name: "ct1"},
+				Spec: argorolloutsv1.AnalysisTemplateSpec{
+					Metrics: []argorolloutsv1.Metric{metric("cluster-metric"), metric("cluster-metric")},
+				},
+			}},
+			expectedTemplateMetrics:        nil,
+			expectedClusterTemplateMetrics: map[string][]string{"ct1": {"cluster-metric"}},
+		},
+		{
+			name: "both template types with duplicate metric names",
+			templates: []*argorolloutsv1.AnalysisTemplate{{
+				ObjectMeta: metav1.ObjectMeta{Name: "at"},
+				Spec: argorolloutsv1.AnalysisTemplateSpec{
+					Metrics: []argorolloutsv1.Metric{metric("m1"), metric("m1"), metric("m2")},
+				},
+			}},
+			clusterTemplates: []*argorolloutsv1.ClusterAnalysisTemplate{{
+				ObjectMeta: metav1.ObjectMeta{Name: "ct1"},
+				Spec: argorolloutsv1.AnalysisTemplateSpec{
+					Metrics: []argorolloutsv1.Metric{metric("cluster-metric"), metric("cluster-metric")},
+				},
+			}},
+			expectedTemplateMetrics:        map[string][]string{"at": {"m1", "m2"}},
+			expectedClusterTemplateMetrics: map[string][]string{"ct1": {"cluster-metric"}},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			gotT, gotC, err := dedupMetricsForAnalysisTemplates(ctx, tc.templates, tc.clusterTemplates)
+			assert.NoError(t, err)
+
+			if tc.expectedTemplateMetrics == nil {
+				assert.Nil(t, gotT)
+			} else {
+				assert.Len(t, gotT, len(tc.expectedTemplateMetrics))
+				for _, tpl := range gotT {
+					wantNames, ok := tc.expectedTemplateMetrics[tpl.Name]
+					assert.True(t, ok, "unexpected AnalysisTemplate name %q", tpl.Name)
+					assert.Equal(t, wantNames, metricNames(tpl.Spec.Metrics), "AnalysisTemplate %q", tpl.Name)
+				}
+			}
+
+			if tc.expectedClusterTemplateMetrics == nil {
+				assert.Nil(t, gotC)
+			} else {
+				assert.Len(t, gotC, len(tc.expectedClusterTemplateMetrics))
+				for _, tpl := range gotC {
+					wantNames, ok := tc.expectedClusterTemplateMetrics[tpl.Name]
+					assert.True(t, ok, "unexpected ClusterAnalysisTemplate name %q", tpl.Name)
+					assert.Equal(t, wantNames, metricNames(tpl.Spec.Metrics), "ClusterAnalysisTemplate %q", tpl.Name)
+				}
+			}
+		})
+	}
+}
diff --git a/tests/e2e/progressive-analysis-monovertex-e2e/analysis_monovertex_test.go b/tests/e2e/progressive-analysis-monovertex-e2e/analysis_monovertex_test.go
@@ -130,7 +130,7 @@ OR
 )`,
 					},
 				},
-				SuccessCondition: "result[0] > 0",
+				SuccessCondition: "len(result) > 0 && result[0] > 0",
 			},
 		},
 		Args: []argov1alpha1.Argument{
@@ -204,7 +204,7 @@ var _ = Describe("Progressive MonoVertex E2E", Serial, func() {
 		VerifyMonoVertexProgressiveFailure(monoVertexRolloutName, monoVertexScaleMinMaxJSONString, updatedMonoVertexSpec, monoVertexScaleTo, false)
 
 		// Verify the AnalysisRun status is Failed
-		VerifyAnalysisRunStatus("mvtx-example-1", GetInstanceName(analysisRunName, 1), argov1alpha1.AnalysisPhaseError)
+		VerifyAnalysisRunStatus("mvtx-example-1", GetInstanceName(analysisRunName, 1), argov1alpha1.AnalysisPhaseFailed)
 		VerifyAnalysisRunStatus("mvtx-example-2", GetInstanceName(analysisRunName, 1), argov1alpha1.AnalysisPhaseSuccessful)
 
 		DeleteMonoVertexRollout(monoVertexRolloutName)
diff --git a/tests/e2e/progressive-analysis-pipeline-e2e/analysis_pipeline_test.go b/tests/e2e/progressive-analysis-pipeline-e2e/analysis_pipeline_test.go
