Add support for from usage in Pipeline Conditions

Resources in Pipeline Conditions can now declare that they depend on the
output of previous tasks using the `from` clause. Using `from` in a
conditional resource implies ordering for the pipeline task i.e. if
task B has a condition (say, C) that takes in an output resource from
task A, task A will run first, followed by the conditional C, and then B

Signed-off-by: Dibyo Mukherjee <[email protected]>

docs/pipelines.md

@@ -325,6 +325,31 @@ tasks:
 In this example, `my-condition` refers to a [Condition](#conditions) custom resource. The `build-push` 
 task will only be executed if the condition evaluates to true. 
 
+Resources in conditions can also use the [`from`](#from) field to indicate that they 
+expect the output of a previous task as input. As with regular Pipeline Tasks, using `from`
+implies ordering --  if task has a condition that takes in an output resource from
+another task, the task producing the output resource will run first:
+
+```yaml
+tasks:
+  - name: first-create-file
+    taskRef:
+      name: create-file
+    resources:
+      outputs:
+        - name: workspace
+          resource: source-repo
+  - name: then-check
+    conditions:
+      - conditionRef: "file-exists"
+        resources:
+          - name: workspace
+            resource: source-repo
+            from: [first-create-file]
+    taskRef:
+      name: echo-hello
+```
+
 ## Ordering
 
 The [Pipeline Tasks](#pipeline-tasks) in a `Pipeline` can be connected and run

examples/pipelineruns/conditional-pipelinerun.yaml

@@ -27,21 +27,31 @@ spec:
 apiVersion: tekton.dev/v1alpha1
 kind: Task
 metadata:
-  name: list-files
+  name: create-readme-file
 spec:
-  inputs:
+  outputs:
     resources:
       - name: workspace
         type: git
   steps:
-    - name: run-ls
+    - name: write-new-stuff
       image: ubuntu
-      script: 'ls -al $(inputs.resources.workspace.path)'
+      script: 'touch $(outputs.resources.workspace.path)/README.md'
+---
+apiVersion: tekton.dev/v1alpha1
+kind: Task
+metadata:
+  name: echo-hello
+spec:
+  steps:
+    - name: echo
+      image: ubuntu
+      script: 'echo hello'
 ---
 apiVersion: tekton.dev/v1alpha1
 kind: Pipeline
 metadata:
-  name: list-files-pipeline
+  name: conditional-pipeline
 spec:
   resources:
     - name: source-repo
@@ -50,9 +60,14 @@ spec:
     - name: "path"
       default: "README.md"
   tasks:
-    - name: list-files-1
+    - name: first-create-file
       taskRef:
-        name: list-files
+        name: create-readme-file
+      resources:
+        outputs:
+          - name: workspace
+            resource: source-repo
+    - name: then-check
       conditions:
         - conditionRef: "file-exists"
           params:
@@ -61,18 +76,17 @@ spec:
           resources:
             - name: workspace
               resource: source-repo
-      resources:
-        inputs:
-          - name: workspace
-            resource: source-repo
+              from: [first-create-file]
+      taskRef:
+        name: echo-hello
 ---
 apiVersion: tekton.dev/v1alpha1
 kind: PipelineRun
 metadata:
-  name: demo-condtional-pr
+  name: condtional-pr
 spec:
   pipelineRef:
-    name: list-files-pipeline
+    name: conditional-pipeline
   serviceAccountName: 'default'
   resources:
     - name: source-repo

pkg/apis/pipeline/v1alpha1/pipeline_types.go

@@ -130,6 +130,12 @@ func (pt PipelineTask) Deps() []string {
 			deps = append(deps, rd.From...)
 		}
 	}
+	// Add any dependents from conditional resources.
+	for _, cond := range pt.Conditions {
+		for _, rd := range cond.Resources {
+			deps = append(deps, rd.From...)
+		}
+	}
 	return deps
 }
 
@@ -160,7 +166,7 @@ type PipelineTaskCondition struct {
 	Params []Param `json:"params,omitempty"`
 
 	// Resources declare the resources provided to this Condition as input
-	Resources []PipelineConditionResource `json:"resources,omitempty"`
+	Resources []PipelineTaskInputResource `json:"resources,omitempty"`
 }
 
 // PipelineDeclaredResource is used by a Pipeline to declare the types of the
@@ -176,15 +182,6 @@ type PipelineDeclaredResource struct {
 	Type PipelineResourceType `json:"type"`
 }
 
-// PipelineConditionResource allows a Pipeline to declare how its DeclaredPipelineResources
-// should be provided to a Condition as its inputs.
-type PipelineConditionResource struct {
-	// Name is the name of the PipelineResource as declared by the Condition.
-	Name string `json:"name"`
-	// Resource is the name of the DeclaredPipelineResource to use.
-	Resource string `json:"resource"`
-}
-
 // PipelineTaskResources allows a Pipeline to declare how its DeclaredPipelineResources
 // should be provided to a Task as its inputs and outputs.
 type PipelineTaskResources struct {

pkg/apis/pipeline/v1alpha1/pipeline_validation.go

@@ -66,7 +66,7 @@ func validateDeclaredResources(ps *PipelineSpec) error {
 	}
 	missing := list.DiffLeft(required, provided)
 	if len(missing) > 0 {
-		return fmt.Errorf("Pipeline declared resources didn't match usage in Tasks: Didn't provide required values: %s", missing)
+		return fmt.Errorf("pipeline declared resources didn't match usage in Tasks: Didn't provide required values: %s", missing)
 	}
 	return nil
 }
@@ -93,16 +93,23 @@ func validateFrom(tasks []PipelineTask) error {
 		taskOutputs[task.Name] = to
 	}
 	for _, t := range tasks {
+		inputResources := []PipelineTaskInputResource{}
 		if t.Resources != nil {
-			for _, rd := range t.Resources.Inputs {
-				for _, pb := range rd.From {
-					outputs, found := taskOutputs[pb]
-					if !found {
-						return fmt.Errorf("expected resource %s to be from task %s, but task %s doesn't exist", rd.Resource, pb, pb)
-					}
-					if !isOutput(outputs, rd.Resource) {
-						return fmt.Errorf("the resource %s from %s must be an output but is an input", rd.Resource, pb)
-					}
+			inputResources = append(inputResources, t.Resources.Inputs...)
+		}
+
+		for _, c := range t.Conditions {
+			inputResources = append(inputResources, c.Resources...)
+		}
+
+		for _, rd := range inputResources {
+			for _, pt := range rd.From {
+				outputs, found := taskOutputs[pt]
+				if !found {
+					return fmt.Errorf("expected resource %s to be from task %s, but task %s doesn't exist", rd.Resource, pt, pt)
+				}
+				if !isOutput(outputs, rd.Resource) {
+					return fmt.Errorf("the resource %s from %s must be an output but is an input", rd.Resource, pt)
 				}
 			}
 		}

pkg/apis/pipeline/v1alpha1/pipeline_validation_test.go

@@ -132,7 +132,9 @@ func TestPipelineSpec_Validate(t *testing.T) {
 				tb.PipelineTaskCondition("some-condition",
 					tb.PipelineTaskConditionResource("some-workspace", "great-resource"))),
 			tb.PipelineTask("foo", "foo-task",
-				tb.PipelineTaskInputResource("wow-image", "wonderful-resource", tb.From("bar"))),
+				tb.PipelineTaskInputResource("wow-image", "wonderful-resource", tb.From("bar")),
+				tb.PipelineTaskCondition("some-condition-2",
+					tb.PipelineTaskConditionResource("wow-image", "wonderful-resource", "bar"))),
 		)),
 		failureExpected: false,
 	}, {
@@ -220,6 +222,15 @@ func TestPipelineSpec_Validate(t *testing.T) {
 					tb.PipelineTaskConditionResource("some-workspace", "missing-resource"))),
 		)),
 		failureExpected: true,
+	}, {
+		name: "invalid from in condition",
+		p: tb.Pipeline("pipeline", "namespace", tb.PipelineSpec(
+			tb.PipelineTask("foo", "foo-task"),
+			tb.PipelineTask("bar", "bar-task",
+				tb.PipelineTaskCondition("some-condition",
+					tb.PipelineTaskConditionResource("some-workspace", "missing-resource", "foo"))),
+		)),
+		failureExpected: true,
 	}, {
 		name: "from resource isn't output by task",
 		p: tb.Pipeline("pipeline", "namespace", tb.PipelineSpec(

pkg/reconciler/pipeline/dag/dag_test.go

@@ -440,6 +440,14 @@ func TestBuild_Invalid(t *testing.T) {
 		RunAfter: []string{"none"},
 	}
 
+	invalidConditionalTask := v1alpha1.PipelineTask{
+		Name: "b",
+		Conditions: []v1alpha1.PipelineTaskCondition{{
+			ConditionRef: "some-condition",
+			Resources:    []v1alpha1.PipelineTaskInputResource{{From: []string{"none"}}},
+		}},
+	}
+
 	tcs := []struct {
 		name string
 		spec v1alpha1.PipelineSpec
@@ -467,6 +475,9 @@ func TestBuild_Invalid(t *testing.T) {
 	}, {
 		name: "invalid-task-name-after",
 		spec: v1alpha1.PipelineSpec{Tasks: []v1alpha1.PipelineTask{invalidTaskAfter}},
+	}, {
+		name: "invalid-task-name-from-conditional",
+		spec: v1alpha1.PipelineSpec{Tasks: []v1alpha1.PipelineTask{invalidConditionalTask}},
 	},
 	}
 	for _, tc := range tcs {
@@ -481,3 +492,82 @@ func TestBuild_Invalid(t *testing.T) {
 		})
 	}
 }
+
+func TestBuild_ConditionResources(t *testing.T) {
+	// a,b, c are regular tasks
+	a := v1alpha1.PipelineTask{Name: "a"}
+	b := v1alpha1.PipelineTask{Name: "b"}
+	c := v1alpha1.PipelineTask{Name: "c"}
+
+	// Condition that depends on Task a output
+	cond1DependsOnA := v1alpha1.PipelineTaskCondition{
+		Resources: []v1alpha1.PipelineTaskInputResource{{From: []string{"a"}}},
+	}
+	// Condition that depends on Task b output
+	cond2DependsOnB := v1alpha1.PipelineTaskCondition{
+		Resources: []v1alpha1.PipelineTaskInputResource{{From: []string{"b"}}},
+	}
+
+	// x indirectly depends on A,B via its conditions
+	xDependsOnAAndB := v1alpha1.PipelineTask{
+		Name:       "x",
+		Conditions: []v1alpha1.PipelineTaskCondition{cond1DependsOnA, cond2DependsOnB},
+	}
+
+	// y depends on a both directly + via its conditional
+	yDependsOnA := v1alpha1.PipelineTask{
+		Name: "y",
+		Resources: &v1alpha1.PipelineTaskResources{
+			Inputs: []v1alpha1.PipelineTaskInputResource{{From: []string{"a"}}},
+		},
+		Conditions: []v1alpha1.PipelineTaskCondition{cond1DependsOnA},
+	}
+
+	// y depends on b both directly + via its conditional
+	zDependsOnBRunsAfterC := v1alpha1.PipelineTask{
+		Name:       "z",
+		RunAfter:   []string{"c"},
+		Conditions: []v1alpha1.PipelineTaskCondition{cond2DependsOnB},
+	}
+
+	//  a   b   c
+	// / \ / \ /
+	// y  x   z
+	nodeA := &dag.Node{Task: a}
+	nodeB := &dag.Node{Task: b}
+	nodeC := &dag.Node{Task: c}
+	nodeX := &dag.Node{Task: xDependsOnAAndB}
+	nodeY := &dag.Node{Task: yDependsOnA}
+	nodeZ := &dag.Node{Task: zDependsOnBRunsAfterC}
+
+	nodeA.Next = []*dag.Node{nodeX, nodeY}
+	nodeB.Next = []*dag.Node{nodeX, nodeZ}
+	nodeC.Next = []*dag.Node{nodeZ}
+	nodeX.Prev = []*dag.Node{nodeA, nodeB}
+	nodeY.Prev = []*dag.Node{nodeA}
+	nodeZ.Prev = []*dag.Node{nodeB, nodeC}
+
+	expectedDAG := &dag.Graph{
+		Nodes: map[string]*dag.Node{
+			"a": nodeA,
+			"b": nodeB,
+			"c": nodeC,
+			"x": nodeX,
+			"y": nodeY,
+			"z": nodeZ,
+		},
+	}
+
+	p := &v1alpha1.Pipeline{
+		ObjectMeta: metav1.ObjectMeta{Name: "pipeline"},
+		Spec: v1alpha1.PipelineSpec{
+			Tasks: []v1alpha1.PipelineTask{a, b, c, xDependsOnAAndB, yDependsOnA, zDependsOnBRunsAfterC},
+		},
+	}
+
+	g, err := dag.Build(v1alpha1.PipelineTaskList(p.Spec.Tasks))
+	if err != nil {
+		t.Errorf("didn't expect error creating valid Pipeline %v but got %v", p, err)
+	}
+	assertSameDAG(t, expectedDAG, g)
+}

pkg/reconciler/pipelinerun/resources/conditionresolution.go

@@ -130,7 +130,8 @@ func convertParamTemplates(step *v1alpha1.Step, params []v1alpha1.ParamSpec) {
 	v1alpha1.ApplyStepReplacements(step, replacements, map[string][]string{})
 }
 
-// ApplyResourceSubstitution applies resource attribute variable substitution.
+// ApplyResources applies the substitution from values in resources which are referenced
+// in spec as subitems of the replacementStr.
 func ApplyResourceSubstitution(step *v1alpha1.Step, resolvedResources map[string]*v1alpha1.PipelineResource, conditionResources []v1alpha1.ResourceDeclaration, images pipeline.Images) error {
 	replacements := make(map[string]string)
 	for _, cr := range conditionResources {