logstash - Filebeat/Logstash 多行系统日志解析

Question

我正在将系统日志解析到 ELK 堆栈中。系统日志示例 Jul 19 10:47:21 host-abc systemd: Started myservice Jul 19 10:47:29 host-abc systemd: Started service. Jul 19 10:47:29 host-abc systemd: Starting service...

理想的做法是将第二行和第三行聚合为一条消息，例如返回：Started Service. Starting service... 因此我希望时间戳、主机名和程序名称在合并行之前匹配。

标签： logstashelastic-stacklogstash-grokfilebeat

Answer 1

您可以使用aggregate过滤器来实现您想要的。该aggregate过滤器支持基于公共字段值将多个日志行聚合到一个事件中。在您的情况下，公共字段将是@timestamp,hostname和的组合program_name。

由于syslog输入已经正确解析了 syslog 行，我们不需要了解任何内容，因此我们可以aggregate立即利用过滤器。我们根据字段聚合行，该SYSLOGBASE2字段将包含直到冒号字符的所有内容:。然后我们简单地收集所有消息，最后我们将消息连接成一个字符串。它是这样的：

input {
  syslog {
    ...
  }
}
filter {
  aggregate {
    task_id => "%{SYSLOGBASE2}"
    code => "map['message'] ||= []; map['message'].push(event.get('message'));"
    push_map_as_event_on_timeout => true
    timeout_task_id_field => "user_id"
    timeout => 1 # 1 second timeout
    timeout_tags => ['_aggregatetimeout']
    timeout_code => "event.set('message', map['message'].join(' '))"
  }
}
output {
  ...
}