用 psutil 采集进程 RSS 并生成趋势 CSV 的脚本

线上排障时最常被问的一句话是「这个服务的内存是什么时候涨上去的」。监控平台往往只保留 7～15 天的高精度曲线，而内存泄漏的斜率可能一个月才看得出来；容器重启后历史数据又断了一截。psutil 是 Python 里直接读 /proc 的轻量库，几十行代码就能在自己的机器上落一份「按分钟采样、按天归档」的 RSS 趋势文件，不受监控系统保留策略限制。下面这份脚本解决三件事：稳定地按固定间隔采样指定进程（按 PID 或按进程名匹配）、把每行数据追加写入 CSV 且进程重启后自动重新绑定、以及在数据量变大后做降采样分析。文中涉及的每个参数、每条报错都来自实际跑过的场景。

一、先确认 psutil 能拿到什么：RSS 的几种取值来源

动手之前需要先厘清一个容易踩坑的点：RSS（Resident Set Size）在不同接口下含义不完全一致。

• Process.memory_info().rss：来自 /proc/<pid>/statm 的第二个字段乘以页大小，单位字节。这是最常用的值，包含共享库、共享内存中实际驻留的部分。对单个进程看趋势足够，但不要把多个进程的 RSS 相加当作「总内存占用」，因为共享页会被重复计算。
• Process.memory_full_info().uss：Unique Set Size，只统计该进程独占的物理页。需要内核开启 CONFIG_PROC_PAGE_MONITOR，读取 /proc/<pid>/smaps_rollup（内核 4.14+），开销比 rss 大得多，高频采样时不建议用。
• Process.memory_percent()：默认按物理内存总量计算百分比，也可以传 memtype=psutil.PSEUDOMEM 改为按 swap 计算。

容器环境要特别注意：/proc/meminfo 在多数容器里仍是宿主机的总量，因此 memory_percent() 在容器里算出来的百分比偏小。判断内存水位应该用 rss 绝对值配合 cgroup 的 memory.max 自己算，不要直接信 memory_percent()。

安装与版本核对：

python3 -m venv venv
source venv/bin/activate
pip install "psutil>=5.9"
python3 -c "import psutil, sys; print(psutil.__version__, sys.version)"

memory_full_info() 与 smaps_rollup 相关的能力在 5.9 之后完善，低于这个版本在部分内核上会抛 NotImplementedError 或直接回退读取全量 smaps 导致单次采样耗时几十毫秒，所以建议锁 >=5.9。

二、采样脚本：进程失效、重启、权限不足都要兜住

下面这个脚本是完整可运行的版本。核心设计点：Process 对象在进程退出后会失效，必须捕获 psutil.NoSuchProcess 并重新按名字查找；采样使用「对齐到下一个整分钟」的等待方式而不是裸 sleep(interval)，避免采样点随时间漂移；CSV 采用追加写并带表头判断，方便中断后重启。

#!/usr/bin/env python3
"""采集指定进程 RSS 并追加写入 CSV。用法：
    python3 rss_tracer.py --pid 1234 --out /var/log/rss/nginx.csv
    python3 rss_tracer.py --name gunicorn --out /var/log/rss/gunicorn.csv --interval 60
"""
import argparse
import csv
import os
import signal
import sys
import time
from datetime import datetime, timezone

import psutil

FIELDS = ["ts_iso", "ts_epoch", "pid", "rss_bytes", "vms_bytes",
          "num_threads", "cpu_percent", "create_time"]

running = True


def _stop(signum, frame):
    global running
    running = False
    print(f"received signal {signum}, stopping...", file=sys.stderr)


def find_pids(name, pid=None):
    """返回候选 PID 列表。pid 优先，否则按进程名精确匹配。"""
    if pid:
        return [pid]
    pids = []
    for p in psutil.process_iter(["pid", "name"]):
        try:
            if p.info["name"] == name:
                pids.append(p.info["pid"])
        except (psutil.NoSuchProcess, psutil.AccessDenied):
            continue
    return pids


def sample(proc):
    with proc.oneshot():
        mi = proc.memory_info()
        return {
            "pid": proc.pid,
            "rss_bytes": mi.rss,
            "vms_bytes": mi.vms,
            "num_threads": proc.num_threads(),
            "cpu_percent": proc.cpu_percent(interval=None),
            "create_time": int(proc.create_time()),
        }


def main():
    ap = argparse.ArgumentParser()
    g = ap.add_mutually_exclusive_group(required=True)
    g.add_argument("--pid", type=int, help="监控的 PID")
    g.add_argument("--name", help="按进程名监控，如 gunicorn")
    ap.add_argument("--out", required=True, help="CSV 输出路径")
    ap.add_argument("--interval", type=int, default=60, help="采样间隔秒，默认 60")
    ap.add_argument("--max-pids", type=int, default=1,
                    help="按名字匹配时最多监控前 N 个进程")
    args = ap.parse_args()

    signal.signal(signal.SIGTERM, _stop)
    signal.signal(signal.SIGINT, _stop)

    out_dir = os.path.dirname(os.path.abspath(args.out))
    os.makedirs(out_dir, exist_ok=True)
    need_header = not os.path.exists(args.out) or os.path.getsize(args.out) == 0

    fh = open(args.out, "a", newline="", encoding="utf-8")
    writer = csv.DictWriter(fh, fieldnames=FIELDS)
    if need_header:
        writer.writeheader()
        fh.flush()

    procs = []
    next_tick = time.time()
    while running:
        # 重新解析进程句柄（首次进入或全部失效后）
        if not procs:
            pids = find_pids(args.name, args.pid)[: args.max_pids]
            if not pids:
                next_tick = time.time() + args.interval
                time.sleep(max(0.0, next_tick - time.time()))
                continue
            procs = [psutil.Process(p) for p in pids]
            for p in procs:
                # 首次调用只用于初始化 cpu_percent 的内部基线
                try:
                    p.cpu_percent(interval=None)
                except psutil.Error:
                    pass

        alive = []
        now = datetime.now(timezone.utc).astimezone()
        for p in procs:
            try:
                row = sample(p)
            except psutil.NoSuchProcess:
                print(f"pid disappeared, will re-resolve", file=sys.stderr)
                continue
            except psutil.AccessDenied:
                print(f"access denied on pid {p.pid}, skip", file=sys.stderr)
                continue
            row["ts_iso"] = now.isoformat(timespec="seconds")
            row["ts_epoch"] = int(now.timestamp())
            writer.writerow(row)
            alive.append(p)
        fh.flush()
        procs = alive

        next_tick += args.interval
        sleep_for = next_tick - time.time()
        if sleep_for < 0:
            # 系统挂起或单次采样过慢，跳过错过的周期，避免追赶式风暴
            next_tick = time.time() + args.interval
            sleep_for = args.interval
        time.sleep(sleep_for)

    fh.close()


if __name__ == "__main__":
    main()

几个实现细节值得展开：

• proc.oneshot() 会把多次属性读取合并到一次 /proc/<pid>/* 访问，在采集多个字段时能显著降低开销，否则 memory_info()、num_threads()、cpu_percent() 会分别打开文件。
• create_time 落进 CSV 是为了辨析 PID 复用。Linux 上 PID 会被回收再分配，如果某天趋势图突然出现记忆体从 0 跳到 800MB，先看 create_time 有没有变。
• cpu_percent(interval=None) 的含义是「距上次调用的平均占用」，首次调用必然返回 0.0，所以脚本在绑定进程后立刻预热了一次，首条数据仍然是 0 但第二条起就准了。
• 时间戳落两列：ts_iso 带本地时区偏移方便人看，ts_epoch 是整数秒方便 pandas 直接 pd.to_datetime(unit="s")。

三、让它长期活着：systemd unit 与 cgroup 的两个坑

脚本本身没有守护能力，用 systemd 托管比 nohup 靠谱得多。一个可以直接用的 unit：

[Unit]
Description=RSS tracer for gunicorn
After=network.target

[Service]
Type=simple
User=monitor
WorkingDirectory=/opt/rss-tracer
Environment=PYTHONUNBUFFERED=1
ExecStart=/opt/rss-tracer/venv/bin/python3 /opt/rss-tracer/rss_tracer.py \
    --name gunicorn --out /var/log/rss/gunicorn.csv --interval 60 --max-pids 4
Restart=always
RestartSec=5
# 内存和文件描述符都不大，但把开关显式写上便于排查
LimitNOFILE=4096
MemoryMax=128M

[Install]
WantedBy=multi-user.target

systemctl enable --now rss-tracer 启用后，用 journalctl -u rss-tracer -f 看日志。两个实际踩过的坑：

1. 被监控进程在别的 cgroup 里看不到。如果目标是容器内进程，而 tracer 跑在宿主机上，process_iter(["name"]) 通常仍能找到（同一 PID namespace 下可见），但 psutil.Process.cmdline()、cwd() 这类需要读 /proc/<pid>/ 下符号链接的字段在权限不足时会抛 AccessDenied。用 --pid 配合 docker inspect -f '{{.State.Pid}}' <container> 更稳；更好的做法是把 tracer 直接放进业务容器的 sidecar 或同一个 Pod 里跑。
2. --name 匹配的是 comm 字段，不是完整命令行。Linux 的 comm 最长 15 个字符（TASK_COMM_LEN=16，含结尾空字节），所以 java -jar app.jar 的 comm 是 java，gunicorn 的 worker 进程有时 comm 就是 gunicorn。用 ps -eo pid,comm,args | grep xxx 先确认匹配不到的原因，必要时改写脚本用 cmdline() 做子串匹配。

如果坚持不用 systemd，用 bash 包一层也行，但一定加 PYTHONUNBUFFERED=1，否则日志缓冲会让你以为脚本挂了：

PYTHONUNBUFFERED=1 setsid python3 rss_tracer.py \
  --pid "$(pgrep -f 'gunicorn.*master')" \
  --out /var/log/rss/gunicorn.csv >> /var/log/rss/tracer.log 2>&1 &

四、CSV 变成趋势：降采样与泄漏判定

按 60 秒采样，一个月就是 43200 行，单个进程的文件不过 4MB 左右，Excel 打开会略卡但完全可用。真正的问题不是存储而是噪声：RSS 会因为请求量、GC、页缓存回收而在几十 MB 范围内抖动，直接看原始点根本看不出泄漏。正确的做法是做 1 小时均值，再看跨天的基线是否单调上移。

import pandas as pd

df = pd.read_csv("/var/log/rss/gunicorn.csv")
df["ts"] = pd.to_datetime(df["ts_epoch"], unit="s", utc=True).dt.tz_convert("Asia/Shanghai")
df = df.set_index("ts").sort_index()

# 一小时均值 + 同窗口的峰值，均值看趋势，峰值看抖动
hourly = df["rss_bytes"].resample("1h").agg(["mean", "max", "count"])
hourly["mean_mb"] = hourly["mean"] / 1024 / 1024
hourly["max_mb"] = hourly["max"] / 1024 / 1024

print(hourly.tail(24)[["mean_mb", "max_mb", "count"]].round(1))

# 简化泄漏判定：取每天均值，看最近 7 天是否单调不减
daily = df["rss_bytes"].resample("1D").mean().dropna()
tail = daily.tail(7)
slope = (tail.iloc[-1] - tail.iloc[0]) / 1024 / 1024
print(f"7日净增长: {slope:.1f} MB, 单调性: {tail.is_monotonic_increasing}")

判据经验值：连续 7 天日均 RSS 单调上升，且净增长超过初始基线的 5%，基本可以定性为泄漏；如果只是工作日高、周末低，那是负载驱动的正常波动。另一个必看的指标是 count，它等于该小时内的有效采样条数，正常应该是 60 上下。低于 30 说明脚本被重启过或者机器睡过去了，此时趋势图上那段平坦的线不是内存降了，是没数据。

注意：resample("1h") 生成的时间索引是左闭右开的桶，跨时区转换要在 resample 之前做，否则夏令时地区会出现 23 或 25 条记录的小时。国内无夏令时，问题不大，但代码里显式 dt.tz_convert 是好习惯。

五、常见报错与处理

报错 / 现象	原因	处理
psutil.NoSuchProcess: process no longer exists (pid=1234)	采样间隙进程退出，通常发生在 memory_info() 或 num_threads() 调用时	脚本已捕获并触发重新解析；如果日志里频繁刷这条，说明业务在反复重启，优先查业务
psutil.AccessDenied: (pid=1234)	tracer 的运行用户与目标进程不同，且未开启 ptrace_scope 豁免；读 statm 一般不受限，读 smaps、environ 才受限	用 root 跑或用同 uid；只取 memory_info() 时不要开 memory_full_info()
MemoryMax 触发后 unit 被 OOM kill，journalctl 出现 Failed with result 'oom-kill'	把上限设得太小，pandas 分析脚本又和 tracer 共用 unit	tracer 与离线分析分离，tracer 的 MemoryMax 给 128M 足够，pandas 脚本单独跑
CSV 表头重复出现若干次	多个 tracer 实例并发写同一个文件，各自判断文件为空并写入表头	确认只有一个实例：pgrep -af rss_tracer.py，必要时用 flock 包一层
采样间隔实际为 60.3～61 秒且逐次累积	用了裸 time.sleep(interval)，采样耗时被计入间隔	本脚本用 next_tick += interval 对齐；系统挂起后用 sleep_for < 0 分支重置，不做追赶

小结

这套方案的价值不在代码量，而在把采集口径固定下来：memory_info().rss 作为主指标、create_time 作为 PID 复用的校验位、ts_epoch 作为分析键、小时级 mean/max/count 作为趋势判读入口。落地时按顺序做三件事：先用 ps -eo pid,comm 确认 --name 能匹配上，再用 systemd 托管并给 MemoryMax 兜底，最后把降采样脚本挂到 crontab 每天跑一次输出到日报。三件事都做完，下次有人说「内存好像涨了」，你能在十秒内给出确切的时间和斜率。

FAQ

为什么不直接用 psutil 的 Process.memory_percent()？

因为它按 /proc/meminfo 的 MemTotal 算分母，容器内这个值通常仍是宿主机总量，百分比会被严重低估。要算配额水位，应读 cgroup v2 的 /sys/fs/cgroup/memory.max（v1 为 /sys/fs/cgroup/memory/memory.limit_in_bytes）作为分母。

按进程名匹配到多个 worker，会重复计数吗？

会，但 RSS 本身就不能简单相加。脚本默认 --max-pids 1 只跟踪一个进程；如果确实要看整个 worker 池，建议每个 PID 单独一行，分析时用 groupby("pid") 分开看，而不是先求和。想看整体内存水位，直接读 cgroup 的 memory.current 更准。

脚本能监控 Java 进程的堆内存吗？

不能。RSS 包含堆、元空间、线程栈、JIT 代码缓存和 mmap 的文件页，与 JVM 的 -Xmx 没有线性关系，堆外内存（direct buffer、Metaspace）也混在里面。要看堆内趋势请用 JMX 或 jcmd <pid> GC.heap_info，用本脚本作为辅助看总物理占用。

CSV 文件会无限增长吗？

按 60 秒采集单进程，一年约 52 万行、50MB 上下，问题不大。若采样多进程或间隔更短，建议加 logrotate：用 postrotate 给 tracer 发 SIGHUP 不现实（脚本未实现重开文件），更简单的做法是让 tracer 按天写不同文件（--out 里带上日期并在 systemd ExecStart 前用 shell 展开），再由 logrotate 压缩归档。

采到的 cpu_percent 一直是 0.0 怎么办？

两种情况：一是进程刚绑定，第一次调用本身就是基线，返回 0.0 属正常；二是 interval=None 依赖上一次调用的时间戳，若进程对象被替换（重新解析后新建了 psutil.Process），计数基线也会重置。如果需要精确的 CPU 占用，改用两次读取 proc.cpu_times() 的差值除以实际间隔自行计算。