fix: fix traceroute output and tests

Author: teo-lohrer-su

src/algorithms/diamond_miner.rs

@@ -84,8 +84,8 @@ impl DiamondMiner {
 
     // echo replies and destination unreachable replies should count towards successors counts
     pub fn links_by_ttl(&self) -> HashMap<TTL, Vec<Link>> {
-        get_links_by_ttl(&self.replies())
-        // get_links_by_ttl(&self.time_exceeded_replies())
+        // get_links_by_ttl(&self.replies())
+        get_links_by_ttl(&self.time_exceeded_replies())
     }
 
     pub fn n_links_by_ttl(&self) -> HashMap<TTL, usize> {
@@ -170,11 +170,17 @@ impl DiamondMiner {
             //     continue;
             // }
             // if the node is in the same subnet as the destination
-            let prefix_length = (32 - (128 - self.mapper_v4.prefix_size.leading_zeros())) as u8;
+            let prefix_length = (1 + 32 - (128 - self.mapper_v4.prefix_size.leading_zeros())) as u8;
+
             let dst_network =
                 ip_network::IpNetwork::new_truncate(self.dst_addr, prefix_length).unwrap();
 
             if dst_network.contains(node) {
+                // println!("network: {:?}", dst_network);
+                // println!(
+                //     "Node {} is in the same subnet as the destination {}, with prefix length {}",
+                //     node, self.dst_addr, prefix_length
+                // );
                 continue;
             }
 
@@ -209,13 +215,13 @@ impl DiamondMiner {
                 // .filter(|l| l.near_ip == Some(node))
                 .count();
 
-            // if a node has no successors, but is not the destination, it is unresolved
             if n_probes >= n_k || node == self.dst_addr {
                 // node is resolved
                 continue;
             }
 
-            if n_probes < n_k && n_successors > 0 {
+            // if n_probes < n_k && n_successors > 0 {
+            if n_probes < n_k {
                 // node is unresolved
                 unresolved_nodes.insert(node);
                 if estimate_successors {

src/algorithms/diamond_miner/tests.rs

@@ -31,7 +31,7 @@ const DEST: [&str; 10] = [
 
 fn diamond_miner() -> DiamondMiner {
     DiamondMiner::new(
-        IpAddr::V4(IP[0].parse().unwrap()),
+        IpAddr::V4(DEST[0].parse().unwrap()),
         1,
         20,
         24000,
@@ -92,9 +92,9 @@ fn test_unresolved_nodes_at_ttl_basic() {
         IpAddr::V4(IP[2].parse().unwrap()),
     ];
 
-    // let links_by_ttl = miner.links_by_ttl();
-    // println!(">> miner.replies_by_round: {:?}", miner.replies_by_round);
-    // println!(">>links_by_ttl: {:?}", links_by_ttl);
+    let links_by_ttl = miner.links_by_ttl();
+    println!(">> miner.replies_by_round: {:?}", miner.replies_by_round);
+    println!(">>links_by_ttl: {:?}", links_by_ttl);
 
     // we fetch the unresolved nodes at TTL 1
     let (unresolved_nodes, max_weighted_threshold) = miner.unresolved_nodes_at_ttl(1, false);
@@ -103,8 +103,9 @@ fn test_unresolved_nodes_at_ttl_basic() {
     assert_eq!(
         unresolved_nodes.len(),
         1,
-        "Unresolved nodes number. unresolved_nodes: {:?}",
-        unresolved_nodes
+        "Unresolved nodes number. unresolved_nodes: {:?}, max_weighted_threshold: {}",
+        unresolved_nodes,
+        max_weighted_threshold
     );
     assert_eq!(max_weighted_threshold, 6, "Max weighted threshold");
     assert_eq!(
@@ -180,6 +181,8 @@ fn test_unresolved_nodes_at_ttl_missing_link() {
     // we should have IP[1] as unresolved at TTL 1
     // since we do not know where the response from IP[3] went through
     // we have to hypothesize that IP[3] is a potential successor of IP[1]
+    // ^^^ this is not true, we would need to hypothesize that all nodes
+    // at ttl n+1 are connected to all nodes at ttl n
     let replies = vec![
         reply(1, IP[1], DEST[1]),
         reply(1, IP[1], DEST[2]),
@@ -208,16 +211,12 @@ fn test_unresolved_nodes_at_ttl_missing_link() {
 
     assert_eq!(
         unresolved_nodes.len(),
-        1,
+        // 1,
+        0,
         "Unresolved nodes number. unresolved_nodes: {:?}",
         unresolved_nodes
     );
-    assert_eq!(max_weighted_threshold, 11, "Max weighted threshold");
-    assert_eq!(
-        unresolved_nodes.into_iter().next().unwrap(),
-        IpAddr::V4(IP[1].parse().unwrap()),
-        "Unresolved nodes"
-    );
+    assert_eq!(max_weighted_threshold, 0, "Max weighted threshold");
 }
 
 fn probes_to_count(probes: Vec<Probe>) -> HashMap<TTL, usize> {

src/classic_traceroute.rs

@@ -5,14 +5,15 @@ use std::net::IpAddr;
 use anyhow::Result;
 
 use itertools::Itertools;
-use pantrace::formats::internal::Traceroute;
+use pantrace::formats::internal::{Traceroute, TracerouteHop};
 use pantrace::traits::TracerouteWriter;
 
 pub struct ClassicTracerouteWriter<W: Write> {
     output: W,
     min_ttl: u8,
     max_ttl: u8,
     dst_addr: IpAddr,
+    total_flows: usize,
 }
 
 impl<W: Write> ClassicTracerouteWriter<W> {
@@ -21,12 +22,14 @@ impl<W: Write> ClassicTracerouteWriter<W> {
         min_ttl: u8,
         max_ttl: u8,
         dst_addr: IpAddr,
+        total_flows: usize,
     ) -> ClassicTracerouteWriter<W> {
         ClassicTracerouteWriter {
             output,
             min_ttl,
             max_ttl,
             dst_addr,
+            total_flows,
         }
     }
 }
@@ -39,20 +42,37 @@ where
         let packet_size = traceroute.flows[0].hops[0].probes[0].size;
         write!(
             self.output,
-            "traceroute to {}({}), {} hops max, {} bytes packets",
-            traceroute.dst_addr, traceroute.dst_addr, self.max_ttl, packet_size
+            "traceroute to {}({}), {} hops max, {} bytes packets, flow {}/{}\n",
+            traceroute.dst_addr,
+            traceroute.dst_addr,
+            self.max_ttl,
+            packet_size,
+            1,
+            self.total_flows
         )
         .unwrap();
-        let flow = &traceroute.flows[0];
-        let hops_by_ttl = flow.hops.iter().group_by(|hop| hop.ttl);
-        let hops_by_ttl = hops_by_ttl
-            .into_iter()
-            .fold(HashMap::new(), |mut acc, (ttl, hops)| {
-                acc.insert(ttl, hops.collect_vec());
-                acc
-            });
 
-        write!(self.output, "\n").unwrap();
+        let all_hops = traceroute
+            .flows
+            .iter()
+            .flat_map(|flow| flow.hops.iter())
+            // .map(|flow| &flow.hops)
+            // .flatten()
+            .collect::<Vec<_>>();
+
+        let hops_by_ttl = all_hops
+            .into_iter()
+            .group_by(|hop| hop.ttl)
+            .into_iter()
+            .fold(
+                HashMap::<u8, Vec<&TracerouteHop>>::new(),
+                |mut acc, (ttl, group)| {
+                    acc.entry(ttl)
+                        .or_insert_with(Vec::new)
+                        .extend(group.into_iter());
+                    acc
+                },
+            );
 
         let mut found_dst = false;
 
@@ -62,29 +82,40 @@ where
             }
             write!(self.output, "{}", ttl).unwrap();
             if let Some(hops) = hops_by_ttl.get(&ttl) {
-                let hop = &hops[0];
-                let probes_by_host = hop
-                    .probes
-                    .iter()
-                    .filter(|probe| probe.reply.is_some())
-                    .group_by(|probe| probe.reply.as_ref().map(|r| r.addr));
-                for (ip, probes) in probes_by_host.into_iter() {
-                    let probes = probes.collect_vec();
-                    if let Some(ip) = ip {
-                        write!(self.output, "   {} ({})", ip, ip).unwrap();
-                        for probe in probes.iter() {
-                            let reply = probe.reply.as_ref().unwrap();
-                            write!(self.output, "  {:.3} ms", reply.rtt / 10.0).unwrap();
-                            found_dst = found_dst || (ip == self.dst_addr);
-                        }
-                        write!(self.output, "\n").unwrap();
-                    } else {
-                        write!(self.output, "   *\n").unwrap();
-                    }
+                let all_probes = hops
+                    .into_iter()
+                    .flat_map(|hop| hop.probes.iter())
+                    .collect::<Vec<_>>();
+                let all_replies_by_host = all_probes
+                    .into_iter()
+                    .filter_map(|probe| probe.reply.as_ref().map(|reply| (reply.addr, reply.rtt)))
+                    .fold(
+                        HashMap::<IpAddr, Vec<f64>>::new(),
+                        |mut acc, (addr, rtt)| {
+                            acc.entry(addr).or_insert_with(Vec::new).push(rtt);
+                            acc
+                        },
+                    );
+
+                for (ip, rtts) in all_replies_by_host.into_iter() {
+                    found_dst |= ip == self.dst_addr;
+
+                    write!(self.output, "   {} ({})", ip, ip).unwrap();
+                    let mean_rtt = rtts.iter().sum::<f64>() / rtts.len() as f64;
+                    write!(
+                        self.output,
+                        "  {:.3} ms ({} probes)",
+                        mean_rtt / 10.0,
+                        rtts.len()
+                    )
+                    .unwrap();
+                    write!(self.output, "\n").unwrap();
                 }
+                // write!(self.output, "\n").unwrap();
             } else {
                 write!(self.output, "   *\n").unwrap();
             }
+            write!(self.output, "\n").unwrap();
         }
 
         Ok(())

src/links/tests.rs

@@ -113,11 +113,6 @@ fn test_get_pairs_by_flow() {
         (
             flow_2,
             vec![
-                ReplyPair {
-                    ttl: 1,
-                    first_reply: None,
-                    second_reply: Some(&replies[1]),
-                },
                 ReplyPair {
                     ttl: 2,
                     first_reply: Some(&replies[1]),

src/main.rs

@@ -14,7 +14,7 @@ use pantrace::formats::internal::{Protocol, Traceroute};
 use pantrace::traits::TracerouteWriter;
 use voyage::algorithms::diamond_miner::DiamondMiner;
 use voyage::classic_traceroute::ClassicTracerouteWriter;
-use voyage::pantrace_builder::{replies_to_pantrace_flows, replies_to_single_pantrace_flow};
+use voyage::pantrace_builder::replies_to_pantrace_flows;
 
 use anyhow::Result;
 use voyage::probe::probe;
@@ -98,7 +98,7 @@ struct Args {
     estimate_successors: bool,
 
     /// Output format
-    #[arg(short, long, value_enum, default_value_t = OutputFormat::Atlas)]
+    #[arg(short, long, value_enum, default_value_t = OutputFormat::Traceroute)]
     output_format: OutputFormat,
 
     /// Receiver wait time in seconds
@@ -271,7 +271,8 @@ fn main() -> Result<()> {
         }
     }
 
-    let pantrace_flows = replies_to_pantrace_flows(&alg.time_exceeded_replies());
+    // let pantrace_flows = replies_to_pantrace_flows(&alg.time_exceeded_replies());
+    let pantrace_flows = replies_to_pantrace_flows(&alg.replies());
 
     let traceroute: Traceroute = Traceroute {
         measurement_name: "diamond_miner".to_string(),
@@ -286,18 +287,22 @@ fn main() -> Result<()> {
         flows: pantrace_flows,
     };
 
+    println!(
+        ">>> total probes in flows: {}",
+        traceroute
+            .flows
+            .iter()
+            .map(|f| f.hops.iter().map(|h| h.probes.len()).sum::<usize>())
+            .sum::<usize>()
+    );
+
     match args.output_format {
         OutputFormat::Traceroute => {
             debug!("--- Traceroute output ---");
             let stdout = std::io::stdout();
-            let classic_traceroute_flow = replies_to_single_pantrace_flow(&alg.replies());
-            // replies_to_single_pantrace_flow(&alg.time_exceeded_replies());
-            let traceroute = Traceroute {
-                flows: vec![classic_traceroute_flow],
-                ..traceroute
-            };
+            let total_flows = traceroute.flows.len();
             let mut traceroute_writer =
-                ClassicTracerouteWriter::new(stdout, min_ttl, max_ttl, dst_addr);
+                ClassicTracerouteWriter::new(stdout, min_ttl, max_ttl, dst_addr, total_flows);
             traceroute_writer.write_traceroute(&traceroute)?;
         }
         OutputFormat::Atlas => {
@@ -361,17 +366,6 @@ fn main() -> Result<()> {
                     }
                 }
             }
-            // debug!(
-            //     "Links: {}",
-            //     alg.links_by_ttl()
-            //         .values()
-            //         .flatten()
-            //         .filter(|link| link.near_ip.is_some() && link.far_ip.is_some())
-            //         .unique()
-            //         .map(|link| format!("{:?}", link))
-            //         .collect::<Vec<_>>()
-            //         .join(", ")
-            // );
         }
     }
 

src/pantrace_builder.rs

@@ -68,7 +68,3 @@ pub fn replies_to_pantrace_flows(replies: &[&Reply]) -> Vec<TracerouteFlow> {
         .map(|replies| generate_pantrace_traceroute_flow(&replies))
         .collect()
 }
-
-pub fn replies_to_single_pantrace_flow(replies: &[&Reply]) -> TracerouteFlow {
-    generate_pantrace_traceroute_flow(replies)
-}