engine: http_server: Add on-demand flush mechanism

include/fluent-bit/flb_config.h

@@ -21,6 +21,7 @@
 #define FLB_CONFIG_H
 
 #include <time.h>
+#include <stdint.h>
 
 #include <fluent-bit/flb_info.h>
 #include <fluent-bit/flb_pipe.h>
@@ -70,10 +71,11 @@ struct flb_config {
      * shutdown when all remaining tasks are flushed
      */
     int grace;
-    int grace_count;          /* Count of grace shutdown tries              */
-    int grace_input;          /* Shutdown grace to keep inputs ingesting    */
-    flb_pipefd_t flush_fd;    /* Timer FD associated to flush               */
-    int convert_nan_to_null;  /* Convert null to nan ?                      */
+    int grace_count;              /* Count of grace shutdown tries           */
+    int grace_input;              /* Shutdown grace to keep inputs ingesting */
+    flb_pipefd_t flush_fd;        /* Timer FD associated to flush            */
+    uint64_t flush_now_count;     /* count of on-demand flush requests       */
+    int convert_nan_to_null;      /* Convert null to nan ?                   */
 
     int daemon;               /* Run as a daemon ?              */
     flb_pipefd_t shutdown_fd; /* Shutdown FD, 5 seconds         */

include/fluent-bit/flb_engine.h

@@ -34,6 +34,7 @@ int flb_engine_flush(struct flb_config *config,
                      struct flb_input_plugin *in_force);
 int flb_engine_exit(struct flb_config *config);
 int flb_engine_exit_status(struct flb_config *config, int status);
+int flb_engine_flush_request(struct flb_config *config);
 int flb_engine_shutdown(struct flb_config *config);
 int flb_engine_destroy_tasks(struct mk_list *tasks);
 void flb_engine_reschedule_retries(struct flb_config *config);

include/fluent-bit/flb_engine_macros.h

@@ -42,16 +42,18 @@
 #define FLB_ENGINE_EV_NOTIFICATION  (1 << 19)                          /* 524288 */
 
 /* Engine events: all engine events set the left 32 bits to '1' */
-#define FLB_ENGINE_EV_STARTED   FLB_BITS_U64_SET(1, 1) /* Engine started    */
-#define FLB_ENGINE_EV_FAILED    FLB_BITS_U64_SET(1, 2) /* Engine started    */
-#define FLB_ENGINE_EV_STOP      FLB_BITS_U64_SET(1, 3) /* Requested to stop */
-#define FLB_ENGINE_EV_SHUTDOWN  FLB_BITS_U64_SET(1, 4) /* Engine shutdown   */
+#define FLB_ENGINE_EV_STARTED   FLB_BITS_U64_SET(1, 1) /* Engine started          */
+#define FLB_ENGINE_EV_FAILED    FLB_BITS_U64_SET(1, 2) /* Engine started          */
+#define FLB_ENGINE_EV_STOP      FLB_BITS_U64_SET(1, 3) /* Requested to stop       */
+#define FLB_ENGINE_EV_SHUTDOWN  FLB_BITS_U64_SET(1, 4) /* Engine shutdown         */
+#define FLB_ENGINE_EV_FLUSH_NOW FLB_BITS_U64_SET(1, 5) /* On-demand flush request */
 
 /* Similar to engine events, but used as return values */
 #define FLB_ENGINE_STARTED      FLB_BITS_U64_LOW(FLB_ENGINE_EV_STARTED)
 #define FLB_ENGINE_FAILED       FLB_BITS_U64_LOW(FLB_ENGINE_EV_FAILED)
 #define FLB_ENGINE_STOP         FLB_BITS_U64_LOW(FLB_ENGINE_EV_STOP)
 #define FLB_ENGINE_SHUTDOWN     FLB_BITS_U64_LOW(FLB_ENGINE_EV_SHUTDOWN)
+#define FLB_ENGINE_FLUSH_NOW    FLB_BITS_U64_LOW(FLB_ENGINE_EV_FLUSH_NOW)
 
 /* Engine signals: Task, it only refer to the type */
 #define FLB_ENGINE_TASK         2

src/flb_engine.c

@@ -23,6 +23,7 @@
 #include <string.h>
 
 #include <monkey/mk_core.h>
+#include <cfl/cfl_atomic.h>
 #include <fluent-bit/flb_bucket_queue.h>
 #include <fluent-bit/flb_event_loop.h>
 #include <fluent-bit/flb_time.h>
@@ -705,6 +706,14 @@ static inline int flb_engine_manager(flb_pipefd_t fd, struct flb_config *config)
             flb_engine_flush(config, NULL);
             return FLB_ENGINE_STOP;
         }
+        else if (key == FLB_ENGINE_FLUSH_NOW) {
+            flb_trace("[engine] on-demand flush requested");
+            flb_engine_reschedule_retries(config);
+            flb_engine_flush(config, NULL);
+            cfl_atomic_store(&config->flush_now_count,
+                             cfl_atomic_load(&config->flush_now_count) + 1);
+            return 0;
+        }
     }
 
     return 0;
@@ -1421,6 +1430,16 @@ int flb_engine_exit(struct flb_config *config)
     return ret;
 }
 
+int flb_engine_flush_request(struct flb_config *config)
+{
+    int ret;
+    uint64_t val;
+
+    val = FLB_ENGINE_EV_FLUSH_NOW;
+    ret = flb_pipe_w(config->ch_manager[1], &val, sizeof(uint64_t));
+    return ret;
+}
+
 /* Stop ingestion and pause all inputs */
 void flb_engine_stop_ingestion(struct flb_config *config)
 {

src/http_server/api/v2/CMakeLists.txt

@@ -3,6 +3,7 @@ set(src
   health.c
   metrics.c
   reload.c
+  flush.c
   register.c
   )
 

src/http_server/api/v2/flush.c

@@ -0,0 +1,173 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2015-2026 The Fluent Bit Authors
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+#include <fluent-bit/flb_info.h>
+#include <fluent-bit/flb_pack.h>
+#include <fluent-bit/flb_sds.h>
+#include <fluent-bit/flb_time.h>
+#include <fluent-bit/flb_engine.h>
+#include <fluent-bit/http_server/flb_hs_utils.h>
+#include <cfl/cfl_atomic.h>
+#include "flush.h"
+
+#include <fluent-bit/flb_http_server.h>
+
+/* Bounded wait for the engine thread to acknowledge a dispatched flush */
+#define FLB_HS_FLUSH_ACK_TIMEOUT_MS 2000
+
+static int wait_for_flush_ack(uint64_t *counter, uint64_t baseline,
+                              int timeout_ms)
+{
+    int waited_ms = 0;
+    const int interval_ms = 2;
+
+    while (cfl_atomic_load(counter) == baseline) {
+        if (waited_ms >= timeout_ms) {
+            return FLB_FALSE;
+        }
+        flb_time_msleep(interval_ms);
+        waited_ms += interval_ms;
+    }
+
+    return FLB_TRUE;
+}
+
+static int handle_flush_request(struct flb_http_response *response,
+                                struct flb_config *config)
+{
+    int ret;
+    int acked;
+    uint64_t baseline;
+    flb_sds_t out_buf;
+    size_t out_size;
+    msgpack_packer mp_pck;
+    msgpack_sbuffer mp_sbuf;
+    int http_status;
+
+    baseline = cfl_atomic_load(&config->flush_now_count);
+
+    ret = flb_engine_flush_request(config);
+    if (ret == -1) {
+        flb_http_response_set_status(response, 500);
+        return flb_http_response_commit(response);
+    }
+
+    acked = wait_for_flush_ack(&config->flush_now_count, baseline,
+                               FLB_HS_FLUSH_ACK_TIMEOUT_MS);
+
+    /* initialize buffers */
+    msgpack_sbuffer_init(&mp_sbuf);
+    msgpack_packer_init(&mp_pck, &mp_sbuf, msgpack_sbuffer_write);
+
+    msgpack_pack_map(&mp_pck, 2);
+    msgpack_pack_str(&mp_pck, 5);
+    msgpack_pack_str_body(&mp_pck, "flush", 5);
+
+    if (acked == FLB_TRUE) {
+        http_status = 200;
+        msgpack_pack_str(&mp_pck, 4);
+        msgpack_pack_str_body(&mp_pck, "done", 4);
+    }
+    else {
+        /* dispatch was requested but not acknowledged within the timeout */
+        http_status = 503;
+        msgpack_pack_str(&mp_pck, 7);
+        msgpack_pack_str_body(&mp_pck, "timeout", 7);
+    }
+
+    msgpack_pack_str(&mp_pck, 15);
+    msgpack_pack_str_body(&mp_pck, "flush_now_count", 15);
+    msgpack_pack_int64(&mp_pck, cfl_atomic_load(&config->flush_now_count));
+
+    /* Export to JSON */
+    out_buf = flb_msgpack_raw_to_json_sds(mp_sbuf.data, mp_sbuf.size, FLB_TRUE);
+    msgpack_sbuffer_destroy(&mp_sbuf);
+    if (!out_buf) {
+        flb_http_response_set_status(response, 500);
+        return flb_http_response_commit(response);
+    }
+    out_size = flb_sds_len(out_buf);
+
+    flb_hs_response_set_payload(response, http_status,
+                                FLB_HS_CONTENT_TYPE_JSON,
+                                out_buf, out_size);
+
+    flb_sds_destroy(out_buf);
+    return 0;
+}
+
+static int handle_get_flush_status(struct flb_http_response *response,
+                                   struct flb_config *config)
+{
+    flb_sds_t out_buf;
+    size_t out_size;
+    msgpack_packer mp_pck;
+    msgpack_sbuffer mp_sbuf;
+
+    /* initialize buffers */
+    msgpack_sbuffer_init(&mp_sbuf);
+    msgpack_packer_init(&mp_pck, &mp_sbuf, msgpack_sbuffer_write);
+
+    msgpack_pack_map(&mp_pck, 1);
+    msgpack_pack_str(&mp_pck, 15);
+    msgpack_pack_str_body(&mp_pck, "flush_now_count", 15);
+    msgpack_pack_int64(&mp_pck, cfl_atomic_load(&config->flush_now_count));
+
+    /* Export to JSON */
+    out_buf = flb_msgpack_raw_to_json_sds(mp_sbuf.data, mp_sbuf.size, FLB_TRUE);
+    msgpack_sbuffer_destroy(&mp_sbuf);
+    if (!out_buf) {
+        flb_http_response_set_status(response, 500);
+        return flb_http_response_commit(response);
+    }
+    out_size = flb_sds_len(out_buf);
+
+    flb_hs_response_set_payload(response, 200,
+                                FLB_HS_CONTENT_TYPE_JSON,
+                                out_buf, out_size);
+
+    flb_sds_destroy(out_buf);
+    return 0;
+}
+
+static int cb_flush(struct flb_hs *hs,
+                    struct flb_http_request *request,
+                    struct flb_http_response *response)
+{
+    struct flb_config *config = hs->config;
+
+    if (request->method == HTTP_METHOD_POST ||
+        request->method == HTTP_METHOD_PUT) {
+        return handle_flush_request(response, config);
+    }
+    else if (request->method == HTTP_METHOD_GET) {
+        return handle_get_flush_status(response, config);
+    }
+
+    flb_http_response_set_status(response, 405);
+    flb_http_response_set_header(response, "Allow", 5, "GET, POST, PUT", 14);
+    return flb_http_response_commit(response);
+}
+
+/* Perform registration */
+int api_v2_flush(struct flb_hs *hs)
+{
+    return flb_hs_register_endpoint(hs, "/api/v2/flush",
+                                    FLB_HS_ROUTE_EXACT, cb_flush);
+}

src/http_server/api/v2/flush.h

@@ -0,0 +1,28 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2015-2026 The Fluent Bit Authors
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+#ifndef FLB_HS_API_V2_FLUSH_H
+#define FLB_HS_API_V2_FLUSH_H
+
+#include <fluent-bit/flb_info.h>
+#include <fluent-bit/flb_http_server.h>
+
+int api_v2_flush(struct flb_hs *hs);
+
+#endif

src/http_server/api/v2/register.c

@@ -23,6 +23,7 @@
 #include "health.h"
 #include "metrics.h"
 #include "reload.h"
+#include "flush.h"
 
 int api_v2_registration(struct flb_hs *hs)
 {
@@ -43,5 +44,10 @@ int api_v2_registration(struct flb_hs *hs)
         return ret;
     }
 
+    ret = api_v2_flush(hs);
+    if (ret != 0) {
+        return ret;
+    }
+
     return 0;
 }

tests/runtime/CMakeLists.txt

@@ -32,6 +32,9 @@ if(NOT FLB_SYSTEM_WINDOWS)
   FLB_RT_CORE_TEST(FLB_CORE_SHUTDOWN_SPIN   "core_shutdown_spin.c")
 endif()
 FLB_RT_CORE_TEST(1 "http_client_chunked.c")
+if(FLB_IN_LIB AND FLB_OUT_LIB)
+  FLB_RT_TEST(1 "core_engine_flush_now.c")
+endif()
 
 FLB_RT_TEST(FLB_CHUNK_TRACE "core_chunk_trace.c")