foxhunt/testing/api-load/src/metrics/collector.rs

use super::*;
use dashmap::DashMap;
use hdrhistogram::Histogram;
use parking_lot::RwLock;
use std::sync::Arc;
use std::time::Instant;
use tokio::sync::mpsc;

pub struct MetricsCollector {
    receiver: mpsc::UnboundedReceiver<RequestMetric>,
    histogram: Arc<RwLock<Histogram<u64>>>,
    service_histograms: Arc<DashMap<ServiceType, Histogram<u64>>>,
    counters: Arc<RwLock<Counters>>,
    start_time: Instant,
    time_series: Arc<RwLock<Vec<TimeSeriesPoint>>>,
}

#[derive(Debug, Default)]
struct Counters {
    total_requests: u64,
    successful_requests: u64,
    failed_requests: u64,
    timeout_requests: u64,
    rate_limited_requests: u64,
    circuit_breaker_requests: u64,
}

impl MetricsCollector {
    pub fn new(receiver: mpsc::UnboundedReceiver<RequestMetric>) -> Self {
        Self {
            receiver,
            histogram: Arc::new(RwLock::new(Histogram::new(3).unwrap())),
            service_histograms: Arc::new(DashMap::new()),
            counters: Arc::new(RwLock::new(Counters::default())),
            start_time: Instant::now(),
            time_series: Arc::new(RwLock::new(Vec::new())),
        }
    }

    pub async fn run(&mut self, num_clients: usize) {
        let mut interval = tokio::time::interval(tokio::time::Duration::from_secs(1));
        let mut last_snapshot_time = Instant::now();
        let mut last_request_count = 0u64;

        loop {
            tokio::select! {
                Some(metric) = self.receiver.recv() => {
                    self.process_metric(metric);
                }
                _ = interval.tick() => {
                    // Create time series snapshot every second
                    let current_time = Instant::now();
                    let elapsed = current_time.duration_since(last_snapshot_time);

                    let counters = self.counters.read();
                    let current_request_count = counters.total_requests;
                    let requests_in_interval = current_request_count - last_request_count;

                    let rps = f64::from(u32::try_from(requests_in_interval).unwrap_or(u32::MAX)) / elapsed.as_secs_f64();

                    let histogram = self.histogram.read();
                    let p99_latency_ms = if !histogram.is_empty() {
                        f64::from(u32::try_from(histogram.value_at_quantile(0.99)).unwrap_or(u32::MAX)) / 1_000_000.0 // Convert nanoseconds to ms
                    } else {
                        0.0
                    };

                    let error_rate_pct = if counters.total_requests > 0 {
                        (f64::from(u32::try_from(counters.failed_requests).unwrap_or(u32::MAX)) / f64::from(u32::try_from(counters.total_requests).unwrap_or(u32::MAX))) * 100.0
                    } else {
                        0.0
                    };

                    drop(counters);
                    drop(histogram);

                    self.time_series.write().push(TimeSeriesPoint {
                        timestamp: chrono::Utc::now(),
                        rps,
                        p99_latency_ms,
                        error_rate_pct,
                        active_clients: num_clients,
                    });

                    last_snapshot_time = current_time;
                    last_request_count = current_request_count;
                }
                else => break,
            }
        }
    }

    fn process_metric(&mut self, metric: RequestMetric) {
        let latency_ns = u64::try_from(metric.latency.as_nanos()).unwrap_or(u64::MAX);

        // Update global histogram
        let mut histogram = self.histogram.write();
        let _ = histogram.record(latency_ns);
        drop(histogram);

        // Update per-service histogram
        self.service_histograms
            .entry(metric.service)
            .or_insert_with(|| Histogram::new(3).unwrap())
            .record(latency_ns)
            .ok();

        // Update counters
        let mut counters = self.counters.write();
        counters.total_requests += 1;

        match metric.status {
            RequestStatus::Success => counters.successful_requests += 1,
            RequestStatus::Error => counters.failed_requests += 1,
            RequestStatus::Timeout => counters.timeout_requests += 1,
            RequestStatus::RateLimited => counters.rate_limited_requests += 1,
            RequestStatus::CircuitBreakerOpen => counters.circuit_breaker_requests += 1,
        }
    }

    pub fn generate_report(&self, test_name: String, config: TestConfig) -> LoadTestReport {
        let end_time = chrono::Utc::now();
        let duration = self.start_time.elapsed();

        let counters = self.counters.read();
        let histogram = self.histogram.read();

        let total_requests = counters.total_requests;
        let successful_requests = counters.successful_requests;
        let failed_requests = counters.failed_requests;

        let requests_per_second =
            f64::from(u32::try_from(total_requests).unwrap_or(u32::MAX)) / duration.as_secs_f64();
        let error_rate_pct = if total_requests > 0 {
            (f64::from(u32::try_from(failed_requests).unwrap_or(u32::MAX))
                / f64::from(u32::try_from(total_requests).unwrap_or(u32::MAX)))
                * 100.0
        } else {
            0.0
        };

        let latency_stats = if !histogram.is_empty() {
            LatencyStats {
                min_ms: f64::from(u32::try_from(histogram.min()).unwrap_or(u32::MAX)) / 1_000_000.0,
                max_ms: f64::from(u32::try_from(histogram.max()).unwrap_or(u32::MAX)) / 1_000_000.0,
                mean_ms: histogram.mean() / 1_000_000.0,
                p50_ms: f64::from(
                    u32::try_from(histogram.value_at_quantile(0.50)).unwrap_or(u32::MAX),
                ) / 1_000_000.0,
                p90_ms: f64::from(
                    u32::try_from(histogram.value_at_quantile(0.90)).unwrap_or(u32::MAX),
                ) / 1_000_000.0,
                p95_ms: f64::from(
                    u32::try_from(histogram.value_at_quantile(0.95)).unwrap_or(u32::MAX),
                ) / 1_000_000.0,
                p99_ms: f64::from(
                    u32::try_from(histogram.value_at_quantile(0.99)).unwrap_or(u32::MAX),
                ) / 1_000_000.0,
                p99_9_ms: f64::from(
                    u32::try_from(histogram.value_at_quantile(0.999)).unwrap_or(u32::MAX),
                ) / 1_000_000.0,
                stddev_ms: histogram.stdev() / 1_000_000.0,
            }
        } else {
            LatencyStats {
                min_ms: 0.0,
                max_ms: 0.0,
                mean_ms: 0.0,
                p50_ms: 0.0,
                p90_ms: 0.0,
                p95_ms: 0.0,
                p99_ms: 0.0,
                p99_9_ms: 0.0,
                stddev_ms: 0.0,
            }
        };

        // Calculate per-service stats
        let mut per_service_stats = std::collections::HashMap::new();
        for entry in self.service_histograms.iter() {
            let service = *entry.key();
            let hist = entry.value();

            if !hist.is_empty() {
                let service_latency_stats = LatencyStats {
                    min_ms: f64::from(u32::try_from(hist.min()).unwrap_or(u32::MAX)) / 1_000_000.0,
                    max_ms: f64::from(u32::try_from(hist.max()).unwrap_or(u32::MAX)) / 1_000_000.0,
                    mean_ms: hist.mean() / 1_000_000.0,
                    p50_ms: f64::from(
                        u32::try_from(hist.value_at_quantile(0.50)).unwrap_or(u32::MAX),
                    ) / 1_000_000.0,
                    p90_ms: f64::from(
                        u32::try_from(hist.value_at_quantile(0.90)).unwrap_or(u32::MAX),
                    ) / 1_000_000.0,
                    p95_ms: f64::from(
                        u32::try_from(hist.value_at_quantile(0.95)).unwrap_or(u32::MAX),
                    ) / 1_000_000.0,
                    p99_ms: f64::from(
                        u32::try_from(hist.value_at_quantile(0.99)).unwrap_or(u32::MAX),
                    ) / 1_000_000.0,
                    p99_9_ms: f64::from(
                        u32::try_from(hist.value_at_quantile(0.999)).unwrap_or(u32::MAX),
                    ) / 1_000_000.0,
                    stddev_ms: hist.stdev() / 1_000_000.0,
                };

                per_service_stats.insert(
                    service,
                    ServiceStats {
                        total_requests: hist.len(),
                        successful_requests: hist.len(), // Simplified for now
                        error_rate_pct: 0.0,             // Simplified for now
                        latency_stats: service_latency_stats,
                    },
                );
            }
        }

        let metrics = AggregatedMetrics {
            total_requests,
            successful_requests,
            failed_requests,
            timeout_requests: counters.timeout_requests,
            rate_limited_requests: counters.rate_limited_requests,
            circuit_breaker_requests: counters.circuit_breaker_requests,
            duration,
            requests_per_second,
            error_rate_pct,
            latency_stats,
            per_service_stats,
            system_metrics: None, // To be filled by system monitor
        };

        LoadTestReport {
            test_name,
            start_time: end_time - chrono::Duration::from_std(duration).expect("INVARIANT: Duration should fit in chrono::Duration"),
            end_time,
            config,
            metrics,
            time_series: self.time_series.read().clone(),
            capacity_recommendation: None, // To be filled based on test type
        }
    }
}