zc2 0.0.29

//! `ContainerRuntime`: the reconciler's only door to Docker, so
//! `reconcile::drive` can be driven in tests by a fake instead of the real
//! CLI. `DockerCli` shells out the same way `mesh.rs`'s `docker()` helper
//! does (Command::new("docker") + captured stdout/stderr), extended with the
//! handful of subcommands a reconciler needs that a static local mesh never
//! did: pull, per-container run/stop/rm, health inspection, and label-based
//! listing.
//!
//! Networking (post-drill fix): the broker driving this reconciler is
//! usually ITSELF a container (mounted `/var/run/docker.sock`, talking to
//! the HOST's daemon) — a live drill against the dev stack found that
//! publishing container ports to the docker HOST's `127.0.0.1` (the
//! original design) is unreachable from the broker's own network namespace;
//! only a host-native `zc` shares that namespace. So `run` no longer
//! publishes any ports at all — it joins `ZAKURO_DEPLOY_NETWORK` (or
//! `bridge`) instead, and `container_ip` resolves the environment's own
//! address on that network, which any container on the same network
//! (including this one, once it's also joined to it) can reach directly.

use std::collections::HashMap;
use std::process::Command;
use std::time::Duration;

use super::{Healthcheck, Port};

/// Everything needed to `docker run` one deployment version.
pub struct RunSpec<'a> {
    pub name: String,
    pub image: &'a str,
    pub cmd: Option<&'a str>,
    pub ports: &'a [Port],
    pub env: &'a HashMap<String, String>,
    /// `(key, value)` pairs applied as `--label key=value`.
    pub labels: Vec<(String, String)>,
}

/// The docker network new environment containers join. `ZAKURO_DEPLOY_NETWORK`
/// when set and non-blank (e.g. `hubnet`, so the broker and every environment
/// it starts share a network the broker itself was also joined to at deploy
/// time), else docker's own default `bridge`.
pub fn deploy_network() -> String {
    std::env::var("ZAKURO_DEPLOY_NETWORK")
        .ok()
        .map(|s| s.trim().to_string())
        .filter(|s| !s.is_empty())
        .unwrap_or_else(|| "bridge".to_string())
}

/// Abstraction over "a place that can run OCI containers", so the reconciler
/// logic in `reconcile::drive` is testable without Docker installed and
/// without touching the network.
pub trait ContainerRuntime {
    /// Whether this runtime is usable at all on this host (e.g. `docker`
    /// resolves on PATH and the daemon answers). Checked once per tick;
    /// `deploy::tick` reports every running-desired deployment `failed` and
    /// does nothing else when this is `false`.
    fn available(&self) -> bool;
    fn pull(&self, image: &str) -> Result<(), String>;
    /// Starts a fresh container per `spec`, returning its id. Idempotent
    /// w.r.t. the container *name*: a stale container from a crashed prior
    /// attempt with the same name is removed first, since docker refuses to
    /// reuse a name that's still allocated. Joins `deploy_network()` — no
    /// port publishing (see this module's doc comment).
    fn run(&self, spec: &RunSpec) -> Result<String, String>;
    fn stop(&self, container_id: &str) -> Result<(), String>;
    fn rm(&self, container_id: &str) -> Result<(), String>;
    /// `docker start` on an already-created (but currently stopped)
    /// container — used to bring a rollback's surviving previous container
    /// back up if it isn't running for some unrelated reason.
    fn restart(&self, container_id: &str) -> Result<(), String>;
    fn is_running(&self, container_id: &str) -> bool;
    /// The container's own IP address on its docker network (first
    /// non-empty entry across `NetworkSettings.Networks`) — the address the
    /// broker actually proxies/health-checks against, NOT a host-published
    /// port (see this module's doc comment for why). A real implementation
    /// retries a few times: docker assigns this address a moment after
    /// `run` returns, not synchronously with it. `Err`/empty when the
    /// container has no network IP at all (e.g. `--network host`), which
    /// callers fall back on treating as "reachable on this host's own
    /// loopback" — the container isn't on its own network in that mode.
    fn container_ip(&self, container_id: &str) -> Result<String, String>;
    /// One readiness probe against `container_id` per `hc` (TCP connect, or
    /// an HTTP GET expecting 2xx — see `health_http_url`). Split out as its
    /// own trait method, rather than composed from `container_ip` + a raw
    /// socket/HTTP call inline in `reconcile::wait_healthy`, specifically so
    /// a fake runtime in tests can answer it directly instead of needing a
    /// real container or a real socket to poll.
    fn health_check(&self, container_id: &str, hc: &Healthcheck) -> bool;
    /// Last `lines` lines of the container's logs, best-effort (empty string
    /// on any failure) — used to give the hub something more useful than
    /// "exited" in a `failed` status report.
    fn logs_tail(&self, container_id: &str, lines: u32) -> String;
    /// Container ids currently labelled `zakuro.deployment=<id>` (any
    /// version), for reconciling local state against reality after a broker
    /// restart.
    fn list_by_label(&self, id: &str) -> Vec<String>;
}

/// The container name this reconciler gives a deployment version. Unique per
/// (deployment, version) so a version bump never collides with the container
/// it's replacing while both are briefly alive during the health check.
pub fn container_name(id: &str, version: u64) -> String {
    format!("zk-dep-{id}-v{version}")
}

/// Labels stamped on every container this reconciler creates, so
/// `list_by_label`/`docker ps --filter` can find them again.
pub fn deployment_labels(id: &str, version: u64) -> Vec<(String, String)> {
    vec![
        ("zakuro.deployment".to_string(), id.to_string()),
        ("zakuro.version".to_string(), version.to_string()),
    ]
}

/// The URL an `"http"` healthcheck (or the `/serve` proxy) targets on the
/// container's own address. Split out from `DockerCli::health_check` so the
/// path-joining logic (default `"/"`, tolerate a path with or without a
/// leading slash) is unit-testable without a real container or socket.
/// `host` is a bare IP or hostname — no scheme, no port.
pub fn health_http_url(host: &str, hc: &Healthcheck) -> String {
    let path = hc.path.as_deref().unwrap_or("/");
    let path = if path.starts_with('/') {
        path.to_string()
    } else {
        format!("/{path}")
    };
    format!("http://{host}:{}{path}", hc.port)
}

/// `docker` CLI-backed `ContainerRuntime`.
pub struct DockerCli;

/// Whether THIS host can run containers at all — `DockerCli::available()` for
/// callers that hold no runtime of their own.
///
/// The register path (`node_sync::register_node`) reports this to the hub so
/// broker eligibility for a deployment is judged on it. It stays a plain
/// probe rather than a cached value because nothing else computes one: the
/// reconciler's own answer lives inside `tick`, which returns before asking
/// whenever the hub has no desired deployments for this node — precisely the
/// broker the hub still needs to judge. One `docker info` per call, so keep
/// it on the ~60s node-sync cadence in `server.rs`'s health-tick loop, next
/// to the roster refresh's own round-trips, and off anything hotter.
pub fn runtime_available() -> bool {
    DockerCli.available()
}

/// Run `docker <args>`, returning trimmed stdout or the trimmed stderr as an
/// error. Same shape as `mesh.rs::docker()`, kept separate rather than
/// shared: `mesh.rs` is the local static-mesh bring-up path, this is the
/// hub-driven reconciler, and coupling them would make either harder to
/// change independently.
fn docker(args: &[&str]) -> Result<String, String> {
    let out = Command::new("docker")
        .args(args)
        .output()
        .map_err(|e| format!("docker not available: {e}"))?;
    if out.status.success() {
        Ok(String::from_utf8_lossy(&out.stdout).trim().to_string())
    } else {
        let stderr = String::from_utf8_lossy(&out.stderr).trim().to_string();
        Err(if stderr.is_empty() {
            format!("docker {} failed", args.first().copied().unwrap_or(""))
        } else {
            stderr
        })
    }
}

/// How many times `DockerCli::container_ip` retries an empty result, and how
/// long it sleeps between attempts. The IP shows up in
/// `NetworkSettings.Networks` a moment after `docker run` returns — not
/// synchronously with it — so the first inspect or two can legitimately come
/// back empty even though the container is fine.
const CONTAINER_IP_RETRIES: u32 = 10;
const CONTAINER_IP_RETRY_DELAY: Duration = Duration::from_millis(300);

impl ContainerRuntime for DockerCli {
    fn available(&self) -> bool {
        Command::new("docker")
            .arg("info")
            .output()
            .map(|o| o.status.success())
            .unwrap_or(false)
    }

    fn pull(&self, image: &str) -> Result<(), String> {
        docker(&["pull", image]).map(|_| ())
    }

    fn run(&self, spec: &RunSpec) -> Result<String, String> {
        // Best-effort: clear a stale container occupying this exact name
        // (e.g. left over from a crashed prior reconcile attempt at the same
        // version). Docker errors on a name collision otherwise.
        let _ = docker(&["rm", "-f", &spec.name]);

        let network = deploy_network();
        let mut args: Vec<String> = vec![
            "run".into(),
            "-d".into(),
            "--name".into(),
            spec.name.clone(),
            "--network".into(),
            network,
        ];
        for (k, v) in &spec.labels {
            args.push("--label".into());
            args.push(format!("{k}={v}"));
        }
        // No `-p`/port publishing: see this module's doc comment — the
        // broker reaches the environment by its own network IP
        // (`container_ip`), not a host-published port, because the broker
        // driving this is usually itself a container with its own netns.
        for (k, v) in spec.env {
            args.push("-e".into());
            args.push(format!("{k}={v}"));
        }
        args.push(spec.image.to_string());
        if let Some(cmd) = spec.cmd {
            // Split on whitespace: these commands come from the hub's
            // deployment config (`cmd: "python -m http.server 8000"`), not
            // arbitrary shell — no quoting/escaping semantics to preserve.
            args.extend(cmd.split_whitespace().map(|s| s.to_string()));
        }

        let arg_refs: Vec<&str> = args.iter().map(|s| s.as_str()).collect();
        docker(&arg_refs)
    }

    fn stop(&self, container_id: &str) -> Result<(), String> {
        docker(&["stop", container_id]).map(|_| ())
    }

    fn rm(&self, container_id: &str) -> Result<(), String> {
        docker(&["rm", "-f", container_id]).map(|_| ())
    }

    fn restart(&self, container_id: &str) -> Result<(), String> {
        docker(&["start", container_id]).map(|_| ())
    }

    fn is_running(&self, container_id: &str) -> bool {
        docker(&["inspect", "-f", "{{.State.Running}}", container_id])
            .map(|out| out == "true")
            .unwrap_or(false)
    }

    fn container_ip(&self, container_id: &str) -> Result<String, String> {
        for attempt in 0..CONTAINER_IP_RETRIES {
            let out = docker(&[
                "inspect",
                "-f",
                "{{range .NetworkSettings.Networks}}{{.IPAddress}} {{end}}",
                container_id,
            ])?;
            // First non-empty token: a container can be on more than one
            // network (rare here, but the format string doesn't rule it
            // out), and an entry with no assigned address yet renders as an
            // empty string between the spaces rather than being omitted.
            if let Some(ip) = out.split_whitespace().find(|s| !s.is_empty()) {
                return Ok(ip.to_string());
            }
            if attempt + 1 < CONTAINER_IP_RETRIES {
                std::thread::sleep(CONTAINER_IP_RETRY_DELAY);
            }
        }
        Err(format!(
            "container {container_id} has no network IP after {CONTAINER_IP_RETRIES} attempts \
             (host network mode, or the container exited)"
        ))
    }

    fn health_check(&self, container_id: &str, hc: &Healthcheck) -> bool {
        // `--network host` (or any other mode docker gives no per-container
        // IP for) has nothing of its own to address: the environment's
        // process is reachable exactly where it bound, on THIS docker
        // host's own loopback — which is only actually correct when this
        // broker shares that host's netns (host-native `zc`, or a broker
        // container itself run with `--network host`). Anything more
        // specific than that fallback would need the hub to tell us which
        // mode a deployment wants, which isn't part of the contract today.
        let host = self
            .container_ip(container_id)
            .unwrap_or_else(|_| "127.0.0.1".to_string());
        match hc.kind.as_str() {
            "tcp" => std::net::TcpStream::connect((host.as_str(), hc.port)).is_ok(),
            _ => ureq::get(health_http_url(&host, hc))
                .config()
                .timeout_global(Some(Duration::from_secs(3)))
                .http_status_as_error(false)
                .build()
                .call()
                .map(|r| (200..300).contains(&r.status().as_u16()))
                .unwrap_or(false),
        }
    }

    fn logs_tail(&self, container_id: &str, lines: u32) -> String {
        docker(&["logs", "--tail", &lines.to_string(), container_id]).unwrap_or_default()
    }

    fn list_by_label(&self, id: &str) -> Vec<String> {
        docker(&[
            "ps",
            "-a",
            "-q",
            "--filter",
            &format!("label=zakuro.deployment={id}"),
        ])
        .map(|out| out.lines().map(|l| l.trim().to_string()).collect())
        .unwrap_or_default()
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn container_name_is_unique_per_version() {
        assert_eq!(container_name("dep_1", 3), "zk-dep-dep_1-v3");
        assert_ne!(container_name("dep_1", 3), container_name("dep_1", 4));
    }

    #[test]
    fn deployment_labels_carry_id_and_version() {
        let labels = deployment_labels("dep_1", 3);
        assert!(labels.contains(&("zakuro.deployment".to_string(), "dep_1".to_string())));
        assert!(labels.contains(&("zakuro.version".to_string(), "3".to_string())));
    }

    fn hc(kind: &str, path: Option<&str>) -> Healthcheck {
        Healthcheck {
            kind: kind.to_string(),
            port: 8000,
            path: path.map(|s| s.to_string()),
            timeout_s: 60,
        }
    }

    #[test]
    fn health_http_url_defaults_path_to_root() {
        assert_eq!(
            health_http_url("172.17.0.5", &hc("http", None)),
            "http://172.17.0.5:8000/"
        );
    }

    #[test]
    fn health_http_url_tolerates_a_path_without_leading_slash() {
        assert_eq!(
            health_http_url("172.17.0.5", &hc("http", Some("healthz"))),
            "http://172.17.0.5:8000/healthz"
        );
    }

    #[test]
    fn health_http_url_keeps_an_already_leading_slash() {
        assert_eq!(
            health_http_url("172.17.0.5", &hc("http", Some("/api/health"))),
            "http://172.17.0.5:8000/api/health"
        );
    }

    #[test]
    fn health_http_url_uses_the_container_ip_not_loopback() {
        // The whole point of the drill fix: this must never say 127.0.0.1
        // when a real container IP is available.
        let url = health_http_url("10.13.13.10", &hc("http", None));
        assert!(url.starts_with("http://10.13.13.10:"));
        assert!(!url.contains("127.0.0.1"));
    }

    #[test]
    fn deploy_network_defaults_to_bridge_when_unset() {
        let _lock = crate::credentials::HOME_ENV_LOCK.lock();
        let prev = std::env::var_os("ZAKURO_DEPLOY_NETWORK");
        std::env::remove_var("ZAKURO_DEPLOY_NETWORK");
        assert_eq!(deploy_network(), "bridge");
        if let Some(v) = prev {
            std::env::set_var("ZAKURO_DEPLOY_NETWORK", v);
        }
    }

    #[test]
    fn deploy_network_honours_the_env_override() {
        let _lock = crate::credentials::HOME_ENV_LOCK.lock();
        let prev = std::env::var_os("ZAKURO_DEPLOY_NETWORK");
        std::env::set_var("ZAKURO_DEPLOY_NETWORK", "hubnet");
        assert_eq!(deploy_network(), "hubnet");
        match prev {
            Some(v) => std::env::set_var("ZAKURO_DEPLOY_NETWORK", v),
            None => std::env::remove_var("ZAKURO_DEPLOY_NETWORK"),
        }
    }

    #[test]
    fn deploy_network_blank_value_falls_back_to_bridge() {
        let _lock = crate::credentials::HOME_ENV_LOCK.lock();
        let prev = std::env::var_os("ZAKURO_DEPLOY_NETWORK");
        std::env::set_var("ZAKURO_DEPLOY_NETWORK", "   ");
        assert_eq!(deploy_network(), "bridge");
        match prev {
            Some(v) => std::env::set_var("ZAKURO_DEPLOY_NETWORK", v),
            None => std::env::remove_var("ZAKURO_DEPLOY_NETWORK"),
        }
    }
}