From 25dbd15b647bbd4aa006d695a3daadddb1b7c298 Mon Sep 17 00:00:00 2001 From: JamesMurkin Date: Thu, 27 Aug 2026 11:35:26 +0100 Subject: [PATCH] Create a new metric for the main scheduler loop New metric - 'armada_scheduler_main_loop_cycle_time' - Labels type (reconciliation/scheduling) and outcome (success/failure) This will supersede existing metrics: - armada_scheduler_schedule_loop_cycle_times - armada_scheduler_schedule_loop_outcome - armada_scheduler_reconciliation_loop_cycle_times The reasons to do this are: - Simpler in code - No risk of the metric being inconsistent with itself. I.e if we set cycle time but not outcome, they could get out sync - Allow more flexible querying - Can now see duration of failing cycles separately to successful ones - Can get an average of all cycles regardless of type, without needing to manaully merge 2 metrics - Less confusing naming. armada_scheduler_schedule_* and armada_scheduler_scheduling_* mean different things but are very similar. Now the main loop is explicitly named that in the metrics, avoiding confusion with the scheduling duration Signed-off-by: JamesMurkin --- internal/scheduler/metrics/cycle_metrics.go | 29 +++++++++++++++ .../scheduler/metrics/cycle_metrics_test.go | 37 ++++++++++++++----- internal/scheduler/scheduler.go | 3 ++ 3 files changed, 59 insertions(+), 10 deletions(-) diff --git a/internal/scheduler/metrics/cycle_metrics.go b/internal/scheduler/metrics/cycle_metrics.go index 4a2e5899e10..01dc8d38ab9 100644 --- a/internal/scheduler/metrics/cycle_metrics.go +++ b/internal/scheduler/metrics/cycle_metrics.go @@ -31,11 +31,19 @@ var ( poolAndShapeAndReasonLabels = []string{poolLabel, jobShapeLabel, unschedulableReasonLabel} poolQueueAndResourceLabels = []string{poolLabel, queueLabel, resourceLabel} poolAndOutcomeLabels = []string{poolLabel, outcomeLabel, terminationReasonLabel} + loopTypeAndOutcomeLabels = []string{typeLabel, outcomeLabel} nodeLabels = []string{poolLabel, nodeLabel, clusterLabel, nodeTypeLabel, resourceLabel, reservationLabel, schedulableLabel, overAllocatedLabel, physicalPoolLabel, capacityClassLabel, scalableUnitLabel} defaultType = "unknown" reconcilerFailureType = "reconciler" ) +type LoopType string + +const ( + Reconciliation LoopType = "reconciliation" + Scheduling LoopType = "scheduling" +) + type perCycleMetrics struct { consideredJobs *prometheus.GaugeVec fairShare *prometheus.GaugeVec @@ -386,6 +394,7 @@ type cycleMetrics struct { schedulingDuration prometheus.Histogram scheduleCycleOutcome *prometheus.CounterVec scheduleCycleTime prometheus.Histogram + mainLoopCycleTime *prometheus.HistogramVec reconciliationCycleTime prometheus.Histogram submitCheckDuration *prometheus.HistogramVec latestCycleMetrics atomic.Pointer[perCycleMetrics] @@ -433,6 +442,15 @@ func newCycleMetrics(publisher pulsarutils.Publisher[*metricevents.Event], scala []string{outcomeLabel}, ) + mainLoopCycleTime := prometheus.NewHistogramVec( + prometheus.HistogramOpts{ + Name: ArmadaSchedulerMetricsPrefix + "main_loop_cycle_time", + Help: "Time taken for a main loop iteration, by loop type and outcome, in milliseconds.", + Buckets: prometheus.ExponentialBuckets(10.0, 1.1, 110), + }, + loopTypeAndOutcomeLabels, + ) + reconciliationCycleTime := prometheus.NewHistogram( prometheus.HistogramOpts{ Name: ArmadaSchedulerMetricsPrefix + "reconciliation_cycle_times", @@ -486,6 +504,7 @@ func newCycleMetrics(publisher pulsarutils.Publisher[*metricevents.Event], scala schedulingDuration: schedulingDuration, scheduleCycleTime: scheduleCycleTime, scheduleCycleOutcome: scheduleCycleOutcome, + mainLoopCycleTime: mainLoopCycleTime, reconciliationCycleTime: reconciliationCycleTime, submitCheckDuration: submitCheckDuration, latestCycleMetrics: atomic.Pointer[perCycleMetrics]{}, @@ -512,6 +531,14 @@ func (m *cycleMetrics) resetLeaderMetrics() { m.latestCycleMetrics.Store(newPerCycleMetrics()) } +func (m *cycleMetrics) ReportMainLoopCycleCompleted(cycleTime time.Duration, success bool, loopType LoopType) { + result := SchedulingOutcomeSuccess + if !success { + result = SchedulingOutcomeFailure + } + m.mainLoopCycleTime.WithLabelValues(string(loopType), result).Observe(float64(cycleTime.Milliseconds())) +} + func (m *cycleMetrics) ReportScheduleCycleTime(cycleTime time.Duration) { m.scheduleCycleTime.Observe(float64(cycleTime.Milliseconds())) } @@ -769,6 +796,7 @@ func (m *cycleMetrics) describe(ch chan<- *prometheus.Desc) { cycleMetrics.nodePoolSize.Describe(ch) } + m.mainLoopCycleTime.Describe(ch) m.reconciliationCycleTime.Describe(ch) } @@ -820,6 +848,7 @@ func (m *cycleMetrics) collect(ch chan<- prometheus.Metric) { currentCycle.nodePoolSize.Collect(ch) } + m.mainLoopCycleTime.Collect(ch) m.reconciliationCycleTime.Collect(ch) } diff --git a/internal/scheduler/metrics/cycle_metrics_test.go b/internal/scheduler/metrics/cycle_metrics_test.go index f80a11833a4..43e7642aeb8 100644 --- a/internal/scheduler/metrics/cycle_metrics_test.go +++ b/internal/scheduler/metrics/cycle_metrics_test.go @@ -142,20 +142,36 @@ func TestReportSubmitCheckDuration(t *testing.T) { "queue3": 723 * time.Microsecond, }) - assertHistogramObservation(t, m.submitCheckDuration, "queue1", 1, 250.0) - assertHistogramObservation(t, m.submitCheckDuration, "queue2", 1, 50.0) - assertHistogramObservation(t, m.submitCheckDuration, "queue3", 1, 0.723) + assertHistogramObservation(t, m.submitCheckDuration, 1, 250.0, "queue1") + assertHistogramObservation(t, m.submitCheckDuration, 1, 50.0, "queue2") + assertHistogramObservation(t, m.submitCheckDuration, 1, 0.723, "queue3") m.ReportSubmitCheckDuration(map[string]time.Duration{ "queue1": 100 * time.Millisecond, }) - assertHistogramObservation(t, m.submitCheckDuration, "queue1", 2, 350.0) - assertHistogramObservation(t, m.submitCheckDuration, "queue2", 1, 50.0) + assertHistogramObservation(t, m.submitCheckDuration, 2, 350.0, "queue1") + assertHistogramObservation(t, m.submitCheckDuration, 1, 50.0, "queue2") } -func assertHistogramObservation(t *testing.T, vec *prometheus.HistogramVec, queue string, wantCount uint64, wantSumMillis float64) { +func TestReportMainLoopCycle(t *testing.T) { + m := newCycleMetrics(pulsarutils.NoOpPublisher[*metricevents.Event]{}, "") + + m.ReportMainLoopCycleCompleted(100*time.Millisecond, true, Scheduling) + m.ReportMainLoopCycleCompleted(300*time.Millisecond, true, Scheduling) + m.ReportMainLoopCycleCompleted(70*time.Millisecond, false, Scheduling) + m.ReportMainLoopCycleCompleted(20*time.Millisecond, true, Reconciliation) + m.ReportMainLoopCycleCompleted(50*time.Millisecond, false, Reconciliation) + + // Each loop type / outcome combination is recorded independently. + assertHistogramObservation(t, m.mainLoopCycleTime, 2, 400.0, string(Scheduling), SchedulingOutcomeSuccess) + assertHistogramObservation(t, m.mainLoopCycleTime, 1, 70.0, string(Scheduling), SchedulingOutcomeFailure) + assertHistogramObservation(t, m.mainLoopCycleTime, 1, 20.0, string(Reconciliation), SchedulingOutcomeSuccess) + assertHistogramObservation(t, m.mainLoopCycleTime, 1, 50.0, string(Reconciliation), SchedulingOutcomeFailure) +} + +func assertHistogramObservation(t *testing.T, vec *prometheus.HistogramVec, wantCount uint64, wantSumMillis float64, labelValues ...string) { t.Helper() - obs, err := vec.GetMetricWithLabelValues(queue) + obs, err := vec.GetMetricWithLabelValues(labelValues...) require.NoError(t, err) metric := &dto.Metric{} require.NoError(t, obs.(prometheus.Metric).Write(metric)) @@ -248,6 +264,7 @@ func TestDisableLeaderMetrics(t *testing.T) { m.scheduleCycleTime.Observe(float64(1000)) m.scheduleCycleOutcome.WithLabelValues(SchedulingOutcomeSuccess) m.reconciliationCycleTime.Observe(float64(1000)) + m.ReportMainLoopCycleCompleted(20*time.Millisecond, true, Reconciliation) m.poolSchedulingCycleTime.WithLabelValues("pool1").Observe(float64(1000)) m.poolSchedulingOutcome.WithLabelValues("pool1", SchedulingOutcomeSuccess, "reason") m.latestCycleMetrics.Load().gangsConsidered.WithLabelValues("pool1", "queue1").Inc() @@ -272,15 +289,15 @@ func TestDisableLeaderMetrics(t *testing.T) { } // Enabled - assert.True(t, len(collect(m)) > 1) + assert.True(t, len(collect(m)) > 2) // Disabled m.disableLeaderMetrics() - assert.Equal(t, 1, len(collect(m))) + assert.Equal(t, 2, len(collect(m))) // Enabled m.enableLeaderMetrics() - assert.True(t, len(collect(m)) > 1) + assert.True(t, len(collect(m)) > 2) } func TestPublishCycleMetrics(t *testing.T) { diff --git a/internal/scheduler/scheduler.go b/internal/scheduler/scheduler.go index 770cb2aeb90..c9412c7377e 100644 --- a/internal/scheduler/scheduler.go +++ b/internal/scheduler/scheduler.go @@ -249,8 +249,10 @@ func (s *Scheduler) Run(ctx *armadacontext.Context) error { schedulingAttempted, err := s.cycle(ctx, fullUpdate, leaderToken, shouldGetSchedulerResult, cycleNumber) cycleTime := s.clock.Since(start) + loopType := metrics.Reconciliation if schedulingAttempted { + loopType = metrics.Scheduling // Only the leader does real scheduling rounds. s.metrics.ReportScheduleCycleTime(cycleTime) s.metrics.ReportScheduleCycleOutcome(err == nil) @@ -259,6 +261,7 @@ func (s *Scheduler) Run(ctx *armadacontext.Context) error { s.metrics.ReportReconcileCycleTime(cycleTime) ctx.Infof("reconciliation cycle completed in %s", cycleTime) } + s.metrics.ReportMainLoopCycleCompleted(cycleTime, err == nil, loopType) if err != nil { // If there is an error, we can't guarantee that the scheduler-internal state is consistent