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 NORMAL load test: {} clients for {}s", num_clients, duration_secs ); 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 }) }; // Spawn client tasks 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 }); } // 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); } } // 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( "Normal Load Test".to_string(), TestConfig { num_clients, duration_secs, test_type: "normal_load".to_string(), }, ); // Add capacity recommendation if report.metrics.error_rate_pct < 1.0 && report.metrics.latency_stats.p99_ms < 10.0 { report.capacity_recommendation = Some(crate::metrics::CapacityRecommendation { max_sustainable_clients: num_clients, max_sustainable_rps: report.metrics.requests_per_second, bottleneck_identified: None, recommendation: format!( "System handled {} clients with {:.2}% error rate and {:.2}ms P99 latency. \ System is performing well under normal load.", num_clients, report.metrics.error_rate_pct, report.metrics.latency_stats.p99_ms ), }); } else { let bottleneck = if report.metrics.error_rate_pct >= 1.0 { "High error rate indicates potential backend service overload" } else { "High latency indicates potential performance bottleneck" }; report.capacity_recommendation = Some(crate::metrics::CapacityRecommendation { max_sustainable_clients: usize::try_from( (f64::from(u32::try_from(num_clients).unwrap_or(u32::MAX)) * 0.8) as u64, ) .unwrap_or(usize::MAX), // Estimate 80% as safe max_sustainable_rps: report.metrics.requests_per_second * 0.8, bottleneck_identified: Some(bottleneck.to_string()), recommendation: format!( "System showed degradation at {} clients ({:.2}% error rate, {:.2}ms P99). \ Recommend staying below {} concurrent clients for production.", num_clients, report.metrics.error_rate_pct, report.metrics.latency_stats.p99_ms, usize::try_from( (f64::from(u32::try_from(num_clients).unwrap_or(u32::MAX)) * 0.8) as u64 ) .unwrap_or(usize::MAX) ), }); } tracing::info!("Normal load test completed: {:?}", report.metrics); Ok(report) }