Skip to content

Commit 623b338

Browse files
authored
Merge branch 'llm-d:main' into main
2 parents 405797a + 6181fd9 commit 623b338

124 files changed

Lines changed: 5832 additions & 837 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

.github/workflows/ci-pr-checks.yaml

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -113,6 +113,16 @@ jobs:
113113
GO_BUILD_CACHE_VOL: ${{ steps.go-cache.outputs.build }}
114114
run: make test-unit
115115

116+
# Anti-rot: execute the benchmark bodies once so runtime breakage (panics,
117+
# deadlocks) is caught. `go test` only compiles benchmarks; it never runs
118+
# them without -bench.
119+
- name: Run benchmark smoke
120+
shell: bash
121+
env:
122+
GO_MOD_CACHE_VOL: ${{ steps.go-cache.outputs.mod }}
123+
GO_BUILD_CACHE_VOL: ${{ steps.go-cache.outputs.build }}
124+
run: make bench-smoke
125+
116126
- name: Run hermetic integration tests
117127
shell: bash
118128
env:

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

Lines changed: 43 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
name: Apply kind labels from PR body
1+
name: Apply kind/area labels and milestone from PR body
22

33
on:
44
pull_request_target:
@@ -23,14 +23,53 @@ jobs:
2323
gh pr view "$PR_NUMBER" --repo "$REPO" --json labels \
2424
-q '.labels[].name | select(startswith("kind/"))' | \
2525
while IFS= read -r label; do
26-
gh pr edit "$PR_NUMBER" --repo "$REPO" --remove-label "$label"
26+
gh pr edit "$PR_NUMBER" --repo "$REPO" --remove-label "$label" || echo "::warning::Failed to remove label '$label'"
2727
done
2828
2929
# Extract /kind values from PR body and apply labels
30-
printf '%s' "$PR_BODY" | tr -d '\r' | grep '^/kind ' | \
31-
sed 's|^/kind ||' | sort -u | \
30+
printf '%s' "$PR_BODY" | tr -d '\r' | sed -n 's|^/kind[[:space:]][[:space:]]*||p' | sed 's|[[:space:]]*$||' | sort -u | \
3231
while IFS= read -r kind; do
3332
[ -z "$kind" ] && continue
3433
gh pr edit "$PR_NUMBER" --repo "$REPO" --add-label "kind/$kind" || \
3534
echo "::warning::Label 'kind/$kind' not found in repository"
3635
done
36+
37+
- name: Set area labels
38+
env:
39+
GH_TOKEN: ${{ github.token }}
40+
PR_BODY: ${{ github.event.pull_request.body }}
41+
PR_NUMBER: ${{ github.event.pull_request.number }}
42+
REPO: ${{ github.repository }}
43+
run: |
44+
# Absence of a directive is a no-op. Area labels are also set by the
45+
# /area comment command and clearing here would wipe those.
46+
areas=$(printf '%s' "$PR_BODY" | tr -d '\r' | sed -n 's|^/area[[:space:]][[:space:]]*||p' | \
47+
sed 's|[[:space:]]*$||' | sed '/^$/d' | sort -u)
48+
[ -z "$areas" ] && exit 0
49+
50+
# Remove existing area/* labels
51+
gh pr view "$PR_NUMBER" --repo "$REPO" --json labels \
52+
-q '.labels[].name | select(startswith("area/"))' | \
53+
while IFS= read -r label; do
54+
gh pr edit "$PR_NUMBER" --repo "$REPO" --remove-label "$label" || echo "::warning::Failed to remove label '$label'"
55+
done
56+
57+
printf '%s\n' "$areas" | \
58+
while IFS= read -r area; do
59+
gh pr edit "$PR_NUMBER" --repo "$REPO" --add-label "area/$area" || \
60+
echo "::warning::Label 'area/$area' not found in repository"
61+
done
62+
63+
- name: Set milestone
64+
env:
65+
GH_TOKEN: ${{ github.token }}
66+
PR_BODY: ${{ github.event.pull_request.body }}
67+
PR_NUMBER: ${{ github.event.pull_request.number }}
68+
REPO: ${{ github.repository }}
69+
run: |
70+
# Use the last /milestone directive in the PR body
71+
milestone=$(printf '%s' "$PR_BODY" | tr -d '\r' | sed -n 's|^/milestone[[:space:]][[:space:]]*||p' | \
72+
sed 's|[[:space:]]*$||' | sed '/^$/d' | tail -n 1)
73+
[ -z "$milestone" ] && exit 0
74+
gh pr edit "$PR_NUMBER" --repo "$REPO" --milestone "$milestone" || \
75+
echo "::warning::Milestone '$milestone' not found in repository"

CODEOWNERS

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -14,9 +14,9 @@
1414
* @llm-d/router-maintainers
1515

1616
# Flow control framework + plugins
17-
/pkg/epp/flowcontrol/ @LukeAVanDrie @shmuelk @llm-d/router-maintainers
18-
/pkg/epp/framework/interface/flowcontrol/ @LukeAVanDrie @shmuelk @llm-d/router-maintainers
19-
/pkg/epp/framework/plugins/flowcontrol/ @LukeAVanDrie @shmuelk @llm-d/router-maintainers
17+
/pkg/epp/flowcontrol/ @LukeAVanDrie @RishabhSaini @shmuelk @llm-d/router-maintainers
18+
/pkg/epp/framework/interface/flowcontrol/ @LukeAVanDrie @RishabhSaini @shmuelk @llm-d/router-maintainers
19+
/pkg/epp/framework/plugins/flowcontrol/ @LukeAVanDrie @RishabhSaini @shmuelk @llm-d/router-maintainers
2020

2121
# Envoy ext_proc message layer
2222
/pkg/common/envoy/ @shmuelk @llm-d/router-maintainers

Makefile

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -315,6 +315,11 @@ bench-tokenizer: image-build-builder ## Run external tokenizer + scorer benchmar
315315
@printf "Run 'EXTERNAL_TOKENIZER_ENABLED=true KV_CACHE_ENABLED=true make env-dev-kind' first.\n\n"
316316
$(BUILDER_RUN_CLUSTER) 'go test -bench=. -benchmem -count=5 -timeout=5m ./test/profiling/tokenizerbench/'
317317

318+
.PHONY: bench-smoke
319+
bench-smoke: image-build-builder ## Smoke-run the flowcontrol benchmarks once (-benchtime=1x) to catch runtime rot
320+
@printf "\033[33;1m==== Running Flow Control Benchmark Smoke ====\033[0m\n"
321+
$(BUILDER_RUN) 'go test -run=^$$ -bench=. -benchtime=1x -timeout=5m ./pkg/epp/flowcontrol/benchmark/...'
322+
318323
.PHONY: post-deploy-test
319324
post-deploy-test: ## Run post deployment tests
320325
@echo "Success!"

Makefile.coord.mk

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -116,7 +116,7 @@ endif
116116
# Env vars forwarded into the e2e test container.
117117
E2E_ENV_VARS = COORDINATOR_IMAGE VLLM_IMAGE EPP_IMAGE VLLM_RENDER_IMAGE VLLM_RENDER_PORT \
118118
E2E_GATEWAY_PORT E2E_KEEP_CLUSTER_ON_FAILURE \
119-
E2E_PRINT_COORDINATOR_LOGS K8S_CONTEXT READY_TIMEOUT MODEL_NAME
119+
E2E_PRINT_LOGS K8S_CONTEXT READY_TIMEOUT MODEL_NAME
120120
BUILDER_E2E_ENV_FLAGS = $(foreach v,$(E2E_ENV_VARS),$(if $($(v)),-e '$(v)=$($(v))'))
121121
ifneq ($(filter command line environment,$(origin NAMESPACE)),)
122122
BUILDER_E2E_ENV_FLAGS += -e NAMESPACE=$(NAMESPACE)

README.coord.md

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -341,7 +341,7 @@ kubectl --context kind-e2e-coordinator-tests get pods
341341
|---|---|---|
342342
| `E2E_KEEP_CLUSTER_ON_FAILURE` | `false` | Preserve the Kind cluster when the suite fails |
343343
| `E2E_GATEWAY_PORT` | `30080` | Host port mapped to the gateway NodePort |
344-
| `E2E_PRINT_COORDINATOR_LOGS` | `false` | Print coordinator pod logs during the run |
344+
| `E2E_PRINT_LOGS` | `false` | Print all pod logs (coordinator, EPPs, Envoy, workers) for every spec, not just on failure |
345345
| `CONTAINER_RUNTIME` | `docker` | Container runtime used to load images into Kind (`docker` or `podman`) |
346346
| `EPP_IMAGE` | `ghcr.io/llm-d/llm-d-router-endpoint-picker:dev` | EPP image loaded into the Kind cluster |
347347
| `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

Lines changed: 17 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -351,23 +351,25 @@ type FlowControlConfig struct {
351351
// levels. If this limit is exceeded, new requests will be rejected even if their specific
352352
// priority band has capacity.
353353
// Accepts standard Kubernetes resource quantities (e.g., "1Gi", "500M").
354-
// If omitted, no global byte limit is enforced.
354+
// If omitted or "0", no global byte limit is enforced.
355355
MaxBytes *resource.Quantity `json:"maxBytes,omitempty"`
356356

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

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

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

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

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

469476
// +optional

cmd/epp/runner/feature_gate_test.go

Lines changed: 104 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,10 +18,15 @@ package runner
1818

1919
import (
2020
"context"
21+
"os"
2122
"testing"
2223

2324
"github.com/stretchr/testify/require"
2425

26+
"github.com/llm-d/llm-d-router/pkg/epp/datastore"
27+
"github.com/llm-d/llm-d-router/pkg/epp/flowcontrol"
28+
"github.com/llm-d/llm-d-router/pkg/epp/flowcontrol/contracts"
29+
"github.com/llm-d/llm-d-router/pkg/epp/requestcontrol"
2530
runserver "github.com/llm-d/llm-d-router/pkg/epp/server"
2631
)
2732

@@ -50,3 +55,102 @@ featureGates:
5055
require.False(t, r.featureGates[runserver.HAPopulateNonLeaderDatastoreFeatureGate])
5156
})
5257
}
58+
59+
// TestFlowControlFeatureGateAdmissionControlWiring exercises the flowControl feature gate through
60+
// the production config path (parseConfigurationPhaseOne -> parseConfigurationPhaseTwo ->
61+
// initAdmissionControl) in both directions:
62+
// - gate on: the FlowControlAdmissionController is wired, the loader emits a non-nil
63+
// FlowControlConfig, and the flow registry is exposed as the priority band control plane;
64+
// - gate off: the LegacyAdmissionController is wired and no flow control config is built.
65+
//
66+
// The "no featureGates stanza" case reads the gate's registered default from the parsed
67+
// feature-gate map instead of hardcoding it, so this test keeps passing unchanged when the gate
68+
// flips to enabled-by-default (#2104) and pins that the flip actually changes the default wiring.
69+
func TestFlowControlFeatureGateAdmissionControlWiring(t *testing.T) {
70+
// The deprecated ENABLE_EXPERIMENTAL_FLOW_CONTROL_LAYER env var appends the gate to the config
71+
// during phase two; clear it so only the featureGates stanza under test drives the outcome.
72+
if v, ok := os.LookupEnv(enableExperimentalFlowControlLayer); ok {
73+
require.NoError(t, os.Unsetenv(enableExperimentalFlowControlLayer))
74+
t.Cleanup(func() { _ = os.Setenv(enableExperimentalFlowControlLayer, v) })
75+
}
76+
77+
boolPtr := func(b bool) *bool { return &b }
78+
testCases := []struct {
79+
name string
80+
configText string
81+
// wantEnabled nil means "expect whatever default the runner registered for the gate",
82+
// read programmatically from the feature gates parsed out of the stanza-less config.
83+
wantEnabled *bool
84+
}{
85+
{
86+
name: "no featureGates stanza follows the registered default",
87+
configText: `apiVersion: llm-d.ai/v1alpha1
88+
kind: EndpointPickerConfig
89+
`,
90+
wantEnabled: nil,
91+
},
92+
{
93+
name: "flowControl gate enabled wires the flow control admission controller",
94+
configText: `apiVersion: llm-d.ai/v1alpha1
95+
kind: EndpointPickerConfig
96+
featureGates:
97+
- flowControl
98+
`,
99+
wantEnabled: boolPtr(true),
100+
},
101+
{
102+
name: "flowControl=false restores the legacy admission controller",
103+
configText: `apiVersion: llm-d.ai/v1alpha1
104+
kind: EndpointPickerConfig
105+
featureGates:
106+
- flowControl=false
107+
`,
108+
wantEnabled: boolPtr(false),
109+
},
110+
}
111+
112+
for _, tc := range testCases {
113+
t.Run(tc.name, func(t *testing.T) {
114+
ctx, cancel := context.WithCancel(context.Background())
115+
defer cancel()
116+
117+
opts := runserver.NewOptions()
118+
opts.ConfigText = tc.configText
119+
opts.PoolName = "test-pool"
120+
121+
r := NewRunner()
122+
rawConfig, err := r.parseConfigurationPhaseOne(ctx, opts)
123+
require.NoError(t, err)
124+
125+
wantEnabled := r.featureGates[flowcontrol.FeatureGate] // Registered default.
126+
if tc.wantEnabled != nil {
127+
wantEnabled = *tc.wantEnabled
128+
require.Equal(t, wantEnabled, r.featureGates[flowcontrol.FeatureGate],
129+
"the loader should honor the explicit featureGates stanza")
130+
}
131+
132+
ds := datastore.NewDatastore(ctx, r.setupMetricsCollection(opts))
133+
eppConfig, err := r.parseConfigurationPhaseTwo(ctx, rawConfig, ds)
134+
require.NoError(t, err)
135+
136+
endpointCandidates := contracts.EndpointCandidates(
137+
requestcontrol.NewDatastoreEndpointCandidates(ds))
138+
_, admissionController, controlPlane :=
139+
r.initAdmissionControl(ctx, opts, eppConfig, endpointCandidates)
140+
141+
if wantEnabled {
142+
require.IsType(t, &requestcontrol.FlowControlAdmissionController{}, admissionController)
143+
require.NotNil(t, eppConfig.FlowControlConfig,
144+
"the loader should build a flow control config when the gate is on")
145+
require.NotNil(t, controlPlane,
146+
"the flow registry should be exposed as the priority band control plane")
147+
} else {
148+
require.IsType(t, &requestcontrol.LegacyAdmissionController{}, admissionController)
149+
require.Nil(t, eppConfig.FlowControlConfig,
150+
"the loader should not build a flow control config when the gate is off")
151+
require.Nil(t, controlPlane,
152+
"no priority band control plane should exist when the gate is off")
153+
}
154+
})
155+
}
156+
}

cmd/epp/runner/runner.go

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -85,6 +85,7 @@ import (
8585
"github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/flowcontrol/saturationdetector/utilization"
8686
"github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/flowcontrol/usagelimits"
8787
"github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/flowcontrol/usagelimits/priorityholdback"
88+
"github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/flowcontrol/usagelimits/softreflectiveceiling"
8889
"github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/requestcontrol/admitter/latencyslo"
8990
"github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/requestcontrol/admitter/probabilisticadmitter"
9091
reqdataprodprefix "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/requestcontrol/dataproducer/approximateprefix"
@@ -610,6 +611,7 @@ func (r *Runner) registerInTreePlugins() {
610611
fwkplugin.Register(slodeadline.SLODeadlineOrderingPolicyType, slodeadline.SLODeadlineOrderingPolicyFactory)
611612
fwkplugin.Register(usagelimits.StaticUsageLimitPolicyType, usagelimits.StaticPolicyFactory)
612613
fwkplugin.Register(priorityholdback.PolicyType, priorityholdback.PolicyFactory)
614+
fwkplugin.Register(softreflectiveceiling.PolicyType, softreflectiveceiling.Factory)
613615

614616
// Register Request level data producer plugins as defaults for their respective data keys.
615617
fwkplugin.RegisterAsDefaultProducer(reqdataprodprefix.ApproxPrefixCachePluginType, reqdataprodprefix.ApproxPrefixCacheFactory, attrprefix.PrefixCacheMatchInfoDataKey)

cmd/epp/runner/test_runner.go

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@ package runner
1919
import (
2020
"context"
2121
"encoding/json"
22+
"net"
2223

2324
"k8s.io/client-go/rest"
2425
ctrl "sigs.k8s.io/controller-runtime"
@@ -33,8 +34,9 @@ import (
3334
// NewTestRunnerSetup creates a setup runner dedicated for integration tests. When mockDataSource is
3435
// non-nil, its plugin type is registered as a factory that returns the provided instance, so the
3536
// YAML config can reference it by type name and the runner wires it into the endpoint factory
36-
// automatically.
37-
func NewTestRunnerSetup(ctx context.Context, cfg *rest.Config, opts *runserver.Options, mockDataSource fwkdl.DataSource) (*Runner, ctrl.Manager, datastore.Datastore, error) {
37+
// automatically. When grpcListener is non-nil the ext_proc server serves on it and
38+
// opts.GRPCPort is ignored.
39+
func NewTestRunnerSetup(ctx context.Context, cfg *rest.Config, opts *runserver.Options, mockDataSource fwkdl.DataSource, grpcListener net.Listener) (*Runner, ctrl.Manager, datastore.Datastore, error) {
3840
runner := NewRunner()
3941

4042
if mockDataSource != nil {
@@ -51,13 +53,17 @@ func NewTestRunnerSetup(ctx context.Context, cfg *rest.Config, opts *runserver.O
5153
managerOverrides := []func(*ctrl.Options){
5254
func(o *ctrl.Options) {
5355
o.Controller.SkipNameValidation = &skipNameValidation
56+
// The kernel assigns the port, so the bind cannot lose a race to
57+
// another listener the way a port number chosen in advance can.
58+
o.Metrics.BindAddress = "127.0.0.1:0"
5459
},
5560
}
5661

5762
manager, ds, err := runner.setup(ctx, cfg, opts, managerOverrides)
5863
if err != nil {
5964
return runner, manager, ds, err
6065
}
66+
runner.serverRunner.GrpcListener = grpcListener
6167

6268
// Production runs the ext_proc and health servers on a context that outlives
6369
// the manager (see Runner.runWithGracefulShutdown). Integration tests drive

0 commit comments

Comments
 (0)