如何在Logstash中聚合多日志条目生成单条堆栈跟踪？

核心方案：使用Logstash aggregate 插件实现日志聚合 #

由于你只能修改Logstash配置，aggregate插件是最适合的工具——它可以基于自定义关联键，在流式处理中把分散的日志条目合并为单条。以下是具体实现步骤：

1. 标记日志的"起始行"和"后续行"

首先需要区分哪些日志是堆栈的起始（比如带时间戳、日志级别的错误首行），哪些是堆栈的后续行。假设你的日志格式是：

• 起始行示例：2024-05-20 10:00:00,123 ERROR [main] com.example.App: 初始化失败
• 堆栈后续行示例： at com.example.App.init(App.java:45)

用grok或正则判断并打标签：

filter {
  # 匹配起始行，打上start_line标签
  grok {
    match => { "message" => "^%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:loglevel} %{DATA:thread} %{DATA:class}:.*" }
    add_tag => ["start_line"]
    tag_on_failure => [] # 匹配失败不打_failure标签
  }

  # 非起始行打上follow_line标签
  if "start_line" not in [tags] {
    mutate {
      add_tag => ["follow_line"]
    }
  }
}

2. 用aggregate插件聚合同组日志

选择一个唯一关联键（比如Kubernetes的Pod ID+容器ID，或者日志自带的trace ID），把同组的起始行和后续行合并：

filter {
  aggregate {
    # 关联键：确保同一容器/同一请求的日志被分到一组
    task_id => "%{[kubernetes][pod][id]}-%{[kubernetes][container][id]}"
    # 聚合逻辑：起始行初始化数据，后续行追加内容
    code => "
      if event.get('start_line')
        # 起始行：初始化聚合的基础字段
        map['message'] = event.get('message')
        map['timestamp'] = event.get('timestamp')
        map['loglevel'] = event.get('loglevel')
        # 按需添加其他需要保留的字段，比如thread、class等
      else
        # 后续行：追加到message末尾
        map['message'] += '\n' + event.get('message')
      end
    "
    # 超时时间：30秒内无新日志则输出聚合结果（避免内存泄漏）
    timeout => 30
    # 超时后的处理：把聚合好的数据写入事件，清理标签
    timeout_code => "
      event.set('message', map['message'])
      event.set('timestamp', map['timestamp'])
      event.set('loglevel', map['loglevel'])
      event.remove('start_line')
      event.remove('follow_line')
    "
    # 只处理标记过的日志行
    when => "('start_line' in [tags] or 'follow_line' in [tags])"
    # 仅在超时后输出聚合事件，避免重复输出起始行
    push_map_as_event_on_timeout => true
    # 丢弃原事件（只保留最终聚合后的事件）
    discard_event => true
  }
}

关键注意事项 #

• 关联键的选择：如果日志带有trace_id（同一请求的唯一标识），用%{trace_id}作为task_id会更准确，能精准聚合同一请求的所有日志（包括堆栈）；如果没有trace ID，用Pod+容器ID是退而求其次的选择。
• 超时时间调整：根据你的日志输出频率调整timeout值——如果堆栈行间隔较长，适当调大（比如60秒），避免聚合不完整；如果日志流量大，调小减少延迟和内存占用。
• 起始行匹配准确性：如果grok正则匹配不准，改用ruby判断（比如检查message是否以时间戳开头），确保起始行和后续行的标记正确，否则聚合逻辑会失效。

备选方案：用Ruby插件自定义聚合逻辑 #

如果aggregate插件的灵活性不够，可以直接用ruby插件维护内存中的哈希表，手动处理聚合：

filter {
  ruby {
    init => "
      # 初始化内存哈希，存储正在聚合的日志
      @log_groups = {}
      # 定时清理超时的日志组（每60秒）
      Thread.new {
        loop do
          sleep 60
          now = Time.now.to_i
          @log_groups.delete_if { |k, v| now - v[:timestamp] > 30 }
        end
      }
    "
    code => "
      pod_id = event.get('[kubernetes][pod][id]')
      container_id = event.get('[kubernetes][container][id]')
      group_key = \"#{pod_id}-#{container_id}\"

      # 判断是否为起始行
      if event.get('message') =~ /^\d{4}-\d{2}-\d{2}/
        # 起始行：创建新的日志组
        @log_groups[group_key] = {
          message: event.get('message'),
          timestamp: Time.now.to_i,
          fields: event.to_hash.reject { |k, v| k == 'message' }
        }
        event.cancel # 暂不输出起始行
      else
        # 后续行：追加到对应日志组
        if @log_groups.key?(group_key)
          @log_groups[group_key][:message] += '\n' + event.get('message')
          event.cancel # 丢弃后续行
        else
          # 找不到对应起始行，直接输出（避免丢失日志）
        end
      end

      # 检查超时的日志组，输出并清理
      now = Time.now.to_i
      @log_groups.each do |k, v|
        if now - v[:timestamp] > 30
          # 创建新事件输出聚合结果
          new_event = LogStash::Event.new(v[:fields])
          new_event.set('message', v[:message])
          new_event.set('timestamp', v[:timestamp])
          pipeline.output(new_event)
          @log_groups.delete(k)
        end
      end
    "
  }
}

这个方案更灵活，但需要注意内存占用——定时清理超时的日志组是必须的，避免OOM。

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