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