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

WIP: Allow jobs to be submitted in paused status #104318

Closed
David Zhang wants to merge 32 commits from David-Zhang-10/flamenco:submit-as-paused into main
pull from: David-Zhang-10/flamenco:submit-as-paused
merge into: studio:main

When changing the target branch, be careful to rebase the branch in your fork to match. See documentation.
Conversation 0 Commits 32 Files Changed 23 +36 -31
1 changed files with 36 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 8f067da901 - Show all commits

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

@ -166,11 +166,26 @@ func (sm *StateMachine) jobStatusIfAThenB(
return sm.JobStatusChange(ctx, job, thenStatus, reason) return sm.JobStatusChange(ctx, job, thenStatus, reason)
} }
func (sm *StateMachine) shouldJobBePaused(ctx context.Context, logger zerolog.Logger, job *persistence.Job) (bool, error) {
if job.Status == api.JobStatusPauseRequested {
numActive, _, err := sm.persist.CountTasksOfJobInStatus(ctx, job, api.TaskStatusActive)
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
}
// updateJobOnTaskStatusCanceled conditionally escalates the cancellation of a task to cancel the job. // updateJobOnTaskStatusCanceled conditionally escalates the cancellation of a task to cancel the job.
func (sm *StateMachine) updateJobOnTaskStatusCanceled(ctx context.Context, logger zerolog.Logger, job *persistence.Job) error { func (sm *StateMachine) updateJobOnTaskStatusCanceled(ctx context.Context, logger zerolog.Logger, job *persistence.Job) error {
// If no more tasks can run, cancel the job. // If no more tasks can run, cancel the job.
numRunnable, _, err := sm.persist.CountTasksOfJobInStatus(ctx, job, numRunnable, _, err := sm.persist.CountTasksOfJobInStatus(ctx, job,
api.TaskStatusActive, api.TaskStatusQueued, api.TaskStatusSoftFailed, api.TaskStatusPaused) api.TaskStatusActive, api.TaskStatusQueued, api.TaskStatusSoftFailed)
if err != nil { if err != nil {
return err return err
} }
@ -181,16 +196,12 @@ func (sm *StateMachine) updateJobOnTaskStatusCanceled(ctx context.Context, logge
} }
// Deal with the special case when the job is in pause-requested status. // Deal with the special case when the job is in pause-requested status.
if job.Status == api.JobStatusPauseRequested { toBePaused, err := sm.shouldJobBePaused(ctx, logger, job)
numActive, _, err := sm.persist.CountTasksOfJobInStatus(ctx, job, api.TaskStatusActive) if err != nil {
if err != nil { return err
return err }
} if toBePaused {
if numActive == 0 { return sm.JobStatusChange(ctx, job, api.JobStatusPaused, "no more active tasks after task cancellation")
// 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 sm.JobStatusChange(ctx, job, api.JobStatusPaused, "all tasks completed")
}
} }
return nil return nil
@ -218,16 +229,13 @@ func (sm *StateMachine) updateJobOnTaskStatusFailed(ctx context.Context, logger
// If the job didn't fail, this failure indicates that at least the job is active. // If the job didn't fail, this failure indicates that at least the job is active.
failLogger.Info().Msg("task failed, but not enough to fail the job") failLogger.Info().Msg("task failed, but not enough to fail the job")
if job.Status == api.JobStatusPauseRequested { // Deal with the special case when the job is in pause-requested status.
numActive, _, err := sm.persist.CountTasksOfJobInStatus(ctx, job, api.TaskStatusActive) toBePaused, err := sm.shouldJobBePaused(ctx, logger, job)
if err != nil { if err != nil {
return err return err
} }
if numActive == 0 { if toBePaused {
// There is no active task, and the job is in pause-requested status, so we can pause the job. return sm.JobStatusChange(ctx, job, api.JobStatusPaused, "no more active tasks after task failure")
failLogger.Info().Msg("No more active tasks, job is paused")
return sm.JobStatusChange(ctx, job, api.JobStatusPaused, "all tasks completed")
}
} }
return sm.jobStatusIfAThenB(ctx, logger, job, api.JobStatusQueued, api.JobStatusActive, return sm.jobStatusIfAThenB(ctx, logger, job, api.JobStatusQueued, api.JobStatusActive,
@ -245,16 +253,13 @@ func (sm *StateMachine) updateJobOnTaskStatusCompleted(ctx context.Context, logg
return sm.JobStatusChange(ctx, job, api.JobStatusCompleted, "all tasks completed") return sm.JobStatusChange(ctx, job, api.JobStatusCompleted, "all tasks completed")
} }
if job.Status == api.JobStatusPauseRequested { // Deal with the special case when the job is in pause-requested status.
numActive, _, err := sm.persist.CountTasksOfJobInStatus(ctx, job, api.TaskStatusActive) toBePaused, err := sm.shouldJobBePaused(ctx, logger, job)
if err != nil { if err != nil {
return err return err
} }
if numActive == 0 { if toBePaused {
// There is no active task, and the job is in pause-requested status, so we can pause the job. return sm.JobStatusChange(ctx, job, api.JobStatusPaused, "no more active tasks after task completion")
logger.Info().Msg("No more active tasks, job is paused")
return sm.JobStatusChange(ctx, job, api.JobStatusPaused, "all tasks completed")
}
} }
logger.Info(). logger.Info().