How to index json output data from docker containers into elastic search

[zerodefect]

I have a set of docker containers running in a local stack (via docker-compose) that are outputting JSON logs that look like the following:

{"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"}

This is a custom JSON output format.

The fluentd log shows the log it receives:

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"}

The corresponding fluentd conf is the following:

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

<filter docker.**>
  @type record_transformer
  renew_record true
  keep_keys log
  enable_ruby true
  <record>
    message ${record["msg"].to_json}
  </record>
</filter>

<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 password
    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>

The problems I'm trying to resolve:

• None of the fields in 'log' are indexable via Kibana/Elastic Search
• Transform log["msg"] => message (I want log["msg"] to replace message").

Any tips would be much appreciated.