deploy_health_check.py
7.38 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
"""c_deploy_supervisor_health_check (03-C §3.15 / 04-P §4.6 / design doc §6):
three pre-cutover checks against a staged candidate -- `process_alive`,
`http_health`, `canary_task`. All confirmed fail-closed and non-retrying per
03-C's own text: "任一检查项 fail 立即整体判定 health_check_failed，不等待
其余检查项跑完" and "不重试——重试属于'给坏版本第二次机会'". Timeouts are the
MVP-fixed draft values from 05 §5.9 (`HEALTH_CHECK_STARTUP_TIMEOUT`=30s,
`HEALTH_CHECK_HTTP_TIMEOUT`=10s, `HEALTH_CHECK_CANARY_TIMEOUT`=120s) --
05 §5.9's own text already flags these as engineering drafts needing
real-world calibration, not authoritative constants to bake into assertions
elsewhere.
"""
import json
import shutil
import subprocess
import tempfile
import time
import urllib.error
import urllib.request
import uuid
from supervisor import deploy_supervisor
HEALTH_CHECK_STARTUP_TIMEOUT = 30
HEALTH_CHECK_HTTP_TIMEOUT = 10
HEALTH_CHECK_CANARY_TIMEOUT = 120
_SIGNAL_TYPES = ("process_alive", "http_health", "canary_task")
def check_process_alive(pid, timeout=HEALTH_CHECK_STARTUP_TIMEOUT, sleep=time.sleep, now=time.time):
deadline = now() + timeout
while now() < deadline:
if deploy_supervisor.is_alive(pid):
return {"signal_type": "process_alive", "status": "ok", "detail": None}
sleep(0.5)
return {
"signal_type": "process_alive", "status": "fail",
"detail": "pid {} not alive within {}s".format(pid, timeout),
}
def check_http_health(port, timeout=HEALTH_CHECK_HTTP_TIMEOUT):
try:
with urllib.request.urlopen("http://127.0.0.1:{}/api/health".format(port), timeout=timeout) as resp:
if resp.status == 200:
return {"signal_type": "http_health", "status": "ok", "detail": None}
return {
"signal_type": "http_health", "status": "fail",
"detail": "unexpected HTTP status {}".format(resp.status),
}
except urllib.error.HTTPError as e:
# HTTPError IS the response object (a non-2xx status raises rather
# than returning) -- close it explicitly, `except` doesn't run our
# `with` block's __exit__ for us.
detail = "unexpected HTTP status {}".format(e.code)
e.close()
return {"signal_type": "http_health", "status": "fail", "detail": detail}
except (urllib.error.URLError, OSError) as e:
return {"signal_type": "http_health", "status": "fail", "detail": str(e)}
def _post_json(port, path, payload, timeout):
body = json.dumps(payload).encode("utf-8")
req = urllib.request.Request(
"http://127.0.0.1:{}{}".format(port, path), data=body,
headers={"Content-Type": "application/json"}, method="POST",
)
with urllib.request.urlopen(req, timeout=timeout) as resp:
return resp.status, json.loads(resp.read())
def _get_json(port, path, timeout):
with urllib.request.urlopen("http://127.0.0.1:{}{}".format(port, path), timeout=timeout) as resp:
return resp.status, json.loads(resp.read())
def check_canary_task(port, timeout=HEALTH_CHECK_CANARY_TIMEOUT):
""""结果可预期的最小任务" (design doc §6): a trivial project + task
creation round-trip. Doesn't run a real Org Graph (no real `claude`
calls, no real cost/latency) -- `c_task_create` itself lands
`status='queued'` synchronously, which is exactly the kind of
deterministic, always-should-pass result a canary needs; the point is
proving the candidate's HTTP+DB write path genuinely works, not
exercising the full pipeline."""
canary_dir = tempfile.mkdtemp(prefix="deploy-canary-")
try:
subprocess.run(["git", "init", "-q"], cwd=canary_dir, check=True, capture_output=True)
subprocess.run(
["git", "-c", "user.name=graphlemon", "-c", "user.email=graphlemon@localhost",
"commit", "--allow-empty", "-q", "-m", "canary"],
cwd=canary_dir, check=True, capture_output=True,
)
status, project = _post_json(
port, "/api/projects", {"target_folder_path": canary_dir, "name": "deploy-canary"}, timeout
)
if status != 200:
return {"signal_type": "canary_task", "status": "fail", "detail": "project create: HTTP {}".format(status)}
goal_text = "deploy-supervisor canary " + uuid.uuid4().hex
status, task = _post_json(
port, "/api/tasks", {"project_id": project["project_id"], "goal_text": goal_text}, timeout
)
if status != 200:
return {"signal_type": "canary_task", "status": "fail", "detail": "task create: HTTP {}".format(status)}
status, fetched = _get_json(port, "/api/tasks/" + task["task_id"], timeout)
if status != 200 or fetched.get("status") != "queued" or fetched.get("goal_text") != goal_text:
return {
"signal_type": "canary_task", "status": "fail",
"detail": "unexpected task state after create: HTTP {} {!r}".format(status, fetched),
}
return {"signal_type": "canary_task", "status": "ok", "detail": None}
except urllib.error.HTTPError as e:
detail = "unexpected HTTP status {}".format(e.code)
e.close()
return {"signal_type": "canary_task", "status": "fail", "detail": detail}
except (urllib.error.URLError, OSError, subprocess.CalledProcessError, KeyError) as e:
return {"signal_type": "canary_task", "status": "fail", "detail": str(e)}
finally:
shutil.rmtree(canary_dir, ignore_errors=True)
def _record_signal(conn, deployment_id, signal):
conn.execute(
"INSERT INTO deploy_health_signal (signal_id, deployment_id, sampled_at, signal_type, status, detail) "
"VALUES (?, ?, CURRENT_TIMESTAMP, ?, ?, ?)",
(str(uuid.uuid4()), deployment_id, signal["signal_type"], signal["status"], signal["detail"]),
)
conn.commit()
def run_health_check(
conn, deployment_id, pid, port,
startup_timeout=HEALTH_CHECK_STARTUP_TIMEOUT, http_timeout=HEALTH_CHECK_HTTP_TIMEOUT,
canary_timeout=HEALTH_CHECK_CANARY_TIMEOUT, sleep=time.sleep, now=time.time,
):
"""c_deploy_supervisor_health_check. Returns {status, signals}. Marks
`deployment.status='health_check_running'` on entry, then
'health_check_passed'/'health_check_failed' once done -- fails closed
and stops at the FIRST failing check (no point running http_health/
canary_task against a process that isn't even alive yet)."""
conn.execute("UPDATE deployment SET status = 'health_check_running' WHERE deployment_id = ?", (deployment_id,))
conn.commit()
signals = []
checks = (
lambda: check_process_alive(pid, timeout=startup_timeout, sleep=sleep, now=now),
lambda: check_http_health(port, timeout=http_timeout),
lambda: check_canary_task(port, timeout=canary_timeout),
)
for check in checks:
signal = check()
_record_signal(conn, deployment_id, signal)
signals.append(signal)
if signal["status"] != "ok":
conn.execute(
"UPDATE deployment SET status = 'health_check_failed' WHERE deployment_id = ?", (deployment_id,)
)
conn.commit()
return {"status": "health_check_failed", "signals": signals}
conn.execute("UPDATE deployment SET status = 'health_check_passed' WHERE deployment_id = ?", (deployment_id,))
conn.commit()
return {"status": "health_check_passed", "signals": signals}