OXYGENHUB
# 7974adb0be66 oxygenhub / crates / hub / src / web / runners.rs
10.4 KiB · Rust · last changed Runners serve the site, an owner or one repository (refs #41, part 2) 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
}
79
80
async fn page(
81
    app: &Shared,
82
    shell: Shell,
83
    new_token: Option<String>,
84
    notice: Option<String>,
85
    error: Option<String>,
86
) -> Result<Page<RunnersPage>, Error> {
87
    let e = |e| Error::from_db(shell.clone(), e);
88
    let list = runners::list(&app.db).await.map_err(e)?;
89
    let running: HashMap<i64, i64> = runners::running(&app.db).await.map_err(e)?.into_iter().collect();
90
    let pending = runners::pending_registrations(&app.db).await.map_err(e)?;
91
    let mut rows = Vec::with_capacity(list.len());
92
    for r in list {
93
        let scope = scope_words(app, &r.scope()).await;
94
        rows.push(RunnerRow { running: running.get(&r.id).copied().unwrap_or(0), scope, r });
95
    }
96
    Ok(Page(RunnersPage { hub: app.config.base_url().to_string(), shell, runners: rows, pending, new_token, notice, error }))
97
}
98
99
/// `GET /admin/runners`.
100
pub async fn list(State(app): State<Shared>, viewer: Viewer) -> Result<Page<RunnersPage>, Error> {
101
    let shell = Shell::new(&app, "/admin/runners", &viewer);
102
    require_admin(&viewer, &shell)?;
103
    page(&app, shell, None, None, None).await
104
}
105
106
#[derive(Deserialize)]
107
pub struct CsrfOnly {
108
    #[serde(rename = "_csrf")]
109
    csrf_token: String,
110
}
111
112
#[derive(Deserialize)]
113
pub struct ScopeForm {
114
    #[serde(rename = "_csrf")]
115
    csrf_token: String,
116
    /// Empty for the whole site, an owner, or `owner/repository`.
117
    #[serde(default)]
118
    scope: String,
119
}
120
121
/// `POST /admin/runners/token`: a one-time registration token, shown once,
122
/// for a runner of the scope given.
123
pub async fn new_token(State(app): State<Shared>, viewer: Viewer, Form(f): Form<ScopeForm>) -> Result<Page<RunnersPage>, Error> {
124
    let shell = Shell::new(&app, "/admin/runners", &viewer);
125
    let me = require_admin(&viewer, &shell)?;
126
    csrf(&viewer, &f.csrf_token, &shell)?;
127
    let scope = match parse_scope(&app, &f.scope).await {
128
        Ok(s) => s,
129
        Err(e) => return page(&app, shell, None, None, Some(e)).await,
130
    };
131
    let token = format!("flowreg_{}", crate::auth::random_hex(24));
132
    runners::add_registration(&app.db, &blake3::hash(token.as_bytes()).to_hex(), &me.username, &scope)
133
        .await
134
        .map_err(|e| Error::from_db(shell.clone(), e))?;
135
    let words = scope_words(&app, &scope).await;
136
    accounts::audit(&app.db, Some(&me), "flow.runner.token", "", &format!("registration token created for {words}")).await.ok();
137
    page(&app, shell, Some(token), Some(format!("The runner it registers will take jobs from {words}.")), None).await
138
}
139
140
/// `POST /admin/runners/{id}/scope`: change the repositories a runner serves.
141
pub async fn set_scope(
142
    State(app): State<Shared>,
143
    viewer: Viewer,
144
    Path(id): Path<i64>,
145
    Form(f): Form<ScopeForm>,
146
) -> Result<Page<RunnersPage>, Error> {
147
    let shell = Shell::new(&app, "/admin/runners", &viewer);
148
    let me = require_admin(&viewer, &shell)?;
149
    csrf(&viewer, &f.csrf_token, &shell)?;
150
    let r =
151
        runners::get(&app.db, id).await.map_err(|e| Error::from_db(shell.clone(), e))?.ok_or_else(|| Error::not_found(shell.clone()))?;
152
    let scope = match parse_scope(&app, &f.scope).await {
153
        Ok(s) => s,
154
        Err(e) => return page(&app, shell, None, None, Some(e)).await,
155
    };
156
    runners::set_scope(&app.db, id, &scope).await.map_err(|e| Error::from_db(shell.clone(), e))?;
157
    let words = scope_words(&app, &scope).await;
158
    accounts::audit(&app.db, Some(&me), "flow.runner.scope", &r.name, &words).await.ok();
159
    page(&app, shell, None, Some(format!("{} now takes jobs from {words}.", r.name)), None).await
160
}
161
162
#[derive(Deserialize)]
163
pub struct StateForm {
164
    #[serde(rename = "_csrf")]
165
    csrf_token: String,
166
    /// `active`, `draining` or `disabled`.
167
    state: String,
168
}
169
170
/// `POST /admin/runners/{id}/state`: drain, disable or resume a runner.
171
pub async fn set_state(
172
    State(app): State<Shared>,
173
    viewer: Viewer,
174
    Path(id): Path<i64>,
175
    Form(f): Form<StateForm>,
176
) -> Result<Response, Error> {
177
    let shell = Shell::new(&app, "/admin/runners", &viewer);
178
    let me = require_admin(&viewer, &shell)?;
179
    csrf(&viewer, &f.csrf_token, &shell)?;
180
    if !matches!(f.state.as_str(), "active" | "draining" | "disabled") {
181
        return Err(Error::bad(shell, "a runner is active, draining or disabled"));
182
    }
183
    let r =
184
        runners::get(&app.db, id).await.map_err(|e| Error::from_db(shell.clone(), e))?.ok_or_else(|| Error::not_found(shell.clone()))?;
185
    runners::set_state(&app.db, id, &f.state).await.map_err(|e| Error::from_db(shell.clone(), e))?;
186
    accounts::audit(&app.db, Some(&me), "flow.runner.state", &r.name, &f.state).await.ok();
187
    Ok(Redirect::to("/admin/runners").into_response())
188
}
189
190
/// `POST /admin/runners/{id}/remove`: forget a runner that runs nothing.
191
pub async fn remove(
192
    State(app): State<Shared>,
193
    viewer: Viewer,
194
    Path(id): Path<i64>,
195
    Form(f): Form<CsrfOnly>,
196
) -> Result<Page<RunnersPage>, Error> {
197
    let shell = Shell::new(&app, "/admin/runners", &viewer);
198
    let me = require_admin(&viewer, &shell)?;
199
    csrf(&viewer, &f.csrf_token, &shell)?;
200
    let e = |e| Error::from_db(shell.clone(), e);
201
    let r = runners::get(&app.db, id).await.map_err(e)?.ok_or_else(|| Error::not_found(shell.clone()))?;
202
    let busy = runners::running(&app.db).await.map_err(e)?.into_iter().any(|(rid, n)| rid == id && n > 0);
203
    if busy {
204
        return page(
205
            &app,
206
            shell,
207
            None,
208
            None,
209
            Some(format!("{} is running jobs: drain it first, then remove it when they finish.", r.name)),
210
        )
211
        .await;
212
    }
213
    runners::remove(&app.db, id).await.map_err(e)?;
214
    accounts::audit(&app.db, Some(&me), "flow.runner.remove", &r.name, "").await.ok();
215
    let also = if r.source == "config" {
216
        " It is configured in hub.toml; remove its token there too, or it comes back on its next claim."
217
    } else {
218
        ""
219
    };
220
    page(&app, shell, None, Some(format!("Removed {}.{also}", r.name)), None).await
221
}
222
223
#[derive(Template)]
224
#[template(path = "admin_runner.html")]
225
pub struct RunnerPage {
226
    shell: Shell,
227
    r: runners::Runner,
228
    running: i64,
229
    history: Vec<runners::RunnerJob>,
230
    older: Option<i64>,
231
    jobs: i64,
232
    passed: i64,
233
    busy_ms: i64,
234
}
235
236
impl RunnerPage {
237
    /// Time spent running jobs, as pages show durations.
238
    pub fn busy(&self) -> String {
239
        crate::flow::human_ms(self.busy_ms)
240
    }
241
}
242
243
#[derive(Deserialize, Default)]
244
pub struct HistoryQuery {
245
    before: Option<i64>,
246
}
247
248
const PER_PAGE: i64 = 50;
249
250
/// `GET /admin/runners/{id}`: a runner and the jobs it ran.
251
pub async fn show(
252
    State(app): State<Shared>,
253
    viewer: Viewer,
254
    Path(id): Path<i64>,
255
    Query(q): Query<HistoryQuery>,
256
) -> Result<Page<RunnerPage>, Error> {
257
    let shell = Shell::new(&app, &format!("/admin/runners/{id}"), &viewer);
258
    require_admin(&viewer, &shell)?;
259
    let e = |e| Error::from_db(shell.clone(), e);
260
    let r = runners::get(&app.db, id).await.map_err(e)?.ok_or_else(|| Error::not_found(shell.clone()))?;
261
    let mut history = runners::history(&app.db, id, PER_PAGE + 1, q.before).await.map_err(e)?;
262
    let older = (history.len() as i64 > PER_PAGE).then(|| history[PER_PAGE as usize - 1].started_ms).flatten();
263
    history.truncate(PER_PAGE as usize);
264
    let (jobs, passed, busy_ms) = runners::totals(&app.db, id).await.map_err(e)?;
265
    let running = runners::running(&app.db).await.map_err(e)?.into_iter().find(|(rid, _)| *rid == id).map(|(_, n)| n).unwrap_or(0);
266
    Ok(Page(RunnerPage { shell, r, running, history, older, jobs, passed, busy_ms }))
267
}
268
269
pub fn duration(j: &runners::RunnerJob) -> String {
270
    match (j.started_ms, j.finished_ms) {
271
        (Some(a), Some(b)) => crate::flow::human_ms(b - a),
272
        (Some(a), None) => crate::flow::human_ms(db::now_ms() - a),
273
        _ => "–".into(),
274
    }
275
}