Implement Skipping

When `WhenExpressions` evaluate to False, the guarded `Task` and
its branch (dependent `Tasks`) are skipped

A `Task` is dependent on and in the branch of another `Task` as specified
by ordering using `runAfter` or by resources using `Results`, `Workspaces`
 and `Resources`

In some use cases of `Conditions`, when a `Condition` evaluates to `False`,
users need to skip the guarded `Task` only and allow ordering-dependent
`Tasks` to execute

When  `WhenExpressions` evaluate to `False`, it is now possible to allow
for execution of ordering-dependent `Tasks` as specified by [`runAfter`](#using-the-runafter-parameter)
using the `continueAfterSkip` field by setting it to `true` or `yes`

`continueAfterSkip` is only supported in `Tasks` with `WhenExpressions`;
there will be a validation error if it is specified in `Tasks` without
`WhenExpressions`

docs/pipelines.md

@@ -352,6 +352,25 @@ tasks:
         name: echo-file-exists
 ```
 
+When  `WhenExpressions` evaluate to `False`, it is possible to allow for execution of ordering-dependent `Tasks` as specified by [`runAfter`](#using-the-runafter-parameter) using the `continueAfterSkip` field by setting it to `true` or `yes`. In this example, `task-should-be-skipped` will be skipped and `task-should-be-executed` will be executed.  
+
+```yaml
+tasks:
+  - name: task-should-be-skipped
+    when:
+      - input: "foo"
+        operator: in
+        values: ["bar"]
+    continueAfterSkip: 'true'
+    taskRef:
+        name: exit-1
+  - name: task-should-be-executed
+    runAfter:
+    - task-should-be-skipped
+    taskRef:
+        name: exit-0  
+```
+
 For an end-to-end example, see [PipelineRun with WhenExpressions](../examples/v1beta1/pipelineruns/pipelinerun-with-when-expressions.yaml).
 
 When `WhenExpressions` are specified in a `Task`, [`Conditions`](#guard-task-execution-using-conditions) should not be specified in the same `Task`. The `Pipeline` will be rejected as invalid if both `WhenExpressions` and `Conditions` are included.

examples/v1beta1/pipelineruns/pipelinerun-with-when-expressions.yaml

@@ -71,11 +71,24 @@ spec:
         - input: "$(params.path)"
           operator: notin
           values: ["README.md"]
+      continueAfterSkip: 'true'
       taskSpec:
         steps:
           - name: echo
             image: ubuntu
             script: exit 1
+    - name: task-should-be-executed-after-skipped-parent-task
+      runAfter:
+        - task-should-be-skipped 
+      when:
+        - input: "$(params.path)"
+          operator: in
+          values: ["README.md"]
+      taskSpec:
+        steps:
+          - name: echo
+            image: ubuntu
+            script: 'echo created README.md'
 ---
 apiVersion: tekton.dev/v1beta1
 kind: PipelineRun

internal/builder/v1beta1/pipeline.go

@@ -345,6 +345,14 @@ func PipelineTaskWhenExpression(input string, operator selection.Operator, value
 	}
 }
 
+// PipelineTaskContinueAfterSkip adds a string indicating whether the ordering-dependent Tasks, as specified by runAfter,
+// should be executed after the parent Task has been skipped due to its When Expressions evaluating to false.
+func PipelineTaskContinueAfterSkip(continueAfterSkip string) PipelineTaskOp {
+	return func(pt *v1beta1.PipelineTask) {
+		pt.ContinueAfterSkip = continueAfterSkip
+	}
+}
+
 // PipelineTaskWorkspaceBinding adds a workspace with the specified name, workspace and subpath on a PipelineTask.
 func PipelineTaskWorkspaceBinding(name, workspace, subPath string) PipelineTaskOp {
 	return func(pt *v1beta1.PipelineTask) {

pkg/apis/pipeline/v1beta1/pipeline_types.go

@@ -132,6 +132,10 @@ type PipelineTask struct {
 	// +optional
 	WhenExpressions WhenExpressions `json:"when,omitempty"`
 
+	// ContinueAfterSkip is a string used to indicate whether ordering-dependent tasks should be executed
+	// +optional
+	ContinueAfterSkip string `json:"continueAfterSkip,omitempty"`
+
 	// Retries represents how many times this task should be retried in case of task failure: ConditionSucceeded set to False
 	// +optional
 	Retries int `json:"retries,omitempty"`

pkg/apis/pipeline/v1beta1/pipeline_validation.go

@@ -519,6 +519,9 @@ func validateWhenExpressions(tasks []PipelineTask) *apis.FieldError {
 		if err := validateOneOfWhenExpressionsOrConditions(i, t); err != nil {
 			return err
 		}
+		if err := validateContinueAfterSkipInTasksWithWhenExpressionsOnly(i, t); err != nil {
+			return err
+		}
 		if err := t.WhenExpressions.validate(); err != nil {
 			return err
 		}
@@ -533,3 +536,10 @@ func validateOneOfWhenExpressionsOrConditions(i int, t PipelineTask) *apis.Field
 	}
 	return nil
 }
+
+func validateContinueAfterSkipInTasksWithWhenExpressionsOnly(i int, t PipelineTask) *apis.FieldError {
+	if t.ContinueAfterSkip != "" && t.WhenExpressions == nil {
+		return apis.ErrDisallowedFields("continueAfterSkip not allowed in tasks without WhenExpressions", fmt.Sprintf("spec.tasks[%d].continueAfterSkip", i))
+	}
+	return nil
+}

pkg/apis/pipeline/v1beta1/pipeline_validation_test.go

@@ -259,6 +259,19 @@ func TestPipelineSpec_Validate_Failure(t *testing.T) {
 				WhenExpressions: []WhenExpression{{}},
 			}},
 		},
+	}, {
+		name: "invalid pipeline with one pipeline task having continueAfterSkip without WhenExpressions",
+		ps: &PipelineSpec{
+			Description: "this is an invalid pipeline with invalid pipeline task",
+			Tasks: []PipelineTask{{
+				Name:    "valid-pipeline-task",
+				TaskRef: &TaskRef{Name: "foo-task"},
+			}, {
+				Name:              "invalid-pipeline-task",
+				TaskRef:           &TaskRef{Name: "foo-task"},
+				ContinueAfterSkip: "true",
+			}},
+		},
 	}, {
 		name: "invalid pipeline with pipeline task having reference to resources which does not exist",
 		ps: &PipelineSpec{

pkg/reconciler/pipelinerun/pipelinerun_test.go

@@ -2191,6 +2191,82 @@ func TestReconcileWithWhenExpressionsWithTaskResults(t *testing.T) {
 	}
 }
 
+func TestReconcileWithWhenExpressionsContinueAfterSkip(t *testing.T) {
+	names.TestingSeed()
+	ps := []*v1beta1.Pipeline{tb.Pipeline("test-pipeline", tb.PipelineNamespace("foo"), tb.PipelineSpec(
+		tb.PipelineTask("a-task", "a-task",
+			tb.PipelineTaskWhenExpression("foo", selection.In, []string{"bar"}),
+			tb.PipelineTaskContinueAfterSkip("true"),
+		),
+		tb.PipelineTask("b-task", "b-task"),
+	))}
+	prs := []*v1beta1.PipelineRun{tb.PipelineRun("test-pipeline-run-different-service-accs", tb.PipelineRunNamespace("foo"),
+		tb.PipelineRunSpec("test-pipeline",
+			tb.PipelineRunServiceAccountName("test-sa-0"),
+		),
+	)}
+	ts := []*v1beta1.Task{
+		tb.Task("a-task", tb.TaskNamespace("foo")),
+		tb.Task("b-task", tb.TaskNamespace("foo")),
+	}
+	trs := []*v1beta1.TaskRun{}
+
+	d := test.Data{
+		PipelineRuns: prs,
+		Pipelines:    ps,
+		Tasks:        ts,
+		TaskRuns:     trs,
+	}
+	prt := NewPipelineRunTest(d, t)
+	defer prt.Cancel()
+
+	wantEvents := []string{
+		"Normal Started",
+		"Normal Running Tasks Completed: 0 \\(Failed: 0, Cancelled 0\\), Incomplete: 1, Skipped: 1",
+	}
+	pipelineRun, clients := prt.reconcileRun("foo", "test-pipeline-run-different-service-accs", wantEvents, false)
+
+	expectedTaskRunName := "test-pipeline-run-different-service-accs-b-task-mz4c7"
+	expectedTaskRun := tb.TaskRun(expectedTaskRunName,
+		tb.TaskRunNamespace("foo"),
+		tb.TaskRunOwnerReference("PipelineRun", "test-pipeline-run-different-service-accs",
+			tb.OwnerReferenceAPIVersion("tekton.dev/v1beta1"),
+			tb.Controller, tb.BlockOwnerDeletion,
+		),
+		tb.TaskRunLabel("tekton.dev/pipeline", "test-pipeline"),
+		tb.TaskRunLabel("tekton.dev/pipelineRun", "test-pipeline-run-different-service-accs"),
+		tb.TaskRunLabel("tekton.dev/pipelineTask", "b-task"),
+		tb.TaskRunSpec(
+			tb.TaskRunTaskRef("b-task"),
+			tb.TaskRunServiceAccountName("test-sa-0"),
+		),
+	)
+	// Check that the expected TaskRun was created
+	actual, err := clients.Pipeline.TektonV1beta1().TaskRuns("foo").List(metav1.ListOptions{
+		LabelSelector: "tekton.dev/pipelineTask=b-task,tekton.dev/pipelineRun=test-pipeline-run-different-service-accs",
+		Limit:         1,
+	})
+
+	if err != nil {
+		t.Fatalf("Failure to list TaskRun's %s", err)
+	}
+	if len(actual.Items) != 1 {
+		t.Fatalf("Expected 1 TaskRuns got %d", len(actual.Items))
+	}
+	actualTaskRun := actual.Items[0]
+	if d := cmp.Diff(&actualTaskRun, expectedTaskRun, ignoreResourceVersion); d != "" {
+		t.Errorf("expected to see TaskRun %v created. Diff %s", expectedTaskRunName, diff.PrintWantGot(d))
+	}
+
+	actualSkippedTasks := pipelineRun.Status.SkippedTasks
+	expectedSkippedTasks := []v1beta1.SkippedTask{{
+		Name: "a-task",
+	}}
+	if d := cmp.Diff(actualSkippedTasks, expectedSkippedTasks); d != "" {
+		t.Errorf("expected to find Skipped Tasks %v. Diff %s", expectedSkippedTasks, diff.PrintWantGot(d))
+	}
+}
+
 // TestReconcileWithAffinityAssistantStatefulSet tests that given a pipelineRun with workspaces,
 // an Affinity Assistant StatefulSet is created for each PVC workspace and
 // that the Affinity Assistant names is propagated to TaskRuns.

pkg/reconciler/pipelinerun/resources/pipelinerunresolution.go

@@ -21,6 +21,7 @@ import (
 	"fmt"
 	"reflect"
 	"strconv"
+	"strings"
 
 	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
@@ -141,6 +142,23 @@ func (t ResolvedPipelineRunTask) IsStarted() bool {
 	return true
 }
 
+func (t *ResolvedPipelineRunTask) isOrderingDependent(parentName string) bool {
+	for _, orderingDep := range t.PipelineTask.RunAfter {
+		if orderingDep == parentName {
+			return true
+		}
+	}
+	return false
+}
+
+func (t *ResolvedPipelineRunTask) shouldContinueAfterSkip() bool {
+	trueMap := map[string]bool{"true": true, "yes": true, "y": true}
+	if _, ok := trueMap[strings.ToLower(t.PipelineTask.ContinueAfterSkip)]; ok {
+		return true
+	}
+	return false
+}
+
 // Skip returns true if a PipelineTask will not be run because
 // (1) its When Expressions evaluated to false
 // (2) its Condition Checks failed
@@ -180,8 +198,11 @@ func (t *ResolvedPipelineRunTask) Skip(state PipelineRunState, d *dag.Graph) boo
 	node := d.Nodes[t.PipelineTask.Name]
 	if isTaskInGraph(t.PipelineTask.Name, d) {
 		for _, p := range node.Prev {
-			if stateMap[p.Task.HashKey()].Skip(state, d) {
-				return true
+			parentTask := stateMap[p.Task.HashKey()]
+			if parentTask.Skip(state, d) {
+				if !t.isOrderingDependent(parentTask.PipelineTask.Name) || !parentTask.shouldContinueAfterSkip() {
+					return true
+				}
 			}
 		}
 	}

pkg/reconciler/pipelinerun/resources/pipelinerunresolution_test.go

@@ -95,6 +95,15 @@ var pts = []v1beta1.PipelineTask{{
 		Operator: selection.NotIn,
 		Values:   []string{"foo", "bar"},
 	}},
+}, {
+	Name:    "mytask12",
+	TaskRef: &v1beta1.TaskRef{Name: "taskWithWhenExpressions"},
+	WhenExpressions: []v1beta1.WhenExpression{{
+		Input:    "foo",
+		Operator: selection.NotIn,
+		Values:   []string{"foo", "bar"},
+	}},
+	ContinueAfterSkip: "true",
 }}
 
 var p = &v1beta1.Pipeline{
@@ -1175,6 +1184,52 @@ func TestIsSkipped(t *testing.T) {
 			},
 		}},
 		expected: true,
+	}, {
+		name:     "tasks-when-expression-skip-from-parent",
+		taskName: "mytask13",
+		state: PipelineRunState{{
+			PipelineTask: &pts[10],
+			TaskRunName:  "pipelinerun-guardedtask",
+			TaskRun:      nil,
+			ResolvedTaskResources: &resources.ResolvedTaskResources{
+				TaskSpec: &task.Spec,
+			},
+		}, {
+			PipelineTask: &v1beta1.PipelineTask{
+				Name:     "mytask13",
+				TaskRef:  &v1beta1.TaskRef{Name: "task"},
+				RunAfter: []string{"mytask11"},
+			}, // mytask13 runAfter mytask11
+			TaskRunName: "ordering-dependent-task",
+			TaskRun:     nil,
+			ResolvedTaskResources: &resources.ResolvedTaskResources{
+				TaskSpec: &task.Spec,
+			},
+		}},
+		expected: true,
+	}, {
+		name:     "tasks-when-expression-continue-after-skip",
+		taskName: "mytask13",
+		state: PipelineRunState{{
+			PipelineTask: &pts[11],
+			TaskRunName:  "pipelinerun-guardedtask",
+			TaskRun:      nil,
+			ResolvedTaskResources: &resources.ResolvedTaskResources{
+				TaskSpec: &task.Spec,
+			},
+		}, {
+			PipelineTask: &v1beta1.PipelineTask{
+				Name:     "mytask13",
+				TaskRef:  &v1beta1.TaskRef{Name: "task"},
+				RunAfter: []string{"mytask12"},
+			}, // mytask13 runAfter mytask12
+			TaskRunName: "ordering-dependent-task",
+			TaskRun:     nil,
+			ResolvedTaskResources: &resources.ResolvedTaskResources{
+				TaskSpec: &task.Spec,
+			},
+		}},
+		expected: false,
 	}}
 
 	for _, tc := range tcs {