Implement benchmarker framework

monitoring/benchmarker/.gitignore

@@ -0,0 +1 @@
+output/

monitoring/benchmarker/artifacts/generation.py

@@ -0,0 +1,26 @@
+import os
+
+from monitoring.benchmarker.artifacts.matplotlib.matplotlib_figure import (
+    generate_matplotlib_figure,
+)
+from monitoring.benchmarker.artifacts.raw_report import generate_raw_report
+from monitoring.benchmarker.configurations.artifacts.artifact import (
+    ArtifactSpecification,
+)
+from monitoring.benchmarker.reports.report import BenchmarkRunReport
+
+
+def generate_artifacts(
+    artifacts_specs: list[ArtifactSpecification],
+    report: BenchmarkRunReport,
+    output_dir: str,
+) -> None:
+    """Generate and save all configured artifacts."""
+    os.makedirs(output_dir, exist_ok=True)
+
+    for spec in artifacts_specs:
+        if "raw_report" in spec and spec.raw_report is not None:
+            generate_raw_report(report, spec.raw_report, output_dir)
+
+        if "matplotlib_figure" in spec and spec.matplotlib_figure is not None:
+            generate_matplotlib_figure(report, spec.matplotlib_figure, output_dir)

monitoring/benchmarker/artifacts/matplotlib/matplotlib_figure.py

@@ -0,0 +1,36 @@
+import os
+
+import matplotlib
+from loguru import logger
+
+matplotlib.use("Agg")  # Non-interactive backend suitable for headless benchmarking
+import matplotlib.pyplot as plt
+
+from monitoring.benchmarker.configurations.artifacts.matplotlib_figure import (
+    MatplotlibFigureSpecification,
+)
+from monitoring.benchmarker.reports.report import BenchmarkRunReport
+
+
+def generate_matplotlib_figure(
+    report: BenchmarkRunReport, fig_spec: MatplotlibFigureSpecification, output_dir: str
+) -> None:
+    filename = (
+        fig_spec.name
+        if fig_spec.name.lower().endswith(".png")
+        else f"{fig_spec.name}.png"
+    )
+    out_path = os.path.join(output_dir, filename)
+    logger.info(
+        f"Generating Matplotlib figure artifact '{fig_spec.name}' -> {out_path}"
+    )
+
+    fig = plt.figure(
+        figsize=(8 * fig_spec.n_subfigure_cols, 5 * fig_spec.n_subfigure_rows)
+    )
+
+    # TODO: Populate graph
+
+    plt.tight_layout()
+    fig.savefig(out_path, bbox_inches="tight")
+    plt.close(fig)

monitoring/benchmarker/artifacts/raw_report.py

@@ -0,0 +1,23 @@
+import json
+import os
+
+from loguru import logger
+
+from monitoring.benchmarker.configurations.artifacts.raw_report import (
+    RawReportSpecification,
+)
+from monitoring.benchmarker.reports.report import BenchmarkRunReport
+
+
+def generate_raw_report(
+    report: BenchmarkRunReport, raw_spec: RawReportSpecification, output_dir: str
+) -> None:
+    filename = (
+        raw_spec.name
+        if raw_spec.name.lower().endswith(".json")
+        else f"{raw_spec.name}.json"
+    )
+    out_path = os.path.join(output_dir, filename)
+    logger.info(f"Writing raw report artifact to {out_path}")
+    with open(out_path, "w") as f:
+        json.dump(report, f, sort_keys=True)

monitoring/benchmarker/benchmark.py

@@ -1,18 +1,14 @@
 #!env/bin/python3
 
 import argparse
-import json
 import os
 import sys
 
-from implicitdict import ImplicitDict
 from loguru import logger
 
-from monitoring.benchmarker.configurations.configuration import BenchmarkConfiguration
-from monitoring.benchmarker.validation import validate_config
-from monitoring.uss_qualifier.fileio import (
-    load_dict_with_references,
-)  # TODO: factor fileio out of uss_qualifier into monitorlib
+from monitoring.benchmarker.artifacts.generation import generate_artifacts
+from monitoring.benchmarker.engine.engine import run_benchmark
+from monitoring.benchmarker.validation import load_config
 
 
 def parseArgs() -> argparse.Namespace:
@@ -24,6 +20,15 @@ def parseArgs() -> argparse.Namespace:
         required=True,
     )
 
+    parser.add_argument(
+        "-o",
+        "--output",
+        "--output-dir",
+        dest="output_dir",
+        default=os.path.join(os.path.dirname(__file__), "output"),
+        help="Folder to which output artifacts should be written (default: monitoring/benchmarker/output)",
+    )
+
     parser.add_argument(
         "--skip-validation",
         action="store_true",
@@ -36,23 +41,18 @@ def parseArgs() -> argparse.Namespace:
 def run_config(
     config_name: str,
     skip_validation: bool,
+    output_dir: str,
 ) -> int:
-    config_src = load_dict_with_references(config_name)
-
-    if not skip_validation:
-        logger.info("Validating configuration...")
-        validation_errors = validate_config(config_src)
-        if validation_errors:
-            for e in validation_errors:
-                logger.error("[{}]: {}", e.json_path, e.message)
-            raise ValueError(
-                f"{len(validation_errors)} benchmark configuration validation errors indicated above.  Hint: resolve the clearest error first and then rerun validation."
-            )
+    config = load_config(config_name, skip_validation)
 
-    whole_config = ImplicitDict.parse(config_src, BenchmarkConfiguration)
+    run_report = run_benchmark(config)
 
-    # Print the resolved configuration to console as JSON
-    print(json.dumps(whole_config, indent=2, sort_keys=True))
+    if "artifacts" in config and config.artifacts:
+        logger.info("Generating configured artifacts...")
+        generate_artifacts(config.artifacts, run_report, output_dir)
+        logger.info("Artifact generation complete.")
+    else:
+        logger.warning("No artifacts specified in configuration.")
 
     return os.EX_OK
 
@@ -66,6 +66,7 @@ def main() -> int:
     exit_code = run_config(
         args.config,
         args.skip_validation,
+        args.output_dir,
     )
     if exit_code != os.EX_OK:
         return exit_code

monitoring/benchmarker/engine/__init__.py

@@ -0,0 +1 @@
+"""Benchmarker execution engine and supporting modules."""

monitoring/benchmarker/engine/actions/actions.py

@@ -0,0 +1,28 @@
+from monitoring.benchmarker.configurations.actions import (
+    BenchmarkActionName,
+    BenchmarkActionSpecification,
+)
+from monitoring.benchmarker.engine.actions.run_command import run_command
+
+
+def run_scenario_actions(
+    action_names: list[BenchmarkActionName] | None,
+    action_specs: dict[BenchmarkActionName, BenchmarkActionSpecification],
+) -> None:
+    """Run a sequence of scenario setup or teardown actions by name."""
+    if not action_names:
+        return
+
+    for action_name in action_names:
+        if action_name not in action_specs:
+            raise ValueError(
+                f"Scenario action '{action_name}' not defined in configuration.actions"
+            )
+        action_spec = action_specs[action_name]
+
+        if "run_command" in action_spec and action_spec.run_command is not None:
+            run_command(action_spec.run_command)
+        else:
+            raise NotImplementedError(
+                f"Action '{action_name}' has no recognized action specification implemented in this version of benchmarker"
+            )

monitoring/benchmarker/engine/actions/run_command.py

@@ -0,0 +1,20 @@
+import os
+import subprocess
+
+from loguru import logger
+
+from monitoring.benchmarker.configurations.actions import RunCommandActionSpecification
+from monitoring.monitorlib.versioning import get_repo_root
+
+
+def run_command(cmd_spec: RunCommandActionSpecification) -> None:
+    repo_root = get_repo_root()
+    cwd = cmd_spec.path.replace("$REPO_ROOT", repo_root)
+    env = os.environ.copy()
+    if "env" in cmd_spec and cmd_spec.env:
+        for k, v in cmd_spec.env.items():
+            env[k] = str(v)
+
+    logger.info(f"Action: Running command `{cmd_spec.command}` in {cwd}")
+
+    subprocess.run(cmd_spec.command, shell=True, cwd=cwd, env=env, check=True)

monitoring/benchmarker/engine/engine.py

@@ -0,0 +1,104 @@
+import asyncio
+from concurrent.futures import ThreadPoolExecutor
+
+from loguru import logger
+
+from monitoring.benchmarker.configurations.configuration import BenchmarkConfiguration
+from monitoring.benchmarker.engine.actions.actions import run_scenario_actions
+from monitoring.benchmarker.engine.loads.loads import run_scenario_load
+from monitoring.benchmarker.engine.operations import (
+    group_operations,
+)
+from monitoring.benchmarker.engine.resources import instantiate_resources
+from monitoring.benchmarker.reports.report import (
+    BenchmarkReport,
+    BenchmarkRunReport,
+    BenchmarkScenarioReport,
+)
+from monitoring.monitorlib.versioning import get_code_version, get_commit_hash
+
+
+async def _run_benchmark_async(
+    config: BenchmarkConfiguration,
+    codebase_version: str,
+    commit_hash: str,
+) -> BenchmarkRunReport:
+    logger.info("Instantiating declared resources...")
+    resource_pool = instantiate_resources(config)
+
+    user_specs_map = {u.name: u for u in config.user_types}
+    loads_map = {load.name: load for load in config.loads}
+
+    max_io_threads = 400
+    executor = ThreadPoolExecutor(
+        max_workers=max_io_threads, thread_name_prefix="benchmarker_io_"
+    )
+
+    scenarios_reports: list[BenchmarkScenarioReport] = []
+
+    action_list = config.actions if "actions" in config and config.actions else []
+    action_specs = {action_spec.name: action_spec for action_spec in action_list}
+
+    try:
+        for scenario_spec in config.scenarios:
+            logger.info(
+                f"========== Starting Scenario '{scenario_spec.name}' =========="
+            )
+
+            # Run setup actions
+            setup_actions = scenario_spec.setup if "setup" in scenario_spec else None
+            run_scenario_actions(setup_actions, action_specs)
+
+            # Run load
+            if scenario_spec.load not in loads_map:
+                raise ValueError(
+                    f"Scenario load '{scenario_spec.load}' not defined in configuration.loads"
+                )
+            load_spec = loads_map[scenario_spec.load]
+            scenario_ops, scenario_steps = await run_scenario_load(
+                load_spec,
+                user_specs_map,
+                resource_pool,
+                executor,
+            )
+
+            # Run teardown actions
+            teardown_actions = (
+                scenario_spec.teardown if "teardown" in scenario_spec else None
+            )
+            run_scenario_actions(teardown_actions, action_specs)
+
+            # Generate and record scenario report
+            scenario_report = BenchmarkScenarioReport(
+                operations=group_operations(scenario_ops),
+                steps=scenario_steps,
+            )
+            if "metadata" in scenario_spec:
+                scenario_report.metadata = (
+                    dict(scenario_spec.metadata)
+                    if scenario_spec.metadata is not None
+                    else None
+                )
+            scenarios_reports.append(scenario_report)
+
+            logger.info(
+                f"========== Completed Scenario '{scenario_spec.name}' =========="
+            )
+    finally:
+        executor.shutdown(wait=True)
+
+    run_report = BenchmarkRunReport(
+        codebase_version=codebase_version,
+        commit_hash=commit_hash,
+        configuration=config,
+        report=BenchmarkReport(scenarios=scenarios_reports),
+    )
+
+    return run_report
+
+
+def run_benchmark(config: BenchmarkConfiguration) -> BenchmarkRunReport:
+    """Execute the benchmarker engine for the provided configuration and return the resulting report."""
+    codebase_version = get_code_version()
+    commit_hash = get_commit_hash()
+    return asyncio.run(_run_benchmark_async(config, codebase_version, commit_hash))