Implemented Conditionally annotate outputs of script template only when consumed #1359

cmd/argoexec/commands/wait.go

@@ -67,5 +67,6 @@ func waitContainer() error {
 		wfExecutor.AddError(err)
 		return err
 	}
+
 	return nil
 }

workflow/common/common.go

@@ -34,6 +34,7 @@ const (
 
 	// AnnotationKeyNodeName is the pod metadata annotation key containing the workflow node name
 	AnnotationKeyNodeName = workflow.FullName + "/node-name"
+
 	// AnnotationKeyNodeMessage is the pod metadata annotation key the executor will use to
 	// communicate errors encountered by the executor during artifact load/save, etc...
 	AnnotationKeyNodeMessage = workflow.FullName + "/node-message"
@@ -130,6 +131,8 @@ type ExecutionControl struct {
 	// It is used to signal the executor to terminate a daemoned container. In the future it will be
 	// used to support workflow or steps/dag level timeouts.
 	Deadline *time.Time `json:"deadline,omitempty"`
+	// IncludeScriptOutput is containing flag to include script output
+	IncludeScriptOutput bool `json:"includeScriptOutput,omitempty"`
 }
 
 type ResourceInterface interface {

workflow/controller/dag.go

@@ -66,7 +66,7 @@ func (d *dagContext) assertBranchFinished(targetTaskName string) bool {
 	// We should ensure that from the bottom to the top,
 	// all the nodes of this branch have at least one failure.
 	// If successful, we should continue to run down until the leaf node
-	taskNode := d.getTaskNode(targetTaskName)
+	taskNode := d.GetTaskNode(targetTaskName)
 	if taskNode == nil {
 		taskObject := d.getTask(targetTaskName)
 		if taskObject != nil {

workflow/controller/exec_control.go

@@ -48,32 +48,33 @@ func (woc *wfOperationCtx) applyExecutionControl(pod *apiv1.Pod, wfNodesLock *sy
 		}
 	}
 
-	// Now ensure the pod's current annotation matches our desired deadline
-	desiredExecCtl := common.ExecutionControl{
-		Deadline: woc.workflowDeadline,
-	}
 	var podExecCtl common.ExecutionControl
 	if execCtlStr, ok := pod.Annotations[common.AnnotationKeyExecutionControl]; ok && execCtlStr != "" {
 		err := json.Unmarshal([]byte(execCtlStr), &podExecCtl)
 		if err != nil {
 			woc.log.Warnf("Failed to unmarshal execution control from pod %s", pod.Name)
 		}
 	}
-	if podExecCtl.Deadline == nil && desiredExecCtl.Deadline == nil {
+	if podExecCtl.Deadline == nil && woc.workflowDeadline == nil {
 		return nil
-	} else if podExecCtl.Deadline != nil && desiredExecCtl.Deadline != nil {
-		if podExecCtl.Deadline.Equal(*desiredExecCtl.Deadline) {
+	} else if podExecCtl.Deadline != nil && woc.workflowDeadline != nil {
+		if podExecCtl.Deadline.Equal(*woc.workflowDeadline) {
 			return nil
 		}
 	}
+
 	if podExecCtl.Deadline != nil && podExecCtl.Deadline.IsZero() {
 		// If the pod has already been explicitly signaled to terminate, then do nothing.
 		// This can happen when daemon steps are terminated.
 		woc.log.Infof("Skipping sync of execution control of pod %s. pod has been signaled to terminate", pod.Name)
 		return nil
 	}
-	woc.log.Infof("Execution control for pod %s out-of-sync desired: %v, actual: %v", pod.Name, desiredExecCtl.Deadline, podExecCtl.Deadline)
-	return woc.updateExecutionControl(pod.Name, desiredExecCtl)
+
+	// Assign new deadline value to PodExeCtl
+	podExecCtl.Deadline = woc.workflowDeadline
+
+	woc.log.Infof("Execution control for pod %s out-of-sync desired: %v, actual: %v", pod.Name, woc.workflowDeadline, podExecCtl.Deadline)
+	return woc.updateExecutionControl(pod.Name, podExecCtl)
 }
 
 // killDaemonedChildren kill any daemoned pods of a steps or DAG template node.

workflow/controller/operator.go

@@ -1293,7 +1293,7 @@ func (woc *wfOperationCtx) executeContainer(nodeName string, tmpl *wfv1.Template
 		return node
 	}
 	woc.log.Debugf("Executing node %s with container template: %v\n", nodeName, tmpl)
-	_, err := woc.createWorkflowPod(nodeName, *tmpl.Container, tmpl)
+	_, err := woc.createWorkflowPod(nodeName, *tmpl.Container, tmpl, false)
 	if err != nil {
 		return woc.initializeNode(nodeName, wfv1.NodeTypePod, tmpl.Name, boundaryID, wfv1.NodeError, err.Error())
 	}
@@ -1365,14 +1365,62 @@ func getTemplateOutputsFromScope(tmpl *wfv1.Template, scope *wfScope) (*wfv1.Out
 	return &outputs, nil
 }
 
+func checkVariableRefInTmpl(template *wfv1.Template, variable string) bool {
+	jsonValue, err := json.Marshal(template)
+	if err != nil {
+		log.Warnf("Unable to marshal the template. %v, %v", template, err)
+	}
+	jsonStr := string(jsonValue)
+
+	return strings.Contains(jsonStr, variable)
+}
+
+// hasOutputResultRef will check given template output has any reference
+func hasOutputResultRef(name string, parentTmpl *wfv1.Template) bool {
+
+	var variableRefName string
+	if parentTmpl.DAG != nil {
+		variableRefName = "{{tasks." + name + ".outputs.result}}"
+	} else if parentTmpl.Steps != nil {
+		variableRefName = "{{steps." + name + ".outputs.result}}"
+	}
+
+	return checkVariableRefInTmpl(parentTmpl, variableRefName)
+}
+
+// getStepOrDAGTaskName will extract the node from NodeStatus Name
+func getStepOrDAGTaskName(nodeName string, hasRetryStrategy bool) string {
+	if strings.Contains(nodeName, ".") {
+		name := nodeName[strings.LastIndex(nodeName, ".")+1:]
+		// Check retry scenario
+		if hasRetryStrategy {
+			if indx := strings.LastIndex(name, "("); indx > 0 {
+				return name[0:indx]
+			}
+		}
+		return name
+	}
+	return nodeName
+}
+
 func (woc *wfOperationCtx) executeScript(nodeName string, tmpl *wfv1.Template, boundaryID string) *wfv1.NodeStatus {
+
+	boundaryNode := woc.wf.Status.Nodes[boundaryID]
+	parentTemplate := woc.wf.GetTemplate(boundaryNode.TemplateName)
+
+	includeScriptOutput := false
+	if parentTemplate != nil {
+		name := getStepOrDAGTaskName(nodeName, tmpl.RetryStrategy != nil)
+		includeScriptOutput = hasOutputResultRef(name, parentTemplate)
+	}
 	node := woc.getNodeByName(nodeName)
+
 	if node != nil {
 		return node
 	}
 	mainCtr := tmpl.Script.Container
 	mainCtr.Args = append(mainCtr.Args, common.ExecutorScriptSourcePath)
-	_, err := woc.createWorkflowPod(nodeName, mainCtr, tmpl)
+	_, err := woc.createWorkflowPod(nodeName, mainCtr, tmpl, includeScriptOutput)
 	if err != nil {
 		return woc.initializeNode(nodeName, wfv1.NodeTypePod, tmpl.Name, boundaryID, wfv1.NodeError, err.Error())
 	}
@@ -1607,7 +1655,7 @@ func (woc *wfOperationCtx) executeResource(nodeName string, tmpl *wfv1.Template,
 	mainCtr.VolumeMounts = []apiv1.VolumeMount{
 		volumeMountPodMetadata,
 	}
-	_, err = woc.createWorkflowPod(nodeName, *mainCtr, tmpl)
+	_, err = woc.createWorkflowPod(nodeName, *mainCtr, tmpl, false)
 	if err != nil {
 		return woc.initializeNode(nodeName, wfv1.NodeTypePod, tmpl.Name, boundaryID, wfv1.NodeError, err.Error())
 	}

workflow/controller/operator_test.go

@@ -3,6 +3,7 @@ package controller
 import (
 	"fmt"
 	"github.com/argoproj/argo/workflow/config"
+	"strings"
 	"testing"
 
 	wfv1 "github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
@@ -1160,6 +1161,127 @@ func TestResourceWithOwnerReferenceTemplate(t *testing.T) {
 	}
 }
 
+var stepScriptTmpl = `
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  generateName: scripts-bash-
+spec:
+  entrypoint: bash-script-example
+  templates:
+  - name: bash-script-example
+    steps:
+    - - name: generate
+        template: gen-random-int
+    - - name: print
+        template: print-message
+        arguments:
+          parameters:
+          - name: message
+            value: "{{steps.generate.outputs.result}}"
+
+  - name: gen-random-int
+    script:
+      image: debian:9.4
+      command: [bash]
+      source: |
+        cat /dev/urandom | od -N2 -An -i | awk -v f=1 -v r=100 '{printf "%i\n", f + r * $1 / 65536}'
+
+  - name: print-message
+    inputs:
+      parameters:
+      - name: message
+    container:
+      image: alpine:latest
+      command: [sh, -c]
+      args: ["echo result was: {{inputs.parameters.message}}"]
+`
+
+var dagScriptTmpl = `
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  generateName: dag-target-
+spec:
+  entrypoint: dag-target
+  arguments:
+    parameters:
+    - name: target
+      value: E
+
+  templates:
+  - name: dag-target
+    dag:
+      tasks:
+      - name: A
+        template: echo
+        arguments:
+          parameters: [{name: message, value: A}]
+      - name: B
+        template: echo
+        arguments:
+          parameters: [{name: message, value: B}]
+      - name: C
+        dependencies: [A]
+        template: echo
+        arguments:
+          parameters: [{name: message, value: "{{tasks.A.outputs.result}}"}]
+  - name: echo
+    script:
+      image: debian:9.4
+      command: [bash]
+      source: |
+        cat /dev/urandom | od -N2 -An -i | awk -v f=1 -v r=100 '{printf "%i\n", f + r * $1 / 65536}'`
+
+func TestStepWFGetNodeName(t *testing.T) {
+
+	controller := newController()
+	wfcset := controller.wfclientset.ArgoprojV1alpha1().Workflows("")
+
+	// operate the workflow. it should create a pod.
+	wf := unmarshalWF(stepScriptTmpl)
+	wf, err := wfcset.Create(wf)
+	assert.Nil(t, err)
+	assert.True(t, hasOutputResultRef("generate", &wf.Spec.Templates[0]))
+	assert.False(t, hasOutputResultRef("print-message", &wf.Spec.Templates[0]))
+	woc := newWorkflowOperationCtx(wf, controller)
+	woc.operate()
+	wf, err = wfcset.Get(wf.ObjectMeta.Name, metav1.GetOptions{})
+	assert.Nil(t, err)
+	for _, node := range wf.Status.Nodes {
+		if strings.Contains(node.Name, "generate") {
+			assert.True(t, getStepOrDAGTaskName(node.Name, &wf.Spec.Templates[0].RetryStrategy != nil) == "generate")
+		} else if strings.Contains(node.Name, "print-message") {
+			assert.True(t, getStepOrDAGTaskName(node.Name, &wf.Spec.Templates[0].RetryStrategy != nil) == "print-message")
+		}
+	}
+}
+
+func TestDAGWFGetNodeName(t *testing.T) {
+
+	controller := newController()
+	wfcset := controller.wfclientset.ArgoprojV1alpha1().Workflows("")
+
+	// operate the workflow. it should create a pod.
+	wf := unmarshalWF(dagScriptTmpl)
+	wf, err := wfcset.Create(wf)
+	assert.Nil(t, err)
+	assert.True(t, hasOutputResultRef("A", &wf.Spec.Templates[0]))
+	assert.False(t, hasOutputResultRef("B", &wf.Spec.Templates[0]))
+	woc := newWorkflowOperationCtx(wf, controller)
+	woc.operate()
+	wf, err = wfcset.Get(wf.ObjectMeta.Name, metav1.GetOptions{})
+	assert.Nil(t, err)
+	for _, node := range wf.Status.Nodes {
+		if strings.Contains(node.Name, ".A") {
+			assert.True(t, getStepOrDAGTaskName(node.Name, wf.Spec.Templates[0].RetryStrategy != nil) == "A")
+		}
+		if strings.Contains(node.Name, ".B") {
+			assert.True(t, getStepOrDAGTaskName(node.Name, wf.Spec.Templates[0].RetryStrategy != nil) == "B")
+		}
+	}
+}
+
 var withParamAsJsonList = `
 apiVersion: argoproj.io/v1alpha1
 kind: Workflow
@@ -1181,14 +1303,14 @@ spec:
           - name: message
             value: "{{item}}"
         withParam: "{{workflow.parameters.input}}"
-  - name: whalesay
+  - name: whalesay 
     inputs:
       parameters:
       - name: message
-    container:
+    script:
       image: alpine:latest
       command: [sh, -c]
-      args: ["echo "]
+      args: ["echo result was: {{inputs.parameters.message}}"]
 `
 
 func TestWithParamAsJsonList(t *testing.T) {

workflow/controller/workflowpod.go

@@ -85,7 +85,7 @@ func (woc *wfOperationCtx) getVolumeDockerSock() apiv1.Volume {
 	}
 }
 
-func (woc *wfOperationCtx) createWorkflowPod(nodeName string, mainCtr apiv1.Container, tmpl *wfv1.Template) (*apiv1.Pod, error) {
+func (woc *wfOperationCtx) createWorkflowPod(nodeName string, mainCtr apiv1.Container, tmpl *wfv1.Template, includeScriptOutput bool) (*apiv1.Pod, error) {
 	nodeID := woc.wf.NodeID(nodeName)
 	woc.log.Debugf("Creating Pod: %s (%s)", nodeName, nodeID)
 	tmpl = tmpl.DeepCopy()
@@ -168,7 +168,7 @@ func (woc *wfOperationCtx) createWorkflowPod(nodeName string, mainCtr apiv1.Cont
 	}
 
 	addSchedulingConstraints(pod, wfSpec, tmpl)
-	woc.addMetadata(pod, tmpl)
+	woc.addMetadata(pod, tmpl, includeScriptOutput)
 
 	err = addVolumeReferences(pod, woc.volumes, tmpl, woc.wf.Status.PersistentVolumeClaims)
 	if err != nil {
@@ -446,21 +446,28 @@ func isResourcesSpecified(ctr *apiv1.Container) bool {
 }
 
 // addMetadata applies metadata specified in the template
-func (woc *wfOperationCtx) addMetadata(pod *apiv1.Pod, tmpl *wfv1.Template) {
+func (woc *wfOperationCtx) addMetadata(pod *apiv1.Pod, tmpl *wfv1.Template, includeScriptOutput bool) {
 	for k, v := range tmpl.Metadata.Annotations {
 		pod.ObjectMeta.Annotations[k] = v
 	}
 	for k, v := range tmpl.Metadata.Labels {
 		pod.ObjectMeta.Labels[k] = v
 	}
+
+	execCtl := common.ExecutionControl{
+		IncludeScriptOutput: includeScriptOutput,
+	}
+
 	if woc.workflowDeadline != nil {
-		execCtl := common.ExecutionControl{
-			Deadline: woc.workflowDeadline,
-		}
+		execCtl.Deadline = woc.workflowDeadline
+
+	}
+	if woc.workflowDeadline != nil || includeScriptOutput {
 		execCtlBytes, err := json.Marshal(execCtl)
 		if err != nil {
 			panic(err)
 		}
+
 		pod.ObjectMeta.Annotations[common.AnnotationKeyExecutionControl] = string(execCtlBytes)
 	}
 }

workflow/executor/executor.go

@@ -732,6 +732,11 @@ func (we *WorkflowExecutor) GetMainContainerID() (string, error) {
 
 // CaptureScriptResult will add the stdout of a script template as output result
 func (we *WorkflowExecutor) CaptureScriptResult() error {
+
+	if we.ExecutionControl == nil || !we.ExecutionControl.IncludeScriptOutput {
+		log.Infof("No Script output reference in workflow. Capturing script output ignored")
+		return nil
+	}
 	if we.Template.Script == nil {
 		return nil
 	}