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/// What `trim_shared` measured and emptied.
190#[derive(Debug, Clone, PartialEq, Eq)]
191pub struct Trimmed {
192    /// Every volume's size in bytes, by name, as measured before emptying any.
193    pub sizes: BTreeMap<String, u64>,
194    /// The shared caches it emptied.
195    pub emptied: Vec<Emptied>,
196}
197
198/// A shared cache `trim_shared` emptied.
199#[derive(Debug, Clone, PartialEq, Eq)]
200pub struct Emptied {
201    /// The volume.
202    pub name: String,
203    /// How large it had grown.
204    pub size: ByteSize,
205}
206
207/// Empties the shared caches that outgrew their limit; call only while no runner runs.
208pub async fn trim_shared(
209    docker: &impl DockerApi,
210    names: &Names,
211    caches: &config::Caches,
212) -> Result<Trimmed, Error> {
213    let sizes = docker.volume_sizes().await?;
214    let mut emptied = Vec::new();
215    for (name, limit) in [
216        (names.pnpm(), caches.pnpm.max_size),
217        (names.maven(), caches.maven.max_size),
218    ] {
219        if let Some(&size) = sizes.get(&name)
220            && size > limit.0
221        {
222            docker.remove_volume(&name).await?;
223            emptied.push(Emptied {
224                name,
225                size: ByteSize(size),
226            });
227        }
228    }
229    Ok(Trimmed { sizes, emptied })
230}
231
232/// Every cache volume the config's profiles use, what it holds and its limit, with size 0.
233pub fn volumes(names: &Names, config: &config::Config) -> Vec<api::CacheVolume> {
234    let volume = |name: String, kind: api::CacheKind, limit: Option<u64>| api::CacheVolume {
235        name,
236        kind,
237        bytes: 0,
238        limit,
239    };
240    let mut volumes = Vec::new();
241    for (profile, settings) in &config.profiles {
242        volumes.push(volume(
243            names.base(profile),
244            api::CacheKind::Snapshot {
245                profile: profile.clone(),
246            },
247            None,
248        ));
249        for slot in 1..=settings.slots as usize {
250            volumes.push(volume(
251                names.slot(profile, slot),
252                api::CacheKind::Slot {
253                    profile: profile.clone(),
254                    slot,
255                },
256                None,
257            ));
258        }
259    }
260    volumes.push(volume(
261        names.pnpm(),
262        api::CacheKind::Pnpm,
263        Some(config.caches.pnpm.max_size.0),
264    ));
265    volumes.push(volume(
266        names.maven(),
267        api::CacheKind::Maven,
268        Some(config.caches.maven.max_size.0),
269    ));
270    if config.caches.mirror.enabled {
271        volumes.push(volume(names.mirror(), api::CacheKind::Mirror, None));
272    }
273    volumes.sort_by(|a, b| a.name.cmp(&b.name));
274    volumes
275}
276
277/// What `reseed` runs on: the profile's image and limits, its slots and the isolation backend.
278#[derive(Debug)]
279pub struct Reseed<'profile> {
280    /// Where the profile's volumes are.
281    pub names: &'profile Names,
282    /// The profile to reseed.
283    pub profile: &'profile str,
284    /// The profile's runner image, which holds the seed script.
285    pub image: &'profile str,
286    /// How many slots the profile has; each slot's layer changes are dropped.
287    pub slots: usize,
288    /// The systemd slice the seed container runs in.
289    pub slice: &'profile str,
290    /// The profile's CPUs.
291    pub cpus: u32,
292    /// The profile's memory.
293    pub memory: ByteSize,
294    /// Slot 1's layers must have changed at least this much (`caches.reseed_at`).
295    pub min_changes: ByteSize,
296    /// How the seed container is isolated, as runners are.
297    pub isolation: &'profile dyn Isolation,
298}
299
300/// Makes slot 1's layers the profile's new snapshot and drops every slot's own layer changes (`image/seed.sh`),
301/// if slot 1 changed at least `min_changes`. Call only while none of the profile's slots runs. Returns the seed
302/// container's output.
303pub async fn reseed(docker: &impl DockerApi, reseed: &Reseed<'_>) -> Result<String, Error> {
304    let names = reseed.names;
305    let mut mounts = vec![volume(names.base(reseed.profile), "/base", false)];
306    mounts.extend((1..=reseed.slots).map(|slot| {
307        volume(
308            names.slot(reseed.profile, slot),
309            &format!("/workers/{slot}"),
310            false,
311        )
312    }));
313    for mount in &mounts {
314        if let Some(source) = &mount.source {
315            docker.ensure_volume(source).await?;
316        }
317    }
318    let name = names.seed(reseed.profile);
319    let spec = RunnerSpec {
320        name: name.clone(),
321        image: reseed.image.to_owned(),
322        profile: reseed.profile.to_owned(),
323        slice: reseed.slice.to_owned(),
324        cpus: reseed.cpus,
325        memory: reseed.memory,
326        env: Vec::new(),
327        mounts,
328        network: Some("none".to_owned()),
329        labels: BTreeMap::from([("runnerd.seed".to_owned(), String::new())]),
330        entrypoint: Some(vec![
331            "/usr/local/sbin/runnerd-seed".to_owned(),
332            reseed.min_changes.0.to_string(),
333        ]),
334        jit_config: None,
335    };
336    // A seed container left by a daemon that died while seeding.
337    match docker.remove(&name).await {
338        Err(error) if !is_not_found(&error) => return Err(error),
339        _ => {}
340    }
341    let id = docker.start(&spec, reseed.isolation).await?;
342    finish_reseed(docker, &id).await
343}
344
345/// Waits for a seed container to exit, removes it, and returns its output. Also takes over a seed container an
346/// earlier daemon left running: stopping it between its `rm` and `mv` would lose that snapshot layer.
347pub async fn finish_reseed(docker: &impl DockerApi, container: &str) -> Result<String, Error> {
348    let result = async {
349        let code = docker.wait(container).await?;
350        let output = docker.logs(container).await?;
351        if code == 0 {
352            Ok(output)
353        } else {
354            Err(Error::DockerContainerWaitError {
355                error: format!("seeding exited {code}: {}", output.trim()),
356                code,
357            })
358        }
359    }
360    .await;
361    let removed = docker.remove(container).await;
362    let output = result?;
363    removed?;
364    Ok(output)
365}
runnerd_core/cache.rs

runnerd_core/
cache.rs
