workers: reap pid_verified stale workers with pre-kill audit and post-kill verification

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()

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。