fix(v2): scheduler metrics collector panics (#3990)

* fix(v2): scheduler metrics collector panics

* fix(v2): add scheduler metrics collector test

Author: kolesnikovae

pkg/experiment/metastore/compaction/scheduler/metrics.go

@@ -81,6 +81,10 @@ func (c *statsCollector) collectStats(fn func(level int, stats queueStats)) {
 	defer c.s.mu.Unlock()
 
 	for i, q := range c.s.queue.levels {
+		// Note that some levels may be empty.
+		if q == nil || q.jobs == nil {
+			continue
+		}
 		var stats queueStats
 		for _, e := range *q.jobs {
 			switch {

pkg/experiment/metastore/compaction/scheduler/metrics_test.go

@@ -1,10 +1,15 @@
 package scheduler
 
 import (
+	"bytes"
+	"os"
 	"testing"
 	"time"
 
 	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/testutil"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 
 	metastorev1 "github.com/grafana/pyroscope/api/gen/proto/go/metastore/v1"
 	"github.com/grafana/pyroscope/api/gen/proto/go/metastore/v1/raft_log"
@@ -27,3 +32,26 @@ func TestCollectorRegistration(t *testing.T) {
 		sc.queue.delete("a")
 	}
 }
+
+func TestCollectorCollect(t *testing.T) {
+	reg := prometheus.NewRegistry()
+	config := Config{
+		MaxFailures:   5,
+		LeaseDuration: 15 * time.Second,
+	}
+
+	sc := NewScheduler(config, nil, reg)
+	sc.queue.put(&raft_log.CompactionJobState{
+		Name: "a", CompactionLevel: 0, Token: 1,
+		Status: metastorev1.CompactionJobStatus_COMPACTION_STATUS_UNSPECIFIED,
+	})
+	sc.queue.put(&raft_log.CompactionJobState{
+		Name: "b", CompactionLevel: 2, Token: 1,
+		Status: metastorev1.CompactionJobStatus_COMPACTION_STATUS_UNSPECIFIED,
+	})
+	sc.queue.delete("a")
+
+	buf, err := os.ReadFile("testdata/metrics.txt")
+	require.NoError(t, err)
+	assert.NoError(t, testutil.GatherAndCompare(reg, bytes.NewReader(buf)))
+}

pkg/experiment/metastore/compaction/scheduler/scheduler_queue.go

@@ -9,7 +9,8 @@ import (
 )
 
 type schedulerQueue struct {
-	jobs   map[string]*jobEntry
+	jobs map[string]*jobEntry
+	// Sparse array of job queues, indexed by compaction level.
 	levels []*jobQueue
 }
 

pkg/experiment/metastore/compaction/scheduler/testdata/metrics.txt

@@ -0,0 +1,30 @@
+# HELP compaction_scheduler_queue_added_jobs_total The total number of jobs added to the queue.
+# TYPE compaction_scheduler_queue_added_jobs_total counter
+compaction_scheduler_queue_added_jobs_total{level="0"} 1
+compaction_scheduler_queue_added_jobs_total{level="2"} 1
+# HELP compaction_scheduler_queue_assigned_jobs_total The total number of jobs assigned.
+# TYPE compaction_scheduler_queue_assigned_jobs_total counter
+compaction_scheduler_queue_assigned_jobs_total{level="0"} 0
+compaction_scheduler_queue_assigned_jobs_total{level="2"} 0
+# HELP compaction_scheduler_queue_completed_jobs_total The total number of jobs completed.
+# TYPE compaction_scheduler_queue_completed_jobs_total counter
+compaction_scheduler_queue_completed_jobs_total{level="0"} 1
+compaction_scheduler_queue_completed_jobs_total{level="2"} 0
+# HELP compaction_scheduler_queue_evicted_jobs_total The total number of jobs evicted.
+# TYPE compaction_scheduler_queue_evicted_jobs_total counter
+compaction_scheduler_queue_evicted_jobs_total{level="0"} 0
+compaction_scheduler_queue_evicted_jobs_total{level="2"} 0
+# HELP compaction_scheduler_queue_jobs The total number of jobs in the queue.
+# TYPE compaction_scheduler_queue_jobs gauge
+compaction_scheduler_queue_jobs{level="0",status="assigned"} 0
+compaction_scheduler_queue_jobs{level="0",status="failed"} 0
+compaction_scheduler_queue_jobs{level="0",status="reassigned"} 0
+compaction_scheduler_queue_jobs{level="0",status="unassigned"} 0
+compaction_scheduler_queue_jobs{level="2",status="assigned"} 0
+compaction_scheduler_queue_jobs{level="2",status="failed"} 0
+compaction_scheduler_queue_jobs{level="2",status="reassigned"} 0
+compaction_scheduler_queue_jobs{level="2",status="unassigned"} 1
+# HELP compaction_scheduler_queue_reassigned_jobs_total The total number of jobs reassigned.
+# TYPE compaction_scheduler_queue_reassigned_jobs_total counter
+compaction_scheduler_queue_reassigned_jobs_total{level="0"} 0
+compaction_scheduler_queue_reassigned_jobs_total{level="2"} 0