Logstash 基础与配置
Logstash 是 ELK 栈里的数据处理引擎，说白了就是一个管道，把日志从源头接进来，中间洗一洗、筛一筛，最后送出去。我最早接触它是在一个电商平台的审计日志项目中，每天几十 GB 的访问日志要实时处理，Logstash 的插件机制让我既爱又恨——灵活是真灵活，但配置写不好，性能瞬间拉胯。这一节咱们就把它拆开揉碎了讲清楚。
输入、过滤、输出插件机制
Logstash 的架构就三个阶段：Input → Filter → Output。每个阶段都是插件组成的，你可以在配置里任意组合。我习惯把这种模型叫作“三明治架构”，中间那层 Filter 是最有花样的。
| 阶段 | 作用 | 常用插件 |
|---|---|---|
| Input | 从数据源拉取或监听日志 | file, beats, tcp, syslog, http |
| Filter | 解析、转换、过滤、丰富数据 | grok, mutate, date, drop, geoip |
| Output | 把处理后的数据发到目标存储 | elasticsearch, stdout, csv, kafka |
插件之间用 filter { ... } 块串联，你可以写多个 filter 插件，它们按顺序执行。举个例子，先用 grok 解析一行 Nginx 日志，再用 mutate 把 IP 字段重命名，最后用 date 把时间戳转成 ES 需要的格式。
核心原则：每个插件只做一件事，保持管道清晰。别在一个 mutate 里同时干 rename、add_field、convert，拆成多个 mutate 块更容易维护。
配置文件结构
Logstash 的配置文件是类 Ruby 语法，块结构用花括号 { } 包起来。我个人习惯把输入、过滤、输出分成三个文件，用 --config 参数指定目录，这样多人协作时不会互相踩脚。
一个最基本的配置文件长这样：
input {
file {
path => "/var/log/nginx/access.log"
start_position => "beginning"
sincedb_path => "/dev/null"
}
}
filter {
grok {
match => { "message" => '%{COMBINEDAPACHELOG}' }
}
}
output {
elasticsearch {
hosts => ["localhost:9200"]
index => "nginx-access-%{+YYYY.MM.dd}"
}
stdout { codec => rubydebug }
}注意 sincedb_path => "/dev/null"，这是调试时的常用技巧——每次重启都从文件头开始读。但生产环境千万别这么用，否则重启后全量重发，ES 会报警。我曾经在压测环境忘了改回来，结果把测试集群撑爆了，嗯，后来吸取教训。
sincedb_path 持久化路径，否则 Logstash 重启后会丢失文件读取偏移量，要么重复读，要么漏读。
配置文件的几个关键点：
- codec：输入输出都可以指定 codec，比如 json、multiline、plain。处理多行日志（如 Java 堆栈）时用 multiline 最省事。
- if 条件：可以在 filter 或 output 中用
if [field] == "value" { ... }做分支处理，实现按日志级别分流到不同索引。 - @timestamp：每个事件默认带一个
@timestamp字段，你可以用 date 插件覆盖它，否则 ES 会用 Logstash 收到的当前时间。
简单日志采集示例
咱们跑一个完整的采集流程。假设你要收集某台服务器的 syslog 审计日志，格式是这样的：
Mar 15 10:20:30 hostname sshd[12345]: Failed password for root from 10.0.0.1 port 22 ssh2目标：把这条日志解析出时间、主机名、进程、源 IP、认证结果，然后存入 Elasticsearch 的 syslog-YYYY.MM.dd 索引。
第一步：编写 Logstash 配置
input {
beats {
port => 5044
ssl => false
}
}
filter {
grok {
match => { "message" => "%{SYSLOGTIMESTAMP:timestamp} %{SYSLOGHOST:hostname} %{DATA:program}(?:\[%{POSINT:pid}\])?: %{GREEDYDATA:message_content}" }
}
mutate {
rename => { "message_content" => "raw_message" }
}
if [raw_message] =~ /Failed password/ {
grok {
match => { "raw_message" => "Failed password for %{USER:username} from %{IP:source_ip} port %{INT:port} %{WORD:protocol}" }
}
mutate {
add_field => { "auth_result" => "failed" }
}
} else if [raw_message] =~ /Accepted password/ {
grok {
match => { "raw_message" => "Accepted password for %{USER:username} from %{IP:source_ip} port %{INT:port} %{WORD:protocol}" }
}
mutate {
add_field => { "auth_result" => "success" }
}
}
date {
match => [ "timestamp", "MMM dd HH:mm:ss", "MMM d HH:mm:ss" ]
target => "@timestamp"
}
}
output {
elasticsearch {
hosts => ["172.16.0.10:9200"]
index => "syslog-audit-%{+YYYY.MM.dd}"
}
stdout { codec => rubydebug }
}第二步：启动 Logstash
假设配置文件叫 syslog-pipeline.conf，执行：
bin/logstash -f syslog-pipeline.conf --config.reload.automatic--config.reload.automatic 这个参数很实用，改完配置不用重启，Logstash 自动 reload。我第一次用的时候还以为是玄学，后来发现它确实靠谱。
第三步：发一条测试数据
用 Filebeat 或者直接 echo 一行日志到 TCP 端口都可以。我习惯用 nc 模拟：
echo 'Mar 15 10:20:30 hostname sshd[12345]: Failed password for root from 10.0.0.1 port 22 ssh2' | nc localhost 5044如果配置了 stdout 输出，控制台会显示解析后的 JSON 事件，包含 auth_result: "failed"、source_ip: "10.0.0.1" 等字段。
验证技巧：在 output 里配上 stdout { codec => dots }，每收到一条事件就打印一个点，快速确认管道吞吐量是否正常。我在做压力测试时经常这么干。
Logstash Pipeline 流程图
下面这张图展示了我们刚才搭建的 Syslog 审计日志采集管线。数据从左到右流经 Input、Filter、Output，每个阶段内部可拆分多个插件。我画这张图是为了让你直观理解“三明治架构”。
看到这张图你可能想问：为什么 Filter 要写那么多步骤？直接一个 grok 全部解析不行吗？我试过，但 grok 模式越复杂越难调试，而且一旦匹配失败整条事件就被标记为 _grokparsefailure。拆成多个条件分支，匹配失败只丢那部分字段，不会影响其他字段解析。这是我从一个线上事故里总结出来的教训——有一次因为日志格式小变化导致全量日志解析失败，ES 索引全是空的，排查了半天才发现是 grok 模式太“贪心”。
常见问题与调优建议
- 性能瓶颈：Filter 里如果用了复杂的 grok 正则，或者 geoip 查库，会导致处理延迟。我建议把耗时的 filter 放在队列后面，或者用多 pipeline 分离重型处理。
- 内存管理：Logstash 默认堆内存 1GB，生产环境可能不够。在 jvm.options 里调整
-Xms2g -Xmx4g，别超过机器内存的 50%，要给 OS 留缓存。 - 队列持久化：如果担心 Logstash 崩溃丢数据，可以开启持久化队列（
queue.type: persisted），但会降低吞吐量，适合对数据完整性要求极高的场景。我一般只在审计日志场景开启，普通业务日志用内存队列。