Logstash日志子消息解析失败：配置问题排查咨询

我有三类日志消息，已通过Logstash的Grok插件解析出时间戳、级别、来源及日志消息字段，但部分包含子消息（如"[11:45:29 INF] Generation finished"）的日志无法成功解析。

日志样例 #

2024-07-22 11:45:29.125 +02:00 [Information] [EventService] Events generation finished
2024-07-22 11:45:29.125 +02:00 [Information] [Process] Result: "[11:45:29 INF] Generation finished"
2024-07-22 11:45:29.267 +02:00 [Information] [] Worker is stopping gracefully

原Logstash配置 #

input {
  file {
    path => "/data/logs/**/*.log"
    start_position => "beginning"
    sincedb_path => "/dev/null"
  }
}

filter {
  grok {
    match => {
      "message" => [
        "%{TIMESTAMP_ISO8601:event_time} %{ISO8601_TIMEZONE:timezone} \[%{WORD:level}\] \[%{GREEDYDATA:source}\] %{GREEDYDATA:log_message}"
      ]
    }
  }
  
  # Additional grok pattern to handle sub-message if present
  grok {
    match => {
      "log_message" => [
        '(?m)%{QUOTEDSTRING:sub_message}"(?: %{GREEDYDATA:remainder})?'
      ]
    }
    tag_on_failure => []
  }

  # Parse the sub-message if present
  grok {
    match => {
      "sub_message" => [
        '\[%{TIME:sub_time} %{WORD:sub_level}\] %{GREEDYDATA:sub_log_message}'
      ]
    }
    tag_on_failure => []
  }

  if "_grokparsefailure" in [tags] {
    drop { }
  }

  mutate {
    add_field => { "timestamp" => "%{event_time} %{timezone}" }
  }
  date {
    match => ["timestamp", "yyyy-MM-dd HH:mm:ss.SSS Z"]
    timezone => "UTC"
    target => "@timestamp"
  }
  mutate {
    remove_field => [ "timestamp", "event_time", "timezone" ]
  }
}

output {
  elasticsearch {
    hosts => ["http://elasticsearch:9200"]
    index => "logs-%{+YYYY.MM.dd}"
  }
  stdout {
    codec => rubydebug
  }
}

问题分析 #

• 第一个Grok的source字段匹配错误：使用GREEDYDATA会贪婪匹配到后续的]及之后的内容，导致log_message字段无法正确获取完整日志内容。应改为匹配[和]之间的内容（允许为空）。
• 子消息匹配的Grok模式错误：QUOTEDSTRING已包含前后双引号，模式末尾多余的"会导致匹配失败；同时原模式中的(?m)多行模式无需启用，子消息都在单行内。
• 无差别解析子消息：不管sub_message是否存在，第三个Grok都会尝试解析，做无效操作，应添加条件判断仅在子消息存在时执行解析。

修正后的配置 #

input {
  file {
    path => "/data/logs/**/*.log"
    start_position => "beginning"
    sincedb_path => "/dev/null"
  }
}

filter {
  grok {
    match => {
      "message" => [
        "%{TIMESTAMP_ISO8601:event_time} %{ISO8601_TIMEZONE:timezone} \[%{WORD:level}\] \[(?<source>[^\]]*)\] %{GREEDYDATA:log_message}"
      ]
    }
  }
  
  # 匹配日志消息中的带引号子消息
  grok {
    match => {
      "log_message" => [
        '%{QUOTEDSTRING:sub_message}(?: %{GREEDYDATA:remainder})?'
      ]
    }
    tag_on_failure => []
  }

  # 仅当sub_message存在时解析子消息内容
  if [sub_message] {
    grok {
      match => {
        "sub_message" => [
          '\[%{TIME:sub_time} %{WORD:sub_level}\] %{GREEDYDATA:sub_log_message}'
        ]
      }
      tag_on_failure => []
    }
    # 移除sub_message首尾的引号
    mutate {
      gsub => [ "sub_message", '^"|"$', '' ]
    }
  }

  if "_grokparsefailure" in [tags] {
    drop { }
  }

  mutate {
    add_field => { "timestamp" => "%{event_time} %{timezone}" }
  }
  date {
    match => ["timestamp", "yyyy-MM-dd HH:mm:ss.SSS Z"]
    timezone => "UTC"
    target => "@timestamp"
  }
  mutate {
    remove_field => [ "timestamp", "event_time", "timezone" ]
  }
}

output {
  elasticsearch {
    hosts => ["http://elasticsearch:9200"]
    index => "logs-%{+YYYY.MM.dd}"
  }
  stdout {
    codec => rubydebug
  }
}

修正说明 #

1. 修复source字段匹配：用\[(?<source>[^\]]*)\]精准匹配[和]之间的所有内容（包括空值），确保log_message能正确获取后续内容。
2. 简化子消息匹配模式：移除多余的双引号和无效的多行模式，让QUOTEDSTRING正确匹配带引号的子消息。
3. 条件解析子消息：添加if [sub_message]判断，仅当存在子消息时才执行解析，避免无效操作。
4. 清理子消息引号：通过mutate的gsub移除sub_message首尾的双引号，让子消息内容更干净。

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