Logstash：拆分后基于 If 的事件丢弃

Question

情况：

我的输入日志如下所示：

{
"key1":"value1"
"key2":"value2"
"events":
[{
"Level":"Information",
"Code":"100"
},
{
"Level":"Information",
"SomeKey":"SomeValue"
},
{
"Level":"Error",
"Message":"Something went wrong"
}
]}

我想：

1. 拆分事件数组以创建单独的对象，所有外部字段仍填充（key1 和 key2）。

2. 在拆分后选择性地删除日志。如果它们包含“代码”属性，我只想保留“信息级”日志。

我的 logstash 配置看起来像

filter {
 split {
  field => "[events]"
  }
}

 filter {
    if ![events][Code]
    { drop {} }
 }

 output {
   elasticsearch {}
 }

问题：

在执行第二个过滤器之前，Logstash 似乎没有分离事件。

换句话说，如果日志中的任何事件没有“代码”字段，则整个日志都将被删除，包括我必须保留的“错误”级别的信息。

我已经为此工作了大约一整天，这真的让我很紧张。我会手动尝试创建自己的插件，但我从未使用过任何 Ruby。

我很确定这无关紧要，但我在 Docker 中运行 ELK 堆栈。我相信配置文件可以正确加载，并且 Logstash 可以使用它们。

Answer 1

这实际上似乎是某种 logstash 错误。这是一个错误的演示：

配置文件：

input {
    stdin { codec => "json" }
}
filter {
    split { field => "events" }
    if ([events] == "" or [events][Code] == "") { 
        drop {} 
    }
}  
output {
    stdout { codec => "rubydebug" }
}

命令行：

echo '{"key1":"value1","key2":"value2","events":[{"Level":"Information","Code":"100"},{"Level":"Information","SomeKey":"SomeValue"},{"Level":"Error","Message":"Something went wrong"}]}' | bin/logstash -f test.conf

结果：

Exception in thread "[main]>worker7" java.lang.NumberFormatException: For input string: "Code"
    at java.lang.NumberFormatException.forInputString(NumberFormatException.java:65)
    at java.lang.Integer.parseInt(Integer.java:580)
    at java.lang.Integer.parseInt(Integer.java:615)
    at org.logstash.Accessors.fetch(Accessors.java:130)
    at org.logstash.Accessors.get(Accessors.java:20)
    at org.logstash.Event.getUnconvertedField(Event.java:160)
    at org.logstash.ext.JrubyEventExtLibrary$RubyEvent.ruby_get_field(JrubyEventExtLibrary.java:113)
    at org.logstash.ext.JrubyEventExtLibrary$RubyEvent$INVOKER$i$1$0$ruby_get_field.call(JrubyEventExtLibrary$RubyEvent$INVOKER$i$1$0$ruby_get_field.gen)

所以基本上它仍然认为 events 是一个数组，并试图通过“代码”对其进行索引

这似乎解决了这个问题：

    mutate {
            add_field => ["code", "%{[events][Code]}"]
    }
    if ([code] == "%{[events][Code]}") {
            drop {}
    }
    mutate {
            remove_field => ["code"]
    }