Logstash 数据采集与处理：输入、过滤器、输出
我把 Logstash 的 pipeline 理解成一个流水线车间。
输入是原料入口，过滤器是加工环节，输出是把成品送到仓库。告警数据的质量，就看这条流水线设计得够不够精细。
1. 输入阶段：数据从哪来
我最早做 ELK 时，习惯把所有日志都塞进 Logstash，结果发现告警触发了一堆假的。后来才明白——输入阶段就要筛选，不要什么数据都往里倒。
| 输入插件 | 适用场景 | 告警注意事项 |
|---|---|---|
| beats | Filebeat / Metricbeat 推送 | 建议带上 @metadata 字段，别在告警 payload 里混入无用指标 |
| tcp | 网络设备 syslog | 注意端口和安全组，我曾经因为默认 tcp 端口被墙耽误了半天排查 |
| file | 本地文件 tail | 要配合 sincedb，重启后不会重读已处理行，这点容易踩坑 |
| http | Webhook 推送 | 建议加 basic auth 或 验证，不然外部恶意推送能刷爆你的告警 |
type 标签，这样后面 filter 可以按类型做不同处理，代码清爽很多。2. 过滤器：告警数据的核心加工
告警规则好不好用，很大程度上取决于过滤器的配置。说白了，就是要从杂乱日志中把关键字段提取出来，并且清洗干净。
2.1 Grok：正则提取的艺术
Grok 是 Logstash 最常用的解析方式。它把正则表达式包装成 %{PATTERN:field} 的语法，省了你手写正则的痛苦。
filter {
grok {
match => { "message" => "%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:level} \[%{DATA:service}\] %{GREEDYDATA:message_body}" }
overwrite => [ "message" ]
}
}我在项目中遇到过这样的事：有一条 RDS 慢查询日志，时间戳格式里混了毫秒和时区，默认的 TIMESTAMP_ISO8601 匹配不上。后来我自己写了个 %{YEAR}-%{MONTHNUM}-%{MONTHDAY} %{HOUR}:%{MINUTE}:%{SECOND},%{INT:msec} 才搞定。记住，不要完全依赖内置模式，日志格式千奇百怪，要灵活组合。
_grokparsefailure 的 tag，然后输出到一个 dead letter queue，方便事后排查。2.2 Mutate：字段整形
告警场景里，经常需要把字符串转数字、改名、删掉冗余字段。Mutate 插件就是干这个的。
filter {
mutate {
convert => { "[response_time]" => "float" }
rename => { "host" => "[host][name]" }
remove_field => [ "path", "offset", "type" ]
}
}你想想看，如果不把 response_time 转成数值，后面告警阈值比较根本没法做。我见过有同事直接把字符串 "120" 扔进告警规则，结果永远是字符串比较，出不来正确结果。
2.3 Date：时间归一化
Logstash 默认用 @timestamp 作为事件时间，但很多日志自带的时间戳格式不标准。用 date 插件把它转成统一格式，能让告警时序更准确。
filter {
date {
match => [ "timestamp", "ISO8601", "yyyy-MM-dd HH:mm:ss,SSS" ]
target => "@timestamp"
timezone => "Asia/Shanghai"
}
}2.4 其他常用过滤器
- kv：解析键值对格式，比如 URL 参数或应用日志里的 k=v
- geoip：从 IP 获取地理位置，用于告警源 IP 分析
- drop：直接丢弃不需要的事件，节省下游算力
- clone：复制事件，便于同一份数据走不同告警流程
3. 输出阶段：数据去哪
告警数据经过过滤器处理之后，需要送到告警引擎或者 ES 索引里。输出阶段也要做基本的校验，避免把垃圾数据写入。
| 输出插件 | 告警用途 | 注意事项 |
|---|---|---|
| elasticsearch | 写入 ES 索引，供告警规则查询 | 建议按天或按类型分索引，设置 ILM 策略，防止磁盘爆满 |
| 直接发邮件告警（简单场景） | 频率限制很重要，我曾经一个错误日志循环触发上千封邮件，把公司邮箱打挂了 | |
| http | 调用外部 webhook（例如钉钉、Slack） | 注意重试机制和超时时间，网络波动会导致告警丢失 |
| stdout | 调试用 | 别在生产环境保留 stdout 输出，日志量大会撑爆日志文件 |
if [type] == "alert" 的分支，专门把告警相关的事件写到专用索引，这样查询更快。4. 完整 pipeline 示例：为 CPU 告警准备数据
我拿一个实际案例来说。假设我们用 Metricbeat 采集服务器 CPU 指标，想要在 CPU 使用率超过 90% 时触发告警。
input {
beats {
port => 5044
codec => "plain" # 有时 metricbeat 会发 json，这里用 plain 做演示
}
}
filter {
# 只处理 CPU 指标，过滤掉其他 system 数据
if [metricset][module] == "system" and [metricset][name] == "cpu" {
grok {
match => { "message" => "host=%{DATA:host} cpu_pct=%{NUMBER:cpu_pct:float} time=%{TIMESTAMP_ISO8601:timestamp}" }
}
mutate {
convert => { "cpu_pct" => "float" }
# 帮告警规则简化判断：直接生成一个布尔字段
if [cpu_pct] > 90.0 {
mutate { add_field => { "[alert][critical]" => "true" } }
}
}
date {
match => [ "timestamp", "ISO8601" ]
target => "@timestamp"
}
} else {
drop { }
}
}
output {
if [alert][critical] == "true" or [cpu_pct] > 85.0 {
elasticsearch {
hosts => ["localhost:9200"]
index => "cpu-alert-%{+YYYY.MM.dd}"
manage_template => false # 我用自建模板
}
# 同时发一份到消息队列，供其他系统消费
kafka {
topic_id => "alert-cpu"
codec => "json"
}
} else {
# 非告警数据也保存，用于趋势分析
elasticsearch {
hosts => ["localhost:9200"]
index => "cpu-non-alert-%{+YYYY.MM.dd}"
}
}
}这个配置里，我故意在 filter 阶段就打了 alert 标记，输出时根据这个标记分流。这样做的好处是——告警规则写起来特别简单，直接查 alert.critical: true 就行，不用再写复杂的计算逻辑。
manage_template => false。我刚开始没关，Logstash 自动创建了一个很松的 mapping，后面字段类型全乱了，告警阈值判断全失效。建议自己设计 ES 模板，然后关掉自动管理。5. 本章知识体系图
下面这张图总结了 Logstash 告警数据处理的核心流程——从输入到输出，每一步都在为告警规则准备好数据。
6. 避坑与实战建议
- 测试先行：我在本地用
stdout { codec => dots }观察流量，确认 pipeline 没问题再上生产 - 不要在一个 filter 块里做太多事：拆成多个 filter 块，每个只负责一件事，便于调试和复用
- 善用条件判断：Logstash 的
if语句很强大，但别写太复杂，可以把判断逻辑拆到不同 stage - 监控 pipeline 健康度：用
monitoring.elasticsearch.hosts把运行指标上报到 ES，实时看丢弃率和处理延迟