runnerd_core/scheduler/
warm.rs

1//! How many warm runners each repository's pool wants (ADR 0050, 0058): its `warm` while the repository is active, by
2//! its jobs of the last hour and the hours of the day its jobs usually come, plus the jobs for the pool its workflow
3//! runs seen so far are expected to queue still, from the history. Pure: the engine reads the history and the clock
4//! and passes them in.
5
6use std::collections::{BTreeMap, BTreeSet};
7use std::time::Duration;
8
9use super::{IDLE_TIMEOUT, Pool};
10
11/// How long after a repository's last job it stays active.
12pub const ACTIVE_AFTER: Duration = Duration::from_secs(3600);
13
14/// How far back the history is read, in milliseconds: 14 days.
15pub const LOOKBACK: u64 = 14 * DAY;
16
17/// An hour of the day is a repository's usual one when its jobs were queued in it on this many days of `LOOKBACK`.
18const USUAL_DAYS: usize = 3;
19
20/// A workflow's runs before its jobs per run are expected of the next.
21const MIN_RUNS: usize = 3;
22
23/// How long a run is remembered after its first job, whether or not it still counts.
24const FORGET_AFTER: Duration = Duration::from_hours(24);
25
26/// How long a run whose workflow is not known yet counts.
27const UNKNOWN_SPAN: Duration = Duration::from_secs(3600);
28
29const HOUR: u64 = 3600 * 1000;
30
31/// A run's jobs for one pool as the history tells them apart: repository, pool's profile, workflow, run ID and
32/// attempt.
33type RunKey<'seen> = (&'seen str, &'seen str, &'seen str, u64, u32);
34
35/// A workflow's runs for one pool: repository, pool's profile and workflow.
36type WorkflowKey<'seen> = (&'seen str, &'seen str, &'seen str);
37const DAY: u64 = 24 * HOUR;
38
39/// A job the history saw queued.
40#[derive(Debug, Clone, PartialEq, Eq)]
41pub struct Seen {
42    /// `owner/name`.
43    pub repository: String,
44    /// The labels it asked for, which tell the pool it was for.
45    pub labels: Vec<String>,
46    /// The workflow's name, when known.
47    pub workflow: Option<String>,
48    /// The workflow run it belongs to.
49    pub run_id: u64,
50    /// 1 for the run's first attempt.
51    pub run_attempt: u32,
52    /// When GitHub queued it, in milliseconds since the epoch.
53    pub queued_at: u64,
54}
55
56/// What a workflow's runs are expected to queue for one pool.
57#[derive(Debug, Clone, Copy, PartialEq, Eq)]
58pub struct Burst {
59    /// The median number of jobs per run.
60    pub jobs: usize,
61    /// The 90th percentile of the time from a run's first job queued to its last.
62    pub span: Duration,
63}
64
65/// What warm runners learn from the history: when each repository's jobs come, and how many jobs its workflows'
66/// runs queue.
67#[derive(Debug, Clone, Default, PartialEq, Eq)]
68pub struct Learned {
69    /// The hours of the day (UTC) each repository's jobs usually come in.
70    usual: BTreeMap<String, BTreeSet<u64>>,
71    /// When each repository's last job was queued, in milliseconds.
72    last: BTreeMap<String, u64>,
73    /// By pool and workflow.
74    bursts: BTreeMap<(Pool, String), Burst>,
75}
76
77impl Learned {
78    /// Learns from the jobs queued in the `LOOKBACK` before `now` (ms). `profile` names the pool of its repository
79    /// each job was for, by its labels; `None` for one no pool serves now, which counts for no burst.
80    pub fn new(seen: &[Seen], now: u64, profile: impl Fn(&Seen) -> Option<String>) -> Learned {
81        let since = now.saturating_sub(LOOKBACK);
82        let mut learned = Learned::default();
83        let mut days: BTreeMap<(&str, u64), BTreeSet<u64>> = BTreeMap::new();
84        let profiles: Vec<Option<String>> = seen.iter().map(profile).collect();
85        // Each run's jobs per pool, and when its first and last were queued.
86        let mut runs: BTreeMap<RunKey<'_>, (usize, u64, u64)> = BTreeMap::new();
87        // When each run attempt's first job was queued, in any pool: a pool's span runs from there, as a run seen
88        // live counts from its first job (`Run::first`), so a pool whose jobs wait for another's still counts.
89        let mut attempts: BTreeMap<(&str, u64, u32), u64> = BTreeMap::new();
90        for (job, profile) in seen
91            .iter()
92            .zip(&profiles)
93            .filter(|(job, _)| job.queued_at >= since)
94        {
95            learned.note(&job.repository, job.queued_at);
96            days.entry((&job.repository, job.queued_at % DAY / HOUR))
97                .or_default()
98                .insert(job.queued_at / DAY);
99            if let (Some(workflow), Some(profile)) = (&job.workflow, profile) {
100                let run = runs
101                    .entry((
102                        &job.repository,
103                        profile,
104                        workflow,
105                        job.run_id,
106                        job.run_attempt,
107                    ))
108                    .or_insert((0, job.queued_at, job.queued_at));
109                run.0 += 1;
110                run.1 = run.1.min(job.queued_at);
111                run.2 = run.2.max(job.queued_at);
112                let first = attempts
113                    .entry((&job.repository, job.run_id, job.run_attempt))
114                    .or_insert(job.queued_at);
115                *first = (*first).min(job.queued_at);
116            }
117        }
118        for ((repository, hour), days) in days {
119            if days.len() >= USUAL_DAYS {
120                learned
121                    .usual
122                    .entry(repository.to_owned())
123                    .or_default()
124                    .insert(hour);
125            }
126        }
127        // By repository, pool's profile and workflow: each run's jobs and span.
128        let mut workflows: BTreeMap<WorkflowKey<'_>, Vec<(usize, u64)>> = BTreeMap::new();
129        for ((repository, profile, workflow, run_id, run_attempt), (jobs, first, last)) in runs {
130            let first = attempts
131                .get(&(repository, run_id, run_attempt))
132                .copied()
133                .unwrap_or(first);
134            workflows
135                .entry((repository, profile, workflow))
136                .or_default()
137                .push((jobs, last - first));
138        }
139        for ((repository, profile, workflow), runs) in workflows {
140            if runs.len() < MIN_RUNS {
141                continue;
142            }
143            let mut jobs: Vec<usize> = runs.iter().map(|(jobs, _)| *jobs).collect();
144            let mut spans: Vec<u64> = runs.iter().map(|(_, span)| *span).collect();
145            jobs.sort_unstable();
146            spans.sort_unstable();
147            let pool = Pool {
148                repository: repository.to_owned(),
149                profile: profile.to_owned(),
150            };
151            learned.bursts.insert(
152                (pool, workflow.to_owned()),
153                Burst {
154                    jobs: jobs[(jobs.len() - 1) / 2],
155                    span: Duration::from_millis(spans[(spans.len() * 90).div_ceil(100) - 1]),
156                },
157            );
158        }
159        learned
160    }
161
162    /// Keeps the last jobs `older` noted that this one has not read yet.
163    pub fn absorb(&mut self, older: &Learned) {
164        for (repository, at) in &older.last {
165            self.note(repository, *at);
166        }
167    }
168
169    /// Notes a job of `repository` queued or taken at `at` (ms).
170    pub fn note(&mut self, repository: &str, at: u64) {
171        let last = self.last.entry(repository.to_owned()).or_default();
172        *last = (*last).max(at);
173    }
174
175    /// Whether `repository` is active at `now` (ms): its last job came within `ACTIVE_AFTER`, or `now` is in an hour
176    /// of the day its jobs usually come in.
177    pub fn active(&self, repository: &str, now: u64) -> bool {
178        let after = u64::try_from(ACTIVE_AFTER.as_millis()).unwrap_or(u64::MAX);
179        self.last
180            .get(repository)
181            .is_some_and(|last| now.saturating_sub(*last) < after)
182            || self
183                .usual
184                .get(repository)
185                .is_some_and(|hours| hours.contains(&(now % DAY / HOUR)))
186    }
187
188    /// What runs of `workflow` are expected to queue for `pool`, once it has `MIN_RUNS` runs with jobs for it.
189    pub fn burst(&self, pool: &Pool, workflow: &str) -> Option<Burst> {
190        self.bursts
191            .get(&(pool.clone(), workflow.to_owned()))
192            .copied()
193    }
194}
195
196/// A workflow run seen lately, with the jobs of it seen so far.
197#[derive(Debug, Clone, PartialEq, Eq)]
198struct Run {
199    repository: String,
200    workflow: Option<String>,
201    /// Each job's pool, by its profile.
202    jobs: BTreeMap<u64, String>,
203    /// When its first job was queued or seen, in milliseconds.
204    first: u64,
205}
206
207/// An attempt of a workflow run: a re-run keeps the run's ID, with the next attempt, and queues its jobs anew.
208#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
209pub struct Attempt {
210    /// The workflow run's ID.
211    pub run_id: u64,
212    /// 1 for the run's first attempt.
213    pub run_attempt: u32,
214}
215
216/// The workflow runs seen lately, by attempt: their jobs not queued yet get warm runners.
217#[derive(Debug, Clone, Default, PartialEq, Eq)]
218pub struct Runs(BTreeMap<Attempt, Run>);
219
220impl Runs {
221    /// Notes `job` of `attempt` for `pool`, queued or taken at `at` (ms), and its workflow when known.
222    pub fn note(
223        &mut self,
224        pool: &Pool,
225        attempt: Attempt,
226        job: u64,
227        workflow: Option<&str>,
228        at: u64,
229    ) {
230        let run = self.0.entry(attempt).or_insert_with(|| Run {
231            repository: pool.repository.clone(),
232            workflow: None,
233            jobs: BTreeMap::new(),
234            first: at,
235        });
236        run.jobs.insert(job, pool.profile.clone());
237        run.first = run.first.min(at);
238        if run.workflow.is_none() {
239            run.workflow = workflow.map(str::to_owned);
240        }
241    }
242
243    /// Moves `job` to `attempt` once GitHub said which attempt it is of, when it was noted under another of the
244    /// run's: a job a runner took is first noted under its run's latest attempt known, which for a re-run's first
245    /// job is the attempt before. An attempt it alone made is forgotten, and its time goes with the job; a job that
246    /// joined an older attempt counts from `at` (ms). A job not followed stays so, so a run forgotten while its job
247    /// ran is not counted again.
248    pub fn place(
249        &mut self,
250        pool: &Pool,
251        attempt: Attempt,
252        job: u64,
253        workflow: Option<&str>,
254        at: u64,
255    ) {
256        // When the job was seen: an attempt it alone made began with it, while one it joined began before it.
257        let mut seen: Option<u64> = None;
258        let mut emptied = Vec::new();
259        for (other, run) in self.attempts_mut(attempt.run_id) {
260            if *other != attempt && run.jobs.remove(&job).is_some() {
261                if run.jobs.is_empty() {
262                    emptied.push(*other);
263                    seen = Some(run.first.min(seen.unwrap_or(at)));
264                } else {
265                    seen = Some(seen.unwrap_or(at));
266                }
267            }
268        }
269        for other in emptied {
270            self.0.remove(&other);
271        }
272        if let Some(seen) = seen {
273            self.note(pool, attempt, job, workflow, seen);
274        }
275    }
276
277    /// Names the workflow of run `run_id`'s attempts followed whose workflow was not known.
278    pub fn name(&mut self, run_id: u64, workflow: &str) {
279        for (_, run) in self.attempts_mut(run_id) {
280            if run.workflow.is_none() {
281                run.workflow = Some(workflow.to_owned());
282            }
283        }
284    }
285
286    /// The latest attempt of run `run_id` followed, or its first.
287    pub fn latest(&self, run_id: u64) -> Attempt {
288        self.0
289            .range(
290                Attempt {
291                    run_id,
292                    run_attempt: 0,
293                }..=Attempt {
294                    run_id,
295                    run_attempt: u32::MAX,
296                },
297            )
298            .next_back()
299            .map_or(
300                Attempt {
301                    run_id,
302                    run_attempt: 1,
303                },
304                |(attempt, _)| *attempt,
305            )
306    }
307
308    fn attempts_mut(&mut self, run_id: u64) -> impl Iterator<Item = (&Attempt, &mut Run)> {
309        self.0.range_mut(
310            Attempt {
311                run_id,
312                run_attempt: 0,
313            }..=Attempt {
314                run_id,
315                run_attempt: u32::MAX,
316            },
317        )
318    }
319
320    /// Forgets the runs whose first job came `FORGET_AFTER` before `now` (ms). One that stopped counting is kept
321    /// until then, so a job of it seen again, as after a restart, does not count it again as a new run.
322    pub fn expire(&mut self, now: u64) {
323        let after = u64::try_from(FORGET_AFTER.as_millis()).unwrap_or(u64::MAX);
324        self.0
325            .retain(|_, run| now.saturating_sub(run.first) < after);
326    }
327
328    /// The jobs for `pool` its repository's runs are expected to queue still at `now` (ms): per run that still
329    /// counts for it, its workflow's jobs per run for the pool less the jobs of it seen for the pool.
330    pub fn remaining(&self, pool: &Pool, learned: &Learned, now: u64) -> usize {
331        self.0
332            .values()
333            .filter(|run| run.repository == pool.repository && run.counts(pool, learned, now))
334            .filter_map(|run| {
335                let burst = learned.burst(pool, run.workflow.as_deref()?)?;
336                let seen = run
337                    .jobs
338                    .values()
339                    .filter(|profile| **profile == pool.profile)
340                    .count();
341                Some(burst.jobs.saturating_sub(seen))
342            })
343            .sum()
344    }
345}
346
347impl Run {
348    /// Whether it still counts for `pool` at `now` (ms): until its workflow's span for the pool plus
349    /// `IDLE_TIMEOUT` after its first job.
350    fn counts(&self, pool: &Pool, learned: &Learned, now: u64) -> bool {
351        let span = self
352            .workflow
353            .as_deref()
354            .and_then(|workflow| learned.burst(pool, workflow))
355            .map_or(UNKNOWN_SPAN, |burst| burst.span + IDLE_TIMEOUT);
356        now.saturating_sub(self.first) < u64::try_from(span.as_millis()).unwrap_or(u64::MAX)
357    }
358}
359
360/// How many warm runners a pool with `warm` configured wants: `warm` while it is `active`, plus the jobs its
361/// runs are expected to queue still (`remaining`). None with `warm = 0`.
362pub fn target(warm: u32, active: bool, remaining: usize) -> usize {
363    if warm == 0 {
364        return 0;
365    }
366    let base = if active {
367        usize::try_from(warm).unwrap_or(usize::MAX)
368    } else {
369        0
370    };
371    base.saturating_add(remaining)
372}
373
374#[cfg(test)]
375mod tests;
runnerd_core/scheduler/warm.rs

runnerd_core/scheduler/
warm.rs
