Metric: Adds are only counted if the object didn't exist yet

Author: alvaroaleman

.golangci.yml

@@ -172,7 +172,7 @@ issues:
         - unused
       # Seems to incorrectly trigger on the two implementations that are only
       # used through an interface and not directly..?
-      path: pkg/controllerworkqueue/metrics\.go
+      path: pkg/controller/priorityqueue/metrics\.go
 
 run:
   go: "1.23"

pkg/controller/priorityqueue/metrics.go

@@ -11,6 +11,9 @@ import (
 
 // This file is mostly a copy of unexported code from
 // https://github.com/kubernetes/kubernetes/blob/1d8828ce707ed9dd7a6a9756385419cce1d202ac/staging/src/k8s.io/client-go/util/workqueue/metrics.go
+//
+// The only difference is the addition of mapLock in defaultQueueMetrics, we want to avoid the need of synchronizing updateUnfinishedWork()
+// with the queue.
 
 type queueMetrics[T comparable] interface {
 	add(item T)

pkg/controller/priorityqueue/priorityqueue.go

@@ -56,9 +56,10 @@ func New[T comparable](name string, o ...Opt[T]) PriorityQueue[T] {
 		opts.MetricProvider = metrics.WorkqueueMetricsProvider{}
 	}
 
-	cwq := &priorityqueue[T]{
-		items: map[T]*item[T]{},
-		queue: btree.NewG(32, less[T]),
+	pq := &priorityqueue[T]{
+		items:   map[T]*item[T]{},
+		queue:   btree.NewG(32, less[T]),
+		metrics: newQueueMetrics[T](opts.MetricProvider, name, clock.RealClock{}),
 		// itemOrWaiterAdded indicates that an item or
 		// waiter was added. It must be buffered, because
 		// if we currently process items we can't tell
@@ -72,25 +73,30 @@ func New[T comparable](name string, o ...Opt[T]) PriorityQueue[T] {
 		tick:              time.Tick,
 	}
 
-	go cwq.spin()
+	go pq.spin()
+	if _, ok := pq.metrics.(noMetrics[T]); !ok {
+		go pq.updateUnfinishedWorkLoop()
+	}
 
-	return wrapWithMetrics(cwq, name, opts.MetricProvider)
+	return pq
 }
 
 type priorityqueue[T comparable] struct {
-	// lock has to be acquired for any access to either items or queue
-	lock  sync.Mutex
-	items map[T]*item[T]
-	queue *btree.BTreeG[*item[T]]
-
-	itemOrWaiterAdded chan struct{}
-
-	rateLimiter workqueue.TypedRateLimiter[T]
+	// lock has to be acquired for any access any of items, queue, addedCounter
+	// or metrics.
+	lock    sync.Mutex
+	items   map[T]*item[T]
+	queue   *btree.BTreeG[*item[T]]
+	metrics queueMetrics[T]
 
 	// addedCounter is a counter of elements added, we need it
 	// because unixNano is not guaranteed to be unique.
 	addedCounter uint64
 
+	itemOrWaiterAdded chan struct{}
+
+	rateLimiter workqueue.TypedRateLimiter[T]
+
 	// locked contains the keys we handed out through Get() and that haven't
 	// yet been returned through Done().
 	locked     sets.Set[T]
@@ -136,6 +142,7 @@ func (w *priorityqueue[T]) AddWithOpts(o AddOpts, items ...T) {
 			}
 			w.items[key] = item
 			w.queue.ReplaceOrInsert(item)
+			w.metrics.add(key)
 			w.addedCounter++
 			continue
 		}
@@ -204,11 +211,12 @@ func (w *priorityqueue[T]) spin() {
 					return true
 				}
 
-				w.get <- *item
+				w.metrics.get(item.key)
 				w.locked.Insert(item.key)
 				w.waiters.Add(-1)
 				delete(w.items, item.key)
 				w.queue.Delete(item)
+				w.get <- *item
 
 				return true
 			})
@@ -258,6 +266,7 @@ func (w *priorityqueue[T]) Done(item T) {
 	w.lockedLock.Lock()
 	defer w.lockedLock.Unlock()
 	w.locked.Delete(item)
+	w.metrics.done(item)
 	w.notifyItemOrWaiterAdded()
 }
 
@@ -306,52 +315,13 @@ type item[T comparable] struct {
 	readyAt         *time.Time
 }
 
-func wrapWithMetrics[T comparable](q *priorityqueue[T], name string, provider workqueue.MetricsProvider) PriorityQueue[T] {
-	mwq := &metricWrappedQueue[T]{
-		priorityqueue: q,
-		metrics:       newQueueMetrics[T](provider, name, clock.RealClock{}),
-	}
-
-	go mwq.updateUnfinishedWorkLoop()
-
-	return mwq
-}
-
-type metricWrappedQueue[T comparable] struct {
-	*priorityqueue[T]
-	metrics queueMetrics[T]
-}
-
-func (m *metricWrappedQueue[T]) AddWithOpts(o AddOpts, items ...T) {
-	for _, item := range items {
-		m.metrics.add(item)
-	}
-	m.priorityqueue.AddWithOpts(o, items...)
-}
-
-func (m *metricWrappedQueue[T]) GetWithPriority() (T, int, bool) {
-	item, priority, shutdown := m.priorityqueue.GetWithPriority()
-	m.metrics.get(item)
-	return item, priority, shutdown
-}
-
-func (m *metricWrappedQueue[T]) Get() (T, bool) {
-	item, _, shutdown := m.GetWithPriority()
-	return item, shutdown
-}
-
-func (m *metricWrappedQueue[T]) Done(item T) {
-	m.metrics.done(item)
-	m.priorityqueue.Done(item)
-}
-
-func (m *metricWrappedQueue[T]) updateUnfinishedWorkLoop() {
-	t := time.NewTicker(time.Millisecond)
+func (w *priorityqueue[T]) updateUnfinishedWorkLoop() {
+	t := time.NewTicker(500 * time.Millisecond) // borrowed from workqueue: https://github.com/kubernetes/kubernetes/blob/67a807bf142c7a2a5ecfdb2a5d24b4cdea4cc79c/staging/src/k8s.io/client-go/util/workqueue/queue.go#L182
 	defer t.Stop()
 	for range t.C {
-		if m.priorityqueue.ShuttingDown() {
+		if w.shutdown.Load() {
 			return
 		}
-		m.metrics.updateUnfinishedWork()
+		w.metrics.updateUnfinishedWork()
 	}
 }

pkg/controller/priorityqueue/priorityqueue_test.go

@@ -90,12 +90,12 @@ var _ = Describe("Controllerworkqueue", func() {
 
 		Consistently(q.Len).Should(Equal(1))
 
-		cwq := q.(*metricWrappedQueue[string])
+		cwq := q.(*priorityqueue[string])
 		cwq.lockedLock.Lock()
 		Expect(cwq.locked.Len()).To(Equal(0))
 
 		Expect(metrics.depth["test"]).To(Equal(1))
-		Expect(metrics.adds["test"]).To(Equal(2))
+		Expect(metrics.adds["test"]).To(Equal(1))
 	})
 
 	It("retains the highest priority", func() {
@@ -112,7 +112,7 @@ var _ = Describe("Controllerworkqueue", func() {
 		Expect(q.Len()).To(Equal(0))
 
 		Expect(metrics.depth["test"]).To(Equal(0))
-		Expect(metrics.adds["test"]).To(Equal(2))
+		Expect(metrics.adds["test"]).To(Equal(1))
 	})
 
 	It("gets pushed to the front if the priority increases", func() {
@@ -131,7 +131,7 @@ var _ = Describe("Controllerworkqueue", func() {
 		Expect(q.Len()).To(Equal(2))
 
 		Expect(metrics.depth["test"]).To(Equal(2))
-		Expect(metrics.adds["test"]).To(Equal(4))
+		Expect(metrics.adds["test"]).To(Equal(3))
 	})
 
 	It("retains the lowest after duration", func() {
@@ -147,7 +147,7 @@ var _ = Describe("Controllerworkqueue", func() {
 
 		Expect(q.Len()).To(Equal(0))
 		Expect(metrics.depth["test"]).To(Equal(0))
-		Expect(metrics.adds["test"]).To(Equal(2))
+		Expect(metrics.adds["test"]).To(Equal(1))
 	})
 
 	It("returns an item only after after has passed", func() {
@@ -158,7 +158,7 @@ var _ = Describe("Controllerworkqueue", func() {
 		nowLock := sync.Mutex{}
 		tick := make(chan time.Time)
 
-		cwq := q.(*metricWrappedQueue[string])
+		cwq := q.(*priorityqueue[string])
 		cwq.now = func() time.Time {
 			nowLock.Lock()
 			defer nowLock.Unlock()
@@ -199,7 +199,7 @@ var _ = Describe("Controllerworkqueue", func() {
 		nowLock := sync.Mutex{}
 		tick := make(chan time.Time)
 
-		cwq := q.(*metricWrappedQueue[string])
+		cwq := q.(*priorityqueue[string])
 		cwq.now = func() time.Time {
 			nowLock.Lock()
 			defer nowLock.Unlock()