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.

docs/events.md

@@ -51,10 +51,15 @@ retrieving those events using the `kubectl describe` command. Tekton can also em
 When you [configure a sink](install.md#configuring-cloudevents-notifications), Tekton emits
 events as described in the table below.
 
-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`
+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`
+`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)