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:53:13 +02:00
parent 406fa6259b
commit 992615c116
9 changed files with 290 additions and 133 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/cluster.rs
+28 -19
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,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<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 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] {
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
+29 -20
View File
@@ -16,27 +16,36 @@ 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> {
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<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
+21 -9
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,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)
}