input: add thread.ring_buffer.retry_limit option

When a threaded input plugin's ring buffer is full, the input thread
retries writing to the buffer before dropping data. Previously, this
retry limit was hardcoded to 10 (1 second total with 100ms sleep
between retries).

This patch makes the retry limit configurable via the new
'thread.ring_buffer.retry_limit' option. The default value remains 10
for backward compatibility. Increasing this value allows the system
to handle temporary backpressure situations without dropping data.

Fixes #11393

Signed-off-by: jinyong.choi <[email protected]>

Author: jinyongchoi

include/fluent-bit/flb_input.h

@@ -383,6 +383,7 @@ struct flb_input_instance {
     struct flb_ring_buffer *rb;
     size_t ring_buffer_size;           /* ring buffer size */
     uint8_t ring_buffer_window;        /* ring buffer window percentage */
+    int ring_buffer_retry_limit;       /* ring buffer write retry limit */
 
     /* List of upstreams */
     struct mk_list upstreams;

src/flb_input.c

@@ -67,11 +67,17 @@ pthread_key_t libco_in_param_key;
  * ring buffer will emit a flush request whenever the window threshold is reached.
  * The window percentage can be tuned per input instance using the
  * 'thread.ring_buffer.window' property.
+ *
+ * Ring buffer retry limit: when the ring buffer is full, the input thread will
+ * retry writing to the buffer up to 'retry_limit' times (with 100ms sleep between
+ * retries) before dropping the data. The default is 10 retries (1 second total).
+ * This can be tuned per input instance using 'thread.ring_buffer.retry_limit'.
  */
 
 #define FLB_INPUT_RING_BUFFER_CAPACITY 1024
 #define FLB_INPUT_RING_BUFFER_SIZE   (sizeof(void *) * FLB_INPUT_RING_BUFFER_CAPACITY)
 #define FLB_INPUT_RING_BUFFER_WINDOW (5)
+#define FLB_INPUT_RING_BUFFER_RETRY_LIMIT (10)
 
 /* config map to register options available for all input plugins */
 struct flb_config_map input_global_properties[] = {
@@ -138,6 +144,12 @@ struct flb_config_map input_global_properties[] = {
         0, FLB_FALSE, 0,
         "Set custom ring buffer window percentage for threaded inputs"
     },
+    {
+        FLB_CONFIG_MAP_INT, "thread.ring_buffer.retry_limit",
+        STR(FLB_INPUT_RING_BUFFER_RETRY_LIMIT),
+        0, FLB_FALSE, 0,
+        "Set maximum retry attempts when ring buffer is full before dropping data"
+    },
 
     {0}
 };
@@ -420,9 +432,10 @@ struct flb_input_instance *flb_input_new(struct flb_config *config,
 
         }
 
-        /* set default ring buffer size and window */
+        /* set default ring buffer size, window, and retry limit */
         instance->ring_buffer_size = FLB_INPUT_RING_BUFFER_SIZE;
         instance->ring_buffer_window = FLB_INPUT_RING_BUFFER_WINDOW;
+        instance->ring_buffer_retry_limit = FLB_INPUT_RING_BUFFER_RETRY_LIMIT;
 
         /* allocate a ring buffer */
         instance->rb = flb_ring_buffer_create(instance->ring_buffer_size);
@@ -759,6 +772,15 @@ int flb_input_set_property(struct flb_input_instance *ins,
         }
         ins->ring_buffer_window = (uint8_t) ret;
     }
+    else if (prop_key_check("thread.ring_buffer.retry_limit", k, len) == 0 && tmp) {
+        ret = atoi(tmp);
+        flb_sds_destroy(tmp);
+        if (ret <= 0) {
+            flb_error("[input] thread.ring_buffer.retry_limit must be greater than 0");
+            return -1;
+        }
+        ins->ring_buffer_retry_limit = ret;
+    }
     else if (prop_key_check("storage.pause_on_chunks_overlimit", k, len) == 0 && tmp) {
         ret = flb_utils_bool(tmp);
         flb_sds_destroy(tmp);

src/flb_input_chunk.c

@@ -2945,7 +2945,6 @@ static int append_to_ring_buffer(struct flb_input_instance *ins,
 {
     int ret;
     int retries = 0;
-    int retry_limit = 10;
     struct input_chunk_raw *cr;
 
     if (buf_size == 0) {
@@ -2995,9 +2994,9 @@ static int append_to_ring_buffer(struct flb_input_instance *ins,
     /*
      * There is a little chance that the ring buffer is full or due to saturation
      * from the main thread the data is not being consumed. On this scenario we
-     * retry up to 'retry_limit' times with a little wait time.
+     * retry up to 'ring_buffer_retry_limit' times with a little wait time.
      */
-    if (retries >= retry_limit) {
+    if (retries >= ins->ring_buffer_retry_limit) {
         flb_plg_error(ins, "could not enqueue records into the ring buffer");
         destroy_chunk_raw(cr);