manycastr/
main.rs

1//! # MAnycastR
2//!
3//! MAnycastR (Measure Anycast Routing) performs synchronized Internet measurements from a
4//! distributed set of anycast Points of Presence (PoPs): catchment mapping, anycast and unicast
5//! latency, anycast traceroute, and anycast censuses. IPv4 and IPv6 are both supported, over
6//! ICMP, UDP (DNS), and TCP.
7//!
8//! These pages document the internals. For installation, usage, measurement types, and output
9//! formats, see the
10//! [README](https://github.com/rhendriks/MAnycastR#readme), or run `manycastr cli start --help`.
11//!
12//! # The components
13//!
14//! A deployment consists of three components, each a subcommand of the `manycastr` binary:
15//!
16//! * [Orchestrator](orchestrator) - a central controller orchestrating measurements
17//! * [CLI](cli) - command-line interface scheduling measurements at the Orchestrator and collecting results
18//! * [Worker](worker) - deployed on anycast PoPs, performing measurements
19//!
20//! The CLI sends a measurement definition to the Orchestrator, which instructs the Workers to
21//! start the measurement. Workers send probes and receive replies, streaming results back to the
22//! Orchestrator, which aggregates them (creating follow-up tasks where the measurement type calls
23//! for it) and forwards them to the CLI, which writes the output file.
24//!
25//! Supporting modules: [net] (packet construction and parsing), [tls] (transport security for the
26//! inter-component gRPC connections), and [custom_module] (generated gRPC types).
27use clap::builder::{ArgPredicate, PossibleValuesParser};
28use clap::{ArgAction, ArgGroup, ArgMatches, Command, arg, value_parser};
29use log::{error, info};
30use pretty_env_logger::formatted_builder;
31use std::io::Write;
32use std::process::exit;
33
34mod cli;
35mod custom_module;
36mod net;
37mod orchestrator;
38mod tls;
39mod worker;
40
41pub const ALL_WORKERS: u32 = 0;
42pub const ALL_ORIGINS: u32 = 0;
43pub const SINGLE_ORIGIN: u32 = 0; // Used for single Origin measurements
44
45/// Get 6-bits from the measurement ID for the DNS identifier for filtering.
46#[inline]
47pub fn dns_identifier(m_id: u32) -> u8 {
48    (m_id & 0x3F) as u8
49}
50
51/// Used for `--responsive` and `--sessions` enabled when using `-m feed`.
52#[inline]
53pub fn probe_id(m_id: u32, session_id: u32) -> u32 {
54    (m_id << 16) | (session_id & 0xFFFF)
55}
56
57/// Get the 16-bit measurement ID from a probe ID.
58#[inline]
59pub fn m_id_of(probe_id: u32) -> u32 {
60    probe_id >> 16
61}
62
63/// Get the 16-bit session ID from a probe ID.
64#[inline]
65pub fn session_id_of(probe_id: u32) -> u32 {
66    probe_id & 0xFFFF
67}
68
69/// Parse command line input and start MAnycastR orchestrator, worker, or CLI
70///
71/// Sets up logging, parses the command-line arguments, runs the appropriate initialization function.
72fn main() {
73    // Initialize logging with timestamps
74    formatted_builder()
75        .parse_env(pretty_env_logger::env_logger::Env::default().default_filter_or("info"))
76        .format(|buf, record| {
77            writeln!(
78                buf,
79                "{} [{}] > {}",
80                chrono::Local::now().format("%Y-%m-%d %H:%M:%S"),
81                record.level(),
82                record.args()
83            )
84        })
85        .init();
86    // Parse the command-line arguments
87    let matches = parse_cmd();
88
89    if let Some(worker_matches) = matches.subcommand_matches("worker") {
90        info!("[Main] Executing Worker version {}", env!("GIT_HASH"));
91
92        let rt = tokio::runtime::Builder::new_current_thread()
93            .enable_all()
94            .build()
95            .unwrap();
96
97        rt.block_on(async {
98            if let Err(e) = worker::Worker::new(worker_matches).await {
99                error!(
100                    "[Worker] Unable to connect to the Orchestrator (check -a, --tls, and that the Orchestrator is running): {e}"
101                );
102                exit(1);
103            }
104        });
105    } else if let Some(cli_matches) = matches.subcommand_matches("cli") {
106        info!("[Main] Executing CLI version {}", env!("GIT_HASH"));
107
108        if let Err(e) = cli::execute(cli_matches) {
109            error!("[CLI] {e}");
110            exit(1);
111        }
112    } else if let Some(server_matches) = matches.subcommand_matches("orchestrator") {
113        info!("[Main] Executing Orchestrator version {}", env!("GIT_HASH"));
114
115        let rt = tokio::runtime::Builder::new_current_thread()
116            .enable_all()
117            .build()
118            .unwrap();
119
120        rt.block_on(async {
121            if let Err(e) = orchestrator::start(server_matches).await {
122                error!("[Orchestrator] {e}");
123                exit(1);
124            }
125        });
126    } else {
127        error!("[Main] No valid subcommand provided, use --help for more information");
128    }
129}
130
131/// Parse command line arguments using clap
132fn parse_cmd() -> ArgMatches {
133    Command::new("manycastr")
134        .version(env!("GIT_HASH"))
135        .author("Remi Hendriks <remi.hendriks@utwente.nl>")
136        .about("Performs synchronized Internet measurement from a distributed set of anycast Points of Presence (PoPs)")
137        .subcommand_required(true)
138        .subcommand(
139            Command::new("orchestrator").about("Launches the MAnycastR Orchestrator")
140                .arg(arg!(-p --port <PORT> "Port to listen on").value_parser(value_parser!(u16)).default_value("50001"))
141                .arg(arg!(--cli_port <PORT> "Port for CLI (default: CLI shares the --port listener)")
142                    .value_parser(value_parser!(u16)))
143                .arg(arg!(--tls <CERT> "Enable TLS with the certificate at the given path (e.g., ./tls/orchestrator.crt)"))
144                .arg(arg!(--tls_key <KEY> "Path to the TLS private key (default: the --tls path with a .key extension)")
145                    .requires("tls"))
146                .arg(arg!(-c --config <FILE> "Worker hostname to IDs configuration").value_parser(value_parser!(String)))
147                .arg(arg!(--max_rate <RATE> "Maximum probing rate allowed for measurements (probes per second, per Worker; optional)")
148                    .value_parser(value_parser!(u32)))
149                .arg(arg!(--origins <FILE> "Origin allow-list restricting the origins CLIs may use ('src_addr, protocol[, protocol...]' per line; 'all' allows all protocols)")
150                    .value_parser(value_parser!(String)))
151        )
152        .subcommand(
153            Command::new("worker").about("Launches the MAnycastR Worker")
154                .arg(arg!(-a --orchestrator <ADDR> "address:port of the Orchestrator (e.g., 10.0.0.0:50001, [::1]:50001, or orchestrator.example.net:50001)").required(true))
155                .arg(arg!(-n --hostname <NAME> "hostname for this Worker (default: $HOSTNAME)"))
156                .arg(arg!(--tls <CERT> "Enable TLS, authenticating the Orchestrator against the certificate at the given path (its own certificate, or the CA that issued it)"))
157                .arg(arg!(--tls_system "Enable TLS, authenticating the Orchestrator against the host's system trust store"))
158                .group(ArgGroup::new("tls_mode").args(["tls", "tls_system"]))
159                .arg(arg!(--tls_domain <NAME> "Name to authenticate the Orchestrator as (default: the host in -a)")
160                    .requires("tls_mode"))
161        )
162        .subcommand(
163            Command::new("cli").about("MAnycastR CLI")
164                .arg(arg!(-a --orchestrator <ADDR> "address:port of the Orchestrator (e.g., 10.0.0.0:50001, [::1]:50001, or orchestrator.example.net:50001)").required(true))
165                .arg(arg!(--tls <CERT> "Enable TLS, authenticating the Orchestrator against the certificate at the given path (its own certificate, or the CA that issued it)"))
166                .arg(arg!(--tls_system "Enable TLS, authenticating the Orchestrator against the host's system trust store"))
167                .group(ArgGroup::new("tls_mode").args(["tls", "tls_system"]))
168                .arg(arg!(--tls_domain <NAME> "Name to authenticate the Orchestrator as (default: the host in -a)")
169                    .requires("tls_mode"))
170                .subcommand(Command::new("worker-list").about("retrieves a list of currently connected Workers from the Orchestrator"))
171                .subcommand(Command::new("start").about("performs a hitlist-based measurement")
172                    .arg(arg!(--hitlist <PATH> "Path to the hitlist file (can be .gz or .bz2 compressed; ISI fsdb hitlists are detected automatically)")
173                        .value_parser(value_parser!(String))
174                        .conflicts_with("target"))
175                    .arg(arg!(-t --target <TARGETS> "Comma-separated target address(es), e.g. '1.1.1.1' or '1.1.1.1,8.8.8.8' (alternative to --hitlist)")
176                        .value_parser(value_parser!(String)))
177                    .arg(arg!(-p --p_type <TYPE> "Protocols to use")
178                        .value_parser(PossibleValuesParser::new(["icmp", "dns", "tcp", "chaos"]))
179                        .value_delimiter(',')// Allow for multiple protocols
180                        .action(ArgAction::Append)
181                        .default_value("icmp")
182                        .ignore_case(true))
183                    .arg(arg!(-m --m_type <MODE> "Measurement type to perform")
184                        .value_parser(PossibleValuesParser::new(["laces", "catchment", "latency", "anycast-traceroute", "tracemap", "feed", "feed-trace"]))
185                        .default_value("laces")
186                        .ignore_case(true))
187                    .arg(arg!(-a --address <ADDR> "Anycast source address, or 'unicastv4'/'unicastv6' to probe from each Worker's local unicast address")
188                        .conflicts_with("configuration")
189                        .required_unless_present("configuration"))
190                    .arg(arg!(-f --configuration <CONF> "Path to config file").conflicts_with_all(["address", "sport", "dport", "p_type"]))
191                    .arg(arg!(-r --rate <RATE> "Probing rate at each Worker (packets per second)")
192                        .value_parser(value_parser!(u32))
193                        .default_value_ifs([
194                            ("m_type", ArgPredicate::Equals("anycast-traceroute".into()), Some("10")),
195                            ("m_type", ArgPredicate::Equals("tracemap".into()), Some("10")),
196                        ])
197                        .default_value("1000"))
198                    .arg(arg!(selective: -x --selective <IDS> "List of Worker IDs/hostnames that send probes [worker_id1,worker_id2,...]"))
199                    .arg(arg!(-o --out <PATH> "Optional path/filename to write output").default_value("./"))
200                    .arg(arg!(--parquet "Write as .parquet (instead of .csv.gz)").action(ArgAction::SetTrue))
201                    .arg(arg!(--stream "Stream to stdout").action(ArgAction::SetTrue))
202                    .arg(arg!(--shuffle "Shuffle hitlist").action(ArgAction::SetTrue))
203                    .arg(arg!(--responsive "Check responsiveness of targets for multi-target hitlists and multi-probe measurements.").action(ArgAction::SetTrue))
204                    .arg(arg!(--sessions "Enable feed sessions (-m feed only): NDJSON targets may carry a 'session' field (1-65535), reported per reply in the output's 'session' column").action(ArgAction::SetTrue))
205                    .arg(arg!(--trace_max_failures <N> "Maximum number of consecutive failures (tracemap: confirmation window past a silent midpoint, default 3)")
206                        .value_parser(value_parser!(u32))
207                        .default_value_if("m_type", ArgPredicate::Equals("tracemap".into()), Some("3"))
208                        .default_value("5"))
209                    .arg(arg!(--trace_timeout <N> "Timeout for hops (in seconds)").value_parser(value_parser!(u32)).default_value("3"))
210                    .arg(arg!(--trace_max_hop <N> "Maximum TTL value (covers >99% of Internet path lengths)")
211                        .value_parser(value_parser!(u32))
212                        .default_value("25"))
213                    .arg(arg!(--trace_initial_hop <N> "Starting TTL value (skips hops within the PoP's own network)")
214                        .value_parser(value_parser!(u32))
215                        .default_value("4"))
216                    .arg(arg!(--trace_star <BOOL> "Emit a '*' hop to the output for unresponsive (timed-out) hops").value_parser(value_parser!(bool)).default_value("true"))
217                    .arg(arg!(-w --worker_interval <N> "Interval between Workers for probes to the same target").value_parser(value_parser!(u32)).default_value("1"))
218                    .arg(arg!(-i --probe_interval <N> "Interval between probes from the same Worker to the same target").value_parser(value_parser!(u32)).default_value("1"))
219                    .arg(arg!(-c --nprobes <N> "Number of probes to send for each origin,target pair [NOTE: violates probing rate]").value_parser(value_parser!(u32)).default_value("1"))
220                    .arg(arg!(-s --sport <PORT> "Source port to use (DNS,UDP)").value_parser(value_parser!(u16)).default_value("62321"))
221                    .arg(arg!(-d --dport <PORT> "Destination port to use (default DNS/CHAOS: 53, TCP: 63853)")
222                        .value_parser(value_parser!(u16))
223                        .default_value_ifs([
224                            ("p_type", ArgPredicate::Equals("dns".into()), Some("53")),
225                            ("p_type", ArgPredicate::Equals("chaos".into()), Some("53")),
226                        ])
227                        .default_value("63853")
228                    )
229                    .arg(arg!(-q --query <QUERY> "Specify DNS record to request (TXT (CHAOS) default: hostname.bind, A default: example.org)")
230                        .default_value_ifs([
231                            ("p_type", ArgPredicate::Equals("chaos".into()), Some("hostname.bind")),
232                            ("p_type", ArgPredicate::Equals("dns".into()), Some("example.org")),
233                        ]))
234                    .arg(arg!(-u --url <URL> "URL encoded in probe payload (e.g., opt-out URL)"))
235                )
236            )
237        .get_matches()
238}
manycastr/main.rs

manycastr/
main.rs
