Add pipeline run support for cloud events

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"
@@ -30,7 +31,7 @@ import (
 	taskruninformer "github.com/tektoncd/pipeline/pkg/client/injection/informers/pipeline/v1beta1/taskrun"
 	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/pipelinerun/config"
+	cloudeventclient "github.com/tektoncd/pipeline/pkg/reconciler/events/cloudevent"
 	"github.com/tektoncd/pipeline/pkg/reconciler/volumeclaim"
 	"github.com/tektoncd/pipeline/pkg/timeout"
 	"k8s.io/client-go/tools/cache"
@@ -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

@@ -39,6 +39,7 @@ import (
 	resourcelisters "github.com/tektoncd/pipeline/pkg/client/resource/listers/resource/v1alpha1"
 	"github.com/tektoncd/pipeline/pkg/contexts"
 	"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"
@@ -113,6 +114,7 @@ type Reconciler struct {
 	clusterTaskLister listers.ClusterTaskLister
 	resourceLister    resourcelisters.PipelineResourceLister
 	conditionLister   listersv1alpha1.ConditionLister
+	cloudEventClient  cloudevent.CEClient
 	tracker           tracker.Interface
 	timeoutHandler    *timeout.Handler
 	metrics           *Recorder
@@ -129,6 +131,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)