xianyun_cluster/driver/

k3d.rs

//! Port of `internal/cluster/k3d/driver.go` and `internal/cluster/k3d/ports.go`:
//! a [`super::Driver`] implementation using k3d as the backend.
//!
//! This is the second highest-churn driver file in the Go codebase (51
//! commits per VCS hotspot analysis, the highest of all), so its tests are
//! ported with extra care for edge cases in `tests/k3d_driver.rs` and
//! `tests/k3d_ports.rs`.
//!
//! **Executor injection.** Following the pattern established by
//! `external::Driver` and `colima::Driver` (see their module doc
//! comments), this driver takes its `Arc<dyn CommandRunner + Send + Sync>`
//! at construction rather than through an ambient context. [`Driver::new`]
//! wires in the real `xianyun_executor::exec::Client` so existing callers
//! (`driver::auto`) need no change; [`Driver::with_runner`] is the
//! test-facing constructor.
//!
//! **`docker.Backend`.** The Go constructor also takes an injected
//! `docker.Backend` and panics if it's nil, but nothing in `driver.go` or
//! `ports.go` ever calls a method on it -- grepping the Go source for
//! `d.docker` or `.docker.` turns up nothing beyond the field assignment
//! in `NewDriver`. `xianyun-backends::docker` is also still a stub crate
//! (see `PORT_NOTES.md`), so this port omits the parameter entirely rather
//! than carry a genuinely dead field forward.

use std::io::Write as _;
use std::net::TcpStream;
use std::sync::Arc;

use super::{Driver as DriverTrait, DriverError, State, Status};
use xianyun_core::config::types::ClusterConfig;
use xianyun_executor::exec::CommandRunner;

pub const DRIVER_NAME: &str = "k3d";

/// Port of `k3d.Driver`.
pub struct Driver {
    pub cfg: ClusterConfig,
    pub profile_name: String,
    /// Port of `Driver.portInUse`: the seam tests substitute with a fake
    /// probe so unit tests never touch the network. Boxed rather than a
    /// bare `fn` pointer because tests need to capture state (e.g. record
    /// every address probed).
    pub port_in_use: Box<dyn Fn(&str) -> bool>,
    runner: Arc<dyn CommandRunner + Send + Sync>,
}

impl Driver {
    /// Port of `k3d.NewDriver`, wired to the real subprocess executor and
    /// a real TCP-connect port probe (`portprobe.InUse`'s equivalent).
    #[must_use]
    pub fn new(cfg: ClusterConfig, profile_name: String) -> Self {
        Self::with_runner(
            cfg,
            profile_name,
            Arc::new(xianyun_executor::exec::Client::new()),
        )
    }

    /// Test-facing constructor: takes the injected `CommandRunner`
    /// explicitly. `port_in_use` still defaults to the real TCP probe;
    /// override the field directly (it's `pub`) when a test needs a fake.
    #[must_use]
    pub fn with_runner(
        cfg: ClusterConfig,
        profile_name: String,
        runner: Arc<dyn CommandRunner + Send + Sync>,
    ) -> Self {
        Self {
            cfg,
            profile_name,
            port_in_use: Box::new(probe_address),
            runner,
        }
    }
}

impl DriverTrait for Driver {
    fn name(&self) -> &str {
        DRIVER_NAME
    }

    fn start(&self) -> Result<(), DriverError> {
        let cfg = self.cfg.clone();

        if !is_k3d_available(self) {
            return Err(DriverError::DriverNotFound("k3d".to_string()));
        }

        let Some(k3d_cfg) = cfg.k3d.clone() else {
            return Err(DriverError::MissingDriverConfig);
        };

        let status = status_impl(self).ok();

        if let Some(status) = &status
            && status.state == Some(State::Ready)
        {
            return verify_config(status, &cfg);
        }

        if let Some(status) = &status
            && status.exists
            && matches!(status.state, Some(State::Stopped) | Some(State::NotReady))
        {
            tracing::info!(
                profile = %self.profile_name,
                state = ?status.state,
                "starting existing k3d cluster"
            );

            let args = vec![
                "cluster".to_string(),
                "start".to_string(),
                self.profile_name.clone(),
            ];

            return self
                .runner
                .exec_streamed_command(None, "k3d", &args)
                .map_err(|e| DriverError::Other(format!("starting k3d: {e}")));
        }

        let port_mappings = merge_port_mappings(&k3d_cfg.port_mappings);

        check_port_conflicts(self, &port_mappings)?;

        let args = build_cluster_create_args(self, &cfg, &port_mappings)
            .map_err(|e| DriverError::Other(format!("building cluster create args: {e}")))?;

        warn_deprecated_disable_args(&k3d_cfg.extra_args);

        tracing::info!(
            profile = %self.profile_name,
            agents = k3d_cfg.agents,
            servers = k3d_cfg.servers,
            port_mappings = k3d_cfg.port_mappings.len(),
            extra_args = k3d_cfg.extra_args.len(),
            "starting k3d cluster"
        );

        self.runner
            .exec_streamed_command(None, "k3d", &args)
            .map_err(|e| DriverError::Other(format!("starting k3d: {e}")))
    }

    fn stop(&self) -> Result<(), DriverError> {
        let status = status_impl(self)?;

        if status.state == Some(State::Stopped) {
            return Ok(());
        }

        tracing::info!(profile = %self.profile_name, "stopping k3d cluster");

        let args = vec![
            "cluster".to_string(),
            "stop".to_string(),
            self.profile_name.clone(),
        ];

        self.runner
            .exec_streamed_command(None, "k3d", &args)
            .map_err(|e| DriverError::Other(format!("stopping k3d: {e}")))
    }

    fn destroy(&self) -> Result<(), DriverError> {
        if !is_k3d_available(self) {
            return Err(DriverError::DriverNotFound("k3d".to_string()));
        }

        let status = status_impl(self)?;

        if status.state == Some(State::Stopped) && !status.exists {
            return Ok(());
        }

        tracing::info!(profile = %self.profile_name, "destroying k3d cluster");

        let (opts, _guard) = disposable_kubeconfig_opts()
            .map_err(|e| DriverError::Other(format!("creating disposable kubeconfig: {e}")))?;

        let args = vec![
            "cluster".to_string(),
            "delete".to_string(),
            self.profile_name.clone(),
        ];

        self.runner
            .exec_streamed_command(Some(opts), "k3d", &args)
            .map_err(|e| DriverError::Other(format!("destroying k3d: {e}")))
    }

    fn status(&self) -> Result<Status, DriverError> {
        status_impl(self)
    }

    fn get_kubeconfig(&self) -> Result<String, DriverError> {
        if !is_k3d_available(self) {
            return Err(DriverError::DriverNotFound("k3d".to_string()));
        }

        let status = status_impl(self)?;

        if status.state != Some(State::Ready) {
            return Err(DriverError::ClusterNotRunning);
        }

        let args = vec![
            "kubeconfig".to_string(),
            "get".to_string(),
            self.profile_name.clone(),
        ];

        self.runner
            .read_stdout("k3d", &args)
            .map_err(|e| DriverError::Other(format!("reading kubeconfig: {e}")))
    }
}

/// An empty temp file created for [`disposable_kubeconfig_opts`], deleted
/// on drop.
struct TempKubeconfigGuard(std::path::PathBuf);

impl Drop for TempKubeconfigGuard {
    fn drop(&mut self) {
        let _ = std::fs::remove_file(&self.0);
    }
}

/// Creates a throwaway temp file to stand in for `kubeconfig.Create("")`
/// (out of scope for this session -- `kubeconfig` is owned by a parallel
/// worktree, and `tempfile` is only a dev-dependency of this crate).
/// Returns exec options with `KUBECONFIG` pointed at it and a guard that
/// deletes the file on drop.
fn disposable_kubeconfig_opts() -> std::io::Result<(
    xianyun_executor::exec::ExecStreamedOpts,
    TempKubeconfigGuard,
)> {
    let pid = std::process::id();
    let nanos = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_nanos())
        .unwrap_or_default();
    let path = std::env::temp_dir().join(format!("caproni-kubeconfig-{pid}-{nanos}"));

    let mut file = std::fs::File::create(&path)?;
    file.write_all(b"")?;

    let mut env: Vec<String> = std::env::vars().map(|(k, v)| format!("{k}={v}")).collect();
    env.push(format!("KUBECONFIG={}", path.display()));

    Ok((
        xianyun_executor::exec::ExecStreamedOpts {
            use_scrolling_area: true,
            env: Some(env),
            ..Default::default()
        },
        TempKubeconfigGuard(path),
    ))
}

fn status_impl(driver: &Driver) -> Result<Status, DriverError> {
    if !is_k3d_available(driver) {
        return Err(DriverError::DriverNotFound("k3d".to_string()));
    }

    let mut status = Status {
        driver: DRIVER_NAME.to_string(),
        state: Some(State::Stopped),
        kube_context: format!("k3d-{}", driver.profile_name),
        ..Default::default()
    };

    let args = vec![
        "cluster".to_string(),
        "list".to_string(),
        driver.profile_name.clone(),
        "-o".to_string(),
        "json".to_string(),
    ];

    let stdout = match driver.runner.read_stdout("k3d", &args) {
        Ok(out) => out,
        Err(_) => return Ok(status),
    };

    let clusters: Vec<K3dCluster> = serde_json::from_str(&stdout)
        .map_err(|e| DriverError::Other(format!("parsing k3d cluster list JSON: {e}")))?;

    let Some(target) = clusters.iter().find(|c| c.name == driver.profile_name) else {
        return Ok(status);
    };

    status.exists = true;
    status.kubernetes_version = node_version(&target.nodes);
    status.ip_address = node_ip(&target.nodes);
    status.host_ip = host_ip(std::env::consts::OS, &status.ip_address);

    status.state = if target.servers_running == 0 {
        Some(State::Stopped)
    } else if target.servers_running == target.servers_count {
        Some(State::Ready)
    } else {
        Some(State::NotReady)
    };

    Ok(status)
}

/// Port of `k3d.hostIP`.
#[must_use]
pub fn host_ip(goos: &str, ip: &str) -> String {
    if goos == "darwin" {
        "127.0.0.1".to_string()
    } else {
        ip.to_string()
    }
}

/// Port of `(*k3d.Driver).isK3dAvailable`.
pub fn is_k3d_available(driver: &Driver) -> bool {
    driver.runner.look_path("k3d").is_ok()
}

/// Port of `(*k3d.Driver).verifyConfig`.
pub fn verify_config(status: &Status, _cfg: &ClusterConfig) -> Result<(), DriverError> {
    if status.driver != DRIVER_NAME {
        return Err(DriverError::ConfigMismatch);
    }

    Ok(())
}

/// Port of `k3d.nodeVersion`.
#[must_use]
pub fn node_version(nodes: &[K3dNode]) -> String {
    for node in nodes {
        if node.role == "server" {
            return node
                .runtime_labels
                .get("k3d.version")
                .cloned()
                .unwrap_or_default();
        }
    }

    String::new()
}

/// Port of `k3d.nodeIP`.
#[must_use]
pub fn node_ip(nodes: &[K3dNode]) -> String {
    let mut server_ip = String::new();

    for node in nodes {
        if node.role == "loadbalancer" && !node.ip.ip.is_empty() {
            return node.ip.ip.clone();
        }
        if node.role == "server" && !node.ip.ip.is_empty() && server_ip.is_empty() {
            server_ip = node.ip.ip.clone();
        }
    }

    server_ip
}

/// Matches an `extra_args` entry that turns off a packaged k3s component,
/// in either form k3d accepts: joined onto `--k3s-arg`, or as the token
/// that follows it. Port of `k3d.deprecatedDisableArg`.
fn is_deprecated_disable_arg(arg: &str) -> bool {
    if let Some(rest) = arg.strip_prefix("--disable") {
        return rest.starts_with('=') || rest.starts_with(' ');
    }
    if let Some(idx) = arg.find("=--disable") {
        let rest = &arg[idx + "=--disable".len()..];
        return rest.starts_with('=') || rest.starts_with(' ');
    }
    false
}

/// Port of `k3d.warnDeprecatedDisableArgs`.
pub fn warn_deprecated_disable_args(extra_args: &[String]) {
    for arg in extra_args {
        if !is_deprecated_disable_arg(arg) {
            continue;
        }

        tracing::warn!(
            extra_arg = %arg,
            "disabling k3s components through extra_args is deprecated and is invisible to \
            caproni; declare it under cluster.k3d.k3s.disable instead"
        );
    }
}

/// Port of `(*k3d.Driver).buildClusterCreateArgs`.
pub fn build_cluster_create_args(
    driver: &Driver,
    cfg: &ClusterConfig,
    port_mappings: &[PortMapping],
) -> Result<Vec<String>, DriverError> {
    let k3d_cfg = cfg.k3d.as_ref().ok_or(DriverError::MissingDriverConfig)?;

    let mut args = vec![
        "cluster".to_string(),
        "create".to_string(),
        driver.profile_name.clone(),
        "--k3s-arg".to_string(),
        "--kubelet-arg=feature-gates=KubeletInUserNamespace=true@server:*".to_string(),
        "--kubeconfig-update-default=false".to_string(),
    ];

    if k3d_cfg.agents > 0 {
        args.push("--agents".to_string());
        args.push(k3d_cfg.agents.to_string());
    }

    if k3d_cfg.servers > 0 && k3d_cfg.servers != 1 {
        args.push("--servers".to_string());
        args.push(k3d_cfg.servers.to_string());
    }

    if !k3d_cfg.image.is_empty() {
        args.push("--image".to_string());
        args.push(k3d_cfg.image.clone());
    }

    for mapping in port_mappings {
        args.push("-p".to_string());
        args.push(mapping.as_spec().to_string());
    }

    if let Some(mirror) = &cfg.registry_mirror {
        let path = write_registries_yaml(&driver.profile_name, mirror)
            .map_err(|e| DriverError::Other(format!("writing k3d registries.yaml: {e}")))?;
        args.push("--registry-config".to_string());
        args.push(path);
    }

    if let Some(network) = &cfg.network
        && !network.cluster_domain.is_empty()
    {
        args.push("--k3s-arg".to_string());
        args.push(format!(
            "--cluster-domain={}@server:*",
            network.cluster_domain
        ));
    }

    if let Some(k3s) = &k3d_cfg.k3s {
        for component in &k3s.disable {
            args.push("--k3s-arg".to_string());
            args.push(format!("--disable={component}@server:*"));
        }
    }

    args.extend(k3d_cfg.extra_args.iter().cloned());

    Ok(args)
}

/// Port of `k3d.writeRegistriesYAML`. Resolves the registries.yaml path
/// directly under `$HOME/.local/share/caproni/<profile>` rather than
/// through the out-of-scope `xianyun-executor::datadir` module's
/// ambient-HOME entry point (see `colima::certs_dir`'s identical
/// deviation).
fn write_registries_yaml(
    profile: &str,
    mirror: &xianyun_core::config::types::RegistryMirrorConfig,
) -> std::io::Result<String> {
    let home = std::env::var("HOME")
        .map_err(|e| std::io::Error::new(std::io::ErrorKind::NotFound, e.to_string()))?;

    let dir = std::path::PathBuf::from(home)
        .join(".local/share/caproni")
        .join(profile);
    std::fs::create_dir_all(&dir)?;

    let path = dir.join("k3d-registries.yaml");

    let endpoint = mirror.host_port();
    let endpoint_url = format!("http://{endpoint}");

    let mut mirrors = serde_yaml::Mapping::new();
    for upstream in &mirror.upstreams {
        let mut entry = serde_yaml::Mapping::new();
        entry.insert(
            serde_yaml::Value::String("endpoint".to_string()),
            serde_yaml::Value::Sequence(vec![serde_yaml::Value::String(endpoint_url.clone())]),
        );
        mirrors.insert(
            serde_yaml::Value::String(upstream.clone()),
            serde_yaml::Value::Mapping(entry),
        );
    }

    let mut tls = serde_yaml::Mapping::new();
    tls.insert(
        serde_yaml::Value::String("insecureSkipVerify".to_string()),
        serde_yaml::Value::Bool(true),
    );
    let mut config_entry = serde_yaml::Mapping::new();
    config_entry.insert(
        serde_yaml::Value::String("tls".to_string()),
        serde_yaml::Value::Mapping(tls),
    );
    let mut configs = serde_yaml::Mapping::new();
    configs.insert(
        serde_yaml::Value::String(endpoint),
        serde_yaml::Value::Mapping(config_entry),
    );

    let mut root = serde_yaml::Mapping::new();
    root.insert(
        serde_yaml::Value::String("mirrors".to_string()),
        serde_yaml::Value::Mapping(mirrors),
    );
    root.insert(
        serde_yaml::Value::String("configs".to_string()),
        serde_yaml::Value::Mapping(configs),
    );

    let body = serde_yaml::to_string(&serde_yaml::Value::Mapping(root))
        .map_err(|e| std::io::Error::other(e.to_string()))?;

    std::fs::write(&path, body)?;

    Ok(path.to_string_lossy().into_owned())
}

/// Port of `k3d.k3dCluster`, the JSON shape of `k3d cluster list -o json`.
#[derive(Debug, Clone, Default, serde::Deserialize)]
#[serde(rename_all = "camelCase", default)]
pub struct K3dCluster {
    pub name: String,
    pub servers_count: i32,
    pub servers_running: i32,
    pub agents_count: i32,
    pub agents_running: i32,
    pub nodes: Vec<K3dNode>,
}

/// Port of `k3d.k3dNode`.
#[derive(Debug, Clone, Default, serde::Deserialize)]
#[serde(default)]
pub struct K3dNode {
    pub name: String,
    pub role: String,
    #[serde(rename = "IP")]
    pub ip: K3dNodeIp,
    #[serde(rename = "State")]
    pub state: K3dNodeState,
    #[serde(rename = "runtimeLabels")]
    pub runtime_labels: std::collections::HashMap<String, String>,
}

/// Port of `k3d.k3dNodeIP`.
#[derive(Debug, Clone, Default, serde::Deserialize)]
pub struct K3dNodeIp {
    #[serde(rename = "IP", default)]
    pub ip: String,
    #[serde(rename = "Static", default)]
    pub static_: bool,
}

/// Port of `k3d.k3dNodeState`.
#[derive(Debug, Clone, Default, serde::Deserialize)]
pub struct K3dNodeState {
    #[serde(rename = "Running", default)]
    pub running: bool,
    #[serde(rename = "Status", default)]
    pub status: String,
}

/// Port of `k3d.portMapping`: a single `k3d cluster create -p` spec, parsed
/// once when it enters the driver and passed around in that form.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct PortMapping {
    pub raw: String,
    pub host_ip: String,
    pub host_port: String,
    pub container_port: String,
    pub tcp: bool,
}

impl PortMapping {
    /// Port of `(portMapping).String`.
    #[must_use]
    pub fn as_spec(&self) -> &str {
        &self.raw
    }

    /// Port of `(portMapping).hostAddress`.
    #[must_use]
    pub fn host_address(&self) -> Option<String> {
        if self.host_port.is_empty() || !self.tcp {
            return None;
        }

        let host = if self.host_ip.is_empty() || self.host_ip == "0.0.0.0" {
            "127.0.0.1"
        } else {
            &self.host_ip
        };

        Some(join_host_port(host, &self.host_port))
    }

    /// Port of `(portMapping).tcpContainerPort`.
    #[must_use]
    pub fn tcp_container_port(&self) -> Option<&str> {
        if !self.tcp {
            return None;
        }

        Some(&self.container_port)
    }
}

/// Port of `net.JoinHostPort` for the plain (non-IPv6-bracket-needing)
/// hosts this driver ever probes -- every mapping with a bracketed IPv6
/// host fails to parse upstream (see [`parse_port_mapping`]) and is never
/// probed in the first place.
fn join_host_port(host: &str, port: &str) -> String {
    format!("{host}:{port}")
}

/// Port of `k3d.parsePortMappings`.
#[must_use]
pub fn parse_port_mappings(mappings: &[String]) -> Vec<PortMapping> {
    mappings.iter().map(|m| parse_port_mapping(m)).collect()
}

/// Port of `k3d.parsePortMapping`.
#[must_use]
pub fn parse_port_mapping(mapping: &str) -> PortMapping {
    let parsed = PortMapping {
        raw: mapping.to_string(),
        ..Default::default()
    };

    let spec = mapping.split('@').next().unwrap_or(mapping);
    let (spec, protocol, has_protocol) = match spec.split_once('/') {
        Some((spec, protocol)) => (spec, protocol, true),
        None => (spec, "", false),
    };

    let fields: Vec<&str> = spec.split(':').collect();

    let (host_ip, host_port, container_port) = match fields.as_slice() {
        [container_port] => (String::new(), String::new(), (*container_port).to_string()),
        [host_port, container_port] => (
            String::new(),
            (*host_port).to_string(),
            (*container_port).to_string(),
        ),
        [host_ip, host_port, container_port] => (
            (*host_ip).to_string(),
            (*host_port).to_string(),
            (*container_port).to_string(),
        ),
        _ => return parsed,
    };

    if !is_port_number(&container_port) {
        return parsed;
    }

    if !host_port.is_empty() && !is_port_number(&host_port) {
        return parsed;
    }

    PortMapping {
        raw: mapping.to_string(),
        host_ip,
        host_port,
        container_port,
        tcp: !has_protocol || protocol == "tcp",
    }
}

fn is_port_number(s: &str) -> bool {
    match s.parse::<u32>() {
        Ok(port) => port > 0 && port <= u32::from(u16::MAX),
        Err(_) => false,
    }
}

/// Port of `k3d.mergePortMappings`.
#[must_use]
pub fn merge_port_mappings(user_mappings: &[String]) -> Vec<PortMapping> {
    let defaults = parse_port_mappings(&[
        "80:80@loadbalancer".to_string(),
        "443:443@loadbalancer".to_string(),
    ]);
    let user = parse_port_mappings(user_mappings);

    let overridden: std::collections::HashSet<String> = user
        .iter()
        .filter_map(|m| m.tcp_container_port().map(str::to_string))
        .collect();

    let mut result = Vec::with_capacity(defaults.len() + user.len());

    for def in defaults {
        let is_overridden = def
            .tcp_container_port()
            .is_some_and(|p| overridden.contains(p));
        if !is_overridden {
            result.push(def);
        }
    }

    result.extend(user);
    result
}

/// Port of `(*k3d.Driver).checkPortConflicts`.
pub fn check_port_conflicts(driver: &Driver, mappings: &[PortMapping]) -> Result<(), DriverError> {
    let mut conflicts = Vec::new();

    for mapping in mappings {
        let Some(address) = mapping.host_address() else {
            continue;
        };

        if !(driver.port_in_use)(&address) {
            continue;
        }

        tracing::debug!(mapping = mapping.as_spec(), address = %address, "host port already has a listener");

        conflicts.push(address);
    }

    if conflicts.is_empty() {
        return Ok(());
    }

    Err(DriverError::PortInUse(format!(
        "{}; stop whatever is listening there, or publish the cluster on other host ports with \
        cluster.k3d.port_mappings",
        conflicts.join(", ")
    )))
}

/// Port of `k3d.probeAddress` (the seam `Driver.portInUse` is stubbed by
/// tests): dials `address` over TCP and reports whether a listener
/// answered.
#[must_use]
pub fn probe_address(address: &str) -> bool {
    TcpStream::connect(address).is_ok()
}