如何在Filebeat配置中规范化发送至Kafka的JSON日志

要实现将message字段中的嵌套JSON（含单引号格式的RequestBody）转换为完全结构化的嵌套JSON，可通过Filebeat的processors链完成，核心是先修复非标准JSON格式，再逐层解析嵌套结构。

配置步骤与示例 #

以下是完整的Filebeat配置片段，包含关键处理器的说明：

filebeat.inputs:
  - type: log
    paths:
      - /path/to/your/target/logs/*.log

processors:
  # 1. 修复非标准JSON：将所有单引号替换为双引号，转为标准JSON格式
  - replace:
      field: message
      pattern: "'"
      replacement: '"'
      ignore_missing: true
      fail_on_error: false

  # 2. 解析外层message为结构化JSON
  - decode_json_fields:
      fields: ["message"]
      target: ""  # 解析后字段直接放在根层级，如需单独归类可改为"parsed_log"
      overwrite_keys: true
      ignore_missing: true
      fail_on_error: false

  # 3. 解析RequestBody字段的嵌套JSON字符串
  - decode_json_fields:
      fields: ["RequestBody"]
      target: "RequestBody"  # 解析后嵌套在原RequestBody字段下
      overwrite_keys: true
      ignore_missing: true
      fail_on_error: false

  # 4. 解析Result字段的嵌套JSON字符串
  - decode_json_fields:
      fields: ["Result"]
      target: "Result"  # 解析后嵌套在原Result字段下
      overwrite_keys: true
      ignore_missing: true
      fail_on_error: false

output.kafka:
  hosts: ["your-kafka-broker:9092"]
  topic: "your-target-topic"

关键细节说明 #

1. 单引号替换：
   标准JSON仅支持双引号，因此先将message中所有单引号替换为双引号，确保后续解析能正常进行。若日志中存在其他非JSON用途的单引号，可改用精准正则匹配仅替换RequestBody/Result字段内的单引号：

   - replace:
       field: message
       pattern: '(?<=("|\')RequestBody("|\'):("|\'))(.*?)(?=("|\))'
       replacement: '${0//\'/"}'
       flags: [DOTALL]
   

2. 逐层解析：
   先解析外层message得到结构化字段，再单独解析RequestBody和Result这两个字符串格式的JSON字段，最终生成嵌套的结构化JSON。

3. 容错配置：
   ignore_missing和fail_on_error设为false，确保个别异常日志不会中断整个Filebeat的日志处理流程。

内容的提问来源于stack exchange，提问作者Vishal Srivastava