oxspeak_server/src/rtc/metrics.rs

use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, Instant};

use crate::metrics::{Metrics, MetricsSnapshot};

/// Compteurs atomiques pour les métriques de la voix / WebRTC.
#[derive(Debug, Default)]
pub struct VoiceMetrics {
    /// Nombre total de datagrammes / paquets reçus.
    pub packets_received: AtomicU64,
    /// Volume total d'octets reçus.
    pub bytes_received: AtomicU64,
    /// Nombre total de datagrammes / paquets retransmis.
    pub packets_sent: AtomicU64,
    /// Volume total d'octets retransmis.
    pub bytes_sent: AtomicU64,
    /// Paquets ignorés ou rejetés.
    pub packets_dropped: AtomicU64,
    /// Nombre d'erreurs d'émission.
    pub send_errors: AtomicU64,
    /// Nombre d'erreurs de réception.
    pub recv_errors: AtomicU64,
}

impl VoiceMetrics {
    /// Crée un jeu de métriques vide enveloppé dans un [`Arc`].
    pub fn new() -> Arc<Self> {
        Arc::new(Self::default())
    }

    /// Enregistre la réception d'un paquet de `bytes` octets.
    #[inline]
    pub fn inc_received(&self, bytes: u64) {
        self.packets_received.fetch_add(1, Ordering::Relaxed);
        self.bytes_received.fetch_add(bytes, Ordering::Relaxed);
    }

    /// Enregistre l'émission d'un paquet de `bytes` octets.
    #[inline]
    pub fn inc_sent(&self, bytes: u64) {
        self.packets_sent.fetch_add(1, Ordering::Relaxed);
        self.bytes_sent.fetch_add(bytes, Ordering::Relaxed);
    }

    /// Enregistre un paquet ignoré.
    #[inline]
    pub fn inc_dropped(&self) {
        self.packets_dropped.fetch_add(1, Ordering::Relaxed);
    }

    /// Enregistre un échec d'émission non fatal.
    #[inline]
    pub fn inc_send_error(&self) {
        self.send_errors.fetch_add(1, Ordering::Relaxed);
    }

    /// Enregistre un échec de réception.
    #[inline]
    pub fn inc_recv_error(&self) {
        self.recv_errors.fetch_add(1, Ordering::Relaxed);
    }

    /// Prend un instantané cohérent de tous les compteurs.
    pub fn snapshot(&self) -> VoiceMetricsSnapshot {
        VoiceMetricsSnapshot {
            taken_at: Instant::now(),
            packets_received: self.packets_received.load(Ordering::Relaxed),
            bytes_received: self.bytes_received.load(Ordering::Relaxed),
            packets_sent: self.packets_sent.load(Ordering::Relaxed),
            bytes_sent: self.bytes_sent.load(Ordering::Relaxed),
            packets_dropped: self.packets_dropped.load(Ordering::Relaxed),
            send_errors: self.send_errors.load(Ordering::Relaxed),
            recv_errors: self.recv_errors.load(Ordering::Relaxed),
        }
    }
}

impl Metrics for VoiceMetrics {
    type Snapshot = VoiceMetricsSnapshot;

    fn snapshot(&self) -> VoiceMetricsSnapshot {
        self.snapshot()
    }
}

/// Lecture cohérente de l'ensemble des compteurs à un instant T.
#[derive(Debug, Clone, Copy)]
pub struct VoiceMetricsSnapshot {
    pub taken_at: Instant,
    pub packets_received: u64,
    pub bytes_received: u64,
    pub packets_sent: u64,
    pub bytes_sent: u64,
    pub packets_dropped: u64,
    pub send_errors: u64,
    pub recv_errors: u64,
}

impl VoiceMetricsSnapshot {
    /// Calcule les taux moyens par seconde depuis un snapshot précédent.
    pub fn rates_since(&self, previous: &Self) -> VoiceRates {
        let secs = self
            .taken_at
            .duration_since(previous.taken_at)
            .as_secs_f64()
            .max(f64::EPSILON);

        VoiceRates {
            packets_received_per_sec: self
                .packets_received
                .saturating_sub(previous.packets_received)
                as f64
                / secs,
            bytes_received_per_sec: self.bytes_received.saturating_sub(previous.bytes_received)
                as f64
                / secs,
            packets_sent_per_sec: self.packets_sent.saturating_sub(previous.packets_sent) as f64
                / secs,
            bytes_sent_per_sec: self.bytes_sent.saturating_sub(previous.bytes_sent) as f64 / secs,
            packets_dropped_per_sec: self
                .packets_dropped
                .saturating_sub(previous.packets_dropped)
                as f64
                / secs,
        }
    }
}

impl MetricsSnapshot for VoiceMetricsSnapshot {
    fn taken_at(&self) -> Instant {
        self.taken_at
    }
}

/// Taux moyens par seconde calculés entre deux [`VoiceMetricsSnapshot`].
#[derive(Debug, Clone, Copy)]
pub struct VoiceRates {
    pub packets_received_per_sec: f64,
    pub bytes_received_per_sec: f64,
    pub packets_sent_per_sec: f64,
    pub bytes_sent_per_sec: f64,
    pub packets_dropped_per_sec: f64,
}

pub fn spawn_reporter(metrics: Arc<VoiceMetrics>, interval: Duration) {
    tokio::spawn(async move {
        let mut ticker = tokio::time::interval(interval);
        ticker.tick().await;

        let mut prev_snapshot = metrics.snapshot();

        loop {
            ticker.tick().await;

            let current = metrics.snapshot();
            let rates = current.rates_since(&prev_snapshot);

            tracing::info!(
                pkts_rx = current.packets_received,
                bytes_rx = current.bytes_received,
                pkts_tx = current.packets_sent,
                bytes_tx = current.bytes_sent,
                pkts_dropped = current.packets_dropped,
                send_errors = current.send_errors,
                recv_errors = current.recv_errors,
                pkts_rx_s = format!("{:.1}", rates.packets_received_per_sec),
                bytes_rx_s = format!("{:.0}", rates.bytes_received_per_sec),
                pkts_tx_s = format!("{:.1}", rates.packets_sent_per_sec),
                bytes_tx_s = format!("{:.0}", rates.bytes_sent_per_sec),
                pkts_dropped_s = format!("{:.1}", rates.packets_dropped_per_sec),
                "Voice metrics"
            );

            prev_snapshot = current;
        }
    });
}