fix: Merge templateDefaults into dag task tmpl. Fixes #12821

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))
 }

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}}"]

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)