Logstash 聚合返回空消息答案

【问题标题】：Logstash aggregation return empty messageLogstash 聚合返回空消息
【发布时间】：2021-07-26 11:37:15
【问题描述】：

我有一个测试环境，可以在投入生产之前测试一些 logstash 插件。

目前，我正在使用 kiwi syslog 生成器来生成一些用于测试的 syslog。

我的字段如下：

@timestamp
message
+ elastic medatadata

从这个基本字段开始，我开始过滤我的数据。首先是在timestamp and message的基础上添加一个新字段，如下：

input {
  syslog {
    port => 514
 }
}
filter {
  prune {
    whitelist_names =>["timestamp","message","newfield", "message_count"]
  }
  mutate {
        add_field => {"newfield" => "%{@timestamp}%{message}"}
  }
}

prune 只是为了不处理不需要的数据。

这很好用，因为我得到了一个包含这 2 个值的新字段。

下一步是根据消息的特定内容运行一些聚合，例如消息是否包含logged in 或logged out

为此，我使用了聚合过滤器

  grok {
    match => {
        "message" => [
        "(?<[@metadata][event_type]>logged out)",
            "(?<[@metadata][event_type]>logged in)",
            "(?<[@metadata][event_type]>workstation locked)"

   ]

 }
}
   aggregate {
        task_id => "%{message}"
        code => "
        map['message_count'] ||= 0; map['message_count'] += 1;
        "
        push_map_as_event_on_timeout => true
        timeout_timestamp_field => "@timestamp"
        timeout => 60
        inactivity_timeout => 50
        timeout_tags => ['_aggregatetimeout']
           
        
      }
    
}

这按预期工作，但我在这里遇到了问题。当聚合超时时。为特定聚合填充的唯一字段是 message_count

正如你在上面的截图中看到的那样，newfield 和消息（最左边的那个，很抱歉它不适合截图）都是空的。

出于演示和测试目的，这绝对没问题，但如果我每秒收到数百个系统日志而不知道message_count 所指的消息，那将是无法管理的。

拜托，我在这里苦苦挣扎，我不知道如何解决这个问题，请有人帮助我了解如何用它所指的消息内容填写newfield？

这是我的整个 logstash 配置，以使其更容易。

input {
  syslog {
    port => 514
 }
}
filter {
  prune {
    whitelist_names =>["timestamp","message","newfield", "message_count"]
  }
  mutate {
        add_field => {"newfield" => "%{@timestamp}%{message}"}
  }
  grok {
    match => {
        "message" => [
        "(?<[@metadata][event_type]>logged out)",
            "(?<[@metadata][event_type]>logged in)",
            "(?<[@metadata][event_type]>workstation locked)"

   ]

 }
}
   aggregate {
        task_id => "%{message}"
        code => "
        map['message_count'] ||= 0; map['message_count'] += 1;
        "
        push_map_as_event_on_timeout => true
        timeout_timestamp_field => "@timestamp"
        timeout => 60
        inactivity_timeout => 50
        timeout_tags => ['_aggregatetimeout']
           
        
      }
    
}
output {
  elasticsearch {
     hosts => ["localhost:9200"]
         index => "logstash_index"
 }
  stdout {
    codec => rubydebug
 }
  csv {
    path => "C:\Users\adminuser\Desktop\syslog\syslogs-%{+yyyy.MM.dd}.csv"
    fields => ["timestamp", "message", "message_count", "newfield"]
 }
}

【问题讨论】：

标签： logstash logstash-grok logstash-configuration

【解决方案1】：

push_map_as_event_on_timeout => true

当您使用它并发生超时时，它会使用地图的内容创建一个新事件。如果您希望原始消息中的字段出现在新事件中，则必须将它们添加到地图中。对于 task_id，有一个简写符号可以使用过滤器上的 timeout_task_id_field 选项执行此操作，否则您已明确添加它们

map['newfield'] ||= event.get('newfield');

【讨论】：