#![allow(clippy::integer_division)] use anyhow::Result; use tokio::sync::mpsc; use tokio::task::JoinSet; use crate::clients::{AuthenticatedClient, MixedWorkloadClient}; use crate::metrics::{collector::MetricsCollector, LoadTestReport, TestConfig}; pub async fn run( gateway_url: String, num_clients: usize, duration_secs: u64, ) -> Result { tracing::info!( "Starting SUSTAINED load test: {} clients for {}s ({}h)", num_clients, duration_secs, duration_secs / 3600 ); let (metrics_tx, metrics_rx) = mpsc::unbounded_channel(); let mut collector = MetricsCollector::new(metrics_rx); // Spawn collector task let collector_handle = { let num_clients_clone = num_clients; tokio::spawn(async move { collector.run(num_clients_clone).await; collector }) }; // Warm-up phase (30 seconds) tracing::info!("Starting warm-up phase (30s)..."); let warm_up_duration = std::time::Duration::from_secs(30); let mut join_set = JoinSet::new(); for client_id in 0..num_clients { let gateway_url = gateway_url.clone(); let metrics_tx_clone = metrics_tx.clone(); join_set.spawn(async move { let auth_client = AuthenticatedClient::new( gateway_url, "test-secret-key-for-load-testing", &format!("user-{}", client_id), &format!("loadtest-user-{}", client_id), ) .await?; let mut mixed_client = MixedWorkloadClient::new(auth_client, client_id, metrics_tx_clone); mixed_client.run_mixed_workload(warm_up_duration).await }); } // Wait for warm-up to complete while let Some(result) = join_set.join_next().await { if let Err(e) = result { tracing::error!("Warm-up client task failed: {:?}", e); } } tracing::info!("Warm-up complete, starting sustained load test..."); // Main sustained load test let mut join_set = JoinSet::new(); let duration = std::time::Duration::from_secs(duration_secs); for client_id in 0..num_clients { let gateway_url = gateway_url.clone(); let metrics_tx = metrics_tx.clone(); join_set.spawn(async move { let auth_client = AuthenticatedClient::new( gateway_url, "test-secret-key-for-load-testing", &format!("user-{}", client_id), &format!("loadtest-user-{}", client_id), ) .await?; let mut mixed_client = MixedWorkloadClient::new(auth_client, client_id, metrics_tx); mixed_client.run_mixed_workload(duration).await }); } // Progress reporting for long-running tests let progress_handle = tokio::spawn(async move { let start = std::time::Instant::now(); let total_duration = std::time::Duration::from_secs(duration_secs); loop { tokio::time::sleep(std::time::Duration::from_secs(300)).await; // Report every 5 minutes let elapsed = start.elapsed(); let progress_pct = (elapsed.as_secs_f64() / total_duration.as_secs_f64()) * 100.0; if elapsed >= total_duration { break; } tracing::info!( "Sustained load test progress: {:.1}% complete ({} / {} seconds)", progress_pct, elapsed.as_secs(), duration_secs ); } }); // Wait for all clients to complete while let Some(result) = join_set.join_next().await { if let Err(e) = result { tracing::error!("Client task failed: {:?}", e); } } // Cancel progress reporting progress_handle.abort(); // Drop the sender to signal collector to finish drop(metrics_tx); // Wait for collector to finish and get the report let collector = collector_handle.await?; let mut report = collector.generate_report( "Sustained Load Test".to_owned(), TestConfig { num_clients, duration_secs, test_type: "sustained_load".to_owned(), }, ); // Analyze for memory leaks and latency degradation let time_series = &report.time_series; let latency_trend = analyze_latency_trend(time_series); let error_rate_stable = analyze_error_rate_stability(time_series); report.capacity_recommendation = Some(crate::metrics::CapacityRecommendation { max_sustainable_clients: num_clients, max_sustainable_rps: report.metrics.requests_per_second, bottleneck_identified: if latency_trend > 10.0 { Some("Latency degradation detected over time".to_owned()) } else if !error_rate_stable { Some("Error rate instability detected".to_owned()) } else { None }, recommendation: if latency_trend < 5.0 && error_rate_stable { format!( "System remained stable over {}h with {} clients. Latency drift: {:.2}%, \ no memory leaks detected. Safe for production.", duration_secs / 3600_u64, num_clients, latency_trend ) } else if latency_trend >= 5.0 { format!( "Latency increased by {:.2}% over test duration. Potential memory leak or \ resource exhaustion. Investigate GC behavior and connection pooling.", latency_trend ) } else { "Error rate fluctuations detected. Review application logs and backend \ service health during sustained load." .to_string() }, }); tracing::info!("Sustained load test completed: {:?}", report.metrics); Ok(report) } fn analyze_latency_trend(time_series: &[crate::metrics::TimeSeriesPoint]) -> f64 { if time_series.len() < 10 { return 0.0; } // Compare first 10% vs last 10% of samples let sample_size = time_series.len() / 10; let first_samples = time_series.get(0..sample_size).unwrap_or(&[]); let last_samples = time_series .get(time_series.len().saturating_sub(sample_size)..time_series.len()) .unwrap_or(&[]); let first_avg: f64 = first_samples.iter().map(|p| p.p99_latency_ms).sum::() / f64::from(u32::try_from(first_samples.len()).unwrap_or(u32::MAX)); let last_avg: f64 = last_samples.iter().map(|p| p.p99_latency_ms).sum::() / f64::from(u32::try_from(last_samples.len()).unwrap_or(u32::MAX)); if first_avg > 0.0 { ((last_avg - first_avg) / first_avg) * 100.0 } else { 0.0 } } fn analyze_error_rate_stability(time_series: &[crate::metrics::TimeSeriesPoint]) -> bool { if time_series.is_empty() { return true; } // Calculate standard deviation of error rates let error_rates: Vec = time_series.iter().map(|p| p.error_rate_pct).collect(); let mean: f64 = error_rates.iter().sum::() / f64::from(u32::try_from(error_rates.len()).unwrap_or(u32::MAX)); let variance: f64 = error_rates .iter() .map(|&rate| { let diff = rate - mean; diff * diff }) .sum::() / f64::from(u32::try_from(error_rates.len()).unwrap_or(u32::MAX)); let stddev = variance.sqrt(); // Consider stable if stddev is less than 2% stddev < 2.0 }