use anyhow::Result; use tracing::{info, warn}; use crate::config::{Cluster, Config, OnFailure, PreServeCommand, Recipe}; /// Run pre-serve commands on every node of the cluster before launching /// containers or tmux sessions. Combines cluster-level config /// (`[before_serve.""]`) with recipe-level overrides /// (`before_serve = [...]` in recipe.toml). Cluster-level commands run /// first, then recipe-level — each in declared order — so the resulting /// state is predictable. /// /// Per-command behaviour on a non-zero exit is controlled by /// [`PreServeCommand::on_failure`]: /// - `abort` (default) — fail-fast: abort the whole serve /// - `warn` — log warning, continue with remaining commands /// - `continue`— stay silent, continue /// /// If `skip` is true the function returns without running anything. pub async fn run( config: &Config, cluster: &Cluster, recipe: &Recipe, skip: bool, dry_run: bool, ) -> Result<()> { if skip { info!("⏭ Skipping pre-serve hooks (--skip-before-serve)"); return Ok(()); } let cmds = combined_commands(config, cluster, recipe); if cmds.is_empty() { return Ok(()); } info!( "🔧 Pre-serve hooks: {} command(s) on {} node(s)", cmds.len(), cluster.nodes.len() ); // Run all commands for one node sequentially (so order matters per // node), and the nodes in parallel. let mut handles = Vec::with_capacity(cluster.nodes.len()); for node in &cluster.nodes { let config = config.clone(); let node = node.clone(); let cmds = cmds.clone(); let dry = dry_run; handles.push(tokio::spawn(async move { ( node.name.clone(), run_for_node(&config, &node, &cmds, dry).await, ) })); } let mut any_failed = false; for h in handles { let (name, result) = h.await?; if let Err(e) = result { any_failed = true; warn!( "{}", fluent_i18n::t!("pre-serve-failed", { "name" => &name, "error" => e.to_string() }) ); } } if any_failed { anyhow::bail!("Pre-serve hooks failed on at least one node (see warnings above)"); } Ok(()) } /// Combine cluster-level + recipe-level commands into one ordered list. fn combined_commands( config: &Config, cluster: &Cluster, recipe: &Recipe, ) -> Vec<(&'static str, PreServeCommand)> { let mut out: Vec<(&'static str, PreServeCommand)> = Vec::new(); if let Some(table) = config.before_serve.get(&cluster.name) { for c in &table.commands { out.push(("cluster", c.clone())); } } for c in &recipe.before_serve { out.push(("recipe", c.clone())); } out } /// Run a fixed list of commands on a single node, sequentially. /// Returns the first command that failed under `on_failure = abort`. async fn run_for_node( config: &Config, node: &crate::config::Node, cmds: &[(&'static str, PreServeCommand)], dry_run: bool, ) -> Result<()> { for (origin, cmd) in cmds { let command = cmd.command(); let on_failure = cmd.on_failure(); if dry_run { info!( "{}", fluent_i18n::t!("pre-serve-run", { "node" => &node.name, "tag" => &origin, "cmd" => truncate(command, 120) }) ); continue; } let result = crate::ssh::run_ssh(config, node, command.to_string(), dry_run).await; let status = match &result { Ok(_) => "ok", Err(_) => "fail", }; let label = format!("[{}|{}|{}]", node.name, origin, status); info!(" {} $ {}", label, command); if let Err(ref e) = result { match on_failure { OnFailure::Abort => { return Err(anyhow::anyhow!( "pre-serve command failed on {}: {}", node.name, e )); } OnFailure::Warn => { warn!( "{}", fluent_i18n::t!("pre-serve-ignored", { "cmd" => &node.name, "error" => e.to_string() }) ); } OnFailure::Continue => { // silent } } } } Ok(()) } fn truncate(s: &str, max: usize) -> String { if s.len() <= max { s.to_string() } else { format!("{}…", &s[..max]) } } #[cfg(test)] mod tests { use super::*; use crate::config::{Node, PreServeCommand}; use std::collections::HashMap; fn test_cluster() -> Cluster { Cluster { name: "c1".to_string(), master_ib_ip: "1.2.3.4".to_string(), nodes: vec![ Node { name: "n0".to_string(), ssh_ip: "10.0.0.1".to_string(), ib_ip: "1.2.3.4".to_string(), rank: 0, host_key: None, entity: None, }, Node { name: "n1".to_string(), ssh_ip: "10.0.0.2".to_string(), ib_ip: "1.2.3.5".to_string(), rank: 1, host_key: None, entity: None, }, ], extra_entities: vec![], } } fn test_config(before_serve: HashMap) -> Config { Config { ssh_user: "u".to_string(), ssh_key_path: "/k".into(), clusters: vec![test_cluster()], nccl: HashMap::new(), hf_cache_dir: None, init_port: 25000, before_serve, exec_aliases: HashMap::new(), home_assistant: None, } } #[test] fn combined_orders_cluster_then_recipe() { let mut bs = HashMap::new(); bs.insert( "c1".to_string(), crate::config::BeforeServeTable { commands: vec![PreServeCommand::Bare("cluster_cmd_1".into())], }, ); let cfg = test_config(bs); let recipe = Recipe { before_serve: vec![ PreServeCommand::Bare("recipe_cmd_1".into()), PreServeCommand::Bare("recipe_cmd_2".into()), ], ..test_recipe() }; let cmds = combined_commands(&cfg, &test_cluster(), &recipe); let names: Vec<&str> = cmds.iter().map(|(_, c)| c.command()).collect(); let origins: Vec<&str> = cmds.iter().map(|(o, _)| *o).collect(); assert_eq!(names, vec!["cluster_cmd_1", "recipe_cmd_1", "recipe_cmd_2"]); assert_eq!(origins, vec!["cluster", "recipe", "recipe"]); } #[test] fn combined_empty_when_unconfigured() { let cfg = test_config(HashMap::new()); let recipe = Recipe { before_serve: vec![], ..test_recipe() }; assert!(combined_commands(&cfg, &test_cluster(), &recipe).is_empty()); } fn test_recipe() -> Recipe { Recipe { engine: "vllm".to_string(), name: "Test".to_string(), docker_image: String::new(), model: "x".to_string(), port: 30000, required_nodes: None, pipeline_parallel_size: None, tensor_parallel_size: 1, args: vec![], env: HashMap::new(), cluster: None, entrypoint: vec![], distributed_backend: "mp".to_string(), before_serve: vec![], shm_size: String::new(), } } // ── Deserialization tests (the ones that live in this module) ── #[test] fn parses_cluster_level_before_serve() { let toml_str = r#" ssh_user = "u" ssh_key_path = "/k" [[clusters]] name = "c1" master_ib_ip = "1.1.1.1" nodes = [ { name = "n0", ssh_ip = "2.2.2.2", ib_ip = "1.1.1.1", rank = 0 }, ] [before_serve."c1"] commands = [ "sudo nvidia-smi -lgc 300,2300", { command = "sudo nvidia-smi -pm 1", on_failure = "warn" }, ] "#; let cfg: Config = toml::from_str(toml_str).unwrap(); assert!(cfg.before_serve.contains_key("c1")); let list = &cfg.before_serve["c1"].commands; assert_eq!(list.len(), 2); assert_eq!(list[0].command(), "sudo nvidia-smi -lgc 300,2300"); assert_eq!(list[0].on_failure(), OnFailure::Abort); // default assert_eq!(list[1].command(), "sudo nvidia-smi -pm 1"); assert_eq!(list[1].on_failure(), OnFailure::Warn); } #[test] fn parses_recipe_level_before_serve() { let toml_str = r#" name = "Demo" model = "x" before_serve = [ "echo a", { command = "echo b", on_failure = "continue" }, ] "#; let r: Recipe = toml::from_str(toml_str).unwrap(); assert_eq!(r.before_serve.len(), 2); assert_eq!(r.before_serve[0].command(), "echo a"); assert_eq!(r.before_serve[0].on_failure(), OnFailure::Abort); assert_eq!(r.before_serve[1].command(), "echo b"); assert_eq!(r.before_serve[1].on_failure(), OnFailure::Continue); } }