in_emitter: Add backpressure limitation for filesystem storage

Signed-off-by: Hiroshi Hatake <[email protected]>

Author: cosmo0920

plugins/in_emitter/emitter.c

@@ -25,6 +25,7 @@
 #include <fluent-bit/flb_sds.h>
 #include <fluent-bit/flb_scheduler.h>
 #include <fluent-bit/flb_ring_buffer.h>
+#include <fluent-bit/flb_storage.h>
 
 #include <sys/types.h>
 #include <sys/stat.h>
@@ -318,6 +319,29 @@ static int cb_emitter_init(struct flb_input_instance *in,
         return -1;
     }
 
+    /*
+     * The emitter is used internally by filters such as rewrite_tag. When the
+     * downstream outputs experience backpressure, the emitter needs to pause
+     * its upstream senders to avoid holding an arbitrary number of "up"
+     * chunks in memory. Without pausing on the filesystem storage limit, the
+     * emitter can continue to accumulate in-memory chunks (for example, in a
+     * rewrite_tag pipeline) even though storage.max_chunks_up intends to cap
+     * usage. Enable pausing on the storage chunks limit by default when
+     * filesystem storage is in use so the configured storage.max_chunks_up
+     * limit is honored.
+     */
+    if (in->storage_type == FLB_STORAGE_FS &&
+        in->storage_pause_on_chunks_overlimit == FLB_FALSE) {
+        in->storage_pause_on_chunks_overlimit = FLB_TRUE;
+        flb_plg_debug(in, "enable pause on storage chunks overlimit for emitter");
+    }
+    else if (in->storage_type != FLB_STORAGE_FS &&
+             in->storage_pause_on_chunks_overlimit == FLB_TRUE) {
+        flb_plg_debug(in,
+                      "storage.pause_on_chunks_overlimit reset: storage.type is not filesystem");
+        in->storage_pause_on_chunks_overlimit = FLB_FALSE;
+    }
+
     if (in->is_threaded == FLB_TRUE && ctx->ring_buffer_size == 0) {
         ctx->ring_buffer_size = DEFAULT_EMITTER_RING_BUFFER_FLUSH_FREQUENCY;
         flb_plg_debug(in, "threaded: enable emitter ring buffer (size=%u)",