Logstash Grok多日志行识别异常：如何区分新条目与续行？

核心思路 #

要区分新日志条目和异常续行，关键是利用multiline codec将非起始行合并到上一条日志中，再用Grok正常解析合并后的完整条目。以下是具体配置和注意事项：

1. 正确配置Multiline Codec #

根据你的日志格式，选择合适的匹配规则，将异常续行合并到上一条日志：

场景1：新日志以时间戳/日志级别开头（如Java标准日志）

input {
  file {
    path => "/path/to/your/logs/*.log"
    codec => multiline {
      # 匹配新日志的起始特征（这里用ISO时间戳或日志级别）
      pattern => "^%{TIMESTAMP_ISO8601}|^(DEBUG|INFO|WARN|ERROR|FATAL)"
      negate => true  # 不匹配该规则的行，合并到上一条
      what => "previous"  # 将续行追加到上一条日志末尾
      auto_flush_interval => 5  # 5秒自动刷新未完成的日志条目，避免卡顿
    }
  }
}

场景2：异常续行有固定特征（如Java堆栈的at 、Caused by:）

input {
  file {
    path => "/path/to/your/logs/*.log"
    codec => multiline {
      # 匹配异常续行的特征
      pattern => "^(\s+at |Caused by:)"
      negate => false  # 匹配该规则的行，合并到上一条
      what => "previous"
    }
  }
}

2. Grok表达式适配合并后的日志 #

合并后的多行日志已成为一个完整条目，直接用Grok匹配单条日志格式即可，用GREEDYDATA字段捕获包含换行的异常堆栈：

filter {
  grok {
    match => { "message" => "%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:loglevel} %{DATA:class} - %{GREEDYDATA:content}" }
  }
}

其中GREEDYDATA会将日志正文+异常堆栈全部捕获到content字段中。

3. 常见错误排查 #

• 不要同时在input和filter中使用multiline插件，会导致重复合并或识别混乱
• 确保pattern精准匹配新日志的起始行，比如日志开头有空格的话，要在pattern中加入^\s*
• 先通过stdout输出测试multiline合并效果，确认日志条目拆分正确后，再添加Grok过滤

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