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

runnerd_core/scheduler/
engine.rs
