Provisioning: Allow setting concurrent jobs settings and add metric (#111451)

pkg/operators/provisioning/jobs_operator.go

@@ -112,11 +112,11 @@ func RunJobController(deps server.OperatorDependencies) error {
 
 	// This is basically our own JobQueue system
 	driver, err := jobs.NewConcurrentJobDriver(
-		3,              // 3 drivers for now
-		20*time.Minute, // Max time for each job
-		time.Minute,    // Cleanup jobs
-		30*time.Second, // Periodically look for new jobs
-		30*time.Second, // Lease renewal interval
+		controllerCfg.concurrentDrivers,
+		controllerCfg.maxJobTimeout,
+		controllerCfg.cleanupInterval,
+		controllerCfg.jobInterval,
+		controllerCfg.leaseRenewalInterval,
 		jobStore,
 		repoGetter,
 		jobHistoryWriter,
@@ -150,7 +150,12 @@ func RunJobController(deps server.OperatorDependencies) error {
 
 type jobsControllerConfig struct {
 	provisioningControllerConfig
-	historyExpiration time.Duration
+	historyExpiration    time.Duration
+	maxJobTimeout        time.Duration
+	cleanupInterval      time.Duration
+	jobInterval          time.Duration
+	leaseRenewalInterval time.Duration
+	concurrentDrivers    int
 }
 
 func setupJobsControllerFromConfig(cfg *setting.Cfg, registry prometheus.Registerer) (*jobsControllerConfig, error) {
@@ -162,6 +167,11 @@ func setupJobsControllerFromConfig(cfg *setting.Cfg, registry prometheus.Registe
 	return &jobsControllerConfig{
 		provisioningControllerConfig: *controllerCfg,
 		historyExpiration:            cfg.SectionWithEnvOverrides("operator").Key("history_expiration").MustDuration(0),
+		concurrentDrivers:            cfg.SectionWithEnvOverrides("operator").Key("concurrent_drivers").MustInt(3),
+		maxJobTimeout:                cfg.SectionWithEnvOverrides("operator").Key("max_job_timeout").MustDuration(20 * time.Minute),
+		cleanupInterval:              cfg.SectionWithEnvOverrides("operator").Key("cleanup_interval").MustDuration(time.Minute),
+		jobInterval:                  cfg.SectionWithEnvOverrides("operator").Key("job_interval").MustDuration(30 * time.Second),
+		leaseRenewalInterval:         cfg.SectionWithEnvOverrides("operator").Key("lease_renewal_interval").MustDuration(30 * time.Second),
 	}, nil
 }
 

pkg/registry/apis/provisioning/jobs/concurrent_driver.go

@@ -21,7 +21,6 @@ type ConcurrentJobDriver struct {
 	repoGetter           RepoGetter
 	historicJobs         HistoryWriter
 	workers              []Worker
-	registry             prometheus.Registerer
 	notifications        chan struct{}
 }
 
@@ -58,6 +57,8 @@ func NewConcurrentJobDriver(
 		cleanupInterval = 5 * time.Minute // Maximum cleanup interval
 	}
 
+	recordConcurrentDriverMetric(registry, numDrivers)
+
 	return &ConcurrentJobDriver{
 		numDrivers:           numDrivers,
 		jobTimeout:           jobTimeout,
@@ -69,7 +70,6 @@ func NewConcurrentJobDriver(
 		historicJobs:         historicJobs,
 		workers:              workers,
 		notifications:        notifications,
-		registry:             registry,
 	}, nil
 }
 

pkg/registry/apis/provisioning/jobs/metrics.go

@@ -110,3 +110,15 @@ func getResourceCountBucket(count int) string {
 		return "1000+"
 	}
 }
+
+func recordConcurrentDriverMetric(registry prometheus.Registerer, numDrivers int) {
+	concurrentDriver := prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Name: "grafana_provisioning_jobs_concurrent_driver_num_drivers",
+			Help: "Number of concurrent job drivers",
+		},
+		[]string{},
+	)
+	registry.MustRegister(concurrentDriver)
+	concurrentDriver.WithLabelValues().Set(float64(numDrivers))
+}