diff --git a/.gitignore b/.gitignore index c824efdf..3ab73834 100644 --- a/.gitignore +++ b/.gitignore @@ -131,6 +131,10 @@ venv/ docker/ scripts/ *.json +# ... except the schemas madengine ships and loads at runtime. The blanket rule above is +# for the model JSONs that land in a dev checkout; these are package data, and hatchling +# picks files by VCS status, so an ignored schema would also be missing from the wheel. +!src/madengine/schemas/*.json .*_env/ .vscode/ diff --git a/docs/cli-reference.md b/docs/cli-reference.md index 842e0fdf..71c95aee 100644 --- a/docs/cli-reference.md +++ b/docs/cli-reference.md @@ -522,9 +522,24 @@ madengine uses standard exit codes so scripts and CI (e.g. Jenkins) can detect s | `2` | `BUILD_FAILURE` | One or more image builds failed (e.g. Docker build error) | | `3` | `RUN_FAILURE` | One or more model executions failed | | `4` | `INVALID_ARGS` | Invalid command-line arguments or configuration | +| `5` | `NO_METRIC` | The workload ran to completion but reported no performance metric | **Failure recording:** Pre-run failures (e.g. image pull, setup) and run failures are recorded in the performance table (`perf.csv`) with status `FAILURE`, so all attempted models appear in the CSV. The file is created automatically if missing. +**`NO_METRIC` versus `RUN_FAILURE`:** a crashed workload and a workload that finished +without producing a number are different problems — the first is a broken run, the second a +broken contract between the model script and madengine — so they get different exit codes +and the row in `perf.csv` reads `NO_METRIC` rather than `FAILURE`. + +**Multi-node verdicts:** every node records its exit code, and the submit node reads them +all back. A metric outweighs an exit code, the same way it does for a single node: where a +framework reports throughput from one rank only, the nodes that collected nothing exit +non-zero on a perfectly healthy run, so a run that produced a metric is reported as a +success with the node outcomes listed as warnings. With no metric anywhere, the node +evidence is all there is and it decides: a node that exited non-zero or never reported at +all gives `RUN_FAILURE` naming the node, while a clean set of exit codes and no metric +gives `NO_METRIC`. + **Example usage in scripts / CI:** ```bash diff --git a/docs/configuration.md b/docs/configuration.md index 4831cc4f..e2f0af3e 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -252,6 +252,11 @@ This allows you to rebuild only changed models while maintaining references to e ### Environment Variables +Host-side variables (`MODEL_DIR`, cache roots, `MAD_DOCKER_BUILDS`) can be collected in a +shell env file that the manifest names via `deployment_config.env_file`; see +[Environment file](deployment.md#environment-file-env_file). The variables below are the +container's, and are set from the run configuration. + Pass environment variables to containers: ```json diff --git a/docs/deployment.md b/docs/deployment.md index fa03e7f5..bdccfcb8 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -252,6 +252,74 @@ The deployment target is automatically detected from the `slurm` key in the conf See [examples/slurm-configs/](../examples/slurm-configs/) for complete examples. +### Environment file (`env_file`) + +A cluster run usually depends on a handful of host variables that say where things live — +the model directory, the cache roots, the shared image store. Instead of requiring every +operator to `source mad.env` in the shell that launches madengine, name the file in the +build manifest and madengine loads it itself, on the submit node and on every worker: + +```json +{ + "deployment_config": { + "target": "slurm", + "env_file": "mad.env" + } +} +``` + +A relative path is resolved against the manifest's directory, so a run directory holding +the manifest and its `mad.env` side by side stays movable. The file is sourced with +`bash`, exactly as `source mad.env` would, so `${VAR:-default}`, command substitution and +conditionals all work — and, like the manifest that names it, the file is trusted input. +Values in the file override what madengine inherited from the launching shell. madengine +logs the names of the variables it applied, never their values, so an env file may carry +tokens. + +The field is optional. Without it madengine behaves exactly as it did before, and nothing +here requires `MAD_DOCKER_BUILDS` or any other variable to be set. What is fatal is naming +a file that is not there: that stops the run at startup rather than leaving a variable to +resolve to the empty string mid-run. + +### Shared image store (`MAD_DOCKER_BUILDS`) + +For a multi-node run every worker needs the same image. Set `MAD_DOCKER_BUILDS` to a +directory on shared storage (usually from the `env_file` above) and madengine saves the +built image there once, then loads it from the tar on workers whose local image ID differs: + +```bash +export MAD_DOCKER_BUILDS=/shared/mad/docker_builds +``` + +Leaving it unset is supported but leaves image distribution to the operator: workers that +do not already have the image cannot reconcile it, and the run fails on those nodes with a +message saying so. + +### How a multi-node run is judged + +Every node writes its exit code to `///node_/node.status` +before it copies anything else, so the outcome survives even when the artifacts do not. The +submit node reads them all back, and weighs them the way a single-node run does — a metric +outweighs an exit code: + +| What came back | Verdict | +| --- | --- | +| A metric, all nodes zero | success | +| A metric, some node non-zero or silent | success, with the node outcomes as warnings | +| No metric, some node non-zero or silent | `RUN_FAILURE`, naming the node | +| No metric, all nodes zero | `NO_METRIC` | + +The warning row is not a technicality. Where a framework reports throughput from one rank +only — Primus/Megatron reports from the last global rank — every other node finds no metric +locally and exits non-zero on a completely healthy run. A verdict that trusted exit codes +over results would fail every such run, so the exit codes are reported and the numbers are +kept. + +For a workload where every rank really is expected to exit zero, set +`slurm.kill_on_bad_exit` to tear the step down on the first bad exit instead of letting the +survivors block on a peer that will never answer. It is off by default for the reason above: +the node exiting non-zero may be the one whose peer holds the numbers. + ### Multi-Node Training For distributed training across SLURM nodes: diff --git a/pyproject.toml b/pyproject.toml index dd9c7566..83b2978c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -25,6 +25,7 @@ dependencies = [ "rich>=13.0.0", "click>=8.0.0", "jinja2>=3.0.0", + "jsonschema>=4.0.0", "pyyaml>=6.0", "kubernetes>=28.0.0", "pytest>=7.0", diff --git a/src/madengine/cli/commands/run.py b/src/madengine/cli/commands/run.py index c961cbe5..9c99743b 100644 --- a/src/madengine/cli/commands/run.py +++ b/src/madengine/cli/commands/run.py @@ -280,16 +280,35 @@ def run( save_summary_with_feedback(execution_summary, summary_output, "Execution") failed_runs = len(execution_summary.get("failed_runs", [])) - if failed_runs == 0: + no_metric_runs = len(execution_summary.get("no_metric_runs", [])) + incomplete = execution_summary.get("incomplete") + for warning in execution_summary.get("warnings", []): + console.print(f"⚠️ [yellow]{warning}[/yellow]") + if incomplete: console.print( - "🎉 [bold green]All model executions completed successfully![/bold green]" + f"💥 [bold red]Run did not complete and produced no metric: " + f"{incomplete.get('reason', 'unknown reason')}[/bold red]" ) - raise typer.Exit(ExitCode.SUCCESS) - else: + raise typer.Exit(ExitCode.RUN_FAILURE) + if failed_runs: console.print( - f"💥 [bold red]Execution failed for {failed_runs} models[/bold red]" + f"💥 [bold red]Execution failed for {failed_runs} " + f"{'run' if failed_runs == 1 else 'runs'}[/bold red]" ) raise typer.Exit(ExitCode.RUN_FAILURE) + if no_metric_runs: + # The workload ran to completion and produced nothing to measure. That is + # not a successful benchmark, and it is not the same failure as a crash: + # what broke is the contract between the model script and madengine. + console.print( + f"📉 [bold yellow]{no_metric_runs} models ran but reported no " + f"performance metric[/bold yellow]" + ) + raise typer.Exit(ExitCode.NO_METRIC) + console.print( + "🎉 [bold green]All model executions completed successfully![/bold green]" + ) + raise typer.Exit(ExitCode.SUCCESS) else: # MAD_CONTAINER_IMAGE handling is done in RunOrchestrator diff --git a/src/madengine/cli/constants.py b/src/madengine/cli/constants.py index b437fa30..bac819bf 100644 --- a/src/madengine/cli/constants.py +++ b/src/madengine/cli/constants.py @@ -17,6 +17,9 @@ class ExitCode(IntEnum): BUILD_FAILURE = 2 RUN_FAILURE = 3 INVALID_ARGS = 4 + #: The workload finished but produced no performance metric. Distinct from + #: RUN_FAILURE so a caller can tell a crashed run from a broken result contract. + NO_METRIC = 5 # Valid values for validation diff --git a/src/madengine/core/env_file.py b/src/madengine/core/env_file.py new file mode 100644 index 00000000..2deb266a --- /dev/null +++ b/src/madengine/core/env_file.py @@ -0,0 +1,169 @@ +#!/usr/bin/env python3 +""" +Load a shell env file (the `mad.env` convention) into the run's environment. + +Multi-node runs depend on a set of variables that describe *where things live* on the +cluster: `MODEL_DIR`, the cache roots, and `MAD_DOCKER_BUILDS`. Until now the only way to +supply them was to `source mad.env` in the same shell before every `madengine run`, and +forgetting produced failures far from the cause — an empty `MODEL_DIR` makes the run +script path resolve to nothing, and a `MAD_DOCKER_BUILDS` that is not on shared storage +makes every worker rebuild the image or fail to find it. + +A manifest can now name the file (`deployment_config.env_file`) and madengine loads it +itself. The file is executed by `bash`, exactly as sourcing it would, so the usual shell +constructs (`${VAR:-default}`, `$(cat ~/.token)`, conditionals) behave the same — which +also means an env file is trusted input, on par with the manifest that names it. + +Copyright (c) Advanced Micro Devices, Inc. All rights reserved. +""" + +import os +import shlex +import subprocess +from pathlib import Path +from typing import Dict, Optional + +from madengine.core.errors import ErrorContext, ValidationError + +#: A shell that hangs (waiting on a prompt, say) must not hang the run. +SOURCE_TIMEOUT_SECONDS = 120 + +#: Separates the before/after environment dumps in the helper shell's output. Both dumps +#: come from the same shell, so anything bash sets on its own (COLUMNS, SHLVL, ...) +#: appears in both and is not mistaken for something the env file did. +_BOUNDARY = "__madengine_env_file_boundary__" + +#: Bash's own bookkeeping, which differs between the two dumps for reasons unrelated to +#: the file's contents. +_SHELL_BOOKKEEPING = frozenset({"_", "SHLVL", "PWD", "OLDPWD"}) + +#: What each file already applied in this process, keyed by resolved path. A run reaches +#: :func:`apply_env_file` twice for the same manifest -- once in the orchestrator, once +#: when the deployment layer loads the manifest -- and sourcing a file again is not a +#: no-op: `PATH="$PATH:/opt/x"` appends a second time and `$(...)` runs a second time. +_APPLIED: Dict[str, Dict[str, str]] = {} + + +def resolve_env_file(env_file: str, base_dir: Optional[str] = None) -> Path: + """The path an *env_file* names, relative paths resolved against *base_dir*.""" + path = Path(env_file) + if not path.is_absolute() and base_dir: + path = Path(base_dir) / path + return path + + +def load_env_file(env_file: str, base_dir: Optional[str] = None) -> Dict[str, str]: + """ + Source an env file with bash and return the variables it sets or changes. + + Args: + env_file: path to the file; relative paths resolve against `base_dir` + base_dir: directory to resolve a relative `env_file` against (typically the + manifest's directory), defaults to the current working directory + + Returns: + Dict[str, str]: variables the file added or changed, relative to the environment + madengine is running with + + Raises: + ValidationError: the file is missing, or bash failed to source it + """ + path = resolve_env_file(env_file, base_dir) + + context = ErrorContext( + operation="env_file loading", component="core.env_file", file_path=str(path) + ) + if not path.is_file(): + raise ValidationError( + f"env_file not found: {path}", + context=context, + suggestions=[ + "deployment_config.env_file is resolved relative to the manifest", + ], + ) + + # `set -a` is what makes plain `KEY=value` lines exported, matching what an operator + # gets from `source mad.env` in a shell configured the usual way. Sourcing is checked + # explicitly: a syntax error makes `.` fail but would otherwise be masked by the + # `env` that follows it. + script = ( + f"env -0; printf '%s\\0' {shlex.quote(_BOUNDARY)}; " + f"set -a; . {shlex.quote(str(path))} || exit 42; env -0" + ) + try: + result = subprocess.run( + ["bash", "-c", script], + capture_output=True, + text=True, + timeout=SOURCE_TIMEOUT_SECONDS, + check=False, + ) + except subprocess.TimeoutExpired as exc: + raise ValidationError( + f"Timed out after {SOURCE_TIMEOUT_SECONDS}s sourcing env_file: {path}", + context=context, + cause=exc, + ) from exc + + if result.returncode != 0: + detail = result.stderr.strip() or f"bash exited with {result.returncode}" + raise ValidationError( + f"Failed to source env_file {path}: {detail}", + context=context, + ) + + before_dump, _, after_dump = result.stdout.partition(f"{_BOUNDARY}\0") + before = _parse_env_dump(before_dump) + + loaded: Dict[str, str] = {} + for key, value in _parse_env_dump(after_dump).items(): + # A variable that already held this value is not something the file changed. + if key in _SHELL_BOOKKEEPING or before.get(key) == value: + continue + loaded[key] = value + return loaded + + +def _parse_env_dump(dump: str) -> Dict[str, str]: + """ + Parse NUL-delimited `env -0` output into a mapping. + + Args: + dump: raw `env -0` output + + Returns: + Dict[str, str]: the variables in the dump + """ + parsed: Dict[str, str] = {} + for entry in dump.split("\0"): + if not entry or "=" not in entry: + continue + key, value = entry.split("=", 1) + parsed[key] = value + return parsed + + +def apply_env_file(env_file: str, base_dir: Optional[str] = None) -> Dict[str, str]: + """ + Load an env file and apply it to `os.environ`, as sourcing it would, once per process. + + The file wins over the inherited environment, so behaviour matches what the operator + gets by sourcing it before the run. A second call for the same file returns what the + first one applied without running it again: the submit side of a cluster run reaches + this twice for one manifest, and a file is shell, not a list of assignments. + + Args: + env_file: path to the file; relative paths resolve against `base_dir` + base_dir: directory to resolve a relative `env_file` against + + Returns: + Dict[str, str]: the variables that were applied + """ + key = str(resolve_env_file(env_file, base_dir).resolve()) + if key in _APPLIED: + return dict(_APPLIED[key]) + + loaded = load_env_file(env_file, base_dir) + os.environ.update(loaded) + _APPLIED[key] = dict(loaded) + return loaded diff --git a/src/madengine/deployment/base.py b/src/madengine/deployment/base.py index 69e1367e..be72ddb2 100644 --- a/src/madengine/deployment/base.py +++ b/src/madengine/deployment/base.py @@ -19,6 +19,9 @@ from jinja2 import Environment, FileSystemLoader from rich.console import Console +from madengine.core.env_file import apply_env_file +from madengine.schemas import perf_csv_header, validate_build_manifest + # Regex for parsing "performance: " log lines. # Value: optional sign, integer/decimal, scientific notation (e or E). @@ -122,8 +125,8 @@ def __init__(self, config: DeploymentConfig): config: Deployment configuration """ self.config = config - self.manifest = self._load_manifest(config.manifest_file) self.console = Console() + self.manifest = self._load_manifest(config.manifest_file) def _load_manifest(self, manifest_file: str) -> Dict: """ @@ -152,6 +155,20 @@ def _load_manifest(self, manifest_file: str) -> Dict: if missing: raise ValueError(f"Invalid manifest, missing: {missing}") + for warning in validate_build_manifest(manifest, source=str(manifest_path)): + self.console.print(f"[yellow]⚠ {warning}[/yellow]") + + env_file = manifest.get("deployment_config", {}).get("env_file") + if env_file: + # The submit side needs these too: MAD_DOCKER_BUILDS and the cache roots are + # read while rendering the job script, before any node starts. + applied = apply_env_file(env_file, base_dir=str(manifest_path.resolve().parent)) + # Names only: an env file legitimately carries secrets. + self.console.print( + f"[cyan]Loaded env_file {env_file}: " + f"{', '.join(sorted(applied)) or '(no new variables)'}[/cyan]" + ) + return manifest # Template Method - defines workflow @@ -590,14 +607,7 @@ def _ensure_perf_csv_exists(self) -> None: perf_csv_path = Path("perf.csv") if perf_csv_path.exists(): return - standard_header = ( - "model,n_gpus,nnodes,gpus_per_node,training_precision,pipeline,args,tags," - "docker_file,base_docker,docker_sha,docker_image,git_commit,machine_name," - "deployment_type,launcher,gpu_architecture,performance,metric,relative_change," - "status,build_duration,test_duration,dataname,data_provider_type,data_size," - "data_download_duration,build_number,additional_docker_run_options" - ) - perf_csv_path.write_text(standard_header + "\n", encoding="utf-8") + perf_csv_path.write_text(perf_csv_header() + "\n", encoding="utf-8") def _write_to_perf_csv(self, perf_data: Dict[str, Any]) -> None: """ diff --git a/src/madengine/deployment/k8s_results.py b/src/madengine/deployment/k8s_results.py index 6da189b5..8567433d 100644 --- a/src/madengine/deployment/k8s_results.py +++ b/src/madengine/deployment/k8s_results.py @@ -16,6 +16,7 @@ from typing import Any, Dict, List, Optional from .common import normalize_launcher +from madengine.schemas import perf_csv_header from madengine.utils.path_utils import scripts_base_dir_from from madengine.utils.run_details import flatten_tags_in_place, get_build_number, get_pipeline @@ -48,15 +49,6 @@ def collector_pod_name(deployment_id: str) -> str: class KubernetesResultsMixin: """Results collection and performance reporting for Kubernetes deployments.""" - # Standard perf.csv header (must match container_runner.ensure_perf_csv_exists) - _PERF_CSV_HEADER = ( - "model,n_gpus,nnodes,gpus_per_node,training_precision,pipeline,args,tags," - "docker_file,base_docker,docker_sha,docker_image,git_commit,machine_name," - "deployment_type,launcher,gpu_architecture,performance,metric,relative_change," - "status,build_duration,test_duration,dataname,data_provider_type,data_size," - "data_download_duration,build_number,additional_docker_run_options" - ) - def collect_results(self, deployment_id: str) -> Dict[str, Any]: """ Enhanced results collection from K8s pods following vLLM multi-node best practices. @@ -977,7 +969,7 @@ def _ensure_perf_csv_exists(self) -> None: """Ensure perf.csv exists with standard header (same as Docker container_runner).""" perf_csv_path = Path("perf.csv") if not perf_csv_path.exists(): - perf_csv_path.write_text(self._PERF_CSV_HEADER + "\n", encoding="utf-8") + perf_csv_path.write_text(perf_csv_header() + "\n", encoding="utf-8") self.console.print("[dim]Created perf.csv with standard header[/dim]") def _build_perf_entry_from_aggregated( diff --git a/src/madengine/deployment/slurm.py b/src/madengine/deployment/slurm.py index 088a3fb2..af723c47 100644 --- a/src/madengine/deployment/slurm.py +++ b/src/madengine/deployment/slurm.py @@ -80,6 +80,11 @@ def __init__(self, config: DeploymentConfig): self.reservation = self.slurm_config.get("reservation", None) # Some clusters expose no GPU GRES, so sbatch rejects --gpus-per-node. self.skip_gpus_directive = self.slurm_config.get("skip_gpus_directive", False) + # Off by default: in a run whose framework reports throughput from one rank only, + # the node that collected nothing exits non-zero on a healthy run, and tearing the + # step down there would kill the node that holds the numbers. Turn it on for a + # workload where any non-zero rank really does mean the run is over. + self.kill_on_bad_exit = self.slurm_config.get("kill_on_bad_exit", False) # Setup Jinja2 template engine template_dir = Path(__file__).parent / "templates" / "slurm" @@ -694,6 +699,7 @@ def debug(self, msg): "nodes": self.nodes, "gpus_per_node": resolved_gpus_per_node, # Use resolved GPU count "skip_gpus_directive": self.skip_gpus_directive, + "kill_on_bad_exit": self.kill_on_bad_exit, "time_limit": self.time_limit, "output_dir": str(self.output_dir), "master_port": master_port, @@ -1728,6 +1734,157 @@ def _select_best_multiple_results_csv(self, candidates: List[Path]) -> Optional[ return best_candidate + def _job_exit_state(self, job_id: str) -> Optional[str]: + """SLURM's own verdict on the job. + + Args: + job_id: the job to ask about + + Returns: + Optional[str]: the state, e.g. COMPLETED or FAILED, or None when sacct cannot + answer — in which case nothing is claimed on its behalf + """ + try: + result = subprocess.run( + ["sacct", "-j", job_id, "-n", "-X", "-o", "State"], + capture_output=True, + text=True, + timeout=10, + ) + except Exception: + return None + if result.returncode != 0: + return None + first_line = (result.stdout or "").strip().splitlines() + return first_line[0].strip().upper() if first_line else None + + def _read_node_statuses(self, job_dir: Path) -> List[Dict[str, Any]]: + """Read the per-node status markers the job script leaves behind. + + Args: + job_dir: collection directory for this job + + Returns: + List[Dict[str, Any]]: one entry per node found, sorted by node rank, each with + `node`, `host` and `exit_code` + """ + statuses: List[Dict[str, Any]] = [] + for status_path in sorted(job_dir.glob("node_*/node.status")): + fields: Dict[str, Any] = {} + try: + for line in status_path.read_text(encoding="utf-8", errors="ignore").splitlines(): + key, _, value = line.partition("=") + if key: + fields[key.strip()] = value.strip() + except OSError: + continue + try: + fields["node"] = int(fields.get("node", status_path.parent.name.replace("node_", ""))) + fields["exit_code"] = int(fields.get("exit_code", 1)) + except ValueError: + continue + statuses.append(fields) + return sorted(statuses, key=lambda s: s["node"]) + + def _report_node_statuses(self, statuses: List[Dict[str, Any]]) -> List[Dict[str, Any]]: + """Print the per-node outcome and return the nodes that failed. + + Args: + statuses: entries from `_read_node_statuses` + + Returns: + List[Dict[str, Any]]: the entries whose exit code is non-zero + """ + failures = [s for s in statuses if s["exit_code"] != 0] + missing = [n for n in range(self.nodes) if n not in {s["node"] for s in statuses}] + + if failures: + self.console.print("[red]Nodes that failed:[/red]") + for status in failures: + self.console.print( + f"[red] node {status['node']} ({status.get('host', 'unknown host')}): " + f"exit code {status['exit_code']}[/red]" + ) + if missing and statuses: + # A node that never wrote a marker did not reach the end of the task script: + # it was killed, or the node itself went away. Only meaningful once at least + # one node has reported, since a job may predate these markers entirely. + self.console.print( + f"[red]Nodes that reported no status (killed or lost): " + f"{', '.join(str(n) for n in missing)}[/red]" + ) + failures = failures + [ + {"node": n, "host": "", "exit_code": None, "missing": True} for n in missing + ] + return failures + + @staticmethod + def _incomplete_reason( + node_failures: List[Dict[str, Any]], job_state: Optional[str] + ) -> str: + """Say why the run is incomplete, in the words of whoever noticed. + + Args: + node_failures: entries from `_report_node_statuses` + job_state: the state sacct reports, if it could be read + + Returns: + str: a phrase naming the failing nodes, or SLURM's own verdict + """ + if node_failures: + parts = [] + for failure in node_failures: + if failure.get("missing"): + parts.append(f"node {failure['node']} reported no status (killed or lost)") + else: + parts.append(f"node {failure['node']} exited {failure['exit_code']}") + return ", ".join(parts) + if job_state: + return f"SLURM reports the job as {job_state}" + return "the job did not complete" + + def _record_verdict( + self, + results: Dict[str, Any], + model: str, + node_failures: List[Dict[str, Any]], + job_state: Optional[str], + rows_collected: int, + ) -> None: + """Record what a run that did not end cleanly amounts to. + + A metric outweighs an exit code, the way it already does for a single node + (`resolve_run_status`): a node exiting non-zero is routine in a multi-node run + whose framework reports throughput from one rank only, so the node that collected + nothing fails locally while the run as a whole did the work. What such a node + earns is a warning, not a verdict. With no metric anywhere, the node evidence is + all there is, and it decides. + + Args: + results: collection results, updated in place + model: model the run belongs to + node_failures: entries from `_report_node_statuses` + job_state: the state sacct reports, if it could be read + rows_collected: how many metric rows the run produced + """ + reason = self._incomplete_reason(node_failures, job_state) + if rows_collected: + warning = ( + f"{reason}, while the run produced {rows_collected} metric row(s); " + f"reported as measurements, check the node logs before trusting them" + ) + results.setdefault("warnings", []).append(warning) + self.console.print(f"[yellow]⚠ {warning}[/yellow]") + return + results["incomplete"] = { + "model": model, + "reason": reason, + "rows_collected": 0, + } + self.console.print( + f"[red]✗ Job did not complete and produced no metric: {reason}[/red]" + ) + def collect_results(self, deployment_id: str) -> Dict[str, Any]: """Collect performance results from SLURM output files. @@ -1751,6 +1908,7 @@ def collect_results(self, deployment_id: str) -> Dict[str, Any]: "logs": [], "successful_runs": [], "failed_runs": [], + "no_metric_runs": [], "session_start_row": session_start_row, } @@ -1786,6 +1944,22 @@ def collect_results(self, deployment_id: str) -> Dict[str, Any]: job_dir = self.output_dir / model_name_for_path / deployment_id job_dir.mkdir(parents=True, exist_ok=True) + # Per-node outcomes come first: a metric parsed out of a log says nothing about + # whether the other nodes finished, and a run where one rank died is not a result. + node_statuses = self._read_node_statuses(job_dir) + node_failures = self._report_node_statuses(node_statuses) + job_state = self._job_exit_state(deployment_id) + # SLURM's verdict counts even when no node marker says so — a step killed by the + # scheduler leaves no marker at all. + job_failed = bool(node_failures) or ( + job_state is not None and "COMPLETED" not in job_state + ) + if job_failed and not node_failures and job_state: + self.console.print(f"[red]SLURM reports job {deployment_id} as {job_state}[/red]") + results["node_statuses"] = node_statuses + results["node_failures"] = node_failures + results["job_state"] = job_state + # Gather log content per node: from job_dir/node_N/ (new) or flat output_dir .out files per_node_log_contents: List[tuple] = [] flat_out_files = sorted(self.output_dir.glob(f"madengine-*_{deployment_id}_*.out")) @@ -1921,12 +2095,18 @@ def collect_results(self, deployment_id: str) -> Dict[str, Any]: ) results["perf_files"] = [str(Path("perf.csv").resolve())] import csv as _csv + # A row with a number in it is a measurement, and it stays one even when the + # run around it did not finish: a model that reports four precisions and + # crashes on two still measured the other two. The verdict on the run as a + # whole travels separately, in results["incomplete"]. + rows_collected = 0 try: with open(resolved_csv, "r", encoding="utf-8", errors="ignore") as f: reader = _csv.DictReader(f) for row in reader: row = {k.strip(): v for k, v in row.items() if k} if row.get("performance") and row.get("metric"): + rows_collected += 1 results["successful_runs"].append({ "model": model_info_for_entry.get("name", "") + "_" + row.get("model", ""), "status": "SUCCESS", @@ -1939,9 +2119,18 @@ def collect_results(self, deployment_id: str) -> Dict[str, Any]: }) except Exception: pass - self.console.print( - f"[green]✓ Updated perf.csv, perf_super.* from multiple_results (Docker-compatible)[/green]" - ) + if job_failed: + self._record_verdict( + results, + model_info_for_entry.get("name", model_name), + node_failures, + job_state, + rows_collected, + ) + else: + self.console.print( + f"[green]✓ Updated perf.csv, perf_super.* from multiple_results (Docker-compatible)[/green]" + ) return results # multiple_results set but CSV not found: fall through to single-result path (may write FAILURE) @@ -2021,6 +2210,28 @@ def collect_results(self, deployment_id: str) -> Dict[str, Any]: ) if run_details_dict is not None: + # Three outcomes, not two. A workload that failed and a workload that ran to + # completion without producing a metric need different answers: the first is a + # broken run, the second is a broken contract between the model script and + # madengine, and reporting both as FAILURE hides which one happened. + has_metric = bool(str(run_details_dict.get("performance") or "").strip()) + if job_failed: + self._record_verdict( + results, + run_details_dict.get("model", model_name), + node_failures, + job_state, + 1 if has_metric else 0, + ) + if not has_metric: + run_details_dict["status"] = "FAILURE" + elif not has_metric: + run_details_dict["status"] = "NO_METRIC" + self.console.print( + "[yellow]⚠ The workload finished on every node but no performance " + "metric was collected; recording the run as NO_METRIC[/yellow]" + ) + perf_entry_path = Path("perf_entry.json") with open(perf_entry_path, "w") as f: json.dump(run_details_dict, f, indent=2) @@ -2065,6 +2276,8 @@ def collect_results(self, deployment_id: str) -> Dict[str, Any]: } if run_details_dict.get("status") == "SUCCESS": results["successful_runs"].append(run_data) + elif run_details_dict.get("status") == "NO_METRIC": + results.setdefault("no_metric_runs", []).append(run_data) else: results["failed_runs"].append(run_data) summary = { diff --git a/src/madengine/deployment/templates/slurm/job.sh.j2 b/src/madengine/deployment/templates/slurm/job.sh.j2 index 4cdddba0..54becfd4 100644 --- a/src/madengine/deployment/templates/slurm/job.sh.j2 +++ b/src/madengine/deployment/templates/slurm/job.sh.j2 @@ -675,6 +675,11 @@ echo "[DEBUG] $(date -Iseconds) Node ${SLURM_PROCID} ($(hostname)): about to run # Run madengine with output redirected to node-specific log files # Use append (2>>) for stderr so debug lines written above are not overwritten # Environment variables (MASTER_ADDR, MAD_MULTI_NODE_RUNNER, etc.) are inherited +# +# errexit is off across this call on purpose: under `set -e` a failing madengine ends the +# task script right here, so the node writes no status marker and copies no artifacts — +# the run loses exactly the evidence a failed node has to leave behind. +set +e $MAD_CLI_COMMAND run \ --manifest-file "$EXEC_MANIFEST" \ --timeout {{ timeout | default(3600) }} \ @@ -683,6 +688,7 @@ $MAD_CLI_COMMAND run \ > "${NODE_LOG_OUT}" 2>> "${NODE_LOG_ERR}" TASK_EXIT=$? +set -e echo "[DEBUG] $(date -Iseconds) Node ${SLURM_PROCID} ($(hostname)): madengine exited with code $TASK_EXIT" >> "${NODE_LOG_ERR}" echo "" echo "Task completed with exit code: $TASK_EXIT" @@ -700,6 +706,16 @@ JOB_COLLECTION_DIR="${SUBMISSION_DIR}/{{ output_dir }}/{{ model_name }}/${SLURM_ NODE_COLLECTION_DIR="${JOB_COLLECTION_DIR}/node_${SLURM_PROCID}" mkdir -p "$NODE_COLLECTION_DIR" +# Record how this node ended, whatever happened. The submit node reads these to tell a +# workload failure from a missing metric; without them a dead worker is only visible as a +# number in someone's terminal. +{ + echo "node=${SLURM_PROCID}" + echo "host=$(hostname)" + echo "exit_code=${TASK_EXIT}" + echo "timestamp=$(date -Iseconds)" +} > "$NODE_COLLECTION_DIR/node.status" + if [ $TASK_EXIT -eq 0 ]; then echo "" echo "========================================================================" @@ -736,7 +752,12 @@ TASK_SCRIPT_EOF chmod +x "$TASK_SCRIPT" echo "Launching tasks on {{ nodes }} nodes..." -srun bash "$TASK_SCRIPT" +# slurm.kill_on_bad_exit tears the step down on the first non-zero rank instead of leaving +# the survivors to block on a peer that will never answer. It is off by default because a +# rank that exits non-zero is not always a dead run: where the framework reports throughput +# from one rank only, the other nodes fail locally on a perfectly healthy run, and killing +# the step there would take the node holding the numbers with it. +srun {% if kill_on_bad_exit %}--kill-on-bad-exit=1 {% endif %}bash "$TASK_SCRIPT" EXIT_CODE=$? # Cleanup task script diff --git a/src/madengine/execution/container_runner.py b/src/madengine/execution/container_runner.py index eab4af7d..54f71559 100644 --- a/src/madengine/execution/container_runner.py +++ b/src/madengine/execution/container_runner.py @@ -32,6 +32,7 @@ flatten_tags, ) from madengine.reporting.update_perf_super import update_perf_super_json, update_perf_super_csv +from madengine.reporting import result_csv from madengine.utils.gpu_config import resolve_runtime_gpus from madengine.deployment.common import canonicalize_distributed_launcher from madengine.utils.config_parser import ConfigParser @@ -1706,40 +1707,42 @@ def run_container( multiple_results, model_dir ) if not resolved_path: + # This is a contract, and the model script held up + # its end or it did not: say which paths were + # searched and what the container was told, so the + # next person does not have to read this code. self.rich_console.print( - f"[yellow]Warning: Could not find multiple results file " - f"(tried cwd and {model_dir}/): {multiple_results}[/yellow]" + f"[yellow]Warning: model '{model_info.get('name', '')}' " + f"declares multiple_results='{multiple_results}' but no " + f"such file was produced.[/yellow]" ) + self.rich_console.print( + f"[yellow] Searched: {os.path.abspath(multiple_results)}" + f" and {os.path.join(os.path.abspath(model_dir), multiple_results)}[/yellow]" + ) + self.rich_console.print( + f"[yellow] The container was given " + f"MAD_OUTPUT_CSV='{multiple_results}'; the run script must " + f"write a CSV there with 'performance' and 'metric' " + f"columns, relative to its working directory.[/yellow]" + ) + # A typo in the card and a script that wrote nothing + # look identical from here, so name the files that do + # look like results CSVs. + for line in result_csv.suggestion_lines( + [model_dir, os.getcwd()] + ): + self.rich_console.print(f"[yellow] {line}[/yellow]") run_results["performance"] = None else: run_results["performance"] = resolved_path - # Validate multiple results file format using proper CSV parsing - try: - import csv - with open(resolved_path, "r") as f: - csv_reader = csv.DictReader(f) - - # Strip whitespace from fieldnames to handle headers like "model, performance, metric" - csv_reader.fieldnames = [f.strip() for f in csv_reader.fieldnames] - - # Check if 'performance' column exists - if 'performance' not in csv_reader.fieldnames: - print("Error: 'performance' column not found in multiple results file.") - run_results["performance"] = None - else: - # Check if at least one row has a non-empty performance value - has_valid_perf = False - for row in csv_reader: - if row.get('performance', '').strip(): - has_valid_perf = True - break - - if not has_valid_perf: - run_results["performance"] = None - print("Error: Performance metric is empty in all rows of multiple results file.") - except Exception as e: - self.rich_console.print( - f"[yellow]Warning: Could not validate multiple results file: {e}[/yellow]" + problem = result_csv.metric_rejection_reason( + resolved_path + ) + if problem: + print( + f"Error: {resolved_path} produced no metric: " + f"{problem}." ) run_results["performance"] = None else: @@ -1782,8 +1785,23 @@ def run_container( run_results["metric"] = "samples_per_second" print(f"✓ Extracted performance (HuggingFace format): {run_results['performance']} {run_results['metric']}") else: - # No performance metrics found - print("Warning: Performance metric not found in expected format 'performance: NUMBER METRIC' or 'train_samples_per_second'") + # Nothing measured. A model reports a + # number either in a results CSV or in its + # log, and this card declared no CSV, so + # say both halves instead of only the one + # that was tried. + print( + "Warning: no metric found. The model card " + "declares no multiple_results, so " + "MAD_OUTPUT_CSV was not exported, and the " + "log has no 'performance: NUMBER METRIC' " + "or 'train_samples_per_second' line: " + f"{log_file_path}" + ) + for line in result_csv.suggestion_lines( + [model_dir, os.getcwd()] + ): + print(f" {line}") run_results["performance"] = None run_results["metric"] = None diff --git a/src/madengine/orchestration/run_orchestrator.py b/src/madengine/orchestration/run_orchestrator.py index 148c953c..3a09e4db 100644 --- a/src/madengine/orchestration/run_orchestrator.py +++ b/src/madengine/orchestration/run_orchestrator.py @@ -16,7 +16,7 @@ import shlex import subprocess from pathlib import Path -from typing import Dict, Optional +from typing import Any, Dict, List, Optional from rich.console import Console as RichConsole from rich.panel import Panel @@ -25,18 +25,74 @@ from madengine.core.auth import load_credentials from madengine.core.context import Context from madengine.core.dataprovider import Data +from madengine.core.env_file import apply_env_file from madengine.core.errors import ( BuildError, ConfigurationError, ExecutionError, create_error_context, ) +from madengine.schemas import validate_build_manifest from madengine.utils.session_tracker import SessionTracker from madengine.orchestration.image_filtering import ( filter_images_by_gpu_compatibility as _filter_by_gpu_compat, filter_images_by_skip_gpu_arch as _filter_by_skip_gpu_arch, ) +#: Blocks the deployment layer reads out of additional_context rather than out of the +#: manifest, so the manifest's copy has to be carried over to reach it. +DEPLOYMENT_CONFIG_KEYS = ( + "slurm", + "k8s", + "kubernetes", + "distributed", + "vllm", + "env_vars", + "debug", +) + + +def merge_deployment_config( + deployment_config: Dict[str, Any], additional_context: Dict[str, Any] +) -> List[str]: + """ + Carry the manifest's deployment blocks into *additional_context*, in place. + + `--additional-context` wins, and it wins whole blocks: a block given there replaces + the manifest's, so a field the manifest declares and the command line does not is + dropped rather than merged. That is worth saying out loud, because the manifest was + just reported as the place the block "belongs" and a reader will expect its values + to apply. + + Args: + deployment_config: the manifest's `deployment_config` + additional_context: what the run was given on the command line, modified in place + + Returns: + List[str]: one warning per block the command line replaced + """ + warnings: List[str] = [] + for key in DEPLOYMENT_CONFIG_KEYS: + if key not in deployment_config: + continue + if key not in additional_context: + additional_context[key] = deployment_config[key] + continue + + from_manifest = deployment_config[key] + from_cli = additional_context[key] + dropped = ( + sorted(set(from_manifest) - set(from_cli)) + if isinstance(from_manifest, dict) and isinstance(from_cli, dict) + else [] + ) + detail = f"; not applied: {', '.join(dropped)}" if dropped else "" + warnings.append( + f"--additional-context '{key}' replaces deployment_config.{key} from the " + f"manifest{detail}" + ) + return warnings + class RunOrchestrator: """ @@ -221,18 +277,36 @@ def execute( # (with optional runtime override) with open(manifest_file) as f: manifest = json.load(f) - + + # Fails fast on a malformed manifest, and folds any top-level deployment + # block into deployment_config so the target is read from one place. + for warning in validate_build_manifest(manifest, source=str(manifest_file)): + self.rich_console.print(f"[yellow]⚠ {warning}[/yellow]") + deployment_config = manifest.get("deployment_config", {}) - + + if deployment_config.get("env_file"): + applied = apply_env_file( + deployment_config["env_file"], + base_dir=str(Path(manifest_file).resolve().parent), + ) + # Names only: an env file legitimately carries secrets (MAD_SECRETS_*). + self.rich_console.print( + f"[cyan]Loaded env_file {deployment_config['env_file']}: " + f"{', '.join(sorted(applied)) or '(no new variables)'}[/cyan]" + ) + # Update additional_context with deployment_config for deployment layer if not self.additional_context: self.additional_context = {} # Merge deployment_config into additional_context (for deployment layer to use) - for key in ["slurm", "k8s", "kubernetes", "distributed", "vllm", "env_vars", "debug"]: - if key in deployment_config and key not in self.additional_context: - self.additional_context[key] = deployment_config[key] - + for warning in merge_deployment_config( + deployment_config, self.additional_context + ): + self.rich_console.print(f"[yellow]⚠ {warning}[/yellow]") + + # Display manifest entries: context (from build) and deployment_config (run/deploy) self.rich_console.print("[bold blue]Build manifest breakdown[/bold blue]\n") manifest_context = manifest.get("context", {}) @@ -749,15 +823,103 @@ def _execute_distributed(self, target: str, manifest_file: str) -> Dict: self.rich_console.print(f"[dim]{'=' * 60}[/dim]\n") - # Return metrics in the format expected by display_results_table - # Extract successful_runs and failed_runs from metrics if available - if result.metrics: - return { - "successful_runs": result.metrics.get("successful_runs", []), - "failed_runs": result.metrics.get("failed_runs", []), - } - else: - return {"successful_runs": [], "failed_runs": []} + return self._summarise_deployment(result, target, manifest_file) + + def _summarise_deployment( + self, result, target: str, manifest_file: str + ) -> Dict[str, Any]: + """Turn a deployment result into the summary the CLI reports on. + + Args: + result: what the deployment layer returned + target: deployment target name, for the records + manifest_file: manifest the run was launched from + + Returns: + Dict[str, Any]: runs by outcome, plus any warnings and the deployment status + """ + metrics = result.metrics or {} + summary = { + "successful_runs": metrics.get("successful_runs", []), + "failed_runs": list(metrics.get("failed_runs", [])), + "no_metric_runs": metrics.get("no_metric_runs", []), + "deployment_status": result.status.value, + } + + # A run that measured something despite a node ending badly is still a run that + # measured something; what the node did is a warning the caller has to see, not a + # verdict that throws the numbers away. + warnings = metrics.get("warnings") + if warnings: + summary["warnings"] = list(warnings) + + # No metric anywhere, on the other hand, leaves the node evidence as the only + # account of what happened, and it decides. + incomplete = metrics.get("incomplete") + if incomplete: + summary["incomplete"] = incomplete + summary["failed_runs"].append( + { + "model": incomplete.get("model", ""), + "status": "FAILURE", + "performance": "", + "metric": "", + "duration": "", + "gpu_arch": "", + "deployment": target, + "machine": result.deployment_id, + "error": incomplete.get("reason", ""), + } + ) + + # A job the scheduler calls failed is a failed run even when nothing could be + # parsed to say so. Without this the caller sees an empty failure list and reports + # success for a job that died. Measurements change that: the scheduler's verdict + # follows from a rank exiting non-zero, which is routine in a multi-node run whose + # framework reports from one rank only, so with results in hand it is a warning. + if not result.is_success and summary["successful_runs"] and not summary["failed_runs"]: + # The deployment layer may already have said which node ended how; only speak + # up when nothing else accounted for the scheduler's verdict. + if not summary.get("warnings"): + summary["warnings"] = [ + f"{target} reported the run as {result.status.value} " + f"({result.message}), while " + f"{len(summary['successful_runs'])} measurement(s) came back" + ] + elif not result.is_success and not summary["failed_runs"]: + summary["failed_runs"].append( + { + "model": self._model_name_from_manifest(manifest_file), + "status": "FAILURE", + "performance": "", + "metric": "", + "duration": "", + "gpu_arch": "", + "deployment": target, + "machine": result.deployment_id, + "error": result.message, + } + ) + + return summary + + @staticmethod + def _model_name_from_manifest(manifest_file: str) -> str: + """Name the run in a failure record, for a job that produced no results of its own. + + Args: + manifest_file: manifest the run was launched from + + Returns: + str: the first model's name, or "unknown" + """ + try: + with open(manifest_file) as f: + models = json.load(f).get("built_models") or {} + except (OSError, ValueError): + return "unknown" + first = next(iter(models.values()), {}) + return first.get("name") or "unknown" def _show_node_info(self): """Show node ROCm information.""" diff --git a/src/madengine/reporting/result_csv.py b/src/madengine/reporting/result_csv.py new file mode 100644 index 00000000..7603a9a7 --- /dev/null +++ b/src/madengine/reporting/result_csv.py @@ -0,0 +1,151 @@ +"""Recognise a results CSV by the shape of its header, to explain a missing metric. + +A model card declares where its results CSV lives (``multiple_results``), and that +declaration stays the only thing a run reports from. The field is optional, though, and +its value has to agree with a path a script in another repository builds by hand, so when +the two disagree a run has to say something better than "metric not found in expected +format": a results CSV is one whose header carries ``model``, ``performance`` and +``metric``, in any order and with any number of extra columns, and a run that read no +metric can at least name the files beside it that look like one. + +Nothing here chooses a file to report from. A CSV nobody declared is named in a message +and never read for a verdict, so a typo in a card and a card that declares nothing read +differently while both still fail. + +Copyright (c) Advanced Micro Devices, Inc. All rights reserved. +""" + +import csv +import os +import typing +from pathlib import Path + +#: A results CSV is recognised by these three columns, whatever else it carries. +REQUIRED_COLUMNS = ("model", "performance", "metric") + +#: madengine's own outputs satisfy the predicate by construction, so naming one as a +#: candidate would point a reader at a run's own previous verdict. +OWN_OUTPUT_NAMES = frozenset({"perf.csv"}) +OWN_OUTPUT_PREFIXES = ("perf_super", "perf_entry") + + +def normalise_columns(fieldnames: typing.Optional[typing.Iterable[str]]) -> typing.List[str]: + """Column names as they are compared: stripped of space, quotes and case.""" + return [ + (name or "").strip().strip('"').strip("'").strip().lower() + for name in (fieldnames or []) + ] + + +def read_columns(path: typing.Union[str, Path]) -> typing.Optional[typing.List[str]]: + """The normalised header of *path*, or None when it cannot be read as CSV.""" + try: + # utf-8-sig so a byte-order mark does not hide behind the first column name. + with open(path, "r", newline="", encoding="utf-8-sig", errors="ignore") as handle: + for row in csv.reader(handle): + return normalise_columns(row) + except (OSError, csv.Error): + return None + return None + + +def missing_columns(path: typing.Union[str, Path]) -> typing.Optional[typing.List[str]]: + """Which required columns *path* lacks; None when it is not readable as CSV.""" + columns = read_columns(path) + if columns is None: + return None + return [name for name in REQUIRED_COLUMNS if name not in columns] + + +def has_result_shape(path: typing.Union[str, Path]) -> bool: + """True when *path* parses as CSV and its header carries all required columns.""" + return missing_columns(path) == [] + + +def is_own_output(path: typing.Union[str, Path]) -> bool: + """True for the files madengine writes itself (perf.csv and the super/entry family).""" + name = os.path.basename(str(path)).lower() + return name in OWN_OUTPUT_NAMES or name.startswith(OWN_OUTPUT_PREFIXES) + + +def count_rows(path: typing.Union[str, Path]) -> typing.Tuple[int, int]: + """(rows with a non-empty ``performance``, total rows) in *path*.""" + with_metric = 0 + total = 0 + try: + with open(path, "r", newline="", encoding="utf-8-sig", errors="ignore") as handle: + reader = csv.DictReader(handle) + reader.fieldnames = normalise_columns(reader.fieldnames) + if "performance" not in (reader.fieldnames or []): + return (0, 0) + for row in reader: + total += 1 + value = row.get("performance") or "" + if str(value).strip(): + with_metric += 1 + except (OSError, csv.Error): + return (0, 0) + return (with_metric, total) + + +def metric_rejection_reason(path: typing.Union[str, Path]) -> typing.Optional[str]: + """Why *path* yields no metric, or None when it does. + + This is the question to ask about the file a model card named: it was declared, so it + is trusted to be the results CSV, and all that is left to check is whether a metric + can be read out of it. A card that reports through a two-column CSV keeps working, + which is why the three-column shape is not required here. + """ + columns = read_columns(path) + if columns is None: + return "not readable as CSV" + if "performance" not in columns: + found = ", ".join(columns) if columns else "(no header)" + return f"no 'performance' column; found: {found}" + if count_rows(path)[0] == 0: + return "every row has an empty 'performance' value" + return None + + +def suggest_candidates( + search_dirs: typing.Sequence[typing.Union[str, Path]], +) -> typing.List[Path]: + """CSVs lying directly in *search_dirs* whose header says they are results files. + + Depth 1 only, and only ever to phrase a message: a training run can leave hundreds of + CSVs behind, and walking the tree would turn a diagnostic into a scan. + """ + found: typing.Dict[str, Path] = {} + for directory in search_dirs: + if directory is None: + continue + as_path = Path(directory) + if not as_path.is_dir(): + continue + for entry in sorted(as_path.glob("*.csv")): + real = os.path.realpath(str(entry)) + if real in found or not entry.is_file(): + continue + if is_own_output(entry) or not has_result_shape(entry): + continue + found[real] = entry + return list(found.values()) + + +def suggestion_lines( + search_dirs: typing.Sequence[typing.Union[str, Path]], limit: int = 3 +) -> typing.List[str]: + """Lines naming the results-looking CSVs beside a run, or saying there were none.""" + where = ", ".join(str(Path(d)) for d in search_dirs if d) or "(nowhere to look)" + candidates = suggest_candidates(search_dirs) + if not candidates: + return [f"No CSV with model, performance and metric columns in: {where}"] + lines = [ + "These files look like results CSVs; declare one in the model card's " + "multiple_results:" + ] + lines.extend(f" {path}" for path in candidates[:limit]) + remaining = len(candidates) - limit + if remaining > 0: + lines.append(f" ... and {remaining} more") + return lines diff --git a/src/madengine/reporting/update_perf_csv.py b/src/madengine/reporting/update_perf_csv.py index f298efa2..d0122af4 100644 --- a/src/madengine/reporting/update_perf_csv.py +++ b/src/madengine/reporting/update_perf_csv.py @@ -14,14 +14,11 @@ # third-party imports import pandas as pd -# Standard header for perf CSV; must match ContainerRunner.ensure_perf_csv_exists() -PERF_CSV_HEADER = ( - "model,n_gpus,nnodes,gpus_per_node,training_precision,pipeline,args,tags," - "docker_file,base_docker,docker_sha,docker_image,git_commit,machine_name," - "deployment_type,launcher,gpu_architecture,performance,metric,relative_change," - "status,build_duration,test_duration,dataname,data_provider_type,data_size," - "data_download_duration,build_number,additional_docker_run_options" -) +from madengine.schemas import perf_csv_header + +# The columns are declared in schemas/perf_csv.schema.json, which every writer reads, so +# the four copies of this string can no longer drift apart. +PERF_CSV_HEADER = perf_csv_header() def df_strip_columns(df: pd.DataFrame) -> pd.DataFrame: diff --git a/src/madengine/schemas/__init__.py b/src/madengine/schemas/__init__.py new file mode 100644 index 00000000..f282af73 --- /dev/null +++ b/src/madengine/schemas/__init__.py @@ -0,0 +1,227 @@ +#!/usr/bin/env python3 +""" +Declared shapes for the files madengine reads, and the validator that enforces them. + +The build manifest used to be checked for three top-level keys and nothing else, so a +typo in a transport variable or a node count that disagreed with itself surfaced as a +failed multi-node job minutes later instead of as a startup error. `build_manifest.schema.json` +declares the shape; `validate_build_manifest` reports the first violation with its JSON +pointer, plus the cross-field checks a schema cannot express. + +The schema is also where the deployment target is defined to live: under +`deployment_config`. A manifest that carries a top-level `slurm`/`k8s`/`distributed` block +is migrated into `deployment_config` with a warning, so the two used to disagree silently +and now cannot. + +The result side is declared the same way. `perf_csv.schema.json` gives the columns of +`perf.csv` in order, replacing the header string that each of the three writers spelled +out for itself under a comment asking the reader to keep them in sync. Columns that +restate a manifest field point at it, so the row and the manifest are two views of one +shape rather than two lists that happen to agree today. + +Copyright (c) Advanced Micro Devices, Inc. All rights reserved. +""" + +import functools +import json +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple + +from madengine.core.errors import ErrorContext, ValidationError + +SCHEMA_DIR = Path(__file__).parent + +#: Blocks that describe *where* a run is deployed. They belong under `deployment_config`. +DEPLOYMENT_BLOCKS = ("slurm", "k8s", "kubernetes", "distributed") + + +def load_schema(name: str = "build_manifest.schema.json") -> Dict[str, Any]: + """Load a bundled JSON Schema by file name.""" + with open(SCHEMA_DIR / name) as f: + return json.load(f) + + +@functools.lru_cache(maxsize=1) +def perf_csv_columns() -> Tuple[str, ...]: + """ + Column names of `perf.csv`, in order. + + Order is the declaration order in `perf_csv.schema.json`: a row is written positionally + into an existing file, so the schema owns the order as well as the names. + + Returns: + Tuple[str, ...]: the column names + """ + return tuple(load_schema("perf_csv.schema.json")["properties"]) + + +@functools.lru_cache(maxsize=1) +def perf_csv_header() -> str: + """ + The `perf.csv` header line, without a trailing newline. + + Returns: + str: comma-separated column names + """ + return ",".join(perf_csv_columns()) + + +def unknown_perf_columns(row: Dict[str, Any]) -> List[str]: + """ + Report keys of a result row that no column accepts. + + A row is written with `extrasaction="ignore"`, so a key the schema does not declare is + dropped on the floor rather than reported. Callers that care can ask. + + Args: + row: a result row keyed by column name + + Returns: + List[str]: keys that are not declared columns, sorted + """ + return sorted(set(row) - set(perf_csv_columns())) + + +def _pointer(path) -> str: + """Render a jsonschema error path as a JSON pointer (RFC 6901).""" + return "/" + "/".join(str(part) for part in path) if path else "/" + + +def migrate_top_level_deployment_blocks(manifest: Dict[str, Any]) -> List[str]: + """ + Move top-level deployment blocks under `deployment_config`, in place. + + Historically a manifest could carry a top-level `slurm` block that selected the SLURM + target while the values that took effect were the ones under `deployment_config`. Fold + the former into the latter so there is one place to read. + + Args: + manifest: parsed manifest, modified in place + + Returns: + List[str]: human-readable warnings, one per migrated or ignored block + """ + warnings: List[str] = [] + for block in DEPLOYMENT_BLOCKS: + if block not in manifest: + continue + value = manifest.pop(block) + deployment_config = manifest.setdefault("deployment_config", {}) + if block in deployment_config: + warnings.append( + f"manifest carries both '{block}' at the top level and under " + f"'deployment_config'; the top-level block was ignored" + ) + else: + deployment_config[block] = value + warnings.append( + f"manifest carries '{block}' at the top level; it belongs under " + f"'deployment_config' and was moved there" + ) + return warnings + + +def _semantic_warnings(manifest: Dict[str, Any]) -> List[str]: + """Cross-field checks that are advisory rather than fatal.""" + warnings: List[str] = [] + + images = manifest.get("built_images") or {} + models = manifest.get("built_models") or {} + unused = sorted(set(images) - set(models)) + if unused: + warnings.append( + f"built_images entries with no matching built_models entry: {', '.join(unused)}" + ) + + deployment_config = manifest.get("deployment_config") or {} + target = deployment_config.get("target") + if target == "slurm" and "slurm" not in deployment_config: + warnings.append( + "deployment_config.target is 'slurm' but there is no deployment_config.slurm " + "block; cluster defaults will be used" + ) + return warnings + + +def _semantic_errors(manifest: Dict[str, Any]) -> List[str]: + """Cross-field checks a JSON Schema cannot express, and that break a run.""" + errors: List[str] = [] + + images = manifest.get("built_images") or {} + models = manifest.get("built_models") or {} + orphans = sorted(set(models) - set(images)) + if orphans: + errors.append( + f"built_models entries with no matching built_images entry: " + f"{', '.join(orphans)}. The two dicts are joined by key." + ) + + deployment_config = manifest.get("deployment_config") or {} + nodes = (deployment_config.get("slurm") or {}).get("nodes") + nnodes = (deployment_config.get("distributed") or {}).get("nnodes") + if nodes is not None and nnodes is not None and nodes != nnodes: + errors.append( + f"deployment_config.slurm.nodes ({nodes}) != " + f"deployment_config.distributed.nnodes ({nnodes}); sbatch and the launcher " + f"would disagree on the world size" + ) + return errors + + +def validate_build_manifest( + manifest: Dict[str, Any], + source: Optional[str] = None, + migrate: bool = True, +) -> List[str]: + """ + Validate a build manifest against the bundled schema, fail-fast on the first error. + + Args: + manifest: parsed manifest; modified in place when `migrate` is set + source: manifest path, for error messages + migrate: fold top-level deployment blocks into `deployment_config` + + Returns: + List[str]: non-fatal warnings + + Raises: + ValidationError: on the first schema violation or failed cross-field check + """ + import jsonschema + + warnings = migrate_top_level_deployment_blocks(manifest) if migrate else [] + + where = f" in {source}" if source else "" + validator = jsonschema.Draft202012Validator(load_schema()) + # Ordered by pointer string rather than by the raw path: a jsonschema path mixes + # property names with array indices, and comparing two of them elementwise can + # raise TypeError instead of reporting the manifest error the caller asked about. + first = next( + iter(sorted(validator.iter_errors(manifest), key=lambda e: _pointer(e.absolute_path))), + None, + ) + if first is not None: + raise ValidationError( + f"Invalid manifest{where}: {_pointer(first.absolute_path)}: {first.message}", + context=ErrorContext( + operation="manifest validation", + component="schemas", + file_path=source, + ), + suggestions=[ + "Check the field against src/madengine/schemas/build_manifest.schema.json", + ], + ) + + errors = _semantic_errors(manifest) + if errors: + raise ValidationError( + f"Invalid manifest{where}: {errors[0]}", + context=ErrorContext( + operation="manifest validation", + component="schemas", + file_path=source, + ), + ) + + return warnings + _semantic_warnings(manifest) diff --git a/src/madengine/schemas/build_manifest.schema.json b/src/madengine/schemas/build_manifest.schema.json new file mode 100644 index 00000000..9f1edfda --- /dev/null +++ b/src/madengine/schemas/build_manifest.schema.json @@ -0,0 +1,150 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://github.com/ROCm/madengine/schemas/build_manifest.schema.json", + "title": "madengine build manifest", + "description": "Shape of the manifest produced by `madengine build` and consumed by `madengine run`. Unknown keys are accepted everywhere so a manifest can carry consumer-specific metadata; the constraints below cover the fields madengine itself reads.", + "type": "object", + "required": ["built_images", "built_models", "context"], + "properties": { + "built_images": { + "description": "One entry per built image, keyed by image key. The same keys must appear in built_models.", + "type": "object", + "additionalProperties": { "$ref": "#/$defs/builtImage" } + }, + "built_models": { + "description": "One entry per model to run, keyed by the matching built_images key.", + "type": "object", + "additionalProperties": { "$ref": "#/$defs/builtModel" } + }, + "context": { "$ref": "#/$defs/context" }, + "deployment_config": { "$ref": "#/$defs/deploymentConfig" }, + "credentials_required": { "type": "array", "items": { "type": "string" } }, + "summary": { "type": "object" } + }, + "$defs": { + "envMap": { + "description": "Environment variables. Values are rendered into a shell command, so they must be scalars. The block itself may be null: that is what a model declaring no variables writes.", + "type": ["object", "null"], + "additionalProperties": { "type": ["string", "number", "boolean"] } + }, + "builtImage": { + "description": "An optional field carrying null means the model declared it as null in models.json; that is what madengine writes, so the schema accepts it.", + "type": "object", + "properties": { + "model": { "type": ["string", "null"] }, + "docker_image": { "type": ["string", "null"] }, + "dockerfile": { "type": ["string", "null"] }, + "base_docker": { "type": ["string", "null"] }, + "local_image": { "type": ["boolean", "null"] }, + "registry_image": { "type": ["string", "null"] }, + "registry": { "type": ["string", "null"] }, + "gpu_vendor": { "type": ["string", "null"] }, + "build_duration": { "type": ["number", "string", "null"] } + } + }, + "builtModel": { + "type": "object", + "required": ["name"], + "properties": { + "name": { "type": "string" }, + "url": { "type": ["string", "null"] }, + "dockerfile": { "type": ["string", "null"] }, + "scripts": { + "description": "Run script, resolved under MODEL_DIR.", + "type": ["string", "null"] + }, + "n_gpus": { "type": ["integer", "string", "null"] }, + "owner": { "type": ["string", "null"] }, + "training_precision": { "type": ["string", "null"] }, + "multiple_results": { + "description": "Name of the per-model CSV madengine collects instead of a single metric.", + "type": ["string", "null"] + }, + "tags": { "type": ["array", "null"], "items": { "type": "string" } }, + "timeout": { "type": ["integer", "string", "null"] }, + "args": { "type": ["string", "null"] }, + "additional_docker_run_options": { "type": ["string", "null"] }, + "env_vars": { "$ref": "#/$defs/envMap" }, + "slurm": { "$ref": "#/$defs/slurm" }, + "distributed": { "$ref": "#/$defs/distributed" } + } + }, + "context": { + "type": "object", + "properties": { + "docker_env_vars": { "$ref": "#/$defs/envMap" }, + "docker_mounts": { + "description": "Bind mounts keyed {container_path: host_path}; madengine renders each as -v :.", + "type": ["object", "null"], + "additionalProperties": { "type": "string" } + }, + "docker_build_arg": { "$ref": "#/$defs/envMap" }, + "docker_gpus": { "type": ["string", "integer", "null"] }, + "gpu_vendor": { "type": ["string", "null"] }, + "guest_os": { "type": ["string", "null"] }, + "skip_perf_collection": { "type": ["boolean", "null"] } + } + }, + "deploymentConfig": { + "description": "The single source of truth for where and how the run is deployed. Deployment blocks belong here, not at the top level of the manifest.", + "type": "object", + "properties": { + "target": { + "type": "string", + "enum": ["local", "docker", "slurm", "k8s", "kubernetes"] + }, + "slurm": { "$ref": "#/$defs/slurm" }, + "k8s": { "type": ["object", "null"] }, + "kubernetes": { "type": ["object", "null"] }, + "distributed": { "$ref": "#/$defs/distributed" }, + "env_vars": { "$ref": "#/$defs/envMap" }, + "env_file": { + "description": "Path to a shell env file (e.g. mad.env) that madengine sources before the run.", + "type": "string" + }, + "debug": { "type": "boolean" }, + "docker_gpus": { "type": ["string", "integer"] }, + "gpus_per_node": { "type": "integer", "minimum": 1 } + } + }, + "slurm": { + "type": ["object", "null"], + "properties": { + "partition": { "type": "string" }, + "account": { "type": "string" }, + "qos": { "type": "string" }, + "reservation": { "type": "string" }, + "nodelist": { "type": "string" }, + "exclude": { "type": "string" }, + "constraint": { "type": "string" }, + "nodes": { "type": "integer", "minimum": 1 }, + "gpus_per_node": { "type": "integer", "minimum": 1 }, + "skip_gpus_directive": { + "description": "Omit #SBATCH --gpus-per-node, for clusters that advertise no GPU GRES.", + "type": "boolean" + }, + "kill_on_bad_exit": { + "description": "Tear the srun step down as soon as one rank exits non-zero (default false). Only for workloads where every rank is expected to exit zero: where throughput is reported from one rank only, the other ranks exit non-zero on a healthy run.", + "type": "boolean" + }, + "time": { "type": "string" }, + "output_dir": { "type": "string" }, + "results_dir": { "type": "string" }, + "shared_workspace": { "type": "string" }, + "exclusive": { "type": "boolean" }, + "network_interface": { "type": "string" }, + "modules": { "type": "array", "items": { "type": "string" } } + } + }, + "distributed": { + "type": ["object", "null"], + "properties": { + "launcher": { "type": "string" }, + "backend": { "type": "string" }, + "port": { "type": "integer", "minimum": 1, "maximum": 65535 }, + "nnodes": { "type": "integer", "minimum": 1 }, + "nproc_per_node": { "type": "integer", "minimum": 1 } + } + } + } +} diff --git a/src/madengine/schemas/perf_csv.schema.json b/src/madengine/schemas/perf_csv.schema.json new file mode 100644 index 00000000..32a272fe --- /dev/null +++ b/src/madengine/schemas/perf_csv.schema.json @@ -0,0 +1,140 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://github.com/ROCm/madengine/schemas/perf_csv.schema.json", + "title": "madengine performance result row", + "description": "Shape of one row of perf.csv. Property order is column order. Columns that restate a manifest field carry x-manifest-source, a JSON pointer into build_manifest.schema.json, so the two shapes cannot be renamed apart without a test failing.", + "type": "object", + "properties": { + "model": { + "description": "Model name as it appears in the manifest.", + "type": ["string", "null"], + "x-manifest-source": "/$defs/builtModel/properties/name" + }, + "n_gpus": { + "description": "GPUs the model asked for; -1 means all visible GPUs.", + "type": ["integer", "string", "null"], + "x-manifest-source": "/$defs/builtModel/properties/n_gpus" + }, + "nnodes": { + "description": "Nodes the run spanned.", + "type": ["integer", "string", "null"], + "x-manifest-source": "/$defs/distributed/properties/nnodes" + }, + "gpus_per_node": { + "description": "GPUs used per node.", + "type": ["integer", "string", "null"], + "x-manifest-source": "/$defs/slurm/properties/gpus_per_node" + }, + "training_precision": { + "description": "Precision the model was run at, e.g. bf16, fp8.", + "type": ["string", "null"], + "x-manifest-source": "/$defs/builtModel/properties/training_precision" + }, + "pipeline": { + "description": "Pipeline the model belongs to, from its model card.", + "type": ["string", "null"] + }, + "args": { + "description": "Arguments passed to the run script.", + "type": ["string", "null"], + "x-manifest-source": "/$defs/builtModel/properties/args" + }, + "tags": { + "description": "Model card tags, flattened to a comma-separated string.", + "type": ["string", "null"], + "x-manifest-source": "/$defs/builtModel/properties/tags" + }, + "docker_file": { + "description": "Dockerfile the image was built from.", + "type": ["string", "null"], + "x-manifest-source": "/$defs/builtImage/properties/dockerfile" + }, + "base_docker": { + "description": "Base image the Dockerfile started from.", + "type": ["string", "null"], + "x-manifest-source": "/$defs/builtImage/properties/base_docker" + }, + "docker_sha": { + "description": "Image digest, as reported by the container runtime at run time.", + "type": ["string", "null"] + }, + "docker_image": { + "description": "Image tag the run used.", + "type": ["string", "null"], + "x-manifest-source": "/$defs/builtImage/properties/docker_image" + }, + "git_commit": { + "description": "Commit of the model repository the run script came from.", + "type": ["string", "null"] + }, + "machine_name": { + "description": "Host the run reported from; the head node for a multi-node run.", + "type": ["string", "null"] + }, + "deployment_type": { + "description": "Where the run was deployed.", + "type": ["string", "null"], + "x-manifest-source": "/$defs/deploymentConfig/properties/target" + }, + "launcher": { + "description": "Distributed launcher, e.g. torchrun, mpirun.", + "type": ["string", "null"], + "x-manifest-source": "/$defs/distributed/properties/launcher" + }, + "gpu_architecture": { + "description": "GPU architecture detected at run time, e.g. gfx950.", + "type": ["string", "null"] + }, + "performance": { + "description": "The metric value. Empty when the workload produced no metric.", + "type": ["number", "string", "null"] + }, + "metric": { + "description": "Unit of the performance value, e.g. tokens/s/GPU.", + "type": ["string", "null"] + }, + "relative_change": { + "description": "Change against the recorded baseline, when one exists.", + "type": ["number", "string", "null"] + }, + "status": { + "description": "Outcome of the run. SUCCESS only when the workload succeeded and a metric was collected.", + "type": ["string", "null"] + }, + "build_duration": { + "description": "Seconds spent building the image.", + "type": ["number", "string", "null"], + "x-manifest-source": "/$defs/builtImage/properties/build_duration" + }, + "test_duration": { + "description": "Seconds spent running the workload.", + "type": ["number", "string", "null"] + }, + "dataname": { + "description": "Dataset the run used.", + "type": ["string", "null"] + }, + "data_provider_type": { + "description": "How the dataset was made available, e.g. nas, minio, custom.", + "type": ["string", "null"] + }, + "data_size": { + "description": "Dataset size as reported by the provider.", + "type": ["number", "string", "null"] + }, + "data_download_duration": { + "description": "Seconds spent fetching the dataset.", + "type": ["number", "string", "null"] + }, + "build_number": { + "description": "CI build number, when the run came from CI.", + "type": ["integer", "string", "null"] + }, + "additional_docker_run_options": { + "description": "Extra options passed to docker run.", + "type": ["string", "null"], + "x-manifest-source": "/$defs/builtModel/properties/additional_docker_run_options" + } + }, + "additionalProperties": true +} diff --git a/tests/unit/test_env_file.py b/tests/unit/test_env_file.py new file mode 100644 index 00000000..fcdf2231 --- /dev/null +++ b/tests/unit/test_env_file.py @@ -0,0 +1,249 @@ +#!/usr/bin/env python3 +""" +Tests for `deployment_config.env_file` loading. + +The contract these lock down is the one an operator already relies on when they run +`source mad.env` by hand: the file is real shell, the values it sets win over whatever +was inherited, and a path that does not exist stops the run at submit time instead of +surfacing as an empty `MODEL_DIR` twenty minutes into an allocation. + +Copyright (c) Advanced Micro Devices, Inc. All rights reserved. +""" + +import os + +import pytest + +from madengine.core import env_file as env_file_module +from madengine.core.env_file import apply_env_file, load_env_file +from madengine.core.errors import ValidationError + + +@pytest.fixture +def env_dir(tmp_path): + """Directory holding env files, standing in for a run directory.""" + return tmp_path + + +@pytest.fixture(autouse=True) +def restore_environ(): + """Keep `apply_env_file` from leaking into the rest of the suite.""" + saved = os.environ.copy() + yield + os.environ.clear() + os.environ.update(saved) + + +@pytest.fixture(autouse=True) +def forget_applied_files(monkeypatch): + """Each test starts with nothing applied, as a fresh process would.""" + monkeypatch.setattr(env_file_module, "_APPLIED", {}) + + +class TestLoadEnvFile: + """Reading values out of an env file.""" + + def test_plain_assignments_are_returned(self, env_dir): + """`KEY=value` lines come back without needing an explicit export.""" + env_file = env_dir / "mad.env" + env_file.write_text("MODEL_DIR=/shared/models\nMAD_DOCKER_BUILDS=/shared/builds\n") + + loaded = load_env_file(str(env_file)) + + assert loaded["MODEL_DIR"] == "/shared/models" + assert loaded["MAD_DOCKER_BUILDS"] == "/shared/builds" + + def test_shell_constructs_are_evaluated(self, env_dir): + """The file is sourced, so expansion and defaults behave as in a shell.""" + env_file = env_dir / "mad.env" + env_file.write_text( + 'MAD_STORAGE=/shared/mad\n' + 'MAD_DOCKER_BUILDS="$MAD_STORAGE/docker_builds"\n' + 'MAD_PARTITION="${MAD_PARTITION:-gpu}"\n' + ) + + loaded = load_env_file(str(env_file)) + + assert loaded["MAD_DOCKER_BUILDS"] == "/shared/mad/docker_builds" + assert loaded["MAD_PARTITION"] == "gpu" + + def test_inherited_value_wins_over_default(self, env_dir, monkeypatch): + """A `${VAR:-default}` respects what the caller already exported.""" + monkeypatch.setenv("MAD_PARTITION", "debug") + env_file = env_dir / "mad.env" + env_file.write_text('MAD_PARTITION="${MAD_PARTITION:-gpu}"\n') + + loaded = load_env_file(str(env_file)) + + # Unchanged relative to the current environment, so nothing to report. + assert "MAD_PARTITION" not in loaded + + def test_unchanged_variables_are_not_reported(self, env_dir, monkeypatch): + """Only what the file actually changes is returned.""" + monkeypatch.setenv("MODEL_DIR", "/shared/models") + env_file = env_dir / "mad.env" + env_file.write_text("MODEL_DIR=/shared/models\nEXTRA=1\n") + + loaded = load_env_file(str(env_file)) + + assert loaded == {"EXTRA": "1"} + + def test_bash_bookkeeping_is_dropped(self, env_dir): + """`_`, `SHLVL` and friends are the subshell's, not the file's.""" + env_file = env_dir / "mad.env" + env_file.write_text("MODEL_DIR=/shared/models\n") + + loaded = load_env_file(str(env_file)) + + assert set(loaded) == {"MODEL_DIR"} + + def test_values_with_newlines_survive(self, env_dir): + """NUL-delimited output keeps multi-line values intact.""" + env_file = env_dir / "mad.env" + env_file.write_text('MAD_BANNER="line one\nline two"\n') + + loaded = load_env_file(str(env_file)) + + assert loaded["MAD_BANNER"] == "line one\nline two" + + def test_relative_path_resolves_against_base_dir(self, env_dir): + """A manifest names its env file relative to itself, not to the CWD.""" + (env_dir / "mad.env").write_text("MODEL_DIR=/shared/models\n") + + loaded = load_env_file("mad.env", base_dir=str(env_dir)) + + assert loaded["MODEL_DIR"] == "/shared/models" + + def test_absolute_path_ignores_base_dir(self, env_dir, tmp_path): + """An absolute path is taken as given.""" + env_file = env_dir / "mad.env" + env_file.write_text("MODEL_DIR=/shared/models\n") + + loaded = load_env_file(str(env_file), base_dir=str(tmp_path / "elsewhere")) + + assert loaded["MODEL_DIR"] == "/shared/models" + + def test_path_with_spaces_is_quoted(self, env_dir): + """The path reaches bash as one word.""" + directory = env_dir / "run dir" + directory.mkdir() + env_file = directory / "mad.env" + env_file.write_text("MODEL_DIR=/shared/models\n") + + loaded = load_env_file(str(env_file)) + + assert loaded["MODEL_DIR"] == "/shared/models" + + +class TestEnvFileFailures: + """A bad env file has to stop the run where it can still be explained.""" + + def test_missing_file_names_the_resolved_path(self, env_dir): + """The error shows where madengine looked, not just what the manifest said.""" + with pytest.raises(ValidationError) as exc_info: + load_env_file("absent.env", base_dir=str(env_dir)) + + assert str(env_dir / "absent.env") in str(exc_info.value) + + def test_directory_is_not_a_file(self, env_dir): + """Pointing at a directory fails the same way a missing file does.""" + with pytest.raises(ValidationError): + load_env_file(str(env_dir)) + + def test_shell_error_is_reported(self, env_dir): + """A non-zero exit from bash carries stderr into the message.""" + env_file = env_dir / "mad.env" + env_file.write_text("exit 3\n") + + with pytest.raises(ValidationError) as exc_info: + load_env_file(str(env_file)) + + assert "Failed to source" in str(exc_info.value) + + def test_syntax_error_is_reported(self, env_dir): + """Malformed shell is a fatal, named error.""" + env_file = env_dir / "mad.env" + env_file.write_text('MODEL_DIR="/unterminated\n') + + with pytest.raises(ValidationError): + load_env_file(str(env_file)) + + +class TestApplyEnvFile: + """Applying the file to the running process.""" + + def test_variables_land_in_environ(self, env_dir): + """What the file sets is visible to everything downstream.""" + env_file = env_dir / "mad.env" + env_file.write_text("MAD_DOCKER_BUILDS=/shared/builds\n") + + apply_env_file(str(env_file)) + + assert os.environ["MAD_DOCKER_BUILDS"] == "/shared/builds" + + def test_file_wins_over_inherited_value(self, env_dir, monkeypatch): + """Sourcing overwrites, and so does this.""" + monkeypatch.setenv("MAD_DOCKER_BUILDS", "/tmp/stale") + env_file = env_dir / "mad.env" + env_file.write_text("MAD_DOCKER_BUILDS=/shared/builds\n") + + apply_env_file(str(env_file)) + + assert os.environ["MAD_DOCKER_BUILDS"] == "/shared/builds" + + def test_unrelated_variables_are_left_alone(self, env_dir, monkeypatch): + """Loading an env file is additive, not a replacement of the environment.""" + monkeypatch.setenv("MAD_KEEP_ME", "yes") + env_file = env_dir / "mad.env" + env_file.write_text("MODEL_DIR=/shared/models\n") + + apply_env_file(str(env_file)) + + assert os.environ["MAD_KEEP_ME"] == "yes" + + +class TestAppliedOncePerProcess: + """A submit-side run reaches the same file twice; the file must run once.""" + + def test_an_append_does_not_happen_twice(self, env_dir, monkeypatch): + """`PATH="$PATH:/opt/x"` is the reason this is not simply idempotent.""" + monkeypatch.setenv("PATH", "/usr/bin") + env_file = env_dir / "mad.env" + env_file.write_text('PATH="$PATH:/opt/x"\n') + + apply_env_file(str(env_file)) + apply_env_file(str(env_file)) + + assert os.environ["PATH"] == "/usr/bin:/opt/x" + + def test_the_second_call_reports_what_the_first_applied(self, env_dir): + """The caller logs the names either way, so both calls return the same map.""" + env_file = env_dir / "mad.env" + env_file.write_text("MODEL_DIR=/shared/models\n") + + assert apply_env_file(str(env_file)) == apply_env_file(str(env_file)) + + def test_the_same_file_under_two_spellings_runs_once(self, env_dir): + """The manifest names it relatively, the deployment layer absolutely.""" + (env_dir / "mad.env").write_text('MAD_COUNTER="${MAD_COUNTER:-}x"\n') + + apply_env_file("mad.env", base_dir=str(env_dir)) + apply_env_file(str(env_dir / "mad.env")) + + assert os.environ["MAD_COUNTER"] == "x" + + def test_a_different_file_is_still_applied(self, env_dir): + (env_dir / "first.env").write_text("MAD_FIRST=1\n") + (env_dir / "second.env").write_text("MAD_SECOND=2\n") + + apply_env_file(str(env_dir / "first.env")) + apply_env_file(str(env_dir / "second.env")) + + assert os.environ["MAD_FIRST"] == "1" + assert os.environ["MAD_SECOND"] == "2" + + def test_a_missing_file_still_raises_every_time(self, env_dir): + """Nothing is remembered about a file that was never sourced.""" + for _ in range(2): + with pytest.raises(ValidationError): + apply_env_file(str(env_dir / "absent.env")) diff --git a/tests/unit/test_manifest_schema.py b/tests/unit/test_manifest_schema.py new file mode 100644 index 00000000..017e9316 --- /dev/null +++ b/tests/unit/test_manifest_schema.py @@ -0,0 +1,201 @@ +#!/usr/bin/env python3 +""" +Unit tests for build-manifest schema validation. + +The manifest used to be checked for three top-level keys, so the failures below all +surfaced minutes later as a failed multi-node job. Each test pins one of them to a +startup error instead: + +1. A field with the wrong type is reported with its JSON pointer. +2. `built_models` and `built_images` are joined by key, so an orphan model is fatal. +3. `slurm.nodes` and `distributed.nnodes` must agree or sbatch and the launcher + disagree on the world size. +4. A top-level deployment block is folded into `deployment_config`, which is the one + place the target is read from. + +Copyright (c) Advanced Micro Devices, Inc. All rights reserved. +""" + +import copy + +import pytest + +from madengine.core.errors import ValidationError +from madengine.schemas import ( + load_schema, + migrate_top_level_deployment_blocks, + validate_build_manifest, +) + + +VALID_MANIFEST = { + "built_images": { + "img": { + "model": "dummy", + "docker_image": "rocm/dummy:latest", + "dockerfile": "docker/dummy.ubuntu.amd.Dockerfile", + "local_image": True, + } + }, + "built_models": { + "img": { + "name": "dummy", + "scripts": "scripts/dummy/run.sh", + "n_gpus": "-1", + "tags": ["pyt", "training"], + "timeout": -1, + "args": "--model_repo dummy", + "multiple_results": "perf_dummy.csv", + } + }, + "context": { + "docker_env_vars": {"NCCL_DEBUG": "INFO"}, + "docker_mounts": {"/dev/infiniband": "/dev/infiniband"}, + "docker_build_arg": {}, + "gpu_vendor": "AMD", + "guest_os": "UBUNTU", + "docker_gpus": "0,1,2,3,4,5,6,7", + }, + "deployment_config": { + "target": "slurm", + "slurm": {"partition": "gpu", "nodes": 2, "gpus_per_node": 8}, + "distributed": {"launcher": "torchrun", "nnodes": 2, "nproc_per_node": 8}, + "env_vars": {}, + }, +} + + +@pytest.fixture +def manifest(): + return copy.deepcopy(VALID_MANIFEST) + + +class TestSchemaIsWellFormed: + def test_schema_loads(self): + schema = load_schema() + assert schema["$schema"].startswith("https://json-schema.org/draft/2020-12") + assert set(schema["required"]) == {"built_images", "built_models", "context"} + + def test_valid_manifest_passes_without_warnings(self, manifest): + assert validate_build_manifest(manifest) == [] + + +class TestSchemaViolations: + def test_wrong_type_is_reported_with_json_pointer(self, manifest): + manifest["deployment_config"]["slurm"]["nodes"] = "2" + with pytest.raises(ValidationError) as excinfo: + validate_build_manifest(manifest) + assert "/deployment_config/slurm/nodes" in str(excinfo.value) + + def test_nested_env_var_value_is_rejected(self, manifest): + """Env values are rendered into a shell command, so they must be scalars.""" + manifest["context"]["docker_env_vars"]["NCCL_IB_HCA"] = {"device": "mlx5_0"} + with pytest.raises(ValidationError) as excinfo: + validate_build_manifest(manifest) + assert "/context/docker_env_vars/NCCL_IB_HCA" in str(excinfo.value) + + def test_unknown_deployment_target_is_rejected(self, manifest): + manifest["deployment_config"]["target"] = "sluurm" + with pytest.raises(ValidationError) as excinfo: + validate_build_manifest(manifest) + assert "/deployment_config/target" in str(excinfo.value) + + def test_unknown_keys_are_allowed(self, manifest): + """A manifest may carry consumer-specific metadata madengine does not read.""" + manifest["built_models"]["img"]["consumer_note"] = "keep me" + manifest["rccl_ci"] = {"branch": "develop"} + assert validate_build_manifest(manifest) == [] + + def test_null_optional_fields_are_accepted(self, manifest): + """`madengine build` copies a null from models.json straight through. + + The schema describes what madengine writes, so rejecting its own output would only + break runs that worked: a model declaring `"timeout": null` is not a broken manifest. + """ + manifest["built_models"]["img"].update( + { + "timeout": None, + "scripts": None, + "n_gpus": None, + "args": None, + "multiple_results": None, + "tags": None, + "env_vars": None, + "slurm": None, + "distributed": None, + } + ) + manifest["built_images"]["img"]["dockerfile"] = None + manifest["context"]["docker_gpus"] = None + + assert validate_build_manifest(manifest) == [] + + def test_error_names_the_source_file(self, manifest): + manifest["deployment_config"]["distributed"]["port"] = 70000 + with pytest.raises(ValidationError) as excinfo: + validate_build_manifest(manifest, source="build_manifest.json") + assert "build_manifest.json" in str(excinfo.value) + + +class TestCrossFieldChecks: + def test_model_without_matching_image_is_fatal(self, manifest): + manifest["built_models"]["orphan"] = {"name": "orphan"} + with pytest.raises(ValidationError) as excinfo: + validate_build_manifest(manifest) + assert "orphan" in str(excinfo.value) + + def test_node_count_mismatch_is_fatal(self, manifest): + manifest["deployment_config"]["distributed"]["nnodes"] = 4 + with pytest.raises(ValidationError) as excinfo: + validate_build_manifest(manifest) + assert "world size" in str(excinfo.value) + + def test_image_without_model_is_a_warning(self, manifest): + manifest["built_images"]["spare"] = {"docker_image": "rocm/spare:latest"} + warnings = validate_build_manifest(manifest) + assert any("spare" in w for w in warnings) + + def test_slurm_target_without_slurm_block_is_a_warning(self, manifest): + del manifest["deployment_config"]["slurm"] + warnings = validate_build_manifest(manifest) + assert any("cluster defaults" in w for w in warnings) + + +class TestSingleSourceOfTruthForDeploymentTarget: + """A top-level deployment block used to select the target while its values were ignored.""" + + def test_top_level_block_is_moved_under_deployment_config(self, manifest): + top_level = {"partition": "batch", "nodes": 2} + manifest["slurm"] = top_level + del manifest["deployment_config"]["slurm"] + + warnings = validate_build_manifest(manifest) + + assert "slurm" not in manifest + assert manifest["deployment_config"]["slurm"] == top_level + assert any("belongs under" in w for w in warnings) + + def test_deployment_config_wins_when_both_are_present(self, manifest): + manifest["slurm"] = {"partition": "ignored-partition", "nodes": 2} + + warnings = validate_build_manifest(manifest) + + assert "slurm" not in manifest + assert manifest["deployment_config"]["slurm"]["partition"] == "gpu" + assert any("was ignored" in w for w in warnings) + + def test_migration_can_be_disabled(self, manifest): + manifest["slurm"] = {"partition": "batch", "nodes": 2} + validate_build_manifest(manifest, migrate=False) + assert "slurm" in manifest + + def test_every_deployment_block_is_migrated(self): + manifest = { + "built_images": {}, + "built_models": {}, + "context": {}, + "distributed": {"launcher": "torchrun", "nnodes": 1}, + "kubernetes": {"namespace": "mad"}, + } + migrate_top_level_deployment_blocks(manifest) + assert set(manifest["deployment_config"]) == {"distributed", "kubernetes"} diff --git a/tests/unit/test_multinode_failure.py b/tests/unit/test_multinode_failure.py new file mode 100644 index 00000000..aeb58ebe --- /dev/null +++ b/tests/unit/test_multinode_failure.py @@ -0,0 +1,316 @@ +#!/usr/bin/env python3 +""" +Tests for reporting the truth about a multi-node run. + +A two-node run that ended with the scheduler recording exit code 3 used to finish with +"All model executions completed successfully!", because the only thing anyone looked at was +whether a number could be parsed out of a log. Three separate defects made that possible: +per-node exit codes were never collected, nothing distinguished a crashed workload from a +missing metric, and a failed deployment whose results could not be parsed produced an empty +failure list. These tests pin all three. + +Copyright (c) Advanced Micro Devices, Inc. All rights reserved. +""" + +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + +from madengine.deployment.slurm import SlurmDeployment + + +@pytest.fixture +def deployment(): + """A SLURM deployment with the scheduler and manifest stubbed out.""" + instance = object.__new__(SlurmDeployment) + instance.nodes = 2 + instance.gpus_per_node = 8 + instance.console = MagicMock() + return instance + + +def write_status(job_dir, node, exit_code, host="node-a"): + """Write the marker a task script leaves behind.""" + node_dir = job_dir / f"node_{node}" + node_dir.mkdir(parents=True, exist_ok=True) + (node_dir / "node.status").write_text( + f"node={node}\nhost={host}\nexit_code={exit_code}\ntimestamp=2026-08-06T12:00:00+00:00\n" + ) + return node_dir + + +class TestReadingNodeStatuses: + """Per-node exit codes, which nothing collected before.""" + + def test_statuses_are_read_in_node_order(self, deployment, tmp_path): + """Node 10 sorts after node 2, unlike its directory name.""" + deployment.nodes = 11 + for node in (0, 2, 10): + write_status(tmp_path, node, 0) + + statuses = deployment._read_node_statuses(tmp_path) + + assert [s["node"] for s in statuses] == [0, 2, 10] + + def test_exit_code_and_host_are_kept(self, deployment, tmp_path): + """Which node failed matters as much as that one did.""" + write_status(tmp_path, 1, 137, host="node-b") + + status = deployment._read_node_statuses(tmp_path)[0] + + assert status["exit_code"] == 137 + assert status["host"] == "node-b" + + def test_missing_markers_are_not_invented(self, deployment, tmp_path): + """A job from before these markers existed reports nothing, not failure.""" + assert deployment._read_node_statuses(tmp_path) == [] + + def test_unparseable_marker_is_skipped(self, deployment, tmp_path): + """A truncated marker must not take down result collection.""" + node_dir = tmp_path / "node_0" + node_dir.mkdir() + (node_dir / "node.status").write_text("exit_code=not-a-number\n") + + assert deployment._read_node_statuses(tmp_path) == [] + + +class TestReportingNodeStatuses: + """Turning markers into a verdict.""" + + def test_all_zero_is_no_failure(self, deployment, tmp_path): + """The happy path stays quiet.""" + write_status(tmp_path, 0, 0) + write_status(tmp_path, 1, 0) + + assert deployment._report_node_statuses(deployment._read_node_statuses(tmp_path)) == [] + + def test_non_zero_exit_is_a_failure(self, deployment, tmp_path): + """The case that used to be reported as success.""" + write_status(tmp_path, 0, 0) + write_status(tmp_path, 1, 3, host="node-b") + + failures = deployment._report_node_statuses(deployment._read_node_statuses(tmp_path)) + + assert [f["node"] for f in failures] == [1] + assert failures[0]["exit_code"] == 3 + + def test_a_node_that_never_reported_is_a_failure(self, deployment, tmp_path): + """A node killed mid-run writes no marker; silence is not success.""" + write_status(tmp_path, 0, 0) + + failures = deployment._report_node_statuses(deployment._read_node_statuses(tmp_path)) + + assert [f["node"] for f in failures] == [1] + assert failures[0]["missing"] is True + + def test_no_markers_at_all_means_no_verdict(self, deployment, tmp_path): + """Without a single marker there is nothing to conclude from their absence.""" + assert deployment._report_node_statuses([]) == [] + + +class TestIncompleteReason: + """A verdict has to say what happened, not just that something did.""" + + def test_a_failed_node_is_named_with_its_exit_code(self, deployment): + """"Execution failed" alone sends the reader back into the logs.""" + reason = deployment._incomplete_reason( + [{"node": 0, "host": "node-a", "exit_code": 3}], "FAILED" + ) + + assert reason == "node 0 exited 3" + + def test_a_lost_node_is_described_as_lost(self, deployment): + """A node with no marker did not exit; it disappeared.""" + reason = deployment._incomplete_reason( + [{"node": 1, "host": "", "exit_code": None, "missing": True}], None + ) + + assert reason == "node 1 reported no status (killed or lost)" + + def test_every_failing_node_is_listed(self, deployment): + """Two nodes down is a different story from one.""" + reason = deployment._incomplete_reason( + [ + {"node": 0, "host": "node-a", "exit_code": 3}, + {"node": 1, "host": "node-b", "exit_code": 137}, + ], + "FAILED", + ) + + assert reason == "node 0 exited 3, node 1 exited 137" + + def test_without_markers_slurm_speaks(self, deployment): + """The step the scheduler killed leaves nothing behind but its state.""" + assert deployment._incomplete_reason([], "TIMEOUT") == ( + "SLURM reports the job as TIMEOUT" + ) + + def test_with_no_evidence_at_all_the_verdict_stays_plain(self, deployment): + """Nothing to quote, so claim nothing specific.""" + assert deployment._incomplete_reason([], None) == "the job did not complete" + + +class TestVerdictWhenANodeEndsBadly: + """A metric outweighs an exit code, as it already does for a single node.""" + + def test_a_measured_run_keeps_its_numbers_and_gains_a_warning(self, deployment): + """The rank that reports no throughput exits non-zero on a healthy run.""" + results: dict = {} + + deployment._record_verdict( + results, "llama", [{"node": 0, "host": "node-a", "exit_code": 3}], "FAILED", 8 + ) + + assert "incomplete" not in results + assert len(results["warnings"]) == 1 + assert "node 0 exited 3" in results["warnings"][0] + assert "8 metric row" in results["warnings"][0] + + def test_a_run_with_nothing_measured_is_a_failure(self, deployment): + """With no metric anywhere the node evidence is all there is.""" + results: dict = {} + + deployment._record_verdict( + results, "llama", [{"node": 1, "host": "node-b", "exit_code": 137}], "FAILED", 0 + ) + + assert "warnings" not in results + assert results["incomplete"]["reason"] == "node 1 exited 137" + assert results["incomplete"]["model"] == "llama" + + def test_a_lost_node_is_named_in_the_warning(self, deployment): + """Whatever the verdict, the operator hears which node went missing.""" + results: dict = {} + + deployment._record_verdict( + results, "llama", [{"node": 1, "exit_code": None, "missing": True}], None, 4 + ) + + assert "reported no status" in results["warnings"][0] + + +class TestSummaryTheCliReportsOn: + """What the orchestrator hands the CLI, for a job the scheduler calls failed.""" + + @staticmethod + def _summarise(metrics, is_success=False): + from madengine.orchestration.run_orchestrator import RunOrchestrator + + orchestrator = object.__new__(RunOrchestrator) + result = MagicMock() + result.metrics = metrics + result.is_success = is_success + result.status.value = "failed" + result.message = "Job 24505 failed: FAILED" + result.deployment_id = "24505" + return orchestrator._summarise_deployment(result, "slurm", "manifest.json") + + def test_measurements_survive_a_failed_job_state(self): + """The scheduler's verdict follows from a rank that exits non-zero by design.""" + summary = self._summarise({"successful_runs": [{"model": "llama"}]}) + + assert summary["failed_runs"] == [] + assert "measurement(s) came back" in summary["warnings"][0] + + def test_the_deployment_does_not_repeat_a_warning_already_made(self): + """The node-level warning is the specific one; two say no more than one.""" + summary = self._summarise( + {"successful_runs": [{"model": "llama"}], "warnings": ["node 0 exited 3"]} + ) + + assert summary["warnings"] == ["node 0 exited 3"] + + def test_a_failed_job_with_nothing_measured_is_still_a_failure(self): + """Otherwise the caller sees an empty failure list and reports success.""" + summary = self._summarise({"successful_runs": []}) + + assert len(summary["failed_runs"]) == 1 + assert summary["failed_runs"][0]["error"] == "Job 24505 failed: FAILED" + + def test_an_incomplete_run_names_the_model_that_did_not_finish(self): + """The deployment layer's own verdict, carried through.""" + summary = self._summarise( + {"successful_runs": [], "incomplete": {"model": "llama", "reason": "node 1 exited 9"}} + ) + + assert summary["incomplete"]["reason"] == "node 1 exited 9" + assert [f["model"] for f in summary["failed_runs"]] == ["llama"] + + +class TestJobExitState: + """SLURM's own verdict, for the failures that leave no marker.""" + + def test_state_is_read_from_sacct(self, deployment): + """The state, upper-cased, as sacct reports it.""" + with patch("madengine.deployment.slurm.subprocess.run") as run: + run.return_value = MagicMock(returncode=0, stdout="COMPLETED\n") + + assert deployment._job_exit_state("123") == "COMPLETED" + + def test_failed_state_is_reported(self, deployment): + """A job the scheduler killed.""" + with patch("madengine.deployment.slurm.subprocess.run") as run: + run.return_value = MagicMock(returncode=0, stdout="CANCELLED by 1001\n") + + assert deployment._job_exit_state("123") == "CANCELLED BY 1001" + + def test_unavailable_sacct_claims_nothing(self, deployment): + """No accounting database is not evidence either way.""" + with patch("madengine.deployment.slurm.subprocess.run") as run: + run.return_value = MagicMock(returncode=1, stdout="") + + assert deployment._job_exit_state("123") is None + + def test_exception_claims_nothing(self, deployment): + """Neither is a missing sacct binary.""" + with patch("madengine.deployment.slurm.subprocess.run", side_effect=OSError): + assert deployment._job_exit_state("123") is None + + +class TestJobScriptRecordsOutcomes: + """What the rendered job script does on the nodes.""" + + @pytest.fixture + def rendered(self, tmp_path): + """The two-node job script, rendered exactly as prepare() renders it.""" + from tests.unit.test_slurm_job_template import ( + MODEL_ENTRY, + _build_deployment, + ) + + deployment = _build_deployment(tmp_path) + context = deployment._prepare_template_context(MODEL_ENTRY) + return deployment.jinja_env.get_template("job.sh.j2").render(**context) + + def test_every_node_records_its_exit_code(self, rendered): + """The marker the submit node reads back.""" + assert "exit_code=${TASK_EXIT}" in rendered + assert "node.status" in rendered + + def test_the_step_is_not_torn_down_on_a_bad_exit_by_default(self, rendered): + """The rank that collects nothing exits non-zero on a healthy multi-node run. + + Tearing the step down there would kill the rank that holds the numbers, so the + teardown is something a workload opts into. + """ + assert "--kill-on-bad-exit" not in rendered.split("srun ")[-1] + + def test_a_failing_run_still_reaches_the_collection_block(self, rendered): + """Under errexit the script died on the madengine line, losing marker and artifacts.""" + madengine_call = rendered.index("$MAD_CLI_COMMAND run") + assert rendered.rindex("set +e", 0, madengine_call) < madengine_call + assert "TASK_EXIT=$?\nset -e" in rendered + + def test_teardown_can_be_turned_on(self, tmp_path): + """A workload where every rank must exit zero can stop holding the allocation.""" + from tests.unit.test_slurm_job_template import ( + MODEL_ENTRY, + _build_deployment, + ) + + deployment = _build_deployment(tmp_path, {"kill_on_bad_exit": True}) + context = deployment._prepare_template_context(MODEL_ENTRY) + rendered = deployment.jinja_env.get_template("job.sh.j2").render(**context) + + assert "srun --kill-on-bad-exit=1" in rendered diff --git a/tests/unit/test_orchestration.py b/tests/unit/test_orchestration.py index 54a11bd5..652a6e30 100644 --- a/tests/unit/test_orchestration.py +++ b/tests/unit/test_orchestration.py @@ -14,7 +14,10 @@ DEFAULT_GUEST_OS, ) from madengine.orchestration.build_orchestrator import BuildOrchestrator -from madengine.orchestration.run_orchestrator import RunOrchestrator +from madengine.orchestration.run_orchestrator import ( + RunOrchestrator, + merge_deployment_config, +) from madengine.core.errors import ConfigurationError @@ -209,6 +212,7 @@ def test_skip_after_build_calls_execute_local(self, mock_cleanup, tmp_path): "deployment_config": {"target": "local"}, "context": {}, "built_images": {}, + "built_models": {}, } ) ) @@ -253,6 +257,7 @@ def test_skip_run_only_still_calls_execute_local( "deployment_config": {"target": "local"}, "context": {}, "built_images": {}, + "built_models": {}, } ) ) @@ -286,6 +291,7 @@ def test_skip_model_run_calls_execute_local(self, mock_cleanup, tmp_path): "deployment_config": {"target": "local"}, "context": {}, "built_images": {}, + "built_models": {}, } ) ) @@ -359,3 +365,44 @@ def test_distributed_warns_on_local_only_flags(self, tmp_path): assert "--keep-alive" in printed assert "--skip-model-run" in printed assert "--keep-model-dir" not in printed # was False, must not appear + + +class TestMergeDeploymentConfig: + """What the manifest's deployment blocks do when the command line names them too.""" + + def test_a_block_the_command_line_does_not_name_is_carried_over(self): + context = {} + warnings = merge_deployment_config({"slurm": {"partition": "gpu"}}, context) + assert context["slurm"] == {"partition": "gpu"} + assert warnings == [] + + def test_the_command_line_wins_and_says_so(self): + context = {"slurm": {"partition": "debug"}} + warnings = merge_deployment_config({"slurm": {"partition": "gpu"}}, context) + assert context["slurm"] == {"partition": "debug"} + assert len(warnings) == 1 + assert "deployment_config.slurm" in warnings[0] + + def test_fields_only_the_manifest_declares_are_named(self): + """The whole block is replaced, so `qos` never reaches the deployment layer.""" + context = {"slurm": {"partition": "debug"}} + warnings = merge_deployment_config( + {"slurm": {"partition": "gpu", "qos": "high", "nodes": 2}}, context + ) + assert "not applied: nodes, qos" in warnings[0] + + def test_an_identical_block_still_warns_but_names_nothing(self): + context = {"slurm": {"partition": "gpu"}} + warnings = merge_deployment_config({"slurm": {"partition": "gpu"}}, context) + assert "not applied" not in warnings[0] + + def test_a_non_dict_block_is_reported_without_a_field_list(self): + context = {"debug": False} + warnings = merge_deployment_config({"debug": True}, context) + assert context["debug"] is False + assert "not applied" not in warnings[0] + + def test_blocks_madengine_does_not_carry_are_left_alone(self): + context = {} + merge_deployment_config({"target": "slurm", "env_file": "mad.env"}, context) + assert context == {} diff --git a/tests/unit/test_perf_csv_schema.py b/tests/unit/test_perf_csv_schema.py new file mode 100644 index 00000000..1336019a --- /dev/null +++ b/tests/unit/test_perf_csv_schema.py @@ -0,0 +1,204 @@ +#!/usr/bin/env python3 +""" +Tests for the declared shape of a perf.csv row. + +Three writers create perf.csv — the container runner, the SLURM/base deployment path and +the Kubernetes results mixin — and each used to carry its own copy of the header under a +comment asking the reader to keep them in sync. These tests hold the writers to the +schema, and hold the schema to the manifest it borrows field names from. + +Copyright (c) Advanced Micro Devices, Inc. All rights reserved. +""" + +import json + +import pytest + +from madengine.schemas import ( + load_schema, + perf_csv_columns, + perf_csv_header, + unknown_perf_columns, +) + + +@pytest.fixture(scope="module") +def perf_schema(): + """The declared result-row shape.""" + return load_schema("perf_csv.schema.json") + + +@pytest.fixture(scope="module") +def manifest_schema(): + """The declared manifest shape the result row borrows names from.""" + return load_schema("build_manifest.schema.json") + + +def _resolve_pointer(document, pointer): + """Resolve an RFC 6901 JSON pointer, returning None when it does not exist.""" + node = document + for part in pointer.lstrip("/").split("/"): + part = part.replace("~1", "/").replace("~0", "~") + if not isinstance(node, dict) or part not in node: + return None + node = node[part] + return node + + +class TestPerfCsvColumns: + """The column list itself.""" + + def test_header_is_the_columns_in_order(self): + """The header line is the declaration order, not an independently kept string.""" + assert perf_csv_header() == ",".join(perf_csv_columns()) + + def test_columns_match_the_published_order(self): + """Rows are appended positionally into existing files; the order is the contract.""" + assert perf_csv_columns() == ( + "model", + "n_gpus", + "nnodes", + "gpus_per_node", + "training_precision", + "pipeline", + "args", + "tags", + "docker_file", + "base_docker", + "docker_sha", + "docker_image", + "git_commit", + "machine_name", + "deployment_type", + "launcher", + "gpu_architecture", + "performance", + "metric", + "relative_change", + "status", + "build_duration", + "test_duration", + "dataname", + "data_provider_type", + "data_size", + "data_download_duration", + "build_number", + "additional_docker_run_options", + ) + + def test_every_column_is_documented(self, perf_schema): + """A column nobody can explain is a column nobody can consume.""" + undocumented = [ + name + for name, spec in perf_schema["properties"].items() + if not spec.get("description") + ] + assert undocumented == [] + + def test_no_duplicate_columns(self): + """JSON object keys make this hard to get wrong; say so anyway.""" + assert len(perf_csv_columns()) == len(set(perf_csv_columns())) + + def test_schema_is_valid(self, perf_schema): + """The schema itself has to be a schema.""" + import jsonschema + + jsonschema.Draft202012Validator.check_schema(perf_schema) + + +class TestManifestLinkage: + """Columns that restate a manifest field say which one.""" + + def test_manifest_pointers_resolve(self, perf_schema, manifest_schema): + """Renaming a manifest field without updating the result contract fails here.""" + dangling = { + column: spec["x-manifest-source"] + for column, spec in perf_schema["properties"].items() + if "x-manifest-source" in spec + and _resolve_pointer(manifest_schema, spec["x-manifest-source"]) is None + } + assert dangling == {} + + def test_linked_columns_cover_the_obvious_ones(self, perf_schema): + """The fields an operator reads off a manifest are the ones that must stay linked.""" + linked = { + column + for column, spec in perf_schema["properties"].items() + if "x-manifest-source" in spec + } + assert {"model", "docker_image", "nnodes", "launcher", "deployment_type"} <= linked + + def test_pointers_reach_a_declared_property(self, perf_schema, manifest_schema): + """A pointer must land on a property definition, not on some intermediate node.""" + for column, spec in perf_schema["properties"].items(): + pointer = spec.get("x-manifest-source") + if pointer is None: + continue + target = _resolve_pointer(manifest_schema, pointer) + assert isinstance(target, dict), column + assert "type" in target or "$ref" in target, column + + +class TestWritersUseTheSchema: + """No writer keeps its own copy of the header.""" + + def test_reporting_header_comes_from_the_schema(self): + """update_perf_csv exports the header other code imports.""" + from madengine.reporting.update_perf_csv import PERF_CSV_HEADER + + assert PERF_CSV_HEADER == perf_csv_header() + + def test_deployment_writes_the_schema_header(self, tmp_path, monkeypatch): + """The SLURM/base aggregation path creates perf.csv with the declared columns.""" + from madengine.deployment.base import BaseDeployment + + monkeypatch.chdir(tmp_path) + # The method touches no instance state, and BaseDeployment is abstract. + BaseDeployment._ensure_perf_csv_exists(None) + + assert (tmp_path / "perf.csv").read_text().strip() == perf_csv_header() + + def test_existing_file_is_left_alone(self, tmp_path, monkeypatch): + """A perf.csv from an earlier run keeps its own column order.""" + from madengine.deployment.base import BaseDeployment + + monkeypatch.chdir(tmp_path) + (tmp_path / "perf.csv").write_text("model,performance\n") + BaseDeployment._ensure_perf_csv_exists(None) + + assert (tmp_path / "perf.csv").read_text() == "model,performance\n" + + def test_no_writer_hardcodes_the_column_list(self): + """The header string should exist once, in the schema.""" + from pathlib import Path + + import madengine + + src = Path(madengine.__file__).parent + needle = "data_download_duration,build_number" + offenders = [ + str(path.relative_to(src)) + for path in src.rglob("*.py") + if needle in path.read_text(encoding="utf-8", errors="replace") + ] + assert offenders == [] + + +class TestUnknownColumns: + """Rows are written with extrasaction='ignore'; callers can still ask what was dropped.""" + + def test_declared_keys_are_not_reported(self): + """A row made of columns reports nothing.""" + row = {name: "" for name in perf_csv_columns()} + + assert unknown_perf_columns(row) == [] + + def test_undeclared_keys_are_reported(self): + """A typo'd key would otherwise vanish silently into the CSV writer.""" + row = {"model": "llama", "perfromance": 1.0, "nnodes": 2} + + assert unknown_perf_columns(row) == ["perfromance"] + + def test_partial_rows_are_fine(self): + """Missing columns are normal: not every run has data provider fields.""" + assert unknown_perf_columns({"model": "llama"}) == [] diff --git a/tests/unit/test_result_csv.py b/tests/unit/test_result_csv.py new file mode 100644 index 00000000..fcf7f567 --- /dev/null +++ b/tests/unit/test_result_csv.py @@ -0,0 +1,186 @@ +"""Recognising a results CSV by its header, and saying so when a run read no metric. + +The header fixtures here are copied verbatim out of MAD's own scripts rather than +invented, so a change in what the scripts emit shows up as a failure here: + +- ``model,performance,metric`` scripts/dummy/run_multi.sh:1 +- ``model, performance, metric`` scripts/pyt_chai1_inference/run.sh:39 +- ``model,performance,metric,mode,precision,...`` scripts/primus_megatron-lm/ + primus_megatron-lm_benchmark_report.sh:283 +- ``hf_pipeline_tag,model,...,performance,metric,unit`` scripts/atom/run_atom.py:43 + +and the near-misses that must stay rejected: + +- ``Model,xP/yD,ISL,...`` scripts/sglang_disagg/benchmark_parser.py:188 +- ``model_name,model_unique_name,...`` scripts/kvcache_transfer_bench/ + kv_cache_estimator.py:1319 + +Copyright (c) Advanced Micro Devices, Inc. All rights reserved. +""" + +from pathlib import Path + +import pytest + +from madengine.reporting import result_csv + + +# Real headers, as the scripts write them. +DUMMY_HEADER = "model,performance,metric" +CHAI_HEADER = "model, performance, metric" +PRIMUS_HEADER = ( + "model,performance,metric,mode,precision,batch_size,global_batch_size," + "seq_len,device,num_gpus" +) +# The three columns are neither first nor adjacent here. +ATOM_HEADER = ( + "hf_pipeline_tag,model,benchmark,tp,inp,out,kv_cache_dtype,num_prompts," + "max_concurrency,bs,cmd,performance,metric,unit" +) +DISAGG_HEADER = "Model,xP/yD,ISL,OSL,concurrency" +ESTIMATOR_HEADER = "model_name,model_unique_name,num_layers" + + +def write_csv(path: Path, header: str, *rows: str) -> Path: + """Write a CSV with *header* and *rows* verbatim, and return the path.""" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text("\n".join([header, *rows]) + "\n", encoding="utf-8") + return path + + +def make_result_csv(path: Path, header: str = DUMMY_HEADER, count: int = 1) -> Path: + """A results CSV with *count* measured rows.""" + if header == PRIMUS_HEADER: + rows = [ + f"m{i},{100 + i},tok/s/GPU,train,BF16,1,8,4096,gfx950,16" + for i in range(count) + ] + else: + rows = [f"m{i},{100 + i},tokens_per_second" for i in range(count)] + return write_csv(path, header, *rows) + + +class TestHeaderPredicate: + """A results CSV is recognised by its columns, however they are spelled.""" + + @pytest.mark.parametrize("header", [DUMMY_HEADER, CHAI_HEADER, PRIMUS_HEADER]) + def test_real_headers_are_recognised(self, tmp_path, header): + assert result_csv.has_result_shape(make_result_csv(tmp_path / "r.csv", header)) + + def test_the_three_columns_need_not_be_first_or_adjacent(self, tmp_path): + path = write_csv( + tmp_path / "r.csv", ATOM_HEADER, "text,m," + ",".join([""] * 9) + ",42,tok/s,x" + ) + assert result_csv.has_result_shape(path) + + @pytest.mark.parametrize("header", [DISAGG_HEADER, ESTIMATOR_HEADER]) + def test_near_misses_are_rejected(self, tmp_path, header): + """A capitalised ``Model`` is fine, but a header without all three is not.""" + assert not result_csv.has_result_shape(write_csv(tmp_path / "r.csv", header, "a,b,c")) + + def test_case_and_quoting_do_not_matter(self, tmp_path): + path = write_csv(tmp_path / "r.csv", '"Model", PERFORMANCE ,"metric"', "m,1,t/s") + assert result_csv.has_result_shape(path) + + def test_a_missing_file_is_not_a_results_csv(self, tmp_path): + assert result_csv.missing_columns(tmp_path / "absent.csv") is None + assert not result_csv.has_result_shape(tmp_path / "absent.csv") + + def test_an_empty_file_has_no_header(self, tmp_path): + empty = tmp_path / "r.csv" + empty.write_text("", encoding="utf-8") + assert result_csv.read_columns(empty) is None + + +class TestOwnOutputs: + """madengine's own files satisfy the predicate, so they are never candidates.""" + + @pytest.mark.parametrize( + "name", ["perf.csv", "PERF.CSV", "perf_super.csv", "perf_entry_1.csv"] + ) + def test_own_outputs_are_recognised(self, name): + assert result_csv.is_own_output(name) + + @pytest.mark.parametrize("name", ["results.csv", "perf_of_model.csv", "my_perf.csv"]) + def test_other_files_are_not(self, name): + assert not result_csv.is_own_output(name) + + +class TestMetricRejectionReason: + """What a declared file is asked: can a metric be read out of it?""" + + def test_a_measured_file_is_accepted(self, tmp_path): + assert result_csv.metric_rejection_reason(make_result_csv(tmp_path / "r.csv")) is None + + def test_a_two_column_csv_still_reports(self, tmp_path): + """A declared file is trusted, so the three-column shape is not demanded of it.""" + path = write_csv(tmp_path / "r.csv", "model,performance", "m,42") + assert result_csv.metric_rejection_reason(path) is None + + def test_a_missing_performance_column_names_what_was_found(self, tmp_path): + path = write_csv(tmp_path / "r.csv", "model,throughput", "m,42") + reason = result_csv.metric_rejection_reason(path) + assert "no 'performance' column" in reason + assert "throughput" in reason + + def test_rows_without_a_number_are_no_metric(self, tmp_path): + path = write_csv(tmp_path / "r.csv", DUMMY_HEADER, "m,,tokens_per_second", "m2, ,x") + assert result_csv.metric_rejection_reason(path) == ( + "every row has an empty 'performance' value" + ) + + def test_an_unreadable_file_says_so(self, tmp_path): + assert result_csv.metric_rejection_reason(tmp_path / "absent.csv") == ( + "not readable as CSV" + ) + + def test_counting_measured_rows(self, tmp_path): + path = write_csv(tmp_path / "r.csv", DUMMY_HEADER, "m,1,t/s", "m2,,t/s", "m3,3,t/s") + assert result_csv.count_rows(path) == (2, 3) + + +class TestSuggestions: + """What a run says about the files beside it when it read no metric.""" + + def test_shape_matching_files_are_suggested(self, tmp_path): + make_result_csv(tmp_path / "results.csv") + write_csv(tmp_path / "params.csv", "name,value", "lr,3e-4") + assert result_csv.suggest_candidates([tmp_path]) == [tmp_path / "results.csv"] + + def test_own_outputs_are_not_suggested(self, tmp_path): + make_result_csv(tmp_path / "perf.csv") + make_result_csv(tmp_path / "perf_super.csv") + assert result_csv.suggest_candidates([tmp_path]) == [] + + def test_the_search_is_depth_one(self, tmp_path): + make_result_csv(tmp_path / "nested" / "results.csv") + assert result_csv.suggest_candidates([tmp_path]) == [] + + def test_directories_that_do_not_exist_are_skipped(self, tmp_path): + make_result_csv(tmp_path / "results.csv") + found = result_csv.suggest_candidates([tmp_path / "absent", None, tmp_path]) + assert found == [tmp_path / "results.csv"] + + def test_the_same_file_is_offered_once(self, tmp_path): + make_result_csv(tmp_path / "results.csv") + assert result_csv.suggest_candidates([tmp_path, tmp_path]) == [ + tmp_path / "results.csv" + ] + + def test_lines_name_the_candidates_and_the_field_to_declare(self, tmp_path): + make_result_csv(tmp_path / "results.csv") + lines = result_csv.suggestion_lines([tmp_path]) + assert "multiple_results" in lines[0] + assert str(tmp_path / "results.csv") in lines[1] + + def test_lines_say_where_it_looked_when_nothing_matches(self, tmp_path): + write_csv(tmp_path / "params.csv", "name,value", "lr,3e-4") + lines = result_csv.suggestion_lines([tmp_path]) + assert len(lines) == 1 + assert str(tmp_path) in lines[0] + + def test_a_long_list_is_cut_off(self, tmp_path): + for index in range(5): + make_result_csv(tmp_path / f"r{index}.csv") + lines = result_csv.suggestion_lines([tmp_path], limit=2) + assert lines[-1] == " ... and 3 more"