feat: 完成 issue #61 ⑥ 可用性监控探针（≥99.8%）
This commit is contained in:
@@ -0,0 +1,55 @@
|
||||
# 可用性监控探针（Availability Probe）
|
||||
|
||||
对应 EPIC #8「⑥ 部署底座」子任务 **#61**（0.5d，对齐 PRD 5.6 验收）：
|
||||
对已部署服务按周期轮询健康端点（`GET /health`），统计窗口可用率，
|
||||
目标 **≥ 99.8%**。
|
||||
|
||||
## 文件
|
||||
|
||||
```
|
||||
deploy/k8s/healthz/
|
||||
├── probe_availability.py 探针实现（AvailabilityProbe + CLI）
|
||||
├── test_probe.py 单元测试（本地 HTTP 端点模拟）
|
||||
└── README.md
|
||||
```
|
||||
|
||||
## 用法（CLI）
|
||||
|
||||
```bash
|
||||
# 轮询 60 轮（每 1s 一次），目标 99.8%
|
||||
python probe_availability.py --endpoints http://10.20.0.30:8000 \
|
||||
--rounds 60 --interval 1 --target 0.998
|
||||
|
||||
# 多端点
|
||||
python probe_availability.py --endpoints http://a:8000,http://b:8000 \
|
||||
--rounds 120 --interval 5
|
||||
```
|
||||
|
||||
退出码：`0` = 可用率达标（PASS）；`1` = 低于目标（FAIL），
|
||||
供 CronJob / 监控告警使用。
|
||||
|
||||
## 编程接口
|
||||
|
||||
```python
|
||||
from probe_availability import AvailabilityProbe
|
||||
|
||||
probe = AvailabilityProbe(["http://iaop-inference:8000"])
|
||||
report = probe.run(rounds=60, interval=1)
|
||||
print(report.to_dict())
|
||||
# {'total': 60, 'success': 60, 'availability': 1.0,
|
||||
# 'target': 0.998, 'meets_target': True, 'failures': 0, ...}
|
||||
```
|
||||
|
||||
## 验收口径（PRD 5.6）
|
||||
|
||||
- 可用率 = 成功探测数 / 总探测数，窗口内 ≥ 99.8%；
|
||||
- 探针**只读**（GET /health），不影响服务状态；
|
||||
- 失败明细（URL/原因/延迟）随报告输出，便于定位。
|
||||
|
||||
## 测试
|
||||
|
||||
```bash
|
||||
python -m unittest test_probe -v
|
||||
```
|
||||
覆盖：稳定端点 100% 达标、单次故障（1/500=0.2%）恰好达标、
|
||||
50% 故障率判定 FAIL、失败明细记录。
|
||||
@@ -0,0 +1,153 @@
|
||||
# -*- 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())
|
||||
@@ -0,0 +1,104 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""可用性监控探针测试（issue #61）。
|
||||
|
||||
用本地 ThreadingHTTPServer 模拟健康端点：
|
||||
1. 稳定 200 → 可用率 100% ≥ 99.8%；
|
||||
2. 间歇 503 → 可用率按失败比例下降，探针统计正确；
|
||||
3. 阈值判断：间歇失败超 0.2% 时 meets_target=False；
|
||||
4. 失败明细记录（URL + 原因）。
|
||||
"""
|
||||
import http.server
|
||||
import os
|
||||
import sys
|
||||
import threading
|
||||
import unittest
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||||
from probe_availability import AvailabilityProbe # noqa: E402
|
||||
|
||||
|
||||
class _HealthHandler(http.server.BaseHTTPRequestHandler):
|
||||
"""可配置成功/失败模式的 /health 处理器。"""
|
||||
|
||||
fail_mode = [] # [True/False...] 按请求序号循环
|
||||
|
||||
def do_GET(self):
|
||||
if self.path == "/health":
|
||||
if self.fail_mode and self.fail_mode[0]:
|
||||
self.fail_mode[0] = False # 一次性失败
|
||||
self.send_response(503)
|
||||
else:
|
||||
self.send_response(200)
|
||||
else:
|
||||
self.send_response(404)
|
||||
self.end_headers()
|
||||
|
||||
def log_message(self, *args): # 静默日志
|
||||
pass
|
||||
|
||||
|
||||
def start_server(fail_once=False):
|
||||
handler = _HealthHandler
|
||||
handler.fail_mode = [True] if fail_once else []
|
||||
srv = http.server.ThreadingHTTPServer(("127.0.0.1", 0), handler)
|
||||
thread = threading.Thread(target=srv.serve_forever, daemon=True)
|
||||
thread.start()
|
||||
return srv
|
||||
|
||||
|
||||
class TestProbe(unittest.TestCase):
|
||||
"""探针统计与阈值判断。"""
|
||||
|
||||
def test_stable_endpoint_100pct(self):
|
||||
srv = start_server()
|
||||
try:
|
||||
url = f"http://127.0.0.1:{srv.server_port}"
|
||||
probe = AvailabilityProbe([url], timeout_seconds=2)
|
||||
report = probe.run(rounds=10, interval=0)
|
||||
finally:
|
||||
srv.shutdown()
|
||||
self.assertEqual(report.total, 10)
|
||||
self.assertEqual(report.success, 10)
|
||||
self.assertEqual(report.availability, 1.0)
|
||||
self.assertTrue(report.meets_target())
|
||||
|
||||
def test_single_failure_below_threshold(self):
|
||||
srv = start_server(fail_once=True)
|
||||
try:
|
||||
url = f"http://127.0.0.1:{srv.server_port}"
|
||||
probe = AvailabilityProbe([url], timeout_seconds=2)
|
||||
report = probe.run(rounds=500, interval=0) # 1/500 = 0.2%
|
||||
finally:
|
||||
srv.shutdown()
|
||||
self.assertEqual(len(report.failures), 1)
|
||||
# 1/500 = 0.002 = 99.8% —— 恰好达标（>= target 判定通过）
|
||||
self.assertGreaterEqual(report.availability, report.target)
|
||||
self.assertTrue(report.meets_target())
|
||||
# urllib 对非 200 抛 HTTPError → 记为不可用
|
||||
self.assertFalse(report.failures[0].ok)
|
||||
self.assertIn("503", report.failures[0].detail)
|
||||
|
||||
def test_high_failure_rate_fails_threshold(self):
|
||||
# 构造 50% 失败的端点（fail_mode 每次循环交替：用两个处理器状态不可行，
|
||||
# 改为直接 mock probe_once）
|
||||
probe = AvailabilityProbe(["http://x"], timeout_seconds=1)
|
||||
results = [probe.probe_once.__self__.__class__] # noqa - 占位
|
||||
from unittest import mock
|
||||
ok_result = mock.MagicMock()
|
||||
ok_result.ok = True
|
||||
bad_result = mock.MagicMock()
|
||||
bad_result.ok = False
|
||||
bad_result.url = "http://x"
|
||||
bad_result.detail = "http-503"
|
||||
bad_result.latency_ms = 1.0
|
||||
# 交替 200/503 → 6 次探测 3 成功 → 50% 可用率，远低于 99.8%
|
||||
with mock.patch.object(probe, "probe_once",
|
||||
side_effect=[ok_result, bad_result] * 3):
|
||||
report = probe.run(rounds=6, interval=0)
|
||||
self.assertEqual(report.availability, 0.5)
|
||||
self.assertFalse(report.meets_target())
|
||||
self.assertEqual(len(report.failures), 3)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user