runnerd_core/api/jobs/
facts.rs

1//! What every client shows of a job: its name, its outcome and its steps', how long it waited and ran, its phases
2//! and milestones, its figures as samples come in, what it used over a step, and how it compares with the same
3//! job's latest runs. Each returns meaning, a label, a figure or an [`OutcomeKind`]; the dashboard, the TUI and
4//! `runnerctl` only draw it, each colouring a kind its own way. Nothing here formats for one client.
5
6use std::borrow::Cow;
7use std::fmt;
8
9use super::{Job, JobState, JobSummary, Sample, SampleSeries, Step, Timeline};
10use crate::api::{QueuedJob, Record};
11
12/// What kind of outcome a job's or a step's is, for each client to colour as it colours that meaning.
13#[derive(Debug, Clone, Copy, PartialEq, Eq)]
14pub enum OutcomeKind {
15    /// It succeeded.
16    Succeeded,
17    /// It failed.
18    Failed,
19    /// It is under way.
20    Running,
21    /// It has not begun: waiting for a runner, or a step not reached yet.
22    Waiting,
23    /// It ended without succeeding or failing: cancelled, skipped, run elsewhere, or ended before any runner took it.
24    Ended,
25}
26
27/// A job's or a step's outcome: its word and what it means.
28#[derive(Debug, Clone, PartialEq, Eq)]
29pub struct Outcome {
30    /// Such as `success`, `timed out` or `running`.
31    pub label: Cow<'static, str>,
32    /// What it means.
33    pub kind: OutcomeKind,
34}
35
36impl Outcome {
37    fn new(label: &'static str, kind: OutcomeKind) -> Outcome {
38        Outcome {
39            label: Cow::Borrowed(label),
40            kind,
41        }
42    }
43
44    /// A conclusion of GitHub's, its underscores as spaces: `timed_out` is `timed out`.
45    fn concluded(conclusion: &str) -> Outcome {
46        let kind = match conclusion {
47            "success" => OutcomeKind::Succeeded,
48            failed if failed_conclusion(failed) => OutcomeKind::Failed,
49            _ => OutcomeKind::Ended,
50        };
51        Outcome {
52            label: Cow::Owned(conclusion.replace('_', " ")),
53            kind,
54        }
55    }
56}
57
58/// The conclusions that count as a failure, as `GET /v1/stats` and the insights count them.
59fn failed_conclusion(conclusion: &str) -> bool {
60    matches!(conclusion, "failure" | "timed_out")
61}
62
63impl JobSummary {
64    /// Whether it failed, as `GET /v1/stats` counts failures: it concluded `failure` or `timed_out`, or it lost its
65    /// runner. A job another machine ran is not this daemon's to count.
66    pub fn is_failure(&self) -> bool {
67        match self.state {
68            JobState::Lost => true,
69            JobState::Completed => self.conclusion.as_deref().is_some_and(failed_conclusion),
70            JobState::Queued | JobState::Running | JobState::Elsewhere => false,
71        }
72    }
73
74    /// Its outcome: where it is while it can change, then how it ended. Its kind is [`OutcomeKind::Failed`] exactly
75    /// when it [failed](JobSummary::is_failure). A job that left the queue without starting, as one a newer push
76    /// cancelled, did not run elsewhere: it reads `cancelled in queue`, or as it concluded.
77    pub fn outcome(&self) -> Outcome {
78        let started = self.timeline.started_at.is_some();
79        match (self.state, self.conclusion.as_deref()) {
80            (JobState::Queued, _) => Outcome::new("queued", OutcomeKind::Waiting),
81            (JobState::Running, _) => Outcome::new("running", OutcomeKind::Running),
82            (JobState::Lost, _) => Outcome::new("lost", OutcomeKind::Failed),
83            (JobState::Elsewhere, Some("cancelled")) if !started => {
84                Outcome::new("cancelled in queue", OutcomeKind::Ended)
85            }
86            (JobState::Elsewhere, Some(conclusion)) if !started => Outcome {
87                kind: OutcomeKind::Ended,
88                ..Outcome::concluded(conclusion)
89            },
90            (JobState::Elsewhere, _) => Outcome::new("elsewhere", OutcomeKind::Ended),
91            (JobState::Completed, None) => Outcome::new("completed", OutcomeKind::Ended),
92            (JobState::Completed, Some(conclusion)) => Outcome::concluded(conclusion),
93        }
94    }
95
96    /// Seconds it waited for a runner: until it started, or until `now` while it may still start. `None` for a job
97    /// that ended without starting here.
98    pub fn waited(&self, now: u64) -> Option<u64> {
99        let end = match self.timeline.started_at {
100            Some(started) => started,
101            None if !self.state.is_final() => now,
102            None => return None,
103        };
104        Some(end.saturating_sub(self.timeline.queued_at))
105    }
106
107    /// Seconds it ran: until it finished, or until `now` while it may still finish. `None` before it started, and
108    /// for a final job without a finish time, which no client claims ran for as long as it has been since.
109    pub fn ran(&self, now: u64) -> Option<u64> {
110        let started = self.timeline.started_at?;
111        let end = match self.timeline.finished_at {
112            Some(finished) => finished,
113            None if !self.state.is_final() => now,
114            None => return None,
115        };
116        Some(end.saturating_sub(started))
117    }
118
119    /// Its phases in order, each from when it began to when it ended, `None` while it goes on. A phase without an
120    /// end goes on while the job can change; a final job's ends when the job finished, or is left out when no time
121    /// says how long it took. Cleaning up follows the finish of a job whose runner started here, until its cleanup.
122    pub fn phases(&self) -> Vec<Phase> {
123        let timeline = &self.timeline;
124        let open = !self.state.is_final();
125        // `Some(end)` for a phase to show, `None` for one to leave out.
126        let until = |end: Option<u64>| match end {
127            Some(end) => Some(Some(end)),
128            None if open => Some(None),
129            None => timeline.finished_at.map(Some),
130        };
131        let runner = timeline.runner_from();
132        let mut phases = Vec::new();
133        if let Some(to) = until(runner.or(timeline.started_at)) {
134            phases.push(Phase {
135                kind: PhaseKind::Waiting,
136                from: timeline.queued_at,
137                to,
138            });
139        }
140        if let Some(from) = runner
141            && let Some(to) = until(timeline.started_at)
142        {
143            phases.push(Phase {
144                kind: PhaseKind::Starting,
145                from,
146                to,
147            });
148        }
149        if let Some(from) = timeline.started_at
150            && let Some(to) = timeline.finished_at.map(Some).or(open.then_some(None))
151        {
152            phases.push(Phase {
153                kind: PhaseKind::Running,
154                from,
155                to,
156            });
157        }
158        if let Some(from) = timeline.finished_at
159            && runner.is_some()
160        {
161            phases.push(Phase {
162                kind: PhaseKind::CleaningUp,
163                from,
164                to: timeline.cleaned_at,
165            });
166        }
167        phases
168    }
169
170    /// Its name as every client writes it: `workflow / job`.
171    pub fn title(&self) -> JobTitle<'_> {
172        job_title(self.workflow.as_deref(), &self.name, Some(self.id))
173    }
174}
175
176/// A stretch of a job's life.
177#[derive(Debug, Clone, Copy, PartialEq, Eq)]
178pub struct Phase {
179    /// Which.
180    pub kind: PhaseKind,
181    /// When it began.
182    pub from: u64,
183    /// When it ended; `None` while it goes on.
184    pub to: Option<u64>,
185}
186
187impl Phase {
188    /// How long it took, or has taken by `now`.
189    pub fn seconds(&self, now: u64) -> u64 {
190        self.to.unwrap_or(now).saturating_sub(self.from)
191    }
192}
193
194/// The phases of a job, in the order they happen.
195#[derive(Debug, Clone, Copy, PartialEq, Eq)]
196pub enum PhaseKind {
197    /// Queued, without a runner.
198    Waiting,
199    /// A runner took it and is starting.
200    Starting,
201    /// The runner runs it.
202    Running,
203    /// It finished; its container and registration are being removed.
204    CleaningUp,
205}
206
207impl PhaseKind {
208    /// Its name, as every client labels it.
209    pub fn label(self) -> &'static str {
210        match self {
211            PhaseKind::Waiting => "Waiting for a runner",
212            PhaseKind::Starting => "Runner starting",
213            PhaseKind::Running => "Running",
214            PhaseKind::CleaningUp => "Cleaning up",
215        }
216    }
217}
218
219/// A moment of a job's life.
220#[derive(Debug, Clone, Copy, PartialEq, Eq)]
221pub struct Milestone {
222    /// Such as `Container started`.
223    pub label: &'static str,
224    /// When.
225    pub at: u64,
226}
227
228impl Timeline {
229    /// The moments it has times for, earliest first: a warm runner's container started before the job was queued.
230    pub fn milestones(&self) -> Vec<Milestone> {
231        let mut milestones: Vec<Milestone> = [
232            ("Queued", Some(self.queued_at)),
233            ("Container started", self.container_started_at),
234            ("Job started", self.started_at),
235            ("Finished", self.finished_at),
236            ("Cleaned up", self.cleaned_at),
237        ]
238        .into_iter()
239        .filter_map(|(label, at)| Some(Milestone { label, at: at? }))
240        .collect();
241        milestones.sort_by_key(|milestone| milestone.at);
242        milestones
243    }
244}
245
246impl Step {
247    /// Its outcome: how it concluded, else whether it runs or waits. A step that timed out failed, one that was
248    /// cancelled or skipped did not.
249    pub fn outcome(&self) -> Outcome {
250        match (self.conclusion.as_deref(), self.status.as_str()) {
251            (Some(conclusion), _) => Outcome::concluded(conclusion),
252            (None, "in_progress") => Outcome::new("running", OutcomeKind::Running),
253            (None, "completed") => Outcome::new("completed", OutcomeKind::Ended),
254            (None, _) => Outcome::new("pending", OutcomeKind::Waiting),
255        }
256    }
257
258    /// Whether it runs now.
259    pub fn is_running(&self) -> bool {
260        self.status == "in_progress" && self.conclusion.is_none()
261    }
262
263    /// Seconds it took, until `now` while it runs; `None` before it started and for a step that ended without a
264    /// completion time.
265    pub fn took(&self, now: u64) -> Option<u64> {
266        let started = self.started_at?;
267        let end = match self.completed_at {
268            Some(completed) => completed,
269            None if self.is_running() => now,
270            None => return None,
271        };
272        Some(end.saturating_sub(started))
273    }
274}
275
276/// A job's name as every client writes it, in parts for a page that sets the workflow apart.
277#[derive(Debug, Clone, PartialEq, Eq)]
278pub struct JobTitle<'job> {
279    /// The workflow's name, when known.
280    pub workflow: Option<&'job str>,
281    /// The job's name, or `job <id>` until the daemon knows it.
282    pub name: Cow<'job, str>,
283}
284
285/// `workflow / job`, or the job's name alone; for a job whose name the daemon does not know yet, as for one its
286/// runner took before the daemon saw it queued, `job <id>`, or `a job` without an ID.
287pub fn job_title<'job>(
288    workflow: Option<&'job str>,
289    name: &'job str,
290    id: Option<u64>,
291) -> JobTitle<'job> {
292    let name = match (name, id) {
293        ("", Some(id)) => Cow::Owned(format!("job {id}")),
294        ("", None) => Cow::Borrowed("a job"),
295        (name, _) => Cow::Borrowed(name),
296    };
297    JobTitle { workflow, name }
298}
299
300impl fmt::Display for JobTitle<'_> {
301    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
302        match self.workflow {
303            Some(workflow) => write!(f, "{workflow} / {}", self.name),
304            None => f.write_str(&self.name),
305        }
306    }
307}
308
309impl QueuedJob {
310    /// Its name as [`JobSummary::title`] writes a job's.
311    pub fn title(&self) -> JobTitle<'_> {
312        job_title(self.workflow.as_deref(), &self.name, Some(self.id))
313    }
314}
315
316impl Record {
317    /// Its job's name as [`JobSummary::title`] writes a job's.
318    pub fn title(&self) -> JobTitle<'_> {
319        job_title(
320            self.workflow.as_deref(),
321            self.name.as_deref().unwrap_or_default(),
322            Some(self.job),
323        )
324    }
325}
326
327impl Job {
328    /// Takes a sample of its container as it arrives: while the job runs, its peaks rise to it, high-water marks
329    /// that deleting old samples does not lower. A final job's figures are the daemon's totals and stay.
330    pub fn observe(&mut self, sample: &Sample) {
331        let metrics = &mut self.summary.metrics;
332        if !self.summary.state.is_final() {
333            metrics.memory_peak = metrics.memory_peak.max(Some(sample.memory));
334            metrics.cpu_peak = metrics.cpu_peak.max(Some(sample.cpu));
335        }
336    }
337
338    /// Takes the job as read again. While it runs, its peaks keep the highest seen, since the store's lag the
339    /// samples a client saw arrive; once it is final, the read is the daemon's totals.
340    pub fn reread(&mut self, read: Job) {
341        let seen = self.summary.metrics.clone();
342        *self = read;
343        if !self.summary.state.is_final() {
344            let metrics = &mut self.summary.metrics;
345            metrics.memory_peak = metrics.memory_peak.max(seen.memory_peak);
346            metrics.cpu_peak = metrics.cpu_peak.max(seen.cpu_peak);
347        }
348    }
349
350    /// How it compares with the same job's latest `runs`, oldest first as given: this job among them as it is now,
351    /// not as the runs were read, since it may have started, finished or peaked higher since.
352    pub fn compare(&self, runs: &[JobSummary], now: u64) -> RunComparison {
353        let runs: Vec<JobSummary> = runs
354            .iter()
355            .map(|run| {
356                if run.id == self.summary.id {
357                    self.summary.clone()
358                } else {
359                    run.clone()
360                }
361            })
362            .collect();
363        // Another machine's runs are not this daemon's to compare.
364        let finished: Vec<&JobSummary> = runs
365            .iter()
366            .filter(|run| {
367                matches!(run.state, JobState::Completed | JobState::Lost)
368                    && run.timeline.started_at.is_some()
369            })
370            .collect();
371        // The lower of the middle two for an even count: a duration one of the runs took.
372        let median = |mut values: Vec<u64>| {
373            values.sort_unstable();
374            values.get(values.len().saturating_sub(1) / 2).copied()
375        };
376        RunComparison {
377            median_ran: median(finished.iter().filter_map(|run| run.ran(now)).collect()),
378            median_waited: median(finished.iter().filter_map(|run| run.waited(now)).collect()),
379            finished: finished.len(),
380            failed: finished.iter().filter(|run| run.is_failure()).count(),
381            largest_peak: runs.iter().filter_map(|run| run.metrics.memory_peak).max(),
382            runs,
383        }
384    }
385}
386
387/// A job's latest runs and what they say together, from [`Job::compare`].
388#[derive(Debug, Clone, PartialEq, Eq)]
389pub struct RunComparison {
390    /// Oldest first, this job as it is now.
391    pub runs: Vec<JobSummary>,
392    /// The median of the finished runs' run times.
393    pub median_ran: Option<u64>,
394    /// The median of the finished runs' waits.
395    pub median_waited: Option<u64>,
396    /// The runs that started and ended here: those the medians and failures count.
397    pub finished: usize,
398    /// Of those, the ones that [failed](JobSummary::is_failure).
399    pub failed: usize,
400    /// The largest memory peak of any run.
401    pub largest_peak: Option<u64>,
402}
403
404impl SampleSeries {
405    /// What the container used from `from` to `to`, from the samples taken after `from` up to `to`: each covers the
406    /// interval before it, so steps that meet at a second share no sample. `None` when none was taken in it.
407    pub fn usage_between(&self, from: u64, to: u64) -> Option<SampleUsage> {
408        let inside: Vec<&Sample> = self
409            .samples
410            .iter()
411            .filter(|sample| from < sample.at && sample.at <= to)
412            .collect();
413        let count = u64::try_from(inside.len())
414            .ok()
415            .filter(|count| *count > 0)?;
416        Some(SampleUsage {
417            memory_peak: inside.iter().map(|sample| sample.memory).max()?,
418            cpu_average: inside.iter().map(|sample| sample.cpu).sum::<u64>() / count,
419        })
420    }
421}
422
423/// What a container used over a stretch of its run, from [`SampleSeries::usage_between`].
424#[derive(Debug, Clone, Copy, PartialEq, Eq)]
425pub struct SampleUsage {
426    /// The highest memory of the samples, without page cache.
427    pub memory_peak: u64,
428    /// The samples' CPU use averaged, in millicores.
429    pub cpu_average: u64,
430}
431
432impl SampleUsage {
433    /// `peak 4.1G`.
434    pub fn memory(&self) -> String {
435        format!("peak {}", crate::config::ByteSize(self.memory_peak))
436    }
437
438    /// `average 1.5 cores`.
439    pub fn cpu(&self) -> String {
440        format!("average {}", super::super::in_cores(self.cpu_average))
441    }
442
443    /// Both, `peak 4.1G, average 1.5 cores`.
444    pub fn description(&self) -> String {
445        format!("{}, {}", self.memory(), self.cpu())
446    }
447}
448
449#[cfg(test)]
450mod tests;
runnerd_core/api/jobs/facts.rs

runnerd_core/api/jobs/
facts.rs
