From a9e2f70d9e3831d1af8e4671741360b00ca1f367 Mon Sep 17 00:00:00 2001 From: host Date: Sat, 31 Oct 2026 01:26:28 +0000 Subject: [PATCH] workers: reap pid_verified stale workers with pre-kill audit and post-kill verification --- cloud/tests/test_workers_pid_kill.py | 278 +++++++++++++++++++++++++++ cloud/workers.py | 151 +++++++++++++++ 2 files changed, 429 insertions(+) create mode 100644 cloud/tests/test_workers_pid_kill.py diff --git a/cloud/tests/test_workers_pid_kill.py b/cloud/tests/test_workers_pid_kill.py new file mode 100644 index 0000000..01f7b1c --- /dev/null +++ b/cloud/tests/test_workers_pid_kill.py @@ -0,0 +1,278 @@ +"""workers 按 pid 精确杀执行路径单测：复用判定、只杀可杀档。 + +只用标准库 + 本地 SQLite 临时库，不联网、不杀任何真实存活 +进程、不碰任何容器：可杀档的“真杀”一律用注入的 kill_fn +测试替身（记录调用，不发真实信号）；默认杀路径只对一个 +已退出并 wait 过的子进程 pid（或等价的已死 pid）跑，且该 +pid 在判定与执行时均经 pid_is_dead 确认为已死。 +不可杀档（异机、无 pid、pid 存活、未失联、恰等于阈值） +只列不杀：kill_fn 绝不收到它们的 pid，杀审计也绝不含它们。 +""" + +import os +import sqlite3 +import subprocess +import sys +import tempfile +import time +import unittest +from pathlib import Path + +try: + from cloud.workers import ( + classify_stale_workers, + reap_stale_workers, + register_worker, + ) +except ImportError: + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.workers import ( + classify_stale_workers, + reap_stale_workers, + register_worker, + ) + + +def _set_last_seen(db_path, worker_id, last_seen): + conn = sqlite3.connect(db_path) + try: + conn.execute("UPDATE workers SET last_seen = ? WHERE worker_id = ?", + (last_seen, worker_id)) + conn.commit() + finally: + conn.close() + + +def _by_id(rows, worker_id): + for row in rows: + if row["worker_id"] == worker_id: + return row + raise AssertionError("missing worker %s" % worker_id) + + +def _audit_rows(db_path): + conn = sqlite3.connect(db_path) + try: + return conn.execute( + "SELECT action, detail FROM audit_log ORDER BY id").fetchall() + finally: + conn.close() + + +class PidKillReapTestCase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "control.db") + self.now = time.time() + register_worker(self.db_path, "w-kill", host="h-local", pid=9001) + register_worker(self.db_path, "w-eq", host="h-local", pid=9002) + register_worker(self.db_path, "w-foreign", host="h-other", + pid=9003) + register_worker(self.db_path, "w-nopid", host="h-local") + register_worker(self.db_path, "w-alive", host="h-local", pid=9004) + register_worker(self.db_path, "w-fresh", host="h-local", pid=9005) + _set_last_seen(self.db_path, "w-kill", self.now - 121) + _set_last_seen(self.db_path, "w-eq", self.now - 120) + _set_last_seen(self.db_path, "w-foreign", self.now - 200) + _set_last_seen(self.db_path, "w-nopid", self.now - 200) + _set_last_seen(self.db_path, "w-alive", self.now - 200) + _set_last_seen(self.db_path, "w-fresh", self.now - 10) + self.dead = {9001, 9002, 9003} + self.events = [] + self.kills = [] + + def tearDown(self): + self.tmp.cleanup() + + def _checker(self, pid): + return pid in self.dead + + def _kill_fn(self, pid): + self.events.append(("kill", pid)) + self.kills.append(pid) + + def _audit_fn(self, action, detail): + self.events.append(("audit", action, detail)) + + def _reap(self): + return reap_stale_workers( + self.db_path, stale_seconds=120, local_host="h-local", + pid_is_dead_fn=self._checker, kill_fn=self._kill_fn, + audit_fn=self._audit_fn, now=self.now) + + def test_only_kill_row_is_killed_and_audited_before_kill(self): + rows = self._reap() + # 真杀只发生在唯一可杀档，且杀前已有审计（事件顺序为证）。 + self.assertEqual(self.kills, [9001]) + kill_idx = self.events.index(("kill", 9001)) + audit_idx = next( + i for i, e in enumerate(self.events) + if e[0] == "audit" and e[1] == "worker_pid_kill") + self.assertLess(audit_idx, kill_idx) + row = _by_id(rows, "w-kill") + self.assertTrue(row["executed"]) + self.assertEqual(row["outcome"], "killed") + self.assertTrue(row["verified_dead"]) + self.assertTrue(row["audit_before"]) + self.assertTrue(row["audit_after"]) + self.assertIsNone(row["kill_error"]) + # 杀后审计在位，detail 含 outcome。 + after = [e for e in self.events + if e[0] == "audit" + and e[1] == "worker_pid_kill_verified"] + self.assertEqual(len(after), 1) + self.assertIn("outcome=killed", after[0][2]) + self.assertIn("worker_id=w-kill", after[0][2]) + + def test_non_kill_rows_are_listed_only(self): + rows = self._reap() + for worker_id in ("w-eq", "w-foreign", "w-nopid", "w-alive", + "w-fresh"): + row = _by_id(rows, worker_id) + self.assertFalse(row["executed"], worker_id) + self.assertEqual(row["outcome"], "listed", worker_id) + self.assertEqual(row["action"], "list_only", worker_id) + # 审计 detail 里绝不出现不可杀档的 worker_id。 + details = " ".join(e[2] for e in self.events if e[0] == "audit") + for worker_id in ("w-eq", "w-foreign", "w-nopid", "w-alive", + "w-fresh"): + self.assertNotIn(worker_id, details) + + def test_default_audit_writes_audit_log_rows(self): + # 不注 audit_fn 时默认写 audit_log 表：杀前与杀后各一条， + # 且只含 w-kill。 + rows = reap_stale_workers( + self.db_path, stale_seconds=120, local_host="h-local", + pid_is_dead_fn=self._checker, kill_fn=self._kill_fn, + now=self.now) + self.assertEqual(_by_id(rows, "w-kill")["outcome"], "killed") + audit = _audit_rows(self.db_path) + actions = [a for a, _ in audit] + self.assertEqual(actions, ["worker_pid_kill", + "worker_pid_kill_verified"]) + self.assertIn("worker_id=w-kill", audit[0][1]) + self.assertIn("outcome=killed", audit[1][1]) + + def test_pid_alive_at_execution_is_skipped_not_killed(self): + # 判定后 pid 被复用（checker 后续调用变 False）：执行时 + # 再验发现已不死，跳过不杀——判定快照不能当执行凭证。 + # 按 pid 计次：9001 首次调用（判定内）为 True，此后 + # 一律 False；其它 pid 按 dead 集判定（worker 行按 + # worker_id 排序，w-alive 会先于 w-kill 调 checker， + # 全局计次会在此误伤判定）。 + counts = {} + + def flip_checker(pid): + counts[pid] = counts.get(pid, 0) + 1 + if pid == 9001 and counts[pid] > 1: + return False + return pid in self.dead + + rows = reap_stale_workers( + self.db_path, stale_seconds=120, local_host="h-local", + pid_is_dead_fn=flip_checker, kill_fn=self._kill_fn, + audit_fn=self._audit_fn, now=self.now) + self.assertEqual(self.kills, []) + row = _by_id(rows, "w-kill") + self.assertFalse(row["executed"]) + self.assertEqual(row["outcome"], + "skipped_pid_alive_at_execution") + self.assertFalse(row["audit_before"]) + + def test_audit_failure_blocks_kill(self): + def failing_audit(action, detail): + raise RuntimeError("audit store down") + + rows = reap_stale_workers( + self.db_path, stale_seconds=120, local_host="h-local", + pid_is_dead_fn=self._checker, kill_fn=self._kill_fn, + audit_fn=failing_audit, now=self.now) + self.assertEqual(self.kills, []) + row = _by_id(rows, "w-kill") + self.assertFalse(row["executed"]) + self.assertEqual(row["outcome"], "audit_failed") + self.assertFalse(row["audit_before"]) + + def test_kill_error_with_dead_pid_still_killed(self): + # kill_fn 抛非 Lookup 的 OSError，但复核已死：成败以 + # 复核为准，kill_error 如实记录，不谎报失败。 + def boom_kill(pid): + self.kills.append(pid) + raise OSError("boom") + + rows = reap_stale_workers( + self.db_path, stale_seconds=120, local_host="h-local", + pid_is_dead_fn=self._checker, kill_fn=boom_kill, + audit_fn=self._audit_fn, now=self.now) + row = _by_id(rows, "w-kill") + self.assertEqual(row["outcome"], "killed") + self.assertTrue(row["verified_dead"]) + self.assertEqual(row["kill_error"], "boom") + + def test_kill_error_with_live_pid_is_failed(self): + # 判定 True、执行再验 True、杀后复核 False：杀失败， + # outcome 如实为 kill_failed_pid_alive。 + seq = {"9001": [True, True, False]} + + def seq_checker(pid): + values = seq.get(str(pid)) + if values: + return values.pop(0) + return pid in self.dead + + rows = reap_stale_workers( + self.db_path, stale_seconds=120, local_host="h-local", + pid_is_dead_fn=seq_checker, kill_fn=self._kill_fn, + audit_fn=self._audit_fn, now=self.now) + row = _by_id(rows, "w-kill") + self.assertEqual(row["outcome"], "kill_failed_pid_alive") + self.assertFalse(row["verified_dead"]) + self.assertEqual(self.kills, [9001]) + + def test_reap_reuses_classification(self): + # 复用证据：同一数据、同一 checker 下，reap 的判定字段 + # 与直接调 classify_stale_workers 完全一致，不另造判定。 + classified = classify_stale_workers( + self.db_path, stale_seconds=120, local_host="h-local", + pid_is_dead_fn=self._checker, now=self.now) + rows = self._reap() + for worker_id in ("w-kill", "w-eq", "w-foreign", "w-nopid", + "w-alive", "w-fresh"): + self.assertEqual(_by_id(rows, worker_id)["action"], + _by_id(classified, worker_id)["action"], + worker_id) + self.assertEqual(_by_id(rows, worker_id)["reason"], + _by_id(classified, worker_id)["reason"], + worker_id) + + +class DefaultKillDeadPidTestCase(unittest.TestCase): + def test_default_kill_on_truly_dead_pid(self): + # 默认杀路径：子进程退出并 wait 后 pid 确实已死， + # 默认 SIGKILL 遇 ProcessLookupError 当已死处理， + # 复核 pid_is_dead 为 True，outcome=killed。 + proc = subprocess.Popen( + [sys.executable, "-c", "pass"], + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + proc.wait() + dead_pid = proc.pid + tmp = tempfile.TemporaryDirectory() + try: + db_path = os.path.join(tmp.name, "control.db") + register_worker(db_path, "w-dead", host="h-local", + pid=dead_pid) + _set_last_seen(db_path, "w-dead", time.time() - 1000) + rows = reap_stale_workers( + db_path, stale_seconds=120, local_host="h-local") + row = _by_id(rows, "w-dead") + self.assertEqual(row["action"], "kill") + self.assertEqual(row["outcome"], "killed") + self.assertTrue(row["verified_dead"]) + self.assertTrue(row["audit_before"]) + self.assertTrue(row["audit_after"]) + finally: + tmp.cleanup() + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/workers.py b/cloud/workers.py index ccf209e..58c2b3f 100644 --- a/cloud/workers.py +++ b/cloud/workers.py @@ -32,9 +32,21 @@ 前缀枚举/删除容器：失联清理只做判定（见 classify_stale_workers），判定为 kill 的唯一可杀对象是该行 的 pid 本人，且必须是本机 + pid 已死（pid_verified）。 + 执行路径（本轨新增，见 reap_stale_workers）：复用 + classify_stale_workers 的判定，不另造判定函数；只对 + action == "kill" 的行调 kill_fn，且执行时先用同一 checker + 再验一次已死（防判定快照与执行之间 pid 被复用），杀前 + 写审计（审计写失败则不杀，fail-closed），杀后用同一 + checker 复核 pid 已死；异机、无 pid、pid 存活或未知一律 + 只列不杀。执行路径不枚举容器、不按名/前缀匹配、不碰 + 基础设施容器、不删 workers 注册行：杀的是 pid 本人， + 注册行的去留是另一条决策（删行会改 /workers 列表语义）， + 本轨不擅自删。亦不新增 HTTP 路由（会改 v1.1 契约）：调 + 用方（管理员/清扫器）在有权限的上下文里直接调本函数。 """ import os +import signal import socket import time @@ -350,6 +362,145 @@ def classify_stale_workers(db_path, stale_seconds=None, local_host=None, return out +def _default_kill(pid): + """默认真杀：只给该 pid 本人发 SIGKILL，不发进程组。 + + pid 已死时 os.kill 抛 ProcessLookupError，调用方 + （reap_stale_workers）把它当“已死、无需再杀”处理， + 不是失败。绝不把信号发给进程组（会误伤同组无关进程）， + 绝不按名/前缀找进程。 + """ + os.kill(pid, signal.SIGKILL) + + +def _default_audit(db_path, action, detail): + """默认审计写 audit_log：user_id=system、project_id=""。 + + 表不存在时先建（PG 与 SQLite 各一套 DDL，同 + cloud/automations.py 的兜底写法：PG DDL 在 SQLite 下 + 失败即回滚再试 SQLite DDL，PG 适配层下第一条即成功）。 + 写失败抛给调用方：reap 的杀前审计失败时不杀 + （fail-closed），杀后审计失败时只记 audit_after=False。 + """ + conn = _db_connect(str(db_path)) + try: + for ddl in ( + "CREATE TABLE IF NOT EXISTS audit_log(" + "id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, " + "user_id TEXT, project_id TEXT, action TEXT, detail TEXT, " + "created_at DOUBLE PRECISION)", + "CREATE TABLE IF NOT EXISTS audit_log(" + "id INTEGER PRIMARY KEY AUTOINCREMENT, " + "user_id TEXT, project_id TEXT, action TEXT, detail TEXT, " + "created_at REAL)"): + try: + conn.execute(ddl) + conn.execute( + "INSERT INTO audit_log(user_id, project_id, action," + " detail, created_at) VALUES(?, ?, ?, ?, ?)", + ("system", "", action, detail, time.time())) + conn.commit() + return + except Exception: + try: + conn.rollback() + except Exception: + pass + raise RuntimeError("audit_log write failed") + finally: + conn.close() + + +def reap_stale_workers(db_path, stale_seconds=None, local_host=None, + pid_is_dead_fn=None, kill_fn=None, audit_fn=None, + now=None): + """失联 worker 执行路径：复用判定，只杀 pid_verified。 + + 流程（每行）： + - 先调 classify_stale_workers（不另造判定）：action 不是 + kill 的行一律只列不杀，executed=False、outcome="listed"， + 绝不调 kill_fn、不写杀审计。 + - kill 行执行时先用同一 checker 再验一次 pid 已死：若此时 + 已不死（判定后 pid 被复用等），跳过不杀， + outcome="skipped_pid_alive_at_execution"，不调 kill_fn。 + - 杀前写审计：audit_fn 未给时走 _default_audit 写 + audit_log（action=worker_pid_kill）；审计写失败则不杀， + outcome="audit_failed"（fail-closed）。 + - 杀：kill_fn 未给时走 _default_kill（SIGKILL 只给该 pid + 本人）；ProcessLookupError 当已死处理，不是失败；其它 + OSError 记入 kill_error，仍以后续复核定成败。 + - 杀后用同一 checker 复核：已死 -> outcome="killed"； + 还活着 -> outcome="kill_failed_pid_alive"；随后写 + 杀后审计（action=worker_pid_kill_verified，detail 含 + outcome），审计成败只记 audit_after，不改 outcome。 + - 本函数不枚举容器、不按名/前缀匹配、不碰基础设施容器、 + 不删 workers 行；返回每行的判定字段 + executed / + outcome / verified_dead / audit_before / audit_after / + kill_error。verified_dead 仅在杀后复核后才有真值， + 未执行时为 None（不是“还活着”的断言）。 + """ + rows = classify_stale_workers( + db_path, stale_seconds=stale_seconds, local_host=local_host, + pid_is_dead_fn=pid_is_dead_fn, now=now) + checker = pid_is_dead_fn if pid_is_dead_fn is not None else pid_is_dead + killer = kill_fn if kill_fn is not None else _default_kill + results = [] + for row in rows: + result = dict(row) + result["executed"] = False + result["verified_dead"] = None + result["audit_before"] = False + result["audit_after"] = False + result["kill_error"] = None + if row.get("action") != "kill": + result["outcome"] = "listed" + results.append(result) + continue + pid = row.get("pid") + if pid is None or not checker(pid): + result["outcome"] = "skipped_pid_alive_at_execution" + result["verified_dead"] = False + results.append(result) + continue + detail = ("worker_id=%s host=%s pid=%s reason=%s" + % (row.get("worker_id"), row.get("host"), pid, + row.get("reason"))) + try: + if audit_fn is not None: + audit_fn("worker_pid_kill", detail) + else: + _default_audit(db_path, "worker_pid_kill", detail) + result["audit_before"] = True + except Exception: + result["outcome"] = "audit_failed" + results.append(result) + continue + result["executed"] = True + try: + killer(pid) + except ProcessLookupError: + pass + except Exception as exc: + result["kill_error"] = str(exc) + verified = bool(checker(pid)) + result["verified_dead"] = verified + result["outcome"] = ( + "killed" if verified else "kill_failed_pid_alive") + after_detail = detail + " outcome=%s verified_dead=%s" % ( + result["outcome"], verified) + try: + if audit_fn is not None: + audit_fn("worker_pid_kill_verified", after_detail) + else: + _default_audit( + db_path, "worker_pid_kill_verified", after_detail) + result["audit_after"] = True + except Exception: + result["audit_after"] = False + results.append(result) + return results + + def register_worker(db_path, worker_id, executor=None, capacity=1, running=None, host=None, pid=None, address=None): """注册（或刷新）一个 worker：不存在 INSERT，存在 UPDATE。