Ops batch: deploy automation, backup restore drill, monitoring, multi-user verification

Deploy: deploy.sh builds tarball, backs up current code (keep 3),
deploys, restarts, health-gates 60s (ok/backend=postgres/db_ok),
auto-rolls back on failure; rollback.sh restores latest backup.
Rollback is code-only, DB migrations are additive-only so old
code can still read new columns.
Backup: restore_postgres refuses to restore into the production
DB name unless explicitly allowed (drill never allows it);
restore-drill.sh backs up fresh, restores to a test DB, compares
tasks/projects/users/audit row counts, drops the test DB.
Monitoring: worker writes heartbeat file every loop (atomic
replace, failure only logged); GET /health adds
worker_last_seen_seconds_ago + worker_ok (180s default, env
tunable, still HTTP 200 when worker is stale); monitor.sh
checks server/worker/health/pg_isready/docker-postgres/disk/
app dir as one JSON line, exit non-zero on any fail, plus
systemd timer every 5 minutes. Push alerting to chat still
needs a separate channel decision (SSH key or webhook URL).
Multi-user: verify-multiuser.py creates test users/projects
in memory-key only, checks RBAC via real HTTP API (owner/
maintainer/contributor/viewer, cross-project isolation,
merge 200/403/409, no-key 401), test tasks carry a never-
satisfiable dependency so the production worker never claims
them, cleanup deletes only multiuser-test- rows (audit log
is append-only and keeps test entries by design).
Tests: host 27 pass, cloud 298 pass (8 skipped are
environment-dependent real-Postgres tests).
This commit is contained in:
host committed 2026-10-26 04:37:40 +08:00
1 parent 6afef78ca8
commit 89a918b543
14 files changed
+1386 -2

No files matched your search

+56
View File
@@ -570,3 +570,59 @@ interrupted 且终态任务 cancel 返回 400、心跳后 sweep 不扫刚心跳
任务仍需 worker 认领执行;`CloudExecutor` 仍是占位符;公网 https
入口仍需部署侧反向代理 + TLS。
## 运维批(本轮新增:部署、恢复演练、监控、心跳、多用户验证)
细则在 `deploy/README.md`,这里只记代码侧的接口与默认值:
- 部署与回滚(`deploy/deploy.sh` / `deploy/rollback.sh`,VM 侧跑,
不碰 `cloud/` 代码):部署前在主机上备份当前 `/opt/codex-app-server`
到 `/opt/codex-app-server-backups/`(时间戳命名,只留最近 3 份,排除
work/),解包新代码、chown codexsvc、重启 server+worker、循环验
`localhost:8765/health` 最多 60 秒(要求 ok=true、backend=postgres、
db_ok=true),失败自动恢复备份并重启;回滚只回代码,不回数据库
(迁移只加列、幂等,老代码仍能读新列,见 deploy/README 的回滚边界)。
两个脚本都不接受密码参数、不读 env 文件、不打印密钥;需经代理时用
环境变量 `SSH_PROXY_CMD` 传入 ProxyCommand,未设则直连。
- 备份恢复演练(`cloud.backup.restore_postgres` + `deploy/restore-drill.sh`,
主机侧跑):`restore_postgres(backup_file, target_dbname, runner=None)`,
用 psql 把 .sql 备份恢复到目标库;目标库名等于 `CLOUD_DB_NAME` 时
直接拒绝(除非显式传 `allow_production=True`,演练脚本永远不设它);
备份文件不存在或为空时报错;密码只经 PGPASSWORD 环境变量传,绝不进
命令行、返回值或日志。演练脚本先跑一次 `backup_postgres` 生成新鲜
备份,再恢复到 `codex_control_restore_test_<时间戳>` 测试库,比对
tasks/projects/users/audit_log 四张表行数(只比行数,不打印行内容),
比对完立即删测试库(trap 保证中途失败也删);绝不恢复到生产库,
绝不删生产库。
- 监控与心跳:worker 的 `run_loop` 每次循环结束时写心跳文件,内容只有
`{"last_seen": epoch秒, "pid": 进程号}`,写失败只记日志不炸 worker;
路径由 `--heartbeat-file` 指定,未指定时看环境变量
`CLOUD_WORKER_HEARTBEAT_FILE`,再未指定则为 `./worker-heartbeat.json`
(VM 本地默认,主机路径不在代码里硬编码;主机部署时两边指到同一个
`/var/lib/codex-app-server/worker-heartbeat.json`,server 侧在 env 文件
里加 `CLOUD_WORKER_HEARTBEAT_FILE` 一行,worker 单元加
`--heartbeat-file` 参数,deploy.sh 不自动改 env 文件)。`GET /health`
在原有 ok/service/backend/db_ok 之外加 `worker_last_seen_seconds_ago`
(无文件或读失败时为 None)与 `worker_ok`(心跳 180 秒内为 true,
阈值可经 `CLOUD_WORKER_HEARTBEAT_MAX_AGE_SECONDS` 调,默认 180);
health 本身仍返回 200,worker 不 ok 由调用方看字段判断,不要改成 503。
`deploy/monitor.sh`(主机侧,codexsvc 可读,不读 env 文件)检查
server/worker 的 systemctl 状态、health 四项、pg_isready、docker ps 里
codex-postgres 在运行、磁盘 <90%、app 目录存在,输出一行 JSON,任一
项 fail 则退出非零;`--self-test` 只验证脚本机制(health 解析 + JSON
合法),不验证主机健康。`deploy/monitor.service` /
`deploy/monitor.timer` 每 5 分钟跑一次,日志进 journal。
- 多用户真实验证(`deploy/verify-multiuser.py`,主机侧以 codexsvc 身份
跑):内存里建 4 个测试用户与 2 个测试项目(P2 只有 owner 是成员),
用 HTTP API 验证建任务、读任务、merge 的角色权限与跨项目隔离;
merge 用的 completed 测试任务是直接在 DB 里置 completed,不跑模型;
测试任务全部带永不存在的依赖 `multiuser-test-never-completes`,生产
worker 只派依赖全部 completed 的任务,所以测试任务永远不会被认领去
跑模型,不花模型钱、不改仓库;测试行全部用 `multiuser-test-` 前缀,
清理时只删这些行,清理失败退出非零并列出残留 id(不含 key);审计
日志是 append-only 的,验证审计条目会留下,脚本不删审计表。
- 仍未做(不要宣称已做):推送告警通道(monitor 失败时通知到 Muse 聊天
的通道还没定);回滚不回数据库(改类型/删列的迁移需单独评估);恢复
演练只比行数,恢复到生产库的真实灾难恢复流程还没演练过;部署脚本、
恢复演练、monitor、多用户验证四项的主机实跑还没做(本轮只在 VM 上
跑了单测、deploy 干跑、monitor 自测与 VM 实跑)。
+55
View File
@@ -101,6 +101,61 @@ def backup_postgres(out_dir, keep=3, timestamp=None, runner=None,
return dest
def restore_postgres(backup_file, target_dbname, runner=None,
psql_cmd="psql", allow_production=False):
"""Postgres 恢复:用 psql 把 .sql 备份恢复到目标库,返回目标库名。
安全规则(恢复演练的核心,绝不放宽):
- 目标库名绝不能等于生产库名(CLOUD_DB_NAME 环境变量读到的值),
相等时直接报错拒绝;只有调用方显式传 allow_production=True
才放行,而部署脚本(deploy/restore-drill.sh)永远不设这个值。
- 恢复前要求备份文件存在且非空,否则报错,不跑 psql。
- 连接参数与 backup_postgres 同款:host/port/user 走 CLOUD_DB_*
环境变量,密码只以 PGPASSWORD 环境变量传给子进程,绝不
出现在命令行参数、返回值或日志里。
- 目标库必须已存在(调用方先 CREATE DATABASE,见恢复演练
脚本);psql 非零退出时明确报错(stderr 截断,不含密码)。
- 本函数只恢复,不删库、不改生产库;删测试库是调用方的事。
"""
if not backup_file or not os.path.isfile(backup_file):
raise FileNotFoundError("备份文件不存在:%s" % backup_file)
if os.path.getsize(backup_file) == 0:
raise ValueError("备份文件为空,拒绝恢复:%s" % backup_file)
if not target_dbname or not isinstance(target_dbname, str) \
or not target_dbname.strip() \
or any(c.isspace() for c in target_dbname):
raise ValueError("目标库名非法:%r" % (target_dbname,))
production = os.environ.get("CLOUD_DB_NAME") or ""
if production and target_dbname == production \
and not allow_production:
raise ValueError(
"拒绝恢复到生产库 %s:目标库名与 CLOUD_DB_NAME 相同;"
"恢复演练必须用测试库名" % production)
host = os.environ.get("CLOUD_DB_HOST") or "127.0.0.1"
port = os.environ.get("CLOUD_DB_PORT") or "5432"
user = os.environ.get("CLOUD_DB_USER")
if not user:
raise RuntimeError("Postgres 恢复需要 CLOUD_DB_USER")
if shutil.which(psql_cmd) is None and runner is None:
raise RuntimeError("psql 不可用:本机找不到 %s" % psql_cmd)
cmd = [psql_cmd, "-h", host, "-p", str(port), "-U", user,
"-d", target_dbname, "-v", "ON_ERROR_STOP=1",
"-f", backup_file]
env = dict(os.environ)
password = os.environ.get("CLOUD_DB_PASSWORD")
if password:
env["PGPASSWORD"] = password
if runner is not None:
result = runner(cmd, env)
else:
result = subprocess.run(cmd, capture_output=True, text=True,
env=env)
if getattr(result, "returncode", 1) != 0:
err = (getattr(result, "stderr", "") or "")[:500]
raise RuntimeError("psql 恢复失败:%s" % err)
return target_dbname
def main(argv=None):
parser = argparse.ArgumentParser(description="cloud control-plane backup")
parser.add_argument("--db", default=None,
+48
View File
@@ -156,6 +156,47 @@ def _db_backend_name():
return (os.environ.get("CLOUD_DB_BACKEND") or "sqlite").strip().lower()
def _worker_heartbeat_status(server=None):
"""读 worker 运维心跳,返回 (age_seconds_or_None, worker_ok)。
心跳文件路径:测试注入优先(server.worker_heartbeat_file),
其次环境变量 CLOUD_WORKER_HEARTBEAT_FILE,未设则默认
./worker-heartbeat.json——与 worker 的默认路径保持一致,主机
部署时两边都要指到同一个主机路径(见 deploy/README)。
阈值:server.worker_heartbeat_max_age_seconds(测试注入)或
环境变量 CLOUD_WORKER_HEARTBEAT_MAX_AGE_SECONDS,默认 180 秒。
文件不存在、读失败、JSON 非法、last_seen 非数字时一律返回
(None, False),不抛异常、不影响 health 本身的 200。
"""
path = None
max_age = 180
if server is not None:
path = getattr(server, "worker_heartbeat_file", None)
override = getattr(server, "worker_heartbeat_max_age_seconds", None)
if override is not None:
try:
max_age = float(override)
except (TypeError, ValueError):
max_age = 180
if not path:
path = os.environ.get("CLOUD_WORKER_HEARTBEAT_FILE") or \
"./worker-heartbeat.json"
if max_age == 180:
try:
max_age = float(os.environ.get(
"CLOUD_WORKER_HEARTBEAT_MAX_AGE_SECONDS") or 180)
except (TypeError, ValueError):
max_age = 180
try:
with open(path, encoding="utf-8") as f:
payload = json.load(f)
last_seen = float(payload.get("last_seen"))
except Exception:
return None, False
age = time.time() - last_seen
return round(age, 1), bool(age <= max_age)
def _connect(db_path):
"""每个请求新建一个连接;后端按 CLOUD_DB_BACKEND 分发。
@@ -448,9 +489,16 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
conn.close()
except Exception:
db_ok = False
# worker 字段:worker_ok 只表示“worker 心跳新鲜”,worker
# 不 ok 不代表 server 挂了,所以 health 本身仍返回 200,
# 调用方看 worker_ok 字段判断,不要把 health 改成 503
# 破坏已有调用(monitor.sh 也是按字段判断,不按状态码)。
age, worker_ok = _worker_heartbeat_status(self.server)
self._send_json(200, {
"ok": True, "service": "cloud-control-plane",
"backend": _db_backend_name(), "db_ok": db_ok,
"worker_last_seen_seconds_ago": age,
"worker_ok": worker_ok,
})
return
if path.startswith("/oauth/"):
+206
View File
@@ -0,0 +1,206 @@
"""运维批单测:恢复演练、worker 心跳、health worker 字段、monitor 自测。
只用标准库,不跑真实模型回合、不联网、不连真实 GitHub、不连生产库;
restore 的 psql 与 monitor 的 systemctl/docker 一律用假 runner 或
--self-test 的 canned server 验证,不在本机起真实服务。
"""
import http.client
import json
import os
import subprocess
import sys
import tempfile
import threading
import time
import unittest
from pathlib import Path
from types import SimpleNamespace
try:
from cloud.backup import restore_postgres
from cloud.server import create_server
from cloud.worker import Worker
except ImportError:
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from cloud.backup import restore_postgres
from cloud.server import create_server
from cloud.worker import Worker
def _fake_result(returncode=0, stdout="", stderr=""):
return SimpleNamespace(returncode=returncode, stdout=stdout,
stderr=stderr)
class RestorePostgresTestCase(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.backup = os.path.join(self.tmp.name, "backup.sql")
with open(self.backup, "w") as f:
f.write("-- fake dump\n")
self.old_env = dict(os.environ)
os.environ["CLOUD_DB_HOST"] = "127.0.0.1"
os.environ["CLOUD_DB_PORT"] = "5432"
os.environ["CLOUD_DB_NAME"] = "prod_db"
os.environ["CLOUD_DB_USER"] = "codexsvc"
os.environ["CLOUD_DB_PASSWORD"] = "test-secret-pw"
def tearDown(self):
os.environ.clear()
os.environ.update(self.old_env)
self.tmp.cleanup()
def test_target_equal_to_production_rejected(self):
# 目标库名等于生产库名时直接拒绝,allow_production 默认 False。
with self.assertRaises(ValueError):
restore_postgres(self.backup, "prod_db",
runner=lambda c, e: _fake_result())
def test_target_equal_to_production_allowed_only_explicitly(self):
calls = []
def runner(cmd, env):
calls.append((list(cmd), dict(env)))
return _fake_result()
out = restore_postgres(self.backup, "prod_db", runner=runner,
allow_production=True)
self.assertEqual(out, "prod_db")
self.assertEqual(len(calls), 1)
def test_missing_backup_file_rejected(self):
with self.assertRaises(FileNotFoundError):
restore_postgres(os.path.join(self.tmp.name, "nope.sql"),
"restore_test",
runner=lambda c, e: _fake_result())
def test_empty_backup_file_rejected(self):
empty = os.path.join(self.tmp.name, "empty.sql")
open(empty, "w").close()
with self.assertRaises(ValueError):
restore_postgres(empty, "restore_test",
runner=lambda c, e: _fake_result())
def test_password_only_in_env_not_in_command(self):
calls = []
def runner(cmd, env):
calls.append((list(cmd), dict(env)))
return _fake_result()
out = restore_postgres(self.backup, "restore_test", runner=runner)
self.assertEqual(out, "restore_test")
cmd, env = calls[0]
self.assertNotIn("test-secret-pw", " ".join(cmd))
self.assertEqual(env.get("PGPASSWORD"), "test-secret-pw")
self.assertIn("-f", cmd)
self.assertIn(self.backup, cmd)
self.assertIn("restore_test", cmd)
def test_psql_failure_raises(self):
def runner(cmd, env):
return _fake_result(returncode=1, stderr="psql boom")
with self.assertRaises(RuntimeError):
restore_postgres(self.backup, "restore_test", runner=runner)
class WorkerHeartbeatTestCase(unittest.TestCase):
def test_heartbeat_file_written_with_last_seen_and_pid(self):
tmp = tempfile.TemporaryDirectory()
hb = os.path.join(tmp.name, "worker-heartbeat.json")
worker = Worker(os.path.join(tmp.name, "db.db"), "/nonexistent/host.py",
env_key="TEST_KEY", heartbeat_file=hb)
worker._write_heartbeat()
with open(hb) as f:
payload = json.load(f)
self.assertIn("last_seen", payload)
self.assertEqual(payload["pid"], os.getpid())
self.assertLess(abs(payload["last_seen"] - time.time()), 5)
tmp.cleanup()
def test_heartbeat_write_failure_does_not_crash_worker(self):
# 心跳目录不可写时只记日志,不抛异常(写到一个文件路径下当目录)。
tmp = tempfile.TemporaryDirectory()
blocker = os.path.join(tmp.name, "blocker")
open(blocker, "w").close()
worker = Worker(os.path.join(tmp.name, "db.db"), "/nonexistent/host.py",
env_key="TEST_KEY",
heartbeat_file=os.path.join(blocker, "hb.json"))
worker._write_heartbeat() # 不抛异常即通过
tmp.cleanup()
class HealthWorkerFieldTestCase(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.db_path = os.path.join(self.tmp.name, "control.db")
self.server = create_server(self.db_path, "127.0.0.1", 0)
self.port = self.server.server_address[1]
self.thread = threading.Thread(
target=self.server.serve_forever, daemon=True)
self.thread.start()
def tearDown(self):
self.server.shutdown()
self.server.server_close()
self.thread.join(timeout=5)
self.tmp.cleanup()
def _health(self):
conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10)
conn.request("GET", "/health")
resp = conn.getresponse()
body = json.loads(resp.read())
conn.close()
return resp.status, body
def test_health_worker_fresh_ok(self):
hb = os.path.join(self.tmp.name, "hb.json")
with open(hb, "w") as f:
json.dump({"last_seen": time.time(), "pid": 123}, f)
self.server.worker_heartbeat_file = hb
status, body = self._health()
self.assertEqual(status, 200)
self.assertTrue(body["ok"])
self.assertTrue(body["worker_ok"])
self.assertIsNotNone(body["worker_last_seen_seconds_ago"])
self.assertLess(body["worker_last_seen_seconds_ago"], 5)
def test_health_worker_stale_not_ok_but_still_200(self):
hb = os.path.join(self.tmp.name, "hb.json")
with open(hb, "w") as f:
json.dump({"last_seen": time.time() - 1000, "pid": 123}, f)
self.server.worker_heartbeat_file = hb
status, body = self._health()
# worker 不 ok 不代表 server 挂了,health 本身仍返回 200。
self.assertEqual(status, 200)
self.assertFalse(body["worker_ok"])
self.assertGreater(body["worker_last_seen_seconds_ago"], 500)
def test_health_worker_missing_file_none_false(self):
self.server.worker_heartbeat_file = os.path.join(
self.tmp.name, "no-such-hb.json")
status, body = self._health()
self.assertEqual(status, 200)
self.assertIsNone(body["worker_last_seen_seconds_ago"])
self.assertFalse(body["worker_ok"])
class MonitorSelfTestCase(unittest.TestCase):
def test_monitor_self_test_passes(self):
# monitor.sh --self-test 只验证脚本机制(health 解析 + JSON 合法),
# 不断言主机健康;VM 上其它项 fail 是正常的,self-test 仍应通过。
repo_root = Path(__file__).resolve().parents[2]
script = repo_root / "deploy" / "monitor.sh"
self.assertTrue(script.exists())
result = subprocess.run(
["sh", str(script), "--self-test"],
capture_output=True, text=True, timeout=60)
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
self.assertIn("SELF-TEST PASS", result.stdout)
if __name__ == "__main__":
unittest.main()
+40 -2
View File
@@ -208,7 +208,7 @@ class Worker:
cost_store=None, process_factory=None,
task_timeout=1800, max_retries=1,
retry_backoff_seconds=5.0, repos_dir="./repos",
repos_runner=None):
repos_runner=None, heartbeat_file=None):
self.db_path = str(db_path)
self.host_py_path = host_py_path
self.config_path = config_path
@@ -252,6 +252,31 @@ class Worker:
# repos_runner 可注入(单测用假 git runner),缺省走真实 git。
self.repos_dir = repos_dir
self.repos_runner = repos_runner
# 运维心跳:路径优先显式参数,其次环境变量,默认当前目录下
# 的 worker-heartbeat.json;主机路径不在代码里硬编码,主机
# 部署时用 --heartbeat-file 或环境变量传入(见 deploy/README)。
# 文件里只有 last_seen 与 pid,绝无任何密钥。
self.heartbeat_file = heartbeat_file or \
os.environ.get("CLOUD_WORKER_HEARTBEAT_FILE") or \
"./worker-heartbeat.json"
def _write_heartbeat(self):
"""写一次运维心跳:写失败只记日志,绝不炸 worker。
先写临时文件再 os.replace,保证 server 读到的要么是旧的
完整文件、要么是新的完整文件,不会读到写一半的 JSON。
"""
try:
payload = {"last_seen": time.time(), "pid": os.getpid()}
tmp = "%s.tmp" % self.heartbeat_file
parent = os.path.dirname(os.path.abspath(self.heartbeat_file))
os.makedirs(parent, exist_ok=True)
with open(tmp, "w", encoding="utf-8") as f:
f.write(json.dumps(payload))
os.replace(tmp, self.heartbeat_file)
except Exception as exc:
log_event(self.logger, "worker_heartbeat_failed",
message="worker heartbeat write failed: %s" % exc)
# -- 成本存储(惰性建表,避免只认领不执行时也建表)--------------
@property
@@ -954,6 +979,9 @@ class Worker:
# 常驻启动清理兜底:清掉上一轮残留的 codex- 容器,
# 失败只记日志(见 cleanup_stale_containers),不炸循环。
self.cleanup_stale_containers()
# 运维心跳:启动时先写一次,之后每轮结束再写,server 的
# /health 据此算 worker_ok(见 server.py 的健康检查)。
self._write_heartbeat()
started = time.time()
all_results = []
try:
@@ -972,6 +1000,9 @@ class Worker:
message="worker_loop_error: %s" % exc)
results = []
all_results.extend(results)
# 每轮结束写心跳:哪怕本轮一个任务都没派,worker 活着
# 这件事本身也要被 server 看到。
self._write_heartbeat()
# 按 0.2 秒切片睡,期间可响应 stop_event/_loop_stop。
deadline = time.time() + interval
while time.time() < deadline:
@@ -1025,6 +1056,12 @@ def main(argv=None):
help="host.py 路径,缺省为 cloud/ 上一级的 host.py")
parser.add_argument("--config", default=None)
parser.add_argument("--repo", default=None)
parser.add_argument("--heartbeat-file", default=None,
help="运维心跳文件路径(默认:环境变量 "
"CLOUD_WORKER_HEARTBEAT_FILE,未设则为 "
"./worker-heartbeat.json;主机部署时用此"
"参数指到 /var/lib/codex-app-server/ 下,"
"与 server 的 health 读取路径保持一致)")
parser.add_argument("--repos-dir", default="./repos",
help="项目仓库检出根目录(默认 ./repos);项目有"
" repo_url 时 worker 在此 clone/fetch 各项目"
@@ -1081,7 +1118,8 @@ def main(argv=None):
task_timeout=args.task_timeout,
max_retries=args.max_retries,
retry_backoff_seconds=args.retry_backoff_seconds,
repos_dir=args.repos_dir)
repos_dir=args.repos_dir,
heartbeat_file=args.heartbeat_file)
# CLI 启动清理兜底(常驻与单次都做):清掉上一轮残留容器。
worker.cleanup_stale_containers()
# CLI 起 worker 前先建表/迁移:全新库直接起循环不应因缺表而炸。