Flow: jobs that ask for runner labels go to runners with them (refs #41, part 3) c4c87f481d05

/// The labels a compiled job and its sandbox ask a runner for (`runs-on`).

fn labels_in(spec: &str, sandbox: &str) -> Vec<String> {

    let mut out: Vec<String> = [spec, sandbox]

        .iter()

        .filter_map(|j| serde_json::from_str::<serde_json::Value>(j).ok())

        .flat_map(|v| v["runs_on"].as_array().cloned().unwrap_or_default())

        .filter_map(|l| l.as_str().map(String::from))

        .collect();

    out.sort();

    out.dedup();

    out

}

impl Job {

    /// The runner labels this job needs.

    pub fn needs(&self) -> Vec<String> {

        labels_in(&self.spec, &self.sandbox)

    }

}

    pub spec: String,

    pub sandbox: String,

}

impl QueueEntry {

    /// The runner labels the job needs.

    pub fn needs(&self) -> Vec<String> {

        labels_in(&self.spec, &self.sandbox)

    }

        "SELECT j.id AS job_id, j.run_id, r.repo_id, p.owner, p.name, p.visibility, r.number AS run_number, j.name AS job, j.status, j.runner, j.spec, j.sandbox \

    claim_within(pool, runtimes, runner, token_hash, Room::default(), &Default::default(), &[]).await

    labels: &[String],

            continue;

        }

        // Every label the job and its sandbox ask for (`runs-on`).

        if !j.needs().iter().all(|l| labels.contains(l)) {

    // The labels it reports now, or those it registered with.

    let labels: Vec<String> = match &req.info {

        Some(i) => i.labels.clone(),

        None => me.label_list().into_iter().map(String::from).collect(),

    };

    let Some(job) =

        fdb::claim_within(&app.db, &req.runtimes, &runner, &crate::auth::token_hash(&secret), room, &me.scope(), &labels).await?

    else {

            spec: String::new(),

            sandbox: String::new(),

/// runs-on (oxygenhub#41, part 3): a job that asks for a label goes only to

/// a runner that has it, and waits for one otherwise.

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

async fn jobs_that_ask_for_labels_wait_for_a_runner_with_them() {

    use oxy_core::identity::SecretKey;

    let h = hub().await;

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

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

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

    let cross = "job \"cross\" {\n    sandbox \"alpine\"\n    runs-on \"riscv64\"\n    exec \"true\"\n}\n";

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

    c.snapshot("t").unwrap();

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

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

    assert!(!runs_of(&h.app, "andy", "demo").await.is_empty());

    // The site runner has no labels: it takes the others, never cross.

    let mut taken = Vec::new();

    while let Some(j) = claim(&h.app).await {

        taken.push(j["job"]["name"].as_str().unwrap().to_string());

    }

    assert!(!taken.contains(&"cross".to_string()) && taken.contains(&"lint".to_string()), "{taken:?}");

    let queue = db::flow::queue(&h.app.db).await.unwrap();

    let waiting = queue.iter().find(|q| q.job == "cross").unwrap();

    assert_eq!((waiting.status.as_str(), waiting.needs()), ("queued", vec!["riscv64".to_string()]));

    // A runner registered with the label takes it.

    let token = "flowreg_riscv";

    db::runners::add_registration(&h.app.db, &blake3::hash(token.as_bytes()).to_hex(), "admin", &Default::default()).await.unwrap();

    let identity = SecretKey::generate().identity("visionfive", "", 1);

    let info = serde_json::json!({ "arch": "riscv64", "os": "linux", "version": "0.1.0", "labels": ["riscv64"], "max_jobs": 1 });

    let reg = serde_json::json!({ "token": token, "name": "visionfive", "identity": hex::encode(identity.encode()), "info": info });

    let (st, body) = call(&h.app, "POST", "/api/v1/flow/runners/register", None, Some(reg.to_string())).await;

    assert_eq!(st, StatusCode::OK, "{body}");

    let credential = serde_json::from_str::<serde_json::Value>(&body).unwrap()["credential"].as_str().unwrap().to_string();

    let claim_body = serde_json::json!({ "runner": "x", "runtimes": ["microvm"], "info": info }).to_string();

    let (st, body) = call(&h.app, "POST", "/api/v1/flow/claim", Some(&credential), Some(claim_body)).await;

    assert_eq!(st, StatusCode::OK, "{body}");

    assert_eq!(serde_json::from_str::<serde_json::Value>(&body).unwrap()["job"]["name"], "cross");

}

  - **Scopes** (oxygenhub#41): a runner serves the whole site, one owner's repositories (a user or an organisation), or one repository. The registration token carries the scope (typed as empty, `owner` or `owner/repository` when it is made) and the runner gets it; an administrator can change it on the Runners page. A claim returns only queued jobs of repositories in the runner's scope, filtered before the queue is cut short, so a scoped runner never receives another scope's jobs and never waits behind them. A job whose pipeline says `runs-on` (OxygenFlow spec 3.1) goes only to a runner that has all of its job's and sandbox's labels, from those the runner reports on the claim or registered with; until one claims it, the queue says which labels it waits for. `hub.toml` runners serve the whole site. Changes are audited (`flow.runner.scope`). **Settings → Runners** on a repository (its admins) and on an organisation (its owners) lists the runners that may take its jobs, with what each serves; there they make a registration token for a runner of their own scope (the repository, or the organisation's repositories) and remove such runners when they run nothing. Only site administrators manage site runners.

  {% for q in queue %}<tr><td><a href="/{{ q.owner }}/{{ q.name }}">{{ q.owner }}/{{ q.name }}</a></td><td><a href="/{{ q.owner }}/{{ q.name }}/flow/{{ q.run_number }}">#{{ q.run_number }}</a></td><td>{{ q.job }}</td><td>{{ q.status }}</td><td>{% if q.runner.is_empty() %}<span class="mute">waiting{% let needs = q.needs() %}{% if !needs.is_empty() %} for a runner with {% for l in needs %}<span class="lbl">{{ l }}</span> {% endfor %}{% endif %}</span>{% else %}{{ q.runner }}{% endif %}</td></tr>{% endfor %}