Skip to content
Merged
Show file tree
Hide file tree
Changes from 3 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
Original file line number Diff line number Diff line change
Expand Up @@ -122,9 +122,9 @@ def add_arguments(self, parser: argparse.ArgumentParser) -> None:
flag_name="--event-plane",
env_var="DYN_EVENT_PLANE",
default=None,
Comment thread
zhongdaor-nv marked this conversation as resolved.
Resolved
help="Determines how events are published. If unset, auto-detected from "
"--discovery-backend: 'zmq' for file/mem (no external services), 'nats' "
"for etcd/kubernetes.",
help="Determines how events are published. If unset, defaults to 'zmq' for "
Comment thread
zhongdaor-nv marked this conversation as resolved.
Resolved
"all discovery backends. Set to 'nats' to use a NATS-based event bus "
"(required for durable/JetStream KV events).",
choices=["nats", "zmq"],
)
add_argument(
Expand Down
4 changes: 2 additions & 2 deletions components/src/dynamo/frontend/frontend_args.py
Original file line number Diff line number Diff line change
Expand Up @@ -365,8 +365,8 @@ def add_arguments(self, parser) -> None:
env_var="DYN_EVENT_PLANE",
default=None,
help="Determines how events are published [nats|zmq]. If unset, "
"auto-detected from --discovery-backend (zmq for file/mem, nats "
"for etcd/kubernetes).",
"defaults to 'zmq' for all discovery backends. Set to 'nats' to use a "
"NATS-based event bus (required for durable/JetStream KV events).",
choices=["nats", "zmq"],
)
add_negatable_bool_argument(
Expand Down
6 changes: 3 additions & 3 deletions docs/components/router/router-guide.md
Original file line number Diff line number Diff line change
Expand Up @@ -155,9 +155,9 @@ When using KV routing, the router needs to know what each worker has cached. The

| Event Mode | How to Enable | Description |
|------------|---------------|-------------|
| **NATS Core (local indexer)** | Router default (no router flag) | Workers maintain a local indexer; configure backend-side KV event publishing so the router can recover state and receive events via NATS Core |
| **JetStream (durable)** | `--router-durable-kv-events` | Events persisted in NATS JetStream; supports snapshots and durable consumers. *Deprecated.* |
| **ZMQ** | `--event-plane zmq` | Workers publish via ZMQ PUB sockets; the standalone `dynamo.indexer` service aggregates events |
| **ZMQ (local indexer)** | Router default (no router flag) | Workers maintain a local indexer and publish KV events via ZMQ PUB sockets; the router recovers state by querying live workers. This is the default event plane for all backends |
| **NATS Core (local indexer)** | `--event-plane nats` (or `DYN_EVENT_PLANE=nats`) | Same local-indexer model, but events flow over NATS Core instead of ZMQ. Opt in when you want a NATS-based event bus |
| **JetStream (durable)** | `--router-durable-kv-events` (requires `--event-plane nats`) | Events persisted in NATS JetStream; supports snapshots and durable consumers. *Deprecated.* |
| **Approximate (no events)** | `--no-router-kv-events` | No events consumed; router predicts cache state from its own routing decisions with TTL-based expiration |

### Aggregated vs. Disaggregated Topology
Expand Down
26 changes: 12 additions & 14 deletions docs/design-docs/event-plane.md
Original file line number Diff line number Diff line change
Expand Up @@ -20,11 +20,11 @@ Key use cases:

The event plane supports two transports:

| | NATS (default) | ZMQ |
| | NATS | ZMQ (default) |
|---|---|---|
| **External infrastructure** | Requires a NATS server | None (peer-to-peer) |
| **Setup complexity** | Simple -- point at a NATS server | Automatic -- workers bind sockets and register via discovery |
| **Best for** | Large-scale deployments | Low operational overhead |
| **Best for** | NATS-based event bus / durable KV events | Default; low operational overhead, no extra services |
Comment thread
zhongdaor-nv marked this conversation as resolved.
Resolved

## Configuration

Expand All @@ -33,11 +33,11 @@ The event plane supports two transports:
Set the `DYN_EVENT_PLANE` environment variable to choose a transport:

```bash
# Use NATS (default -- no need to set explicitly)
export DYN_EVENT_PLANE=nats

# Use ZMQ
# Use ZMQ (default -- no need to set explicitly)
export DYN_EVENT_PLANE=zmq

# Use NATS (opt-in)
export DYN_EVENT_PLANE=nats
```

Python components also accept this as a CLI flag:
Expand All @@ -54,19 +54,17 @@ python3 -m dynamo.vllm --event-plane zmq --model Qwen/Qwen3-0.6B

| Variable | Description | Default |
|----------|-------------|---------|
| `DYN_EVENT_PLANE` | Transport: `nats` or `zmq` | Context-dependent (see below) |
| `DYN_EVENT_PLANE` | Transport: `nats` or `zmq` | `zmq` (all backends) |
| `NATS_SERVER` | NATS server URL (NATS transport only) | `nats://localhost:4222` |

When `DYN_EVENT_PLANE` is not set, the default is chosen based on the discovery backend:

- `--discovery-backend file` or `mem` (local backends): defaults to **zmq** — no external services required.
- `--discovery-backend etcd` or `kubernetes` (distributed backends): defaults to **nats**.

Set `DYN_EVENT_PLANE` explicitly to override this automatic selection.
When `DYN_EVENT_PLANE` is not set, the default is **zmq** for every discovery backend
Comment thread
zhongdaor-nv marked this conversation as resolved.
Resolved
(`file`, `mem`, `etcd`, `kubernetes`) — no external NATS server is required. Set
`DYN_EVENT_PLANE=nats` to opt into the NATS transport (also required for durable/JetStream
Comment thread
zhongdaor-nv marked this conversation as resolved.
Resolved
KV events).

## NATS Transport

When using NATS (`DYN_EVENT_PLANE=nats`, or unset with a distributed backend):
When using NATS (`DYN_EVENT_PLANE=nats`):

- Requires a running NATS server. Set `NATS_SERVER` if it is not on `localhost:4222`.
- Events are published to NATS subjects scoped by namespace and component.
Expand Down
5 changes: 4 additions & 1 deletion lib/llm/src/kv_router/indexer/recovery/subscriber.rs
Original file line number Diff line number Diff line change
Expand Up @@ -129,7 +129,10 @@ pub async fn start_subscriber(
);
if transport_kind != EventTransportKind::Nats {
anyhow::bail!(
"--durable-kv-events requires NATS event plane, but runtime is configured for {transport_kind:?}"
"--durable-kv-events requires the NATS event plane, but the runtime resolved to \
{transport_kind:?} (ZMQ is now the default event plane). Either set \
DYN_EVENT_PLANE=nats to keep using durable/JetStream KV events, or drop \
--durable-kv-events to use the default (non-durable) event-plane subscriber."
);
}
tracing::info!("Using JetStream subscription (--durable-kv-events enabled)");
Expand Down
29 changes: 14 additions & 15 deletions lib/runtime/src/discovery/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -31,47 +31,46 @@ pub use utils::watch_and_extract_field;
#[serde(rename_all = "snake_case")]
pub enum EventTransportKind {
/// NATS Core pub/sub
#[default]
Nats,
/// ZMQ pub/sub
#[default]
Zmq,
}

impl EventTransportKind {
/// Parse from environment variable `DYN_EVENT_PLANE`.
///
/// Returns `Nats` if the variable is not set or is empty, which is the correct
/// default for distributed deployments (etcd/kubernetes backends). For local-only
/// workflows (`--discovery-backend file` or `mem`) this context-unaware default
/// may be incorrect — prefer [`DistributedRuntime::default_event_transport_kind`]
/// when you have access to a runtime, as it derives the correct default from the
/// configured discovery backend.
/// Returns `Zmq` if the variable is not set or is empty: ZMQ is the default
/// event plane for all backends. NATS remains available as an explicit opt-in
/// (`DYN_EVENT_PLANE=nats`). When you have access to a runtime, prefer
/// [`DistributedRuntime::default_event_transport_kind`], which resolves the same
/// default through the configured discovery backend.
///
/// Returns an error for unrecognised values.
pub fn from_env() -> Result<Self> {
match std::env::var(crate::config::environment_names::event_plane::DYN_EVENT_PLANE)
.as_deref()
{
Ok("nats") | Ok("") | Err(_) => Ok(Self::Nats),
Ok("zmq") => Ok(Self::Zmq),
Ok("nats") => Ok(Self::Nats),
Ok("zmq") | Ok("") | Err(_) => Ok(Self::Zmq),
Ok(other) => anyhow::bail!(
"Invalid DYN_EVENT_PLANE value '{}'. Valid values: 'nats', 'zmq'",
other
),
}
}

/// Parse from environment variable, defaulting to NATS when the variable is unset.
/// Parse from environment variable, defaulting to ZMQ when the variable is unset.
///
/// This default is suitable for distributed deployments. For local-only workflows
/// prefer [`DistributedRuntime::default_event_transport_kind`], which automatically
/// selects ZMQ when running with a `file` or `mem` discovery backend.
/// ZMQ is the default event plane for all backends; NATS is an explicit opt-in.
/// When you have access to a runtime, prefer
/// [`DistributedRuntime::default_event_transport_kind`].
///
/// Logs a warning if an invalid value is encountered.
pub fn from_env_or_default() -> Self {
Self::from_env().unwrap_or_else(|e| {
tracing::warn!("{e}, defaulting to NATS");
Self::Nats
tracing::warn!("{e}, defaulting to ZMQ");
Self::Zmq
})
}

Expand Down
39 changes: 12 additions & 27 deletions lib/runtime/src/distributed.rs
Original file line number Diff line number Diff line change
Expand Up @@ -433,12 +433,11 @@ impl DistributedRuntime {
/// Returns the event transport kind this runtime was configured with.
///
/// The value is resolved once at construction time by `DiscoveryBackend::resolve_event_transport_kind`:
/// if `DYN_EVENT_PLANE` is set explicitly that value wins; otherwise the discovery
/// backend drives the default (ZMQ for `file`/`mem`, NATS for `etcd`/`kubernetes`).
/// if `DYN_EVENT_PLANE` is set explicitly that value wins; otherwise the default is
/// ZMQ for every discovery backend (NATS is an explicit opt-in).
///
/// Use this instead of [`EventTransportKind::from_env_or_default`] wherever you have
/// access to a `DistributedRuntime`, so that local-only workflows work without
/// setting `DYN_EVENT_PLANE` explicitly.
/// access to a `DistributedRuntime`.
pub fn default_event_transport_kind(&self) -> crate::discovery::EventTransportKind {
self.event_transport_kind
}
Expand Down Expand Up @@ -611,8 +610,6 @@ impl DiscoveryBackend {
/// Returns true if this backend requires no external services (file or in-memory).
///
/// Local backends do not need etcd, NATS, or any other infrastructure daemon.
/// This is used to drive smart defaults: for example, the event plane defaults to
/// ZMQ (not NATS) when a local backend is in use and `DYN_EVENT_PLANE` is not set.
pub fn is_local(&self) -> bool {
matches!(
self,
Expand All @@ -623,10 +620,10 @@ impl DiscoveryBackend {

/// Resolve the event transport kind for this backend.
///
/// This is the single authoritative mapping of `(DYN_EVENT_PLANE, backend)` →
/// `EventTransportKind`. When `DYN_EVENT_PLANE` is unset or empty the backend
/// drives the default: local backends (`file`/`mem`) → ZMQ, distributed backends
/// (`etcd`/`kubernetes`) → NATS.
/// This is the single authoritative mapping of `DYN_EVENT_PLANE` →
/// `EventTransportKind`. ZMQ is the default event plane for all backends
/// (`file`/`mem`/`etcd`/`kubernetes`); NATS is an explicit opt-in via
/// `DYN_EVENT_PLANE=nats`.
///
/// Call this once at startup and store the result; do not call it repeatedly.
pub fn resolve_event_transport_kind(&self) -> crate::discovery::EventTransportKind {
Comment thread
zhongdaor-nv marked this conversation as resolved.
Resolved
Expand All @@ -635,27 +632,15 @@ impl DiscoveryBackend {
match std::env::var(DYN_EVENT_PLANE).as_deref() {
Ok("nats") => EventTransportKind::Nats,
Ok("zmq") => EventTransportKind::Zmq,
// Unset or empty: derive from backend type.
Ok("") | Err(_) => {
if self.is_local() {
EventTransportKind::Zmq
} else {
EventTransportKind::Nats
}
}
// Unset or empty: ZMQ is the default for every backend.
Ok("") | Err(_) => EventTransportKind::Zmq,
Ok(other) => {
let default_kind = if self.is_local() {
EventTransportKind::Zmq
} else {
EventTransportKind::Nats
};
tracing::warn!(
Comment thread
zhongdaor-nv marked this conversation as resolved.
Resolved
"Invalid DYN_EVENT_PLANE value '{}'. Valid values: 'nats', 'zmq'. \
Defaulting to {:?}.",
other,
default_kind
Defaulting to Zmq.",
Comment thread
zhongdaor-nv marked this conversation as resolved.
Resolved
other
);
default_kind
EventTransportKind::Zmq
}
}
}
Expand Down
53 changes: 28 additions & 25 deletions tests/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -1145,7 +1145,7 @@ def runtime_services(request, discovery_backend, request_plane):

@pytest.fixture()
def runtime_services_dynamic_ports(
request, discovery_backend, request_plane, durable_kv_events
request, discovery_backend, request_plane, durable_kv_events, monkeypatch
):
"""Provide NATS and Etcd servers with truly dynamic ports per test.

Expand All @@ -1159,41 +1159,44 @@ def runtime_services_dynamic_ports(
leak across workers.

- If discovery_backend != "etcd", etcd is not started (returns None)
- NATS is always started when etcd is used, because KV events require NATS
regardless of the request_plane (tcp/nats only affects request transport)
- NATS Core mode (no JetStream) is the default; JetStream is enabled when durable_kv_events=True
- When etcd is used this fixture pins DYN_EVENT_PLANE=nats, reproducing the
pre-existing per-backend default (etcd/kubernetes -> NATS). The product
default is now ZMQ for all backends, but the in-process e2e harness delivers
Comment thread
zhongdaor-nv marked this conversation as resolved.
Resolved
KV events between a separate engine subprocess and an in-process router over
a naive direct ZMQ PUB/SUB path with no broker; that path suffers the
slow-joiner problem (events published before the SUB connects are dropped),
so the router would observe zero KV events. NATS Core buffers and supports
request/reply, so these tests rely on it. Both the in-process router
(get_runtime(event_plane=None)) and engine subprocesses (env =
os.environ.copy()) inherit this; explicit event_plane="zmq" overrides at the
call site still win. file/mem backends are left on the ZMQ default.
- NATS Core mode (no JetStream) is the default; JetStream is enabled when durable_kv_events=True.

Returns a tuple of (nats_process, etcd_process) where each has a .port attribute.
"""
import os

# Port cleanup is now handled in NatsServer and EtcdServer __exit__ methods
# Always start NATS when etcd is used - KV events require NATS regardless of request_plane
# When durable_kv_events=False (default), disable JetStream for faster startup
# Start NATS when etcd is used so the NATS opt-in paths stay available.
# When durable_kv_events=False (default), disable JetStream for faster startup.
if discovery_backend == "etcd":
with NatsServer(
request, port=0, disable_jetstream=not durable_kv_events
) as nats_process:
with EtcdServer(request, port=0) as etcd_process:
# Save original env vars (may be set by session-scoped fixture)
orig_nats = os.environ.get("NATS_SERVER")
orig_etcd = os.environ.get("ETCD_ENDPOINTS")

# Set environment variables for this test's dynamic ports
os.environ["NATS_SERVER"] = f"nats://localhost:{nats_process.port}"
os.environ["ETCD_ENDPOINTS"] = f"http://localhost:{etcd_process.port}"
# Set environment variables for this test's dynamic ports.
# monkeypatch.setenv auto-restores them (including any values set by
# the session-scoped fixture) at test teardown.
monkeypatch.setenv(
"NATS_SERVER", f"nats://localhost:{nats_process.port}"
)
monkeypatch.setenv(
"ETCD_ENDPOINTS", f"http://localhost:{etcd_process.port}"
)
# Pin the NATS event plane for etcd-backed tests: the in-process e2e
# harness cannot reliably deliver KV events over the (now-default) ZMQ
# plane without a broker (slow-joiner loss). See the fixture docstring.
monkeypatch.setenv("DYN_EVENT_PLANE", "nats")

yield nats_process, etcd_process

# Restore original env vars (or remove if they weren't set)
if orig_nats is not None:
os.environ["NATS_SERVER"] = orig_nats
else:
os.environ.pop("NATS_SERVER", None)
if orig_etcd is not None:
os.environ["ETCD_ENDPOINTS"] = orig_etcd
else:
os.environ.pop("ETCD_ENDPOINTS", None)
elif request_plane == "nats":
with NatsServer(
request, port=0, disable_jetstream=not durable_kv_events
Expand Down
Loading