Skip to content
Closed
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
2 changes: 1 addition & 1 deletion deploy/inference-gateway/ext-proc/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -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 }
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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)

Expand Down
2 changes: 1 addition & 1 deletion lib/backend-common/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -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 }
Expand Down
2 changes: 1 addition & 1 deletion lib/backend-common/examples/mocker/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -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"] }
10 changes: 7 additions & 3 deletions lib/llm/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -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"]

Expand All @@ -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"]
Expand Down Expand Up @@ -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 }
Expand Down Expand Up @@ -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"
Expand Down
12 changes: 12 additions & 0 deletions lib/llm/src/request_trace/sink.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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<Self> {
let nats_client = nats::ClientOptions::default()
Expand All @@ -76,6 +80,7 @@ impl NatsRequestTraceSink {
}
}

#[cfg(feature = "nats")]
#[async_trait]
impl RequestTraceSink for NatsRequestTraceSink {
fn name(&self) -> &'static str {
Expand Down Expand Up @@ -191,6 +196,13 @@ async fn parse_sinks_from_env() -> anyhow::Result<Vec<Arc<dyn RequestTraceSink>>
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?))
}
Expand Down
2 changes: 1 addition & 1 deletion lib/mocker/servers/vllm/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -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"] }
8 changes: 6 additions & 2 deletions lib/runtime/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand All @@ -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 }
Expand Down Expand Up @@ -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"] }
Expand Down
19 changes: 12 additions & 7 deletions lib/runtime/src/component.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -63,6 +59,7 @@ mod component;
mod endpoint;
mod namespace;
mod registry;
#[cfg(feature = "nats")]
pub mod service;

pub(crate) use client::EndpointDiscoverySource;
Expand Down Expand Up @@ -95,6 +92,7 @@ pub enum DeviceType {

#[derive(Default)]
pub struct RegistryInner {
#[cfg(feature = "nats")]
pub(crate) services: HashMap<String, Service>,
}

Expand Down Expand Up @@ -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.
Expand Down
8 changes: 7 additions & 1 deletion lib/runtime/src/component/endpoint.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -17,7 +19,6 @@ use crate::{
},
protocols::EndpointId,
traits::DistributedRuntimeProvider,
transports::nats,
};

fn endpoint_device_type() -> Option<DeviceType> {
Expand Down Expand Up @@ -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,
Expand Down
Loading
Loading