output: added a flush abortion mechanism

This change introduces an additional check in output_pre_cb_flush to
ensure that chunks can be dropped if they were scheduled to be flushed
but the flush coroutine has not entered the plugins entry point yet.

Signed-off-by: Leonardo Alminana <[email protected]>

Author: Leonardo Alminana

include/fluent-bit/flb_output.h

@@ -498,6 +498,12 @@ struct flb_out_flush_params {
 
 extern FLB_TLS_DEFINE(struct flb_out_flush_params, out_flush_params);
 
+#define FLB_OUTPUT_RETURN(x)                                            \
+    flb_output_return_do(x);                                            \
+    return
+
+static inline void flb_output_return_do(int x);
+
 static FLB_INLINE void output_params_set(struct flb_output_flush *out_flush,
                                          struct flb_coro *coro,
                                          struct flb_task *task,
@@ -537,6 +543,7 @@ static FLB_INLINE void output_params_set(struct flb_output_flush *out_flush,
 
 static FLB_INLINE void output_pre_cb_flush(void)
 {
+    int route_status;
     struct flb_coro *coro;
     struct flb_output_plugin *out_p;
     struct flb_out_flush_params *params;
@@ -557,11 +564,29 @@ static FLB_INLINE void output_pre_cb_flush(void)
      */
     coro = params->coro;
     persisted_params = *params;
+
     co_switch(coro->caller);
 
     /* Continue, we will resume later */
     out_p = persisted_params.out_plugin;
 
+    flb_task_acquire_lock(persisted_params.out_flush->task);
+
+    route_status = flb_task_get_route_status(
+                    persisted_params.out_flush->task,
+                    persisted_params.out_flush->o_ins);
+
+    if (route_status == FLB_TASK_ROUTE_DROPPED) {
+        flb_task_release_lock(persisted_params.out_flush->task);
+
+        FLB_OUTPUT_RETURN(FLB_ERROR);
+    }
+
+    flb_task_activate_route(persisted_params.out_flush->task,
+                            persisted_params.out_flush->o_ins);
+
+    flb_task_release_lock(persisted_params.out_flush->task);
+
     out_p->cb_flush(persisted_params.event_chunk,
                     persisted_params.out_flush,
                     persisted_params.i_ins,
@@ -909,6 +934,12 @@ static inline void flb_output_return(int ret, struct flb_coro *co) {
     o_ins = out_flush->o_ins;
     task = out_flush->task;
 
+    flb_task_acquire_lock(task);
+
+    flb_task_deactivate_route(task, o_ins);
+
+    flb_task_release_lock(task);
+
     if (out_flush->processed_event_chunk) {
         if (task->event_chunk->data != out_flush->processed_event_chunk->data) {
             flb_free(out_flush->processed_event_chunk->data);