runnerd_core/
docker.rs

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

runnerd_core/
docker.rs
