仅使用 grok 过滤器将键值对作为 logstash 输出
Have a key value pair as logstash output, by only using grok filter
我正在开发一个 spring 引导项目并使用 ELK 堆栈进行日志记录和审计。我需要一个 logstash.conf 文件来处理日志,输出可以有动态键值对。此输出数据将用于审计。
为更清楚起见添加示例
示例:
示例日志:
[INFO] [3b1d04f219fc43d18ccb6cb22db6cff4] 2021-10-13_13:43:09.074 Audit_ key1:value1| key2:value2| key3:value3| keyN:valueN
所需的 logstash 输出:
{
"logLevel": [
[
"INFO"
]
],
"threadId": [
[
"3b1d04f219fc43d18ccb6cb22db6cff4"
]
],
"timeStamp": [
[
"2021-10-13_13:43:09.074"
]
],
"class": [
[
"Audit_"
]
],
"key1": [
[
"value1"
]
],
"key2": [
[
"value2"
]
],
"key3": [
[
"value3"
]
],
"keyN": [
[
"valueN"
]
]
}
注:
- “key”永远是一个词或字符串值
- "value"可以是单词、数字或句子(带空格的字符串)
- ":"是key和value之间的分隔符
- “|”是键值对之间的分隔符
- 键值对的数量可以变化。
有人 suggest/help 我可以在这里使用匹配模式吗?我只允许使用 grok 过滤器。
感谢 Filip 和 leandrojmp 的指导!
仅仅为此使用 grok 过滤器,会使它变得非常复杂,而且它不支持动态键值对。
所以我结合了 grok 和 kv 过滤器。这种方法对我有用。
示例日志:
[INFO] [3b1d04f219fc43d18ccb6cb22db6cff4] 2021-10-13_13:43:09.074 _Audit_ key1:value1| key2:value2| key3:value3| keyN:valueN
logstash.conf 文件:
input {
beats {
port => "5044"
}
}
filter {
grok {
match => {"message" => "\[%{LOGLEVEL:logLevel}\]\ \[%{WORD:traceId}\]\ (?<timestamp>[0-9\-_:\.]*)\ %{WORD:class}\ %{GREEDYDATA:message}"]}
overwrite => [ "message" ]
}
if [class] == "_Audit_" {
kv {
source => "message"
field_split => "&"
value_split => "="
remove_field => ["message"]
}
}
}
output {
if [class] == "_Audit_" {
elasticsearch {
hosts => ["localhost:9200"]
index => "audit-logs-%{+YYYY.MM.dd}"
}
}
else {
elasticsearch {
hosts => ["localhost:9200"]
index => "normal-logs-%{+YYYY.MM.dd}"
}
}
}
我正在开发一个 spring 引导项目并使用 ELK 堆栈进行日志记录和审计。我需要一个 logstash.conf 文件来处理日志,输出可以有动态键值对。此输出数据将用于审计。
为更清楚起见添加示例
示例:
示例日志:
[INFO] [3b1d04f219fc43d18ccb6cb22db6cff4] 2021-10-13_13:43:09.074 Audit_ key1:value1| key2:value2| key3:value3| keyN:valueN
所需的 logstash 输出:
{
"logLevel": [
[
"INFO"
]
],
"threadId": [
[
"3b1d04f219fc43d18ccb6cb22db6cff4"
]
],
"timeStamp": [
[
"2021-10-13_13:43:09.074"
]
],
"class": [
[
"Audit_"
]
],
"key1": [
[
"value1"
]
],
"key2": [
[
"value2"
]
],
"key3": [
[
"value3"
]
],
"keyN": [
[
"valueN"
]
]
}
注:
- “key”永远是一个词或字符串值
- "value"可以是单词、数字或句子(带空格的字符串)
- ":"是key和value之间的分隔符
- “|”是键值对之间的分隔符
- 键值对的数量可以变化。
有人 suggest/help 我可以在这里使用匹配模式吗?我只允许使用 grok 过滤器。
感谢 Filip 和 leandrojmp 的指导!
仅仅为此使用 grok 过滤器,会使它变得非常复杂,而且它不支持动态键值对。
所以我结合了 grok 和 kv 过滤器。这种方法对我有用。
示例日志:
[INFO] [3b1d04f219fc43d18ccb6cb22db6cff4] 2021-10-13_13:43:09.074 _Audit_ key1:value1| key2:value2| key3:value3| keyN:valueN
logstash.conf 文件:
input {
beats {
port => "5044"
}
}
filter {
grok {
match => {"message" => "\[%{LOGLEVEL:logLevel}\]\ \[%{WORD:traceId}\]\ (?<timestamp>[0-9\-_:\.]*)\ %{WORD:class}\ %{GREEDYDATA:message}"]}
overwrite => [ "message" ]
}
if [class] == "_Audit_" {
kv {
source => "message"
field_split => "&"
value_split => "="
remove_field => ["message"]
}
}
}
output {
if [class] == "_Audit_" {
elasticsearch {
hosts => ["localhost:9200"]
index => "audit-logs-%{+YYYY.MM.dd}"
}
}
else {
elasticsearch {
hosts => ["localhost:9200"]
index => "normal-logs-%{+YYYY.MM.dd}"
}
}
}