Rsyslog的omprog模块可将日志实时传给Python脚本处理，需v8.19.0+版本、正确配置模块加载、行协议解析、长运行脚本及可靠队列参数。
可以通过 Rsyslog 的 omprog 模块将日志实时传递给 Python 脚本处理，无需中间文件或队列，实现低延迟报警触发。关键在于正确配置模块行为、输入格式、进程生命周期及错误处理。
确保 Rsyslog 版本支持 omprog 并启用模块
Rsyslog v8.19.0+ 原生支持 omprog，旧版本需编译时开启 --enable-omprog。运行以下命令确认：
rsyslogd -v | grep -i "omprog"
若未启用，需升级或重编译。启用后，在 /etc/rsyslog.conf 或 /etc/rsyslog.d/ 下的配置文件中加载模块：
module(load="omprog")
编写符合 omprog 协议的 Python 脚本
omprog 以行协议通信：每行一条 syslog 消息（含 PRI、TIMESTAMP、HOSTNAME、TAG、CONTENT），脚本需从 stdin 逐行读取，处理后立即输出（不缓存）、不退出（保持长运行）。示例脚本 /opt/bin/alert_handler.py：
立即学习“Python免费学习笔记（深入）”；
#!/usr/bin/env python3
import sys
import json
import re
<p>def parse_syslog_line(line):</p><h1>简单解析：PRI TIMESTAMP HOST TAG CONTENT（rsyslog 默认 template）</h1><pre class="brush:php;toolbar:false;">parts = line.strip().split(None, 4)
if len(parts) < 5:
return None
pri, timestamp, hostname, tag, content = parts
return {
"pri": pri,
"timestamp": timestamp,
"hostname": hostname,
"tag": tag.rstrip(':'),
"content": content.strip()
}def should_alert(log_dict):
示例：匹配 ERROR、CRITICAL 或特定关键词
return ("ERROR" in log_dict["content"] or
"CRITICAL" in log_dict["content"] or
re.search(r"fail|down|timeout", log_dict["content"], re.I))def send_alert(log_dict):
此处替换为实际报警逻辑：钉钉/企微/Webhook/邮件等
print(f"[ALERT] {log_dict['hostname']} {log_dict['tag']}: {log_dict['content'][:100]}")
# requests.post(...) 或 subprocess.call(...) 发送通知if name == "main":
关键：禁用 stdout 缓存，保证实时传递
sys.stdout.flush()
while True:
try:
line = sys.stdin.readline()
if not line:
break
log = parse_syslog_line(line)
if log and should_alert(log):
send_alert(log)
except KeyboardInterrupt:
break
except Exception as e:
# 记录错误到系统日志，避免脚本崩溃导致日志丢失
sys.stderr.write(f"ERR: {e}\n")
sys.stderr.flush()赋予执行权限：
chmod +x /opt/bin/alert_handler.py
在 rsyslog 中配置 omprog 规则并指定输入模板
默认日志格式可能缺失字段，建议定义清晰的 template 供 omprog 使用。在 /etc/rsyslog.d/50-alert.conf 中添加：
# 定义简洁、可解析的日志模板（空格分隔，无多余符号）
template(name="OmprogFormat" type="string" string="%PRI% %TIMESTAMP:::date-rfc3339% %HOSTNAME% %syslogtag:1:32% %msg%\n")
<h1>将匹配的日志（如 authpriv.<em>、</em>.err 或自定义 facility）发送给 Python 脚本</h1><p>if $syslogfacility-text == 'authpriv' or $syslogseverity >= 3 then {
action(
type="omprog"
binary="/opt/bin/alert_handler.py"
template="OmprogFormat"</p><h1>可选：设置超时和重启策略</h1><pre class="brush:php;toolbar:false;"> timeout="60"
ignorecase="off"
queue.size="10000"
queue.dequeuebatchsize="100"
queue.maxdiskspace="1g"
queue.saveonshutdown="on"
)
# 可选：同时保留原始日志到文件便于调试
stop}
重启前验证配置
rsyslogd -N1
说明：
-
template 必须末尾带
\n，否则 omprog 无法按行识别 -
queue.*参数提升可靠性：当日志洪峰或脚本临时卡顿时，rsyslog 自动缓存并重试 -
stop防止重复处理（尤其多规则场景） - 测试配置语法：
rsyslogd -N1；重载服务：systemctl reload rsyslog
调试与运维要点
常见问题集中在通信中断、脚本退出、权限不足或解析失败：
- 脚本异常退出会导致 rsyslog 自动重启它（默认行为），但频繁崩溃会刷屏日志。用
journalctl -u rsyslog -f观察 stderr 输出 - Python 脚本中避免使用阻塞操作（如未设 timeout 的 requests 请求），否则会拖慢整个 rsyslog 队列
- 若需环境变量（如 API token），在
omprogaction 中用env="KEY=VALUE"显式传入，不要依赖用户 shell 环境 - 测试触发：用
logger -p authpriv.err "Test ERROR from logger"查看是否被脚本捕获 - 性能敏感场景下，可在脚本中加简单计数器或采样逻辑，避免每条 ERROR 都发告警
