diff --git a/include/fluent-bit/flb_config.h b/include/fluent-bit/flb_config.h index 3ebcb5edbae..1df9c762722 100644 --- a/include/fluent-bit/flb_config.h +++ b/include/fluent-bit/flb_config.h @@ -21,6 +21,7 @@ #define FLB_CONFIG_H #include +#include #include #include @@ -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 */ diff --git a/include/fluent-bit/flb_engine.h b/include/fluent-bit/flb_engine.h index 0fcd522ca53..d8ff0214a6f 100644 --- a/include/fluent-bit/flb_engine.h +++ b/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); diff --git a/include/fluent-bit/flb_engine_macros.h b/include/fluent-bit/flb_engine_macros.h index ef05a97cce8..4ce4462518f 100644 --- a/include/fluent-bit/flb_engine_macros.h +++ b/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 diff --git a/src/flb_engine.c b/src/flb_engine.c index 63178da843c..51bbaf16e2e 100644 --- a/src/flb_engine.c +++ b/src/flb_engine.c @@ -23,6 +23,7 @@ #include #include +#include #include #include #include @@ -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) { diff --git a/src/http_server/api/v2/CMakeLists.txt b/src/http_server/api/v2/CMakeLists.txt index 6b8dc283a0c..2a58c0b3d2a 100644 --- a/src/http_server/api/v2/CMakeLists.txt +++ b/src/http_server/api/v2/CMakeLists.txt @@ -3,6 +3,7 @@ set(src health.c metrics.c reload.c + flush.c register.c ) diff --git a/src/http_server/api/v2/flush.c b/src/http_server/api/v2/flush.c new file mode 100644 index 00000000000..21bfeadbced --- /dev/null +++ b/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 +#include +#include +#include +#include +#include +#include +#include "flush.h" + +#include + +/* 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); +} diff --git a/src/http_server/api/v2/flush.h b/src/http_server/api/v2/flush.h new file mode 100644 index 00000000000..6ad92e2df18 --- /dev/null +++ b/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 +#include + +int api_v2_flush(struct flb_hs *hs); + +#endif diff --git a/src/http_server/api/v2/register.c b/src/http_server/api/v2/register.c index a57ac41480c..bec827eedea 100644 --- a/src/http_server/api/v2/register.c +++ b/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; } diff --git a/tests/runtime/CMakeLists.txt b/tests/runtime/CMakeLists.txt index a7884894b08..5322b86a979 100644 --- a/tests/runtime/CMakeLists.txt +++ b/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") diff --git a/tests/runtime/core_engine_flush_now.c b/tests/runtime/core_engine_flush_now.c new file mode 100644 index 00000000000..991b5194109 --- /dev/null +++ b/tests/runtime/core_engine_flush_now.c @@ -0,0 +1,147 @@ +/* -*- 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 +#include +#include +#include + +#include "flb_tests_runtime.h" + +/* + * Long enough that the periodic flush timer cannot plausibly fire during + * the test window, so a record only arrives if flb_engine_flush_request() + * actually dispatched it on demand. + */ +#define TEST_FLUSH_INTERVAL_SEC "60" +#define TEST_WAIT_TIMEOUT_MS 3000 + +static pthread_mutex_t result_mutex = PTHREAD_MUTEX_INITIALIZER; +static int result_count = 0; + +static int get_result_count(void) +{ + int ret; + + pthread_mutex_lock(&result_mutex); + ret = result_count; + pthread_mutex_unlock(&result_mutex); + + return ret; +} + +static void inc_result_count(void) +{ + pthread_mutex_lock(&result_mutex); + result_count++; + pthread_mutex_unlock(&result_mutex); +} + +static int cb_count_record(void *record, size_t size, void *data) +{ + (void) size; + (void) data; + + inc_result_count(); + flb_free(record); + return 0; +} + +static void wait_for_result(uint32_t timeout_ms, int *count) +{ + struct flb_time start_time; + struct flb_time end_time; + struct flb_time diff_time; + uint64_t elapsed_ms; + + flb_time_get(&start_time); + + while (true) { + *count = get_result_count(); + if (*count > 0) { + return; + } + + flb_time_msleep(20); + flb_time_get(&end_time); + flb_time_diff(&end_time, &start_time, &diff_time); + elapsed_ms = flb_time_to_nanosec(&diff_time) / 1000000; + + if (elapsed_ms > timeout_ms) { + return; + } + } +} + +/* + * flb_engine_flush_request() must dispatch a buffered record immediately, + * without waiting for the (deliberately very long) periodic flush timer. + */ +void flb_test_flush_now_dispatches_immediately(void) +{ + flb_ctx_t *ctx; + struct flb_lib_out_cb cb_data; + int in_ffd; + int out_ffd; + int ret; + int count = 0; + char *input_json = "[1, {\"msg\": \"flush now test\"}]"; + + ctx = flb_create(); + TEST_CHECK(ctx != NULL); + + TEST_CHECK(flb_service_set(ctx, + "Flush", TEST_FLUSH_INTERVAL_SEC, + "Grace", "1", + "Log_Level", "error", + NULL) == 0); + + in_ffd = flb_input(ctx, (char *) "lib", NULL); + TEST_CHECK(in_ffd >= 0); + + cb_data.cb = cb_count_record; + cb_data.data = NULL; + + out_ffd = flb_output(ctx, (char *) "lib", (void *) &cb_data); + TEST_CHECK(out_ffd >= 0); + TEST_CHECK(flb_output_set(ctx, out_ffd, "match", "*", NULL) == 0); + + ret = flb_start(ctx); + TEST_CHECK_(ret == 0, "starting engine"); + + ret = flb_lib_push(ctx, in_ffd, input_json, strlen(input_json)); + TEST_CHECK_(ret >= 0, "pushing record"); + + TEST_CHECK(flb_engine_flush_request(ctx->config) >= 0); + + wait_for_result(TEST_WAIT_TIMEOUT_MS, &count); + TEST_CHECK_(count > 0, + "expected the record to arrive within %dms via on-demand " + "flush (flush interval is %ss); got count=%d", + TEST_WAIT_TIMEOUT_MS, TEST_FLUSH_INTERVAL_SEC, count); + + flb_stop(ctx); + flb_destroy(ctx); +} + +/* Test list */ +TEST_LIST = { + {"flush_now_dispatches_immediately", flb_test_flush_now_dispatches_immediately}, + {NULL, NULL} +};