diff --git a/src/cluster.rs b/src/cluster.rs index 697f4be..bd3c670 100644 --- a/src/cluster.rs +++ b/src/cluster.rs @@ -4,6 +4,7 @@ use std::str::FromStr; use async_walkdir::WalkDir; use futures::StreamExt; +use futures::future::join_all; use futures::stream::FuturesUnordered; use minijinja::Environment; use optional_struct::{Applicable, optional_struct}; @@ -87,7 +88,11 @@ impl Cluster { let default: OptionalClusterDeserialize = if path.exists() { let content = tokio::fs::read_to_string(path).await.unwrap(); - serde_yaml::from_str(&content).unwrap() + + // The deserialization makes a blocking web request + tokio::task::spawn_blocking(move || serde_yaml::from_str(&content).unwrap()) + .await + .unwrap() } else { Default::default() }; @@ -134,29 +139,35 @@ impl Cluster { pub async fn get_all() -> Vec { let env = PathEnvironment::new_patches(); - WalkDir::new(get_talos_path().join("clusters")) + let names: Vec<_> = WalkDir::new(get_talos_path().join("clusters")) .filter_map(|e| async { e.ok() }) .filter(|e| { let e = e.clone(); async move { e.metadata().await.unwrap().is_file() } }) - .filter_map(|e| { - let env = env.clone(); - async move { - let mut path = PathBuf::from_str(&e.file_name().to_string_lossy()).unwrap(); - path.set_extension(""); + .filter_map(|e| async move { + let mut path = PathBuf::from_str(&e.file_name().to_string_lossy()).unwrap(); + path.set_extension(""); - let name = path.to_string_lossy().to_string(); + let name = path.to_string_lossy().to_string(); - if name == "default" { - return None; - } - - Some(Cluster::get(&name, &env).await) + if name == "default" { + return None; } + + Some(name) }) .collect() - .await + .await; + + // Get the clusters concurrently + join_all(names.iter().map(|name| { + let env = env.clone(); + let name = name.clone(); + + async move { Cluster::get(&name, &env).await } + })) + .await } pub fn nodes(&self) -> &[Node] { diff --git a/src/main.rs b/src/main.rs index 08d4c51..4cbb473 100644 --- a/src/main.rs +++ b/src/main.rs @@ -8,6 +8,7 @@ use crete::cluster::Cluster; use crete::environment::PathEnvironment; use crete::{get_configs_path, get_repo_path, set_repo_path}; use futures::StreamExt; +use futures::future::join_all; use minijinja::context; use thiserror::Error; use tokio::process::Command; @@ -69,13 +70,25 @@ async fn generate(opts: &GlobalOpts) -> Result<(), Error> { tokio::fs::create_dir_all(&path).await.unwrap(); // Generate config files - for cluster in clusters { - for node in cluster.nodes() { - run_command(node.talosctl_gen_config_command(&cluster)).await; - } + join_all(clusters.iter().map(|cluster| async move { + join_all(cluster.nodes().iter().map(|node| { + let cluster = cluster.clone(); + async move { + run_command(node.talosctl_gen_config_command(&cluster)).await; + } + })) + .await; + // Run these commands in parallel with the the node commands + // NOTE: These do need to happen in the correct order run_command(cluster.talosctl_gen_config_command()).await; run_command(cluster.talosctl_add_endpoint_command()).await; + })) + .await; + + // Merge all the cluster talosconfig files into a single file + // Since this all modifies the same output file this has to happen sequentially + for cluster in clusters { run_command(cluster.talosctl_merge_command()).await; } diff --git a/src/schematic.rs b/src/schematic.rs index ace2f72..794f4b1 100644 --- a/src/schematic.rs +++ b/src/schematic.rs @@ -29,21 +29,20 @@ where // We can't use async directly inside a custom deserializer, and using the blocking client also // does not work as it will try to start tokio inside of tokio. // So we have do this to work around it. - let response = tokio::task::block_in_place(|| { - Handle::current().block_on(async { - let res = client - .post("https://factory.talos.dev/schematics") - .body(content) - .send() - .await + // NOTE: This requires that the deserialization happens inside spawn_blocking + let response = Handle::current().block_on(async { + let res = client + .post("https://factory.talos.dev/schematics") + .body(content) + .send() + .await + .map_err(serde::de::Error::custom)?; + + let response: Response = + serde_json::from_str(&res.text().await.map_err(serde::de::Error::custom)?) .map_err(serde::de::Error::custom)?; - let response: Response = - serde_json::from_str(&res.text().await.map_err(serde::de::Error::custom)?) - .map_err(serde::de::Error::custom)?; - - Ok(response) - }) + Ok(response) })?; Ok(response.id)