adding metadata to taskSpec in PipelineTask

docs/pipelineruns.md

@@ -119,6 +119,28 @@ spec:
 In the [`taskSpec` in `pipelineSpec` example](../examples/v1beta1/pipelineruns/pipelinerun-with-pipelinespec-and-taskspec.yaml)
 it's `Tasks` all the way down!
 
+You can also specify labels and annotations with `taskSpec` which are propagated to each `taskRun` and then to the
+respective pods. These labels can be used to identify and filter pods for further actions (such as collecting pod metrics,
+and cleaning up completed pod with certain labels, etc) even being part of one single Pipeline.
+
+```yaml
+spec:
+  pipelineSpec:
+    tasks:
+    - name: task1
+      metadata:
+        labels:
+          pipeline-sdk-type: kfp
+      taskSpec:
+       ...
+    - name: task2
+      metadata:
+        labels:
+          pipeline-sdk-type: tfx
+      taskSpec:
+       ...
+```
+
 ## Specifying `Resources`
 
 A `Pipeline` requires [`PipelineResources`](resources.md) to provide inputs and store outputs

examples/v1beta1/pipelineruns/pipelinerun-with-pipelinespec-and-taskspec.yaml

@@ -7,6 +7,9 @@ spec:
     tasks:
       - name: echo-good-morning
         taskSpec:
+          metadata:
+            labels:
+              app: "example"
           steps:
             - name: echo
               image: ubuntu

internal/builder/v1beta1/pipeline.go

@@ -198,7 +198,20 @@ func PipelineRunResult(name, value string) PipelineRunStatusOp {
 // PipelineTaskSpec sets the TaskSpec on a PipelineTask.
 func PipelineTaskSpec(spec *v1beta1.TaskSpec) PipelineTaskOp {
 	return func(pt *v1beta1.PipelineTask) {
-		pt.TaskSpec = spec
+		if pt.TaskSpec == nil {
+			pt.TaskSpec = &v1beta1.EmbeddedTask{}
+		}
+		pt.TaskSpec.TaskSpec = spec
+	}
+}
+
+// TaskSpecMetadata sets the Metadata on a TaskSpec within PipelineTask.
+func TaskSpecMetadata(metadata v1beta1.PipelineTaskMetadata) PipelineTaskOp {
+	return func(pt *v1beta1.PipelineTask) {
+		if pt.TaskSpec == nil {
+			pt.TaskSpec = &v1beta1.EmbeddedTask{}
+		}
+		pt.TaskSpec.Metadata = metadata
 	}
 }
 

internal/builder/v1beta1/pipeline_test.go

@@ -57,12 +57,14 @@ func TestPipeline(t *testing.T) {
 			tb.RunAfter("foo"),
 			tb.PipelineTaskTimeout(5*time.Second),
 		),
-		tb.PipelineTask("foo", "", tb.PipelineTaskSpec(&v1beta1.TaskSpec{
-			Steps: []v1beta1.Step{{Container: corev1.Container{
-				Name:  "step",
-				Image: "myimage",
-			}}},
-		})),
+		tb.PipelineTask("foo", "", tb.PipelineTaskSpec(getTaskSpec())),
+		tb.PipelineTask("task-with-taskSpec", "",
+			tb.TaskSpecMetadata(v1beta1.PipelineTaskMetadata{
+				Labels:      map[string]string{"label": "labelvalue"},
+				Annotations: map[string]string{"annotation": "annotationvalue"}},
+			),
+			tb.PipelineTaskSpec(getTaskSpec()),
+		),
 		tb.PipelineWorkspaceDeclaration("workspace1"),
 	),
 		tb.PipelineCreationTimestamp(creationTime),
@@ -137,13 +139,19 @@ func TestPipeline(t *testing.T) {
 				Timeout:  &metav1.Duration{Duration: 5 * time.Second},
 			}, {
 				Name: "foo",
-				TaskSpec: &v1beta1.TaskSpec{
-					Steps: []v1beta1.Step{{Container: corev1.Container{
-						Name:  "step",
-						Image: "myimage",
-					}}},
+				TaskSpec: &v1beta1.EmbeddedTask{
+					TaskSpec: getTaskSpec()},
+			}, {
+				Name: "task-with-taskSpec",
+				TaskSpec: &v1beta1.EmbeddedTask{
+					Metadata: v1beta1.PipelineTaskMetadata{
+						Labels:      map[string]string{"label": "labelvalue"},
+						Annotations: map[string]string{"annotation": "annotationvalue"},
+					},
+					TaskSpec: getTaskSpec(),
 				},
-			}},
+			},
+			},
 			Workspaces: []v1beta1.PipelineWorkspaceDeclaration{{
 				Name: "workspace1",
 			}},
@@ -426,5 +434,13 @@ func TestPipelineRunWithFinalTask(t *testing.T) {
 	if diff := cmp.Diff(expectedPipelineRun, pipelineRun); diff != "" {
 		t.Fatalf("PipelineRun diff -want, +got: %s", diff)
 	}
+}
 
+func getTaskSpec() *v1beta1.TaskSpec {
+	return &v1beta1.TaskSpec{
+		Steps: []v1beta1.Step{{Container: corev1.Container{
+			Name:  "step",
+			Image: "myimage",
+		}}},
+	}
 }

pkg/apis/pipeline/v1alpha1/pipeline_conversion.go

@@ -61,8 +61,8 @@ func (source *PipelineTask) ConvertTo(ctx context.Context, sink *v1beta1.Pipelin
 	sink.Name = source.Name
 	sink.TaskRef = source.TaskRef
 	if source.TaskSpec != nil {
-		sink.TaskSpec = &v1beta1.TaskSpec{}
-		if err := source.TaskSpec.ConvertTo(ctx, sink.TaskSpec); err != nil {
+		sink.TaskSpec = &v1beta1.EmbeddedTask{TaskSpec: &v1beta1.TaskSpec{}}
+		if err := source.TaskSpec.ConvertTo(ctx, sink.TaskSpec.TaskSpec); err != nil {
 			return err
 		}
 	}
@@ -112,7 +112,7 @@ func (sink *PipelineTask) ConvertFrom(ctx context.Context, source v1beta1.Pipeli
 	sink.TaskRef = source.TaskRef
 	if source.TaskSpec != nil {
 		sink.TaskSpec = &TaskSpec{}
-		if err := sink.TaskSpec.ConvertFrom(ctx, source.TaskSpec); err != nil {
+		if err := sink.TaskSpec.ConvertFrom(ctx, source.TaskSpec.TaskSpec); err != nil {
 			return err
 		}
 	}

pkg/apis/pipeline/v1beta1/pipeline_defaults_test.go

@@ -85,41 +85,49 @@ func TestPipelineSpec_SetDefaults(t *testing.T) {
 		desc: "pipeline task with taskSpec - default param type must be " + string(v1beta1.ParamTypeString),
 		ps: &v1beta1.PipelineSpec{
 			Tasks: []v1beta1.PipelineTask{{
-				Name: "foo", TaskSpec: &v1beta1.TaskSpec{
-					Params: []v1beta1.ParamSpec{{
-						Name: "string-param",
-					}},
+				Name: "foo", TaskSpec: &v1beta1.EmbeddedTask{
+					TaskSpec: &v1beta1.TaskSpec{
+						Params: []v1beta1.ParamSpec{{
+							Name: "string-param",
+						}},
+					},
 				},
 			}},
 		},
 		want: &v1beta1.PipelineSpec{
 			Tasks: []v1beta1.PipelineTask{{
-				Name: "foo", TaskSpec: &v1beta1.TaskSpec{
-					Params: []v1beta1.ParamSpec{{
-						Name: "string-param",
-						Type: v1beta1.ParamTypeString,
-					}},
+				Name: "foo", TaskSpec: &v1beta1.EmbeddedTask{
+					TaskSpec: &v1beta1.TaskSpec{
+						Params: []v1beta1.ParamSpec{{
+							Name: "string-param",
+							Type: v1beta1.ParamTypeString,
+						}},
+					},
 				},
 			}},
 		},
 	}, {
 		desc: "final pipeline task with taskSpec - default param type must be " + string(v1beta1.ParamTypeString),
 		ps: &v1beta1.PipelineSpec{
 			Finally: []v1beta1.PipelineTask{{
-				Name: "foo", TaskSpec: &v1beta1.TaskSpec{
-					Params: []v1beta1.ParamSpec{{
-						Name: "string-param",
-					}},
+				Name: "foo", TaskSpec: &v1beta1.EmbeddedTask{
+					TaskSpec: &v1beta1.TaskSpec{
+						Params: []v1beta1.ParamSpec{{
+							Name: "string-param",
+						}},
+					},
 				},
 			}},
 		},
 		want: &v1beta1.PipelineSpec{
 			Finally: []v1beta1.PipelineTask{{
-				Name: "foo", TaskSpec: &v1beta1.TaskSpec{
-					Params: []v1beta1.ParamSpec{{
-						Name: "string-param",
-						Type: v1beta1.ParamTypeString,
-					}},
+				Name: "foo", TaskSpec: &v1beta1.EmbeddedTask{
+					TaskSpec: &v1beta1.TaskSpec{
+						Params: []v1beta1.ParamSpec{{
+							Name: "string-param",
+							Type: v1beta1.ParamTypeString,
+						}},
+					},
 				},
 			}},
 		},

pkg/apis/pipeline/v1beta1/pipeline_types.go

@@ -90,6 +90,23 @@ type PipelineResult struct {
 	Value string `json:"value"`
 }
 
+type PipelineTaskMetadata struct {
+	// +optional
+	Labels map[string]string `json:"labels,omitempty"`
+
+	// +optional
+	Annotations map[string]string `json:"annotations,omitempty"`
+}
+
+type EmbeddedTask struct {
+	// +optional
+	Metadata PipelineTaskMetadata `json:"metadata,omitempty"`
+
+	// TaskSpec is a specification of a task
+	// +optional
+	*TaskSpec `json:",inline,omitempty"`
+}
+
 // PipelineTask defines a task in a Pipeline, passing inputs from both
 // Params and from the output of previous tasks.
 type PipelineTask struct {
@@ -104,7 +121,7 @@ type PipelineTask struct {
 
 	// TaskSpec is a specification of a task
 	// +optional
-	TaskSpec *TaskSpec `json:"taskSpec,omitempty"`
+	TaskSpec *EmbeddedTask `json:"taskSpec,inline,omitempty"`
 
 	// Conditions is a list of conditions that need to be true for the task to run
 	// +optional
@@ -139,6 +156,10 @@ type PipelineTask struct {
 	Timeout *metav1.Duration `json:"timeout,omitempty"`
 }
 
+func (pt *PipelineTask) TaskSpecMetadata() PipelineTaskMetadata {
+	return pt.TaskSpec.Metadata
+}
+
 func (pt PipelineTask) HashKey() string {
 	return pt.Name
 }

pkg/apis/pipeline/v1beta1/pipeline_validation_test.go

@@ -152,13 +152,9 @@ func TestPipelineSpec_Validate_Failure(t *testing.T) {
 				Name:    "valid-pipeline-task",
 				TaskRef: &TaskRef{Name: "foo-task"},
 			}, {
-				Name:    "invalid-pipeline-task",
-				TaskRef: &TaskRef{Name: "foo-task"},
-				TaskSpec: &TaskSpec{
-					Steps: []Step{{
-						Container: corev1.Container{Name: "foo", Image: "bar"},
-					}},
-				},
+				Name:     "invalid-pipeline-task",
+				TaskRef:  &TaskRef{Name: "foo-task"},
+				TaskSpec: &EmbeddedTask{TaskSpec: getTaskSpec()},
 			}},
 		},
 	}, {
@@ -250,12 +246,8 @@ func TestValidatePipelineTasks_Success(t *testing.T) {
 	}, {
 		name: "pipeline task with valid taskspec",
 		tasks: []PipelineTask{{
-			Name: "foo",
-			TaskSpec: &TaskSpec{
-				Steps: []Step{{
-					Container: corev1.Container{Name: "foo", Image: "bar"},
-				}},
-			},
+			Name:     "foo",
+			TaskSpec: &EmbeddedTask{TaskSpec: getTaskSpec()},
 		}},
 	}}
 	for _, tt := range tests {
@@ -280,19 +272,15 @@ func TestValidatePipelineTasks_Failure(t *testing.T) {
 	}, {
 		name: "pipeline task with both taskref and taskspec",
 		tasks: []PipelineTask{{
-			Name:    "foo",
-			TaskRef: &TaskRef{Name: "foo-task"},
-			TaskSpec: &TaskSpec{
-				Steps: []Step{{
-					Container: corev1.Container{Name: "foo", Image: "bar"},
-				}},
-			},
+			Name:     "foo",
+			TaskRef:  &TaskRef{Name: "foo-task"},
+			TaskSpec: &EmbeddedTask{TaskSpec: getTaskSpec()},
 		}},
 	}, {
 		name: "pipeline task with invalid taskspec",
 		tasks: []PipelineTask{{
 			Name:     "foo",
-			TaskSpec: &TaskSpec{},
+			TaskSpec: &EmbeddedTask{TaskSpec: &TaskSpec{}},
 		}},
 	}, {
 		name: "pipeline tasks invalid (duplicate tasks)",
@@ -660,14 +648,14 @@ func TestValidateGraph_Failure(t *testing.T) {
 func TestValidateParamResults_Success(t *testing.T) {
 	desc := "valid pipeline task referencing task result along with parameter variable"
 	tasks := []PipelineTask{{
-		TaskSpec: &TaskSpec{
+		TaskSpec: &EmbeddedTask{TaskSpec: &TaskSpec{
 			Results: []TaskResult{{
 				Name: "output",
 			}},
 			Steps: []Step{{
 				Container: corev1.Container{Name: "foo", Image: "bar"},
 			}},
-		},
+		}},
 		Name: "a-task",
 	}, {
 		Name:    "foo",
@@ -1041,12 +1029,8 @@ func TestValidatePipelineWithFinalTasks_Success(t *testing.T) {
 					Name:    "final-task-1",
 					TaskRef: &TaskRef{Name: "final-task"},
 				}, {
-					Name: "final-task-2",
-					TaskSpec: &TaskSpec{
-						Steps: []Step{{
-							Container: corev1.Container{Name: "foo", Image: "bar"},
-						}},
-					},
+					Name:     "final-task-2",
+					TaskSpec: &EmbeddedTask{TaskSpec: getTaskSpec()},
 				}},
 			},
 		},
@@ -1200,13 +1184,9 @@ func TestValidatePipelineWithFinalTasks_Failure(t *testing.T) {
 					TaskRef: &TaskRef{Name: "non-final-task"},
 				}},
 				Finally: []PipelineTask{{
-					Name:    "final-task",
-					TaskRef: &TaskRef{Name: "non-final-task"},
-					TaskSpec: &TaskSpec{
-						Steps: []Step{{
-							Container: corev1.Container{Name: "foo", Image: "bar"},
-						}},
-					},
+					Name:     "final-task",
+					TaskRef:  &TaskRef{Name: "non-final-task"},
+					TaskSpec: &EmbeddedTask{TaskSpec: getTaskSpec()},
 				}},
 			},
 		},
@@ -1491,3 +1471,11 @@ func TestContextInvalid(t *testing.T) {
 		})
 	}
 }
+
+func getTaskSpec() *TaskSpec {
+	return &TaskSpec{
+		Steps: []Step{{
+			Container: corev1.Container{Name: "foo", Image: "bar"},
+		}},
+	}
+}