Merge branch 'main' into fix/flowcontrol-finalize-requests-2099

Author: LukeAVanDrie

.github/workflows/pr-kind-label.yaml

@@ -1,4 +1,4 @@
-name: Apply kind labels from PR body
+name: Apply kind/area labels and milestone from PR body
 
 on:
   pull_request_target:
@@ -23,14 +23,53 @@ jobs:
           gh pr view "$PR_NUMBER" --repo "$REPO" --json labels \
             -q '.labels[].name | select(startswith("kind/"))' | \
             while IFS= read -r label; do
-              gh pr edit "$PR_NUMBER" --repo "$REPO" --remove-label "$label"
+              gh pr edit "$PR_NUMBER" --repo "$REPO" --remove-label "$label" || echo "::warning::Failed to remove label '$label'"
             done
 
           # Extract /kind values from PR body and apply labels
-          printf '%s' "$PR_BODY" | tr -d '\r' | grep '^/kind ' | \
-            sed 's|^/kind ||' | sort -u | \
+          printf '%s' "$PR_BODY" | tr -d '\r' | sed -n 's|^/kind[[:space:]][[:space:]]*||p' | sed 's|[[:space:]]*$||' | sort -u | \
             while IFS= read -r kind; do
               [ -z "$kind" ] && continue
               gh pr edit "$PR_NUMBER" --repo "$REPO" --add-label "kind/$kind" || \
                 echo "::warning::Label 'kind/$kind' not found in repository"
             done
+
+      - name: Set area labels
+        env:
+          GH_TOKEN: ${{ github.token }}
+          PR_BODY: ${{ github.event.pull_request.body }}
+          PR_NUMBER: ${{ github.event.pull_request.number }}
+          REPO: ${{ github.repository }}
+        run: |
+          # Absence of a directive is a no-op. Area labels are also set by the
+          # /area comment command and clearing here would wipe those.
+          areas=$(printf '%s' "$PR_BODY" | tr -d '\r' | sed -n 's|^/area[[:space:]][[:space:]]*||p' | \
+            sed 's|[[:space:]]*$||' | sed '/^$/d' | sort -u)
+          [ -z "$areas" ] && exit 0
+
+          # Remove existing area/* labels
+          gh pr view "$PR_NUMBER" --repo "$REPO" --json labels \
+            -q '.labels[].name | select(startswith("area/"))' | \
+            while IFS= read -r label; do
+              gh pr edit "$PR_NUMBER" --repo "$REPO" --remove-label "$label" || echo "::warning::Failed to remove label '$label'"
+            done
+
+          printf '%s\n' "$areas" | \
+            while IFS= read -r area; do
+              gh pr edit "$PR_NUMBER" --repo "$REPO" --add-label "area/$area" || \
+                echo "::warning::Label 'area/$area' not found in repository"
+            done
+
+      - name: Set milestone
+        env:
+          GH_TOKEN: ${{ github.token }}
+          PR_BODY: ${{ github.event.pull_request.body }}
+          PR_NUMBER: ${{ github.event.pull_request.number }}
+          REPO: ${{ github.repository }}
+        run: |
+          # Use the last /milestone directive in the PR body
+          milestone=$(printf '%s' "$PR_BODY" | tr -d '\r' | sed -n 's|^/milestone[[:space:]][[:space:]]*||p' | \
+             sed 's|[[:space:]]*$||' | sed '/^$/d' | tail -n 1)
+          [ -z "$milestone" ] && exit 0
+          gh pr edit "$PR_NUMBER" --repo "$REPO" --milestone "$milestone" || \
+            echo "::warning::Milestone '$milestone' not found in repository"

CODEOWNERS

@@ -14,9 +14,9 @@
 *                                                                         @llm-d/router-maintainers
 
 # Flow control framework + plugins
-/pkg/epp/flowcontrol/                                                     @LukeAVanDrie @shmuelk @llm-d/router-maintainers
-/pkg/epp/framework/interface/flowcontrol/                                 @LukeAVanDrie @shmuelk @llm-d/router-maintainers
-/pkg/epp/framework/plugins/flowcontrol/                                   @LukeAVanDrie @shmuelk @llm-d/router-maintainers
+/pkg/epp/flowcontrol/                                                     @LukeAVanDrie @RishabhSaini @shmuelk @llm-d/router-maintainers
+/pkg/epp/framework/interface/flowcontrol/                                 @LukeAVanDrie @RishabhSaini @shmuelk @llm-d/router-maintainers
+/pkg/epp/framework/plugins/flowcontrol/                                   @LukeAVanDrie @RishabhSaini @shmuelk @llm-d/router-maintainers
 
 # Envoy ext_proc message layer
 /pkg/common/envoy/                                                        @shmuelk @llm-d/router-maintainers

Makefile.coord.mk

@@ -116,7 +116,7 @@ endif
 # Env vars forwarded into the e2e test container.
 E2E_ENV_VARS = COORDINATOR_IMAGE VLLM_IMAGE EPP_IMAGE VLLM_RENDER_IMAGE VLLM_RENDER_PORT \
                E2E_GATEWAY_PORT E2E_KEEP_CLUSTER_ON_FAILURE \
-               E2E_PRINT_COORDINATOR_LOGS K8S_CONTEXT READY_TIMEOUT MODEL_NAME
+               E2E_PRINT_LOGS K8S_CONTEXT READY_TIMEOUT MODEL_NAME
 BUILDER_E2E_ENV_FLAGS = $(foreach v,$(E2E_ENV_VARS),$(if $($(v)),-e '$(v)=$($(v))'))
 ifneq ($(filter command line environment,$(origin NAMESPACE)),)
 BUILDER_E2E_ENV_FLAGS += -e NAMESPACE=$(NAMESPACE)

README.coord.md

@@ -341,7 +341,7 @@ kubectl --context kind-e2e-coordinator-tests get pods
 |---|---|---|
 | `E2E_KEEP_CLUSTER_ON_FAILURE` | `false` | Preserve the Kind cluster when the suite fails |
 | `E2E_GATEWAY_PORT` | `30080` | Host port mapped to the gateway NodePort |
-| `E2E_PRINT_COORDINATOR_LOGS` | `false` | Print coordinator pod logs during the run |
+| `E2E_PRINT_LOGS` | `false` | Print all pod logs (coordinator, EPPs, Envoy, workers) for every spec, not just on failure |
 | `CONTAINER_RUNTIME` | `docker` | Container runtime used to load images into Kind (`docker` or `podman`) |
 | `EPP_IMAGE` | `ghcr.io/llm-d/llm-d-router-endpoint-picker:dev` | EPP image loaded into the Kind cluster |
 | `VLLM_IMAGE` | `ghcr.io/llm-d/llm-d-inference-sim:v0.10.2` | vLLM image loaded into the Kind cluster |

apix/config/v1alpha1/endpointpickerconfig_types.go

@@ -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
@@ -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"`
 
@@ -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

pkg/epp/config/loader/flowcontrol_test.go

@@ -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")
 			},
 		},
 
@@ -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 ---
 		{
@@ -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 ---
 		{

pkg/epp/flowcontrol/README.md

@@ -99,13 +99,14 @@ their justifications, please refer to the detailed documentation within the rele
     processing loop. It manages a worker that distributes incoming requests, apply policies, and dispatch
     requests to the backends. Its design focuses on high throughput and backpressure.
 
-2.  **Pluggable `Policy` Framework (`./framework`)**: This defines the core interfaces for all pluggable logic. It
-    features a two-tier policy system for `Fairness` (decisions *between* different flows) and `Ordering`
-    (decisions *within* a single flow) logic, covering both request dispatch and displacement.
-
-3.  **Extensible `SafeQueue` System (`./framework`)**: This defines the `framework.SafeQueue` interface for
-    concurrent-safe request storage. It uses a `QueueCapability` system that allows for diverse and extensible queue
-    implementations (e.g., FIFO, Priority Heap) while maintaining a stable interface.
+2.  **Pluggable `Policy` Contracts (`/pkg/epp/framework/interface/flowcontrol`)**: The plugin SDK defines the core
+    interfaces for all pluggable logic. It features a two-tier policy system for `Fairness` (decisions *between*
+    different flows) and `Ordering` (decisions *within* a single flow) logic, covering both request dispatch and
+    displacement.
+
+3.  **The `SafeQueue` (`./contracts`, `./queue`)**: The `contracts.SafeQueue` interface defines concurrent-safe
+    request storage, and `./queue` provides the priority-queue implementation whose dispatch order is determined by
+    the flow's configured `OrderingPolicy`.
 
 4.  **The `FlowRegistry` (`./registry`, `./contracts`)**: This is the stateful control plane of the system. It manages
     the configuration and lifecycle of all flows, policies, and queues. It presents a view of its state to the

pkg/epp/flowcontrol/controller/config.go

@@ -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
 	// 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.
@@ -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.
@@ -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,
 	}

pkg/epp/flowcontrol/controller/config_test.go

@@ -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,
@@ -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")
 			},
 		},
 		{