from __future__ import annotations import argparse import asyncio import json import sys import time from pathlib import Path from typing import Any if __package__ in {None, ""}: sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from scripts import load_test, track7_check else: from scripts import load_test, track7_check PROFILE_THRESHOLDS = { "step_250_250": { "require_success_rate": 99.0, "require_p95_seconds": 1.5, "require_p99_seconds": 3.0, }, "target_500_500": { "require_success_rate": 99.0, "require_p95_seconds": 2.0, "require_p99_seconds": 3.5, }, } def build_stage_root(report_root: str | None = None) -> Path: if report_root: target = Path(report_root) else: timestamp = time.strftime("%Y%m%d_%H%M%S") target = Path(".artifacts") / "track7" / f"staged_{timestamp}" target.mkdir(parents=True, exist_ok=True) return target def resolve_profiles(profiles_raw: str) -> list[str]: requested = [item.strip() for item in profiles_raw.split(",") if item.strip()] if not requested: raise ValueError("At least one profile is required") unsupported = [name for name in requested if name not in PROFILE_THRESHOLDS] if unsupported: raise ValueError(f"Unsupported staged profiles: {', '.join(unsupported)}") return requested def evaluate_stage( *, namespace: str, report_dir: Path, kubectl_context: str | None, thresholds: dict[str, float], ) -> list[str]: summary = track7_check.load_summary(report_dir) issues: list[str] = [] issues.extend( track7_check.evaluate_summary( summary, require_success_rate=thresholds["require_success_rate"], require_p95_seconds=thresholds["require_p95_seconds"], require_p99_seconds=thresholds["require_p99_seconds"], ) ) issues.extend(track7_check.check_pod_health(namespace, kubectl_context=kubectl_context)) issues.extend(track7_check.check_kubectl_top(namespace, kubectl_context=kubectl_context)) issues.extend(track7_check.check_hpa(namespace, kubectl_context=kubectl_context)) return issues def write_acceptance_pack(stage_root: Path, payload: dict[str, Any]) -> None: (stage_root / "acceptance_summary.json").write_text(json.dumps(payload, indent=2), encoding="utf-8") lines = [ "# Track 7 Staged Validation", "", f"- Namespace: `{payload['namespace']}`", f"- Base URL: `{payload['base_url']}`", f"- Auth mode: `{payload['auth_mode']}`", f"- Include read traffic: `{int(payload['include_read_traffic'])}`", f"- Started at: `{payload['started_at']}`", f"- Completed at: `{payload['completed_at']}`", f"- Passed: `{payload['passed']}`", "", "## Stages", "", ] for stage in payload["stages"]: lines.extend( [ f"### {stage['profile']}", "", f"- Report dir: `{stage['report_dir']}`", f"- Passed: `{stage['passed']}`", f"- Success rate: `{stage['results']['success_rate']:.2f}%`", f"- P95: `{stage['results']['p95_seconds']:.4f}s`", f"- P99: `{stage['results']['p99_seconds']:.4f}s`", ] ) if stage["issues"]: lines.append("- Issues:") for issue in stage["issues"]: lines.append(f" - {issue}") lines.append("") (stage_root / "acceptance_summary.md").write_text("\n".join(lines), encoding="utf-8") async def run_staged_validation( *, base_url: str, namespace: str, profiles: list[str], auth_mode: str, include_read_traffic: bool, kubectl_context: str | None, report_root: str | None, ) -> tuple[dict[str, Any], Path]: stage_root = build_stage_root(report_root) started_at = time.strftime("%Y-%m-%dT%H:%M:%S") stages: list[dict[str, Any]] = [] for profile_name in profiles: thresholds = PROFILE_THRESHOLDS[profile_name] report_dir = stage_root / profile_name summary, out_dir = await load_test.run_profile( base_url=base_url, profile_name=profile_name, ramp_seconds=None, hold_seconds=None, voice=None, digital=None, include_read_traffic=include_read_traffic, auth_mode=auth_mode, report_dir=str(report_dir), require_success_rate=thresholds["require_success_rate"], require_p95_seconds=thresholds["require_p95_seconds"], require_p99_seconds=thresholds["require_p99_seconds"], ) issues = evaluate_stage( namespace=namespace, report_dir=out_dir, kubectl_context=kubectl_context, thresholds=thresholds, ) stage_payload = { "profile": profile_name, "report_dir": str(out_dir), "thresholds": thresholds, "results": summary["results"], "passed": not issues, "issues": issues, } stages.append(stage_payload) if issues: break completed_at = time.strftime("%Y-%m-%dT%H:%M:%S") payload = { "namespace": namespace, "base_url": base_url, "auth_mode": auth_mode, "include_read_traffic": include_read_traffic, "started_at": started_at, "completed_at": completed_at, "stages": stages, "passed": all(stage["passed"] for stage in stages) and len(stages) == len(profiles), } write_acceptance_pack(stage_root, payload) return payload, stage_root async def main() -> None: parser = argparse.ArgumentParser(description="Run staged Track 7 validation and write an acceptance pack.") parser.add_argument("--base-url", required=True, help="Gateway base URL") parser.add_argument("--namespace", required=True, help="Kubernetes namespace to validate") parser.add_argument( "--profiles", default="step_250_250,target_500_500", help="Comma-separated staged profiles. Supported: step_250_250,target_500_500", ) parser.add_argument( "--auth-mode", choices=["legacy_headers", "bearer"], default="bearer", help="Authentication mode for load traffic", ) parser.add_argument( "--include-read-traffic", type=int, choices=[0, 1], default=1, help="Include background supervisor/reporting reads during staged runs", ) parser.add_argument("--kubectl-context", default=None, help="Optional kubectl context override") parser.add_argument("--report-root", default=None, help="Root directory for stage artifacts") args = parser.parse_args() profiles = resolve_profiles(args.profiles) payload, stage_root = await run_staged_validation( base_url=args.base_url, namespace=args.namespace, profiles=profiles, auth_mode=args.auth_mode, include_read_traffic=bool(args.include_read_traffic), kubectl_context=args.kubectl_context, report_root=args.report_root, ) print("Track 7 staged validation:") print(json.dumps(payload, indent=2)) print(f"Acceptance pack: {stage_root}") if not payload["passed"]: raise SystemExit(1) if __name__ == "__main__": asyncio.run(main())