stroblmeandClaude Fable 5
b33be3fdd4
A standalone harness, never part of the test run, that drives a real stack through the durable path — the webhook, which journals every trigger — and then stops Redis, kills the engine mid-cascade and restarts the broker under a live subscription. Latest-value-wins with concurrent cascades means counter equality is not a promise, so what it asserts is that the queue drains, that state ends on the last value sent, and that nothing reached the dead-letter stream. Every docker verb goes through one helper that checks the compose label before it acts and refuses anything outside this project, because the machines this runs on host unrelated services. Two invariants are deliberately looser than they look. The queue belongs to the whole stack, so "nothing pending" would be measuring other people's traffic; the harness waits on the age of the oldest unacknowledged entry instead, which a stuck item always dominates. And the observability tables are cleared only after the collector's flush interval has passed, since deleting a flow publishes an audit event of its own. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_017MeiWk3Yq12n2pTvnQWYvt
974 lines
34 KiB
Python
974 lines
34 KiB
Python
"""Soak and chaos harness: does the engine keep its promises under load and failure?
|
|
|
|
Standalone on purpose, and never part of the test run. It drives the running
|
|
stack through the *durable* path — the webhook, which journals every trigger to
|
|
the work queue — and then takes pieces of the stack away to see what survives.
|
|
|
|
make soak
|
|
make soak SOAK_ARGS="--minutes 30 --rps 25"
|
|
make soak SOAK_ARGS="--scenario redis"
|
|
make soak SOAK_ARGS="--dry-run"
|
|
|
|
Every docker verb goes through :func:`docker`, which refuses any container that
|
|
is not part of this project's compose stack. The machine this runs on hosts
|
|
unrelated services; nothing here may ever reach them.
|
|
|
|
The engine's semantic is latest-value-wins with cascades running concurrently,
|
|
so counter equality is not a promise and is not asserted. What is asserted is
|
|
that the queue drains, that state ends on the last value sent, and that nothing
|
|
was dead-lettered.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import logging
|
|
import secrets as token
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
from collections.abc import Callable
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from dataclasses import dataclass, field
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
from app.core.config import settings
|
|
from app.flow.metrics import FLUSH_INTERVAL_S
|
|
|
|
log = logging.getLogger("soak")
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# Host safety
|
|
#
|
|
# This host runs critical services that have nothing to do with fluksio. The
|
|
# allow-list is checked first, the compose label second, and both before
|
|
# anything is executed — a typo must never become a restart of someone else's
|
|
# database.
|
|
# -----------------------------------------------------------------------------
|
|
|
|
COMPOSE_PROJECT = "fluksio-app"
|
|
REDIS = "fluksio-redis"
|
|
API = "fluksio-api"
|
|
BROKER = "fluksio-app-mosquitto-1"
|
|
DB = "fluksio-db"
|
|
#: The only containers this harness may ever name. The first three are the
|
|
#: chaos targets; the database is read and cleaned up, never stopped.
|
|
ALLOWED = frozenset({REDIS, API, BROKER, DB})
|
|
|
|
|
|
class HostSafety(RuntimeError):
|
|
"""A docker call was about to touch something outside this stack."""
|
|
|
|
|
|
def _compose_project(container: str) -> str:
|
|
"""Which compose project owns a container, or "" if there is no such container."""
|
|
result = subprocess.run(
|
|
[
|
|
"docker",
|
|
"inspect",
|
|
"-f",
|
|
'{{index .Config.Labels "com.docker.compose.project"}}',
|
|
container,
|
|
],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=30,
|
|
)
|
|
return result.stdout.strip() if result.returncode == 0 else ""
|
|
|
|
|
|
def docker(container: str, *argv: str, timeout: float = 120.0) -> str:
|
|
"""Run ``docker <argv>`` against a container of this stack, and nothing else."""
|
|
if container not in ALLOWED:
|
|
raise HostSafety(f"'{container}' is not one of {sorted(ALLOWED)} — refusing")
|
|
if container not in argv:
|
|
raise HostSafety(f"'{container}' is guarded but absent from {argv} — refusing")
|
|
project = _compose_project(container)
|
|
if project != COMPOSE_PROJECT:
|
|
raise HostSafety(
|
|
f"'{container}' belongs to compose project '{project}', "
|
|
f"not '{COMPOSE_PROJECT}' — refusing"
|
|
)
|
|
result = subprocess.run(
|
|
["docker", *argv], capture_output=True, text=True, timeout=timeout
|
|
)
|
|
if result.returncode != 0:
|
|
raise RuntimeError(f"docker {' '.join(argv)}: {result.stderr.strip()}")
|
|
return result.stdout
|
|
|
|
|
|
def is_ours(container: str) -> bool:
|
|
"""Does this container exist and belong to our stack? Inspect only."""
|
|
return _compose_project(container) == COMPOSE_PROJECT
|
|
|
|
|
|
def is_running(container: str) -> bool:
|
|
return (
|
|
docker(container, "inspect", "-f", "{{.State.Running}}", container).strip()
|
|
== "true"
|
|
)
|
|
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# Thresholds
|
|
# -----------------------------------------------------------------------------
|
|
|
|
MAX_RPS = 50
|
|
HTTP_TIMEOUT = 10.0
|
|
P95_LIMIT_MS = 500.0
|
|
ACCEPT_RATE = 0.99
|
|
#: Cascades coalesce — a value superseded before its node ran is not a loss.
|
|
#: The counter is therefore a floor, not an equality.
|
|
COUNT_FLOOR = 0.9
|
|
#: How long a quiet queue is given to prove it after the load stops.
|
|
DRAIN_S = 90.0
|
|
#: An unacknowledged entry younger than this is a cascade that is running, not
|
|
#: work that was lost. See :func:`is_quiet`.
|
|
QUIET_AGE_S = 5.0
|
|
#: Redelivery after an engine death waits on the reclaim path, not on the
|
|
#: engine: XAUTOCLAIM only takes entries idle for RECLAIM_IDLE_MS (60 s) and the
|
|
#: reaper runs every RECLAIM_INTERVAL_S (30 s) — see app/flow/executor.py. Do
|
|
#: not tighten this below their sum plus room to actually run the work.
|
|
ENGINE_DRAIN_S = 150.0
|
|
HEALTH_RECOVER_S = 90.0
|
|
REDIS_RECOVER_S = 60.0
|
|
#: A subscription is restarted with backoff 1, 5, 30, 60 s (app/flow/
|
|
#: supervision.py), so a reconnect can legitimately take half a minute.
|
|
BROKER_RECOVER_S = 60.0
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# Fixtures
|
|
# -----------------------------------------------------------------------------
|
|
|
|
#: Flow names are lowercase identifiers (app/flow/schemas.py NAME_PATTERN), so
|
|
#: the marker is an underscore rather than a hyphen.
|
|
PREFIX = "soak_"
|
|
FLOW_LOAD = f"{PREFIX}load"
|
|
FLOW_CHAIN = f"{PREFIX}chain"
|
|
FLOW_MQTT = f"{PREFIX}mqtt"
|
|
#: Deletion order: whatever reads another flow's messages goes first.
|
|
FLOWS = (FLOW_MQTT, FLOW_CHAIN, FLOW_LOAD)
|
|
|
|
COUNT_SOURCE = '''"""Count what made it through, reading the total it also writes."""
|
|
|
|
|
|
def process(tick, total=0, params=None):
|
|
return {"total": int(total) + 1}
|
|
'''
|
|
|
|
|
|
def spec(name: str, dtype: str = "int", trigger: bool = True) -> dict[str, Any]:
|
|
return {"name": name, "dtype": dtype, "trigger": trigger}
|
|
|
|
|
|
def load_flow(secret: str) -> dict[str, Any]:
|
|
"""A webhook feeding an accumulator: the durable path, end to end."""
|
|
return {
|
|
"name": FLOW_LOAD,
|
|
"title": "Soak load",
|
|
"version": 0,
|
|
"nodes": [
|
|
{
|
|
"id": "hook",
|
|
"type": "http",
|
|
"title": "Tick",
|
|
"position": {"x": 40, "y": 40},
|
|
"params": {"url": "/tick", "method": "POST", "secret": secret},
|
|
"provides": [spec("tick")],
|
|
},
|
|
{
|
|
"id": "count",
|
|
"type": "python",
|
|
"title": "Count",
|
|
"position": {"x": 360, "y": 40},
|
|
# `total` does not trigger: the node reads the message it is
|
|
# about to write, which is how a node keeps a running total.
|
|
"requires": [spec("tick"), spec("total", trigger=False)],
|
|
"provides": [spec("total")],
|
|
},
|
|
],
|
|
"inputs": [{"spec": spec("total"), "initial": 0}],
|
|
}
|
|
|
|
|
|
def chain_flow() -> dict[str, Any]:
|
|
"""A second flow reading the first one's message, so cascades cross flows."""
|
|
return {
|
|
"name": FLOW_CHAIN,
|
|
"title": "Soak chain",
|
|
"version": 0,
|
|
"nodes": [
|
|
{
|
|
"id": "double",
|
|
"type": "change",
|
|
"title": "Double",
|
|
"position": {"x": 40, "y": 40},
|
|
"params": {"scale": 2.0, "offset": 0.0},
|
|
"requires": [spec(f"{FLOW_LOAD}.total")],
|
|
"provides": [spec("doubled", dtype="float")],
|
|
}
|
|
],
|
|
}
|
|
|
|
|
|
def mqtt_flow() -> dict[str, Any]:
|
|
"""A subscriber on the dev stack's own broker, for the broker scenario."""
|
|
return {
|
|
"name": FLOW_MQTT,
|
|
"title": "Soak broker",
|
|
"version": 0,
|
|
"nodes": [
|
|
{
|
|
"id": "sub",
|
|
"type": "mqtt",
|
|
"title": "Broker",
|
|
"position": {"x": 40, "y": 40},
|
|
"params": {
|
|
"topic": "soak/tick",
|
|
"broker_host": "mosquitto",
|
|
"broker_port": 1883,
|
|
"client_id": "fluksio-soak",
|
|
},
|
|
"provides": [spec("tick", dtype="float")],
|
|
}
|
|
],
|
|
}
|
|
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# The API
|
|
# -----------------------------------------------------------------------------
|
|
|
|
|
|
def connect(base_url: str) -> httpx.Client:
|
|
"""A logged-in client for the JSON API."""
|
|
client = httpx.Client(base_url=base_url, timeout=HTTP_TIMEOUT)
|
|
response = client.post(
|
|
"/api/v1/login/access-token",
|
|
data={
|
|
"username": settings.FIRST_SUPERUSER,
|
|
"password": settings.FIRST_SUPERUSER_PASSWORD,
|
|
},
|
|
)
|
|
response.raise_for_status()
|
|
client.headers["Authorization"] = f"Bearer {response.json()['access_token']}"
|
|
return client
|
|
|
|
|
|
def health_code(client: httpx.Client) -> int:
|
|
"""The deep health check's status: 200 while the engine can do its job."""
|
|
try:
|
|
return client.get("/api/v1/utils/health/").status_code
|
|
except httpx.HTTPError:
|
|
return 0
|
|
|
|
|
|
def queue_stats(client: httpx.Client) -> dict[str, Any]:
|
|
"""Queue depth and age. Empty when the engine cannot say."""
|
|
try:
|
|
response = client.get("/api/v1/observability/summary")
|
|
response.raise_for_status()
|
|
queue: dict[str, Any] = response.json()["queue"]
|
|
except (httpx.HTTPError, KeyError, ValueError):
|
|
return {}
|
|
return {} if queue.get("error") else queue
|
|
|
|
|
|
def message_value(client: httpx.Client, flow: str, message: str) -> Any:
|
|
"""The last value on a message, or None when it has none — or the API is down."""
|
|
try:
|
|
response = client.get(f"/api/v1/flows/{flow}/state")
|
|
response.raise_for_status()
|
|
entry = response.json()["values"].get(f"{flow}.{message}")
|
|
except (httpx.HTTPError, KeyError, ValueError):
|
|
return None
|
|
return entry["value"] if entry else None
|
|
|
|
|
|
def node_health(client: httpx.Client, flow: str, node_id: str) -> str:
|
|
"""How a node says its connection is doing: ok, degraded or down."""
|
|
try:
|
|
response = client.get(f"/api/v1/flows/{flow}")
|
|
response.raise_for_status()
|
|
except httpx.HTTPError:
|
|
return "unknown"
|
|
for status in response.json()["nodes"]:
|
|
if status["id"] == f"{flow}.{node_id}":
|
|
return str(status["health"])
|
|
return "unknown"
|
|
|
|
|
|
def dead_letters(client: httpx.Client) -> dict[str, dict[str, Any]]:
|
|
"""Work the engine gave up on, keyed by entry id."""
|
|
try:
|
|
response = client.get(
|
|
"/api/v1/observability/dead-letter", params={"limit": 200}
|
|
)
|
|
response.raise_for_status()
|
|
except httpx.HTTPError:
|
|
return {}
|
|
return {str(row["id"]): row for row in response.json()}
|
|
|
|
|
|
def create_flow(
|
|
client: httpx.Client,
|
|
definition: dict[str, Any],
|
|
sources: dict[str, str] | None = None,
|
|
) -> None:
|
|
"""Draft, node source, publish — the sequence the editor uses."""
|
|
name = definition["name"]
|
|
response = client.put(f"/api/v1/flows/{name}", json=definition)
|
|
response.raise_for_status()
|
|
version = response.json()["definition"]["version"]
|
|
|
|
for node_id, code in (sources or {}).items():
|
|
client.put(
|
|
f"/api/v1/flows/{name}/nodes/{node_id}/source", json={"code": code}
|
|
).raise_for_status()
|
|
|
|
response = client.post(f"/api/v1/flows/{name}/publish", json={"version": version})
|
|
response.raise_for_status()
|
|
issues = response.json()["issues"]
|
|
if issues:
|
|
raise RuntimeError(
|
|
f"'{name}' would not run: {[issue['message'] for issue in issues]}"
|
|
)
|
|
log.info("created flow '%s'", name)
|
|
|
|
|
|
def delete_flows(client: httpx.Client) -> None:
|
|
for name in FLOWS:
|
|
try:
|
|
response = client.delete(f"/api/v1/flows/{name}", timeout=30.0)
|
|
if response.status_code == 200:
|
|
log.info("deleted flow '%s'", name)
|
|
except httpx.HTTPError as exc:
|
|
log.error("could not delete flow '%s': %s", name, exc)
|
|
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# Waiting
|
|
# -----------------------------------------------------------------------------
|
|
|
|
|
|
def wait_for(
|
|
predicate: Callable[[], bool], timeout: float, interval: float = 1.0
|
|
) -> float | None:
|
|
"""How many seconds until ``predicate`` held, or None if it never did."""
|
|
started = time.monotonic()
|
|
while True:
|
|
if predicate():
|
|
return time.monotonic() - started
|
|
if time.monotonic() - started >= timeout:
|
|
return None
|
|
time.sleep(interval)
|
|
|
|
|
|
def is_quiet(stats: dict[str, Any]) -> bool:
|
|
"""Is nothing *waiting* in the queue?
|
|
|
|
The queue belongs to the whole stack, not to this run, so "pending is zero"
|
|
is the wrong question on a machine where other flows keep triggering. What
|
|
is asked instead is whether anything has been sitting there: an item that
|
|
was claimed a moment ago belongs to a cascade that is simply running, while
|
|
work the engine has lost hold of only gets older. ``oldest_pending_s`` is
|
|
the age of the *oldest* unacknowledged entry, so a stuck item always shows
|
|
even while newer ones come and go.
|
|
|
|
``delayed`` is deliberately not part of it: a rate-limited output keeps a
|
|
flush item parked there for as long as the flow runs, and no soak fixture
|
|
declares an interval, so none of that backlog is this run's.
|
|
"""
|
|
return bool(stats) and (
|
|
not stats.get("pending")
|
|
or float(stats.get("oldest_pending_s") or 0) < QUIET_AGE_S
|
|
)
|
|
|
|
|
|
def settle(client: httpx.Client, timeout: float) -> tuple[bool, dict[str, Any]]:
|
|
"""Wait for the work queue to go quiet, two clear polls in a row.
|
|
|
|
Two, because a freshly journaled item sits in the stream unclaimed for up to
|
|
the consumer's block window without ever showing up as pending.
|
|
"""
|
|
quiet = 0
|
|
stats: dict[str, Any] = {}
|
|
deadline = time.monotonic() + timeout
|
|
while True:
|
|
stats = queue_stats(client)
|
|
quiet = quiet + 1 if is_quiet(stats) else 0
|
|
if quiet >= 2:
|
|
return True, stats
|
|
if time.monotonic() >= deadline:
|
|
return False, stats
|
|
time.sleep(1.5)
|
|
|
|
|
|
def seconds(value: float | None) -> str:
|
|
return f"{value:.1f}s" if value is not None else "never"
|
|
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# The load
|
|
# -----------------------------------------------------------------------------
|
|
|
|
|
|
class Driver:
|
|
"""Fires the webhook at a fixed rate from a small pool of threads."""
|
|
|
|
def __init__(self, url: str, rps: int) -> None:
|
|
self.url = url
|
|
self.sent = 0
|
|
self.accepted = 0
|
|
self.failed = 0
|
|
self.skipped = 0
|
|
self.tick = 0
|
|
self.latencies: list[float] = []
|
|
self._interval = 1.0 / rps
|
|
self._client = httpx.Client(timeout=HTTP_TIMEOUT)
|
|
self._pool = ThreadPoolExecutor(
|
|
max_workers=min(32, rps + 4), thread_name_prefix="soak"
|
|
)
|
|
# Bounds what a stalled backend can make this script hold: a request
|
|
# with no free slot is skipped rather than queued forever.
|
|
self._slots = threading.Semaphore(4 * rps)
|
|
self._lock = threading.Lock()
|
|
self._stop = threading.Event()
|
|
self._pacer = threading.Thread(target=self._pace, name="soak-pacer")
|
|
|
|
def start(self) -> None:
|
|
self._pacer.start()
|
|
|
|
def stop(self) -> None:
|
|
self._stop.set()
|
|
self._pacer.join()
|
|
self._pool.shutdown(wait=True)
|
|
|
|
def close(self) -> None:
|
|
self._client.close()
|
|
|
|
def post(self, tick: int) -> bool:
|
|
"""One webhook call, timed. Returns whether it was accepted."""
|
|
started = time.perf_counter()
|
|
try:
|
|
accepted = self._client.post(self.url, json={"tick": tick}).is_success
|
|
except httpx.HTTPError:
|
|
accepted = False
|
|
elapsed_ms = (time.perf_counter() - started) * 1000
|
|
with self._lock:
|
|
self.latencies.append(elapsed_ms)
|
|
if accepted:
|
|
self.accepted += 1
|
|
else:
|
|
self.failed += 1
|
|
return accepted
|
|
|
|
def sentinel(self) -> int | None:
|
|
"""One last tick, sent on its own — what state has to end on."""
|
|
with self._lock:
|
|
self.tick += 1
|
|
self.sent += 1
|
|
tick = self.tick
|
|
return tick if self.post(tick) else None
|
|
|
|
def _fire(self, tick: int) -> None:
|
|
try:
|
|
self.post(tick)
|
|
finally:
|
|
self._slots.release()
|
|
|
|
def _pace(self) -> None:
|
|
due = time.monotonic()
|
|
while not self._stop.is_set():
|
|
due += self._interval
|
|
if self._slots.acquire(blocking=False):
|
|
with self._lock:
|
|
self.tick += 1
|
|
self.sent += 1
|
|
tick = self.tick
|
|
self._pool.submit(self._fire, tick)
|
|
else:
|
|
with self._lock:
|
|
self.skipped += 1
|
|
self._stop.wait(max(0.0, due - time.monotonic()))
|
|
|
|
|
|
def percentile(values: list[float], fraction: float) -> float:
|
|
if not values:
|
|
return 0.0
|
|
ordered = sorted(values)
|
|
return ordered[min(len(ordered) - 1, int(fraction * len(ordered)))]
|
|
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# Reporting
|
|
# -----------------------------------------------------------------------------
|
|
|
|
|
|
@dataclass
|
|
class Report:
|
|
"""What one scenario measured, and which of its promises held."""
|
|
|
|
name: str
|
|
stats: list[str] = field(default_factory=list)
|
|
checks: list[tuple[str, bool]] = field(default_factory=list)
|
|
notes: list[str] = field(default_factory=list)
|
|
skipped: str = ""
|
|
|
|
def stat(self, line: str) -> None:
|
|
self.stats.append(line)
|
|
|
|
def note(self, line: str) -> None:
|
|
self.notes.append(line)
|
|
|
|
def check(self, description: str, ok: bool) -> None:
|
|
self.checks.append((description, bool(ok)))
|
|
|
|
@property
|
|
def ok(self) -> bool:
|
|
return all(ok for _, ok in self.checks)
|
|
|
|
def show(self) -> None:
|
|
log.info("")
|
|
log.info("── %s %s", self.name, "─" * max(3, 56 - len(self.name)))
|
|
if self.skipped:
|
|
log.info(" SKIP %s", self.skipped)
|
|
return
|
|
for line in self.stats:
|
|
log.info(" %s", line)
|
|
for description, ok in self.checks:
|
|
log.info(" %s %s", "pass" if ok else "FAIL", description)
|
|
for line in self.notes:
|
|
log.info(" note %s", line)
|
|
|
|
|
|
def report_load(report: Report, driver: Driver) -> None:
|
|
report.stat(
|
|
f"posts sent {driver.sent} accepted {driver.accepted} "
|
|
f"failed {driver.failed} skipped {driver.skipped}"
|
|
)
|
|
report.stat(
|
|
f"latency p50 {percentile(driver.latencies, 0.5):.0f} ms "
|
|
f"p95 {percentile(driver.latencies, 0.95):.0f} ms"
|
|
)
|
|
|
|
|
|
def finish_load(
|
|
client: httpx.Client,
|
|
driver: Driver,
|
|
report: Report,
|
|
drain_s: float,
|
|
dead_before: dict[str, dict[str, Any]],
|
|
) -> None:
|
|
"""Stop the load and check how the engine ended up: drained, fresh, nothing lost."""
|
|
driver.stop()
|
|
report_load(report, driver)
|
|
|
|
drained, _ = settle(client, drain_s)
|
|
report.check(f"queue drains within {drain_s:.0f}s of the load stopping", drained)
|
|
|
|
# Sent alone, after the drain: with four cascades running at once, only a
|
|
# tick that had the queue to itself is guaranteed to be the last one
|
|
# written. Latest-value-wins says nothing about which of two racing
|
|
# cascades lands second.
|
|
tick = driver.sentinel()
|
|
report.check("the last webhook was accepted", tick is not None)
|
|
|
|
drained, stats = settle(client, drain_s)
|
|
report.check("nothing left waiting in the queue at the end", drained)
|
|
report.stat(
|
|
f"queue (shared with the rest of the stack) pending {stats.get('pending', '?')}"
|
|
f" oldest {stats.get('oldest_pending_s', '?')}s"
|
|
f" delayed {stats.get('delayed', '?')} parked {stats.get('parked', '?')}"
|
|
)
|
|
|
|
value = message_value(client, FLOW_LOAD, "tick")
|
|
total = message_value(client, FLOW_LOAD, "total")
|
|
report.stat(f"state tick {value} total {total}")
|
|
report.check(
|
|
f"state ends on the last tick sent (sent {tick}, holds {value})",
|
|
tick is not None and value == tick,
|
|
)
|
|
|
|
new_dead = [
|
|
row for key, row in dead_letters(client).items() if key not in dead_before
|
|
]
|
|
ours = [row for row in new_dead if str(row.get("flow", "")).startswith(PREFIX)]
|
|
report.stat(f"dead letters: {len(ours)} for soak flows, {len(new_dead)} in total")
|
|
report.check("nothing was dead-lettered", not ours)
|
|
|
|
report.check("engine healthy at the end", health_code(client) == 200)
|
|
driver.close()
|
|
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# Scenarios
|
|
# -----------------------------------------------------------------------------
|
|
|
|
|
|
def scenario_load(client: httpx.Client, args: argparse.Namespace, url: str) -> Report:
|
|
"""(a) Steady load down the durable path."""
|
|
report = Report("load")
|
|
dead_before = dead_letters(client)
|
|
before = message_value(client, FLOW_LOAD, "total") or 0
|
|
|
|
driver = Driver(url, args.rps)
|
|
log.info("load: %d rps for %.0fs", args.rps, args.seconds)
|
|
driver.start()
|
|
time.sleep(args.seconds)
|
|
finish_load(client, driver, report, DRAIN_S, dead_before)
|
|
|
|
rate = driver.accepted / driver.sent if driver.sent else 0.0
|
|
report.check(
|
|
f"at least {ACCEPT_RATE:.0%} of posts accepted ({rate:.1%})",
|
|
rate >= ACCEPT_RATE,
|
|
)
|
|
p95 = percentile(driver.latencies, 0.95)
|
|
report.check(f"p95 under {P95_LIMIT_MS:.0f} ms ({p95:.0f} ms)", p95 < P95_LIMIT_MS)
|
|
|
|
total = message_value(client, FLOW_LOAD, "total")
|
|
counted = (total or 0) - before
|
|
report.check(
|
|
f"counter kept up with the load ({counted} of {driver.accepted} accepted)",
|
|
total is not None and counted >= COUNT_FLOOR * driver.accepted,
|
|
)
|
|
doubled = message_value(client, FLOW_CHAIN, "doubled")
|
|
report.check(
|
|
f"the downstream flow saw the final total ({doubled} vs {total})",
|
|
total is not None and doubled == total * 2,
|
|
)
|
|
report.show()
|
|
return report
|
|
|
|
|
|
def scenario_redis(client: httpx.Client, args: argparse.Namespace, url: str) -> Report:
|
|
"""(b) The state and journal backend goes away mid-load."""
|
|
report = Report("redis")
|
|
dead_before = dead_letters(client)
|
|
|
|
driver = Driver(url, args.rps)
|
|
log.info("redis: %d rps, taking %s away halfway through", args.rps, REDIS)
|
|
driver.start()
|
|
time.sleep(args.seconds / 2)
|
|
|
|
before = message_value(client, FLOW_LOAD, "total")
|
|
# Past redis' appendfsync-everysec window, so what was just read is on disk
|
|
# and its survival is a real claim about the append-only file.
|
|
time.sleep(1.5)
|
|
|
|
docker(REDIS, "stop", REDIS)
|
|
degraded = wait_for(lambda: health_code(client) != 200, 30.0, 0.5)
|
|
report.check("health reports trouble while redis is gone", degraded is not None)
|
|
|
|
docker(REDIS, "start", REDIS)
|
|
recovered = wait_for(lambda: health_code(client) == 200, REDIS_RECOVER_S, 1.0)
|
|
report.check(
|
|
f"health back to 200 within {REDIS_RECOVER_S:.0f}s of redis returning",
|
|
recovered is not None,
|
|
)
|
|
report.stat(
|
|
f"degraded after {seconds(degraded)}, healthy again {seconds(recovered)} "
|
|
"after the restart"
|
|
)
|
|
|
|
after = message_value(client, FLOW_LOAD, "total")
|
|
report.check(
|
|
f"state journaled before the outage survived it ({before} → {after})",
|
|
before is not None and after is not None and after >= before,
|
|
)
|
|
|
|
time.sleep(args.seconds / 2)
|
|
finish_load(client, driver, report, DRAIN_S, dead_before)
|
|
report.show()
|
|
return report
|
|
|
|
|
|
def scenario_engine(client: httpx.Client, args: argparse.Namespace, url: str) -> Report:
|
|
"""(c) The engine is killed mid-cascade and has to pick the work back up."""
|
|
report = Report("engine")
|
|
dead_before = dead_letters(client)
|
|
|
|
driver = Driver(url, args.rps)
|
|
log.info("engine: %d rps, killing %s halfway through", args.rps, API)
|
|
driver.start()
|
|
time.sleep(args.seconds / 2)
|
|
|
|
at_kill = message_value(client, FLOW_LOAD, "total")
|
|
# Read after the total, so work already accepted but not yet counted makes
|
|
# the claim below easier rather than impossible.
|
|
accepted_at_kill = driver.accepted
|
|
|
|
docker(API, "kill", API)
|
|
time.sleep(2.0)
|
|
# The dev stack sets restart:"no" for the backend, so nothing else brings
|
|
# it back — and it must never be left down.
|
|
docker(API, "start", API)
|
|
recovered = wait_for(lambda: health_code(client) == 200, HEALTH_RECOVER_S, 2.0)
|
|
report.stat(f"engine answered healthy again {seconds(recovered)} after restart")
|
|
report.check(
|
|
f"health back to 200 within {HEALTH_RECOVER_S:.0f}s of the restart",
|
|
recovered is not None,
|
|
)
|
|
if recovered is None:
|
|
log.error("engine health after restart: %s", health_code(client))
|
|
|
|
time.sleep(args.seconds / 2)
|
|
finish_load(client, driver, report, ENGINE_DRAIN_S, dead_before)
|
|
|
|
accepted_after = driver.accepted - accepted_at_kill
|
|
after = message_value(client, FLOW_LOAD, "total")
|
|
floor = (at_kill or 0) + COUNT_FLOOR * accepted_after
|
|
report.check(
|
|
f"counter carried on across the kill ({at_kill} → {after}, "
|
|
f"{accepted_after} accepted after it)",
|
|
at_kill is not None and after is not None and after >= floor,
|
|
)
|
|
report.note(
|
|
"alert delivery is not asserted — that needs a channel on the receiving end"
|
|
)
|
|
report.show()
|
|
return report
|
|
|
|
|
|
def scenario_broker(client: httpx.Client) -> Report:
|
|
"""(e) The MQTT broker restarts under a live subscription."""
|
|
report = Report("broker")
|
|
if not is_ours(BROKER):
|
|
report.skipped = f"no {BROKER} container — the broker lives in the dev stack"
|
|
report.show()
|
|
return report
|
|
|
|
def subscribed() -> bool:
|
|
return node_health(client, FLOW_MQTT, "sub") == "ok"
|
|
|
|
def publish(value: str) -> None:
|
|
docker(BROKER, "exec", BROKER, "mosquitto_pub", "-t", "soak/tick", "-m", value)
|
|
|
|
report.check("the subscriber connects", wait_for(subscribed, 60.0, 2.0) is not None)
|
|
publish("1")
|
|
landed = wait_for(
|
|
lambda: message_value(client, FLOW_MQTT, "tick") == 1.0, 30.0, 1.0
|
|
)
|
|
report.check("a published value reaches the flow", landed is not None)
|
|
|
|
docker(BROKER, "stop", BROKER)
|
|
down = wait_for(lambda: node_health(client, FLOW_MQTT, "sub") == "down", 60.0, 1.0)
|
|
report.check(
|
|
"the subscriber reports down while the broker is gone", down is not None
|
|
)
|
|
|
|
docker(BROKER, "start", BROKER)
|
|
back = wait_for(subscribed, BROKER_RECOVER_S, 2.0)
|
|
report.check(
|
|
f"the subscriber is ok again within {BROKER_RECOVER_S:.0f}s", back is not None
|
|
)
|
|
report.stat(f"reported down after {seconds(down)}, ok again {seconds(back)} later")
|
|
|
|
publish("2")
|
|
landed = wait_for(
|
|
lambda: message_value(client, FLOW_MQTT, "tick") == 2.0, 30.0, 1.0
|
|
)
|
|
report.check("a value published after recovery lands", landed is not None)
|
|
report.show()
|
|
return report
|
|
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# Cleanup
|
|
# -----------------------------------------------------------------------------
|
|
|
|
#: Mirrors FlowController.forget_flow: the messages themselves, and the
|
|
#: engine's bookkeeping about them, which is keyed by message name too.
|
|
REDIS_PATTERNS = (f"pipeline:{PREFIX}*", f"pipeline:*:{PREFIX}*")
|
|
|
|
|
|
def clear_redis() -> None:
|
|
"""Belt and braces: deleting a flow clears its keys through the engine,
|
|
which is no help when the engine was the thing that died."""
|
|
try:
|
|
for pattern in REDIS_PATTERNS:
|
|
keys = docker(
|
|
REDIS, "exec", REDIS, "redis-cli", "--scan", "--pattern", pattern
|
|
).split()
|
|
for start in range(0, len(keys), 200):
|
|
docker(
|
|
REDIS, "exec", REDIS, "redis-cli", "del", *keys[start : start + 200]
|
|
)
|
|
if keys:
|
|
log.info("deleted %d redis key(s) matching %s", len(keys), pattern)
|
|
except Exception as exc:
|
|
log.error("redis keys may be left behind: %s", exc)
|
|
|
|
|
|
def clear_observability() -> None:
|
|
"""The collector wrote rollups, runs and events for the soak flows."""
|
|
names = "', '".join(FLOWS)
|
|
statement = "; ".join(
|
|
f"DELETE FROM {table} WHERE flow IN ('{names}')"
|
|
for table in ("metric_minute", "flow_run", "engine_event")
|
|
)
|
|
try:
|
|
docker(
|
|
DB,
|
|
"exec",
|
|
DB,
|
|
"psql",
|
|
"-q",
|
|
"-U",
|
|
settings.POSTGRES_USER,
|
|
"-d",
|
|
settings.POSTGRES_DB,
|
|
"-c",
|
|
statement,
|
|
)
|
|
log.info("deleted observability rows for the soak flows")
|
|
except Exception as exc:
|
|
log.error("observability rows may be left behind: %s", exc)
|
|
|
|
|
|
def cleanup(client: httpx.Client | None) -> None:
|
|
"""Undo everything this run created, whatever went wrong on the way."""
|
|
log.info("")
|
|
log.info("cleaning up")
|
|
for container in (REDIS, API, BROKER):
|
|
try:
|
|
if is_ours(container) and not is_running(container):
|
|
log.info("starting %s", container)
|
|
docker(container, "start", container)
|
|
except Exception as exc:
|
|
log.error("could not start %s: %s", container, exc)
|
|
|
|
if client is not None:
|
|
wait_for(lambda: health_code(client) == 200, HEALTH_RECOVER_S, 2.0)
|
|
delete_flows(client)
|
|
|
|
clear_redis()
|
|
# Deleting a flow publishes an audit event of its own, and the metrics
|
|
# collector buffers what it hears for up to FLUSH_INTERVAL_S before writing
|
|
# it. Clearing the tables any sooner races that flush and leaves the rows
|
|
# this very cleanup caused.
|
|
time.sleep(FLUSH_INTERVAL_S + 5.0)
|
|
clear_observability()
|
|
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# Entry point
|
|
# -----------------------------------------------------------------------------
|
|
|
|
|
|
def parse_args(argv: list[str] | None) -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(
|
|
description="Soak and chaos harness for the running fluksio stack.",
|
|
)
|
|
parser.add_argument(
|
|
"--minutes", type=float, default=2.0, help="Load time per scenario (default 2)."
|
|
)
|
|
parser.add_argument(
|
|
"--rps",
|
|
type=int,
|
|
default=10,
|
|
help=f"Webhooks per second, capped at {MAX_RPS} (default 10).",
|
|
)
|
|
parser.add_argument(
|
|
"--scenario",
|
|
choices=("all", "load", "redis", "engine", "broker"),
|
|
default="all",
|
|
help="Which scenario to run (default all).",
|
|
)
|
|
parser.add_argument(
|
|
"--base-url",
|
|
default="http://api.localhost",
|
|
help="Where the API is (default http://api.localhost).",
|
|
)
|
|
parser.add_argument(
|
|
"--dry-run",
|
|
action="store_true",
|
|
help="Report what the harness can see and change nothing.",
|
|
)
|
|
args = parser.parse_args(argv)
|
|
args.rps = max(1, min(MAX_RPS, args.rps))
|
|
args.seconds = max(10.0, args.minutes * 60)
|
|
return args
|
|
|
|
|
|
def dry_run(client: httpx.Client, args: argparse.Namespace) -> None:
|
|
"""Everything the harness needs, read-only: login, flows, health, containers."""
|
|
log.info("logged in as %s at %s", settings.FIRST_SUPERUSER, args.base_url)
|
|
flows = client.get("/api/v1/flows/").json()["data"]
|
|
log.info("flows: %s", ", ".join(flow["name"] for flow in flows) or "none")
|
|
response = client.get("/api/v1/utils/health/")
|
|
body = response.json()
|
|
log.info("health %s: %s %s", response.status_code, body["status"], body["problems"])
|
|
log.info("queue: %s", body["queue"])
|
|
for container in sorted(ALLOWED):
|
|
ours = is_ours(container)
|
|
log.info(
|
|
"%s: ours=%s running=%s",
|
|
container,
|
|
ours,
|
|
is_running(container) if ours else "-",
|
|
)
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
args = parse_args(argv)
|
|
logging.basicConfig(level=logging.INFO, format="%(message)s", stream=sys.stdout)
|
|
logging.getLogger("httpx").setLevel(logging.WARNING)
|
|
|
|
client = connect(args.base_url)
|
|
if args.dry_run:
|
|
dry_run(client, args)
|
|
return 0
|
|
|
|
# The webhook is mounted without authentication, so its secret is the only
|
|
# thing guarding it — even for a fixture that lives for a few minutes.
|
|
secret = token.token_hex(16)
|
|
hook = f"{args.base_url.rstrip('/')}/hooks/{FLOW_LOAD}/tick/{secret}"
|
|
wanted = ("load", "redis", "engine") if args.scenario == "all" else (args.scenario,)
|
|
reports: list[Report] = []
|
|
failure = ""
|
|
|
|
try:
|
|
# A previous run that died before its cleanup would otherwise collide.
|
|
delete_flows(client)
|
|
if args.scenario != "broker":
|
|
create_flow(client, load_flow(secret), {"count": COUNT_SOURCE})
|
|
create_flow(client, chain_flow())
|
|
if args.scenario in ("all", "broker") and is_ours(BROKER):
|
|
create_flow(client, mqtt_flow())
|
|
|
|
if "load" in wanted:
|
|
reports.append(scenario_load(client, args, hook))
|
|
if "redis" in wanted:
|
|
reports.append(scenario_redis(client, args, hook))
|
|
if "engine" in wanted:
|
|
reports.append(scenario_engine(client, args, hook))
|
|
if args.scenario in ("all", "broker"):
|
|
reports.append(scenario_broker(client))
|
|
except KeyboardInterrupt:
|
|
failure = "interrupted"
|
|
except Exception as exc:
|
|
failure = f"{type(exc).__name__}: {exc}"
|
|
log.exception("the run stopped early")
|
|
finally:
|
|
cleanup(client)
|
|
|
|
checks = sum(len(report.checks) for report in reports)
|
|
ok = bool(reports) and not failure and all(report.ok for report in reports)
|
|
log.info("")
|
|
log.info(
|
|
"SOAK %s — %d check(s) across %d scenario(s)%s",
|
|
"PASS" if ok else "FAIL",
|
|
checks,
|
|
len(reports),
|
|
f" — {failure}" if failure else "",
|
|
)
|
|
return 0 if ok else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|