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

runnerd_core/
cache.rs
