Added status of previous steps as variables (argoproj#1681)

docs/variables.md

@@ -13,17 +13,19 @@ The following variables are made available to reference various metadata of a wo
 | Variable | Description|
 |----------|------------|
 | `steps.<STEPNAME>.ip` | IP address of a previous daemon container step |
-| `steps.<STEPNAME>.outputs.result` | Output result of a previous script step |
-| `steps.<STEPNAME>.outputs.parameters.<NAME>` | Output parameter of a previous step |
-| `steps.<STEPNAME>.outputs.artifacts.<NAME>` | Output artifact of a previous step |
+| `steps.<STEPNAME>.status` | Phase status of any previous script step |
+| `steps.<STEPNAME>.outputs.result` | Output result of any previous script step |
+| `steps.<STEPNAME>.outputs.parameters.<NAME>` | Output parameter of any previous step |
+| `steps.<STEPNAME>.outputs.artifacts.<NAME>` | Output artifact of any previous step |
 
 ## DAG Templates:
 | Variable | Description|
 |----------|------------|
 | `tasks.<TASKNAME>.ip` | IP address of a previous daemon container task |
-| `tasks.<TASKNAME>.outputs.result` | Output result of a previous script task |
-| `tasks.<TASKNAME>.outputs.parameters.<NAME>` | Output parameter of a previous task |
-| `tasks.<TASKNAME>.outputs.artifacts.<NAME>` | Output artifact of a previous task |
+| `tasks.<STEPNAME>.status` | Phase status of any previous task step |
+| `tasks.<TASKNAME>.outputs.result` | Output result of any previous script task |
+| `tasks.<TASKNAME>.outputs.parameters.<NAME>` | Output parameter of any previous task |
+| `tasks.<TASKNAME>.outputs.artifacts.<NAME>` | Output artifact of any previous task |
 
 ## Container/Script Templates:
 | Variable | Description|

examples/status-reference.yaml

@@ -0,0 +1,41 @@
+# The status reference example combines the use of a status result,
+# along with conditionals, to take a dynamic path in the
+# workflow. In this example, depending on the status of 'flakey-container'
+# the template will either run the 'succeeded' step or the 'failed' step.
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  generateName: status-reference-
+spec:
+  entrypoint: status-reference
+  templates:
+    - name: status-reference
+      steps:
+        - - name: flakey-container
+            template: flakey-container
+            continueOn:
+              failed: true
+        - - name: failed
+            template: failed
+            when: "{{steps.flakey-container.status}} == Failed"
+          - name: succeeded
+            template: succeeded
+            when: "{{steps.flakey-container.status}} == Succeeded"
+
+    - name: flakey-container
+      script:
+        image: alpine:3.6
+        command: [sh, -c]
+        args: ["exit 1"]
+
+    - name: failed
+      container:
+        image: alpine:3.6
+        command: [sh, -c]
+        args: ["echo \"the flakey container failed\""]
+
+    - name: succeeded
+      container:
+        image: alpine:3.6
+        command: [sh, -c]
+        args: ["echo \"the flakey container passed\""]

workflow/controller/operator.go

@@ -1602,6 +1602,10 @@ func (woc *wfOperationCtx) processNodeOutputs(scope *wfScope, prefix string, nod
 		key := fmt.Sprintf("%s.ip", prefix)
 		scope.addParamToScope(key, node.PodIP)
 	}
+	if node.Phase != "" {
+		key := fmt.Sprintf("%s.status", prefix)
+		scope.addParamToScope(key, string(node.Phase))
+	}
 	woc.addOutputsToScope(prefix, node.Outputs, scope)
 }
 

workflow/controller/operator_test.go

@@ -6,11 +6,11 @@ import (
 	"strings"
 	"testing"
 
-	"sigs.k8s.io/yaml"
 	"github.com/stretchr/testify/assert"
 	apiv1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"sigs.k8s.io/yaml"
 
 	wfv1 "github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
 	"github.com/argoproj/argo/test"
@@ -1102,6 +1102,60 @@ func TestResolvePlaceholdersInOutputValues(t *testing.T) {
 	assert.Equal(t, "output-value-placeholders-wf", *parameterValue)
 }
 
+var outputStatuses = `
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  generateName: scripts-bash-
+spec:
+  entrypoint: bash-script-example
+  templates:
+  - name: bash-script-example
+    steps:
+    - - name: first
+        template: flakey-container
+        continueOn:
+          failed: true
+    - - name: print
+        template: print-message
+        arguments:
+          parameters:
+          - name: message
+            value: "{{steps.first.status}}"
+
+
+  - name: flakey-container
+    script:
+      image: busybox
+      command: [sh, -c]
+      args: ["exit 0"]
+
+  - name: print-message
+    inputs:
+      parameters:
+      - name: message
+    container:
+      image: alpine:latest
+      command: [sh, -c]
+      args: ["echo result was: {{inputs.parameters.message}}"]
+`
+
+func TestResolveStatuses(t *testing.T) {
+
+	controller := newController()
+	wfcset := controller.wfclientset.ArgoprojV1alpha1().Workflows("")
+
+	// operate the workflow. it should create a pod.
+	wf := unmarshalWF(outputStatuses)
+	wf, err := wfcset.Create(wf)
+	assert.Nil(t, err)
+	jsonValue, err := json.Marshal(&wf.Spec.Templates[0])
+	assert.NoError(t, err)
+
+	assert.Contains(t, string(jsonValue), "{{steps.first.status}}")
+	assert.NotContains(t, string(jsonValue), "{{steps.print.status}}")
+}
+
 var resourceTemplate = `
 apiVersion: argoproj.io/v1alpha1
 kind: Workflow

workflow/validate/validate.go

@@ -556,6 +556,7 @@ func (ctx *templateValidationCtx) validateSteps(scope map[string]interface{}, tm
 			}
 			stepNames[step.Name] = true
 			prefix := fmt.Sprintf("steps.%s", step.Name)
+			scope[fmt.Sprintf("%s.status", prefix)] = true
 			err := addItemsToScope(prefix, step.WithItems, step.WithParam, step.WithSequence, scope)
 			if err != nil {
 				return errors.Errorf(errors.CodeBadRequest, "templates.%s.steps[%d].%s %s", tmpl.Name, i, step.Name, err.Error())
@@ -581,7 +582,7 @@ func (ctx *templateValidationCtx) validateSteps(scope map[string]interface{}, tm
 		for i, step := range stepGroup {
 			aggregate := len(step.WithItems) > 0 || step.WithParam != ""
 			resolvedTmpl := resolvedTemplates[step.Name]
-			ctx.addOutputsToScope(resolvedTmpl, fmt.Sprintf("steps.%s", step.Name), scope, aggregate)
+			ctx.addOutputsToScope(resolvedTmpl, fmt.Sprintf("steps.%s", step.Name), scope, aggregate, false)
 
 			// Validate the template again with actual arguments.
 			_, err = ctx.validateTemplateHolder(&step, tmplCtx, &step.Arguments, scope)
@@ -634,7 +635,7 @@ func addItemsToScope(prefix string, withItems []wfv1.Item, withParam string, wit
 	return nil
 }
 
-func (ctx *templateValidationCtx) addOutputsToScope(tmpl *wfv1.Template, prefix string, scope map[string]interface{}, aggregate bool) {
+func (ctx *templateValidationCtx) addOutputsToScope(tmpl *wfv1.Template, prefix string, scope map[string]interface{}, aggregate bool, isAncestor bool) {
 	if tmpl.Daemon != nil && *tmpl.Daemon {
 		scope[fmt.Sprintf("%s.ip", prefix)] = true
 	}
@@ -665,6 +666,9 @@ func (ctx *templateValidationCtx) addOutputsToScope(tmpl *wfv1.Template, prefix
 			scope[fmt.Sprintf("%s.outputs.parameters", prefix)] = true
 		}
 	}
+	if isAncestor {
+		scope[fmt.Sprintf("%s.status", prefix)] = true
+	}
 }
 
 func validateOutputs(scope map[string]interface{}, tmpl *wfv1.Template) error {
@@ -883,7 +887,7 @@ func (ctx *templateValidationCtx) validateDAG(scope map[string]interface{}, tmpl
 			return errors.Errorf(errors.CodeBadRequest, "templates.%s.tasks.%s %s", tmpl.Name, task.Name, err.Error())
 		}
 		prefix := fmt.Sprintf("tasks.%s", task.Name)
-		ctx.addOutputsToScope(resolvedTmpl, prefix, scope, false)
+		ctx.addOutputsToScope(resolvedTmpl, prefix, scope, false, false)
 		resolvedTemplates[task.Name] = resolvedTmpl
 		dupDependencies := make(map[string]bool)
 		for j, depName := range task.Dependencies {
@@ -917,7 +921,7 @@ func (ctx *templateValidationCtx) validateDAG(scope map[string]interface{}, tmpl
 		resolvedTmpl := resolvedTemplates[task.Name]
 		// add all tasks outputs to scope so that a nested DAGs can have outputs
 		prefix := fmt.Sprintf("tasks.%s", task.Name)
-		ctx.addOutputsToScope(resolvedTmpl, prefix, scope, false)
+		ctx.addOutputsToScope(resolvedTmpl, prefix, scope, false, false)
 		taskBytes, err := json.Marshal(task)
 		if err != nil {
 			return errors.InternalWrapError(err)
@@ -932,7 +936,7 @@ func (ctx *templateValidationCtx) validateDAG(scope map[string]interface{}, tmpl
 			resolvedTmpl := resolvedTemplates[ancestor]
 			ancestorPrefix := fmt.Sprintf("tasks.%s", ancestor)
 			aggregate := len(ancestorTask.WithItems) > 0 || ancestorTask.WithParam != ""
-			ctx.addOutputsToScope(resolvedTmpl, ancestorPrefix, taskScope, aggregate)
+			ctx.addOutputsToScope(resolvedTmpl, ancestorPrefix, taskScope, aggregate, true)
 		}
 		err = addItemsToScope(prefix, task.WithItems, task.WithParam, task.WithSequence, taskScope)
 		if err != nil {