make separate binary for load tests

Author: ghubertpalo

Cargo.lock

@@ -973,12 +973,12 @@ impl DependenciesBuilder {
                 .get_current_epoch()
                 .await
                 .map_err(|e| DependenciesBuilderError::Initialization {
-                    message: "cannot create aggregator runner".to_string(),
+                    message: "cannot create aggregator runner: failed to retrieve current epoch."
+                        .to_string(),
                     error: Some(e.into()),
                 })?
                 .ok_or(DependenciesBuilderError::Initialization {
-                    message: "cannot build aggregator runner: impossible to retrieve current epoch"
-                        .to_string(),
+                    message: "cannot build aggregator runner: no epoch returned.".to_string(),
                     error: None,
                 })?;
             let (work_epoch, epoch_to_sign, next_epoch_to_sign) = match current_epoch {

mithril-aggregator/src/dependency_injection/builder.rs

@@ -8,11 +8,18 @@ homepage = { workspace = true }
 license = { workspace = true }
 repository = { workspace = true }
 
+[[bin]]
+name = "load-aggregator"
+test = false
+bench = false
+
 [dependencies]
+anyhow = "1.0.71"
 async-trait = "0.1.52"
 clap = { version = "4.0.18", features = ["derive"] }
 glob = "0.3"
 hex = "0.4.3"
+indicatif = { version = "0.17.5", features = ["tokio"] }
 mithril-common = { path = "../../mithril-common" }
 reqwest = { version = "0.11", features = ["json"] }
 serde = { version = "1.0", features = ["derive"] }

mithril-test-lab/mithril-end-to-end/Cargo.toml

@@ -0,0 +1,253 @@
+use std::{path::PathBuf, sync::Arc, time::Duration};
+
+use anyhow::Context;
+use clap::Parser;
+
+use indicatif::{ProgressBar, ProgressDrawTarget};
+use mithril_common::{
+    digesters::DummyImmutablesDbBuilder,
+    entities::{Epoch, PartyId, ProtocolParameters},
+    messages::RegisterSignerMessage,
+    test_utils::{MithrilFixture, MithrilFixtureBuilder},
+    StdResult,
+};
+
+use mithril_end_to_end::{Aggregator, BftNode};
+use reqwest::StatusCode;
+use slog::Level;
+use thiserror::Error;
+use tokio::{select, task::JoinSet, time::sleep};
+
+#[derive(Debug, Error)]
+pub enum LoadError {
+    #[error("Registering signer party_id={party_id}, expected HTTP code {expected_http_code} got {got_http_code} with the message: {error_message}.")]
+    SignerRegistrationError {
+        party_id: PartyId,
+        expected_http_code: u32,
+        got_http_code: u32,
+        error_message: String,
+    },
+}
+
+fn init_logger(opts: &MainOpts) -> slog_scope::GlobalLoggerGuard {
+    use slog::Drain;
+
+    let decorator = slog_term::TermDecorator::new().build();
+    let drain = slog_term::FullFormat::new(decorator).build().fuse();
+    let drain = slog_async::Async::new(drain).build().fuse();
+    let drain = slog::LevelFilter::new(drain, opts.log_level()).fuse();
+
+    slog_scope::set_global_logger(slog::Logger::root(Arc::new(drain), slog::o!()))
+}
+
+pub fn generate_signer_data(number_of_signers: usize) -> MithrilFixture {
+    MithrilFixtureBuilder::default()
+        .with_signers(number_of_signers)
+        .build()
+}
+
+/// Generate signer registration
+pub fn generate_register_message(signers_fixture: &MithrilFixture) -> Vec<RegisterSignerMessage> {
+    let epoch = Epoch(2);
+    signers_fixture
+        .signers()
+        .into_iter()
+        .map(|signer| RegisterSignerMessage {
+            epoch: Some(epoch),
+            party_id: signer.party_id,
+            verification_key: signer.verification_key,
+            verification_key_signature: signer.verification_key_signature,
+            operational_certificate: signer.operational_certificate,
+            kes_period: signer.kes_period,
+        })
+        .collect::<Vec<_>>()
+}
+
+#[derive(Debug, Parser)]
+#[command(author, version, about, long_about = None)]
+pub struct MainOpts {
+    /// Location of the Cardano CLI binary
+    #[arg(short, long)]
+    pub cardano_cli_path: PathBuf,
+
+    /// Temporary location for logs, databases etc.
+    #[arg(short, long)]
+    pub temporary_path: Option<PathBuf>,
+
+    /// Path of the Aggregator binary
+    #[arg(short, long, default_value = "./target/debug")]
+    pub aggregator_dir: PathBuf,
+
+    /// Number of concurrent signers
+    #[arg(long, default_value = "2")]
+    pub num_signers: usize,
+
+    /// Mithril technical Era
+    #[arg(long, default_value = "thales")]
+    pub mithril_era: String,
+
+    /// Aggregator HTTP port
+    #[arg(short = 'p', long, default_value = "8888")]
+    server_port: u32,
+
+    /// Log level
+    #[arg(short='v', action = clap::ArgAction::Count)]
+    verbose: u8,
+}
+
+impl MainOpts {
+    /// get log level from parameters
+    pub fn log_level(&self) -> Level {
+        match self.verbose {
+            0 => Level::Warning,
+            1 => Level::Info,
+            2 => Level::Debug,
+            _ => Level::Trace,
+        }
+    }
+}
+
+#[derive(Debug)]
+pub struct AggregatorParameters {
+    server_port: u32,
+    bft_node: BftNode,
+    cardano_cli_path: PathBuf,
+    work_dir: PathBuf,
+    bin_dir: PathBuf,
+    mithril_era: String,
+}
+
+impl AggregatorParameters {
+    fn new(opts: &MainOpts) -> StdResult<Self> {
+        // configure a dummy immutable db
+        let immutable_db = DummyImmutablesDbBuilder::new("load-tester")
+            .with_immutables(&[1, 2, 3])
+            .append_immutable_trio()
+            .build();
+
+        let bft_node = BftNode {
+            db_path: immutable_db.dir,
+            socket_path: PathBuf::new(),
+        };
+        let tmp_dir = opts
+            .temporary_path
+            .as_ref()
+            .cloned()
+            .unwrap_or_else(|| std::env::temp_dir().join("load-aggregator"));
+
+        if tmp_dir.exists() {
+            std::fs::remove_dir_all(&tmp_dir).with_context(|| {
+                format!(
+                    "Could not remove existing temp directory '{}'.",
+                    tmp_dir.display()
+                )
+            })?;
+        }
+        std::fs::create_dir_all(&tmp_dir)
+            .with_context(|| format!("Could not create temp directory '{}'.", tmp_dir.display()))?;
+
+        let cardano_cli_path = {
+            if !opts.cardano_cli_path.exists() {
+                Err(format!(
+                    "Given cardano-cli path does not exist: {}",
+                    opts.cardano_cli_path.display()
+                ))?
+            }
+
+            opts.cardano_cli_path.canonicalize().with_context(|| {
+                format!(
+                    "Could not canonicalize path to the cardano-cli, path: {}",
+                    opts.cardano_cli_path.display()
+                )
+            })?
+        };
+
+        let aggregator_parameters = AggregatorParameters {
+            bft_node,
+            bin_dir: opts.aggregator_dir.clone(),
+            cardano_cli_path,
+            server_port: opts.server_port,
+            work_dir: tmp_dir,
+            mithril_era: opts.mithril_era.clone(),
+        };
+
+        Ok(aggregator_parameters)
+    }
+}
+
+#[tokio::main]
+async fn main() -> StdResult<()> {
+    let opts = MainOpts::parse();
+    let _logger = init_logger(&opts);
+    let args = AggregatorParameters::new(&opts)?;
+    println!("OPTIONS={opts:?}");
+    let signers_fixture = generate_signer_data(opts.num_signers);
+    let register_messages = generate_register_message(&signers_fixture);
+    let mut aggregator = Aggregator::new(
+        args.server_port as u64,
+        &args.bft_node,
+        &args.cardano_cli_path,
+        &args.work_dir,
+        &args.bin_dir,
+        &args.mithril_era,
+    )
+    .unwrap();
+    let progress_bar = ProgressBar::new_spinner().with_message("starting Aggregator process…");
+    aggregator.set_protocol_parameters(&ProtocolParameters::default());
+    aggregator.serve().unwrap();
+    let spinner = async move {
+        loop {
+            progress_bar.tick();
+            sleep(Duration::from_millis(50)).await;
+        }
+    };
+
+    select! {
+        _ = spinner => (),
+        _ = sleep(Duration::from_secs(10)) => (),
+    }
+
+    let mut join_set: JoinSet<StdResult<()>> = JoinSet::new();
+    let progress_bar =
+        ProgressBar::with_draw_target(Some(opts.num_signers as u64), ProgressDrawTarget::stdout());
+
+    for register in register_messages {
+        let endpoint = aggregator.endpoint();
+        join_set.spawn(async move {
+            let response = reqwest::Client::new()
+                .post(format!("{}/register-signer", endpoint))
+                .json(&register)
+                .send()
+                .await
+                .unwrap();
+
+            match response.status() {
+                StatusCode::CREATED => Ok(()),
+                status => Err(LoadError::SignerRegistrationError {
+                    expected_http_code: 201,
+                    got_http_code: status.as_u16() as u32,
+                    party_id: register.party_id,
+                    error_message: response.text().await.unwrap(),
+                }
+                .into()),
+            }
+        });
+    }
+    let mut errors = 0;
+
+    while let Some(res) = join_set.join_next().await {
+        let res = res.expect("Tokio task join failed!");
+        progress_bar.inc(1);
+
+        if res.is_err() {
+            // eprintln!("Signer error caught: {res:?}");
+            errors += 1;
+        }
+    }
+
+    assert_eq!(opts.num_signers - 1, errors);
+
+    // ensure POSTing payload gives 200
+    aggregator.stop().await.unwrap();
+    Ok(())
+}

mithril-test-lab/mithril-end-to-end/src/bin/load-aggregator/main.rs

@@ -5,7 +5,7 @@ mod mithril;
 mod run_only;
 mod utils;
 
-pub use devnet::Devnet;
+pub use devnet::*;
 pub use end_to_end_spec::Spec;
 pub use mithril::*;
 pub use run_only::RunOnly;