main.rs 34.1%

126 of 369 lines run, 14 of 106 functions · run #181 on main · run missed
//! `flow`: check pipelines, run them locally, and run jobs for OxygenHub
//! (docs/spec.md, docs/m0.md).

use flow::{facts, home, images, microvm, runner, syscall, wait, wasm, workspace};

use anyhow::{bail, Context, Result};
use clap::{Parser, Subcommand};
use flow_core::api::Record;
use flow_core::{Pipeline, Runtime, PIPELINE_PATH};
use home::Home;
use std::collections::BTreeSet;
use std::path::{Path, PathBuf};
use std::time::Duration;

#[derive(Parser)]
#[command(name = "flow", version = env!("OXY_VERSION_LONG"), about = "OxygenFlow: hermetic pipelines for Oxygen repositories")]
struct Cli {
    /// The Flow home (default $FLOW_HOME, else ~/.local/share/flow).
    #[arg(long, global = true)]
    home: Option<PathBuf>,
    #[command(subcommand)]
    cmd: Cmd,
}

#[derive(Subcommand)]
enum RunnerCmd {
    /// Register this runner with a hub using a one-time token from its Admin → Runners page
    Register {
        /// The hub, e.g. https://earth.example:7343
        hub: String,
        /// The registration token (read from stdin when not given)
        #[arg(long)]
        token: Option<String>,
        /// The runner's name (default: the host name)
        #[arg(long)]
        name: Option<String>,
        /// The hub's CA certificate, when self-signed
        #[arg(long)]
        ca: Option<PathBuf>,
        /// A label the runner offers (repeatable)
        #[arg(long)]
        label: Vec<String>,
    },
    /// Measure this machine as a runner does: its speed score (100 on the
    /// reference machine) and class, free disk and load (spec §11.1)
    Bench,
}

#[derive(Subcommand)]
enum Cmd {
    /// Validate a pipeline and print its jobs and hash
    Check {
        /// The pipeline file (default: .flow/pipeline.kdl)
        file: Option<PathBuf>,
    },
    /// Print a file's BLAKE3, for pinning in flow.toml
    Hash { file: PathBuf },
    /// Rootfs images for sandboxes
    Rootfs {
        #[command(subcommand)]
        cmd: RootfsCmd,
    },
    /// Run the pipeline on this directory's files, locally
    Run {
        /// Only these jobs (and what they depend on)
        jobs: Vec<String>,
        /// The directory to build (default: the current one)
        #[arg(long, default_value = ".")]
        dir: PathBuf,
        /// Minutes before a job is stopped
        #[arg(long, default_value_t = 60)]
        timeout: u64,
        /// A variable for every job, as the hub's settings would give it;
        /// the pipeline's own `env` wins
        #[arg(long = "var", value_name = "NAME=VALUE")]
        vars: Vec<String>,
        /// A secret a job's `secret` block names, as the hub would supply
        /// it; mounted as a file, never in the environment
        #[arg(long = "secret", value_name = "NAME=VALUE")]
        secrets: Vec<String>,
    },
    /// Claim and run jobs for the hub in flow.toml's [runner] (or the
    /// runner.toml `flow runner register` wrote)
    Runner {
        #[command(subcommand)]
        cmd: Option<RunnerCmd>,
    },
    /// Follow the checks of a pushed state on OxygenHub until they finish:
    /// exit 0 when all passed, 1 when one failed, 2 at the time limit.
    /// Uses $OXY_TOKEN and $OXY_CA as oxy does.
    Wait {
        /// The repository's URL on the hub (https://hub/owner/name)
        url: String,
        /// The state hash the push moved the bookmark to (`oxy push --json` prints it)
        state: String,
        /// Minutes to wait at most
        #[arg(long, default_value_t = 60)]
        timeout: u64,
    },
    /// Show which isolation tiers this machine can run jobs in, and why
    /// the others cannot
    Tiers,
    /// The syscall tier's sandbox helper, started by the runner for each job
    #[command(name = "__sandbox", hide = true)]
    Sandbox { spec: PathBuf },
    /// Check that this machine can run the syscall tier
    #[command(name = "__sandbox-probe", hide = true)]
    SandboxProbe,
    /// The WASM tier's helper, started by the runner for each job
    #[command(name = "__wasm", hide = true)]
    Wasm { spec: PathBuf },
}

#[derive(Subcommand)]
enum RootfsCmd {
    /// Make an ext4 rootfs image from a root file system tarball and store it by hash
    Import {
        tarball: PathBuf,
        /// Refuse the tarball unless its SHA-256 matches (as its publisher states it)
        #[arg(long)]
        sha256: Option<String>,
        /// The image's own environment, NAME=value (repeatable): what a
        /// Dockerfile's ENV sets and `docker export` loses. Jobs start from
        /// it; their sandbox's and their own `env` override it
        #[arg(long = "env", value_name = "NAME=VALUE")]
        env: Vec<String>,
    },
    /// List the images this home holds
    List,
    /// Send an image to a hub, so its runners fetch it instead of having it
    /// copied to them: only the chunks the hub lacks travel. Signs in as
    /// `oh` is signed in to that hub, or with $OXY_TOKEN and $OXY_CA
    Push {
        /// The image's hash, as pipelines pin it (cas://blob/HASH)
        hash: String,
        /// The hub (default: $OH_HUB, else the runner's, from runner.toml or flow.toml)
        #[arg(long)]
        hub: Option<String>,
    },
    /// Fetch an image a hub holds into this home, as a runner does for a
    /// job whose image it lacks. Signs in as `oh` is, or with $OXY_TOKEN
    Pull {
        hash: String,
        #[arg(long)]
        hub: Option<String>,
    },
}

fn main() {
    let cli = Cli::parse();
    if let Err(e) = run(cli) {
        eprintln!("flow: {e:#}");
        std::process::exit(1);
    }
}

fn run(cli: Cli) -> Result<()> {
    match cli.cmd {
        Cmd::Sandbox { spec } => std::process::exit(syscall::sandbox_main(&spec)),
        Cmd::SandboxProbe => std::process::exit(syscall::probe_main()),
        Cmd::Wasm { spec } => std::process::exit(wasm::wasm_main(&spec)),
        Cmd::Check { file } => check(&file.unwrap_or_else(|| PathBuf::from(PIPELINE_PATH))),
        Cmd::Hash { file } => {
            println!("{}", home::hash_file(&file)?);
            Ok(())
        }
        Cmd::Rootfs { cmd } => {
            let home = Home::locate(cli.home.as_deref())?;
            match cmd {
                RootfsCmd::Import { tarball, sha256, env } => rootfs_import(&home, &tarball, sha256.as_deref(), &env),
                RootfsCmd::Push { hash, hub } => {
                    let client = image_client(&home, hub)?;
                    eprintln!("Sending image {} …", &hash[..12.min(hash.len())]);
                    let m = client.push(&home, &hash)?;
                    println!(
                        "Sent image {}: {} chunks the hub lacked, {} in {:.1} s",
                        &hash[..12],
                        m.chunks,
                        images::human(m.bytes),
                        m.elapsed.as_secs_f64()
                    );
                    Ok(())
                }
                RootfsCmd::Pull { hash, hub } => {
                    let client = image_client(&home, hub)?;
                    match client.fetch(&home, &hash)? {
                        Some(m) => {
                            println!(
                                "Fetched image {}: {} chunks, {} in {:.1} s",
                                &hash[..12],
                                m.chunks,
                                images::human(m.bytes),
                                m.elapsed.as_secs_f64()
                            );
                            Ok(())
                        }
                        None => anyhow::bail!("the hub does not hold image {}; send it with flow rootfs push", &hash[..12.min(hash.len())]),
                    }
                }
                RootfsCmd::List => {
                    for e in std::fs::read_dir(home.root.join("cas"))?.flatten() {
                        let size = e.metadata().map(|m| m.len()).unwrap_or(0);
                        if let Some(h) = e.file_name().to_string_lossy().strip_suffix(".ext4") {
                            println!("cas://blob/{h}  {} MiB", size >> 20);
                        }
                    }
                    Ok(())
                }
            }
        }
        Cmd::Run { jobs, dir, timeout, vars, secrets } => {
            let home = Home::locate(cli.home.as_deref())?;
            let pairs = |what: &str, list: &[String]| {
                list.iter()
                    .map(|e| {
                        e.split_once('=')
                            .map(|(k, v)| (k.to_string(), v.to_string()))
                            .ok_or_else(|| anyhow::anyhow!("--{what} {e:?}: expected NAME=VALUE"))
                    })
                    .collect::<Result<Vec<_>>>()
            };
            let (vars, secrets) = (pairs("var", &vars)?, pairs("secret", &secrets)?);
            run_local(&home, &dir, &jobs, Duration::from_secs(timeout * 60), &vars, &secrets)
        }
        Cmd::Runner { cmd: Some(RunnerCmd::Bench) } => {
            let home = Home::locate(cli.home.as_deref())?;
            let (speed, ms) = facts::bench();
            let f = facts::now(&home, Some(speed));
            println!("speed {speed} (class {}), benchmark {ms} ms", flow_core::api::Facts::class_of(speed));
            println!("free disk {} MiB at {}", f.free_disk_mib.unwrap_or(0), home.root.display());
            println!("load {:.2}", f.load_x100.unwrap_or(0) as f64 / 100.0);
            if let Some(k) = f.kernel {
                println!("kernel {k}");
            }
            Ok(())
        }
        Cmd::Runner { cmd: Some(RunnerCmd::Register { hub, token, name, ca, label }) } => {
            let home = Home::locate(cli.home.as_deref())?;
            if std::env::var_os("OXY_HOME").is_none() {
                std::env::set_var("OXY_HOME", home.root.join("oxy"));
            }
            let name = name.unwrap_or_else(|| {
                std::fs::read_to_string("/etc/hostname").map(|h| h.trim().to_string()).unwrap_or_else(|_| "flow".into())
            });
            let token = match token {
                Some(t) => t,
                None => {
                    eprintln!("Paste the registration token from the hub's Admin → Runners, then press Enter:");
                    let mut line = String::new();
                    std::io::stdin().read_line(&mut line)?;
                    line.trim().to_string()
                }
            };
            let got = runner::register(&home, &hub, &token, &name, ca, label)?;
            println!("registered as {} (runner {}); start it with `flow runner`", got.name, got.id);
            Ok(())
        }
        Cmd::Runner { cmd: None } => {
            let home = Home::locate(cli.home.as_deref())?;
            // The runner's oxy state (identity, user config) stays in its home.
            if std::env::var_os("OXY_HOME").is_none() {
                std::env::set_var("OXY_HOME", home.root.join("oxy"));
            }
            runner::serve(&home)
        }
        Cmd::Tiers => {
            let home = Home::locate(cli.home.as_deref())?;
            match runner::tiers(&home) {
                Ok((have, missing)) => {
                    for t in have {
                        println!("{t}: available");
                    }
                    for why in missing {
                        println!("{}", why.replacen(": ", ": unavailable: ", 1));
                    }
                }
                Err(why) => println!("{why}"),
            }
            Ok(())
        }
        Cmd::Wait { url, state, timeout } => {
            let token = std::env::var("OXY_TOKEN").ok();
            let ca = std::env::var_os("OXY_CA").map(PathBuf::from);
            let code =
                wait::wait(&url, &state, token.as_deref(), ca.as_deref(), Duration::from_secs(timeout * 60), Duration::from_secs(15))?;
            std::process::exit(code);
        }
    }
}

/// The hub image commands talk to: `--hub`, else $OH_HUB, else the
/// runner's. The token: $OXY_TOKEN, else the login `oh auth login` (or
/// `oxy auth login`) saved for that hub in Oxygen's credential store
/// (oxygenflow#44), else the runner's own.
fn image_client(home: &Home, hub: Option<String>) -> Result<images::Client> {
    let runner = home.config.runner.clone();
    let hub = hub
        .or_else(|| std::env::var("OH_HUB").ok().filter(|h| !h.is_empty()))
        .or_else(|| runner.as_ref().map(|r| r.hub.clone()))
        .context("name the hub with --hub https://HUB (or set OH_HUB)")?;
    let (saved_token, saved_ca) = oxy_core::credentials::Credentials::user().lookup(&hub);
    let token = std::env::var("OXY_TOKEN")
        .ok()
        .filter(|t| !t.is_empty())
        .or(saved_token)
        .or_else(|| runner.as_ref().map(|r| r.token.clone()))
        .with_context(|| format!("no token for {hub}: sign in with `oh auth login --hub {hub}`, or set OXY_TOKEN"))?;
    let ca = std::env::var_os("OXY_CA").map(PathBuf::from).or(saved_ca).or_else(|| runner.as_ref().and_then(|r| r.ca.clone()));
    images::Client::new(&hub, &token, ca.as_deref())
}

fn load(file: &Path) -> Result<Pipeline> {
    let src = std::fs::read_to_string(file).with_context(|| format!("read {}", file.display()))?;
    Pipeline::parse(&src).map_err(|p| anyhow::anyhow!("{}", not_valid(file, &p.to_string())))
}

/// Why a pipeline did not parse. When it names something this flow does
/// not know (a node, an output, an event), the pipeline may simply be
/// newer than this flow (oxygenflow#43): say so, with this flow's version.
fn not_valid(file: &Path, problems: &str) -> String {
    if problems.contains("unknown ") {
        format!(
            "{} uses something this flow ({}) does not know:\n{problems}\n\
             If the pipeline was written for a newer OxygenFlow, update flow (build it from oxygenflow main); \
             otherwise fix the lines named.",
            file.display(),
            env!("OXY_VERSION_LONG")
        )
    } else {
        format!("{} is not valid:\n{problems}", file.display())
    }
}

fn check(file: &Path) -> Result<()> {
    let p = load(file)?;
    let depths = p.depths();
    for j in p.order() {
        let sb = &p.sandboxes[&j.sandbox];
        let after = if j.depends_on.is_empty() { String::new() } else { format!(" after {}", j.depends_on.join(", ")) };
        println!("{}{}  {} [{}]{after}", "  ".repeat(depths[j.name.as_str()]), j.name, j.exec.join(" "), sb.runtime.as_str());
    }
    println!("pipeline {}", hex::encode(p.hash()));
    Ok(())
}

/// The image environment file flow-init reads (`/etc/flow/env`): one
/// `NAME=value` per line, names checked as a pipeline's are.
fn image_env_file(env: &[String]) -> Result<String> {
    let mut out = String::new();
    for e in env {
        let Some((k, v)) = e.split_once('=') else { bail!("--env {e:?}: expected NAME=value") };
        let ok = !k.is_empty() && !k.starts_with(|c: char| c.is_ascii_digit()) && k.bytes().all(|b| b.is_ascii_alphanumeric() || b == b'_');
        if !ok || v.contains('\n') || v.contains('\0') {
            bail!("--env {e:?}: a name is letters, digits and _, and a value is one line");
        }
        out.push_str(&format!("{k}={v}\n"));
    }
    Ok(out)
}

/// The architecture an ELF file was built for, by its header.
fn elf_arch(path: &Path) -> Option<&'static str> {
    let mut head = [0u8; 20];
    std::io::Read::read_exact(&mut std::fs::File::open(path).ok()?, &mut head).ok()?;
    jet::rootfs::elf_arch(&head)
}

/// The architecture of an unpacked rootfs: its shell's, or the first
/// program's found under bin.
fn tree_arch(tree: &Path) -> Option<&'static str> {
    for rel in ["bin/sh", "bin/busybox", "usr/bin/env", "bin/bash", "usr/bin/sh"] {
        let p = tree.join(rel);
        let p = std::fs::canonicalize(&p).unwrap_or(p);
        if let Some(a) = elf_arch(&p) {
            return Some(a);
        }
    }
    None
}

fn rootfs_import(home: &Home, tarball: &Path, sha256: Option<&str>, env: &[String]) -> Result<()> {
    let env_file = image_env_file(env)?;
    if let Some(want) = sha256 {
        let out = std::process::Command::new("sha256sum").arg(tarball).output().context("run sha256sum")?;
        let got = String::from_utf8_lossy(&out.stdout).split_whitespace().next().unwrap_or("").to_string();
        if got != want.to_lowercase() {
            bail!("{} has SHA-256 {got}, expected {want}", tarball.display());
        }
    }
    let tmp = home.root.join("work").join(format!("rootfs-{}", std::process::id()));
    let _ = std::fs::remove_dir_all(&tmp);
    let tree = tmp.join("tree");
    std::fs::create_dir_all(&tree)?;
    let result = (|| -> Result<String> {
        let status = std::process::Command::new("tar").arg("-xf").arg(tarball).arg("-C").arg(&tree).arg("--no-same-owner").status()?;
        if !status.success() {
            bail!("tar could not unpack {}", tarball.display());
        }
        // The image's architecture must be this machine's: there is no
        // emulation tier (spec §11.2).
        let arch = tree_arch(&tree).unwrap_or("unknown");
        if arch != "unknown" && arch != std::env::consts::ARCH {
            bail!(
                "{} is a {arch} image; this runner is {}, and jobs here run no other architecture",
                tarball.display(),
                std::env::consts::ARCH
            );
        }
        // Mount points flow-init needs (docs/m0.md §2).
        for d in ["src", "tmp", "proc", "dev", "run"] {
            std::fs::create_dir_all(tree.join(d))?;
        }
        if !env_file.is_empty() {
            std::fs::create_dir_all(tree.join("etc/flow"))?;
            std::fs::write(tree.join("etc/flow/env"), &env_file)?;
        }
        let used: u64 = ignore::WalkBuilder::new(&tree)
            .hidden(false)
            .ignore(false)
            .git_ignore(false)
            .build()
            .flatten()
            .filter_map(|e| e.metadata().ok())
            .map(|m| m.len().max(4096))
            .sum();
        let mib = used * 13 / 10 / (1 << 20) + 64;
        let img = tmp.join("rootfs.ext4");
        let status = std::process::Command::new("mkfs.ext4")
            .args(["-q", "-F", "-L", "rootfs", "-E", "root_owner=0:0", "-d"])
            .arg(&tree)
            .arg(&img)
            .arg(format!("{mib}M"))
            .status()
            .context("run mkfs.ext4")?;
        if !status.success() {
            bail!("mkfs.ext4 failed");
        }
        let hash = home::hash_file(&img)?;
        std::fs::rename(&img, home.rootfs(&hash))?;
        let manifest = serde_json::json!({ "arch": if arch == "unknown" { std::env::consts::ARCH } else { arch }, "imported_ms": std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_millis() as u64).unwrap_or(0) });
        std::fs::write(home.image_manifest(&hash), manifest.to_string())?;
        Ok(hash)
    })();
    let _ = std::fs::remove_dir_all(&tmp);
    let hash = result?;
    println!("rootfs \"cas://blob/{hash}\"");
    Ok(())
}

fn run_local(
    home: &Home,
    dir: &Path,
    only: &[String],
    timeout: Duration,
    vars: &[(String, String)],
    secrets: &[(String, String)],
) -> Result<()> {
    let p = load(&dir.join(PIPELINE_PATH))?;
    // The named jobs and everything they depend on.
    let mut wanted: BTreeSet<String> = only.iter().cloned().collect();
    for name in only {
        if !p.jobs.contains_key(name) {
            bail!("no job {name:?} in the pipeline");
        }
    }
    if !only.is_empty() {
        loop {
            let more: Vec<String> = wanted.iter().flat_map(|j| p.jobs[j].depends_on.clone()).filter(|d| !wanted.contains(d)).collect();
            if more.is_empty() {
                break;
            }
            wanted.extend(more);
        }
    }
    let scratch = home.root.join("work").join(format!("local-{}", std::process::id()));
    let _ = std::fs::remove_dir_all(&scratch);
    std::fs::create_dir_all(scratch.join("tree"))?;
    workspace::copy_tree(dir, &scratch.join("tree"))?;
    let mut failed: BTreeSet<String> = BTreeSet::new();
    let mut worst = 0;
    for job in p.order() {
        if !only.is_empty() && !wanted.contains(&job.name) {
            continue;
        }
        if let Some(d) = job.depends_on.iter().find(|d| failed.contains(*d)) {
            println!("── {} skipped: {d} failed", job.name);
            failed.insert(job.name.clone());
            continue;
        }
        let sb = &p.sandboxes[&job.sandbox];
        println!("── {} ({})", job.name, job.exec.join(" "));
        let js = scratch.join(&job.name);
        std::fs::create_dir_all(&js)?;
        let job = &job
            .in_sandbox(sb)
            .with_flow_env(&[("FLOW_JOB".into(), job.name.clone()), ("FLOW_RUN".into(), "local".into())])
            .with_flow_env(vars);
        // Identities come from a hub: a local run has none.
        for i in &job.identities {
            println!("── {}: identity {:?} is not available locally; {} will be missing", job.name, i.audience, i.mount_path);
        }
        // Each secret the job declares must be given, as the hub would.
        let inputs = job
            .secrets
            .iter()
            .map(|s| {
                let value = secrets.iter().find(|(k, _)| *k == s.name).map(|(_, v)| v.clone());
                value
                    .map(|value| flow_core::api::SecretInput { name: s.name.clone(), path: s.mount_path.clone(), value })
                    .ok_or_else(|| anyhow::anyhow!("job {}: secret {:?} is not given; pass --secret {}=VALUE", job.name, s.name, s.name))
            })
            .collect::<Result<Vec<_>>>()?;
        // Each job starts from the same files: jobs do not share writes.
        let out = if sb.runtime == Runtime::Wasm {
            workspace::copy_tree(&scratch.join("tree"), &js.join("tree"))?;
            wasm::run(sb, job, &js.join("tree"), &js, timeout, &mut |r: Record| print_record(&r))?
        } else if sb.runtime == Runtime::Syscall {
            workspace::copy_tree(&scratch.join("tree"), &js.join("tree"))?;
            let group = format!("local-{}-{}", std::process::id(), js.file_name().unwrap_or_default().to_string_lossy());
            syscall::run(home, sb, job, &inputs, &js.join("tree"), &js, timeout, &mut |r: Record| print_record(&r), &group)?
        } else {
            workspace::image(&scratch.join("tree"), &js.join("workspace.ext4"))?;
            microvm::run(home, sb, job, &inputs, &js.join("workspace.ext4"), &js, timeout, &mut |r: Record| print_record(&r))?
        };
        let _ = workspace::remove_tree(&js);
        match out.exit_code {
            Some(0) => println!("── {} passed in {} ms", job.name, out.duration_ms),
            other => {
                println!(
                    "── {} failed ({}) in {} ms",
                    job.name,
                    other.map(|c| format!("exit {c}")).unwrap_or_else(|| "no exit code".into()),
                    out.duration_ms
                );
                failed.insert(job.name.clone());
                worst = 1;
            }
        }
    }
    let _ = workspace::remove_tree(&scratch);
    if worst != 0 {
        std::process::exit(worst);
    }
    Ok(())
}

fn print_record(r: &Record) {
    match r.stream.as_str() {
        "out" => println!("   {}", r.text),
        "err" => eprintln!("   {}", r.text),
        s => eprintln!("   [{s}] {}", r.text),
    }
}

#[cfg(test)]
mod tests {
    #[test]
    fn a_pipeline_newer_than_flow_is_named_as_such() {
        let m = super::not_valid(std::path::Path::new(".flow/pipeline.kdl"), "line 92: unknown output `sbom`; outputs are …");
        assert!(m.contains("does not know") && m.contains(env!("OXY_VERSION_LONG")) && m.contains("newer OxygenFlow"), "{m}");
        let m = super::not_valid(std::path::Path::new(".flow/pipeline.kdl"), "line 3: a job needs a sandbox");
        assert!(m.contains("is not valid") && !m.contains("newer"), "{m}");
    }

    use super::image_env_file;

    #[test]
    fn image_environments_are_checked_lines() {
        let f = image_env_file(&["PATH=/opt/bin:/usr/bin".into(), "X16EMU=/opt/x16emu".into(), "EMPTY=".into()]).unwrap();
        assert_eq!(f, "PATH=/opt/bin:/usr/bin\nX16EMU=/opt/x16emu\nEMPTY=\n");
        assert!(image_env_file(&["NOEQUALS".into()]).is_err());
        assert!(image_env_file(&["1X=y".into()]).is_err());
        assert!(image_env_file(&["X=two\nlines".into()]).is_err());
        assert_eq!(image_env_file(&[]).unwrap(), "");
    }
}
`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet
Coverage / crates/flow/src/main.rs 34.1%

Keyboard
