🌱 Add priority label to PQ depth metric

pkg/controller/priorityqueue/metrics.go

@@ -6,6 +6,7 @@ import (
 
 	"k8s.io/client-go/util/workqueue"
 	"k8s.io/utils/clock"
+	"sigs.k8s.io/controller-runtime/pkg/internal/metrics"
 )
 
 // This file is mostly a copy of unexported code from
@@ -14,8 +15,9 @@ import (
 // The only two differences are the addition of mapLock in defaultQueueMetrics and converging retryMetrics into queueMetrics.
 
 type queueMetrics[T comparable] interface {
-	add(item T)
-	get(item T)
+	add(item T, priority int)
+	get(item T, priority int)
+	updateDepthWithPriorityMetric(oldPriority, newPriority int)
 	done(item T)
 	updateUnfinishedWork()
 	retry()
@@ -25,9 +27,9 @@ func newQueueMetrics[T comparable](mp workqueue.MetricsProvider, name string, cl
 	if len(name) == 0 {
 		return noMetrics[T]{}
 	}
-	return &defaultQueueMetrics[T]{
+
+	dqm := &defaultQueueMetrics[T]{
 		clock:                   clock,
-		depth:                   mp.NewDepthMetric(name),
 		adds:                    mp.NewAddsMetric(name),
 		latency:                 mp.NewLatencyMetric(name),
 		workDuration:            mp.NewWorkDurationMetric(name),
@@ -37,14 +39,22 @@ func newQueueMetrics[T comparable](mp workqueue.MetricsProvider, name string, cl
 		processingStartTimes:    map[T]time.Time{},
 		retries:                 mp.NewRetriesMetric(name),
 	}
+
+	if mpp, ok := mp.(metrics.MetricsProviderWithPriority); ok {
+		dqm.depthWithPriority = mpp.NewDepthMetricWithPriority(name)
+	} else {
+		dqm.depth = mp.NewDepthMetric(name)
+	}
+	return dqm
 }
 
 // defaultQueueMetrics expects the caller to lock before setting any metrics.
 type defaultQueueMetrics[T comparable] struct {
 	clock clock.Clock
 
 	// current depth of a workqueue
-	depth workqueue.GaugeMetric
+	depth             workqueue.GaugeMetric
+	depthWithPriority metrics.DepthMetricWithPriority
 	// total number of adds handled by a workqueue
 	adds workqueue.CounterMetric
 	// how long an item stays in a workqueue
@@ -64,13 +74,17 @@ type defaultQueueMetrics[T comparable] struct {
 }
 
 // add is called for ready items only
-func (m *defaultQueueMetrics[T]) add(item T) {
+func (m *defaultQueueMetrics[T]) add(item T, priority int) {
 	if m == nil {
 		return
 	}
 
 	m.adds.Inc()
-	m.depth.Inc()
+	if m.depthWithPriority != nil {
+		m.depthWithPriority.Inc(priority)
+	} else {
+		m.depth.Inc()
+	}
 
 	m.mapLock.Lock()
 	defer m.mapLock.Unlock()
@@ -80,12 +94,16 @@ func (m *defaultQueueMetrics[T]) add(item T) {
 	}
 }
 
-func (m *defaultQueueMetrics[T]) get(item T) {
+func (m *defaultQueueMetrics[T]) get(item T, priority int) {
 	if m == nil {
 		return
 	}
 
-	m.depth.Dec()
+	if m.depthWithPriority != nil {
+		m.depthWithPriority.Dec(priority)
+	} else {
+		m.depth.Dec()
+	}
 
 	m.mapLock.Lock()
 	defer m.mapLock.Unlock()
@@ -97,6 +115,13 @@ func (m *defaultQueueMetrics[T]) get(item T) {
 	}
 }
 
+func (m *defaultQueueMetrics[T]) updateDepthWithPriorityMetric(oldPriority, newPriority int) {
+	if m.depthWithPriority != nil {
+		m.depthWithPriority.Dec(oldPriority)
+		m.depthWithPriority.Inc(newPriority)
+	}
+}
+
 func (m *defaultQueueMetrics[T]) done(item T) {
 	if m == nil {
 		return
@@ -139,8 +164,9 @@ func (m *defaultQueueMetrics[T]) retry() {
 
 type noMetrics[T any] struct{}
 
-func (noMetrics[T]) add(item T)            {}
-func (noMetrics[T]) get(item T)            {}
-func (noMetrics[T]) done(item T)           {}
-func (noMetrics[T]) updateUnfinishedWork() {}
-func (noMetrics[T]) retry()                {}
+func (noMetrics[T]) add(item T, priority int)                                   {}
+func (noMetrics[T]) get(item T, priority int)                                   {}
+func (noMetrics[T]) updateDepthWithPriorityMetric(oldPriority, newPriority int) {}
+func (noMetrics[T]) done(item T)                                                {}
+func (noMetrics[T]) updateUnfinishedWork()                                      {}
+func (noMetrics[T]) retry()                                                     {}

pkg/controller/priorityqueue/metrics_test.go

@@ -4,11 +4,12 @@ import (
 	"sync"
 
 	"k8s.io/client-go/util/workqueue"
+	"sigs.k8s.io/controller-runtime/pkg/internal/metrics"
 )
 
 func newFakeMetricsProvider() *fakeMetricsProvider {
 	return &fakeMetricsProvider{
-		depth:                   make(map[string]int),
+		depth:                   make(map[string]map[int]int),
 		adds:                    make(map[string]int),
 		latency:                 make(map[string][]float64),
 		workDuration:            make(map[string][]float64),
@@ -19,8 +20,10 @@ func newFakeMetricsProvider() *fakeMetricsProvider {
 	}
 }
 
+var _ metrics.MetricsProviderWithPriority = &fakeMetricsProvider{}
+
 type fakeMetricsProvider struct {
-	depth                   map[string]int
+	depth                   map[string]map[int]int
 	adds                    map[string]int
 	latency                 map[string][]float64
 	workDuration            map[string][]float64
@@ -31,9 +34,13 @@ type fakeMetricsProvider struct {
 }
 
 func (f *fakeMetricsProvider) NewDepthMetric(name string) workqueue.GaugeMetric {
+	panic("Should never be called. Expected NewDepthMetricWithPriority to be called instead")
+}
+
+func (f *fakeMetricsProvider) NewDepthMetricWithPriority(name string) metrics.DepthMetricWithPriority {
 	f.mu.Lock()
 	defer f.mu.Unlock()
-	f.depth[name] = 0
+	f.depth[name] = map[int]int{}
 	return &fakeGaugeMetric{m: &f.depth, mu: &f.mu, name: name}
 }
 
@@ -80,21 +87,21 @@ func (f *fakeMetricsProvider) NewRetriesMetric(name string) workqueue.CounterMet
 }
 
 type fakeGaugeMetric struct {
-	m    *map[string]int
+	m    *map[string]map[int]int
 	mu   *sync.Mutex
 	name string
 }
 
-func (fg *fakeGaugeMetric) Inc() {
+func (fg *fakeGaugeMetric) Inc(priority int) {
 	fg.mu.Lock()
 	defer fg.mu.Unlock()
-	(*fg.m)[fg.name]++
+	(*fg.m)[fg.name][priority]++
 }
 
-func (fg *fakeGaugeMetric) Dec() {
+func (fg *fakeGaugeMetric) Dec(priority int) {
 	fg.mu.Lock()
 	defer fg.mu.Unlock()
-	(*fg.m)[fg.name]--
+	(*fg.m)[fg.name][priority]--
 }
 
 type fakeCounterMetric struct {

pkg/controller/priorityqueue/priorityqueue.go

@@ -156,7 +156,7 @@ func (w *priorityqueue[T]) AddWithOpts(o AddOpts, items ...T) {
 			w.items[key] = item
 			w.queue.ReplaceOrInsert(item)
 			if item.ReadyAt == nil {
-				w.metrics.add(key)
+				w.metrics.add(key, item.Priority)
 			}
 			w.addedCounter++
 			continue
@@ -166,12 +166,16 @@ func (w *priorityqueue[T]) AddWithOpts(o AddOpts, items ...T) {
 		// will affect the order - Just delete and re-add.
 		item, _ := w.queue.Delete(w.items[key])
 		if o.Priority > item.Priority {
+			// Update depth metric only if the item in the queue was already added to the depth metric.
+			if item.ReadyAt == nil || w.becameReady.Has(key) {
+				w.metrics.updateDepthWithPriorityMetric(item.Priority, o.Priority)
+			}
 			item.Priority = o.Priority
 		}
 
 		if item.ReadyAt != nil && (readyAt == nil || readyAt.Before(*item.ReadyAt)) {
 			if readyAt == nil && !w.becameReady.Has(key) {
-				w.metrics.add(key)
+				w.metrics.add(key, item.Priority)
 			}
 			item.ReadyAt = readyAt
 		}
@@ -223,7 +227,7 @@ func (w *priorityqueue[T]) spin() {
 						return false
 					}
 					if !w.becameReady.Has(item.Key) {
-						w.metrics.add(item.Key)
+						w.metrics.add(item.Key, item.Priority)
 						w.becameReady.Insert(item.Key)
 					}
 				}
@@ -239,7 +243,7 @@ func (w *priorityqueue[T]) spin() {
 					return true
 				}
 
-				w.metrics.get(item.Key)
+				w.metrics.get(item.Key, item.Priority)
 				w.locked.Insert(item.Key)
 				w.waiters.Add(-1)
 				delete(w.items, item.Key)