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 container that rebuilds the profile's snapshot.
67    pub fn seed(&self, profile: &str) -> String {
68        format!("{}-{profile}-seed", self.prefix)
69    }
70
71    /// The mirror's container and its storage volume.
72    pub fn mirror(&self) -> String {
73        format!("{}-mirror", self.prefix)
74    }
75
76    /// What `GET /v1/caches` and `runnerctl caches clean` call the shared build cache: a directory of the daemon's,
77    /// not a Docker volume.
78    pub fn build_cache(&self) -> String {
79        format!("{}-build-cache", self.prefix)
80    }
81}
82
83fn volume(source: String, target: &str, read_only: bool) -> Mount {
84    Mount {
85        target: Some(target.to_owned()),
86        source: Some(source),
87        typ: Some(MountType::VOLUME),
88        read_only: Some(read_only),
89        ..Default::default()
90    }
91}
92
93/// What a runner container in slot `slot` (1-based) of the profile mounts (paths from `image/Dockerfile`).
94pub fn runner_mounts(names: &Names, profile: &str, slot: usize) -> Vec<Mount> {
95    vec![
96        volume(names.base(profile), "/base", true),
97        volume(names.slot(profile, slot), "/state", false),
98        volume(names.pnpm(), "/shared/pnpm", false),
99        volume(names.maven(), "/shared/maven", false),
100    ]
101}
102
103/// The mirror container: Docker Hub through a proxy registry, which refuses pushes, so a job can only make it
104/// fetch what Docker Hub serves. Content not pulled again within the TTL expires. No published port.
105pub fn mirror_body(names: &Names, mirror: &config::Mirror, slice: &str) -> ContainerCreateBody {
106    let memory = i64::try_from(mirror.memory.0).unwrap_or(i64::MAX);
107    ContainerCreateBody {
108        image: Some(MIRROR_IMAGE.to_owned()),
109        env: Some(vec![
110            "REGISTRY_PROXY_REMOTEURL=https://registry-1.docker.io".to_owned(),
111            format!("REGISTRY_PROXY_TTL={}s", mirror.ttl.as_secs()),
112            // Expiry deletes content.
113            "REGISTRY_STORAGE_DELETE_ENABLED=true".to_owned(),
114        ]),
115        labels: Some(HashMap::from([(
116            "runnerd.mirror".to_owned(),
117            String::new(),
118        )])),
119        host_config: Some(HostConfig {
120            cgroup_parent: Some(slice.to_owned()),
121            memory: Some(memory),
122            memory_swap: Some(memory),
123            nano_cpus: Some(i64::from(mirror.cpus) * 1_000_000_000),
124            // When the slice runs out of memory, which measured admission allows, the kernel kills a job's process
125            // before the mirror every runner pulls through.
126            oom_score_adj: Some(MIRROR_OOM_SCORE_ADJ),
127            mounts: Some(vec![volume(names.mirror(), "/var/lib/registry", false)]),
128            network_mode: Some(names.network()),
129            ..Default::default()
130        }),
131        networking_config: Some(NetworkingConfig {
132            endpoints_config: Some(HashMap::from([(
133                names.network(),
134                EndpointSettings {
135                    aliases: Some(vec![MIRROR_ALIAS.to_owned()]),
136                    ..Default::default()
137                },
138            )])),
139        }),
140        ..Default::default()
141    }
142}
143
144/// Creates what a profile's runner in `slot` (1-based) mounts and joins, if missing.
145pub async fn prepare(
146    docker: &impl DockerApi,
147    names: &Names,
148    profile: &str,
149    slot: usize,
150) -> Result<(), Error> {
151    docker.ensure_network(&names.network()).await?;
152    for name in [
153        names.base(profile),
154        names.slot(profile, slot),
155        names.pnpm(),
156        names.maven(),
157    ] {
158        docker.ensure_volume(&name).await?;
159    }
160    Ok(())
161}
162
163/// Starts the mirror unless it runs.
164pub async fn mirror_up(
165    docker: &impl DockerApi,
166    names: &Names,
167    mirror: &config::Mirror,
168    slice: &str,
169) -> Result<(), Error> {
170    match docker.inspect(&names.mirror()).await {
171        Ok(state) if state.running => return Ok(()),
172        // Stopped on its own: started again from scratch, its storage is the volume.
173        Ok(_) => docker.remove(&names.mirror()).await?,
174        Err(error) if is_not_found(&error) => {}
175        Err(error) => return Err(error),
176    }
177    docker.ensure_network(&names.network()).await?;
178    if !docker.image_present(MIRROR_IMAGE).await? {
179        docker.pull(MIRROR_IMAGE).await?;
180    }
181    docker
182        .run(&names.mirror(), mirror_body(names, mirror, slice))
183        .await?;
184    Ok(())
185}
186
187/// Stops the mirror; its storage stays.
188pub async fn mirror_down(docker: &impl DockerApi, names: &Names) -> Result<(), Error> {
189    match docker.remove(&names.mirror()).await {
190        Err(error) if is_not_found(&error) => Ok(()),
191        other => other,
192    }
193}
194
195/// Whether the mirror's container is down, so its storage can be removed.
196#[derive(Debug, Clone, Copy, PartialEq, Eq)]
197pub enum Mirror {
198    /// It runs, or may: an earlier daemon may have left it up.
199    Up,
200    /// It is stopped.
201    Down,
202}
203
204/// What `trim_shared` measured and emptied.
205#[derive(Debug, Clone, PartialEq, Eq)]
206pub struct Trimmed {
207    /// Every volume's size in bytes, by name, as measured before emptying any.
208    pub sizes: BTreeMap<String, u64>,
209    /// The shared caches it emptied.
210    pub emptied: Vec<Emptied>,
211}
212
213/// A shared cache `trim_shared` emptied.
214#[derive(Debug, Clone, PartialEq, Eq)]
215pub struct Emptied {
216    /// The volume.
217    pub name: String,
218    /// How large it had grown.
219    pub size: ByteSize,
220}
221
222/// Empties the shared caches that outgrew their limit, the mirror's storage among them once the mirror is down
223/// (`Mirror::Down`); call only while no runner runs.
224pub async fn trim_shared(
225    docker: &impl DockerApi,
226    names: &Names,
227    caches: &config::Caches,
228    mirror: Mirror,
229) -> Result<Trimmed, Error> {
230    let sizes = docker.volume_sizes().await?;
231    let mut emptied = Vec::new();
232    let mut limits = vec![
233        (names.pnpm(), caches.pnpm.max_size),
234        (names.maven(), caches.maven.max_size),
235    ];
236    if mirror == Mirror::Down {
237        limits.push((names.mirror(), caches.mirror.max_size));
238    }
239    for (name, limit) in limits {
240        if let Some(&size) = sizes.get(&name)
241            && size > limit.0
242        {
243            docker.remove_volume(&name).await?;
244            emptied.push(Emptied {
245                name,
246                size: ByteSize(size),
247            });
248        }
249    }
250    Ok(Trimmed { sizes, emptied })
251}
252
253/// Every cache volume the config's profiles use, what it holds and its limit, with size 0.
254pub fn volumes(names: &Names, config: &config::Config) -> Vec<api::CacheVolume> {
255    let volume = |name: String, kind: api::CacheKind, limit: Option<u64>| api::CacheVolume {
256        name,
257        kind,
258        bytes: 0,
259        limit,
260    };
261    let mut volumes = Vec::new();
262    for (profile, settings) in &config.profiles {
263        volumes.push(volume(
264            names.base(profile),
265            api::CacheKind::Snapshot {
266                profile: profile.clone(),
267            },
268            Some(config.caches.snapshot_max.0),
269        ));
270        for slot in 1..=settings.slots as usize {
271            volumes.push(volume(
272                names.slot(profile, slot),
273                api::CacheKind::Slot {
274                    profile: profile.clone(),
275                    slot,
276                },
277                None,
278            ));
279        }
280    }
281    volumes.push(volume(
282        names.pnpm(),
283        api::CacheKind::Pnpm,
284        Some(config.caches.pnpm.max_size.0),
285    ));
286    volumes.push(volume(
287        names.maven(),
288        api::CacheKind::Maven,
289        Some(config.caches.maven.max_size.0),
290    ));
291    if config.caches.mirror.enabled {
292        volumes.push(volume(
293            names.mirror(),
294            api::CacheKind::Mirror,
295            Some(config.caches.mirror.max_size.0),
296        ));
297    }
298    if config.caches.build.enabled {
299        volumes.push(volume(
300            names.build_cache(),
301            api::CacheKind::Build,
302            Some(config.caches.build.max_size.0),
303        ));
304    }
305    volumes.sort_by(|a, b| a.name.cmp(&b.name));
306    volumes
307}
308
309/// What `reseed` runs on: the profile's image and limits, its slots and the isolation backend.
310#[derive(Debug)]
311pub struct Reseed<'profile> {
312    /// Where the profile's volumes are.
313    pub names: &'profile Names,
314    /// The profile to reseed.
315    pub profile: &'profile str,
316    /// The profile's runner image, which holds the seed script.
317    pub image: &'profile str,
318    /// How many slots the profile has; each slot's layer changes are dropped.
319    pub slots: usize,
320    /// The systemd slice the seed container runs in.
321    pub slice: &'profile str,
322    /// The profile's CPUs.
323    pub cpus: u32,
324    /// The profile's memory.
325    pub memory: ByteSize,
326    /// Slot 1's layers must have changed at least this much (`caches.reseed_at`).
327    pub min_changes: ByteSize,
328    /// A snapshot that would outgrow this starts over (`caches.snapshot_max`).
329    pub max_size: ByteSize,
330    /// How the seed container is isolated, as runners are.
331    pub isolation: &'profile dyn Isolation,
332}
333
334/// Makes slot 1's layers the profile's new snapshot and drops every slot's own layer changes (`image/seed.sh`),
335/// if slot 1 changed at least `min_changes`. A snapshot that would outgrow `max_size` starts over from slot 1's
336/// changes alone, or from the image when those outgrow it too. Call only while none of the profile's slots runs. Returns the seed
337/// container's output.
338pub async fn reseed(docker: &impl DockerApi, reseed: &Reseed<'_>) -> Result<String, Error> {
339    let names = reseed.names;
340    let mut mounts = vec![volume(names.base(reseed.profile), "/base", false)];
341    mounts.extend((1..=reseed.slots).map(|slot| {
342        volume(
343            names.slot(reseed.profile, slot),
344            &format!("/workers/{slot}"),
345            false,
346        )
347    }));
348    for mount in &mounts {
349        if let Some(source) = &mount.source {
350            docker.ensure_volume(source).await?;
351        }
352    }
353    let name = names.seed(reseed.profile);
354    let spec = RunnerSpec {
355        name: name.clone(),
356        image: reseed.image.to_owned(),
357        profile: reseed.profile.to_owned(),
358        slice: reseed.slice.to_owned(),
359        cpus: reseed.cpus,
360        memory: reseed.memory,
361        env: Vec::new(),
362        mounts,
363        network: Some("none".to_owned()),
364        labels: BTreeMap::from([("runnerd.seed".to_owned(), String::new())]),
365        entrypoint: Some(vec![
366            "/usr/local/sbin/runnerd-seed".to_owned(),
367            reseed.min_changes.0.to_string(),
368            reseed.max_size.0.to_string(),
369        ]),
370        jit_config: None,
371    };
372    // A seed container left by a daemon that died while seeding.
373    match docker.remove(&name).await {
374        Err(error) if !is_not_found(&error) => return Err(error),
375        _ => {}
376    }
377    let id = docker.start(&spec, reseed.isolation).await?;
378    finish_reseed(docker, &id).await
379}
380
381/// Waits for a seed container to exit, removes it, and returns its output. Also takes over a seed container an
382/// earlier daemon left running: stopping it between its `rm` and `mv` would lose that snapshot layer.
383pub async fn finish_reseed(docker: &impl DockerApi, container: &str) -> Result<String, Error> {
384    let result = async {
385        let code = docker.wait(container).await?;
386        let output = docker.logs(container).await?;
387        if code == 0 {
388            Ok(output)
389        } else {
390            Err(Error::DockerContainerWaitError {
391                error: format!("seeding exited {code}: {}", output.trim()),
392                code,
393            })
394        }
395    }
396    .await;
397    let removed = docker.remove(container).await;
398    let output = result?;
399    removed?;
400    Ok(output)
401}
runnerd_core/cache.rs

runnerd_core/
cache.rs
