start_command/
docker_snapshot_safety.rs

1//! Serialize snapshots separately from the execution store and bound disk usage.
2use crate::docker_cleanup::docker_command;
3use crate::execution_control::CommandRunner;
4use crate::isolation::isolation_log::shell_quote;
5use crate::store_lock::LockManager;
6use serde_json::Value;
7use std::path::PathBuf;
8use std::time::{Duration, Instant};
9
10const GIB: u64 = 1024 * 1024 * 1024;
11pub const SNAPSHOT_IMAGE_LABEL: &str = "start-command.snapshot-image";
12
13/// sysexits.h EX_TEMPFAIL: the resume can succeed when retried later.
14pub const EX_TEMPFAIL: i32 = 75;
15
16/// A refused snapshot; `exit_code` is [`EX_TEMPFAIL`] when retrying can help.
17#[derive(Debug, Clone, PartialEq, Eq)]
18pub struct SnapshotFailure {
19    pub message: String,
20    pub exit_code: Option<i32>,
21}
22
23impl SnapshotFailure {
24    fn tempfail(message: String) -> Self {
25        Self {
26            message,
27            exit_code: Some(EX_TEMPFAIL),
28        }
29    }
30}
31
32impl From<String> for SnapshotFailure {
33    fn from(message: String) -> Self {
34        Self {
35            message,
36            exit_code: None,
37        }
38    }
39}
40
41impl From<&str> for SnapshotFailure {
42    fn from(message: &str) -> Self {
43        message.to_string().into()
44    }
45}
46
47/// Seconds as the wait messages print them: "600s", "1.5s".
48pub fn format_seconds(seconds: f64) -> String {
49    format!("{}s", (seconds * 1000.0).round() / 1000.0)
50}
51
52fn default_lock_path() -> PathBuf {
53    std::env::var_os("START_SNAPSHOT_LOCK")
54        .map(PathBuf::from)
55        .unwrap_or_else(|| {
56            if cfg!(unix) {
57                PathBuf::from("/tmp/start-command-docker-snapshot.lock")
58            } else {
59                std::env::temp_dir().join("start-command-docker-snapshot.lock")
60            }
61        })
62}
63
64fn notify_stderr(line: &str) {
65    eprintln!("{}", line);
66}
67
68/// How to take the host-wide snapshot lock (issue #224).
69pub struct SnapshotLockRequest<'a> {
70    pub lock_path: Option<PathBuf>,
71    /// Container about to be snapshotted, published for other waiters.
72    pub holder: Option<&'a str>,
73    /// Wait this long for a busy lock; 0 fails at once.
74    pub wait_seconds: f64,
75    pub notify: &'a dyn Fn(&str),
76}
77
78impl Default for SnapshotLockRequest<'_> {
79    fn default() -> Self {
80        Self {
81            lock_path: None,
82            holder: None,
83            wait_seconds: 0.0,
84            notify: &notify_stderr,
85        }
86    }
87}
88
89pub fn acquire_snapshot_lock() -> Result<LockManager, String> {
90    acquire_snapshot_lock_with(&SnapshotLockRequest::default()).map_err(|e| e.message)
91}
92
93/// Take the snapshot lock, waiting `wait_seconds` for a busy one. The wait is
94/// announced once with the holder's name.
95pub fn acquire_snapshot_lock_with(
96    request: &SnapshotLockRequest,
97) -> Result<LockManager, SnapshotFailure> {
98    let path = request.lock_path.clone().unwrap_or_else(default_lock_path);
99    let mut lock = LockManager::new(path);
100    if let Some(holder) = request.holder {
101        lock = lock.with_metadata("holder", Value::from(holder));
102    }
103    // Same-host LockManager ownership uses PID liveness, not the 60s stale
104    // age used for other hosts. Long commits keep their serialization lock.
105    let wait_ms = (request.wait_seconds.max(0.0) * 1000.0) as u64;
106    if lock.acquire(if wait_ms == 0 {
107        1000
108    } else {
109        wait_ms.min(1000)
110    }) {
111        return Ok(lock);
112    }
113    let holder = lock
114        .read_lock_file()
115        .and_then(|data| data.get("holder").and_then(Value::as_str).map(String::from));
116    let busy = match &holder {
117        Some(holder) => format!(
118            "A Docker snapshot resume of {} is in progress on this host.",
119            holder
120        ),
121        None => "A Docker snapshot resume is in progress on this host.".to_string(),
122    };
123    if wait_ms <= 1000 {
124        return Err(SnapshotFailure::tempfail(format!(
125            "{} Retry after it finishes, or wait for it with --snapshot-wait <seconds>.",
126            busy
127        )));
128    }
129    let waited = format_seconds(wait_ms as f64 / 1000.0);
130    (request.notify)(&format!(
131        "[Resume] Waiting up to {} for the snapshot of {} to finish before snapshotting {}.",
132        waited,
133        holder.as_deref().unwrap_or("another container"),
134        request.holder.unwrap_or("this container")
135    ));
136    if lock.acquire(wait_ms - 1000) {
137        return Ok(lock);
138    }
139    Err(SnapshotFailure::tempfail(format!(
140        "{} Still busy after waiting {}; retry after it finishes.",
141        busy, waited
142    )))
143}
144
145pub fn containerd_root(config: &str) -> String {
146    for line in config.lines().map(str::trim) {
147        if line.starts_with('[') {
148            break;
149        }
150        if let Some((key, value)) = line.split_once('=') {
151            let value = value.trim();
152            if key.trim() == "root" && (value.starts_with('"') || value.starts_with('\'')) {
153                let quote = value.chars().next().unwrap();
154                if let Some(end) = value[1..].find(quote) {
155                    return value[1..end + 1].into();
156                }
157            }
158        }
159    }
160    "/var/lib/containerd".into()
161}
162
163fn default_containerd_root() -> Result<String, String> {
164    if let Ok(root) = std::env::var("START_CONTAINERD_ROOT") {
165        return Ok(root);
166    }
167    match std::fs::read_to_string("/etc/containerd/config.toml") {
168        Ok(config) => Ok(containerd_root(&config)),
169        Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(containerd_root("")),
170        Err(e) => Err(format!(
171            "Cannot read containerd storage configuration: {}",
172            e
173        )),
174    }
175}
176
177/// Free bytes at `root` according to `df -Pk`.
178pub fn df_free_bytes<R: CommandRunner + ?Sized>(runner: &R, root: &str) -> Result<u64, String> {
179    let result = runner.run("df", &["-Pk".into(), root.into()]);
180    if !result.success {
181        return Err(format!(
182            "Cannot check free disk at {}: {}",
183            root,
184            result.stderr.trim()
185        ));
186    }
187    result
188        .stdout
189        .lines()
190        .last()
191        .and_then(|line| line.split_whitespace().nth(3))
192        .and_then(|s| s.parse::<u64>().ok())
193        .and_then(|n| n.checked_mul(1024))
194        .ok_or_else(|| format!("Cannot determine free disk at {}", root))
195}
196
197pub fn preflight_snapshot<R: CommandRunner + ?Sized>(
198    name: &str,
199    runner: &R,
200) -> Result<String, String> {
201    preflight_snapshot_with(name, runner, None, &|root| df_free_bytes(runner, root))
202}
203
204pub fn preflight_snapshot_with<R: CommandRunner + ?Sized>(
205    name: &str,
206    runner: &R,
207    containerd_override: Option<&str>,
208    free: &dyn Fn(&str) -> Result<u64, String>,
209) -> Result<String, String> {
210    let options = PreflightOptions {
211        containerd_root: containerd_override,
212        ..Default::default()
213    };
214    preflight_snapshot_guarded(name, runner, &options, free).map_err(|e| e.message)
215}
216
217fn sleep_for(duration: Duration) {
218    std::thread::sleep(duration);
219}
220
221fn monotonic_seconds() -> f64 {
222    static START: std::sync::OnceLock<Instant> = std::sync::OnceLock::new();
223    START.get_or_init(Instant::now).elapsed().as_secs_f64()
224}
225
226/// Options of the disk preflight (issue #224).
227pub struct PreflightOptions<'a> {
228    /// Writable-layer size to use instead of `docker inspect --size`.
229    pub size_hint: Option<u64>,
230    /// Re-check free disk this long instead of refusing at once.
231    pub disk_wait_seconds: f64,
232    /// Poll interval; defaults to START_SNAPSHOT_DISK_POLL or 10 seconds.
233    pub disk_poll_seconds: Option<f64>,
234    pub containerd_root: Option<&'a str>,
235    pub notify: &'a dyn Fn(&str),
236    pub sleep: &'a dyn Fn(Duration),
237    /// Monotonic clock in seconds.
238    pub now: &'a dyn Fn() -> f64,
239}
240
241impl Default for PreflightOptions<'_> {
242    fn default() -> Self {
243        Self {
244            size_hint: None,
245            disk_wait_seconds: 0.0,
246            disk_poll_seconds: None,
247            containerd_root: None,
248            notify: &notify_stderr,
249            sleep: &sleep_for,
250            now: &monotonic_seconds,
251        }
252    }
253}
254
255fn measure_writable_layer<R: CommandRunner + ?Sized>(
256    name: &str,
257    runner: &R,
258    docker: &str,
259    size_hint: Option<u64>,
260) -> Result<(u64, Option<&'static str>), SnapshotFailure> {
261    if let Some(size) = size_hint {
262        return Ok((size, Some("size hint")));
263    }
264    let result = runner.run(
265        docker,
266        &[
267            "inspect".into(),
268            "--size".into(),
269            "--format".into(),
270            "{{.SizeRw}}".into(),
271            name.into(),
272        ],
273    );
274    let size = result
275        .stdout
276        .trim()
277        .parse::<u64>()
278        .ok()
279        .filter(|_| result.success)
280        .ok_or("Cannot determine the writable layer size; refusing Docker snapshot resume.")?;
281    Ok((size, None))
282}
283
284fn image_store_roots<R: CommandRunner + ?Sized>(
285    runner: &R,
286    docker: &str,
287    containerd_override: Option<&str>,
288) -> Result<Vec<String>, SnapshotFailure> {
289    let result = runner.run(
290        docker,
291        &["info".into(), "--format".into(), "{{json .}}".into()],
292    );
293    let info: Value = serde_json::from_str(&result.stdout).unwrap_or(Value::Null);
294    let root = info
295        .get("DockerRootDir")
296        .and_then(Value::as_str)
297        .filter(|root| !root.is_empty() && result.success)
298        .ok_or("Cannot determine Docker data root; refusing Docker snapshot resume.")?;
299    let mut roots = vec![root.to_string()];
300    if info
301        .get("DriverStatus")
302        .is_some_and(|d| d.to_string().contains("io.containerd.snapshotter"))
303    {
304        let root = match containerd_override {
305            Some(root) => root.to_string(),
306            None => default_containerd_root()?,
307        };
308        if !roots.contains(&root) {
309            roots.push(root);
310        }
311    }
312    Ok(roots)
313}
314
315/// The first root without `required` free bytes, with what it has.
316fn find_short_root(
317    roots: &[String],
318    required: u64,
319    free: &dyn Fn(&str) -> Result<u64, String>,
320) -> Result<Option<(String, u64)>, String> {
321    for root in roots {
322        let available = free(root)?;
323        if available < required {
324            return Ok(Some((root.clone(), available)));
325        }
326    }
327    Ok(None)
328}
329
330fn gib(bytes: u64) -> f64 {
331    bytes as f64 / GIB as f64
332}
333
334/// Refuse a snapshot that could fill the disk. The writable layer is measured
335/// once (or taken from `size_hint`); with `disk_wait_seconds` free space is
336/// re-checked every poll interval until it suffices. A disk that stays short
337/// fails with [`EX_TEMPFAIL`].
338pub fn preflight_snapshot_guarded<R: CommandRunner + ?Sized>(
339    name: &str,
340    runner: &R,
341    options: &PreflightOptions,
342    free: &dyn Fn(&str) -> Result<u64, String>,
343) -> Result<String, SnapshotFailure> {
344    let docker = docker_command().to_string_lossy().to_string();
345    let (size, source) = measure_writable_layer(name, runner, &docker, options.size_hint)?;
346    let roots = image_store_roots(runner, &docker, options.containerd_root)?;
347    let required = size
348        .checked_mul(2)
349        .and_then(|n| n.checked_add(10 * GIB))
350        .ok_or("Snapshot size exceeds the supported range.")?;
351    let wait = options.disk_wait_seconds.max(0.0);
352    let poll = options
353        .disk_poll_seconds
354        .or_else(|| {
355            std::env::var("START_SNAPSHOT_DISK_POLL")
356                .ok()
357                .and_then(|value| value.parse::<f64>().ok())
358        })
359        .filter(|seconds| seconds.is_finite() && *seconds > 0.0)
360        .unwrap_or(10.0);
361    let started = (options.now)();
362    let mut waited = false;
363    while let Some((root, available)) = find_short_root(&roots, required, free)? {
364        let detail = format!(
365            "Insufficient disk at {}: need {:.2} GiB (2 × writable layer plus reserve), available {:.2} GiB.",
366            root,
367            gib(required),
368            gib(available)
369        );
370        let left = wait - ((options.now)() - started);
371        if left <= 0.0 {
372            return Err(SnapshotFailure::tempfail(if waited {
373                format!(
374                    "{} Still short after waiting {}; Docker snapshot resume refused.",
375                    detail,
376                    format_seconds(wait)
377                )
378            } else {
379                format!("{} Docker snapshot resume refused.", detail)
380            }));
381        }
382        if !waited {
383            (options.notify)(&format!(
384                "[Resume] Waiting up to {} for disk: {}",
385                format_seconds(wait),
386                detail
387            ));
388            waited = true;
389        }
390        (options.sleep)(Duration::from_secs_f64(poll.min(left)));
391    }
392    let measured = match source {
393        Some(source) => format!("{:.2} GiB ({})", gib(size), source),
394        None => format!("{:.2} GiB", gib(size)),
395    };
396    Ok(format!(
397        "[Resume] snapshotting {}; disk preflight passed ({:.2} GiB required).",
398        measured,
399        gib(required)
400    ))
401}
402
403pub fn snapshot_image_cleanup_snippets(name: &str, redirection: &str) -> (String, String) {
404    let format = shell_quote(&format!(
405        "{{{{index .Config.Labels \"{}\"}}}}",
406        SNAPSHOT_IMAGE_LABEL
407    ));
408    (format!("__start_snapshot_image=$(docker inspect -f {} {} 2>/dev/null || true)", format, shell_quote(name)),
409     format!("case \"$__start_snapshot_image\" in start-command-resume/*) docker rmi \"$__start_snapshot_image\" {} || true;; esac", redirection))
410}
411
412#[cfg(test)]
413mod tests {
414    use super::*;
415    use crate::execution_control::CommandRunOutput;
416    use std::cell::RefCell;
417
418    struct Runner;
419    impl CommandRunner for Runner {
420        fn run(&self, _bin: &str, args: &[String]) -> CommandRunOutput {
421            CommandRunOutput {
422                success: true,
423                stdout: if args[0] == "info" {
424                    "{\"DockerRootDir\":\"/docker\",\"DriverStatus\":[[\"driver-type\",\"io.containerd.snapshotter.v1\"]]}".into()
425                } else {
426                    (8 * GIB).to_string()
427                },
428                ..Default::default()
429            }
430        }
431    }
432
433    #[test]
434    fn checks_both_roots_and_refuses_insufficient_containerd_disk() {
435        let checked = RefCell::new(vec![]);
436        let error = preflight_snapshot_with("box", &Runner, Some("/containerd"), &|root| {
437            checked.borrow_mut().push(root.to_string());
438            Ok(if root == "/docker" {
439                100 * GIB
440            } else {
441                25 * GIB
442            })
443        })
444        .unwrap_err();
445        assert!(error.contains("/containerd"));
446        assert_eq!(*checked.borrow(), vec!["/docker", "/containerd"]);
447    }
448
449    #[test]
450    fn reads_only_top_level_containerd_root() {
451        assert_eq!(
452            containerd_root("root = '/custom'\n[plugins]\nroot = '/wrong'"),
453            "/custom"
454        );
455    }
456}
start_command/docker_snapshot_safety.rs

start_command/
docker_snapshot_safety.rs
