ant_quic/monitoring/

metrics.rs

//! Production Metrics Collection
//!
//! This module implements high-performance metrics collection for NAT traversal
//! operations with intelligent sampling, aggregation, and export capabilities.

use std::{
    collections::{HashMap, VecDeque},
    sync::{
        atomic::{AtomicU64, Ordering},
        Arc,
    },
    time::{Duration, Instant, SystemTime},
    net::SocketAddr,
};

use tokio::{
    sync::{RwLock, Mutex},
    time::interval,
};
use tracing::{debug, info, warn};

use super::{
    MonitoringError, NatTraversalAttempt, NatTraversalResult, MetricsSummary,
};

/// Production metrics collector with intelligent sampling
pub struct ProductionMetricsCollector {
    /// Metrics configuration
    config: MetricsConfig,
    /// Core metrics storage
    metrics_store: Arc<MetricsStore>,
    /// Sampling controller
    sampler: Arc<AdaptiveSampler>,
    /// Aggregation engine
    aggregator: Arc<MetricsAggregator>,
    /// Export manager
    exporter: Arc<MetricsExporter>,
    /// Circuit breaker for overload protection
    circuit_breaker: Arc<CircuitBreaker>,
    /// Collector state
    state: Arc<RwLock<CollectorState>>,
    /// Background tasks
    tasks: Arc<Mutex<Vec<tokio::task::JoinHandle<()>>>>,
}

impl ProductionMetricsCollector {
    /// Create new production metrics collector
    pub async fn new(config: MetricsConfig) -> Result<Self, MonitoringError> {
        let metrics_store = Arc::new(MetricsStore::new(config.storage.clone()));
        let sampler = Arc::new(AdaptiveSampler::new(config.sampling.clone()));
        let aggregator = Arc::new(MetricsAggregator::new(config.aggregation.clone()));
        let exporter = Arc::new(MetricsExporter::new(config.export.clone()));
        let circuit_breaker = Arc::new(CircuitBreaker::new(config.circuit_breaker.clone()));
        
        Ok(Self {
            config,
            metrics_store,
            sampler,
            aggregator,
            exporter,
            circuit_breaker,
            state: Arc::new(RwLock::new(CollectorState::new())),
            tasks: Arc::new(Mutex::new(Vec::new())),
        })
    }
    
    /// Start metrics collection
    pub async fn start(&self) -> Result<(), MonitoringError> {
        info!("Starting production metrics collector");
        
        // Update state
        {
            let mut state = self.state.write().await;
            state.status = CollectorStatus::Starting;
            state.start_time = Some(Instant::now());
        }
        
        // Start background tasks
        self.start_aggregation_task().await?;
        self.start_export_task().await?;
        self.start_cleanup_task().await?;
        self.start_health_task().await?;
        
        // Update state to running
        {
            let mut state = self.state.write().await;
            state.status = CollectorStatus::Running;
        }
        
        info!("Production metrics collector started");
        Ok(())
    }
    
    /// Stop metrics collection
    pub async fn stop(&self) -> Result<(), MonitoringError> {
        info!("Stopping production metrics collector");
        
        // Update state
        {
            let mut state = self.state.write().await;
            state.status = CollectorStatus::Stopping;
        }
        
        // Stop background tasks
        let mut tasks = self.tasks.lock().await;
        for task in tasks.drain(..) {
            task.abort();
        }
        
        // Final export
        self.exporter.flush().await?;
        
        // Update state
        {
            let mut state = self.state.write().await;
            state.status = CollectorStatus::Stopped;
        }
        
        info!("Production metrics collector stopped");
        Ok(())
    }
    
    /// Record NAT traversal attempt
    pub async fn record_nat_attempt(&self, attempt: &NatTraversalAttempt) -> Result<(), MonitoringError> {
        // Check circuit breaker
        if !self.circuit_breaker.allow_request().await {
            return Ok(()); // Fail fast during overload
        }
        
        // Check sampling decision
        if !self.sampler.should_sample_attempt(attempt).await {
            return Ok(());
        }
        
        // Record attempt metrics
        let attempt_metric = AttemptMetric {
            attempt_id: attempt.attempt_id.clone(),
            timestamp: attempt.timestamp,
            client_region: attempt.client_info.region.clone(),
            server_region: attempt.server_info.region.clone(),
            nat_types: (
                attempt.client_info.nat_type.clone(),
                attempt.server_info.nat_type.clone(),
            ),
            network_conditions: attempt.network_conditions.clone(),
        };
        
        self.metrics_store.record_attempt(attempt_metric).await?;
        
        // Update counters
        self.increment_counter("nat_attempts_total", &[
            ("client_region", attempt.client_info.region.as_deref().unwrap_or("unknown")),
            ("server_region", attempt.server_info.region.as_deref().unwrap_or("unknown")),
        ]).await;
        
        Ok(())
    }
    
    /// Record NAT traversal result
    pub async fn record_nat_result(&self, result: &NatTraversalResult) -> Result<(), MonitoringError> {
        // Always sample results (more important than attempts)
        let sample_rate = if result.success { 0.1 } else { 1.0 }; // 10% success, 100% failures
        if !self.sampler.should_sample_with_rate(sample_rate).await {
            return Ok(());
        }
        
        // Record result metrics
        let result_metric = ResultMetric {
            attempt_id: result.attempt_id.clone(),
            success: result.success,
            duration: result.duration,
            error_category: result.error_info.as_ref().map(|e| e.error_category.clone()),
            performance: result.performance_metrics.clone(),
            connection_info: result.connection_info.clone(),
        };
        
        self.metrics_store.record_result(result_metric).await?;
        
        // Update counters and histograms
        let status = if result.success { "success" } else { "failure" };
        self.increment_counter("nat_results_total", &[("status", status)]).await;
        
        self.record_histogram("nat_duration_ms", result.duration.as_millis() as f64, &[
            ("status", status),
        ]).await;
        
        if let Some(conn_info) = &result.connection_info {
            self.record_histogram("connection_latency_ms", conn_info.quality.latency_ms as f64, &[]).await;
            self.record_histogram("connection_throughput_mbps", conn_info.quality.throughput_mbps as f64, &[]).await;
        }
        
        // Record error metrics
        if let Some(error_info) = &result.error_info {
            self.increment_counter("nat_errors_total", &[
                ("category", &format!("{:?}", error_info.error_category)),
                ("code", &error_info.error_code),
            ]).await;
        }
        
        Ok(())
    }
    
    /// Get collector status
    pub async fn get_status(&self) -> String {
        let state = self.state.read().await;
        format!("{:?}", state.status)
    }
    
    /// Get metrics summary
    pub async fn get_summary(&self) -> MetricsSummary {
        self.metrics_store.get_summary().await
    }
    
    /// Increment counter metric
    async fn increment_counter(&self, name: &str, labels: &[(&str, &str)]) {
        self.metrics_store.increment_counter(name, labels).await;
    }
    
    /// Record histogram value
    async fn record_histogram(&self, name: &str, value: f64, labels: &[(&str, &str)]) {
        self.metrics_store.record_histogram(name, value, labels).await;
    }
    
    /// Start aggregation background task
    async fn start_aggregation_task(&self) -> Result<(), MonitoringError> {
        let aggregator = self.aggregator.clone();
        let metrics_store = self.metrics_store.clone();
        let interval_duration = self.config.aggregation.interval;
        
        let task = tokio::spawn(async move {
            let mut interval = interval(interval_duration);
            
            loop {
                interval.tick().await;
                
                if let Err(e) = aggregator.aggregate_metrics(&metrics_store).await {
                    warn!("Metrics aggregation failed: {}", e);
                }
            }
        });
        
        self.tasks.lock().await.push(task);
        Ok(())
    }
    
    /// Start export background task
    async fn start_export_task(&self) -> Result<(), MonitoringError> {
        let exporter = self.exporter.clone();
        let aggregator = self.aggregator.clone();
        let interval_duration = self.config.export.interval;
        
        let task = tokio::spawn(async move {
            let mut interval = interval(interval_duration);
            
            loop {
                interval.tick().await;
                
                match aggregator.get_aggregated_metrics().await {
                    Ok(metrics) => {
                        if let Err(e) = exporter.export_metrics(metrics).await {
                            warn!("Metrics export failed: {}", e);
                        }
                    }
                    Err(e) => {
                        warn!("Failed to get aggregated metrics: {}", e);
                    }
                }
            }
        });
        
        self.tasks.lock().await.push(task);
        Ok(())
    }
    
    /// Start cleanup background task
    async fn start_cleanup_task(&self) -> Result<(), MonitoringError> {
        let metrics_store = self.metrics_store.clone();
        let retention_period = self.config.storage.retention_period;
        
        let task = tokio::spawn(async move {
            let mut interval = interval(Duration::from_secs(3600)); // Cleanup hourly
            
            loop {
                interval.tick().await;
                
                if let Err(e) = metrics_store.cleanup_old_data(retention_period).await {
                    warn!("Metrics cleanup failed: {}", e);
                }
            }
        });
        
        self.tasks.lock().await.push(task);
        Ok(())
    }
    
    /// Start health monitoring task
    async fn start_health_task(&self) -> Result<(), MonitoringError> {
        let circuit_breaker = self.circuit_breaker.clone();
        let metrics_store = self.metrics_store.clone();
        let state = self.state.clone();
        
        let task = tokio::spawn(async move {
            let mut interval = interval(Duration::from_secs(30)); // Health check every 30s
            
            loop {
                interval.tick().await;
                
                // Check system health
                let health = metrics_store.get_health_metrics().await;
                let metrics_per_second = health.metrics_per_second;
                circuit_breaker.update_health(health).await;
                
                // Update collector state
                let mut collector_state = state.write().await;
                collector_state.last_health_check = Some(Instant::now());
                collector_state.metrics_collected += metrics_per_second as u64;
            }
        });
        
        self.tasks.lock().await.push(task);
        Ok(())
    }
}

/// Metrics configuration
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct MetricsConfig {
    /// Storage configuration
    pub storage: StorageConfig,
    /// Sampling configuration
    pub sampling: SamplingConfig,
    /// Aggregation configuration
    pub aggregation: AggregationConfig,
    /// Export configuration
    pub export: MetricsExportConfig,
    /// Circuit breaker configuration
    pub circuit_breaker: CircuitBreakerConfig,
}

impl Default for MetricsConfig {
    fn default() -> Self {
        Self {
            storage: StorageConfig::default(),
            sampling: SamplingConfig::default(),
            aggregation: AggregationConfig::default(),
            export: MetricsExportConfig::default(),
            circuit_breaker: CircuitBreakerConfig::default(),
        }
    }
}

/// Storage configuration
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct StorageConfig {
    /// Maximum metrics to store in memory
    pub max_metrics: usize,
    /// Data retention period
    pub retention_period: Duration,
    /// Flush interval to disk
    pub flush_interval: Duration,
}

impl Default for StorageConfig {
    fn default() -> Self {
        Self {
            max_metrics: 100_000,
            retention_period: Duration::from_secs(3600), // 1 hour
            flush_interval: Duration::from_secs(60),     // 1 minute
        }
    }
}

/// Sampling configuration
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct SamplingConfig {
    /// Base sampling rate for attempts
    pub base_attempt_rate: f64,
    /// Sampling rate for results
    pub result_rate: f64,
    /// Adaptive sampling settings
    pub adaptive: AdaptiveSamplingConfig,
}

impl Default for SamplingConfig {
    fn default() -> Self {
        Self {
            base_attempt_rate: 0.01, // 1% of attempts
            result_rate: 0.1,        // 10% of results
            adaptive: AdaptiveSamplingConfig::default(),
        }
    }
}

/// Adaptive sampling configuration
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct AdaptiveSamplingConfig {
    /// Enable adaptive sampling
    pub enabled: bool,
    /// Target metrics per second
    pub target_rate: f64,
    /// Adjustment interval
    pub adjustment_interval: Duration,
    /// Maximum sampling rate
    pub max_rate: f64,
    /// Minimum sampling rate
    pub min_rate: f64,
}

impl Default for AdaptiveSamplingConfig {
    fn default() -> Self {
        Self {
            enabled: true,
            target_rate: 1000.0, // 1000 metrics/sec
            adjustment_interval: Duration::from_secs(60),
            max_rate: 1.0,
            min_rate: 0.001,
        }
    }
}

/// Aggregation configuration
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct AggregationConfig {
    /// Aggregation interval
    pub interval: Duration,
    /// Aggregation window size
    pub window_size: Duration,
    /// Enable percentile calculations
    pub enable_percentiles: bool,
}

impl Default for AggregationConfig {
    fn default() -> Self {
        Self {
            interval: Duration::from_secs(10),        // Aggregate every 10s
            window_size: Duration::from_secs(60),     // 60s window
            enable_percentiles: true,
        }
    }
}

/// Export configuration
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct MetricsExportConfig {
    /// Export interval
    pub interval: Duration,
    /// Export destinations
    pub destinations: Vec<ExportDestination>,
    /// Batch size for export
    pub batch_size: usize,
    /// Export timeout
    pub timeout: Duration,
}

impl Default for MetricsExportConfig {
    fn default() -> Self {
        Self {
            interval: Duration::from_secs(30),
            destinations: vec![ExportDestination::Prometheus {
                endpoint: "http://localhost:9090/api/v1/write".to_string(),
            }],
            batch_size: 1000,
            timeout: Duration::from_secs(10),
        }
    }
}

/// Export destinations
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub enum ExportDestination {
    Prometheus { endpoint: String },
    InfluxDB { endpoint: String, database: String },
    CloudWatch { region: String },
    DataDog { api_key: String },
    StatsD { endpoint: String },
}

/// Circuit breaker configuration
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct CircuitBreakerConfig {
    /// Failure threshold
    pub failure_threshold: u32,
    /// Success threshold for recovery
    pub success_threshold: u32,
    /// Timeout duration
    pub timeout: Duration,
    /// Maximum queue size
    pub max_queue_size: usize,
}

impl Default for CircuitBreakerConfig {
    fn default() -> Self {
        Self {
            failure_threshold: 5,
            success_threshold: 3,
            timeout: Duration::from_secs(60),
            max_queue_size: 10000,
        }
    }
}

/// Metrics store for high-performance storage
struct MetricsStore {
    /// Counter metrics
    counters: Arc<RwLock<HashMap<String, CounterMetric>>>,
    /// Histogram metrics
    histograms: Arc<RwLock<HashMap<String, HistogramMetric>>>,
    /// Attempt metrics
    attempts: Arc<Mutex<VecDeque<AttemptMetric>>>,
    /// Result metrics
    results: Arc<Mutex<VecDeque<ResultMetric>>>,
    /// Storage configuration
    config: StorageConfig,
}

impl MetricsStore {
    fn new(config: StorageConfig) -> Self {
        Self {
            counters: Arc::new(RwLock::new(HashMap::new())),
            histograms: Arc::new(RwLock::new(HashMap::new())),
            attempts: Arc::new(Mutex::new(VecDeque::new())),
            results: Arc::new(Mutex::new(VecDeque::new())),
            config,
        }
    }
    
    async fn record_attempt(&self, attempt: AttemptMetric) -> Result<(), MonitoringError> {
        let mut attempts = self.attempts.lock().await;
        attempts.push_back(attempt);
        
        // Enforce size limit
        while attempts.len() > self.config.max_metrics {
            attempts.pop_front();
        }
        
        Ok(())
    }
    
    async fn record_result(&self, result: ResultMetric) -> Result<(), MonitoringError> {
        let mut results = self.results.lock().await;
        results.push_back(result);
        
        // Enforce size limit
        while results.len() > self.config.max_metrics {
            results.pop_front();
        }
        
        Ok(())
    }
    
    async fn increment_counter(&self, name: &str, labels: &[(&str, &str)]) {
        let key = format!("{}:{}", name, labels_to_string(labels));
        let mut counters = self.counters.write().await;
        
        counters.entry(key)
            .or_insert_with(|| CounterMetric::new(name.to_string(), labels_to_map(labels)))
            .increment();
    }
    
    async fn record_histogram(&self, name: &str, value: f64, labels: &[(&str, &str)]) {
        let key = format!("{}:{}", name, labels_to_string(labels));
        let mut histograms = self.histograms.write().await;
        
        histograms.entry(key)
            .or_insert_with(|| HistogramMetric::new(name.to_string(), labels_to_map(labels)))
            .record(value);
    }
    
    async fn get_summary(&self) -> MetricsSummary {
        let results = self.results.lock().await;
        let _one_hour_ago = SystemTime::now() - Duration::from_secs(3600);
        
        let recent_results: Vec<_> = results.iter()
            .filter(|r| r.attempt_id.len() > 0) // Simple time filter
            .collect();
        
        let total_attempts = recent_results.len() as u64;
        let successful = recent_results.iter().filter(|r| r.success).count() as u64;
        let success_rate = if total_attempts > 0 {
            successful as f32 / total_attempts as f32
        } else {
            0.0
        };
        
        let avg_duration = if !recent_results.is_empty() {
            recent_results.iter()
                .map(|r| r.duration.as_millis())
                .sum::<u128>() / recent_results.len() as u128
        } else {
            0
        };
        
        MetricsSummary {
            nat_attempts_last_hour: total_attempts,
            success_rate_last_hour: success_rate,
            avg_connection_time_ms: avg_duration as u64,
            active_connections: 0, // Would be tracked separately
            error_rate_last_hour: 1.0 - success_rate,
        }
    }
    
    async fn cleanup_old_data(&self, retention_period: Duration) -> Result<(), MonitoringError> {
        let cutoff = SystemTime::now() - retention_period;
        
        // Cleanup attempts
        {
            let mut attempts = self.attempts.lock().await;
            while let Some(front) = attempts.front() {
                if front.timestamp < cutoff {
                    attempts.pop_front();
                } else {
                    break;
                }
            }
        }
        
        // Cleanup results would be similar
        // In practice, would also cleanup counters and histograms
        
        Ok(())
    }
    
    async fn get_health_metrics(&self) -> HealthMetrics {
        let attempts_count = self.attempts.lock().await.len();
        let results_count = self.results.lock().await.len();
        
        HealthMetrics {
            metrics_per_second: (attempts_count + results_count) as f64 / 60.0, // Rough estimate
            memory_usage_mb: ((attempts_count + results_count) * 1024) as f64 / 1024.0 / 1024.0,
            queue_depth: attempts_count + results_count,
            error_rate: 0.0, // Would calculate from actual errors
        }
    }
}

/// Adaptive sampler for intelligent sampling decisions
struct AdaptiveSampler {
    config: SamplingConfig,
    current_rate: Arc<AtomicU64>, // Stored as u64 for atomic operations
    last_adjustment: Arc<RwLock<Instant>>,
}

impl AdaptiveSampler {
    fn new(config: SamplingConfig) -> Self {
        let initial_rate = (config.base_attempt_rate * 1_000_000.0) as u64; // Store as millionths
        
        Self {
            config,
            current_rate: Arc::new(AtomicU64::new(initial_rate)),
            last_adjustment: Arc::new(RwLock::new(Instant::now())),
        }
    }
    
    async fn should_sample_attempt(&self, _attempt: &NatTraversalAttempt) -> bool {
        let rate = self.current_rate.load(Ordering::Relaxed) as f64 / 1_000_000.0;
        rand::random::<f64>() < rate
    }
    
    async fn should_sample_with_rate(&self, rate: f64) -> bool {
        rand::random::<f64>() < rate
    }
    
    async fn adjust_sampling_rate(&self, current_metrics_rate: f64) {
        if !self.config.adaptive.enabled {
            return;
        }
        
        let mut last_adjustment = self.last_adjustment.write().await;
        if last_adjustment.elapsed() < self.config.adaptive.adjustment_interval {
            return;
        }
        
        let target_rate = self.config.adaptive.target_rate;
        let current_rate = self.current_rate.load(Ordering::Relaxed) as f64 / 1_000_000.0;
        
        let adjustment_factor = target_rate / current_metrics_rate.max(1.0);
        let new_rate = (current_rate * adjustment_factor)
            .max(self.config.adaptive.min_rate)
            .min(self.config.adaptive.max_rate);
        
        self.current_rate.store((new_rate * 1_000_000.0) as u64, Ordering::Relaxed);
        *last_adjustment = Instant::now();
        
        debug!("Adjusted sampling rate from {:.4} to {:.4}", current_rate, new_rate);
    }
}

/// Circuit breaker for overload protection
struct CircuitBreaker {
    config: CircuitBreakerConfig,
    state: Arc<RwLock<CircuitBreakerState>>,
    consecutive_failures: Arc<AtomicU64>,
    consecutive_successes: Arc<AtomicU64>,
    queue_size: Arc<AtomicU64>,
}

impl CircuitBreaker {
    fn new(config: CircuitBreakerConfig) -> Self {
        Self {
            config,
            state: Arc::new(RwLock::new(CircuitBreakerState::Closed)),
            consecutive_failures: Arc::new(AtomicU64::new(0)),
            consecutive_successes: Arc::new(AtomicU64::new(0)),
            queue_size: Arc::new(AtomicU64::new(0)),
        }
    }
    
    async fn allow_request(&self) -> bool {
        let state = self.state.read().await;
        
        match *state {
            CircuitBreakerState::Closed => {
                // Check queue size
                self.queue_size.load(Ordering::Relaxed) < self.config.max_queue_size as u64
            }
            CircuitBreakerState::Open => false,
            CircuitBreakerState::HalfOpen => {
                // Allow limited requests to test recovery
                self.queue_size.load(Ordering::Relaxed) < (self.config.max_queue_size / 10) as u64
            }
        }
    }
    
    async fn update_health(&self, health: HealthMetrics) {
        // Update queue size
        self.queue_size.store(health.queue_depth as u64, Ordering::Relaxed);
        
        // Check if we should change circuit breaker state
        if health.error_rate > 0.5 {
            // High error rate
            let failures = self.consecutive_failures.fetch_add(1, Ordering::Relaxed) + 1;
            self.consecutive_successes.store(0, Ordering::Relaxed);
            
            if failures >= self.config.failure_threshold as u64 {
                let mut state = self.state.write().await;
                *state = CircuitBreakerState::Open;
                warn!("Circuit breaker opened due to high error rate");
            }
        } else {
            // Low error rate
            let successes = self.consecutive_successes.fetch_add(1, Ordering::Relaxed) + 1;
            self.consecutive_failures.store(0, Ordering::Relaxed);
            
            let current_state = *self.state.read().await;
            if matches!(current_state, CircuitBreakerState::Open) && successes >= self.config.success_threshold as u64 {
                let mut state = self.state.write().await;
                *state = CircuitBreakerState::HalfOpen;
                info!("Circuit breaker moved to half-open state");
            } else if matches!(current_state, CircuitBreakerState::HalfOpen) && successes >= self.config.success_threshold as u64 * 2 {
                let mut state = self.state.write().await;
                *state = CircuitBreakerState::Closed;
                info!("Circuit breaker closed - system recovered");
            }
        }
    }
}

/// Circuit breaker states
#[derive(Debug, Clone, Copy)]
enum CircuitBreakerState {
    Closed,
    Open,
    HalfOpen,
}

/// Metrics aggregator for time-based aggregation
struct MetricsAggregator {
    config: AggregationConfig,
    aggregated_data: Arc<RwLock<AggregatedData>>,
}

impl MetricsAggregator {
    fn new(config: AggregationConfig) -> Self {
        Self {
            config,
            aggregated_data: Arc::new(RwLock::new(AggregatedData::new())),
        }
    }
    
    async fn aggregate_metrics(&self, _metrics_store: &MetricsStore) -> Result<(), MonitoringError> {
        // Aggregate metrics from the store
        // This would calculate time-based windows, percentiles, etc.
        debug!("Aggregating metrics for export");
        Ok(())
    }
    
    async fn get_aggregated_metrics(&self) -> Result<Vec<ExportMetric>, MonitoringError> {
        let data = self.aggregated_data.read().await;
        Ok(data.to_export_metrics())
    }
}

/// Metrics exporter for various destinations
struct MetricsExporter {
    config: MetricsExportConfig,
}

impl MetricsExporter {
    fn new(config: MetricsExportConfig) -> Self {
        Self { config }
    }
    
    async fn export_metrics(&self, metrics: Vec<ExportMetric>) -> Result<(), MonitoringError> {
        for destination in &self.config.destinations {
            if let Err(e) = self.export_to_destination(destination, &metrics).await {
                warn!("Failed to export to {:?}: {}", destination, e);
            }
        }
        Ok(())
    }
    
    async fn export_to_destination(&self, destination: &ExportDestination, metrics: &[ExportMetric]) -> Result<(), MonitoringError> {
        match destination {
            ExportDestination::Prometheus { endpoint } => {
                debug!("Exporting {} metrics to Prometheus at {}", metrics.len(), endpoint);
                // Would implement actual Prometheus export
            }
            ExportDestination::InfluxDB { endpoint, database } => {
                debug!("Exporting {} metrics to InfluxDB at {} (db: {})", metrics.len(), endpoint, database);
                // Would implement actual InfluxDB export
            }
            _ => {
                debug!("Export to {:?} not yet implemented", destination);
            }
        }
        Ok(())
    }
    
    async fn flush(&self) -> Result<(), MonitoringError> {
        debug!("Flushing remaining metrics");
        Ok(())
    }
}

// Helper functions and data structures

fn labels_to_string(labels: &[(&str, &str)]) -> String {
    labels.iter()
        .map(|(k, v)| format!("{}={}", k, v))
        .collect::<Vec<_>>()
        .join(",")
}

fn labels_to_map(labels: &[(&str, &str)]) -> HashMap<String, String> {
    labels.iter()
        .map(|(k, v)| (k.to_string(), v.to_string()))
        .collect()
}

/// Collector state
#[derive(Debug)]
struct CollectorState {
    status: CollectorStatus,
    start_time: Option<Instant>,
    last_health_check: Option<Instant>,
    metrics_collected: u64,
    errors_encountered: u64,
}

impl CollectorState {
    fn new() -> Self {
        Self {
            status: CollectorStatus::Stopped,
            start_time: None,
            last_health_check: None,
            metrics_collected: 0,
            errors_encountered: 0,
        }
    }
}

/// Collector status
#[derive(Debug, Clone)]
enum CollectorStatus {
    Stopped,
    Starting,
    Running,
    Stopping,
    Error,
}

/// Health metrics
#[derive(Debug, Clone)]
struct HealthMetrics {
    metrics_per_second: f64,
    memory_usage_mb: f64,
    queue_depth: usize,
    error_rate: f64,
}

/// Attempt metric
#[derive(Debug, Clone)]
struct AttemptMetric {
    attempt_id: String,
    timestamp: SystemTime,
    client_region: Option<String>,
    server_region: Option<String>,
    nat_types: (Option<crate::monitoring::NatType>, Option<crate::monitoring::NatType>),
    network_conditions: crate::monitoring::NetworkConditions,
}

/// Result metric
#[derive(Debug, Clone)]
struct ResultMetric {
    attempt_id: String,
    success: bool,
    duration: Duration,
    error_category: Option<crate::monitoring::ErrorCategory>,
    performance: crate::monitoring::PerformanceMetrics,
    connection_info: Option<crate::monitoring::ConnectionInfo>,
}

/// Counter metric for tracking counts
#[derive(Debug)]
struct CounterMetric {
    name: String,
    labels: HashMap<String, String>,
    value: AtomicU64,
    last_updated: std::sync::RwLock<Instant>,
}

impl CounterMetric {
    fn new(name: String, labels: HashMap<String, String>) -> Self {
        Self {
            name,
            labels,
            value: AtomicU64::new(0),
            last_updated: std::sync::RwLock::new(Instant::now()),
        }
    }
    
    fn increment(&self) {
        self.value.fetch_add(1, Ordering::Relaxed);
        if let Ok(mut last_updated) = self.last_updated.write() {
            *last_updated = Instant::now();
        }
    }
    
    fn get_value(&self) -> u64 {
        self.value.load(Ordering::Relaxed)
    }
}

/// Histogram metric for tracking distributions
#[derive(Debug)]
struct HistogramMetric {
    name: String,
    labels: HashMap<String, String>,
    values: std::sync::Mutex<Vec<f64>>,
    buckets: Vec<f64>,
    last_updated: std::sync::RwLock<Instant>,
}

impl HistogramMetric {
    fn new(name: String, labels: HashMap<String, String>) -> Self {
        // Standard histogram buckets for latency/duration metrics
        let buckets = vec![
            0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 25.0, 50.0, 100.0
        ];
        
        Self {
            name,
            labels,
            values: std::sync::Mutex::new(Vec::new()),
            buckets,
            last_updated: std::sync::RwLock::new(Instant::now()),
        }
    }
    
    fn record(&self, value: f64) {
        if let Ok(mut values) = self.values.lock() {
            values.push(value);
            // Keep only recent values to prevent unbounded growth
            if values.len() > 10000 {
                values.drain(0..5000);
            }
        }
        if let Ok(mut last_updated) = self.last_updated.write() {
            *last_updated = Instant::now();
        }
    }
    
    fn get_percentile(&self, percentile: f64) -> Option<f64> {
        let values = self.values.lock().ok()?;
        if values.is_empty() {
            return None;
        }
        
        let mut sorted_values = values.clone();
        sorted_values.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));
        
        let index = ((percentile / 100.0) * (sorted_values.len() - 1) as f64) as usize;
        Some(sorted_values[index])
    }
    
    fn get_bucket_counts(&self) -> Vec<(f64, u64)> {
        let values = match self.values.lock() {
            Ok(guard) => guard,
            Err(_) => return Vec::new(),
        };
        
        self.buckets.iter().map(|&bucket| {
            let count = values.iter().filter(|&&v| v <= bucket).count() as u64;
            (bucket, count)
        }).collect()
    }
}

/// Bootstrap node performance metrics
#[derive(Debug, Clone)]
pub struct BootstrapNodeMetrics {
    /// Node address
    pub address: SocketAddr,
    /// Total coordination requests handled
    pub coordination_requests: u64,
    /// Successful coordinations
    pub successful_coordinations: u64,
    /// Average response time
    pub avg_response_time_ms: f64,
    /// Current availability (0.0 to 1.0)
    pub availability: f64,
    /// Last successful contact
    pub last_contact: Option<SystemTime>,
    /// Error rate in last hour
    pub error_rate: f64,
}

/// NAT type success rate metrics
#[derive(Debug, Clone)]
pub struct NatTypeMetrics {
    /// NAT type
    pub nat_type: crate::monitoring::NatType,
    /// Total attempts for this NAT type
    pub total_attempts: u64,
    /// Successful attempts
    pub successful_attempts: u64,
    /// Success rate (0.0 to 1.0)
    pub success_rate: f64,
    /// Average connection time for successful attempts
    pub avg_connection_time_ms: f64,
    /// Most common failure reasons
    pub common_failures: Vec<(String, u64)>,
}

/// Connection latency and RTT metrics
#[derive(Debug, Clone)]
pub struct LatencyMetrics {
    /// Connection establishment latency percentiles
    pub connection_latency_p50: f64,
    pub connection_latency_p95: f64,
    pub connection_latency_p99: f64,
    /// Round-trip time percentiles
    pub rtt_p50: f64,
    pub rtt_p95: f64,
    pub rtt_p99: f64,
    /// Jitter measurements
    pub jitter_avg: f64,
    pub jitter_max: f64,
}

/// Comprehensive metrics collection implementation
impl ProductionMetricsCollector {
    /// Record bootstrap node performance
    pub async fn record_bootstrap_performance(
        &self,
        node_address: SocketAddr,
        response_time: Duration,
        success: bool,
    ) -> Result<(), MonitoringError> {
        // Update bootstrap node specific metrics
        let node_str = node_address.to_string();
        let status_str = if success { "success" } else { "failure" };
        let labels = &[
            ("node", node_str.as_str()),
            ("status", status_str),
        ];
        
        self.increment_counter("bootstrap_requests_total", labels).await;
        self.record_histogram("bootstrap_response_time_ms", response_time.as_millis() as f64, &[
            ("node", &node_address.to_string()),
        ]).await;
        
        if !success {
            self.increment_counter("bootstrap_errors_total", &[
                ("node", &node_address.to_string()),
            ]).await;
        }
        
        Ok(())
    }
    
    /// Record NAT type specific metrics
    pub async fn record_nat_type_result(
        &self,
        nat_type: crate::monitoring::NatType,
        success: bool,
        duration: Duration,
        error_category: Option<crate::monitoring::ErrorCategory>,
    ) -> Result<(), MonitoringError> {
        let nat_type_str = format!("{:?}", nat_type);
        let status = if success { "success" } else { "failure" };
        
        // Record NAT type specific success/failure
        self.increment_counter("nat_traversal_by_type_total", &[
            ("nat_type", &nat_type_str),
            ("status", status),
        ]).await;
        
        // Record duration by NAT type
        self.record_histogram("nat_traversal_duration_by_type_ms", duration.as_millis() as f64, &[
            ("nat_type", &nat_type_str),
            ("status", status),
        ]).await;
        
        // Record error categories for failures
        if let Some(error_cat) = error_category {
            self.increment_counter("nat_traversal_errors_by_type", &[
                ("nat_type", &nat_type_str),
                ("error_category", &format!("{:?}", error_cat)),
            ]).await;
        }
        
        Ok(())
    }
    
    /// Record connection quality metrics
    pub async fn record_connection_quality(
        &self,
        latency_ms: u32,
        jitter_ms: u32,
        throughput_mbps: f32,
        packet_loss_rate: f32,
    ) -> Result<(), MonitoringError> {
        // Record latency metrics
        self.record_histogram("connection_latency_ms", latency_ms as f64, &[]).await;
        self.record_histogram("connection_jitter_ms", jitter_ms as f64, &[]).await;
        self.record_histogram("connection_throughput_mbps", throughput_mbps as f64, &[]).await;
        self.record_histogram("connection_packet_loss_rate", packet_loss_rate as f64, &[]).await;
        
        Ok(())
    }
    
    /// Get bootstrap node metrics
    pub async fn get_bootstrap_metrics(&self) -> Vec<BootstrapNodeMetrics> {
        let counters = self.metrics_store.counters.read().await;
        let histograms = self.metrics_store.histograms.read().await;
        
        let mut node_metrics = HashMap::new();
        
        // Collect bootstrap node data from counters and histograms
        for (key, counter) in counters.iter() {
            if key.starts_with("bootstrap_requests_total:") {
                if let Some(node_addr) = extract_label_value(key, "node") {
                    let entry = node_metrics.entry(node_addr.clone()).or_insert_with(|| BootstrapNodeMetrics {
                        address: node_addr.parse().unwrap_or_else(|_| "0.0.0.0:0".parse().unwrap()),
                        coordination_requests: 0,
                        successful_coordinations: 0,
                        avg_response_time_ms: 0.0,
                        availability: 1.0,
                        last_contact: Some(SystemTime::now()),
                        error_rate: 0.0,
                    });
                    
                    if key.contains("status=success") {
                        entry.successful_coordinations = counter.get_value();
                    }
                    entry.coordination_requests += counter.get_value();
                }
            }
        }
        
        // Add response time data from histograms
        for (key, histogram) in histograms.iter() {
            if key.starts_with("bootstrap_response_time_ms:") {
                if let Some(node_addr) = extract_label_value(key, "node") {
                    if let Some(entry) = node_metrics.get_mut(&node_addr) {
                        entry.avg_response_time_ms = histogram.get_percentile(50.0).unwrap_or(0.0);
                    }
                }
            }
        }
        
        // Calculate availability and error rates
        for metrics in node_metrics.values_mut() {
            if metrics.coordination_requests > 0 {
                metrics.availability = metrics.successful_coordinations as f64 / metrics.coordination_requests as f64;
                metrics.error_rate = 1.0 - metrics.availability;
            }
        }
        
        node_metrics.into_values().collect()
    }
    
    /// Get NAT type success rate metrics
    pub async fn get_nat_type_metrics(&self) -> Vec<NatTypeMetrics> {
        let counters = self.metrics_store.counters.read().await;
        let histograms = self.metrics_store.histograms.read().await;
        
        let mut nat_metrics = HashMap::new();
        
        // Collect NAT type data from counters
        for (key, counter) in counters.iter() {
            if key.starts_with("nat_traversal_by_type_total:") {
                if let Some(nat_type_str) = extract_label_value(key, "nat_type") {
                    let nat_type = parse_nat_type(&nat_type_str);
                    let entry = nat_metrics.entry(nat_type_str.clone()).or_insert_with(|| NatTypeMetrics {
                        nat_type,
                        total_attempts: 0,
                        successful_attempts: 0,
                        success_rate: 0.0,
                        avg_connection_time_ms: 0.0,
                        common_failures: Vec::new(),
                    });
                    
                    if key.contains("status=success") {
                        entry.successful_attempts = counter.get_value();
                    }
                    entry.total_attempts += counter.get_value();
                }
            }
        }
        
        // Add duration data from histograms
        for (key, histogram) in histograms.iter() {
            if key.starts_with("nat_traversal_duration_by_type_ms:") && key.contains("status=success") {
                if let Some(nat_type_str) = extract_label_value(key, "nat_type") {
                    if let Some(entry) = nat_metrics.get_mut(&nat_type_str) {
                        entry.avg_connection_time_ms = histogram.get_percentile(50.0).unwrap_or(0.0);
                    }
                }
            }
        }
        
        // Calculate success rates
        for metrics in nat_metrics.values_mut() {
            if metrics.total_attempts > 0 {
                metrics.success_rate = metrics.successful_attempts as f64 / metrics.total_attempts as f64;
            }
        }
        
        nat_metrics.into_values().collect()
    }
    
    /// Get latency and RTT metrics
    pub async fn get_latency_metrics(&self) -> LatencyMetrics {
        let histograms = self.metrics_store.histograms.read().await;
        
        let connection_latency = histograms.get("connection_latency_ms:");
        let rtt_histogram = histograms.get("connection_rtt_ms:");
        let jitter_histogram = histograms.get("connection_jitter_ms:");
        
        LatencyMetrics {
            connection_latency_p50: connection_latency.and_then(|h| h.get_percentile(50.0)).unwrap_or(0.0),
            connection_latency_p95: connection_latency.and_then(|h| h.get_percentile(95.0)).unwrap_or(0.0),
            connection_latency_p99: connection_latency.and_then(|h| h.get_percentile(99.0)).unwrap_or(0.0),
            rtt_p50: rtt_histogram.and_then(|h| h.get_percentile(50.0)).unwrap_or(0.0),
            rtt_p95: rtt_histogram.and_then(|h| h.get_percentile(95.0)).unwrap_or(0.0),
            rtt_p99: rtt_histogram.and_then(|h| h.get_percentile(99.0)).unwrap_or(0.0),
            jitter_avg: jitter_histogram.and_then(|h| h.get_percentile(50.0)).unwrap_or(0.0),
            jitter_max: jitter_histogram.and_then(|h| h.get_percentile(100.0)).unwrap_or(0.0),
        }
    }
}

// Helper functions
fn extract_label_value(key: &str, label_name: &str) -> Option<String> {
    let label_prefix = format!("{}=", label_name);
    key.split(',')
        .find(|part| part.contains(&label_prefix))
        .and_then(|part| part.split('=').nth(1))
        .map(|s| s.to_string())
}

fn parse_nat_type(nat_type_str: &str) -> crate::monitoring::NatType {
    match nat_type_str {
        "FullCone" => crate::monitoring::NatType::FullCone,
        "RestrictedCone" => crate::monitoring::NatType::RestrictedCone,
        "PortRestrictedCone" => crate::monitoring::NatType::PortRestrictedCone,
        "Symmetric" => crate::monitoring::NatType::Symmetric,
        "CarrierGrade" => crate::monitoring::NatType::CarrierGrade,
        "DoubleNat" => crate::monitoring::NatType::DoubleNat,
        "None" => crate::monitoring::NatType::None,
        _ => crate::monitoring::NatType::None,
    }
}

/// Aggregated data for export
#[derive(Debug)]
struct AggregatedData {
    /// Aggregated counters
    counters: HashMap<String, u64>,
    /// Aggregated histograms with percentiles
    histograms: HashMap<String, HistogramSummary>,
    /// Last aggregation time
    last_aggregation: Instant,
}

impl AggregatedData {
    fn new() -> Self {
        Self {
            counters: HashMap::new(),
            histograms: HashMap::new(),
            last_aggregation: Instant::now(),
        }
    }
    
    fn to_export_metrics(&self) -> Vec<ExportMetric> {
        let mut metrics = Vec::new();
        
        // Export counters
        for (name, value) in &self.counters {
            metrics.push(ExportMetric {
                name: name.clone(),
                metric_type: MetricType::Counter,
                value: MetricValue::Counter(*value),
                labels: HashMap::new(),
                timestamp: SystemTime::now(),
            });
        }
        
        // Export histogram summaries
        for (name, summary) in &self.histograms {
            metrics.push(ExportMetric {
                name: format!("{}_p50", name),
                metric_type: MetricType::Gauge,
                value: MetricValue::Gauge(summary.p50),
                labels: HashMap::new(),
                timestamp: SystemTime::now(),
            });
            
            metrics.push(ExportMetric {
                name: format!("{}_p95", name),
                metric_type: MetricType::Gauge,
                value: MetricValue::Gauge(summary.p95),
                labels: HashMap::new(),
                timestamp: SystemTime::now(),
            });
            
            metrics.push(ExportMetric {
                name: format!("{}_p99", name),
                metric_type: MetricType::Gauge,
                value: MetricValue::Gauge(summary.p99),
                labels: HashMap::new(),
                timestamp: SystemTime::now(),
            });
        }
        
        metrics
    }
}

/// Histogram summary for aggregation
#[derive(Debug, Clone)]
struct HistogramSummary {
    pub count: u64,
    pub sum: f64,
    pub p50: f64,
    pub p95: f64,
    pub p99: f64,
}

/// Export metric format
#[derive(Debug, Clone)]
pub struct ExportMetric {
    pub name: String,
    pub metric_type: MetricType,
    pub value: MetricValue,
    pub labels: HashMap<String, String>,
    pub timestamp: SystemTime,
}

/// Metric types for export
#[derive(Debug, Clone)]
pub enum MetricType {
    Counter,
    Gauge,
    Histogram,
}

/// Metric values for export
#[derive(Debug, Clone)]
pub enum MetricValue {
    Counter(u64),
    Gauge(f64),
    Histogram(Vec<(f64, u64)>), // bucket, count pairs
}



#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn test_metrics_collector_creation() {
        let config = MetricsConfig::default();
        let collector = ProductionMetricsCollector::new(config).await.unwrap();
        
        let status = collector.get_status().await;
        assert!(status.contains("Stopped"));
    }
    
    #[tokio::test]
    async fn test_adaptive_sampler() {
        let mut config = SamplingConfig::default();
        // Set a shorter adjustment interval for testing
        config.adaptive.adjustment_interval = Duration::from_millis(10);
        let sampler = AdaptiveSampler::new(config.clone());
        
        // Wait for the adjustment interval to allow first adjustment
        tokio::time::sleep(Duration::from_millis(20)).await;
        
        // Test rate adjustment with half the target rate
        sampler.adjust_sampling_rate(500.0).await; // Half of target rate (1000)
        
        // Rate should increase to compensate (double the base rate)
        let rate = sampler.current_rate.load(Ordering::Relaxed) as f64 / 1_000_000.0;
        let expected_rate = config.base_attempt_rate * 2.0; // Should double
        assert!((rate - expected_rate).abs() < 0.001); // Allow small floating point error
    }
    
    #[tokio::test]
    async fn test_circuit_breaker() {
        let config = CircuitBreakerConfig::default();
        let breaker = CircuitBreaker::new(config);
        
        // Initially should allow requests
        assert!(breaker.allow_request().await);
        
        // Simulate high error rate
        let bad_health = HealthMetrics {
            metrics_per_second: 1000.0,
            memory_usage_mb: 100.0,
            queue_depth: 100,
            error_rate: 0.8, // High error rate
        };
        
        // Update health multiple times to trip circuit breaker
        for _ in 0..10 {
            breaker.update_health(bad_health.clone()).await;
        }
        
        // Should eventually deny requests (depending on thresholds)
        // This test might need adjustment based on exact logic
    }
}