isothrum-runner 1.0.0

//! The `compose` driver: Docker Compose projects, named by the manifest's identity. Its five
//! actions run the native commands run-manifest-v0.md §6 fixes; its status is the manifest's
//! `lifecycle.status` (`ps --all --format json`) matched to the manifest's machines.

mod actions;
mod config;
mod pins;
mod ps;

use std::collections::BTreeMap;
use std::path::PathBuf;
use std::time::Duration;

use isoloom_manifest::{Action, ComposeIdentity, SnakeName, Step, Template, access};

use super::{Driver, Observed};
use crate::Result;
use crate::error::Error;
use crate::executor::{Cx, Echo, call_ok};
use crate::process::ProcessSpec;
use crate::state::Op;
use crate::status::ShellCommand;
use crate::store::write_private;

/// Every machine of `manifest` as `ps --all --format json`'s output `out` shows it, and the host
/// port each published machine port got: a project on this computer's Docker, or one Compose runs
/// inside a VM (`docker-vm`'s guest).
pub(crate) fn read_ps(manifest: &isoloom_manifest::Manifest, out: &str) -> (Vec<crate::status::MachineStatus>, BTreeMap<crate::vars::PortKey, u16>) {
    ps::machines(manifest, &ps::parse_ps(out))
}

/// What an empty pinned-ports file holds: Compose reads it as no override.
const NO_PINS: &str = "services: {}\n";

/// How long one of an action's own Docker calls may take.
const CALL_TIMEOUT: Duration = Duration::from_secs(120);

/// How long hot-adding the access box may take: its image may still need pulling.
const ATTACH_TIMEOUT: Duration = Duration::from_secs(900);

/// The `compose` driver of one manifest.
#[derive(Debug, Clone)]
pub(crate) struct Compose {
    identity: ComposeIdentity,
}

impl Compose {
    pub(crate) fn new(identity: ComposeIdentity) -> Compose {
        Compose { identity }
    }

    /// The Compose project (`${run.name}`, filled).
    fn project(&self, cx: &Cx<'_>) -> Result<String> {
        cx.vars.render(&self.identity.project, "identity.compose.project")
    }

    /// The pinned-ports file, absolute, when the identity names one.
    fn pins(&self, cx: &Cx<'_>) -> Result<Option<PathBuf>> {
        let Some(file) = &self.identity.pinned_ports_file else {
            return Ok(None);
        };
        let path = cx.vars.render(file.template(), "identity.compose.pinned_ports_file")?;
        Ok(Some(cx.run.dir().join(path)))
    }

    /// `docker compose --project-name P --file F… <rest>`'s arguments.
    fn compose_args(&self, cx: &Cx<'_>, rest: &[&str]) -> Result<Vec<String>> {
        let mut args = vec!["compose".to_owned(), "--project-name".into(), self.project(cx)?];
        for file in &self.identity.files {
            args.push("--file".into());
            args.push(cx.vars.render(file.template(), "identity.compose.files")?);
        }
        args.extend(rest.iter().map(|s| (*s).to_owned()));
        Ok(args)
    }

    /// Runs `docker compose … <rest>` for `step` with `env`, returning its standard output.
    async fn compose(&self, cx: &Cx<'_>, step: &Step, rest: &[&str], env: &[(String, String)]) -> Result<String> {
        let args = self.compose_args(cx, rest)?;
        docker(cx, step, args, env).await
    }
}

/// Runs `docker <args>` in the manifest's folder for `step`, returning its standard output.
pub(super) async fn docker(cx: &Cx<'_>, step: &Step, args: Vec<String>, env: &[(String, String)]) -> Result<String> {
    let spec = ProcessSpec {
        program: "docker".into(),
        args,
        cwd: Some(cx.run.dir().to_path_buf()),
        env: env.to_vec(),
        stdin: None,
    };
    call_ok(cx, step.id.as_str(), &spec, CALL_TIMEOUT, Echo::Quiet).await
}

impl Driver for Compose {
    /// A Compose run has nothing to check before it starts (its tools are probed).
    fn preflight(&self, _cx: &Cx<'_>) -> impl Future<Output = Result<()>> {
        std::future::ready(Ok(()))
    }

    fn env(&self, _cx: &Cx<'_>) -> Result<Vec<(String, String)>> {
        Ok(Vec::new())
    }

    fn prepare(&self, cx: &Cx<'_>) -> Result<()> {
        // Every Compose call names the pinned-ports file: it must exist before the first.
        if let Some(pins) = self.pins(cx)?
            && !pins.exists()
        {
            write_private(&pins, NO_PINS.as_bytes()).map_err(Error::io(&pins, "written"))?;
        }
        Ok(())
    }

    async fn action(&self, cx: &Cx<'_>, step: &Step, action: Action, with: &BTreeMap<SnakeName, Template>) -> Result<()> {
        match action {
            Action::ComposePinPorts => actions::pin_ports(self, cx, step).await,
            Action::ComposeCheckPortsFree => actions::check_ports_free(self, cx, step).await,
            Action::ComposeRemoveStaleNetworks => actions::remove_stale_networks(self, cx, step).await,
            Action::ComposeRemoveContainers => actions::remove_containers(self, cx, step, with).await,
            Action::ComposeRemoveNetworks => actions::remove_networks(self, cx, step).await,
            other => Err(super::foreign_action("Compose", step, other)),
        }
    }

    async fn observe(&self, cx: &Cx<'_>) -> Result<Observed> {
        let out = super::status_report(cx).await?;
        let (machines, published) = ps::machines(cx.run.manifest(), &ps::parse_ps(&out));
        // Compose publishes on this computer.
        let hosts = published.keys().map(|(machine, _)| (machine.clone(), "127.0.0.1".to_owned())).collect();
        Ok(Observed {
            machines,
            published,
            hosts,
            ..Observed::default()
        })
    }

    fn after_down(&self, cx: &Cx<'_>) -> Result<()> {
        // The next start is a new run: it picks its ports again.
        if let Some(pins) = self.pins(cx)? {
            write_private(&pins, NO_PINS.as_bytes()).map_err(Error::io(&pins, "written"))?;
        }
        Ok(())
    }

    /// Brings the access service up in its profile, so the initial `up` left it down:
    /// `docker compose --profile access up -d --no-deps <service>`, with the picked image in
    /// [`access::IMAGE_ENV`] (the service's `${VAR:-default}` keeps its default otherwise). When a
    /// network is named, the started container is also plugged into it, best effort.
    async fn attach(&self, cx: &Cx<'_>, native: &str, image: Option<&str>, network: Option<&str>) -> Result<()> {
        let mut env = Vec::new();
        if let Some(image) = image {
            env.push((access::IMAGE_ENV.to_owned(), image.to_owned()));
        }
        let args = self.compose_args(cx, &["--profile", access::PROFILE, "up", "-d", "--no-deps", native])?;
        // The box's image may still need pulling, so this gets the long boot budget, not a call's.
        call_ok(cx, Op::Attach.as_str(), &docker_spec(cx, args, env), ATTACH_TIMEOUT, Echo::Shown).await?;
        if let Some(network) = network {
            self.connect_network(cx, native, network).await;
        }
        Ok(())
    }

    /// Stops and removes the access service, leaving the rest of the run up:
    /// `docker compose --profile access rm -sf <service>`.
    async fn detach(&self, cx: &Cx<'_>, native: &str) -> Result<()> {
        let args = self.compose_args(cx, &["--profile", access::PROFILE, "rm", "-sf", native])?;
        call_ok(cx, Op::Detach.as_str(), &docker_spec(cx, args, Vec::new()), CALL_TIMEOUT, Echo::Shown).await?;
        Ok(())
    }

    /// `docker compose … exec <service> bash`: Compose addresses the box by service, so no
    /// container name is reconstructed. `exec` is interactive with a TTY by default.
    fn shell_command(&self, cx: &Cx<'_>, native: &str) -> Result<ShellCommand> {
        Ok(ShellCommand {
            program: "docker".into(),
            args: self.compose_args(cx, &["exec", native, "bash"])?,
            cwd: Some(cx.run.dir().to_path_buf()),
        })
    }
}

impl Compose {
    /// Plugs the just-started access container into the lab network `network` (its project
    /// network), best effort: a box already on it, or a network it can't name, is left as it is.
    async fn connect_network(&self, cx: &Cx<'_>, native: &str, network: &str) {
        let Ok(project) = self.project(cx) else { return };
        let Ok(id_args) = self.compose_args(cx, &["ps", "-q", native]) else { return };
        let Ok(out) = call_ok(cx, Op::Attach.as_str(), &docker_spec(cx, id_args, Vec::new()), CALL_TIMEOUT, Echo::Quiet).await else {
            return;
        };
        let Some(container) = out.lines().map(str::trim).find(|s| !s.is_empty()) else {
            return;
        };
        let net = format!("{project}_{network}");
        let connect = vec!["network".into(), "connect".into(), net, container.to_owned()];
        // Already-connected is a non-zero exit we deliberately ignore: the box is where we want it.
        if let Err(e) = call_ok(cx, Op::Attach.as_str(), &docker_spec(cx, connect, Vec::new()), CALL_TIMEOUT, Echo::Quiet).await {
            tracing::debug!(%e, "access box already on the network, or it can't be joined");
        }
    }
}

/// A `docker <args>` spec in the run's folder, with `env`.
fn docker_spec(cx: &Cx<'_>, args: Vec<String>, env: Vec<(String, String)>) -> ProcessSpec {
    ProcessSpec {
        program: "docker".into(),
        args,
        cwd: Some(cx.run.dir().to_path_buf()),
        env,
        stdin: None,
    }
}