use crate::config::{Cluster, Config, Node}; use crate::ssh::{run_ssh, run_ssh_long_running}; use anyhow::Result; use tracing::info; use super::shq; /// Get the Docker image digest (short ID) from a node. async fn get_image_digest(config: &Config, node: &Node, image: &str) -> Result> { let cmd = format!("docker image ls --format '{{{{.ID}}}}' {}", shq(image)); let out = run_ssh(config, node, cmd, false).await?; let digest = out.trim().to_string(); if digest.is_empty() { Ok(None) } else { Ok(Some(digest)) } } /// Get the Docker image creation timestamp. async fn get_image_created(config: &Config, node: &Node, image: &str) -> Option { let cmd = format!( "docker image inspect --format '{{{{.Created}}}}' {} 2>/dev/null", shq(image) ); run_ssh(config, node, cmd, false) .await .ok() .map(|c| c.trim().to_string()) } /// Sync Docker image across cluster nodes. /// Finds the node with the *most recently built* image (by creation timestamp), /// then pipes `docker save | ssh docker load` to nodes missing or with a different digest. /// /// **Security note:** The daisy-chain SSH pipe (`docker save | ssh -o /// StrictHostKeyChecking=no ... docker load`) bypasses host key verification for /// the *inner* SSH hop. The outer SSH session (via russh) is verified, but the /// piped `ssh` command trusts the target host unconditionally for the duration /// of the pipe. This is a design trade-off: re-using the verified russh session /// for the pipe would require piping the data through the agent, which adds /// complexity. Acceptable for trusted-cluster deployments where the config /// file is the trust anchor. pub async fn sync_image_to_nodes( config: &Config, cluster: &Cluster, image: &str, explicit_source: Option<&str>, ) -> Result<()> { // Collect digests and creation timestamps per node. let mut digests: Vec<(&Node, Option)> = Vec::new(); let mut created: Vec> = Vec::new(); for node in &cluster.nodes { let d = get_image_digest(config, node, image).await?; let c = if d.is_some() { get_image_created(config, node, image).await } else { None }; digests.push((node, d)); created.push(c); } let source_idx = if let Some(src_name) = explicit_source { digests .iter() .position(|(n, _)| n.name == src_name) .ok_or_else(|| { anyhow::anyhow!("sync-image-source '{}' not found in cluster", src_name) })? } else { let mut indexed: Vec = digests .iter() .enumerate() .filter(|(_, (_, d))| d.is_some()) .map(|(i, _)| i) .collect(); indexed.sort_by(|&a, &b| { created[b] .as_deref() .unwrap_or("") .cmp(created[a].as_deref().unwrap_or("")) }); *indexed .first() .ok_or_else(|| anyhow::anyhow!("Image '{}' not found on any node", image))? }; let (src_node, src_digest_opt) = &digests[source_idx]; let source_digest = src_digest_opt.as_ref().unwrap(); info!( "Syncing image '{}' from '{}' (most recent build) to peers...", image, src_node.name ); for (target, target_digest) in &digests { if target.name == src_node.name { continue; } if target_digest.as_deref() == Some(source_digest) { info!( "{}", fluent_i18n::t!("sync-digest-match", { "name" => &target.name }) ); continue; } info!( "{}", fluent_i18n::t!("sync-syncing", { "name" => &target.name, "ip" => &target.ssh_ip }) ); let pipe_cmd = format!( "docker save {} | ssh -o StrictHostKeyChecking=no -i {} {}@{} docker load", shq(image), shq(&config.ssh_key_path.display().to_string()), shq(&config.ssh_user), shq(&target.ssh_ip) ); run_ssh_long_running(config, src_node, pipe_cmd, false).await?; let verify = get_image_digest(config, target, image).await?; if verify.as_deref() != Some(source_digest) { anyhow::bail!( "Image sync to '{}' failed: digest mismatch (source={}, target={:?})", target.name, source_digest, verify ); } info!( "{}", fluent_i18n::t!("sync-synced", { "name" => &target.name }) ); } Ok(()) }