🐛 Priorityqueue: Yet another queue_depth metric fix

Inside the priorityqueues `spin` we call the metrics `add` if an item becomes ready so that the `queue_depth` metric gets incremented. To avoid doing this multiple times for the same item, we track the key in a map and remove it there when we hand the item out. If an item gets added without `RequeueAfter` that is already on the queue but with a `RequeueAfter` we also call the metrics `add` - But if we already did that in `spin` we will count the item twice.

Author: alvaroaleman

pkg/controller/priorityqueue/metrics.go

@@ -85,11 +85,11 @@ func (m *defaultQueueMetrics[T]) get(item T) {
 return
 }
 
+m.depth.Dec()
+
 m.mapLock.Lock()
 defer m.mapLock.Unlock()
 
-m.depth.Dec()
-
 m.processingStartTimes[item] = m.clock.Now()
 if startTime, exists := m.addTimes[item]; exists {
 m.latency.Observe(m.sinceInSeconds(startTime))

pkg/controller/priorityqueue/priorityqueue.go

@@ -168,7 +168,7 @@ func (w *priorityqueue[T]) AddWithOpts(o AddOpts, items ...T) {
 }
 
 if item.ReadyAt != nil && (readyAt == nil || readyAt.Before(*item.ReadyAt)) {
-if readyAt == nil {
+if readyAt == nil && !w.becameReady.Has(key) {
 w.metrics.add(key)
 }
 item.ReadyAt = readyAt

pkg/controller/priorityqueue/priorityqueue_test.go

@@ -396,6 +396,24 @@ var _ = Describe("Controllerworkqueue", func() {
 metrics.mu.Lock()
 Expect(metrics.depth["test"]).To(Equal(1))
 })
+
+It("Updates metrics correctly for an item whose requeueAfter expired that gets added again without requeueAfter", func() {
+q, metrics := newQueue()
+defer q.ShutDown()
+
+q.AddWithOpts(AddOpts{After: 50 * time.Millisecond}, "foo")
+time.Sleep(100 * time.Millisecond)
+
+Expect(q.Len()).To(Equal(1))
+metrics.mu.Lock()
+Expect(metrics.depth["test"]).To(Equal(1))
+metrics.mu.Unlock()
+
+q.AddWithOpts(AddOpts{}, "foo")
+Expect(q.Len()).To(Equal(1))
+metrics.mu.Lock()
+Expect(metrics.depth["test"]).To(Equal(1))
+})
 })
 
 func BenchmarkAddGetDone(b *testing.B) {