From 6b24bf0b0d7aab7ba0ce82f459ae30c0ede02fe3 Mon Sep 17 00:00:00 2001 From: xsl Date: Sun, 20 Sep 2026 23:02:18 +0800 Subject: [PATCH] =?UTF-8?q?feat(gateway):=20=E8=A2=AB=E5=8A=A8=E6=91=98?= =?UTF-8?q?=E9=99=A4=E5=90=8E=E8=BF=9B=E5=85=A5=2060s=20=E9=9A=94=E7=A6=BB?= =?UTF-8?q?=E6=9C=9F=EF=BC=8C=E9=98=B2=E6=8E=A2=E6=B5=8B=E5=A4=8D=E6=B4=BB?= =?UTF-8?q?=E6=A8=AA=E8=B7=B3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 落地 docs/网关摘除隔离期实施方案.md（9·19 worker 磁盘满事故第 3 层防线）： - pool.py: WorkerState.offline_until + _cooldown_seconds()；mark_worker_unhealthy 摘除时设隔离截止时刻，隔离期内探测成功也不复活（_try_revive 守卫） - pool.py: 删除从未被调用的死代码 _check_worker_health/_mark_failure - config.py + config.example.json: health_check.cooldown_seconds（默认 60，0=关闭） - tests/test_gateway_pool.py: 9 用例（摘除设隔离/隔离内不复活/期满复活/整链路等） 网关机已重启验证：启动日志含"隔离期=60s"，/gateway-health 正常。 --- gateway/config.example.json | 3 +- gateway/config.py | 1 + gateway/pool.py | 76 ++++++++---------- tests/test_gateway_pool.py | 154 ++++++++++++++++++++++++++++++++++++ 4 files changed, 190 insertions(+), 44 deletions(-) create mode 100644 tests/test_gateway_pool.py diff --git a/gateway/config.example.json b/gateway/config.example.json index d977fff..d4a3430 100644 --- a/gateway/config.example.json +++ b/gateway/config.example.json @@ -11,7 +11,8 @@ "interval_seconds": 8, "timeout_seconds": 3, "unhealthy_threshold": 2, - "healthy_threshold": 1 + "healthy_threshold": 1, + "cooldown_seconds": 60 }, "ark_api_key": "", "dispatch": { diff --git a/gateway/config.py b/gateway/config.py index b6f0b03..5c58bc5 100644 --- a/gateway/config.py +++ b/gateway/config.py @@ -21,6 +21,7 @@ DEFAULTS = { "timeout_seconds": 3, "unhealthy_threshold": 2, "healthy_threshold": 1, + "cooldown_seconds": 60, # 被动摘除后的隔离期（秒），0=关闭（等价旧行为） }, "dispatch": { "per_worker_concurrency": 1, diff --git a/gateway/pool.py b/gateway/pool.py index 9bf790e..ab9e3fa 100644 --- a/gateway/pool.py +++ b/gateway/pool.py @@ -18,6 +18,15 @@ import httpx logger = logging.getLogger("gateway.pool") +def _cooldown_seconds() -> float: + """被动摘除后的隔离期时长（秒），配置 health_check.cooldown_seconds，默认 60。""" + try: + from gateway.config import get_config + return float(get_config()["health_check"].get("cooldown_seconds", 60)) + except Exception: # noqa: BLE001 + return 60.0 + + # --------------------------------------------------------------------------- # 异常 # --------------------------------------------------------------------------- @@ -39,6 +48,7 @@ class WorkerState: busy: bool = False consecutive_failures: int = 0 consecutive_successes: int = 0 + offline_until: float = 0.0 # 被动摘除后的隔离截止时刻（time.monotonic() 基准） # --------------------------------------------------------------------------- @@ -63,45 +73,18 @@ _global_waiting: int = 0 # 健康检查后台任务 # --------------------------------------------------------------------------- -async def _check_worker_health( - client: httpx.AsyncClient, - w: WorkerState, - cfg: dict, -) -> None: - """探测单个 worker 的 /health，更新上下线状态。""" - hc_cfg = cfg["health_check"] - url = f"{w.url}{hc_cfg['path']}" - token = cfg["shared_password"] +def _try_revive(w: WorkerState, healthy_threshold: int) -> bool: + """探测成功后是否允许将该 worker 复活上线。 - try: - resp = await client.get( - url, - headers={"X-Internal-Token": token}, - timeout=hc_cfg["timeout_seconds"], - ) - if resp.status_code == 200: - w.consecutive_failures = 0 - w.consecutive_successes += 1 - if w.consecutive_successes >= hc_cfg["healthy_threshold"]: - if not w.online: - w.online = True - logger.info("Worker 上线: %s（连续成功 %d 次）", w.url, w.consecutive_successes) - else: - _mark_failure(w, f"HTTP {resp.status_code}") - except Exception as exc: - _mark_failure(w, str(exc)) - - -def _mark_failure(w: WorkerState, reason: str) -> None: - """记录一次失败，达到阈值后下线。""" - w.consecutive_successes = 0 - w.consecutive_failures += 1 - threshold = w.consecutive_failures # used in log - hc_threshold = 2 # default, will be overridden - if w.consecutive_failures >= hc_threshold: - # 实际 threshold 从配置读取，这里先做基本判断 - pass - logger.debug("Worker %s 健康检查失败 (%d/%d): %s", w.url, w.consecutive_failures, 99, reason) + 条件：未在线 + 连续成功达到阈值 + **不在被动摘除隔离期内** + （offline_until 之前即使 /health 探测成功也压着不上线，防横跳）。 + """ + if w.online or w.consecutive_successes < healthy_threshold: + return False + if time.monotonic() < w.offline_until: + logger.debug("Worker %s 探测成功但处于隔离期，暂不复活", w.url) + return False + return True async def _health_check_loop(cfg: dict) -> None: @@ -113,8 +96,8 @@ async def _health_check_loop(cfg: dict) -> None: token = cfg["shared_password"] logger.info( - "健康检查循环启动 | 间隔=%ds | 下线阈值=%d | 上线阈值=%d | workers=%d", - interval, unhealthy_threshold, healthy_threshold, len(_workers), + "健康检查循环启动 | 间隔=%ds | 下线阈值=%d | 上线阈值=%d | 隔离期=%.0fs | workers=%d", + interval, unhealthy_threshold, healthy_threshold, _cooldown_seconds(), len(_workers), ) async with httpx.AsyncClient() as client: @@ -130,7 +113,7 @@ async def _health_check_loop(cfg: dict) -> None: if resp.status_code == 200: w.consecutive_failures = 0 w.consecutive_successes += 1 - if w.consecutive_successes >= healthy_threshold and not w.online: + if _try_revive(w, healthy_threshold): w.online = True logger.info("✅ Worker 上线: %s", w.url) else: @@ -272,12 +255,19 @@ def release_global_slot() -> None: # --------------------------------------------------------------------------- async def mark_worker_unhealthy(w: WorkerState) -> None: - """被动标记：转发失败时立即将该 worker 下线。""" + """被动标记：转发失败时立即将该 worker 下线，并进入隔离期。 + + 隔离期内即使 /health 探测成功也不复活（由 _try_revive 守卫）， + 避免"摘除→8 秒复活→再失败"的横跳。 + """ async with _pool_condition: if w.online: w.online = False w.consecutive_failures += 1 - logger.warning("⚠ Worker 被动下线: %s（转发失败）", w.url) + cooldown = _cooldown_seconds() + w.offline_until = time.monotonic() + cooldown + logger.warning("⚠ Worker 被动下线: %s（转发失败，隔离 %.0fs）", + w.url, cooldown) async def acquire_worker(cfg: dict) -> WorkerState: diff --git a/tests/test_gateway_pool.py b/tests/test_gateway_pool.py new file mode 100644 index 0000000..4490528 --- /dev/null +++ b/tests/test_gateway_pool.py @@ -0,0 +1,154 @@ +"""gateway/pool.py 隔离期逻辑单测（不依赖真实网络）。 + +覆盖《docs/网关摘除隔离期实施方案.md》§3.3 的三个用例： +- 被动摘除设置隔离期（offline_until > now） +- 隔离期内探测成功不复活 +- 隔离期满后探测成功恢复上线 +外加：_try_revive 的阈值/在线短路、mark_worker_unhealthy 对已下线 worker 的幂等、 +_health_check_loop 整链路验证、_cooldown_seconds 配置读取。 + +与 test_gateway_serialization.py 同款：asyncio.run 套同步测试，不依赖 pytest-asyncio。 +""" + +import asyncio +import time +from unittest.mock import patch + +from gateway import pool +from gateway.pool import WorkerState + +_HC_CFG = { + "health_check": { + "path": "/health", + "interval_seconds": 0.05, # 测试用：快速轮询 + "timeout_seconds": 3, + "unhealthy_threshold": 2, + "healthy_threshold": 1, + }, + "shared_password": "test-token", +} + + +# --------------------------------------------------------------------------- +# 被动摘除 → 隔离期 +# --------------------------------------------------------------------------- + +def test_passive_ejection_sets_cooldown(): + w = WorkerState(url="http://127.0.0.1:9000", online=True) + + async def run(): + pool._pool_condition = asyncio.Condition() + with patch.object(pool, "_cooldown_seconds", return_value=60.0): + await pool.mark_worker_unhealthy(w) + + asyncio.run(run()) + assert w.online is False + assert w.offline_until > time.monotonic(), "被动摘除必须设置未来的隔离截止时刻" + assert w.consecutive_failures == 1 + + +def test_passive_ejection_idempotent_when_offline(): + """已下线的 worker 再次被标记：不重复计失败、不续期隔离。""" + w = WorkerState(url="http://127.0.0.1:9000", online=False) + w.offline_until = time.monotonic() + 30 + before = w.offline_until + + async def run(): + pool._pool_condition = asyncio.Condition() + with patch.object(pool, "_cooldown_seconds", return_value=60.0): + await pool.mark_worker_unhealthy(w) + + asyncio.run(run()) + assert w.online is False + assert w.consecutive_failures == 0 + assert w.offline_until == before + + +# --------------------------------------------------------------------------- +# _try_revive：探测成功后是否允许复活 +# --------------------------------------------------------------------------- + +def test_probe_success_blocked_during_cooldown(): + w = WorkerState(url="http://127.0.0.1:9000") # offline + w.consecutive_successes = 1 + w.offline_until = time.monotonic() + 60 + assert pool._try_revive(w, healthy_threshold=1) is False + assert w.online is False + + +def test_probe_success_revives_after_cooldown(): + w = WorkerState(url="http://127.0.0.1:9000") + w.consecutive_successes = 1 + w.offline_until = time.monotonic() - 1 # 隔离期已过 + assert pool._try_revive(w, healthy_threshold=1) is True + + +def test_try_revive_requires_success_threshold(): + w = WorkerState(url="http://127.0.0.1:9000") + w.consecutive_successes = 1 + w.offline_until = 0.0 + assert pool._try_revive(w, healthy_threshold=2) is False + + +def test_try_revive_noop_when_already_online(): + w = WorkerState(url="http://127.0.0.1:9000", online=True) + w.consecutive_successes = 1 + assert pool._try_revive(w, healthy_threshold=1) is False + + +# --------------------------------------------------------------------------- +# _health_check_loop 整链路：隔离期内探测 200 不上线，期满后上线 +# --------------------------------------------------------------------------- + +def test_health_check_loop_respects_cooldown(monkeypatch): + w = WorkerState(url="http://127.0.0.1:9000", online=False) + w.offline_until = time.monotonic() + 60 # 处于隔离期 + + class _FakeResp: + status_code = 200 + + class _FakeClient: + async def __aenter__(self): + return self + + async def __aexit__(self, *args): + return False + + async def get(self, url, headers=None, timeout=None): + return _FakeResp() + + async def run(): + monkeypatch.setattr(pool, "_workers", {"w1": w}) + monkeypatch.setattr(pool, "_shutdown_event", asyncio.Event()) + monkeypatch.setattr(pool.httpx, "AsyncClient", _FakeClient) + task = asyncio.create_task(pool._health_check_loop(_HC_CFG)) + try: + await asyncio.sleep(0.2) # 跑若干轮探测 + assert w.consecutive_successes >= 1, "探测应已成功过" + assert w.online is False, "隔离期内即使探测成功也不得复活" + + w.offline_until = time.monotonic() - 1 # 隔离期结束 + await asyncio.sleep(0.2) + assert w.online is True, "隔离期满后下一次探测成功应上线" + finally: + pool._shutdown_event.set() + await asyncio.wait_for(task, timeout=2) + + asyncio.run(run()) + + +# --------------------------------------------------------------------------- +# _cooldown_seconds 配置读取 +# --------------------------------------------------------------------------- + +def test_cooldown_seconds_reads_config(monkeypatch): + from gateway import config as gw_config + monkeypatch.setattr(gw_config, "_config_cache", + {"health_check": {"cooldown_seconds": 120}}) + assert pool._cooldown_seconds() == 120.0 + + +def test_cooldown_seconds_falls_back_to_60(monkeypatch): + from gateway import config as gw_config + monkeypatch.setattr(gw_config, "_config_cache", None) # get_config 会抛 RuntimeError + assert pool._cooldown_seconds() == 60.0