register Elbencho as a CBT benchmark

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):

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.")

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)

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

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)
 

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)