Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
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
19 changes: 17 additions & 2 deletions components/src/dynamo/sglang/publisher.py
Original file line number Diff line number Diff line change
Expand Up @@ -163,6 +163,21 @@ def format_zmq_endpoint(endpoint_template: str, ip_address: str) -> str:
return NetworkAddress(ip_address, parsed.port).to_tcp()


_WILDCARD_HOSTS = {"*", "0.0.0.0", "::"}


def kv_event_connect_ip(endpoint: str) -> str:
"""Return the address the local subscriber uses to reach SGLang's KV event publisher.

SGLang binds a wildcard endpoint without ZMQ_IPV6, so it listens on IPv4 only.
Loopback reaches that listener on every host, including IPv6-only hosts where the
host's own address is IPv6.
"""
if urlparse(endpoint).hostname in _WILDCARD_HOSTS:
return "127.0.0.1"
return get_local_ip_auto()


# Note: We use SGLang's ZmqEventPublisher.offset_endpoint_port() directly
# to ensure perfect alignment between publisher (SGLang) and subscriber (dynamo).
# This is the same pattern used by dynamo+vLLM.
Expand Down Expand Up @@ -384,7 +399,7 @@ def init_kv_event_publish(self) -> List[KvEventPublisher]:
raise ValueError(
"sglang kv_events_config is set but missing 'endpoint'"
)
local_ip = get_local_ip_auto()
connect_ip = kv_event_connect_ip(base_ep)

# Determine DP attention configuration
dp_ranks = get_local_dp_rank_range(self.server_args)
Expand All @@ -406,7 +421,7 @@ def init_kv_event_publish(self) -> List[KvEventPublisher]:
)
continue

zmq_ep = format_zmq_endpoint(zmq_ep, local_ip)
zmq_ep = format_zmq_endpoint(zmq_ep, connect_ip)

logging.info(
f"Setting up ZMQ kv event subscriber for dp_rank={dp_rank} "
Expand Down
17 changes: 17 additions & 0 deletions components/src/dynamo/sglang/tests/test_sglang_publisher.py
Original file line number Diff line number Diff line change
Expand Up @@ -489,6 +489,23 @@ def cleanup(self):
assert metrics_task.cancelled()


@pytest.mark.parametrize(
("endpoint", "expected"),
[
("tcp://*:5557", "127.0.0.1"),
("tcp://0.0.0.0:5557", "127.0.0.1"),
("tcp://[::]:5557", "127.0.0.1"),
("tcp://192.0.2.10:5557", "2001:db8::7"),
],
)
def test_kv_event_connect_ip_uses_loopback_for_wildcard_binds(
monkeypatch, endpoint, expected
):
monkeypatch.setattr(publisher_mod, "get_local_ip_auto", lambda: "2001:db8::7")

assert publisher_mod.kv_event_connect_ip(endpoint) == expected


def test_init_kv_event_publish_uses_worker_id_override(monkeypatch):
calls = []

Expand Down
9 changes: 8 additions & 1 deletion deploy/operator/internal/dynamo/backend_vllm.go
Original file line number Diff line number Diff line change
Expand Up @@ -168,9 +168,16 @@ def _k8s_api():
ns = open(f"{SA}/namespace").read()
return f"https://kubernetes.default.svc/api/v1/namespaces/{ns}/pods"

def resolve_leader_ip():
# gethostbyname is IPv4-only; fall back to IPv6 for IPv6-only clusters.
try:
return socket.gethostbyname(host)
except socket.gaierror:
return socket.getaddrinfo(host, None, socket.AF_INET6, socket.SOCK_STREAM)[0][4][0]

def leader_pod_is_healthy():
try:
ip = socket.gethostbyname(host)
ip = resolve_leader_ip()
except socket.gaierror:
return False, "DNS resolution failed", None, None
try:
Expand Down
5 changes: 3 additions & 2 deletions docs/fern/pages/kubernetes/kv-aware-routing/kv-dc-relay.md
Original file line number Diff line number Diff line change
Expand Up @@ -236,8 +236,9 @@ Relay discovers direct ZMQ publishers through Kubernetes metadata and connects t
advertised TCP addresses. Allow those pod-to-pod connections, worker recovery requests, and
the return path to Relay's response port. Publisher ports may be dynamically allocated;
opening only `5561` is insufficient. On multi-interface workers, set `DYN_EVENT_PLANE_HOST`
to a reachable pod IP if automatic selection advertises the wrong address. It changes the
advertised address, not the listener's bind address; use routable IPv4 addresses for direct ZMQ.
to a reachable IPv4 or IPv6 pod address if automatic selection advertises the wrong address.
Direct ZMQ publishers bind `0.0.0.0` for an advertised IPv4 address or `[::]` for an advertised
IPv6 address. Relay connects to the advertised address and the port assigned to that listener.

Discovery RBAC, the WAN Service, optional mTLS sidecar, and the gRPC checks below are unchanged.

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,8 @@ Every CLI argument has a corresponding environment variable. The CLI argument ta
## HTTP and networking

<ParamField path="--http-host" type="string" default="0.0.0.0">
HTTP listen address.
HTTP listen address: an IPv4 or IPv6 address, with optional brackets around IPv6. `0.0.0.0`
accepts IPv4 connections only; on IPv6-only hosts, set `::`.

Environment variable: `DYN_HTTP_HOST`
</ParamField>
Expand Down
31 changes: 27 additions & 4 deletions docs/fern/pages/reference/components/runtime-configuration.mdx
Original file line number Diff line number Diff line change
Expand Up @@ -210,15 +210,29 @@ Unless a field is marked environment-only, it has both a CLI flag and an environ
usable non-loopback IPv4 address first, then IPv6. On Unix, automatic selection excludes down
interfaces. If neither family has a usable non-loopback address, Dynamo selects IPv4 loopback,
then IPv6 loopback, and falls back to `127.0.0.1` if neither is found. Remote subscribers cannot
reach a loopback address.
reach a loopback address. Automatic selection can choose an IPv4 link-local address or a bridge
address. On hosts with several interfaces, set this variable to a reachable IP address or interface
if automatic selection chooses the wrong address.

Direct ZMQ publishers bind `0.0.0.0` for an advertised IPv4 address or `[::]` for an advertised
IPv6 address. Discovery publishes the concrete address with the port assigned to that listener.
An explicit override must identify an address reachable by subscribers. Event-plane PUB/SUB
sockets support IPv4 and IPv6, including connections to configured ZMQ brokers. This variable
does not affect NATS or ZMQ broker mode.
An explicit override must identify an address reachable by subscribers. The IPv6 wildcard can
also accept IPv4 connections; keep ZMQ listeners on trusted networks.

Event-plane PUB/SUB sockets support IPv4 addresses and bracketed IPv6 addresses, including
connections to configured ZMQ brokers. Hostnames retain IPv4 resolution. For an IPv6 broker,
use a bracketed address such as `tcp://[2001:db8::10]:5555`; Dynamo rejects an unbracketed IPv6
address. This variable does not affect NATS
or ZMQ broker mode.
</ParamField>

<Note>
On IPv6-only hosts, set `DYN_HTTP_HOST=::` on the Frontend and `DYN_SYSTEM_HOST=::` on workers. Both
Comment thread
jthomson04 marked this conversation as resolved.
listeners default to `0.0.0.0`, which accepts IPv4 connections only. The request plane and the direct
ZMQ event plane select an IPv6 address automatically when the host has no usable non-loopback IPv4
address.
</Note>

<ParamField path="DYN_ZMQ_IO_THREADS" type="integer" default="4">
Number of libzmq background I/O threads in the process-wide event-plane context. All event-plane
PUB/SUB sockets in a frontend or worker process share this context. This setting does not change
Expand Down Expand Up @@ -358,6 +372,15 @@ Unless a field is marked environment-only, it has both a CLI flag and an environ
effective per-worker cap is the engine limit plus this queue limit.
</ParamField>

<ParamField path="DYN_SYSTEM_HOST" type="string" default="0.0.0.0">
Listen address for the system status server on `DYN_SYSTEM_PORT`, which serves health, metrics,
and self-hosted model metadata. `0.0.0.0` accepts IPv4 connections only; on IPv6-only hosts, set
`::`. For a wildcard address, Dynamo advertises a non-loopback address from the same family; `::`
Comment thread
jthomson04 marked this conversation as resolved.
also accepts IPv4 connections, so it falls back to a non-loopback IPv4 address. If none exists,
Dynamo advertises loopback and logs a warning; remote callers such as the Frontend cannot reach
that address.
</ParamField>

## Related pages

<CardGroup cols={2}>
Expand Down
47 changes: 47 additions & 0 deletions lib/kv-router/src/services/common/zmq.rs
Original file line number Diff line number Diff line change
Expand Up @@ -59,6 +59,7 @@ impl ZmqSocket {
}

pub(crate) fn connect(&self, endpoint: &str) -> Result<()> {
set_ipv6_for(self.socket(), endpoint)?;
self.socket().connect(endpoint)?;
Ok(())
}
Expand Down Expand Up @@ -161,6 +162,13 @@ impl ZmqSocket {
}
}

/// libzmq resolves TCP hosts as IPv4 unless `ZMQ_IPV6` is set, and snapshots the option per
/// bind or connect. [`validate_endpoint`] rejects unbracketed IPv6 literals.
fn set_ipv6_for(socket: &zmq::Socket, endpoint: &str) -> Result<()> {
socket.set_ipv6(endpoint.starts_with("tcp://["))?;
Ok(())
}

fn configure_common_socket(socket: &zmq::Socket) -> Result<()> {
socket.set_linger(ZMQ_LINGER_MS)?;
socket.set_reconnect_ivl(ZMQ_RECONNECT_IVL_MS)?;
Expand Down Expand Up @@ -231,6 +239,7 @@ pub(crate) fn connect_sub_socket(endpoint: &str) -> Result<SharedSocket> {
pub(crate) fn connect_dealer_socket(endpoint: &str) -> Result<SharedSocket> {
Ok(Arc::new(Mutex::new(build_socket(zmq::DEALER, |socket| {
configure_bidirectional_socket(socket)?;
set_ipv6_for(socket, endpoint)?;
socket.connect(endpoint)?;
Ok(())
})?)))
Expand All @@ -244,6 +253,7 @@ pub(crate) fn connect_dealer_socket(endpoint: &str) -> Result<SharedSocket> {
pub(crate) fn create_bound_pub_socket(endpoint: &str) -> Result<ZmqSocket> {
build_socket(zmq::PUB, |socket| {
configure_send_socket(socket)?;
set_ipv6_for(socket, endpoint)?;
socket.bind(endpoint)?;
Ok(())
})
Expand Down Expand Up @@ -304,3 +314,40 @@ pub(crate) fn validate_endpoint(endpoint: &str) -> Result<()> {
)),
}
}

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

#[tokio::test]
async fn bracketed_ipv6_endpoints_connect() {
if let Err(error) = std::net::TcpListener::bind("[::1]:0") {
eprintln!("Skipping IPv6 ZMQ test: {error}");
return;
}
let mut publisher = create_bound_pub_socket("tcp://[::1]:*").unwrap();
let endpoint = publisher.socket().get_last_endpoint().unwrap().unwrap();
validate_endpoint(&endpoint).unwrap();
let mut subscriber = create_sub_socket(b"").unwrap();
subscriber.connect(&endpoint).unwrap();

tokio::time::timeout(Duration::from_secs(5), async {
loop {
publisher
.send_multipart(vec![b"ipv6".to_vec()])
.await
.unwrap();
if let Ok(frames) =
tokio::time::timeout(Duration::from_millis(50), subscriber.recv_multipart())
.await
{
assert_eq!(frames.unwrap(), vec![b"ipv6".to_vec()]);
break;
}
}
})
.await
.expect("IPv6 SUB socket should receive the message");
}
}
52 changes: 43 additions & 9 deletions lib/llm/src/http/service/service_v2.rs
Original file line number Diff line number Diff line change
Expand Up @@ -45,7 +45,7 @@ use dynamo_runtime::metrics::{
tokio_perf::{ensure_tokio_perf_metrics_registered_prometheus, tokio_metrics_and_canary_loop},
transport_metrics::ensure_transport_metrics_registered_prometheus,
};
use std::net::SocketAddr;
use std::net::{IpAddr, SocketAddr};
use tokio::sync::Notify;
use tokio::task::JoinHandle;
use tokio_util::sync::CancellationToken;
Expand Down Expand Up @@ -926,9 +926,7 @@ impl HttpService {
Use run()/spawn() (which bind internally) when enable_tls is set."
));
}
let addr: SocketAddr = address
.parse()
.map_err(|e| anyhow::anyhow!("Invalid address '{}': {}", address, e))?;
let addr = listen_socket_addr(&self.host, self.port)?;
let cert_path = self
.tls_cert_path
.as_ref()
Expand Down Expand Up @@ -1007,9 +1005,7 @@ impl HttpService {
let listener = match listener {
Some(l) => l,
None => {
let addr: SocketAddr = address
.parse()
.map_err(|e| anyhow::anyhow!("Invalid address '{}': {}", address, e))?;
let addr = listen_socket_addr(&self.host, self.port)?;
Comment thread
jthomson04 marked this conversation as resolved.
bind_listener(addr).map_err(|e| {
tracing::error!(
protocol = %protocol,
Expand Down Expand Up @@ -1075,11 +1071,11 @@ impl HttpService {
let Some(rl_router) = self.rl_router.clone() else {
return Ok(());
};
let rl_addr = format!("{}:{}", self.host, self.rl_port);
let rl_addr = listen_socket_addr(&self.host, self.rl_port)?;
// Bind eagerly and fail fast: when RL discovery is enabled, a bind failure
// should abort service startup rather than silently leave RL discovery
// unavailable while the main HTTP service keeps running.
let listener = tokio::net::TcpListener::bind(&rl_addr).await.map_err(|e| {
let listener = tokio::net::TcpListener::bind(rl_addr).await.map_err(|e| {
tracing::error!(
address = %rl_addr,
error = %e,
Expand Down Expand Up @@ -1155,6 +1151,20 @@ fn listen_backlog() -> u32 {
parse_listen_backlog(std::env::var(env_llm::DYN_HTTP_LISTEN_BACKLOG))
}

/// Build the listen address from an IP literal host. IPv6 brackets are optional,
/// so `::` and `[::]` are equivalent.
fn listen_socket_addr(host: &str, port: u16) -> Result<SocketAddr> {
let literal = host.trim();
let literal = literal
.strip_prefix('[')
.and_then(|inner| inner.strip_suffix(']'))
.unwrap_or(literal);
let ip: IpAddr = literal
.parse()
.map_err(|e| anyhow::anyhow!("Invalid address '{host}:{port}': {e}"))?;
Ok(SocketAddr::new(ip, port))
}

/// `tokio::net::TcpListener::bind` listens with a backlog of 128. A few thousand
/// clients connecting within seconds overflow that, and overflowed connections
/// are delayed or, depending on host TCP settings, fail.
Expand Down Expand Up @@ -2774,6 +2784,30 @@ mod tests {
assert_eq!(peer, client.local_addr().unwrap());
}

#[test]
fn test_listen_socket_addr_accepts_ip_literals() {
for (host, expected) in [
("0.0.0.0", "0.0.0.0:8000"),
("127.0.0.1", "127.0.0.1:8000"),
("::", "[::]:8000"),
("[::]", "[::]:8000"),
(" [::1] ", "[::1]:8000"),
] {
assert_eq!(
listen_socket_addr(host, 8000).unwrap(),
expected.parse::<SocketAddr>().unwrap(),
"{host}"
);
}
for host in ["localhost", "[::", "::1]", ""] {
let error = listen_socket_addr(host, 8000).unwrap_err().to_string();
assert!(
error.starts_with(&format!("Invalid address '{host}:8000'")),
"{host}: {error}"
);
}
}

#[test]
fn test_sse_keep_alive_for_deferred_response() {
let configured = Duration::from_millis(5000);
Expand Down
Loading
Loading