Repository navigation
in_cloudwatch log messages getting dropped #158
Description
Problem
I'm using cloudwatch_in to fetch logs from cloudwatch and publishing them to kafka topic. But I see that some of the log messages are missing in kafka topic. I cannot tell whether the messages are getting dropped while pushing it to kafka topic or the plugin is not scrapping them. I see "Rate exceeded" errors in the logs while connecting to cloudwatch and I'm working with Amazon to increase the limits on the cloudwatch for our account.
Can someone tell me if the plugin will retry when it gets "Rate exceeded" error and is this the reason for missing log messages?
...
Steps to replicate
<source>
@type cloudwatch_logs
aws_key_id {{ aws_access_key_id}}
aws_sec_key {{ aws_secret_access_key}}
region {{ aws_region}}
tag ordercapture-cloudwatch.out
log_group_name /aws/lambda/ordercapture-preprod_api
use_log_stream_name_prefix false
use_todays_log_stream true
fetch_interval 30
state_file /var/tmp/ordercapture-preprod_api.out.state
json_handler json
format none
</source>
<source>
@type cloudwatch_logs
aws_key_id {{ aws_access_key_id}}
aws_sec_key {{ aws_secret_access_key}}
region {{ aws_region}}
tag ordercapture-cloudwatch.out
log_group_name /aws/lambda/ordercapture-preprod_authorizer
use_log_stream_name_prefix false
use_todays_log_stream true
fetch_interval 30
state_file /var/tmp/ordercapture-preprod_authorizer.out.state
json_handler json
format none
</source>
<source>
@type cloudwatch_logs
aws_key_id {{ aws_access_key_id}}
aws_sec_key {{ aws_secret_access_key}}
region {{ aws_region}}
tag ordercapture-cloudwatch.out
log_group_name /aws/lambda/ordercapture-preprod_csv_convert
use_log_stream_name_prefix false
use_todays_log_stream true
fetch_interval 30
state_file /var/tmp/ordercapture-preprod_csv_convert.out.state
json_handler json
format none
</source>
<source>
@type cloudwatch_logs
aws_key_id {{ aws_access_key_id}}
aws_sec_key {{ aws_secret_access_key}}
region {{ aws_region}}
tag ordercapture-cloudwatch.out
log_group_name /aws/lambda/ordercapture-preprod_execution
use_log_stream_name_prefix false
use_todays_log_stream true
fetch_interval 30
state_file /var/tmp/ordercapture-preprod_execution.out.state
json_handler json
format none
</source>
<filter *.**>
@type record_transformer
<record>
topic ${tag_prefix[0]}
</record>
</filter>
<match *.**>
@type kafka_buffered
brokers {{ broker }}
default_topic topic
output_data_type json
ssl_ca_cert {{ sslcert}}
<buffer>
@type memory
#flush_interval 1s
flush_at_shutdown true
retry_max_times 5
flush_mode immediate
#path /var/log/fluent/s3/ingest
#timekey 3600 # 1 hour partition
#timekey_wait 10m
#timekey_use_utc true # use utc
#chunk_limit_size 128m
</buffer>
</match>
Expected Behavior or What you need to ask
All the log messages from above mentioned streams should be pushed to kafka topic
Using Fluentd and CloudWatchLogs plugin versions
- OS version
NAME="Alpine Linux"
ID=alpine
VERSION_ID=3.8.1
PRETTY_NAME="Alpine Linux v3.8"
-
Bare Metal or within Docker or Kubernetes or others?
Kubernetes -
Fluentd v1.3.2
-
Fluent-gem list
activesupport (5.2.3)
addressable (2.6.0)
aws-eventstream (1.0.3)
aws-partitions (1.156.0)
aws-sdk-cloudwatchlogs (1.17.0)
aws-sdk-core (3.49.0)
aws-sigv4 (1.1.0)
bigdecimal (1.3.5)
cmath (default: 1.0.0)
concurrent-ruby (1.1.5)
cool.io (1.5.3)
csv (default: 1.0.0)
date (default: 1.0.0)
dig_rb (1.0.1)
digest-crc (0.4.1)
domain_name (0.5.20180417)
etc (default: 1.0.0)
fcntl (default: 1.0.0)
fileutils (default: 1.0.2)
fluent-plugin-cloudwatch-logs (0.7.3)
fluent-plugin-kafka (0.9.2)
fluent-plugin-kubernetes_metadata_filter (2.1.6)
fluentd (1.3.2)
http (0.9.8)
http-cookie (1.0.3)
http-form_data (1.0.3)
http_parser.rb (0.6.0)
i18n (1.6.0)
ipaddr (default: 1.2.0)
jmespath (1.4.0)
json (2.1.0)
kubeclient (1.1.4)
lru_redux (1.1.0)
ltsv (0.1.2)
mime-types (3.2.2)
mime-types-data (3.2019.0331)
minitest (5.11.3)
msgpack (1.2.4)
netrc (0.11.0)
oj (3.3.10)
openssl (default: 2.1.2)
psych (default: 3.0.2)
public_suffix (3.0.3)
recursive-open-struct (1.0.0)
rest-client (2.0.2)
ruby-kafka (0.7.6)
scanf (default: 1.0.0)
serverengine (2.1.0)
sigdump (0.2.4)
stringio (default: 0.0.1)
strptime (0.2.3)
strscan (default: 1.0.0)
thread_safe (0.3.6)
tzinfo (1.2.5)
tzinfo-data (1.2018.7)
unf (0.1.4)
unf_ext (0.0.7.6)
webrick (default: 1.4.2)
yajl-ruby (1.4.1)
zlib (default: 1.0.0)
Reacted by Rohit Verma
Activity
Watson1978 commentedon Jul 19, 2026
Thanks for the report, and sorry for the very late reply.
Short answer to your question: in the version you were on (0.7.3), the input plugin did not retry on
Rate exceeded.A
Rate exceedederror is anAws::CloudWatchLogs::Errors::ThrottlingException, and back then it simply propagated and that fetch cycle was skipped.That said, a throttling error by itself should not permanently drop messages on the input side. The plugin stores
next_forward_tokenonly after a successfulget_log_eventsresponse, so when aThrottlingExceptionis raised the token is not advanced and the nextfetch_intervalretries from the same position.Input-side gaps are only expected in edge cases — e.g. throttling persists long enough that the plugin falls far behind and the
next_tokenexpires (tokens are valid for ~24h), or the stream rolls over while usinguse_todays_log_stream true.Since then we've added an option to handle exactly this case:
So my recommendations:
throttling_retry_secondsso throttled fetches are retried rather than skipped.fetch_intervalor reducing the number of concurrent sources also lowers request pressure.max 5 retries), records can be dropped on buffer overflow or once retries are exhausted, independently of anything on the CloudWatch side. It's worth confirming whether the gap is at ingestion or at delivery.This overlaps with #114, and the throttling behavior is addressed by
throttling_retry_seconds, so I'll close this as resolved/stale. If you can still reproduce message loss on a current version withthrottling_retry_secondsset, please reopen with the plugin version and logs and we'll take another look.Thanks!