feat: add a mach saturate command to saturate a network and determine its capacity

Cargo.toml

@@ -1,11 +1,11 @@
 [workspace]
 members = ["crates/*", "cli"]
-resolver = "2"
+resolver = "3"
 
 [workspace.package]
 version = "3.3.0"
-repository = "https://github.com/cloudflare/foundations"
-edition = "2021"
+repository = "https://github.com/cloudflare/networkquality-rs"
+edition = "2024"
 authors = [
     "Fisher Darling <fisher@cloudflare.com>",
     "Lina Baquero <lina@cloudflare.com>",
@@ -28,21 +28,25 @@ nq-tokio-network = { path = "./crates/nq-tokio-network" }
 
 anyhow = "1.0"
 async-trait = { version = "0.1" }
-boring = "4.11.0"
+boring = "5.0"
+bytes = "1.6.0"
 clap = "4.3"
 clap-verbosity-flag = "2.1"
 http = "1.0"
 http-body-util = "0.1.2"
+humansize = "2.1.3"
 hyper = "1.0"
 hyper-util = "0.1"
 pin-project-lite = "0.2"
 rand = "0.8"
+rustls = { version = "0.23.10", default-features = false, features = ["std", "ring"] }
+rustls-native-certs = "0.7.0"
 serde = "1.0"
 serde_json = { version = "1.0", features = ["preserve_order"] }
 tracing = "0.1"
 tracing-subscriber = "0.3"
 tokio = "1.43"
 tokio-util = "0.7"
-tokio-boring = "4.11.0"
+tokio-boring = "5.0"
 url = "2.4"
 webrtc = "0.12.0"

cli/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "mach-cli"
-version = "0.1.0"
+version = "0.2.0"
 authors = ["Fisher Darling <fisher@cloudflare.com>"]
 edition = "2021"
 
@@ -27,4 +27,4 @@ serde_json = { workspace = true }
 tokio = { workspace = true, features = ["rt-multi-thread", "time", "net", "macros",] }
 tokio-util = { workspace = true }
 tracing = { workspace = true }
-tracing-subscriber = { workspace = true, features = ["env-filter", "fmt"] }
+tracing-subscriber = { workspace = true, features = ["env-filter", "fmt"] }

cli/src/aim_report.rs

@@ -132,9 +132,9 @@ impl CloudflareAimResults {
 #[serde(rename_all = "camelCase")]
 pub struct BpsMeasurement {
     /// The total number of bytes.
-    bytes: usize,
+    pub(crate) bytes: usize,
     /// The bits per second of the transfer.
-    bps: usize,
+    pub(crate) bps: usize,
 }
 
 impl BpsMeasurement {
@@ -154,7 +154,7 @@ impl BpsMeasurement {
     }
 
     /// Use the test duration and network capacity to create a synthetic bps result.
-    fn from_rpm_result(rpm_result: &ResponsivenessResult) -> BpsMeasurement {
+    pub(crate) fn from_rpm_result(rpm_result: &ResponsivenessResult) -> BpsMeasurement {
         let throughput = rpm_result.throughput().unwrap_or(0) as f64;
 
         let bytes = throughput * rpm_result.duration.as_secs_f64();
@@ -190,7 +190,7 @@ impl Default for PacketLossMeasurement {
 /// https://developers.cloudflare.com/speed/aim/
 #[derive(Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
-#[allow(missing_docs)]
+#[allow(missing_docs, dead_code)]
 pub enum AimScore {
     Streaming {
         points: usize,

cli/src/args/mod.rs

@@ -3,13 +3,17 @@
 
 pub(crate) mod packet_loss;
 pub(crate) mod rpm;
+pub(crate) mod saturate;
 pub(crate) mod up_down;
 
 use clap::{Parser, Subcommand, ValueEnum};
+use nq_core::ConnectionType;
 use packet_loss::PacketLossArgs;
 
 use crate::args::rpm::RpmArgs;
-use crate::args::up_down::DownloadArgs;
+use crate::args::saturate::SaturateArgs;
+// use crate::args::saturate::SaturateArgs;
+use crate::args::up_down::{DownloadArgs, UploadArgs};
 
 /// mach runs multiple different network performance tests. The main focus of
 /// mach and this tool is to implement the IETF draft: "Responsiveness under
@@ -39,7 +43,7 @@ pub enum Command {
     Download(DownloadArgs),
     /// Upload data (POST) to an endpoint,  reporting latency measurements and total
     /// throughput.
-    // Upload(UploadArgs),
+    Upload(UploadArgs),
     /// Determine the Round-Trip-Time (RTT), or latency, of a link using the
     /// time it takes to establish a TCP connection.
     ///
@@ -57,6 +61,10 @@ pub enum Command {
     },
     /// Send UDP packets to a TURN server, reporting lost packets.
     PacketLoss(PacketLossArgs),
+    /// Saturate the network in some direction and report maximum goodput. The
+    /// direction, `up`, `down`, `both` must be specified. By default, the
+    /// command runs for 20s.
+    Saturate(SaturateArgs),
 }
 
 // todo(fisher): figure out proxy chaining. Preparsing args or using the -- sentinal?
@@ -77,7 +85,19 @@ pub enum Command {
 /// Describes which underlying transport a connection uses.
 #[derive(Debug, Clone, ValueEnum)]
 pub enum ConnType {
+    H1ClearText,
     H1,
     H2,
     H3,
 }
+
+impl From<ConnType> for ConnectionType {
+    fn from(conn_type: ConnType) -> Self {
+        match conn_type {
+            ConnType::H1ClearText => ConnectionType::H1 { use_tls: false },
+            ConnType::H1 => ConnectionType::H1 { use_tls: true },
+            ConnType::H2 => ConnectionType::H2,
+            ConnType::H3 => ConnectionType::H3,
+        }
+    }
+}

cli/src/args/saturate.rs

@@ -0,0 +1,86 @@
+// Copyright (c) 2023-2024 Cloudflare, Inc.
+// Licensed under the BSD-3-Clause license found in the LICENSE file or at https://opensource.org/licenses/BSD-3-Clause
+
+//! Arguments for running a network saturation test.
+
+use crate::args::ConnType;
+
+/// Download data (GET) from an endpoint, reporting latency measurements and
+/// total throughput.
+#[derive(Debug, clap::Args)]
+pub struct SaturateArgs {
+    /// The connection type to use.
+    #[clap(short = 't', long, default_value = "h1-clear-text")]
+    pub(crate) conn_type: ConnType,
+    /// Which direction to saturate: `up`, `down` or `both`.
+    #[clap(subcommand)]
+    pub(crate) direction: Direction,
+    /// The duration in seconds to saturate the network for. Defaults
+    /// to 20s.
+    #[clap(long, default_value = "20")]
+    pub(crate) duration: u64,
+}
+
+#[derive(Debug, Clone, clap::Subcommand)]
+pub enum Direction {
+    /// Saturate the download (ingress) side of the network.
+    Down {
+        /// The URL to upload data to.
+        #[clap(
+            short,
+            long,
+            default_value = "http://h3.speed.cloudflare.com/__down?bytes=10000000000"
+        )]
+        download_url: String,
+    },
+    /// Saturate the upload (egress) side of the network.
+    Up {
+        /// The URL to upload data to.
+        #[clap(short, long, default_value = "http://h3.speed.cloudflare.com/__up")]
+        upload_url: String,
+    },
+    /// Saturate both the download (ingress) and upload (egress) side of the network.
+    Both {
+        /// The URL to download data from.
+        #[clap(
+            short,
+            long,
+            default_value = "http://h3.speed.cloudflare.com/__down?bytes=10000000000"
+        )]
+        download_url: String,
+        /// The URL to upload data to.
+        #[clap(short, long, default_value = "http://h3.speed.cloudflare.com/__up")]
+        upload_url: String,
+    },
+}
+
+impl Default for Direction {
+    fn default() -> Self {
+        Self::Down {
+            download_url: "http://h3.speed.cloudflare.com/__down?bytes=10000000000".into(),
+        }
+    }
+}
+
+pub(crate) struct Urls {
+    pub(crate) upload: String,
+    pub(crate) download: String,
+}
+
+impl Direction {
+    pub(crate) fn urls(&self) -> Urls {
+        match self {
+            Direction::Up { upload_url: url } | Direction::Down { download_url: url } => Urls {
+                upload: url.clone(),
+                download: url.clone(),
+            },
+            Direction::Both {
+                download_url,
+                upload_url,
+            } => Urls {
+                download: download_url.clone(),
+                upload: upload_url.clone(),
+            },
+        }
+    }
+}

cli/src/args/up_down.rs

@@ -34,7 +34,7 @@ pub struct UploadArgs {
     /// can be set.
     #[clap(short, long)]
     pub(crate) bytes: Option<usize>,
-    /// Upload the contents of a file. Only one of `bytes` or `file` can be set.
+    // /// Upload the contents of a file. Only one of `bytes` or `file` can be set.
     // #[clap(short, long)]
     // pub(crate) file: Option<PathBuf>,
     /// Headers to add to the request.

cli/src/latency.rs

@@ -45,10 +45,8 @@ pub async fn run(url: String, runs: usize) -> anyhow::Result<()> {
 pub async fn run_test(config: &LatencyConfig) -> anyhow::Result<LatencyResult> {
     let shutdown = CancellationToken::new();
     let time = Arc::new(TokioTime::new()) as Arc<dyn Time>;
-    let network = Arc::new(TokioNetwork::new(
-        Arc::clone(&time),
-        shutdown.clone(),
-    )) as Arc<dyn Network>;
+    let network =
+        Arc::new(TokioNetwork::new(Arc::clone(&time), shutdown.clone())) as Arc<dyn Network>;
 
     let rtt = Latency::new(config.clone());
     let results = rtt.run_test(network, time, shutdown).await?;

cli/src/main.rs

@@ -7,6 +7,7 @@ mod latency;
 mod packet_loss;
 mod report;
 mod rpm;
+mod saturate;
 mod up_down;
 mod util;
 
@@ -30,9 +31,10 @@ async fn main() -> anyhow::Result<()> {
     match command {
         Command::Rpm(config) => rpm::run(config).await?,
         Command::Download(config) => up_down::download(config).await?,
-        // Command::Upload(config) => up_down::upload(config).await?,
+        Command::Upload(config) => up_down::upload(config).await?,
         Command::Rtt { url, runs } => latency::run(url, runs).await?,
         Command::PacketLoss(config) => packet_loss::run(config).await?,
+        Command::Saturate(config) => saturate::run(config).await?,
     }
 
     Ok(())

cli/src/packet_loss.rs

@@ -59,7 +59,7 @@ async fn fetch_turn_server_creds(
     headers.append(hyper::header::HOST, HeaderValue::from_str(host)?);
 
     let response = Client::default()
-        .new_connection(ConnectionType::H1)
+        .new_connection(ConnectionType::h1())
         .method("GET")
         .headers(headers)
         .send(
@@ -80,4 +80,4 @@ async fn fetch_turn_server_creds(
     let creds = serde_json::from_slice(&response.into_body().collect().await?.to_bytes())
         .context("parsing json creds from turn server url")?;
     Ok(creds)
-}
+}

cli/src/rpm.rs

@@ -35,9 +35,12 @@ pub async fn run(cli_config: RpmArgs) -> anyhow::Result<()> {
         }
         None => {
             let urls = RpmUrls {
+                small_download_url: cli_config.small_download_url.clone(),
                 small_https_download_url: cli_config.small_download_url,
+                large_download_url: cli_config.large_download_url.clone(),
                 large_https_download_url: cli_config.large_download_url,
-                https_upload_url: cli_config.upload_url,
+                https_upload_url: cli_config.upload_url.clone(),
+                upload_url: cli_config.upload_url,
             };
             info!("using default configuration urls: {urls:?}");
 
@@ -74,6 +77,8 @@ pub async fn run(cli_config: RpmArgs) -> anyhow::Result<()> {
         trimmed_mean_percent: cli_config.trimmed_mean_percent,
         std_tolerance: cli_config.std_tolerance,
         max_loaded_connections: cli_config.max_loaded_connections,
+        conn_type: ConnectionType::H2,
+        determine_load_only: false,
     };
 
     info!("running download test");
@@ -121,10 +126,8 @@ async fn run_test(
 ) -> anyhow::Result<ResponsivenessResult> {
     let shutdown = CancellationToken::new();
     let time = Arc::new(TokioTime::new()) as Arc<dyn Time>;
-    let network = Arc::new(TokioNetwork::new(
-        Arc::clone(&time),
-        shutdown.clone().into(),
-    )) as Arc<dyn Network>;
+    let network =
+        Arc::new(TokioNetwork::new(Arc::clone(&time), shutdown.clone())) as Arc<dyn Network>;
 
     let rpm = Responsiveness::new(config.clone(), download)?;
     let result = rpm.run_test(network, time, shutdown.clone()).await?;
@@ -145,11 +148,11 @@ pub struct RpmServerConfig {
 
 #[derive(Debug, Serialize, Deserialize)]
 pub struct RpmUrls {
-    #[serde(alias = "small_download_url")]
+    small_download_url: String,
     small_https_download_url: String,
-    #[serde(alias = "large_download_url")]
+    large_download_url: String,
     large_https_download_url: String,
-    #[serde(alias = "upload_url")]
+    upload_url: String,
     https_upload_url: String,
 }