Configure Fluentd with Kubernetes DaemonSet and log routing for centralized collection

Create Fluentd namespace and RBAC configuration

Set up the namespace and service account with cluster-wide permissions to read container logs from all nodes.

apiVersion: v1
kind: Namespace
metadata:
  name: fluentd-system
---
apiVersion: v1
kind: ServiceAccount
metadata:
  name: fluentd
  namespace: fluentd-system
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: fluentd
rules:
apiGroups: [""]
  resources:
    - pods
    - namespaces
  verbs:
    - get
    - list
    - watch
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
  name: fluentd
roleRef:
  kind: ClusterRole
  name: fluentd
  apiGroup: rbac.authorization.k8s.io
subjects:
kind: ServiceAccount
  name: fluentd
  namespace: fluentd-system

kubectl apply -f fluentd-rbac.yaml

Create Fluentd configuration with log routing

Configure Fluentd with input sources, filtering, parsing, and multiple output destinations including Elasticsearch and S3.

apiVersion: v1
kind: ConfigMap
metadata:
  name: fluentd-config
  namespace: fluentd-system
data:
  fluent.conf: |
    
      @type tail
      @id in_tail_container_logs
      path /var/log/containers/*.log
      pos_file /var/log/fluentd-containers.log.pos
      tag kubernetes.*
      read_from_head true
      
        @type cri
      
    

    
      @type tail
      @id in_tail_minion
      path /var/log/salt/minion
      pos_file /var/log/fluentd-salt.log.pos
      tag salt
      
        @type regexp
        expression /^(?[^ ] [^ ,])[^\[]\[(?[^\]])\s\][^:]:?(?.*)$/
        time_format %Y-%m-%d %H:%M:%S
      
    

    
      @type tail
      @id in_tail_startupscript
      path /var/log/startupscript.log
      pos_file /var/log/fluentd-startupscript.log.pos
      tag startupscript
      
        @type syslog
      
    

    
      @type tail
      @id in_tail_docker
      path /var/log/docker.log
      pos_file /var/log/fluentd-docker.log.pos
      tag docker
      
        @type regexp
        expression /^time="(?[^)])" level=(?[^ ]) msg="(?[^"]*)"/
        time_format %Y-%m-%dT%H:%M:%S.%NZ
      
    

    
      @type tail
      @id in_tail_etcd
      path /var/log/etcd.log
      pos_file /var/log/fluentd-etcd.log.pos
      tag etcd
      
        @type regexp
        expression /^(?[^ ] [^ ]) (?[^ ]) | (?.)$/
        time_format %Y-%m-%d %H:%M:%S
      
    

    
      @type tail
      @id in_tail_kubelet
      multiline_flush_interval 5s
      path /var/log/kubelet.log
      pos_file /var/log/fluentd-kubelet.log.pos
      tag kubelet
      
        @type kubernetes
      
    

    
      @type tail
      @id in_tail_kube_proxy
      multiline_flush_interval 5s
      path /var/log/kube-proxy.log
      pos_file /var/log/fluentd-kube-proxy.log.pos
      tag kube-proxy
      
        @type kubernetes
      
    

    
      @type tail
      @id in_tail_kube_apiserver
      multiline_flush_interval 5s
      path /var/log/kube-apiserver.log
      pos_file /var/log/fluentd-kube-apiserver.log.pos
      tag kube-apiserver
      
        @type kubernetes
      
    

    
      @type tail
      @id in_tail_kube_controller_manager
      multiline_flush_interval 5s
      path /var/log/kube-controller-manager.log
      pos_file /var/log/fluentd-kube-controller-manager.log.pos
      tag kube-controller-manager
      
        @type kubernetes
      
    

    
      @type tail
      @id in_tail_kube_scheduler
      multiline_flush_interval 5s
      path /var/log/kube-scheduler.log
      pos_file /var/log/fluentd-kube-scheduler.log.pos
      tag kube-scheduler
      
        @type kubernetes
      
    

    
      @type kubernetes_metadata
      @id filter_kube_metadata
      kubernetes_url "#{ENV['KUBERNETES_SERVICE_HOST']}:#{ENV['KUBERNETES_SERVICE_PORT_HTTPS']}"
      verify_ssl "#{ENV['KUBERNETES_VERIFY_SSL'] || true}"
      ca_file "#{ENV['KUBERNETES_CA_FILE']}"
      skip_labels false
      skip_container_metadata false
      skip_master_url false
      skip_namespace_metadata false
    

    
      @type record_transformer
      @id filter_containers_stream_transformer
      
        hostname ${hostname}
        raw ${record['log']}
      
    

    fluentd.log>
      @type null
      @id ignore_fluentd_logs
    

    kube-system.log>
      @type elasticsearch
      @id out_es_kube_system
      @log_level info
      include_tag_key true
      host "#{ENV['FLUENT_ELASTICSEARCH_HOST'] || 'elasticsearch.logging.svc.cluster.local'}"
      port "#{ENV['FLUENT_ELASTICSEARCH_PORT'] || '9200'}"
      path "#{ENV['FLUENT_ELASTICSEARCH_PATH'] || ''}"
      scheme "#{ENV['FLUENT_ELASTICSEARCH_SCHEME'] || 'http'}"
      ssl_verify "#{ENV['FLUENT_ELASTICSEARCH_SSL_VERIFY'] || 'true'}"
      ssl_version "#{ENV['FLUENT_ELASTICSEARCH_SSL_VERSION'] || 'TLSv1_2'}"
      user "#{ENV['FLUENT_ELASTICSEARCH_USER'] || use_default}"
      password "#{ENV['FLUENT_ELASTICSEARCH_PASSWORD'] || use_default}"
      reload_connections false
      reconnect_on_error true
      reload_on_failure true
      log_es_400_reason false
      logstash_prefix kube-system
      logstash_dateformat %Y.%m.%d
      logstash_format true
      index_name kube-system
      target_index_key
      type_name _doc
      include_timestamp false
      template_name kube-system
      template_file
      template_overwrite false
      sniffer_class_name Fluent::Plugin::ElasticsearchSimpleSniffer
      request_timeout 5s
      application_name default
      suppress_type_name true
      enable_ilm false
      ilm_policy_id logstash-policy
      ilm_policy
      ilm_policy_overwrite false
      
        flush_thread_count 8
        flush_interval 5s
        chunk_limit_size 2M
        queue_limit_length 32
        retry_max_interval 30
        retry_forever true
      
    

    
      @type copy
      
        @type elasticsearch
        @id out_es_containers
        @log_level info
        include_tag_key true
        host "#{ENV['FLUENT_ELASTICSEARCH_HOST'] || 'elasticsearch.logging.svc.cluster.local'}"
        port "#{ENV['FLUENT_ELASTICSEARCH_PORT'] || '9200'}"
        path "#{ENV['FLUENT_ELASTICSEARCH_PATH'] || ''}"
        scheme "#{ENV['FLUENT_ELASTICSEARCH_SCHEME'] || 'http'}"
        ssl_verify "#{ENV['FLUENT_ELASTICSEARCH_SSL_VERIFY'] || 'true'}"
        ssl_version "#{ENV['FLUENT_ELASTICSEARCH_SSL_VERSION'] || 'TLSv1_2'}"
        user "#{ENV['FLUENT_ELASTICSEARCH_USER'] || use_default}"
        password "#{ENV['FLUENT_ELASTICSEARCH_PASSWORD'] || use_default}"
        reload_connections false
        reconnect_on_error true
        reload_on_failure true
        log_es_400_reason false
        logstash_prefix containers
        logstash_dateformat %Y.%m.%d
        logstash_format true
        index_name containers
        target_index_key
        type_name _doc
        include_timestamp false
        template_name containers
        template_file
        template_overwrite false
        sniffer_class_name Fluent::Plugin::ElasticsearchSimpleSniffer
        request_timeout 5s
        application_name default
        suppress_type_name true
        enable_ilm false
        ilm_policy_id logstash-policy
        ilm_policy
        ilm_policy_overwrite false
        
          flush_thread_count 8
          flush_interval 5s
          chunk_limit_size 2M
          queue_limit_length 32
          retry_max_interval 30
          retry_forever true
        
      
      
        @type s3
        @id out_s3_containers
        aws_key_id "#{ENV['AWS_ACCESS_KEY_ID']}"
        aws_sec_key "#{ENV['AWS_SECRET_ACCESS_KEY']}"
        s3_bucket "#{ENV['S3_BUCKET_NAME']}"
        s3_region "#{ENV['AWS_REGION'] || 'us-west-2'}"
        path logs/containers/%Y/%m/%d/
        s3_object_key_format %{path}%{time_slice}_%{index}.%{file_extension}
        buffer_type file
        buffer_path /var/log/fluent/s3
        time_slice_format %Y%m%d%H
        time_slice_wait 1m
        format json
        include_time_key true
        include_tag_key true
        
          timekey 3600
          timekey_wait 60
          chunk_limit_size 100m
        
      
    

    
      @type elasticsearch
      @id out_es_default
      @log_level info
      include_tag_key true
      host "#{ENV['FLUENT_ELASTICSEARCH_HOST'] || 'elasticsearch.logging.svc.cluster.local'}"
      port "#{ENV['FLUENT_ELASTICSEARCH_PORT'] || '9200'}"
      path "#{ENV['FLUENT_ELASTICSEARCH_PATH'] || ''}"
      scheme "#{ENV['FLUENT_ELASTICSEARCH_SCHEME'] || 'http'}"
      ssl_verify "#{ENV['FLUENT_ELASTICSEARCH_SSL_VERIFY'] || 'true'}"
      ssl_version "#{ENV['FLUENT_ELASTICSEARCH_SSL_VERSION'] || 'TLSv1_2'}"
      user "#{ENV['FLUENT_ELASTICSEARCH_USER'] || use_default}"
      password "#{ENV['FLUENT_ELASTICSEARCH_PASSWORD'] || use_default}"
      reload_connections false
      reconnect_on_error true
      reload_on_failure true
      log_es_400_reason false
      logstash_prefix system
      logstash_dateformat %Y.%m.%d
      logstash_format true
      index_name system
      target_index_key
      type_name _doc
      include_timestamp false
      template_name system
      template_file
      template_overwrite false
      sniffer_class_name Fluent::Plugin::ElasticsearchSimpleSniffer
      request_timeout 5s
      application_name default
      suppress_type_name true
      enable_ilm false
      ilm_policy_id logstash-policy
      ilm_policy
      ilm_policy_overwrite false
      
        flush_thread_count 8
        flush_interval 5s
        chunk_limit_size 2M
        queue_limit_length 32
        retry_max_interval 30
        retry_forever true
      
    

kubectl apply -f fluentd-configmap.yaml

Create secrets for external services

Configure authentication credentials for Elasticsearch and AWS S3 access.

kubectl create secret generic fluentd-secrets -n fluentd-system \
  --from-literal=FLUENT_ELASTICSEARCH_USER=elastic \
  --from-literal=FLUENT_ELASTICSEARCH_PASSWORD=your-elasticsearch-password \
  --from-literal=AWS_ACCESS_KEY_ID=your-aws-access-key \
  --from-literal=AWS_SECRET_ACCESS_KEY=your-aws-secret-key \
  --from-literal=S3_BUCKET_NAME=your-logs-bucket \
  --from-literal=AWS_REGION=us-west-2

Deploy Fluentd DaemonSet

Create the DaemonSet to run Fluentd on every node with proper resource limits and volume mounts.

apiVersion: apps/v1
kind: DaemonSet
metadata:
  name: fluentd
  namespace: fluentd-system
  labels:
    k8s-app: fluentd-logging
    version: v1
spec:
  selector:
    matchLabels:
      k8s-app: fluentd-logging
      version: v1
  template:
    metadata:
      labels:
        k8s-app: fluentd-logging
        version: v1
    spec:
      serviceAccount: fluentd
      serviceAccountName: fluentd
      tolerations:
      - key: node-role.kubernetes.io/control-plane
        effect: NoSchedule
      - key: node-role.kubernetes.io/master
        effect: NoSchedule
      containers:
      - name: fluentd
        image: fluent/fluentd-kubernetes-daemonset:v1.16-debian-elasticsearch7-1
        env:
          - name: FLUENT_ELASTICSEARCH_HOST
            value: "elasticsearch.logging.svc.cluster.local"
          - name: FLUENT_ELASTICSEARCH_PORT
            value: "9200"
          - name: FLUENT_ELASTICSEARCH_SCHEME
            value: "http"
          - name: FLUENTD_SYSTEMD_CONF
            value: disable
          - name: FLUENT_CONTAINER_TAIL_EXCLUDE_PATH
            value: /var/log/containers/fluent*
          - name: FLUENT_ELASTICSEARCH_SSL_VERIFY
            value: "true"
          - name: FLUENT_ELASTICSEARCH_SSL_VERSION
            value: "TLSv1_2"
          - name: FLUENT_ELASTICSEARCH_USER
            valueFrom:
              secretKeyRef:
                name: fluentd-secrets
                key: FLUENT_ELASTICSEARCH_USER
          - name: FLUENT_ELASTICSEARCH_PASSWORD
            valueFrom:
              secretKeyRef:
                name: fluentd-secrets
                key: FLUENT_ELASTICSEARCH_PASSWORD
          - name: AWS_ACCESS_KEY_ID
            valueFrom:
              secretKeyRef:
                name: fluentd-secrets
                key: AWS_ACCESS_KEY_ID
          - name: AWS_SECRET_ACCESS_KEY
            valueFrom:
              secretKeyRef:
                name: fluentd-secrets
                key: AWS_SECRET_ACCESS_KEY
          - name: S3_BUCKET_NAME
            valueFrom:
              secretKeyRef:
                name: fluentd-secrets
                key: S3_BUCKET_NAME
          - name: AWS_REGION
            valueFrom:
              secretKeyRef:
                name: fluentd-secrets
                key: AWS_REGION
        resources:
          limits:
            memory: 512Mi
            cpu: 200m
          requests:
            cpu: 100m
            memory: 200Mi
        volumeMounts:
        - name: fluentd-config
          mountPath: /fluentd/etc/fluent.conf
          subPath: fluent.conf
        - name: varlog
          mountPath: /var/log
        - name: varlibdockercontainers
          mountPath: /var/lib/docker/containers
          readOnly: true
        - name: runlogjournal
          mountPath: /run/log/journal
          readOnly: true
        - name: dmesg
          mountPath: /var/log/dmesg
          readOnly: true
      terminationGracePeriodSeconds: 30
      volumes:
      - name: fluentd-config
        configMap:
          name: fluentd-config
      - name: varlog
        hostPath:
          path: /var/log
      - name: varlibdockercontainers
        hostPath:
          path: /var/lib/docker/containers
      - name: runlogjournal
        hostPath:
          path: /run/log/journal
      - name: dmesg
        hostPath:
          path: /var/log/dmesg

kubectl apply -f fluentd-daemonset.yaml

Configure custom application log parsing

Add additional parsing rules for specific application formats like JSON logs and multiline stack traces.

apiVersion: v1
kind: ConfigMap
metadata:
  name: fluentd-custom-parsing
  namespace: fluentd-system
data:
  custom-parsing.conf: |
    
      @type parser
      @id filter_parser
      key_name log
      reserve_time true
      reserve_data true
      remove_key_name_field false
      replace_invalid_sequence true
      emit_invalid_record_to_error false
      
        @type multi_format
        
          format json
          time_key timestamp
          time_type string
          time_format %Y-%m-%dT%H:%M:%S.%L%z
        
        
          format regexp
          expression /^(?\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z)\s+(?\w+)\s+(?.*)$/
          time_key timestamp
          time_format %Y-%m-%dT%H:%M:%S.%LZ
        
        
          format regexp
          expression /^\[(?[^\]]+)\]\s+(?\w+)\s+-\s+(?.*)$/
          time_key timestamp
          time_format %Y-%m-%d %H:%M:%S
        
        
          format none
        
      
    

    
      @type grep
      @id filter_grep_exclude_healthcheck
      
        key log
        pattern /health|ping|status/
      
    

    
      @type record_transformer
      @id filter_add_cluster_info
      enable_ruby
      auto_typecast true
      
        cluster_name "#{ENV['CLUSTER_NAME'] || 'production'}"
        environment "#{ENV['ENVIRONMENT'] || 'prod'}"
        region "#{ENV['AWS_REGION'] || 'us-west-2'}"
        log_level ${record.dig('kubernetes', 'labels', 'app.kubernetes.io/component') == 'database' ? 'INFO' : record['level']}
      
    

kubectl apply -f fluentd-custom-parsing.yaml

Update DaemonSet to include custom parsing

Add the custom parsing configuration to the DaemonSet volume mounts.

kubectl patch daemonset fluentd -n fluentd-system --type='json' -p='[
  {
    "op": "add",
    "path": "/spec/template/spec/containers/0/volumeMounts/-",
    "value": {
      "name": "fluentd-custom-parsing",
      "mountPath": "/fluentd/etc/conf.d/custom-parsing.conf",
      "subPath": "custom-parsing.conf"
    }
  },
  {
    "op": "add",
    "path": "/spec/template/spec/volumes/-",
    "value": {
      "name": "fluentd-custom-parsing",
      "configMap": {
        "name": "fluentd-custom-parsing"
      }
    }
  }
]'

Configure log monitoring and alerting

Set up monitoring for Fluentd performance and log flow health.

apiVersion: v1
kind: Service
metadata:
  name: fluentd-metrics
  namespace: fluentd-system
  labels:
    k8s-app: fluentd-logging
  annotations:
    prometheus.io/scrape: "true"
    prometheus.io/port: "24231"
spec:
  ports:
  - name: prometheus
    port: 24231
    protocol: TCP
    targetPort: 24231
  selector:
    k8s-app: fluentd-logging
---
apiVersion: monitoring.coreos.com/v1
kind: ServiceMonitor
metadata:
  name: fluentd-metrics
  namespace: fluentd-system
spec:
  selector:
    matchLabels:
      k8s-app: fluentd-logging
  endpoints:
  - port: prometheus
    interval: 30s
    path: /metrics

kubectl apply -f fluentd-monitoring.yaml

Set up log routing by namespace

Configure different output destinations based on Kubernetes namespace and application labels.

apiVersion: v1
kind: ConfigMap
metadata:
  name: fluentd-routing
  namespace: fluentd-system
data:
  routing.conf: |
    production.log>
      @type copy
      
        @type elasticsearch
        @id out_es_production
        host "#{ENV['FLUENT_ELASTICSEARCH_HOST']}"
        port "#{ENV['FLUENT_ELASTICSEARCH_PORT']}"
        index_name production-logs
        logstash_format true
        logstash_prefix production
        
          flush_interval 1s
          chunk_limit_size 1M
        
      
      
        @type s3
        @id out_s3_production
        aws_key_id "#{ENV['AWS_ACCESS_KEY_ID']}"
        aws_sec_key "#{ENV['AWS_SECRET_ACCESS_KEY']}"
        s3_bucket "#{ENV['S3_BUCKET_NAME']}"
        path logs/production/%Y/%m/%d/
        
          timekey 3600
          timekey_wait 60
        
      
      
        @type kafka2
        @id out_kafka_production
        brokers "#{ENV['KAFKA_BROKERS'] || 'kafka:9092'}"
        topic_key topic
        default_topic production-logs
        
          @type json
        
        
          flush_interval 1s
        
      
    

    staging.log>
      @type elasticsearch
      @id out_es_staging
      host "#{ENV['FLUENT_ELASTICSEARCH_HOST']}"
      port "#{ENV['FLUENT_ELASTICSEARCH_PORT']}"
      index_name staging-logs
      logstash_format true
      logstash_prefix staging
      
        flush_interval 5s
        chunk_limit_size 2M
      
    

    monitoring.log>
      @type forward
      @id out_forward_monitoring
      
        name monitoring-fluentd
        host monitoring-fluentd.monitoring.svc.cluster.local
        port 24224
        weight 60
      
      
        flush_interval 1s
        retry_max_interval 30
        retry_forever true
      
    

    critical>
      @type copy
      
        @type elasticsearch
        @id out_es_critical
        host "#{ENV['FLUENT_ELASTICSEARCH_HOST']}"
        port "#{ENV['FLUENT_ELASTICSEARCH_PORT']}"
        index_name critical-logs
        logstash_format true
        logstash_prefix critical
        
          flush_interval 1s
        
      
      
        @type slack
        @id out_slack_critical
        webhook_url "#{ENV['SLACK_WEBHOOK_URL']}"
        channel alerts
        username fluentd-bot
        icon_emoji :warning:
        title Critical Log Alert
        message %s
        message_keys message
        
          flush_interval 1s
        
      
    

kubectl apply -f fluentd-routing.yaml

Configure log retention and archival

Set up automated log retention policies and long-term archival to cost-effective storage.

apiVersion: v1
kind: ConfigMap
metadata:
  name: fluentd-retention
  namespace: fluentd-system
data:
  retention.conf: |
    
      @type copy
      
        @type relabel
        @label @mainstream
      
      
        @type relabel
        @label @archive
      
    

    
      
        @type elasticsearch
        @id out_es_mainstream
        host "#{ENV['FLUENT_ELASTICSEARCH_HOST']}"
        port "#{ENV['FLUENT_ELASTICSEARCH_PORT']}"
        logstash_format true
        logstash_prefix mainstream
        template_name mainstream-template
        template_overwrite true
        template {
          "index_patterns": ["mainstream-*"],
          "settings": {
            "index": {
              "number_of_shards": 1,
              "number_of_replicas": 1,
              "refresh_interval": "30s",
              "lifecycle": {
                "name": "mainstream-logs-policy",
                "rollover_alias": "mainstream-logs"
              }
            }
          }
        }
        
          flush_interval 5s
          chunk_limit_size 2M
          queue_limit_length 32
        
      
    

    
      
        @type record_transformer
        enable_ruby
        
          archive_date ${Time.now.strftime("%Y-%m-%d")}
          retention_days 90
        
      
      
        @type s3
        @id out_s3_archive
        aws_key_id "#{ENV['AWS_ACCESS_KEY_ID']}"
        aws_sec_key "#{ENV['AWS_SECRET_ACCESS_KEY']}"
        s3_bucket "#{ENV['S3_ARCHIVE_BUCKET'] || ENV['S3_BUCKET_NAME']}"
        s3_region "#{ENV['AWS_REGION']}"
        path archive/logs/%Y/%m/%d/
        s3_object_key_format %{path}cluster=%{cluster_name}/namespace=%{kubernetes_namespace}/pod=%{kubernetes_pod_name}/%{time_slice}_%{index}.%{file_extension}
        store_as gzip_command
        
          timekey 86400
          timekey_wait 3600
          chunk_limit_size 100m
          compress gzip
        
      
    

kubectl apply -f fluentd-retention.yaml