feat: Implement latency profiling and guardrails for performance monitoring
CI / lint-test-build (push) Failing after 19s
CI / lint-test-build (push) Failing after 19s
- Added synthetic latency profiler scenarios and CLI scripts for baseline generation and regression checks. - Introduced latency baseline and threshold artifacts for CI enforcement. - Enhanced CI workflow with latency guardrail checks. - Updated documentation to include latency profiling commands and performance metrics. - Added unit tests for latency guardrail evaluation.
This commit is contained in:
@@ -0,0 +1,54 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
from datetime import UTC, datetime
|
||||
from pathlib import Path
|
||||
|
||||
from arbitrade.perf.latency import run_latency_profile
|
||||
|
||||
|
||||
def _format_summary(profile: dict[str, object]) -> str:
|
||||
scenarios = profile.get("scenarios")
|
||||
if not isinstance(scenarios, dict):
|
||||
return "No scenarios found."
|
||||
|
||||
lines = ["Latency profiling summary:"]
|
||||
for scenario_name, payload in scenarios.items():
|
||||
if not isinstance(payload, dict):
|
||||
continue
|
||||
lines.append(f"- {scenario_name}")
|
||||
stages = payload.get("stages")
|
||||
if not isinstance(stages, dict):
|
||||
continue
|
||||
for stage_name, stage_payload in stages.items():
|
||||
if not isinstance(stage_payload, dict):
|
||||
continue
|
||||
p95 = float(stage_payload.get("p95_ms", 0.0))
|
||||
p99 = float(stage_payload.get("p99_ms", 0.0))
|
||||
lines.append(f" - {stage_name}: p95={p95:.4f}ms p99={p99:.4f}ms")
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description="Profile synthetic latency scenarios.")
|
||||
parser.add_argument("--iterations", type=int, default=600)
|
||||
parser.add_argument("--output", type=Path, default=None)
|
||||
args = parser.parse_args()
|
||||
|
||||
profile = run_latency_profile(iterations=args.iterations)
|
||||
profile["generated_at"] = datetime.now(UTC).isoformat()
|
||||
|
||||
print(_format_summary(profile))
|
||||
|
||||
if args.output is not None:
|
||||
args.output.parent.mkdir(parents=True, exist_ok=True)
|
||||
args.output.write_text(json.dumps(profile, indent=2), encoding="utf-8")
|
||||
print(f"Wrote profile JSON to {args.output}")
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user