仅使用 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"
    ]
  ]
}

注:

有人 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}"
      }
  }
}