out_cloudwatch_logs: bug fix: self throttle if too many calls per flu…

…sh (fluent#2618)

Signed-off-by: Wesley Pettit <[email protected]>

plugins/out_cloudwatch_logs/cloudwatch_api.c

@@ -41,6 +41,7 @@
 #include <msgpack.h>
 #include <string.h>
 #include <stdio.h>
+#include <unistd.h>
 
 #include "cloudwatch_api.h"
 
@@ -991,6 +992,17 @@ int put_log_events(struct flb_cloudwatch *ctx, struct cw_flush *buf,
     flb_sds_t error;
     int num_headers = 1;
 
+    buf->put_events_calls++;
+
+    if (buf->put_events_calls >= 4) {
+        /*
+         * In normal execution, even under high throughput, 4+ calls per flush
+         * should be extremely rare. This is needed for edge cases basically.
+         */
+        flb_plg_debug(ctx->ins, "Too many calls this flush, sleeping for 250 ms");
+        usleep(250000);
+    }
+
     flb_plg_debug(ctx->ins, "Sending log events to log stream %s", stream->name);
 
     /* stream is being used, update expiration */

plugins/out_cloudwatch_logs/cloudwatch_logs.c

@@ -353,6 +353,8 @@ static void cb_cloudwatch_flush(const void *data, size_t bytes,
     (void) i_ins;
     (void) config;
 
+    ctx->buf->put_events_calls = 0;
+
     if (ctx->create_group == FLB_TRUE && ctx->group_created == FLB_FALSE) {
         ret = create_log_group(ctx);
         if (ret < 0) {

plugins/out_cloudwatch_logs/cloudwatch_logs.h

@@ -52,6 +52,17 @@ struct cw_flush {
     /* buffer used to temporarily hold an event during processing */
     char *event_buf;
     size_t event_buf_size;
+
+    /*
+     * According to the docs:
+     * PutLogEvents: 5 requests per second per log stream.
+     * Additional requests are throttled. This quota can't be changed.
+     * This plugin fast. A single flush might make more than 5 calls,
+     * Then fail, then retry, then be too fast again, on and on.
+     * I have seen this happen.
+     * So we throttle ourselves if more than 5 calls are made per flush
+     */
+    int put_events_calls;
 };
 
 struct cw_event {