feat: add alpha load-test methodology

deploy/load-smoke.py

@@ -0,0 +1,80 @@
+#!/usr/bin/env python3
+"""Small read-only latency probe for alpha sizing; not a capacity benchmark."""
+from __future__ import annotations
+
+import argparse
+import concurrent.futures
+import json
+import math
+import os
+import statistics
+import time
+import urllib.error
+import urllib.request
+
+
+PUBLIC = {
+    "activity": "/api/v1/activity?hours=72&limit=20",
+    "records": "/api/v1/records?limit=50",
+}
+ADMIN = {
+    "staging": "/api/v1/admin/external-observations?status=review&limit=50",
+    "moderation": "/api/v1/admin/catch-reports?status=pending&limit=50",
+}
+
+
+def percentile(values: list[float], percent: float) -> float:
+    return sorted(values)[max(0, math.ceil(len(values) * percent) - 1)]
+
+
+def request_once(url: str, token: str | None, timeout: float) -> tuple[float, int]:
+    headers = {"Accept": "application/json", "User-Agent": "RF4-Spotter-load-smoke/0.1"}
+    if token:
+        headers["Authorization"] = f"Bearer {token}"
+    started = time.perf_counter()
+    try:
+        with urllib.request.urlopen(urllib.request.Request(url, headers=headers), timeout=timeout) as response:
+            response.read()
+            status = response.status
+    except urllib.error.HTTPError as exc:
+        exc.read()
+        status = exc.code
+    return (time.perf_counter() - started) * 1000, status
+
+
+def probe(name: str, url: str, *, token: str | None, requests: int, concurrency: int, timeout: float) -> dict:
+    request_once(url, token, timeout)  # one explicit warm-up excluded from statistics
+    with concurrent.futures.ThreadPoolExecutor(max_workers=concurrency) as pool:
+        samples = list(pool.map(lambda _: request_once(url, token, timeout), range(requests)))
+    latencies = [sample[0] for sample in samples]
+    statuses: dict[str, int] = {}
+    for _, status in samples:
+        statuses[str(status)] = statuses.get(str(status), 0) + 1
+    return {
+        "scenario": name, "requests": requests, "concurrency": concurrency,
+        "p50_ms": round(statistics.median(latencies), 2),
+        "p95_ms": round(percentile(latencies, .95), 2),
+        "max_ms": round(max(latencies), 2), "statuses": statuses,
+    }
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="Read-only RF4 alpha latency probe")
+    parser.add_argument("--base-url", required=True)
+    parser.add_argument("--requests", type=int, default=100)
+    parser.add_argument("--concurrency", type=int, default=5)
+    parser.add_argument("--timeout", type=float, default=10)
+    args = parser.parse_args()
+    if not 1 <= args.requests <= 10_000 or not 1 <= args.concurrency <= 100:
+        parser.error("requests must be 1..10000 and concurrency 1..100")
+    base = args.base_url.rstrip("/")
+    token = os.environ.get("ADMIN_TOKEN")
+    results = [probe(name, base + path, token=None, requests=args.requests, concurrency=args.concurrency, timeout=args.timeout) for name, path in PUBLIC.items()]
+    if token:
+        results.extend(probe(name, base + path, token=token, requests=args.requests, concurrency=args.concurrency, timeout=args.timeout) for name, path in ADMIN.items())
+    print(json.dumps({"base_url": base, "admin_scenarios_skipped": not bool(token), "results": results}, ensure_ascii=False, indent=2))
+    return 1 if any(any(not status.startswith("2") for status in result["statuses"]) for result in results) else 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())