enabling consuming task results in finally

Final tasks can be configured to consume Results of PipelineTasks from
tasks section

Author: pritidesai

docs/pipelines.md

@@ -788,6 +788,32 @@ spec:
           value: "someURL"
 ```
 
+#### Consuming `Task` execution results in `finally`
+
+Final tasks can be configured to consume `Results` of `PipelineTask` from the `tasks` section:
+
+```yaml
+spec:
+  tasks:
+    - name: clone-app-repo
+      taskRef:
+        name: git-clone
+  finally:
+    - name: discover-git-commit
+      params:
+        - name: commit
+          value: $(tasks.clone-app-repo.results.commit)
+```
+**Note:** The scheduling of such final task does not change, it will still be executed in parallel with other
+final tasks after all non-final tasks are done.
+
+The controller resolves task results before executing the finally task `discover-git-commit`. If the task
+`clone-app-repo` failed or skipped with [when expression](#guard-task-execution-using-whenexpressions) resulting in
+uninitialized task result `commit`, the finally Task `discover-git-commit` will be included in the list of
+`skippedTasks` and continues executing rest of the final tasks. The pipeline exits with `completion` instead of
+`success` if a finally task is added to the list of `skippedTasks`.
+
+
 ### `PipelineRun` Status with `finally`
 
 With `finally`, `PipelineRun` status is calculated based on `PipelineTasks` under `tasks` section and final tasks.
@@ -900,35 +926,6 @@ no `runAfter` can be specified in final tasks.
 final tasks are guaranteed to be executed after all `PipelineTasks` therefore no `conditions` can be specified in
 final tasks.
 
-#### Cannot configure `Task` execution results with `finally`
-
-Final tasks can not be configured to consume `Results` of `PipelineTask` from `tasks` section i.e. the following
-example is not supported right now but we are working on adding support for the same (tracked in issue
-[#2557](https://github.com/tektoncd/pipeline/issues/2557)).
-
-```yaml
-spec:
-  tasks:
-    - name: count-comments-before
-      taskRef:
-        Name: count-comments
-    - name: add-comment
-      taskRef:
-        Name: add-comment
-    - name: count-comments-after
-      taskRef:
-        Name: count-comments
-  finally:
-    - name: check-count
-      taskRef:
-        Name: check-count
-      params:
-        - name: before-count
-          value: $(tasks.count-comments-before.results.count) #invalid
-        - name: after-count
-          value: $(tasks.count-comments-after.results.count) #invalid
-```
-
 #### Cannot configure `Pipeline` result with `finally`
 
 Final tasks can emit `Results` but results emitted from the final tasks can not be configured in the

examples/v1beta1/pipelineruns/pipelinerun-with-final-tasks.yaml

@@ -164,6 +164,20 @@ spec:
       workspaces:
         - name: source
           workspace: git-source
+    - name: check-git-commit
+      params:
+        - name: commit
+          value: $(tasks.clone-app-repo.results.commit)
+      taskSpec:
+        params:
+          - name: commit
+        steps:
+          - name: check-commit-initialized
+            image: alpine
+            script: |
+              if [[ ! $(params.commit) ]]; then
+                exit 1
+              fi
 ---
 
 # PipelineRun to execute pipeline - clone-into-workspace-and-cleanup-workspace

pkg/apis/pipeline/v1beta1/pipeline_validation.go

@@ -68,7 +68,7 @@ func (ps *PipelineSpec) Validate(ctx context.Context) (errs *apis.FieldError) {
 	// Validate the pipeline's results
 	errs = errs.Also(validatePipelineResults(ps.Results))
 	errs = errs.Also(validateTasksAndFinallySection(ps))
-	errs = errs.Also(validateFinalTasks(ps.Finally))
+	errs = errs.Also(validateFinalTasks(ps.Tasks, ps.Finally))
 	errs = errs.Also(validateWhenExpressions(ps.Tasks))
 	return errs
 }
@@ -420,7 +420,7 @@ func validateTasksAndFinallySection(ps *PipelineSpec) *apis.FieldError {
 	return nil
 }
 
-func validateFinalTasks(finalTasks []PipelineTask) *apis.FieldError {
+func validateFinalTasks(tasks []PipelineTask, finalTasks []PipelineTask) *apis.FieldError {
 	for idx, f := range finalTasks {
 		if len(f.RunAfter) != 0 {
 			return apis.ErrInvalidValue(fmt.Sprintf("no runAfter allowed under spec.finally, final task %s has runAfter specified", f.Name), "").ViaFieldIndex("finally", idx)
@@ -433,7 +433,10 @@ func validateFinalTasks(finalTasks []PipelineTask) *apis.FieldError {
 		}
 	}
 
-	if err := validateTaskResultReferenceNotUsed(finalTasks).ViaField("finally"); err != nil {
+	ts := PipelineTaskList(tasks).Names()
+	fts := PipelineTaskList(finalTasks).Names()
+
+	if err := validateTaskResultReference(finalTasks, ts, fts).ViaField("finally"); err != nil {
 		return err
 	}
 
@@ -444,14 +447,22 @@ func validateFinalTasks(finalTasks []PipelineTask) *apis.FieldError {
 	return nil
 }
 
-func validateTaskResultReferenceNotUsed(tasks []PipelineTask) *apis.FieldError {
-	for idx, t := range tasks {
+func validateTaskResultReference(finalTasks []PipelineTask, ts, fts sets.String) *apis.FieldError {
+	for idx, t := range finalTasks {
 		for _, p := range t.Params {
 			expressions, ok := GetVarSubstitutionExpressionsForParam(p)
 			if ok {
 				if LooksLikeContainsResultRefs(expressions) {
-					return apis.ErrInvalidValue(fmt.Sprintf("no task result allowed under params,"+
-						"final task param %s has set task result as its value", p.Name), "params").ViaIndex(idx)
+					resultRefs := NewResultRefs(expressions)
+					for _, resultRef := range resultRefs {
+						if fts.Has(resultRef.PipelineTask) {
+							return apis.ErrInvalidValue(fmt.Sprintf("invalid task result reference, "+
+								"final task param %s has task result reference from a final task", p.Name), "params").ViaIndex(idx)
+						} else if !ts.Has(resultRef.PipelineTask) {
+							return apis.ErrInvalidValue(fmt.Sprintf("invalid task result reference, "+
+								"final task param %s has task result reference from a task which is not defined in the pipeline", p.Name), "params").ViaIndex(idx)
+						}
+					}
 				}
 			}
 		}

pkg/apis/pipeline/v1beta1/pipeline_validation_test.go

@@ -1614,6 +1614,24 @@ func TestValidatePipelineWithFinalTasks_Success(t *testing.T) {
 				}},
 			},
 		},
+	}, {
+		name: "valid pipeline with final tasks referring to task results from a dag task",
+		p: &Pipeline{
+			ObjectMeta: metav1.ObjectMeta{Name: "pipeline"},
+			Spec: PipelineSpec{
+				Tasks: []PipelineTask{{
+					Name:    "non-final-task",
+					TaskRef: &TaskRef{Name: "non-final-task"},
+				}},
+				Finally: []PipelineTask{{
+					Name:    "final-task-1",
+					TaskRef: &TaskRef{Name: "final-task"},
+					Params: []Param{{
+						Name: "param1", Value: ArrayOrString{Type: ParamTypeString, StringVal: "$(tasks.non-final-task.results.output)"},
+					}},
+				}},
+			},
+		},
 	}}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
@@ -1918,6 +1936,7 @@ func TestValidateTasksAndFinallySection_Failure(t *testing.T) {
 func TestValidateFinalTasks_Failure(t *testing.T) {
 	tests := []struct {
 		name          string
+		tasks         []PipelineTask
 		finalTasks    []PipelineTask
 		expectedError apis.FieldError
 	}{{
@@ -1960,16 +1979,32 @@ func TestValidateFinalTasks_Failure(t *testing.T) {
 			Paths:   []string{"finally[0].resources.inputs[0]"},
 		},
 	}, {
-		name: "invalid pipeline with final tasks having reference to task results",
+		name: "invalid pipeline with final tasks having task results reference from a final task",
+		finalTasks: []PipelineTask{{
+			Name:    "final-task-1",
+			TaskRef: &TaskRef{Name: "final-task"},
+		}, {
+			Name:    "final-task-2",
+			TaskRef: &TaskRef{Name: "final-task"},
+			Params: []Param{{
+				Name: "param1", Value: ArrayOrString{Type: ParamTypeString, StringVal: "$(tasks.final-task-1.results.output)"},
+			}},
+		}},
+		expectedError: apis.FieldError{
+			Message: `invalid value: invalid task result reference, final task param param1 has task result reference from a final task`,
+			Paths:   []string{"finally[1].params"},
+		},
+	}, {
+		name: "invalid pipeline with final tasks having task results reference from non existent dag task",
 		finalTasks: []PipelineTask{{
 			Name:    "final-task",
 			TaskRef: &TaskRef{Name: "final-task"},
 			Params: []Param{{
-				Name: "param1", Value: ArrayOrString{Type: ParamTypeString, StringVal: "$(tasks.a-task.results.output)"},
+				Name: "param1", Value: ArrayOrString{Type: ParamTypeString, StringVal: "$(tasks.no-dag-task-1.results.output)"},
 			}},
 		}},
 		expectedError: apis.FieldError{
-			Message: `invalid value: no task result allowed under params,final task param param1 has set task result as its value`,
+			Message: `invalid value: invalid task result reference, final task param param1 has task result reference from a task which is not defined in the pipeline`,
 			Paths:   []string{"finally[0].params"},
 		},
 	}, {
@@ -1990,7 +2025,7 @@ func TestValidateFinalTasks_Failure(t *testing.T) {
 	}}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			err := validateFinalTasks(tt.finalTasks)
+			err := validateFinalTasks(tt.tasks, tt.finalTasks)
 			if err == nil {
 				t.Errorf("Pipeline.ValidateFinalTasks() did not return error for invalid pipeline")
 			}

pkg/reconciler/pipelinerun/pipelinerun.go

@@ -581,7 +581,18 @@ func (c *Reconciler) runNextSchedulableTask(ctx context.Context, pr *v1beta1.Pip
 	if len(fnextRprts) != 0 {
 		// apply the runtime context just before creating taskRuns for final tasks in queue
 		resources.ApplyPipelineTaskContext(fnextRprts, pipelineRunFacts.GetPipelineTaskStatus(ctx))
-		nextRprts = append(nextRprts, fnextRprts...)
+
+		// Before creating TaskRun for scheduled final task, check if it's consuming a task result
+		// Resolve and apply task result wherever applicable, report warning in case resolution fails
+		for _, rprt := range fnextRprts {
+			resolvedResultRefs, err := resources.ResolveResultRef(pipelineRunFacts.State, rprt)
+			if err != nil {
+				logger.Infof("Final task %q is not executed as it could not resolve task params for %q: %v", rprt.PipelineTask.Name, pr.Name, err)
+				continue
+			}
+			resources.ApplyTaskResults(resources.PipelineRunState{rprt}, resolvedResultRefs)
+			nextRprts = append(nextRprts, rprt)
+		}
 	}
 
 	for _, rprt := range nextRprts {