Logstash 过滤器：聚合 - 超时自动保存

Question

我在 AWS 中有一个 Lambda 函数，可以将日志报告给 ELK 实例。 lambda 函数的每次调用都会生成一个唯一的invocation_id，该invocation_id 会随每个日志事件一起发送，因此可以在 ELK 中识别来自单个调用的事件。在操作结束时，我发送一个“Done”事件。

Lambda 函数可能会失败或超时，然后不会发送“完成”事件。

我想使用logstash aggregate filter 来识别失败的调用。含义 - 每个invocation_id 将是聚合映射中的task_id，“完成”事件将是end_of_task。

我需要告诉它“在超时（X 时间后没有收到完成事件）保存聚合事件，状态=失败”。

这个过滤器可以吗？如果是这样，语法是什么？从文档中不清楚..

Answer 1

Logstash 聚合过滤器从 2.3.0 版本开始支持超时事件生成。以下是如何使用此功能实现您想要的：

if [action] == "BEGIN" {
  aggregate {
    task_id => "%{id}"
    code => "map['bytes'] = 0"
    map_action => "create"
  }
} elseif [action] == "DONE" {
  aggregate {
    task_id => "%{id}"
    code => "event['bytes'] += map['bytes']"
    timeout_code => "event.tag('failed')"
    map_action => "update"
    end_of_task => true
    timeout => 10
    push_map_as_event_on_timeout => true
} else {
    aggregate {
    task_id => "%{id}"
    code => "map['bytes'] += event['bytes']"
    map_action => "update"
    add_tag => [ "drop" ]
}