amberjs 1.17.0

// 分布式指标收集模块
// 提供实时性能指标收集、聚合和监控功能
//
// Stage 29.7: 分布式监控与调试 - 实时性能指标和监控

use std::collections::HashMap;
use std::sync::{Arc, Mutex, RwLock};

use tracing::{debug, info, warn};
use std::time::{Duration, Instant};
use std::time::SystemTime;
use std::hash::Hash;
/// 指标类型
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub enum MetricType {
    /// 集群级别Nodes指标
    Cluster,
    ClusterThroughput,
    ClusterLatency,
    ClusterAvailability,
    /// 节点级别指标
    NodeCpuUsage,
    NodeMemoryUsage,
    NodeNetworkIO,
    NodeDiskIO,
    NodeActiveTasks,
    NodeTaskQueueSize,
    /// 任务级别指标
    TaskExecutionTime,
    TaskSuccessRate,
    TaskFailureRate,
    TaskQueueTime,
    /// 系统资源指标
    SystemLoadAverage,
    SystemMemoryPressure,
    SystemNetworkLatency,
    SystemDiskUtilization,
}
/// 指标值
#[derive(Debug, Clone)]
pub enum MetricValue {
    Counter(u64),
    Gauge(f64),
    Histogram(Vec<f64>),
    Summary(Percentiles),
}
/// 百分位数统计
#[derive(Debug, Clone)]
pub struct Percentiles {
    pub p50: f64,
    pub p90: f64,
    pub p95: f64,
    pub p99: f64,
    pub p999: f64,
}
/// 指标数据点
#[derive(Debug, Clone)]
pub struct MetricPoint {
    pub metric_type: MetricType,
    pub value: MetricValue,
    pub timestamp: u64, // 使用 u64 而不是 Instant，便于序列化
    pub labels: HashMap<String, String>,
    pub node_id: String,
}
/// 实时指标
#[derive(Debug, Clone)]
pub struct RealTimeMetrics {
    pub cluster_summary: ClusterMetricsSummary,
    pub node_metrics: HashMap<String, NodeMetrics>,
    pub system_metrics: SystemMetrics,
    pub timestamp: u64, // 使用 u64 而不是 Instant，便于序列化
}
/// 集群指标摘要
#[derive(Debug, Clone)]
pub struct ClusterMetricsSummary {
    pub total_nodes: u32,
    pub healthy_nodes: u32,
    pub total_tasks: u64,
    pub active_tasks: u64,
    pub completed_tasks: u64,
    pub failed_tasks: u64,
    pub average_throughput: f64, // tasks/sec
    pub average_latency: f64, // ms
    pub availability: f64, // 0-1
}
/// 节点指标
#[derive(Debug, Clone)]
pub struct NodeMetrics {
    pub node_id: String,
    pub cpu_usage: f64, // 0-100
    pub memory_usage: f64, // 0-100
    pub memory_used_gb: f64,
    pub memory_total_gb: f64,
    pub network_rx_mbps: f64,
    pub network_tx_mbps: f64,
    pub disk_read_mbps: f64,
    pub disk_write_mbps: f64,
    pub active_tasks: u32,
    pub task_queue_size: u32,
    pub load_average: f64,
    pub uptime_seconds: u64,
}
/// 系统指标
#[derive(Debug, Clone)]
pub struct SystemMetrics {
    pub load_average: f64,
    pub memory_pressure: f64, // 0-1
    pub network_latency_ms: f64,
    pub disk_utilization: f64, // 0-100
    pub gc_collection_time_ms: f64,
    pub jit_compilation_time_ms: f64,
}
/// 指标配置
#[derive(Debug, Clone)]
pub struct MetricsConfig {
    pub collection_interval: Duration,
    pub retention_period: Duration,
    pub enable_real_time: bool,
    pub enable_aggregation: bool,
    pub max_metric_points: usize,
}
/// 分布式指标收集器
#[derive(Clone, Debug)]
pub struct DistributedMetrics {
    config: MetricsConfig,
    node_manager: Arc<NodeManager>,
    task_executor: Arc<TaskExecutor>,
    task_scheduler: Arc<TaskScheduler>,
    metric_points: Arc<RwLock<Vec<MetricPoint>>>,
    real_time_metrics: Arc<RwLock<Option<RealTimeMetrics>>>,
    historical_stats: Arc<RwLock<HashMap<MetricType, Vec<MetricPoint>>>>,
}
impl DistributedMetrics {
    /// 创建新的指标收集器
    pub fn new(
        config: MetricsConfig,
        node_manager: Arc<NodeManager>,
        task_executor: Arc<TaskExecutor>,
        task_scheduler: Arc<TaskScheduler>,
    ) -> Self {
        Self {
            config,
            node_manager,
            task_executor,
            task_scheduler,
            metric_points: Arc::new(Mutex::new(Vec::new())),
            real_time_metrics: Arc::new(Mutex::new(None)),
            historical_stats: Arc::new(Mutex::new(HashMap::new())),
        }
    }
    /// 启动指标收集
    pub async fn start(&self) -> Result<(), String> {
        info!("Starting distributed metrics collector...");
        let metrics_collector: _ = self.clone();
        tokio::spawn(async move {
            let mut interval_timer = interval(metrics_collector.config.collection_interval);
            loop {
                interval_timer.tick().await;
                if let Err(e) = metrics_collector.collect_metrics().await {
                    warn!("Failed to collect metrics: {}", e);
                }
                if let Err(e) = metrics_collector.update_real_time_metrics().await {
                    warn!("Failed to update real-time metrics: {}", e);
                }
                if metrics_collector.config.enable_aggregation {
                    if let Err(e) = metrics_collector.aggregate_metrics().await {
                        warn!("Failed to aggregate metrics: {}", e);
                    }
                }
                if let Err(e) = metrics_collector.cleanup_old_metrics().await {
                    warn!("Failed to cleanup old metrics: {}", e);
                }
            }
        });
        info!("Distributed metrics collector started");
        Ok(())
    }
    /// 收集所有指标
    async fn collect_metrics(&self) -> Result<(), String> {
        debug!("Collecting metrics...");
        // 收集集群指标
        self.collect_cluster_metrics().await?;
        // 收集节点指标
        self.collect_node_metrics().await?;
        // 收集任务指标
        self.collect_task_metrics().await?;
        // 收集系统指标
        self.collect_system_metrics().await?;
        Ok(())
    }
    /// 收集集群级别指标
    async fn collect_cluster_metrics(&self) -> Result<(), String> {
        let cluster_topology: _ = self.node_manager.get_cluster_topology().await;
        // 节点数量指标
        let node_count_metric: _ = MetricPoint {
            metric_type: MetricType::Cluster,
            value: MetricValue::Gauge(cluster_topology.total_nodes as f64),
            timestamp: std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_secs(),
            labels: HashMap::new(),
            node_id: "cluster".to_string(),
        };
        self.add_metric_point(node_count_metric).await;
        // 吞吐量指标（模拟计算）
        let throughput: _ = self.calculate_cluster_throughput().await;
        let throughput_metric: _ = MetricPoint {
            metric_type: MetricType::ClusterThroughput,
            value: MetricValue::Gauge(throughput),
            timestamp: std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_secs(),
            labels: HashMap::new(),
            node_id: "cluster".to_string(),
        };
        self.add_metric_point(throughput_metric).await;
        // 可用性指标
        let availability: _ = self.calculate_cluster_availability().await;
        let availability_metric: _ = MetricPoint {
            metric_type: MetricType::ClusterAvailability,
            value: MetricValue::Gauge(availability * 100.0),
            timestamp: std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_secs(),
            labels: HashMap::new(),
            node_id: "cluster".to_string(),
        };
        self.add_metric_point(availability_metric).await;
        Ok(())
    }
    /// 收集节点级别指标
    async fn collect_node_metrics(&self) -> Result<(), String> {
        let cluster_topology: _ = self.node_manager.get_cluster_topology().await;
        // 生成每个节点的指标
        for region in cluster_topology.regions.values() {
            for i in 0..region.node_count {
                let node_id: _ = format!("{}-node-{}", region.location, i);
                // CPU 使用率（模拟数据）
                let cpu_usage: _ = self.simulate_cpu_usage(&node_id);
                let cpu_metric: _ = MetricPoint {
                    metric_type: MetricType::NodeCpuUsage,
                    value: MetricValue::Gauge(cpu_usage),
                    timestamp: std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_secs(),
                    labels: HashMap::new(),
                    node_id: node_id.clone(),
                };
                self.add_metric_point(cpu_metric).await;
                // 内存使用率（模拟数据）
                let memory_usage: _ = self.simulate_memory_usage(&node_id);
                let memory_metric: _ = MetricPoint {
                    metric_type: MetricType::NodeMemoryUsage,
                    value: MetricValue::Gauge(memory_usage),
                    timestamp: std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_secs(),
                    labels: HashMap::new(),
                    node_id: node_id.clone(),
                };
                self.add_metric_point(memory_metric).await;
                // 活跃任务数（模拟数据）
                let active_tasks: _ = self.simulate_active_tasks(&node_id);
                let task_metric: _ = MetricPoint {
                    metric_type: MetricType::NodeActiveTasks,
                    value: MetricValue::Gauge(active_tasks as f64),
                    timestamp: std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_secs(),
                    labels: HashMap::new(),
                    node_id: node_id.clone(),
                };
                self.add_metric_point(task_metric).await;
            }
        }
        Ok(())
    }
    /// 收集任务级别指标
    async fn collect_task_metrics(&self) -> Result<(), String> {
        // 任务执行时间（模拟数据）
        let execution_times: _ = self.simulate_execution_times();
        let execution_metric: _ = MetricPoint {
            metric_type: MetricType::TaskExecutionTime,
            value: MetricValue::Histogram(execution_times),
            timestamp: std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_secs(),
            labels: HashMap::new(),
            node_id: "cluster".to_string(),
        };
        self.add_metric_point(execution_metric).await;
        // 成功率（模拟数据）
        let success_rate: _ = self.simulate_success_rate();
        let success_metric: _ = MetricPoint {
            metric_type: MetricType::TaskSuccessRate,
            value: MetricValue::Gauge(success_rate * 100.0),
            timestamp: std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_secs(),
            labels: HashMap::new(),
            node_id: "cluster".to_string(),
        };
        self.add_metric_point(success_metric).await;
        Ok(())
    }
    /// 收集系统级别指标
    async fn collect_system_metrics(&self) -> Result<(), String> {
        // 系统负载（模拟数据）
        let load_average: _ = self.simulate_load_average();
        let load_metric: _ = MetricPoint {
            metric_type: MetricType::SystemLoadAverage,
            value: MetricValue::Gauge(load_average),
            timestamp: std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_secs(),
            labels: HashMap::new(),
            node_id: "system".to_string(),
        };
        self.add_metric_point(load_metric).await;
        // 内存压力（模拟数据）
        let memory_pressure: _ = self.simulate_memory_pressure();
        let pressure_metric: _ = MetricPoint {
            metric_type: MetricType::SystemMemoryPressure,
            value: MetricValue::Gauge(memory_pressure),
            timestamp: std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_secs(),
            labels: HashMap::new(),
            node_id: "system".to_string(),
        };
        self.add_metric_point(pressure_metric).await;
        Ok(())
    }
    /// 添加指标点
    async fn add_metric_point(&self, metric: MetricPoint) {
        let mut points = self.metric_points.write().await;
        points.push(metric);
        // 限制指标点数量
        if points.len() > self.config.max_metric_points {
            points.remove(0);
        }
    }
    /// 更新实时指标
    async fn update_real_time_metrics(&self) -> Result<(), String> {
        let cluster_topology: _ = self.node_manager.get_cluster_topology().await;
        let mut node_metrics = HashMap::new();
        for region in cluster_topology.regions.values() {
            for i in 0..region.node_count {
                let node_id: _ = format!("{}-node-{}", region.location, i);
                let node_metric: _ = NodeMetrics {
                    node_id: node_id.clone(),
                    cpu_usage: self.simulate_cpu_usage(&node_id),
                    memory_usage: self.simulate_memory_usage(&node_id),
                    memory_used_gb: 8.0 + ((node_id.len() as f64) % 4.0),
                    memory_total_gb: 16.0,
                    network_rx_mbps: self.simulate_network_rx(&node_id),
                    network_tx_mbps: self.simulate_network_tx(&node_id),
                    disk_read_mbps: self.simulate_disk_read(&node_id),
                    disk_write_mbps: self.simulate_disk_write(&node_id),
                    active_tasks: self.simulate_active_tasks(&node_id),
                    task_queue_size: self.simulate_queue_size(&node_id),
                    load_average: self.simulate_load_average(),
                    uptime_seconds: 3600 + (node_id.len() as u64 * 60),
                };
                node_metrics.insert(node_id, node_metric);
            }
        }
        let system_metrics: _ = SystemMetrics {
            load_average: self.simulate_load_average(),
            memory_pressure: self.simulate_memory_pressure(),
            network_latency_ms: self.simulate_network_latency(),
            disk_utilization: self.simulate_disk_utilization(),
            gc_collection_time_ms: self.simulate_gc_time(),
            jit_compilation_time_ms: self.simulate_jit_time(),
        };
        let cluster_summary: _ = ClusterMetricsSummary {
            total_nodes: cluster_topology.total_nodes as u32,
            healthy_nodes: cluster_topology.online_nodes as u32,
            total_tasks: 10000,
            active_tasks: 150,
            completed_tasks: 9850,
            failed_tasks: 0,
            average_throughput: 250.0,
            average_latency: 45.5,
            availability: 0.999,
        };
        let real_time: _ = RealTimeMetrics {
            cluster_summary,
            node_metrics,
            system_metrics,
            timestamp: std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_secs(),
        };
        let mut rt_metrics = self.real_time_metrics.write().await;
        *rt_metrics = Some(real_time);
        Ok(())
    }
    /// 获取实时指标
    pub async fn get_real_time_metrics(&self) -> Option<RealTimeMetrics> {
        let metrics: _ = self.real_time_metrics.read().await;
        metrics.clone()
    }
    /// 获取历史指标
    pub async fn get_historical_metrics(&self, metric_type: MetricType) -> Vec<MetricPoint> {
        let stats: _ = self.historical_stats.read().await;
        stats.get(&metric_type).cloned().unwrap_or_default()
    }
    /// 聚合指标
    async fn aggregate_metrics(&self) -> Result<(), String> {
        let points: _ = self.metric_points.read().await;
        let mut aggregated: HashMap<MetricType, Vec<MetricPoint>> = HashMap::new();
        for point in points.iter() {
            aggregated.entry(point.metric_type.clone())
                .or_insert_with(Vec::new)
                .push(point.clone());
        }
        let mut stats = self.historical_stats.write().await;
        *stats = aggregated;
        Ok(())
    }
    /// 清理旧指标
    async fn cleanup_old_metrics(&self) -> Result<(), String> {
        let cutoff: _ = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_secs()
            - self.config.retention_period.as_secs();
        let mut points = self.metric_points.write().await;
        points.retain(|point| point.timestamp > cutoff);
        Ok(())
    }
    // 模拟数据生成方法（实际实现中会从系统获取真实数据）
    fn simulate_cpu_usage(&self, _node_id: &str) -> f64 {
        20.0 + (rand::random::<f64>() * 60.0)
    }
    fn simulate_memory_usage(&self, _node_id: &str) -> f64 {
        30.0 + (rand::random::<f64>() * 50.0)
    }
    fn simulate_active_tasks(&self, _node_id: &str) -> u32 {
        (rand::random::<u32>() % 50) + 10
    }
    fn simulate_queue_size(&self, _node_id: &str) -> u32 {
        rand::random::<u32>() % 20
    }
    fn simulate_network_rx(&self, _node_id: &str) -> f64 {
        rand::random::<f64>() * 100.0
    }
    fn simulate_network_tx(&self, _node_id: &str) -> f64 {
        rand::random::<f64>() * 80.0
    }
    fn simulate_disk_read(&self, _node_id: &str) -> f64 {
        rand::random::<f64>() * 50.0
    }
    fn simulate_disk_write(&self, _node_id: &str) -> f64 {
        rand::random::<f64>() * 40.0
    }
    fn simulate_load_average(&self) -> f64 {
        0.5 + (rand::random::<f64>() * 2.0)
    }
    fn simulate_memory_pressure(&self) -> f64 {
        rand::random::<f64>() * 0.8
    }
    fn simulate_network_latency(&self) -> f64 {
        0.5 + (rand::random::<f64>() * 5.0)
    }
    fn simulate_disk_utilization(&self) -> f64 {
        rand::random::<f64>() * 70.0
    }
    fn simulate_gc_time(&self) -> f64 {
        rand::random::<f64>() * 10.0
    }
    fn simulate_jit_time(&self) -> f64 {
        rand::random::<f64>() * 20.0
    }
    async fn calculate_cluster_throughput(&self) -> f64 {
        // 模拟集群吞吐量计算
        200.0 + (rand::random::<f64>() * 100.0)
    }
    async fn calculate_cluster_availability(&self) -> f64 {
        // 模拟集群可用性计算
        0.95 + (rand::random::<f64>() * 0.05)
    }
    fn simulate_execution_times(&self) -> Vec<f64> {
        (0..100).map(|_| {
            10.0 + (rand::random::<f64>() * 90.0)
        }).collect()
    }
    fn simulate_success_rate(&self) -> f64 {
        0.95 + (rand::random::<f64>() * 0.05)
    }
}