input: make input threads ring_buffer capacity and window flush configurable

Input plugins running in threaded mode use a per-instance ring buffer to enqueue records into
the pipeline, which are then consumed by the engine. By default, each ring buffer is sized for
1024 entries.

In high-throughput scenarios, the default capacity may not be sufficient. This patch introduces two
new configuration options to allow fine-tuning of the ring buffer behavior per input plugin:

 - thread.ring_buffer.capacity: Sets the maximum number of entries the ring buffer can hold (default: 1024).

 - thread.ring_buffer.window: Sets the flush window as a percentage of capacity, used to trigger a flush
                              request when the threshold is reached (default: 5).

Signed-off-by: Eduardo Silva <[email protected]>

Author: edsiper

include/fluent-bit/flb_input.h

@@ -379,6 +379,8 @@ struct flb_input_instance {
      * in the ring buffer.
      */
     struct flb_ring_buffer *rb;
+    size_t ring_buffer_size;           /* ring buffer size */
+    uint8_t ring_buffer_window;        /* ring buffer window percentage */
 
     /* List of upstreams */
     struct mk_list upstreams;

src/flb_input.c

@@ -56,16 +56,20 @@ pthread_key_t libco_in_param_key;
 #define protcmp(a, b)  strncasecmp(a, b, strlen(a))
 
 /*
- * Ring buffer size: we make space for 512 entries that each input instance can
- * use to enqueue data. Note that this value is fixed and only affect input plugins
- * which runs in threaded mode (separate thread)
+ * Ring buffer capacity: by default we make space for 1024 entries that each
+ * input instance can use to enqueue data. The capacity can be customized per
+ * input instance through the 'thread.ring_buffer.capacity' property. The value
+ * represents the number of slots and is converted to bytes when the ring buffer
+ * is created. This affects only input plugins running in threaded mode.
  *
  * Ring buffer window: the current window size is set to 5% which means that the
- * ring buffer will emit a flush request whenever there are 51 records or more
- * awaiting to be consumed.
+ * 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.
  */
 
-#define FLB_INPUT_RING_BUFFER_SIZE   (sizeof(void *) * 1024)
+#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)
 
 /* config map to register options available for all input plugins */
@@ -123,6 +127,16 @@ struct flb_config_map input_global_properties[] = {
         0, FLB_FALSE, 0,
         "Enable threading on an input"
     },
+    {
+        FLB_CONFIG_MAP_INT, "thread.ring_buffer.capacity", "0",
+        0, FLB_FALSE, 0,
+        "Set custom ring buffer capacity when the input runs in threaded mode"
+    },
+    {
+        FLB_CONFIG_MAP_INT, "thread.ring_buffer.window", "0",
+        0, FLB_FALSE, 0,
+        "Set custom ring buffer window percentage for threaded inputs"
+    },
 
     {0}
 };
@@ -404,8 +418,12 @@ struct flb_input_instance *flb_input_new(struct flb_config *config,
 
         }
 
+        /* set default ring buffer size and window */
+        instance->ring_buffer_size = FLB_INPUT_RING_BUFFER_SIZE;
+        instance->ring_buffer_window = FLB_INPUT_RING_BUFFER_WINDOW;
+
         /* allocate a ring buffer */
-        instance->rb = flb_ring_buffer_create(FLB_INPUT_RING_BUFFER_SIZE);
+        instance->rb = flb_ring_buffer_create(instance->ring_buffer_size);
         if (!instance->rb) {
             flb_error("instance %s could not initialize ring buffer",
                       flb_input_name(instance));
@@ -704,6 +722,30 @@ int flb_input_set_property(struct flb_input_instance *ins,
 
         ins->is_threaded = enabled;
     }
+    else if (prop_key_check("thread.ring_buffer.capacity", k, len) == 0 && tmp) {
+        ret = atoi(tmp);
+        flb_sds_destroy(tmp);
+        if (ret <= 0) {
+            return -1;
+        }
+        ins->ring_buffer_size = (size_t) ret * sizeof(void *);
+        if (ins->rb) {
+            flb_ring_buffer_destroy(ins->rb);
+            ins->rb = flb_ring_buffer_create(ins->ring_buffer_size);
+            if (!ins->rb) {
+                flb_error("instance %s could not initialize ring buffer", flb_input_name(ins));
+                return -1;
+            }
+        }
+    }
+    else if (prop_key_check("thread.ring_buffer.window", k, len) == 0 && tmp) {
+        ret = atoi(tmp);
+        flb_sds_destroy(tmp);
+        if (ret <= 0 || ret > 100) {
+            return -1;
+        }
+        ins->ring_buffer_window = (uint8_t) ret;
+    }
     else if (prop_key_check("storage.pause_on_chunks_overlimit", k, len) == 0 && tmp) {
         ret = flb_utils_bool(tmp);
         flb_sds_destroy(tmp);
@@ -1320,7 +1362,7 @@ int flb_input_instance_init(struct flb_input_instance *ins,
             }
 
             /* register the ring buffer */
-            ret = flb_ring_buffer_add_event_loop(ins->rb, config->evl, FLB_INPUT_RING_BUFFER_WINDOW);
+            ret = flb_ring_buffer_add_event_loop(ins->rb, config->evl, ins->ring_buffer_window);
             if (ret) {
                 flb_error("failed while registering ring buffer events on input %s",
                           ins->name);