minke.yu/sglang
1
0
Fork 0
Code Issues Pull Requests Actions 12 Packages Projects Releases Wiki Activity

[sgl-router] refactor - config and organize CLI options (#39867)

Browse Source
This commit is contained in:
Kan Wu
2026-09-20 19:40:04 +08:00
committed by GitHub
parent efa7be2091
commit c610c40399
4 changed files with 1021 additions and 1476 deletions
Download Patch File Download Diff File Expand all files Collapse all files
experimental/sgl-router/src/config/cli.rs
+768 -748
View File
File diff suppressed because it is too large Load Diff
experimental/sgl-router/src/config/mod.rs
+98 -176
View File
@@ -5,51 +5,29 @@ pub use cli::Cli;
pub use sampling::*; pub use sampling::*;
pub use types::*; pub use types::*;
use anyhow::{anyhow, Result}; use anyhow::{anyhow, ensure, Result};
/// The k8s default `terminationGracePeriodSeconds`, assumed when the operator /// Default pod termination grace period when none is declared.
/// has not declared the pod's real one. A `shutdown_drain_secs` at or above the
/// grace period leaves no time for the in-flight drain, so the pod is SIGKILLed
/// before it finishes — the opposite of what the drain is for.
pub const K8S_DEFAULT_GRACE_SECS: u64 = 30; pub const K8S_DEFAULT_GRACE_SECS: u64 = 30;
/// Ceiling on `shutdown_drain_secs`, enforced by [`Config::validate`]. Sized /// Maximum shutdown pause; deployments must also allow time to drain in-flight requests.
/// for the workload rather than for the k8s default grace period: a single /// This is the hard typo gate (an extra digit, seconds confused with milliseconds);
/// streaming completion can hold the router for many minutes, so a deployment /// whether a legal drain fits a particular grace period is [`shutdown_drain_advisory`]'s
/// that does not want terminations cutting one off runs a /// job, because the operator can raise the budget.
/// `terminationGracePeriodSeconds` in the tens of minutes and a drain to match.
/// Deciding whether a particular drain fits a particular grace period is
/// [`shutdown_drain_advisory`]'s job — advice, because the operator can raise
/// the budget. This constant is the separate, harder gate: it rejects a value
/// that is not a drain at all, an extra digit or seconds confused with
/// milliseconds, which no grace period could ever service.
pub const MAX_SHUTDOWN_DRAIN_SECS: u64 = 1800; pub const MAX_SHUTDOWN_DRAIN_SECS: u64 = 1800;
/// A `shutdown_drain_secs` that leaves no room under the grace period for the /// A shutdown pause that exhausts the declared or assumed pod grace period.
/// in-flight drain that follows the pause. Carries the compared values as
/// fields so the caller logs a static message with structured data rather than
/// interpolating the numbers into the message text.
#[derive(Debug, Clone, Copy, PartialEq, Eq)] #[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ShutdownDrainAdvisory { pub struct ShutdownDrainAdvisory {
pub shutdown_drain_secs: u64, pub shutdown_drain_secs: u64,
/// The budget the drain was compared against. /// The budget the drain was compared against.
pub termination_grace_secs: u64, pub termination_grace_secs: u64,
/// Whether that budget came from the operator or from /// Whether the operator declared the grace period.
/// [`K8S_DEFAULT_GRACE_SECS`]. An assumed budget makes the advisory a
/// guess; a declared one makes it a fact.
pub grace_declared: bool, pub grace_declared: bool,
} }
/// Advisory (not a hard error: the drain may well be right and the grace period /// Warn when the pause leaves no time for in-flight draining.
/// raised to match) for a drain that leaves no room for the in-flight drain. /// An undeclared grace period uses the Kubernetes default.
/// The bound is `>=`, not `>`: a drain of exactly the grace period already
/// consumes all of it.
///
/// `termination_grace_secs` is the pod's real `terminationGracePeriodSeconds`
/// when the operator declared it. The router cannot read its own pod spec, so
/// `None` falls back to the k8s default — which is why declaring the real value
/// is the way to silence this on a deployment that raised the grace period
/// deliberately, rather than lowering a drain that was correct.
pub fn shutdown_drain_advisory( pub fn shutdown_drain_advisory(
shutdown_drain_secs: u64, shutdown_drain_secs: u64,
termination_grace_secs: Option<u64>, termination_grace_secs: Option<u64>,
@@ -63,51 +41,35 @@ pub fn shutdown_drain_advisory(
} }
impl Config { impl Config {
/// Check invariants the type system and `clap` don't already enforce. /// Validate invariants not enforced by the CLI parser.
/// Called by [`cli::Cli::into_config`] after assembling the `Config`
/// from flags. Unknown policy names and `--cb-threshold 0` are
/// rejected at parse time (`ValueEnum` / `NonZeroU32`); only the
/// remaining value-level invariants are checked here.
pub(crate) fn validate(&self) -> Result<()> { pub(crate) fn validate(&self) -> Result<()> {
if self.model.id.is_empty() { ensure!(!self.model.id.is_empty(), "model id must be non-empty");
return Err(anyhow!("model id must be non-empty"));
}
if let Some(bucket_config) = self.model.bucket_config.as_ref() { if let Some(bucket_config) = self.model.bucket_config.as_ref() {
validate_bucket_config(bucket_config)?; validate_bucket_config(bucket_config)?;
} }
self.model.sampling_overrides.validate()?; self.model.sampling_overrides.validate()?;
if self.server.shutdown_drain_secs > MAX_SHUTDOWN_DRAIN_SECS { ensure!(
return Err(anyhow!( self.server.shutdown_drain_secs <= MAX_SHUTDOWN_DRAIN_SECS,
"shutdown_drain_secs must be at most {MAX_SHUTDOWN_DRAIN_SECS} (got {}); \ "shutdown_drain_secs must be at most {MAX_SHUTDOWN_DRAIN_SECS} (got {}); \
past the ceiling a value is a typo rather than a drain, and the pod \ past the ceiling a value is a typo rather than a drain. A long but \
would be SIGKILLed long before the pause elapsed. A long but deliberate \ deliberate drain is fine — declare --termination-grace-secs so startup \
drain is fine — declare --termination-grace-secs so startup can check \ can check it against the pod's real budget",
it against the pod's real budget", self.server.shutdown_drain_secs,
self.server.shutdown_drain_secs, );
));
}
match &self.discovery { match &self.discovery {
DiscoveryBackend::StaticUrls(s) => { DiscoveryBackend::StaticUrls(s) => {
if s.urls.is_empty() { ensure!(
return Err(anyhow!( !s.urls.is_empty(),
"discovery.static_urls.urls must be a non-empty list" "discovery.static_urls.urls must be a non-empty list"
)); );
} // Normalize URLs before deduplication so trailing slashes cannot register a worker twice.
// Validate every entry up front so typos surface at
// startup with a precise diagnostic instead of as
// per-worker introspect failures or as two registry
// entries pointing at the same SGLang (trailing-slash
// near-duplicates). Dedupe runs against a normalized
// form (trimmed + trailing `/` stripped) so
// `"http://x:30000"` and `"http://x:30000/"` collide.
let mut seen = std::collections::HashSet::new(); let mut seen = std::collections::HashSet::new();
for raw in &s.urls { for raw in &s.urls {
let trimmed = raw.trim(); let trimmed = raw.trim();
if trimmed.is_empty() { ensure!(
return Err(anyhow!( !trimmed.is_empty(),
"discovery.static_urls.urls contains an empty or whitespace-only entry" "discovery.static_urls.urls contains an empty or whitespace-only entry"
)); );
}
let parsed = url::Url::parse(trimmed).map_err(|e| { let parsed = url::Url::parse(trimmed).map_err(|e| {
anyhow!("discovery.static_urls.urls entry {raw:?} is not a valid URL: {e}") anyhow!("discovery.static_urls.urls entry {raw:?} is not a valid URL: {e}")
})?; })?;
@@ -120,17 +82,13 @@ impl Config {
} }
} }
let normalized = parsed.as_str().trim_end_matches('/').to_string(); let normalized = parsed.as_str().trim_end_matches('/').to_string();
if !seen.insert(normalized.clone()) { ensure!(
return Err(anyhow!( seen.insert(normalized.clone()),
"discovery.static_urls.urls contains duplicate entry {raw:?} (normalized: {normalized:?})" "discovery.static_urls.urls contains duplicate entry {raw:?} (normalized: {normalized:?})"
)); );
}
} }
} }
// K8s selector validity is resolved at construction time // Kubernetes selector combinations are validated by `resolve_mode` during construction.
// (`resolve_mode` in `Cli::build_discovery`), so the stored
// `K8sDiscoveryMode` is already valid here. Any namespace
// (including empty, for a cluster-wide watch) is accepted.
DiscoveryBackend::K8s(_) => {} DiscoveryBackend::K8s(_) => {}
} }
Ok(()) Ok(())
@@ -138,50 +96,44 @@ impl Config {
} }
fn validate_bucket_config(bucket_config: &BucketConfig) -> Result<()> { fn validate_bucket_config(bucket_config: &BucketConfig) -> Result<()> {
if bucket_config.buckets.is_empty() { ensure!(
return Err(anyhow!( !bucket_config.buckets.is_empty(),
"bucket_config.buckets must be non-empty when configured" "bucket_config.buckets must be non-empty when configured"
)); );
}
let mut ids = std::collections::HashSet::new(); let mut ids = std::collections::HashSet::new();
let mut ranks = std::collections::HashSet::new(); let mut ranks = std::collections::HashSet::new();
let mut stage_workers = std::collections::HashSet::new(); let mut stage_workers = std::collections::HashSet::new();
let mut has_prefill_bucket = false; let mut has_prefill_bucket = false;
for bucket in &bucket_config.buckets { for bucket in &bucket_config.buckets {
has_prefill_bucket |= bucket.stage == BucketStage::Prefill; has_prefill_bucket |= bucket.stage == BucketStage::Prefill;
if bucket.id.is_empty() || !ids.insert(bucket.id.as_str()) { ensure!(
return Err(anyhow!( !bucket.id.is_empty() && ids.insert(bucket.id.as_str()),
"bucket_config bucket id must be non-empty and unique: {:?}", "bucket_config bucket id must be non-empty and unique: {:?}",
bucket.id bucket.id
)); );
} ensure!(
if !ranks.insert((bucket.stage, bucket.rank)) { ranks.insert((bucket.stage, bucket.rank)),
return Err(anyhow!( "bucket_config rank must be unique within each stage: {}",
"bucket_config rank must be unique within each stage: {}", bucket.rank
bucket.rank );
)); ensure!(
} !bucket.worker_ids.is_empty(),
if bucket.worker_ids.is_empty() { "bucket_config bucket {:?} has no worker_ids",
return Err(anyhow!( bucket.id
"bucket_config bucket {:?} has no worker_ids", );
bucket.id
));
}
let mut worker_ids = std::collections::HashSet::new(); let mut worker_ids = std::collections::HashSet::new();
for worker_id in &bucket.worker_ids { for worker_id in &bucket.worker_ids {
if worker_id.is_empty() || !worker_ids.insert(worker_id.as_str()) { ensure!(
return Err(anyhow!( !worker_id.is_empty() && worker_ids.insert(worker_id.as_str()),
"bucket_config bucket {:?} has an empty or duplicate worker id", "bucket_config bucket {:?} has an empty or duplicate worker id",
bucket.id bucket.id
)); );
} ensure!(
if !stage_workers.insert((bucket.stage, worker_id.as_str())) { stage_workers.insert((bucket.stage, worker_id.as_str())),
return Err(anyhow!( "bucket_config worker {:?} belongs to more than one {:?} bucket",
"bucket_config worker {:?} belongs to more than one {:?} bucket", worker_id,
worker_id, bucket.stage
bucket.stage );
));
}
} }
validate_range( validate_range(
bucket.min_extend_tokens, bucket.min_extend_tokens,
@@ -195,33 +147,28 @@ fn validate_bucket_config(bucket_config: &BucketConfig) -> Result<()> {
&bucket.id, &bucket.id,
"sequence", "sequence",
)?; )?;
if bucket.max_context_tokens == Some(0) { ensure!(
return Err(anyhow!( bucket.max_context_tokens != Some(0),
"bucket_config bucket {:?} max_context_tokens must be > 0", "bucket_config bucket {:?} max_context_tokens must be > 0",
bucket.id bucket.id
)); );
} ensure!(
if bucket.ttft_p95_at_capacity_ms == Some(0) { bucket.ttft_p95_at_capacity_ms != Some(0),
return Err(anyhow!( "bucket_config bucket {:?} TTFT p95 must be > 0",
"bucket_config bucket {:?} TTFT p95 must be > 0", bucket.id
bucket.id );
)); ensure!(
} bucket
if bucket .tps_p05_at_capacity
.tps_p05_at_capacity .is_none_or(|value| value.is_finite() && value > 0.0),
.is_some_and(|value| !value.is_finite() || value <= 0.0) "bucket_config bucket {:?} TPS p05 must be finite and > 0",
{ bucket.id
return Err(anyhow!( );
"bucket_config bucket {:?} TPS p05 must be finite and > 0", ensure!(
bucket.id bucket.max_pending_prefill_tokens != Some(0),
)); "bucket_config bucket {:?} max_pending_prefill_tokens must be > 0",
} bucket.id
if bucket.max_pending_prefill_tokens == Some(0) { );
return Err(anyhow!(
"bucket_config bucket {:?} max_pending_prefill_tokens must be > 0",
bucket.id
));
}
match bucket.stage { match bucket.stage {
BucketStage::Prefill BucketStage::Prefill
if bucket.min_sequence_tokens.is_some() if bucket.min_sequence_tokens.is_some()
@@ -247,20 +194,19 @@ fn validate_bucket_config(bucket_config: &BucketConfig) -> Result<()> {
_ => {} _ => {}
} }
} }
if !has_prefill_bucket { ensure!(
return Err(anyhow!( has_prefill_bucket,
"bucket_config must contain at least one Prefill bucket; enabling Bucket routing otherwise leaves every request without a Prefill domain" "bucket_config must contain at least one Prefill bucket; enabling Bucket \
)); routing otherwise leaves every request without a Prefill domain"
} );
Ok(()) Ok(())
} }
fn validate_range(min: Option<u64>, max: Option<u64>, id: &str, name: &str) -> Result<()> { fn validate_range(min: Option<u64>, max: Option<u64>, id: &str, name: &str) -> Result<()> {
if min.zip(max).is_some_and(|(min, max)| min > max) { ensure!(
return Err(anyhow!( min.zip(max).is_none_or(|(min, max)| min <= max),
"bucket_config bucket {id:?} has invalid {name} range: min > max" "bucket_config bucket {id:?} has invalid {name} range: min > max"
)); );
}
Ok(()) Ok(())
} }
@@ -268,10 +214,6 @@ fn validate_range(min: Option<u64>, max: Option<u64>, id: &str, name: &str) -> R
mod tests { mod tests {
use super::*; use super::*;
/// Build a minimal valid-shape `Config` with the given static worker
/// URLs and model id, so the `validate()` branches can be exercised
/// directly. CLI parsing and the static-vs-k8s mapping are covered in
/// the `cli` module tests; the k8s selector grammar in `types`.
fn cfg(model_id: &str, urls: &[&str]) -> Config { fn cfg(model_id: &str, urls: &[&str]) -> Config {
Config { Config {
server: ServerConfig::default(), server: ServerConfig::default(),
@@ -549,19 +491,13 @@ mod tests {
#[test] #[test]
fn shutdown_drain_advisory_is_silent_below_the_k8s_default_grace() { fn shutdown_drain_advisory_is_silent_below_the_k8s_default_grace() {
// Anything strictly under the k8s default terminationGracePeriodSeconds
// (30 s) still leaves room for the in-flight drain, so it is safe
// without operator action.
assert!(shutdown_drain_advisory(29, None).is_none()); assert!(shutdown_drain_advisory(29, None).is_none());
assert!(shutdown_drain_advisory(0, None).is_none()); assert!(shutdown_drain_advisory(0, None).is_none());
} }
/// The default drain is exactly the assumed grace period, so out of the box /// Pinned because this is the one case an operator meets without choosing it:
/// it leaves nothing for the in-flight drain and says so on every startup. /// an edit to either constant that silenced the warning would change the default
/// Asserted rather than left implicit because it is the one case an /// deployment's behaviour, and should have to say so here.
/// operator meets without choosing it: a later edit to either constant that
/// silenced the warning would be changing the default deployment's
/// behaviour, and should have to say so here.
#[test] #[test]
fn the_default_drain_warns_until_the_grace_period_is_raised() { fn the_default_drain_warns_until_the_grace_period_is_raised() {
let advisory = shutdown_drain_advisory(default_shutdown_drain_secs(), None) let advisory = shutdown_drain_advisory(default_shutdown_drain_secs(), None)
@@ -585,11 +521,6 @@ mod tests {
#[test] #[test]
fn shutdown_drain_advisory_warns_once_the_drain_consumes_the_whole_grace() { fn shutdown_drain_advisory_warns_once_the_drain_consumes_the_whole_grace() {
// A drain of exactly the 30 s k8s default leaves zero seconds for the
// in-flight drain, so the pod is SIGKILLed mid-drain — the boundary
// itself must warn, not just values past it. The ceiling is in the list
// because it is startable: `validate` accepts it, so the advisory is
// the only thing left to say it does not fit the default grace period.
for drain in [K8S_DEFAULT_GRACE_SECS, 120, MAX_SHUTDOWN_DRAIN_SECS] { for drain in [K8S_DEFAULT_GRACE_SECS, 120, MAX_SHUTDOWN_DRAIN_SECS] {
let advisory = shutdown_drain_advisory(drain, None) let advisory = shutdown_drain_advisory(drain, None)
.unwrap_or_else(|| panic!("{drain}s must warn")); .unwrap_or_else(|| panic!("{drain}s must warn"));
@@ -602,9 +533,6 @@ mod tests {
} }
} }
/// The advisory's whole purpose is to be silenceable by declaring the real
/// budget: a 60 s drain under a 120 s grace period is a correct
/// configuration, and warning about it trains operators to ignore the line.
#[test] #[test]
fn shutdown_drain_advisory_respects_a_declared_grace_period() { fn shutdown_drain_advisory_respects_a_declared_grace_period() {
assert!( assert!(
@@ -624,18 +552,12 @@ mod tests {
shutdown_drain_advisory(10, Some(10)).is_some(), shutdown_drain_advisory(10, Some(10)).is_some(),
"a short declared grace period must still be compared against", "a short declared grace period must still be compared against",
); );
// The configuration the ceiling was raised for: a completion streaming
// for minutes wants a drain of minutes, under a grace period declared
// to match. That is correct, not merely tolerated, so it must be silent.
assert!( assert!(
shutdown_drain_advisory(MAX_SHUTDOWN_DRAIN_SECS, Some(3600)).is_none(), shutdown_drain_advisory(MAX_SHUTDOWN_DRAIN_SECS, Some(3600)).is_none(),
"a long drain under a grace period declared to cover it must not warn", "a long drain under a grace period declared to cover it must not warn",
); );
} }
/// `validate` is the hard gate the advisory deliberately is not: past the
/// ceiling the value can only be a typo, and starting on it would make
/// every later termination a SIGKILL.
#[test] #[test]
fn validate_rejects_a_shutdown_drain_past_the_ceiling() { fn validate_rejects_a_shutdown_drain_past_the_ceiling() {
let mut config = cfg("qwen3-0.6b", &["http://10.0.0.1:30000"]); let mut config = cfg("qwen3-0.6b", &["http://10.0.0.1:30000"]);
experimental/sgl-router/src/config/sampling.rs
+80 -217
View File
@@ -1,58 +1,30 @@
// SPDX-FileCopyrightText: Copyright (c) 2026 The SGLang Authors // SPDX-FileCopyrightText: Copyright (c) 2026 The SGLang Authors
// SPDX-License-Identifier: Apache-2.0 // SPDX-License-Identifier: Apache-2.0
//! Fleet-wide sampling contract (`--override-sampling-params` / //! Fleet sampling defaults and constraints. Custom JSON visitors preserve duplicate
//! `--sampling-param-conflict`): the parameters an operator fixes for every //! keys so validation can reject them and report the offending parameter.
//! request this router serves, and what a request that disagrees gets.
//! //!
//! WHY this is parsed by hand rather than with `serde(deny_unknown_fields)`: //! The flag is read once, at startup, on a router that crash-loops if it is wrong,
//! the flag is read once, at startup, on a router that crash-loops if it is //! so the message an operator reads out of `kubectl logs` is the whole debugging
//! wrong, so the message an operator reads out of `kubectl logs` is the whole //! session: every rejection names the offending key, the value it saw, and the
//! debugging session. Every rejection here names the offending key, the value //! domain it violated. (A `serde_json::Map` would keep only the last of a repeated
//! it saw, and the domain it violated. The same reasoning is why both the //! key and silently enforce a value the operator did not write.)
//! outer object and a band are decoded as ordered ENTRIES instead of a
//! `serde_json::Map`: a map keeps only the last of a repeated key, so
//! `{"temperature": 0, "temperature": 1}` would start cleanly and enforce a
//! value the operator did not write.
use anyhow::{anyhow, Result}; use anyhow::{anyhow, ensure, Result};
use std::collections::BTreeMap; use std::collections::BTreeMap;
/// Sampling parameters fixed fleet-wide, and what to do with a request that /// Fleet sampling defaults. Exact values fill absent fields; [`ConflictPolicy`]
/// disagrees. /// determines whether differing client values are rejected or forwarded.
///
/// A configured parameter is always injected into the forwarded body when the
/// request OMITS it, so the engine's own defaults cannot drift from what the
/// operator declared. What differs between the two [`ConflictPolicy`] modes is
/// only the request that DOES send the field: `Reject` makes the value an
/// immutability contract (400 before admission — never a silent rewrite),
/// while `Allow` lets the client value through untouched, degrading the
/// configured value to a fleet-wide default.
#[derive(Debug, Clone, Default, PartialEq)] #[derive(Debug, Clone, Default, PartialEq)]
pub struct SamplingOverrides { pub struct SamplingOverrides {
/// Configured parameters, keyed so enforcement and injection are one loop /// Parameters in deterministic injection order.
/// over whatever the operator set instead of a per-field ladder repeated
/// at each site. Iterating a `BTreeMap` keyed by the field enum is what
/// fixes the order values are injected in, so a forwarded body is
/// byte-identical across runs.
pub params: BTreeMap<SamplingField, ParamSpec>, pub params: BTreeMap<SamplingField, ParamSpec>,
/// Applies to every configured parameter: there is deliberately no /// Conflict behavior shared by all configured parameters.
/// per-parameter mode, so an operator reads one knob off one manifest.
pub conflict: ConflictPolicy, pub conflict: ConflictPolicy,
} }
impl SamplingOverrides { impl SamplingOverrides {
/// Re-check every invariant [`parse_sampling_overrides`] enforces, on an /// Validate parsed and programmatically constructed overrides alike.
/// already-built value.
///
/// WHY this is separate from the parser: the parser turns a raw JSON
/// string into this struct and is reachable only from the CLI, but the
/// struct itself is reachable from anywhere — a test fixture, a future
/// config file, an admin API. [`crate::config::Config::validate`] calls
/// this so no such path can hold a spec the flag would have refused to
/// start with (an out-of-domain exact value, an inverted band whose
/// `contains` rejects every value, or a band under `allow`, which names
/// nothing to inject and rejects nothing).
pub(crate) fn validate(&self) -> Result<()> { pub(crate) fn validate(&self) -> Result<()> {
for (&field, spec) in &self.params { for (&field, spec) in &self.params {
validate_spec(field, spec, self.conflict)?; validate_spec(field, spec, self.conflict)?;
@@ -65,10 +37,7 @@ impl SamplingOverrides {
/// (`--sampling-param-conflict`). /// (`--sampling-param-conflict`).
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, clap::ValueEnum)] #[derive(Debug, Clone, Copy, PartialEq, Eq, Default, clap::ValueEnum)]
pub enum ConflictPolicy { pub enum ConflictPolicy {
/// 400 before admission, quoting the configured value. The default: the /// Reject differing client values with 400 before admission.
/// point of declaring a fleet-wide sampling contract is usually that it
/// holds, and silently serving something other than what the client asked
/// for is the one behavior no client can detect.
#[default] #[default]
Reject, Reject,
/// Forward the client's value to the engine untouched. The configured /// Forward the client's value to the engine untouched. The configured
@@ -80,30 +49,16 @@ pub enum ConflictPolicy {
/// accepted ones. /// accepted ones.
#[derive(Debug, Clone, PartialEq)] #[derive(Debug, Clone, PartialEq)]
pub enum ParamSpec { pub enum ParamSpec {
/// A single value: injected when the request omits the field, and under /// Injected when absent; under [`ConflictPolicy::Reject`], the only accepted value.
/// [`ConflictPolicy::Reject`] the only value a request may send. /// JSON numbers preserve integer wire types.
///
/// Held as the parsed JSON number rather than an `f64` so injection
/// re-emits the operator's literal — `"n": 1` stays `1` and does not
/// become `1.0` on the wire for the integer-typed fields.
Exact(serde_json::Number), Exact(serde_json::Number),
/// An inclusive `[lo, hi]` band of accepted values, for a contract that /// Inclusive bounds for supplied values; never injects a default.
/// fixes most sampling knobs but leaves one tunable inside a range. A band /// Requires [`ConflictPolicy::Reject`]. Omitted fields use the engine default,
/// names no single value, so it never injects; it only rejects /// which may lie outside the band.
/// out-of-band values, which is why a band under [`ConflictPolicy::Allow`]
/// is a startup error rather than a no-op.
///
/// A band therefore constrains only the requests that NAME the parameter.
/// A request that omits it gets the model's own default (the engine reads
/// `generation_config`), which the router cannot see and which may itself
/// lie outside the band. An operator who needs the omitting majority
/// pinned too wants an exact value, not a band.
Range { lo: f64, hi: f64 }, Range { lo: f64, hi: f64 },
} }
/// A sampling parameter that can be fixed fleet-wide. The enum is what makes a /// Supported fleet sampling parameters.
/// typo in the `--override-sampling-params` JSON a startup error instead of a
/// key that silently never matches a request field.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub enum SamplingField { pub enum SamplingField {
Temperature, Temperature,
@@ -129,15 +84,7 @@ impl SamplingField {
Self::N, Self::N,
]; ];
/// This field's slot in [`Self::ALL`], and in the request probe's /// Slot in [`Self::ALL`] and in the request probe array.
/// fixed-size array of probed values.
///
/// Declaration order IS the slot order, so this cannot assign a wrong
/// one. What still needs checking is that [`Self::ALL`] agrees — see the
/// assertion below; `from_wire_name` and `supported_fields` both iterate
/// `ALL`, so a field missing from it is rejected at startup as an unknown
/// key, the failure that is invisible to any test that also iterates
/// `ALL`.
pub const fn index(self) -> usize { pub const fn index(self) -> usize {
self as usize self as usize
} }
@@ -190,12 +137,11 @@ pub(crate) fn parse_sampling_overrides(
'{{\"temperature\": 1, \"top_p\": 0.95}}': {e}" '{{\"temperature\": 1, \"top_p\": 0.95}}': {e}"
) )
})?; })?;
if entries.is_empty() { ensure!(
return Err(anyhow!( !entries.is_empty(),
"--override-sampling-params is empty: pass at least one of {}, or omit the flag", "--override-sampling-params is empty: pass at least one of {}, or omit the flag",
supported_fields() supported_fields()
)); );
}
let mut params = BTreeMap::new(); let mut params = BTreeMap::new();
for (key, value) in entries { for (key, value) in entries {
let field = SamplingField::from_wire_name(&key).ok_or_else(|| { let field = SamplingField::from_wire_name(&key).ok_or_else(|| {
@@ -216,26 +162,20 @@ pub(crate) fn parse_sampling_overrides(
)) ))
} }
}; };
if params.insert(field, spec).is_some() { ensure!(
return Err(anyhow!( params.insert(field, spec).is_none(),
"--override-sampling-params: {} is set more than once", "--override-sampling-params: {} is set more than once",
field.wire_name() field.wire_name()
)); );
}
} }
let overrides = SamplingOverrides { params, conflict }; let overrides = SamplingOverrides { params, conflict };
// Re-checks the domains `checked_value` already covered above. That first // Validate complete specs too, including bands and programmatically built overrides.
// pass is not redundant: it is what quotes the operator's own literal // The earlier exact-value check protects the integer cast in `canonical_number`.
// (`1e2`, not `100`) and what guards `canonical_number`'s saturating
// `as i64` cast before it runs. This pass is what a hand-built
// `SamplingOverrides` gets, and is the only check a band's bounds see.
overrides.validate()?; overrides.validate()?;
Ok(overrides) Ok(overrides)
} }
/// Check one already-built spec. Shared by [`parse_sampling_overrides`] and /// Validate a spec independently of how it was constructed.
/// [`SamplingOverrides::validate`] so a hand-built `SamplingOverrides` is held
/// to exactly the domain the flag is.
fn validate_spec(field: SamplingField, spec: &ParamSpec, conflict: ConflictPolicy) -> Result<()> { fn validate_spec(field: SamplingField, spec: &ParamSpec, conflict: ConflictPolicy) -> Result<()> {
let key = field.wire_name(); let key = field.wire_name();
match spec { match spec {
@@ -245,31 +185,24 @@ fn validate_spec(field: SamplingField, spec: &ParamSpec, conflict: ConflictPolic
&ParamSpec::Range { lo, hi } => { &ParamSpec::Range { lo, hi } => {
check_domain(field, lo, &lo.to_string())?; check_domain(field, lo, &lo.to_string())?;
check_domain(field, hi, &hi.to_string())?; check_domain(field, hi, &hi.to_string())?;
if lo > hi { ensure!(
return Err(anyhow!( lo <= hi,
"--override-sampling-params: {key} band needs min <= max, got min {lo} > max {hi}" "--override-sampling-params: {key} band needs min <= max, got min {lo} > max {hi}"
)); );
} // `top_k = -1` disables filtering; it cannot bound a band that would admit zero.
// Bounds are checked one at a time, which is only sufficient for a ensure!(
// contiguous domain. `top_k`'s is not ({-1} U [1, inf)): `{"min": -1, field != SamplingField::TopK || lo >= 1.0,
// "max": 100}` has two individually legal bounds and would admit "--override-sampling-params: top_k band bounds must both be >= 1 \
// `top_k: 0`, which is rejected as an exact value. -1 is a sentinel,
// not a range endpoint.
if field == SamplingField::TopK && lo < 1.0 {
return Err(anyhow!(
"--override-sampling-params: top_k band bounds must both be >= 1 \
(-1 disables top_k entirely and cannot bound a range)" (-1 disables top_k entirely and cannot bound a range)"
)); );
}
// A band only ever rejects, so under `allow` it would be dead config // A band only ever rejects, so under `allow` it would be dead config
// that silently accepts everything. // that silently accepts everything.
if conflict == ConflictPolicy::Allow { ensure!(
return Err(anyhow!( conflict == ConflictPolicy::Reject,
"--override-sampling-params: the {key} band requires \ "--override-sampling-params: the {key} band requires \
--sampling-param-conflict reject — under `allow` nothing is rejected \ --sampling-param-conflict reject — under `allow` nothing is rejected \
and a band names no value to inject" and a band names no value to inject"
)); );
}
} }
} }
Ok(()) Ok(())
@@ -294,11 +227,10 @@ fn parse_band(
)) ))
} }
}; };
if slot.is_some() { ensure!(
return Err(anyhow!( slot.is_none(),
"--override-sampling-params: {key} band sets \"{bound}\" more than once" "--override-sampling-params: {key} band sets \"{bound}\" more than once"
)); );
}
let serde_json::Value::Number(n) = v else { let serde_json::Value::Number(n) = v else {
return Err(anyhow!( return Err(anyhow!(
"--override-sampling-params: {key} band needs numeric bounds, got {bound}: {v}" "--override-sampling-params: {key} band needs numeric bounds, got {bound}: {v}"
@@ -312,96 +244,61 @@ fn parse_band(
{{\"min\": LO, \"max\": HI}} with numeric bounds" {{\"min\": LO, \"max\": HI}} with numeric bounds"
)); ));
}; };
// `lo <= hi`, `top_k`'s discontiguous domain and the band-under-`allow` // Finished-spec constraints are checked by `validate_spec` for all construction paths.
// rule are all properties of the finished spec, so they live in
// `validate_spec` and hold for a hand-built `SamplingOverrides` too.
Ok(ParamSpec::Range { lo, hi }) Ok(ParamSpec::Range { lo, hi })
} }
/// Check one configured value against its parameter's domain, at startup /// Validate before normalization so diagnostics retain the configured number.
/// instead of per request. Written as positive containment so a NaN bound /// Domains follow the OpenAI contract plus engine-specific parameters — deliberately
/// fails too. /// NARROWER than what the engine accepts: these values are injected into request
/// /// bodies, and a fleet contract outside the range every OpenAI client library
/// These are the OpenAI API's domains, which are NARROWER than what the engine /// validates against is far more likely a typo than an intent.
/// itself accepts (`SamplingParams.verify` requires only that `temperature` be
/// non-negative and finite, so it would take `temperature: 5`). Narrower is
/// deliberate: the values here are injected into request bodies, and a fleet
/// contract outside the range every OpenAI client library validates against is
/// far more likely a typo than an intent. The one exception is `top_k`, where
/// `-1` is the engine's own "disable / whole vocabulary" spelling and its
/// default — a legitimate thing to fix fleet-wide. Note `top_k: 1` is greedy
/// decoding, NOT "disabled".
fn checked_value(field: SamplingField, n: &serde_json::Number) -> Result<f64> { fn checked_value(field: SamplingField, n: &serde_json::Number) -> Result<f64> {
let name = field.wire_name(); let name = field.wire_name();
// `as_f64` is infallible for a JSON number unless serde_json's // Handle conversion failure even if serde_json arbitrary precision is enabled later.
// `arbitrary_precision` is on (it is not); kept total rather than
// `expect`-ing, so enabling that feature can't turn config into a panic.
let v = n.as_f64().ok_or_else(|| { let v = n.as_f64().ok_or_else(|| {
anyhow!("--override-sampling-params: {name} ({n}) is not a finite number") anyhow!("--override-sampling-params: {name} ({n}) is not a finite number")
})?; })?;
// The operator's own literal is what the message quotes, not the parsed
// f64: `1e2` should read back as `1e2`.
check_domain(field, v, &n.to_string())?; check_domain(field, v, &n.to_string())?;
Ok(v) Ok(v)
} }
/// The domain half of [`checked_value`], over an f64 that may not have come /// Validate a numeric domain; `shown` is the value quoted in diagnostics.
/// from a literal (a band's bounds are stored as f64). `shown` is what the
/// error quotes back to the operator.
fn check_domain(field: SamplingField, v: f64, shown: &str) -> Result<()> { fn check_domain(field: SamplingField, v: f64, shown: &str) -> Result<()> {
let name = field.wire_name(); let name = field.wire_name();
let (ok, domain) = match field { let (ok, domain) = match field {
SamplingField::Temperature => ((0.0..=2.0).contains(&v), "in [0, 2]"), SamplingField::Temperature => ((0.0..=2.0).contains(&v), "in [0, 2]"),
SamplingField::TopP => (v > 0.0 && v <= 1.0, "in (0, 1]"), SamplingField::TopP => (v > 0.0 && v <= 1.0, "in (0, 1]"),
SamplingField::TopK => (v >= 1.0 || v == -1.0, ">= 1, or -1 to disable"), SamplingField::TopK => (v >= 1.0 || v == -1.0, ">= 1, or -1 to disable"),
// Not an OpenAI parameter: `min_p` is the engine's own nucleus floor, // Engine-specific: zero disables `min_p`.
// and 0 is its default (disabled), so the whole [0, 1] range is
// legitimate to fix fleet-wide.
SamplingField::MinP => ((0.0..=1.0).contains(&v), "in [0, 1]"), SamplingField::MinP => ((0.0..=1.0).contains(&v), "in [0, 1]"),
// Also engine-only. 1.0 is "no penalty"; the engine requires > 0, and // Engine-specific: one disables the penalty; cap fleet defaults at two.
// values above ~2 degrade output badly enough that a fleet-wide pin
// there is far more likely a typo than an intent.
SamplingField::RepetitionPenalty => (v > 0.0 && v <= 2.0, "in (0, 2]"), SamplingField::RepetitionPenalty => (v > 0.0 && v <= 2.0, "in (0, 2]"),
SamplingField::FrequencyPenalty | SamplingField::PresencePenalty => { SamplingField::FrequencyPenalty | SamplingField::PresencePenalty => {
((-2.0..=2.0).contains(&v), "in [-2, 2]") ((-2.0..=2.0).contains(&v), "in [-2, 2]")
} }
// OpenAI caps `n` at 128. Unbounded here, a typo'd digit would be // Bound sequence fan-out when `n` is injected into requests.
// injected into every request that omits `n` and fan each one out to
// that many sequences at the engine — the exact per-request failure
// this startup check exists to convert into a launch failure.
SamplingField::N => ((1.0..=128.0).contains(&v), "in [1, 128]"), SamplingField::N => ((1.0..=128.0).contains(&v), "in [1, 128]"),
}; };
if !ok { ensure!(
return Err(anyhow!( ok,
"--override-sampling-params: {name} ({shown}) must be {domain}" "--override-sampling-params: {name} ({shown}) must be {domain}"
)); );
}
if field.is_integral() { if field.is_integral() {
if v.fract() != 0.0 { ensure!(
return Err(anyhow!( v.fract() == 0.0,
"--override-sampling-params: {name} ({shown}) must be a whole number" "--override-sampling-params: {name} ({shown}) must be a whole number"
)); );
} // Float-to-i64 casts saturate. Use [-2^63, 2^63): `i64::MAX as f64` rounds up.
// `canonical_number` casts to `i64`, and a Rust float-to-int cast ensure!(
// SATURATES rather than failing, so a literal past the i64 range would (i64::MIN as f64..i64::MAX as f64).contains(&v),
// silently become `i64::MAX` in every forwarded body. The exactly "--override-sampling-params: {name} ({shown}) is too large to forward"
// convertible f64s are [-2^63, 2^63), which is this half-open range );
// written as positive containment — `i64::MAX as f64` rounds UP to
// 2^63, so an inclusive `<=` against it would admit 2^63 itself and
// saturate exactly as described.
if !(i64::MIN as f64..i64::MAX as f64).contains(&v) {
return Err(anyhow!(
"--override-sampling-params: {name} ({shown}) is too large to forward"
));
}
} }
Ok(()) Ok(())
} }
/// Normalize an integer-typed parameter's literal so injection writes `1` /// Emit integer-typed parameters as integers; preserve other JSON numbers.
/// rather than `1.0` for a config that spelled it `1.0` — the engine types
/// these fields as `int`, and the forwarded body should look like what a
/// client would have sent. Non-integral fields keep the operator's literal.
fn canonical_number( fn canonical_number(
field: SamplingField, field: SamplingField,
value: f64, value: f64,
@@ -423,8 +320,7 @@ fn supported_fields() -> String {
.join(", ") .join(", ")
} }
/// A JSON object decoded to its entries IN ORDER, keeping a repeated key /// Ordered JSON entries preserve duplicate keys for validation.
/// instead of collapsing it. See the module WHY note.
struct ObjectEntries(Vec<(String, ParamValue)>); struct ObjectEntries(Vec<(String, ParamValue)>);
impl<'de> serde::Deserialize<'de> for ObjectEntries { impl<'de> serde::Deserialize<'de> for ObjectEntries {
@@ -454,10 +350,7 @@ impl<'de> serde::Deserialize<'de> for ObjectEntries {
} }
} }
/// One parameter's raw value: a number, a band's entries, or anything else. /// Raw parameter value; `Other` retains invalid values for precise diagnostics.
/// `Other` keeps the offending value so the caller can name it, rather than
/// degrading a wrong-type message into a serde type error behind the outer
/// object's context.
enum ParamValue { enum ParamValue {
Number(serde_json::Number), Number(serde_json::Number),
Band(Vec<(String, serde_json::Value)>), Band(Vec<(String, serde_json::Value)>),
@@ -510,9 +403,7 @@ impl<'de> serde::Deserialize<'de> for ParamValue {
Ok(ParamValue::Other(v.into())) Ok(ParamValue::Other(v.into()))
} }
/// JSON `null`. There is deliberately no `visit_none`: this type /// `deserialize_any` routes JSON null to `visit_unit`.
/// is only ever reached through `deserialize_any`, which routes
/// null here and never to the `Option` hook.
fn visit_unit<E>(self) -> Result<ParamValue, E> { fn visit_unit<E>(self) -> Result<ParamValue, E> {
Ok(ParamValue::Other(serde_json::Value::Null)) Ok(ParamValue::Other(serde_json::Value::Null))
} }
@@ -654,18 +545,12 @@ mod tests {
assert_eq!(exact_of(&o, SamplingField::TopK), Some(1000.0)); assert_eq!(exact_of(&o, SamplingField::TopK), Some(1000.0));
} }
/// `top_k: -1` is the engine's own "disable / whole vocabulary" spelling
/// (and its default), so a fleet may legitimately fix `top_k` to it —
/// unlike every other parameter, whose domain is the OpenAI one.
#[test] #[test]
fn top_k_accepts_the_engines_disable_sentinel() { fn top_k_accepts_the_engines_disable_sentinel() {
let o = parse(r#"{"top_k": -1}"#).unwrap(); let o = parse(r#"{"top_k": -1}"#).unwrap();
assert_eq!(exact_of(&o, SamplingField::TopK), Some(-1.0)); assert_eq!(exact_of(&o, SamplingField::TopK), Some(-1.0));
} }
/// An integer-typed parameter spelled as a float is normalized, so the
/// forwarded body carries `1` and not `1.0` for a field the engine types
/// as `int`.
#[test] #[test]
fn integral_params_are_normalized_to_integers() { fn integral_params_are_normalized_to_integers() {
let o = parse(r#"{"n": 1.0, "top_k": 20.0}"#).unwrap(); let o = parse(r#"{"n": 1.0, "top_k": 20.0}"#).unwrap();
@@ -689,11 +574,7 @@ mod tests {
} }
assert_eq!(SamplingField::from_wire_name("max_tokens"), None); assert_eq!(SamplingField::from_wire_name("max_tokens"), None);
} }
/// `canonical_number` casts to `i64` and a Rust float-to-int cast /// Float-to-i64 casts saturate. Use [-2^63, 2^63): `i64::MAX as f64` rounds up.
/// SATURATES, so a literal past the i64 range must fail the launch rather
/// than be injected as `i64::MAX`. The boundary case is the trap: `i64::MAX
/// as f64` rounds UP to 2^63, so a `>` comparison against it admits 2^63
/// itself.
#[test] #[test]
fn integral_literals_beyond_i64_fail_the_launch() { fn integral_literals_beyond_i64_fail_the_launch() {
for raw in [ for raw in [
@@ -716,12 +597,6 @@ mod tests {
); );
} }
/// `min_p` and `repetition_penalty` are the only two parameters besides
/// `temperature`/`top_p`/`top_k` that the engine resolves from the model's
/// own `generation_config`, so they are exactly the ones a fleet-wide pin
/// exists to stop drifting when an image is swapped. Rejecting them as
/// unknown keys would crash-loop the router for the operator who needs the
/// flag most.
#[test] #[test]
fn governs_the_engine_defaulted_parameters() { fn governs_the_engine_defaulted_parameters() {
let o = parse(r#"{"min_p": 0.05, "repetition_penalty": 1.1}"#).unwrap(); let o = parse(r#"{"min_p": 0.05, "repetition_penalty": 1.1}"#).unwrap();
@@ -744,16 +619,8 @@ mod tests {
} }
} }
/// `ALL` is what `from_wire_name` and `supported_fields` iterate, so a
/// field missing from it is silently rejected at startup as an unknown
/// key — invisible to any test that also iterates `ALL`. Pin the length
/// and the slot mapping against the wire names instead.
#[test] #[test]
fn all_covers_every_field_exactly_once() { fn all_covers_every_field_exactly_once() {
// The slot mapping itself is asserted at compile time (see the
// `const _` block above `parse_sampling_overrides`); what only a test
// can catch is a field missing from `ALL` entirely, which is why the
// names below are written out rather than derived from it.
let names: std::collections::BTreeSet<_> = let names: std::collections::BTreeSet<_> =
SamplingField::ALL.iter().map(|f| f.wire_name()).collect(); SamplingField::ALL.iter().map(|f| f.wire_name()).collect();
assert_eq!(names.len(), SamplingField::ALL.len(), "duplicate wire name"); assert_eq!(names.len(), SamplingField::ALL.len(), "duplicate wire name");
@@ -771,10 +638,6 @@ mod tests {
} }
} }
/// The struct-level invariants must hold for a `SamplingOverrides` that
/// never went through the parser — a test fixture, a future config file or
/// admin API. An inverted band is the nastiest of these: `(lo..=hi)`
/// contains nothing, so it would 400 every request naming the parameter.
#[test] #[test]
fn validate_rejects_hand_built_specs_the_parser_would_refuse() { fn validate_rejects_hand_built_specs_the_parser_would_refuse() {
let bad = [ let bad = [
experimental/sgl-router/src/config/types.rs
+75 -335
View File
@@ -2,32 +2,22 @@ use crate::config::sampling::SamplingOverrides;
use serde::Deserialize; use serde::Deserialize;
use std::num::NonZeroU32; use std::num::NonZeroU32;
/// In-memory router configuration, built from CLI flags by /// Single-model configuration built and validated by [`crate::config::Cli::into_config`].
/// [`crate::config::cli::Cli::into_config`] and validated by
/// [`Config::validate`]. The router serves exactly one model.
#[derive(Debug, Clone)] #[derive(Debug, Clone)]
pub struct Config { pub struct Config {
pub server: ServerConfig, pub server: ServerConfig,
pub observability: ObservabilityConfig, pub observability: ObservabilityConfig,
pub model: ModelConfig, pub model: ModelConfig,
/// Selected discovery backend. Built from CLI flags by /// Discovery mode resolved from CLI options; static URLs are checked by [`Config::validate`].
/// [`crate::config::cli::Cli::into_config`]: the static-vs-k8s choice
/// and the k8s selector grammar are resolved there (the latter via
/// [`resolve_mode`]); static worker-URL validity is checked by
/// [`Config::validate`].
pub discovery: DiscoveryBackend, pub discovery: DiscoveryBackend,
pub proxy: ProxyConfig, pub proxy: ProxyConfig,
pub active_load: ActiveLoadConfig, pub active_load: ActiveLoadConfig,
} }
/// Outbound proxy tuning. Default mirrors SGLang's typical prefill / /// Outbound request timeout settings.
/// decode latency budget; e2e tests lower it so per-request failures
/// trip the circuit breaker within the test's wall-time.
#[derive(Debug, Clone, Copy)] #[derive(Debug, Clone, Copy)]
pub struct ProxyConfig { pub struct ProxyConfig {
/// Maximum time to wait for a single upstream HTTP request to /// Timeout for upstream response headers and body. Counts as a circuit-breaker failure.
/// return headers + body. Default 300 s. The circuit breaker
/// records a failure when this fires.
pub request_timeout_secs: u64, pub request_timeout_secs: u64,
} }
@@ -43,15 +33,10 @@ impl Default for ProxyConfig {
} }
} }
/// Active-load (per-request) tracking. Production default (10 min) /// Request-tracking timeout; defaults above the proxy timeout.
/// sits above `proxy.request_timeout_secs` so the proxy timeout is the
/// one users hit first for normal slow upstreams; tests lower it to
/// let the janitor fire within their wall-time budget.
#[derive(Debug, Clone, Copy)] #[derive(Debug, Clone, Copy)]
pub struct ActiveLoadConfig { pub struct ActiveLoadConfig {
/// How long a request entry can live in the registry before the /// Maximum request-entry lifetime before cancellation with 504 `stale_request_expired`.
/// janitor fires its `cancel_token` and the chat handler returns
/// 504 `stale_request_expired`. Default 600 s.
pub stale_request_timeout_secs: u64, pub stale_request_timeout_secs: u64,
} }
@@ -67,13 +52,7 @@ impl Default for ActiveLoadConfig {
} }
} }
/// Routing policy selector — the enum form lets `clap` reject unknown /// Routing strategies accepted by `--policy`.
/// values at parse time and removes the runtime string match in the
/// policy factory.
///
/// Accepted on the CLI (`--policy`) as `round_robin` / `random` /
/// `power_of_two` / `load_based` / `fused_score` / `score_policy` /
/// `session_aware` / `cache_aware` / `sticky`.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, clap::ValueEnum)] #[derive(Debug, Clone, Copy, PartialEq, Eq, Default, clap::ValueEnum)]
pub enum PolicyKind { pub enum PolicyKind {
#[default] #[default]
@@ -98,11 +77,7 @@ pub enum PolicyKind {
/// Selects cache-affine prefill candidates from the configured prefix provider. /// Selects cache-affine prefill candidates from the configured prefix provider.
#[value(name = "cache_aware")] #[value(name = "cache_aware")]
CacheAware, CacheAware,
/// Sticky-session routing: pins a routing key (read from a /// Pin a request-header routing key to a worker.
/// configurable request header) to a worker via an in-memory map, so
/// stateful sessions land on the same backend. Tuning — header name,
/// keyless-fallback policy, and TTL eviction — lives on
/// `ModelConfig::sticky`.
#[value(name = "sticky")] #[value(name = "sticky")]
Sticky, Sticky,
} }
@@ -248,35 +223,16 @@ impl std::fmt::Display for StickyFallbackKind {
pub struct ServerConfig { pub struct ServerConfig {
pub host: String, pub host: String,
pub port: u16, pub port: u16,
/// Seconds to keep serving after SIGTERM — with `/readyz` flipped to 503 — /// Pause after SIGTERM with `/readyz` returning 503 before stopping accepts.
/// before the HTTP server stops accepting. The default covers both /// Allows endpoint removal or readiness-probe failures to reach load balancers.
/// deregistration paths: endpoint removal reaching kube-proxy after the /// Leave time in the pod grace period for in-flight draining; 0 disables the pause.
/// pod's `deletionTimestamp` is stamped, and a probe-driven load balancer,
/// which cannot act until `failureThreshold * periodSeconds` of `/readyz`
/// failures have accumulated.
///
/// Note that it equals the k8s default `terminationGracePeriodSeconds`, so
/// a pod that has not raised its grace period is left with nothing for the
/// in-flight drain that follows the pause, and
/// [`shutdown_drain_advisory`](crate::config::shutdown_drain_advisory)
/// warns at every startup. That is the intended reading rather than a
/// misconfigured default: a router whose completions stream for minutes
/// cannot terminate cleanly inside 30 s at all, and the grace period is the
/// thing to raise. 0 disables the pause.
pub shutdown_drain_secs: u64, pub shutdown_drain_secs: u64,
/// The pod's actual `terminationGracePeriodSeconds`, when the operator /// Declared pod termination grace period; `None` uses the Kubernetes default for advisories.
/// declares it. The router cannot read its own pod spec, so without this
/// the startup advisory can only compare the drain against the k8s
/// default — and warns, wrongly, about a deployment that raised the grace
/// period on purpose. `None` means "assume the default".
pub termination_grace_secs: Option<u64>, pub termination_grace_secs: Option<u64>,
} }
impl ServerConfig { impl ServerConfig {
/// [`Self::shutdown_drain_secs`] as a `Duration`. Keeps the seconds-to- /// Shutdown pause as a duration.
/// `Duration` conversion in the library, where a test can pin it, rather
/// than in `main.rs` where a `from_secs`/`from_millis` slip would silently
/// shorten every drain by a factor of 1000.
pub fn shutdown_drain(&self) -> std::time::Duration { pub fn shutdown_drain(&self) -> std::time::Duration {
std::time::Duration::from_secs(self.shutdown_drain_secs) std::time::Duration::from_secs(self.shutdown_drain_secs)
} }
@@ -294,10 +250,7 @@ pub fn default_shutdown_drain_secs() -> u64 {
30 30
} }
/// Exists so test fixtures can spell out only the fields they care about /// Defaults for config construction; the CLI mapping remains exhaustive.
/// (`tests/` is a separate crate, so a `#[cfg(test)]` constructor cannot reach
/// the integration fixtures). Keep `Cli::into_config` exhaustive so adding a
/// field still forces a decision on the production path.
impl Default for ServerConfig { impl Default for ServerConfig {
fn default() -> Self { fn default() -> Self {
Self { Self {
@@ -312,10 +265,7 @@ impl Default for ServerConfig {
#[derive(Debug, Clone)] #[derive(Debug, Clone)]
pub struct ObservabilityConfig { pub struct ObservabilityConfig {
pub log_level: String, pub log_level: String,
/// Selects the tracing-subscriber output format. `clap` rejects /// Tracing output format.
/// unrecognized values at parse time (`--log-format jsonl` and
/// similar typos surface as an error instead of silently degrading
/// to text).
pub log_format: LogFormat, pub log_format: LogFormat,
} }
@@ -346,9 +296,8 @@ impl Default for ObservabilityConfig {
#[derive(Debug, Clone)] #[derive(Debug, Clone)]
pub struct ModelConfig { pub struct ModelConfig {
pub id: String, pub id: String,
/// Tokenizer source: a local `tokenizer.json` path or a HuggingFace repo /// Local tokenizer.json or HuggingFace repo id; defaults to `id`.
/// id (downloaded on demand). Defaults to `id` when `--tokenizer-path` /// Resolved by [`crate::tokenizer::adapter::load`].
/// is omitted. Resolved by [`crate::tokenizer::adapter::load`].
pub tokenizer_path: String, pub tokenizer_path: String,
/// Disable router-generated input IDs for this model; keep routing tokenization. /// Disable router-generated input IDs for this model; keep routing tokenization.
/// Use when workers have rendering defaults or template stops the router cannot see. /// Use when workers have rendering defaults or template stops the router cannot see.
@@ -361,25 +310,15 @@ pub struct ModelConfig {
pub circuit_breaker: Option<CircuitBreakerConfig>, pub circuit_breaker: Option<CircuitBreakerConfig>,
/// Cache-Aware prefix configuration. /// Cache-Aware prefix configuration.
pub cache_aware: Option<CacheAwareConfig>, pub cache_aware: Option<CacheAwareConfig>,
/// Tuning for the sticky-session policy. `Some` exactly when /// Present only for the sticky policy; the header supplies the request routing key.
/// `policy = "sticky"` (built by [`crate::config::cli::Cli::into_config`]).
/// The chat handler reads `sticky.header_name` to populate
/// [`crate::policies::SelectionContext::routing_key`].
pub sticky: Option<StickyConfig>, pub sticky: Option<StickyConfig>,
/// Session and cache-affinity tuning. /// Session and cache-affinity tuning.
pub affinity: Option<AffinityConfig>, pub affinity: Option<AffinityConfig>,
/// Terms the score-composition policy sums. `Some` exactly when /// Terms for `fused_score` or `score_policy`; defaults to [`DEFAULT_FUSE`].
/// `policy = "fused_score"` or `policy = "score_policy"` (built by
/// [`crate::config::cli::Cli::into_config`]), defaulting to
/// [`DEFAULT_FUSE`] when `--fuse` is omitted.
pub fused: Option<Vec<FusedTerm>>, pub fused: Option<Vec<FusedTerm>>,
/// Hard constraints applied before policy selection. /// Hard constraints applied before policy selection.
pub eligibility: Option<EligibilityConfig>, pub eligibility: Option<EligibilityConfig>,
/// Sampling parameters fixed fleet-wide for this model, and what happens /// Fleet sampling defaults and conflict behavior. See [`SamplingOverrides`].
/// to a request that sends a different value: a 400 before admission, or
/// the client value forwarded untouched. Either way the configured value
/// is injected when the request omits the field — see
/// [`SamplingOverrides`]. Empty (default) preserves today's behavior.
pub sampling_overrides: SamplingOverrides, pub sampling_overrides: SamplingOverrides,
} }
@@ -458,9 +397,7 @@ pub struct CacheAwareConfig {
pub kv_indexer_endpoint: Option<KvIndexerEndpointConfig>, pub kv_indexer_endpoint: Option<KvIndexerEndpointConfig>,
} }
/// Default routing-key header for the sticky policy. The `x-sgl-` prefix /// Default request header for sticky routing.
/// matches the router's other emitted/consumed metadata headers
/// (`x-sgl-decode-url`, `x-sgl-router-error-code`).
pub const DEFAULT_STICKY_HEADER: &str = "x-sgl-routing-key"; pub const DEFAULT_STICKY_HEADER: &str = "x-sgl-routing-key";
/// Default request header for session-aware routing. /// Default request header for session-aware routing.
@@ -520,66 +457,23 @@ pub struct AffinityConfig {
pub cache_candidate_ratio: f64, pub cache_candidate_ratio: f64,
pub cache_candidate_max_workers: usize, pub cache_candidate_max_workers: usize,
pub cache_switch_margin_tokens: u64, pub cache_switch_margin_tokens: u64,
/// Queue gate (`--worker-queue-limit`): a worker whose engine reports at /// Waiting-request limit for cache affinity; `None` disables.
/// least this many *waiting* requests cannot win a selection on cache /// Gates on the engine-published *waiting* count rather than total depth because
/// affinity — the request goes to another worker holding the same /// waiting is the question the request cares about — will it sit behind other work —
/// prefix, or failing that to the least-loaded worker that is not /// while depth proxies it badly (an engine can queue far below its running cap on
/// queueing. `None` disables the gate. /// long-prompt traffic). Fails open without a fresh sample: the router-side
/// /// in-flight counter cannot separate running from waiting requests.
/// Gating on the queue rather than on total depth is what makes this /// Counts sum across DP ranks, so scale the limit with `dp_size`.
/// targeted: `num_waiting_reqs` IS the question the request cares about
/// — will I sit behind other work before my prefill starts — whereas
/// depth only proxies it, and proxies it badly (an engine can queue at
/// 7-8 running on long-prompt traffic, far below its running cap, so a
/// depth threshold either fires on healthy busy workers or misses the
/// workers actually making requests wait).
///
/// The gate reads the engine-published load sample and fails OPEN on a
/// worker with no fresh sample: the router-side in-flight counter cannot
/// separate a running request from a waiting one, so there is no honest
/// substitute to compare the limit against.
///
/// Note the firing point scales with `dp_size`: the sample sums `waiting`
/// across a worker's DP ranks while a request lands on one of them, so
/// scale the limit with `--dp-size` on DP-attention deployments.
///
/// The companion `saturation_queue_floor` cancels the gate's diversions
/// when they have no payoff (nothing in the fleet reads below the
/// floor).
pub worker_queue_limit: Option<u64>, pub worker_queue_limit: Option<u64>,
/// Saturation pin (`--saturation-queue-floor`): cancels queue-gate /// Keep the least-pressured prefix owner when the queue gate rejects all admitted
/// diversions that have no payoff. When no cache candidate survives /// cache candidates and no fresh fleet queue is below this floor. Unknown queues
/// both `worker_queue_limit` and hard admission, at least one was over /// do not count as idle — the opposite of the gate's fail-open, deliberately: the
/// the limit, AND no worker in the routable fleet has a fresh queue /// pin asks whether a provably better destination exists, and an unknown queue is
/// reading strictly below this floor, the diverted request would wait /// not proof. Requires `floor <= worker_queue_limit`; scale with `dp_size`.
/// wherever it lands — so it pins to the least-pressured prefix owner
/// instead of cold-prefilling on a non-owner (which evicts other
/// prefixes and manufactures the next round of misses). `None` — the
/// default — preserves the pure gate behavior.
///
/// Polarity note: a worker with no fresh sample does NOT count as idle
/// — the opposite of the gate's fail-open, and deliberately so. The
/// gate keeps affinity because that is the safe default action; the
/// pin asks whether a *provably better* destination exists, and an
/// unknown queue is not proof. Both polarities leave the request with
/// its prefix owner when the signal is missing.
///
/// The CLI enforces `floor <= worker_queue_limit` and requires the
/// gate; like the limit, scale the floor with `dp_size`.
pub saturation_queue_floor: Option<u64>, pub saturation_queue_floor: Option<u64>,
/// Number of random candidates sampled for the min-load fallback /// Min-load fallback sample size; defaults to power-of-two. At least the pool
/// (`--min-load-choices`); the least-pressured of the sample wins. /// size chooses the exact minimum with random ties; 1 draws uniformly without
/// [`DEFAULT_MIN_LOAD_CHOICES`] is the pre-existing power-of-2 /// a backup for admission or pressure guards. Separate from cache-owner limits.
/// behavior, so upgrading changes nothing. `k >= pool` skips the
/// shuffle and returns the exact minimum, with ties broken randomly
/// (an idle fleet ties on every comparison, so a fixed order would pin
/// every fallback dispatch to one worker); `k = 1` is a uniform draw
/// within the tier, and its sample has no second member, so the
/// proposal carries no backup and admission loses its backup-admission
/// and pressure-guard paths. The
/// `--cache-candidate-*` knobs bound the cache-affinity OWNER candidate
/// set; this bounds the min-load FALLBACK sample used when no owner is
/// usable.
pub min_load_choices: usize, pub min_load_choices: usize,
} }
@@ -610,18 +504,13 @@ impl Default for AffinityConfig {
} }
} }
/// Per-model sticky-session tuning. Built from the `--routing-key-header` /// Sticky routing settings; the CLI validates the header name and positive durations.
/// / `--sticky-*` flags by [`crate::config::cli::Cli::into_config`], which
/// also validates that `header_name` parses as an HTTP header name.
#[derive(Debug, Clone)] #[derive(Debug, Clone)]
pub struct StickyConfig { pub struct StickyConfig {
/// Request header carrying the routing key. Validated to parse as a /// Request header carrying the routing key. Validated to parse as a
/// `http::HeaderName` at config-build time. /// `http::HeaderName` at config-build time.
pub header_name: String, pub header_name: String,
/// Policy used to pick a worker when a request has no routing key, and /// Fallback for new or missing routing keys.
/// to pick the initial worker when a new key is first seen. One of
/// `round_robin` / `random` / `power_of_two` / `load_based` — the
/// dependency-free policies the factory can build standalone.
pub fallback_policy: StickyFallbackKind, pub fallback_policy: StickyFallbackKind,
/// Evict an assignment after it has been idle (unreferenced) this many /// Evict an assignment after it has been idle (unreferenced) this many
/// seconds. Bounds the map against unbounded routing-key cardinality. /// seconds. Bounds the map against unbounded routing-key cardinality.
@@ -650,10 +539,7 @@ impl Default for StickyConfig {
#[derive(Debug, Clone)] #[derive(Debug, Clone)]
pub struct CircuitBreakerConfig { pub struct CircuitBreakerConfig {
/// Consecutive failures required before the breaker opens. Encoded /// Consecutive failures before opening the breaker; zero is invalid.
/// as `NonZeroU32` so `--cb-threshold 0` (which would open the
/// breaker before any failure) is rejected at CLI-parse time rather
/// than silently behaving as "always open".
pub threshold: NonZeroU32, pub threshold: NonZeroU32,
pub cool_down_secs: u64, pub cool_down_secs: u64,
} }
@@ -670,43 +556,15 @@ pub enum DiscoveryBackend {
K8s(K8sDiscoveryConfig), K8s(K8sDiscoveryConfig),
} }
/// Fixed list of worker URLs. Each URL is registered once at startup; /// Workers registered at startup. Roles, models, and bootstrap ports come from
/// `mode`, `model_ids`, and `bootstrap_port` are resolved per-worker /// `/server_info`; topology changes require a restart.
/// from `/server_info` (see [`crate::workers::introspect`]).
///
/// No file watcher, no hot-reload: topology change requires a restart.
#[derive(Debug, Clone)] #[derive(Debug, Clone)]
pub struct StaticUrlsDiscoveryConfig { pub struct StaticUrlsDiscoveryConfig {
pub urls: Vec<String>, pub urls: Vec<String>,
} }
/// Configuration for the Kubernetes `EndpointSlice` discovery backend. /// Kubernetes EndpointSlice discovery. Selectors classify slices; worker roles
/// Built from the `--service-discovery*` / `--selector` / `--prefill-selector` /// and bootstrap ports come from `/server_info` introspection.
/// / `--decode-selector` flags by [`crate::config::cli::Cli::build_discovery`].
///
/// Two operating modes, distinguished by which selector flags are set:
///
/// 1. **Plain** — all matched workers share the same role:
/// `--service-discovery-namespace default --selector app=sglang`
///
/// 2. **PD disaggregation** — prefill and decode workers are separated by
/// different selectors:
/// `--service-discovery-namespace default
/// --prefill-selector app=sglang,role=prefill
/// --decode-selector app=sglang,role=decode`
///
/// In PD mode, the selectors drive **slice-classification** (which
/// EndpointSlices feed the prefill pool vs the decode pool). The actual
/// `WorkerMode` and `bootstrap_port` for each worker are filled in by
/// the worker manager from each worker's `/server_info` introspection,
/// so PD works without any pod-level annotations — see
/// [`crate::workers::introspect`] for the `disaggregation_mode` and
/// `disaggregation_bootstrap_port` extraction.
///
/// [`resolve_mode`] validates the selector flags and produces the
/// resolved [`K8sDiscoveryMode`] once, at construction in
/// [`crate::config::cli::Cli::build_discovery`] — so an invalid selector
/// combination is unrepresentable here.
#[derive(Debug, Clone)] #[derive(Debug, Clone)]
pub struct K8sDiscoveryConfig { pub struct K8sDiscoveryConfig {
pub namespace: String, pub namespace: String,
@@ -714,14 +572,8 @@ pub struct K8sDiscoveryConfig {
pub mode: K8sDiscoveryMode, pub mode: K8sDiscoveryMode,
} }
/// Resolved discovery mode, produced by [`resolve_mode`] from the CLI /// Validated selector mode. Plain selectors run server-side; PD selectors
/// selector flags and stored on [`K8sDiscoveryConfig`]. /// classify EndpointSlices client-side.
///
/// The discovery backend uses this to:
/// * pick the server-side `LIST` label selector (Plain: the single selector;
/// PD: empty, with classification done client-side per slice), and
/// * assign each `EndpointSlice` a [`crate::discovery::WorkerMode`] in
/// `extract_workers`.
#[derive(Debug, Clone, PartialEq, Eq)] #[derive(Debug, Clone, PartialEq, Eq)]
pub enum K8sDiscoveryMode { pub enum K8sDiscoveryMode {
/// One global label selector; every matched EndpointSlice becomes a /// One global label selector; every matched EndpointSlice becomes a
@@ -774,52 +626,31 @@ pub enum ConfigError {
IdenticalPdSelectors, IdenticalPdSelectors,
} }
/// Returns `true` when `selector` has zero non-empty terms after /// An empty selector matches every slice, which is invalid for a PD role.
/// trimming and splitting on `,`. `labels_match_selector` then returns
/// `true` for every label set, which is the "matches everything"
/// degenerate case PD mode must reject.
fn is_selector_empty(selector: &str) -> bool { fn is_selector_empty(selector: &str) -> bool {
selector.split(',').all(|t| t.trim().is_empty()) selector.split(',').all(|t| t.trim().is_empty())
} }
/// Canonicalize a comma-separated equality selector to a sorted list of /// Normalize validated equality terms for comparison, matching runtime whitespace
/// parsed `(key, value)` tuples. Comparison happens at the parsed-term /// and `=`/`==` handling. Term order does not affect matching.
/// level — *not* the raw string level — because `labels_match_selector` fn canonical_selector(selector: &str) -> Vec<(&str, &str)> {
/// already strips whitespace and treats `key=value` and `key==value` as let mut terms: Vec<_> = selector
/// the same equality test. Comparing raw strings would let
/// `"app=sglang"` vs `"app==sglang"` (and `"app = sglang"` vs
/// `"app=sglang"`) past the identical-selector check, even though
/// `classify_mode` would treat them identically at runtime — exactly
/// the silent decode-pool-empty failure mode this check exists to
/// prevent.
///
/// Returns an empty `Vec` for selectors with no parseable terms
/// (whitespace-only, comma-only, or any term that doesn't match the
/// `key=value` / `key==value` grammar). Callers must run
/// [`is_equality_selector`] before this to surface malformed
/// selectors as `UnsupportedSelectorGrammar`.
fn canonical_selector(selector: &str) -> Vec<(String, String)> {
let mut terms: Vec<(String, String)> = selector
.split(',') .split(',')
.filter_map(|raw| { .filter_map(|raw| {
let term = raw.trim(); let term = raw.trim();
if term.is_empty() { if term.is_empty() {
return None; return None;
} }
// Mirror `labels_match_selector`: prefer the `==` alias so a // Prefer `==` so its second equals sign does not become part of the value.
// term like `key==value` parses to `(key, value)` instead of
// `(key, =value)`.
let (k, v) = term.split_once("==").or_else(|| term.split_once('='))?; let (k, v) = term.split_once("==").or_else(|| term.split_once('='))?;
Some((k.trim().to_string(), v.trim().to_string())) Some((k.trim(), v.trim()))
}) })
.collect(); .collect();
terms.sort(); terms.sort_unstable();
terms terms
} }
/// Returns `true` when `selector` parses as a comma-separated equality /// Check the equality-only grammar supported by client-side PD matching.
/// selector — every term has the shape `key=value` or `key==value`.
/// See [`ConfigError::UnsupportedSelectorGrammar`] for rationale.
fn is_equality_selector(selector: &str) -> bool { fn is_equality_selector(selector: &str) -> bool {
for term in selector.split(',') { for term in selector.split(',') {
let term = term.trim(); let term = term.trim();
@@ -835,9 +666,7 @@ fn is_equality_selector(selector: &str) -> bool {
continue; continue;
} }
if let Some((k, _value)) = term.split_once('=') { if let Some((k, _value)) = term.split_once('=') {
// Reject `!=` (rendered as `key!` + `=value` by split_once). // Reject `!=`; empty label values are valid.
// Empty value is legal in K8s — `label_selector = "tier="`
// matches pods with `tier=""` — so we don't constrain it.
if k.trim().is_empty() || k.trim().ends_with('!') { if k.trim().is_empty() || k.trim().ends_with('!') {
return false; return false;
} }
@@ -849,10 +678,7 @@ fn is_equality_selector(selector: &str) -> bool {
true true
} }
/// Validate the selector combination and return the resolved /// Resolve and validate the plain or prefill/decode selector combination.
/// [`K8sDiscoveryMode`]. Called once at construction by
/// [`crate::config::cli::Cli::build_discovery`], so an invalid
/// combination can never be stored on a [`K8sDiscoveryConfig`].
pub fn resolve_mode( pub fn resolve_mode(
label_selector: Option<&str>, label_selector: Option<&str>,
prefill_selector: Option<&str>, prefill_selector: Option<&str>,
@@ -860,57 +686,30 @@ pub fn resolve_mode(
) -> Result<K8sDiscoveryMode, ConfigError> { ) -> Result<K8sDiscoveryMode, ConfigError> {
match (label_selector, prefill_selector, decode_selector) { match (label_selector, prefill_selector, decode_selector) {
(Some(label), None, None) => { (Some(label), None, None) => {
// Plain mode pushes `label` to the K8s API as the // Plain selectors run on the Kubernetes API, which supports the full grammar.
// server-side `labelSelector` of the EndpointSlice // PD selectors are checked client-side and only support equality.
// watcher (`watcher::Config::default().labels(&label)`
// in `discovery::k8s::spawn`). K8s itself parses the
// full label-selector grammar — equality, set-based
// (`in` / `notin`), presence (`key` / `!key`), and
// `!=` — and rejects malformed selectors at
// watch-start time. So we don't grammar-check `label`
// here and let the K8s API be the syntax authority. PD
// mode, in contrast, evaluates selectors client-side via
// `labels_match_selector` which only understands
// equality — so PD selectors are still grammar-checked
// below.
Ok(K8sDiscoveryMode::Plain { Ok(K8sDiscoveryMode::Plain {
label_selector: label.to_string(), label_selector: label.to_string(),
}) })
} }
(None, Some(prefill), Some(decode)) => { (None, Some(prefill), Some(decode)) => {
// Both selectors validated individually so the operator // Validate both grammars before checking for empty or identical selectors.
// sees which one is malformed. WorkerMode + bootstrap_port let selectors = [("prefill", prefill), ("decode", decode)];
// for each prefill pod are filled in by the worker for (selector, value) in selectors {
// manager from each worker's `/server_info` — these if !is_equality_selector(value) {
// selectors only drive client-side classification per return Err(ConfigError::UnsupportedSelectorGrammar {
// EndpointSlice (see `classify_mode` in discovery/k8s.rs). selector,
if !is_equality_selector(prefill) { value: value.to_string(),
return Err(ConfigError::UnsupportedSelectorGrammar { });
selector: "prefill", }
value: prefill.to_string(),
});
} }
if !is_equality_selector(decode) { // Empty PD selectors match everything and starve the opposite role.
return Err(ConfigError::UnsupportedSelectorGrammar { for (selector, value) in selectors {
selector: "decode", if is_selector_empty(value) {
value: decode.to_string(), return Err(ConfigError::EmptyPdSelector { selector });
}); }
} }
// Empty PD selector matches every EndpointSlice at // Prefill wins when both selectors match, leaving decode empty.
// runtime; combined with classify_mode's prefill-first
// ordering, an empty selector would silently funnel all
// workers into one role. Reject up front.
if is_selector_empty(prefill) {
return Err(ConfigError::EmptyPdSelector {
selector: "prefill",
});
}
if is_selector_empty(decode) {
return Err(ConfigError::EmptyPdSelector { selector: "decode" });
}
// Identical selectors degrade the same way as an empty
// one: every slice matches both, prefill wins, decode
// stays empty.
if canonical_selector(prefill) == canonical_selector(decode) { if canonical_selector(prefill) == canonical_selector(decode) {
return Err(ConfigError::IdenticalPdSelectors); return Err(ConfigError::IdenticalPdSelectors);
} }
@@ -931,10 +730,6 @@ mod k8s_discovery_config_tests {
#[test] #[test]
fn mode_constructs_pd_disaggregation_from_prefill_and_decode_selectors() { fn mode_constructs_pd_disaggregation_from_prefill_and_decode_selectors() {
// K8s PD now works without per-pod annotations: each worker's
// `/server_info` carries `disaggregation_bootstrap_port`, and the
// worker manager applies it post-discovery. The K8s config layer's
// job is just to validate the selector combination.
let m = resolve_mode(None, Some("app=sglang,role=p"), Some("app=sglang,role=d")) let m = resolve_mode(None, Some("app=sglang,role=p"), Some("app=sglang,role=d"))
.expect("PD mode is now valid"); .expect("PD mode is now valid");
assert_eq!( assert_eq!(
@@ -948,9 +743,6 @@ mod k8s_discovery_config_tests {
#[test] #[test]
fn mode_pd_rejects_set_based_prefill_selector() { fn mode_pd_rejects_set_based_prefill_selector() {
// Both PD selectors get the same equality-only grammar check as
// the plain label_selector. A set-based prefill selector would
// silently match zero pods at runtime → fail-fast at load.
let err = let err =
resolve_mode(None, Some("app in (sglang, vllm)"), Some("app=sglang")).unwrap_err(); resolve_mode(None, Some("app in (sglang, vllm)"), Some("app=sglang")).unwrap_err();
assert!( assert!(
@@ -992,12 +784,8 @@ mod k8s_discovery_config_tests {
); );
} }
/// Plain mode pushes its selector to the K8s API server-side /// Plain selectors run on the Kubernetes API, which supports the full grammar.
/// (`watcher::Config::default().labels(&selector)` in /// PD selectors are checked client-side and only support equality.
/// `discovery::k8s::spawn`), so the full K8s label-selector grammar
/// — including set-based operators like `app in (a,b)` — is
/// supported and must not be grammar-checked at startup. PD mode
/// (checked client-side) is the opposite; see the PD tests below.
#[test] #[test]
fn mode_accepts_set_based_selector_in_plain_mode() { fn mode_accepts_set_based_selector_in_plain_mode() {
let m = resolve_mode(Some("app in (sglang,sglang-small)"), None, None) let m = resolve_mode(Some("app in (sglang,sglang-small)"), None, None)
@@ -1010,9 +798,6 @@ mod k8s_discovery_config_tests {
); );
} }
/// `notin`, presence (`key`), absence (`!key`), and inequality (`!=`)
/// are all valid K8s server-side selector grammar — plain mode must
/// pass them through.
#[test] #[test]
fn mode_accepts_other_set_based_forms_in_plain_mode() { fn mode_accepts_other_set_based_forms_in_plain_mode() {
for raw in [ for raw in [
@@ -1033,15 +818,6 @@ mod k8s_discovery_config_tests {
} }
} }
/// PD mode evaluates selectors *client-side* via
/// `labels_match_selector`, which only handles equality. A set-based
/// PD selector would silently match zero pods → fail-fast at load.
/// Pins the plain-server-side / PD-client-side asymmetry: relaxing
/// the grammar check for plain (see `mode_accepts_set_based_*`
/// above) must not accidentally relax it for PD selectors. Uses
/// `notin` so this test covers a different set-based form than
/// `mode_pd_rejects_set_based_prefill_selector` (which uses `in`)
/// — both must keep failing.
#[test] #[test]
fn mode_pd_rejects_notin_prefill_selector() { fn mode_pd_rejects_notin_prefill_selector() {
let err = let err =
@@ -1101,9 +877,6 @@ mod k8s_discovery_config_tests {
); );
} }
/// Empty plain `label_selector` is valid — matches every
/// EndpointSlice in the namespace (documented K8s behavior; the
/// operator opts in by setting plain mode at all).
#[test] #[test]
fn mode_accepts_empty_plain_label_selector() { fn mode_accepts_empty_plain_label_selector() {
let m = resolve_mode(Some(""), None, None).unwrap(); let m = resolve_mode(Some(""), None, None).unwrap();
@@ -1115,12 +888,6 @@ mod k8s_discovery_config_tests {
); );
} }
/// PD mode is the *opposite* of plain: an empty selector would match
/// every EndpointSlice, and since `classify_mode` checks prefill
/// before decode, an empty `prefill_selector` would classify
/// everything as Prefill — decode pool stays empty and the resolver
/// surfaces the wrong `no_decode_workers_available` error. Fail-fast
/// at config load.
#[test] #[test]
fn mode_pd_rejects_empty_prefill_selector() { fn mode_pd_rejects_empty_prefill_selector() {
let err = resolve_mode(None, Some(""), Some("role=decode")).unwrap_err(); let err = resolve_mode(None, Some(""), Some("role=decode")).unwrap_err();
@@ -1144,9 +911,6 @@ mod k8s_discovery_config_tests {
); );
} }
/// Whitespace-only / comma-only PD selector parses to zero terms in
/// `labels_match_selector` and matches every slice at runtime — same
/// failure mode as a literal empty string.
#[test] #[test]
fn mode_pd_rejects_whitespace_only_prefill_selector() { fn mode_pd_rejects_whitespace_only_prefill_selector() {
let err = resolve_mode(None, Some(" , "), Some("role=decode")).unwrap_err(); let err = resolve_mode(None, Some(" , "), Some("role=decode")).unwrap_err();
@@ -1161,9 +925,6 @@ mod k8s_discovery_config_tests {
); );
} }
/// Identical prefill and decode selectors degrade silently: every
/// slice matches both, but `classify_mode` returns `Prefill` first,
/// so the decode pool stays empty.
#[test] #[test]
fn mode_pd_rejects_identical_prefill_and_decode_selectors() { fn mode_pd_rejects_identical_prefill_and_decode_selectors() {
let err = resolve_mode(None, Some("app=sglang"), Some("app=sglang")).unwrap_err(); let err = resolve_mode(None, Some("app=sglang"), Some("app=sglang")).unwrap_err();
@@ -1184,14 +945,6 @@ mod k8s_discovery_config_tests {
); );
} }
/// `labels_match_selector` accepts both `key=value` and `key==value`
/// for equality and parses them to the same `(key, value)` tuple.
/// Two selectors that differ only in this alias choice are runtime-
/// equivalent — they'd match the same EndpointSlices, then
/// `classify_mode`'s prefill-first ordering would funnel every slice
/// into Prefill, leaving decode empty. The check must canonicalize
/// at the term level (parsed `(key, value)` tuples), not the raw
/// string level.
#[test] #[test]
fn mode_pd_rejects_identical_selectors_under_eq_alias() { fn mode_pd_rejects_identical_selectors_under_eq_alias() {
let err = resolve_mode(None, Some("app=sglang"), Some("app==sglang")).unwrap_err(); let err = resolve_mode(None, Some("app=sglang"), Some("app==sglang")).unwrap_err();
@@ -1201,11 +954,6 @@ mod k8s_discovery_config_tests {
); );
} }
/// Inner whitespace inside a term (`"app = sglang"`) is the same
/// label as no whitespace (`"app=sglang"`) — the runtime
/// `labels_match_selector` trims key and value independently
/// (see `key.trim()` / `expected.trim()` in `k8s.rs`). Canonical
/// form must agree.
#[test] #[test]
fn mode_pd_rejects_identical_selectors_under_inner_whitespace() { fn mode_pd_rejects_identical_selectors_under_inner_whitespace() {
let err = resolve_mode(None, Some("app=sglang"), Some("app = sglang")).unwrap_err(); let err = resolve_mode(None, Some("app=sglang"), Some("app = sglang")).unwrap_err();
@@ -1215,11 +963,6 @@ mod k8s_discovery_config_tests {
); );
} }
/// Term order doesn't matter for label matching, so `"a=1,b=2"` and
/// `"b=2,a=1"` must be treated as identical. (Implied by the sort
/// in `canonical_selector`, but pinned explicitly so a future
/// "preserve user order for diagnostics" refactor can't silently
/// reintroduce the silent-failure bug.)
#[test] #[test]
fn mode_pd_rejects_identical_selectors_under_term_order_permutation() { fn mode_pd_rejects_identical_selectors_under_term_order_permutation() {
let err = let err =
@@ -1230,9 +973,6 @@ mod k8s_discovery_config_tests {
); );
} }
/// Sanity: two selectors that genuinely differ at the term level
/// must still pass validation — the canonicalizer must not be so
/// aggressive that it false-positives on legitimate PD configs.
#[test] #[test]
fn mode_pd_accepts_truly_distinct_selectors() { fn mode_pd_accepts_truly_distinct_selectors() {
let m = resolve_mode( let m = resolve_mode(