feat(flowcontrol): add per-band and global queue utilization gauges

Signed-off-by: Alok Behera <alokbeherak061@gmail.com>

Author: sudoalok

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

@@ -356,6 +356,36 @@ func (p *Processor) hasCapacity(priority int, itemByteSize uint64) (bool, contra
 	return true, stats
 }
 
+// recordQueueUtilization emits occupancy/limit ratio gauges for each priority band and, when a global limit is
+// configured, the global aggregate (priority ""). It reads a single Stats() snapshot; the data source is expected to
+// move with the engine-merge refactor, but the metric contract (names, labels, semantics) stays stable (#2102).
+// A dimension with no configured limit (capacity 0) is omitted rather than reported as a misleading 0.
+func (p *Processor) recordQueueUtilization() {
+	stats := p.registry.Stats()
+
+	for priority, band := range stats.PerPriorityBandStats {
+		priorityStr := strconv.Itoa(priority)
+		if band.CapacityRequests > 0 {
+			metrics.RecordFlowControlQueueUtilizationRequests(priorityStr, p.poolName,
+				float64(band.Len)/float64(band.CapacityRequests))
+		}
+		if band.CapacityBytes > 0 {
+			metrics.RecordFlowControlQueueUtilizationBytes(priorityStr, p.poolName,
+				float64(band.ByteSize)/float64(band.CapacityBytes))
+		}
+	}
+
+	// Global aggregate across all bands, only when a global limit is configured.
+	if stats.TotalCapacityRequests > 0 {
+		metrics.RecordFlowControlQueueUtilizationRequests("", p.poolName,
+			float64(stats.TotalLen)/float64(stats.TotalCapacityRequests))
+	}
+	if stats.TotalCapacityBytes > 0 {
+		metrics.RecordFlowControlQueueUtilizationBytes("", p.poolName,
+			float64(stats.TotalByteSize)/float64(stats.TotalCapacityBytes))
+	}
+}
+
 // dispatchCycle attempts to dispatch a single item by iterating through priority bands from highest to lowest.
 // It applies the configured policies for each band to select an item and then attempts to dispatch it.
 // It returns true if an item was successfully dispatched, and false otherwise.
@@ -380,6 +410,9 @@ func (p *Processor) dispatchCycle(ctx context.Context) bool {
 	// Record pool saturation metric
 	metrics.RecordFlowControlPoolSaturation(p.poolName, saturation)
 
+	// Record queue utilization ratios (the demand-side twin of saturation) from the same periodic sample.
+	p.recordQueueUtilization()
+
 	priorities := p.registry.AllOrderedPriorityLevels()
 	ceilings := p.usageLimitPolicy.ComputeLimit(ctx, saturation, priorities)
 

pkg/epp/flowcontrol/integration_test.go

@@ -19,6 +19,7 @@ package flowcontrol_test
 import (
 	"context"
 	"fmt"
+	"strconv"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -1163,7 +1164,9 @@ func TestFlowControlMetricsEmitted(t *testing.T) {
 	eppmetrics.Register()
 
 	detector := newBlockedDetector()
-	h := newHarness(t, harnessOpts{detector: detector})
+	// bandMaxRequests bounds the band so the request-count utilization ratio is computable
+	// (occupancy/limit); without a configured limit the ratio series is intentionally omitted.
+	h := newHarness(t, harnessOpts{detector: detector, bandMaxRequests: 10})
 
 	key := flowcontrol.FlowKey{ID: "metrics-flow", Priority: 0}
 
@@ -1186,6 +1189,17 @@ func TestFlowControlMetricsEmitted(t *testing.T) {
 	require.Greater(t, queueSizeGaugeSum(t, key.ID), 0.0,
 		"queue_size should be > 0 while a request is actively queued")
 
+	// The band is bounded (bandMaxRequests above), so its utilization ratio is occupancy/limit.
+	// Unlike queue_size, these gauges are refreshed by the dispatch cycle rather than synchronously
+	// on enqueue, so they trail admission by up to one tick.
+	priorityStr := strconv.Itoa(key.Priority)
+	require.Eventually(t, func() bool {
+		return queueUtilizationGauge(t, priorityStr) > 0
+	}, time.Second, time.Millisecond,
+		"utilization ratio should be > 0 while a request is actively queued")
+	require.LessOrEqual(t, queueUtilizationGauge(t, priorityStr), 1.0,
+		"utilization ratio must stay within [0,1]")
+
 	// Unblock the detector so the request finalizes deterministically via dispatch.
 	detector.Unblock(1)
 
@@ -1201,6 +1215,33 @@ func TestFlowControlMetricsEmitted(t *testing.T) {
 	// gauge is already back at 0 -- no polling needed.
 	require.Zero(t, queueSizeGaugeSum(t, key.ID),
 		"queue_size should return to 0 after the request finalizes")
+
+	// The utilization gauges are dispatch-cycle driven, so they trail the drain by up to one tick.
+	require.Eventually(t, func() bool {
+		return queueUtilizationGauge(t, priorityStr) == 0
+	}, time.Second, time.Millisecond,
+		"utilization ratio should return to 0 after the queue drains")
+}
+
+// queueUtilizationGauge returns the request-count utilization ratio recorded for the given priority
+// band, or -1 if no series exists for that band.
+func queueUtilizationGauge(t *testing.T, priority string) float64 {
+	t.Helper()
+	families, err := ctrlmetrics.Registry.Gather()
+	require.NoError(t, err)
+	for _, f := range families {
+		if f.GetName() != "llm_d_epp_flow_control_queue_utilization_requests" {
+			continue
+		}
+		for _, m := range f.GetMetric() {
+			for _, lp := range m.GetLabel() {
+				if lp.GetName() == "priority" && lp.GetValue() == priority {
+					return m.GetGauge().GetValue()
+				}
+			}
+		}
+	}
+	return -1
 }
 
 // ============================================================================

pkg/epp/metrics/llm_d_router_metrics.go

@@ -380,6 +380,24 @@ var (
 		[]string{"inference_pool"},
 	)
 
+	llmdFlowControlQueueUtilizationRequests = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Subsystem: LLMDRouterEndpointPickerSubsystem,
+			Name:      "flow_control_queue_utilization_requests",
+			Help:      metricsutil.HelpMsgWithStability("Fraction of the request-count limit currently occupied by the Flow Control queue (0.0-1.0), per priority band. The global aggregate is reported with priority=\"\" and only when a global request limit is configured; a dimension with no limit is omitted rather than reported as 0.", compbasemetrics.ALPHA),
+		},
+		[]string{"priority", "inference_pool"},
+	)
+
+	llmdFlowControlQueueUtilizationBytes = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Subsystem: LLMDRouterEndpointPickerSubsystem,
+			Name:      "flow_control_queue_utilization_bytes",
+			Help:      metricsutil.HelpMsgWithStability("Fraction of the byte-size limit currently occupied by the Flow Control queue (0.0-1.0), per priority band. The global aggregate is reported with priority=\"\" and only when a global byte limit is configured; a dimension with no limit is omitted rather than reported as 0.", compbasemetrics.ALPHA),
+		},
+		[]string{"priority", "inference_pool"},
+	)
+
 	llmdFlowControlRequestsTotal = prometheus.NewCounterVec(
 		prometheus.CounterOpts{
 			Subsystem: LLMDRouterEndpointPickerSubsystem,

pkg/epp/metrics/metrics.go

@@ -472,6 +472,8 @@ func Register(customCollectors ...prometheus.Collector) {
 		metrics.Registry.MustRegister(llmdFlowControlQueueBytes)
 		metrics.Registry.MustRegister(flowControlPoolSaturation)
 		metrics.Registry.MustRegister(llmdFlowControlPoolSaturation)
+		metrics.Registry.MustRegister(llmdFlowControlQueueUtilizationRequests)
+		metrics.Registry.MustRegister(llmdFlowControlQueueUtilizationBytes)
 		metrics.Registry.MustRegister(flowControlRequestEnqueueDuration)
 		metrics.Registry.MustRegister(llmdFlowControlRequestEnqueueDuration)
 		metrics.Registry.MustRegister(llmdFlowControlRequestsTotal)
@@ -542,6 +544,8 @@ func Reset() {
 	llmdFlowControlQueueBytes.Reset()
 	flowControlPoolSaturation.Reset()
 	llmdFlowControlPoolSaturation.Reset()
+	llmdFlowControlQueueUtilizationRequests.Reset()
+	llmdFlowControlQueueUtilizationBytes.Reset()
 	flowControlRequestEnqueueDuration.Reset()
 	llmdFlowControlRequestEnqueueDuration.Reset()
 	flowControlDispatchCycleDuration.Reset()
@@ -906,6 +910,20 @@ func RecordFlowControlPoolSaturation(inferencePool string, saturation float64) {
 	llmdFlowControlPoolSaturation.WithLabelValues(inferencePool).Set(saturation)
 }
 
+// RecordFlowControlQueueUtilizationRequests sets the request-count queue utilization ratio (occupancy/limit, 0.0-1.0)
+// for a priority band, or the global aggregate when priority is "". Callers only invoke this for a bounded dimension,
+// so an unlimited limit yields no series rather than a misleading 0.
+func RecordFlowControlQueueUtilizationRequests(priority, inferencePool string, ratio float64) {
+	llmdFlowControlQueueUtilizationRequests.WithLabelValues(priority, inferencePool).Set(ratio)
+}
+
+// RecordFlowControlQueueUtilizationBytes sets the byte-size queue utilization ratio (occupancy/limit, 0.0-1.0) for a
+// priority band, or the global aggregate when priority is "". Callers only invoke this for a bounded dimension, so an
+// unlimited limit yields no series rather than a misleading 0.
+func RecordFlowControlQueueUtilizationBytes(priority, inferencePool string, ratio float64) {
+	llmdFlowControlQueueUtilizationBytes.WithLabelValues(priority, inferencePool).Set(ratio)
+}
+
 // IncFlowControlRequestsTotal increments the total request counter for a given outcome.
 func IncFlowControlRequestsTotal(outcome, priority, inferencePool string) {
 	llmdFlowControlRequestsTotal.WithLabelValues(outcome, priority, inferencePool).Inc()

release-notes.d/unreleased/2190.md

@@ -0,0 +1,7 @@
+---
+pr: 2190
+url: https://github.com/llm-d/llm-d-router/pull/2190
+author: sudoalok
+date: 2026-07-25
+---
+Added flow control queue utilization gauges: `llm_d_epp_flow_control_queue_utilization_requests` and `llm_d_epp_flow_control_queue_utilization_bytes`, reporting queue occupancy as a fraction of the configured limit (0.0-1.0) per priority band, plus a global aggregate series (priority="") when a global limit is configured. Operators can alert on "queue at N% of its limit" directly, without joining config values into PromQL.