system_status_server.rs 29.7 KB
Newer Older
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
// SPDX-FileCopyrightText: Copyright (c) 2024-2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
// SPDX-License-Identifier: Apache-2.0
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

16
use crate::config::HealthStatus;
17
use crate::logging::make_request_span;
18
use crate::logging::TraceParent;
19
20
use crate::metrics::MetricsRegistry;
use crate::traits::DistributedRuntimeProvider;
21
use axum::{body, http::StatusCode, response::IntoResponse, routing::get, Router};
22
23
use serde_json::json;
use std::collections::HashMap;
24
use std::sync::Arc;
25
26
use std::sync::OnceLock;
use std::time::Instant;
27
use tokio::{net::TcpListener, task::JoinHandle};
28
use tokio_util::sync::CancellationToken;
29
30
use tower_http::trace::DefaultMakeSpan;
use tower_http::trace::TraceLayer;
31

32
/// System status server information containing socket address and handle
33
#[derive(Debug)]
34
pub struct SystemStatusServerInfo {
35
36
37
38
    pub socket_addr: std::net::SocketAddr,
    pub handle: Option<Arc<JoinHandle<()>>>,
}

39
impl SystemStatusServerInfo {
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
    pub fn new(socket_addr: std::net::SocketAddr, handle: Option<JoinHandle<()>>) -> Self {
        Self {
            socket_addr,
            handle: handle.map(Arc::new),
        }
    }

    pub fn address(&self) -> String {
        self.socket_addr.to_string()
    }

    pub fn hostname(&self) -> String {
        self.socket_addr.ip().to_string()
    }

    pub fn port(&self) -> u16 {
        self.socket_addr.port()
    }
}

60
impl Clone for SystemStatusServerInfo {
61
62
63
64
65
66
67
68
    fn clone(&self) -> Self {
        Self {
            socket_addr: self.socket_addr,
            handle: self.handle.clone(),
        }
    }
}

69
70
/// System status server state containing metrics and uptime tracking
pub struct SystemStatusState {
71
72
73
    // global drt registry is for printing out the entire Prometheus format output
    root_drt: Arc<crate::DistributedRuntime>,
    start_time: OnceLock<Instant>,
74
    uptime_gauge: prometheus::Gauge,
75
76
}

77
78
impl SystemStatusState {
    /// Create new system status server state with the provided metrics registry
79
    pub fn new(drt: Arc<crate::DistributedRuntime>) -> anyhow::Result<Self> {
80
        // Note: This metric is created at the DRT level (no namespace), so it will be prefixed with "dynamo_component_"
81
        let uptime_gauge = drt.as_ref().create_gauge(
82
            "uptime_seconds",
83
84
85
86
87
88
89
90
91
92
            "Total uptime of the DistributedRuntime in seconds",
            &[],
        )?;
        let state = Self {
            root_drt: drt,
            start_time: OnceLock::new(),
            uptime_gauge,
        };
        Ok(state)
    }
93

94
95
96
97
98
99
    /// Initialize the start time (can only be called once)
    pub fn initialize_start_time(&self) -> Result<(), &'static str> {
        self.start_time
            .set(Instant::now())
            .map_err(|_| "Start time already initialized")
    }
100

101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
    pub fn uptime(&self) -> Result<std::time::Duration, &'static str> {
        self.start_time
            .get()
            .ok_or("Start time not initialized")
            .map(|start_time| start_time.elapsed())
    }

    /// Get a reference to the distributed runtime
    pub fn drt(&self) -> &crate::DistributedRuntime {
        &self.root_drt
    }

    /// Update the uptime gauge with current value
    pub fn update_uptime_gauge(&self) {
        if let Ok(uptime) = self.uptime() {
            let uptime_seconds = uptime.as_secs_f64();
            self.uptime_gauge.set(uptime_seconds);
        } else {
            tracing::warn!("Failed to update uptime gauge: start time not initialized");
        }
121
122
123
    }
}

124
125
/// Start system status server with metrics support
pub async fn spawn_system_status_server(
126
127
128
129
    host: &str,
    port: u16,
    cancel_token: CancellationToken,
    drt: Arc<crate::DistributedRuntime>,
130
) -> anyhow::Result<(std::net::SocketAddr, tokio::task::JoinHandle<()>)> {
131
132
    // Create system status server state with the provided metrics registry
    let server_state = Arc::new(SystemStatusState::new(drt)?);
133
134
135
136
137
138
139
140
141
142
143
144
145
146
    let health_path = server_state
        .drt()
        .system_health
        .lock()
        .unwrap()
        .health_path
        .clone();
    let live_path = server_state
        .drt()
        .system_health
        .lock()
        .unwrap()
        .live_path
        .clone();
147

148
149
150
151
152
    // Initialize the start time
    server_state
        .initialize_start_time()
        .map_err(|e| anyhow::anyhow!("Failed to initialize start time: {}", e))?;

153
    let app = Router::new()
154
        .route(
155
            &health_path,
156
157
            get({
                let state = Arc::clone(&server_state);
158
                move || health_handler(state)
159
160
161
            }),
        )
        .route(
162
            &live_path,
163
164
            get({
                let state = Arc::clone(&server_state);
165
                move || health_handler(state)
166
167
            }),
        )
168
169
170
171
        .route(
            "/metrics",
            get({
                let state = Arc::clone(&server_state);
172
                move || metrics_handler(state)
173
            }),
174
        )
175
176
177
178
179
        .fallback(|| async {
            tracing::info!("[fallback handler] called");
            (StatusCode::NOT_FOUND, "Route not found").into_response()
        })
        .layer(TraceLayer::new_for_http().make_span_with(make_request_span));
180
181

    let address = format!("{}:{}", host, port);
182
    tracing::info!("[spawn_system_status_server] binding to: {}", address);
183
184
185
186
187

    let listener = match TcpListener::bind(&address).await {
        Ok(listener) => {
            // get the actual address and port, print in debug level
            let actual_address = listener.local_addr()?;
188
            tracing::info!(
189
                "[spawn_system_status_server] system status server bound to: {}",
190
191
192
                actual_address
            );
            (listener, actual_address)
193
194
195
196
197
198
        }
        Err(e) => {
            tracing::error!("Failed to bind to address {}: {}", address, e);
            return Err(anyhow::anyhow!("Failed to bind to address: {}", e));
        }
    };
199
    let (listener, actual_address) = listener;
200
201

    let observer = cancel_token.child_token();
202
203
204
205
206
207
    // Spawn the server in the background and return the handle
    let handle = tokio::spawn(async move {
        if let Err(e) = axum::serve(listener, app)
            .with_graceful_shutdown(observer.cancelled_owned())
            .await
        {
208
            tracing::error!("System status server error: {}", e);
209
210
        }
    });
211

212
    Ok((actual_address, handle))
213
214
}

215
/// Health handler
216
217
#[tracing::instrument(skip_all, level = "trace")]
async fn health_handler(state: Arc<SystemStatusState>) -> impl IntoResponse {
218
219
220
221
222
223
    let (mut healthy, endpoints) = state
        .drt()
        .system_health
        .lock()
        .unwrap()
        .get_health_status();
224
225
    let uptime = match state.uptime() {
        Ok(uptime_state) => Some(uptime_state),
226
227
        Err(e) => {
            tracing::error!("Failed to get uptime: {}", e);
228
229
            healthy = false;
            None
230
        }
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
    };

    let healthy_string = if healthy { "ready" } else { "notready" };
    let status_code = if healthy {
        StatusCode::OK
    } else {
        StatusCode::SERVICE_UNAVAILABLE
    };

    let response = json!({
        "status": healthy_string,
        "uptime": uptime,
        "endpoints": endpoints
    });

    tracing::trace!("Response {}", response.to_string());

    (status_code, response.to_string())
249
}
250
251

/// Metrics handler with DistributedRuntime uptime
252
253
#[tracing::instrument(skip_all, level = "trace")]
async fn metrics_handler(state: Arc<SystemStatusState>) -> impl IntoResponse {
254
    // Update the uptime gauge with current value
255
    state.update_uptime_gauge();
256

257
258
259
260
261
262
263
264
265
266
267
268
    // Execute all the callbacks starting at the DistributedRuntime level
    assert!(state.drt().basename() == "");
    let callback_results = state
        .drt()
        .execute_metrics_callbacks(&state.drt().hierarchy());
    for result in callback_results {
        if let Err(e) = result {
            tracing::error!("Error executing metrics callback: {}", e);
        }
    }

    // Get all metrics from DistributedRuntime (top-level)
269
270
    match state.drt().prometheus_metrics_fmt() {
        Ok(response) => (StatusCode::OK, response),
271
        Err(e) => {
272
            tracing::error!("Failed to get metrics from registry: {}", e);
273
274
            (
                StatusCode::INTERNAL_SERVER_ERROR,
275
                "Failed to get metrics".to_string(),
276
277
278
279
280
            )
        }
    }
}

281
282
// Regular tests: cargo test system_status_server --lib
// Integration tests: cargo test system_status_server --lib --features integration
283
284

#[cfg(test)]
285
286
/// Helper function to create a DRT instance for basic unit tests
/// Uses from_current to leverage existing tokio runtime without environment configuration
287
288
289
290
291
292
293
async fn create_test_drt_async() -> crate::DistributedRuntime {
    let rt = crate::Runtime::from_current().unwrap();
    crate::DistributedRuntime::from_settings_without_discovery(rt)
        .await
        .unwrap()
}

294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
#[cfg(test)]
/// Helper function to create a DRT instance for integration tests
/// Uses spawn_blocking to create runtime safely without ownership issues
/// Enables system status server for integration testing
/// Note: This function uses environment variables to configure and create the DistributedRuntime.
async fn create_test_drt_with_settings_async() -> crate::DistributedRuntime {
    // Create runtime in blocking context where it can be safely dropped
    let handle = tokio::task::spawn_blocking(|| {
        // Load configuration from environment/settings
        let config = crate::config::RuntimeConfig::from_settings().unwrap();

        // Create runtime with the configuration and extract handle
        let runtime = config.create_runtime().unwrap();
        let handle = runtime.handle().clone();

        // Runtime will be automatically dropped when it goes out of scope
        handle
    })
    .await
    .unwrap();

    // Create Runtime using external handle (no ownership)
    let rt = crate::Runtime::from_handle(handle).unwrap();
    crate::DistributedRuntime::from_settings_without_discovery(rt)
        .await
        .unwrap()
}

322
323
324
#[cfg(test)]
mod tests {
    use super::*;
325
    use crate::logging::tests::load_log;
326
    use crate::metrics::MetricsRegistry;
327
328
329
    use anyhow::{anyhow, Result};
    use chrono::{DateTime, Utc};
    use jsonschema::{Draft, JSONSchema};
330
    use rstest::rstest;
331
332
333
    use serde_json::Value;
    use std::fs::File;
    use std::io::{BufRead, BufReader};
334
    use std::sync::Arc;
335
    use stdio_override::*;
336
337
    use tokio::time::{sleep, Duration};

338
    // This is a basic test to verify the HTTP server is working before testing other more complicated tests
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
    #[tokio::test]
    async fn test_http_server_lifecycle() {
        let cancel_token = CancellationToken::new();
        let cancel_token_for_server = cancel_token.clone();

        // Test basic HTTP server lifecycle without DistributedRuntime
        let app = Router::new().route("/test", get(|| async { (StatusCode::OK, "test") }));

        // start HTTP server
        let server_handle = tokio::spawn(async move {
            let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
            let _ = axum::serve(listener, app)
                .with_graceful_shutdown(cancel_token_for_server.cancelled_owned())
                .await;
        });

355
        // server starts immediately, no need to wait
356
357
358
359
360
361
362
363
364
365
366
367

        // cancel token
        cancel_token.cancel();

        // wait for the server to shut down
        let result = tokio::time::timeout(Duration::from_secs(5), server_handle).await;
        assert!(
            result.is_ok(),
            "HTTP server should shut down when cancel token is cancelled"
        );
    }

368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
    #[cfg(feature = "integration")]
    #[tokio::test]
    async fn test_uptime_without_initialization() {
        // Test that uptime returns an error if start time is not initialized
        temp_env::async_with_vars([("DYN_SYSTEM_ENABLED", Some("false"))], async {
            let drt = create_test_drt_async().await;
            let system_status = SystemStatusState::new(Arc::new(drt)).unwrap();

            // This should return an error because start time is not initialized
            let result = system_status.uptime();
            assert!(result.is_err());
            assert_eq!(result.unwrap_err(), "Start time not initialized");
        })
        .await;
    }

384
    #[cfg(feature = "integration")]
385
    #[tokio::test]
386
387
    async fn test_runtime_metrics_initialization_and_namespace() {
        // Test that metrics have correct namespace
388
389
390
        temp_env::async_with_vars([("DYN_SYSTEM_ENABLED", Some("false"))], async {
            let drt = create_test_drt_async().await;
            let system_status = SystemStatusState::new(Arc::new(drt)).unwrap();
391

392
393
            // Initialize start time
            system_status.initialize_start_time().unwrap();
394

395
            system_status.uptime_gauge.set(42.0);
396

397
398
            let response = system_status.drt().prometheus_metrics_fmt().unwrap();
            println!("Full metrics response:\n{}", response);
399

400
401
            // Filter out NATS client metrics for comparison
            use crate::metrics::prometheus_names::nats as nats_metrics;
402

403
404
405
406
407
            let filtered_response: String = response
                .lines()
                .filter(|line| !line.contains(nats_metrics::PREFIX))
                .collect::<Vec<_>>()
                .join("\n");
408

409
410
411
412
413
414
415
            let expected = "\
# HELP dynamo_component_uptime_seconds Total uptime of the DistributedRuntime in seconds
# TYPE dynamo_component_uptime_seconds gauge
dynamo_component_uptime_seconds 42";
            assert_eq!(filtered_response, expected);
        })
        .await;
416
417
    }

418
    #[cfg(feature = "integration")]
419
    #[tokio::test]
420
421
    async fn test_start_time_initialization() {
        // Test that start time can only be initialized once
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
        temp_env::async_with_vars([("DYN_SYSTEM_ENABLED", Some("false"))], async {
            let drt = create_test_drt_async().await;
            let system_status = SystemStatusState::new(Arc::new(drt)).unwrap();

            // First initialization should succeed
            assert!(system_status.initialize_start_time().is_ok());

            // Second initialization should fail
            assert!(system_status.initialize_start_time().is_err());

            // Sleep for 100ms and verify uptime increases
            tokio::time::sleep(std::time::Duration::from_millis(100)).await;
            let uptime_after_sleep = system_status.uptime().unwrap();
            assert!(
                uptime_after_sleep >= std::time::Duration::from_millis(100),
                "Uptime should be at least 100ms after sleep, got: {:?}",
                uptime_after_sleep
            );
440

441
442
443
            // If we get here, uptime calculation works correctly
        })
        .await;
444
    }
445

446
447
448
449
450
451
    /// This test verifies the health and liveness endpoints of the system status server.
    /// It checks that the endpoints respond with the correct HTTP status codes and bodies
    /// based on the initial health status and any custom endpoint paths provided via environment variables.
    /// The test is parameterized using multiple #[case] attributes to cover various scenarios,
    /// including different initial health states ("ready" and "notready"), default and custom endpoint paths,
    /// and expected response codes and bodies.
452
    #[rstest]
453
454
455
456
457
458
459
460
461
462
463
    #[case("ready", 200, "ready", None, None, 3)]
    #[case("notready", 503, "notready", None, None, 3)]
    #[case("ready", 200, "ready", Some("/custom/health"), Some("/custom/live"), 5)]
    #[case(
        "notready",
        503,
        "notready",
        Some("/custom/health"),
        Some("/custom/live"),
        5
    )]
464
    #[tokio::test]
465
    #[cfg(feature = "integration")]
466
467
468
469
    async fn test_health_endpoints(
        #[case] starting_health_status: &'static str,
        #[case] expected_status: u16,
        #[case] expected_body: &'static str,
470
471
472
        #[case] custom_health_path: Option<&'static str>,
        #[case] custom_live_path: Option<&'static str>,
        #[case] expected_num_tests: usize,
473
474
475
476
477
478
479
    ) {
        use std::sync::Arc;
        // use tokio::io::{AsyncReadExt, AsyncWriteExt};
        // use reqwest for HTTP requests

        // Closure call is needed here to satisfy async_with_vars

480
481
        crate::logging::init();

482
483
        #[allow(clippy::redundant_closure_call)]
        temp_env::async_with_vars(
484
            [
485
486
                ("DYN_SYSTEM_ENABLED", Some("true")),
                ("DYN_SYSTEM_PORT", Some("0")),
487
488
489
490
491
492
493
                (
                    "DYN_SYSTEM_STARTING_HEALTH_STATUS",
                    Some(starting_health_status),
                ),
                ("DYN_SYSTEM_HEALTH_PATH", custom_health_path),
                ("DYN_SYSTEM_LIVE_PATH", custom_live_path),
            ],
494
            (async || {
495
496
497
498
499
500
501
502
                let drt = Arc::new(create_test_drt_with_settings_async().await);

                // Get system status server info from DRT (instead of manually spawning)
                let system_info = drt
                    .system_status_server_info()
                    .expect("System status server should be started by DRT");
                let addr = system_info.socket_addr;

503
                let client = reqwest::Client::new();
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526

                // Prepare test cases
                let mut test_cases = vec![];
                if custom_health_path.is_none() {
                    // When using default paths, test the default paths
                    test_cases.push(("/health", expected_status, expected_body));
                } else {
                    // When using custom paths, default paths should not exist
                    test_cases.push(("/health", 404, "Route not found"));
                    test_cases.push((custom_health_path.unwrap(), expected_status, expected_body));
                }
                if custom_live_path.is_none() {
                    // When using default paths, test the default paths
                    test_cases.push(("/live", expected_status, expected_body));
                } else {
                    // When using custom paths, default paths should not exist
                    test_cases.push(("/live", 404, "Route not found"));
                    test_cases.push((custom_live_path.unwrap(), expected_status, expected_body));
                }
                test_cases.push(("/someRandomPathNotFoundHere", 404, "Route not found"));
                assert_eq!(test_cases.len(), expected_num_tests);

                for (path, expect_status, expect_body) in test_cases {
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
                    println!("[test] Sending request to {}", path);
                    let url = format!("http://{}{}", addr, path);
                    let response = client.get(&url).send().await.unwrap();
                    let status = response.status();
                    let body = response.text().await.unwrap();
                    println!(
                        "[test] Response for {}: status={}, body={:?}",
                        path, status, body
                    );
                    assert_eq!(
                        status, expect_status,
                        "Response: status={}, body={:?}",
                        status, body
                    );
                    assert!(
                        body.contains(expect_body),
                        "Response: status={}, body={:?}",
                        status,
                        body
                    );
                }
            })(),
        )
        .await;
    }

553
554
555
556
557
558
559
560
561
562
    #[tokio::test]
    #[cfg(feature = "integration")]
    async fn test_health_endpoint_tracing() -> Result<()> {
        use std::sync::Arc;

        // Closure call is needed here to satisfy async_with_vars

        #[allow(clippy::redundant_closure_call)]
        let _ = temp_env::async_with_vars(
            [
563
564
                ("DYN_SYSTEM_ENABLED", Some("true")),
                ("DYN_SYSTEM_PORT", Some("0")),
565
566
567
568
569
570
571
572
573
574
                ("DYN_SYSTEM_STARTING_HEALTH_STATUS", Some("ready")),
                ("DYN_LOGGING_JSONL", Some("1")),
                ("DYN_LOG", Some("trace")),
            ],
            (async || {
                // TODO Add proper testing for
                // trace id and parent id

                crate::logging::init();

575
576
577
578
579
580
581
                let drt = Arc::new(create_test_drt_with_settings_async().await);

                // Get system status server info from DRT (instead of manually spawning)
                let system_info = drt
                    .system_status_server_info()
                    .expect("System status server should be started by DRT");
                let addr = system_info.socket_addr;
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
                let client = reqwest::Client::new();
                for path in [("/health"), ("/live"), ("/someRandomPathNotFoundHere")] {
                    let traceparent_value =
                        "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01";
                    let tracestate_value = "vendor1=opaqueValue1,vendor2=opaqueValue2";
                    let mut headers = reqwest::header::HeaderMap::new();
                    headers.insert(
                        reqwest::header::HeaderName::from_static("traceparent"),
                        reqwest::header::HeaderValue::from_str(traceparent_value)?,
                    );
                    headers.insert(
                        reqwest::header::HeaderName::from_static("tracestate"),
                        reqwest::header::HeaderValue::from_str(tracestate_value)?,
                    );
                    let url = format!("http://{}{}", addr, path);
                    let response = client.get(&url).headers(headers).send().await.unwrap();
                    let status = response.status();
                    let body = response.text().await.unwrap();
                    tracing::info!(body = body, status = status.to_string());
                }

                Ok::<(), anyhow::Error>(())
            })(),
        )
        .await;
        Ok(())
    }

610
611
    #[cfg(feature = "integration")]
    #[tokio::test]
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
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
    async fn test_health_endpoint_with_changing_health_status() {
        // Test health endpoint starts in not ready status, then becomes ready
        // when endpoints are created (DYN_SYSTEM_USE_ENDPOINT_HEALTH_STATUS=generate)
        const ENDPOINT_NAME: &str = "generate";
        const ENDPOINT_HEALTH_CONFIG: &str = "[\"generate\"]";
        temp_env::async_with_vars(
            [
                ("DYN_SYSTEM_ENABLED", Some("true")),
                ("DYN_SYSTEM_PORT", Some("0")),
                ("DYN_SYSTEM_STARTING_HEALTH_STATUS", Some("notready")),
                ("DYN_SYSTEM_USE_ENDPOINT_HEALTH_STATUS", Some(ENDPOINT_HEALTH_CONFIG)),
            ],
            async {
                let drt = Arc::new(create_test_drt_with_settings_async().await);

                // Check if system status server was started
                let system_info_opt = drt.system_status_server_info();

                // Ensure system status server was spawned by DRT
                assert!(
                    system_info_opt.is_some(),
                    "System status server was not spawned by DRT. Expected DRT to spawn server when DYN_SYSTEM_ENABLED=true, but system_status_server_info() returned None. Environment: DYN_SYSTEM_ENABLED={:?}, DYN_SYSTEM_PORT={:?}",
                    std::env::var("DYN_SYSTEM_ENABLED"),
                    std::env::var("DYN_SYSTEM_PORT")
                );

                // Get the system status server info from DRT - this should never fail now due to above check
                let system_info = system_info_opt.unwrap();
                let addr = system_info.socket_addr;

                // Initially check health - should be not ready
                let client = reqwest::Client::new();
                let health_url = format!("http://{}/health", addr);

                let response = client.get(&health_url).send().await.unwrap();
                let status = response.status();
                let body = response.text().await.unwrap();

                // Health should be not ready (503) initially
                assert_eq!(status, 503, "Health should be 503 (not ready) initially, got: {}", status);
                assert!(body.contains("\"status\":\"notready\""), "Health should contain status notready");

                // Now create a namespace, component, and endpoint to make the system healthy
                let namespace = drt.namespace("ns1234").unwrap();
                let component = namespace.component("comp1234").unwrap();

                // Create a simple test handler
                use crate::pipeline::{async_trait, network::Ingress, AsyncEngine, AsyncEngineContextProvider, Error, ManyOut, SingleIn};
                use crate::protocols::annotated::Annotated;

                struct TestHandler;

                #[async_trait]
                impl AsyncEngine<SingleIn<String>, ManyOut<Annotated<String>>, Error> for TestHandler {
                    async fn generate(&self, input: SingleIn<String>) -> crate::Result<ManyOut<Annotated<String>>> {
                        let (data, ctx) = input.into_parts();
                        let response = Annotated::from_data(format!("You responded: {}", data));
                        Ok(crate::pipeline::ResponseStream::new(
                            Box::pin(crate::stream::iter(vec![response])),
                            ctx.context()
                        ))
                    }
                }

                // Create the ingress and start the endpoint service
                let ingress = Ingress::for_engine(std::sync::Arc::new(TestHandler)).unwrap();

                // Start the service and endpoint
                tokio::spawn(async move {
                    let _ = component
                        .service_builder()
                        .create()
                        .await
                        .unwrap()
                        .endpoint(ENDPOINT_NAME)
                        .endpoint_builder()
                        .handler(ingress)
                        .start()
                        .await;
                });

                // Hit health endpoint 200 times to verify consistency
                let mut success_count = 0;
                let mut failures = Vec::new();

                for i in 1..=200 {
                    let response = client.get(&health_url).send().await.unwrap();
                    let status = response.status();
                    let body = response.text().await.unwrap();

                    if status == 200 && body.contains("\"status\":\"ready\"") {
                        success_count += 1;
                    } else {
                        failures.push((i, status.as_u16(), body.clone()));
                        if failures.len() <= 5 {  // Only log first 5 failures
                            tracing::warn!("Request {}: status={}, body={}", i, status, body);
                        }
                    }
                }

                tracing::info!("Health endpoint test results: {}/200 requests succeeded", success_count);
                if !failures.is_empty() {
                    tracing::warn!("Failed requests: {}", failures.len());
                }
716

717
718
719
720
721
                // Expect at least 150 out of 200 requests to be successful
                assert!(success_count >= 150, "Expected at least 150 out of 200 requests to succeed, but only {} succeeded", success_count);
            },
        )
        .await;
722
723
724
    }

    #[cfg(feature = "integration")]
725
    #[tokio::test]
726
    async fn test_spawn_system_status_server_endpoints() {
727
        // use reqwest for HTTP requests
728
        temp_env::async_with_vars(
729
730
731
732
733
            [
                ("DYN_SYSTEM_ENABLED", Some("true")),
                ("DYN_SYSTEM_PORT", Some("0")),
                ("DYN_SYSTEM_STARTING_HEALTH_STATUS", Some("ready")),
            ],
734
            async {
735
736
737
738
739
740
741
                let drt = Arc::new(create_test_drt_with_settings_async().await);

                // Get system status server info from DRT (instead of manually spawning)
                let system_info = drt
                    .system_status_server_info()
                    .expect("System status server should be started by DRT");
                let addr = system_info.socket_addr;
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
                let client = reqwest::Client::new();
                for (path, expect_200, expect_body) in [
                    ("/health", true, "ready"),
                    ("/live", true, "ready"),
                    ("/someRandomPathNotFoundHere", false, "Route not found"),
                ] {
                    println!("[test] Sending request to {}", path);
                    let url = format!("http://{}{}", addr, path);
                    let response = client.get(&url).send().await.unwrap();
                    let status = response.status();
                    let body = response.text().await.unwrap();
                    println!(
                        "[test] Response for {}: status={}, body={:?}",
                        path, status, body
                    );
                    if expect_200 {
                        assert_eq!(status, 200, "Response: status={}, body={:?}", status, body);
                    } else {
                        assert_eq!(status, 404, "Response: status={}, body={:?}", status, body);
                    }
                    assert!(
                        body.contains(expect_body),
                        "Response: status={}, body={:?}",
                        status,
                        body
                    );
768
                }
769
                // DRT handles server cleanup automatically
770
771
772
            },
        )
        .await;
773
    }
774
}