|
|
@@ -1,212 +1,38 @@
|
|
|
-use anyhow::{Context, Result};
|
|
|
-use clap::{Parser, Subcommand};
|
|
|
-use serde::Deserialize;
|
|
|
-use std::collections::HashMap;
|
|
|
-use tokio::fs;
|
|
|
-use tokio::time::{Duration, sleep};
|
|
|
-use tracing::info;
|
|
|
+use anyhow::Result;
|
|
|
+use clap::Parser;
|
|
|
|
|
|
+mod cli;
|
|
|
+mod config;
|
|
|
+mod docker;
|
|
|
+mod doctor;
|
|
|
mod health;
|
|
|
mod ssh;
|
|
|
mod tmux;
|
|
|
|
|
|
-use crate::ssh::*;
|
|
|
-
|
|
|
-#[derive(Parser)]
|
|
|
-#[command(author, version, about = "DGX Spark Launcher для vLLM и SGLang")]
|
|
|
-struct Cli {
|
|
|
- #[arg(long, short, default_value = "config.toml")]
|
|
|
- config: String,
|
|
|
-
|
|
|
- #[arg(long, short, help = "Показать команды без выполнения")]
|
|
|
- dry_run: bool,
|
|
|
-
|
|
|
- #[command(subcommand)]
|
|
|
- command: Commands,
|
|
|
-}
|
|
|
-
|
|
|
-#[derive(Subcommand)]
|
|
|
-enum Commands {
|
|
|
- /// Запустить кластер
|
|
|
- Serve {
|
|
|
- #[arg(long, required = true)]
|
|
|
- recipe: String,
|
|
|
-
|
|
|
- #[arg(long, default_value = "auto")]
|
|
|
- cluster: String,
|
|
|
- },
|
|
|
-
|
|
|
- /// Показать логи контейнеров
|
|
|
- Logs {
|
|
|
- #[arg(long, required = true)]
|
|
|
- recipe: String,
|
|
|
-
|
|
|
- #[arg(long, action = clap::ArgAction::SetTrue)]
|
|
|
- follow: bool,
|
|
|
- },
|
|
|
-
|
|
|
- /// Остановить кластер
|
|
|
- Stop {
|
|
|
- #[arg(long, required = true)]
|
|
|
- recipe: String,
|
|
|
- },
|
|
|
-
|
|
|
- /// Проверить здоровье кластера
|
|
|
- Health {
|
|
|
- #[arg(
|
|
|
- long,
|
|
|
- help = "Имя кластера (например, 4nodes-switch). Если не указано - проверяет все."
|
|
|
- )]
|
|
|
- cluster: Option<String>,
|
|
|
- #[arg(long, help = "Показать хост-ключи узлов (для добавления в config.toml)")]
|
|
|
- fingerprint: bool,
|
|
|
- },
|
|
|
-
|
|
|
- /// Проверить и исправить рецепт (doctor mode)
|
|
|
- Doctor {
|
|
|
- #[arg(long, required = true, help = "Имя рецепта для проверки")]
|
|
|
- recipe: String,
|
|
|
- #[arg(long, default_value = "auto", help = "Кластер (если не auto — использует recipe.cluster)")]
|
|
|
- cluster: String,
|
|
|
- #[arg(long, help = "Автоматически исправить проблемы (создать Dockerfile + собрать образ)")]
|
|
|
- repair: bool,
|
|
|
- },
|
|
|
-
|
|
|
- /// Управление tmux-сессиями (альтернатива Docker)
|
|
|
- #[command(subcommand)]
|
|
|
- Tmux(TmuxCommands),
|
|
|
-}
|
|
|
-
|
|
|
-#[derive(Subcommand)]
|
|
|
-pub enum TmuxCommands {
|
|
|
- /// Запустить кластер через tmux
|
|
|
- Serve {
|
|
|
- #[arg(long, required = true)]
|
|
|
- recipe: String,
|
|
|
- #[arg(long, default_value = "auto")]
|
|
|
- cluster: String,
|
|
|
- },
|
|
|
- /// Остановить tmux-сессии
|
|
|
- Stop {
|
|
|
- #[arg(long, required = true)]
|
|
|
- recipe: String,
|
|
|
- #[arg(long, default_value = "auto")]
|
|
|
- cluster: String,
|
|
|
- },
|
|
|
- /// Статус tmux-сессий (подробный)
|
|
|
- Status {
|
|
|
- #[arg(long, required = true)]
|
|
|
- recipe: String,
|
|
|
- #[arg(long, default_value = "auto")]
|
|
|
- cluster: String,
|
|
|
- },
|
|
|
- /// Показать логи tmux-сессий
|
|
|
- Logs {
|
|
|
- #[arg(long, required = true)]
|
|
|
- recipe: String,
|
|
|
- #[arg(long, action = clap::ArgAction::SetTrue)]
|
|
|
- follow: bool,
|
|
|
- #[arg(long, default_value = "auto")]
|
|
|
- cluster: String,
|
|
|
- },
|
|
|
- /// Подключиться к tmux-сессии головного узла
|
|
|
- Attach {
|
|
|
- #[arg(long, required = true)]
|
|
|
- recipe: String,
|
|
|
- #[arg(long, default_value = "auto")]
|
|
|
- cluster: String,
|
|
|
- },
|
|
|
-}
|
|
|
-
|
|
|
-#[derive(Clone, Deserialize)]
|
|
|
-struct Node {
|
|
|
- name: String,
|
|
|
- ssh_ip: String,
|
|
|
- ib_ip: String,
|
|
|
- rank: u32,
|
|
|
- #[serde(default)]
|
|
|
- host_key: Option<String>,
|
|
|
-}
|
|
|
-
|
|
|
-#[derive(Clone, Deserialize)]
|
|
|
-struct NcclConfig {
|
|
|
- ib_hca: String,
|
|
|
- socket_ifname: String,
|
|
|
- ib_gid_index: u32,
|
|
|
- ib_disable: u32,
|
|
|
-}
|
|
|
-
|
|
|
-#[derive(Clone, Deserialize)]
|
|
|
-struct Cluster {
|
|
|
- name: String,
|
|
|
- master_ib_ip: String,
|
|
|
- nodes: Vec<Node>,
|
|
|
-}
|
|
|
-
|
|
|
-#[derive(Clone, Deserialize)]
|
|
|
-struct Config {
|
|
|
- ssh_user: String,
|
|
|
- ssh_key_path: String,
|
|
|
- clusters: Vec<Cluster>,
|
|
|
- #[serde(default)]
|
|
|
- nccl: HashMap<String, NcclConfig>,
|
|
|
-}
|
|
|
-
|
|
|
-#[derive(Clone, Deserialize)]
|
|
|
-struct Recipe {
|
|
|
- #[serde(default = "default_engine")]
|
|
|
- engine: String,
|
|
|
- name: String,
|
|
|
- #[serde(default)]
|
|
|
- docker_image: String,
|
|
|
- #[serde(default)]
|
|
|
- model: String,
|
|
|
- #[serde(default)]
|
|
|
- port: u16,
|
|
|
- #[serde(default)]
|
|
|
- required_nodes: Option<u32>,
|
|
|
- #[serde(default)]
|
|
|
- pipeline_parallel_size: Option<u32>,
|
|
|
- #[serde(default)]
|
|
|
- args: Vec<String>,
|
|
|
- #[serde(default)]
|
|
|
- env: HashMap<String, String>,
|
|
|
- #[serde(default)]
|
|
|
- cluster: Option<String>,
|
|
|
- #[serde(default)]
|
|
|
- entrypoint: Vec<String>,
|
|
|
- #[serde(default = "default_distributed_backend")]
|
|
|
- distributed_backend: String,
|
|
|
-}
|
|
|
-
|
|
|
-fn default_engine() -> String {
|
|
|
- "vllm".to_string()
|
|
|
-}
|
|
|
-
|
|
|
-fn default_distributed_backend() -> String {
|
|
|
- "mp".to_string()
|
|
|
-}
|
|
|
+use crate::cli::{Cli, Commands};
|
|
|
+use crate::config::{Config, Cluster, select_cluster_by_name, resolve_cluster_for_recipe};
|
|
|
+use crate::ssh::fetch_host_key;
|
|
|
|
|
|
#[tokio::main]
|
|
|
async fn main() -> Result<()> {
|
|
|
tracing_subscriber::fmt::init();
|
|
|
let cli = Cli::parse();
|
|
|
|
|
|
- let config: Config = toml::from_str(&fs::read_to_string(&cli.config).await?)?;
|
|
|
-
|
|
|
+ let raw_config = tokio::fs::read_to_string(&cli.config).await?;
|
|
|
+ let config: Config = toml::from_str(&raw_config)?;
|
|
|
let dry_run = cli.dry_run;
|
|
|
|
|
|
match cli.command {
|
|
|
Commands::Serve { recipe, cluster } => {
|
|
|
- serve(&config, &recipe, &cluster, dry_run).await?;
|
|
|
+ docker::serve(&config, &recipe, &cluster, dry_run).await?;
|
|
|
}
|
|
|
Commands::Logs { recipe, follow } => {
|
|
|
- let (cluster, recipe_name) = get_cluster_for_recipe(&config, &recipe).await?;
|
|
|
- show_logs(&config, &cluster, &recipe_name, follow, dry_run).await?;
|
|
|
+ let (cluster, recipe) = resolve_cluster_for_recipe(&config, &recipe).await?;
|
|
|
+ docker::show_logs(&config, &cluster, &recipe.name, follow, dry_run).await?;
|
|
|
}
|
|
|
Commands::Stop { recipe } => {
|
|
|
- let (cluster, recipe_name) = get_cluster_for_recipe(&config, &recipe).await?;
|
|
|
- stop_cluster(&config, &cluster, &recipe_name, dry_run).await?;
|
|
|
+ let (cluster, recipe) = resolve_cluster_for_recipe(&config, &recipe).await?;
|
|
|
+ docker::stop_cluster(&config, &cluster, &recipe.name, dry_run).await?;
|
|
|
}
|
|
|
Commands::Health {
|
|
|
cluster: cluster_name,
|
|
|
@@ -221,8 +47,14 @@ async fn main() -> Result<()> {
|
|
|
println!("\n=== Кластер: {} ===", cluster.name);
|
|
|
for node in &cluster.nodes {
|
|
|
match fetch_host_key(&config, node).await {
|
|
|
- Ok(fp) => println!("{} ({}) -> host_key = \"{}\"", node.name, node.ssh_ip, fp),
|
|
|
- Err(e) => println!("{} ({}) -> error: {}", node.name, node.ssh_ip, e),
|
|
|
+ Ok(fp) => println!(
|
|
|
+ "{} ({}) -> host_key = \"{}\"",
|
|
|
+ node.name, node.ssh_ip, fp
|
|
|
+ ),
|
|
|
+ Err(e) => println!(
|
|
|
+ "{} ({}) -> error: {}",
|
|
|
+ node.name, node.ssh_ip, e
|
|
|
+ ),
|
|
|
}
|
|
|
}
|
|
|
}
|
|
|
@@ -230,1145 +62,27 @@ async fn main() -> Result<()> {
|
|
|
match cluster_name {
|
|
|
Some(ref name) => {
|
|
|
let cluster = select_cluster_by_name(&config, name)?;
|
|
|
- health_check(&config, &cluster, dry_run).await?;
|
|
|
+ docker::health_check(&config, &cluster, dry_run).await?;
|
|
|
}
|
|
|
None => {
|
|
|
for cluster in &config.clusters {
|
|
|
- health_check(&config, cluster, dry_run).await?;
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
- },
|
|
|
- Commands::Doctor { recipe, cluster, repair } => {
|
|
|
- doctor(&config, &recipe, &cluster, repair, dry_run).await?;
|
|
|
- },
|
|
|
- Commands::Tmux(tmux_cmd) => {
|
|
|
- tmux::handle_tmux(&config, tmux_cmd, dry_run).await?;
|
|
|
- },
|
|
|
- }
|
|
|
-
|
|
|
- Ok(())
|
|
|
-}
|
|
|
-
|
|
|
-// ====================== Serve ======================
|
|
|
-async fn serve(
|
|
|
- config: &Config,
|
|
|
- recipe_name: &str,
|
|
|
- cluster_name: &str,
|
|
|
- dry_run: bool,
|
|
|
-) -> Result<()> {
|
|
|
- let recipe_path = format!("recipes/{}/recipe.toml", recipe_name);
|
|
|
- let recipe: Recipe = toml::from_str(&fs::read_to_string(&recipe_path).await?)?;
|
|
|
-
|
|
|
- info!("🚀 Запуск рецепта: {} ({}) [backend: {}]", recipe.name, recipe.engine, recipe.distributed_backend);
|
|
|
-
|
|
|
- let cluster_choice = if cluster_name != "auto" {
|
|
|
- cluster_name.to_string()
|
|
|
- } else if let Some(ref rc) = recipe.cluster {
|
|
|
- rc.clone()
|
|
|
- } else {
|
|
|
- "auto".to_string()
|
|
|
- };
|
|
|
- let (cluster, nccl) = select_cluster(config, &cluster_choice)?;
|
|
|
-
|
|
|
- if let Some(required) = recipe.required_nodes {
|
|
|
- let actual = cluster.nodes.len() as u32;
|
|
|
- if actual != required {
|
|
|
- anyhow::bail!(
|
|
|
- "Recipe '{}' requires {} nodes, but cluster '{}' has {} nodes",
|
|
|
- recipe.name, required, cluster.name, actual
|
|
|
- );
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- // Check for repaired image
|
|
|
- let image = try_repaired_image(config, &cluster, &recipe).await?;
|
|
|
- let mut recipe = recipe;
|
|
|
- if image != recipe.docker_image {
|
|
|
- recipe.docker_image = image.clone();
|
|
|
- }
|
|
|
-
|
|
|
- start_cluster(config, &cluster, nccl, recipe_name, &recipe, dry_run).await?;
|
|
|
-
|
|
|
- if !dry_run {
|
|
|
- info!("\n📜 Последние логи контейнеров:");
|
|
|
- show_logs(config, &cluster, recipe_name, false, dry_run).await?;
|
|
|
- }
|
|
|
-
|
|
|
- Ok(())
|
|
|
-}
|
|
|
-
|
|
|
-// ====================== Логи ======================
|
|
|
-async fn show_logs(config: &Config, cluster: &Cluster, recipe_name: &str, follow: bool, dry_run: bool) -> Result<()> {
|
|
|
- if follow {
|
|
|
- let head_node = cluster.nodes.first().context("No nodes in cluster")?;
|
|
|
- info!("📜 Following logs for '{}' (head node)...", head_node.name);
|
|
|
- let cmd = format!(
|
|
|
- "docker ps -q --filter \"label=sm121.recipe={}\" --filter \"label=sm121.rank=0\" | xargs -r docker logs -f",
|
|
|
- recipe_name
|
|
|
- );
|
|
|
- run_ssh_follow(config, head_node, cmd, dry_run).await?;
|
|
|
- } else {
|
|
|
- let flag = "--tail=100";
|
|
|
- info!(
|
|
|
- "📜 Логи кластера '{}' ({} узлов):",
|
|
|
- cluster.name,
|
|
|
- cluster.nodes.len()
|
|
|
- );
|
|
|
-
|
|
|
- let mut handles = Vec::new();
|
|
|
- for node in &cluster.nodes {
|
|
|
- let cmd = format!(
|
|
|
- "docker ps -q --filter \"label=sm121.recipe={}\" --filter \"label=sm121.rank={}\" | xargs -r docker logs {} 2>&1",
|
|
|
- recipe_name, node.rank, flag
|
|
|
- );
|
|
|
- let config = config.clone();
|
|
|
- let node = node.clone();
|
|
|
- let dry = dry_run;
|
|
|
- handles.push(tokio::spawn(async move {
|
|
|
- let output = run_ssh_with_retry(&config, &node, cmd, dry, 1).await;
|
|
|
- (node.name, output)
|
|
|
- }));
|
|
|
- }
|
|
|
-
|
|
|
- let results = futures::future::join_all(handles).await;
|
|
|
- for result in results {
|
|
|
- if let Ok((name, Ok(output))) = result
|
|
|
- && !output.trim().is_empty()
|
|
|
- {
|
|
|
- println!("\n--- {} ---", name);
|
|
|
- println!("{}", output);
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
- Ok(())
|
|
|
-}
|
|
|
-
|
|
|
-// ====================== Stop ======================
|
|
|
-async fn stop_cluster(config: &Config, cluster: &Cluster, recipe_name: &str, dry_run: bool) -> Result<()> {
|
|
|
- info!("🛑 Останавливаем кластер '{}'...", cluster.name);
|
|
|
- let mut failures = Vec::new();
|
|
|
-
|
|
|
- for node in &cluster.nodes {
|
|
|
- let stop_cmd = format!(
|
|
|
- "docker ps -q --filter \"label=sm121.recipe={}\" | xargs -r docker stop",
|
|
|
- recipe_name
|
|
|
- );
|
|
|
- if let Err(e) = run_ssh(config, node, stop_cmd, dry_run).await {
|
|
|
- tracing::warn!("Stop failed on {}: {}", node.name, e);
|
|
|
- failures.push((node.name.clone(), "stop".to_string(), e));
|
|
|
- }
|
|
|
-
|
|
|
- let rm_cmd = format!(
|
|
|
- "docker ps -qa --filter \"label=sm121.recipe={}\" | xargs -r docker rm",
|
|
|
- recipe_name
|
|
|
- );
|
|
|
- if let Err(e) = run_ssh(config, node, rm_cmd, dry_run).await {
|
|
|
- tracing::warn!("Remove failed on {}: {}", node.name, e);
|
|
|
- failures.push((node.name.clone(), "rm".to_string(), e));
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- if !failures.is_empty() {
|
|
|
- for (name, action, e) in &failures {
|
|
|
- tracing::warn!("Failed to {} on {}: {}", action, name, e);
|
|
|
- }
|
|
|
- anyhow::bail!("Failed to stop containers on {} node(s)", failures.len());
|
|
|
- }
|
|
|
-
|
|
|
- info!("✅ Все контейнеры остановлены.");
|
|
|
- Ok(())
|
|
|
-}
|
|
|
-
|
|
|
-// ====================== Health ======================
|
|
|
-async fn health_check(config: &Config, cluster: &Cluster, dry_run: bool) -> Result<()> {
|
|
|
- println!("\n=== Кластер: {} ===", cluster.name);
|
|
|
-
|
|
|
- let mut handles = Vec::new();
|
|
|
- for node in &cluster.nodes {
|
|
|
- let config = config.clone();
|
|
|
- let node = node.clone();
|
|
|
- let dry = dry_run;
|
|
|
- handles.push(tokio::spawn(async move {
|
|
|
- let result = run_ssh_with_retry(
|
|
|
- &config,
|
|
|
- &node,
|
|
|
- "docker ps --filter \"label=sm121\" --format '{{.Names}} {{.Status}}' 2>&1".to_string(),
|
|
|
- dry,
|
|
|
- 1,
|
|
|
- ).await;
|
|
|
- (node.name, node.ib_ip, result)
|
|
|
- }));
|
|
|
- }
|
|
|
-
|
|
|
- let mut all_healthy = true;
|
|
|
- let results = futures::future::join_all(handles).await;
|
|
|
- for result in results {
|
|
|
- match result {
|
|
|
- Ok((name, ib_ip, Ok(docker_check))) => {
|
|
|
- print!("{} ({}): ", name, ib_ip);
|
|
|
- if docker_check.is_empty() && dry_run {
|
|
|
- continue;
|
|
|
- }
|
|
|
- if docker_check.starts_with("docker:") || docker_check.contains("permission denied") {
|
|
|
- println!("✗ Docker недоступен");
|
|
|
- if let Some(line) = docker_check.lines().next() {
|
|
|
- println!(" └─ {}", line);
|
|
|
- }
|
|
|
- all_healthy = false;
|
|
|
- } else {
|
|
|
- let containers: Vec<&str> = docker_check.lines().filter(|l| !l.is_empty()).collect();
|
|
|
- if containers.is_empty() {
|
|
|
- println!("✗ Нет контейнеров");
|
|
|
- all_healthy = false;
|
|
|
- } else {
|
|
|
- println!("✓ {} контейнер(ов)", containers.len());
|
|
|
- for c in containers {
|
|
|
- println!(" └─ {}", c);
|
|
|
+ docker::health_check(&config, cluster, dry_run).await?;
|
|
|
}
|
|
|
}
|
|
|
}
|
|
|
}
|
|
|
- Ok((name, ib_ip, Err(e))) => {
|
|
|
- println!("{} ({}): ✗ SSH ошибка: {}", name, ib_ip, e);
|
|
|
- all_healthy = false;
|
|
|
- }
|
|
|
- Err(join_err) => {
|
|
|
- println!("✗ Task join error: {}", join_err);
|
|
|
- all_healthy = false;
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- if !dry_run {
|
|
|
- if all_healthy {
|
|
|
- info!("✅ Кластер здоров!");
|
|
|
- } else {
|
|
|
- info!("⚠️ Кластер имеет проблемы");
|
|
|
- }
|
|
|
- }
|
|
|
- Ok(())
|
|
|
-}
|
|
|
-
|
|
|
-fn select_cluster_by_name(config: &Config, cluster_name: &str) -> Result<Cluster> {
|
|
|
- let cluster = config
|
|
|
- .clusters
|
|
|
- .iter()
|
|
|
- .find(|c| c.name == cluster_name)
|
|
|
- .context(format!("Кластер {} не найден", cluster_name))?;
|
|
|
- Ok(cluster.clone())
|
|
|
-}
|
|
|
-
|
|
|
-// ====================== Вспомогательные функции ======================
|
|
|
-async fn get_cluster_for_recipe(config: &Config, recipe_name: &str) -> Result<(Cluster, String)> {
|
|
|
- let recipe_path = format!("recipes/{}/recipe.toml", recipe_name);
|
|
|
- let recipe: Recipe = toml::from_str(&fs::read_to_string(&recipe_path).await?)?;
|
|
|
- let cluster_choice = recipe.cluster.as_deref().unwrap_or("auto");
|
|
|
- let (cluster, _) = select_cluster(config, cluster_choice)?;
|
|
|
- Ok((cluster, recipe_name.to_string()))
|
|
|
-}
|
|
|
-
|
|
|
-fn select_cluster<'c>(config: &'c Config, cluster_name: &str) -> Result<(Cluster, &'c NcclConfig)> {
|
|
|
- if cluster_name != "auto" {
|
|
|
- let cluster = config
|
|
|
- .clusters
|
|
|
- .iter()
|
|
|
- .find(|c| c.name == cluster_name)
|
|
|
- .context(format!("Кластер {} не найден", cluster_name))?;
|
|
|
- let nccl = config
|
|
|
- .nccl
|
|
|
- .get(&cluster.name)
|
|
|
- .context("NCCL конфиг не найден")?;
|
|
|
-
|
|
|
- return Ok((cluster.clone(), nccl));
|
|
|
- }
|
|
|
-
|
|
|
- for cluster in &config.clusters {
|
|
|
- if let Some(nccl) = config.nccl.get(&cluster.name) {
|
|
|
- info!("✅ Автоподбор кластера: {}", cluster.name);
|
|
|
- return Ok((cluster.clone(), nccl));
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- anyhow::bail!("Не удалось подобрать подходящий кластер");
|
|
|
-}
|
|
|
-
|
|
|
-async fn start_cluster(
|
|
|
- config: &Config,
|
|
|
- cluster: &Cluster,
|
|
|
- nccl: &NcclConfig,
|
|
|
- recipe_name: &str,
|
|
|
- recipe: &Recipe,
|
|
|
- dry_run: bool,
|
|
|
-) -> Result<()> {
|
|
|
- let nnodes = cluster.nodes.len() as u32;
|
|
|
- let pp_size = recipe.pipeline_parallel_size.unwrap_or(0);
|
|
|
- let pp_size = if pp_size == 0 { nnodes } else { pp_size };
|
|
|
-
|
|
|
- let recipe_clone = (*recipe).clone();
|
|
|
-
|
|
|
- let mut handles = Vec::new();
|
|
|
- for node in &cluster.nodes {
|
|
|
- let is_head = node.rank == 0;
|
|
|
- let cleanup_cmd = format!("docker rm -f sm121_{}_{} 2>/dev/null; true", recipe_name, node.rank);
|
|
|
-
|
|
|
- let config = config.clone();
|
|
|
- let cluster = cluster.clone();
|
|
|
- let nccl = nccl.clone();
|
|
|
- let node = node.clone();
|
|
|
- let recipe = recipe_clone.clone();
|
|
|
- let recipe_name = recipe_name.to_string();
|
|
|
- let dry = dry_run;
|
|
|
- let engine = recipe.engine.clone();
|
|
|
-
|
|
|
- handles.push(tokio::spawn(async move {
|
|
|
- // Cleanup leftover container
|
|
|
- if !dry {
|
|
|
- let _ = run_ssh(&config, &node, cleanup_cmd, dry).await;
|
|
|
- }
|
|
|
-
|
|
|
- let docker_cmd = build_docker_command(
|
|
|
- &cluster, &nccl, &recipe, &recipe_name, &node, is_head, pp_size, nnodes,
|
|
|
- );
|
|
|
- let full_cmd = docker_cmd.join(" ");
|
|
|
-
|
|
|
- let result = run_ssh_with_retry(&config, &node, full_cmd, dry, 2).await;
|
|
|
- (node.name, node.rank, engine, result)
|
|
|
- }));
|
|
|
- }
|
|
|
-
|
|
|
- let results = futures::future::join_all(handles).await;
|
|
|
- for result in results {
|
|
|
- match result {
|
|
|
- Ok((name, rank, engine, Ok(_output))) => {
|
|
|
- info!("✅ {} запущен на {} (rank {})", engine.to_uppercase(), name, rank);
|
|
|
- }
|
|
|
- Ok((name, rank, engine, Err(e))) => {
|
|
|
- anyhow::bail!("Failed to start {} on {} (rank {}): {}", engine.to_uppercase(), name, rank, e);
|
|
|
- }
|
|
|
- Err(join_err) => {
|
|
|
- anyhow::bail!("Task join error: {}", join_err);
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- if !dry_run {
|
|
|
- sleep(Duration::from_secs(10)).await;
|
|
|
- info!(
|
|
|
- "🎉 Кластер успешно запущен! API → http://{}:{}",
|
|
|
- cluster.master_ib_ip, recipe.port
|
|
|
- );
|
|
|
- }
|
|
|
- Ok(())
|
|
|
-}
|
|
|
-
|
|
|
-#[allow(clippy::too_many_arguments)]
|
|
|
-fn build_docker_command(
|
|
|
- cluster: &Cluster,
|
|
|
- nccl: &NcclConfig,
|
|
|
- recipe: &Recipe,
|
|
|
- recipe_name: &str,
|
|
|
- node: &Node,
|
|
|
- is_head: bool,
|
|
|
- pp_size: u32,
|
|
|
- nnodes: u32,
|
|
|
-) -> Vec<String> {
|
|
|
- let mut cmd = vec![
|
|
|
- "docker run -d --gpus all --ipc=host --privileged --network=host".to_string(),
|
|
|
- format!("-e NCCL_IB_DISABLE={}", nccl.ib_disable),
|
|
|
- format!("-e NCCL_IB_HCA={}", nccl.ib_hca),
|
|
|
- format!("-e NCCL_SOCKET_IFNAME={}", nccl.socket_ifname),
|
|
|
- format!("-e NCCL_IB_GID_INDEX={}", nccl.ib_gid_index),
|
|
|
- format!("-e VLLM_HOST_IP={}", node.ib_ip),
|
|
|
- ];
|
|
|
-
|
|
|
- for (k, v) in &recipe.env {
|
|
|
- cmd.push(format!("-e {}={}", k, v));
|
|
|
- }
|
|
|
-
|
|
|
- if is_head {
|
|
|
- cmd.push(format!("-p {}:{}" , recipe.port, recipe.port));
|
|
|
- }
|
|
|
-
|
|
|
- // Container name and labels
|
|
|
- cmd.push(format!("--name sm121_{}_{}", recipe_name, node.rank));
|
|
|
- cmd.push(format!("--label sm121.recipe={}", recipe_name));
|
|
|
- cmd.push(format!("--label sm121.rank={}", node.rank));
|
|
|
-
|
|
|
- // Ray backend: need RAY_ADDRESS + custom entrypoint that starts Ray before vLLM
|
|
|
- if recipe.distributed_backend == "ray" {
|
|
|
- cmd.push(format!("-e RAY_ADDRESS={}:6379", cluster.master_ib_ip));
|
|
|
- cmd.push(format!("-e RAY_PORT={}", recipe.port));
|
|
|
- }
|
|
|
-
|
|
|
- cmd.push(recipe.docker_image.clone());
|
|
|
-
|
|
|
- // Engine entrypoint (optional — empty for vLLM)
|
|
|
- if !recipe.entrypoint.is_empty() {
|
|
|
- cmd.extend(recipe.entrypoint.clone());
|
|
|
- }
|
|
|
-
|
|
|
- cmd.push(recipe.model.clone());
|
|
|
- cmd.extend(recipe.args.clone());
|
|
|
-
|
|
|
- // Backend-specific flags
|
|
|
- match recipe.distributed_backend.as_str() {
|
|
|
- "ray" => {
|
|
|
- cmd.extend(vec![
|
|
|
- "--tensor-parallel-size".to_string(),
|
|
|
- "1".to_string(),
|
|
|
- "--pipeline-parallel-size".to_string(),
|
|
|
- pp_size.to_string(),
|
|
|
- "--nnodes".to_string(),
|
|
|
- nnodes.to_string(),
|
|
|
- "--node-rank".to_string(),
|
|
|
- node.rank.to_string(),
|
|
|
- "--master-addr".to_string(),
|
|
|
- cluster.master_ib_ip.clone(),
|
|
|
- "--distributed-executor-backend".to_string(),
|
|
|
- "ray".to_string(),
|
|
|
- ]);
|
|
|
- }
|
|
|
- "mp" => {
|
|
|
- cmd.extend(vec![
|
|
|
- "--tensor-parallel-size".to_string(),
|
|
|
- "1".to_string(),
|
|
|
- "--pipeline-parallel-size".to_string(),
|
|
|
- pp_size.to_string(),
|
|
|
- "--nnodes".to_string(),
|
|
|
- nnodes.to_string(),
|
|
|
- "--node-rank".to_string(),
|
|
|
- node.rank.to_string(),
|
|
|
- "--master-addr".to_string(),
|
|
|
- cluster.master_ib_ip.clone(),
|
|
|
- "--distributed-executor-backend".to_string(),
|
|
|
- "mp".to_string(),
|
|
|
- ]);
|
|
|
- }
|
|
|
- _ => {}
|
|
|
- }
|
|
|
-
|
|
|
- cmd
|
|
|
-}
|
|
|
-
|
|
|
-// ====================== Doctor ======================
|
|
|
-
|
|
|
-/// Doctor check result for a single node/container
|
|
|
-struct DoctorResult {
|
|
|
- node_name: String,
|
|
|
- checks: Vec<DoctorCheck>,
|
|
|
-}
|
|
|
-
|
|
|
-struct DoctorCheck {
|
|
|
- name: String,
|
|
|
- status: DoctorStatus,
|
|
|
- detail: String,
|
|
|
-}
|
|
|
-
|
|
|
-#[derive(PartialEq)]
|
|
|
-enum DoctorStatus {
|
|
|
- Ok,
|
|
|
- Warn,
|
|
|
- Fail,
|
|
|
-}
|
|
|
-
|
|
|
-impl std::fmt::Display for DoctorStatus {
|
|
|
- fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
|
|
- match self {
|
|
|
- DoctorStatus::Ok => write!(f, "✓"),
|
|
|
- DoctorStatus::Warn => write!(f, "⚠"),
|
|
|
- DoctorStatus::Fail => write!(f, "✗"),
|
|
|
}
|
|
|
- }
|
|
|
-}
|
|
|
-
|
|
|
-/// Run doctor checks on the cluster for a recipe
|
|
|
-async fn doctor(
|
|
|
- config: &Config,
|
|
|
- recipe_name: &str,
|
|
|
- cluster_name: &str,
|
|
|
- repair: bool,
|
|
|
- dry_run: bool,
|
|
|
-) -> Result<()> {
|
|
|
- let recipe_path = format!("recipes/{}/recipe.toml", recipe_name);
|
|
|
- let recipe: Recipe = toml::from_str(&fs::read_to_string(&recipe_path).await?)?;
|
|
|
-
|
|
|
- println!("\n🩺 Doctor: {} ({}) [backend: {}]", recipe.name, recipe.engine, recipe.distributed_backend);
|
|
|
-
|
|
|
- // Resolve cluster
|
|
|
- let cluster_choice = if cluster_name != "auto" {
|
|
|
- cluster_name.to_string()
|
|
|
- } else if let Some(ref rc) = recipe.cluster {
|
|
|
- rc.clone()
|
|
|
- } else {
|
|
|
- "auto".to_string()
|
|
|
- };
|
|
|
- let (cluster, _nccl) = select_cluster(config, &cluster_choice)?;
|
|
|
-
|
|
|
- // Check each node
|
|
|
- let mut all_results: Vec<DoctorResult> = Vec::new();
|
|
|
- let mut handles = Vec::new();
|
|
|
-
|
|
|
- for node in &cluster.nodes {
|
|
|
- let config = config.clone();
|
|
|
- let node = node.clone();
|
|
|
- let engine = recipe.engine.clone();
|
|
|
- let backend = recipe.distributed_backend.clone();
|
|
|
- let image = recipe.docker_image.clone();
|
|
|
- handles.push(tokio::spawn(async move {
|
|
|
- check_node(&config, &node, &engine, &backend, &image).await
|
|
|
- }));
|
|
|
- }
|
|
|
-
|
|
|
- let results = futures::future::join_all(handles).await;
|
|
|
- for r in results.into_iter().flatten() {
|
|
|
- all_results.push(r);
|
|
|
- }
|
|
|
-
|
|
|
- // Print results
|
|
|
- let mut has_failures = false;
|
|
|
- for result in &all_results {
|
|
|
- println!("\n--- {} ---", result.node_name);
|
|
|
- for check in &result.checks {
|
|
|
- println!(" {} {} {}", check.status, check.name, check.detail);
|
|
|
- if check.status == DoctorStatus::Fail {
|
|
|
- has_failures = true;
|
|
|
- }
|
|
|
+ Commands::Doctor {
|
|
|
+ recipe,
|
|
|
+ cluster,
|
|
|
+ repair,
|
|
|
+ } => {
|
|
|
+ doctor::doctor(&config, &recipe, &cluster, repair, dry_run).await?;
|
|
|
}
|
|
|
- }
|
|
|
-
|
|
|
- if has_failures {
|
|
|
- if repair {
|
|
|
- println!("\n🔧 Repairing image...");
|
|
|
- let repaired = run_cluster_repair(config, &cluster, &recipe, dry_run).await?;
|
|
|
- println!("✅ Repaired image: {}", repaired);
|
|
|
- } else {
|
|
|
- println!("\n⚠ Some checks failed. Run with --repair to auto-fix.");
|
|
|
- std::process::exit(1);
|
|
|
+ Commands::Tmux(tmux_cmd) => {
|
|
|
+ crate::tmux::handle_tmux(&config, tmux_cmd, dry_run).await?;
|
|
|
}
|
|
|
- } else {
|
|
|
- println!("\n✅ All checks passed!");
|
|
|
}
|
|
|
|
|
|
Ok(())
|
|
|
}
|
|
|
-
|
|
|
-/// Check a single node for required software
|
|
|
-async fn check_node(
|
|
|
- config: &Config,
|
|
|
- node: &Node,
|
|
|
- engine: &str,
|
|
|
- backend: &str,
|
|
|
- image: &str,
|
|
|
-) -> DoctorResult {
|
|
|
- let mut checks = Vec::new();
|
|
|
-
|
|
|
- // Node-level checks
|
|
|
- checks.push(node_check(config, node, "Docker", "docker --version").await);
|
|
|
- checks.push(node_check(config, node, "NVIDIA GPU", "nvidia-smi").await);
|
|
|
- checks.push(node_check(config, node, "NCCL libs", "ldconfig -p | grep -q nccl && echo OK").await);
|
|
|
-
|
|
|
- // Container-level checks
|
|
|
- checks.push(container_check(config, node, image, "Engine", format!("{} version", engine)).await);
|
|
|
-
|
|
|
- if backend == "ray" {
|
|
|
- checks.push(container_check(config, node, image, "Ray", "ray --version".to_string()).await);
|
|
|
- }
|
|
|
-
|
|
|
- // Check container entrypoint
|
|
|
- checks.push(container_entrypoint_check(config, node, image).await);
|
|
|
-
|
|
|
- DoctorResult {
|
|
|
- node_name: node.name.clone(),
|
|
|
- checks,
|
|
|
- }
|
|
|
-}
|
|
|
-
|
|
|
-/// Run a command on the node via SSH and return a check result
|
|
|
-async fn node_check(config: &Config, node: &Node, name: &str, command: &str) -> DoctorCheck {
|
|
|
- match run_ssh(config, node, command.to_string(), false).await {
|
|
|
- Ok(output) if !output.trim().is_empty() => DoctorCheck {
|
|
|
- name: name.to_string(),
|
|
|
- status: DoctorStatus::Ok,
|
|
|
- detail: format!("OK: {}", output.trim().lines().next().unwrap_or(output.trim())),
|
|
|
- },
|
|
|
- Ok(_) => DoctorCheck {
|
|
|
- name: name.to_string(),
|
|
|
- status: DoctorStatus::Fail,
|
|
|
- detail: "not found".to_string(),
|
|
|
- },
|
|
|
- Err(e) => DoctorCheck {
|
|
|
- name: name.to_string(),
|
|
|
- status: DoctorStatus::Fail,
|
|
|
- detail: format!("error: {}", e),
|
|
|
- },
|
|
|
- }
|
|
|
-}
|
|
|
-
|
|
|
-/// Run a command inside a container on the node and return a check result
|
|
|
-async fn container_check(config: &Config, node: &Node, image: &str, name: &str, command: String) -> DoctorCheck {
|
|
|
- let docker_cmd = format!(
|
|
|
- "docker run --rm {} {} 2>&1 | head -1",
|
|
|
- image, command
|
|
|
- );
|
|
|
- match run_ssh(config, node, docker_cmd, false).await {
|
|
|
- Ok(output) if !output.contains("not found") && !output.contains("No such file") => DoctorCheck {
|
|
|
- name: name.to_string(),
|
|
|
- status: DoctorStatus::Ok,
|
|
|
- detail: format!("OK: {}", output.trim()),
|
|
|
- },
|
|
|
- Ok(output) => DoctorCheck {
|
|
|
- name: name.to_string(),
|
|
|
- status: DoctorStatus::Fail,
|
|
|
- detail: format!("not found: {}", output.trim()),
|
|
|
- },
|
|
|
- Err(e) => DoctorCheck {
|
|
|
- name: name.to_string(),
|
|
|
- status: DoctorStatus::Fail,
|
|
|
- detail: format!("error: {}", e),
|
|
|
- },
|
|
|
- }
|
|
|
-}
|
|
|
-
|
|
|
-/// Check the container entrypoint
|
|
|
-async fn container_entrypoint_check(config: &Config, node: &Node, image: &str) -> DoctorCheck {
|
|
|
- let docker_cmd = format!(
|
|
|
- "docker inspect --format '{{{{json .Config.Entrypoint}}}}' {} 2>/dev/null || echo 'null'",
|
|
|
- image
|
|
|
- );
|
|
|
- match run_ssh(config, node, docker_cmd, false).await {
|
|
|
- Ok(output) => {
|
|
|
- let entrypoint = output.trim();
|
|
|
- if entrypoint == "null" || entrypoint == "[]" {
|
|
|
- DoctorCheck {
|
|
|
- name: "Entrypoint".to_string(),
|
|
|
- status: DoctorStatus::Ok,
|
|
|
- detail: "default (shell) entrypoint".to_string(),
|
|
|
- }
|
|
|
- } else {
|
|
|
- DoctorCheck {
|
|
|
- name: "Entrypoint".to_string(),
|
|
|
- status: DoctorStatus::Warn,
|
|
|
- detail: format!("custom: {}", entrypoint),
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
- Err(e) => DoctorCheck {
|
|
|
- name: "Entrypoint".to_string(),
|
|
|
- status: DoctorStatus::Fail,
|
|
|
- detail: format!("error: {}", e),
|
|
|
- },
|
|
|
- }
|
|
|
-}
|
|
|
-
|
|
|
-/// Generate Dockerfile and build repaired image on head node
|
|
|
-async fn run_cluster_repair(
|
|
|
- config: &Config,
|
|
|
- cluster: &Cluster,
|
|
|
- recipe: &Recipe,
|
|
|
- dry_run: bool,
|
|
|
-) -> Result<String> {
|
|
|
- let head_node = cluster.nodes.first().context("No nodes in cluster")?;
|
|
|
-
|
|
|
- // Generate repaired image name
|
|
|
- let repaired_tag = format!("{}.repaired", recipe.docker_image);
|
|
|
-
|
|
|
- // Generate Dockerfile
|
|
|
- let dockerfile = generate_repair_dockerfile(&repaired_tag, recipe);
|
|
|
-
|
|
|
- // Write Dockerfile locally
|
|
|
- let dockerfile_path = format!("recipes/{}/Dockerfile.repaired", "temp");
|
|
|
- fs::write(&dockerfile_path, &dockerfile).await
|
|
|
- .context("Failed to write Dockerfile")?;
|
|
|
-
|
|
|
- // SSH to head node, create Dockerfile, build image
|
|
|
- let create_dockerfile = format!(
|
|
|
- "mkdir -p /tmp/sm121_docker && cat > /tmp/sm121_docker/Dockerfile << 'DOCKERFILE_END'\n{}DOCKERFILE_END",
|
|
|
- dockerfile
|
|
|
- );
|
|
|
-
|
|
|
- if !dry_run {
|
|
|
- run_ssh(config, head_node, create_dockerfile, false).await?;
|
|
|
-
|
|
|
- let build_cmd = format!(
|
|
|
- "cd /tmp/sm121_docker && docker build -t {} .",
|
|
|
- repaired_tag
|
|
|
- );
|
|
|
- let output = run_ssh_with_retry(config, head_node, build_cmd, false, 1).await
|
|
|
- .context("Failed to build repaired image")?;
|
|
|
- println!("Build output: {}", output.lines().last().unwrap_or(&output));
|
|
|
- } else {
|
|
|
- println!(" (dry-run) Would build {} on {}", repaired_tag, head_node.name);
|
|
|
- }
|
|
|
-
|
|
|
- // Cleanup
|
|
|
- fs::remove_file(&dockerfile_path).await.ok();
|
|
|
-
|
|
|
- Ok(repaired_tag)
|
|
|
-}
|
|
|
-
|
|
|
-/// Generate a Dockerfile to repair a container
|
|
|
-fn generate_repair_dockerfile(base_image: &str, recipe: &Recipe) -> String {
|
|
|
- let mut lines = vec![format!("FROM {}", base_image)];
|
|
|
-
|
|
|
- // Install ray if needed
|
|
|
- if recipe.distributed_backend == "ray" {
|
|
|
- lines.push("RUN pip install ray[default] || pip install ray".to_string());
|
|
|
- }
|
|
|
-
|
|
|
- // Install engine if needed
|
|
|
- let engine_install = match recipe.engine.as_str() {
|
|
|
- "vllm" => "pip install vllm",
|
|
|
- "sglang" => "pip install sglang",
|
|
|
- _ => "",
|
|
|
- };
|
|
|
- if !engine_install.is_empty() {
|
|
|
- lines.push(format!("RUN {} || true", engine_install));
|
|
|
- }
|
|
|
-
|
|
|
- // Ensure entrypoint is shell-compatible
|
|
|
- lines.push("ENTRYPOINT [\"/bin/bash\", \"-c\"]".to_string());
|
|
|
-
|
|
|
- lines.join("\n")
|
|
|
-}
|
|
|
-
|
|
|
-// ====================== serve() repair image check ======================
|
|
|
-
|
|
|
-async fn try_repaired_image(config: &Config, cluster: &Cluster, recipe: &Recipe) -> Result<String> {
|
|
|
- let head_node = cluster.nodes.first().context("No nodes in cluster")?;
|
|
|
- let repaired_tag = format!("{}.repaired", recipe.docker_image);
|
|
|
-
|
|
|
- // Check if repaired image exists on head node
|
|
|
- let check_cmd = format!(
|
|
|
- "docker images -q {} | head -1",
|
|
|
- repaired_tag
|
|
|
- );
|
|
|
-
|
|
|
- match run_ssh(config, head_node, check_cmd, false).await {
|
|
|
- Ok(output) if !output.trim().is_empty() => {
|
|
|
- info!("🩺 Found repaired image: {}", repaired_tag);
|
|
|
- Ok(repaired_tag)
|
|
|
- }
|
|
|
- _ => Ok(recipe.docker_image.clone()),
|
|
|
- }
|
|
|
-}
|
|
|
-
|
|
|
-#[cfg(test)]
|
|
|
-mod tests {
|
|
|
- use super::*;
|
|
|
- use std::collections::HashMap;
|
|
|
-
|
|
|
- fn test_cluster() -> Cluster {
|
|
|
- Cluster {
|
|
|
- name: "test-cluster".to_string(),
|
|
|
- master_ib_ip: "192.168.1.1".to_string(),
|
|
|
- nodes: vec![
|
|
|
- Node {
|
|
|
- name: "node0".to_string(),
|
|
|
- ssh_ip: "10.0.0.1".to_string(),
|
|
|
- ib_ip: "192.168.1.1".to_string(),
|
|
|
- rank: 0,
|
|
|
- host_key: None,
|
|
|
- },
|
|
|
- Node {
|
|
|
- name: "node1".to_string(),
|
|
|
- ssh_ip: "10.0.0.2".to_string(),
|
|
|
- ib_ip: "192.168.1.2".to_string(),
|
|
|
- rank: 1,
|
|
|
- host_key: None,
|
|
|
- },
|
|
|
- ],
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- fn test_nccl() -> NcclConfig {
|
|
|
- NcclConfig {
|
|
|
- ib_hca: "rocep1s0f0".to_string(),
|
|
|
- socket_ifname: "enp1s0f0np0".to_string(),
|
|
|
- ib_gid_index: 3,
|
|
|
- ib_disable: 0,
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- fn test_recipe() -> Recipe {
|
|
|
- Recipe {
|
|
|
- engine: "vllm".to_string(),
|
|
|
- name: "Test Model".to_string(),
|
|
|
- docker_image: "test/image:latest".to_string(),
|
|
|
- model: "test/model".to_string(),
|
|
|
- port: 8000,
|
|
|
- required_nodes: Some(2),
|
|
|
- pipeline_parallel_size: Some(2),
|
|
|
- args: vec!["--max-model-len".to_string(), "4096".to_string()],
|
|
|
- env: HashMap::from([("NCCL_DEBUG".to_string(), "INFO".to_string())]),
|
|
|
- cluster: None,
|
|
|
- entrypoint: vec![],
|
|
|
- distributed_backend: "mp".to_string(),
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- // ── build_docker_command tests ──────────────────────────────
|
|
|
-
|
|
|
- #[test]
|
|
|
- fn test_build_docker_command_mp_backend() {
|
|
|
- let cluster = test_cluster();
|
|
|
- let nccl = test_nccl();
|
|
|
- let recipe = test_recipe();
|
|
|
- let node = &cluster.nodes[0];
|
|
|
-
|
|
|
- let cmd = build_docker_command(&cluster, &nccl, &recipe, "test-recipe", node, true, 2, 2);
|
|
|
-
|
|
|
- let joined = cmd.join(" ");
|
|
|
-
|
|
|
- // Basic docker flags
|
|
|
- assert!(joined.contains("docker run -d --gpus all --ipc=host --privileged --network=host"));
|
|
|
- // NCCL env
|
|
|
- assert!(joined.contains("-e NCCL_IB_HCA=rocep1s0f0"));
|
|
|
- assert!(joined.contains("-e NCCL_SOCKET_IFNAME=enp1s0f0np0"));
|
|
|
- // Recipe env
|
|
|
- assert!(joined.contains("-e NCCL_DEBUG=INFO"));
|
|
|
- // Port only on head
|
|
|
- assert!(joined.contains("-p 8000:8000"));
|
|
|
- // Container name & labels
|
|
|
- assert!(joined.contains("--name sm121_test-recipe_0"));
|
|
|
- assert!(joined.contains("--label sm121.recipe=test-recipe"));
|
|
|
- assert!(joined.contains("--label sm121.rank=0"));
|
|
|
- // Image & model
|
|
|
- assert!(joined.contains("test/image:latest"));
|
|
|
- assert!(joined.contains("test/model"));
|
|
|
- // Args
|
|
|
- assert!(joined.contains("--max-model-len 4096"));
|
|
|
- // MP-specific flags
|
|
|
- assert!(joined.contains("--tensor-parallel-size 1"));
|
|
|
- assert!(joined.contains("--pipeline-parallel-size 2"));
|
|
|
- assert!(joined.contains("--nnodes 2"));
|
|
|
- assert!(joined.contains("--node-rank 0"));
|
|
|
- assert!(joined.contains("--master-addr 192.168.1.1"));
|
|
|
- assert!(joined.contains("--distributed-executor-backend mp"));
|
|
|
- // No Ray env
|
|
|
- assert!(!joined.contains("RAY_ADDRESS"));
|
|
|
- }
|
|
|
-
|
|
|
- #[test]
|
|
|
- fn test_build_docker_command_worker_no_port() {
|
|
|
- let cluster = test_cluster();
|
|
|
- let nccl = test_nccl();
|
|
|
- let recipe = test_recipe();
|
|
|
- let node = &cluster.nodes[1]; // rank 1
|
|
|
-
|
|
|
- let cmd = build_docker_command(&cluster, &nccl, &recipe, "test-recipe", node, false, 2, 2);
|
|
|
- let joined = cmd.join(" ");
|
|
|
-
|
|
|
- // No port mapping on worker
|
|
|
- assert!(!joined.contains("-p 8000:8000"));
|
|
|
- // Correct rank
|
|
|
- assert!(joined.contains("--node-rank 1"));
|
|
|
- assert!(joined.contains("--label sm121.rank=1"));
|
|
|
- assert!(joined.contains("--name sm121_test-recipe_1"));
|
|
|
- }
|
|
|
-
|
|
|
- #[test]
|
|
|
- fn test_build_docker_command_ray_backend() {
|
|
|
- let cluster = test_cluster();
|
|
|
- let nccl = test_nccl();
|
|
|
- let mut recipe = test_recipe();
|
|
|
- recipe.distributed_backend = "ray".to_string();
|
|
|
- let node = &cluster.nodes[0];
|
|
|
-
|
|
|
- let cmd = build_docker_command(&cluster, &nccl, &recipe, "test-recipe", node, true, 2, 2);
|
|
|
- let joined = cmd.join(" ");
|
|
|
-
|
|
|
- // Ray-specific env
|
|
|
- assert!(joined.contains("-e RAY_ADDRESS=192.168.1.1:6379"));
|
|
|
- assert!(joined.contains("-e RAY_PORT=8000"));
|
|
|
- // Ray-specific flags
|
|
|
- assert!(joined.contains("--tensor-parallel-size 1"));
|
|
|
- assert!(joined.contains("--pipeline-parallel-size 2"));
|
|
|
- assert!(joined.contains("--nnodes 2"));
|
|
|
- assert!(joined.contains("--node-rank 0"));
|
|
|
- assert!(joined.contains("--master-addr 192.168.1.1"));
|
|
|
- assert!(joined.contains("--distributed-executor-backend ray"));
|
|
|
- }
|
|
|
-
|
|
|
- #[test]
|
|
|
- fn test_build_docker_command_auto_pp_size() {
|
|
|
- let cluster = test_cluster();
|
|
|
- let nccl = test_nccl();
|
|
|
- let mut recipe = test_recipe();
|
|
|
- recipe.pipeline_parallel_size = Some(0); // auto
|
|
|
- let node = &cluster.nodes[0];
|
|
|
-
|
|
|
- // pp_size = 0 should become nnodes (2) in start_cluster, but build_docker_command
|
|
|
- // takes pp_size directly. test with pp_size = nnodes = 2.
|
|
|
- let cmd = build_docker_command(&cluster, &nccl, &recipe, "test-recipe", node, true, 2, 2);
|
|
|
- let joined = cmd.join(" ");
|
|
|
-
|
|
|
- assert!(joined.contains("--pipeline-parallel-size 2"));
|
|
|
- }
|
|
|
-
|
|
|
- #[test]
|
|
|
- fn test_build_docker_command_entrypoint() {
|
|
|
- let cluster = test_cluster();
|
|
|
- let nccl = test_nccl();
|
|
|
- let mut recipe = test_recipe();
|
|
|
- recipe.entrypoint = vec!["python".to_string(), "-m".to_string(), "custom.module".to_string()];
|
|
|
- let node = &cluster.nodes[0];
|
|
|
-
|
|
|
- let cmd = build_docker_command(&cluster, &nccl, &recipe, "test-recipe", node, true, 2, 2);
|
|
|
- let joined = cmd.join(" ");
|
|
|
-
|
|
|
- // Entrypoint should appear after image name, before model
|
|
|
- let image_pos = joined.find("test/image:latest").unwrap();
|
|
|
- let model_pos = joined.find("test/model").unwrap();
|
|
|
- let entrypoint_pos = joined.find("python -m custom.module").unwrap();
|
|
|
- assert!(image_pos < entrypoint_pos && entrypoint_pos < model_pos,
|
|
|
- "entrypoint should be between image and model");
|
|
|
- }
|
|
|
-
|
|
|
- // ── select_cluster tests ────────────────────────────────────
|
|
|
-
|
|
|
- #[test]
|
|
|
- fn test_select_cluster_by_name() {
|
|
|
- let mut config = Config {
|
|
|
- ssh_user: "user".to_string(),
|
|
|
- ssh_key_path: "/path/to/key".to_string(),
|
|
|
- clusters: vec![test_cluster()],
|
|
|
- nccl: HashMap::new(),
|
|
|
- };
|
|
|
- config.nccl.insert("test-cluster".to_string(), test_nccl());
|
|
|
-
|
|
|
- let (cluster, nccl) = select_cluster(&config, "test-cluster").unwrap();
|
|
|
- assert_eq!(cluster.name, "test-cluster");
|
|
|
- assert_eq!(nccl.ib_hca, "rocep1s0f0");
|
|
|
- }
|
|
|
-
|
|
|
- #[test]
|
|
|
- fn test_select_cluster_auto() {
|
|
|
- let mut config = Config {
|
|
|
- ssh_user: "user".to_string(),
|
|
|
- ssh_key_path: "/path/to/key".to_string(),
|
|
|
- clusters: vec![test_cluster()],
|
|
|
- nccl: HashMap::new(),
|
|
|
- };
|
|
|
- config.nccl.insert("test-cluster".to_string(), test_nccl());
|
|
|
-
|
|
|
- let (cluster, nccl) = select_cluster(&config, "auto").unwrap();
|
|
|
- assert_eq!(cluster.name, "test-cluster");
|
|
|
- assert_eq!(nccl.ib_hca, "rocep1s0f0");
|
|
|
- }
|
|
|
-
|
|
|
- #[test]
|
|
|
- fn test_select_cluster_auto_no_match() {
|
|
|
- let config = Config {
|
|
|
- ssh_user: "user".to_string(),
|
|
|
- ssh_key_path: "/path/to/key".to_string(),
|
|
|
- clusters: vec![test_cluster()],
|
|
|
- nccl: HashMap::new(),
|
|
|
- };
|
|
|
- // No NCCL config for test-cluster → auto should fail
|
|
|
- let result = select_cluster(&config, "auto");
|
|
|
- assert!(result.is_err());
|
|
|
- }
|
|
|
-
|
|
|
- #[test]
|
|
|
- fn test_select_cluster_not_found() {
|
|
|
- let config = Config {
|
|
|
- ssh_user: "user".to_string(),
|
|
|
- ssh_key_path: "/path/to/key".to_string(),
|
|
|
- clusters: vec![],
|
|
|
- nccl: HashMap::new(),
|
|
|
- };
|
|
|
- let result = select_cluster(&config, "nonexistent");
|
|
|
- assert!(result.is_err());
|
|
|
- }
|
|
|
-
|
|
|
- // ── Recipe deserialization tests ─────────────────────────────
|
|
|
-
|
|
|
- #[test]
|
|
|
- fn test_recipe_deserialize_minimal() {
|
|
|
- let toml_str = r#"
|
|
|
-name = "Minimal"
|
|
|
-model = "test/model"
|
|
|
-"#;
|
|
|
- let recipe: Recipe = toml::from_str(toml_str).unwrap();
|
|
|
- assert_eq!(recipe.name, "Minimal");
|
|
|
- assert_eq!(recipe.model, "test/model");
|
|
|
- assert_eq!(recipe.engine, "vllm"); // default
|
|
|
- assert_eq!(recipe.port, 0); // default
|
|
|
- assert!(recipe.args.is_empty());
|
|
|
- assert!(recipe.env.is_empty());
|
|
|
- assert_eq!(recipe.distributed_backend, "mp"); // default
|
|
|
- }
|
|
|
-
|
|
|
- #[test]
|
|
|
- fn test_recipe_deserialize_full() {
|
|
|
- let toml_str = r#"
|
|
|
-engine = "sglang"
|
|
|
-name = "Full Recipe"
|
|
|
-docker_image = "org/image:tag"
|
|
|
-model = "org/model"
|
|
|
-port = 30000
|
|
|
-required_nodes = 4
|
|
|
-pipeline_parallel_size = 2
|
|
|
-distributed_backend = "ray"
|
|
|
-cluster = "my-cluster"
|
|
|
-
|
|
|
-args = ["--arg1", "val1", "--arg2"]
|
|
|
-
|
|
|
-[env]
|
|
|
-KEY = "value"
|
|
|
-OTHER = "123"
|
|
|
-"#;
|
|
|
- let recipe: Recipe = toml::from_str(toml_str).unwrap();
|
|
|
- assert_eq!(recipe.engine, "sglang");
|
|
|
- assert_eq!(recipe.name, "Full Recipe");
|
|
|
- assert_eq!(recipe.docker_image, "org/image:tag");
|
|
|
- assert_eq!(recipe.model, "org/model");
|
|
|
- assert_eq!(recipe.port, 30000);
|
|
|
- assert_eq!(recipe.required_nodes, Some(4));
|
|
|
- assert_eq!(recipe.pipeline_parallel_size, Some(2));
|
|
|
- assert_eq!(recipe.distributed_backend, "ray");
|
|
|
- assert_eq!(recipe.cluster, Some("my-cluster".to_string()));
|
|
|
- assert_eq!(recipe.args, vec!["--arg1", "val1", "--arg2"]);
|
|
|
- assert_eq!(recipe.env.get("KEY").unwrap(), "value");
|
|
|
- assert_eq!(recipe.env.get("OTHER").unwrap(), "123");
|
|
|
- }
|
|
|
-
|
|
|
- #[test]
|
|
|
- fn test_recipe_deserialize_no_required_nodes() {
|
|
|
- let toml_str = r#"
|
|
|
-name = "Flexible"
|
|
|
-model = "test/model"
|
|
|
-"#;
|
|
|
- let recipe: Recipe = toml::from_str(toml_str).unwrap();
|
|
|
- assert_eq!(recipe.required_nodes, None); // None, not Some(0)
|
|
|
- assert_eq!(recipe.pipeline_parallel_size, None);
|
|
|
- }
|
|
|
-
|
|
|
- // ── Config deserialization tests ───────────────────────────────
|
|
|
-
|
|
|
- #[test]
|
|
|
- fn test_config_deserialize() {
|
|
|
- let toml_str = r#"
|
|
|
-ssh_user = "admin"
|
|
|
-ssh_key_path = "/home/admin/.ssh/id_rsa"
|
|
|
-
|
|
|
-[[clusters]]
|
|
|
-name = "prod"
|
|
|
-master_ib_ip = "10.0.0.1"
|
|
|
-nodes = [
|
|
|
- { name = "node1", ssh_ip = "10.0.0.10", ib_ip = "10.0.0.1", rank = 0 },
|
|
|
- { name = "node2", ssh_ip = "10.0.0.11", ib_ip = "10.0.0.2", rank = 1 },
|
|
|
-]
|
|
|
-
|
|
|
-[nccl."prod"]
|
|
|
-ib_hca = "mlx5_0"
|
|
|
-socket_ifname = "eth0"
|
|
|
-ib_gid_index = 3
|
|
|
-ib_disable = 0
|
|
|
-"#;
|
|
|
- let config: Config = toml::from_str(toml_str).unwrap();
|
|
|
- assert_eq!(config.ssh_user, "admin");
|
|
|
- assert_eq!(config.clusters.len(), 1);
|
|
|
- assert_eq!(config.clusters[0].nodes.len(), 2);
|
|
|
- assert_eq!(config.clusters[0].nodes[0].host_key, None);
|
|
|
- assert!(config.nccl.contains_key("prod"));
|
|
|
- assert_eq!(config.nccl["prod"].ib_hca, "mlx5_0");
|
|
|
- }
|
|
|
-
|
|
|
- #[test]
|
|
|
- fn test_config_deserialize_with_host_key() {
|
|
|
- let toml_str = r#"
|
|
|
-ssh_user = "admin"
|
|
|
-ssh_key_path = "/path/to/key"
|
|
|
-
|
|
|
-[[clusters]]
|
|
|
-name = "secure"
|
|
|
-master_ib_ip = "1.1.1.1"
|
|
|
-nodes = [
|
|
|
- { name = "n1", ssh_ip = "2.2.2.2", ib_ip = "1.1.1.1", rank = 0, host_key = "SHA256:abc123" },
|
|
|
-]
|
|
|
-"#;
|
|
|
- let config: Config = toml::from_str(toml_str).unwrap();
|
|
|
- assert_eq!(
|
|
|
- config.clusters[0].nodes[0].host_key,
|
|
|
- Some("SHA256:abc123".to_string())
|
|
|
- );
|
|
|
- }
|
|
|
-
|
|
|
- // ── generate_repair_dockerfile tests ───────────────────────────
|
|
|
-
|
|
|
- #[test]
|
|
|
- fn test_generate_repair_dockerfile_vllm() {
|
|
|
- let recipe = test_recipe();
|
|
|
- let dockerfile = generate_repair_dockerfile("base/image:tag", &recipe);
|
|
|
-
|
|
|
- assert!(dockerfile.contains("FROM base/image:tag"));
|
|
|
- assert!(dockerfile.contains("pip install vllm"));
|
|
|
- assert!(!dockerfile.contains("ray"));
|
|
|
- assert!(dockerfile.contains("ENTRYPOINT"));
|
|
|
- }
|
|
|
-
|
|
|
- #[test]
|
|
|
- fn test_generate_repair_dockerfile_ray() {
|
|
|
- let mut recipe = test_recipe();
|
|
|
- recipe.distributed_backend = "ray".to_string();
|
|
|
- let dockerfile = generate_repair_dockerfile("base/image:tag", &recipe);
|
|
|
-
|
|
|
- assert!(dockerfile.contains("FROM base/image:tag"));
|
|
|
- assert!(dockerfile.contains("pip install vllm"));
|
|
|
- assert!(dockerfile.contains("pip install ray"));
|
|
|
- }
|
|
|
-
|
|
|
- #[test]
|
|
|
- fn test_generate_repair_dockerfile_sglang() {
|
|
|
- let mut recipe = test_recipe();
|
|
|
- recipe.engine = "sglang".to_string();
|
|
|
- let dockerfile = generate_repair_dockerfile("base/image:tag", &recipe);
|
|
|
-
|
|
|
- assert!(dockerfile.contains("pip install sglang"));
|
|
|
- assert!(!dockerfile.contains("vllm"));
|
|
|
- }
|
|
|
-
|
|
|
- // ── select_cluster_by_name tests ───────────────────────────────
|
|
|
-
|
|
|
- #[test]
|
|
|
- fn test_select_cluster_by_name_found() {
|
|
|
- let mut nccl_map = HashMap::new();
|
|
|
- nccl_map.insert("test-cluster".to_string(), test_nccl());
|
|
|
- let config = Config {
|
|
|
- ssh_user: "u".to_string(),
|
|
|
- ssh_key_path: "/k".to_string(),
|
|
|
- clusters: vec![test_cluster()],
|
|
|
- nccl: nccl_map,
|
|
|
- };
|
|
|
- let cluster = select_cluster_by_name(&config, "test-cluster").unwrap();
|
|
|
- assert_eq!(cluster.name, "test-cluster");
|
|
|
- assert_eq!(cluster.nodes.len(), 2);
|
|
|
- }
|
|
|
-
|
|
|
- #[test]
|
|
|
- fn test_select_cluster_by_name_not_found() {
|
|
|
- let config = Config {
|
|
|
- ssh_user: "u".to_string(),
|
|
|
- ssh_key_path: "/k".to_string(),
|
|
|
- clusters: vec![test_cluster()],
|
|
|
- nccl: HashMap::new(),
|
|
|
- };
|
|
|
- let result = select_cluster_by_name(&config, "no-such-cluster");
|
|
|
- assert!(result.is_err());
|
|
|
- }
|
|
|
-
|
|
|
- // ── Defaults ───────────────────────────────────────────────────
|
|
|
-
|
|
|
- #[test]
|
|
|
- fn test_default_engine() {
|
|
|
- assert_eq!(default_engine(), "vllm");
|
|
|
- }
|
|
|
-
|
|
|
- #[test]
|
|
|
- fn test_default_distributed_backend() {
|
|
|
- assert_eq!(default_distributed_backend(), "mp");
|
|
|
- }
|
|
|
-}
|