pipeline refactor and cluster watch #731

chengjoey · 2021-06-30T06:49:51Z

What type of this PR

Add one of the following kinds:
/kind feature

What this PR does / why we need it:

pipeline refactor, add three executor in scheduler plugin(k8sjob, k8sflink, k8sspark), dcos and edas will dispath to scheduler module
get cluster from cluster-manager module, and register cluster hook

Which issue(s) this PR fixes:

Specified Reviewers:

/assign @your-reviewer

codecov · 2021-06-30T07:27:33Z

Codecov Report

Merging #731 (ec2b5a1) into master (93784d3) will increase coverage by 0.03%.
The diff coverage is 0.00%.

❗ Current head ec2b5a1 differs from pull request most recent head a58ff1a. Consider uploading reports for the commit a58ff1a to get more accurate results

@@            Coverage Diff             @@
##           master     #731      +/-   ##
==========================================
+ Coverage   10.14%   10.18%   +0.03%     
==========================================
  Files         940      937       -3     
  Lines       88167    87840     -327     
==========================================
  Hits         8946     8946              
+ Misses      78287    77961     -326     
+ Partials      934      933       -1

Impacted Files	Coverage Δ
...gine/actionexecutor/plugins/scheduler/scheduler.go	`0.00% <0.00%> (ø)`
...dules/pipeline/services/pipelinesvc/clusterinfo.go	`0.00% <0.00%> (ø)`
modules/pipeline/spec/pipeline_configs.go	`0.00% <ø> (ø)`
pkg/mock/mock.go	`24.65% <0.00%> (ø)`

sfwn · 2021-07-01T02:05:14Z

apistructs/cluster.go

+)
+
+const (
+	ClusterTypeDcos = "dcos"


Can you reuse definitions at line 20~22 in this file.

sfwn · 2021-07-02T07:13:09Z

modules/pipeline/endpoints/clusterinfo.go

+	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+		errStr := fmt.Sprintf("decode clusterhook request fail: %v", err)
+		logrus.Error(errStr)
+		return httpserver.HTTPResponse{Status: http.StatusBadRequest, Content: errStr}, nil


use httpserver.ErrResp

sfwn · 2021-07-02T07:14:01Z

modules/pipeline/endpoints/endpoints.go

@@ -245,5 +245,8 @@ func (e *Endpoints) Routes() []httpserver.Endpoint {
 		// reports
 		{Path: "/api/pipeline-reportsets/{pipelineID}", Method: http.MethodGet, Handler: e.queryPipelineReportSet},
 		{Path: "/api/pipeline-reportsets", Method: http.MethodGet, Handler: e.pagingPipelineReportSets},
+
+		// cluster info
+		{Path: "/clusterhook", Method: http.MethodPost, Handler: e.clusterHook},


clusterhook for what, you should specified in path.

sfwn · 2021-07-02T09:16:20Z

modules/pipeline/initialize.go

+	ev := apistructs.CreateHookRequest{
+		Name:   "pipeline_watch_cluster_changed",
+		Events: []string{bundle.ClusterEvent},
+		URL:    strutil.Concat("http://", discover.Pipeline(), "/clusterhook"),


Use same const variable with endpoints.

how about multi instance?

sfwn · 2021-07-02T09:17:03Z

modules/pipeline/endpoints/clusterinfo.go

+func (e *Endpoints) clusterHook(ctx context.Context, r *http.Request, vars map[string]string) (httpserver.Responser, error) {
+	req := apistructs.ClusterEvent{}
+	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+		errStr := fmt.Sprintf("decode clusterhook request fail: %v", err)


failed to xxx

sfwn · 2021-07-02T09:35:25Z

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

+	}
+}
+
+func (m *Manager) updateExecutor(cluster apistructs.ClusterInfo) error {


duplicated with addExecutor.

sfwn · 2021-07-04T14:49:52Z

go.sum

@@ -228,6 +228,8 @@ github.com/cheekybits/is v0.0.0-20150225183255-68e9c0620927/go.mod h1:h/aW8ynjgk
 github.com/cheggaaa/pb/v3 v3.0.1/go.mod h1:SqqeMF/pMOIu3xgGoxtPYhMNQP258xE4x/XRTYua+KU=
 github.com/cheggaaa/pb/v3 v3.0.4 h1:QZEPYOj2ix6d5oEg63fbHmpolrnNiwjUsk+h74Yt4bM=
 github.com/cheggaaa/pb/v3 v3.0.4/go.mod h1:7rgWxLrAUcFMkvJuv09+DYi7mMUYi8nO9iOWcvGJPfw=
+github.com/chengjoey/flink-on-k8s-operator v0.0.1 h1:a4zrgATUtbQXFrfz0OAH1C9crs4fojzp2k25oNlgfl0=


GoogleCloudPlatform/flink-on-k8s-operator#460

sfwn · 2021-07-04T14:54:44Z

modules/pipeline/endpoints/clusterinfo.go

+func (e *Endpoints) clusterHook(ctx context.Context, r *http.Request, vars map[string]string) (httpserver.Responser, error) {
+	req := apistructs.ClusterEvent{}
+	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+		errStr := fmt.Sprintf("failed to decode clusterhook request err: %v", err)


failed to decode clusterhook request, err: %v

sfwn · 2021-07-04T14:54:55Z

modules/pipeline/endpoints/clusterinfo.go

+		return httpserver.ErrResp(http.StatusBadRequest, "", errStr)
+	}
+	if err := e.pipelineSvc.ClusterHook(req); err != nil {
+		errStr := fmt.Sprintf("failed to handle cluster event: %v", err)


failed to handle cluster event, err: %v

sfwn · 2021-07-04T14:55:05Z

modules/pipeline/endpoints/clusterinfo.go

+		logrus.Error(errStr)
+		return httpserver.ErrResp(http.StatusBadRequest, "", errStr)
+	}
+	return httpserver.HTTPResponse{Status: http.StatusOK}, nil


httpserver.OkResp

sfwn · 2021-07-04T14:56:01Z

modules/pipeline/endpoints/endpoints.go

@@ -37,6 +37,10 @@ import (
 	"github.com/erda-project/erda/pkg/http/httpserver"
 )

+const (
+	ClusterHookApiPath = "/api/pipeline-clusters/hook"


/api/{resource}/actions/{concreteAction}

sfwn · 2021-07-04T15:06:21Z

modules/pipeline/initialize.go

@@ -194,6 +201,9 @@ func do() (*httpserver.Server, error) {
 		return nil, err
 	}

+	// register cluster hook
+	registerClusterHook(bdl)


Put registerClusterHook into pkg clusterinfo is better?
Put these two methods together at least.

Or only one method here such as doClusterAbout.

scheduler initialze need cluster info, so clusterinfo.initialize should make first
cluster hook should register after pipeline service start

Move registerClusterHook into clusterinfo.Initialize

sfwn · 2021-07-04T15:08:31Z

modules/pipeline/initialize.go

+		},
+	}
+	if err := bdl.CreateWebhook(ev); err != nil {
+		logrus.Warnf("failed to register watch cluster changed event, %v", err)


Return err and handle err at invoker side.

And I think pipeline should panic if registerClusterHook failed at bootsstrap.

Each time the pod ip is different.
How about old pod ips already registered? Maybe eventbox will continue send msgs to them but failed of course. And the subscribers(pod ips) will be more and more ?

If so, maybe we should continue register with service ip and all instances listen from etcd.

sfwn · 2021-07-04T15:18:36Z

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

+	}
+	go m.listenClusterEventSync(context.Background(), eventChan)
+
+	logrus.Info("pipengine task executor manager Initialize Done .")


pipeline scheduler task executor

Check all others.

sfwn · 2021-07-04T15:21:02Z

modules/pipeline/pipengine/actionexecutor/plugins/scheduler/executor/plugins/k8sflink/flink.go

+)
+
+const (
+	DiceRootDomainKEY  = "DICE_ROOT_DOMAIN"


sfwn · 2021-07-04T15:21:58Z

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

+package k8sjob
+
+var (
+	errPullImage               = "拉取镜像失败"


Later with i18n

sfwn · 2021-07-04T15:25:21Z

modules/pipeline/pipengine/actionexecutor/plugins/scheduler/executor/types/define.go

+	"context"
+	"fmt"
+
+	"github.com/gogap/errors"


Check all error pkg, use fmt.Errorf or unified github.com/pkg/errors.

sfwn · 2021-07-04T15:26:01Z

modules/pipeline/pipengine/actionexecutor/plugins/scheduler/logic/shcedule.go

@@ -0,0 +1,45 @@
+// Copyright (c) 2021 Terminus, Inc.


Typo of file name schedule.

sfwn · 2021-07-04T15:29:36Z

modules/pipeline/pipengine/reconciler/taskrun/taskop/start.go

+	if data == nil {
+		return nil
+	}
+	jobID, ok := data.(string)


You should check task executor type here.
Not all string type data is job id.

only dcos return job id, so remove inJectJobID

sfwn · 2021-07-04T15:29:59Z

modules/pipeline/pipengine/reconciler/taskrun/taskop/start.go

@@ -159,6 +164,19 @@ func (s *start) TuneTriggers() taskrun.TaskOpTuneTriggers {
 	}
 }

+// injectJobID save flink, spark job id after start
+func injectJobID(tr *taskrun.TaskRun, data interface{}) error {


injectJobIDForFlinkSpark is better.

only dcos return str job id,already remove injectJobID

sfwn · 2021-07-04T15:30:24Z

modules/pipeline/services/pipelinesvc/clusterinfo.go

@@ -18,9 +18,11 @@ import (
 	"strings"
 	"time"

+	"github.com/gogap/errors"


sfwn · 2021-07-04T15:31:57Z

modules/pipeline/services/pipelinesvc/clusterinfo.go

+
+// ClusterHook listen and dispatch cluster event from eventbox
+func (s *PipelineSvc) ClusterHook(clusterEvent apistructs.ClusterEvent) error {
+	if !strutil.Equal(clusterEvent.Content.Type, apistructs.K8S, true) &&


Use switch grammar is more clear here.

sfwn · 2021-07-04T15:32:23Z

modules/pipeline/spec/pipeline_task.go

@@ -95,6 +95,8 @@ type PipelineTaskExtra struct {
 	LoopOptions *apistructs.PipelineTaskLoopOptions `json:"loopOptions,omitempty"` // 开始执行后保证不为空

 	AppliedResources apistructs.PipelineAppliedResources `json:"appliedResources,omitempty"`
+
+	JobID string `json:"jobID,omitempty"` // flink, spark job id


Concrete field name.

sfwn · 2021-07-05T01:42:49Z

modules/pipeline/initialize.go

@@ -194,6 +201,9 @@ func do() (*httpserver.Server, error) {
 		return nil, err
 	}

+	// register cluster hook
+	registerClusterHook(bdl)


Move registerClusterHook into clusterinfo.Initialize

…oudPlatform

sfwn · 2021-07-05T03:42:45Z

/approve

chengjoey force-pushed the feature/pipeline-clusterinfo branch from 1fb34d4 to 574c817 Compare June 30, 2021 07:07

Effet requested a review from sfwn July 1, 2021 01:47

Effet added pipeline pipeline service refactor labels Jul 1, 2021

sfwn requested changes Jul 1, 2021

View reviewed changes

apistructs/cluster.go Outdated

)

const (

ClusterTypeDcos = "dcos"

Copy link

Member

sfwn Jul 1, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can you reuse definitions at line 20~22 in this file.

chengjoey force-pushed the feature/pipeline-clusterinfo branch from 574c817 to 695a835 Compare July 1, 2021 13:14

sfwn requested changes Jul 2, 2021

View reviewed changes

chengjoey force-pushed the feature/pipeline-clusterinfo branch from 695a835 to 5fba941 Compare July 4, 2021 03:09

sfwn reviewed Jul 4, 2021

View reviewed changes

sfwn changed the title ~~Feature/pipeline refactor and cluster watch~~ pipeline refactor and cluster watch Jul 4, 2021

sfwn requested changes Jul 4, 2021

View reviewed changes

chengjoey force-pushed the feature/pipeline-clusterinfo branch from 5fba941 to 4c14eb3 Compare July 4, 2021 16:33

chengjoey added 16 commits July 5, 2021 09:32

get clusters when pipeline init

1c458a9

add k8sjob task executor for pipeline

eefbbcb

add pipeline cluster hook

7028ba8

add flink schema

abed09f

add k8s flink task executor

86bf557

add k8s spark task executor

ae839b8

add metronome task executor

83c49cd

add k8sjob task executor for pipeline

cba4637

set cluster info when task executor initialize

e9591dd

add affinity for k8sjob

4320a39

add flink and metronome task executor for pipeline scheduler

4008b55

add field job id for task extra

105ec65

add spark task executor

38d40e2

delete dc/os executor

eaea5bf

delete scheduler k8sjob,k8sflink,k8sspark executor

cea7091

register cluster hook

1c1167b

chengjoey force-pushed the feature/pipeline-clusterinfo branch from 4c14eb3 to d3b5121 Compare July 5, 2021 01:40

sfwn requested changes Jul 5, 2021

View reviewed changes

chengjoey force-pushed the feature/pipeline-clusterinfo branch from d3b5121 to 9ce8942 Compare July 5, 2021 02:00

replace mod pkg github.com/googlecloudplatform to github.com/GoogleCl…

39047d4

…oudPlatform

chengjoey force-pushed the feature/pipeline-clusterinfo branch from 9ce8942 to 39047d4 Compare July 5, 2021 03:25

sfwn approved these changes Jul 5, 2021

View reviewed changes

erda-bot added the approved label Jul 5, 2021

erda-bot approved these changes Jul 5, 2021

View reviewed changes

Merge branch 'master' into feature/pipeline-clusterinfo

a58ff1a

erda-bot merged commit 5860529 into erda-project:master Jul 5, 2021

chengjoey deleted the feature/pipeline-clusterinfo branch July 16, 2021 01:43

pipeline refactor and cluster watch #731

pipeline refactor and cluster watch #731

Conversation

chengjoey commented Jun 30, 2021

What type of this PR

What this PR does / why we need it:

Which issue(s) this PR fixes:

Specified Reviewers:

codecov bot commented Jun 30, 2021 • edited Loading

Codecov Report

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sfwn commented Jul 5, 2021

codecov bot commented Jun 30, 2021 •

edited

Loading