如何实现触发式环形缓冲Packet capture抓包并保存pcap文件

别自己在Python层解析tcpdump的CLI输出做环形缓冲，完全走了弯路，两个核心问题直接用抓包工具原生能力配合轻量进程控制就能解决，全程不需要手动处理pcap格式、不需要自己维护内存缓冲，抓包进程常驻不退出，支持无限次故障触发。

问题解决路径 #

• 针对「仅保留最近10秒流量」的需求：不要手动读CLI流做队列缓冲，直接用抓包工具自带的环形缓冲能力，把缓冲放在/dev/shm内存文件系统上，配置自动覆盖10秒前的旧数据，没有额外内存拷贝开销，不会出现缓冲延迟导致的丢包。
• 针对「输出合规pcap/pcapng文件」的需求：绝对不要捕获tcpdump的人类可读文本输出生成文件，pcap/pcapng是带固定二进制头、包记录头的结构化格式，手动拼接几乎不可能兼容所有分析工具，直接让抓包进程原生输出二进制pcapng格式即可，落盘的文件可以直接被Wireshark、tshark、tcpdump读取。
• 针对「触发保存不终止抓包进程」的需求：不要用网上流传的kill抓包进程轮转文件的方案，进程重启的空窗期很容易漏掉故障瞬间流量，改用用户自定义信号通知常驻抓包进程标记保留缓冲即可，抓包全程不中断。

可直接落地的实现代码 #

依赖安装：先安装wireshark-common组件，自带dumpcap（轻量后台抓包工具，比tcpdump更适合长期运行）和mergecap（pcap文件合并工具），嵌入式IoT网关直接找对应架构的ipk/opk安装包即可，常驻内存占用不到10M。

import subprocess
import signal
import time
import glob
import os
from pathlib import Path

# 配置参数
CAPTURE_NIC = "br-lan"  # 替换成IoT场景对应的实际业务网卡，比如接终端的wlan0/eth1
PRE_CAP_SEC = 10  # 故障前保留流量时长
POST_CAP_SEC = 10 # 故障后抓包时长
SAVE_DIR = Path("./fault_caps")
SAVE_DIR.mkdir(exist_ok=True)
RING_FILE_PREFIX = "/dev/shm/iot_cap_ring" # 内存文件系统路径，无磁盘IO损耗

class PersistentCap:
    def __init__(self):
        self.is_saving = False
        self.cap_proc = None

    def start(self):
        # 启动常驻环形缓冲抓包
        cap_cmd = [
            "dumpcap",
            "-i", CAPTURE_NIC,
            "-w", RING_FILE_PREFIX,
            "--ring-buffer", f"duration:{PRE_CAP_SEC}",
            "--ring-buffer", "files:2", # 固定保留2个10秒的块，旧块自动覆盖
            "-n", # 不做IP/域名解析，减少性能损耗避免IoT高流量场景丢包
            "-s", "0", # 抓完整报文不截断
            "-q" # 静默运行不输出日志
        ]
        self.cap_proc = subprocess.Popen(
            cap_cmd,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL
        )
        # 注册SIGUSR1为故障触发信号，应用检测到断连时给这个进程发SIGUSR1即可
        signal.signal(signal.SIGUSR1, self._save_fault_cap)

    def _save_fault_cap(self, *args):
        if self.is_saving:
            return
        self.is_saving = True
        # 等待故障后10秒流量抓完
        time.sleep(POST_CAP_SEC)
        # 合并环形缓冲里的两个10秒块，生成20秒的完整故障包
        ts = int(time.time())
        target_file = SAVE_DIR / f"net_fault_{ts}.pcapng"
        ring_chunks = glob.glob(f"{RING_FILE_PREFIX}_*")
        merge_cmd = [
            "mergecap",
            "-w", str(target_file),
            "-F", "pcapng"
        ] + ring_chunks
        subprocess.run(merge_cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        self.is_saving = False

if __name__ == "__main__":
    cap_service = PersistentCap()
    cap_service.start()
    print(f"抓包服务已启动，检测到断连时发送SIGUSR1到{os.getpid()}即可保存故障包")
    while True:
        time.sleep(1)

踩坑提示：不要用tcpdump的-G参数做时间轮转，tcpdump对环形缓冲的信号支持很差，触发保存时很容易出现文件截断。如果一定要用tcpdump，需要升级到4.99以上版本，配合--time-stamp-precision nano参数减少时间误差，但稳定性还是不如dumpcap。
应用侧触发保存只需要调用os.kill(抓包进程pid, signal.SIGUSR1)即可，触发后抓包进程不会退出，自动回到等待下一次触发的状态。

内容的提问来源于stack exchange，提问作者Yaron