254 lines
11 KiB
Python
254 lines
11 KiB
Python
#!/usr/bin/env python3
|
||
"""deploy_guard.py — 部署一致性守卫（每 15 分钟 cron，全天候）
|
||
|
||
原则：实际部署的脚本 = MoFin git 最新版代码。
|
||
|
||
五件事：
|
||
1. 代码漂移检测：git 工作区中【被跟踪的代码文件】出现未提交改动
|
||
→ 自动 git checkout 回滚 + 重链硬链接 + XMPP 告警
|
||
（只回滚未提交的工作区改动；git 提交永远安全）
|
||
2. 上游自动部署：session-work 分支领先 master 且可快进
|
||
→ 自动 merge + 重链 + 必要时重启 dashboard + 告警
|
||
3. 硬链接一致性：幂等执行 sync_profile_scripts.sh
|
||
4. cron 引用完整性：jobs.json 引用的脚本必须存在
|
||
5. 状态落盘：gateway/logs/deploy_guard_status.json + deploy_guard.log
|
||
（有动作/有问题才发 XMPP，沉默即正常）
|
||
"""
|
||
import json, os, subprocess, sys, glob
|
||
from datetime import datetime
|
||
|
||
# ── 消息通道统一路由(broadcast/xmpp by delivery) ──
|
||
try:
|
||
from messenger import install_stdio_hook as _msh
|
||
_msh()
|
||
except Exception:
|
||
pass
|
||
|
||
REPO = "/home/hmo/MoFin"
|
||
CANONICAL = f"{REPO}/deploy/profile-scripts"
|
||
PROFILE_DIRS = [
|
||
"/home/hmo/.hermes/profiles/position-analyst/scripts",
|
||
"/home/hmo/.hermes/profiles/default/scripts",
|
||
"/home/hmo/.hermes/profiles/mohe/scripts",
|
||
]
|
||
LOG = f"{REPO}/gateway/logs/deploy_guard.log"
|
||
STATUS_JSON = f"{REPO}/gateway/logs/deploy_guard_status.json"
|
||
SYNC_SCRIPT = f"{CANONICAL}/sync_profile_scripts.sh"
|
||
|
||
# 守卫范围：被 git 跟踪的【代码】路径（运行时产物 data/ logs/ static/*.json 不在其列）
|
||
CODE_PATHS = [
|
||
"deploy/profile-scripts", "deploy/bot", "prompt_manager", "scripts",
|
||
"server.py", "mofin_db.py", "mo_data.py", "strategy_lifecycle.py", "xmpp_logger.py",
|
||
"mo_config.py",
|
||
"static/index.html", "static/mofin_health.html",
|
||
]
|
||
|
||
actions = [] # 采取了的动作（需告警）
|
||
problems = [] # 发现的问题（需告警）
|
||
|
||
|
||
def log(msg):
|
||
line = f"[{datetime.now().isoformat(timespec='seconds')}] {msg}"
|
||
print(line, flush=True)
|
||
try:
|
||
os.makedirs(os.path.dirname(LOG), exist_ok=True)
|
||
with open(LOG, "a", encoding="utf-8") as f:
|
||
f.write(line + "\n")
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def xmpp(msg, level=None):
|
||
try:
|
||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||
from alert_helper import notify, INFO
|
||
notify("部署守卫", msg, level or INFO)
|
||
except Exception as e:
|
||
log(f"XMPP发送失败: {e}")
|
||
|
||
|
||
def git(*args, check=True):
|
||
env = dict(os.environ)
|
||
env["GIT_ALLOW_COMMIT"] = "1" # pre-commit 白名单令牌（知微无此令牌无法提交）
|
||
r = subprocess.run(["git", "-C", REPO] + list(args),
|
||
capture_output=True, text=True, timeout=60, env=env)
|
||
if check and r.returncode != 0:
|
||
raise RuntimeError(f"git {' '.join(args)}: {r.stderr.strip()[:200]}")
|
||
return r.stdout.strip()
|
||
|
||
|
||
# ── 1. 代码漂移检测与回滚 ──
|
||
def check_code_drift():
|
||
# 2026-08-25 假漂移过滤：git status 受 stat 缓存抖动误报 M（touch/mtime 变化即报），
|
||
# 导致 heavy_run.sh/regime_perf_daily.sh 反复被"假回滚"+告警噪音。
|
||
# 先 refresh 索引 stat 缓存，再对报 M 的文件做内容级 diff 二次确认，无实质差异不回滚。
|
||
git("update-index", "--refresh", check=False)
|
||
out = git("status", "--porcelain", "--", *CODE_PATHS)
|
||
if not out:
|
||
return
|
||
modified, untracked = [], []
|
||
for line in out.splitlines():
|
||
if not line.strip():
|
||
continue
|
||
status, path = line[:2], line[2:].strip()
|
||
# 重命名格式 "R old -> new" 取新路径
|
||
if " -> " in path:
|
||
path = path.split(" -> ", 1)[1]
|
||
if status == "??":
|
||
untracked.append(path)
|
||
else:
|
||
modified.append(path)
|
||
if modified:
|
||
# 内容级二次确认（2026-08-25 v2：用 diff --quiet 返回码,权威判定不受stat缓存影响;
|
||
# 上一版用 --name-only 输出判定,仍被 heavy_run 假回滚——原因待查,quiet 返回码兜底）
|
||
def _has_real_diff(f):
|
||
_r = subprocess.run(["git", "-C", REPO, "diff", "--quiet", "HEAD", "--", f],
|
||
capture_output=True, timeout=60,
|
||
env=dict(os.environ, GIT_ALLOW_COMMIT="1"))
|
||
return _r.returncode != 0
|
||
real_modified = [f for f in modified if _has_real_diff(f)]
|
||
if not real_modified:
|
||
log(f"DRIFT假报(stat缓存抖动,内容无实质差异,不回滚): {modified}")
|
||
modified = []
|
||
elif len(real_modified) < len(modified):
|
||
log(f"DRIFT部分假报: 真实修改={real_modified}, 假抖动={ [f for f in modified if f not in real_modified] }")
|
||
modified = real_modified
|
||
if modified:
|
||
diff_stat = git("diff", "--stat", "--", *modified, check=False)
|
||
log(f"DRIFT: 检测到 {len(modified)} 个代码文件未入库改动: {modified}")
|
||
for f in modified:
|
||
git("checkout", "--", f)
|
||
subprocess.run(["bash", SYNC_SCRIPT], capture_output=True, timeout=120)
|
||
actions.append(f"回滚未入库代码改动 {len(modified)} 个: {', '.join(modified[:5])}"
|
||
+ (f" 等(共{len(modified)}个)" if len(modified) > 5 else ""))
|
||
log(f"DRIFT_REVERTED: {modified}\n{diff_stat[:500]}")
|
||
if untracked:
|
||
# 未跟踪的新文件：不删（可能是在途工作），只告警
|
||
problems.append(f"权威目录出现未跟踪文件(需人工确认): {', '.join(untracked[:5])}")
|
||
log(f"UNTRACKED: {untracked}")
|
||
|
||
|
||
# ── 2. session-work → master 自动部署 ──
|
||
def check_upstream_deploy():
|
||
try:
|
||
git("rev-parse", "--verify", "session-work")
|
||
except Exception:
|
||
return # 无此分支
|
||
ahead = git("rev-list", "--count", "master..session-work", check=False)
|
||
if not ahead.isdigit() or int(ahead) == 0:
|
||
return
|
||
# 仅当可快进（master 是 session-work 祖先）才自动合并
|
||
r = subprocess.run(["git", "-C", REPO, "merge-base", "--is-ancestor", "master", "session-work"])
|
||
if r.returncode != 0:
|
||
problems.append(f"session-work 领先 master {ahead} 个提交但存在分叉，需人工合并")
|
||
log(f"UPSTREAM_DIVERGED: ahead={ahead}")
|
||
return
|
||
log(f"UPSTREAM: session-work 领先 {ahead} 个提交，自动合并部署")
|
||
merge_out = git("merge", "session-work", "--no-edit")
|
||
sync_out = subprocess.run(["bash", SYNC_SCRIPT], capture_output=True, text=True, timeout=120)
|
||
# 触及 dashboard 代码则重启
|
||
touched = git("diff", "--name-only", f"HEAD~{ahead}", "HEAD", check=False)
|
||
restarted = ""
|
||
if any(t.startswith(("server.py", "static/")) for t in touched.splitlines()):
|
||
rr = subprocess.run(["sudo", "-n", "systemctl", "restart", "mofin-dashboard"],
|
||
capture_output=True, text=True, timeout=30)
|
||
restarted = " + dashboard已重启" if rr.returncode == 0 else " (dashboard重启失败,需人工)"
|
||
actions.append(f"自动部署 session-work→master ({ahead}个提交){restarted}")
|
||
log(f"UPSTREAM_DEPLOYED: {merge_out[:200]} | sync: {sync_out.stdout.strip()[-100:]}")
|
||
# ── 部署后自动验证（盲区②：提交级回归检测——知微 stale 提交事件的系统补丁）──
|
||
try:
|
||
vr = subprocess.run(["python3", f"{REPO}/scripts/verify_deployment.py"],
|
||
capture_output=True, text=True, timeout=300)
|
||
vout = (vr.stdout or "") + (vr.stderr or "")
|
||
import re as _re
|
||
m = _re.search(r"汇总:\s*(\d+)\s*项通过,\s*(\d+)\s*项失败", vout)
|
||
if m and int(m.group(2)) > 0:
|
||
fm = _re.search(r"失败项:\s*(\[.*?\])", vout)
|
||
problems.append(f"部署后验证发现 {m.group(2)} 项失败: {fm.group(1) if fm else '详见日志'}"
|
||
f"（刚合并的 {ahead} 个提交可能引入回归，需人工核查）")
|
||
log(f"POST_DEPLOY_VERIFY_FAIL: {m.group(2)} failures")
|
||
elif m:
|
||
log(f"POST_DEPLOY_VERIFY_OK: {m.group(1)} passed")
|
||
else:
|
||
log("POST_DEPLOY_VERIFY: 无法解析验证输出")
|
||
except Exception as e:
|
||
log(f"POST_DEPLOY_VERIFY_ERROR: {str(e)[:120]}")
|
||
|
||
|
||
# ── 3. 硬链接一致性 ──
|
||
def check_hardlinks():
|
||
r = subprocess.run(["bash", SYNC_SCRIPT], capture_output=True, text=True, timeout=120)
|
||
tail = (r.stdout or "").strip().splitlines()
|
||
if tail:
|
||
log(f"SYNC: {tail[-1]}")
|
||
if r.returncode != 0:
|
||
problems.append(f"sync_profile_scripts.sh 执行失败: {(r.stderr or '')[:150]}")
|
||
|
||
|
||
# ── 4. cron 引用完整性 ──
|
||
def check_cron_refs():
|
||
missing = []
|
||
for pj in glob.glob("/home/hmo/.hermes/profiles/*/cron/jobs.json"):
|
||
profile = pj.split("/")[-3]
|
||
try:
|
||
with open(pj, encoding="utf-8") as f:
|
||
jobs = json.load(f)
|
||
jobs = jobs if isinstance(jobs, list) else jobs.get("jobs", [])
|
||
except Exception:
|
||
continue
|
||
for j in jobs:
|
||
if not j.get("enabled", True):
|
||
continue
|
||
script = j.get("script") or ""
|
||
if not script:
|
||
continue
|
||
base = os.path.basename(script)
|
||
if not any(os.path.exists(os.path.join(d, base)) for d in PROFILE_DIRS + [CANONICAL]):
|
||
missing.append(f"[{profile}]{j.get('name','?')}→{base}")
|
||
if missing:
|
||
problems.append(f"cron引用缺失脚本 {len(missing)} 个: {', '.join(missing[:5])}")
|
||
log(f"CRON_MISSING: {missing}")
|
||
|
||
|
||
def main():
|
||
log("── deploy_guard 开始 ──")
|
||
for fn in (check_code_drift, check_upstream_deploy, check_hardlinks, check_cron_refs):
|
||
try:
|
||
fn()
|
||
except Exception as e:
|
||
problems.append(f"{fn.__name__} 异常: {str(e)[:150]}")
|
||
log(f"ERROR in {fn.__name__}: {e}")
|
||
|
||
status = {
|
||
"ts": datetime.now().isoformat(timespec="seconds"),
|
||
"ok": not problems,
|
||
"actions": actions,
|
||
"problems": problems,
|
||
}
|
||
try:
|
||
with open(STATUS_JSON, "w", encoding="utf-8") as f:
|
||
json.dump(status, f, ensure_ascii=False, indent=1)
|
||
except Exception:
|
||
pass
|
||
|
||
if actions or problems:
|
||
# 分级：部署后验证失败=需人工核查(ACTION直通)；其余=INFO(限速聚合)
|
||
info_parts = list(actions)
|
||
action_parts = []
|
||
for p in problems:
|
||
if "部署后验证发现" in p:
|
||
action_parts.append(p)
|
||
else:
|
||
info_parts.append(p)
|
||
if info_parts:
|
||
xmpp("\n".join(f"✅ {a}" if a in actions else f"⚠️ {a}" for a in info_parts))
|
||
for p in action_parts:
|
||
from alert_helper import notify as _notify, ACTION as _ACT
|
||
_notify("部署验证", f"⚠️ {p}", _ACT)
|
||
log(f"── 结束: actions={len(actions)} problems={len(problems)} ──")
|
||
return 0 if not problems else 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|