Refactor load aggregator with 'stress_test' module

This makes the core more readable and usable.

Author: jpraynaud

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

@@ -3,6 +3,7 @@ mod devnet;
 mod end_to_end_spec;
 mod mithril;
 mod run_only;
+pub mod stress_test;
 mod utils;
 
 pub use devnet::*;

mithril-test-lab/mithril-end-to-end/src/lib.rs

@@ -0,0 +1,108 @@
+use slog_scope::info;
+use std::time::Duration;
+
+use mithril_common::{entities::Epoch, test_utils::MithrilFixture, StdResult};
+
+use crate::{
+    stress_test::{entities::AggregatorParameters, fake_chain, fake_signer, wait},
+    Aggregator,
+};
+
+/// Bootstrap an aggregator and make it compute its genesis certificate
+pub async fn bootstrap_aggregator(
+    args: &AggregatorParameters,
+    signers_fixture: &MithrilFixture,
+    current_epoch: &mut Epoch,
+) -> StdResult<Aggregator> {
+    info!(">> Launch Aggregator");
+    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();
+
+    fake_chain::set_epoch(&args.mock_epoch_file_path(), *current_epoch);
+    fake_chain::set_stake_distribution(&args.mock_stake_distribution_file_path(), signers_fixture);
+
+    // Extremely large interval since, for the two following starts, only the http_server part
+    // of the aggregator is relevant as we need to send signer registrations.
+    aggregator.change_run_interval(Duration::from_secs(20000));
+    aggregator.set_mock_cardano_cli_file_path(
+        &args.mock_stake_distribution_file_path(),
+        &args.mock_epoch_file_path(),
+    );
+    aggregator.set_protocol_parameters(&signers_fixture.protocol_parameters());
+
+    info!(
+        ">> Starting the aggregator with a large run interval to call the http_server\
+    without being bothered by the state machine cycles"
+    );
+    aggregator.serve().unwrap();
+    wait::for_http_response(
+        &format!("{}/epoch-settings", aggregator.endpoint()),
+        Duration::from_secs(10),
+        "Waiting for the aggregator to start",
+    )
+    .await?;
+
+    info!(">> Send the Signer Key Registrations payloads for the genesis signers");
+    let errors = fake_signer::register_signers_to_aggregator(
+        &aggregator,
+        &signers_fixture.signers(),
+        *current_epoch + 1,
+    )
+    .await?;
+    assert_eq!(0, errors);
+    aggregator.stop().await.unwrap();
+
+    info!(">> Move one epoch forward in order to issue the genesis certificate");
+    *current_epoch += 1;
+    fake_chain::set_epoch(&args.mock_epoch_file_path(), *current_epoch);
+
+    info!(">> Restarting the aggregator still with a large run interval");
+    aggregator.serve().unwrap();
+    wait::for_http_response(
+        &format!("{}/epoch-settings", aggregator.endpoint()),
+        Duration::from_secs(10),
+        "Waiting for the aggregator to start",
+    )
+    .await?;
+
+    info!(">> Send the Signer Key Registrations payloads for next genesis signers");
+    let errors = fake_signer::register_signers_to_aggregator(
+        &aggregator,
+        &signers_fixture.signers(),
+        *current_epoch + 1,
+    )
+    .await?;
+    assert_eq!(0, errors);
+    aggregator.stop().await.unwrap();
+
+    {
+        info!(">> Compute genesis certificate");
+        let mut genesis_aggregator = Aggregator::copy_configuration(&aggregator);
+        genesis_aggregator
+            .bootstrap_genesis()
+            .await
+            .expect("Genesis aggregator should be able to bootstrap genesis");
+    }
+
+    info!(">> Restart aggregator with a normal run interval");
+    aggregator.change_run_interval(Duration::from_secs(3));
+    aggregator.serve().unwrap();
+
+    wait::for_http_response(
+        &format!("{}/epoch-settings", aggregator.endpoint()),
+        Duration::from_secs(10),
+        "Waiting for the aggregator to restart",
+    )
+    .await?;
+
+    info!(">> Aggregator bootrapped");
+
+    Ok(aggregator)
+}

mithril-test-lab/mithril-end-to-end/src/stress_test/aggregator_helpers.rs

@@ -0,0 +1,173 @@
+use anyhow::{anyhow, Context};
+use clap::Parser;
+use mithril_common::StdResult;
+use slog::Level;
+use std::{
+    path::{Path, PathBuf},
+    time::Duration,
+};
+use tokio::time::Instant;
+
+use crate::BftNode;
+
+#[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 = "20")]
+    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 = "8080")]
+    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 {
+    pub server_port: u32,
+    pub bft_node: BftNode,
+    pub cardano_cli_path: PathBuf,
+    pub work_dir: PathBuf,
+    pub bin_dir: PathBuf,
+    pub mithril_era: String,
+}
+
+impl AggregatorParameters {
+    pub fn new(opts: &MainOpts, immutable_db_path: &Path) -> StdResult<Self> {
+        let bft_node = BftNode {
+            db_path: immutable_db_path.to_path_buf(),
+            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(anyhow!(
+                    "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)
+    }
+
+    pub fn mock_stake_distribution_file_path(&self) -> PathBuf {
+        self.work_dir.join("stake_distribution.json")
+    }
+
+    pub fn mock_epoch_file_path(&self) -> PathBuf {
+        self.work_dir.join("epoch.txt")
+    }
+}
+
+pub struct Timing {
+    phase: String,
+    duration: Duration,
+}
+
+pub struct Reporter {
+    number_of_signers: usize,
+    timings: Vec<Timing>,
+    current_timing: Option<(String, Instant)>,
+}
+
+impl Reporter {
+    pub fn new(number_of_signers: usize) -> Self {
+        Self {
+            number_of_signers,
+            timings: vec![],
+            current_timing: None,
+        }
+    }
+
+    pub fn start(&mut self, phase: &str) {
+        self.current_timing = Some((phase.to_owned(), Instant::now()));
+    }
+
+    pub fn stop(&mut self) {
+        match &self.current_timing {
+            Some((phase, instant)) => {
+                let timing = Timing {
+                    phase: phase.clone(),
+                    duration: instant.elapsed(),
+                };
+
+                self.timings.push(timing);
+                self.current_timing = None;
+            }
+            None => (),
+        }
+    }
+
+    pub fn print_report(&self) {
+        println!("number_of_signers\t{}", self.number_of_signers);
+        println!("phase\tduration/ms");
+        for t in &self.timings {
+            println!("{}\t{}", t.phase, t.duration.as_millis());
+        }
+    }
+}

mithril-test-lab/mithril-end-to-end/src/stress_test/entities.rs

@@ -0,0 +1,23 @@
+use slog_scope::debug;
+use std::{fs::File, io::Write, path::Path};
+
+use mithril_common::{entities::Epoch, test_utils::MithrilFixture};
+
+pub fn set_stake_distribution(
+    mock_stake_distribution_file_path: &Path,
+    signers_fixture: &MithrilFixture,
+) {
+    let mock_stake_distribution_file = File::create(mock_stake_distribution_file_path).unwrap();
+    serde_json::to_writer(
+        &mock_stake_distribution_file,
+        &signers_fixture.cardano_cli_stake_distribution(),
+    )
+    .expect("Writing the stake distribution into a file for the mock cardano cli failed");
+}
+
+pub fn set_epoch(mock_epoch_file_path: &Path, epoch: Epoch) {
+    let mock_epoch_file = File::create(mock_epoch_file_path).unwrap();
+    write!(&mock_epoch_file, "{}", *epoch)
+        .expect("Writing the epoch into a file for the mock cardano cli failed");
+    debug!("New Epoch: {epoch}");
+}