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.
5
6use std::collections::BTreeMap;
7
8use crate::api::{self, Advice, Peaks, Spread};
9use crate::config::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/// A peak at or above this share of its container's cap, in percent, likely met the cap.
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.
40    pub memory_peak: u64,
41    /// Its sampled CPU peak, millicores.
42    pub cpu_peak: u64,
43    /// When it was queued.
44    pub queued_at: u64,
45    /// When it started; `None` for a job lost before.
46    pub started_at: Option<u64>,
47    /// When it finished, or was found lost.
48    pub finished_at: Option<u64>,
49    /// Whether the kernel killed a process in its container for memory.
50    pub oom_killed: bool,
51    /// What measured admission reserved for it; `None` under caps or for an adopted runner.
52    pub reserved: Option<u64>,
53}
54
55/// One runner container's time in a slot. Times in milliseconds.
56#[derive(Debug, Clone, PartialEq, Eq)]
57pub struct Held {
58    /// Its profile.
59    pub profile: String,
60    /// When it started.
61    pub started_at: u64,
62    /// When it was removed; `None` while it runs.
63    pub removed_at: Option<u64>,
64}
65
66/// What the store reads for sizing.
67#[derive(Debug, Clone, Default, PartialEq, Eq)]
68pub struct History {
69    /// Finished jobs with a recorded peak, cancelled ones left out.
70    pub jobs: Vec<Ran>,
71    /// Cancelled jobs, for what admission learns from how they ended: a kill for memory, a peak at the cap or past
72    /// what was reserved. Their peaks (0 when never sampled) are not read as a job's size.
73    pub cancelled: Vec<Ran>,
74    /// Runners that held a slot at some point after the history's start.
75    pub runners: Vec<Held>,
76}
77
78/// What `advise` sizes against besides the history.
79#[derive(Debug, Clone, Copy)]
80pub struct Window {
81    /// The start asked for, in milliseconds, at most `now`.
82    pub since: u64,
83    /// Now, in milliseconds.
84    pub now: u64,
85    /// Seconds between two samples of a container.
86    pub sample_interval: u64,
87}
88
89impl Window {
90    /// The window `GET /v1/sizing` reads: from `since` (seconds), default 30 days before `now` (seconds), and
91    /// never after `now`.
92    pub fn new(since: Option<u64>, now: u64) -> Window {
93        Window {
94            since: since
95                .unwrap_or_else(|| now.saturating_sub(30 * 24 * 3600))
96                .min(now)
97                .saturating_mul(1000),
98            now: now.saturating_mul(1000),
99            sample_interval: SAMPLE_INTERVAL.as_secs(),
100        }
101    }
102}
103
104/// The advice for every profile of `config` from `history`, against the slice's `memory_max`.
105pub fn advise(
106    config: &Config,
107    memory_max: Option<u64>,
108    history: &History,
109    window: Window,
110) -> api::Sizing {
111    let since = history
112        .runners
113        .iter()
114        .map(|runner| runner.started_at)
115        .min()
116        .map_or(window.since, |first| first.max(window.since));
117    let span = window.now.saturating_sub(since).max(1);
118    let caps = config.memory_caps().map_or(u64::MAX, |caps| caps.0);
119    let rules = &config.sizing;
120    let profiles = config
121        .profiles
122        .iter()
123        .map(|(name, profile)| {
124            let jobs: Vec<&Ran> = history
125                .jobs
126                .iter()
127                .filter(|job| job.profile == *name)
128                .collect();
129            let share = profile.memory.0.saturating_mul(u64::from(profile.slots));
130            let held: u64 = history
131                .runners
132                .iter()
133                .filter(|runner| runner.profile == *name)
134                .map(|runner| {
135                    let end = runner.removed_at.unwrap_or(window.now).min(window.now);
136                    end.saturating_sub(runner.started_at.max(since))
137                })
138                .sum();
139            let busy = u32::try_from(
140                (held.saturating_mul(100) / span.saturating_mul(u64::from(profile.slots).max(1)))
141                    .min(100),
142            )
143            .unwrap_or(100);
144            let mut keys: BTreeMap<(&str, Option<&str>, &str), Vec<&Ran>> = BTreeMap::new();
145            for job in &jobs {
146                keys.entry((&job.repository, job.workflow.as_deref(), &job.name))
147                    .or_default()
148                    .push(job);
149            }
150            let sized = Sized {
151                cap: profile.memory.0,
152                slots: profile.slots,
153                share,
154                // What the slice leaves this profile once the other profiles and the mirror took theirs.
155                room: memory_max.map(|max| max.saturating_sub(caps.saturating_sub(share))),
156                margin: rules.margin,
157                min_jobs: rules.min_jobs,
158            };
159            let job_sizings: Vec<api::JobSizing> = keys
160                .into_iter()
161                .map(|((repository, workflow, job_name), runs)| {
162                    let peaks = peaks(&runs);
163                    let wait = waits(&runs);
164                    api::JobSizing {
165                        repository: repository.to_owned(),
166                        workflow: workflow.map(str::to_owned),
167                        name: job_name.to_owned(),
168                        advice: sized.advice(&peaks, &wait, busy),
169                        peaks,
170                        wait,
171                    }
172                })
173                .collect();
174            let peaks = peaks(&jobs);
175            let wait = waits(&jobs);
176            api::ProfileSizing {
177                name: name.clone(),
178                memory: profile.memory.0,
179                cpus: profile.cpus,
180                slots: profile.slots,
181                advice: sized.advice(&peaks, &wait, busy),
182                peaks,
183                wait,
184                busy,
185                jobs: job_sizings,
186            }
187        })
188        .collect();
189    api::Sizing {
190        since: since / 1000,
191        until: window.now / 1000,
192        margin: rules.margin,
193        sample_interval: window.sample_interval,
194        min_jobs: rules.min_jobs,
195        memory_max,
196        profiles,
197    }
198}
199
200/// A profile's limits and the rules advice follows.
201struct Sized {
202    cap: u64,
203    slots: u32,
204    /// What its slots take of the budget now: `memory × slots`.
205    share: u64,
206    /// The memory its slots may take of the slice; `None` for an unlimited slice.
207    room: Option<u64>,
208    margin: u32,
209    min_jobs: u32,
210}
211
212impl Sized {
213    /// The memory a set of jobs needs: its largest peak plus the margin, rounded up to `GRANULE`. The largest, not
214    /// a percentile: the cap kills a job that goes over it, so a cap below a peak a job reached kills it again.
215    fn need(&self, peaks: &Peaks) -> u64 {
216        let largest = peaks.memory.max;
217        let need = largest.saturating_add(largest.saturating_mul(u64::from(self.margin)) / 100);
218        need.div_ceil(GRANULE).saturating_mul(GRANULE).max(GRANULE)
219    }
220
221    fn advice(&self, peaks: &Peaks, wait: &Spread, busy: u32) -> Advice {
222        if peaks.jobs < u64::from(self.min_jobs) {
223            return Advice::TooFew { jobs: peaks.jobs };
224        }
225        let need = self.need(peaks);
226        let fit = |memory: u64| u32::try_from(memory / need).unwrap_or(u32::MAX);
227        if need > self.cap {
228            let slots = match self.room {
229                Some(room) => fit(room).min(self.slots),
230                None => self.slots,
231            };
232            if slots == 0 {
233                return Advice::SliceTooSmall { memory: need };
234            }
235            return Advice::Raise {
236                memory: need,
237                slots,
238            };
239        }
240        // A run at its cap, this one or a smaller one it had then, may have been held back or killed by it: its
241        // peak is a floor, not what it needs, so the cap is never lowered under it.
242        if peaks.at_cap == 0 && self.cap - need >= self.cap * WORTH_LOWERING / 100 {
243            let wanted = wait.p95 >= WAITED || busy >= BUSY;
244            return Advice::Lower {
245                memory: need,
246                slots: if wanted {
247                    fit(self.share).max(self.slots)
248                } else {
249                    self.slots
250                },
251            };
252        }
253        Advice::Keep { memory: need }
254    }
255}
256
257fn peaks(jobs: &[&Ran]) -> Peaks {
258    let at_cap = jobs
259        .iter()
260        .filter(|job| {
261            job.memory_cap.is_some_and(|cap| {
262                job.memory_peak.saturating_mul(100) >= cap.saturating_mul(AT_CAP)
263            })
264        })
265        .count();
266    Peaks {
267        jobs: jobs.len() as u64,
268        at_cap: at_cap as u64,
269        memory: spread(jobs.iter().map(|job| job.memory_peak).collect()),
270        cpu: spread(jobs.iter().map(|job| job.cpu_peak).collect()),
271    }
272}
273
274/// Seconds from queued to started of the jobs that started.
275fn waits(jobs: &[&Ran]) -> Spread {
276    spread(
277        jobs.iter()
278            .filter_map(|job| {
279                job.started_at
280                    .map(|at| at.saturating_sub(job.queued_at) / 1000)
281            })
282            .collect(),
283    )
284}
285
286/// Nearest-rank percentiles; all zero for no values.
287pub(crate) fn spread(mut values: Vec<u64>) -> Spread {
288    values.sort_unstable();
289    let rank = |percent: usize| {
290        let index = (values.len() * percent).div_ceil(100).saturating_sub(1);
291        values.get(index).copied().unwrap_or(0)
292    };
293    Spread {
294        p50: rank(50),
295        p95: rank(95),
296        p99: rank(99),
297        max: values.last().copied().unwrap_or(0),
298    }
299}
300
301#[cfg(test)]
302mod tests;
runnerd_core/sizing.rs

runnerd_core/
sizing.rs
