OXYGENHUB
# c8df33a6829f oxygenhub / crates / hub / src / web / runners.rs
15.7 KiB · Rust · last changed Runners decline, and the queue says why a job waits 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
    /// The repositories it serves, in words.
28
    pub scope: String,
29
}
30
31
/// A scope in words: the whole site, an owner's repositories, or one
32
/// repository.
33
async fn scope_words(app: &Shared, s: &runners::Scope) -> String {
34
    match s.repo_id {
35
        Some(id) => match db::repos::get_by_id(&app.db, id).await.ok().flatten() {
36
            Some(r) => format!("{}/{}", r.owner, r.name),
37
            None => "a removed repository".into(),
38
        },
39
        None if s.owner.is_empty() => "the whole site".into(),
40
        None => format!("{}'s repositories", s.owner),
41
    }
42
}
43
44
/// A scope as typed: empty for the whole site, an owner, or
45
/// `owner/repository`.
46
async fn parse_scope(app: &Shared, text: &str) -> Result<runners::Scope, String> {
47
    let text = text.trim().trim_matches('/');
48
    if text.is_empty() {
49
        return Ok(runners::Scope::default());
50
    }
51
    if let Some((owner, name)) = text.split_once('/') {
52
        return match db::repos::get(&app.db, owner, name).await {
53
            Ok(Some(r)) => Ok(runners::Scope { owner: r.owner.clone(), repo_id: Some(r.id) }),
54
            _ => Err(format!("There is no repository {text}.")),
55
        };
56
    }
57
    let user = accounts::account_by_name(&app.db, text).await.ok().flatten().is_some();
58
    let org = accounts::org_by_name(&app.db, text).await.ok().flatten().is_some();
59
    if user || org {
60
        Ok(runners::Scope { owner: text.to_string(), repo_id: None })
61
    } else {
62
        Err(format!("There is no user or organisation {text}."))
63
    }
64
}
65
66
#[derive(Template)]
67
#[template(path = "admin_runners.html")]
68
pub struct RunnersPage {
69
    shell: Shell,
70
    runners: Vec<RunnerRow>,
71
    /// Registration tokens waiting to be used: (created by, created, expires).
72
    pending: Vec<(String, i64, i64)>,
73
    /// A token just made, shown once, with the command that uses it.
74
    new_token: Option<String>,
75
    hub: String,
76
    notice: Option<String>,
77
    error: Option<String>,
78
    /// Running jobs, then queued ones in the order runners claim them.
79
    queue: Vec<db::flow::QueueEntry>,
80
}
81
82
async fn page(
83
    app: &Shared,
84
    shell: Shell,
85
    new_token: Option<String>,
86
    notice: Option<String>,
87
    error: Option<String>,
88
) -> Result<Page<RunnersPage>, Error> {
89
    let e = |e| Error::from_db(shell.clone(), e);
90
    let list = runners::list(&app.db).await.map_err(e)?;
91
    let running: HashMap<i64, i64> = runners::running(&app.db).await.map_err(e)?.into_iter().collect();
92
    let pending = runners::pending_registrations(&app.db).await.map_err(e)?;
93
    let mut rows = Vec::with_capacity(list.len());
94
    for r in list {
95
        let scope = scope_words(app, &r.scope()).await;
96
        rows.push(RunnerRow { running: running.get(&r.id).copied().unwrap_or(0), scope, r });
97
    }
98
    let queue = db::flow::queue(&app.db).await.map_err(e)?;
99
    Ok(Page(RunnersPage { hub: app.config.base_url().to_string(), shell, runners: rows, pending, new_token, notice, error, queue }))
100
}
101
102
/// `GET /admin/runners`.
103
pub async fn list(State(app): State<Shared>, viewer: Viewer) -> Result<Page<RunnersPage>, Error> {
104
    let shell = Shell::new(&app, "/admin/runners", &viewer);
105
    require_admin(&viewer, &shell)?;
106
    page(&app, shell, None, None, None).await
107
}
108
109
#[derive(Deserialize)]
110
pub struct CsrfOnly {
111
    #[serde(rename = "_csrf")]
112
    csrf_token: String,
113
}
114
115
#[derive(Deserialize)]
116
pub struct ScopeForm {
117
    #[serde(rename = "_csrf")]
118
    csrf_token: String,
119
    /// Empty for the whole site, an owner, or `owner/repository`.
120
    #[serde(default)]
121
    scope: String,
122
}
123
124
/// `POST /admin/runners/token`: a one-time registration token, shown once,
125
/// for a runner of the scope given.
126
pub async fn new_token(State(app): State<Shared>, viewer: Viewer, Form(f): Form<ScopeForm>) -> Result<Page<RunnersPage>, Error> {
127
    let shell = Shell::new(&app, "/admin/runners", &viewer);
128
    let me = require_admin(&viewer, &shell)?;
129
    csrf(&viewer, &f.csrf_token, &shell)?;
130
    let scope = match parse_scope(&app, &f.scope).await {
131
        Ok(s) => s,
132
        Err(e) => return page(&app, shell, None, None, Some(e)).await,
133
    };
134
    let token = format!("flowreg_{}", crate::auth::random_hex(24));
135
    runners::add_registration(&app.db, &blake3::hash(token.as_bytes()).to_hex(), &me.username, &scope)
136
        .await
137
        .map_err(|e| Error::from_db(shell.clone(), e))?;
138
    let words = scope_words(&app, &scope).await;
139
    accounts::audit(&app.db, Some(&me), "flow.runner.token", "", &format!("registration token created for {words}")).await.ok();
140
    page(&app, shell, Some(token), Some(format!("The runner it registers will take jobs from {words}.")), None).await
141
}
142
143
/// `POST /admin/runners/{id}/scope`: change the repositories a runner serves.
144
pub async fn set_scope(
145
    State(app): State<Shared>,
146
    viewer: Viewer,
147
    Path(id): Path<i64>,
148
    Form(f): Form<ScopeForm>,
149
) -> Result<Page<RunnersPage>, Error> {
150
    let shell = Shell::new(&app, "/admin/runners", &viewer);
151
    let me = require_admin(&viewer, &shell)?;
152
    csrf(&viewer, &f.csrf_token, &shell)?;
153
    let r =
154
        runners::get(&app.db, id).await.map_err(|e| Error::from_db(shell.clone(), e))?.ok_or_else(|| Error::not_found(shell.clone()))?;
155
    let scope = match parse_scope(&app, &f.scope).await {
156
        Ok(s) => s,
157
        Err(e) => return page(&app, shell, None, None, Some(e)).await,
158
    };
159
    runners::set_scope(&app.db, id, &scope).await.map_err(|e| Error::from_db(shell.clone(), e))?;
160
    let words = scope_words(&app, &scope).await;
161
    accounts::audit(&app.db, Some(&me), "flow.runner.scope", &r.name, &words).await.ok();
162
    page(&app, shell, None, Some(format!("{} now takes jobs from {words}.", r.name)), None).await
163
}
164
165
#[derive(Deserialize)]
166
pub struct LimitsForm {
167
    #[serde(rename = "_csrf")]
168
    csrf_token: String,
169
    /// Comma-separated.
170
    #[serde(default)]
171
    labels: String,
172
    /// Minutes; empty for the site's limit.
173
    #[serde(default)]
174
    timeout_minutes: String,
175
}
176
177
/// `POST /admin/runners/{id}/limits`: the labels an administrator gives a
178
/// runner and its own job time limit (#98).
179
pub async fn set_limits(
180
    State(app): State<Shared>,
181
    viewer: Viewer,
182
    Path(id): Path<i64>,
183
    Form(f): Form<LimitsForm>,
184
) -> Result<Response, Error> {
185
    let shell = Shell::new(&app, &format!("/admin/runners/{id}"), &viewer);
186
    let me = require_admin(&viewer, &shell)?;
187
    csrf(&viewer, &f.csrf_token, &shell)?;
188
    let r =
189
        runners::get(&app.db, id).await.map_err(|e| Error::from_db(shell.clone(), e))?.ok_or_else(|| Error::not_found(shell.clone()))?;
190
    let mut labels: Vec<String> = Vec::new();
191
    for l in f.labels.split([',', ' ']).map(str::trim).filter(|l| !l.is_empty()) {
192
        if !l.chars().all(|c| c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '.')) || l.len() > 40 {
193
            return Err(Error::bad(shell, format!("label {l:?}: letters, digits, - _ and . only, up to 40 characters")));
194
        }
195
        if !labels.iter().any(|x| x == l) {
196
            labels.push(l.to_string());
197
        }
198
    }
199
    let timeout = match f.timeout_minutes.trim() {
200
        "" => None,
201
        t => match t.parse::<i64>() {
202
            Ok(m) if (1..=24 * 60).contains(&m) => Some(m),
203
            _ => return Err(Error::bad(shell, "a time limit is a whole number of minutes from 1 to 1440")),
204
        },
205
    };
206
    runners::set_limits(&app.db, id, &labels.join(","), timeout).await.map_err(|e| Error::from_db(shell.clone(), e))?;
207
    let words = format!(
208
        "labels {}; time limit {}",
209
        if labels.is_empty() { "none".to_string() } else { labels.join(", ") },
210
        timeout.map(|m| format!("{m} min")).unwrap_or_else(|| "the site's".into())
211
    );
212
    accounts::audit(&app.db, Some(&me), "flow.runner.limits", &r.name, &words).await.ok();
213
    Ok(Redirect::to(&format!("/admin/runners/{id}")).into_response())
214
}
215
216
#[derive(Deserialize)]
217
pub struct StateForm {
218
    #[serde(rename = "_csrf")]
219
    csrf_token: String,
220
    /// `active`, `draining` or `disabled`.
221
    state: String,
222
}
223
224
/// `POST /admin/runners/{id}/state`: drain, disable or resume a runner.
225
pub async fn set_state(
226
    State(app): State<Shared>,
227
    viewer: Viewer,
228
    Path(id): Path<i64>,
229
    Form(f): Form<StateForm>,
230
) -> Result<Response, Error> {
231
    let shell = Shell::new(&app, "/admin/runners", &viewer);
232
    let me = require_admin(&viewer, &shell)?;
233
    csrf(&viewer, &f.csrf_token, &shell)?;
234
    if !matches!(f.state.as_str(), "active" | "draining" | "disabled") {
235
        return Err(Error::bad(shell, "a runner is active, draining or disabled"));
236
    }
237
    let r =
238
        runners::get(&app.db, id).await.map_err(|e| Error::from_db(shell.clone(), e))?.ok_or_else(|| Error::not_found(shell.clone()))?;
239
    runners::set_state(&app.db, id, &f.state).await.map_err(|e| Error::from_db(shell.clone(), e))?;
240
    accounts::audit(&app.db, Some(&me), "flow.runner.state", &r.name, &f.state).await.ok();
241
    Ok(Redirect::to("/admin/runners").into_response())
242
}
243
244
/// `POST /admin/runners/{id}/remove`: forget a runner that runs nothing.
245
pub async fn remove(
246
    State(app): State<Shared>,
247
    viewer: Viewer,
248
    Path(id): Path<i64>,
249
    Form(f): Form<CsrfOnly>,
250
) -> Result<Page<RunnersPage>, Error> {
251
    let shell = Shell::new(&app, "/admin/runners", &viewer);
252
    let me = require_admin(&viewer, &shell)?;
253
    csrf(&viewer, &f.csrf_token, &shell)?;
254
    let e = |e| Error::from_db(shell.clone(), e);
255
    let r = runners::get(&app.db, id).await.map_err(e)?.ok_or_else(|| Error::not_found(shell.clone()))?;
256
    let busy = runners::running(&app.db).await.map_err(e)?.into_iter().any(|(rid, n)| rid == id && n > 0);
257
    if busy {
258
        return page(
259
            &app,
260
            shell,
261
            None,
262
            None,
263
            Some(format!("{} is running jobs: drain it first, then remove it when they finish.", r.name)),
264
        )
265
        .await;
266
    }
267
    runners::remove(&app.db, id).await.map_err(e)?;
268
    accounts::audit(&app.db, Some(&me), "flow.runner.remove", &r.name, "").await.ok();
269
    let also = if r.source == "config" {
270
        " It is configured in hub.toml; remove its token there too, or it comes back on its next claim."
271
    } else {
272
        ""
273
    };
274
    page(&app, shell, None, Some(format!("Removed {}.{also}", r.name)), None).await
275
}
276
277
#[derive(Template)]
278
#[template(path = "admin_runner.html")]
279
pub struct RunnerPage {
280
    shell: Shell,
281
    r: runners::Runner,
282
    running: i64,
283
    history: Vec<runners::RunnerJob>,
284
    declines: Vec<crate::db::flow::RunnerDecline>,
285
    older: Option<i64>,
286
    jobs: i64,
287
    passed: i64,
288
    busy_ms: i64,
289
}
290
291
impl RunnerPage {
292
    /// Time spent running jobs, as pages show durations.
293
    pub fn busy(&self) -> String {
294
        crate::flow::human_ms(self.busy_ms)
295
    }
296
}
297
298
#[derive(Deserialize, Default)]
299
pub struct HistoryQuery {
300
    before: Option<i64>,
301
}
302
303
const PER_PAGE: i64 = 50;
304
305
/// `GET /admin/runners/{id}`: a runner and the jobs it ran.
306
pub async fn show(
307
    State(app): State<Shared>,
308
    viewer: Viewer,
309
    Path(id): Path<i64>,
310
    Query(q): Query<HistoryQuery>,
311
) -> Result<Page<RunnerPage>, Error> {
312
    let shell = Shell::new(&app, &format!("/admin/runners/{id}"), &viewer);
313
    require_admin(&viewer, &shell)?;
314
    let e = |e| Error::from_db(shell.clone(), e);
315
    let r = runners::get(&app.db, id).await.map_err(e)?.ok_or_else(|| Error::not_found(shell.clone()))?;
316
    let mut history = runners::history(&app.db, id, PER_PAGE + 1, q.before).await.map_err(e)?;
317
    let older = (history.len() as i64 > PER_PAGE).then(|| history[PER_PAGE as usize - 1].started_ms).flatten();
318
    history.truncate(PER_PAGE as usize);
319
    let (jobs, passed, busy_ms) = runners::totals(&app.db, id).await.map_err(e)?;
320
    let running = runners::running(&app.db).await.map_err(e)?.into_iter().find(|(rid, _)| *rid == id).map(|(_, n)| n).unwrap_or(0);
321
    let declines = crate::db::flow::declines_of_runner(&app.db, id, 20).await.map_err(e)?;
322
    Ok(Page(RunnerPage { shell, r, running, history, declines, older, jobs, passed, busy_ms }))
323
}
324
325
pub fn duration(j: &runners::RunnerJob) -> String {
326
    match (j.started_ms, j.finished_ms) {
327
        (Some(a), Some(b)) => crate::flow::human_ms(b - a),
328
        (Some(a), None) => crate::flow::human_ms(db::now_ms() - a),
329
        _ => "–".into(),
330
    }
331
}
332
333
// ---------------------------------------------------------------- settings
334
335
#[derive(Template)]
336
#[template(path = "admin_runner_settings.html")]
337
pub struct SettingsPage {
338
    shell: Shell,
339
    default_m: u64,
340
    max_m: u64,
341
    grace_m: i64,
342
    /// What hub.toml says, shown beside a setting the page overrides.
343
    file_default_m: u64,
344
    flow_repos: Vec<String>,
345
    notice: Option<String>,
346
    error: Option<String>,
347
}
348
349
async fn settings_page(app: &Shared, shell: Shell, notice: Option<String>, error: Option<String>) -> Page<SettingsPage> {
350
    let l = crate::flow::limits(app).await;
351
    Page(SettingsPage {
352
        shell,
353
        default_m: l.default_s / 60,
354
        max_m: l.max_s / 60,
355
        grace_m: l.grace_ms / 60_000,
356
        file_default_m: app.config.flow.job_timeout_minutes,
357
        flow_repos: app.config.flow.repos.clone(),
358
        notice,
359
        error,
360
    })
361
}
362
363
/// `GET /admin/runners/settings` (oxygenhub#89).
364
pub async fn settings(State(app): State<Shared>, viewer: Viewer) -> Result<Page<SettingsPage>, Error> {
365
    let shell = Shell::new(&app, "/admin/runners/settings", &viewer);
366
    require_admin(&viewer, &shell)?;
367
    Ok(settings_page(&app, shell, None, None).await)
368
}
369
370
#[derive(Deserialize)]
371
pub struct SettingsForm {
372
    #[serde(rename = "_csrf")]
373
    csrf_token: String,
374
    default_minutes: String,
375
    max_minutes: String,
376
    grace_minutes: String,
377
}
378
379
/// `POST /admin/runners/settings`: the site's job time limits.
380
pub async fn save_settings(State(app): State<Shared>, viewer: Viewer, Form(f): Form<SettingsForm>) -> Result<Page<SettingsPage>, Error> {
381
    let shell = Shell::new(&app, "/admin/runners/settings", &viewer);
382
    let me = require_admin(&viewer, &shell)?;
383
    csrf(&viewer, &f.csrf_token, &shell)?;
384
    let num = |v: &str| v.trim().parse::<u64>().ok().filter(|n| *n > 0);
385
    let (Some(d), Some(m), Some(g)) = (num(&f.default_minutes), num(&f.max_minutes), num(&f.grace_minutes)) else {
386
        return Ok(settings_page(&app, shell, None, Some("Each limit is a whole number of minutes, at least 1.".into())).await);
387
    };
388
    if d > m {
389
        return Ok(settings_page(&app, shell, None, Some("The default cannot be longer than the maximum.".into())).await);
390
    }
391
    if m > 24 * 60 {
392
        return Ok(settings_page(&app, shell, None, Some("A job may run for a day at most (1440 minutes).".into())).await);
393
    }
394
    let e = |e| Error::from_db(shell.clone(), e);
395
    accounts::set_setting(&app.db, crate::flow::SETTING_DEFAULT, &d.to_string()).await.map_err(e)?;
396
    accounts::set_setting(&app.db, crate::flow::SETTING_MAX, &m.to_string()).await.map_err(e)?;
397
    accounts::set_setting(&app.db, crate::flow::SETTING_GRACE, &g.to_string()).await.map_err(e)?;
398
    accounts::audit(
399
        &app.db,
400
        Some(&me),
401
        "flow.settings",
402
        "",
403
        &format!("job time limit {d} min by default, {m} min at most; lost after {g} min more"),
404
    )
405
    .await
406
    .ok();
407
    Ok(settings_page(&app, shell, Some("Saved. Jobs claimed from now on run under these limits.".into()), None).await)
408
}