从不一致正则模式提取服务器日志字段的问题排查

原始日志与需求 #

现有一组格式不一致的服务器事件日志：

logs = ['event on SRVDC1.acme.loc created medium alert TS Gateway login failure.',
        'event with source 10.10.13.1 by john.smith on SRVDC1.acme.loc created medium alert More Than 3 Failed Login Attempts Within 1 Hour .',
        'authentication event with process lsass.exe, source 192.168.254.13:63000, by thomas on SRVDC1.acme.loc created medium alert Logon Failure - Unknown user or bad password.',
        'iam event by ANONYMOUS LOGON on SRVDC2.acme.loc created medium alert Computer account added/changed/deleted..']

需要提取以下字段：

• 必现字段：告警名称、域名
• 非必现字段：源IP、操作者（缺失时用-填充）

现有代码与报错 #

当前代码尝试提取字段，但运行时抛出TypeError: argument of type 'NoneType' is not iterable：

import re

host_list = [re.search('(?<=on).*?(?=created)',log).group() for log in logs]
alert_list = [re.search('(?<=alert).*?(?=\.)',log).group() for log in logs]
source_ip_list = [' - ' if '\d+\.\d+\.\d+\.\d+\.' not in re.search('\d+\.\d+\.\d+\.\d+\.',log) else re.search('(?<=source).*?(?=\d+\.\d+\.\d+\.\d+\.)',log).group() for log in logs]
actor_list = [' - ' if 'by' not in re.search('by',log) else re.search('(?<=by).*?(?=on)',log).group() for log in logs]

print(host_list)
print(alert_list)
print(source_ip_list)
print(actor_list)

错误原因 #

报错核心是：当re.search()未找到匹配内容时会返回None，此时直接对None执行in判断或调用.group()方法会触发类型错误。比如：

1. 无IP的日志中，re.search('\d+\.\d+\.\d+\.\d+\.',log)返回None，执行'\d+\.\d+\.\d+\.\d+\.' not in None会报错
2. 无操作者的日志中，re.search('by',log)返回None，执行'by' not in None同样报错

修正后的代码 #

调整逻辑，先判断匹配结果是否存在，再提取内容或填充默认值：

import re

host_list = [re.search('(?<=on).*?(?=created)', log).group().strip() for log in logs]
alert_list = [re.search('(?<=alert).*?(?=\.)', log).group().strip() for log in logs]

# 提取源IP：匹配IP格式，无则返回'-'
source_ip_list = []
for log in logs:
    ip_match = re.search(r'\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b', log)
    source_ip_list.append(ip_match.group() if ip_match else '-')

# 提取操作者：匹配'by'到'on'之间的内容，无则返回'-'
actor_list = []
for log in logs:
    actor_match = re.search(r'(?<=by).*?(?=on)', log)
    actor_list.append(actor_match.group().strip() if actor_match else '-')

# 打印结果
print(host_list)
print(alert_list)
print(source_ip_list)
print(actor_list)

验证输出 #

运行修正后的代码，输出与期望一致：

['SRVDC1.acme.loc', 'SRVDC1.acme.loc', 'SRVDC1.acme.loc', 'SRVDC2.acme.loc']
['TS Gateway login failure', 'More Than 3 Failed Login Attempts Within 1 Hour', 'Logon Failure - Unknown user or bad password', 'Computer account added/changed/deleted']
['-', '10.10.13.1', '192.168.254.13', '-']
['-', 'john.smith', 'thomas', 'ANONYMOUS LOGON']

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