Skip to content
Merged
Show file tree
Hide file tree
Changes from 17 commits
Commits
Show all changes
22 commits
Select commit Hold shift + click to select a range
0ac45f0
feat(global-router): derive Hash for RouterReplica
jioffe502 Oct 29, 2026
5fa5d54
feat(global-router): add a router cache to the Global Discovery client
jioffe502 Oct 30, 2026
8e7193e
feat(pool-relay): poll the Global Router replicas from Global Discovery
jioffe502 Oct 30, 2026
a495868
refactor(global-router): publish the router set before marking it cur…
jioffe502 Oct 30, 2026
e0fd0b2
fix(global-router): keep at most 1 KiB of a discovery error body
jioffe502 Oct 30, 2026
78ee36e
test(pool-relay): read from a free port in the router poller host test
jioffe502 Oct 30, 2026
3c212d7
feat(global-router): add GET /pools to the Global Discovery client
jioffe502 Oct 30, 2026
31e214c
refactor(global-router): share the cache refresh bookkeeping
jioffe502 Oct 30, 2026
440bb72
feat(global-router): display a PoolKey as its name for people
jioffe502 Oct 30, 2026
f77c4be
feat(global-router): add a pool cache to the Global Discovery client
jioffe502 Oct 30, 2026
5efe33c
perf(global-router): merge a pool list without rebuilding it as a map
jioffe502 Oct 30, 2026
955cbe1
test(global-router): cover a dropped pool cache, an empty complete li…
jioffe502 Oct 30, 2026
dc76f1c
Merge remote-tracking branch 'origin/main' into jioffe/llm-213-poll-g…
jioffe502 Nov 2, 2026
c18fbc7
fix(global-router): read at most 1 KiB of a discovery error body
jioffe502 Nov 2, 2026
e5311db
test(global-router): check every cause in error_chain and drop restat…
jioffe502 Nov 2, 2026
31796ca
Merge branch 'jioffe/llm-213-poll-global-routers' into jioffe/llm-233…
jioffe502 Nov 2, 2026
05b8354
test(global-router): drop pool test comments that repeat the assertions
jioffe502 Nov 2, 2026
c4bd8b5
Merge branch 'main' into jioffe/llm-213-poll-global-routers
jioffe502 Nov 2, 2026
c69c6f3
test(global-router): sync the kvbm lockfile and trim RouterCache tests
jioffe502 Nov 3, 2026
91b6a9d
Merge branch 'jioffe/llm-213-poll-global-routers' into jioffe/llm-233…
jioffe502 Nov 3, 2026
5330649
Merge remote-tracking branch 'origin/main' into jioffe/llm-233-pool-c…
jioffe502 Nov 3, 2026
74e715d
Merge remote-tracking branch 'origin/main' into jioffe/llm-233-pool-c…
jioffe502 Nov 3, 2026
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

1 change: 1 addition & 0 deletions Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -68,6 +68,7 @@ aisimulate-core = { version = "=0.13.0-dev.202610290000000087" }

# Local crates
dynamo-runtime = { path = "lib/runtime", version = "1.7.0" }
dynamo-global-discovery-client = { path = "lib/global-discovery-client", version = "1.7.0" }
dynamo-llm = { path = "lib/llm", version = "1.7.0", default-features = false }
dynamo-sglang-generate = { path = "lib/generate-formats/sglang", version = "1.7.0" }
dynamo-vllm-generate = { path = "lib/generate-formats/vllm", version = "1.7.0" }
Expand Down
4 changes: 3 additions & 1 deletion components/src/dynamo/pool_relay/__main__.py
Original file line number Diff line number Diff line change
Expand Up @@ -81,6 +81,7 @@ async def worker(runtime: DistributedRuntime) -> None:
tuning=dict(args.tuning) or None,
sources_file=args.sources_file,
connection_revision=args.connection_revision,
global_discovery_url=args.global_discovery_url,
)
await relay.start()
diagnostics = PoolRelayDiagnostics(relay)
Expand All @@ -89,12 +90,13 @@ async def worker(runtime: DistributedRuntime) -> None:

logger.info(
"Pool Relay started for cluster_id=%s namespaces=%s watch_all=%s "
"endpoint_prefixes=%s wan_bind=%s",
"endpoint_prefixes=%s wan_bind=%s global_discovery_url=%s",
args.cluster_id,
args.namespaces,
args.watch_all,
args.endpoint_prefixes,
args.bind,
args.global_discovery_url,
)
endpoint_tasks: list[asyncio.Future[object]] = []
# A terminal host failure only resolves wait_for_shutdown(); the diagnostics
Expand Down
12 changes: 12 additions & 0 deletions components/src/dynamo/pool_relay/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,7 @@ class PoolRelayCliConfig:
tuning: tuple[tuple[str, int], ...]
sources_file: str | None = None
connection_revision: str | None = None
global_discovery_url: str | None = None


class RelayShutdownWaiter(Protocol):
Expand Down Expand Up @@ -163,6 +164,10 @@ def parse_args(
parser.add_argument("--expected-unique-blocks", type=_positive_int)

parser.add_argument("--bind", help="Plaintext WAN gRPC listen address")
parser.add_argument(
"--global-discovery-url",
help="Global Discovery Service root to read Global Router replicas from",
)

parsed = parser.parse_args(argv)
cluster_id = _string_value(parsed.cluster_id, environment, "DYN_CLUSTER_ID")
Expand Down Expand Up @@ -275,6 +280,12 @@ def parse_args(
if bind is not None and not bind.strip():
parser.error("--bind must not be empty")

global_discovery_url = _string_value(
parsed.global_discovery_url, environment, "DYN_RELAY_GLOBAL_DISCOVERY_URL"
)
if global_discovery_url is not None and not global_discovery_url.strip():
parser.error("--global-discovery-url must not be empty")

tuning: list[tuple[str, int]] = []
for key in TUNING_KEYS:
environment_name = f"DYN_RELAY_{key.upper()}"
Expand Down Expand Up @@ -304,6 +315,7 @@ def parse_args(
tuning=tuple(tuning),
sources_file=sources_file,
connection_revision=connection_revision,
global_discovery_url=global_discovery_url,
)


Expand Down
25 changes: 25 additions & 0 deletions components/src/dynamo/pool_relay/tests/test_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@ def test_default_config() -> None:
assert config.namespaces == ()
assert config.endpoint_prefixes == ()
assert config.bind is None
assert config.global_discovery_url is None
assert config.tuning == ()


Expand Down Expand Up @@ -56,13 +57,16 @@ def test_cli_overrides_environment() -> None:
"[::1]:0",
"--expected-unique-blocks",
"128",
"--global-discovery-url",
"http://cli-discovery:8080/",
],
{
"DYN_CLUSTER_ID": "environment-dc",
"DYN_RELAY_NAMESPACES": "environment-a, environment-b",
"DYN_RELAY_ENDPOINT_PREFIXES": "environment-a.backend",
"DYN_RELAY_BIND": "127.0.0.1:0",
"DYN_RELAY_EXPECTED_UNIQUE_BLOCKS": "64",
"DYN_RELAY_GLOBAL_DISCOVERY_URL": "http://environment-discovery:8080/",
},
)

Expand All @@ -72,6 +76,7 @@ def test_cli_overrides_environment() -> None:
assert config.endpoint_prefixes == ("cli-a.backend", "cli-b.backend")
assert config.bind == "[::1]:0"
assert config.expected_unique_blocks == 128
assert config.global_discovery_url == "http://cli-discovery:8080/"


def test_config_from_environment() -> None:
Expand All @@ -83,6 +88,7 @@ def test_config_from_environment() -> None:
"DYN_RELAY_ENDPOINT_PREFIXES": "prod-a.backend, prod-b.backend",
"DYN_RELAY_BIND": "127.0.0.1:0",
"DYN_RELAY_EXPECTED_UNIQUE_BLOCKS": "64",
"DYN_RELAY_GLOBAL_DISCOVERY_URL": "http://discovery:8080/",
},
)

Expand All @@ -92,6 +98,7 @@ def test_config_from_environment() -> None:
assert config.watch_all is False
assert config.bind == "127.0.0.1:0"
assert config.expected_unique_blocks == 64
assert config.global_discovery_url == "http://discovery:8080/"


@pytest.mark.parametrize(
Expand Down Expand Up @@ -364,6 +371,24 @@ def test_expected_unique_blocks_rejects_invalid_numbers(
)


@pytest.mark.parametrize("source", ["cli", "environment"])
def test_global_discovery_url_rejects_empty(
source: str, capsys: pytest.CaptureFixture[str]
) -> None:
if source == "cli":
argv = ["--global-discovery-url", " "]
environment = {}
else:
argv = []
environment = {"DYN_RELAY_GLOBAL_DISCOVERY_URL": " "}
assert_cli_error(
["--cluster-id", "dc-a", *argv],
environment,
capsys,
"--global-discovery-url must not be empty",
)


def test_sources_file_is_explicitly_empty_until_runtime_loads_sources() -> None:
config = parse_args(
["--cluster-id", "dc", "--sources-file", "/etc/relay/sources.json"],
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ see [Pool Relay Concepts](../../developer-guide/knowledge-base/modular-component
| `--endpoint-prefix` | `DYN_RELAY_ENDPOINT_PREFIXES` | No prefix filter | Repeat the CLI option, or use a comma-separated environment value. Match namespace/component/endpoint segments. |
| `--expected-unique-blocks` | `DYN_RELAY_EXPECTED_UNIQUE_BLOCKS` | `1048576` | Positive expected number of unique blocks per pool; sizes the CKF, not a global Relay memory limit. |
| `--bind` | `DYN_RELAY_BIND` | No listener | Plaintext gRPC socket address, such as `127.0.0.1:5561` or `[::1]:5561`. Requires a numeric IP and port. |
| `--global-discovery-url` | `DYN_RELAY_GLOBAL_DISCOVERY_URL` | Unset | Root of the Global Discovery Service, such as `http://global-discovery:8080/`. The Relay reads the Global Router replica list from it every 5 seconds. When a read fails, it keeps the last list and reports why in the health field `global_routers_stale_reason`. When unset, the Relay reads no list. |
| `--help` | None | — | Print CLI usage and exit. |

The protocol and gRPC server are included in the standard build. Omitting `--bind` disables only
Expand Down Expand Up @@ -163,7 +164,7 @@ of SHA-256 over the UTF-8 cluster ID. These are not HTTP paths or WAN RPCs.

| Endpoint | Availability | Request and response |
| --- | --- | --- |
| `health` | Always | Empty request; reports `healthy`, shutdown state, endpoint counts, and WAN state/errors. |
| `health` | Always | Empty request; reports `healthy`, shutdown state, endpoint counts, WAN state/errors, `global_discovery_enabled`, and `global_routers_stale_reason`. |
| `stats` | `ckf-diagnostics` build feature | Empty request; returns detailed endpoint/pool statistics. |
| `snapshot` | `ckf-diagnostics` build feature | Requires `serving_endpoint`; returns that endpoint's diagnostic snapshot. |

Expand Down
26 changes: 26 additions & 0 deletions lib/bindings/python/Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

8 changes: 8 additions & 0 deletions lib/bindings/python/rust/llm/pool_relay.rs
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@ impl PoolRelay {
tuning=None,
sources_file=None,
connection_revision=None,
global_discovery_url=None,
))]
#[allow(clippy::too_many_arguments)]
fn new(
Expand All @@ -57,6 +58,7 @@ impl PoolRelay {
tuning: Option<std::collections::HashMap<String, u64>>,
sources_file: Option<String>,
connection_revision: Option<String>,
global_discovery_url: Option<String>,
) -> PyResult<Self> {
if namespace_filter.is_some() && namespaces.is_some() {
return Err(PyValueError::new_err(
Expand Down Expand Up @@ -231,6 +233,12 @@ impl PoolRelay {
},
producer,
transport,
global_discovery_url: global_discovery_url
.map(|url| url.parse())
.transpose()
.map_err(|error| {
PyValueError::new_err(format!("invalid global_discovery_url: {error}"))
})?,
},
inner: Arc::new(OnceCell::new()),
})
Expand Down
1 change: 1 addition & 0 deletions lib/bindings/python/src/dynamo/_core.pyi
Original file line number Diff line number Diff line change
Expand Up @@ -2828,6 +2828,7 @@ class PoolRelay:
tuning: Optional[Dict[str, int]] = None,
sources_file: Optional[str] = None,
connection_revision: Optional[str] = None,
global_discovery_url: Optional[str] = None,
) -> None:
...

Expand Down
20 changes: 20 additions & 0 deletions lib/global-common/src/pool.rs
Original file line number Diff line number Diff line change
Expand Up @@ -76,6 +76,18 @@ pub enum PoolKeyError {
InvalidField { field: &'static str },
}

/// The pool's name for people, `<cluster_id>/<namespace>/<dgd_name>`, as logs
/// and metrics show it. A pool's identity is its [`PoolId`].
impl fmt::Display for PoolKey {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(
f,
"{}/{}/{}",
self.cluster_id, self.namespace, self.dgd_name
)
}
}
Comment thread
jioffe502 marked this conversation as resolved.
Resolved

impl PoolKey {
pub fn new(
cluster_id: impl Into<String>,
Expand Down Expand Up @@ -212,6 +224,14 @@ mod tests {
assert_eq!(serde_json::to_string(&id).unwrap(), format!("\"{id}\""));
}

#[test]
fn a_key_displays_as_its_name_for_people() {
assert_eq!(
key("ohio", "dynamo", "mocker-1").to_string(),
"ohio/dynamo/mocker-1"
);
}

#[test]
fn key_json_is_checked_when_read() {
let pool = key("ohio", "dynamo", "mocker-1");
Expand Down
2 changes: 1 addition & 1 deletion lib/global-common/src/replicas.rs
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@ use crate::watch::WatchEvent;
pub struct RouterReplicaId(pub String);

/// A Global Router replica and the address workload clusters reach it at.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[derive(Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub struct RouterReplica {
pub replica: RouterReplicaId,
pub url: Url,
Expand Down
5 changes: 4 additions & 1 deletion lib/global-discovery-client/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -17,8 +17,11 @@ reqwest = { workspace = true }
serde = { workspace = true }
serde_json = { workspace = true }
thiserror = { workspace = true }
tokio = { workspace = true }
tokio-util = { workspace = true }
tracing = { workspace = true }
url = { workspace = true }

[dev-dependencies]
axum = { workspace = true }
tokio = { workspace = true }
tokio = { workspace = true, features = ["test-util"] }
Loading
Loading