Add repair server dynamometer (#1618)

Here's an example usage on dogfood, measuring a region snapshot's
downstairs on another sled:

    # /tmp/jwm-downstairs repair-dynamometer \
        --clone-source '[fd00:1122:3344:107::c]:23009'

    using http://[fd00:1122:3344:107::c]:23009
    The source RegionDefinition is: RegionDefinition { block_size: 512,
extent_size: Block { value: 131072, shift: 9 }, extent_count: 16, uuid:
    31fb7fa7-0432-409e-8bec-571b6613b188, encrypted: true,
    database_read_version: 1, database_write_version: 1 }
    The source mode is: true
    Repair extent 0
eid:0 Found repair files: ["000", "000.db", "000.db-shm", "000.db-wal"]
    Repair extent 1
eid:1 Found repair files: ["001", "001.db", "001.db-shm", "001.db-wal"]
    Repair extent 2
eid:2 Found repair files: ["002", "002.db", "002.db-shm", "002.db-wal"]
    Repair extent 3
eid:3 Found repair files: ["003", "003.db", "003.db-shm", "003.db-wal"]
    bytes per second: 222755540
    Repair extent 4
eid:4 Found repair files: ["004", "004.db", "004.db-shm", "004.db-wal"]
    Repair extent 5
eid:5 Found repair files: ["005", "005.db", "005.db-shm", "005.db-wal"]
    Repair extent 6
eid:6 Found repair files: ["006", "006.db", "006.db-shm", "006.db-wal"]
    bytes per second: 224481150
    Repair extent 7
eid:7 Found repair files: ["007", "007.db", "007.db-shm", "007.db-wal"]
    Repair extent 8
eid:8 Found repair files: ["008", "008.db", "008.db-shm", "008.db-wal"]
    Repair extent 9
eid:9 Found repair files: ["009", "009.db", "009.db-shm", "009.db-wal"]
    Repair extent 10
eid:10 Found repair files: ["00A", "00A.db", "00A.db-shm", "00A.db-wal"]
    bytes per second: 227660750
    Repair extent 11
eid:11 Found repair files: ["00B", "00B.db", "00B.db-shm", "00B.db-wal"]
    Repair extent 12
eid:12 Found repair files: ["00C", "00C.db", "00C.db-shm", "00C.db-wal"]
    Repair extent 13
eid:13 Found repair files: ["00D", "00D.db", "00D.db-shm", "00D.db-wal"]
    bytes per second: 228354910
    Repair extent 14
eid:14 Found repair files: ["00E", "00E.db", "00E.db-shm", "00E.db-wal"]
    Repair extent 15
eid:15 Found repair files: ["00F", "00F.db", "00F.db-shm", "00F.db-wal"]
    B/s: [222755540.0, 224481150.0, 227660750.0, 228354910.0]
    B/S mean 225813090 stddev 2645588
    B/s min 222755540 max 228354910

Bytes per second looks like about 215.4 MiB/s here.

Author: jmpesp

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
+        }
     }
 }