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::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; `None` for one a restart adopted, which reserves
194    /// 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}
201
202impl Occupant {
203    /// What it reserves of the budget under measured admission: its cap if a restart adopted it.
204    fn reserved(&self) -> u64 {
205        self.admission
206            .map_or(self.memory, |admission| admission.reserved)
207    }
208
209    /// The CPU it reserves, in millicores: its CPU limit unless it was admitted by CPU.
210    fn reserved_cpu(&self) -> u64 {
211        self.admission
212            .and_then(|admission| admission.cpu)
213            .map_or(self.cpus, |cpu| cpu.reserved)
214    }
215
216    fn running_job(&self) -> api::RunningJob {
217        api::RunningJob {
218            repository: self.repository.clone(),
219            runner: self.runner_name.clone(),
220            job: self.job,
221        }
222    }
223}
224
225#[derive(Debug, Default)]
226struct RepositoryState {
227    queued: Vec<QueuedJob>,
228    /// Whether the last poll of its queued jobs succeeded.
229    listed: bool,
230    held_until: Option<SystemTime>,
231    /// Runners in a row that ended without taking a job.
232    failures: u32,
233    /// Runners in a row that ended without a job after waiting `IDLE_TIMEOUT` for one.
234    idle_exits: u32,
235    /// Its oldest job held for CPU that memory and a slot had room for, and since when, kept while that job stays
236    /// queued: a tick held by memory or slots in between does not make it younger.
237    cpu_wait: Option<CpuWait>,
238    /// Whether the last plan held its next job for CPU.
239    cpu_held: bool,
240}
241
242/// A job waiting for CPU, as `RepositoryState::cpu_wait` keeps it.
243#[derive(Debug, Clone, Copy)]
244struct CpuWait {
245    /// The job's ID.
246    job: u64,
247    /// When a plan first held it for CPU.
248    since: SystemTime,
249}
250
251/// The scheduler with its side effects, on the real GitHub and Docker unless a test passes fakes.
252#[derive(Debug)]
253pub struct Engine<G = Github, D = Docker> {
254    config: Config,
255    /// Shared with the listings a tick reads while it answers commands.
256    github: Arc<G>,
257    docker: Arc<D>,
258    names: Names,
259    isolation: &'static dyn Isolation,
260    identity: HostIdentity,
261    runner_prefix: String,
262    memory_max: Option<u64>,
263    /// The CPUs the slice may use, in millicores, read at start.
264    cpus: Option<u64>,
265    slots: BTreeMap<String, Vec<Option<Occupant>>>,
266    repositories: BTreeMap<String, RepositoryState>,
267    reseeds: BTreeMap<String, JoinHandle<Result<String, docker::Error>>>,
268    /// The pulled images of the config's profiles, and the removal of those no profile runs.
269    images: Images,
270    /// The memory cap each running reseed's container started with, which a reload does not change.
271    reseed_memory: BTreeMap<String, u64>,
272    /// Profiles the API asked to reseed: they start nothing until their slots are free and the reseed ran.
273    reseed_requested: BTreeSet<String>,
274    /// Repositories a reload removed, polled until none of this machine's runners is registered on them.
275    draining: BTreeSet<String>,
276    /// Profiles a reload removed, kept while their runners or reseed still run.
277    retired: BTreeMap<String, crate::config::Profile>,
278    /// Profiles whose slot 1 ran a job since their last reseed check.
279    reseed_due: BTreeSet<String>,
280    /// Profiles whose snapshot the last check of the caches measured over `snapshot_max`: due, and reseeded however
281    /// little slot 1 changed.
282    oversized: BTreeSet<String>,
283    /// When each profile's last reseed finished: a check of the caches that started before then measured the old
284    /// snapshot.
285    reseeded_at: BTreeMap<String, SystemTime>,
286    /// What the slots' Docker Engines may keep together, from the last check of the caches (`cache::docker_disk`);
287    /// `None` until one measured Docker's disk, when each image keeps its own default.
288    docker_disk: Option<u64>,
289    /// Whether a job ended since the shared caches were last checked.
290    trim_due: bool,
291    /// The running check of the shared caches: Docker takes many seconds to measure volumes.
292    trim: Option<JoinHandle<Result<Trimmed, docker::Error>>>,
293    /// When the last check of the shared caches started; the daemon's start counts as one, as the first runs then
294    /// unless runners were adopted.
295    checked_at: SystemTime,
296    mirror_up: bool,
297    /// The memory cap the running mirror started with.
298    mirror_memory: u64,
299    rate_limit_logged: bool,
300    /// Runners without a container that GitHub still holds busy, already logged.
301    busy_orphans: BTreeSet<u64>,
302    status: watch::Sender<api::Status>,
303    /// The profiles as the last run announced them, until the first publish compares the profiles with them. Their
304    /// slots are older than the `slot.changed` events clients have, so that publish announces every slot instead.
305    announced_profiles: Option<Vec<api::ProfileStatus>>,
306    control: Arc<Control>,
307    commands: Commands,
308    caches: watch::Sender<Vec<api::CacheVolume>>,
309    store: Store,
310    paths: Paths,
311    /// The containers the sampler reads.
312    tracked: watch::Sender<Vec<Tracked>>,
313    found: (mpsc::UnboundedSender<Found>, mpsc::UnboundedReceiver<Found>),
314    captures: BTreeMap<String, JoinHandle<()>>,
315    /// Who held which address, for the mirror's log.
316    leases: Leases,
317    /// How far the mirror's log was counted.
318    mirror_until: MirrorPosition,
319    /// Jobs recorded queued and not yet taken or closed, with their repository.
320    queued_jobs: BTreeMap<u64, String>,
321    /// Runners of runs a forced pause cancelled, with when they are removed if they still run.
322    cancelling: BTreeMap<String, tokio::time::Instant>,
323    /// Jobs whose runner is gone, waiting for GitHub to report them completed.
324    closing: BTreeMap<u64, Closing>,
325    last_prune: Option<SystemTime>,
326    /// The records of failed jobs and the shells open in them.
327    records: records::RecordState,
328    /// Under measured admission, the predictor and the high-water mark.
329    measured: admission::Measured,
330    /// What warm runners learned from the history, and the workflow runs seen lately.
331    warm: warm::Warm,
332    /// The shared build cache.
333    build_cache: Option<BuildCache>,
334    /// Each repository's default branch, once a job of it needed it: the branch whose builds write to the build
335    /// cache when `[caches.build] branches` names none.
336    default_branches: BTreeMap<String, String>,
337}
338
339fn unix_secs(time: SystemTime) -> u64 {
340    time.duration_since(SystemTime::UNIX_EPOCH)
341        .unwrap_or_default()
342        .as_secs()
343}
344
345impl<G: GithubApi + 'static, D: DockerApi + 'static> Engine<G, D> {
346    /// An engine for a validated config. `identity` names this machine's runners, so a restarted daemon recognizes
347    /// them; `memory_max` is the slice's.
348    #[allow(
349        clippy::too_many_lines,
350        reason = "one line per field of the engine's state, which split up would only be listed twice"
351    )]
352    pub fn new(config: Config, github: G, docker: D, setup: Setup) -> (Engine<G, D>, Handle) {
353        let Setup {
354            names,
355            identity,
356            memory_max,
357            cpus,
358            store,
359            announced,
360            paused,
361            paths,
362            build_cache,
363        } = setup;
364        let slots = config
365            .profiles
366            .iter()
367            .map(|(name, profile)| (name.clone(), vec![None; profile.slots as usize]))
368            .collect();
369        let repositories = config
370            .repositories
371            .iter()
372            .map(|repository| (repository.name.clone(), RepositoryState::default()))
373            .collect();
374        let (control, commands) = Control::new(
375            config
376                .repositories
377                .iter()
378                .map(|repository| repository.name.clone()),
379            paused,
380            store.clone(),
381        );
382        let control = Arc::new(control);
383        let (status, receiver) = watch::channel(api::Status {
384            version: env!("CARGO_PKG_VERSION").to_owned(),
385            shutting_down: false,
386            budget: api::Budget {
387                slice: config.host.slice.clone(),
388                memory_max,
389                ..api::Budget::default()
390            },
391            profiles: Vec::new(),
392            // What clients resuming after the last run's events have, so the first publish announces the difference.
393            repositories: announced.repositories,
394            queued: Vec::new(),
395        });
396        let (caches, caches_receiver) = watch::channel(cache::volumes(&names, &config));
397        let (records, records_receiver) = records::RecordState::new();
398        let mut engine = Engine {
399            isolation: isolation::backend(config.host.isolation),
400            // Unknown: an earlier run may have left the mirror up, with the configured cap at best.
401            mirror_memory: config.mirror_memory(),
402            config,
403            github: Arc::new(github),
404            docker: Arc::new(docker),
405            names,
406            runner_prefix: identity.prefix(),
407            identity,
408            memory_max,
409            cpus,
410            slots,
411            repositories,
412            reseeds: BTreeMap::new(),
413            images: Images::new(),
414            reseed_memory: BTreeMap::new(),
415            reseed_requested: BTreeSet::new(),
416            retired: BTreeMap::new(),
417            draining: BTreeSet::new(),
418            reseed_due: BTreeSet::new(),
419            oversized: BTreeSet::new(),
420            reseeded_at: BTreeMap::new(),
421            docker_disk: None,
422            // So the caches are measured once at start.
423            trim_due: true,
424            trim: None,
425            checked_at: SystemTime::now(),
426            // Unknown: an earlier run may have left it up, so the first idle tick stops it.
427            mirror_up: true,
428            rate_limit_logged: false,
429            busy_orphans: BTreeSet::new(),
430            status,
431            announced_profiles: Some(announced.profiles),
432            control: control.clone(),
433            commands,
434            caches,
435            store: store.clone(),
436            paths,
437            tracked: watch::channel(Vec::new()).0,
438            found: mpsc::unbounded_channel(),
439            captures: BTreeMap::new(),
440            leases: Leases::default(),
441            mirror_until: MirrorPosition::default(),
442            queued_jobs: BTreeMap::new(),
443            cancelling: BTreeMap::new(),
444            closing: BTreeMap::new(),
445            last_prune: None,
446            measured: admission::Measured::default(),
447            warm: warm::Warm::default(),
448            records,
449            build_cache: build_cache.clone(),
450            default_branches: BTreeMap::new(),
451        };
452        engine.publish();
453        (
454            engine,
455            Handle {
456                status: receiver,
457                control,
458                store,
459                caches: caches_receiver,
460                records: records_receiver,
461                build_cache,
462            },
463        )
464    }
465
466    /// Runs until a shutdown request is carried out.
467    pub async fn run(mut self) {
468        tokio::spawn(metrics::run(
469            Sampler::new(self.paths.clone(), &self.config.host.slice),
470            self.tracked.subscribe(),
471            self.store.clone(),
472        ));
473        // Until Docker answers: scheduling without knowing the containers already running would start a second
474        // runner on a slot's volume and delete the running one's registration.
475        while let Err(error) = self.recover().await {
476            warn!("cannot list the containers an earlier run left; trying again in 5s: {error}");
477            if self.control.mode() != Shutdown::No {
478                return;
479            }
480            tokio::select! {
481                () = tokio::time::sleep(Duration::from_secs(5)) => {}
482                () = self.control.wake.notified() => {}
483            }
484        }
485        loop {
486            let mode = self.control.mode();
487            self.tick(mode).await;
488            self.publish();
489            let idle = self.slots.values().flatten().all(Option::is_none);
490            if mode != Shutdown::No
491                && idle
492                && self.reseeds.is_empty()
493                && self.trim.is_none()
494                && self.records.recordings.is_empty()
495            {
496                break;
497            }
498            let interval = if mode == Shutdown::No {
499                self.config.host.poll_interval
500            } else {
501                Duration::from_secs(5)
502            };
503            tokio::select! {
504                () = tokio::time::sleep(interval) => {}
505                () = self.control.wake.notified() => {}
506                Some(command) = self.commands.recv() => {
507                    self.command(command).await;
508                    self.publish();
509                }
510                Some(()) = self.records.finished.recv() => {
511                    self.collect_recordings().await;
512                    self.publish();
513                }
514            }
515        }
516        self.close_shells().await;
517        if let Err(error) = cache::mirror_down(&*self.docker, &self.names).await {
518            warn!("cannot stop the mirror: {error}");
519        }
520        info!("stopped");
521    }
522
523    /// Runs `work` as a task of its own, answering commands and publishing after each until it is done, so a command
524    /// waits for a slow GitHub no longer than for the command before it. The task goes on while a command runs: a
525    /// request's timeout runs on the clock, not on polls.
526    async fn serving<T: Send + 'static>(
527        &mut self,
528        work: impl Future<Output = T> + Send + 'static,
529    ) -> T {
530        let mut task = tokio::spawn(work);
531        loop {
532            tokio::select! {
533                biased;
534                done = &mut task => match done.map_err(tokio::task::JoinError::try_into_panic) {
535                    Ok(output) => return output,
536                    // As if it had panicked here.
537                    Err(Ok(panic)) => std::panic::resume_unwind(panic),
538                    // Only while the runtime shuts down, which drops this task too.
539                    Err(Err(_)) => std::future::pending::<()>().await,
540                },
541                Some(command) = self.commands.recv() => {
542                    self.command(command).await;
543                    self.publish();
544                }
545            }
546        }
547    }
548
549    fn occupied(&self) -> impl Iterator<Item = (SlotRef, &Occupant)> {
550        self.slots.iter().flat_map(|(profile, slots)| {
551            slots.iter().enumerate().filter_map(|(index, slot)| {
552                slot.as_ref().map(|occupant| {
553                    (
554                        SlotRef {
555                            profile: profile.clone(),
556                            index,
557                        },
558                        occupant,
559                    )
560                })
561            })
562        })
563    }
564
565    /// Drops the free slots past each profile's configured number, and the profiles a reload removed once nothing
566    /// runs in them.
567    fn drop_closed_slots(&mut self) {
568        for (name, slots) in &mut self.slots {
569            let count = self
570                .config
571                .profiles
572                .get(name)
573                .map_or(0, |profile| profile.slots as usize);
574            while slots.len() > count && slots.last().is_some_and(Option::is_none) {
575                slots.pop();
576            }
577        }
578        let profiles = &self.config.profiles;
579        self.slots
580            .retain(|name, slots| !slots.is_empty() || profiles.contains_key(name));
581        let (slots, reseeds) = (&self.slots, &self.reseeds);
582        self.retired
583            .retain(|name, _| slots.contains_key(name) || reseeds.contains_key(name));
584    }
585}
586
587#[cfg(test)]
588mod tests;
runnerd_core/scheduler/engine.rs

runnerd_core/scheduler/
engine.rs
