Add config and dependency injection to scheduler metrics

config/scheduler/config.yaml

@@ -9,6 +9,15 @@ internedStringsCacheSize: 100000
 metrics:
   port: 9000
   refreshInterval: 30s
+  schedulerMetrics:
+    scheduleCycleTime:
+      start: 1.0
+      factor: 1.1
+      count: 110
+    reconcileCycleTime:
+      start: 1.0
+      factor: 1.1
+      count: 110
 pulsar:
   URL: "pulsar://pulsar:6650"
   jobsetEventsTopic: "events"

internal/armada/configuration/types.go

@@ -329,6 +329,18 @@ type MetricsConfig struct {
 	Port                    uint16
 	RefreshInterval         time.Duration
 	ExposeSchedulingMetrics bool
+	SchedulerMetrics        SchedulerMetricsConfig
+}
+
+type SchedulerMetricsConfig struct {
+	ScheduleCycleTime  HistogramConfig
+	ReconcileCycleTime HistogramConfig
+}
+
+type HistogramConfig struct {
+	Start  float64
+	Factor float64
+	Count  int
 }
 
 type EventApiConfig struct {

internal/scheduler/scheduler.go

@@ -90,6 +90,7 @@ func NewScheduler(
 	executorTimeout time.Duration,
 	maxAttemptedRuns uint,
 	nodeIdLabel string,
+	schedulerMetrics *SchedulerMetrics,
 ) (*Scheduler, error) {
 	jobDb := jobdb.NewJobDb()
 	return &Scheduler{
@@ -110,7 +111,7 @@ func NewScheduler(
 		nodeIdLabel:                nodeIdLabel,
 		jobsSerial:                 -1,
 		runsSerial:                 -1,
-		metrics:                    GetSchedulerMetrics(),
+		metrics:                    schedulerMetrics,
 	}, nil
 }
 
@@ -175,10 +176,10 @@ func (s *Scheduler) Run(ctx context.Context) error {
 
 			if shouldSchedule && leaderToken.leader {
 				// Only the leader token does real scheduling rounds.
-				s.metrics.ReportScheduleCycleTime(float64(cycleTime.Milliseconds()))
+				s.metrics.ReportScheduleCycleTime(cycleTime)
 				log.Infof("scheduling cycle completed in %s", cycleTime)
 			} else {
-				s.metrics.ReportReconcileCycleTime(float64(cycleTime.Milliseconds()))
+				s.metrics.ReportReconcileCycleTime(cycleTime)
 				log.Infof("reconciliation cycle completed in %s", cycleTime)
 			}
 

internal/scheduler/scheduler_metrics.go

@@ -1,9 +1,12 @@
 package scheduler
 
 import (
+	"time"
+
 	"github.com/prometheus/client_golang/prometheus"
 	log "github.com/sirupsen/logrus"
 
+	"github.com/armadaproject/armada/internal/armada/configuration"
 	"github.com/armadaproject/armada/internal/scheduler/interfaces"
 )
 
@@ -23,24 +26,17 @@ type SchedulerMetrics struct {
 	preemptedJobsPerQueue prometheus.GaugeVec
 }
 
-var schedulerMetrics *SchedulerMetrics
-
-func init() {
-	schedulerMetrics = newSchedulerMetrics()
-}
-
-func GetSchedulerMetrics() *SchedulerMetrics {
-	return schedulerMetrics
-}
-
-func newSchedulerMetrics() *SchedulerMetrics {
+func NewSchedulerMetrics(config configuration.SchedulerMetricsConfig) *SchedulerMetrics {
 	scheduleCycleTime := prometheus.NewHistogram(
 		prometheus.HistogramOpts{
 			Namespace: NAMESPACE,
 			Subsystem: SUBSYSTEM,
 			Name:      "schedule_cycle_times",
 			Help:      "Cycle time when in a scheduling round.",
-			Buckets:   prometheus.LinearBuckets(0, 5, 20),
+			Buckets: prometheus.ExponentialBuckets(
+				config.ScheduleCycleTime.Start,
+				config.ScheduleCycleTime.Factor,
+				config.ScheduleCycleTime.Count),
 		},
 	)
 
@@ -50,7 +46,10 @@ func newSchedulerMetrics() *SchedulerMetrics {
 			Subsystem: SUBSYSTEM,
 			Name:      "reconcile_cycle_times",
 			Help:      "Cycle time when outside of a scheduling round.",
-			Buckets:   prometheus.LinearBuckets(0, 5, 20),
+			Buckets: prometheus.ExponentialBuckets(
+				config.ReconcileCycleTime.Start,
+				config.ReconcileCycleTime.Factor,
+				config.ReconcileCycleTime.Count),
 		},
 	)
 
@@ -93,12 +92,12 @@ func newSchedulerMetrics() *SchedulerMetrics {
 	}
 }
 
-func (metrics *SchedulerMetrics) ReportScheduleCycleTime(cycleTime float64) {
-	metrics.scheduleCycleTime.Observe(cycleTime)
+func (metrics *SchedulerMetrics) ReportScheduleCycleTime(cycleTime time.Duration) {
+	metrics.scheduleCycleTime.Observe(float64(cycleTime.Milliseconds()))
 }
 
-func (metrics *SchedulerMetrics) ReportReconcileCycleTime(cycleTime float64) {
-	metrics.reconcileCycleTime.Observe(cycleTime)
+func (metrics *SchedulerMetrics) ReportReconcileCycleTime(cycleTime time.Duration) {
+	metrics.reconcileCycleTime.Observe(float64(cycleTime.Milliseconds()))
 }
 
 func (metrics *SchedulerMetrics) ReportSchedulerResult(result *SchedulerResult) {

internal/scheduler/scheduler_test.go

@@ -76,6 +76,18 @@ var (
 		Version: 2,
 	}
 	updatedSchedulingInfoBytes = protoutil.MustMarshall(updatedSchedulingInfo)
+	schedulerMetrics           = NewSchedulerMetrics(configuration.SchedulerMetricsConfig{
+		ScheduleCycleTime: configuration.HistogramConfig{
+			Start:  1,
+			Factor: 1.1,
+			Count:  100,
+		},
+		ReconcileCycleTime: configuration.HistogramConfig{
+			Start:  1,
+			Factor: 1.1,
+			Count:  100,
+		},
+	})
 )
 
 var queuedJob = jobdb.NewJob(
@@ -502,6 +514,7 @@ func TestScheduler_TestCycle(t *testing.T) {
 				clusterTimeout,
 				maxNumberOfAttempts,
 				nodeIdLabel,
+				schedulerMetrics,
 			)
 			require.NoError(t, err)
 
@@ -665,7 +678,8 @@ func TestRun(t *testing.T) {
 		15*time.Second,
 		1*time.Hour,
 		maxNumberOfAttempts,
-		nodeIdLabel)
+		nodeIdLabel,
+		schedulerMetrics)
 	require.NoError(t, err)
 
 	sched.clock = testClock
@@ -874,7 +888,8 @@ func TestScheduler_TestSyncState(t *testing.T) {
 				5*time.Second,
 				1*time.Hour,
 				maxNumberOfAttempts,
-				nodeIdLabel)
+				nodeIdLabel,
+				schedulerMetrics)
 			require.NoError(t, err)
 
 			// insert initial jobs

internal/scheduler/schedulerapp.go

@@ -206,6 +206,7 @@ func Run(config schedulerconfig.Configuration) error {
 		config.ExecutorTimeout,
 		config.Scheduling.MaxRetries+1,
 		config.Scheduling.Preemption.NodeIdLabel,
+		NewSchedulerMetrics(config.Metrics.SchedulerMetrics),
 	)
 	if err != nil {
 		return errors.WithMessage(err, "error creating scheduler")