infra/crete
1
0
Fork 0
Code Issues 3 Pull Requests 1 Actions Packages Releases 4 Activity

feat: Make application async

Browse Source
This commit is contained in:
Dreaded_X
2026-10-04 05:39:17 +02:00
parent 9440ede0ab
commit bcf716038e
9 changed files with 290 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+158 -1
View File
@@ -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.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "37672978ae0febce7516ae0a85b53e6185159a9a28787391eb63fc44ec36037d"
dependencies = [
"async-fs",
"futures-lite",
"thiserror",
]
[[package]]
name = "atomic-waker"
version = "1.1.2"
@@ -85,6 +136,19 @@ dependencies = [
"generic-array",
]
[[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"
@@ -187,6 +251,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"
@@ -216,9 +289,11 @@ dependencies = [
name = "crete"
version = "0.1.3"
dependencies = [
"async-walkdir",
"byteorder",
"clap",
"clap_complete",
"futures",
"gix-discover",
"minijinja",
"netascii",
@@ -232,9 +307,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 = "crypto-common"
version = "0.1.7"
@@ -325,6 +405,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.0"
@@ -371,6 +471,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"
@@ -387,12 +502,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"
@@ -422,6 +561,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",
@@ -1218,6 +1358,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"
@@ -1253,6 +1399,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"
Cargo.toml
+2 -1
View File
@@ -14,9 +14,11 @@ schemars = { version = "1.2.1", features = ["semver1"] }
serde_json = "1.0.149"
[dependencies]
async-walkdir = "2.1.0"
byteorder = "1.5.0"
clap = { version = "4.6.0", features = ["derive"] }
clap_complete = "4.6.0"
futures = "0.3.34"
gix-discover = { version = "0.55.0", features = ["sha1"] }
minijinja = { version = "2.18.0", features = ["json", "loader"] }
netascii = "0.1.0"
@@ -34,7 +36,6 @@ serde_json = { workspace = true }
serde_yaml = "0.9.34"
thiserror = "2.0.18"
tokio = { version = "1.52.1", features = ["full"] }
walkdir = "2.5.0"
[lib]
test = false
src/cluster.rs
+22 -13
View File
@@ -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,20 +124,25 @@ 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<Cluster> {
pub async fn get_all() -> Vec<Cluster> {
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 env = env.clone();
async move {
let mut path = PathBuf::from_str(&e.file_name().to_string_lossy()).unwrap();
path.set_extension("");
@@ -145,9 +152,11 @@ impl Cluster {
return None;
}
Some(Cluster::get(&name, &env))
Some(Cluster::get(&name, &env).await)
}
})
.collect()
.await
}
pub fn nodes(&self) -> &[Node] {
src/environment.rs
+20 -23
View File
@@ -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<Item = String> {
// 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<Item = Self::Item>;
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()
})
}
}
src/lib.rs
+1 -1
View File
@@ -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;
src/main.rs
+25 -21
View File
@@ -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(),
src/node.rs
+6 -6
View File
@@ -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 }
}
src/patch.rs
+16 -7
View File
@@ -16,15 +16,21 @@ pub(crate) struct Patches<T> {
pub(crate) type OptionalPatchesString = OptionalPatches<String>;
fn render(
async fn render(
patches: Vec<String>,
env: &Environment,
env: &Environment<'static>,
cluster: &Cluster,
node: &Node,
) -> Vec<serde_yaml::Value> {
let env = env.clone();
let cluster = cluster.clone();
let node = node.clone();
// Minijinja will still have blocking I/O internally
tokio::task::spawn_blocking(move || {
patches
.into_iter()
.map(|name| {
.map(move |name| {
let content = env
.get_template(&name)
.unwrap()
@@ -37,6 +43,9 @@ fn render(
serde_yaml::from_str(&content).unwrap()
})
.collect()
})
.await
.unwrap()
}
impl Patches<String> {
@@ -50,15 +59,15 @@ impl Patches<String> {
}
}
pub(crate) fn render(
pub(crate) async fn render(
self,
env: &Environment,
env: &Environment<'static>,
cluster: &Cluster,
node: &Node,
) -> Patches<serde_yaml::Value> {
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,
}
}
}
src/schematic.rs
+20 -8
View File
@@ -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,20 +21,31 @@ 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)?)
// 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)
}