diff --git a/Cargo.lock b/Cargo.lock index c550337..b1305fa 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -52,6 +52,57 @@ dependencies = [ "windows-sys", ] +[[package]] +name = "async-channel" +version = "2.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "924ed96dd52d1b75e9c1a3e6275715fd320f5f9439fb5a4a11fa51f4221158d2" +dependencies = [ + "concurrent-queue", + "event-listener-strategy", + "futures-core", + "pin-project-lite", +] + +[[package]] +name = "async-fs" +version = "2.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8034a681df4aed8b8edbd7fbe472401ecf009251c8b40556b304567052e294c5" +dependencies = [ + "async-lock", + "blocking", + "futures-lite", +] + +[[package]] +name = "async-lock" +version = "3.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "290f7f2596bd5b78a9fec8088ccd89180d7f9f55b94b0576823bbbdc72ee8311" +dependencies = [ + "event-listener", + "event-listener-strategy", + "pin-project-lite", +] + +[[package]] +name = "async-task" +version = "4.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b75356056920673b02621b35afd0f7dda9306d03c79a30f5c56c44cf256e3de" + +[[package]] +name = "async-walkdir" +version = "2.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8ed40ca66d45a6775511920324e4ae05c81673f9133c187c4970a3089c3ae27c" +dependencies = [ + "async-fs", + "futures-lite", + "thiserror", +] + [[package]] name = "atomic-waker" version = "1.1.2" @@ -82,6 +133,19 @@ version = "2.13.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3ded4057c258ba199e2d26386d3af3780957ecaee6c4ef4041c6b4b8b97c0b06" +[[package]] +name = "blocking" +version = "1.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a70e4329df6cb94385eed412ec92375c3cdd8a6e502493d1229b6414e4036dfa" +dependencies = [ + "async-channel", + "async-task", + "futures-io", + "futures-lite", + "piper", +] + [[package]] name = "bstr" version = "1.13.1" @@ -184,6 +248,15 @@ version = "1.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570" +[[package]] +name = "concurrent-queue" +version = "2.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4ca0197aee26d1ae37445ee532fefce43251d24cc7c166799f4d46817f1d3973" +dependencies = [ + "crossbeam-utils", +] + [[package]] name = "core-foundation" version = "0.10.1" @@ -204,9 +277,11 @@ checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b" name = "crete" version = "0.1.3" dependencies = [ + "async-walkdir", "byteorder", "clap", "clap_complete", + "futures", "gix-discover", "minijinja", "netascii", @@ -220,9 +295,14 @@ dependencies = [ "serde_yaml", "thiserror", "tokio", - "walkdir", ] +[[package]] +name = "crossbeam-utils" +version = "0.8.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a31eee39dddec8330830986fcd7625edb5a24ec90ea038215273bbc3adb08ac6" + [[package]] name = "defmt" version = "0.3.100" @@ -302,6 +382,26 @@ dependencies = [ "windows-sys", ] +[[package]] +name = "event-listener" +version = "5.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a23add41df1562121a9393cb065eab5146a1242410f23a644851e90cfd669d2" +dependencies = [ + "parking", + "pin-project-lite", +] + +[[package]] +name = "event-listener-strategy" +version = "0.5.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8be9f3dfaaffdae2972880079a491a1a8bb7cbed0b8dd7a347f668b4150a3b93" +dependencies = [ + "event-listener", + "pin-project-lite", +] + [[package]] name = "faster-hex" version = "0.10.1" @@ -350,6 +450,21 @@ dependencies = [ "percent-encoding", ] +[[package]] +name = "futures" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9a31d2a3fbaaeb2af2368bbdd904aa8e812d3c04a1ee10d3171f52d556e5d0a3" +dependencies = [ + "futures-channel", + "futures-core", + "futures-executor", + "futures-io", + "futures-sink", + "futures-task", + "futures-util", +] + [[package]] name = "futures-channel" version = "0.3.34" @@ -366,12 +481,36 @@ version = "0.3.34" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "92d699e522242e69e3003b94ecc1f960f3a5e015aa7c5d7486e65ad01dd94f5e" +[[package]] +name = "futures-executor" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "031b47cf1a3c6cc8bc2fc76cd437f521619387907d469316e7c0bc278f1f5432" +dependencies = [ + "futures-core", + "futures-task", + "futures-util", +] + [[package]] name = "futures-io" version = "0.3.34" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "53c0fa8157de1303bfffdaa1cc2a673bfffb60102f76b0ef4441659124373fed" +[[package]] +name = "futures-lite" +version = "2.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f78e10609fe0e0b3f4157ffab1876319b5b0db102a2c60dc4626306dc46b44ad" +dependencies = [ + "fastrand", + "futures-core", + "futures-io", + "parking", + "pin-project-lite", +] + [[package]] name = "futures-macro" version = "0.3.34" @@ -401,6 +540,7 @@ version = "0.3.34" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0d50a92467f8ba5dd6e3ee5d4bd04d73ab2e4e1c44474a0674821dfce14b79bc" dependencies = [ + "futures-channel", "futures-core", "futures-io", "futures-macro", @@ -1191,6 +1331,12 @@ dependencies = [ "syn 2.0.119", ] +[[package]] +name = "parking" +version = "2.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f38d5652c16fde515bb1ecef450ab0f6a219d619a7274976324d5e377f7dceba" + [[package]] name = "parking_lot" version = "0.12.5" @@ -1226,6 +1372,17 @@ version = "0.2.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" +[[package]] +name = "piper" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c835479a4443ded371d6c535cbfd8d31ad92c5d23ae9770a61bc155e4992a3c1" +dependencies = [ + "atomic-waker", + "fastrand", + "futures-io", +] + [[package]] name = "pkg-config" version = "0.3.34" @@ -1370,9 +1527,7 @@ checksum = "16a1cfa75cc186dd73d5818e510e042e40927bccc9c236b061cea97e1eb08029" dependencies = [ "base64", "bytes", - "futures-channel", "futures-core", - "futures-util", "http", "http-body", "http-body-util", @@ -1388,14 +1543,12 @@ dependencies = [ "sync_wrapper", "tokio", "tokio-native-tls", - "tokio-util", "tower", "tower-http", "tower-service", "url", "wasm-bindgen", "wasm-bindgen-futures", - "wasm-streams", "web-sys", ] @@ -1768,19 +1921,6 @@ dependencies = [ "tokio", ] -[[package]] -name = "tokio-util" -version = "0.7.19" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "494815d09bf52b5548659851081238f0ca39ff638363907596da739561c62c52" -dependencies = [ - "bytes", - "futures-core", - "futures-sink", - "pin-project-lite", - "tokio", -] - [[package]] name = "toml_datetime" version = "1.1.1+spec-1.1.0" @@ -2013,19 +2153,6 @@ dependencies = [ "unicode-ident", ] -[[package]] -name = "wasm-streams" -version = "0.5.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9d1ec4f6517c9e11ae630e200b2b65d193279042e28edd4a2cda233e46670bbb" -dependencies = [ - "futures-util", - "js-sys", - "wasm-bindgen", - "wasm-bindgen-futures", - "web-sys", -] - [[package]] name = "web-sys" version = "0.3.106" diff --git a/Cargo.toml b/Cargo.toml index cbee3bc..d6e5bd7 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -14,17 +14,17 @@ schemars = { version = "1.2.2", features = ["semver1"] } serde_json = "1.0.151" [dependencies] +async-walkdir = "2.2.0" byteorder = "1.5.0" clap = { version = "4.6.7", features = ["derive"] } clap_complete = "4.6.11" +futures = "0.3.34" gix-discover = { version = "0.56.0", features = ["sha1"] } minijinja = { version = "2.24.0", features = ["json", "loader"] } netascii = "0.1.0" num_enum = "0.7.6" optional_struct = "0.5.2" reqwest = { version = "0.13.5", default-features = false, features = [ - "blocking", - "stream", "native-tls", ] } schemars = { workspace = true } @@ -34,7 +34,6 @@ serde_json = { workspace = true } serde_yaml = "0.9.34" thiserror = "2.0.21" tokio = { version = "1.52.1", features = ["full"] } -walkdir = "2.5.0" [lib] test = false diff --git a/src/cluster.rs b/src/cluster.rs index bce3f70..697f4be 100644 --- a/src/cluster.rs +++ b/src/cluster.rs @@ -1,13 +1,15 @@ use std::net::Ipv4Addr; use std::path::PathBuf; -use std::process::Command; use std::str::FromStr; +use async_walkdir::WalkDir; +use futures::StreamExt; +use futures::stream::FuturesUnordered; use minijinja::Environment; use optional_struct::{Applicable, optional_struct}; use schemars::JsonSchema; use serde::{Deserialize, Serialize}; -use walkdir::WalkDir; +use tokio::process::Command; use crate::environment::PathEnvironment; use crate::node::{Node, OptionalNodeDeserialize}; @@ -68,7 +70,7 @@ struct ClusterDeserialize { base: Base, } -#[derive(Debug, Serialize)] +#[derive(Debug, Serialize, Clone)] #[serde(rename_all = "camelCase")] pub struct Cluster { pub(crate) name: String, @@ -80,11 +82,11 @@ pub struct Cluster { } impl Cluster { - pub(crate) fn get(cluster_name: &str, env: &Environment) -> Self { + pub(crate) async fn get(cluster_name: &str, env: &Environment<'static>) -> Self { let path = get_talos_path().join("clusters").join("default.yaml"); let default: OptionalClusterDeserialize = if path.exists() { - let content = std::fs::read_to_string(path).unwrap(); + let content = tokio::fs::read_to_string(path).await.unwrap(); serde_yaml::from_str(&content).unwrap() } else { Default::default() @@ -92,7 +94,7 @@ impl Cluster { let mut path = get_talos_path().join("clusters").join(cluster_name); path.add_extension("yaml"); - let content = std::fs::read_to_string(path).unwrap(); + let content = tokio::fs::read_to_string(path).await.unwrap(); let mut cluster: OptionalClusterDeserialize = serde_yaml::from_str(&content).unwrap(); @@ -122,32 +124,39 @@ impl Cluster { let nodes = nodes .iter() - .map(|name| Node::get(name, env, &cluster, &default, &base)) - .collect(); + .map(|name| async { Node::get(name, env, &cluster, &default, &base).await }); + + let nodes = FuturesUnordered::from_iter(nodes).collect().await; Self { nodes, ..cluster } } - pub fn get_all() -> Vec { + pub async fn get_all() -> Vec { let env = PathEnvironment::new_patches(); WalkDir::new(get_talos_path().join("clusters")) - .into_iter() - .filter_map(|e| e.ok()) - .filter(|e| e.metadata().unwrap().is_file()) + .filter_map(|e| async { e.ok() }) + .filter(|e| { + let e = e.clone(); + async move { e.metadata().await.unwrap().is_file() } + }) .filter_map(|e| { - let mut path = PathBuf::from_str(&e.file_name().to_string_lossy()).unwrap(); - path.set_extension(""); + let env = env.clone(); + 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; + if name == "default" { + return None; + } + + Some(Cluster::get(&name, &env).await) } - - Some(Cluster::get(&name, &env)) }) .collect() + .await } pub fn nodes(&self) -> &[Node] { diff --git a/src/environment.rs b/src/environment.rs index 904e36e..615a2ec 100644 --- a/src/environment.rs +++ b/src/environment.rs @@ -2,15 +2,17 @@ use std::net::Ipv4Addr; use std::ops::Deref; use std::path::{Path, PathBuf}; +use async_walkdir::WalkDir; +use futures::{Stream, StreamExt}; use minijinja::value::ViaDeserialize; use minijinja::{AutoEscape, Environment, path_loader}; -use walkdir::WalkDir; use crate::{get_configs_path, get_talos_path}; /// Transparent wrapper around minijinja::Environment that loads templates from a path and /// configures better defaults. It also implements IntoIter, making it possible to iterate over all /// the templates. +#[derive(Clone)] pub struct PathEnvironment<'a> { env: Environment<'a>, path: PathBuf, @@ -76,6 +78,23 @@ impl<'a> PathEnvironment<'a> { pub(crate) fn new_patches() -> Self { Self::new(&get_talos_path().join("patches")) } + + pub fn files(&self) -> impl Stream { + // Find all templates in path + WalkDir::new(&self.path) + .filter_map(|e| async { e.ok() }) + .filter(|e| { + let e = e.clone(); + async move { e.metadata().await.unwrap().is_file() } + }) + .map(move |e| { + e.path() + .strip_prefix(&self.path) + .expect("All paths should have prefix") + .to_string_lossy() + .to_string() + }) + } } // Make PathEnvironment act like a normal Environment transparently @@ -86,25 +105,3 @@ impl<'a> Deref for PathEnvironment<'a> { &self.env } } - -// Iterate over all the files in the path -impl<'a, 'b> IntoIterator for &'b PathEnvironment<'a> { - type Item = String; - - type IntoIter = impl Iterator; - - fn into_iter(self) -> Self::IntoIter { - // Find all templates in path - WalkDir::new(&self.path) - .into_iter() - .filter_map(|e| e.ok()) - .filter(|e| e.metadata().unwrap().is_file()) - .map(|e| { - e.path() - .strip_prefix(&self.path) - .expect("All paths should have prefix") - .to_string_lossy() - .to_string() - }) - } -} diff --git a/src/lib.rs b/src/lib.rs index 7685d78..549d4b1 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,4 +1,4 @@ -#![feature(impl_trait_in_assoc_type, path_absolute_method)] +#![feature(path_absolute_method)] use std::path::{Path, PathBuf}; use std::sync::OnceLock; diff --git a/src/main.rs b/src/main.rs index b69af7f..08d4c51 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,14 +1,16 @@ mod cli; -use std::process::Command; +use std::pin::pin; use clap::{CommandFactory, Parser}; use clap_complete::{Shell, generate as generate_complete}; use crete::cluster::Cluster; use crete::environment::PathEnvironment; use crete::{get_configs_path, get_repo_path, set_repo_path}; +use futures::StreamExt; use minijinja::context; use thiserror::Error; +use tokio::process::Command; use crate::cli::{Cli, Commands, GlobalOpts}; @@ -18,8 +20,8 @@ enum Error { NoClustersFound, } -fn run_command(mut command: Command) { - let output = command.output().unwrap(); +async fn run_command(mut command: Command) { + let output = command.output().await.unwrap(); if !output.stdout.is_empty() { println!("{}", String::from_utf8_lossy(&output.stdout).trim_end()); } @@ -28,24 +30,25 @@ fn run_command(mut command: Command) { } } -fn generate(opts: &GlobalOpts) -> Result<(), Error> { +async fn generate(opts: &GlobalOpts) -> Result<(), Error> { set_repo_path(&opts.repo); - let clusters = Cluster::get_all(); + let clusters = Cluster::get_all().await; if clusters.is_empty() { return Err(Error::NoClustersFound); } let path = opts.repo.join("rendered"); - if path.exists() { - std::fs::remove_dir_all(&path).unwrap(); - } - std::fs::create_dir(&path).unwrap(); + // We don't really care if this fails, as that likely mean that the directory did not exist + tokio::fs::remove_dir_all(&path).await.ok(); + tokio::fs::create_dir_all(&path).await.unwrap(); // Render templates let template_env = PathEnvironment::new(&get_repo_path().join("templates")); - for template_name in &template_env { + + let mut stream = pin!(template_env.files()); + while let Some(template_name) = stream.next().await { let template = template_env.get_template(&template_name).unwrap(); let content = template @@ -55,35 +58,36 @@ fn generate(opts: &GlobalOpts) -> Result<(), Error> { }) .unwrap(); - std::fs::write(path.join(template_name), content).unwrap(); + tokio::fs::write(path.join(template_name), content) + .await + .unwrap(); } // Remove existing config files and create output directory let path = get_configs_path(); - if path.exists() { - std::fs::remove_dir_all(&path).unwrap(); - } - std::fs::create_dir(&path).unwrap(); + tokio::fs::remove_dir_all(&path).await.ok(); + 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)); + run_command(node.talosctl_gen_config_command(&cluster)).await; } - run_command(cluster.talosctl_gen_config_command()); - run_command(cluster.talosctl_add_endpoint_command()); - run_command(cluster.talosctl_merge_command()); + run_command(cluster.talosctl_gen_config_command()).await; + run_command(cluster.talosctl_add_endpoint_command()).await; + run_command(cluster.talosctl_merge_command()).await; } Ok(()) } -fn main() -> Result<(), Error> { +#[tokio::main] +async fn main() -> Result<(), Error> { let cli = Cli::parse(); match cli.command { - Commands::Generate => generate(&cli.global_opts)?, + Commands::Generate => generate(&cli.global_opts).await?, Commands::ShellCompletions => generate_complete( Shell::from_env().unwrap_or(Shell::Bash), &mut Cli::command(), diff --git a/src/node.rs b/src/node.rs index ab5c166..e4539da 100644 --- a/src/node.rs +++ b/src/node.rs @@ -1,10 +1,10 @@ use std::net::Ipv4Addr; -use std::process::Command; use minijinja::Environment; use optional_struct::{Applicable, optional_struct}; use schemars::JsonSchema; use serde::{Deserialize, Serialize}; +use tokio::process::Command; use crate::cluster::{Base, Cluster}; use crate::patch::{OptionalPatches, OptionalPatchesString, Patches}; @@ -90,7 +90,7 @@ pub(crate) struct NodeDeserialize { sops: Secret, } -#[derive(Debug, Serialize)] +#[derive(Debug, Serialize, Clone)] #[serde(rename_all = "camelCase")] pub struct Node { hostname: String, @@ -106,9 +106,9 @@ pub struct Node { } impl Node { - pub(crate) fn get( + pub(crate) async fn get( node_name: &str, - env: &Environment, + env: &Environment<'static>, cluster: &Cluster, default: &OptionalNodeDeserialize, base: &Base, @@ -121,7 +121,7 @@ impl Node { .to_string(); path.add_extension("yaml"); - let content = std::fs::read_to_string(path).unwrap(); + let content = tokio::fs::read_to_string(path).await.unwrap(); let node: OptionalNodeDeserialize = serde_yaml::from_str(&content).unwrap(); @@ -172,7 +172,7 @@ impl Node { }; // Render patches - let patches = patches.render(env, cluster, &node); + let patches = patches.render(env, cluster, &node).await; Node { patches, ..node } } diff --git a/src/patch.rs b/src/patch.rs index 4121bdb..a0b939f 100644 --- a/src/patch.rs +++ b/src/patch.rs @@ -16,27 +16,36 @@ pub(crate) struct Patches { pub(crate) type OptionalPatchesString = OptionalPatches; -fn render( +async fn render( patches: Vec, - env: &Environment, + env: &Environment<'static>, cluster: &Cluster, node: &Node, ) -> Vec { - patches - .into_iter() - .map(|name| { - let content = env - .get_template(&name) - .unwrap() - .render(context! { - node, - cluster - }) - .unwrap(); + let env = env.clone(); + let cluster = cluster.clone(); + let node = node.clone(); - serde_yaml::from_str(&content).unwrap() - }) - .collect() + // Minijinja will still have blocking I/O internally + tokio::task::spawn_blocking(move || { + patches + .into_iter() + .map(move |name| { + let content = env + .get_template(&name) + .unwrap() + .render(context! { + node, + cluster + }) + .unwrap(); + + serde_yaml::from_str(&content).unwrap() + }) + .collect() + }) + .await + .unwrap() } impl Patches { @@ -50,15 +59,15 @@ impl Patches { } } - pub(crate) fn render( + pub(crate) async fn render( self, - env: &Environment, + env: &Environment<'static>, cluster: &Cluster, node: &Node, ) -> Patches { Patches { - all: render(self.all.clone(), env, cluster, node), - control_plane: render(self.control_plane.clone(), env, cluster, node), + all: render(self.all.clone(), env, cluster, node).await, + control_plane: render(self.control_plane.clone(), env, cluster, node).await, } } } diff --git a/src/schematic.rs b/src/schematic.rs index e6f35d7..ace2f72 100644 --- a/src/schematic.rs +++ b/src/schematic.rs @@ -1,5 +1,6 @@ use schemars::JsonSchema; use serde::{Deserialize, Deserializer, Serialize}; +use tokio::runtime::Handle; use crate::get_talos_path; @@ -11,7 +12,7 @@ where let path = get_talos_path().join("schematics").join(name); let content = std::fs::read_to_string(path).unwrap().trim().to_owned(); - let client = reqwest::blocking::ClientBuilder::new() + let client = reqwest::ClientBuilder::new() .user_agent(format!( "{}/{}", std::env!("CARGO_PKG_NAME"), @@ -20,19 +21,30 @@ where .build() .unwrap(); - let res = client - .post("https://factory.talos.dev/schematics") - .body(content) - .send() - .map_err(serde::de::Error::custom)?; - #[derive(Debug, Deserialize)] struct Response { id: String, } - let response: Response = serde_json::from_str(&res.text().map_err(serde::de::Error::custom)?) - .map_err(serde::de::Error::custom)?; + // 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 + .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.id) }