Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 17 additions & 10 deletions apix/config/v1alpha1/endpointpickerconfig_types.go
Original file line number Diff line number Diff line change
Expand Up @@ -351,23 +351,25 @@ type FlowControlConfig struct {
// levels. If this limit is exceeded, new requests will be rejected even if their specific
// priority band has capacity.
// Accepts standard Kubernetes resource quantities (e.g., "1Gi", "500M").
// If omitted, no global byte limit is enforced.
// If omitted or "0", no global byte limit is enforced.
MaxBytes *resource.Quantity `json:"maxBytes,omitempty"`

// +optional
// MaxRequests defines the global maximum number of concurrent requests across all priority
// levels. If this limit is exceeded, new requests will be rejected even if their specific
// priority band has capacity.
// Accepts standard Kubernetes resource quantities (e.g., "100", "1k").
// If omitted, no global request limit is enforced.
// If omitted or "0", no global request limit is enforced.
MaxRequests *resource.Quantity `json:"maxRequests,omitempty"`

// +optional
// DefaultRequestTTL serves as a fallback timeout for requests that do not specify their own
// deadline.
// It ensures that requests do not hang indefinitely in the queue.
// If 0 or omitted, it defaults to the client context deadline, meaning requests may wait
// indefinitely unless cancelled by the client.
// DefaultRequestTTL bounds how long a request may wait in the queue before it is evicted.
// If omitted, it defaults to 60s. This is a queue-wait budget: a request that cannot dispatch
// within it is shed with a retryable backpressure error rather than served late, and it is the
// only bound on queue wait when neither the client nor the gateway enforces a request deadline.
// Where such deadlines exist and fire sooner, they evict the request first (client disconnect).
// An explicit "0s" disables the TTL: requests then wait until client disconnect or controller
// shutdown.
DefaultRequestTTL *metav1.Duration `json:"defaultRequestTTL,omitempty"`

// +optional
Expand All @@ -382,7 +384,10 @@ type FlowControlConfig struct {
// +optional
// DefaultNegativePriorityBand allows you to define a separate template for priority levels
// strictly below zero. This enables designating negative-priority traffic as sheddable by
// setting lower capacity limits (e.g., maxBytes: "0" to drop immediately).
// setting lower capacity limits (e.g., a small maxRequests, so that under saturation the band
// fills quickly and subsequent requests are rejected immediately rather than queued).
// Note that a value of "0" is treated as unset and receives the system default, not zero
// capacity.
// If not specified, negative priorities fall back to DefaultPriorityBand.
DefaultNegativePriorityBand *PriorityBandConfig `json:"defaultNegativePriorityBand,omitempty"`

Expand Down Expand Up @@ -457,13 +462,15 @@ type PriorityBandConfig struct {
// +optional
// MaxBytes is the maximum number of bytes allowed for this priority band.
// Accepts standard Kubernetes resource quantities (e.g., "1Gi", "500M").
// If omitted, the system default is used.
// If omitted or "0", the system default (1G) is used. Per-band limits are always bounded; to
// effectively remove the bound, set an explicit large value.
MaxBytes *resource.Quantity `json:"maxBytes,omitempty"`

// +optional
// MaxRequests is the maximum number of concurrent requests allowed for this priority band.
// Accepts standard Kubernetes resource quantities (e.g., "100", "1k").
// If omitted, no request limit is enforced.
// If omitted or "0", the system default (5000) is used. Per-band limits are always bounded; to
// effectively remove the bound, set an explicit large value.
MaxRequests *resource.Quantity `json:"maxRequests,omitempty"`

// +optional
Expand Down
79 changes: 48 additions & 31 deletions pkg/epp/config/loader/flowcontrol_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -154,10 +154,13 @@ func TestBuildRegistryConfig(t *testing.T) {
apiConfig: nil,
assertion: func(t *testing.T, cfg *registry.Config) {
assert.Equal(t, uint64(0), cfg.MaxBytes, "Default global limit should be 0 (unlimited)")
assert.Equal(t, uint64(0), cfg.MaxRequests, "Default global request limit should be 0 (unlimited)")
require.NotNil(t, cfg.DefaultPriorityBand,
"Default priority band template should be initialized automatically")
assert.Equal(t, uint64(1_000_000_000) /* registry default: 1 GB */, cfg.DefaultPriorityBand.MaxBytes,
"Default template should use system default capacity")
assert.Equal(t, uint64(5000) /* registry default */, cfg.DefaultPriorityBand.MaxRequests,
"Default template should use the system default request-count capacity")
},
},

Expand Down Expand Up @@ -207,6 +210,49 @@ func TestBuildRegistryConfig(t *testing.T) {
"Explicit 0 in DefaultPriorityBand template should be treated as 'Use Default'")
},
},
{
name: "ShouldApplyDefault_WhenBandMaxRequestsIsNilOrZero",
apiConfig: &configapi.FlowControlConfig{
PriorityBands: []configapi.PriorityBandConfig{
{
Priority: 1,
// MaxRequests omitted
},
{
Priority: 2,
MaxRequests: ptr.To(resource.MustParse("0")), // Explicitly zero
},
{
Priority: 3,
MaxRequests: ptr.To(resource.MustParse("250")),
},
},
},
assertion: func(t *testing.T, cfg *registry.Config) {
require.Contains(t, cfg.PriorityBands, 1)
assert.Equal(t, uint64(5000) /* registry default */, cfg.PriorityBands[1].MaxRequests,
"Omitted MaxRequests (nil) should result in the system default (5000)")
require.Contains(t, cfg.PriorityBands, 2)
assert.Equal(t, uint64(5000) /* registry default */, cfg.PriorityBands[2].MaxRequests,
"Explicit MaxRequests (0) should be treated as 'Use Default' (5000)")
require.Contains(t, cfg.PriorityBands, 3)
assert.Equal(t, uint64(250), cfg.PriorityBands[3].MaxRequests,
"Explicit MaxRequests should be preserved")
},
},
{
name: "ShouldApplyDefault_WhenNegativeBandTemplateMaxRequestsIsZero",
apiConfig: &configapi.FlowControlConfig{
DefaultNegativePriorityBand: &configapi.PriorityBandConfig{
MaxRequests: ptr.To(resource.MustParse("0")), // Explicitly zero
},
},
assertion: func(t *testing.T, cfg *registry.Config) {
require.NotNil(t, cfg.DefaultNegativePriorityBand)
assert.Equal(t, uint64(5000) /* registry default */, cfg.DefaultNegativePriorityBand.MaxRequests,
"Explicit 0 in the negative band template should be treated as 'Use Default', not zero capacity")
},
},

// --- Validation Errors ---
{
Expand Down Expand Up @@ -299,37 +345,8 @@ func TestBuildRegistryConfig(t *testing.T) {
},

// --- MaxRequests: Defaulting Logic ---
{
name: "ShouldDefaultToZero_WhenMaxRequestsIsNil",
apiConfig: &configapi.FlowControlConfig{
PriorityBands: []configapi.PriorityBandConfig{
{
Priority: 1,
},
},
},
assertion: func(t *testing.T, cfg *registry.Config) {
require.Contains(t, cfg.PriorityBands, 1)
assert.Equal(t, uint64(0), cfg.PriorityBands[1].MaxRequests,
"Omitted MaxRequests should default to 0 (no request limit)")
},
},
{
name: "ShouldDefaultToZero_WhenMaxRequestsIsZero",
apiConfig: &configapi.FlowControlConfig{
PriorityBands: []configapi.PriorityBandConfig{
{
Priority: 1,
MaxRequests: ptr.To(resource.MustParse("0")),
},
},
},
assertion: func(t *testing.T, cfg *registry.Config) {
require.Contains(t, cfg.PriorityBands, 1)
assert.Equal(t, uint64(0), cfg.PriorityBands[1].MaxRequests,
"Explicit MaxRequests=0 should remain 0 (no request limit)")
},
},
// Nil and explicit-zero band MaxRequests defaulting is covered by
// ShouldApplyDefault_WhenBandMaxRequestsIsNilOrZero above.

// --- MaxRequests: Validation Errors ---
{
Expand Down
17 changes: 14 additions & 3 deletions pkg/epp/flowcontrol/controller/config.go
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,13 @@ import (
)

const (
// defaultRequestTTL is the default Time-To-Live applied to queued requests when the configuration
// does not specify one. It is a queue-wait budget: a request still undispatched after this long is
// shed with a retryable backpressure signal instead of being served with severely degraded
// time-to-first-token. It is also the only bound on queue wait when neither the client nor the
// gateway enforces a request deadline (the well-lit guides configure no gateway request timeout);
// where such deadlines exist and fire sooner, context cancellation evicts the request first.
defaultRequestTTL = 60 * time.Second

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Requests can come with a timeout in their context, right? if so, what happens if this is lower than the timeout set by the client?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The two deadlines compose with minimum-wins semantics.

We derive the queue context via context.WithDeadlineCause(ctx, enqueueTime.Add(TTL), ErrTTLExpired), where ctx is the inbound request context.

  • On client deadline sooner than the TTL, the parent's deadline wins and fires first. Crucially, the ErrTTLExpired cause only attaches if the TTL's own deadline is the one that expires; when the parent fires, the cause is the parent's cancellation. So it flows down the QueueOutcomeEvictedContextCancelled path and is reported as client disconnect/timeout, not as a TTL shed. Attribution stays honest in metrics and status mapping.
  • On client deadline later (or absent), the 60s TTL fires first with the ErrTTLExpired cause → EvictedTTL → the TTLExpired dropped-reason response. The client's larger budget is deliberately cut at 60s of queue wait. The remaining budget is preserved for generation. I.e., TTL is "queue-wait budget" not "client-deadline proxy".

In short, if the client sets a higher timeout, we still enforce maximum of 60s queue-wait budget by default.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Long-term I would like to be able to express TTL at the per-priority (InfObj) and request level, but that is not in scope for this change as it would be a new transport mechanism.

// defaultExpiryCleanupInterval is the default frequency for scanning for expired items.
defaultExpiryCleanupInterval = 1 * time.Second
// defaultEnqueueChannelBufferSize is the default size of a worker's incoming request buffer.
Expand All @@ -32,9 +39,12 @@ const (

// Config holds the configuration for the `FlowController`.
type Config struct {
// DefaultRequestTTL is the default Time-To-Live applied to requests that do not
// specify their own TTL hint.
// Optional: If zero, no TTL is applied by default and we rely solely on request context cancellation.
// DefaultRequestTTL is the default Time-To-Live applied to requests that do not specify their own
// TTL hint. Because the admission adapter does not currently plumb a per-request hint, this value
// governs every request entering flow control.
// Optional: Defaults to `defaultRequestTTL` (60s). An explicit zero disables the TTL entirely, in
// which case queued requests are bounded only by request context cancellation (client disconnect
// or gateway timeout).
DefaultRequestTTL time.Duration

// ExpiryCleanupInterval is the interval at which each processor scans its queues for expired items.
Expand Down Expand Up @@ -76,6 +86,7 @@ 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{
DefaultRequestTTL: defaultRequestTTL,
ExpiryCleanupInterval: defaultExpiryCleanupInterval,
EnqueueChannelBufferSize: defaultEnqueueChannelBufferSize,
}
Expand Down
14 changes: 13 additions & 1 deletion pkg/epp/flowcontrol/controller/config_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,18 @@ func TestNewConfig(t *testing.T) {
name: "Defaults_ShouldBeApplied_WhenNoOptionsProvided",
opts: nil,
expectErr: false,
expectedCfg: Config{
DefaultRequestTTL: defaultRequestTTL,
ExpiryCleanupInterval: defaultExpiryCleanupInterval,
EnqueueChannelBufferSize: defaultEnqueueChannelBufferSize,
},
},
{
name: "ZeroDefaultRequestTTL_ShouldDisableTTL",
opts: []ConfigOption{
WithDefaultRequestTTL(0),
},
expectErr: false,
expectedCfg: Config{
DefaultRequestTTL: 0,
ExpiryCleanupInterval: defaultExpiryCleanupInterval,
Expand Down Expand Up @@ -136,7 +148,7 @@ func TestNewConfigFromAPI(t *testing.T) {
"ExpiryCleanupInterval should be defaulted")
assert.Equal(t, defaultEnqueueChannelBufferSize, cfg.EnqueueChannelBufferSize,
"EnqueueChannelBufferSize should be defaulted")
assert.Equal(t, time.Duration(0), cfg.DefaultRequestTTL, "DefaultRequestTTL should default to 0 (disabled)")
assert.Equal(t, defaultRequestTTL, cfg.DefaultRequestTTL, "DefaultRequestTTL should be defaulted when unset")
},
},
{
Expand Down
28 changes: 20 additions & 8 deletions pkg/epp/flowcontrol/registry/config.go
Original file line number Diff line number Diff line change
Expand Up @@ -39,9 +39,14 @@ const (
)

const (
// defaultPriorityBandMaxBytes is the default global capacity for a priority band if not explicitly configured.
// It is set to 1 GB.
// defaultPriorityBandMaxBytes is the default byte-size capacity for a priority band if not explicitly
// configured. It is set to 1 GB.
defaultPriorityBandMaxBytes uint64 = 1_000_000_000
// defaultPriorityBandMaxRequests is the default request-count capacity for a priority band if not
// explicitly configured. Together with the default request TTL (60s), it only binds when a band's
// arrival rate under a dispatch halt exceeds maxRequests/TTL (~83 req/s); below that, TTL expiry
// bounds occupancy first.
defaultPriorityBandMaxRequests uint64 = 5000
// defaultFlowGCTimeout is the default duration of inactivity after which an idle flow is garbage collected.
// This also serves as the interval for the periodic garbage collection scan.
defaultFlowGCTimeout time.Duration = 5 * time.Minute
Expand Down Expand Up @@ -85,14 +90,15 @@ type Config struct {
DefaultPriorityBand *PriorityBandConfig

// DefaultNegativePriorityBand is an optional template for dynamically provisioning priority bands when a request
// arrives with a priority level strictly below zero. This allows setting lower capacity limits (or zero) for
// negative-priority traffic to designate it as sheddable.
// arrives with a priority level strictly below zero. This allows setting lower capacity limits for
// negative-priority traffic to designate it as sheddable (a value of 0 is treated as unset and receives the
// default).
// If nil, negative priorities fall back to DefaultPriorityBand.
DefaultNegativePriorityBand *PriorityBandConfig

// FlowGCTimeout defines the interval at which the registry scans for and garbage collects idle flows.
// A flow is collected if it has been observed to be Idle for at least one full scan interval.
// Optional: Defaults to `defaultFlowGCTimeout` (1 hour).
// Optional: Defaults to `defaultFlowGCTimeout` (5 minutes).
FlowGCTimeout time.Duration

// PriorityBandGCTimeout defines the duration of inactivity after which a dynamically provisioned priority band
Expand Down Expand Up @@ -136,12 +142,15 @@ type PriorityBandConfig struct {
FairnessPolicy flowcontrol.FairnessPolicy

// MaxBytes defines the maximum total byte size for this priority band.
// Optional: Defaults to defaultPriorityBandMaxBytes (1 GB).
// Optional: Defaults to defaultPriorityBandMaxBytes (1 GB). A value of 0 is treated as unset and receives the
// default; per-band limits are always bounded, unlike the optional global limits. To effectively remove the
// bound, set an explicit large value.
MaxBytes uint64

// MaxRequests defines the maximum total request count for this priority band.
// A value of 0 signifies no request-count limit is enforced.
// Optional: Defaults to defaultPriorityBandMaxRequests (5000).
// Optional: Defaults to defaultPriorityBandMaxRequests (5000). A value of 0 is treated as unset and receives
// the default; per-band limits are always bounded, unlike the optional global limits. To effectively remove
// the bound, set an explicit large value.
MaxRequests uint64
}

Expand Down Expand Up @@ -386,6 +395,9 @@ func (p *PriorityBandConfig) applyDefaults(defaults PriorityBandPolicyDefaults)
if p.MaxBytes == 0 {
p.MaxBytes = defaultPriorityBandMaxBytes
}
if p.MaxRequests == 0 {
p.MaxRequests = defaultPriorityBandMaxRequests
}
if p.FairnessPolicy == nil {
if defaults.FairnessPolicy == nil {
return fmt.Errorf("no default fairness policy provided and none set on band %d", p.Priority)
Expand Down
14 changes: 12 additions & 2 deletions pkg/epp/flowcontrol/registry/config_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -88,6 +88,7 @@ func TestNewConfig(t *testing.T) {
require.NotNil(t, band.FairnessPolicy)
assert.Equal(t, DefaultFairnessPolicyRef, band.FairnessPolicy.TypedName().Name)
assert.Equal(t, defaultPriorityBandMaxBytes, band.MaxBytes)
assert.Equal(t, defaultPriorityBandMaxRequests, band.MaxRequests)
},
},
{
Expand Down Expand Up @@ -129,9 +130,17 @@ func TestNewConfig(t *testing.T) {
assertion: func(t *testing.T, cfg *Config) {
assert.Len(t, cfg.PriorityBands, 1, "PriorityBands should contain only the always-injected priority 0")
assert.Contains(t, cfg.PriorityBands, 0, "PriorityBands should contain priority 0")
assert.Equal(t, defaultPriorityBandMaxBytes, cfg.PriorityBands[0].MaxBytes,
"Auto-provisioned priority 0 band should receive the byte-size default")
assert.Equal(t, defaultPriorityBandMaxRequests, cfg.PriorityBands[0].MaxRequests,
"Auto-provisioned priority 0 band should receive the request-count default")
require.NotNil(t, cfg.DefaultPriorityBand, "DefaultPriorityBand template must be initialized")
assert.NotNil(t, cfg.DefaultPriorityBand.FairnessPolicy)
assert.Equal(t, DefaultFairnessPolicyRef, cfg.DefaultPriorityBand.FairnessPolicy.TypedName().Name)
assert.Equal(t, defaultPriorityBandMaxBytes, cfg.DefaultPriorityBand.MaxBytes,
"Dynamic provisioning template should receive the byte-size default")
assert.Equal(t, defaultPriorityBandMaxRequests, cfg.DefaultPriorityBand.MaxRequests,
"Dynamic provisioning template should receive the request-count default")
},
},
{
Expand Down Expand Up @@ -372,15 +381,16 @@ func TestNewConfig_DefaultNegativePriorityBand(t *testing.T) {
"Defaults should be applied to DefaultNegativePriorityBand")
})

t.Run("ShouldAllowZeroMaxBytes_ForSheddableTraffic", func(t *testing.T) {
t.Run("ShouldApplyCapacityDefaults_ToEmptyNegativeBandTemplate", func(t *testing.T) {
t.Parallel()
cfg, err := NewConfig(defaults,
WithDefaultNegativePriorityBand(&PriorityBandConfig{}),
)
require.NoError(t, err)
require.NotNil(t, cfg.DefaultNegativePriorityBand)
// MaxBytes=0 gets defaulted to 1GB via applyDefaults
// Zero capacity values are treated as unset: bands are always bounded, never zero-capacity.
assert.Equal(t, defaultPriorityBandMaxBytes, cfg.DefaultNegativePriorityBand.MaxBytes)
assert.Equal(t, defaultPriorityBandMaxRequests, cfg.DefaultNegativePriorityBand.MaxRequests)
})

t.Run("ShouldCloneNegativeBandTemplate", func(t *testing.T) {
Expand Down
8 changes: 7 additions & 1 deletion pkg/epp/requestcontrol/admission.go
Original file line number Diff line number Diff line change
Expand Up @@ -194,7 +194,13 @@ func (r *flowControlRequest) ID() string {
}
return r.inferenceRequest.RequestID
}
func (r *flowControlRequest) InitialEffectiveTTL() time.Duration { return 0 } // Use controller default.

// InitialEffectiveTTL returns 0 to defer to the controller-level default TTL, which is therefore the only TTL
// source for every request today.
// TODO(https://github.com/llm-d/llm-d-router/issues/1090): plumb more specific TTL scopes and resolve
// most-specific-wins: per-request (clamped), then per-band (effectively priority band config, eventually
// codifiable in the InferenceObjective CRD), then the controller default.
func (r *flowControlRequest) InitialEffectiveTTL() time.Duration { return 0 }
func (r *flowControlRequest) ByteSize() uint64 { return r.requestByteSize }

func (r *flowControlRequest) InferenceRequest() *scheduling.InferenceRequest {
Expand Down
Loading