Only fetch the definitions once 🧙

Since we "dereference" fetched definition and store them into the
`status`, we can use this as a source of truth. This allows us to
fetch only once the definitions (Task, Pipeline) *and* has the benefit
to make the runs (PipelineRun, TaskRun) immune to change to what they
refer, while the are executing.

Without this change, if you run a PipelineRun that reference a
Pipeline that is deleted during execution, your PipelineRun will fail
at some point because it cannot fetch the definition anymore, even if
it stored them in its status. This is the case for TaskRun too.

Signed-off-by: Vincent Demeester <[email protected]>

Author: vdemeester

pkg/reconciler/pipelinerun/pipelinerun.go

@@ -207,8 +207,8 @@ func (c *Reconciler) ReconcileKind(ctx context.Context, pr *v1beta1.PipelineRun)
 		return c.finishReconcileUpdateEmitEvents(ctx, pr, before, nil)
 	}
 
+	// If the pipelinerun is cancelled, cancel tasks and update status
 	if pr.IsCancelled() {
-		// If the pipelinerun is cancelled, cancel tasks and update status
 		err := cancelPipelineRun(ctx, logger, pr, c.PipelineClientSet)
 		return c.finishReconcileUpdateEmitEvents(ctx, pr, before, err)
 	}
@@ -278,7 +278,7 @@ func (c *Reconciler) resolvePipelineState(
 	pst := resources.PipelineRunState{}
 	// Resolve each task individually because they each could have a different reference context (remote or local).
 	for _, task := range tasks {
-		fn, _, err := tresources.GetTaskFunc(ctx, c.KubeClientSet, c.PipelineClientSet, task.TaskRef, pr.Namespace, pr.Spec.ServiceAccountName)
+		fn, err := tresources.GetTaskFunc(ctx, c.KubeClientSet, c.PipelineClientSet, task.TaskRef, pr.Namespace, pr.Spec.ServiceAccountName)
 		if err != nil {
 			// This Run has failed, so we need to mark it as failed and stop reconciling it
 			pr.Status.MarkFailed(ReasonCouldntGetTask, "Pipeline %s/%s can't be Run; task %s could not be fetched: %s",

pkg/reconciler/pipelinerun/resources/pipelineref.go

@@ -37,6 +37,18 @@ func GetPipelineFunc(ctx context.Context, k8s kubernetes.Interface, tekton clien
 	cfg := config.FromContextOrDefaults(ctx)
 	pr := pipelineRun.Spec.PipelineRef
 	namespace := pipelineRun.Namespace
+	// if the spec is already in the status, do not try to fetch it again, just use it as source of truth
+	if pipelineRun.Status.PipelineSpec != nil {
+		return func(_ context.Context, name string) (v1beta1.PipelineObject, error) {
+			return &v1beta1.Pipeline{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      name,
+					Namespace: namespace,
+				},
+				Spec: *pipelineRun.Status.PipelineSpec,
+			}, nil
+		}, nil
+	}
 	switch {
 	case cfg.FeatureFlags.EnableTektonOCIBundles && pr != nil && pr.Bundle != "":
 		// Return an inline function that implements GetTask by calling Resolver.Get with the specified task type and

pkg/reconciler/pipelinerun/resources/pipelineref_test.go

@@ -146,8 +146,7 @@ func TestGetPipelineFunc(t *testing.T) {
 			Name: "simple",
 		},
 		expected: tb.Pipeline("simple", tb.PipelineType, tb.PipelineNamespace("default"), tb.PipelineSpec(tb.PipelineTask("something", "something"))),
-	},
-	}
+	}}
 
 	for _, tc := range testcases {
 		t.Run(tc.name, func(t *testing.T) {
@@ -175,7 +174,7 @@ func TestGetPipelineFunc(t *testing.T) {
 				t.Fatalf("failed to get pipeline fn: %s", err.Error())
 			}
 
-			pipeline, err := fn(context.Background(), tc.ref.Name)
+			pipeline, err := fn(ctx, tc.ref.Name)
 			if err != nil {
 				t.Fatalf("failed to call pipelinefn: %s", err.Error())
 			}
@@ -186,3 +185,58 @@ func TestGetPipelineFunc(t *testing.T) {
 		})
 	}
 }
+
+func TestGetPipelineFuncSpecAlreadyFetched(t *testing.T) {
+	ctx := context.Background()
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+	tektonclient := fake.NewSimpleClientset(simplePipeline, dummyPipeline)
+	kubeclient := fakek8s.NewSimpleClientset(&v1.ServiceAccount{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: "default",
+			Name:      "default",
+		},
+	})
+
+	name := "anyname-really"
+	pipelineSpec := v1beta1.PipelineSpec{
+		Tasks: []v1beta1.PipelineTask{{
+			Name:    "task1",
+			TaskRef: &v1beta1.TaskRef{Name: "task"},
+		}},
+	}
+	pipelineRun := &v1beta1.PipelineRun{
+		ObjectMeta: metav1.ObjectMeta{Namespace: "default"},
+		Spec: v1beta1.PipelineRunSpec{
+			PipelineRef: &v1beta1.PipelineRef{
+				// Using simple here to show that, it won't fetch the simple pipelinespec,
+				// which is different from the pipelineSpec above
+				Name: "simple",
+			},
+			ServiceAccountName: "default",
+		},
+		Status: v1beta1.PipelineRunStatus{PipelineRunStatusFields: v1beta1.PipelineRunStatusFields{
+			PipelineSpec: &pipelineSpec,
+		}},
+	}
+	expectedPipeline := &v1beta1.Pipeline{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: "default",
+		},
+		Spec: pipelineSpec,
+	}
+
+	fn, err := resources.GetPipelineFunc(ctx, kubeclient, tektonclient, pipelineRun)
+	if err != nil {
+		t.Fatalf("failed to get pipeline fn: %s", err.Error())
+	}
+	actualPipeline, err := fn(ctx, name)
+	if err != nil {
+		t.Fatalf("failed to call pipelinefn: %s", err.Error())
+	}
+
+	if diff := cmp.Diff(actualPipeline, expectedPipeline); expectedPipeline != nil && diff != "" {
+		t.Error(diff)
+	}
+}

pkg/reconciler/pipelinerun/resources/pipelinerunresolution.go

@@ -386,16 +386,32 @@ func ResolvePipelineRunTask(
 			kind     v1beta1.TaskKind
 		)
 
+		taskRun, err := getTaskRun(rprt.TaskRunName)
+		if err != nil {
+			if !errors.IsNotFound(err) {
+				return nil, fmt.Errorf("error retrieving TaskRun %s: %w", rprt.TaskRunName, err)
+			}
+		}
+		if taskRun != nil {
+			rprt.TaskRun = taskRun
+		}
+
 		if task.TaskRef != nil {
-			t, err = getTask(ctx, task.TaskRef.Name)
-			if err != nil {
-				return nil, &TaskNotFoundError{
-					Name: task.TaskRef.Name,
-					Msg:  err.Error(),
+			// If the TaskRun has already a store TaskSpec in its status, use it as source of truth
+			if taskRun != nil && taskRun.Status.TaskSpec != nil {
+				spec = *taskRun.Status.TaskSpec
+				taskName = task.TaskRef.Name
+			} else {
+				t, err = getTask(ctx, task.TaskRef.Name)
+				if err != nil {
+					return nil, &TaskNotFoundError{
+						Name: task.TaskRef.Name,
+						Msg:  err.Error(),
+					}
 				}
+				spec = t.TaskSpec()
+				taskName = t.TaskMetadata().Name
 			}
-			spec = t.TaskSpec()
-			taskName = t.TaskMetadata().Name
 			kind = task.TaskRef.Kind
 		} else {
 			spec = task.TaskSpec.TaskSpec
@@ -408,16 +424,6 @@ func ResolvePipelineRunTask(
 
 		rprt.ResolvedTaskResources = rtr
 
-		taskRun, err := getTaskRun(rprt.TaskRunName)
-		if err != nil {
-			if !errors.IsNotFound(err) {
-				return nil, fmt.Errorf("error retrieving TaskRun %s: %w", rprt.TaskRunName, err)
-			}
-		}
-		if taskRun != nil {
-			rprt.TaskRun = taskRun
-		}
-
 		// Get all conditions that this pipelineTask will be using, if any
 		if len(task.Conditions) > 0 {
 			rcc, err := resolveConditionChecks(&task, pipelineRun.Status.TaskRuns, rprt.TaskRunName, getTaskRun, getCondition, providedResources)

pkg/reconciler/taskrun/resources/taskref.go

@@ -31,11 +31,40 @@ import (
 	"k8s.io/client-go/kubernetes"
 )
 
+// GetTaskKind returns the referenced Task kind (Task, ClusterTask, ...) if the TaskRun is using TaskRef.
+func GetTaskKind(taskrun *v1beta1.TaskRun) v1beta1.TaskKind {
+	kind := v1alpha1.NamespacedTaskKind
+	if taskrun.Spec.TaskRef != nil && taskrun.Spec.TaskRef.Kind != "" {
+		kind = taskrun.Spec.TaskRef.Kind
+	}
+	return kind
+}
+
+// GetTaskFuncFromTaskRun is a factory function that will use the given TaskRef as context to return a valid GetTask function. It
+// also requires a kubeclient, tektonclient, namespace, and service account in case it needs to find that task in
+// cluster or authorize against an external repositroy. It will figure out whether it needs to look in the cluster or in
+// a remote image to fetch the  reference. It will also return the "kind" of the task being referenced.
+func GetTaskFuncFromTaskRun(ctx context.Context, k8s kubernetes.Interface, tekton clientset.Interface, taskrun *v1beta1.TaskRun) (GetTask, error) {
+	// if the spec is already in the status, do not try to fetch it again, just use it as source of truth
+	if taskrun.Status.TaskSpec != nil {
+		return func(_ context.Context, name string) (v1beta1.TaskObject, error) {
+			return &v1beta1.Task{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      name,
+					Namespace: taskrun.Namespace,
+				},
+				Spec: *taskrun.Status.TaskSpec,
+			}, nil
+		}, nil
+	}
+	return GetTaskFunc(ctx, k8s, tekton, taskrun.Spec.TaskRef, taskrun.Namespace, taskrun.Spec.ServiceAccountName)
+}
+
 // GetTaskFunc is a factory function that will use the given TaskRef as context to return a valid GetTask function. It
 // also requires a kubeclient, tektonclient, namespace, and service account in case it needs to find that task in
 // cluster or authorize against an external repositroy. It will figure out whether it needs to look in the cluster or in
 // a remote image to fetch the  reference. It will also return the "kind" of the task being referenced.
-func GetTaskFunc(ctx context.Context, k8s kubernetes.Interface, tekton clientset.Interface, tr *v1beta1.TaskRef, namespace, saName string) (GetTask, v1beta1.TaskKind, error) {
+func GetTaskFunc(ctx context.Context, k8s kubernetes.Interface, tekton clientset.Interface, tr *v1beta1.TaskRef, namespace, saName string) (GetTask, error) {
 	cfg := config.FromContextOrDefaults(ctx)
 	kind := v1alpha1.NamespacedTaskKind
 	if tr != nil && tr.Kind != "" {
@@ -84,15 +113,15 @@ func GetTaskFunc(ctx context.Context, k8s kubernetes.Interface, tekton clientset
 			}
 
 			return nil, fmt.Errorf("failed to convert obj %s into Task", obj.GetObjectKind().GroupVersionKind().String())
-		}, kind, nil
+		}, nil
 	default:
 		// Even if there is no task ref, we should try to return a local resolver.
 		local := &LocalTaskRefResolver{
 			Namespace:    namespace,
 			Kind:         kind,
 			Tektonclient: tekton,
 		}
-		return local.GetTask, kind, nil
+		return local.GetTask, nil
 	}
 }
 

pkg/reconciler/taskrun/resources/taskref_test.go

@@ -226,15 +226,11 @@ func TestGetTaskFunc(t *testing.T) {
 				t.Fatalf("failed to upload test image: %s", err.Error())
 			}
 
-			fn, kind, err := resources.GetTaskFunc(ctx, kubeclient, tektonclient, tc.ref, "default", "default")
+			fn, err := resources.GetTaskFunc(ctx, kubeclient, tektonclient, tc.ref, "default", "default")
 			if err != nil {
 				t.Fatalf("failed to get task fn: %s", err.Error())
 			}
 
-			if string(tc.expectedKind) != string(kind) {
-				t.Errorf("expected kind %s did not match actual kind %s", tc.expectedKind, kind)
-			}
-
 			task, err := fn(ctx, tc.ref.Name)
 			if err != nil {
 				t.Fatalf("failed to call taskfn: %s", err.Error())
@@ -246,3 +242,63 @@ func TestGetTaskFunc(t *testing.T) {
 		})
 	}
 }
+
+func TestGetTaskFuncFromTaskRunSpecAlreadyFetched(t *testing.T) {
+	ctx := context.Background()
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+	tektonclient := fake.NewSimpleClientset(tb.Task("simple", tb.TaskType, tb.TaskNamespace("default"), tb.TaskSpec(tb.Step("something"))))
+	kubeclient := fakek8s.NewSimpleClientset(&v1.ServiceAccount{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: "default",
+			Name:      "default",
+		},
+	})
+
+	name := "anyname-really"
+	TaskSpec := v1beta1.TaskSpec{
+		Steps: []v1beta1.Step{{
+			Container: corev1.Container{
+				Image: "myimage",
+			},
+			Script: `
+#!/usr/bin/env bash
+echo hello
+`,
+		}},
+	}
+	TaskRun := &v1beta1.TaskRun{
+		ObjectMeta: metav1.ObjectMeta{Namespace: "default"},
+		Spec: v1beta1.TaskRunSpec{
+			TaskRef: &v1beta1.TaskRef{
+				// Using simple here to show that, it won't fetch the simple Taskspec,
+				// which is different from the TaskSpec above
+				Name: "simple",
+			},
+			ServiceAccountName: "default",
+		},
+		Status: v1beta1.TaskRunStatus{TaskRunStatusFields: v1beta1.TaskRunStatusFields{
+			TaskSpec: &TaskSpec,
+		}},
+	}
+	expectedTask := &v1beta1.Task{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: "default",
+		},
+		Spec: TaskSpec,
+	}
+
+	fn, err := resources.GetTaskFuncFromTaskRun(ctx, kubeclient, tektonclient, TaskRun)
+	if err != nil {
+		t.Fatalf("failed to get Task fn: %s", err.Error())
+	}
+	actualTask, err := fn(ctx, name)
+	if err != nil {
+		t.Fatalf("failed to call Taskfn: %s", err.Error())
+	}
+
+	if diff := cmp.Diff(actualTask, expectedTask); expectedTask != nil && diff != "" {
+		t.Error(diff)
+	}
+}

pkg/reconciler/taskrun/taskrun.go

@@ -271,7 +271,7 @@ func (c *Reconciler) prepare(ctx context.Context, tr *v1beta1.TaskRun) (*v1beta1
 	// and may not have had all of the assumed default specified.
 	tr.SetDefaults(contexts.WithUpgradeViaDefaulting(ctx))
 
-	getTaskfunc, kind, err := resources.GetTaskFunc(ctx, c.KubeClientSet, c.PipelineClientSet, tr.Spec.TaskRef, tr.Namespace, tr.Spec.ServiceAccountName)
+	getTaskfunc, err := resources.GetTaskFuncFromTaskRun(ctx, c.KubeClientSet, c.PipelineClientSet, tr)
 	if err != nil {
 		logger.Errorf("Failed to fetch task reference %s: %v", tr.Spec.TaskRef.Name, err)
 		tr.Status.SetCondition(&apis.Condition{
@@ -324,7 +324,7 @@ func (c *Reconciler) prepare(ctx context.Context, tr *v1beta1.TaskRun) (*v1beta1
 		inputs = tr.Spec.Resources.Inputs
 		outputs = tr.Spec.Resources.Outputs
 	}
-	rtr, err := resources.ResolveTaskResources(taskSpec, taskMeta.Name, kind, inputs, outputs, c.resourceLister.PipelineResources(tr.Namespace).Get)
+	rtr, err := resources.ResolveTaskResources(taskSpec, taskMeta.Name, resources.GetTaskKind(tr), inputs, outputs, c.resourceLister.PipelineResources(tr.Namespace).Get)
 	if err != nil {
 		if k8serrors.IsNotFound(err) && tknreconciler.IsYoungResource(tr) {
 			// For newly created resources, don't fail immediately.