81 lines
3.3 KiB
Python
81 lines
3.3 KiB
Python
#!/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())
|