Logstash无法解析键值对间含空格的JSON日志问题求助

问题描述 #

日志中包含JSON对象，当JSON键值对为"key":"value"或"key" :"value"格式时可正常处理，但为"key": "value"（冒号后带空格）时无法正常解析。

当前Logstash配置 #

input {
  syslog {
    port => 3011
  }
}

filter {
  grok {
    match => { "message" =>
    [
      "%{SYSLOGTIMESTAMP:timestamp4} %{DATA:time_ms}|%{DATA:field1}|%{DATA:field2}|99|%{DATA:field3}|%{DATA:field4}|%{DATA:field5}|%{DATA:field6}|%{DATA:field7}|%{DATA:field8}|%{DATA:field9}|%{DATA:field10}|%{DATA:field11}|%{DATA:field12}|%{GREEDYDATA:field13}"
    ]
    }
  }
  date {
    match => ["timestamp4", "MMM dd HH:mm:ss"]
  }
  if [field13] {
    mutate {
      add_field => {"log_type" => "my-logs"}
    }
  }
}

output {
  if [log_type] == "my-logs" {
    stdout { codec => rubydebug }
    elasticsearch {
      hosts => ["ES_HOST:9200"]
      index => "my-logs-000001"
    }
  }
}

日志示例 #

可正常处理的日志 #

echo "Mar 21 13:27:11 11:11.366293|dataadwhw1|ebsmp4713user5_@maiator|99|4064|22|SUCCESS|data|19|UA101|10.1.1.70|https|data.com|{\"wrg_id\":\"200000337\"}|200" | nc localhost 3011
echo "Mar 21 13:27:11 11:11.366293|dataadwhw1|ebsmp4713user5_@maiator|99|4064|22|SUCCESS|data|19|UA101|10.1.1.70|https|data.com|{\"wrg_id\" :\"200000337\"}|200" | nc localhost 3011

无法处理的日志 #

echo "Mar 21 13:27:11 11:11.366293|dataadwhw1|ebsmp4713user5_@maiator|99|4064|22|SUCCESS|data|19|UA101|10.1.1.70|https|data.com|{\"wrg_id\": \"200000337\"}|200" | nc localhost 3011

原因分析 #

1. Grok模式匹配冗余：当前使用%{GREEDYDATA:field13}会将JSON字符串和最后的200一并纳入field13，后续若解析JSON结构，多余的|200会导致解析失败。
2. 缺少JSON结构化解析：配置仅提取了包含JSON的字段field13，未对其进行结构化转换。虽然JSON规范允许键值对冒号前后存在空格，但需通过专门的过滤器完成字符串到结构化数据的转换。

解决方案 #

步骤1：修正Grok模式，精准提取字段 #

将Grok模式最后一段改为%{DATA:field13}|%{DATA:field14}，单独提取JSON字符串和状态码：

grok {
  match => { "message" =>
  [
    "%{SYSLOGTIMESTAMP:timestamp4} %{DATA:time_ms}|%{DATA:field1}|%{DATA:field2}|99|%{DATA:field3}|%{DATA:field4}|%{DATA:field5}|%{DATA:field6}|%{DATA:field7}|%{DATA:field8}|%{DATA:field9}|%{DATA:field10}|%{DATA:field11}|%{DATA:field12}|%{DATA:field13}|%{DATA:field14}"
  ]
  }
}

%{DATA:field13}匹配|之间的JSON字符串，%{DATA:field14}匹配最后的200，避免冗余内容干扰后续解析。

步骤2：添加JSON过滤器解析结构化数据 #

在mutate过滤器前添加json过滤器，将field13的JSON字符串转换为结构化字段：

if [field13] {
  json {
    source => "field13"
    target => "json_data" # 可选：指定解析后数据的存储字段，不指定则直接合并到根字段
  }
  mutate {
    add_field => {"log_type" => "my-logs"}
  }
}

该过滤器会自动兼容JSON规范中的空格格式，无论冒号前后是否有空格都能正常解析。

完整修正后的配置 #

input {
  syslog {
    port => 3011
  }
}

filter {
  grok {
    match => { "message" =>
    [
      "%{SYSLOGTIMESTAMP:timestamp4} %{DATA:time_ms}|%{DATA:field1}|%{DATA:field2}|99|%{DATA:field3}|%{DATA:field4}|%{DATA:field5}|%{DATA:field6}|%{DATA:field7}|%{DATA:field8}|%{DATA:field9}|%{DATA:field10}|%{DATA:field11}|%{DATA:field12}|%{DATA:field13}|%{DATA:field14}"
    ]
    }
  }
  date {
    match => ["timestamp4", "MMM dd HH:mm:ss"]
  }
  if [field13] {
    json {
      source => "field13"
      target => "json_data"
    }
    mutate {
      add_field => {"log_type" => "my-logs"}
    }
  }
}

output {
  if [log_type] == "my-logs" {
    stdout { codec => rubydebug }
    elasticsearch {
      hosts => ["ES_HOST:9200"]
      index => "my-logs-000001"
    }
  }
}

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