Add repair server dynamometer

downstairs/src/dynamometer.rs

@@ -1,6 +1,8 @@
 // Copyright 2023 Oxide Computer Company
 use super::*;
 
+use futures::TryStreamExt;
+
 pub enum DynoFlushConfig {
     FlushPerIops(usize),
     FlushPerBlocks(usize),
@@ -187,3 +189,106 @@ pub fn dynamometer(
 
     Ok(())
 }
+
+pub async fn repair_dynamometer(clone_source: SocketAddr) -> Result<()> {
+    let mut bytes_received = 0;
+    let mut measurement_time = Instant::now();
+    let mut bytes_per_second: Vec<f32> = vec![];
+
+    let url = format!("http://{:?}", clone_source);
+    println!("using {url}");
+    let repair = repair_client::Client::new(&url);
+
+    let source_def = match repair.get_region_info().await {
+        Ok(def) => def.into_inner(),
+        Err(e) => {
+            bail!("Failed to get source region definition: {e}");
+        }
+    };
+
+    println!("The source RegionDefinition is: {:?}", source_def);
+
+    let source_ro_mode = match repair.get_region_mode().await {
+        Ok(ro) => ro.into_inner(),
+        Err(e) => {
+            bail!("Failed to get source mode: {e}");
+        }
+    };
+
+    println!("The source mode is: {:?}", source_ro_mode);
+    if !source_ro_mode {
+        bail!("Source downstairs is not read only");
+    }
+
+    for eid in (0..source_def.extent_count()).map(ExtentId) {
+        println!("Repair extent {eid}");
+
+        let mut repair_files = match repair.get_files_for_extent(eid.0).await {
+            Ok(f) => f.into_inner(),
+            Err(e) => {
+                bail!("Failed to get repair files: {:?}", e,);
+            }
+        };
+
+        repair_files.sort();
+        println!("eid:{} Found repair files: {:?}", eid, repair_files);
+
+        let mut stream = match repair
+            .get_extent_file(eid.0, repair_client::types::FileType::Data)
+            .await
+        {
+            Ok(rs) => rs,
+            Err(e) => {
+                bail!("Failed to get extent {} db file: {:?}", eid, e,);
+            }
+        };
+
+        loop {
+            match stream.try_next().await {
+                Ok(Some(bytes)) => {
+                    bytes_received += bytes.len();
+
+                    let elapsed = measurement_time.elapsed();
+
+                    if elapsed > Duration::from_secs(1) {
+                        let fractional_seconds: f32 = elapsed.as_secs() as f32
+                            + (elapsed.subsec_nanos() as f32 / 1e9);
+
+                        println!(
+                            "bytes per second: {}",
+                            bytes_received as f32 / fractional_seconds
+                        );
+                        bytes_per_second
+                            .push(bytes_received as f32 / fractional_seconds);
+                        bytes_received = 0;
+                        measurement_time = Instant::now();
+                    }
+                }
+
+                Ok(None) => break,
+
+                Err(e) => {
+                    bail!("repair stream error: {:?}", e);
+                }
+            }
+        }
+    }
+
+    println!("B/s: {:?}", bytes_per_second);
+    println!(
+        "B/S mean {} stddev {}",
+        statistical::mean(&bytes_per_second),
+        statistical::standard_deviation(&bytes_per_second, None),
+    );
+
+    bytes_per_second
+        .sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));
+
+    println!(
+        "B/s min {} max {}",
+        bytes_per_second.first().unwrap(),
+        bytes_per_second.last().unwrap(),
+    );
+
+    Ok(())
+}

downstairs/src/main.rs

@@ -1,5 +1,4 @@
 // Copyright 2023 Oxide Computer Company
-
 use std::net::{IpAddr, SocketAddr};
 use std::path::PathBuf;
 use std::time::Duration;
@@ -221,7 +220,7 @@ enum Args {
         bind_addr: SocketAddr,
     },
     Version,
-    /// Measure an isolated downstairs
+    /// Measure an isolated downstairs' disk usage
     Dynamometer {
         #[clap(long, default_value_t = 512)]
         block_size: u64,
@@ -258,6 +257,11 @@ enum Args {
         #[clap(long, value_parser = parse_duration, conflicts_with_all = ["flush_per_iops", "flush_per_blocks"])]
         flush_per_ms: Option<Duration>,
     },
+    /// Measure a downstairs' repair server
+    RepairDynamometer {
+        #[clap(long, value_name = "SOURCE", action)]
+        clone_source: SocketAddr,
+    },
 }
 
 fn parse_duration(arg: &str) -> Result<Duration, std::num::ParseIntError> {
@@ -499,5 +503,8 @@ async fn main() -> Result<()> {
 
             dynamometer(region, num_writes, samples, flush_config)
         }
+        Args::RepairDynamometer { clone_source } => {
+            repair_dynamometer(clone_source).await
+        }
     }
 }