rusty_engine/monitoring/

collector.rs

//! Metric collector implementation using lock-free ring buffers
//!
//! This module provides the MetricCollector that uses lock-free ring buffers
//! for zero-latency metric recording in the hot path.

use crate::monitoring::MonitoringConfig;
use crate::monitoring::metrics::Metric;
use crate::monitoring::ring_buffer::SharedRingBuffer;
use crate::timing::TimingUpkeep;
use anyhow::{Result, anyhow};
use flume::Sender;
use parking_lot::RwLock;
use std::sync::Arc;
use std::thread;
use std::time::Duration;

/// A metric collector that uses lock-free ring buffers
pub struct MetricCollector {
    /// Name of this collector
    name: String,
    /// Ring buffer for metric storage
    buffer: SharedRingBuffer<Metric>,
    /// Channel sender for async processing
    sender: Option<Sender<Vec<Metric>>>,
    /// Configuration
    config: MonitoringConfig,
    /// Collection statistics
    stats: Arc<RwLock<CollectorStats>>,
}

/// Statistics for a metric collector
#[derive(Debug, Default)]
pub struct CollectorStats {
    /// Total number of metrics recorded by this collector
    pub metrics_recorded: u64,
    /// Number of metrics successfully sent for processing
    pub metrics_sent: u64,
    /// Number of metrics dropped due to channel backpressure
    pub metrics_dropped: u64,
    /// Number of times the ring buffer overflowed
    pub buffer_overflows: u64,
    /// Timestamp of the last flush operation in nanoseconds
    pub last_flush_ns: u64,
    /// Total number of flush operations performed
    pub flush_count: u64,
}

impl MetricCollector {
    /// Create a new metric collector
    #[must_use]
    pub fn new(name: String, config: MonitoringConfig) -> Self {
        Self {
            name,
            buffer: SharedRingBuffer::new(config.ring_buffer_size),
            sender: None,
            config,
            stats: Arc::new(RwLock::new(CollectorStats::default())),
        }
    }

    /// Create a collector connected to a processing channel
    pub fn with_channel(
        name: String,
        config: MonitoringConfig,
        sender: Sender<Vec<Metric>>,
    ) -> Self {
        Self {
            name,
            buffer: SharedRingBuffer::new(config.ring_buffer_size),
            sender: Some(sender),
            config,
            stats: Arc::new(RwLock::new(CollectorStats::default())),
        }
    }

    /// Record a metric with minimal overhead
    #[inline(always)]
    pub fn record(&self, metric: Metric) {
        // Fast path - push to ring buffer
        if self.buffer.push(metric) {
            // Use relaxed ordering for stats update to minimize overhead
            let mut stats = self.stats.write();
            stats.metrics_recorded += 1;
        } else {
            // Buffer overflow - update stats
            let mut stats = self.stats.write();
            stats.metrics_dropped += 1;
            stats.buffer_overflows = self.buffer.overflow_count() as u64;
        }
    }

    /// Start the background flushing thread
    pub fn start_flusher(&self) -> Result<thread::JoinHandle<()>> {
        if self.sender.is_none() {
            return Err(anyhow!("No channel configured for flushing"));
        }

        let buffer = self.buffer.clone();
        let sender = self.sender.as_ref().unwrap().clone();
        let flush_interval = Duration::from_millis(self.config.flush_interval_ms);
        let stats = self.stats.clone();

        let handle = thread::spawn(move || {
            loop {
                thread::sleep(flush_interval);

                // Drain metrics from ring buffer
                let metrics = buffer.drain(1024); // Drain up to 1024 metrics at a time

                if !metrics.is_empty() {
                    let count = metrics.len() as u64;

                    // Send to processing channel
                    if sender.send(metrics).is_ok() {
                        let mut stats = stats.write();
                        stats.metrics_sent += count;
                        stats.last_flush_ns = TimingUpkeep::now_ns();
                        stats.flush_count += 1;
                    } else {
                        // Channel closed, exit
                        break;
                    }
                }
            }
        });

        Ok(handle)
    }

    /// Manually flush metrics from the buffer
    #[must_use]
    pub fn flush(&self) -> Vec<Metric> {
        let metrics = self.buffer.drain(usize::MAX);

        if !metrics.is_empty() {
            let mut stats = self.stats.write();
            stats.metrics_sent += metrics.len() as u64;
            stats.last_flush_ns = TimingUpkeep::now_ns();
            stats.flush_count += 1;
        }

        metrics
    }

    /// Get collector statistics
    #[must_use]
    pub fn stats(&self) -> CollectorStats {
        self.stats.read().clone()
    }

    /// Get the collector name
    #[must_use]
    pub fn name(&self) -> &str {
        &self.name
    }

    /// Get the current buffer size
    #[must_use]
    pub fn buffer_size(&self) -> usize {
        self.buffer.len()
    }
}

/// A collection of metric collectors
pub struct CollectorRegistry {
    collectors: Arc<RwLock<Vec<Arc<MetricCollector>>>>,
}

impl CollectorRegistry {
    /// Creates a new collector registry for managing multiple metric collectors
    ///
    /// The registry provides thread-safe storage and retrieval of metric collectors,
    /// allowing different components of the system to register their own collectors.
    ///
    /// # Example
    ///
    /// ```ignore
    /// let registry = CollectorRegistry::new();
    /// let collector = Arc::new(MetricCollector::new("orders".into(), config));
    /// registry.register(collector);
    /// ```
    #[must_use]
    pub fn new() -> Self {
        Self {
            collectors: Arc::new(RwLock::new(Vec::new())),
        }
    }

    /// Register a collector
    pub fn register(&self, collector: Arc<MetricCollector>) {
        self.collectors.write().push(collector);
    }

    /// Get all collectors
    #[must_use]
    pub fn collectors(&self) -> Vec<Arc<MetricCollector>> {
        self.collectors.read().clone()
    }

    /// Flush all collectors
    #[must_use]
    pub fn flush_all(&self) -> Vec<Metric> {
        let mut all_metrics = Vec::new();

        for collector in self.collectors.read().iter() {
            all_metrics.extend(collector.flush());
        }

        all_metrics
    }
}

impl Default for CollectorRegistry {
    fn default() -> Self {
        Self::new()
    }
}

impl Clone for CollectorStats {
    fn clone(&self) -> Self {
        Self {
            metrics_recorded: self.metrics_recorded,
            metrics_sent: self.metrics_sent,
            metrics_dropped: self.metrics_dropped,
            buffer_overflows: self.buffer_overflows,
            last_flush_ns: self.last_flush_ns,
            flush_count: self.flush_count,
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::monitoring::metrics::{Counter, Gauge};

    #[test]
    fn test_metric_collector_basic() {
        let config = MonitoringConfig {
            ring_buffer_size: 64,
            ..Default::default()
        };

        let collector = MetricCollector::new("test".into(), config);

        // Record some metrics
        let counter = Metric::Counter(Counter {
            name: "test_counter".into(),
            value: 1,
            timestamp_ns: 1000,
        });

        collector.record(counter.clone());
        collector.record(counter.clone());

        assert_eq!(collector.buffer_size(), 2);

        // Flush metrics
        let metrics = collector.flush();
        assert_eq!(metrics.len(), 2);
        assert_eq!(collector.buffer_size(), 0);

        // Check stats
        let stats = collector.stats();
        assert_eq!(stats.metrics_recorded, 2);
        assert_eq!(stats.metrics_sent, 2);
        assert_eq!(stats.metrics_dropped, 0);
    }

    #[test]
    fn test_metric_collector_with_channel() {
        let config = MonitoringConfig {
            ring_buffer_size: 64,
            flush_interval_ms: 100,
            ..Default::default()
        };

        let (tx, rx) = flume::unbounded();
        let collector = Arc::new(MetricCollector::with_channel("test".into(), config, tx));

        // Start flusher
        let handle = collector.start_flusher().unwrap();

        // Record metrics
        for i in 0..10 {
            let gauge = Metric::Gauge(Gauge {
                name: "test_gauge".into(),
                value: i as f64,
                timestamp_ns: i as u64,
            });
            collector.record(gauge);
        }

        // Wait for flush
        thread::sleep(Duration::from_millis(200));

        // Check that metrics were sent
        let received = rx.try_recv().unwrap();
        assert_eq!(received.len(), 10);

        // Cleanup
        drop(rx); // Close channel to stop flusher
        let _ = handle.join();
    }

    #[test]
    fn test_collector_registry() {
        let registry = CollectorRegistry::new();

        let config = MonitoringConfig::default();
        let collector1 = Arc::new(MetricCollector::new("collector1".into(), config.clone()));
        let collector2 = Arc::new(MetricCollector::new("collector2".into(), config));

        registry.register(collector1.clone());
        registry.register(collector2.clone());

        // Record metrics in both collectors
        let counter = Metric::Counter(Counter {
            name: "test".into(),
            value: 1,
            timestamp_ns: 1000,
        });

        collector1.record(counter.clone());
        collector2.record(counter.clone());
        collector2.record(counter);

        // Flush all
        let all_metrics = registry.flush_all();
        assert_eq!(all_metrics.len(), 3);
    }

    #[test]
    fn test_buffer_overflow() {
        let config = MonitoringConfig {
            ring_buffer_size: 4, // Very small buffer
            ..Default::default()
        };

        let collector = MetricCollector::new("test".into(), config);

        // Fill buffer (can hold 3 items due to ring buffer implementation)
        for i in 0..5 {
            let counter = Metric::Counter(Counter {
                name: "test".into(),
                value: i,
                timestamp_ns: i,
            });
            collector.record(counter);
        }

        let stats = collector.stats();
        assert!(stats.metrics_dropped > 0);
        assert!(stats.buffer_overflows > 0);
    }
}