studio/flamenco
studio/flamenco
12
30
Fork 38
Code Issues 64 Pull Requests 5 Releases Wiki Activity

Support pausing jobs #104313

Manually merged
Sybren A. Stüvel merged 28 commits from David-Zhang-10/flamenco:paused-job-status into main 2024-07-01 17:53:44 +02:00
Conversation 1 Commits 28 Files Changed 12 +14 -10
2 changed files with 14 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 4b4511f9b6 - Show all commits

22
internal/manager/task_state_machine/task_state_machine.go
View File

@ -180,16 +180,22 @@ func (sm *StateMachine) updateJobOnTaskStatusCanceled(ctx context.Context, logge
return sm.JobStatusChange(ctx, job, api.JobStatusCanceled, "canceled task was last runnable task of job, canceling job") return sm.JobStatusChange(ctx, job, api.JobStatusCanceled, "canceled task was last runnable task of job, canceling job")
} }
numActive, _, err := sm.persist.CountTasksOfJobInStatus(ctx, job, api.TaskStatusActive) // Deal with the special case when the job is in pause-requested status.
if err != nil { if job.Status != api.JobStatusPauseRequested {
return err return nil
} } else {
if numActive == 0 && job.Status == api.JobStatusPauseRequested { numActive, _, err := sm.persist.CountTasksOfJobInStatus(ctx, job, api.TaskStatusActive)
// there is no active task, and the job is in pause-requested status, so we can pause the job if err != nil {
logger.Info().Msg("all tasks of job are completed, job is paused") return err
return sm.JobStatusChange(ctx, job, api.JobStatusPaused, "all tasks completed") }
if numActive == 0 {
// there is no active task, and the job is in pause-requested status, so we can pause the job
logger.Info().Msg("all tasks of job are completed, job is paused")
return sm.JobStatusChange(ctx, job, api.JobStatusPaused, "all tasks completed")
}
} }
// Execution should not reach here.
return nil return nil
} }

2
internal/manager/task_state_machine/task_state_machine_test.go
View File

@ -191,8 +191,6 @@ func TestTaskStatusChangeCancelSingleTask(t *testing.T) {
mocks.persist.EXPECT().CountTasksOfJobInStatus(ctx, job, mocks.persist.EXPECT().CountTasksOfJobInStatus(ctx, job,
api.TaskStatusActive, api.TaskStatusQueued, api.TaskStatusSoftFailed, api.TaskStatusPaused). api.TaskStatusActive, api.TaskStatusQueued, api.TaskStatusSoftFailed, api.TaskStatusPaused).
Return(1, 2, nil) Return(1, 2, nil)
mocks.persist.EXPECT().CountTasksOfJobInStatus(ctx, job,
api.TaskStatusActive).Return(0, 2, nil)
require.NoError(t, sm.TaskStatusChange(ctx, task, api.TaskStatusCanceled)) require.NoError(t, sm.TaskStatusChange(ctx, task, api.TaskStatusCanceled))
// T2: queued > cancelled --> J: cancel-requested > canceled // T2: queued > cancelled --> J: cancel-requested > canceled