//! Admin → Runners (hub spec §13.2): the OxygenFlow runners of this hub, //! what they are and are running, their history, registration tokens, and //! draining, disabling and removing them. use super::settings::csrf; use super::{filters, Error, Page, Shell}; use crate::auth::Viewer; use crate::db::{self, accounts, runners}; use crate::Shared; use askama::Template; use axum::extract::{Form, Path, Query, State}; use axum::response::{IntoResponse, Redirect, Response}; use serde::Deserialize; use std::collections::HashMap; fn require_admin(viewer: &Viewer, shell: &Shell) -> Result { match viewer.account() { Some(a) if a.is_admin => Ok(a.clone()), _ => Err(Error::not_found(shell.clone())), } } pub struct RunnerRow { pub r: runners::Runner, /// Jobs it is running now. pub running: i64, /// The repositories it serves, in words. pub scope: String, /// Where its peers reach it in the image swarm, and its site /// (oxygenhub#144); empty when it is not in one. pub swarm: String, } /// A scope in words: the whole site, an owner's repositories, or one /// repository. async fn scope_words(app: &Shared, s: &runners::Scope) -> String { match s.repo_id { Some(id) => match db::repos::get_by_id(&app.db, id).await.ok().flatten() { Some(r) => format!("{}/{}", r.owner, r.name), None => "a removed repository".into(), }, None if s.owner.is_empty() => "the whole site".into(), None => format!("{}'s repositories", s.owner), } } /// A scope as typed: empty for the whole site, an owner, or /// `owner/repository`. async fn parse_scope(app: &Shared, text: &str) -> Result { let text = text.trim().trim_matches('/'); if text.is_empty() { return Ok(runners::Scope::default()); } if let Some((owner, name)) = text.split_once('/') { return match db::repos::get(&app.db, owner, name).await { Ok(Some(r)) => Ok(runners::Scope { owner: r.owner.clone(), repo_id: Some(r.id) }), _ => Err(format!("There is no repository {text}.")), }; } let user = accounts::account_by_name(&app.db, text).await.ok().flatten().is_some(); let org = accounts::org_by_name(&app.db, text).await.ok().flatten().is_some(); if user || org { Ok(runners::Scope { owner: text.to_string(), repo_id: None }) } else { Err(format!("There is no user or organisation {text}.")) } } #[derive(Template)] #[template(path = "admin_runners.html")] pub struct RunnersPage { shell: Shell, runners: Vec, /// Registration tokens waiting to be used: (created by, created, expires). pending: Vec<(String, i64, i64)>, /// A token just made, shown once, with the command that uses it. new_token: Option, hub: String, notice: Option, error: Option, /// Running jobs, then queued ones in the order runners claim them. queue: Vec, } async fn page( app: &Shared, shell: Shell, new_token: Option, notice: Option, error: Option, ) -> Result, Error> { let e = |e| Error::from_db(shell.clone(), e); let list = runners::list(&app.db).await.map_err(e)?; let running: HashMap = runners::running(&app.db).await.map_err(e)?.into_iter().collect(); let pending = runners::pending_registrations(&app.db).await.map_err(e)?; let mut rows = Vec::with_capacity(list.len()); for r in list { let scope = scope_words(app, &r.scope()).await; let swarm = app .swarm .addr(r.id) .map(|s| if s.site.is_empty() { s.addr } else { format!("{} · site {}", s.addr, s.site) }) .unwrap_or_default(); rows.push(RunnerRow { running: running.get(&r.id).copied().unwrap_or(0), scope, swarm, r }); } 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 })) } /// `GET /admin/runners`. pub async fn list(State(app): State, viewer: Viewer) -> Result, Error> { let shell = Shell::new(&app, "/admin/runners", &viewer); require_admin(&viewer, &shell)?; page(&app, shell, None, None, None).await } #[derive(Deserialize)] pub struct CsrfOnly { #[serde(rename = "_csrf")] csrf_token: String, } #[derive(Deserialize)] pub struct ScopeForm { #[serde(rename = "_csrf")] csrf_token: String, /// Empty for the whole site, an owner, or `owner/repository`. #[serde(default)] scope: String, } /// `POST /admin/runners/token`: a one-time registration token, shown once, /// for a runner of the scope given. pub async fn new_token(State(app): State, viewer: Viewer, Form(f): Form) -> Result, Error> { let shell = Shell::new(&app, "/admin/runners", &viewer); let me = require_admin(&viewer, &shell)?; csrf(&viewer, &f.csrf_token, &shell)?; let scope = match parse_scope(&app, &f.scope).await { Ok(s) => s, Err(e) => return page(&app, shell, None, None, Some(e)).await, }; let token = format!("flowreg_{}", crate::auth::random_hex(24)); runners::add_registration(&app.db, &blake3::hash(token.as_bytes()).to_hex(), &me.username, &scope) .await .map_err(|e| Error::from_db(shell.clone(), e))?; let words = scope_words(&app, &scope).await; accounts::audit(&app.db, Some(&me), "flow.runner.token", "", &format!("registration token created for {words}")).await.ok(); page(&app, shell, Some(token), Some(format!("The runner it registers will take jobs from {words}.")), None).await } /// `POST /admin/runners/{id}/scope`: change the repositories a runner serves. pub async fn set_scope( State(app): State, viewer: Viewer, Path(id): Path, Form(f): Form, ) -> Result, Error> { let shell = Shell::new(&app, "/admin/runners", &viewer); let me = require_admin(&viewer, &shell)?; csrf(&viewer, &f.csrf_token, &shell)?; let r = runners::get(&app.db, id).await.map_err(|e| Error::from_db(shell.clone(), e))?.ok_or_else(|| Error::not_found(shell.clone()))?; let scope = match parse_scope(&app, &f.scope).await { Ok(s) => s, Err(e) => return page(&app, shell, None, None, Some(e)).await, }; runners::set_scope(&app.db, id, &scope).await.map_err(|e| Error::from_db(shell.clone(), e))?; let words = scope_words(&app, &scope).await; accounts::audit(&app.db, Some(&me), "flow.runner.scope", &r.name, &words).await.ok(); page(&app, shell, None, Some(format!("{} now takes jobs from {words}.", r.name)), None).await } #[derive(Deserialize)] pub struct LimitsForm { #[serde(rename = "_csrf")] csrf_token: String, /// Comma-separated. #[serde(default)] labels: String, /// Minutes; empty for the site's limit. #[serde(default)] timeout_minutes: String, } /// `POST /admin/runners/{id}/limits`: the labels an administrator gives a /// runner and its own job time limit (#98). pub async fn set_limits( State(app): State, viewer: Viewer, Path(id): Path, Form(f): Form, ) -> Result { let shell = Shell::new(&app, &format!("/admin/runners/{id}"), &viewer); let me = require_admin(&viewer, &shell)?; csrf(&viewer, &f.csrf_token, &shell)?; let r = runners::get(&app.db, id).await.map_err(|e| Error::from_db(shell.clone(), e))?.ok_or_else(|| Error::not_found(shell.clone()))?; let mut labels: Vec = Vec::new(); for l in f.labels.split([',', ' ']).map(str::trim).filter(|l| !l.is_empty()) { if !l.chars().all(|c| c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '.')) || l.len() > 40 { return Err(Error::bad(shell, format!("label {l:?}: letters, digits, - _ and . only, up to 40 characters"))); } if !labels.iter().any(|x| x == l) { labels.push(l.to_string()); } } let timeout = match f.timeout_minutes.trim() { "" => None, t => match t.parse::() { Ok(m) if (1..=24 * 60).contains(&m) => Some(m), _ => return Err(Error::bad(shell, "a time limit is a whole number of minutes from 1 to 1440")), }, }; runners::set_limits(&app.db, id, &labels.join(","), timeout).await.map_err(|e| Error::from_db(shell.clone(), e))?; let words = format!( "labels {}; time limit {}", if labels.is_empty() { "none".to_string() } else { labels.join(", ") }, timeout.map(|m| format!("{m} min")).unwrap_or_else(|| "the site's".into()) ); accounts::audit(&app.db, Some(&me), "flow.runner.limits", &r.name, &words).await.ok(); Ok(Redirect::to(&format!("/admin/runners/{id}")).into_response()) } #[derive(Deserialize)] pub struct StateForm { #[serde(rename = "_csrf")] csrf_token: String, /// `active`, `draining` or `disabled`. state: String, } /// `POST /admin/runners/{id}/state`: drain, disable or resume a runner. pub async fn set_state( State(app): State, viewer: Viewer, Path(id): Path, Form(f): Form, ) -> Result { let shell = Shell::new(&app, "/admin/runners", &viewer); let me = require_admin(&viewer, &shell)?; csrf(&viewer, &f.csrf_token, &shell)?; if !matches!(f.state.as_str(), "active" | "draining" | "disabled") { return Err(Error::bad(shell, "a runner is active, draining or disabled")); } let r = runners::get(&app.db, id).await.map_err(|e| Error::from_db(shell.clone(), e))?.ok_or_else(|| Error::not_found(shell.clone()))?; runners::set_state(&app.db, id, &f.state).await.map_err(|e| Error::from_db(shell.clone(), e))?; accounts::audit(&app.db, Some(&me), "flow.runner.state", &r.name, &f.state).await.ok(); Ok(Redirect::to("/admin/runners").into_response()) } /// `POST /admin/runners/{id}/remove`: forget a runner that runs nothing. pub async fn remove( State(app): State, viewer: Viewer, Path(id): Path, Form(f): Form, ) -> Result, Error> { let shell = Shell::new(&app, "/admin/runners", &viewer); let me = require_admin(&viewer, &shell)?; csrf(&viewer, &f.csrf_token, &shell)?; let e = |e| Error::from_db(shell.clone(), e); let r = runners::get(&app.db, id).await.map_err(e)?.ok_or_else(|| Error::not_found(shell.clone()))?; let busy = runners::running(&app.db).await.map_err(e)?.into_iter().any(|(rid, n)| rid == id && n > 0); if busy { return page( &app, shell, None, None, Some(format!("{} is running jobs: drain it first, then remove it when they finish.", r.name)), ) .await; } runners::remove(&app.db, id).await.map_err(e)?; accounts::audit(&app.db, Some(&me), "flow.runner.remove", &r.name, "").await.ok(); let also = if r.source == "config" { " It is configured in hub.toml; remove its token there too, or it comes back on its next claim." } else { "" }; page(&app, shell, None, Some(format!("Removed {}.{also}", r.name)), None).await } #[derive(Template)] #[template(path = "admin_runner.html")] pub struct RunnerPage { shell: Shell, r: runners::Runner, running: i64, history: Vec, declines: Vec, older: Option, jobs: i64, passed: i64, busy_ms: i64, } impl RunnerPage { /// Time spent running jobs, as pages show durations. pub fn busy(&self) -> String { crate::flow::human_ms(self.busy_ms) } } #[derive(Deserialize, Default)] pub struct HistoryQuery { before: Option, } const PER_PAGE: i64 = 50; /// `GET /admin/runners/{id}`: a runner and the jobs it ran. pub async fn show( State(app): State, viewer: Viewer, Path(id): Path, Query(q): Query, ) -> Result, Error> { let shell = Shell::new(&app, &format!("/admin/runners/{id}"), &viewer); require_admin(&viewer, &shell)?; let e = |e| Error::from_db(shell.clone(), e); let r = runners::get(&app.db, id).await.map_err(e)?.ok_or_else(|| Error::not_found(shell.clone()))?; let mut history = runners::history(&app.db, id, PER_PAGE + 1, q.before).await.map_err(e)?; let older = (history.len() as i64 > PER_PAGE).then(|| history[PER_PAGE as usize - 1].started_ms).flatten(); history.truncate(PER_PAGE as usize); let (jobs, passed, busy_ms) = runners::totals(&app.db, id).await.map_err(e)?; let running = runners::running(&app.db).await.map_err(e)?.into_iter().find(|(rid, _)| *rid == id).map(|(_, n)| n).unwrap_or(0); let declines = crate::db::flow::declines_of_runner(&app.db, id, 20).await.map_err(e)?; Ok(Page(RunnerPage { shell, r, running, history, declines, older, jobs, passed, busy_ms })) } pub fn duration(j: &runners::RunnerJob) -> String { match (j.started_ms, j.finished_ms) { (Some(a), Some(b)) => crate::flow::human_ms(b - a), (Some(a), None) => crate::flow::human_ms(db::now_ms() - a), _ => "–".into(), } } // ---------------------------------------------------------------- settings #[derive(Template)] #[template(path = "admin_runner_settings.html")] pub struct SettingsPage { shell: Shell, default_m: u64, max_m: u64, grace_m: i64, /// What hub.toml says, shown beside a setting the page overrides. file_default_m: u64, flow_repos: Vec, notice: Option, error: Option, } async fn settings_page(app: &Shared, shell: Shell, notice: Option, error: Option) -> Page { let l = crate::flow::limits(app).await; Page(SettingsPage { shell, default_m: l.default_s / 60, max_m: l.max_s / 60, grace_m: l.grace_ms / 60_000, file_default_m: app.config.flow.job_timeout_minutes, flow_repos: app.config.flow.repos.clone(), notice, error, }) } /// `GET /admin/runners/settings` (oxygenhub#89). pub async fn settings(State(app): State, viewer: Viewer) -> Result, Error> { let shell = Shell::new(&app, "/admin/runners/settings", &viewer); require_admin(&viewer, &shell)?; Ok(settings_page(&app, shell, None, None).await) } #[derive(Deserialize)] pub struct SettingsForm { #[serde(rename = "_csrf")] csrf_token: String, default_minutes: String, max_minutes: String, grace_minutes: String, } /// `POST /admin/runners/settings`: the site's job time limits. pub async fn save_settings(State(app): State, viewer: Viewer, Form(f): Form) -> Result, Error> { let shell = Shell::new(&app, "/admin/runners/settings", &viewer); let me = require_admin(&viewer, &shell)?; csrf(&viewer, &f.csrf_token, &shell)?; let num = |v: &str| v.trim().parse::().ok().filter(|n| *n > 0); let (Some(d), Some(m), Some(g)) = (num(&f.default_minutes), num(&f.max_minutes), num(&f.grace_minutes)) else { return Ok(settings_page(&app, shell, None, Some("Each limit is a whole number of minutes, at least 1.".into())).await); }; if d > m { return Ok(settings_page(&app, shell, None, Some("The default cannot be longer than the maximum.".into())).await); } if m > 24 * 60 { return Ok(settings_page(&app, shell, None, Some("A job may run for a day at most (1440 minutes).".into())).await); } let e = |e| Error::from_db(shell.clone(), e); accounts::set_setting(&app.db, crate::flow::SETTING_DEFAULT, &d.to_string()).await.map_err(e)?; accounts::set_setting(&app.db, crate::flow::SETTING_MAX, &m.to_string()).await.map_err(e)?; accounts::set_setting(&app.db, crate::flow::SETTING_GRACE, &g.to_string()).await.map_err(e)?; accounts::audit( &app.db, Some(&me), "flow.settings", "", &format!("job time limit {d} min by default, {m} min at most; lost after {g} min more"), ) .await .ok(); Ok(settings_page(&app, shell, Some("Saved. Jobs claimed from now on run under these limits.".into()), None).await) } /// 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, pub places: Vec, /// 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, on_hub: usize, /// Runners in the image swarm, as "name (site)". swarm: Vec, } /// `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, viewer: Viewer) -> Result, 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 = HashMap::new(); let row = |rows: &mut HashMap, 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 = 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 })) }