engine_dispatch: remove chunks from memory if the task fails to be cr…

…eated

When the tasks_map is filled and the new arriving chunks try to create a new
task, then an error is returned. Additionally, under that condition the new
chunks don't get deleted from memory, and they occupy a space of the
storage.max_chunks_up. Eventually, the new chunks end up using the entire space
of the storage.max_chunks_up. This causes the old chunks that have a task
assigned to be unable to be brought up into memory, and therefore they never
get flushed.

This pr fixes the problem by deleting the new chunks from memory in case they
failed to create a task and the filesystem storage is in use. Additionally, it
runs a check to not bring it up to memory in the first place if possible.

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

src/flb_engine_dispatch.c

@@ -29,6 +29,7 @@
 #include <fluent-bit/flb_engine.h>
 #include <fluent-bit/flb_task.h>
 #include <fluent-bit/flb_event.h>
+#include <chunkio/chunkio.h>
 
 
 /* It creates a new output thread using a 'Retry' context */
@@ -270,6 +271,14 @@ int flb_engine_dispatch(uint64_t id, struct flb_input_instance *in,
             continue;
         }
 
+        if (flb_task_map_get_task_id(config) == -1) {
+            /*
+             * There isn't a task available, no more chunks can have a task
+             * assigned.
+             */
+            break;
+        }
+
         /* There is a match, get the buffer */
         buf_data = flb_input_chunk_flush(ic, &buf_size);
         if (buf_size == 0) {
@@ -312,6 +321,12 @@ int flb_engine_dispatch(uint64_t id, struct flb_input_instance *in,
              */
             if (t_err == FLB_TRUE) {
                 flb_input_chunk_release_lock(ic);
+
+                /*
+                 * If the Storage type is 'filesystem' we need to put
+                 * the file content down.
+                 */
+                flb_input_chunk_down(ic);
             }
             continue;
         }