Add docs and validation for providedBy

While working on creating an end to end test for #168, @dlorenc and I
had a very hard time determining why our pipeline wasn't working as
expected - it turned out that we were using the wrong resource name in
our `providedBy` clauses.

While adding this validation I _really_ struggled to understand how the
`providedBy` clause actually works. I've updated the docs with my best
understanding of how it is currently working.

Fortunately in #320 we will be simplifying how this works!

This is follow-up work for #124

docs/using.md

@@ -37,6 +37,59 @@ need.
   expressed explicitly using this key since a task needing a resource from a
   another task would have to run after.
 - The name used in the `providedBy` is the name of `PipelineTask`
+- The name of the `PipelineResource` must correspond to a `PipelineResource` from
+  the `Task` that the referenced `PipelineTask` provides as an output
+
+For example see this `Pipeline` spec:
+
+```yaml
+  - name: build-skaffold-app
+    taskRef:
+      name: build-push
+    params:
+    - name: pathToDockerFile
+      value: Dockerfile
+    - name: pathToContext
+      value: /workspace/examples/microservices/leeroy-app
+  - name: deploy-app
+    taskRef:
+      name: demo-deploy-kubectl
+    resources:
+    - name: image
+      providedBy:
+      - build-skaffold-app
+```
+
+The `image` resource is expected to be provided to the `deploy-app` `Task` from the
+`build-skaffold-app` `Task`. This means that the `PipelineResource` bound to the `image`
+input for `deploy-app` must be bound to the same `PipelineResource` as an output from
+`build-skaffold-app`.
+
+This is the corresponding `PipelineRun` spec:
+
+```yaml
+  - name: build-skaffold-app
+    ...
+    outputs:
+    - name: builtImage
+      resourceRef:
+        name: skaffold-image-leeroy-app
+  - name: deploy-app
+    ...
+    - name: image
+      resourceRef:
+        name: skaffold-image-leeroy-app
+```
+
+You can see that the `builtImage` output from `build-skaffold-app` is bound to the
+`skaffold-image-leeroy-app` `PipelineResource`, and the same `PipelineResource` is bound
+to `image` for `deploy-app`.
+
+This controls two things:
+
+1. The order the `Tasks` are executed in: `deploy-app` must come after `build-skaffold-app`
+2. The state of the `PipelineResources`: the image provided to `deploy-app` may be changed
+   by `build-skaffold-app` (WIP, see [#216](https://github.com/knative/build-pipeline/issues/216))
 
 ## Creating a Task
 

pkg/apis/pipeline/v1alpha1/pipeline_types.go

@@ -90,7 +90,7 @@ type PipelineTaskParam struct {
 type ResourceDependency struct {
 	// Name is the name of the Task's input that this Resource should be used for.
 	Name string `json:"name"`
-	// ProvidedBy is the list of Task names that the resource has to come from.
+	// ProvidedBy is the list of PipelineTask names that the resource has to come from.
 	// +optional
 	ProvidedBy []string `json:"providedBy,omitempty"`
 }

pkg/reconciler/v1alpha1/pipelinerun/pipelinerun.go

@@ -197,6 +197,19 @@ func (c *Reconciler) reconcile(ctx context.Context, pr *v1alpha1.PipelineRun) er
 		}
 		return fmt.Errorf("error getting Tasks for Pipeline %s: %s", p.Name, err)
 	}
+
+	if err := resources.ValidateProvidedBy(pipelineState); err != nil {
+		// This Run has failed, so we need to mark it as failed and stop reconciling it
+		pr.Status.SetCondition(&duckv1alpha1.Condition{
+			Type:   duckv1alpha1.ConditionSucceeded,
+			Status: corev1.ConditionFalse,
+			Reason: ReasonFailedValidation,
+			Message: fmt.Sprintf("Pipeline %s can't be Run; it invalid input/output linkages: %s",
+				fmt.Sprintf("%s/%s", p.Namespace, pr.Name), err),
+		})
+		return nil
+	}
+
 	for _, rprt := range pipelineState {
 		err := taskrun.ValidateResolvedTaskResources(rprt.PipelineTask.Params, rprt.ResolvedTaskResources)
 		if err != nil {

pkg/reconciler/v1alpha1/pipelinerun/resources/pipelinerunresolution.go

@@ -234,3 +234,51 @@ func GetPipelineConditionStatus(prName string, state []*ResolvedPipelineRunTask,
 		Message: "All Tasks have completed executing",
 	}
 }
+
+func findReferencedTask(pb string, state []*ResolvedPipelineRunTask) *ResolvedPipelineRunTask {
+	for _, rprtRef := range state {
+		if rprtRef.PipelineTask.Name == pb {
+			return rprtRef
+		}
+	}
+	return nil
+}
+
+// ValidateProvidedBy will look at any `providedBy` clauses in the resolved PipelineRun state
+// and validate it: the `providedBy` must specify an input of the current `Task`. The `PipelineTask`
+// it is provided by must actually exist in the `Pipeline`. The `PipelineResource` that is bound to the input
+// must be the same `PipelineResource` that was bound to the output of the previous `Task`. If the state is
+// not valid, it will return an error.
+func ValidateProvidedBy(state []*ResolvedPipelineRunTask) error {
+	for _, rprt := range state {
+		for _, dep := range rprt.PipelineTask.ResourceDependencies {
+			inputBinding, ok := rprt.ResolvedTaskResources.Inputs[dep.Name]
+			if !ok {
+				return fmt.Errorf("PipelineTask %s is trying to declare dependency for Resource %s which is not an input of PipelineTask %q", rprt.PipelineTask.Name, dep.Name, rprt.PipelineTask.Name)
+			}
+
+			for _, pb := range dep.ProvidedBy {
+				if pb == rprt.PipelineTask.Name {
+					return fmt.Errorf("PipelineTask %s is trying to depend on a PipelineResource from itself", pb)
+				}
+				depTask := findReferencedTask(pb, state)
+				if depTask == nil {
+					return fmt.Errorf("pipelineTask %s is trying to depend on previous Task %q but it does not exist", rprt.PipelineTask.Name, pb)
+				}
+
+				sameBindingExists := false
+				for _, output := range depTask.ResolvedTaskResources.Outputs {
+					if output.Name == inputBinding.Name {
+						sameBindingExists = true
+					}
+				}
+				if !sameBindingExists {
+					return fmt.Errorf("providedBy is ambiguous: input %q for PipelineTask %q is bound to %q but no outputs in PipelineTask %q are bound to same resource",
+						dep.Name, rprt.PipelineTask.Name, inputBinding.Name, depTask.PipelineTask.Name)
+				}
+			}
+		}
+	}
+
+	return nil
+}