OXYGENHUB
# 38833ed78707 oxygenhub / crates / hub / src / web / runners.rs
7.2 KiB · Rust · last changed Runners: registration, Admin -> Runners, history, drain/disable/remove 1 month ago source blame history raw
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
16
fn require_admin(viewer: &Viewer, shell: &Shell) -> Result<accounts::Account, Error> {
17
    match viewer.account() {
18
        Some(a) if a.is_admin => Ok(a.clone()),
19
        _ => Err(Error::not_found(shell.clone())),
20
    }
21
}
22
23
pub struct RunnerRow {
24
    pub r: runners::Runner,
25
    /// Jobs it is running now.
26
    pub running: i64,
27
}
28
29
#[derive(Template)]
30
#[template(path = "admin_runners.html")]
31
pub struct RunnersPage {
32
    shell: Shell,
33
    runners: Vec<RunnerRow>,
34
    /// Registration tokens waiting to be used: (created by, created, expires).
35
    pending: Vec<(String, i64, i64)>,
36
    /// A token just made, shown once, with the command that uses it.
37
    new_token: Option<String>,
38
    hub: String,
39
    notice: Option<String>,
40
    error: Option<String>,
41
}
42
43
async fn page(
44
    app: &Shared,
45
    shell: Shell,
46
    new_token: Option<String>,
47
    notice: Option<String>,
48
    error: Option<String>,
49
) -> Result<Page<RunnersPage>, Error> {
50
    let e = |e| Error::from_db(shell.clone(), e);
51
    let list = runners::list(&app.db).await.map_err(e)?;
52
    let running: HashMap<i64, i64> = runners::running(&app.db).await.map_err(e)?.into_iter().collect();
53
    let pending = runners::pending_registrations(&app.db).await.map_err(e)?;
54
    let rows = list.into_iter().map(|r| RunnerRow { running: running.get(&r.id).copied().unwrap_or(0), r }).collect();
55
    Ok(Page(RunnersPage { hub: app.config.base_url().to_string(), shell, runners: rows, pending, new_token, notice, error }))
56
}
57
58
/// `GET /admin/runners`.
59
pub async fn list(State(app): State<Shared>, viewer: Viewer) -> Result<Page<RunnersPage>, Error> {
60
    let shell = Shell::new(&app, "/admin/runners", &viewer);
61
    require_admin(&viewer, &shell)?;
62
    page(&app, shell, None, None, None).await
63
}
64
65
#[derive(Deserialize)]
66
pub struct CsrfOnly {
67
    #[serde(rename = "_csrf")]
68
    csrf_token: String,
69
}
70
71
/// `POST /admin/runners/token`: a one-time registration token, shown once.
72
pub async fn new_token(State(app): State<Shared>, viewer: Viewer, Form(f): Form<CsrfOnly>) -> Result<Page<RunnersPage>, Error> {
73
    let shell = Shell::new(&app, "/admin/runners", &viewer);
74
    let me = require_admin(&viewer, &shell)?;
75
    csrf(&viewer, &f.csrf_token, &shell)?;
76
    let token = format!("flowreg_{}", crate::auth::random_hex(24));
77
    runners::add_registration(&app.db, &blake3::hash(token.as_bytes()).to_hex(), &me.username)
78
        .await
79
        .map_err(|e| Error::from_db(shell.clone(), e))?;
80
    accounts::audit(&app.db, Some(&me), "flow.runner.token", "", "registration token created").await.ok();
81
    page(&app, shell, Some(token), None, None).await
82
}
83
84
#[derive(Deserialize)]
85
pub struct StateForm {
86
    #[serde(rename = "_csrf")]
87
    csrf_token: String,
88
    /// `active`, `draining` or `disabled`.
89
    state: String,
90
}
91
92
/// `POST /admin/runners/{id}/state`: drain, disable or resume a runner.
93
pub async fn set_state(
94
    State(app): State<Shared>,
95
    viewer: Viewer,
96
    Path(id): Path<i64>,
97
    Form(f): Form<StateForm>,
98
) -> Result<Response, Error> {
99
    let shell = Shell::new(&app, "/admin/runners", &viewer);
100
    let me = require_admin(&viewer, &shell)?;
101
    csrf(&viewer, &f.csrf_token, &shell)?;
102
    if !matches!(f.state.as_str(), "active" | "draining" | "disabled") {
103
        return Err(Error::bad(shell, "a runner is active, draining or disabled"));
104
    }
105
    let r =
106
        runners::get(&app.db, id).await.map_err(|e| Error::from_db(shell.clone(), e))?.ok_or_else(|| Error::not_found(shell.clone()))?;
107
    runners::set_state(&app.db, id, &f.state).await.map_err(|e| Error::from_db(shell.clone(), e))?;
108
    accounts::audit(&app.db, Some(&me), "flow.runner.state", &r.name, &f.state).await.ok();
109
    Ok(Redirect::to("/admin/runners").into_response())
110
}
111
112
/// `POST /admin/runners/{id}/remove`: forget a runner that runs nothing.
113
pub async fn remove(
114
    State(app): State<Shared>,
115
    viewer: Viewer,
116
    Path(id): Path<i64>,
117
    Form(f): Form<CsrfOnly>,
118
) -> Result<Page<RunnersPage>, Error> {
119
    let shell = Shell::new(&app, "/admin/runners", &viewer);
120
    let me = require_admin(&viewer, &shell)?;
121
    csrf(&viewer, &f.csrf_token, &shell)?;
122
    let e = |e| Error::from_db(shell.clone(), e);
123
    let r = runners::get(&app.db, id).await.map_err(e)?.ok_or_else(|| Error::not_found(shell.clone()))?;
124
    let busy = runners::running(&app.db).await.map_err(e)?.into_iter().any(|(rid, n)| rid == id && n > 0);
125
    if busy {
126
        return page(
127
            &app,
128
            shell,
129
            None,
130
            None,
131
            Some(format!("{} is running jobs: drain it first, then remove it when they finish.", r.name)),
132
        )
133
        .await;
134
    }
135
    runners::remove(&app.db, id).await.map_err(e)?;
136
    accounts::audit(&app.db, Some(&me), "flow.runner.remove", &r.name, "").await.ok();
137
    let also = if r.source == "config" {
138
        " It is configured in hub.toml; remove its token there too, or it comes back on its next claim."
139
    } else {
140
        ""
141
    };
142
    page(&app, shell, None, Some(format!("Removed {}.{also}", r.name)), None).await
143
}
144
145
#[derive(Template)]
146
#[template(path = "admin_runner.html")]
147
pub struct RunnerPage {
148
    shell: Shell,
149
    r: runners::Runner,
150
    running: i64,
151
    history: Vec<runners::RunnerJob>,
152
    older: Option<i64>,
153
    jobs: i64,
154
    passed: i64,
155
    busy_ms: i64,
156
}
157
158
impl RunnerPage {
159
    /// Time spent running jobs, as pages show durations.
160
    pub fn busy(&self) -> String {
161
        crate::flow::human_ms(self.busy_ms)
162
    }
163
}
164
165
#[derive(Deserialize, Default)]
166
pub struct HistoryQuery {
167
    before: Option<i64>,
168
}
169
170
const PER_PAGE: i64 = 50;
171
172
/// `GET /admin/runners/{id}`: a runner and the jobs it ran.
173
pub async fn show(
174
    State(app): State<Shared>,
175
    viewer: Viewer,
176
    Path(id): Path<i64>,
177
    Query(q): Query<HistoryQuery>,
178
) -> Result<Page<RunnerPage>, Error> {
179
    let shell = Shell::new(&app, &format!("/admin/runners/{id}"), &viewer);
180
    require_admin(&viewer, &shell)?;
181
    let e = |e| Error::from_db(shell.clone(), e);
182
    let r = runners::get(&app.db, id).await.map_err(e)?.ok_or_else(|| Error::not_found(shell.clone()))?;
183
    let mut history = runners::history(&app.db, id, PER_PAGE + 1, q.before).await.map_err(e)?;
184
    let older = (history.len() as i64 > PER_PAGE).then(|| history[PER_PAGE as usize - 1].started_ms).flatten();
185
    history.truncate(PER_PAGE as usize);
186
    let (jobs, passed, busy_ms) = runners::totals(&app.db, id).await.map_err(e)?;
187
    let running = runners::running(&app.db).await.map_err(e)?.into_iter().find(|(rid, _)| *rid == id).map(|(_, n)| n).unwrap_or(0);
188
    Ok(Page(RunnerPage { shell, r, running, history, older, jobs, passed, busy_ms }))
189
}
190
191
pub fn duration(j: &runners::RunnerJob) -> String {
192
    match (j.started_ms, j.finished_ms) {
193
        (Some(a), Some(b)) => crate::flow::human_ms(b - a),
194
        (Some(a), None) => crate::flow::human_ms(db::now_ms() - a),
195
        _ => "–".into(),
196
    }
197
}