Propagate labels from Pipeline/Task to PipelineRun/TaskRun

With this change, labels are propagated from Pipeline and Task to
PipelineRun and TaskRun, respectively, giving us full label propagation
from Pipeline to PipelineRun to TaskRun to Pod and Task to TaskRun to
Pod.

This commit also adds a label whose key is pipeline.knative.dev/task
to all TaskRuns that refer to a Task with a TaskRef (the label is not
added to TaskRuns using an embedded TaskSpec) that contains the name of
the Task.

In addition, this commit introduces a builder for Pod values to increase
the readability of the taskrun reconciliation tests.

Fixes tektoncd#501

docs/pipelineruns.md

@@ -15,6 +15,7 @@ Creation of a `PipelineRun` will trigger the creation of
 - [Syntax](#syntax)
   - [Resources](#resources)
   - [Service account](#service-account)
+- [Labels](#labels)
 - [Cancelling a PipelineRun](#cancelling-a-pipelinerun)
 - [Examples](#examples)
 
@@ -110,11 +111,11 @@ labels will be added automatically:
 These labels make it easier to find the resources that are associated with a
 given pipeline.
 
-For example, to find all `Pods` created by a `Pipeline` named test-pipeline, you
+For example, to find all `Pods` created by a `PipelineRun` named test-pipelinerun, you
 could use the following command:
 
 ```shell
-kubectl get pods --all-namespaces -l pipeline.knative.dev/pipeline=test-pipeline
+kubectl get pods --all-namespaces -l pipeline.knative.dev/pipelineRun=test-pipelinerun
 ```
 
 ## Cancelling a PipelineRun

docs/pipelines.md

@@ -9,6 +9,7 @@ This document defines `Pipelines` and their capabilities.
   - [Parameters](#parameters)
   - [Pipeline Tasks](#pipeline-tasks)
     - [From](#from)
+- [Labels](#labels)
 - [Examples](#examples)
 
 ## Syntax
@@ -201,6 +202,26 @@ The resource `my-image` is expected to be given to the `deploy-app` `Task` from
 the `build-app` `Task`. This means that the `PipelineResource` `my-image` must
 also be declared as an output of `build-app`.
 
+## Labels
+
+Any labels specified in the metadata field of a `Pipeline` will be propagated
+to any `PipelineRuns` that refer to the `Task` (and then propagated by those
+`PipelineRuns` to the `TaskRuns` created automatically for each `Task` in the
+`Pipeline` and then to the `Pods` created for those `TaskRuns`). In addition,
+the following label will be added automatically:
+
+- `pipeline.knative.dev/pipeline` will contain the name of the `Pipeline`
+
+This label makes it easier to find the resources that are associated with a
+given `Pipeline`.
+
+For example, to find all `PipelineRuns` for a `Pipeline` named test-pipeline, you
+could use the following command:
+
+```shell
+kubectl get pipelineruns --all-namespaces -l pipeline.knative.dev/pipeline=test-pipeline
+```
+
 ## Examples
 
 For complete examples, see

docs/taskruns.md

@@ -229,9 +229,10 @@ spec:
 ## Labels
 
 Any labels specified in the metadata field of a `TaskRun` will be propagated to
-the `Pod` created to execute the `Task`. In addition, the following label will
+the `Pod` created to execute the `Task`. In addition, the following labels will
 be added automatically:
 
+- `pipeline.knative.dev/task` will contain the name of the `Task`
 - `pipeline.knative.dev/taskRun` will contain the name of the `TaskRun`
 
 If the `TaskRun` was created automatically by a `PipelineRun`, then the

docs/tasks.md

@@ -23,6 +23,7 @@ entire Kubernetes cluster.
   - [Controlling where resources are mounted](#controlling-where-resources-are-mounted)
   - [Volumes](#volumes)
   - [Templating](#templating)
+- [Labels](#labels)
 - [Examples](#examples)
 
 ## ClusterTask
@@ -340,6 +341,24 @@ ${inputs.params.<name>}
 ```
 **Note**: Task volume names and volume source(current support includes only configmap) can also be parameterized as shown in [example](#using-kubernetes-configmap-as-volume-source)
 
+## Labels
+
+Any labels specified in the metadata field of a `Task` will be propagated to any
+`TaskRuns` that refer to the task (and then to any `Pods` created to execute
+those `TaskRuns`). In addition, the following label will be added automatically:
+
+- `pipeline.knative.dev/task` will contain the name of the `Task`
+
+This label makes it easier to find the resources that are associated with a
+given `Task`.
+
+For example, to find all `TaskRuns` for a `Task` named test-task, you
+could use the following command:
+
+```shell
+kubectl get taskruns --all-namespaces -l pipeline.knative.dev/task=test-task
+```
+
 ## Examples
 
 Use these code snippets to help you understand how to define your `Tasks`.

pkg/apis/pipeline/register.go

@@ -19,6 +19,7 @@ package pipeline
 // GroupName is the Kubernetes resource group name for Pipeline types.
 const (
 	GroupName           = "pipeline.knative.dev"
+	TaskLabelKey        = "/task"
 	TaskRunLabelKey     = "/taskRun"
 	PipelineLabelKey    = "/pipeline"
 	PipelineRunLabelKey = "/pipelineRun"

pkg/reconciler/v1alpha1/pipelinerun/pipelinerun.go

@@ -171,15 +171,16 @@ func (c *Reconciler) Reconcile(ctx context.Context, key string) error {
 		return err
 	}
 
-	// Reconcile this copy of the task run and then write back any status
+	// Reconcile this copy of the task run and then write back any status or label
 	// updates regardless of whether the reconciliation errored out.
 	err = c.reconcile(ctx, pr)
-	if equality.Semantic.DeepEqual(original.Status, pr.Status) {
-		// If we didn't change anything then don't call updateStatus.
+	if equality.Semantic.DeepEqual(original.Status, pr.Status) &&
+		reflect.DeepEqual(original.ObjectMeta.Labels, pr.ObjectMeta.Labels) {
+		// If we didn't change anything then don't call updateStatusAndLabels.
 		// This is important because the copy we loaded from the informer's
 		// cache may be stale and we don't want to overwrite a prior update
 		// to status with this stale state.
-	} else if _, err := c.updateStatus(pr); err != nil {
+	} else if _, err := c.updateStatusAndLabels(pr); err != nil {
 		c.Logger.Warn("Failed to update PipelineRun status", zap.Error(err))
 		c.Recorder.Event(pr, corev1.EventTypeWarning, eventReasonFailed, "PipelineRun failed to update")
 		return err
@@ -225,6 +226,15 @@ func (c *Reconciler) reconcile(ctx context.Context, pr *v1alpha1.PipelineRun) er
 	// Apply parameter templating from the PipelineRun
 	p = resources.ApplyParameters(p, pr)
 
+	// Propagate labels from Pipeline to PipelineRun.
+	if pr.ObjectMeta.Labels == nil {
+		pr.ObjectMeta.Labels = make(map[string]string, len(p.ObjectMeta.Labels)+1)
+	}
+	for key, value := range p.ObjectMeta.Labels {
+		pr.ObjectMeta.Labels[key] = value
+	}
+	pr.ObjectMeta.Labels[pipeline.GroupName+pipeline.PipelineLabelKey] = p.Name
+
 	pipelineState, err := resources.ResolvePipelineRun(
 		*pr,
 		func(name string) (v1alpha1.TaskInterface, error) {
@@ -360,11 +370,11 @@ func (c *Reconciler) createTaskRun(logger *zap.SugaredLogger, rprt *resources.Re
 		taskRunTimeout = nil
 	}
 
-	labels := make(map[string]string, len(pr.ObjectMeta.Labels)+2)
+	// Propagate labels from PipelineRun to TaskRun.
+	labels := make(map[string]string, len(pr.ObjectMeta.Labels)+1)
 	for key, val := range pr.ObjectMeta.Labels {
 		labels[key] = val
 	}
-	labels[pipeline.GroupName+pipeline.PipelineLabelKey] = pr.Spec.PipelineRef.Name
 	labels[pipeline.GroupName+pipeline.PipelineRunLabelKey] = pr.Name
 
 	tr := &v1alpha1.TaskRun{
@@ -392,14 +402,16 @@ func (c *Reconciler) createTaskRun(logger *zap.SugaredLogger, rprt *resources.Re
 	return c.PipelineClientSet.PipelineV1alpha1().TaskRuns(pr.Namespace).Create(tr)
 }
 
-func (c *Reconciler) updateStatus(pr *v1alpha1.PipelineRun) (*v1alpha1.PipelineRun, error) {
+func (c *Reconciler) updateStatusAndLabels(pr *v1alpha1.PipelineRun) (*v1alpha1.PipelineRun, error) {
 	newPr, err := c.pipelineRunLister.PipelineRuns(pr.Namespace).Get(pr.Name)
 	if err != nil {
 		return nil, fmt.Errorf("Error getting PipelineRun %s when updating status: %s", pr.Name, err)
 	}
-	if !reflect.DeepEqual(pr.Status, newPr.Status) {
+	if !reflect.DeepEqual(pr.Status, newPr.Status) ||
+		!reflect.DeepEqual(pr.ObjectMeta.Labels, newPr.ObjectMeta.Labels) {
 		newPr.Status = pr.Status
-		return c.PipelineClientSet.PipelineV1alpha1().PipelineRuns(pr.Namespace).UpdateStatus(newPr)
+		newPr.ObjectMeta.Labels = pr.ObjectMeta.Labels
+		return c.PipelineClientSet.PipelineV1alpha1().PipelineRuns(pr.Namespace).Update(newPr)
 	}
 	return newPr, nil
 }

pkg/reconciler/v1alpha1/taskrun/resources/taskspec.go

@@ -20,6 +20,7 @@ import (
 	"fmt"
 
 	"github.com/knative/build-pipeline/pkg/apis/pipeline/v1alpha1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
 // GetTask is a function used to retrieve Tasks.
@@ -28,24 +29,25 @@ type GetTask func(string) (v1alpha1.TaskInterface, error)
 // GetClusterTask is a function that will retrieve the Task from name and namespace.
 type GetClusterTask func(name string) (v1alpha1.TaskInterface, error)
 
-// GetTaskSpec will retrieve the Task Spec associated with the provieded TaskRun. This can come from a
-// reference Task or from an embedded Task spec.
-func GetTaskSpec(taskRunSpec *v1alpha1.TaskRunSpec, taskRunName string, getTask GetTask) (*v1alpha1.TaskSpec, string, error) {
+// GetTaskData will retrieve the Task metadata and Spec associated with the
+// provided TaskRun. This can come from a reference Task or from the TaskRun's
+// metadata and embedded TaskSpec.
+func GetTaskData(taskRun *v1alpha1.TaskRun, getTask GetTask) (*metav1.ObjectMeta, *v1alpha1.TaskSpec, error) {
+	taskMeta := metav1.ObjectMeta{}
 	taskSpec := v1alpha1.TaskSpec{}
-	taskName := ""
-	if taskRunSpec.TaskRef != nil && taskRunSpec.TaskRef.Name != "" {
+	if taskRun.Spec.TaskRef != nil && taskRun.Spec.TaskRef.Name != "" {
 		// Get related task for taskrun
-		t, err := getTask(taskRunSpec.TaskRef.Name)
+		t, err := getTask(taskRun.Spec.TaskRef.Name)
 		if err != nil {
-			return nil, taskName, fmt.Errorf("error when listing tasks for taskRun %s %v", taskRunName, err)
+			return nil, nil, fmt.Errorf("error when listing tasks for taskRun %s %v", taskRun.Name, err)
 		}
+		taskMeta = t.TaskMetadata()
 		taskSpec = t.TaskSpec()
-		taskName = t.TaskMetadata().Name
-	} else if taskRunSpec.TaskSpec != nil {
-		taskSpec = *taskRunSpec.TaskSpec
-		taskName = taskRunName
+	} else if taskRun.Spec.TaskSpec != nil {
+		taskMeta = taskRun.ObjectMeta
+		taskSpec = *taskRun.Spec.TaskSpec
 	} else {
-		return &taskSpec, taskName, fmt.Errorf("TaskRun %s not providing TaskRef or TaskSpec", taskRunName)
+		return nil, nil, fmt.Errorf("TaskRun %s not providing TaskRef or TaskSpec", taskRun.Name)
 	}
-	return &taskSpec, taskName, nil
+	return &taskMeta, &taskSpec, nil
 }

pkg/reconciler/v1alpha1/taskrun/resources/taskspec_test.go

@@ -36,20 +36,25 @@ func TestGetTaskSpec_Ref(t *testing.T) {
 			}},
 		},
 	}
-	spec := &v1alpha1.TaskRunSpec{
-		TaskRef: &v1alpha1.TaskRef{
-			Name: "orchestrate",
+	tr := &v1alpha1.TaskRun{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "mytaskrun",
+		},
+		Spec: v1alpha1.TaskRunSpec{
+			TaskRef: &v1alpha1.TaskRef{
+				Name: "orchestrate",
+			},
 		},
 	}
 	gt := func(n string) (v1alpha1.TaskInterface, error) { return task, nil }
-	taskSpec, name, err := GetTaskSpec(spec, "mytaskrun", gt)
+	taskMeta, taskSpec, err := GetTaskData(tr, gt)
 
 	if err != nil {
 		t.Fatalf("Did not expect error getting task spec but got: %s", err)
 	}
 
-	if name != "orchestrate" {
-		t.Errorf("Expected task name to be `orchestrate` but was %q", name)
+	if taskMeta.Name != "orchestrate" {
+		t.Errorf("Expected task name to be `orchestrate` but was %q", taskMeta.Name)
 	}
 
 	if len(taskSpec.Steps) != 1 || taskSpec.Steps[0].Name != "step1" {
@@ -58,21 +63,27 @@ func TestGetTaskSpec_Ref(t *testing.T) {
 }
 
 func TestGetTaskSpec_Embedded(t *testing.T) {
-	spec := &v1alpha1.TaskRunSpec{
-		TaskSpec: &v1alpha1.TaskSpec{
-			Steps: []corev1.Container{{
-				Name: "step1",
-			}},
-		}}
+	tr := &v1alpha1.TaskRun{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "mytaskrun",
+		},
+		Spec: v1alpha1.TaskRunSpec{
+			TaskSpec: &v1alpha1.TaskSpec{
+				Steps: []corev1.Container{{
+					Name: "step1",
+				}},
+			},
+		},
+	}
 	gt := func(n string) (v1alpha1.TaskInterface, error) { return nil, fmt.Errorf("shouldn't be called") }
-	taskSpec, name, err := GetTaskSpec(spec, "mytaskrun", gt)
+	taskMeta, taskSpec, err := GetTaskData(tr, gt)
 
 	if err != nil {
 		t.Fatalf("Did not expect error getting task spec but got: %s", err)
 	}
 
-	if name != "mytaskrun" {
-		t.Errorf("Expected task name for embedded task to default to name of task run but was %q", name)
+	if taskMeta.Name != "mytaskrun" {
+		t.Errorf("Expected task name for embedded task to default to name of task run but was %q", taskMeta.Name)
 	}
 
 	if len(taskSpec.Steps) != 1 || taskSpec.Steps[0].Name != "step1" {
@@ -81,22 +92,31 @@ func TestGetTaskSpec_Embedded(t *testing.T) {
 }
 
 func TestGetTaskSpec_Invalid(t *testing.T) {
-	spec := &v1alpha1.TaskRunSpec{}
+	tr := &v1alpha1.TaskRun{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "mytaskrun",
+		},
+	}
 	gt := func(n string) (v1alpha1.TaskInterface, error) { return nil, fmt.Errorf("shouldn't be called") }
-	_, _, err := GetTaskSpec(spec, "mytaskrun", gt)
+	_, _, err := GetTaskData(tr, gt)
 	if err == nil {
 		t.Fatalf("Expected error resolving spec with no embedded or referenced task spec but didn't get error")
 	}
 }
 
 func TestGetTaskSpec_Error(t *testing.T) {
-	spec := &v1alpha1.TaskRunSpec{
-		TaskRef: &v1alpha1.TaskRef{
-			Name: "orchestrate",
+	tr := &v1alpha1.TaskRun{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "mytaskrun",
+		},
+		Spec: v1alpha1.TaskRunSpec{
+			TaskRef: &v1alpha1.TaskRef{
+				Name: "orchestrate",
+			},
 		},
 	}
 	gt := func(n string) (v1alpha1.TaskInterface, error) { return nil, fmt.Errorf("something went wrong") }
-	_, _, err := GetTaskSpec(spec, "mytaskrun", gt)
+	_, _, err := GetTaskData(tr, gt)
 	if err == nil {
 		t.Fatalf("Expected error when unable to find referenced Task but got none")
 	}