Logstash双层JSON解析异常及_jsonparsefailure错误排查

问题背景 #

使用Logstash处理日志时，message字段出现双重JSON序列化问题，需要将其还原为单层JSON格式。尝试添加json filter的source配置后，出现_jsonparsefailure标签，导致日志解析失败。

现有配置 #

input {
  file {
    id => "my_lt_log"
    path => "/logs/logtransformer.log"
    type => "log"
    start_position => "beginning"
  }
}
filter {
if [type] == "log" {
        mutate {
            remove_field => [ "kubernetes"]
        }
        mutate {
          gsub => [ "message", "(\W)at(\W)", '\1""\2' ]
        }
        if [message][metadata][proc_id] {
            mutate {
              add_field => { "[metadata][proc_id]" => "%{[message][metadata][proc_id]}" }
            }
            }
        if "_jsonparsefailure" in [tags] {
          mutate {
            add_field => {
              "logplane" => "adp-app-logs"
              "abc" => "%{[message]}"
            }
            remove_field => [ "message", "kubernetes" ]
          }
        }
        else {
          mutate {
              rename => {
                "path" => "filename"
              }
              add_field => {
                "def" => "%{[message]}"
                
                "message" => "%{[message][message]}"
                "timestamp" => "%{[message][timestamp]}"
              }
            }
            
      }
}
output {
...
}

输入输出样例 #

• 输入文件样例：

{"version": "1.1.0", "timestamp": "2023-02-06T13:42:59.634Z", "severity": "info", "service_id": "eric-log-transformer", "metadata" : {"namespace": "roshan", "pod_name": "eric-log-transformer-5bc84c4cb-c8qtw", "node_name": "node-10-63-142-135", "pod_uid": "00632d7b-d151-4b0c-84fe-8a6ee6b64b35", "container_name": "logtransformer"}, "message": "Starting pipeline {:pipeline_id=>'logstash', 'pipeline.workers'=>2, 'pipeline.batch.size'=>2048, 'pipeline.batch.delay'=>50, 'pipeline.max_inflight'=>4096, 'pipeline.sources'=>['/opt/logstash/resource/logstash.conf'], :thread=>'#<Thread:0x3d8b1518 run>'}"}

• 当前输出（def字段为序列化字符串）：

{
        "_index" : "adp-app-logs-2023.02.03",
        "_type" : "_doc",
        "_id" : "PpiDF4YBCMtUNdxoMJFW",
        "_score" : 0.79323065,
        "_source" : {
          "@version" : "1",
          "service_id" : "%{[message][service_id]}",
          "def" : "{\"version\": \"1.1.0\", \"timestamp\": \"2023-02-03T13:41:43.034Z\", \"severity\": \"info\", \"service_id\": \"eric-log-transformer\", \"metadata\" : {\"namespace\": \"zyadros\", \"pod_name\": \"eric-log-transformer-7b64896976-s6h5r\", \"node_name\": \"node-10-63-142-135\", \"pod_uid\": \"336c9706-41a9-41c0-b459-2eb4e9f6e2b4\", \"container_name\": \"logtransformer\"}, \"message\": \"Starting pipeline {:pipeline_id=>'opensearch', 'pipeline.workers'=>2, 'pipeline.batch.size'=>2048, 'pipeline.batch.delay'=>50, 'pipeline.max_inflight'=>4096, 'pipeline.sources'=>['/opt/logstash/resource/searchengine.conf'], :thread=>'#<Thread:0x7649ae47 run>'}\"}",
          "@timestamp" : "2023-02-03T13:41:58.044976Z",
          "severity" : "%{[message][severity]}"
        }
      }

• 期望def字段输出：

{"version": "1.1.0", "timestamp": "2023-02-06T06:18:33.647Z", "severity": "info", "service_id": "eric-log-transformer", "metadata" : {"namespace": "zyadros", "pod_name": "eric-log-transformer-5cb7dbc6b5-ghrsc", "node_name": "node-10-63-142-135", "pod_uid": "52b8e6fe-9547-4091-9034-36e1141f4391", "container_name": "logtransformer"}, "message": "Starting tcp input listener {:address=>'0.0.0.0:5015', :ssl_enable=>true}"}

错误日志（添加json {source}后） #

{
        "_index" : "%{logplane}-2023.02.07",
        "_type" : "_doc",
        "_id" : "9d6xKoYBCoUR1nQuuhlp",
        "_score" : 2.969562E-4,
        "_source" : {
          "path" : "/logs/logtransformer.log",
          "@version" : "1",
          "tags" : [
            "_jsonparsefailure"
          ],
          "@timestamp" : "2023-02-07T07:05:22.465556Z",
          "host" : "eric-log-transformer-d6dddd6f9-lp6d7",
          "message" : " at [Source: (byte[])' at [Source: (byte[])' at [Source: (byte[])'{'version': '1.1.0', 'timestamp': '2023-02-07T07:05:06.163Z', 'severity': 'warning', 'service_id': 'eric-log-transformer', 'metadata' : {'namespace': 'zyadros', 'pod_name': 'eric-log-transformer-d6dddd6f9-lp6d7', 'node_name': 'node-10-63-142-138', 'pod_uid': '02659e5c-c9ac-49c8-a4bd-74b9477e846d', 'container_name': 'logtransformer'}, 'message': 'Error parsing json {:source=>'message', :raw=>'{\\'version\\': \\'1.1.0\\', \\'timestamp\\': \\'2023-02-07T07:05:02'[truncated 68 bytes]; line: 1, column: 5]>}\"}"
        }
      }

问题分析 #

1. 解析顺序错误：先执行gsub修改message字段，破坏了原始JSON的结构，导致后续json filter无法正确解析。
2. 字段操作逻辑混乱：在未解析JSON的情况下直接引用[message][metadata][proc_id]，导致字段引用失效，同时条件判断顺序不合理，触发错误分支。

解决方案 #

修正步骤 #

1. 优先解析原始JSON：在任何字段修改前，先使用json filter解析原始message字段，将其转换为结构化数据。
2. 调整gsub作用对象：将gsub操作从原始message字段改为解析后的内部[message]字段，避免破坏顶层JSON结构。
3. 优化条件判断顺序：先处理解析成功的逻辑，再处理解析失败的异常分支，确保逻辑流转正确。

修正后的完整配置 #

input {
  file {
    id => "my_lt_log"
    path => "/logs/logtransformer.log"
    type => "log"
    start_position => "beginning"
  }
}
filter {
  if [type] == "log" {
    # 第一步：先解析原始message字段的JSON
    json {
      source => "message"
      remove_field => "message" # 解析后移除原始字符串，避免重复
    }

    mutate {
      remove_field => [ "kubernetes" ]
    }

    # 第二步：修改解析后的内部message字段，不破坏顶层结构
    mutate {
      gsub => [ "[message]", "(\W)at(\W)", '\1""\2' ]
    }

    # 处理proc_id字段
    if [metadata][proc_id] {
      mutate {
        add_field => { "[metadata][proc_id]" => "%{[metadata][proc_id]}" }
      }
    }

    # 正常解析分支：处理字段重命名和赋值
    mutate {
      rename => {
        "path" => "filename"
      }
      add_field => {
        "def" => "%{@metadata[_json_input]}" # 保存原始JSON字符串到def字段
        "message" => "%{[message]}"
        "timestamp" => "%{[timestamp]}"
        "service_id" => "%{[service_id]}"
        "severity" => "%{[severity]}"
      }
    }

    # 解析失败分支：捕获_jsonparsefailure标签
    if "_jsonparsefailure" in [tags] {
      mutate {
        add_field => {
          "logplane" => "adp-app-logs"
          "abc" => "%{[message]}"
        }
        remove_field => [ "message", "kubernetes" ]
      }
    }
  }
}
output {
  # 保留原输出配置
  ...
}

关键说明 #

• json filter前置：确保在修改字段前完成JSON解析，避免破坏原始结构。
• @metadata[_json_input]用法：json filter会自动将原始输入字符串保存到@metadata[_json_input]，直接引用该字段可以得到未修改的原始JSON字符串，满足def字段的期望输出。
• 调整gsub目标：将操作目标改为[message]（解析后的内部消息字段），避免影响顶层JSON结构。

效果验证 #

修正后，def字段将输出原始的单层JSON字符串，message字段会被替换为解析后的日志内容，同时不会出现_jsonparsefailure错误标签，所有字段引用（如service_id、severity）都能正确解析赋值。

内容的提问来源于stack exchange，提问作者Roshan Yadav