Logstash 如何正确处理多行日志（如 Java 堆栈）？

核心：在 file input 中使用 codec => multiline，通过 pattern 正则定义"新日志行的开头"（如时间戳），negate => true + what => "previous" 表示不匹配的行追加到上一条日志。

完整配置

input {
  file {
    path => "/var/log/myapp/*.log"
    start_position => "beginning"

    codec => multiline {
      pattern => "^\["          # 以 [ 开头的行是新日志
      negate => true            # 不匹配 pattern 的行...
      what => "previous"        # ...追加到上一条事件
    }
  }
}

原理解释

假设日志格式如下：

[2024-01-15 10:30:00] ERROR 处理请求失败
java.lang.NullPointerException
    at com.example.Service.process(Service.java:42)
    at com.example.Controller.handle(Controller.java:15)
[2024-01-15 10:30:01] INFO 请求完成

• pattern => "^\["：以 [ 开头的行是新日志的开始；
• negate => true：不以 [ 开头的行（堆栈行）不是新日志；
• what => "previous"：这些行追加到上一条日志中。

结果：Java 堆栈被合并为一条完整的日志事件，而不是每一行都是独立事件。

常见日志格式的 pattern

# 时间戳开头（YYYY-MM-DD）
pattern => "^\d{4}-\d{2}-\d{2}"

# log4j 风格 [LEVEL]
pattern => "^\[(TRACE|DEBUG|INFO|WARN|ERROR)\]"

# Python logging 默认格式
pattern => "^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}"

观测云对照

观测云 DataKit 的日志采集配置支持多行合并：通过 multiline_match 参数指定新日志行的正则（如 ^\\d{4}-\\d{2}-\\d{2}），堆栈自动合并为一条完整日志。Pipeline 解析时不再需要处理拆散的堆栈行。

常见问题（FAQ）

Q：多行日志超过一定行数会截断吗？
A：会。multiline codec 有 max_lines 参数（默认 500 行），超长的堆栈会被截断。可调整：max_lines => 1000。

Q：多行事件会等待超时吗？
A：如果最后一条日志没有后续新行触发"刷新"，它会一直等待。设置 auto_flush_interval => 5（秒）确保最后一条也能及时输出。

Q：Filebeat 也能处理多行吗？
A：能。Filebeat 的 multiline.pattern + multiline.negate + multiline.match 配置与 Logstash 类似，在采集端就做合并。