fix: process aggregate outputs for nodes with retries. Fixes #14228

Signed-off-by: oninowang <[email protected]>

Author: jswxstw

test/e2e/functional/parameter-aggregation-dag-with-retry.yaml

@@ -0,0 +1,34 @@
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  name: parameter-aggregation-dag-with-retry
+spec:
+  retryStrategy:
+    limit: 1
+  entrypoint: fanout-dag-with-output
+  templates:
+    - name: echo-value
+      inputs:
+        parameters:
+          - name: message
+      container:
+        image: argoproj/argosay:v2
+      outputs:
+        parameters:
+          - name: dummy-output
+            value: '{{inputs.parameters.message}}'
+    - name: fanout-dag-with-output
+      dag:
+        tasks:
+          - name: echo-list
+            template: echo-value
+            arguments:
+              parameters:
+                - name: message
+                  value: '{{item}}'
+            withItems: [1, 2, 3]
+      outputs:
+        parameters:
+          - name: dummy-dag-output
+            valueFrom:
+              parameter: '{{tasks.echo-list.outputs.parameters.dummy-output}}'

test/e2e/functional_test.go

@@ -582,6 +582,24 @@ func (s *FunctionalSuite) TestParameterAggregationFromOutputs() {
 		})
 }
 
+func (s *FunctionalSuite) TestParameterAggregationDAGWithRetry() {
+	s.Given().
+		Workflow("@functional/parameter-aggregation-dag-with-retry.yaml").
+		When().
+		SubmitWorkflow().
+		WaitForWorkflow(time.Second * 90).
+		Then().
+		ExpectWorkflow(func(t *testing.T, _ *metav1.ObjectMeta, status *wfv1.WorkflowStatus) {
+			assert.Equal(t, wfv1.WorkflowSucceeded, status.Phase)
+			nodeStatus := status.Nodes.FindByDisplayName("parameter-aggregation-dag-with-retry(0)")
+			require.NotNil(t, nodeStatus)
+			assert.Equal(t, wfv1.NodeSucceeded, nodeStatus.Phase)
+			require.NotNil(t, nodeStatus.Outputs)
+			assert.Len(t, nodeStatus.Outputs.Parameters, 1)
+			assert.Equal(t, `["1","2","3"]`, nodeStatus.Outputs.Parameters[0].Value.String())
+		})
+}
+
 func (s *FunctionalSuite) TestDAGDepends() {
 	s.Given().
 		Workflow("@functional/dag-depends.yaml").

workflow/controller/operator.go

@@ -3236,7 +3236,8 @@ func (n loopNodes) Len() int {
 }
 
 func parseLoopIndex(s string) int {
-	s = strings.SplitN(s, "(", 2)[1]
+	splits := strings.Split(s, "(")
+	s = splits[len(splits)-1]
 	s = strings.SplitN(s, ":", 2)[0]
 	val, err := strconv.Atoi(s)
 	if err != nil {
@@ -3276,7 +3277,7 @@ func (woc *wfOperationCtx) processAggregateNodeOutputs(scope *wfScope, prefix st
 	outputParamValueLists := make(map[string][]string)
 	resultsList := make([]wfv1.Item, 0)
 	for _, node := range childNodes {
-		if node.Outputs == nil || node.Phase != wfv1.NodeSucceeded || node.Type == wfv1.NodeTypeRetry {
+		if node.Outputs == nil || node.Phase != wfv1.NodeSucceeded {
 			continue
 		}
 		if len(node.Outputs.Parameters) > 0 {