add job Queued state (#4121)

Improves visibility of queued jobs by adding a dedicated `Queued` state

pkg/models/job.go

@@ -22,6 +22,10 @@ const (
 	// yet scheduled.
 	JobStateTypePending
 
+	// JobStateTypeQueued is the state of a job that has been evaluated but no
+	// matching nodes are available yet.
+	JobStateTypeQueued
+
 	// JobStateTypeRunning is the state of a job that has been scheduled, with at
 	// least one active execution.
 	JobStateTypeRunning

pkg/models/plan.go

@@ -96,6 +96,12 @@ func (p *Plan) MarkJobRunningIfEligible() {
 	p.DesiredJobState = JobStateTypeRunning
 }
 
+// MarkJobQueued marks the job as pending.
+func (p *Plan) MarkJobQueued(event Event) {
+	p.DesiredJobState = JobStateTypeQueued
+	p.Event = event
+}
+
 func (p *Plan) MarkJobFailed(event Event) {
 	p.DesiredJobState = JobStateTypeFailed
 	p.Event = event

pkg/orchestrator/events.go

@@ -10,13 +10,15 @@ import (
 const (
 	EventTopicJobSubmission    models.EventTopic = "Submission"
 	EventTopicJobScheduling    models.EventTopic = "Scheduling"
+	EventTopicJobQueueing      models.EventTopic = "Queueing"
 	EventTopicExecutionTimeout models.EventTopic = "Exec Timeout"
 	EventTopicJobTimeout       models.EventTopic = "Job Timeout"
 )
 
 const (
 	jobSubmittedMessage        = "Job submitted"
 	jobTranslatedMessage       = "Job tasks translated to new type"
+	jobQueuedMessage           = "Job queued"
 	jobStopRequestedMessage    = "Job requested to stop before completion"
 	jobExhaustedRetriesMessage = "Job failed because it has been retried too many times"
 	JobTimeoutMessage          = "Job timed out"
@@ -70,6 +72,14 @@ func JobTimeoutEvent(timeout time.Duration) models.Event {
 	return *e
 }
 
+func JobQueueingEvent(reason string) models.Event {
+	message := jobQueuedMessage
+	if reason != "" {
+		message = fmt.Sprintf("%s. %s", message, reason)
+	}
+	return *models.NewEvent(EventTopicJobQueueing).WithMessage(message)
+}
+
 func ExecStoppedByJobStopEvent() models.Event {
 	return event(EventTopicJobScheduling, execStoppedByJobStopMessage, map[string]string{})
 }

pkg/orchestrator/scheduler/batch_service_job.go

@@ -224,6 +224,16 @@ func (b *BatchServiceJobScheduler) createMissingExecs(
 		plan.AppendEvaluation(delayedEvaluation)
 		log.Ctx(ctx).Debug().Msgf("Creating delayed evaluation %s to retry scheduling job %s in %s due to: %s",
 			delayedEvaluation.ID, job.ID, waitUntil.Sub(b.clock.Now()), comment)
+
+		// if not a single node was matched, then the job if fully queued and we should reflect that
+		// in the job state and events
+		if len(matching) == 0 {
+			// only update the state if the is running, or pending and triggered by job registration
+			if job.State.StateType == models.JobStateTypeRunning ||
+				plan.Eval.TriggeredBy == models.EvalTriggerJobRegister {
+				plan.MarkJobQueued(orchestrator.JobQueueingEvent(comment))
+			}
+		}
 	}
 	return nil
 }