feat(metrics_collection): add configuration for metrics collection

Author: leo-marble

.env.example

@@ -142,5 +142,9 @@ SENTRY_DSN=
 # OFFLOADING_SAVE_POINTS=1000
 # OFFLOADING_WRITES_PER_SEC=200
 
+# Configure metrics collection.
+# METRICS_COLLECTION_ENABLED=true
+# METRICS_COLLECTION_JOB_INTERVAL=1h
+
 # Abort and exit if license cannot be validated.
 KILL_IF_READ_LICENSE_ERROR=false

cmd/worker.go

@@ -100,6 +100,11 @@ func RunTaskQueue(apiVersion string) error {
 
 	offloadingConfig.ValidateAndFix(ctx)
 
+	metricCollectionConfig := infra.MetricCollectionConfig{
+		Enabled:     utils.GetEnv("METRICS_COLLECTION_ENABLED", true),
+		JobInterval: utils.GetEnvDuration("METRICS_COLLECTION_JOB_INTERVAL", 1*time.Hour),
+	}
+
 	infra.SetupSentry(workerConfig.sentryDsn, workerConfig.env, apiVersion)
 	defer sentry.Flush(3 * time.Second)
 
@@ -157,21 +162,25 @@ func RunTaskQueue(apiVersion string) error {
 		return err
 	}
 
-	// For non-org queues
+	// For non-org
 	nonOrgQueues := make(map[string]river.QueueConfig)
-	maps.Copy(nonOrgQueues, usecases.QueueMetrics())
-	queueWhitelist := slices.Collect(maps.Keys(nonOrgQueues))
+	queueWhitelist := []string{}
+	globalPeriodics := []*river.PeriodicJob{}
+
+	if metricCollectionConfig.Enabled {
+		maps.Copy(nonOrgQueues, usecases.QueueMetrics())
+		queueWhitelist = append(queueWhitelist, slices.Collect(maps.Keys(nonOrgQueues))...)
+		globalPeriodics = append(globalPeriodics,
+			scheduled_execution.NewMetricsCollectionPeriodicJob(metricCollectionConfig))
+	}
 
 	// Add the metrics queue
 	maps.Copy(queues, nonOrgQueues)
 
 	// Periodics always contain the per-org tasks retrieved above. Add other, non-organization-scoped periodics below
 	periodics := append(
 		orgPeriodics,
-		[]*river.PeriodicJob{
-			// Add periodic jobs here
-			scheduled_execution.NewMetricsCollectionPeriodicJob(),
-		}...,
+		globalPeriodics...,
 	)
 
 	riverClient, err = river.NewClient(riverpgxv5.New(pool), &river.Config{
@@ -216,7 +225,9 @@ func RunTaskQueue(apiVersion string) error {
 	if offloadingConfig.Enabled {
 		river.AddWorker(workers, adminUc.NewOffloadingWorker())
 	}
-	river.AddWorker(workers, uc.NewMetricsCollectionWorker())
+	if metricCollectionConfig.Enabled {
+		river.AddWorker(workers, uc.NewMetricsCollectionWorker())
+	}
 
 	if err := riverClient.Start(ctx); err != nil {
 		utils.LogAndReportSentryError(ctx, err)

infra/metrics_collection.go

@@ -0,0 +1,8 @@
+package infra
+
+import "time"
+
+type MetricCollectionConfig struct {
+	Enabled     bool
+	JobInterval time.Duration
+}

usecases/scheduled_execution/metrics_collection_job.go

@@ -5,25 +5,23 @@ import (
 	"encoding/json"
 	"time"
 
+	"github.com/checkmarble/marble-backend/infra"
 	"github.com/checkmarble/marble-backend/models"
 	"github.com/checkmarble/marble-backend/repositories"
 	"github.com/checkmarble/marble-backend/usecases/executor_factory"
 	"github.com/checkmarble/marble-backend/utils"
 	"github.com/riverqueue/river"
 )
 
-// const METRIC_COLLECTION_WORKER_INTERVAL = 24 * time.Hour // Run daily
-const METRICS_COLLECTION_WORKER_INTERVAL = 10 * time.Second // Run every minute for testing
-
-func NewMetricsCollectionPeriodicJob() *river.PeriodicJob {
+func NewMetricsCollectionPeriodicJob(config infra.MetricCollectionConfig) *river.PeriodicJob {
 	return river.NewPeriodicJob(
-		river.PeriodicInterval(METRICS_COLLECTION_WORKER_INTERVAL),
+		river.PeriodicInterval(config.JobInterval),
 		func() (river.JobArgs, *river.InsertOpts) {
 			return models.MetricsCollectionArgs{}, &river.InsertOpts{
 				Queue: "metrics",
 				UniqueOpts: river.UniqueOpts{
 					ByQueue:  true,
-					ByPeriod: METRICS_COLLECTION_WORKER_INTERVAL,
+					ByPeriod: config.JobInterval,
 				},
 			}
 		},