feat(cloudevents-server): support notify by event subject (#102)

Signed-off-by: wuhuizuo <wuhuizuo@126.com>

Signed-off-by: wuhuizuo <wuhuizuo@126.com>

cloudevents-server/configs/example-config-sqlite3.yaml

@@ -9,9 +9,15 @@ tibuild:
   trigger_sink_url: http://el-xxx.tekton.url:8080 # url of tekton event listener.
 tekton:
   dashboard_base_url: https://tekton.local.com/dashboard-path
-  receivers:
-    "*": [all-level-receiver]
-    dev.tekton.event.pipelinerun.started.v1: [started-receiver]
-    dev.tekton.event.pipelinerun.running.v1: [running-receiver]
-    dev.tekton.event.pipelinerun.successful.v1: [successful-receiver]
-    dev.tekton.event.pipelinerun.failed.v1: [failure-receiver]
+  notifications:
+    - event_type: ".*"
+      receivers: [all-level-receiver]
+    - event_type: dev.tekton.event.pipelinerun.started.v1
+      receivers: [started-receiver]
+    - event_type: dev.tekton.event.pipelinerun.running.v1
+      receivers: [running-receiver]
+    - event_type: dev.tekton.event.pipelinerun.successful.v1
+      receivers: [successful-receiver]
+    - event_type: dev.tekton.event.pipelinerun.failed.v1
+      event_subject_reg: ^xxx-from-.*
+      receivers: [failure-receiver]

cloudevents-server/configs/example-config.yaml

@@ -9,9 +9,15 @@ tibuild:
   trigger_sink_url: http://el-xxx.tekton.url:8080 # url of tekton event listener.
 tekton:
   dashboard_base_url: https://tekton.local.com/dashboard-path
-  receivers:
-    "*": [all-level-receiver]
-    dev.tekton.event.pipelinerun.started.v1: [started-receiver]
-    dev.tekton.event.pipelinerun.running.v1: [running-receiver]
-    dev.tekton.event.pipelinerun.successful.v1: [successful-receiver]
-    dev.tekton.event.pipelinerun.failed.v1: [failure-receiver]
+  notifications:
+    - event_type: ".*"
+      receivers: [all-level-receiver]
+    - event_type: dev.tekton.event.pipelinerun.started.v1
+      receivers: [started-receiver]
+    - event_type: dev.tekton.event.pipelinerun.running.v1
+      receivers: [running-receiver]
+    - event_type: dev.tekton.event.pipelinerun.successful.v1
+      receivers: [successful-receiver]
+    - event_type: dev.tekton.event.pipelinerun.failed.v1
+      event_subject_reg: ^xxx-from-.*
+      receivers: [failure-receiver]

cloudevents-server/pkg/config/config.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
+	"regexp"
 
 	"gopkg.in/yaml.v3"
 )
@@ -19,10 +20,33 @@ type LarkBotApp struct {
 	AppSecret string `yaml:"app_secret,omitempty" json:"app_secret,omitempty"`
 }
 
+// TektonNotification notify config for tekton run events, the `Type` and `Subject` is "AND" logic.
+type TektonNotification struct {
+	// EventType of the cloud event, if you want it send all types, set value as "*".
+	EventType string `yaml:"event_type,omitempty" json:"event_type,omitempty"`
+	// EventSubjectReg match regex of the cloud event, if you want it send for all
+	//	subjects, set the value as "" or ".*".
+	EventSubjectReg string `yaml:"event_subject_reg,omitempty" json:"event_subject_reg,omitempty"`
+	// Receivers that will send to.
+	Receivers []string `yaml:"receivers,omitempty" json:"receivers,omitempty"`
+}
+
+func (c *TektonNotification) IsMatched(typ, subject string) bool {
+	if (c.EventType == "" || c.EventType == "*") || c.EventType == typ {
+		if c.EventSubjectReg == "" {
+			return true
+		}
+
+		ok, _ := regexp.MatchString(c.EventSubjectReg, subject)
+		return ok
+	}
+
+	return false
+}
+
 type Tekton struct {
-	DashboardBaseURL string `yaml:"dashboard_base_url,omitempty" json:"dashboard_base_url,omitempty"`
-	// Receivers receivers list of the event type, if you want it send all types, set the key "*".
-	Receivers map[string][]string `yaml:"receivers,omitempty" json:"receiver,omitempty"`
+	DashboardBaseURL string               `yaml:"dashboard_base_url,omitempty" json:"dashboard_base_url,omitempty"`
+	Notifications    []TektonNotification `yaml:"notifications,omitempty" json:"notifications,omitempty"`
 }
 
 type Config struct {

cloudevents-server/pkg/events/custom/tekton/handler.go

@@ -12,6 +12,7 @@ import (
 	"github.com/tektoncd/pipeline/pkg/apis/pipeline/v1beta1"
 	tektoncloudevent "github.com/tektoncd/pipeline/pkg/reconciler/events/cloudevent"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/sets"
 	"knative.dev/pkg/apis"
 
 	"github.com/PingCAP-QE/ee-apps/cloudevents-server/pkg/config"
@@ -46,6 +47,20 @@ func getTriggerUser(run AnnotationsGetter) string {
 	return contextData[eventContextAnnotationInnerKeyUser]
 }
 
+func getReceivers(event cloudevents.Event, cfgs []config.TektonNotification) []string {
+	eventType := event.Type()
+	eventSub := event.Subject()
+
+	ret := sets.NewString()
+	for _, cfg := range cfgs {
+		if cfg.IsMatched(eventType, eventSub) {
+			ret.Insert(cfg.Receivers...)
+		}
+	}
+
+	return ret.List()
+}
+
 // <dashboard base url>/#/namespaces/<namespace>/<run-type>s/<run-name>
 // source: /apis///namespaces/<namespace>//<run-name>
 // https://tekton.abc.com/tekton/apis/tekton.dev/v1beta1/namespaces/ee-cd/pipelineruns/auto-compose-multi-arch-image-run-g5hqv

cloudevents-server/pkg/events/custom/tekton/handler_pipelinerun.go

@@ -49,7 +49,7 @@ func (h *pipelineRunHandler) Handle(event cloudevents.Event) cloudevents.Result
 		if receiver := getTriggerUser(data.PipelineRun); receiver != "" {
 			receivers = []string{receiver}
 		} else {
-			receivers = append(h.Receivers[defaultReceiversKey], h.Receivers[event.Type()]...)
+			receivers = getReceivers(event, h.Notifications)
 		}
 
 		log.Debug().

cloudevents-server/pkg/events/custom/tekton/handler_pipelinerun_test.go

@@ -53,7 +53,7 @@ func Test_pipelineRunHandler_Handle(t *testing.T) {
 	h := &pipelineRunHandler{
 		LarkClient: lark.NewClient(larkAppID, larkAppSecret, lark.WithLogReqAtDebug(true), lark.WithEnableTokenCache(true)),
 		Tekton: config.Tekton{
-			Receivers:        map[string][]string{"*": {receiver}},
+			Notifications:    []config.TektonNotification{{Receivers: []string{receiver}}},
 			DashboardBaseURL: baseURL,
 		},
 	}

cloudevents-server/pkg/events/custom/tekton/handler_taskrun.go

@@ -45,7 +45,7 @@ func (h *taskRunHandler) Handle(event cloudevents.Event) cloudevents.Result {
 		if receiver := getTriggerUser(data.TaskRun); receiver != "" {
 			receivers = []string{receiver}
 		} else {
-			receivers = append(h.Receivers[defaultReceiversKey], h.Receivers[event.Type()]...)
+			receivers = getReceivers(event, h.Notifications)
 		}
 
 		log.Debug().

cloudevents-server/pkg/events/custom/tekton/handler_taskrun_test.go

@@ -25,6 +25,8 @@ var (
 	taskRunSuccessfulEventBytes []byte
 	//go:embed testdata/event-taskrun.unknown.json
 	taskRunUnknownEventBytes []byte
+	//go:embed testdata/event-taskrun.failed-standalone.json
+	standaloneTaskRunFailedEventBytes []byte
 )
 
 func Test_taskRunHandler_Handle(t *testing.T) {
@@ -43,12 +45,16 @@ func Test_taskRunHandler_Handle(t *testing.T) {
 		{name: tektoncloudevent.TaskRunStartedEventV1, eventJSON: taskRunStartedEventBytes, want: cloudevents.ResultACK},
 		{name: tektoncloudevent.TaskRunSuccessfulEventV1, eventJSON: taskRunSuccessfulEventBytes, want: cloudevents.ResultACK},
 		{name: tektoncloudevent.TaskRunUnknownEventV1, eventJSON: taskRunUnknownEventBytes, want: cloudevents.ResultACK},
+		{name: "standalone-taskrun-failed", eventJSON: standaloneTaskRunFailedEventBytes, want: cloudevents.ResultACK},
 	}
 
 	h := &taskRunHandler{
 		LarkClient: lark.NewClient(larkAppID, larkAppSecret, lark.WithLogReqAtDebug(true), lark.WithEnableTokenCache(true)),
 		Tekton: config.Tekton{
-			Receivers:        map[string][]string{"*": {receiver}},
+			Notifications: []config.TektonNotification{
+				{Receivers: []string{receiver}},
+				{Receivers: []string{receiver}, EventSubjectReg: "^fake-task-run-standalone-"},
+			},
 			DashboardBaseURL: baseURL,
 		},
 	}