distributed.rs 30.2 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
use crate::component::{Component, Instance};
5
use crate::pipeline::PipelineError;
6
use crate::pipeline::network::manager::NetworkManager;
7
use crate::service::{ServiceClient, ServiceSet};
8
use crate::storage::kv;
Ryan Olson's avatar
Ryan Olson committed
9
use crate::{
10
    component::{self, ComponentBuilder, Endpoint, Namespace},
11
    discovery::Discovery,
12
13
    metrics::PrometheusUpdateCallback,
    metrics::{MetricsHierarchy, MetricsRegistry},
Ryan Olson's avatar
Ryan Olson committed
14
15
    transports::{etcd, nats, tcp},
};
16
use crate::{discovery, system_status_server, transports};
Ryan Olson's avatar
Ryan Olson committed
17

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

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

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

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

37
38
type InstanceMap = HashMap<Endpoint, Weak<Receiver<Vec<Instance>>>>;

39
40
41
42
43
44
45
46
/// 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>,
47
    network_manager: Arc<NetworkManager>,
48
49
    tcp_server: Arc<OnceCell<Arc<transports::tcp::server::TcpStreamServer>>>,
    system_status_server: Arc<OnceLock<Arc<system_status_server::SystemStatusServerInfo>>>,
50
    request_plane: RequestPlaneMode,
51
52
53
54

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

55
56
57
58
    // 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>>>,

59
60
61
62
63
64
65
    // 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,

66
    instance_sources: Arc<tokio::sync::Mutex<InstanceMap>>,
67
68
69
70

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

71
72
73
    // Local endpoint registry for in-process calls
    local_endpoint_registry: crate::local_endpoint_registry::LocalEndpointRegistry,

74
75
    // This hierarchy's own metrics registry
    metrics_registry: MetricsRegistry,
76
77
78

    // Registry for /engine/* route callbacks
    engine_routes: crate::engine_routes::EngineRouteRegistry,
79
80
}

81
impl MetricsHierarchy for DistributedRuntime {
82
83
84
85
    fn basename(&self) -> String {
        "".to_string() // drt has no basename. Basename only begins with the Namespace.
    }

86
87
88
89
90
91
    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
92
93
94
    }
}

Ryan Olson's avatar
Ryan Olson committed
95
96
97
98
99
100
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
101
102
impl DistributedRuntime {
    pub async fn new(runtime: Runtime, config: DistributedConfig) -> Result<Self> {
103
        let (discovery_backend, nats_config, request_plane) = config.dissolve();
Ryan Olson's avatar
Ryan Olson committed
104

105
106
107
108
        let nats_client = match nats_config {
            Some(nc) => Some(nc.connect().await?),
            None => None,
        };
Ryan Olson's avatar
Ryan Olson committed
109

110
        // Start system status server for health and metrics if enabled in configuration
111
112
113
114
115
116
117
118
        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
        };
119
120
        let starting_health_status = config.starting_health_status.clone();
        let use_endpoint_health_status = config.use_endpoint_health_status.clone();
121
122
        let health_endpoint_path = config.system_health_path.clone();
        let live_endpoint_path = config.system_live_path.clone();
123
        let system_health = Arc::new(parking_lot::Mutex::new(SystemHealth::new(
124
125
            starting_health_status,
            use_endpoint_health_status,
126
127
            health_endpoint_path,
            live_endpoint_path,
128
        )));
129

130
        // Initialize discovery client based on backend configuration
131
132
        let (discovery_client, discovery_metadata) = match discovery_backend {
            DiscoveryBackend::Kubernetes => {
133
134
135
136
137
138
139
140
141
142
143
144
145
146
                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))
            }
147
            DiscoveryBackend::KvStore(kv_selector) => {
148
                tracing::info!("Initializing KV store discovery backend");
149
150
151
152
153
154
155
156
157
158
                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(),
                };
159
160
                use crate::discovery::KVStoreDiscovery;
                (
161
162
                    Arc::new(KVStoreDiscovery::new(store, runtime.primary_token()))
                        as Arc<dyn Discovery>,
163
164
165
                    None,
                )
            }
166
167
        };

168
        let component_registry = component::Registry::new();
169

170
171
172
173
174
175
176
177
        // 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,
        );

178
        let distributed_runtime = Self {
Ryan Olson's avatar
Ryan Olson committed
179
            runtime,
180
            network_manager: Arc::new(network_manager),
Ryan Olson's avatar
Ryan Olson committed
181
182
            nats_client,
            tcp_server: Arc::new(OnceCell::new()),
183
            system_status_server: Arc::new(OnceLock::new()),
184
            discovery_client,
185
            discovery_metadata,
186
            component_registry,
187
            instance_sources: Arc::new(Mutex::new(HashMap::new())),
188
            metrics_registry: crate::MetricsRegistry::new(),
189
            system_health,
190
            request_plane,
191
            local_endpoint_registry: crate::local_endpoint_registry::LocalEndpointRegistry::new(),
192
            engine_routes: crate::engine_routes::EngineRouteRegistry::new(),
193
194
        };

195
196
197
198
199
200
        // Initialize the uptime gauge in SystemHealth
        distributed_runtime
            .system_health
            .lock()
            .initialize_uptime_gauge(&distributed_runtime)?;

201
        // Handle system status server initialization
202
        if let Some(cancel_token) = cancel_token {
203
            // System server is enabled - start both the state and HTTP server
204
            let host = config.system_host.clone();
205
            let port = config.system_port as u16;
206

207
            // Start system status server (it creates SystemStatusState internally)
208
            match crate::system_status_server::spawn_system_status_server(
209
210
211
212
                &host,
                port,
                cancel_token,
                Arc::new(distributed_runtime.clone()),
213
                distributed_runtime.discovery_metadata.clone(),
214
215
216
            )
            .await
            {
217
                Ok((addr, handle)) => {
218
                    tracing::info!("System status server started successfully on {}", addr);
219

220
221
222
223
224
225
                    // Store system status server information
                    let system_status_server_info =
                        crate::system_status_server::SystemStatusServerInfo::new(
                            addr,
                            Some(handle),
                        );
226

227
                    // Initialize the system_status_server field
228
                    distributed_runtime
229
230
231
                        .system_status_server
                        .set(Arc::new(system_status_server_info))
                        .expect("System status server info should only be set once");
232
233
                }
                Err(e) => {
234
                    tracing::error!("System status server startup failed: {}", e);
235
                }
236
            }
237
        } else {
238
            // System server HTTP is disabled, but uptime metrics are still being tracked via SystemHealth
239
240
241
            tracing::debug!(
                "System status server HTTP endpoints disabled, but uptime metrics are being tracked"
            );
242
243
        }

244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
        // 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
                ),
                Err(e) => tracing::error!("Health check manager failed to start: {}", e),
            }
        }

269
        Ok(distributed_runtime)
Ryan Olson's avatar
Ryan Olson committed
270
271
272
    }

    pub async fn from_settings(runtime: Runtime) -> Result<Self> {
273
        let config = DistributedConfig::from_settings();
Ryan Olson's avatar
Ryan Olson committed
274
275
276
277
278
279
280
        Self::new(runtime, config).await
    }

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

281
282
283
284
    pub fn primary_token(&self) -> CancellationToken {
        self.runtime.primary_token()
    }

285
286
287
288
289
290
291
292
293
294
295
    // 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()
    }

296
297
298
299
300
301
302
    /// 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
    }

303
304
305
306
307
    /// Get the engine route registry for registering custom /engine/* routes
    pub fn engine_routes(&self) -> &crate::engine_routes::EngineRouteRegistry {
        &self.engine_routes
    }

308
    pub fn connection_id(&self) -> u64 {
309
        self.discovery_client.instance_id()
Ryan Olson's avatar
Ryan Olson committed
310
311
312
313
    }

    pub fn shutdown(&self) {
        self.runtime.shutdown();
314
        self.discovery_client.shutdown();
Ryan Olson's avatar
Ryan Olson committed
315
316
317
318
    }

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

322
323
324
    /// Returns the discovery interface for service registration and discovery
    pub fn discovery(&self) -> Arc<dyn Discovery> {
        self.discovery_client.clone()
325
326
    }

327
    pub async fn tcp_server(&self) -> Result<Arc<tcp::server::TcpStreamServer>> {
Ryan Olson's avatar
Ryan Olson committed
328
329
330
331
332
        Ok(self
            .tcp_server
            .get_or_try_init(async move {
                let options = tcp::server::ServerOptions::default();
                let server = tcp::server::TcpStreamServer::new(options).await?;
333
                Ok::<_, PipelineError>(server)
Ryan Olson's avatar
Ryan Olson committed
334
335
336
337
338
            })
            .await?
            .clone())
    }

339
    /// Get the network manager
340
341
342
    ///
    /// The network manager consolidates all network configuration and provides
    /// unified access to request plane servers and clients.
343
344
    pub fn network_manager(&self) -> Arc<NetworkManager> {
        self.network_manager.clone()
345
346
347
348
349
350
351
352
353
    }

    /// 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>>
    {
354
        self.network_manager().server().await
Ryan Olson's avatar
Ryan Olson committed
355
356
    }

357
358
359
360
361
    /// 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()
362
363
    }

364
365
366
367
368
    /// How the frontend should talk to the backend.
    pub fn request_plane(&self) -> RequestPlaneMode {
        self.request_plane
    }

369
370
371
    pub fn child_token(&self) -> CancellationToken {
        self.runtime.child_token()
    }
372

373
374
375
376
    pub(crate) fn graceful_shutdown_tracker(&self) -> Arc<GracefulShutdownTracker> {
        self.runtime.graceful_shutdown_tracker()
    }

377
    pub fn instance_sources(&self) -> Arc<Mutex<InstanceMap>> {
378
379
        self.instance_sources.clone()
    }
380
381
382

    /// 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.
383
384
385
    ///
    /// 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.
386
    pub async fn kv_router_nats_publish(
387
388
389
390
391
        &self,
        subject: String,
        payload: bytes::Bytes,
    ) -> anyhow::Result<()> {
        let Some(nats_client) = self.nats_client.as_ref() else {
392
393
394
            // NATS not available - this is expected in approximate mode (--no-kv-events)
            tracing::trace!("Skipping NATS publish (NATS not configured): {}", subject);
            return Ok(());
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
        };
        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?)
    }

411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
    /// 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)
    }

430
431
432
    /// 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.
433
434
435
436
437
438
439
440
441
442
    ///
    /// 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);

443
444
445
446
        let drt = self.clone();
        self.runtime().secondary().spawn(async move {
            let service_name = component.service_name();

447
448
449
450
451
452
453
454
455
456
457
458
            // 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");
459
460
                // Signal success - service already exists
                let _ = tx.send(Ok(())).await;
461
462
                return;
            }
463

464
465
            let Some(nats_client) = drt.nats_client.as_ref() else {
                tracing::error!("Cannot create NATS service without NATS.");
466
467
468
                let _ = tx
                    .send(Err("Cannot create NATS service without NATS".to_string()))
                    .await;
469
470
471
472
473
474
475
476
                return;
            };
            let description = None;
            let nats_service = match crate::component::service::build_nats_service(
                nats_client,
                &component,
                description,
            )
477
            .await
478
479
480
481
            {
                Ok(service) => service,
                Err(err) => {
                    tracing::error!(error = %err, component = service_name, "Failed to build NATS service");
482
                    let _ = tx.send(Err(format!("Failed to build NATS service: {err}"))).await;
483
484
485
                    return;
                }
            };
486

487
488
489
490
            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);
491

492
                tracing::info!("Added NATS service {service_name}");
493

494
495
496
497
498
499
500
                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?
501
            }
502
503
504

            // Signal completion - service registered successfully
            let _ = tx.send(Ok(())).await;
505
        });
506
507

        rx
508
    }
Ryan Olson's avatar
Ryan Olson committed
509
510
}

511
512
513
514
515
516
517
518
519
/// 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
520
521
#[derive(Dissolve)]
pub struct DistributedConfig {
522
    pub discovery_backend: DiscoveryBackend,
523
    pub nats_config: Option<nats::ClientOptions>,
524
    pub request_plane: RequestPlaneMode,
Ryan Olson's avatar
Ryan Olson committed
525
526
527
}

impl DistributedConfig {
528
    pub fn from_settings() -> DistributedConfig {
529
        let request_plane = RequestPlaneMode::from_env();
530
531
532
533
534
535
536
537
538
        // 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.
        // If a NATS server is configured via env, enable the client regardless of request plane.
        let nats_enabled = request_plane.is_nats()
            || std::env::var(crate::config::environment_names::nats::NATS_SERVER).is_ok();
539

540
541
542
543
        // 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());
544

545
546
547
548
549
550
551
552
553
554
555
556
557
558
        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)
            }
559
560
        };

Ryan Olson's avatar
Ryan Olson committed
561
        DistributedConfig {
562
            discovery_backend,
563
            nats_config: if nats_enabled {
564
565
566
567
568
                Some(nats::ClientOptions::default())
            } else {
                None
            },
            request_plane,
Ryan Olson's avatar
Ryan Olson committed
569
570
        }
    }
Ryan Olson's avatar
Ryan Olson committed
571
572

    pub fn for_cli() -> DistributedConfig {
573
574
575
576
        let etcd_config = etcd::ClientOptions {
            attach_lease: false,
            ..Default::default()
        };
577
        let request_plane = RequestPlaneMode::from_env();
578
579
        let nats_enabled = request_plane.is_nats()
            || std::env::var(crate::config::environment_names::nats::NATS_SERVER).is_ok();
580
        DistributedConfig {
581
            discovery_backend: DiscoveryBackend::KvStore(kv::Selector::Etcd(Box::new(etcd_config))),
582
            nats_config: if nats_enabled {
583
584
585
586
587
                Some(nats::ClientOptions::default())
            } else {
                None
            },
            request_plane,
588
589
        }
    }
590
591
592
593
594

    /// A DistributedConfig that isn't distributed, for when the frontend and backend are in the
    /// same process.
    pub fn process_local() -> DistributedConfig {
        DistributedConfig {
595
            discovery_backend: DiscoveryBackend::KvStore(kv::Selector::Memory),
596
597
598
599
600
601
            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,
        }
    }
602
603
604
605
606
}

/// Request plane transport mode configuration
///
/// This determines how requests are distributed from routers to workers:
607
/// - `Nats`: Use NATS for request distribution (legacy)
608
/// - `Http`: Use HTTP/2 for request distribution
609
/// - `Tcp`: Use raw TCP for request distribution with msgpack support (default)
610
611
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RequestPlaneMode {
612
    /// Use NATS for request plane
613
614
615
616
617
618
619
620
621
    Nats,
    /// Use HTTP/2 for request plane
    Http,
    /// Use raw TCP for request plane with msgpack support
    Tcp,
}

impl Default for RequestPlaneMode {
    fn default() -> Self {
622
        Self::Tcp
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
    }
}

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
            )),
648
        }
Ryan Olson's avatar
Ryan Olson committed
649
    }
Ryan Olson's avatar
Ryan Olson committed
650
}
651

652
653
654
655
656
657
658
659
660
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()
    }
661
662
663
664

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

667
pub mod distributed_test_utils {
668
669
    //! Common test helper functions for DistributedRuntime tests

670
    /// Helper function to create a DRT instance for integration-only tests.
671
    /// Uses from_current to leverage existing tokio runtime
672
    /// Note: Settings are read from environment variables inside DistributedRuntime::from_settings
673
    #[cfg(feature = "integration")]
674
    pub async fn create_test_drt_async() -> super::DistributedRuntime {
675
        use crate::transports::nats;
676

677
        let rt = crate::Runtime::from_current().unwrap();
678
        let config = super::DistributedConfig {
679
680
681
            discovery_backend: super::DiscoveryBackend::KvStore(
                crate::storage::kv::Selector::Memory,
            ),
682
            nats_config: Some(nats::ClientOptions::default()),
683
            request_plane: crate::distributed::RequestPlaneMode::default(),
684
685
        };
        super::DistributedRuntime::new(rt, config).await.unwrap()
686
    }
687
688
689
690
691
692
693
694
695
696

    /// 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 {
697
        use crate::transports::nats;
698
699
700

        let rt = crate::Runtime::from_current().unwrap();
        let config = super::DistributedConfig {
701
702
703
            discovery_backend: super::DiscoveryBackend::KvStore(
                crate::storage::kv::Selector::File(store_path.to_path_buf()),
            ),
704
705
706
707
708
            nats_config: Some(nats::ClientOptions::default()),
            request_plane: crate::distributed::RequestPlaneMode::default(),
        };
        super::DistributedRuntime::new(rt, config).await.unwrap()
    }
709
}
710

711
#[cfg(all(test, feature = "integration"))]
712
mod tests {
713
    use super::RequestPlaneMode;
714
715
716
717
    use super::distributed_test_utils::create_test_drt_async;

    #[tokio::test]
    async fn test_drt_uptime_after_delay_system_disabled() {
718
        use crate::config::environment_names::runtime::system as env_system;
719
        // Test uptime with system status server disabled
720
        temp_env::async_with_vars([(env_system::DYN_SYSTEM_PORT, None::<&str>)], async {
721
722
723
724
725
726
727
            // 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
728
            let uptime = drt.system_health.lock().uptime();
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
            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() {
745
        use crate::config::environment_names::runtime::system as env_system;
746
        // Test uptime with system status server enabled
747
        temp_env::async_with_vars([(env_system::DYN_SYSTEM_PORT, Some("8081"))], async {
748
749
750
751
752
753
754
            // 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
755
            let uptime = drt.system_health.lock().uptime();
756
757
758
759
760
761
762
763
764
765
766
767
768
            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;
    }
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804

    #[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");
    }
805
}