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

runnerd_core/
sizing.rs
