Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .github/codeowners/areas.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -76,6 +76,7 @@ areas:
- docs/fern/pages/kubernetes/kv-aware-routing/kv-dc-relay.md
- docs/fern/pages/reference/components/kv-dc-relay-configuration.md
- lib/bindings/python/rust/llm/kv_dc_relay.rs
- lib/global-discovery-client/
- lib/llm/src/kv_dc_relay.rs
- lib/llm/src/kv_dc_relay/
- lib/llm/tests/fixtures/relay-sources-canonical.json
Expand Down
1 change: 1 addition & 0 deletions CODEOWNERS
Original file line number Diff line number Diff line change
Expand Up @@ -88,6 +88,7 @@
/components/src/dynamo/thunderagent_router/ @ai-dynamo/dynamo-router-codeowners

# === global-router (@ai-dynamo/dynamo-global-router-codeowners) ===
/lib/global-discovery-client/ @ai-dynamo/dynamo-global-router-codeowners
/components/src/dynamo/kv_dc_relay/ @ai-dynamo/dynamo-global-router-codeowners

# === backend-vllm (@ai-dynamo/dynamo-backend-vllm-codeowners) ===
Expand Down
12 changes: 12 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 @@ -12,6 +12,7 @@ members = [
"lib/kv-hashing",
"lib/mocker",
"lib/kv-router",
"lib/global-discovery-client",
"lib/router-plugins/catalog",
"lib/router-plugins/builtin",
"examples/router/custom-policy-example/soft-pin-repin",
Expand Down
22 changes: 22 additions & 0 deletions lib/global-discovery-client/Cargo.toml
Original file line number Diff line number Diff line change
@@ -0,0 +1,22 @@
# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0

[package]
name = "dynamo-global-discovery-client"
description = "HTTP client for the Global Router's Global Discovery Service"
version.workspace = true
edition.workspace = true
authors.workspace = true
license.workspace = true
homepage.workspace = true
repository.workspace = true

[dependencies]
reqwest = { workspace = true }
serde = { workspace = true }
thiserror = { workspace = true }
url = { workspace = true }

[dev-dependencies]
axum = { workspace = true }
tokio = { workspace = true }
154 changes: 154 additions & 0 deletions lib/global-discovery-client/src/client.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,154 @@
// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
// SPDX-License-Identifier: Apache-2.0

use std::time::Duration;

use reqwest::StatusCode;
use thiserror::Error;
use url::Url;

use crate::replicas::ReplicaEndpoint;
use crate::wire::{PoolRegistration, RoutersResponse};

#[derive(Debug, Error)]
#[non_exhaustive]
pub enum DiscoveryClientError {
/// The base URL cannot have a path appended, for example `mailto:`.
#[error("{0} cannot be a base URL")]
InvalidBaseUrl(Url),
#[error("building the HTTP client failed")]
Build(#[source] reqwest::Error),
#[error("request to {url} failed")]
Request {
url: Url,
#[source]
source: reqwest::Error,
},
#[error("{url} returned {status}: {body}")]
Status {
url: Url,
status: StatusCode,
body: String,
},
/// The body is not in the shape this client expects, for example a
/// renamed field or a router URL that does not parse.
#[error("{url} returned a body this client cannot parse")]
Decode {
url: Url,
#[source]
source: reqwest::Error,
},
}

pub struct GlobalDiscoveryClient {
http: reqwest::Client,
routers_url: Url,
pools_url: Url,
}

impl GlobalDiscoveryClient {
Comment thread
jioffe502 marked this conversation as resolved.
Resolved
/// `base_url` is the root of the service, for example
/// `http://global-discovery:8080/`. The client keeps any path in it as a
/// prefix. `request_timeout` bounds each call.
pub fn new(base_url: Url, request_timeout: Duration) -> Result<Self, DiscoveryClientError> {
let http = reqwest::Client::builder()
.timeout(request_timeout)
.build()
.map_err(DiscoveryClientError::Build)?;
Ok(Self {
http,
routers_url: endpoint(&base_url, "routers")?,
pools_url: endpoint(&base_url, "pools")?,
})
}

/// Calls `GET /routers` and returns every Global Router replica the
/// service lists.
///
/// Each call returns the full current set. If any entry does not parse,
/// the whole call fails, so a caller never acts on a partial list.
Comment thread
jioffe502 marked this conversation as resolved.
Resolved
pub async fn routers(&self) -> Result<Vec<ReplicaEndpoint>, DiscoveryClientError> {
let url = &self.routers_url;
let response = self.send(self.http.get(url.clone()), url).await?;
Comment thread
jioffe502 marked this conversation as resolved.
Resolved
let body: RoutersResponse =
response
.json()
.await
.map_err(|source| DiscoveryClientError::Decode {
url: url.clone(),
source,
})?;
Ok(body.routers)
}

/// Calls `POST /pools` to register a pool or refresh its registration.
/// The PoolRelay calls this on a schedule, and each call is also the
/// pool's heartbeat.
pub async fn register_pool(
&self,
registration: &PoolRegistration,
) -> Result<(), DiscoveryClientError> {
let url = &self.pools_url;
self.send(self.http.post(url.clone()).json(registration), url)
.await?;
Ok(())
}

async fn send(
&self,
request: reqwest::RequestBuilder,
url: &Url,
) -> Result<reqwest::Response, DiscoveryClientError> {
let response = request
.send()
.await
.map_err(|source| DiscoveryClientError::Request {
url: url.clone(),
source,
})?;
let status = response.status();
if status.is_success() {
return Ok(response);
}
Comment thread
jioffe502 marked this conversation as resolved.
Resolved
Err(DiscoveryClientError::Status {
url: url.clone(),
status,
body: response.text().await.unwrap_or_default(),
Comment thread
jioffe502 marked this conversation as resolved.
Resolved
})
}
}

/// `base_url` with `segment` appended, keeping any path it already has.
fn endpoint(base_url: &Url, segment: &str) -> Result<Url, DiscoveryClientError> {
let mut url = base_url.clone();
url.path_segments_mut()
.map_err(|()| DiscoveryClientError::InvalidBaseUrl(base_url.clone()))?
.pop_if_empty()
.push(segment);
Ok(url)
}

#[cfg(test)]
mod tests {
use super::*;

#[test]
fn endpoint_keeps_the_base_path() {
for (base, expected) in [
("http://hub:8080", "http://hub:8080/pools"),
(
"http://hub:8080/discovery",
"http://hub:8080/discovery/pools",
),
(
"http://hub:8080/discovery/",
"http://hub:8080/discovery/pools",
),
] {
let url = endpoint(&base.parse().unwrap(), "pools").unwrap();
assert_eq!(url.as_str(), expected, "{base}");
}
let error = endpoint(&"mailto:ops@example.com".parse().unwrap(), "pools").unwrap_err();
assert!(matches!(error, DiscoveryClientError::InvalidBaseUrl(_)));
}
}
20 changes: 20 additions & 0 deletions lib/global-discovery-client/src/lib.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,20 @@
// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
// SPDX-License-Identifier: Apache-2.0

//! Client for the two calls a PoolRelay makes to the Global Discovery
//! Service: listing every Global Router replica
//! ([`GlobalDiscoveryClient::routers`]) and registering its pool
//! ([`GlobalDiscoveryClient::register_pool`]).
//!
//! The crate also defines the read-side types that the client and the service
//! share: the router replica types and the list-and-watch events.

mod client;
mod replicas;
mod watch;
mod wire;

pub use client::{DiscoveryClientError, GlobalDiscoveryClient};
pub use replicas::{ReplicaEndpoint, ReplicaEvent, ReplicaId};
pub use watch::{Completeness, Snapshot, WatchEvent};
pub use wire::{Capacity, PoolKey, PoolLocation, PoolRegistration};
22 changes: 22 additions & 0 deletions lib/global-discovery-client/src/replicas.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,22 @@
// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
// SPDX-License-Identifier: Apache-2.0

//! Global Router replicas a PoolRelay connects to. Every replica needs every
//! pool's state, so a PoolRelay opens one connection per replica.

use serde::{Deserialize, Serialize};
use url::Url;

use crate::watch::WatchEvent;

#[derive(Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub struct ReplicaId(pub String);

/// A replica's address as seen from workload clusters.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct ReplicaEndpoint {
pub replica: ReplicaId,
pub url: Url,
}

pub type ReplicaEvent = WatchEvent<ReplicaEndpoint, ReplicaId>;
32 changes: 32 additions & 0 deletions lib/global-discovery-client/src/watch.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,32 @@
// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
// SPDX-License-Identifier: Apache-2.0

//! List-and-watch events, shared by the Global Discovery Service and its
//! clients.
//!
//! A watch starts with one `Added` per current item, then one `Resync` with the
//! same set, then changes. A `Resync` replaces all earlier state.

#[derive(Clone, Debug)]
pub enum WatchEvent<T, K> {
/// A new or changed item. It replaces any earlier item with the same key.
Added(T),
Removed(K),
Resync(Snapshot<T>),
}

#[derive(Clone, Debug)]
pub struct Snapshot<T> {
pub items: Vec<T>,
pub completeness: Completeness,
}

/// Whether a snapshot lists every item, so a new router replica can tell a
/// complete list from a partial one.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Completeness {
/// Read from a shared store, so it lists every existing item.
Exact,
/// Built from the updates this replica has received so far.
Estimated,
}
53 changes: 53 additions & 0 deletions lib/global-discovery-client/src/wire.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,53 @@
// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
// SPDX-License-Identifier: Apache-2.0

//! JSON exchanged with the Global Discovery Service. The client ignores
//! response fields it does not know, so the service can add fields first.

use serde::{Deserialize, Serialize};
use url::Url;

use crate::replicas::ReplicaEndpoint;

/// `GET /routers` response body.
Comment thread
jioffe502 marked this conversation as resolved.
Resolved
#[derive(Debug, Deserialize)]
pub(crate) struct RoutersResponse {
pub routers: Vec<ReplicaEndpoint>,
}

/// `POST /pools` request body. A PoolRelay sends it on every registration.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct PoolRegistration {
pub key: PoolKey,
pub location: PoolLocation,
/// The model or checkpoint the pool serves.
pub model: String,
pub capabilities: Vec<String>,
/// Local frontends the Global Router dispatches requests to.
pub frontends: Vec<Url>,
pub capacity: Vec<Capacity>,
}

/// `namespace` is the Kubernetes namespace containing the DGD, not the
/// Dynamo runtime namespace derived by the operator.
#[derive(Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub struct PoolKey {
pub site_id: String,
pub namespace: String,
pub dgd_name: String,
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct PoolLocation {
pub region: String,
pub availability_zone: Option<String>,
pub cluster: Option<String>,
pub datacenter: Option<String>,
}

/// The workers of one hardware SKU.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct Capacity {
pub sku: String,
pub count: u32,
}
Loading
Loading