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

runnerd_core/
sizing.rs
