diff --git a/apix/config/v1alpha1/endpointpickerconfig_types.go b/apix/config/v1alpha1/endpointpickerconfig_types.go index 3926f924e5..3b8112abe5 100644 --- a/apix/config/v1alpha1/endpointpickerconfig_types.go +++ b/apix/config/v1alpha1/endpointpickerconfig_types.go @@ -401,6 +401,14 @@ type FlowControlConfig struct { // SaturationDetector specifies which saturation detector plugin to use for both Admission and // Flow Control. If omitted, "utilization-detector" is used by default. SaturationDetector *SaturationDetectorConfig `json:"saturationDetector,omitempty"` + + // +optional + // EnableEviction enables demand-driven in-flight eviction. When higher-priority requests are + // blocked by pool saturation, lower-priority in-flight requests (priority < 0) may be + // terminated to reclaim capacity. Pacing and sizing self-configure from the selected + // saturation detector. See docs/flow-control-eviction.md. + // Defaults to false. + EnableEviction bool `json:"enableEviction,omitempty"` } func (fcc *FlowControlConfig) String() string { @@ -445,6 +453,10 @@ func (fcc *FlowControlConfig) String() string { parts = append(parts, fmt.Sprintf("SaturationDetector: %v", fcc.SaturationDetector)) } + if fcc.EnableEviction { + parts = append(parts, "EnableEviction: true") + } + return "{" + strings.Join(parts, ", ") + "}" } diff --git a/cmd/epp/runner/runner.go b/cmd/epp/runner/runner.go index c2fb54efbe..5ecaa9fd39 100644 --- a/cmd/epp/runner/runner.go +++ b/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...) diff --git a/pkg/epp/flowcontrol/controller/config.go b/pkg/epp/flowcontrol/controller/config.go index b154d4a8be..8596306df9 100644 --- a/pkg/epp/flowcontrol/controller/config.go +++ b/pkg/epp/flowcontrol/controller/config.go @@ -28,6 +28,16 @@ const ( defaultExpiryCleanupInterval = 1 * time.Second // defaultEnqueueChannelBufferSize is the default size of a worker's incoming request buffer. defaultEnqueueChannelBufferSize = 100 + // defaultMaxRevocationsPerDecision caps revocations per reclamation decision, bounding the + // damage from mean-footprint misestimation. + defaultMaxRevocationsPerDecision = 2 + // defaultEvictionConfirmationGrace covers the reclaiming stage (engine abort, KV GC, scrape, + // staleness window) for the default utilization detector. Deployments pairing eviction with the + // concurrency detector can lower it substantially. + defaultEvictionConfirmationGrace = 500 * time.Millisecond + // defaultEvictionConfirmationTimeout bounds how long an unconfirmed revocation can hold the + // reclamation pacing gate closed. + defaultEvictionConfirmationTimeout = 10 * time.Second ) // Config holds the configuration for the `FlowController`. @@ -46,6 +56,33 @@ type Config struct { // serial execution loop and allowing the system to handle short bursts of traffic without blocking. // Optional: Defaults to `defaultEnqueueChannelBufferSize` (100). EnqueueChannelBufferSize int + + // EnableEviction enables demand-driven in-flight eviction: when higher-priority requests are + // blocked by pool saturation, lower-priority in-flight requests may be terminated to reclaim + // capacity. Requires the eviction plumbing to be wired (see Deps.InFlightEvictor). + // See docs/flow-control-eviction.md. + // Optional: Defaults to false. + EnableEviction bool + + // MaxRevocationsPerDecision caps how many revocations a single reclamation decision may issue. + // Not exposed through the API configuration: benchmark data shows sizing is deficit-bound, so + // this cap rarely binds and is not worth a user-facing knob. + // Optional: Defaults to `defaultMaxRevocationsPerDecision` (2). + MaxRevocationsPerDecision int + + // EvictionConfirmationGrace is how long a confirmed revocation's pending-reclaim debit keeps + // suppressing further reclamation, covering the saturation signal's confirmation-to-visibility + // lag. Not exposed through the API configuration: the EPP wiring derives it from the selected + // saturation detector, since the correct value is a property of the sensor, not a preference. + // Optional: Defaults to `defaultEvictionConfirmationGrace` (500ms), the conservative value for + // scraped sensors. + EvictionConfirmationGrace time.Duration + + // EvictionConfirmationTimeout bounds how long an unconfirmed revocation can hold the + // reclamation pacing gate closed before being treated as confirmed. Not exposed through the + // API configuration. + // Optional: Defaults to `defaultEvictionConfirmationTimeout` (10s). + EvictionConfirmationTimeout time.Duration } func (c *Config) String() string { @@ -64,11 +101,14 @@ type ConfigOption func(*Config) // NewConfigFromAPI creates a new Config from the API configuration. func NewConfigFromAPI(apiConfig *configapi.FlowControlConfig) (*Config, error) { - opts := make([]ConfigOption, 0, 1) + opts := make([]ConfigOption, 0, 4) if apiConfig != nil { if apiConfig.DefaultRequestTTL != nil { opts = append(opts, WithDefaultRequestTTL(apiConfig.DefaultRequestTTL.Duration)) } + if apiConfig.EnableEviction { + opts = append(opts, WithEnableEviction(true)) + } } return NewConfig(opts...) } @@ -76,8 +116,11 @@ func NewConfigFromAPI(apiConfig *configapi.FlowControlConfig) (*Config, error) { // NewConfig creates a new Config with the given options, applying defaults and validation. func NewConfig(opts ...ConfigOption) (*Config, error) { c := &Config{ - ExpiryCleanupInterval: defaultExpiryCleanupInterval, - EnqueueChannelBufferSize: defaultEnqueueChannelBufferSize, + ExpiryCleanupInterval: defaultExpiryCleanupInterval, + EnqueueChannelBufferSize: defaultEnqueueChannelBufferSize, + MaxRevocationsPerDecision: defaultMaxRevocationsPerDecision, + EvictionConfirmationGrace: defaultEvictionConfirmationGrace, + EvictionConfirmationTimeout: defaultEvictionConfirmationTimeout, } for _, opt := range opts { @@ -111,6 +154,34 @@ func WithEnqueueChannelBufferSize(size int) ConfigOption { } } +// WithEnableEviction enables demand-driven in-flight eviction. +func WithEnableEviction(enabled bool) ConfigOption { + return func(c *Config) { + c.EnableEviction = enabled + } +} + +// WithMaxRevocationsPerDecision sets the per-decision revocation cap. +func WithMaxRevocationsPerDecision(n int) ConfigOption { + return func(c *Config) { + c.MaxRevocationsPerDecision = n + } +} + +// WithEvictionConfirmationGrace sets the post-confirmation grace period. +func WithEvictionConfirmationGrace(d time.Duration) ConfigOption { + return func(c *Config) { + c.EvictionConfirmationGrace = d + } +} + +// WithEvictionConfirmationTimeout sets the confirmation timeout. +func WithEvictionConfirmationTimeout(d time.Duration) ConfigOption { + return func(c *Config) { + c.EvictionConfirmationTimeout = d + } +} + // validate checks the configuration for validity. func (c *Config) validate() error { if c.DefaultRequestTTL < 0 { @@ -122,5 +193,14 @@ func (c *Config) validate() error { if c.EnqueueChannelBufferSize < 0 { return fmt.Errorf("EnqueueChannelBufferSize cannot be negative, but got %d", c.EnqueueChannelBufferSize) } + if c.MaxRevocationsPerDecision < 1 { + return fmt.Errorf("MaxRevocationsPerDecision must be at least 1, but got %d", c.MaxRevocationsPerDecision) + } + if c.EvictionConfirmationGrace < 0 { + return fmt.Errorf("EvictionConfirmationGrace cannot be negative, but got %v", c.EvictionConfirmationGrace) + } + if c.EvictionConfirmationTimeout <= 0 { + return fmt.Errorf("EvictionConfirmationTimeout must be positive, but got %v", c.EvictionConfirmationTimeout) + } return nil } diff --git a/pkg/epp/flowcontrol/controller/config_test.go b/pkg/epp/flowcontrol/controller/config_test.go index 89773a3c18..5d91059ba1 100644 --- a/pkg/epp/flowcontrol/controller/config_test.go +++ b/pkg/epp/flowcontrol/controller/config_test.go @@ -41,9 +41,12 @@ func TestNewConfig(t *testing.T) { opts: nil, expectErr: false, expectedCfg: Config{ - DefaultRequestTTL: 0, - ExpiryCleanupInterval: defaultExpiryCleanupInterval, - EnqueueChannelBufferSize: defaultEnqueueChannelBufferSize, + DefaultRequestTTL: 0, + ExpiryCleanupInterval: defaultExpiryCleanupInterval, + EnqueueChannelBufferSize: defaultEnqueueChannelBufferSize, + MaxRevocationsPerDecision: defaultMaxRevocationsPerDecision, + EvictionConfirmationGrace: defaultEvictionConfirmationGrace, + EvictionConfirmationTimeout: defaultEvictionConfirmationTimeout, }, }, { @@ -53,9 +56,12 @@ func TestNewConfig(t *testing.T) { }, expectErr: false, expectedCfg: Config{ - DefaultRequestTTL: 10 * time.Second, - ExpiryCleanupInterval: defaultExpiryCleanupInterval, - EnqueueChannelBufferSize: defaultEnqueueChannelBufferSize, + DefaultRequestTTL: 10 * time.Second, + ExpiryCleanupInterval: defaultExpiryCleanupInterval, + EnqueueChannelBufferSize: defaultEnqueueChannelBufferSize, + MaxRevocationsPerDecision: defaultMaxRevocationsPerDecision, + EvictionConfirmationGrace: defaultEvictionConfirmationGrace, + EvictionConfirmationTimeout: defaultEvictionConfirmationTimeout, }, }, { @@ -67,9 +73,12 @@ func TestNewConfig(t *testing.T) { }, expectErr: false, expectedCfg: Config{ - DefaultRequestTTL: 10 * time.Second, - ExpiryCleanupInterval: 2 * time.Second, - EnqueueChannelBufferSize: 50, + DefaultRequestTTL: 10 * time.Second, + ExpiryCleanupInterval: 2 * time.Second, + EnqueueChannelBufferSize: 50, + MaxRevocationsPerDecision: defaultMaxRevocationsPerDecision, + EvictionConfirmationGrace: defaultEvictionConfirmationGrace, + EvictionConfirmationTimeout: defaultEvictionConfirmationTimeout, }, }, { @@ -100,6 +109,45 @@ func TestNewConfig(t *testing.T) { }, expectErr: true, }, + { + name: "WithEvictionOptions_ShouldUpdateConfig", + opts: []ConfigOption{ + WithEnableEviction(true), + WithMaxRevocationsPerDecision(5), + WithEvictionConfirmationGrace(50 * time.Millisecond), + WithEvictionConfirmationTimeout(30 * time.Second), + }, + expectErr: false, + expectedCfg: Config{ + ExpiryCleanupInterval: defaultExpiryCleanupInterval, + EnqueueChannelBufferSize: defaultEnqueueChannelBufferSize, + EnableEviction: true, + MaxRevocationsPerDecision: 5, + EvictionConfirmationGrace: 50 * time.Millisecond, + EvictionConfirmationTimeout: 30 * time.Second, + }, + }, + { + name: "ZeroMaxRevocationsPerDecision_ShouldError", + opts: []ConfigOption{ + WithMaxRevocationsPerDecision(0), + }, + expectErr: true, + }, + { + name: "NegativeEvictionConfirmationGrace_ShouldError", + opts: []ConfigOption{ + WithEvictionConfirmationGrace(-1 * time.Second), + }, + expectErr: true, + }, + { + name: "ZeroEvictionConfirmationTimeout_ShouldError", + opts: []ConfigOption{ + WithEvictionConfirmationTimeout(0), + }, + expectErr: true, + }, } for _, tc := range testCases { @@ -175,6 +223,19 @@ func TestNewConfigFromAPI(t *testing.T) { }, expectedErr: "DefaultRequestTTL cannot be negative", }, + { + name: "EnableEviction_ShouldTranslate_WithInternalDefaults", + apiConfig: &configapi.FlowControlConfig{ + EnableEviction: true, + }, + assertion: func(t *testing.T, cfg *Config) { + assert.True(t, cfg.EnableEviction, "EnableEviction should be translated") + // Pacing and sizing parameters are internal: defaulted here, derived at wiring time. + assert.Equal(t, defaultMaxRevocationsPerDecision, cfg.MaxRevocationsPerDecision) + assert.Equal(t, defaultEvictionConfirmationGrace, cfg.EvictionConfirmationGrace) + assert.Equal(t, defaultEvictionConfirmationTimeout, cfg.EvictionConfirmationTimeout) + }, + }, } for _, tc := range testCases { diff --git a/pkg/epp/flowcontrol/controller/controller.go b/pkg/epp/flowcontrol/controller/controller.go index 2bcb2293be..8f5de80d65 100644 --- a/pkg/epp/flowcontrol/controller/controller.go +++ b/pkg/epp/flowcontrol/controller/controller.go @@ -60,6 +60,7 @@ type processorFactory func( cleanupSweepInterval time.Duration, enqueueChannelBufferSize int, logger logr.Logger, + reclamation *internal.ReclamationController, ) processor var _ processor = &internal.Processor{} @@ -105,6 +106,11 @@ type Deps struct { UsageLimitPolicy flowcontrol.UsageLimitPolicy Clock clock.WithTicker ProcessorFactory processorFactory + + // InFlightEvictor enables demand-driven in-flight eviction when Config.EnableEviction is set. + // Satisfied by *eviction.RequestEvictor. The FlowController registers the reclamation + // controller's Confirm as the evictor's eviction-terminated listener. + InFlightEvictor internal.InFlightEvictor } // NewFlowController creates and starts a new FlowController instance. @@ -147,6 +153,7 @@ func NewFlowController( cleanupSweepInterval time.Duration, enqueueChannelBufferSize int, logger logr.Logger, + reclamation *internal.ReclamationController, ) processor { return internal.NewProcessor( ctx, @@ -160,12 +167,36 @@ func NewFlowController( cleanupSweepInterval, enqueueChannelBufferSize, logger, + reclamation, ) } } else { fc.processorFactory = deps.ProcessorFactory } + // Demand-driven in-flight eviction is active only when both the config enables it and the + // eviction plumbing was wired in. The reclamation controller's Confirm becomes the evictor's + // eviction-terminated listener, closing the confirmation loop. + var reclamation *internal.ReclamationController + if config.EnableEviction && deps.InFlightEvictor != nil { + reclamation = internal.NewReclamationController( + internal.ReclamationConfig{ + MaxRevocationsPerDecision: config.MaxRevocationsPerDecision, + ConfirmationGrace: config.EvictionConfirmationGrace, + ConfirmationTimeout: config.EvictionConfirmationTimeout, + }, + deps.InFlightEvictor, + deps.Clock, + fc.logger, + poolName, + ) + deps.InFlightEvictor.SetEvictionTerminatedListener(reclamation.Confirm) + fc.logger.V(logutil.DEFAULT).Info("Demand-driven in-flight eviction enabled.", + "maxRevocationsPerDecision", config.MaxRevocationsPerDecision, + "confirmationGrace", config.EvictionConfirmationGrace, + "confirmationTimeout", config.EvictionConfirmationTimeout) + } + // Construct a new worker, but do not start its goroutine yet. fc.processor = fc.processorFactory( fc.parentCtx, @@ -178,6 +209,7 @@ func NewFlowController( fc.config.ExpiryCleanupInterval, fc.config.EnqueueChannelBufferSize, fc.logger, + reclamation, ) fc.logger.V(logutil.DEFAULT).Info("Starting the Processor.") diff --git a/pkg/epp/flowcontrol/controller/controller_test.go b/pkg/epp/flowcontrol/controller/controller_test.go index 68e4dd3db7..ecac47448f 100644 --- a/pkg/epp/flowcontrol/controller/controller_test.go +++ b/pkg/epp/flowcontrol/controller/controller_test.go @@ -287,6 +287,7 @@ func (f *mockProcessorFactory) new( _ time.Duration, _ int, _ logr.Logger, + _ *internal.ReclamationController, ) processor { if f.processor != nil { return f.processor diff --git a/pkg/epp/flowcontrol/controller/internal/processor.go b/pkg/epp/flowcontrol/controller/internal/processor.go index d8c32d257e..121cf1f222 100644 --- a/pkg/epp/flowcontrol/controller/internal/processor.go +++ b/pkg/epp/flowcontrol/controller/internal/processor.go @@ -75,6 +75,10 @@ type Processor struct { cleanupSweepInterval time.Duration logger logr.Logger + // reclamation, when non-nil, enables demand-driven in-flight eviction on HoL blocking. + // See docs/flow-control-eviction.md. + reclamation *ReclamationController + // lifecycleCtx controls the processor's lifetime. Monitored by Submit* methods for safe shutdown. lifecycleCtx context.Context @@ -106,6 +110,7 @@ func NewProcessor( cleanupSweepInterval time.Duration, enqueueChannelBufferSize int, logger logr.Logger, + reclamation *ReclamationController, ) *Processor { return &Processor{ registry: registry, @@ -119,6 +124,7 @@ func NewProcessor( logger: logger, lifecycleCtx: ctx, enqueueChan: make(chan *FlowItem, enqueueChannelBufferSize), + reclamation: reclamation, } } @@ -377,6 +383,9 @@ func (p *Processor) dispatchCycle(ctx context.Context) bool { if saturation >= usageLimit { p.logger.V(logutil.DEBUG).Info("Priority band is saturated; enforcing HoL blocking.", "priority", priority, "saturation", saturation, "usageLimit", usageLimit) + if p.reclamation != nil { + p.maybeReclaim(ctx, saturation, priorities, ceilings, i) + } // 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 diff --git a/pkg/epp/flowcontrol/controller/internal/processor_test.go b/pkg/epp/flowcontrol/controller/internal/processor_test.go index 20be2f473b..921dc172d0 100644 --- a/pkg/epp/flowcontrol/controller/internal/processor_test.go +++ b/pkg/epp/flowcontrol/controller/internal/processor_test.go @@ -142,7 +142,8 @@ func newTestHarness(t *testing.T, expiryCleanupInterval time.Duration) *testHarn h.clock, expiryCleanupInterval, 100, - h.logger) + h.logger, + nil) require.NotNil(t, h.processor, "NewProcessor should not return nil") t.Cleanup(func() { h.Stop() }) diff --git a/pkg/epp/flowcontrol/controller/internal/reclamation.go b/pkg/epp/flowcontrol/controller/internal/reclamation.go new file mode 100644 index 0000000000..f2bb7ac0f2 --- /dev/null +++ b/pkg/epp/flowcontrol/controller/internal/reclamation.go @@ -0,0 +1,337 @@ +/* +Copyright 2026 The llm-d Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package internal + +import ( + "context" + "math" + "strconv" + "sync" + "time" + + "github.com/go-logr/logr" + "k8s.io/utils/clock" + + logutil "github.com/llm-d/llm-d-router/pkg/common/observability/logging" + "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/flowcontrol" + "github.com/llm-d/llm-d-router/pkg/epp/metrics" +) + +// InFlightEvictor is the narrow view of the eviction subsystem the reclamation controller needs. +// It is satisfied by *eviction.RequestEvictor. +type InFlightEvictor interface { + // EvictN evicts up to n requests in victim-policy order, stopping at the first victim whose + // priority is not strictly below priorityBound, and returns the IDs actually evicted. + EvictN(ctx context.Context, n int, priorityBound int) ([]string, error) + // Stats returns the number of tracked in-flight requests and how many are evictable. + Stats() (inFlight int, evictable int) + // PeekVictimPriority returns the priority of the next victim, or false if none are evictable. + PeekVictimPriority() (priority int, ok bool) + // SetEvictionTerminatedListener registers the confirmation callback, invoked once per evicted + // request when its stream terminates. + SetEvictionTerminatedListener(listener func(requestID string)) +} + +// ReclamationConfig holds the tuning parameters for the ReclamationController. +// See docs/flow-control-eviction.md for the derivation of each parameter. +type ReclamationConfig struct { + // MaxRevocationsPerDecision caps how many revocations a single decision may issue. It bounds the + // damage from mean-footprint misestimation. + MaxRevocationsPerDecision int + + // ConfirmationGrace is the wait after the last confirmation before the next decision may run. It + // covers the reclaiming stage: engine abort, KV GC, metrics scrape, and staleness window. + ConfirmationGrace time.Duration + + // ConfirmationTimeout bounds how long an unconfirmed revocation can hold the pacing gate closed. + // After it expires the revocation is treated as confirmed and a health metric is incremented. + ConfirmationTimeout time.Duration +} + +// revocation is one issued, not-yet-confirmed eviction. +type revocation struct { + // credit is the pending-reclaim debit taken for this revocation, in saturation-gauge units, + // retained at its issue-time value until the debit expires. + credit float64 + issuedAt time.Time +} + +// coolingDebit is a confirmed revocation's debit held through the reclaiming stage: the stream is +// dead, but the freed capacity is not yet visible in the gauge (engine GC, scrape, staleness). The +// debit expires ConfirmationGrace after confirmation. +type coolingDebit struct { + credit float64 + expiresAt time.Time +} + +// ReclamationController decides when in-flight eviction fires and how many leases it revokes. +// It implements the sizing and pacing rules from docs/flow-control-eviction.md: +// +// - Sizing: deficit in saturation-gauge units against a mean-footprint credit estimate, capped by +// MaxRevocationsPerDecision. +// - Pacing: confirmation-gated. A new decision may run only once every previously issued +// revocation is confirmed (its ext_proc stream terminated) and ConfirmationGrace has elapsed, +// so the controller never acts twice on a gauge that has not absorbed its prior actions. +// +// Concurrency: Reclaim and GateOpen are called only from the processor's single-writer run loop. +// Confirm is called from ext_proc handler goroutines. The mutex is held across EvictN inside +// Reclaim so a confirmation arriving mid-decision blocks until its revocation is registered; the +// evictor never calls back into the controller synchronously, so this cannot deadlock. +type ReclamationController struct { + evictor InFlightEvictor + cfg ReclamationConfig + clock clock.PassiveClock + logger logr.Logger + poolName string + + mu sync.Mutex + outstanding map[string]revocation + cooling []coolingDebit + pendingReclaim float64 +} + +// NewReclamationController constructs a controller. The caller is responsible for registering +// Confirm as the evictor's eviction-terminated listener. +func NewReclamationController( + cfg ReclamationConfig, + evictor InFlightEvictor, + clk clock.PassiveClock, + logger logr.Logger, + poolName string, +) *ReclamationController { + return &ReclamationController{ + evictor: evictor, + cfg: cfg, + clock: clk, + logger: logger.WithName("reclamation-controller"), + poolName: poolName, + outstanding: make(map[string]revocation), + } +} + +// GateOpen reports whether a new reclamation decision may run: stop-and-wait, open only when +// nothing is outstanding and nothing is cooling, i.e. every prior revocation has confirmed and had +// ConfirmationGrace to become gauge-visible. It first retires timed-out revocations and expired +// cooling debits. This is the cheap early-exit the dispatch cycle calls on every HoL break. +// +// A sliding-window relaxation (issue while outstanding < W) was prototyped and benchmarked with no +// measurable benefit: sizing is deficit-bound at the ceiling boundary, so the epoch rate, not the +// gate, limits reclaim throughput. See docs/flow-control-eviction.md, Alternatives. +func (c *ReclamationController) GateOpen(now time.Time) bool { + c.mu.Lock() + defer c.mu.Unlock() + c.sweepTimeoutsLocked(now) + c.expireCoolingLocked(now) + return len(c.outstanding) == 0 && len(c.cooling) == 0 +} + +// VictimPriority returns the priority of the next lease the selector would revoke. +func (c *ReclamationController) VictimPriority() (int, bool) { + return c.evictor.PeekVictimPriority() +} + +// Reclaim runs one sizing decision and issues the resulting revocations. The caller must have +// observed GateOpen() == true in the same dispatch cycle. +// +// All quantities are dimensionless, in the saturation gauge's own units: +// +// reclaimTarget = saturation - ceiling(blocked band with eligible demand) +// credit = saturation / inFlight (mean lease footprint estimate) +// n = max(1, ceil((reclaimTarget - pendingReclaim) / credit)), +// capped at MaxRevocationsPerDecision and at the evictable lease count +// +// demandPriority is the eligible demand band's priority, passed to the actuator as a strict upper +// bound on victim priority so that no victim in the decision sits at or above the demand band +// (the victim-head check in the trigger does not cover later victims of a multi-revocation +// decision). +func (c *ReclamationController) Reclaim(ctx context.Context, saturation, ceiling float64, demandPriority int) { + inFlight, evictable := c.evictor.Stats() + if inFlight <= 0 || evictable <= 0 { + return + } + credit := saturation / float64(inFlight) + if credit <= 0 || math.IsNaN(credit) || math.IsInf(credit, 0) { + return + } + + c.mu.Lock() + defer c.mu.Unlock() + + reclaimTarget := saturation - ceiling + netTarget := reclaimTarget - c.pendingReclaim + metrics.RecordFlowControlReclaimTarget(c.poolName, reclaimTarget) + // Under stop-and-wait pacing the gate opens only once every debit has expired, so + // pendingReclaim is zero here and this branch never executes through the dispatch path. The + // subtraction makes sizing safe independently of pacing: defense in depth for out-of-band + // calls, and the invariant any relaxed pacing scheme would rely on. + if netTarget < 0 || (netTarget == 0 && c.pendingReclaim > 0) { + return + } + + // The dispatch gate blocks on saturation >= ceiling, so unblocking requires reclaiming strictly + // more than the deficit. At the boundary (netTarget == 0 with nothing pending) the quotient + // rounds to zero, which would leave the band blocked with no reclamation; issue at least one + // revocation instead. + n := max(1, int(math.Ceil(netTarget/credit))) + if n > c.cfg.MaxRevocationsPerDecision { + n = c.cfg.MaxRevocationsPerDecision + } + if n > evictable { + n = evictable + } + + // The mutex is held across EvictN so that a confirmation racing the registration below blocks in + // Confirm until the revocation is present in the outstanding set. + evicted, err := c.evictor.EvictN(ctx, n, demandPriority) + if err != nil { + c.logger.Error(err, "Revocation batch failed", "requested", n) + return + } + now := c.clock.Now() + for _, requestID := range evicted { + c.outstanding[requestID] = revocation{credit: credit, issuedAt: now} + c.pendingReclaim += credit + } + metrics.RecordFlowControlRevocationsIssued(c.poolName, strconv.Itoa(demandPriority), len(evicted)) + metrics.RecordFlowControlPendingReclaim(c.poolName, c.pendingReclaim) + + c.logger.V(logutil.DEBUG).Info("Revocations issued", + "saturation", saturation, "ceiling", ceiling, + "reclaimTarget", reclaimTarget, "credit", credit, + "requested", n, "issued", len(evicted)) +} + +// Confirm records that an evicted request's stream has terminated. Safe for concurrent use; called +// from ext_proc handler goroutines via the evictor's eviction-terminated listener. IDs that are +// not outstanding (e.g. already retired by timeout) are ignored. +func (c *ReclamationController) Confirm(requestID string) { + c.mu.Lock() + defer c.mu.Unlock() + r, ok := c.outstanding[requestID] + if !ok { + return + } + delete(c.outstanding, requestID) + now := c.clock.Now() + c.startCoolingLocked(r, now) + metrics.RecordFlowControlRevocations(c.poolName, metrics.RevocationOutcomeConfirmed, 1) + metrics.RecordFlowControlRevocationConfirmationDuration(c.poolName, now.Sub(r.issuedAt)) +} + +// sweepTimeoutsLocked retires revocations whose confirmation never arrived within +// ConfirmationTimeout, so a hung stream cannot hold the pacing gate closed forever. +func (c *ReclamationController) sweepTimeoutsLocked(now time.Time) { + for requestID, r := range c.outstanding { + if now.Sub(r.issuedAt) < c.cfg.ConfirmationTimeout { + continue + } + delete(c.outstanding, requestID) + c.startCoolingLocked(r, now) + metrics.RecordFlowControlRevocations(c.poolName, metrics.RevocationOutcomeTimedOut, 1) + c.logger.V(logutil.DEFAULT).Info("Revocation confirmation timed out; treating as confirmed", + "requestID", requestID, "timeout", c.cfg.ConfirmationTimeout) + } +} + +// startCoolingLocked moves a confirmed revocation's debit into the reclaiming (cooling) stage: the +// stream is dead, but the freed capacity is assumed gauge-invisible for another ConfirmationGrace, +// so the debit keeps suppressing re-sizing until then. +func (c *ReclamationController) startCoolingLocked(r revocation, now time.Time) { + c.cooling = append(c.cooling, coolingDebit{credit: r.credit, expiresAt: now.Add(c.cfg.ConfirmationGrace)}) +} + +// expireCoolingLocked releases cooling debits whose grace has elapsed. +func (c *ReclamationController) expireCoolingLocked(now time.Time) { + kept := c.cooling[:0] + changed := false + for _, d := range c.cooling { + if now.Before(d.expiresAt) { + kept = append(kept, d) + continue + } + changed = true + c.pendingReclaim -= d.credit + } + c.cooling = kept + if changed { + if c.pendingReclaim < 0 { + c.pendingReclaim = 0 + } + metrics.RecordFlowControlPendingReclaim(c.poolName, c.pendingReclaim) + } +} + +// maybeReclaim evaluates the eviction trigger on a HoL-blocking break at band index breakIdx. +// The pacing gate is checked before any queue scan, so the common saturated-but-gated case costs a +// single comparison. Demand is eligible when a blocked band holds queued requests whose priority +// is strictly greater than the current victim head's priority (no same-band churn). +// +// priorities is ordered highest first, so once a band's priority is not strictly greater than the +// victim's, no later band can be eligible and the scan stops. +func (p *Processor) maybeReclaim( + ctx context.Context, + saturation float64, + priorities []int, + ceilings []float64, + breakIdx int, +) { + if !p.reclamation.GateOpen(p.clock.Now()) { + return + } + victimPriority, ok := p.reclamation.VictimPriority() + if !ok { + return // Nothing is evictable. + } + for j := breakIdx; j < len(priorities); j++ { + if priorities[j] <= victimPriority { + return + } + // A zero ceiling is a policy statement that the band must not dispatch regardless of load. + // No amount of reclamation can unblock it (saturation >= 0 always holds), so treating its + // queue as demand would revoke leases in a loop for no benefit. + if ceilings[j] <= 0 { + continue + } + // Each band's ceiling is checked directly; monotonicity across bands is not assumed. An + // unblocked band needs no eviction because it will dispatch on a subsequent cycle. + if saturation < ceilings[j] { + continue + } + band, err := p.registry.PriorityBandAccessor(priorities[j]) + if err != nil { + continue + } + if !bandHasQueuedItems(band) { + continue + } + p.reclamation.Reclaim(ctx, saturation, ceilings[j], priorities[j]) + return + } +} + +// bandHasQueuedItems reports whether any flow queue in the band is non-empty. +func bandHasQueuedItems(band flowcontrol.PriorityBandAccessor) bool { + found := false + band.IterateQueues(func(q flowcontrol.FlowQueueAccessor) bool { + if q.Len() > 0 { + found = true + return false + } + return true + }) + return found +} diff --git a/pkg/epp/flowcontrol/controller/internal/reclamation_test.go b/pkg/epp/flowcontrol/controller/internal/reclamation_test.go new file mode 100644 index 0000000000..cca8f94248 --- /dev/null +++ b/pkg/epp/flowcontrol/controller/internal/reclamation_test.go @@ -0,0 +1,375 @@ +/* +Copyright 2026 The llm-d Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package internal + +import ( + "context" + "fmt" + "sync" + "testing" + "time" + + "github.com/go-logr/logr" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + testclock "k8s.io/utils/clock/testing" + + fwkdl "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/datalayer" +) + +// fakeInFlightEvictor is a configurable InFlightEvictor for controller tests. +type fakeInFlightEvictor struct { + mu sync.Mutex + inFlight int + evictable int + victimPriority int + hasVictim bool + evictErr error + + evictCalls []int // n per EvictN call + boundCalls []int // priorityBound per EvictN call + nextID int + listener func(requestID string) +} + +func (f *fakeInFlightEvictor) EvictN(_ context.Context, n int, priorityBound int) ([]string, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.evictCalls = append(f.evictCalls, n) + f.boundCalls = append(f.boundCalls, priorityBound) + if f.evictErr != nil { + return nil, f.evictErr + } + if n > f.evictable { + n = f.evictable + } + ids := make([]string, 0, n) + for range n { + ids = append(ids, fmt.Sprintf("req-%d", f.nextID)) + f.nextID++ + } + f.evictable -= n + f.inFlight -= n + return ids, nil +} + +func (f *fakeInFlightEvictor) Stats() (int, int) { + f.mu.Lock() + defer f.mu.Unlock() + return f.inFlight, f.evictable +} + +func (f *fakeInFlightEvictor) PeekVictimPriority() (int, bool) { + f.mu.Lock() + defer f.mu.Unlock() + return f.victimPriority, f.hasVictim +} + +func (f *fakeInFlightEvictor) SetEvictionTerminatedListener(listener func(requestID string)) { + f.mu.Lock() + defer f.mu.Unlock() + f.listener = listener +} + +func (f *fakeInFlightEvictor) totalEvictCalls() []int { + f.mu.Lock() + defer f.mu.Unlock() + if len(f.evictCalls) == 0 { + return nil + } + out := make([]int, len(f.evictCalls)) + copy(out, f.evictCalls) + return out +} + +func newTestReclamationController( + evictor *fakeInFlightEvictor, + clk *testclock.FakeClock, + cfg ReclamationConfig, +) *ReclamationController { + return NewReclamationController(cfg, evictor, clk, logr.Discard(), "test-pool") +} + +var testReclamationConfig = ReclamationConfig{ + MaxRevocationsPerDecision: 2, + ConfirmationGrace: 100 * time.Millisecond, + ConfirmationTimeout: 5 * time.Second, +} + +func TestReclamationController_Sizing(t *testing.T) { + t.Parallel() + + testCases := []struct { + name string + inFlight int + evictable int + saturation float64 + ceiling float64 + expectedN []int // expected EvictN call args; empty means no call + }{ + { + // credit = 1.2/6 = 0.2; deficit = 0.2; n = 1. + name: "SmallDeficit_EvictsOne", + inFlight: 6, evictable: 6, saturation: 1.2, ceiling: 1.0, + expectedN: []int{1}, + }, + { + // credit = 2.0/4 = 0.5; deficit = 1.0; n = 2 = cap. + name: "DeepOverload_CappedByMaxPerDecision", + inFlight: 4, evictable: 4, saturation: 2.0, ceiling: 1.0, + expectedN: []int{2}, + }, + { + // n would be 2 but only 1 lease is evictable. + name: "CappedByEvictable", + inFlight: 4, evictable: 1, saturation: 2.0, ceiling: 1.0, + expectedN: []int{1}, + }, + { + // Saturation below the ceiling: no deficit, no eviction. + name: "NoDeficit_NoEviction", + inFlight: 6, evictable: 6, saturation: 0.9, ceiling: 1.0, + expectedN: nil, + }, + { + // The dispatch gate blocks at saturation == ceiling, but the deficit is exactly zero. + // The controller must still issue one revocation or the band deadlocks until churn. + name: "ExactBoundary_EvictsOne", + inFlight: 10, evictable: 10, saturation: 1.0, ceiling: 1.0, + expectedN: []int{1}, + }, + { + // No tracked leases: credit is undefined and there are no victims. + name: "ZeroLeases_NoEviction", + inFlight: 0, evictable: 0, saturation: 1.5, ceiling: 1.0, + expectedN: nil, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + clk := testclock.NewFakeClock(time.Now()) + evictor := &fakeInFlightEvictor{inFlight: tc.inFlight, evictable: tc.evictable} + rc := newTestReclamationController(evictor, clk, testReclamationConfig) + + rc.Reclaim(context.Background(), tc.saturation, tc.ceiling, 0) + + assert.Equal(t, tc.expectedN, evictor.totalEvictCalls(), "EvictN calls should match expected sizing") + }) + } +} + +func TestReclamationController_GateLifecycle(t *testing.T) { + t.Parallel() + clk := testclock.NewFakeClock(time.Now()) + evictor := &fakeInFlightEvictor{inFlight: 4, evictable: 4} + rc := newTestReclamationController(evictor, clk, testReclamationConfig) + + require.True(t, rc.GateOpen(clk.Now()), "gate should be open initially") + + // Issue a decision: deficit 1.0, credit 0.5, n = 2. + rc.Reclaim(context.Background(), 2.0, 1.0, 0) + require.Equal(t, []int{2}, evictor.totalEvictCalls(), "decision should issue revocations") + assert.False(t, rc.GateOpen(clk.Now()), "gate should be closed while revocations are outstanding") + + // One confirmation is not enough. + rc.Confirm("req-0") + assert.False(t, rc.GateOpen(clk.Now()), "gate should stay closed until all revocations confirm") + + // All confirmed, but grace has not elapsed. + rc.Confirm("req-1") + assert.False(t, rc.GateOpen(clk.Now()), "gate should stay closed during the grace window") + + // Grace elapsed: gate reopens. + clk.Step(testReclamationConfig.ConfirmationGrace + time.Millisecond) + assert.True(t, rc.GateOpen(clk.Now()), "gate should reopen after all confirmations plus grace") +} + +func TestReclamationController_ConfirmUnknownID_Ignored(t *testing.T) { + t.Parallel() + clk := testclock.NewFakeClock(time.Now()) + evictor := &fakeInFlightEvictor{inFlight: 4, evictable: 4} + rc := newTestReclamationController(evictor, clk, testReclamationConfig) + + require.NotPanics(t, func() { rc.Confirm("req-unknown") }) + assert.True(t, rc.GateOpen(clk.Now()), "unknown confirmations must not perturb the gate") +} + +func TestReclamationController_ConfirmationTimeout_ReopensGate(t *testing.T) { + t.Parallel() + clk := testclock.NewFakeClock(time.Now()) + evictor := &fakeInFlightEvictor{inFlight: 4, evictable: 4} + rc := newTestReclamationController(evictor, clk, testReclamationConfig) + + rc.Reclaim(context.Background(), 2.0, 1.0, 0) + require.False(t, rc.GateOpen(clk.Now()), "gate closed while outstanding") + + // Confirmation never arrives; the timeout retires the revocations. + clk.Step(testReclamationConfig.ConfirmationTimeout + time.Millisecond) + require.False(t, rc.GateOpen(clk.Now()), + "first check after timeout retires entries but grace restarts from retirement") + clk.Step(testReclamationConfig.ConfirmationGrace + time.Millisecond) + assert.True(t, rc.GateOpen(clk.Now()), "gate should reopen after timeout retirement plus grace") + + // A late confirmation for a timed-out revocation is a no-op: no second outcome, and the gate + // stays open. + require.NotPanics(t, func() { rc.Confirm("req-0") }) + assert.True(t, rc.GateOpen(clk.Now()), "late confirmation must not re-close the gate") +} + +func TestReclamationController_PendingDebit_SuppressesRepeatDecision(t *testing.T) { + t.Parallel() + clk := testclock.NewFakeClock(time.Now()) + evictor := &fakeInFlightEvictor{inFlight: 4, evictable: 4} + rc := newTestReclamationController(evictor, clk, testReclamationConfig) + + rc.Reclaim(context.Background(), 2.0, 1.0, 0) + require.Equal(t, []int{2}, evictor.totalEvictCalls()) + + // Even if a caller bypassed the gate, the pending debit covers the deficit and no further + // revocations are issued for the same signal. + rc.Reclaim(context.Background(), 2.0, 1.0, 0) + assert.Equal(t, []int{2}, evictor.totalEvictCalls(), + "a second decision against the same stale signal must not issue more revocations") +} + +// --- Integration: dispatchCycle -> maybeReclaim --- + +// withReclamation attaches a reclamation controller backed by the fake evictor to the harness +// processor. +func withReclamation(h *testHarness, evictor *fakeInFlightEvictor, cfg ReclamationConfig) { + h.processor.reclamation = NewReclamationController(cfg, evictor, h.clock, logr.Discard(), "test-pool") +} + +func TestDispatchCycle_HoLBlock_TriggersReclamation(t *testing.T) { + t.Parallel() + h := newTestHarness(t, testCleanupTick) + evictor := &fakeInFlightEvictor{inFlight: 6, evictable: 3, victimPriority: -1, hasVictim: true} + withReclamation(h, evictor, testReclamationConfig) + + q := h.addQueue(testFlow) // Priority 10. + require.NoError(t, q.Add(h.newTestItem("req-blocked", testFlow, testTTL))) + + h.saturationDetector.SaturationFunc = func(context.Context, []fwkdl.Endpoint) float64 { return 1.2 } + + dispatched := h.processor.dispatchCycle(h.ctx) + + assert.False(t, dispatched, "saturated cycle must not dispatch") + // credit = 1.2/6 = 0.2; deficit = 0.2; n = 1. + assert.Equal(t, []int{1}, evictor.totalEvictCalls(), "HoL blocking with eligible demand should revoke") +} + +func TestDispatchCycle_NoQueuedDemand_NoReclamation(t *testing.T) { + t.Parallel() + h := newTestHarness(t, testCleanupTick) + evictor := &fakeInFlightEvictor{inFlight: 6, evictable: 3, victimPriority: -1, hasVictim: true} + withReclamation(h, evictor, testReclamationConfig) + + h.addQueue(testFlow) // Registered band, but its queue is empty. + + h.saturationDetector.SaturationFunc = func(context.Context, []fwkdl.Endpoint) float64 { return 1.2 } + + h.processor.dispatchCycle(h.ctx) + assert.Empty(t, evictor.totalEvictCalls(), "no queued demand means nothing to reclaim for") +} + +func TestDispatchCycle_VictimNotStrictlyLower_NoReclamation(t *testing.T) { + t.Parallel() + h := newTestHarness(t, testCleanupTick) + // Victim priority equals the blocked band's priority: same-band churn is forbidden. + evictor := &fakeInFlightEvictor{inFlight: 6, evictable: 3, victimPriority: testFlow.Priority, hasVictim: true} + withReclamation(h, evictor, testReclamationConfig) + + q := h.addQueue(testFlow) + require.NoError(t, q.Add(h.newTestItem("req-blocked", testFlow, testTTL))) + + h.saturationDetector.SaturationFunc = func(context.Context, []fwkdl.Endpoint) float64 { return 1.2 } + + h.processor.dispatchCycle(h.ctx) + assert.Empty(t, evictor.totalEvictCalls(), "demand must dominate the victim priority strictly") +} + +func TestDispatchCycle_NoVictims_NoReclamation(t *testing.T) { + t.Parallel() + h := newTestHarness(t, testCleanupTick) + evictor := &fakeInFlightEvictor{inFlight: 0, evictable: 0, hasVictim: false} + withReclamation(h, evictor, testReclamationConfig) + + q := h.addQueue(testFlow) + require.NoError(t, q.Add(h.newTestItem("req-blocked", testFlow, testTTL))) + + h.saturationDetector.SaturationFunc = func(context.Context, []fwkdl.Endpoint) float64 { return 1.2 } + + h.processor.dispatchCycle(h.ctx) + assert.Empty(t, evictor.totalEvictCalls(), "no evictable leases means no decision") +} + +func TestDispatchCycle_GateClosed_SkipsDemandScan(t *testing.T) { + t.Parallel() + h := newTestHarness(t, testCleanupTick) + evictor := &fakeInFlightEvictor{inFlight: 6, evictable: 6, victimPriority: -1, hasVictim: true} + withReclamation(h, evictor, testReclamationConfig) + + q := h.addQueue(testFlow) + require.NoError(t, q.Add(h.newTestItem("req-blocked", testFlow, testTTL))) + + h.saturationDetector.SaturationFunc = func(context.Context, []fwkdl.Endpoint) float64 { return 1.2 } + + // First cycle issues; the gate then closes until confirmation. + h.processor.dispatchCycle(h.ctx) + require.Equal(t, []int{1}, evictor.totalEvictCalls()) + + // Repeated cycles against the same stale gauge must not issue more revocations. + for range 10 { + h.processor.dispatchCycle(h.ctx) + } + assert.Equal(t, []int{1}, evictor.totalEvictCalls(), "gate must suppress repeat decisions until confirmation") +} + +func TestMaybeReclaim_ZeroCeilingBand_NotEligibleDemand(t *testing.T) { + t.Parallel() + h := newTestHarness(t, testCleanupTick) + evictor := &fakeInFlightEvictor{inFlight: 6, evictable: 6, victimPriority: -1, hasVictim: true} + withReclamation(h, evictor, testReclamationConfig) + + q := h.addQueue(testFlow) + require.NoError(t, q.Add(h.newTestItem("req-blocked", testFlow, testTTL))) + + // A zero-ceiling band is permanently blocked by policy; no amount of reclamation can unblock + // it, so its queue must not count as eviction demand. + h.processor.maybeReclaim(h.ctx, 1.0, []int{testFlow.Priority}, []float64{0}, 0) + assert.Empty(t, evictor.totalEvictCalls(), "a zero-ceiling band must not trigger reclamation") + + // Positive control: the same demand against an attainable ceiling reclaims. + h.processor.maybeReclaim(h.ctx, 1.0, []int{testFlow.Priority}, []float64{0.9}, 0) + assert.Equal(t, []int{1}, evictor.totalEvictCalls(), "an attainable blocked ceiling must reclaim") +} + +func TestReclaim_PassesDemandPriorityAsVictimBound(t *testing.T) { + t.Parallel() + clk := testclock.NewFakeClock(time.Now()) + evictor := &fakeInFlightEvictor{inFlight: 4, evictable: 4} + rc := newTestReclamationController(evictor, clk, testReclamationConfig) + + rc.Reclaim(context.Background(), 2.0, 1.0, -1) + + evictor.mu.Lock() + defer evictor.mu.Unlock() + require.Equal(t, []int{-1}, evictor.boundCalls, + "the demand band's priority must reach the actuator as the victim bound") +} diff --git a/pkg/epp/flowcontrol/eviction/request_evictor.go b/pkg/epp/flowcontrol/eviction/request_evictor.go index 6cd1e39e5a..7b85f6e4ea 100644 --- a/pkg/epp/flowcontrol/eviction/request_evictor.go +++ b/pkg/epp/flowcontrol/eviction/request_evictor.go @@ -19,6 +19,7 @@ package eviction import ( "context" "net" + "sync" "time" "sigs.k8s.io/controller-runtime/pkg/log" @@ -37,6 +38,17 @@ type RequestEvictor struct { queue *EvictionQueue evictor Evictor evictionRegistry *EvictionRegistry + + // mu guards the fields below. + mu sync.Mutex + // pendingEvictions holds the IDs of requests whose eviction has been issued but whose stream has + // not yet terminated. Populated by EvictN before the eviction signal is sent; consumed exactly + // once by cleanupRequest when the stream terminates. + pendingEvictions map[string]struct{} + // evictionTerminatedListener, if set, is invoked with the request ID each time an evicted + // request's stream terminates. It may be called from ext_proc handler goroutines and must be + // safe for concurrent use. + evictionTerminatedListener func(requestID string) } // NewRequestEvictor creates a RequestEvictor with the given policies and evictor. @@ -53,9 +65,30 @@ func NewRequestEvictor( queue: NewEvictionQueue(ordering, filter), evictor: evictor, evictionRegistry: registry, + pendingEvictions: make(map[string]struct{}), } } +// SetEvictionTerminatedListener registers a callback invoked once per evicted request when its +// stream terminates. This is the confirmation signal for eviction pacing: the request is dead from +// the EPP's perspective, even though the engine may not have freed its resources yet. +// Must be called before the RequestEvictor is in use. +func (p *RequestEvictor) SetEvictionTerminatedListener(listener func(requestID string)) { + p.mu.Lock() + defer p.mu.Unlock() + p.evictionTerminatedListener = listener +} + +// PeekVictimPriority returns the priority of the next request that would be evicted, or false if +// no evictable requests exist. It does not modify the queue. +func (p *RequestEvictor) PeekVictimPriority() (priority int, ok bool) { + item := p.queue.Peek() + if item == nil { + return 0, false + } + return item.Priority, true +} + // EvictionRegistry returns the shared eviction registry. // The ext_proc Process() goroutine uses this to look up eviction channels for dispatched requests. func (p *RequestEvictor) EvictionRegistry() *EvictionRegistry { @@ -142,11 +175,14 @@ func (p *RequestEvictor) ResponseBody( "inFlight", p.queue.InFlightLen()) } -// EvictN attempts to evict up to n requests from the eviction queue. -// Each request is only removed from tracking after a successful eviction. If the eviction fails, -// the request remains in the queue for a future eviction attempt. +// EvictN attempts to evict up to n requests from the eviction queue, in victim-policy order, +// stopping at the first victim whose priority is not strictly below priorityBound. The bound +// enforces strict priority dominance across a whole multi-revocation decision: the victim head +// check alone does not cover later victims, which come off the heap at priorities at or above the +// head's. Each request is only removed from tracking after a successful eviction. If the eviction +// fails, the request remains in the queue for a future eviction attempt. // Returns the request IDs that were successfully evicted. -func (p *RequestEvictor) EvictN(ctx context.Context, n int) ([]string, error) { +func (p *RequestEvictor) EvictN(ctx context.Context, n int, priorityBound int) ([]string, error) { logger := log.FromContext(ctx) evicted := make([]string, 0, n) @@ -157,8 +193,23 @@ func (p *RequestEvictor) EvictN(ctx context.Context, n int) ([]string, error) { } item := items[0] + if item.Priority >= priorityBound { + // The heap orders victims lowest-priority first, so no remaining victim can be below the + // bound either. Re-tracking races a concurrent completion: if cleanupRequest ran between + // the pop and this Track, the entry is re-inserted dead and survives until a future + // decision evicts it, which then stalls the pacing gate for one ConfirmationTimeout. The + // window is sub-microsecond and the cost is bounded, so it is tolerated. + p.queue.Track(item) + break + } + + // Mark before the eviction signal is sent so that cleanupRequest observes the marker no matter + // how quickly the stream terminates. + p.markPendingEviction(item.RequestID) if err := p.evictor.Evict(ctx, item); err != nil { logger.Error(err, "Failed to evict request, re-tracking", "requestID", item.RequestID, "targetURL", item.TargetURL) + p.unmarkPendingEviction(item.RequestID) + // Same tolerated re-track-vs-completion window as the bound branch above. p.queue.Track(item) continue } @@ -179,12 +230,43 @@ func (p *RequestEvictor) Stats() (inFlight int, evictable int) { // cleanupRequest removes a request from all tracking structures. // If the evictor supports cleanup (e.g., ImmediateResponseEvictor), it also // cleans up evictor-internal state to prevent unbounded map growth. +// If the request had been evicted, the eviction-terminated listener is notified exactly once. func (p *RequestEvictor) cleanupRequest(requestID string) { p.queue.Untrack(requestID) p.evictionRegistry.Deregister(requestID) if c, ok := p.evictor.(EvictorWithCleanup); ok { c.Cleanup(requestID) } + if listener := p.consumePendingEviction(requestID); listener != nil { + listener(requestID) + } +} + +// markPendingEviction records that an eviction has been issued for the request. +func (p *RequestEvictor) markPendingEviction(requestID string) { + p.mu.Lock() + defer p.mu.Unlock() + p.pendingEvictions[requestID] = struct{}{} +} + +// unmarkPendingEviction removes the pending-eviction marker after a failed eviction attempt. +func (p *RequestEvictor) unmarkPendingEviction(requestID string) { + p.mu.Lock() + defer p.mu.Unlock() + delete(p.pendingEvictions, requestID) +} + +// consumePendingEviction removes the pending-eviction marker if present and returns the listener +// to notify, or nil if the request was not evicted or was already consumed. Consuming under the +// same lock as marking guarantees at-most-once notification across the idempotent cleanup paths. +func (p *RequestEvictor) consumePendingEviction(requestID string) func(requestID string) { + p.mu.Lock() + defer p.mu.Unlock() + if _, ok := p.pendingEvictions[requestID]; !ok { + return nil + } + delete(p.pendingEvictions, requestID) + return p.evictionTerminatedListener } // EvictorWithCleanup is an optional interface for evictors that maintain per-request state diff --git a/pkg/epp/flowcontrol/eviction/request_evictor_test.go b/pkg/epp/flowcontrol/eviction/request_evictor_test.go index 18d5b2687a..e9cc409ab1 100644 --- a/pkg/epp/flowcontrol/eviction/request_evictor_test.go +++ b/pkg/epp/flowcontrol/eviction/request_evictor_test.go @@ -102,7 +102,7 @@ func TestRequestEvictor_EvictN_ClosesEvictChannel(t *testing.T) { evictCh := re.EvictionRegistry().Get("req-1") require.NotNil(t, evictCh) - evicted, err := re.EvictN(ctx, 1) + evicted, err := re.EvictN(ctx, 1, 0) require.NoError(t, err) require.Equal(t, []string{"req-1"}, evicted) @@ -123,7 +123,7 @@ func TestRequestEvictor_EvictN_ReTracksOnFailure(t *testing.T) { ctx := context.Background() re.PreRequest(ctx, makeInferenceRequest("req-1", -1), makeSchedulingResult()) - evicted, err := re.EvictN(ctx, 1) + evicted, err := re.EvictN(ctx, 1, 0) require.NoError(t, err) assert.Empty(t, evicted) @@ -149,7 +149,7 @@ func TestRequestEvictor_RaceBetweenEvictAndCompletion(t *testing.T) { go func() { defer wg.Done() for range 5 { - _, _ = re.EvictN(ctx, 1) + _, _ = re.EvictN(ctx, 1, 0) time.Sleep(time.Millisecond) } }() @@ -204,7 +204,7 @@ func TestRequestEvictor_CleanupCallsEvictorCleanup(t *testing.T) { re.PreRequest(ctx, makeInferenceRequest("req-1", -1), makeSchedulingResult()) // Evict to create a sync.Once entry in the evictor. - _, _ = re.EvictN(ctx, 1) + _, _ = re.EvictN(ctx, 1, 0) // Complete the request — this should call cleanupRequest which calls evictor.Cleanup. // After cleanup, the sync.Once entry for "req-1" should be removed. @@ -222,7 +222,7 @@ func TestRequestEvictor_CleanupCallsEvictorCleanup(t *testing.T) { evictCh := re.EvictionRegistry().Get("req-1") require.NotNil(t, evictCh) - evicted, err := re.EvictN(ctx, 1) + evicted, err := re.EvictN(ctx, 1, 0) require.NoError(t, err) require.Len(t, evicted, 1) @@ -252,3 +252,102 @@ type failingEvictor struct{} func (e *failingEvictor) Evict(_ context.Context, _ *flowcontrol.EvictionItem) error { return assert.AnError } + +func TestRequestEvictor_EvictionTerminatedListener_NotifiedOncePerEvictedRequest(t *testing.T) { + t.Parallel() + re := NewRequestEvictor(&testOrdering{}, &acceptAllFilter{}, NewImmediateResponseEvictor()) + + var mu sync.Mutex + notified := make(map[string]int) + re.SetEvictionTerminatedListener(func(requestID string) { + mu.Lock() + defer mu.Unlock() + notified[requestID]++ + }) + + ctx := context.Background() + re.PreRequest(ctx, makeInferenceRequest("req-evicted", -1), makeSchedulingResult()) + re.PreRequest(ctx, makeInferenceRequest("req-natural", -1), makeSchedulingResult()) + + evicted, err := re.EvictN(ctx, 1, 0) + require.NoError(t, err) + require.Len(t, evicted, 1) + evictedID := evicted[0] + + // Stream termination for the evicted request notifies; the idempotent second cleanup does not. + re.ResponseBody(ctx, makeInferenceRequest(evictedID, -1), &requestcontrol.Response{EndOfStream: true}, nil) + re.ResponseBody(ctx, makeInferenceRequest(evictedID, -1), &requestcontrol.Response{EndOfStream: true}, nil) + + // A naturally completing request must not notify. + naturalID := "req-natural" + if evictedID == naturalID { + naturalID = "req-evicted" + } + re.ResponseBody(ctx, makeInferenceRequest(naturalID, -1), &requestcontrol.Response{EndOfStream: true}, nil) + + mu.Lock() + defer mu.Unlock() + assert.Equal(t, map[string]int{evictedID: 1}, notified, + "listener should fire exactly once, only for the evicted request") +} + +func TestRequestEvictor_FailedEviction_DoesNotNotifyListener(t *testing.T) { + t.Parallel() + re := NewRequestEvictor(&testOrdering{}, &acceptAllFilter{}, &failingEvictor{}) + + var mu sync.Mutex + var notifications []string + re.SetEvictionTerminatedListener(func(requestID string) { + mu.Lock() + defer mu.Unlock() + notifications = append(notifications, requestID) + }) + + ctx := context.Background() + re.PreRequest(ctx, makeInferenceRequest("req-1", -1), makeSchedulingResult()) + + evicted, err := re.EvictN(ctx, 1, 0) + require.NoError(t, err) + assert.Empty(t, evicted, "failed eviction should evict nothing") + + // The request later completes naturally: still no notification. + re.ResponseBody(ctx, makeInferenceRequest("req-1", -1), &requestcontrol.Response{EndOfStream: true}, nil) + + mu.Lock() + defer mu.Unlock() + assert.Empty(t, notifications, "a failed eviction must not mark the request as evicted") +} + +func TestRequestEvictor_PeekVictimPriority(t *testing.T) { + t.Parallel() + re := NewRequestEvictor(&testOrdering{}, &acceptAllFilter{}, &NoOpEvictor{}) + + _, ok := re.PeekVictimPriority() + assert.False(t, ok, "no victims when nothing is tracked") + + ctx := context.Background() + re.PreRequest(ctx, makeInferenceRequest("req-1", -2), makeSchedulingResult()) + + priority, ok := re.PeekVictimPriority() + require.True(t, ok) + assert.Equal(t, -2, priority) +} + +func TestRequestEvictor_EvictN_PriorityBound(t *testing.T) { + t.Parallel() + re := NewRequestEvictor(&testOrdering{}, &acceptAllFilter{}, NewImmediateResponseEvictor()) + + ctx := context.Background() + re.PreRequest(ctx, makeInferenceRequest("req-low", -5), makeSchedulingResult()) + re.PreRequest(ctx, makeInferenceRequest("req-high", -1), makeSchedulingResult()) + + // Demand at priority -1: only victims strictly below -1 may be evicted, even though n allows + // two. The -1 victim would be same-band churn against the demand. + evicted, err := re.EvictN(ctx, 2, -1) + require.NoError(t, err) + assert.Equal(t, []string{"req-low"}, evicted, "only victims strictly below the bound may be evicted") + + inFlight, evictable := re.Stats() + assert.Equal(t, 1, inFlight, "the at-bound victim must remain tracked") + assert.Equal(t, 1, evictable, "the at-bound victim must remain evictable for future decisions") +} diff --git a/pkg/epp/flowcontrol/integration_test.go b/pkg/epp/flowcontrol/integration_test.go index c3b03697a4..bc8d6311bc 100644 --- a/pkg/epp/flowcontrol/integration_test.go +++ b/pkg/epp/flowcontrol/integration_test.go @@ -695,7 +695,7 @@ func TestEvictionPipeline(t *testing.T) { sheddableCh := reg.Get("shed-1") require.NotNil(t, sheddableCh, "sheddable request should have an eviction channel") - evictedIDs, err := requestEvictor.EvictN(ctx, 1) + evictedIDs, err := requestEvictor.EvictN(ctx, 1, 0) require.NoError(t, err) require.Equal(t, []string{"shed-1"}, evictedIDs) diff --git a/pkg/epp/metrics/llm_d_router_metrics.go b/pkg/epp/metrics/llm_d_router_metrics.go index 26e9d06f32..eb231991fd 100644 --- a/pkg/epp/metrics/llm_d_router_metrics.go +++ b/pkg/epp/metrics/llm_d_router_metrics.go @@ -364,6 +364,52 @@ var ( }, []string{"outcome", "priority", "inference_pool"}, ) + + llmdFlowControlRevocationsIssuedTotal = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Subsystem: LLMDRouterEndpointPickerSubsystem, + Name: "flow_control_revocations_issued_total", + Help: metricsutil.HelpMsgWithStability("Total number of in-flight eviction revocations issued, labeled by the demand band's priority.", compbasemetrics.ALPHA), + }, + []string{"priority", "inference_pool"}, + ) + + llmdFlowControlRevocationsTotal = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Subsystem: LLMDRouterEndpointPickerSubsystem, + Name: "flow_control_revocations_total", + Help: metricsutil.HelpMsgWithStability("Total number of in-flight eviction revocations by terminal outcome (confirmed, timed_out). Every issued revocation eventually increments exactly one outcome.", compbasemetrics.ALPHA), + }, + []string{"outcome", "inference_pool"}, + ) + + llmdFlowControlReclaimTarget = prometheus.NewGaugeVec( + prometheus.GaugeOpts{ + Subsystem: LLMDRouterEndpointPickerSubsystem, + Name: "flow_control_reclaim_target", + Help: metricsutil.HelpMsgWithStability("Last computed reclamation deficit, in saturation-gauge units.", compbasemetrics.ALPHA), + }, + []string{"inference_pool"}, + ) + + llmdFlowControlPendingReclaim = prometheus.NewGaugeVec( + prometheus.GaugeOpts{ + Subsystem: LLMDRouterEndpointPickerSubsystem, + Name: "flow_control_pending_reclaim", + Help: metricsutil.HelpMsgWithStability("Sum of outstanding and cooling pending-reclaim debits, in saturation-gauge units.", compbasemetrics.ALPHA), + }, + []string{"inference_pool"}, + ) + + llmdFlowControlRevocationConfirmationDuration = prometheus.NewHistogramVec( + prometheus.HistogramOpts{ + Subsystem: LLMDRouterEndpointPickerSubsystem, + Name: "flow_control_revocation_confirmation_seconds", + Help: metricsutil.HelpMsgWithStability("Time from revocation issue to confirmed stream termination.", compbasemetrics.ALPHA), + Buckets: []float64{0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10}, + }, + []string{"inference_pool"}, + ) ) // --- llm-d Inference Model Rewrite Metrics --- diff --git a/pkg/epp/metrics/metrics.go b/pkg/epp/metrics/metrics.go index 044d557b88..20c23ceeb5 100644 --- a/pkg/epp/metrics/metrics.go +++ b/pkg/epp/metrics/metrics.go @@ -473,6 +473,11 @@ func Register(customCollectors ...prometheus.Collector) { metrics.Registry.MustRegister(flowControlRequestEnqueueDuration) metrics.Registry.MustRegister(llmdFlowControlRequestEnqueueDuration) metrics.Registry.MustRegister(llmdFlowControlRequestsTotal) + metrics.Registry.MustRegister(llmdFlowControlRevocationsIssuedTotal) + metrics.Registry.MustRegister(llmdFlowControlRevocationsTotal) + metrics.Registry.MustRegister(llmdFlowControlReclaimTarget) + metrics.Registry.MustRegister(llmdFlowControlPendingReclaim) + metrics.Registry.MustRegister(llmdFlowControlRevocationConfirmationDuration) metrics.Registry.MustRegister(inferenceModelRewriteDecisionsTotal) metrics.Registry.MustRegister(llmdInferenceModelRewriteDecisionsTotal) metrics.Registry.MustRegister(DataLayerPollErrorsTotal) @@ -543,6 +548,11 @@ func Reset() { flowControlDispatchCycleDuration.Reset() llmdFlowControlDispatchCycleDuration.Reset() llmdFlowControlRequestsTotal.Reset() + llmdFlowControlRevocationsIssuedTotal.Reset() + llmdFlowControlRevocationsTotal.Reset() + llmdFlowControlReclaimTarget.Reset() + llmdFlowControlPendingReclaim.Reset() + llmdFlowControlRevocationConfirmationDuration.Reset() inferenceModelRewriteDecisionsTotal.Reset() llmdInferenceModelRewriteDecisionsTotal.Reset() DataLayerPollErrorsTotal.Reset() @@ -874,6 +884,39 @@ func IncFlowControlRequestsTotal(outcome, priority, inferencePool string) { llmdFlowControlRequestsTotal.WithLabelValues(outcome, priority, inferencePool).Inc() } +// Terminal revocation outcomes for the flow control revocations counter. Every issued revocation +// eventually increments exactly one outcome. +const ( + RevocationOutcomeConfirmed = "confirmed" + RevocationOutcomeTimedOut = "timed_out" +) + +// RecordFlowControlRevocationsIssued counts revocations at issue time, labeled by the demand +// band's priority. +func RecordFlowControlRevocationsIssued(inferencePool, priority string, n int) { + llmdFlowControlRevocationsIssuedTotal.WithLabelValues(priority, inferencePool).Add(float64(n)) +} + +// RecordFlowControlRevocations increments the revocation counter for a terminal outcome. +func RecordFlowControlRevocations(inferencePool, outcome string, n int) { + llmdFlowControlRevocationsTotal.WithLabelValues(outcome, inferencePool).Add(float64(n)) +} + +// RecordFlowControlReclaimTarget records the last computed reclamation deficit. +func RecordFlowControlReclaimTarget(inferencePool string, target float64) { + llmdFlowControlReclaimTarget.WithLabelValues(inferencePool).Set(target) +} + +// RecordFlowControlPendingReclaim records the capacity debited for unconfirmed revocations. +func RecordFlowControlPendingReclaim(inferencePool string, pending float64) { + llmdFlowControlPendingReclaim.WithLabelValues(inferencePool).Set(pending) +} + +// RecordFlowControlRevocationConfirmationDuration records issue-to-confirmation latency. +func RecordFlowControlRevocationConfirmationDuration(inferencePool string, duration time.Duration) { + llmdFlowControlRevocationConfirmationDuration.WithLabelValues(inferencePool).Observe(duration.Seconds()) +} + // RecordInferenceModelRewriteDecision records the routing decision for InferenceModelRewrite. // The rewrite name and target come from configuration; only the source model name is // request-derived and needs bounding (a generic rule matches arbitrary model names). diff --git a/pkg/epp/metrics/metrics_test.go b/pkg/epp/metrics/metrics_test.go index ca0c00a685..aded10f76f 100644 --- a/pkg/epp/metrics/metrics_test.go +++ b/pkg/epp/metrics/metrics_test.go @@ -1460,3 +1460,33 @@ func TestInferenceModelRewriteDecisionsTotalMetric(t *testing.T) { require.NoError(t, err) require.Equal(t, 1.0, valNew) } + +func TestFlowControlEvictionMetrics(t *testing.T) { + RecordFlowControlRevocationsIssued("pool-evict", "10", 2) + RecordFlowControlRevocations("pool-evict", RevocationOutcomeConfirmed, 1) + RecordFlowControlRevocations("pool-evict", RevocationOutcomeTimedOut, 1) + RecordFlowControlReclaimTarget("pool-evict", 0.1) + RecordFlowControlPendingReclaim("pool-evict", 0.05) + RecordFlowControlRevocationConfirmationDuration("pool-evict", 5*time.Millisecond) + + for name, testdata := range map[string]string{ + "llm_d_epp_flow_control_revocations_issued_total": "testdata/llm_d_flow_control_revocations_issued_metric", + "llm_d_epp_flow_control_revocations_total": "testdata/llm_d_flow_control_revocations_metric", + "llm_d_epp_flow_control_reclaim_target": "testdata/llm_d_flow_control_reclaim_target_metric", + "llm_d_epp_flow_control_pending_reclaim": "testdata/llm_d_flow_control_pending_reclaim_metric", + } { + want, err := os.Open(testdata) + if err != nil { + t.Fatal(err) + } + defer want.Close() + if err := promtestutil.GatherAndCompare(metrics.Registry, want, name); err != nil { + t.Error(err) + } + } + + if got := promtestutil.CollectAndCount(llmdFlowControlRevocationConfirmationDuration, + "llm_d_epp_flow_control_revocation_confirmation_seconds"); got != 1 { + t.Errorf("confirmation duration histogram series = %d, want 1", got) + } +} diff --git a/pkg/epp/metrics/testdata/llm_d_flow_control_pending_reclaim_metric b/pkg/epp/metrics/testdata/llm_d_flow_control_pending_reclaim_metric new file mode 100644 index 0000000000..ed87a92ca6 --- /dev/null +++ b/pkg/epp/metrics/testdata/llm_d_flow_control_pending_reclaim_metric @@ -0,0 +1,3 @@ +# HELP llm_d_epp_flow_control_pending_reclaim [ALPHA] Sum of outstanding and cooling pending-reclaim debits, in saturation-gauge units. +# TYPE llm_d_epp_flow_control_pending_reclaim gauge +llm_d_epp_flow_control_pending_reclaim{inference_pool="pool-evict"} 0.05 diff --git a/pkg/epp/metrics/testdata/llm_d_flow_control_reclaim_target_metric b/pkg/epp/metrics/testdata/llm_d_flow_control_reclaim_target_metric new file mode 100644 index 0000000000..92f75401f2 --- /dev/null +++ b/pkg/epp/metrics/testdata/llm_d_flow_control_reclaim_target_metric @@ -0,0 +1,3 @@ +# HELP llm_d_epp_flow_control_reclaim_target [ALPHA] Last computed reclamation deficit, in saturation-gauge units. +# TYPE llm_d_epp_flow_control_reclaim_target gauge +llm_d_epp_flow_control_reclaim_target{inference_pool="pool-evict"} 0.1 diff --git a/pkg/epp/metrics/testdata/llm_d_flow_control_revocations_issued_metric b/pkg/epp/metrics/testdata/llm_d_flow_control_revocations_issued_metric new file mode 100644 index 0000000000..1c32b2bcbf --- /dev/null +++ b/pkg/epp/metrics/testdata/llm_d_flow_control_revocations_issued_metric @@ -0,0 +1,3 @@ +# HELP llm_d_epp_flow_control_revocations_issued_total [ALPHA] Total number of in-flight eviction revocations issued, labeled by the demand band's priority. +# TYPE llm_d_epp_flow_control_revocations_issued_total counter +llm_d_epp_flow_control_revocations_issued_total{inference_pool="pool-evict",priority="10"} 2 diff --git a/pkg/epp/metrics/testdata/llm_d_flow_control_revocations_metric b/pkg/epp/metrics/testdata/llm_d_flow_control_revocations_metric new file mode 100644 index 0000000000..04ee53dc85 --- /dev/null +++ b/pkg/epp/metrics/testdata/llm_d_flow_control_revocations_metric @@ -0,0 +1,4 @@ +# HELP llm_d_epp_flow_control_revocations_total [ALPHA] Total number of in-flight eviction revocations by terminal outcome (confirmed, timed_out). Every issued revocation eventually increments exactly one outcome. +# TYPE llm_d_epp_flow_control_revocations_total counter +llm_d_epp_flow_control_revocations_total{inference_pool="pool-evict",outcome="confirmed"} 1 +llm_d_epp_flow_control_revocations_total{inference_pool="pool-evict",outcome="timed_out"} 1 diff --git a/pkg/epp/requestcontrol/director.go b/pkg/epp/requestcontrol/director.go index 4e33e980f0..28afef87eb 100644 --- a/pkg/epp/requestcontrol/director.go +++ b/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 diff --git a/pkg/epp/server/runserver.go b/pkg/epp/server/runserver.go index fdd5b84bf8..3f31f1e43f 100644 --- a/pkg/epp/server/runserver.go +++ b/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 {