diff --git a/doc/reporting.rst b/doc/reporting.rst index 1bdeb8d92..0b907480a 100644 --- a/doc/reporting.rst +++ b/doc/reporting.rst @@ -31,6 +31,23 @@ 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 +------------------------- + +CloudAI writes ``experiment.json`` in each scenario's results directory. + +The file contains scenario details, the configured system name, test cases, status, timing, and result paths. Standalone +execution also records each run's process ID, status, iteration, timing, and workload metrics. + +CloudAI updates the file when standalone runs start and finish, then finalizes it when scenario execution succeeds or +fails. Timestamps use UTC; durations use seconds. Unknown timestamps are ``null``. Dry runs also produce scenario and +test-case details without launching workloads. + +Metrics come from ``TestDefinition.metric_observations()``, independently of reporter settings. A completed ordinary +test with one run also includes those metrics on the test case. Each file update replaces the previous snapshot +atomically. Metric extraction or write errors produce warnings without affecting execution. + + .. _general-flow: General Flow diff --git a/src/cloudai/_core/base_runner.py b/src/cloudai/_core/base_runner.py index 10f6bc01c..17f85c2c9 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 +import cloudai.models.output +import cloudai.output + from .base_job import BaseJob from .command_gen_strategy import CommandGenStrategy from .exceptions import JobFailureError, JobSubmissionError @@ -69,6 +73,49 @@ 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 = cloudai.models.output.Experiment( + id=output_path.name, + name=self.test_scenario.name, + system_name=self.system.name, + status="running", + path=str(output_path), + start=datetime.datetime.now(datetime.timezone.utc), + tests=[ + cloudai.models.output.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 = cloudai.output.ExperimentOutput(experiment, output_path) + + def get_run_output( + self, job: BaseJob, tr: TestRun, result: JobStatusResult | None = None + ) -> cloudai.models.output.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 +157,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 +299,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 +313,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..71af6811f --- /dev/null +++ b/src/cloudai/models/output.py @@ -0,0 +1,90 @@ +# 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 +from typing import Literal + +import pydantic + +Status = Literal["pending", "running", "completed", "failed", "cancelled", "unknown"] + + +class Dimension(pydantic.BaseModel): + """One coordinate of a metric measurement.""" + + name: str + value: str + unit: str = "" + is_x: bool = False + + +class Metric(pydantic.BaseModel): + """A canonical measurement at a dimension point.""" + + name: str + value: str | int | pydantic.FiniteFloat | bool + unit: str = "" + dimensions: list[Dimension] = pydantic.Field(default_factory=list) + + +class Run(pydantic.BaseModel): + """A logical execution with normalized metadata and original measurements.""" + + path: str + jobid: str + status: Status = "unknown" + metrics: list[Metric] = pydantic.Field(default_factory=list) + start: datetime.datetime | None = None + finish: datetime.datetime | None = None + duration: pydantic.FiniteFloat | None = None + iteration: int | None = None + step: int | None = None + + +class DSE(pydantic.BaseModel): + """Search space and recommendation for one test case.""" + + space: dict[str, list[str | int | pydantic.FiniteFloat]] + best_config: dict[str, str | int | pydantic.FiniteFloat] | None = None + best_step: int | None = None + + +class Test(pydantic.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] = pydantic.Field(default_factory=list) + runs: list[Run] = pydantic.Field(default_factory=list) + dse: DSE | None = None + + +class Experiment(pydantic.BaseModel): + """Full snapshot spanning the entire scenario, including all DSE trials.""" + + id: str + name: str + system_name: str + description: str | None = None + status: Status = "pending" + path: str + start: datetime.datetime | None = None + finish: datetime.datetime | None = None + duration: pydantic.FiniteFloat | None = None + tests: list[Test] = pydantic.Field(default_factory=list) diff --git a/src/cloudai/output.py b/src/cloudai/output.py new file mode 100644 index 000000000..7d3e1aec0 --- /dev/null +++ b/src/cloudai/output.py @@ -0,0 +1,132 @@ +# 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 + +import cloudai.models.output + + +def elapsed_seconds(start: datetime.datetime | None, finish: datetime.datetime | None) -> float | None: + """Return the nonnegative elapsed time between two timestamps.""" + if start is None or finish is None: + return None + return max((finish - start).total_seconds(), 0.0) + + +class ExperimentOutput: + """Collect experiment results and publish snapshots.""" + + def __init__(self, experiment: cloudai.models.output.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: cloudai.models.output.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: cloudai.models.output.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) -> cloudai.models.output.Experiment: + """Return an independent snapshot without finalizing the experiment.""" + full = self.experiment.model_copy(deep=True) + now = datetime.datetime.now(datetime.timezone.utc) + if full.status == "running": + full.duration = elapsed_seconds(full.start, now) + for test in full.tests: + for run in test.runs: + if run.status == "running": + run.duration = elapsed_seconds(run.start, now) + 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: cloudai.models.output.Status, finish: datetime.datetime | None) -> None: + self.experiment.status = status + self.experiment.finish = finish + self.experiment.duration = elapsed_seconds(self.experiment.start, finish) + for test in self.experiment.tests: + if test.dse is None: + test.metrics = self._single_run_metrics(test) + if status == "completed": + for test in self.experiment.tests: + if test.status not in ("failed", "cancelled"): + test.status = "completed" + self.write() + + @staticmethod + def _single_run_metrics(test: cloudai.models.output.Test) -> list[cloudai.models.output.Metric]: + """Return metrics when one ordinary run completed successfully.""" + if len(test.runs) != 1: + return [] + run = test.runs[0] + if run.status != "completed" or run.step not in (None, 0): + return [] + return [metric.model_copy(deep=True) for metric in run.metrics] + + @staticmethod + def _update_test_status(test: cloudai.models.output.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" diff --git a/src/cloudai/systems/slurm/single_sbatch_runner.py b/src/cloudai/systems/slurm/single_sbatch_runner.py index 0476e2f92..5db16fc30 100644 --- a/src/cloudai/systems/slurm/single_sbatch_runner.py +++ b/src/cloudai/systems/slurm/single_sbatch_runner.py @@ -190,6 +190,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) @@ -207,6 +208,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 f47e129a2..93a177962 100644 --- a/src/cloudai/systems/slurm/slurm_runner.py +++ b/src/cloudai/systems/slurm/slurm_runner.py @@ -64,9 +64,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..6d31304e1 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,15 @@ # 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 +import cloudai.models.output +import cloudai.output +from cloudai.core import BaseJob, BaseRunner, JobIdRetrievalError, JobStatusResult, System, TestRun, TestScenario from cloudai.util import CommandShell from .standalone_job import StandaloneJob @@ -35,17 +40,67 @@ 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 + ) -> cloudai.models.output.Run: + standalone_job = cast(StandaloneJob, job) + status: cloudai.models.output.Status = "running" + metrics: list[cloudai.models.output.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 cloudai.models.output.Run( + path=str(tr.output_path.absolute()), + jobid=str(job.id), + status=status, + metrics=metrics, + start=standalone_job.start, + finish=standalone_job.finish, + duration=cloudai.output.elapsed_seconds(standalone_job.start, 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) -> cloudai.models.output.Metric: + dimensions = [ + cloudai.models.output.Dimension( + name=cloudai.metrics.dimension_label(key), + value=str(value), + ) + for key, value in sorted(observation.dimensions.items()) + ] + return cloudai.models.output.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..8a3e437d7 --- /dev/null +++ b/tests/systems/standalone/test_runner.py @@ -0,0 +1,90 @@ +# 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": 3, + "iteration": 0, + "step": 0, + } + runner.experiment_output.update_run(test_run.name, run) + runner.experiment_output.finish("completed", job.finish) + assert runner.experiment_output.experiment.tests[0].metrics == run.metrics diff --git a/tests/test_acceptance.py b/tests/test_acceptance.py index 080799d77..d3bf181e4 100644 --- a/tests/test_acceptance.py +++ b/tests/test_acceptance.py @@ -25,6 +25,7 @@ import pytest import toml +import cloudai.models.output 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 @@ -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,38 @@ 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 = cloudai.models.output.Experiment.model_validate_json( + (results_output / "experiment.json").read_text() + ) + + assert experiment.model_dump() == { + "id": results_output.name, + "name": toml.load(scenario["path"])["name"], + "system_name": "example-cluster", + "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..dae5cce65 --- /dev/null +++ b/tests/test_output.py @@ -0,0 +1,109 @@ +# 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.models.output +import cloudai.output + + +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 = cloudai.models.output.Experiment( + id="experiment", + name="scenario", + system_name="test-system", + status="running", + path=str(tmp_path), + start=start, + tests=[cloudai.models.output.Test(id="case", name="workload", path=str(tmp_path / "case"))], + ) + experiment_output = cloudai.output.ExperimentOutput(experiment, tmp_path) + first_run = cloudai.models.output.Run( + path=str(tmp_path / "case" / "0"), + jobid="101", + status="completed", + start=start, + finish=start + datetime.timedelta(seconds=2), + duration=2, + iteration=0, + step=0, + ) + second_run = cloudai.models.output.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) + second_run.duration = 2 + experiment_output.update_run("case", second_run) + experiment_output.finish("failed", start + datetime.timedelta(seconds=5)) + + stored = cloudai.models.output.Experiment.model_validate_json((tmp_path / "experiment.json").read_text()) + assert stored.model_dump() == { + "id": "experiment", + "name": "scenario", + "system_name": "test-system", + "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, + } + ], + }