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 {