studio/flamenco
studio/flamenco
11
29
Fork 38
Code Issues 62 Pull Requests 8 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 +7 -12
1 changed files with 7 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 4b20f8fad8 - Show all commits

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

@ -167,7 +167,7 @@ func (sm *StateMachine) jobStatusIfAThenB(
}
// isJobPausingComplete returns true when the job status is pause-requested and there are no more active tasks.
func (sm *StateMachine) isJobPausingComplete(ctx context.Context, logger zerolog.Logger, job *persistence.Job) (bool, error) {
func (sm *StateMachine) isJobPausingComplete(ctx context.Context, job *persistence.Job) (bool, error) {
if job.Status != api.JobStatusPauseRequested {
return false, nil
}
@ -175,12 +175,7 @@ func (sm *StateMachine) isJobPausingComplete(ctx context.Context, logger zerolog
if err != nil {
return false, err
}
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("No more active tasks, job is paused")
return true, nil
}
return false, nil
return numActive == 0, nil
}
// updateJobOnTaskStatusCanceled conditionally escalates the cancellation of a task to cancel the job.
@ -198,7 +193,7 @@ func (sm *StateMachine) updateJobOnTaskStatusCanceled(ctx context.Context, logge
}
// Deal with the special case when the job is in pause-requested status.
toBePaused, err := sm.isJobPausingComplete(ctx, logger, job)
toBePaused, err := sm.isJobPausingComplete(ctx, job)
if err != nil {
return err
}
@ -232,7 +227,7 @@ func (sm *StateMachine) updateJobOnTaskStatusFailed(ctx context.Context, logger
failLogger.Info().Msg("task failed, but not enough to fail the job")
// Deal with the special case when the job is in pause-requested status.
toBePaused, err := sm.isJobPausingComplete(ctx, logger, job)
toBePaused, err := sm.isJobPausingComplete(ctx, job)
if err != nil {
return err
}
@ -256,7 +251,7 @@ func (sm *StateMachine) updateJobOnTaskStatusCompleted(ctx context.Context, logg
}
// Deal with the special case when the job is in pause-requested status.
toBePaused, err := sm.isJobPausingComplete(ctx, logger, job)
toBePaused, err := sm.isJobPausingComplete(ctx, job)
if err != nil {
return err
}
@ -511,9 +506,9 @@ func (sm *StateMachine) pauseTasks(
}
// If pausing was requested, it has now happened, so the job can transition.
toBePaused, err := sm.isJobPausingComplete(ctx, logger, job)
toBePaused, err := sm.isJobPausingComplete(ctx, job)
if err != nil {
return "", fmt.Errorf("error when accessing number of active tasks")
return "", err
}
if toBePaused {
logger.Info().Msg("all tasks of job paused, job can go to 'paused' status")