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

runnerd_core/scheduler/
engine.rs
