feat(cloudevents-server): support set receivers by event types

Signed-off-by: wuhuizuo <[email protected]>

cloudevents-server/pkg/config/config.go

@@ -14,16 +14,21 @@ type Store struct {
 	DSN    string `yaml:"dsn,omitempty" json:"dsn,omitempty"`
 }
 
-type Lark struct {
+type LarkBotApp struct {
 	AppID     string `yaml:"app_id,omitempty" json:"app_id,omitempty"`
 	AppSecret string `yaml:"app_secret,omitempty" json:"app_secret,omitempty"`
-	// TODO: how to get the receiver?
-	Receivers []string `yaml:"receivers,omitempty" json:"receiver,omitempty"`
+}
+
+type Tekton struct {
+	DashboardBaseURL string `yaml:"web_base_url,omitempty" json:"web_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"`
 }
 
 type Config struct {
-	Store   Store `yaml:"store,omitempty" json:"store,omitempty"`
-	Lark    Lark  `yaml:"lark,omitempty" json:"lark,omitempty"`
+	Store   Store      `yaml:"store,omitempty" json:"store,omitempty"`
+	Lark    LarkBotApp `yaml:"lark,omitempty" json:"lark,omitempty"`
+	Tekton  Tekton     `yaml:"tekton,omitempty" json:"tekton,omitempty"`
 	TiBuild struct {
 		ResultSinkURL  string `yaml:"result_sink_url,omitempty" json:"result_sink_url,omitempty"`
 		TriggerSinkURL string `yaml:"trigger_sink_url,omitempty" json:"trigger_sink_url,omitempty"`

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

@@ -1,17 +1,38 @@
 package tekton
 
 import (
+	"encoding/json"
+
 	"github.com/PingCAP-QE/ee-apps/cloudevents-server/pkg/config"
 	"github.com/PingCAP-QE/ee-apps/cloudevents-server/pkg/events/handler"
+	"github.com/PingCAP-QE/ee-apps/cloudevents-server/pkg/lark"
+)
+
+const (
+	eventContextAnnotationKey          = "tekton.dev/ce-context"
+	eventContextAnnotationInnerKeyUser = "user"
 )
 
-func NewHandler(cfg config.Lark) (handler.EventHandler, error) {
-	larkClient := newLarkClient(cfg)
-	baseURL := "https://do.pingcap.net/tekton"
+func NewHandler(cfg config.Config) (handler.EventHandler, error) {
+	larkClient := lark.NewClient(cfg.Lark.AppID, cfg.Lark.AppSecret)
 	ret := new(handler.CompositeEventHandler).AddHandlers(
-		&pipelineRunHandler{LarkClient: larkClient, Receivers: cfg.Receivers, RunDetailBaseURL: baseURL},
-		&taskRunHandler{LarkClient: larkClient, Receivers: cfg.Receivers, RunDetailBaseURL: baseURL},
+		&pipelineRunHandler{LarkClient: larkClient, Tekton: cfg.Tekton},
+		&taskRunHandler{LarkClient: larkClient, Tekton: cfg.Tekton},
 	)
 
 	return ret, nil
 }
+
+func getTriggerUser(run AnnotationsGetter) string {
+	eventContext := run.GetAnnotations()[eventContextAnnotationKey]
+	if eventContext == "" {
+		return ""
+	}
+
+	contextData := make(map[string]string)
+	if err := json.Unmarshal([]byte(eventContext), &contextData); err != nil {
+		return ""
+	}
+
+	return contextData[eventContextAnnotationInnerKeyUser]
+}

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

@@ -1,25 +1,26 @@
 package tekton
 
 import (
-	"encoding/json"
 	"net/http"
 
+	"github.com/PingCAP-QE/ee-apps/cloudevents-server/pkg/config"
 	cloudevents "github.com/cloudevents/sdk-go/v2"
 	lark "github.com/larksuite/oapi-sdk-go/v3"
 	"github.com/rs/zerolog/log"
-	"github.com/tektoncd/pipeline/pkg/apis/pipeline/v1beta1"
 	tektoncloudevent "github.com/tektoncd/pipeline/pkg/reconciler/events/cloudevent"
 )
 
 const (
-	eventContextAnnotationKey          = "tekton.dev/ce-context"
-	eventContextAnnotationInnerKeyUser = "user"
+	defaultReceiversKey = "*"
 )
 
 type pipelineRunHandler struct {
-	LarkClient       *lark.Client
-	RunDetailBaseURL string
-	Receivers        []string
+	config.Tekton
+	LarkClient *lark.Client
+}
+
+type AnnotationsGetter interface {
+	GetAnnotations() map[string]string
 }
 
 func (h *pipelineRunHandler) SupportEventTypes() []string {
@@ -42,28 +43,17 @@ func (h *pipelineRunHandler) Handle(event cloudevents.Event) cloudevents.Result
 	case tektoncloudevent.PipelineRunStartedEventV1,
 		tektoncloudevent.PipelineRunFailedEventV1,
 		tektoncloudevent.PipelineRunSuccessfulEventV1:
-		receivers := h.Receivers
+		var receivers []string
+		// send notify to the trigger user if it's existed, else send to the receivers configurated by type.
 		if receiver := getTriggerUser(data.PipelineRun); receiver != "" {
-			receivers = append(receivers, receiver)
+			receivers = []string{receiver}
+		} else {
+			receivers = append(h.Receivers[defaultReceiversKey], h.Receivers[event.Type()]...)
 		}
 
-		return sendLarkMessages(h.LarkClient, receivers, event, h.RunDetailBaseURL)
+		return sendLarkMessages(h.LarkClient, receivers, event, h.DashboardBaseURL)
 	default:
 		log.Debug().Str("ce-type", event.Type()).Msg("skip notifing for the event type.")
 		return cloudevents.ResultACK
 	}
 }
-
-func getTriggerUser(pr *v1beta1.PipelineRun) string {
-	eventContext := pr.Annotations[eventContextAnnotationKey]
-	if eventContext == "" {
-		return ""
-	}
-
-	contextData := make(map[string]string)
-	if err := json.Unmarshal([]byte(eventContext), &contextData); err != nil {
-		return ""
-	}
-
-	return contextData[eventContextAnnotationInnerKeyUser]
-}

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

@@ -9,6 +9,8 @@ import (
 	lark "github.com/larksuite/oapi-sdk-go/v3"
 	tektoncloudevent "github.com/tektoncd/pipeline/pkg/reconciler/events/cloudevent"
 
+	"github.com/PingCAP-QE/ee-apps/cloudevents-server/pkg/config"
+
 	_ "embed"
 )
 
@@ -49,9 +51,11 @@ func Test_pipelineRunHandler_Handle(t *testing.T) {
 	}
 
 	h := &pipelineRunHandler{
-		LarkClient:       lark.NewClient(larkAppID, larkAppSecret, lark.WithLogReqAtDebug(true), lark.WithEnableTokenCache(true)),
-		Receivers:        []string{receiver},
-		RunDetailBaseURL: baseURL,
+		LarkClient: lark.NewClient(larkAppID, larkAppSecret, lark.WithLogReqAtDebug(true), lark.WithEnableTokenCache(true)),
+		Tekton: config.Tekton{
+			Receivers:        map[string][]string{"*": {receiver}},
+			DashboardBaseURL: baseURL,
+		},
 	}
 	for _, tt := range tests {
 		t.Run(string(tt.name), func(t *testing.T) {

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

@@ -3,16 +3,16 @@ package tekton
 import (
 	"net/http"
 
+	"github.com/PingCAP-QE/ee-apps/cloudevents-server/pkg/config"
 	cloudevents "github.com/cloudevents/sdk-go/v2"
 	lark "github.com/larksuite/oapi-sdk-go/v3"
 	"github.com/rs/zerolog/log"
 	tektoncloudevent "github.com/tektoncd/pipeline/pkg/reconciler/events/cloudevent"
 )
 
 type taskRunHandler struct {
-	LarkClient       *lark.Client
-	RunDetailBaseURL string
-	Receivers        []string
+	config.Tekton
+	LarkClient *lark.Client
 }
 
 func (h *taskRunHandler) SupportEventTypes() []string {
@@ -33,7 +33,15 @@ func (h *taskRunHandler) Handle(event cloudevents.Event) cloudevents.Result {
 
 	switch event.Type() {
 	case string(tektoncloudevent.TaskRunFailedEventV1):
-		return sendLarkMessages(h.LarkClient, h.Receivers, event, h.RunDetailBaseURL)
+		var receivers []string
+		// send notify to the trigger user if it's existed, else send to the receivers configurated by type.
+		if receiver := getTriggerUser(data.TaskRun); receiver != "" {
+			receivers = []string{receiver}
+		} else {
+			receivers = append(h.Receivers[defaultReceiversKey], h.Receivers[event.Type()]...)
+		}
+
+		return sendLarkMessages(h.LarkClient, receivers, event, h.DashboardBaseURL)
 	default:
 		log.Debug().Str("ce-type", event.Type()).Msg("skip notifing for the event type.")
 		return cloudevents.ResultACK

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

@@ -5,6 +5,7 @@ import (
 	"reflect"
 	"testing"
 
+	"github.com/PingCAP-QE/ee-apps/cloudevents-server/pkg/config"
 	cloudevents "github.com/cloudevents/sdk-go/v2"
 	lark "github.com/larksuite/oapi-sdk-go/v3"
 	tektoncloudevent "github.com/tektoncd/pipeline/pkg/reconciler/events/cloudevent"
@@ -45,9 +46,11 @@ func Test_taskRunHandler_Handle(t *testing.T) {
 	}
 
 	h := &taskRunHandler{
-		LarkClient:       lark.NewClient(larkAppID, larkAppSecret, lark.WithLogReqAtDebug(true), lark.WithEnableTokenCache(true)),
-		Receivers:        []string{receiver},
-		RunDetailBaseURL: baseURL,
+		LarkClient: lark.NewClient(larkAppID, larkAppSecret, lark.WithLogReqAtDebug(true), lark.WithEnableTokenCache(true)),
+		Tekton: config.Tekton{
+			Receivers:        map[string][]string{"*": {receiver}},
+			DashboardBaseURL: baseURL,
+		},
 	}
 	for _, tt := range tests {
 		t.Run(string(tt.name), func(t *testing.T) {

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

@@ -3,7 +3,6 @@ package tekton
 import (
 	"bytes"
 	"context"
-	"crypto/tls"
 	"encoding/json"
 	"fmt"
 	"net/http"
@@ -23,8 +22,6 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"knative.dev/pkg/apis"
 
-	"github.com/PingCAP-QE/ee-apps/cloudevents-server/pkg/config"
-
 	_ "embed"
 )
 
@@ -72,21 +69,11 @@ func newMessageReq(receiver string, messageRawStr string) *larkim.CreateMessageR
 		Build()
 }
 
-func newLarkClient(cfg config.Lark) *lark.Client {
-	// Disable certificate verification
-	tr := &http.Transport{
-		TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
+func sendLarkMessages(client *lark.Client, receivers []string, event cloudevents.Event, detailBaseUrl string) protocol.Result {
+	if len(receivers) == 0 {
+		return cloudevents.ResultACK
 	}
-	httpClient := &http.Client{Transport: tr}
 
-	return lark.NewClient(cfg.AppID, cfg.AppSecret,
-		lark.WithLogReqAtDebug(true),
-		lark.WithEnableTokenCache(true),
-		lark.WithHttpClient(httpClient),
-	)
-}
-
-func sendLarkMessages(client *lark.Client, receivers []string, event cloudevents.Event, detailBaseUrl string) protocol.Result {
 	createMsgReqs, err := newLarkMessages(receivers, event, detailBaseUrl)
 	if err != nil {
 		log.Error().Err(err).Msg("compose lark message failed")

cloudevents-server/pkg/lark/client.go

@@ -0,0 +1,22 @@
+package lark
+
+import (
+	"crypto/tls"
+	"net/http"
+
+	larksdk "github.com/larksuite/oapi-sdk-go/v3"
+)
+
+func NewClient(appID, appSecret string) *larksdk.Client {
+	// Disable certificate verification
+	tr := &http.Transport{
+		TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
+	}
+	httpClient := &http.Client{Transport: tr}
+
+	return larksdk.NewClient(appID, appSecret,
+		larksdk.WithLogReqAtDebug(true),
+		larksdk.WithEnableTokenCache(true),
+		larksdk.WithHttpClient(httpClient),
+	)
+}