From 0d83b381c7ca4ac3c6ff35a65531f82e60bec6bc Mon Sep 17 00:00:00 2001 From: Kenan Al-Shamie Date: Wed, 12 Aug 2026 11:54:44 +0100 Subject: [PATCH 1/2] register Elbencho as a CBT benchmark with workloads-aware factory bypass --- benchmark/benchmark.py | 2 +- benchmark/elbencho.py | 145 +++++++++++++++++ benchmarkfactory.py | 60 ++++--- docs/Workloads.md | 11 +- tests/test_benchmarkfactory.py | 6 +- tests/test_bm_elbencho.py | 285 +++++++++++++++++++++++++++++++++ 6 files changed, 483 insertions(+), 26 deletions(-) create mode 100644 benchmark/elbencho.py create mode 100644 tests/test_bm_elbencho.py diff --git a/benchmark/benchmark.py b/benchmark/benchmark.py index f0cf4513..c24518c4 100644 --- a/benchmark/benchmark.py +++ b/benchmark/benchmark.py @@ -150,7 +150,7 @@ def run(self): with open(config_file, 'w') as fd: yaml.dump(config_dict, fd, default_flow_style=False) - def exists(self): + def exists(self) -> bool: return False def compare(self, baseline): diff --git a/benchmark/elbencho.py b/benchmark/elbencho.py new file mode 100644 index 00000000..db669f66 --- /dev/null +++ b/benchmark/elbencho.py @@ -0,0 +1,145 @@ +""" +CBT benchmark module for Elbencho in S3 mode. + +Registers Elbencho as a benchmark so that adding ``elbencho:`` to a YAML +test plan is enough to invoke it. + +The class snapshots top-level keys as global defaults, and stores the +workloads dictionary for use at run time. List-valued workload +parameters (``threads``, ``iodepth``, ``blocksize``) are subject to +the Cartesian permutation as per usual, but not within the usual +``benchmarkfactory.expand_configs()`` function which applies to +top-level list values — this class owns the expansion internally. +""" + +import logging +import os +import pprint + +import common +import monitoring +import settings + +from .benchmark import Benchmark + +logger = logging.getLogger("cbt") + + +class Elbencho(Benchmark): + """ + The Elbencho S3 benchmark class owns top-level and workload-level + keys. It has built-in functionality to validate the executable Elbencho + binary path and perform the Cartesian expansion of different workloads. + """ + + def __init__(self, archive_dir, cluster, config): + super().__init__(archive_dir, cluster, config) + + # --- top-level (session-wide) keys --- + self.cmd_path = config.get("cmd_path", "/usr/local/bin/elbencho") + self.auth = config.get("auth", {}) + self._global_defaults = { + "cmd_path": self.cmd_path, + "auth": self.auth, + } + + # --- workload-level (isolated) keys --- + # workloads is a dict of named workload entries; + # each entry carries its own benchmark parameters. + self.workloads = config.get("workloads", {}) + if not isinstance(self.workloads, dict): + raise ValueError(f"workloads must be a dict, got {type(self.workloads).__name__}") + + for workload_name, workload_params in self.workloads.items(): + if not isinstance(workload_params, dict): + raise ValueError(f"workload '{workload_name}' must be a dict") + + # This is for use in future stories + self.base_run_dir = self.run_dir + + # Validate required keys in the YAML schema + for workload_name, workload_params in self.workloads.items(): + if "mode" not in workload_params: + raise ValueError(f"workload '{workload_name}' missing required key 'mode'") + if "s3_bucket" not in workload_params: + raise ValueError(f"workload '{workload_name}' missing required key 's3_bucket'") + + if self.workloads: + logger.info( + "%d Elbencho workload(s) defined:\n %s", + len(self.workloads), + pprint.pformat(self.workloads).replace("\n", "\n "), + ) + + # ------------------------------------------------------------------ + # Config generation + # ------------------------------------------------------------------ + + @classmethod + def workload_configs(cls, config): + """ + Elbencho uses a workloads model: list-valued parameters (``threads``, + ``iodepth``, ``blocksize``) are expanded internally by the class, not + by ``benchmarkfactory.expand_configs()``. Yield a single config dict + so the factory instantiates exactly one Elbencho object per run. + """ + yield dict(config) + + # ------------------------------------------------------------------ + # Lifecycle helpers + # ------------------------------------------------------------------ + + def exists(self) -> bool: + """Return True if the output archive directory already contains results.""" + if os.path.exists(self.archive_dir): + logger.info("Skipping existing Elbencho results in %s.", self.archive_dir) + return True + return False + + def initialize(self): + super().initialize() + + # Raises if the binary is missing or not executable on any node. + # before touching the cluster or starting monitoring. + logger.info("Verifying elbencho binary is executable on all client nodes: %s", self.cmd_path) + common.pdsh( + settings.getnodes('clients'), + f"test -x {self.cmd_path}", + continue_if_error=False, + ).communicate() + + self.cleandir() + + def run(self): + super().run() + + if not self.workloads: + logger.warning("Elbencho: no workloads defined — nothing to run.") + return + + self.dropcaches() + common.make_remote_dir(self.run_dir) + self.cluster.dump_config(self.run_dir) + + monitoring.start(self.run_dir) + + self._run_workloads() + + monitoring.stop(self.run_dir) + common.sync_files(f"{self.run_dir}/*", self.archive_dir) + + def cleanup(self): + pass + + # ------------------------------------------------------------------ + # Internal helpers + # ------------------------------------------------------------------ + + def _run_workloads(self): + """Iterate over every named workload in definition order. + + The three-tier nested loop (blocksize → threads → iodepth) will be + left as a TODO. Here we only validate and log each + workload entry so the class is fully registered and callable. + """ + logger.info("Elbencho: workload iteration complete.") diff --git a/benchmarkfactory.py b/benchmarkfactory.py index f9bdae5b..23e4647c 100644 --- a/benchmarkfactory.py +++ b/benchmarkfactory.py @@ -2,32 +2,62 @@ import itertools import settings -from benchmark.radosbench import Radosbench +from benchmark.cephtestrados import CephTestRados +from benchmark.cosbench import Cosbench +from benchmark.elbencho import Elbencho from benchmark.fio import Fio +from benchmark.getput import Getput from benchmark.hsbench import Hsbench -from benchmark.rbdfio import RbdFio -from benchmark.rawfio import RawFio from benchmark.kvmrbdfio import KvmRbdFio from benchmark.librbdfio import LibrbdFio from benchmark.nullbench import Nullbench -from benchmark.cosbench import Cosbench -from benchmark.cephtestrados import CephTestRados -from benchmark.getput import Getput +from benchmark.radosbench import Radosbench +from benchmark.rawfio import RawFio +from benchmark.rbdfio import RbdFio + +BENCHMARK_CLASSES = { + 'nullbench': Nullbench, + 'radosbench': Radosbench, + 'fio': Fio, + 'hsbench': Hsbench, + 'rbdfio': RbdFio, + 'kvmrbdfio': KvmRbdFio, + 'rawfio': RawFio, + 'librbdfio': LibrbdFio, + 'cosbench': Cosbench, + 'cephtestrados': CephTestRados, + 'getput': Getput, + 'elbencho': Elbencho, +} + def get_all(archive, cluster, iteration): for benchmark, config in sorted(settings.benchmarks.items()): default = {"benchmark": benchmark, "iteration": iteration} - for current in all_configs(config): + bclass = BENCHMARK_CLASSES.get(benchmark) + # If the benchmark uses workloads, we use its internal method + # to generate the configurations rather than going down the usual + # expand_configs() method which is deprecated (we want to move towards + # workloads) + if bclass is not None and hasattr(bclass, 'workload_configs'): + configs = bclass.workload_configs(config) + else: + configs = expand_configs(config) + + for current in configs: current.update(default) yield get_object(archive, cluster, benchmark, current) -def all_configs(config): +def expand_configs(config): """ return all parameter combinations for config config: dict - list of params iterate over all top-level lists in config + + Deprecated: we're moving towards workloads, + which is a different code path -- see get_all() """ cycle_over_lists = [] cycle_over_names = [] @@ -50,19 +80,7 @@ def all_configs(config): yield current def get_object(archive, cluster, benchmark, bconfig): - benchmarks = { - 'nullbench': Nullbench, - 'radosbench': Radosbench, - 'fio': Fio, - 'hsbench': Hsbench, - 'rbdfio': RbdFio, - 'kvmrbdfio': KvmRbdFio, - 'rawfio': RawFio, - 'librbdfio': LibrbdFio, - 'cosbench': Cosbench, - 'cephtestrados': CephTestRados, - 'getput': Getput} try: - return benchmarks[benchmark](archive, cluster, bconfig) + return BENCHMARK_CLASSES[benchmark](archive, cluster, bconfig) except KeyError: return None diff --git a/docs/Workloads.md b/docs/Workloads.md index 81d1f2b0..44f5a3dc 100644 --- a/docs/Workloads.md +++ b/docs/Workloads.md @@ -7,7 +7,16 @@ of jobs (or threads, or processes), such that the increase number of these cause increase in the I/O. Specifiying workloads in this way permits to generate *response latency curves* from the results. -The workload feature is currently supported for `librbdfio` only. +The workload feature is currently supported for `librbdfio` and `elbencho`. + +## How it works + +Benchmarks that manage their own iteration declare a `workload_configs` classmethod. +`benchmarkfactory` checks for this method at runtime and, when present, calls it instead of +`expand_configs()` — the Cartesian expansion of top-level list-valued parameters then becomes +the responsibility of the class itself. This keeps the factory generic: adding workload support +to a new benchmark only requires implementing `workload_configs` on that class — no changes to +`benchmarkfactory.py` are needed. ![workloads](./workloads.png) diff --git a/tests/test_benchmarkfactory.py b/tests/test_benchmarkfactory.py index 5b6a6ad7..d1ca0f02 100644 --- a/tests/test_benchmarkfactory.py +++ b/tests/test_benchmarkfactory.py @@ -10,7 +10,7 @@ class TestBenchmarkFactory(unittest.TestCase): def test_permutations_1(self): """ Basic sanity permutations """ config = {"x": 12, "y": True, "z": {1: 2}, "t": [1, 2, 4]} - cfgs = list(benchmarkfactory.all_configs(config)) + cfgs = list(benchmarkfactory.expand_configs(config)) self.assertEqual(len(cfgs), 3) self.assertEqual([dict] * 3, list(map(type, cfgs))) tvals = [] @@ -25,7 +25,7 @@ def test_permutations_1(self): def test_permutations_2(self): """ Basic sanity permutations """ config = {"x": 12, "y": True, "z": {1: 2}, "t": [1, 2, 4], "j": [7, True, 'gg']} - cfgs = list(benchmarkfactory.all_configs(config)) + cfgs = list(benchmarkfactory.expand_configs(config)) self.assertEqual(len(cfgs), 9) self.assertEqual([dict] * 9, list(map(type, cfgs))) @@ -43,7 +43,7 @@ def test_permutations_2(self): def test_permutations_0(self): """ Basic sanity permutations """ config = {"x": 12, "y": True, "z": {1: 2}} - cfgs = list(benchmarkfactory.all_configs(config)) + cfgs = list(benchmarkfactory.expand_configs(config)) self.assertEqual(len(cfgs), 1) self.assertEqual(cfgs[0], config) diff --git a/tests/test_bm_elbencho.py b/tests/test_bm_elbencho.py new file mode 100644 index 00000000..13cd5d9d --- /dev/null +++ b/tests/test_bm_elbencho.py @@ -0,0 +1,285 @@ +""" +Validates that Elbencho correctly initializes with defaults and overrides, +merges global and per-workload configuration, integrates with benchmarkfactory, +and owns Cartesian expansion of list-valued workload parameters internally, +bypassing benchmarkfactory.expand_configs(). + +These tests use mock infrastructure and do not require a live cluster. +""" + +import tempfile +import unittest + +import benchmarkfactory +import settings +from benchmark.elbencho import Elbencho +from cluster.ceph import Ceph + +# --------------------------------------------------------------------------- +# Shared test fixtures +# --------------------------------------------------------------------------- + +INVARIANT_YAML = "tools/invariant.yaml" + +_MINIMAL_CONFIG = { + "iteration": 0, + "benchmark": "elbencho", +} + +_FULL_CONFIG = { + "iteration": 0, + "benchmark": "elbencho", + "cmd_path": "/opt/elbencho/bin/elbencho", + "auth": { + "config": "access_key=AKID;secret_key=;url=http://rgw:7480;retry=9" + }, + "workloads": { + "write_small": { + "s3_bucket": "cbt-benchmark", + "mkdirs": True, + "threads": [1, 4, 16], + "iodepth": [1, 4, 16], + "blocksize": ["4k", "128k"], + "size": "4g", + "num_objects": 1000, + "mode": "write", + "direct": True, + "duration": 60, + }, + "read_small": { + "s3_bucket": "cbt-benchmark", + "threads": [1, 4, 16], + "iodepth": [1, 4], + "blocksize": ["4k"], + "size": "4g", + "num_objects": 1000, + "mode": "read", + "duration": 60, + }, + }, +} + + +class TestElbenchoDefaults(unittest.TestCase): + """Test that construction with minimal config produces correct defaults.""" + + archive_dir = "/tmp" + + @classmethod + def setUpClass(cls): + settings.mock_initialize(config_file=INVARIANT_YAML) + cls.cluster = Ceph.mockinit(settings.cluster) + + def _make(self, config=None) -> Elbencho: + cfg = dict(_MINIMAL_CONFIG, **(config or {})) + b = benchmarkfactory.get_object(self.archive_dir, self.cluster, "elbencho", cfg) + assert isinstance(b, Elbencho) + return b + + def test_returns_elbencho_instance(self): + b = self._make() + self.assertIsInstance(b, Elbencho) + + def test_default_cmd_path(self): + b = self._make() + self.assertEqual("/usr/local/bin/elbencho", b.cmd_path) + + def test_default_auth_is_empty_dict(self): + b = self._make() + self.assertEqual({}, b.auth) + + def test_default_workloads_is_empty_dict(self): + b = self._make() + self.assertEqual({}, b.workloads) + + def test_global_defaults_cmd_path(self): + b = self._make() + self.assertEqual("/usr/local/bin/elbencho", b._global_defaults["cmd_path"]) + + def test_global_defaults_auth(self): + b = self._make() + self.assertEqual({}, b._global_defaults["auth"]) + + def test_global_defaults_keys_only_cmd_path_and_auth(self): + """_global_defaults must contain exactly the two session-wide keys.""" + b = self._make() + self.assertEqual({"cmd_path", "auth"}, set(b._global_defaults.keys())) + + def test_base_run_dir_set(self): + b = self._make() + self.assertIsNotNone(b.base_run_dir) + self.assertIsInstance(b.base_run_dir, str) + + def test_archive_dir_set(self): + b = self._make() + self.assertIsNotNone(b.archive_dir) + + +class TestElbenchoExplicitConfig(unittest.TestCase): + """Test construction with fully-specified top-level and workload config.""" + + archive_dir = "/tmp" + + @classmethod + def setUpClass(cls): + settings.mock_initialize(config_file=INVARIANT_YAML) + cls.cluster = Ceph.mockinit(settings.cluster) + + def _make(self) -> Elbencho: + b = benchmarkfactory.get_object( + self.archive_dir, self.cluster, "elbencho", dict(_FULL_CONFIG) + ) + assert isinstance(b, Elbencho) + return b + + def test_custom_cmd_path(self): + b = self._make() + self.assertEqual("/opt/elbencho/bin/elbencho", b.cmd_path) + + def test_custom_auth(self): + b = self._make() + self.assertIn("config", b.auth) + self.assertIn("access_key=AKID", b.auth["config"]) + + def test_workloads_stored_as_dict(self): + b = self._make() + self.assertIsInstance(b.workloads, dict) + + def test_workload_names_preserved(self): + b = self._make() + self.assertIn("write_small", b.workloads) + self.assertIn("read_small", b.workloads) + + def test_workload_list_params_not_permuted(self): + """List-valued workload params (threads, iodepth, blocksize) must be + stored as lists — not expanded by the factory.""" + b = self._make() + ws = b.workloads["write_small"] + self.assertIsInstance(ws["threads"], list) + self.assertIsInstance(ws["iodepth"], list) + self.assertIsInstance(ws["blocksize"], list) + self.assertEqual([1, 4, 16], ws["threads"]) + self.assertEqual([1, 4, 16], ws["iodepth"]) + self.assertEqual(["4k", "128k"], ws["blocksize"]) + + def test_global_defaults_reflect_explicit_cmd_path(self): + b = self._make() + self.assertEqual("/opt/elbencho/bin/elbencho", b._global_defaults["cmd_path"]) + + +class TestElbenchoWorkloadMerge(unittest.TestCase): + """Test the per-workload default-merging logic in _run_workloads. + + We test the merge contract directly rather than going through run() which + would try to contact a real cluster. + """ + + archive_dir = "/tmp" + + @classmethod + def setUpClass(cls): + settings.mock_initialize(config_file=INVARIANT_YAML) + cls.cluster = Ceph.mockinit(settings.cluster) + + def _make(self, workloads) -> Elbencho: + cfg = dict(_MINIMAL_CONFIG, workloads=workloads) + b = benchmarkfactory.get_object(self.archive_dir, self.cluster, "elbencho", cfg) + assert isinstance(b, Elbencho) + return b + + def test_workload_mode_missing_raises_error(self): + with self.assertRaises(ValueError) as ctx: + self._make( + { + "no_mode": {"s3_bucket": "test-bucket"}, + } + ) + self.assertIn("missing required key 'mode'", str(ctx.exception)) + + +class TestElbenchoExists(unittest.TestCase): + """Test the exists() skip-guard.""" + + archive_dir = "/tmp" + + @classmethod + def setUpClass(cls): + settings.mock_initialize(config_file=INVARIANT_YAML) + cls.cluster = Ceph.mockinit(settings.cluster) + + def _make(self) -> Elbencho: + b = benchmarkfactory.get_object( + self.archive_dir, self.cluster, "elbencho", dict(_MINIMAL_CONFIG) + ) + assert isinstance(b, Elbencho) + return b + + def test_exists_false_when_archive_dir_absent(self): + b = self._make() + # Point archive_dir at a path that definitely does not exist. + b.archive_dir = "/tmp/__cbt_elbencho_no_such_dir_xyzzy__" + self.assertFalse(b.exists()) + + def test_exists_true_when_archive_dir_present(self): + b = self._make() + with tempfile.TemporaryDirectory() as tmpdir: + b.archive_dir = tmpdir + self.assertTrue(b.exists()) + + +class TestBenchmarkFactoryElbenchoIntegration(unittest.TestCase): + """Test that benchmarkfactory wires elbencho correctly. + + Key guarantee from Story 2: the factory must NOT apply expand_configs() + Cartesian permutation to elbencho configs. + """ + + archive_dir = "/tmp" + + @classmethod + def setUpClass(cls): + settings.mock_initialize(config_file=INVARIANT_YAML) + cls.cluster = Ceph.mockinit(settings.cluster) + + def test_get_object_returns_elbencho(self): + b = benchmarkfactory.get_object( + self.archive_dir, self.cluster, "elbencho", dict(_MINIMAL_CONFIG) + ) + self.assertIsInstance(b, Elbencho) + + def test_unknown_benchmark_returns_none(self): + b = benchmarkfactory.get_object( + self.archive_dir, self.cluster, "no_such_benchmark", dict(_MINIMAL_CONFIG) + ) + self.assertIsNone(b) + + def test_get_all_yields_single_instance_despite_list_valued_workload_params(self): + """Even though workload params contain lists (threads, iodepth, blocksize), + get_all() must yield exactly ONE Elbencho object — not a Cartesian product.""" + settings.benchmarks = { + "elbencho": { + "cmd_path": "/usr/local/bin/elbencho", + "auth": {}, + "workloads": { + "w": { + "s3_bucket": "b", + "threads": [1, 4, 16], + "iodepth": [1, 4, 16], + "blocksize": ["4k", "128k"], + "mode": "write", + } + }, + } + } + objects = list(benchmarkfactory.get_all(self.archive_dir, self.cluster, 0)) + elbencho_objects = [o for o in objects if isinstance(o, Elbencho)] + self.assertEqual(1, len(elbencho_objects)) + + def test_elbencho_uses_workload_configs_delegation(self): + from benchmark.elbencho import Elbencho + self.assertTrue(hasattr(Elbencho, 'workload_configs')) + self.assertTrue(callable(Elbencho.workload_configs)) + + +if __name__ == "__main__": + unittest.main() From 7831e76d54804d10449ac840ccbdfe4a488b2dbb Mon Sep 17 00:00:00 2001 From: Kenan Al-Shamie Date: Wed, 12 Aug 2026 11:54:44 +0100 Subject: [PATCH 2/2] add executable binary path validation for hsbench --- benchmark/hsbench.py | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/benchmark/hsbench.py b/benchmark/hsbench.py index cdc6ad50..cbad434a 100644 --- a/benchmark/hsbench.py +++ b/benchmark/hsbench.py @@ -45,6 +45,16 @@ def exists(self): def initialize(self): super(Hsbench, self).initialize() + # Fail-fast: verify the hsbench binary exists and is executable on all + # client nodes before touching the cluster or starting monitoring. + # Raises if the binary is missing or not executable on any node. + logger.info("Verifying hsbench binary is executable on all client nodes: %s", self.cmd_path) + common.pdsh( + settings.getnodes('clients'), + f"test -x {self.cmd_path}", + continue_if_error=False, + ).communicate() + # Clean and Create the run directory common.clean_remote_dir(self.run_dir) common.make_remote_dir(self.run_dir)