453 lines
16 KiB
Python
453 lines
16 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import asyncio
|
|
import csv
|
|
import json
|
|
import math
|
|
import statistics
|
|
import time
|
|
from collections import Counter
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
|
|
PROFILE_DEFAULTS = {
|
|
"baseline_100_100": {"voice": 100, "digital": 100, "ramp_seconds": 15, "hold_seconds": 60},
|
|
"step_250_250": {"voice": 250, "digital": 250, "ramp_seconds": 30, "hold_seconds": 120},
|
|
"target_500_500": {"voice": 500, "digital": 500, "ramp_seconds": 60, "hold_seconds": 180},
|
|
}
|
|
DIGITAL_DISTRIBUTION = (("telegram", 0.4), ("webchat", 0.3), ("email", 0.3))
|
|
READ_TRAFFIC_INTERVAL_SECONDS = 1.0
|
|
MAX_ERROR_SAMPLES = 200
|
|
WORKER_PAUSE_SECONDS = 1.0
|
|
|
|
|
|
def resolve_profile(
|
|
name: str,
|
|
ramp_seconds: int | None = None,
|
|
hold_seconds: int | None = None,
|
|
voice: int | None = None,
|
|
digital: int | None = None,
|
|
) -> dict[str, int | str]:
|
|
if name not in PROFILE_DEFAULTS:
|
|
raise ValueError(f"Unsupported profile: {name}")
|
|
profile = dict(PROFILE_DEFAULTS[name])
|
|
profile["profile"] = name
|
|
if ramp_seconds is not None:
|
|
profile["ramp_seconds"] = ramp_seconds
|
|
if hold_seconds is not None:
|
|
profile["hold_seconds"] = hold_seconds
|
|
if voice is not None:
|
|
profile["voice"] = voice
|
|
if digital is not None:
|
|
profile["digital"] = digital
|
|
return profile
|
|
|
|
|
|
def split_digital_mix(total: int) -> dict[str, int]:
|
|
remaining = total
|
|
result: dict[str, int] = {}
|
|
for idx, (name, ratio) in enumerate(DIGITAL_DISTRIBUTION):
|
|
if idx == len(DIGITAL_DISTRIBUTION) - 1:
|
|
count = remaining
|
|
else:
|
|
count = int(round(total * ratio))
|
|
remaining -= count
|
|
result[name] = count
|
|
return result
|
|
|
|
|
|
def build_mix_profile(profile: dict[str, int | str], include_read_traffic: bool) -> dict[str, Any]:
|
|
voice = int(profile["voice"])
|
|
digital = int(profile["digital"])
|
|
digital_mix = split_digital_mix(digital)
|
|
payload = {
|
|
"profile": profile["profile"],
|
|
"voice_workers": voice,
|
|
"digital_workers": digital,
|
|
"ramp_seconds": int(profile["ramp_seconds"]),
|
|
"hold_seconds": int(profile["hold_seconds"]),
|
|
"channels": {"voice": voice, **digital_mix},
|
|
"include_read_traffic": include_read_traffic,
|
|
"read_endpoints": [
|
|
"/proxy/supervisor/supervisor/realtime",
|
|
"/proxy/reporting/reports/kpi",
|
|
]
|
|
if include_read_traffic
|
|
else [],
|
|
}
|
|
return payload
|
|
|
|
|
|
def _build_voice_payload(idx: int) -> tuple[str, str, dict[str, Any]]:
|
|
return (
|
|
"POST",
|
|
"/proxy/interaction/interactions",
|
|
{
|
|
"channel": "voice",
|
|
"subject": f"Track7 voice load #{idx}",
|
|
"customer_id": None,
|
|
"queue_id": "q_load_voice",
|
|
"priority": 3,
|
|
},
|
|
)
|
|
|
|
|
|
def _build_telegram_payload(idx: int) -> tuple[str, str, dict[str, Any]]:
|
|
return (
|
|
"POST",
|
|
"/proxy/telegram/integrations/telegram/webhook",
|
|
{
|
|
"chat_id": f"load_chat_{idx}",
|
|
"text": f"Track7 telegram load #{idx}",
|
|
"customer_external_id": None,
|
|
"payload": {"source": "track7-load"},
|
|
},
|
|
)
|
|
|
|
|
|
def _build_webchat_payload(idx: int) -> tuple[str, str, dict[str, Any]]:
|
|
return (
|
|
"POST",
|
|
"/proxy/webchat/integrations/webchat/messages",
|
|
{
|
|
"session_id": f"load_session_{idx}",
|
|
"text": f"Track7 webchat load #{idx}",
|
|
"visitor_name": "Load Test Visitor",
|
|
"customer_external_id": None,
|
|
"queue_id": "q_load_webchat",
|
|
"priority": 3,
|
|
"payload": {"source": "track7-load"},
|
|
},
|
|
)
|
|
|
|
|
|
def _build_email_payload(idx: int) -> tuple[str, str, dict[str, Any]]:
|
|
return (
|
|
"POST",
|
|
"/proxy/email/integrations/email/messages",
|
|
{
|
|
"from_email": f"load{idx}@example.test",
|
|
"subject": f"Track7 email load #{idx}",
|
|
"body": "Synthetic email for scale validation.",
|
|
"customer_external_id": None,
|
|
"queue_id": "q_load_email",
|
|
"priority": 3,
|
|
"payload": {"source": "track7-load"},
|
|
},
|
|
)
|
|
|
|
|
|
def build_request(target: str, idx: int) -> tuple[str, str, dict[str, Any]]:
|
|
if target == "voice":
|
|
return _build_voice_payload(idx)
|
|
if target == "telegram":
|
|
return _build_telegram_payload(idx)
|
|
if target == "webchat":
|
|
return _build_webchat_payload(idx)
|
|
if target == "email":
|
|
return _build_email_payload(idx)
|
|
raise ValueError(f"Unsupported target: {target}")
|
|
|
|
|
|
def build_worker_targets(profile: dict[str, int | str]) -> list[str]:
|
|
targets = ["voice"] * int(profile["voice"])
|
|
digital_mix = split_digital_mix(int(profile["digital"]))
|
|
for name, count in digital_mix.items():
|
|
targets.extend([name] * count)
|
|
return targets
|
|
|
|
|
|
def percentile(samples: list[float], ratio: float) -> float:
|
|
if not samples:
|
|
return 0.0
|
|
sorted_samples = sorted(samples)
|
|
index = max(0, math.ceil(len(sorted_samples) * ratio) - 1)
|
|
return sorted_samples[index]
|
|
|
|
|
|
def ensure_report_dir(report_dir: str | None = None) -> Path:
|
|
if report_dir:
|
|
target = Path(report_dir)
|
|
else:
|
|
timestamp = time.strftime("%Y%m%d_%H%M%S")
|
|
target = Path(".artifacts") / "track7" / timestamp
|
|
target.mkdir(parents=True, exist_ok=True)
|
|
return target
|
|
|
|
|
|
async def build_auth_headers(client: httpx.AsyncClient, auth_mode: str) -> dict[str, str]:
|
|
if auth_mode == "legacy_headers":
|
|
return {"X-User": "admin", "X-Role": "admin"}
|
|
if auth_mode != "bearer":
|
|
raise ValueError(f"Unsupported auth mode: {auth_mode}")
|
|
|
|
response = await client.post(
|
|
"/proxy/auth/auth/login",
|
|
json={"username": "admin", "password": "admin123"},
|
|
timeout=10,
|
|
)
|
|
response.raise_for_status()
|
|
token = response.json()["access_token"]
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
|
|
def _record_result(
|
|
results: list[dict[str, Any]],
|
|
errors: list[dict[str, Any]],
|
|
target: str,
|
|
started_at: float,
|
|
elapsed: float,
|
|
status_code: int,
|
|
ok: bool,
|
|
error_message: str | None = None,
|
|
) -> None:
|
|
results.append(
|
|
{
|
|
"ts": round(started_at, 6),
|
|
"target": target,
|
|
"status_code": status_code,
|
|
"latency_seconds": round(elapsed, 6),
|
|
"ok": ok,
|
|
}
|
|
)
|
|
if (not ok or status_code >= 500 or error_message) and len(errors) < MAX_ERROR_SAMPLES:
|
|
errors.append(
|
|
{
|
|
"ts": round(started_at, 6),
|
|
"target": target,
|
|
"status_code": status_code,
|
|
"latency_seconds": round(elapsed, 6),
|
|
"error": error_message,
|
|
}
|
|
)
|
|
|
|
|
|
async def _run_worker(
|
|
client: httpx.AsyncClient,
|
|
headers: dict[str, str],
|
|
target: str,
|
|
idx: int,
|
|
stop_event: asyncio.Event,
|
|
results: list[dict[str, Any]],
|
|
errors: list[dict[str, Any]],
|
|
) -> None:
|
|
method, path, payload = build_request(target, idx)
|
|
while not stop_event.is_set():
|
|
started = time.perf_counter()
|
|
try:
|
|
response = await client.request(method, path, headers=headers, json=payload, timeout=20)
|
|
elapsed = time.perf_counter() - started
|
|
ok = response.status_code < 500 and response.status_code < 400
|
|
_record_result(results, errors, target, started, elapsed, response.status_code, ok)
|
|
except Exception as exc:
|
|
elapsed = time.perf_counter() - started
|
|
_record_result(results, errors, target, started, elapsed, 0, False, str(exc))
|
|
await asyncio.sleep(WORKER_PAUSE_SECONDS)
|
|
|
|
|
|
async def _run_read_probe(
|
|
client: httpx.AsyncClient,
|
|
headers: dict[str, str],
|
|
path: str,
|
|
stop_event: asyncio.Event,
|
|
results: list[dict[str, Any]],
|
|
errors: list[dict[str, Any]],
|
|
) -> None:
|
|
target = f"read:{path.rsplit('/', 1)[-1]}"
|
|
while not stop_event.is_set():
|
|
started = time.perf_counter()
|
|
try:
|
|
response = await client.get(path, headers=headers, timeout=20)
|
|
elapsed = time.perf_counter() - started
|
|
ok = response.status_code < 500 and response.status_code < 400
|
|
_record_result(results, errors, target, started, elapsed, response.status_code, ok)
|
|
except Exception as exc:
|
|
elapsed = time.perf_counter() - started
|
|
_record_result(results, errors, target, started, elapsed, 0, False, str(exc))
|
|
await asyncio.sleep(READ_TRAFFIC_INTERVAL_SECONDS)
|
|
|
|
|
|
def build_summary(
|
|
results: list[dict[str, Any]],
|
|
mix_profile: dict[str, Any],
|
|
thresholds: dict[str, float | None],
|
|
started_at: str,
|
|
completed_at: str,
|
|
) -> dict[str, Any]:
|
|
latencies = [item["latency_seconds"] for item in results]
|
|
total = len(results)
|
|
success = sum(1 for item in results if item["ok"])
|
|
failures = total - success
|
|
five_xx = sum(1 for item in results if int(item["status_code"]) >= 500 or int(item["status_code"]) == 0)
|
|
by_target = Counter(item["target"] for item in results)
|
|
summary = {
|
|
"profile": mix_profile["profile"],
|
|
"started_at": started_at,
|
|
"completed_at": completed_at,
|
|
"traffic": mix_profile,
|
|
"results": {
|
|
"total_requests": total,
|
|
"success": success,
|
|
"failed": failures,
|
|
"five_xx_or_transport": five_xx,
|
|
"success_rate": round((success / total) * 100, 2) if total else 0.0,
|
|
"five_xx_rate": round((five_xx / total) * 100, 2) if total else 0.0,
|
|
"avg_seconds": round(statistics.mean(latencies), 4) if latencies else 0.0,
|
|
"p95_seconds": round(percentile(latencies, 0.95), 4) if latencies else 0.0,
|
|
"p99_seconds": round(percentile(latencies, 0.99), 4) if latencies else 0.0,
|
|
"by_target": dict(by_target),
|
|
},
|
|
"thresholds": thresholds,
|
|
}
|
|
summary["passed"] = evaluate_thresholds(summary)
|
|
return summary
|
|
|
|
|
|
def evaluate_thresholds(summary: dict[str, Any]) -> bool:
|
|
results = summary["results"]
|
|
thresholds = summary["thresholds"]
|
|
if thresholds.get("require_success_rate") is not None and results["success_rate"] < thresholds["require_success_rate"]:
|
|
return False
|
|
if thresholds.get("require_p95_seconds") is not None and results["p95_seconds"] > thresholds["require_p95_seconds"]:
|
|
return False
|
|
if thresholds.get("require_p99_seconds") is not None and results["p99_seconds"] > thresholds["require_p99_seconds"]:
|
|
return False
|
|
return True
|
|
|
|
|
|
def write_report(report_dir: Path, summary: dict[str, Any], results: list[dict[str, Any]], errors: list[dict[str, Any]]) -> None:
|
|
(report_dir / "summary.json").write_text(json.dumps(summary, indent=2), encoding="utf-8")
|
|
(report_dir / "mix_profile.json").write_text(json.dumps(summary["traffic"], indent=2), encoding="utf-8")
|
|
(report_dir / "error_samples.json").write_text(json.dumps(errors, indent=2), encoding="utf-8")
|
|
with (report_dir / "latency_samples.csv").open("w", encoding="utf-8", newline="") as handle:
|
|
writer = csv.DictWriter(handle, fieldnames=["ts", "target", "status_code", "latency_seconds", "ok"])
|
|
writer.writeheader()
|
|
writer.writerows(results)
|
|
|
|
|
|
async def run_profile(
|
|
*,
|
|
base_url: str,
|
|
profile_name: str,
|
|
ramp_seconds: int | None,
|
|
hold_seconds: int | None,
|
|
voice: int | None,
|
|
digital: int | None,
|
|
include_read_traffic: bool,
|
|
auth_mode: str,
|
|
report_dir: str | None,
|
|
require_success_rate: float | None,
|
|
require_p95_seconds: float | None,
|
|
require_p99_seconds: float | None,
|
|
) -> tuple[dict[str, Any], Path]:
|
|
profile = resolve_profile(
|
|
profile_name,
|
|
ramp_seconds=ramp_seconds,
|
|
hold_seconds=hold_seconds,
|
|
voice=voice,
|
|
digital=digital,
|
|
)
|
|
mix_profile = build_mix_profile(profile, include_read_traffic)
|
|
out_dir = ensure_report_dir(report_dir)
|
|
started_at_iso = time.strftime("%Y-%m-%dT%H:%M:%S")
|
|
results: list[dict[str, Any]] = []
|
|
errors: list[dict[str, Any]] = []
|
|
stop_event = asyncio.Event()
|
|
worker_targets = build_worker_targets(profile)
|
|
|
|
async with httpx.AsyncClient(base_url=base_url) as client:
|
|
headers = await build_auth_headers(client, auth_mode)
|
|
workers: list[asyncio.Task[None]] = []
|
|
ramp_window = max(1, int(profile["ramp_seconds"]))
|
|
spacing = ramp_window / max(1, len(worker_targets))
|
|
for idx, target in enumerate(worker_targets):
|
|
workers.append(asyncio.create_task(_run_worker(client, headers, target, idx, stop_event, results, errors)))
|
|
if spacing > 0:
|
|
await asyncio.sleep(spacing)
|
|
|
|
read_tasks: list[asyncio.Task[None]] = []
|
|
if include_read_traffic:
|
|
read_tasks = [
|
|
asyncio.create_task(
|
|
_run_read_probe(client, headers, "/proxy/supervisor/supervisor/realtime", stop_event, results, errors)
|
|
),
|
|
asyncio.create_task(
|
|
_run_read_probe(client, headers, "/proxy/reporting/reports/kpi", stop_event, results, errors)
|
|
),
|
|
]
|
|
|
|
await asyncio.sleep(int(profile["hold_seconds"]))
|
|
stop_event.set()
|
|
await asyncio.gather(*workers, *read_tasks, return_exceptions=True)
|
|
|
|
completed_at_iso = time.strftime("%Y-%m-%dT%H:%M:%S")
|
|
thresholds = {
|
|
"require_success_rate": require_success_rate,
|
|
"require_p95_seconds": require_p95_seconds,
|
|
"require_p99_seconds": require_p99_seconds,
|
|
}
|
|
summary = build_summary(results, mix_profile, thresholds, started_at_iso, completed_at_iso)
|
|
write_report(out_dir, summary, results, errors)
|
|
return summary, out_dir
|
|
|
|
|
|
async def main() -> None:
|
|
parser = argparse.ArgumentParser(description="Track 7 mixed workload load harness through the gateway.")
|
|
parser.add_argument("--base-url", default="http://localhost:8080", help="Gateway URL")
|
|
parser.add_argument(
|
|
"--profile",
|
|
default="baseline_100_100",
|
|
choices=sorted(PROFILE_DEFAULTS),
|
|
help="Named load profile",
|
|
)
|
|
parser.add_argument("--ramp-seconds", type=int, default=None, help="Override ramp duration")
|
|
parser.add_argument("--hold-seconds", type=int, default=None, help="Override hold duration")
|
|
parser.add_argument("--voice", type=int, default=None, help="Backward-compatible voice worker override")
|
|
parser.add_argument("--digital", type=int, default=None, help="Backward-compatible digital worker override")
|
|
parser.add_argument("--report-dir", default=None, help="Directory for JSON/CSV reports")
|
|
parser.add_argument("--require-success-rate", type=float, default=None, help="Optional success-rate threshold")
|
|
parser.add_argument("--require-p95-seconds", type=float, default=None, help="Optional p95 threshold")
|
|
parser.add_argument("--require-p99-seconds", type=float, default=None, help="Optional p99 threshold")
|
|
parser.add_argument(
|
|
"--include-read-traffic",
|
|
type=int,
|
|
choices=[0, 1],
|
|
default=1,
|
|
help="Include background supervisor/reporting reads during the hold window",
|
|
)
|
|
parser.add_argument(
|
|
"--auth-mode",
|
|
choices=["legacy_headers", "bearer"],
|
|
default="legacy_headers",
|
|
help="Authentication mode for load traffic",
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
summary, out_dir = await run_profile(
|
|
base_url=args.base_url,
|
|
profile_name=args.profile,
|
|
ramp_seconds=args.ramp_seconds,
|
|
hold_seconds=args.hold_seconds,
|
|
voice=args.voice,
|
|
digital=args.digital,
|
|
include_read_traffic=bool(args.include_read_traffic),
|
|
auth_mode=args.auth_mode,
|
|
report_dir=args.report_dir,
|
|
require_success_rate=args.require_success_rate,
|
|
require_p95_seconds=args.require_p95_seconds,
|
|
require_p99_seconds=args.require_p99_seconds,
|
|
)
|
|
print("Track 7 load test result:")
|
|
print(json.dumps(summary, indent=2))
|
|
print(f"Report dir: {out_dir}")
|
|
if not summary["passed"]:
|
|
raise SystemExit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|