runnerd_core/
sizing.rs

1//! Right-sizing advice from the job history: the spread of each profile's and each job's sampled peaks, waits and
2//! slot use, and the `memory` and `slots` to set. Pure: the store reads the history, the daemon serves the result
3//! (`api::Sizing`). It never changes the config: a client applies advice through `PUT /v1/config`, which validates
4//! it like any other config, and under `[sizing] auto` the daemon applies what `auto::changes` picks the same way.
5
6use std::collections::BTreeMap;
7
8use crate::api::{self, Advice, Peaks, Spread};
9use crate::config::{Admission, Config};
10use crate::observe::metrics::SAMPLE_INTERVAL;
11
12/// Advice rounds memory up to a multiple of this: 256M.
13pub(crate) const GRANULE: u64 = 256 << 20;
14
15/// Without its container's `memory.events`, a peak at or above this share of its cap, in percent, likely met it.
16const AT_CAP: u64 = 95;
17
18/// Lowering a cap is advised once it saves at least this share of it, in percent.
19const WORTH_LOWERING: u64 = 10;
20
21/// Jobs that waited this long at p95, in seconds, or slots busy this share of the time, in percent, want more
22/// slots.
23const WAITED: u64 = 60;
24const BUSY: u32 = 50;
25
26/// One finished job of this daemon's runners, as sizing reads it. Times in milliseconds.
27#[derive(Debug, Clone, PartialEq, Eq)]
28pub struct Ran {
29    /// `owner/name`.
30    pub repository: String,
31    /// The workflow's name, when known.
32    pub workflow: Option<String>,
33    /// The job's name.
34    pub name: String,
35    /// The profile its runner ran in.
36    pub profile: String,
37    /// Its container's memory cap, when recorded.
38    pub memory_cap: Option<u64>,
39    /// Its sampled memory peak, bytes, as `measure` counts memory.
40    pub memory_peak: u64,
41    /// Its sampled CPU peak, millicores.
42    pub cpu_peak: u64,
43    /// Its CPU time over its run time, millicores; `None` when not recorded.
44    pub cpu_average: Option<u64>,
45    /// When it was queued.
46    pub queued_at: u64,
47    /// When it started; `None` for a job lost before.
48    pub started_at: Option<u64>,
49    /// When it finished, or was found lost.
50    pub finished_at: Option<u64>,
51    /// Whether the kernel killed a process in its container for memory.
52    pub oom_killed: bool,
53    /// Whether its container's memory reached a limit with nothing left to reclaim (`memory.events` `oom`); `None`
54    /// when its events were not recorded.
55    pub hit_cap: Option<bool>,
56    /// What its sampled memory counts.
57    pub measure: Measure,
58    /// Whether it completed with success: a failed or lost run may have stopped before its heaviest step.
59    pub succeeded: bool,
60    /// What measured admission reserved for it; `None` under caps or for an adopted runner.
61    pub reserved: Option<u64>,
62}
63
64/// What a run's sampled memory counts.
65#[derive(Debug, Clone, Copy, PartialEq, Eq)]
66pub enum Measure {
67    /// `memory.current` less `inactive_file`, as runners started before schema version 8 were sampled: page cache
68    /// read twice (`active_file`) counts, so peaks overstate what a job needs.
69    WithActiveFile,
70    /// `memory.current` less the file LRU lists (`active_file`, `inactive_file`): an estimate of what a job holds
71    /// without the page cache the kernel can drop.
72    WithoutPageCache,
73}
74
75impl Ran {
76    /// Whether it met its container's `cap`: killed for memory, or at the cap with nothing left to reclaim. Without
77    /// its container's events, whether its peak came within `AT_CAP` percent of the cap.
78    pub fn met_cap(&self, cap: u64) -> bool {
79        self.oom_killed
80            || self.hit_cap.unwrap_or_else(|| {
81                self.memory_peak.saturating_mul(100) >= cap.saturating_mul(AT_CAP)
82            })
83    }
84}
85
86/// One runner container's time in a slot. Times in milliseconds.
87#[derive(Debug, Clone, PartialEq, Eq)]
88pub struct Held {
89    /// Its profile.
90    pub profile: String,
91    /// When it started.
92    pub started_at: u64,
93    /// When it was removed; `None` while it runs.
94    pub removed_at: Option<u64>,
95}
96
97/// What the store reads for sizing.
98#[derive(Debug, Clone, Default, PartialEq, Eq)]
99pub struct History {
100    /// Finished jobs with a recorded peak, cancelled ones left out.
101    pub jobs: Vec<Ran>,
102    /// Cancelled jobs, for what admission learns from how they ended: a kill for memory, a peak at the cap or past
103    /// what was reserved. Their peaks (0 when never sampled) are not read as a job's size.
104    pub cancelled: Vec<Ran>,
105    /// Runners that held a slot at some point after the history's start.
106    pub runners: Vec<Held>,
107}
108
109/// What `advise` sizes against besides the history.
110#[derive(Debug, Clone, Copy)]
111pub struct Window {
112    /// The start asked for, in milliseconds, at most `now`.
113    pub since: u64,
114    /// Now, in milliseconds.
115    pub now: u64,
116    /// Seconds between two samples of a container.
117    pub sample_interval: u64,
118}
119
120impl Window {
121    /// The window `GET /v1/sizing` reads: from `since` (seconds), default 30 days before `now` (seconds), and
122    /// never after `now`.
123    pub fn new(since: Option<u64>, now: u64) -> Window {
124        Window {
125            since: since
126                .unwrap_or_else(|| now.saturating_sub(30 * 24 * 3600))
127                .min(now)
128                .saturating_mul(1000),
129            now: now.saturating_mul(1000),
130            sample_interval: SAMPLE_INTERVAL.as_secs(),
131        }
132    }
133}
134
135/// The advice for every profile of `config` from `history`, against the slice's `memory_max` as `config`'s
136/// admission policy budgets it.
137pub fn advise(
138    config: &Config,
139    memory_max: Option<u64>,
140    history: &History,
141    window: Window,
142) -> api::Sizing {
143    let since = history
144        .runners
145        .iter()
146        .map(|runner| runner.started_at)
147        .min()
148        .map_or(window.since, |first| first.max(window.since));
149    let span = window.now.saturating_sub(since).max(1);
150    let caps = config.memory_caps().map_or(u64::MAX, |caps| caps.0);
151    let mirror = config.mirror_memory();
152    let rules = &config.sizing;
153    let profiles = config
154        .profiles
155        .iter()
156        .map(|(name, profile)| {
157            let jobs: Vec<&Ran> = history
158                .jobs
159                .iter()
160                .filter(|job| job.profile == *name)
161                .collect();
162            let share = profile.memory.0.saturating_mul(u64::from(profile.slots));
163            let held: u64 = history
164                .runners
165                .iter()
166                .filter(|runner| runner.profile == *name)
167                .map(|runner| {
168                    let end = runner.removed_at.unwrap_or(window.now).min(window.now);
169                    end.saturating_sub(runner.started_at.max(since))
170                })
171                .sum();
172            let busy = u32::try_from(
173                (held.saturating_mul(100) / span.saturating_mul(u64::from(profile.slots).max(1)))
174                    .min(100),
175            )
176            .unwrap_or(100);
177            let mut keys: BTreeMap<(&str, Option<&str>, &str), Vec<&Ran>> = BTreeMap::new();
178            for job in &jobs {
179                keys.entry((&job.repository, job.workflow.as_deref(), &job.name))
180                    .or_default()
181                    .push(job);
182            }
183            let wait = waits(&jobs);
184            let sized = Sized {
185                cap: profile.memory.0,
186                slots: profile.slots,
187                share,
188                admission: config.host.admission,
189                room: memory_max.map(|max| match config.host.admission {
190                    // What the slice leaves this profile once the other profiles and the mirror took theirs.
191                    Admission::Caps => max.saturating_sub(caps.saturating_sub(share)),
192                    // Measured admission reserves predicted peaks, not caps: the config check only holds one
193                    // runner of each profile beside the mirror.
194                    Admission::Measured => max.saturating_sub(mirror),
195                }),
196                wants_slots: wait.p95 >= WAITED || busy >= BUSY,
197                margin: rules.margin,
198                min_jobs: rules.min_jobs,
199            };
200            let job_sizings: Vec<api::JobSizing> = keys
201                .into_iter()
202                .map(|((repository, workflow, job_name), runs)| {
203                    let peaks = peaks(&runs);
204                    let wait = waits(&runs);
205                    api::JobSizing {
206                        repository: repository.to_owned(),
207                        workflow: workflow.map(str::to_owned),
208                        name: job_name.to_owned(),
209                        advice: sized.advice(&peaks, wait.p95 >= WAITED || busy >= BUSY),
210                        peaks,
211                        wait,
212                    }
213                })
214                .collect();
215            let peaks = peaks(&jobs);
216            api::ProfileSizing {
217                name: name.clone(),
218                memory: profile.memory.0,
219                cpus: profile.cpus,
220                slots: profile.slots,
221                advice: sized.advice(&peaks, sized.wants_slots),
222                peaks,
223                wait,
224                busy,
225                wants_slots: sized.wants_slots,
226                jobs: job_sizings,
227            }
228        })
229        .collect();
230    api::Sizing {
231        since: since / 1000,
232        until: window.now / 1000,
233        margin: rules.margin,
234        sample_interval: window.sample_interval,
235        min_jobs: rules.min_jobs,
236        memory_max,
237        admission: match config.host.admission {
238            Admission::Caps => api::AdmissionPolicy::Caps,
239            Admission::Measured => api::AdmissionPolicy::Measured,
240        },
241        profiles,
242        auto: config.sizing.auto,
243        changes: Vec::new(),
244    }
245}
246
247/// A profile's limits and the rules advice follows.
248struct Sized {
249    cap: u64,
250    slots: u32,
251    /// What its slots take of the budget now: `memory × slots`.
252    share: u64,
253    admission: Admission,
254    /// The memory the slice holds for it, as `admission` budgets it: all its slots at their caps under `caps`, one
255    /// runner under `measured`; `None` for an unlimited slice.
256    room: Option<u64>,
257    /// Whether its jobs waited for a slot at p95, or its slots were busy, enough to want more.
258    wants_slots: bool,
259    margin: u32,
260    min_jobs: u32,
261}
262
263impl Sized {
264    /// The memory a set of jobs needs: its largest peak plus the margin, rounded up to `GRANULE`. The largest, not
265    /// a percentile: the cap kills a job that goes over it, so a cap below a peak a job reached kills it again.
266    fn need(&self, peaks: &Peaks) -> u64 {
267        let largest = peaks.memory.max;
268        let need = largest.saturating_add(largest.saturating_mul(u64::from(self.margin)) / 100);
269        need.div_ceil(GRANULE).saturating_mul(GRANULE).max(GRANULE)
270    }
271
272    fn advice(&self, peaks: &Peaks, wants_slots: bool) -> Advice {
273        if peaks.jobs < u64::from(self.min_jobs) {
274            return Advice::TooFew { jobs: peaks.jobs };
275        }
276        let need = self.need(peaks);
277        let fit = |memory: u64| u32::try_from(memory / need).unwrap_or(u32::MAX);
278        if need > self.cap {
279            let slots = match (self.admission, self.room) {
280                (Admission::Caps, Some(room)) => fit(room).min(self.slots),
281                (Admission::Measured, Some(room)) if room < need => 0,
282                (_, _) => self.slots,
283            };
284            if slots == 0 {
285                return Advice::SliceTooSmall { memory: need };
286            }
287            return Advice::Raise {
288                memory: need,
289                slots,
290            };
291        }
292        // A run at its cap, this one or a smaller one it had then, may have been held back or killed by it: its
293        // peak is a floor, not what it needs, so the cap is never lowered under it.
294        if peaks.at_cap == 0 && self.cap - need >= self.cap * WORTH_LOWERING / 100 {
295            // Under measured admission slots are not bought with caps: the advice line says when more may help.
296            return Advice::Lower {
297                memory: need,
298                slots: if wants_slots && self.admission == Admission::Caps {
299                    fit(self.share).max(self.slots)
300                } else {
301                    self.slots
302                },
303            };
304        }
305        Advice::Keep { memory: need }
306    }
307}
308
309fn peaks(jobs: &[&Ran]) -> Peaks {
310    let at_cap = jobs
311        .iter()
312        .filter(|job| job.memory_cap.is_some_and(|cap| job.met_cap(cap)))
313        .count();
314    Peaks {
315        jobs: jobs.len() as u64,
316        at_cap: at_cap as u64,
317        memory: spread(jobs.iter().map(|job| job.memory_peak).collect()),
318        cpu: spread(jobs.iter().map(|job| job.cpu_peak).collect()),
319    }
320}
321
322/// Seconds from queued to started of the jobs that started.
323fn waits(jobs: &[&Ran]) -> Spread {
324    spread(
325        jobs.iter()
326            .filter_map(|job| {
327                job.started_at
328                    .map(|at| at.saturating_sub(job.queued_at) / 1000)
329            })
330            .collect(),
331    )
332}
333
334/// Nearest-rank percentiles; all zero for no values.
335pub(crate) fn spread(mut values: Vec<u64>) -> Spread {
336    values.sort_unstable();
337    let rank = |percent: usize| {
338        let index = (values.len() * percent).div_ceil(100).saturating_sub(1);
339        values.get(index).copied().unwrap_or(0)
340    };
341    Spread {
342        p50: rank(50),
343        p95: rank(95),
344        p99: rank(99),
345        max: values.last().copied().unwrap_or(0),
346    }
347}
348
349pub mod auto;
350
351#[cfg(test)]
352mod tests;
runnerd_core/sizing.rs

runnerd_core/
sizing.rs
