OXYGENHUB

Coverage / crates/hub/src/web/runners.rs 77.8%

383 of 492 lines run, 59 of 176 functions · run #518 on main · run missed

1
//! Admin → Runners (hub spec §13.2): the OxygenFlow runners of this hub,
2
//! what they are and are running, their history, registration tokens, and
3
//! draining, disabling and removing them.
4
5
use super::settings::csrf;
6
use super::{filters, Error, Page, Shell};
7
use crate::auth::Viewer;
8
use crate::db::{self, accounts, runners};
9
use crate::Shared;
10
use askama::Template;
11
use axum::extract::{Form, Path, Query, State};
12
use axum::response::{IntoResponse, Redirect, Response};
13
use serde::Deserialize;
14
use std::collections::HashMap;
15
1633×
fn require_admin(viewer: &Viewer, shell: &Shell) -> Result<accounts::Account, Error> {
1733×
    match viewer.account() {
1829×
        Some(a) if a.is_admin => Ok(a.clone()),
194×
        _ => Err(Error::not_found(shell.clone())),
20
    }
2133×
}
22
23
pub struct RunnerRow {
24
    pub r: runners::Runner,
25
    /// Jobs it is running now.
26
    pub running: i64,
27
    /// The repositories it serves, in words.
28
    pub scope: String,
29
    /// Where its peers reach it in the image swarm, and its site
30
    /// (oxygenhub#144); empty when it is not in one.
31
    pub swarm: String,
32
    /// The jobs it runs now: (repository, run, job, started).
33
    pub jobs: Vec<(String, i64, String, Option<i64>)>,
34
}
35
36
impl RunnerRow {
37
    /// What it is doing (oxygenhub#243): `disabled`, `draining`, `idle`,
38
    /// or `running N of M`.
3917×
    pub fn doing(&self) -> String {
4017×
        match self.r.state.as_str() {
4117×
            "disabled" => "disabled".into(),
4217×
            "draining" if self.running > 0 => format!("draining, {} left", self.running),
430×
            "draining" => "draining".into(),
4416×
            _ if self.running > 0 => format!("running {} of {}", self.running, self.r.max_jobs),
4511×
            _ => "idle".into(),
46
        }
4717×
    }
48
49
    /// The jobs it runs, for the status's tooltip.
506×
    pub fn jobs_line(&self) -> String {
516×
        let now = crate::db::now_ms();
526×
        self.jobs
536×
            .iter()
546×
            .map(|(repo, run, job, started)| match started {
556×
                Some(t) => format!("{repo} #{run} {job}, {} min", (now - t).max(0) / 60_000),
560×
                None => format!("{repo} #{run} {job}"),
576×
            })
586×
            .collect::<Vec<_>>()
596×
            .join("; ")
606×
    }
61
}
62
63
/// The queue's filters (oxygenhub#243), kept in the URL.
64
#[derive(Deserialize, Default, Clone)]
65
pub struct QueueFilter {
66
    #[serde(default)]
67
    pub repo: String,
68
    #[serde(default)]
69
    pub status: String,
70
    #[serde(default)]
71
    pub runner: String,
72
}
73
74
impl QueueFilter {
7520×
    fn keeps(&self, q: &db::flow::QueueEntry) -> bool {
7620×
        (self.repo.is_empty() || format!("{}/{}", q.owner, q.name) == self.repo)
7720×
            && (self.status.is_empty() || q.status == self.status)
7818×
            && (self.runner.is_empty() || q.runner == self.runner)
7920×
    }
80
8126×
    pub fn any(&self) -> bool {
8226×
        !(self.repo.is_empty() && self.status.is_empty() && self.runner.is_empty())
8326×
    }
84
}
85
86
/// A scope in words: the whole site, an owner's repositories, or one
87
/// repository.
8822×
async fn scope_words(app: &Shared, s: &runners::Scope) -> String {
8922×
    match s.repo_id {
900×
        Some(id) => match db::repos::get_by_id(&app.db, id).await.ok().flatten() {
910×
            Some(r) => format!("{}/{}", r.owner, r.name),
920×
            None => "a removed repository".into(),
93
        },
9422×
        None if s.owner.is_empty() => "the whole site".into(),
950×
        None => format!("{}'s repositories", s.owner),
96
    }
9722×
}
98
99
/// A scope as typed: empty for the whole site, an owner, or
100
/// `owner/repository`.
1015×
async fn parse_scope(app: &Shared, text: &str) -> Result<runners::Scope, String> {
1025×
    let text = text.trim().trim_matches('/');
1035×
    if text.is_empty() {
1045×
        return Ok(runners::Scope::default());
1050×
    }
1060×
    if let Some((owner, name)) = text.split_once('/') {
1070×
        return match db::repos::get(&app.db, owner, name).await {
1080×
            Ok(Some(r)) => Ok(runners::Scope { owner: r.owner.clone(), repo_id: Some(r.id) }),
1090×
            _ => Err(format!("There is no repository {text}.")),
110
        };
1110×
    }
1120×
    let user = accounts::account_by_name(&app.db, text).await.ok().flatten().is_some();
1130×
    let org = accounts::org_by_name(&app.db, text).await.ok().flatten().is_some();
1140×
    if user || org {
1150×
        Ok(runners::Scope { owner: text.to_string(), repo_id: None })
116
    } else {
1170×
        Err(format!("There is no user or organisation {text}."))
118
    }
1195×
}
120
121
#[derive(Template)]
122
#[template(path = "admin_runners.html")]
123
pub struct RunnersPage {
124
    shell: Shell,
125
    runners: Vec<RunnerRow>,
126
    /// Registration tokens waiting to be used: (created by, created, expires).
127
    pending: Vec<(String, i64, i64)>,
128
    /// A token just made, shown once, with the command that uses it.
129
    new_token: Option<String>,
130
    hub: String,
131
    notice: Option<String>,
132
    error: Option<String>,
133
    /// Running jobs, then queued ones in the order runners claim them,
134
    /// as the filter keeps them.
135
    queue: Vec<db::flow::QueueEntry>,
136
    /// How many there are, unfiltered.
137
    queue_total: usize,
138
    filter: QueueFilter,
139
    /// The repositories and runners in the queue, for the filter's choices.
140
    queue_repos: Vec<String>,
141
    queue_runners: Vec<String>,
142
}
143
1447×
async fn page(
1457×
    app: &Shared,
1467×
    shell: Shell,
1477×
    new_token: Option<String>,
1487×
    notice: Option<String>,
1497×
    error: Option<String>,
1507×
) -> Result<Page<RunnersPage>, Error> {
1517×
    page_filtered(app, shell, new_token, notice, error, QueueFilter::default()).await
1527×
}
153
15416×
async fn page_filtered(
15516×
    app: &Shared,
15616×
    shell: Shell,
15716×
    new_token: Option<String>,
15816×
    notice: Option<String>,
15916×
    error: Option<String>,
16016×
    filter: QueueFilter,
16116×
) -> Result<Page<RunnersPage>, Error> {
16216×
    let e = |e| Error::from_db(shell.clone(), e);
16316×
    let list = runners::list(&app.db).await.map_err(e)?;
16416×
    let running: HashMap<i64, i64> = runners::running(&app.db).await.map_err(e)?.into_iter().collect();
16516×
    let pending = runners::pending_registrations(&app.db).await.map_err(e)?;
16616×
    let all = db::flow::queue(&app.db).await.map_err(e)?;
16716×
    let mut rows = Vec::with_capacity(list.len());
16817×
    for r in list {
16917×
        let scope = scope_words(app, &r.scope()).await;
17017×
        let swarm = app
17117×
            .swarm
17217×
            .addr(r.id)
17317×
            .map(|s| if s.site.is_empty() { s.addr } else { format!("{} · site {}", s.addr, s.site) })
17417×
            .unwrap_or_default();
17517×
        let jobs = all
17617×
            .iter()
17738×
            .filter(|q| q.status == "running" && q.runner == r.name)
17817×
            .map(|q| (format!("{}/{}", q.owner, q.name), q.run_number, q.job.clone(), q.started_ms))
17917×
            .collect();
18017×
        rows.push(RunnerRow { running: running.get(&r.id).copied().unwrap_or(0), scope, swarm, jobs, r });
181
    }
18220×
    let mut queue_repos: Vec<String> = all.iter().map(|q| format!("{}/{}", q.owner, q.name)).collect();
18316×
    queue_repos.sort();
18416×
    queue_repos.dedup();
18520×
    let mut queue_runners: Vec<String> = all.iter().filter(|q| !q.runner.is_empty()).map(|q| q.runner.clone()).collect();
18616×
    queue_runners.sort();
18716×
    queue_runners.dedup();
18816×
    let queue_total = all.len();
18920×
    let queue = all.into_iter().filter(|q| filter.keeps(q)).collect();
19016×
    Ok(Page(RunnersPage {
19116×
        hub: app.config.base_url().to_string(),
19216×
        shell,
19316×
        runners: rows,
19416×
        pending,
19516×
        new_token,
19616×
        notice,
19716×
        error,
19816×
        queue,
19916×
        queue_total,
20016×
        filter,
20116×
        queue_repos,
20216×
        queue_runners,
20316×
    }))
20416×
}
205
206
/// `GET /admin/runners`.
20711×
pub async fn list(State(app): State<Shared>, viewer: Viewer, Query(filter): Query<QueueFilter>) -> Result<Page<RunnersPage>, Error> {
20811×
    let shell = Shell::new(&app, "/admin/runners", &viewer);
20911×
    require_admin(&viewer, &shell)?;
2109×
    page_filtered(&app, shell, None, None, None, filter).await
21111×
}
212
213
#[derive(Deserialize)]
214
pub struct CsrfOnly {
215
    #[serde(rename = "_csrf")]
216
    csrf_token: String,
217
}
218
219
#[derive(Deserialize)]
220
pub struct ScopeForm {
221
    #[serde(rename = "_csrf")]
222
    csrf_token: String,
223
    /// Empty for the whole site, an owner, or `owner/repository`.
224
    #[serde(default)]
225
    scope: String,
226
}
227
228
/// `POST /admin/runners/token`: a one-time registration token, shown once,
229
/// for a runner of the scope given.
2305×
pub async fn new_token(State(app): State<Shared>, viewer: Viewer, Form(f): Form<ScopeForm>) -> Result<Page<RunnersPage>, Error> {
2315×
    let shell = Shell::new(&app, "/admin/runners", &viewer);
2325×
    let me = require_admin(&viewer, &shell)?;
2335×
    csrf(&viewer, &f.csrf_token, &shell)?;
2345×
    let scope = match parse_scope(&app, &f.scope).await {
2355×
        Ok(s) => s,
2360×
        Err(e) => return page(&app, shell, None, None, Some(e)).await,
237
    };
2385×
    let token = format!("flowreg_{}", crate::auth::random_hex(24));
2395×
    runners::add_registration(&app.db, &blake3::hash(token.as_bytes()).to_hex(), &me.username, &scope)
2405×
        .await
2415×
        .map_err(|e| Error::from_db(shell.clone(), e))?;
2425×
    let words = scope_words(&app, &scope).await;
2435×
    accounts::audit(&app.db, Some(&me), "flow.runner.token", "", &format!("registration token created for {words}")).await.ok();
2445×
    page(&app, shell, Some(token), Some(format!("The runner it registers will take jobs from {words}.")), None).await
2455×
}
246
247
/// `POST /admin/runners/{id}/scope`: change the repositories a runner serves.
2480×
pub async fn set_scope(
2490×
    State(app): State<Shared>,
2500×
    viewer: Viewer,
2510×
    Path(id): Path<i64>,
2520×
    Form(f): Form<ScopeForm>,
2530×
) -> Result<Page<RunnersPage>, Error> {
2540×
    let shell = Shell::new(&app, "/admin/runners", &viewer);
2550×
    let me = require_admin(&viewer, &shell)?;
2560×
    csrf(&viewer, &f.csrf_token, &shell)?;
2570×
    let r =
2580×
        runners::get(&app.db, id).await.map_err(|e| Error::from_db(shell.clone(), e))?.ok_or_else(|| Error::not_found(shell.clone()))?;
2590×
    let scope = match parse_scope(&app, &f.scope).await {
2600×
        Ok(s) => s,
2610×
        Err(e) => return page(&app, shell, None, None, Some(e)).await,
262
    };
2630×
    runners::set_scope(&app.db, id, &scope).await.map_err(|e| Error::from_db(shell.clone(), e))?;
2640×
    let words = scope_words(&app, &scope).await;
2650×
    accounts::audit(&app.db, Some(&me), "flow.runner.scope", &r.name, &words).await.ok();
2660×
    page(&app, shell, None, Some(format!("{} now takes jobs from {words}.", r.name)), None).await
2670×
}
268
269
#[derive(Deserialize)]
270
pub struct LimitsForm {
271
    #[serde(rename = "_csrf")]
272
    csrf_token: String,
273
    /// Comma-separated.
274
    #[serde(default)]
275
    labels: String,
276
    /// Minutes; empty for the site's limit.
277
    #[serde(default)]
278
    timeout_minutes: String,
279
}
280
281
/// `POST /admin/runners/{id}/limits`: the labels an administrator gives a
282
/// runner and its own job time limit (#98).
2832×
pub async fn set_limits(
2842×
    State(app): State<Shared>,
2852×
    viewer: Viewer,
2862×
    Path(id): Path<i64>,
2872×
    Form(f): Form<LimitsForm>,
2882×
) -> Result<Response, Error> {
2892×
    let shell = Shell::new(&app, &format!("/admin/runners/{id}"), &viewer);
2902×
    let me = require_admin(&viewer, &shell)?;
2912×
    csrf(&viewer, &f.csrf_token, &shell)?;
2922×
    let r =
2932×
        runners::get(&app.db, id).await.map_err(|e| Error::from_db(shell.clone(), e))?.ok_or_else(|| Error::not_found(shell.clone()))?;
2942×
    let mut labels: Vec<String> = Vec::new();
2955×
    for l in f.labels.split([',', ' ']).map(str::trim).filter(|l| !l.is_empty()) {
29616×
        if !l.chars().all(|c| c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '.')) || l.len() > 40 {
2971×
            return Err(Error::bad(shell, format!("label {l:?}: letters, digits, - _ and . only, up to 40 characters")));
2983×
        }
2993×
        if !labels.iter().any(|x| x == l) {
3003×
            labels.push(l.to_string());
3013×
        }
302
    }
3031×
    let timeout = match f.timeout_minutes.trim() {
3041×
        "" => None,
3051×
        t => match t.parse::<i64>() {
3061×
            Ok(m) if (1..=24 * 60).contains(&m) => Some(m),
3070×
            _ => return Err(Error::bad(shell, "a time limit is a whole number of minutes from 1 to 1440")),
308
        },
309
    };
3101×
    runners::set_limits(&app.db, id, &labels.join(","), timeout).await.map_err(|e| Error::from_db(shell.clone(), e))?;
3111×
    let words = format!(
312
        "labels {}; time limit {}",
3131×
        if labels.is_empty() { "none".to_string() } else { labels.join(", ") },
3141×
        timeout.map(|m| format!("{m} min")).unwrap_or_else(|| "the site's".into())
315
    );
3161×
    accounts::audit(&app.db, Some(&me), "flow.runner.limits", &r.name, &words).await.ok();
3171×
    Ok(Redirect::to(&format!("/admin/runners/{id}")).into_response())
3182×
}
319
320
#[derive(Deserialize)]
321
pub struct StateForm {
322
    #[serde(rename = "_csrf")]
323
    csrf_token: String,
324
    /// `active`, `draining` or `disabled`.
325
    state: String,
326
}
327
328
/// `POST /admin/runners/{id}/state`: drain, disable or resume a runner.
3292×
pub async fn set_state(
3302×
    State(app): State<Shared>,
3312×
    viewer: Viewer,
3322×
    Path(id): Path<i64>,
3332×
    Form(f): Form<StateForm>,
3342×
) -> Result<Response, Error> {
3352×
    let shell = Shell::new(&app, "/admin/runners", &viewer);
3362×
    let me = require_admin(&viewer, &shell)?;
3372×
    csrf(&viewer, &f.csrf_token, &shell)?;
3382×
    if !matches!(f.state.as_str(), "active" | "draining" | "disabled") {
3390×
        return Err(Error::bad(shell, "a runner is active, draining or disabled"));
3402×
    }
3412×
    let r =
3422×
        runners::get(&app.db, id).await.map_err(|e| Error::from_db(shell.clone(), e))?.ok_or_else(|| Error::not_found(shell.clone()))?;
3432×
    runners::set_state(&app.db, id, &f.state).await.map_err(|e| Error::from_db(shell.clone(), e))?;
3442×
    accounts::audit(&app.db, Some(&me), "flow.runner.state", &r.name, &f.state).await.ok();
3452×
    Ok(Redirect::to("/admin/runners").into_response())
3462×
}
347
348
/// `POST /admin/runners/{id}/remove`: forget a runner that runs nothing.
3492×
pub async fn remove(
3502×
    State(app): State<Shared>,
3512×
    viewer: Viewer,
3522×
    Path(id): Path<i64>,
3532×
    Form(f): Form<CsrfOnly>,
3542×
) -> Result<Page<RunnersPage>, Error> {
3552×
    let shell = Shell::new(&app, "/admin/runners", &viewer);
3562×
    let me = require_admin(&viewer, &shell)?;
3572×
    csrf(&viewer, &f.csrf_token, &shell)?;
3582×
    let e = |e| Error::from_db(shell.clone(), e);
3592×
    let r = runners::get(&app.db, id).await.map_err(e)?.ok_or_else(|| Error::not_found(shell.clone()))?;
3602×
    let busy = runners::running(&app.db).await.map_err(e)?.into_iter().any(|(rid, n)| rid == id && n > 0);
3612×
    if busy {
3621×
        return page(
3631×
            &app,
3641×
            shell,
3651×
            None,
3661×
            None,
3671×
            Some(format!("{} is running jobs: drain it first, then remove it when they finish.", r.name)),
3681×
        )
3691×
        .await;
3701×
    }
3711×
    runners::remove(&app.db, id).await.map_err(e)?;
3721×
    accounts::audit(&app.db, Some(&me), "flow.runner.remove", &r.name, "").await.ok();
3731×
    let also = if r.source == "config" {
3740×
        " It is configured in hub.toml; remove its token there too, or it comes back on its next claim."
375
    } else {
3761×
        ""
377
    };
3781×
    page(&app, shell, None, Some(format!("Removed {}.{also}", r.name)), None).await
3792×
}
380
381
#[derive(Template)]
382
#[template(path = "admin_runner.html")]
383
pub struct RunnerPage {
384
    shell: Shell,
385
    r: runners::Runner,
386
    running: i64,
387
    history: Vec<runners::RunnerJob>,
388
    declines: Vec<crate::db::flow::RunnerDecline>,
389
    older: Option<i64>,
390
    jobs: i64,
391
    passed: i64,
392
    busy_ms: i64,
393
}
394
395
impl RunnerPage {
396
    /// Time spent running jobs, as pages show durations.
3973×
    pub fn busy(&self) -> String {
3983×
        crate::flow::human_ms(self.busy_ms)
3993×
    }
400
}
401
402
#[derive(Deserialize, Default)]
403
pub struct HistoryQuery {
404
    before: Option<i64>,
405
}
406
407
const PER_PAGE: i64 = 50;
408
409
/// `GET /admin/runners/{id}`: a runner and the jobs it ran.
4103×
pub async fn show(
4113×
    State(app): State<Shared>,
4123×
    viewer: Viewer,
4133×
    Path(id): Path<i64>,
4143×
    Query(q): Query<HistoryQuery>,
4153×
) -> Result<Page<RunnerPage>, Error> {
4163×
    let shell = Shell::new(&app, &format!("/admin/runners/{id}"), &viewer);
4173×
    require_admin(&viewer, &shell)?;
4183×
    let e = |e| Error::from_db(shell.clone(), e);
4193×
    let r = runners::get(&app.db, id).await.map_err(e)?.ok_or_else(|| Error::not_found(shell.clone()))?;
4203×
    let mut history = runners::history(&app.db, id, PER_PAGE + 1, q.before).await.map_err(e)?;
4213×
    let older = (history.len() as i64 > PER_PAGE).then(|| history[PER_PAGE as usize - 1].started_ms).flatten();
4223×
    history.truncate(PER_PAGE as usize);
4233×
    let (jobs, passed, busy_ms) = runners::totals(&app.db, id).await.map_err(e)?;
4243×
    let running = runners::running(&app.db).await.map_err(e)?.into_iter().find(|(rid, _)| *rid == id).map(|(_, n)| n).unwrap_or(0);
4253×
    let declines = crate::db::flow::declines_of_runner(&app.db, id, 20).await.map_err(e)?;
4263×
    Ok(Page(RunnerPage { shell, r, running, history, declines, older, jobs, passed, busy_ms }))
4273×
}
428
4293×
pub fn duration(j: &runners::RunnerJob) -> String {
4303×
    match (j.started_ms, j.finished_ms) {
4310×
        (Some(a), Some(b)) => crate::flow::human_ms(b - a),
4323×
        (Some(a), None) => crate::flow::human_ms(db::now_ms() - a),
4330×
        _ => "–".into(),
434
    }
4353×
}
436
437
// ---------------------------------------------------------------- settings
438
439
#[derive(Template)]
440
#[template(path = "admin_runner_settings.html")]
441
pub struct SettingsPage {
442
    shell: Shell,
443
    default_m: u64,
444
    max_m: u64,
445
    grace_m: i64,
446
    /// What hub.toml says, shown beside a setting the page overrides.
447
    file_default_m: u64,
448
    flow_repos: Vec<String>,
449
    /// The site's priority for jobs (oxygenhub#244): low, normal or high.
450
    priority: String,
451
    /// The most expedited runs one owner may have going at once.
452
    expedite_cap: i64,
453
    notice: Option<String>,
454
    error: Option<String>,
455
}
456
4574×
async fn settings_page(app: &Shared, shell: Shell, notice: Option<String>, error: Option<String>) -> Page<SettingsPage> {
4584×
    let l = crate::flow::limits(app).await;
459
    Page(SettingsPage {
4604×
        shell,
4614×
        default_m: l.default_s / 60,
4624×
        max_m: l.max_s / 60,
4634×
        grace_m: l.grace_ms / 60_000,
4644×
        file_default_m: app.config.flow.job_timeout_minutes,
4654×
        flow_repos: app.config.flow.repos.clone(),
4664×
        expedite_cap: accounts::setting(&app.db, "flow_expedite_cap").await.ok().flatten().and_then(|v| v.parse().ok()).unwrap_or(2),
4674×
        priority: accounts::setting(&app.db, "flow_priority")
4684×
            .await
4694×
            .ok()
4704×
            .flatten()
4714×
            .filter(|p| !p.is_empty())
4724×
            .unwrap_or_else(|| "normal".into()),
4734×
        notice,
4744×
        error,
475
    })
4764×
}
477
478
#[derive(Deserialize)]
479
pub struct PriorityForm {
480
    #[serde(rename = "_csrf")]
481
    csrf_token: String,
482
    priority: String,
483
    #[serde(default)]
484
    expedite_cap: Option<String>,
485
}
486
487
/// `POST /admin/runners/priority` (oxygenhub#244): the level every job has
488
/// unless its organisation, repository or run says otherwise.
4890×
pub async fn save_priority(State(app): State<Shared>, viewer: Viewer, Form(f): Form<PriorityForm>) -> Result<Page<SettingsPage>, Error> {
4900×
    let shell = Shell::new(&app, "/admin/runners/settings", &viewer);
4910×
    let me = require_admin(&viewer, &shell)?;
4920×
    csrf(&viewer, &f.csrf_token, &shell)?;
4930×
    if !["low", "normal", "high"].contains(&f.priority.as_str()) {
4940×
        return Ok(settings_page(&app, shell, None, Some("Pick low, normal or high (urgent is for expediting one run).".into())).await);
4950×
    }
4960×
    let cap = match f.expedite_cap.as_deref().map(str::trim) {
4970×
        None | Some("") => None,
4980×
        Some(v) => match v.parse::<i64>() {
4990×
            Ok(n) if (0..=100).contains(&n) => Some(n),
5000×
            _ => return Ok(settings_page(&app, shell, None, Some("The cap is a whole number from 0 to 100.".into())).await),
501
        },
502
    };
5030×
    accounts::set_setting(&app.db, "flow_priority", &f.priority).await.map_err(|e| Error::from_db(shell.clone(), e))?;
5040×
    if let Some(n) = cap {
5050×
        accounts::set_setting(&app.db, "flow_expedite_cap", &n.to_string()).await.map_err(|e| Error::from_db(shell.clone(), e))?;
5060×
    }
5070×
    accounts::audit(&app.db, Some(&me), "flow.priority", "site", &f.priority).await.ok();
5080×
    Ok(settings_page(
5090×
        &app,
5100×
        shell,
5110×
        Some(format!("Saved: jobs are {} unless their organisation, repository or run says otherwise.", f.priority)),
5120×
        None,
5130×
    )
5140×
    .await)
5150×
}
516
517
/// `GET /admin/runners/settings` (oxygenhub#89).
5183×
pub async fn settings(State(app): State<Shared>, viewer: Viewer) -> Result<Page<SettingsPage>, Error> {
5193×
    let shell = Shell::new(&app, "/admin/runners/settings", &viewer);
5203×
    require_admin(&viewer, &shell)?;
5212×
    Ok(settings_page(&app, shell, None, None).await)
5223×
}
523
524
#[derive(Deserialize)]
525
pub struct SettingsForm {
526
    #[serde(rename = "_csrf")]
527
    csrf_token: String,
528
    default_minutes: String,
529
    max_minutes: String,
530
    grace_minutes: String,
531
}
532
533
/// `POST /admin/runners/settings`: the site's job time limits.
5342×
pub async fn save_settings(State(app): State<Shared>, viewer: Viewer, Form(f): Form<SettingsForm>) -> Result<Page<SettingsPage>, Error> {
5352×
    let shell = Shell::new(&app, "/admin/runners/settings", &viewer);
5362×
    let me = require_admin(&viewer, &shell)?;
5372×
    csrf(&viewer, &f.csrf_token, &shell)?;
5386×
    let num = |v: &str| v.trim().parse::<u64>().ok().filter(|n| *n > 0);
5392×
    let (Some(d), Some(m), Some(g)) = (num(&f.default_minutes), num(&f.max_minutes), num(&f.grace_minutes)) else {
5400×
        return Ok(settings_page(&app, shell, None, Some("Each limit is a whole number of minutes, at least 1.".into())).await);
541
    };
5422×
    if d > m {
5431×
        return Ok(settings_page(&app, shell, None, Some("The default cannot be longer than the maximum.".into())).await);
5441×
    }
5451×
    if m > 24 * 60 {
5460×
        return Ok(settings_page(&app, shell, None, Some("A job may run for a day at most (1440 minutes).".into())).await);
5471×
    }
5481×
    let e = |e| Error::from_db(shell.clone(), e);
5491×
    accounts::set_setting(&app.db, crate::flow::SETTING_DEFAULT, &d.to_string()).await.map_err(e)?;
5501×
    accounts::set_setting(&app.db, crate::flow::SETTING_MAX, &m.to_string()).await.map_err(e)?;
5511×
    accounts::set_setting(&app.db, crate::flow::SETTING_GRACE, &g.to_string()).await.map_err(e)?;
5521×
    accounts::audit(
5531×
        &app.db,
5541×
        Some(&me),
5551×
        "flow.settings",
5561×
        "",
5571×
        &format!("job time limit {d} min by default, {m} min at most; lost after {g} min more"),
5581×
    )
5591×
    .await
5601×
    .ok();
5611×
    Ok(settings_page(&app, shell, Some("Saved. Jobs claimed from now on run under these limits.".into()), None).await)
5622×
}
563
564
/// Where one image is: the hub, a runner holding it, or a runner fetching
565
/// it from the swarm now.
566
pub struct Place {
567
    pub name: String,
568
    /// `hub`, `holds` or `fetching`.
569
    pub kind: &'static str,
570
    pub title: String,
571
}
572
573
pub struct ImageRow {
574
    pub hash: String,
575
    pub arch: String,
576
    /// Its size, when the hub keeps it (runners do not say).
577
    pub size: Option<i64>,
578
    pub places: Vec<Place>,
579
    /// Who sent it to the hub, and when.
580
    pub sent: Option<(String, i64)>,
581
    /// Repositories whose jobs ran on it in the last 90 days.
582
    pub uses: Vec<(String, i64, i64)>,
583
    pub last_used: i64,
584
}
585
586
impl ImageRow {
587
    /// Nowhere a job could get it: not on the hub, on no runner.
588
    /// Nowhere a job could get it: nobody holds it, and the hub has at most
589
    /// its manifest (oxygenflow#94).
5908×
    pub fn stranded(&self) -> bool {
5918×
        self.places.iter().all(|p| p.kind == "peer")
5928×
    }
593
}
594
595
#[derive(Template)]
596
#[template(path = "admin_images.html")]
597
pub struct ImagesPage {
598
    shell: Shell,
599
    images: Vec<ImageRow>,
600
    on_hub: usize,
601
    /// Runners in the image swarm, as "name (site)".
602
    swarm: Vec<String>,
603
    /// The swarm backdrop's data (docs/backgrounds.md §2): the hub and the
604
    /// runners, and for each image who holds it and who fetches it now.
605
    backdrop: String,
606
}
607
608
/// The swarm's data, `{scene, nodes: [name], images: [{held, fetching}]}`,
609
/// the hub first, from the page's own rows.
6102×
fn swarm_json(images: &[ImageRow]) -> String {
6112×
    let mut nodes: Vec<String> = vec!["hub".into()];
6122×
    let mut list = Vec::new();
6134×
    for img in images {
6144×
        let (mut held, mut fetching) = (Vec::new(), Vec::new());
6154×
        for p in &img.places {
6163×
            let i = match p.kind {
6173×
                "hub" => 0,
6183×
                _ => match nodes.iter().position(|n| *n == p.name) {
6190×
                    Some(i) => i,
620
                    None => {
6212×
                        nodes.push(p.name.clone());
6222×
                        nodes.len() - 1
623
                    }
624
                },
625
            };
6263×
            if p.kind == "fetching" {
6270×
                fetching.push(i);
6283×
            } else {
6293×
                held.push(i);
6303×
            }
631
        }
6324×
        list.push(serde_json::json!({ "held": held, "fetching": fetching }));
633
    }
6342×
    serde_json::json!({ "scene": "swarm", "nodes": nodes, "images": list }).to_string().replace('<', "\\u003c")
6352×
}
636
637
/// `GET /admin/runners/images` (C0, docs/cylinder.md): every runner image,
638
/// where it is and what uses it.
6393×
pub async fn images(State(app): State<Shared>, viewer: Viewer) -> Result<Page<ImagesPage>, Error> {
6403×
    let shell = Shell::new(&app, "/admin/runners/images", &viewer);
6413×
    require_admin(&viewer, &shell)?;
6422×
    let e = |e| Error::from_db(shell.clone(), e);
6432×
    let kept = crate::images::kept(&app.db).await.map_err(e)?;
6442×
    let uses = crate::images::uses(&app.db, db::now_ms() - 90 * 24 * 3600 * 1000).await.map_err(e)?;
6452×
    let archs = crate::images::reported_archs(&app.db).await.map_err(e)?;
6462×
    let list = runners::list(&app.db).await.map_err(e)?;
6472×
    let fetching = app.swarm.fetching_now();
6482×
    let mut rows: HashMap<String, ImageRow> = HashMap::new();
6494×
    let row = |rows: &mut HashMap<String, ImageRow>, hash: &str| -> () {
6504×
        rows.entry(hash.to_ascii_lowercase()).or_insert_with(|| ImageRow {
6514×
            hash: hash.to_ascii_lowercase(),
6524×
            arch: archs.get(&hash.to_ascii_lowercase()).cloned().unwrap_or_default(),
6534×
            size: None,
6544×
            places: Vec::new(),
6554×
            sent: None,
6564×
            uses: Vec::new(),
657
            last_used: 0,
6584×
        });
6594×
    };
6602×
    for k in &kept {
6612×
        row(&mut rows, &k.hash);
6622×
        let r = rows.get_mut(&k.hash.to_ascii_lowercase()).expect("added");
6632×
        r.size = Some(k.size);
6642×
        if r.arch.is_empty() {
6651×
            r.arch = k.arch.clone();
6661×
        }
6672×
        r.sent = Some((k.uploader.clone(), k.created_ms));
668
        // An image only runners hold (oxygenflow#94): the hub keeps its
669
        // manifest, and the runners holding it serve its chunks.
6702×
        r.places.push(if k.source == "peer" {
6711×
            Place {
6721×
                name: "manifest".into(),
6731×
                kind: "peer",
6741×
                title: format!("The hub keeps only its manifest, sent by {}; runners holding it serve its {} chunks", k.holder, k.chunks),
6751×
            }
676
        } else {
6771×
            Place { name: "hub".into(), kind: "hub", title: format!("{} chunks kept by the hub", k.chunks) }
678
        });
679
    }
6802×
    let mut swarm = Vec::new();
6812×
    for runner in &list {
6821×
        if let Some(a) = app.swarm.addr(runner.id) {
6831×
            swarm.push(if a.site.is_empty() { runner.name.clone() } else { format!("{} ({})", runner.name, a.site) });
6840×
        }
6851×
        for h in runner.image_list().unwrap_or_default() {
6861×
            row(&mut rows, &h);
6871×
            rows.get_mut(&h.to_ascii_lowercase()).expect("added").places.push(Place {
6881×
                name: runner.name.clone(),
6891×
                kind: "holds",
6901×
                title: format!("{} holds it ({})", runner.name, runner.status()),
6911×
            });
6921×
        }
693
    }
6942×
    for (image, who) in &fetching {
6950×
        row(&mut rows, image);
6960×
        for id in who {
6970×
            if let Some(runner) = list.iter().find(|r| r.id == *id) {
6980×
                rows.get_mut(image).expect("added").places.push(Place {
6990×
                    name: runner.name.clone(),
7000×
                    kind: "fetching",
7010×
                    title: format!("{} is fetching it from the swarm now", runner.name),
7020×
                });
7030×
            }
704
        }
705
    }
7062×
    for u in uses {
7071×
        if !jet::images::is_hash(&u.image) {
7080×
            continue;
7091×
        }
7101×
        row(&mut rows, &u.image);
7111×
        let r = rows.get_mut(&u.image).expect("added");
7121×
        r.last_used = r.last_used.max(u.last_ms);
7131×
        r.uses.push((format!("{}/{}", u.owner, u.name), u.jobs, u.last_ms));
714
    }
7152×
    let mut images: Vec<ImageRow> = rows.into_values().collect();
7166×
    images.sort_by(|a, b| b.last_used.cmp(&a.last_used).then_with(|| a.hash.cmp(&b.hash)));
7172×
    let on_hub = kept.len();
7182×
    let backdrop = swarm_json(&images);
7192×
    Ok(Page(ImagesPage { shell, images, on_hub, swarm, backdrop }))
7203×
}