crates/flow/src/main.rs at ecce3719da80 · andymccall/oxygenflow

# ecce3719da80 oxygenflow / crates / flow / src / main.rs

22.6 KiB · Rust · last changed Jetstream: runners share images peer to peer 24 days ago source blame history raw

//! `flow`: check pipelines, run them locally, and run jobs for OxygenHub
//! (docs/spec.md, docs/m0.md).

mod cpio;
mod facts;
/// The job-starting code `flow-init` uses in a microVM, shared with the
/// syscall tier.
#[path = "../../flow-init/src/job.rs"]
mod guest;
mod home;
mod images;
mod microvm;
mod outputs;
mod runner;
mod swarm;
mod syscall;
mod wait;
mod 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,
}

#[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. Uses
    /// $OXY_TOKEN (a token that may write repositories) and $OXY_CA
    Push {
        /// The image's hash, as pipelines pin it (cas://blob/HASH)
        hash: String,
        /// The hub (default: 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. Uses $OXY_TOKEN and $OXY_CA
    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::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 the runner's; with
/// $OXY_TOKEN, else the runner's own token.
fn image_client(home: &Home, hub: Option<String>) -> Result<images::Client> {
    let runner = home.config.runner.clone();
    let hub = hub.or_else(|| runner.as_ref().map(|r| r.hub.clone())).context("name the hub with --hub https://HUB")?;
    let token = std::env::var("OXY_TOKEN")
        .ok()
        .or_else(|| runner.as_ref().map(|r| r.token.clone()))
        .context("set OXY_TOKEN to a token of yours on the hub")?;
    let ca = std::env::var_os("OXY_CA").map(PathBuf::from).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!("{} is not valid:\n{p}", 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 bytes = std::fs::read(path).ok()?;
    if bytes.len() < 20 || &bytes[..4] != b"\x7fELF" {
        return None;
    }
    let little = bytes[5] == 1;
    let machine = if little { u16::from_le_bytes([bytes[18], bytes[19]]) } else { u16::from_be_bytes([bytes[18], bytes[19]]) };
    Some(match machine {
        0x3E => "x86_64",
        0xB7 => "aarch64",
        0xF3 => "riscv64",
        0x03 => "x86",
        0x28 => "arm",
        0x15 => "powerpc64",
        0x16 => "s390x",
        _ => return None,
    })
}

/// 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];
        if sb.runtime == Runtime::Wasm {
            bail!("job {}: runtime {} is not available in this build yet", job.name, sb.runtime.as_str());
        }
        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::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 {
    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

1	//! `flow`: check pipelines, run them locally, and run jobs for OxygenHub
2	//! (docs/spec.md, docs/m0.md).
3
4	mod cpio;
5	mod facts;
6	/// The job-starting code `flow-init` uses in a microVM, shared with the
7	/// syscall tier.
8	#[path = "../../flow-init/src/job.rs"]
9	mod guest;
10	mod home;
11	mod images;
12	mod microvm;
13	mod outputs;
14	mod runner;
15	mod swarm;
16	mod syscall;
17	mod wait;
18	mod workspace;
19
20	use anyhow::{bail, Context, Result};
21	use clap::{Parser, Subcommand};
22	use flow_core::api::Record;
23	use flow_core::{Pipeline, Runtime, PIPELINE_PATH};
24	use home::Home;
25	use std::collections::BTreeSet;
26	use std::path::{Path, PathBuf};
27	use std::time::Duration;
28
29	#[derive(Parser)]
30	#[command(name = "flow", version = env!("OXY_VERSION_LONG"), about = "OxygenFlow: hermetic pipelines for Oxygen repositories")]
31	struct Cli {
32	/// The Flow home (default $FLOW_HOME, else ~/.local/share/flow).
33	#[arg(long, global = true)]
34	home: Option<PathBuf>,
35	#[command(subcommand)]
36	cmd: Cmd,
37	}
38
39	#[derive(Subcommand)]
40	enum RunnerCmd {
41	/// Register this runner with a hub using a one-time token from its Admin → Runners page
42	Register {
43	/// The hub, e.g. https://earth.example:7343
44	hub: String,
45	/// The registration token (read from stdin when not given)
46	#[arg(long)]
47	token: Option<String>,
48	/// The runner's name (default: the host name)
49	#[arg(long)]
50	name: Option<String>,
51	/// The hub's CA certificate, when self-signed
52	#[arg(long)]
53	ca: Option<PathBuf>,
54	/// A label the runner offers (repeatable)
55	#[arg(long)]
56	label: Vec<String>,
57	},
58	/// Measure this machine as a runner does: its speed score (100 on the
59	/// reference machine) and class, free disk and load (spec §11.1)
60	Bench,
61	}
62
63	#[derive(Subcommand)]
64	enum Cmd {
65	/// Validate a pipeline and print its jobs and hash
66	Check {
67	/// The pipeline file (default: .flow/pipeline.kdl)
68	file: Option<PathBuf>,
69	},
70	/// Print a file's BLAKE3, for pinning in flow.toml
71	Hash { file: PathBuf },
72	/// Rootfs images for sandboxes
73	Rootfs {
74	#[command(subcommand)]
75	cmd: RootfsCmd,
76	},
77	/// Run the pipeline on this directory's files, locally
78	Run {
79	/// Only these jobs (and what they depend on)
80	jobs: Vec<String>,
81	/// The directory to build (default: the current one)
82	#[arg(long, default_value = ".")]
83	dir: PathBuf,
84	/// Minutes before a job is stopped
85	#[arg(long, default_value_t = 60)]
86	timeout: u64,
87	/// A variable for every job, as the hub's settings would give it;
88	/// the pipeline's own `env` wins
89	#[arg(long = "var", value_name = "NAME=VALUE")]
90	vars: Vec<String>,
91	/// A secret a job's `secret` block names, as the hub would supply
92	/// it; mounted as a file, never in the environment
93	#[arg(long = "secret", value_name = "NAME=VALUE")]
94	secrets: Vec<String>,
95	},
96	/// Claim and run jobs for the hub in flow.toml's [runner] (or the
97	/// runner.toml `flow runner register` wrote)
98	Runner {
99	#[command(subcommand)]
100	cmd: Option<RunnerCmd>,
101	},
102	/// Follow the checks of a pushed state on OxygenHub until they finish:
103	/// exit 0 when all passed, 1 when one failed, 2 at the time limit.
104	/// Uses $OXY_TOKEN and $OXY_CA as oxy does.
105	Wait {
106	/// The repository's URL on the hub (https://hub/owner/name)
107	url: String,
108	/// The state hash the push moved the bookmark to (`oxy push --json` prints it)
109	state: String,
110	/// Minutes to wait at most
111	#[arg(long, default_value_t = 60)]
112	timeout: u64,
113	},
114	/// Show which isolation tiers this machine can run jobs in, and why
115	/// the others cannot
116	Tiers,
117	/// The syscall tier's sandbox helper, started by the runner for each job
118	#[command(name = "__sandbox", hide = true)]
119	Sandbox { spec: PathBuf },
120	/// Check that this machine can run the syscall tier
121	#[command(name = "__sandbox-probe", hide = true)]
122	SandboxProbe,
123	}
124
125	#[derive(Subcommand)]
126	enum RootfsCmd {
127	/// Make an ext4 rootfs image from a root file system tarball and store it by hash
128	Import {
129	tarball: PathBuf,
130	/// Refuse the tarball unless its SHA-256 matches (as its publisher states it)
131	#[arg(long)]
132	sha256: Option<String>,
133	/// The image's own environment, NAME=value (repeatable): what a
134	/// Dockerfile's ENV sets and `docker export` loses. Jobs start from
135	/// it; their sandbox's and their own `env` override it
136	#[arg(long = "env", value_name = "NAME=VALUE")]
137	env: Vec<String>,
138	},
139	/// List the images this home holds
140	List,
141	/// Send an image to a hub, so its runners fetch it instead of having it
142	/// copied to them: only the chunks the hub lacks travel. Uses
143	/// $OXY_TOKEN (a token that may write repositories) and $OXY_CA
144	Push {
145	/// The image's hash, as pipelines pin it (cas://blob/HASH)
146	hash: String,
147	/// The hub (default: the runner's, from runner.toml or flow.toml)
148	#[arg(long)]
149	hub: Option<String>,
150	},
151	/// Fetch an image a hub holds into this home, as a runner does for a
152	/// job whose image it lacks. Uses $OXY_TOKEN and $OXY_CA
153	Pull {
154	hash: String,
155	#[arg(long)]
156	hub: Option<String>,
157	},
158	}
159
160	fn main() {
161	let cli = Cli::parse();
162	if let Err(e) = run(cli) {
163	eprintln!("flow: {e:#}");
164	std::process::exit(1);
165	}
166	}
167
168	fn run(cli: Cli) -> Result<()> {
169	match cli.cmd {
170	Cmd::Sandbox { spec } => std::process::exit(syscall::sandbox_main(&spec)),
171	Cmd::SandboxProbe => std::process::exit(syscall::probe_main()),
172	Cmd::Check { file } => check(&file.unwrap_or_else(\|\| PathBuf::from(PIPELINE_PATH))),
173	Cmd::Hash { file } => {
174	println!("{}", home::hash_file(&file)?);
175	Ok(())
176	}
177	Cmd::Rootfs { cmd } => {
178	let home = Home::locate(cli.home.as_deref())?;
179	match cmd {
180	RootfsCmd::Import { tarball, sha256, env } => rootfs_import(&home, &tarball, sha256.as_deref(), &env),
181	RootfsCmd::Push { hash, hub } => {
182	let client = image_client(&home, hub)?;
183	eprintln!("Sending image {} …", &hash[..12.min(hash.len())]);
184	let m = client.push(&home, &hash)?;
185	println!(
186	"Sent image {}: {} chunks the hub lacked, {} in {:.1} s",
187	&hash[..12],
188	m.chunks,
189	images::human(m.bytes),
190	m.elapsed.as_secs_f64()
191	);
192	Ok(())
193	}
194	RootfsCmd::Pull { hash, hub } => {
195	let client = image_client(&home, hub)?;
196	match client.fetch(&home, &hash)? {
197	Some(m) => {
198	println!(
199	"Fetched image {}: {} chunks, {} in {:.1} s",
200	&hash[..12],
201	m.chunks,
202	images::human(m.bytes),
203	m.elapsed.as_secs_f64()
204	);
205	Ok(())
206	}
207	None => anyhow::bail!("the hub does not hold image {}; send it with flow rootfs push", &hash[..12.min(hash.len())]),
208	}
209	}
210	RootfsCmd::List => {
211	for e in std::fs::read_dir(home.root.join("cas"))?.flatten() {
212	let size = e.metadata().map(\|m\| m.len()).unwrap_or(0);
213	if let Some(h) = e.file_name().to_string_lossy().strip_suffix(".ext4") {
214	println!("cas://blob/{h} {} MiB", size >> 20);
215	}
216	}
217	Ok(())
218	}
219	}
220	}
221	Cmd::Run { jobs, dir, timeout, vars, secrets } => {
222	let home = Home::locate(cli.home.as_deref())?;
223	let pairs = \|what: &str, list: &[String]\| {
224	list.iter()
225	.map(\|e\| {
226	e.split_once('=')
227	.map(\|(k, v)\| (k.to_string(), v.to_string()))
228	.ok_or_else(\|\| anyhow::anyhow!("--{what} {e:?}: expected NAME=VALUE"))
229	})
230	.collect::<Result<Vec<_>>>()
231	};
232	let (vars, secrets) = (pairs("var", &vars)?, pairs("secret", &secrets)?);
233	run_local(&home, &dir, &jobs, Duration::from_secs(timeout * 60), &vars, &secrets)
234	}
235	Cmd::Runner { cmd: Some(RunnerCmd::Bench) } => {
236	let home = Home::locate(cli.home.as_deref())?;
237	let (speed, ms) = facts::bench();
238	let f = facts::now(&home, Some(speed));
239	println!("speed {speed} (class {}), benchmark {ms} ms", flow_core::api::Facts::class_of(speed));
240	println!("free disk {} MiB at {}", f.free_disk_mib.unwrap_or(0), home.root.display());
241	println!("load {:.2}", f.load_x100.unwrap_or(0) as f64 / 100.0);
242	if let Some(k) = f.kernel {
243	println!("kernel {k}");
244	}
245	Ok(())
246	}
247	Cmd::Runner { cmd: Some(RunnerCmd::Register { hub, token, name, ca, label }) } => {
248	let home = Home::locate(cli.home.as_deref())?;
249	if std::env::var_os("OXY_HOME").is_none() {
250	std::env::set_var("OXY_HOME", home.root.join("oxy"));
251	}
252	let name = name.unwrap_or_else(\|\| {
253	std::fs::read_to_string("/etc/hostname").map(\|h\| h.trim().to_string()).unwrap_or_else(\|_\| "flow".into())
254	});
255	let token = match token {
256	Some(t) => t,
257	None => {
258	eprintln!("Paste the registration token from the hub's Admin → Runners, then press Enter:");
259	let mut line = String::new();
260	std::io::stdin().read_line(&mut line)?;
261	line.trim().to_string()
262	}
263	};
264	let got = runner::register(&home, &hub, &token, &name, ca, label)?;
265	println!("registered as {} (runner {}); start it with `flow runner`", got.name, got.id);
266	Ok(())
267	}
268	Cmd::Runner { cmd: None } => {
269	let home = Home::locate(cli.home.as_deref())?;
270	// The runner's oxy state (identity, user config) stays in its home.
271	if std::env::var_os("OXY_HOME").is_none() {
272	std::env::set_var("OXY_HOME", home.root.join("oxy"));
273	}
274	runner::serve(&home)
275	}
276	Cmd::Tiers => {
277	let home = Home::locate(cli.home.as_deref())?;
278	match runner::tiers(&home) {
279	Ok((have, missing)) => {
280	for t in have {
281	println!("{t}: available");
282	}
283	for why in missing {
284	println!("{}", why.replacen(": ", ": unavailable: ", 1));
285	}
286	}
287	Err(why) => println!("{why}"),
288	}
289	Ok(())
290	}
291	Cmd::Wait { url, state, timeout } => {
292	let token = std::env::var("OXY_TOKEN").ok();
293	let ca = std::env::var_os("OXY_CA").map(PathBuf::from);
294	let code =
295	wait::wait(&url, &state, token.as_deref(), ca.as_deref(), Duration::from_secs(timeout * 60), Duration::from_secs(15))?;
296	std::process::exit(code);
297	}
298	}
299	}
300
301	/// The hub image commands talk to: `--hub`, else the runner's; with
302	/// $OXY_TOKEN, else the runner's own token.
303	fn image_client(home: &Home, hub: Option<String>) -> Result<images::Client> {
304	let runner = home.config.runner.clone();
305	let hub = hub.or_else(\|\| runner.as_ref().map(\|r\| r.hub.clone())).context("name the hub with --hub https://HUB")?;
306	let token = std::env::var("OXY_TOKEN")
307	.ok()
308	.or_else(\|\| runner.as_ref().map(\|r\| r.token.clone()))
309	.context("set OXY_TOKEN to a token of yours on the hub")?;
310	let ca = std::env::var_os("OXY_CA").map(PathBuf::from).or_else(\|\| runner.as_ref().and_then(\|r\| r.ca.clone()));
311	images::Client::new(&hub, &token, ca.as_deref())
312	}
313
314	fn load(file: &Path) -> Result<Pipeline> {
315	let src = std::fs::read_to_string(file).with_context(\|\| format!("read {}", file.display()))?;
316	Pipeline::parse(&src).map_err(\|p\| anyhow::anyhow!("{} is not valid:\n{p}", file.display()))
317	}
318
319	fn check(file: &Path) -> Result<()> {
320	let p = load(file)?;
321	let depths = p.depths();
322	for j in p.order() {
323	let sb = &p.sandboxes[&j.sandbox];
324	let after = if j.depends_on.is_empty() { String::new() } else { format!(" after {}", j.depends_on.join(", ")) };
325	println!("{}{} {} [{}]{after}", " ".repeat(depths[j.name.as_str()]), j.name, j.exec.join(" "), sb.runtime.as_str());
326	}
327	println!("pipeline {}", hex::encode(p.hash()));
328	Ok(())
329	}
330
331	/// The image environment file flow-init reads (`/etc/flow/env`): one
332	/// `NAME=value` per line, names checked as a pipeline's are.
333	fn image_env_file(env: &[String]) -> Result<String> {
334	let mut out = String::new();
335	for e in env {
336	let Some((k, v)) = e.split_once('=') else { bail!("--env {e:?}: expected NAME=value") };
337	let ok = !k.is_empty() && !k.starts_with(\|c: char\| c.is_ascii_digit()) && k.bytes().all(\|b\| b.is_ascii_alphanumeric() \|\| b == b'_');
338	if !ok \|\| v.contains('\n') \|\| v.contains('\0') {
339	bail!("--env {e:?}: a name is letters, digits and _, and a value is one line");
340	}
341	out.push_str(&format!("{k}={v}\n"));
342	}
343	Ok(out)
344	}
345
346	/// The architecture an ELF file was built for, by its header.
347	fn elf_arch(path: &Path) -> Option<&'static str> {
348	let bytes = std::fs::read(path).ok()?;
349	if bytes.len() < 20 \|\| &bytes[..4] != b"\x7fELF" {
350	return None;
351	}
352	let little = bytes[5] == 1;
353	let machine = if little { u16::from_le_bytes([bytes[18], bytes[19]]) } else { u16::from_be_bytes([bytes[18], bytes[19]]) };
354	Some(match machine {
355	0x3E => "x86_64",
356	0xB7 => "aarch64",
357	0xF3 => "riscv64",
358	0x03 => "x86",
359	0x28 => "arm",
360	0x15 => "powerpc64",
361	0x16 => "s390x",
362	_ => return None,
363	})
364	}
365
366	/// The architecture of an unpacked rootfs: its shell's, or the first
367	/// program's found under bin.
368	fn tree_arch(tree: &Path) -> Option<&'static str> {
369	for rel in ["bin/sh", "bin/busybox", "usr/bin/env", "bin/bash", "usr/bin/sh"] {
370	let p = tree.join(rel);
371	let p = std::fs::canonicalize(&p).unwrap_or(p);
372	if let Some(a) = elf_arch(&p) {
373	return Some(a);
374	}
375	}
376	None
377	}
378
379	fn rootfs_import(home: &Home, tarball: &Path, sha256: Option<&str>, env: &[String]) -> Result<()> {
380	let env_file = image_env_file(env)?;
381	if let Some(want) = sha256 {
382	let out = std::process::Command::new("sha256sum").arg(tarball).output().context("run sha256sum")?;
383	let got = String::from_utf8_lossy(&out.stdout).split_whitespace().next().unwrap_or("").to_string();
384	if got != want.to_lowercase() {
385	bail!("{} has SHA-256 {got}, expected {want}", tarball.display());
386	}
387	}
388	let tmp = home.root.join("work").join(format!("rootfs-{}", std::process::id()));
389	let _ = std::fs::remove_dir_all(&tmp);
390	let tree = tmp.join("tree");
391	std::fs::create_dir_all(&tree)?;
392	let result = (\|\| -> Result<String> {
393	let status = std::process::Command::new("tar").arg("-xf").arg(tarball).arg("-C").arg(&tree).arg("--no-same-owner").status()?;
394	if !status.success() {
395	bail!("tar could not unpack {}", tarball.display());
396	}
397	// The image's architecture must be this machine's: there is no
398	// emulation tier (spec §11.2).
399	let arch = tree_arch(&tree).unwrap_or("unknown");
400	if arch != "unknown" && arch != std::env::consts::ARCH {
401	bail!(
402	"{} is a {arch} image; this runner is {}, and jobs here run no other architecture",
403	tarball.display(),
404	std::env::consts::ARCH
405	);
406	}
407	// Mount points flow-init needs (docs/m0.md §2).
408	for d in ["src", "tmp", "proc", "dev", "run"] {
409	std::fs::create_dir_all(tree.join(d))?;
410	}
411	if !env_file.is_empty() {
412	std::fs::create_dir_all(tree.join("etc/flow"))?;
413	std::fs::write(tree.join("etc/flow/env"), &env_file)?;
414	}
415	let used: u64 = ignore::WalkBuilder::new(&tree)
416	.hidden(false)
417	.ignore(false)
418	.git_ignore(false)
419	.build()
420	.flatten()
421	.filter_map(\|e\| e.metadata().ok())
422	.map(\|m\| m.len().max(4096))
423	.sum();
424	let mib = used * 13 / 10 / (1 << 20) + 64;
425	let img = tmp.join("rootfs.ext4");
426	let status = std::process::Command::new("mkfs.ext4")
427	.args(["-q", "-F", "-L", "rootfs", "-E", "root_owner=0:0", "-d"])
428	.arg(&tree)
429	.arg(&img)
430	.arg(format!("{mib}M"))
431	.status()
432	.context("run mkfs.ext4")?;
433	if !status.success() {
434	bail!("mkfs.ext4 failed");
435	}
436	let hash = home::hash_file(&img)?;
437	std::fs::rename(&img, home.rootfs(&hash))?;
438	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) });
439	std::fs::write(home.image_manifest(&hash), manifest.to_string())?;
440	Ok(hash)
441	})();
442	let _ = std::fs::remove_dir_all(&tmp);
443	let hash = result?;
444	println!("rootfs \"cas://blob/{hash}\"");
445	Ok(())
446	}
447
448	fn run_local(
449	home: &Home,
450	dir: &Path,
451	only: &[String],
452	timeout: Duration,
453	vars: &[(String, String)],
454	secrets: &[(String, String)],
455	) -> Result<()> {
456	let p = load(&dir.join(PIPELINE_PATH))?;
457	// The named jobs and everything they depend on.
458	let mut wanted: BTreeSet<String> = only.iter().cloned().collect();
459	for name in only {
460	if !p.jobs.contains_key(name) {
461	bail!("no job {name:?} in the pipeline");
462	}
463	}
464	if !only.is_empty() {
465	loop {
466	let more: Vec<String> = wanted.iter().flat_map(\|j\| p.jobs[j].depends_on.clone()).filter(\|d\| !wanted.contains(d)).collect();
467	if more.is_empty() {
468	break;
469	}
470	wanted.extend(more);
471	}
472	}
473	let scratch = home.root.join("work").join(format!("local-{}", std::process::id()));
474	let _ = std::fs::remove_dir_all(&scratch);
475	std::fs::create_dir_all(scratch.join("tree"))?;
476	workspace::copy_tree(dir, &scratch.join("tree"))?;
477	let mut failed: BTreeSet<String> = BTreeSet::new();
478	let mut worst = 0;
479	for job in p.order() {
480	if !only.is_empty() && !wanted.contains(&job.name) {
481	continue;
482	}
483	if let Some(d) = job.depends_on.iter().find(\|d\| failed.contains(*d)) {
484	println!("── {} skipped: {d} failed", job.name);
485	failed.insert(job.name.clone());
486	continue;
487	}
488	let sb = &p.sandboxes[&job.sandbox];
489	if sb.runtime == Runtime::Wasm {
490	bail!("job {}: runtime {} is not available in this build yet", job.name, sb.runtime.as_str());
491	}
492	println!("── {} ({})", job.name, job.exec.join(" "));
493	let js = scratch.join(&job.name);
494	std::fs::create_dir_all(&js)?;
495	let job = &job
496	.in_sandbox(sb)
497	.with_flow_env(&[("FLOW_JOB".into(), job.name.clone()), ("FLOW_RUN".into(), "local".into())])
498	.with_flow_env(vars);
499	// Identities come from a hub: a local run has none.
500	for i in &job.identities {
501	println!("── {}: identity {:?} is not available locally; {} will be missing", job.name, i.audience, i.mount_path);
502	}
503	// Each secret the job declares must be given, as the hub would.
504	let inputs = job
505	.secrets
506	.iter()
507	.map(\|s\| {
508	let value = secrets.iter().find(\|(k, _)\| *k == s.name).map(\|(_, v)\| v.clone());
509	value
510	.map(\|value\| flow_core::api::SecretInput { name: s.name.clone(), path: s.mount_path.clone(), value })
511	.ok_or_else(\|\| anyhow::anyhow!("job {}: secret {:?} is not given; pass --secret {}=VALUE", job.name, s.name, s.name))
512	})
513	.collect::<Result<Vec<_>>>()?;
514	// Each job starts from the same files: jobs do not share writes.
515	let out = if sb.runtime == Runtime::Syscall {
516	workspace::copy_tree(&scratch.join("tree"), &js.join("tree"))?;
517	let group = format!("local-{}-{}", std::process::id(), js.file_name().unwrap_or_default().to_string_lossy());
518	syscall::run(home, sb, job, &inputs, &js.join("tree"), &js, timeout, &mut \|r: Record\| print_record(&r), &group)?
519	} else {
520	workspace::image(&scratch.join("tree"), &js.join("workspace.ext4"))?;
521	microvm::run(home, sb, job, &inputs, &js.join("workspace.ext4"), &js, timeout, &mut \|r: Record\| print_record(&r))?
522	};
523	let _ = workspace::remove_tree(&js);
524	match out.exit_code {
525	Some(0) => println!("── {} passed in {} ms", job.name, out.duration_ms),
526	other => {
527	println!(
528	"── {} failed ({}) in {} ms",
529	job.name,
530	other.map(\|c\| format!("exit {c}")).unwrap_or_else(\|\| "no exit code".into()),
531	out.duration_ms
532	);
533	failed.insert(job.name.clone());
534	worst = 1;
535	}
536	}
537	}
538	let _ = workspace::remove_tree(&scratch);
539	if worst != 0 {
540	std::process::exit(worst);
541	}
542	Ok(())
543	}
544
545	fn print_record(r: &Record) {
546	match r.stream.as_str() {
547	"out" => println!(" {}", r.text),
548	"err" => eprintln!(" {}", r.text),
549	s => eprintln!(" [{s}] {}", r.text),
550	}
551	}
552
553	#[cfg(test)]
554	mod tests {
555	use super::image_env_file;
556
557	#[test]
558	fn image_environments_are_checked_lines() {
559	let f = image_env_file(&["PATH=/opt/bin:/usr/bin".into(), "X16EMU=/opt/x16emu".into(), "EMPTY=".into()]).unwrap();
560	assert_eq!(f, "PATH=/opt/bin:/usr/bin\nX16EMU=/opt/x16emu\nEMPTY=\n");
561	assert!(image_env_file(&["NOEQUALS".into()]).is_err());
562	assert!(image_env_file(&["1X=y".into()]).is_err());
563	assert!(image_env_file(&["X=two\nlines".into()]).is_err());
564	assert_eq!(image_env_file(&[]).unwrap(), "");
565	}
566	}

Keyboard