From 0c3661b11dac4db795cc4e8a61bcbc39098a105d Mon Sep 17 00:00:00 2001 From: Ivan Podkidyshev Date: Fri, 11 Sep 2026 20:53:25 +0200 Subject: [PATCH 01/17] Add unified results for completed Slurm scenarios --- doc/reporting.rst | 15 ++ src/cloudai/metrics.py | 4 +- src/cloudai/systems/slurm/slurm_runner.py | 31 ++- src/cloudai/unified_output.py | 242 ++++++++++++++++++++++ tests/test_unified_output.py | 179 ++++++++++++++++ 5 files changed, 469 insertions(+), 2 deletions(-) create mode 100644 src/cloudai/unified_output.py create mode 100644 tests/test_unified_output.py diff --git a/doc/reporting.rst b/doc/reporting.rst index 1bdeb8d92..99b4bc707 100644 --- a/doc/reporting.rst +++ b/doc/reporting.rst @@ -31,6 +31,21 @@ Per-test reports are linked to a particular workload type (e.g. ``NcclTest``). A To list all available reports, users can use ``cloudai list-reports``. Use verbose output to also print report configurations. +Unified experiment output +------------------------- + +Ordinary Slurm scenarios write ``experiment.json`` in the scenario results directory when execution finishes or fails. +The file contains experiment metadata, test cases, submitted runs, statuses, timing, and canonical metrics from +``TestDefinition.metric_observations()``. NCCL and NIXLBench provide these metrics; other workloads have empty metric lists. +Test-level metrics are arithmetic means of successful iterations at matching metric and dimension points. Per-run +measurements retain their original values, and missing measurements are not treated as zero. + +Experiment timing covers scenario execution, including gaps between jobs. Run timestamps without timezone information +are null. Metric extraction and output-write failures produce warnings without changing execution behavior. The file is +replaced atomically and is independent of reporter configuration. Dry runs, DSE, and single-sbatch execution do not produce +this artifact. + + .. _general-flow: General Flow diff --git a/src/cloudai/metrics.py b/src/cloudai/metrics.py index abf4c066a..f2cee83f1 100644 --- a/src/cloudai/metrics.py +++ b/src/cloudai/metrics.py @@ -44,6 +44,8 @@ class DimensionDefinition: key: str label: str value_type: Any + unit: str = "" + is_x: bool = False def validate(self, value: Any) -> MetricValue: """Validate one configured or observed dimension value.""" @@ -97,7 +99,7 @@ def validate_dimensions(cls, values: Mapping[str, Any]) -> MetricDimensions: return {key: cls.get_dimension(key).validate(value) for key, value in values.items()} -SIZE_BYTES = DimensionDefinition("size_bytes", "Size", Annotated[int, Field(strict=True, ge=0)]) +SIZE_BYTES = DimensionDefinition("size_bytes", "Size", Annotated[int, Field(strict=True, ge=0)], unit="B", is_x=True) BATCH_SIZE = DimensionDefinition("batch_size", "Batch size", Annotated[int, Field(strict=True, gt=0)]) OPERATION = DimensionDefinition("operation", "Operation", Annotated[str, Field(strict=True, min_length=1)]) PLACEMENT = DimensionDefinition("placement", "Placement", Literal["in_place", "out_of_place"]) diff --git a/src/cloudai/systems/slurm/slurm_runner.py b/src/cloudai/systems/slurm/slurm_runner.py index 7cbcb0c54..f0db71c64 100644 --- a/src/cloudai/systems/slurm/slurm_runner.py +++ b/src/cloudai/systems/slurm/slurm_runner.py @@ -21,7 +21,8 @@ import toml -from cloudai.core import BaseJob, BaseRunner, JobIdRetrievalError, System, TestRun, TestScenario +from cloudai.core import BaseJob, BaseRunner, JobIdRetrievalError, JobStatusResult, System, TestRun, TestScenario +from cloudai.unified_output import ExperimentOutput from cloudai.util import CommandShell from .slurm_command_gen_strategy import SlurmCommandGenStrategy @@ -43,6 +44,34 @@ def __init__(self, mode: str, system: System, test_scenario: TestScenario, outpu self.system = cast(SlurmSystem, system) self.cmd_shell = CommandShell() self.pinned_nodes: dict[str, list[str]] = {} + self._experiment: ExperimentOutput | None = None + + def run(self) -> None: + if self.mode != "run" or any(tr.is_dse_job or tr.step > 0 for tr in self.test_scenario.test_runs): + super().run() + return + + try: + self._experiment = ExperimentOutput(self.test_scenario, self.scenario_root) + except Exception as exc: + logging.warning("Cannot initialize unified experiment output: %s", exc) + completed = False + try: + super().run() + completed = True + finally: + if self._experiment is not None: + self._experiment.finish(self.system, self.jobs, completed) + self._experiment = None + + def get_job_status(self, job: BaseJob) -> JobStatusResult: + result = super().get_job_status(job) + if self._experiment is not None: + try: + self._experiment.capture(self.system, job, result) + except Exception as exc: + logging.warning("Cannot capture unified output for job %s: %s", job.id, exc) + return result def submit_test(self, tr: TestRun) -> None: if tr.pin_nodes and tr.name in self.pinned_nodes: diff --git a/src/cloudai/unified_output.py b/src/cloudai/unified_output.py new file mode 100644 index 000000000..3e467e057 --- /dev/null +++ b/src/cloudai/unified_output.py @@ -0,0 +1,242 @@ +# SPDX-FileCopyrightText: NVIDIA CORPORATION & AFFILIATES +# Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import logging +from collections import defaultdict +from dataclasses import replace +from datetime import datetime, timezone +from pathlib import Path +from statistics import fmean +from tempfile import NamedTemporaryFile +from typing import Literal + +import toml +from pydantic import BaseModel, Field, FiniteFloat + +from cloudai.core import BaseJob, JobStatusResult, System, TestScenario +from cloudai.metrics import MetricCatalog, MetricObservation, MetricValue + +Status = Literal["pending", "running", "completed", "failed", "cancelled", "unknown"] + + +class Dimension(BaseModel): + """A metric coordinate in the unified output contract.""" + + name: str + value: str + unit: str = "" + is_x: bool = False + + +class Metric(BaseModel): + """A canonical measurement and its dimension point.""" + + name: str + value: FiniteFloat + unit: str + dimensions: list[Dimension] + + +class Run(BaseModel): + """One submitted execution, independent of the mutable TestRun.""" + + path: str + jobid: str + status: Status = "unknown" + metrics: list[Metric] = Field(default_factory=list) + start: datetime | None = None + finish: datetime | None = None + duration: FiniteFloat | None = None + iteration: int + step: int + + +class TestResult(BaseModel): + """A scenario test case and all its executed iterations.""" + + id: str + name: str + description: str + status: Status = "pending" + path: str + metrics: list[Metric] = Field(default_factory=list) + runs: list[Run] = Field(default_factory=list) + + +class Experiment(BaseModel): + """The API Schema v0.2 Experiment representation.""" + + id: str + name: str + status: Status = "unknown" + path: str + start: datetime | None + finish: datetime | None = None + duration: FiniteFloat | None = None + tests: list[TestResult] + + +def _timestamp(value: str) -> datetime | None: + try: + timestamp = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError: + return None + # sacct's default timestamps carry no timezone; the login host may use a different one. + return timestamp.astimezone(timezone.utc) if timestamp.tzinfo is not None else None + + +def _metric(observation: MetricObservation) -> Metric: + dimensions = [] + for key, value in sorted(observation.dimensions.items()): + definition = MetricCatalog.get_dimension(key) + dimensions.append( + Dimension(name=definition.label, value=str(value), unit=definition.unit, is_x=definition.is_x) + ) + return Metric( + name=observation.metric.display_name, + value=observation.value, + unit=observation.metric.unit, + dimensions=dimensions, + ) + + +def _status(statuses: set[Status]) -> Status: + if "failed" in statuses: + return "failed" + if "cancelled" in statuses: + return "cancelled" + if statuses == {"pending"}: + return "pending" + return "completed" if statuses == {"completed"} else "unknown" + + +class ExperimentOutput: + """Collect completed Slurm runs and atomically write one experiment.json.""" + + def __init__(self, scenario: TestScenario, output_path: Path): + self.scenario = scenario + self.output_path = output_path.absolute() + self.tests = { + tr.name: TestResult( + id=tr.name, + name=tr.name, + description=tr.test.description, + path=str(self.output_path / tr.name), + ) + for tr in scenario.test_runs + } + self.observations: dict[str, list[MetricObservation]] = defaultdict(list) + self.experiment = Experiment( + id=self.output_path.name, + name=scenario.name, + path=str(self.output_path), + start=datetime.now(timezone.utc), + tests=list(self.tests.values()), + ) + + def capture(self, system: System, job: BaseJob, result: JobStatusResult | None = None) -> None: + tr = job.test_run + test = self.tests[tr.name] + if any(run.jobid == str(job.id) and run.path == str(tr.output_path.absolute()) for run in test.runs): + return + run = Run( + path=str(tr.output_path.absolute()), + jobid=str(job.id), + iteration=tr.current_iteration, + step=tr.step, + status="failed" if result is not None and not result.is_successful else "unknown", + ) + test.runs.append(run) + metadata_path = tr.output_path / "slurm-job.toml" + try: + metadata = toml.load(metadata_path) + state = metadata["state"].split()[0].rstrip("+") + if state == "CANCELLED": + run.status = "cancelled" + elif state in { + "FAILED", + "TIMEOUT", + "NODE_FAIL", + "OUT_OF_MEMORY", + "BOOT_FAIL", + "DEADLINE", + "PREEMPTED", + "REVOKED", + "SPECIAL_EXIT", + } or metadata["exit_code"] not in {"0", "0:0", ""}: + run.status = "failed" + elif state == "COMPLETED" and result is not None and result.is_successful: + run.status = "completed" + run.start = _timestamp(metadata["start_time"]) + run.finish = _timestamp(metadata["end_time"]) + run.duration = metadata["elapsed_time_sec"] + except (OSError, ValueError, KeyError, TypeError, IndexError) as exc: + if result is not None or metadata_path.exists(): + logging.warning("Cannot read unified output metadata for job %s: %s", job.id, exc) + + if result is None: + return + try: + observations = sorted( + tr.test.metric_observations(system, tr), + key=lambda observation: (observation.metric.key, sorted(observation.dimensions.items())), + ) + run.metrics = [_metric(observation) for observation in observations] + if run.status == "completed": + self.observations[tr.name].extend(observations) + except Exception as exc: + logging.warning("Cannot extract unified output metrics for job %s: %s", job.id, exc) + + def finish(self, system: System, unfinished_jobs: list[BaseJob], completed: bool) -> None: + temporary_path = None + try: + for job in unfinished_jobs: + self.capture(system, job) + for tr in self.scenario.test_runs: + test = self.tests[tr.name] + statuses: set[Status] = {run.status for run in test.runs} + if len(test.runs) != tr.iterations: + statuses.add("unknown" if test.runs else "pending") + test.status = _status(statuses) + + groups: dict[tuple[str, tuple[tuple[str, MetricValue], ...]], list[MetricObservation]] = defaultdict( + list + ) + for observation in self.observations[tr.name]: + groups[(observation.metric.key, tuple(sorted(observation.dimensions.items())))].append(observation) + test.metrics = [ + _metric(replace(group[0], value=fmean(observation.value for observation in group))) + for _, group in sorted(groups.items()) + ] + + self.experiment.status = _status({test.status for test in self.tests.values()}) if completed else "failed" + self.experiment.finish = datetime.now(timezone.utc) + if self.experiment.start is not None: + self.experiment.duration = (self.experiment.finish - self.experiment.start).total_seconds() + content = self.experiment.model_dump_json(indent=2) + self.output_path.mkdir(parents=True, exist_ok=True) + with NamedTemporaryFile(mode="w", encoding="utf-8", dir=self.output_path, delete=False) as temporary: + temporary_path = Path(temporary.name) + temporary.write(content + "\n") + temporary_path.replace(self.output_path / "experiment.json") + except Exception as exc: + logging.warning("Cannot write unified experiment output: %s", exc) + finally: + if temporary_path is not None: + try: + temporary_path.unlink(missing_ok=True) + except OSError as exc: + logging.warning("Cannot remove temporary unified output %s: %s", temporary_path, exc) diff --git a/tests/test_unified_output.py b/tests/test_unified_output.py new file mode 100644 index 000000000..7392ef794 --- /dev/null +++ b/tests/test_unified_output.py @@ -0,0 +1,179 @@ +# SPDX-FileCopyrightText: NVIDIA CORPORATION & AFFILIATES +# Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import json +from pathlib import Path +from unittest.mock import Mock + +import pytest +import toml + +from cloudai._core.exceptions import JobFailureError +from cloudai.core import BaseJob, TestRun, TestScenario +from cloudai.systems.slurm import SlurmJob, SlurmRunner, SlurmSystem +from cloudai.unified_output import Experiment, ExperimentOutput +from cloudai.workloads.nccl_test import NCCLTestDefinition + + +@pytest.fixture +def runner(slurm_system: SlurmSystem, nccl_tr: TestRun, monkeypatch: pytest.MonkeyPatch) -> SlurmRunner: + stdout = (nccl_tr.output_path / "stdout.txt").read_text() + "# Out of bounds values : 0\n" + nccl_tr.iterations = 2 + slurm_system.output_path.mkdir() + runner = SlurmRunner( + "run", slurm_system, TestScenario("repeats", [nccl_tr]), slurm_system.output_path / "repeats_2026-09-11" + ) + + def submit(tr: TestRun) -> SlurmJob: + content = stdout + if tr.current_iteration: + content = "\n".join(line for line in stdout.splitlines() if not line.lstrip().startswith("12000000")) + content = content.replace("20.20", "40.20") + (tr.output_path / "stdout.txt").write_text(content) + return SlurmJob(tr, id=100 + tr.current_iteration) + + def complete(job: BaseJob) -> None: + offset = "+02:00" if job.test_run.current_iteration == 0 else "" + metadata = { + "state": "COMPLETED", + "exit_code": "0:0", + "start_time": f"2026-09-11T12:00:00{offset}", + "end_time": f"2026-09-11T12:00:02{offset}", + "elapsed_time_sec": 2, + } + (job.test_run.output_path / "slurm-job.toml").write_text(toml.dumps(metadata)) + + monkeypatch.setattr(runner, "on_job_submit", Mock()) + monkeypatch.setattr(runner, "_submit_test", submit) + monkeypatch.setattr(runner, "on_job_completion", complete) + monkeypatch.setattr(SlurmSystem, "is_job_completed", lambda self, job: True) + monkeypatch.setattr(SlurmSystem, "is_job_running", lambda self, job: False) + monkeypatch.setattr(SlurmSystem, "kill", Mock()) + return runner + + +def test_completed_experiment_and_repeated_metrics(runner: SlurmRunner): + runner.run() + + path = runner.scenario_root / "experiment.json" + experiment = Experiment.model_validate_json(path.read_text()) + data = json.loads(path.read_text()) + assert set(data) == {"id", "name", "status", "path", "start", "finish", "duration", "tests"} + assert experiment.id == runner.scenario_root.name + assert experiment.name == "repeats" + assert experiment.status == "completed" + assert experiment.start is not None and experiment.finish is not None + assert experiment.duration == (experiment.finish - experiment.start).total_seconds() + test = experiment.tests[0] + assert test.id == test.name == "nccl_test" + assert test.status == "completed" + assert test.path == str(runner.scenario_root / "nccl_test") + assert [run.iteration for run in test.runs] == [0, 1] + assert [run.step for run in test.runs] == [0, 0] + assert [run.jobid for run in test.runs] == ["100", "101"] + assert [run.path for run in test.runs] == [str(Path(test.path) / str(i)) for i in range(2)] + assert data["tests"][0]["runs"][0]["start"] == "2026-09-11T10:00:00Z" + assert test.runs[1].start is None and test.runs[1].finish is None + assert [run.duration for run in test.runs] == [2, 2] + + bandwidth = [ + metric + for metric in test.metrics + if metric.name == "Bandwidth" + and any(dimension.name == "Placement" and dimension.value == "out_of_place" for dimension in metric.dimensions) + ] + assert [metric.value for metric in bandwidth] == pytest.approx([30.20, 30.30, 130.40]) + assert all(metric.unit == "GB/s" for metric in bandwidth) + sizes = [dimension for metric in bandwidth for dimension in metric.dimensions if dimension.name == "Size"] + assert [size.value for size in sizes] == ["1000000", "2000000", "12000000"] + assert all(size.unit == "B" and size.is_x for size in sizes) + assert len(test.runs[0].metrics) == 12 + assert len(test.runs[1].metrics) == 8 + assert test.runs[0].metrics != test.runs[1].metrics + assert list(runner.scenario_root.glob("*.json")) == [path] + + +@pytest.mark.parametrize("failure,abort", [("workload", True), ("workload", False), ("scheduler", False)]) +def test_failed_run_preserves_earlier_results( + runner: SlurmRunner, monkeypatch: pytest.MonkeyPatch, failure: str, abort: bool +): + complete = runner.on_job_completion + runner.test_scenario.job_status_check = abort + + def fail_second_job(job: BaseJob) -> None: + complete(job) + if job.test_run.current_iteration != 1: + return + if failure == "workload": + with (job.test_run.output_path / "stdout.txt").open("a") as stdout: + stdout.write("\nTest NCCL failure\n") + else: + path = job.test_run.output_path / "slurm-job.toml" + path.write_text(path.read_text().replace("COMPLETED", "FAILED")) + + monkeypatch.setattr(runner, "on_job_completion", fail_second_job) + if abort: + with pytest.raises(JobFailureError): + runner.run() + else: + runner.run() + + experiment = Experiment.model_validate_json((runner.scenario_root / "experiment.json").read_text()) + test = experiment.tests[0] + assert experiment.status == test.status == "failed" + assert [run.status for run in test.runs] == ["completed", "failed"] + assert [run.iteration for run in test.runs] == [0, 1] + assert test.runs[0].path != test.runs[1].path + assert len(test.runs[1].metrics) == 8 + assert test.metrics == test.runs[0].metrics + + +def test_output_failures_are_nonfatal( + runner: SlurmRunner, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +): + extract = NCCLTestDefinition.metric_observations + + def failing_extract(self, system, tr): + if tr.current_iteration == 1: + raise ValueError("malformed measurement") + return extract(self, system, tr) + + monkeypatch.setattr(NCCLTestDefinition, "metric_observations", failing_extract) + runner.run() + + path = runner.scenario_root / "experiment.json" + original = path.read_bytes() + experiment = Experiment.model_validate_json(original) + assert experiment.status == "completed" + assert experiment.tests[0].runs[1].metrics == [] + assert experiment.tests[0].metrics == experiment.tests[0].runs[0].metrics + assert "Cannot extract unified output metrics for job 101: malformed measurement" in caplog.text + + monkeypatch.setattr(Path, "replace", Mock(side_effect=OSError("write unavailable"))) + output = ExperimentOutput(runner.test_scenario, runner.scenario_root) + output.finish(runner.system, [], completed=False) + assert path.read_bytes() == original + assert "Cannot write unified experiment output: write unavailable" in caplog.text + assert sorted(p.name for p in runner.scenario_root.iterdir()) == ["experiment.json", "nccl_test"] + + monkeypatch.setattr( + "cloudai.systems.slurm.slurm_runner.ExperimentOutput", Mock(side_effect=OSError("initialization unavailable")) + ) + runner.test_scenario.test_runs[0].current_iteration = 0 + runner.run() + assert len(runner.jobs) == 0 + assert path.read_bytes() == original + assert "Cannot initialize unified experiment output: initialization unavailable" in caplog.text From f655642e3ffba421d3e0751d5c88fc3288ed73bf Mon Sep 17 00:00:00 2001 From: Ivan Podkidyshev Date: Mon, 14 Sep 2026 17:26:33 +0200 Subject: [PATCH 02/17] Emit UTC timestamps from Slurm CLI accounting Request UTC ISO timestamps from sacct so the existing unified-output parser can retain job start and finish across supported Python versions. --- src/cloudai/systems/slurm/slurm_system.py | 1 + tests/systems/slurm/test_system.py | 31 +++++++++++++++++++++-- 2 files changed, 30 insertions(+), 2 deletions(-) diff --git a/src/cloudai/systems/slurm/slurm_system.py b/src/cloudai/systems/slurm/slurm_system.py index edef2989b..7d39c6d2f 100644 --- a/src/cloudai/systems/slurm/slurm_system.py +++ b/src/cloudai/systems/slurm/slurm_system.py @@ -362,6 +362,7 @@ def is_job_completed(self, job: BaseJob, retry_threshold: int = 3) -> bool: def get_job_status(self, job: BaseJob, retry_threshold: int = 3) -> list[SlurmStepMetadata]: retry_count = 0 command = ( + "TZ=UTC SLURM_TIME_FORMAT='%Y-%m-%dT%H:%M:%SZ' " f"sacct -j {job.id} --format=JobID,JobName,State,ExitCode,Start,End,ElapsedRAW,SubmitLine " "--delimiter='|' -p --noheader" ) diff --git a/tests/systems/slurm/test_system.py b/tests/systems/slurm/test_system.py index 0f1b69ebf..17515b014 100644 --- a/tests/systems/slurm/test_system.py +++ b/tests/systems/slurm/test_system.py @@ -673,8 +673,30 @@ def test_env_vars_order(): assert actual_order == expected_order -@pytest.mark.parametrize("stdout,stderr, expected", [("", "error", None)]) -def test_get_job_status(slurm_system: SlurmSystem, stdout: str, stderr: str, expected: tuple): +@pytest.mark.parametrize( + "stdout,stderr,expected", + [ + ("", "error", None), + ( + "1|job|COMPLETED|0:0|2026-09-14T14:41:02Z|2026-09-14T14:42:41Z|99|sbatch job.sh|\n", + "", + [ + SlurmStepMetadata( + job_id=1, + step_id="", + name="job", + state="COMPLETED", + exit_code="0:0", + start_time="2026-09-14T14:41:02Z", + end_time="2026-09-14T14:42:41Z", + elapsed_time_sec=99, + submit_line="sbatch job.sh", + ) + ], + ), + ], +) +def test_get_job_status(slurm_system: SlurmSystem, stdout: str, stderr: str, expected: list[SlurmStepMetadata] | None): job = BaseJob(test_run=Mock(), id=1) pp = Mock() pp.communicate = Mock(return_value=(stdout, stderr)) @@ -685,6 +707,11 @@ def test_get_job_status(slurm_system: SlurmSystem, stdout: str, stderr: str, exp slurm_system.get_job_status(job) else: assert slurm_system.get_job_status(job) == expected + slurm_system.cmd_shell.execute.assert_called_once_with( + "TZ=UTC SLURM_TIME_FORMAT='%Y-%m-%dT%H:%M:%SZ' " + "sacct -j 1 --format=JobID,JobName,State,ExitCode,Start,End,ElapsedRAW,SubmitLine " + "--delimiter='|' -p --noheader" + ) sacct_output = """2623913,job,COMPLETED,0:0,2025-05-09T01:34:52,2025-05-09T01:59:27,1475,sbatch sbatch_script.sh, From d2798f59600a3ec9a8e41b346c5124d21b3ce0e0 Mon Sep 17 00:00:00 2001 From: Ivan Podkidyshev Date: Mon, 14 Sep 2026 20:03:29 +0200 Subject: [PATCH 03/17] Replace Slurm output implementation with an experiment output skeleton --- doc/reporting.rst | 36 ++-- src/cloudai/metrics.py | 4 +- src/cloudai/models/output.py | 107 ++++++++++ src/cloudai/output.py | 55 +++++ src/cloudai/systems/slurm/slurm_runner.py | 31 +-- src/cloudai/systems/slurm/slurm_system.py | 1 - src/cloudai/unified_output.py | 242 ---------------------- tests/systems/slurm/test_system.py | 31 +-- tests/test_unified_output.py | 179 ---------------- 9 files changed, 189 insertions(+), 497 deletions(-) create mode 100644 src/cloudai/models/output.py create mode 100644 src/cloudai/output.py delete mode 100644 src/cloudai/unified_output.py delete mode 100644 tests/test_unified_output.py diff --git a/doc/reporting.rst b/doc/reporting.rst index 99b4bc707..d4fc0aab1 100644 --- a/doc/reporting.rst +++ b/doc/reporting.rst @@ -31,19 +31,29 @@ Per-test reports are linked to a particular workload type (e.g. ``NcclTest``). A To list all available reports, users can use ``cloudai list-reports``. Use verbose output to also print report configurations. -Unified experiment output -------------------------- - -Ordinary Slurm scenarios write ``experiment.json`` in the scenario results directory when execution finishes or fails. -The file contains experiment metadata, test cases, submitted runs, statuses, timing, and canonical metrics from -``TestDefinition.metric_observations()``. NCCL and NIXLBench provide these metrics; other workloads have empty metric lists. -Test-level metrics are arithmetic means of successful iterations at matching metric and dimension points. Per-run -measurements retain their original values, and missing measurements are not treated as zero. - -Experiment timing covers scenario execution, including gaps between jobs. Run timestamps without timezone information -are null. Metric extraction and output-write failures produce warnings without changing execution behavior. The file is -replaced atomically and is independent of reporter configuration. Dry runs, DSE, and single-sbatch execution do not produce -this artifact. +Experiment output skeleton +-------------------------- + +``cloudai.models.output`` declares the full and short experiment models. +``cloudai.output.ExperimentOutput`` declares the collector interface; its operations raise +``NotImplementedError``. No runner is connected to it, and execution does not generate either JSON file. + +The interface defines these boundaries for implementation: + +- One collector belongs to the whole experiment, including ordinary iterations, DSE trials, or single-sbatch tests. + Scenario orchestration owns its lifetime; runners supply logical run updates before mutable test state advances. +- Slurm CLI/REST metadata normalization stays in the Slurm backend. Standalone supplies process status and UTC timing. + Canonical measurements come from ``TestDefinition.metric_observations()``. +- Per-run measurements remain intact. Summary aggregation groups successful repeats at matching metric and dimension + points within the same configuration; DSE summaries use the selected configuration. +- ``snapshot()`` derives short and full views from the same experiment state. ``write()`` publishes + ``experiment-summary.json`` and ``experiment.json`` at the scenario root, with atomic replacement per file. + Unknown timestamps remain null; extraction and write failures warn without changing benchmark behavior. +- Snapshot writing is separate from finalization so ongoing progress can use the same interface. + ``finish()`` belongs to the completion/failure boundary of the whole experiment, including all DSE trials. + +The runnable Slurm implementation, its tests, and the UTC accounting change are available at Git tag +``ipod/unified-output-v1`` (``f655642e``). .. _general-flow: diff --git a/src/cloudai/metrics.py b/src/cloudai/metrics.py index f2cee83f1..abf4c066a 100644 --- a/src/cloudai/metrics.py +++ b/src/cloudai/metrics.py @@ -44,8 +44,6 @@ class DimensionDefinition: key: str label: str value_type: Any - unit: str = "" - is_x: bool = False def validate(self, value: Any) -> MetricValue: """Validate one configured or observed dimension value.""" @@ -99,7 +97,7 @@ def validate_dimensions(cls, values: Mapping[str, Any]) -> MetricDimensions: return {key: cls.get_dimension(key).validate(value) for key, value in values.items()} -SIZE_BYTES = DimensionDefinition("size_bytes", "Size", Annotated[int, Field(strict=True, ge=0)], unit="B", is_x=True) +SIZE_BYTES = DimensionDefinition("size_bytes", "Size", Annotated[int, Field(strict=True, ge=0)]) BATCH_SIZE = DimensionDefinition("batch_size", "Batch size", Annotated[int, Field(strict=True, gt=0)]) OPERATION = DimensionDefinition("operation", "Operation", Annotated[str, Field(strict=True, min_length=1)]) PLACEMENT = DimensionDefinition("placement", "Placement", Literal["in_place", "out_of_place"]) diff --git a/src/cloudai/models/output.py b/src/cloudai/models/output.py new file mode 100644 index 000000000..80fcf75cc --- /dev/null +++ b/src/cloudai/models/output.py @@ -0,0 +1,107 @@ +# SPDX-FileCopyrightText: NVIDIA CORPORATION & AFFILIATES +# Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Backend-independent experiment output models based on API Schema v0.2.""" + +from datetime import datetime +from typing import Literal + +from pydantic import BaseModel, Field, FiniteFloat + +Status = Literal["pending", "running", "completed", "failed", "cancelled", "unknown"] + + +class Dimension(BaseModel): + """One coordinate of a metric measurement.""" + + name: str + value: str + unit: str = "" + is_x: bool = False + + +class Metric(BaseModel): + """A canonical measurement at a dimension point.""" + + name: str + value: str | int | FiniteFloat | bool + unit: str = "" + dimensions: list[Dimension] = Field(default_factory=list) + + +class Run(BaseModel): + """A logical execution with normalized metadata and original measurements.""" + + path: str + jobid: str + status: Status = "unknown" + metrics: list[Metric] = Field(default_factory=list) + start: datetime | None = None + finish: datetime | None = None + duration: FiniteFloat | None = None + iteration: int | None = None + step: int | None = None + + +class DSE(BaseModel): + """Search space and recommendation for one test case.""" + + space: dict[str, list[str | int | FiniteFloat]] + best_config: dict[str, str | int | FiniteFloat] | None = None + best_step: int | None = None + + +class TestShort(BaseModel): + """Test identity, status, and summary metrics.""" + + id: str + name: str + description: str | None = None + status: Status = "pending" + path: str + metrics: list[Metric] = Field(default_factory=list) + + +class Test(TestShort): + """A test case with all logical executions and optional DSE metadata.""" + + runs: list[Run] = Field(default_factory=list) + dse: DSE | None = None + + +class _ExperimentMetadata(BaseModel): + """Metadata shared by full and short experiment snapshots.""" + + id: str + name: str + description: str | None = None + status: Status = "pending" + path: str + start: datetime | None = None + finish: datetime | None = None + duration: FiniteFloat | None = None + + +class ExperimentShort(_ExperimentMetadata): + """Catalog snapshot without individual runs or DSE details.""" + + tests: list[TestShort] = Field(default_factory=list) + + +class Experiment(_ExperimentMetadata): + """Full snapshot spanning the entire scenario, including all DSE trials.""" + + tests: list[Test] = Field(default_factory=list) diff --git a/src/cloudai/output.py b/src/cloudai/output.py new file mode 100644 index 000000000..78ac0aaab --- /dev/null +++ b/src/cloudai/output.py @@ -0,0 +1,55 @@ +# SPDX-FileCopyrightText: NVIDIA CORPORATION & AFFILIATES +# Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Experiment output interface skeleton; no runner integration or persistence yet.""" + +from datetime import datetime +from pathlib import Path + +from cloudai.models.output import Experiment, ExperimentShort, Run, Status, Test + + +class ExperimentOutput: + """ + One collector per experiment, shared across iterations, DSE trials, and batch jobs. + + Runners provide normalized records, keeping scheduler-specific parsing outside this module. + The methods below describe the intended interface and are deliberately unimplemented. + """ + + def __init__(self, experiment: Experiment, output_path: Path) -> None: + self.experiment = experiment + self.output_path = output_path + + def update_run(self, test_id: str, run: Run) -> None: + """Upsert a logical run by test ID and run path before its mutable TestRun advances.""" + raise NotImplementedError + + def update_test(self, test: Test) -> None: + """Update test metadata, summary metrics, and optional DSE results.""" + raise NotImplementedError + + def snapshot(self) -> tuple[ExperimentShort, Experiment]: + """Derive consistent short and full snapshots without finalizing the experiment.""" + raise NotImplementedError + + def write(self) -> None: + """Publish experiment-summary.json and experiment.json, atomically per file, warning on errors.""" + raise NotImplementedError + + def finish(self, status: Status, finish: datetime | None) -> None: + """Finalize and publish after the whole experiment completes or fails.""" + raise NotImplementedError(f"Finalizing experiment output with status {status} at {finish} is not implemented") diff --git a/src/cloudai/systems/slurm/slurm_runner.py b/src/cloudai/systems/slurm/slurm_runner.py index f0db71c64..7cbcb0c54 100644 --- a/src/cloudai/systems/slurm/slurm_runner.py +++ b/src/cloudai/systems/slurm/slurm_runner.py @@ -21,8 +21,7 @@ import toml -from cloudai.core import BaseJob, BaseRunner, JobIdRetrievalError, JobStatusResult, System, TestRun, TestScenario -from cloudai.unified_output import ExperimentOutput +from cloudai.core import BaseJob, BaseRunner, JobIdRetrievalError, System, TestRun, TestScenario from cloudai.util import CommandShell from .slurm_command_gen_strategy import SlurmCommandGenStrategy @@ -44,34 +43,6 @@ def __init__(self, mode: str, system: System, test_scenario: TestScenario, outpu self.system = cast(SlurmSystem, system) self.cmd_shell = CommandShell() self.pinned_nodes: dict[str, list[str]] = {} - self._experiment: ExperimentOutput | None = None - - def run(self) -> None: - if self.mode != "run" or any(tr.is_dse_job or tr.step > 0 for tr in self.test_scenario.test_runs): - super().run() - return - - try: - self._experiment = ExperimentOutput(self.test_scenario, self.scenario_root) - except Exception as exc: - logging.warning("Cannot initialize unified experiment output: %s", exc) - completed = False - try: - super().run() - completed = True - finally: - if self._experiment is not None: - self._experiment.finish(self.system, self.jobs, completed) - self._experiment = None - - def get_job_status(self, job: BaseJob) -> JobStatusResult: - result = super().get_job_status(job) - if self._experiment is not None: - try: - self._experiment.capture(self.system, job, result) - except Exception as exc: - logging.warning("Cannot capture unified output for job %s: %s", job.id, exc) - return result def submit_test(self, tr: TestRun) -> None: if tr.pin_nodes and tr.name in self.pinned_nodes: diff --git a/src/cloudai/systems/slurm/slurm_system.py b/src/cloudai/systems/slurm/slurm_system.py index 7d39c6d2f..edef2989b 100644 --- a/src/cloudai/systems/slurm/slurm_system.py +++ b/src/cloudai/systems/slurm/slurm_system.py @@ -362,7 +362,6 @@ def is_job_completed(self, job: BaseJob, retry_threshold: int = 3) -> bool: def get_job_status(self, job: BaseJob, retry_threshold: int = 3) -> list[SlurmStepMetadata]: retry_count = 0 command = ( - "TZ=UTC SLURM_TIME_FORMAT='%Y-%m-%dT%H:%M:%SZ' " f"sacct -j {job.id} --format=JobID,JobName,State,ExitCode,Start,End,ElapsedRAW,SubmitLine " "--delimiter='|' -p --noheader" ) diff --git a/src/cloudai/unified_output.py b/src/cloudai/unified_output.py deleted file mode 100644 index 3e467e057..000000000 --- a/src/cloudai/unified_output.py +++ /dev/null @@ -1,242 +0,0 @@ -# SPDX-FileCopyrightText: NVIDIA CORPORATION & AFFILIATES -# Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. -# SPDX-License-Identifier: Apache-2.0 -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -import logging -from collections import defaultdict -from dataclasses import replace -from datetime import datetime, timezone -from pathlib import Path -from statistics import fmean -from tempfile import NamedTemporaryFile -from typing import Literal - -import toml -from pydantic import BaseModel, Field, FiniteFloat - -from cloudai.core import BaseJob, JobStatusResult, System, TestScenario -from cloudai.metrics import MetricCatalog, MetricObservation, MetricValue - -Status = Literal["pending", "running", "completed", "failed", "cancelled", "unknown"] - - -class Dimension(BaseModel): - """A metric coordinate in the unified output contract.""" - - name: str - value: str - unit: str = "" - is_x: bool = False - - -class Metric(BaseModel): - """A canonical measurement and its dimension point.""" - - name: str - value: FiniteFloat - unit: str - dimensions: list[Dimension] - - -class Run(BaseModel): - """One submitted execution, independent of the mutable TestRun.""" - - path: str - jobid: str - status: Status = "unknown" - metrics: list[Metric] = Field(default_factory=list) - start: datetime | None = None - finish: datetime | None = None - duration: FiniteFloat | None = None - iteration: int - step: int - - -class TestResult(BaseModel): - """A scenario test case and all its executed iterations.""" - - id: str - name: str - description: str - status: Status = "pending" - path: str - metrics: list[Metric] = Field(default_factory=list) - runs: list[Run] = Field(default_factory=list) - - -class Experiment(BaseModel): - """The API Schema v0.2 Experiment representation.""" - - id: str - name: str - status: Status = "unknown" - path: str - start: datetime | None - finish: datetime | None = None - duration: FiniteFloat | None = None - tests: list[TestResult] - - -def _timestamp(value: str) -> datetime | None: - try: - timestamp = datetime.fromisoformat(value.replace("Z", "+00:00")) - except ValueError: - return None - # sacct's default timestamps carry no timezone; the login host may use a different one. - return timestamp.astimezone(timezone.utc) if timestamp.tzinfo is not None else None - - -def _metric(observation: MetricObservation) -> Metric: - dimensions = [] - for key, value in sorted(observation.dimensions.items()): - definition = MetricCatalog.get_dimension(key) - dimensions.append( - Dimension(name=definition.label, value=str(value), unit=definition.unit, is_x=definition.is_x) - ) - return Metric( - name=observation.metric.display_name, - value=observation.value, - unit=observation.metric.unit, - dimensions=dimensions, - ) - - -def _status(statuses: set[Status]) -> Status: - if "failed" in statuses: - return "failed" - if "cancelled" in statuses: - return "cancelled" - if statuses == {"pending"}: - return "pending" - return "completed" if statuses == {"completed"} else "unknown" - - -class ExperimentOutput: - """Collect completed Slurm runs and atomically write one experiment.json.""" - - def __init__(self, scenario: TestScenario, output_path: Path): - self.scenario = scenario - self.output_path = output_path.absolute() - self.tests = { - tr.name: TestResult( - id=tr.name, - name=tr.name, - description=tr.test.description, - path=str(self.output_path / tr.name), - ) - for tr in scenario.test_runs - } - self.observations: dict[str, list[MetricObservation]] = defaultdict(list) - self.experiment = Experiment( - id=self.output_path.name, - name=scenario.name, - path=str(self.output_path), - start=datetime.now(timezone.utc), - tests=list(self.tests.values()), - ) - - def capture(self, system: System, job: BaseJob, result: JobStatusResult | None = None) -> None: - tr = job.test_run - test = self.tests[tr.name] - if any(run.jobid == str(job.id) and run.path == str(tr.output_path.absolute()) for run in test.runs): - return - run = Run( - path=str(tr.output_path.absolute()), - jobid=str(job.id), - iteration=tr.current_iteration, - step=tr.step, - status="failed" if result is not None and not result.is_successful else "unknown", - ) - test.runs.append(run) - metadata_path = tr.output_path / "slurm-job.toml" - try: - metadata = toml.load(metadata_path) - state = metadata["state"].split()[0].rstrip("+") - if state == "CANCELLED": - run.status = "cancelled" - elif state in { - "FAILED", - "TIMEOUT", - "NODE_FAIL", - "OUT_OF_MEMORY", - "BOOT_FAIL", - "DEADLINE", - "PREEMPTED", - "REVOKED", - "SPECIAL_EXIT", - } or metadata["exit_code"] not in {"0", "0:0", ""}: - run.status = "failed" - elif state == "COMPLETED" and result is not None and result.is_successful: - run.status = "completed" - run.start = _timestamp(metadata["start_time"]) - run.finish = _timestamp(metadata["end_time"]) - run.duration = metadata["elapsed_time_sec"] - except (OSError, ValueError, KeyError, TypeError, IndexError) as exc: - if result is not None or metadata_path.exists(): - logging.warning("Cannot read unified output metadata for job %s: %s", job.id, exc) - - if result is None: - return - try: - observations = sorted( - tr.test.metric_observations(system, tr), - key=lambda observation: (observation.metric.key, sorted(observation.dimensions.items())), - ) - run.metrics = [_metric(observation) for observation in observations] - if run.status == "completed": - self.observations[tr.name].extend(observations) - except Exception as exc: - logging.warning("Cannot extract unified output metrics for job %s: %s", job.id, exc) - - def finish(self, system: System, unfinished_jobs: list[BaseJob], completed: bool) -> None: - temporary_path = None - try: - for job in unfinished_jobs: - self.capture(system, job) - for tr in self.scenario.test_runs: - test = self.tests[tr.name] - statuses: set[Status] = {run.status for run in test.runs} - if len(test.runs) != tr.iterations: - statuses.add("unknown" if test.runs else "pending") - test.status = _status(statuses) - - groups: dict[tuple[str, tuple[tuple[str, MetricValue], ...]], list[MetricObservation]] = defaultdict( - list - ) - for observation in self.observations[tr.name]: - groups[(observation.metric.key, tuple(sorted(observation.dimensions.items())))].append(observation) - test.metrics = [ - _metric(replace(group[0], value=fmean(observation.value for observation in group))) - for _, group in sorted(groups.items()) - ] - - self.experiment.status = _status({test.status for test in self.tests.values()}) if completed else "failed" - self.experiment.finish = datetime.now(timezone.utc) - if self.experiment.start is not None: - self.experiment.duration = (self.experiment.finish - self.experiment.start).total_seconds() - content = self.experiment.model_dump_json(indent=2) - self.output_path.mkdir(parents=True, exist_ok=True) - with NamedTemporaryFile(mode="w", encoding="utf-8", dir=self.output_path, delete=False) as temporary: - temporary_path = Path(temporary.name) - temporary.write(content + "\n") - temporary_path.replace(self.output_path / "experiment.json") - except Exception as exc: - logging.warning("Cannot write unified experiment output: %s", exc) - finally: - if temporary_path is not None: - try: - temporary_path.unlink(missing_ok=True) - except OSError as exc: - logging.warning("Cannot remove temporary unified output %s: %s", temporary_path, exc) diff --git a/tests/systems/slurm/test_system.py b/tests/systems/slurm/test_system.py index 17515b014..0f1b69ebf 100644 --- a/tests/systems/slurm/test_system.py +++ b/tests/systems/slurm/test_system.py @@ -673,30 +673,8 @@ def test_env_vars_order(): assert actual_order == expected_order -@pytest.mark.parametrize( - "stdout,stderr,expected", - [ - ("", "error", None), - ( - "1|job|COMPLETED|0:0|2026-09-14T14:41:02Z|2026-09-14T14:42:41Z|99|sbatch job.sh|\n", - "", - [ - SlurmStepMetadata( - job_id=1, - step_id="", - name="job", - state="COMPLETED", - exit_code="0:0", - start_time="2026-09-14T14:41:02Z", - end_time="2026-09-14T14:42:41Z", - elapsed_time_sec=99, - submit_line="sbatch job.sh", - ) - ], - ), - ], -) -def test_get_job_status(slurm_system: SlurmSystem, stdout: str, stderr: str, expected: list[SlurmStepMetadata] | None): +@pytest.mark.parametrize("stdout,stderr, expected", [("", "error", None)]) +def test_get_job_status(slurm_system: SlurmSystem, stdout: str, stderr: str, expected: tuple): job = BaseJob(test_run=Mock(), id=1) pp = Mock() pp.communicate = Mock(return_value=(stdout, stderr)) @@ -707,11 +685,6 @@ def test_get_job_status(slurm_system: SlurmSystem, stdout: str, stderr: str, exp slurm_system.get_job_status(job) else: assert slurm_system.get_job_status(job) == expected - slurm_system.cmd_shell.execute.assert_called_once_with( - "TZ=UTC SLURM_TIME_FORMAT='%Y-%m-%dT%H:%M:%SZ' " - "sacct -j 1 --format=JobID,JobName,State,ExitCode,Start,End,ElapsedRAW,SubmitLine " - "--delimiter='|' -p --noheader" - ) sacct_output = """2623913,job,COMPLETED,0:0,2025-05-09T01:34:52,2025-05-09T01:59:27,1475,sbatch sbatch_script.sh, diff --git a/tests/test_unified_output.py b/tests/test_unified_output.py deleted file mode 100644 index 7392ef794..000000000 --- a/tests/test_unified_output.py +++ /dev/null @@ -1,179 +0,0 @@ -# SPDX-FileCopyrightText: NVIDIA CORPORATION & AFFILIATES -# Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. -# SPDX-License-Identifier: Apache-2.0 -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -import json -from pathlib import Path -from unittest.mock import Mock - -import pytest -import toml - -from cloudai._core.exceptions import JobFailureError -from cloudai.core import BaseJob, TestRun, TestScenario -from cloudai.systems.slurm import SlurmJob, SlurmRunner, SlurmSystem -from cloudai.unified_output import Experiment, ExperimentOutput -from cloudai.workloads.nccl_test import NCCLTestDefinition - - -@pytest.fixture -def runner(slurm_system: SlurmSystem, nccl_tr: TestRun, monkeypatch: pytest.MonkeyPatch) -> SlurmRunner: - stdout = (nccl_tr.output_path / "stdout.txt").read_text() + "# Out of bounds values : 0\n" - nccl_tr.iterations = 2 - slurm_system.output_path.mkdir() - runner = SlurmRunner( - "run", slurm_system, TestScenario("repeats", [nccl_tr]), slurm_system.output_path / "repeats_2026-09-11" - ) - - def submit(tr: TestRun) -> SlurmJob: - content = stdout - if tr.current_iteration: - content = "\n".join(line for line in stdout.splitlines() if not line.lstrip().startswith("12000000")) - content = content.replace("20.20", "40.20") - (tr.output_path / "stdout.txt").write_text(content) - return SlurmJob(tr, id=100 + tr.current_iteration) - - def complete(job: BaseJob) -> None: - offset = "+02:00" if job.test_run.current_iteration == 0 else "" - metadata = { - "state": "COMPLETED", - "exit_code": "0:0", - "start_time": f"2026-09-11T12:00:00{offset}", - "end_time": f"2026-09-11T12:00:02{offset}", - "elapsed_time_sec": 2, - } - (job.test_run.output_path / "slurm-job.toml").write_text(toml.dumps(metadata)) - - monkeypatch.setattr(runner, "on_job_submit", Mock()) - monkeypatch.setattr(runner, "_submit_test", submit) - monkeypatch.setattr(runner, "on_job_completion", complete) - monkeypatch.setattr(SlurmSystem, "is_job_completed", lambda self, job: True) - monkeypatch.setattr(SlurmSystem, "is_job_running", lambda self, job: False) - monkeypatch.setattr(SlurmSystem, "kill", Mock()) - return runner - - -def test_completed_experiment_and_repeated_metrics(runner: SlurmRunner): - runner.run() - - path = runner.scenario_root / "experiment.json" - experiment = Experiment.model_validate_json(path.read_text()) - data = json.loads(path.read_text()) - assert set(data) == {"id", "name", "status", "path", "start", "finish", "duration", "tests"} - assert experiment.id == runner.scenario_root.name - assert experiment.name == "repeats" - assert experiment.status == "completed" - assert experiment.start is not None and experiment.finish is not None - assert experiment.duration == (experiment.finish - experiment.start).total_seconds() - test = experiment.tests[0] - assert test.id == test.name == "nccl_test" - assert test.status == "completed" - assert test.path == str(runner.scenario_root / "nccl_test") - assert [run.iteration for run in test.runs] == [0, 1] - assert [run.step for run in test.runs] == [0, 0] - assert [run.jobid for run in test.runs] == ["100", "101"] - assert [run.path for run in test.runs] == [str(Path(test.path) / str(i)) for i in range(2)] - assert data["tests"][0]["runs"][0]["start"] == "2026-09-11T10:00:00Z" - assert test.runs[1].start is None and test.runs[1].finish is None - assert [run.duration for run in test.runs] == [2, 2] - - bandwidth = [ - metric - for metric in test.metrics - if metric.name == "Bandwidth" - and any(dimension.name == "Placement" and dimension.value == "out_of_place" for dimension in metric.dimensions) - ] - assert [metric.value for metric in bandwidth] == pytest.approx([30.20, 30.30, 130.40]) - assert all(metric.unit == "GB/s" for metric in bandwidth) - sizes = [dimension for metric in bandwidth for dimension in metric.dimensions if dimension.name == "Size"] - assert [size.value for size in sizes] == ["1000000", "2000000", "12000000"] - assert all(size.unit == "B" and size.is_x for size in sizes) - assert len(test.runs[0].metrics) == 12 - assert len(test.runs[1].metrics) == 8 - assert test.runs[0].metrics != test.runs[1].metrics - assert list(runner.scenario_root.glob("*.json")) == [path] - - -@pytest.mark.parametrize("failure,abort", [("workload", True), ("workload", False), ("scheduler", False)]) -def test_failed_run_preserves_earlier_results( - runner: SlurmRunner, monkeypatch: pytest.MonkeyPatch, failure: str, abort: bool -): - complete = runner.on_job_completion - runner.test_scenario.job_status_check = abort - - def fail_second_job(job: BaseJob) -> None: - complete(job) - if job.test_run.current_iteration != 1: - return - if failure == "workload": - with (job.test_run.output_path / "stdout.txt").open("a") as stdout: - stdout.write("\nTest NCCL failure\n") - else: - path = job.test_run.output_path / "slurm-job.toml" - path.write_text(path.read_text().replace("COMPLETED", "FAILED")) - - monkeypatch.setattr(runner, "on_job_completion", fail_second_job) - if abort: - with pytest.raises(JobFailureError): - runner.run() - else: - runner.run() - - experiment = Experiment.model_validate_json((runner.scenario_root / "experiment.json").read_text()) - test = experiment.tests[0] - assert experiment.status == test.status == "failed" - assert [run.status for run in test.runs] == ["completed", "failed"] - assert [run.iteration for run in test.runs] == [0, 1] - assert test.runs[0].path != test.runs[1].path - assert len(test.runs[1].metrics) == 8 - assert test.metrics == test.runs[0].metrics - - -def test_output_failures_are_nonfatal( - runner: SlurmRunner, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture -): - extract = NCCLTestDefinition.metric_observations - - def failing_extract(self, system, tr): - if tr.current_iteration == 1: - raise ValueError("malformed measurement") - return extract(self, system, tr) - - monkeypatch.setattr(NCCLTestDefinition, "metric_observations", failing_extract) - runner.run() - - path = runner.scenario_root / "experiment.json" - original = path.read_bytes() - experiment = Experiment.model_validate_json(original) - assert experiment.status == "completed" - assert experiment.tests[0].runs[1].metrics == [] - assert experiment.tests[0].metrics == experiment.tests[0].runs[0].metrics - assert "Cannot extract unified output metrics for job 101: malformed measurement" in caplog.text - - monkeypatch.setattr(Path, "replace", Mock(side_effect=OSError("write unavailable"))) - output = ExperimentOutput(runner.test_scenario, runner.scenario_root) - output.finish(runner.system, [], completed=False) - assert path.read_bytes() == original - assert "Cannot write unified experiment output: write unavailable" in caplog.text - assert sorted(p.name for p in runner.scenario_root.iterdir()) == ["experiment.json", "nccl_test"] - - monkeypatch.setattr( - "cloudai.systems.slurm.slurm_runner.ExperimentOutput", Mock(side_effect=OSError("initialization unavailable")) - ) - runner.test_scenario.test_runs[0].current_iteration = 0 - runner.run() - assert len(runner.jobs) == 0 - assert path.read_bytes() == original - assert "Cannot initialize unified experiment output: initialization unavailable" in caplog.text From 4e8b2b0b8fc9c2ea26c439b706cce115b7c18282 Mon Sep 17 00:00:00 2001 From: Ivan Podkidyshev Date: Mon, 14 Sep 2026 20:49:15 +0200 Subject: [PATCH 04/17] Connect experiment output skeleton to runner lifecycle hooks --- doc/reporting.rst | 8 +++- src/cloudai/_core/base_runner.py | 39 ++++++++++++++++++- src/cloudai/cli/handlers.py | 21 ++++++---- src/cloudai/output.py | 2 +- .../systems/slurm/single_sbatch_runner.py | 2 + src/cloudai/systems/slurm/slurm_runner.py | 12 ++++-- .../systems/standalone/standalone_runner.py | 10 ++++- 7 files changed, 78 insertions(+), 16 deletions(-) diff --git a/doc/reporting.rst b/doc/reporting.rst index d4fc0aab1..73b665e99 100644 --- a/doc/reporting.rst +++ b/doc/reporting.rst @@ -36,12 +36,18 @@ Experiment output skeleton ``cloudai.models.output`` declares the full and short experiment models. ``cloudai.output.ExperimentOutput`` declares the collector interface; its operations raise -``NotImplementedError``. No runner is connected to it, and execution does not generate either JSON file. +``NotImplementedError``. Runner call sites are guarded by an optional collector. Its creation hook returns ``None``, +so execution does not generate either JSON file. The interface defines these boundaries for implementation: - One collector belongs to the whole experiment, including ordinary iterations, DSE trials, or single-sbatch tests. Scenario orchestration owns its lifetime; runners supply logical run updates before mutable test state advances. +- ``BaseRunner`` calls ``update_run_output()`` after submission and workload validation, before advancing an iteration. + Slurm and standalone declare normalization stubs. Single-sbatch calls the same hook from its separate execution loop, + expanding the allocation into logical runs through ``completed_test_runs()``. +- The CLI scenario boundary calls creation, initial snapshot, and finalization hooks. DSE reuses the attached collector + across trials; individual runner invocations do not finalize it. Final status and timing are still placeholders. - Slurm CLI/REST metadata normalization stays in the Slurm backend. Standalone supplies process status and UTC timing. Canonical measurements come from ``TestDefinition.metric_observations()``. - Per-run measurements remain intact. Summary aggregation groups successful repeats at matching metric and dimension diff --git a/src/cloudai/_core/base_runner.py b/src/cloudai/_core/base_runner.py index 10f6bc01c..43a6b2765 100644 --- a/src/cloudai/_core/base_runner.py +++ b/src/cloudai/_core/base_runner.py @@ -18,7 +18,11 @@ import time from abc import ABC, abstractmethod from pathlib import Path -from typing import Dict, List +from typing import TYPE_CHECKING, Dict, List + +if TYPE_CHECKING: + from cloudai.models.output import Run + from cloudai.output import ExperimentOutput from .base_job import BaseJob from .command_gen_strategy import CommandGenStrategy @@ -69,6 +73,36 @@ def __init__(self, mode: str, system: System, test_scenario: TestScenario, outpu self.testrun_to_job_map: Dict[TestRun, BaseJob] = {} logging.debug(f"{self.__class__.__name__} initialized") self.shutting_down = False + self.experiment_output: ExperimentOutput | None = None + + def create_experiment_output(self) -> "ExperimentOutput | None": + """Initialize from the original scenario; collector creation is not implemented yet.""" + return None + + def get_run_output(self, job: BaseJob, tr: TestRun, result: JobStatusResult | None = None) -> "Run": + """Normalize one logical execution; result is absent at submission.""" + raise NotImplementedError + + def completed_test_runs(self, job: BaseJob) -> list[TestRun]: + """Return logical executions represented by a scheduler job.""" + return [job.test_run] + + def update_run_output(self, job: BaseJob, result: JobStatusResult | None = None) -> None: + """Capture logical runs before iteration or DSE state advances, then publish a snapshot.""" + if self.mode == "run" and self.experiment_output is not None: + for tr in self.completed_test_runs(job): + self.experiment_output.update_run(str(tr.name), self.get_run_output(job, tr, result)) + self.write_output() + + def write_output(self) -> None: + """Publish a progress snapshot when a collector is attached.""" + if self.mode == "run" and self.experiment_output is not None: + self.experiment_output.write() + + def finish_output(self) -> None: + """Finalize the whole experiment; aggregate status and timing remain placeholders.""" + if self.mode == "run" and self.experiment_output is not None: + self.experiment_output.finish(status="unknown", finish=None) def shutdown(self): """Gracefully shut down the runner, terminating all outstanding jobs.""" @@ -110,6 +144,7 @@ def submit_test(self, tr: TestRun): job = self._submit_test(tr) self.jobs.append(job) self.testrun_to_job_map[tr] = job + self.update_run_output(job) except JobSubmissionError as e: logging.error(e) exit(1) @@ -251,6 +286,7 @@ def monitor_jobs(self) -> int: else: if self.test_scenario.job_status_check: job_status_result = self.get_job_status(job) + self.update_run_output(job, job_status_result) if job_status_result.is_successful: successful_jobs_count += 1 self.handle_job_completion(job) @@ -264,6 +300,7 @@ def monitor_jobs(self) -> int: raise JobFailureError(job.test_run.name, error_message, job_status_result.error_message) else: job_status_result = self.get_job_status(job) + self.update_run_output(job, job_status_result) if not job_status_result.is_successful: error_message = ( f"Job {job.id} for test {job.test_run.name} failed: {job_status_result.error_message}" diff --git a/src/cloudai/cli/handlers.py b/src/cloudai/cli/handlers.py index 1fc620544..2d3e39026 100644 --- a/src/cloudai/cli/handlers.py +++ b/src/cloudai/cli/handlers.py @@ -352,16 +352,21 @@ def handle_dry_run_and_run(args: argparse.Namespace) -> int: register_signal_handlers(runner.cancel_on_signal) logging.info(f"Scenario results will be stored at: {runner.runner.scenario_root}") - has_dse = any(tr.is_dse_job for tr in test_scenario.test_runs) - if args.single_sbatch or not has_dse: # in this mode cases are unrolled using grid search - handle_non_dse_job(runner, args) - return 0 + runner.runner.experiment_output = runner.runner.create_experiment_output() + try: + runner.runner.write_output() + has_dse = any(tr.is_dse_job for tr in test_scenario.test_runs) + if args.single_sbatch or not has_dse: # in this mode cases are unrolled using grid search + handle_non_dse_job(runner, args) + return 0 - if all(tr.is_dse_job for tr in test_scenario.test_runs): - return handle_dse_job(runner, args) + if all(tr.is_dse_job for tr in test_scenario.test_runs): + return handle_dse_job(runner, args) - logging.error("Mixing DSE and non-DSE jobs is not allowed.") - return 1 + logging.error("Mixing DSE and non-DSE jobs is not allowed.") + return 1 + finally: + runner.runner.finish_output() def handle_generate_report(args: argparse.Namespace) -> int: diff --git a/src/cloudai/output.py b/src/cloudai/output.py index 78ac0aaab..0108cc2f5 100644 --- a/src/cloudai/output.py +++ b/src/cloudai/output.py @@ -14,7 +14,7 @@ # See the License for the specific language governing permissions and # limitations under the License. -"""Experiment output interface skeleton; no runner integration or persistence yet.""" +"""Experiment output interface skeleton; runner hooks are inactive until collector creation is implemented.""" from datetime import datetime from pathlib import Path diff --git a/src/cloudai/systems/slurm/single_sbatch_runner.py b/src/cloudai/systems/slurm/single_sbatch_runner.py index 3a49041f4..e17959925 100644 --- a/src/cloudai/systems/slurm/single_sbatch_runner.py +++ b/src/cloudai/systems/slurm/single_sbatch_runner.py @@ -191,6 +191,7 @@ def run(self): tr = self.test_scenario.test_runs[0] job = self._submit_test(tr) self.jobs.append(job) + self.update_run_output(job) if self.shutting_down: self.system.kill(job) @@ -208,6 +209,7 @@ def run(self): self.handle_dse() self.on_job_completion(job) + self.update_run_output(job) def handle_dse(self): registry = Registry() diff --git a/src/cloudai/systems/slurm/slurm_runner.py b/src/cloudai/systems/slurm/slurm_runner.py index 7cbcb0c54..3499477fe 100644 --- a/src/cloudai/systems/slurm/slurm_runner.py +++ b/src/cloudai/systems/slurm/slurm_runner.py @@ -17,11 +17,11 @@ import logging import re from pathlib import Path -from typing import cast +from typing import TYPE_CHECKING, cast import toml -from cloudai.core import BaseJob, BaseRunner, JobIdRetrievalError, System, TestRun, TestScenario +from cloudai.core import BaseJob, BaseRunner, JobIdRetrievalError, JobStatusResult, System, TestRun, TestScenario from cloudai.util import CommandShell from .slurm_command_gen_strategy import SlurmCommandGenStrategy @@ -29,6 +29,9 @@ from .slurm_metadata import SlurmJobMetadata, SlurmStepMetadata from .slurm_system import SlurmSystem +if TYPE_CHECKING: + from cloudai.models.output import Run + class SlurmRunner(BaseRunner): """ @@ -92,8 +95,9 @@ def on_job_submit(self, tr: TestRun) -> None: cmd_gen = self.get_cmd_gen_strategy(self.system, tr) cmd_gen.store_test_run() - def completed_test_runs(self, job: BaseJob) -> list[TestRun]: - return [cast(SlurmJob, job).test_run] + def get_run_output(self, job: BaseJob, tr: TestRun, result: JobStatusResult | None = None) -> "Run": + """Normalize CLI/REST metadata and canonical metrics without assuming per-test allocation timing.""" + raise NotImplementedError def on_job_completion(self, job: BaseJob) -> None: logging.debug(f"Job completion callback for job {job.id}") diff --git a/src/cloudai/systems/standalone/standalone_runner.py b/src/cloudai/systems/standalone/standalone_runner.py index 016fdbabc..a25639c6b 100644 --- a/src/cloudai/systems/standalone/standalone_runner.py +++ b/src/cloudai/systems/standalone/standalone_runner.py @@ -16,12 +16,16 @@ import logging from pathlib import Path +from typing import TYPE_CHECKING -from cloudai.core import BaseRunner, JobIdRetrievalError, System, TestRun, TestScenario +from cloudai.core import BaseJob, BaseRunner, JobIdRetrievalError, JobStatusResult, System, TestRun, TestScenario from cloudai.util import CommandShell from .standalone_job import StandaloneJob +if TYPE_CHECKING: + from cloudai.models.output import Run + class StandaloneRunner(BaseRunner): """ @@ -35,6 +39,10 @@ def __init__(self, mode: str, system: System, test_scenario: TestScenario, outpu super().__init__(mode, system, test_scenario, output_path) self.cmd_shell = CommandShell() + def get_run_output(self, job: BaseJob, tr: TestRun, result: JobStatusResult | None = None) -> "Run": + """Normalize PID, process outcome, UTC timing, and canonical metrics; process tracking is pending.""" + raise NotImplementedError + def _submit_test(self, tr: TestRun) -> StandaloneJob: logging.info(f"Running test: {tr.name}") tr.output_path = self.get_job_output_path(tr) From 65db5ab22c304a1779f446b14402becba3c0a72f Mon Sep 17 00:00:00 2001 From: Ivan Podkidyshev Date: Wed, 16 Sep 2026 11:59:15 +0200 Subject: [PATCH 05/17] Revert skeleton documentation changes --- doc/reporting.rst | 42 +++++++++++++----------------------------- 1 file changed, 13 insertions(+), 29 deletions(-) diff --git a/doc/reporting.rst b/doc/reporting.rst index 73b665e99..99b4bc707 100644 --- a/doc/reporting.rst +++ b/doc/reporting.rst @@ -31,35 +31,19 @@ Per-test reports are linked to a particular workload type (e.g. ``NcclTest``). A To list all available reports, users can use ``cloudai list-reports``. Use verbose output to also print report configurations. -Experiment output skeleton --------------------------- - -``cloudai.models.output`` declares the full and short experiment models. -``cloudai.output.ExperimentOutput`` declares the collector interface; its operations raise -``NotImplementedError``. Runner call sites are guarded by an optional collector. Its creation hook returns ``None``, -so execution does not generate either JSON file. - -The interface defines these boundaries for implementation: - -- One collector belongs to the whole experiment, including ordinary iterations, DSE trials, or single-sbatch tests. - Scenario orchestration owns its lifetime; runners supply logical run updates before mutable test state advances. -- ``BaseRunner`` calls ``update_run_output()`` after submission and workload validation, before advancing an iteration. - Slurm and standalone declare normalization stubs. Single-sbatch calls the same hook from its separate execution loop, - expanding the allocation into logical runs through ``completed_test_runs()``. -- The CLI scenario boundary calls creation, initial snapshot, and finalization hooks. DSE reuses the attached collector - across trials; individual runner invocations do not finalize it. Final status and timing are still placeholders. -- Slurm CLI/REST metadata normalization stays in the Slurm backend. Standalone supplies process status and UTC timing. - Canonical measurements come from ``TestDefinition.metric_observations()``. -- Per-run measurements remain intact. Summary aggregation groups successful repeats at matching metric and dimension - points within the same configuration; DSE summaries use the selected configuration. -- ``snapshot()`` derives short and full views from the same experiment state. ``write()`` publishes - ``experiment-summary.json`` and ``experiment.json`` at the scenario root, with atomic replacement per file. - Unknown timestamps remain null; extraction and write failures warn without changing benchmark behavior. -- Snapshot writing is separate from finalization so ongoing progress can use the same interface. - ``finish()`` belongs to the completion/failure boundary of the whole experiment, including all DSE trials. - -The runnable Slurm implementation, its tests, and the UTC accounting change are available at Git tag -``ipod/unified-output-v1`` (``f655642e``). +Unified experiment output +------------------------- + +Ordinary Slurm scenarios write ``experiment.json`` in the scenario results directory when execution finishes or fails. +The file contains experiment metadata, test cases, submitted runs, statuses, timing, and canonical metrics from +``TestDefinition.metric_observations()``. NCCL and NIXLBench provide these metrics; other workloads have empty metric lists. +Test-level metrics are arithmetic means of successful iterations at matching metric and dimension points. Per-run +measurements retain their original values, and missing measurements are not treated as zero. + +Experiment timing covers scenario execution, including gaps between jobs. Run timestamps without timezone information +are null. Metric extraction and output-write failures produce warnings without changing execution behavior. The file is +replaced atomically and is independent of reporter configuration. Dry runs, DSE, and single-sbatch execution do not produce +this artifact. .. _general-flow: From 4b9f53d6fa391e3b0c82ae1d89ea79410ebbb8db Mon Sep 17 00:00:00 2001 From: Ivan Podkidyshev Date: Wed, 16 Sep 2026 16:59:07 +0200 Subject: [PATCH 06/17] Implement experiment output collection and snapshot writing --- src/cloudai/_core/base_runner.py | 5 +- src/cloudai/cli/handlers.py | 1 - src/cloudai/models/output.py | 4 - src/cloudai/output.py | 126 +++++++++++++++--- src/cloudai/systems/slurm/slurm_runner.py | 1 - .../systems/standalone/standalone_runner.py | 1 - 6 files changed, 106 insertions(+), 32 deletions(-) diff --git a/src/cloudai/_core/base_runner.py b/src/cloudai/_core/base_runner.py index 43a6b2765..cda5370ad 100644 --- a/src/cloudai/_core/base_runner.py +++ b/src/cloudai/_core/base_runner.py @@ -73,10 +73,9 @@ def __init__(self, mode: str, system: System, test_scenario: TestScenario, outpu self.testrun_to_job_map: Dict[TestRun, BaseJob] = {} logging.debug(f"{self.__class__.__name__} initialized") self.shutting_down = False - self.experiment_output: ExperimentOutput | None = None + self.experiment_output: ExperimentOutput | None = self.create_experiment_output() def create_experiment_output(self) -> "ExperimentOutput | None": - """Initialize from the original scenario; collector creation is not implemented yet.""" return None def get_run_output(self, job: BaseJob, tr: TestRun, result: JobStatusResult | None = None) -> "Run": @@ -95,12 +94,10 @@ def update_run_output(self, job: BaseJob, result: JobStatusResult | None = None) self.write_output() def write_output(self) -> None: - """Publish a progress snapshot when a collector is attached.""" if self.mode == "run" and self.experiment_output is not None: self.experiment_output.write() def finish_output(self) -> None: - """Finalize the whole experiment; aggregate status and timing remain placeholders.""" if self.mode == "run" and self.experiment_output is not None: self.experiment_output.finish(status="unknown", finish=None) diff --git a/src/cloudai/cli/handlers.py b/src/cloudai/cli/handlers.py index 2d3e39026..91984add4 100644 --- a/src/cloudai/cli/handlers.py +++ b/src/cloudai/cli/handlers.py @@ -352,7 +352,6 @@ def handle_dry_run_and_run(args: argparse.Namespace) -> int: register_signal_handlers(runner.cancel_on_signal) logging.info(f"Scenario results will be stored at: {runner.runner.scenario_root}") - runner.runner.experiment_output = runner.runner.create_experiment_output() try: runner.runner.write_output() has_dse = any(tr.is_dse_job for tr in test_scenario.test_runs) diff --git a/src/cloudai/models/output.py b/src/cloudai/models/output.py index 80fcf75cc..c3fd88aff 100644 --- a/src/cloudai/models/output.py +++ b/src/cloudai/models/output.py @@ -14,8 +14,6 @@ # See the License for the specific language governing permissions and # limitations under the License. -"""Backend-independent experiment output models based on API Schema v0.2.""" - from datetime import datetime from typing import Literal @@ -83,8 +81,6 @@ class Test(TestShort): class _ExperimentMetadata(BaseModel): - """Metadata shared by full and short experiment snapshots.""" - id: str name: str description: str | None = None diff --git a/src/cloudai/output.py b/src/cloudai/output.py index 0108cc2f5..d090d18c0 100644 --- a/src/cloudai/output.py +++ b/src/cloudai/output.py @@ -14,42 +14,126 @@ # See the License for the specific language governing permissions and # limitations under the License. -"""Experiment output interface skeleton; runner hooks are inactive until collector creation is implemented.""" - -from datetime import datetime +import logging +from collections import defaultdict +from datetime import datetime, timezone from pathlib import Path +from statistics import fmean +from tempfile import NamedTemporaryFile -from cloudai.models.output import Experiment, ExperimentShort, Run, Status, Test +from cloudai.models.output import Experiment, ExperimentShort, Metric, Run, Status, Test class ExperimentOutput: - """ - One collector per experiment, shared across iterations, DSE trials, and batch jobs. - - Runners provide normalized records, keeping scheduler-specific parsing outside this module. - The methods below describe the intended interface and are deliberately unimplemented. - """ + """Collect experiment results and publish full and short snapshots.""" def __init__(self, experiment: Experiment, output_path: Path) -> None: - self.experiment = experiment + self.experiment = experiment.model_copy(deep=True) self.output_path = output_path def update_run(self, test_id: str, run: Run) -> None: - """Upsert a logical run by test ID and run path before its mutable TestRun advances.""" - raise NotImplementedError + """Store the latest state of a run, identified by test and output path.""" + test = next((test for test in self.experiment.tests if test.id == test_id), None) + if test is None: + raise KeyError(f"Unknown experiment test: {test_id}") + recorded = run.model_copy(deep=True) + for index, current in enumerate(test.runs): + if current.path == run.path: + test.runs[index] = recorded + return + test.runs.append(recorded) def update_test(self, test: Test) -> None: - """Update test metadata, summary metrics, and optional DSE results.""" - raise NotImplementedError + """Update a test while retaining its previously recorded runs.""" + recorded = test.model_copy(deep=True) + for index, current in enumerate(self.experiment.tests): + if current.id == test.id: + incoming_runs = recorded.runs + recorded.runs = current.runs + self.experiment.tests[index] = recorded + for run in incoming_runs: + self.update_run(test.id, run) + return + self.experiment.tests.append(recorded) def snapshot(self) -> tuple[ExperimentShort, Experiment]: - """Derive consistent short and full snapshots without finalizing the experiment.""" - raise NotImplementedError + """Return independent short and full views without finalizing the experiment.""" + full = self.experiment.model_copy(deep=True) + self._update_timing(full) + for test in full.tests: + for run in test.runs: + self._update_timing(run) + if test.metrics or test.dse is not None or any(run.step not in (None, 0) for run in test.runs): + continue + test.metrics = self._aggregate_metrics(test.runs) + short = ExperimentShort.model_validate(full.model_dump()) + return short, full def write(self) -> None: - """Publish experiment-summary.json and experiment.json, atomically per file, warning on errors.""" - raise NotImplementedError + """Atomically replace each output file, warning on failure.""" + pending: list[tuple[Path, Path]] = [] + try: + short, full = self.snapshot() + contents = [ + ("experiment.json", full.model_dump_json(indent=2)), + ("experiment-summary.json", short.model_dump_json(indent=2)), + ] + self.output_path.mkdir(parents=True, exist_ok=True) + for filename, content in contents: + with NamedTemporaryFile( + mode="w", encoding="utf-8", dir=self.output_path, prefix=f".{filename}.", delete=False + ) as temporary: + pending.append((Path(temporary.name), self.output_path / filename)) + temporary.write(content + "\n") + for temporary_path, destination in pending: + temporary_path.replace(destination) + except Exception as exc: + logging.warning("Cannot write experiment output: %s", exc) + finally: + for temporary_path, _ in pending: + try: + temporary_path.unlink(missing_ok=True) + except OSError as exc: + logging.warning("Cannot remove temporary experiment output %s: %s", temporary_path, exc) def finish(self, status: Status, finish: datetime | None) -> None: - """Finalize and publish after the whole experiment completes or fails.""" - raise NotImplementedError(f"Finalizing experiment output with status {status} at {finish} is not implemented") + self.experiment.status = status + self.experiment.finish = finish + self._update_timing(self.experiment) + self.write() + + @staticmethod + def _update_timing(record: Experiment | Run) -> None: + for field in ("start", "finish"): + value = getattr(record, field) + if value is not None: + setattr(record, field, value.astimezone(timezone.utc) if value.utcoffset() is not None else None) + end = record.finish + if end is None and record.status == "running": + end = datetime.now(timezone.utc) + if record.start is not None and end is not None: + record.duration = max((end - record.start).total_seconds(), 0.0) + + @staticmethod + def _aggregate_metrics(runs: list[Run]) -> list[Metric]: + groups: dict[tuple[str, str, tuple[tuple[str, str, str], ...]], list[Metric]] = defaultdict(list) + for run in runs: + if run.status != "completed": + continue + for metric in run.metrics: + point = tuple( + sorted((dimension.name, dimension.value, dimension.unit) for dimension in metric.dimensions) + ) + groups[(metric.name, metric.unit, point)].append(metric) + + metrics: list[Metric] = [] + for group in groups.values(): + metric = group[0].model_copy(deep=True) + values = [item.value for item in group] + if all(isinstance(value, (int, float)) and not isinstance(value, bool) for value in values): + metric.value = fmean(float(value) for value in values) + elif not all(type(value) is type(metric.value) and value == metric.value for value in values): + logging.warning("Cannot aggregate conflicting values for metric %s", metric.name) + continue + metrics.append(metric) + return metrics diff --git a/src/cloudai/systems/slurm/slurm_runner.py b/src/cloudai/systems/slurm/slurm_runner.py index 3499477fe..ae00baa09 100644 --- a/src/cloudai/systems/slurm/slurm_runner.py +++ b/src/cloudai/systems/slurm/slurm_runner.py @@ -96,7 +96,6 @@ def on_job_submit(self, tr: TestRun) -> None: cmd_gen.store_test_run() def get_run_output(self, job: BaseJob, tr: TestRun, result: JobStatusResult | None = None) -> "Run": - """Normalize CLI/REST metadata and canonical metrics without assuming per-test allocation timing.""" raise NotImplementedError def on_job_completion(self, job: BaseJob) -> None: diff --git a/src/cloudai/systems/standalone/standalone_runner.py b/src/cloudai/systems/standalone/standalone_runner.py index a25639c6b..211ae3d3b 100644 --- a/src/cloudai/systems/standalone/standalone_runner.py +++ b/src/cloudai/systems/standalone/standalone_runner.py @@ -40,7 +40,6 @@ def __init__(self, mode: str, system: System, test_scenario: TestScenario, outpu self.cmd_shell = CommandShell() def get_run_output(self, job: BaseJob, tr: TestRun, result: JobStatusResult | None = None) -> "Run": - """Normalize PID, process outcome, UTC timing, and canonical metrics; process tracking is pending.""" raise NotImplementedError def _submit_test(self, tr: TestRun) -> StandaloneJob: From 5c108994d1af8819bcc037e580f99335a9557e0c Mon Sep 17 00:00:00 2001 From: Ivan Podkidyshev Date: Wed, 16 Sep 2026 19:42:29 +0200 Subject: [PATCH 07/17] Use module imports in experiment output --- src/cloudai/output.py | 48 +++++++++++++++++++++++-------------------- 1 file changed, 26 insertions(+), 22 deletions(-) diff --git a/src/cloudai/output.py b/src/cloudai/output.py index d090d18c0..9addaee8e 100644 --- a/src/cloudai/output.py +++ b/src/cloudai/output.py @@ -14,24 +14,24 @@ # See the License for the specific language governing permissions and # limitations under the License. +import collections +import datetime import logging -from collections import defaultdict -from datetime import datetime, timezone -from pathlib import Path -from statistics import fmean -from tempfile import NamedTemporaryFile +import pathlib +import statistics +import tempfile -from cloudai.models.output import Experiment, ExperimentShort, Metric, Run, Status, Test +from cloudai.models import output as output_models class ExperimentOutput: """Collect experiment results and publish full and short snapshots.""" - def __init__(self, experiment: Experiment, output_path: Path) -> None: + def __init__(self, experiment: output_models.Experiment, output_path: pathlib.Path) -> None: self.experiment = experiment.model_copy(deep=True) self.output_path = output_path - def update_run(self, test_id: str, run: Run) -> None: + def update_run(self, test_id: str, run: output_models.Run) -> None: """Store the latest state of a run, identified by test and output path.""" test = next((test for test in self.experiment.tests if test.id == test_id), None) if test is None: @@ -43,7 +43,7 @@ def update_run(self, test_id: str, run: Run) -> None: return test.runs.append(recorded) - def update_test(self, test: Test) -> None: + def update_test(self, test: output_models.Test) -> None: """Update a test while retaining its previously recorded runs.""" recorded = test.model_copy(deep=True) for index, current in enumerate(self.experiment.tests): @@ -56,7 +56,7 @@ def update_test(self, test: Test) -> None: return self.experiment.tests.append(recorded) - def snapshot(self) -> tuple[ExperimentShort, Experiment]: + def snapshot(self) -> tuple[output_models.ExperimentShort, output_models.Experiment]: """Return independent short and full views without finalizing the experiment.""" full = self.experiment.model_copy(deep=True) self._update_timing(full) @@ -66,12 +66,12 @@ def snapshot(self) -> tuple[ExperimentShort, Experiment]: if test.metrics or test.dse is not None or any(run.step not in (None, 0) for run in test.runs): continue test.metrics = self._aggregate_metrics(test.runs) - short = ExperimentShort.model_validate(full.model_dump()) + short = output_models.ExperimentShort.model_validate(full.model_dump()) return short, full def write(self) -> None: """Atomically replace each output file, warning on failure.""" - pending: list[tuple[Path, Path]] = [] + pending: list[tuple[pathlib.Path, pathlib.Path]] = [] try: short, full = self.snapshot() contents = [ @@ -80,10 +80,10 @@ def write(self) -> None: ] self.output_path.mkdir(parents=True, exist_ok=True) for filename, content in contents: - with NamedTemporaryFile( + with tempfile.NamedTemporaryFile( mode="w", encoding="utf-8", dir=self.output_path, prefix=f".{filename}.", delete=False ) as temporary: - pending.append((Path(temporary.name), self.output_path / filename)) + pending.append((pathlib.Path(temporary.name), self.output_path / filename)) temporary.write(content + "\n") for temporary_path, destination in pending: temporary_path.replace(destination) @@ -96,27 +96,31 @@ def write(self) -> None: except OSError as exc: logging.warning("Cannot remove temporary experiment output %s: %s", temporary_path, exc) - def finish(self, status: Status, finish: datetime | None) -> None: + def finish(self, status: output_models.Status, finish: datetime.datetime | None) -> None: self.experiment.status = status self.experiment.finish = finish self._update_timing(self.experiment) self.write() @staticmethod - def _update_timing(record: Experiment | Run) -> None: + def _update_timing(record: output_models.Experiment | output_models.Run) -> None: for field in ("start", "finish"): value = getattr(record, field) if value is not None: - setattr(record, field, value.astimezone(timezone.utc) if value.utcoffset() is not None else None) + setattr( + record, field, value.astimezone(datetime.timezone.utc) if value.utcoffset() is not None else None + ) end = record.finish if end is None and record.status == "running": - end = datetime.now(timezone.utc) + end = datetime.datetime.now(datetime.timezone.utc) if record.start is not None and end is not None: record.duration = max((end - record.start).total_seconds(), 0.0) @staticmethod - def _aggregate_metrics(runs: list[Run]) -> list[Metric]: - groups: dict[tuple[str, str, tuple[tuple[str, str, str], ...]], list[Metric]] = defaultdict(list) + def _aggregate_metrics(runs: list[output_models.Run]) -> list[output_models.Metric]: + groups: dict[tuple[str, str, tuple[tuple[str, str, str], ...]], list[output_models.Metric]] = ( + collections.defaultdict(list) + ) for run in runs: if run.status != "completed": continue @@ -126,12 +130,12 @@ def _aggregate_metrics(runs: list[Run]) -> list[Metric]: ) groups[(metric.name, metric.unit, point)].append(metric) - metrics: list[Metric] = [] + metrics: list[output_models.Metric] = [] for group in groups.values(): metric = group[0].model_copy(deep=True) values = [item.value for item in group] if all(isinstance(value, (int, float)) and not isinstance(value, bool) for value in values): - metric.value = fmean(float(value) for value in values) + metric.value = statistics.fmean(float(value) for value in values) elif not all(type(value) is type(metric.value) and value == metric.value for value in values): logging.warning("Cannot aggregate conflicting values for metric %s", metric.name) continue From 453a0e33342f5a9ddad90a1145bf9e4465f1d1e0 Mon Sep 17 00:00:00 2001 From: Ivan Podkidyshev Date: Wed, 16 Sep 2026 19:55:21 +0200 Subject: [PATCH 08/17] Keep only full experiment output --- src/cloudai/models/output.py | 24 +++++------------------- src/cloudai/output.py | 35 ++++++++++++++--------------------- 2 files changed, 19 insertions(+), 40 deletions(-) diff --git a/src/cloudai/models/output.py b/src/cloudai/models/output.py index c3fd88aff..4d62e0233 100644 --- a/src/cloudai/models/output.py +++ b/src/cloudai/models/output.py @@ -62,8 +62,8 @@ class DSE(BaseModel): best_step: int | None = None -class TestShort(BaseModel): - """Test identity, status, and summary metrics.""" +class Test(BaseModel): + """A test case with all logical executions and optional DSE metadata.""" id: str name: str @@ -71,16 +71,13 @@ class TestShort(BaseModel): status: Status = "pending" path: str metrics: list[Metric] = Field(default_factory=list) - - -class Test(TestShort): - """A test case with all logical executions and optional DSE metadata.""" - runs: list[Run] = Field(default_factory=list) dse: DSE | None = None -class _ExperimentMetadata(BaseModel): +class Experiment(BaseModel): + """Full snapshot spanning the entire scenario, including all DSE trials.""" + id: str name: str description: str | None = None @@ -89,15 +86,4 @@ class _ExperimentMetadata(BaseModel): start: datetime | None = None finish: datetime | None = None duration: FiniteFloat | None = None - - -class ExperimentShort(_ExperimentMetadata): - """Catalog snapshot without individual runs or DSE details.""" - - tests: list[TestShort] = Field(default_factory=list) - - -class Experiment(_ExperimentMetadata): - """Full snapshot spanning the entire scenario, including all DSE trials.""" - tests: list[Test] = Field(default_factory=list) diff --git a/src/cloudai/output.py b/src/cloudai/output.py index 9addaee8e..469296a6e 100644 --- a/src/cloudai/output.py +++ b/src/cloudai/output.py @@ -25,7 +25,7 @@ class ExperimentOutput: - """Collect experiment results and publish full and short snapshots.""" + """Collect experiment results and publish snapshots.""" def __init__(self, experiment: output_models.Experiment, output_path: pathlib.Path) -> None: self.experiment = experiment.model_copy(deep=True) @@ -56,8 +56,8 @@ def update_test(self, test: output_models.Test) -> None: return self.experiment.tests.append(recorded) - def snapshot(self) -> tuple[output_models.ExperimentShort, output_models.Experiment]: - """Return independent short and full views without finalizing the experiment.""" + def snapshot(self) -> output_models.Experiment: + """Return an independent snapshot without finalizing the experiment.""" full = self.experiment.model_copy(deep=True) self._update_timing(full) for test in full.tests: @@ -66,31 +66,24 @@ def snapshot(self) -> tuple[output_models.ExperimentShort, output_models.Experim if test.metrics or test.dse is not None or any(run.step not in (None, 0) for run in test.runs): continue test.metrics = self._aggregate_metrics(test.runs) - short = output_models.ExperimentShort.model_validate(full.model_dump()) - return short, full + return full def write(self) -> None: - """Atomically replace each output file, warning on failure.""" - pending: list[tuple[pathlib.Path, pathlib.Path]] = [] + """Atomically replace experiment.json, warning on failure.""" + temporary_path: pathlib.Path | None = None try: - short, full = self.snapshot() - contents = [ - ("experiment.json", full.model_dump_json(indent=2)), - ("experiment-summary.json", short.model_dump_json(indent=2)), - ] + content = self.snapshot().model_dump_json(indent=2) self.output_path.mkdir(parents=True, exist_ok=True) - for filename, content in contents: - with tempfile.NamedTemporaryFile( - mode="w", encoding="utf-8", dir=self.output_path, prefix=f".{filename}.", delete=False - ) as temporary: - pending.append((pathlib.Path(temporary.name), self.output_path / filename)) - temporary.write(content + "\n") - for temporary_path, destination in pending: - temporary_path.replace(destination) + with tempfile.NamedTemporaryFile( + mode="w", encoding="utf-8", dir=self.output_path, prefix=".experiment.json.", delete=False + ) as temporary: + temporary_path = pathlib.Path(temporary.name) + temporary.write(content + "\n") + temporary_path.replace(self.output_path / "experiment.json") except Exception as exc: logging.warning("Cannot write experiment output: %s", exc) finally: - for temporary_path, _ in pending: + if temporary_path is not None: try: temporary_path.unlink(missing_ok=True) except OSError as exc: From bd12e57de59595133ef8178baa0015571c2ea2af Mon Sep 17 00:00:00 2001 From: Ivan Podkidyshev Date: Wed, 16 Sep 2026 20:02:50 +0200 Subject: [PATCH 09/17] Leave test metric summaries to callers --- src/cloudai/output.py | 31 ------------------------------- 1 file changed, 31 deletions(-) diff --git a/src/cloudai/output.py b/src/cloudai/output.py index 469296a6e..e134b608f 100644 --- a/src/cloudai/output.py +++ b/src/cloudai/output.py @@ -14,11 +14,9 @@ # See the License for the specific language governing permissions and # limitations under the License. -import collections import datetime import logging import pathlib -import statistics import tempfile from cloudai.models import output as output_models @@ -63,9 +61,6 @@ def snapshot(self) -> output_models.Experiment: for test in full.tests: for run in test.runs: self._update_timing(run) - if test.metrics or test.dse is not None or any(run.step not in (None, 0) for run in test.runs): - continue - test.metrics = self._aggregate_metrics(test.runs) return full def write(self) -> None: @@ -108,29 +103,3 @@ def _update_timing(record: output_models.Experiment | output_models.Run) -> None end = datetime.datetime.now(datetime.timezone.utc) if record.start is not None and end is not None: record.duration = max((end - record.start).total_seconds(), 0.0) - - @staticmethod - def _aggregate_metrics(runs: list[output_models.Run]) -> list[output_models.Metric]: - groups: dict[tuple[str, str, tuple[tuple[str, str, str], ...]], list[output_models.Metric]] = ( - collections.defaultdict(list) - ) - for run in runs: - if run.status != "completed": - continue - for metric in run.metrics: - point = tuple( - sorted((dimension.name, dimension.value, dimension.unit) for dimension in metric.dimensions) - ) - groups[(metric.name, metric.unit, point)].append(metric) - - metrics: list[output_models.Metric] = [] - for group in groups.values(): - metric = group[0].model_copy(deep=True) - values = [item.value for item in group] - if all(isinstance(value, (int, float)) and not isinstance(value, bool) for value in values): - metric.value = statistics.fmean(float(value) for value in values) - elif not all(type(value) is type(metric.value) and value == metric.value for value in values): - logging.warning("Cannot aggregate conflicting values for metric %s", metric.name) - continue - metrics.append(metric) - return metrics From fa1e90be233ae2fb4a0d6f8fef2b161af0ec29cf Mon Sep 17 00:00:00 2001 From: Ivan Podkidyshev Date: Wed, 16 Sep 2026 20:12:10 +0200 Subject: [PATCH 10/17] Generate experiment output for standalone runs --- src/cloudai/_core/base_runner.py | 6 +- src/cloudai/_core/runner.py | 4 +- src/cloudai/cli/handlers.py | 14 ++- src/cloudai/output.py | 18 ++++ .../systems/standalone/standalone_job.py | 8 +- .../systems/standalone/standalone_runner.py | 98 +++++++++++++++++-- .../systems/standalone/standalone_system.py | 5 + 7 files changed, 135 insertions(+), 18 deletions(-) diff --git a/src/cloudai/_core/base_runner.py b/src/cloudai/_core/base_runner.py index cda5370ad..d22027b16 100644 --- a/src/cloudai/_core/base_runner.py +++ b/src/cloudai/_core/base_runner.py @@ -14,6 +14,7 @@ # See the License for the specific language governing permissions and # limitations under the License. +import datetime import logging import time from abc import ABC, abstractmethod @@ -97,9 +98,10 @@ def write_output(self) -> None: if self.mode == "run" and self.experiment_output is not None: self.experiment_output.write() - def finish_output(self) -> None: + def finish_output(self, successful: bool) -> None: if self.mode == "run" and self.experiment_output is not None: - self.experiment_output.finish(status="unknown", finish=None) + status = "completed" if successful else "failed" + self.experiment_output.finish(status=status, finish=datetime.datetime.now(datetime.timezone.utc)) def shutdown(self): """Gracefully shut down the runner, terminating all outstanding jobs.""" diff --git a/src/cloudai/_core/runner.py b/src/cloudai/_core/runner.py index b6647a895..ee636dd30 100644 --- a/src/cloudai/_core/runner.py +++ b/src/cloudai/_core/runner.py @@ -79,13 +79,15 @@ def create_runner(self, mode: str, system: System, test_scenario: TestScenario) return runner_class(mode, system, test_scenario, results_root) - def run(self): + def run(self) -> bool: """Run the test scenario using the instantiated runner.""" try: self.runner.run() logging.debug("All jobs finished successfully.") + return True except JobFailureError as exc: logging.debug(f"Runner failed JobFailure exception: {exc}", exc_info=True) + return False def cancel_on_signal( self, diff --git a/src/cloudai/cli/handlers.py b/src/cloudai/cli/handlers.py index 91984add4..e0d65d647 100644 --- a/src/cloudai/cli/handlers.py +++ b/src/cloudai/cli/handlers.py @@ -239,10 +239,11 @@ def generate_reports( logging.debug(e, exc_info=True) -def handle_non_dse_job(runner: Runner, args: argparse.Namespace) -> None: - runner.run() +def handle_non_dse_job(runner: Runner, args: argparse.Namespace) -> bool: + successful = runner.run() generate_reports(runner.runner.system, runner.runner.test_scenario, runner.runner.scenario_root) logging.info("All jobs are complete.") + return successful def register_signal_handlers(signal_handler: Callable) -> None: @@ -352,20 +353,23 @@ def handle_dry_run_and_run(args: argparse.Namespace) -> int: register_signal_handlers(runner.cancel_on_signal) logging.info(f"Scenario results will be stored at: {runner.runner.scenario_root}") + successful = False try: runner.runner.write_output() has_dse = any(tr.is_dse_job for tr in test_scenario.test_runs) if args.single_sbatch or not has_dse: # in this mode cases are unrolled using grid search - handle_non_dse_job(runner, args) + successful = handle_non_dse_job(runner, args) return 0 if all(tr.is_dse_job for tr in test_scenario.test_runs): - return handle_dse_job(runner, args) + result = handle_dse_job(runner, args) + successful = result == 0 + return result logging.error("Mixing DSE and non-DSE jobs is not allowed.") return 1 finally: - runner.runner.finish_output() + runner.runner.finish_output(successful) def handle_generate_report(args: argparse.Namespace) -> int: diff --git a/src/cloudai/output.py b/src/cloudai/output.py index e134b608f..c9f44b273 100644 --- a/src/cloudai/output.py +++ b/src/cloudai/output.py @@ -38,8 +38,10 @@ def update_run(self, test_id: str, run: output_models.Run) -> None: for index, current in enumerate(test.runs): if current.path == run.path: test.runs[index] = recorded + self._update_test_status(test) return test.runs.append(recorded) + self._update_test_status(test) def update_test(self, test: output_models.Test) -> None: """Update a test while retaining its previously recorded runs.""" @@ -87,9 +89,25 @@ def write(self) -> None: def finish(self, status: output_models.Status, finish: datetime.datetime | None) -> None: self.experiment.status = status self.experiment.finish = finish + if status == "completed": + for test in self.experiment.tests: + if test.status not in ("failed", "cancelled"): + test.status = "completed" self._update_timing(self.experiment) self.write() + @staticmethod + def _update_test_status(test: output_models.Test) -> None: + statuses = {run.status for run in test.runs} + if "failed" in statuses: + test.status = "failed" + elif "cancelled" in statuses: + test.status = "cancelled" + elif "running" in statuses: + test.status = "running" + elif statuses == {"completed"}: + test.status = "completed" + @staticmethod def _update_timing(record: output_models.Experiment | output_models.Run) -> None: for field in ("start", "finish"): diff --git a/src/cloudai/systems/standalone/standalone_job.py b/src/cloudai/systems/standalone/standalone_job.py index d57befb4b..0e68be19f 100644 --- a/src/cloudai/systems/standalone/standalone_job.py +++ b/src/cloudai/systems/standalone/standalone_job.py @@ -14,7 +14,9 @@ # See the License for the specific language governing permissions and # limitations under the License. -from dataclasses import dataclass +import datetime +import subprocess +from dataclasses import dataclass, field from cloudai.core import BaseJob @@ -23,4 +25,6 @@ class StandaloneJob(BaseJob): """A job class for standalone execution.""" - pass + process: subprocess.Popen[str] | None = field(default=None, repr=False) + start: datetime.datetime | None = None + finish: datetime.datetime | None = None diff --git a/src/cloudai/systems/standalone/standalone_runner.py b/src/cloudai/systems/standalone/standalone_runner.py index 211ae3d3b..3042f9ecf 100644 --- a/src/cloudai/systems/standalone/standalone_runner.py +++ b/src/cloudai/systems/standalone/standalone_runner.py @@ -14,18 +14,19 @@ # See the License for the specific language governing permissions and # limitations under the License. +import datetime import logging from pathlib import Path -from typing import TYPE_CHECKING +from typing import cast +import cloudai.metrics +from cloudai import output from cloudai.core import BaseJob, BaseRunner, JobIdRetrievalError, JobStatusResult, System, TestRun, TestScenario +from cloudai.models import output as output_models from cloudai.util import CommandShell from .standalone_job import StandaloneJob -if TYPE_CHECKING: - from cloudai.models.output import Run - class StandaloneRunner(BaseRunner): """ @@ -39,8 +40,85 @@ def __init__(self, mode: str, system: System, test_scenario: TestScenario, outpu super().__init__(mode, system, test_scenario, output_path) self.cmd_shell = CommandShell() - def get_run_output(self, job: BaseJob, tr: TestRun, result: JobStatusResult | None = None) -> "Run": - raise NotImplementedError + def create_experiment_output(self) -> output.ExperimentOutput | None: + if self.mode != "run": + return None + output_path = self.scenario_root.absolute() + experiment = output_models.Experiment( + id=output_path.name, + name=self.test_scenario.name, + status="running", + path=str(output_path), + start=datetime.datetime.now(datetime.timezone.utc), + tests=[ + output_models.Test( + id=str(tr.name), + name=tr.test.name, + description=tr.test.description, + path=str(output_path / str(tr.name)), + ) + for tr in self.test_scenario.test_runs + ], + ) + return output.ExperimentOutput(experiment, output_path) + + def get_run_output(self, job: BaseJob, tr: TestRun, result: JobStatusResult | None = None) -> output_models.Run: + standalone_job = cast(StandaloneJob, job) + status: output_models.Status = "running" + metrics: list[output_models.Metric] = [] + if result is not None: + status = "completed" if result.is_successful else "failed" + if job.terminated_by_dependency: + status = "cancelled" + if result.is_successful: + try: + observations = tr.test.metric_observations(self.system, tr) + metrics = [self._metric_output(observation) for observation in observations] + except Exception as exc: + logging.warning("Cannot extract output metrics for standalone job %s: %s", job.id, exc) + return output_models.Run( + path=str(tr.output_path.absolute()), + jobid=str(job.id), + status=status, + metrics=metrics, + start=standalone_job.start, + finish=standalone_job.finish, + iteration=tr.current_iteration, + step=tr.step, + ) + + def get_runner_job_status(self, job: BaseJob) -> JobStatusResult: + standalone_job = cast(StandaloneJob, job) + if standalone_job.terminated_by_dependency: + return JobStatusResult(is_successful=True) + if standalone_job.process is None: + return JobStatusResult(is_successful=True) + return_code = standalone_job.process.poll() + if return_code == 0: + return JobStatusResult(is_successful=True) + return JobStatusResult(is_successful=False, error_message=f"Process exited with status {return_code}") + + def on_job_completion(self, job: BaseJob) -> None: + standalone_job = cast(StandaloneJob, job) + standalone_job.finish = datetime.datetime.now(datetime.timezone.utc) + if standalone_job.process is not None: + standalone_job.process.communicate() + + @staticmethod + def _metric_output(observation: cloudai.metrics.MetricObservation) -> output_models.Metric: + dimensions = [ + output_models.Dimension( + name=cloudai.metrics.dimension_label(key), + value=str(value), + ) + for key, value in sorted(observation.dimensions.items()) + ] + return output_models.Metric( + name=observation.metric.display_name, + value=observation.value, + unit=observation.metric.unit, + dimensions=dimensions, + ) def _submit_test(self, tr: TestRun) -> StandaloneJob: logging.info(f"Running test: {tr.name}") @@ -48,11 +126,15 @@ def _submit_test(self, tr: TestRun) -> StandaloneJob: exec_cmd = self.get_cmd_gen_strategy(self.system, tr).gen_exec_command() logging.info(f"Executing command for test {tr.name}: {exec_cmd}") job_id = 0 + process = None + start = None if self.mode == "run": - pid = self.cmd_shell.execute(exec_cmd).pid + start = datetime.datetime.now(datetime.timezone.utc) + process = self.cmd_shell.execute(exec_cmd) + pid = process.pid job_id = pid if job_id is None: raise JobIdRetrievalError( test_name=str(tr.name), command=exec_cmd, stdout="", stderr="", message="Failed to retrieve job ID." ) - return StandaloneJob(tr, id=job_id) + return StandaloneJob(tr, id=job_id, process=process, start=start) diff --git a/src/cloudai/systems/standalone/standalone_system.py b/src/cloudai/systems/standalone/standalone_system.py index 57d99487d..ccdb20516 100644 --- a/src/cloudai/systems/standalone/standalone_system.py +++ b/src/cloudai/systems/standalone/standalone_system.py @@ -19,6 +19,8 @@ from cloudai.core import BaseJob, System from cloudai.util import CommandShell +from .standalone_job import StandaloneJob + class StandaloneSystem(System): """ @@ -49,6 +51,9 @@ def is_job_running(self, job: BaseJob) -> bool: Returns: bool: True if the job is running, False otherwise. """ + if isinstance(job, StandaloneJob) and job.process is not None: + return job.process.poll() is None + command = f"ps -p {job.id}" logging.debug(f"Checking job status with command: {command}") stdout = self.cmd_shell.execute(command).communicate()[0] From 19893a8df4110a1aaa47ad6a2f02970f2c89f98d Mon Sep 17 00:00:00 2001 From: Ivan Podkidyshev Date: Thu, 17 Sep 2026 13:37:00 +0200 Subject: [PATCH 11/17] Create experiment output for all runners --- src/cloudai/_core/base_runner.py | 44 ++++++++++++++----- src/cloudai/systems/slurm/slurm_runner.py | 10 +---- .../systems/standalone/standalone_job.py | 4 +- .../systems/standalone/standalone_runner.py | 42 +----------------- .../systems/standalone/standalone_system.py | 5 --- 5 files changed, 38 insertions(+), 67 deletions(-) diff --git a/src/cloudai/_core/base_runner.py b/src/cloudai/_core/base_runner.py index d22027b16..085d66dc6 100644 --- a/src/cloudai/_core/base_runner.py +++ b/src/cloudai/_core/base_runner.py @@ -19,11 +19,10 @@ import time from abc import ABC, abstractmethod from pathlib import Path -from typing import TYPE_CHECKING, Dict, List +from typing import Dict, List -if TYPE_CHECKING: - from cloudai.models.output import Run - from cloudai.output import ExperimentOutput +from cloudai.models import output as output_models +from cloudai.output import ExperimentOutput from .base_job import BaseJob from .command_gen_strategy import CommandGenStrategy @@ -74,14 +73,35 @@ def __init__(self, mode: str, system: System, test_scenario: TestScenario, outpu self.testrun_to_job_map: Dict[TestRun, BaseJob] = {} logging.debug(f"{self.__class__.__name__} initialized") self.shutting_down = False - self.experiment_output: ExperimentOutput | None = self.create_experiment_output() - - def create_experiment_output(self) -> "ExperimentOutput | None": - return None + self.experiment_output = self.create_experiment_output() + + def create_experiment_output(self) -> ExperimentOutput | None: + if self.mode != "run": + return None + output_path = self.scenario_root.absolute() + experiment = output_models.Experiment( + id=output_path.name, + name=self.test_scenario.name, + status="running", + path=str(output_path), + start=datetime.datetime.now(datetime.timezone.utc), + tests=[ + output_models.Test( + id=str(tr.name), + name=tr.test.name, + description=tr.test.description, + path=str(output_path / str(tr.name)), + ) + for tr in self.test_scenario.test_runs + ], + ) + return ExperimentOutput(experiment, output_path) - def get_run_output(self, job: BaseJob, tr: TestRun, result: JobStatusResult | None = None) -> "Run": + def get_run_output( + self, job: BaseJob, tr: TestRun, result: JobStatusResult | None = None + ) -> output_models.Run | None: """Normalize one logical execution; result is absent at submission.""" - raise NotImplementedError + return None def completed_test_runs(self, job: BaseJob) -> list[TestRun]: """Return logical executions represented by a scheduler job.""" @@ -91,7 +111,9 @@ def update_run_output(self, job: BaseJob, result: JobStatusResult | None = None) """Capture logical runs before iteration or DSE state advances, then publish a snapshot.""" if self.mode == "run" and self.experiment_output is not None: for tr in self.completed_test_runs(job): - self.experiment_output.update_run(str(tr.name), self.get_run_output(job, tr, result)) + run = self.get_run_output(job, tr, result) + if run is not None: + self.experiment_output.update_run(str(tr.name), run) self.write_output() def write_output(self) -> None: diff --git a/src/cloudai/systems/slurm/slurm_runner.py b/src/cloudai/systems/slurm/slurm_runner.py index ae00baa09..3f09f9523 100644 --- a/src/cloudai/systems/slurm/slurm_runner.py +++ b/src/cloudai/systems/slurm/slurm_runner.py @@ -17,11 +17,11 @@ import logging import re from pathlib import Path -from typing import TYPE_CHECKING, cast +from typing import cast import toml -from cloudai.core import BaseJob, BaseRunner, JobIdRetrievalError, JobStatusResult, System, TestRun, TestScenario +from cloudai.core import BaseJob, BaseRunner, JobIdRetrievalError, System, TestRun, TestScenario from cloudai.util import CommandShell from .slurm_command_gen_strategy import SlurmCommandGenStrategy @@ -29,9 +29,6 @@ from .slurm_metadata import SlurmJobMetadata, SlurmStepMetadata from .slurm_system import SlurmSystem -if TYPE_CHECKING: - from cloudai.models.output import Run - class SlurmRunner(BaseRunner): """ @@ -95,9 +92,6 @@ def on_job_submit(self, tr: TestRun) -> None: cmd_gen = self.get_cmd_gen_strategy(self.system, tr) cmd_gen.store_test_run() - def get_run_output(self, job: BaseJob, tr: TestRun, result: JobStatusResult | None = None) -> "Run": - raise NotImplementedError - def on_job_completion(self, job: BaseJob) -> None: logging.debug(f"Job completion callback for job {job.id}") slurm_job = cast(SlurmJob, job) diff --git a/src/cloudai/systems/standalone/standalone_job.py b/src/cloudai/systems/standalone/standalone_job.py index 0e68be19f..b30a130c9 100644 --- a/src/cloudai/systems/standalone/standalone_job.py +++ b/src/cloudai/systems/standalone/standalone_job.py @@ -15,8 +15,7 @@ # limitations under the License. import datetime -import subprocess -from dataclasses import dataclass, field +from dataclasses import dataclass from cloudai.core import BaseJob @@ -25,6 +24,5 @@ class StandaloneJob(BaseJob): """A job class for standalone execution.""" - process: subprocess.Popen[str] | None = field(default=None, repr=False) start: datetime.datetime | None = None finish: datetime.datetime | None = None diff --git a/src/cloudai/systems/standalone/standalone_runner.py b/src/cloudai/systems/standalone/standalone_runner.py index 3042f9ecf..9f08aa58b 100644 --- a/src/cloudai/systems/standalone/standalone_runner.py +++ b/src/cloudai/systems/standalone/standalone_runner.py @@ -20,7 +20,6 @@ from typing import cast import cloudai.metrics -from cloudai import output from cloudai.core import BaseJob, BaseRunner, JobIdRetrievalError, JobStatusResult, System, TestRun, TestScenario from cloudai.models import output as output_models from cloudai.util import CommandShell @@ -40,28 +39,6 @@ def __init__(self, mode: str, system: System, test_scenario: TestScenario, outpu super().__init__(mode, system, test_scenario, output_path) self.cmd_shell = CommandShell() - def create_experiment_output(self) -> output.ExperimentOutput | None: - if self.mode != "run": - return None - output_path = self.scenario_root.absolute() - experiment = output_models.Experiment( - id=output_path.name, - name=self.test_scenario.name, - status="running", - path=str(output_path), - start=datetime.datetime.now(datetime.timezone.utc), - tests=[ - output_models.Test( - id=str(tr.name), - name=tr.test.name, - description=tr.test.description, - path=str(output_path / str(tr.name)), - ) - for tr in self.test_scenario.test_runs - ], - ) - return output.ExperimentOutput(experiment, output_path) - def get_run_output(self, job: BaseJob, tr: TestRun, result: JobStatusResult | None = None) -> output_models.Run: standalone_job = cast(StandaloneJob, job) status: output_models.Status = "running" @@ -87,22 +64,9 @@ def get_run_output(self, job: BaseJob, tr: TestRun, result: JobStatusResult | No step=tr.step, ) - def get_runner_job_status(self, job: BaseJob) -> JobStatusResult: - standalone_job = cast(StandaloneJob, job) - if standalone_job.terminated_by_dependency: - return JobStatusResult(is_successful=True) - if standalone_job.process is None: - return JobStatusResult(is_successful=True) - return_code = standalone_job.process.poll() - if return_code == 0: - return JobStatusResult(is_successful=True) - return JobStatusResult(is_successful=False, error_message=f"Process exited with status {return_code}") - def on_job_completion(self, job: BaseJob) -> None: standalone_job = cast(StandaloneJob, job) standalone_job.finish = datetime.datetime.now(datetime.timezone.utc) - if standalone_job.process is not None: - standalone_job.process.communicate() @staticmethod def _metric_output(observation: cloudai.metrics.MetricObservation) -> output_models.Metric: @@ -126,15 +90,13 @@ def _submit_test(self, tr: TestRun) -> StandaloneJob: exec_cmd = self.get_cmd_gen_strategy(self.system, tr).gen_exec_command() logging.info(f"Executing command for test {tr.name}: {exec_cmd}") job_id = 0 - process = None start = None if self.mode == "run": start = datetime.datetime.now(datetime.timezone.utc) - process = self.cmd_shell.execute(exec_cmd) - pid = process.pid + pid = self.cmd_shell.execute(exec_cmd).pid job_id = pid if job_id is None: raise JobIdRetrievalError( test_name=str(tr.name), command=exec_cmd, stdout="", stderr="", message="Failed to retrieve job ID." ) - return StandaloneJob(tr, id=job_id, process=process, start=start) + return StandaloneJob(tr, id=job_id, start=start) diff --git a/src/cloudai/systems/standalone/standalone_system.py b/src/cloudai/systems/standalone/standalone_system.py index ccdb20516..57d99487d 100644 --- a/src/cloudai/systems/standalone/standalone_system.py +++ b/src/cloudai/systems/standalone/standalone_system.py @@ -19,8 +19,6 @@ from cloudai.core import BaseJob, System from cloudai.util import CommandShell -from .standalone_job import StandaloneJob - class StandaloneSystem(System): """ @@ -51,9 +49,6 @@ def is_job_running(self, job: BaseJob) -> bool: Returns: bool: True if the job is running, False otherwise. """ - if isinstance(job, StandaloneJob) and job.process is not None: - return job.process.poll() is None - command = f"ps -p {job.id}" logging.debug(f"Checking job status with command: {command}") stdout = self.cmd_shell.execute(command).communicate()[0] From 8f111478af352900c06d30b94197fdd28cbccf88 Mon Sep 17 00:00:00 2001 From: Ivan Podkidyshev Date: Thu, 17 Sep 2026 13:41:59 +0200 Subject: [PATCH 12/17] Write experiment output for dry runs --- src/cloudai/_core/base_runner.py | 28 +++++++++++----------------- 1 file changed, 11 insertions(+), 17 deletions(-) diff --git a/src/cloudai/_core/base_runner.py b/src/cloudai/_core/base_runner.py index 085d66dc6..f53746a67 100644 --- a/src/cloudai/_core/base_runner.py +++ b/src/cloudai/_core/base_runner.py @@ -73,11 +73,6 @@ def __init__(self, mode: str, system: System, test_scenario: TestScenario, outpu self.testrun_to_job_map: Dict[TestRun, BaseJob] = {} logging.debug(f"{self.__class__.__name__} initialized") self.shutting_down = False - self.experiment_output = self.create_experiment_output() - - def create_experiment_output(self) -> ExperimentOutput | None: - if self.mode != "run": - return None output_path = self.scenario_root.absolute() experiment = output_models.Experiment( id=output_path.name, @@ -95,7 +90,7 @@ def create_experiment_output(self) -> ExperimentOutput | None: for tr in self.test_scenario.test_runs ], ) - return ExperimentOutput(experiment, output_path) + self.experiment_output = ExperimentOutput(experiment, output_path) def get_run_output( self, job: BaseJob, tr: TestRun, result: JobStatusResult | None = None @@ -109,21 +104,20 @@ def completed_test_runs(self, job: BaseJob) -> list[TestRun]: def update_run_output(self, job: BaseJob, result: JobStatusResult | None = None) -> None: """Capture logical runs before iteration or DSE state advances, then publish a snapshot.""" - if self.mode == "run" and self.experiment_output is not None: - for tr in self.completed_test_runs(job): - run = self.get_run_output(job, tr, result) - if run is not None: - self.experiment_output.update_run(str(tr.name), run) - self.write_output() + if self.mode != "run": + return + for tr in self.completed_test_runs(job): + run = self.get_run_output(job, tr, result) + if run is not None: + self.experiment_output.update_run(str(tr.name), run) + self.write_output() def write_output(self) -> None: - if self.mode == "run" and self.experiment_output is not None: - self.experiment_output.write() + self.experiment_output.write() def finish_output(self, successful: bool) -> None: - if self.mode == "run" and self.experiment_output is not None: - status = "completed" if successful else "failed" - self.experiment_output.finish(status=status, finish=datetime.datetime.now(datetime.timezone.utc)) + status = "completed" if successful else "failed" + self.experiment_output.finish(status=status, finish=datetime.datetime.now(datetime.timezone.utc)) def shutdown(self): """Gracefully shut down the runner, terminating all outstanding jobs.""" From f20fa358b12c63aa738fff74e09b16d0b36c4ec3 Mon Sep 17 00:00:00 2001 From: Ivan Podkidyshev Date: Thu, 17 Sep 2026 13:45:24 +0200 Subject: [PATCH 13/17] Remove experiment output write wrapper --- src/cloudai/_core/base_runner.py | 3 --- src/cloudai/cli/handlers.py | 2 +- 2 files changed, 1 insertion(+), 4 deletions(-) diff --git a/src/cloudai/_core/base_runner.py b/src/cloudai/_core/base_runner.py index f53746a67..d0c15eeb3 100644 --- a/src/cloudai/_core/base_runner.py +++ b/src/cloudai/_core/base_runner.py @@ -110,9 +110,6 @@ def update_run_output(self, job: BaseJob, result: JobStatusResult | None = None) run = self.get_run_output(job, tr, result) if run is not None: self.experiment_output.update_run(str(tr.name), run) - self.write_output() - - def write_output(self) -> None: self.experiment_output.write() def finish_output(self, successful: bool) -> None: diff --git a/src/cloudai/cli/handlers.py b/src/cloudai/cli/handlers.py index e0d65d647..313cbbed2 100644 --- a/src/cloudai/cli/handlers.py +++ b/src/cloudai/cli/handlers.py @@ -355,7 +355,7 @@ def handle_dry_run_and_run(args: argparse.Namespace) -> int: successful = False try: - runner.runner.write_output() + runner.runner.experiment_output.write() has_dse = any(tr.is_dse_job for tr in test_scenario.test_runs) if args.single_sbatch or not has_dse: # in this mode cases are unrolled using grid search successful = handle_non_dse_job(runner, args) From 4f0be98a4b577ecd763341a674edf01c6b5baa1f Mon Sep 17 00:00:00 2001 From: Ivan Podkidyshev Date: Thu, 17 Sep 2026 14:21:11 +0200 Subject: [PATCH 14/17] Validate unified output in dry runs --- tests/test_acceptance.py | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/tests/test_acceptance.py b/tests/test_acceptance.py index 080799d77..0e5ebd485 100644 --- a/tests/test_acceptance.py +++ b/tests/test_acceptance.py @@ -28,6 +28,7 @@ from cloudai.cli import setup_logging from cloudai.cli.handlers import handle_dry_run_and_run from cloudai.core import CommandGenStrategy, GitRepo, TestDefinition, TestRun, TestScenario +from cloudai.models.output import Experiment from cloudai.models.scenario import TestRunDetails from cloudai.systems.slurm import SlurmCommandGenStrategy, SlurmRunner, SlurmSystem from cloudai.workloads.ai_dynamo import ( @@ -170,6 +171,22 @@ def test_details_is_dumped_and_valid(self, do_dry_run: tuple[Path, dict]) -> Non for details_toml in details_tomls: TestRunDetails.model_validate(toml.load(details_toml)) + def test_experiment_output_is_dumped_and_valid(self, do_dry_run: tuple[Path, dict]) -> None: + tmp_path, scenario = do_dry_run + results_output = next(path for path in tmp_path.iterdir() if path.is_dir()) + experiment = Experiment.model_validate_json((results_output / "experiment.json").read_text()) + + assert experiment.id == results_output.name + assert experiment.name == toml.load(scenario["path"])["name"] + assert experiment.status == "completed" + assert experiment.path == str(results_output.absolute()) + assert experiment.start is not None + assert experiment.finish is not None + assert experiment.duration is not None + assert len(experiment.tests) == scenario["expected_dirs_number"] + assert all(test.status == "completed" for test in experiment.tests) + assert all(not test.runs for test in experiment.tests) + @pytest.fixture def partial_tr(slurm_system: SlurmSystem) -> partial[TestRun]: From 6b12ebd2bb8276d50e1fd93411200ff75630bd23 Mon Sep 17 00:00:00 2001 From: Ivan Podkidyshev Date: Thu, 17 Sep 2026 14:56:40 +0200 Subject: [PATCH 15/17] Test unified experiment output --- tests/systems/standalone/test_runner.py | 89 +++++++++++++++++++++++++ tests/test_output.py | 67 +++++++++++++++++++ 2 files changed, 156 insertions(+) create mode 100644 tests/systems/standalone/test_runner.py create mode 100644 tests/test_output.py diff --git a/tests/systems/standalone/test_runner.py b/tests/systems/standalone/test_runner.py new file mode 100644 index 000000000..aed4462cf --- /dev/null +++ b/tests/systems/standalone/test_runner.py @@ -0,0 +1,89 @@ +# SPDX-FileCopyrightText: NVIDIA CORPORATION & AFFILIATES +# Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import datetime +import pathlib + +import pytest + +import cloudai.metrics +from cloudai.core import JobStatusResult, System, TestDefinition, TestRun, TestScenario +from cloudai.models.workload import CmdArgs +from cloudai.systems.standalone import StandaloneJob, StandaloneRunner, StandaloneSystem + + +class MetricWorkload(TestDefinition): + successful: bool + + def was_run_successful(self, tr: TestRun) -> JobStatusResult: + if self.successful: + return JobStatusResult(is_successful=True) + return JobStatusResult(is_successful=False, error_message="workload result failed") + + def metric_observations(self, system: System, tr: TestRun) -> list[cloudai.metrics.MetricObservation]: + return [ + cloudai.metrics.MetricObservation( + metric=cloudai.metrics.BANDWIDTH, + value=12.5, + dimensions={"size_bytes": 1024}, + ) + ] + + +@pytest.mark.parametrize("successful", [True, False]) +def test_standalone_run_output_uses_workload_status_and_metrics( + tmp_path: pathlib.Path, standalone_system: StandaloneSystem, successful: bool +) -> None: + workload = MetricWorkload( + name="metric-workload", + description="metric workload", + test_template_name="MetricWorkload", + cmd_args=CmdArgs(), + successful=successful, + ) + test_run = TestRun( + name="case", + test=workload, + num_nodes=1, + nodes=[], + output_path=tmp_path / "case" / "0", + ) + runner = StandaloneRunner("run", standalone_system, TestScenario(name="scenario", test_runs=[test_run]), tmp_path) + start = datetime.datetime(2026, 1, 2, 3, 4, 5, tzinfo=datetime.timezone.utc) + job = StandaloneJob( + test_run, + id=123, + start=start, + finish=start + datetime.timedelta(seconds=3), + ) + + run = runner.get_run_output(job, test_run, runner.get_job_status(job)) + + assert run.status == ("completed" if successful else "failed") + assert run.jobid == "123" + assert run.start == start + assert run.finish == start + datetime.timedelta(seconds=3) + if successful: + assert [metric.model_dump() for metric in run.metrics] == [ + { + "name": "Bandwidth", + "value": 12.5, + "unit": "GB/s", + "dimensions": [{"name": "Size", "value": "1024", "unit": "", "is_x": False}], + } + ] + else: + assert run.metrics == [] diff --git a/tests/test_output.py b/tests/test_output.py new file mode 100644 index 000000000..4c76de1d4 --- /dev/null +++ b/tests/test_output.py @@ -0,0 +1,67 @@ +# SPDX-FileCopyrightText: NVIDIA CORPORATION & AFFILIATES +# Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import datetime +import pathlib + +from cloudai.models import output as output_models +from cloudai.output import ExperimentOutput + + +def test_experiment_output_preserves_runs_and_finalizes_failure(tmp_path: pathlib.Path) -> None: + start = datetime.datetime(2026, 1, 2, 3, 4, 5, tzinfo=datetime.timezone.utc) + experiment = output_models.Experiment( + id="experiment", + name="scenario", + status="running", + path=str(tmp_path), + start=start, + tests=[output_models.Test(id="case", name="workload", path=str(tmp_path / "case"))], + ) + experiment_output = ExperimentOutput(experiment, tmp_path) + first_run = output_models.Run( + path=str(tmp_path / "case" / "0"), + jobid="101", + status="completed", + start=start, + finish=start + datetime.timedelta(seconds=2), + iteration=0, + step=0, + ) + second_run = output_models.Run( + path=str(tmp_path / "case" / "1"), + jobid="102", + status="running", + start=start + datetime.timedelta(seconds=2), + iteration=1, + step=0, + ) + + experiment_output.update_run("case", first_run) + experiment_output.update_run("case", second_run) + second_run.status = "failed" + second_run.finish = start + datetime.timedelta(seconds=4) + experiment_output.update_run("case", second_run) + experiment_output.finish("failed", start + datetime.timedelta(seconds=5)) + + stored = output_models.Experiment.model_validate_json((tmp_path / "experiment.json").read_text()) + assert stored.status == "failed" + assert stored.duration == 5 + assert stored.tests[0].status == "failed" + assert [(run.jobid, run.status, run.duration) for run in stored.tests[0].runs] == [ + ("101", "completed", 2), + ("102", "failed", 2), + ] From 955224d7e152eb3fa20eed70fc6d08b5defd4b6f Mon Sep 17 00:00:00 2001 From: Ivan Podkidyshev Date: Thu, 17 Sep 2026 16:30:10 +0200 Subject: [PATCH 16/17] Update standalone copyright years --- src/cloudai/systems/standalone/standalone_job.py | 2 +- src/cloudai/systems/standalone/standalone_runner.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/src/cloudai/systems/standalone/standalone_job.py b/src/cloudai/systems/standalone/standalone_job.py index b30a130c9..66e3a3d63 100644 --- a/src/cloudai/systems/standalone/standalone_job.py +++ b/src/cloudai/systems/standalone/standalone_job.py @@ -1,5 +1,5 @@ # SPDX-FileCopyrightText: NVIDIA CORPORATION & AFFILIATES -# Copyright (c) 2024-2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# Copyright (c) 2024-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 # # Licensed under the Apache License, Version 2.0 (the "License"); diff --git a/src/cloudai/systems/standalone/standalone_runner.py b/src/cloudai/systems/standalone/standalone_runner.py index 9f08aa58b..1cb2af22f 100644 --- a/src/cloudai/systems/standalone/standalone_runner.py +++ b/src/cloudai/systems/standalone/standalone_runner.py @@ -1,5 +1,5 @@ # SPDX-FileCopyrightText: NVIDIA CORPORATION & AFFILIATES -# Copyright (c) 2024-2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# Copyright (c) 2024-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 # # Licensed under the Apache License, Version 2.0 (the "License"); From 7dd755b6a528e22d0cf57d9a87762bc619ca6818 Mon Sep 17 00:00:00 2001 From: Ivan Podkidyshev Date: Thu, 17 Sep 2026 17:57:31 +0200 Subject: [PATCH 17/17] Compare complete experiment output in tests Signed-off-by: Ivan Podkidyshev --- tests/systems/standalone/test_runner.py | 52 ++++++++++++------------- tests/test_acceptance.py | 51 ++++++++++++++++++------ tests/test_output.py | 52 +++++++++++++++++++++---- 3 files changed, 110 insertions(+), 45 deletions(-) diff --git a/tests/systems/standalone/test_runner.py b/tests/systems/standalone/test_runner.py index aed4462cf..33a7ef994 100644 --- a/tests/systems/standalone/test_runner.py +++ b/tests/systems/standalone/test_runner.py @@ -17,23 +17,18 @@ import datetime import pathlib -import pytest - +import cloudai.core import cloudai.metrics -from cloudai.core import JobStatusResult, System, TestDefinition, TestRun, TestScenario -from cloudai.models.workload import CmdArgs from cloudai.systems.standalone import StandaloneJob, StandaloneRunner, StandaloneSystem -class MetricWorkload(TestDefinition): - successful: bool - - def was_run_successful(self, tr: TestRun) -> JobStatusResult: - if self.successful: - return JobStatusResult(is_successful=True) - return JobStatusResult(is_successful=False, error_message="workload result failed") +class MetricWorkload(cloudai.core.TestDefinition): + def was_run_successful(self, tr: cloudai.core.TestRun) -> cloudai.core.JobStatusResult: + return cloudai.core.JobStatusResult(is_successful=True) - def metric_observations(self, system: System, tr: TestRun) -> list[cloudai.metrics.MetricObservation]: + def metric_observations( + self, system: cloudai.core.System, tr: cloudai.core.TestRun + ) -> list[cloudai.metrics.MetricObservation]: return [ cloudai.metrics.MetricObservation( metric=cloudai.metrics.BANDWIDTH, @@ -43,25 +38,25 @@ def metric_observations(self, system: System, tr: TestRun) -> list[cloudai.metri ] -@pytest.mark.parametrize("successful", [True, False]) def test_standalone_run_output_uses_workload_status_and_metrics( - tmp_path: pathlib.Path, standalone_system: StandaloneSystem, successful: bool + tmp_path: pathlib.Path, standalone_system: StandaloneSystem ) -> None: workload = MetricWorkload( name="metric-workload", description="metric workload", test_template_name="MetricWorkload", - cmd_args=CmdArgs(), - successful=successful, + cmd_args=cloudai.core.CmdArgs(), ) - test_run = TestRun( + test_run = cloudai.core.TestRun( name="case", test=workload, num_nodes=1, nodes=[], output_path=tmp_path / "case" / "0", ) - runner = StandaloneRunner("run", standalone_system, TestScenario(name="scenario", test_runs=[test_run]), tmp_path) + runner = StandaloneRunner( + "run", standalone_system, cloudai.core.TestScenario(name="scenario", test_runs=[test_run]), tmp_path + ) start = datetime.datetime(2026, 1, 2, 3, 4, 5, tzinfo=datetime.timezone.utc) job = StandaloneJob( test_run, @@ -72,18 +67,21 @@ def test_standalone_run_output_uses_workload_status_and_metrics( run = runner.get_run_output(job, test_run, runner.get_job_status(job)) - assert run.status == ("completed" if successful else "failed") - assert run.jobid == "123" - assert run.start == start - assert run.finish == start + datetime.timedelta(seconds=3) - if successful: - assert [metric.model_dump() for metric in run.metrics] == [ + assert run.model_dump() == { + "path": str((tmp_path / "case" / "0").absolute()), + "jobid": "123", + "status": "completed", + "metrics": [ { "name": "Bandwidth", "value": 12.5, "unit": "GB/s", "dimensions": [{"name": "Size", "value": "1024", "unit": "", "is_x": False}], } - ] - else: - assert run.metrics == [] + ], + "start": start, + "finish": start + datetime.timedelta(seconds=3), + "duration": None, + "iteration": 0, + "step": 0, + } diff --git a/tests/test_acceptance.py b/tests/test_acceptance.py index 0e5ebd485..5816a1cbb 100644 --- a/tests/test_acceptance.py +++ b/tests/test_acceptance.py @@ -89,11 +89,27 @@ from cloudai.workloads.vllm import VllmArgs, VllmCmdArgs, VllmRayStartArgs, VllmTestDefinition SLURM_TEST_SCENARIOS = [ - {"path": Path("conf/common/test_scenario/sleep.toml"), "expected_dirs_number": 4, "log_file": "sleep_debug.log"}, + { + "path": Path("conf/common/test_scenario/sleep.toml"), + "expected_dirs_number": 4, + "log_file": "sleep_debug.log", + "tests": [ + ("Tests.sleep1", "sleep", "sleep test"), + ("Tests.sleep5", "sleep", "sleep test"), + ("Tests.sleep5_2", "sleep", "sleep test"), + ("Tests.sleep20", "sleep", "sleep test"), + ], + }, { "path": Path("conf/common/test_scenario/ucc_test.toml"), "expected_dirs_number": 4, "log_file": "ucc_test_debug.log", + "tests": [ + ("Tests.alltoall", "ucc_base_test", "UCC alltoall"), + ("Tests.allgather", "ucc_base_test", "UCC allgather"), + ("Tests.allreduce", "ucc_base_test", "UCC allreduce"), + ("Tests.reduce_scatter", "ucc_base_test", "UCC reduce_scatter"), + ], }, ] @@ -176,16 +192,29 @@ def test_experiment_output_is_dumped_and_valid(self, do_dry_run: tuple[Path, dic results_output = next(path for path in tmp_path.iterdir() if path.is_dir()) experiment = Experiment.model_validate_json((results_output / "experiment.json").read_text()) - assert experiment.id == results_output.name - assert experiment.name == toml.load(scenario["path"])["name"] - assert experiment.status == "completed" - assert experiment.path == str(results_output.absolute()) - assert experiment.start is not None - assert experiment.finish is not None - assert experiment.duration is not None - assert len(experiment.tests) == scenario["expected_dirs_number"] - assert all(test.status == "completed" for test in experiment.tests) - assert all(not test.runs for test in experiment.tests) + assert experiment.model_dump() == { + "id": results_output.name, + "name": toml.load(scenario["path"])["name"], + "description": None, + "status": "completed", + "path": str(results_output.absolute()), + "start": experiment.start, + "finish": experiment.finish, + "duration": experiment.duration, + "tests": [ + { + "id": test_id, + "name": name, + "description": description, + "status": "completed", + "path": str(results_output.absolute() / test_id), + "metrics": [], + "runs": [], + "dse": None, + } + for test_id, name, description in scenario["tests"] + ], + } @pytest.fixture diff --git a/tests/test_output.py b/tests/test_output.py index 4c76de1d4..632a3a654 100644 --- a/tests/test_output.py +++ b/tests/test_output.py @@ -58,10 +58,48 @@ def test_experiment_output_preserves_runs_and_finalizes_failure(tmp_path: pathli experiment_output.finish("failed", start + datetime.timedelta(seconds=5)) stored = output_models.Experiment.model_validate_json((tmp_path / "experiment.json").read_text()) - assert stored.status == "failed" - assert stored.duration == 5 - assert stored.tests[0].status == "failed" - assert [(run.jobid, run.status, run.duration) for run in stored.tests[0].runs] == [ - ("101", "completed", 2), - ("102", "failed", 2), - ] + assert stored.model_dump() == { + "id": "experiment", + "name": "scenario", + "description": None, + "status": "failed", + "path": str(tmp_path), + "start": start, + "finish": start + datetime.timedelta(seconds=5), + "duration": 5, + "tests": [ + { + "id": "case", + "name": "workload", + "description": None, + "status": "failed", + "path": str(tmp_path / "case"), + "metrics": [], + "runs": [ + { + "path": str(tmp_path / "case" / "0"), + "jobid": "101", + "status": "completed", + "metrics": [], + "start": start, + "finish": start + datetime.timedelta(seconds=2), + "duration": 2, + "iteration": 0, + "step": 0, + }, + { + "path": str(tmp_path / "case" / "1"), + "jobid": "102", + "status": "failed", + "metrics": [], + "start": start + datetime.timedelta(seconds=2), + "finish": start + datetime.timedelta(seconds=4), + "duration": 2, + "iteration": 1, + "step": 0, + }, + ], + "dse": None, + } + ], + }