仅使用 grok 过滤器将键值对作为 logstash 输出答案

【问题标题】：Have a key value pair as logstash output, by only using grok filter仅使用 grok 过滤器将键值对作为 logstash 输出
【发布时间】：2021-10-19 14:52:04
【问题描述】：

我正在开发一个 Spring Boot 项目，并使用 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”总是一个单词或字符串值
“值”可以是单词、数字或句子（带空格的字符串）
“:”是key和value的分隔符
“|”是键值对之间的分隔符
键值对的数量可能会有所不同。

有人可以建议/帮助我这里使用的匹配模式吗？我只能使用 grok 过滤器。

【问题讨论】：

有什么理由只使用 grok？这很容易使用 grok 和 kv 过滤器完成，但我不知道是否可以仅使用 grok，因为字段的数量和这些字段的值可能会有所不同，其中一些可能会破坏您的 grok 模式。
一个 logstash.conf 文件已经在运行，我想使用同一个 conf 文件。当前的 conf 文件只有 grok 过滤器。如果我在现有的 logstash.conf 文件中添加 kv 过滤器对现有流程有影响吗？
您可以随时使用logstash条件并检查事件属于哪个数据流。我不能告诉你究竟是怎么回事，因为你从来没有提到过其他流程或其他有问题的 logstash 配置。但我看不出你为什么不使用kv 过滤器的原因，我很确定你不能只使用grok。

标签： elasticsearch logstash logstash-grok elk

【解决方案1】：

感谢 Filip 和leadrojmp 的指导！

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

【讨论】：