deploy_health_check.py
9.31 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
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
"""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
# MVP engineering estimate per 05 §5.9 -- needs real-world calibration, not an authoritative constant.
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):
"""Uses `reap_if_exited` (not plain `is_alive`) so a candidate that
crashed immediately after being spawned by THIS process reads as dead
rather than a false "alive" zombie reading -- see
`deploy_supervisor.is_alive`'s own docstring for why the distinction
matters here specifically (this module's caller is always the same
process that Popen'd the candidate)."""
deadline = now() + timeout
while now() < deadline:
if not deploy_supervisor.reap_if_exited(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, sleep=time.sleep, now=time.time):
"""Polls until `timeout` elapses, same as `check_process_alive` --
NOT the "重试属于给坏版本第二次机会" policy this module's own docstring
warns against (real self-hosting runs, 2026-08-03, hit this: `pid`
reads alive within milliseconds of `Popen`, well before the candidate
process has finished importing modules and actually bound its listen
socket -- `urlopen` on a port nothing is listening on yet raises
ECONNREFUSED essentially instantly, `timeout=` only bounds how long to
wait for a RESPONSE once connected, not how long to wait for the port
to start accepting connections at all). `check_process_alive` already
established the precedent that tolerating normal startup latency
within ONE health-check pass isn't "retrying a failed check" -- this
just extends the same tolerance to the next check in the same pass.
A genuine non-2xx HTTP response IS a definitive answer from a server
that's actually up, so that still fails immediately without polling,
preserving the real "don't give a bad build a second chance" intent.
"""
deadline = now() + timeout
last_detail = None
while now() < deadline:
try:
with urllib.request.urlopen("http://127.0.0.1:{}/api/health".format(port), timeout=1) 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. A real response from a live
# server, not a startup race -- fail immediately, no polling.
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:
last_detail = str(e)
sleep(0.5)
return {"signal_type": "http_health", "status": "fail", "detail": last_detail}
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, sleep=sleep, now=now),
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}