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

runnerd_core/
cache.rs
