4、脚本化perf采集：用Python subprocess封装perf命令，实现批量、定时、条件触发的性能数据采集

说实话，很多工程师用perf的时候，都是手动敲命令。一次两次还行，但你要做长期性能监控、批量对比测试，手敲就太累了。我个人的习惯是——能用脚本解决的，绝不手动重复。

这一章，我们就来聊聊怎么用Python的subprocess模块，把perf命令封装成自动化采集工具。说白了，就是让脚本替我们干活：批量跑、定时跑、条件触发跑。

4.1 为什么需要脚本化？

先说说我在项目里遇到的真实场景。有一次，我需要对比三个不同内核版本下，同一个网络应用的性能表现。每个版本要跑10次，每次采集CPU周期、缓存命中率、分支预测失败率三个指标。手动敲？30次命令，每次还要等结果、记录数据……我估计半天就没了。

用脚本封装后，一条命令搞定。睡个午觉回来，数据全在CSV里躺着。

脚本化带来的好处很明显：

• 可重复性：同样的采集参数，每次跑结果一致
• 可编排：批量、定时、条件触发，组合起来用
• 可集成：采集完直接喂给数据分析脚本，形成工具链

4.2 subprocess基础：怎么调perf？

Python调用外部命令，最常用的就是subprocess模块。我个人偏爱subprocess.run()，简单直接。

来看一个最基础的封装：

import subprocess
import json

def run_perf(cmd, event_list, duration=5):
    """
    封装perf stat命令
    :param cmd: 要监控的目标命令
    :param event_list: 性能事件列表，如 ['cycles', 'cache-misses']
    :param duration: 采样时长（秒）
    """
    events = ','.join(event_list)
    perf_cmd = [
        'perf', 'stat',
        '-e', events,
        '--duration', str(duration),
        '--field-separator', ',',
        '-x', ',',   # 输出CSV格式
        cmd
    ]
    
    result = subprocess.run(
        perf_cmd,
        capture_output=True,
        text=True,
        timeout=duration + 10
    )
    
    return result.stdout

# 使用示例
output = run_perf('stress --cpu 4 --timeout 10', 
                  ['cycles', 'instructions', 'cache-misses'],
                  duration=10)
print(output)

嗯，这里要注意一点：timeout参数一定要设。我曾经没设，结果目标进程卡死了，脚本也一直挂在那。加个超时，至少能保证脚本不会无限等待。

4.3 批量采集：一次跑多个场景

批量采集是我用得最多的功能。比如我要对比不同负载下的缓存表现：

import csv
import time

def batch_perf_collect(scenarios, events, output_file='perf_data.csv'):
    """
    批量采集多个场景的性能数据
    :param scenarios: 列表，每个元素是 (场景名, 命令)
    :param events: 性能事件列表
    :param output_file: 输出CSV文件
    """
    with open(output_file, 'w', newline='') as f:
        writer = csv.writer(f)
        # 写表头
        header = ['scenario', 'timestamp'] + events
        writer.writerow(header)
        
        for name, cmd in scenarios:
            print(f'采集场景: {name}')
            output = run_perf(cmd, events, duration=10)
            
            # 解析perf输出（简化版）
            values = parse_perf_output(output, events)
            row = [name, time.time()] + values
            writer.writerow(row)
            
            # 场景之间留点间隔，避免干扰
            time.sleep(2)

# 使用示例
scenarios = [
    ('cpu-bound', 'stress --cpu 4 --timeout 15'),
    ('io-bound', 'stress --io 4 --timeout 15'),
    ('mem-bound', 'stress --vm 2 --vm-bytes 512M --timeout 15')
]

batch_perf_collect(scenarios, ['cycles', 'instructions', 'cache-misses'])

为什么场景之间要sleep 2秒？我在项目中遇到过，连续跑高负载场景，系统温度还没降下来，第二个场景的数据会受第一个场景的余热影响。加个间隔，数据更干净。

4.4 定时采集：像cron一样工作

定时采集适合做长期监控。比如每5分钟采集一次系统级性能数据：

import schedule
import datetime

def scheduled_perf_collect():
    """定时采集系统级性能数据"""
    timestamp = datetime.datetime.now().strftime('%Y%m%d_%H%M%S')
    output_file = f'perf_sys_{timestamp}.csv'
    
    # 采集系统级事件
    events = ['cpu-cycles', 'instructions', 'cache-references', 'cache-misses']
    
    # 用sleep命令模拟一个空监控（perf stat -a 采集整个系统）
    cmd = ['perf', 'stat', '-a', '-e', ','.join(events), 
           '--field-separator', ',', '-x', ',',
           'sleep', '10']
    
    result = subprocess.run(cmd, capture_output=True, text=True)
    
    # 保存结果
    with open(output_file, 'w') as f:
        f.write(result.stdout)
    
    print(f'[{timestamp}] 采集完成: {output_file}')

# 每5分钟执行一次
schedule.every(5).minutes.do(scheduled_perf_collect)

# 也可以定时在整点执行
# schedule.every().hour.at(':00').do(scheduled_perf_collect)

print('定时采集已启动，每5分钟采集一次...')
while True:
    schedule.run_pending()
    time.sleep(1)

4.5 条件触发采集：只在需要的时候抓

条件触发是最智能的方式。比如CPU温度超过80度时，自动触发perf采集，看看是哪个进程在搞鬼。

import psutil
import threading

class ConditionTriggeredPerf:
    """条件触发的perf采集器"""
    
    def __init__(self, condition_func, events, duration=10):
        """
        :param condition_func: 返回bool的函数，True时触发采集
        :param events: 性能事件列表
        :param duration: 每次采集时长
        """
        self.condition_func = condition_func
        self.events = events
        self.duration = duration
        self.monitoring = False
        
    def start_monitor(self, interval=1):
        """启动监控线程"""
        self.monitoring = True
        thread = threading.Thread(target=self._monitor_loop, args=(interval,))
        thread.daemon = True
        thread.start()
        
    def _monitor_loop(self, interval):
        while self.monitoring:
            if self.condition_func():
                print('条件触发！开始采集...')
                self._collect_and_save()
                # 触发后暂停一段时间，避免重复触发
                time.sleep(60)
            time.sleep(interval)
    
    def _collect_and_save(self):
        timestamp = datetime.datetime.now().strftime('%Y%m%d_%H%M%S')
        output_file = f'perf_trigger_{timestamp}.csv'
        
        # 采集指定进程（这里示例采集所有进程）
        cmd = ['perf', 'stat', '-a', 
               '-e', ','.join(self.events),
               '--field-separator', ',', '-x', ',',
               'sleep', str(self.duration)]
        
        result = subprocess.run(cmd, capture_output=True, text=True)
        
        with open(output_file, 'w') as f:
            f.write(result.stdout)
        
        print(f'触发采集完成: {output_file}')

# 使用示例：CPU温度超过80度时触发
def cpu_temp_high():
    # 需要系统支持读取CPU温度
    try:
        with open('/sys/class/thermal/thermal_zone0/temp', 'r') as f:
            temp = int(f.read().strip()) / 1000
        return temp > 80
    except:
        return False

trigger = ConditionTriggeredPerf(
    condition_func=cpu_temp_high,
    events=['cycles', 'instructions', 'cache-misses', 'branch-misses'],
    duration=15
)

trigger.start_monitor(interval=5)
print('条件触发监控已启动，CPU温度超过80度时自动采集...')

# 保持主线程运行
try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    trigger.monitoring = False
    print('监控已停止')

这个功能我是在一次线上问题排查时想到的。当时服务器偶尔出现卡顿，但手动采集时又正常。用条件触发后，终于抓到了内存压力大时的性能数据——原来是某个进程在内存紧张时频繁触发swap。

4.6 数据解析与格式化

采集完数据，还得解析。perf stat的CSV输出格式有点怪，需要处理一下：

def parse_perf_stat_csv(raw_output):
    """
    解析perf stat的CSV输出
    返回事件名到值的字典
    """
    result = {}
    for line in raw_output.strip().split('\n'):
        if not line or line.startswith('#'):
            continue
        parts = line.split(',')
        if len(parts) >= 3:
            try:
                value = float(parts[0])
                event = parts[2].strip()
                result[event] = value
            except ValueError:
                continue
    return result

# 更完整的解析，支持单位转换
def parse_perf_stat_detailed(raw_output):
    """
    详细解析，支持K/M/G单位
    """
    result = {}
    for line in raw_output.strip().split('\n'):
        if not line or line.startswith('#'):
            continue
        parts = line.split(',')
        if len(parts) >= 3:
            try:
                raw_value = parts[0].strip()
                event = parts[2].strip()
                
                # 处理单位
                if raw_value.endswith('K'):
                    value = float(raw_value[:-1]) * 1000
                elif raw_value.endswith('M'):
                    value = float(raw_value[:-1]) * 1000000
                elif raw_value.endswith('G'):
                    value = float(raw_value[:-1]) * 1000000000
                else:
                    value = float(raw_value)
                
                result[event] = value
            except (ValueError, IndexError):
                continue
    return result

4.7 完整工具链示例

最后，我把这些功能整合成一个完整的工具类。你直接拿去用，改改参数就行：

class PerfCollector:
    """完整的perf采集工具类"""
    
    def __init__(self, events=None, duration=10):
        self.events = events or ['cycles', 'instructions', 'cache-misses']
        self.duration = duration
        self.data_dir = 'perf_data'
        os.makedirs(self.data_dir, exist_ok=True)
    
    def collect_single(self, cmd, label=''):
        """单次采集"""
        output = self._run_perf(cmd)
        parsed = parse_perf_stat_detailed(output)
        
        timestamp = datetime.datetime.now().isoformat()
        record = {
            'timestamp': timestamp,
            'label': label,
            'command': cmd,
            **parsed
        }
        return record
    
    def collect_batch(self, scenarios):
        """批量采集"""
        records = []
        for label, cmd in scenarios:
            print(f'采集: {label}')
            record = self.collect_single(cmd, label)
            records.append(record)
            time.sleep(2)
        return records
    
    def collect_triggered(self, condition_func, interval=5):
        """条件触发采集"""
        print('启动条件触发模式...')
        while True:
            if condition_func():
                record = self.collect_single('sleep 10', 'triggered')
                self._save_record(record)
                print(f'触发采集完成: {record["timestamp"]}')
                time.sleep(60)  # 避免重复触发
            time.sleep(interval)
    
    def _run_perf(self, cmd):
        perf_cmd = [
            'perf', 'stat',
            '-e', ','.join(self.events),
            '--field-separator', ',',
            '-x', ',',
            cmd
        ]
        result = subprocess.run(
            perf_cmd,
            capture_output=True,
            text=True,
            timeout=self.duration + 10
        )
        return result.stdout
    
    def _save_record(self, record):
        filename = f'{self.data_dir}/perf_{record["timestamp"][:10]}.json'
        with open(filename, 'a') as f:
            f.write(json.dumps(record) + '\n')

# 使用示例
collector = PerfCollector(
    events=['cycles', 'instructions', 'cache-misses', 'branch-misses'],
    duration=10
)

# 批量采集
scenarios = [
    ('轻负载', 'stress --cpu 2 --timeout 15'),
    ('中负载', 'stress --cpu 4 --timeout 15'),
    ('重负载', 'stress --cpu 8 --timeout 15')
]
results = collector.collect_batch(scenarios)
print(f'采集完成，共{len(results)}条记录')

这个工具类我用了很久，基本覆盖了日常80%的perf采集需求。你拿到项目里，改改事件列表和采集时长，就能直接用。

嗯，脚本化perf采集的核心思路就这些。说白了就是：把perf当成一个可编程的传感器，用Python给它加上大脑。批量、定时、条件触发，三种模式组合起来，基本能应对所有性能采集场景。

下一章，我们会聊聊怎么把这些采集到的数据，用Python做自动化分析。到时候你会发现，采集和分析连起来，才是真正的工具链。