fix: calculate pool saturation as endpoint average

Pool saturation is defined as average of saturation across each
endpoint, this is exactly how it is computed but in multi-dimensional
saturation cases such as hybrid, this wasn't true. This fixes it.

Signed-off-by: mohammadkhan <mohammadkhan@digitalocean.com>

Author: asharkhan3101

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

@@ -15,7 +15,9 @@ The detector implements the `SaturationDetector` interface to provide a utilizat
 
 In token mode, both numerator and denominator are evaluated in tokens: the aggregate inflight token count divided by the sum of all endpoints' MaxTokenConcurrency.
 
-**Heterogeneous Deployments:** Because this detector calculates saturation globally as a single aggregate fraction, it utilizes an aggregate queueing model. In deployments with heterogeneous compute (e.g., mixing H100 and L4 nodes), this heavily biases the pool saturation metric toward the state of the larger nodes. Contrast this with the Utilization Detector, which evaluates saturation as an unweighted average of individual endpoint scores.
+Hybrid mode is the exception: rather than one aggregate fraction, it evaluates each endpoint's saturation as the larger of its request and token ratios and reports the unweighted average across endpoints. This prevents distinct endpoints saturating on different dimensions from being masked by aggregate ratios that each remain low.
+
+**Heterogeneous Deployments:** Because this detector calculates saturation globally as a single aggregate fraction (in requests and tokens mode), it utilizes an aggregate queueing model. In deployments with heterogeneous compute (e.g., mixing H100 and L4 nodes), this heavily biases the pool saturation metric toward the state of the larger nodes. Contrast this with the Utilization Detector, which evaluates saturation as an unweighted average of individual endpoint scores.
 
 ### Role in Scheduling (The Traffic Shaper)
 The detector implements the `Filter` interface to protect individual endpoints. It removes endpoints from candidate lists if their local inflight count exceeds the safety limit:
@@ -34,7 +36,7 @@ The plugin internally tracks active concurrency by hooking into the request life
 
 The plugin accepts JSON parameters decoding to the following fields:
 
-- `concurrencyMode` (`string`): Evaluation mode. Valid values are `"requests"`, `"tokens"`, or `"hybrid"`. In `"hybrid"` mode both request and token accounting are evaluated and the more constraining dimension wins: pool saturation is the larger of the two ratios, and an endpoint is filtered out when either its request load or its token load reaches the limit. (Default: `"requests"`)
+- `concurrencyMode` (`string`): Evaluation mode. Valid values are `"requests"`, `"tokens"`, or `"hybrid"`. In `"hybrid"` mode both request and token accounting are evaluated. Pool saturation is computed per endpoint as the larger of that endpoint's request and token ratios, then averaged across endpoints, so an endpoint saturated on either dimension is reflected even when distinct endpoints saturate on different dimensions. An endpoint is filtered out when either its request load or its token load reaches the limit. (Default: `"requests"`)
 - `maxConcurrency` (`int64`): Maximum requests in flight. Serves as the "ideal" request capacity for a single endpoint. Must be > 0. (Default: `100`)
 - `maxTokenConcurrency` (`int64`): Maximum tokens in flight. The "tokens" mode equivalent of `maxConcurrency`. Must be > 0. (Default: `1000000`)
 - `headroom` (`float64`): Allowed burst capacity above the ideal threshold, expressed as a fraction (e.g., `0.2` for 20%). Must be >= 0.0. (Default: `0.0`)

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

@@ -121,24 +121,28 @@ func (d *detector) getLoad(m datalayer.AttributeMap) *attrconcurrency.InFlightLo
 
 // Saturation calculates the saturation level of the pool.
 //
-// It returns an aggregate saturation signal where each dimension is evaluated as:
+// In "requests" and "tokens" mode it returns an aggregate signal, evaluated as:
 //
 //	Saturation = Total Inflight / Total Capacity.
 //
-// In "hybrid" mode both the request and token dimensions are computed and the larger
-// (more saturated) of the two is returned, so the pool reports saturated as soon as
-// either dimension is exhausted.
+// In "hybrid" mode saturation is instead evaluated per endpoint as
+// max(requestRatio, tokenRatio) and averaged across endpoints. Evaluating each
+// endpoint independently ensures an endpoint saturated on either dimension is
+// reflected in the pool signal.
 func (d *detector) Saturation(_ context.Context, endpoints []datalayer.Endpoint) float64 {
 	if len(endpoints) == 0 {
 		return 1.0
 	}
 
 	var reqInflight, reqCapacity, tokInflight, tokCapacity int64
+	var endpointCount int
+	var hybridSatSum float64
 	for _, e := range endpoints {
 		if e == nil {
 			continue
 		}
 
+		endpointCount++
 		reqCapacity += d.config.maxConcurrency
 		tokCapacity += d.config.maxTokenConcurrency
 
@@ -149,13 +153,20 @@ func (d *detector) Saturation(_ context.Context, endpoints []datalayer.Endpoint)
 		load := d.getLoad(e.GetAttributes())
 		reqInflight += load.Requests
 		tokInflight += load.Tokens
+		hybridSatSum += max(
+			ratio(load.Requests, d.config.maxConcurrency),
+			ratio(load.Tokens, d.config.maxTokenConcurrency),
+		)
 	}
 
 	switch d.config.mode {
 	case modeTokens:
 		return ratio(tokInflight, tokCapacity)
 	case modeHybrid:
-		return max(ratio(reqInflight, reqCapacity), ratio(tokInflight, tokCapacity))
+		if endpointCount == 0 {
+			return 1.0
+		}
+		return hybridSatSum / float64(endpointCount)
 	default:
 		return ratio(reqInflight, reqCapacity)
 	}

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

@@ -495,7 +495,8 @@ func TestDetector_TokenDeleteEndpoint(t *testing.T) {
 	require.InDelta(t, 0.0, detector.Saturation(ctx, candidates), 1e-6, "expected clean state after DeleteEndpoint")
 }
 
-// TestDetector_HybridSaturation verifies hybrid mode reports the more saturated dimension.
+// TestDetector_HybridSaturation verifies hybrid mode evaluates each endpoint as the more
+// saturated of its request and token dimensions and averages the result across endpoints.
 func TestDetector_HybridSaturation(t *testing.T) {
 	t.Parallel()
 
@@ -505,17 +506,35 @@ func TestDetector_HybridSaturation(t *testing.T) {
 		maxTokenConcurrency: 100,
 	}
 
+	type endpointLoad struct {
+		requests int64
+		tokens   int64
+	}
+
 	tests := []struct {
 		name           string
-		requests       int64
-		tokens         int64
+		endpoints      map[string]endpointLoad
 		wantSaturation float64
 	}{
-		{name: "empty", requests: 0, tokens: 0, wantSaturation: 0.0},
-		{name: "tokens_dominate", requests: 2, tokens: 80, wantSaturation: 0.8},   // max(0.2, 0.8)
-		{name: "requests_dominate", requests: 9, tokens: 10, wantSaturation: 0.9}, // max(0.9, 0.1)
-		{name: "request_dimension_saturates_first", requests: 10, tokens: 5, wantSaturation: 1.0},
-		{name: "token_dimension_saturates_first", requests: 1, tokens: 100, wantSaturation: 1.0},
+		{name: "empty", endpoints: map[string]endpointLoad{"endpoint-a": {0, 0}}, wantSaturation: 0.0},
+		// Single endpoint: saturation is the more saturated of its two dimensions.
+		{name: "tokens_dominate", endpoints: map[string]endpointLoad{"endpoint-a": {2, 80}}, wantSaturation: 0.8},   // max(0.2, 0.8)
+		{name: "requests_dominate", endpoints: map[string]endpointLoad{"endpoint-a": {9, 10}}, wantSaturation: 0.9}, // max(0.9, 0.1)
+		{name: "request_dimension_saturates_first", endpoints: map[string]endpointLoad{"endpoint-a": {10, 5}}, wantSaturation: 1.0},
+		{name: "token_dimension_saturates_first", endpoints: map[string]endpointLoad{"endpoint-a": {1, 100}}, wantSaturation: 1.0},
+		// Distinct endpoints saturated on different dimensions each report 1.0, so the pool
+		// averages to 1.0.
+		{
+			name:           "endpoints_saturated_on_different_dimensions",
+			endpoints:      map[string]endpointLoad{"endpoint-a": {10, 0}, "endpoint-b": {0, 100}},
+			wantSaturation: 1.0,
+		},
+		// One saturated endpoint and one idle endpoint average to the midpoint.
+		{
+			name:           "per_endpoint_average",
+			endpoints:      map[string]endpointLoad{"endpoint-a": {10, 0}, "endpoint-b": {0, 0}},
+			wantSaturation: 0.5, // mean(1.0, 0.0)
+		},
 	}
 
 	for _, tc := range tests {
@@ -524,13 +543,17 @@ func TestDetector_HybridSaturation(t *testing.T) {
 			reg := newLocalRegistry()
 			ctx := context.Background()
 			detector := newDetector("test-detector", config, logr.Discard())
-			endpointName := "endpoint-a"
-			reg.update(fullEndpointName(endpointName), func(load *attrconcurrency.InFlightLoad) {
-				load.Requests = tc.requests
-				load.Tokens = tc.tokens
-			})
 
-			got := detector.Saturation(ctx, []datalayer.Endpoint{newFakeEndpoint(reg, endpointName)})
+			candidates := make([]datalayer.Endpoint, 0, len(tc.endpoints))
+			for name, l := range tc.endpoints {
+				reg.update(fullEndpointName(name), func(load *attrconcurrency.InFlightLoad) {
+					load.Requests = l.requests
+					load.Tokens = l.tokens
+				})
+				candidates = append(candidates, newFakeEndpoint(reg, name))
+			}
+
+			got := detector.Saturation(ctx, candidates)
 			require.InDelta(t, tc.wantSaturation, got, 1e-6, "hybrid saturation mismatch")
 		})
 	}