Logstash Pipeline 是什么
Logstash 管道是经典的三段式结构：input 负责收取数据，filter 负责解析与改写，output 负责发送到目标。一条管道写在一个配置文件里，多管道可以在 pipelines.yml 中分别声明，各管道独立配置队列与线程，是日志平台里最常见的清洗层。
基本信息
| 结构 | input / filter / output 三段 |
|---|---|
| 配置文件 | .conf，或 pipelines.yml 声明多管道 |
| 常见 input | beats、kafka、file、tcp |
| 常见 output | elasticsearch、kafka、stdout |
| 调试手段 | -e 临时配置 + rubydebug |
| 并发模型 | pipeline workers 数决定处理能力 |
典型使用场景
- 接收 Filebeat 送来的日志，解析后写入 Elasticsearch
- 从 Kafka 消费日志做清洗，再分发到多个下游
- 对非结构化文本做 grok 解析与字段类型转换
- 用条件判断给不同来源的日志打上不同标签与索引
- 在写入前做字段裁剪与脱敏，控制存储与合规风险
- 用 stdout 与 rubydebug 在本地验证解析逻辑
单管道示例
单管道示例
input {
beats { port => 5044 } # 接收 Filebeat 数据
}
filter {
grok { match => { "message" => "%{TIMESTAMP_ISO8601:log_ts} %{LOGLEVEL:level} %{GREEDYDATA:msg}" } }
date { match => ["log_ts", "ISO8601"] } # 用日志时间覆盖事件时间
}
output {
elasticsearch {
hosts => ["http://es:9200"]
index => "applogs-%{+YYYY.MM.dd}"
}
}三段职责
| 段 | 职责 | 常见插件 |
|---|---|---|
| input | 接收数据 | beats、kafka、file、tcp |
| filter | 解析与改写 | grok、date、mutate、dissect |
| output | 发送数据 | elasticsearch、kafka、stdout |
| codec | 编码解码 | json、plain、rubydebug |
调试用 -e 跑临时配置、stdout { codec => rubydebug } 输出中间结果，验证后再上生产，比直接改线上配置安全得多。
date 过滤器很关键：不把日志时间覆盖到 @timestamp 时，索引会按接收时间归档，回放历史日志会全部落在当天。
管道吞吐受 workers 与批大小影响，调参前先用监控确认瓶颈在 CPU、堆内存还是下游写入。
多管道场景下用 pipelines.yml 的 id 区分，命名统一便于排障；各管道的队列设置相互独立。
持久化队列能在进程重启时保住未处理完的事件，对不能丢日志的场景建议开启并评估磁盘占用。
用 mutate 做字段裁剪与重命名，把不需要的字段在写入前删掉，既省存储也降低下游解析成本。
Logstash 自身指标（事件吞吐、队列积压、JVM 堆）要一并监控，否则出现延迟时很难判断瓶颈在哪一环。