Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Handle concurrent access to a pipeline's status field #1794

Merged
merged 7 commits into from
Aug 21, 2024
Merged

Handle concurrent access to a pipeline's status field #1794

Show file tree
Hide file tree
Changes from 6 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
d9b1148
pipeline.Instance: guard access to status field
hariso Aug 16, 2024
b8aed97
fix tests
hariso Aug 19, 2024
0c66bc9
linter, comment
hariso Aug 19, 2024
087968c
Merge branch 'main' into haris/pipeline-recovery/atomic-status
hariso Aug 19, 2024
9e4d1a4
fix test
hariso Aug 19, 2024
7e940eb
Merge branch 'haris/pipeline-recovery/atomic-status' of github.com:Co…
hariso Aug 19, 2024
a2eaffe
Merge branch 'main' into haris/pipeline-recovery/atomic-status
hariso Aug 21, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions pkg/orchestrator/connectors.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,7 +61,7 @@ func (c *ConnectorOrchestrator) Create(
if pl.ProvisionedBy != pipeline.ProvisionTypeAPI {
return nil, cerrors.Errorf("cannot add a connector to the pipeline %q: %w", pl.ID, ErrImmutableProvisionedByConfig)
}
if pl.Status == pipeline.StatusRunning {
if pl.GetStatus() == pipeline.StatusRunning {
return nil, cerrors.Errorf("cannot create connector: %w", pipeline.ErrPipelineRunning)
}

Expand Down Expand Up @@ -124,7 +124,7 @@ func (c *ConnectorOrchestrator) Delete(ctx context.Context, id string) error {
if err != nil {
return err
}
if pl.Status == pipeline.StatusRunning {
if pl.GetStatus() == pipeline.StatusRunning {
return pipeline.ErrPipelineRunning
}
err = c.connectors.Delete(ctx, id, c.connectorPlugins)
Expand Down Expand Up @@ -171,7 +171,7 @@ func (c *ConnectorOrchestrator) Update(ctx context.Context, id string, config co
if err != nil {
return nil, err
}
if pl.Status == pipeline.StatusRunning {
if pl.GetStatus() == pipeline.StatusRunning {
return nil, pipeline.ErrPipelineRunning
}

Expand Down
61 changes: 36 additions & 25 deletions pkg/orchestrator/connectors_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,9 +36,10 @@ func TestConnectorOrchestrator_Create_Success(t *testing.T) {
plsMock, consMock, procsMock, connPluginMock, procPluginMock := newMockServices(t)

pl := &pipeline.Instance{
ID: uuid.NewString(),
Status: pipeline.StatusSystemStopped,
ID: uuid.NewString(),
}
pl.SetStatus(pipeline.StatusSystemStopped)

want := &connector.Instance{
ID: uuid.NewString(),
Type: connector.TypeSource,
Expand Down Expand Up @@ -110,9 +111,9 @@ func TestConnectorOrchestrator_Create_PipelineRunning(t *testing.T) {
plsMock, consMock, procsMock, connPluginMock, procPluginMock := newMockServices(t)

pl := &pipeline.Instance{
ID: uuid.NewString(),
Status: pipeline.StatusRunning,
ID: uuid.NewString(),
}
pl.SetStatus(pipeline.StatusRunning)

plsMock.EXPECT().
Get(gomock.AssignableToTypeOf(ctxType), pl.ID).
Expand All @@ -133,9 +134,9 @@ func TestConnectorOrchestrator_Create_PipelineProvisionByConfig(t *testing.T) {

pl := &pipeline.Instance{
ID: uuid.NewString(),
Status: pipeline.StatusRunning,
ProvisionedBy: pipeline.ProvisionTypeConfig,
}
pl.SetStatus(pipeline.StatusRunning)

plsMock.EXPECT().
Get(gomock.AssignableToTypeOf(ctxType), pl.ID).
Expand All @@ -155,9 +156,10 @@ func TestConnectorOrchestrator_Create_CreateConnectorError(t *testing.T) {
plsMock, consMock, procsMock, connPluginMock, procPluginMock := newMockServices(t)

pl := &pipeline.Instance{
ID: uuid.NewString(),
Status: pipeline.StatusSystemStopped,
ID: uuid.NewString(),
}
pl.SetStatus(pipeline.StatusSystemStopped)

config := connector.Config{}
wantErr := cerrors.New("test error")
plsMock.EXPECT().
Expand Down Expand Up @@ -195,9 +197,10 @@ func TestConnectorOrchestrator_Create_AddConnectorError(t *testing.T) {
plsMock, consMock, procsMock, connPluginMock, procPluginMock := newMockServices(t)

pl := &pipeline.Instance{
ID: uuid.NewString(),
Status: pipeline.StatusSystemStopped,
ID: uuid.NewString(),
}
pl.SetStatus(pipeline.StatusSystemStopped)

conn := &connector.Instance{
ID: uuid.NewString(),
Type: connector.TypeSource,
Expand Down Expand Up @@ -257,9 +260,10 @@ func TestConnectorOrchestrator_Delete_Success(t *testing.T) {
plsMock, consMock, procsMock, connPluginMock, procPluginMock := newMockServices(t)

pl := &pipeline.Instance{
ID: uuid.NewString(),
Status: pipeline.StatusSystemStopped,
ID: uuid.NewString(),
}
pl.SetStatus(pipeline.StatusSystemStopped)

conn := &connector.Instance{
ID: uuid.NewString(),
PipelineID: pl.ID,
Expand Down Expand Up @@ -308,9 +312,10 @@ func TestConnectorOrchestrator_Delete_PipelineRunning(t *testing.T) {
plsMock, consMock, procsMock, connPluginMock, procPluginMock := newMockServices(t)

pl := &pipeline.Instance{
ID: uuid.NewString(),
Status: pipeline.StatusRunning,
ID: uuid.NewString(),
}
pl.SetStatus(pipeline.StatusRunning)

conn := &connector.Instance{
ID: uuid.NewString(),
PipelineID: pl.ID,
Expand All @@ -336,9 +341,10 @@ func TestConnectorOrchestrator_Delete_ProcessorAttached(t *testing.T) {
plsMock, consMock, procsMock, connPluginMock, procPluginMock := newMockServices(t)

pl := &pipeline.Instance{
ID: uuid.NewString(),
Status: pipeline.StatusRunning,
ID: uuid.NewString(),
}
pl.SetStatus(pipeline.StatusRunning)

conn := &connector.Instance{
ID: uuid.NewString(),
PipelineID: pl.ID,
Expand All @@ -362,9 +368,10 @@ func TestConnectorOrchestrator_Delete_Fail(t *testing.T) {
plsMock, consMock, procsMock, connPluginMock, procPluginMock := newMockServices(t)

pl := &pipeline.Instance{
ID: uuid.NewString(),
Status: pipeline.StatusSystemStopped,
ID: uuid.NewString(),
}
pl.SetStatus(pipeline.StatusSystemStopped)

conn := &connector.Instance{
ID: uuid.NewString(),
PipelineID: pl.ID,
Expand Down Expand Up @@ -394,9 +401,10 @@ func TestConnectorOrchestrator_Delete_RemoveConnectorFailed(t *testing.T) {
plsMock, consMock, procsMock, connPluginMock, procPluginMock := newMockServices(t)

pl := &pipeline.Instance{
ID: uuid.NewString(),
Status: pipeline.StatusSystemStopped,
ID: uuid.NewString(),
}
pl.SetStatus(pipeline.StatusSystemStopped)

conn := &connector.Instance{
ID: uuid.NewString(),
Plugin: "test-plugin",
Expand Down Expand Up @@ -442,9 +450,10 @@ func TestConnectorOrchestrator_Update_Success(t *testing.T) {
plsMock, consMock, procsMock, connPluginMock, procPluginMock := newMockServices(t)

pl := &pipeline.Instance{
ID: uuid.NewString(),
Status: pipeline.StatusSystemStopped,
ID: uuid.NewString(),
}
pl.SetStatus(pipeline.StatusSystemStopped)

conn := &connector.Instance{
ID: uuid.NewString(),
Plugin: "test-plugin",
Expand Down Expand Up @@ -511,9 +520,10 @@ func TestConnectorOrchestrator_Update_PipelineRunning(t *testing.T) {
plsMock, consMock, procsMock, connPluginMock, procPluginMock := newMockServices(t)

pl := &pipeline.Instance{
ID: uuid.NewString(),
Status: pipeline.StatusRunning,
ID: uuid.NewString(),
}
pl.SetStatus(pipeline.StatusRunning)

conn := &connector.Instance{
ID: uuid.NewString(),
PipelineID: pl.ID,
Expand All @@ -540,9 +550,10 @@ func TestConnectorOrchestrator_Update_Fail(t *testing.T) {
plsMock, consMock, procsMock, connPluginMock, procPluginMock := newMockServices(t)

pl := &pipeline.Instance{
ID: uuid.NewString(),
Status: pipeline.StatusSystemStopped,
ID: uuid.NewString(),
}
pl.SetStatus(pipeline.StatusSystemStopped)

conn := &connector.Instance{
ID: uuid.NewString(),
Type: connector.TypeDestination,
Expand Down
6 changes: 3 additions & 3 deletions pkg/orchestrator/pipelines.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,7 +57,7 @@ func (po *PipelineOrchestrator) Update(ctx context.Context, id string, cfg pipel
return nil, cerrors.Errorf("pipeline %q cannot be updated: %w", pl.ID, ErrImmutableProvisionedByConfig)
}
// TODO lock pipeline
if pl.Status == pipeline.StatusRunning {
if pl.GetStatus() == pipeline.StatusRunning {
return nil, pipeline.ErrPipelineRunning
}
return po.pipelines.Update(ctx, pl.ID, cfg)
Expand All @@ -72,7 +72,7 @@ func (po *PipelineOrchestrator) Delete(ctx context.Context, id string) error {
if pl.ProvisionedBy != pipeline.ProvisionTypeAPI {
return cerrors.Errorf("pipeline %q cannot be deleted: %w", pl.ID, ErrImmutableProvisionedByConfig)
}
if pl.Status == pipeline.StatusRunning {
if pl.GetStatus() == pipeline.StatusRunning {
return pipeline.ErrPipelineRunning
}
if len(pl.ConnectorIDs) != 0 {
Expand All @@ -94,7 +94,7 @@ func (po *PipelineOrchestrator) UpdateDLQ(ctx context.Context, id string, dlq pi
return nil, cerrors.Errorf("pipeline %q cannot be updated: %w", pl.ID, ErrImmutableProvisionedByConfig)
}
// TODO lock pipeline
if pl.Status == pipeline.StatusRunning {
if pl.GetStatus() == pipeline.StatusRunning {
return nil, pipeline.ErrPipelineRunning
}

Expand Down
Loading