"docs/observability/metrics.md" did not exist on "1df620b4d0121970c5b85ee91fa34c0bb89a9aae"
distributed.rs 29.3 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::{self, Store as _};
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
    store: kv::Manager,
48
    network_manager: Arc<NetworkManager>,
49
50
    tcp_server: Arc<OnceCell<Arc<transports::tcp::server::TcpStreamServer>>>,
    system_status_server: Arc<OnceLock<Arc<system_status_server::SystemStatusServerInfo>>>,
51
    request_plane: RequestPlaneMode,
52
53
54
55

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

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

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

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

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

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

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

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

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

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

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

        let runtime_clone = runtime.clone();

108
        let store = match selected_kv_store {
109
            kv::Selector::Etcd(etcd_config) => {
110
111
112
113
                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."))?;
114
                kv::Manager::etcd(etcd_client)
115
            }
116
            kv::Selector::File(root) => kv::Manager::file(runtime.primary_token(), root),
117
            kv::Selector::Memory => kv::Manager::memory(),
118
        };
Ryan Olson's avatar
Ryan Olson committed
119

120
121
122
123
        let nats_client = match nats_config {
            Some(nc) => Some(nc.connect().await?),
            None => None,
        };
Ryan Olson's avatar
Ryan Olson committed
124

125
        // Start system status server for health and metrics if enabled in configuration
126
127
128
129
130
131
132
133
        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
        };
134
135
        let starting_health_status = config.starting_health_status.clone();
        let use_endpoint_health_status = config.use_endpoint_health_status.clone();
136
137
        let health_endpoint_path = config.system_health_path.clone();
        let live_endpoint_path = config.system_live_path.clone();
138
        let system_health = Arc::new(parking_lot::Mutex::new(SystemHealth::new(
139
140
            starting_health_status,
            use_endpoint_health_status,
141
142
            health_endpoint_path,
            live_endpoint_path,
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
171
172
173
174
175
        // 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,
                )
            }
176
177
        };

178
        let component_registry = component::Registry::new();
179

180
181
182
183
184
185
186
187
        // 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,
        );

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

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

212
        // Handle system status server initialization
213
        if let Some(cancel_token) = cancel_token {
214
            // System server is enabled - start both the state and HTTP server
215
            let host = config.system_host.clone();
216
            let port = config.system_port as u16;
217

218
            // Start system status server (it creates SystemStatusState internally)
219
            match crate::system_status_server::spawn_system_status_server(
220
221
222
223
                &host,
                port,
                cancel_token,
                Arc::new(distributed_runtime.clone()),
224
                distributed_runtime.discovery_metadata.clone(),
225
226
227
            )
            .await
            {
228
                Ok((addr, handle)) => {
229
                    tracing::info!("System status server started successfully on {}", addr);
230

231
232
233
234
235
236
                    // Store system status server information
                    let system_status_server_info =
                        crate::system_status_server::SystemStatusServerInfo::new(
                            addr,
                            Some(handle),
                        );
237

238
                    // Initialize the system_status_server field
239
                    distributed_runtime
240
241
242
                        .system_status_server
                        .set(Arc::new(system_status_server_info))
                        .expect("System status server info should only be set once");
243
244
                }
                Err(e) => {
245
                    tracing::error!("System status server startup failed: {}", e);
246
                }
247
            }
248
        } else {
249
            // System server HTTP is disabled, but uptime metrics are still being tracked via SystemHealth
250
251
252
            tracing::debug!(
                "System status server HTTP endpoints disabled, but uptime metrics are being tracked"
            );
253
254
        }

255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
        // 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),
            }
        }

280
        Ok(distributed_runtime)
Ryan Olson's avatar
Ryan Olson committed
281
282
283
    }

    pub async fn from_settings(runtime: Runtime) -> Result<Self> {
284
        let config = DistributedConfig::from_settings();
Ryan Olson's avatar
Ryan Olson committed
285
286
287
288
289
290
291
        Self::new(runtime, config).await
    }

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

292
293
294
295
    pub fn primary_token(&self) -> CancellationToken {
        self.runtime.primary_token()
    }

296
297
298
299
300
301
302
303
304
305
306
    // 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()
    }

307
308
309
310
311
312
313
    /// 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
    }

314
315
316
317
318
    /// Get the engine route registry for registering custom /engine/* routes
    pub fn engine_routes(&self) -> &crate::engine_routes::EngineRouteRegistry {
        &self.engine_routes
    }

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

    /// 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.
400
401
402
    ///
    /// 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.
403
    pub async fn kv_router_nats_publish(
404
405
406
407
408
        &self,
        subject: String,
        payload: bytes::Bytes,
    ) -> anyhow::Result<()> {
        let Some(nats_client) = self.nats_client.as_ref() else {
409
410
411
            // NATS not available - this is expected in approximate mode (--no-kv-events)
            tracing::trace!("Skipping NATS publish (NATS not configured): {}", subject);
            return Ok(());
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
        };
        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?)
    }

428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
    /// 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)
    }

447
448
449
    /// 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.
450
451
452
453
454
455
456
457
458
459
    ///
    /// 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);

460
461
462
463
        let drt = self.clone();
        self.runtime().secondary().spawn(async move {
            let service_name = component.service_name();

464
465
466
467
468
469
470
471
472
473
474
475
            // 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");
476
477
                // Signal success - service already exists
                let _ = tx.send(Ok(())).await;
478
479
                return;
            }
480

481
482
            let Some(nats_client) = drt.nats_client.as_ref() else {
                tracing::error!("Cannot create NATS service without NATS.");
483
484
485
                let _ = tx
                    .send(Err("Cannot create NATS service without NATS".to_string()))
                    .await;
486
487
488
489
490
491
492
493
                return;
            };
            let description = None;
            let nats_service = match crate::component::service::build_nats_service(
                nats_client,
                &component,
                description,
            )
494
            .await
495
496
497
498
            {
                Ok(service) => service,
                Err(err) => {
                    tracing::error!(error = %err, component = service_name, "Failed to build NATS service");
499
                    let _ = tx.send(Err(format!("Failed to build NATS service: {err}"))).await;
500
501
502
                    return;
                }
            };
503

504
505
506
507
            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);
508

509
                tracing::info!("Added NATS service {service_name}");
510

511
512
513
514
515
516
517
                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?
518
            }
519
520
521

            // Signal completion - service registered successfully
            let _ = tx.send(Ok(())).await;
522
        });
523
524

        rx
525
    }
Ryan Olson's avatar
Ryan Olson committed
526
527
528
529
}

#[derive(Dissolve)]
pub struct DistributedConfig {
530
    pub store_backend: kv::Selector,
531
    pub nats_config: Option<nats::ClientOptions>,
532
    pub request_plane: RequestPlaneMode,
Ryan Olson's avatar
Ryan Olson committed
533
534
535
}

impl DistributedConfig {
536
    pub fn from_settings() -> DistributedConfig {
537
        let request_plane = RequestPlaneMode::from_env();
538
539
540
541
542
543
544
545
546
        // 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();
Ryan Olson's avatar
Ryan Olson committed
547
        DistributedConfig {
548
            store_backend: kv::Selector::Etcd(Box::default()),
549
            nats_config: if nats_enabled {
550
551
552
553
554
                Some(nats::ClientOptions::default())
            } else {
                None
            },
            request_plane,
Ryan Olson's avatar
Ryan Olson committed
555
556
        }
    }
Ryan Olson's avatar
Ryan Olson committed
557
558

    pub fn for_cli() -> DistributedConfig {
559
560
561
562
        let etcd_config = etcd::ClientOptions {
            attach_lease: false,
            ..Default::default()
        };
563
        let request_plane = RequestPlaneMode::from_env();
564
565
        let nats_enabled = request_plane.is_nats()
            || std::env::var(crate::config::environment_names::nats::NATS_SERVER).is_ok();
566
        DistributedConfig {
567
            store_backend: kv::Selector::Etcd(Box::new(etcd_config)),
568
            nats_config: if nats_enabled {
569
570
571
572
573
                Some(nats::ClientOptions::default())
            } else {
                None
            },
            request_plane,
574
575
        }
    }
576
577
578
579
580

    /// A DistributedConfig that isn't distributed, for when the frontend and backend are in the
    /// same process.
    pub fn process_local() -> DistributedConfig {
        DistributedConfig {
581
            store_backend: kv::Selector::Memory,
582
583
584
585
586
587
            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,
        }
    }
588
589
590
591
592
}

/// Request plane transport mode configuration
///
/// This determines how requests are distributed from routers to workers:
593
/// - `Nats`: Use NATS for request distribution (legacy)
594
/// - `Http`: Use HTTP/2 for request distribution
595
/// - `Tcp`: Use raw TCP for request distribution with msgpack support (default)
596
597
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RequestPlaneMode {
598
    /// Use NATS for request plane
599
600
601
602
603
604
605
606
607
    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 {
608
        Self::Tcp
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
    }
}

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
            )),
634
        }
Ryan Olson's avatar
Ryan Olson committed
635
    }
Ryan Olson's avatar
Ryan Olson committed
636
}
637

638
639
640
641
642
643
644
645
646
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()
    }
647
648
649
650

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

653
pub mod distributed_test_utils {
654
655
    //! Common test helper functions for DistributedRuntime tests

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

663
        let rt = crate::Runtime::from_current().unwrap();
664
        let config = super::DistributedConfig {
665
            store_backend: kv::Selector::Memory,
666
            nats_config: Some(nats::ClientOptions::default()),
667
            request_plane: crate::distributed::RequestPlaneMode::default(),
668
669
        };
        super::DistributedRuntime::new(rt, config).await.unwrap()
670
    }
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690

    /// 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 {
        use crate::{storage::kv, transports::nats};

        let rt = crate::Runtime::from_current().unwrap();
        let config = super::DistributedConfig {
            store_backend: kv::Selector::File(store_path.to_path_buf()),
            nats_config: Some(nats::ClientOptions::default()),
            request_plane: crate::distributed::RequestPlaneMode::default(),
        };
        super::DistributedRuntime::new(rt, config).await.unwrap()
    }
691
}
692

693
#[cfg(all(test, feature = "integration"))]
694
mod tests {
695
    use super::RequestPlaneMode;
696
697
698
699
    use super::distributed_test_utils::create_test_drt_async;

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

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