Add volumeClaimTemplate as a Workspace volume source

An existing PersistentVolumeClaim can currently be used as a Workspace
volume source. There is two ways of using an existing PVC as volume:

 - Reuse an existing PVC
 - Create a new PVC before each PipelineRun.

There is disadvantages by reusing the same PVC for every PipelineRun:

 - You need to clean the PVC at the end of the Pipeline
 - All Tasks using the workspace will be scheduled to the node where
   the PV is bound
 - Concurrent PipelineRuns may interfere, an artifact or file from one
   PipelineRun may slip in to or be used in another PipelineRun, with
   very few audit tracks.

There is also disadvantages by creating a new PVC before each PipelineRun:

 - This can not (easily) be done declaratively
 - This is hard to do programmatically, because it is hard to know when
   to delete the PVC. The PipelineRun can not be set as OwnerReference since
   the PVC must be created first

 This commit adds 'volumeClaimTemplate' as a volume source for workspaces. This
 has several advantages:

 - The syntax is used in k8s StatefulSet and other k8s projects so it is
   familiar in the kubernetes ecosystem
 - It is possible to declaratively declare that a PVC should be created for each
   PipelineRun, e.g. from a TriggerTemplate.
 - The user can choose storageClass (or omit to get the cluster default) to e.g.
   get a faster SSD volume, or to get a volume compatible with e.g. Windows.
 - The user can adapt the size to the job, e.g. use 5Gi for apps that contains
   machine learning models, or 1Gi for microservice apps. It can be changed on
   demand in a configuration that lives in the users namespace e.g. in a
   TriggerTemplate.
 - The size affects the storage quota that is set on the namespace and it may affect
   billing and cost depending on the cluster environment.
 - The PipelineRun or TaskRun with the template is created first, and is used
   as OwnerReference on the PVC. That means that the PVC will have the same lifecycle
   as the PipelineRun.

 Related to tektoncd#1986

 See also:
  - tektoncd#2174
  - tektoncd#2218
  - tektoncd/triggers#476
  - tektoncd/triggers#482
  - kubeflow/kfp-tekton#51

docs/workspaces.md

@@ -19,8 +19,8 @@
 
 `Workspaces` allow `Tasks` to declare parts of the filesystem that need to be provided
 at runtime by `TaskRuns`. A `TaskRun` can make these parts of the filesystem available
-in many ways: using a read-only `ConfigMap` or `Secret`, a `PersistentVolumeClaim`
-shared with other Tasks, or simply an `emptyDir` that is discarded when the `TaskRun`
+in many ways: using a read-only `ConfigMap` or `Secret`, an existing `PersistentVolumeClaim`
+shared with other Tasks, create a `PersistentVolumeClaim` from a provided `VolumeClaimTemplate`, or simply an `emptyDir` that is discarded when the `TaskRun`
 completes.
 
 `Workspaces` are similar to `Volumes` except that they allow a `Task` author 
@@ -294,9 +294,15 @@ However, they work well for single `TaskRuns` where the data stored in the `empt
 
 #### `persistentVolumeClaim`
 
-The `persistentVolumeClaim` field references a [`persistentVolumeClaim` volume](https://kubernetes.io/docs/concepts/storage/volumes/#persistentvolumeclaim).
+The `persistentVolumeClaim` field references an existing [`persistentVolumeClaim` volume](https://kubernetes.io/docs/concepts/storage/volumes/#persistentvolumeclaim).
 `PersistentVolumeClaim` volumes are a good choice for sharing data among `Tasks` within a `Pipeline`.
 
+#### `volumeClaimTemplate`
+
+The `volumeClaimTemplate` is a template of a [`persistentVolumeClaim` volume](https://kubernetes.io/docs/concepts/storage/volumes/#persistentvolumeclaim), created for each `PipelineRun` or `TaskRun`. 
+When the volume is created from a template in a `PipelineRun` or `TaskRun` it will be deleted when the `PipelineRun` or `TaskRun` is deleted.
+`volumeClaimTemplate` volumes are a good choice for sharing data among `Tasks` within a `Pipeline` when the volume is only used during a `PipelineRun` or `TaskRun`.
+
 #### `configMap`
 
 The `configMap` field references a [`configMap` volume](https://kubernetes.io/docs/concepts/storage/volumes/#configmap).

pkg/apis/pipeline/v1alpha1/pipelinerun_types.go

@@ -208,3 +208,14 @@ func (pr *PipelineRun) GetServiceAccountName(pipelineTaskName string) string {
 	}
 	return serviceAccountName
 }
+
+// HasVolumeClaimTemplate returns true if PipelineRun contains volumeClaimTemplates that is
+// used for creating PersistentVolumeClaims with an OwnerReference for each run
+func (pr *PipelineRun) HasVolumeClaimTemplate() bool {
+	for _, ws := range pr.Spec.Workspaces {
+		if ws.VolumeClaimTemplate != nil {
+			return true
+		}
+	}
+	return false
+}

pkg/apis/pipeline/v1alpha1/pipelinerun_types_test.go

@@ -130,6 +130,25 @@ func TestPipelineRunIsCancelled(t *testing.T) {
 	}
 }
 
+func TestPipelineRunHasVolumeClaimTemplate(t *testing.T) {
+	pr := &v1alpha1.PipelineRun{
+		Spec: v1alpha1.PipelineRunSpec{
+			Workspaces: []v1alpha1.WorkspaceBinding{{
+				Name: "my-workspace",
+				VolumeClaimTemplate: &corev1.PersistentVolumeClaim{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "pvc",
+					},
+					Spec: corev1.PersistentVolumeClaimSpec{},
+				},
+			}},
+		},
+	}
+	if !pr.HasVolumeClaimTemplate() {
+		t.Fatal("Expected pipelinerun to have a volumeClaimTemplate workspace")
+	}
+}
+
 func TestPipelineRunKey(t *testing.T) {
 	pr := tb.PipelineRun("prunname", "testns")
 	expectedKey := fmt.Sprintf("PipelineRun/%p", pr)

pkg/apis/pipeline/v1alpha1/taskrun_types.go

@@ -23,9 +23,18 @@ import (
 	"github.com/tektoncd/pipeline/pkg/apis/pipeline/v1beta1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	"knative.dev/pkg/apis"
 )
 
+var (
+	taskRunGroupVersionKind = schema.GroupVersionKind{
+		Group:   SchemeGroupVersion.Group,
+		Version: SchemeGroupVersion.Version,
+		Kind:    pipeline.TaskRunControllerName,
+	}
+)
+
 // TaskRunSpec defines the desired state of TaskRun
 type TaskRunSpec struct {
 	// +optional
@@ -165,6 +174,11 @@ func (tr *TaskRun) GetBuildPodRef() corev1.ObjectReference {
 	}
 }
 
+// GetOwnerReference gets the task run as owner reference for any related objects
+func (tr *TaskRun) GetOwnerReference() metav1.OwnerReference {
+	return *metav1.NewControllerRef(tr, taskRunGroupVersionKind)
+}
+
 // GetPipelineRunPVCName for taskrun gets pipelinerun
 func (tr *TaskRun) GetPipelineRunPVCName() string {
 	if tr == nil {
@@ -228,3 +242,14 @@ func (tr *TaskRun) IsPartOfPipeline() (bool, string, string) {
 
 	return false, "", ""
 }
+
+// HasVolumeClaimTemplate returns true if TaskRun contains volumeClaimTemplates that is
+// used for creating PersistentVolumeClaims with an OwnerReference for each run
+func (tr *TaskRun) HasVolumeClaimTemplate() bool {
+	for _, ws := range tr.Spec.Workspaces {
+		if ws.VolumeClaimTemplate != nil {
+			return true
+		}
+	}
+	return false
+}

pkg/apis/pipeline/v1alpha1/taskrun_types_test.go

@@ -112,6 +112,25 @@ func TestTaskRunIsCancelled(t *testing.T) {
 	}
 }
 
+func TestTaskRunHasVolumeClaimTemplate(t *testing.T) {
+	tr := &v1alpha1.TaskRun{
+		Spec: v1alpha1.TaskRunSpec{
+			Workspaces: []v1alpha1.WorkspaceBinding{{
+				Name: "my-workspace",
+				VolumeClaimTemplate: &corev1.PersistentVolumeClaim{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "pvc",
+					},
+					Spec: corev1.PersistentVolumeClaimSpec{},
+				},
+			}},
+		},
+	}
+	if !tr.HasVolumeClaimTemplate() {
+		t.Fatal("Expected taskrun to have a volumeClaimTemplate workspace")
+	}
+}
+
 func TestTaskRunKey(t *testing.T) {
 	tr := tb.TaskRun("taskrunname", "")
 	expectedKey := fmt.Sprintf("TaskRun/%p", tr)

pkg/apis/pipeline/v1beta1/workspace_types.go

@@ -55,6 +55,10 @@ type WorkspaceBinding struct {
 	// for this binding (i.e. the volume will be mounted at this sub directory).
 	// +optional
 	SubPath string `json:"subPath,omitempty"`
+	// VolumeClaimTemplate is a template for a claim that will be created in the same namespace.
+	// The PipelineRun controller is responsible for creating a unique claim for each instance of PipelineRun.
+	// +optional
+	VolumeClaimTemplate *corev1.PersistentVolumeClaim `json:"volumeClaimTemplate,omitempty"`
 	// PersistentVolumeClaimVolumeSource represents a reference to a
 	// PersistentVolumeClaim in the same namespace. Either this OR EmptyDir can be used.
 	// +optional

pkg/apis/pipeline/v1beta1/workspace_validation.go

@@ -27,6 +27,7 @@ import (
 // WorkspaceBinding may include.
 var allVolumeSourceFields []string = []string{
 	"workspace.persistentvolumeclaim",
+	"workspace.volumeclaimtemplate",
 	"workspace.emptydir",
 	"workspace.configmap",
 	"workspace.secret",
@@ -72,6 +73,9 @@ func (b *WorkspaceBinding) Validate(ctx context.Context) *apis.FieldError {
 // has been configured with.
 func (b *WorkspaceBinding) numSources() int {
 	n := 0
+	if b.VolumeClaimTemplate != nil {
+		n++
+	}
 	if b.PersistentVolumeClaim != nil {
 		n++
 	}

pkg/apis/pipeline/v1beta1/workspace_validation_test.go

@@ -21,6 +21,8 @@ import (
 	"testing"
 
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
 func TestWorkspaceBindingValidateValid(t *testing.T) {
@@ -35,6 +37,24 @@ func TestWorkspaceBindingValidateValid(t *testing.T) {
 				ClaimName: "pool-party",
 			},
 		},
+	}, {
+		name: "Valid volumeClaimTemplate",
+		binding: &WorkspaceBinding{
+			Name: "beth",
+			VolumeClaimTemplate: &corev1.PersistentVolumeClaim{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "mypvc",
+				},
+				Spec: corev1.PersistentVolumeClaimSpec{
+					AccessModes: []corev1.PersistentVolumeAccessMode{corev1.ReadWriteOnce},
+					Resources: corev1.ResourceRequirements{
+						Requests: corev1.ResourceList{
+							"storage": resource.MustParse("1Gi"),
+						},
+					},
+				},
+			},
+		},
 	}, {
 		name: "Valid emptyDir",
 		binding: &WorkspaceBinding{

pkg/reconciler/pipelinerun/controller.go

@@ -31,6 +31,7 @@ import (
 	resourceinformer "github.com/tektoncd/pipeline/pkg/client/resource/injection/informers/resource/v1alpha1/pipelineresource"
 	"github.com/tektoncd/pipeline/pkg/reconciler"
 	"github.com/tektoncd/pipeline/pkg/reconciler/pipelinerun/config"
+	"github.com/tektoncd/pipeline/pkg/reconciler/volumeclaim"
 	"k8s.io/client-go/tools/cache"
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
 	"knative.dev/pkg/configmap"
@@ -80,6 +81,7 @@ func NewController(images pipeline.Images) func(context.Context, configmap.Watch
 			conditionLister:   conditionInformer.Lister(),
 			timeoutHandler:    timeoutHandler,
 			metrics:           metrics,
+			pvcHandler:        volumeclaim.NewPVCHandler(kubeclientset, logger),
 		}
 		impl := controller.NewImpl(c, c.Logger, pipeline.PipelineRunControllerName)
 

pkg/reconciler/pipelinerun/pipelinerun.go

@@ -36,6 +36,7 @@ import (
 	"github.com/tektoncd/pipeline/pkg/reconciler/pipeline/dag"
 	"github.com/tektoncd/pipeline/pkg/reconciler/pipelinerun/resources"
 	"github.com/tektoncd/pipeline/pkg/reconciler/taskrun"
+	"github.com/tektoncd/pipeline/pkg/reconciler/volumeclaim"
 	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/equality"
@@ -105,6 +106,7 @@ type Reconciler struct {
 	configStore       configStore
 	timeoutHandler    *reconciler.TimeoutSet
 	metrics           *Recorder
+	pvcHandler        volumeclaim.PvcHandler
 }
 
 var (
@@ -436,6 +438,21 @@ func (c *Reconciler) reconcile(ctx context.Context, pr *v1alpha1.PipelineRun) er
 		return err
 	}
 
+	if pipelineState.IsBeforeFirstTaskRun() && pr.HasVolumeClaimTemplate() {
+		// create workspace PVC from template
+		if err = c.pvcHandler.CreatePersistentVolumeClaimsForWorkspaces(pr.Spec.Workspaces, pr.GetOwnerReference()[0], pr.Namespace); err != nil {
+			c.Logger.Errorf("Failed to create PVC for PipelineRun %s: %v", pr.Name, err)
+			pr.Status.SetCondition(&apis.Condition{
+				Type:   apis.ConditionSucceeded,
+				Status: corev1.ConditionFalse,
+				Reason: volumeclaim.ReasonCouldntCreateWorkspacePVC,
+				Message: fmt.Sprintf("Failed to create PVC for PipelineRun %s Workspaces correctly: %s",
+					fmt.Sprintf("%s/%s", pr.Namespace, pr.Name), err),
+			})
+			return nil
+		}
+	}
+
 	candidateTasks, err := dag.GetSchedulable(d, pipelineState.SuccessfulPipelineTaskNames()...)
 	if err != nil {
 		c.Logger.Errorf("Error getting potential next tasks for valid pipelinerun %s: %v", pr.Name, err)
@@ -603,9 +620,7 @@ func (c *Reconciler) createTaskRun(rprt *resources.ResolvedPipelineRunTask, pr *
 	for _, ws := range rprt.PipelineTask.Workspaces {
 		taskWorkspaceName, pipelineWorkspaceName := ws.Name, ws.Workspace
 		if b, hasBinding := pipelineRunWorkspaces[pipelineWorkspaceName]; hasBinding {
-			binding := *b.DeepCopy()
-			binding.Name = taskWorkspaceName
-			tr.Spec.Workspaces = append(tr.Spec.Workspaces, binding)
+			tr.Spec.Workspaces = append(tr.Spec.Workspaces, taskWorkspaceByWorkspaceVolumeSource(b, taskWorkspaceName, pr.GetOwnerReference()[0]))
 		} else {
 			return nil, fmt.Errorf("expected workspace %q to be provided by pipelinerun for pipeline task %q", pipelineWorkspaceName, rprt.PipelineTask.Name)
 		}
@@ -616,6 +631,26 @@ func (c *Reconciler) createTaskRun(rprt *resources.ResolvedPipelineRunTask, pr *
 	return c.PipelineClientSet.TektonV1alpha1().TaskRuns(pr.Namespace).Create(tr)
 }
 
+// taskWorkspaceByWorkspaceVolumeSource is returning the WorkspaceBinding with the TaskRun specified name.
+// If the volume source is a volumeClaimTemplate, the template is applied and passed to TaskRun as a persistentVolumeClaim
+func taskWorkspaceByWorkspaceVolumeSource(wb v1alpha1.WorkspaceBinding, taskWorkspaceName string, owner metav1.OwnerReference) v1alpha1.WorkspaceBinding {
+	if wb.VolumeClaimTemplate == nil {
+		binding := *wb.DeepCopy()
+		binding.Name = taskWorkspaceName
+		return binding
+	}
+
+	// apply template
+	binding := v1alpha1.WorkspaceBinding{
+		SubPath: wb.SubPath,
+		PersistentVolumeClaim: &corev1.PersistentVolumeClaimVolumeSource{
+			ClaimName: volumeclaim.GetPersistentVolumeClaimName(wb.VolumeClaimTemplate, wb, owner),
+		},
+	}
+	binding.Name = taskWorkspaceName
+	return binding
+}
+
 func addRetryHistory(tr *v1alpha1.TaskRun) {
 	newStatus := *tr.Status.DeepCopy()
 	newStatus.RetriesStatus = nil

pkg/reconciler/pipelinerun/pipelinerun_test.go

@@ -1778,3 +1778,84 @@ func TestReconcileWithTaskResults(t *testing.T) {
 		t.Errorf("expected to see TaskRun %v created. Diff %s", expectedTaskRunName, d)
 	}
 }
+
+// TestReconcileWithVolumeClaimTemplateWorkspace tests that given a pipeline with volumeClaimTemplate workspace,
+// a PVC is created and that the workspace appears as a PersistentVolumeClaim workspace for TaskRuns.
+func TestReconcileWithVolumeClaimTemplateWorkspace(t *testing.T) {
+	workspaceName := "ws1"
+	claimName := "myclaim"
+	pipelineRunName := "test-pipeline-run"
+	ps := []*v1alpha1.Pipeline{tb.Pipeline("test-pipeline", "foo", tb.PipelineSpec(
+		tb.PipelineTask("hello-world-1", "hello-world", tb.PipelineTaskWorkspaceBinding("taskWorkspaceName", workspaceName)),
+		tb.PipelineTask("hello-world-2", "hello-world"),
+		tb.PipelineWorkspaceDeclaration(workspaceName),
+	))}
+
+	prs := []*v1alpha1.PipelineRun{tb.PipelineRun(pipelineRunName, "foo",
+		tb.PipelineRunSpec("test-pipeline", tb.PipelineRunWorkspaceBindingVolumeClaimTemplate(workspaceName, claimName))),
+	}
+	ts := []*v1alpha1.Task{tb.Task("hello-world", "foo")}
+
+	d := test.Data{
+		PipelineRuns: prs,
+		Pipelines:    ps,
+		Tasks:        ts,
+	}
+
+	testAssets, cancel := getPipelineRunController(t, d)
+	defer cancel()
+	c := testAssets.Controller
+	clients := testAssets.Clients
+
+	err := c.Reconciler.Reconcile(context.Background(), "foo/test-pipeline-run")
+	if err != nil {
+		t.Errorf("Did not expect to see error when reconciling PipelineRun but saw %s", err)
+	}
+
+	// Check that the PipelineRun was reconciled correctly
+	reconciledRun, err := clients.Pipeline.TektonV1alpha1().PipelineRuns("foo").Get("test-pipeline-run", metav1.GetOptions{})
+	if err != nil {
+		t.Fatalf("Somehow had error getting reconciled run out of fake client: %s", err)
+	}
+
+	// Check that the expected PVC was created
+	pvcNames := make([]string, 0)
+	for _, a := range clients.Kube.Actions() {
+		if ca, ok := a.(ktesting.CreateAction); ok {
+			obj := ca.GetObject()
+			if pvc, ok := obj.(*corev1.PersistentVolumeClaim); ok {
+				pvcNames = append(pvcNames, pvc.Name)
+			}
+		}
+	}
+
+	if len(pvcNames) != 1 {
+		t.Errorf("expected one PVC created. %d was created", len(pvcNames))
+	}
+
+	expectedPVCName := fmt.Sprintf("%s-%s-%s", claimName, workspaceName, pipelineRunName)
+	if pvcNames[0] != expectedPVCName {
+		t.Errorf("expected the created PVC to be named %s. It was named %s", expectedPVCName, pvcNames[0])
+	}
+
+	taskRuns, err := clients.Pipeline.TektonV1alpha1().TaskRuns("foo").List(metav1.ListOptions{})
+	if err != nil {
+		t.Fatalf("unexpected error when listing TaskRuns: %v", err)
+	}
+
+	for _, tr := range taskRuns.Items {
+		for _, ws := range tr.Spec.Workspaces {
+			if ws.VolumeClaimTemplate != nil {
+				t.Fatalf("found volumeClaimTemplate workspace. Did not expect to find any taskruns with volumeClaimTemplate workspaces")
+			}
+
+			if ws.PersistentVolumeClaim == nil {
+				t.Fatalf("found taskRun workspace that is not PersistentVolumeClaim workspace. Did only expect PersistentVolumeClaims workspaces")
+			}
+		}
+	}
+
+	if !reconciledRun.Status.GetCondition(apis.ConditionSucceeded).IsUnknown() {
+		t.Errorf("Expected PipelineRun to be running, but condition status is %s", reconciledRun.Status.GetCondition(apis.ConditionSucceeded))
+	}
+}