cherry pick feat(framework): add custom pipeline notification service #…

…7920 to v1.0 (#7924)

* feat(framework): add custom pipeline notification service

* fix(e2e): fix errors

backend/server/api/api.go

@@ -59,6 +59,15 @@ func Init() {
 	basicRes = services.GetBasicRes()
 }
 
+func InjectCustomService(pipelineNotifier services.PipelineNotificationService) errors.Error {
+	if pipelineNotifier != nil {
+		if err := services.InjectCustomService(pipelineNotifier); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 // @title  DevLake Swagger API
 // @version 0.1
 // @description  <h2>This is the main page of devlake api</h2>

backend/server/services/init.go

@@ -125,6 +125,14 @@ func Init() {
 	registerPluginsMigrationScripts()
 }
 
+func InjectCustomService(pipelineNotifier PipelineNotificationService) errors.Error {
+	if pipelineNotifier == nil {
+		return errors.Default.New("pipeline notifier is nil")
+	}
+	customPipelineNotificationService = pipelineNotifier
+	return nil
+}
+
 var statusLock sync.Mutex
 
 // ExecuteMigration executes all pending migration scripts and initialize services module

backend/server/services/pipeline.go

@@ -41,7 +41,7 @@ import (
 	"golang.org/x/sync/semaphore"
 )
 
-var notificationService *NotificationService
+var defaultNotificationService *DefaultPipelineNotificationService
 var globalPipelineLog = logruslog.Global.Nested("pipeline service")
 var pluginOptionSanitizers = map[string]func(map[string]interface{}){
 	"gitextractor": func(options map[string]interface{}) {
@@ -85,7 +85,7 @@ func pipelineServiceInit() {
 	var notificationEndpoint = cfg.GetString("NOTIFICATION_ENDPOINT")
 	var notificationSecret = cfg.GetString("NOTIFICATION_SECRET")
 	if strings.TrimSpace(notificationEndpoint) != "" {
-		notificationService = NewNotificationService(notificationEndpoint, notificationSecret)
+		defaultNotificationService = NewDefaultPipelineNotificationService(notificationEndpoint, notificationSecret)
 	}
 
 	// standalone mode: reset pipeline status
@@ -226,8 +226,10 @@ func GetPipeline(pipelineId uint64, shouldSanitize bool) (*models.Pipeline, erro
 	if err != nil {
 		return nil, err
 	}
-	if err := SanitizePipeline(dbPipeline); err != nil {
-		return nil, errors.Convert(err)
+	if shouldSanitize {
+		if err := SanitizePipeline(dbPipeline); err != nil {
+			return nil, errors.Convert(err)
+		}
 	}
 	return dbPipeline, nil
 }
@@ -352,23 +354,45 @@ func RunPipelineInQueue(pipelineMaxParallel int64) {
 	}
 }
 
+func getProjectName(pipeline *models.Pipeline) (string, errors.Error) {
+	if pipeline == nil {
+		return "", errors.Default.New("pipeline is nil")
+	}
+	blueprintId := pipeline.BlueprintId
+	dbBlueprint := &models.Blueprint{}
+	err := db.First(dbBlueprint, dal.Where("id = ?", blueprintId))
+	if err != nil {
+		if db.IsErrorNotFound(err) {
+			return "", errors.NotFound.New(fmt.Sprintf("blueprint(id: %d) not found", blueprintId))
+		}
+		return "", errors.Internal.Wrap(err, "error getting the blueprint from database")
+	}
+	return dbBlueprint.ProjectName, nil
+}
+
 // NotifyExternal FIXME ...
 func NotifyExternal(pipelineId uint64) errors.Error {
-	if notificationService == nil {
+	notification := GetPipelineNotificationService()
+	if notification == nil {
 		return nil
 	}
 	// send notification to an external web endpoint
 	pipeline, err := GetPipeline(pipelineId, true)
 	if err != nil {
 		return err
 	}
-	err = notificationService.PipelineStatusChanged(PipelineNotification{
-		PipelineID: pipeline.ID,
-		CreatedAt:  pipeline.CreatedAt,
-		UpdatedAt:  pipeline.UpdatedAt,
-		BeganAt:    pipeline.BeganAt,
-		FinishedAt: pipeline.FinishedAt,
-		Status:     pipeline.Status,
+	projectName, err := getProjectName(pipeline)
+	if err != nil {
+		return err
+	}
+	err = notification.PipelineStatusChanged(PipelineNotificationParam{
+		ProjectName: projectName,
+		PipelineID:  pipeline.ID,
+		CreatedAt:   pipeline.CreatedAt,
+		UpdatedAt:   pipeline.UpdatedAt,
+		BeganAt:     pipeline.BeganAt,
+		FinishedAt:  pipeline.FinishedAt,
+		Status:      pipeline.Status,
 	})
 	if err != nil {
 		globalPipelineLog.Error(err, "failed to send notification: %v", err)

backend/server/services/pipeline_notification.go

@@ -0,0 +1,49 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements.  See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License.  You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package services
+
+import (
+	"github.com/apache/incubator-devlake/core/errors"
+	"time"
+)
+
+type PipelineNotificationParam struct {
+	ProjectName string
+	PipelineID  uint64
+	CreatedAt   time.Time
+	UpdatedAt   time.Time
+	BeganAt     *time.Time
+	FinishedAt  *time.Time
+	Status      string
+}
+
+type PipelineNotificationService interface {
+	PipelineStatusChanged(params PipelineNotificationParam) errors.Error
+}
+
+var customPipelineNotificationService PipelineNotificationService
+
+func GetPipelineNotificationService() PipelineNotificationService {
+	if customPipelineNotificationService != nil {
+		return customPipelineNotificationService
+	}
+	if defaultNotificationService != nil {
+		return defaultNotificationService
+	}
+	return nil
+}

backend/server/services/notification.go → backend/server/services/pipeline_notification_impl.go

@@ -22,46 +22,34 @@ import (
 	"encoding/hex"
 	"encoding/json"
 	"fmt"
-	"io"
-	"net/http"
-	"strings"
-	"time"
-
 	"github.com/apache/incubator-devlake/core/errors"
 	"github.com/apache/incubator-devlake/core/models"
 	"github.com/apache/incubator-devlake/core/utils"
+	"io"
+	"net/http"
+	"strings"
 )
 
-// NotificationService FIXME ...
-type NotificationService struct {
+// DefaultPipelineNotificationService FIXME ...
+type DefaultPipelineNotificationService struct {
 	EndPoint string
 	Secret   string
 }
 
-// NewNotificationService FIXME ...
-func NewNotificationService(endpoint, secret string) *NotificationService {
-	return &NotificationService{
+// NewDefaultPipelineNotificationService creates a new DefaultPipelineNotificationService
+func NewDefaultPipelineNotificationService(endpoint, secret string) *DefaultPipelineNotificationService {
+	return &DefaultPipelineNotificationService{
 		EndPoint: endpoint,
 		Secret:   secret,
 	}
 }
 
-// PipelineNotification FIXME ...
-type PipelineNotification struct {
-	PipelineID uint64
-	CreatedAt  time.Time
-	UpdatedAt  time.Time
-	BeganAt    *time.Time
-	FinishedAt *time.Time
-	Status     string
-}
-
 // PipelineStatusChanged FIXME ...
-func (n *NotificationService) PipelineStatusChanged(params PipelineNotification) errors.Error {
+func (n *DefaultPipelineNotificationService) PipelineStatusChanged(params PipelineNotificationParam) errors.Error {
 	return n.sendNotification(models.NotificationPipelineStatusChanged, params)
 }
 
-func (n *NotificationService) sendNotification(notificationType models.NotificationType, data interface{}) errors.Error {
+func (n *DefaultPipelineNotificationService) sendNotification(notificationType models.NotificationType, data interface{}) errors.Error {
 	var dataJson, err = json.Marshal(data)
 	if err != nil {
 		return errors.Convert(err)
@@ -99,7 +87,7 @@ func (n *NotificationService) sendNotification(notificationType models.Notificat
 	return db.Update(notification)
 }
 
-func (n *NotificationService) signature(input, nouce string) string {
+func (n *DefaultPipelineNotificationService) signature(input, nouce string) string {
 	sum := sha256.Sum256([]byte(input + n.Secret + nouce))
 	return hex.EncodeToString(sum[:])
 }