Logstash 拆分日志并单独插入到elasticsearch中答案

【问题标题】：Logstash split log and insert it separately into elasticsearchLogstash 拆分日志并单独插入到elasticsearch中
【发布时间】：2021-09-14 05:55:54
【问题描述】：

我正在编写一个 logstash 配置文件，而我收到的一个日志给我带来了问题，团队向我发送了多个日志，例如合并为一个。

消息： [logitem(aaa=1, bbb=1, ccc=1), logitem(aaa=2, bbb=2, ccc=2), logitem(aaa=3, bbb=3, ccc=3)]

是否可以将这些日志拆分为 3 个并将它们单独插入到 elasticsearch 中？ (3 条记录)

【问题讨论】：

它们会一直是logitem 的三个项目吗？还是数量不同？
不，它不会总是 3，它会有所不同
这应该可以使用split 过滤插件来实现。如果您想要详细的配置答案，您应该发布一个更好的事件示例（如 logstash 所见） - 使用 stdout 输出插件并粘贴输出。

标签： ruby elasticsearch logstash

【解决方案1】：

使用红宝石过滤器

这种方式应该可行（请参阅下面的 cmets 进行讨论和参考）。您可能需要在几个地方调整 grok / scan 正则表达式。

grok {
    match => {
        "message" => "^\[%{GREEDYDATA:logitems}\]$"
    }
}

ruby {
    code => "event.set('logitem', event.get('message').scan(/logitem\([^\)]+\)/))"
}

split {
    field => "logitem"
}

grok {
    match => {
        "logitem" => "^logitem\(aaa=%{DATA:field_a}, bbb=%{DATA:field_b}, ccc=%{DATA:field_c}\)"
    }
}

扫描正则表达式的目的是匹配一个字符串：

以logitem开头
然后是( 字符
然后是除) 之外的任何字符中的一个或多个
以)结尾

使用 grok

令人惊讶的是，这种方式行不通。有关更多详细信息，请参阅this github issue。 TL;DR... grok 不会将重复的匹配项放入数组中。

filter {
  grok {
    match => {
      "message" => "^\[*(logitem\(%{DATA:logitem}\), )*logitem\(%{DATA:logitem}\)\]$"
    }
  }
  split {
    field => "logitem"
  }
}

如果您确定消息将始终采用aaa=, bbb= 格式，您可以更加明确。

[编辑 1：将 grok 方法标记为不工作并添加了 ruby 方法。 2：为了更好的流程重新排序了一些东西]

【讨论】：

嗨，汤姆，这几乎成功了！设法将日志分成两个不同的条目，但是 logitem 正则表达式部分只占用 logitem 中的最后 2 个条目
也许this answer 是更好的方法？
和 this github issue 有更多细节 - 看起来你确实需要使用 ruby 过滤器，令人惊讶！至少，如果您不知道该字段将重复多少次。
正在尝试重复正则表达式模式，但 ruby 扫描看起来不错，让我试试
ruby 代码方法对我有用，需要为 ruby 获取正则表达式，而 event.put 对我不起作用，所以我使用了 event.set，你想更新你的答案吗？另外我想知道是否可以修改您使用 grok 的最初想法以接受重复出现的模式