olai_stack_runtime/
docker.rs

1//! [`DockerCli`]: the [`ComposeEngine`] that runs the `docker` CLI.
2
3use std::path::{Path, PathBuf};
4use std::process::Stdio;
5
6use async_trait::async_trait;
7use futures::StreamExt;
8use tokio::io::{AsyncBufReadExt, BufReader};
9use tokio::process::{Child, Command};
10
11use crate::engine::{ComposeEngine, DownOptions, EngineStream, LogOptions, Progress, UpOptions};
12use crate::error::{Error, Result};
13use crate::events::{EngineEvent, EventFilter, PROJECT_LABEL, parse_event};
14use crate::logs::{LogLine, parse_log_line};
15use crate::status::{ProjectStatus, ServiceStatus, parse_ls, parse_ps};
16
17/// Runs the `docker` CLI (with the Compose plugin). It honours `DOCKER_HOST` and the current
18/// Docker context like the CLI does; [`with_context`](Self::with_context) pins one.
19#[derive(Clone, Debug)]
20pub struct DockerCli {
21    program: PathBuf,
22    context: Option<String>,
23}
24
25impl Default for DockerCli {
26    fn default() -> Self {
27        DockerCli {
28            program: PathBuf::from("docker"),
29            context: None,
30        }
31    }
32}
33
34/// What the CLI prints when it can't reach a daemon (Docker CLI and Compose wordings).
35const UNREACHABLE: &[&str] = &[
36    "Cannot connect to the Docker daemon",
37    "failed to connect to the docker API",
38    "error during connect",
39    "Is the docker daemon running",
40];
41
42impl DockerCli {
43    /// The `docker` on `PATH`, under the current context.
44    pub fn new() -> Self {
45        Self::default()
46    }
47
48    /// Run `program` instead of `docker` (e.g. an absolute path).
49    pub fn with_program(mut self, program: impl Into<PathBuf>) -> Self {
50        self.program = program.into();
51        self
52    }
53
54    /// Talk to the daemon of Docker context `context` (`docker --context <context>`).
55    pub fn with_context(mut self, context: impl Into<String>) -> Self {
56        self.context = Some(context.into());
57        self
58    }
59
60    fn command(&self, args: &[&str]) -> Command {
61        let mut cmd = Command::new(&self.program);
62        if let Some(context) = &self.context {
63            cmd.args(["--context", context]);
64        }
65        cmd.args(args).stdin(Stdio::null()).kill_on_drop(true);
66        cmd
67    }
68
69    /// Run `args` to completion, capturing stdout; a failure becomes an [`Error`].
70    async fn output(&self, args: &[&str]) -> Result<String> {
71        let line = args.join(" ");
72        let out = self
73            .command(args)
74            .output()
75            .await
76            .map_err(|e| spawn_error(&line, e))?;
77        if !out.status.success() {
78            return Err(failure(line, out.status, &out.stderr));
79        }
80        Ok(String::from_utf8_lossy(&out.stdout).into_owned())
81    }
82
83    /// Run a compose lifecycle command in `dir`, with `env` added and progress as asked.
84    async fn lifecycle(
85        &self,
86        dir: &Path,
87        args: &[&str],
88        env: &std::collections::BTreeMap<String, String>,
89        progress: Progress,
90    ) -> Result<()> {
91        let line = args.join(" ");
92        let mut cmd = self.command(args);
93        cmd.current_dir(dir).envs(env);
94        match progress {
95            Progress::Inherit => {
96                cmd.stdout(Stdio::inherit()).stderr(Stdio::inherit());
97                let status = cmd.status().await.map_err(|e| spawn_error(&line, e))?;
98                if !status.success() {
99                    return Err(failure(line, status, &[]));
100                }
101            }
102            Progress::Quiet => {
103                let out = cmd.output().await.map_err(|e| spawn_error(&line, e))?;
104                if !out.status.success() {
105                    return Err(failure(line, out.status, &out.stderr));
106                }
107            }
108        }
109        Ok(())
110    }
111
112    /// Spawn `args` with stdout piped, as a stream of its lines.
113    fn lines(&self, args: &[&str]) -> Result<EngineStream<String>> {
114        let line = args.join(" ");
115        let mut child = self
116            .command(args)
117            .stdout(Stdio::piped())
118            .stderr(Stdio::null())
119            .spawn()
120            .map_err(|e| spawn_error(&line, e))?;
121        let stdout = child.stdout.take().expect("stdout is piped");
122        let lines = BufReader::new(stdout).lines();
123        // The child rides along in the stream state, so dropping the stream kills it.
124        let stream = futures::stream::unfold(
125            (lines, child, line),
126            |(mut lines, child, line): (_, Child, String)| async move {
127                match lines.next_line().await {
128                    Ok(Some(l)) => Some((Ok(l), (lines, child, line))),
129                    Ok(None) => None,
130                    Err(source) => Some((
131                        Err(Error::Io {
132                            command: line.clone(),
133                            source,
134                        }),
135                        (lines, child, line),
136                    )),
137                }
138            },
139        );
140        Ok(stream.boxed())
141    }
142}
143
144fn spawn_error(command: &str, source: std::io::Error) -> Error {
145    if source.kind() == std::io::ErrorKind::NotFound {
146        Error::Unavailable {
147            reason: format!("the `docker` command was not found ({source})"),
148        }
149    } else {
150        Error::Io {
151            command: command.to_owned(),
152            source,
153        }
154    }
155}
156
157fn failure(command: String, status: std::process::ExitStatus, stderr: &[u8]) -> Error {
158    let stderr = String::from_utf8_lossy(stderr).trim().to_owned();
159    if UNREACHABLE.iter().any(|needle| stderr.contains(needle)) {
160        return Error::Unavailable { reason: stderr };
161    }
162    Error::Failed {
163        command,
164        status,
165        stderr,
166    }
167}
168
169#[async_trait]
170impl ComposeEngine for DockerCli {
171    async fn up(&self, dir: &Path, opts: &UpOptions) -> Result<()> {
172        let timeout = opts.wait_timeout.map(|t| t.as_secs().max(1).to_string());
173        let mut args = vec!["compose", "up", "-d"];
174        if opts.wait {
175            args.push("--wait");
176            if let Some(timeout) = &timeout {
177                args.extend(["--wait-timeout", timeout]);
178            }
179        }
180        self.lifecycle(dir, &args, &opts.env, opts.progress).await
181    }
182
183    async fn down(&self, dir: &Path, opts: &DownOptions) -> Result<()> {
184        let mut args = vec!["compose", "down"];
185        if opts.volumes {
186            args.push("--volumes");
187        }
188        self.lifecycle(dir, &args, &opts.env, opts.progress).await
189    }
190
191    async fn ls(&self) -> Result<Vec<ProjectStatus>> {
192        let args = ["compose", "ls", "--all", "--format", "json"];
193        let out = self.output(&args).await?;
194        parse_ls(&out).map_err(|source| Error::Parse {
195            command: args.join(" "),
196            source,
197        })
198    }
199
200    async fn ps(&self, project: &str) -> Result<Vec<ServiceStatus>> {
201        let args = ["compose", "-p", project, "ps", "--all", "--format", "json"];
202        let out = self.output(&args).await?;
203        parse_ps(&out).map_err(|source| Error::Parse {
204            command: args.join(" "),
205            source,
206        })
207    }
208
209    async fn logs(&self, project: &str, opts: &LogOptions) -> Result<EngineStream<LogLine>> {
210        let tail = opts.tail.map(|t| t.to_string());
211        let mut args = vec![
212            "compose",
213            "-p",
214            project,
215            "logs",
216            "--timestamps",
217            "--no-color",
218        ];
219        if opts.follow {
220            args.push("--follow");
221        }
222        if let Some(tail) = &tail {
223            args.extend(["--tail", tail]);
224        }
225        args.extend(opts.services.iter().map(String::as_str));
226        Ok(self
227            .lines(&args)?
228            .map(|line| line.map(|l| parse_log_line(&l)))
229            .boxed())
230    }
231
232    async fn events(&self, filter: &EventFilter) -> Result<EngineStream<EngineEvent>> {
233        let label = match filter {
234            EventFilter::Project(name) => format!("label={PROJECT_LABEL}={name}"),
235            // Docker filters on a label's exact value only; a prefix is matched below.
236            EventFilter::Prefix(_) | EventFilter::All => format!("label={PROJECT_LABEL}"),
237        };
238        let args = [
239            "events",
240            "--format",
241            "{{json .}}",
242            "--filter",
243            "type=container",
244            "--filter",
245            &label,
246        ];
247        let command = args.join(" ");
248        let filter = filter.clone();
249        Ok(self
250            .lines(&args)?
251            .filter_map(move |line| {
252                let event = line.and_then(|l| {
253                    parse_event(&l).map_err(|source| Error::Parse {
254                        command: command.clone(),
255                        source,
256                    })
257                });
258                let keep = match event {
259                    Ok(Some(e)) if filter.matches(&e) => Some(Ok(e)),
260                    Ok(_) => None,
261                    Err(e) => Some(Err(e)),
262                };
263                futures::future::ready(keep)
264            })
265            .boxed())
266    }
267
268    async fn network_rm(&self, name: &str) -> Result<bool> {
269        match self.output(&["network", "rm", name]).await {
270            Ok(_) => Ok(true),
271            Err(Error::Failed { .. }) => Ok(false),
272            Err(e) => Err(e),
273        }
274    }
275}
olai_stack_runtime/docker.rs

olai_stack_runtime/
docker.rs
