feat: pool-wide saturation computation + expose as gauge (kubernetes-sigs#2343)

* refactor: pool-wide saturation metrics + prometheus

Signed-off-by: Edoardo Vacchi <evacchi@users.noreply.github.com>

* remove redundant field

Signed-off-by: Edoardo Vacchi <evacchi@users.noreply.github.com>

* address review comments

Signed-off-by: Edoardo Vacchi <evacchi@users.noreply.github.com>

* rebase issue

Signed-off-by: Edoardo Vacchi <evacchi@users.noreply.github.com>

* fix rebase

Signed-off-by: Edoardo Vacchi <evacchi@users.noreply.github.com>

---------

Signed-off-by: Edoardo Vacchi <evacchi@users.noreply.github.com>

Author: evacchi

cmd/epp/runner/runner.go

@@ -341,6 +341,7 @@ func (r *Runner) setup(ctx context.Context, cfg *rest.Config, opts *runserver.Op
 		}
 		fc, err := fccontroller.NewFlowController(
 			ctx,
+			opts.PoolName,
 			eppConfig.FlowControlConfig.Controller,
 			registry, saturationDetector,
 			locator,

pkg/epp/flowcontrol/controller/controller.go

@@ -128,6 +128,7 @@ type flowControllerOption func(*FlowController)
 // The provided context governs the lifecycle of the controller and all its workers.
 func NewFlowController(
 	ctx context.Context,
+	poolName string,
 	config *Config,
 	registry contracts.FlowRegistry,
 	sd contracts.SaturationDetector,
@@ -156,6 +157,7 @@ func NewFlowController(
 	) shardProcessor {
 		return internal.NewShardProcessor(
 			ctx,
+			poolName,
 			shard,
 			saturationDetector,
 			podLocator,

pkg/epp/flowcontrol/controller/controller_test.go

@@ -132,7 +132,7 @@ func newUnitHarness(
 		withClock(harnessOpts.clock),
 		withShardProcessorFactory(mockProcessorFactory.new),
 	}
-	fc, err := NewFlowController(ctx, cfg, registry, mockDetector, mockPodLocator, fcOpts...)
+	fc, err := NewFlowController(ctx, "test-pool", cfg, registry, mockDetector, mockPodLocator, fcOpts...)
 	require.NoError(t, err, "failed to create FlowController for unit test harness")
 
 	h := &testHarness{
@@ -167,7 +167,7 @@ func newIntegrationHarness(t *testing.T, ctx context.Context, cfg *Config, regis
 		withRegistryClient(registry),
 		withClock(mockClock),
 	}
-	fc, err := NewFlowController(ctx, cfg, registry, mockDetector, mockPodLocator, opts...)
+	fc, err := NewFlowController(ctx, "test-pool", cfg, registry, mockDetector, mockPodLocator, opts...)
 	require.NoError(t, err, "failed to create FlowController for integration test harness")
 
 	h := &testHarness{

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

@@ -65,6 +65,7 @@ var ErrProcessorBusy = errors.New("shard processor is busy")
 // loop is the sole writer for all state-mutating operations. This makes complex transactions (like capacity checks)
 // inherently atomic without coarse-grained locks.
 type ShardProcessor struct {
+	poolName             string
 	shard                contracts.RegistryShard
 	saturationDetector   contracts.SaturationDetector
 	podLocator           contracts.PodLocator
@@ -87,6 +88,7 @@ type ShardProcessor struct {
 // NewShardProcessor creates a new ShardProcessor instance.
 func NewShardProcessor(
 	ctx context.Context,
+	poolName string,
 	shard contracts.RegistryShard,
 	saturationDetector contracts.SaturationDetector,
 	podLocator contracts.PodLocator,
@@ -97,6 +99,7 @@ func NewShardProcessor(
 ) *ShardProcessor {
 	return &ShardProcessor{
 		shard:                shard,
+		poolName:             poolName,
 		saturationDetector:   saturationDetector,
 		podLocator:           podLocator,
 		clock:                clock,
@@ -311,6 +314,20 @@ func (sp *ShardProcessor) dispatchCycle(ctx context.Context) bool {
 		metrics.RecordFlowControlDispatchCycleDuration(time.Since(dispatchCycleStart))
 	}()
 
+	pool := sp.podLocator.Locate(ctx, nil)
+	saturation := sp.saturationDetector.Saturation(ctx, pool)
+
+	// Record pool saturation metric
+	metrics.RecordFlowControlPoolSaturation(sp.poolName, saturation)
+
+	// --- Viability Check (Pool-Wide Saturation) ---
+	if saturation >= 1.0 {
+		sp.logger.V(logutil.DEBUG).Info("Pool is saturated; enforcing HoL blocking.",
+			"poolName", sp.poolName)
+		// Short-circuit
+		return false
+	}
+
 	for _, priority := range sp.shard.AllOrderedPriorityLevels() {
 		originalBand, err := sp.shard.PriorityBandAccessor(priority)
 		if err != nil {
@@ -328,16 +345,7 @@ func (sp *ShardProcessor) dispatchCycle(ctx context.Context) bool {
 			continue
 		}
 
-		// --- Viability Check (Saturation/HoL Blocking) ---
 		req := item.OriginalRequest()
-		candidates := sp.podLocator.Locate(ctx, req.GetMetadata())
-		if sp.saturationDetector.Saturation(ctx, candidates) >= 1.0 {
-			sp.logger.V(logutil.DEBUG).Info("Policy's chosen item is saturated; enforcing HoL blocking.",
-				"flowKey", req.FlowKey(), "reqID", req.ID(), "priorityName", originalBand.PriorityName())
-			// Stop the dispatch cycle entirely to respect strict policy decision and prevent priority inversion where
-			// lower-priority work might exacerbate the saturation affecting high-priority work.
-			return false
-		}
 
 		// --- Dispatch ---
 		if err := sp.dispatchItem(item); err != nil {

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

@@ -120,6 +120,7 @@ func newTestHarness(t *testing.T, expiryCleanupInterval time.Duration) *testHarn
 
 	h.processor = NewShardProcessor(
 		h.ctx,
+		"test-pool",
 		h,
 		h.saturationDetector,
 		h.podLocator,
@@ -546,6 +547,7 @@ func TestShardProcessor(t *testing.T) {
 			// A successful test is one that completes without panicking.
 			time.Sleep(50 * time.Millisecond)
 		})
+
 	})
 
 	t.Run("Unit", func(t *testing.T) {

pkg/epp/metrics/metrics.go

@@ -431,6 +431,15 @@ var (
 		},
 		append([]string{"fairness_id", "priority", "inference_pool"}, modelLabels...),
 	)
+
+	flowControlPoolSaturation = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Subsystem: inferenceExtension,
+			Name:      "flow_control_pool_saturation",
+			Help:      metricsutil.HelpMsgWithStability("Current saturation level of the inference pool (0.0 = empty, 1.0 = fully saturated).", compbasemetrics.ALPHA),
+		},
+		[]string{"inference_pool"},
+	)
 )
 
 // --- Inference Model Rewrite Metrics ---
@@ -487,6 +496,7 @@ func Register(customCollectors ...prometheus.Collector) {
 		metrics.Registry.MustRegister(flowControlDispatchCycleDuration)
 		metrics.Registry.MustRegister(flowControlQueueSize)
 		metrics.Registry.MustRegister(flowControlQueueBytes)
+		metrics.Registry.MustRegister(flowControlPoolSaturation)
 		metrics.Registry.MustRegister(flowControlRequestEnqueueDuration)
 		metrics.Registry.MustRegister(inferenceModelRewriteDecisionsTotal)
 		for _, collector := range customCollectors {
@@ -535,6 +545,7 @@ func Reset() {
 	flowControlRequestQueueDuration.Reset()
 	flowControlQueueSize.Reset()
 	flowControlQueueBytes.Reset()
+	flowControlPoolSaturation.Reset()
 	flowControlRequestEnqueueDuration.Reset()
 	inferenceModelRewriteDecisionsTotal.Reset()
 }
@@ -849,6 +860,11 @@ func SubFlowControlQueueBytes(fairnessID, priority, inferencePool, modelName, ta
 	flowControlQueueBytes.WithLabelValues(fairnessID, priority, inferencePool, modelName, targetModelName).Sub(float64(bytes))
 }
 
+// RecordFlowControlPoolSaturation records the current saturation level for an inference pool.
+func RecordFlowControlPoolSaturation(inferencePool string, saturation float64) {
+	flowControlPoolSaturation.WithLabelValues(inferencePool).Set(saturation)
+}
+
 // SetTTFTSLOThreshold sets the TTFT SLO threshold for a model.
 // This allows dynamic threshold management and makes the threshold visible in metrics.
 func SetTTFTSLOThreshold(modelName, targetModelName string, threshold float64) {

pkg/epp/metrics/metrics_test.go

@@ -1120,6 +1120,47 @@ func TestFlowControlQueueBytesMetric(t *testing.T) {
 	require.Equal(t, 0.0, val, "Gauge value for non-existent labels should be 0")
 }
 
+func TestFlowControlPoolSaturationMetric(t *testing.T) {
+	Reset()
+
+	const pool = "test-pool"
+
+	// Set saturation to 0.5
+	RecordFlowControlPoolSaturation(pool, 0.5)
+	val, err := testutil.GetGaugeMetricValue(flowControlPoolSaturation.WithLabelValues(pool))
+	require.NoError(t, err, "Failed to get gauge value for pool saturation")
+	require.Equal(t, 0.5, val, "Gauge value should be 0.5")
+
+	// Update saturation to 1.0 (fully saturated)
+	RecordFlowControlPoolSaturation(pool, 1.0)
+	val, err = testutil.GetGaugeMetricValue(flowControlPoolSaturation.WithLabelValues(pool))
+	require.NoError(t, err, "Failed to get gauge value after update")
+	require.Equal(t, 1.0, val, "Gauge value should be 1.0 after update")
+
+	// Update saturation to 0.0 (empty)
+	RecordFlowControlPoolSaturation(pool, 0.0)
+	val, err = testutil.GetGaugeMetricValue(flowControlPoolSaturation.WithLabelValues(pool))
+	require.NoError(t, err, "Failed to get gauge value for empty pool")
+	require.Equal(t, 0.0, val, "Gauge value should be 0.0 for empty pool")
+
+	// Multiple pools
+	RecordFlowControlPoolSaturation("pool-a", 0.3)
+	RecordFlowControlPoolSaturation("pool-b", 0.7)
+
+	valA, err := testutil.GetGaugeMetricValue(flowControlPoolSaturation.WithLabelValues("pool-a"))
+	require.NoError(t, err, "Failed to get gauge value for pool-a")
+	require.Equal(t, 0.3, valA, "Gauge value should be 0.3 for pool-a")
+
+	valB, err := testutil.GetGaugeMetricValue(flowControlPoolSaturation.WithLabelValues("pool-b"))
+	require.NoError(t, err, "Failed to get gauge value for pool-b")
+	require.Equal(t, 0.7, valB, "Gauge value should be 0.7 for pool-b")
+
+	// Non-existent pool
+	val, err = testutil.GetGaugeMetricValue(flowControlPoolSaturation.WithLabelValues("non-existent"))
+	require.NoError(t, err, "Failed to get gauge value for non-existent pool")
+	require.Equal(t, 0.0, val, "Gauge value for non-existent pool should be 0")
+}
+
 func TestInferenceModelRewriteDecisionsTotalMetric(t *testing.T) {
 	Reset()
 

site-src/guides/metrics-and-observability.md

@@ -64,6 +64,7 @@ These metrics provide insights into the [Flow Control layer](flow-control.md) wi
 | inference_extension_flow_control_queue_bytes | Gauge | The current size in bytes of all requests being actively managed by the Flow Control layer. This includes requests from the moment they enter the `EnqueueAndWait` function until they reach a final outcome. | `fairness_id`=&lt;flow-id&gt; <br> `priority`=&lt;flow-priority&gt; <br> `inference_pool`=&lt;pool-name&gt; <br> `model_name`=&lt;model-name&gt; <br> `target_model_name`=&lt;target-model-name&gt; | ALPHA |
 | inference_extension_flow_control_dispatch_cycle_duration_seconds | Histogram | The time taken for each dispatch cycle in the Flow Control layer. |  | ALPHA |
 | inference_extension_flow_control_request_enqueue_duration_seconds | Gauge | The time taken to enqueue requests by the EPP Flow Control layer. | `fairness_id`=&lt;flow-id&gt; <br> `priority`=&lt;flow-priority&gt; <br> `outcome`=&lt;QueueOutcome&gt; | ALPHA |
+| inference_extension_flow_control_pool_saturation | Gauge | Current saturation level of the inference pool (0.0 = empty, 1.0 = fully saturated). | `inference_pool`=&lt;pool-name&gt; | ALPHA |
 
 
 ## Scrape Metrics & Pprof profiles