feat: Support per-output parameter aggregation (#4374)

Signed-off-by: Alexey Volkov <alexey.volkov@ark-kun.com>

docs/variables.md

@@ -47,8 +47,8 @@ The following variables are made available to reference various metadata of a wo
 | `steps.<STEPNAME>.startedAt` | Timestamp when the step started |
 | `steps.<STEPNAME>.finishedAt` | Timestamp when the step finished |
 | `steps.<STEPNAME>.outputs.result` | Output result of any previous container or script step |
-| `steps.<STEPNAME>.outputs.parameters` | When the previous step uses 'withItems', this contains a JSON array of the output parameters of each invocation |
-| `steps.<STEPNAME>.outputs.parameters.<NAME>` | Output parameter of any previous step |
+| `steps.<STEPNAME>.outputs.parameters` | When the previous step uses 'withItems' or 'withParams', this contains a JSON array of the output parameter maps of each invocation |
+| `steps.<STEPNAME>.outputs.parameters.<NAME>` | Output parameter of any previous step. When the previous step uses 'withItems' or 'withParams', this contains a JSON array of the output parameter values of each invocation |
 | `steps.<STEPNAME>.outputs.artifacts.<NAME>` | Output artifact of any previous step |
 
 ## DAG Templates
@@ -61,8 +61,8 @@ The following variables are made available to reference various metadata of a wo
 | `tasks.<TASKNAME>.startedAt` | Timestamp when the task started |
 | `tasks.<TASKNAME>.finishedAt` | Timestamp when the task finished |
 | `tasks.<TASKNAME>.outputs.result` | Output result of any previous container or script task |
-| `tasks.<TASKNAME>.outputs.parameters` | When the previous task uses 'withItems', this contains a JSON array of the output parameters of each invocation |
-| `tasks.<TASKNAME>.outputs.parameters.<NAME>` | Output parameter of any previous task |
+| `tasks.<TASKNAME>.outputs.parameters` | When the previous task uses 'withItems' or 'withParams', this contains a JSON array of the output parameter maps of each invocation |
+| `tasks.<TASKNAME>.outputs.parameters.<NAME>` | Output parameter of any previous task. When the previous task uses 'withItems' or 'withParams', this contains a JSON array of the output parameter values of each invocation |
 | `tasks.<TASKNAME>.outputs.artifacts.<NAME>` | Output artifact of any previous task |
 
 ## Container/Script Templates

examples/parameter-aggregation-dag.yaml

@@ -18,6 +18,20 @@ spec:
           - name: num
             value: "{{item}}"
         withItems: [1, 2, 3, 4]
+      - name: print-nums
+        template: whalesay
+        dependencies: [odd-or-even]
+        arguments:
+          parameters:
+          - name: message
+            value: "{{tasks.odd-or-even.outputs.parameters.num}}"
+      - name: print-evenness
+        template: whalesay
+        dependencies: [odd-or-even]
+        arguments:
+          parameters:
+          - name: message
+            value: "{{tasks.odd-or-even.outputs.parameters.evenness}}"
       # Next, for each even number, divide it by two (using a script template).
       # Skip odd numbers using a `when` clause.
       - name: divide-by-2

examples/parameter-aggregation.yaml

@@ -17,6 +17,18 @@ spec:
           - name: num
             value: "{{item}}"
         withItems: [1, 2, 3, 4]
+    - - name: print-nums
+        template: whalesay
+        arguments:
+          parameters:
+          - name: message
+            value: "{{steps.odd-or-even.outputs.parameters.num}}"
+    - - name: print-evenness
+        template: whalesay
+        arguments:
+          parameters:
+          - name: message
+            value: "{{steps.odd-or-even.outputs.parameters.evenness}}"
     # Next, for each even number, divide it by two (using a script template).
     # Skip odd numbers using a `when` clause.
     - - name: divide-by-2

workflow/controller/operator.go

@@ -2442,6 +2442,7 @@ func (woc *wfOperationCtx) processAggregateNodeOutputs(tmpl *wfv1.Template, scop
 	// need to sort the child node list so that the order of outputs are preserved
 	sort.Sort(loopNodes(childNodes))
 	paramList := make([]map[string]string, 0)
+	outputParamValueLists := make(map[string][]string)
 	resultsList := make([]wfv1.Item, 0)
 	for _, node := range childNodes {
 		if node.Outputs == nil {
@@ -2451,6 +2452,9 @@ func (woc *wfOperationCtx) processAggregateNodeOutputs(tmpl *wfv1.Template, scop
 			param := make(map[string]string)
 			for _, p := range node.Outputs.Parameters {
 				param[p.Name] = p.Value.String()
+				outputParamValueList := outputParamValueLists[p.Name]
+				outputParamValueList = append(outputParamValueList, p.Value.String())
+				outputParamValueLists[p.Name] = outputParamValueList
 			}
 			paramList = append(paramList, param)
 		}
@@ -2478,6 +2482,15 @@ func (woc *wfOperationCtx) processAggregateNodeOutputs(tmpl *wfv1.Template, scop
 	}
 	key := fmt.Sprintf("%s.outputs.parameters", prefix)
 	scope.addParamToScope(key, string(outputsJSON))
+	// Adding per-output aggregated value placeholders
+	for outputName, valueList := range outputParamValueLists {
+		key = fmt.Sprintf("%s.outputs.parameters.%s", prefix, outputName)
+		valueListJSON, err := json.Marshal(valueList)
+		if err != nil {
+			return err
+		}
+		scope.addParamToScope(key, string(valueListJSON))
+	}
 	return nil
 }