feat(epp): wire in-flight eviction behind flowControl.enableEviction

The confirmation grace is derived at wiring time from the selected
saturation detector (dispatch-tick scale for the concurrency detector;
refresh interval + staleness + engine reclaim budget for scraped
sensors); remaining controller parameters are internal. The Director
tracks dispatches and stream termination; the ext_proc server receives
the evict-channel lookup.

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

Co-authored-by: Rishabh Saini <rishabhsaini01@gmail.com>
Signed-off-by: Luke Van Drie <lukevandrie@google.com>

Author: LukeAVanDrie

cmd/epp/runner/runner.go

@@ -56,8 +56,10 @@ import (
 	"github.com/llm-d/llm-d-router/pkg/epp/flowcontrol"
 	"github.com/llm-d/llm-d-router/pkg/epp/flowcontrol/contracts"
 	fccontroller "github.com/llm-d/llm-d-router/pkg/epp/flowcontrol/controller"
+	fceviction "github.com/llm-d/llm-d-router/pkg/epp/flowcontrol/eviction"
 	fcregistry "github.com/llm-d/llm-d-router/pkg/epp/flowcontrol/registry"
 	fwkdl "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/datalayer"
+	fwkfc "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"
 	attrconcurrency "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/attribute/concurrency"
 	attrlatency "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/attribute/latency"
@@ -70,6 +72,8 @@ import (
 	sourcemetrics "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/source/metrics"
 	srcmodels "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/source/models"
 	sourcenotifications "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/source/notifications"
+	evictfiltering "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/flowcontrol/eviction/filtering"
+	evictordering "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/flowcontrol/eviction/ordering"
 	"github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/flowcontrol/fairness/globalstrict"
 	programaware "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/flowcontrol/fairness/program-aware"
 	"github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/flowcontrol/fairness/roundrobin"
@@ -448,9 +452,12 @@ func (r *Runner) setup(ctx context.Context, cfg *rest.Config, opts *runserver.Op
 
 	endpointCandidates := contracts.EndpointCandidates(requestcontrol.NewDatastoreEndpointCandidates(ds,
 		requestcontrol.WithDisableEndpointSubsetFilter(opts.DisableEndpointSubsetFilter)))
-	endpointCandidates, admissionController, priorityBandControlPlane := r.initAdmissionControl(ctx, opts, eppConfig, endpointCandidates)
+	endpointCandidates, admissionController, priorityBandControlPlane, requestEvictor := r.initAdmissionControl(ctx, opts, eppConfig, endpointCandidates)
 
 	director := requestcontrol.NewDirectorWithConfig(ds, scheduler, admissionController, endpointCandidates, r.requestControlConfig)
+	if requestEvictor != nil {
+		director.SetRequestEvictor(requestEvictor)
+	}
 
 	serverRunner := &runserver.ExtProcServerRunner{
 		GrpcPort:                         opts.GRPCPort,
@@ -471,6 +478,9 @@ func (r *Runner) setup(ctx context.Context, cfg *rest.Config, opts *runserver.Op
 		GRPCMaxSendMsgSize:               opts.GRPCMaxSendMsgSize,
 		EnableGRPCStreamMetrics:          opts.EnableGRPCStreamMetrics,
 	}
+	if requestEvictor != nil {
+		serverRunner.EvictChannelLookup = requestEvictor.EvictionRegistry()
+	}
 
 	if err := serverRunner.SetupWithManager(mgr); err != nil {
 		setupLog.Error(err, "Failed to setup EPP controllers")
@@ -876,28 +886,87 @@ func (r *Runner) initAdmissionControl(
 	opts *runserver.Options,
 	eppConfig *config.Config,
 	endpointCandidates contracts.EndpointCandidates,
-) (contracts.EndpointCandidates, requestcontrol.AdmissionController, contracts.PriorityBandControlPlane) {
+) (contracts.EndpointCandidates, requestcontrol.AdmissionController, contracts.PriorityBandControlPlane, *fceviction.RequestEvictor) {
 	if !r.featureGates[flowcontrol.FeatureGate] {
 		setupLog.Info("Experimental Flow Control layer is disabled, using legacy admission control")
 		return endpointCandidates,
 			requestcontrol.NewLegacyAdmissionController(eppConfig.SaturationDetector, endpointCandidates),
+			nil,
 			nil
 	}
 	endpointCandidates = requestcontrol.NewCachedEndpointCandidates(ctx, endpointCandidates, 50*time.Millisecond)
 	setupLog.Info("Initializing experimental Flow Control layer")
 	registry := fcregistry.NewFlowRegistry(eppConfig.FlowControlConfig.Registry, setupLog)
-	fc := fccontroller.NewFlowController(
-		ctx,
-		opts.PoolName,
-		eppConfig.FlowControlConfig.Controller,
-		fccontroller.Deps{
-			Registry:           registry,
-			SaturationDetector: eppConfig.SaturationDetector,
-			EndpointCandidates: endpointCandidates,
-			UsageLimitPolicy:   eppConfig.FlowControlConfig.UsageLimitPolicy,
-		},
-	)
-	return endpointCandidates, requestcontrol.NewFlowControlAdmissionController(fc, opts.PoolName), registry
+
+	deps := fccontroller.Deps{
+		Registry:           registry,
+		SaturationDetector: eppConfig.SaturationDetector,
+		EndpointCandidates: endpointCandidates,
+		UsageLimitPolicy:   eppConfig.FlowControlConfig.UsageLimitPolicy,
+	}
+
+	var requestEvictor *fceviction.RequestEvictor
+	if eppConfig.FlowControlConfig.Controller.EnableEviction {
+		var err error
+		requestEvictor, err = buildRequestEvictor()
+		if err != nil {
+			setupLog.Error(err, "Failed to build eviction plumbing; in-flight eviction disabled")
+		} else {
+			deps.InFlightEvictor = requestEvictor
+			grace := deriveEvictionConfirmationGrace(eppConfig.SaturationDetector, opts)
+			eppConfig.FlowControlConfig.Controller.EvictionConfirmationGrace = grace
+			setupLog.Info("In-flight eviction plumbing initialized",
+				"filter", evictfiltering.SheddableFilterType,
+				"ordering", evictordering.PriorityThenTimeOrderingType,
+				"confirmationGrace", grace)
+		}
+	}
+
+	fc := fccontroller.NewFlowController(ctx, opts.PoolName, eppConfig.FlowControlConfig.Controller, deps)
+	return endpointCandidates, requestcontrol.NewFlowControlAdmissionController(fc, opts.PoolName), registry, requestEvictor
+}
+
+const (
+	// evictionEngineReclaimBudget covers the engine-side abort and KV garbage-collection time
+	// between a revoked request's stream termination and the freed capacity appearing in scraped
+	// metrics.
+	evictionEngineReclaimBudget = 250 * time.Millisecond
+	// evictionLeadingSensorGrace covers scheduling jitter for sensors whose gauge updates in the
+	// same event chain as the confirmation (a few dispatch ticks).
+	evictionLeadingSensorGrace = 10 * time.Millisecond
+)
+
+// deriveEvictionConfirmationGrace computes the reclamation pacing grace from the paired saturation
+// detector's confirmation-to-visibility lag. The grace is a property of the sensor, not a
+// preference, so it is derived rather than configured (see docs/flow-control-eviction.md).
+func deriveEvictionConfirmationGrace(detector fwkfc.SaturationDetector, opts *runserver.Options) time.Duration {
+	if detector != nil && detector.TypedName().Type == concurrency.ConcurrencyDetectorType {
+		// The concurrency detector's counters decrement when the stream terminates: the gauge leads
+		// physical reclamation, and only dispatch-loop jitter separates confirmation from visibility.
+		return evictionLeadingSensorGrace
+	}
+	// Scraped sensors (utilization detector, or any custom detector, conservatively): the freed
+	// capacity is visible only after the engine reclaims it and the next non-stale scrape lands.
+	return opts.RefreshMetricsInterval + opts.MetricsStalenessThreshold + evictionEngineReclaimBudget
+}
+
+// buildRequestEvictor assembles the in-flight eviction plumbing: the victim filter and ordering
+// policies, the ImmediateResponse eviction mechanism, and the tracking queue that ties them
+// together. See docs/flow-control-eviction.md.
+func buildRequestEvictor() (*fceviction.RequestEvictor, error) {
+	orderingPlugin, err := evictordering.PriorityThenTimeOrderingFactory(evictordering.PriorityThenTimeOrderingType, nil, nil)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create eviction ordering policy: %w", err)
+	}
+	filterPlugin, err := evictfiltering.SheddableFilterFactory(evictfiltering.SheddableFilterType, nil, nil)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create eviction filter policy: %w", err)
+	}
+	return fceviction.NewRequestEvictor(
+		orderingPlugin.(fwkfc.EvictionOrderingPolicy),
+		filterPlugin.(fwkfc.EvictionFilterPolicy),
+		fceviction.NewImmediateResponseEvictor(),
+	), nil
 }
 
 // runWithFileDiscovery handles the execution path when a discovery plugin is configured.
@@ -974,8 +1043,11 @@ func (r *Runner) runWithFileDiscovery(ctx context.Context, opts *runserver.Optio
 		requestcontrol.WithDisableEndpointSubsetFilter(opts.DisableEndpointSubsetFilter)))
 	// File-discovery mode has no InferenceObjective reconciler to drive the
 	// control plane; static bands from config apply at registry construction.
-	endpointCandidates, admissionController, _ := r.initAdmissionControl(ctx, opts, eppConfig, endpointCandidates)
+	endpointCandidates, admissionController, _, requestEvictor := r.initAdmissionControl(ctx, opts, eppConfig, endpointCandidates)
 	director := requestcontrol.NewDirectorWithConfig(ds, scheduler, admissionController, endpointCandidates, r.requestControlConfig)
+	if requestEvictor != nil {
+		director.SetRequestEvictor(requestEvictor)
+	}
 
 	gknn := common.GKNN{
 		NamespacedName: types.NamespacedName{Name: poolName, Namespace: namespace},
@@ -998,6 +1070,9 @@ func (r *Runner) runWithFileDiscovery(ctx context.Context, opts *runserver.Optio
 		GRPCMaxSendMsgSize:               opts.GRPCMaxSendMsgSize,
 		EnableGRPCStreamMetrics:          opts.EnableGRPCStreamMetrics,
 	}
+	if requestEvictor != nil {
+		serverRunner.EvictChannelLookup = requestEvictor.EvictionRegistry()
+	}
 
 	r.customCollectors = append(r.customCollectors, collectors.NewInferencePoolMetricsCollector(ds))
 	metrics.Register(r.customCollectors...)

pkg/epp/requestcontrol/director.go

@@ -43,6 +43,7 @@ import (
 	"github.com/llm-d/llm-d-router/pkg/epp/datalayer"
 	"github.com/llm-d/llm-d-router/pkg/epp/datastore"
 	"github.com/llm-d/llm-d-router/pkg/epp/flowcontrol/contracts"
+	"github.com/llm-d/llm-d-router/pkg/epp/flowcontrol/eviction"
 	fwkdl "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/datalayer"
 	fwkrc "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/requestcontrol"
 	fwkrh "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/requesthandling"
@@ -210,6 +211,10 @@ type Director struct {
 	// streaming response path. The request context key avoids coupling independent streams that reuse the same
 	// x-request-id header.
 	responseBodyQueues sync.Map
+
+	// requestEvictor, when set, tracks dispatched requests for demand-driven in-flight eviction.
+	// See docs/flow-control-eviction.md.
+	requestEvictor *eviction.RequestEvictor
 }
 
 // getInferenceObjective fetches the inferenceObjective from the datastore otherwise creates a new one based on reqCtx.
@@ -483,9 +488,19 @@ func (d *Director) prepareRequest(ctx context.Context, reqCtx *handlers.RequestC
 
 	d.runPreRequestPlugins(ctx, reqCtx.SchedulingRequest, result)
 
+	if d.requestEvictor != nil {
+		d.requestEvictor.PreRequest(ctx, reqCtx.SchedulingRequest, result)
+	}
+
 	return reqCtx, nil
 }
 
+// SetRequestEvictor wires the in-flight eviction tracker into the request lifecycle.
+// Must be called before the Director serves traffic.
+func (d *Director) SetRequestEvictor(re *eviction.RequestEvictor) {
+	d.requestEvictor = re
+}
+
 func (d *Director) toSchedulerEndpoints(endpoints []fwkdl.Endpoint) []fwksched.Endpoint {
 	result := make([]fwksched.Endpoint, len(endpoints))
 	for i, endpoint := range endpoints {
@@ -522,6 +537,16 @@ func (d *Director) HandleResponseHeader(ctx context.Context, reqCtx *handlers.Re
 func (d *Director) HandleResponseBody(ctx context.Context, reqCtx *handlers.RequestContext, endOfStream bool) *handlers.RequestContext {
 	logger := log.FromContext(ctx).WithValues("stage", "bodyChunk")
 	logger.V(logutil.TRACE).Info("Entering HandleResponseBodyChunk")
+
+	// The eviction tracker must observe stream termination even when no streaming plugins are
+	// registered, so this runs before the early return below.
+	if endOfStream && d.requestEvictor != nil {
+		d.requestEvictor.ResponseBody(ctx, reqCtx.SchedulingRequest, &fwkrc.Response{
+			RequestID:   reqCtx.Request.Headers[reqcommon.RequestIDHeaderKey],
+			EndOfStream: true,
+		}, reqCtx.TargetPod)
+	}
+
 	if len(d.requestControlPlugins.responseStreamingPlugins) == 0 {
 		logger.V(logutil.TRACE).Info("Exiting HandleResponseBodyChunk")
 		return reqCtx

pkg/epp/server/runserver.go

@@ -66,6 +66,9 @@ type ExtProcServerRunner struct {
 	GRPCMaxRecvMsgSize               int
 	GRPCMaxSendMsgSize               int
 	EnableGRPCStreamMetrics          bool
+	// EvictChannelLookup, when set, enables the ext_proc server to terminate in-flight requests
+	// selected for eviction by flow control. See docs/flow-control-eviction.md.
+	EvictChannelLookup handlers.EvictChannelLookup
 }
 
 // NewDefaultExtProcServerRunner creates a runner with default values.
@@ -214,6 +217,9 @@ func (r *ExtProcServerRunner) AsRunnable(logger logr.Logger) manager.Runnable {
 			poolCap = 4 * 1024 * 1024 // gRPC default 4MB
 		}
 		extProcServer := handlers.NewStreamingServer(r.Datastore, r.Director, r.ParserRegistry, poolCap)
+		if r.EvictChannelLookup != nil {
+			extProcServer.SetEvictChannelLookup(r.EvictChannelLookup)
+		}
 		extProcPb.RegisterExternalProcessorServer(srv, extProcServer)
 
 		if r.HealthChecking {