diff --git a/deploy/inference-gateway/ext-proc/Cargo.toml b/deploy/inference-gateway/ext-proc/Cargo.toml index daf82d9aed7b..a0532ea73636 100644 --- a/deploy/inference-gateway/ext-proc/Cargo.toml +++ b/deploy/inference-gateway/ext-proc/Cargo.toml @@ -22,7 +22,7 @@ dynamo-kv-router = { workspace = true, features = [ "standalone-selection", ] } dynamo-protocols = { workspace = true } -dynamo-runtime = { workspace = true, features = ["etcd"] } +dynamo-runtime = { workspace = true, features = ["etcd", "nats"] } anyhow = { workspace = true } axum = { workspace = true } diff --git a/docs/fern/pages/developer-guide/knowledge-base/concepts/system-architecture/sidecar-backends.md b/docs/fern/pages/developer-guide/knowledge-base/concepts/system-architecture/sidecar-backends.md index 5d94c0e799a7..4a2ac9f4a9ec 100644 --- a/docs/fern/pages/developer-guide/knowledge-base/concepts/system-architecture/sidecar-backends.md +++ b/docs/fern/pages/developer-guide/knowledge-base/concepts/system-architecture/sidecar-backends.md @@ -138,11 +138,13 @@ succeeded. Across Kubernetes pods, use `DYN_DISCOVERY_BACKEND=kubernetes`, or the selection-catalog registration above, in place of the file backend. At build time, `dynamo-runtime` gates the etcd transport and discovery backend -behind the `etcd` Cargo feature (on by default). A binary built with -`--no-default-features` does not link `etcd-client`; `DYN_DISCOVERY_BACKEND` -then defaults to `mem`, and `etcd` is rejected with an explicit error. The -file, memory, and Kubernetes backends, the TCP request plane, and the ZMQ event -plane are always compiled in. +behind the `etcd` Cargo feature and the NATS request plane, NATS event plane, +and NATS service registration behind the `nats` feature (both on by default). +A binary built with `--no-default-features` links neither `etcd-client` nor +`async-nats`; `DYN_DISCOVERY_BACKEND` then defaults to `mem`, and selecting +`etcd` or a NATS plane is rejected with an explicit error naming the feature. +The file, memory, and Kubernetes discovery backends, the TCP request plane, and +the ZMQ event plane are always compiled in. ## Running Without A Distributed Runtime (Experimental) diff --git a/lib/backend-common/Cargo.toml b/lib/backend-common/Cargo.toml index 8bc40c1f087f..890efbd23c39 100644 --- a/lib/backend-common/Cargo.toml +++ b/lib/backend-common/Cargo.toml @@ -22,7 +22,7 @@ testing = [] integration = ["dynamo-runtime/integration"] [dependencies] -dynamo-runtime = { workspace = true, features = ["etcd"] } +dynamo-runtime = { workspace = true, features = ["etcd", "nats"] } dynamo-llm = { path = "../llm", default-features = false } dynamo-protocols = { workspace = true } dynamo-rl = { workspace = true } diff --git a/lib/backend-common/examples/mocker/Cargo.toml b/lib/backend-common/examples/mocker/Cargo.toml index 8a287f224667..9a0a1eba4879 100644 --- a/lib/backend-common/examples/mocker/Cargo.toml +++ b/lib/backend-common/examples/mocker/Cargo.toml @@ -34,4 +34,4 @@ uuid = { workspace = true } [dev-dependencies] dynamo-backend-common = { workspace = true, features = ["testing"] } -dynamo-runtime = { workspace = true, features = ["etcd"] } +dynamo-runtime = { workspace = true, features = ["etcd", "nats"] } diff --git a/lib/llm/Cargo.toml b/lib/llm/Cargo.toml index c73dc3287ef8..fd091f9d8cb8 100644 --- a/lib/llm/Cargo.toml +++ b/lib/llm/Cargo.toml @@ -12,7 +12,7 @@ repository.workspace = true description = "Dynamo LLM Library" [features] -default = ["block-manager", "etcd"] +default = ["block-manager", "etcd", "nats"] # todo(ops): get this working in CI as a default. # default = ["block-manager", "testing-full"] @@ -25,6 +25,9 @@ nvtx = ["dynamo-runtime/nvtx"] # etcd discovery/transport in dynamo-runtime. Off (`--no-default-features`) leaves # the file, memory, and Kubernetes discovery backends and does not link etcd-client. etcd = ["dynamo-runtime/etcd"] +# NATS request/event planes in dynamo-runtime. Off leaves the TCP request plane and +# ZMQ event plane and does not link async-nats through the runtime. +nats = ["dynamo-runtime/nats", "dep:async-nats"] block-manager = ["dep:nixl-sys", "dep:cudarc", "dep:nix", "dep:aligned-vec"] nccl = ["dep:cudarc", "cudarc/nccl"] # Enable NCCL collective operations block-manager-bench = ["block-manager", "testing-full", "dep:clap", "dep:indicatif"] @@ -92,7 +95,7 @@ dynamo-parsers = { workspace = true } dynamo-parsers-v2 = { workspace = true } async-stream = { workspace = true } async-trait = { workspace = true } -async-nats = { workspace = true } +async-nats = { workspace = true, optional = true } bytes = { workspace = true } chrono = { workspace = true } dashmap = { workspace = true } @@ -230,7 +233,8 @@ llm-multimodal = { version = "=1.9.0", optional = true } llm-tokenizer = { version = "=1.6.0", optional = true } [dev-dependencies] -dynamo-runtime = { workspace = true, features = ["testing", "etcd"] } +async-nats = { workspace = true } +dynamo-runtime = { workspace = true, features = ["testing", "etcd", "nats"] } tokio = { workspace = true, features = ["test-util"] } approx = "0.5" assert_matches = "1.5" diff --git a/lib/llm/src/request_trace/sink.rs b/lib/llm/src/request_trace/sink.rs index 076ec082447f..b311bba54fcf 100644 --- a/lib/llm/src/request_trace/sink.rs +++ b/lib/llm/src/request_trace/sink.rs @@ -9,9 +9,11 @@ use std::sync::{ use std::time::Duration; use anyhow::{Context as _, anyhow}; +#[cfg(feature = "nats")] use async_nats::jetstream; use async_trait::async_trait; use dynamo_runtime::config::environment_names::llm::request_trace as env_request_trace; +#[cfg(feature = "nats")] use dynamo_runtime::transports::nats; use tokio::sync::broadcast; use tokio_util::sync::CancellationToken; @@ -53,11 +55,13 @@ impl RequestTraceSink for StderrRequestTraceSink { } } +#[cfg(feature = "nats")] pub struct NatsRequestTraceSink { js: jetstream::Context, subject: String, } +#[cfg(feature = "nats")] impl NatsRequestTraceSink { async fn from_policy(policy: &RequestTracePolicy) -> anyhow::Result { let nats_client = nats::ClientOptions::default() @@ -76,6 +80,7 @@ impl NatsRequestTraceSink { } } +#[cfg(feature = "nats")] #[async_trait] impl RequestTraceSink for NatsRequestTraceSink { fn name(&self) -> &'static str { @@ -191,6 +196,13 @@ async fn parse_sinks_from_env() -> anyhow::Result> for sink_kind in &policy.sinks { match sink_kind { RequestTraceSinkKind::Stderr => sinks.push(Arc::new(StderrRequestTraceSink)), + #[cfg(not(feature = "nats"))] + RequestTraceSinkKind::Nats => { + return Err(anyhow!( + "the NATS request-trace sink is not compiled into this build (dynamo-llm feature `nats`)" + )); + } + #[cfg(feature = "nats")] RequestTraceSinkKind::Nats => { sinks.push(Arc::new(NatsRequestTraceSink::from_policy(policy).await?)) } diff --git a/lib/mocker/servers/vllm/Cargo.toml b/lib/mocker/servers/vllm/Cargo.toml index 6ecff541ec2e..a0ad3236a028 100644 --- a/lib/mocker/servers/vllm/Cargo.toml +++ b/lib/mocker/servers/vllm/Cargo.toml @@ -40,7 +40,7 @@ dynamo-kv-router = { workspace = true, features = ["standalone-selection"] } dynamo-llm = { workspace = true } tempfile = { workspace = true } tokio-util = { workspace = true } -dynamo-runtime = { workspace = true, features = ["etcd"] } +dynamo-runtime = { workspace = true, features = ["etcd", "nats"] } dynamo-sidecar-common = { workspace = true } serde_json = { workspace = true } tokio-stream = { workspace = true, features = ["net"] } diff --git a/lib/runtime/Cargo.toml b/lib/runtime/Cargo.toml index aaf042cfdf04..7931aef201bf 100644 --- a/lib/runtime/Cargo.toml +++ b/lib/runtime/Cargo.toml @@ -15,10 +15,13 @@ description = "Dynamo Runtime Library" ignored = ["mio"] [features] -default = ["etcd"] +default = ["etcd", "nats"] # etcd discovery backend (`DYN_DISCOVERY_BACKEND=etcd`) and the etcd transport. # Off: the file, memory, and Kubernetes backends remain; `etcd-client` is not linked. etcd = ["dep:etcd-client"] +# NATS request plane, NATS event plane, and NATS service registration. Off: the +# TCP request plane and ZMQ event plane remain; `async-nats` is not linked. +nats = ["dep:async-nats"] integration = [] testing-etcd = ["etcd"] # Tests that require an active ETCD server # White-box helpers for cross-crate lifecycle tests. Never enabled by a normal build. @@ -36,7 +39,7 @@ dynamo-truthy = { workspace = true } # Use workspace dependencies where available anyhow = { workspace = true } -async-nats = { workspace = true } +async-nats = { workspace = true, optional = true } async-stream = { workspace = true } async-trait = { workspace = true } tmq = { workspace = true } @@ -121,6 +124,7 @@ k8s-openapi = { version = "0.26.0", features = ["v1_32"] } [dev-dependencies] criterion = { version = "0.5", features = ["async_tokio"] } +tokio = { workspace = true, features = ["test-util"] } rcgen = { workspace = true } rstest = { version = "0.23.0" } temp-env = { version = "0.3.6" , features=["async_closure"] } diff --git a/lib/runtime/src/component.rs b/lib/runtime/src/component.rs index 47a5bad95ee4..7f8c1ace002d 100644 --- a/lib/runtime/src/component.rs +++ b/lib/runtime/src/component.rs @@ -35,20 +35,16 @@ use crate::{ config::HealthStatus, distributed::RequestPlaneMode, metrics::{MetricsHierarchy, MetricsRegistry, prometheus_names}, - service::ServiceClient, - service::ServiceSet, }; -use super::{DistributedRuntime, Runtime, traits::*, transports::nats::Slug, utils::Duration}; +use super::{DistributedRuntime, Runtime, slug::Slug, traits::*, utils::Duration}; use crate::pipeline::network::{ PushWorkHandler, RequestPlanePayloadCodec, ingress::push_endpoint::PushEndpoint, }; use crate::protocols::EndpointId; -use async_nats::{ - rustls::quic, - service::{Service, ServiceExt}, -}; +#[cfg(feature = "nats")] +use async_nats::service::Service; use dashmap::DashMap; use derive_builder::Builder; use derive_getters::Getters; @@ -63,6 +59,7 @@ mod component; mod endpoint; mod namespace; mod registry; +#[cfg(feature = "nats")] pub mod service; pub(crate) use client::EndpointDiscoverySource; @@ -95,6 +92,7 @@ pub enum DeviceType { #[derive(Default)] pub struct RegistryInner { + #[cfg(feature = "nats")] pub(crate) services: HashMap, } @@ -322,6 +320,13 @@ impl ComponentBuilder { // This prevents a race condition where serve_endpoint() tries to look up the service // before it's registered in the component registry. let drt = component.drt(); + #[cfg(not(feature = "nats"))] + if drt.request_plane().is_nats() { + return Err(anyhow::anyhow!( + "NATS request plane is not compiled into this build (dynamo-runtime feature `nats`)" + )); + } + #[cfg(feature = "nats")] if drt.request_plane().is_nats() { let mut rx = drt.register_nats_service(component.clone()); // Wait synchronously for the NATS service registration to complete. diff --git a/lib/runtime/src/component/endpoint.rs b/lib/runtime/src/component/endpoint.rs index d9e9f114ed4e..9c966fbfdf30 100644 --- a/lib/runtime/src/component/endpoint.rs +++ b/lib/runtime/src/component/endpoint.rs @@ -9,6 +9,8 @@ use derive_getters::Dissolve; use educe::Educe; use tokio_util::sync::CancellationToken; +#[cfg(feature = "nats")] +use crate::transports::nats; use crate::{ component::{DeviceType, Endpoint, Instance, TransportType}, distributed::RequestPlaneMode, @@ -17,7 +19,6 @@ use crate::{ }, protocols::EndpointId, traits::DistributedRuntimeProvider, - transports::nats, }; fn endpoint_device_type() -> Option { @@ -345,6 +346,11 @@ fn build_transport_type_inner( Ok(TransportType::Tcp(tcp_endpoint)) } + #[cfg(not(feature = "nats"))] + RequestPlaneMode::Nats => Err(anyhow::anyhow!( + "NATS request plane is not compiled into this build (dynamo-runtime feature `nats`)" + )), + #[cfg(feature = "nats")] RequestPlaneMode::Nats => Ok(TransportType::Nats(nats::instance_subject( endpoint_id, connection_id, diff --git a/lib/runtime/src/distributed.rs b/lib/runtime/src/distributed.rs index abd7a7f3cd72..12a365f6a4ba 100644 --- a/lib/runtime/src/distributed.rs +++ b/lib/runtime/src/distributed.rs @@ -7,6 +7,7 @@ use crate::component::{ use crate::config::environment_names::tcp_response_stream; use crate::pipeline::PipelineError; use crate::pipeline::network::manager::NetworkManager; +#[cfg(feature = "nats")] use crate::service::{ServiceClient, ServiceSet}; use crate::storage::kv; use crate::{discovery, system_status_server, transports}; @@ -14,7 +15,7 @@ use crate::{ discovery::{Discovery, DiscoverySpec, EndpointRegistrationLease, EndpointRegistrationManager}, metrics::PrometheusUpdateCallback, metrics::{MetricsHierarchy, MetricsRegistry}, - transports::{nats, tcp}, + transports::tcp, }; use super::utils::GracefulShutdownTracker; @@ -54,6 +55,7 @@ pub struct DistributedRuntime { // local runtime runtime: Runtime, + #[cfg(feature = "nats")] nats_client: Option, network_manager: Arc, tcp_server: Arc>>, @@ -127,10 +129,18 @@ impl DistributedRuntime { let (discovery_backend, nats_config, request_plane, event_transport_kind) = config.dissolve(); + #[cfg(feature = "nats")] let nats_client = match nats_config { Some(nc) => Some(nc.connect().await?), None => None, }; + #[cfg(feature = "nats")] + let network_nats_client = nats_client.clone().map(|c| c.client().clone()); + #[cfg(not(feature = "nats"))] + let network_nats_client: Option = { + let _: Option = nats_config; + None + }; // Start system status server for health and metrics if enabled in configuration let config = crate::config::RuntimeConfig::from_settings().unwrap_or_default(); @@ -196,7 +206,7 @@ impl DistributedRuntime { // NetworkManager for request plane let network_manager = NetworkManager::new( runtime.child_token(), - nats_client.clone().map(|c| c.client().clone()), + network_nats_client, component_registry.clone(), request_plane, ); @@ -209,6 +219,7 @@ impl DistributedRuntime { let distributed_runtime = Self { runtime, network_manager: Arc::new(network_manager), + #[cfg(feature = "nats")] nats_client, tcp_server: Arc::new(OnceCell::new()), system_status_server: Arc::new(OnceLock::new()), @@ -492,6 +503,7 @@ impl DistributedRuntime { self.routing_occupancy_states.clone() } + #[cfg(feature = "nats")] /// TODO: This is a temporary KV router measure for component/component.rs EventPublisher impl for /// Component, to allow it to publish to NATS. KV Router is the only user. /// @@ -506,6 +518,7 @@ impl DistributedRuntime { .await } + #[cfg(feature = "nats")] pub(crate) async fn kv_router_nats_publish_subject( &self, subject: async_nats::Subject, @@ -519,6 +532,7 @@ impl DistributedRuntime { Ok(nats_client.client().publish(subject, payload).await?) } + #[cfg(feature = "nats")] /// TODO: This is a temporary KV router measure for component/component.rs EventSubscriber impl for /// Component, to allow it to subscribe to NATS. KV Router is the only user. pub(crate) async fn kv_router_nats_subscribe( @@ -531,6 +545,7 @@ impl DistributedRuntime { Ok(nats_client.client().subscribe(subject).await?) } + #[cfg(feature = "nats")] /// TODO (karenc): This is a temporary KV router measure for worker query requests. /// Allows KV Router to perform request/reply with workers. (versus the pub/sub pattern above) /// KV Router is the only user, made public for use in dynamo-llm crate @@ -550,6 +565,7 @@ impl DistributedRuntime { Ok(response) } + #[cfg(feature = "nats")] /// DEPRECATED: This method exists only for NATS request plane support. /// Once everything uses the TCP request plane, this can be removed along with /// the NATS service registration infrastructure. @@ -683,7 +699,7 @@ impl DiscoveryBackend { #[derive(Dissolve)] pub struct DistributedConfig { pub discovery_backend: DiscoveryBackend, - pub nats_config: Option, + pub nats_config: Option, pub request_plane: RequestPlaneMode, /// Resolved event transport kind — computed once at config time from /// `DYN_EVENT_PLANE` and the discovery backend, then stored on the runtime @@ -697,6 +713,25 @@ const DEFAULT_DISCOVERY_BACKEND: &str = "etcd"; #[cfg(not(feature = "etcd"))] const DEFAULT_DISCOVERY_BACKEND: &str = "mem"; +/// NATS connection options when NATS is required by the configuration. +/// Without the `nats` feature a configuration that needs NATS is a startup +/// error, matching the unknown-backend handling above. +pub(crate) fn nats_config_if(nats_enabled: bool) -> Option { + if !nats_enabled { + return None; + } + #[cfg(feature = "nats")] + { + Some(transports::nats::ClientOptions::default()) + } + #[cfg(not(feature = "nats"))] + { + panic!( + "NATS is required by DYN_REQUEST_PLANE/DYN_EVENT_PLANE/NATS_SERVER but not compiled into this build (dynamo-runtime feature `nats`)" + ) + } +} + impl DistributedConfig { pub fn from_settings() -> DistributedConfig { let request_plane = RequestPlaneMode::from_env(); @@ -745,11 +780,7 @@ impl DistributedConfig { DistributedConfig { discovery_backend, - nats_config: if nats_enabled { - Some(nats::ClientOptions::default()) - } else { - None - }, + nats_config: crate::distributed::nats_config_if(nats_enabled), request_plane, event_transport_kind, } @@ -776,11 +807,7 @@ impl DistributedConfig { ); DistributedConfig { discovery_backend, - nats_config: if nats_enabled { - Some(nats::ClientOptions::default()) - } else { - None - }, + nats_config: crate::distributed::nats_config_if(nats_enabled), request_plane, event_transport_kind, } @@ -861,14 +888,12 @@ pub mod distributed_test_utils { /// Note: Settings are read from environment variables inside DistributedRuntime::from_settings #[cfg(feature = "integration")] pub async fn create_test_drt_async() -> super::DistributedRuntime { - use crate::transports::nats; - let rt = crate::Runtime::from_current().unwrap(); let config = super::DistributedConfig { discovery_backend: super::DiscoveryBackend::KvStore( crate::storage::kv::Selector::Memory, ), - nats_config: Some(nats::ClientOptions::default()), + nats_config: crate::distributed::nats_config_if(true), request_plane: crate::distributed::RequestPlaneMode::default(), event_transport_kind: crate::discovery::EventTransportKind::Nats, }; @@ -884,14 +909,12 @@ pub mod distributed_test_utils { pub async fn create_test_shared_drt_async( store_path: &std::path::Path, ) -> super::DistributedRuntime { - use crate::transports::nats; - let rt = crate::Runtime::from_current().unwrap(); let config = super::DistributedConfig { discovery_backend: super::DiscoveryBackend::KvStore( crate::storage::kv::Selector::File(store_path.to_path_buf()), ), - nats_config: Some(nats::ClientOptions::default()), + nats_config: crate::distributed::nats_config_if(true), request_plane: crate::distributed::RequestPlaneMode::default(), event_transport_kind: crate::discovery::EventTransportKind::Nats, }; diff --git a/lib/runtime/src/lib.rs b/lib/runtime/src/lib.rs index ad66cf1a8d55..536af9e36ffc 100644 --- a/lib/runtime/src/lib.rs +++ b/lib/runtime/src/lib.rs @@ -42,6 +42,7 @@ pub mod protocols; mod routing_policy; pub mod runnable; pub mod runtime; +#[cfg(feature = "nats")] pub mod service; pub mod slug; pub mod storage; diff --git a/lib/runtime/src/logging.rs b/lib/runtime/src/logging.rs index c691763ac7e8..55978a597107 100644 --- a/lib/runtime/src/logging.rs +++ b/lib/runtime/src/logging.rs @@ -54,6 +54,7 @@ use crate::config::{ ConsoleLogFormat, console_log_format, disable_ansi_logging, env_is_truthy, legacy_jsonl_logging_enabled, span_events_enabled, }; +#[cfg(feature = "nats")] use async_nats::{HeaderMap, HeaderValue}; use axum::extract::FromRequestParts; use axum::http; @@ -487,6 +488,7 @@ impl GenericHeaders for TraceparentHeader<'_> { } } +#[cfg(feature = "nats")] impl GenericHeaders for async_nats::HeaderMap { fn get(&self, key: &str) -> Option<&str> { async_nats::HeaderMap::get(self, key).map(|value| value.as_str()) @@ -678,6 +680,7 @@ pub fn make_system_request_span(req: &Request) -> Span { } /// Create a handle_payload span from NATS headers with component context +#[cfg(feature = "nats")] pub fn make_handle_payload_span( headers: &async_nats::HeaderMap, component: &str, @@ -779,6 +782,7 @@ pub fn make_handle_payload_span_from_tcp_headers( } /// Extract OpenTelemetry trace context from NATS headers for distributed tracing +#[cfg(feature = "nats")] pub fn extract_otel_context_from_nats_headers( headers: &async_nats::HeaderMap, ) -> ( @@ -791,6 +795,7 @@ pub fn extract_otel_context_from_nats_headers( } /// Inject OpenTelemetry trace context into NATS headers using W3C Trace Context propagation +#[cfg(feature = "nats")] pub fn inject_otel_context_into_nats_headers( headers: &mut async_nats::HeaderMap, context: Option, @@ -810,6 +815,7 @@ pub fn inject_otel_context_into_nats_headers( } /// Inject trace context from current span into NATS headers +#[cfg(feature = "nats")] pub fn inject_current_trace_into_nats_headers(headers: &mut async_nats::HeaderMap) { inject_otel_context_into_nats_headers(headers, None); } @@ -841,16 +847,13 @@ pub fn inject_trace_headers_into_map(headers: &mut std::collections::HashMap Option { - let mut headers = async_nats::HeaderMap::new(); - headers.insert("traceparent", ctx.create_traceparent()); - + // Plain string headers: the extraction is transport-agnostic. + let mut headers = std::collections::HashMap::new(); + headers.insert("traceparent".to_string(), ctx.create_traceparent()); if let Some(ref tracestate) = ctx.tracestate { - headers.insert("tracestate", tracestate.as_str()); + headers.insert("tracestate".to_string(), tracestate.clone()); } - - let (otel_context, _trace_id, _parent_span_id) = - extract_otel_context_from_nats_headers(&headers); - otel_context + extract_trace_parent(&headers).1 } /// Create a client_request span linked to the parent trace context @@ -2384,10 +2387,10 @@ pub mod tests { #[test] fn trace_parent_from_headers_preserves_unsampled_flag() { - let mut headers = async_nats::HeaderMap::new(); + let mut headers = std::collections::HashMap::new(); headers.insert( - "traceparent", - "00-11111111111111111111111111111111-2222222222222222-00", + "traceparent".to_string(), + "00-11111111111111111111111111111111-2222222222222222-00".to_string(), ); let trace_parent = TraceParent::from_headers(&headers); diff --git a/lib/runtime/src/pipeline/error.rs b/lib/runtime/src/pipeline/error.rs index 7d7fa3c9f1b1..fc161a02aefa 100644 --- a/lib/runtime/src/pipeline/error.rs +++ b/lib/runtime/src/pipeline/error.rs @@ -3,6 +3,8 @@ //! Pipeline Error // +#[cfg(feature = "nats")] +#[cfg(feature = "nats")] use async_nats::error::Error as NatsError; pub use anyhow::{Context, Error, Result, anyhow, anyhow as error, bail, ensure}; @@ -78,27 +80,35 @@ pub enum PipelineError { #[error("An endpoint URL must have the format: namespace/component/endpoint")] InvalidEndpointFormat, + #[cfg(feature = "nats")] #[error("NATS Request Error: {0}")] NatsRequestError(#[from] NatsError), + #[cfg(feature = "nats")] #[error("NATS Get Stream Error: {0}")] NatsGetStreamError(#[from] NatsError), + #[cfg(feature = "nats")] #[error("NATS Create Stream Error: {0}")] NatsCreateStreamError(#[from] NatsError), + #[cfg(feature = "nats")] #[error("NATS Consumer Error: {0}")] NatsConsumerError(#[from] NatsError), + #[cfg(feature = "nats")] #[error("NATS Batch Error: {0}")] NatsBatchError(#[from] NatsError), + #[cfg(feature = "nats")] #[error("NATS Publish Error: {0}")] NatsPublishError(#[from] NatsError), + #[cfg(feature = "nats")] #[error("NATS Connect Error: {0}")] NatsConnectError(#[from] NatsError), + #[cfg(feature = "nats")] #[error("NATS Subscriber Error: {0}")] NatsSubscriberError(#[from] async_nats::SubscribeError), diff --git a/lib/runtime/src/pipeline/network/egress.rs b/lib/runtime/src/pipeline/network/egress.rs index 7c3cfc50455f..62a734605b43 100644 --- a/lib/runtime/src/pipeline/network/egress.rs +++ b/lib/runtime/src/pipeline/network/egress.rs @@ -2,6 +2,7 @@ // SPDX-License-Identifier: Apache-2.0 pub mod addressed_router; +#[cfg(feature = "nats")] pub mod nats_client; pub mod push_router; pub mod route_span; diff --git a/lib/runtime/src/pipeline/network/ingress.rs b/lib/runtime/src/pipeline/network/ingress.rs index 906e4b4a8481..f488aaf33077 100644 --- a/lib/runtime/src/pipeline/network/ingress.rs +++ b/lib/runtime/src/pipeline/network/ingress.rs @@ -1,6 +1,7 @@ // SPDX-FileCopyrightText: Copyright (c) 2024-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. // SPDX-License-Identifier: Apache-2.0 +#[cfg(feature = "nats")] pub mod nats_server; pub mod push_endpoint; pub mod push_handler; diff --git a/lib/runtime/src/pipeline/network/ingress/push_endpoint.rs b/lib/runtime/src/pipeline/network/ingress/push_endpoint.rs index f7d71c110dae..a4ee5e871647 100644 --- a/lib/runtime/src/pipeline/network/ingress/push_endpoint.rs +++ b/lib/runtime/src/pipeline/network/ingress/push_endpoint.rs @@ -6,9 +6,11 @@ use std::sync::atomic::{AtomicU64, Ordering}; use super::*; use crate::SystemHealth; use crate::config::HealthStatus; +#[cfg(feature = "nats")] use crate::logging::make_handle_payload_span; use crate::protocols::LeaseId; use anyhow::Result; +#[cfg(feature = "nats")] use async_nats::service::endpoint::Endpoint; use derive_builder::Builder; use parking_lot::Mutex; @@ -33,6 +35,7 @@ impl PushEndpoint { PushEndpointBuilder::default() } + #[cfg(feature = "nats")] pub async fn start( self, endpoint: Endpoint, diff --git a/lib/runtime/src/pipeline/network/manager.rs b/lib/runtime/src/pipeline/network/manager.rs index af55dc3f2280..e16aecde76da 100644 --- a/lib/runtime/src/pipeline/network/manager.rs +++ b/lib/runtime/src/pipeline/network/manager.rs @@ -81,14 +81,14 @@ struct NetworkConfig { tcp_client_config: super::egress::tcp_client::TcpRequestConfig, // NATS configuration (provided externally, not from env) - nats_client: Option, + nats_client: Option, } impl NetworkConfig { /// Load configuration from environment variables /// /// This is the ONLY place where network-related environment variables are read. - fn from_env(nats_client: Option) -> Self { + fn from_env(nats_client: Option) -> Self { Self { // TCP server configuration // If DYN_TCP_RPC_PORT is set, use that port; otherwise None means OS will assign a free port @@ -161,7 +161,7 @@ impl NetworkManager { /// Returns an Arc-wrapped NetworkManager ready to create servers and clients. pub fn new( cancellation_token: CancellationToken, - nats_client: Option, + nats_client: Option, component_registry: crate::component::Registry, mode: RequestPlaneMode, ) -> Self { @@ -298,6 +298,14 @@ impl NetworkManager { Ok(server.clone() as Arc) } + #[cfg(not(feature = "nats"))] + async fn create_nats_server(&self) -> Result> { + anyhow::bail!( + "NATS request plane is not compiled into this build (dynamo-runtime feature `nats`); use DYN_REQUEST_PLANE=tcp" + ) + } + + #[cfg(feature = "nats")] async fn create_nats_server(&self) -> Result> { use super::ingress::nats_server::NatsMultiplexedServer; @@ -329,6 +337,14 @@ impl NetworkManager { )?)) } + #[cfg(not(feature = "nats"))] + fn create_nats_client(&self) -> Result> { + anyhow::bail!( + "NATS request plane is not compiled into this build (dynamo-runtime feature `nats`); use DYN_REQUEST_PLANE=tcp" + ) + } + + #[cfg(feature = "nats")] fn create_nats_client(&self) -> Result> { use super::egress::nats_client::NatsRequestClient; @@ -369,7 +385,13 @@ mod tests { "expected NATS mode without NATS client to fail, got {} client", client.transport_name() ), - Err(err) => assert!(err.to_string().contains("NATS client required")), + // With the `nats` feature the manager needs a client; without it the + // request plane itself is not compiled in. + Err(err) => assert!( + err.to_string().contains("NATS client required") + || err.to_string().contains("not compiled into this build"), + "{err}" + ), } } } diff --git a/lib/runtime/src/storage/kv.rs b/lib/runtime/src/storage/kv.rs index 87bd2a123463..2122929d4de1 100644 --- a/lib/runtime/src/storage/kv.rs +++ b/lib/runtime/src/storage/kv.rs @@ -22,7 +22,9 @@ use serde::{Deserialize, Serialize}; mod mem; pub use mem::MemoryStore; +#[cfg(feature = "nats")] mod nats; +#[cfg(feature = "nats")] pub use nats::NATSStore; #[cfg(feature = "etcd")] mod etcd; @@ -191,6 +193,7 @@ impl TryFrom for Selector { #[allow(clippy::large_enum_variant)] enum KeyValueStoreEnum { Memory(MemoryStore), + #[cfg(feature = "nats")] Nats(NATSStore), #[cfg(feature = "etcd")] Etcd(EtcdStore), @@ -207,6 +210,7 @@ impl KeyValueStoreEnum { use KeyValueStoreEnum::*; Ok(match self { Memory(x) => Box::new(x.get_or_create_bucket(bucket_name, ttl).await?), + #[cfg(feature = "nats")] Nats(x) => Box::new(x.get_or_create_bucket(bucket_name, ttl).await?), #[cfg(feature = "etcd")] Etcd(x) => Box::new(x.get_or_create_bucket(bucket_name, ttl).await?), @@ -221,6 +225,7 @@ impl KeyValueStoreEnum { .get_bucket(bucket_name) .await? .map(|b| Box::new(b) as Box), + #[cfg(feature = "nats")] Nats(x) => x .get_bucket(bucket_name) .await? @@ -244,6 +249,7 @@ impl KeyValueStoreEnum { Memory(x) => x.connection_id(), #[cfg(feature = "etcd")] Etcd(x) => x.connection_id(), + #[cfg(feature = "nats")] Nats(x) => x.connection_id(), File(x) => x.connection_id(), } @@ -255,6 +261,7 @@ impl KeyValueStoreEnum { Memory(x) => x.shutdown(), #[cfg(feature = "etcd")] Etcd(x) => x.shutdown(), + #[cfg(feature = "nats")] Nats(x) => x.shutdown(), File(x) => x.shutdown(), } diff --git a/lib/runtime/src/transports.rs b/lib/runtime/src/transports.rs index af17ddac71c7..253be2070c01 100644 --- a/lib/runtime/src/transports.rs +++ b/lib/runtime/src/transports.rs @@ -9,7 +9,22 @@ #[cfg(feature = "etcd")] pub mod etcd; pub mod event_plane; +#[cfg(feature = "nats")] pub mod nats; + +/// The NATS client handle threaded through the network manager. Without the +/// `nats` feature it is uninhabited, so every `Option` of it is `None`. +#[cfg(feature = "nats")] +pub type NatsClientHandle = async_nats::Client; +#[cfg(not(feature = "nats"))] +pub type NatsClientHandle = std::convert::Infallible; + +/// NATS connection options carried by `DistributedConfig`; uninhabited without +/// the `nats` feature. +#[cfg(feature = "nats")] +pub type NatsClientOptions = nats::ClientOptions; +#[cfg(not(feature = "nats"))] +pub type NatsClientOptions = std::convert::Infallible; pub mod tcp; mod utils; pub mod zmq; diff --git a/lib/runtime/src/transports/etcd.rs b/lib/runtime/src/transports/etcd.rs index 41e90dd4be96..b6bf08856f21 100644 --- a/lib/runtime/src/transports/etcd.rs +++ b/lib/runtime/src/transports/etcd.rs @@ -4,7 +4,6 @@ use crate::runtime::Runtime; use anyhow::{Context, Result}; -use async_nats::jetstream::kv; use derive_builder::Builder; use derive_getters::Dissolve; use futures::StreamExt; diff --git a/lib/runtime/src/transports/event_plane/mod.rs b/lib/runtime/src/transports/event_plane/mod.rs index 43937421c920..ea7537a934d5 100644 --- a/lib/runtime/src/transports/event_plane/mod.rs +++ b/lib/runtime/src/transports/event_plane/mod.rs @@ -6,6 +6,7 @@ mod codec; mod dynamic_subscriber; mod frame; +#[cfg(feature = "nats")] mod nats_transport; mod traits; mod transport; @@ -405,12 +406,14 @@ impl EventPublisher { // Use Msgpack codec for all transports enum TransportSetup { + #[cfg(feature = "nats")] Nats(Arc, Arc), ZmqDirect(Arc, Arc, String), // includes public endpoint ZmqBroker(Arc, Arc), } let transport_setup = match transport_kind { + #[cfg(feature = "nats")] EventTransportKind::Nats => { let transport = Arc::new(nats_transport::NatsTransport::new_publisher( drt.clone(), @@ -419,6 +422,10 @@ impl EventPublisher { let codec = Arc::new(Codec::Msgpack(MsgpackCodec)); TransportSetup::Nats(transport as Arc, codec) } + #[cfg(not(feature = "nats"))] + EventTransportKind::Nats => anyhow::bail!( + "NATS event plane is not compiled into this build (dynamo-runtime feature `nats`); use DYN_EVENT_PLANE=zmq" + ), EventTransportKind::Zmq => { // Check for broker mode if let Some(broker) = resolve_zmq_broker(drt, &scope).await? { @@ -480,6 +487,7 @@ impl EventPublisher { // Extract transport and codec, and register if needed let (tx, codec, discovery_instance) = match transport_setup { + #[cfg(feature = "nats")] TransportSetup::Nats(tx, codec) => { let transport_config = EventTransport::nats(scope.subject_prefix()); let spec = DiscoverySpec::EventChannel { @@ -744,12 +752,17 @@ impl EventSubscriber { // Use Msgpack codec for all transports let (wire_stream, codec): (WireStream, Arc) = match transport_kind { + #[cfg(feature = "nats")] EventTransportKind::Nats => { let transport = nats_transport::NatsTransport::new(drt.clone()); let stream = transport.subscribe(&routing_key).await?; let codec = Arc::new(Codec::Msgpack(MsgpackCodec)); (stream, codec) } + #[cfg(not(feature = "nats"))] + EventTransportKind::Nats => anyhow::bail!( + "NATS event plane is not compiled into this build (dynamo-runtime feature `nats`); use DYN_EVENT_PLANE=zmq" + ), EventTransportKind::Zmq => { // Check for broker mode if let Some(broker) = resolve_zmq_broker(drt, &scope).await? { diff --git a/lib/sidecar/sglang/Cargo.toml b/lib/sidecar/sglang/Cargo.toml index e0944e1b4989..2eb54ab42182 100644 --- a/lib/sidecar/sglang/Cargo.toml +++ b/lib/sidecar/sglang/Cargo.toml @@ -21,7 +21,7 @@ path = "src/main.rs" [dependencies] dynamo-backend-common = { workspace = true } -dynamo-runtime = { workspace = true, features = ["etcd"] } +dynamo-runtime = { workspace = true, features = ["etcd", "nats"] } dynamo-sidecar-common = { workspace = true } anyhow = { workspace = true }