OXYGENHUB
# c30444e35f2e oxygenhub / crates / hub / src / insights.rs
13.4 KiB · Rust · last changed Insights: OxygenFlow's standard charts, for the site, an owner and a repository 17 days ago source blame history raw
1 
//! Insights (oxygenhub#257): OxygenFlow's standard numbers over time, for
2 
//! the site, an owner or a repository: how long jobs wait for a runner, how
3 
//! many run and how they end, how long they take, how deep the queue is and
4 
//! how busy the runners are. All from `flow_jobs` (migration 0080's queue
5 
//! and ready times) and the runners' rows, so nothing new is collected;
6 
//! power users point Grafana at `/metrics` for anything more.
7
8 
use crate::db::{self, Pool};
9 
use sqlx::FromRow;
10
11 
/// The span a page shows.
12 
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
13 
pub enum Range {
14 
    Hour,
15 
    Day,
16 
    Week,
17 
}
18
19 
impl Range {
20 
    pub fn parse(s: &str) -> Range {
21 
        match s {
22 
            "1h" => Range::Hour,
23 
            "7d" => Range::Week,
24 
            _ => Range::Day,
25 
        }
26 
    }
27
28 
    pub fn key(self) -> &'static str {
29 
        match self {
30 
            Range::Hour => "1h",
31 
            Range::Day => "24h",
32 
            Range::Week => "7d",
33 
        }
34 
    }
35
36 
    pub fn ms(self) -> i64 {
37 
        match self {
38 
            Range::Hour => 3_600_000,
39 
            Range::Day => 86_400_000,
40 
            Range::Week => 7 * 86_400_000,
41 
        }
42 
    }
43
44 
    /// How many buckets the charts have.
45 
    pub fn buckets(self) -> usize {
46 
        match self {
47 
            Range::Hour => 60,
48 
            Range::Day => 48,
49 
            Range::Week => 56,
50 
        }
51 
    }
52
53 
    pub fn bucket_ms(self) -> i64 {
54 
        self.ms() / self.buckets() as i64
55 
    }
56 
}
57
58 
/// Which jobs count.
59 
#[derive(Clone, Debug, Default)]
60 
pub struct Scope {
61 
    /// An owner's repositories.
62 
    pub owner: Option<String>,
63 
    /// One repository.
64 
    pub repo_id: Option<i64>,
65 
    /// One runner's jobs.
66 
    pub runner: Option<String>,
67 
}
68
69 
#[derive(Clone, Debug, FromRow)]
70 
struct JobRow {
71 
    status: String,
72 
    runner: String,
73 
    owner: String,
74 
    repo: String,
75 
    queued_ms: Option<i64>,
76 
    ready_ms: Option<i64>,
77 
    started_ms: Option<i64>,
78 
    finished_ms: Option<i64>,
79 
}
80
81 
impl JobRow {
82 
    /// Seconds from ready to started, when it started.
83 
    fn pickup(&self) -> Option<f64> {
84 
        Some((self.started_ms? - self.ready_ms.or(self.queued_ms)?).max(0) as f64 / 1000.0)
85 
    }
86
87 
    fn duration(&self) -> Option<f64> {
88 
        Some((self.finished_ms? - self.started_ms?).max(0) as f64 / 1000.0)
89 
    }
90
91 
    /// When it left the queue: started, or ended without starting.
92 
    fn left_queue(&self) -> Option<i64> {
93 
        self.started_ms.or(self.finished_ms)
94 
    }
95
96 
    fn ended(&self) -> bool {
97 
        !matches!(self.status.as_str(), "queued" | "running")
98 
    }
99 
}
100
101 
/// One chart's line or stack: a value per bucket.
102 
#[derive(Clone, Debug)]
103 
pub struct Series {
104 
    pub name: &'static str,
105 
    /// The CSS class that colours it (`s-ok`, `s-bad`, …).
106 
    pub class: &'static str,
107 
    pub values: Vec<Option<f64>>,
108 
}
109
110 
/// One runner's row.
111 
#[derive(Clone, Debug)]
112 
pub struct RunnerRow {
113 
    pub name: String,
114 
    pub status: &'static str,
115 
    pub running: i64,
116 
    pub max_jobs: i64,
117 
    pub jobs: i64,
118 
    pub failed: i64,
119 
    pub pickup_median: Option<f64>,
120 
    pub duration_median: Option<f64>,
121 
    pub last_seen_ms: Option<i64>,
122 
}
123
124 
/// One repository's row.
125 
#[derive(Clone, Debug)]
126 
pub struct RepoRow {
127 
    pub slug: String,
128 
    pub jobs: i64,
129 
    pub failed: i64,
130 
    pub pickup_median: Option<f64>,
131 
    pub pickup_p95: Option<f64>,
132 
    pub queued: i64,
133 
}
134
135 
/// Everything a page draws.
136 
#[derive(Clone, Debug)]
137 
pub struct Insights {
138 
    pub range: Range,
139 
    /// The first bucket's start; buckets follow every `range.bucket_ms()`.
140 
    pub start_ms: i64,
141 
    pub now_ms: i64,
142 
    /// Seconds: median and 95th percentile per bucket, by the job's start.
143 
    pub pickup: Vec<Series>,
144 
    /// Jobs that ended per bucket, by result.
145 
    pub results: Vec<Series>,
146 
    /// Seconds: median and 95th percentile per bucket, by the job's end.
147 
    pub durations: Vec<Series>,
148 
    /// Jobs in the queue and running, at each bucket's end.
149 
    pub load: Vec<Series>,
150 
    /// Running jobs over the online runners' slots, percent, per bucket.
151 
    pub utilisation: Vec<Series>,
152 
    pub ended: i64,
153 
    pub passed: i64,
154 
    pub failed: i64,
155 
    pub pickup_median: Option<f64>,
156 
    pub pickup_p95: Option<f64>,
157 
    pub duration_median: Option<f64>,
158 
    pub queued_now: i64,
159 
    pub running_now: i64,
160 
    pub runners_online: i64,
161 
    pub runners_total: i64,
162 
    pub slots_busy: i64,
163 
    pub slots_total: i64,
164 
    pub runners: Vec<RunnerRow>,
165 
    pub repos: Vec<RepoRow>,
166 
}
167
168 
/// The `p`-th percentile (0..=1) of `v`, nearest rank.
169 
pub fn percentile(v: &mut [f64], p: f64) -> Option<f64> {
170 
    if v.is_empty() {
171 
        return None;
172 
    }
173 
    v.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));
174 
    let rank = ((p * v.len() as f64).ceil() as usize).clamp(1, v.len());
175 
    Some(v[rank - 1])
176 
}
177
178 
async fn rows(pool: &Pool, scope: &Scope, since: i64) -> sqlx::Result<Vec<JobRow>> {
179 
    sqlx::query_as(
180 
        "SELECT j.status, j.runner, p.owner, p.name AS repo, COALESCE(j.queued_ms, r.created_ms) AS queued_ms, j.ready_ms,
181 
                j.started_ms, j.finished_ms
182 
         FROM flow_jobs j JOIN flow_runs r ON r.id = j.run_id JOIN repositories p ON p.id = r.repo_id
183 
         WHERE (j.finished_ms >= $1 OR j.status IN ('queued', 'running'))
184 
           AND ($2 = '' OR p.owner = $3) AND ($4 IS NULL OR p.id = $5) AND ($6 = '' OR j.runner = $7)",
185 
    )
186 
    .bind(since)
187 
    .bind(scope.owner.as_deref().unwrap_or(""))
188 
    .bind(scope.owner.as_deref().unwrap_or(""))
189 
    .bind(scope.repo_id)
190 
    .bind(scope.repo_id)
191 
    .bind(scope.runner.as_deref().unwrap_or(""))
192 
    .bind(scope.runner.as_deref().unwrap_or(""))
193 
    .fetch_all(pool)
194 
    .await
195 
}
196
197 
/// Gather a scope's numbers over `range`, ending now.
198 
pub async fn gather(pool: &Pool, scope: &Scope, range: Range) -> sqlx::Result<Insights> {
199 
    let now = db::now_ms();
200 
    let start = now - range.ms();
201 
    let jobs = rows(pool, scope, start).await?;
202 
    let runners = db::runners::list(pool).await?;
203 
    let n = range.buckets();
204 
    let width = range.bucket_ms();
205 
    let bucket = |t: i64| -> Option<usize> { (t >= start && t <= now).then(|| (((t - start) / width) as usize).min(n - 1)) };
206
207 
    let mut pickups: Vec<Vec<f64>> = vec![Vec::new(); n];
208 
    let mut durations: Vec<Vec<f64>> = vec![Vec::new(); n];
209 
    let (mut passed, mut failed, mut other) = (vec![0f64; n], vec![0f64; n], vec![0f64; n]);
210 
    let (mut all_pickups, mut all_durations) = (Vec::new(), Vec::new());
211 
    for j in &jobs {
212 
        if let (Some(p), Some(b)) = (j.pickup(), j.started_ms.and_then(bucket)) {
213 
            pickups[b].push(p);
214 
            all_pickups.push(p);
215 
        }
216 
        if let (Some(f), true) = (j.finished_ms, j.ended()) {
217 
            if let Some(b) = bucket(f) {
218 
                match j.status.as_str() {
219 
                    "passed" => passed[b] += 1.0,
220 
                    "failed" => failed[b] += 1.0,
221 
                    _ => other[b] += 1.0,
222 
                }
223 
                if let Some(d) = j.duration() {
224 
                    durations[b].push(d);
225 
                    all_durations.push(d);
226 
                }
227 
            }
228 
        }
229 
    }
230 
    // The queue and the running jobs at each bucket's end.
231 
    let (mut queued_at, mut running_at) = (vec![0f64; n], vec![0f64; n]);
232 
    for (i, (q, r)) in queued_at.iter_mut().zip(running_at.iter_mut()).enumerate() {
233 
        let t = (start + (i as i64 + 1) * width).min(now);
234 
        for j in &jobs {
235 
            let entered = j.queued_ms.unwrap_or(i64::MAX);
236 
            if entered <= t && j.left_queue().is_none_or(|l| l > t) && !(j.ended() && j.finished_ms.is_none()) {
237 
                *q += 1.0;
238 
            }
239 
            if j.started_ms.is_some_and(|s| s <= t) && j.finished_ms.is_none_or(|f| f > t) && !(j.ended() && j.finished_ms.is_none()) {
240 
                *r += 1.0;
241 
            }
242 
        }
243 
    }
244 
    // Slots: the runners serving the scope that are online now.
245 
    let serving: Vec<&db::runners::Runner> =
246 
        runners.iter().filter(|r| scope.runner.as_deref().is_none_or(|n| n == r.name)).filter(|r| r.state != "removed").collect();
247 
    let slots_total: i64 = serving.iter().filter(|r| r.online()).map(|r| r.max_jobs.max(1)).sum();
248 
    let utilisation: Vec<Option<f64>> =
249 
        running_at.iter().map(|r| (slots_total > 0).then(|| (r / slots_total as f64 * 100.0).min(100.0))).collect();
250
251 
    let quantiles = |sets: &mut Vec<Vec<f64>>, p: f64| -> Vec<Option<f64>> { sets.iter_mut().map(|v| percentile(v, p)).collect() };
252 
    let pickup = vec![
253 
        Series { name: "median", class: "s-main", values: quantiles(&mut pickups, 0.5) },
254 
        Series { name: "95th percentile", class: "s-warn", values: quantiles(&mut pickups, 0.95) },
255 
    ];
256 
    let durations_series = vec![
257 
        Series { name: "median", class: "s-main", values: quantiles(&mut durations, 0.5) },
258 
        Series { name: "95th percentile", class: "s-warn", values: quantiles(&mut durations, 0.95) },
259 
    ];
260 
    let some = |v: Vec<f64>| v.into_iter().map(Some).collect::<Vec<_>>();
261 
    let ended_total = |v: &[f64]| v.iter().sum::<f64>() as i64;
262 
    let (p_total, f_total, o_total) = (ended_total(&passed), ended_total(&failed), ended_total(&other));
263
264 
    // Runners: what each did in the range.
265 
    let mut runner_rows: Vec<RunnerRow> = serving
266 
        .iter()
267 
        .filter(|r| scope.owner.is_none() && scope.repo_id.is_none() || jobs.iter().any(|j| j.runner == r.name))
268 
        .map(|r| {
269 
            let mine: Vec<&JobRow> = jobs.iter().filter(|j| j.runner == r.name).collect();
270 
            let mut p: Vec<f64> = mine.iter().filter(|j| j.started_ms.is_some_and(|s| s >= start)).filter_map(|j| j.pickup()).collect();
271 
            let mut d: Vec<f64> = mine.iter().filter(|j| j.ended()).filter_map(|j| j.duration()).collect();
272 
            RunnerRow {
273 
                name: r.name.clone(),
274 
                status: r.status(),
275 
                running: mine.iter().filter(|j| j.status == "running").count() as i64,
276 
                max_jobs: r.max_jobs.max(1),
277 
                jobs: mine.iter().filter(|j| j.ended()).count() as i64,
278 
                failed: mine.iter().filter(|j| j.status == "failed").count() as i64,
279 
                pickup_median: percentile(&mut p, 0.5),
280 
                duration_median: percentile(&mut d, 0.5),
281 
                last_seen_ms: r.last_seen_ms,
282 
            }
283 
        })
284 
        .collect();
285 
    runner_rows.sort_by(|a, b| b.jobs.cmp(&a.jobs).then_with(|| a.name.cmp(&b.name)));
286
287 
    // Repositories, busiest first (a repository's own page has just itself).
288 
    let mut by_repo: std::collections::BTreeMap<String, (i64, i64, Vec<f64>, i64)> = Default::default();
289 
    for j in &jobs {
290 
        let e = by_repo.entry(format!("{}/{}", j.owner, j.repo)).or_default();
291 
        if j.ended() && j.finished_ms.is_some_and(|f| f >= start) {
292 
            e.0 += 1;
293 
            if j.status == "failed" {
294 
                e.1 += 1;
295 
            }
296 
        }
297 
        if let Some(p) = j.pickup().filter(|_| j.started_ms.is_some_and(|s| s >= start)) {
298 
            e.2.push(p);
299 
        }
300 
        if j.status == "queued" {
301 
            e.3 += 1;
302 
        }
303 
    }
304 
    let mut repos: Vec<RepoRow> = by_repo
305 
        .into_iter()
306 
        .map(|(slug, (jobs, failed, mut p, queued))| RepoRow {
307 
            slug,
308 
            jobs,
309 
            failed,
310 
            pickup_median: percentile(&mut p, 0.5),
311 
            pickup_p95: percentile(&mut p, 0.95),
312 
            queued,
313 
        })
314 
        .collect();
315 
    repos.sort_by(|a, b| (b.jobs + b.queued).cmp(&(a.jobs + a.queued)).then_with(|| a.slug.cmp(&b.slug)));
316 
    repos.truncate(15);
317
318 
    Ok(Insights {
319 
        range,
320 
        start_ms: start,
321 
        now_ms: now,
322 
        pickup,
323 
        results: vec![
324 
            Series { name: "passed", class: "s-ok", values: some(passed) },
325 
            Series { name: "failed", class: "s-bad", values: some(failed) },
326 
            Series { name: "other (error, cancelled, skipped)", class: "s-mute", values: some(other) },
327 
        ],
328 
        durations: durations_series,
329 
        load: vec![
330 
            Series { name: "queued", class: "s-warn", values: some(queued_at) },
331 
            Series { name: "running", class: "s-main", values: some(running_at) },
332 
        ],
333 
        utilisation: vec![Series { name: "busy slots", class: "s-main", values: utilisation }],
334 
        ended: p_total + f_total + o_total,
335 
        passed: p_total,
336 
        failed: f_total,
337 
        pickup_median: percentile(&mut all_pickups, 0.5),
338 
        pickup_p95: percentile(&mut all_pickups, 0.95),
339 
        duration_median: percentile(&mut all_durations, 0.5),
340 
        queued_now: jobs.iter().filter(|j| j.status == "queued").count() as i64,
341 
        running_now: jobs.iter().filter(|j| j.status == "running").count() as i64,
342 
        runners_online: serving.iter().filter(|r| r.online()).count() as i64,
343 
        runners_total: serving.len() as i64,
344 
        slots_busy: jobs.iter().filter(|j| j.status == "running").count() as i64,
345 
        slots_total,
346 
        runners: runner_rows,
347 
        repos,
348 
    })
349 
}
350
351 
/// A span of seconds, short: `42 s`, `3.5 min`, `1.2 h`.
352 
pub fn seconds(s: f64) -> String {
353 
    if s < 60.0 {
354 
        format!("{s:.0} s")
355 
    } else if s < 3600.0 {
356 
        let m = s / 60.0;
357 
        if m < 10.0 {
358 
            format!("{m:.1} min")
359 
        } else {
360 
            format!("{m:.0} min")
361 
        }
362 
    } else {
363 
        format!("{:.1} h", s / 3600.0)
364 
    }
365 
}
366
367 
#[cfg(test)]
368 
mod tests {
369 
    use super::*;
370
371 
    #[test]
372 
    fn percentiles_are_nearest_rank() {
373 
        let mut v = vec![5.0, 1.0, 3.0, 2.0, 4.0];
374 
        assert_eq!(percentile(&mut v, 0.5), Some(3.0));
375 
        assert_eq!(percentile(&mut v, 0.95), Some(5.0));
376 
        assert_eq!(percentile(&mut [], 0.5), None);
377 
        assert_eq!(seconds(42.0), "42 s");
378 
        assert_eq!(seconds(210.0), "3.5 min");
379 
        assert_eq!(seconds(4320.0), "1.2 h");
380 
    }
381 
}