如何在Kibana的fluentd数据视图中转换message字段以显示其JSON内的msg值？

版本信息 #

Elasticsearch / Kibana:  8.6.0
AWS EKS k8s:  1.24
k8s logging operator:  ghcr.io/banzaicloud/logging-operator:3.17.10

问题说明 #

日志栈架构为服务日志 → Fluentd → Elasticsearch → Kibana，当前使用索引过滤器fluentd*的数据视图。message字段存储的是完整JSON对象（包含file、func、level、msg、time等属性），需要把message字段直接替换为JSON内msg对应的字符串内容。

解决方案 #

方案1：在Fluentd层面处理（建议优先采用）

在日志传输阶段完成字段转换，减少Elasticsearch的存储压力和后续处理开销。通过Banzai Cloud Logging Operator配置Fluentd的解析与字段重写：

修改Elasticsearch输出的CR配置：

apiVersion: logging.banzaicloud.io/v1beta1
kind: Output
metadata:
  name: elasticsearch-output
spec:
  elasticsearch:
    host: elasticsearch-cluster
    port: 9200
    index_name: fluentd-%Y%m%d
    # 第一步：解析message字段的JSON内容
    filter:
      parser:
        key_name: message
        reserve_data: false  # 不需要保留原JSON其他字段时设为false，需要保留则改为true
        remove_key_name_field: true
        parse:
          type: json
    # 第二步：将解析出的msg值覆盖原message字段
    filter:
      record_transformer:
        enable_ruby: true
        record:
          message: "${record['msg']}"
          # 若需保留原JSON的其他字段，可在此追加，例如：level: "${record['level']}"

方案2：在Elasticsearch层面用Ingest Pipeline处理

如果无法调整Fluentd配置，可通过Elasticsearch的Ingest Pipeline在数据写入时完成转换：

1. 创建处理管道：

PUT _ingest/pipeline/rewrite-message-field
{
  "description": "把message字段替换为内部msg属性值",
  "processors": [
    {
      "json": {
        "field": "message",
        "target_field": "message_parsed"
      }
    },
    {
      "set": {
        "field": "message",
        "value": "{{message_parsed.msg}}"
      }
    },
    {
      "remove": {
        "field": "message_parsed"
      }
    }
  ]
}

2. 关联到索引模板：
   修改fluentd*对应的索引模板，设置index.default_pipeline: "rewrite-message-field"，确保新写入的日志自动应用该转换规则。

验证 #

配置生效后，新写入的日志在Kibana数据视图中，message字段将直接显示msg对应的字符串内容。

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