fix(flowcontrol): surface stale-metrics saturation and fix pool_saturation help text

The utilization detector scores endpoints with missing or stale metrics
as fully saturated, and both detectors score an empty pool as 1.0.
Dispatch halts at saturation 1.0, so a fleet-wide scrape failure stops
all dispatch and nothing points at metrics collection as the cause. The
pool_saturation help text also claimed a 0.0-1.0 range, but both
detectors return values above 1 when the pool is oversubscribed.

- Add flow_control_stale_endpoints{detector}: candidate endpoints
  scored as saturated because their metrics are missing or older than
  the staleness threshold. Emitted under the llm_d_epp prefix only.
- Log the stale condition from the utilization detector at default
  verbosity, at most once per 30s. Saturation runs every dispatch cycle
  (~1ms), so the log is bounded by time, not call count.
- Fix the pool_saturation help text on both gauges: 1.0 is the gating
  set point, values above 1.0 measure oversubscription, and empty-pool
  and stale-metrics conditions read as 1.0 (fail-closed). Values stay
  unclamped.
- Keep fail-closed for stale metrics on a non-empty pool and document
  the decision in the detector README: admitting blind on missing data
  would overload model servers with no backpressure signal at all.

Fixes #2100
Part of #1187

Signed-off-by: Luke Van Drie <lukevandrie@google.com>

Author: LukeAVanDrie

docs/metrics.md

@@ -242,10 +242,25 @@ Exposed when the `flowControl` feature gate is enabled.
 
 *   **Type:** Gauge
 *   **Labels:** `inference_pool`
-*   **Description:** Current saturation level of the inference pool (0.0 = empty, 1.0 = fully
-    saturated).
+*   **Description:** Pool saturation signal gating dispatch. 1.0 is the gating set point; values
+    above 1.0 indicate the magnitude of oversubscription past it (deliberately not clamped). An
+    empty pool reads as 1.0, and with the default utilization detector, endpoints with missing or
+    stale metrics score as fully saturated (fail-closed).
 *   **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.
+    dispatching and requests remain queued. A reading pinned at exactly 1.0 can be fail-closed
+    stale-metrics or an empty pool rather than genuine overload (which typically reads above 1.0);
+    check `flow_control_stale_endpoints` to disambiguate.
+
+#### `flow_control_stale_endpoints`
+
+*   **Type:** Gauge
+*   **Labels:** `detector`
+*   **Description:** Number of candidate endpoints whose metrics are missing or older than the
+    staleness threshold, as of the most recent saturation evaluation. Recorded by the utilization
+    saturation detector; emitted under the `llm_d_epp` prefix only (no deprecated
+    `inference_extension_*` twin).
+*   **Usage:** A nonzero value during a dispatch stall indicates a model-server metrics collection
+    problem (scrape path, port, TLS, auth) rather than genuine overload.
 
 #### `flow_control_requests_total`
 

pkg/epp/framework/plugins/flowcontrol/saturationdetector/utilization/README.md

@@ -23,7 +23,14 @@ The global pool saturation is then evaluated across all candidate endpoints as a
     PoolSaturation = Average(EndpointScore)
 
 **Heterogeneous Deployments:** Because this detector calculates saturation as an unweighted average of individual endpoint scores, it treats all endpoints equally regardless of their physical capacity. In deployments with heterogeneous compute (e.g., mixing H100 and L4 nodes), a small, saturated endpoint has the exact same impact on global backpressure as a massive, saturated endpoint. Contrast this with the Concurrency Detector, which evaluates saturation as a single aggregate fraction, biasing toward larger endpoints.
-*Note: Endpoints with missing or stale metrics are aggressively scored as 100% saturated.*
+*Note: Endpoints with missing or stale metrics are aggressively scored as 100% saturated (fail-closed).*
+
+**Operational dependency:** because stale endpoints score as saturated, Flow Control dispatch depends on the
+health of model-server metrics collection (scrape path, port, TLS, auth). A fleet-wide scrape outage pins
+`flow_control_pool_saturation` at 1.0 and halts dispatch entirely. The `flow_control_stale_endpoints` gauge and a
+rate-limited detector log distinguish this from genuine overload: stale endpoints read exactly 1.0, while genuine
+oversubscription typically reads above 1.0. This fail-closed posture is deliberate — admitting blind on missing
+data risks overloading model servers with no backpressure signal at all.
 
 ### Role in Scheduling (The Traffic Shaper)
 The detector implements the `Filter` interface to protect individual endpoints. It removes endpoints from candidate lists if their telemetry is stale, or if they exceed specific safety limits:

pkg/epp/framework/plugins/flowcontrol/saturationdetector/utilization/detector.go

@@ -24,6 +24,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"sync/atomic"
 	"time"
 
 	"github.com/go-logr/logr"
@@ -34,11 +35,17 @@ import (
 	"github.com/llm-d/llm-d-router/pkg/epp/framework/interface/flowcontrol"
 	fwkplugin "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/plugin"
 	fwksched "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/scheduling"
+	"github.com/llm-d/llm-d-router/pkg/epp/metrics"
 )
 
 const (
 	// UtilizationDetectorType is the unique identifier for this plugin.
 	UtilizationDetectorType = "utilization-detector"
+
+	// staleWarnInterval bounds how often the detector logs about endpoints with missing or stale
+	// metrics. Saturation is evaluated every dispatch cycle (~1ms), so the condition must be logged
+	// on a time budget, never per evaluation.
+	staleWarnInterval = 30 * time.Second
 )
 
 // UtilizationDetectorFactory instantiates the detector plugin using the provided JSON parameters.
@@ -69,6 +76,11 @@ var (
 type Detector struct {
 	config    Config
 	typedName fwkplugin.TypedName
+	logger    logr.Logger
+
+	// lastStaleWarnNanos is the wall-clock time (UnixNano) of the most recent stale-metrics log,
+	// accessed atomically because Saturation may be called concurrently.
+	lastStaleWarnNanos atomic.Int64
 }
 
 // NewDetector creates a new instance of the Utilization Detector.
@@ -95,6 +107,7 @@ func NewDetector(name string, cfg Config, logger logr.Logger) *Detector {
 	return &Detector{
 		config:    cfg,
 		typedName: typedName,
+		logger:    pluginLogger,
 	}
 }
 
@@ -114,28 +127,61 @@ func (d *Detector) TypedName() fwkplugin.TypedName {
 //	PodScore = Max(WaitingQueue / QueueThreshold, KVCacheUsage / KVCacheThreshold)
 func (d *Detector) Saturation(_ context.Context, candidates []datalayer.Endpoint) float64 {
 	if len(candidates) == 0 {
+		// No candidates means no stale endpoints. Keeping the gauge current here prevents a stale
+		// reading from a previous evaluation misattributing an empty-pool stall to a metrics
+		// collection failure.
+		metrics.RecordFlowControlStaleEndpoints(d.typedName.Name, 0)
 		return 1.0
 	}
 
 	var totalScore float64
+	staleCount := 0
 	for _, e := range candidates {
-		metrics := e.GetMetrics()
+		podMetrics := e.GetMetrics()
 
-		if metrics == nil || time.Since(metrics.UpdateTime) > d.config.MetricsStalenessThreshold {
+		if podMetrics == nil || time.Since(podMetrics.UpdateTime) > d.config.MetricsStalenessThreshold {
+			// Fail closed: an endpoint whose metrics are missing or stale scores as fully saturated. A
+			// fleet-wide metrics collection failure therefore halts dispatch entirely rather than
+			// admitting blind; the gauge and the rate-limited log below exist so operators can tell that
+			// stall apart from genuine overload (which typically scores above 1.0).
 			totalScore += 1.0
+			staleCount++
 			continue
 		}
 
-		qRatio := float64(metrics.WaitingQueueSize) / float64(d.config.QueueDepthThreshold)
-		kvRatio := metrics.KVCacheUsagePercent / d.config.KVCacheUtilThreshold
+		qRatio := float64(podMetrics.WaitingQueueSize) / float64(d.config.QueueDepthThreshold)
+		kvRatio := podMetrics.KVCacheUsagePercent / d.config.KVCacheUtilThreshold
 
 		// Roofline Analysis: The pod is saturated if either resource is exhausted.
 		totalScore += max(qRatio, kvRatio)
 	}
 
+	metrics.RecordFlowControlStaleEndpoints(d.typedName.Name, staleCount)
+	if staleCount > 0 {
+		d.maybeLogStaleEndpoints(staleCount, len(candidates))
+	}
+
 	return totalScore / float64(len(candidates))
 }
 
+// maybeLogStaleEndpoints logs the stale-metrics condition at most once per staleWarnInterval.
+func (d *Detector) maybeLogStaleEndpoints(staleCount, total int) {
+	now := time.Now().UnixNano()
+	last := d.lastStaleWarnNanos.Load()
+	if now-last < int64(staleWarnInterval) {
+		return
+	}
+	if !d.lastStaleWarnNanos.CompareAndSwap(last, now) {
+		return // Another goroutine logged concurrently.
+	}
+	d.logger.V(logutil.DEFAULT).Info(
+		"Endpoints with missing or stale metrics are scored as fully saturated (fail-closed); "+
+			"if dispatch is stalled, check model-server metrics collection (scrape path, port, TLS, auth)",
+		"staleEndpoints", staleCount,
+		"totalEndpoints", total,
+		"metricsStalenessThreshold", d.config.MetricsStalenessThreshold.String())
+}
+
 // Filter blocks traffic to specific pods that are physically saturated or exceeding their safety limits.
 //
 // It applies a relaxed limit (Threshold * (1 + Headroom)) to allow for scheduling flexibility and burst tolerance.
@@ -151,12 +197,12 @@ func (d *Detector) Filter(
 	filtered := make([]fwksched.Endpoint, 0, len(endpoints))
 
 	for _, endpoint := range endpoints {
-		metrics := endpoint.GetMetrics()
-		if metrics == nil || time.Since(metrics.UpdateTime) > d.config.MetricsStalenessThreshold {
+		podMetrics := endpoint.GetMetrics()
+		if podMetrics == nil || time.Since(podMetrics.UpdateTime) > d.config.MetricsStalenessThreshold {
 			continue
 		}
 
-		if float64(metrics.WaitingQueueSize) < qLimit && metrics.KVCacheUsagePercent < kvLimit {
+		if float64(podMetrics.WaitingQueueSize) < qLimit && podMetrics.KVCacheUsagePercent < kvLimit {
 			filtered = append(filtered, endpoint)
 		}
 	}

pkg/epp/framework/plugins/flowcontrol/saturationdetector/utilization/detector_test.go

@@ -24,10 +24,12 @@ import (
 	"github.com/go-logr/logr"
 	"github.com/stretchr/testify/require"
 	"k8s.io/apimachinery/pkg/types"
+	ctrlmetrics "sigs.k8s.io/controller-runtime/pkg/metrics"
 
 	fwkdl "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/datalayer"
 	fwkplugin "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/plugin"
 	fwksched "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/scheduling"
+	eppmetrics "github.com/llm-d/llm-d-router/pkg/epp/metrics"
 )
 
 func makePodMetric(name string, queueDepth int, kvUsage float64, updateTime time.Time) fwkdl.Endpoint {
@@ -370,3 +372,70 @@ func TestDetector_Filter(t *testing.T) {
 		})
 	}
 }
+
+// TestDetector_StaleEndpointObservability verifies that Saturation records the stale-endpoint
+// gauge (keyed by detector name) and that the stale-metrics log is time-bounded.
+func TestDetector_StaleEndpointObservability(t *testing.T) {
+	t.Parallel()
+
+	eppmetrics.Register()
+
+	// A wide staleness threshold keeps fresh pods deterministically fresh on slow CI machines;
+	// the stale pod is stamped far past the threshold.
+	config := Config{
+		QueueDepthThreshold:       5,
+		KVCacheUtilThreshold:      0.90,
+		MetricsStalenessThreshold: time.Hour,
+	}
+	// A unique detector name isolates this test's gauge series from parallel tests.
+	detectorName := "stale-observability-test"
+	detector := NewDetector(detectorName, config, logr.Discard())
+
+	staleGaugeValue := func() float64 {
+		families, err := ctrlmetrics.Registry.Gather()
+		require.NoError(t, err)
+		for _, f := range families {
+			if f.GetName() != "llm_d_epp_flow_control_stale_endpoints" {
+				continue
+			}
+			for _, m := range f.GetMetric() {
+				for _, l := range m.GetLabel() {
+					if l.GetName() == "detector" && l.GetValue() == detectorName {
+						return m.GetGauge().GetValue()
+					}
+				}
+			}
+		}
+		return -1 // Series absent.
+	}
+
+	baseTime := time.Now()
+	pods := []fwkdl.Endpoint{
+		makePodMetric("fresh", 1, 0.1, baseTime),
+		makePodMetric("stale", 1, 0.1, baseTime.Add(-2*time.Hour)),
+		fwkdl.NewEndpoint(&fwkdl.EndpointMetadata{
+			NamespacedName: types.NamespacedName{Name: "nil-metrics", Namespace: "ns1"},
+		}, nil),
+	}
+
+	detector.Saturation(context.Background(), pods)
+	require.Equal(t, 2.0, staleGaugeValue(), "stale and nil-metrics endpoints should both be counted")
+	firstWarn := detector.lastStaleWarnNanos.Load()
+	require.NotZero(t, firstWarn, "first stale observation should record a log timestamp")
+
+	// A second observation within staleWarnInterval must not log again.
+	detector.Saturation(context.Background(), pods)
+	require.Equal(t, firstWarn, detector.lastStaleWarnNanos.Load(),
+		"stale-metrics logging must be time-bounded, not per evaluation")
+
+	// An empty candidate list has no stale endpoints; the gauge must not stay pinned at its last
+	// value, or an empty-pool stall reads as a metrics collection failure.
+	detector.Saturation(context.Background(), []fwkdl.Endpoint{})
+	require.Equal(t, 0.0, staleGaugeValue(), "gauge should read zero for an empty candidate list")
+
+	// Re-observe staleness, then confirm fresh metrics clear it.
+	detector.Saturation(context.Background(), pods)
+	require.Equal(t, 2.0, staleGaugeValue(), "staleness should be re-observed after the empty list")
+	detector.Saturation(context.Background(), []fwkdl.Endpoint{makePodMetric("fresh", 1, 0.1, time.Now())})
+	require.Equal(t, 0.0, staleGaugeValue(), "gauge should return to zero when staleness clears")
+}

pkg/epp/metrics/llm_d_router_metrics.go

@@ -351,11 +351,30 @@ var (
 		prometheus.GaugeOpts{
 			Subsystem: LLMDRouterEndpointPickerSubsystem,
 			Name:      "flow_control_pool_saturation",
-			Help:      metricsutil.HelpMsgWithStability("Current saturation level of the inference pool (0.0 = empty, 1.0 = fully saturated).", compbasemetrics.ALPHA),
+			Help: metricsutil.HelpMsgWithStability(
+				"Pool saturation signal gating Flow Control dispatch. 1.0 is the gating set point; values above 1.0 "+
+					"indicate the magnitude of oversubscription past it. An empty pool reads as 1.0. With the default "+
+					"utilization detector, endpoints with missing or stale metrics score as fully saturated "+
+					"(fail-closed; see flow_control_stale_endpoints).",
+				compbasemetrics.ALPHA),
 		},
 		[]string{"inference_pool"},
 	)
 
+	llmdFlowControlStaleEndpoints = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Subsystem: LLMDRouterEndpointPickerSubsystem,
+			Name:      "flow_control_stale_endpoints",
+			Help: metricsutil.HelpMsgWithStability(
+				"Number of candidate endpoints whose metrics are missing or older than the staleness threshold, as of "+
+					"the most recent saturation evaluation. Recorded by the utilization saturation detector, which scores "+
+					"these endpoints as fully saturated in flow_control_pool_saturation (fail-closed): a nonzero value "+
+					"during a dispatch stall indicates a metrics collection problem rather than genuine overload.",
+				compbasemetrics.ALPHA),
+		},
+		[]string{"detector"},
+	)
+
 	llmdFlowControlRequestsTotal = prometheus.NewCounterVec(
 		prometheus.CounterOpts{
 			Subsystem: LLMDRouterEndpointPickerSubsystem,

pkg/epp/metrics/metrics.go

@@ -368,7 +368,13 @@ var (
 		prometheus.GaugeOpts{
 			Subsystem: inferenceExtension,
 			Name:      "flow_control_pool_saturation",
-			Help:      metricsutil.HelpMsgWithStability("[Deprecated: Use llm_d_epp_flow_control_pool_saturation] Current saturation level of the inference pool (0.0 = empty, 1.0 = fully saturated).", compbasemetrics.ALPHA),
+			Help: metricsutil.HelpMsgWithStability(
+				"[Deprecated: Use llm_d_epp_flow_control_pool_saturation] Pool saturation signal gating Flow Control "+
+					"dispatch. 1.0 is the gating set point; values above 1.0 indicate the magnitude of oversubscription "+
+					"past it. An empty pool reads as 1.0. With the default utilization detector, endpoints with missing "+
+					"or stale metrics score as fully saturated (fail-closed; see "+
+					"llm_d_epp_flow_control_stale_endpoints).",
+				compbasemetrics.ALPHA),
 		},
 		[]string{"inference_pool"},
 	)
@@ -470,6 +476,9 @@ func Register(customCollectors ...prometheus.Collector) {
 		metrics.Registry.MustRegister(llmdFlowControlQueueBytes)
 		metrics.Registry.MustRegister(flowControlPoolSaturation)
 		metrics.Registry.MustRegister(llmdFlowControlPoolSaturation)
+		// No deprecated inference_extension twin: new flow control metrics are emitted under the
+		// llm_d_epp prefix only.
+		metrics.Registry.MustRegister(llmdFlowControlStaleEndpoints)
 		metrics.Registry.MustRegister(flowControlRequestEnqueueDuration)
 		metrics.Registry.MustRegister(llmdFlowControlRequestEnqueueDuration)
 		metrics.Registry.MustRegister(llmdFlowControlRequestsTotal)
@@ -538,6 +547,7 @@ func Reset() {
 	llmdFlowControlQueueBytes.Reset()
 	flowControlPoolSaturation.Reset()
 	llmdFlowControlPoolSaturation.Reset()
+	llmdFlowControlStaleEndpoints.Reset()
 	flowControlRequestEnqueueDuration.Reset()
 	llmdFlowControlRequestEnqueueDuration.Reset()
 	flowControlDispatchCycleDuration.Reset()
@@ -890,6 +900,12 @@ func RecordFlowControlPoolSaturation(inferencePool string, saturation float64) {
 	llmdFlowControlPoolSaturation.WithLabelValues(inferencePool).Set(saturation)
 }
 
+// RecordFlowControlStaleEndpoints records how many candidate endpoints the given saturation
+// detector scored as fully saturated because their metrics were missing or stale.
+func RecordFlowControlStaleEndpoints(detector string, count int) {
+	llmdFlowControlStaleEndpoints.WithLabelValues(detector).Set(float64(count))
+}
+
 // IncFlowControlRequestsTotal increments the total request counter for a given outcome.
 func IncFlowControlRequestsTotal(outcome, priority, inferencePool string) {
 	llmdFlowControlRequestsTotal.WithLabelValues(outcome, priority, inferencePool).Inc()