HA stage 0: health reads workers registry, daily pg_dump timer, backup freshness in monitor (#48)

Co-authored-by: host <host@test.local>
This commit is contained in:
openqjandhost authored and GitHub committed 2026-10-26 23:51:54 +08:00
1 parent 275de39148
commit 455513dd33
6 files changed
+150 -10

No files matched your search

+47 -6
View File
@@ -205,10 +205,33 @@ def _db_backend_name():
return (os.environ.get("CLOUD_DB_BACKEND") or "sqlite").strip().lower()
def _deployed_commit():
"""当前上线代码的 commit:deploy.py 部署时在包根写的 DEPLOYED_COMMIT。
文件不存在、读失败、内容为空时返回 None(老包、本地开发都走这支),
health 绝不因此失败。commit 是公开的版本标识,不是密钥。
路径取 cloud/ 的父目录(即仓库根),不依赖当前工作目录。"""
path = os.path.join(
os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
"DEPLOYED_COMMIT")
try:
with open(path, encoding="utf-8") as f:
value = f.read().strip()
except Exception:
return None
return value or None
def _worker_heartbeat_status(server=None):
"""读 worker 运维心跳,返回 (age_seconds_or_None, worker_ok)。
心跳文件路径:测试注入优先(server.worker_heartbeat_file),
优先读 workers 注册表(多机前提):所有 worker 每轮都往
workers 表写 last_seen(见 worker._registry_heartbeat),而文件
心跳只在本机存在——多 server 下 server 读不到别机 worker 的
文件,注册表是唯一跨机可靠的心跳源。表不存在、无行、查库
失败时回落文件心跳(单机旧行为与单测保持兼容)。
文件心跳路径:测试注入优先(server.worker_heartbeat_file),
其次环境变量 CLOUD_WORKER_HEARTBEAT_FILE,未设则默认
./worker-heartbeat.json——与 worker 的默认路径保持一致,主机
部署时两边都要指到同一个主机路径(见 deploy/README)。
@@ -217,25 +240,40 @@ def _worker_heartbeat_status(server=None):
文件不存在、读失败、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
db_path = getattr(server, "db_path", None) if server is not None else None
if db_path:
try:
conn = _connect(db_path)
try:
row = conn.execute(
"SELECT MAX(last_seen) FROM workers").fetchone()
finally:
conn.close()
if row is not None and row[0] is not None:
age = time.time() - float(row[0])
return round(age, 1), bool(age <= max_age)
except Exception:
pass
path = None
if server is not None:
path = getattr(server, "worker_heartbeat_file", None)
if not path:
path = os.environ.get("CLOUD_WORKER_HEARTBEAT_FILE") or \
"./worker-heartbeat.json"
try:
with open(path, encoding="utf-8") as f:
payload = json.load(f)
@@ -574,6 +612,9 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
"backend": _db_backend_name(), "db_ok": db_ok,
"worker_last_seen_seconds_ago": age,
"worker_ok": worker_ok,
# 上线证据:health 通过不再单独证明新代码,
# 部署验真必须同时核对 commit 等于期望值(见 deploy.py)。
"commit": _deployed_commit(),
})
return
if path.startswith("/oauth/"):
+31
View File
@@ -21,11 +21,13 @@ try:
from cloud.backup import restore_postgres
from cloud.server import create_server
from cloud.worker import Worker
from cloud.workers import heartbeat_worker, register_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
from cloud.workers import heartbeat_worker, register_worker
def _fake_result(returncode=0, stdout="", stderr=""):
@@ -187,6 +189,35 @@ class HealthWorkerFieldTestCase(unittest.TestCase):
self.assertIsNone(body["worker_last_seen_seconds_ago"])
self.assertFalse(body["worker_ok"])
def test_health_worker_db_registry_fresh_ok_without_file(self):
# 多机前提:文件心跳缺失时,DB 注册表的新鲜心跳同样算 ok。
register_worker(self.db_path, "w-db", executor="docker", capacity=1)
heartbeat_worker(self.db_path, "w-db", executor="docker")
self.server.worker_heartbeat_file = os.path.join(
self.tmp.name, "no-such-hb.json")
status, body = self._health()
self.assertEqual(status, 200)
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_db_registry_stale_not_ok(self):
register_worker(self.db_path, "w-db", executor="docker", capacity=1)
import sqlite3
conn = sqlite3.connect(self.db_path)
try:
conn.execute("UPDATE workers SET last_seen = ?",
(time.time() - 1000,))
conn.commit()
finally:
conn.close()
self.server.worker_heartbeat_file = os.path.join(
self.tmp.name, "no-such-hb.json")
status, body = self._health()
self.assertEqual(status, 200)
self.assertFalse(body["worker_ok"])
self.assertGreater(body["worker_last_seen_seconds_ago"], 500)
class MonitorSelfTestCase(unittest.TestCase):
def test_monitor_self_test_passes(self):