config: engine: Implement adaptive flushing

conf/fluent-bit.conf

@@ -4,6 +4,57 @@
     # set an interval of seconds before to flush records to a destination
     flush        1
 
+    # flush.adaptive
+    # --------------
+    # Enable adaptive flush interval adjustments based on output chunk
+    # backpressure.
+    #
+    # flush.adaptive off
+
+    # flush.adaptive.min_interval
+    # ---------------------------
+    # lower bound for adaptive flush interval in seconds.
+    #
+    # flush.adaptive.min_interval 0.5
+
+    # flush.adaptive.max_interval
+    # ---------------------------
+    # upper bound for adaptive flush interval in seconds.
+    #
+    # flush.adaptive.max_interval 2.0
+
+    # flush.adaptive.low_pressure
+    # ---------------------------
+    # output chunk pressure (%) threshold considered mostly idle.
+    #
+    # flush.adaptive.low_pressure 25
+
+    # flush.adaptive.medium_pressure
+    # ------------------------------
+    # output chunk pressure (%) threshold considered moderate pressure.
+    #
+    # flush.adaptive.medium_pressure 50
+
+    # flush.adaptive.high_pressure
+    # ----------------------------
+    # output chunk pressure (%) threshold considered sustained pressure.
+    #
+    # flush.adaptive.high_pressure 75
+
+    # flush.adaptive.up_steps
+    # -----------------------
+    # consecutive pressure samples required before moving to a faster flush
+    # step. minimum: 1
+    #
+    # flush.adaptive.up_steps 2
+
+    # flush.adaptive.down_steps
+    # -------------------------
+    # consecutive idle samples required before moving to a slower flush step.
+    # minimum: 1
+    #
+    # flush.adaptive.down_steps 3
+
     # Daemon
     # ======
     # instruct Fluent Bit to run in foreground or background mode.

include/fluent-bit/flb_config.h

@@ -64,6 +64,18 @@ struct flb_config {
     int is_shutting_down;     /* is the service shutting down ? */
     int is_running;           /* service running ?              */
     double flush;             /* Flush timeout                  */
+    int flush_adaptive;       /* Enable adaptive flush interval */
+    double flush_adaptive_min_interval;
+    double flush_adaptive_max_interval;
+    double flush_adaptive_low_pressure;
+    double flush_adaptive_medium_pressure;
+    double flush_adaptive_high_pressure;
+    int flush_adaptive_up_steps;
+    int flush_adaptive_down_steps;
+    int flush_adaptive_level;
+    int flush_adaptive_hits;
+    int flush_adaptive_direction;
+    double flush_adaptive_current_interval;
 
     /*
      * Maximum grace time on shutdown. If set to -1, the engine will
@@ -368,6 +380,14 @@ enum conf_type {
 };
 
 #define FLB_CONF_STR_FLUSH        "Flush"
+#define FLB_CONF_STR_FLUSH_ADAPTIVE "flush.adaptive"
+#define FLB_CONF_STR_FLUSH_ADAPTIVE_MIN "flush.adaptive.min_interval"
+#define FLB_CONF_STR_FLUSH_ADAPTIVE_MAX "flush.adaptive.max_interval"
+#define FLB_CONF_STR_FLUSH_ADAPTIVE_LOW "flush.adaptive.low_pressure"
+#define FLB_CONF_STR_FLUSH_ADAPTIVE_MEDIUM "flush.adaptive.medium_pressure"
+#define FLB_CONF_STR_FLUSH_ADAPTIVE_HIGH "flush.adaptive.high_pressure"
+#define FLB_CONF_STR_FLUSH_ADAPTIVE_UP_STEPS "flush.adaptive.up_steps"
+#define FLB_CONF_STR_FLUSH_ADAPTIVE_DOWN_STEPS "flush.adaptive.down_steps"
 #define FLB_CONF_STR_GRACE        "Grace"
 #define FLB_CONF_STR_DAEMON       "Daemon"
 #define FLB_CONF_STR_LOGFILE      "Log_File"

include/fluent-bit/flb_engine.h

@@ -39,6 +39,12 @@ int flb_engine_destroy_tasks(struct mk_list *tasks);
 void flb_engine_reschedule_retries(struct flb_config *config);
 void flb_engine_stop_ingestion(struct flb_config *config);
 
+/* Adaptive flush helpers (also used by internal tests) */
+int flb_engine_adaptive_flush_target_level(struct flb_config *config,
+                                           double pressure);
+double flb_engine_adaptive_flush_interval(struct flb_config *config,
+                                          int level);
+
 /* Engine event loop */
 void flb_engine_evl_init();
 struct mk_event_loop *flb_engine_evl_get();

src/flb_config.c

@@ -55,6 +55,38 @@ struct flb_service_config service_configs[] = {
      FLB_CONF_TYPE_DOUBLE,
      offsetof(struct flb_config, flush)},
 
+    {FLB_CONF_STR_FLUSH_ADAPTIVE,
+     FLB_CONF_TYPE_BOOL,
+     offsetof(struct flb_config, flush_adaptive)},
+
+    {FLB_CONF_STR_FLUSH_ADAPTIVE_MIN,
+     FLB_CONF_TYPE_DOUBLE,
+     offsetof(struct flb_config, flush_adaptive_min_interval)},
+
+    {FLB_CONF_STR_FLUSH_ADAPTIVE_MAX,
+     FLB_CONF_TYPE_DOUBLE,
+     offsetof(struct flb_config, flush_adaptive_max_interval)},
+
+    {FLB_CONF_STR_FLUSH_ADAPTIVE_LOW,
+     FLB_CONF_TYPE_DOUBLE,
+     offsetof(struct flb_config, flush_adaptive_low_pressure)},
+
+    {FLB_CONF_STR_FLUSH_ADAPTIVE_MEDIUM,
+     FLB_CONF_TYPE_DOUBLE,
+     offsetof(struct flb_config, flush_adaptive_medium_pressure)},
+
+    {FLB_CONF_STR_FLUSH_ADAPTIVE_HIGH,
+     FLB_CONF_TYPE_DOUBLE,
+     offsetof(struct flb_config, flush_adaptive_high_pressure)},
+
+    {FLB_CONF_STR_FLUSH_ADAPTIVE_UP_STEPS,
+     FLB_CONF_TYPE_INT,
+     offsetof(struct flb_config, flush_adaptive_up_steps)},
+
+    {FLB_CONF_STR_FLUSH_ADAPTIVE_DOWN_STEPS,
+     FLB_CONF_TYPE_INT,
+     offsetof(struct flb_config, flush_adaptive_down_steps)},
+
     {FLB_CONF_STR_GRACE,
      FLB_CONF_TYPE_INT,
      offsetof(struct flb_config, grace)},
@@ -270,6 +302,18 @@ struct flb_config *flb_config_init()
 
     /* Flush */
     config->flush        = FLB_CONFIG_FLUSH_SECS;
+    config->flush_adaptive = FLB_FALSE;
+    config->flush_adaptive_min_interval = 0.5;
+    config->flush_adaptive_max_interval = 2.0;
+    config->flush_adaptive_low_pressure = 25.0;
+    config->flush_adaptive_medium_pressure = 50.0;
+    config->flush_adaptive_high_pressure = 75.0;
+    config->flush_adaptive_up_steps = 2;
+    config->flush_adaptive_down_steps = 3;
+    config->flush_adaptive_level = 1;
+    config->flush_adaptive_hits = 0;
+    config->flush_adaptive_direction = 0;
+    config->flush_adaptive_current_interval = FLB_CONFIG_FLUSH_SECS;
     config->daemon       = FLB_FALSE;
     config->init_time    = time(NULL);
     config->kernel       = flb_kernel_info();