register cluster hook

apistructs/cluster.go

@@ -42,12 +42,6 @@ const (
 	ClusterActionDelete = "delete"
 )
 
-const (
-	ClusterTypeDcos = "dcos"
-	ClusterTypeK8S  = "k8s"
-	ClusterTypeEdas = "edas"
-)
-
 // token / client-cert&client-key / proxy(dialer)
 const (
 	ManageToken = "token"

modules/pipeline/initialize.go

@@ -48,12 +48,14 @@ import (
 	"github.com/erda-project/erda/modules/pipeline/services/reportsvc"
 	"github.com/erda-project/erda/modules/pipeline/services/snippetsvc"
 	"github.com/erda-project/erda/modules/pkg/websocket"
+	"github.com/erda-project/erda/pkg/discover"
 	"github.com/erda-project/erda/pkg/http/httpserver"
 	"github.com/erda-project/erda/pkg/jsonstore"
 	"github.com/erda-project/erda/pkg/jsonstore/etcd"
 	"github.com/erda-project/erda/pkg/loop"
 	"github.com/erda-project/erda/pkg/pipeline_network_hook_client"
 	"github.com/erda-project/erda/pkg/pipeline_snippet_client"
+	"github.com/erda-project/erda/pkg/strutil"
 	// "terminus.io/dice/telemetry/promxp"
 )
 
@@ -200,6 +202,9 @@ func do() (*httpserver.Server, error) {
 		return nil, err
 	}
 
+	// register cluster hook
+	registerClusterHook(bdl)
+
 	return server, nil
 }
 
@@ -225,6 +230,23 @@ func registerSnippetClient(dbclient *dbclient.Client) error {
 	return nil
 }
 
+func registerClusterHook(bdl *bundle.Bundle) {
+	ev := apistructs.CreateHookRequest{
+		Name:   "pipeline_watch_cluster_changed",
+		Events: []string{bundle.ClusterEvent},
+		URL:    strutil.Concat("http://", discover.Pipeline(), "/clusterhook"),
+		Active: true,
+		HookLocation: apistructs.HookLocation{
+			Org:         "-1",
+			Project:     "-1",
+			Application: "-1",
+		},
+	}
+	if err := bdl.CreateWebhook(ev); err != nil {
+		logrus.Warnf("failed to register watch cluster changed event, %v", err)
+	}
+}
+
 func doCrondAbout(crondSvc *crondsvc.CrondSvc, pipelineSvc *pipelinesvc.PipelineSvc) error {
 	// 加载定时配置
 	logs, err := crondSvc.ReloadCrond(pipelineSvc.RunCronPipelineFunc)

modules/pipeline/pipengine/actionexecutor/plugins/scheduler/executor/manager.go

@@ -29,10 +29,6 @@ import (
 	"github.com/erda-project/erda/modules/pipeline/pkg/clusterinfo"
 )
 
-const (
-	CLUSTERTYPEK8S = "k8s"
-)
-
 // Manager for scheduler task executor
 type Manager struct {
 	sync.RWMutex
@@ -54,7 +50,7 @@ func (m *Manager) Initialize(cfgs []apistructs.ClusterInfo) error {
 	logrus.Infof("pipeline scheduler task executor Inititalize ...")
 
 	for i := range cfgs {
-		if cfgs[i].Type != apistructs.ClusterTypeK8S {
+		if cfgs[i].Type != apistructs.K8S {
 			continue
 		}
 		if err := m.addExecutor(cfgs[i]); err != nil {
@@ -91,7 +87,7 @@ func (m *Manager) addExecutor(cluster apistructs.ClusterInfo) error {
 	defer m.Unlock()
 
 	switch cluster.Type {
-	case CLUSTERTYPEK8S:
+	case apistructs.K8S:
 		k8sjobCreate, ok := m.factory[k8sjob.Kind]
 		if ok {
 			name := types.Name(fmt.Sprintf("%sfor%s", cluster.Name, k8sjob.Kind))
@@ -187,7 +183,7 @@ func (m *Manager) deleteExecutor(cluster apistructs.ClusterInfo) {
 	defer m.Unlock()
 
 	switch cluster.Type {
-	case CLUSTERTYPEK8S:
+	case apistructs.K8S:
 		name := types.Name(fmt.Sprintf("%sfor%s", cluster.Name, k8sjob.Kind))
 		if _, exist := m.executors[name]; exist {
 			delete(m.executors, name)
@@ -212,7 +208,7 @@ func (m *Manager) updateExecutor(cluster apistructs.ClusterInfo) error {
 	defer m.Unlock()
 
 	switch cluster.Type {
-	case CLUSTERTYPEK8S:
+	case apistructs.K8S:
 		k8sjobCreate, ok := m.factory[k8sjob.Kind]
 		if ok {
 			name := types.Name(fmt.Sprintf("%sfor%s", cluster.Name, k8sjob.Kind))

modules/pipeline/pipengine/actionexecutor/plugins/scheduler/executor/plugins/k8sjob/k8sjob.go

@@ -290,12 +290,12 @@ func (k *K8sJob) Remove(ctx context.Context, task *spec.PipelineTask) (data inte
 func (k *K8sJob) BatchDelete(ctx context.Context, tasks []*spec.PipelineTask) (data interface{}, err error) {
 	for _, task := range tasks {
 		if len(task.Extra.UUID) <= 0 {
-		continue
-	}
+			continue
+		}
 		_, err = k.Remove(ctx, task)
 		if err != nil {
-		return nil, err
-	}
+			return nil, err
+		}
 	}
 	return nil, nil
 }

modules/pipeline/services/pipelinesvc/clusterinfo.go

@@ -62,9 +62,9 @@ func (s *PipelineSvc) retryQueryClusterInfo(clusterName string, pipelineID uint6
 
 // ClusterHook listen and dispatch cluster event from eventbox
 func (s *PipelineSvc) ClusterHook(clusterEvent apistructs.ClusterEvent) error {
-	if !strutil.Equal(clusterEvent.Content.Type, apistructs.ClusterTypeK8S, true) &&
-		!strutil.Equal(clusterEvent.Content.Type, apistructs.ClusterTypeEdas, true) &&
-		!strutil.Equal(clusterEvent.Content.Type, apistructs.ClusterTypeDcos, true) {
+	if !strutil.Equal(clusterEvent.Content.Type, apistructs.K8S, true) &&
+		!strutil.Equal(clusterEvent.Content.Type, apistructs.EDAS, true) &&
+		!strutil.Equal(clusterEvent.Content.Type, apistructs.DCOS, true) {
 		return errors.Errorf("invalid cluster event type: %s", clusterEvent.Content.Type)
 	}