Add component tests for threshold engine decisions

tests/component/test_threshold_engine.py

@@ -0,0 +1,310 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright (c) 2026 Nutanix, Inc. All rights reserved.
+#
+# Author: Leonardo Forchini <leonardo.forchini@nutanix.com>
+
+"""Test threshold engine behavior"""
+
+import time
+from collections.abc import Callable
+from typing import Any
+
+import pytest
+from harness.common import POLL_S, wait_for
+from harness.controller import Controller
+from harness.fake_backend import FakeBackend
+
+BASELINE_IO = 10_000
+
+
+@pytest.fixture
+def validating_threshold() -> Any:
+    """Same thresholds, with a hold long enough to change I/O before the sample."""
+    return {
+        "scale_up_threshold_percent": 60,
+        "scale_down_sustain_secs": POLL_S * 1,
+        "max_scale_down_step": 1,
+        "scale_up_min_gain_percent": 10,
+        "scale_down_revert_drop_percent": 10,
+        "scale_validation_delay_secs": POLL_S * 3,
+    }
+
+
+def _ticked(controller: Controller) -> bool:
+    """True once the first poll reports the healthy VM over D-Bus."""
+    stats = controller.stats()
+    return stats is not None and len(stats.get("vms", [])) >= 1
+
+
+def test_hold_below_scale_up_threshold(
+    controller: Controller,
+    fake_backend: FakeBackend,
+    run_controller: Callable,
+    fast_threshold: Any,
+) -> None:
+    """Utilisation under the up threshold, and not low enough to shrink, holds."""
+    fake_backend.prepare_healthy_vm(threads=2, util=0.5)
+    run_controller(fast_threshold)
+
+    wait_for(
+        lambda: _ticked(controller),
+        timeout=10,
+        description="first poll classified the held VM",
+    )
+    assert fake_backend.calls() == [], controller.logs()
+    assert fake_backend.thread_count() == 2
+
+
+def test_scale_up_one_thread(
+    fake_backend: FakeBackend, run_controller: Callable, fast_threshold: Any
+) -> None:
+    """A saturated worker grows by one and stops at the controller maximum."""
+    fake_backend.prepare_healthy_vm(threads=1, util=0.95)
+    run_controller(fast_threshold, max_thread_count=2)
+
+    wait_for(
+        lambda: fake_backend.calls() == [2],
+        timeout=10,
+        description="scale up by one thread",
+    )
+    assert fake_backend.thread_count() == 2
+
+
+def test_scale_down_one_thread(
+    fake_backend: FakeBackend, run_controller: Callable, fast_threshold: Any
+) -> None:
+    """Low utilisation must last for the sustain window, then drops by one step."""
+    fake_backend.prepare_healthy_vm(threads=4, util=0.0)
+    run_controller(fast_threshold)
+
+    wait_for(
+        lambda: fake_backend.calls() == [3],
+        timeout=10,
+        description="scale down by one thread",
+    )
+    assert fake_backend.thread_count() == 3
+
+
+def test_scale_down_step_limit(
+    controller: Controller,
+    fake_backend: FakeBackend,
+    run_controller: Callable,
+    fast_threshold: Any,
+) -> None:
+    """An idle pool sheds one thread per successful scale, not the whole surplus."""
+    fake_backend.prepare_healthy_vm(threads=6, util=0.0)
+    fast_threshold["max_scale_down_step"] = 1
+    run_controller(fast_threshold)
+
+    wait_for(
+        lambda: fake_backend.calls() == [5, 4, 3, 2, 1],
+        timeout=10,
+        description="step down to the controller minimum",
+    )
+    assert fake_backend.thread_count() == 1, controller.logs()
+
+
+def test_scale_up_reverts_when_iops_rate_falls(
+    controller: Controller,
+    fake_backend: FakeBackend,
+    run_controller: Callable,
+    validating_threshold: Any,
+) -> None:
+    """A scale-up is restored when its validation-window IOPS rate falls."""
+    lifetime = 10_000_000
+    fast_step = 20_000
+    slow_step = 200
+    io = lifetime
+    fake_backend.prepare_healthy_vm(threads=2, util=0.5, io=io)
+    run_controller(validating_threshold)
+
+    wait_for(lambda: _ticked(controller), timeout=10, description="first baseline poll")
+    baseline_until = time.monotonic() + POLL_S * 4
+    while time.monotonic() < baseline_until:
+        io += fast_step
+        fake_backend.set_io_counts(io, 0, 0)
+        time.sleep(POLL_S)
+
+    fake_backend.set_util(0.95)
+    deadline = time.monotonic() + 10
+    while fake_backend.calls() != [3] and time.monotonic() < deadline:
+        io += fast_step
+        fake_backend.set_io_counts(io, 0, 0)
+        time.sleep(POLL_S)
+    assert fake_backend.calls() == [3], controller.logs()
+
+    deadline = time.monotonic() + 10
+    while fake_backend.calls()[:2] != [3, 2] and time.monotonic() < deadline:
+        io += slow_step
+        fake_backend.set_io_counts(io, 0, 0)
+        time.sleep(POLL_S)
+    assert fake_backend.calls()[:2] == [3, 2], controller.logs()
+    assert fake_backend.thread_count() == 2
+
+
+def test_scale_up_kept_when_io_gains(
+    fake_backend: FakeBackend, run_controller: Callable, validating_threshold: Any
+) -> None:
+    """Enough extra I/O keeps the new thread and allows the next scale-up."""
+    fake_backend.prepare_healthy_vm(threads=2, util=0.95, io=BASELINE_IO)
+    run_controller(validating_threshold)
+
+    wait_for(
+        lambda: fake_backend.calls() == [3],
+        timeout=10,
+        description="scale up before validation",
+    )
+
+    fake_backend.set_io_counts(BASELINE_IO * 2, 0, 0)
+    wait_for(
+        lambda: len(fake_backend.calls()) == 2,
+        timeout=10,
+        description="next scale-up after validation",
+    )
+    assert fake_backend.calls() == [3, 4]
+
+
+def test_scale_down_kept_when_io_keeps_rising(
+    controller: Controller,
+    fake_backend: FakeBackend,
+    run_controller: Callable,
+    validating_threshold: Any,
+) -> None:
+    """A scale-down stays when cumulative I/O keeps increasing.
+
+    QEMU ``query-blockstats`` counters only grow. A lower total is not a
+    regression this backend can report, so validation must leave the new
+    count in place while the counter climbs.
+    """
+    fake_backend.prepare_healthy_vm(threads=4, util=0.0, io=BASELINE_IO)
+    run_controller(validating_threshold)
+
+    wait_for(
+        lambda: fake_backend.calls() == [3],
+        timeout=10,
+        description="scale down before validation",
+    )
+    fake_backend.set_util(0.5)
+    fake_backend.set_io_counts(BASELINE_IO * 2, 0, 0)
+    time.sleep(POLL_S * 5)
+    assert fake_backend.calls() == [3], controller.logs()
+    assert fake_backend.thread_count() == 3
+
+
+def test_scale_down_reverts_when_iops_rate_falls(
+    controller: Controller,
+    fake_backend: FakeBackend,
+    run_controller: Callable,
+    validating_threshold: Any,
+) -> None:
+    """A collapsed IOPS rate reverts a scale-down even though the counter rises.
+
+    QEMU blockstats are lifetime totals. They do not fall when the guest slows
+    down; the loss shows up as a smaller delta between polls. This pool's
+    counter keeps climbing, but the climb slows from about 100k IOPS to about
+    1k IOPS, well past the 10% allowance. The previous thread count should be
+    restored.
+
+    The engine compares IOPS rates, so the slower post-scale window restores
+    the previous thread count.
+    """
+    # 20_000 ops before each 0.2s poll is 100_000 IOPS. 200 ops/poll is 1_000.
+    lifetime = 10_000_000
+    fast_step = 20_000
+    slow_step = 200
+    config = dict(validating_threshold)
+    config["scale_down_sustain_secs"] = POLL_S * 5
+    io = lifetime
+    fake_backend.prepare_healthy_vm(threads=4, util=0.0, io=io)
+    run_controller(config)
+
+    deadline = time.monotonic() + 10
+    while not fake_backend.calls() and time.monotonic() < deadline:
+        io += fast_step
+        fake_backend.set_io_counts(io, 0, 0)
+        time.sleep(POLL_S)
+    assert fake_backend.calls() == [3], controller.logs()
+    count_at_scale = io
+
+    fake_backend.set_util(0.95)
+    deadline = time.monotonic() + 10
+    while fake_backend.calls()[:2] != [3, 4] and time.monotonic() < deadline:
+        io += slow_step
+        fake_backend.set_io_counts(io, 0, 0)
+        time.sleep(POLL_S)
+    assert io > count_at_scale
+    assert fake_backend.calls()[:2] == [3, 4], controller.logs()
+    assert fake_backend.thread_count() == 4
+
+
+def test_scale_up_kept_when_iops_rate_rises(
+    controller: Controller,
+    fake_backend: FakeBackend,
+    run_controller: Callable,
+    validating_threshold: Any,
+) -> None:
+    """A doubled IOPS rate keeps a scale-up even though the lifetime counter barely moves.
+
+    The counter starts at 10_000_000. After the pool grows, completed I/O
+    rises from about 100k IOPS to about 200k IOPS, twice the 10% allowance.
+    That extra work is only a few hundred thousand operations, far under 10%
+    of the lifetime total. Rate-based validation keeps the new worker count.
+    """
+    # 20_000 ops before each 0.2s poll is 100_000 IOPS. 40_000 ops/poll is 200_000.
+    lifetime = 10_000_000
+    baseline_step = 20_000
+    gained_step = 40_000
+    io = lifetime
+    fake_backend.prepare_healthy_vm(threads=2, util=0.5, io=io)
+    run_controller(validating_threshold)
+
+    wait_for(
+        lambda: _ticked(controller),
+        timeout=10,
+        description="first poll before the scale-up",
+    )
+    baseline_until = time.monotonic() + POLL_S * 4
+    while time.monotonic() < baseline_until:
+        io += baseline_step
+        fake_backend.set_io_counts(io, 0, 0)
+        time.sleep(POLL_S)
+    assert fake_backend.calls() == [], controller.logs()
+
+    fake_backend.set_util(0.95)
+    deadline = time.monotonic() + 10
+    while fake_backend.calls() != [3] and time.monotonic() < deadline:
+        io += baseline_step
+        fake_backend.set_io_counts(io, 0, 0)
+        time.sleep(POLL_S)
+    assert fake_backend.calls() == [3], controller.logs()
+    count_at_scale = io
+
+    settled = time.monotonic() + POLL_S * 8
+    while time.monotonic() < settled:
+        io += gained_step
+        fake_backend.set_io_counts(io, 0, 0)
+        time.sleep(POLL_S)
+    assert count_at_scale < io < count_at_scale + count_at_scale // 10
+    calls = fake_backend.calls()
+    assert calls[0] == 3 and calls[1] == 4, controller.logs()
+    assert 2 not in calls, controller.logs()
+
+
+def test_scale_down_kept_when_io_holds(
+    controller: Controller,
+    fake_backend: FakeBackend,
+    run_controller: Callable,
+    validating_threshold: Any,
+) -> None:
+    """A scale-down whose I/O stays inside the drop allowance is left in place."""
+    fake_backend.prepare_healthy_vm(threads=4, util=0.0, io=BASELINE_IO)
+    run_controller(validating_threshold)
+
+    wait_for(
+        lambda: fake_backend.calls() == [3],
+        timeout=10,
+        description="scale down before validation",
+    )
+    fake_backend.set_util(0.5)
+    assert fake_backend.calls() == [3], controller.logs()
+    assert fake_backend.thread_count() == 3