Skip to content
Sign up

flagon-io/g1t

public
0
419 linesCodeBlameRaw
1//! Docker Engines of the runner's own, for `nested-container` isolation
2//! (sandbox.rs):
3//!
4//! - **The runner's engine.** In the runner's image under Sysbox
5//! (`G1T_RUNNER_SANDBOX=nested`), the runner starts `dockerd` inside its
6//! own container and runs every job's container on it. It is a child of
7//! the runner, which `docker-init` (tini) supervises as PID 1, and is
8//! started again before a job if it stopped. Started the way g1t's own
9//! sandboxes start a job's engine (crate::docker::engine): the cgroup
10//! controllers handed down first, as moby's `hack/dind` does, and plain
11//! copies when the disk takes no overlays. Unlike there, Sysbox lets it
12//! keep its packet filter, so each job gets a bridge network of its own.
13//! - **A job's engine.** A job that uses Docker (`services:`, a Docker
14//! action, `docker` in a step) gets an engine of its own, started the
15//! first time the job's Docker socket is used: a `docker:dind` container
16//! beside the job's, in the job's network namespace, so what it
17//! publishes is on the job's `localhost`. The job sees only a socket
18//! that the runner passes on. The engine is privileged within the
19//! runner's container, never on the machine (Sysbox maps its root to an
20//! unprivileged user there), and is removed with its data after the job.
21
22use std::io::{self, Read, Write};
23use std::os::unix::net::{UnixListener, UnixStream};
24use std::path::{Path, PathBuf};
25use std::process::{Child, Command, Stdio};
26use std::sync::atomic::{AtomicBool, Ordering};
27use std::sync::{Arc, Mutex, OnceLock};
28use std::time::{Duration, Instant};
29
30use anyhow::{Context, Result, bail};
31use serde_json::{Value, json};
32
33use super::log;
34
35/// Everything of the runner's engine that is not its data.
36pub const DIR: &str = "/run/g1t-runner";
37/// The runner's engine's socket: not `/var/run/docker.sock`, so a socket
38/// there is never the runner's own (sandbox.rs looks for one).
39pub const SOCKET: &str = "/run/g1t-runner/docker.sock";
40/// Where a job's sockets are kept, a folder each.
41pub const JOBS_DIR: &str = "/run/g1t-runner/jobs";
42/// Where the job sees its Docker socket.
43pub const JOB_SOCKET_DIR: &str = "/run/g1t-docker";
44/// The Docker CLI in the runner's image, lent to a job read-only.
45pub const DOCKER_CLI: &str = "/usr/local/bin/docker";
46/// The image a job's engine runs.
47pub const DEFAULT_JOB_ENGINE_IMAGE: &str = "docker:28-dind";
48
49const LOG: &str = "/run/g1t-runner/dockerd.log";
50/// From moby's `hack/dind`, as crate::docker::engine does it.
51const CGROUP_NESTING: &str = "if [ -f /sys/fs/cgroup/cgroup.controllers ] && [ -z \"$(cat /sys/fs/cgroup/cgroup.subtree_control)\" ]; then \
52 mkdir -p /sys/fs/cgroup/init && xargs -rn1 < /sys/fs/cgroup/cgroup.procs > /sys/fs/cgroup/init/cgroup.procs 2>/dev/null; \
53 sed -e 's/ / +/g' -e 's/^/+/' < /sys/fs/cgroup/cgroup.controllers > /sys/fs/cgroup/cgroup.subtree_control; fi";
54
55/// The engine the runner's `docker` commands use: its own, once started.
56static HOST: OnceLock<String> = OnceLock::new();
57/// The runner's engine, while it runs.
58static ENGINE: Mutex<Option<Child>> = Mutex::new(None);
59
60/// Whether the runner uses an engine of its own.
61pub fn in_use() -> bool {
62 HOST.get().is_some()
63}
64
65/// `docker`, pointed at the runner's own engine when it has one.
66pub fn docker() -> Command {
67 let mut command = Command::new("docker");
68 if let Some(host) = HOST.get() {
69 command.env("DOCKER_HOST", host);
70 }
71 command
72}
73
74/// The runner's engine's `daemon.json`. `copies`: the disk takes no
75/// overlays. Its own address ranges, away from Docker's defaults, which
76/// the machine's engine (and so the runner's own address) already uses.
77pub fn daemon_config(copies: bool) -> Value {
78 let mut config = json!({
79 "hosts": [format!("unix://{SOCKET}")],
80 "pidfile": format!("{DIR}/dockerd.pid"),
81 "bip": "172.31.254.1/24",
82 "default-address-pools": [{ "base": "172.30.0.0/16", "size": 24 }],
83 "registry-mirrors": [crate::docker::engine::MIRROR],
84 "log-driver": "json-file",
85 "log-opts": { "max-size": "20m", "max-file": "2" },
86 });
87 if copies {
88 config["storage-driver"] = json!("vfs");
89 }
90 config
91}
92
93/// A request to an engine's socket; its status.
94fn ping(socket: &Path) -> Option<u16> {
95 let mut stream = UnixStream::connect(socket).ok()?;
96 stream.set_read_timeout(Some(Duration::from_secs(5))).ok()?;
97 stream.write_all(b"GET /_ping HTTP/1.1\r\nHost: docker\r\nConnection: close\r\n\r\n").ok()?;
98 let mut head = [0u8; 12];
99 stream.read_exact(&mut head).ok()?;
100 std::str::from_utf8(&head[9..12]).ok()?.parse().ok()
101}
102
103fn tail(path: &Path, lines: usize) -> String {
104 let text = std::fs::read_to_string(path).unwrap_or_default();
105 let all: Vec<&str> = text.lines().filter(|l| !l.trim().is_empty()).collect();
106 all[all.len().saturating_sub(lines)..].join("\n")
107}
108
109/// Starts the runner's own engine, or finds it running (the runner
110/// restarted after an update); its version.
111pub fn start() -> Result<String> {
112 let _ = HOST.set(format!("unix://{SOCKET}"));
113 let mut engine = ENGINE.lock().unwrap_or_else(|p| p.into_inner());
114 if let Some(child) = engine.as_mut()
115 && matches!(child.try_wait(), Ok(None))
116 && ping(Path::new(SOCKET)) == Some(200)
117 {
118 return Ok(version());
119 }
120 if engine.is_none() && ping(Path::new(SOCKET)) == Some(200) {
121 return Ok(version());
122 }
123 std::fs::create_dir_all(DIR).with_context(|| format!("could not make {DIR}"))?;
124 let _ = Command::new("sh").args(["-c", CGROUP_NESTING]).status();
125 let mut last = String::new();
126 for copies in [false, true] {
127 std::fs::write(format!("{DIR}/daemon.json"), serde_json::to_vec_pretty(&daemon_config(copies))?)?;
128 let _ = std::fs::remove_file(SOCKET);
129 let out = std::fs::File::create(LOG).with_context(|| format!("could not write {LOG}"))?;
130 let err = out.try_clone()?;
131 let mut child = Command::new("dockerd")
132 .args(["--config-file", &format!("{DIR}/daemon.json")])
133 .env_remove("DOCKER_HOST")
134 .stdin(Stdio::null())
135 .stdout(out)
136 .stderr(err)
137 .spawn()
138 .context("could not run dockerd: the runner's image has it; is this the runner's image?")?;
139 let until = Instant::now() + Duration::from_secs(90);
140 loop {
141 if ping(Path::new(SOCKET)) == Some(200) {
142 *engine = Some(child);
143 let version = version();
144 return Ok(if copies { format!("{version} (plain-copy storage: this disk takes no overlays)") } else { version });
145 }
146 if let Ok(Some(status)) = child.try_wait() {
147 last = format!("dockerd stopped ({status}):\n{}", tail(Path::new(LOG), 15));
148 break;
149 }
150 if Instant::now() >= until {
151 let _ = child.kill();
152 let _ = child.wait();
153 last = format!("dockerd did not answer in 90 s:\n{}", tail(Path::new(LOG), 15));
154 break;
155 }
156 std::thread::sleep(Duration::from_millis(200));
157 }
158 }
159 bail!(
160 "the runner's own Docker Engine could not start. Under Sysbox it needs no more than `docker run --runtime=sysbox-runc`; without Sysbox it cannot start without --privileged, which this runner does not use.\n{last}"
161 )
162}
163
164/// Before a job: the runner's engine is running, started again if not.
165pub fn ensure_running() -> Result<()> {
166 let stopped = {
167 let mut engine = ENGINE.lock().unwrap_or_else(|p| p.into_inner());
168 match engine.as_mut() {
169 Some(child) => !matches!(child.try_wait(), Ok(None)),
170 None => ping(Path::new(SOCKET)) != Some(200),
171 }
172 };
173 if stopped {
174 log("The runner's Docker Engine stopped; starting it again.");
175 start()?;
176 }
177 Ok(())
178}
179
180fn version() -> String {
181 docker()
182 .args(["version", "--format", "{{.Server.Version}}"])
183 .output()
184 .ok()
185 .filter(|o| o.status.success())
186 .map(|o| String::from_utf8_lossy(&o.stdout).trim().to_owned())
187 .filter(|v| !v.is_empty())
188 .unwrap_or_else(|| "?".into())
189}
190
191// --- A job's own engine -------------------------------------------------
192
193/// A job's Docker: a socket now, an engine the first time it is used.
194pub struct JobEngine {
195 /// `JOBS_DIR/<job container>`: `proxy/docker.sock` (the job's) and
196 /// `engine/docker.sock` (its engine's).
197 pub dir: PathBuf,
198 /// The engine's container and its data's volume.
199 pub container: String,
200 pub volume: String,
201 stop: Arc<AtomicBool>,
202}
203
204/// What starting it needs.
205#[derive(Clone)]
206struct EngineSpec {
207 image: String,
208 container: String,
209 volume: String,
210 /// The job's container, whose network the engine joins.
211 job: String,
212 label: String,
213 engine_dir: PathBuf,
214}
215
216/// The `docker run` for a job's engine. Privileged within the runner's
217/// container: Sysbox maps its root to an unprivileged user on the
218/// machine, and gives the runner's container no devices of the machine's.
219fn engine_args(spec: &EngineSpec) -> Vec<String> {
220 let mut args: Vec<String> = ["run", "-d", "--rm", "--name", &spec.container, "--label", &spec.label, "--privileged", "--pull", "missing"]
221 .iter()
222 .map(|s| s.to_string())
223 .collect();
224 args.extend(["--network".into(), format!("container:{}", spec.job)]);
225 // No TLS certificates and no TCP port: only the socket, which the
226 // runner passes on.
227 args.extend(["-e".into(), "DOCKER_TLS_CERTDIR=".into()]);
228 args.extend(["-v".into(), format!("{}:/run/g1t-engine", spec.engine_dir.display())]);
229 args.extend(["-v".into(), format!("{}:/var/lib/docker", spec.volume)]);
230 args.extend([spec.image.clone(), "dockerd".into(), "--host=unix:///run/g1t-engine/docker.sock".into()]);
231 args
232}
233
234fn start_job_engine(spec: &EngineSpec) -> Result<(), String> {
235 let begun = Instant::now();
236 let output = docker().args(engine_args(spec)).output().map_err(|e| format!("could not run docker: {e}"))?;
237 if !output.status.success() {
238 return Err(format!("could not start the job's Docker Engine: {}", String::from_utf8_lossy(&output.stderr).trim()));
239 }
240 let socket = spec.engine_dir.join("docker.sock");
241 while begun.elapsed() < Duration::from_secs(120) {
242 if ping(&socket) == Some(200) {
243 log(&format!("Started the job's own Docker Engine in {:.1}s.", begun.elapsed().as_secs_f64()));
244 return Ok(());
245 }
246 std::thread::sleep(Duration::from_millis(200));
247 }
248 Err("the job's Docker Engine did not answer in 120 s".into())
249}
250
251/// Passes one connection on to the job's engine, starting it first.
252fn serve(mut client: UnixStream, spec: &EngineSpec, started: &Mutex<Option<Result<(), String>>>) {
253 let ready = {
254 let mut started = started.lock().unwrap_or_else(|p| p.into_inner());
255 if started.is_none() {
256 *started = Some(start_job_engine(spec));
257 }
258 started.clone().unwrap_or(Ok(()))
259 };
260 if let Err(problem) = ready {
261 let body = serde_json::to_string(&json!({ "message": problem })).unwrap_or_default();
262 let _ = write!(client, "HTTP/1.1 503 Service Unavailable\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", body.len());
263 return;
264 }
265 let Ok(engine) = UnixStream::connect(spec.engine_dir.join("docker.sock")) else { return };
266 let (Ok(mut client_read), Ok(mut engine_write)) = (client.try_clone(), engine.try_clone()) else { return };
267 let upstream = std::thread::spawn(move || {
268 let _ = io::copy(&mut client_read, &mut engine_write);
269 let _ = engine_write.shutdown(std::net::Shutdown::Write);
270 });
271 let mut engine_read = engine;
272 let _ = io::copy(&mut engine_read, &mut client);
273 let _ = client.shutdown(std::net::Shutdown::Both);
274 let _ = upstream.join();
275}
276
277impl JobEngine {
278 /// Sets a job's Docker socket up, with no engine behind it yet.
279 pub fn prepare(job: &str, runner: &str, image: &str) -> Result<JobEngine> {
280 let dir = Path::new(JOBS_DIR).join(job);
281 let _ = std::fs::remove_dir_all(&dir);
282 let (proxy, engine_dir) = (dir.join("proxy"), dir.join("engine"));
283 std::fs::create_dir_all(&proxy).and_then(|()| std::fs::create_dir_all(&engine_dir)).with_context(|| format!("could not make {}", dir.display()))?;
284 let socket = proxy.join("docker.sock");
285 let listener = UnixListener::bind(&socket).with_context(|| format!("could not listen on {}", socket.display()))?;
286 {
287 use std::os::unix::fs::PermissionsExt;
288 // Whoever the job's image runs as.
289 std::fs::set_permissions(&socket, std::fs::Permissions::from_mode(0o666))?;
290 }
291 listener.set_nonblocking(true)?;
292 let spec = EngineSpec {
293 image: image.to_owned(),
294 container: format!("{job}-docker"),
295 volume: format!("{job}-docker"),
296 job: job.to_owned(),
297 label: format!("sh.g1t.runner={runner}"),
298 engine_dir,
299 };
300 let stop = Arc::new(AtomicBool::new(false));
301 let (stopping, thread_spec) = (stop.clone(), spec.clone());
302 std::thread::spawn(move || {
303 let started: Arc<Mutex<Option<Result<(), String>>>> = Arc::new(Mutex::new(None));
304 while !stopping.load(Ordering::SeqCst) {
305 match listener.accept() {
306 Ok((client, _)) => {
307 let _ = client.set_nonblocking(false);
308 let (spec, started) = (thread_spec.clone(), started.clone());
309 std::thread::spawn(move || serve(client, &spec, &started));
310 }
311 Err(error) if error.kind() == io::ErrorKind::WouldBlock => std::thread::sleep(Duration::from_millis(100)),
312 Err(_) => std::thread::sleep(Duration::from_millis(100)),
313 }
314 }
315 });
316 Ok(JobEngine { dir, container: spec.container, volume: spec.volume, stop })
317 }
318
319 /// What the job's `docker run` adds: its socket's folder, read-only
320 /// (a socket on a read-only mount still connects), the Docker CLI, and
321 /// `DOCKER_HOST`.
322 pub fn job_args(&self) -> Vec<String> {
323 let mut args = vec!["-v".into(), format!("{}:{JOB_SOCKET_DIR}:ro", self.dir.join("proxy").display())];
324 if Path::new(DOCKER_CLI).exists() {
325 args.extend(["-v".into(), format!("{DOCKER_CLI}:/usr/local/bin/docker:ro")]);
326 }
327 args.extend(["-e".into(), format!("DOCKER_HOST=unix://{JOB_SOCKET_DIR}/docker.sock")]);
328 args
329 }
330
331 /// Stops the engine and removes it, its data and its sockets.
332 pub fn remove(&self) {
333 self.stop.store(true, Ordering::SeqCst);
334 let quiet = |args: &[&str]| {
335 let _ = docker().args(args).stdout(Stdio::null()).stderr(Stdio::null()).status();
336 };
337 quiet(&["rm", "-f", "-v", &self.container]);
338 quiet(&["volume", "rm", "-f", &self.volume]);
339 let _ = std::fs::remove_dir_all(&self.dir);
340 }
341}
342
343#[cfg(test)]
344mod tests {
345 use super::*;
346
347 #[test]
348 fn the_runners_engine_keeps_its_own_socket_and_addresses() {
349 let config = daemon_config(false);
350 assert_eq!(config["hosts"][0], "unix:///run/g1t-runner/docker.sock");
351 assert_ne!(SOCKET, "/var/run/docker.sock");
352 assert!(config.get("storage-driver").is_none());
353 assert_eq!(daemon_config(true)["storage-driver"], "vfs");
354 // Not Docker's default range, which the machine's engine has.
355 assert!(!config["bip"].as_str().unwrap().starts_with("172.17."));
356 }
357
358 #[test]
359 fn a_jobs_engine_is_in_the_jobs_network_and_mounts_nothing_of_the_machine() {
360 let spec = EngineSpec {
361 image: DEFAULT_JOB_ENGINE_IMAGE.into(),
362 container: "g1t-job_1-docker".into(),
363 volume: "g1t-job_1-docker".into(),
364 job: "g1t-job_1".into(),
365 label: "sh.g1t.runner=rnr_1".into(),
366 engine_dir: PathBuf::from("/run/g1t-runner/jobs/g1t-job_1/engine"),
367 };
368 let args = engine_args(&spec);
369 let joined = args.join(" ");
370 assert!(joined.contains("--network container:g1t-job_1"));
371 assert!(joined.contains("--rm"));
372 assert!(!joined.contains("docker.sock:/"), "no engine socket is mounted in: {joined}");
373 // Its volumes: the runner's own folder for its socket, and a volume.
374 let mounts: Vec<&String> = args.windows(2).filter(|w| w[0] == "-v").map(|w| &w[1]).collect();
375 assert_eq!(mounts.len(), 2);
376 assert!(mounts[0].starts_with("/run/g1t-runner/jobs/"));
377 assert!(mounts[1].starts_with("g1t-job_1-docker:"));
378 assert!(args.iter().any(|a| a == "DOCKER_TLS_CERTDIR="));
379 }
380
381 /// Against a real Docker Engine, as root, with `docker:28-dind` at
382 /// hand: `G1T_TEST_DOCKER=1 cargo test -- --ignored job_engine`.
383 #[test]
384 #[ignore]
385 fn a_jobs_engine_starts_when_the_job_first_uses_it() {
386 if std::env::var("G1T_TEST_DOCKER").as_deref() != Ok("1") {
387 return;
388 }
389 let job = "g1t-engine-test";
390 let quiet = |args: &[&str]| {
391 let _ = docker().args(args).stdout(Stdio::null()).stderr(Stdio::null()).status();
392 };
393 quiet(&["rm", "-f", job]);
394 let engine = JobEngine::prepare(job, "test", DEFAULT_JOB_ENGINE_IMAGE).unwrap();
395 // A job's container with the job's Docker: the dind image has a CLI.
396 let mut run = docker();
397 run.args(["run", "-d", "--rm", "--name", job, "--cap-drop", "ALL", "--security-opt", "no-new-privileges"]);
398 let mut job_args = engine.job_args();
399 // This machine's CLI may not run in Alpine; the image has its own.
400 if let Some(at) = job_args.iter().position(|a| a.starts_with(DOCKER_CLI)) {
401 job_args.drain(at - 1..=at);
402 }
403 run.args(&job_args).args(["--entrypoint", "sleep", DEFAULT_JOB_ENGINE_IMAGE, "300"]);
404 assert!(run.status().unwrap().success());
405 // Nothing runs behind the socket until it is used.
406 assert!(!docker().args(["inspect", &engine.container]).stdout(Stdio::null()).stderr(Stdio::null()).status().unwrap().success());
407 let version = docker().args(["exec", job, "docker", "version", "--format", "{{.Server.Version}}"]).output().unwrap();
408 let shown = String::from_utf8_lossy(&version.stdout).trim().to_owned();
409 let errors = String::from_utf8_lossy(&version.stderr).trim().to_owned();
410 // The job cannot change its socket's folder.
411 let wrote = docker().args(["exec", job, "touch", "/run/g1t-docker/x"]).stderr(Stdio::null()).status().unwrap();
412 quiet(&["rm", "-f", job]);
413 engine.remove();
414 assert!(version.status.success(), "docker version in the job: {errors}");
415 assert!(shown.starts_with("28."), "{shown}");
416 assert!(!wrote.success());
417 assert!(!engine.dir.exists());
418 }
419}