|
@@ -1,10 +1,62 @@
|
|
|
use crate::config::{Cluster, Config, Node};
|
|
use crate::config::{Cluster, Config, Node};
|
|
|
use crate::ssh::{run_ssh, run_ssh_long_running};
|
|
use crate::ssh::{run_ssh, run_ssh_long_running};
|
|
|
use anyhow::Result;
|
|
use anyhow::Result;
|
|
|
|
|
+use std::time::Duration;
|
|
|
use tracing::info;
|
|
use tracing::info;
|
|
|
|
|
|
|
|
use super::shq;
|
|
use super::shq;
|
|
|
|
|
|
|
|
|
|
+/// The containerd snapshotter (`io.containerd.snapshotter.v1` — every DGX
|
|
|
|
|
+/// Spark node) returns `docker load` BEFORE the snapshot is fully unpacked,
|
|
|
|
|
+/// so a `docker run` issued seconds after the sync can die with
|
|
|
|
|
+/// "unable to start container process ... stat ...: no such file or
|
|
|
|
|
+/// directory" (dgx3 lost exactly that race on the first Mistral Large 4
|
|
|
|
|
+/// serve, 2026-10-29: rank 2 died silently, `--rm` removed the evidence and
|
|
|
|
|
+/// the rendezvous hung until its 30-min timeout). Probe the exec path with
|
|
|
|
|
+/// retries: starting a trivial `/bin/sh` container forces the snapshot
|
|
|
|
|
+/// materialization — once that passes, the engine's own exec works too.
|
|
|
|
|
+async fn verify_image_exec_ready(config: &Config, target: &Node, image: &str) -> Result<()> {
|
|
|
|
|
+ const PROBE_TRIES: u32 = 24; // 24 x 5 s = 2 min worst-case cap
|
|
|
|
|
+ const PROBE_SECS: u64 = 5;
|
|
|
|
|
+ let probe = exec_probe_command(image);
|
|
|
|
|
+ for attempt in 1..=PROBE_TRIES {
|
|
|
|
|
+ match run_ssh(config, target, probe.clone(), false).await {
|
|
|
|
|
+ Ok(_) => {
|
|
|
|
|
+ if attempt > 1 {
|
|
|
|
|
+ info!(
|
|
|
|
|
+ "image unpack probe passed on '{}' (attempt {attempt})",
|
|
|
|
|
+ target.name
|
|
|
|
|
+ );
|
|
|
|
|
+ }
|
|
|
|
|
+ return Ok(());
|
|
|
|
|
+ }
|
|
|
|
|
+ Err(e) => {
|
|
|
|
|
+ if attempt == 1 {
|
|
|
|
|
+ info!(
|
|
|
|
|
+ "image unpack probe on '{}' failed (attempt 1): {e:#} — retrying",
|
|
|
|
|
+ target.name
|
|
|
|
|
+ );
|
|
|
|
|
+ }
|
|
|
|
|
+ tokio::time::sleep(Duration::from_secs(PROBE_SECS)).await;
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ anyhow::bail!(
|
|
|
|
|
+ "image '{image}' on '{}' is still not exec-ready after {PROBE_TRIES} probe attempts — the containerd snapshotter has not finished unpacking it; refusing to start containers against a half-materialized image",
|
|
|
|
|
+ target.name
|
|
|
|
|
+ )
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+/// The exec-readiness probe: starting a trivial `/bin/sh` container forces
|
|
|
|
|
+/// the containerd snapshotter to materialize the rootfs — after this passes,
|
|
|
|
|
+/// the engine's own exec works too.
|
|
|
|
|
+fn exec_probe_command(image: &str) -> String {
|
|
|
|
|
+ format!(
|
|
|
|
|
+ "docker run --rm --entrypoint /bin/sh {} -c 'exit 0'",
|
|
|
|
|
+ shq(image)
|
|
|
|
|
+ )
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
/// Get the Docker image digest (short ID) from a node.
|
|
/// Get the Docker image digest (short ID) from a node.
|
|
|
async fn get_image_digest(config: &Config, node: &Node, image: &str) -> Result<Option<String>> {
|
|
async fn get_image_digest(config: &Config, node: &Node, image: &str) -> Result<Option<String>> {
|
|
|
// `docker image ls` does not accept digest refs (`repo@sha256:...`) —
|
|
// `docker image ls` does not accept digest refs (`repo@sha256:...`) —
|
|
@@ -287,6 +339,17 @@ pub async fn sync_image_to_nodes(
|
|
|
let verify = get_image_digest(&config, &target, &image).await;
|
|
let verify = get_image_digest(&config, &target, &image).await;
|
|
|
match verify {
|
|
match verify {
|
|
|
Ok(v) if v.as_deref() == Some(expected.as_str()) => {
|
|
Ok(v) if v.as_deref() == Some(expected.as_str()) => {
|
|
|
|
|
+ // Digest present ≠ snapshot unpacked (the
|
|
|
|
|
+ // containerd lazy-unpack race): probe exec
|
|
|
|
|
+ // readiness before declaring the sync done.
|
|
|
|
|
+ // A probe failure is not a connectivity
|
|
|
|
|
+ // problem — the other IP won't help either.
|
|
|
|
|
+ if let Err(e) =
|
|
|
|
|
+ verify_image_exec_ready(&config, &target, &image).await
|
|
|
|
|
+ {
|
|
|
|
|
+ last_err = Some(format!("image exec-ready probe: {e}"));
|
|
|
|
|
+ break;
|
|
|
|
|
+ }
|
|
|
synced = true;
|
|
synced = true;
|
|
|
break;
|
|
break;
|
|
|
}
|
|
}
|
|
@@ -415,7 +478,7 @@ pub async fn handle_sync_image(
|
|
|
|
|
|
|
|
#[cfg(test)]
|
|
#[cfg(test)]
|
|
|
mod tests {
|
|
mod tests {
|
|
|
- use super::safe_name;
|
|
|
|
|
|
|
+ use super::{exec_probe_command, safe_name};
|
|
|
|
|
|
|
|
#[test]
|
|
#[test]
|
|
|
fn safe_image_name_is_filename_safe() {
|
|
fn safe_image_name_is_filename_safe() {
|
|
@@ -423,4 +486,12 @@ mod tests {
|
|
|
assert!(!safe.contains('/') && !safe.contains(':') && !safe.contains('@'));
|
|
assert!(!safe.contains('/') && !safe.contains(':') && !safe.contains('@'));
|
|
|
assert!(!safe.is_empty());
|
|
assert!(!safe.is_empty());
|
|
|
}
|
|
}
|
|
|
|
|
+
|
|
|
|
|
+ #[test]
|
|
|
|
|
+ fn exec_probe_forces_snapshot_materialization() {
|
|
|
|
|
+ let cmd = exec_probe_command("mistral-large4-vllm:local");
|
|
|
|
|
+ assert!(cmd.starts_with("docker run --rm --entrypoint /bin/sh "));
|
|
|
|
|
+ assert!(cmd.ends_with("-c 'exit 0'"));
|
|
|
|
|
+ assert!(cmd.contains("'mistral-large4-vllm:local'"));
|
|
|
|
|
+ }
|
|
|
}
|
|
}
|