distributed.rs 28.1 KB
Newer Older
1
2
// SPDX-FileCopyrightText: Copyright (c) 2024-2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
// 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
7
use crate::pipeline::network::manager::NetworkManager;
use crate::service::{ComponentNatsServerPrometheusMetrics, ServiceClient, ServiceSet};
8
use crate::storage::kv::{self, Store as _};
9
use crate::transports::nats::DRTNatsClientPrometheusMetrics;
Ryan Olson's avatar
Ryan Olson committed
10
use crate::{
11
    component::{self, ComponentBuilder, Endpoint, Namespace},
12
    discovery::Discovery,
13
14
    metrics::PrometheusUpdateCallback,
    metrics::{MetricsHierarchy, MetricsRegistry},
Ryan Olson's avatar
Ryan Olson committed
15
16
    transports::{etcd, nats, tcp},
};
17
use crate::{discovery, system_status_server, transports};
Ryan Olson's avatar
Ryan Olson committed
18

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

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

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

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

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

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

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

57
58
59
60
    // 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>>>,

61
62
63
64
65
66
67
    // 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,

68
    instance_sources: Arc<tokio::sync::Mutex<InstanceMap>>,
69
70
71
72
73
74
75
76

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

    // This hierarchy's own metrics registry
    metrics_registry: MetricsRegistry,
}

77
impl MetricsHierarchy for DistributedRuntime {
78
79
80
81
    fn basename(&self) -> String {
        "".to_string() // drt has no basename. Basename only begins with the Namespace.
    }

82
83
84
85
86
87
    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
88
89
90
    }
}

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

        let runtime_clone = runtime.clone();

103
        let store = match selected_kv_store {
104
            kv::Selector::Etcd(etcd_config) => {
105
106
107
108
                let etcd_client = etcd::Client::new(*etcd_config, runtime_clone).await.inspect_err(|err|
                    // The returned error doesn't show because of a dropped runtime error, so
                    // log it first.
                    tracing::error!(%err, "Could not connect to etcd. Pass `--store-kv ..` to use a different backend or start etcd."))?;
109
                kv::Manager::etcd(etcd_client)
110
            }
111
112
            kv::Selector::File(root) => kv::Manager::file(root),
            kv::Selector::Memory => kv::Manager::memory(),
113
        };
Ryan Olson's avatar
Ryan Olson committed
114

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

120
        // Start system status server for health and metrics if enabled in configuration
121
122
123
124
125
126
127
128
        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
        };
129
130
        let starting_health_status = config.starting_health_status.clone();
        let use_endpoint_health_status = config.use_endpoint_health_status.clone();
131
132
        let health_endpoint_path = config.system_health_path.clone();
        let live_endpoint_path = config.system_live_path.clone();
133
        let system_health = Arc::new(parking_lot::Mutex::new(SystemHealth::new(
134
135
            starting_health_status,
            use_endpoint_health_status,
136
137
            health_endpoint_path,
            live_endpoint_path,
138
        )));
139

140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
        // Initialize discovery client based on backend configuration
        let discovery_backend =
            std::env::var("DYN_DISCOVERY_BACKEND").unwrap_or_else(|_| "kv_store".to_string());

        let (discovery_client, discovery_metadata) = match discovery_backend.as_str() {
            "kubernetes" => {
                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))
            }
            _ => {
                tracing::info!("Initializing KV store discovery backend");
                use crate::discovery::KVStoreDiscovery;
                (
                    Arc::new(KVStoreDiscovery::new(
                        store.clone(),
                        runtime.primary_token(),
                    )) as Arc<dyn Discovery>,
                    None,
                )
            }
171
172
        };

173
        let component_registry = component::Registry::new();
174
175
        let nats_client_for_metrics = nats_client.clone();

176
177
178
179
180
181
182
183
        // 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,
        );

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

200
201
202
203
204
        if let Some(nats_client_for_metrics) = nats_client_for_metrics {
            let nats_client_metrics = DRTNatsClientPrometheusMetrics::new(
                &distributed_runtime,
                nats_client_for_metrics.client().clone(),
            )?;
205
            // Register a callback to update NATS client metrics on the DRT's metrics registry
206
207
208
209
210
211
212
213
            let nats_client_callback = Arc::new({
                let nats_client_clone = nats_client_metrics.clone();
                move || {
                    nats_client_clone.set_from_client_stats();
                    Ok(())
                }
            });
            distributed_runtime
214
215
                .metrics_registry
                .add_update_callback(nats_client_callback);
216
        }
217

218
219
220
221
222
223
        // Initialize the uptime gauge in SystemHealth
        distributed_runtime
            .system_health
            .lock()
            .initialize_uptime_gauge(&distributed_runtime)?;

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

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

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

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

267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
        // 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),
            }
        }

292
        Ok(distributed_runtime)
Ryan Olson's avatar
Ryan Olson committed
293
294
295
    }

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

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

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

308
309
310
311
312
313
314
315
316
317
318
    // 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()
    }

319
    pub fn connection_id(&self) -> u64 {
320
        self.discovery_client.instance_id()
Ryan Olson's avatar
Ryan Olson committed
321
322
323
324
    }

    pub fn shutdown(&self) {
        self.runtime.shutdown();
325
        self.store.shutdown();
Ryan Olson's avatar
Ryan Olson committed
326
327
328
329
    }

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

333
334
335
    /// Returns the discovery interface for service registration and discovery
    pub fn discovery(&self) -> Arc<dyn Discovery> {
        self.discovery_client.clone()
336
337
    }

338
    pub async fn tcp_server(&self) -> Result<Arc<tcp::server::TcpStreamServer>> {
Ryan Olson's avatar
Ryan Olson committed
339
340
341
342
343
        Ok(self
            .tcp_server
            .get_or_try_init(async move {
                let options = tcp::server::ServerOptions::default();
                let server = tcp::server::TcpStreamServer::new(options).await?;
344
                Ok::<_, PipelineError>(server)
Ryan Olson's avatar
Ryan Olson committed
345
346
347
348
349
            })
            .await?
            .clone())
    }

350
    /// Get the network manager
351
352
353
    ///
    /// The network manager consolidates all network configuration and provides
    /// unified access to request plane servers and clients.
354
355
    pub fn network_manager(&self) -> Arc<NetworkManager> {
        self.network_manager.clone()
356
357
358
359
360
361
362
363
364
    }

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

368
369
370
371
372
    /// 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()
373
374
    }

375
    /// An interface to store things outside of the process. Usually backed by something like etcd.
376
    /// Currently does key-value, but will grow to include whatever we need to store.
377
    pub fn store(&self) -> &kv::Manager {
378
        &self.store
379
380
    }

381
382
383
384
385
    /// How the frontend should talk to the backend.
    pub fn request_plane(&self) -> RequestPlaneMode {
        self.request_plane
    }

386
387
388
    pub fn child_token(&self) -> CancellationToken {
        self.runtime.child_token()
    }
389

390
391
392
393
    pub(crate) fn graceful_shutdown_tracker(&self) -> Arc<GracefulShutdownTracker> {
        self.runtime.graceful_shutdown_tracker()
    }

394
    pub fn instance_sources(&self) -> Arc<Mutex<InstanceMap>> {
395
396
        self.instance_sources.clone()
    }
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541

    /// 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.
    pub(crate) async fn kv_router_nats_publish(
        &self,
        subject: String,
        payload: bytes::Bytes,
    ) -> anyhow::Result<()> {
        let Some(nats_client) = self.nats_client.as_ref() else {
            anyhow::bail!("KV router's EventPublisher requires NATS");
        };
        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?)
    }

    /// Start NATS metrics service in the background to isolate the async,
    /// and because we don't need it yet.
    /// TODO: This and the things it calls should be in a nats module somewhere.
    pub fn start_stats_service(&self, component: Component) {
        let drt = self.clone();
        self.runtime().secondary().spawn(async move {
            let service_name = component.service_name();
            if let Err(err) = drt.add_stats_service(component).await {
                tracing::error!(error = %err, component = service_name, "Failed starting stats service");
            }
        });
    }

    /// Gather NATS metrics
    async fn add_stats_service(&self, component: Component) -> anyhow::Result<()> {
        let service_name = component.service_name();

        // Pre-check to save cost of creating the service, but don't hold the lock
        if self
            .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");
            return Ok(());
        }

        let Some(nats_client) = self.nats_client.as_ref() else {
            anyhow::bail!("Cannot create NATS service without NATS.");
        };
        let description = None;
        let (nats_service, stats_reg) =
            crate::component::service::build_nats_service(nats_client, &component, description)
                .await?;

        let mut guard = self.component_registry().inner.lock().await;
        if !guard.services.contains_key(&service_name) {
            // Normal case
            guard.services.insert(service_name.clone(), nats_service);
            guard.stats_handlers.insert(service_name.clone(), stats_reg);

            tracing::info!("Added NATS / stats service {service_name}");

            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?
            return Ok(());
        }

        let cancel_token = self.primary_token();
        let service_client = self
            .nats_client
            .as_ref()
            .map(|nc| ServiceClient::new(nc.clone()))
            .ok_or_else(|| {
                anyhow::anyhow!("Stats service requires NATS client to collect service metrics.")
            })?;
        // If there is another component with the same service name, this will fail.
        let component_metrics = ComponentNatsServerPrometheusMetrics::new(&component)?;

        self.runtime().secondary().spawn(nats_metrics_worker(
            cancel_token,
            service_client,
            component_metrics,
            component,
        ));
        Ok(())
    }
}

/// Add Prometheus metrics for this component's NATS service stats.
///
/// Starts a background task that periodically requests service statistics from NATS
/// and updates the corresponding Prometheus metrics. The first scrape happens immediately,
/// then subsequent scrapes occur at a fixed interval of 9.8 seconds (MAX_WAIT_MS),
/// which should be near or smaller than typical Prometheus scraping intervals to ensure
/// metrics are fresh when Prometheus collects them.
async fn nats_metrics_worker(
    cancel_token: CancellationToken,
    service_client: ServiceClient,
    component_metrics: ComponentNatsServerPrometheusMetrics,
    component: Component,
) {
    const MAX_WAIT_MS: Duration = Duration::from_millis(9800); // Should be <= Prometheus scrape interval
    let timeout = Duration::from_millis(500);
    let mut interval = tokio::time::interval(MAX_WAIT_MS);
    interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);

    let service_name = component.service_name();
    loop {
        tokio::select! {
            result = service_client.collect_services(&service_name, timeout) => {
                match result {
                    Ok(service_set) => {
                        component_metrics.update_from_service_set(&service_set);
                    }
                    Err(err) => {
                        tracing::error!("Background scrape failed for {service_name}: {err}",);
                        component_metrics.reset_to_zeros();
                    }
                }
            }
            _ = cancel_token.cancelled() => {
                tracing::trace!("nats_metrics_worker stopped");
                break;
            }
        }

        interval.tick().await;
    }
Ryan Olson's avatar
Ryan Olson committed
542
543
544
545
}

#[derive(Dissolve)]
pub struct DistributedConfig {
546
    pub store_backend: kv::Selector,
547
    pub nats_config: Option<nats::ClientOptions>,
548
    pub request_plane: RequestPlaneMode,
Ryan Olson's avatar
Ryan Olson committed
549
550
551
}

impl DistributedConfig {
552
    pub fn from_settings() -> DistributedConfig {
553
        let request_plane = RequestPlaneMode::from_env();
Ryan Olson's avatar
Ryan Olson committed
554
        DistributedConfig {
555
            store_backend: kv::Selector::Etcd(Box::default()),
556
557
558
559
560
561
            nats_config: if request_plane.is_nats() {
                Some(nats::ClientOptions::default())
            } else {
                None
            },
            request_plane,
Ryan Olson's avatar
Ryan Olson committed
562
563
        }
    }
Ryan Olson's avatar
Ryan Olson committed
564
565

    pub fn for_cli() -> DistributedConfig {
566
567
568
569
        let etcd_config = etcd::ClientOptions {
            attach_lease: false,
            ..Default::default()
        };
570
        let request_plane = RequestPlaneMode::from_env();
571
        DistributedConfig {
572
            store_backend: kv::Selector::Etcd(Box::new(etcd_config)),
573
574
575
576
577
578
            nats_config: if request_plane.is_nats() {
                Some(nats::ClientOptions::default())
            } else {
                None
            },
            request_plane,
579
580
        }
    }
581
582
583
584
585

    /// A DistributedConfig that isn't distributed, for when the frontend and backend are in the
    /// same process.
    pub fn process_local() -> DistributedConfig {
        DistributedConfig {
586
            store_backend: kv::Selector::Memory,
587
588
589
590
591
592
            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,
        }
    }
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
}

/// Request plane transport mode configuration
///
/// This determines how requests are distributed from routers to workers:
/// - `Nats`: Use NATS for request distribution (default, legacy)
/// - `Http`: Use HTTP/2 for request distribution
/// - `Tcp`: Use raw TCP for request distribution with msgpack support
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RequestPlaneMode {
    /// Use NATS for request plane (default for backward compatibility)
    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 {
        Self::Nats
    }
}

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
            )),
639
        }
Ryan Olson's avatar
Ryan Olson committed
640
    }
Ryan Olson's avatar
Ryan Olson committed
641
}
642

643
644
645
646
647
648
649
650
651
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()
    }
652
653
654
655

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

658
pub mod distributed_test_utils {
659
660
    //! Common test helper functions for DistributedRuntime tests

661
    /// Helper function to create a DRT instance for integration-only tests.
662
    /// Uses from_current to leverage existing tokio runtime
663
    /// Note: Settings are read from environment variables inside DistributedRuntime::from_settings
664
    #[cfg(feature = "integration")]
665
    pub async fn create_test_drt_async() -> super::DistributedRuntime {
666
        use crate::{storage::kv, transports::nats};
667

668
        let rt = crate::Runtime::from_current().unwrap();
669
        let config = super::DistributedConfig {
670
            store_backend: kv::Selector::Memory,
671
            nats_config: Some(nats::ClientOptions::default()),
672
            request_plane: crate::distributed::RequestPlaneMode::default(),
673
674
        };
        super::DistributedRuntime::new(rt, config).await.unwrap()
675
676
    }
}
677

678
#[cfg(all(test, feature = "integration"))]
679
mod tests {
680
    use super::RequestPlaneMode;
681
682
683
684
    use super::distributed_test_utils::create_test_drt_async;

    #[tokio::test]
    async fn test_drt_uptime_after_delay_system_disabled() {
685
        use crate::config::environment_names::runtime::system as env_system;
686
        // Test uptime with system status server disabled
687
        temp_env::async_with_vars([(env_system::DYN_SYSTEM_PORT, None::<&str>)], async {
688
689
690
691
692
693
694
            // 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
695
            let uptime = drt.system_health.lock().uptime();
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
            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() {
712
        use crate::config::environment_names::runtime::system as env_system;
713
        // Test uptime with system status server enabled
714
        temp_env::async_with_vars([(env_system::DYN_SYSTEM_PORT, Some("8081"))], async {
715
716
717
718
719
720
721
            // 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
722
            let uptime = drt.system_health.lock().uptime();
723
724
725
726
727
728
729
730
731
732
733
734
735
            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;
    }
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771

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