Logs saved in overlay2/merged/tmp/fluent/backup folder and not written to output store

[TarunDhan]

Hi Team,

We have deployed fluent-bit in EKS along with our application, which tails the logs and sends to fluentd(running on docker) which then writes to elasticsearch.
It works for some time but then all the logs are getting saved in below folder and nothing is written in elasticsearch /var/lib/docker/overlay2/5baca98c799dd930d1c277868518ca365/merged/tmp/fluent/backup/

We see the below errors in the logs -
2023-04-04 06:45:57 +0000 [warn]: #1 got unrecoverable error in primary and no secondary error_class=NoMethodError error="undefined method `emit_stream' for nil:NilClass\n\n router.emit_stream(emit_tag, e.retry_stream)\n ^^^^^^^^^^^^"

Below Fluentd docker details

 "Description": "Fluentd docker image",
  "Vendor": "Fluent Organization",
  "Version": "1.15.1",
  "maintainer": "Eduardo Silva <eduardoÉtreasure-data.com>"

<system>
  workers 6
</system>

<source>
  @type http
  port 8888
  bind 0.0.0.0
  body_size_limit 32m
  keepalive_timeout 10s
</source>

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

# expose metrics in prometheus format

<source>
  @type prometheus
  bind 0.0.0.0
  port 24231
  metrics_path /metrics
</source>

<source>
  @type prometheus_output_monitor
  interval 10
  <labels>
    hostname ${hostname}
  </labels>
</source>

# Ignore logs fluentd itself
<match fluent.**>
  @type null
</match>

# Ref: https://github.com/repeatedly/fluent-plugin-record-modifier
# Add new key "app_name" from kubernetes name label (app.kubernetes.io/name OR k8s-app OR app OR component)
# Remove few unwanted keys from logs to reduce size
<filter kube.**>
  @type record_modifier
  <record>
    app_name ${ if record['kubernetes']['labels']['app.kubernetes.io/name'] != nil; record['app_name'] = record['kubernetes']['labels']['app.kubernetes.io/name']; elsif record['kubernetes']['labels']['app_kubernetes_io/name'] != nil; record['app_name'] = record['kubernetes']['labels']['app_kubernetes_io/name']; elsif record['kubernetes']['labels']['app'] != nil; record['app_name'] = record['kubernetes']['labels']['app']; elsif record['kubernetes']['labels']['component'] != nil; record['app_name'] = record['kubernetes']['labels']['component']; elsif record['kubernetes']['labels']['k8s-app'] != nil; record['app_name'] = record['kubernetes']['labels']['k8s-app']; else; record['app_name'] = "no_label"; end;}
    log_time ${Time.at(time).to_s}
    tag ${record['kubernetes']['namespace_name']}.${record['app_name']}.${record['kubernetes']['container_name']}.${record['stream']}
    stream ${record['stream']}
    log ${record['log']}
    host ${record['kubernetes']['host']}
    instance ${record['kubernetes']['labels']['app.kubernetes.io/instance']}
    pod_name ${record['kubernetes']['pod_name']}
    #container_image ${record['kubernetes']['container_image']}
  </record>

  # Comment below line for getting all kube metadata
  whitelist_keys ""
</filter>

<filter kube.**>
  @type prometheus
  <metric>
    name fluentd_input_status_num_records_total
    type counter
    desc The total number of incoming records
    <labels>
      tag ${tag}
      hostname ${hostname}
    </labels>
  </metric>
</filter>

<match kube.**>
  @type rewrite_tag_filter
  <rule>
    key tag
    pattern ^(.+)$
    tag kube_rewrite.$1
  </rule>
</match>

<match kube_rewrite.prod.a.**>
  @type relabel
  @label @highvolume
</match>
<match kube_rewrite.prod.app1.**>
  @type relabel
  @label @highvolume
</match>

<label @highvolume>
<match kube_rewrite.prod.**>
  @type copy
  <store>
    @type s3
    s3_bucket logging-backup
    s3_region us-1
    path "/${tag[1]}/${tag[2]}/%Y/%m/%d/%H/${tag[3]}-${tag[4]}-#{worker_id}-#{hostname}-"
    store_as gzip_command
    <buffer tag, time>
      @type file
      path /var/log/fluent-buffer/s3-l/
      flush_thread_count 4
      timekey 30m
      timekey_wait 10m
      timekey_zone "Asia/Kolkata"
      chunk_limit_size 256m
    </buffer>
    <format>
      @type json
    </format>
  </store>

  <store>
    @type elasticsearch
    hosts 123.11.5.190:5610,123.11.5.66:5610,123.11.5.177:5610,123.11.5.148:5610,123.11.5.44:5610
    logstash_format true
    logstash_prefix logs-${tag[1]}-${tag[2]}
    request_timeout 120s
    time_key log_time
    logstash_dateformat %Y.%m.%d.%H

    <buffer>
      @type file
      path /var/log/fluent-buffer/es-el/
      flush_thread_count 12
      timekey_zone "Asia/Kolkata"
      timekey 15m
      timekey_wait 1s
      #flush_mode interval
      #flush_interval 10s
      chunk_limit_size 10m
    </buffer>
  </store>

  <store>
    @type prometheus
    <metric>
      name fluentd_output_status_num_records_total
      type counter
      desc The total number of outgoing records
      <labels>
        tag ${tag}
        hostname ${hostname}
      </labels>
    </metric>
  </store>

</match>
</label>

<match kube_rewrite.**>
  @type copy
  <store>
    @type s3
    s3_bucket logging-backup
    s3_region us-1
    path "/${tag[1]}/${tag[2]}/%Y/%m/%d/%H/${tag[3]}-${tag[4]}-#{worker_id}-#{hostname}-"
    store_as gzip_command
    <buffer tag, time>
      @type file
      path /var/log/fluent-buffer/s3-new/
      flush_thread_count 4
      timekey 30m
      timekey_wait 10m
      timekey_zone "Asia/Kolkata"
      chunk_limit_size 256m
    </buffer>
    <format>
      @type json
    </format>
  </store>

  <store>
    @type elasticsearch
    hosts 123.11.5.190:5610,123.11.5.66:5610,123.11.5.177:5610,123.11.5.148:5610,123.11.5.44:5610
    logstash_format true
    logstash_prefix logs-${tag[1]}-${tag[2]}
    request_timeout 120s
    time_key log_time
    logstash_dateformat %Y.%m.%d

    <buffer>
      @type file
      path /var/log/fluent-buffer/es/
      flush_thread_count 12
      timekey_zone "Asia/Kolkata"
      timekey 15m
      timekey_wait 1s
      #flush_mode interval
      #flush_interval 10s
      chunk_limit_size 10m
    </buffer>
  </store>

  <store>
    @type prometheus
    <metric>
      name fluentd_output_status_num_records_total
      type counter
      desc The total number of outgoing records
      <labels>
        tag ${tag}
        hostname ${hostname}
      </labels>
    </metric>
  </store>

</match>

<match stats.**>
  @type copy
  <store>
    @type elasticsearch
    hosts 123.11.5.190:5610,123.11.5.66:5610,123.11.5.177:5610,123.11.5.148:5610,123.11.5.44:5610
    logstash_format true
    logstash_prefix stats-${tag[1]}-${tag[2]}
    request_timeout 120s
    time_key datetime
    logstash_dateformat %Y.%m.%d

    <buffer>
      @type file
      path /var/log/fluent-buffer/stats/
      flush_thread_count 8
      #timekey_zone "Asia/Kolkata"
      timekey 15m
      timekey_wait 1s
      #flush_mode interval
      #flush_interval 1s
      chunk_limit_size 1m
    </buffer>
  </store>
</match>

[peterroth]

Hello,

We are running into a similar situation, the /tmp/fluent/backup/ directory is written with tremendous amount of files and they are kept for ever. I'm trying to solve this phenomenon and this ticket seems to be relevant: 1856. As I currently understand, there is a retry mechanism to ingest data into Elasticsearch, but after an amount of retries Fluentd gives up and writes the data into the directory above instead.