227 lines
9.4 KiB
Python
227 lines
9.4 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
|
|
|
|
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",
|
|
]
|
|
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():
|
|
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:
|
|
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())
|