feat(concurrency-detector): Add hybrid mode to concurrency detector (#1991)

* feat(concurrency-detector): Add hybrid mode to concurrency detector

"hybrid" mode is a new concurrency detector mode that has been added
which takes into account both concurrency based on token and request.

Its saturation is computed as:

`saturation = max(requestSaturation, tokenSaturation)`

Using just request based concurrency is not enough as different
workloads such as prefill heavy tends to consume more GPU compute, using
just the tokenSaturation suffers from wrong estimation of output tokens
which falls down in low ISL and high OSL workloads.

Hybrid concurrency detector attempts to solve this problem by
essentially capping whichever saturates first due to high ISL/OSL workload.

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

* ci: run formatting

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

* 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>

* docs: review and fix the code comments

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

---------

Signed-off-by: mohammadkhan <mohammadkhan@digitalocean.com>
Co-authored-by: Abdullah Gharaibeh <40361897+ahg-g@users.noreply.github.com>
Co-authored-by: Luke Van Drie <lukevandrie@google.com>

Author: 3 people

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"` or `"tokens"`. (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/config.go

@@ -62,6 +62,8 @@ type apiConfig struct {
 	// Valid values are:
 	// - "requests": use discrete request counts for capacity accounting.
 	// - "tokens": use estimated token counts for capacity accounting.
+	// - "hybrid": evaluate each endpoint as the more constraining of its request and
+	//   token ratios; pool saturation is the average of these across endpoints.
 	//
 	// Defaults to "requests" if unset.
 	ConcurrencyMode *concurrencyMode `json:"concurrencyMode,omitempty"`
@@ -85,6 +87,8 @@ const (
 	modeRequests concurrencyMode = "requests"
 	// modeTokens uses token count for concurrency detection.
 	modeTokens concurrencyMode = "tokens"
+	// modeHybrid uses endpoint average saturation across both request and token counts.
+	modeHybrid concurrencyMode = "hybrid"
 )
 
 const (
@@ -164,7 +168,7 @@ func validateConfig(cfg *apiConfig) error {
 
 	if cfg.ConcurrencyMode != nil {
 		switch *cfg.ConcurrencyMode {
-		case modeRequests, modeTokens:
+		case modeRequests, modeTokens, modeHybrid:
 			// Valid
 		default:
 			errs = append(errs, fmt.Errorf("unsupported concurrencyMode: %q", *cfg.ConcurrencyMode))

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

@@ -121,49 +121,69 @@ func (d *detector) getLoad(m datalayer.AttributeMap) *attrconcurrency.InFlightLo
 
 // Saturation calculates the saturation level of the pool.
 //
-// It returns an aggregate saturation signal where:
+// In "requests" and "tokens" mode it returns an aggregate signal, evaluated as:
 //
-//	Saturation = Total Inflight Requests / Total MaxConcurrency Capacity.
+//	Saturation = Total Inflight / Total Capacity.
+//
+// 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 totalInflight, totalCapacity int64
+	var reqInflight, reqCapacity, tokInflight, tokCapacity int64
+	var endpointCount int
+	var hybridSatSum float64
 	for _, e := range endpoints {
 		if e == nil {
 			continue
 		}
 
-		if d.config.mode == modeTokens {
-			totalCapacity += d.config.maxTokenConcurrency
-		} else {
-			totalCapacity += d.config.maxConcurrency
-		}
+		endpointCount++
+		reqCapacity += d.config.maxConcurrency
+		tokCapacity += d.config.maxTokenConcurrency
 
 		if e.GetMetadata() == nil {
 			continue
 		}
 
 		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),
+		)
+	}
 
-		if d.config.mode == modeTokens {
-			totalInflight += load.Tokens
-		} else {
-			totalInflight += load.Requests
+	switch d.config.mode {
+	case modeTokens:
+		return ratio(tokInflight, tokCapacity)
+	case modeHybrid:
+		if endpointCount == 0 {
+			return 1.0
 		}
+		return hybridSatSum / float64(endpointCount)
+	default:
+		return ratio(reqInflight, reqCapacity)
 	}
+}
 
-	if totalCapacity == 0 {
+// ratio computes inflight/capacity, failing closed (1.0) when capacity is zero.
+func ratio(inflight, capacity int64) float64 {
+	if capacity == 0 {
 		return 1.0
 	}
-
-	return float64(totalInflight) / float64(totalCapacity)
+	return float64(inflight) / float64(capacity)
 }
 
 // Filter blocks traffic to specific endpoints that are physically saturated or exceeding their safety limits.
 //
-// It applies a relaxed limit (MaxConcurrency * (1 + Headroom)) to allow for scheduling flexibility and burst tolerance.
+// It applies a relaxed limit (Capacity * (1 + Headroom)) to allow for scheduling flexibility and burst tolerance.
+// In "hybrid" mode an endpoint is dropped when either its request load or its token load reaches the limit.
 func (d *detector) Filter(
 	_ context.Context,
 	_ *fwksched.InferenceRequest,
@@ -172,28 +192,30 @@ func (d *detector) Filter(
 	// Pre-allocate assuming most endpoints will pass the filter to minimize allocations.
 	filtered := make([]fwksched.Endpoint, 0, len(endpoints))
 
-	var limit int64
-	if d.config.mode == modeTokens {
-		limit = int64(float64(d.config.maxTokenConcurrency) * (1.0 + d.config.headroom))
-	} else {
-		limit = int64(float64(d.config.maxConcurrency) * (1.0 + d.config.headroom))
-	}
+	reqLimit := int64(float64(d.config.maxConcurrency) * (1.0 + d.config.headroom))
+	tokLimit := int64(float64(d.config.maxTokenConcurrency) * (1.0 + d.config.headroom))
 
 	for _, e := range endpoints {
 		if e == nil {
 			continue
 		}
 		load := d.getLoad(e)
 
-		if d.config.mode == modeTokens {
-			if load.Tokens < limit {
-				filtered = append(filtered, e)
-			}
-		} else {
-			if load.Requests < limit {
-				filtered = append(filtered, e)
-			}
+		if d.admits(load, reqLimit, tokLimit) {
+			filtered = append(filtered, e)
 		}
 	}
 	return filtered
 }
+
+// admits reports whether an endpoint is below its safety limit for the active mode.
+func (d *detector) admits(load *attrconcurrency.InFlightLoad, reqLimit, tokLimit int64) bool {
+	switch d.config.mode {
+	case modeTokens:
+		return load.Tokens < tokLimit
+	case modeHybrid:
+		return load.Requests < reqLimit && load.Tokens < tokLimit
+	default:
+		return load.Requests < reqLimit
+	}
+}

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

@@ -110,6 +110,11 @@ func TestConcurrencyDetectorFactory(t *testing.T) {
 			configJSON: []byte(`{"headroom": -0.5}`),
 			wantError:  true,
 		},
+		{
+			name:       "valid hybrid mode",
+			configJSON: []byte(`{"concurrencyMode": "hybrid", "maxConcurrency": 10, "maxTokenConcurrency": 100}`),
+			wantError:  false,
+		},
 		{
 			name:       "invalid mode",
 			configJSON: []byte(`{"concurrencyMode": "magic"}`),
@@ -490,6 +495,111 @@ 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 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()
+
+	config := config{
+		mode:                modeHybrid,
+		maxConcurrency:      10,
+		maxTokenConcurrency: 100,
+	}
+
+	type endpointLoad struct {
+		requests int64
+		tokens   int64
+	}
+
+	tests := []struct {
+		name           string
+		endpoints      map[string]endpointLoad
+		wantSaturation float64
+	}{
+		{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 {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			reg := newLocalRegistry()
+			ctx := context.Background()
+			detector := newDetector("test-detector", config, logr.Discard())
+
+			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")
+		})
+	}
+}
+
+// TestDetector_HybridFilter verifies hybrid mode drops an endpoint when either dimension hits its limit.
+func TestDetector_HybridFilter(t *testing.T) {
+	t.Parallel()
+
+	// headroom 0.0 -> limits are exactly maxConcurrency (10) and maxTokenConcurrency (100).
+	config := config{
+		mode:                modeHybrid,
+		maxConcurrency:      10,
+		maxTokenConcurrency: 100,
+	}
+	ctx := context.Background()
+
+	tests := []struct {
+		name     string
+		requests int64
+		tokens   int64
+		wantKept int
+	}{
+		{name: "below_both_limits", requests: 5, tokens: 50, wantKept: 1},
+		{name: "request_limit_reached", requests: 10, tokens: 50, wantKept: 0},
+		{name: "token_limit_reached", requests: 5, tokens: 100, wantKept: 0},
+		{name: "both_over", requests: 20, tokens: 200, wantKept: 0},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			reg := newLocalRegistry()
+			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
+			})
+
+			kept := detector.Filter(ctx, nil, []fwksched.Endpoint{newStubSchedulingEndpoint(reg, endpointName)})
+			require.Len(t, kept, tc.wantKept, "hybrid filter mismatch")
+		})
+	}
+}
+
 // TestDetector_ConcurrencyStress performs race condition check.
 func TestDetector_ConcurrencyStress(t *testing.T) {
 	t.Parallel()