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

cmd/argoexec/commands/wait.go

@@ -1,6 +1,10 @@
 package commands
 
 import (
+	"github.com/argoproj/argo/workflow/common"
+	"github.com/argoproj/argo/workflow/executor"
+	"strconv"
+	"strings"
 	"time"
 
 	"github.com/argoproj/pkg/stats"
@@ -28,8 +32,18 @@ func waitContainer() error {
 	defer stats.LogStats()
 	stats.StartStatsTicker(5 * time.Minute)
 
+	includeOutput, err := executor.GetAnnotationField(wfExecutor.PodAnnotationsPath, common.AnnotationIncludeOutputs)
+	if err != nil {
+		wfExecutor.AddError(err)
+	}
+
+	includeOutputFlag, err := strconv.ParseBool(strings.Trim(includeOutput, "\""))
+
+	if err != nil {
+		wfExecutor.AddError(err)
+	}
 	// Wait for main container to complete
-	err := wfExecutor.Wait()
+	err = wfExecutor.Wait()
 	if err != nil {
 		wfExecutor.AddError(err)
 		// do not return here so we can still try to kill sidecars & save outputs
@@ -56,16 +70,20 @@ func waitContainer() error {
 		wfExecutor.AddError(err)
 		return err
 	}
-	// Capture output script result
-	err = wfExecutor.CaptureScriptResult()
-	if err != nil {
-		wfExecutor.AddError(err)
-		return err
-	}
-	err = wfExecutor.AnnotateOutputs(logArt)
-	if err != nil {
-		wfExecutor.AddError(err)
-		return err
+
+	if includeOutputFlag {
+		// Capture output script result
+		err = wfExecutor.CaptureScriptResult()
+		if err != nil {
+			wfExecutor.AddError(err)
+			return err
+		}
+
+		err = wfExecutor.AnnotateOutputs(logArt)
+		if err != nil {
+			wfExecutor.AddError(err)
+			return err
+		}
 	}
 	return nil
 }

workflow/common/common.go

@@ -34,6 +34,10 @@ const (
 
 	// AnnotationKeyNodeName is the pod metadata annotation key containing the workflow node name
 	AnnotationKeyNodeName = workflow.FullName + "/node-name"
+
+	// AnnotationIncludeOutputs is the pod metadata annotation key containing flag to include script output
+	AnnotationIncludeOutputs = workflow.FullName + "/include-output"
+
 	// 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"

workflow/controller/operator.go

@@ -1290,7 +1290,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())
 	}
@@ -1362,14 +1362,83 @@ func getTemplateOutputsFromScope(tmpl *wfv1.Template, scope *wfScope) (*wfv1.Out
 	return &outputs, nil
 }
 
+func HasOutputResultRef(name string, parentTmpl *wfv1.Template) bool {
+
+	nodeName := "." + name + "."
+	fmt.Println(nodeName)
+
+	if parentTmpl.DAG != nil {
+		for _, dagTask := range parentTmpl.DAG.Tasks {
+
+			if strings.Contains(dagTask.When, nodeName) {
+				return true
+			}
+
+			for _, artf := range dagTask.Arguments.Parameters {
+				if strings.Contains(*artf.Value, "{{") && (strings.Contains(*artf.Value, nodeName) || (artf.ValueFrom != nil && strings.Contains(artf.ValueFrom.Parameter, nodeName))) {
+					return true
+				}
+			}
+		}
+	}
+
+	if parentTmpl.Steps != nil {
+		for idx := range parentTmpl.Steps {
+			stepGroup := parentTmpl.Steps[idx]
+			for _, step := range stepGroup {
+
+				if strings.Contains(step.When, nodeName) {
+					return true
+				}
+				for _, artf := range step.Arguments.Parameters {
+					if strings.Contains(*artf.Value, "{{") && (strings.Contains(*artf.Value, nodeName) || (artf.ValueFrom != nil && strings.Contains(artf.ValueFrom.Parameter, nodeName))) {
+						return true
+					}
+				}
+			}
+		}
+	}
+
+	for _, param := range parentTmpl.Outputs.Parameters {
+		if strings.Contains(*param.Value, "{{") && (strings.Contains(*param.Value, nodeName) || (param.ValueFrom != nil && strings.Contains(param.ValueFrom.Parameter, nodeName))) {
+			return true
+		}
+	}
+
+	return false
+}
+
+func GetStepOrDAGTaskName(nodeName string) string {
+	if strings.Contains(nodeName, ".") {
+		name := nodeName[strings.LastIndex(nodeName, ".")+1:]
+		// Check retry scenario
+		if strings.Contains(name, "(") {
+			return name[0:strings.Index(name, "(")]
+		}
+		return nodeName[strings.LastIndex(nodeName, ".")+1:]
+	}
+	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)
+		includeScriptOutput = HasOutputResultRef(name, parentTemplate)
+	}
+	fmt.Println(includeScriptOutput)
 	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())
 	}
@@ -1604,7 +1673,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

@@ -1159,3 +1159,73 @@ func TestResourceWithOwnerReferenceTemplate(t *testing.T) {
 		assert.Equal(t, "resource-with-ownerreference-template", objectMetas["resource-cm-3"].OwnerReferences[1].Name)
 	}
 }
+
+var stepScriptTmpl = `
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  name: resource-with-ownerreference-template
+spec:
+  entrypoint: start
+  templates:
+  - name: start
+    steps:
+    - - name: resource-1
+        template: resource-1
+      - name: resource-2
+        template: resource-2	
+        arguments:
+           parameters: [{name: message, value: "{{steps.resource-1.output.result}}"}]
+      - name: resource-3
+        template: resource-3 `
+
+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
+        dependencies: [A]
+        template: echo
+        arguments:
+          parameters: [{name: message, value: B}]
+      - name: C
+        dependencies: [A]
+        template: echo
+        arguments:
+          parameters: [{name: message, value: "{{dag.A.output.result}}"}]
+      - name: D
+        dependencies: [B, C]
+        template: echo
+        arguments:
+          parameters: [{name: message, value: D}]`
+
+func TestGetNodeName(t *testing.T) {
+
+	// Steps Workflow
+	wf := unmarshalWF(stepScriptTmpl)
+
+	assert.True(t, HasOutputResultRef("resource-1", &wf.Spec.Templates[0]))
+	assert.False(t, HasOutputResultRef("resource-2", &wf.Spec.Templates[0]))
+
+	// DAG workflow
+	wf = unmarshalWF(dagScriptTmpl)
+
+	assert.True(t, HasOutputResultRef("A", &wf.Spec.Templates[0]))
+	assert.False(t, HasOutputResultRef("B", &wf.Spec.Templates[0]))
+}

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()
@@ -105,7 +105,8 @@ func (woc *wfOperationCtx) createWorkflowPod(nodeName string, mainCtr apiv1.Cont
 				common.LabelKeyCompleted: "false",                // Allows filtering by incomplete workflow pods
 			},
 			Annotations: map[string]string{
-				common.AnnotationKeyNodeName: nodeName,
+				common.AnnotationKeyNodeName:    nodeName,
+				common.AnnotationIncludeOutputs: strconv.FormatBool(includeScriptOutput),
 			},
 			OwnerReferences: []metav1.OwnerReference{
 				*metav1.NewControllerRef(woc.wf, wfv1.SchemaGroupVersionKind),

workflow/executor/executor.go

@@ -69,6 +69,8 @@ type WorkflowExecutor struct {
 	// list of errors that occurred during execution.
 	// the first of these is used as the overall message of the node
 	errors []error
+
+	ExcludeOutput bool
 }
 
 // ContainerRuntimeExecutor is the interface for interacting with a container runtime (e.g. docker)
@@ -1101,6 +1103,58 @@ func LoadTemplate(path string) (*wfv1.Template, error) {
 	return &tmpl, nil
 }
 
+func GetAnnotationField(filePath string, key string) (string, error) {
+	// Read the annotation file
+	file, err := os.Open(filePath)
+	if err != nil {
+		log.Errorf("ERROR opening annotation file from %s", filePath)
+		return "", errors.InternalWrapError(err)
+	}
+	content := ""
+	defer func() {
+		_ = file.Close()
+	}()
+	reader := bufio.NewReader(file)
+
+	// Prefix of key property in the annotation file
+	prefix := fmt.Sprintf("%s=", key)
+
+	for {
+		// Read line-by-line
+		var buffer bytes.Buffer
+		var l []byte
+		var isPrefix bool
+		for {
+			l, isPrefix, err = reader.ReadLine()
+			buffer.Write(l)
+			// If we've reached the end of the line, stop reading.
+			if !isPrefix {
+				break
+			}
+			// If we're just at the EOF, break
+			if err != nil {
+				break
+			}
+		}
+
+		line := buffer.String()
+
+		// Read property
+		if strings.HasPrefix(line, prefix) {
+			// Trim the prefix
+			content = strings.TrimPrefix(line, prefix)
+			break
+		}
+
+		// The end of the annotation file
+		if err == io.EOF {
+			break
+		}
+	}
+
+	return content, nil
+}
+
 // unmarshalAnnotationField unmarshals the value of an annotation key into the supplied interface
 // from the downward api annotation volume file
 func unmarshalAnnotationField(filePath string, key string, into interface{}) error {