From 6a9272f321e2bc80daf2ad8736a216030d42abc5 Mon Sep 17 00:00:00 2001 From: Eduardo Silva Date: Mon, 3 Aug 2026 13:22:47 -0600 Subject: [PATCH 01/21] network: preserve nonblocking accept errors Signed-off-by: Eduardo Silva --- src/flb_network.c | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/src/flb_network.c b/src/flb_network.c index d4572846791..61bfdb07f06 100644 --- a/src/flb_network.c +++ b/src/flb_network.c @@ -1924,10 +1924,14 @@ flb_sockfd_t flb_net_accept(flb_sockfd_t server_fd) SOCK_NONBLOCK | SOCK_CLOEXEC); #else remote_fd = accept(server_fd, (struct sockaddr*)&sock_addr, &socket_size); - flb_net_socket_nonblocking(remote_fd); + + if (remote_fd != FLB_INVALID_SOCKET) { + flb_net_socket_nonblocking(remote_fd); + } #endif - if (remote_fd == -1) { + if (remote_fd == FLB_INVALID_SOCKET && + !FLB_WOULDBLOCK()) { perror("accept4"); } From 7b927bd048a478a6dfc80602014cf20586c778e7 Mon Sep 17 00:00:00 2001 From: Eduardo Silva Date: Mon, 3 Aug 2026 13:22:52 -0600 Subject: [PATCH 02/21] downstream: accept connections in event coroutines Signed-off-by: Eduardo Silva --- include/fluent-bit/flb_connection.h | 6 + include/fluent-bit/flb_downstream.h | 7 + src/flb_downstream.c | 233 +++++++++++++++++++++++++++- 3 files changed, 240 insertions(+), 6 deletions(-) diff --git a/include/fluent-bit/flb_connection.h b/include/fluent-bit/flb_connection.h index 3eed648e61d..f546cfa1d70 100644 --- a/include/fluent-bit/flb_connection.h +++ b/include/fluent-bit/flb_connection.h @@ -62,6 +62,9 @@ struct flb_connection; typedef void (*flb_connection_drop_notification_callback)( struct flb_connection *connection); +typedef int (*flb_connection_accept_callback)( + struct flb_connection *connection, + void *data); typedef int (*flb_connection_event_callback)(void *data); /* Base network connection */ @@ -79,6 +82,9 @@ struct flb_connection { * teardown immediately after the callback returns. */ flb_connection_drop_notification_callback drop_notification_callback; + flb_connection_accept_callback accept_callback; + void *accept_callback_data; + int event_registration_mask; /* Socket */ flb_sockfd_t fd; diff --git a/include/fluent-bit/flb_downstream.h b/include/fluent-bit/flb_downstream.h index 58c92155efb..fe8108533b9 100644 --- a/include/fluent-bit/flb_downstream.h +++ b/include/fluent-bit/flb_downstream.h @@ -86,11 +86,18 @@ void flb_downstream_pause(struct flb_downstream *stream); void flb_downstream_resume(struct flb_downstream *stream); int flb_downstream_conn_release(struct flb_connection *connection); +int flb_downstream_conn_release_all(struct flb_downstream *stream); /* * The callback and any ingestion it invokes run on config->coro_stack_size. * Callers must size that stack for their complete callback path. */ +int flb_downstream_conn_event_accept( + struct flb_downstream *stream, + flb_connection_accept_callback accept_callback, + void *accept_callback_data, + flb_connection_event_callback event_callback, + int mask); int flb_downstream_conn_event_register(struct flb_connection *connection, int (*callback)(void *data), int mask); diff --git a/src/flb_downstream.c b/src/flb_downstream.c index 55b2bcdd33f..63522e362a6 100644 --- a/src/flb_downstream.c +++ b/src/flb_downstream.c @@ -34,15 +34,18 @@ #include static inline int prepare_destroy_conn_safe(struct flb_connection *connection); +static void resume_pending_event_coroutines(struct flb_downstream *stream); -static void flb_downstream_conn_event_coro(void) +static void flb_downstream_conn_event_coro_terminate(struct flb_coro *coro) { - struct flb_coro *coro; - struct flb_connection *connection; - - coro = flb_coro_get(); - connection = coro->data; + while (FLB_TRUE) { + flb_coro_yield(coro, FLB_FALSE); + } +} +static void flb_downstream_conn_event_loop(struct flb_connection *connection, + struct flb_coro *coro) +{ while (FLB_TRUE) { flb_coro_yield(coro, FLB_FALSE); @@ -70,6 +73,92 @@ static void flb_downstream_conn_event_coro(void) } } +static void flb_downstream_conn_async_event_loop( + struct flb_connection *connection, + struct flb_coro *coro) +{ + while (FLB_TRUE) { + connection->busy_flag = FLB_TRUE; + connection->event_callback(connection); + connection->busy_flag = FLB_FALSE; + + if (connection->event_release_pending == FLB_TRUE) { + connection->event_release_pending = FLB_FALSE; + prepare_destroy_conn_safe(connection); + } + + if (connection->fd == FLB_INVALID_SOCKET) { + connection->coroutine = NULL; + flb_downstream_conn_event_coro_terminate(coro); + } + + connection->coroutine = coro; + } +} + +static void flb_downstream_conn_event_coro(void) +{ + struct flb_coro *coro; + struct flb_connection *connection; + + coro = flb_coro_get(); + connection = coro->data; + + flb_downstream_conn_event_loop(connection, coro); +} + +static void flb_downstream_conn_accept_event_coro(void) +{ + int result; + struct flb_coro *coro; + struct flb_connection *connection; + + coro = flb_coro_get(); + connection = coro->data; + + flb_connection_reset_connection_timeout(connection); + result = flb_io_net_accept(connection, coro); + flb_connection_unset_connection_timeout(connection); + + if (result == 0 && + connection->event_release_pending == FLB_FALSE && + connection->downstream->paused == FLB_FALSE) { + result = connection->accept_callback( + connection, + connection->accept_callback_data); + } + + if (result == 0 && + connection->event_release_pending == FLB_FALSE && + connection->fd != FLB_INVALID_SOCKET) { + flb_connection_reset_io_timeout(connection); + + result = mk_event_add(connection->evl, + connection->fd, + FLB_ENGINE_EV_THREAD, + connection->event_registration_mask, + &connection->event); + } + + connection->busy_flag = FLB_FALSE; + + if (result != 0 || + connection->event_release_pending == FLB_TRUE || + connection->fd == FLB_INVALID_SOCKET) { + connection->event_release_pending = FLB_FALSE; + prepare_destroy_conn_safe(connection); + connection->coroutine = NULL; + flb_downstream_conn_event_coro_terminate(coro); + } + + /* + * Async connection callbacks own their read wait. Run them continuously + * so TLS application data buffered during a handshake is consumed without + * requiring another kernel readiness edge. + */ + flb_downstream_conn_async_event_loop(connection, coro); +} + /* Config map for Downstream networking setup */ struct flb_config_map downstream_net[] = { { @@ -584,6 +673,138 @@ int flb_downstream_conn_release(struct flb_connection *connection) return ret; } +int flb_downstream_conn_release_all(struct flb_downstream *stream) +{ + struct flb_connection *connection; + struct mk_list *head; + struct mk_list *tmp; + + if (stream == NULL) { + return -1; + } + + flb_stream_acquire_lock(&stream->base, FLB_TRUE); + + mk_list_foreach_safe(head, tmp, &stream->busy_queue) { + connection = mk_list_entry(head, struct flb_connection, _head); + + if (connection->event_coroutine != NULL && + connection->busy_flag == FLB_TRUE) { + connection->event_release_pending = FLB_TRUE; + + if (flb_coro_get() != connection->event_coroutine) { + wake_event_coroutine(connection, ECANCELED); + } + } + else { + prepare_destroy_conn(connection); + } + } + + flb_stream_release_lock(&stream->base); + + if (flb_stream_is_thread_safe(&stream->base)) { + resume_pending_event_coroutines(stream); + } + + return 0; +} + +int flb_downstream_conn_event_accept( + struct flb_downstream *stream, + flb_connection_accept_callback accept_callback, + void *accept_callback_data, + flb_connection_event_callback event_callback, + int mask) +{ + int ret; + size_t stack_size; + flb_sockfd_t connection_fd; + struct flb_coro *coro; + struct flb_coro *previous_coro; + struct flb_connection *connection; + struct flb_config *config; + + if (stream == NULL || accept_callback == NULL || event_callback == NULL || + (mask & (MK_EVENT_READ | MK_EVENT_WRITE)) == 0 || + (stream->base.transport != FLB_TRANSPORT_TCP && + stream->base.transport != FLB_TRANSPORT_UNIX_STREAM)) { + return -1; + } + + if (stream->paused == FLB_TRUE) { + connection_fd = flb_net_accept(stream->server_fd); + if (connection_fd >= 0) { + flb_socket_close(connection_fd); + + return 0; + } + + return -1; + } + + config = stream->base.config; + if (config == NULL || flb_downstream_is_async(stream) == FLB_FALSE) { + return -1; + } + + connection = flb_connection_create(FLB_INVALID_SOCKET, + FLB_DOWNSTREAM_CONNECTION, + stream, + flb_engine_evl_get(), + NULL); + if (connection == NULL) { + return -1; + } + + coro = flb_coro_create(connection); + if (coro == NULL) { + flb_connection_destroy(connection); + return -1; + } + + coro->caller = co_active(); + coro->callee = co_create(config->coro_stack_size, + flb_downstream_conn_accept_event_coro, + &stack_size); + if (coro->callee == NULL) { + flb_coro_destroy(coro); + flb_connection_destroy(connection); + return -1; + } + +#ifdef FLB_HAVE_VALGRIND + coro->valgrind_stack_id = VALGRIND_STACK_REGISTER( + coro->callee, + ((char *) coro->callee) + stack_size); +#endif + + connection->accept_callback = accept_callback; + connection->accept_callback_data = accept_callback_data; + connection->event_callback = event_callback; + connection->event_registration_mask = mask; + connection->event_coroutine = coro; + connection->coroutine = coro; + connection->busy_flag = FLB_TRUE; + flb_connection_enable_flags(connection, FLB_IO_ASYNC); + + flb_stream_acquire_lock(&stream->base, FLB_TRUE); + mk_list_add(&connection->_head, &stream->busy_queue); + flb_stream_release_lock(&stream->base); + + previous_coro = flb_coro_get(); + flb_coro_resume(coro); + flb_coro_set(previous_coro); + + ret = 0; + if (connection->fd == FLB_INVALID_SOCKET && + connection->coroutine == NULL) { + ret = -1; + } + + return ret; +} + int flb_downstream_conn_event_register(struct flb_connection *connection, flb_connection_event_callback callback, int mask) From 6a5e2ebd029c646cc5b47c526910c6b85d93bee7 Mon Sep 17 00:00:00 2001 From: Eduardo Silva Date: Mon, 3 Aug 2026 13:24:23 -0600 Subject: [PATCH 03/21] input: support fallible pause callbacks Signed-off-by: Eduardo Silva --- include/fluent-bit/flb_input.h | 4 +++ src/flb_input.c | 52 ++++++++++++++++++++++++++++++---- 2 files changed, 50 insertions(+), 6 deletions(-) diff --git a/include/fluent-bit/flb_input.h b/include/fluent-bit/flb_input.h index e524fd7e337..080fccb0106 100644 --- a/include/fluent-bit/flb_input.h +++ b/include/fluent-bit/flb_input.h @@ -203,6 +203,8 @@ struct flb_input_plugin { */ void (*cb_pause) (void *, struct flb_config *); void (*cb_resume) (void *, struct flb_config *); + int (*cb_pause_checked) (void *, struct flb_config *); + int (*cb_resume_checked) (void *, struct flb_config *); /* * Optional callback that can be used from a parent caller to ingest @@ -903,6 +905,8 @@ void *flb_input_flush(struct flb_input_instance *ins, size_t *size); int flb_input_test_pause_resume(struct flb_input_instance *ins, int sleep_seconds); int flb_input_pause(struct flb_input_instance *ins); int flb_input_pause_all(struct flb_config *config); +int flb_input_plugin_pause(struct flb_input_instance *ins); +int flb_input_plugin_resume(struct flb_input_instance *ins); int flb_input_resume(struct flb_input_instance *ins); #ifdef FLB_HAVE_METRICS void flb_input_rate_update(struct flb_input_instance *ins, diff --git a/src/flb_input.c b/src/flb_input.c index da78b511f40..3b74d7d094a 100644 --- a/src/flb_input.c +++ b/src/flb_input.c @@ -2984,6 +2984,46 @@ static void flb_input_ingestion_resumed(struct flb_input_instance *ins) } } +int flb_input_plugin_pause(struct flb_input_instance *ins) +{ + int ret; + + ret = 0; + + if (ins->p->cb_pause_checked != NULL) { + ret = ins->p->cb_pause_checked(ins->context, ins->config); + } + else if (ins->p->cb_pause != NULL) { + ins->p->cb_pause(ins->context, ins->config); + } + + if (ret == 0) { + flb_input_ingestion_paused(ins); + } + + return ret; +} + +int flb_input_plugin_resume(struct flb_input_instance *ins) +{ + int ret; + + ret = 0; + + if (ins->p->cb_resume_checked != NULL) { + ret = ins->p->cb_resume_checked(ins->context, ins->config); + } + else if (ins->p->cb_resume != NULL) { + ins->p->cb_resume(ins->context, ins->config); + } + + if (ret == 0) { + flb_input_ingestion_resumed(ins); + } + + return ret; +} + int flb_input_pause(struct flb_input_instance *ins) { /* if the instance is already paused, just return */ @@ -2992,14 +3032,14 @@ int flb_input_pause(struct flb_input_instance *ins) } /* Pause only if a callback is set and a local context exists */ - if (ins->p->cb_pause && ins->context) { + if ((ins->p->cb_pause || ins->p->cb_pause_checked) && ins->context) { if (flb_input_is_threaded(ins)) { /* signal the thread event loop about the 'pause' operation */ - flb_input_thread_instance_pause(ins); + return flb_input_thread_instance_pause(ins); } else { flb_info("[input] pausing %s", flb_input_name(ins)); - ins->p->cb_pause(ins->context, ins->config); + return flb_input_plugin_pause(ins); } } @@ -3010,14 +3050,14 @@ int flb_input_pause(struct flb_input_instance *ins) int flb_input_resume(struct flb_input_instance *ins) { - if (ins->p->cb_resume && ins->context) { + if ((ins->p->cb_resume || ins->p->cb_resume_checked) && ins->context) { if (flb_input_is_threaded(ins)) { /* signal the thread event loop about the 'resume' operation */ - flb_input_thread_instance_resume(ins); + return flb_input_thread_instance_resume(ins); } else { flb_info("[input] resume %s", flb_input_name(ins)); - ins->p->cb_resume(ins->context, ins->config); + return flb_input_plugin_resume(ins); } } From 08c70ae5b712d3a82d0ef05a69a8bb5409cb1dc0 Mon Sep 17 00:00:00 2001 From: Eduardo Silva Date: Mon, 3 Aug 2026 13:24:24 -0600 Subject: [PATCH 04/21] input_thread: handle fallible pause callbacks Signed-off-by: Eduardo Silva --- src/flb_input_thread.c | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/src/flb_input_thread.c b/src/flb_input_thread.c index e9f2650b9c4..e8b2865a7f8 100644 --- a/src/flb_input_thread.c +++ b/src/flb_input_thread.c @@ -75,13 +75,17 @@ static inline int handle_input_event(flb_pipefd_t fd, struct flb_input_instance } else if (type == FLB_INPUT_THREAD_TO_THREAD) { if (operation == FLB_INPUT_THREAD_PAUSE) { - if (ins->p->cb_pause && ins->context) { - ins->p->cb_pause(ins->context, ins->config); + if ((ins->p->cb_pause || ins->p->cb_pause_checked) && ins->context) { + if (flb_input_plugin_pause(ins) != 0) { + flb_plg_error(ins, "could not pause input instance"); + } } } else if (operation == FLB_INPUT_THREAD_RESUME) { - if (ins->p->cb_resume) { - ins->p->cb_resume(ins->context, ins->config); + if ((ins->p->cb_resume || ins->p->cb_resume_checked) && ins->context) { + if (flb_input_plugin_resume(ins) != 0) { + flb_plg_error(ins, "could not resume input instance"); + } } } else if (operation == FLB_INPUT_THREAD_EXIT) { From 33d111e323ac4ef7d166410a153df8576e1992cc Mon Sep 17 00:00:00 2001 From: Eduardo Silva Date: Mon, 3 Aug 2026 13:23:01 -0600 Subject: [PATCH 05/21] http_server: add pause and resume controls Signed-off-by: Eduardo Silva --- .../fluent-bit/http_server/flb_http_server.h | 5 + src/http_server/flb_http_server.c | 206 +++++++++++++++--- 2 files changed, 184 insertions(+), 27 deletions(-) diff --git a/include/fluent-bit/http_server/flb_http_server.h b/include/fluent-bit/http_server/flb_http_server.h index 5e75263b6fe..3927a0539cd 100755 --- a/include/fluent-bit/http_server/flb_http_server.h +++ b/include/fluent-bit/http_server/flb_http_server.h @@ -169,6 +169,7 @@ struct flb_http_server_session { int releasable; int drop_pending; + int destroying; int connection_slot_reserved; struct flb_connection *connection; @@ -219,6 +220,10 @@ int flb_http_server_init_with_options(struct flb_http_server *session, int flb_http_server_start(struct flb_http_server *session); +int flb_http_server_pause(struct flb_http_server *session); + +int flb_http_server_resume(struct flb_http_server *session); + int flb_http_server_stop(struct flb_http_server *session); int flb_http_server_destroy(struct flb_http_server *session); diff --git a/src/http_server/flb_http_server.c b/src/http_server/flb_http_server.c index 35deef0d937..e3befbb2c61 100644 --- a/src/http_server/flb_http_server.c +++ b/src/http_server/flb_http_server.c @@ -34,12 +34,18 @@ /* PRIVATE */ +#define FLB_HTTP_SERVER_ACCEPT_BATCH_SIZE 64 + struct flb_http_server_worker_context { struct flb_http_server server; struct flb_net_setup net_setup; }; static void flb_http_server_runtime_stop(struct flb_http_server *session); + +static void flb_http_server_pause_on_event_loop(struct flb_http_server *server); + +static void flb_http_server_resume_on_event_loop(struct flb_http_server *server); static int flb_http_server_running_on_caller_context( struct flb_http_server *session) { @@ -317,6 +323,10 @@ static int flb_http_server_should_connection_be_closed( server = parent_session->parent; downstream = server->downstream; + if (downstream->paused == FLB_TRUE) { + return FLB_TRUE; + } + /* Version behaviors implemented in the following block : * HTTP/0.9 keep-alive is opt-in * HTTP/1.0 keep-alive is opt-in @@ -399,6 +409,14 @@ static int flb_http_server_client_activity_event_handler(void *data) server = session->parent; + if (event->mask & MK_EVENT_READ && + server->downstream != NULL && + server->downstream->paused == FLB_TRUE) { + flb_http_server_session_destroy(session); + + return -1; + } + if (event->mask & MK_EVENT_READ) { result = flb_http_server_session_read(session); @@ -462,28 +480,21 @@ static int flb_http_server_client_activity_event_handler(void *data) return 0; } -static int flb_http_server_client_connection_event_handler(void *data) +static int flb_http_server_client_connection_initialize( + struct flb_connection *connection, + void *data) { - struct flb_connection *connection; struct flb_http_server_session *session; struct flb_http_server *server; int result; server = (struct flb_http_server *) data; - connection = flb_downstream_conn_get(server->downstream); - - if (connection == NULL) { - return -1; - } - if (server->max_connections > 0) { flb_http_server_reap_stale_sessions(server); } if (!flb_http_server_connection_slot_reserve(server)) { - flb_downstream_conn_release(connection); - return -5; } @@ -491,7 +502,6 @@ static int flb_http_server_client_connection_event_handler(void *data) if (session == NULL) { flb_http_server_connection_slot_release(server); - flb_downstream_conn_release(connection); return -2; } @@ -504,24 +514,8 @@ static int flb_http_server_client_connection_event_handler(void *data) session->http1.stream.user_data = server->user_data; } - MK_EVENT_NEW(&connection->event); - connection->user_data = (void *) session; connection->drop_notification_callback = flb_http_server_connection_drop; - connection->event.type = FLB_ENGINE_EV_CUSTOM; - connection->event.handler = flb_http_server_client_activity_event_handler; - - result = mk_event_add(server->event_loop, - connection->fd, - FLB_ENGINE_EV_CUSTOM, - MK_EVENT_READ, - &connection->event); - - if (result == -1) { - flb_http_server_session_destroy(session); - - return -3; - } cfl_list_add(&session->_head, &server->clients); @@ -536,6 +530,62 @@ static int flb_http_server_client_connection_event_handler(void *data) return 0; } +static int flb_http_server_client_connection_event_handler(void *data) +{ + int accepted_connections; + int result; + struct flb_connection *connection; + struct flb_http_server *server; + + server = (struct flb_http_server *) data; + + if (flb_downstream_is_async(server->downstream) == FLB_FALSE) { + connection = flb_downstream_conn_get(server->downstream); + if (connection == NULL) { + return -1; + } + + result = flb_http_server_client_connection_initialize(connection, server); + if (result != 0) { + if (connection->fd != FLB_INVALID_SOCKET) { + flb_downstream_conn_release(connection); + } + + return result; + } + + result = flb_downstream_conn_event_register( + connection, + flb_http_server_client_activity_event_handler, + MK_EVENT_READ); + if (result != 0) { + flb_http_server_session_destroy(connection->user_data); + } + + return result; + } + + accepted_connections = 0; + + do { + result = flb_downstream_conn_event_accept( + server->downstream, + flb_http_server_client_connection_initialize, + server, + flb_http_server_client_activity_event_handler, + MK_EVENT_READ); + accepted_connections++; + } while (result == 0 && + accepted_connections < FLB_HTTP_SERVER_ACCEPT_BATCH_SIZE); + + /* + * EAGAIN is the normal termination condition. The batch limit prevents a + * continuously refilled accept queue from starving control and timeout + * events; a still-readable listener will be dispatched again. + */ + return 0; +} + static void flb_http_server_worker_maintenance(struct flb_downstream_worker *worker, void *worker_context) { @@ -923,6 +973,8 @@ int flb_http_server_start(struct flb_http_server *session) return -1; } + flb_stream_enable_async_mode(&session->downstream->base); + session->listener_event.type = FLB_ENGINE_EV_CUSTOM; session->listener_event.handler = flb_http_server_client_connection_event_handler; @@ -955,6 +1007,101 @@ int flb_http_server_start(struct flb_http_server *session) return 0; } +static void flb_http_server_pause_on_event_loop(struct flb_http_server *server) +{ + struct cfl_list *iterator_backup; + struct cfl_list *iterator; + struct flb_http_server_session *session; + + if (server->downstream != NULL) { + flb_downstream_pause(server->downstream); + } + + /* + * Active downstream callbacks may be suspended in asynchronous I/O. + * Releasing through the downstream interface wakes those coroutines and + * defers the drop notification until their callbacks have unwound. + */ + cfl_list_foreach_safe(iterator, + iterator_backup, + &server->clients) { + session = cfl_list_entry(iterator, + struct flb_http_server_session, + _head); + + if (session->connection != NULL) { + session->drop_pending = FLB_TRUE; + } + } + + if (server->downstream != NULL) { + flb_downstream_conn_release_all(server->downstream); + } +} + +static void flb_http_server_resume_on_event_loop(struct flb_http_server *server) +{ + flb_http_server_reap_stale_sessions(server); + + if (server->downstream != NULL) { + flb_downstream_resume(server->downstream); + } +} + +static void flb_http_server_worker_pause(struct flb_downstream_worker *worker, + void *worker_context, + void *data) +{ + struct flb_http_server_worker_context *context; + + (void) worker; + (void) data; + + context = worker_context; + + flb_http_server_pause_on_event_loop(&context->server); +} + +static void flb_http_server_worker_resume(struct flb_downstream_worker *worker, + void *worker_context, + void *data) +{ + struct flb_http_server_worker_context *context; + + (void) worker; + (void) data; + + context = worker_context; + + flb_http_server_resume_on_event_loop(&context->server); +} + +int flb_http_server_pause(struct flb_http_server *server) +{ + if (server->runtime != NULL) { + return flb_downstream_worker_runtime_foreach(server->runtime, + flb_http_server_worker_pause, + NULL); + } + + flb_http_server_pause_on_event_loop(server); + + return 0; +} + +int flb_http_server_resume(struct flb_http_server *server) +{ + if (server->runtime != NULL) { + return flb_downstream_worker_runtime_foreach(server->runtime, + flb_http_server_worker_resume, + NULL); + } + + flb_http_server_resume_on_event_loop(server); + + return 0; +} + int flb_http_server_stop(struct flb_http_server *server) { struct cfl_list *iterator_backup; @@ -1137,6 +1284,11 @@ void flb_http_server_session_destroy(struct flb_http_server_session *session) struct flb_connection *connection; if (session != NULL) { + if (session->destroying == FLB_TRUE) { + return; + } + session->destroying = FLB_TRUE; + connection = session->connection; session->connection = NULL; From adb1bef91d0f3fe8797d23843bb0ea8da2badd6f Mon Sep 17 00:00:00 2001 From: Eduardo Silva Date: Mon, 3 Aug 2026 13:23:08 -0600 Subject: [PATCH 06/21] in_http: implement pause and resume callbacks Signed-off-by: Eduardo Silva --- plugins/in_http/http.c | 36 ++++++++++++++++++++++++++++++++++-- 1 file changed, 34 insertions(+), 2 deletions(-) diff --git a/plugins/in_http/http.c b/plugins/in_http/http.c index c8e4d3cae82..3010c1e021f 100644 --- a/plugins/in_http/http.c +++ b/plugins/in_http/http.c @@ -138,6 +138,38 @@ static int in_http_exit(void *data, struct flb_config *config) return 0; } +static int in_http_pause(void *data, struct flb_config *config) +{ + struct flb_http *ctx; + + (void) config; + + ctx = data; + + if (flb_http_server_pause(&ctx->http_server) != 0) { + flb_plg_error(ctx->ins, "could not pause HTTP server"); + return -1; + } + + return 0; +} + +static int in_http_resume(void *data, struct flb_config *config) +{ + struct flb_http *ctx; + + (void) config; + + ctx = data; + + if (flb_http_server_resume(&ctx->http_server) != 0) { + flb_plg_error(ctx->ins, "could not resume HTTP server"); + return -1; + } + + return 0; +} + /* Configuration properties map */ static struct flb_config_map config_map[] = { { @@ -188,8 +220,8 @@ struct flb_input_plugin in_http_plugin = { .cb_pre_run = NULL, .cb_collect = NULL, .cb_flush_buf = NULL, - .cb_pause = NULL, - .cb_resume = NULL, + .cb_pause_checked = in_http_pause, + .cb_resume_checked = in_http_resume, .cb_exit = in_http_exit, .config_map = config_map, .flags = FLB_INPUT_NET_SERVER | FLB_INPUT_HTTP_SERVER | FLB_IO_OPT_TLS From da4d1582dc15756abb1a9253e12152b1a3e11217 Mon Sep 17 00:00:00 2001 From: Eduardo Silva Date: Mon, 3 Aug 2026 13:23:08 -0600 Subject: [PATCH 07/21] in_opentelemetry: implement pause and resume callbacks Signed-off-by: Eduardo Silva --- plugins/in_opentelemetry/opentelemetry.c | 36 ++++++++++++++++++++++-- 1 file changed, 34 insertions(+), 2 deletions(-) diff --git a/plugins/in_opentelemetry/opentelemetry.c b/plugins/in_opentelemetry/opentelemetry.c index d4229941d4f..6ff23d5d66f 100644 --- a/plugins/in_opentelemetry/opentelemetry.c +++ b/plugins/in_opentelemetry/opentelemetry.c @@ -137,6 +137,38 @@ static int in_opentelemetry_exit(void *data, struct flb_config *config) return 0; } +static int in_opentelemetry_pause(void *data, struct flb_config *config) +{ + struct flb_opentelemetry *ctx; + + (void) config; + + ctx = data; + + if (flb_http_server_pause(&ctx->http_server) != 0) { + flb_plg_error(ctx->ins, "could not pause HTTP server"); + return -1; + } + + return 0; +} + +static int in_opentelemetry_resume(void *data, struct flb_config *config) +{ + struct flb_opentelemetry *ctx; + + (void) config; + + ctx = data; + + if (flb_http_server_resume(&ctx->http_server) != 0) { + flb_plg_error(ctx->ins, "could not resume HTTP server"); + return -1; + } + + return 0; +} + /* Configuration properties map */ static struct flb_config_map config_map[] = { { @@ -197,8 +229,8 @@ struct flb_input_plugin in_opentelemetry_plugin = { .cb_pre_run = NULL, .cb_collect = NULL, .cb_flush_buf = NULL, - .cb_pause = NULL, - .cb_resume = NULL, + .cb_pause_checked = in_opentelemetry_pause, + .cb_resume_checked = in_opentelemetry_resume, .cb_exit = in_opentelemetry_exit, .config_map = config_map, .flags = FLB_INPUT_NET_SERVER | FLB_INPUT_HTTP_SERVER | FLB_IO_OPT_TLS From 3b8083527dbcf47e0400907570b33b6165cff84e Mon Sep 17 00:00:00 2001 From: Eduardo Silva Date: Mon, 3 Aug 2026 13:23:09 -0600 Subject: [PATCH 08/21] in_elasticsearch: implement pause and resume callbacks Signed-off-by: Eduardo Silva --- plugins/in_elasticsearch/in_elasticsearch.c | 34 +++++++++++++++++++-- 1 file changed, 32 insertions(+), 2 deletions(-) diff --git a/plugins/in_elasticsearch/in_elasticsearch.c b/plugins/in_elasticsearch/in_elasticsearch.c index a89a5203ff7..0fb016c3a0d 100644 --- a/plugins/in_elasticsearch/in_elasticsearch.c +++ b/plugins/in_elasticsearch/in_elasticsearch.c @@ -153,6 +153,36 @@ static int in_elasticsearch_bulk_exit(void *data, struct flb_config *config) return 0; } +static int in_elasticsearch_bulk_pause(void *data, struct flb_config *config) +{ + struct flb_in_elasticsearch *ctx; + + (void) config; + + ctx = data; + if (flb_http_server_pause(&ctx->http_server) != 0) { + flb_plg_error(ctx->ins, "could not pause HTTP server"); + return -1; + } + + return 0; +} + +static int in_elasticsearch_bulk_resume(void *data, struct flb_config *config) +{ + struct flb_in_elasticsearch *ctx; + + (void) config; + + ctx = data; + if (flb_http_server_resume(&ctx->http_server) != 0) { + flb_plg_error(ctx->ins, "could not resume HTTP server"); + return -1; + } + + return 0; +} + /* Configuration properties map */ static struct flb_config_map config_map[] = { { @@ -191,8 +221,8 @@ struct flb_input_plugin in_elasticsearch_plugin = { .cb_pre_run = NULL, .cb_collect = NULL, .cb_flush_buf = NULL, - .cb_pause = NULL, - .cb_resume = NULL, + .cb_pause_checked = in_elasticsearch_bulk_pause, + .cb_resume_checked = in_elasticsearch_bulk_resume, .cb_exit = in_elasticsearch_bulk_exit, .config_map = config_map, .flags = FLB_INPUT_NET_SERVER | FLB_INPUT_HTTP_SERVER | FLB_IO_OPT_TLS From 3033477661a4ab7cd87e33254cd10326ceba536f Mon Sep 17 00:00:00 2001 From: Eduardo Silva Date: Mon, 3 Aug 2026 13:23:09 -0600 Subject: [PATCH 09/21] in_prometheus_remote_write: implement pause callbacks Signed-off-by: Eduardo Silva --- plugins/in_prometheus_remote_write/prom_rw.c | 34 ++++++++++++++++++-- 1 file changed, 32 insertions(+), 2 deletions(-) diff --git a/plugins/in_prometheus_remote_write/prom_rw.c b/plugins/in_prometheus_remote_write/prom_rw.c index 528e9a24a7a..f5966b34afd 100644 --- a/plugins/in_prometheus_remote_write/prom_rw.c +++ b/plugins/in_prometheus_remote_write/prom_rw.c @@ -119,6 +119,36 @@ static int prom_rw_exit(void *data, struct flb_config *config) return 0; } +static int prom_rw_pause(void *data, struct flb_config *config) +{ + struct flb_prom_remote_write *ctx; + + (void) config; + + ctx = data; + if (flb_http_server_pause(&ctx->http_server) != 0) { + flb_plg_error(ctx->ins, "could not pause HTTP server"); + return -1; + } + + return 0; +} + +static int prom_rw_resume(void *data, struct flb_config *config) +{ + struct flb_prom_remote_write *ctx; + + (void) config; + + ctx = data; + if (flb_http_server_resume(&ctx->http_server) != 0) { + flb_plg_error(ctx->ins, "could not resume HTTP server"); + return -1; + } + + return 0; +} + /* Configuration properties map */ static struct flb_config_map config_map[] = { { @@ -150,8 +180,8 @@ struct flb_input_plugin in_prometheus_remote_write_plugin = { .cb_pre_run = NULL, .cb_collect = NULL, .cb_flush_buf = NULL, - .cb_pause = NULL, - .cb_resume = NULL, + .cb_pause_checked = prom_rw_pause, + .cb_resume_checked = prom_rw_resume, .cb_exit = prom_rw_exit, .config_map = config_map, .flags = FLB_INPUT_NET_SERVER | FLB_INPUT_HTTP_SERVER | FLB_IO_OPT_TLS From 835844bfc8fd051efaf7ed8b7f880b11a43d46e7 Mon Sep 17 00:00:00 2001 From: Eduardo Silva Date: Mon, 3 Aug 2026 13:23:09 -0600 Subject: [PATCH 10/21] in_splunk: implement pause and resume callbacks Signed-off-by: Eduardo Silva --- plugins/in_splunk/splunk.c | 34 ++++++++++++++++++++++++++++++++-- 1 file changed, 32 insertions(+), 2 deletions(-) diff --git a/plugins/in_splunk/splunk.c b/plugins/in_splunk/splunk.c index 1027fcfa553..00fa975ae55 100644 --- a/plugins/in_splunk/splunk.c +++ b/plugins/in_splunk/splunk.c @@ -108,6 +108,36 @@ static int in_splunk_exit(void *data, struct flb_config *config) return 0; } +static int in_splunk_pause(void *data, struct flb_config *config) +{ + struct flb_splunk *ctx; + + (void) config; + + ctx = data; + if (flb_http_server_pause(&ctx->http_server) != 0) { + flb_plg_error(ctx->ins, "could not pause HTTP server"); + return -1; + } + + return 0; +} + +static int in_splunk_resume(void *data, struct flb_config *config) +{ + struct flb_splunk *ctx; + + (void) config; + + ctx = data; + if (flb_http_server_resume(&ctx->http_server) != 0) { + flb_plg_error(ctx->ins, "could not resume HTTP server"); + return -1; + } + + return 0; +} + /* Configuration properties map */ static struct flb_config_map config_map[] = { { @@ -163,8 +193,8 @@ struct flb_input_plugin in_splunk_plugin = { .cb_pre_run = NULL, .cb_collect = NULL, .cb_flush_buf = NULL, - .cb_pause = NULL, - .cb_resume = NULL, + .cb_pause_checked = in_splunk_pause, + .cb_resume_checked = in_splunk_resume, .cb_exit = in_splunk_exit, .config_map = config_map, .flags = FLB_INPUT_NET_SERVER | FLB_INPUT_HTTP_SERVER | FLB_IO_OPT_TLS From 1e93a09aa658da604a2e85b8a9ffe400b2c787f3 Mon Sep 17 00:00:00 2001 From: Eduardo Silva Date: Mon, 3 Aug 2026 13:23:21 -0600 Subject: [PATCH 11/21] tests: internal: preserve nonblocking accept errors Signed-off-by: Eduardo Silva --- tests/internal/network.c | 21 +++++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/tests/internal/network.c b/tests/internal/network.c index b2941c81759..427f884dad2 100644 --- a/tests/internal/network.c +++ b/tests/internal/network.c @@ -181,9 +181,30 @@ void test_ipv6_bracketed_listen() } } +void test_accept_empty_nonblocking_listener() +{ + flb_sockfd_t fd_remote; + flb_sockfd_t fd_server; + + fd_server = flb_net_server("0", TEST_HOSTv4, + FLB_NETWORK_DEFAULT_BACKLOG_SIZE, + FLB_FALSE); + if (!TEST_CHECK(fd_server != FLB_INVALID_SOCKET)) { + return; + } + + fd_remote = flb_net_accept(fd_server); + + TEST_CHECK(fd_remote == FLB_INVALID_SOCKET); + TEST_CHECK(FLB_WOULDBLOCK()); + + flb_socket_close(fd_server); +} + TEST_LIST = { { "ipv4_client_server", test_ipv4_client_server}, { "ipv6_client_server", test_ipv6_client_server}, { "ipv6_bracketed_listen", test_ipv6_bracketed_listen}, + { "accept_empty_nonblocking_listener", test_accept_empty_nonblocking_listener}, { 0 } }; From b8c4b9a362708e5da23c06ef500c85da8d92706f Mon Sep 17 00:00:00 2001 From: Eduardo Silva Date: Mon, 3 Aug 2026 13:23:22 -0600 Subject: [PATCH 12/21] tests: internal: cover fallible pause callbacks Signed-off-by: Eduardo Silva --- tests/internal/CMakeLists.txt | 1 + tests/internal/input_pause.c | 77 +++++++++++++++++++++++++++++++++++ 2 files changed, 78 insertions(+) create mode 100644 tests/internal/input_pause.c diff --git a/tests/internal/CMakeLists.txt b/tests/internal/CMakeLists.txt index 7ef7ed35dd6..746d5039342 100644 --- a/tests/internal/CMakeLists.txt +++ b/tests/internal/CMakeLists.txt @@ -29,6 +29,7 @@ set(UNIT_TESTS_FILES mp_chunk_cobj.c input_chunk.c input_chunk_routes.c + input_pause.c flb_time.c file.c csv.c diff --git a/tests/internal/input_pause.c b/tests/internal/input_pause.c new file mode 100644 index 00000000000..79f8a397882 --- /dev/null +++ b/tests/internal/input_pause.c @@ -0,0 +1,77 @@ +/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */ + +#include + +#include + +#include "flb_tests_internal.h" + +struct input_pause_test_context { + int pause_calls; + int resume_calls; + int result; +}; + +static int checked_pause(void *data, struct flb_config *config) +{ + struct input_pause_test_context *context; + + (void) config; + + context = data; + context->pause_calls++; + + return context->result; +} + +static int checked_resume(void *data, struct flb_config *config) +{ + struct input_pause_test_context *context; + + (void) config; + + context = data; + context->resume_calls++; + + return context->result; +} + +static void test_checked_pause_resume_result() +{ + int ret; + struct flb_input_plugin plugin; + struct flb_input_instance instance; + struct input_pause_test_context context; + + memset(&plugin, 0, sizeof(plugin)); + memset(&instance, 0, sizeof(instance)); + memset(&context, 0, sizeof(context)); + + plugin.cb_pause_checked = checked_pause; + plugin.cb_resume_checked = checked_resume; + instance.p = &plugin; + instance.context = &context; + + context.result = -1; + ret = flb_input_plugin_pause(&instance); + TEST_CHECK(ret == -1); + TEST_CHECK(context.pause_calls == 1); + + ret = flb_input_plugin_resume(&instance); + TEST_CHECK(ret == -1); + TEST_CHECK(context.resume_calls == 1); + + context.result = 0; + ret = flb_input_plugin_pause(&instance); + TEST_CHECK(ret == 0); + TEST_CHECK(context.pause_calls == 2); + + ret = flb_input_plugin_resume(&instance); + TEST_CHECK(ret == 0); + TEST_CHECK(context.resume_calls == 2); +} + +TEST_LIST = { + { "checked_pause_resume_result", test_checked_pause_resume_result }, + { 0 } +}; From aa885e0872de51e2075431742f94f29d44d64815 Mon Sep 17 00:00:00 2001 From: Eduardo Silva Date: Mon, 3 Aug 2026 13:23:22 -0600 Subject: [PATCH 13/21] tests: internal: cover HTTP server session teardown Signed-off-by: Eduardo Silva --- tests/internal/http_server.c | 34 ++++++++++++++++++++++++++++++++++ 1 file changed, 34 insertions(+) diff --git a/tests/internal/http_server.c b/tests/internal/http_server.c index a79251d8af1..49d4a2da600 100644 --- a/tests/internal/http_server.c +++ b/tests/internal/http_server.c @@ -683,6 +683,38 @@ void test_http_server_session_destroy_clears_drop_pending() flb_free(session); } +void test_http_server_session_destroy_is_reentrant_safe() +{ + struct flb_connection connection; + struct flb_http_server_session *session; + + memset(&connection, 0, sizeof(struct flb_connection)); + connection.fd = FLB_INVALID_SOCKET; + + session = flb_http_server_session_create(HTTP_PROTOCOL_VERSION_11); + if (!TEST_CHECK(session != NULL)) { + return; + } + + session->connection = &connection; + session->releasable = FLB_FALSE; + session->destroying = FLB_TRUE; + connection.user_data = session; + + flb_http_server_session_destroy(session); + + TEST_CHECK(session->connection == &connection); + TEST_CHECK(connection.user_data == session); + + session->destroying = FLB_FALSE; + flb_http_server_session_destroy(session); + + TEST_CHECK(connection.user_data == NULL); + TEST_CHECK(session->connection == NULL); + + flb_free(session); +} + TEST_LIST = { { "http_server_options_defaults", test_http_server_options_defaults }, { "http_server_options_multi_worker_magic", test_http_server_options_multi_worker_magic }, @@ -703,5 +735,7 @@ TEST_LIST = { test_http_server_session_destroy_with_closed_connection }, { "http_server_session_destroy_clears_drop_pending", test_http_server_session_destroy_clears_drop_pending }, + { "http_server_session_destroy_is_reentrant_safe", + test_http_server_session_destroy_is_reentrant_safe }, { 0 } }; From 3c610d31c449bc38980a4b4b713cc81572364a59 Mon Sep 17 00:00:00 2001 From: Eduardo Silva Date: Mon, 3 Aug 2026 13:23:31 -0600 Subject: [PATCH 14/21] tests: integration: expand input pause resume coverage Signed-off-by: Eduardo Silva --- .../config/in_elasticsearch_pause_resume.yaml | 19 ++ ...in_elasticsearch_pause_resume_workers.yaml | 20 ++ .../tests/test_in_elasticsearch_001.py | 42 +++ .../config/in_http_accept_timeout_tls.yaml | 21 ++ .../in_http/config/in_http_pause_resume.yaml | 20 ++ .../in_http_pause_resume_http2_tls.yaml | 22 ++ .../in_http/tests/test_in_http_001.py | 141 ++++++++ .../config/otlp_pause_resume_workers.yaml | 19 ++ .../tests/test_in_opentelemetry_001.py | 43 +++ .../config/receiver_pause_resume.yaml | 20 ++ .../config/receiver_pause_resume_workers.yaml | 21 ++ .../test_in_prometheus_remote_write_001.py | 108 +++++- .../in_splunk/config/splunk_pause_resume.yaml | 20 ++ .../config/splunk_pause_resume_workers.yaml | 21 ++ .../in_splunk/tests/test_in_splunk_001.py | 42 +++ .../src/utils/input_pause_resume.py | 314 ++++++++++++++++++ 16 files changed, 892 insertions(+), 1 deletion(-) create mode 100644 tests/integration/scenarios/in_elasticsearch/config/in_elasticsearch_pause_resume.yaml create mode 100644 tests/integration/scenarios/in_elasticsearch/config/in_elasticsearch_pause_resume_workers.yaml create mode 100644 tests/integration/scenarios/in_http/config/in_http_accept_timeout_tls.yaml create mode 100644 tests/integration/scenarios/in_http/config/in_http_pause_resume.yaml create mode 100644 tests/integration/scenarios/in_http/config/in_http_pause_resume_http2_tls.yaml create mode 100644 tests/integration/scenarios/in_opentelemetry/config/otlp_pause_resume_workers.yaml create mode 100644 tests/integration/scenarios/in_prometheus_remote_write/config/receiver_pause_resume.yaml create mode 100644 tests/integration/scenarios/in_prometheus_remote_write/config/receiver_pause_resume_workers.yaml create mode 100644 tests/integration/scenarios/in_splunk/config/splunk_pause_resume.yaml create mode 100644 tests/integration/scenarios/in_splunk/config/splunk_pause_resume_workers.yaml create mode 100644 tests/integration/src/utils/input_pause_resume.py diff --git a/tests/integration/scenarios/in_elasticsearch/config/in_elasticsearch_pause_resume.yaml b/tests/integration/scenarios/in_elasticsearch/config/in_elasticsearch_pause_resume.yaml new file mode 100644 index 00000000000..294ec0ac03e --- /dev/null +++ b/tests/integration/scenarios/in_elasticsearch/config/in_elasticsearch_pause_resume.yaml @@ -0,0 +1,19 @@ +service: + flush: 12 + grace: 1 + log_level: info + http_server: on + http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT} + +pipeline: + inputs: + - name: elasticsearch + listen: 0.0.0.0 + port: ${FLUENT_BIT_TEST_LISTENER_PORT} + mem_buf_limit: 8KB + http2: off + tls: off + + outputs: + - name: stdout + match: '*' diff --git a/tests/integration/scenarios/in_elasticsearch/config/in_elasticsearch_pause_resume_workers.yaml b/tests/integration/scenarios/in_elasticsearch/config/in_elasticsearch_pause_resume_workers.yaml new file mode 100644 index 00000000000..b5c6908dcad --- /dev/null +++ b/tests/integration/scenarios/in_elasticsearch/config/in_elasticsearch_pause_resume_workers.yaml @@ -0,0 +1,20 @@ +service: + flush: 12 + grace: 1 + log_level: info + http_server: on + http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT} + +pipeline: + inputs: + - name: elasticsearch + listen: 0.0.0.0 + port: ${FLUENT_BIT_TEST_LISTENER_PORT} + mem_buf_limit: 8KB + http2: off + tls: off + http_server.workers: 4 + + outputs: + - name: stdout + match: '*' diff --git a/tests/integration/scenarios/in_elasticsearch/tests/test_in_elasticsearch_001.py b/tests/integration/scenarios/in_elasticsearch/tests/test_in_elasticsearch_001.py index b8f41819bea..531d06f2610 100644 --- a/tests/integration/scenarios/in_elasticsearch/tests/test_in_elasticsearch_001.py +++ b/tests/integration/scenarios/in_elasticsearch/tests/test_in_elasticsearch_001.py @@ -4,6 +4,7 @@ from server.otlp_server import data_storage from utils.http_matrix import PROTOCOL_CASES, run_curl_request +from utils.input_pause_resume import assert_pause_resume_cycles, open_partial_http_request from utils.test_service import FluentBitTestService logger = logging.getLogger(__name__) @@ -273,6 +274,47 @@ def test_in_elasticsearch_rejects_unknown_bulk_operation(): assert details["status"] == 400 +@pytest.mark.parametrize( + "config_file", + [ + "in_elasticsearch_pause_resume.yaml", + "in_elasticsearch_pause_resume_workers.yaml", + ], + ids=["single_listener", "workers_4"], +) +def test_in_elasticsearch_pause_resume_cycles(config_file): + service = Service(config_file) + + try: + service.start() + large_document = '{"index":{}}\n{"message":"' + ("x" * 6144) + '"}\n' + small_document = '{"index":{}}\n{"message":"resume-check"}\n' + + def open_active_connections(): + return [ + open_partial_http_request( + "127.0.0.1", + service.flb_listener_port, + ) + for _ in range(8) + ] + + assert_pause_resume_cycles( + service.flb, + f"http://localhost:{service.flb_listener_port}/_bulk", + large_document, + ["Content-Type: application/x-ndjson"], + input_name="elasticsearch.0", + success_status=200, + cycles=2, + pause_trigger_requests=2, + resume_payload=small_document, + active_connection_factory=open_active_connections, + ) + finally: + service.stop() + + @pytest.mark.parametrize( "case", IN_ELASTICSEARCH_SMALL_BUFFER_REGRESSION_CASES, diff --git a/tests/integration/scenarios/in_http/config/in_http_accept_timeout_tls.yaml b/tests/integration/scenarios/in_http/config/in_http_accept_timeout_tls.yaml new file mode 100644 index 00000000000..6694b4b324f --- /dev/null +++ b/tests/integration/scenarios/in_http/config/in_http_accept_timeout_tls.yaml @@ -0,0 +1,21 @@ +service: + flush: 1 + grace: 1 + log_level: info + http_server: on + http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT} + +pipeline: + inputs: + - name: http + port: ${FLUENT_BIT_TEST_LISTENER_PORT} + http2: on + workers: 4 + net.accept_timeout: 2s + tls: on + tls.crt_file: ${CERTIFICATE_TEST} + tls.key_file: ${PRIVATE_KEY_TEST} + + outputs: + - name: stdout + match: "*" diff --git a/tests/integration/scenarios/in_http/config/in_http_pause_resume.yaml b/tests/integration/scenarios/in_http/config/in_http_pause_resume.yaml new file mode 100644 index 00000000000..64744150ff8 --- /dev/null +++ b/tests/integration/scenarios/in_http/config/in_http_pause_resume.yaml @@ -0,0 +1,20 @@ +service: + flush: 12 + grace: 1 + log_level: info + http_server: on + http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT} + +pipeline: + inputs: + - name: http + port: ${FLUENT_BIT_TEST_LISTENER_PORT} + mem_buf_limit: 8KB + enable_health_endpoint: on + http2: off + workers: 4 + tls: off + + outputs: + - name: stdout + match: '*' diff --git a/tests/integration/scenarios/in_http/config/in_http_pause_resume_http2_tls.yaml b/tests/integration/scenarios/in_http/config/in_http_pause_resume_http2_tls.yaml new file mode 100644 index 00000000000..ab81500aef7 --- /dev/null +++ b/tests/integration/scenarios/in_http/config/in_http_pause_resume_http2_tls.yaml @@ -0,0 +1,22 @@ +service: + flush: 12 + grace: 1 + log_level: info + http_server: on + http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT} + +pipeline: + inputs: + - name: http + port: ${FLUENT_BIT_TEST_LISTENER_PORT} + mem_buf_limit: 8KB + enable_health_endpoint: on + http2: on + workers: 4 + tls: on + tls.crt_file: ${CERTIFICATE_TEST} + tls.key_file: ${PRIVATE_KEY_TEST} + + outputs: + - name: stdout + match: '*' diff --git a/tests/integration/scenarios/in_http/tests/test_in_http_001.py b/tests/integration/scenarios/in_http/tests/test_in_http_001.py index 519af7ae538..d0ffe6d4564 100644 --- a/tests/integration/scenarios/in_http/tests/test_in_http_001.py +++ b/tests/integration/scenarios/in_http/tests/test_in_http_001.py @@ -9,6 +9,16 @@ from server.http_server import data_storage, http_server_run from utils.http_matrix import PROTOCOL_CASES, run_curl_request +from utils.input_pause_resume import ( + ConnectionFlood, + assert_connection_closed, + assert_pause_resume_cycles, + is_valgrind, + large_json_payload, + open_partial_http_request, + open_stalled_tcp_connection, + wait_for_input_pause_state, +) from utils.test_service import FluentBitTestService logger = logging.getLogger(__name__) @@ -141,6 +151,137 @@ def test_in_http_rejects_get_requests(): assert result["status_code"] >= 400 +@pytest.mark.parametrize( + "case", + [ + { + "id": "http1_cleartext_workers", + "config": "in_http_pause_resume.yaml", + "scheme": "http", + "http_mode": "http1.1", + "stalled_connection": open_partial_http_request, + }, + { + "id": "http2_tls_workers", + "config": "in_http_pause_resume_http2_tls.yaml", + "scheme": "https", + "http_mode": "http2", + "stalled_connection": open_stalled_tcp_connection, + }, + ], + ids=lambda case: case["id"], +) +def test_in_http_pause_resume_cycles(case): + service = Service(case["config"]) + + try: + service.start() + + def open_active_connections(): + return [ + case["stalled_connection"]( + "127.0.0.1", + service.flb_listener_port, + ) + for _ in range(8) + ] + + assert_pause_resume_cycles( + service.flb, + f"{case['scheme']}://localhost:{service.flb_listener_port}/", + large_json_payload(size=6144), + ["Content-Type: application/json"], + input_name="http.0", + success_status=201, + cycles=3, + http_mode=case["http_mode"], + pause_trigger_requests=2, + ca_cert_path=service.tls_crt_file if case["scheme"] == "https" else None, + active_connection_factory=open_active_connections, + ) + finally: + service.stop() + + +def test_in_http_shutdown_while_paused_with_active_connections(): + service = Service("in_http_pause_resume.yaml") + service.start() + connection_flood = ConnectionFlood( + "127.0.0.1", + service.flb_listener_port, + ) + stalled_connections = [] + + try: + for _ in range(8): + stalled_connections.append( + open_partial_http_request( + "127.0.0.1", + service.flb_listener_port, + ) + ) + + connection_flood.start() + connection_flood.wait_for_attempts(256) + + for _ in range(2): + result = run_curl_request( + f"http://localhost:{service.flb_listener_port}/", + large_json_payload(size=6144), + headers=["Content-Type: application/json"], + http_mode="http1.1", + ) + assert result["status_code"] == 201, result + + wait_for_input_pause_state( + service.flb, + "http.0", + True, + timeout=20 if is_valgrind() else 10, + ) + + shutdown_started = time.monotonic() + service.stop() + shutdown_elapsed = time.monotonic() - shutdown_started + + shutdown_limit = 20 if is_valgrind() else 5 + assert shutdown_elapsed < shutdown_limit + + for connection in stalled_connections: + assert_connection_closed(connection) + finally: + connection_flood.stop() + for connection in stalled_connections: + connection.close() + service.stop() + + +def test_in_http_async_tls_accept_timeout(): + service = Service("in_http_accept_timeout_tls.yaml") + service.start() + stalled_connection = None + + try: + stalled_connection = open_stalled_tcp_connection( + "127.0.0.1", + service.flb_listener_port, + ) + assert_connection_closed(stalled_connection, timeout=10) + + result = run_curl_request( + f"https://localhost:{service.flb_listener_port}/", + '{"message":"accept-timeout-recovered"}', + headers=["Content-Type: application/json"], + http_mode="http2", + ca_cert_path=service.tls_crt_file, + ) + assert result["status_code"] == 201, result + finally: + if stalled_connection is not None: + stalled_connection.close() + service.stop() + + @pytest.mark.parametrize("case", PROTOCOL_CASES, ids=[case["id"] for case in PROTOCOL_CASES]) def test_in_http_health_endpoint(case): service = Service(IN_HTTP_PROTOCOL_CONFIGS[case["config_key"]]) diff --git a/tests/integration/scenarios/in_opentelemetry/config/otlp_pause_resume_workers.yaml b/tests/integration/scenarios/in_opentelemetry/config/otlp_pause_resume_workers.yaml new file mode 100644 index 00000000000..b536906cc0d --- /dev/null +++ b/tests/integration/scenarios/in_opentelemetry/config/otlp_pause_resume_workers.yaml @@ -0,0 +1,19 @@ +service: + flush: 12 + grace: 1 + log_level: info + http_server: on + http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT} + +pipeline: + inputs: + - name: opentelemetry + port: ${FLUENT_BIT_TEST_LISTENER_PORT} + mem_buf_limit: 8KB + http2: on + tls: off + http_server.workers: 4 + + outputs: + - name: stdout + match: '*' diff --git a/tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py b/tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py index 3f4b1675e3c..69e35b0533b 100644 --- a/tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py +++ b/tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py @@ -36,6 +36,7 @@ # local imports from utils.data_utils import read_json_file from utils.http_matrix import PROTOCOL_CASES, run_curl_request +from utils.input_pause_resume import assert_pause_resume_cycles, open_stalled_tcp_connection from utils.test_service import FluentBitTestService from server.http_server import http_server_run @@ -1562,6 +1563,48 @@ def send_job(job): assert "/v1/traces" in paths_seen +@pytest.mark.parametrize( + "signal_type,json_input,endpoint", + [ + ("logs", "test_logs_001.in.json", "/v1/logs"), + ("metrics", "test_metrics_001.in.json", "/v1/metrics"), + ("traces", "test_traces_001.in.json", "/v1/traces"), + ], + ids=["logs", "metrics", "traces"], +) +def test_in_opentelemetry_pause_resume_workers(signal_type, json_input, endpoint): + service = Service("otlp_pause_resume_workers.yaml") + + try: + service.start() + resume_payload = service.build_otel_payload(json_input, signal_type) + + def open_active_connections(): + return [ + open_stalled_tcp_connection( + "127.0.0.1", + service.flb_listener_port, + ) + for _ in range(8) + ] + + assert_pause_resume_cycles( + service.flb, + f"http://localhost:{service.flb_listener_port}{endpoint}", + resume_payload, + ["Content-Type: application/x-protobuf"], + input_name="opentelemetry.0", + success_status=201, + cycles=2, + http_mode="http2-prior-knowledge", + pause_trigger_requests=64, + resume_payload=resume_payload, + active_connection_factory=open_active_connections, + ) + finally: + service.stop() + + def test_in_opentelemetry_http_workers_export_ingress_queue_metrics(): service = Service(IN_OPENTELEMETRY_WORKER_PROTOCOL_CONFIGS["http1_cleartext"]) service.start() diff --git a/tests/integration/scenarios/in_prometheus_remote_write/config/receiver_pause_resume.yaml b/tests/integration/scenarios/in_prometheus_remote_write/config/receiver_pause_resume.yaml new file mode 100644 index 00000000000..cc86891cee5 --- /dev/null +++ b/tests/integration/scenarios/in_prometheus_remote_write/config/receiver_pause_resume.yaml @@ -0,0 +1,20 @@ +service: + flush: 4 + grace: 1 + log_level: info + http_server: on + http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT} + +pipeline: + inputs: + - name: prometheus_remote_write + listen: 127.0.0.1 + port: ${PROM_RW_RECEIVER_PORT} + uri: /write + mem_buf_limit: 8KB + http2: off + successful_response_code: 201 + + outputs: + - name: stdout + match: "*" diff --git a/tests/integration/scenarios/in_prometheus_remote_write/config/receiver_pause_resume_workers.yaml b/tests/integration/scenarios/in_prometheus_remote_write/config/receiver_pause_resume_workers.yaml new file mode 100644 index 00000000000..bc228b6f042 --- /dev/null +++ b/tests/integration/scenarios/in_prometheus_remote_write/config/receiver_pause_resume_workers.yaml @@ -0,0 +1,21 @@ +service: + flush: 4 + grace: 1 + log_level: info + http_server: on + http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT} + +pipeline: + inputs: + - name: prometheus_remote_write + listen: 127.0.0.1 + port: ${PROM_RW_RECEIVER_PORT} + uri: /write + mem_buf_limit: 8KB + http2: off + http_server.workers: 4 + successful_response_code: 201 + + outputs: + - name: stdout + match: "*" diff --git a/tests/integration/scenarios/in_prometheus_remote_write/tests/test_in_prometheus_remote_write_001.py b/tests/integration/scenarios/in_prometheus_remote_write/tests/test_in_prometheus_remote_write_001.py index 88bf30f941c..014ed7e29b2 100644 --- a/tests/integration/scenarios/in_prometheus_remote_write/tests/test_in_prometheus_remote_write_001.py +++ b/tests/integration/scenarios/in_prometheus_remote_write/tests/test_in_prometheus_remote_write_001.py @@ -4,6 +4,12 @@ import pytest from utils.fluent_bit_manager import FluentBitManager +from utils.input_pause_resume import ( + assert_connection_closed, + open_partial_http_request, + open_stalled_tcp_connection, + wait_for_input_pause_state, +) from utils.network import find_available_port @@ -72,7 +78,7 @@ def _restore_env(self): os.environ[key] = value self._previous_env.clear() - def start(self): + def start(self, *, start_sender=True): self._set_env("PROM_RW_RECEIVER_PORT", find_available_port()) self._set_env("CERTIFICATE_TEST", self.tls_crt_file) self._set_env("PRIVATE_KEY_TEST", self.tls_key_file) @@ -82,6 +88,10 @@ def start(self): self.receiver_port = int(os.environ["PROM_RW_RECEIVER_PORT"]) self.wait_for_log(self.receiver.log_file, f"listening on 127.0.0.1:{self.receiver_port}") + if start_sender: + self.start_sender() + + def start_sender(self): self.sender = FluentBitManager(self.sender_config) self.sender.start() @@ -103,6 +113,17 @@ def wait_for_log(self, path, pattern, *, timeout=20, interval=0.5): time.sleep(interval) raise TimeoutError(f"Timed out waiting for {pattern} in {path}") + def wait_for_log_count(self, path, pattern, minimum, *, timeout=20, interval=0.5): + deadline = time.time() + timeout + while time.time() < deadline: + count = _read_file(path).count(pattern) + if count >= minimum: + return count + time.sleep(interval) + raise TimeoutError( + f"Timed out waiting for {minimum} occurrences of {pattern} in {path}" + ) + @pytest.mark.parametrize("workers_enabled", [False, True], ids=["single_listener", "workers_4"]) @pytest.mark.parametrize("case", PROM_RW_CASES, ids=[case["id"] for case in PROM_RW_CASES]) @@ -128,3 +149,88 @@ def test_in_prometheus_remote_write_matrix(case, workers_enabled): assert "fluentbit_input_metrics_scrapes_total" in receiver_log finally: service.stop() + + +@pytest.mark.parametrize( + "receiver_config", + [ + "receiver_pause_resume.yaml", + "receiver_pause_resume_workers.yaml", + ], + ids=["single_listener", "workers_4"], +) +def test_in_prometheus_remote_write_pause_resume_and_shutdown(receiver_config): + service = Service( + receiver_config, + "sender_cleartext.yaml", + ) + stalled_connections = [] + paused_connection = None + + try: + service.start(start_sender=False) + for _ in range(8): + stalled_connections.append( + open_partial_http_request( + "127.0.0.1", + service.receiver_port, + ) + ) + + service.start_sender() + wait_for_input_pause_state( + service.receiver, + "prometheus_remote_write.0", + True, + timeout=30, + ) + + for connection in stalled_connections: + assert_connection_closed(connection) + + paused_connection = open_stalled_tcp_connection( + "127.0.0.1", + service.receiver_port, + ) + assert_connection_closed(paused_connection) + + service.sender.stop() + service.sender = None + + wait_for_input_pause_state( + service.receiver, + "prometheus_remote_write.0", + False, + timeout=30, + ) + service.wait_for_log( + service.receiver.log_file, + "fluentbit_input_metrics_scrapes_total", + timeout=30, + interval=0.5, + ) + + delivered_before_resume = _read_file(service.receiver.log_file).count( + "fluentbit_input_metrics_scrapes_total" + ) + service.start_sender() + service.wait_for_log_count( + service.receiver.log_file, + "fluentbit_input_metrics_scrapes_total", + delivered_before_resume + 1, + timeout=30, + interval=0.5, + ) + wait_for_input_pause_state( + service.receiver, + "prometheus_remote_write.0", + True, + timeout=30, + ) + service.stop() + finally: + for connection in stalled_connections: + connection.close() + if paused_connection is not None: + paused_connection.close() + service.stop() diff --git a/tests/integration/scenarios/in_splunk/config/splunk_pause_resume.yaml b/tests/integration/scenarios/in_splunk/config/splunk_pause_resume.yaml new file mode 100644 index 00000000000..ddc5a3599dd --- /dev/null +++ b/tests/integration/scenarios/in_splunk/config/splunk_pause_resume.yaml @@ -0,0 +1,20 @@ +service: + flush: 12 + grace: 1 + log_level: info + http_server: on + http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT} + +pipeline: + inputs: + - name: splunk + port: ${FLUENT_BIT_TEST_LISTENER_PORT} + host: 0.0.0.0 + mem_buf_limit: 8KB + http2: off + net.keepalive: on + tls: off + + outputs: + - name: stdout + match: '*' diff --git a/tests/integration/scenarios/in_splunk/config/splunk_pause_resume_workers.yaml b/tests/integration/scenarios/in_splunk/config/splunk_pause_resume_workers.yaml new file mode 100644 index 00000000000..05b1a63833e --- /dev/null +++ b/tests/integration/scenarios/in_splunk/config/splunk_pause_resume_workers.yaml @@ -0,0 +1,21 @@ +service: + flush: 12 + grace: 1 + log_level: info + http_server: on + http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT} + +pipeline: + inputs: + - name: splunk + port: ${FLUENT_BIT_TEST_LISTENER_PORT} + host: 0.0.0.0 + mem_buf_limit: 8KB + http2: off + net.keepalive: on + tls: off + http_server.workers: 4 + + outputs: + - name: stdout + match: '*' diff --git a/tests/integration/scenarios/in_splunk/tests/test_in_splunk_001.py b/tests/integration/scenarios/in_splunk/tests/test_in_splunk_001.py index 3da7c34f5da..b3644c7b673 100644 --- a/tests/integration/scenarios/in_splunk/tests/test_in_splunk_001.py +++ b/tests/integration/scenarios/in_splunk/tests/test_in_splunk_001.py @@ -6,6 +6,7 @@ import requests from server.http_server import configure_http_response, data_storage, http_server_run +from utils.input_pause_resume import assert_pause_resume_cycles, open_partial_http_request from utils.test_service import FluentBitTestService from utils.http_matrix import PROTOCOL_CASES, run_curl_request @@ -257,6 +258,47 @@ def test_in_splunk_hec_auth_status_codes(case): assert result["body"] == case["body"] +@pytest.mark.parametrize( + "config_file", + [ + "splunk_pause_resume.yaml", + "splunk_pause_resume_workers.yaml", + ], + ids=["single_listener", "workers_4"], +) +def test_in_splunk_pause_resume_cycles(config_file): + service = Service(config_file) + + try: + service.start() + large_event = json.dumps({"event": "x" * 6144}) + small_event = json.dumps({"event": "resume-check"}) + + def open_active_connections(): + return [ + open_partial_http_request( + "127.0.0.1", + service.flb_listener_port, + ) + for _ in range(8) + ] + + assert_pause_resume_cycles( + service.flb, + f"http://localhost:{service.flb_listener_port}/services/collector", + large_event, + create_splunk_headers(), + input_name="splunk.0", + success_status=200, + cycles=2, + pause_trigger_requests=2, + resume_payload=small_event, + active_connection_factory=open_active_connections, + ) + finally: + service.stop() + + def test_in_splunk_to_out_splunk_prefers_configured_output_token(): service = ForwardingService("in_splunk_to_out_splunk.yaml") service.start() diff --git a/tests/integration/src/utils/input_pause_resume.py b/tests/integration/src/utils/input_pause_resume.py new file mode 100644 index 00000000000..2f6cd0e30be --- /dev/null +++ b/tests/integration/src/utils/input_pause_resume.py @@ -0,0 +1,314 @@ +import re +import socket +import subprocess +import threading +import time + +import requests + +from utils.http_matrix import run_curl_request + + +INGESTION_PAUSED_RE = re.compile( + r'^fluentbit_input_ingestion_paused\{name="([^"]+)"\}\s+([0-9.]+)' +) + + +class ConnectionFlood: + def __init__(self, host, port, workers=16): + self.host = host + self.port = port + self.workers = workers + self.attempts = 0 + self.attempts_lock = threading.Lock() + self.stop_event = threading.Event() + self.threads = [] + + def _run(self): + while not self.stop_event.is_set(): + connection = None + try: + connection = socket.create_connection( + (self.host, self.port), + timeout=0.5, + ) + connection.sendall(b"G") + except OSError: + pass + finally: + if connection is not None: + connection.close() + + with self.attempts_lock: + self.attempts += 1 + + def start(self): + for _ in range(self.workers): + thread = threading.Thread(target=self._run, daemon=True) + thread.start() + self.threads.append(thread) + + def wait_for_attempts(self, minimum, timeout=10): + deadline = time.monotonic() + timeout + + while time.monotonic() < deadline: + with self.attempts_lock: + if self.attempts >= minimum: + return + + time.sleep(0.05) + + raise TimeoutError( + f"Connection flood made fewer than {minimum} attempts" + ) + + def stop(self): + self.stop_event.set() + + for thread in self.threads: + thread.join(timeout=2) + + +def large_json_payload(size=65536): + return '{"message":"' + ("x" * size) + '"}' + + +def payload_bytes(payload): + if isinstance(payload, bytes): + return payload + + return payload.encode() + + +def open_partial_http_request(host, port): + connection = socket.create_connection((host, port), timeout=5) + request = ( + "POST / HTTP/1.1\r\n" + f"Host: {host}\r\n" + "Content-Type: application/json\r\n" + "Content-Length: 65536\r\n" + "Connection: keep-alive\r\n" + "\r\n" + '{"message":"partial' + ) + + connection.sendall(request.encode()) + time.sleep(0.25) + + return connection + + +def open_stalled_tcp_connection(host, port): + connection = socket.create_connection((host, port), timeout=5) + time.sleep(0.25) + return connection + + +def assert_connection_closed(connection, *, timeout=5): + deadline = time.time() + timeout + connection.settimeout(0.25) + + while time.time() < deadline: + try: + if connection.recv(1) == b"": + return + except socket.timeout: + continue + except OSError: + return + + raise AssertionError("HTTP connection remained open after the input paused") + + +def wait_for_input_pause_state(flb, input_name, expected, *, timeout=15, interval=0.25): + deadline = time.time() + timeout + expected_value = 1 if expected else 0 + + while time.time() < deadline: + if input_pause_state(flb, input_name) == expected_value: + return + + time.sleep(interval) + + state = "paused" if expected else "resumed" + raise TimeoutError(f"Timed out waiting for input to become {state}") + + +def input_pause_state(flb, input_name): + try: + response = requests.get( + f"http://127.0.0.1:{flb.http_monitoring_port}/api/v2/metrics/prometheus", + timeout=5 if is_valgrind() else 2, + ) + response.raise_for_status() + except requests.exceptions.RequestException: + return None + + for line in response.text.splitlines(): + match = INGESTION_PAUSED_RE.match(line) + if match and match.group(1) == input_name: + return int(float(match.group(2))) + + return None + + +def curl_status_code(result): + match = re.search(rb"__META__(\d{3})", result.stdout) + if match is None: + return 0 + + return int(match.group(1)) + + +def assert_pause_resume_cycles( + flb, + url, + payload, + headers, + *, + input_name, + success_status, + cycles=2, + http_mode="http1.1", + paused_attempts=4, + pause_trigger_requests=1, + resume_payload=None, + resume_requests=4, + ca_cert_path=None, + active_connection_factory=None, +): + if resume_payload is None: + resume_payload = '{"message":"resume-check"}' + + for _ in range(cycles): + active_connections = [] + if active_connection_factory is not None: + active_connections = active_connection_factory() + if not isinstance(active_connections, (list, tuple)): + active_connections = [active_connections] + + for _ in range(pause_trigger_requests): + result = run_curl_without_check( + url, + payload, + headers=headers, + http_mode=http_mode, + max_time=10, + ca_cert_path=ca_cert_path, + ) + paused = input_pause_state(flb, input_name) == 1 + if paused: + break + + if result.returncode != 0 or curl_status_code(result) != success_status: + wait_for_input_pause_state(flb, input_name, True, timeout=2) + break + + wait_for_input_pause_state( + flb, + input_name, + True, + timeout=20 if is_valgrind() else 10, + ) + + for connection in active_connections: + assert_connection_closed( + connection, + timeout=20 if is_valgrind() else 5, + ) + connection.close() + + rejected_attempts = 0 + for _ in range(paused_attempts): + if input_pause_state(flb, input_name) != 1: + break + + paused_result = run_curl_without_check( + url, + payload, + headers, + http_mode=http_mode, + max_time=2, + ca_cert_path=ca_cert_path, + ) + if paused_result.returncode != 0 or b"__META__000" in paused_result.stdout: + rejected_attempts += 1 + else: + assert input_pause_state(flb, input_name) != 1, ( + "HTTP request succeeded while the input remained paused" + ) + + assert rejected_attempts > 0, "No HTTP request was rejected while the input was paused" + + wait_for_input_pause_state( + flb, + input_name, + False, + timeout=30 if is_valgrind() else 15, + ) + + for _ in range(resume_requests): + result = run_curl_request( + url, + resume_payload, + headers=headers, + http_mode=http_mode, + ca_cert_path=ca_cert_path, + ) + assert result["status_code"] == success_status, result + + +def run_curl_without_check( + url, + payload, + headers, + *, + http_mode, + max_time, + ca_cert_path=None, +): + command = [ + "curl", + "--silent", + "--show-error", + "--output", + "-", + "--write-out", + "\n__META__%{http_code} %{http_version}", + "--max-time", + str(max_time), + "-X", + "POST", + ] + + for header in headers: + command.extend(["-H", header]) + + command.extend(["--data-binary", "@-"]) + + if http_mode == "http1.1": + command.append("--http1.1") + elif http_mode == "http2": + command.append("--http2") + elif http_mode == "http2-prior-knowledge": + command.append("--http2-prior-knowledge") + else: + raise ValueError(f"Unsupported HTTP mode {http_mode}") + + if ca_cert_path is not None: + command.extend(["--cacert", ca_cert_path]) + + command.append(url) + + return subprocess.run( + command, + input=payload_bytes(payload), + capture_output=True, + check=False, + ) + + +def is_valgrind(): + import os + + return bool(os.environ.get("VALGRIND")) From 43b1f44857b465484bc2506d67078b66fc11a3d8 Mon Sep 17 00:00:00 2001 From: Eduardo Silva Date: Mon, 3 Aug 2026 13:23:31 -0600 Subject: [PATCH 15/21] workflows: add input HTTP lifecycle tests Signed-off-by: Eduardo Silva --- .github/workflows/input-http-pause-tests.yaml | 113 ++++++++++++++++++ 1 file changed, 113 insertions(+) create mode 100644 .github/workflows/input-http-pause-tests.yaml diff --git a/.github/workflows/input-http-pause-tests.yaml b/.github/workflows/input-http-pause-tests.yaml new file mode 100644 index 00000000000..afb852f3e75 --- /dev/null +++ b/.github/workflows/input-http-pause-tests.yaml @@ -0,0 +1,113 @@ +name: Input HTTP pause and lifecycle tests + +on: + pull_request: + branches: + - master + paths: + - '.github/workflows/input-http-pause-tests.yaml' + - 'include/fluent-bit/flb_connection.h' + - 'include/fluent-bit/flb_downstream.h' + - 'include/fluent-bit/http_server/**' + - 'plugins/in_elasticsearch/**' + - 'plugins/in_http/**' + - 'plugins/in_opentelemetry/**' + - 'plugins/in_prometheus_remote_write/**' + - 'plugins/in_splunk/**' + - 'src/flb_downstream.c' + - 'src/flb_network.c' + - 'src/http_server/**' + - 'tests/integration/scenarios/in_elasticsearch/**' + - 'tests/integration/scenarios/in_http/**' + - 'tests/integration/scenarios/in_http_max_connections/**' + - 'tests/integration/scenarios/in_opentelemetry/**' + - 'tests/integration/scenarios/in_prometheus_remote_write/**' + - 'tests/integration/scenarios/in_splunk/**' + - 'tests/integration/src/**' + - 'tests/internal/http_server.c' + push: + branches: + - master + paths: + - 'include/fluent-bit/flb_connection.h' + - 'include/fluent-bit/flb_downstream.h' + - 'include/fluent-bit/http_server/**' + - 'plugins/in_elasticsearch/**' + - 'plugins/in_http/**' + - 'plugins/in_opentelemetry/**' + - 'plugins/in_prometheus_remote_write/**' + - 'plugins/in_splunk/**' + - 'src/flb_downstream.c' + - 'src/flb_network.c' + - 'src/http_server/**' + - 'tests/integration/scenarios/in_elasticsearch/**' + - 'tests/integration/scenarios/in_http/**' + - 'tests/integration/scenarios/in_http_max_connections/**' + - 'tests/integration/scenarios/in_opentelemetry/**' + - 'tests/integration/scenarios/in_prometheus_remote_write/**' + - 'tests/integration/scenarios/in_splunk/**' + - 'tests/integration/src/**' + - 'tests/internal/http_server.c' + workflow_dispatch: + +permissions: + contents: read + +jobs: + focused-linux: + name: Focused Linux integration and Valgrind + runs-on: ubuntu-22.04 + timeout-minutes: 30 + + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + persist-credentials: false + + - name: Install build and test dependencies + run: | + sudo apt-get update + sudo apt-get install -y \ + bison \ + build-essential \ + cmake \ + flex \ + libbpf-dev \ + libssl-dev \ + libsystemd-dev \ + libyaml-dev \ + python3-venv \ + valgrind + + - name: Configure and build + run: | + tests/integration/setup-venv.sh + cmake -S . -B build \ + -DFLB_TESTS_RUNTIME=On \ + -DFLB_TESTS_INTERNAL=On + cmake --build build -j8 + + - name: Run focused integration tests + run: | + tests/integration/.venv/bin/python -m pytest -q \ + tests/integration/scenarios/in_http/tests/test_in_http_001.py \ + tests/integration/scenarios/in_http_max_connections/tests/test_in_http_max_connections_001.py \ + tests/integration/scenarios/in_elasticsearch/tests/test_in_elasticsearch_001.py \ + tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py \ + tests/integration/scenarios/in_prometheus_remote_write/tests/test_in_prometheus_remote_write_001.py \ + tests/integration/scenarios/in_splunk/tests/test_in_splunk_001.py \ + -k 'pause_resume or shutdown_while or async_tls_accept_timeout or max_connections or idle_timeout' + + - name: Run focused integration tests with strict Valgrind + env: + VALGRIND: 1 + VALGRIND_STRICT: 1 + run: | + tests/integration/.venv/bin/python -m pytest -q \ + tests/integration/scenarios/in_http/tests/test_in_http_001.py \ + tests/integration/scenarios/in_http_max_connections/tests/test_in_http_max_connections_001.py \ + tests/integration/scenarios/in_elasticsearch/tests/test_in_elasticsearch_001.py \ + tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py \ + tests/integration/scenarios/in_prometheus_remote_write/tests/test_in_prometheus_remote_write_001.py \ + tests/integration/scenarios/in_splunk/tests/test_in_splunk_001.py \ + -k 'pause_resume or shutdown_while or async_tls_accept_timeout or max_connections or idle_timeout' From 58f516ec0aebbbadc7e661c2aa5bbe9ac48853d4 Mon Sep 17 00:00:00 2001 From: Eduardo Silva Date: Mon, 3 Aug 2026 15:59:40 -0600 Subject: [PATCH 16/21] tests: internal: cover pause callback failures Signed-off-by: Eduardo Silva --- tests/internal/input_pause.c | 129 +++++++++++++++++++++++++++++++++-- 1 file changed, 123 insertions(+), 6 deletions(-) diff --git a/tests/internal/input_pause.c b/tests/internal/input_pause.c index 79f8a397882..108c95450fe 100644 --- a/tests/internal/input_pause.c +++ b/tests/internal/input_pause.c @@ -1,6 +1,11 @@ /* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */ #include +#include +#include + +#include +#include #include @@ -36,7 +41,58 @@ static int checked_resume(void *data, struct flb_config *config) return context->result; } -static void test_checked_pause_resume_result() +static int input_pause_test_metrics_init(struct flb_input_instance *instance) +{ + int ret; + char *label_keys[] = {"name"}; + char *label_values[] = {instance->name}; + + instance->cmt = cmt_create(); + if (instance->cmt == NULL) { + return -1; + } + + instance->cmt_ingestion_paused = cmt_gauge_create( + instance->cmt, + "fluentbit", "input", "ingestion_paused", + "Is the input paused or not?", 1, label_keys); + if (instance->cmt_ingestion_paused == NULL) { + cmt_destroy(instance->cmt); + instance->cmt = NULL; + return -1; + } + + ret = cmt_gauge_set(instance->cmt_ingestion_paused, 0, 0, 1, label_values); + if (ret != 0) { + cmt_destroy(instance->cmt); + instance->cmt = NULL; + instance->cmt_ingestion_paused = NULL; + } + + return ret; +} + +static double input_pause_test_metric_value(struct flb_input_instance *instance) +{ + int ret; + double value; + char *label_values[] = {instance->name}; + + value = -1; + ret = cmt_gauge_get_val(instance->cmt_ingestion_paused, 1, label_values, &value); + TEST_CHECK(ret == 0); + + return value; +} + +static void input_pause_test_metrics_destroy(struct flb_input_instance *instance) +{ + cmt_destroy(instance->cmt); + instance->cmt = NULL; + instance->cmt_ingestion_paused = NULL; +} + +static void test_checked_pause_resume_result_and_metric() { int ret; struct flb_input_plugin plugin; @@ -51,27 +107,88 @@ static void test_checked_pause_resume_result() plugin.cb_resume_checked = checked_resume; instance.p = &plugin; instance.context = &context; + strncpy(instance.name, "test.0", sizeof(instance.name) - 1); + + ret = input_pause_test_metrics_init(&instance); + if (!TEST_CHECK(ret == 0)) { + return; + } context.result = -1; ret = flb_input_plugin_pause(&instance); TEST_CHECK(ret == -1); TEST_CHECK(context.pause_calls == 1); - - ret = flb_input_plugin_resume(&instance); - TEST_CHECK(ret == -1); - TEST_CHECK(context.resume_calls == 1); + TEST_CHECK(input_pause_test_metric_value(&instance) == 0); context.result = 0; ret = flb_input_plugin_pause(&instance); TEST_CHECK(ret == 0); TEST_CHECK(context.pause_calls == 2); + TEST_CHECK(input_pause_test_metric_value(&instance) == 1); + context.result = -1; + ret = flb_input_plugin_resume(&instance); + TEST_CHECK(ret == -1); + TEST_CHECK(context.resume_calls == 1); + TEST_CHECK(input_pause_test_metric_value(&instance) == 1); + + context.result = 0; ret = flb_input_plugin_resume(&instance); TEST_CHECK(ret == 0); TEST_CHECK(context.resume_calls == 2); + TEST_CHECK(input_pause_test_metric_value(&instance) == 0); + + input_pause_test_metrics_destroy(&instance); +} + +static void test_threaded_dispatch_failure_preserves_state() +{ + int ret; + struct flb_input_plugin plugin; + struct flb_input_instance instance; + struct flb_input_thread_instance thread_instance; + struct input_pause_test_context context; + + memset(&plugin, 0, sizeof(plugin)); + memset(&instance, 0, sizeof(instance)); + memset(&thread_instance, 0, sizeof(thread_instance)); + memset(&context, 0, sizeof(context)); + + plugin.cb_pause_checked = checked_pause; + plugin.cb_resume_checked = checked_resume; + instance.p = &plugin; + instance.context = &context; + instance.is_threaded = FLB_TRUE; + instance.thi = &thread_instance; + thread_instance.ch_parent_events[1] = FLB_INVALID_SOCKET; + strncpy(instance.name, "threaded.0", sizeof(instance.name) - 1); + + ret = input_pause_test_metrics_init(&instance); + if (!TEST_CHECK(ret == 0)) { + return; + } + + ret = flb_input_pause(&instance); + TEST_CHECK(ret == -1); + TEST_CHECK(context.pause_calls == 0); + TEST_CHECK(input_pause_test_metric_value(&instance) == 0); + + ret = cmt_gauge_set(instance.cmt_ingestion_paused, 0, 1, 1, + (char *[]) {instance.name}); + TEST_CHECK(ret == 0); + + ret = flb_input_resume(&instance); + TEST_CHECK(ret == -1); + TEST_CHECK(context.resume_calls == 0); + TEST_CHECK(input_pause_test_metric_value(&instance) == 1); + + input_pause_test_metrics_destroy(&instance); } TEST_LIST = { - { "checked_pause_resume_result", test_checked_pause_resume_result }, + { "checked_pause_resume_result_and_metric", + test_checked_pause_resume_result_and_metric }, + { "threaded_dispatch_failure_preserves_state", + test_threaded_dispatch_failure_preserves_state }, { 0 } }; From 354bf54589ca8e6d49651ddf4fc49292efcfe800 Mon Sep 17 00:00:00 2001 From: Eduardo Silva Date: Mon, 3 Aug 2026 15:59:45 -0600 Subject: [PATCH 17/21] tests: integration: cover input lifecycle edge cases Signed-off-by: Eduardo Silva --- .../tests/test_in_elasticsearch_001.py | 35 ++++++++- .../config/in_http_pause_resume_single.yaml | 20 ++++++ .../in_http/tests/test_in_http_001.py | 72 +++++++------------ .../config/otlp_pause_resume_single.yaml | 19 +++++ .../tests/test_in_opentelemetry_001.py | 63 +++++++++++++--- .../test_in_prometheus_remote_write_001.py | 19 +++++ .../in_splunk/tests/test_in_splunk_001.py | 31 +++++++- .../src/utils/input_pause_resume.py | 70 ++++++++++++++++++ 8 files changed, 272 insertions(+), 57 deletions(-) create mode 100644 tests/integration/scenarios/in_http/config/in_http_pause_resume_single.yaml create mode 100644 tests/integration/scenarios/in_opentelemetry/config/otlp_pause_resume_single.yaml diff --git a/tests/integration/scenarios/in_elasticsearch/tests/test_in_elasticsearch_001.py b/tests/integration/scenarios/in_elasticsearch/tests/test_in_elasticsearch_001.py index 531d06f2610..a8e81184ecf 100644 --- a/tests/integration/scenarios/in_elasticsearch/tests/test_in_elasticsearch_001.py +++ b/tests/integration/scenarios/in_elasticsearch/tests/test_in_elasticsearch_001.py @@ -4,7 +4,11 @@ from server.otlp_server import data_storage from utils.http_matrix import PROTOCOL_CASES, run_curl_request -from utils.input_pause_resume import assert_pause_resume_cycles, open_partial_http_request +from utils.input_pause_resume import ( + assert_pause_resume_cycles, + assert_shutdown_while_paused, + open_partial_http_request, +) from utils.test_service import FluentBitTestService logger = logging.getLogger(__name__) @@ -315,6 +319,35 @@ def open_active_connections(): service.stop() +@pytest.mark.parametrize( + "config_file", + [ + "in_elasticsearch_pause_resume.yaml", + "in_elasticsearch_pause_resume_workers.yaml", + ], + ids=["single_listener", "workers_4"], +) +def test_in_elasticsearch_shutdown_while_paused(config_file): + service = Service(config_file) + + try: + service.start() + large_document = '{"index":{}}\n{"message":"' + ("x" * 6144) + '"}\n' + assert_shutdown_while_paused( + service.flb, + service.stop, + "127.0.0.1", + service.flb_listener_port, + f"http://localhost:{service.flb_listener_port}/_bulk", + large_document, + ["Content-Type: application/x-ndjson"], + input_name="elasticsearch.0", + success_status=200, + ) + finally: + service.stop() + + @pytest.mark.parametrize( "case", IN_ELASTICSEARCH_SMALL_BUFFER_REGRESSION_CASES, diff --git a/tests/integration/scenarios/in_http/config/in_http_pause_resume_single.yaml b/tests/integration/scenarios/in_http/config/in_http_pause_resume_single.yaml new file mode 100644 index 00000000000..660352a3b2a --- /dev/null +++ b/tests/integration/scenarios/in_http/config/in_http_pause_resume_single.yaml @@ -0,0 +1,20 @@ +service: + flush: 12 + grace: 1 + log_level: info + http_server: on + http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT} + +pipeline: + inputs: + - name: http + port: ${FLUENT_BIT_TEST_LISTENER_PORT} + mem_buf_limit: 8KB + enable_health_endpoint: on + http2: off + workers: 1 + tls: off + + outputs: + - name: stdout + match: '*' diff --git a/tests/integration/scenarios/in_http/tests/test_in_http_001.py b/tests/integration/scenarios/in_http/tests/test_in_http_001.py index d0ffe6d4564..a82d65307ea 100644 --- a/tests/integration/scenarios/in_http/tests/test_in_http_001.py +++ b/tests/integration/scenarios/in_http/tests/test_in_http_001.py @@ -10,14 +10,13 @@ from server.http_server import data_storage, http_server_run from utils.http_matrix import PROTOCOL_CASES, run_curl_request from utils.input_pause_resume import ( - ConnectionFlood, assert_connection_closed, assert_pause_resume_cycles, + assert_shutdown_while_paused, is_valgrind, large_json_payload, open_partial_http_request, open_stalled_tcp_connection, - wait_for_input_pause_state, ) from utils.test_service import FluentBitTestService @@ -154,6 +153,13 @@ def test_in_http_rejects_get_requests(): @pytest.mark.parametrize( "case", [ + { + "id": "http1_cleartext_single_listener", + "config": "in_http_pause_resume_single.yaml", + "scheme": "http", + "http_mode": "http1.1", + "stalled_connection": open_partial_http_request, + }, { "id": "http1_cleartext_workers", "config": "in_http_pause_resume.yaml", @@ -203,56 +209,28 @@ def open_active_connections(): service.stop() -def test_in_http_shutdown_while_paused_with_active_connections(): - service = Service("in_http_pause_resume.yaml") - service.start() - connection_flood = ConnectionFlood( - "127.0.0.1", - service.flb_listener_port, - ) - stalled_connections = [] +@pytest.mark.parametrize( + "config_file", + ["in_http_pause_resume_single.yaml", "in_http_pause_resume.yaml"], + ids=["single_listener", "workers_4"], +) +def test_in_http_shutdown_while_paused_with_active_connections(config_file): + service = Service(config_file) try: - for _ in range(8): - stalled_connections.append( - open_partial_http_request( - "127.0.0.1", - service.flb_listener_port, - ) - ) - - connection_flood.start() - connection_flood.wait_for_attempts(256) - - for _ in range(2): - result = run_curl_request( - f"http://localhost:{service.flb_listener_port}/", - large_json_payload(size=6144), - headers=["Content-Type: application/json"], - http_mode="http1.1", - ) - assert result["status_code"] == 201, result - - wait_for_input_pause_state( + service.start() + assert_shutdown_while_paused( service.flb, - "http.0", - True, - timeout=20 if is_valgrind() else 10, + service.stop, + "127.0.0.1", + service.flb_listener_port, + f"http://localhost:{service.flb_listener_port}/", + large_json_payload(size=6144), + ["Content-Type: application/json"], + input_name="http.0", + success_status=201, ) - - shutdown_started = time.monotonic() - service.stop() - shutdown_elapsed = time.monotonic() - shutdown_started - - shutdown_limit = 20 if is_valgrind() else 5 - assert shutdown_elapsed < shutdown_limit - - for connection in stalled_connections: - assert_connection_closed(connection) finally: - connection_flood.stop() - for connection in stalled_connections: - connection.close() service.stop() diff --git a/tests/integration/scenarios/in_opentelemetry/config/otlp_pause_resume_single.yaml b/tests/integration/scenarios/in_opentelemetry/config/otlp_pause_resume_single.yaml new file mode 100644 index 00000000000..8cac6e6cfcb --- /dev/null +++ b/tests/integration/scenarios/in_opentelemetry/config/otlp_pause_resume_single.yaml @@ -0,0 +1,19 @@ +service: + flush: 12 + grace: 1 + log_level: info + http_server: on + http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT} + +pipeline: + inputs: + - name: opentelemetry + port: ${FLUENT_BIT_TEST_LISTENER_PORT} + mem_buf_limit: 8KB + http2: on + tls: off + http_server.workers: 1 + + outputs: + - name: stdout + match: '*' diff --git a/tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py b/tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py index 69e35b0533b..e1446e91ded 100644 --- a/tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py +++ b/tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py @@ -36,7 +36,11 @@ # local imports from utils.data_utils import read_json_file from utils.http_matrix import PROTOCOL_CASES, run_curl_request -from utils.input_pause_resume import assert_pause_resume_cycles, open_stalled_tcp_connection +from utils.input_pause_resume import ( + assert_pause_resume_cycles, + assert_shutdown_while_paused, + open_stalled_tcp_connection, +) from utils.test_service import FluentBitTestService from server.http_server import http_server_run @@ -1564,16 +1568,30 @@ def send_job(job): @pytest.mark.parametrize( - "signal_type,json_input,endpoint", + "config_file,signal_type,json_input,endpoint", [ - ("logs", "test_logs_001.in.json", "/v1/logs"), - ("metrics", "test_metrics_001.in.json", "/v1/metrics"), - ("traces", "test_traces_001.in.json", "/v1/traces"), + (config_file, signal_type, json_input, endpoint) + for config_file in [ + "otlp_pause_resume_single.yaml", + "otlp_pause_resume_workers.yaml", + ] + for signal_type, json_input, endpoint in [ + ("logs", "test_logs_001.in.json", "/v1/logs"), + ("metrics", "test_metrics_001.in.json", "/v1/metrics"), + ("traces", "test_traces_001.in.json", "/v1/traces"), + ] + ], + ids=[ + f"{'single_listener' if 'single' in config_file else 'workers_4'}-{signal_type}" + for config_file in [ + "otlp_pause_resume_single.yaml", + "otlp_pause_resume_workers.yaml", + ] + for signal_type in ["logs", "metrics", "traces"] ], - ids=["logs", "metrics", "traces"], ) -def test_in_opentelemetry_pause_resume_workers(signal_type, json_input, endpoint): - service = Service("otlp_pause_resume_workers.yaml") +def test_in_opentelemetry_pause_resume(config_file, signal_type, json_input, endpoint): + service = Service(config_file) try: service.start() @@ -1605,6 +1623,35 @@ def open_active_connections(): service.stop() +@pytest.mark.parametrize( + "config_file", + ["otlp_pause_resume_single.yaml", "otlp_pause_resume_workers.yaml"], + ids=["single_listener", "workers_4"], +) +def test_in_opentelemetry_shutdown_while_paused(config_file): + service = Service(config_file) + + try: + service.start() + payload = service.build_otel_payload("test_logs_001.in.json", "logs") + assert_shutdown_while_paused( + service.flb, + service.stop, + "127.0.0.1", + service.flb_listener_port, + f"http://localhost:{service.flb_listener_port}/v1/logs", + payload, + ["Content-Type: application/x-protobuf"], + input_name="opentelemetry.0", + success_status=201, + connection_factory=open_stalled_tcp_connection, + pause_trigger_requests=64, + http_mode="http2-prior-knowledge", + ) + finally: + service.stop() + + def test_in_opentelemetry_http_workers_export_ingress_queue_metrics(): service = Service(IN_OPENTELEMETRY_WORKER_PROTOCOL_CONFIGS["http1_cleartext"]) service.start() diff --git a/tests/integration/scenarios/in_prometheus_remote_write/tests/test_in_prometheus_remote_write_001.py b/tests/integration/scenarios/in_prometheus_remote_write/tests/test_in_prometheus_remote_write_001.py index 014ed7e29b2..ba18d7fe609 100644 --- a/tests/integration/scenarios/in_prometheus_remote_write/tests/test_in_prometheus_remote_write_001.py +++ b/tests/integration/scenarios/in_prometheus_remote_write/tests/test_in_prometheus_remote_write_001.py @@ -6,6 +6,7 @@ from utils.fluent_bit_manager import FluentBitManager from utils.input_pause_resume import ( assert_connection_closed, + is_valgrind, open_partial_http_request, open_stalled_tcp_connection, wait_for_input_pause_state, @@ -165,6 +166,7 @@ def test_in_prometheus_remote_write_pause_resume_and_shutdown(receiver_config): "sender_cleartext.yaml", ) stalled_connections = [] + shutdown_connections = [] paused_connection = None try: @@ -213,6 +215,14 @@ def test_in_prometheus_remote_write_pause_resume_and_shutdown(receiver_config): delivered_before_resume = _read_file(service.receiver.log_file).count( "fluentbit_input_metrics_scrapes_total" ) + for _ in range(8): + shutdown_connections.append( + open_partial_http_request( + "127.0.0.1", + service.receiver_port, + ) + ) + service.start_sender() service.wait_for_log_count( service.receiver.log_file, @@ -227,10 +237,19 @@ def test_in_prometheus_remote_write_pause_resume_and_shutdown(receiver_config): True, timeout=30, ) + + for connection in shutdown_connections: + assert_connection_closed(connection) + + shutdown_started = time.monotonic() service.stop() + shutdown_elapsed = time.monotonic() - shutdown_started + assert shutdown_elapsed < (30 if is_valgrind() else 8) finally: for connection in stalled_connections: connection.close() + for connection in shutdown_connections: + connection.close() if paused_connection is not None: paused_connection.close() service.stop() diff --git a/tests/integration/scenarios/in_splunk/tests/test_in_splunk_001.py b/tests/integration/scenarios/in_splunk/tests/test_in_splunk_001.py index b3644c7b673..cf4cf5c8fff 100644 --- a/tests/integration/scenarios/in_splunk/tests/test_in_splunk_001.py +++ b/tests/integration/scenarios/in_splunk/tests/test_in_splunk_001.py @@ -6,7 +6,11 @@ import requests from server.http_server import configure_http_response, data_storage, http_server_run -from utils.input_pause_resume import assert_pause_resume_cycles, open_partial_http_request +from utils.input_pause_resume import ( + assert_pause_resume_cycles, + assert_shutdown_while_paused, + open_partial_http_request, +) from utils.test_service import FluentBitTestService from utils.http_matrix import PROTOCOL_CASES, run_curl_request @@ -299,6 +303,31 @@ def open_active_connections(): service.stop() +@pytest.mark.parametrize( + "config_file", + ["splunk_pause_resume.yaml", "splunk_pause_resume_workers.yaml"], + ids=["single_listener", "workers_4"], +) +def test_in_splunk_shutdown_while_paused(config_file): + service = Service(config_file) + + try: + service.start() + assert_shutdown_while_paused( + service.flb, + service.stop, + "127.0.0.1", + service.flb_listener_port, + f"http://localhost:{service.flb_listener_port}/services/collector", + json.dumps({"event": "x" * 6144}), + create_splunk_headers(), + input_name="splunk.0", + success_status=200, + ) + finally: + service.stop() + + def test_in_splunk_to_out_splunk_prefers_configured_output_token(): service = ForwardingService("in_splunk_to_out_splunk.yaml") service.start() diff --git a/tests/integration/src/utils/input_pause_resume.py b/tests/integration/src/utils/input_pause_resume.py index 2f6cd0e30be..b40ae319e8e 100644 --- a/tests/integration/src/utils/input_pause_resume.py +++ b/tests/integration/src/utils/input_pause_resume.py @@ -258,6 +258,76 @@ def assert_pause_resume_cycles( assert result["status_code"] == success_status, result +def assert_shutdown_while_paused( + flb, + stop_service, + host, + port, + url, + payload, + headers, + *, + input_name, + success_status, + connection_factory=open_partial_http_request, + connection_count=8, + pause_trigger_requests=2, + http_mode="http1.1", + ca_cert_path=None, +): + active_connections = [] + connection_flood = ConnectionFlood(host, port) + + try: + for _ in range(connection_count): + active_connections.append(connection_factory(host, port)) + + connection_flood.start() + connection_flood.wait_for_attempts(128) + + for _ in range(pause_trigger_requests): + result = run_curl_without_check( + url, + payload, + headers=headers, + http_mode=http_mode, + max_time=10, + ca_cert_path=ca_cert_path, + ) + if input_pause_state(flb, input_name) == 1: + break + + if result.returncode != 0 or curl_status_code(result) == 0: + wait_for_input_pause_state(flb, input_name, True, timeout=2) + break + + assert curl_status_code(result) == success_status, result.stdout + + wait_for_input_pause_state( + flb, + input_name, + True, + timeout=30 if is_valgrind() else 15, + ) + + shutdown_started = time.monotonic() + stop_service() + shutdown_elapsed = time.monotonic() - shutdown_started + + shutdown_limit = 30 if is_valgrind() else 8 + assert shutdown_elapsed < shutdown_limit + + for connection in active_connections: + assert_connection_closed( + connection, + timeout=20 if is_valgrind() else 5, + ) + finally: + connection_flood.stop() + for connection in active_connections: + connection.close() + + def run_curl_without_check( url, payload, From a5d52b03a9b33bd4fd8a98beb1e33816382c23f3 Mon Sep 17 00:00:00 2001 From: Eduardo Silva Date: Mon, 3 Aug 2026 20:04:43 -0600 Subject: [PATCH 18/21] tests: integration: synchronize max connection checks Signed-off-by: Eduardo Silva --- .../tests/test_in_http_max_connections_001.py | 123 +++++++++--------- 1 file changed, 65 insertions(+), 58 deletions(-) diff --git a/tests/integration/scenarios/in_http_max_connections/tests/test_in_http_max_connections_001.py b/tests/integration/scenarios/in_http_max_connections/tests/test_in_http_max_connections_001.py index a0500778e6e..330edbe1bc9 100644 --- a/tests/integration/scenarios/in_http_max_connections/tests/test_in_http_max_connections_001.py +++ b/tests/integration/scenarios/in_http_max_connections/tests/test_in_http_max_connections_001.py @@ -110,6 +110,45 @@ def _wait_for_accepted_request(service, payload, http_mode, timeout=10, interval return {"status_code": 0, "http_version": ""} +def _wait_for_rejected_request(service, payload, http_mode, timeout=10, interval=0.05): + """Wait until the held connection is accepted and consumes the only slot.""" + deadline = time.monotonic() + timeout + + while time.monotonic() < deadline: + try: + response = run_curl_request( + f"http://127.0.0.1:{service.flb_listener_port}/", + payload=payload, + headers=["Content-Type: application/json"], + http_mode=http_mode, + ) + if response["status_code"] != 201: + return + except Exception: + return + + time.sleep(interval) + + raise AssertionError("held connection never occupied the configured connection slot") + + +def _wait_for_forwarded_message(service, message): + def find_message(): + for payload in data_storage["payloads"]: + for record in payload: + if record.get("message") == message: + return record + + return None + + return service.service.wait_for_condition( + find_message, + timeout=10, + interval=0.5, + description=f"forwarded payload {message}", + ) + + def test_in_http_max_connections_blocks_and_recovers(): service = Service() accepted = {"status_code": 0} @@ -122,20 +161,13 @@ def test_in_http_max_connections_blocks_and_recovers(): try: held_connection = socket.create_connection(("127.0.0.1", service.flb_listener_port), timeout=2) held_connection.settimeout(2) + held_connection.sendall(b"POST / HTTP/1.1\r\n") - overflow_rejected = False - try: - response = run_curl_request( - f"http://127.0.0.1:{service.flb_listener_port}/", - payload='{"message":"max-connections"}', - headers=["Content-Type: application/json"], - http_mode="http1.1", - ) - overflow_rejected = response["status_code"] != 201 - except Exception: - overflow_rejected = True - - assert overflow_rejected + _wait_for_rejected_request( + service, + payload='{"message":"max-connections-probe"}', + http_mode="http1.1", + ) finally: if held_connection: held_connection.close() @@ -146,17 +178,12 @@ def test_in_http_max_connections_blocks_and_recovers(): headers=["Content-Type: application/json"], http_mode="http1.1", ) - forwarded_payloads = service.service.wait_for_condition( - lambda: data_storage["payloads"] if data_storage["payloads"] else None, - timeout=10, - interval=0.5, - description="forwarded max-connections payload", - ) + forwarded_payloads = _wait_for_forwarded_message(service, "max-connections") finally: service.stop() assert accepted["status_code"] == 201 - assert forwarded_payloads[0][0]["message"] == "max-connections" + assert forwarded_payloads["message"] == "max-connections" def test_in_http_idle_timeout_evicts_partial_request_connection(): @@ -173,19 +200,11 @@ def test_in_http_idle_timeout_evicts_partial_request_connection(): held_connection.settimeout(5) held_connection.sendall(b"POST / HTTP/1.1\r\n") - overflow_rejected = False - try: - response = run_curl_request( - f"http://127.0.0.1:{service.flb_listener_port}/", - payload='{"message":"idle-timeout-blocked"}', - headers=["Content-Type: application/json"], - http_mode="http1.1", - ) - overflow_rejected = response["status_code"] != 201 - except Exception: - overflow_rejected = True - - assert overflow_rejected + _wait_for_rejected_request( + service, + payload='{"message":"idle-timeout-blocked"}', + http_mode="http1.1", + ) response = _wait_for_accepted_request( service, @@ -196,17 +215,15 @@ def test_in_http_idle_timeout_evicts_partial_request_connection(): if held_connection: held_connection.close() held_connection = None - forwarded_payloads = service.service.wait_for_condition( - lambda: data_storage["payloads"] if data_storage["payloads"] else None, - timeout=10, - interval=0.5, - description="forwarded idle-timeout payload", + forwarded_payloads = _wait_for_forwarded_message( + service, + "idle-timeout-recovered", ) finally: service.stop() assert response["status_code"] == 201 - assert forwarded_payloads[0][0]["message"] == "idle-timeout-recovered" + assert forwarded_payloads["message"] == "idle-timeout-recovered" def test_in_http_idle_timeout_evicts_partial_http2_preface_connection(): @@ -226,19 +243,11 @@ def test_in_http_idle_timeout_evicts_partial_http2_preface_connection(): held_connection.settimeout(5) held_connection.sendall(b"PRI * HTTP/2.0\r\n\r\nSM\r\n") - overflow_rejected = False - try: - response = run_curl_request( - f"http://127.0.0.1:{service.flb_listener_port}/", - payload='{"message":"idle-timeout-http2-blocked"}', - headers=["Content-Type: application/json"], - http_mode="http2-prior-knowledge", - ) - overflow_rejected = response["status_code"] != 201 - except Exception: - overflow_rejected = True - - assert overflow_rejected + _wait_for_rejected_request( + service, + payload='{"message":"idle-timeout-http2-blocked"}', + http_mode="http2-prior-knowledge", + ) response = _wait_for_accepted_request( service, @@ -249,15 +258,13 @@ def test_in_http_idle_timeout_evicts_partial_http2_preface_connection(): if held_connection: held_connection.close() held_connection = None - forwarded_payloads = service.service.wait_for_condition( - lambda: data_storage["payloads"] if data_storage["payloads"] else None, - timeout=10, - interval=0.5, - description="forwarded idle-timeout http2 payload", + forwarded_payloads = _wait_for_forwarded_message( + service, + "idle-timeout-http2-recovered", ) finally: service.stop() assert response["status_code"] == 201 assert response["http_version"] == "2" - assert forwarded_payloads[0][0]["message"] == "idle-timeout-http2-recovered" + assert forwarded_payloads["message"] == "idle-timeout-http2-recovered" From d3607d544e116c910fa0dea7efa68d25b327d97d Mon Sep 17 00:00:00 2001 From: Eduardo Silva Date: Mon, 3 Aug 2026 20:57:32 -0600 Subject: [PATCH 19/21] downstream: dispatch callbacks on parent stack Signed-off-by: Eduardo Silva --- include/fluent-bit/flb_connection.h | 3 ++ include/fluent-bit/flb_downstream.h | 9 +++++ src/flb_downstream.c | 60 ++++++++++++++++++++++++----- 3 files changed, 63 insertions(+), 9 deletions(-) diff --git a/include/fluent-bit/flb_connection.h b/include/fluent-bit/flb_connection.h index f546cfa1d70..4315424f59f 100644 --- a/include/fluent-bit/flb_connection.h +++ b/include/fluent-bit/flb_connection.h @@ -162,6 +162,9 @@ struct flb_connection { /* Downstream-owned event callback coroutine */ struct flb_coro *event_coroutine; flb_connection_event_callback event_callback; + flb_connection_event_callback event_parent_callback; + void *event_parent_callback_data; + int event_parent_callback_result; int event_wakeup_pending; int event_release_pending; diff --git a/include/fluent-bit/flb_downstream.h b/include/fluent-bit/flb_downstream.h index fe8108533b9..6bb446c056d 100644 --- a/include/fluent-bit/flb_downstream.h +++ b/include/fluent-bit/flb_downstream.h @@ -101,6 +101,15 @@ int flb_downstream_conn_event_accept( int flb_downstream_conn_event_register(struct flb_connection *connection, int (*callback)(void *data), int mask); + +/* + * Suspend an event coroutine while callback runs on its parent stack. This is + * required for code which depends on native thread stack bounds, such as WAMR. + */ +int flb_downstream_conn_event_call_parent( + struct flb_connection *connection, + flb_connection_event_callback callback, + void *callback_data); void flb_downstream_conn_event_resume(struct flb_connection *connection); int flb_downstream_conn_pending_destroy_list(struct mk_list *list); diff --git a/src/flb_downstream.c b/src/flb_downstream.c index 63522e362a6..b211a33db67 100644 --- a/src/flb_downstream.c +++ b/src/flb_downstream.c @@ -721,7 +721,6 @@ int flb_downstream_conn_event_accept( size_t stack_size; flb_sockfd_t connection_fd; struct flb_coro *coro; - struct flb_coro *previous_coro; struct flb_connection *connection; struct flb_config *config; @@ -792,9 +791,7 @@ int flb_downstream_conn_event_accept( mk_list_add(&connection->_head, &stream->busy_queue); flb_stream_release_lock(&stream->base); - previous_coro = flb_coro_get(); - flb_coro_resume(coro); - flb_coro_set(previous_coro); + flb_downstream_conn_event_resume(connection); ret = 0; if (connection->fd == FLB_INVALID_SOCKET && @@ -812,7 +809,6 @@ int flb_downstream_conn_event_register(struct flb_connection *connection, int ret; size_t stack_size; struct flb_coro *coro; - struct flb_coro *previous_coro; struct flb_config *config; if (connection == NULL || callback == NULL || @@ -859,9 +855,7 @@ int flb_downstream_conn_event_register(struct flb_connection *connection, flb_trace("[downstream] register event coroutine for connection #%i", connection->fd); - previous_coro = flb_coro_get(); - flb_coro_resume(coro); - flb_coro_set(previous_coro); + flb_downstream_conn_event_resume(connection); ret = mk_event_add(connection->evl, connection->fd, @@ -880,13 +874,61 @@ int flb_downstream_conn_event_register(struct flb_connection *connection, return 0; } +int flb_downstream_conn_event_call_parent( + struct flb_connection *connection, + flb_connection_event_callback callback, + void *callback_data) +{ + struct flb_coro *coro; + + if (connection == NULL || callback == NULL) { + return -1; + } + + coro = flb_coro_get(); + if (coro == NULL || coro != connection->event_coroutine) { + return callback(callback_data); + } + + if (connection->event_parent_callback != NULL) { + return -1; + } + + connection->event_parent_callback = callback; + connection->event_parent_callback_data = callback_data; + flb_coro_yield(coro, FLB_FALSE); + + return connection->event_parent_callback_result; +} + void flb_downstream_conn_event_resume(struct flb_connection *connection) { + int result; + void *callback_data; + flb_connection_event_callback callback; struct flb_coro *previous_coro; previous_coro = flb_coro_get(); connection->event_wakeup_pending = FLB_FALSE; - flb_coro_resume(connection->event_coroutine); + + while (connection->event_coroutine != NULL) { + flb_coro_resume(connection->event_coroutine); + flb_coro_set(previous_coro); + + callback = connection->event_parent_callback; + if (callback == NULL) { + break; + } + + callback_data = connection->event_parent_callback_data; + connection->event_parent_callback = NULL; + connection->event_parent_callback_data = NULL; + + result = callback(callback_data); + connection->event_parent_callback_result = result; + connection->event_wakeup_pending = FLB_FALSE; + } + flb_coro_set(previous_coro); } From ead676146f3689b7ee731b7845e9902f6c0c9af6 Mon Sep 17 00:00:00 2001 From: Eduardo Silva Date: Mon, 3 Aug 2026 20:57:39 -0600 Subject: [PATCH 20/21] http_server: run request callbacks on parent stack Signed-off-by: Eduardo Silva --- src/http_server/flb_http_server.c | 26 +++++++++++++++++++++++++- 1 file changed, 25 insertions(+), 1 deletion(-) diff --git a/src/http_server/flb_http_server.c b/src/http_server/flb_http_server.c index e3befbb2c61..6f8e1ef6bc5 100644 --- a/src/http_server/flb_http_server.c +++ b/src/http_server/flb_http_server.c @@ -374,6 +374,22 @@ static int flb_http_server_should_connection_be_closed( return FLB_TRUE; } +struct flb_http_server_request_callback_context { + struct flb_http_server *server; + struct flb_http_request *request; + struct flb_http_response *response; +}; + +static int flb_http_server_request_callback_dispatch(void *data) +{ + struct flb_http_server_request_callback_context *context; + + context = data; + + return context->server->request_callback(context->request, + context->response); +} + static int flb_http_server_client_activity_event_handler(void *data) { int close_connection; @@ -385,6 +401,7 @@ static int flb_http_server_client_activity_event_handler(void *data) struct flb_http_server_session *session; struct flb_http_server *server; struct flb_http_stream *stream; + struct flb_http_server_request_callback_context callback_context; int result; struct mk_event *event; @@ -453,7 +470,14 @@ static int flb_http_server_client_activity_event_handler(void *data) } if (server->request_callback != NULL) { - result = server->request_callback(request, response); + callback_context.server = server; + callback_context.request = request; + callback_context.response = response; + + result = flb_downstream_conn_event_call_parent( + connection, + flb_http_server_request_callback_dispatch, + &callback_context); } else { /* Report */ From d7fd23291cecf7b1336dea6dac149a35d1cc57a8 Mon Sep 17 00:00:00 2001 From: Eduardo Silva Date: Mon, 3 Aug 2026 20:57:39 -0600 Subject: [PATCH 21/21] tests: cover parent stack callback dispatch Signed-off-by: Eduardo Silva --- tests/internal/network.c | 84 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 84 insertions(+) diff --git a/tests/internal/network.c b/tests/internal/network.c index 427f884dad2..c7c79fa5e4f 100644 --- a/tests/internal/network.c +++ b/tests/internal/network.c @@ -7,6 +7,8 @@ #include #include #include +#include +#include #include #include "flb_tests_internal.h" @@ -19,6 +21,43 @@ #define TEST_EV_CLIENT MK_EVENT_NOTIFICATION #define TEST_EV_SERVER MK_EVENT_CUSTOM +struct parent_callback_context { + struct flb_connection connection; + struct flb_coro *parent_coro; + int callback_on_parent; + int callback_result; + int coroutine_done; +}; + +static int parent_callback(void *data) +{ + struct parent_callback_context *context; + + context = data; + context->callback_on_parent = flb_coro_get() == context->parent_coro; + + return 73; +} + +static void parent_callback_coro(void) +{ + struct flb_coro *coro; + struct parent_callback_context *context; + + coro = flb_coro_get(); + context = coro->data; + + context->callback_result = flb_downstream_conn_event_call_parent( + &context->connection, + parent_callback, + context); + context->coroutine_done = FLB_TRUE; + + while (FLB_TRUE) { + flb_coro_yield(coro, FLB_FALSE); + } +} + static int socket_check_ok(flb_sockfd_t fd) { int ret; @@ -201,10 +240,55 @@ void test_accept_empty_nonblocking_listener() flb_socket_close(fd_server); } +void test_downstream_event_callback_runs_on_parent_stack() +{ + size_t stack_size; + struct flb_coro *coro; + struct parent_callback_context context; + + memset(&context, 0, sizeof(context)); + + flb_coro_thread_init(); + + coro = flb_coro_create(&context); + if (!TEST_CHECK(coro != NULL)) { + return; + } + + coro->caller = co_active(); + coro->callee = co_create(test_env_config->coro_stack_size, + parent_callback_coro, + &stack_size); + if (!TEST_CHECK(coro->callee != NULL)) { + flb_coro_destroy(coro); + return; + } + +#ifdef FLB_HAVE_VALGRIND + coro->valgrind_stack_id = VALGRIND_STACK_REGISTER( + coro->callee, + ((char *) coro->callee) + stack_size); +#endif + + context.parent_coro = flb_coro_get(); + context.connection.event_coroutine = coro; + + flb_downstream_conn_event_resume(&context.connection); + + TEST_CHECK(context.callback_on_parent == FLB_TRUE); + TEST_CHECK(context.callback_result == 73); + TEST_CHECK(context.coroutine_done == FLB_TRUE); + + context.connection.event_coroutine = NULL; + flb_coro_destroy(coro); +} + TEST_LIST = { { "ipv4_client_server", test_ipv4_client_server}, { "ipv6_client_server", test_ipv6_client_server}, { "ipv6_bracketed_listen", test_ipv6_bracketed_listen}, { "accept_empty_nonblocking_listener", test_accept_empty_nonblocking_listener}, + { "downstream_event_callback_runs_on_parent_stack", + test_downstream_event_callback_runs_on_parent_stack }, { 0 } };