Graceful Pipeline Run Termination

The implementation of TEP-0058: Graceful Pipeline Run Termination.

The new `spec.Status` values:

- `StoppedRunFinally` - To "stop" (i.e. let the tasks complete, then execute finally tasks) a Pipeline
- `CancelledRunFinally` - To "cancel" (i.e. interrupt any executing non finally tasks, then execute finally tasks)
- `Cancelled` - Same as today's `PipelineRunCancelled` - i.e. interrupt any executing tasks without running finally tasks

`PipelineRunCancelled` is deprecated (replaced by `Cancelled`)

Author: rafalbigaj

docs/pipelineruns.md

@@ -525,8 +525,44 @@ Task Runs:
 ## Cancelling a `PipelineRun`
 
 To cancel a `PipelineRun` that's currently executing, update its definition
-to mark it as cancelled. When you do so, the spawned `TaskRuns` are also marked
-as cancelled and all associated `Pods` are deleted. For example:
+to mark it as "Cancelled". When you do so, the spawned `TaskRuns` are also marked
+as cancelled and all associated `Pods` are deleted. Pending final tasks are not scheduled. 
+For example:
+
+```yaml
+apiVersion: tekton.dev/v1beta1
+kind: PipelineRun
+metadata:
+  name: go-example-git
+spec:
+  # […]
+  status: "Cancelled"
+```
+
+## Gracefully cancelling a `PipelineRun`
+
+To gracefully cancel a `PipelineRun` that's currently executing, update its definition
+to mark it as "CancelledRunFinally". When you do so, the spawned `TaskRuns` are also marked
+as cancelled and all associated `Pods` are deleted. Final tasks are scheduled normally. 
+For example:
+
+```yaml
+apiVersion: tekton.dev/v1beta1
+kind: PipelineRun
+metadata:
+  name: go-example-git
+spec:
+  # […]
+  status: "CancelledRunFinally"
+```
+
+
+## Gracefully stopping a `PipelineRun`
+
+To gracefully stop a `PipelineRun` that's currently executing, update its definition
+to mark it as "StoppedRunFinally". When you do so, the spawned `TaskRuns` are completed normally,
+but no new non-final task is scheduled. Final tasks are executed afterwards.
+For example:
 
 ```yaml
 apiVersion: tekton.dev/v1beta1
@@ -535,7 +571,7 @@ metadata:
   name: go-example-git
 spec:
   # […]
-  status: "PipelineRunCancelled"
+  status: "StoppedRunFinally"
 ```
 
 ## Pending `PipelineRuns`

internal/builder/v1beta1/pipeline.go

@@ -120,6 +120,16 @@ func PipelineRunCancelled(spec *v1beta1.PipelineRunSpec) {
 	spec.Status = v1beta1.PipelineRunSpecStatusCancelled
 }
 
+// PipelineRunCancelledRunFinally sets the status to cancel and run finally the PipelineRunSpec.
+func PipelineRunCancelledRunFinally(spec *v1beta1.PipelineRunSpec) {
+	spec.Status = v1beta1.PipelineRunSpecStatusCancelledRunFinally
+}
+
+// PipelineRunStoppedRunFinally sets the status to stop and run finally the PipelineRunSpec.
+func PipelineRunStoppedRunFinally(spec *v1beta1.PipelineRunSpec) {
+	spec.Status = v1beta1.PipelineRunSpecStatusStoppedRunFinally
+}
+
 // PipelineRunPending sets the status to pending to the PipelineRunSpec.
 func PipelineRunPending(spec *v1beta1.PipelineRunSpec) {
 	spec.Status = v1beta1.PipelineRunSpecStatusPending

pkg/apis/pipeline/v1alpha1/pipelinerun_types.go

@@ -101,7 +101,7 @@ type PipelineRunSpecStatus = v1beta1.PipelineRunSpecStatus
 const (
 	// PipelineRunSpecStatusCancelled indicates that the user wants to cancel the task,
 	// if not already cancelled or terminated
-	PipelineRunSpecStatusCancelled = v1beta1.PipelineRunSpecStatusCancelled
+	PipelineRunSpecStatusCancelled = v1beta1.PipelineRunSpecStatusCancelledDeprecated
 )
 
 // PipelineResourceRef can be used to refer to a specific instance of a Resource

pkg/apis/pipeline/v1beta1/pipelinerun_types.go

@@ -85,7 +85,17 @@ func (pr *PipelineRun) HasStarted() bool {
 
 // IsCancelled returns true if the PipelineRun's spec status is set to Cancelled state
 func (pr *PipelineRun) IsCancelled() bool {
-	return pr.Spec.Status == PipelineRunSpecStatusCancelled
+	return pr.Spec.Status == PipelineRunSpecStatusCancelled || pr.Spec.Status == PipelineRunSpecStatusCancelledDeprecated
+}
+
+// IsStopped returns true if the PipelineRun's spec status is set to Stopped state
+func (pr *PipelineRun) IsStopped() bool {
+	return pr.Spec.Status == PipelineRunSpecStatusCancelledRunFinally
+}
+
+// IsStopping returns true if the PipelineRun's spec status is set to Stopping state
+func (pr *PipelineRun) IsStopping() bool {
+	return pr.Spec.Status == PipelineRunSpecStatusStoppedRunFinally
 }
 
 func (pr *PipelineRun) GetTimeout(ctx context.Context) time.Duration {
@@ -193,9 +203,22 @@ type PipelineRunSpec struct {
 type PipelineRunSpecStatus string
 
 const (
+	// PipelineRunSpecStatusCancelledDeprecated indicates that the user wants to cancel the task,
+	// if not already cancelled or terminated (deprecated: replaced by PipelineRunSpecStatusCancelled)
+	PipelineRunSpecStatusCancelledDeprecated = "PipelineRunCancelled"
+
 	// PipelineRunSpecStatusCancelled indicates that the user wants to cancel the task,
 	// if not already cancelled or terminated
-	PipelineRunSpecStatusCancelled = "PipelineRunCancelled"
+	PipelineRunSpecStatusCancelled = "Cancelled"
+
+	// PipelineRunSpecStatusCancelledRunFinally indicates that the user wants to cancel the pipeline run,
+	// if not already cancelled or terminated, but ensure finally is run normally
+	PipelineRunSpecStatusCancelledRunFinally = "CancelledRunFinally"
+
+	// PipelineRunSpecStatusStoppedRunFinally indicates that the user wants to stop the pipeline run,
+	// wait for already running tasks to be completed and run finally
+	// if not already cancelled or terminated
+	PipelineRunSpecStatusStoppedRunFinally = "StoppedRunFinally"
 
 	// PipelineRunSpecStatusPending indicates that the user wants to postpone starting a PipelineRun
 	// until some condition is met

pkg/apis/pipeline/v1beta1/pipelinerun_validation.go

@@ -82,11 +82,7 @@ func (ps *PipelineRunSpec) Validate(ctx context.Context) (errs *apis.FieldError)
 		}
 	}
 
-	if ps.Status != "" {
-		if ps.Status != PipelineRunSpecStatusCancelled && ps.Status != PipelineRunSpecStatusPending {
-			errs = errs.Also(apis.ErrInvalidValue(fmt.Sprintf("%s should be %s or %s", ps.Status, PipelineRunSpecStatusCancelled, PipelineRunSpecStatusPending), "status"))
-		}
-	}
+	errs = errs.Also(validateSpecStatus(ps.Status))
 
 	if ps.Workspaces != nil {
 		wsNames := make(map[string]int)
@@ -101,3 +97,18 @@ func (ps *PipelineRunSpec) Validate(ctx context.Context) (errs *apis.FieldError)
 
 	return errs
 }
+
+func validateSpecStatus(status PipelineRunSpecStatus) *apis.FieldError {
+	switch status {
+	case "":
+		return nil
+	case PipelineRunSpecStatusCancelledDeprecated,
+		PipelineRunSpecStatusCancelled,
+		PipelineRunSpecStatusCancelledRunFinally,
+		PipelineRunSpecStatusStoppedRunFinally:
+		return nil
+	}
+	return apis.ErrInvalidValue(fmt.Sprintf("%s should be %s, %s, %s or %s", status,
+		PipelineRunSpecStatusCancelled, PipelineRunSpecStatusCancelledRunFinally,
+		PipelineRunSpecStatusStoppedRunFinally, PipelineRunSpecStatusPending), "status")
+}

pkg/reconciler/pipelinerun/cancel.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"github.com/tektoncd/pipeline/pkg/reconciler/pipeline/dag"
 	"log"
 	"strings"
 	"time"
@@ -59,11 +60,51 @@ func init() {
 
 // cancelPipelineRun marks the PipelineRun as cancelled and any resolved TaskRun(s) too.
 func cancelPipelineRun(ctx context.Context, logger *zap.SugaredLogger, pr *v1beta1.PipelineRun, clientSet clientset.Interface) error {
+	errs := cancelPipelineTaskRuns(ctx, logger, pr, nil, clientSet)
+
+	// If we successfully cancelled all the TaskRuns and Runs, we can consider the PipelineRun cancelled.
+	if len(errs) == 0 {
+		pr.Status.SetCondition(&apis.Condition{
+			Type:    apis.ConditionSucceeded,
+			Status:  corev1.ConditionFalse,
+			Reason:  ReasonCancelled,
+			Message: fmt.Sprintf("PipelineRun %q was cancelled", pr.Name),
+		})
+		// update pr completed time
+		pr.Status.CompletionTime = &metav1.Time{Time: time.Now()}
+	} else {
+		e := strings.Join(errs, "\n")
+		// Indicate that we failed to cancel the PipelineRun
+		pr.Status.SetCondition(&apis.Condition{
+			Type:    apis.ConditionSucceeded,
+			Status:  corev1.ConditionUnknown,
+			Reason:  ReasonCouldntCancel,
+			Message: fmt.Sprintf("PipelineRun %q was cancelled but had errors trying to cancel TaskRuns and/or Runs: %s", pr.Name, e),
+		})
+		return fmt.Errorf("error(s) from cancelling TaskRun(s) from PipelineRun %s: %s", pr.Name, e)
+	}
+	return nil
+}
+
+// cancelPipelineTaskRuns patches `TaskRun` and `Run` with canceled status
+func cancelPipelineTaskRuns(ctx context.Context, logger *zap.SugaredLogger, pr *v1beta1.PipelineRun, omitTaskGraph *dag.Graph, clientSet clientset.Interface) []string {
 	errs := []string{}
 
+	isTaskOmittedFromCancel := func(taskName string) bool {
+		if omitTaskGraph == nil {
+			return false
+		}
+		_, ok := omitTaskGraph.Nodes[taskName]
+		return ok
+	}
+
 	// Loop over the TaskRuns in the PipelineRun status.
 	// If a TaskRun is not in the status yet we should not cancel it anyways.
-	for taskRunName := range pr.Status.TaskRuns {
+	for taskRunName, taskRun := range pr.Status.TaskRuns {
+		if isTaskOmittedFromCancel(taskRun.PipelineTaskName) {
+			continue
+		}
+
 		logger.Infof("cancelling TaskRun %s", taskRunName)
 
 		if _, err := clientSet.TektonV1beta1().TaskRuns(pr.Namespace).Patch(ctx, taskRunName, types.JSONPatchType, cancelTaskRunPatchBytes, metav1.PatchOptions{}, ""); err != nil {
@@ -72,25 +113,28 @@ func cancelPipelineRun(ctx context.Context, logger *zap.SugaredLogger, pr *v1bet
 		}
 	}
 	// Loop over the Runs in the PipelineRun status.
-	for runName := range pr.Status.Runs {
+	for runName, run := range pr.Status.Runs {
+		if isTaskOmittedFromCancel(run.PipelineTaskName) {
+			continue
+		}
+
 		logger.Infof("cancelling Run %s", runName)
 
 		if _, err := clientSet.TektonV1alpha1().Runs(pr.Namespace).Patch(ctx, runName, types.JSONPatchType, cancelRunPatchBytes, metav1.PatchOptions{}, ""); err != nil {
 			errs = append(errs, fmt.Errorf("Failed to patch Run `%s` with cancellation: %s", runName, err).Error())
 			continue
 		}
 	}
-	// If we successfully cancelled all the TaskRuns and Runs, we can consider the PipelineRun cancelled.
-	if len(errs) == 0 {
-		pr.Status.SetCondition(&apis.Condition{
-			Type:    apis.ConditionSucceeded,
-			Status:  corev1.ConditionFalse,
-			Reason:  ReasonCancelled,
-			Message: fmt.Sprintf("PipelineRun %q was cancelled", pr.Name),
-		})
-		// update pr completed time
-		pr.Status.CompletionTime = &metav1.Time{Time: time.Now()}
-	} else {
+
+	return errs
+}
+
+// stopPipelineRun marks the PipelineRun as stopping, any resolved TaskRun(s) as cancelled and runs finally.
+func stopPipelineRun(ctx context.Context, logger *zap.SugaredLogger, pr *v1beta1.PipelineRun, finalTaskGraph *dag.Graph, clientSet clientset.Interface) error {
+	errs := cancelPipelineTaskRuns(ctx, logger, pr, finalTaskGraph, clientSet)
+
+	// If we successfully cancelled all the TaskRuns and Runs, we can proceed with the PipelineRun reconciliation to trigger finally.
+	if len(errs) > 0 {
 		e := strings.Join(errs, "\n")
 		// Indicate that we failed to cancel the PipelineRun
 		pr.Status.SetCondition(&apis.Condition{

pkg/reconciler/pipelinerun/cancel_test.go

@@ -95,6 +95,14 @@ func TestCancelPipelineRun(t *testing.T) {
 			{ObjectMeta: metav1.ObjectMeta{Name: "t1"}},
 			{ObjectMeta: metav1.ObjectMeta{Name: "t2"}},
 		},
+	}, {
+		name: "deprecated-state",
+		pipelineRun: &v1beta1.PipelineRun{
+			ObjectMeta: metav1.ObjectMeta{Name: "test-pipeline-run-cancelled"},
+			Spec: v1beta1.PipelineRunSpec{
+				Status: v1beta1.PipelineRunSpecStatusCancelledDeprecated,
+			},
+		},
 	}}
 	for _, tc := range testCases {
 		tc := tc

pkg/reconciler/pipelinerun/pipelinerun.go

@@ -390,6 +390,15 @@ func (c *Reconciler) reconcile(ctx context.Context, pr *v1beta1.PipelineRun, get
 		return controller.NewPermanentError(err)
 	}
 
+	// once the list of final tasks is ready, check if pipeline is not stopped
+	if pr.IsStopped() {
+		// If the pipelinerun is stopped, cancel tasks, but run finally
+		err := stopPipelineRun(ctx, logger, pr, dfinally, c.PipelineClientSet)
+		if err != nil {
+			return controller.NewPermanentError(err)
+		}
+	}
+
 	if err := pipelineSpec.Validate(ctx); err != nil {
 		// This Run has failed, so we need to mark it as failed and stop reconciling it
 		pr.Status.MarkFailed(ReasonFailedValidation,
@@ -489,6 +498,7 @@ func (c *Reconciler) reconcile(ctx context.Context, pr *v1beta1.PipelineRun, get
 	// dag tasks graph and final tasks graph
 	pipelineRunFacts := &resources.PipelineRunFacts{
 		State:           pipelineRunState,
+		SpecStatus:      pr.Spec.Status,
 		TasksGraph:      d,
 		FinalTasksGraph: dfinally,
 	}