FileBeat无法解码JSON消息：日志JSON字段解析失败求助

尝试通过FileBeat结合Kafka将日志发送至Elasticsearch，计划在发送前将message字段中的JSON字符串解码为独立字段，但在Kibana中查看时解析操作未生效，无法拆分出独立字段。

示例日志消息 #

{
  "@timestamp": [
    "2023-04-28T07:27:25.682Z"
  ],
  "@version": [
    "1"
  ],
  "@version.keyword": [
    "1"
  ],
  "event.original": [
    "{\"@timestamp\":\"2023-04-28T07:27:24.678Z\",\"@metadata\":{\"beat\":\"filebeat\",\"type\":\"_doc\",\"version\":\"8.7.0\"},\"message\":\"[Apr 28, 2023 00:27:24,080][INFO][pool-28-thread-6][CacheDataReaderImpl()-()] - Retrieving Cache Data of table: MYSQL_SYSTEM_PARAMS with WHERE CLAUSE: [param_id|EMAIL_INVALID_JSON] , from Instance: FMI , for Group Id: GRP_AI\",\"log\":{\"offset\":3691343,\"file\":{\"path\":\"/y/mcpdata/AIRestfulService/oltpj-cache-layer.log\"}}}"
  ],
  "event.original.keyword": [
    "{\"@timestamp\":\"2023-04-28T07:27:24.678Z\",\"@metadata\":{\"beat\":\"filebeat\",\"type\":\"_doc\",\"version\":\"8.7.0\"},\"message\":\"[Apr 28, 2023 00:27:24,080][INFO][pool-28-thread-6][CacheDataReaderImpl()-()] - Retrieving Cache Data of table: MYSQL_SYSTEM_PARAMS with WHERE CLAUSE: [param_id|EMAIL_INVALID_JSON] , from Instance: FMI , for Group Id: GRP_AI\",\"log\":{\"offset\":3691343,\"file\":{\"path\":\"/y/mcpdata/AIRestfulService/oltpj-cache-layer.log\"}}}"
  ],
  "message": [
    "{\"@timestamp\":\"2023-04-28T07:27:24.678Z\",\"@metadata\":{\"beat\":\"filebeat\",\"type\":\"_doc\",\"version\":\"8.7.0\"},\"message\":\"[Apr 28, 2023 00:27:24,080][INFO][pool-28-thread-6][CacheDataReaderImpl()-()] - Retrieving Cache Data of table: MYSQL_SYSTEM_PARAMS with WHERE CLAUSE: [param_id|EMAIL_INVALID_JSON] , from Instance: FMI , for Group Id: GRP_AI\",\"log\":{\"offset\":3691343,\"file\":{\"path\":\"/y/mcpdata/AIRestfulService/oltpj-cache-layer.log\"}}}"
  ],
  "message.keyword": [
    "{\"@timestamp\":\"2023-04-28T07:27:24.678Z\",\"@metadata\":{\"beat\":\"filebeat\",\"type\":\"_doc\",\"version\":\"8.7.0\"},\"message\":\"[Apr 28, 2023 00:27:24,080][INFO][pool-28-thread-6][CacheDataReaderImpl()-()] - Retrieving Cache Data of table: MYSQL_SYSTEM_PARAMS with WHERE CLAUSE: [param_id|EMAIL_INVALID_JSON] , from Instance: FMI , for Group Id: GRP_AI\",\"log\":{\"offset\":3691343,\"file\":{\"path\":\"/y/mcpdata/AIRestfulService/oltpj-cache-layer.log\"}}}"
  ],
  "_id": "087CxocBQy_JTnXHdv8o",
  "_index": "applogs_ai_rest-2023-04-28",
  "_score": null
}

当前FileBeat配置 #

filebeat.config.modules.path: /etc/filebeat/modules.d/*.yml
name: filebeat-AI-logs
logging.level: info
logging.to_files: true
logging.files:
  path: /var/log/filebeat
  name: filebeat
  keepfiles: 7
  permissions: 0644

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /y/mcpdata/AIRestfulService/*.log
  multiline.pattern: '^\[\w{3} \d{1,2}, \d{4} \d{2}:\d{2}:\d{2},\d{3}\]'
  multiline.negate: true
  multiline.match: after
processors:
  - drop_fields:
      fields: ["@timestamp","@metadata", "host", "offset", "agent", "ecs", "input", "service", "tags"]
      ignore_missing: true
  - decode_json_fields:
      fields: ["message"]
      process_array: false
      max_depth: 5
      target: "message"
      overwrite_keys: true
      add_error_key: false
output.kafka:
  hosts: ["***.***.***.***:****"]
  topic: 'applogs_ai_rest'
  required_acks: 1
  compression: gzip
  max_message_bytes: 100000
  bulk_max_size: 2048
  flush_interval: 500ms
#output.console:
# pretty: true

解决方案 #

1. 调整decode_json_fields的target配置 #

当前配置target: "message"会将解码后的JSON嵌套在原message字段内部，而非展开到根级别。若要让解码后的字段成为独立的根字段，需删除target配置（默认行为就是将字段展开到根级别），或设置target: ""。

2. 调整处理器执行顺序 #

当前先执行drop_fields删除了@timestamp，但解码后的JSON中包含业务日志的@timestamp，建议先解码再处理字段，避免提前删除需要保留的字段。

3. 开启错误排查机制 #

将add_error_key设为true，若解码失败，FileBeat会添加error.message字段，方便排查JSON格式是否合法。

4. 验证处理结果 #

临时启用output.console并开启pretty: true，直接查看FileBeat处理后的输出，确认解码逻辑是否生效。

修改后的FileBeat配置示例 #

filebeat.config.modules.path: /etc/filebeat/modules.d/*.yml
name: filebeat-AI-logs
logging.level: info
logging.to_files: true
logging.files:
  path: /var/log/filebeat
  name: filebeat
  keepfiles: 7
  permissions: 0644

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /y/mcpdata/AIRestfulService/*.log
  multiline.pattern: '^\[\w{3} \d{1,2}, \d{4} \d{2}:\d{2}:\d{2},\d{3}\]'
  multiline.negate: true
  multiline.match: after
processors:
  # 先解码JSON，再处理字段
  - decode_json_fields:
      fields: ["message"]
      process_array: false
      max_depth: 5
      # 移除target配置，将解码字段展开到根级别
      overwrite_keys: true
      add_error_key: true # 开启错误键，便于排查解码失败问题
  - drop_fields:
      fields: ["@metadata", "host", "offset", "agent", "ecs", "input", "service", "tags"]
      ignore_missing: true
output.kafka:
  hosts: ["***.***.***.***:****"]
  topic: 'applogs_ai_rest'
  required_acks: 1
  compression: gzip
  max_message_bytes: 100000
  bulk_max_size: 2048
  flush_interval: 500ms
# 临时启用控制台输出验证处理结果
#output.console:
#  pretty: true

内容的提问来源于stack exchange，提问作者Obaid Ur Rehman