feat(scheduler): add metric for time spend in queue (#109346)

pkg/util/scheduler/queue.go

@@ -23,9 +23,17 @@ var ErrTenantQueueFull = errors.New("tenant queue full")
 var ErrNilRunnable = errors.New("cannot enqueue nil runnable")
 var ErrMissingTenantID = errors.New("item requires TenantID")
 
+// queuedItem represents a runnable task annotated with its enqueue timestamp.
+// The timestamp allows us to compute how long the task spent waiting in the queue
+// before being picked up by a worker.
+type queuedItem struct {
+	runnable   func()
+	enqueuedAt time.Time
+}
+
 type tenantQueue struct {
 	id       string
-	items    []func()
+	items    []queuedItem
 	isActive bool
 }
 
@@ -43,7 +51,7 @@ func (tq *tenantQueue) isFull(maxSize int) bool {
 	return maxSize > 0 && len(tq.items) >= maxSize
 }
 func (tq *tenantQueue) addItem(runnable func()) {
-	tq.items = append(tq.items, runnable)
+	tq.items = append(tq.items, queuedItem{runnable: runnable, enqueuedAt: time.Now()})
 }
 
 type enqueueRequest struct {
@@ -106,7 +114,7 @@ type Queue struct {
 	// Metrics
 	queueLength       *prometheus.GaugeVec
 	discardedRequests *prometheus.CounterVec
-	enqueueDuration   prometheus.Histogram
+	queueWaitDuration *prometheus.HistogramVec
 }
 
 type QueueOptions struct {
@@ -148,11 +156,13 @@ func NewQueue(opts *QueueOptions) *Queue {
 		Name: "discarded_requests_total",
 		Help: "Total number of discarded requests",
 	}, []string{"tenant", "reason"})
-	q.enqueueDuration = promauto.With(opts.Registerer).NewHistogram(prometheus.HistogramOpts{
-		Name:    "enqueue_duration_seconds",
-		Help:    "Duration of enqueue operation in seconds",
-		Buckets: prometheus.DefBuckets,
-	})
+	q.queueWaitDuration = promauto.With(opts.Registerer).NewHistogramVec(prometheus.HistogramOpts{
+		Name:                            "queue_wait_duration_seconds",
+		Help:                            "Time items spend waiting in the queue before being dequeued, in seconds",
+		NativeHistogramBucketFactor:     1.1,
+		NativeHistogramMaxBucketNumber:  160,
+		NativeHistogramMinResetDuration: time.Hour,
+	}, []string{"tenant"})
 
 	q.Service = services.NewBasicService(nil, q.dispatcherLoop, q.stopping)
 
@@ -175,8 +185,11 @@ func (q *Queue) scheduleRoundRobin() {
 		tq := tenantElem.Value.(*tenantQueue)
 
 		// Get and deliver the runnable item
-		item := tq.items[0]
-		req.respChan <- dequeueResponse{runnable: item, err: nil}
+		qi := tq.items[0]
+		req.respChan <- dequeueResponse{runnable: qi.runnable, err: nil}
+
+		// Observe how long the item spent waiting in the queue.
+		q.queueWaitDuration.WithLabelValues(tq.id).Observe(time.Since(qi.enqueuedAt).Seconds())
 
 		// Update bookkeeping
 		q.pendingDequeueRequests.Remove(reqElem)
@@ -200,7 +213,7 @@ func (q *Queue) handleEnqueueRequest(req enqueueRequest) {
 	if !exists {
 		tq = &tenantQueue{
 			id:    req.tenantID,
-			items: make([]func(), 0, 8),
+			items: make([]queuedItem, 0, 8),
 		}
 		q.tenantQueues[req.tenantID] = tq
 	}
@@ -275,8 +288,6 @@ func (q *Queue) Enqueue(ctx context.Context, tenantID string, runnable func()) e
 		return ErrQueueClosed
 	}
 
-	start := time.Now()
-
 	respChan := make(chan error, 1)
 	req := enqueueRequest{
 		tenantID: tenantID,
@@ -295,7 +306,6 @@ func (q *Queue) Enqueue(ctx context.Context, tenantID string, runnable func()) e
 		q.discardedRequests.WithLabelValues(tenantID, "context_canceled").Inc()
 		err = ctx.Err()
 	}
-	q.enqueueDuration.Observe(time.Since(start).Seconds())
 
 	return err
 }