02_日志分析：从C程序日志提取统计

日志分析：从 C 程序日志提取统计 (Log Analysis)

---

章节概述

你写的 C 服务器跑了三天，生成了 800MB 的日志文件。现在需要从中统计错误频率、请求耗时分布、高峰时段。手动 grep 不够用了——本章教你用 Python 在 C 程序日志上完成”读取→正则提取→统计→输出报告”的完整工作流。先从 python -c 临时分析起步，再演进到可复用的 .py 脚本。

核心理念：C 语言写高性能服务器，Python 写日志分析——这是典型的”C 做运算，Python 做分析”分工模式。C 程序只需按固定格式输出日志（CSV、JSON Lines 或自定义格式），Python 就能在事后或实时读取并生成任何维度的统计报告。你不需要在 C 里内嵌一个统计引擎。

---

第一节：C 程序日志格式设计

---

日志分析的第一步是为 C 程序设计机器可解析的日志格式。以下是三种从简单到复杂的实用方案。

方案一：分隔符文本（推荐入门）

// 在你的 C 程序中
printf("[%s] %s | %s | %dms | %s\n",
 timestamp, level, endpoint, latency, status);

输出示例：

[2025-03-01 14:23:05] INFO | /api/login | 12ms | 200
[2025-03-01 14:23:06] ERROR | /api/upload | 5234ms | 500
[2025-03-01 14:23:07] INFO | /api/login | 8ms | 200

用 | 分隔字段，Python 一行即可解析：l.strip().split(' | ')。

方案二：JSON Lines（推荐生产环境）

// printf 输出单行 JSON（使用 cJSON 库或手写）
printf("{\"ts\":\"%s\",\"lvl\":\"%s\",\"ep\":\"%s\",\"lat\":%d,\"st\":%d}\n",
 timestamp, level, endpoint, latency, status);

输出示例（每行一个 JSON 对象）：

{"ts":"2025-03-01 14:23:05","lvl":"INFO","ep":"/api/login","lat":12,"st":200}
{"ts":"2025-03-01 14:23:06","lvl":"ERROR","ep":"/api/upload","lat":5234,"st":500}

Python 读取：json.loads(line) —— 比手写分隔符解析更可靠，且支持嵌套结构。

方案三：syslog 标准格式（接入现有监控体系）

#include <syslog.h>
syslog(LOG_INFO, "/api/login latency=%dms status=%d", latency, status);

Python 可读取系统 syslog 文件或通过 journald 接口获取。

无论选哪种方案，核心原则是：日志格式要对机器友好。人类可读性是加分项，但解析便利性才是生产力的关键。

---

第二节：快速分析——python -c 临时统计

---

拿到日志文件后，第一件事不应该是打开编辑器写脚本，而是用 python -c 快速探查。

示例日志（server.log）：

[2025-03-01 14:23:05] INFO | /api/login | 12ms | 200
[2025-03-01 14:23:06] ERROR | /api/upload | 5234ms | 500
[2025-03-01 14:23:07] INFO | /api/login | 8ms | 200
[2025-03-01 14:23:08] WARN | /api/search | 156ms | 200
[2025-03-01 14:23:09] ERROR | /api/login | 1023ms | 503
[2025-03-01 14:23:10] INFO | /api/upload | 45ms | 200
[2025-03-01 14:23:11] INFO | /api/login | 11ms | 200
[2025-03-01 14:23:12] INFO | /api/search | 89ms | 200
[2025-03-01 14:23:13] ERROR | /api/login | 2001ms | 500

探查 1：各日志级别数量统计

python -c "
import sys, collections
c = collections.Counter()
for line in open('server.log'):
 if ' | ' in line:
 c[line.split('|')[0].split()[-1]] += 1
for level, count in c.most_common():
 print(f'{level:>6}: {count}')
"

输出：

 INFO: 5
 ERROR: 3
 WARN: 1

collections.Counter 一行搞定分类计数——在 C 中这需要哈希表 + 排序，在 bash 中需要 awk '{print $2}' | sort | uniq -c。

探查 2：各接口的平均耗时

python -c "
import sys, collections
from statistics import mean
 
stats = collections.defaultdict(list)
 
for line in open('server.log'):
 parts = [p.strip() for p in line.split('|') if p.strip()]
 if len(parts) >= 3:
 endpoint = parts[1]
 latency = int(parts[2].replace('ms', ''))
 stats[endpoint].append(latency)
 
for ep, lats in sorted(stats.items()):
 print(f'{ep:>16}: 平均 {mean(lats):.0f}ms (共{len(lats)}次)')
"

输出：

 /api/login: 平均 348ms (共5次)
 /api/search: 平均 123ms (共2次)
 /api/upload: 平均 2640ms (共2次)

statistics.mean 来自标准库，collections.defaultdict(list) 免去手动初始化的麻烦。

探查 3：按小时统计请求量分布

python -c "
import sys, collections
c = collections.Counter()
for line in open('server.log'):
 if line[0] == '[':
 hour = line[12:14] # 时间戳的"时"字段
 c[hour] += 1
for h in sorted(c):
 print(f'{h}:00 → {c[h]} 条')
"

---

第三节：完整分析脚本——从临时到可复用

---

python -c 适合 5 分钟以内的临时探查。一旦分析需求复杂化，或者需要每天跑一次，就应该写 .py 脚本。以下是一个完整的日志分析脚本。

脚本：log_analyzer.py

#!/usr/bin/env python3
"""C 程序日志分析工具 —— 读取分隔符格式日志，输出统计报告"""
import sys
import re
import json
import csv
from collections import Counter, defaultdict
from statistics import mean, median, stdev
from datetime import datetime
 
def parse_line(line):
 """解析一行 [时间] 级别 | 接口 | 耗时 | 状态码"""
 pattern = r'\[(.+?)\]\s+(\w+)\s+\|\s+(\S+)\s+\|\s+(\d+)ms\s+\|\s+(\d+)'
 m = re.match(pattern, line)
 if not m:
 return None
 return {
 'time': datetime.strptime(m.group(1), '%Y-%m-%d %H:%M:%S'),
 'level': m.group(2),
 'endpoint': m.group(3),
 'latency': int(m.group(4)),
 'status': int(m.group(5)),
 }
 
def analyze(logfile, report_format='text'):
 entries = []
 for line in open(logfile, encoding='utf-8', errors='ignore'):
 entry = parse_line(line.strip())
 if entry:
 entries.append(entry)
 
 if not entries:
 print("未解析到任何日志条目", file=sys.stderr)
 sys.exit(1)
 
 # 统计 1：各级别计数
 level_counts = Counter(e['level'] for e in entries)
 
 # 统计 2：各接口耗时统计
 ep_stats = defaultdict(list)
 for e in entries:
 ep_stats[e['endpoint']].append(e['latency'])
 
 # 统计 3：各状态码计数
 status_counts = Counter(e['status'] for e in entries)
 
 # 统计 4：每小时请求量（高峰时段检测）
 hour_counts = Counter(e['time'].hour for e in entries)
 
 # 统计 5：错误（status >= 400）的详细信息
 errors = [e for e in entries if e['status'] >= 400]
 
 report = {
 'total_entries': len(entries),
 'time_range': f"{min(e['time'] for e in entries)} ~ {max(e['time'] for e in entries)}",
 'level_distribution': dict(level_counts),
 'endpoint_stats': {},
 'status_distribution': dict(status_counts),
 'hourly_distribution': dict(sorted(hour_counts.items())),
 'error_count': len(errors),
 'error_rate': f"{len(errors)/len(entries)*100:.2f}%",
 }
 
 for ep, lats in ep_stats.items():
 report['endpoint_stats'][ep] = {
 'count': len(lats),
 'avg_ms': round(mean(lats), 1),
 'median_ms': round(median(lats), 1),
 'p99_ms': round(sorted(lats)[int(len(lats)*0.99)] if len(lats) >= 100 else max(lats), 1),
 'min_ms': min(lats),
 'max_ms': max(lats),
 }
 
 if report_format == 'json':
 print(json.dumps(report, indent=2, ensure_ascii=False, default=str))
 elif report_format == 'csv':
 writer = csv.writer(sys.stdout)
 writer.writerow(['endpoint', 'count', 'avg_ms', 'max_ms'])
 for ep, stats in report['endpoint_stats'].items():
 writer.writerow([ep, stats['count'], stats['avg_ms'], stats['max_ms']])
 else:
 # 文本报告
 print("=" * 60)
 print(" C 程序日志分析报告")
 print("=" * 60)
 print(f"总日志量: {report['total_entries']} 条")
 print(f"时间范围: {report['time_range']}")
 print(f"错误率: {report['error_rate']}")
 print()
 print("--- 日志级别分布 ---")
 for level, count in level_counts.most_common():
 print(f" {level:>6}: {count}")
 print()
 print("--- 接口耗时统计 ---")
 print(f" {'接口':<18} {'次数':>6} {'平均':>8} {'中位':>8} {'最小':>6} {'最大':>6}")
 for ep, stats in sorted(report['endpoint_stats'].items()):
 s = stats
 print(f" {ep:<18} {s['count']:>6} {s['avg_ms']:>7.0f}ms {s['median_ms']:>7.0f}ms {s['min_ms']:>5}ms {s['max_ms']:>5}ms")
 print()
 print("--- 小时分布 ---")
 for h, c in sorted(hour_counts.items()):
 bar = '█' * (c * 2)
 print(f" {h:>02d}:00 {c:>4} {bar}")
 
if __name__ == '__main__':
 import argparse
 parser = argparse.ArgumentParser(description='C 程序日志分析工具')
 parser.add_argument('logfile', help='日志文件路径')
 parser.add_argument('--format', choices=['text','json','csv'], default='text',
 help='输出格式 (default: text)')
 args = parser.parse_args()
 analyze(args.logfile, args.format)

使用方式：

# 标准文本报告
python log_analyzer.py server.log
 
# JSON 输出（供其他程序消费）
python log_analyzer.py server.log --format json > report.json
 
# CSV 输出（导入 Excel）
python log_analyzer.py server.log --format csv > report.csv

这个脚本的设计模式：parse_line() 负责解析单行（正则提取）→ analyze() 负责统计（Counter/mean/median）→ 主函数负责输出（text/json/csv 多格式）。argparse 来自标准库，无需第三方依赖。

---

第四节：正则表达式深度提取

---

真实日志比示例复杂得多。以下是几个常见场景的正则表达式写法。

场景 1：提取 URL 中的查询参数

import re
 
line = '[2025-03-01 14:23:05] INFO | /api/search?q=python&page=3 | 45ms | 200'
endpoint = line.split('|')[1].strip()
 
# 提取搜索关键词
q_match = re.search(r'[?&]q=([^&]+)', endpoint)
if q_match:
 print(f'搜索关键词: {q_match.group(1)}')

场景 2：从堆栈跟踪中提取文件名和行号

C 程序崩溃时可能会输出如下格式：

#0 0x00007f... in do_work () at src/net/server.c:342
#1 0x00007f... in main () at src/main.c:56

用 python -c 提取所有源文件位置：

python -c "
import re, sys
for line in sys.stdin:
 m = re.search(r'at (\S+):(\d+)', line)
 if m:
 print(f'{m.group(1)}:{m.group(2)}')
" < crash.log

场景 3：多行日志记录合并

有些日志记录跨多行（如异常堆栈），需要将属于同一条记录的行合并：

import re
 
def parse_multiline_log(filepath):
 """将以时间戳开头的行作为新记录开始的标志"""
 entries = []
 current = []
 ts_pattern = re.compile(r'^\[\d{4}-\d{2}-\d{2}')
 for line in open(filepath):
 line = line.rstrip('\n')
 if ts_pattern.match(line):
 if current:
 entries.append(''.join(current))
 current = [line]
 else:
 current.append(line)
 if current:
 entries.append(''.join(current))
 return entries

---

第五节：实时日志分析与管道

---

日志不一定要分析静态文件——tail -f + Python 管道可以实现实时监控。

实时错误率监控（滑动窗口）

# 监听实时日志，每 10 条输出一次错误率
tail -f server.log | python -c "
import sys, collections
 
window = collections.deque(maxlen=100)
count = 0
 
for line in sys.stdin:
 is_error = 'ERROR' in line
 window.append(is_error)
 count += 1
 if count % 10 == 0:
 rate = sum(window) / len(window) * 100
 print(f'[滑动窗口={len(window)}] 错误率: {rate:.1f}%')
"

collections.deque(maxlen=100) 自动维护固定长度的滑动窗口——新元素入队时，最老的自动出队。在 C 中实现同样的效果需要自己维护环形缓冲区。

触发式告警

tail -f server.log | python -c "
import sys
 
for line in sys.stdin:
 if 'ERROR' in line and '500' in line:
 print(f' 告警：5xx 错误 —— {line.strip()[:120]}')
 if 'lat' in line:
 import re
 m = re.search(r'(\d+)ms', line)
 if m and int(m.group(1)) > 5000:
 print(f' 慢请求：{m.group(0)} —— {line.strip()[:120]}')
"

---

练习

以下题目用于验证本章所学内容：

题号	题目	链接	涉及知识点
14	最长公共前缀	https://leetcode.cn/problems/longest-common-prefix/	字符串处理、模式匹配
20	有效的括号	https://leetcode.cn/problems/valid-parentheses/	栈、字符串解析
387	字符串中的第一个唯一字符	https://leetcode.cn/problems/first-unique-character-in-a-string/	字符串分析与计数