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 crate::admission::Key;
7use crate::admission::recent::Recent;
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, or a descendant's, reached a limit with nothing left to reclaim
57 /// (`memory.events` `oom`); `None` when its events were not recorded.
58 pub hit_cap: Option<bool>,
59 /// Whether its container's own limit failed reclaim (`memory.events.local` `oom`), not a descendant's; `None`
60 /// when not recorded, as before schema 14.
61 pub own_cap: Option<bool>,
62 /// What its sampled memory counts.
63 pub measure: Measure,
64 /// Whether it completed with success: a failed or lost run may have stopped before its heaviest step.
65 pub succeeded: bool,
66 /// Whether it completed with `failure` or `timed_out`: not lost, not cancelled.
67 pub failed: bool,
68 /// Its workflow run on GitHub: a re-run of its failed jobs has the same.
69 pub run_id: u64,
70 /// The branch it ran for, when known.
71 pub branch: Option<String>,
72 /// What measured admission reserved for it; `None` under caps or for an adopted runner.
73 pub reserved: Option<u64>,
74 /// How long of its wait it waited because the host's memory could not take its runner, in ms: no more slots
75 /// would have started it sooner.
76 pub host_waited: u64,
77}
78
79/// What a run's sampled memory counts.
80#[derive(Debug, Clone, Copy, PartialEq, Eq)]
81pub enum Measure {
82 /// `memory.current` less `inactive_file`, as runners started before schema version 8 were sampled: page cache
83 /// read twice (`active_file`) counts, so peaks overstate what a job needs.
84 WithActiveFile,
85 /// `memory.current` less the file LRU lists (`active_file`, `inactive_file`): an estimate of what a job holds
86 /// without the page cache the kernel can drop.
87 WithoutPageCache,
88}
89
90/// What a run says about its container's memory cap (ADR 0070).
91#[derive(Debug, Clone, Copy, PartialEq, Eq)]
92pub enum Capped {
93 /// The container's own limit failed reclaim: it needed more than the cap.
94 Met,
95 /// Recorded without the local counter: by the reading before it, a kill or a limit hit in the container, or
96 /// without events a peak within `AT_CAP` percent of the cap. Holds and floors as `Met`, never raises a cap.
97 Assumed,
98 /// A process in it was killed for memory while its own limit held: the slice or the host ran out.
99 Killed,
100 /// None of those. A peak near the cap with no `oom` at it is clear: a JVM fills the room it has.
101 Clear,
102}
103
104impl Capped {
105 /// Whether it says the job needed its cap: `Met` or `Assumed`.
106 pub fn met(self) -> bool {
107 matches!(self, Capped::Met | Capped::Assumed)
108 }
109
110 /// Whether it puts its job on its cap: anything but `Clear`.
111 pub fn holds(self) -> bool {
112 self != Capped::Clear
113 }
114}
115
116impl Ran {
117 /// What it says about its container's `cap`. A kill that only a descendant's own limit explains, the counters
118 /// showing a limit hit below the container and none at it, is a service container the job capped itself: clear.
119 /// A kill no sample saw an `oom` before, its last sample within `AT_CAP` percent of the cap, met it in the last
120 /// interval.
121 pub fn capped(&self, cap: u64) -> Capped {
122 match self.own_cap {
123 Some(true) => Capped::Met,
124 // No sample saw `oom` at any level and its last came near the cap: its own limit killed it before the
125 // next sample could count it.
126 Some(false)
127 if self.oom_killed
128 && self.hit_cap == Some(false)
129 && self.memory_peak.saturating_mul(100) >= cap.saturating_mul(AT_CAP) =>
130 {
131 Capped::Met
132 }
133 Some(false) if self.oom_killed && self.hit_cap != Some(true) => Capped::Killed,
134 None if self.oom_killed
135 || self.hit_cap.unwrap_or_else(|| {
136 self.memory_peak.saturating_mul(100) >= cap.saturating_mul(AT_CAP)
137 }) =>
138 {
139 Capped::Assumed
140 }
141 Some(false) | None => Capped::Clear,
142 }
143 }
144}
145
146/// One runner container's time in a slot for a job. Times in milliseconds.
147#[derive(Debug, Clone, PartialEq, Eq)]
148pub struct Held {
149 /// Its profile.
150 pub profile: String,
151 /// When it started holding the slot for a job: its start, or when its job was queued if that came later, as for
152 /// a warm runner.
153 pub started_at: u64,
154 /// When it was removed; `None` while it runs.
155 pub removed_at: Option<u64>,
156}
157
158/// A profile's `cpus` as the daemon ran it from `at` (ms), by its `profile.changed` event: one when it started with
159/// the profile, and after each reload or restart that changed the profile's image, CPUs, memory or slots.
160#[derive(Debug, Clone, PartialEq, Eq)]
161pub struct CpusSet {
162 /// The profile.
163 pub profile: String,
164 /// Its `cpus` from then.
165 pub cpus: u32,
166 /// When, in milliseconds.
167 pub at: u64,
168}
169
170/// A profile's `memory` as the daemon ran it from `at` (ms), by its `profile.changed` event, as `CpusSet` has its
171/// `cpus`.
172#[derive(Debug, Clone, PartialEq, Eq)]
173pub struct MemorySet {
174 /// The profile.
175 pub profile: String,
176 /// Its `memory` from then, in bytes.
177 pub memory: u64,
178 /// When, in milliseconds.
179 pub at: u64,
180}
181
182/// What the store reads for sizing.
183#[derive(Debug, Clone, Default, PartialEq, Eq)]
184pub struct History {
185 /// Finished jobs with a recorded peak, cancelled ones left out.
186 pub jobs: Vec<Ran>,
187 /// Cancelled jobs, for what admission learns from how they ended: a kill for memory, a peak at the cap or past
188 /// what was reserved. Their peaks (0 when never sampled) are not read as a job's size.
189 pub cancelled: Vec<Ran>,
190 /// Runners that held a slot at some point after the history's start.
191 pub runners: Vec<Held>,
192 /// Auto sizing's changes, oldest first, from `CHANGES_READ` before now: what a revert reads (ADR 0070). Empty
193 /// where nothing reads them, as for admission.
194 pub changes: Vec<api::SizingChange>,
195 /// Each profile's `memory` as the daemon ran it, oldest first: since when a raise reads runs at it.
196 pub memory_set: Vec<MemorySet>,
197}
198
199/// What `advise` sizes against besides the history.
200#[derive(Debug, Clone, Copy)]
201pub struct Window {
202 /// The start asked for, in milliseconds, at most `now`.
203 pub since: u64,
204 /// Now, in milliseconds.
205 pub now: u64,
206 /// Seconds between two samples of a container.
207 pub sample_interval: u64,
208}
209
210impl Window {
211 /// The window `GET /v1/sizing` reads: from `since` (seconds), default 30 days before `now` (seconds), and
212 /// never after `now`.
213 pub fn new(since: Option<u64>, now: u64) -> Window {
214 Window {
215 since: since
216 .unwrap_or_else(|| now.saturating_sub(30 * 24 * 3600))
217 .min(now)
218 .saturating_mul(1000),
219 now: now.saturating_mul(1000),
220 sample_interval: SAMPLE_INTERVAL.as_secs(),
221 }
222 }
223}
224
225/// The advice for every profile of `config` from `history`, against the slice's `memory_max` as `config`'s
226/// admission policy budgets it. Peaks, and the memory advice, go by each job's sizing window over the history's last
227/// 30 days (`memory`, ADR 0070); `window.since` narrows only the waits, the busy share and the starts.
228pub fn advise(
229 config: &Config,
230 memory_max: Option<u64>,
231 history: &History,
232 window: Window,
233) -> api::Sizing {
234 let since = first_runner(history, window.since);
235 let caps = config.memory_caps().map_or(u64::MAX, |caps| caps.0);
236 let mirror = config.mirror_memory();
237 let rules = &config.sizing;
238 let recent = memory::windows(config, history, window.now);
239 let profiles = config
240 .profiles
241 .iter()
242 .map(|(name, profile)| {
243 let jobs: Vec<&Ran> = history
244 .jobs
245 .iter()
246 .filter(|job| job.profile == *name && job.queued_at >= window.since)
247 .collect();
248 let share = profile.memory.0.saturating_mul(u64::from(profile.slots));
249 let busy = busy(history, name, profile.slots, since, window.now);
250 let wait = waits(&jobs);
251 let starts = starts(&jobs);
252 let cap = profile.memory.0;
253 let took = memory::took(&history.memory_set, name, cap, window.now);
254 let sized = Sized {
255 cap,
256 slots: profile.slots,
257 share,
258 admission: config.host.admission,
259 room: memory_max.map(|max| match config.host.admission {
260 // What the slice leaves this profile once the other profiles and the mirror took theirs.
261 Admission::Caps => max.saturating_sub(caps.saturating_sub(share)),
262 // Measured admission reserves predicted peaks, not caps: the config check only holds one
263 // runner of each profile beside the mirror.
264 Admission::Measured => max.saturating_sub(mirror),
265 }),
266 wants_slots: wait.p95 >= WAITED || busy >= BUSY,
267 min_jobs: rules.min_jobs,
268 };
269 let raises = Raises {
270 history,
271 recent: &recent,
272 profile: name,
273 cap,
274 took,
275 margin: rules.margin,
276 };
277 let memory = job_memory(&raises, &jobs, &sized, window.now, busy);
278 let peaks = memory::peaks(
279 recent.jobs_of(name).flat_map(|(_, runs)| runs.window()),
280 memory.on_cap,
281 );
282 let current = api::Resize {
283 memory: cap,
284 slots: profile.slots,
285 cpus: None,
286 };
287 let revert = memory::revert(config, history, name, current);
288 let advice = match &revert {
289 Some((_, to)) => Advice::Raise {
290 memory: to.memory,
291 slots: to.slots,
292 },
293 None => sized.advice(
294 &peaks,
295 Needs {
296 raise: raises.need(None),
297 floor: memory.floor.max(memory::revert_floor(history, name)),
298 },
299 sized.wants_slots,
300 ),
301 };
302 api::ProfileSizing {
303 name: name.clone(),
304 memory: cap,
305 cpus: profile.cpus,
306 slots: profile.slots,
307 advice,
308 revert: revert.map(|(revert, _)| revert),
309 peaks,
310 wait,
311 starts,
312 busy,
313 wants_slots: sized.wants_slots,
314 jobs: memory.jobs,
315 }
316 })
317 .collect();
318 api::Sizing {
319 since: since / 1000,
320 until: window.now / 1000,
321 margin: rules.margin,
322 sample_interval: window.sample_interval,
323 min_jobs: rules.min_jobs,
324 memory_max,
325 admission: config.host.admission.policy(),
326 profiles,
327 auto: config.sizing.auto,
328 changes: Vec::new(),
329 auto_plan: None,
330 }
331}
332
333/// What a raise of one profile reads (`memory::raise_need`).
334struct Raises<'history> {
335 history: &'history History,
336 recent: &'history Recent,
337 profile: &'history str,
338 /// Its `memory`.
339 cap: u64,
340 /// Since when it has had it, in ms.
341 took: u64,
342 margin: u32,
343}
344
345impl Raises<'_> {
346 /// What a raise needs from the job `key`, or from every job for `None`.
347 fn need(&self, key: Option<&Key>) -> u64 {
348 memory::raise_need(
349 self.history,
350 self.recent,
351 self.profile,
352 self.cap,
353 self.took,
354 self.margin,
355 key,
356 )
357 }
358}
359
360/// What a profile's jobs say about its `memory`.
361struct JobMemory {
362 /// Each job's sizing.
363 jobs: Vec<api::JobSizing>,
364 /// How many of them are on their cap.
365 on_cap: u64,
366 /// The largest of their floors.
367 floor: u64,
368}
369
370/// Each job of `raises.profile` with runs in its window: its peaks, waits and advice; `jobs` are the profile's runs
371/// since the window's start, for the waits.
372fn job_memory(raises: &Raises<'_>, jobs: &[&Ran], sized: &Sized, now: u64, busy: u32) -> JobMemory {
373 let mut on_cap = 0;
374 let mut floor = 0;
375 let sizings = raises
376 .recent
377 .jobs_of(raises.profile)
378 .filter(|(_, runs)| runs.runs() > 0)
379 .map(|(key, runs)| {
380 let held = memory::on_cap(runs, now);
381 on_cap += u64::from(held);
382 let peaks = memory::peaks(runs.window(), u64::from(held));
383 let needs = Needs {
384 raise: raises.need(Some(key)),
385 floor: memory::floor(runs, raises.margin),
386 };
387 floor = floor.max(needs.floor);
388 let runs_of: Vec<&Ran> = jobs
389 .iter()
390 .copied()
391 .filter(|job| Key::of(job) == *key)
392 .collect();
393 let wait = waits(&runs_of);
394 api::JobSizing {
395 repository: key.repository.clone(),
396 workflow: key.workflow.clone(),
397 name: key.name.clone(),
398 advice: sized.advice(&peaks, needs, wait.p95 >= WAITED || busy >= BUSY),
399 peaks,
400 wait,
401 }
402 })
403 .collect();
404 JobMemory {
405 jobs: sizings,
406 on_cap,
407 floor,
408 }
409}
410
411/// Where a history read from `start` (ms) begins: at its first runner when that started later, as a new daemon's
412/// history does.
413fn first_runner(history: &History, start: u64) -> u64 {
414 history
415 .runners
416 .iter()
417 .map(|runner| runner.started_at)
418 .min()
419 .map_or(start, |first| first.max(start))
420}
421
422/// The percent of the time from `since` to `now` (ms) that the profile's runners held its `slots`.
423fn busy(history: &History, profile: &str, slots: u32, since: u64, now: u64) -> u32 {
424 let span = now.saturating_sub(since).max(1);
425 let held: u64 = history
426 .runners
427 .iter()
428 .filter(|runner| runner.profile == profile)
429 .map(|runner| {
430 let end = runner.removed_at.unwrap_or(now).min(now);
431 end.saturating_sub(runner.started_at.max(since))
432 })
433 .sum();
434 u32::try_from(
435 (held.saturating_mul(100) / span.saturating_mul(u64::from(slots).max(1))).min(100),
436 )
437 .unwrap_or(100)
438}
439
440/// A profile's limits and the rules advice follows.
441struct Sized {
442 cap: u64,
443 slots: u32,
444 /// What its slots take of the budget now: `memory × slots`.
445 share: u64,
446 admission: Admission,
447 /// The memory the slice holds for it, as `admission` budgets it: all its slots at their caps under `caps`, one
448 /// runner under `measured`; `None` for an unlimited slice.
449 room: Option<u64>,
450 /// Whether its jobs waited for a slot at p95, or its slots were busy, enough to want more.
451 wants_slots: bool,
452 min_jobs: u32,
453}
454
455/// What a set of jobs' runs say its `memory` needs (`memory`, ADR 0070).
456#[derive(Debug, Clone, Copy)]
457struct Needs {
458 /// What a raise goes to: from runs that met the current cap since the profile took it; 0 without one.
459 raise: u64,
460 /// What a lowering stays at or above: every windowed run's need, and any revert's floor.
461 floor: u64,
462}
463
464impl Sized {
465 /// A raise toward `needs.raise` when it is above the cap; a lowering toward `needs.floor` when that saves a tenth
466 /// and no job is on its cap; else keep, with `needs.floor` as what the jobs need, above the cap or not.
467 fn advice(&self, peaks: &Peaks, needs: Needs, wants_slots: bool) -> Advice {
468 if peaks.jobs < u64::from(self.min_jobs) {
469 return Advice::TooFew { jobs: peaks.jobs };
470 }
471 if needs.raise > self.cap {
472 let need = needs.raise;
473 let fit = |memory: u64| u32::try_from(memory / need).unwrap_or(u32::MAX);
474 let slots = match (self.admission, self.room) {
475 (Admission::Caps, Some(room)) => fit(room).min(self.slots),
476 (Admission::Measured, Some(room)) if room < need => 0,
477 (_, _) => self.slots,
478 };
479 if slots == 0 {
480 return Advice::SliceTooSmall { memory: need };
481 }
482 return Advice::Raise {
483 memory: need,
484 slots,
485 };
486 }
487 let need = needs.floor.max(GRANULE);
488 // A job on its cap may have been held back or killed by it: no lowering until it ran clear of it again.
489 if peaks.on_cap == 0
490 && need <= self.cap
491 && self.cap - need >= self.cap * WORTH_LOWERING / 100
492 {
493 let fit = |memory: u64| u32::try_from(memory / need).unwrap_or(u32::MAX);
494 // Under measured admission slots are not bought with caps: the advice line says when more may help.
495 return Advice::Lower {
496 memory: need,
497 slots: if wants_slots && self.admission == Admission::Caps {
498 fit(self.share).max(self.slots)
499 } else {
500 self.slots
501 },
502 };
503 }
504 Advice::Keep { memory: need }
505 }
506}
507
508/// Seconds from queued to started of the jobs that started, less what they waited for the host's memory.
509fn waits(jobs: &[&Ran]) -> Spread {
510 spread(
511 jobs.iter()
512 .filter_map(|job| {
513 job.started_at.map(|at| {
514 at.saturating_sub(job.queued_at)
515 .saturating_sub(job.host_waited)
516 / 1000
517 })
518 })
519 .collect(),
520 )
521}
522
523/// The jobs that started on a warm runner and the others, and their median waits.
524fn starts(jobs: &[&Ran]) -> api::Starts {
525 let (warm, cold): (Vec<&Ran>, Vec<&Ran>) = jobs
526 .iter()
527 .filter(|job| job.started_at.is_some())
528 .partition(|job| job.runner_started_at < job.queued_at);
529 let count = |jobs: &[&Ran]| u32::try_from(jobs.len()).unwrap_or(u32::MAX);
530 api::Starts {
531 warm: count(&warm),
532 warm_p50: waits(&warm).p50,
533 cold: count(&cold),
534 cold_p50: waits(&cold).p50,
535 }
536}
537
538/// Nearest-rank percentiles; all zero for no values.
539pub(crate) fn spread(mut values: Vec<u64>) -> Spread {
540 values.sort_unstable();
541 let rank = |percent: usize| {
542 let index = (values.len() * percent).div_ceil(100).saturating_sub(1);
543 values.get(index).copied().unwrap_or(0)
544 };
545 Spread {
546 p50: rank(50),
547 p95: rank(95),
548 p99: rank(99),
549 max: values.last().copied().unwrap_or(0),
550 }
551}
552
553pub mod auto;
554pub mod memory;
555
556#[cfg(test)]
557mod tests;