distributed.rs 32.9 KB
Newer Older
1
// SPDX-FileCopyrightText: Copyright (c) 2024-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
2
// SPDX-License-Identifier: Apache-2.0
Ryan Olson's avatar
Ryan Olson committed
3

4
5
6
use crate::component::{
    self, Component, ComponentBuilder, Endpoint, Instance, Namespace, RoutingOccupancyState,
};
7
use crate::config::environment_names::tcp_response_stream;
8
use crate::pipeline::PipelineError;
9
use crate::pipeline::network::manager::NetworkManager;
10
use crate::service::{ServiceClient, ServiceSet};
11
use crate::storage::kv;
12
use crate::{discovery, system_status_server, transports};
Ryan Olson's avatar
Ryan Olson committed
13
use crate::{
14
    discovery::Discovery,
15
16
    metrics::PrometheusUpdateCallback,
    metrics::{MetricsHierarchy, MetricsRegistry},
Ryan Olson's avatar
Ryan Olson committed
17
18
19
    transports::{etcd, nats, tcp},
};

20
use super::utils::GracefulShutdownTracker;
21
22
use crate::SystemHealth;
use crate::runtime::Runtime;
Ryan Olson's avatar
Ryan Olson committed
23

24
// Used instead of std::cell::OnceCell because get_or_try_init there is nightly
25
use async_once_cell::OnceCell;
26

27
use std::fmt;
28
use std::sync::{Arc, OnceLock, Weak};
29
use std::time::Duration;
30
use tokio::sync::watch::Receiver;
31
32

use anyhow::Result;
Ryan Olson's avatar
Ryan Olson committed
33
34
use derive_getters::Dissolve;
use figment::error;
35
36
use std::collections::HashMap;
use tokio::sync::Mutex;
37
use tokio_util::sync::CancellationToken;
Ryan Olson's avatar
Ryan Olson committed
38

39
type InstanceMap = HashMap<Endpoint, Weak<Receiver<Vec<Instance>>>>;
40
type RoutingOccupancyMap = HashMap<Endpoint, Weak<RoutingOccupancyState>>;
41

42
43
44
45
46
47
48
49
/// Distributed [Runtime] which provides access to shared resources across the cluster, this includes
/// communication protocols and transports.
#[derive(Clone)]
pub struct DistributedRuntime {
    // local runtime
    runtime: Runtime,

    nats_client: Option<transports::nats::Client>,
50
    network_manager: Arc<NetworkManager>,
51
52
    tcp_server: Arc<OnceCell<Arc<transports::tcp::server::TcpStreamServer>>>,
    system_status_server: Arc<OnceLock<Arc<system_status_server::SystemStatusServerInfo>>>,
53
    request_plane: RequestPlaneMode,
54
55
56
57

    // Service discovery client
    discovery_client: Arc<dyn discovery::Discovery>,

58
59
60
61
    // Discovery metadata (only used for Kubernetes backend)
    // Shared with system status server to expose via /metadata endpoint
    discovery_metadata: Option<Arc<tokio::sync::RwLock<discovery::DiscoveryMetadata>>>,

62
63
64
65
66
67
68
    // local registry for components
    // the registry allows us to use share runtime resources across instances of the same component object.
    // take for example two instances of a client to the same remote component. The registry allows us to use
    // a single endpoint watcher for both clients, this keeps the number background tasking watching specific
    // paths in etcd to a minimum.
    component_registry: component::Registry,

69
    instance_sources: Arc<tokio::sync::Mutex<InstanceMap>>,
70
    routing_occupancy_states: Arc<tokio::sync::Mutex<RoutingOccupancyMap>>,
71
72
73
74

    // Health Status
    system_health: Arc<parking_lot::Mutex<SystemHealth>>,

75
76
77
    // Local endpoint registry for in-process calls
    local_endpoint_registry: crate::local_endpoint_registry::LocalEndpointRegistry,

78
79
    // This hierarchy's own metrics registry
    metrics_registry: MetricsRegistry,
80
81
82

    // Registry for /engine/* route callbacks
    engine_routes: crate::engine_routes::EngineRouteRegistry,
83
84
}

85
impl MetricsHierarchy for DistributedRuntime {
86
87
88
89
    fn basename(&self) -> String {
        "".to_string() // drt has no basename. Basename only begins with the Namespace.
    }

90
91
92
93
94
95
    fn parent_hierarchies(&self) -> Vec<&dyn MetricsHierarchy> {
        vec![] // drt is the root, so no parent hierarchies
    }

    fn get_metrics_registry(&self) -> &MetricsRegistry {
        &self.metrics_registry
96
    }
97
98
99
100

    fn connection_id(&self) -> Option<u64> {
        Some(self.discovery_client.instance_id())
    }
101
102
}

Ryan Olson's avatar
Ryan Olson committed
103
104
105
106
107
108
impl std::fmt::Debug for DistributedRuntime {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "DistributedRuntime")
    }
}

Ryan Olson's avatar
Ryan Olson committed
109
110
impl DistributedRuntime {
    pub async fn new(runtime: Runtime, config: DistributedConfig) -> Result<Self> {
111
        let (discovery_backend, nats_config, request_plane) = config.dissolve();
Ryan Olson's avatar
Ryan Olson committed
112

113
114
115
116
        let nats_client = match nats_config {
            Some(nc) => Some(nc.connect().await?),
            None => None,
        };
Ryan Olson's avatar
Ryan Olson committed
117

118
        // Start system status server for health and metrics if enabled in configuration
119
120
121
122
123
124
125
126
        let config = crate::config::RuntimeConfig::from_settings().unwrap_or_default();
        // IMPORTANT: We must extract cancel_token from runtime BEFORE moving runtime into the struct below.
        // This is because after moving, runtime is no longer accessible in this scope (ownership rules).
        let cancel_token = if config.system_server_enabled() {
            Some(runtime.clone().child_token())
        } else {
            None
        };
127
128
        let starting_health_status = config.starting_health_status.clone();
        let use_endpoint_health_status = config.use_endpoint_health_status.clone();
129
130
        let health_endpoint_path = config.system_health_path.clone();
        let live_endpoint_path = config.system_live_path.clone();
131
        let system_health = Arc::new(parking_lot::Mutex::new(SystemHealth::new(
132
133
            starting_health_status,
            use_endpoint_health_status,
134
135
            health_endpoint_path,
            live_endpoint_path,
136
        )));
137

138
        // Initialize discovery client based on backend configuration
139
140
        let (discovery_client, discovery_metadata) = match discovery_backend {
            DiscoveryBackend::Kubernetes => {
141
142
143
144
145
146
147
148
149
150
151
152
153
154
                tracing::info!("Initializing Kubernetes discovery backend");
                let metadata = Arc::new(tokio::sync::RwLock::new(
                    crate::discovery::DiscoveryMetadata::new(),
                ));
                let client = crate::discovery::KubeDiscoveryClient::new(
                    metadata.clone(),
                    runtime.primary_token(),
                )
                .await
                .inspect_err(
                    |err| tracing::error!(%err, "Failed to initialize Kubernetes discovery client"),
                )?;
                (Arc::new(client) as Arc<dyn Discovery>, Some(metadata))
            }
155
            DiscoveryBackend::KvStore(kv_selector) => {
156
                tracing::info!("Initializing KV store discovery backend: {kv_selector}");
157
158
159
160
161
162
163
164
165
166
                let runtime_clone = runtime.clone();
                let store = match kv_selector {
                    kv::Selector::Etcd(etcd_config) => {
                        let etcd_client = etcd::Client::new(*etcd_config, runtime_clone).await.inspect_err(|err|
                            tracing::error!(%err, "Could not connect to etcd. Pass `--discovery-backend ..` to use a different backend or start etcd."))?;
                        kv::Manager::etcd(etcd_client)
                    }
                    kv::Selector::File(root) => kv::Manager::file(runtime.primary_token(), root),
                    kv::Selector::Memory => kv::Manager::memory(),
                };
167
168
                use crate::discovery::KVStoreDiscovery;
                (
169
170
                    Arc::new(KVStoreDiscovery::new(store, runtime.primary_token()))
                        as Arc<dyn Discovery>,
171
172
173
                    None,
                )
            }
174
175
        };

176
        let component_registry = component::Registry::new();
177

178
179
180
181
182
183
184
185
        // NetworkManager for request plane
        let network_manager = NetworkManager::new(
            runtime.child_token(),
            nats_client.clone().map(|c| c.client().clone()),
            component_registry.clone(),
            request_plane,
        );

186
        let distributed_runtime = Self {
Ryan Olson's avatar
Ryan Olson committed
187
            runtime,
188
            network_manager: Arc::new(network_manager),
Ryan Olson's avatar
Ryan Olson committed
189
190
            nats_client,
            tcp_server: Arc::new(OnceCell::new()),
191
            system_status_server: Arc::new(OnceLock::new()),
192
            discovery_client,
193
            discovery_metadata,
194
            component_registry,
195
            instance_sources: Arc::new(Mutex::new(HashMap::new())),
196
            routing_occupancy_states: Arc::new(Mutex::new(HashMap::new())),
197
            metrics_registry: crate::MetricsRegistry::new(),
198
            system_health,
199
            request_plane,
200
            local_endpoint_registry: crate::local_endpoint_registry::LocalEndpointRegistry::new(),
201
            engine_routes: crate::engine_routes::EngineRouteRegistry::new(),
202
203
        };

204
205
206
207
208
209
        // Initialize the uptime gauge in SystemHealth
        distributed_runtime
            .system_health
            .lock()
            .initialize_uptime_gauge(&distributed_runtime)?;

210
211
212
213
214
215
216
217
218
219
220
221
        // Register an update callback so the uptime gauge is refreshed before
        // every Prometheus scrape (both system status server and frontend).
        {
            let system_health = distributed_runtime.system_health.clone();
            distributed_runtime
                .metrics_registry
                .add_update_callback(std::sync::Arc::new(move || {
                    system_health.lock().update_uptime_gauge();
                    Ok(())
                }));
        }

222
        // Handle system status server initialization
223
        if let Some(cancel_token) = cancel_token {
224
            // System server is enabled - start both the state and HTTP server
225
            let host = config.system_host.clone();
226
            let port = config.system_port as u16;
227

228
            // Start system status server (it creates SystemStatusState internally)
229
            match crate::system_status_server::spawn_system_status_server(
230
231
232
233
                &host,
                port,
                cancel_token,
                Arc::new(distributed_runtime.clone()),
234
                distributed_runtime.discovery_metadata.clone(),
235
236
237
            )
            .await
            {
238
                Ok((addr, handle)) => {
239
                    tracing::info!("System status server started successfully on {addr}");
240

241
242
243
244
245
246
                    // Store system status server information
                    let system_status_server_info =
                        crate::system_status_server::SystemStatusServerInfo::new(
                            addr,
                            Some(handle),
                        );
247

248
                    // Initialize the system_status_server field
249
                    distributed_runtime
250
251
252
                        .system_status_server
                        .set(Arc::new(system_status_server_info))
                        .expect("System status server info should only be set once");
253
254
                }
                Err(e) => {
255
                    tracing::error!("System status server startup failed: {e}");
256
                }
257
            }
258
        } else {
259
            // System server HTTP is disabled, but uptime metrics are still being tracked via SystemHealth
260
261
262
            tracing::debug!(
                "System status server HTTP endpoints disabled, but uptime metrics are being tracked"
            );
263
264
        }

265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
        // Start health check manager if enabled
        if config.health_check_enabled {
            let health_check_config = crate::health_check::HealthCheckConfig {
                canary_wait_time: std::time::Duration::from_secs(config.canary_wait_time_secs),
                request_timeout: std::time::Duration::from_secs(
                    config.health_check_request_timeout_secs,
                ),
            };

            // Start the health check manager (spawns per-endpoint monitoring tasks)
            match crate::health_check::start_health_check_manager(
                distributed_runtime.clone(),
                Some(health_check_config),
            )
            .await
            {
                Ok(()) => tracing::info!(
                    "Health check manager started (canary_wait_time: {}s, request_timeout: {}s)",
                    config.canary_wait_time_secs,
                    config.health_check_request_timeout_secs
                ),
286
                Err(e) => tracing::error!("Health check manager failed to start: {e}"),
287
288
289
            }
        }

290
        Ok(distributed_runtime)
Ryan Olson's avatar
Ryan Olson committed
291
292
293
    }

    pub async fn from_settings(runtime: Runtime) -> Result<Self> {
294
        let config = DistributedConfig::from_settings();
Ryan Olson's avatar
Ryan Olson committed
295
296
297
298
299
300
301
        Self::new(runtime, config).await
    }

    pub fn runtime(&self) -> &Runtime {
        &self.runtime
    }

302
303
304
305
    pub fn primary_token(&self) -> CancellationToken {
        self.runtime.primary_token()
    }

306
307
308
309
310
311
312
313
314
315
316
    // TODO: Don't hand out pointers, instead have methods to use the registry in friendly ways
    // (without being aware of async locks and so on)
    pub fn component_registry(&self) -> &component::Registry {
        &self.component_registry
    }

    // TODO: Don't hand out pointers, instead provide system health related services.
    pub fn system_health(&self) -> Arc<parking_lot::Mutex<SystemHealth>> {
        self.system_health.clone()
    }

317
318
319
320
321
322
323
    /// Get the local endpoint registry for in-process endpoint calls
    pub fn local_endpoint_registry(
        &self,
    ) -> &crate::local_endpoint_registry::LocalEndpointRegistry {
        &self.local_endpoint_registry
    }

324
325
326
327
328
    /// Get the engine route registry for registering custom /engine/* routes
    pub fn engine_routes(&self) -> &crate::engine_routes::EngineRouteRegistry {
        &self.engine_routes
    }

329
    pub fn connection_id(&self) -> u64 {
330
        self.discovery_client.instance_id()
Ryan Olson's avatar
Ryan Olson committed
331
332
333
334
    }

    pub fn shutdown(&self) {
        self.runtime.shutdown();
335
        self.discovery_client.shutdown();
Ryan Olson's avatar
Ryan Olson committed
336
337
338
339
    }

    /// Create a [`Namespace`]
    pub fn namespace(&self, name: impl Into<String>) -> Result<Namespace> {
340
        Namespace::new(self.clone(), name.into())
Ryan Olson's avatar
Ryan Olson committed
341
342
    }

343
344
345
    /// Returns the discovery interface for service registration and discovery
    pub fn discovery(&self) -> Arc<dyn Discovery> {
        self.discovery_client.clone()
346
347
    }

348
    pub async fn tcp_server(&self) -> Result<Arc<tcp::server::TcpStreamServer>> {
Ryan Olson's avatar
Ryan Olson committed
349
350
351
        Ok(self
            .tcp_server
            .get_or_try_init(async move {
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
                let port = match std::env::var(tcp_response_stream::DYN_TCP_RESPONSE_STREAM_PORT) {
                    Ok(p) => p.parse::<u16>().map_err(|_| {
                        PipelineError::Generic(format!(
                            "invalid {}: '{}' is not a valid port number",
                            tcp_response_stream::DYN_TCP_RESPONSE_STREAM_PORT,
                            p
                        ))
                    })?,
                    Err(_) => 0,
                };
                let interface = std::env::var(tcp_response_stream::DYN_TCP_RESPONSE_STREAM_HOST)
                    .ok()
                    .filter(|h| !h.is_empty());

                let host_suffix = interface
                    .as_ref()
                    .map_or(String::new(), |h| format!(" on host {h}"));
                if port == 0 {
                    tracing::info!(
                        "TCP response stream server using OS-assigned port{host_suffix}"
                    );
                } else {
                    tracing::info!(
                        "TCP response stream server using fixed port {port}{host_suffix}"
                    );
                }

                let options = tcp::server::ServerOptions { port, interface };
Ryan Olson's avatar
Ryan Olson committed
380
                let server = tcp::server::TcpStreamServer::new(options).await?;
381
                Ok::<_, PipelineError>(server)
Ryan Olson's avatar
Ryan Olson committed
382
383
384
385
386
            })
            .await?
            .clone())
    }

387
    /// Get the network manager
388
389
390
    ///
    /// The network manager consolidates all network configuration and provides
    /// unified access to request plane servers and clients.
391
392
    pub fn network_manager(&self) -> Arc<NetworkManager> {
        self.network_manager.clone()
393
394
395
396
397
398
399
400
401
    }

    /// Get the request plane server (convenience method)
    ///
    /// This is a shortcut for `network_manager().await?.server().await`.
    pub async fn request_plane_server(
        &self,
    ) -> Result<Arc<dyn crate::pipeline::network::ingress::unified_server::RequestPlaneServer>>
    {
402
        self.network_manager().server().await
Ryan Olson's avatar
Ryan Olson committed
403
404
    }

405
406
407
408
409
    /// Get system status server information if available
    pub fn system_status_server_info(
        &self,
    ) -> Option<Arc<crate::system_status_server::SystemStatusServerInfo>> {
        self.system_status_server.get().cloned()
410
411
    }

412
413
414
415
416
    /// How the frontend should talk to the backend.
    pub fn request_plane(&self) -> RequestPlaneMode {
        self.request_plane
    }

417
418
419
    pub fn child_token(&self) -> CancellationToken {
        self.runtime.child_token()
    }
420

421
422
423
424
    pub(crate) fn graceful_shutdown_tracker(&self) -> Arc<GracefulShutdownTracker> {
        self.runtime.graceful_shutdown_tracker()
    }

425
    pub fn instance_sources(&self) -> Arc<Mutex<InstanceMap>> {
426
427
        self.instance_sources.clone()
    }
428

429
430
431
432
    pub(crate) fn routing_occupancy_states(&self) -> Arc<Mutex<RoutingOccupancyMap>> {
        self.routing_occupancy_states.clone()
    }

433
434
    /// 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.
435
436
437
    ///
    /// When NATS is not available (e.g., running in approximate mode with --no-kv-events),
    /// this function returns Ok(()) silently since publishing is optional in that mode.
438
    pub async fn kv_router_nats_publish(
439
440
441
442
443
        &self,
        subject: String,
        payload: bytes::Bytes,
    ) -> anyhow::Result<()> {
        let Some(nats_client) = self.nats_client.as_ref() else {
444
            // NATS not available - this is expected in approximate mode (--no-kv-events)
445
            tracing::trace!("Skipping NATS publish (NATS not configured): {subject}");
446
            return Ok(());
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
        };
        Ok(nats_client.client().publish(subject, payload).await?)
    }

    /// 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(
        &self,
        subject: String,
    ) -> Result<async_nats::Subscriber> {
        let Some(nats_client) = self.nats_client.as_ref() else {
            anyhow::bail!("KV router's EventSubscriber requires NATS");
        };
        Ok(nats_client.client().subscribe(subject).await?)
    }

463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
    /// 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
    pub async fn kv_router_nats_request(
        &self,
        subject: String,
        payload: bytes::Bytes,
        timeout: std::time::Duration,
    ) -> anyhow::Result<async_nats::Message> {
        let Some(nats_client) = self.nats_client.as_ref() else {
            anyhow::bail!("KV router's request requires NATS");
        };
        let response =
            tokio::time::timeout(timeout, nats_client.client().request(subject, payload))
                .await
                .map_err(|_| anyhow::anyhow!("Request timed out after {:?}", timeout))??;
        Ok(response)
    }

482
483
484
    /// 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.
485
486
487
488
489
490
491
492
493
494
    ///
    /// Returns a receiver that signals when the NATS service registration is complete.
    /// The caller should use `blocking_recv()` to wait for completion.
    pub fn register_nats_service(
        &self,
        component: Component,
    ) -> tokio::sync::mpsc::Receiver<Result<(), String>> {
        // Create a oneshot-style channel (capacity 1) to signal completion
        let (tx, rx) = tokio::sync::mpsc::channel::<Result<(), String>>(1);

495
496
497
498
        let drt = self.clone();
        self.runtime().secondary().spawn(async move {
            let service_name = component.service_name();

499
500
501
502
503
504
505
506
507
508
509
510
            // Pre-check to save cost of creating the service, but don't hold the lock
            if drt
                .component_registry()
                .inner
                .lock()
                .await
                .services
                .contains_key(&service_name)
            {
                // The NATS service is per component, but it is called from `serve_endpoint`, and there
                // are often multiple endpoints for a component (e.g. `clear_kv_blocks` and `generate`).
                tracing::trace!("Service {service_name} already exists");
511
512
                // Signal success - service already exists
                let _ = tx.send(Ok(())).await;
513
514
                return;
            }
515

516
517
            let Some(nats_client) = drt.nats_client.as_ref() else {
                tracing::error!("Cannot create NATS service without NATS.");
518
519
520
                let _ = tx
                    .send(Err("Cannot create NATS service without NATS".to_string()))
                    .await;
521
522
523
524
525
526
527
528
                return;
            };
            let description = None;
            let nats_service = match crate::component::service::build_nats_service(
                nats_client,
                &component,
                description,
            )
529
            .await
530
531
532
533
            {
                Ok(service) => service,
                Err(err) => {
                    tracing::error!(error = %err, component = service_name, "Failed to build NATS service");
534
                    let _ = tx.send(Err(format!("Failed to build NATS service: {err}"))).await;
535
536
537
                    return;
                }
            };
538

539
540
541
542
            let mut guard = drt.component_registry().inner.lock().await;
            if !guard.services.contains_key(&service_name) {
                // Normal case
                guard.services.insert(service_name.clone(), nats_service);
543

544
                tracing::info!("Added NATS service {service_name}");
545

546
547
548
549
550
551
552
                drop(guard);
            } else {
                drop(guard);
                let _ = nats_service.stop().await;
                // The NATS service is per component, but it is called from `serve_endpoint`, and there
                // are often multiple endpoints for a component (e.g. `clear_kv_blocks` and `generate`).
                // TODO: Is this still true?
553
            }
554
555
556

            // Signal completion - service registered successfully
            let _ = tx.send(Ok(())).await;
557
        });
558
559

        rx
560
    }
Ryan Olson's avatar
Ryan Olson committed
561
562
}

563
564
565
566
567
568
569
570
571
/// Selects which discovery backend to use and, for KV store backends, which KV store.
#[derive(Clone, Debug)]
pub enum DiscoveryBackend {
    /// Use Kubernetes API for service discovery (no KV store needed)
    Kubernetes,
    /// Use a KV store (etcd, file, or memory) for service discovery
    KvStore(kv::Selector),
}

Ryan Olson's avatar
Ryan Olson committed
572
573
#[derive(Dissolve)]
pub struct DistributedConfig {
574
    pub discovery_backend: DiscoveryBackend,
575
    pub nats_config: Option<nats::ClientOptions>,
576
    pub request_plane: RequestPlaneMode,
Ryan Olson's avatar
Ryan Olson committed
577
578
579
}

impl DistributedConfig {
580
    pub fn from_settings() -> DistributedConfig {
581
        let request_plane = RequestPlaneMode::from_env();
582
583
584
585
586
587
        // NATS is used for more than just NATS request-plane RPC:
        // - KV router events (JetStream or NATS core + local indexer)
        // - inter-router replica sync (NATS core)
        //
        // Historically we only connected to NATS when the request plane was NATS, which made
        // `DYN_REQUEST_PLANE=tcp|http` incompatible with KV routing modes that rely on NATS.
588
589
590
591
592
593
594
595
596
        // Enable the NATS client when any of these hold:
        // 1. Request plane is NATS
        // 2. NATS_SERVER is explicitly configured
        // 3. Event plane is NATS (the default)
        let event_plane_is_nats =
            std::env::var(crate::config::environment_names::event_plane::DYN_EVENT_PLANE)
                .map(|v| v.eq_ignore_ascii_case("nats"))
                .unwrap_or(true);

597
        let nats_enabled = request_plane.is_nats()
598
599
            || std::env::var(crate::config::environment_names::nats::NATS_SERVER).is_ok()
            || event_plane_is_nats;
600

601
602
603
604
        // DYN_DISCOVERY_BACKEND selects the discovery mechanism
        // Valid values: "kubernetes", "etcd" (default), "file", "mem"
        let backend_str =
            std::env::var("DYN_DISCOVERY_BACKEND").unwrap_or_else(|_| "etcd".to_string());
605

606
607
608
609
610
611
612
613
614
615
616
617
618
619
        let discovery_backend = match backend_str.as_str() {
            "kubernetes" => {
                tracing::info!("Using Kubernetes discovery backend");
                DiscoveryBackend::Kubernetes
            }
            other => {
                let selector: kv::Selector = other.parse().unwrap_or_else(|_| {
                    panic!(
                        "Unknown DYN_DISCOVERY_BACKEND value: '{other}'. \
                         Valid options: kubernetes, etcd, file, mem"
                    )
                });
                DiscoveryBackend::KvStore(selector)
            }
620
621
        };

Ryan Olson's avatar
Ryan Olson committed
622
        DistributedConfig {
623
            discovery_backend,
624
            nats_config: if nats_enabled {
625
626
627
628
629
                Some(nats::ClientOptions::default())
            } else {
                None
            },
            request_plane,
Ryan Olson's avatar
Ryan Olson committed
630
631
        }
    }
Ryan Olson's avatar
Ryan Olson committed
632
633

    pub fn for_cli() -> DistributedConfig {
634
635
636
637
        let etcd_config = etcd::ClientOptions {
            attach_lease: false,
            ..Default::default()
        };
638
        let request_plane = RequestPlaneMode::from_env();
639
640
641
642
        let event_plane_is_nats =
            std::env::var(crate::config::environment_names::event_plane::DYN_EVENT_PLANE)
                .map(|v| v.eq_ignore_ascii_case("nats"))
                .unwrap_or(true);
643
        let nats_enabled = request_plane.is_nats()
644
645
            || std::env::var(crate::config::environment_names::nats::NATS_SERVER).is_ok()
            || event_plane_is_nats;
646
        DistributedConfig {
647
            discovery_backend: DiscoveryBackend::KvStore(kv::Selector::Etcd(Box::new(etcd_config))),
648
            nats_config: if nats_enabled {
649
650
651
652
653
                Some(nats::ClientOptions::default())
            } else {
                None
            },
            request_plane,
654
655
        }
    }
656
657
658
659
660

    /// A DistributedConfig that isn't distributed, for when the frontend and backend are in the
    /// same process.
    pub fn process_local() -> DistributedConfig {
        DistributedConfig {
661
            discovery_backend: DiscoveryBackend::KvStore(kv::Selector::Memory),
662
663
664
665
666
667
            nats_config: None,
            // This won't be used in process local, so we likely need a "none" option to
            // communicate that and avoid opening the ports.
            request_plane: RequestPlaneMode::Tcp,
        }
    }
668
669
670
671
672
}

/// Request plane transport mode configuration
///
/// This determines how requests are distributed from routers to workers:
673
/// - `Nats`: Use NATS for request distribution (legacy)
674
/// - `Http`: Use HTTP/2 for request distribution
675
/// - `Tcp`: Use raw TCP for request distribution with msgpack support (default)
676
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
677
pub enum RequestPlaneMode {
678
    /// Use NATS for request plane
679
680
681
682
    Nats,
    /// Use HTTP/2 for request plane
    Http,
    /// Use raw TCP for request plane with msgpack support
683
    #[default]
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
    Tcp,
}

impl fmt::Display for RequestPlaneMode {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Nats => write!(f, "nats"),
            Self::Http => write!(f, "http"),
            Self::Tcp => write!(f, "tcp"),
        }
    }
}

impl std::str::FromStr for RequestPlaneMode {
    type Err = anyhow::Error;

    fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
        match s.to_lowercase().as_str() {
            "nats" => Ok(Self::Nats),
            "http" => Ok(Self::Http),
            "tcp" => Ok(Self::Tcp),
            _ => Err(anyhow::anyhow!(
                "Invalid request plane mode: '{}'. Valid options are: 'nats', 'http', 'tcp'",
                s
            )),
709
        }
Ryan Olson's avatar
Ryan Olson committed
710
    }
Ryan Olson's avatar
Ryan Olson committed
711
}
712

713
714
715
716
717
718
719
720
721
impl RequestPlaneMode {
    /// Get the request plane mode from environment variable (uncached)
    /// Reads from `DYN_REQUEST_PLANE` environment variable.
    fn from_env() -> Self {
        std::env::var("DYN_REQUEST_PLANE")
            .ok()
            .and_then(|s| s.parse().ok())
            .unwrap_or_default()
    }
722
723
724
725

    pub fn is_nats(&self) -> bool {
        matches!(self, RequestPlaneMode::Nats)
    }
726
727
}

728
pub mod distributed_test_utils {
729
730
    //! Common test helper functions for DistributedRuntime tests

731
    /// Helper function to create a DRT instance for integration-only tests.
732
    /// Uses from_current to leverage existing tokio runtime
733
    /// Note: Settings are read from environment variables inside DistributedRuntime::from_settings
734
    #[cfg(feature = "integration")]
735
    pub async fn create_test_drt_async() -> super::DistributedRuntime {
736
        use crate::transports::nats;
737

738
        let rt = crate::Runtime::from_current().unwrap();
739
        let config = super::DistributedConfig {
740
741
742
            discovery_backend: super::DiscoveryBackend::KvStore(
                crate::storage::kv::Selector::Memory,
            ),
743
            nats_config: Some(nats::ClientOptions::default()),
744
            request_plane: crate::distributed::RequestPlaneMode::default(),
745
746
        };
        super::DistributedRuntime::new(rt, config).await.unwrap()
747
    }
748
749
750
751
752
753
754
755
756
757

    /// Helper function to create a DRT instance which points at
    /// a (shared) file-backed KV store and ephemeral NATS transport so that
    /// multiple DRT instances may observe the same registration state.
    /// NOTE: This gets around the fact that create_test_drt_async() is
    /// hardcoded to spin up a memory-backed discovery store
    /// which means we can't share discovery state across runtimes.
    pub async fn create_test_shared_drt_async(
        store_path: &std::path::Path,
    ) -> super::DistributedRuntime {
758
        use crate::transports::nats;
759
760
761

        let rt = crate::Runtime::from_current().unwrap();
        let config = super::DistributedConfig {
762
763
764
            discovery_backend: super::DiscoveryBackend::KvStore(
                crate::storage::kv::Selector::File(store_path.to_path_buf()),
            ),
765
766
767
768
769
            nats_config: Some(nats::ClientOptions::default()),
            request_plane: crate::distributed::RequestPlaneMode::default(),
        };
        super::DistributedRuntime::new(rt, config).await.unwrap()
    }
770
}
771

772
#[cfg(all(test, feature = "integration"))]
773
mod tests {
774
    use super::RequestPlaneMode;
775
776
777
778
    use super::distributed_test_utils::create_test_drt_async;

    #[tokio::test]
    async fn test_drt_uptime_after_delay_system_disabled() {
779
        use crate::config::environment_names::runtime::system as env_system;
780
        // Test uptime with system status server disabled
781
        temp_env::async_with_vars([(env_system::DYN_SYSTEM_PORT, None::<&str>)], async {
782
783
784
785
786
787
788
            // Start a DRT
            let drt = create_test_drt_async().await;

            // Wait 50ms
            tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;

            // Check that uptime is 50+ ms
789
            let uptime = drt.system_health.lock().uptime();
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
            assert!(
                uptime >= std::time::Duration::from_millis(50),
                "Expected uptime to be at least 50ms, but got {:?}",
                uptime
            );

            println!(
                "✓ DRT uptime test passed (system disabled): uptime = {:?}",
                uptime
            );
        })
        .await;
    }

    #[tokio::test]
    async fn test_drt_uptime_after_delay_system_enabled() {
806
        use crate::config::environment_names::runtime::system as env_system;
807
        // Test uptime with system status server enabled
808
        temp_env::async_with_vars([(env_system::DYN_SYSTEM_PORT, Some("8081"))], async {
809
810
811
812
813
814
815
            // Start a DRT
            let drt = create_test_drt_async().await;

            // Wait 50ms
            tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;

            // Check that uptime is 50+ ms
816
            let uptime = drt.system_health.lock().uptime();
817
818
819
820
821
822
823
824
825
826
827
828
829
            assert!(
                uptime >= std::time::Duration::from_millis(50),
                "Expected uptime to be at least 50ms, but got {:?}",
                uptime
            );

            println!(
                "✓ DRT uptime test passed (system enabled): uptime = {:?}",
                uptime
            );
        })
        .await;
    }
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865

    #[test]
    fn test_request_plane_mode_from_str() {
        assert_eq!(
            "nats".parse::<RequestPlaneMode>().unwrap(),
            RequestPlaneMode::Nats
        );
        assert_eq!(
            "http".parse::<RequestPlaneMode>().unwrap(),
            RequestPlaneMode::Http
        );
        assert_eq!(
            "tcp".parse::<RequestPlaneMode>().unwrap(),
            RequestPlaneMode::Tcp
        );
        assert_eq!(
            "NATS".parse::<RequestPlaneMode>().unwrap(),
            RequestPlaneMode::Nats
        );
        assert_eq!(
            "HTTP".parse::<RequestPlaneMode>().unwrap(),
            RequestPlaneMode::Http
        );
        assert_eq!(
            "TCP".parse::<RequestPlaneMode>().unwrap(),
            RequestPlaneMode::Tcp
        );
        assert!("invalid".parse::<RequestPlaneMode>().is_err());
    }

    #[test]
    fn test_request_plane_mode_display() {
        assert_eq!(RequestPlaneMode::Nats.to_string(), "nats");
        assert_eq!(RequestPlaneMode::Http.to_string(), "http");
        assert_eq!(RequestPlaneMode::Tcp.to_string(), "tcp");
    }
866
}