diff --git a/scripts/bge-failover-proxy.py b/scripts/bge-failover-proxy.py index 24835304..c7b5e79c 100644 --- a/scripts/bge-failover-proxy.py +++ b/scripts/bge-failover-proxy.py @@ -16,9 +16,10 @@ import json import logging import os import socket +import subprocess import threading import time -from http.server import HTTPServer, BaseHTTPRequestHandler +from http.server import HTTPServer, BaseHTTPRequestHandler, ThreadingHTTPServer from urllib import request as urlrequest from urllib.error import URLError @@ -38,7 +39,14 @@ logging.basicConfig( log = logging.getLogger("bge-proxy") _state_lock = threading.Lock() -_state = {"remote_ok": None, "last_check": 0.0, "using": None} +_state = {"remote_ok": None, "last_check": 0.0, "using": None, + "local_started_by_proxy": False, "remote_ok_since": None, + "remote_retry_after": 0.0} # embedding 实发失败后的冷却截止（秒，300s） + +# 按需拉起本地 fallback 的参数（2026-09-05: bge-embed 不再常驻，104 故障时才拉起） +LOCAL_START_TIMEOUT = float(os.environ.get("LOCAL_START_TIMEOUT", "30")) # 拉起后等待就绪上限 +LOCAL_STOP_GRACE = float(os.environ.get("LOCAL_STOP_GRACE", "60")) # 远端稳定多久后停本地 +BGE_EMBED_SERVICE = os.environ.get("BGE_EMBED_SERVICE", "bge-embed.service") def _probe_remote() -> bool: @@ -54,23 +62,97 @@ def _probe_remote() -> bool: return False +def _local_healthy() -> bool: + """探测本地 fallback (8001) /health。""" + try: + req = urlrequest.Request(LOCAL_EMBED_URL + HEALTH_PATH) + with urlrequest.urlopen(req, timeout=1.5) as r: + if r.status != 200: + return False + return "ok" in r.read(200).decode("utf-8", "ignore") + except Exception: + return False + + +def _ensure_local() -> bool: + """远端不可用时确保本地可用：未启动则 systemctl --user 拉起 bge-embed。 + + 返回 True = 本地可用；False = 本地拉起失败/超时（转发将 502）。 + """ + if _local_healthy(): + return True + with _state_lock: + if not _state["local_started_by_proxy"]: + log.info("远端不可用 → 拉起 %s (按需)", BGE_EMBED_SERVICE) + try: + subprocess.Popen( + ["systemctl", "--user", "start", BGE_EMBED_SERVICE], + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, + ) + _state["local_started_by_proxy"] = True + except Exception as e: + log.error("拉起 %s 失败: %s", BGE_EMBED_SERVICE, e) + # 轮询等待模型加载（int8 onnx ~10-20s） + deadline = time.time() + LOCAL_START_TIMEOUT + while time.time() < deadline: + if _local_healthy(): + return True + time.sleep(1) + log.error("本地 fallback %s 拉起超时 (%ss)", BGE_EMBED_SERVICE, LOCAL_START_TIMEOUT) + return False + + +def _maybe_stop_local(remote_ok: bool): + """远端已恢复且稳定 → 停掉 proxy 拉起的本地 fallback，回到省电模式。""" + now = time.time() + with _state_lock: + if remote_ok and _state["local_started_by_proxy"]: + if _state["remote_ok_since"] is None: + _state["remote_ok_since"] = now + elif now - _state["remote_ok_since"] >= LOCAL_STOP_GRACE: + log.info("远端已稳定 %ss → 停 %s (释放内存)", LOCAL_STOP_GRACE, BGE_EMBED_SERVICE) + try: + subprocess.Popen( + ["systemctl", "--user", "stop", BGE_EMBED_SERVICE], + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, + ) + _state["local_started_by_proxy"] = False + _state["remote_ok_since"] = None + except Exception as e: + log.error("停止 %s 失败: %s", BGE_EMBED_SERVICE, e) + elif not remote_ok: + _state["remote_ok_since"] = None + + def choose_target() -> tuple[str, bool]: - """返回 (目标URL, 是否远端)。带 3s 冷却 + 状态缓存。""" + """返回 (目标URL, 是否远端)。带 3s 冷却 + 状态缓存 + 按需拉起本地 fallback。""" global _state now = time.time() with _state_lock: - if _state["remote_ok"] is not None and now - _state["last_check"] < 3.0: + cache_fresh = _state["remote_ok"] is not None and now - _state["last_check"] < 3.0 + in_cooldown = _state["remote_ok"] is False and now < _state["remote_retry_after"] + if cache_fresh or in_cooldown: remote_ok = _state["remote_ok"] else: remote_ok = _probe_remote() _state["remote_ok"] = remote_ok _state["last_check"] = now - target = REMOTE_EMBED_URL if remote_ok else LOCAL_EMBED_URL - using = "remote(104-GPU)" if remote_ok else "local(106-CPU)" + _maybe_stop_local(remote_ok) + if remote_ok: + using = "remote(104-GPU)" if using != _state["using"]: log.info("切换 → %s", using) + with _state_lock: + _state["using"] = using + return REMOTE_EMBED_URL, True + # 远端不可用 → 确保本地可用 + local_ok = _ensure_local() + using = "local(106-CPU)" if local_ok else "local(106-CPU, 未就绪)" + if using != _state["using"]: + log.info("切换 → %s", using) + with _state_lock: _state["using"] = using - return target, remote_ok + return LOCAL_EMBED_URL, False def _forward(url: str, method: str, body: bytes, headers: dict) -> tuple[int, dict, bytes]: @@ -103,6 +185,21 @@ class Handler(BaseHTTPRequestHandler): status, resp_headers, resp_body = _forward( target + self.path, self.command, body, dict(self.headers) ) + # 远端 embedding 实发失败（500/502/503/504，health 探测不出的故障） + # → 标故障 + 30s 冷却 + 立即降级 local 重试一次 + if status >= 500 and target.startswith(REMOTE_EMBED_URL): + with _state_lock: + _state["remote_ok"] = False + # 300s 冷却: 104 可能出现 health OK 但 embedding 500（实发故障）， + # 短冷却会导致反复 remote→500→local 抖动并可能反复启停 bge-embed + _state["remote_retry_after"] = time.time() + 300 + _state["using"] = None # 强制下次 log 切换 + log.info("远端 embedding 实发失败(%s) → 降级 local 重试", status) + if _ensure_local(): + target = LOCAL_EMBED_URL + status, resp_headers, resp_body = _forward( + target + self.path, self.command, body, dict(self.headers) + ) self.send_response(status) # 只转发安全响应头；Content-Length 必须用实际长度 for k, v in resp_headers.items(): @@ -112,7 +209,11 @@ class Handler(BaseHTTPRequestHandler): self.send_header("X-BGE-Proxy", "remote-104" if target.startswith(REMOTE_EMBED_URL) else "local-106") self.end_headers() if self.command != "HEAD": - self.wfile.write(resp_body) + try: + self.wfile.write(resp_body) + except (BrokenPipeError, ConnectionResetError): + # 客户端提前断开（探测超时/取消）——不影响其他请求（多线程） + pass do_GET = _handle do_POST = _handle @@ -131,11 +232,13 @@ def _startup_probe(): if __name__ == "__main__": - log.info("bge 故障切换代理启动 0.0.0.0:%s", PROXY_PORT) + log.info("bge 故障切换代理启动 0.0.0.0:%s (Threading)", PROXY_PORT) log.info(" 远端: %s", REMOTE_EMBED_URL) log.info(" 本地 fallback: %s", LOCAL_EMBED_URL) _startup_probe() - server = HTTPServer(("0.0.0.0", PROXY_PORT), Handler) + # ThreadingHTTPServer: 单请求卡住不阻塞 /health 探测和其他请求 + server = ThreadingHTTPServer(("0.0.0.0", PROXY_PORT), Handler) + server.daemon_threads = True try: server.serve_forever() except KeyboardInterrupt: