# -*- coding: utf-8 -*- """主程序进程守护（看门狗）。 周期探测主程序健康： - 连续 3 次失败 → 认为主程序宕机 → 强杀残留 → 调启动命令拉起 - 拉起后等待健康恢复；连续 3 次拉起失败 → 上报告警（ALERT） """ import logging import subprocess import threading import time log = logging.getLogger(__name__) # 模块级「升级中」标志：升级流程开始时置 True，结束后置 False。 # 看门狗探测到主程序不健康时若此标志为 True，说明是升级主动停的，跳过拉起， # 避免升级替换 jar 时被看门狗拉起的新进程抢占文件锁。 _upgrading = threading.Event() # 拉起后的宽限期（秒）：主程序重启后 JVM 启动 + DB 建连可能耗时数分钟 # （如机器休眠唤醒后网络未就绪，实测 4 分钟）。宽限期内探测失败只记录不累计， # 避免重启循环（重启→DB 慢→又判宕机→再重启）。 GRACE_PERIOD_SEC = 300 class ProcessGuard: """看门狗：周期健康探测 + 宕机拉起 + 告警上报。""" def __init__(self, config, main_app_client, client): self.config = config self.main_app_client = main_app_client self.client = client self._fail_count = 0 # 连续健康探测失败次数 self._restart_count = 0 # 连续拉起次数 self._grace_until = 0.0 # 拉起后的宽限期截止时刻（monotonic 秒） self._stop = threading.Event() self._thread: threading.Thread | None = None def start(self): """启动守护（首检延迟 30 秒，之后每 15 秒探测一次）。""" if not self.config.guard_enabled: log.info("进程守护未启用（guard.enable=false）") return self._thread = threading.Thread(target=self._loop, name="process-guard", daemon=True) self._thread.start() log.info("进程守护已启动：每 15 秒探测 %s", self.config.main_health_url) def stop(self): """停止守护。""" self._stop.set() # ================= 探测循环 ================= def _loop(self): """探测主循环：先等 30 秒（给主程序启动时间），再周期探测。""" if self._stop.wait(30): return while not self._stop.wait(15): try: self._check() except Exception as e: log.warning("守护探测异常：%s", e) def _check(self): """单次探测：健康则清零计数；连续 3 次失败则拉起（升级期间/宽限期内跳过）。""" health = self.main_app_client.health() if health.get("ok") is True: self._fail_count = 0 if self._grace_until and time.monotonic() < self._grace_until: log.info("宽限期内主程序恢复健康，结束宽限") self._grace_until = 0.0 return # 拉起后宽限期内：主程序可能在启动/DB 建连中，只记录不累计失败 if self._grace_until and time.monotonic() < self._grace_until: log.info("主程序拉起后宽限期内（剩 %d 秒），本次探测失败不累计：%s", int(self._grace_until - time.monotonic()), health.get("error")) return # 升级流程主动停的主程序，看门狗不接管，避免抢占 jar 锁 if _upgrading.is_set(): log.info("主程序不可达，但升级流程进行中，看门狗跳过拉起") self._fail_count = 0 # 不累计，升级结束后自然恢复 return self._fail_count += 1 log.warning("主程序健康探测失败（%d/3）：%s", self._fail_count, health.get("error")) if self._fail_count < 3: return # 连续 3 次失败 → 拉起 self._fail_count = 0 self._restart_count += 1 attempts = self._restart_count if attempts > 3: self.client.send("ALERT", { "level": "CRITICAL", "msg": f"主程序连续 {attempts} 次拉起失败，请人工介入", }) # 重置计数，之后继续尝试（降频可后续优化） self._restart_count = 3 return # 拉起方式：配置了服务名则走 nssm restart（幂等），否则回退启动命令 if self.config.main_service_name: start_desc = f"nssm restart {self.config.main_service_name}" else: start_desc = self.config.main_start_cmd log.error("主程序宕机，执行第 %d 次拉起：%s", attempts, start_desc) self.client.send("ALERT", { "level": "WARN", "msg": f"主程序宕机，Agent 正在执行第 {attempts} 次拉起", }) try: if self.config.main_service_name: ok, reason = restart_main_service(self.config.nssm_path, self.config.main_service_name) if not ok: log.error("nssm 拉起失败：%s", reason) else: proc = start_main_app(self.config.main_start_cmd) log.info("启动命令已执行，pid=%s", proc.pid if proc else None) # 拉起后进入宽限期：JVM 启动 + DB 建连期间不累计探测失败，防重启循环 self._grace_until = time.monotonic() + GRACE_PERIOD_SEC except Exception as e: log.error("拉起失败：%s", e) # ================= 公共工具（upgrade.py 复用） ================= def restart_main_service(nssm_path: str, service_name: str) -> tuple[bool, str]: """ 通过 NSSM 重启主程序服务（幂等：无论服务处于停止/运行态均到运行态）。 nssm_path 为 nssm.exe 完整路径；service_name 为注册的 Windows 服务名。 返回 (是否成功, 输出/原因)。nssm 输出打印到日志，权限不足（拒绝访问）等 失败原因一眼可见。 """ try: proc = subprocess.run( [nssm_path, "restart", service_name], capture_output=True, timeout=60, ) # nssm 输出为 UTF-16 编码（宽字符），需按 utf-16 解码 out = ((proc.stdout or b"") + (proc.stderr or b"")).decode("utf-16", errors="replace").strip() ok = proc.returncode == 0 if ok: log.info("nssm restart %s 成功：%s", service_name, out or "(无输出)") else: log.warning("nssm restart %s 失败（exit=%s）：%s", service_name, proc.returncode, out) if "拒绝访问" in out or "Access" in out: out += "（提示：重启服务需要管理员权限，请将 Agent 以 Windows 服务方式运行）" return ok, out except Exception as e: log.warning("nssm restart %s 执行异常：%s", service_name, e) return False, str(e) def stop_main_service(nssm_path: str, service_name: str) -> tuple[bool, str]: """ 通过 NSSM 停止主程序服务（nssm stop 会等待服务进程退出后返回）。 用于升级流程：停止成功即文件锁释放，替换 jar 不再被 WinError 32 挡住。 nssm_path 为 nssm.exe 完整路径；service_name 为注册的 Windows 服务名。 返回 (是否成功, 输出/原因)。 """ try: proc = subprocess.run( [nssm_path, "stop", service_name], capture_output=True, timeout=120, ) out = ((proc.stdout or b"") + (proc.stderr or b"")).decode("utf-16", errors="replace").strip() ok = proc.returncode == 0 if ok: log.info("nssm stop %s 成功：%s", service_name, out or "(无输出)") else: log.warning("nssm stop %s 失败（exit=%s）：%s", service_name, proc.returncode, out) return ok, out except Exception as e: log.warning("nssm stop %s 执行异常：%s", service_name, e) return False, str(e) def start_main_service(nssm_path: str, service_name: str) -> tuple[bool, str]: """ 通过 NSSM 启动主程序服务（nssm start 触发服务进入运行态，应用初始化由健康检查轮询确认）。 返回 (是否成功, 输出/原因)。 """ try: proc = subprocess.run( [nssm_path, "start", service_name], capture_output=True, timeout=60, ) out = ((proc.stdout or b"") + (proc.stderr or b"")).decode("utf-16", errors="replace").strip() ok = proc.returncode == 0 if ok: log.info("nssm start %s 成功：%s", service_name, out or "(无输出)") else: log.warning("nssm start %s 失败（exit=%s）：%s", service_name, proc.returncode, out) return ok, out except Exception as e: log.warning("nssm start %s 执行异常：%s", service_name, e) return False, str(e) def start_main_app(start_cmd: str) -> subprocess.Popen | None: """ 拉起主程序。 Windows 下 .bat 走 cmd /c（路径转反斜杠）；其余走 sh。 返回 Popen 对象（不等待进程结束）。 """ cmd = start_cmd.replace("/", "\\") # cmd/c 要求反斜杠路径 if cmd.lower().endswith(".bat"): args = ["cmd", "/c", cmd, "start"] else: args = ["sh", start_cmd, "start"] return subprocess.Popen( args, stdout=subprocess.DEVNULL, stderr=subprocess.STDOUT, # 不设 cwd，继承 Agent 工作目录（与主程序同根目录部署） )