Add dev tooling for creating cluster arrangements from a config

Author: spacekookie

Cargo.lock

@@ -36,6 +36,7 @@ indexmap = { version = "2.1.0", features = ["serde"] }
 itertools = { version = "0.10.5" }
 metrics = "0.22.0"
 metrics-exporter-prometheus = { version = "0.13.0", default-features = false, features = ["http-listener"] }
+nom = "7.0"
 once_cell = "1.17.1"
 opentelemetry = { version = "0.20.0", features = ["rt-tokio"] }
 opentelemetry-otlp = { version = "0.13.0" }
@@ -47,6 +48,7 @@ quinn-proto = "0.10.5"
 quinn-plaintext = { version = "0.2.0" }
 quoted-string = "0.6.1"
 rand = { version = "0.8.5", features = ["small_rng"] }
+random_graphs = "0.1"
 rangemap = { version = "1.4.0" }
 rcgen = { version = "0.11.1", features = ["x509-parser"] }
 rhai = { version = "1.15.1", features = ["sync"] }

Cargo.toml

@@ -0,0 +1,13 @@
+[package]
+name = "corro-devcluster"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+serde = { workspace = true, features = [ "derive" ] }
+serde_json = { workspace = true }
+thiserror = { workspace = true }
+rand = { workspace = true }
+random_graphs = { workspace = true }
+nom = { workspace = true }
+clap = { workspace = true }

crates/corro-devcluster/Cargo.toml

@@ -0,0 +1,11 @@
+# corro-devcluster
+
+A tool to quickly start a corrosion cluster for development.
+
+- Build corrosion with nix (keep in mind that changes to the
+  `Cargo.lock` file need to be checked into git to be visible to a nix
+  build)
+- Provide a topology config (`.txt` is fine, see
+  [example_topologies](example_topologies/))
+- Provide a state directory.  Node state and log output will be kept
+  in separated directories.

crates/corro-devcluster/README.md

@@ -0,0 +1 @@
+A -> B

crates/corro-devcluster/example_topologies/a_b.txt

@@ -0,0 +1,3 @@
+A -> B
+B -> C
+C -> A

crates/corro-devcluster/example_topologies/a_b_c_circle.txt

@@ -0,0 +1,259 @@
+mod topology;
+
+use clap::{Parser, Subcommand};
+use serde::Deserialize;
+use std::{
+    collections::{BTreeMap, HashMap},
+    env,
+    fs::File,
+    io::{Read, Stdout, Write},
+    path::{Path, PathBuf},
+    process::{Command, Stdio},
+    sync::mpsc::{channel, RecvError, Sender},
+    thread,
+    time::Duration,
+};
+
+use crate::topology::Simple;
+
+#[derive(Parser)]
+#[clap(version = env!("CARGO_PKG_VERSION"))]
+struct Cli {
+    /// Set the state directory path.  If not set the environment
+    /// variable CORRO_DEVCLUSTER_STATE_DIR will be used
+    #[clap(long = "statedir", short = 'd', global = true)]
+    state_directory: Option<PathBuf>,
+
+    /// Set the state directory path.  If not set the environment
+    /// variable CORRO_DEVCLUSTER_SCHEMA_DIR will be used
+    #[clap(long = "schemadir", short = 's', global = true)]
+    schema_directory: Option<PathBuf>,
+
+    /// Provide the binary path for corrosion.  If none is provided,
+    /// corrosion will be built with nix (which may take a minute)
+    #[clap(long = "binpath", short = 'b', global = true)]
+    binary_path: Option<String>,
+
+    #[command(subcommand)]
+    command: CliCommand,
+}
+
+#[derive(Subcommand)]
+enum CliCommand {
+    /// Create a simple topology in format `A -> B`, `B -> C`, etc
+    Simple {
+        /// Set the topology file path
+        topology_path: PathBuf,
+    },
+}
+
+fn main() {
+    let cli: Cli = Cli::parse();
+
+    let state_dir = match cli
+        .state_directory
+        .or(env::var("CORRO_DEVCLUSTER_STATE_DIR")
+            .ok()
+            .map(|path| PathBuf::new().join(path)))
+    {
+        Some(dir) => dir,
+        None => {
+            eprintln!("FAILED: either pass `--statedir` or set 'CORRO_DEVCLUSTER_STATE_DIR' environment variable!");
+            std::process::exit(1);
+        }
+    };
+
+    let schema_dir = match cli
+        .schema_directory
+        .or(env::var("CORRO_DEVCLUSTER_SCHEMA_DIR")
+            .ok()
+            .map(|path| PathBuf::new().join(path)))
+    {
+        Some(dir) => dir,
+        None => {
+            eprintln!("FAILED: either pass `--statedir` or set 'CORRO_DEVCLUSTER_STATE_DIR' environment variable!");
+            std::process::exit(1);
+        }
+    };
+
+    let bin_path = cli
+        .binary_path
+        .or_else(|| build_corrosion().map(|h| h.path))
+        .expect("failed to determine corrosion binary location!");
+
+    match cli.command {
+        CliCommand::Simple { topology_path } => {
+            let mut topo_config = File::open(topology_path).expect("failed to open topology-file!");
+            let mut topo_buffer = String::new();
+            topo_config
+                .read_to_string(&mut topo_buffer)
+                .expect("failed to read topology-file!");
+
+            let mut topology = Simple::default();
+            topo_buffer.lines().for_each(|line| {
+                topology
+                    .parse_edge(line)
+                    .expect("Syntax error in topology-file!");
+            });
+
+            run_simple_topology(topology, bin_path, state_dir, schema_dir);
+        }
+    }
+
+    // let handle = build_corrosion(env::args().next().map(|s| PathBuf::new().join(s)).unwrap());
+    // println!("{:#?}", handle);
+}
+
+fn run_simple_topology(topo: Simple, bin_path: String, state_dir: PathBuf, schema_dir: PathBuf) {
+    println!("//// Creating topology: \n{:#?}", topo);
+    let nodes = topo.get_all_nodes();
+
+    let mut port_map = BTreeMap::default();
+
+    // First go assign ports to all the nodes
+    for node_name in &nodes {
+        // Generate a port in range 1025 - 32768
+        let node_port: u16 = 1025 + rand::random::<u16>() % (32 * 1024) - 1025;
+        port_map.insert(node_name.clone(), node_port);
+    }
+
+    // Then generate each config with the appropriate bootstrap_set
+    for node_name in &nodes {
+        let node_port = port_map.get(node_name).unwrap(); // We just put it there
+        let node_state = state_dir.join(node_name);
+
+        // Delete / create the node state directory
+        let _ = std::fs::remove_dir(&node_state);
+        let _ = std::fs::create_dir_all(&node_state);
+
+        let mut bootstrap_set = vec![];
+        for link in topo.inner.get(node_name).unwrap() {
+            bootstrap_set.push(format!(
+                "\"[::1]:{}\"", // only connect locally
+                port_map.get(link).expect("Port for node not set!")
+            ));
+        }
+
+        let node_config = generate_config(
+            node_state.to_str().unwrap(),
+            schema_dir.to_str().unwrap(),
+            *node_port,
+            bootstrap_set,
+        );
+
+        println!(
+            "Generated config for node '{}': \n{}",
+            node_name, node_config
+        );
+
+        let mut config_file = File::create(node_state.join("config.toml"))
+            .expect("failed to create node config file");
+        config_file
+            .write_all(node_config.as_bytes())
+            .expect("failed to write node config file");
+    }
+
+    let (tx, rx) = channel::<()>();
+
+    // Spawn nodes those without bootstraps first if they exist.
+    for (pure_responder, _) in topo.inner.iter().filter(|(_, vec)| vec.is_empty()) {
+        run_corrosion(tx.clone(), bin_path.clone(), state_dir.join(pure_responder));
+        thread::sleep(Duration::from_millis(250)); // give the start thread a bit of time to breathe
+    }
+
+    for (initiator, _) in topo.inner.iter().filter(|(_, vec)| !vec.is_empty()) {
+        run_corrosion(tx.clone(), bin_path.clone(), state_dir.join(initiator));
+        thread::sleep(Duration::from_millis(250)); // give the start thread a bit of time to breathe
+    }
+
+    // wait for the threads
+    while let Ok(()) = rx.recv() {}
+    Command::new("pkill")
+        .arg("corrosion")
+        .output()
+        .expect("failed to gracefully kill corrosions. They've become sentient!!!");
+}
+
+fn generate_config(
+    state_dir: &str,
+    schema_dir: &str,
+    port: u16,
+    bootstrap_set: Vec<String>,
+) -> String {
+    let bootstrap = bootstrap_set.join(",");
+    format!(
+        r#"[db]
+path = "{state_dir}/corrosion.db"
+schema_paths = ["{schema_dir}"]
+
+[gossip]
+addr = "[::]:{port}"
+external_addr = "[::1]:{port}"
+bootstrap = [{bootstrap}]
+plaintext = true
+
+[api]
+addr = "127.0.0.1:{api_port}"
+
+[admin]
+path = "{state_dir}/admin.sock"
+"#,
+        state_dir = state_dir,
+        schema_dir = schema_dir,
+        port = port,
+        // the chances of a collision here are very very small since
+        // every port is random
+        api_port = port + 1,
+        bootstrap = bootstrap
+    )
+}
+
+#[derive(Debug)]
+struct BinHandle {
+    path: String,
+}
+
+fn nix_output(vec: &Vec<u8>) -> Vec<HashMap<String, serde_json::Value>> {
+    serde_json::from_slice(vec).unwrap()
+}
+
+fn run_corrosion(tx: Sender<()>, bin_path: String, state_path: PathBuf) {
+    let node_log = File::create(state_path.join("node.log")).expect("couldn't create log file");
+    let mut cmd = Command::new(bin_path);
+
+    cmd.args([
+        "-c",
+        state_path.join("config.toml").to_str().unwrap(),
+        "agent",
+    ]);
+
+    cmd.stdout(node_log);
+    let mut cmd_handle = cmd.spawn().expect("failed to spawn corrosion!");
+
+    thread::spawn(move || {
+        println!("Waiting for node...");
+        cmd_handle
+            .wait()
+            .expect("corrosion node has encountered an error!");
+        tx.send(()).unwrap();
+        println!("Node completed")
+    });
+}
+
+fn build_corrosion() -> Option<BinHandle> {
+    println!("Running 'nix build' ...");
+    let build_output = Command::new("nix")
+        .args(["build", "--json"])
+        .output()
+        .ok()?;
+
+    let json = nix_output(&build_output.stdout).remove(0);
+
+    Some(BinHandle {
+        path: json
+            .get("outputs")?
+            .get("out")?
+            .to_string()
+            .replace("\"", ""),
+    })
+}

crates/corro-devcluster/src/main.rs

@@ -0,0 +1,52 @@
+use nom::{
+    bytes::complete::tag,
+    character::complete::{alpha1, multispace0},
+    sequence::delimited,
+    IResult,
+};
+use std::collections::{BTreeMap, BTreeSet};
+
+/// Topology definition as a set of graph edges
+#[derive(Debug, Default)]
+pub struct Simple {
+    pub(crate) inner: BTreeMap<String, Vec<String>>,
+}
+
+impl Simple {
+    /// Parse a topology line in the following format:
+    ///
+    /// A -> B
+    /// B -> C
+    /// A -> C
+    /// etc
+    pub fn parse_edge<'top, 'input>(&mut self, input: &'input str) -> IResult<&'input str, ()> {
+        let (input, first) = alpha1(input)?;
+        let (input, _) = delimited(multispace0, tag("->"), multispace0)(input)?;
+        let (input, second) = alpha1(input)?;
+
+        // Add first -> second edge
+        self.inner
+            .entry(first.to_string())
+            .or_default()
+            .push(second.to_string());
+        // Add second to the map if it doesn't exist yet but don't
+        // create the connection edge
+        self.inner.entry(second.to_string()).or_default();
+
+        Ok((input, ()))
+    }
+
+    pub fn get_all_nodes(&self) -> Vec<String> {
+        self.inner
+            .keys()
+            .fold(BTreeSet::new(), |mut acc, key| {
+                acc.insert(key.clone());
+                self.inner.get(key).unwrap().iter().for_each(|value| {
+                    acc.insert(value.clone());
+                });
+                acc
+            })
+            .into_iter()
+            .collect()
+    }
+}