聚合 logstash 过滤器配置

Question

我的目标是在logstash 中结合基于pId 的事件。但是我发现具有相同 pId 的事件没有被合并为一个事件。添加聚合后我看不到任何变化。请帮助

日志如下所示：

June 1st 2017, 11:51:26.992 {id} {pId} ClassName:methodName:99 [DEBUG] - Received request:
June 1st 2017, 11:51:26.993 {id} {pId} ClassName:methodName:100 [DEBUG] - Id: abbababcajdfbjasndflsdlf
June 1st 2017, 11:51:26.993 {id} {pId} ClassName:methodName:100 [DEBUG] - unique id: AAAAA
June 1st 2017, 11:51:26.993 {id} {pId} ClassName:methodName:100 [DEBUG] Total time: 12

这是我的配置：

filter {
grok{
match => { "message" => "%{DATESTAMP:log_timestamp} %{DATA:id} %{DATA:pId} %{DATA:ClassName} [%{LOGLEVEL:severity}] - %{GREEDYDATA:message}" }
}
if [message] =~ /Received request:/ {
aggregate {
task_id => "%{pId}"
code => "map['message'] = event['message']"
map_action => "create"
}
}
else if [message] =~ /Total time:^/ {
aggregate {
task_id => "%{pId}"
code => "map['new_message'] = event['message'];event['new_message'] = map['new_message']"
map_action => "update"
end_of_task => true
timeout => 120
}
}
else {
aggregate {
task_id => "%{pId}"
code => "map['new_message'] = event['message'];event['new_message'] = map['new_message']"
map_action => "update"
}
}
}

Answer 1

聚合是那些真的难以正确处理的过滤器之一。在很大程度上，由于 Logstash 被设计为并行处理管道，过滤器堆栈中的每个 aggregate 调用对于管道来说都是唯一的，并且您不能确定所有事件都会通过相同的管道。开箱即用。

如果您使用 -w 1 参数运行 logstash 以强制所有内容通过单个管道，您将获得此行为。

在这种情况下，我建议改用input 上的multiline 编解码器。这会将所有日志线集中在一个事件中，您可以稍后在过滤器阶段对其进行解析。当然，这假设这些多行事件中的每一个都同时被放入并且永远不会多路复用。如果你得到多路复用，那么聚合将需要失去你的并行性。

input {
  file {
    path => "/var/log/app/debug_logs.log"
    codec => multiline {
       pattern => "Received request:"
       negate => true
       what => previous
    }
  }
}

这将搜索 不 匹配您的Received request: 正则表达式的事件并将它们附加到上一行。当它确实看到Received request: 时，它将开始一个新事件。然后你的filter {} 阶段就会看到这个

message => "June 1st 2017, 11:51:26.992 {id} {pId} ClassName:methodName:99 [DEBUG] - Received request:\nJune 1st 2017, 11:51:26.993 {id} {pId} ClassName:methodName:100 [DEBUG] - Id: abbababcajdfbjasndflsdlf\nJune 1st 2017, 11:51:26.993 {id} {pId} ClassName:methodName:100 [DEBUG] - unique id: AAAAA\nJune 1st 2017, 11:51:26.993 {id} {pId} ClassName:methodName:100 [DEBUG] Total time: 12"

这在并行上下文中更容易操作。