Pipeline created from API gets overwritten by config file with duplicate ID

pkg/provisioning/errors.go

@@ -17,5 +17,6 @@ package provisioning
 import "github.com/conduitio/conduit/pkg/foundation/cerrors"
 
 var (
-	ErrDuplicatedPipelineID = cerrors.New("duplicated pipeline ID")
+	ErrDuplicatedPipelineID    = cerrors.New("duplicated pipeline ID")
+	ErrDuplicatedAPIPipelineID = cerrors.New("duplicated pipeline ID already exists from API")
 )

pkg/provisioning/service.go

@@ -102,6 +102,20 @@ func (s *Service) Init(ctx context.Context) error {
 		allPls = append(allPls, duplicateID)
 	}
 
+	// remove pipelines with duplicate IDs from API pipelines
+	var existingPipelinesFromAPI []int
+	for i, pl := range configs {
+		_, err := s.pipelineService.Get(ctx, pl.ID)
+		// if err is nil then a pipeline exists already, so we add towards duplicate ID list
+		if err == nil {
+			multierr = cerrors.Errorf("pipelines with ID %q will be skipped: %w", pl.ID, ErrDuplicatedAPIPipelineID)
+			existingPipelinesFromAPI = append(existingPipelinesFromAPI, i)
+		} else if !cerrors.Is(err, pipeline.ErrInstanceNotFound) {
+			multierr = cerrors.Errorf("could not get pipeline with ID %v: %w", pl.ID, err)
+		}
+	}
+	configs = s.deleteIndexes(configs, existingPipelinesFromAPI)
+
 	// contains pipelineIDs of successfully provisioned pipelines.
 	var successPls []string
 	for _, cfg := range configs {

pkg/provisioning/service_test.go

@@ -110,10 +110,12 @@ func TestProvision_Create(t *testing.T) {
 	service, pipelineService, connService, procService, plugService := newTestService(ctrl, logger)
 	service.pipelinesPath = "./test/pipelines1"
 
+	// no duplicate pipeline from API
+	pipelineService.EXPECT().Get(anyCtx, p1.P1.ID).Return(nil, pipeline.ErrInstanceNotFound)
+
 	pipelineService.EXPECT().List(anyCtx)
 	// pipeline doesn't exist
 	pipelineService.EXPECT().Get(anyCtx, p1.P1.ID).Return(nil, pipeline.ErrInstanceNotFound)
-
 	// create pipeline
 	pipelineService.EXPECT().CreateWithInstance(anyCtx, p1.P1)
 	pipelineService.EXPECT().UpdateDLQ(anyCtx, p1.P1.ID, p1.P1.DLQ)
@@ -144,6 +146,9 @@ func TestProvision_Update(t *testing.T) {
 	service.pipelinesPath = "./test/pipelines1"
 
 	pipelineService.EXPECT().List(anyCtx)
+	// no duplicate pipeline from API
+	pipelineService.EXPECT().Get(anyCtx, p1.P1.ID).Return(nil, pipeline.ErrInstanceNotFound)
+
 	// pipeline exists
 	pipelineService.EXPECT().Get(anyCtx, p1.P1.ID).Return(oldPipelineInstance, nil)
 
@@ -204,6 +209,9 @@ func TestProvision_NoRollbackOnFailedStart(t *testing.T) {
 	service, pipelineService, connService, procService, plugService := newTestService(ctrl, logger)
 	service.pipelinesPath = "./test/pipelines1"
 
+	// no duplicate pipeline from API
+	pipelineService.EXPECT().Get(anyCtx, p1.P1.ID).Return(nil, pipeline.ErrInstanceNotFound)
+
 	pipelineService.EXPECT().List(anyCtx)
 	pipelineService.EXPECT().Get(anyCtx, p1.P1.ID).Return(nil, pipeline.ErrInstanceNotFound)
 
@@ -237,6 +245,9 @@ func TestProvision_RollbackCreate(t *testing.T) {
 	service, pipelineService, connService, procService, plugService := newTestService(ctrl, logger)
 	service.pipelinesPath = "./test/pipelines1"
 
+	// no duplicate pipeline from API
+	pipelineService.EXPECT().Get(anyCtx, p1.P1.ID).Return(nil, pipeline.ErrInstanceNotFound)
+
 	pipelineService.EXPECT().List(anyCtx)
 	pipelineService.EXPECT().Get(anyCtx, p1.P1.ID).Return(nil, pipeline.ErrInstanceNotFound)
 
@@ -273,6 +284,9 @@ func TestProvision_RollbackUpdate(t *testing.T) {
 	service, pipelineService, connService, procService, _ := newTestService(ctrl, logger)
 	service.pipelinesPath = "./test/pipelines1"
 
+	// no duplicate pipeline from API
+	pipelineService.EXPECT().Get(anyCtx, p1.P1.ID).Return(nil, pipeline.ErrInstanceNotFound)
+
 	pipelineService.EXPECT().List(anyCtx)
 	// pipeline exists
 	pipelineService.EXPECT().Get(anyCtx, p1.P1.ID).Return(oldPipelineInstance, nil)
@@ -310,6 +324,9 @@ func TestProvision_MultiplePipelinesDuplicatedPipelineID(t *testing.T) {
 	service, pipelineService, connService, procService, plugService := newTestService(ctrl, logger)
 	service.pipelinesPath = "./test/pipelines2"
 
+	// no duplicate pipeline from API
+	pipelineService.EXPECT().Get(anyCtx, p2.P2.ID).Return(nil, pipeline.ErrInstanceNotFound)
+
 	pipelineService.EXPECT().List(anyCtx)
 	pipelineService.EXPECT().Get(anyCtx, p2.P2.ID).Return(nil, pipeline.ErrInstanceNotFound)
 
@@ -336,6 +353,10 @@ func TestProvision_MultiplePipelines(t *testing.T) {
 	service, pipelineService, connService, procService, plugService := newTestService(ctrl, logger)
 	service.pipelinesPath = "./test/pipelines3"
 
+	// no duplicate pipeline from API
+	pipelineService.EXPECT().Get(anyCtx, p3.P1.ID).Return(nil, pipeline.ErrInstanceNotFound)
+	pipelineService.EXPECT().Get(anyCtx, p3.P2.ID).Return(nil, pipeline.ErrInstanceNotFound)
+
 	pipelineService.EXPECT().List(anyCtx)
 	pipelineService.EXPECT().Get(anyCtx, p3.P1.ID).Return(nil, pipeline.ErrInstanceNotFound)
 	pipelineService.EXPECT().Get(anyCtx, p3.P2.ID).Return(nil, pipeline.ErrInstanceNotFound)
@@ -366,6 +387,59 @@ func TestProvision_MultiplePipelines(t *testing.T) {
 	is.NoErr(err)
 }
 
+func TestProvision_DuplicatePipelineFromAPI(t *testing.T) {
+	is := is.New(t)
+	logger := log.Nop()
+	ctrl := gomock.NewController(t)
+
+	service, pipelineService, _, _, _ := newTestService(ctrl, logger)
+	service.pipelinesPath = "./test/pipelines1"
+
+	// duplicate pipeline found from API
+	pipelineService.EXPECT().Get(anyCtx, p1.P1.ID).Return(oldPipelineInstance, nil)
+
+	pipelineService.EXPECT().List(anyCtx)
+
+	err := service.Init(context.Background())
+	is.True(cerrors.Is(err, ErrDuplicatedAPIPipelineID))
+}
+
+func TestProvision_DuplicatePipelineFromAPI_Error(t *testing.T) {
+	is := is.New(t)
+	logger := log.Nop()
+	ctrl := gomock.NewController(t)
+	otherErr := cerrors.New("error1")
+
+	service, pipelineService, connService, procService, plugService := newTestService(ctrl, logger)
+	service.pipelinesPath = "./test/pipelines1"
+
+	// duplicate pipeline found from API
+	pipelineService.EXPECT().Get(anyCtx, p1.P1.ID).Return(nil, otherErr)
+
+	pipelineService.EXPECT().List(anyCtx)
+	// pipeline doesn't exist
+	pipelineService.EXPECT().Get(anyCtx, p1.P1.ID).Return(nil, pipeline.ErrInstanceNotFound)
+	// create pipeline
+	pipelineService.EXPECT().CreateWithInstance(anyCtx, p1.P1)
+	pipelineService.EXPECT().UpdateDLQ(anyCtx, p1.P1.ID, p1.P1.DLQ)
+	pipelineService.EXPECT().AddConnector(anyCtx, p1.P1.ID, p1.P1.ConnectorIDs[0])
+	pipelineService.EXPECT().AddConnector(anyCtx, p1.P1.ID, p1.P1.ConnectorIDs[1])
+	pipelineService.EXPECT().AddProcessor(anyCtx, p1.P1.ID, p1.P1.ProcessorIDs[0])
+
+	connService.EXPECT().CreateWithInstance(anyCtx, p1.P1C1)
+	connService.EXPECT().CreateWithInstance(anyCtx, p1.P1C2)
+	connService.EXPECT().AddProcessor(anyCtx, p1.P1C2.ID, p1.P1C2.ProcessorIDs[0])
+
+	procService.EXPECT().CreateWithInstance(anyCtx, p1.P1C2P1)
+	procService.EXPECT().CreateWithInstance(anyCtx, p1.P1P1)
+
+	// start pipeline
+	pipelineService.EXPECT().Start(anyCtx, connService, procService, plugService, p1.P1.ID)
+
+	err := service.Init(context.Background())
+	is.True(cerrors.Is(err, otherErr))
+}
+
 func TestProvision_IntegrationTestServices(t *testing.T) {
 	is := is.New(t)
 	ctx, killAll := context.WithCancel(context.Background())