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

runnerd_core/scheduler/
engine.rs
