From 01205af6677cb624b2729ac3c1c823fb460206c0 Mon Sep 17 00:00:00 2001 From: Leonardo Forchini Date: Fri, 25 Sep 2026 09:11:23 +0000 Subject: [PATCH] Add component-tests for controller logic Add component tests which ensure the controller enforces configured limits such as vCPU cap and thread bounds. Also include the fixtures necessary for these tests. Signed-off-by: Leonardo Forchini --- .github/workflows/pull_request.yaml | 2 +- .gitignore | 1 + Makefile | 8 +- pytest.ini | 3 + tests/component/conftest.py | 462 ++++++++++++++++++ tests/component/test_controller.py | 206 ++++++++ .../test_runaway_scale_regression.py | 24 +- 7 files changed, 689 insertions(+), 17 deletions(-) create mode 100644 pytest.ini create mode 100644 tests/component/conftest.py create mode 100644 tests/component/test_controller.py diff --git a/.github/workflows/pull_request.yaml b/.github/workflows/pull_request.yaml index bee3d37..a6e81b5 100644 --- a/.github/workflows/pull_request.yaml +++ b/.github/workflows/pull_request.yaml @@ -25,5 +25,5 @@ jobs: key: ${{ runner.os }}-${{ matrix.toolchain }}-${{ hashFiles('**/Cargo.lock') }} restore-keys: | ${{ runner.os }}-${{ matrix.toolchain }}- - - run: sudo apt-get install libvirt-dev fuse3 + - run: sudo apt-get install libvirt-dev fuse3 dbus python3-pytest python3-dbus - run: make pre-push diff --git a/.gitignore b/.gitignore index ea8c4bf..30e6c96 100644 --- a/.gitignore +++ b/.gitignore @@ -1 +1,2 @@ /target +/tests/component/__pycache__ diff --git a/Makefile b/Makefile index af4fc30..ff3104a 100644 --- a/Makefile +++ b/Makefile @@ -48,8 +48,14 @@ clean: unit-test: RUST_BACKTRACE=1 cargo test --all-features +.PHONY: component-test +component-test: + cargo build --locked --no-default-features \ + --features threshold-engine,fake-backend,mockfs + python3 -m pytest tests/component -q + .PHONY: test -test: unit-test +test: unit-test component-test .PHONY: pre-push pre-push: check test diff --git a/pytest.ini b/pytest.ini new file mode 100644 index 0000000..3ed7ec0 --- /dev/null +++ b/pytest.ini @@ -0,0 +1,3 @@ +[pytest] +pythonpath = tests/component +testpaths = tests/component diff --git a/tests/component/conftest.py b/tests/component/conftest.py new file mode 100644 index 0000000..d33d8fc --- /dev/null +++ b/tests/component/conftest.py @@ -0,0 +1,462 @@ +# SPDX-License-Identifier: BSD-3-Clause +# Copyright (c) 2026 Nutanix, Inc. All rights reserved. +# +# Author: Leonardo Forchini + +"""Local component-test harness.""" + +import fcntl +import json +import os +import signal +import subprocess +import time +from pathlib import Path + +import dbus +import pytest + +REPO_ROOT = Path(__file__).resolve().parents[2] +DEFAULT_VM = "vm-a" +CONFIG_FILE = "/etc/io-thread-controller.json" +ENGINE_DIR = "/etc/io-thread-controller.d/engines" +BACKEND_DIR = "/etc/io-thread-controller.d/backends" +STATE_DIR = "/var/fake" +OWNERSHIP_FILE = "/run/io-thread-controller/vm-ownership.json" +MOCKFS_BUS_NAME = "com.nutanix.mockfs1" +MOCKFS_OBJECT_PATH = "/com/nutanix/mockfs1" +MOCKFS_INTERFACE = "com.nutanix.mockfs1" +POLL_S = 0.2 + + +def wait_for(predicate, timeout, description): + """Poll ``predicate`` until it is true or ``timeout`` seconds elapse.""" + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if predicate(): + return + time.sleep(0.05) + raise AssertionError("timed out waiting for %s" % (description,)) + + +def _physical(root, logical): + return root / logical.lstrip("/") + + +def _target_dir(): + raw = subprocess.check_output( + ["cargo", "metadata", "--format-version", "1", "--no-deps", "--offline"], + cwd=REPO_ROOT, + text=True, + ) + return Path(json.loads(raw)["target_directory"]) + + +def _write_json(path, payload): + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(payload)) + + +class FakeBackend: + """JSON VM directory the ``fake`` backend discovers.""" + + def __init__(self, root): + self.root = root + _physical(root, STATE_DIR).mkdir(parents=True, exist_ok=True) + self.state_dir = _physical(root, STATE_DIR) + self.default_vm = DEFAULT_VM + self.add_vm(DEFAULT_VM) + + def add_vm(self, vm_id, **fields): + defaults = { + "thread_count": 1, + "vcpu_count": 1, + "per_thread_util": 0.0, + "managed": True, + "fail_snapshot": False, + "read_io_count": 0, + "write_io_count": 0, + "other_io_count": 0, + } + defaults.update(fields) + self._update(vm_id, **defaults) + return vm_id + + def set_util(self, util, vm=None): + self._update(vm or self.default_vm, per_thread_util=util) + + def set_threads(self, count, vm=None): + self._update(vm or self.default_vm, thread_count=count) + + def set_vcpu_count(self, count, vm=None): + self._update(vm or self.default_vm, vcpu_count=count) + + def fail_snapshot(self, failed=True, vm=None): + self._update(vm or self.default_vm, fail_snapshot=failed) + + def set_managed(self, managed, vm=None): + self._update(vm or self.default_vm, managed=managed) + + def set_io_counts(self, read=0, write=0, other=0, vm=None): + """Set cumulative read, write, and other I/O counts.""" + self._update( + vm or self.default_vm, + read_io_count=read, + write_io_count=write, + other_io_count=other, + ) + + def thread_count(self, vm=None): + return self._read(vm or self.default_vm)["thread_count"] + + def clear_calls(self): + path = self.state_dir / "calls.json" + if path.exists(): + path.unlink() + + def calls(self, vm=None): + """Thread-count targets ``set_thread_count`` recorded, in order.""" + vm_id = vm or self.default_vm + path = self.state_dir / "calls.json" + if not path.exists(): + return [] + recorded = [] + for line in path.read_text().splitlines(): + if not line.strip(): + continue + try: + item = json.loads(line) + except json.JSONDecodeError: + continue + if item.get("vm") == vm_id: + recorded.append(item["threads"]) + return recorded + + def _vm_path(self, vm_id): + return self.state_dir / ("%s.json" % vm_id) + + def _read(self, vm_id): + return json.loads(self._vm_path(vm_id).read_text()) + + def _update(self, vm_id, **fields): + path = self._vm_path(vm_id) + lock_path = path.with_suffix(".lock") + path.parent.mkdir(parents=True, exist_ok=True) + with open(lock_path, "a+") as lock: + fcntl.flock(lock, fcntl.LOCK_EX) + data = {} + if path.exists(): + data = json.loads(path.read_text()) + data.update(fields) + temporary = path.with_suffix(".json.tmp-pytest") + temporary.write_text(json.dumps(data)) + os.replace(temporary, path) + fcntl.flock(lock, fcntl.LOCK_UN) + + +def _prepare(fake_backend, threads, util, io=0, vcpu=16): + """Set the default VM's threads, utilisation, vCPU cap, and read I/O.""" + fake_backend.set_threads(threads) + fake_backend.set_vcpu_count(vcpu) + fake_backend.set_util(util) + fake_backend.set_io_counts(io, 0, 0) + + +def _setup_fake_backend(root): + """Write ``fake.json`` so the daemon reads this tree's VM directory.""" + _write_json( + _physical(root, LOGICAL_BACKEND_DIR) / "fake.json", + {"state_dir": LOGICAL_STATE_DIR}, + ) + + +class DBusServer: + def __init__(self, root): + self._socket = root / "dbus.sock" + self._config = root / "dbus.xml" + self._config.write_text( + """ + + system + unix:path=%s + EXTERNAL + + + + + + +""" + % self._socket + ) + self._proc = None + self.address = None + + def start(self): + self._proc = subprocess.Popen( + [ + "dbus-daemon", + "--nofork", + "--config-file=%s" % self._config, + "--print-address", + ], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + assert self._proc.stdout is not None + self.address = self._proc.stdout.readline().strip() + if not self.address: + raise RuntimeError("dbus-daemon did not print an address") + deadline = time.monotonic() + 5 + while time.monotonic() < deadline: + if self._socket.exists(): + return + if self._proc.poll() is not None: + raise RuntimeError("dbus-daemon exited before listening") + time.sleep(0.01) + raise RuntimeError("dbus socket %s did not appear" % self._socket) + + def stop(self): + if self._proc is None or self._proc.poll() is not None: + return + self._proc.send_signal(signal.SIGTERM) + try: + self._proc.wait(timeout=5) + except subprocess.TimeoutExpired: + self._proc.kill() + self._proc.wait(timeout=5) + + +class MockProc: + """mockfsd mounted at ``/proc`` and controlled over D-Bus. + + The process claims ``com.nutanix.mockfs1`` on the harness system bus. + ``Mount`` exposes the FUSE tree; ``RegisterSequence`` feeds ``/stat``. + """ + + def __init__(self, root, binary, dbus_server): + self.mount = root / "proc" + self.mount.mkdir(parents=True, exist_ok=True) + self._proc = None + self._bus = None + self._iface = None + self._stderr_path = root / "mockfsd.stderr" + self._stderr = open(self._stderr_path, "w") + try: + env = os.environ.copy() + env["DBUS_SYSTEM_BUS_ADDRESS"] = dbus_server.address + self._proc = subprocess.Popen( + [str(binary)], + stdout=subprocess.PIPE, + stderr=self._stderr, + text=True, + env=env, + ) + assert self._proc.stdout is not None + name = self._proc.stdout.readline().strip() + if self._proc.poll() is not None or name != MOCKFS_BUS_NAME: + raise RuntimeError( + "mockfsd did not claim %s (printed %r): %s" + % (MOCKFS_BUS_NAME, name, self._stderr_text()) + ) + self._bus = dbus.bus.BusConnection(dbus_server.address) + self._iface = dbus.Interface( + self._bus.get_object(MOCKFS_BUS_NAME, MOCKFS_OBJECT_PATH), + MOCKFS_INTERFACE, + ) + wait_for(self._healthy, timeout=5, description="mockfsd ping") + self._iface.Mount(str(self.mount)) + except Exception: + self.stop() + raise + + def _stderr_text(self): + self._stderr.flush() + return self._stderr_path.read_text(errors="replace").strip() + + def _healthy(self): + if self._proc.poll() is not None: + raise RuntimeError("mockfsd exited during startup: %s" % self._stderr_text()) + try: + return self._iface.Ping() == "ok" + except dbus.exceptions.DBusException: + return False + + def register_sequence(self, path, values): + """Register successive file bodies for ``path`` on this mount.""" + payload = dbus.Array( + [dbus.ByteArray(bytes(value)) for value in values], + signature="ay", + ) + self._iface.RegisterSequence(str(self.mount), path, payload) + + def stop(self): + if self._iface is not None and self._proc is not None and self._proc.poll() is None: + try: + self._iface.Unmount(str(self.mount)) + except dbus.exceptions.DBusException: + pass + if self._proc is not None and self._proc.poll() is None: + self._proc.send_signal(signal.SIGINT) + try: + self._proc.wait(timeout=5) + except subprocess.TimeoutExpired: + self._proc.kill() + self._proc.wait(timeout=5) + if self._stderr is not None and not self._stderr.closed: + self._stderr.close() + + +class Controller: + """Starts ``io-thread-controller`` against the fake backend.""" + + def __init__(self, root, binary, dbus_server): + self.root = root + self._binary = binary + self._dbus = dbus_server + self._proc = None + self._log_path = root / "daemon.log" + self.ownership_path = _physical(root, OWNERSHIP_FILE) + + def __call__(self, engine, engine_config, controller_overrides): + self.stop() + config = { + "scale_poll_secs": 0.2, + "enable_per_vm_status_line": True, + "enable_aggregate_status_line": True, + "min_thread_count": 1, + "max_thread_count": 8, + "host_cpu_scale_up_ceiling_percent": 0, + "cooldown_secs": 0.0, + "engine": engine, + "engine_config_dir": ENGINE_DIR, + "backend_config_dir": BACKEND_DIR, + "vm_state_path": OWNERSHIP_FILE, + "dry_run": False, + "print_status_header": False, + } + config.update(controller_overrides) + _write_json(_physical(self.root, CONFIG_FILE), config) + _write_json( + _physical(self.root, ENGINE_DIR) / ("%s.json" % engine), + engine_config, + ) + _write_json( + _physical(self.root, BACKEND_DIR) / "fake.json", + {"state_dir": STATE_DIR}, + ) + self.start() + + def start(self): + log_fd = os.open( + self._log_path, os.O_CREAT | os.O_WRONLY | os.O_TRUNC, 0o644 + ) + self._proc = subprocess.Popen( + [ + str(self._binary), + "--config", + CONFIG_FILE, + "--log-style", + "human", + "--log-level", + "info", + ], + cwd=REPO_ROOT, + env={ + **os.environ, + "IO_THREAD_CONTROLLER_ROOT_PATH": str(self.root), + "DBUS_SYSTEM_BUS_ADDRESS": self._dbus.address, + "RUST_LOG": "info", + }, + stdout=log_fd, + stderr=log_fd, + ) + os.close(log_fd) + + def logs(self): + if not self._log_path.exists(): + return "" + return self._log_path.read_text(errors="replace") + + def stop(self): + if self._proc is None or self._proc.poll() is not None: + self._proc = None + return + self._proc.send_signal(signal.SIGINT) + try: + self._proc.wait(timeout=5) + except subprocess.TimeoutExpired: + self._proc.kill() + self._proc.wait(timeout=5) + self._proc = None + + +@pytest.fixture +def fake_backend(tmp_path): + return FakeBackend(tmp_path) + + +@pytest.fixture +def dbus_server(tmp_path): + server = DBusServer(tmp_path) + server.start() + yield server + server.stop() + + +@pytest.fixture +def mock_proc(tmp_path, dbus_server): + binary = _target_dir() / "debug" / "mockfsd" + if not binary.exists(): + raise RuntimeError("mockfsd is not built at %s" % binary) + proc = MockProc(tmp_path, binary, dbus_server) + yield proc + proc.stop() + + +@pytest.fixture +def controller(tmp_path, fake_backend, dbus_server): + del fake_backend # created first so the state directory exists + binary = _target_dir() / "debug" / "io-thread-controller" + if not binary.exists(): + raise RuntimeError("io-thread-controller is not built at %s" % binary) + launched = Controller(tmp_path, binary, dbus_server) + yield launched + launched.stop() + + +@pytest.fixture +def fast_threshold(): + """Decisions land on the next eligible poll, with validation disabled.""" + return { + "scale_up_threshold_percent": 60, + "scale_down_sustain_polls": 1, + "max_scale_down_step": 1, + "scale_up_min_gain_percent": 0, + "scale_down_revert_drop_percent": 0, + "scale_validation_sample_polls": 0, + } + + +@pytest.fixture +def run_threshold(controller): + """Start the daemon on the threshold engine with open controller bounds.""" + + def start(engine_config, **overrides): + settings = { + "scale_poll_secs": POLL_S, + "min_thread_count": 1, + "max_thread_count": 8, + "host_cpu_scale_up_ceiling_percent": 0, + "cooldown_secs": 0, + } + settings.update(overrides) + controller( + engine="threshold", + engine_config=engine_config, + controller_overrides=settings, + ) + + return start diff --git a/tests/component/test_controller.py b/tests/component/test_controller.py new file mode 100644 index 0000000..4428074 --- /dev/null +++ b/tests/component/test_controller.py @@ -0,0 +1,206 @@ +# SPDX-License-Identifier: BSD-3-Clause +# Copyright (c) 2026 Nutanix, Inc. All rights reserved. +# +# Author: Leonardo Forchini + +"""Controller policy observed through a running daemon and the fake backend.""" + +import json +import re +import time + +from conftest import POLL_S, _prepare, wait_for + +HEALTHY_VM = "vm-a" +FAILED_VM = "vm-bad" + +_ANSI_RE = re.compile(r"\x1b\[[0-9;]*[A-Za-z]") +_VM_RE = re.compile(r'\bvm="?([^"\s]+)"?') +_THR_RE = re.compile(r"\bthr=(\d+)\b") +_TRACKED_RE = re.compile(r"\btracked=(\d+)") +_ACTION_RE = re.compile(r'\baction="?([A-Za-z]+)"?') + + +def _plain(logs): + """Drop SGR color codes so field matchers see ``tracked=1`` and ``status:``.""" + return _ANSI_RE.sub("", logs) + + +def _stat(user, idle): + body = "cpu %d 0 0 %d 0 0 0 0 0 0\n" % (user, idle) + body += "cpu0 %d 0 0 %d 0 0 0 0 0 0\n" % (user, idle) + body += "ctxt 1\nbtime 1\nprocesses 1\n" + return body.encode("ascii") + + +def _high_host_sequence(): + """Baseline sample, then busy deltas that pin host utilisation near 1.""" + samples = [_stat(0, 1000)] + idle = 1000 + user = 0 + for _ in range(40): + user += 10000 + samples.append(_stat(user, idle)) + return samples + + +def _status_threads(logs, vm_id): + found = None + for line in _plain(logs).splitlines(): + if " status:" not in line or "aggregate" in line: + continue + vm = _VM_RE.search(line) + thr = _THR_RE.search(line) + if vm and thr and vm.group(1) == vm_id: + found = int(thr.group(1)) + return found + + +def _tracked(logs): + found = None + for line in _plain(logs).splitlines(): + if "aggregate" not in line: + continue + match = _TRACKED_RE.search(line) + if match: + found = int(match.group(1)) + return found + + +def _engine_actions(logs): + actions = [] + for line in _plain(logs).splitlines(): + if " engine:" not in line: + continue + vm = _VM_RE.search(line) + action = _ACTION_RE.search(line) + if vm and action: + actions.append((vm.group(1), action.group(1))) + return actions + + +def test_actuation_enforces_vcpu_cap(controller, fake_backend, run_threshold, fast_threshold): + """A managed VM's pool stays within its vCPU count, and an unmanaged VM does not move.""" + _prepare(fake_backend, threads=3, util=0.95, vcpu=4) + run_threshold(fast_threshold) + + wait_for( + lambda: 4 in fake_backend.calls(), + timeout=10, + description="scale up to the vCPU count", + ) + time.sleep(POLL_S * 4) + assert fake_backend.calls() == [4], controller.logs() + assert fake_backend.thread_count() == 4 + + fake_backend.set_util(0.0) + wait_for( + lambda: fake_backend.calls()[-1:] == [3], + timeout=10, + description="scale down to 3", + ) + assert fake_backend.thread_count() == 3 + + controller.stop() + controller.ownership_path.parent.mkdir(parents=True, exist_ok=True) + controller.ownership_path.write_text( + json.dumps({"managed_vms": [], "unmanaged_vms": [HEALTHY_VM]}) + ) + before = list(fake_backend.calls()) + controller.start() + fake_backend.set_util(0.95) + time.sleep(POLL_S * 4) + fake_backend.set_util(0.0) + time.sleep(POLL_S * 4) + assert fake_backend.calls() == before, controller.logs() + assert fake_backend.thread_count() == 3 + + +def test_actuation_enforces_controller_bounds_and_host_ceiling( + controller, fake_backend, mock_proc, run_threshold, fast_threshold +): + """Controller min, max, host CPU ceiling, and cooldown gate ordinary scales.""" + _prepare(fake_backend, threads=2, util=0.95, vcpu=8) + run_threshold( + fast_threshold, + min_thread_count=2, + max_thread_count=3, + cooldown_secs=30, + ) + wait_for( + lambda: fake_backend.calls() == [3], + timeout=10, + description="scale up to the controller maximum", + ) + time.sleep(POLL_S * 4) + assert fake_backend.calls() == [3], controller.logs() + assert fake_backend.thread_count() == 3 + + fake_backend.set_util(0.0) + time.sleep(POLL_S * 4) + assert fake_backend.calls() == [3], controller.logs() + assert fake_backend.thread_count() == 3 + + controller.stop() + fake_backend.clear_calls() + fake_backend.set_threads(4) + fake_backend.set_util(0.0) + controller( + engine="threshold", + engine_config=fast_threshold, + controller_overrides={ + "scale_poll_secs": POLL_S, + "min_thread_count": 2, + "max_thread_count": 8, + "host_cpu_scale_up_ceiling_percent": 0, + "cooldown_secs": 0, + }, + ) + wait_for( + lambda: fake_backend.calls() == [3, 2], + timeout=10, + description="scale down to the controller minimum", + ) + time.sleep(POLL_S * 4) + assert fake_backend.calls() == [3, 2], controller.logs() + assert fake_backend.thread_count() == 2 + + controller.stop() + mock_proc.register_sequence("/stat", _high_host_sequence()) + fake_backend.clear_calls() + _prepare(fake_backend, threads=1, util=0.95, vcpu=8) + run_threshold(fast_threshold, host_cpu_scale_up_ceiling_percent=50) + wait_for( + lambda: fake_backend.calls() == [2], + timeout=10, + description="scale up before the host ceiling is known", + ) + time.sleep(POLL_S * 4) + assert fake_backend.calls() == [2], controller.logs() + assert fake_backend.thread_count() == 2 + + +def test_tick_refreshes_evaluates_and_drops_failed_instances( + controller, fake_backend, run_threshold, fast_threshold +): + """One poll refreshes a healthy VM and drops a VM whose snapshot fails.""" + _prepare(fake_backend, threads=4, util=0.95, vcpu=1) + fake_backend.add_vm( + FAILED_VM, + thread_count=2, + vcpu_count=4, + per_thread_util=0.95, + fail_snapshot=True, + ) + run_threshold(fast_threshold) + + wait_for( + lambda: _tracked(controller.logs()) == 1 + and _status_threads(controller.logs(), HEALTHY_VM) == fake_backend.thread_count(), + timeout=10, + description="healthy VM status after the failed VM is dropped", + ) + assert fake_backend.calls(FAILED_VM) == [], controller.logs() + actions = _engine_actions(controller.logs()) + assert actions, controller.logs() + assert {vm for vm, _action in actions} == {HEALTHY_VM}, controller.logs() diff --git a/tests/component/test_runaway_scale_regression.py b/tests/component/test_runaway_scale_regression.py index ecadff1..5f80b40 100644 --- a/tests/component/test_runaway_scale_regression.py +++ b/tests/component/test_runaway_scale_regression.py @@ -18,7 +18,7 @@ import time -from conftest import wait_for +from conftest import _prepare, wait_for # Values picked so a saturated workload can grow at most one @@ -50,25 +50,19 @@ def _threshold_config(): } -def test_saturated_workload_does_not_burst_scale(controller, fake_backend): +def test_saturated_workload_does_not_burst_scale(fake_backend, run_threshold): """A hot workload should NOT burn through every scale-up slot in a couple of polls. We prove this by measuring inter-scale spacing and asserting it stays above the validation-window floor. """ - fake_backend.set_util(0.95) - controller( - engine="threshold", - engine_config=_threshold_config(), - controller_overrides={ - "scale_poll_secs": POLL_INTERVAL_S, - "min_thread_count": 1, - # Cap low so the test finishes fast; the assertion - # measures inter-scale spacing, not the target count. - "max_thread_count": 6, - "host_cpu_scale_up_ceiling_percent": 0, - "cooldown_secs": 0, - }, + _prepare(fake_backend, threads=1, util=0.95, vcpu=8) + run_threshold( + _threshold_config(), + scale_poll_secs=POLL_INTERVAL_S, + # Cap low so the test finishes fast; the assertion + # measures inter-scale spacing, not the target count. + max_thread_count=6, ) # Wait for the first scale to land so we know the