From 233152c13671fd958b1a0869b6bd254f5452bc79 Mon Sep 17 00:00:00 2001
From: jswxstw <jswxstw@gmail.com>
Date: Fri, 22 Mar 2024 16:02:29 +0800
Subject: [PATCH] fix: Merge templateDefaults into dag task tmpl. Fixes #12821

Signed-off-by: oninowang <oninowang@tencent.com>
---
 test/e2e/metrics_test.go                    | 17 ++++++++
 test/e2e/testdata/workflow-dag-metrics.yaml | 47 +++++++++++++++++++++
 workflow/controller/dag.go                  |  4 ++
 3 files changed, 68 insertions(+)
 create mode 100644 test/e2e/testdata/workflow-dag-metrics.yaml

diff --git a/test/e2e/metrics_test.go b/test/e2e/metrics_test.go
index d52ddfefefc3..febdf06f38e6 100644
--- a/test/e2e/metrics_test.go
+++ b/test/e2e/metrics_test.go
@@ -77,6 +77,23 @@ func (s *MetricsSuite) TestRetryMetrics() {
 		})
 }
 
+func (s *MetricsSuite) TestDAGMetrics() {
+	s.Given().
+		Workflow(`@testdata/workflow-dag-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(`argo_workflows_result_counter{status="Succeeded"} 5`)
+		})
+}
+
 func TestMetricsSuite(t *testing.T) {
 	suite.Run(t, new(MetricsSuite))
 }
diff --git a/test/e2e/testdata/workflow-dag-metrics.yaml b/test/e2e/testdata/workflow-dag-metrics.yaml
new file mode 100644
index 000000000000..a7573bf41a4b
--- /dev/null
+++ b/test/e2e/testdata/workflow-dag-metrics.yaml
@@ -0,0 +1,47 @@
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  generateName: dag-diamond-
+spec:
+  entrypoint: diamond
+  templateDefaults:
+    metrics:
+      prometheus:
+        - name: result_counter
+          help: "Count of step execution by result status"
+          labels:
+            - key: status
+              value: "{{status}}"
+          counter:
+            value: "1"
+  templates:
+    - name: diamond
+      dag:
+        tasks:
+          - name: A
+            template: echo
+            arguments:
+              parameters: [{name: message, value: A}]
+          - name: B
+            depends: "A"
+            template: echo
+            arguments:
+              parameters: [{name: message, value: B}]
+          - name: C
+            depends: "A"
+            template: echo
+            arguments:
+              parameters: [{name: message, value: C}]
+          - name: D
+            depends: "B && C"
+            template: echo
+            arguments:
+              parameters: [{name: message, value: D}]
+
+    - name: echo
+      inputs:
+        parameters:
+          - name: message
+      container:
+        image: alpine:3.7
+        command: [echo, "{{inputs.parameters.message}}"]
diff --git a/workflow/controller/dag.go b/workflow/controller/dag.go
index fd60b5262eed..f82b0b3d2af6 100644
--- a/workflow/controller/dag.go
+++ b/workflow/controller/dag.go
@@ -432,6 +432,10 @@ func (woc *wfOperationCtx) executeDAGTask(ctx context.Context, dagCtx *dagContex
 	if node != nil && node.Fulfilled() {
 		// Collect the completed task metrics
 		_, tmpl, _, _ := dagCtx.tmplCtx.ResolveTemplate(task)
+		if err := woc.mergedTemplateDefaultsInto(tmpl); err != nil {
+			woc.markNodeError(node.Name, err)
+			return
+		}
 		if tmpl != nil && tmpl.Metrics != nil {
 			if prevNodeStatus, ok := woc.preExecutionNodePhases[node.ID]; ok && !prevNodeStatus.Fulfilled() {
 				localScope, realTimeScope := woc.prepareMetricScope(node)