enabling consuming task results in finally

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

docs/pipelines.md

@@ -688,6 +688,36 @@ spec:
  value: "someURL"
 ```
 
+#### Consuming `Task` execution results in `finally`
+
+Final tasks can be configured to consume `Results` of `PipelineTask` from `tasks` section:
+
+```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
+```
+**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.
+
+
 ### `PipelineRun` Status with `finally`
 
 With `finally`, `PipelineRun` status is calculated based on `PipelineTasks` under `tasks` section and final tasks.
@@ -765,35 +795,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

pkg/apis/pipeline/v1beta1/pipeline_validation.go

@@ -472,32 +472,13 @@ func validateFinalTasks(finalTasks []PipelineTask) *apis.FieldError {
  }
  }
 
- if err := validateTaskResultReferenceNotUsed(finalTasks); err != nil {
- return err
- }
-
  if err := validateTasksInputFrom(finalTasks); err != nil {
  return err
  }
 
  return nil
 }
 
-func validateTaskResultReferenceNotUsed(tasks []PipelineTask) *apis.FieldError {
- for _, t := range tasks {
- 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), "spec.finally.task.params")
- }
- }
- }
- }
- return nil
-}
-
 func validateTasksInputFrom(tasks []PipelineTask) *apis.FieldError {
  for _, t := range tasks {
  inputResources := []PipelineTaskInputResource{}

pkg/apis/pipeline/v1beta1/pipeline_validation_test.go

@@ -1524,15 +1524,6 @@ func TestValidateFinalTasks_Failure(t *testing.T) {
  }},
  },
  }},
- }, {
- name: "invalid pipeline with final tasks having reference to task results",
- 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: "invalid pipeline with final task specifying when expressions",
  finalTasks: []PipelineTask{{

pkg/reconciler/pipeline/dag/dag.go

@@ -67,17 +67,30 @@ func (g *Graph) addPipelineTask(t Task) (*Node, error) {
 }
 
 // Build returns a valid pipeline Graph. Returns error if the pipeline is invalid
-func Build(tasks Tasks) (*Graph, error) {
+func BuildGraph(tasks Tasks) (*Graph, error) {
  d := newGraph()
 
- deps := map[string][]string{}
  // Add all Tasks mentioned in the `PipelineSpec`
  for _, pt := range tasks.Items() {
  if _, err := d.addPipelineTask(pt); err != nil {
  return nil, fmt.Errorf("task %s is already present in Graph, can't add it again: %w", pt.HashKey(), err)
  }
+ }
+ return d, nil
+}
+
+// Build returns a valid pipeline Graph. Returns error if the pipeline is invalid
+func Build(tasks Tasks) (*Graph, error) {
+ d, err := BuildGraph(tasks)
+ if err != nil {
+ return nil, err
+ }
+
+ deps := map[string][]string{}
+ for _, pt := range tasks.Items() {
  deps[pt.HashKey()] = pt.Deps()
  }
+
  // Process all from and runAfter constraints to add task dependency
  for pt, taskDeps := range deps {
  for _, previousTask := range taskDeps {

pkg/reconciler/pipelinerun/pipelinerun.go

@@ -307,7 +307,7 @@ func (c *Reconciler) reconcile(ctx context.Context, pr *v1beta1.PipelineRun) err
  // if a task in PipelineRunState is final task or not
  // the finally section is optional and might not exist
  // dfinally holds an empty Graph in the absence of finally clause
- dfinally, err := dag.Build(v1beta1.PipelineTaskList(pipelineSpec.Finally))
+ dfinally, err := dag.BuildGraph(v1beta1.PipelineTaskList(pipelineSpec.Finally))
  if err != nil {
  // This Run has failed, so we need to mark it as failed and stop reconciling it
  pr.Status.MarkFailed(ReasonInvalidGraph,
@@ -495,7 +495,7 @@ func (c *Reconciler) runNextSchedulableTask(ctx context.Context, pr *v1beta1.Pip
  logger := logging.FromContext(ctx)
  recorder := controller.GetEventRecorder(ctx)
 
- var nextRprts []*resources.ResolvedPipelineRunTask
+ var nextRprts resources.PipelineRunState
 
  // when pipeline run is stopping, do not schedule any new task and only
  // wait for all running tasks to complete and report their status
@@ -521,7 +521,18 @@ func (c *Reconciler) runNextSchedulableTask(ctx context.Context, pr *v1beta1.Pip
  resources.ApplyTaskResults(nextRprts, resolvedResultRefs)
 
  // GetFinalTasks only returns tasks when a DAG is complete
- nextRprts = append(nextRprts, pipelineRunState.GetFinalTasks(d, dfinally)...)
+ finallyRprts := pipelineRunState.GetFinalTasks(d, dfinally)
+
+ for _, rprt := range finallyRprts {
+ resolvedResultRefs, err := resources.ResolveResultRefs(pipelineRunState, resources.PipelineRunState{rprt})
+ if err != nil {
+ rprt.InvalidTaskResultsInFinally = true
+ logger.Warnf("Declaring final task %q failed as it failed to resolve task params for %q with error %v", rprt.PipelineTask.Name, pr.Name, err)
+ continue
+ }
+ resources.ApplyTaskResults(resources.PipelineRunState{rprt}, resolvedResultRefs)
+ nextRprts = append(nextRprts, rprt)
+ }
 
  for _, rprt := range nextRprts {
  if rprt == nil || rprt.Skip(pipelineRunState, d) {

pkg/reconciler/pipelinerun/pipelinerun_test.go

@@ -3923,6 +3923,132 @@ func TestReconcilePipeline_TaskSpecMetadata(t *testing.T) {
  }
 }
 
+func TestReconcileWithTaskResultsInFinalTasks(t *testing.T) {
+ names.TestingSeed()
+ ps := []*v1beta1.Pipeline{tb.Pipeline("test-pipeline", tb.PipelineNamespace("foo"), tb.PipelineSpec(
+ tb.PipelineTask("dag-task-1", "dag-task"),
+ tb.PipelineTask("dag-task-2", "dag-task"),
+ tb.FinalPipelineTask("final-task-1", "final-task-1",
+ tb.PipelineTaskParam("finalParam", "$(tasks.dag-task-1.results.aResult)"),
+ ),
+ tb.FinalPipelineTask("final-task-2", "final-task-2",
+ tb.PipelineTaskParam("finalParam", "$(tasks.dag-task-2.results.aResult)"),
+ ),
+ ))}
+ prs := []*v1beta1.PipelineRun{tb.PipelineRun("test-pipeline-run-final-task-results", tb.PipelineRunNamespace("foo"),
+ tb.PipelineRunSpec("test-pipeline",
+ tb.PipelineRunServiceAccountName("test-sa-0"),
+ ),
+ )}
+ ts := []*v1beta1.Task{
+ tb.Task("dag-task", tb.TaskNamespace("foo")),
+ tb.Task("final-task-1", tb.TaskNamespace("foo"),
+ tb.TaskSpec(
+ tb.TaskParam("finalParam", v1beta1.ParamTypeString),
+ ),
+ ),
+ tb.Task("final-task-2", tb.TaskNamespace("foo"),
+ tb.TaskSpec(
+ tb.TaskParam("finalParam", v1beta1.ParamTypeString),
+ ),
+ ),
+ }
+ trs := []*v1beta1.TaskRun{
+ tb.TaskRun("test-pipeline-run-final-task-results-dag-task-1-xxyyy",
+ tb.TaskRunNamespace("foo"),
+ tb.TaskRunOwnerReference("PipelineRun", "test-pipeline-run-final-task-results",
+ tb.OwnerReferenceAPIVersion("tekton.dev/v1beta1"),
+ tb.Controller, tb.BlockOwnerDeletion,
+ ),
+ tb.TaskRunLabel("tekton.dev/pipeline", "test-pipeline"),
+ tb.TaskRunLabel("tekton.dev/pipelineRun", "test-pipeline-run-final-task-results"),
+ tb.TaskRunLabel("tekton.dev/pipelineTask", "dag-task-1"),
+ tb.TaskRunSpec(
+ tb.TaskRunTaskRef("hello-world"),
+ tb.TaskRunServiceAccountName("test-sa"),
+ ),
+ tb.TaskRunStatus(
+ tb.StatusCondition(
+ apis.Condition{
+ Type: apis.ConditionSucceeded,
+ Status: corev1.ConditionTrue,
+ },
+ ),
+ tb.TaskRunResult("aResult", "aResultValue"),
+ ),
+ ),
+ tb.TaskRun("test-pipeline-run-final-task-results-dag-task-2-xxyyy",
+ tb.TaskRunNamespace("foo"),
+ tb.TaskRunOwnerReference("PipelineRun", "test-pipeline-run-final-task-results",
+ tb.OwnerReferenceAPIVersion("tekton.dev/v1beta1"),
+ tb.Controller, tb.BlockOwnerDeletion,
+ ),
+ tb.TaskRunLabel("tekton.dev/pipeline", "test-pipeline"),
+ tb.TaskRunLabel("tekton.dev/pipelineRun", "test-pipeline-run-final-task-results"),
+ tb.TaskRunLabel("tekton.dev/pipelineTask", "dag-task-2"),
+ tb.TaskRunSpec(
+ tb.TaskRunTaskRef("hello-world"),
+ tb.TaskRunServiceAccountName("test-sa"),
+ ),
+ tb.TaskRunStatus(
+ tb.StatusCondition(
+ apis.Condition{
+ Type: apis.ConditionSucceeded,
+ Status: corev1.ConditionFalse,
+ },
+ ),
+ ),
+ ),
+ }
+ d := test.Data{
+ PipelineRuns: prs,
+ Pipelines: ps,
+ Tasks: ts,
+ TaskRuns: trs,
+ }
+ prt := NewPipelineRunTest(d, t)
+ defer prt.Cancel()
+
+ reconciledRun, clients := prt.reconcileRun("foo", "test-pipeline-run-final-task-results", []string{}, false)
+
+ expectedTaskRunName := "test-pipeline-run-final-task-results-final-task-1-9l9zj"
+ expectedTaskRun := tb.TaskRun(expectedTaskRunName,
+ tb.TaskRunNamespace("foo"),
+ tb.TaskRunOwnerReference("PipelineRun", "test-pipeline-run-final-task-results",
+ tb.OwnerReferenceAPIVersion("tekton.dev/v1beta1"),
+ tb.Controller, tb.BlockOwnerDeletion,
+ ),
+ tb.TaskRunLabel("tekton.dev/pipeline", "test-pipeline"),
+ tb.TaskRunLabel("tekton.dev/pipelineRun", "test-pipeline-run-final-task-results"),
+ tb.TaskRunLabel("tekton.dev/pipelineTask", "final-task-1"),
+ tb.TaskRunSpec(
+ tb.TaskRunTaskRef("final-task-1"),
+ tb.TaskRunServiceAccountName("test-sa-0"),
+ tb.TaskRunParam("finalParam", "aResultValue"),
+ ),
+ )
+ // Check that the expected TaskRun was created
+ actual, err := clients.Pipeline.TektonV1beta1().TaskRuns("foo").List(metav1.ListOptions{
+ LabelSelector: "tekton.dev/pipelineTask=final-task-1,tekton.dev/pipelineRun=test-pipeline-run-final-task-results",
+ 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))
+ }
+
+ if s := reconciledRun.Status.TaskRuns["test-pipeline-run-final-task-results-final-task-2-mz4c7"].Status.GetCondition(apis.ConditionSucceeded); s.Status != corev1.ConditionFalse {
+ t.Fatalf("Status expected to be %s but is %s", corev1.ConditionFalse, s.Status)
+ }
+}
+
 // NewPipelineRunTest returns PipelineRunTest with a new PipelineRun controller created with specified state through data
 // This PipelineRunTest can be reused for multiple PipelineRuns by calling reconcileRun for each pipelineRun
 func NewPipelineRunTest(data test.Data, t *testing.T) *PipelineRunTest {

pkg/reconciler/pipelinerun/resources/pipelinerunresolution.go

@@ -69,6 +69,9 @@ type ResolvedPipelineRunTask struct {
  ResolvedTaskResources *resources.ResolvedTaskResources
  // ConditionChecks ~~TaskRuns but for evaling conditions
  ResolvedConditionChecks TaskConditionCheckState // Could also be a TaskRun or maybe just a Pod?
+ // this flag is set for a task when task result resolution fails for that task
+ // and mainly used to create a task run object with failure for that task
+ InvalidTaskResultsInFinally bool
 }
 
 func (t ResolvedPipelineRunTask) IsDone() bool {

pkg/reconciler/pipelinerun/resources/pipelinerunstate.go

@@ -28,6 +28,12 @@ import (
  "knative.dev/pkg/apis"
 )
 
+const (
+ // ReasonInvalidTaskResultsInFinallyTask indicates that the final task was not able to resolve
+ // task result of a DAG task
+ ReasonInvalidTaskResultsInFinallyTask = "InvalidTaskResultsInFinallyTask"
+)
+
 // PipelineRunState is a slice of ResolvedPipelineRunTasks the represents the current execution
 // state of the PipelineRun.
 type PipelineRunState []*ResolvedPipelineRunTask
@@ -264,7 +270,7 @@ func (state PipelineRunState) GetSkippedTasks(pr *v1beta1.PipelineRun, d *dag.Gr
 func (state PipelineRunState) GetTaskRunsStatus(pr *v1beta1.PipelineRun) map[string]*v1beta1.PipelineRunTaskRunStatus {
  status := make(map[string]*v1beta1.PipelineRunTaskRunStatus)
  for _, rprt := range state {
- if rprt.TaskRun == nil && rprt.ResolvedConditionChecks == nil {
+ if rprt.TaskRun == nil && rprt.ResolvedConditionChecks == nil && !rprt.InvalidTaskResultsInFinally {
  continue
  }
 
@@ -305,6 +311,17 @@ func (state PipelineRunState) GetTaskRunsStatus(pr *v1beta1.PipelineRun) map[str
  })
  }
  }
+ if rprt.InvalidTaskResultsInFinally {
+ if prtrs.Status == nil {
+ prtrs.Status = &v1beta1.TaskRunStatus{}
+ }
+ prtrs.Status.SetCondition(&apis.Condition{
+ Type: apis.ConditionSucceeded,
+ Status: corev1.ConditionFalse,
+ Reason: ReasonInvalidTaskResultsInFinallyTask,
+ Message: fmt.Sprintf("Finally Task %s in PipelineRun %s was having invalid task results", rprt.PipelineTask.Name, pr.Name),
+ })
+ }
  status[rprt.TaskRunName] = prtrs
  }
  return status