//! 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. #[allow(clippy::too_many_arguments)] pub async fn prepare_cluster( config: &Config, recipe: &Recipe, cluster_name: &str, node_name: Option<&str>, auto_power_on: bool, prober: &P, power: Option<&C>, pinger: &N, ) -> Result<(Cluster, NcclConfig)> where P: NodeProber, C: PowerController, N: NodePinger, { let state = 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 auto_power_on && state_has_unreachable(&state, config) { let (c, _nccl) = resolve_cluster_simple(config, recipe, cluster_name)?; if let Some(pwr) = power { power_on_if_off(&state, &c, pwr).await?; } // Wait for nodes to become reachable. wait_for_all_nodes(config, &c.nodes, pinger).await?; // Re-probe after power-on. prober.probe(config).await } else { state }; 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 && !ns.occupied) }) .collect(); let c = Cluster { nodes: avail, ..c }; (c, n.clone()) }; // 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. fn resolve_solo_node(config: &Config, node_name: &str) -> Result<(Cluster, NcclConfig)> { 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() .unwrap_or_else(|| NcclConfig { ib_hca: String::new(), socket_ifname: "lo".to_string(), ib_gid_index: 3, ib_disable: 1, gloo_socket_ifname: None, ucx_net_devices: None, nccl_cross_nic: None, nccl_proto: None, nccl_p2p_level: None, }); (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. fn resolve_cluster_simple<'c>( config: &'c Config, recipe: &Recipe, cluster_name: &str, ) -> Result<(Cluster, &'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) } /// 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( _state: &ClusterState, 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, ) -> Result<()> { info!("{}", fluent_i18n::t!("cluster-waiting-for-nodes")); let boot_wait_secs = 60; info!("⏳ Waiting {}s for nodes to boot...", boot_wait_secs); tokio::time::sleep(Duration::from_secs(boot_wait_secs)).await; let max_poll_secs = 120; let poll_interval_secs = 10; let mut waited = 0; while waited < 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(poll_interval_secs)).await; waited += poll_interval_secs; } anyhow::bail!("{}", fluent_i18n::t!("cluster-nodes-not-available-timeout")); } #[cfg(test)] #[path = "tests.rs"] mod tests;