out_s3: implement retry_limit parameter

retry_limit parameter is not honored and is set to 5. This feature provides dynamic retry_limit based on configuration for out_s3 plugin.

Signed-off-by: usharma <[email protected]>

out_s3: fix retry_limit as it is part of default instance

Fixed the retry_limit as it was not needed and it part of default instance

Signed-off-by: usharma <[email protected]>

out_s3: fix infinite retry_limit in cb_s3_init

Fixes the infinite retry limit to MAX_UPLOAD_ERRORS(old behaviour) in cb_s3_init. If negative retry limit is set, the retry limit is set to MAX_UPLOAD_ERRORS

Signed-off-by: usharma <[email protected]>

out_s3: fix infinite retry_limit in cb_s3_init

Fixes the infinite retry limit to MAX_UPLOAD_ERRORS(old behaviour) in cb_s3_init. If negative retry limit is set, the retry limit is set to MAX_UPLOAD_ERRORS

Signed-off-by: usharma <[email protected]>

Author: usharma

plugins/out_s3/s3.c

@@ -617,6 +617,10 @@ static int cb_s3_init(struct flb_output_instance *ins,
     ctx->retry_time = 0;
     ctx->upload_queue_success = FLB_FALSE;
 
+    if(ctx->ins->retry_limit < 0) {
+        ctx->ins->retry_limit = MAX_UPLOAD_ERRORS;
+    }
+
     /* Export context */
     flb_output_set_context(ins, ctx);
 
@@ -1367,11 +1371,10 @@ static int put_all_chunks(struct flb_s3 *ctx)
                 continue;
             }
 
-            if (chunk->failures >= MAX_UPLOAD_ERRORS) {
+            if (chunk->failures >= ctx->ins->retry_limit) {
                 flb_plg_warn(ctx->ins,
-                             "Chunk for tag %s failed to send %i times, "
-                             "will not retry",
-                             (char *) fsf->meta_buf, MAX_UPLOAD_ERRORS);
+                             "Chunk for tag %s failed to send %d/%d times, will not retry",
+                             (char *) fsf->meta_buf, chunk->failures, ctx->ins->retry_limit);
                 flb_fstore_file_inactive(ctx->fs, fsf);
                 continue;
             }
@@ -1656,7 +1659,7 @@ static struct multipart_upload *get_upload(struct flb_s3 *ctx,
         if (tmp_upload->upload_state == MULTIPART_UPLOAD_STATE_COMPLETE_IN_PROGRESS) {
             continue;
         }
-        if (tmp_upload->upload_errors >= MAX_UPLOAD_ERRORS) {
+        if (tmp_upload->upload_errors >= ctx->ins->retry_limit) {
             tmp_upload->upload_state = MULTIPART_UPLOAD_STATE_COMPLETE_IN_PROGRESS;
             flb_plg_error(ctx->ins, "Upload for %s has reached max upload errors",
                           tmp_upload->s3_key);
@@ -1902,7 +1905,7 @@ static void s3_upload_queue(struct flb_config *config, void *out_context)
 
             /* If retry limit was reached, discard file and remove file from queue */
             upload_contents->retry_counter++;
-            if (upload_contents->retry_counter >= MAX_UPLOAD_ERRORS) {
+            if (upload_contents->retry_counter >= ctx->ins->retry_limit) {
                 flb_plg_warn(ctx->ins, "Chunk file failed to send %d times, will not "
                              "retry", upload_contents->retry_counter);
                 s3_store_file_inactive(ctx, upload_contents->upload_file);
@@ -3303,6 +3306,13 @@ static void cb_s3_upload(struct flb_config *config, void *data)
         if (ret != FLB_OK) {
             flb_plg_error(ctx->ins, "Could not send chunk with tag %s",
                           (char *) fsf->meta_buf);
+            if(chunk->failures >= ctx->ins->retry_limit){
+                flb_plg_warn(ctx->ins,
+                             "Chunk for tag %s failed to send %d/%d times, will not retry",
+                             (char *) fsf->meta_buf, chunk->failures, ctx->ins->retry_limit);
+                flb_fstore_file_inactive(ctx->fs, fsf);
+                continue;
+            }
         }
     }
 
@@ -3311,7 +3321,7 @@ static void cb_s3_upload(struct flb_config *config, void *data)
         m_upload = mk_list_entry(head, struct multipart_upload, _head);
         complete = FLB_FALSE;
 
-        if (m_upload->complete_errors >= MAX_UPLOAD_ERRORS) {
+        if (m_upload->complete_errors >= ctx->ins->retry_limit) {
             flb_plg_error(ctx->ins,
                           "Upload for %s has reached max completion errors, "
                           "plugin will give up", m_upload->s3_key);
@@ -3823,10 +3833,10 @@ static void cb_s3_flush(struct flb_event_chunk *event_chunk,
                         m_upload_file, file_first_log_time);
     }
 
-    /* Discard upload_file if it has failed to upload MAX_UPLOAD_ERRORS times */
-    if (upload_file != NULL && upload_file->failures >= MAX_UPLOAD_ERRORS) {
-        flb_plg_warn(ctx->ins, "File with tag %s failed to send %d times, will not "
-                     "retry", event_chunk->tag, MAX_UPLOAD_ERRORS);
+    /* Discard upload_file if it has failed to upload retry_limit times */
+    if (upload_file != NULL && upload_file->failures >= ctx->ins->retry_limit) {
+        flb_plg_warn(ctx->ins, "File with tag %s failed to send %d/%d times, will not retry",
+                     event_chunk->tag, upload_file->failures, ctx->ins->retry_limit);
         s3_store_file_inactive(ctx, upload_file);
         upload_file = NULL;
     }

plugins/out_s3/s3.h

@@ -48,6 +48,8 @@
 
 #define DEFAULT_UPLOAD_TIMEOUT 3600
 
+#define MAX_UPLOAD_ERRORS 5
+
 /*
  * If we see repeated errors on an upload/chunk, we will discard it
  * This saves us from scenarios where something goes wrong and an upload can
@@ -56,8 +58,9 @@
  *
  * The same is done for chunks, just to be safe, even though realistically
  * I can't think of a reason why a chunk could become unsendable.
+ *
+ * The retry limit is now configurable via the retry_limit parameter.
  */
-#define MAX_UPLOAD_ERRORS 5
 
 struct upload_queue {
     struct s3_file *upload_file;
@@ -96,7 +99,7 @@ struct multipart_upload {
 
     struct mk_list _head;
 
-    /* see note for MAX_UPLOAD_ERRORS */
+    /* see note for retry_limit configuration */
     int upload_errors;
     int complete_errors;
 };