Logstash aggregate filter：新Start事件触发时关闭旧聚合的方法

嘿，这个问题我之前在处理Logstash日志聚合的时候也踩过坑！超时方案确实在并发的Start事件面前不靠谱，不过咱们可以用强制结束上一个聚合任务的思路来解决，核心是利用aggregate filter的end_task动作结合全局变量追踪最近的任务ID，具体配置和思路如下：

核心思路 #

当新的Start事件进来时，先触发结束上一个未完成的聚合任务，再创建新的聚合任务。这样就能保证每个Start事件对应的聚合都是独立的，不会被后续的Start事件打乱。

具体配置步骤 #

1. 为每个Start事件生成唯一任务ID

首先用Ruby filter给每个Start事件分配一个唯一的task_id，同时追踪最近的任务ID，用来触发上一个任务的结束：

ruby {
  code => "
    # 取出之前保存的上一个任务ID
    last_task_id = @last_task_id
    if event.get('message') =~ /^Start/
      # 生成新的唯一任务ID（也可以用日志里自带的唯一标识代替UUID）
      new_task_id = SecureRandom.uuid
      event.set('task_id', new_task_id)
      # 如果存在上一个任务，标记要结束它
      if last_task_id
        event.set('end_prev_task', last_task_id)
      end
      # 更新全局变量，保存当前任务ID
      @last_task_id = new_task_id
    else
      # 非Start事件，绑定到最近的任务ID上
      event.set('task_id', @last_task_id)
    end
  "
  # 初始化全局变量，避免空指针
  init => "@last_task_id = nil"
}

2. 配置两个Aggregate Filter

第一个用来结束上一个任务，第二个用来处理当前的聚合：

# 第一个Aggregate：结束上一个未完成的聚合任务
aggregate {
  task_id => "%{end_prev_task}"
  action => "end_task"
  # 只有当标记了要结束上一个任务时才触发
  condition => [ "end_prev_task", "not", "nil" ]
  timeout => 0 # 立即结束任务
  # 结束时输出聚合好的内容
  timeout_code => "
    event.set('aggregated_message', map['aggregated_message'])
  "
}

# 第二个Aggregate：处理当前的聚合任务
aggregate {
  task_id => "%{task_id}"
  action => "create_or_update"
  # 确保只有绑定了有效任务ID的事件才会被处理
  condition => [ "task_id", "not", "nil" ]
  # 初始化聚合内容为Start消息
  map => { "aggregated_message" => "%{message}" }
  # 后续非Start消息追加到聚合内容中
  code => "
    if event.get('message') !~ /^Start/
      map['aggregated_message'] += ' ' + event.get('message')
    end
  "
  # 给最后一个任务留个超时兜底，防止没有后续Start事件导致丢失
  inactivity_timeout => 300 # 可根据实际业务调整时长，这里是5分钟
  # 超时或任务结束时输出聚合结果
  timeout_code => "
    event.set('aggregated_message', map['aggregated_message'])
  "
}

关键注意事项 #

• 单Worker运行：因为我们用了Logstash的全局变量@last_task_id，每个Worker会维护自己的变量副本，如果开启多Worker，会导致任务追踪混乱。如果必须用多Worker，可以改用Redis等外部存储来共享最近的任务ID，但配置复杂度会上升。
• 用业务唯一ID替代UUID：如果你的Start日志里自带唯一标识（比如订单ID、请求ID），直接用这个ID作为task_id会比UUID更高效，也更贴合业务场景。

这样配置后，不管两个Start事件间隔多短，新的Start都会先结束上一个聚合任务，再开启新的，完美解决你遇到的聚合混乱问题！

内容的提问来源于stack exchange，提问作者herm