runnerd_core/
docker.rs

1//! Runner containers through the Docker Engine API: the connection, container, volume and image requests, and the
2//! `DockerApi` trait the engine and the caches call. It decides nothing about scheduling, and a JIT config only
3//! ever reaches a container as a root-only file, never the create request or a log line.
4
5use std::collections::{BTreeMap, HashMap};
6use std::fmt;
7use std::time::{Duration, Instant};
8
9use bollard::exec::{CreateExecOptions, StartExecResults};
10use bollard::models::{
11    ContainerCreateBody, CreateImageInfo, HostConfig, Mount, NetworkCreateRequest, SystemInfo,
12    VolumeCreateRequest,
13};
14use bollard::query_parameters::{
15    CreateContainerOptionsBuilder, CreateImageOptionsBuilder, DataUsageOptionsBuilder,
16    ListContainersOptionsBuilder, LogsOptionsBuilder, RemoveContainerOptionsBuilder,
17    RemoveVolumeOptionsBuilder, UploadToContainerOptionsBuilder, WaitContainerOptionsBuilder,
18};
19use futures_util::stream::BoxStream;
20use futures_util::{StreamExt, TryStreamExt};
21use tracing::info;
22
23use crate::config::ByteSize;
24use crate::isolation::Isolation;
25
26pub use bollard::errors::Error;
27
28/// One runner container: a profile's image stage with the profile's limits.
29#[derive(Debug, Clone)]
30pub struct RunnerSpec {
31    /// The container's name; for a runner, also the runner's.
32    pub name: String,
33    /// The image, such as `runnerd-runner:base`.
34    pub image: String,
35    /// The profile, put on the `runnerd.profile` label.
36    pub profile: String,
37    /// The systemd slice the container is placed in.
38    pub slice: String,
39    /// CPUs the container may use.
40    pub cpus: u32,
41    /// Memory the container may use, swap included.
42    pub memory: ByteSize,
43    /// `NAME=value` environment variables.
44    pub env: Vec<String>,
45    /// Volumes to mount.
46    pub mounts: Vec<Mount>,
47    /// The network the container joins; `None` is Docker's default bridge.
48    pub network: Option<String>,
49    /// Labels on top of `runnerd.profile` and `runnerd.isolation`.
50    pub labels: BTreeMap<String, String>,
51    /// Replaces the image's entrypoint; `None` runs the runner.
52    pub entrypoint: Option<Vec<String>>,
53    /// The runner's just-in-time config, which the entrypoint reads from `JIT_CONFIG_PATH`.
54    pub jit_config: Option<JitConfig>,
55}
56
57/// Where `start` puts the just-in-time config in the container: a root-only file, never the create request,
58/// which Docker stores and clients may log.
59pub const JIT_CONFIG_PATH: &str = "/run/runnerd/jitconfig";
60
61/// A runner's just-in-time config from GitHub: credentials, never printed.
62#[derive(Clone)]
63pub struct JitConfig(pub String);
64
65impl fmt::Debug for JitConfig {
66    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
67        f.write_str("JitConfig(<redacted>)")
68    }
69}
70
71/// What `inspect` reports about a runner container.
72#[derive(Debug, Clone, PartialEq, Eq)]
73pub struct RunnerState {
74    /// Whether the container runs.
75    pub running: bool,
76    /// Its exit code, once it stopped.
77    pub exit_code: Option<i64>,
78    /// True under `unsafe-dind`: a job in the container can take over the host.
79    pub privileged: bool,
80    /// The runtime Docker runs it with, such as `sysbox-runc`.
81    pub runtime: Option<String>,
82    /// Its main process on the host, while it runs.
83    pub pid: Option<u32>,
84    /// Its address on its first network.
85    pub address: Option<String>,
86    /// When it started, in milliseconds since the epoch.
87    pub started_at: Option<u64>,
88    /// Its memory cap in bytes; `None` when it has none.
89    pub memory: Option<u64>,
90    /// Its CPU limit in millicores; `None` when it has none.
91    pub cpus: Option<u64>,
92}
93
94/// What a command run in a container prints on stdout, as it prints it.
95pub type Output = BoxStream<'static, Result<Vec<u8>, Error>>;
96
97/// The create-container request for a runner container under an isolation backend.
98pub fn create_body(spec: &RunnerSpec, isolation: &dyn Isolation) -> ContainerCreateBody {
99    // Clamped to 8 EiB, the most Docker takes.
100    let memory = i64::try_from(spec.memory.0).unwrap_or(i64::MAX);
101    let mut host_config = HostConfig {
102        cgroup_parent: Some(spec.slice.clone()),
103        memory: Some(memory),
104        // Swap equal to memory: the container gets no swap on top of its memory.
105        memory_swap: Some(memory),
106        nano_cpus: Some(i64::from(spec.cpus) * 1_000_000_000),
107        init: Some(true),
108        mounts: Some(spec.mounts.clone()),
109        network_mode: spec.network.clone(),
110        ..Default::default()
111    };
112    isolation.configure(&mut host_config);
113    let mut labels: HashMap<String, String> = spec.labels.clone().into_iter().collect();
114    labels.insert("runnerd.profile".to_owned(), spec.profile.clone());
115    labels.insert("runnerd.isolation".to_owned(), isolation.name().to_owned());
116    ContainerCreateBody {
117        image: Some(spec.image.clone()),
118        env: Some(spec.env.clone()),
119        entrypoint: spec.entrypoint.clone(),
120        labels: Some(labels),
121        host_config: Some(host_config),
122        ..Default::default()
123    }
124}
125
126/// A container `list` found.
127#[derive(Debug, Clone, PartialEq, Eq)]
128pub struct Container {
129    /// The Docker container ID.
130    pub id: String,
131    /// Without Docker's leading `/`.
132    pub name: String,
133    /// Whether it runs.
134    pub running: bool,
135    /// Its labels.
136    pub labels: HashMap<String, String>,
137}
138
139/// Whether a Docker error says the container, volume or image does not exist.
140pub fn is_not_found(error: &Error) -> bool {
141    matches!(
142        error,
143        Error::DockerResponseServerError {
144            status_code: 404,
145            ..
146        }
147    )
148}
149
150/// The Docker Engine calls the scheduler and the caches make. `Docker` is the real one; the engine's tests use a
151/// fake.
152pub trait DockerApi: Send + Sync {
153    /// Creates and starts a runner container; returns its ID.
154    fn start(
155        &self,
156        spec: &RunnerSpec,
157        isolation: &dyn Isolation,
158    ) -> impl Future<Output = Result<String, Error>> + Send;
159    /// A container's state and isolation.
160    fn inspect(&self, container: &str) -> impl Future<Output = Result<RunnerState, Error>> + Send;
161    /// Removes a runner container, stopping it first if it runs.
162    fn remove(&self, container: &str) -> impl Future<Output = Result<(), Error>> + Send;
163    /// Creates and starts a container that is not a runner, such as the mirror; returns its ID.
164    fn run(
165        &self,
166        name: &str,
167        body: ContainerCreateBody,
168    ) -> impl Future<Output = Result<String, Error>> + Send;
169    /// Every container, running or not, that has the label.
170    fn list(&self, label: &str) -> impl Future<Output = Result<Vec<Container>, Error>> + Send;
171    /// Waits for a container to exit; returns its exit code.
172    fn wait(&self, container: &str) -> impl Future<Output = Result<i64, Error>> + Send;
173    /// A container's output so far, stdout and stderr interleaved.
174    fn logs(&self, container: &str) -> impl Future<Output = Result<String, Error>> + Send;
175    /// A container's output since `since` (seconds since the epoch), each line prefixed with Docker's RFC 3339
176    /// timestamp and a space.
177    fn logs_since(
178        &self,
179        container: &str,
180        since: u64,
181    ) -> impl Future<Output = Result<String, Error>> + Send;
182    /// Runs a command as root in a running container; its stdout streams until it exits or the container stops.
183    /// Its stderr is dropped.
184    fn exec(
185        &self,
186        container: &str,
187        command: Vec<String>,
188    ) -> impl Future<Output = Result<Output, Error>> + Send;
189    /// Whether Docker has an image.
190    fn image_present(&self, image: &str) -> impl Future<Output = Result<bool, Error>> + Send;
191    /// Pulls an image, logging its progress every few seconds.
192    fn pull(&self, image: &str) -> impl Future<Output = Result<(), Error>> + Send;
193    /// Creates a bridge network unless it exists.
194    fn ensure_network(&self, name: &str) -> impl Future<Output = Result<(), Error>> + Send;
195    /// Creates a volume unless it exists.
196    fn ensure_volume(&self, name: &str) -> impl Future<Output = Result<(), Error>> + Send;
197    /// Removes a volume; a volume that does not exist is not an error.
198    fn remove_volume(&self, name: &str) -> impl Future<Output = Result<(), Error>> + Send;
199    /// The disk size of every volume Docker knows, by name. Slow: Docker measures each volume.
200    fn volume_sizes(&self) -> impl Future<Output = Result<BTreeMap<String, u64>, Error>> + Send;
201}
202
203/// The local Docker Engine.
204#[derive(Debug)]
205pub struct Docker {
206    api: bollard::Docker,
207}
208
209impl Docker {
210    /// Connects to the Docker Engine at `DOCKER_HOST`, or its local socket. Only `unix://` hosts are supported.
211    pub fn connect() -> Result<Docker, Error> {
212        connect_to(std::env::var("DOCKER_HOST").ok().as_deref())
213    }
214
215    /// The Engine's system information.
216    pub async fn info(&self) -> Result<SystemInfo, Error> {
217        self.api.info().await
218    }
219
220    /// An image's labels; `None` when Docker does not have it.
221    pub async fn image_labels(
222        &self,
223        image: &str,
224    ) -> Result<Option<HashMap<String, String>>, Error> {
225        match self.api.inspect_image(image).await {
226            Ok(inspect) => Ok(Some(
227                inspect
228                    .config
229                    .and_then(|config| config.labels)
230                    .unwrap_or_default(),
231            )),
232            Err(error) if is_not_found(&error) => Ok(None),
233            Err(error) => Err(error),
234        }
235    }
236
237    /// Asks the image's registry for its manifest, without pulling it, and returns the platforms it is built for,
238    /// such as `linux/amd64`. An error means `pull` could not get it.
239    pub async fn registry_platforms(&self, image: &str) -> Result<Vec<String>, Error> {
240        let inspect = self.api.inspect_registry_image(image, None).await?;
241        Ok(inspect
242            .platforms
243            .into_iter()
244            .filter_map(|platform| Some(format!("{}/{}", platform.os?, platform.architecture?)))
245            // Attestation manifests in an index carry the platform unknown/unknown.
246            .filter(|platform| platform != "unknown/unknown")
247            .collect())
248    }
249}
250
251impl DockerApi for Docker {
252    async fn start(&self, spec: &RunnerSpec, isolation: &dyn Isolation) -> Result<String, Error> {
253        let options = CreateContainerOptionsBuilder::new()
254            .name(&spec.name)
255            .build();
256        let id = self
257            .api
258            .create_container(Some(options), create_body(spec, isolation))
259            .await?
260            .id;
261        let started = async {
262            if let Some(jit_config) = &spec.jit_config {
263                let options = UploadToContainerOptionsBuilder::new().path("/").build();
264                let tar = jit_config_tar(jit_config)?;
265                // As a stream: bollard traces a full body's bytes, but a stream only as "(stream)".
266                let tar = futures_util::stream::once(std::future::ready(tar.into()));
267                self.api
268                    .upload_to_container(&id, Some(options), bollard::body_stream(tar))
269                    .await?;
270            }
271            self.api.start_container(&id, None).await
272        };
273        if let Err(error) = started.await {
274            // Best effort: the start error is the one worth reporting.
275            let _ = self.remove(&id).await;
276            return Err(error);
277        }
278        Ok(id)
279    }
280
281    async fn inspect(&self, container: &str) -> Result<RunnerState, Error> {
282        let response = self.api.inspect_container(container, None).await?;
283        let state = response.state.unwrap_or_default();
284        let host_config = response.host_config.unwrap_or_default();
285        let address = response
286            .network_settings
287            .and_then(|settings| settings.networks)
288            .and_then(|networks| {
289                networks
290                    .into_values()
291                    .find_map(|network| network.ip_address.filter(|address| !address.is_empty()))
292            });
293        Ok(RunnerState {
294            running: state.running.unwrap_or(false),
295            exit_code: state.exit_code,
296            privileged: host_config.privileged.unwrap_or(false),
297            runtime: host_config.runtime,
298            pid: state
299                .pid
300                .and_then(|pid| u32::try_from(pid).ok())
301                .filter(|pid| *pid > 0),
302            address,
303            started_at: state.started_at.as_deref().and_then(crate::time::parse_ms),
304            memory: host_config
305                .memory
306                .and_then(|memory| u64::try_from(memory).ok())
307                .filter(|memory| *memory > 0),
308            cpus: host_config
309                .nano_cpus
310                .and_then(|nano| u64::try_from(nano / 1_000_000).ok())
311                .filter(|cpus| *cpus > 0),
312        })
313    }
314
315    async fn remove(&self, container: &str) -> Result<(), Error> {
316        let options = RemoveContainerOptionsBuilder::new().force(true).build();
317        self.api.remove_container(container, Some(options)).await
318    }
319
320    async fn run(&self, name: &str, body: ContainerCreateBody) -> Result<String, Error> {
321        let options = CreateContainerOptionsBuilder::new().name(name).build();
322        let id = self.api.create_container(Some(options), body).await?.id;
323        if let Err(error) = self.api.start_container(&id, None).await {
324            let _ = self.remove(&id).await;
325            return Err(error);
326        }
327        Ok(id)
328    }
329
330    async fn list(&self, label: &str) -> Result<Vec<Container>, Error> {
331        let options = ListContainersOptionsBuilder::new()
332            .all(true)
333            .filters(&HashMap::from([("label", vec![label])]))
334            .build();
335        let containers = self.api.list_containers(Some(options)).await?;
336        Ok(containers
337            .into_iter()
338            .map(|container| Container {
339                id: container.id.unwrap_or_default(),
340                name: container
341                    .names
342                    .and_then(|names| names.into_iter().next())
343                    .unwrap_or_default()
344                    .trim_start_matches('/')
345                    .to_owned(),
346                running: container.state
347                    == Some(bollard::models::ContainerSummaryStateEnum::RUNNING),
348                labels: container.labels.unwrap_or_default(),
349            })
350            .collect())
351    }
352
353    async fn wait(&self, container: &str) -> Result<i64, Error> {
354        let options = WaitContainerOptionsBuilder::new()
355            .condition("not-running")
356            .build();
357        let mut responses = self.api.wait_container(container, Some(options));
358        match responses.next().await {
359            Some(Ok(response)) => Ok(response.status_code),
360            // Bollard reports a non-zero exit as an error.
361            Some(Err(Error::DockerContainerWaitError { code, .. })) => Ok(code),
362            Some(Err(error)) => Err(error),
363            None => Ok(0),
364        }
365    }
366
367    async fn logs(&self, container: &str) -> Result<String, Error> {
368        let options = LogsOptionsBuilder::new().stdout(true).stderr(true).build();
369        let chunks: Vec<_> = self
370            .api
371            .logs(container, Some(options))
372            .try_collect()
373            .await?;
374        Ok(chunks.iter().map(ToString::to_string).collect())
375    }
376
377    async fn logs_since(&self, container: &str, since: u64) -> Result<String, Error> {
378        let options = LogsOptionsBuilder::new()
379            .stdout(true)
380            .stderr(true)
381            .timestamps(true)
382            .since(i32::try_from(since).unwrap_or(i32::MAX))
383            .build();
384        let chunks: Vec<_> = self
385            .api
386            .logs(container, Some(options))
387            .try_collect()
388            .await?;
389        Ok(chunks.iter().map(ToString::to_string).collect())
390    }
391
392    async fn exec(&self, container: &str, command: Vec<String>) -> Result<Output, Error> {
393        let options = CreateExecOptions {
394            cmd: Some(command),
395            user: Some("root".to_owned()),
396            attach_stdout: Some(true),
397            attach_stderr: Some(true),
398            ..Default::default()
399        };
400        let exec = self.api.create_exec(container, options).await?;
401        match self.api.start_exec(&exec.id, None).await? {
402            StartExecResults::Attached { output, .. } => Ok(output
403                .try_filter_map(|output| {
404                    std::future::ready(Ok(match output {
405                        bollard::container::LogOutput::StdOut { message } => Some(message.to_vec()),
406                        _ => None,
407                    }))
408                })
409                .boxed()),
410            StartExecResults::Detached => Ok(futures_util::stream::empty().boxed()),
411        }
412    }
413
414    async fn image_present(&self, image: &str) -> Result<bool, Error> {
415        Ok(self.image_labels(image).await?.is_some())
416    }
417
418    async fn pull(&self, image: &str) -> Result<(), Error> {
419        let options = CreateImageOptionsBuilder::new().from_image(image).build();
420        let mut stream = self.api.create_image(Some(options), None, None);
421        let mut progress = PullProgress::default();
422        let mut logged = Instant::now();
423        info!("pulling {image}");
424        while let Some(update) = stream.try_next().await? {
425            progress.update(&update);
426            if logged.elapsed() >= PULL_LOG_INTERVAL {
427                logged = Instant::now();
428                info!("pulling {image}: {progress}");
429            }
430        }
431        info!("pulled {image}");
432        Ok(())
433    }
434
435    async fn ensure_network(&self, name: &str) -> Result<(), Error> {
436        match self.api.inspect_network(name, None).await {
437            Err(error) if is_not_found(&error) => {
438                let request = NetworkCreateRequest {
439                    name: name.to_owned(),
440                    driver: Some("bridge".to_owned()),
441                    ..Default::default()
442                };
443                self.api.create_network(request).await.map(|_| ())
444            }
445            other => other.map(|_| ()),
446        }
447    }
448
449    async fn ensure_volume(&self, name: &str) -> Result<(), Error> {
450        // Creating a volume that exists is a no-op in Docker.
451        let request = VolumeCreateRequest {
452            name: Some(name.to_owned()),
453            ..Default::default()
454        };
455        self.api.create_volume(request).await.map(|_| ())
456    }
457
458    async fn remove_volume(&self, name: &str) -> Result<(), Error> {
459        let options = RemoveVolumeOptionsBuilder::new().build();
460        match self.api.remove_volume(name, Some(options)).await {
461            Err(error) if is_not_found(&error) => Ok(()),
462            other => other,
463        }
464    }
465
466    async fn volume_sizes(&self) -> Result<BTreeMap<String, u64>, Error> {
467        // Without a type filter: bollard cannot encode one in the query.
468        let options = DataUsageOptionsBuilder::new().verbose(true).build();
469        let usage = self.api.df(Some(options)).await?;
470        // API 1.53 leaves the items untyped; they are volumes as `GET /volumes` describes them.
471        Ok(usage
472            .volume_usage
473            .and_then(|volumes| volumes.items)
474            .unwrap_or_default()
475            .into_iter()
476            .filter_map(|volume| {
477                let name = volume.get("Name")?.as_str()?.to_owned();
478                let size = volume.get("UsageData")?.get("Size")?.as_u64()?;
479                Some((name, size))
480            })
481            .collect())
482    }
483}
484
485/// How often `pull` logs its progress.
486const PULL_LOG_INTERVAL: Duration = Duration::from_secs(10);
487
488/// A pull's layers so far, from the statuses Docker streams per layer.
489#[derive(Debug, Default)]
490struct PullProgress {
491    layers: BTreeMap<String, Layer>,
492}
493
494#[derive(Debug, Default)]
495struct Layer {
496    downloaded: u64,
497    size: u64,
498    done: bool,
499}
500
501impl PullProgress {
502    fn update(&mut self, update: &CreateImageInfo) {
503        let (Some(id), Some(status)) = (&update.id, &update.status) else {
504            return;
505        };
506        let bytes = |value: Option<i64>| value.and_then(|value| u64::try_from(value).ok());
507        let detail = update.progress_detail.as_ref();
508        match status.as_str() {
509            "Pulling fs layer" | "Waiting" => {
510                self.layers.entry(id.clone()).or_default();
511            }
512            "Downloading" => {
513                let layer = self.layers.entry(id.clone()).or_default();
514                layer.downloaded =
515                    bytes(detail.and_then(|detail| detail.current)).unwrap_or(layer.downloaded);
516                layer.size = bytes(detail.and_then(|detail| detail.total)).unwrap_or(layer.size);
517            }
518            "Download complete" => {
519                if let Some(layer) = self.layers.get_mut(id) {
520                    layer.downloaded = layer.size;
521                }
522            }
523            "Pull complete" | "Already exists" => {
524                let layer = self.layers.entry(id.clone()).or_default();
525                layer.downloaded = layer.size;
526                layer.done = true;
527            }
528            _ => {}
529        }
530    }
531}
532
533impl fmt::Display for PullProgress {
534    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
535        let done = self.layers.values().filter(|layer| layer.done).count();
536        let downloaded: u64 = self.layers.values().map(|layer| layer.downloaded).sum();
537        let size: u64 = self.layers.values().map(|layer| layer.size).sum();
538        write!(
539            f,
540            "{done} of {} layers done, {} of {} downloaded",
541            self.layers.len(),
542            ByteSize(downloaded),
543            ByteSize(size)
544        )
545    }
546}
547
548fn connect_to(docker_host: Option<&str>) -> Result<Docker, Error> {
549    let api = match docker_host.filter(|host| !host.is_empty()) {
550        None => bollard::Docker::connect_with_socket_defaults()?,
551        Some(host) if host.starts_with("unix://") => {
552            bollard::Docker::connect_with_socket(host, 120, bollard::API_DEFAULT_VERSION)?
553        }
554        // Bollard would quietly fall back to the local socket: a different Engine than the one asked for.
555        Some(host) => {
556            return Err(Error::UnsupportedURISchemeError {
557                uri: host.to_owned(),
558            });
559        }
560    };
561    Ok(Docker { api })
562}
563
564/// A tar archive, to extract at `/`, holding the config as a root-only file at `JIT_CONFIG_PATH`.
565fn jit_config_tar(jit_config: &JitConfig) -> std::io::Result<Vec<u8>> {
566    let mut tar = tar::Builder::new(Vec::new());
567    let path = JIT_CONFIG_PATH.trim_start_matches('/');
568    let mut dir = tar::Header::new_gnu();
569    dir.set_entry_type(tar::EntryType::Directory);
570    dir.set_mode(0o700);
571    dir.set_size(0);
572    tar.append_data(
573        &mut dir,
574        path.rsplit_once('/').map_or("", |(dir, _)| dir),
575        std::io::empty(),
576    )?;
577    let mut file = tar::Header::new_gnu();
578    file.set_mode(0o400);
579    file.set_size(jit_config.0.len() as u64);
580    tar.append_data(&mut file, path, jit_config.0.as_bytes())?;
581    tar.into_inner()
582}
583
584#[cfg(test)]
585mod tests;
runnerd_core/docker.rs

runnerd_core/
docker.rs
