From 9b03b357a69fad63f21ee1f034905bdb29793bff Mon Sep 17 00:00:00 2001 From: Michael Renner Date: Wed, 5 Aug 2026 13:57:40 +0200 Subject: [PATCH 1/3] input_chunk: support storage limits with many outputs The storage limit path used a signed int as a secondary output selection mask. Shifting by output IDs at or above 32 is undefined and can alias a high-ID output to a lower output on common architectures. Check each routed output against its configured limit directly before releasing queue space. This keeps limit enforcement aligned with the dynamically sized routing masks. Signed-off-by: Michael Renner --- src/flb_input_chunk.c | 27 +++++++++++++++------------ 1 file changed, 15 insertions(+), 12 deletions(-) diff --git a/src/flb_input_chunk.c b/src/flb_input_chunk.c index 2c11af562a3..7728ebde691 100644 --- a/src/flb_input_chunk.c +++ b/src/flb_input_chunk.c @@ -1102,7 +1102,7 @@ int flb_input_chunk_release_space_compound( * will drop the the oldest chunks when the limitation on local disk is reached. */ int flb_input_chunk_find_space_new_data(struct flb_input_chunk *ic, - size_t chunk_size, int overlimit) + size_t chunk_size) { int count; int result; @@ -1111,20 +1111,23 @@ int flb_input_chunk_find_space_new_data(struct flb_input_chunk *ic, size_t local_release_requirement; /* - * For each output instances that will be over the limit after adding the new chunk, - * we have to determine how many chunks needs to be removed. We will adjust the - * routes_mask to only route to the output plugin that have enough space after - * deleting some chunks fome the queue. + * For each output instance that will be over the limit after adding the new chunk, + * we have to determine how many chunks need to be removed. We will adjust the + * routes_mask to only route to the output plugin that has enough space after + * deleting some chunks from the queue. */ count = 0; mk_list_foreach(head, &ic->in->config->outputs) { o_ins = mk_list_entry(head, struct flb_output_instance, _head); - if ((o_ins->total_limit_size == -1) || ((1 << o_ins->id) & overlimit) == 0 || - (flb_routes_mask_get_bit(ic->routes_mask, - o_ins->id, - o_ins->config->router) == 0)) { + if ((o_ins->total_limit_size == -1) || + (flb_routes_mask_get_bit(ic->routes_mask, + o_ins->id, + o_ins->config->router) == 0) || + (o_ins->fs_chunks_size + + o_ins->fs_backlog_chunks_size + + chunk_size) <= o_ins->total_limit_size) { continue; } @@ -1182,7 +1185,7 @@ int flb_input_chunk_has_overlimit_routes(struct flb_input_chunk *ic, if ((o_ins->fs_chunks_size + o_ins->fs_backlog_chunks_size + chunk_size) > o_ins->total_limit_size) { - overlimit |= (1 << o_ins->id); + overlimit = FLB_TRUE; } } @@ -1195,13 +1198,13 @@ int flb_input_chunk_has_overlimit_routes(struct flb_input_chunk *ic, int flb_input_chunk_place_new_chunk(struct flb_input_chunk *ic, size_t chunk_size) { int result; - int overlimit; + int overlimit; struct flb_input_instance *i_ins = ic->in; if (i_ins->storage_type == CIO_STORE_FS) { overlimit = flb_input_chunk_has_overlimit_routes(ic, chunk_size); if (overlimit != 0) { - result = flb_input_chunk_find_space_new_data(ic, chunk_size, overlimit); + result = flb_input_chunk_find_space_new_data(ic, chunk_size); if (result != 0) { return 0; From b1128fa38e8d4eb2941a30b1fd99a9d4a25339e2 Mon Sep 17 00:00:00 2001 From: Michael Renner Date: Wed, 5 Aug 2026 13:57:40 +0200 Subject: [PATCH 2/3] tests: internal: cover storage limits with many outputs Create 33 outputs and place the constrained output at ID 32. Verify that eviction removes only its route while preserving the unrelated route at ID 0. Signed-off-by: Michael Renner --- tests/internal/input_chunk.c | 162 +++++++++++++++++++++++++++++++++++ 1 file changed, 162 insertions(+) diff --git a/tests/internal/input_chunk.c b/tests/internal/input_chunk.c index d9b51c85d07..1d75edfc9d2 100644 --- a/tests/internal/input_chunk.c +++ b/tests/internal/input_chunk.c @@ -1123,6 +1123,166 @@ void flb_test_input_chunk_prefers_deletable_files_on_limit(void) flb_free(root_path); } +void flb_test_input_chunk_limit_with_many_outputs(void) +{ + int i; + int records; + struct flb_input_instance *i_ins; + struct flb_output_instance *o_low; + struct flb_output_instance *o_filler; + struct flb_output_instance *o_high; + struct mk_list *head; + struct mk_list *tmp; + struct flb_input_chunk *ic; + struct flb_task *task; + struct flb_config *cfg; + struct cio_ctx *cio; + struct mk_event_loop *evl; + struct cio_options opts = {0}; + char *root_path; + char temp_path[128]; + char buf[2048]; + size_t first_chunk_size; + + snprintf(temp_path, sizeof(temp_path) - 1, + "/input-chunk-many-outputs-%i/", getpid()); + temp_path[sizeof(temp_path) - 1] = '\0'; + + root_path = flb_test_tmpdir_cat(temp_path); + TEST_CHECK(root_path != NULL); + if (!root_path) { + return; + } + + memset(buf, 0x5A, sizeof(buf)); + + flb_init_env(); + cfg = flb_config_init(); + evl = mk_event_loop_create(256); + + TEST_CHECK(evl != NULL); + if (!evl) { + flb_config_exit(cfg); + flb_free(root_path); + return; + } + + cfg->evl = evl; + flb_log_create(cfg, FLB_LOG_STDERR, FLB_LOG_DEBUG, NULL); + + i_ins = flb_input_new(cfg, "dummy", NULL, FLB_TRUE); + TEST_CHECK(i_ins != NULL); + if (!i_ins) { + flb_config_exit(cfg); + flb_free(root_path); + return; + } + i_ins->storage_type = CIO_STORE_FS; + + cio_options_init(&opts); + opts.root_path = root_path; + opts.log_cb = log_cb; + opts.log_level = CIO_LOG_DEBUG; + opts.flags = CIO_OPEN; + + cio = cio_create(&opts); + TEST_CHECK(cio != NULL); + if (!cio) { + flb_input_exit_all(cfg); + flb_output_exit(cfg); + flb_config_exit(cfg); + flb_free(root_path); + return; + } + + flb_storage_input_create(cio, i_ins); + flb_input_init_all(cfg); + + o_low = flb_output_new(cfg, "http", NULL, FLB_TRUE); + TEST_CHECK(o_low != NULL); + if (!o_low) { + cio_destroy(cio); + flb_input_exit_all(cfg); + flb_output_exit(cfg); + flb_config_exit(cfg); + flb_free(root_path); + return; + } + flb_output_set_property(o_low, "match", "many.*"); + flb_output_set_property(o_low, "storage.total_limit_size", "10M"); + + for (i = 0; i < 31; i++) { + o_filler = flb_output_new(cfg, "http", NULL, FLB_TRUE); + TEST_CHECK(o_filler != NULL); + if (!o_filler) { + cio_destroy(cio); + flb_input_exit_all(cfg); + flb_output_exit(cfg); + flb_config_exit(cfg); + flb_free(root_path); + return; + } + flb_output_set_property(o_filler, "match", "filler.*"); + } + + o_high = flb_output_new(cfg, "http", NULL, FLB_TRUE); + TEST_CHECK(o_high != NULL); + if (!o_high) { + cio_destroy(cio); + flb_input_exit_all(cfg); + flb_output_exit(cfg); + flb_config_exit(cfg); + flb_free(root_path); + return; + } + TEST_CHECK(o_low->id == 0); + TEST_CHECK(o_high->id == 32); + flb_output_set_property(o_high, "match", "many.*"); + flb_output_set_property(o_high, "storage.total_limit_size", "10M"); + + TEST_CHECK(flb_routes_mask_set_size(mk_list_size(&cfg->outputs), + cfg->router) == 0); + TEST_CHECK_(flb_router_io_set(cfg) != -1, "unable to router"); + + records = flb_mp_count(buf, sizeof(buf)); + + TEST_CHECK(flb_input_chunk_append_raw(i_ins, FLB_INPUT_LOGS, + records, "many.one", 8, + buf, sizeof(buf)) == 0); + ic = mk_list_entry_last(&i_ins->chunks, struct flb_input_chunk, _head); + first_chunk_size = flb_input_chunk_get_real_size(ic); + o_high->total_limit_size = first_chunk_size + (first_chunk_size / 2); + + TEST_CHECK(flb_input_chunk_append_raw(i_ins, FLB_INPUT_LOGS, + records, "many.two", 8, + buf, sizeof(buf)) == 0); + + TEST_CHECK(mk_list_size(&i_ins->chunks) == 2); + ic = mk_list_entry_first(&i_ins->chunks, struct flb_input_chunk, _head); + TEST_CHECK(flb_routes_mask_get_bit(ic->routes_mask, + o_low->id, cfg->router) == 1); + TEST_CHECK(flb_routes_mask_get_bit(ic->routes_mask, + o_high->id, cfg->router) == 0); + TEST_CHECK(o_high->fs_chunks_size <= o_high->total_limit_size); + + mk_list_foreach_safe(head, tmp, &i_ins->tasks) { + task = mk_list_entry(head, struct flb_task, _head); + flb_task_destroy(task, FLB_TRUE); + } + + mk_list_foreach_safe(head, tmp, &i_ins->chunks) { + ic = mk_list_entry(head, struct flb_input_chunk, _head); + flb_input_chunk_destroy(ic, FLB_TRUE); + } + + cio_destroy(cio); + flb_router_exit(cfg); + flb_input_exit_all(cfg); + flb_output_exit(cfg); + flb_config_exit(cfg); + flb_free(root_path); +} + /* Test list */ TEST_LIST = { @@ -1136,5 +1296,7 @@ TEST_LIST = { flb_test_input_chunk_grouped_release_space_drop_counters}, {"input_chunk_prefers_deletable_files_on_limit", flb_test_input_chunk_prefers_deletable_files_on_limit}, + {"input_chunk_limit_with_many_outputs", + flb_test_input_chunk_limit_with_many_outputs}, {NULL, NULL} }; From 3355097b9872b9152e2a6a4d8676244c13c8790c Mon Sep 17 00:00:00 2001 From: Michael Renner Date: Wed, 5 Aug 2026 15:03:33 +0200 Subject: [PATCH 3/3] tests: internal: validate storage limit chunk size Keep the real chunk size signed until its error result is checked. Avoid deriving the test limit from a failed size lookup. Signed-off-by: Michael Renner --- tests/internal/input_chunk.c | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/tests/internal/input_chunk.c b/tests/internal/input_chunk.c index 1d75edfc9d2..0aa8b407453 100644 --- a/tests/internal/input_chunk.c +++ b/tests/internal/input_chunk.c @@ -1142,6 +1142,7 @@ void flb_test_input_chunk_limit_with_many_outputs(void) char *root_path; char temp_path[128]; char buf[2048]; + ssize_t chunk_real_size; size_t first_chunk_size; snprintf(temp_path, sizeof(temp_path) - 1, @@ -1250,7 +1251,12 @@ void flb_test_input_chunk_limit_with_many_outputs(void) records, "many.one", 8, buf, sizeof(buf)) == 0); ic = mk_list_entry_last(&i_ins->chunks, struct flb_input_chunk, _head); - first_chunk_size = flb_input_chunk_get_real_size(ic); + chunk_real_size = flb_input_chunk_get_real_size(ic); + TEST_CHECK(chunk_real_size > 0); + if (chunk_real_size <= 0) { + goto cleanup; + } + first_chunk_size = (size_t) chunk_real_size; o_high->total_limit_size = first_chunk_size + (first_chunk_size / 2); TEST_CHECK(flb_input_chunk_append_raw(i_ins, FLB_INPUT_LOGS, @@ -1265,6 +1271,7 @@ void flb_test_input_chunk_limit_with_many_outputs(void) o_high->id, cfg->router) == 0); TEST_CHECK(o_high->fs_chunks_size <= o_high->total_limit_size); +cleanup: mk_list_foreach_safe(head, tmp, &i_ins->tasks) { task = mk_list_entry(head, struct flb_task, _head); flb_task_destroy(task, FLB_TRUE);