cargo update

Cargo.toml

@@ -19,16 +19,18 @@ clap = { version = "4.5.40", features = ["derive", "env"] }
 clap_complete = { version = "4.5.54", features = ["unstable-dynamic"] }
 env_logger = "0.11.8"
 eyre = "0.6.12"
+fastrand = "2.3.0"
 futures = "0.3.31"
 futures-util = "0.3.31"
 glob = "0.3.2"
 hex = "0.4.3"
+human-units = "0.5.3"
 log = "0.4.27"
 lz4 = "1.28.1"
 nix = { version = "0.30.1", features = ["user"] }
 openssl = "0.10.73"
 page_size = "0.6.0"
-reqwest = { version = "0.12.20", features = ["json", "stream", "native-tls"] }
+reqwest = { version = "0.13.1", features = ["json", "stream", "native-tls"] }
 serde = { version = "1.0.219", features = ["derive"] }
 serde_json = "1.0.140"
 serde_yaml = "0.9.34"

src/bin/dkl.rs

@@ -1,6 +1,8 @@
 use clap::{CommandFactory, Parser, Subcommand};
 use eyre::{format_err, Result};
+use human_units::Duration;
 use log::{debug, error};
+use std::net::SocketAddr;
 use tokio::fs;
 
 #[derive(Parser)]
@@ -65,6 +67,17 @@ enum Command {
         #[arg(long, default_value = "/")]
         chroot: std::path::PathBuf,
     },
+    Proxy {
+        #[arg(long, short = 'l')]
+        listen: Vec<SocketAddr>,
+        targets: Vec<SocketAddr>,
+        /// target polling interval
+        #[arg(long, default_value = "30s")]
+        poll: Duration,
+        /// connect or check timeout
+        #[arg(long, default_value = "5s")]
+        timeout: Duration,
+    },
 }
 
 #[tokio::main(flavor = "current_thread")]
@@ -126,6 +139,20 @@ async fn main() -> Result<()> {
             .install(layer, version)
             .await
         }
+        C::Proxy {
+            listen,
+            targets,
+            poll,
+            timeout,
+        } => Ok(dkl::proxy::Proxy {
+            listen_addrs: listen,
+            targets,
+            poll: poll.into(),
+            timeout: timeout.into(),
+        }
+        .run()
+        .await
+        .map(|_| ())?),
     }
 }
 

src/bootstrap.rs

@@ -2,7 +2,7 @@ use std::collections::BTreeMap as Map;
 
 pub const TAKE_ALL: i16 = -1;
 
-#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
+#[derive(Clone, Debug, Default, serde::Deserialize, serde::Serialize)]
 pub struct Config {
     pub anti_phishing_code: String,
 
@@ -42,21 +42,11 @@ impl Config {
     pub fn new(bootstrap_dev: String) -> Self {
         Self {
             anti_phishing_code: "Direktil<3".into(),
-            keymap: None,
-            modules: None,
-            resolv_conf: None,
-            vpns: Map::new(),
-            networks: vec![],
-            auths: vec![],
-            ssh: Default::default(),
-            pre_lvm_crypt: vec![],
-            lvm: vec![],
-            crypt: vec![],
-            signer_public_key: None,
             bootstrap: Bootstrap {
                 dev: bootstrap_dev,
-                seed: None,
+                ..Default::default()
             },
+            ..Default::default()
         }
     }
 }
@@ -88,6 +78,17 @@ pub struct NetworkInterface {
     pub udev: Option<UdevFilter>,
 }
 
+impl Default for NetworkInterface {
+    fn default() -> Self {
+        Self {
+            var: "iface".into(),
+            n: 1,
+            regexps: Vec::new(),
+            udev: Some(UdevFilter::Eq("INTERFACE".into(), "eth0".into())),
+        }
+    }
+}
+
 #[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
 pub struct SSHServer {
     pub listen: String,
@@ -244,7 +245,7 @@ pub struct Raid {
     pub stripes: Option<u8>,
 }
 
-#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
+#[derive(Clone, Debug, Default, serde::Deserialize, serde::Serialize)]
 pub struct Bootstrap {
     pub dev: String,
     #[serde(skip_serializing_if = "Option::is_none")]

src/dls.rs

@@ -159,6 +159,30 @@ impl<'t> Host<'t> {
     }
 }
 
+#[derive(Default, serde::Deserialize, serde::Serialize)]
+#[serde(rename_all = "PascalCase")]
+pub struct Config {
+    #[serde(default, deserialize_with = "deserialize_null_as_default")]
+    pub clusters: Vec<ClusterConfig>,
+    #[serde(default, deserialize_with = "deserialize_null_as_default")]
+    pub hosts: Vec<HostConfig>,
+    #[serde(default, deserialize_with = "deserialize_null_as_default")]
+    pub host_templates: Vec<HostConfig>,
+    #[serde(default, rename = "SSLConfig")]
+    pub ssl_config: String,
+}
+
+// compensate for go's encoder pitfalls
+use serde::{Deserialize, Deserializer};
+fn deserialize_null_as_default<'de, D, T>(deserializer: D) -> std::result::Result<T, D::Error>
+where
+    T: Default + Deserialize<'de>,
+    D: Deserializer<'de>,
+{
+    let opt = Option::deserialize(deserializer)?;
+    Ok(opt.unwrap_or_default())
+}
+
 #[derive(serde::Deserialize, serde::Serialize)]
 #[serde(rename_all = "PascalCase")]
 pub struct ClusterConfig {
@@ -171,15 +195,15 @@ pub struct ClusterConfig {
 #[serde(rename_all = "PascalCase")]
 pub struct HostConfig {
     pub name: String,
-    #[serde(skip_serializing_if = "Option::is_none")]
+    #[serde(default, skip_serializing_if = "Option::is_none")]
     pub cluster_name: Option<String>,
 
     #[serde(rename = "IPs")]
     pub ips: Vec<IpAddr>,
 
-    #[serde(skip_serializing_if = "Map::is_empty")]
+    #[serde(default, skip_serializing_if = "Map::is_empty")]
     pub labels: Map<String, String>,
-    #[serde(skip_serializing_if = "Map::is_empty")]
+    #[serde(default, skip_serializing_if = "Map::is_empty")]
     pub annotations: Map<String, String>,
 
     #[serde(rename = "IPXE", skip_serializing_if = "Option::is_none")]
@@ -189,10 +213,13 @@ pub struct HostConfig {
     pub kernel: String,
     pub versions: Map<String, String>,
 
+    /// initrd config template
     pub bootstrap_config: String,
-    #[serde(skip_serializing_if = "Map::is_empty")]
+    /// files to add to the final initrd config, with rendering
-    pub initrd_files: Map<String, String>,
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub initrd_files: Vec<crate::File>,
 
+    /// system config template
     pub config: String,
 }
 

src/lib.rs

@@ -1,4 +1,5 @@
 pub mod apply;
+pub mod proxy;
 pub mod bootstrap;
 pub mod dls;
 pub mod dynlay;
@@ -52,7 +53,7 @@ pub struct User {
     pub gid: Option<u32>,
 }
 
-#[derive(Debug, serde::Deserialize, serde::Serialize)]
+#[derive(Debug, PartialEq, Eq, serde::Deserialize, serde::Serialize)]
 pub struct File {
     pub path: String,
     #[serde(skip_serializing_if = "Option::is_none")]
@@ -61,10 +62,21 @@ pub struct File {
     pub kind: FileKind,
 }
 
-#[derive(Debug, serde::Deserialize, serde::Serialize)]
+#[derive(Debug, PartialEq, Eq, serde::Deserialize, serde::Serialize)]
 #[serde(rename_all = "snake_case")]
 pub enum FileKind {
     Content(String),
     Symlink(String),
     Dir(bool),
 }
+
+// ------------------------------------------------------------------------
+
+impl Config {
+    pub fn has_file(&self, path: &str) -> bool {
+        self.files.iter().any(|f| f.path == path)
+    }
+    pub fn file(&self, path: &str) -> Option<&File> {
+        self.files.iter().find(|f| f.path == path)
+    }
+}

src/proxy.rs

@@ -0,0 +1,136 @@
+use log::{info, log_enabled, warn};
+use std::convert::Infallible;
+use std::io;
+use std::net::SocketAddr;
+use std::sync::atomic::{AtomicBool, Ordering::Relaxed};
+use std::time::Duration;
+use thiserror::Error;
+use tokio::net::{TcpListener, TcpStream};
+use tokio::time;
+
+pub struct Proxy {
+    pub listen_addrs: Vec<SocketAddr>,
+    pub targets: Vec<SocketAddr>,
+    pub poll: Duration,
+    pub timeout: Duration,
+}
+
+#[derive(Debug, Error)]
+pub enum Error {
+    #[error("failed to listen on {0}: {1}")]
+    ListenFailed(SocketAddr, std::io::Error),
+}
+
+pub type Result<T> = std::result::Result<T, Error>;
+
+impl Proxy {
+    pub async fn run(self) -> Result<Infallible> {
+        let mut listeners = Vec::with_capacity(self.listen_addrs.len());
+        for addr in self.listen_addrs {
+            listeners.push(
+                TcpListener::bind(&addr)
+                    .await
+                    .map_err(|e| Error::ListenFailed(addr, e))?,
+            );
+            info!("listening on {addr}");
+        }
+
+        // all targets are initially ok (better land on a down one than just fail)
+        let targets: Vec<_> = (self.targets.into_iter())
+            .map(|addr| TargetStatus {
+                addr,
+                up: AtomicBool::new(true),
+                timeout: self.timeout,
+            })
+            .collect();
+
+        // the proxy runs forever -> using 'static is not a leak
+        let targets = targets.leak();
+
+        for listener in listeners {
+            tokio::spawn(proxy_listener(listener, targets));
+        }
+
+        check_targets(targets, self.poll).await
+    }
+}
+
+struct TargetStatus {
+    addr: SocketAddr,
+    up: AtomicBool,
+    timeout: Duration,
+}
+impl TargetStatus {
+    fn is_up(&self) -> bool {
+        self.up.load(Relaxed)
+    }
+
+    fn set_up(&self, is_up: bool) {
+        let prev = self.up.swap(is_up, Relaxed);
+        if prev != is_up {
+            if is_up {
+                info!("{} is up", self.addr);
+            } else {
+                warn!("{} is down", self.addr);
+            }
+        }
+    }
+
+    async fn connect(&self) -> io::Result<TcpStream> {
+        let r = match time::timeout(self.timeout, TcpStream::connect(self.addr)).await {
+            Ok(r) => r,
+            Err(e) => Err(io::Error::new(io::ErrorKind::TimedOut, e)),
+        };
+
+        self.set_up(r.is_ok());
+        r
+    }
+}
+
+async fn check_targets(targets: &'static [TargetStatus], poll: Duration) -> ! {
+    use tokio::time;
+    let mut poll_ticker = time::interval(poll);
+    poll_ticker.set_missed_tick_behavior(time::MissedTickBehavior::Skip);
+
+    loop {
+        poll_ticker.tick().await;
+
+        let mut tasks = tokio::task::JoinSet::new();
+
+        for target in targets {
+            tasks.spawn(target.connect());
+        }
+
+        tasks.join_all().await;
+
+        if log_enabled!(log::Level::Info) {
+            let mut infos = String::new();
+            for ts in targets.iter() {
+                infos.push_str(&format!("{} ", ts.addr));
+                infos.push_str(if ts.is_up() { "up   " } else { "down " });
+            }
+            info!("{infos}");
+        }
+    }
+}
+
+async fn proxy_listener(listener: TcpListener, targets: &'static [TargetStatus]) {
+    let mut rng = fastrand::Rng::new();
+
+    loop {
+        let mut active = Vec::with_capacity(targets.len());
+        let (mut src, _) = listener.accept().await.expect("listener.accept() failed");
+
+        active.extend((targets.iter().enumerate()).filter_map(|(i, ts)| ts.is_up().then_some(i)));
+        rng.shuffle(&mut active);
+
+        tokio::spawn(async move {
+            for i in active {
+                if let Ok(mut dst) = targets[i].connect().await {
+                    let _ = tokio::io::copy_bidirectional(&mut src, &mut dst).await;
+                    break;
+                }
+            }
+        });
+    }
+}