runnerd_core/
cache.rs

1//! The warm caches runner containers share and keep between jobs: each profile's read-only snapshot, each slot's
2//! volume (its layers' changes and its Docker Engine's data), the shared pnpm store and Maven repository, and the
3//! Docker Hub mirror on the runners' network. Owns their names, mounts and lifecycle; it never decides when a
4//! runner starts, and it never touches a volume a running container uses.
5
6use std::collections::{BTreeMap, HashMap};
7
8use bollard::models::{
9    ContainerCreateBody, EndpointSettings, HostConfig, Mount, MountType, NetworkingConfig,
10};
11
12use crate::api;
13use crate::config::{self, ByteSize};
14use crate::docker::{DockerApi, Error, RunnerSpec, is_not_found};
15use crate::isolation::Isolation;
16
17/// The Docker Hub pull-through cache (`registry` 3.1.2, multi-platform index).
18pub const MIRROR_IMAGE: &str =
19    "registry:3.1.2@sha256:01cd312e67026f06f5d89c3e133a4a9adc7cb9eeb9b1194e0b2954dfd770efd2";
20
21/// The host name runner containers reach the mirror at (`image/daemon.json`).
22const MIRROR_ALIAS: &str = "mirror";
23
24/// The mirror's `oom_score_adj`: the kernel picks it after the runners' processes, which keep the default 0.
25const MIRROR_OOM_SCORE_ADJ: i64 = -500;
26
27/// The names of everything runnerd creates in Docker, under one prefix: `runnerd`, or another for tests.
28#[derive(Debug, Clone)]
29pub struct Names {
30    prefix: String,
31}
32
33impl Names {
34    /// Names under `prefix`.
35    pub fn new(prefix: &str) -> Names {
36        Names {
37            prefix: prefix.to_owned(),
38        }
39    }
40
41    /// The network runners and the mirror share.
42    pub fn network(&self) -> String {
43        self.prefix.clone()
44    }
45
46    /// The profile's read-only snapshot.
47    pub fn base(&self, profile: &str) -> String {
48        format!("{}-{profile}-base", self.prefix)
49    }
50
51    /// Slot `slot` (1-based) of the profile: its layers' changes and its Docker Engine's data.
52    pub fn slot(&self, profile: &str, slot: usize) -> String {
53        format!("{}-{profile}-slot-{slot}", self.prefix)
54    }
55
56    /// The pnpm store all slots share.
57    pub fn pnpm(&self) -> String {
58        format!("{}-pnpm", self.prefix)
59    }
60
61    /// The Maven repository all slots share.
62    pub fn maven(&self) -> String {
63        format!("{}-maven", self.prefix)
64    }
65
66    /// The slot (1-based) of the profile whose volume `volume` is, if it is one of its slots'.
67    pub fn slot_of(&self, profile: &str, volume: &str) -> Option<usize> {
68        volume
69            .strip_prefix(&format!("{}-{profile}-slot-", self.prefix))?
70            .parse()
71            .ok()
72    }
73
74    /// The container that rebuilds the profile's snapshot.
75    pub fn seed(&self, profile: &str) -> String {
76        format!("{}-{profile}-seed", self.prefix)
77    }
78
79    /// The mirror's container and its storage volume.
80    pub fn mirror(&self) -> String {
81        format!("{}-mirror", self.prefix)
82    }
83
84    /// What `GET /v1/caches` and `runnerctl caches clean` call the shared build cache: a directory of the daemon's,
85    /// not a Docker volume.
86    pub fn build_cache(&self) -> String {
87        format!("{}-build-cache", self.prefix)
88    }
89}
90
91fn volume(source: String, target: &str, read_only: bool) -> Mount {
92    Mount {
93        target: Some(target.to_owned()),
94        source: Some(source),
95        typ: Some(MountType::VOLUME),
96        read_only: Some(read_only),
97        ..Default::default()
98    }
99}
100
101/// What a runner container in slot `slot` (1-based) of the profile mounts (paths from `image/Dockerfile`).
102pub fn runner_mounts(names: &Names, profile: &str, slot: usize) -> Vec<Mount> {
103    vec![
104        volume(names.base(profile), "/base", true),
105        volume(names.slot(profile, slot), "/state", false),
106        volume(names.pnpm(), "/shared/pnpm", false),
107        volume(names.maven(), "/shared/maven", false),
108    ]
109}
110
111/// The mirror container: Docker Hub through a proxy registry, which refuses pushes, so a job can only make it
112/// fetch what Docker Hub serves. Content not pulled again within the TTL expires. No published port.
113pub fn mirror_body(names: &Names, mirror: &config::Mirror, slice: &str) -> ContainerCreateBody {
114    let memory = i64::try_from(mirror.memory.0).unwrap_or(i64::MAX);
115    ContainerCreateBody {
116        image: Some(MIRROR_IMAGE.to_owned()),
117        env: Some(vec![
118            "REGISTRY_PROXY_REMOTEURL=https://registry-1.docker.io".to_owned(),
119            format!("REGISTRY_PROXY_TTL={}s", mirror.ttl.as_secs()),
120            // Expiry deletes content.
121            "REGISTRY_STORAGE_DELETE_ENABLED=true".to_owned(),
122        ]),
123        labels: Some(HashMap::from([(
124            "runnerd.mirror".to_owned(),
125            String::new(),
126        )])),
127        host_config: Some(HostConfig {
128            cgroup_parent: Some(slice.to_owned()),
129            memory: Some(memory),
130            memory_swap: Some(memory),
131            nano_cpus: Some(i64::from(mirror.cpus) * 1_000_000_000),
132            // When the slice runs out of memory, which measured admission allows, the kernel kills a job's process
133            // before the mirror every runner pulls through.
134            oom_score_adj: Some(MIRROR_OOM_SCORE_ADJ),
135            mounts: Some(vec![volume(names.mirror(), "/var/lib/registry", false)]),
136            network_mode: Some(names.network()),
137            ..Default::default()
138        }),
139        networking_config: Some(NetworkingConfig {
140            endpoints_config: Some(HashMap::from([(
141                names.network(),
142                EndpointSettings {
143                    aliases: Some(vec![MIRROR_ALIAS.to_owned()]),
144                    ..Default::default()
145                },
146            )])),
147        }),
148        ..Default::default()
149    }
150}
151
152/// Creates what a profile's runner in `slot` (1-based) mounts and joins, if missing.
153pub async fn prepare(
154    docker: &impl DockerApi,
155    names: &Names,
156    profile: &str,
157    slot: usize,
158) -> Result<(), Error> {
159    docker.ensure_network(&names.network()).await?;
160    for name in [
161        names.base(profile),
162        names.slot(profile, slot),
163        names.pnpm(),
164        names.maven(),
165    ] {
166        docker.ensure_volume(&name).await?;
167    }
168    Ok(())
169}
170
171/// Starts the mirror unless it runs.
172pub async fn mirror_up(
173    docker: &impl DockerApi,
174    names: &Names,
175    mirror: &config::Mirror,
176    slice: &str,
177) -> Result<(), Error> {
178    match docker.inspect(&names.mirror()).await {
179        Ok(state) if state.running => return Ok(()),
180        // Stopped on its own: started again from scratch, its storage is the volume.
181        Ok(_) => docker.remove(&names.mirror()).await?,
182        Err(error) if is_not_found(&error) => {}
183        Err(error) => return Err(error),
184    }
185    docker.ensure_network(&names.network()).await?;
186    if !docker.image_present(MIRROR_IMAGE).await? {
187        docker.pull(MIRROR_IMAGE).await?;
188    }
189    docker
190        .run(&names.mirror(), mirror_body(names, mirror, slice))
191        .await?;
192    Ok(())
193}
194
195/// Stops the mirror; its storage stays.
196pub async fn mirror_down(docker: &impl DockerApi, names: &Names) -> Result<(), Error> {
197    match docker.remove(&names.mirror()).await {
198        Err(error) if is_not_found(&error) => Ok(()),
199        other => other,
200    }
201}
202
203/// Whether the mirror's container is down, so its storage can be removed.
204#[derive(Debug, Clone, Copy, PartialEq, Eq)]
205pub enum Mirror {
206    /// It runs, or may: an earlier daemon may have left it up.
207    Up,
208    /// It is stopped.
209    Down,
210}
211
212/// What `trim_shared` measured and emptied.
213#[derive(Debug, Clone, PartialEq, Eq)]
214pub struct Trimmed {
215    /// Every volume's size in bytes, by name, as measured before emptying any.
216    pub sizes: BTreeMap<String, u64>,
217    /// The shared caches it emptied.
218    pub emptied: Vec<Emptied>,
219    /// The slot volumes past their profile's slot count it removed.
220    pub closed: Vec<Emptied>,
221    /// The volumes it was to empty or remove that Docker refused to, as one a stopped container still uses: tried
222    /// again at the next check.
223    pub refused: Vec<Refused>,
224    /// Docker's free disk in bytes, as measured before emptying any; `None` when it could not be measured.
225    pub free: Option<u64>,
226}
227
228/// A volume `trim_shared` was to remove that Docker refused to.
229#[derive(Debug, Clone, PartialEq, Eq)]
230pub struct Refused {
231    /// The volume.
232    pub name: String,
233    /// What Docker said.
234    pub error: String,
235}
236
237/// A shared cache `trim_shared` emptied.
238#[derive(Debug, Clone, PartialEq, Eq)]
239pub struct Emptied {
240    /// The volume.
241    pub name: String,
242    /// How large it had grown.
243    pub size: ByteSize,
244}
245
246/// Empties the shared caches that outgrew their limit, the mirror's storage among them once the mirror is down
247/// (`Mirror::Down`), and removes the volumes of slots past the count `slots` gives their profile, as a lowered count
248/// leaves them: each holds a slot's layers and Docker store, which no job prunes again (docs/adr/0054). Call only
249/// while no runner runs, and leave out of `slots` a profile whose reseed runs, which mounts its slots' volumes.
250pub async fn trim_shared(
251    docker: &impl DockerApi,
252    names: &Names,
253    caches: &config::Caches,
254    mirror: Mirror,
255    slots: &BTreeMap<String, usize>,
256) -> Result<Trimmed, Error> {
257    let sizes = docker.volume_sizes().await?;
258    let free = docker.free_space().await.ok();
259    let mut emptied = Vec::new();
260    let mut refused = Vec::new();
261    let mut limits = vec![
262        (names.pnpm(), caches.pnpm.max_size),
263        (names.maven(), caches.maven.max_size),
264    ];
265    if mirror == Mirror::Down {
266        limits.push((names.mirror(), caches.mirror.max_size));
267    }
268    for (name, limit) in limits {
269        if let Some(&size) = sizes.get(&name)
270            && size > limit.0
271        {
272            // One Docker refuses stays to the next check; the others are still trimmed, and the sizes kept.
273            if let Err(error) = docker.remove_volume(&name).await {
274                refused.push(Refused {
275                    name,
276                    error: error.to_string(),
277                });
278                continue;
279            }
280            emptied.push(Emptied {
281                name,
282                size: ByteSize(size),
283            });
284        }
285    }
286    let mut closed = Vec::new();
287    for (name, &size) in &sizes {
288        let past = slots.iter().any(|(profile, &count)| {
289            names
290                .slot_of(profile, name)
291                .is_some_and(|slot| slot > count)
292        });
293        if past {
294            // As a stopped container kept as a failed job's record still uses it.
295            if let Err(error) = docker.remove_volume(name).await {
296                refused.push(Refused {
297                    name: name.clone(),
298                    error: error.to_string(),
299                });
300                continue;
301            }
302            closed.push(Emptied {
303                name: name.clone(),
304                size: ByteSize(size),
305            });
306        }
307    }
308    Ok(Trimmed {
309        sizes,
310        emptied,
311        closed,
312        refused,
313        free,
314    })
315}
316
317/// What the slots' Docker Engines may keep together, in bytes: a quarter of Docker's disk not otherwise used, which
318/// is what is free plus what the slot volumes already hold, measured together (docs/adr/0054).
319pub fn docker_disk(
320    names: &Names,
321    config: &config::Config,
322    free: u64,
323    sizes: &BTreeMap<String, u64>,
324) -> u64 {
325    let held: u64 = config
326        .profiles
327        .iter()
328        .flat_map(|(profile, settings)| {
329            (1..=settings.slots as usize).map(move |slot| names.slot(profile, slot))
330        })
331        .filter_map(|name| sizes.get(&name))
332        .sum();
333    free.saturating_add(held) / DOCKER_DISK_SHARE
334}
335
336/// One slot's share of `disk` among `slots`, in whole GiB as `image/layers.sh` takes it: at least 1, at most
337/// `DOCKER_SLOT_MAX_GIB`.
338pub fn docker_budget_gib(disk: u64, slots: u64) -> u64 {
339    ((disk / slots.max(1)) >> 30).clamp(1, DOCKER_SLOT_MAX_GIB)
340}
341
342/// The slots' Docker Engines together keep at most one part in this of Docker's disk not otherwise used.
343const DOCKER_DISK_SHARE: u64 = 4;
344
345/// The most one slot's Docker Engine keeps, in GiB: the image's own default, for a daemon that sets none.
346const DOCKER_SLOT_MAX_GIB: u64 = 20;
347
348/// Every cache volume the config's profiles use, what it holds and its limit, with size 0.
349pub fn volumes(names: &Names, config: &config::Config) -> Vec<api::CacheVolume> {
350    let volume = |name: String, kind: api::CacheKind, limit: Option<u64>| api::CacheVolume {
351        name,
352        kind,
353        bytes: 0,
354        limit,
355    };
356    let mut volumes = Vec::new();
357    for (profile, settings) in &config.profiles {
358        volumes.push(volume(
359            names.base(profile),
360            api::CacheKind::Snapshot {
361                profile: profile.clone(),
362            },
363            Some(config.caches.snapshot_max.0),
364        ));
365        for slot in 1..=settings.slots as usize {
366            volumes.push(volume(
367                names.slot(profile, slot),
368                api::CacheKind::Slot {
369                    profile: profile.clone(),
370                    slot,
371                },
372                None,
373            ));
374        }
375    }
376    volumes.push(volume(
377        names.pnpm(),
378        api::CacheKind::Pnpm,
379        Some(config.caches.pnpm.max_size.0),
380    ));
381    volumes.push(volume(
382        names.maven(),
383        api::CacheKind::Maven,
384        Some(config.caches.maven.max_size.0),
385    ));
386    if config.caches.mirror.enabled {
387        volumes.push(volume(
388            names.mirror(),
389            api::CacheKind::Mirror,
390            Some(config.caches.mirror.max_size.0),
391        ));
392    }
393    if config.caches.build.enabled {
394        volumes.push(volume(
395            names.build_cache(),
396            api::CacheKind::Build,
397            Some(config.caches.build.max_size.0),
398        ));
399    }
400    volumes.sort_by(|a, b| a.name.cmp(&b.name));
401    volumes
402}
403
404/// What `reseed` runs on: the profile's image and limits, its slots and the isolation backend.
405#[derive(Debug)]
406pub struct Reseed<'profile> {
407    /// Where the profile's volumes are.
408    pub names: &'profile Names,
409    /// The profile to reseed.
410    pub profile: &'profile str,
411    /// The profile's runner image, which holds the seed script.
412    pub image: &'profile str,
413    /// How many slots the profile has; each slot's layer changes are dropped.
414    pub slots: usize,
415    /// The systemd slice the seed container runs in.
416    pub slice: &'profile str,
417    /// The profile's CPUs.
418    pub cpus: u32,
419    /// The profile's memory.
420    pub memory: ByteSize,
421    /// Slot 1's layers must have changed at least this much (`caches.reseed_at`).
422    pub min_changes: ByteSize,
423    /// A snapshot that would outgrow this starts over (`caches.snapshot_max`).
424    pub max_size: ByteSize,
425    /// How the seed container is isolated, as runners are.
426    pub isolation: &'profile dyn Isolation,
427}
428
429/// Makes slot 1's layers the profile's new snapshot and drops every slot's own layer changes (`image/seed.sh`),
430/// if slot 1 changed at least `min_changes`. A snapshot that would outgrow `max_size` starts over from the image,
431/// as every slot's changes go too (docs/adr/0065); after a reseed that did not finish, it starts over whatever the
432/// limits (docs/adr/0066). Call only while none of the profile's slots runs. Returns the seed container's output.
433pub async fn reseed(docker: &impl DockerApi, reseed: &Reseed<'_>) -> Result<String, Error> {
434    let names = reseed.names;
435    let mut mounts = vec![volume(names.base(reseed.profile), "/base", false)];
436    mounts.extend((1..=reseed.slots).map(|slot| {
437        volume(
438            names.slot(reseed.profile, slot),
439            &format!("/workers/{slot}"),
440            false,
441        )
442    }));
443    for mount in &mounts {
444        if let Some(source) = &mount.source {
445            docker.ensure_volume(source).await?;
446        }
447    }
448    let name = names.seed(reseed.profile);
449    let spec = RunnerSpec {
450        name: name.clone(),
451        image: reseed.image.to_owned(),
452        profile: reseed.profile.to_owned(),
453        slice: reseed.slice.to_owned(),
454        cpus: reseed.cpus,
455        memory: reseed.memory,
456        env: Vec::new(),
457        mounts,
458        network: Some("none".to_owned()),
459        labels: BTreeMap::from([("runnerd.seed".to_owned(), String::new())]),
460        entrypoint: Some(vec![
461            "/usr/local/sbin/runnerd-seed".to_owned(),
462            reseed.min_changes.0.to_string(),
463            reseed.max_size.0.to_string(),
464        ]),
465        jit_config: None,
466    };
467    // A seed container left by a daemon that died while seeding.
468    match docker.remove(&name).await {
469        Err(error) if !is_not_found(&error) => return Err(error),
470        _ => {}
471    }
472    let id = docker.start(&spec, reseed.isolation).await?;
473    finish_reseed(docker, &id).await
474}
475
476/// Waits for a seed container to exit, removes it, and returns its output. Also takes over a seed container an
477/// earlier daemon left running: stopping it between its `rm` and `mv` would lose that snapshot layer.
478pub async fn finish_reseed(docker: &impl DockerApi, container: &str) -> Result<String, Error> {
479    let result = async {
480        let code = docker.wait(container).await?;
481        let output = docker.logs(container).await?;
482        if code == 0 {
483            Ok(output)
484        } else {
485            Err(Error::DockerContainerWaitError {
486                error: format!("seeding exited {code}: {}", output.trim()),
487                code,
488            })
489        }
490    }
491    .await;
492    let removed = docker.remove(container).await;
493    let output = result?;
494    removed?;
495    Ok(output)
496}
497
498#[cfg(test)]
499mod tests {
500    use super::docker_budget_gib;
501
502    #[test]
503    fn a_slots_docker_budget_is_its_share_between_1_and_20_gib() {
504        assert_eq!(docker_budget_gib(100 << 30, 10), 10);
505        assert_eq!(
506            docker_budget_gib(1 << 40, 4),
507            20,
508            "a large disk still stops at 20 GiB"
509        );
510        assert_eq!(
511            docker_budget_gib(1 << 30, 13),
512            1,
513            "a full disk still keeps 1 GiB"
514        );
515        assert_eq!(docker_budget_gib(1 << 30, 0), 1);
516    }
517}
runnerd_core/cache.rs

runnerd_core/
cache.rs
