runnerd_core/scheduler/
engine.rs

1//! The scheduling loop: polls GitHub, pulls the profiles' images, applies `plan`'s decisions with GitHub and
2//! Docker, removes each runner's container when its job ends, keeps the mirror up while any runner runs, reseeds
3//! idle profiles and recovers runners and containers a previous daemon left. It also feeds the job history: queued
4//! jobs, the job each runner took (from `observe::capture`), steps from GitHub, mirror bytes, cache sizes and slot
5//! and budget changes. It owns slot state; it never serves the API (it publishes `api::Status` and takes commands
6//! through `Control`) and never logs the token or a JIT config.
7//!
8//! This module holds `Engine`'s state, its construction and its loop; the submodules each add one concern's
9//! methods: `recover` (startup), `tick` (polling and planning), `runners` (a slot's runner from start to removal),
10//! `images` (pulling the profiles' images), `caches` (reseeds, the mirror and the shared caches), `history` (what
11//! the store records), `warm` (how many warm runners each repository wants), `control` (the API's commands) and
12//! `status` (what it publishes).
13
14mod admission;
15mod caches;
16mod control;
17mod history;
18mod images;
19mod records;
20mod recover;
21mod runners;
22mod status;
23mod tick;
24mod warm;
25
26use std::collections::{BTreeMap, BTreeSet};
27use std::sync::Arc;
28use std::time::{Duration, SystemTime};
29
30use tokio::sync::{mpsc, watch};
31use tokio::task::JoinHandle;
32use tracing::{info, warn};
33
34pub use self::control::{
35    Command, Commands, Commit, Control, ControlError, RunningJobs, RunningSlots, Shutdown, Target,
36};
37use self::history::{Closing, Found};
38use self::images::Images;
39pub use self::records::RecordSelection;
40use super::SlotRef;
41use crate::admission::Key;
42use crate::api;
43use crate::cache::{self, Names, Trimmed};
44use crate::config::Config;
45use crate::docker::{self, Docker, DockerApi};
46use crate::github::{Github, GithubApi, QueuedJob};
47use crate::isolation::{self, Isolation};
48use crate::observe::metrics::{self, Paths, Sampler, Tracked};
49use crate::observe::mirror::Leases;
50use crate::store::{Announced, MirrorPosition, Store};
51
52/// This machine as its runner names carry it: `runnerd-<id>-<host>-<profile>-<slot>-<secs>`. Only `runnerd-<id>-`
53/// identifies the machine, so a changed host name still matches the runners started before the change, and two
54/// machines with the same or overlapping host names never take each other's runners for their own.
55#[derive(Debug, Clone, PartialEq, Eq)]
56pub struct HostIdentity {
57    /// 8 hex digits: a hash of the systemd machine ID, never the ID itself, as machine-id(5) asks.
58    id: String,
59    /// The host name in lower case, letters, digits and dashes, at most 20 characters: for people reading names.
60    host: String,
61}
62
63impl HostIdentity {
64    /// This machine's identity, from `/etc/machine-id` and its host name.
65    pub fn read() -> std::io::Result<HostIdentity> {
66        let hostname = std::fs::read_to_string("/proc/sys/kernel/hostname").unwrap_or_default();
67        let machine_id = std::fs::read_to_string("/etc/machine-id")
68            .or_else(|_| std::fs::read_to_string("/var/lib/dbus/machine-id"))
69            .map_err(|error| {
70                std::io::Error::new(
71                    error.kind(),
72                    format!("cannot read /etc/machine-id: {error}"),
73                )
74            })?;
75        HostIdentity::new(&hostname, &machine_id).ok_or_else(|| {
76            std::io::Error::new(
77                std::io::ErrorKind::InvalidData,
78                "/etc/machine-id does not hold a machine ID",
79            )
80        })
81    }
82
83    /// The identity for a host name and a machine ID; `None` unless the ID is 32 hex digits.
84    pub fn new(hostname: &str, machine_id: &str) -> Option<HostIdentity> {
85        let machine_id = machine_id.trim();
86        if machine_id.len() != 32 || !machine_id.chars().all(|c| c.is_ascii_hexdigit()) {
87            return None;
88        }
89        // FNV-1a: stable across Rust releases, unlike std's hasher, so a new build recognizes the runners of the old.
90        let hash = format!("runnerd:{}", machine_id.to_ascii_lowercase())
91            .bytes()
92            .fold(0xcbf2_9ce4_8422_2325_u64, |hash, byte| {
93                (hash ^ u64::from(byte)).wrapping_mul(0x0000_0100_0000_01b3)
94            });
95        let host: String = hostname
96            .trim()
97            .chars()
98            .map(|c| {
99                if c.is_ascii_alphanumeric() {
100                    c.to_ascii_lowercase()
101                } else {
102                    '-'
103                }
104            })
105            .take(20)
106            .collect();
107        let host = host.trim_matches('-');
108        Some(HostIdentity {
109            id: format!("{:08x}", hash >> 32),
110            host: if host.is_empty() { "host" } else { host }.to_owned(),
111        })
112    }
113
114    /// What every runner name of this machine starts with. The ID's fixed length keeps it from starting another
115    /// machine's names.
116    pub fn prefix(&self) -> String {
117        format!("runnerd-{}-", self.id)
118    }
119
120    /// The name of a runner, and of its container, started in `slot` (1-based) of `profile` at `secs`.
121    pub fn runner_name(&self, profile: &str, slot: usize, secs: u64) -> String {
122        format!("{}{}-{profile}-{slot}-{secs}", self.prefix(), self.host)
123    }
124}
125
126/// What the API needs from a running engine.
127#[derive(Debug, Clone)]
128pub struct Handle {
129    /// The latest status, replaced after every tick.
130    pub status: watch::Receiver<api::Status>,
131    /// Pause, resume and shutdown requests.
132    pub control: Arc<Control>,
133    /// The job history.
134    pub store: Store,
135    /// Every cache volume of the config, with size 0: the API fills in the sizes it measured. Replaced when the
136    /// config is.
137    pub caches: watch::Receiver<Vec<api::CacheVolume>>,
138    /// The records of failed jobs, replaced whenever one is made or removed.
139    pub records: watch::Receiver<api::Records>,
140}
141
142/// Where the engine runs, besides its config and its GitHub and Docker clients.
143#[derive(Debug)]
144pub struct Setup {
145    /// The names of what runnerd creates in Docker.
146    pub names: Names,
147    /// This machine, which names its runners, so a restarted daemon recognizes them.
148    pub identity: HostIdentity,
149    /// The slice's `MemoryMax`.
150    pub memory_max: Option<u64>,
151    /// The CPUs the slice may use, in millicores (`config::slice_cpus`); `None` admits no runner by CPU.
152    pub cpus: Option<u64>,
153    /// The job history.
154    pub store: Store,
155    /// The profiles and repositories as the last run announced them (`Store::announced`), so the first publish
156    /// announces only the difference, including what the config no longer has.
157    pub announced: Announced,
158    /// The repositories recorded paused (`Store::paused_repositories`): they stay paused from the first tick.
159    pub paused: Vec<String>,
160    /// Where the cgroup and proc file systems are.
161    pub paths: Paths,
162}
163
164/// Why a runner starts.
165#[derive(Debug, Clone, Copy, PartialEq, Eq)]
166enum Purpose {
167    /// For a queued job.
168    Job,
169    /// As a warm runner, for jobs not queued yet.
170    Warm,
171}
172
173/// A slot's runner and its container.
174#[derive(Debug, Clone)]
175struct Occupant {
176    repository: String,
177    runner_name: String,
178    runner_id: u64,
179    container: String,
180    busy: bool,
181    started_at: SystemTime,
182    /// Its container's memory cap, which a reload does not change.
183    memory: u64,
184    /// Its container's CPU limit in millicores, which a reload does not change.
185    cpus: u64,
186    /// How it was admitted, with what it reserved of the budget; `None` for one a restart adopted, which reserves
187    /// its cap.
188    admission: Option<api::Admission>,
189    /// The job it took, once its log capture found it.
190    job: Option<u64>,
191    /// That job as admission tells jobs apart, once the history had its record.
192    key: Option<Key>,
193}
194
195impl Occupant {
196    /// What it reserves of the budget under measured admission: its cap if a restart adopted it.
197    fn reserved(&self) -> u64 {
198        self.admission
199            .map_or(self.memory, |admission| admission.reserved)
200    }
201
202    /// The CPU it reserves, in millicores: its CPU limit unless it was admitted by CPU.
203    fn reserved_cpu(&self) -> u64 {
204        self.admission
205            .and_then(|admission| admission.cpu)
206            .map_or(self.cpus, |cpu| cpu.reserved)
207    }
208
209    fn running_job(&self) -> api::RunningJob {
210        api::RunningJob {
211            repository: self.repository.clone(),
212            runner: self.runner_name.clone(),
213            job: self.job,
214        }
215    }
216}
217
218#[derive(Debug, Default)]
219struct RepositoryState {
220    queued: Vec<QueuedJob>,
221    /// Whether the last poll of its queued jobs succeeded.
222    listed: bool,
223    held_until: Option<SystemTime>,
224    /// Runners in a row that ended without taking a job.
225    failures: u32,
226    /// Runners in a row that ended without a job after waiting `IDLE_TIMEOUT` for one.
227    idle_exits: u32,
228    /// Its oldest job held for CPU that memory and a slot had room for, and since when, kept while that job stays
229    /// queued: a tick held by memory or slots in between does not make it younger.
230    cpu_wait: Option<CpuWait>,
231    /// Whether the last plan held its next job for CPU.
232    cpu_held: bool,
233}
234
235/// A job waiting for CPU, as `RepositoryState::cpu_wait` keeps it.
236#[derive(Debug, Clone, Copy)]
237struct CpuWait {
238    /// The job's ID.
239    job: u64,
240    /// When a plan first held it for CPU.
241    since: SystemTime,
242}
243
244/// The scheduler with its side effects, on the real GitHub and Docker unless a test passes fakes.
245#[derive(Debug)]
246pub struct Engine<G = Github, D = Docker> {
247    config: Config,
248    /// Shared with the listings a tick reads while it answers commands.
249    github: Arc<G>,
250    docker: Arc<D>,
251    names: Names,
252    isolation: &'static dyn Isolation,
253    identity: HostIdentity,
254    runner_prefix: String,
255    memory_max: Option<u64>,
256    /// The CPUs the slice may use, in millicores, read at start.
257    cpus: Option<u64>,
258    slots: BTreeMap<String, Vec<Option<Occupant>>>,
259    repositories: BTreeMap<String, RepositoryState>,
260    reseeds: BTreeMap<String, JoinHandle<Result<String, docker::Error>>>,
261    /// The pulled images of the config's profiles, and the removal of those no profile runs.
262    images: Images,
263    /// The memory cap each running reseed's container started with, which a reload does not change.
264    reseed_memory: BTreeMap<String, u64>,
265    /// Profiles the API asked to reseed: they start nothing until their slots are free and the reseed ran.
266    reseed_requested: BTreeSet<String>,
267    /// Repositories a reload removed, polled until none of this machine's runners is registered on them.
268    draining: BTreeSet<String>,
269    /// Profiles a reload removed, kept while their runners or reseed still run.
270    retired: BTreeMap<String, crate::config::Profile>,
271    /// Profiles whose slot 1 ran a job since their last reseed check.
272    reseed_due: BTreeSet<String>,
273    /// Whether a job ended since the shared caches were last checked.
274    trim_due: bool,
275    /// The running check of the shared caches: Docker takes many seconds to measure volumes.
276    trim: Option<JoinHandle<Result<Trimmed, docker::Error>>>,
277    /// When the last check of the shared caches started; the daemon's start counts as one, as the first runs then
278    /// unless runners were adopted.
279    checked_at: SystemTime,
280    mirror_up: bool,
281    /// The memory cap the running mirror started with.
282    mirror_memory: u64,
283    rate_limit_logged: bool,
284    /// Runners without a container that GitHub still holds busy, already logged.
285    busy_orphans: BTreeSet<u64>,
286    status: watch::Sender<api::Status>,
287    /// The profiles as the last run announced them, until the first publish compares the profiles with them. Their
288    /// slots are older than the `slot.changed` events clients have, so that publish announces every slot instead.
289    announced_profiles: Option<Vec<api::ProfileStatus>>,
290    control: Arc<Control>,
291    commands: Commands,
292    caches: watch::Sender<Vec<api::CacheVolume>>,
293    store: Store,
294    paths: Paths,
295    /// The containers the sampler reads.
296    tracked: watch::Sender<Vec<Tracked>>,
297    found: (mpsc::UnboundedSender<Found>, mpsc::UnboundedReceiver<Found>),
298    captures: BTreeMap<String, JoinHandle<()>>,
299    /// Who held which address, for the mirror's log.
300    leases: Leases,
301    /// How far the mirror's log was counted.
302    mirror_until: MirrorPosition,
303    /// Jobs recorded queued and not yet taken or closed, with their repository.
304    queued_jobs: BTreeMap<u64, String>,
305    /// Runners of runs a forced pause cancelled, with when they are removed if they still run.
306    cancelling: BTreeMap<String, tokio::time::Instant>,
307    /// Jobs whose runner is gone, waiting for GitHub to report them completed.
308    closing: BTreeMap<u64, Closing>,
309    last_prune: Option<SystemTime>,
310    /// The records of failed jobs and the shells open in them.
311    records: records::RecordState,
312    /// Under measured admission, the predictor and the high-water mark.
313    measured: admission::Measured,
314    /// What warm runners learned from the history, and the workflow runs seen lately.
315    warm: warm::Warm,
316}
317
318fn unix_secs(time: SystemTime) -> u64 {
319    time.duration_since(SystemTime::UNIX_EPOCH)
320        .unwrap_or_default()
321        .as_secs()
322}
323
324impl<G: GithubApi + 'static, D: DockerApi + 'static> Engine<G, D> {
325    /// An engine for a validated config. `identity` names this machine's runners, so a restarted daemon recognizes
326    /// them; `memory_max` is the slice's.
327    pub fn new(config: Config, github: G, docker: D, setup: Setup) -> (Engine<G, D>, Handle) {
328        let Setup {
329            names,
330            identity,
331            memory_max,
332            cpus,
333            store,
334            announced,
335            paused,
336            paths,
337        } = setup;
338        let slots = config
339            .profiles
340            .iter()
341            .map(|(name, profile)| (name.clone(), vec![None; profile.slots as usize]))
342            .collect();
343        let repositories = config
344            .repositories
345            .iter()
346            .map(|repository| (repository.name.clone(), RepositoryState::default()))
347            .collect();
348        let (control, commands) = Control::new(
349            config
350                .repositories
351                .iter()
352                .map(|repository| repository.name.clone()),
353            paused,
354            store.clone(),
355        );
356        let control = Arc::new(control);
357        let (status, receiver) = watch::channel(api::Status {
358            version: env!("CARGO_PKG_VERSION").to_owned(),
359            shutting_down: false,
360            budget: api::Budget {
361                slice: config.host.slice.clone(),
362                memory_max,
363                ..api::Budget::default()
364            },
365            profiles: Vec::new(),
366            // What clients resuming after the last run's events have, so the first publish announces the difference.
367            repositories: announced.repositories,
368            queued: Vec::new(),
369        });
370        let (caches, caches_receiver) = watch::channel(cache::volumes(&names, &config));
371        let (records, records_receiver) = records::RecordState::new();
372        let mut engine = Engine {
373            isolation: isolation::backend(config.host.isolation),
374            // Unknown: an earlier run may have left the mirror up, with the configured cap at best.
375            mirror_memory: config.mirror_memory(),
376            config,
377            github: Arc::new(github),
378            docker: Arc::new(docker),
379            names,
380            runner_prefix: identity.prefix(),
381            identity,
382            memory_max,
383            cpus,
384            slots,
385            repositories,
386            reseeds: BTreeMap::new(),
387            images: Images::new(),
388            reseed_memory: BTreeMap::new(),
389            reseed_requested: BTreeSet::new(),
390            retired: BTreeMap::new(),
391            draining: BTreeSet::new(),
392            reseed_due: BTreeSet::new(),
393            // So the caches are measured once at start.
394            trim_due: true,
395            trim: None,
396            checked_at: SystemTime::now(),
397            // Unknown: an earlier run may have left it up, so the first idle tick stops it.
398            mirror_up: true,
399            rate_limit_logged: false,
400            busy_orphans: BTreeSet::new(),
401            status,
402            announced_profiles: Some(announced.profiles),
403            control: control.clone(),
404            commands,
405            caches,
406            store: store.clone(),
407            paths,
408            tracked: watch::channel(Vec::new()).0,
409            found: mpsc::unbounded_channel(),
410            captures: BTreeMap::new(),
411            leases: Leases::default(),
412            mirror_until: MirrorPosition::default(),
413            queued_jobs: BTreeMap::new(),
414            cancelling: BTreeMap::new(),
415            closing: BTreeMap::new(),
416            last_prune: None,
417            measured: admission::Measured::default(),
418            warm: warm::Warm::default(),
419            records,
420        };
421        engine.publish();
422        (
423            engine,
424            Handle {
425                status: receiver,
426                control,
427                store,
428                caches: caches_receiver,
429                records: records_receiver,
430            },
431        )
432    }
433
434    /// Runs until a shutdown request is carried out.
435    pub async fn run(mut self) {
436        tokio::spawn(metrics::run(
437            Sampler::new(self.paths.clone(), &self.config.host.slice),
438            self.tracked.subscribe(),
439            self.store.clone(),
440        ));
441        // Until Docker answers: scheduling without knowing the containers already running would start a second
442        // runner on a slot's volume and delete the running one's registration.
443        while let Err(error) = self.recover().await {
444            warn!("cannot list the containers an earlier run left; trying again in 5s: {error}");
445            if self.control.mode() != Shutdown::No {
446                return;
447            }
448            tokio::select! {
449                () = tokio::time::sleep(Duration::from_secs(5)) => {}
450                () = self.control.wake.notified() => {}
451            }
452        }
453        loop {
454            let mode = self.control.mode();
455            self.tick(mode).await;
456            self.publish();
457            let idle = self.slots.values().flatten().all(Option::is_none);
458            if mode != Shutdown::No
459                && idle
460                && self.reseeds.is_empty()
461                && self.trim.is_none()
462                && self.records.recordings.is_empty()
463            {
464                break;
465            }
466            let interval = if mode == Shutdown::No {
467                self.config.host.poll_interval
468            } else {
469                Duration::from_secs(5)
470            };
471            tokio::select! {
472                () = tokio::time::sleep(interval) => {}
473                () = self.control.wake.notified() => {}
474                Some(command) = self.commands.recv() => {
475                    self.command(command).await;
476                    self.publish();
477                }
478                Some(()) = self.records.finished.recv() => {
479                    self.collect_recordings().await;
480                    self.publish();
481                }
482            }
483        }
484        self.close_shells().await;
485        if let Err(error) = cache::mirror_down(&*self.docker, &self.names).await {
486            warn!("cannot stop the mirror: {error}");
487        }
488        info!("stopped");
489    }
490
491    /// Runs `work` as a task of its own, answering commands and publishing after each until it is done, so a command
492    /// waits for a slow GitHub no longer than for the command before it. The task goes on while a command runs: a
493    /// request's timeout runs on the clock, not on polls.
494    async fn serving<T: Send + 'static>(
495        &mut self,
496        work: impl Future<Output = T> + Send + 'static,
497    ) -> T {
498        let mut task = tokio::spawn(work);
499        loop {
500            tokio::select! {
501                biased;
502                done = &mut task => match done.map_err(tokio::task::JoinError::try_into_panic) {
503                    Ok(output) => return output,
504                    // As if it had panicked here.
505                    Err(Ok(panic)) => std::panic::resume_unwind(panic),
506                    // Only while the runtime shuts down, which drops this task too.
507                    Err(Err(_)) => std::future::pending::<()>().await,
508                },
509                Some(command) = self.commands.recv() => {
510                    self.command(command).await;
511                    self.publish();
512                }
513            }
514        }
515    }
516
517    fn occupied(&self) -> impl Iterator<Item = (SlotRef, &Occupant)> {
518        self.slots.iter().flat_map(|(profile, slots)| {
519            slots.iter().enumerate().filter_map(|(index, slot)| {
520                slot.as_ref().map(|occupant| {
521                    (
522                        SlotRef {
523                            profile: profile.clone(),
524                            index,
525                        },
526                        occupant,
527                    )
528                })
529            })
530        })
531    }
532
533    /// Drops the free slots past each profile's configured number, and the profiles a reload removed once nothing
534    /// runs in them.
535    fn drop_closed_slots(&mut self) {
536        for (name, slots) in &mut self.slots {
537            let count = self
538                .config
539                .profiles
540                .get(name)
541                .map_or(0, |profile| profile.slots as usize);
542            while slots.len() > count && slots.last().is_some_and(Option::is_none) {
543                slots.pop();
544            }
545        }
546        let profiles = &self.config.profiles;
547        self.slots
548            .retain(|name, slots| !slots.is_empty() || profiles.contains_key(name));
549        let (slots, reseeds) = (&self.slots, &self.reseeds);
550        self.retired
551            .retain(|name, _| slots.contains_key(name) || reseeds.contains_key(name));
552    }
553}
554
555#[cfg(test)]
556mod tests;
runnerd_core/scheduler/engine.rs

runnerd_core/scheduler/
engine.rs
