config.rs 17 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488
  1. pub use crate::pre_serve::{BeforeServeTable, OnFailure, PreServeCommand};
  2. use anyhow::{Context, Result};
  3. use serde::Deserialize;
  4. use std::collections::HashMap;
  5. use tracing::info;
  6. #[derive(Clone, Deserialize)]
  7. pub struct Node {
  8. pub name: String,
  9. pub ssh_ip: String,
  10. pub ib_ip: String,
  11. pub rank: u32,
  12. #[serde(default)]
  13. pub host_key: Option<String>,
  14. #[serde(default)]
  15. pub entity: Option<String>,
  16. }
  17. #[derive(Clone, Deserialize)]
  18. pub struct NcclConfig {
  19. pub ib_hca: String,
  20. pub socket_ifname: String,
  21. pub ib_gid_index: u32,
  22. pub ib_disable: u32,
  23. }
  24. #[derive(Clone, Deserialize)]
  25. pub struct Cluster {
  26. pub name: String,
  27. pub master_ib_ip: String,
  28. pub nodes: Vec<Node>,
  29. /// Extra Home Assistant switches to toggle alongside the nodes during power on/off.
  30. #[serde(default)]
  31. pub extra_entities: Vec<String>,
  32. }
  33. /// Top-level config (clusters + per-cluster pre-serve commands).
  34. #[derive(Clone, Deserialize)]
  35. pub struct Config {
  36. pub ssh_user: String,
  37. pub ssh_key_path: std::path::PathBuf,
  38. pub clusters: Vec<Cluster>,
  39. #[serde(default)]
  40. pub nccl: HashMap<String, NcclConfig>,
  41. #[serde(default)]
  42. pub hf_cache_dir: Option<String>,
  43. /// TCP port for cluster init (vLLM/SGLang)
  44. #[serde(default = "default_init_port")]
  45. pub init_port: u16,
  46. /// Per-cluster pre-serve commands, command aliases, HA config
  47. #[serde(default)]
  48. pub before_serve: HashMap<String, BeforeServeTable>, // <-- move here
  49. #[serde(default)]
  50. pub exec_aliases: HashMap<String, String>,
  51. #[serde(default)]
  52. pub home_assistant: Option<HomeAssistantConfig>,
  53. }
  54. /// Home Assistant config for `power on`/`power off`.
  55. #[derive(Clone, Deserialize)]
  56. pub struct HomeAssistantConfig {
  57. pub url: String,
  58. pub token: String,
  59. #[serde(default = "default_shutdown_delay")]
  60. pub shutdown_delay_seconds: u64,
  61. }
  62. fn default_shutdown_delay() -> u64 {
  63. 30
  64. }
  65. fn default_init_port() -> u16 {
  66. 25000
  67. }
  68. fn default_port() -> u16 {
  69. 30000
  70. }
  71. impl Config {
  72. pub fn resolved_hf_cache_dir(&self) -> std::path::PathBuf {
  73. if let Some(ref dir) = self.hf_cache_dir {
  74. return std::path::PathBuf::from(dir);
  75. }
  76. let home = if self.ssh_user == "root" {
  77. "/root"
  78. } else {
  79. // Safe to return a temporary String; PathBuf::from copies it
  80. return std::path::PathBuf::from(format!("/home/{}/.cache/huggingface", self.ssh_user));
  81. };
  82. std::path::PathBuf::from(format!("{}/.cache/huggingface", home))
  83. }
  84. }
  85. #[derive(Clone, Deserialize)]
  86. pub struct Recipe {
  87. #[serde(default = "default_engine")]
  88. pub engine: String,
  89. pub name: String,
  90. #[serde(default)]
  91. pub docker_image: String,
  92. #[serde(default)]
  93. pub model: String,
  94. #[serde(default = "default_port")]
  95. pub port: u16,
  96. #[serde(default)]
  97. pub required_nodes: Option<u32>,
  98. #[serde(default)]
  99. pub pipeline_parallel_size: Option<u32>,
  100. #[serde(default = "default_tensor_parallel_size")]
  101. pub tensor_parallel_size: u32,
  102. #[serde(default)]
  103. pub args: Vec<String>,
  104. #[serde(default)]
  105. pub env: HashMap<String, String>,
  106. #[serde(default)]
  107. pub cluster: Option<String>,
  108. #[serde(default)]
  109. pub entrypoint: Vec<String>,
  110. #[serde(default = "default_distributed_backend")]
  111. pub distributed_backend: String,
  112. /// Shared memory size for the container (e.g. "64gb"). Omitted when empty.
  113. #[serde(default = "default_shm_size")]
  114. pub shm_size: String,
  115. /// Recipe-level pre-serve commands (appended after cluster-level).
  116. #[serde(default)]
  117. pub before_serve: Vec<PreServeCommand>,
  118. }
  119. /// Resolves cluster by exact name, then unique prefix. Errors if none or ambiguous.
  120. fn resolve_cluster_name<'a>(clusters: &'a [Cluster], name: &str) -> Result<&'a Cluster> {
  121. if let Some(c) = clusters.iter().find(|c| c.name == name) {
  122. return Ok(c);
  123. }
  124. let matches: Vec<&Cluster> = clusters
  125. .iter()
  126. .filter(|c| c.name.starts_with(name))
  127. .collect();
  128. match matches.len() {
  129. 0 => anyhow::bail!(fluent_i18n::t!("cluster-not-found", { "name" => name.to_string() })),
  130. 1 => Ok(matches[0]),
  131. n => anyhow::bail!(
  132. "Pref '{}' matches {} clusters: {}",
  133. name,
  134. n,
  135. matches
  136. .iter()
  137. .map(|c| c.name.as_str())
  138. .collect::<Vec<_>>()
  139. .join(", ")
  140. ),
  141. }
  142. }
  143. fn default_engine() -> String {
  144. "vllm".to_string()
  145. }
  146. fn default_distributed_backend() -> String {
  147. "mp".to_string()
  148. }
  149. fn default_shm_size() -> String {
  150. String::new()
  151. }
  152. fn default_tensor_parallel_size() -> u32 {
  153. 1
  154. }
  155. /// Selects a cluster + NCCL config by name or auto-picks the smallest
  156. /// cluster with NCCL config satisfying `required_nodes`.
  157. pub fn select_cluster<'c>(
  158. config: &'c Config,
  159. cluster_name: &str,
  160. required_nodes: Option<u32>,
  161. ) -> Result<(Cluster, &'c NcclConfig)> {
  162. if cluster_name != "auto" {
  163. let cluster = resolve_cluster_name(&config.clusters, cluster_name)?;
  164. let nccl = config
  165. .nccl
  166. .get(&cluster.name)
  167. .context(fluent_i18n::t!("nccl-config-missing"))?;
  168. return Ok((cluster.clone(), nccl));
  169. }
  170. let mut candidates: Vec<(&Cluster, &NcclConfig)> = config
  171. .clusters
  172. .iter()
  173. .filter_map(|c| Some((c, config.nccl.get(&c.name)?)))
  174. .filter(|(c, _)| required_nodes.is_none_or(|r| (c.nodes.len() as u32) == r))
  175. .collect();
  176. candidates.sort_by_key(|(c, _)| c.nodes.len());
  177. if let Some((cluster, nccl)) = candidates.into_iter().next() {
  178. info!(
  179. "{}",
  180. fluent_i18n::t!("cluster-selected", {
  181. "name" => cluster.name.to_string(),
  182. "nodes" => cluster.nodes.len(),
  183. })
  184. );
  185. return Ok((cluster.clone(), nccl));
  186. }
  187. anyhow::bail!(fluent_i18n::t!("cluster-auto-fail"));
  188. }
  189. /// Looks up a cluster by name (or unique prefix) without requiring NCCL config.
  190. pub fn select_cluster_by_name(config: &Config, cluster_name: &str) -> Result<Cluster> {
  191. Ok(resolve_cluster_name(&config.clusters, cluster_name)?.clone())
  192. }
  193. /// Resolves the cluster for a recipe, applying recipe-level cluster pinning.
  194. pub fn resolve_cluster<'c>(
  195. config: &'c Config,
  196. recipe: &Recipe,
  197. cluster_name: &str,
  198. ) -> Result<(Cluster, &'c NcclConfig)> {
  199. let choice = if cluster_name != "auto" {
  200. cluster_name.to_string()
  201. } else if let Some(ref rc) = recipe.cluster {
  202. rc.clone()
  203. } else {
  204. "auto".to_string()
  205. };
  206. select_cluster(config, &choice, recipe.required_nodes)
  207. }
  208. /// Loads a recipe (cwd first, then `<default_config_dir>/recipes/<name>/recipe.toml`,
  209. /// falling back to cwd for the error).
  210. pub async fn load_recipe(recipe_name: &str) -> Result<Recipe> {
  211. let cwd_path = format!("recipes/{}/recipe.toml", recipe_name);
  212. if let Ok(content) = tokio::fs::read_to_string(&cwd_path).await {
  213. return Ok(toml::from_str(&content)?);
  214. }
  215. if let Some(dir) = default_config_dir() {
  216. let in_default = dir.join("recipes").join(recipe_name).join("recipe.toml");
  217. if let Ok(content) = tokio::fs::read_to_string(&in_default).await {
  218. return Ok(toml::from_str(&content)?);
  219. }
  220. }
  221. // Neither found — let the cwd path produce a clean error.
  222. let content = tokio::fs::read_to_string(&cwd_path).await?;
  223. Ok(toml::from_str(&content)?)
  224. }
  225. /// OS-appropriate default config dir (e.g. `~/.config/sm121-launcher`),
  226. /// or `None` for sandboxes with no known config dir.
  227. pub fn default_config_dir() -> Option<std::path::PathBuf> {
  228. dirs::config_dir().map(|p| p.join("sm121-launcher"))
  229. }
  230. #[cfg(test)]
  231. mod tests {
  232. use super::*;
  233. fn test_cluster() -> Cluster {
  234. Cluster {
  235. name: "test-cluster".to_string(),
  236. master_ib_ip: "192.168.1.1".to_string(),
  237. nodes: vec![
  238. Node {
  239. name: "node0".to_string(),
  240. ssh_ip: "10.0.0.1".to_string(),
  241. ib_ip: "192.168.1.1".to_string(),
  242. rank: 0,
  243. host_key: None,
  244. entity: None,
  245. },
  246. Node {
  247. name: "node1".to_string(),
  248. ssh_ip: "10.0.0.2".to_string(),
  249. ib_ip: "192.168.1.2".to_string(),
  250. rank: 1,
  251. host_key: None,
  252. entity: None,
  253. },
  254. ],
  255. extra_entities: vec![],
  256. }
  257. }
  258. fn test_config() -> Config {
  259. Config {
  260. ssh_user: "u".to_string(),
  261. ssh_key_path: "/k".into(),
  262. clusters: vec![test_cluster()],
  263. nccl: HashMap::new(),
  264. hf_cache_dir: None,
  265. init_port: 25000,
  266. before_serve: HashMap::new(),
  267. exec_aliases: HashMap::new(),
  268. home_assistant: None,
  269. }
  270. }
  271. fn test_nccl() -> NcclConfig {
  272. NcclConfig {
  273. ib_hca: "rocep1s0f0".to_string(),
  274. socket_ifname: "enp1s0f0np0".to_string(),
  275. ib_gid_index: 3,
  276. ib_disable: 0,
  277. }
  278. }
  279. // ── Deserialization tests ──────────────────────────────────
  280. #[test]
  281. fn test_recipe_deserialize_minimal() {
  282. let toml_str = r#"
  283. name = "Minimal"
  284. model = "test/model"
  285. "#;
  286. let recipe: Recipe = toml::from_str(toml_str).unwrap();
  287. assert_eq!(recipe.name, "Minimal");
  288. assert_eq!(recipe.model, "test/model");
  289. assert_eq!(recipe.engine, "vllm");
  290. assert_eq!(recipe.port, 30000);
  291. assert!(recipe.args.is_empty());
  292. assert!(recipe.env.is_empty());
  293. assert_eq!(recipe.distributed_backend, "mp");
  294. assert_eq!(recipe.tensor_parallel_size, 1);
  295. }
  296. #[test]
  297. fn test_recipe_deserialize_full() {
  298. let toml_str = r#"
  299. engine = "sglang"
  300. name = "Full Recipe"
  301. docker_image = "org/image:tag"
  302. model = "org/model"
  303. port = 30000
  304. required_nodes = 4
  305. pipeline_parallel_size = 2
  306. distributed_backend = "ray"
  307. cluster = "my-cluster"
  308. args = ["--arg1", "val1", "--arg2"]
  309. [env]
  310. KEY = "value"
  311. OTHER = "123"
  312. "#;
  313. let recipe: Recipe = toml::from_str(toml_str).unwrap();
  314. assert_eq!(recipe.engine, "sglang");
  315. assert_eq!(recipe.name, "Full Recipe");
  316. assert_eq!(recipe.docker_image, "org/image:tag");
  317. assert_eq!(recipe.model, "org/model");
  318. assert_eq!(recipe.port, 30000);
  319. assert_eq!(recipe.required_nodes, Some(4));
  320. assert_eq!(recipe.pipeline_parallel_size, Some(2));
  321. assert_eq!(recipe.tensor_parallel_size, 1);
  322. assert_eq!(recipe.distributed_backend, "ray");
  323. assert_eq!(recipe.cluster, Some("my-cluster".to_string()));
  324. assert_eq!(recipe.args, vec!["--arg1", "val1", "--arg2"]);
  325. assert_eq!(recipe.env.get("KEY").unwrap(), "value");
  326. assert_eq!(recipe.env.get("OTHER").unwrap(), "123");
  327. }
  328. #[test]
  329. fn test_recipe_deserialize_no_required_nodes() {
  330. let toml_str = r#"
  331. name = "Flexible"
  332. model = "test/model"
  333. "#;
  334. let recipe: Recipe = toml::from_str(toml_str).unwrap();
  335. assert_eq!(recipe.required_nodes, None);
  336. assert_eq!(recipe.pipeline_parallel_size, None);
  337. }
  338. #[test]
  339. fn test_config_deserialize() {
  340. let toml_str = r#"
  341. ssh_user = "admin"
  342. ssh_key_path = "/home/admin/.ssh/id_rsa"
  343. [[clusters]]
  344. name = "prod"
  345. master_ib_ip = "10.0.0.1"
  346. nodes = [
  347. { name = "node1", ssh_ip = "10.0.0.10", ib_ip = "10.0.0.1", rank = 0 },
  348. { name = "node2", ssh_ip = "10.0.0.11", ib_ip = "10.0.0.2", rank = 1 },
  349. ]
  350. [nccl."prod"]
  351. ib_hca = "mlx5_0"
  352. socket_ifname = "eth0"
  353. ib_gid_index = 3
  354. ib_disable = 0
  355. "#;
  356. let config: Config = toml::from_str(toml_str).unwrap();
  357. assert_eq!(config.ssh_user, "admin");
  358. assert_eq!(config.clusters.len(), 1);
  359. assert_eq!(config.clusters[0].nodes.len(), 2);
  360. assert_eq!(config.clusters[0].nodes[0].host_key, None);
  361. assert!(config.nccl.contains_key("prod"));
  362. assert_eq!(config.nccl["prod"].ib_hca, "mlx5_0");
  363. }
  364. #[test]
  365. fn test_config_deserialize_with_host_key() {
  366. let toml_str = r#"
  367. ssh_user = "admin"
  368. ssh_key_path = "/path/to/key"
  369. [[clusters]]
  370. name = "secure"
  371. master_ib_ip = "1.1.1.1"
  372. nodes = [
  373. { name = "n1", ssh_ip = "2.2.2.2", ib_ip = "1.1.1.1", rank = 0, host_key = "SHA256:abc123" },
  374. ]
  375. "#;
  376. let config: Config = toml::from_str(toml_str).unwrap();
  377. assert_eq!(
  378. config.clusters[0].nodes[0].host_key,
  379. Some("SHA256:abc123".to_string())
  380. );
  381. }
  382. #[test]
  383. fn test_select_cluster_by_name() {
  384. let mut config = Config {
  385. ssh_user: "user".to_string(),
  386. ssh_key_path: "/path/to/key".into(),
  387. clusters: vec![test_cluster()],
  388. nccl: HashMap::new(),
  389. hf_cache_dir: None,
  390. init_port: 25000,
  391. before_serve: HashMap::new(),
  392. exec_aliases: HashMap::new(),
  393. home_assistant: None,
  394. };
  395. config.nccl.insert("test-cluster".to_string(), test_nccl());
  396. let (cluster, nccl) = select_cluster(&config, "test-cluster", None).unwrap();
  397. assert_eq!(cluster.name, "test-cluster");
  398. assert_eq!(nccl.ib_hca, "rocep1s0f0");
  399. }
  400. #[test]
  401. fn test_select_cluster_auto() {
  402. let mut config = Config {
  403. ssh_user: "user".to_string(),
  404. ssh_key_path: "/path/to/key".into(),
  405. clusters: vec![test_cluster()],
  406. nccl: HashMap::new(),
  407. hf_cache_dir: None,
  408. init_port: 25000,
  409. before_serve: HashMap::new(),
  410. exec_aliases: HashMap::new(),
  411. home_assistant: None,
  412. };
  413. config.nccl.insert("test-cluster".to_string(), test_nccl());
  414. let (cluster, nccl) = select_cluster(&config, "auto", None).unwrap();
  415. assert_eq!(cluster.name, "test-cluster");
  416. assert_eq!(nccl.ib_hca, "rocep1s0f0");
  417. }
  418. #[test]
  419. fn test_select_cluster_auto_no_match() {
  420. let config = test_config();
  421. let result = select_cluster(&config, "auto", None);
  422. assert!(result.is_err());
  423. }
  424. #[test]
  425. fn test_select_cluster_not_found() {
  426. let mut config = test_config();
  427. config.clusters = vec![];
  428. let result = select_cluster(&config, "nonexistent", None);
  429. assert!(result.is_err());
  430. }
  431. #[test]
  432. fn test_select_cluster_by_name_found() {
  433. let mut nccl_map = HashMap::new();
  434. nccl_map.insert("test-cluster".to_string(), test_nccl());
  435. let mut config = test_config();
  436. config.nccl = nccl_map;
  437. let cluster = select_cluster_by_name(&config, "test-cluster").unwrap();
  438. assert_eq!(cluster.name, "test-cluster");
  439. assert_eq!(cluster.nodes.len(), 2);
  440. }
  441. #[test]
  442. fn test_select_cluster_by_name_not_found() {
  443. let config = test_config();
  444. let result = select_cluster_by_name(&config, "no-such-cluster");
  445. assert!(result.is_err());
  446. }
  447. #[test]
  448. fn test_select_cluster_by_prefix() {
  449. let mut config = test_config();
  450. config.nccl.insert("test-cluster".to_string(), test_nccl());
  451. // Unique prefix works
  452. let cluster = select_cluster_by_name(&config, "test-c").unwrap();
  453. assert_eq!(cluster.name, "test-cluster");
  454. // No match errors
  455. let result = select_cluster_by_name(&config, "no-such");
  456. assert!(result.is_err());
  457. }
  458. #[test]
  459. fn test_russian_pluralization() {
  460. let _prev = fluent_i18n::set_locale(Some("ru")).ok();
  461. let s1 = fluent_i18n::t!("cluster-selected", { "name" => "c", "nodes" => 1u64 });
  462. assert!(s1.contains("1 узел"), "got: {}", s1);
  463. let s2 = fluent_i18n::t!("cluster-selected", { "name" => "c", "nodes" => 2u64 });
  464. assert!(s2.contains("2 узла"), "got: {}", s2);
  465. let s5 = fluent_i18n::t!("cluster-selected", { "name" => "c", "nodes" => 5u64 });
  466. assert!(s5.contains("5 узлов"), "got: {}", s5);
  467. }
  468. #[test]
  469. fn test_default_engine_is_vllm() {
  470. let recipe: Recipe = toml::from_str("name = \"x\"\nmodel = \"m\"").unwrap();
  471. assert_eq!(recipe.engine, "vllm");
  472. }
  473. #[test]
  474. fn test_default_distributed_backend_is_mp() {
  475. let recipe: Recipe = toml::from_str("name = \"x\"\nmodel = \"m\"").unwrap();
  476. assert_eq!(recipe.distributed_backend, "mp");
  477. }
  478. // ── Default config dir ─────────────────────────────────────
  479. #[test]
  480. fn default_config_dir_appends_sm121_launcher() {
  481. let Some(dir) = default_config_dir() else {
  482. // No config dir on this system; nothing to assert.
  483. return;
  484. };
  485. assert!(dir.ends_with("sm121-launcher"));
  486. assert!(dir.parent().is_some());
  487. }
  488. }