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/controller.go

@@ -44,6 +44,7 @@ const (
 	resyncPeriod = 10 * time.Hour
 )
 
+// NewController instantiates a new controller.Impl from knative.dev/pkg/controller
 func NewController(images pipeline.Images) func(context.Context, configmap.Watcher) *controller.Impl {
 	return func(ctx context.Context, cmw configmap.Watcher) *controller.Impl {
 		logger := logging.FromContext(ctx)

pkg/reconciler/pipelinerun/metrics.go

@@ -48,6 +48,7 @@ var (
 		stats.UnitDimensionless)
 )
 
+// Recorder holds keys for Tekton metrics
 type Recorder struct {
 	initialized bool
 

pkg/reconciler/pipelinerun/pipelinerun.go

@@ -158,9 +158,10 @@ func (c *Reconciler) Reconcile(ctx context.Context, key string) error {
 
 	// In case of reconcile errors, we store the error in a multierror, attempt
 	// to update, and return the original error combined with any update error
-	var merr error
+	var merr *multierror.Error
 
-	if pr.IsDone() {
+	switch {
+	case pr.IsDone():
 		// We may be reading a version of the object that was stored at an older version
 		// and may not have had all of the assumed default specified.
 		pr.SetDefaults(contexts.WithUpgradeViaDefaulting(ctx))
@@ -181,7 +182,13 @@ func (c *Reconciler) Reconcile(ctx context.Context, key string) error {
 				c.Logger.Warnf("Failed to log the metrics : %v", err)
 			}
 		}(c.metrics)
-	} else {
+	case pr.IsCancelled():
+		// If the pipelinerun is cancelled, cancel tasks and update status
+		before := pr.Status.GetCondition(apis.ConditionSucceeded)
+		merr = multierror.Append(merr, cancelPipelineRun(c.Logger, pr, c.PipelineClientSet))
+		after := pr.Status.GetCondition(apis.ConditionSucceeded)
+		reconciler.EmitEvent(c.Recorder, before, after, pr)
+	default:
 		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)
 			c.Recorder.Event(pr, corev1.EventTypeWarning, eventReasonFailed, "Failed to create tracker for TaskRuns for PipelineRun")
@@ -228,7 +235,7 @@ func (c *Reconciler) Reconcile(ctx context.Context, key string) error {
 		}(c.metrics)
 	}
 
-	return merr
+	return merr.ErrorOrNil()
 }
 
 func (c *Reconciler) updatePipelineResults(ctx context.Context, pr *v1alpha1.PipelineRun) {
@@ -527,15 +534,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 {