Introduce additional metastore compaction metrics (#4625)

* Add a few more metastore compaction metrics

* Update metric description

* Improve misconfiguration handling, variable names

* Rework stats collection approach, to avoid data races

* Bring back level label

* Fix backlog jobs calculation

* Calculate backlog jobs from batches

Author: aleks-p

pkg/metastore/compaction/compactor/compaction_queue.go

@@ -23,9 +23,11 @@ type compactionKey struct {
 }
 
 type compactionQueue struct {
-	config     Config
-	registerer prometheus.Registerer
-	levels     []*blockQueue
+	config         Config
+	registerer     prometheus.Registerer
+	levels         []*blockQueue
+	globalStats    *globalQueueStats
+	statsCollector *globalQueueStatsCollector
 }
 
 // blockQueue stages blocks as they are being added. Once a batch of blocks
@@ -40,9 +42,10 @@ type compactionQueue struct {
 // the queue is through explicit removal. Batch and block iterators provide
 // the read access.
 type blockQueue struct {
-	config     Config
-	registerer prometheus.Registerer
-	staged     map[compactionKey]*stagedBlocks
+	config      Config
+	registerer  prometheus.Registerer
+	staged      map[compactionKey]*stagedBlocks
+	globalStats *globalQueueStats
 	// Batches ordered by arrival.
 	head, tail *batch
 	// Priority queue by last update: we need to flush
@@ -103,10 +106,17 @@ type batch struct {
 }
 
 func newCompactionQueue(config Config, registerer prometheus.Registerer) *compactionQueue {
-	return &compactionQueue{
-		config:     config,
-		registerer: registerer,
+	globalStats := newGlobalQueueStats(len(config.Levels))
+	q := &compactionQueue{
+		config:      config,
+		registerer:  registerer,
+		globalStats: globalStats,
 	}
+	if registerer != nil {
+		q.statsCollector = newGlobalQueueStatsCollector(q)
+		util.RegisterOrGet(registerer, q.statsCollector)
+	}
+	return q
 }
 
 func (q *compactionQueue) reset() {
@@ -145,18 +155,19 @@ func (q *compactionQueue) blockQueue(l uint32) *blockQueue {
 	}
 	level := q.levels[l]
 	if level == nil {
-		level = newBlockQueue(q.config, q.registerer)
+		level = newBlockQueue(q.config, q.registerer, q.globalStats)
 		q.levels[l] = level
 	}
 	return level
 }
 
-func newBlockQueue(config Config, registerer prometheus.Registerer) *blockQueue {
+func newBlockQueue(config Config, registerer prometheus.Registerer, globalStats *globalQueueStats) *blockQueue {
 	return &blockQueue{
-		config:     config,
-		registerer: registerer,
-		staged:     make(map[compactionKey]*stagedBlocks),
-		updates:    new(priorityBlockQueue),
+		config:      config,
+		registerer:  registerer,
+		staged:      make(map[compactionKey]*stagedBlocks),
+		globalStats: globalStats,
+		updates:     new(priorityBlockQueue),
 	}
 }
 
@@ -172,6 +183,7 @@ func (q *blockQueue) stagedBlocks(k compactionKey) *stagedBlocks {
 		staged.resetBatch()
 		q.staged[k] = staged
 		heap.Push(q.updates, staged)
+		q.globalStats.AddQueues(k, 1)
 		if q.registerer != nil {
 			staged.collector = newQueueStatsCollector(staged)
 			util.RegisterOrGet(q.registerer, staged.collector)
@@ -189,6 +201,7 @@ func (q *blockQueue) removeStaged(s *stagedBlocks) {
 		panic("bug: attempt to delete compaction queue with an invalid priority index")
 	}
 	heap.Remove(q.updates, s.heapIndex)
+	q.globalStats.AddQueues(s.key, -1)
 }
 
 func (s *stagedBlocks) push(block blockEntry) bool {
@@ -203,6 +216,7 @@ func (s *stagedBlocks) push(block blockEntry) bool {
 	}
 	s.batch.size++
 	s.stats.blocks.Add(1)
+	s.queue.globalStats.AddBlocks(s.key, 1)
 	if s.queue.config.exceedsMaxSize(s.batch) ||
 		s.queue.config.exceedsMaxAge(s.batch, s.updatedAt) {
 		s.flush()
@@ -245,6 +259,7 @@ func (s *stagedBlocks) delete(block string) blockEntry {
 	ref.batch.blocks[ref.index] = zeroBlockEntry
 	ref.batch.size--
 	s.stats.blocks.Add(-1)
+	s.queue.globalStats.AddBlocks(s.key, -1)
 	if ref.batch.size == 0 {
 		if ref.batch != s.batch {
 			// We should never ever try to delete the staging batch from the
@@ -311,6 +326,7 @@ func (q *blockQueue) pushBatch(b *batch) bool {
 	b.staged.tail = b
 
 	b.staged.stats.batches.Add(1)
+	q.globalStats.AddBatches(b.staged.key, 1)
 	return true
 }
 
@@ -357,6 +373,7 @@ func (q *blockQueue) removeBatch(b *batch) bool {
 	b.prev = nil
 
 	b.staged.stats.batches.Add(-1)
+	q.globalStats.AddBatches(b.staged.key, -1)
 	return true
 }
 

pkg/metastore/compaction/compactor/compaction_queue_test.go

@@ -14,8 +14,13 @@ import (
 
 func testBlockEntry(id int) blockEntry { return blockEntry{id: strconv.Itoa(id)} }
 
+func testBlockQueue(cfg Config) *blockQueue {
+	stats := newGlobalQueueStats(len(cfg.Levels))
+	return newBlockQueue(cfg, nil, stats)
+}
+
 func TestBlockQueue_Push(t *testing.T) {
-	q := newBlockQueue(Config{Levels: []LevelConfig{{MaxBlocks: 3}}}, nil)
+	q := testBlockQueue(Config{Levels: []LevelConfig{{MaxBlocks: 3}}})
 	key := compactionKey{tenant: "t", shard: 1}
 
 	result := q.stagedBlocks(key).push(testBlockEntry(1))
@@ -46,7 +51,7 @@ func TestBlockQueue_Push(t *testing.T) {
 }
 
 func TestBlockQueue_DuplicateBlock(t *testing.T) {
-	q := newBlockQueue(Config{Levels: []LevelConfig{{MaxBlocks: 3}}}, nil)
+	q := testBlockQueue(Config{Levels: []LevelConfig{{MaxBlocks: 3}}})
 	key := compactionKey{tenant: "t", shard: 1}
 
 	require.True(t, q.stagedBlocks(key).push(testBlockEntry(1)))
@@ -56,7 +61,7 @@ func TestBlockQueue_DuplicateBlock(t *testing.T) {
 }
 
 func TestBlockQueue_Remove(t *testing.T) {
-	q := newBlockQueue(Config{Levels: []LevelConfig{{MaxBlocks: 3}}}, nil)
+	q := testBlockQueue(Config{Levels: []LevelConfig{{MaxBlocks: 3}}})
 	key := compactionKey{tenant: "t", shard: 1}
 	q.stagedBlocks(key).push(testBlockEntry(1))
 	q.stagedBlocks(key).push(testBlockEntry(2))
@@ -73,7 +78,7 @@ func TestBlockQueue_Remove(t *testing.T) {
 }
 
 func TestBlockQueue_RemoveNotFound(t *testing.T) {
-	q := newBlockQueue(Config{Levels: []LevelConfig{{MaxBlocks: 3}}}, nil)
+	q := testBlockQueue(Config{Levels: []LevelConfig{{MaxBlocks: 3}}})
 	key := compactionKey{tenant: "t", shard: 1}
 	remove(q, key, "1")
 	q.stagedBlocks(key).push(testBlockEntry(1))
@@ -85,7 +90,7 @@ func TestBlockQueue_RemoveNotFound(t *testing.T) {
 }
 
 func TestBlockQueue_Linking(t *testing.T) {
-	q := newBlockQueue(Config{Levels: []LevelConfig{{MaxBlocks: 2}}}, nil)
+	q := testBlockQueue(Config{Levels: []LevelConfig{{MaxBlocks: 2}}})
 	key := compactionKey{tenant: "t", shard: 1}
 
 	q.stagedBlocks(key).push(testBlockEntry(1))
@@ -154,7 +159,7 @@ func TestBlockQueue_EmptyQueue(t *testing.T) {
 		numBlocksPerKey = 100
 	)
 
-	q := newBlockQueue(Config{Levels: []LevelConfig{{MaxBlocks: 3}}}, nil)
+	q := testBlockQueue(Config{Levels: []LevelConfig{{MaxBlocks: 3}}})
 	keys := make([]compactionKey, numKeys)
 	for i := 0; i < numKeys; i++ {
 		keys[i] = compactionKey{
@@ -254,7 +259,7 @@ func TestBlockQueue_FlushByAge(t *testing.T) {
 }
 
 func TestBlockQueue_BatchIterator(t *testing.T) {
-	q := newBlockQueue(Config{Levels: []LevelConfig{{MaxBlocks: 3}}}, nil)
+	q := testBlockQueue(Config{Levels: []LevelConfig{{MaxBlocks: 3}}})
 	keys := []compactionKey{
 		{tenant: "t-1", shard: 1},
 		{tenant: "t-2", shard: 2},

pkg/metastore/compaction/compactor/metrics.go

@@ -2,6 +2,7 @@ package compactor
 
 import (
 	"strconv"
+	"sync/atomic"
 
 	"github.com/prometheus/client_golang/prometheus"
 )
@@ -66,3 +67,85 @@ func (b *queueStatsCollector) Collect(m chan<- prometheus.Metric) {
 	m <- prometheus.MustNewConstMetric(b.rejected, prometheus.CounterValue, float64(b.stats.rejected.Load()))
 	m <- prometheus.MustNewConstMetric(b.missed, prometheus.CounterValue, float64(b.stats.missed.Load()))
 }
+
+type globalQueueStats struct {
+	blocksPerLevel  []atomic.Int32
+	queuesPerLevel  []atomic.Int32
+	batchesPerLevel []atomic.Int32
+}
+
+func newGlobalQueueStats(numLevels int) *globalQueueStats {
+	return &globalQueueStats{
+		blocksPerLevel:  make([]atomic.Int32, numLevels),
+		queuesPerLevel:  make([]atomic.Int32, numLevels),
+		batchesPerLevel: make([]atomic.Int32, numLevels),
+	}
+}
+
+func (g *globalQueueStats) AddBlocks(key compactionKey, delta int32) {
+	g.blocksPerLevel[key.level].Add(delta)
+}
+
+func (g *globalQueueStats) AddQueues(key compactionKey, delta int32) {
+	g.queuesPerLevel[key.level].Add(delta)
+}
+
+func (g *globalQueueStats) AddBatches(key compactionKey, delta int32) {
+	g.batchesPerLevel[key.level].Add(delta)
+}
+
+type globalQueueStatsCollector struct {
+	compactionQueue *compactionQueue
+
+	blocks  *prometheus.Desc
+	queues  *prometheus.Desc
+	batches *prometheus.Desc
+}
+
+const globalQueueMetricsPrefix = "compaction_global_queue_"
+
+func newGlobalQueueStatsCollector(compactionQueue *compactionQueue) *globalQueueStatsCollector {
+	variableLabels := []string{"level"}
+
+	return &globalQueueStatsCollector{
+		compactionQueue: compactionQueue,
+
+		blocks: prometheus.NewDesc(
+			globalQueueMetricsPrefix+"blocks_current",
+			"The current number of blocks across all queues, for a compaction level.",
+			variableLabels, nil,
+		),
+
+		queues: prometheus.NewDesc(
+			globalQueueMetricsPrefix+"queues_current",
+			"The current number of queues, for a compaction level.",
+			variableLabels, nil,
+		),
+
+		batches: prometheus.NewDesc(
+			globalQueueMetricsPrefix+"batches_current",
+			"The current number of batches (jobs that are not yet created), for a compaction level.",
+			variableLabels, nil,
+		),
+	}
+}
+
+func (c *globalQueueStatsCollector) Describe(ch chan<- *prometheus.Desc) {
+	ch <- c.blocks
+	ch <- c.queues
+	ch <- c.batches
+}
+
+func (c *globalQueueStatsCollector) Collect(ch chan<- prometheus.Metric) {
+	for levelIdx := range c.compactionQueue.config.Levels {
+		blocksAtLevel := c.compactionQueue.globalStats.blocksPerLevel[levelIdx].Load()
+		queuesAtLevel := c.compactionQueue.globalStats.queuesPerLevel[levelIdx].Load()
+		batchesAtLevel := c.compactionQueue.globalStats.batchesPerLevel[levelIdx].Load()
+
+		levelLabel := strconv.Itoa(levelIdx)
+
+		ch <- prometheus.MustNewConstMetric(c.blocks, prometheus.GaugeValue, float64(blocksAtLevel), levelLabel)
+		ch <- prometheus.MustNewConstMetric(c.queues, prometheus.GaugeValue, float64(queuesAtLevel), levelLabel)
+		ch <- prometheus.MustNewConstMetric(c.batches, prometheus.GaugeValue, float64(batchesAtLevel), levelLabel)
+	}
+}

pkg/metastore/compaction/compactor/metrics_test.go

@@ -28,3 +28,80 @@ func TestCollectorRegistration(t *testing.T) {
 		}
 	}
 }
+
+func TestBlockQueueAggregatedMetrics(t *testing.T) {
+	reg := prometheus.NewRegistry()
+	c := NewCompactor(testConfig, nil, nil, reg)
+
+	entries := []compaction.BlockEntry{
+		{ID: "block1", Tenant: "A", Shard: 0, Level: 0},
+		{ID: "block2", Tenant: "A", Shard: 0, Level: 0},
+		{ID: "block3", Tenant: "A", Shard: 0, Level: 0},
+		{ID: "block4", Tenant: "A", Shard: 1, Level: 0},
+		{ID: "block5", Tenant: "B", Shard: 0, Level: 1},
+		{ID: "block6", Tenant: "B", Shard: 0, Level: 1},
+		{ID: "block7", Tenant: "B", Shard: 0, Level: 1},
+	}
+
+	for _, e := range entries {
+		c.enqueue(e)
+	}
+
+	metrics, err := reg.Gather()
+	if err != nil {
+		t.Fatalf("failed to gather metrics: %v", err)
+	}
+
+	var blocksTotal, queuesTotal, batchesTotal float64
+	var foundBlocks, foundQueues, foundBatches bool
+
+	for _, mf := range metrics {
+		if mf.GetName() == "compaction_global_queue_blocks_current" {
+			for _, m := range mf.GetMetric() {
+				blocksTotal += m.GetGauge().GetValue()
+				foundBlocks = true
+			}
+		}
+		if mf.GetName() == "compaction_global_queue_queues_current" {
+			for _, m := range mf.GetMetric() {
+				queuesTotal += m.GetGauge().GetValue()
+				foundQueues = true
+			}
+		}
+
+		if mf.GetName() == "compaction_global_queue_batches_current" {
+			for _, m := range mf.GetMetric() {
+				batchesTotal += m.GetGauge().GetValue()
+				foundBatches = true
+			}
+		}
+	}
+
+	if !foundBlocks {
+		t.Fatal("compaction_global_queue_blocks metric not found")
+	}
+	if !foundQueues {
+		t.Fatal("compaction_global_queue_queues metric not found")
+	}
+	if !foundBatches {
+		t.Fatal("compaction_global_queue_batches_current metric not found")
+	}
+
+	if blocksTotal != 7 {
+		t.Errorf("expected 7 total blocks, got %v", blocksTotal)
+	}
+
+	if queuesTotal != 3 {
+		t.Errorf("expected 3 total queues, got %v", queuesTotal)
+	}
+
+	// testConfig.Levels[0].MaxBlocks = 3
+	// testConfig.Levels[1].MaxBlocks = 2
+	// (A,0): 3 blocks → 3/3 = 1 batch
+	// (A,1): 1 block → 1/2 = 0 batches
+	// (B,1): 3 blocks → 3/2 = 1 batch
+	// Total = 2 batches
+	if batchesTotal != 2 {
+		t.Errorf("expected 2 total batches, got %v", batchesTotal)
+	}
+}