ELK、Loki、Kafka 三种日志告警联动方案全解析（附实战 Demo）
原创
ELK、Loki、Kafka 三种日志告警联动方案全解析（附实战 Demo）
原创
摘要
我们经常遇到这样的场景：**生产环境系统出问题了，日志早就写好了报错信息，但团队没人知道。**系统看起来“运行正常”，其实用户早就被 Bug 打败了。这篇文章会从实战角度出发，介绍如何将日志系统与告警通道打通：让日志真正“叫得出声”，第一时间触发钉钉、邮箱、短信等渠道告警。
引言
随着微服务架构的普及，一个用户请求背后可能跨越十几个服务节点。我们在每个节点记录日志，但如果没有统一的监控和告警机制，这些日志就像写在墙上的涂鸦——没人看，也没人管。
现代的日志系统已经不再是“写到文件里就完事”，它应该是系统可观测性的一部分。我们可以通过：
- ELK + Watcher 实现日志字段级触发
- Loki + Promtail + Prometheus + Alertmanager 实现 Grafana 风格日志告警
- Kafka 日志流 + 实时消费 + 条件匹配触发钉钉机器人
下面，我们从简单到复杂，逐一讲解实现方式。
ELK + Watcher：老牌告警组合
什么是 Watcher？
Watcher 是 Elasticsearch 的一款 X-Pack 插件，用于监控数据变动并触发行为。可以设定日志字段中的关键字（如“ERROR”、“OutOfMemory”），超过阈值或出现即触发告警。
Demo：只要日志里出现 ERROR，马上发邮件告警
前置条件
- 已搭建好 ELK（Filebeat → Logstash → Elasticsearch → Kibana）
- Kibana 中可以搜索到 ERROR 日志
Watcher 创建脚本（可通过 Kibana Dev Tools 执行）
PUT _watcher/watch/error_alert_watch
{
"trigger": {
"schedule": {
"interval": "1m"
}
},
"input": {
"search": {
"request": {
"indices": ["logstash-*"],
"body": {
"query": {
"match": {
"message": "ERROR"
}
}
}
}
}
},
"condition": {
"compare": {
"ctx.payload.hits.total.value": {
"gt": 0
}
}
},
"actions": {
"send_email": {
"email": {
"to": "ops@example.com",
"subject": "日志告警：发现 ERROR 日志",
"body": "请检查系统日志，发现 ERROR 报错。"
}
}
}
}Loki + Promtail + Alertmanager：现代云原生方案
为什么推荐？
Loki 是 Grafana 推出的日志系统，支持 PromQL 查询语法，轻量级、原生支持日志与指标结合，配合 Alertmanager 可直接推送钉钉、微信、邮件等。
Loki 示例配置
Promtail 收集日志 → Loki 聚合 → Alertmanager 告警
groups:
- name: error-logs
rules:
- alert: AppErrorLog
expr: count_over_time({job="app"} |= "ERROR"[1m]) > 0
for: 1m
labels:
severity: warning
annotations:
summary: "发现 ERROR 日志"
description: "服务出现错误日志，请立即检查。"Alertmanager 配置（钉钉）
receivers:
- name: dingtalk
webhook_configs:
- url: 'https://oapi.dingtalk.com/robot/send?access_token=xxx'可以结合 Webhook 转换器（如 dingtalk-webhook-adapter）将 Prometheus 告警转为钉钉格式。
Kafka + 日志消费 + 自定义告警
什么场景适合？
适用于日志量超大、不适合用全文索引系统查询的情况，比如：
- 高频日志（百万级）
- 特殊告警逻辑（非简单字符串匹配）
实现思路
- 所有服务日志通过 Filebeat 发送到 Kafka
- 编写一个 Node.js 或 Python 消费者
- 识别日志中的关键词（如“OutOfMemory”、“user_id=0”等）
- 匹配即触发钉钉/邮件告警
Node.js Kafka 消费 + 钉钉推送示例
const { Kafka } = require('kafkajs');
const axios = require('axios');
const kafka = new Kafka({ clientId: 'log-alert', brokers: ['localhost:9092'] });
const consumer = kafka.consumer({ groupId: 'log-alert-group' });
(async () => {
await consumer.connect();
await consumer.subscribe({ topic: 'logs', fromBeginning: false });
await consumer.run({
eachMessage: async ({ message }) => {
const log = message.value.toString();
if (log.includes('ERROR') || log.includes('OutOfMemory')) {
await axios.post('https://oapi.dingtalk.com/robot/send?access_token=xxx', {
msgtype: 'text',
text: { content: `日志告警：${log}` }
});
}
},
});
})();真实场景案例：生产支付服务异常自动提醒
场景：
支付服务日志中记录了一个字段 payment_status = failed，但因为日志没被及时处理，业务侧连续丢单 20 分钟。
解决方案：
- 将日志聚合到 Loki
- 使用如下 PromQL 规则
count_over_time({app="payment"} |= "payment_status=failed"[5m]) > 10- 超过 10 条记录就触发钉钉告警，精度达到分钟级别，从“事后补救”升级为“实时自愈”。
QA 环节
Q1: 日志告警太频繁怎么办？
A: 可以通过设置 抖动时间（for: 5m）、分组告警（group_by）、聚合告警（count > N） 降低骚扰频率。
Q2: 告警能包含堆栈信息吗？
A: 可以。ELK 中 watcher 动作支持引用字段，Loki 支持 line_format，Kafka 消费者也可以截取关键字段作为告警内容。
Q3: 可以指定某些服务日志不告警吗？
A: 可以通过正则或标签过滤，比如：
{job!="dev"} |= "ERROR"避免开发环境日志触发告警。
总结
日志不止用来看，更重要的是“能触发”，也就是我们说的日志驱动告警系统。只要你把日志打得够规范，后续无论是用 Watcher、Loki 还是 Kafka，都能帮你实现实时响应的问题发现体系。
原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。
如有侵权，请联系 cloudcommunity@tencent.com 删除。
原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。
如有侵权，请联系 cloudcommunity@tencent.com 删除。
腾讯云开发者
