I am running Fluentd in GKE to get my containers' logs in Stackdriver using my customized Fluentd configuration file.

Some logs which are json objects in the container log files at /var/log/containers/, are having text-payload at Stackdriver. I want them to show up as json-payload.

When I use the built in Fluentd agent of Stackdriver (i.e. not using my customized Fluentd config file), those logs have json-payload. What in my config file could cause this problem?

Here is my config file:

<source>
  type tail
  format json
  time_key time
  path /var/log/containers/*.log
  pos_file /var/log/gcp-containers.log.pos
  time_format %Y-%m-%dT%H:%M:%S.%N%Z
  tag reform.*
  read_from_head true
</source>

 <filter reform.**>
  type parser
  format /^(?<time>\d{2}[^\s]*) : \[[^\]]*\] \[[^\]]*\] \[[^\]]*\] \[[^\]]*\] (?<severity>[^ ]*) (.*)/
  reserve_data true
  suppress_parse_error_log true
  key_name log
</filter>


<filter reform.**>
  type parser
  format /^(?<severity>\w)(?<time>\d{4} [^\s]*)\s+(?<pid>\d+)\s+(?<source>[^      \]]+)\] (?<log>.*)/
  reserve_data true
  suppress_parse_error_log true
  key_name log
</filter>
<filter reform.**>
  @type concat
  key log
  n_lines 1
</filter>

<filter reform.**>
  @type record_transformer
  enable_ruby true
  <record>
  log ${record["log"].gsub(/[0-9]{4} *[0-9]{4} *[0-9]{4} *[0-9]{4}/,"xxxx xxxx xxxx xxxx")}
  </record>
</filter>

<match reform.**>
  type record_reformer
  enable_ruby true
  tag raw.kubernetes.${tag_suffix[4].split('-')[0..-2].join('-')}
</match>                 
 # Detect exceptions in the log output and forward them as one log entry.
<match raw.kubernetes.**>
  @type copy

  <store>
    @type prometheus

    <metric>
      type counter
      name logging_line_count
      desc Total number of lines generated by application containers
    </metric>
  </store>
  <store>
    @type detect_exceptions

    remove_tag_prefix raw
    message log
    stream stream
    multiline_flush_interval 5
    max_bytes 500000
    max_lines 1000
  </store>
</match>

<source>
  type tail
  format /^(?<time>[^ ]* [^ ,]*)[^\[]*\[[^\]]*\]\[(?<severity>[^ \]]*) *\] (?<message>.*)$/
  time_format %Y-%m-%d %H:%M:%S
  path /var/log/salt/minion
  pos_file /var/log/gcp-salt.pos
  tag salt
</source>

# Example:
# Dec 21 23:17:22 gke-foo-1-1-4b5cbd14-node-4eoj startupscript: Finished running startup script /var/run/google.startup.script
<source>
  type tail
  format syslog
  path /var/log/startupscript.log
  pos_file /var/log/gcp-startupscript.log.pos
  tag startupscript
</source>                     
 # Examples:
# time="2016-02-04T06:51:03.053580605Z" level=info msg="GET /containers/json"
# time="2016-02-04T07:53:57.505612354Z" level=error msg="HTTP Error" err="No such image: -f" statusCode=404
<source>
  type tail
  format /^time="(?<time>[^)]*)" level=(?<severity>[^ ]*) msg="(?<message>[^"]*)"( err="(?<error>[^"]*)")?( statusCode=(
lt;status_code>\d+))?/
  path /var/log/docker.log
  pos_file /var/log/gcp-docker.log.pos
  tag docker
</source>

# Example:
# 2016/02/04 06:52:38 filePurge: successfully removed file /var/etcd/data/member/wal/00000000000006d0-00000000010a23d1.wal
<source>
  type tail
  # Not parsing this, because it doesn't have anything particularly useful to
  # parse out of it (like severities).
  format none
  path /var/log/etcd.log
  pos_file /var/log/gcp-etcd.log.pos
  tag etcd
</source>

# Example:
# I0204 07:32:30.020537    3368 server.go:1048] POST /stats/container/: (13.972191ms) 200 [[Go-http-client/1.1] 10.244.1.3:40537]
<source>
  type tail
  format multiline
  multiline_flush_interval 5s
  format_firstline /^\w\d{4}/
  format1 /^(?<severity>\w)(?<time>\d{4} [^\s]*)\s+(?<pid>\d+)\s+(?<source>[^ \]]+)\] (?<message>.*)/
  time_format %m%d %H:%M:%S.%N
  path /var/log/kubelet.log
  pos_file /var/log/gcp-kubelet.log.pos
  tag kubelet
</source>             
# Example:
# I1118 21:26:53.975789       6 proxier.go:1096] Port "nodePort for kube-system/default-http-backend:http" (:31429/tcp) was open before and is still needed
<source>
  type tail
  format multiline
  multiline_flush_interval 5s
  format_firstline /^\w\d{4}/
  format1 /^(?<severity>\w)(?<time>\d{4} [^\s]*)\s+(?<pid>\d+)\s+(?<source>[^ \]]+)\] (?<message>.*)/
  time_format %m%d %H:%M:%S.%N
  path /var/log/kube-proxy.log
  pos_file /var/log/gcp-kube-proxy.log.pos
  tag kube-proxy
</source>

# Example:
# I0204 07:00:19.604280       5 handlers.go:131] GET /api/v1/nodes: (1.624207ms) 200 [[kube-controller-manager/v1.1.3 (linux/amd64) kubernetes/6a81b50] 127.0.0.1:38266]
<source>
  type tail
  format multiline
  multiline_flush_interval 5s
  format_firstline /^\w\d{4}/
  format1 /^(?<severity>\w)(?<time>\d{4} [^\s]*)\s+(?<pid>\d+)\s+(?<source>[^ \]]+)\] (?<message>.*)/
  time_format %m%d %H:%M:%S.%N
  path /var/log/kube-apiserver.log
  pos_file /var/log/gcp-kube-apiserver.log.pos
  tag kube-apiserver
</source>

# Example:
# 2017-02-09T00:15:57.992775796Z AUDIT: id="90c73c7c-97d6-4b65-9461-f94606ff825f" ip="104.132.1.72" method="GET" user="kubecfg" as="<self>" asgroups="<lookup>" namespace="default" uri="/api/v1/namespaces/default/pods"
# 2017-02-09T00:15:57.993528822Z AUDIT: id="90c73c7c-97d6-4b65-9461-f94606ff825f" response="200"
<source>
  type tail
  format multiline
  multiline_flush_interval 5s
  format_firstline /^\S+\s+AUDIT:/
  # Fields must be explicitly captured by name to be parsed into the record.
  # Fields may not always be present, and order may change, so this just looks
  # for a list of key="\"quoted\" value" pairs separated by spaces.
  # Unknown fields are ignored.
  # Note: We can't separate query/response lines as format1/format2 because
  #       they don't always come one after the other for a given query.
  # TODO: Maybe add a JSON output mode to audit log so we can get rid of this?
  format1 /^(?<time>\S+) AUDIT:(?: (?:id="(?<id>(?:[^"\\]|\\.)*)"|ip="(?<ip>(?:[^"\\]|\\.)*)"|method="(?<method>(?:[^"\\]|\\.)*)"|user="(?<user>(?:[^"\\]|\\.)*)"|groups="(?<groups>(?:[^"\\]|\\.)*)"|as="(?<as>(?:[^"\\]|\\.)*)"|asgroups="(?<asgroups>(?:[^"\\]|\\.)*)"|namespace="(?<namespace>(?:[^"\\]|\\.)*)"|uri="(?<uri>(?:[^"\\]|\\.)*)"|response="(?<response>(?:[^"\\]|\\.)*)"|\w+="(?:[^"\\]|\\.)*"))*/
  time_format %FT%T.%L%Z
  path /var/log/kube-apiserver-audit.log
  pos_file /var/log/gcp-kube-apiserver-audit.log.pos
  tag kube-apiserver-audit
</source>

# Example:
# I0204 06:55:31.872680       5 servicecontroller.go:277] LB already exists and doesn't need update for service kube-system/kube-ui
<source>
  type tail
  format multiline
  multiline_flush_interval 5s
  format_firstline /^\w\d{4}/
  format1 /^(?<severity>\w)(?<time>\d{4} [^\s]*)\s+(?<pid>\d+)\s+(?<source>[^ \]]+)\] (?<message>.*)/
  time_format %m%d %H:%M:%S.%N
  path /var/log/kube-controller-manager.log
  pos_file /var/log/gcp-kube-controller-manager.log.pos
  tag kube-controller-manager
</source>
<source>
  type tail
  format multiline
  multiline_flush_interval 5s
  format_firstline /^\w\d{4}/
  format1 /^(?<severity>\w)(?<time>\d{4} [^\s]*)\s+(?<pid>\d+)\s+(?<source>[^ \]]+)\] (?<message>.*)/
  time_format %m%d %H:%M:%S.%N
  path /var/log/kube-scheduler.log
  pos_file /var/log/gcp-kube-scheduler.log.pos
  tag kube-scheduler
</source>

# Example:
# I1104 10:36:20.242766       5 rescheduler.go:73] Running Rescheduler
<source>
  type tail
  format multiline
  multiline_flush_interval 5s
  format_firstline /^\w\d{4}/
  format1 /^(?<severity>\w)(?<time>\d{4} [^\s]*)\s+(?<pid>\d+)\s+(?<source>[^ \]]+)\] (?<message>.*)/
  time_format %m%d %H:%M:%S.%N
  path /var/log/rescheduler.log
  pos_file /var/log/gcp-rescheduler.log.pos
  tag rescheduler
</source>

# Example:
# I0603 15:31:05.793605       6 cluster_manager.go:230] Reading config from path /etc/gce.conf
<source>
  type tail
  format multiline
  multiline_flush_interval 5s
  format_firstline /^\w\d{4}/
  format1 /^(?<severity>\w)(?<time>\d{4} [^\s]*)\s+(?<pid>\d+)\s+(?<source>[^ \]]+)\] (?<message>.*)/
  time_format %m%d %H:%M:%S.%N
  path /var/log/glbc.log
  pos_file /var/log/gcp-glbc.log.pos
  tag glbc
</source>

<source>
  type tail
  format multiline
  multiline_flush_interval 5s
  format_firstline /^\w\d{4}/
  format1 /^(?<severity>\w)(?<time>\d{4} [^\s]*)\s+(?<pid>\d+)\s+(?<source>[^ \]]+)\] (?<message>.*)/
  time_format %m%d %H:%M:%S.%N
  path /var/log/cluster-autoscaler.log
  pos_file /var/log/gcp-cluster-autoscaler.log.pos
  tag cluster-autoscaler
</source>

# Logs from systemd-journal for interesting services.
<source>
  type systemd
  filters [{ "_SYSTEMD_UNIT": "docker.service" }]
  pos_file /var/log/gcp-journald-docker.pos
  read_from_head true
  tag docker
</source>

<source>
  type systemd
  filters [{ "_SYSTEMD_UNIT": "kubelet.service" }]
  pos_file /var/log/gcp-journald-kubelet.pos
  read_from_head true
  tag kubelet
</source>


monitoring.conf: |- 

# Prometheus monitoring
<source>
  @type prometheus
  port 80
</source>
<source>
  @type prometheus_monitor
</source>
# This source is used to acquire approximate process start timestamp,
# which purpose is explained before the corresponding output plugin.
<source>
  @type exec
  command /bin/sh -c 'date +%s'
  tag process_start
  time_format %Y-%m-%d %H:%M:%S
  keys process_start_timestamp
</source>

# This filter is used to convert process start timestamp to integer
# value for correct ingestion in the prometheus output plugin.
<filter process_start>
  @type record_transformer
  enable_ruby true
  auto_typecast true
  <record>
    process_start_timestamp ${record["process_start_timestamp"].to_i}
  </record>
</filter>
  output.conf: |-
# This match is placed before the all-matching output to provide metric
# exporter with a process start timestamp for correct exporting of
# cumulative metrics to Stackdriver.
<match process_start>
  @type prometheus
  <metric>
    type gauge
    name process_start_time_seconds
    desc Timestamp of the process start in seconds
    key process_start_timestamp
  </metric>
</match>
<match kubernetes.**>
  @type copy

  <store>
    @type google_cloud

    # Set the buffer type to file to improve the reliability and reduce the memory consumption
    buffer_type file
    buffer_path /var/log/fluentd-buffers/kubernetes.containers.buffer
    # Set queue_full action to block because we want to pause gracefully
    # in case of the off-the-limits load instead of throwing an exception
    buffer_queue_full_action block
    # Set the chunk limit conservatively to avoid exceeding the GCL limit
    # of 10MiB per write request.
    buffer_chunk_limit 2M
    # Cap the combined memory usage of this buffer and the one below to
    # 2MiB/chunk * (6 + 2) chunks = 16 MiB
    buffer_queue_limit 6
    # Never wait more than 5 seconds before flushing logs in the non-error case.
    flush_interval 5s
    # Never wait longer than 30 seconds between retries.
    max_retry_wait 30
    # Disable the limit on the number of retries (retry forever).
    disable_retry_limit
    # Use multiple threads for processing.
    num_threads 2
  </store>
  <store>
    @type prometheus

    <metric>
      type counter
      name logging_entry_count
      desc Total number of log entries generated by either an application container or a system component
      <labels>
        component container
      </labels>
    </metric>
  </store>
</match>

<match **>
  @type copy

  <store>
    @type google_cloud

    detect_subservice false
    buffer_type file
    buffer_path /var/log/fluentd-buffers/kubernetes.system.buffer
    buffer_queue_full_action block
    buffer_chunk_limit 2M
    buffer_queue_limit 2
    flush_interval 5s
    max_retry_wait 30
    disable_retry_limit
    num_threads 2
  </store>
  <store>
    @type prometheus

    <metric>
      type counter
      name logging_entry_count
      desc Total number of log entries generated by either an application         container or a system component
      <labels>
        component system
      </labels>
    </metric>
  </store>
</match>