From 065889bc000b3c3d3547d06a22b9b70b2e559d53 Mon Sep 17 00:00:00 2001 From: Kenan Al-Shamie Date: Fri, 4 Sep 2026 09:48:05 +0100 Subject: [PATCH 1/3] add a RemoteExecutor abstraction for pdsh-free parallel SSH pdsh is being dropped from Rocky Linux 10, so the command fan-out mechanism is being moved behind an interface it can be swapped out from incrementally rather than replaced in place. The new remote/ package holds a RemoteExecutor abstract base class (run_command / run_command_with_error_checking) and a stdlib-only AsyncSSHExecutor implementation (asyncio + the system ssh binary, no third-party packages), plus the pdsh-free cluster helpers make_remote_dir/clean_remote_dir/ sync_files. common.py's pdsh code is left untouched so every existing benchmark keeps working unchanged; docs/ReplacingPdsh.md tracks the remaining migration (ceph tracker #80193). Signed-off-by: Kenan Al-Shamie Assisted-by: Claude-v2.1.212:claude-opus-4-8 --- docs/ReplacingPdsh.md | 122 +++++++++++++++++++++++ remote/__init__.py | 0 remote/async_ssh.py | 199 ++++++++++++++++++++++++++++++++++++++ remote/remote_executor.py | 41 ++++++++ tests/test_async_ssh.py | 153 +++++++++++++++++++++++++++++ 5 files changed, 515 insertions(+) create mode 100644 docs/ReplacingPdsh.md create mode 100644 remote/__init__.py create mode 100644 remote/async_ssh.py create mode 100644 remote/remote_executor.py create mode 100644 tests/test_async_ssh.py diff --git a/docs/ReplacingPdsh.md b/docs/ReplacingPdsh.md new file mode 100644 index 00000000..7efe1d09 --- /dev/null +++ b/docs/ReplacingPdsh.md @@ -0,0 +1,122 @@ +# Replacing pdsh with parallel OpenSSH + +## Background + +CBT has historically used `pdsh` (Parallel Distributed Shell) to fan commands +out to remote nodes. pdsh is being dropped from Rocky Linux 10 and is no longer +a safe long-term dependency. This document tracks the work to replace it. + +> **Tracker:** the full pdsh-removal effort is tracked in +> [ceph tracker #80193](https://tracker.ceph.com/issues/80193). Update that +> ticket as call sites are migrated so the remaining work below is not lost. + +## The approach: a RemoteExecutor abstraction + +Rather than swap `pdsh` for `asyncio` in place, the fan-out mechanism is moved +behind an interface so it can be replaced incrementally — one benchmark at a +time — without a big-bang migration. The `remote/` package holds this: + +- **`remote/remote_executor.py`** — `RemoteExecutor`, an abstract base class with + `run_command(nodes, command, continue_if_error=True)`, + `run_command_with_error_checking(nodes, command)`, and the directory/result + helpers `make_remote_dir`, `clean_remote_dir`, and `sync_files`. This is the + single interface the rest of the code calls; the transport is an implementation + detail. + +- **`remote/async_ssh.py`** — `AsyncSSHExecutor(RemoteExecutor)`, a stdlib-only + implementation using Python's `asyncio` and the system `/usr/bin/ssh` binary + (no third-party packages; the name is unrelated to the `asyncssh` PyPI + package). It fans commands out concurrently, takes the SSH user from + `settings.cluster['user']`, and always passes `-o BatchMode=yes` so a host that + needs interactive auth fails fast rather than hanging. `run_command` returns a + list of `(host, stdout, stderr, exit_status)` tuples — one per node — and + raises `RuntimeError` when `continue_if_error` is False and any host fails. + + The pdsh-free cluster helpers Elbencho needs — `make_remote_dir`, + `clean_remote_dir`, and `sync_files` (parallel `scp -r` pull-back) — are + methods on `RemoteExecutor` itself, so swapping the executor swaps directory + setup and result collection along with the command fan-out. Nothing reaches + around the interface to module-level functions. + +`common.py` is left untouched: `pdsh`, `pdsh_check`, `pdcp`, `rpdcp`, and `scp` +all remain, and every benchmark other than Elbencho still uses them. + +## First consumer: Elbencho + +[`benchmark/elbencho.py`](../benchmark/elbencho.py) holds a `RemoteExecutor` (an +`AsyncSSHExecutor`) and runs its own lifecycle through it — binary check, +`dropcaches`, directory setup, the per-cell workload fan-out, and result +pull-back are all pdsh-free. + +Elbencho also drives the shared **Workloads pipeline** the same way `librbdfio` +does: commands are generated by `Workload._create_command_class` (which now has +an `elbencho` branch returning an `ElbenchoCommand`) and iterated via +`Workloads.command_groups()` — a new, execution-agnostic generator that yields +`(output_directory, [command, ...])` per run cell. Elbencho fans the resulting +command strings out through its `RemoteExecutor`. + +It deliberately does **not** call `Benchmark.run()` or `Workloads.run()` yet, +because both still fan out via `pdsh`; Elbencho reproduces only the base steps it +needs (OSD read-ahead, the `benchmark_config.yaml` snapshot) locally, and wraps +monitoring per command group to match `Workloads.run()`. `cluster/*` operations +and `monitoring.py` remain pdsh-based and are called as-is — so a fully migrated +Elbencho run still shows pdsh for the ceph config dump, OSD read-ahead, and +collectl monitoring; only Elbencho's own fan-out uses ssh. + +## Known deviations to unwind later + +These are compromises Elbencho makes to stay contained (elbencho-only) while +the base class is still pdsh-based. They are safe today but should collapse once +the executor seam reaches the base class and `Workloads.run()`. + +1. **`Elbencho.run()` duplicates base `Benchmark.run()` steps.** Because the base + `run()` fans out via pdsh, Elbencho does not call `super().run()`; it instead + reproduces the base steps it needs (OSD read-ahead, the + `benchmark_config.yaml` snapshot). This means any future change to + `Benchmark.run()` (new config-snapshot logic, new OSD handling, etc.) must be + mirrored into `Elbencho.run()` by hand. The fix is item 2 in *What remains*: + route the base lifecycle through `self._remote` so Elbencho can call + `super().run()` again and delete the duplicated steps. + +2. **`Workload._create_commands_from_options` is fio-shaped.** For every + permutation it reads `volumes_per_client` (S3 has no block volumes), splits + `iodepth` across those targets, and sets a `target_number` per command. + `ElbenchoCommand` ignores `target_number` entirely, and with + `volumes_per_client` defaulting to 1 the iodepth split is a no-op (one target, + all iodepth passed through) — so the machinery is inert for Elbencho today, not + wrong. It is left untouched deliberately: generalising the target/volume + concept out of the shared pipeline would mean editing the fio path, which is + out of scope for this PR. Revisit when the pipeline is made benchmark-agnostic. + +## What remains for a complete replacement + +1. **A pdsh `RemoteExecutor`** — add `PdshExecutor(RemoteExecutor)` wrapping the + existing `common.pdsh` calls. With that, `Benchmark` can hold a `self._remote` + defaulting to `PdshExecutor` (behaviour-preserving for every current + benchmark), and Elbencho simply swaps in `AsyncSSHExecutor`. + +2. **Base `Benchmark` lifecycle** — `run()` (its `rm -rf`), `dropcaches`, + `cleandir`, and `sync` in [`benchmark/benchmark.py`](../benchmark/benchmark.py) + call `common.pdsh` directly. Route them through `self._remote` so Elbencho can + call `super().run()` instead of reproducing base steps. + +3. **`Workloads.run()`** — currently fans out via `pdsh` and owns per-group + monitoring / ramp-time / pre-workload-script handling. Parameterise it with a + `RemoteExecutor` so Elbencho can use it directly (inheriting that + monitoring/ramp/script handling) instead of its own loop. + +4. **`common.pdcp` / `common.rpdcp`** — file distribution/retrieval used by + benchmarks other than Elbencho; needs an executor-based replacement. + +5. **All other benchmark modules** — fio, radosbench, hsbench, etc. Once the base + class and `Workloads.run()` are executor-driven, each benchmark migrates by + swapping its executor. + +6. **Localhost short-circuit** — `pdsh` falls back to a local `subprocess` call + when the target resolves to localhost; `AsyncSSHExecutor` currently SSHes even + for localhost. A `get_localnode` guard (already present in `pdsh`) should be + added so single-node local runs avoid the round-trip. + +7. **Remove pdsh from prerequisites** — once all call sites are migrated, pdsh can + be dropped from the setup docs and any CI dependency lists. Tracked in + [ceph tracker #80193](https://tracker.ceph.com/issues/80193). diff --git a/remote/__init__.py b/remote/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/remote/async_ssh.py b/remote/async_ssh.py new file mode 100644 index 00000000..fb90456c --- /dev/null +++ b/remote/async_ssh.py @@ -0,0 +1,199 @@ +""" +Async parallel SSH via the system OpenSSH client. + +This module implements :class:`~remote.remote_executor.RemoteExecutor` using +Python's stdlib ``asyncio`` and the system ``/usr/bin/ssh`` binary — no +third-party packages are required. The directory-setup and result-collection +helpers (``make_remote_dir``, ``clean_remote_dir``, ``sync_files``) are methods +on the executor, so swapping the fan-out mechanism swaps them too. + +The naming here should not be confused with the external ``asyncssh`` PyPI +package; only built-in Python packages are used. +""" + +import asyncio +import logging +import os + +import settings +from common import expanded_node_list +from remote.remote_executor import RemoteExecutor + +logger = logging.getLogger("cbt") + + +async def _ssh_exec_one(host, command, ssh_args): + proc = await asyncio.create_subprocess_exec( + *ssh_args, host, command, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + stdout_bytes, stderr_bytes = await proc.communicate() + return ( + host, + stdout_bytes.decode(errors="replace"), + stderr_bytes.decode(errors="replace"), + proc.returncode, + ) + + +async def _ssh_exec_all(node_list, command, ssh_args): + tasks = [_ssh_exec_one(h, command, ssh_args) for h in node_list] + return await asyncio.gather(*tasks, return_exceptions=True) + + +class AsyncSSHExecutor(RemoteExecutor): + """Run commands on cluster nodes concurrently via the system ``ssh`` binary.""" + + def _build_ssh_args(self): + """Build the base ``ssh`` argv shared by every node invocation. + + ``-o BatchMode=yes`` is always passed so that a host requiring + interactive authentication fails immediately rather than hanging. + The SSH user, if any, is taken from ``settings.cluster['user']``. + """ + ssh_args = ["ssh", "-o", "BatchMode=yes", "-o", "StrictHostKeyChecking=no"] + user = settings.cluster.get("user") + if user: + ssh_args += ["-l", user] + return ssh_args + + def run_command(self, nodes, command, continue_if_error=True): + node_list = expanded_node_list(nodes) + ssh_args = self._build_ssh_args() + + raw = asyncio.run(_ssh_exec_all(node_list, command, ssh_args)) + + results = [] + errors = [] + for item in raw: + if isinstance(item, BaseException): + errors.append(str(item)) + logger.warning("ssh: failed to launch process: %s", item) + continue + host, stdout, stderr, exit_status = item + logger.debug("ssh [%s] exit=%d", host, exit_status) + if stdout: + logger.debug("ssh [%s] stdout: %s", host, stdout.rstrip()) + if stderr: + logger.debug("ssh [%s] stderr: %s", host, stderr.rstrip()) + if exit_status != 0: + detail = (stdout or stderr).rstrip() + msg = f"ssh [{host}] exited {exit_status}: {detail}" + if not continue_if_error: + errors.append(msg) + else: + logger.warning(msg) + results.append((host, stdout, stderr, exit_status)) + + if errors: + raise RuntimeError( + "asyncssh_exec failed on one or more hosts:\n" + "\n".join(errors) + ) + + return results + + def run_command_with_error_checking(self, nodes, command): + self.run_command(nodes, command, continue_if_error=False) + + # ------------------------------------------------------------------ + # pdsh-free cluster helpers (RemoteExecutor interface) + # ------------------------------------------------------------------ + + def make_remote_dir(self, remote_dir): + """Create *remote_dir* on all cluster nodes via parallel SSH.""" + self.run_command_with_error_checking( + all_cluster_nodes(), f'mkdir -p -m0755 -- {remote_dir}' + ) + + def clean_remote_dir(self, remote_dir): + """Remove *remote_dir* from all cluster nodes via parallel SSH.""" + if remote_dir == "/" or not os.path.isabs(remote_dir): + raise SystemExit("Cleaning the remote dir doesn't seem safe, bailing.") + self.run_command_with_error_checking( + all_cluster_nodes(), + f'if [ -d "{remote_dir}" ]; then rm -rf {remote_dir}; fi', + ) + + def sync_files(self, remote_dir, local_dir): + """Pull *remote_dir* from all cluster nodes into *local_dir* via parallel ``scp -r``.""" + nodes_str = all_cluster_nodes() + node_list = expanded_node_list(nodes_str) + + if not os.path.exists(local_dir): + os.makedirs(local_dir) + + if 'user' in settings.cluster: + self.run_command_with_error_checking( + nodes_str, + 'sudo chown -R {0}.{0} {1}'.format(settings.cluster['user'], remote_dir), + ) + + user = settings.cluster.get("user") + scp_base_args = _build_scp_args() + + def _bare_host(h): + return h.split("@", 1)[-1] + + async def _pull_all(): + tasks = [ + _scp_pull_one(_bare_host(h), remote_dir, local_dir, scp_base_args, user=user) + for h in node_list + ] + return await asyncio.gather(*tasks, return_exceptions=True) + + raw = asyncio.run(_pull_all()) + + errors = [] + for item in raw: + if isinstance(item, BaseException): + errors.append(str(item)) + logger.warning("scp: failed to launch process: %s", item) + continue + host, stdout, stderr, exit_status = item + if exit_status != 0: + detail = (stderr or stdout).rstrip() + errors.append(f"scp [{host}] exited {exit_status}: {detail}") + + if errors: + raise RuntimeError( + "async_sync_files failed on one or more hosts:\n" + "\n".join(errors) + ) + + +# --------------------------------------------------------------------------- +# Module-level helpers used by AsyncSSHExecutor +# --------------------------------------------------------------------------- + +def all_cluster_nodes(): + """Return every node in the cluster (clients, osds, mons, rgws, mds). + + Single source of truth for the "all nodes" set used by the cluster-wide + helpers above, so the node groups are declared in exactly one place. + """ + return settings.getnodes('clients', 'osds', 'mons', 'rgws', 'mds') + + +def _build_scp_args(): + """Build the base ``scp`` argv shared by every pull invocation.""" + return ["scp", "-r", "-o", "BatchMode=yes", "-o", "StrictHostKeyChecking=no"] + + +async def _scp_pull_one(host, remote_path, local_dir, scp_base_args, user=None): + dest = os.path.join(local_dir, host) + os.makedirs(dest, exist_ok=True) + src_host = f"{user}@{host}" if user else host + proc = await asyncio.create_subprocess_exec( + *scp_base_args, + f"{src_host}:{remote_path}", + dest, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + stdout_bytes, stderr_bytes = await proc.communicate() + return ( + host, + stdout_bytes.decode(errors="replace"), + stderr_bytes.decode(errors="replace"), + proc.returncode, + ) diff --git a/remote/remote_executor.py b/remote/remote_executor.py new file mode 100644 index 00000000..eab1d5db --- /dev/null +++ b/remote/remote_executor.py @@ -0,0 +1,41 @@ +""" +Abstract interface for running a shell command across a set of cluster nodes. + +CBT historically fanned commands out to remote nodes with ``pdsh`` (see +``common.py``). pdsh is being retired (tracked in +https://tracker.ceph.com/issues/80193), so the fan-out mechanism is being +moved behind this interface. The first concrete implementation is +``remote.async_ssh.AsyncSSHExecutor`` (stdlib asyncio + the system ``ssh`` +binary); a pdsh-backed implementation can follow without changing callers. +""" + +from abc import ABC, abstractmethod + + +class RemoteExecutor(ABC): + """A way of running a single command on one or more cluster nodes.""" + + @abstractmethod + def run_command(self, nodes, command, continue_if_error=True) -> list: + """Run *command* on all *nodes* in parallel. + + Return a list of ``(host, stdout, stderr, exit_status)`` tuples, one + per node. If *continue_if_error* is False and any node exits non-zero + (or fails to launch), raise ``RuntimeError``. + """ + + @abstractmethod + def run_command_with_error_checking(self, nodes, command) -> None: + """Run *command* on all *nodes*; raise ``RuntimeError`` if any fail.""" + + @abstractmethod + def make_remote_dir(self, remote_dir) -> None: + """Create *remote_dir* on every cluster node.""" + + @abstractmethod + def clean_remote_dir(self, remote_dir) -> None: + """Remove *remote_dir* from every cluster node.""" + + @abstractmethod + def sync_files(self, remote_dir, local_dir) -> None: + """Pull *remote_dir* from every cluster node into *local_dir*.""" diff --git a/tests/test_async_ssh.py b/tests/test_async_ssh.py new file mode 100644 index 00000000..9e8071d8 --- /dev/null +++ b/tests/test_async_ssh.py @@ -0,0 +1,153 @@ +"""Unit tests for the async parallel SSH module (remote/async_ssh.py). + +These tests verify that the SSH/scp argv is constructed as expected and that +error conditions (non-zero exit, launch failure, mixed multi-node results) are +handled correctly. The actual ``asyncio.create_subprocess_exec`` call is +mocked — we do not exercise real asyncio/ssh here. +""" + +import unittest +from unittest.mock import AsyncMock, MagicMock, patch + +from remote.async_ssh import AsyncSSHExecutor, _build_scp_args + +_SSH_BASE = ["ssh", "-o", "BatchMode=yes", "-o", "StrictHostKeyChecking=no"] + + +def _make_proc(returncode=0, stdout=b"out", stderr=b"err"): + """Build a fake asyncio subprocess with an awaitable communicate().""" + proc = MagicMock() + proc.communicate = AsyncMock(return_value=(stdout, stderr)) + proc.returncode = returncode + return proc + + +class TestSshArgvConstruction(unittest.TestCase): + + @patch.dict("settings.cluster", {}, clear=True) + def test_ssh_args_no_user(self): + self.assertEqual(_SSH_BASE, AsyncSSHExecutor()._build_ssh_args()) + + @patch.dict("settings.cluster", {"user": "bob"}, clear=True) + def test_ssh_args_with_user(self): + self.assertEqual(_SSH_BASE + ["-l", "bob"], AsyncSSHExecutor()._build_ssh_args()) + + def test_scp_args(self): + self.assertEqual( + ["scp", "-r", "-o", "BatchMode=yes", "-o", "StrictHostKeyChecking=no"], + _build_scp_args(), + ) + + @patch.dict("settings.cluster", {}, clear=True) + @patch("remote.async_ssh.asyncio.create_subprocess_exec", new_callable=AsyncMock) + def test_run_command_passes_expected_argv(self, mock_exec): + mock_exec.return_value = _make_proc(returncode=0) + AsyncSSHExecutor().run_command("h1", "ls -l") + self.assertEqual(list(mock_exec.call_args.args), _SSH_BASE + ["h1", "ls -l"]) + + @patch.dict("settings.cluster", {"user": "bob"}, clear=True) + @patch("remote.async_ssh.asyncio.create_subprocess_exec", new_callable=AsyncMock) + def test_run_command_includes_user_in_argv(self, mock_exec): + mock_exec.return_value = _make_proc(returncode=0) + AsyncSSHExecutor().run_command("h1", "whoami") + self.assertEqual(list(mock_exec.call_args.args), _SSH_BASE + ["-l", "bob", "h1", "whoami"]) + + +class TestRunCommandResults(unittest.TestCase): + + @patch.dict("settings.cluster", {}, clear=True) + @patch("remote.async_ssh.asyncio.create_subprocess_exec", new_callable=AsyncMock) + def test_success_returns_result_tuple(self, mock_exec): + mock_exec.return_value = _make_proc(returncode=0, stdout=b"hi", stderr=b"") + results = AsyncSSHExecutor().run_command("h1", "ls") + self.assertEqual([("h1", "hi", "", 0)], results) + + @patch.dict("settings.cluster", {}, clear=True) + @patch("remote.async_ssh.asyncio.create_subprocess_exec", new_callable=AsyncMock) + def test_fans_out_to_all_nodes(self, mock_exec): + mock_exec.return_value = _make_proc(returncode=0) + results = AsyncSSHExecutor().run_command("h1,h2,h3", "ls") + self.assertEqual(3, mock_exec.call_count) + self.assertEqual(3, len(results)) + + +class TestRunCommandErrorHandling(unittest.TestCase): + + @patch.dict("settings.cluster", {}, clear=True) + @patch("remote.async_ssh.asyncio.create_subprocess_exec", new_callable=AsyncMock) + def test_nonzero_exit_raises_when_continue_false(self, mock_exec): + mock_exec.return_value = _make_proc(returncode=1, stdout=b"", stderr=b"boom") + with self.assertRaises(RuntimeError) as ctx: + AsyncSSHExecutor().run_command("h1", "false", continue_if_error=False) + self.assertIn("h1", str(ctx.exception)) + self.assertIn("exited 1", str(ctx.exception)) + + @patch.dict("settings.cluster", {}, clear=True) + @patch("remote.async_ssh.asyncio.create_subprocess_exec", new_callable=AsyncMock) + def test_run_command_with_error_checking_raises(self, mock_exec): + mock_exec.return_value = _make_proc(returncode=2, stderr=b"nope") + with self.assertRaises(RuntimeError): + AsyncSSHExecutor().run_command_with_error_checking("h1", "false") + + @patch.dict("settings.cluster", {}, clear=True) + @patch("remote.async_ssh.asyncio.create_subprocess_exec", new_callable=AsyncMock) + def test_nonzero_exit_continues_when_true(self, mock_exec): + mock_exec.return_value = _make_proc(returncode=1, stderr=b"warn") + with self.assertLogs("cbt", level="WARNING"): + results = AsyncSSHExecutor().run_command("h1", "false", continue_if_error=True) + self.assertEqual([("h1", "out", "warn", 1)], results) + + @patch.dict("settings.cluster", {}, clear=True) + @patch("remote.async_ssh.asyncio.create_subprocess_exec", new_callable=AsyncMock) + def test_launch_failure_is_captured_and_raised(self, mock_exec): + # A failure to even launch the process is always surfaced as a + # RuntimeError, even with continue_if_error=True. + mock_exec.side_effect = OSError("cannot spawn ssh") + with self.assertLogs("cbt", level="WARNING"): + with self.assertRaises(RuntimeError): + AsyncSSHExecutor().run_command("h1", "ls", continue_if_error=True) + + @patch.dict("settings.cluster", {}, clear=True) + @patch("remote.async_ssh.asyncio.create_subprocess_exec", new_callable=AsyncMock) + def test_multinode_one_failure_names_failing_host(self, mock_exec): + # gather preserves task-creation order: h1 then h2. + mock_exec.side_effect = [ + _make_proc(returncode=0), + _make_proc(returncode=1, stderr=b"bad"), + ] + with self.assertRaises(RuntimeError) as ctx: + AsyncSSHExecutor().run_command("h1,h2", "cmd", continue_if_error=False) + self.assertIn("h2", str(ctx.exception)) + self.assertNotIn("[h1]", str(ctx.exception)) + + @patch.dict("settings.cluster", {}, clear=True) + @patch("remote.async_ssh.asyncio.create_subprocess_exec", new_callable=AsyncMock) + def test_error_detail_prefers_stdout_over_stderr(self, mock_exec): + # The error detail is (stdout or stderr): when a failing command wrote + # to stdout, that text — not stderr — is what surfaces in the message. + mock_exec.return_value = _make_proc(returncode=1, stdout=b"stdout detail", stderr=b"stderr detail") + with self.assertRaises(RuntimeError) as ctx: + AsyncSSHExecutor().run_command("h1", "cmd", continue_if_error=False) + self.assertIn("stdout detail", str(ctx.exception)) + self.assertNotIn("stderr detail", str(ctx.exception)) + + @patch.dict("settings.cluster", {}, clear=True) + @patch("remote.async_ssh.asyncio.create_subprocess_exec", new_callable=AsyncMock) + def test_error_detail_falls_back_to_stderr_when_stdout_empty(self, mock_exec): + mock_exec.return_value = _make_proc(returncode=1, stdout=b"", stderr=b"stderr detail") + with self.assertRaises(RuntimeError) as ctx: + AsyncSSHExecutor().run_command("h1", "cmd", continue_if_error=False) + self.assertIn("stderr detail", str(ctx.exception)) + + @patch.dict("settings.cluster", {}, clear=True) + @patch("remote.async_ssh.asyncio.create_subprocess_exec", new_callable=AsyncMock) + def test_zero_exit_with_stderr_is_not_a_failure(self, mock_exec): + # A command that writes to stderr but exits 0 (e.g. warnings) succeeds: + # only the exit status decides failure, never the presence of stderr. + mock_exec.return_value = _make_proc(returncode=0, stdout=b"", stderr=b"just a warning") + results = AsyncSSHExecutor().run_command("h1", "cmd", continue_if_error=False) + self.assertEqual([("h1", "", "just a warning", 0)], results) + + +if __name__ == "__main__": + unittest.main() From 03c5edf41d5e70feda6508a97ac66858881eb256 Mon Sep 17 00:00:00 2001 From: Kenan Al-Shamie Date: Fri, 4 Sep 2026 09:48:22 +0100 Subject: [PATCH 2/3] add Elbencho S3 benchmark driven by the Workloads pipeline MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Introduce the Elbencho benchmark for S3 object workloads. It builds commands through the shared Workloads pipeline the same way librbdfio does — an ElbenchoCommand (command/elbencho_command.py) created via Workload._create_command_class, with list-valued params expanded by all_configs() — and fans the generated command strings out through a RemoteExecutor, so its whole lifecycle (binary check, dropcaches, directory setup, workload runs, result sync) is pdsh-free. The one non-obvious piece is auth threading: S3 credentials arrive as a nested dict, which the Workloads global-option collection would stringify, so they are flattened into flat string options before the base class builds the pipeline. A new execution-agnostic Workloads.command_groups() generator exposes the per-cell commands so Elbencho can drive them without the still-pdsh-based Workloads.run(). Includes the unit test suite. Signed-off-by: Kenan Al-Shamie Assisted-by: Claude-v2.1.212:claude-opus-4-8 --- benchmark/benchmark.py | 2 +- benchmark/elbencho.py | 162 +++++++++++++ benchmarkfactory.py | 4 +- command/command.py | 14 +- command/elbencho_command.py | 180 ++++++++++++++ command/fio_command.py | 11 +- command/rbd_fio_command.py | 7 +- remote/async_ssh.py | 137 +++++------ remote/remote_executor.py | 67 ++++-- tests/test_async_ssh.py | 18 +- tests/test_bm_elbencho.py | 428 +++++++++++++++++++++++++++++++++ tests/test_elbencho_command.py | 198 +++++++++++++++ tests/test_workload.py | 20 +- workloads/workload.py | 6 + workloads/workloads.py | 27 +++ 15 files changed, 1158 insertions(+), 123 deletions(-) create mode 100644 benchmark/elbencho.py create mode 100644 command/elbencho_command.py create mode 100644 tests/test_bm_elbencho.py create mode 100644 tests/test_elbencho_command.py diff --git a/benchmark/benchmark.py b/benchmark/benchmark.py index 486ed357..326a3e1e 100644 --- a/benchmark/benchmark.py +++ b/benchmark/benchmark.py @@ -158,7 +158,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..75076d94 --- /dev/null +++ b/benchmark/elbencho.py @@ -0,0 +1,162 @@ +"""Elbencho S3 benchmark, driven by the shared Workloads pipeline. + +The generated commands are fanned out to the client nodes through the pdsh-free +``RemoteExecutor``.""" + +import logging +import os + +import yaml + +import monitoring +import settings +from remote.async_ssh import AsyncSSHExecutor +from remote.remote_executor import RemoteExecutor + +from .benchmark import Benchmark + +logger = logging.getLogger("cbt") + + +class Elbencho(Benchmark): + + def __init__(self, archive_dir: str, cluster, config: dict) -> None: + # auth comes in from the YAML as a nested dict; flatten it to + # strings now so the Workloads pipeline (which stringifies everything) + # doesn't mangle it. + self.auth = config.get("auth", {}) + config["s3_auth_config"] = self.auth.get("config", "") + config["s3_session_token"] = self.auth.get("s3_session_token", "") + config.pop("auth", None) + + super().__init__(archive_dir, cluster, config) + + self.cmd_path = config.get("cmd_path", "/usr/local/bin/elbencho") + + # RemoteExecutor is the ABC which allows us to easily + # swap out AsyncIO as the fan-out tool later if needed. + self._remote: RemoteExecutor = AsyncSSHExecutor() + + self.base_run_dir = self.run_dir + + workloads = config.get("workloads", {}) + if not isinstance(workloads, dict): + raise ValueError(f"workloads must be a dict, got {type(workloads).__name__}") + self._validate_workloads(workloads) + + for wl_name, wl_params in workloads.items(): + logger.info("Elbencho workload '%s': %s", wl_name, wl_params) + + # ------------------------------------------------------------------ + # Lifecycle overrides (pdsh-free) + # ------------------------------------------------------------------ + + def exists(self) -> bool: + 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) -> None: + super().initialize() + + logger.info("Verifying elbencho binary is executable on all client nodes: %s", self.cmd_path) + self._remote.run_command_with_error_checking(settings.getnodes('clients'), f"test -x {self.cmd_path}") + + self.cleandir() + + if not os.path.exists(self.archive_dir): + os.makedirs(self.archive_dir) + + def cleandir(self) -> None: + clients = settings.getnodes('clients') + self._remote.clean_remote_dir(clients, self.run_dir) + self._remote.make_remote_dir(clients, self.run_dir) + + def dropcaches(self) -> None: + nodes = settings.getnodes('clients', 'osds') + self._remote.run_command(nodes, 'sync', continue_if_error=False) + self._remote.run_command( + nodes, + 'echo 3 | sudo tee /proc/sys/vm/drop_caches', + continue_if_error=False, + ) + + def run(self) -> None: + if self.osd_ra and self.osd_ra_changed: + logger.info('Setting OSD Read Ahead to: %s', self.osd_ra) + self.cluster.set_osd_param('read_ahead_kb', self.osd_ra) + + config_file = os.path.join(self.archive_dir, 'benchmark_config.yaml') + if not os.path.exists(self.archive_dir): + os.makedirs(self.archive_dir) + if not os.path.exists(config_file): + config_dict = dict(cluster=self.config) + with open(config_file, 'w') as fd: + yaml.dump(config_dict, fd, default_flow_style=False) + + if not self._workloads.exist(): + logger.warning("Elbencho: no workloads defined — nothing to run.") + return + + self.dropcaches() + # TODO: call super().run() once Benchmark.run() is executor-driven and + # is no longer using AsyncIO. + self._remote.make_remote_dir(settings.getnodes('clients'), self.run_dir) + self.cluster.dump_config(self.run_dir) + + self._run_workloads() + + self._remote.sync_files(settings.getnodes('clients'), self.run_dir, self.archive_dir) + + def cleanup(self) -> None: + pass + + # ------------------------------------------------------------------ + # Validation + # ------------------------------------------------------------------ + + def _validate_workloads(self, workloads: dict) -> None: + """Fail early with a precise error rather than mid-run inside the pipeline.""" + for name, params in workloads.items(): + if not isinstance(params, dict): + raise ValueError(f"workload '{name}' must be a dict") + for field in ("mode", "s3_bucket"): + if field not in params: + raise ValueError(f"workload '{name}' missing required key '{field}'") + for field in ("threads", "iodepth"): + values = params.get(field) + if values is None: + continue + for value in (values if isinstance(values, list) else [values]): + try: + int(value) + except (TypeError, ValueError): + raise ValueError( + f"workload '{name}': {field} value {value!r} is not an integer" + ) + + # ------------------------------------------------------------------ + # Run loop + # ------------------------------------------------------------------ + + def _run_workloads(self) -> None: + clients = settings.getnodes("clients") + + self._workloads.set_benchmark_type("elbencho") + self._workloads.set_executable(self.cmd_path) + + for output_directory, commands in self._workloads.command_groups(): + live_commands = [cmd for cmd in commands if cmd] + if not live_commands: + continue + + self._remote.make_remote_dir(settings.getnodes('clients'), output_directory) + logger.info("Elbencho: running %d command(s) → %s", len(live_commands), output_directory) + monitoring.start(output_directory) + for cmd in live_commands: + logger.debug("Elbencho cmd: %s", cmd) + self._remote.run_command(clients, cmd, continue_if_error=False) + monitoring.stop() + + logger.info("Elbencho: all workloads complete.") diff --git a/benchmarkfactory.py b/benchmarkfactory.py index 3b603c45..d9dfced2 100644 --- a/benchmarkfactory.py +++ b/benchmarkfactory.py @@ -1,6 +1,7 @@ import settings from common import all_configs from benchmark.radosbench import Radosbench +from benchmark.elbencho import Elbencho from benchmark.fio import Fio from benchmark.hsbench import Hsbench from benchmark.rbdfio import RbdFio @@ -32,7 +33,8 @@ def get_object(archive, cluster, benchmark, bconfig): 'librbdfio': LibrbdFio, 'cosbench': Cosbench, 'cephtestrados': CephTestRados, - 'getput': Getput} + 'getput': Getput, + 'elbencho': Elbencho} try: return benchmarks[benchmark](archive, cluster, bconfig) except KeyError: diff --git a/command/command.py b/command/command.py index 29afb34e..5c951b3d 100644 --- a/command/command.py +++ b/command/command.py @@ -7,8 +7,9 @@ """ from abc import ABC, abstractmethod +from collections.abc import Mapping from logging import Logger, getLogger -from typing import Optional +from typing import Any, Optional from cli_options import CliOptions @@ -21,13 +22,16 @@ class Command(ABC): system """ - def __init__(self, options: dict[str, str]) -> None: + # ``options`` is the raw config from the YAML/test plan: heterogeneous + # values (ints, bools, strings). _parse_options() is the boundary that + # normalizes it into the str|None CliOptions store. + def __init__(self, options: Mapping[str, Any]) -> None: self._executable: Optional[str] = None self._output_directory: str = "" self._options: CliOptions = self._parse_options(options) @abstractmethod - def _parse_options(self, options: dict[str, str]) -> CliOptions: + def _parse_options(self, options: Mapping[str, Any]) -> CliOptions: """ Take the options passed in from the configuration yaml file and convert them to a list of key/value pairs that match the parameters @@ -42,7 +46,7 @@ def _generate_full_command(self) -> str: """ @abstractmethod - def _parse_global_options(self, options: dict[str, str]) -> CliOptions: + def _parse_global_options(self, options: Mapping[str, Any]) -> CliOptions: """ Parse the set of global options into the correct format for the command type """ @@ -89,7 +93,7 @@ def set_executable(self, executable_path: str) -> None: """ self._executable = executable_path - def set_global_options(self, global_options: dict[str, str]) -> None: + def set_global_options(self, global_options: Mapping[str, Any]) -> None: """ Update the global options """ diff --git a/command/elbencho_command.py b/command/elbencho_command.py new file mode 100644 index 00000000..0b590e3c --- /dev/null +++ b/command/elbencho_command.py @@ -0,0 +1,180 @@ +"""Builds the elbencho command line for a single S3 workload instance. + +It returns the full executable string that can be used to run a cli command. +It is instantiated by ``Workload._create_command_class`` as part of the +shared Workloads pipeline.""" + +import os +import re +import shlex +from logging import Logger, getLogger + +from cli_options import CliOptions +from command.command import Command + +log: Logger = getLogger("cbt") + +_BS_SUFFIXES = {"k": 1024, "m": 1024 ** 2, "g": 1024 ** 3} + + +class ElbenchoCommand(Command): + """A single elbencho S3 command line for one run cell.""" + + _MODE_FLAGS = { + "write": ["--write"], + "read": ["--read"], + "readwrite": ["--write", "--read"], + "stat": ["--stat"], + "list": ["--s3listobjpar"], + } + + _MODES_NO_BLOCKSIZE = {"stat", "list"} + + def __init__(self, options: dict, workload_output_directory: str) -> None: + # Must be set before super().__init__() because the base constructor calls _parse_options(). + self._workload_output_directory: str = workload_output_directory + super().__init__(options) + + @classmethod + def mode_is_supported(cls, mode: str) -> bool: + """Return True if mode is currently supported (stat/list are not yet).""" + return mode not in cls._MODES_NO_BLOCKSIZE + + @staticmethod + def parse_blocksize_to_bytes(blocksize: str) -> int: + 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[str]: + """Return S3 auth flags for whatever credentials are present; empty auth yields no flags.""" + flags: list[str] = [] + + config_str = auth.get("config", "") + if config_str: + 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 + + # ------------------------------------------------------------------ + # Command ABC implementation + # ------------------------------------------------------------------ + + def _parse_options(self, options: dict) -> CliOptions: + # Populate CliOptions with parsed options and defaults. + parsed_options: CliOptions = CliOptions() + + parsed_options["mode"] = options.get("mode") + parsed_options["s3_bucket"] = options.get("s3_bucket") + parsed_options["s3_region"] = options.get("s3_region", "default") + parsed_options["threads"] = str(options.get("threads", 1)) + parsed_options["blocksize"] = str(options.get("blocksize", "4k")) + parsed_options["iodepth"] = str(options.get("iodepth", 1)) + + # Optional value flags + for key in ("size", "num_objects", "num_dirs", "duration", "hosts"): + value = options.get(key) + parsed_options[key] = str(value) if value is not None else None + + # Boolean presence flags: "true" when truthy, None otherwise. + for key in ("deldirs", "s3nompcheck", "mkdirs"): + parsed_options[key] = "true" if options.get(key) else None + + parsed_options["s3_auth_config"] = options.get("s3_auth_config", "") + parsed_options["s3_session_token"] = options.get("s3_session_token", "") + + return parsed_options + + def _parse_global_options(self, options: dict) -> CliOptions: + return CliOptions(options) + + def _generate_output_directory_path(self) -> str: + # {base}/elbencho/{mode}_{blocksize_bytes}/threads-{NNN}/iodepth-{NNN} + options = self._options + mode = str(options["mode"]) + blocksize = str(options["blocksize"]) + threads = int(str(options["threads"])) + iodepth = int(str(options["iodepth"])) + return os.path.join( + self._workload_output_directory, + self.benchmark, + f"{mode}_{self.parse_blocksize_to_bytes(blocksize)}", + f"threads-{threads:03d}", + f"iodepth-{iodepth:03d}", + ) + + @property + def benchmark(self) -> str: + return "elbencho" + + def _generate_full_command(self) -> str: + if self._executable is None: + return "" + + options = self._options + mode = str(options["mode"]) + + if not self.mode_is_supported(mode): + log.warning( + "Elbencho: mode '%s' is not yet supported by the formatter. " + "Skipping run (blocksize=%s, threads=%s, iodepth=%s).", + mode, options["blocksize"], options["threads"], options["iodepth"], + ) + return "" + + mode_flags = self._MODE_FLAGS.get(mode) + if mode_flags is None: + raise ValueError(f"Unknown elbencho mode: {mode!r}") + + cmd_parts: list[str] = [self._executable] + cmd_parts += mode_flags + cmd_parts += ["--threads", str(options["threads"])] + cmd_parts += ["--block", str(options["blocksize"])] + cmd_parts += ["--iodepth", str(options["iodepth"])] + + if options["size"] is not None: + cmd_parts += ["--size", str(options["size"])] + if options["num_objects"] is not None: + cmd_parts += ["--files", str(options["num_objects"])] + if options["num_dirs"] is not None: + cmd_parts += ["--dirs", str(options["num_dirs"])] + if options["duration"] is not None: + cmd_parts += ["--timelimit", str(options["duration"])] + if options["deldirs"]: + cmd_parts += ["--deldirs"] + if options["s3nompcheck"]: + cmd_parts += ["--s3nompcheck"] + if options["hosts"] is not None: + cmd_parts += ["--hosts", str(options["hosts"])] + + cmd_parts += self.build_auth_flags({ + "config": options["s3_auth_config"] or "", + "s3_session_token": options["s3_session_token"] or "", + }) + cmd_parts += ["--s3region", str(options["s3_region"])] + cmd_parts += ["--resfile", os.path.join(self._generate_output_directory_path(), "result.csv")] + + if options["mkdirs"]: + cmd_parts += ["--mkdirs"] + cmd_parts += [f"s3://{options['s3_bucket']}"] + + # Safely quote arguments for remote shell execution. + return shlex.join(cmd_parts) diff --git a/command/fio_command.py b/command/fio_command.py index 7552e83e..65e2dd25 100644 --- a/command/fio_command.py +++ b/command/fio_command.py @@ -11,8 +11,9 @@ """ from abc import ABC, abstractmethod +from collections.abc import Mapping from logging import Logger, getLogger -from typing import Optional +from typing import Any, Optional from cli_options import CliOptions from command.command import Command @@ -29,25 +30,25 @@ class FioCommand(Command, ABC): _REQUIRED_OPTIONS = {"invalidate": "0", "direct": "1"} _DIRECT_TRANSLATIONS: list[str] = ["numjobs", "iodepth"] - def __init__(self, options: dict[str, str], workload_output_directory: str) -> None: + def __init__(self, options: Mapping[str, Any], workload_output_directory: str) -> None: self._target_number: int = int(options["target_number"]) self._total_iodepth: Optional[str] = options.get("total_iodepth", None) self._workload_output_directory: str = workload_output_directory super().__init__(options) @abstractmethod - def _parse_ioengine_specific_parameters(self, options: dict[str, str]) -> dict[str, str]: + def _parse_ioengine_specific_parameters(self, options: Mapping[str, Any]) -> dict[str, str]: """ Get any options that are specific to the I/O engine being used for this fio run and add them to the CliOptons for this workload """ - def _parse_global_options(self, options: dict[str, str]) -> CliOptions: + def _parse_global_options(self, options: Mapping[str, Any]) -> CliOptions: global_options: CliOptions = CliOptions(options) return global_options - def _parse_options(self, options: dict[str, str]) -> CliOptions: + def _parse_options(self, options: Mapping[str, Any]) -> CliOptions: fio_cli_options: CliOptions = CliOptions() fio_cli_options.update(self._parse_ioengine_specific_parameters(options)) diff --git a/command/rbd_fio_command.py b/command/rbd_fio_command.py index 97008cb4..8f8a821d 100644 --- a/command/rbd_fio_command.py +++ b/command/rbd_fio_command.py @@ -15,6 +15,9 @@ Of these clustername and busy_poll are not currently used by CBT """ +from collections.abc import Mapping +from typing import Any + from command.fio_command import FioCommand from common import get_fqdn_cmd @@ -26,14 +29,14 @@ class RbdFioCommand(FioCommand): _RBD_DEFAULT_OPTIONS: dict[str, str] = {"ioengine": "rbd", "clientname": "admin"} - def __init__(self, options: dict[str, str], workload_output_directory: str) -> None: + def __init__(self, options: Mapping[str, Any], workload_output_directory: str) -> None: super().__init__(options, workload_output_directory) @property def benchmark(self) -> str: return "rbdfio" - def _parse_ioengine_specific_parameters(self, options: dict[str, str]) -> dict[str, str]: + def _parse_ioengine_specific_parameters(self, options: Mapping[str, Any]) -> dict[str, str]: rbd_options: dict[str, str] = self._RBD_DEFAULT_OPTIONS rbd_base_name: str = options.get("rbdname", "cbt-fio") diff --git a/remote/async_ssh.py b/remote/async_ssh.py index fb90456c..d58a51a7 100644 --- a/remote/async_ssh.py +++ b/remote/async_ssh.py @@ -1,19 +1,11 @@ """ -Async parallel SSH via the system OpenSSH client. - -This module implements :class:`~remote.remote_executor.RemoteExecutor` using -Python's stdlib ``asyncio`` and the system ``/usr/bin/ssh`` binary — no -third-party packages are required. The directory-setup and result-collection -helpers (``make_remote_dir``, ``clean_remote_dir``, ``sync_files``) are methods -on the executor, so swapping the fan-out mechanism swaps them too. - -The naming here should not be confused with the external ``asyncssh`` PyPI -package; only built-in Python packages are used. +Parallel SSH fan-out using stdlib asyncio and the system ssh binary. """ import asyncio import logging import os +from typing import Optional, Union import settings from common import expanded_node_list @@ -22,7 +14,11 @@ logger = logging.getLogger("cbt") -async def _ssh_exec_one(host, command, ssh_args): +async def _ssh_exec_one( + host: str, + command: str, + ssh_args: list[str], +) -> tuple[str, str, str, int]: proc = await asyncio.create_subprocess_exec( *ssh_args, host, command, stdout=asyncio.subprocess.PIPE, @@ -33,11 +29,15 @@ async def _ssh_exec_one(host, command, ssh_args): host, stdout_bytes.decode(errors="replace"), stderr_bytes.decode(errors="replace"), - proc.returncode, + proc.returncode if proc.returncode is not None else -1, ) -async def _ssh_exec_all(node_list, command, ssh_args): +async def _ssh_exec_all( + node_list: list[str], + command: str, + ssh_args: list[str], +) -> list[Union[tuple[str, str, str, int], BaseException]]: tasks = [_ssh_exec_one(h, command, ssh_args) for h in node_list] return await asyncio.gather(*tasks, return_exceptions=True) @@ -45,27 +45,27 @@ async def _ssh_exec_all(node_list, command, ssh_args): class AsyncSSHExecutor(RemoteExecutor): """Run commands on cluster nodes concurrently via the system ``ssh`` binary.""" - def _build_ssh_args(self): - """Build the base ``ssh`` argv shared by every node invocation. - - ``-o BatchMode=yes`` is always passed so that a host requiring - interactive authentication fails immediately rather than hanging. - The SSH user, if any, is taken from ``settings.cluster['user']``. - """ + def _build_ssh_args(self) -> list[str]: + # BatchMode=yes so a host that needs interactive auth fails fast. ssh_args = ["ssh", "-o", "BatchMode=yes", "-o", "StrictHostKeyChecking=no"] user = settings.cluster.get("user") if user: - ssh_args += ["-l", user] + ssh_args += ["-l", str(user)] return ssh_args - def run_command(self, nodes, command, continue_if_error=True): - node_list = expanded_node_list(nodes) + def run_command( + self, + nodes: str, + command: str, + continue_if_error: bool = True, + ) -> list[tuple[str, str, str, int]]: + node_list: list[str] = expanded_node_list(nodes) ssh_args = self._build_ssh_args() raw = asyncio.run(_ssh_exec_all(node_list, command, ssh_args)) - results = [] - errors = [] + results: list[tuple[str, str, str, int]] = [] + errors: list[str] = [] for item in raw: if isinstance(item, BaseException): errors.append(str(item)) @@ -78,7 +78,7 @@ def run_command(self, nodes, command, continue_if_error=True): if stderr: logger.debug("ssh [%s] stderr: %s", host, stderr.rstrip()) if exit_status != 0: - detail = (stdout or stderr).rstrip() + detail = (stderr or stdout).rstrip() msg = f"ssh [{host}] exited {exit_status}: {detail}" if not continue_if_error: errors.append(msg) @@ -93,59 +93,29 @@ def run_command(self, nodes, command, continue_if_error=True): return results - def run_command_with_error_checking(self, nodes, command): - self.run_command(nodes, command, continue_if_error=False) - - # ------------------------------------------------------------------ - # pdsh-free cluster helpers (RemoteExecutor interface) - # ------------------------------------------------------------------ - - def make_remote_dir(self, remote_dir): - """Create *remote_dir* on all cluster nodes via parallel SSH.""" - self.run_command_with_error_checking( - all_cluster_nodes(), f'mkdir -p -m0755 -- {remote_dir}' - ) - - def clean_remote_dir(self, remote_dir): - """Remove *remote_dir* from all cluster nodes via parallel SSH.""" - if remote_dir == "/" or not os.path.isabs(remote_dir): - raise SystemExit("Cleaning the remote dir doesn't seem safe, bailing.") - self.run_command_with_error_checking( - all_cluster_nodes(), - f'if [ -d "{remote_dir}" ]; then rm -rf {remote_dir}; fi', - ) - - def sync_files(self, remote_dir, local_dir): - """Pull *remote_dir* from all cluster nodes into *local_dir* via parallel ``scp -r``.""" - nodes_str = all_cluster_nodes() - node_list = expanded_node_list(nodes_str) + def sync_files(self, nodes: str, remote_dir: str, local_dir: str) -> None: + """Pull *remote_dir* from *nodes* into *local_dir* via parallel ``scp -r``.""" + node_list: list[str] = expanded_node_list(nodes) if not os.path.exists(local_dir): os.makedirs(local_dir) if 'user' in settings.cluster: self.run_command_with_error_checking( - nodes_str, + nodes, 'sudo chown -R {0}.{0} {1}'.format(settings.cluster['user'], remote_dir), ) user = settings.cluster.get("user") scp_base_args = _build_scp_args() - def _bare_host(h): - return h.split("@", 1)[-1] + raw = asyncio.run(_scp_pull_all(node_list, remote_dir, local_dir, scp_base_args, user)) - async def _pull_all(): - tasks = [ - _scp_pull_one(_bare_host(h), remote_dir, local_dir, scp_base_args, user=user) - for h in node_list - ] - return await asyncio.gather(*tasks, return_exceptions=True) - - raw = asyncio.run(_pull_all()) - - errors = [] + errors: list[str] = [] for item in raw: + # TODO: the BaseException branch duplicates the same pattern in + # run_command(); consolidate into a shared helper once the pattern + # stabilises. if isinstance(item, BaseException): errors.append(str(item)) logger.warning("scp: failed to launch process: %s", item) @@ -165,21 +135,36 @@ async def _pull_all(): # Module-level helpers used by AsyncSSHExecutor # --------------------------------------------------------------------------- -def all_cluster_nodes(): - """Return every node in the cluster (clients, osds, mons, rgws, mds). - - Single source of truth for the "all nodes" set used by the cluster-wide - helpers above, so the node groups are declared in exactly one place. - """ - return settings.getnodes('clients', 'osds', 'mons', 'rgws', 'mds') +def _bare_host(h: str) -> str: + """Strip any ``user@`` prefix, returning just the hostname.""" + return h.split("@", 1)[-1] + + +async def _scp_pull_all( + node_list: list[str], + remote_dir: str, + local_dir: str, + scp_base_args: list[str], + user: Optional[str], +) -> list[Union[tuple[str, str, str, int], BaseException]]: + tasks = [ + _scp_pull_one(_bare_host(h), remote_dir, local_dir, scp_base_args, user=str(user) if user else None) + for h in node_list + ] + return await asyncio.gather(*tasks, return_exceptions=True) -def _build_scp_args(): - """Build the base ``scp`` argv shared by every pull invocation.""" +def _build_scp_args() -> list[str]: return ["scp", "-r", "-o", "BatchMode=yes", "-o", "StrictHostKeyChecking=no"] -async def _scp_pull_one(host, remote_path, local_dir, scp_base_args, user=None): +async def _scp_pull_one( + host: str, + remote_path: str, + local_dir: str, + scp_base_args: list[str], + user: Optional[str] = None, +) -> tuple[str, str, str, int]: dest = os.path.join(local_dir, host) os.makedirs(dest, exist_ok=True) src_host = f"{user}@{host}" if user else host @@ -195,5 +180,5 @@ async def _scp_pull_one(host, remote_path, local_dir, scp_base_args, user=None): host, stdout_bytes.decode(errors="replace"), stderr_bytes.decode(errors="replace"), - proc.returncode, + proc.returncode if proc.returncode is not None else -1, ) diff --git a/remote/remote_executor.py b/remote/remote_executor.py index eab1d5db..1436cbab 100644 --- a/remote/remote_executor.py +++ b/remote/remote_executor.py @@ -1,14 +1,14 @@ """ -Abstract interface for running a shell command across a set of cluster nodes. - -CBT historically fanned commands out to remote nodes with ``pdsh`` (see -``common.py``). pdsh is being retired (tracked in -https://tracker.ceph.com/issues/80193), so the fan-out mechanism is being -moved behind this interface. The first concrete implementation is -``remote.async_ssh.AsyncSSHExecutor`` (stdlib asyncio + the system ``ssh`` -binary); a pdsh-backed implementation can follow without changing callers. +Abstract interface for fanning a command out to cluster nodes. + +pdsh is being retired (tracker.ceph.com/issues/80193); this interface lets +each benchmark migrate to a new transport without touching the others. + +TODO: add a pdsh-backed RemoteExecutor and move the fio path onto it. Not +done yet — deferred so the fio route stays untouched until it can be tested. """ +import os from abc import ABC, abstractmethod @@ -16,26 +16,43 @@ class RemoteExecutor(ABC): """A way of running a single command on one or more cluster nodes.""" @abstractmethod - def run_command(self, nodes, command, continue_if_error=True) -> list: - """Run *command* on all *nodes* in parallel. + def run_command( + self, + nodes: str, + command: str, + continue_if_error: bool = True, + ) -> list[tuple[str, str, str, int]]: + """Run command on all nodes in parallel. Return a list of ``(host, stdout, stderr, exit_status)`` tuples, one - per node. If *continue_if_error* is False and any node exits non-zero + per node. If continue_if_error is False and any node exits non-zero (or fails to launch), raise ``RuntimeError``. """ @abstractmethod - def run_command_with_error_checking(self, nodes, command) -> None: - """Run *command* on all *nodes*; raise ``RuntimeError`` if any fail.""" - - @abstractmethod - def make_remote_dir(self, remote_dir) -> None: - """Create *remote_dir* on every cluster node.""" - - @abstractmethod - def clean_remote_dir(self, remote_dir) -> None: - """Remove *remote_dir* from every cluster node.""" - - @abstractmethod - def sync_files(self, remote_dir, local_dir) -> None: - """Pull *remote_dir* from every cluster node into *local_dir*.""" + def sync_files(self, nodes: str, remote_dir: str, local_dir: str) -> None: + """Pull remote_dir from nodes into local_dir.""" + + # ------------------------------------------------------------------ + # Concrete helpers built on run_command(); transport-agnostic, so + # every subclass inherits the same implementation. + # ------------------------------------------------------------------ + + def run_command_with_error_checking(self, nodes: str, command: str) -> None: + """Run command on all nodes; raise ``RuntimeError`` if any fail.""" + self.run_command(nodes, command, continue_if_error=False) + + def make_remote_dir(self, nodes: str, remote_dir: str) -> None: + """Create remote_dir on nodes in parallel.""" + self.run_command_with_error_checking( + nodes, f'mkdir -p -m0755 -- {remote_dir}' + ) + + def clean_remote_dir(self, nodes: str, remote_dir: str) -> None: + """Remove remote_dir from nodes in parallel.""" + if remote_dir == "/" or not os.path.isabs(remote_dir): + raise SystemExit("Cleaning the remote dir doesn't seem safe, bailing.") + self.run_command_with_error_checking( + nodes, + f'if [ -d "{remote_dir}" ]; then rm -rf {remote_dir}; fi', + ) diff --git a/tests/test_async_ssh.py b/tests/test_async_ssh.py index 9e8071d8..f55ff489 100644 --- a/tests/test_async_ssh.py +++ b/tests/test_async_ssh.py @@ -122,23 +122,29 @@ def test_multinode_one_failure_names_failing_host(self, mock_exec): @patch.dict("settings.cluster", {}, clear=True) @patch("remote.async_ssh.asyncio.create_subprocess_exec", new_callable=AsyncMock) - def test_error_detail_prefers_stdout_over_stderr(self, mock_exec): - # The error detail is (stdout or stderr): when a failing command wrote - # to stdout, that text — not stderr — is what surfaces in the message. + def test_error_detail_prefers_stderr_over_stdout(self, mock_exec): mock_exec.return_value = _make_proc(returncode=1, stdout=b"stdout detail", stderr=b"stderr detail") with self.assertRaises(RuntimeError) as ctx: AsyncSSHExecutor().run_command("h1", "cmd", continue_if_error=False) - self.assertIn("stdout detail", str(ctx.exception)) - self.assertNotIn("stderr detail", str(ctx.exception)) + self.assertIn("stderr detail", str(ctx.exception)) + self.assertNotIn("stdout detail", str(ctx.exception)) @patch.dict("settings.cluster", {}, clear=True) @patch("remote.async_ssh.asyncio.create_subprocess_exec", new_callable=AsyncMock) - def test_error_detail_falls_back_to_stderr_when_stdout_empty(self, mock_exec): + def test_error_detail_uses_stderr_when_no_stdout(self, mock_exec): mock_exec.return_value = _make_proc(returncode=1, stdout=b"", stderr=b"stderr detail") with self.assertRaises(RuntimeError) as ctx: AsyncSSHExecutor().run_command("h1", "cmd", continue_if_error=False) self.assertIn("stderr detail", str(ctx.exception)) + @patch.dict("settings.cluster", {}, clear=True) + @patch("remote.async_ssh.asyncio.create_subprocess_exec", new_callable=AsyncMock) + def test_error_detail_falls_back_to_stdout_when_stderr_empty(self, mock_exec): + mock_exec.return_value = _make_proc(returncode=1, stdout=b"stdout detail", stderr=b"") + with self.assertRaises(RuntimeError) as ctx: + AsyncSSHExecutor().run_command("h1", "cmd", continue_if_error=False) + self.assertIn("stdout detail", str(ctx.exception)) + @patch.dict("settings.cluster", {}, clear=True) @patch("remote.async_ssh.asyncio.create_subprocess_exec", new_callable=AsyncMock) def test_zero_exit_with_stderr_is_not_a_failure(self, mock_exec): diff --git a/tests/test_bm_elbencho.py b/tests/test_bm_elbencho.py new file mode 100644 index 00000000..2a10251e --- /dev/null +++ b/tests/test_bm_elbencho.py @@ -0,0 +1,428 @@ +"""Unit tests for the Elbencho benchmark class. + +Validates construction, config handling, factory integration, +CLI command building, run loop fan-out, and pdsh-free lifecycle. +""" + +import tempfile +import unittest +from unittest.mock import patch + +import benchmarkfactory +import settings +from benchmark.elbencho import Elbencho +from cluster.ceph import Ceph +from remote.async_ssh import AsyncSSHExecutor + +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", + "duration": 60, + }, + "read_small": { + "s3_bucket": "cbt-benchmark", + "threads": [1, 4], + "iodepth": [1, 4], + "blocksize": ["4k"], + "size": "4g", + "num_objects": 1000, + "mode": "read", + "duration": 60, + }, + }, +} + + +class TestElbenchoDefaults(unittest.TestCase): + + 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): + self.assertIsInstance(self._make(), Elbencho) + + def test_default_cmd_path(self): + self.assertEqual("/usr/local/bin/elbencho", self._make().cmd_path) + + def test_default_auth_is_empty_dict(self): + self.assertEqual({}, self._make().auth) + + def test_no_workloads_registered_by_default(self): + self.assertFalse(self._make()._workloads.exist()) + + def test_base_run_dir_set(self): + b = self._make() + self.assertIsNotNone(b.base_run_dir) + self.assertIsInstance(b.base_run_dir, str) + + +class TestElbenchoExplicitConfig(unittest.TestCase): + + 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): + self.assertEqual("/opt/elbencho/bin/elbencho", self._make().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_registered(self): + self.assertTrue(self._make()._workloads.exist()) + + def test_workload_names_preserved(self): + names = self._make()._workloads.get_names() + self.assertIn("write_small", names) + self.assertIn("read_small", names) + + +class TestElbenchoValidation(unittest.TestCase): + + 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_missing_mode_raises(self): + with self.assertRaises(ValueError) as ctx: + self._make({"no_mode": {"s3_bucket": "test-bucket"}}) + self.assertIn("missing required key 'mode'", str(ctx.exception)) + + def test_missing_s3_bucket_raises(self): + with self.assertRaises(ValueError) as ctx: + self._make({"bad": {"mode": "write"}}) + self.assertIn("missing required key 's3_bucket'", str(ctx.exception)) + + def test_non_integer_threads_raises(self): + with self.assertRaises(ValueError) as ctx: + self._make({"bad": {"mode": "write", "s3_bucket": "b", "threads": ["bad_value"]}}) + self.assertIn("threads value 'bad_value' is not an integer", str(ctx.exception)) + + def test_non_integer_iodepth_raises(self): + with self.assertRaises(ValueError) as ctx: + self._make({"bad": {"mode": "write", "s3_bucket": "b", "iodepth": "not_a_number"}}) + self.assertIn("iodepth value 'not_a_number' is not an integer", str(ctx.exception)) + + def test_string_integer_threads_accepted(self): + # YAML may deserialise quoted integers as strings — "4" should be valid, + # so construction must succeed without raising. + b = self._make({"w": {"mode": "write", "s3_bucket": "b", "threads": ["4", 16]}}) + self.assertTrue(b._workloads.exist()) + + def test_mixed_valid_invalid_threads_raises_on_bad_item(self): + with self.assertRaises(ValueError) as ctx: + self._make({"w": {"mode": "write", "s3_bucket": "b", "threads": [1, 4, "oops"]}}) + self.assertIn("threads value 'oops' is not an integer", str(ctx.exception)) + + +class TestElbenchoExists(unittest.TestCase): + + 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() + 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 TestBenchmarkFactoryIntegration(unittest.TestCase): + + 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): + 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)) + + +# --------------------------------------------------------------------------- +# Run-loop tests +# --------------------------------------------------------------------------- + +class TestRunLoop(unittest.TestCase): + + archive_dir = "/tmp" + + @classmethod + def setUpClass(cls): + settings.mock_initialize(config_file=INVARIANT_YAML) + cls.cluster = Ceph.mockinit(settings.cluster) + + def setUp(self): + # _run_workloads() wraps each command group in monitoring; stub it so + # the run-loop tests don't touch real monitors. + for target in ("monitoring.start", "monitoring.stop"): + patcher = patch(target) + patcher.start() + self.addCleanup(patcher.stop) + + def _make(self, workloads: dict) -> Elbencho: + cfg = dict( + _MINIMAL_CONFIG, + cmd_path="/usr/local/bin/elbencho", + auth={}, + workloads=workloads, + ) + b = benchmarkfactory.get_object(self.archive_dir, self.cluster, "elbencho", cfg) + assert isinstance(b, Elbencho) + return b + + @patch.object(AsyncSSHExecutor, "make_remote_dir") + @patch.object(AsyncSSHExecutor, "run_command") + def test_call_count_matches_run_matrix(self, mock_exec, mock_mkdir): + """2 blocksizes x 2 threads x 2 iodepths = 8 run_command calls.""" + mock_exec.return_value = [] + 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_exec.call_count) + + @patch.object(AsyncSSHExecutor, "make_remote_dir") + @patch.object(AsyncSSHExecutor, "run_command") + def test_run_dir_path_structure(self, mock_exec, mock_mkdir): + mock_exec.return_value = [] + b = self._make({ + "w": { + "s3_bucket": "bkt", + "mode": "write", + "blocksize": ["4k"], + "threads": [16], + "iodepth": [4], + } + }) + b._run_workloads() + mkdir_calls = [c.args[1] for c in mock_mkdir.call_args_list] + self.assertTrue( + any("elbencho/write_4096/threads-016/iodepth-004" in d for d in mkdir_calls), + f"Expected path segment not found in: {mkdir_calls}", + ) + + @patch.object(AsyncSSHExecutor, "make_remote_dir") + @patch.object(AsyncSSHExecutor, "run_command") + def test_stat_workload_skipped(self, mock_exec, mock_mkdir): + mock_exec.return_value = [] + b = self._make({ + "w": {"s3_bucket": "bkt", "mode": "stat", "threads": [1], "iodepth": [1]} + }) + b._run_workloads() + mock_exec.assert_not_called() + + @patch.object(AsyncSSHExecutor, "make_remote_dir") + @patch.object(AsyncSSHExecutor, "run_command") + def test_scalar_threads_and_iodepth(self, mock_exec, mock_mkdir): + mock_exec.return_value = [] + b = self._make({ + "w": { + "s3_bucket": "bkt", + "mode": "write", + "blocksize": "4k", + "threads": 4, + "iodepth": 2, + } + }) + b._run_workloads() + self.assertEqual(1, mock_exec.call_count) + + @patch.object(AsyncSSHExecutor, "make_remote_dir") + @patch.object(AsyncSSHExecutor, "run_command") + def test_cell_emits_matching_command_and_byte_run_dir(self, mock_exec, mock_mkdir): + # The run loop must feed elbencho the human blocksize (128k) while + # naming the run directory with the byte count (131072). Mixing the two + # up is a real regression risk, so pin both from a single cell. + mock_exec.return_value = [] + b = self._make({ + "w": { + "s3_bucket": "bkt", + "mode": "write", + "blocksize": ["128k"], + "threads": [8], + "iodepth": [16], + "size": "4g", + } + }) + b._run_workloads() + + self.assertEqual(1, mock_exec.call_count) + cmd = mock_exec.call_args.args[1] + self.assertIn("--block 128k", cmd) + self.assertNotIn("--block 131072", cmd) # byte count must not reach --block + self.assertIn("--threads 8", cmd) + self.assertIn("--iodepth 16", cmd) + self.assertIn("--size 4g", cmd) + self.assertTrue(cmd.endswith("s3://bkt")) + + run_dirs = [c.args[1] for c in mock_mkdir.call_args_list] + self.assertTrue( + any("elbencho/write_131072/threads-008/iodepth-016" in d for d in run_dirs), + f"byte-based run dir not found in: {run_dirs}", + ) + + +class TestElbenchoNoPdsh(unittest.TestCase): + + archive_dir = "/tmp" + + @classmethod + def setUpClass(cls): + settings.mock_initialize(config_file=INVARIANT_YAML) + cls.cluster = Ceph.mockinit(settings.cluster) + + def _make(self) -> Elbencho: + cfg = dict( + _MINIMAL_CONFIG, + cmd_path="/usr/local/bin/elbencho", + auth={}, + workloads={}, + ) + b = benchmarkfactory.get_object(self.archive_dir, self.cluster, "elbencho", cfg) + assert isinstance(b, Elbencho) + return b + + @patch.object(AsyncSSHExecutor, "make_remote_dir") + @patch.object(AsyncSSHExecutor, "clean_remote_dir") + def test_cleandir_uses_async_helpers(self, mock_clean, mock_mkdir): + b = self._make() + b.cleandir() + clients = mock_clean.call_args.args[0] + self.assertEqual(mock_clean.call_args.args[1], b.run_dir) + self.assertEqual(mock_mkdir.call_args.args[0], clients) + self.assertEqual(mock_mkdir.call_args.args[1], b.run_dir) + + @patch.object(AsyncSSHExecutor, "run_command") + def test_dropcaches_uses_remote_executor(self, mock_exec): + mock_exec.return_value = [] + b = self._make() + b.dropcaches() + self.assertEqual(2, mock_exec.call_count) + commands = [c.args[1] for c in mock_exec.call_args_list] + self.assertIn("sync", commands) + self.assertTrue(any("drop_caches" in cmd for cmd in commands)) + + @patch.object(AsyncSSHExecutor, "sync_files") + @patch.object(AsyncSSHExecutor, "make_remote_dir") + @patch.object(AsyncSSHExecutor, "run_command") + def test_run_does_not_call_pdsh(self, mock_exec, mock_mkdir, mock_sync): + import common as _common + mock_exec.return_value = [] + b = self._make() + with patch.object(b, "dropcaches"), \ + patch.object(b.cluster, "dump_config"), \ + patch.object(b.cluster, "set_osd_param"), \ + patch("monitoring.start"), \ + patch("monitoring.stop"), \ + patch.object(b, "_run_workloads"): + with patch.object(_common, "pdsh", side_effect=AssertionError("pdsh called")): + b.run() + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_elbencho_command.py b/tests/test_elbencho_command.py new file mode 100644 index 00000000..2b7b68b9 --- /dev/null +++ b/tests/test_elbencho_command.py @@ -0,0 +1,198 @@ +"""Unit tests for ElbenchoCommand (command/elbencho_command.py). + +These tests pin the exact elbencho command line produced for a run cell, the +exact S3 auth-flag translation, and blocksize parsing — including the negative +cases (optional flags omitted when unset) and error/edge conditions. +""" + +import unittest + +from command.elbencho_command import ElbenchoCommand + +_AUTH = {"config": "access_key=AK;secret_key=SK;url=http://rgw:7480"} + + +def _cmd(workload, bs="4k", threads=1, iodepth=1, run_dir="/tmp/run", + auth=_AUTH, cmd_path="/usr/local/bin/elbencho", set_executable=True): + # Auth reaches the command as flat string options (as it does through the + # Workloads pipeline), not a separate constructor argument. + options = { + **workload, + "blocksize": bs, + "threads": threads, + "iodepth": iodepth, + "s3_auth_config": auth.get("config", ""), + "s3_session_token": auth.get("s3_session_token", ""), + } + command = ElbenchoCommand(options, run_dir) + if set_executable: + command.set_executable(cmd_path) + return command.get() + + +class TestParseBlocksizeToBytes(unittest.TestCase): + + def test_valid_conversions(self): + cases = { + "4k": 4096, + "128k": 131072, + "1m": 1048576, + "1g": 1073741824, + "512": 512, # no suffix -> raw bytes + "4K": 4096, # case-insensitive + "2M": 2097152, + "1.5k": 1536, # fractional multiplier + " 4k ": 4096, # surrounding whitespace tolerated + } + for text, expected in cases.items(): + with self.subTest(blocksize=text): + self.assertEqual(expected, ElbenchoCommand.parse_blocksize_to_bytes(text)) + + def test_invalid_forms_raise(self): + for text in ["", "abc", "4kb", "1.2.3k", "k", "-4k", "4 k", "4tb"]: + with self.subTest(blocksize=text): + with self.assertRaises(ValueError): + ElbenchoCommand.parse_blocksize_to_bytes(text) + + +class TestBuildAuthFlags(unittest.TestCase): + + def test_full_config_exact_order(self): + # retry=9 is an unknown key and must be dropped; the emitted flags must + # be endpoints -> key -> secret in that order. + flags = ElbenchoCommand.build_auth_flags( + {"config": "access_key=AK;secret_key=SK;url=http://rgw:7480;retry=9"} + ) + self.assertEqual( + ["--s3endpoints", "http://rgw:7480", "--s3key", "AK", "--s3secret", "SK"], + flags, + ) + + def test_config_without_url_omits_endpoints(self): + self.assertEqual( + ["--s3key", "AK", "--s3secret", "SK"], + ElbenchoCommand.build_auth_flags({"config": "access_key=AK;secret_key=SK"}), + ) + + def test_config_url_only(self): + self.assertEqual( + ["--s3endpoints", "http://rgw:7480"], + ElbenchoCommand.build_auth_flags({"config": "url=http://rgw:7480"}), + ) + + def test_malformed_entries_are_skipped(self): + # "garbage" has no '=' and must be ignored rather than crash. + self.assertEqual( + ["--s3key", "AK", "--s3secret", "SK"], + ElbenchoCommand.build_auth_flags({"config": "access_key=AK;garbage;secret_key=SK"}), + ) + + def test_session_token_appended_after_config(self): + self.assertEqual( + ["--s3endpoints", "http://rgw:7480", "--s3key", "AK", "--s3secret", "SK", + "--s3authtoken", "tok"], + ElbenchoCommand.build_auth_flags( + {"config": "access_key=AK;secret_key=SK;url=http://rgw:7480", "s3_session_token": "tok"} + ), + ) + + def test_token_only(self): + self.assertEqual(["--s3authtoken", "tok"], + ElbenchoCommand.build_auth_flags({"s3_session_token": "tok"})) + + def test_empty_auth_and_empty_config_emit_nothing(self): + self.assertEqual([], ElbenchoCommand.build_auth_flags({})) + self.assertEqual([], ElbenchoCommand.build_auth_flags({"config": ""})) + + +class TestGenerateFullCommand(unittest.TestCase): + + def test_minimal_write_command_exact(self): + cmd = _cmd({"s3_bucket": "bkt", "mode": "write"}) + self.assertEqual( + "/usr/local/bin/elbencho --write --threads 1 --block 4k --iodepth 1 " + "--s3endpoints http://rgw:7480 --s3key AK --s3secret SK " + "--s3region default " + "--resfile /tmp/run/elbencho/write_4096/threads-001/iodepth-001/result.csv s3://bkt", + cmd, + ) + + def test_all_options_command_exact(self): + # Every optional field set, plus a session token, at a custom exe/run dir. + # Pins flag ordering end-to-end. + workload = { + "s3_bucket": "bkt", "mode": "write", "size": "4g", "num_objects": 100, + "num_dirs": 4, "duration": 30, "deldirs": True, "s3nompcheck": True, + "hosts": "c1,c2", "s3_region": "us-east-1", "mkdirs": True, + } + cmd = _cmd( + workload, bs="128k", threads=8, iodepth=16, run_dir="/tmp/myrun", + cmd_path="/opt/elbencho", + auth={"config": "access_key=AK;secret_key=SK;url=http://rgw:7480", "s3_session_token": "tok"}, + ) + self.assertEqual( + "/opt/elbencho --write --threads 8 --block 128k --iodepth 16 " + "--size 4g --files 100 --dirs 4 --timelimit 30 --deldirs --s3nompcheck " + "--hosts c1,c2 --s3endpoints http://rgw:7480 --s3key AK --s3secret SK " + "--s3authtoken tok --s3region us-east-1 " + "--resfile /tmp/myrun/elbencho/write_131072/threads-008/iodepth-016/result.csv " + "--mkdirs s3://bkt", + cmd, + ) + + def test_optional_flags_absent_when_unset(self): + # A minimal workload must not leak flags for options the user didn't set + # (e.g. "--files None" or an always-on "--mkdirs"). + cmd = _cmd({"s3_bucket": "bkt", "mode": "write"}) + for flag in ["--size", "--files", "--dirs", "--timelimit", "--deldirs", + "--s3nompcheck", "--hosts", "--mkdirs", "--s3authtoken"]: + with self.subTest(flag=flag): + self.assertNotIn(flag, cmd) + + def test_num_objects_zero_still_emits_files_flag(self): + # 0 is a meaningful value: the builder keys off "is not None", so + # num_objects=0 must produce "--files 0", not be dropped as falsy. + cmd = _cmd({"s3_bucket": "bkt", "mode": "write", "num_objects": 0}) + self.assertIn("--files 0", cmd) + + def test_block_uses_raw_blocksize_not_bytes(self): + # The command must pass elbencho the human blocksize (128k), never the + # byte count used elsewhere for directory naming. + cmd = _cmd({"s3_bucket": "bkt", "mode": "write"}, bs="128k") + self.assertIn("--block 128k", cmd) + self.assertNotIn("--block 131072", cmd) + + def test_write_mode_is_exclusive(self): + cmd = _cmd({"s3_bucket": "bkt", "mode": "write"}) + self.assertIn("--write", cmd) + self.assertNotIn("--read", cmd) + + def test_read_mode_is_exclusive(self): + cmd = _cmd({"s3_bucket": "bkt", "mode": "read"}) + self.assertIn("--read", cmd) + self.assertNotIn("--write", cmd) + + def test_readwrite_emits_both_flags_in_order(self): + cmd = _cmd({"s3_bucket": "bkt", "mode": "readwrite"}) + self.assertIn("--write --read", cmd) + + def test_unknown_mode_raises_naming_the_mode(self): + with self.assertRaises(ValueError) as ctx: + _cmd({"s3_bucket": "bkt", "mode": "bogus"}) + self.assertIn("bogus", str(ctx.exception)) + + def test_unsupported_modes_return_empty_string(self): + # stat/list are recognised by elbencho but not yet formatter-supported, + # so the builder yields "" and the run loop skips them. + for mode in ["stat", "list"]: + with self.subTest(mode=mode): + self.assertEqual("", _cmd({"s3_bucket": "bkt", "mode": mode})) + + def test_no_executable_returns_empty_string(self): + # Without set_executable() there is nothing to run; get() must refuse + # rather than emit a command starting with "None". + self.assertEqual("", _cmd({"s3_bucket": "bkt", "mode": "write"}, set_executable=False)) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_workload.py b/tests/test_workload.py index 173a8d51..d3878add 100644 --- a/tests/test_workload.py +++ b/tests/test_workload.py @@ -226,6 +226,24 @@ def test_create_command_class_rbdfio(self) -> None: self.assertIsNotNone(command) self.assertIsInstance(command, Command) + def test_create_command_class_elbencho(self) -> None: + """Test creating ElbenchoCommand for elbencho benchmark""" + workload = self._create_workload() + workload.set_benchmark_type("elbencho") + + options = { + "mode": "write", + "s3_bucket": "bkt", + "blocksize": "4k", + "threads": "1", + "iodepth": "1", + } + + command = workload._create_command_class(options) + + self.assertIsNotNone(command) + self.assertIsInstance(command, Command) + def test_create_command_class_unsupported(self) -> None: """Test creating command for unsupported benchmark type""" workload = self._create_workload() @@ -250,8 +268,6 @@ def test_workload_str_representation(self) -> None: self.assertIn(self.workload_name, str_repr) self.assertIn("Name:", str_repr) - - if __name__ == "__main__": unittest.main() diff --git a/workloads/workload.py b/workloads/workload.py index e28070e2..827075fb 100644 --- a/workloads/workload.py +++ b/workloads/workload.py @@ -8,6 +8,7 @@ from typing import Optional from command.command import Command +from command.elbencho_command import ElbenchoCommand from command.rbd_fio_command import RbdFioCommand from common import all_configs # pyright: ignore[reportUnknownVariableType] from workloads.workload_types import WorkloadType @@ -132,6 +133,11 @@ def _create_command_class(self, options: dict[str, str]) -> Command: if self._parent_benchmark_type == "rbdfio": return RbdFioCommand(options, f"{self._base_run_directory}{self._name}") + if self._parent_benchmark_type == "elbencho": + # Elbencho builds its own output path (mode/blocksize-bytes/threads/ + # iodepth) from the base run directory, with no workload-name segment. + return ElbenchoCommand(options, self._base_run_directory) + log.error("Benchmark Class %s is not supported by workloads yet", self._parent_benchmark_type) raise NotImplementedError diff --git a/workloads/workloads.py b/workloads/workloads.py index 877285c5..70b6045c 100644 --- a/workloads/workloads.py +++ b/workloads/workloads.py @@ -2,6 +2,7 @@ The workloads class that contains all the Workloads for a given Benchmark run """ +from collections.abc import Generator from logging import Logger, getLogger from time import sleep from typing import Optional, Union @@ -101,6 +102,32 @@ def run(self) -> None: log.info("== Workloads completed ==") + def command_groups(self) -> Generator[tuple[str, list[str]], None, None]: + """Yield (output_directory, [command, ...]) for every run cell without executing. + + This factors the (pdsh-free) command-generation bookkeeping out of + run() so a caller can choose how to fan the commands out — e.g. via a + remote.RemoteExecutor for benchmarks that have moved off pdsh. + + set_benchmark_type() and set_executable() must be called first. + + TODO: pre_workload_script and ramp_time are not yet yielded here. + TODO: collapse with run() once the rbdfio path is also executor-driven, + so we have a single place where we set up and execute command lists. + """ + if not self._benchmark_type: + log.error("Benchmark type has not been set, set_benchmark_type() must be called first.") + return + + if not self._executable: + log.error("Executable path has not been set, set_executable() must be called first.") + return + + for workload in self._workloads: + workload.set_benchmark_type(self._benchmark_type) + workload.set_executable(self._executable) + yield from workload.get_commands_list() + def get_names(self) -> str: """ Get the names for all the workloads From 3ecc16d2e8416ef02afa498c206d3c4bd1a077b1 Mon Sep 17 00:00:00 2001 From: Kenan Al-Shamie Date: Fri, 4 Sep 2026 09:48:36 +0100 Subject: [PATCH 3/3] add Elbencho S3 user documentation User-facing guide for running the Elbencho S3 benchmark via CBT: the test-plan YAML structure, the blocksize/size interaction (single-PUT vs multipart upload and the 5 MB minimum part size), running instructions, and expected result output. Links Elbencho from docs/Workloads.md. Signed-off-by: Kenan Al-Shamie Assisted-by: Claude-v2.1.212:claude-opus-4-8 --- docs/Workloads.md | 3 +- docs/workloads/elbencho-s3.md | 184 ++++++++++++++++++++++++++ example/wip-elbencho/elbencho_ex.yaml | 40 ++++++ 3 files changed, 226 insertions(+), 1 deletion(-) create mode 100644 docs/workloads/elbencho-s3.md create mode 100644 example/wip-elbencho/elbencho_ex.yaml diff --git a/docs/Workloads.md b/docs/Workloads.md index 81d1f2b0..784874b6 100644 --- a/docs/Workloads.md +++ b/docs/Workloads.md @@ -7,7 +7,8 @@ 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`. For Elbencho-specific +usage, see [docs/workloads/elbencho-s3.md](workloads/elbencho-s3.md). ![workloads](./workloads.png) diff --git a/docs/workloads/elbencho-s3.md b/docs/workloads/elbencho-s3.md new file mode 100644 index 00000000..e89f0e62 --- /dev/null +++ b/docs/workloads/elbencho-s3.md @@ -0,0 +1,184 @@ +# 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=;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: ['5m', '16m'] + size: '5g' + num_objects: 100 + duration: 30 + + read_small: + s3_bucket: 'cbt-benchmark' + mode: 'read' + threads: [1, 4] + iodepth: [1, 4] + blocksize: ['5m', '16m'] + size: '5g' + num_objects: 100 + duration: 30 +``` + +Replace ``, ``, and the RGW URL with your cluster's values. +Your RGW user should be setup already as a pre-requisite to this process. To get your username, +simply run `radosgw-admin user list`, and to see your access/secret key, run + +```bash +radosgw-admin user info --uid="" +``` + +### How `blocksize` and `size` control S3 uploads + +In S3 mode, `blocksize` is the **multipart upload part size** and `size` is the **per-object +size**. Elbencho decides between single-PUT and multipart upload based on their relationship: + +- `blocksize >= size` → **single PutObject** (one HTTP PUT per object, no part-size constraints) +- `blocksize < size` → **multipart upload** (each part = `blocksize` bytes, parts per object = `ceil(size / blocksize)`) + +When using multipart upload, S3 requires each part (except the last) to be **at least 5 MB**. +If `blocksize` is smaller than 5 MB in a multipart scenario, RGW rejects the upload. + +With the example config above (`blocksize: ['5m', '16m']`, `size: '5g'`): +- `5m` part size → `ceil(5g / 5m)` = **1024 parts per object** (multipart) +- `16m` part size → `ceil(5g / 16m)` = **320 parts per object** (multipart) + +To test single-PUT performance, set `blocksize >= size` (e.g. `blocksize: '64m'`, `size: '64m'`). + +**Read workloads must match write parameters.** Elbencho locates objects by the same +`blocksize` and `size` used at write time. A mismatch causes HTTP 416 (Range Not Satisfiable). + +```yaml +# Single-PUT: one HTTP request per object, no multipart overhead +blocksize: ['64m'] +size: '64m' + +# Multipart: 320 parts per object +blocksize: ['16m'] +size: '5g' + +# Wrong — blocksize too small for multipart (< 5 MB minimum part size) +blocksize: ['4k'] +size: '1m' + +# Wrong — read doesn't match write +write_small: + blocksize: ['5m'] + size: '5g' +read_small: + blocksize: ['8m'] # ← 416, doesn't match write + size: '10g' # ← 416, doesn't match write + +# Correct — read matches write exactly +write_small: + blocksize: ['5m', '16m'] + size: '5g' +read_small: + blocksize: ['5m', '16m'] # identical to write + size: '5g' # identical to write +``` + +## Running + +```bash +PYTHONPATH=/cbt 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 parallel SSH +2. Expand each workload's list-valued params (`blocksize`, `threads`, `iodepth`) into the cartesian product of run cells through the shared Workloads pipeline. `blocksize` sets the multipart part size, `threads` sets the number of independent S3 client workers, and `iodepth` sets the number of concurrent async HTTP requests per thread. +3. Fan out one elbencho process per client node via parallel SSH for each run cell +4. Sync results back to the archive directory via `scp -r` when complete + +The expected log output for each run cell looks like: + +``` +INFO - Elbencho: running 1 command(s) → /tmp/cbt/00000000/Elbencho/write_131072/threads-004/iodepth-004 +DEBUG - ssh [client1] exit=0 +DEBUG - ssh [client1] stdout: OPERATION RESULT TYPE ... +=========== ================ ========== ========= +MKBUCKETS Elapsed time : 26ms 26ms + Buckets/s : 38 38 + Buckets total : 1 1 +--- +WRITE Elapsed time : 16.248s 16.501s + Objects/s : 24 24 + Throughput MiB/s : 122 121 + Total MiB : 1985 2000 + Objects total : 395 400 +--- +INFO - Elbencho: all workloads complete. +``` + +With 2 blocksizes × 2 thread values × 2 iodepth values × 2 workloads (write + read), this +configuration produces **16 run cells**. + +## Expected result files + +Each run cell produces a `result.csv`. Results are pulled from each client node via `scp -r` +into a per-hostname subdirectory under the archive, mirroring the remote path: + +``` +/tmp/cbt-results//tmp/cbt/00000000/Elbencho/ + 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/example/wip-elbencho/elbencho_ex.yaml b/example/wip-elbencho/elbencho_ex.yaml new file mode 100644 index 00000000..7116a58b --- /dev/null +++ b/example/wip-elbencho/elbencho_ex.yaml @@ -0,0 +1,40 @@ +cluster: + user: 'cbt' + head: "cadmin" + clients: ["cadmin"] + osds: ["inf1", "inf2", "inf3"] + rgws: ["inf1", "inf2", "inf3"] + 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=;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: ['5m', '16m'] + size: '5g' + num_objects: 100 + duration: 30 + + read_small: + s3_bucket: 'cbt-benchmark' + mode: 'read' + threads: [1, 4] + iodepth: [1, 4] + blocksize: ['5m', '16m'] + size: '5g' + num_objects: 100 + duration: 30