runnerd_core/sizing/
memory.rs

1//! What a profile's runs say about its `memory` (ADR 0070): each job's sizing window (`admission::recent`), what a
2//! raise needs (`raise_need`: only runs that met the current cap since the profile took it), what a lowering must
3//! stay above (`lower_floor`: every windowed run's need), and a lowering auto sizing made that went too far, which is
4//! reverted. Pure; `advise` and auto sizing read it, never a client.
5
6use std::collections::BTreeSet;
7
8use crate::admission::Key;
9use crate::admission::recent::{self, Counted, Ending, JobRuns, RECENT_RUNS, Recent, SIZING_TIMES};
10use crate::api::{self, Peaks, Resize, RevertCause, SizingRevert, job_title};
11use crate::config::{Admission, Config};
12
13use super::{Capped, History, MemorySet, Ran, spread};
14
15/// How far back the windows read, in ms: 30 days, whatever `since` a client asked for.
16pub const HISTORY: u64 = 30 * 24 * 3600 * 1000;
17
18/// How far back auto sizing's recorded changes are read for a revert, in ms: 60 days, so a lowering up to 30 days
19/// before the run that met its cap is found while that run is in the history.
20pub const CHANGES_READ: u64 = 2 * HISTORY;
21
22/// A lowering is reverted after this many failures in a row at it, of a job that succeeded at the cap it came from.
23pub const FAILED_RUNS: usize = 3;
24
25/// Every job's sizing window from `history`'s last 30 days to `now` (ms).
26pub(super) fn windows(config: &Config, history: &History, now: u64) -> Recent {
27    let size = SIZING_TIMES.saturating_mul(RECENT_RUNS.max(u64::from(config.sizing.min_jobs)));
28    let mut recent = Recent::new(size);
29    let start = now.saturating_sub(HISTORY);
30    for (run, ending) in recent::rows(history) {
31        if run.queued_at < start {
32            continue;
33        }
34        recent.read(
35            run,
36            &run.profile,
37            ending,
38            cap_of(config, run),
39            recent::finished(run),
40        );
41    }
42    recent
43}
44
45/// The cap a run's container had, else its profile's now.
46fn cap_of(config: &Config, run: &Ran) -> u64 {
47    run.memory_cap
48        .or_else(|| config.profiles.get(&run.profile).map(|p| p.memory.0))
49        .unwrap_or(u64::MAX)
50}
51
52/// The peaks of the runs in some windows; `on_cap` how many of their jobs are on a hold at the cap.
53pub(super) fn peaks<'run>(runs: impl Iterator<Item = &'run Counted>, on_cap: u64) -> Peaks {
54    let runs: Vec<&Counted> = runs.collect();
55    Peaks {
56        jobs: runs.len() as u64,
57        at_cap: runs.iter().filter(|run| run.capped.met()).count() as u64,
58        on_cap,
59        memory: spread(runs.iter().map(|run| run.peak).collect()),
60        cpu: spread(runs.iter().map(|run| run.cpu_peak).collect()),
61    }
62}
63
64/// What a windowed run says its job needs at least: above its cap when it met it, else its peak's need, at most
65/// its cap, since it ran clear of it.
66fn run_need(run: &Counted, margin: u32) -> u64 {
67    if run.capped.met() {
68        api::need(run.peak.max(run.cap), margin)
69    } else {
70        api::need(run.peak, margin).min(run.cap)
71    }
72}
73
74/// The largest need of a job's windowed runs.
75pub(super) fn floor(runs: &JobRuns, margin: u32) -> u64 {
76    runs.window()
77        .map(|run| run_need(run, margin))
78        .max()
79        .unwrap_or(0)
80}
81
82/// What a job's windowed runs at caps above `above` say it needs, each at most its own cap: a raise goes back at
83/// most to a cap the job ran at, and a need above that comes from a run that meets it. A run recorded without the
84/// local counter (`Capped::Assumed`) says nothing here: it never raises a cap.
85fn larger_caps(runs: &JobRuns, margin: u32, above: u64) -> u64 {
86    runs.window()
87        .filter(|run| run.cap > above && run.capped != Capped::Assumed)
88        .map(|run| run_need(run, margin).min(run.cap))
89        .max()
90        .unwrap_or(0)
91}
92
93/// Whether `held` is on its cap for sizing: a hold `recent::CLEAN_RUNS` clear runs have not ended, however old.
94pub(super) fn on_cap(runs: &JobRuns, now: u64) -> bool {
95    runs.held(now, None).is_some()
96}
97
98/// When the daemon started running `profile` with `memory`, in ms, by `set`: its first event since the last one
99/// with other `memory`. `now` when its latest has other `memory`; 0 with no events.
100pub(super) fn took(set: &[MemorySet], profile: &str, memory: u64, now: u64) -> u64 {
101    let mut since = None;
102    let mut seen = false;
103    for event in set.iter().filter(|event| event.profile == profile) {
104        seen = true;
105        if event.memory != memory {
106            since = None;
107        } else if since.is_none() {
108            since = Some(event.at);
109        }
110    }
111    since.unwrap_or(if seen { now } else { 0 })
112}
113
114/// What a raise of `profile`, whose `memory` is `cap` since `took` (ms), needs from the job `key`, `None` for every
115/// job: the largest of `need(max(peak, cap))` over its runs that met `cap` since then, and of that job's need at
116/// larger caps (`larger_caps`), since its last sample before a kill rarely shows what it needed. 0 without such a
117/// run.
118pub(super) fn raise_need(
119    history: &History,
120    recent: &Recent,
121    profile: &str,
122    cap: u64,
123    took: u64,
124    margin: u32,
125    key: Option<&Key>,
126) -> u64 {
127    history
128        .jobs
129        .iter()
130        .chain(&history.cancelled)
131        .filter(|run| {
132            run.profile == profile
133                && run.memory_cap == Some(cap)
134                && recent::finished(run) >= took
135                && run.capped(cap) == Capped::Met
136                && key.is_none_or(|key| Key::of(run) == *key)
137        })
138        .map(|run| {
139            let larger = recent
140                .job(&Key::of(run), profile)
141                .map_or(0, |runs| larger_caps(runs, margin, cap));
142            api::need(run.memory_peak.max(cap), margin).max(larger)
143        })
144        .max()
145        .unwrap_or(0)
146}
147
148/// The lowerings auto sizing made of `profile` that went too far, oldest first: since each, a run at the cap it
149/// lowered to met it, or a job that succeeded at the cap it came from failed `FAILED_RUNS` times in a row at it.
150fn too_far<'change>(
151    history: &'change History,
152    profile: &str,
153) -> impl Iterator<Item = (&'change api::SizingChange, SizingRevert)> {
154    history
155        .changes
156        .iter()
157        .filter(move |change| change.profile == profile && change.to.memory < change.from.memory)
158        .filter_map(move |change| {
159            cause(history, profile, change).map(|(job, cause)| {
160                (
161                    change,
162                    SizingRevert {
163                        from: change.to.memory,
164                        to: change.from.memory,
165                        job,
166                        cause,
167                    },
168                )
169            })
170        })
171}
172
173/// What shows that `lowering` of `profile` went too far, if anything: the job and how.
174fn cause(
175    history: &History,
176    profile: &str,
177    lowering: &api::SizingChange,
178) -> Option<(String, RevertCause)> {
179    let (lowered, from) = (lowering.to.memory, lowering.from.memory);
180    let since = lowering.at.saturating_mul(1000);
181    let after: Vec<(&Ran, Ending)> = recent::rows(history)
182        .into_iter()
183        .filter(|(run, _)| {
184            run.profile == profile
185                && run.memory_cap == Some(lowered)
186                && recent::finished(run) >= since
187        })
188        .collect();
189    let named = |run: &Ran| {
190        format!(
191            "{} {}",
192            run.repository,
193            job_title(run.workflow.as_deref(), &run.name, None)
194        )
195    };
196    if let Some((run, _)) = after
197        .iter()
198        .find(|(run, _)| run.capped(lowered) == Capped::Met)
199    {
200        return Some((named(run), RevertCause::Met));
201    }
202    let jobs: BTreeSet<Key> = after.iter().map(|(run, _)| Key::of(run)).collect();
203    for key in jobs {
204        // Before the revert only runs from before the lowering ran at `from`; after it, its runs at `from` since keep
205        // the floor while the failures are in the history.
206        let succeeded_at_from = history.jobs.iter().any(|run| {
207            run.profile == profile
208                && Key::of(run) == key
209                && run.succeeded
210                && run.memory_cap == Some(from)
211        });
212        if !succeeded_at_from {
213            continue;
214        }
215        let mut row: Vec<(u64, Option<&str>)> = Vec::new();
216        for (run, ending) in after.iter().filter(|(run, _)| Key::of(run) == key) {
217            if *ending == Ending::Cancelled {
218                continue;
219            }
220            if run.succeeded {
221                row.clear();
222            } else if run.failed && !row.iter().any(|(id, _)| *id == run.run_id) {
223                row.push((run.run_id, run.branch.as_deref()));
224            }
225            let branches: BTreeSet<Option<&str>> = row.iter().map(|(_, branch)| *branch).collect();
226            if row.len() >= FAILED_RUNS && branches.len() >= 2 {
227                return Some((
228                    named(run),
229                    RevertCause::Failed {
230                        runs: u32::try_from(row.len()).unwrap_or(u32::MAX),
231                    },
232                ));
233            }
234        }
235    }
236    None
237}
238
239/// The revert auto sizing makes now of `profile`, whose `memory` and `slots` are `current`, if any: its latest
240/// recorded change of `memory` lowered it to what it has, the daemon has run that since (`set`), and that went too
241/// far. Changes of only `slots` or `cpus` since do not hide it. Under caps the slots go back to what they were
242/// before the lowering, so the memory fits the slice.
243pub(super) fn revert(
244    config: &Config,
245    history: &History,
246    profile: &str,
247    current: Resize,
248) -> Option<(SizingRevert, Resize)> {
249    let last = history
250        .changes
251        .iter()
252        .rfind(|change| change.profile == profile && change.to.memory != change.from.memory)?;
253    if last.to.memory != current.memory || last.to.memory >= last.from.memory {
254        return None;
255    }
256    let since = last.at.saturating_mul(1000);
257    let mut events = history
258        .memory_set
259        .iter()
260        .filter(|event| event.profile == profile && event.at >= since)
261        .peekable();
262    events.peek()?;
263    if events.any(|event| event.memory != current.memory) {
264        return None;
265    }
266    let (_, revert) = too_far(history, profile).find(|(change, _)| *change == last)?;
267    let slots = match config.host.admission {
268        Admission::Caps => last.from.slots,
269        Admission::Measured => current.slots,
270    };
271    Some((
272        revert,
273        Resize {
274            memory: last.from.memory,
275            slots,
276            cpus: None,
277        },
278    ))
279}
280
281/// The floor reverts set on `profile`'s lowerings: the largest cap a lowering that went too far came from, while
282/// the run that showed it is in the history.
283pub(super) fn revert_floor(history: &History, profile: &str) -> u64 {
284    too_far(history, profile)
285        .map(|(_, revert)| revert.to)
286        .max()
287        .unwrap_or(0)
288}
runnerd_core/sizing/memory.rs

runnerd_core/sizing/
memory.rs
