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

runnerd_core/
sizing.rs
