feat(scheduler): add per-pool scheduling outcome metrics (#4591)

<!-- Thanks for sending a pull request! Here are some tips for you: -->

#### What type of PR is this?

Enhancement

#### What this PR does / why we need it

Adds per-pool scheduling metrics to track success/failure outcomes for
each pool independently.

Currently a scheduling failure in one pool causes the entire cycle to
fail with a single error. These metrics enable:
  1. Identifying which pool is failing
  2. Alerting on specific pool failures
  3. Tracking pool health over time

**New metrics:**
- `armada_scheduler_pool_scheduling_outcome` - counter with labels
`pool`, `outcome` (success/failure)

#### Which issue(s) this PR fixes
<!--
*Automatically closes linked issue when PR is merged.
Usage: `Fixes #<issue number>`, or `Fixes (paste link of issue)`.
_If PR is about `failing-tests or flakes`, please post the related
issues/tests in a comment and do not use `Fixes`_*
-->
Fixes #

#### Special notes for your reviewer

Signed-off-by: Dejan Zele Pejchev <pejcev.dejan@gmail.com>

Author: dejanzele

internal/scheduler/metrics/constants.go

@@ -23,6 +23,11 @@ const (
 	reservationLabel         = "reservation"
 	jobShapeLabel            = "job_shape"
 	unschedulableReasonLabel = "unschedulable_reason"
+	outcomeLabel             = "outcome"
+	terminationReasonLabel   = "termination_reason"
+
+	PoolSchedulingOutcomeSuccess = "success"
+	PoolSchedulingOutcomeFailure = "failure"
 
 	// Job state strings
 	queued    = "queued"

internal/scheduler/metrics/cycle_metrics.go

@@ -30,6 +30,7 @@ var (
 	poolAndShapeLabels                     = []string{poolLabel, jobShapeLabel}
 	poolAndShapeAndReasonLabels            = []string{poolLabel, jobShapeLabel, unschedulableReasonLabel}
 	poolQueueAndResourceLabels             = []string{poolLabel, queueLabel, resourceLabel}
+	poolAndOutcomeLabels                   = []string{poolLabel, outcomeLabel, terminationReasonLabel}
 	defaultType                            = "unknown"
 )
 
@@ -356,6 +357,7 @@ type cycleMetrics struct {
 
 	scheduledJobs           *prometheus.CounterVec
 	premptedJobs            *prometheus.CounterVec
+	poolSchedulingOutcome   *prometheus.CounterVec
 	scheduleCycleTime       prometheus.Histogram
 	reconciliationCycleTime prometheus.Histogram
 	latestCycleMetrics      atomic.Pointer[perCycleMetrics]
@@ -395,10 +397,19 @@ func newCycleMetrics(publisher pulsarutils.Publisher[*metricevents.Event]) *cycl
 		},
 	)
 
+	poolSchedulingOutcome := prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Name: prefix + "pool_scheduling_outcome",
+			Help: "Number of scheduling attempts per pool by outcome",
+		},
+		poolAndOutcomeLabels,
+	)
+
 	cycleMetrics := &cycleMetrics{
 		leaderMetricsEnabled:    true,
 		scheduledJobs:           scheduledJobs,
 		premptedJobs:            premptedJobs,
+		poolSchedulingOutcome:   poolSchedulingOutcome,
 		scheduleCycleTime:       scheduleCycleTime,
 		reconciliationCycleTime: reconciliationCycleTime,
 		latestCycleMetrics:      atomic.Pointer[perCycleMetrics]{},
@@ -420,6 +431,7 @@ func (m *cycleMetrics) disableLeaderMetrics() {
 func (m *cycleMetrics) resetLeaderMetrics() {
 	m.premptedJobs.Reset()
 	m.scheduledJobs.Reset()
+	m.poolSchedulingOutcome.Reset()
 	m.latestCycleMetrics.Store(newPerCycleMetrics())
 }
 
@@ -438,6 +450,18 @@ func (m *cycleMetrics) ReportJobPreemptedWithType(job *jobdb.Job, preemptionType
 	m.premptedJobs.WithLabelValues(job.LatestRun().Pool(), job.Queue(), job.PriorityClassName(), string(preemptionType)).Inc()
 }
 
+// ReportPoolSchedulingOutcomes reports outcomes for all pools from the scheduler result
+func (m *cycleMetrics) ReportPoolSchedulingOutcomes(outcomes []scheduling.PoolSchedulingOutcome) {
+	for _, o := range outcomes {
+		terminationReason := string(o.TerminationReason)
+		outcome := PoolSchedulingOutcomeSuccess
+		if !o.Success {
+			outcome = PoolSchedulingOutcomeFailure
+		}
+		m.poolSchedulingOutcome.WithLabelValues(o.Pool, outcome, terminationReason).Inc()
+	}
+}
+
 func (m *cycleMetrics) ReportSchedulerResult(ctx *armadacontext.Context, result scheduling.SchedulerResult) {
 	// Metrics that depend on pool
 	currentCycle := newPerCycleMetrics()
@@ -565,6 +589,7 @@ func (m *cycleMetrics) describe(ch chan<- *prometheus.Desc) {
 	if m.leaderMetricsEnabled {
 		m.scheduledJobs.Describe(ch)
 		m.premptedJobs.Describe(ch)
+		m.poolSchedulingOutcome.Describe(ch)
 		m.scheduleCycleTime.Describe(ch)
 
 		currentCycle := m.latestCycleMetrics.Load()
@@ -608,6 +633,7 @@ func (m *cycleMetrics) collect(ch chan<- prometheus.Metric) {
 	if m.leaderMetricsEnabled {
 		m.scheduledJobs.Collect(ch)
 		m.premptedJobs.Collect(ch)
+		m.poolSchedulingOutcome.Collect(ch)
 		m.scheduleCycleTime.Collect(ch)
 
 		currentCycle := m.latestCycleMetrics.Load()

internal/scheduler/metrics/cycle_metrics_test.go

@@ -311,6 +311,47 @@ func TestPublishCycleMetrics(t *testing.T) {
 	m.publishCycleMetrics(ctx, schedulerResult)
 }
 
+func TestReportPoolSchedulingOutcomes(t *testing.T) {
+	m := newCycleMetrics(pulsarutils.NoOpPublisher[*metricevents.Event]{})
+
+	outcomes := []scheduling.PoolSchedulingOutcome{
+		{
+			Pool:              "pool-1",
+			Success:           false,
+			TerminationReason: scheduling.PoolSchedulingTerminationReasonError,
+		},
+		{
+			Pool:              "pool-2",
+			Success:           true,
+			TerminationReason: scheduling.PoolSchedulingTerminationReasonCompleted,
+		},
+		{
+			Pool:              "pool-3",
+			Success:           true,
+			TerminationReason: scheduling.PoolSchedulingTerminationReasonTimeout,
+		},
+		{
+			Pool:              "pool-4",
+			Success:           true,
+			TerminationReason: scheduling.PoolSchedulingTerminationReasonRateLimit,
+		},
+	}
+
+	m.ReportPoolSchedulingOutcomes(outcomes)
+
+	failureCount := testutil.ToFloat64(m.poolSchedulingOutcome.WithLabelValues("pool-1", PoolSchedulingOutcomeFailure, string(scheduling.PoolSchedulingTerminationReasonError)))
+	assert.Equal(t, 1.0, failureCount, "failure outcome metric does not match")
+
+	completedCount := testutil.ToFloat64(m.poolSchedulingOutcome.WithLabelValues("pool-2", PoolSchedulingOutcomeSuccess, string(scheduling.PoolSchedulingTerminationReasonCompleted)))
+	assert.Equal(t, 1.0, completedCount, "completed outcome metric does not match")
+
+	timeoutCount := testutil.ToFloat64(m.poolSchedulingOutcome.WithLabelValues("pool-3", PoolSchedulingOutcomeSuccess, string(scheduling.PoolSchedulingTerminationReasonTimeout)))
+	assert.Equal(t, 1.0, timeoutCount, "timeout outcome metric does not match")
+
+	rateLimitCount := testutil.ToFloat64(m.poolSchedulingOutcome.WithLabelValues("pool-4", PoolSchedulingOutcomeSuccess, string(scheduling.PoolSchedulingTerminationReasonRateLimit)))
+	assert.Equal(t, 1.0, rateLimitCount, "rate limit outcome metric does not match")
+}
+
 func mustParseResourcePtr(qtyStr string) *resource.Quantity {
 	q := resource.MustParse(qtyStr)
 	return &q

internal/scheduler/scheduler.go

@@ -212,6 +212,9 @@ func (s *Scheduler) Run(ctx *armadacontext.Context) error {
 				if shouldSchedule {
 					previousSchedulingRoundEnd = s.clock.Now()
 				}
+
+				s.metrics.ReportPoolSchedulingOutcomes(result.PoolSchedulingOutcomes)
+
 				if err != nil {
 					ctx.Logger().WithStacktrace(err).Error("scheduling cycle failure")
 					leaderToken = leader.InvalidLeaderToken()
@@ -366,6 +369,8 @@ func (s *Scheduler) cycle(ctx *armadacontext.Context, updateAll bool, leaderToke
 		var result *scheduling.SchedulerResult
 		result, err = s.schedulingAlgo.Schedule(ctx, resourceUnits, txn)
 		if err != nil {
+			// Copy outcomes to the returned result so metrics are reported for pools that succeeded before the failure
+			overallSchedulerResult.PoolSchedulingOutcomes = result.PoolSchedulingOutcomes
 			return overallSchedulerResult, err
 		}
 

internal/scheduler/scheduler_test.go

@@ -1870,7 +1870,7 @@ type testSchedulingAlgo struct {
 func (t *testSchedulingAlgo) Schedule(_ *armadacontext.Context, _ map[string]internaltypes.ResourceList, txn *jobdb.Txn) (*scheduling.SchedulerResult, error) {
 	t.numberOfScheduleCalls++
 	if t.shouldError {
-		return nil, errors.New("error scheduling jobs")
+		return &scheduling.SchedulerResult{}, errors.New("error scheduling jobs")
 	}
 	if t.persisted {
 		// Exit right away if decisions have already been persisted.

internal/scheduler/scheduling/result.go

@@ -1,12 +1,14 @@
 package scheduling
 
 import (
+	"context"
 	"time"
 
 	"github.com/armadaproject/armada/internal/scheduler/internaltypes"
 	"github.com/armadaproject/armada/internal/scheduler/jobdb"
 	"github.com/armadaproject/armada/internal/scheduler/nodedb"
-	"github.com/armadaproject/armada/internal/scheduler/scheduling/context"
+	"github.com/armadaproject/armada/internal/scheduler/scheduling/constraints"
+	schedulercontext "github.com/armadaproject/armada/internal/scheduler/scheduling/context"
 )
 
 type QueueStats struct {
@@ -24,6 +26,35 @@ type QueueStats struct {
 	Time                             time.Duration
 }
 
+type PoolSchedulingTerminationReason string
+
+const (
+	PoolSchedulingTerminationReasonCompleted    PoolSchedulingTerminationReason = "completed"
+	PoolSchedulingTerminationReasonTimeout      PoolSchedulingTerminationReason = "timeout"
+	PoolSchedulingTerminationReasonRateLimit    PoolSchedulingTerminationReason = "rate_limit"
+	PoolSchedulingTerminationReasonMaxResources PoolSchedulingTerminationReason = "max_resources"
+	PoolSchedulingTerminationReasonError        PoolSchedulingTerminationReason = "error"
+)
+
+func terminationReasonFromString(reason string) PoolSchedulingTerminationReason {
+	switch reason {
+	case context.Canceled.Error(), context.DeadlineExceeded.Error():
+		return PoolSchedulingTerminationReasonTimeout
+	case constraints.GlobalRateLimitExceededUnschedulableReason:
+		return PoolSchedulingTerminationReasonRateLimit
+	case constraints.MaximumResourcesScheduledUnschedulableReason:
+		return PoolSchedulingTerminationReasonMaxResources
+	default:
+		return PoolSchedulingTerminationReasonCompleted
+	}
+}
+
+type PoolSchedulingOutcome struct {
+	Pool              string
+	Success           bool
+	TerminationReason PoolSchedulingTerminationReason
+}
+
 type PerPoolSchedulingStats struct {
 	// scheduling stats per queue
 	StatsPerQueue map[string]QueueStats
@@ -36,25 +67,27 @@ type PerPoolSchedulingStats struct {
 	// The nodeDb used in the scheduling round
 	NodeDb *nodedb.NodeDb
 	// The jobs scheduled in this cycle
-	ScheduledJobs []*context.JobSchedulingContext
+	ScheduledJobs []*schedulercontext.JobSchedulingContext
 	// The jobs preempted in this cycle
-	PreemptedJobs []*context.JobSchedulingContext
+	PreemptedJobs []*schedulercontext.JobSchedulingContext
 	// Scheduling summary for gang shapes we're interested in. Prices are determined if the job is deemed schedulable.
 	MarketDrivenIndicativePrices IndicativeGangPricesByJobShape
 }
 
 // SchedulerResult is returned by Rescheduler.Schedule().
 type SchedulerResult struct {
 	// Running jobs that should be preempted.
-	PreemptedJobs []*context.JobSchedulingContext
+	PreemptedJobs []*schedulercontext.JobSchedulingContext
 	// Queued jobs that should be scheduled.
-	ScheduledJobs []*context.JobSchedulingContext
+	ScheduledJobs []*schedulercontext.JobSchedulingContext
 	// Each result may bundle the result of several scheduling decisions.
 	// These are the corresponding scheduling contexts.
 	// TODO: This doesn't seem like the right approach.
-	SchedulingContexts []*context.SchedulingContext
+	SchedulingContexts []*schedulercontext.SchedulingContext
 	// scheduling stats
 	PerPoolSchedulingStats map[string]PerPoolSchedulingStats
+	// Pool scheduling outcomes for metrics reporting
+	PoolSchedulingOutcomes []PoolSchedulingOutcome
 }
 
 // PreemptedJobsFromSchedulerResult returns the slice of preempted jobs in the result.

internal/scheduler/scheduling/scheduling_algo.go

@@ -97,6 +97,10 @@ func NewFairSchedulingAlgo(
 // It iterates over each executor in turn (using lexicographical order) and assigns the jobs using a LegacyScheduler, before moving onto the next executor.
 // It maintains state of which executors it has considered already and may take multiple Schedule() calls to consider all executors if scheduling is slow.
 // Newly leased jobs are updated as such in the jobDb using the transaction provided and are also returned to the caller.
+//
+// This function must always return a non-nil SchedulerResult, even when returning an error.
+// The result contains PoolSchedulingOutcomes that track which pools succeeded or failed,
+// and callers depend on this for metrics reporting.
 func (l *FairSchedulingAlgo) Schedule(
 	ctx *armadacontext.Context,
 	resourceUnits map[string]internaltypes.ResourceList,
@@ -134,7 +138,9 @@ func (l *FairSchedulingAlgo) Schedule(
 
 		fsctx, err := l.newFairSchedulingAlgoContext(ctx, txn, pool)
 		if err != nil {
-			return nil, err
+			overallSchedulerResult.PoolSchedulingOutcomes = append(overallSchedulerResult.PoolSchedulingOutcomes,
+				PoolSchedulingOutcome{Pool: pool.Name, Success: false, TerminationReason: PoolSchedulingTerminationReasonError})
+			return overallSchedulerResult, err
 		}
 
 		if fsctx.nodeDb.NumNodes() <= 0 {
@@ -156,12 +162,15 @@ func (l *FairSchedulingAlgo) Schedule(
 
 		ctx.Infof("Scheduled on executor pool %s in %v with error %v", pool.Name, time.Now().Sub(start), err)
 
-		if errors.Is(err, context.DeadlineExceeded) {
-			// We've reached the scheduling time limit;
+		if errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled) {
+			overallSchedulerResult.PoolSchedulingOutcomes = append(overallSchedulerResult.PoolSchedulingOutcomes,
+				PoolSchedulingOutcome{Pool: pool.Name, Success: true, TerminationReason: PoolSchedulingTerminationReasonTimeout})
 			ctx.Info("stopped scheduling early as we have hit the maximum scheduling duration")
 			break
 		} else if err != nil {
-			return nil, err
+			overallSchedulerResult.PoolSchedulingOutcomes = append(overallSchedulerResult.PoolSchedulingOutcomes,
+				PoolSchedulingOutcome{Pool: pool.Name, Success: false, TerminationReason: PoolSchedulingTerminationReasonError})
+			return overallSchedulerResult, err
 		}
 		if l.schedulingContextRepository != nil {
 			l.schedulingContextRepository.StoreSchedulingContext(sctx)
@@ -170,13 +179,21 @@ func (l *FairSchedulingAlgo) Schedule(
 		preemptedJobs := PreemptedJobsFromSchedulerResult(schedulerResult)
 		scheduledJobs := ScheduledJobsFromSchedulerResult(schedulerResult)
 
+		terminationReason := terminationReasonFromString(sctx.TerminationReason)
 		if err := txn.Upsert(preemptedJobs); err != nil {
-			return nil, err
+			overallSchedulerResult.PoolSchedulingOutcomes = append(overallSchedulerResult.PoolSchedulingOutcomes,
+				PoolSchedulingOutcome{Pool: pool.Name, Success: false, TerminationReason: PoolSchedulingTerminationReasonError})
+			return overallSchedulerResult, err
 		}
 		if err := txn.Upsert(scheduledJobs); err != nil {
-			return nil, err
+			overallSchedulerResult.PoolSchedulingOutcomes = append(overallSchedulerResult.PoolSchedulingOutcomes,
+				PoolSchedulingOutcome{Pool: pool.Name, Success: false, TerminationReason: PoolSchedulingTerminationReasonError})
+			return overallSchedulerResult, err
 		}
 
+		overallSchedulerResult.PoolSchedulingOutcomes = append(overallSchedulerResult.PoolSchedulingOutcomes,
+			PoolSchedulingOutcome{Pool: pool.Name, Success: true, TerminationReason: terminationReason})
+
 		// Aggregate changes across executors.
 		overallSchedulerResult.PreemptedJobs = append(overallSchedulerResult.PreemptedJobs, schedulerResult.PreemptedJobs...)
 		overallSchedulerResult.ScheduledJobs = append(overallSchedulerResult.ScheduledJobs, schedulerResult.ScheduledJobs...)