feat(guard): 部署一致性守卫 deploy_guard.py + 知微运维纪律 + 健康JSON移出跟踪
- deploy_guard.py (15min cron): 代码漂移自动回滚(仅未提交改动)+ session-work可快进时自动merge部署+幂等重链+cron引用完整性, 状态落盘JSON, 有动作即XMPP报备 - docs/zhiwei-ops-discipline.md: 知微纪律——禁止直接编辑被跟踪代码/ 禁stale提交/禁直调gateway批量LLM; 自愈白名单(rerun/restart/ sync/switch_key)本不改代码, 与纪律不矛盾; 紧急热修走git流程, 提交到session-work后guard 15min自动部署 - dev-spec 红线#6 补充部署守卫机制 - static/mofin_health.json 移出git跟踪(运行时产物, 常驻dirty 会废掉漂移检测)
This commit is contained in:
@@ -0,0 +1,197 @@
|
||||
#!/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",
|
||||
"server.py", "mofin_db.py", "strategy_lifecycle.py", "xmpp_logger.py",
|
||||
"mo_config.py", "scripts/mo_data.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):
|
||||
try:
|
||||
import urllib.request
|
||||
req = urllib.request.Request(
|
||||
"http://127.0.0.1:5805/",
|
||||
data=json.dumps({"body": msg, "to": "hmo@yoin.fun", "type": "chat"}).encode(),
|
||||
headers={"Content-Type": "application/json"})
|
||||
urllib.request.urlopen(req, timeout=5)
|
||||
except Exception as e:
|
||||
log(f"XMPP发送失败: {e}")
|
||||
|
||||
|
||||
def git(*args, check=True):
|
||||
r = subprocess.run(["git", "-C", REPO] + list(args),
|
||||
capture_output=True, text=True, timeout=60)
|
||||
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():
|
||||
status, path = line[:2], line[3:].strip()
|
||||
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:]}")
|
||||
|
||||
|
||||
# ── 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:
|
||||
msg = "🛡️ 部署守卫\n"
|
||||
for a in actions:
|
||||
msg += f"✅ {a}\n"
|
||||
for p in problems:
|
||||
msg += f"⚠️ {p}\n"
|
||||
xmpp(msg.strip())
|
||||
log(f"── 结束: actions={len(actions)} problems={len(problems)} ──")
|
||||
return 0 if not problems else 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user