Add pipeline run support for cloud events

Replace the pipeline run controller own config store with the
shared one used by the taskrun controller too. The pipeline run
config store is only useful to the artifact storage, however
the artifact storage loads the config by fetching the configmap
via the kube client, so it does not use the config store.

Attaching the shared config store to the controller, along with
the cloud events client, enables the pipeline run controller to
start sending cloud events for all events where we send k8s events
today (except for error ones).

Add a reconciler unit test to verify that events are sent when
the sink is configured.

Drop reconciler/pipelinerun/config because it's not used. It was
injected in the pipeline run controller before, but not used.
We can add the store for artifact configs back in a different
commit, but it wil have to be part of the shared store.

(cherry picked from commit e6c91d2)
Signed-off-by: Vincent Demeester <vincent@sbr.pm>

docs/events.md

@@ -45,22 +45,24 @@ No events are emitted for `Conditions` today (https://github.com/tektoncd/pipeli
   actually starts running.
 - `Succeeded`: this is triggered once all `Tasks` reachable via the DAG are
   executed successfully.
-- `Failed`: this is triggered if the `PipelineRun` is completed, but not
-  successfully. Causes of failure may be: one the `Tasks` failed or the
-  `PipelineRun` was cancelled or timed out. `Failed` events are also triggered
-  in case the `PipelineRun` cannot be executed at all because of validation issues.
+- `Failed`: emitted if the `PipelineRun` finishes running unsuccessfully because a `Task` failed or the
+  `PipelineRun` timed out or was cancelled. A `PipelineRun` also emits `Failed` events if it cannot
+  execute at all due to failing validation.
 
-# CloudEvents
+# Events via `CloudEvents`
 
-When a sink is [configured](./install.md#configuring-cloudevents-notifications), the following events
-will be generated by appropriate controller when a lifecycle event happens for  `TaskRun`.
+When you [configure a sink](install.md#configuring-cloudevents-notifications), Tekton emits
+events as described in the table below.
 
-The complete table of events:
-
-Reasource     |Event    |Event Type
+Resource      |Event    |Event Type
 :-------------|:-------:|:----------------------------------------------------------
-TaskRun       | Started | dev.tekton.event.taskrun.started.v1
-TaskRun       | Running | dev.tekton.event.taskrun.runnning.v1
-TaskRun       | Condition Change while Running | dev.tekton.event.taskrun.unknown.v1
-TaskRun       | Succeed | dev.tekton.event.taskrun.successful.v1
-TaskRun       | Failed  | dev.tekton.event.taskrun.failed.v1
+`TaskRun`     | `Started` | `dev.tekton.event.taskrun.started.v1`
+`TaskRun`     | `Running` | `dev.tekton.event.taskrun.runnning.v1`
+`TaskRun`     | `Condition Change while Running` | `dev.tekton.event.taskrun.unknown.v1`
+`TaskRun`     | `Succeed` | `dev.tekton.event.taskrun.successful.v1`
+`TaskRun`     | `Failed`  | `dev.tekton.event.taskrun.failed.v1`
+`PipelineRun` | `Started` | `dev.tekton.event.pipelinerun.started.v1`
+`PipelineRun` | `Running` | `dev.tekton.event.pipelinerun.runnning.v1`
+`PipelineRun` | `Condition Change while Running` | `dev.tekton.event.pipelinerun.unknown.v1`
+`PipelineRun` | `Succeed` | `dev.tekton.event.pipelinerun.successful.v1`
+`PipelineRun` | `Failed`  | `dev.tekton.event.pipelinerun.failed.v1`

internal/builder/v1beta1/pipeline.go

@@ -361,6 +361,13 @@ func PipelineRunNamespace(namespace string) PipelineRunOp {
 	}
 }
 
+// PipelineRunSelfLink adds a SelfLink
+func PipelineRunSelfLink(selflink string) PipelineRunOp {
+	return func(tr *v1beta1.PipelineRun) {
+		tr.ObjectMeta.SelfLink = selflink
+	}
+}
+
 // PipelineRunSpec sets the PipelineRunSpec, references Pipeline with specified name, to the PipelineRun.
 // Any number of PipelineRunSpec modifier can be passed to transform it.
 func PipelineRunSpec(name string, ops ...PipelineRunSpecOp) PipelineRunOp {

pkg/reconciler/pipelinerun/controller.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"time"
 
+	"github.com/tektoncd/pipeline/pkg/apis/config"
 	"github.com/tektoncd/pipeline/pkg/apis/pipeline"
 	pipelineclient "github.com/tektoncd/pipeline/pkg/client/injection/client"
 	conditioninformer "github.com/tektoncd/pipeline/pkg/client/injection/informers/pipeline/v1alpha1/condition"
@@ -31,7 +32,7 @@ import (
 	pipelinerunreconciler "github.com/tektoncd/pipeline/pkg/client/injection/reconciler/pipeline/v1beta1/pipelinerun"
 	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"
+	cloudeventclient "github.com/tektoncd/pipeline/pkg/reconciler/events/cloudevent"
 	"github.com/tektoncd/pipeline/pkg/reconciler/volumeclaim"
 	"k8s.io/client-go/tools/cache"
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
@@ -72,11 +73,12 @@ func NewController(namespace string, images pipeline.Images) func(context.Contex
 			resourceLister:    resourceInformer.Lister(),
 			conditionLister:   conditionInformer.Lister(),
 			timeoutHandler:    timeoutHandler,
+			cloudEventClient:  cloudeventclient.Get(ctx),
 			metrics:           metrics,
 			pvcHandler:        volumeclaim.NewPVCHandler(kubeclientset, logger),
 		}
 		impl := pipelinerunreconciler.NewImpl(ctx, c, func(impl *controller.Impl) controller.Options {
-			configStore := config.NewStore(images, logger.Named("config-store"))
+			configStore := config.NewStore(logger.Named("config-store"))
 			configStore.WatchConfigs(cmw)
 			return controller.Options{
 				AgentName:   pipeline.PipelineRunControllerName,

pkg/reconciler/pipelinerun/pipelinerun.go

@@ -40,6 +40,7 @@ import (
 	"github.com/tektoncd/pipeline/pkg/contexts"
 	"github.com/tektoncd/pipeline/pkg/reconciler"
 	"github.com/tektoncd/pipeline/pkg/reconciler/events"
+	"github.com/tektoncd/pipeline/pkg/reconciler/events/cloudevent"
 	"github.com/tektoncd/pipeline/pkg/reconciler/pipeline/dag"
 	"github.com/tektoncd/pipeline/pkg/reconciler/pipelinerun/resources"
 	"github.com/tektoncd/pipeline/pkg/reconciler/taskrun"
@@ -116,6 +117,7 @@ type Reconciler struct {
 	clusterTaskLister listers.ClusterTaskLister
 	resourceLister    resourcelisters.PipelineResourceLister
 	conditionLister   listersv1alpha1.ConditionLister
+	cloudEventClient  cloudevent.CEClient
 	tracker           tracker.Interface
 	timeoutHandler    *reconciler.TimeoutSet
 	metrics           *Recorder
@@ -132,6 +134,8 @@ var (
 // resource with the current status of the resource.
 func (c *Reconciler) ReconcileKind(ctx context.Context, pr *v1beta1.PipelineRun) pkgreconciler.Event {
 	logger := logging.FromContext(ctx)
+	ctx = cloudevent.ToContext(ctx, c.cloudEventClient)
+
 	// Read the initial condition
 	before := pr.Status.GetCondition(apis.ConditionSucceeded)