per-peer stats

src/bin/rust-bfdd.rs

@@ -3,7 +3,7 @@ use futures::future;
 use rust_bfd::config::{Config, ConfigError, ResolvedConfig, ResolvedPeer};
 use rust_bfd::events::{EventMessageSink, ScriptHookSink, StateChangeEvent};
 use rust_bfd::{bfd::*, events::EventMessage};
-use rust_bfd::{set_ttl_or_hop_limit, BfdSessionStats};
+use rust_bfd::{set_ttl_or_hop_limit, AtomicInstant, AtomicIntervalStats, BfdSessionStats};
 use std::{
     collections::{HashMap, HashSet},
     error::Error,
@@ -124,8 +124,10 @@ struct BfdSessionState {
     detection_time: Atomic<BfdInterval>, // mutable by rx thread
     poll_mode: Atomic<bool>,             // mutable by rx thread
 
-    control_packets_rx: Atomic<u64>,
-    control_packets_tx: Atomic<u64>,
+    rx_stats: AtomicIntervalStats, // reset each time the session enters Up
+    tx_stats: AtomicIntervalStats, // reset each time the session enters Up
+    last_up: AtomicInstant,
+    last_down: AtomicInstant,
     last_state_change: Mutex<Instant>,
     last_remote_diag: Atomic<BfdDiagnostic>,
 }
@@ -221,11 +223,19 @@ impl BfdSessionState {
     fn set_detection_time(&self, value: BfdInterval) {
         self.detection_time.store(value, ORDERING)
     }
-    fn control_packets_rx(&self) -> u64 {
-        self.control_packets_rx.load(ORDERING)
+    fn record_rx(&self) {
+        self.rx_stats.record(Instant::now());
     }
-    fn control_packets_tx(&self) -> u64 {
-        self.control_packets_tx.load(ORDERING)
+    fn record_tx(&self) {
+        self.tx_stats.record(Instant::now());
+    }
+    fn mark_up(&self) {
+        self.rx_stats.reset();
+        self.tx_stats.reset();
+        self.last_up.store(Instant::now());
+    }
+    fn mark_down(&self) {
+        self.last_down.store(Instant::now());
     }
     fn last_state_change(&self) -> Instant {
         *self.last_state_change.try_lock().unwrap()
@@ -266,8 +276,10 @@ impl From<&BfdSessionState> for BfdSessionStats {
             remote_discr: state.remote_discr(),
             state: state.session_state(),
             last_diag: state.last_remote_diag(),
-            control_packets_rx: state.control_packets_rx(),
-            control_packets_tx: state.control_packets_tx(),
+            rx: state.rx_stats.snapshot(),
+            tx: state.tx_stats.snapshot(),
+            last_up: state.last_up.load(),
+            last_down: state.last_down.load(),
             last_change: state.last_state_change(),
             detect_time: state.detection_time(),
             base_interval: BfdInterval::from_secs(0), // TODO: This is a hack and will be patched up in the session
@@ -483,8 +495,10 @@ impl BfdSession {
                 detection_time: Atomic::new(BfdInterval::from_secs(0)),
                 poll_mode: Atomic::new(false),
 
-                control_packets_rx: Atomic::new(0),
-                control_packets_tx: Atomic::new(0),
+                rx_stats: AtomicIntervalStats::new(),
+                tx_stats: AtomicIntervalStats::new(),
+                last_up: AtomicInstant::new(),
+                last_down: AtomicInstant::new(),
                 last_state_change: Mutex::new(Instant::now()),
                 last_remote_diag: Atomic::new(BfdDiagnostic::None),
             },
@@ -514,6 +528,7 @@ impl BfdSession {
         *self.state.last_state_change.lock().await = Instant::now();
         match (self.state.session_state(), new_state) {
             (BfdState::Up, BfdState::AdminDown | BfdState::Down | BfdState::Init) => {
+                self.state.mark_down();
                 if let Some(watchdog) = self.rx_watchdog.as_mut() {
                     watchdog.send(WatchdogReset::Terminate).await.unwrap();
                 };
@@ -530,6 +545,7 @@ impl BfdSession {
                 self.update_transmit_interval().await;
             }
             (BfdState::AdminDown | BfdState::Down | BfdState::Init, BfdState::Up) => {
+                self.state.mark_up();
                 self.start_watchdog().await;
                 self.state.set_local_diag(BfdDiagnostic::None); // This doesn't seem to be required by the spec but seems sensible and at least FRR does it.
                                                                 // Advertise the configured interval now, but per section 6.8.3 keep transmitting at the old (slower)
@@ -601,7 +617,7 @@ impl BfdSession {
         };
         let socket = self.state.control_sock.clone();
         let dest = self.state.peer_addr;
-        self.state.control_packets_tx.fetch_add(1, ORDERING);
+        self.state.record_tx();
         debug!("tx packet: {:?}", packet);
         if let Err(e) = socket
             .send_to(
@@ -674,7 +690,7 @@ impl BfdSession {
         if let Some(watchdog) = self.rx_watchdog.as_ref() {
             let _ = watchdog.send(WatchdogReset::ReceivedPacket).await;
         }
-        self.state.control_packets_rx.fetch_add(1, ORDERING);
+        self.state.record_rx();
         let received_state = match p.flags.state() {
             Err(_) => {
                 warn!("Invalid state, discarding");

src/lib.rs

@@ -6,7 +6,8 @@ pub mod events;
 use std::{
     fmt::{self, Display},
     net::IpAddr,
-    time::Instant,
+    sync::atomic::{AtomicU64, Ordering},
+    time::{Duration, Instant},
 };
 
 use log::debug;
@@ -15,6 +16,157 @@ use tokio::net::UdpSocket;
 
 use crate::bfd::*;
 
+/// A snapshot of the packet count and inter-packet interval statistics for one direction of a session.
+#[derive(Debug, Clone, Copy, Default)]
+pub struct IntervalStats {
+    count: u64,
+    min: Option<Duration>,
+    max: Option<Duration>,
+    mean: Option<Duration>,
+}
+impl IntervalStats {
+    pub fn count(&self) -> u64 {
+        self.count
+    }
+    pub fn min(&self) -> Option<Duration> {
+        self.min
+    }
+    pub fn max(&self) -> Option<Duration> {
+        self.max
+    }
+    pub fn mean(&self) -> Option<Duration> {
+        self.mean
+    }
+}
+impl Display for IntervalStats {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        let ms = |d: Option<Duration>| {
+            d.map_or("-".to_string(), |d| {
+                format!("{:.0}", d.as_secs_f64() * 1000.0)
+            })
+        };
+        write!(
+            f,
+            "{} pkts, interval min/avg/max {}/{}/{} ms",
+            self.count,
+            ms(self.min),
+            ms(self.mean),
+            ms(self.max)
+        )
+    }
+}
+
+/// An optional `Instant` that can be shared between threads, stored as nanoseconds since its construction.
+/// Instants earlier than the construction time are clamped to it.
+#[derive(Debug)]
+pub struct AtomicInstant {
+    epoch: Instant,
+    nanos: AtomicU64,
+}
+impl AtomicInstant {
+    const NONE: u64 = u64::MAX;
+
+    pub fn new() -> Self {
+        Self {
+            epoch: Instant::now(),
+            nanos: AtomicU64::new(Self::NONE),
+        }
+    }
+    pub fn load(&self) -> Option<Instant> {
+        match self.nanos.load(Ordering::Relaxed) {
+            Self::NONE => None,
+            n => Some(self.epoch + Duration::from_nanos(n)),
+        }
+    }
+    pub fn store(&self, value: Instant) {
+        self.swap(value);
+    }
+    pub fn swap(&self, value: Instant) -> Option<Instant> {
+        let n = u64::try_from(value.saturating_duration_since(self.epoch).as_nanos())
+            .unwrap_or(Self::NONE - 1)
+            .min(Self::NONE - 1);
+        match self.nanos.swap(n, Ordering::Relaxed) {
+            Self::NONE => None,
+            prev => Some(self.epoch + Duration::from_nanos(prev)),
+        }
+    }
+    pub fn clear(&self) {
+        self.nanos.store(Self::NONE, Ordering::Relaxed);
+    }
+}
+impl Default for AtomicInstant {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+/// Lock-free accumulator for packet counts and inter-packet intervals. Updates are not atomic as a group, so it
+/// is intended for a single writer; a concurrent reader may observe a mix of old and new values.
+#[derive(Debug)]
+pub struct AtomicIntervalStats {
+    count: AtomicU64,
+    last: AtomicInstant,
+    min_nanos: AtomicU64,
+    max_nanos: AtomicU64,
+    total_nanos: AtomicU64,
+}
+impl AtomicIntervalStats {
+    pub fn new() -> Self {
+        Self {
+            count: AtomicU64::new(0),
+            last: AtomicInstant::new(),
+            min_nanos: AtomicU64::new(u64::MAX),
+            max_nanos: AtomicU64::new(0),
+            total_nanos: AtomicU64::new(0),
+        }
+    }
+    pub fn record(&self, now: Instant) {
+        self.count.fetch_add(1, Ordering::Relaxed);
+        if let Some(prev) = self.last.swap(now) {
+            let nanos =
+                u64::try_from(now.saturating_duration_since(prev).as_nanos()).unwrap_or(u64::MAX);
+            self.min_nanos.fetch_min(nanos, Ordering::Relaxed);
+            self.max_nanos.fetch_max(nanos, Ordering::Relaxed);
+            self.total_nanos.fetch_add(nanos, Ordering::Relaxed);
+        }
+    }
+    pub fn reset(&self) {
+        self.count.store(0, Ordering::Relaxed);
+        self.last.clear();
+        self.min_nanos.store(u64::MAX, Ordering::Relaxed);
+        self.max_nanos.store(0, Ordering::Relaxed);
+        self.total_nanos.store(0, Ordering::Relaxed);
+    }
+    pub fn snapshot(&self) -> IntervalStats {
+        let count = self.count.load(Ordering::Relaxed);
+        let intervals = count.saturating_sub(1);
+        if intervals == 0 {
+            return IntervalStats {
+                count,
+                ..Default::default()
+            };
+        }
+        let total = self.total_nanos.load(Ordering::Relaxed);
+        IntervalStats {
+            count,
+            min: Some(Duration::from_nanos(self.min_nanos.load(Ordering::Relaxed))),
+            max: Some(Duration::from_nanos(self.max_nanos.load(Ordering::Relaxed))),
+            mean: Some(Duration::from_nanos(total / intervals)),
+        }
+    }
+}
+impl Default for AtomicIntervalStats {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+fn fmt_ago(t: Option<Instant>) -> String {
+    t.map_or("never".to_string(), |t| {
+        format!("{}s ago", t.elapsed().as_secs())
+    })
+}
+
 #[derive(Debug)]
 pub struct BfdSessionStats {
     pub local_ip: IpAddr,
@@ -23,8 +175,11 @@ pub struct BfdSessionStats {
     pub remote_discr: BfdDiscriminator,
     pub state: BfdState,
     pub last_diag: BfdDiagnostic,
-    pub control_packets_rx: u64,
-    pub control_packets_tx: u64,
+    /// Packet statistics since the session last entered Up
+    pub rx: IntervalStats,
+    pub tx: IntervalStats,
+    pub last_up: Option<Instant>,
+    pub last_down: Option<Instant>,
     pub last_change: Instant,
     pub detect_time: BfdInterval,
     pub base_interval: BfdInterval,
@@ -40,9 +195,12 @@ impl Display for BfdSessionStats {
         writeln!(
             f,
             "{:<20} {:<9} {:>6} {:>6}",
-            self.remote_ip, self.state, self.control_packets_rx, self.control_packets_tx
+            self.remote_ip,
+            self.state,
+            self.rx.count(),
+            self.tx.count()
         )?;
-        write!(
+        writeln!(
             f,
             "  LocalDiscr:{} RemoteDiscr:{} DetectTime: {} TxTime: {} Diag:{} Last: {}s",
             self.local_discr,
@@ -51,6 +209,14 @@ impl Display for BfdSessionStats {
             self.base_interval,
             self.last_diag,
             Instant::now().duration_since(self.last_change).as_secs()
+        )?;
+        writeln!(f, "  Rx: {}", self.rx)?;
+        writeln!(f, "  Tx: {}", self.tx)?;
+        write!(
+            f,
+            "  Last Up: {}  Last Down: {}",
+            fmt_ago(self.last_up),
+            fmt_ago(self.last_down)
         )
     }
 }
@@ -70,7 +236,68 @@ pub fn set_ttl_or_hop_limit(sock: &UdpSocket, hop_limit: u8) -> Result<(), std::
             debug!("Got {} after setting {}", res, i32_hop_limit);
             Ok(())
         }
-    } else { 
+    } else {
         panic!("UdpSocket with no address family ?!")
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn interval_stats_empty() {
+        let s = AtomicIntervalStats::new().snapshot();
+        assert_eq!(s.count(), 0);
+        assert_eq!((s.min(), s.max(), s.mean()), (None, None, None));
+    }
+
+    #[test]
+    fn interval_stats_single_packet_has_no_intervals() {
+        let stats = AtomicIntervalStats::new();
+        stats.record(Instant::now());
+        let s = stats.snapshot();
+        assert_eq!(s.count(), 1);
+        assert_eq!((s.min(), s.max(), s.mean()), (None, None, None));
+    }
+
+    #[test]
+    fn interval_stats_min_max_mean() {
+        let stats = AtomicIntervalStats::new();
+        let t0 = Instant::now();
+        for ms in [0, 100, 400, 500] {
+            stats.record(t0 + Duration::from_millis(ms));
+        }
+        let s = stats.snapshot();
+        assert_eq!(s.count(), 4);
+        assert_eq!(s.min(), Some(Duration::from_millis(100)));
+        assert_eq!(s.max(), Some(Duration::from_millis(300)));
+        assert_eq!(s.mean().unwrap().as_millis(), 166);
+    }
+
+    #[test]
+    fn interval_stats_reset() {
+        let stats = AtomicIntervalStats::new();
+        let t0 = Instant::now();
+        stats.record(t0);
+        stats.record(t0 + Duration::from_millis(100));
+        stats.reset();
+        let s = stats.snapshot();
+        assert_eq!(s.count(), 0);
+        assert_eq!((s.min(), s.max(), s.mean()), (None, None, None));
+        stats.record(t0 + Duration::from_millis(900));
+        stats.record(t0 + Duration::from_millis(1000));
+        assert_eq!(stats.snapshot().max(), Some(Duration::from_millis(100)));
+    }
+
+    #[test]
+    fn atomic_instant_roundtrip() {
+        let a = AtomicInstant::new();
+        assert_eq!(a.load(), None);
+        let t = Instant::now() + Duration::from_secs(5);
+        assert_eq!(a.swap(t), None);
+        assert_eq!(a.load(), Some(t));
+        a.clear();
+        assert_eq!(a.load(), None);
+    }
+}