Admin › Runners › Images: what images are where 6ca0b971ff07

{% extends "base.html" %}

{% block title %}Images · Administration{% endblock %}

{% block mainclass %} class="wide"{% endblock %}

{% block content %}

<div class="head"><h1>Runners <small>{{ images.len()|plural("image", "images") }} · {{ on_hub }} on the hub · <a href="/admin">Administration</a></small></h1></div>

<div class="bar"><span class="modes"><a href="/admin/runners">runners</a><a href="/admin/runners/images" class="on">images</a><a href="/admin/runners/settings">settings</a></span></div>

<p class="small mute">Every rootfs image a runner holds, the hub keeps, or a job ran on in the last 90 days, newest use first. {% if swarm.is_empty() %}No runner shares images peer to peer yet: give one a <code>[runner.swarm]</code> section.{% else %}Sharing images peer to peer: {% for s in swarm %}{{ s }}{% if !loop.last %}, {% endif %}{% endfor %}.{% endif %} Send an image to the hub with <code>flow rootfs push HASH</code>.</p>

{% if images.is_empty() %}

<p class="mute">No images yet.</p>

{% else %}

<div class="table"><table class="images">

  <tr><th>Image</th><th>Size</th><th>Where</th><th>Used by</th><th>Sent</th></tr>

  {% for i in images %}

  <tr{% if i.stranded() %} class="stranded"{% endif %}>

    <td><code class="chip" title="{{ i.hash }}">{{ i.hash|short }}</code>{% if !i.arch.is_empty() %} <span class="tag">{{ i.arch }}</span>{% endif %}</td>

    <td class="small">{% if let Some(s) = i.size %}{{ s|bytes }}{% else %}<span class="mute" title="Only the hub's copy is measured">—</span>{% endif %}</td>

    <td>{% if i.stranded() %}<span class="tag bad" title="A job pinning it will wait until it is sent: flow rootfs push">nowhere</span>{% endif %}{% for p in i.places %}<span class="lbl img-{{ p.kind }}" title="{{ p.title }}">{% if p.kind == "fetching" %}⇣ {% endif %}{{ p.name }}</span>{% endfor %}</td>

    <td class="small">{% for (repo, jobs, last) in i.uses %}<div><a href="/{{ repo }}/runs">{{ repo }}</a> <span class="mute">{{ jobs|plural("job", "jobs") }}, <time title="{{ last|when_i }}">{{ last|ago_i }}</time></span></div>{% endfor %}{% if i.uses.is_empty() %}<span class="mute">no recent jobs</span>{% endif %}</td>

    <td class="small">{% if let Some((who, when)) = i.sent %}{{ who }} <time class="mute" title="{{ when|when_i }}">{{ when|ago_i }}</time>{% else %}<span class="mute">not on the hub</span>{% endif %}</td>

  </tr>

  {% endfor %}

</table></div>

{% endif %}

{% endblock %}

/// An image this hub keeps (`jet_images`).

#[derive(sqlx::FromRow)]

pub struct Kept {

    pub hash: String,

    pub size: i64,

    pub arch: String,

    pub chunks: i64,

    pub uploader: String,

    pub created_ms: i64,

}

pub async fn kept(pool: &Pool) -> sqlx::Result<Vec<Kept>> {

    sqlx::query_as("SELECT hash, size, arch, chunks, uploader, created_ms FROM jet_images").fetch_all(pool).await

}

/// One repository's use of one image: its jobs since `since_ms`, and when

/// it last ran one.

#[derive(sqlx::FromRow)]

pub struct Use {

    pub image: String,

    pub owner: String,

    pub name: String,

    pub jobs: i64,

    pub last_ms: i64,

}

/// What each repository's jobs ran on since `since_ms`, newest first.

pub async fn uses(pool: &Pool, since_ms: i64) -> sqlx::Result<Vec<Use>> {

    sqlx::query_as(

        "SELECT lower(json_extract(j.sandbox, '$.rootfs')) AS image, p.owner, p.name, COUNT(*) AS jobs,

                MAX(COALESCE(j.started_ms, r.created_ms)) AS last_ms

         FROM flow_jobs j JOIN flow_runs r ON r.id = j.run_id JOIN repositories p ON p.id = r.repo_id

         WHERE r.created_ms >= ? AND COALESCE(json_extract(j.sandbox, '$.rootfs'), '') != ''

         GROUP BY image, p.id ORDER BY last_ms DESC",

    )

    .bind(since_ms)

    .fetch_all(pool)

    .await

}

/// The architecture runners reported for each image (`flow_images`).

pub async fn reported_archs(pool: &Pool) -> sqlx::Result<std::collections::HashMap<String, String>> {

    Ok(sqlx::query_as::<_, (String, String)>("SELECT hash, arch FROM flow_images").fetch_all(pool).await?.into_iter().collect())

}

    /// Who is fetching which image now, by image.

    pub fn fetching_now(&self) -> HashMap<String, Vec<i64>> {

        let fetching = self.fetching.lock().unwrap_or_else(|p| p.into_inner());

        let now = db::now_ms();

        fetching

            .iter()

            .map(|(image, who)| {

                (image.clone(), who.iter().filter(|(_, seen)| now - **seen < FETCHING_FRESH_MS).map(|(r, _)| *r).collect::<Vec<_>>())

            })

            .filter(|(_, who)| !who.is_empty())

            .collect()

    }

        .route("/admin/runners/images", get(runners::images))

/// Where one image is: the hub, a runner holding it, or a runner fetching

/// it from the swarm now.

pub struct Place {

    pub name: String,

    /// `hub`, `holds` or `fetching`.

    pub kind: &'static str,

    pub title: String,

}

pub struct ImageRow {

    pub hash: String,

    pub arch: String,

    /// Its size, when the hub keeps it (runners do not say).

    pub size: Option<i64>,

    pub places: Vec<Place>,

    /// Who sent it to the hub, and when.

    pub sent: Option<(String, i64)>,

    /// Repositories whose jobs ran on it in the last 90 days.

    pub uses: Vec<(String, i64, i64)>,

    pub last_used: i64,

}

impl ImageRow {

    /// Nowhere a job could get it: not on the hub, on no runner.

    pub fn stranded(&self) -> bool {

        self.places.is_empty()

    }

}

#[derive(Template)]

#[template(path = "admin_images.html")]

pub struct ImagesPage {

    shell: Shell,

    images: Vec<ImageRow>,

    on_hub: usize,

    /// Runners in the image swarm, as "name (site)".

    swarm: Vec<String>,

}

/// `GET /admin/runners/images` (C0, docs/cylinder.md): every runner image,

/// where it is and what uses it.

pub async fn images(State(app): State<Shared>, viewer: Viewer) -> Result<Page<ImagesPage>, Error> {

    let shell = Shell::new(&app, "/admin/runners/images", &viewer);

    require_admin(&viewer, &shell)?;

    let e = |e| Error::from_db(shell.clone(), e);

    let kept = crate::images::kept(&app.db).await.map_err(e)?;

    let uses = crate::images::uses(&app.db, db::now_ms() - 90 * 24 * 3600 * 1000).await.map_err(e)?;

    let archs = crate::images::reported_archs(&app.db).await.map_err(e)?;

    let list = runners::list(&app.db).await.map_err(e)?;

    let fetching = app.swarm.fetching_now();

    let mut rows: HashMap<String, ImageRow> = HashMap::new();

    let row = |rows: &mut HashMap<String, ImageRow>, hash: &str| -> () {

        rows.entry(hash.to_ascii_lowercase()).or_insert_with(|| ImageRow {

            hash: hash.to_ascii_lowercase(),

            arch: archs.get(&hash.to_ascii_lowercase()).cloned().unwrap_or_default(),

            size: None,

            places: Vec::new(),

            sent: None,

            uses: Vec::new(),

            last_used: 0,

        });

    };

    for k in &kept {

        row(&mut rows, &k.hash);

        let r = rows.get_mut(&k.hash.to_ascii_lowercase()).expect("added");

        r.size = Some(k.size);

        if r.arch.is_empty() {

            r.arch = k.arch.clone();

        }

        r.sent = Some((k.uploader.clone(), k.created_ms));

        r.places.push(Place { name: "hub".into(), kind: "hub", title: format!("{} chunks kept by the hub", k.chunks) });

    }

    let mut swarm = Vec::new();

    for runner in &list {

        if let Some(a) = app.swarm.addr(runner.id) {

            swarm.push(if a.site.is_empty() { runner.name.clone() } else { format!("{} ({})", runner.name, a.site) });

        }

        for h in runner.image_list().unwrap_or_default() {

            row(&mut rows, &h);

            rows.get_mut(&h.to_ascii_lowercase()).expect("added").places.push(Place {

                name: runner.name.clone(),

                kind: "holds",

                title: format!("{} holds it ({})", runner.name, runner.status()),

            });

        }

    }

    for (image, who) in &fetching {

        row(&mut rows, image);

        for id in who {

            if let Some(runner) = list.iter().find(|r| r.id == *id) {

                rows.get_mut(image).expect("added").places.push(Place {

                    name: runner.name.clone(),

                    kind: "fetching",

                    title: format!("{} is fetching it from the swarm now", runner.name),

                });

            }

        }

    }

    for u in uses {

        if !jet::images::is_hash(&u.image) {

            continue;

        }

        row(&mut rows, &u.image);

        let r = rows.get_mut(&u.image).expect("added");

        r.last_used = r.last_used.max(u.last_ms);

        r.uses.push((format!("{}/{}", u.owner, u.name), u.jobs, u.last_ms));

    }

    let mut images: Vec<ImageRow> = rows.into_values().collect();

    images.sort_by(|a, b| b.last_used.cmp(&a.last_used).then_with(|| a.hash.cmp(&b.hash)));

    let on_hub = kept.len();

    Ok(Page(ImagesPage { shell, images, on_hub, swarm }))

}

/// Admin › Runners › Images (C0): each image, where it is and what uses it.

#[tokio::test(flavor = "multi_thread")]

async fn the_images_page_says_where_each_image_is_and_what_uses_it() {

    let h = hub().await;

    db::accounts::create_account(&h.app.db, "andy", "[email protected]", "x", "salt", true).await.unwrap();

    create(&h.app, "andy", "demo").await;

    let work = tempfile::tempdir().unwrap();

    let c = client(&format!("{}/andy/demo", h.base), work.path());

    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));

    write(work.path(), "test.sh", "echo ok\n");

    c.commit("Add a pipeline", &[]).unwrap();

    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));

    // A runner holds another image and shares it; the hub keeps a third.

    let held = "cd".repeat(32);

    let kept = "ef".repeat(32);

    let body = serde_json::json!({ "runner": "earth-1", "runtimes": ["wasm"], "images": [held], "swarm": { "addr": "10.0.0.1:7401", "site": "home", "cert": "11".repeat(32) } });

    let (st, out) = call(&h.app, "POST", "/api/v1/flow/claim", Some("runner-secret"), Some(body.to_string())).await;

    assert!(st == StatusCode::NO_CONTENT || st == StatusCode::OK, "{st} {out}");

    sqlx::query("INSERT INTO jet_images (hash, size, arch, chunks, uploader, created_ms) VALUES (?, 1048576, 'x86_64', 9, 'andy', 1)")

        .bind(&kept)

        .execute(&h.app.db)

        .await

        .unwrap();

    let res = oxygenhub::router(h.app.clone())

        .oneshot(

            Request::post("/register")

                .header(header::CONTENT_TYPE, "application/x-www-form-urlencoded")

                .body(Body::from("username=carol&email=carol%40example.com&password=correct-horse-battery"))

                .unwrap(),

        )

        .await

        .unwrap();

    let cookie = res.headers().get(header::SET_COOKIE).unwrap().to_str().unwrap().split(';').next().unwrap().to_string();

    let carol = db::accounts::account_by_name(&h.app.db, "carol").await.unwrap().unwrap();

    db::accounts::set_admin(&h.app.db, carol.id, true).await.unwrap();

    let res = oxygenhub::router(h.app.clone())

        .oneshot(Request::get("/admin/runners/images").header(header::COOKIE, cookie).body(Body::empty()).unwrap())

        .await

        .unwrap();

    assert_eq!(res.status(), StatusCode::OK);

    let page = String::from_utf8_lossy(&res.into_body().collect().await.unwrap().to_bytes()).into_owned();

    let row = |hash: &str| {

        let at = page.find(&format!("title=\"{hash}\"")).unwrap_or_else(|| panic!("{hash} is not on the page"));

        let end = page[at..].find("</tr>").unwrap();

        page[at..at + end].to_string()

    };

    let pinned = row(ROOTFS);

    assert!(pinned.contains("nowhere") && pinned.contains("andy/demo") && pinned.contains("3 jobs"), "{pinned}");

    let on_runner = row(&held);

    assert!(on_runner.contains("img-holds") && on_runner.contains("earth-1") && !on_runner.contains("nowhere"), "{on_runner}");

    let on_hub = row(&kept);

    assert!(on_hub.contains("img-hub") && on_hub.contains("1.0 MiB") && on_hub.contains("x86_64"), "{on_hub}");

    assert!(page.contains("earth-1 (home)"), "the swarm is named");

    assert_eq!(call(&h.app, "GET", "/admin/runners/images", None, None).await.0, StatusCode::NOT_FOUND, "administrators only");

}

- **Architecture** (OxygenFlow spec §11.2–§11.3, oxygenflow#36): a runner reports each image's architecture on its claims (`image_archs`), kept in `flow_images`; a job is offered only to a runner whose architecture matches its sandbox's `arch`, else its image's as known; unknown matches, as before. The queue says which architecture a waiting job asks for. **Images by hash** (oxygenflow#39, Jetstream): `flow rootfs push HASH` sends a rootfs image as the FastCDC chunks the hub lacks (`POST /flow/images/missing`, `PUT /flow/images/chunks/{hash}`, by a token that may write repositories) and then its manifest (`PUT /flow/images/{hash}`), which the hub accepts only when every chunk is held and they make an image with that hash; chunks are kept zstd-compressed under `<data>/images`, shared by every image, and `jet_images` lists what is held. Runners and accounts fetch with `GET /flow/images/{hash}` and `GET /flow/images/chunks/{hash}`. A runner whose claims say `fetches_images` is offered jobs whose image only the hub holds, and fetches it before its preflight. Each claim also carries the runner's `facts` (OxygenFlow spec §11.1, oxygenflow#40): free disk, load, speed score and class, kernel; the hub keeps the latest on the runner, shows them on Admin › Runners and in `GET /runners`. A job's finish carries the most disk and memory it took (`peak_disk_mib`, `peak_memory_mib`, oxygenflow#41); a new job's `need_disk_mib` is the most its last five runs of the same repository and name took, plus a quarter and 64 MiB, and it is offered only to a runner whose claim reports at least that much free disk; the waiting reason says so. **Ranking** (oxygenflow#42): a job that another free runner which polled in the last 10 s would do better (it holds the job's image while the claimant would fetch it, or has a higher speed class) is held from the claimant for up to 15 s, then any runner that fits takes it; each job records why it ran where it did (`placed`: "holds the image, idle, speed 100 (class 3)", with "after N s held for a better runner" when it was), shown on the run page, in the API and by `oh run view`. A run whose jobs pin an image held by neither the hub nor a runner sets a failing `pipeline/images` check on its state (#137), naming the images and how to send them (`flow rootfs push`), so a proposal pinning a new image shows it before it lands; a later run of the state turns it to passed once every image is held. A placement pass every 15 seconds works out why each ready queued job no runner can take waits, from what each online runner offered at its last claim (scope, tier, architecture, image held, labels, memory and CPUs, a recent decline), and shows it on the job (API `waiting`, `oh run view` and `oh run watch`, its panel open on the run page) and in its check's summary, so a proposal says why its check is stuck (#28). **Images page** (#146): Admin › Runners › Images lists every rootfs image a runner holds, the hub keeps, or a job ran on in the last 90 days (from each job's sandbox), newest use first: its size and architecture, where it is (the hub, each runner holding it, each runner fetching it from the swarm now), the repositories whose jobs used it, and who sent it; an image held nowhere is marked. **The swarm** (#144, oxygenflow docs/jetstream.md §8.1): a runner in an image swarm reports its peer address, its site and the BLAKE3 of its TLS certificate with each claim (`swarm`; without a certificate it is not in the swarm), kept in memory and shown on Admin › Runners. `POST /flow/swarm/announce {image, state}` (by a runner; `fetching`, `done` or `stopped`) returns the online runners in its site that hold the image or announced fetching it in the last 45 seconds, each with its certificate's fingerprint, which the fetcher pins its TLS connection to, and a ticket signed with the hub key (JSON naming the runner, its site and an expiry 15 minutes on, hex, then the signature) that peers check against `/.well-known/oxyhub-key` before serving chunks; a runner that has not reported an address is told to add `[runner.swarm]` (409).

  <a href="/admin/runners/images">Images ↗</a>

<div class="bar"><span class="modes"><a href="/admin/runners">runners</a><a href="/admin/runners/images">images</a><a href="/admin/runners/settings" class="on">settings</a></span></div>

<div class="bar"><span class="modes"><a href="/admin/runners" class="on">runners</a><a href="/admin/runners/images">images</a><a href="/admin/runners/settings">settings</a></span></div>

/* Admin › Runners › Images (C0, docs/cylinder.md): where each image is. */

table.images td { vertical-align: top; }

table.images tr.stranded td:first-child { box-shadow: inset 3px 0 0 var(--bad); }

.lbl.img-hub { color: var(--acc); border-color: var(--acc); }

.lbl.img-holds { color: var(--ok); border-color: var(--ok); }

.lbl.img-fetching { color: var(--warn); border-color: var(--warn); }