154 lines
5.5 KiB
Python
154 lines
5.5 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""可用性监控探针 —— 目标 ≥ 99.8%（issue #61 / PRD 5.6「⑥ 部署底座」）。
|
||
|
||
对已部署服务（推理服务 /health、驾驶舱等）按周期轮询健康端点，
|
||
统计窗口内可用率（success / total），断言 ≥ 99.8% 验收阈值：
|
||
|
||
- 探针只读（GET /health），不修改任何状态；
|
||
- 每次探测记录：时间戳 / 状态 / 延迟；
|
||
- 报告输出：窗口可用率、总探测数、失败明细、最近一次延迟；
|
||
- 可作为 CronJob 或边车（sidecar）周期性运行，结果喂给监控告警。
|
||
|
||
用法（CLI）：
|
||
python probe_availability.py --endpoints http://localhost:8000/health \
|
||
--rounds 60 --interval 1 --target 0.998
|
||
退出码：0 = 可用率达标；1 = 低于目标。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import sys
|
||
import time
|
||
import urllib.request
|
||
from dataclasses import dataclass, field
|
||
from typing import List, Optional, Sequence, Tuple
|
||
|
||
|
||
@dataclass
|
||
class ProbeResult:
|
||
"""单次探测结果。"""
|
||
|
||
url: str
|
||
ok: bool
|
||
latency_ms: float = 0.0
|
||
detail: str = ""
|
||
|
||
|
||
@dataclass
|
||
class ProbeReport:
|
||
"""窗口统计报告。"""
|
||
|
||
total: int = 0
|
||
success: int = 0
|
||
availability: float = 0.0
|
||
target: float = 0.998
|
||
failures: List[ProbeResult] = field(default_factory=list)
|
||
last_latency_ms: float = 0.0
|
||
|
||
def meets_target(self) -> bool:
|
||
"""可用率 ≥ 目标（默认 99.8%）。"""
|
||
return self.availability >= self.target
|
||
|
||
def to_dict(self) -> dict:
|
||
return {
|
||
"total": self.total,
|
||
"success": self.success,
|
||
"availability": round(self.availability, 6),
|
||
"target": self.target,
|
||
"meets_target": self.meets_target(),
|
||
"failures": len(self.failures),
|
||
"last_latency_ms": round(self.last_latency_ms, 2),
|
||
}
|
||
|
||
|
||
class AvailabilityProbe:
|
||
"""可用性探针：周期轮询健康端点，统计可用率。"""
|
||
|
||
def __init__(
|
||
self,
|
||
endpoints: Sequence[str],
|
||
timeout_seconds: float = 5.0,
|
||
target: float = 0.998,
|
||
) -> None:
|
||
self.endpoints = [u.rstrip("/") for u in endpoints]
|
||
self.timeout = float(timeout_seconds)
|
||
self.target = float(target)
|
||
|
||
# ------------------------------------------------------------------
|
||
def probe_once(self, url: str) -> ProbeResult:
|
||
"""探测单个端点：GET /health，200 即可用。"""
|
||
started = time.monotonic()
|
||
try:
|
||
with urllib.request.urlopen(
|
||
url + "/health", timeout=self.timeout) as resp:
|
||
ok = resp.status == 200
|
||
return ProbeResult(
|
||
url=url, ok=ok,
|
||
latency_ms=round((time.monotonic() - started) * 1000, 2),
|
||
detail=f"http-{resp.status}")
|
||
except Exception as exc: # noqa: BLE001 - 探测失败即视为不可用
|
||
return ProbeResult(
|
||
url=url, ok=False,
|
||
latency_ms=round((time.monotonic() - started) * 1000, 2),
|
||
detail=f"error: {exc}")
|
||
|
||
def run(
|
||
self,
|
||
rounds: int = 60,
|
||
interval: float = 1.0,
|
||
) -> ProbeReport:
|
||
"""执行 rounds 轮轮询（每轮探测全部端点），返回窗口报告。"""
|
||
report = ProbeReport(target=self.target)
|
||
for _ in range(max(1, rounds)):
|
||
for url in self.endpoints:
|
||
result = self.probe_once(url)
|
||
report.total += 1
|
||
if result.ok:
|
||
report.success += 1
|
||
report.last_latency_ms = result.latency_ms
|
||
else:
|
||
report.failures.append(result)
|
||
if interval > 0:
|
||
time.sleep(interval)
|
||
report.availability = (
|
||
report.success / report.total if report.total else 0.0)
|
||
return report
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# CLI
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def main(argv: Optional[List[str]] = None) -> int:
|
||
parser = argparse.ArgumentParser(
|
||
description="可用性监控探针（目标 ≥ 99.8%，issue #61）")
|
||
parser.add_argument("--endpoints", required=True,
|
||
help="健康端点（逗号分隔，如 http://host:8000）")
|
||
parser.add_argument("--rounds", type=int, default=60, help="轮询轮数")
|
||
parser.add_argument("--interval", type=float, default=1.0, help="轮询间隔（秒）")
|
||
parser.add_argument("--timeout", type=float, default=5.0, help="单次探测超时")
|
||
parser.add_argument("--target", type=float, default=0.998,
|
||
help="可用率目标（默认 0.998 = 99.8%）")
|
||
args = parser.parse_args(argv)
|
||
|
||
probe = AvailabilityProbe(
|
||
endpoints=[u for u in args.endpoints.split(",") if u],
|
||
timeout_seconds=args.timeout, target=args.target)
|
||
report = probe.run(rounds=args.rounds, interval=args.interval)
|
||
|
||
summary = report.to_dict()
|
||
print(f"可用率 {summary['availability']:.4%} "
|
||
f"（{summary['success']}/{summary['total']}，"
|
||
f"目标 {summary['target']:.4%}）"
|
||
f" -> {'PASS' if summary['meets_target'] else 'FAIL'}")
|
||
if report.failures:
|
||
print("失败明细（前 10 条）：")
|
||
for f in report.failures[:10]:
|
||
print(f" - {f.url}: {f.detail} ({f.latency_ms:.0f}ms)")
|
||
return 0 if report.meets_target() else 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|