call-center/scripts/track7_check.py

from __future__ import annotations

import argparse
import json
import subprocess
from pathlib import Path
from typing import Any


def load_summary(report_dir: Path) -> dict[str, Any]:
    summary_path = report_dir / "summary.json"
    if not summary_path.exists():
        raise FileNotFoundError(f"Missing summary.json: {summary_path}")
    return json.loads(summary_path.read_text(encoding="utf-8"))


def evaluate_summary(
    summary: dict[str, Any],
    *,
    require_success_rate: float,
    require_p95_seconds: float,
    require_p99_seconds: float,
) -> list[str]:
    issues: list[str] = []
    results = summary.get("results", {})
    success_rate = float(results.get("success_rate", 0.0))
    p95 = float(results.get("p95_seconds", 0.0))
    p99 = float(results.get("p99_seconds", 0.0))
    five_xx_rate = float(results.get("five_xx_rate", 0.0))

    if success_rate < require_success_rate:
        issues.append(f"Success rate {success_rate:.2f}% is below threshold {require_success_rate:.2f}%")
    if p95 > require_p95_seconds:
        issues.append(f"P95 {p95:.2f}s exceeds threshold {require_p95_seconds:.2f}s")
    if p99 > require_p99_seconds:
        issues.append(f"P99 {p99:.2f}s exceeds threshold {require_p99_seconds:.2f}s")
    if five_xx_rate > 0.5:
        issues.append(f"5xx/transport error rate {five_xx_rate:.2f}% exceeds 0.50%")
    return issues


def _run_kubectl(args: list[str]) -> subprocess.CompletedProcess[str]:
    return subprocess.run(args, capture_output=True, text=True, check=False)


def check_pod_health(namespace: str, kubectl_context: str | None = None) -> list[str]:
    cmd = ["kubectl"]
    if kubectl_context:
        cmd.extend(["--context", kubectl_context])
    cmd.extend(["get", "pods", "-n", namespace, "-o", "json"])
    result = _run_kubectl(cmd)
    if result.returncode != 0:
        return [f"kubectl get pods failed: {result.stderr.strip() or result.stdout.strip()}"]

    payload = json.loads(result.stdout)
    issues: list[str] = []
    for item in payload.get("items", []):
        name = item.get("metadata", {}).get("name", "<unknown>")
        status = item.get("status", {})
        phase = status.get("phase")
        if phase not in {"Running", "Succeeded"}:
            issues.append(f"Pod {name} is in unexpected phase {phase}")
        for container in status.get("containerStatuses", []):
            if container.get("restartCount", 0) > 0:
                issues.append(f"Pod {name} restartCount={container.get('restartCount')}")
            state = container.get("state", {})
            waiting = state.get("waiting") or {}
            terminated = state.get("terminated") or {}
            if waiting.get("reason") == "CrashLoopBackOff":
                issues.append(f"Pod {name} is in CrashLoopBackOff")
            if terminated.get("reason") == "OOMKilled":
                issues.append(f"Pod {name} had OOMKilled termination")
    return issues


def check_kubectl_top(namespace: str, kubectl_context: str | None = None) -> list[str]:
    cmd = ["kubectl"]
    if kubectl_context:
        cmd.extend(["--context", kubectl_context])
    cmd.extend(["top", "pods", "-n", namespace, "--no-headers"])
    result = _run_kubectl(cmd)
    if result.returncode != 0:
        return [f"kubectl top pods failed: {result.stderr.strip() or result.stdout.strip()}"]
    if not result.stdout.strip():
        return ["kubectl top pods returned no rows"]
    return []


def check_hpa(namespace: str, kubectl_context: str | None = None) -> list[str]:
    cmd = ["kubectl"]
    if kubectl_context:
        cmd.extend(["--context", kubectl_context])
    cmd.extend(["get", "hpa", "-n", namespace, "-o", "json"])
    result = _run_kubectl(cmd)
    if result.returncode != 0:
        return [f"kubectl get hpa failed: {result.stderr.strip() or result.stdout.strip()}"]

    payload = json.loads(result.stdout)
    items = payload.get("items", [])
    if not items:
        return ["No HPA objects found in namespace"]

    issues: list[str] = []
    for item in items:
        name = item.get("metadata", {}).get("name", "<unknown>")
        status = item.get("status", {})
        current_metrics = status.get("currentMetrics") or []
        if not current_metrics and status.get("currentCPUUtilizationPercentage") is None:
            issues.append(f"HPA {name} has no current utilization metrics")
    return issues


def main() -> None:
    parser = argparse.ArgumentParser(description="Validate Track 7 scale/hardening evidence.")
    parser.add_argument("--namespace", required=True, help="Kubernetes namespace")
    parser.add_argument("--report-dir", required=True, help="Directory produced by scripts/load_test.py")
    parser.add_argument("--require-success-rate", type=float, default=99.0)
    parser.add_argument("--require-p95-seconds", type=float, default=2.0)
    parser.add_argument("--require-p99-seconds", type=float, default=3.5)
    parser.add_argument("--kubectl-context", default=None)
    args = parser.parse_args()

    report_dir = Path(args.report_dir)
    issues: list[str] = []
    try:
        summary = load_summary(report_dir)
    except Exception as exc:
        raise SystemExit(f"[FAIL] {exc}") from exc

    issues.extend(
        evaluate_summary(
            summary,
            require_success_rate=args.require_success_rate,
            require_p95_seconds=args.require_p95_seconds,
            require_p99_seconds=args.require_p99_seconds,
        )
    )
    issues.extend(check_pod_health(args.namespace, kubectl_context=args.kubectl_context))
    issues.extend(check_kubectl_top(args.namespace, kubectl_context=args.kubectl_context))
    issues.extend(check_hpa(args.namespace, kubectl_context=args.kubectl_context))

    if issues:
        print("[FAIL] Track 7 validation failed:")
        for issue in issues:
            print(f"- {issue}")
        raise SystemExit(1)

    print("[PASS] Track 7 validation checks passed")
    print(f"- namespace: {args.namespace}")
    print(f"- report_dir: {report_dir}")


if __name__ == "__main__":
    main()