Runners: the queue on Admin → Runners, and the runner API (refs #41, part 4) b66e70b7b6c7

/// An OxygenFlow runner (`GET /api/v1/runners`, site administrators).

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]

pub struct Runner {

    pub id: i64,

    pub name: String,

    /// `online`, `offline`, `draining` or `disabled`.

    pub status: String,

    /// What it serves: empty for the whole site, an owner, or `owner/name`.

    pub serves: String,

    pub arch: String,

    pub os: String,

    pub version: String,

    pub labels: Vec<String>,

    pub max_jobs: i64,

    pub memory_mib: Option<i64>,

    pub cpus: Option<i64>,

    /// The key that signs its receipts (KeyId hex), empty until bound.

    pub key: String,

    /// `registered` or `config` (a hub.toml token).

    pub source: String,

    pub last_seen_ms: Option<i64>,

    /// Jobs it is running now.

    pub running: i64,

}

/// A job a runner ran or is running.

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]

pub struct RunnerJob {

    pub repo: String,

    pub run: i64,

    pub job: String,

    pub status: String,

    pub started_ms: Option<i64>,

    pub finished_ms: Option<i64>,

}

/// A job in the queue: running ones first, then queued ones in the order

/// runners claim them.

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]

pub struct QueueItem {

    pub repo: String,

    pub run: i64,

    pub job: String,

    /// `running` or `queued`.

    pub status: String,

    /// The runner holding it, when running.

    pub runner: String,

}

export type Landed = {

  by: string;

  via: string;

  when_ms: number;

};

  landed?: Landed | null;

  requested_reviewers?: string[];

        .route("/runners", get(runners))

        .route("/runners/queue", get(runner_queue))

        .route("/runners/{id}/jobs", get(runner_jobs))

// ----------------------------------------------------------------- runners

/// Site administrators only, with any token of theirs (hub spec §13.2).

fn require_site_admin(c: &Caller) -> Result<&Account, ApiError> {

    match c.account.as_ref() {

        Some(a) if a.is_admin => Ok(a),

        _ => Err(not_found("no such API route")),

    }

}

async fn runner_of(app: &Shared, r: db::runners::Runner, running: i64) -> hub_api::Runner {

    let serves = match r.scope_repo_id {

        Some(id) => db::repos::get_by_id(&app.db, id).await.ok().flatten().map(|p| p.slug()).unwrap_or_default(),

        None => r.scope_owner.clone(),

    };

    hub_api::Runner {

        id: r.id,

        status: r.status().to_string(),

        serves,

        labels: r.label_list().into_iter().map(String::from).collect(),

        name: r.name,

        arch: r.arch,

        os: r.os,

        version: r.version,

        max_jobs: r.max_jobs,

        memory_mib: r.memory_mib,

        cpus: r.cpus,

        key: r.key,

        source: r.source,

        last_seen_ms: r.last_seen_ms,

        running,

    }

}

/// `GET /api/v1/runners`.

async fn runners(State(app): State<Shared>, headers: HeaderMap) -> Result<Json<Vec<hub_api::Runner>>, ApiError> {

    let c = caller(&app, &headers).await?;

    require_site_admin(&c)?;

    let running: std::collections::HashMap<i64, i64> = db::runners::running(&app.db).await?.into_iter().collect();

    let mut out = Vec::new();

    for r in db::runners::list(&app.db).await? {

        let n = running.get(&r.id).copied().unwrap_or(0);

        out.push(runner_of(&app, r, n).await);

    }

    Ok(Json(out))

}

#[derive(Deserialize)]

struct BeforeQuery {

    #[serde(default)]

    before: Option<i64>,

    #[serde(default)]

    limit: Option<i64>,

}

/// `GET /api/v1/runners/{id}/jobs?before=MS&limit=N`: newest first.

async fn runner_jobs(

    State(app): State<Shared>,

    headers: HeaderMap,

    Path(id): Path<i64>,

    Query(q): Query<BeforeQuery>,

) -> Result<Json<Vec<hub_api::RunnerJob>>, ApiError> {

    let c = caller(&app, &headers).await?;

    require_site_admin(&c)?;

    db::runners::get(&app.db, id).await?.ok_or_else(|| not_found("no such runner"))?;

    let jobs = db::runners::history(&app.db, id, q.limit.unwrap_or(50).clamp(1, 200), q.before).await?;

    Ok(Json(

        jobs.into_iter()

            .map(|j| hub_api::RunnerJob {

                repo: format!("{}/{}", j.owner, j.repo),

                run: j.run_number,

                job: j.job,

                status: j.status,

                started_ms: j.started_ms,

                finished_ms: j.finished_ms,

            })

            .collect(),

    ))

}

/// `GET /api/v1/runners/queue`: running jobs, then queued ones in claim order.

async fn runner_queue(State(app): State<Shared>, headers: HeaderMap) -> Result<Json<Vec<hub_api::QueueItem>>, ApiError> {

    let c = caller(&app, &headers).await?;

    require_site_admin(&c)?;

    Ok(Json(

        db::flow::queue(&app.db)

            .await?

            .into_iter()

            .map(|e| hub_api::QueueItem {

                repo: format!("{}/{}", e.owner, e.name),

                run: e.run_number,

                job: e.job,

                status: e.status,

                runner: e.runner,

            })

            .collect(),

    ))

}

    /// The runner holding it, when running.

    pub runner: String,

        "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 \

            runner: String::new(),

    /// Running jobs, then queued ones in the order runners claim them.

    queue: Vec<db::flow::QueueEntry>,

    let queue = db::flow::queue(&app.db).await.map_err(e)?;

    Ok(Page(RunnersPage { hub: app.config.base_url().to_string(), shell, runners: rows, pending, new_token, notice, error, queue }))

    // The queue on Admin → Runners and through the API (#41, part 4).

    let (_, page) = web("GET", "/admin/runners".into(), None).await;

    assert!(page.contains("<h2>Queue</h2>") && page.contains("<td>plex</td>"), "{page}");

    let (_, tokens) = web("GET", "/settings/tokens".into(), None).await;

    let (_, made) = web("POST", "/settings/tokens".into(), Some(format!("_csrf={}&name=admin&scopes=user%3Aread", csrf_of(&tokens)))).await;

    let i = made.find("oxyhub_").unwrap();

    let admin_token: String = made[i..].chars().take_while(|c| c.is_ascii_alphanumeric() || *c == '_').collect();

    let (st, body) = call(&h.app, "GET", "/api/v1/runners", Some(&admin_token), None).await;

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

    let listed: serde_json::Value = serde_json::from_str(&body).unwrap();

    let plex_row = listed.as_array().unwrap().iter().find(|r| r["name"] == "plex").unwrap();

    assert_eq!((plex_row["running"].as_i64(), plex_row["arch"].as_str()), (Some(1), Some("x86_64")));

    let (_, body) = call(&h.app, "GET", "/api/v1/runners/queue", Some(&admin_token), None).await;

    let queue: serde_json::Value = serde_json::from_str(&body).unwrap();

    assert!(queue.as_array().unwrap().iter().any(|q| q["status"] == "running" && q["runner"] == "plex"), "{queue}");

    let (_, body) = call(&h.app, "GET", &format!("/api/v1/runners/{}/jobs", plex.id), Some(&admin_token), None).await;

    let jobs: serde_json::Value = serde_json::from_str(&body).unwrap();

    assert_eq!(jobs[0]["repo"], "andy/demo", "{jobs}");

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

  - **The queue** (on the Runners page): running jobs with the runner holding each, then queued jobs in the order runners claim them.

  - **API** (site administrators): `GET /api/v1/runners` (each runner, what it serves, its machine, key, last seen and jobs running), `GET /api/v1/runners/{id}/jobs?before=&limit=` (its history, newest first) and `GET /api/v1/runners/queue`.

  <h2>Queue</h2>

  {% if queue.is_empty() %}<p class="mute small">Nothing running or waiting.</p>{% else %}

  <div class="table"><table>

  <tr><th>Repository</th><th>Run</th><th>Job</th><th>Status</th><th>Runner</th></tr>

  {% 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</span>{% else %}{{ q.runner }}{% endif %}</td></tr>{% endfor %}

  </table></div>

  <p class="small mute">Queued jobs are in the order runners claim them; a job waits for its dependencies and for a runner whose scope, tier and room fit it.</p>

  {% endif %}

</section>

<section class="panel">