Logs missing during heavy log volume
Author: jicowanCreated Nov 3, 2024Updated Dec 19, 2025
Labelsenhancement
Describe the bug
During heavy log volumes, e.g. >10k log entries per second, fluentd consistently drops logs. It may be related to log rotation (on Kubernetes). When I ran a load test, I see the following entries in the fluentd logs:
2024-11-02 14:06:36 +0000 [warn]: #0 [in_tail_container_logs] Could not follow a file (inode: 101712298) because an existing watcher for that filepath follows a different inode: 101712295 (e.g. keeps watching a already rotated file). If you keep getting this message, please restart Fluentd. filepath="/var/log/containers/logger-deployment-57cc6745c7-mzxxh_default_logger-8bb9a8d2eb65d5c07af7e194aad99176a79941a69c06b6ae390a0d8b9dd06cf1.log"
2024-11-02 14:06:36 +0000 [warn]: #0 [in_tail_container_logs] Could not follow a file (inode: 97581155) because an existing watcher for that filepath follows a different inode: 97581154 (e.g. keeps watching a already rotated file). If you keep getting this message, please restart Fluentd. filepath="/var/log/containers/logger-deployment-57cc6745c7-nrq45_default_logger-2bad2e8722fb2369996c134f02dcf4a2fff8068d43863d3f7173a56ff2a8bbd0.log"
2024-11-02 14:06:36 +0000 [warn]: #0 [in_tail_container_logs] Could not follow a file (inode: 111149786) because an existing watcher for that filepath follows a different inode: 111149782 (e.g. keeps watching a already rotated file). If you keep getting this message, please restart Fluentd. filepath="/var/log/containers/logger-deployment-57cc6745c7-p4rcl_default_logger-88fb9eaab07505f6d59f03e48e2993069eba82902efe44a46098c0d7d44f24c4.log"
2024-11-02 14:06:36 +0000 [warn]: #0 [in_tail_container_logs] Could not follow a file (inode: 77634742) because an existing watcher for that filepath follows a different inode: 77634741 (e.g. keeps watching a already rotated file). If you keep getting this message, please restart Fluentd. filepath="/var/log/containers/logger-deployment-57cc6745c7-ps45w_default_logger-90f54592392569f72662a2dacfdca239a907c1da4c1729f7a75bb50f56bc9663.log"When I added follow_inodes=true and rotate_wait=0 to the container configuration, the errors went away, but large chunks of logs were still missing and the following entries appeared in the fluentd logs.
2024-11-02 17:27:59 +0000 [warn]: #0 stat() for /var/log/containers/logger-deployment-57cc6745c7-hw4ds_default_logger-aba43bbd009d1652e1961dbd30ed45f09e337bfb42d3fa247b12fde7af248909.log failed. Continuing without tailing it.
2024-11-02 17:27:59 +0000 [warn]: #0 stat() for /var/log/containers/logger-deployment-57cc6745c7-jtxmz_default_logger-742ba4e5339168b7b5442745705bbfed1d93c832027ca0c680b193c9c62e796f.log failed. Continuing without tailing it.
2024-11-02 17:27:59 +0000 [warn]: #0 stat() for /var/log/containers/logger-deployment-57cc6745c7-kmrlv_default_logger-7682a4b64550055203e19ff9387b686e316fe4e5e7884b720dede3692659c686.log failed. Continuing without tailing it.I am running the latest version of the fluentd kubernetes daemonset for cloudwatch, fluent/fluentd-kubernetes-daemonset:v1.17.1-debian-cloudwatch-1.2.
During the test, both memory and CPU utilization for fluentd remained fairly low.
To Reproduce
Run multiple replicas of the following program:
import multiprocessing
import os
import time
import random
import sys
from datetime import datetime
def generate_log_entry():
log_levels = ['INFO', 'WARNING', 'ERROR', 'DEBUG']
messages = [
'User logged in',
'Database connection established',
'File not found',
'Memory usage high',
'Network latency detected',
'Cache cleared',
'API request successful',
'Configuration updated'
]
timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]
level = random.choice(log_levels)
message = random.choice(messages)
pod = os.getenv("POD_NAME", "unknown")
return f"{timestamp} {pod} [{level}] {message}"
def worker(queue):
while True:
log_entry = generate_log_entry()
queue.put(log_entry)
time.sleep(0.01) # Small delay to prevent overwhelming the system
def logger(queue, counter):
while True:
log_entry = queue.get()
with counter.get_lock():
counter.value += 1
print(f"[{counter.value}] {log_entry}", flush=True)
if __name__ == '__main__':
num_processes = multiprocessing.cpu_count()
manager = multiprocessing.Manager()
log_queue = manager.Queue()
# Create a shared counter
counter = multiprocessing.Value('i', 0)
# Start worker processes
workers = []
for _ in range(num_processes - 1): # Reserve one process for logging
p = multiprocessing.Process(target=worker, args=(log_queue,))
p.start()
workers.append(p)
# Start logger process
logger_process = multiprocessing.Process(target=logger, args=(log_queue, counter))
logger_process.start()
try:
# Keep the main process running
while True:
time.sleep(1)
# Print the current count every second
print(f"Total logs emitted: {counter.value}", file=sys.stderr, flush=True)
except KeyboardInterrupt:
print("\nStopping log generation...", file=sys.stderr)
# Stop worker processes
for p in workers:
p.terminate()
p.join()
# Stop logger process
logger_process.terminate()
logger_process.join()
print(f"Log generation stopped. Total logs emitted: {counter.value}", file=sys.stderr)
sys.exit(0)Here's the deployment for the test application:
apiVersion: apps/v1
kind: Deployment
metadata:
name: logger-deployment
labels:
app: logger
spec:
replicas: 1 # Adjust the number of replicas as needed
selector:
matchLabels:
app: logger
template:
metadata:
labels:
app: logger
spec:
affinity:
podAffinity:
preferredDuringSchedulingIgnoredDuringExecution:
- weight: 100
podAffinityTerm:
labelSelector:
matchExpressions:
- key: app
operator: In
values:
- logger
topologyKey: "kubernetes.io/hostname"
containers:
- name: logger
image: jicowan/logger:v3.0
resources:
requests:
cpu: 4
memory: 128Mi
limits:
cpu: 4
memory: 256Mi
env:
- name: POD_NAME
valueFrom:
fieldRef:
fieldPath: metadata.nameHere's the container.conf file for fluentd:
<source>
@type tail
@id in_tail_container_core_logs
@label @raw.containers
@log_level debug
path /var/log/containers/*fluentd-cloudwatch*.log,/var/log/containers/*aws-node*.log,/var/log/containers/*kube-proxy*.log,/var/log/containers/*kube-system*.log,/var/log/containers/cloudwatch-agent*.log,/var/log/containers/policy-manager*.log,/var/log/containers/*private-ca*.log,/var/log/containers/metrics-server*.log,/var/log/containers/rbac-controller*.log,/var/log/containers/cluster-autoscaler*.log,/var/log/containers/cwagent*.log,/var/log/containers/*prometheus*.log,/var/log/containers/*nginx*.log,/var/log/containers/*kube-state*.log
pos_file /var/log/fluentd-core-containers.log.pos
tag corecontainers.**
read_from_head true
follow_inodes true
rotate_wait 0
<parse>
@type "#{ENV['FLUENT_CONTAINER_TAIL_PARSER_TYPE'] || 'json'}"
time_format %Y-%m-%dT%H:%M:%S.%N%:z
</parse>
</source>
<source>
@type tail
@id in_tail_container_logs
@label @raw.containers
path /var/log/containers/*.log
exclude_path /var/log/containers/*aws-node*.log,/var/log/containers/*coredns*.log,/var/log/containers/*kube-proxy*.log,/var/log/containers/*kube-system*.log,/var/log/containers/cloudwatch-agent*.log,/var/log/containers/policy-manager*.log,/var/log/containers/*private-ca*.log,/var/log/containers/metrics-server*.log,/var/log/containers/rbac-controller*.log,/var/log/containers/cluster-autoscaler*.log,/var/log/containers/cwagent*.log,/var/log/containers/*prometheus*.log,/var/log/containers/*nginx*.log,/var/log/containers/*opa*.log,/var/log/containers/*fluentd-cloudwatch*.log,/var/log/containers/*datadog-agent*.log,/var/log/containers/*kube-state-metrics*.log,/var/log/containers/*ebs-csi-node*.log,/var/log/containers/*ebs-csi-controller*.log,/var/log/containers/*fsx-csi-node*.log,/var/log/containers/*calico-node*.log
pos_file /var/log/fluentd-containers.log.pos
tag container.**
read_from_head true
follow_inodes true
rotate_wait 0
<parse>
@type "#{ENV['FLUENT_CONTAINER_TAIL_PARSER_TYPE'] || 'json'}"
time_format %Y-%m-%dT%H:%M:%S.%N%:z
</parse>
</source>
<source>
@type tail
@id in_tail_daemonset_logs
@label @containers
path /var/log/containers/*opa*.log,/var/log/containers/*datadog-agent*.log,/var/log/containers/*ebs-csi-node*.log,/var/log/containers/*ebs-csi-controller*.log,/var/log/containers/*fsx-csi-node*.log,/var/log/containers/*calico-node*.log
pos_file /var/log/daemonset.log.pos
tag daemonset.**
read_from_head true
follow_inodes true
rotate_wait 0
<parse>
@type "#{ENV['FLUENT_CONTAINER_TAIL_PARSER_TYPE'] || 'json'}"
time_format %Y-%m-%dT%H:%M:%S.%N%:z
</parse>
</source>
<label @raw.containers>
<match **>
@id raw.detect_exceptions
@type detect_exceptions
remove_tag_prefix raw
@label @containers
multiline_flush_interval 1s
max_bytes 500000
max_lines 1000
</match>
</label>
<label @containers>
<filter corecontainers.**>
@type prometheus
<metric>
name fluentd_input_status_num_corecontainer_records_total
type counter
desc The total number of incoming corecontainer records
</metric>
</filter>
<filter container.**>
@type prometheus
<metric>
name fluentd_input_status_num_container_records_total
type counter
desc The total number of incoming container records
</metric>
</filter>
<filter daemonset.**>
@type prometheus
<metric>
name fluentd_input_status_num_daemonset_records_total
type counter
desc The total number of incoming daemonset records
</metric>
</filter>
<filter **>
@type record_transformer
@id filter_containers_stream_transformer
<record>
seal_id "110628"
cluster_name "logging"
stream_name ${tag_parts[4]}
</record>
</filter>
<filter **>
@type kubernetes_metadata
@id filter_kube_metadata
@log_level error
</filter>
<match corecontainers.**>
@type copy
<store>
@type prometheus
<metric>
name fluentd_output_status_num_corecontainer_records_total
type counter
desc The total number of outgoing corecontainer records
</metric>
</store>
<store>
@type cloudwatch_logs
@id out_cloudwatch_logs_core_containers
region "us-west-2"
log_group_name "/aws/eks/logging/core-containers"
log_stream_name_key stream_name
remove_log_stream_name_key true
auto_create_stream true
<inject>
time_key time_nanoseconds
time_type string
time_format %Y-%m-%dT%H:%M:%S.%N
</inject>
<buffer>
flush_interval 5s
chunk_limit_size 2m
queued_chunks_limit_size 32
retry_forever true
</buffer>
</store>
</match>
<match container.**>
@type copy
<store>
@type prometheus
<metric>
name fluentd_output_status_num_container_records_total
type counter
desc The total number of outgoing container records
</metric>
</store>
<store>
@type cloudwatch_logs
@id out_cloudwatch_logs_containers
region "us-west-2"
log_group_name "/aws/eks/logging/containers"
log_stream_name_key stream_name
remove_log_stream_name_key true
auto_create_stream true
<inject>
time_key time_nanoseconds
time_type string
time_format %Y-%m-%dT%H:%M:%S.%N
</inject>
<buffer>
flush_interval 5s
chunk_limit_size 2m
queued_chunks_limit_size 32
retry_forever true
</buffer>
</store>
</match>
<match daemonset.**>
@type copy
<store>
@type prometheus
<metric>
name fluentd_output_status_num_daemonset_records_total
type counter
desc The total number of outgoing daemonset records
</metric>
</store>
<store>
@type cloudwatch_logs
@id out_cloudwatch_logs_daemonset
region "us-west-2"
log_group_name "/aws/eks/logging/daemonset"
log_stream_name_key stream_name
remove_log_stream_name_key true
auto_create_stream true
<inject>
time_key time_nanoseconds
time_type string
time_format %Y-%m-%dT%H:%M:%S.%N
</inject>
<buffer>
flush_interval 5s
chunk_limit_size 2m
queued_chunks_limit_size 32
retry_forever true
</buffer>
</store>
</match>
</label>Expected behavior
The test application assigns an sequence number to each log entry. I have a Python notebook that flattens the json log output, sorts the logs by sequence number, then finds gaps in the sequence. This is how I know that fluentd is dropping logs. If everything is working as it should there should be no log loss.
I ran the same tests with fluent bit and experience no log loss.
Your Environment
- Fluentd version: v1.17.1
- Package version:
- Operating system: Amazon Linux 2
- Kernel version: 5.10.225-213.878.amzn2.x86_64Your Configuration
data:
containers.conf: |-
<source>
@type tail
@id in_tail_container_core_logs
@label @raw.containers
@log_level debug
path /var/log/containers/*fluentd-cloudwatch*.log,/var/log/containers/*aws-node*.log,/var/log/containers/*kube-proxy*.log,/var/log/containers/*kube-system*.log,/var/log/containers/cloudwatch-agent*.log,/var/log/containers/policy-manager*.log,/var/log/containers/*private-ca*.log,/var/log/containers/metrics-server*.log,/var/log/containers/rbac-controller*.log,/var/log/containers/cluster-autoscaler*.log,/var/log/containers/cwagent*.log,/var/log/containers/*prometheus*.log,/var/log/containers/*nginx*.log,/var/log/containers/*kube-state*.log
pos_file /var/log/fluentd-core-containers.log.pos
tag corecontainers.**
read_from_head true
follow_inodes true
rotate_wait 0
<parse>
@type "#{ENV['FLUENT_CONTAINER_TAIL_PARSER_TYPE'] || 'json'}"
time_format %Y-%m-%dT%H:%M:%S.%N%:z
</parse>
</source>
<source>
@type tail
@id in_tail_container_logs
@label @raw.containers
path /var/log/containers/*.log
exclude_path /var/log/containers/*aws-node*.log,/var/log/containers/*coredns*.log,/var/log/containers/*kube-proxy*.log,/var/log/containers/*kube-system*.log,/var/log/containers/cloudwatch-agent*.log,/var/log/containers/policy-manager*.log,/var/log/containers/*private-ca*.log,/var/log/containers/metrics-server*.log,/var/log/containers/rbac-controller*.log,/var/log/containers/cluster-autoscaler*.log,/var/log/containers/cwagent*.log,/var/log/containers/*prometheus*.log,/var/log/containers/*nginx*.log,/var/log/containers/*opa*.log,/var/log/containers/*fluentd-cloudwatch*.log,/var/log/containers/*datadog-agent*.log,/var/log/containers/*kube-state-metrics*.log,/var/log/containers/*ebs-csi-node*.log,/var/log/containers/*ebs-csi-controller*.log,/var/log/containers/*fsx-csi-node*.log,/var/log/containers/*calico-node*.log
pos_file /var/log/fluentd-containers.log.pos
tag container.**
read_from_head true
follow_inodes true
rotate_wait 0
<parse>
@type "#{ENV['FLUENT_CONTAINER_TAIL_PARSER_TYPE'] || 'json'}"
time_format %Y-%m-%dT%H:%M:%S.%N%:z
</parse>
</source>
<source>
@type tail
@id in_tail_daemonset_logs
@label @containers
path /var/log/containers/*opa*.log,/var/log/containers/*datadog-agent*.log,/var/log/containers/*ebs-csi-node*.log,/var/log/containers/*ebs-csi-controller*.log,/var/log/containers/*fsx-csi-node*.log,/var/log/containers/*calico-node*.log
pos_file /var/log/daemonset.log.pos
tag daemonset.**
read_from_head true
follow_inodes true
rotate_wait 0
<parse>
@type "#{ENV['FLUENT_CONTAINER_TAIL_PARSER_TYPE'] || 'json'}"
time_format %Y-%m-%dT%H:%M:%S.%N%:z
</parse>
</source>
<label @raw.containers>
<match **>
@id raw.detect_exceptions
@type detect_exceptions
remove_tag_prefix raw
@label @containers
multiline_flush_interval 1s
max_bytes 500000
max_lines 1000
</match>
</label>
<label @containers>
<filter corecontainers.**>
@type prometheus
<metric>
name fluentd_input_status_num_corecontainer_records_total
type counter
desc The total number of incoming corecontainer records
</metric>
</filter>
<filter container.**>
@type prometheus
<metric>
name fluentd_input_status_num_container_records_total
type counter
desc The total number of incoming container records
</metric>
</filter>
<filter daemonset.**>
@type prometheus
<metric>
name fluentd_input_status_num_daemonset_records_total
type counter
desc The total number of incoming daemonset records
</metric>
</filter>
<filter **>
@type record_transformer
@id filter_containers_stream_transformer
<record>
seal_id "110628"
cluster_name "logging"
stream_name ${tag_parts[4]}
</record>
</filter>
<filter **>
@type kubernetes_metadata
@id filter_kube_metadata
@log_level error
</filter>
<match corecontainers.**>
@type copy
<store>
@type prometheus
<metric>
name fluentd_output_status_num_corecontainer_records_total
type counter
desc The total numberSource: fluent/fluentd