rusty_bin/monitor/

system_monitor.rs

//! System monitoring and health check implementation
//!
//! This module provides comprehensive system monitoring including health checks,
//! application metrics, and system metrics collection.

use std::{sync::Arc, time::Duration};

use flume::{Receiver, Sender};
use rusty_common::collections::FxHashMap;
use sysinfo::System;
use tokio::{sync::RwLock, task::JoinHandle, time::interval};

use crate::monitor::{
    ApplicationMetrics, HealthCheck, HealthStatus, MonitorConfig, SystemMetrics,
    metrics::MetricsCollector,
};

/// System monitor for health checks and metrics collection
#[derive(Debug)]
pub struct SystemMonitor {
    /// Monitor configuration
    config: MonitorConfig,
    /// Application metrics
    app_metrics: Arc<RwLock<ApplicationMetrics>>,
    /// Metrics collector instance
    metrics_collector: MetricsCollector,
    /// Latest system metrics
    latest_system_metrics: Arc<RwLock<SystemMetrics>>,
    /// Health check results
    health_checks: Arc<RwLock<Vec<HealthCheck>>>,
    /// Shutdown signal sender for monitoring
    shutdown_monitoring_signal_tx: Arc<RwLock<Sender<()>>>,
    /// Shutdown signal receiver for monitoring
    shutdown_monitor_signal_rx: Arc<RwLock<Receiver<()>>>,
    /// Shutdown signal sender for health checks
    shutdown_health_signal_tx: Arc<RwLock<Sender<()>>>,
    /// Shutdown signal receiver for health checks
    shutdown_health_signal_rx: Arc<RwLock<Receiver<()>>>,
}

impl SystemMonitor {
    /// Create a new system monitor
    #[must_use]
    pub fn new(config: MonitorConfig) -> Self {
        let app_metrics = ApplicationMetrics {
            timestamp: crate::monitor::utils::time::now_nanos(),
            trades_processed: 0,
            orderbooks_processed: 0,
            files_written: 0,
            bytes_written: 0,
            compression_ratio: 0.0,
            active_connections: 0,
            error_count: 0,
            latency_p50: 0.0,
            latency_p95: 0.0,
            latency_p99: 0.0,
        };

        let (shutdown_tx, shutdown_rx) = flume::unbounded();
        let (shutdown_health_tx, shutdown_health_rx) = flume::unbounded();

        let metrics_collector = MetricsCollector::new();
        let initial_system_metrics = metrics_collector.collect_system_metrics();

        Self {
            config,
            app_metrics: Arc::new(RwLock::new(app_metrics)),
            metrics_collector,
            latest_system_metrics: Arc::new(RwLock::new(initial_system_metrics)),
            health_checks: Arc::new(RwLock::new(Vec::new())),

            shutdown_monitoring_signal_tx: Arc::new(RwLock::new(shutdown_tx)),
            shutdown_monitor_signal_rx: Arc::new(RwLock::new(shutdown_rx)),

            shutdown_health_signal_tx: Arc::new(RwLock::new(shutdown_health_tx)),
            shutdown_health_signal_rx: Arc::new(RwLock::new(shutdown_health_rx)),
        }
    }

    /// Start monitoring (both metrics and health checks)
    pub async fn start_monitoring(&self) -> JoinHandle<()> {
        log::info!("Starting system monitoring");

        // Start metrics collection in background
        let config = self.config.clone();
        let app_metrics = self.app_metrics.clone();
        let latest_system_metrics = self.latest_system_metrics.clone();
        let metrics_collector = self.metrics_collector.clone();

        let shutdown_monitor_signal_rx = self.shutdown_monitor_signal_rx.clone();

        tokio::spawn(async move {
            let mut interval = interval(Duration::from_secs(
                config.monitoring.metrics_interval_seconds,
            ));

            loop {
                tokio::select! {
                    _ = interval.tick() => {
                        // Update application metrics timestamp
                        {
                            let mut metrics = app_metrics.write().await;
                            metrics.timestamp = crate::monitor::utils::time::now_nanos();
                        }

                        // Collect and update system metrics
                        let metrics_collector_clone = metrics_collector.clone();
                        let new_system_metrics =
                            tokio::task::spawn_blocking(move || metrics_collector_clone.collect_system_metrics())
                                .await
                                .unwrap();

                        {
                            let mut system_metrics = latest_system_metrics.write().await;
                            *system_metrics = new_system_metrics;
                        }
                    }
                    _ = shutdown_monitor_signal_rx.read() => {
                        break;
                    }
                }
            }
        })
    }

    /// Stop monitoring
    pub async fn stop_monitoring(&self) {
        log::info!("Stopping system monitoring");
        let _ = self.shutdown_monitoring_signal_tx.write().await.send(());
        log::info!("System monitoring stopped");
    }

    /// Start health monitoring background task
    ///
    /// Returns a JoinHandle to the spawned health monitoring task that periodically checks system health
    pub async fn start_health_monitoring(&self) -> JoinHandle<()> {
        log::info!("Starting health monitoring");

        // Start health monitoring in background
        let config = self.config.clone();
        let shutdown_health_signal_rx = self.shutdown_health_signal_rx.clone();
        let health_checks = self.health_checks.clone();

        tokio::spawn(async move {
            let mut interval = interval(Duration::from_secs(
                config.monitoring.health_check_interval_seconds,
            ));

            loop {
                tokio::select! {
                    _ = interval.tick() => {
                        // Perform health checks
                        let health_check_results = Self::perform_health_checks_static(&config).await;

                        // Store health check results
                        *health_checks.write().await = health_check_results.clone();

                        // Check for critical issues
                        for check in &health_check_results {
                            match check.status {
                                HealthStatus::Critical => {
                                    log::error!("Critical health check failed: {} - {}", check.name, check.message);
                                    // Here we could send alerts (Discord, email, etc.)
                                }
                                HealthStatus::Warning => {
                                    log::warn!("Health check warning: {} - {}", check.name, check.message);
                                }
                                _ => {}
                            }
                        }
                    }
                    _ = shutdown_health_signal_rx.read() => {
                        break;
                    }
                }
            }
        })
    }

    /// Collect current metrics
    pub async fn get_system_metrics(&self) -> SystemMetrics {
        self.latest_system_metrics.read().await.clone()
    }

    /// Check system health
    pub async fn check_health(&self) -> Vec<HealthCheck> {
        self.health_checks.read().await.clone()
    }

    /// Get latest stored health checks
    pub async fn get_health_checks(&self) -> Vec<HealthCheck> {
        self.health_checks.read().await.clone()
    }

    /// Get application metrics
    pub async fn get_app_metrics(&self) -> ApplicationMetrics {
        self.app_metrics.read().await.clone()
    }

    /// Update application metrics
    pub async fn update_app_metrics<F>(&self, updater: F)
    where
        F: FnOnce(&mut ApplicationMetrics),
    {
        let mut metrics = self.app_metrics.write().await;
        updater(&mut metrics);
    }

    /// Perform health checks (static version)
    async fn perform_health_checks_static(config: &MonitorConfig) -> Vec<HealthCheck> {
        let mut checks = Vec::new();

        // Check disk space
        checks.push(Self::check_disk_space_static(config).await);

        // Check memory usage
        checks.push(Self::check_memory_usage_static().await);

        // Check data directories
        checks.push(Self::check_data_directories_static(config).await);

        // Check file permissions
        checks.push(Self::check_file_permissions_static(config).await);

        checks
    }

    /// Check disk space (static version)
    async fn check_disk_space_static(_config: &MonitorConfig) -> HealthCheck {
        use sysinfo::Disks;

        let disks = Disks::new_with_refreshed_list();

        let mut status = HealthStatus::Healthy;
        let mut message = "Disk space is adequate".to_string();
        let mut details = FxHashMap::default();

        for disk in &disks {
            let total = disk.total_space();
            let available = disk.available_space();
            let usage_percent = ((total - available) as f64 / total as f64) * 100.0;

            let mount_point = disk.mount_point().to_string_lossy().to_string();
            details.insert(
                format!("disk_{}", mount_point.replace('/', "_")),
                format!("{usage_percent:.1}% used"),
            );

            if usage_percent > 90.0 {
                status = HealthStatus::Critical;
                message = format!("Disk space critical: {mount_point} is {usage_percent:.1}% full");
            } else if usage_percent > 80.0 && status == HealthStatus::Healthy {
                status = HealthStatus::Warning;
                message = format!("Disk space warning: {mount_point} is {usage_percent:.1}% full");
            }
        }

        HealthCheck {
            name: "disk_space".to_string(),
            status,
            message,
            timestamp: crate::monitor::utils::time::now_nanos(),
            details,
        }
    }

    /// Check memory usage (static version)
    async fn check_memory_usage_static() -> HealthCheck {
        let mut system = System::new_all();
        system.refresh_memory();

        let memory_total = system.total_memory();
        let memory_used = system.used_memory();
        let usage_percent = (memory_used as f64 / memory_total as f64) * 100.0;

        let (status, message) = if usage_percent > 90.0 {
            (
                HealthStatus::Critical,
                format!("Memory usage critical: {usage_percent:.1}%"),
            )
        } else if usage_percent > 80.0 {
            (
                HealthStatus::Warning,
                format!("Memory usage high: {usage_percent:.1}%"),
            )
        } else {
            (
                HealthStatus::Healthy,
                format!("Memory usage normal: {usage_percent:.1}%"),
            )
        };

        let mut details = FxHashMap::default();
        details.insert("usage_percent".to_string(), format!("{usage_percent:.1}%"));
        details.insert(
            "used_mb".to_string(),
            format!("{}", memory_used / 1024 / 1024),
        );
        details.insert(
            "total_mb".to_string(),
            format!("{}", memory_total / 1024 / 1024),
        );

        HealthCheck {
            name: "memory_usage".to_string(),
            status,
            message,
            timestamp: crate::monitor::utils::time::now_nanos(),
            details,
        }
    }

    /// Check data directories (static version)
    async fn check_data_directories_static(config: &MonitorConfig) -> HealthCheck {
        let directories = [&config.storage.base_path, &config.storage.market_data_path];

        let mut status = HealthStatus::Healthy;
        let mut message = "All data directories accessible".to_string();
        let mut details = FxHashMap::default();

        for (i, dir) in directories.iter().enumerate() {
            let dir_name = format!("directory_{i}");

            if !dir.exists() {
                status = HealthStatus::Critical;
                message = format!("Data directory missing: {}", dir.display());
                details.insert(dir_name, "missing".to_string());
            } else if !dir.is_dir() {
                status = HealthStatus::Critical;
                message = format!("Data path is not a directory: {}", dir.display());
                details.insert(dir_name, "not_directory".to_string());
            } else {
                details.insert(dir_name, "ok".to_string());
            }
        }

        HealthCheck {
            name: "data_directories".to_string(),
            status,
            message,
            timestamp: crate::monitor::utils::time::now_nanos(),
            details,
        }
    }

    /// Check file permissions (static version)
    async fn check_file_permissions_static(config: &MonitorConfig) -> HealthCheck {
        use std::fs::OpenOptions;

        let test_file = config.storage.market_data_path.join(".permission_test");

        let (status, message) = match OpenOptions::new()
            .create(true)
            .write(true)
            .truncate(true)
            .open(&test_file)
        {
            Ok(_) => {
                // Clean up test file
                let _ = std::fs::remove_file(&test_file);
                (
                    HealthStatus::Healthy,
                    "File permissions are adequate".to_string(),
                )
            }
            Err(e) => (
                HealthStatus::Critical,
                format!("Cannot write to data directory: {e}"),
            ),
        };

        HealthCheck {
            name: "file_permissions".to_string(),
            status,
            message,
            timestamp: crate::monitor::utils::time::now_nanos(),
            details: FxHashMap::default(),
        }
    }
}

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

    #[tokio::test]
    async fn test_system_monitor() {
        let config = MonitorConfig::default();
        let system_monitor = SystemMonitor::new(config);

        system_monitor.start_monitoring().await;

        let system_metrics = system_monitor.get_system_metrics().await;
        println!("System metrics: {system_metrics:?}");

        let app_metrics = system_monitor.get_app_metrics().await;
        println!("App metrics: {app_metrics:?}");
    }
}