From 645ac08daa2d51bb6ac2fb75935a6af54d730d8a Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Mon, 13 Apr 2026 15:01:05 +0700 Subject: [PATCH] feat(observability): Export ping related metrics --- src/pinger.rs | 51 +++++++++++++++++++++++++++++++++++-------- src/pinger_icmp.rs | 30 +++++++++++++++++-------- src/pinger_icmp_ts.rs | 31 ++++++++++++++++---------- 3 files changed, 83 insertions(+), 29 deletions(-) diff --git a/src/pinger.rs b/src/pinger.rs index 041e4b6..421725f 100644 --- a/src/pinger.rs +++ b/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 { 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>>, stats_sender: Sender, + metrics_sender: SyncSender, ) -> 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; + fn parse_packet( + &self, + id: u16, + reflector: IpAddr, + measurement_type: MeasurementType, + pkt: Icmpv4Packet, + ) -> Result; } pub trait PingSender { diff --git a/src/pinger_icmp.rs b/src/pinger_icmp.rs index 7b9dfad..f201fb0 100644 --- a/src/pinger_icmp.rs +++ b/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 { + fn parse_packet( + &self, + id: u16, + reflector: IpAddr, + measurement_type: MeasurementType, + pkt: Icmpv4Packet, + ) -> Result { 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_))), } } diff --git a/src/pinger_icmp_ts.rs b/src/pinger_icmp_ts.rs index 3f91ab4..c3bf5ca 100644 --- a/src/pinger_icmp_ts.rs +++ b/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 { + fn parse_packet( + &self, + id: u16, + reflector: IpAddr, + measurement_type: MeasurementType, + pkt: Icmpv4Packet, + ) -> Result { 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_))), } }