跳到主要内容

进程管理与服务检查

除了「执行一条命令」，运维还要看进程是否在跑、资源是否异常、服务是否健康。Python 可用 psutil（需安装）或解析 /proc；服务状态优先 systemctl。

安装 psutil​

cd ~/python-for-ops
uv add psutil

进程枚举与过滤​

#!/usr/bin/env python3
"""按名称查找进程、检查 CPU/内存占用。"""
import psutil

def find_processes(name_substr: str) -> list[psutil.Process]:
    """name_substr 匹配进程名或 cmdline，如 'nginx'、'java'。"""
    matched = []
    for proc in psutil.process_iter(["pid", "name", "cmdline", "cpu_percent", "memory_info"]):
        try:
            info = proc.info
            cmdline = " ".join(info["cmdline"] or [])
            if name_substr in (info["name"] or "") or name_substr in cmdline:
                matched.append(proc)
        except (psutil.NoSuchProcess, psutil.AccessDenied):
            continue
    return matched

def report_process(name: str) -> None:
    procs = find_processes(name)
    if not procs:
        print(f"未找到进程: {name}")
        return
    for p in procs:
        with p.oneshot():  # 批量读，减少 syscall
            mem_mb = p.memory_info().rss / 1024 / 1024
            print(f"PID={p.pid} name={p.name()} mem={mem_mb:.1f}MB cpu={p.cpu_percent():.1f}%")

if __name__ == "__main__":
    report_process("sshd")

systemd 服务检查​

import subprocess

SYSTEMD_UNITS = ["nginx", "docker", "kubelet"]

def check_services(units: list[str]) -> dict[str, str]:
    """返回 unit -> ActiveState（active/inactive/failed）。"""
    status = {}
    for unit in units:
        r = subprocess.run(
            ["systemctl", "is-active", unit],
            capture_output=True,
            text=True,
            timeout=10,
        )
        # is-active：active=0, inactive/failed=非0
        state = r.stdout.strip() if r.returncode == 0 else r.stdout.strip() or "unknown"
        if r.returncode != 0 and not state:
            state = r.stderr.strip() or "inactive"
        status[unit] = state
    return status

def health_summary() -> bool:
    results = check_services(SYSTEMD_UNITS)
    all_ok = True
    for unit, state in results.items():
        ok = state == "active"
        all_ok &= ok
        flag = "OK" if ok else "FAIL"
        print(f"[{flag}] {unit}: {state}")
    return all_ok

启动子进程并等待​

import os
import signal
import subprocess
import time

def run_background(cmd: list[str], log_file: str) -> subprocess.Popen:
    """启动长期运行的子进程（如本地 mock 服务），日志写入文件。"""
    f = open(log_file, "a", encoding="utf-8")
    return subprocess.Popen(
        cmd,
        stdout=f,
        stderr=subprocess.STDOUT,
        start_new_session=True,  # 独立进程组，便于整组 kill
    )

def stop_process_group(proc: subprocess.Popen, grace_sec: int = 5) -> None:
    """先 SIGTERM，超时再 SIGKILL。"""
    if proc.poll() is not None:
        return
    try:
        os.killpg(proc.pid, signal.SIGTERM)
    except ProcessLookupError:
        return
    deadline = time.time() + grace_sec
    while time.time() < deadline:
        if proc.poll() is not None:
            return
        time.sleep(0.2)
    os.killpg(proc.pid, signal.SIGKILL)
备注

生产环境长期服务仍应交给 systemd/supervisor，脚本只做巡检与临时拉起。

无 psutil 时的轻量方案​

import subprocess
from pathlib import Path

def pid_exists(pid: int) -> bool:
    return subprocess.run(["kill", "-0", str(pid)], capture_output=True).returncode == 0

def read_proc_name(pid: int) -> str | None:
    status = Path(f"/proc/{pid}/status")
    if not status.exists():
        return None
    for line in status.read_text().splitlines():
        if line.startswith("Name:"):
            return line.split("\t", 1)[1].strip()
    return None

小结​

  • 服务是否该由 systemd 管：用 systemctl is-active / is-enabled。
  • 进程资源：psutil 跨平台；Linux 也可读 /proc。
  • 脚本拉起的子进程：记得 timeout、日志、优雅退出，避免僵尸进程与孤儿进程占资源。