Improvements on pipeline cancel

When a pipeline is cancelled, detect it early in the reconcile
cycle and cancel all taskruns defined in the pipelinerun status,
so that we don't spend time building the DAG and resolving all
resources.

The patch to cancel a TaskRun is the same for all TaskRuns, so
build it once and apply it to all taskruns.

Author: afrittoli

pkg/reconciler/pipelinerun/cancel.go

@@ -27,15 +27,14 @@ import (
 
 	"github.com/tektoncd/pipeline/pkg/apis/pipeline/v1alpha1"
 	clientset "github.com/tektoncd/pipeline/pkg/client/clientset/versioned"
-	"github.com/tektoncd/pipeline/pkg/reconciler/pipelinerun/resources"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"knative.dev/pkg/apis"
 )
 
 // cancelPipelineRun marks the PipelineRun as cancelled and any resolved TaskRun(s) too.
-func cancelPipelineRun(logger *zap.SugaredLogger, pr *v1alpha1.PipelineRun, pipelineState []*resources.ResolvedPipelineRunTask, clientSet clientset.Interface) error {
+func cancelPipelineRun(logger *zap.SugaredLogger, pr *v1alpha1.PipelineRun, clientSet clientset.Interface) error {
 	pr.Status.SetCondition(&apis.Condition{
 		Type:    apis.ConditionSucceeded,
 		Status:  corev1.ConditionFalse,
@@ -45,23 +44,21 @@ func cancelPipelineRun(logger *zap.SugaredLogger, pr *v1alpha1.PipelineRun, pipe
 	// update pr completed time
 	pr.Status.CompletionTime = &metav1.Time{Time: time.Now()}
 	errs := []string{}
-	for _, rprt := range pipelineState {
-		if rprt.TaskRun == nil {
-			// No taskrun yet, pass
-			continue
-		}
 
-		logger.Infof("cancelling TaskRun %s", rprt.TaskRunName)
+	// Use Patch to update the TaskRuns since the TaskRun controller may be operating on the
+	// TaskRuns at the same time and trying to update the entire object may cause a race
+	b, err := getCancelPatch()
+	if err != nil {
+		return fmt.Errorf("couldn't make patch to update TaskRun cancellation: %v", err)
+	}
 
-		// Use Patch to update the TaskRuns since the TaskRun controller may be operating on the
-		// TaskRuns at the same time and trying to update the entire object may cause a race
-		b, err := getCancelPatch()
-		if err != nil {
-			errs = append(errs, fmt.Errorf("couldn't make patch to update TaskRun cancellation: %v", err).Error())
-			continue
-		}
-		if _, err := clientSet.TektonV1alpha1().TaskRuns(pr.Namespace).Patch(rprt.TaskRunName, types.JSONPatchType, b, ""); err != nil {
-			errs = append(errs, fmt.Errorf("Failed to patch TaskRun `%s` with cancellation: %s", rprt.TaskRunName, err).Error())
+	// 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 {
+		logger.Infof("cancelling TaskRun %s", taskRunName)
+
+		if _, err := clientSet.TektonV1alpha1().TaskRuns(pr.Namespace).Patch(taskRunName, types.JSONPatchType, b, ""); err != nil {
+			errs = append(errs, fmt.Errorf("Failed to patch TaskRun `%s` with cancellation: %s", taskRunName, err).Error())
 			continue
 		}
 	}

pkg/reconciler/pipelinerun/cancel_test.go

@@ -21,7 +21,6 @@ import (
 	"testing"
 
 	"github.com/tektoncd/pipeline/pkg/apis/pipeline/v1alpha1"
-	"github.com/tektoncd/pipeline/pkg/reconciler/pipelinerun/resources"
 	ttesting "github.com/tektoncd/pipeline/pkg/reconciler/testing"
 	tb "github.com/tektoncd/pipeline/test/builder"
 	test "github.com/tektoncd/pipeline/test/v1alpha1"
@@ -32,10 +31,9 @@ import (
 
 func TestCancelPipelineRun(t *testing.T) {
 	testCases := []struct {
-		name          string
-		pipelineRun   *v1alpha1.PipelineRun
-		pipelineState []*resources.ResolvedPipelineRunTask
-		taskRuns      []*v1alpha1.TaskRun
+		name        string
+		pipelineRun *v1alpha1.PipelineRun
+		taskRuns    []*v1alpha1.TaskRun
 	}{{
 		name: "no-resolved-taskrun",
 		pipelineRun: tb.PipelineRun("test-pipeline-run-cancelled", tb.PipelineRunNamespace("foo"),
@@ -44,28 +42,29 @@ func TestCancelPipelineRun(t *testing.T) {
 			),
 		),
 	}, {
-		name: "1-of-resolved-taskrun",
+		name: "1-taskrun",
 		pipelineRun: tb.PipelineRun("test-pipeline-run-cancelled", tb.PipelineRunNamespace("foo"),
 			tb.PipelineRunSpec("test-pipeline",
 				tb.PipelineRunCancelled,
 			),
+			tb.PipelineRunStatus(
+				tb.PipelineRunTaskRunsStatus("t1", &v1alpha1.PipelineRunTaskRunStatus{
+					PipelineTaskName: "task-1",
+				})),
 		),
-		pipelineState: []*resources.ResolvedPipelineRunTask{
-			{TaskRunName: "t1", TaskRun: tb.TaskRun("t1", tb.TaskRunNamespace("foo"))},
-			{TaskRunName: "t2"},
-		},
 		taskRuns: []*v1alpha1.TaskRun{tb.TaskRun("t1", tb.TaskRunNamespace("foo"))},
 	}, {
-		name: "resolved-taskruns",
+		name: "multiple-taskruns",
 		pipelineRun: tb.PipelineRun("test-pipeline-run-cancelled", tb.PipelineRunNamespace("foo"),
 			tb.PipelineRunSpec("test-pipeline",
 				tb.PipelineRunCancelled,
 			),
+			tb.PipelineRunStatus(
+				tb.PipelineRunTaskRunsStatus(
+					"t1", &v1alpha1.PipelineRunTaskRunStatus{PipelineTaskName: "task-1"}),
+				tb.PipelineRunTaskRunsStatus(
+					"t2", &v1alpha1.PipelineRunTaskRunStatus{PipelineTaskName: "task-2"})),
 		),
-		pipelineState: []*resources.ResolvedPipelineRunTask{
-			{TaskRunName: "t1", TaskRun: tb.TaskRun("t1", tb.TaskRunNamespace("foo"))},
-			{TaskRunName: "t2", TaskRun: tb.TaskRun("t2", tb.TaskRunNamespace("foo"))},
-		},
 		taskRuns: []*v1alpha1.TaskRun{tb.TaskRun("t1", tb.TaskRunNamespace("foo")), tb.TaskRun("t2", tb.TaskRunNamespace("foo"))},
 	}}
 	for _, tc := range testCases {
@@ -79,7 +78,7 @@ func TestCancelPipelineRun(t *testing.T) {
 			ctx, cancel := context.WithCancel(ctx)
 			defer cancel()
 			c, _ := test.SeedTestData(t, ctx, d)
-			err := cancelPipelineRun(logtesting.TestLogger(t), tc.pipelineRun, tc.pipelineState, c.Pipeline)
+			err := cancelPipelineRun(logtesting.TestLogger(t), tc.pipelineRun, c.Pipeline)
 			if err != nil {
 				t.Fatal(err)
 			}

pkg/reconciler/pipelinerun/pipelinerun.go

@@ -181,6 +181,12 @@ func (c *Reconciler) Reconcile(ctx context.Context, key string) error {
 				c.Logger.Warnf("Failed to log the metrics : %v", err)
 			}
 		}(c.metrics)
+	} else if pr.IsCancelled() {
+		// If the pipelinerun is cancelled, cancel tasks and update status
+		before := pr.Status.GetCondition(apis.ConditionSucceeded)
+		multierror.Append(merr, cancelPipelineRun(c.Logger, pr, c.PipelineClientSet))
+		after := pr.Status.GetCondition(apis.ConditionSucceeded)
+		reconciler.EmitEvent(c.Recorder, before, after, pr)
 	} else {
 		if err := c.tracker.Track(pr.GetTaskRunRef(), pr); err != nil {
 			c.Logger.Errorf("Failed to create tracker for TaskRuns for PipelineRun %s: %v", pr.Name, err)
@@ -527,15 +533,6 @@ func (c *Reconciler) reconcile(ctx context.Context, pr *v1alpha1.PipelineRun) er
 		}
 	}
 
-	// If the pipelinerun is cancelled, cancel tasks and update status
-	if pr.IsCancelled() {
-		before := pr.Status.GetCondition(apis.ConditionSucceeded)
-		err := cancelPipelineRun(c.Logger, pr, pipelineState, c.PipelineClientSet)
-		after := pr.Status.GetCondition(apis.ConditionSucceeded)
-		reconciler.EmitEvent(c.Recorder, before, after, pr)
-		return err
-	}
-
 	if pipelineState.IsBeforeFirstTaskRun() && pr.HasVolumeClaimTemplate() {
 		// create workspace PVC from template
 		if err = c.pvcHandler.CreatePersistentVolumeClaimsForWorkspaces(pr.Spec.Workspaces, pr.GetOwnerReference(), pr.Namespace); err != nil {