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

runnerd_core/scheduler/
engine.rs
