From 8583dd0e3944198e40050dc619c601bb11d79dc0 Mon Sep 17 00:00:00 2001 From: mkuznet1 Date: Thu, 6 Aug 2026 16:53:17 +0000 Subject: [PATCH 1/8] feat(manifest): validate the manifest schema and settle one source of truth The manifest was checked for three top-level keys and nothing else, so a typo in a transport variable, a nested env value, or a node count that disagreed with itself surfaced as a failed multi-node job minutes later rather than as a startup error. Separately, the deployment target could be read from two places: a top-level slurm block selected the target while the values that took effect were the ones under deployment_config, so a manifest could be configured and ignored at the same time. Declare the shape in src/madengine/schemas/build_manifest.schema.json and validate it where the manifest is loaded, reporting the first violation with its JSON pointer. Unknown keys stay allowed so a manifest can carry consumer metadata. Two cross-field checks a schema cannot express are included: built_models entries must have a matching built_images entry, and slurm.nodes must equal distributed.nnodes. A top-level deployment block is folded into deployment_config with a warning, which leaves one place to read the target from. The schema accepts the nulls madengine itself writes: `madengine build` copies optional model fields straight from models.json, so a model declaring `"timeout": null` reaches the manifest as null rather than as the default the code appears to supply. Optional scalars, the tag list and the env/slurm/distributed blocks therefore allow null, while required fields and value types are unchanged -- a null block is a model that declared nothing, a nested null inside an env map is still an error. The blanket *.json ignore, there to keep model JSONs out of a dev checkout, would have swallowed the schema: runs from a checkout would work while the wheel shipped without it, since hatchling selects wheel contents by VCS status. src/madengine/ schemas/*.json is package data, so it is exempt. --- .gitignore | 4 + pyproject.toml | 1 + src/madengine/deployment/base.py | 7 +- .../orchestration/run_orchestrator.py | 8 +- src/madengine/schemas/__init__.py | 173 +++++++++++++++ .../schemas/build_manifest.schema.json | 146 +++++++++++++ tests/unit/test_manifest_schema.py | 201 ++++++++++++++++++ tests/unit/test_orchestration.py | 3 + 8 files changed, 541 insertions(+), 2 deletions(-) create mode 100644 src/madengine/schemas/__init__.py create mode 100644 src/madengine/schemas/build_manifest.schema.json create mode 100644 tests/unit/test_manifest_schema.py 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/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/deployment/base.py b/src/madengine/deployment/base.py index 69e1367e..9381f729 100644 --- a/src/madengine/deployment/base.py +++ b/src/madengine/deployment/base.py @@ -19,6 +19,8 @@ from jinja2 import Environment, FileSystemLoader from rich.console import Console +from madengine.schemas import validate_build_manifest + # Regex for parsing "performance: " log lines. # Value: optional sign, integer/decimal, scientific notation (e or E). @@ -122,8 +124,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 +154,9 @@ 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]") + return manifest # Template Method - defines workflow diff --git a/src/madengine/orchestration/run_orchestrator.py b/src/madengine/orchestration/run_orchestrator.py index 148c953c..5ad2ee77 100644 --- a/src/madengine/orchestration/run_orchestrator.py +++ b/src/madengine/orchestration/run_orchestrator.py @@ -31,6 +31,7 @@ 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, @@ -221,7 +222,12 @@ 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", {}) # Update additional_context with deployment_config for deployment layer diff --git a/src/madengine/schemas/__init__.py b/src/madengine/schemas/__init__.py new file mode 100644 index 00000000..ae78c6e7 --- /dev/null +++ b/src/madengine/schemas/__init__.py @@ -0,0 +1,173 @@ +#!/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. + +Copyright (c) Advanced Micro Devices, Inc. All rights reserved. +""" + +import json +from pathlib import Path +from typing import Any, Dict, List, Optional + +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) + + +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()) + first = next(iter(sorted(validator.iter_errors(manifest), key=lambda e: list(e.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..efbc08ed --- /dev/null +++ b/src/madengine/schemas/build_manifest.schema.json @@ -0,0 +1,146 @@ +{ + "$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" + }, + "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/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_orchestration.py b/tests/unit/test_orchestration.py index 54a11bd5..b188be10 100644 --- a/tests/unit/test_orchestration.py +++ b/tests/unit/test_orchestration.py @@ -209,6 +209,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 +254,7 @@ def test_skip_run_only_still_calls_execute_local( "deployment_config": {"target": "local"}, "context": {}, "built_images": {}, + "built_models": {}, } ) ) @@ -286,6 +288,7 @@ def test_skip_model_run_calls_execute_local(self, mock_cleanup, tmp_path): "deployment_config": {"target": "local"}, "context": {}, "built_images": {}, + "built_models": {}, } ) ) From 60f25937b9d90b03f30e7a883ceff06fd89bb512 Mon Sep 17 00:00:00 2001 From: mkuznet1 Date: Thu, 6 Aug 2026 16:53:38 +0000 Subject: [PATCH 2/8] feat(config): give the environment and build-dir contract a manifest field A cluster run depends on host variables that say where things live: MODEL_DIR, the cache roots, MAD_DOCKER_BUILDS. The only way to supply them was to source mad.env in the shell that launched madengine, and forgetting produced failures far from the cause -- an empty MODEL_DIR resolves the run script path to nothing, and a MAD_DOCKER_BUILDS off shared storage makes every worker fail to find the image. deployment_config.env_file now names that file and madengine sources it itself, on the submit node while rendering the job and on every worker during the run. Relative paths resolve against the manifest, so a run directory stays movable. A missing file is fatal at startup instead of an empty string mid-run. Only the names of the applied variables are logged; an env file may carry tokens. MAD_DOCKER_BUILDS, until now only visible in the code, is documented alongside it. --- docs/configuration.md | 5 + docs/deployment.md | 43 ++++ src/madengine/core/env_file.py | 150 ++++++++++++++ src/madengine/deployment/base.py | 12 ++ .../orchestration/run_orchestrator.py | 14 +- tests/unit/test_env_file.py | 195 ++++++++++++++++++ 6 files changed, 418 insertions(+), 1 deletion(-) create mode 100644 src/madengine/core/env_file.py create mode 100644 tests/unit/test_env_file.py 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..8e804e7f 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -252,6 +252,49 @@ 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. + ### Multi-Node Training For distributed training across SLURM nodes: diff --git a/src/madengine/core/env_file.py b/src/madengine/core/env_file.py new file mode 100644 index 00000000..22281875 --- /dev/null +++ b/src/madengine/core/env_file.py @@ -0,0 +1,150 @@ +#!/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"}) + + +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 = Path(env_file) + if not path.is_absolute() and base_dir: + path = Path(base_dir) / path + + 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. + + The file wins over the inherited environment, so behaviour matches what the operator + gets by sourcing it before the run. + + 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 + """ + loaded = load_env_file(env_file, base_dir) + os.environ.update(loaded) + return loaded diff --git a/src/madengine/deployment/base.py b/src/madengine/deployment/base.py index 9381f729..2414fb1a 100644 --- a/src/madengine/deployment/base.py +++ b/src/madengine/deployment/base.py @@ -19,6 +19,7 @@ from jinja2 import Environment, FileSystemLoader from rich.console import Console +from madengine.core.env_file import apply_env_file from madengine.schemas import validate_build_manifest @@ -157,6 +158,17 @@ def _load_manifest(self, manifest_file: str) -> Dict: 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 diff --git a/src/madengine/orchestration/run_orchestrator.py b/src/madengine/orchestration/run_orchestrator.py index 5ad2ee77..ad894c69 100644 --- a/src/madengine/orchestration/run_orchestrator.py +++ b/src/madengine/orchestration/run_orchestrator.py @@ -25,6 +25,7 @@ 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, @@ -229,7 +230,18 @@ def execute( 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 = {} diff --git a/tests/unit/test_env_file.py b/tests/unit/test_env_file.py new file mode 100644 index 00000000..8c4921a4 --- /dev/null +++ b/tests/unit/test_env_file.py @@ -0,0 +1,195 @@ +#!/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.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) + + +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" From 89d11ff9415b5eac21135601a7c1a0c642b26b34 Mon Sep 17 00:00:00 2001 From: mkuznet1 Date: Thu, 6 Aug 2026 16:53:39 +0000 Subject: [PATCH 3/8] refactor(reporting): make the result schema a contract, not a column list Three writers create perf.csv -- the container runner, the deployment aggregation path and the Kubernetes results mixin -- and each carried its own copy of the 29-column header under a comment asking the reader to keep them in sync. Nothing enforced that, and a consumer reading the CSV had no declaration to read at all. perf_csv.schema.json now declares the row: names, order, types and a description per column, with the writers reading the header from it. Columns that restate a manifest field carry a JSON pointer to it, so the result row and the build manifest are two views of one shape; a test walks those pointers and fails if a manifest field is renamed out from under the result contract. --- src/madengine/deployment/base.py | 11 +- src/madengine/deployment/k8s_results.py | 12 +- src/madengine/reporting/update_perf_csv.py | 13 +- src/madengine/schemas/__init__.py | 50 ++++- src/madengine/schemas/perf_csv.schema.json | 140 ++++++++++++++ tests/unit/test_perf_csv_schema.py | 204 +++++++++++++++++++++ 6 files changed, 402 insertions(+), 28 deletions(-) create mode 100644 src/madengine/schemas/perf_csv.schema.json create mode 100644 tests/unit/test_perf_csv_schema.py diff --git a/src/madengine/deployment/base.py b/src/madengine/deployment/base.py index 2414fb1a..be72ddb2 100644 --- a/src/madengine/deployment/base.py +++ b/src/madengine/deployment/base.py @@ -20,7 +20,7 @@ from rich.console import Console from madengine.core.env_file import apply_env_file -from madengine.schemas import validate_build_manifest +from madengine.schemas import perf_csv_header, validate_build_manifest # Regex for parsing "performance: " log lines. @@ -607,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/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 index ae78c6e7..45c787da 100644 --- a/src/madengine/schemas/__init__.py +++ b/src/madengine/schemas/__init__.py @@ -13,12 +13,19 @@ 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 +from typing import Any, Dict, List, Optional, Tuple from madengine.core.errors import ErrorContext, ValidationError @@ -34,6 +41,47 @@ def load_schema(name: str = "build_manifest.schema.json") -> Dict[str, Any]: 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 "/" 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_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"}) == [] From fe646238d92c789aebe84923cc677ac3a73dedbd Mon Sep 17 00:00:00 2001 From: mkuznet1 Date: Thu, 6 Aug 2026 16:54:49 +0000 Subject: [PATCH 4/8] fix(slurm): report multinode failures honestly A two-node run finished with the scheduler recording exit code 3 and madengine printing "All model executions completed successfully!". Three defects lined up to make that possible: per-node exit codes were collected nowhere, a metric parsed from a log was treated as proof the whole run succeeded, and a deployment the scheduler called failed produced an empty failure list when its results could not be parsed. Each node now records its exit code, host and rank in a marker the submit node reads back, and result collection consults those markers and the scheduler's own verdict before it believes any number. The task script runs under `set -e`, so a non-zero madengine used to end it on that very line, leaving the failing node with no marker and no artifacts -- exactly the evidence a failed node has to leave behind; errexit is now off across that one call and restored once the exit code is captured. What a non-zero node means is decided the way the single-node path already states it in resolve_run_status: a metric is the strongest evidence a run did the work. Primus/Megatron reports throughput from the last global rank only, so every other node finds no metric locally and exits non-zero even when the training was perfect -- both accepted baseline runs on the reference cluster ended with node 0 at exit 3 and node 1 at 0. With a metric, the node outcomes and the scheduler's verdict are warnings and the numbers stand: a row with a number in it is a measurement whatever happened around it, so the rows stay in successful_runs and the verdict travels separately, naming what did not finish. With no metric anywhere the node evidence is all there is and it decides: a node that exited non-zero or never reported gives a failure naming it, a clean set of exit codes gives NO_METRIC and exit 5, because a broken result contract is not the same event as a crash, and neither is success. The summary is built in _summarise_deployment so the rule can be tested without standing up a deployment. For the same reason srun --kill-on-bad-exit is opt-in through slurm.kill_on_bad_exit rather than the default: the rank exiting non-zero may be the peer of the one holding the results, and killing the step there would throw them away. The multiple_results diagnostics now name the paths searched and the variable the container was given, so a model script that does not hold up its end of that contract can be fixed without reading madengine's source. --- docs/cli-reference.md | 15 + docs/deployment.md | 25 ++ src/madengine/cli/commands/run.py | 29 +- src/madengine/cli/constants.py | 3 + src/madengine/deployment/slurm.py | 219 +++++++++++- .../deployment/templates/slurm/job.sh.j2 | 23 +- src/madengine/execution/container_runner.py | 29 +- .../orchestration/run_orchestrator.py | 108 +++++- .../schemas/build_manifest.schema.json | 4 + tests/unit/test_multinode_failure.py | 316 ++++++++++++++++++ 10 files changed, 748 insertions(+), 23 deletions(-) create mode 100644 tests/unit/test_multinode_failure.py 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/deployment.md b/docs/deployment.md index 8e804e7f..bdccfcb8 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -295,6 +295,31 @@ Leaving it unset is supported but leaves image distribution to the operator: wor 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/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/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..0c2d597b 100644 --- a/src/madengine/execution/container_runner.py +++ b/src/madengine/execution/container_runner.py @@ -1706,9 +1706,24 @@ 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]" ) run_results["performance"] = None else: @@ -1724,7 +1739,10 @@ def run_container( # Check if 'performance' column exists if 'performance' not in csv_reader.fieldnames: - print("Error: 'performance' column not found in multiple results file.") + print( + f"Error: {resolved_path} has no 'performance' column; " + f"found: {', '.join(csv_reader.fieldnames or []) or '(no header)'}" + ) run_results["performance"] = None else: # Check if at least one row has a non-empty performance value @@ -1736,7 +1754,10 @@ def run_container( if not has_valid_perf: run_results["performance"] = None - print("Error: Performance metric is empty in all rows of multiple results file.") + print( + f"Error: every row of {resolved_path} has an empty " + f"'performance' value, so the run produced no metric." + ) except Exception as e: self.rich_console.print( f"[yellow]Warning: Could not validate multiple results file: {e}[/yellow]" diff --git a/src/madengine/orchestration/run_orchestrator.py b/src/madengine/orchestration/run_orchestrator.py index ad894c69..2b5f3724 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, Optional from rich.console import Console as RichConsole from rich.panel import Panel @@ -767,15 +767,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/schemas/build_manifest.schema.json b/src/madengine/schemas/build_manifest.schema.json index efbc08ed..9f1edfda 100644 --- a/src/madengine/schemas/build_manifest.schema.json +++ b/src/madengine/schemas/build_manifest.schema.json @@ -123,6 +123,10 @@ "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" }, 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 From 0c343877656e0137e81aae45aedc4714b45d1ea6 Mon Sep 17 00:00:00 2001 From: mkuznet1 Date: Fri, 7 Aug 2026 16:41:04 +0000 Subject: [PATCH 5/8] fix(reporting): name the reason a run produced no metric A model card declares where its results CSV lives in multiple_results, and the value has to agree with a path a script in another repository builds by hand. When the two disagreed madengine fell back to scraping 'performance: ' from the log, found nothing there because a CSV-reporting model never prints that line, and wrote a row with an empty performance and status=FAILURE. One yellow warning was the only trace, and a typo in the card looked exactly like a card that never declared the field. The declaration stays the contract. What changes is what a run says when it cannot read a metric: madengine.reporting.result_csv recognises a results CSV by a header carrying model, performance and metric, in any order and with any number of extra columns, and that predicate is used only to explain itself. The message now names the CSVs lying beside the run that look like results files and the field to declare them in, so a typo and an omission read differently while both still fail. madengine's own perf.csv and the perf_super/perf_entry family are never named, since they match by construction. Nothing is substituted. A file nobody declared is not read for a verdict, so no run silently reports a number from a file its card never mentioned. The same module also replaces the ad-hoc CSV parsing that validated a declared file inline, so 'no performance column' and 'every row is empty' are worded in one place instead of two. Deliberately not here: searching for and reporting from an undeclared file. Two MAD cards write a conforming CSV without declaring it, and one line in each card fixes that where the omission is. The other 32 such cards write to /run_logs on shared storage, which no depth-1 search would have reached anyway. --- src/madengine/execution/container_runner.py | 67 ++++--- src/madengine/reporting/result_csv.py | 151 ++++++++++++++++ tests/unit/test_result_csv.py | 186 ++++++++++++++++++++ 3 files changed, 369 insertions(+), 35 deletions(-) create mode 100644 src/madengine/reporting/result_csv.py create mode 100644 tests/unit/test_result_csv.py diff --git a/src/madengine/execution/container_runner.py b/src/madengine/execution/container_runner.py index 0c2d597b..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 @@ -1725,42 +1726,23 @@ def run_container( 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( - f"Error: {resolved_path} has no 'performance' column; " - f"found: {', '.join(csv_reader.fieldnames or []) or '(no header)'}" - ) - 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( - f"Error: every row of {resolved_path} has an empty " - f"'performance' value, so the run produced no metric." - ) - 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: @@ -1803,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/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/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" From 1f2c78b4a1cb0adfaa5bf86d2f1bba59b6b93019 Mon Sep 17 00:00:00 2001 From: mkuznet1 Date: Mon, 10 Aug 2026 11:21:22 +0000 Subject: [PATCH 6/8] fix(schemas): order validation errors by pointer, not by a mixed-type path The first schema error reported to the caller was picked by sorting jsonschema's errors on list(e.path). Such a path mixes property names with array indices, and comparing two of them elementwise raises TypeError the moment they diverge into a string against an integer at the same position, replacing the manifest error the caller asked about with a traceback from the sort. No manifest reaches that today: the type of a path element follows the type of the container it indexes, and one container is not both an array and an object. The ordering does not need the hazard either way -- the error's JSON pointer is a string and orders deterministically. --- src/madengine/schemas/__init__.py | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/src/madengine/schemas/__init__.py b/src/madengine/schemas/__init__.py index 45c787da..f282af73 100644 --- a/src/madengine/schemas/__init__.py +++ b/src/madengine/schemas/__init__.py @@ -193,7 +193,13 @@ def validate_build_manifest( where = f" in {source}" if source else "" validator = jsonschema.Draft202012Validator(load_schema()) - first = next(iter(sorted(validator.iter_errors(manifest), key=lambda e: list(e.path))), None) + # 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}", From 34a6a14a8833139764da6fd615ef2ce10ca96bee Mon Sep 17 00:00:00 2001 From: mkuznet1 Date: Mon, 10 Aug 2026 11:21:23 +0000 Subject: [PATCH 7/8] fix(config): source an env file once per process A cluster run reaches apply_env_file twice for one manifest: the orchestrator applies it before it knows the target, and the deployment layer applies it again when it loads the same manifest on the submit node. An env file is shell rather than a list of assignments, so the second pass is not a no-op -- PATH="$PATH:/opt/x" appends a second copy, and a $(...) runs again with whatever that costs. Applied files are now remembered by resolved path, so a second call returns what the first one applied without running bash again, and the two call sites keep logging the same names. A file that is not there still fails every time, because nothing was applied to remember. --- src/madengine/core/env_file.py | 29 ++++++++++++++---- tests/unit/test_env_file.py | 54 ++++++++++++++++++++++++++++++++++ 2 files changed, 78 insertions(+), 5 deletions(-) diff --git a/src/madengine/core/env_file.py b/src/madengine/core/env_file.py index 22281875..2deb266a 100644 --- a/src/madengine/core/env_file.py +++ b/src/madengine/core/env_file.py @@ -37,6 +37,20 @@ #: 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]: """ @@ -54,9 +68,7 @@ def load_env_file(env_file: str, base_dir: Optional[str] = None) -> Dict[str, st Raises: ValidationError: the file is missing, or bash failed to source it """ - path = Path(env_file) - if not path.is_absolute() and base_dir: - path = Path(base_dir) / path + path = resolve_env_file(env_file, base_dir) context = ErrorContext( operation="env_file loading", component="core.env_file", file_path=str(path) @@ -133,10 +145,12 @@ def _parse_env_dump(dump: str) -> Dict[str, str]: 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. + 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. + 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` @@ -145,6 +159,11 @@ def apply_env_file(env_file: str, base_dir: Optional[str] = None) -> Dict[str, s 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/tests/unit/test_env_file.py b/tests/unit/test_env_file.py index 8c4921a4..fcdf2231 100644 --- a/tests/unit/test_env_file.py +++ b/tests/unit/test_env_file.py @@ -14,6 +14,7 @@ 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 @@ -33,6 +34,12 @@ def restore_environ(): 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.""" @@ -193,3 +200,50 @@ def test_unrelated_variables_are_left_alone(self, env_dir, monkeypatch): 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")) From 1e2c005c77536c73cf9204ab89b842ce33c43a8d Mon Sep 17 00:00:00 2001 From: mkuznet1 Date: Mon, 10 Aug 2026 11:21:24 +0000 Subject: [PATCH 8/8] fix(run): say when --additional-context replaces a manifest block A manifest carrying a top-level deployment block is now folded into deployment_config with a warning that it "belongs there and was moved". The merge that follows does not keep that promise when the same block also arrives on the command line: --additional-context wins, and it wins whole blocks, so a field only the manifest declares -- a qos, say -- is dropped rather than merged. Nothing picks it up later either, because the deployment layer reads its configuration out of additional_context and never out of the manifest. The precedence stays as it is; what was missing is that anyone was told. The merge moves into merge_deployment_config, which returns one warning per replaced block and names the fields that did not survive it. --- .../orchestration/run_orchestrator.py | 66 +++++++++++++++++-- tests/unit/test_orchestration.py | 46 ++++++++++++- 2 files changed, 106 insertions(+), 6 deletions(-) diff --git a/src/madengine/orchestration/run_orchestrator.py b/src/madengine/orchestration/run_orchestrator.py index 2b5f3724..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 Any, Dict, Optional +from typing import Any, Dict, List, Optional from rich.console import Console as RichConsole from rich.panel import Panel @@ -39,6 +39,60 @@ 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: """ @@ -247,10 +301,12 @@ def execute( 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", {}) diff --git a/tests/unit/test_orchestration.py b/tests/unit/test_orchestration.py index b188be10..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 @@ -362,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 == {}