feat(observability): Export ping related metrics

src/pinger.rs

@@ -1,10 +1,13 @@
 use crate::MeasurementType;
+use crate::metrics::Metric;
+use crate::time::Time;
+use crate::util::RwLockExt;
 use icmp_socket2::socket::IcmpSocket;
 use icmp_socket2::{IcmpSocket4, Icmpv4Packet};
-use crate::util::RwLockExt;
 use log::{debug, warn};
+use rustix::time::ClockId;
 use std::net::{IpAddr, Ipv4Addr};
-use std::sync::mpsc::Sender;
+use std::sync::mpsc::{Sender, SyncSender};
 use std::sync::{Arc, RwLock};
 use std::time::{Duration, Instant};
 use std::{io, thread};
@@ -24,8 +27,9 @@ pub enum PingError {
 
 pub struct PingReply {
     pub reflector: IpAddr,
+    pub measurement_type: MeasurementType,
     pub seq: u16,
-    pub rtt: i64,
+    pub rtt: f64,
     pub current_time: i64,
     pub down_time: f64,
     pub up_time: f64,
@@ -37,9 +41,7 @@ pub struct PingReply {
 
 fn open_socket(type_: MeasurementType) -> io::Result<IcmpSocket4> {
     match type_ {
-        MeasurementType::Icmp | MeasurementType::IcmpTimestamps => {
-            IcmpSocket4::new()
-        },
+        MeasurementType::Icmp | MeasurementType::IcmpTimestamps => IcmpSocket4::new(),
         _ => {
             unimplemented!()
         }
@@ -53,9 +55,13 @@ pub trait PingListener {
         type_: MeasurementType,
         reflectors_lock: Arc<RwLock<Vec<IpAddr>>>,
         stats_sender: Sender<PingReply>,
+        metrics_sender: SyncSender<Metric>,
     ) -> anyhow::Result<()> {
         let socket = &mut open_socket(type_)?;
 
+        // We need a realtime clock when exporting metrics
+        let clock = Time::new(ClockId::Realtime);
+
         loop {
             let (pkt, sender) = match socket.rcv_from() {
                 Ok(val) => val,
@@ -69,7 +75,7 @@ pub trait PingListener {
                 continue;
             }
 
-            let reply = match self.parse_packet(id, addr, pkt) {
+            let reply = match self.parse_packet(id, addr, type_, pkt) {
                 Ok(val) => val,
                 Err(_) => {
                     // parse_packet will throw an error if it's an unknown protocol etc.
@@ -78,7 +84,28 @@ pub trait PingListener {
                 }
             };
 
-            debug!("Type: {:4}  | Reflector IP: {:>15}  | Seq: {:5}  | Current time: {:8}  |  Originate: {:8}  |  Received time: {:8}  |  Transmit time : {:8}  |  RTT: {:8}  | UL time: {:5}  | DL time: {:5}", "ICMP", addr.to_string(), reply.seq, reply.current_time, reply.originate_timestamp, reply.receive_timestamp, reply.transmit_timestamp, reply.rtt, reply.up_time, reply.down_time);
+            let _ = metrics_sender.try_send(Metric::Ping {
+                reflector: reply.reflector,
+                measurement_type: reply.measurement_type,
+                rtt: reply.rtt,
+                up_time: reply.up_time,
+                down_time: reply.down_time,
+                timestamp_ns: clock.as_nanos(),
+            });
+
+            debug!(
+                "Type: {:4}  | Reflector IP: {:>15}  | Seq: {:5}  | Current time: {:8}  |  Originate: {:8}  |  Received time: {:8}  |  Transmit time : {:8}  |  RTT: {:8}  | UL time: {:5}  | DL time: {:5}",
+                "ICMP",
+                addr.to_string(),
+                reply.seq,
+                reply.current_time,
+                reply.originate_timestamp,
+                reply.receive_timestamp,
+                reply.transmit_timestamp,
+                reply.rtt,
+                reply.up_time,
+                reply.down_time
+            );
             if let Err(e) = stats_sender.send(reply) {
                 warn!("Stats channel closed, stopping listener: {}", e);
                 break Ok(());
@@ -86,7 +113,13 @@ pub trait PingListener {
         }
     }
 
-    fn parse_packet(&self, id: u16, reflector: IpAddr, pkt: Icmpv4Packet) -> Result<PingReply, PingError>;
+    fn parse_packet(
+        &self,
+        id: u16,
+        reflector: IpAddr,
+        measurement_type: MeasurementType,
+        pkt: Icmpv4Packet,
+    ) -> Result<PingReply, PingError>;
 }
 
 pub trait PingSender {

src/pinger_icmp.rs

@@ -1,6 +1,7 @@
 use std::net::IpAddr;
 use std::time::Instant;
 
+use crate::config::MeasurementType;
 use crate::pinger::{PingError, PingListener, PingReply, PingSender};
 use crate::time::Time;
 use icmp_socket2::Icmpv4Message;
@@ -14,7 +15,13 @@ pub struct PingerICMPEchoSender {}
 
 impl PingListener for PingerICMPEchoListener {
     // Result: RTT, down time, up time
-    fn parse_packet(&self, id: u16, reflector: IpAddr, pkt: Icmpv4Packet) -> Result<PingReply, PingError> {
+    fn parse_packet(
+        &self,
+        id: u16,
+        reflector: IpAddr,
+        measurement_type: MeasurementType,
+        pkt: Icmpv4Packet,
+    ) -> Result<PingReply, PingError> {
         match pkt.typ {
             0 => {
                 if let Icmpv4Message::EchoReply {
@@ -33,32 +40,37 @@ impl PingListener for PingerICMPEchoListener {
                     let time_sent = match payload.as_slice().try_into() {
                         Ok(bytes) => u64::from_be_bytes(bytes) as i64,
                         Err(_) => {
-                            return Err(PingError::InvalidPacket(format!("Expected 8 bytes payload, but found {}", payload.len())))
+                            return Err(PingError::InvalidPacket(format!(
+                                "Expected 8 bytes payload, but found {}",
+                                payload.len()
+                            )));
                         }
                     };
 
                     let clock = Time::new(ClockId::Monotonic);
                     let time_ms = clock.to_milliseconds() as i64;
 
-                    let rtt: i64 = time_ms - time_sent;
+                    let rtt = (time_ms - time_sent) as f64;
                     Ok(PingReply {
                         reflector,
+                        measurement_type,
                         seq: sequence,
                         rtt,
                         current_time: time_ms,
-                        down_time: (rtt / 2) as f64,
-                        up_time: (rtt / 2) as f64,
+                        down_time: (rtt / 2.0) as f64,
+                        up_time: (rtt / 2.0) as f64,
                         originate_timestamp: time_sent,
                         receive_timestamp: 0,
                         transmit_timestamp: 0,
                         last_receive_time_s: Instant::now(),
                     })
                 } else {
-                    Err(PingError::InvalidPacket(format!("Packet had type {:?}, but did not match the structure", pkt.typ)))
+                    Err(PingError::InvalidPacket(format!(
+                        "Packet had type {:?}, but did not match the structure",
+                        pkt.typ
+                    )))
+                }
             }
-
-                
-            },
             type_ => Err(PingError::InvalidType(format!("{:?}", type_))),
         }
     }

src/pinger_icmp_ts.rs

@@ -1,8 +1,9 @@
+use crate::config::MeasurementType;
 use crate::pinger::{PingError, PingListener, PingReply, PingSender};
 use crate::time::Time;
-use icmp_socket2::packet::WithTimestampRequest;
 use icmp_socket2::Icmpv4Message;
 use icmp_socket2::Icmpv4Packet;
+use icmp_socket2::packet::WithTimestampRequest;
 use rustix::thread::ClockId;
 use std::net::IpAddr;
 use std::time::Instant;
@@ -13,7 +14,13 @@ pub struct PingerICMPTimestampSender {}
 
 impl PingListener for PingerICMPTimestampListener {
     // Result: RTT, down time, up time
-    fn parse_packet(&self, id: u16, reflector: IpAddr, pkt: Icmpv4Packet) -> Result<PingReply, PingError> {
+    fn parse_packet(
+        &self,
+        id: u16,
+        reflector: IpAddr,
+        measurement_type: MeasurementType,
+        pkt: Icmpv4Packet,
+    ) -> Result<PingReply, PingError> {
         match pkt.typ {
             // 14 = Timestamp reply
             14 => {
@@ -35,28 +42,30 @@ impl PingListener for PingerICMPTimestampListener {
                     let time_now = Time::new(ClockId::Realtime);
                     let time_since_midnight = time_now.get_time_since_midnight();
 
-                    let rtt: i64 = time_since_midnight - originate as i64;
-                    let dl_time: i64 = time_since_midnight - transmit as i64;
-                    let ul_time: i64 = receive as i64 - originate as i64;
+                    let rtt = (time_since_midnight - originate as i64) as f64;
+                    let dl_time = (time_since_midnight - transmit as i64) as f64;
+                    let ul_time = (receive as i64 - originate as i64) as f64;
 
                     Ok(PingReply {
                         reflector,
+                        measurement_type,
                         seq: sequence,
                         rtt,
                         current_time: time_since_midnight,
-                        down_time: dl_time as f64,
-                        up_time: ul_time as f64,
+                        down_time: dl_time,
+                        up_time: ul_time,
                         originate_timestamp: originate as i64,
                         receive_timestamp: receive as i64,
                         transmit_timestamp: transmit as i64,
                         last_receive_time_s: Instant::now(),
                     })
                 } else {
-                    Err(PingError::InvalidPacket(format!("Packet had type {:?}, but did not match the structure", pkt.typ)))
+                    Err(PingError::InvalidPacket(format!(
+                        "Packet had type {:?}, but did not match the structure",
+                        pkt.typ
+                    )))
+                }
             }
-
-                
-            },
             type_ => Err(PingError::InvalidType(format!("{:?}", type_))),
         }
     }