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