Dynamic FluentD Configuration for Kubernetes cluster logs using kubernetes_metadata, rewrite_tag_filter, and forest plugins

fluentd_kubernetes_dynamic_config.yaml

apiVersion: v1
data:
  fluentd.conf: |
    # Capture the kubernetes pod container logs
    <source>
      @type tail
      format json
      path /var/log/containers/*.log
      pos_file /var/log/kubernetes.pos
      time_format %Y-%m-%dT%H:%M:%S
      tag kubernetes.*
      emit_unmatched_lines true
      message_key event
      path_key source_name
      # read_from_head true
    </source>

    # Extract the kubernetes metadata from the log file names
    <filter kubernetes.var.log.containers.**.log>
      @type kubernetes_metadata
    </filter>

    # Create a new field so that we can rewrite the current tag with it
    <filter kubernetes.**>
      @type record_transformer
      enable_ruby true
      <record>
        kubernetes_namespace_pod_container_name ${record["kubernetes"]["namespace_name"]}.${record["kubernetes"]["pod_name"]}.${record["kubernetes"]["container_name"]}
      </record>
    </filter>

    # Retag based on the namespace, pod, and container name of the log message
    <match kubernetes.**>
      @type rewrite_tag_filter
      rewriterule1 kubernetes_namespace_pod_container_name ^(.+)$ kube.$1
    </match>

    # Remove the temporary field
    <filter kube.**>
      @type record_transformer
      remove_keys kubernetes_namespace_pod_container_name
    </filter>

    # Kubernetes and FluentD logs go to aws elasticsearch
    <match kube.kube** kube.**fluentd**>
      <store>
        buffer_chunk_limit 5m
        buffer_path /var/log/td-agent/buffer/kube-system-aws-elasticsearch-service
        buffer_queue_limit 60480
        buffer_type file
        disable_retry_limit true
        <endpoint>
          region ap-southeast-2
          url https://AWS-ELASTICSEARCH-DOMAIN
        </endpoint>
        flush_interval 10s
        logstash_format true
        max_retry_wait 30s
        reload_connections false
        @type aws-elasticsearch-service
        # @type null
      </store>
      @type copy
    </match>

    # App container specific log go to sumologic + s3
    #   Use the forest plugin to instantiate output stores for each set of tag_parts
    #     ${tag_parts[1..-1]} = namespace.pod.container
    #     ${tag_parts[1]} = namespace name
    #     ${tag_parts[2..-2]} = pod name
    #     ${tag_parts[-1]} = container name
    <match kube.**>
      <store>
        @type forest
        subtype sumologic
        <template>
          buffer_chunk_limit 5m
          buffer_path /var/log/td-agent/buffer/${tag_parts[1..-1]}-sumologic
          buffer_queue_limit 604800
          buffer_type file
          disable_retry_limit true
          endpoint https://SUMOLOGIC-HTTPS-URL
          flush_interval 1s
          log_format text
          log_key log
          max_retry_wait 30s
          # format: namespace/pod/container
          source_category ${tag_parts[1]}/${tag_parts[2..-2]}/${tag_parts[-1]}
          source_host ${tag_parts[1..-1]}
          source_name /var/log/containers/${tag_parts[1..-1]}.log
        </template>
      </store>
      <store>
        @type forest
        subtype s3
        <template>
          acl bucket-owner-full-control
          <assume_role_credentials>
            role_arn AWS-IAM-ROLE
            role_session_name AWS-SESSION-NAME
          </assume_role_credentials>
          buffer_chunk_limit 256m
          buffer_path /var/log/td-agent/buffer/${tag_parts[1..-1]}-s3
          buffer_queue_limit 10080
          buffer_type file
          check_apikey_on_start false
          disable_retry_limit true
          flush_interval 60s
          format json
          include_time_key
          max_retry_wait 30s
          s3_bucket AWS-S3-BUCKET-NAME
          # format: namespace/pod/container/time_format_index_file_ext
          s3_object_key_format ${tag_parts[1]}/${tag_parts[2..-2]}/${tag_parts[-1]}/%{time_slice}_%{index}.%{file_extension}
          s3_region ap-southeast-2
          time_slice_format %Y/%m/%d/%H%M
        </template>
      </store>
      @type copy
    </match>
kind: ConfigMap
metadata:
  name: fluentd-dynamic-config