init

src/rtc/metrics.rs

@@ -0,0 +1,178 @@
+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;
+        }
+    });
+}