Elastic摄入管道中JSON日志字段覆盖失败问题排查

在OpenShift上运行多个Spring Boot应用，需要将日志发送至Elastic平台。已在集群部署Elastic Agent，通过托管摄入管道处理日志，且按照Elastic官方指南配置Logback输出ECS格式日志。Agent已将日志解析为消息中的JSON对象，希望用JSON对象中的值覆盖对应常规字段，为此配置了指定的摄入管道，但仅json.message覆盖message生效，json.service.name覆盖service.name、json.log.level覆盖log.level均失败，Kibana中对应字段为空，尝试过点分隔处理器也未解决问题。

日志示例 #

{"@timestamp":"2023-10-02T10:17:47.798Z","log.level": "INFO","message":"Starting Application v1.0.0 using Java 17.0.8 on my-component-15-7hbhs with PID 1 (/deployments/my-component-1.0.0.jar started by 1002400000 in /deployments)","ecs.version": "1.2.0","service.name":"my-component","event.dataset":"my-component","process.thread.name":"main","log.logger":"com.example.Application"}

当前摄入管道配置 #

[
  {
    "set": {
      "field": "log.level",
      "ignore_empty_value": true,
      "copy_from": "json.log.level"
    }
  },
  {
    "set": {
      "field": "message",
      "copy_from": "json.message",
      "ignore_empty_value": true
    }
  },
  {
    "set": {
      "field": "service.name",
      "ignore_empty_value": true,
      "copy_from": "json.service.name"
    }
  },
  {
    "set": {
      "field": "service.version",
      "copy_from": "kubernetes.labels.app_kubernetes_io/version",
      "ignore_empty_value": true
    }
  },
  {
    "pipeline": {
      "name": "logs-kubernetes.container_logs@custom",
      "ignore_missing_pipeline": true
    }
  },
  {
    "reroute": {
      "tag": "kubernetes.container_logs",
      "dataset": [
        "{{kubernetes.annotations.elastic_co/dataset}}",
        "{{data_stream.dataset}}"
      ],
      "namespace": [
        "{{kubernetes.annotations.elastic_co/namespace}}",
        "{{data_stream.namespace}}"
      ],
      "if": "ctx.kubernetes?.annotations != null"
    }
  }
]

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

当前配置中，自定义set处理器先执行，随后调用的logs-kubernetes.container_logs@custom托管管道可能会覆盖你设置的log.level和service.name字段。把自定义set处理器移到托管管道调用之后，确保自定义覆盖逻辑最后执行：

[
  {
    "pipeline": {
      "name": "logs-kubernetes.container_logs@custom",
      "ignore_missing_pipeline": true
    }
  },
  {
    "set": {
      "field": "log.level",
      "ignore_empty_value": true,
      "copy_from": "json.log.level"
    }
  },
  {
    "set": {
      "field": "message",
      "copy_from": "json.message",
      "ignore_empty_value": true
    }
  },
  {
    "set": {
      "field": "service.name",
      "ignore_empty_value": true,
      "copy_from": "json.service.name"
    }
  },
  {
    "set": {
      "field": "service.version",
      "copy_from": "kubernetes.labels.app_kubernetes_io/version",
      "ignore_empty_value": true
    }
  },
  {
    "reroute": {
      "tag": "kubernetes.container_logs",
      "dataset": [
        "{{kubernetes.annotations.elastic_co/dataset}}",
        "{{data_stream.dataset}}"
      ],
      "namespace": [
        "{{kubernetes.annotations.elastic_co/namespace}}",
        "{{data_stream.namespace}}"
      ],
      "if": "ctx.kubernetes?.annotations != null"
    }
  }
]

2. 直接解析JSON到根字段 #

既然Logback已经输出标准ECS格式JSON，无需先解析到json嵌套对象再复制。修改Elastic Agent配置，让json处理器直接将日志解析到文档根字段：

processors:
  - json:
      target: ""
      add_error_key: true

这样日志中的log.level、service.name等字段会直接写入文档根路径，跳过中间复制步骤。

3. 验证管道执行过程 #

使用Kibana的Ingest Pipeline Simulator（路径：Stack Management -> Ingest Pipelines -> Simulate），输入原始日志文档，测试摄入管道的每一步执行结果，定位字段覆盖失败的具体环节。

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