fix the bug of concurrent create snippet pipelines

erda-project · Nov 22, 2021 · 524da96 · 524da96
1 parent 0d67328
commit 524da96
Show file tree

Hide file tree

Showing 4 changed files with 57 additions and 10 deletions.
diff --git a/modules/pipeline/services/pipelinesvc/create_v2.go b/modules/pipeline/services/pipelinesvc/create_v2.go
@@ -31,13 +31,13 @@ import (
 
 func (s *PipelineSvc) CreateV2(req *apistructs.PipelineCreateRequestV2) (*spec.Pipeline, error) {
 	// validate
-	if err := s.validateCreateRequest(req); err != nil {
+	if err := s.ValidateCreateRequest(req); err != nil {
 		return nil, err
 	}
 	// set default
 	setDefault(req)
 
-	p, err := s.makePipelineFromRequestV2(req)
+	p, err := s.MakePipelineFromRequestV2(req)
 	if err != nil {
 		return nil, err
 	}
@@ -79,8 +79,8 @@ const (
 	maxSqlIndexLength = 191
 )
 
-// validateCreateRequest validate pipelineCreateRequestV2
-func (s *PipelineSvc) validateCreateRequest(req *apistructs.PipelineCreateRequestV2) error {
+// ValidateCreateRequest validate pipelineCreateRequestV2
+func (s *PipelineSvc) ValidateCreateRequest(req *apistructs.PipelineCreateRequestV2) error {
 	if req == nil {
 		return apierrors.ErrCreatePipeline.MissingParameter("request")
 	}
@@ -138,7 +138,7 @@ func logCompatibleFailed(key, value string, err error) {
 	logrus.Errorf("compatible from labels failed, key: %s, value: %s, err: %v", key, value, err)
 }
 
-func (s *PipelineSvc) makePipelineFromRequestV2(req *apistructs.PipelineCreateRequestV2) (*spec.Pipeline, error) {
+func (s *PipelineSvc) MakePipelineFromRequestV2(req *apistructs.PipelineCreateRequestV2) (*spec.Pipeline, error) {
 	p := &spec.Pipeline{}
 
 	// 解析 pipeline yml 文件，生成最终 pipeline yml 文件

diff --git a/modules/pipeline/services/pipelinesvc/create_v2_test.go b/modules/pipeline/services/pipelinesvc/create_v2_test.go
@@ -56,7 +56,7 @@ func TestValidateCreateRequest(t *testing.T) {
 			"1": "value",
 		},
 	}
-	err := svc.validateCreateRequest(&req)
+	err := svc.ValidateCreateRequest(&req)
 	assert.NoError(t, err)
 	spew.Dump(req)
 }
@@ -98,7 +98,7 @@ stages:
 	})
 	defer pm.Unpatch()
 
-	p, err := svc.makePipelineFromRequestV2(&req)
+	p, err := svc.MakePipelineFromRequestV2(&req)
 	assert.NoError(t, err)
 	assert.Equal(t, "custom-namespace", p.Extra.Namespace)
 	assert.Equal(t, true, p.Extra.NotPipelineControlledNs)

diff --git a/modules/pipeline/services/pipelinesvc/snippet.go b/modules/pipeline/services/pipelinesvc/snippet.go
@@ -283,8 +283,12 @@ func (s *PipelineSvc) MakeSnippetPipeline4Create(p *spec.Pipeline, snippetTask *
 	for k, v := range snippetTask.Extra.Action.Params {
 		runParams = append(runParams, apistructs.PipelineRunParam{Name: k, Value: v})
 	}
+
+	labels := make(map[string]string)
+	for k, v := range p.Labels {
+		labels[k] = v
+	}
 	// transfer snippetTask to pipeline create request
-	labels := p.Labels
 	for k, v := range snippetConfig.Labels {
 		labels[k] = v
 	}
@@ -301,10 +305,10 @@ func (s *PipelineSvc) MakeSnippetPipeline4Create(p *spec.Pipeline, snippetTask *
 		RunParams:              runParams,
 		IdentityInfo:           p.GenIdentityInfo(),
 	}
-	if err := s.validateCreateRequest(&snippetPipelineCreateReq); err != nil {
+	if err := s.ValidateCreateRequest(&snippetPipelineCreateReq); err != nil {
 		return nil, apierrors.ErrCreateSnippetPipeline.InternalError(err)
 	}
-	snippetP, err := s.makePipelineFromRequestV2(&snippetPipelineCreateReq)
+	snippetP, err := s.MakePipelineFromRequestV2(&snippetPipelineCreateReq)
 	if err != nil {
 		return nil, err
 	}

diff --git a/modules/pipeline/services/pipelinesvc/snippet_test.go b/modules/pipeline/services/pipelinesvc/snippet_test.go
@@ -17,13 +17,16 @@ package pipelinesvc
 import (
 	"encoding/json"
 	"reflect"
+	"strconv"
 	"testing"
 
 	"bou.ke/monkey"
 	"github.com/alecthomas/assert"
 
+
 	"github.com/erda-project/erda/apistructs"
 	"github.com/erda-project/erda/bundle"
+	"github.com/erda-project/erda/modules/pipeline/spec"
 	"github.com/erda-project/erda/pkg/expression"
 	"github.com/erda-project/erda/pkg/pipeline_snippet_client"
 )
@@ -251,3 +254,43 @@ func Test_ActionJson(t *testing.T) {
 		t.Fail()
 	}
 }
+
+func TestMakeSnippetPipeline4Create(t *testing.T) {
+	tasks := make([]*spec.PipelineTask, 0, 10)
+	for i := 0; i < 10; i++ {
+		tasks = append(tasks, &spec.PipelineTask{
+			ID:         uint64(i),
+			PipelineID: 0,
+			StageID:    0,
+			Name:       strconv.Itoa(i),
+			Extra: spec.PipelineTaskExtra{
+				Labels: map[string]string{"setID": strconv.Itoa(i)},
+			},
+		})
+	}
+	p := &spec.Pipeline{
+		Labels: map[string]string{"setID": "100"},
+	}
+
+	s := &PipelineSvc{}
+	monkey.PatchInstanceMethod(reflect.TypeOf(s), "ValidateCreateRequest", func(*PipelineSvc, *apistructs.PipelineCreateRequestV2) error {
+		return nil
+	})
+	monkey.PatchInstanceMethod(reflect.TypeOf(s), "MakePipelineFromRequestV2", func(*PipelineSvc, *apistructs.PipelineCreateRequestV2) (*spec.Pipeline, error) {
+		return nil, nil
+	})
+	defer monkey.UnpatchAll()
+	for i := range tasks {
+		go func() {
+			_, err := s.MakeSnippetPipeline4Create(p, tasks[i], "")
+			if err != nil {
+				t.Error(err)
+				return
+			}
+			if p.Labels["setID"] != "100" {
+				t.Fatal("fail")
+			}
+		}()
+	}
+
+}