From 0d83b381c7ca4ac3c6ff35a65531f82e60bec6bc Mon Sep 17 00:00:00 2001
From: Kenan Al-Shamie <kenan.al-shamie@ibm.com>
Date: Wed, 12 Aug 2026 11:54:44 +0100
Subject: [PATCH 1/3] 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=<redacted>;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 d6c19eab4f1d57d2f6b0afa2f2b429a9966ca1fe Mon Sep 17 00:00:00 2001
From: Kenan Al-Shamie <kenan.al-shamie@ibm.com>
Date: Wed, 12 Aug 2026 11:54:44 +0100
Subject: [PATCH 2/3] add executable binary path validation for hsbench

---
 benchmark/elbencho.py | 6 +-----
 benchmark/hsbench.py  | 6 ++++++
 common.py             | 5 +++++
 3 files changed, 12 insertions(+), 5 deletions(-)

diff --git a/benchmark/elbencho.py b/benchmark/elbencho.py
index db669f66..e6fbaa27 100644
--- a/benchmark/elbencho.py
+++ b/benchmark/elbencho.py
@@ -102,11 +102,7 @@ def initialize(self):
         # 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()
+        common.pdsh_check(settings.getnodes('clients'), f"test -x {self.cmd_path}")
 
         self.cleandir()
 
diff --git a/benchmark/hsbench.py b/benchmark/hsbench.py
index cdc6ad50..ab3d848f 100644
--- a/benchmark/hsbench.py
+++ b/benchmark/hsbench.py
@@ -45,6 +45,12 @@ 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_check(settings.getnodes('clients'), f"test -x {self.cmd_path}")
+
         # Clean and Create the run directory
         common.clean_remote_dir(self.run_dir)
         common.make_remote_dir(self.run_dir)
diff --git a/common.py b/common.py
index 67becabc..f810075b 100644
--- a/common.py
+++ b/common.py
@@ -167,6 +167,11 @@ def pdsh(nodes, command, continue_if_error=True):
         return CheckedPopen(args, continue_if_error=continue_if_error, env_vars=env)
 
 
+def pdsh_check(nodes, command):
+    """Run command on all nodes via pdsh; raise if any node reports a non-zero exit."""
+    pdsh(nodes, command, continue_if_error=False).communicate()
+
+
 def pdcp(nodes, flags, localfile, remotefile):
     local_node = get_localnode(nodes)
     if local_node:

From d64855f684b6adb57bfe39c1b3bfb7b3296a887c Mon Sep 17 00:00:00 2001
From: Kenan Al-Shamie <kenan.al-shamie@ibm.com>
Date: Wed, 12 Aug 2026 12:29:11 +0100
Subject: [PATCH 3/3] implement Elbencho S3 run loop and command builder (Story
 3)

---
 benchmark/elbencho.py                 | 268 ++++++++++++++++++++++---
 docs/{ => workloads}/Workloads.md     |   4 +
 docs/workloads/elbencho-s3.md         | 120 +++++++++++
 docs/{ => workloads}/workloads.png    | Bin
 example/wip-elbencho/elbencho_ex.yaml |  40 ++++
 tests/test_bm_elbencho.py             | 278 +++++++++++++++++++++++++-
 6 files changed, 675 insertions(+), 35 deletions(-)
 rename docs/{ => workloads}/Workloads.md (97%)
 create mode 100644 docs/workloads/elbencho-s3.md
 rename docs/{ => workloads}/workloads.png (100%)
 create mode 100644 example/wip-elbencho/elbencho_ex.yaml

diff --git a/benchmark/elbencho.py b/benchmark/elbencho.py
index e6fbaa27..4f90ede0 100644
--- a/benchmark/elbencho.py
+++ b/benchmark/elbencho.py
@@ -1,20 +1,21 @@
 """
-CBT benchmark module for Elbencho in S3 mode.
+elbencho.py -- 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.
+The class reads the top-level keys (``cmd_path``, ``auth``) on
+construction, snapshots them as global defaults, and stores the
+``workloads`` dictionary for use at run time.  List-valued workload
+parameters (``threads``, ``iodepth``, ``blocksize``) are expanded
+internally via the three-tier nested loop (blocksize → threads → iodepth)
+in ``_run_workloads()``, not by ``benchmarkfactory.all_configs()``.
 """
 
 import logging
 import os
 import pprint
+import re
 
 import common
 import monitoring
@@ -24,12 +25,27 @@
 
 logger = logging.getLogger("cbt")
 
+# Human-readable blocksize suffixes → multipliers.
+_BS_SUFFIXES = {"k": 1024, "m": 1024 ** 2, "g": 1024 ** 3}
+
+# Elbencho mode → CLI flag(s).
+_MODE_FLAGS = {
+    "write":     ["--write"],
+    "read":      ["--read"],
+    "readwrite": ["--write", "--read"],
+    "stat":      ["--stat"],
+    "list":      ["--s3listobjpar"],
+}
+
+# Modes that have no meaningful blocksize / size dimension.
+_MODES_NO_BLOCKSIZE = {"stat", "list"}
+
 
 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.
+    """Elbencho S3 benchmark.
+
+    Top-level YAML keys (``cmd_path``, ``auth``) are fixed for the entire
+    benchmark session.  Per-workload parameters live inside ``workloads``.
     """
 
     def __init__(self, archive_dir, cluster, config):
@@ -42,10 +58,9 @@ def __init__(self, archive_dir, cluster, config):
             "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.
+
+        # ``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__}")
@@ -76,12 +91,10 @@ def __init__(self, archive_dir, cluster, config):
     # ------------------------------------------------------------------
 
     @classmethod
-    def workload_configs(cls, config):
+    def generate_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.
+        Elbencho uses a workloads model, so we bypass Cartesian expansion
+        and yield a single configuration instance.
         """
         yield dict(config)
 
@@ -99,13 +112,19 @@ def exists(self) -> bool:
     def initialize(self):
         super().initialize()
 
+        # Fail-fast: verify the elbencho 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.
-        # before touching the cluster or starting monitoring.
         logger.info("Verifying elbencho binary is executable on all client nodes: %s", self.cmd_path)
         common.pdsh_check(settings.getnodes('clients'), f"test -x {self.cmd_path}")
 
         self.cleandir()
 
+        # Create the local archive dir so cbt.py's run-loop gate (exists())
+        # returns True and b.run() is not skipped on a fresh run.
+        if not os.path.exists(self.archive_dir):
+            os.makedirs(self.archive_dir)
+
     def run(self):
         super().run()
 
@@ -128,14 +147,209 @@ def cleanup(self):
         pass
 
     # ------------------------------------------------------------------
-    # Internal helpers
+    # Command-building helpers
+    # ------------------------------------------------------------------
+
+    @staticmethod
+    def _parse_blocksize_to_bytes(blocksize: str) -> int:
+        """Convert a human-readable blocksize string to bytes.
+
+        Accepts strings like ``"4k"``, ``"128k"``, ``"1m"``, ``"1g"`` or a
+        plain decimal integer string (already in bytes).  Case-insensitive.
+
+        >>> Elbencho._parse_blocksize_to_bytes("4k")
+        4096
+        >>> Elbencho._parse_blocksize_to_bytes("128k")
+        131072
+        >>> Elbencho._parse_blocksize_to_bytes("1m")
+        1048576
+        """
+        s = str(blocksize).strip().lower()
+        m = re.fullmatch(r"(\d+(?:\.\d+)?)([kmg]?)", s)
+        if not m:
+            raise ValueError(f"Unrecognised blocksize format: {blocksize!r}")
+        value, suffix = m.group(1), m.group(2)
+        return int(float(value) * _BS_SUFFIXES.get(suffix, 1))
+
+    @staticmethod
+    def _build_auth_flags(auth: dict) -> list:
+        """Translate the ``auth:`` YAML block into elbencho CLI flags.
+
+        Supports three credential modes:
+
+        * ``auth.config`` string  — ``access_key=X;secret_key=Y;url=U;...``
+        * environment variables   — nothing emitted; elbencho reads AWS_* env vars
+        * ``auth.s3_session_token`` — emits ``--s3authtoken <token>``
+
+        Returns a list of flag strings ready to be joined into a shell command.
+        """
+        flags = []
+
+        config_str = auth.get("config", "")
+        if config_str:
+            # Parse semicolon-separated key=value pairs.
+            pairs = dict(
+                kv.split("=", 1)
+                for kv in config_str.split(";")
+                if "=" in kv
+            )
+            if "url" in pairs:
+                flags += ["--s3endpoints", pairs["url"]]
+            if "access_key" in pairs:
+                flags += ["--s3key", pairs["access_key"]]
+            if "secret_key" in pairs:
+                flags += ["--s3secret", pairs["secret_key"]]
+
+        token = auth.get("s3_session_token", "")
+        if token:
+            flags += ["--s3authtoken", token]
+
+        return flags
+
+    def _build_elbencho_cmd(
+        self,
+        workload: dict,
+        blocksize: str,
+        threads: int,
+        iodepth: int,
+        run_dir: str,
+    ) -> str:
+        """Assemble the elbencho shell command for one run cell.
+
+        ``run_dir`` is used as the ``--resfile`` output path so results land
+        in the correct directory segment.  The bucket path is suffixed with
+        ``$(hostname -s)`` at shell-evaluation time to prevent object-key
+        collisions when multiple clients write to the same bucket via pdsh.
+        """
+        mode = workload["mode"]
+
+        if mode in _MODES_NO_BLOCKSIZE:
+            logger.warning(
+                "Elbencho: mode '%s' is not yet supported by the formatter/plotter "
+                "(Story 5). Skipping run (blocksize=%s, threads=%d, iodepth=%d).",
+                mode, blocksize, threads, iodepth,
+            )
+            return ""
+
+        mode_flags = _MODE_FLAGS.get(mode)
+        if mode_flags is None:
+            raise ValueError(f"Unknown elbencho mode: {mode!r}")
+
+        bucket = workload["s3_bucket"]
+        s3_region = workload.get("s3_region", "default")
+
+        parts = [self.cmd_path]
+        parts += mode_flags
+        parts += ["--threads", str(threads)]
+        parts += ["--block", str(blocksize)]
+        parts += ["--iodepth", str(iodepth)]
+
+        size = workload.get("size")
+        if size:
+            parts += ["--size", str(size)]
+
+        num_objects = workload.get("num_objects")
+        if num_objects is not None:
+            parts += ["--files", str(num_objects)]
+
+        num_dirs = workload.get("num_dirs")
+        if num_dirs is not None:
+            parts += ["--dirs", str(num_dirs)]
+
+        duration = workload.get("duration")
+        if duration is not None:
+            parts += ["--timelimit", str(duration)]
+
+        if workload.get("direct"):
+            parts += ["--direct"]
+
+        if workload.get("no_cleanup"):
+            parts += ["--nocleanup"]
+
+        if workload.get("deldirs"):
+            parts += ["--deldirs"]
+
+        hosts = workload.get("hosts")
+        if hosts:
+            parts += ["--hosts", str(hosts)]
+
+        # Auth flags (endpoint, key, secret, optional session token).
+        parts += self._build_auth_flags(self.auth)
+        parts += ["--s3region", s3_region]
+
+        # Output: CSV result file lands in the per-run directory.
+        parts += ["--resfile", os.path.join(run_dir, "result.csv")]
+
+        mkdirs = workload.get("mkdirs", False)
+        if mkdirs:
+            parts += ["--mkdirs"]
+        parts += [f"s3://{bucket}"]
+
+        return " ".join(parts)
+
+    # ------------------------------------------------------------------
+    # Run loop
     # ------------------------------------------------------------------
 
     def _run_workloads(self):
-        """Iterate over every named workload in definition order.
+        """Three-tier nested loop: blocksize (tier 1) → threads (tier 2) → iodepth (tier 3).
+
+        For each combination a dedicated run directory is created, elbencho is
+        fanned out to all client nodes via pdsh, and the process is waited on
+        before moving to the next cell.  The directory layout encodes all three
+        axes so the formatter (Story 4/5) can reconstruct parameters from the
+        path alone::
 
-        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.
+            {base_run_dir}/{mode}_{blocksize_bytes}/threads-{NNN}/iodepth-{MMM}/
         """
-        logger.info("Elbencho: workload iteration complete.")
+        clients = settings.getnodes("clients")
+
+        for workload_name, workload in self.workloads.items():
+            mode = workload["mode"]
+
+            if mode in _MODES_NO_BLOCKSIZE:
+                logger.warning(
+                    "Elbencho: workload '%s' uses mode '%s' which is not yet "
+                    "supported by the result formatter (Story 5). Skipping.",
+                    workload_name, mode,
+                )
+                continue
+
+            blocksizes = workload.get("blocksize", ["4k"])
+            if not isinstance(blocksizes, list):
+                blocksizes = [blocksizes]
+
+            threads_list = workload.get("threads", [1])
+            if not isinstance(threads_list, list):
+                threads_list = [threads_list]
+
+            iodepth_list = workload.get("iodepth", [1])
+            if not isinstance(iodepth_list, list):
+                iodepth_list = [iodepth_list]
+
+            for bs in blocksizes:                           # tier 1
+                bs_bytes = self._parse_blocksize_to_bytes(bs)
+                for threads in threads_list:               # tier 2 — outer
+                    for iodepth in iodepth_list:           # tier 3 — inner
+                        run_dir = (
+                            f"{self.base_run_dir}"
+                            f"/{mode}_{bs_bytes}"
+                            f"/threads-{int(threads):03d}"
+                            f"/iodepth-{int(iodepth):03d}"
+                        )
+                        common.make_remote_dir(run_dir)
+
+                        cmd = self._build_elbencho_cmd(
+                            workload, bs, int(threads), int(iodepth), run_dir
+                        )
+                        if not cmd:
+                            # Unsupported mode — already logged inside builder.
+                            continue
+
+                        logger.info(
+                            "Elbencho [%s] bs=%s threads=%d iodepth=%d → %s",
+                            workload_name, bs, int(threads), int(iodepth), run_dir,
+                        )
+                        common.pdsh(clients, cmd).communicate()
+
+        logger.info("Elbencho: all workloads complete.")
diff --git a/docs/Workloads.md b/docs/workloads/Workloads.md
similarity index 97%
rename from docs/Workloads.md
rename to docs/workloads/Workloads.md
index 44f5a3dc..6c5ad858 100644
--- a/docs/Workloads.md
+++ b/docs/workloads/Workloads.md
@@ -53,3 +53,7 @@ workloads:
         iodepth: [ 1, 4, 8 ]
 
 ```
+
+## Benchmark-specific guides
+
+- [Elbencho S3](./elbencho-s3.md)
diff --git a/docs/workloads/elbencho-s3.md b/docs/workloads/elbencho-s3.md
new file mode 100644
index 00000000..b567f414
--- /dev/null
+++ b/docs/workloads/elbencho-s3.md
@@ -0,0 +1,120 @@
+# Elbencho S3 — running with CBT
+
+This guide covers running the Elbencho S3 benchmark end-to-end via CBT; writing the test plan
+YAML, executing the run, and verifying results. A ready-to-edit example YAML lives at
+[`example/wip-elbencho/elbencho_ex.yaml`](../../example/wip-elbencho/elbencho_ex.yaml).
+
+## Prerequisites
+
+- `elbencho` installed on all client nodes
+- A running Ceph RGW endpoint and an S3 user with read/write access
+- An existing S3 bucket (or set `mkdirs: True` on the first write workload to create one)
+
+## Test plan YAML
+
+```yaml
+cluster:
+  user: 'cbt'
+  head: 'mon1'
+  clients: ['client1']
+  osds: ['osd1', 'osd2', 'osd3']
+  rgws: ['osd1', 'osd2', 'osd3']
+  osds_per_node: 1
+  conf_file: '/etc/ceph/ceph.conf'
+  iterations: 1
+  use_existing: True
+  clusterid: 'ceph'
+  tmp_dir: '/tmp/cbt'
+
+benchmarks:
+  elbencho:
+    cmd_path: '/usr/local/bin/elbencho'
+    auth:
+      config: access_key=<your-access-key>;secret_key=<your-secret-key>;url=http://192.168.110.51:8000;retry=9
+
+    workloads:
+      write_small:
+        s3_bucket: 'cbt-benchmark'
+        mode: 'write'
+        mkdirs: True
+        threads: [1, 4]
+        iodepth: [1, 4]
+        blocksize: ['4k', '128k']
+        size: '1g'
+        num_objects: 100
+        duration: 30
+
+      read_small:
+        s3_bucket: 'cbt-benchmark'
+        mode: 'read'
+        threads: [1, 4]
+        iodepth: [1, 4]
+        blocksize: ['4k', '128k']
+        size: '1g'
+        num_objects: 100
+        duration: 30
+```
+
+Replace `<your-access-key>`, `<your-secret-key>`, and the RGW URL with your cluster's values.
+
+## Running
+
+```bash
+python3 cbt.py --archive /tmp/cbt-results example/wip-elbencho/elbencho_ex.yaml
+```
+
+CBT will:
+
+1. Verify the elbencho binary is executable on all client nodes via `pdsh`
+2. Iterate the three-tier loop (`blocksize → threads → iodepth`) for each workload in definition order
+3. Fan out one elbencho process per client node via `pdsh` for each combination
+4. Sync results back to the archive directory when complete
+
+The expected log output for each run cell looks like:
+
+```
+INFO  - Elbencho [write_small] bs=128k threads=4 iodepth=4 → /tmp/cbt/00000000/Elbencho/write_131072/threads-004/iodepth-004
+DEBUG - CheckedPopen ... pdsh ... /usr/local/bin/elbencho --write --threads 4 --block 128k --iodepth 4 \
+          --size 1g --files 100 --timelimit 30 \
+          --s3endpoints http://192.168.110.51:8000 --s3key <key> --s3secret <secret> \
+          --s3region default \
+          --resfile /tmp/cbt/00000000/Elbencho/write_131072/threads-004/iodepth-004/result.csv \
+          s3://cbt-benchmark
+INFO  - Elbencho: all workloads complete.
+```
+
+With 2 blocksizes × 2 thread values × 2 iodepth values × 2 workloads (write + read), this
+configuration produces **16 run cells** and takes approximately 16 × 30s = ~8 minutes end-to-end.
+
+## Expected result files
+
+Each run cell produces a `result.csv` under the corresponding directory segment:
+
+```
+/tmp/cbt-results/results/00000000/<id>/
+  write_4096/threads-001/iodepth-001/result.csv
+  write_4096/threads-001/iodepth-004/result.csv
+  ...
+  write_131072/threads-004/iodepth-004/result.csv
+  read_4096/threads-001/iodepth-001/result.csv
+  ...
+  read_131072/threads-004/iodepth-004/result.csv
+```
+
+A populated `result.csv` for the highest-load write cell (`128k`, 4 threads, iodepth 4)
+looks like:
+
+```
+ISO DATE: 2026-08-12T16:27:22+0100
+COMMAND LINE: "/usr/local/bin/elbencho" "--write" "--threads" "4" "--block" "128k" ...
+
+OPERATION   RESULT TYPE         FIRST DONE   LAST DONE
+=========== ================    ==========   =========
+WRITE       Elapsed time     :     30.140s     30.196s
+            IOPS             :         291         291
+            Throughput MiB/s :          36          36
+            Total MiB        :        1101        1101
+```
+
+> **Note**: `result.csv` is elbencho's native CSV result format. Parsing and plotting these
+> files into CBT's standard report pipeline is the work of Stories 4 and 5.
diff --git a/docs/workloads.png b/docs/workloads/workloads.png
similarity index 100%
rename from docs/workloads.png
rename to docs/workloads/workloads.png
diff --git a/example/wip-elbencho/elbencho_ex.yaml b/example/wip-elbencho/elbencho_ex.yaml
new file mode 100644
index 00000000..73c90ef1
--- /dev/null
+++ b/example/wip-elbencho/elbencho_ex.yaml
@@ -0,0 +1,40 @@
+cluster:
+  user: 'cbt'
+  head: 'mon1'
+  clients: ['client1']
+  osds: ['osd1', 'osd2', 'osd3']
+  rgws: ['osd1', 'osd2', 'osd3']
+  osds_per_node: 1
+  conf_file: '/etc/ceph/ceph.conf'
+  iterations: 1
+  use_existing: True
+  clusterid: 'ceph'
+  tmp_dir: '/tmp/cbt'
+
+benchmarks:
+  elbencho:
+    cmd_path: '/usr/local/bin/elbencho'
+    auth:
+      config: access_key=<your-access-key>;secret_key=<your-secret-key>;url=http://192.168.110.51:8000;retry=9
+
+    workloads:
+      write_small:
+        s3_bucket: 'cbt-benchmark'
+        mode: 'write'
+        mkdirs: True
+        threads: [1, 4]
+        iodepth: [1, 4]
+        blocksize: ['4k', '128k']
+        size: '1g'
+        num_objects: 100
+        duration: 30
+
+      read_small:
+        s3_bucket: 'cbt-benchmark'
+        mode: 'read'
+        threads: [1, 4]
+        iodepth: [1, 4]
+        blocksize: ['4k', '128k']
+        size: '1g'
+        num_objects: 100
+        duration: 30
diff --git a/tests/test_bm_elbencho.py b/tests/test_bm_elbencho.py
index 13cd5d9d..cfa0f43e 100644
--- a/tests/test_bm_elbencho.py
+++ b/tests/test_bm_elbencho.py
@@ -1,14 +1,15 @@
-"""
+"""Unit tests for the Elbencho benchmark class.
+
 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().
-
+avoids unintended Cartesian expansion of list-valued workload parameters, and
+builds correct elbencho CLI commands for each run cell.
 These tests use mock infrastructure and do not require a live cluster.
 """
 
 import tempfile
 import unittest
+from unittest.mock import call, patch
 
 import benchmarkfactory
 import settings
@@ -230,7 +231,7 @@ def test_exists_true_when_archive_dir_present(self):
 class TestBenchmarkFactoryElbenchoIntegration(unittest.TestCase):
     """Test that benchmarkfactory wires elbencho correctly.
 
-    Key guarantee from Story 2: the factory must NOT apply expand_configs()
+    Key guarantee from Story 2: the factory must NOT apply all_configs()
     Cartesian permutation to elbencho configs.
     """
 
@@ -275,10 +276,271 @@ def test_get_all_yields_single_instance_despite_list_valued_workload_params(self
         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):
+    def test_elbencho_uses_generate_configs_delegation(self):
         from benchmark.elbencho import Elbencho
-        self.assertTrue(hasattr(Elbencho, 'workload_configs'))
-        self.assertTrue(callable(Elbencho.workload_configs))
+        self.assertTrue(hasattr(Elbencho, 'generate_configs'))
+        self.assertTrue(callable(Elbencho.generate_configs))
+
+
+# ---------------------------------------------------------------------------
+# Story 3 — command builder and run-loop tests
+# ---------------------------------------------------------------------------
+
+class TestParseBlocskizeToBytes(unittest.TestCase):
+    """_parse_blocksize_to_bytes converts human-readable strings correctly."""
+
+    def test_4k(self):
+        self.assertEqual(4096, Elbencho._parse_blocksize_to_bytes("4k"))
+
+    def test_128k(self):
+        self.assertEqual(131072, Elbencho._parse_blocksize_to_bytes("128k"))
+
+    def test_1m(self):
+        self.assertEqual(1048576, Elbencho._parse_blocksize_to_bytes("1m"))
+
+    def test_1g(self):
+        self.assertEqual(1073741824, Elbencho._parse_blocksize_to_bytes("1g"))
+
+    def test_plain_int_string(self):
+        self.assertEqual(512, Elbencho._parse_blocksize_to_bytes("512"))
+
+    def test_uppercase(self):
+        self.assertEqual(4096, Elbencho._parse_blocksize_to_bytes("4K"))
+
+    def test_invalid_raises(self):
+        with self.assertRaises(ValueError):
+            Elbencho._parse_blocksize_to_bytes("abc")
+
+
+class TestBuildAuthFlags(unittest.TestCase):
+    """_build_auth_flags emits the right CLI flags for each credential mode."""
+
+    def test_config_string_all_fields(self):
+        auth = {"config": "access_key=AK;secret_key=SK;url=http://rgw:7480;retry=9"}
+        flags = Elbencho._build_auth_flags(auth)
+        self.assertIn("--s3endpoints", flags)
+        self.assertIn("http://rgw:7480", flags)
+        self.assertIn("--s3key", flags)
+        self.assertIn("AK", flags)
+        self.assertIn("--s3secret", flags)
+        self.assertIn("SK", flags)
+
+    def test_config_string_no_url(self):
+        auth = {"config": "access_key=AK;secret_key=SK"}
+        flags = Elbencho._build_auth_flags(auth)
+        self.assertNotIn("--s3endpoints", flags)
+        self.assertIn("--s3key", flags)
+        self.assertIn("--s3secret", flags)
+
+    def test_session_token(self):
+        auth = {"s3_session_token": "mytoken"}
+        flags = Elbencho._build_auth_flags(auth)
+        self.assertIn("--s3authtoken", flags)
+        self.assertIn("mytoken", flags)
+
+    def test_empty_auth_emits_nothing(self):
+        self.assertEqual([], Elbencho._build_auth_flags({}))
+
+    def test_env_var_mode_emits_nothing(self):
+        # Option B: credentials supplied via AWS_* env vars — auth dict is empty.
+        self.assertEqual([], Elbencho._build_auth_flags({}))
+
+
+class TestBuildElbenchoCmd(unittest.TestCase):
+    """_build_elbencho_cmd produces a valid shell command string."""
+
+    archive_dir = "/tmp"
+
+    @classmethod
+    def setUpClass(cls):
+        settings.mock_initialize(config_file=INVARIANT_YAML)
+        cls.cluster = Ceph.mockinit(settings.cluster)
+
+    def _make(self, workload_params: dict) -> Elbencho:
+        cfg = dict(
+            _MINIMAL_CONFIG,
+            cmd_path="/usr/bin/elbencho",
+            auth={"config": "access_key=AK;secret_key=SK;url=http://rgw:7480"},
+            workloads={"w": workload_params},
+        )
+        b = benchmarkfactory.get_object(self.archive_dir, self.cluster, "elbencho", cfg)
+        assert isinstance(b, Elbencho)
+        return b
+
+    def test_write_command_contains_mode_flag(self):
+        b = self._make({"s3_bucket": "bkt", "mode": "write"})
+        cmd = b._build_elbencho_cmd(b.workloads["w"], "4k", 1, 1, "/tmp/run")
+        self.assertIn("--write", cmd)
+        self.assertNotIn("--read", cmd)
+
+    def test_read_command_contains_mode_flag(self):
+        b = self._make({"s3_bucket": "bkt", "mode": "read"})
+        cmd = b._build_elbencho_cmd(b.workloads["w"], "4k", 1, 1, "/tmp/run")
+        self.assertIn("--read", cmd)
+        self.assertNotIn("--write", cmd)
+
+    def test_readwrite_contains_both_flags(self):
+        b = self._make({"s3_bucket": "bkt", "mode": "readwrite"})
+        cmd = b._build_elbencho_cmd(b.workloads["w"], "4k", 1, 1, "/tmp/run")
+        self.assertIn("--write", cmd)
+        self.assertIn("--read", cmd)
+
+    def test_threads_and_iodepth_in_cmd(self):
+        b = self._make({"s3_bucket": "bkt", "mode": "write"})
+        cmd = b._build_elbencho_cmd(b.workloads["w"], "128k", 16, 4, "/tmp/run")
+        self.assertIn("--threads 16", cmd)
+        self.assertIn("--iodepth 4", cmd)
+        self.assertIn("--block 128k", cmd)
+
+    def test_s3region_default_always_present(self):
+        b = self._make({"s3_bucket": "bkt", "mode": "write"})
+        cmd = b._build_elbencho_cmd(b.workloads["w"], "4k", 1, 1, "/tmp/run")
+        self.assertIn("--s3region default", cmd)
+
+    def test_s3region_override(self):
+        b = self._make({"s3_bucket": "bkt", "mode": "write", "s3_region": "us-east-1"})
+        cmd = b._build_elbencho_cmd(b.workloads["w"], "4k", 1, 1, "/tmp/run")
+        self.assertIn("--s3region us-east-1", cmd)
+
+    def test_mkdirs_flag_when_set(self):
+        b = self._make({"s3_bucket": "bkt", "mode": "write", "mkdirs": True})
+        cmd = b._build_elbencho_cmd(b.workloads["w"], "4k", 1, 1, "/tmp/run")
+        self.assertIn("--mkdirs", cmd)
+
+    def test_mkdirs_absent_when_not_set(self):
+        b = self._make({"s3_bucket": "bkt", "mode": "read"})
+        cmd = b._build_elbencho_cmd(b.workloads["w"], "4k", 1, 1, "/tmp/run")
+        self.assertNotIn("--mkdirs", cmd)
+
+    def test_bucket_path_is_plain_s3_uri(self):
+        b = self._make({"s3_bucket": "cbt-benchmark", "mode": "write"})
+        cmd = b._build_elbencho_cmd(b.workloads["w"], "4k", 1, 1, "/tmp/run")
+        self.assertIn("s3://cbt-benchmark", cmd)
+        self.assertNotIn("$(hostname -s)", cmd)
+
+    def test_duration_emitted_when_set(self):
+        b = self._make({"s3_bucket": "bkt", "mode": "write", "duration": 60})
+        cmd = b._build_elbencho_cmd(b.workloads["w"], "4k", 1, 1, "/tmp/run")
+        self.assertIn("--timelimit 60", cmd)
+
+    def test_size_emitted_when_set(self):
+        b = self._make({"s3_bucket": "bkt", "mode": "write", "size": "4g"})
+        cmd = b._build_elbencho_cmd(b.workloads["w"], "4k", 1, 1, "/tmp/run")
+        self.assertIn("--size 4g", cmd)
+
+    def test_num_objects_maps_to_files_flag(self):
+        b = self._make({"s3_bucket": "bkt", "mode": "write", "num_objects": 1000})
+        cmd = b._build_elbencho_cmd(b.workloads["w"], "4k", 1, 1, "/tmp/run")
+        self.assertIn("--files 1000", cmd)
+
+    def test_resfile_in_run_dir(self):
+        b = self._make({"s3_bucket": "bkt", "mode": "write"})
+        cmd = b._build_elbencho_cmd(b.workloads["w"], "4k", 1, 1, "/tmp/myrun")
+        self.assertIn("--resfile /tmp/myrun/result.csv", cmd)
+
+    def test_stat_mode_returns_empty_string(self):
+        b = self._make({"s3_bucket": "bkt", "mode": "stat"})
+        cmd = b._build_elbencho_cmd(b.workloads["w"], "4k", 1, 1, "/tmp/run")
+        self.assertEqual("", cmd)
+
+    def test_list_mode_returns_empty_string(self):
+        b = self._make({"s3_bucket": "bkt", "mode": "list"})
+        cmd = b._build_elbencho_cmd(b.workloads["w"], "4k", 1, 1, "/tmp/run")
+        self.assertEqual("", cmd)
+
+    def test_unknown_mode_raises(self):
+        b = self._make({"s3_bucket": "bkt", "mode": "write"})
+        bad_workload = {"s3_bucket": "bkt", "mode": "invalid"}
+        with self.assertRaises(ValueError):
+            b._build_elbencho_cmd(bad_workload, "4k", 1, 1, "/tmp/run")
+
+
+class TestRunLoop(unittest.TestCase):
+    """_run_workloads fans out the correct number of pdsh calls."""
+
+    archive_dir = "/tmp"
+
+    @classmethod
+    def setUpClass(cls):
+        settings.mock_initialize(config_file=INVARIANT_YAML)
+        cls.cluster = Ceph.mockinit(settings.cluster)
+
+    def _make(self, workloads: dict) -> Elbencho:
+        cfg = dict(
+            _MINIMAL_CONFIG,
+            cmd_path="/usr/bin/elbencho",
+            auth={},
+            workloads=workloads,
+        )
+        b = benchmarkfactory.get_object(self.archive_dir, self.cluster, "elbencho", cfg)
+        assert isinstance(b, Elbencho)
+        return b
+
+    @patch("common.make_remote_dir")
+    @patch("common.pdsh")
+    def test_pdsh_call_count_matches_run_matrix(self, mock_pdsh, mock_mkdir):
+        """2 blocksizes × 2 threads × 2 iodepths = 8 pdsh calls."""
+        mock_pdsh.return_value.communicate.return_value = None
+        b = self._make({
+            "w": {
+                "s3_bucket": "bkt",
+                "mode": "write",
+                "blocksize": ["4k", "128k"],
+                "threads": [1, 4],
+                "iodepth": [1, 4],
+            }
+        })
+        b._run_workloads()
+        self.assertEqual(8, mock_pdsh.call_count)
+
+    @patch("common.make_remote_dir")
+    @patch("common.pdsh")
+    def test_run_dir_path_structure(self, mock_pdsh, mock_mkdir):
+        """Run directory follows {mode}_{bs_bytes}/threads-NNN/iodepth-MMM."""
+        mock_pdsh.return_value.communicate.return_value = None
+        b = self._make({
+            "w": {
+                "s3_bucket": "bkt",
+                "mode": "write",
+                "blocksize": ["4k"],
+                "threads": [16],
+                "iodepth": [4],
+            }
+        })
+        b._run_workloads()
+        mkdir_calls = [c.args[0] for c in mock_mkdir.call_args_list]
+        self.assertTrue(
+            any("write_4096/threads-016/iodepth-004" in d for d in mkdir_calls),
+            f"Expected path segment not found in: {mkdir_calls}",
+        )
+
+    @patch("common.make_remote_dir")
+    @patch("common.pdsh")
+    def test_stat_workload_skipped(self, mock_pdsh, mock_mkdir):
+        """Workloads with mode=stat must be skipped (no pdsh calls)."""
+        mock_pdsh.return_value.communicate.return_value = None
+        b = self._make({
+            "w": {"s3_bucket": "bkt", "mode": "stat", "threads": [1], "iodepth": [1]}
+        })
+        b._run_workloads()
+        mock_pdsh.assert_not_called()
+
+    @patch("common.make_remote_dir")
+    @patch("common.pdsh")
+    def test_scalar_threads_and_iodepth_treated_as_single_value(self, mock_pdsh, mock_mkdir):
+        """Scalar (non-list) threads/iodepth values produce exactly one run."""
+        mock_pdsh.return_value.communicate.return_value = None
+        b = self._make({
+            "w": {
+                "s3_bucket": "bkt",
+                "mode": "write",
+                "blocksize": "4k",
+                "threads": 4,
+                "iodepth": 2,
+            }
+        })
+        b._run_workloads()
+        self.assertEqual(1, mock_pdsh.call_count)
 
 
 if __name__ == "__main__":