oxirs-cluster 0.2.4

//! Advanced Cluster Analytics

use anyhow::Result;
use scirs2_core::stats::statistical_analysis;
use std::collections::{HashMap, VecDeque};
use std::time::Instant;
use super::types::*;
pub struct AdvancedClusterAnalytics {
    analytics_ml: MLPipeline,
    performance_tracker: PerformanceTracker,
    anomaly_detector: AnomalyDetector,
    trend_analyzer: TrendAnalyzer,
}

impl AdvancedClusterAnalytics {
    async fn new() -> Result<Self> {
        Ok(Self {
            analytics_ml: MLPipeline::new(),
            performance_tracker: PerformanceTracker::new(),
            anomaly_detector: AnomalyDetector::new(),
            trend_analyzer: TrendAnalyzer::new(),
        })
    }

    async fn collect_cluster_metrics(&mut self, cluster_state: &ClusterState) -> Result<()> {
        // Collect performance metrics
        self.performance_tracker.record_performance(cluster_state).await?;

        // Detect anomalies
        let anomalies = self.anomaly_detector.detect_anomalies(cluster_state).await?;
        if !anomalies.is_empty() {
            tracing::warn!("Detected {} anomalies in cluster", anomalies.len());
        }

        // Analyze trends
        self.trend_analyzer.analyze_trends(cluster_state).await?;

        Ok(())
    }

    async fn get_analytics(&self) -> ClusterAnalytics {
        ClusterAnalytics {
            performance_summary: self.performance_tracker.get_performance_summary().await,
            detected_anomalies: self.anomaly_detector.get_recent_anomalies().await,
            trend_analysis: self.trend_analyzer.get_trend_analysis().await,
            recommendations: self.generate_recommendations().await,
        }
    }

    async fn generate_recommendations(&self) -> Vec<String> {
        vec![
            "Consider adding nodes to handle increased load".to_string(),
            "Optimize network topology for better latency".to_string(),
            "Adjust replication factors based on access patterns".to_string(),
        ]
    }
}

/// Performance tracker for cluster analytics
#[derive(Debug)]
pub struct PerformanceTracker {
    performance_history: VecDeque<ClusterPerformanceSnapshot>,
}

impl PerformanceTracker {
    fn new() -> Self {
        Self {
            performance_history: VecDeque::with_capacity(10000),
        }
    }

    async fn record_performance(&mut self, cluster_state: &ClusterState) -> Result<()> {
        let snapshot = ClusterPerformanceSnapshot {
            timestamp: SystemTime::now(),
            metrics: cluster_state.performance_metrics.clone(),
            node_count: cluster_state.nodes.len(),
            total_data_size: cluster_state.nodes.values().map(|n| n.data_size).sum(),
        };

        self.performance_history.push_back(snapshot);

        // Keep history manageable
        while self.performance_history.len() > 10000 {
            self.performance_history.pop_front();
        }

        Ok(())
    }

    async fn get_performance_summary(&self) -> PerformanceSummary {
        if self.performance_history.is_empty() {
            return PerformanceSummary::default();
        }

        let recent_metrics: Vec<&ClusterPerformanceMetrics> = self.performance_history
            .iter()
            .rev()
            .take(100) // Last 100 snapshots
            .map(|s| &s.metrics)
            .collect();

        let avg_throughput = recent_metrics.iter()
            .map(|m| m.query_throughput_qps)
            .sum::<f64>() / recent_metrics.len() as f64;

        let avg_latency = recent_metrics.iter()
            .map(|m| m.consensus_latency_ms)
            .sum::<u64>() / recent_metrics.len() as u64;

        let avg_availability = recent_metrics.iter()
            .map(|m| m.availability)
            .sum::<f64>() / recent_metrics.len() as f64;

        PerformanceSummary {
            average_throughput_qps: avg_throughput,
            average_latency_ms: avg_latency,
            average_availability: avg_availability,
            performance_trend: self.calculate_performance_trend(),
        }
    }

    fn calculate_performance_trend(&self) -> PerformanceTrend {
        // Simple trend calculation based on recent vs older metrics
        if self.performance_history.len() < 10 {
            return PerformanceTrend::Stable;
        }

        let recent_avg = self.performance_history.iter()
            .rev()
            .take(5)
            .map(|s| s.metrics.query_throughput_qps)
            .sum::<f64>() / 5.0;

        let older_avg = self.performance_history.iter()
            .rev()
            .skip(5)
            .take(5)
            .map(|s| s.metrics.query_throughput_qps)
            .sum::<f64>() / 5.0;

        if recent_avg > older_avg * 1.1 {
            PerformanceTrend::Improving
        } else if recent_avg < older_avg * 0.9 {
            PerformanceTrend::Degrading
        } else {
            PerformanceTrend::Stable
        }
    }
}

/// Anomaly detector for cluster monitoring
#[derive(Debug)]
pub struct AnomalyDetector {
    detection_model: MLPipeline,
    detected_anomalies: VecDeque<DetectedAnomaly>,
}

impl AnomalyDetector {
    fn new() -> Self {
        Self {
            detection_model: MLPipeline::new(),
            detected_anomalies: VecDeque::with_capacity(1000),
        }
    }

    async fn detect_anomalies(&mut self, cluster_state: &ClusterState) -> Result<Vec<DetectedAnomaly>> {
        let mut anomalies = Vec::new();

        // Check for performance anomalies
        if cluster_state.performance_metrics.query_throughput_qps < 100.0 {
            anomalies.push(DetectedAnomaly {
                anomaly_type: AnomalyType::LowThroughput,
                severity: AnomalySeverity::High,
                description: "Query throughput below acceptable threshold".to_string(),
                affected_nodes: Vec::new(),
                timestamp: SystemTime::now(),
            });
        }

        // Check for consensus anomalies
        if cluster_state.performance_metrics.consensus_latency_ms > 1000 {
            anomalies.push(DetectedAnomaly {
                anomaly_type: AnomalyType::HighLatency,
                severity: AnomalySeverity::Medium,
                description: "Consensus latency above acceptable threshold".to_string(),
                affected_nodes: Vec::new(),
                timestamp: SystemTime::now(),
            });
        }

        // Store detected anomalies
        for anomaly in &anomalies {
            self.detected_anomalies.push_back(anomaly.clone());
        }

        // Keep anomaly history manageable
        while self.detected_anomalies.len() > 1000 {
            self.detected_anomalies.pop_front();
        }

        Ok(anomalies)
    }

    async fn get_recent_anomalies(&self) -> Vec<DetectedAnomaly> {
        self.detected_anomalies.iter()
            .rev()
            .take(50) // Last 50 anomalies
            .cloned()
            .collect()
    }
}

/// Trend analyzer for predictive insights
#[derive(Debug)]
pub struct TrendAnalyzer {
    analysis_model: MLPipeline,
    trend_data: VecDeque<TrendDataPoint>,
}

impl TrendAnalyzer {
    fn new() -> Self {
        Self {
            analysis_model: MLPipeline::new(),
            trend_data: VecDeque::with_capacity(5000),
        }
    }

    async fn analyze_trends(&mut self, cluster_state: &ClusterState) -> Result<()> {
        let trend_point = TrendDataPoint {
            timestamp: SystemTime::now(),
            throughput: cluster_state.performance_metrics.query_throughput_qps,
            latency: cluster_state.performance_metrics.consensus_latency_ms as f64,
            availability: cluster_state.performance_metrics.availability,
            node_count: cluster_state.nodes.len(),
        };

        self.trend_data.push_back(trend_point);

        // Keep trend data manageable
        while self.trend_data.len() > 5000 {
            self.trend_data.pop_front();
        }

        Ok(())
    }

    async fn get_trend_analysis(&self) -> TrendAnalysis {
        TrendAnalysis {
            throughput_trend: self.calculate_throughput_trend(),
            latency_trend: self.calculate_latency_trend(),
            availability_trend: self.calculate_availability_trend(),
            capacity_projection: self.project_capacity_needs(),
        }
    }

    fn calculate_throughput_trend(&self) -> TrendDirection {
        if self.trend_data.len() < 10 {
            return TrendDirection::Unknown;
        }

        let recent_avg = self.trend_data.iter()
            .rev()
            .take(5)
            .map(|p| p.throughput)
            .sum::<f64>() / 5.0;

        let older_avg = self.trend_data.iter()
            .rev()
            .skip(5)
            .take(5)
            .map(|p| p.throughput)
            .sum::<f64>() / 5.0;

        if recent_avg > older_avg * 1.05 {
            TrendDirection::Increasing
        } else if recent_avg < older_avg * 0.95 {
            TrendDirection::Decreasing
        } else {
            TrendDirection::Stable
        }
    }

    fn calculate_latency_trend(&self) -> TrendDirection {
        if self.trend_data.len() < 10 {
            return TrendDirection::Unknown;
        }

        let recent_avg = self.trend_data.iter()
            .rev()
            .take(5)
            .map(|p| p.latency)
            .sum::<f64>() / 5.0;

        let older_avg = self.trend_data.iter()
            .rev()
            .skip(5)
            .take(5)
            .map(|p| p.latency)
            .sum::<f64>() / 5.0;

        if recent_avg > older_avg * 1.05 {
            TrendDirection::Increasing
        } else if recent_avg < older_avg * 0.95 {
            TrendDirection::Decreasing
        } else {
            TrendDirection::Stable
        }
    }

    fn calculate_availability_trend(&self) -> TrendDirection {
        if self.trend_data.len() < 10 {
            return TrendDirection::Unknown;
        }

        let recent_avg = self.trend_data.iter()
            .rev()
            .take(5)
            .map(|p| p.availability)
            .sum::<f64>() / 5.0;

        let older_avg = self.trend_data.iter()
            .rev()
            .skip(5)
            .take(5)
            .map(|p| p.availability)
            .sum::<f64>() / 5.0;

        if recent_avg > older_avg * 1.01 {
            TrendDirection::Increasing
        } else if recent_avg < older_avg * 0.99 {
            TrendDirection::Decreasing
        } else {
            TrendDirection::Stable
        }
    }

    fn project_capacity_needs(&self) -> CapacityProjection {
        // Simple linear projection based on current trends
        CapacityProjection {
            projected_node_count_1_month: self.trend_data.back().map(|p| p.node_count + 2).unwrap_or(5),
            projected_throughput_1_month: self.trend_data.back().map(|p| p.throughput * 1.2).unwrap_or(1200.0),
            scaling_recommendation: "Consider adding 2-3 nodes within the next month".to_string(),
        }
    }
}