# -*- coding: utf-8 -*- """主程序升级命令。 流程（见 05-部署运维.md SOP）： 1. 从 B 端下载新 jar 到 tmp（SHA-256 校验，防传输损坏） 2. 备份当前 bin/hospital-front.jar → backup/hospital-front-{时间戳}.jar（保留最近 5 个） 3. 优雅停止主程序（调 /api/admin/restart；超时强杀） 4. 替换 jar 5. 调启动命令拉起主程序 6. 轮询健康检查（最长 90 秒） 7. 成功 → 上报；失败 → 用备份 jar 自动回滚并重启 → 上报 期间整个流程串行（upgrade_lock 防并发升级）。 """ import glob import hashlib import logging import os import shutil import subprocess import threading import time import requests from agent.process_guard import start_main_app, stop_main_service, start_main_service, _upgrading log = logging.getLogger(__name__) # 备份保留数量 KEEP_BACKUPS = 5 class UpgradeCommand: """升级 SOP 执行器。""" def __init__(self, config, main_app_client, client): self.config = config self.main_app_client = main_app_client self.client = client # 升级互斥锁（同一时刻只允许一个升级在跑） self._upgrade_lock = threading.Lock() def upgrade(self, version: str, download_url: str, sha256: str | None) -> dict: """ 执行升级。 :param version: 新版本号（仅用于日志与上报） :param download_url: B 端提供的下载地址：相对路径（以 / 开头，经网关下载）或完整 URL :param sha256: 可选校验值 """ with self._upgrade_lock: # 置「升级中」标志：升级主动停主程序期间，看门狗跳过拉起， # 避免新进程抢占 jar 文件锁导致替换失败 _upgrading.set() try: return self._do_upgrade(version, download_url, sha256) finally: _upgrading.clear() def _do_upgrade(self, version: str, download_url: str, sha256: str | None) -> dict: """升级实际执行（由 upgrade 加锁 + 置标志后调用）。""" # 相对路径（B 端下发的文件路径）：拼 网关基础地址 + 下载路由前缀 + 文件路径 # 例：http://网关:19901 + /file/down + /hf-upgrade/20260902/xxx.jar if download_url.startswith("/"): if not self.config.gateway_url: return self._fail(version, "B 端下发了相对下载路径，但 Agent 未配置 gateway.url") prefix = "/" + self.config.download_prefix.strip("/") download_url = self.config.gateway_url.rstrip("/") + prefix + "/" + download_url.strip("/") log.info("开始升级到版本 %s：%s", version, download_url) self.client.send("UPGRADE_PROGRESS", {"version": version, "stage": "download"}) # ===== 1. 下载 ===== tmp_jar = os.path.join("tmp", f"upgrade-{version}.jar") os.makedirs(os.path.dirname(tmp_jar), exist_ok=True) try: with requests.get(download_url, stream=True, timeout=(10, 120)) as resp: if resp.status_code != 200: return self._fail(version, f"下载失败：HTTP {resp.status_code}") content_type = resp.headers.get("Content-Type", "") with open(tmp_jar, "wb") as f: for chunk in resp.iter_content(8192): f.write(chunk) except Exception as e: return self._fail(version, f"下载异常：{e}") size = os.path.getsize(tmp_jar) # 诊断：打印响应类型与文件头（下载到错误页面/JSON 而非 jar 时一眼可见） with open(tmp_jar, "rb") as f: head = f.read(64) head_preview = head[:32].decode("utf-8", errors="replace") log.info("下载完成：%s（%d 字节，Content-Type=%s，文件头=%r）", os.path.basename(tmp_jar), size, content_type, head_preview) # jar 为 zip 格式（PK 魔数）。文件服务器下载失败常返回 200+错误 JSON/HTML， # 此处提前拦截并给出明确原因，避免误报 SHA-256 校验失败 if not head.startswith(b"PK"): # 疑似鉴权/路径错误，读出响应体前 200 字符帮助定位 with open(tmp_jar, "rb") as f: body_preview = f.read(200).decode("utf-8", errors="replace") os.remove(tmp_jar) return self._fail( version, f"下载内容不是 jar（PK 魔数缺失）：大小={size}，Content-Type={content_type}，" f"响应开头={body_preview!r}（疑似下载地址需鉴权或路径错误）", ) # ===== 2. 校验 ===== if sha256: try: actual = sha256_file(tmp_jar) except Exception as e: return self._fail(version, f"SHA-256 计算异常：{e}") if actual.lower() != sha256.lower(): os.remove(tmp_jar) return self._fail(version, f"SHA-256 校验失败：expected={sha256}, actual={actual}") log.info("SHA-256 校验通过") # ===== 3. 备份 ===== # jar 路径来自配置（main.jar.path），适配不同部署结构 #（如 Agent 与主程序分目录、jar 文件名带版本号等） jar_file = self.config.main_jar backup_file = os.path.join("backup", f"hospital-front-{int(time.time() * 1000)}.jar") os.makedirs(os.path.dirname(backup_file), exist_ok=True) if os.path.exists(jar_file): try: shutil.copyfile(jar_file, backup_file) except Exception as e: return self._fail(version, f"备份失败：{e}") log.info("已备份当前版本 → %s", os.path.basename(backup_file)) self._clean_old_backups() # ===== 4. 停止主程序 ===== self.client.send("UPGRADE_PROGRESS", {"version": version, "stage": "stop"}) self._stop_main_app() log.info("主程序已停止") # ===== 5. 替换 ===== # 注意：Agent 为独立 Python 进程，不占用 bin/hospital-front.jar， # 可安全替换；无需再维护 Agent jar 副本（Java 版遗留逻辑）。 try: copy_with_retry(tmp_jar, jar_file, 10) os.remove(tmp_jar) except Exception as e: self._rollback(backup_file) return self._fail(version, f"替换 jar 失败，已回滚：{e}") log.info("jar 已替换") # ===== 6. 拉起 + 健康检查 ===== self.client.send("UPGRADE_PROGRESS", {"version": version, "stage": "restart"}) self._start_main_app() self.client.send("UPGRADE_PROGRESS", {"version": version, "stage": "healthcheck"}) if self._wait_healthy(90): log.info("升级成功：%s", version) return {"ok": True, "version": version, "msg": "升级成功"} # ===== 7. 健康检查失败 → 回滚 ===== log.error("新版本健康检查失败，自动回滚到 %s", os.path.basename(backup_file)) self.client.send("UPGRADE_PROGRESS", {"version": version, "stage": "rollback"}) self._stop_main_app() self._rollback(backup_file) self._start_main_app() if self._wait_healthy(90): return self._fail(version, f"新版本启动失败，已自动回滚到 {os.path.basename(backup_file)} 并恢复服务") self.client.send("ALERT", { "level": "CRITICAL", "msg": f"升级失败且回滚后仍不健康，请人工介入！版本={version}", }) return self._fail(version, "升级失败且回滚后仍不健康，请人工介入") # ================= 内部步骤 ================= def _stop_main_app(self): """停止主程序：优先 nssm stop（等进程退出再返回，文件锁释放可靠）； 未配置服务名时回退 admin API + PID 强杀。""" if self.config.main_service_name: # nssm 管理的服务：nssm stop 等待服务进程退出后返回，文件锁释放可靠 stop_main_service(self.config.nssm_path, self.config.main_service_name) time.sleep(1) # 保险等锁释放 return # 回退：admin API 触发优雅退出 + PID 强杀 pid = read_pid_file(self.config.main_pid) if self.main_app_client.restart(100): # 等待健康探测失败（最长 30 秒） deadline = time.time() + 30 while time.time() < deadline: if self.main_app_client.health().get("ok") is not True: break time.sleep(1) # 兜底：按 PID 文件强杀 kill_by_pid_file(self.config.main_pid) # 等待进程真正退出（Windows 下进程退出后才释放 jar 文件锁） if pid is not None: wait_process_exit(pid, 15) time.sleep(1) def _start_main_app(self): """拉起主程序：优先 nssm start（服务模式），否则回退启动命令。""" if self.config.main_service_name: start_main_service(self.config.nssm_path, self.config.main_service_name) else: try: proc = start_main_app(self.config.main_start_cmd) log.info("启动命令已执行，pid=%s", proc.pid if proc else None) except Exception as e: log.error("主程序拉起失败：%s", e) def _wait_healthy(self, timeout_sec: int) -> bool: """轮询等待主程序健康。""" deadline = time.time() + timeout_sec while time.time() < deadline: if self.main_app_client.health().get("ok") is True: return True time.sleep(3) return False def _rollback(self, backup_file: str): """回滚：备份 jar 覆盖回主程序 jar（路径来自配置）。""" try: if os.path.exists(backup_file): copy_with_retry(backup_file, self.config.main_jar, 10) log.info("已回滚到 %s", os.path.basename(backup_file)) except Exception as e: log.error("回滚失败：%s", e) def _clean_old_backups(self): """清理旧备份（按文件名倒序保留最近 N 个）。""" backups = sorted(glob.glob(os.path.join("backup", "hospital-front-*")), reverse=True) for old in backups[KEEP_BACKUPS:]: try: os.remove(old) except Exception: pass def _fail(self, version: str, msg: str) -> dict: """统一失败上报。""" log.error("升级失败：%s", msg) return {"ok": False, "version": version, "msg": msg} # ================= 模块级工具函数 ================= def read_pid_file(pid_file: str) -> str | None: """读 PID 文件，无或为空返回 None。""" try: if not os.path.exists(pid_file): return None with open(pid_file, "r", encoding="utf-8") as f: s = f.read().strip() return s or None except Exception: return None def kill_by_pid_file(pid_file: str): """按 PID 文件强杀进程（Windows taskkill / Linux kill -9）。""" pid = read_pid_file(pid_file) if not pid: log.warning("强杀跳过：PID 文件 %s 无内容或不存在", pid_file) return log.info("按 PID 强杀：%s（来自 %s）", pid, pid_file) try: if os.name == "nt": r = subprocess.run(["taskkill", "/PID", pid, "/T", "/F"], capture_output=True, timeout=15) out = (r.stdout or b"") + (r.stderr or b"") log.info("taskkill 退出码=%s，输出=%s", r.returncode, out.decode("gbk", errors="replace").strip()) else: r = subprocess.run(["kill", "-9", pid], capture_output=True, timeout=15) log.info("kill -9 退出码=%s", r.returncode) except Exception as e: log.warning("PID 强杀失败：%s", e) def wait_process_exit(pid: str, timeout_sec: int): """轮询等待进程退出（tasklist 查不到即退出）。""" deadline = time.time() + timeout_sec while time.time() < deadline: alive = False try: if os.name == "nt": out = subprocess.run( ["tasklist", "/FI", f"PID eq {pid}"], capture_output=True, timeout=10, ).stdout.decode("gbk", errors="ignore") alive = pid in out else: alive = subprocess.run(["kill", "-0", pid], capture_output=True).returncode == 0 except Exception: alive = False if not alive: log.info("进程 %s 已退出", pid) return time.sleep(0.5) log.warning("等待进程 %s 退出超时（%ds），可能仍占用 jar 锁", pid, timeout_sec) def copy_with_retry(src: str, dst: str, retries: int): """带重试的文件覆盖（Windows 下用「改名腾位」策略绕开文件锁延迟）。 进程退出后 Windows 对 jar 的内存映射句柄释放有滞后，直接 os.remove 会持续 WinError 32。改为：先把旧 jar 改名为 .old（Windows 允许重命名正在使用的文件）， 腾出原文件名写入新 jar；.old 留待进程完全退出后清理。 """ last_error = None old_name = dst + ".old" for i in range(retries): try: # 清理上一次残留的 .old（若进程已退出可删；删不动也无妨，继续） if os.path.exists(old_name): try: os.remove(old_name) except Exception: pass # 把占用中的旧 jar 改名腾位（关键：即使文件被锁，重命名通常也能成功） if os.path.exists(dst): os.replace(dst, old_name) # 原文件名已腾空，写入新 jar shutil.copyfile(src, dst) return except Exception as e: last_error = e log.warning("替换重试 %d/%d 失败：%s（dst=%s）", i + 1, retries, e, dst) time.sleep(1) raise RuntimeError( f"文件替换失败（重试 {retries} 次）：{os.path.basename(dst)}，最后错误：{last_error}" ) from last_error def sha256_file(path: str) -> str: """文件 SHA-256（分块读取，支持大文件）。""" digest = hashlib.sha256() with open(path, "rb") as f: for chunk in iter(lambda: f.read(8192), b""): digest.update(chunk) return digest.hexdigest()