不同使用场景,排查方式不同:
| 使用场景 | 排查重点 |
|---|---|
| Logstash filter grok | 日志是否进入 filter、pattern 是否匹配、_grokparsefailure |
| Elasticsearch Ingest Pipeline | simulate API、processor 返回结果 |
| Filebeat + Elasticsearch | 是否开启 processors、字段是否存在 |
| Graylog / OpenSearch | Grok pattern 是否注册、测试窗口 |
✅ 下面以 Logstash Grok(最常见) 为主,其他场景后面补充。
在 filter 前加 debug 输出:
input {
stdin {}
}
filter {
mutate {
add_tag => ["entered_filter"]
}
}或用 stdout:
output {
stdout {
codec => rubydebug
}
}_grokparsefailureLogstash Grok 匹配失败会自动打 tag:
tags: [
"_grokparsefailure"
]如果出现:
常见错误:字段名写错
filter {
mutate {
add_field => {
"raw_log" => "%{message}"
}
}
}message 还是 log 或 event.original示例:
输入日志:
10.0.0.1 GET /index.html 200 1024
Grok:
%{IP:client} %{WORD:method} %{URIPATH:path} %{NUMBER:status:int} %{NUMBER:bytes:int}日志是:
a b c正则写死:
%{WORD:a} %{WORD:b} %{WORD:c}✅ 解决:
%{WORD:a}\s+%{WORD:b}\s+%{WORD:c}grok {
match => { "log" => "%{IP:client}" }
}但实际字段是 message
✅ 解决:
match => { "message" => "%{IP:client}" }grok {
patterns_dir => ["/etc/logstash/patterns"]
match => { "message" => "%{MY_PATTERN}" }
}✅ 检查:
%{IP:Client} # 不推荐✅ 推荐:
%{IP:client}break_on_match 控制匹配行为默认:匹配第一个成功即停止
grok {
match => {
"message" => [
"pattern1",
"pattern2"
]
}
break_on_match => false
}POST _ingest/pipeline/_simulate
{
"pipeline": {
"processors": [
{
"grok": {
"field": "message",
"patterns": ["%{IP:client}"]
}
}
]
},
"docs": [
{
"_source": {
"message": "192.168.1.1"
}
}
]
}✅ 日志进没进?
✅ 字段名对不对?
✅ 有没有 _grokparsefailure?
✅ pattern 在 debugger 里能不能匹配?
你可以直接贴出:
我可以 直接帮你改到可用 ✅。