diff --git a/deploy/k8s/healthz/README.md b/deploy/k8s/healthz/README.md new file mode 100644 index 0000000..7488c17 --- /dev/null +++ b/deploy/k8s/healthz/README.md @@ -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、失败明细记录。 diff --git a/deploy/k8s/healthz/probe_availability.py b/deploy/k8s/healthz/probe_availability.py new file mode 100644 index 0000000..f5bd358 --- /dev/null +++ b/deploy/k8s/healthz/probe_availability.py @@ -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()) diff --git a/deploy/k8s/healthz/test_probe.py b/deploy/k8s/healthz/test_probe.py new file mode 100644 index 0000000..1dfb935 --- /dev/null +++ b/deploy/k8s/healthz/test_probe.py @@ -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()