fix: metrics don't get emitted properly during retry. Fixes #8207 #10463

 (#10489)

Signed-off-by: Jiacheng Xu <xjcmaxwellcjx@gmail.com>
Co-authored-by: Saravanan Balasubramanian <33908564+sarabala1979@users.noreply.github.com>

test/e2e/metrics_test.go

@@ -7,8 +7,11 @@ import (
  "testing"
 
  "github.com/gavv/httpexpect/v2"
+ "github.com/stretchr/testify/assert"
  "github.com/stretchr/testify/suite"
+ metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
+ wfv1 "github.com/argoproj/argo-workflows/v3/pkg/apis/workflow/v1alpha1"
  "github.com/argoproj/argo-workflows/v3/test/e2e/fixtures"
 )
 
@@ -57,6 +60,23 @@ func (s *MetricsSuite) TestMetricsEndpoint() {
  })
 }
 
+func (s *MetricsSuite) TestRetryMetrics() {
+ s.Given().
+ Workflow(`@testdata/workflow-retry-metrics.yaml`).
+ When().
+ SubmitWorkflow().
+ WaitForWorkflow(fixtures.ToBeSucceeded).
+ Then().
+ ExpectWorkflow(func(t *testing.T, metadata *metav1.ObjectMeta, status *wfv1.WorkflowStatus) {
+ assert.Equal(t, wfv1.WorkflowSucceeded, status.Phase)
+ s.e(s.T()).GET("").
+ Expect().
+ Status(200).
+ Body().
+ Contains(`runs_exit_status_counter{exit_code="1",status="Failed"} 3`)
+ })
+}
+
 func TestMetricsSuite(t *testing.T) {
  suite.Run(t, new(MetricsSuite))
 }

test/e2e/testdata/workflow-retry-metrics.yaml

@@ -0,0 +1,38 @@
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+ generateName: workflow-retry-metrics
+spec:
+ entrypoint: main
+ podGC:
+ strategy: OnWorkflowSuccess
+ templates:
+ - name: main
+ steps:
+ - - continueOn:
+ error: true
+ failed: true
+ name: runTest
+ template: run-test
+ - container:
+ name: runner
+ image: 'argoproj/argosay:v2'
+ args:
+ - exit 1
+ command:
+ - sh
+ - -c
+ metrics:
+ prometheus:
+ - counter:
+ value: "1"
+ help: Count of runs by exit code
+ labels:
+ - key: exit_code
+ value: '{{exitCode}}'
+ - key: status
+ value: "{{status}}"
+ name: runs_exit_status_counter
+ name: run-test
+ retryStrategy:
+ limit: "2"

workflow/controller/operator.go

@@ -1922,8 +1922,15 @@ func (woc *wfOperationCtx) executeTemplate(ctx context.Context, nodeName string,
  return retryParentNode, nil
  }
  retryParentNode = processedRetryParentNode
+ lastChildNode := getChildNodeIndex(retryParentNode, woc.wf.Status.Nodes, -1)
  // The retry node might have completed by now.
  if retryParentNode.Fulfilled() {
+ // If retry node has completed, set the output of the last child node to its output.
+ // Runtime parameters (e.g., `status`, `resourceDuration`) in the output will be used to emit metrics.
+ if lastChildNode != nil {
+ retryParentNode.Outputs = lastChildNode.Outputs.DeepCopy()
+ woc.wf.Status.Nodes[node.ID] = *retryParentNode
+ }
  if processedTmpl.Metrics != nil {
  // In this check, a completed node may or may not have existed prior to this execution. If it did exist, ensure that it wasn't
  // completed before this execution. If it did not exist prior, then we can infer that it was completed during this execution.
@@ -1937,14 +1944,12 @@ func (woc *wfOperationCtx) executeTemplate(ctx context.Context, nodeName string,
  if processedTmpl.Synchronization != nil {
  woc.controller.syncManager.Release(woc.wf, node.ID, processedTmpl.Synchronization)
  }
- lastChildNode := getChildNodeIndex(retryParentNode, woc.wf.Status.Nodes, -1)
- if lastChildNode != nil {
- retryParentNode.Outputs = lastChildNode.Outputs.DeepCopy()
- woc.wf.Status.Nodes[node.ID] = *retryParentNode
- }
  return retryParentNode, nil
+ } else if lastChildNode != nil && lastChildNode.Fulfilled() && processedTmpl.Metrics != nil {
+ // If retry node has not completed and last child node has completed, emit metrics for the last child node.
+ localScope, realTimeScope := woc.prepareMetricScope(lastChildNode)
+ woc.computeMetrics(processedTmpl.Metrics.Prometheus, localScope, realTimeScope, false)
  }
- lastChildNode := getChildNodeIndex(retryParentNode, woc.wf.Status.Nodes, -1)
  if lastChildNode != nil && !lastChildNode.Fulfilled() {
  // Last child node is still running.
  nodeName = lastChildNode.Name