Add request enqueue latency metric

Author: RyanRosario

pkg/epp/flowcontrol/controller/internal/processor.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"strconv"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -208,14 +209,24 @@ func (sp *ShardProcessor) Run(ctx context.Context) {
 // enqueue processes an item received from the enqueueChan.
 // It handles capacity checks, checks for external finalization, and either admits the item to a queue or rejects it.
 func (sp *ShardProcessor) enqueue(item *FlowItem) {
+
+	outcome := item.FinalState()
 	req := item.OriginalRequest()
 	key := req.FlowKey()
+	priorityStr := strconv.Itoa(key.Priority)
+
+	startTime := time.Now()
+
+	defer func() {
+		metrics.RecordFlowControlRequestEnqueueDuration(priorityStr, outcome.Outcome.String(), time.Since(startTime))
+
+	}()
 
 	// --- Optimistic External Finalization Check ---
 	// Check if the item was finalized by the Controller (due to TTL/cancellation) while it was buffered in enqueueChan.
 	// This is an optimistic check to avoid unnecessary processing on items already considered dead.
 	// The ultimate guarantee of cleanup for any races is the runCleanupSweep mechanism.
-	if finalState := item.FinalState(); finalState != nil {
+	if finalState := outcome; finalState != nil {
 		sp.logger.V(logutil.TRACE).Info("Item finalized externally before processing, discarding.",
 			"outcome", finalState.Outcome, "err", finalState.Err, "flowKey", key, "reqID", req.ID())
 		return

pkg/epp/metrics/metrics.go

@@ -399,6 +399,18 @@ var (
 		append([]string{"fairness_id", "priority", "outcome", "inference_pool"}, ModelLabels...),
 	)
 
+	flowControlRequestEnqueueDuration = prometheus.NewHistogramVec(
+		prometheus.HistogramOpts{
+			Subsystem: InferenceExtension,
+			Name:      "flow_control_request_enqueue_duration_seconds",
+			Help:      metricsutil.HelpMsgWithStability("Distribution of the time taken to enqueue requests by the EPP flow control layer.", compbasemetrics.ALPHA),
+			Buckets: []float64{
+				5, 10, 25, 50, 100, 250, 500, 1000,
+			},
+		},
+		[]string{"priority", "outcome"},
+	)
+
 	flowControlQueueSize = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Subsystem: InferenceExtension,
@@ -461,6 +473,7 @@ func Register(customCollectors ...prometheus.Collector) {
 		metrics.Registry.MustRegister(PrefixCacheHitLength)
 		metrics.Registry.MustRegister(flowControlRequestQueueDuration)
 		metrics.Registry.MustRegister(flowControlQueueSize)
+		metrics.Registry.MustRegister(flowControlRequestEnqueueDuration)
 		metrics.Registry.MustRegister(inferenceModelRewriteDecisionsTotal)
 		for _, collector := range customCollectors {
 			metrics.Registry.MustRegister(collector)
@@ -507,6 +520,7 @@ func Reset() {
 	PrefixCacheHitLength.Reset()
 	flowControlRequestQueueDuration.Reset()
 	flowControlQueueSize.Reset()
+	flowControlRequestEnqueueDuration.Reset()
 	inferenceModelRewriteDecisionsTotal.Reset()
 }
 
@@ -787,6 +801,16 @@ func RecordFlowControlRequestQueueDuration(
 	).Observe(duration.Seconds())
 }
 
+// RecordFlowControlRequestQueueDuration records the duration a request was in the enqueuing process in the Flow Control layer.
+func RecordFlowControlRequestEnqueueDuration(
+	priority string, outcome string,
+	duration time.Duration,
+) {
+	flowControlRequestEnqueueDuration.WithLabelValues(
+		priority, outcome,
+	).Observe(duration.Seconds())
+}
+
 // IncFlowControlQueueSize increments the Flow Control queue size gauge.
 func IncFlowControlQueueSize(fairnessID, priority, inferencePool, modelName, targetModelName string) {
 	flowControlQueueSize.WithLabelValues(fairnessID, priority, inferencePool, modelName, targetModelName).Inc()

pkg/epp/metrics/metrics_test.go

@@ -685,6 +685,73 @@ func TestSchedulerE2ELatency(t *testing.T) {
 	}
 }
 
+func TestFlowControlEnqueueDurationMetric(t *testing.T) {
+	Reset()
+
+	scenarios := []struct {
+		name       string
+		priorities []string
+		outcomes   []string
+		durations  []time.Duration
+	}{
+		{
+			name: "multiple enqueue latencies",
+			priorities: []string{
+				"1", "1", "1", "1", "1", "1", "1",
+				"2", "2", "2", "2", "2", "2", "2",
+			},
+			outcomes: []string{
+				"Dispatched", "NotYetFinalized", "RejectedCapacity", "RejectedOther", "EvictedTTL", "EvictedContextCancelled", "EvictedOther",
+				"Dispatched", "NotYetFinalized", "RejectedCapacity", "RejectedOther", "EvictedTTL", "EvictedContextCancelled", "EvictedOther",
+			},
+			durations: []time.Duration{
+				200 * time.Nanosecond,
+				800 * time.Nanosecond,
+				1500 * time.Nanosecond,
+				3 * time.Nanosecond,
+				8 * time.Nanosecond,
+				15 * time.Nanosecond,
+				30 * time.Nanosecond,
+				75 * time.Nanosecond,
+				150 * time.Nanosecond,
+				200 * time.Nanosecond,
+				800 * time.Nanosecond,
+				1500 * time.Nanosecond,
+				3 * time.Nanosecond,
+				15 * time.Nanosecond,
+			},
+		},
+	}
+
+	for _, scenario := range scenarios {
+		t.Run(scenario.name, func(t *testing.T) {
+			// FIX: Loop through the indices to record one event at a time
+			for i := range scenario.priorities {
+				RecordFlowControlRequestEnqueueDuration(
+					scenario.priorities[i],
+					scenario.outcomes[i],
+					scenario.durations[i],
+				)
+			}
+
+			// Validate results
+			func() {
+				wantEnqueueLatency, err := os.Open("testdata/flow_control_enqueue_duration_seconds_metric")
+				if err != nil {
+					t.Fatal(err)
+				}
+				defer wantEnqueueLatency.Close()
+
+				if err := testutil.GatherAndCompare(metrics.Registry, wantEnqueueLatency, "inference_extension_flow_control_enqueue_duration_seconds"); err != nil {
+					t.Error(err)
+				}
+			}()
+		})
+	}
+}
+
+// TODO (7028): Research histogram bins using real-world data to ensure they are optimal.
+
 func TestSchedulerAttemptsTotal(t *testing.T) {
 
 	scenarios := []struct {