Python 运维自动化系列（04）：服务半夜挂了怎么办——写一个进程守护脚本

import psutil

# 识别：遍历进程，按命令行里的标记串匹配
for proc in psutil.process_iter(["pid", "name", "cmdline"]):
    cmdline = proc.info["cmdline"] or []
    if any("my-worker" in (arg or "") for arg in cmdline):
        print("找到了:", proc.info["pid"])

# 观测：oneshot() 一次快照取多个指标，减少系统调用
with proc.oneshot():
    proc.cpu_percent(interval=0.2)
    proc.memory_percent()
    proc.create_time()

# 行动：温和终止 vs 强杀
proc.terminate()   # Linux 发 SIGTERM（进程有机会善后）
proc.kill()        # Linux 发 SIGKILL（无法捕获，立即死）

import subprocess
import sys
import time

import psutil

MARKER = "opsdemo-sleepy-worker"  # 用命令行里的标记串识别「我们的」进程


def spawn_worker() -> psutil.Process:
    p = subprocess.Popen(
        [sys.executable, "-c", f"import time; time.sleep(120)  # {MARKER}"],
    )
    return psutil.Process(p.pid)


def find_worker() -> psutil.Process | None:
    for proc in psutil.process_iter(["pid", "name", "cmdline"]):
        try:
            cmdline = proc.info["cmdline"] or []
            if any(MARKER in (arg or "") for arg in cmdline):
                return proc
        except psutil.NoSuchProcess:  # 迭代期间进程可能刚好退出
            continue
    return None


def wait_gone(pid: int, timeout: float = 5.0) -> bool:
    deadline = time.time() + timeout
    while time.time() < deadline:
        if not psutil.pid_exists(pid):
            return True
        time.sleep(0.1)
    return False


if __name__ == "__main__":
    print("[1] 拉起工作进程")
    worker = spawn_worker()
    print(f"    已启动 pid={worker.pid}")

    print("[2] 巡检：找到它并读取指标")
    found = find_worker()
    with found.oneshot():
        print(f"    pid={found.pid} name={found.name()} "
              f"cpu={found.cpu_percent(interval=0.2):.1f}% "
              f"mem={found.memory_percent():.1f}% "
              f"create_time={time.strftime('%H:%M:%S', time.localtime(found.create_time()))}")

    print("[3] 模拟进程被杀（kill -9 级别的不可捕获退出）")
    worker.kill()
    gone = wait_gone(worker.pid)
    print(f"    进程已消失: {gone}")

    print("[4] 守护逻辑发现掉线，自动重启")
    time.sleep(0.3)
    found = find_worker()
    print(f"    巡检发现存活: {found is not None}")
    if found is None:
        worker2 = spawn_worker()
        print(f"    已自动重启, 新 pid={worker2.pid} (旧 pid={worker.pid})")
        found = find_worker()
        print(f"    重启后巡检确认存活: {found is not None} pid={found.pid if found else '?'}")

    print("[5] 收尾：清理演示进程")
    found.terminate()
    print(f"    清理完成, 进程已退出: {wait_gone(found.pid)}")

[1] 拉起工作进程
    已启动 pid=14124
[2] 巡检：找到它并读取指标
    pid=14124 name=python.exe cpu=0.0% mem=0.1% create_time=20:55:52
[3] 模拟进程被杀（kill -9 级别的不可捕获退出）
    进程已消失: True
[4] 守护逻辑发现掉线，自动重启
    巡检发现存活: False
    已自动重启, 新 pid=4704 (旧 pid=14124)
    重启后巡检确认存活: True pid=4704
[5] 收尾：清理演示进程
    清理完成, 进程已退出: True