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