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/_core/base_runner.py b/src/cloudai/_core/base_runner.py index 10f6bc01c..d0c15eeb3 100644 --- a/src/cloudai/_core/base_runner.py +++ b/src/cloudai/_core/base_runner.py @@ -14,12 +14,16 @@ # 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 from pathlib import Path from typing import Dict, List +from cloudai.models import output as output_models +from cloudai.output import ExperimentOutput + from .base_job import BaseJob from .command_gen_strategy import CommandGenStrategy from .exceptions import JobFailureError, JobSubmissionError @@ -69,6 +73,48 @@ 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 + 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 + ], + ) + self.experiment_output = ExperimentOutput(experiment, output_path) + + 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.""" + return None + + 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": + 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.experiment_output.write() + + def finish_output(self, successful: bool) -> 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.""" @@ -110,6 +156,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 +298,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 +312,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/_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 1fc620544..313cbbed2 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,16 +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}") - 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) - - logging.error("Mixing DSE and non-DSE jobs is not allowed.") - return 1 + successful = False + try: + 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) + return 0 + + if all(tr.is_dse_job for tr in test_scenario.test_runs): + 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(successful) def handle_generate_report(args: argparse.Namespace) -> int: diff --git a/src/cloudai/models/output.py b/src/cloudai/models/output.py new file mode 100644 index 000000000..4d62e0233 --- /dev/null +++ b/src/cloudai/models/output.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. + +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 Test(BaseModel): + """A test case with all logical executions and optional DSE metadata.""" + + id: str + name: str + description: str | None = None + status: Status = "pending" + path: str + metrics: list[Metric] = Field(default_factory=list) + runs: list[Run] = Field(default_factory=list) + dse: DSE | None = None + + +class Experiment(BaseModel): + """Full snapshot spanning the entire scenario, including all DSE trials.""" + + 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 + 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..c9f44b273 --- /dev/null +++ b/src/cloudai/output.py @@ -0,0 +1,123 @@ +# 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 logging +import pathlib +import tempfile + +from cloudai.models import output as output_models + + +class ExperimentOutput: + """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) + self.output_path = output_path + + 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: + 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 + 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.""" + 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) -> 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: + for run in test.runs: + self._update_timing(run) + return full + + def write(self) -> None: + """Atomically replace experiment.json, warning on failure.""" + temporary_path: pathlib.Path | None = None + try: + content = self.snapshot().model_dump_json(indent=2) + self.output_path.mkdir(parents=True, exist_ok=True) + 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: + if temporary_path is not None: + 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: 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"): + value = getattr(record, field) + if value is not 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.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) 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..3f09f9523 100644 --- a/src/cloudai/systems/slurm/slurm_runner.py +++ b/src/cloudai/systems/slurm/slurm_runner.py @@ -92,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 completed_test_runs(self, job: BaseJob) -> list[TestRun]: - return [cast(SlurmJob, job).test_run] - 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 d57befb4b..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"); @@ -14,6 +14,7 @@ # See the License for the specific language governing permissions and # limitations under the License. +import datetime from dataclasses import dataclass from cloudai.core import BaseJob @@ -23,4 +24,5 @@ class StandaloneJob(BaseJob): """A job class for standalone execution.""" - pass + 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 016fdbabc..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"); @@ -14,10 +14,14 @@ # 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 cast -from cloudai.core import BaseRunner, JobIdRetrievalError, System, TestRun, TestScenario +import cloudai.metrics +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 @@ -35,17 +39,64 @@ 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) -> 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 on_job_completion(self, job: BaseJob) -> None: + standalone_job = cast(StandaloneJob, job) + standalone_job.finish = datetime.datetime.now(datetime.timezone.utc) + + @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}") tr.output_path = self.get_job_output_path(tr) 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 + start = None if self.mode == "run": + start = datetime.datetime.now(datetime.timezone.utc) 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) + return StandaloneJob(tr, id=job_id, start=start) diff --git a/tests/systems/standalone/test_runner.py b/tests/systems/standalone/test_runner.py new file mode 100644 index 000000000..33a7ef994 --- /dev/null +++ b/tests/systems/standalone/test_runner.py @@ -0,0 +1,87 @@ +# 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 cloudai.core +import cloudai.metrics +from cloudai.systems.standalone import StandaloneJob, StandaloneRunner, StandaloneSystem + + +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: cloudai.core.System, tr: cloudai.core.TestRun + ) -> list[cloudai.metrics.MetricObservation]: + return [ + cloudai.metrics.MetricObservation( + metric=cloudai.metrics.BANDWIDTH, + value=12.5, + dimensions={"size_bytes": 1024}, + ) + ] + + +def test_standalone_run_output_uses_workload_status_and_metrics( + tmp_path: pathlib.Path, standalone_system: StandaloneSystem +) -> None: + workload = MetricWorkload( + name="metric-workload", + description="metric workload", + test_template_name="MetricWorkload", + cmd_args=cloudai.core.CmdArgs(), + ) + test_run = cloudai.core.TestRun( + name="case", + test=workload, + num_nodes=1, + nodes=[], + output_path=tmp_path / "case" / "0", + ) + 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, + 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.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}], + } + ], + "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 080799d77..5816a1cbb 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 ( @@ -88,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"), + ], }, ] @@ -170,6 +187,35 @@ 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.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 def partial_tr(slurm_system: SlurmSystem) -> partial[TestRun]: diff --git a/tests/test_output.py b/tests/test_output.py new file mode 100644 index 000000000..632a3a654 --- /dev/null +++ b/tests/test_output.py @@ -0,0 +1,105 @@ +# 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.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, + } + ], + }