Add Support for managedBy field in TaskRun and PipelineRun

Added a "managedBy" field to delegate responsibility of controlling
the lifecycle of PipelineRuns/TaskRuns.

The semantics of the field:

Whenever the value is set, and it does not point to the built-in
controller, then we skip the reconciliation.
* The field is immutable
* The field is not defaulted

Author: khrm

pkg/apis/pipeline/register.go

@@ -55,6 +55,10 @@ const (
 	// MemberOfLabelKey is used as the label identifier for a PipelineTask
 	// Set to Tasks/Finally depending on the position of the PipelineTask
 	MemberOfLabelKey = GroupName + "/memberOf"
+
+	// PipelineManagedBy is the value of the "managedBy" field for resources
+	// managed by the Tekton Pipeline controller.
+	PipelineManagedBy = GroupName + "/pipeline"
 )
 
 var (

pkg/apis/pipeline/v1/pipelinerun_types.go

@@ -36,6 +36,12 @@ import (
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 )
 
+const (
+	// PipelineRunByTektonController is the value for the managedBy field to indicate
+	// that the PipelineRun is managed by the Tekton Pipelines controller.
+	PipelineRunByTektonController = "tekton.dev/pipeline"
+)
+
 // +genclient
 // +genreconciler:krshapedlogic=false
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
@@ -283,6 +289,10 @@ type PipelineRunSpec struct {
 	// +optional
 	// +listType=atomic
 	TaskRunSpecs []PipelineTaskRunSpec `json:"taskRunSpecs,omitempty"`
+	// ManagedBy is the platform that is managing this PipelineRun.
+	// This field is immutable.
+	// +optional
+	ManagedBy string `json:"managedBy,omitempty"`
 }
 
 // TimeoutFields allows granular specification of pipeline, task, and finally timeouts

pkg/apis/pipeline/v1/pipelinerun_validation.go

@@ -140,14 +140,18 @@ func (ps *PipelineRunSpec) ValidateUpdate(ctx context.Context) (errs *apis.Field
 	old := &oldObj.Spec
 
 	// If already in the done state, the spec cannot be modified. Otherwise, only the status field can be modified.
-	tips := "Once the PipelineRun is complete, no updates are allowed"
-	if !oldObj.IsDone() {
-		old = old.DeepCopy()
-		old.Status = ps.Status
-		tips = "Once the PipelineRun has started, only status updates are allowed"
-	}
-	if !equality.Semantic.DeepEqual(old, ps) {
-		errs = errs.Also(apis.ErrInvalidValue(tips, ""))
+	if old.ManagedBy != ps.ManagedBy {
+		errs = errs.Also(apis.ErrInvalidValue("managedBy is immutable", "spec.managedBy"))
+	} else {
+		tips := "Once the PipelineRun is complete, no updates are allowed"
+		if !oldObj.IsDone() {
+			old = old.DeepCopy()
+			old.Status = ps.Status
+			tips = "Once the PipelineRun has started, only status updates are allowed"
+		}
+		if !equality.Semantic.DeepEqual(old, ps) {
+			errs = errs.Also(apis.ErrInvalidValue(tips, ""))
+		}
 	}
 
 	return

pkg/apis/pipeline/v1/pipelinerun_validation_test.go

@@ -1667,6 +1667,56 @@ func TestPipelineRunSpec_ValidateUpdate(t *testing.T) {
 				Message: `invalid value: Once the PipelineRun is complete, no updates are allowed`,
 				Paths:   []string{""},
 			},
+		}, {
+			name: "is update ctx, baseline is not done, managedBy changes",
+			baselinePipelineRun: &v1.PipelineRun{
+				Spec: v1.PipelineRunSpec{
+					ManagedBy: "tekton.dev/pipeline",
+				},
+				Status: v1.PipelineRunStatus{
+					Status: duckv1.Status{
+						Conditions: duckv1.Conditions{
+							{Type: apis.ConditionSucceeded, Status: corev1.ConditionUnknown},
+						},
+					},
+				},
+			},
+			pipelineRun: &v1.PipelineRun{
+				Spec: v1.PipelineRunSpec{
+					ManagedBy: "some-other-controller",
+				},
+			},
+			isCreate: false,
+			isUpdate: true,
+			expectedError: apis.FieldError{
+				Message: `invalid value: managedBy is immutable`,
+				Paths:   []string{"spec.managedBy"},
+			},
+		}, {
+			name: "is update ctx, baseline is unknown, managedBy changes",
+			baselinePipelineRun: &v1.PipelineRun{
+				Spec: v1.PipelineRunSpec{
+					ManagedBy: "tekton.dev/pipeline",
+				},
+				Status: v1.PipelineRunStatus{
+					Status: duckv1.Status{
+						Conditions: duckv1.Conditions{
+							{Type: apis.ConditionSucceeded, Status: corev1.ConditionUnknown},
+						},
+					},
+				},
+			},
+			pipelineRun: &v1.PipelineRun{
+				Spec: v1.PipelineRunSpec{
+					ManagedBy: "some-other-controller",
+				},
+			},
+			isCreate: false,
+			isUpdate: true,
+			expectedError: apis.FieldError{
+				Message: `invalid value: managedBy is immutable`,
+				Paths:   []string{"spec.managedBy"},
+			},
 		},
 	}
 

pkg/apis/pipeline/v1/taskrun_types.go

@@ -85,6 +85,10 @@ type TaskRunSpec struct {
 	SidecarSpecs []TaskRunSidecarSpec `json:"sidecarSpecs,omitempty"`
 	// Compute resources to use for this TaskRun
 	ComputeResources *corev1.ResourceRequirements `json:"computeResources,omitempty"`
+	// ManagedBy is the platform that is managing this TaskRun.
+	// This field is immutable.
+	// +optional
+	ManagedBy string `json:"managedBy,omitempty"`
 }
 
 // TaskRunSpecStatus defines the TaskRun spec status the user can provide

pkg/apis/pipeline/v1/taskrun_validation.go

@@ -135,16 +135,20 @@ func (ts *TaskRunSpec) ValidateUpdate(ctx context.Context) (errs *apis.FieldErro
 
 	// If already in the done state, the spec cannot be modified.
 	// Otherwise, only the status, statusMessage field can be modified.
-	tips := "Once the TaskRun is complete, no updates are allowed"
-	if !oldObj.IsDone() {
-		old = old.DeepCopy()
-		old.Status = ts.Status
-		old.StatusMessage = ts.StatusMessage
-		tips = "Once the TaskRun has started, only status and statusMessage updates are allowed"
-	}
+	if old.ManagedBy != ts.ManagedBy {
+		errs = errs.Also(apis.ErrInvalidValue("managedBy is immutable", "spec.managedBy"))
+	} else {
+		tips := "Once the TaskRun is complete, no updates are allowed"
+		if !oldObj.IsDone() {
+			old = old.DeepCopy()
+			old.Status = ts.Status
+			old.StatusMessage = ts.StatusMessage
+			tips = "Once the TaskRun has started, only status and statusMessage updates are allowed"
+		}
 
-	if !equality.Semantic.DeepEqual(old, ts) {
-		errs = errs.Also(apis.ErrInvalidValue(tips, ""))
+		if !equality.Semantic.DeepEqual(old, ts) {
+			errs = errs.Also(apis.ErrInvalidValue(tips, ""))
+		}
 	}
 
 	return

pkg/apis/pipeline/v1/taskrun_validation_test.go

@@ -1107,6 +1107,56 @@ func TestTaskRunSpec_ValidateUpdate(t *testing.T) {
 				Message: `invalid value: Once the TaskRun is complete, no updates are allowed`,
 				Paths:   []string{""},
 			},
+		}, {
+			name: "is update ctx, baseline is not done, managedBy changes",
+			baselineTaskRun: &v1.TaskRun{
+				Spec: v1.TaskRunSpec{
+					ManagedBy: "tekton.dev/pipeline",
+				},
+				Status: v1.TaskRunStatus{
+					Status: duckv1.Status{
+						Conditions: duckv1.Conditions{
+							{Type: apis.ConditionSucceeded, Status: corev1.ConditionUnknown},
+						},
+					},
+				},
+			},
+			taskRun: &v1.TaskRun{
+				Spec: v1.TaskRunSpec{
+					ManagedBy: "some-other-controller",
+				},
+			},
+			isCreate: false,
+			isUpdate: true,
+			expectedError: apis.FieldError{
+				Message: `invalid value: managedBy is immutable`,
+				Paths:   []string{"spec.managedBy"},
+			},
+		}, {
+			name: "is update ctx, baseline is unknown, managedBy changes",
+			baselineTaskRun: &v1.TaskRun{
+				Spec: v1.TaskRunSpec{
+					ManagedBy: "tekton.dev/pipeline",
+				},
+				Status: v1.TaskRunStatus{
+					Status: duckv1.Status{
+						Conditions: duckv1.Conditions{
+							{Type: apis.ConditionSucceeded, Status: corev1.ConditionUnknown},
+						},
+					},
+				},
+			},
+			taskRun: &v1.TaskRun{
+				Spec: v1.TaskRunSpec{
+					ManagedBy: "some-other-controller",
+				},
+			},
+			isCreate: false,
+			isUpdate: true,
+			expectedError: apis.FieldError{
+				Message: `invalid value: managedBy is immutable`,
+				Paths:   []string{"spec.managedBy"},
+			},
 		},
 	}
 

pkg/reconciler/pipelinerun/controller.go

@@ -96,7 +96,17 @@ func NewController(opts *pipeline.Options, clock clock.PassiveClock) func(contex
 			logging.FromContext(ctx).Panicf("Couldn't register Secret informer event handler: %w", err)
 		}
 
-		if _, err := pipelineRunInformer.Informer().AddEventHandler(controller.HandleAll(impl.Enqueue)); err != nil {
+		if _, err := pipelineRunInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
+			FilterFunc: func(obj interface{}) bool {
+				if pr, ok := obj.(*v1.PipelineRun); ok {
+					if pr.Spec.ManagedBy != "" && pr.Spec.ManagedBy != v1.PipelineRunByTektonController {
+						return false
+					}
+				}
+				return true
+			},
+			Handler: controller.HandleAll(impl.Enqueue),
+		}); err != nil {
 			logging.FromContext(ctx).Panicf("Couldn't register PipelineRun informer event handler: %w", err)
 		}
 

pkg/reconciler/pipelinerun/pipelinerun_test.go

@@ -47,6 +47,7 @@ import (
 	"github.com/tektoncd/pipeline/pkg/reconciler/pipelinerun/resources"
 	taskresources "github.com/tektoncd/pipeline/pkg/reconciler/taskrun/resources"
 	th "github.com/tektoncd/pipeline/pkg/reconciler/testing"
+	ttesting "github.com/tektoncd/pipeline/pkg/reconciler/testing"
 	"github.com/tektoncd/pipeline/pkg/reconciler/volumeclaim"
 	resolutioncommon "github.com/tektoncd/pipeline/pkg/resolution/common"
 	remoteresource "github.com/tektoncd/pipeline/pkg/resolution/resource"
@@ -18222,3 +18223,89 @@ spec:
 		})
 	}
 }
+
+func TestReconcile_ManagedBy(t *testing.T) {
+	namespace := "foo"
+	prManagedByTektonName := "test-pr-managed-by-tekton"
+	prManagedByOtherName := "test-pr-managed-by-other"
+
+	ps := []*v1.Pipeline{simpleHelloWorldPipeline}
+	ts := []*v1.Task{simpleHelloWorldTask}
+
+	prs := []*v1.PipelineRun{
+		parse.MustParseV1PipelineRun(t, fmt.Sprintf(`
+metadata:
+  name: %s
+  namespace: %s
+spec:
+  pipelineRef:
+    name: test-pipeline
+  managedBy: tekton.dev/pipeline
+`, prManagedByTektonName, namespace)),
+		parse.MustParseV1PipelineRun(t, fmt.Sprintf(`
+metadata:
+  name: %s
+  namespace: %s
+spec:
+  pipelineRef:
+    name: test-pipeline
+  managedBy: other-controller
+`, prManagedByOtherName, namespace)),
+	}
+
+	// This data is filtered and simulates what the informer would pass to the reconciler.
+	// It only contains the PipelineRun that should be reconciled.
+	filteredData := test.Data{
+		PipelineRuns: []*v1.PipelineRun{prs[0]}, // Only the one managed by Tekton
+		Pipelines:    ps,
+		Tasks:        ts,
+		ConfigMaps:   th.NewFeatureFlagsConfigMapInSlice(),
+	}
+
+	// Initialize controller with filtered data for the listers
+	prt := newPipelineRunTest(t, filteredData)
+	defer prt.Cancel()
+
+	// Create clients with all the data for verification
+	ctx, _ := ttesting.SetupFakeContext(t)
+	clients, _ := test.SeedTestData(t, ctx, test.Data{
+		PipelineRuns: prs,
+		Pipelines:    ps,
+		Tasks:        ts,
+		ConfigMaps:   th.NewFeatureFlagsConfigMapInSlice(),
+	})
+
+	// Verify no TaskRuns were created for the externally managed PipelineRun
+	taskRunsOther := getTaskRunsForPipelineRun(prt.TestAssets.Ctx, t, clients, namespace, prManagedByOtherName)
+	if len(taskRunsOther) != 0 {
+		t.Errorf("Expected no TaskRuns for externally managed PipelineRun, but found %d", len(taskRunsOther))
+	}
+
+	// Verify its status is unchanged
+	reconciledOther, err := clients.Pipeline.TektonV1().PipelineRuns(namespace).Get(prt.TestAssets.Ctx, prManagedByOtherName, metav1.GetOptions{})
+	if err != nil {
+		t.Fatalf("Failed to get externally managed PipelineRun: %v", err)
+	}
+	if len(reconciledOther.Status.Conditions) != 0 {
+		t.Errorf("Expected externally managed PipelineRun to have no conditions, but it did")
+	}
+
+	// 2. Reconcile the PipelineRun managed by "tekton.dev/pipeline"
+	// We expect this one to be processed normally.
+	wantEvents := []string{
+		"Normal Started",
+		"Normal Running Tasks Completed: 0",
+	}
+	reconciledTekton, clients := prt.reconcileRun(namespace, prManagedByTektonName, wantEvents, false)
+
+	// Verify a TaskRun was created for the Tekton-managed PipelineRun
+	taskRunsTekton := getTaskRunsForPipelineRun(prt.TestAssets.Ctx, t, clients, namespace, prManagedByTektonName)
+	if len(taskRunsTekton) != 1 {
+		t.Errorf("Expected 1 TaskRun for Tekton-managed PipelineRun, but found %d", len(taskRunsTekton))
+	}
+
+	// Verify its status was updated
+	if !reconciledTekton.Status.GetCondition(apis.ConditionSucceeded).IsUnknown() {
+		t.Errorf("Expected Tekton-managed PipelineRun to be running, but it was not")
+	}
+}

pkg/reconciler/taskrun/controller.go

@@ -107,7 +107,21 @@ func NewController(opts *pipeline.Options, clock clock.PassiveClock) func(contex
 			logging.FromContext(ctx).Panicf("Couldn't register Secret informer event handler: %w", err)
 		}
 
-		if _, err := taskRunInformer.Informer().AddEventHandler(controller.HandleAll(impl.Enqueue)); err != nil {
+		if _, err := taskRunInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
+			FilterFunc: func(obj interface{}) bool {
+				tr, ok := obj.(*v1.TaskRun)
+				if !ok {
+					// We don't care about anything other than TaskRuns.
+					return true
+				}
+				// The taskrun-controller should not reconcile TaskRuns managed by other controllers.
+				if tr.Spec.ManagedBy != "" && tr.Spec.ManagedBy != v1.PipelineRunByTektonController {
+					return false
+				}
+				return true
+			},
+			Handler: controller.HandleAll(impl.Enqueue),
+		}); err != nil {
 			logging.FromContext(ctx).Panicf("Couldn't register TaskRun informer event handler: %w", err)
 		}