ci_engine/service/
docker.rs

1// SPDX-License-Identifier: Apache-2.0
2//! Docker CLI implementation of the treadle service boundary.
3
4use std::{
5    net::{SocketAddr, TcpStream},
6    time::{Duration, Instant},
7};
8
9use ci_config::Service;
10
11use super::{CommandRunner, RealCommandRunner, RunningServices, ServiceError, ServiceProvider};
12
13const DEFAULT_HEALTH_TIMEOUT_SECS: u64 = 60;
14const READINESS_POLL: Duration = Duration::from_millis(500);
15const TCP_PROBE_TIMEOUT: Duration = Duration::from_millis(500);
16
17/// Docker-CLI provider. Local CLI does not select this automatically.
18pub struct DockerProvider {
19    job: String,
20    runner: Box<dyn CommandRunner>,
21    health_timeout_secs: u64,
22    skip_readiness: bool,
23}
24
25impl std::fmt::Debug for DockerProvider {
26    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
27        formatter
28            .debug_struct("DockerProvider")
29            .field("job", &self.job)
30            .field("health_timeout_secs", &self.health_timeout_secs)
31            .field("skip_readiness", &self.skip_readiness)
32            .finish_non_exhaustive()
33    }
34}
35
36impl DockerProvider {
37    /// Construct a Docker provider for one job namespace.
38    #[must_use]
39    pub fn new(job: impl Into<String>) -> Self {
40        Self {
41            job: job.into(),
42            runner: Box::new(RealCommandRunner),
43            health_timeout_secs: DEFAULT_HEALTH_TIMEOUT_SECS,
44            skip_readiness: false,
45        }
46    }
47
48    /// Inject the external-command boundary and skip real-clock readiness.
49    #[must_use]
50    pub fn with_runner(job: impl Into<String>, runner: Box<dyn CommandRunner>) -> Self {
51        Self {
52            job: job.into(),
53            runner,
54            health_timeout_secs: DEFAULT_HEALTH_TIMEOUT_SECS,
55            skip_readiness: true,
56        }
57    }
58
59    /// Override the per-service readiness timeout.
60    #[must_use]
61    pub fn health_timeout_secs(mut self, seconds: u64) -> Self {
62        self.health_timeout_secs = seconds;
63        self
64    }
65
66    fn container_name(&self, service: &Service) -> String {
67        format!("hci-{}-{}", self.job, service.name)
68    }
69
70    fn run_args(&self, service: &Service) -> Vec<String> {
71        let mut args = vec![
72            "run".to_string(),
73            "-d".to_string(),
74            "--name".to_string(),
75            self.container_name(service),
76        ];
77        for port in &service.ports {
78            args.extend(["-p".to_string(), format!("{port}:{port}")]);
79        }
80        for (key, value) in &service.env {
81            args.extend(["-e".to_string(), format!("{key}={value}")]);
82        }
83        args.push(service.image.clone());
84        args
85    }
86
87    fn await_ready(&self, service: &Service) -> Result<(), ServiceError> {
88        if self.skip_readiness {
89            return Ok(());
90        }
91        let deadline = Instant::now() + Duration::from_secs(self.health_timeout_secs);
92        loop {
93            if self.probe_ready(service) {
94                return Ok(());
95            }
96            if Instant::now() >= deadline {
97                return Err(ServiceError::NotReady {
98                    name: service.name.clone(),
99                    image: service.image.clone(),
100                    timeout_secs: self.health_timeout_secs,
101                });
102            }
103            std::thread::sleep(READINESS_POLL);
104        }
105    }
106
107    fn probe_ready(&self, service: &Service) -> bool {
108        match &service.ready_cmd {
109            Some(command) if !command.is_empty() => {
110                let mut args = vec!["exec".to_string(), self.container_name(service)];
111                args.extend(command.iter().cloned());
112                self.runner
113                    .run("docker", &args)
114                    .map(|outcome| outcome.success)
115                    .unwrap_or(false)
116            }
117            _ => service.ports.first().is_none_or(|port| tcp_probe(*port)),
118        }
119    }
120
121    fn teardown(&self, names: &[String]) {
122        for name in names {
123            let _ = self.runner.run(
124                "docker",
125                &["rm".to_string(), "-f".to_string(), name.clone()],
126            );
127        }
128    }
129}
130
131fn tcp_probe(port: u16) -> bool {
132    let address = SocketAddr::from(([127, 0, 0, 1], port));
133    TcpStream::connect_timeout(&address, TCP_PROBE_TIMEOUT).is_ok()
134}
135
136impl ServiceProvider for DockerProvider {
137    fn up(&self, services: &[Service]) -> Result<RunningServices, ServiceError> {
138        let mut started = Vec::new();
139        for service in services {
140            let name = self.container_name(service);
141            let outcome = self.runner.run("docker", &self.run_args(service))?;
142            if !outcome.success {
143                self.teardown(&started);
144                return Err(ServiceError::Provision(format!(
145                    "`docker run` for service {:?} failed: {}",
146                    service.name,
147                    outcome.output.trim()
148                )));
149            }
150            started.push(name);
151            if let Err(error) = self.await_ready(service) {
152                self.teardown(&started);
153                return Err(error);
154            }
155        }
156        Ok(RunningServices { handles: started })
157    }
158
159    fn down(&self, running: RunningServices) -> Result<(), ServiceError> {
160        self.teardown(&running.handles);
161        Ok(())
162    }
163}
ci_engine/service/docker.rs

ci_engine/service/
docker.rs
