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

runnerd_core/api/jobs/
facts.rs
