//! 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, } #[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, } 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 rows = list.into_iter().map(|r| RunnerRow { running: running.get(&r.id).copied().unwrap_or(0), r }).collect(); Ok(Page(RunnersPage { hub: app.config.base_url().to_string(), shell, runners: rows, pending, new_token, notice, error })) } /// `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, } /// `POST /admin/runners/token`: a one-time registration token, shown once. 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 token = format!("flowreg_{}", crate::auth::random_hex(24)); runners::add_registration(&app.db, &blake3::hash(token.as_bytes()).to_hex(), &me.username) .await .map_err(|e| Error::from_db(shell.clone(), e))?; accounts::audit(&app.db, Some(&me), "flow.runner.token", "", "registration token created").await.ok(); page(&app, shell, Some(token), None, None).await } #[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, 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); Ok(Page(RunnerPage { shell, r, running, history, 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(), } }