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

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

Author: sudoalok

docs/metrics.md

@@ -256,6 +256,29 @@ Exposed when the `flowControl` feature gate is enabled.
 *   **Usage:** When saturation reaches the usage limit threshold, the dispatch cycle skips
     dispatching and requests remain queued. Sustained 1.0 indicates all backends are at capacity.
 
+#### `flow_control_capacity_utilization_requests`
+
+*   **Type:** Gauge
+*   **Labels:** `priority`, `inference_pool`
+*   **Description:** Fraction of the configured request-count capacity currently occupied (0.0-1.0),
+    aggregated over every flow in a priority band. This is not a per-flow-queue metric: `priority`
+    identifies the band, and `priority=""` carries the aggregate across all bands (emitted only when
+    a global request-count capacity is configured). A dimension with no configured capacity is
+    omitted rather than reported as 0.
+*   **Usage:** Lets operators alert on "the band is at N% of its request limit" without joining
+    configured `maxRequests` values into the query. Sustained values near 1.0 precede
+    `flow_control_requests_total{outcome="RejectedCapacity"}` rising.
+
+#### `flow_control_capacity_utilization_bytes`
+
+*   **Type:** Gauge
+*   **Labels:** `priority`, `inference_pool`
+*   **Description:** Byte-size counterpart of `flow_control_capacity_utilization_requests`: the
+    fraction of the configured byte-size capacity currently occupied (0.0-1.0), aggregated over every
+    flow in a priority band, with `priority=""` for the aggregate across all bands.
+*   **Usage:** Memory-pressure equivalent of the request-count ratio; a band can hit its `maxBytes`
+    ceiling long before its `maxRequests` one when payloads are large.
+
 #### `flow_control_requests_total`
 
 *   **Type:** Counter

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

@@ -356,6 +356,37 @@ func (p *Processor) hasCapacity(priority int, itemByteSize uint64) (bool, contra
 	return true, stats
 }
 
+// recordCapacityUtilization emits occupancy/configured-capacity ratio gauges per priority band (aggregated over every
+// flow in the band, never per flow queue) and, when a global capacity is configured, the aggregate across all bands
+// (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 capacity (capacity 0) is omitted rather than reported as a misleading 0.
+func (p *Processor) recordCapacityUtilization() {
+	stats := p.registry.Stats()
+
+	for priority, band := range stats.PerPriorityBandStats {
+		priorityStr := strconv.Itoa(priority)
+		if band.CapacityRequests > 0 {
+			metrics.RecordFlowControlCapacityUtilizationRequests(priorityStr, p.poolName,
+				float64(band.Len)/float64(band.CapacityRequests))
+		}
+		if band.CapacityBytes > 0 {
+			metrics.RecordFlowControlCapacityUtilizationBytes(priorityStr, p.poolName,
+				float64(band.ByteSize)/float64(band.CapacityBytes))
+		}
+	}
+
+	// Aggregate across all bands, only when a global capacity is configured.
+	if stats.TotalCapacityRequests > 0 {
+		metrics.RecordFlowControlCapacityUtilizationRequests("", p.poolName,
+			float64(stats.TotalLen)/float64(stats.TotalCapacityRequests))
+	}
+	if stats.TotalCapacityBytes > 0 {
+		metrics.RecordFlowControlCapacityUtilizationBytes("", 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 +411,9 @@ func (p *Processor) dispatchCycle(ctx context.Context) bool {
 	// Record pool saturation metric
 	metrics.RecordFlowControlPoolSaturation(p.poolName, saturation)
 
+	// Record capacity utilization ratios (the demand-side twin of saturation) from the same periodic sample.
+	p.recordCapacityUtilization()
+
 	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,10 @@ func TestFlowControlMetricsEmitted(t *testing.T) {
 	eppmetrics.Register()
 
 	detector := newBlockedDetector()
-	h := newHarness(t, harnessOpts{detector: detector})
+	// bandMaxRequests and maxRequests bound the band and the registry as a whole, so both the
+	// per-band ratio and the all-bands aggregate are computable (occupancy/configured capacity);
+	// without a configured capacity the ratio series is intentionally omitted.
+	h := newHarness(t, harnessOpts{detector: detector, bandMaxRequests: 10, maxRequests: 4})
 
 	key := flowcontrol.FlowKey{ID: "metrics-flow", Priority: 0}
 
@@ -1186,6 +1190,20 @@ 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")
 
+	// Both dimensions are bounded (bandMaxRequests=10, maxRequests=4 above), so with exactly one
+	// request queued the ratios are occupancy/configured capacity: 1/10 for the band and 1/4 for the
+	// all-bands aggregate. 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 capacityUtilizationGauge(t, priorityStr) > 0
+	}, time.Second, time.Millisecond,
+		"band utilization ratio should be > 0 while a request is actively queued")
+	require.InDelta(t, 0.1, capacityUtilizationGauge(t, priorityStr), 1e-9,
+		"band ratio should equal occupancy/capacity (1 queued / bandMaxRequests=10)")
+	require.InDelta(t, 0.25, capacityUtilizationGauge(t, globalPriorityLabel), 1e-9,
+		"aggregate ratio should equal occupancy/global capacity (1 queued / maxRequests=4)")
+
 	// Unblock the detector so the request finalizes deterministically via dispatch.
 	detector.Unblock(1)
 
@@ -1201,6 +1219,37 @@ 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 capacityUtilizationGauge(t, priorityStr) == 0 &&
+			capacityUtilizationGauge(t, globalPriorityLabel) == 0
+	}, time.Second, time.Millisecond,
+		"band and aggregate utilization ratios should return to 0 after the queue drains")
+}
+
+// globalPriorityLabel is the priority label value carried by the all-bands aggregate series.
+const globalPriorityLabel = ""
+
+// capacityUtilizationGauge returns the request-count capacity utilization ratio recorded for the given
+// priority band, or -1 if no series exists for that band.
+func capacityUtilizationGauge(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_capacity_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"},
 	)
 
+	llmdFlowControlCapacityUtilizationRequests = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Subsystem: LLMDRouterEndpointPickerSubsystem,
+			Name:      "flow_control_capacity_utilization_requests",
+			Help:      metricsutil.HelpMsgWithStability("Fraction of the configured request-count capacity currently occupied, aggregated over every flow in a priority band (0.0-1.0). The aggregate across all bands is reported with priority=\"\" and only when a global request-count capacity is configured; a dimension with no configured capacity is omitted rather than reported as 0.", compbasemetrics.ALPHA),
+		},
+		[]string{"priority", "inference_pool"},
+	)
+
+	llmdFlowControlCapacityUtilizationBytes = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Subsystem: LLMDRouterEndpointPickerSubsystem,
+			Name:      "flow_control_capacity_utilization_bytes",
+			Help:      metricsutil.HelpMsgWithStability("Fraction of the configured byte-size capacity currently occupied, aggregated over every flow in a priority band (0.0-1.0). The aggregate across all bands is reported with priority=\"\" and only when a global byte-size capacity is configured; a dimension with no configured capacity 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(llmdFlowControlCapacityUtilizationRequests)
+		metrics.Registry.MustRegister(llmdFlowControlCapacityUtilizationBytes)
 		metrics.Registry.MustRegister(flowControlRequestEnqueueDuration)
 		metrics.Registry.MustRegister(llmdFlowControlRequestEnqueueDuration)
 		metrics.Registry.MustRegister(llmdFlowControlRequestsTotal)
@@ -542,6 +544,8 @@ func Reset() {
 	llmdFlowControlQueueBytes.Reset()
 	flowControlPoolSaturation.Reset()
 	llmdFlowControlPoolSaturation.Reset()
+	llmdFlowControlCapacityUtilizationRequests.Reset()
+	llmdFlowControlCapacityUtilizationBytes.Reset()
 	flowControlRequestEnqueueDuration.Reset()
 	llmdFlowControlRequestEnqueueDuration.Reset()
 	flowControlDispatchCycleDuration.Reset()
@@ -906,6 +910,20 @@ func RecordFlowControlPoolSaturation(inferencePool string, saturation float64) {
 	llmdFlowControlPoolSaturation.WithLabelValues(inferencePool).Set(saturation)
 }
 
+// RecordFlowControlCapacityUtilizationRequests sets the request-count capacity utilization ratio
+// (occupancy/configured capacity, 0.0-1.0) for a priority band, or the aggregate across all bands when priority is "".
+// Callers only invoke this for a bounded dimension, so an unbounded one yields no series rather than a misleading 0.
+func RecordFlowControlCapacityUtilizationRequests(priority, inferencePool string, ratio float64) {
+	llmdFlowControlCapacityUtilizationRequests.WithLabelValues(priority, inferencePool).Set(ratio)
+}
+
+// RecordFlowControlCapacityUtilizationBytes sets the byte-size capacity utilization ratio (occupancy/configured
+// capacity, 0.0-1.0) for a priority band, or the aggregate across all bands when priority is "". Callers only invoke
+// this for a bounded dimension, so an unbounded one yields no series rather than a misleading 0.
+func RecordFlowControlCapacityUtilizationBytes(priority, inferencePool string, ratio float64) {
+	llmdFlowControlCapacityUtilizationBytes.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()