用 Python 写一个“可复查”的小脚本:从日志里只捞出真正需要处理的那几行
很多线上问题并不是难在定位,而是难在复查:当时是谁先看到报错、同一条错误出现了多少次、前后还有没有关联的警告,翻一遍上万行的日志往往就把关键信息淹没了。直接用 grep 捞关键字当然快,但每次换一个人、换一个时间段,结果就可能不一样,事后也很难说清楚“为什么只处理了这几条”。所以我更愿意把筛选逻辑写成一个小脚本:输入是原始日志,输出是一份可以复查的清单,而不是一堆随手复制的片段。
先定三条筛选规则。第一,只看需要人工干预的级别,例如 ERROR 和 CRITICAL,INFO 和 DEBUG 先排除,但要保留它们在结果里的计数,方便判断是不是偶发。第二,去掉重复:同一条错误在短时间内刷屏一百次,处理清单里只出现一次,同时注明它出现了多少次、第一次和最后一次的时间。第三,保留上下文:每条入选记录至少带上时间、模块、请求标识和原始行号,这样别人拿到清单不用重新翻日志,也能定位到源头。
实现上不用写得很复杂。按行读文件,避免一次性把大文件塞进内存;用正则先拆出时间和级别,再用一个字典按“模块加错误摘要”做聚合。摘要不要用整行原文,因为同一类错误里订单号、时间戳每次都不同,直接用整行去重会失效;可以先把数字和标识符替换成占位符,再聚合。这样做的好处是,清单长度由错误种类决定,而不是由日志总行数决定。
import argparse, re
from collections import Counter, defaultdict
LINE = re.compile(r'^(?P<time>\S+ \S+) \[(?P<level>\w+)\] (?P<module>\S+) (?P<msg>.*)$')
def summarize(msg):
return re.sub(r'\d+', '#', msg)
def main():
parser = argparse.ArgumentParser()
parser.add_argument('logfile')
parser.add_argument('--out', default='report.md')
args = parser.parse_args()
groups = defaultdict(list)
levels = Counter()
with open(args.logfile, encoding='utf-8', errors='ignore') as f:
for no, line in enumerate(f, 1):
m = LINE.match(line.strip())
if not m:
continue
levels[m.group('level')] += 1
if m.group('level') in ('ERROR', 'CRITICAL'):
key = (m.group('module'), summarize(m.group('msg')))
groups[key].append((no, m.group('time')))
with open(args.out, 'w', encoding='utf-8') as out:
out.write('# 日志处理清单\n\n')
for (module, msg), rows in sorted(groups.items(), key=lambda x: -len(x[1])):
out.write(f"- {module} | 出现 {len(rows)} 次 | 首次 {rows[0][1]} | 末次 {rows[-1][1]} | {msg} | 行号 {rows[0][0]}\n")
out.write(f"\n总级别统计:{dict(levels)}\n")
if __name__ == '__main__':
main()
输出建议分成两部分:前面是处理清单,每条写清次数、首次时间、末次时间和一句话判断;后面是统计,写清总行数、过滤后行数和各类错误占比。脚本最后再打印一行校验信息,例如输入文件路径、输出文件路径和生成时间,让这份清单本身也能被复查。跑之前先在小文件上试一次,确认聚合没有把不同错误合并,也没有把同一错误拆成多条,再用到真实日志上。
最后提醒两点。第一,日志里可能有手机号、订单号或内部地址,输出前先脱敏,只保留排查需要的标识。第二,这个脚本不是监控告警的替代品,它解决的是事后复查和交接问题;真正需要实时发现问题,还是要靠告警和指标。把“筛选规则写下来、结果能复现”当成习惯,日志才不会越翻越乱。
- 点赞
- 收藏
- 关注作者
评论(0)