//! Serve orchestrator — combines probing, power control, and node-pinging //! into a single `prepare_cluster` step that `serve` calls. //! //! Every external interaction goes through the trait bounds `P: NodeProber`, //! `C: PowerController`, `N: NodePinger`, making the control flow testable //! with mock implementations. use std::time::Duration; use anyhow::{Context, Result}; use tracing::info; use crate::cluster_state::{ClusterState, select_cluster}; use crate::config::{Cluster, Config, NcclConfig, Node, Recipe}; use crate::ends::{NodePinger, NodeProber, PowerController, cluster_entities}; /// Prepare a cluster for serving: probe, optionally power on, wait, then /// select a cluster with its free nodes. /// /// Returns a `Cluster` whose `nodes` field contains only free (reachable /// and not occupied) nodes, plus the matching NCCL config (or `None` for /// single-node recipes, where NCCL config is not required). /// /// Timing for [`wait_for_all_nodes`] after a power-on: how long to wait /// for the machines to boot before the first poll, the poll interval, and /// the max total poll window. Injectable so tests don't sleep real seconds /// (a previous hardcoded 60s boot wait made the whole suite take 60s). #[derive(Clone, Debug)] pub struct BootTiming { pub boot_wait_secs: u64, pub poll_interval_secs: u64, pub max_poll_secs: u64, } impl Default for BootTiming { fn default() -> Self { Self { boot_wait_secs: 60, poll_interval_secs: 10, max_poll_secs: 120, } } } /// Bundles the injectable end-effectors + power-on flag so /// [`prepare_cluster`] stays under clippy's `too_many_arguments` threshold /// without an `#[allow]`. pub struct PrepareRequest<'a, P: NodeProber, C: PowerController, N: NodePinger> { pub prober: &'a P, pub power: Option<&'a C>, pub pinger: &'a N, pub auto_power_on: bool, pub boot_timing: BootTiming, /// `--force`: skip the occupancy filter during cluster selection. /// Nodes with another container running are kept in the available set /// (reachable is still required — can't SSH to an unreachable node). pub force_occupied: bool, } pub async fn prepare_cluster( config: &Config, recipe: &Recipe, cluster_name: &str, node_name: Option<&str>, req: &PrepareRequest<'_, P, C, N>, ) -> Result<(Cluster, Option)> where P: NodeProber, C: PowerController, N: NodePinger, { let state = req.prober.probe(config).await; let resolved = if let Some(nn) = node_name { resolve_solo_node(config, nn)? } else { // Auto-power-on: if any nodes are unreachable, try powering on. let state = if req.auto_power_on && state_has_unreachable(&state, config) { let (c, _nccl) = resolve_cluster_simple(config, recipe, cluster_name)?; if let Some(pwr) = req.power { power_on_if_off(&c, pwr).await?; } // Wait for nodes to become reachable. wait_for_all_nodes(config, &c.nodes, req.pinger, &req.boot_timing).await?; // Re-probe after power-on. req.prober.probe(config).await } else { state }; // `--force`: clear occupied flags so select_cluster counts reachable // nodes (ignoring occupancy) and the downstream filter keeps them. let mut state = state; if req.force_occupied { for ns in state.nodes.values_mut() { if ns.reachable { ns.occupied = false; } } } let (c, n) = select_cluster(config, cluster_name, recipe.required_nodes, &state)?; let avail: Vec = c .nodes .into_iter() .filter(|n| { state .nodes .get(&n.name) .is_some_and(|ns| ns.reachable && (req.force_occupied || !ns.occupied)) }) .collect(); let c = Cluster { nodes: avail, ..c }; (c, n.cloned()) }; // Validate node count. if let Some(required) = recipe.required_nodes { let actual = resolved.0.nodes.len() as u32; if actual < required { anyhow::bail!( "Recipe '{}' requires {} nodes, but cluster '{}' has only {} nodes", recipe.name, required, resolved.0.name, actual ); } } Ok(resolved) } /// Collect a single node from any cluster. /// /// Returns `None` for the NCCL config when the cluster has no NCCL entry /// — this is valid for single-node deployments where NCCL env vars are /// never emitted. fn resolve_solo_node(config: &Config, node_name: &str) -> Result<(Cluster, Option)> { let (solo_c, solo_n) = config .clusters .iter() .find_map(|c| { c.nodes.iter().find(|n| n.name == node_name).map(|node| { let solo = Cluster { name: c.name.clone(), master_ib_ip: node.ib_ip.clone(), nodes: vec![node.clone()], extra_entities: vec![], }; let nc = config.nccl.get(&c.name).cloned(); (solo, nc) }) }) .context(format!("Node '{}' not found in any cluster", node_name))?; Ok((solo_c, solo_n)) } /// Resolve cluster by name or auto treating every config node as free /// (reachable, not occupied). Only used for the power-on pre-check /// before any probe has been made, so being optimistic is safe: /// a full probe follows immediately after power-on completes. fn resolve_cluster_simple<'c>( config: &'c Config, recipe: &Recipe, cluster_name: &str, ) -> Result<(Cluster, Option<&'c NcclConfig>)> { use crate::cluster_state::NodeState; let mut state = ClusterState::default(); for cluster in &config.clusters { for node in &cluster.nodes { state.nodes.entry(node.name.clone()).or_insert(NodeState { reachable: true, occupied: false, }); } } select_cluster(config, cluster_name, recipe.required_nodes, &state) } /// Dry-run cluster resolution: treat every configured node as reachable and /// free, then select. Skips all SSH probing/power/pinging so a dry-run works /// even when the cluster is off or unreachable (the whole point of a dry-run /// is to preview the generated command, not to verify hardware). /// /// Like [`prepare_cluster`], returns the cluster with **all** its nodes (dry-run /// does not filter for free nodes, since none are actually inspected). pub fn prepare_cluster_dry_run( config: &Config, recipe: &Recipe, cluster_name: &str, node_name: Option<&str>, ) -> Result<(Cluster, Option)> { if let Some(nn) = node_name { let (c, n) = resolve_solo_node(config, nn)?; return Ok((c, n)); } let (c, n) = resolve_cluster_simple(config, recipe, cluster_name)?; Ok((c, n.cloned())) } /// Does the state contain any unreachable nodes? fn state_has_unreachable(state: &ClusterState, config: &Config) -> bool { for cluster in &config.clusters { for node in &cluster.nodes { if state.nodes.get(&node.name).is_some_and(|ns| !ns.reachable) { return true; } } } false } /// Power on HA entities that are not already on. async fn power_on_if_off(cluster: &Cluster, power: &impl PowerController) -> Result { let entities = cluster_entities(cluster); if entities.is_empty() { return Ok(false); } let any_off = { let mut off = false; for e in &entities { match power.entity_is_on(e).await { Some(true) => {} _ => off = true, } } off }; if !any_off { return Ok(false); } info!("{}", fluent_i18n::t!("cluster-switch-off-detected")); for e in &entities { power.turn_on(e).await?; } info!("{}", fluent_i18n::t!("cluster-switch-powered-on")); Ok(true) } /// Wait until every node in the list accepts an SSH ping, with a timeout. async fn wait_for_all_nodes( config: &Config, nodes: &[Node], pinger: &impl NodePinger, timing: &BootTiming, ) -> Result<()> { info!("{}", fluent_i18n::t!("cluster-waiting-for-nodes")); if timing.boot_wait_secs > 0 { info!( "{}", fluent_i18n::t!("boot-wait", { "secs" => timing.boot_wait_secs }) ); tokio::time::sleep(Duration::from_secs(timing.boot_wait_secs)).await; } let mut waited = 0; while waited < timing.max_poll_secs { let mut all_online = true; for node in nodes { if !pinger.is_reachable(config, node).await { all_online = false; break; } } if all_online { info!("{}", fluent_i18n::t!("cluster-nodes-available")); return Ok(()); } tokio::time::sleep(Duration::from_secs(timing.poll_interval_secs)).await; // Advance by at least 1s so a zero poll-interval (tests) can't spin // forever when the pinger reports unreachable. waited += timing.poll_interval_secs.max(1); } anyhow::bail!("{}", fluent_i18n::t!("cluster-nodes-not-available-timeout")); } #[cfg(test)] #[path = "tests.rs"] mod tests;