feat(measurement): integrate IRTT with full configuration and per-stream tracking

Extend reflector configuration to support IRTT-specific parameters (ports, HMAC) and track per-stream measurement state, enabling dynamic selection and authentication for IRTT probes.

src/controller.rs

@@ -1,100 +1,140 @@
-use std::{collections::HashMap, net::IpAddr, time::Duration};
+use std::{
+    collections::HashMap,
+    net::SocketAddr,
+    sync::{Arc, RwLock},
+    time::Duration,
+};
 
 use flume::{Receiver, SendError, Sender};
 use thiserror::Error;
 
-use crate::measurement::{
-    icmp::IcmpBinding,
-    model::{MeasurementSource, MeasurementStream, MeasurementStreamId, ProbeSchedule},
+use crate::{
+    measurement::{
+        icmp::IcmpBinding,
+        irtt::IrttBinding,
+        model::{MeasurementSource, MeasurementStream, MeasurementStreamId, ProbeSchedule},
+    },
+    settings::ReflectorDefinition,
 };
 
 #[derive(Debug, Error)]
 pub enum ControllerError {
     #[error("ICMP channel error")]
     IcmpChannelError(#[from] SendError<Vec<IcmpBinding>>),
+    #[error("IRTT channel error")]
+    IrttChannelError(#[from] SendError<Vec<IrttBinding>>),
 }
 
 pub struct Controller {
-    reflector_rx: Receiver<Vec<IpAddr>>,
+    reflector_rx: Receiver<Vec<SocketAddr>>,
     icmp_binding_tx: Sender<Vec<IcmpBinding>>,
-
+    irtt_binding_tx: Sender<Vec<IrttBinding>>,
+    reflectors: HashMap<SocketAddr, ReflectorDefinition>,
+    active_streams: Arc<RwLock<Vec<MeasurementStream>>>,
+    endpoint_streams: Arc<RwLock<HashMap<SocketAddr, MeasurementStream>>>,
     source: MeasurementSource,
     period: Duration,
-
     next_stream_id: u64,
-    stream_ids: HashMap<(IpAddr, MeasurementSource), MeasurementStreamId>,
+    stream_ids: HashMap<(SocketAddr, MeasurementSource), MeasurementStreamId>,
 }
 
 impl Controller {
     pub fn new(
-        reflector_rx: Receiver<Vec<IpAddr>>,
+        reflector_rx: Receiver<Vec<SocketAddr>>,
         icmp_binding_tx: Sender<Vec<IcmpBinding>>,
+        irtt_binding_tx: Sender<Vec<IrttBinding>>,
+        reflectors: HashMap<SocketAddr, ReflectorDefinition>,
+        active_streams: Arc<RwLock<Vec<MeasurementStream>>>,
+        endpoint_streams: Arc<RwLock<HashMap<SocketAddr, MeasurementStream>>>,
+        source: MeasurementSource,
     ) -> Self {
         Self {
             reflector_rx,
             icmp_binding_tx,
-            source: MeasurementSource::IcmpTimestamp,
+            irtt_binding_tx,
+            reflectors,
+            active_streams,
+            endpoint_streams,
+            source,
             period: Duration::from_millis(500),
             next_stream_id: 0,
             stream_ids: HashMap::new(),
         }
     }
 
-    pub async fn run(mut self, initial_reflectors: Vec<IpAddr>) -> Result<(), ControllerError> {
+    pub async fn run(mut self, initial_reflectors: Vec<SocketAddr>) -> Result<(), ControllerError> {
         self.update_reflectors(initial_reflectors).await?;
-
         while let Ok(reflectors) = self.reflector_rx.recv_async().await {
             self.update_reflectors(reflectors).await?;
         }
-
         Ok(())
     }
 
-    fn stream_for(&mut self, peer: IpAddr) -> MeasurementStream {
-        let key = (peer, self.source);
-
+    fn stream_for(&mut self, endpoint: SocketAddr) -> MeasurementStream {
+        let key = (endpoint, self.source);
         let id = *self.stream_ids.entry(key).or_insert_with(|| {
             let id = MeasurementStreamId(self.next_stream_id);
             self.next_stream_id += 1;
             id
         });
-
         MeasurementStream {
             id,
-            peer,
+            peer: endpoint.ip(),
             source: self.source,
         }
     }
 
-    async fn update_reflectors(&mut self, reflectors: Vec<IpAddr>) -> Result<(), ControllerError> {
-        let count = reflectors.len();
-
-        let bindings = reflectors
-            .into_iter()
-            .enumerate()
-            .map(|(index, peer)| {
-                let stream = self.stream_for(peer);
-
-                IcmpBinding::new(
-                    stream,
-                    ProbeSchedule {
-                        period: self.period,
-                        offset: evenly_spaced_offset(self.period, index, count),
-                    },
+    async fn update_reflectors(
+        &mut self,
+        endpoints: Vec<SocketAddr>,
+    ) -> Result<(), ControllerError> {
+        let count = endpoints.len();
+        let mut icmp = Vec::new();
+        let mut irtt = Vec::new();
+        let mut streams = Vec::with_capacity(count);
+        let mut endpoint_streams = HashMap::new();
+        for (index, endpoint) in endpoints.into_iter().enumerate() {
+            let Some((port, hmac)) = self.reflectors.get(&endpoint).map(|definition| {
+                (
+                    definition.irtt_port.unwrap_or(2112),
+                    definition.irtt_hmac.clone(),
                 )
-            })
-            .collect();
-
-        self.icmp_binding_tx
-            .send_async(bindings)
-            .await
-            .map_err(|e| ControllerError::IcmpChannelError(e))
+            }) else {
+                continue;
+            };
+            let stream = self.stream_for(endpoint);
+            let schedule = ProbeSchedule {
+                period: self.period,
+                offset: evenly_spaced_offset(self.period, index, count),
+            };
+            match self.source {
+                MeasurementSource::IcmpEcho | MeasurementSource::IcmpTimestamp => {
+                    icmp.push(IcmpBinding::new(stream, schedule))
+                }
+                MeasurementSource::Irtt => {
+                    irtt.push(IrttBinding::new(stream, schedule, port, hmac))
+                }
+                _ => unreachable!("unsupported reflector measurement source"),
+            }
+            streams.push(stream);
+            endpoint_streams.insert(endpoint, stream);
+        }
+        *self
+            .active_streams
+            .write()
+            .expect("active stream lock poisoned") = streams;
+        *self
+            .endpoint_streams
+            .write()
+            .expect("endpoint stream lock poisoned") = endpoint_streams;
+        self.icmp_binding_tx.send_async(icmp).await?;
+        self.irtt_binding_tx.send_async(irtt).await?;
+        Ok(())
     }
 }
 
 fn evenly_spaced_offset(period: Duration, index: usize, count: usize) -> Duration {
     let nanos = period.as_nanos() * index as u128 / count as u128;
-
     Duration::new(
         (nanos / 1_000_000_000) as u64,
         (nanos % 1_000_000_000) as u32,