Docker自定义JSON日志转Elasticsearch的Kibana检索配置问询

我将部署若干集成了Golang编写的通用日志框架的Docker容器，日志格式为JSON格式。该自定义JSON日志中包含特定数据，我希望这些数据能在Kibana中被索引和检索。我了解需要对这些数据进行转换/过滤，但即便查阅了文档仍不清楚具体操作方式——我需要从JSON中提取JSON吗？

以下是Docker日志中显示的最小示例应用输出：

{"app_name":"SampleApp","app_port":6666,"app_version":"0.0.2","file":"/build/examples/sample/app/runmain/main.go:131","func":"example.com/code/microservices/examples/sample/app/runmain.mainErr.func1","fw_version":"v0.0.1","level":"info","msg":"listening","time":"2024-01-18T20:31:39.163970213+07:00"}

数据传入Fluentd后记录为：

2024-01-18 20:31:39.000000000 +0000 f905b090d278: {"container_id":"f905b090d278ec2cc2f1f912acdbf8787a0a1c91d8ab7b00ad84e9da20c8c147","container_name":"/fervent_jemison","source":"stdout","log":"{\"app_name\":\"SampleApp\",\"app_port\":6666,\"app_version\":\"0.0.2\",\"file\":\"/build/examples/sample/app/runmain/main.go:131\",\"func\":\"example.com/code/microservices/examples/sample/app/runmain.mainErr.func1\",\"fw_version\":\"v0.0.1\",\"level\":\"info\",\"msg\":\"listening\",\"time\":\"2024-01-18T20:31:39.163970213+07:00\"}\r"}

当前使用的Fluentd配置如下：

<source>
  @type forward
  port 24224
  bind 0.0.0.0
</source>

<filter docker.**>
  @type parser
  key_name log
  reserve_data true
  <parse>
    @type json
  </parse>
</filter>

<match *.**>
  @type copy

  <store>
    @type elasticsearch
    host es
    port 9200
    user elastic
    password elastic
    logstash_format true
    logstash_prefix fluentd
    logstash_dateformat %Y%m%d
    include_tag_key true
    type_name access_log
    tag_key @log_name
    <buffer> 
      flush_interval 1s
    </buffer>
  </store>

  <store>
    @type stdout
  </store>
</match>

我最初会在单台主机上本地部署所有组件，希望能得到相关建议或进一步指导，这对我来说是全新的领域。

核心问题解析 #

你当前的场景确实需要从外层JSON的log字段中提取嵌套的JSON结构——Docker把应用输出的JSON日志当成字符串存到了log字段里，只有解析这个嵌套的JSON，把里面的app_name、level、time等字段单独提取出来，Elasticsearch才能正确索引这些字段，Kibana才能实现精准检索。

现有配置调整优化 #

你的Fluentd配置已经用到了parser过滤器，只需补充几个关键配置即可：

1. 修复标签匹配问题：
   确保docker.**能匹配到实际的Docker日志标签。如果Fluentd接收的日志标签不是以docker.开头（比如日志开头的f905b090d278是标签），可以把过滤器的匹配规则改成**，确保覆盖所有日志。
2. 处理冗余字符：
   注意到log字段末尾有\r，添加trim_whitespace true可以去掉多余空白字符，避免解析失败：

   <filter docker.**>
     @type parser
     key_name log
     reserve_data true
     trim_whitespace true
     <parse>
       @type json
     </parse>
   </filter>
   

3. 使用应用自身的时间戳：
   应用日志里的time字段带时区，比Docker转发的时间更准确，添加time_key和time_format让Fluentd用这个字段作为日志时间戳：

   <filter docker.**>
     @type parser
     key_name log
     reserve_data true
     trim_whitespace true
     time_key time
     time_format %Y-%m-%dT%H:%M:%S.%N%:z
     <parse>
       @type json
     </parse>
   </filter>
   

验证流程 #

1. 调整配置后重启Fluentd服务。
2. 查看Fluentd的stdout输出，确认log字段的嵌套JSON已被解析为独立顶级字段（比如输出中同时存在container_id、app_name、level等字段）。
3. 登录Kibana，创建匹配fluentd-*前缀的索引模式，在Discover页面即可看到所有提取出的字段，并能通过app_name、level等关键词检索日志。

额外优化建议 #

• 标签精细化管理：通过Docker Compose的logging配置给不同容器设置专属标签，这样可以在Fluentd中针对不同应用做差异化过滤处理。
• 清理冗余字段：用record_transformer过滤器删除不需要索引的字段，比如：

  <filter docker.**>
    @type record_transformer
    remove_keys log,func
  </filter>
  

• 错误排查：保留@type stdout的store，实时查看处理后的日志，若解析失败，可在Fluentd日志中找到错误信息排查JSON格式问题。

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