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' diff --git a/include/fluent-bit/flb_connection.h b/include/fluent-bit/flb_connection.h index 3eed648e61d..4315424f59f 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; @@ -156,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 58c92155efb..6bb446c056d 100644 --- a/include/fluent-bit/flb_downstream.h +++ b/include/fluent-bit/flb_downstream.h @@ -86,14 +86,30 @@ 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); + +/* + * 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/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/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/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 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 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 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 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 diff --git a/src/flb_downstream.c b/src/flb_downstream.c index 55b2bcdd33f..b211a33db67 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,135 @@ 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_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); + + flb_downstream_conn_event_resume(connection); + + 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) @@ -591,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 || @@ -638,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, @@ -659,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); } 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); } } 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) { 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"); } diff --git a/src/http_server/flb_http_server.c b/src/http_server/flb_http_server.c index 35deef0d937..6f8e1ef6bc5 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 @@ -364,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; @@ -375,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; @@ -399,6 +426,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); @@ -435,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 */ @@ -462,28 +504,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 +526,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 +538,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 +554,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 +997,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 +1031,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 +1308,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; 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..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,6 +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, + assert_shutdown_while_paused, + open_partial_http_request, +) from utils.test_service import FluentBitTestService logger = logging.getLogger(__name__) @@ -273,6 +278,76 @@ 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( + "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_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/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 519af7ae538..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 @@ -9,6 +9,15 @@ 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 ( + assert_connection_closed, + assert_pause_resume_cycles, + assert_shutdown_while_paused, + is_valgrind, + large_json_payload, + open_partial_http_request, + open_stalled_tcp_connection, +) from utils.test_service import FluentBitTestService logger = logging.getLogger(__name__) @@ -141,6 +150,116 @@ def test_in_http_rejects_get_requests(): assert result["status_code"] >= 400 +@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", + "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() + + +@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: + 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}/", + large_json_payload(size=6144), + ["Content-Type: application/json"], + input_name="http.0", + success_status=201, + ) + finally: + 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_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" 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/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..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,6 +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, + assert_shutdown_while_paused, + open_stalled_tcp_connection, +) from utils.test_service import FluentBitTestService from server.http_server import http_server_run @@ -1562,6 +1567,91 @@ def send_job(job): assert "/v1/traces" in paths_seen +@pytest.mark.parametrize( + "config_file,signal_type,json_input,endpoint", + [ + (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"] + ], +) +def test_in_opentelemetry_pause_resume(config_file, signal_type, json_input, endpoint): + service = Service(config_file) + + 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() + + +@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/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..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 @@ -4,6 +4,13 @@ import pytest 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, +) from utils.network import find_available_port @@ -72,7 +79,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 +89,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 +114,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 +150,106 @@ 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 = [] + shutdown_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" + ) + 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, + "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, + ) + + 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/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..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,6 +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, + assert_shutdown_while_paused, + open_partial_http_request, +) from utils.test_service import FluentBitTestService from utils.http_matrix import PROTOCOL_CASES, run_curl_request @@ -257,6 +262,72 @@ 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() + + +@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 new file mode 100644 index 00000000000..b40ae319e8e --- /dev/null +++ b/tests/integration/src/utils/input_pause_resume.py @@ -0,0 +1,384 @@ +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 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, + 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")) 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/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 } }; diff --git a/tests/internal/input_pause.c b/tests/internal/input_pause.c new file mode 100644 index 00000000000..108c95450fe --- /dev/null +++ b/tests/internal/input_pause.c @@ -0,0 +1,194 @@ +/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */ + +#include +#include +#include + +#include +#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 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; + 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; + 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); + 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_and_metric", + test_checked_pause_resume_result_and_metric }, + { "threaded_dispatch_failure_preserves_state", + test_threaded_dispatch_failure_preserves_state }, + { 0 } +}; diff --git a/tests/internal/network.c b/tests/internal/network.c index b2941c81759..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; @@ -181,9 +220,75 @@ 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); +} + +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 } };