in_cloudwatch log messages getting dropped

[sasi1124]

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)

[Watson1978]

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 exceeded error is an Aws::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_token only after a successful get_log_events response, so when a ThrottlingException is raised the token is not advanced and the next fetch_interval retries 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_token expires (tokens are valid for ~24h), or the stream rolls over while using use_todays_log_stream true.

Since then we've added an option to handle exactly this case:

<source>
  @type cloudwatch_logs
  # ...
  fetch_interval 30
  throttling_retry_seconds 30   # on ThrottlingException, wait N seconds and retry instead of skipping
</source>

So my recommendations:

1. Upgrade off 0.7.3 (and Fluentd 1.3.2) to a recent release and set throttling_retry_seconds so throttled fetches are retried rather than skipped.
2. Keep working with AWS to raise the account/API limits (as you mentioned) — throttling is ultimately an API-rate problem. Increasing fetch_interval or reducing the number of concurrent sources also lowers request pressure.
3. Please also check the Kafka output side as a possible source of the missing messages: with a memory buffer and a small retry limit (e.g. 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 with throttling_retry_seconds set, please reopen with the plugin version and logs and we'll take another look.

Thanks!