runnerd_core/scheduler/
engine.rs

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

runnerd_core/scheduler/
engine.rs
