Massive hack to prevent attempted mount of non-existent PVC 💀

In tektoncd#1007 @dlorenc and I tried to fix the case where a PVC wasn't needed
for output -> input linking and it was being created anyway. What we
forgot to do was check to see where that PVC was being mounted. It turns
out that if a TaskRun has an output and is created by a PipelineRun
(this is checked via the owner reference), the TaskRun assumes it needs
to mount the volume and further adds containers to copy the output data
to the (possibly) non-existent PVC. @castlemilk caught this problem in tektoncd#1068.

The real fix here is probably going to involve an interface change b/c
we can't assume that just being owned by a PipelineRun means that a
linking PVC is going to be involved. This commit is a terrible and
probably race condition hack to make it so that if the PVC the TaskRun
is expecting doesn't exist, it doesn't attempt to add containers that
will copy data to it.

Making the hack even worse is that instead of adding more actual unit
tests, I updated the test to run all the existing unit tests twice, once
with this PVC existing and once with it not, and I manipulated the test
so that in the case where it doesn't exist, the expected outcome is
different. This is a terrible way to write tests and I hope we either
don't merge this or we fix it quickly afterward. @dlorenc and I are
going to work on a better fix tomorrow.

I also modified our end to end PipelineRun test to include an output
resource so we could reproduce the issue that @castlemilk reported.

pkg/reconciler/v1alpha1/pipelinerun/pipelinerun_test.go

@@ -696,20 +696,38 @@ func TestReconcileWithTimeout(t *testing.T) {
 }
 
 func TestReconcileWithoutPVC(t *testing.T) {
+	rs := []*v1alpha1.PipelineResource{tb.PipelineResource("dance-party", "foo",
+		tb.PipelineResourceSpec(v1alpha1.PipelineResourceTypeGit),
+	)}
 	ps := []*v1alpha1.Pipeline{tb.Pipeline("test-pipeline", "foo", tb.PipelineSpec(
-		tb.PipelineTask("hello-world-1", "hello-world"),
-		tb.PipelineTask("hello-world-2", "hello-world"),
+		tb.PipelineDeclaredResource("party", v1alpha1.PipelineResourceTypeGit),
+		// Though these use the same resource as input and output, they are not linked with `from` so no PVC is needed
+		tb.PipelineTask("hello-world-1", "time-output",
+			tb.PipelineTaskOutputResource("time", "party"),
+		),
+		tb.PipelineTask("hello-world-2", "time-input",
+			tb.PipelineTaskInputResource("time", "party"),
+		),
 	))}
 
 	prs := []*v1alpha1.PipelineRun{tb.PipelineRun("test-pipeline-run", "foo",
-		tb.PipelineRunSpec("test-pipeline")),
+		tb.PipelineRunSpec("test-pipeline",
+			tb.PipelineRunResourceBinding("party", tb.PipelineResourceBindingRef("dance-party")),
+		)),
 	}
-	ts := []*v1alpha1.Task{tb.Task("hello-world", "foo")}
+	ts := []*v1alpha1.Task{
+		tb.Task("time-input", "foo", tb.TaskSpec(
+			tb.TaskInputs(tb.InputsResource("time", v1alpha1.PipelineResourceTypeGit)),
+		)),
+		tb.Task("time-output", "foo", tb.TaskSpec(
+			tb.TaskOutputs(tb.OutputsResource("time", v1alpha1.PipelineResourceTypeGit)),
+		))}
 
 	d := test.Data{
-		PipelineRuns: prs,
-		Pipelines:    ps,
-		Tasks:        ts,
+		PipelineResources: rs,
+		PipelineRuns:      prs,
+		Pipelines:         ps,
+		Tasks:             ts,
 	}
 
 	// create fake recorder for testing
@@ -730,7 +748,7 @@ func TestReconcileWithoutPVC(t *testing.T) {
 		t.Fatalf("Somehow had error getting reconciled run out of fake client: %s", err)
 	}
 
-	// Check that the expected TaskRun was created
+	// Check that no PVC was created
 	for _, a := range clients.Kube.Actions() {
 		if ca, ok := a.(ktesting.CreateAction); ok {
 			obj := ca.GetObject()

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

@@ -25,6 +25,8 @@ import (
 	"go.uber.org/zap"
 	"golang.org/x/xerrors"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
 )
 
@@ -43,8 +45,8 @@ var (
 // This function also reads the inputs to check if resources are redeclared in inputs and has any custom
 // target directory.
 // Steps executed:
-//  1. If taskrun has owner reference as pipelinerun then all outputs are copied to parents PVC
-// and also runs any custom upload steps (upload to blob store)
+//  1. If taskrun has owner reference as pipelinerun and the expected parent PVC exists, then all outputs
+// are copied to parents PVC and also runs any custom upload steps (upload to blob store)
 //  2.  If taskrun does not have pipelinerun as owner reference then all outputs resources execute their custom
 // upload steps (like upload to blob store )
 //
@@ -68,8 +70,22 @@ func AddOutputResources(
 
 	pvcName := taskRun.GetPipelineRunPVCName()
 	as, err := artifacts.GetArtifactStorage(pvcName, kubeclient, logger)
+
 	if err != nil {
-		return nil, err
+		return nil, xerrors.Errorf("couldn't determine artifact storage: %w", err)
+	}
+
+	if as.GetType() == v1alpha1.ArtifactStoragePVCType {
+		// TODO(#1068) There is probably a race condition here - this is just a hack to work around the bug in 0.5.0 while we
+		// work on a real long term fix
+		if _, err := kubeclient.CoreV1().PersistentVolumeClaims(taskRun.Namespace).Get(pvcName, metav1.GetOptions{}); err != nil {
+			// If there is no "parent" PVC, we shouldn't try to attach it and upload to it
+			if errors.IsNotFound(err) {
+				as = &artifacts.ArtifactStorageNone{}
+			} else {
+				return nil, xerrors.Errorf("error checking for possible parent PVC volume %s: %w", pvcName, err)
+			}
+		}
 	}
 
 	// track resources that are present in input of task cuz these resources will be copied onto PVC
@@ -91,6 +107,7 @@ func AddOutputResources(
 		if !ok || resource == nil {
 			return nil, xerrors.Errorf("failed to get output pipeline Resource for task %q resource %v", taskName, boundResource)
 		}
+
 		var (
 			resourceContainers []corev1.Container
 			resourceVolumes    []corev1.Volume
@@ -132,6 +149,7 @@ func AddOutputResources(
 		}
 
 		if allowedOutputResources[resource.GetType()] && taskRun.HasPipelineRunOwnerReference() {
+
 			var newSteps []corev1.Container
 			for _, dPath := range boundResource.Paths {
 				containers := as.GetCopyToStorageFromContainerSpec(resource.GetName(), sourcePath, dPath)