From 2c6ca33b36049594e4634c7c90155852f5bf4a63 Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Wed, 28 Oct 2026 16:52:29 +0900 Subject: [PATCH 1/3] out_file: enforce destination limits across workers and tag filenames Signed-off-by: Hiroshi Hatake --- plugins/out_file/file.c | 106 ++++++++++++++++++++++++++-------------- 1 file changed, 70 insertions(+), 36 deletions(-) diff --git a/plugins/out_file/file.c b/plugins/out_file/file.c index 49394e00633..925c1b8faf9 100644 --- a/plugins/out_file/file.c +++ b/plugins/out_file/file.c @@ -24,6 +24,7 @@ #include #include #include +#include #include #include #include @@ -88,12 +89,14 @@ struct flb_file_conf { int missing_field_action; int limit_reached_action; int dynamic_destination; + int limit_tag_files; int enable_strftime; int time_path; int time_file; struct flb_record_accessor *ra_path; struct flb_record_accessor *ra_file; struct flb_hash_table *dynamic_files; + flb_lock_t destination_lock; struct flb_output_instance *ins; }; @@ -225,6 +228,7 @@ static void file_conf_destroy(struct flb_file_conf *ctx) } file_rotate_destroy(ctx->rotation); + flb_lock_destroy(&ctx->destination_lock); flb_free(ctx); } @@ -384,18 +388,57 @@ static int apply_destination_action(struct flb_file_conf *ctx, return -1; } +/* Reserve destinations before rotation or filesystem changes, including failed opens. */ +static int reserve_destination(struct flb_file_conf *ctx, + char *output, size_t output_size) +{ + int ret; + int limit_reached = FLB_FALSE; + void *stored_value; + size_t stored_size; + + if (ctx->dynamic_files == NULL) { + return 0; + } + + if (flb_lock_acquire(&ctx->destination_lock, FLB_LOCK_DEFAULT_RETRY_LIMIT, + FLB_LOCK_DEFAULT_RETRY_DELAY) != 0) { + flb_plg_error(ctx->ins, "could not acquire destination lock"); + return -1; + } + + ret = flb_hash_table_get(ctx->dynamic_files, output, strlen(output), + &stored_value, &stored_size); + if (ret < 0) { + if (ctx->dynamic_files->total_count >= ctx->max_dynamic_files) { + limit_reached = FLB_TRUE; + } + else { + ret = flb_hash_table_add(ctx->dynamic_files, output, strlen(output), "", 0); + } + } + + flb_lock_release(&ctx->destination_lock, FLB_LOCK_DEFAULT_RETRY_LIMIT, + FLB_LOCK_DEFAULT_RETRY_DELAY); + + if (limit_reached == FLB_TRUE) { + return apply_destination_action(ctx, ctx->limit_reached_action, + "max_dynamic_files limit reached", + output, output_size); + } + + return ret < 0 ? -1 : 0; +} + static int resolve_dynamic_destination(struct flb_file_conf *ctx, const char *tag, msgpack_object map, const struct flb_time *timestamp, char *output, - size_t output_size, - int *new_destination) + size_t output_size) { int ret; char sanitized_tag[PATH_MAX]; - void *stored_value; - size_t stored_size; flb_sds_t dynamic_path = NULL; flb_sds_t dynamic_file = NULL; flb_sds_t timestamp_path = NULL; @@ -407,7 +450,6 @@ static int resolve_dynamic_destination(struct flb_file_conf *ctx, const char *path; const char *file; - *new_destination = FLB_FALSE; path = ctx->out_path; file = ctx->out_file; @@ -522,24 +564,7 @@ static int resolve_dynamic_destination(struct flb_file_conf *ctx, return -1; } - if (ctx->dynamic_files == NULL) { - return 0; - } - - ret = flb_hash_table_get(ctx->dynamic_files, output, strlen(output), - &stored_value, &stored_size); - if (ret >= 0) { - return 0; - } - - if (ctx->dynamic_files->total_count >= ctx->max_dynamic_files) { - return apply_destination_action(ctx, ctx->limit_reached_action, - "max_dynamic_files limit reached", - output, output_size); - } - - *new_destination = FLB_TRUE; - return 0; + return reserve_destination(ctx, output, output_size); } @@ -562,6 +587,10 @@ static int cb_file_init(struct flb_output_instance *ins, return -1; } ctx->ins = ins; + if (flb_lock_init(&ctx->destination_lock) != 0) { + flb_free(ctx); + return -1; + } ctx->format = FLB_OUT_FILE_FMT_JSON; /* default */ ctx->delimiter = NULL; ctx->label_delimiter = NULL; @@ -574,6 +603,9 @@ static int cb_file_init(struct flb_output_instance *ins, return -1; } + /* Preserve existing tag behavior unless the limit was explicitly configured. */ + ctx->limit_tag_files = flb_output_get_property("max_dynamic_files", ins) != NULL; + ctx->missing_field_action = parse_dynamic_action(ctx, "on_missing_field", ctx->on_missing_field); ctx->limit_reached_action = parse_dynamic_action(ctx, "on_limit_reached", @@ -639,7 +671,9 @@ static int cb_file_init(struct flb_output_instance *ins, return -1; } - if (ctx->dynamic_destination == FLB_TRUE && ctx->max_dynamic_files > 0) { + if ((ctx->dynamic_destination == FLB_TRUE || + (ctx->limit_tag_files == FLB_TRUE && ctx->out_file == NULL)) && + ctx->max_dynamic_files > 0) { table_size = ctx->max_dynamic_files; if (table_size > 128) { table_size = 128; @@ -1136,7 +1170,6 @@ static int flush_dynamic_logs(struct flb_event_chunk *event_chunk, struct flb_log_event *log_event) { int ret; - int new_destination; FILE *fp; long file_pos; struct file_rotate_entry *rot_entry = NULL; @@ -1153,7 +1186,7 @@ static int flush_dynamic_logs(struct flb_event_chunk *event_chunk, log_event)) == FLB_EVENT_DECODER_SUCCESS) { ret = resolve_dynamic_destination(ctx, event_chunk->tag, *log_event->body, &log_event->timestamp, - output, output_size, &new_destination); + output, output_size); if (ret == 1) { continue; } @@ -1176,15 +1209,6 @@ static int flush_dynamic_logs(struct flb_event_chunk *event_chunk, return FLB_ERROR; } - if (new_destination == FLB_TRUE) { - ret = flb_hash_table_add(ctx->dynamic_files, output, strlen(output), "", 0); - if (ret < 0) { - close_output_file(fp, rot_entry); - flb_log_event_decoder_destroy(log_decoder); - return FLB_ERROR; - } - } - file_pos = ftell(fp); if (ctx->format == FLB_OUT_FILE_FMT_MSGPACK) { if (fwrite(log_decoder->record_base, 1, log_decoder->record_length, fp) != @@ -1284,6 +1308,14 @@ static void cb_file_flush(struct flb_event_chunk *event_chunk, } } + if (ctx->limit_tag_files == FLB_TRUE && ctx->dynamic_destination == FLB_FALSE && + ctx->out_file == NULL) { + ret = reserve_destination(ctx, out_file, sizeof(out_file)); + if (ret != 0) { + FLB_OUTPUT_RETURN(ret == 1 ? FLB_OK : FLB_ERROR); + } + } + if (file_rotate_acquire(ctx->rotation, out_file, &rot_entry) != 0) { FLB_OUTPUT_RETURN(FLB_ERROR); } @@ -1465,7 +1497,9 @@ static struct flb_config_map config_map[] = { { FLB_CONFIG_MAP_INT, "max_dynamic_files", "1024", 0, FLB_TRUE, offsetof(struct flb_file_conf, max_dynamic_files), - "Maximum number of distinct record-accessor destinations. Set to 0 for unlimited" + "Maximum number of distinct dynamic destinations. Also limits tag-derived " + "filenames when explicitly configured. Failed open attempts also count. " + "Set to 0 for unlimited" }, { From e88ca44219fb2c26aa6ed0cb4a93ecb32a00f30a Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Wed, 28 Oct 2026 16:52:39 +0900 Subject: [PATCH 2/3] tests: runtime: cover tag destination limits with rotation Signed-off-by: Hiroshi Hatake --- tests/runtime/out_file.c | 66 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 66 insertions(+) diff --git a/tests/runtime/out_file.c b/tests/runtime/out_file.c index 79bbe3ab3ae..2d46681955f 100644 --- a/tests/runtime/out_file.c +++ b/tests/runtime/out_file.c @@ -36,6 +36,7 @@ void flb_test_file_dynamic_unsafe_fallback(void); void flb_test_file_dynamic_windows_unsafe_fallback(void); void flb_test_file_dynamic_path_traversal_fallback(void); void flb_test_file_dynamic_limit_fallback(void); +void flb_test_file_tag_limit_fallback(void); /* Test list */ TEST_LIST = { @@ -52,6 +53,7 @@ TEST_LIST = { {"dynamic_windows_unsafe_fallback", flb_test_file_dynamic_windows_unsafe_fallback}, {"dynamic_path_traversal_fallback", flb_test_file_dynamic_path_traversal_fallback}, {"dynamic_limit_fallback", flb_test_file_dynamic_limit_fallback}, + {"tag_limit_fallback", flb_test_file_tag_limit_fallback}, {"template", flb_test_file_template}, {"delimiter_ltsv", flb_test_file_delim_ltsv}, {"delimiter_csv", flb_test_file_delim_csv}, @@ -1017,6 +1019,70 @@ void flb_test_file_dynamic_limit_fallback(void) flb_test_rmdir(TEST_LOGPATH); } +void flb_test_file_tag_limit_fallback(void) +{ + int ret; + int bytes; + int first_input; + int second_input; + int out_ffd; + char first[256]; + char second[256]; + char overflow[256]; + flb_ctx_t *ctx; + + snprintf(first, sizeof(first), "%s/first", TEST_LOGPATH); + snprintf(second, sizeof(second), "%s/second", TEST_LOGPATH); + snprintf(overflow, sizeof(overflow), "%s/overflow.log", TEST_LOGPATH); + remove(first); + remove(second); + remove(overflow); + flb_test_rmdir(TEST_LOGPATH); + + ctx = flb_create(); + flb_service_set(ctx, "Flush", "1", "Grace", "1", "Log_Level", "error", NULL); + + first_input = flb_input(ctx, (char *) "lib", NULL); + TEST_CHECK(first_input >= 0); + flb_input_set(ctx, first_input, "tag", "first", NULL); + second_input = flb_input(ctx, (char *) "lib", NULL); + TEST_CHECK(second_input >= 0); + flb_input_set(ctx, second_input, "tag", "second", NULL); + + out_ffd = flb_output(ctx, (char *) "file", NULL); + TEST_CHECK(out_ffd >= 0); + flb_output_set(ctx, out_ffd, "match", "*", NULL); + flb_output_set(ctx, out_ffd, "path", TEST_LOGPATH, NULL); + flb_output_set(ctx, out_ffd, "max_dynamic_files", "1", NULL); + flb_output_set(ctx, out_ffd, "on_limit_reached", "fallback", NULL); + flb_output_set(ctx, out_ffd, "fallback_path", TEST_LOGPATH, NULL); + flb_output_set(ctx, out_ffd, "fallback_file", "overflow.log", NULL); + flb_output_set(ctx, out_ffd, "mkdir", "true", NULL); + flb_output_set(ctx, out_ffd, "rotate", "true", NULL); + + ret = flb_start(ctx); + TEST_CHECK(ret == 0); + + bytes = flb_lib_push(ctx, first_input, JSON_SMALL, strlen(JSON_SMALL)); + TEST_CHECK(bytes == strlen(JSON_SMALL)); + ret = wait_for_file(first, 1, TEST_TIMEOUT); + TEST_CHECK(ret == 0); + + bytes = flb_lib_push(ctx, second_input, JSON_SMALL, strlen(JSON_SMALL)); + TEST_CHECK(bytes == strlen(JSON_SMALL)); + ret = wait_for_file(overflow, 1, TEST_TIMEOUT); + TEST_CHECK(ret == 0); + + flb_stop(ctx); + flb_destroy(ctx); + + TEST_CHECK(access(second, F_OK) != 0); + remove(first); + remove(second); + remove(overflow); + flb_test_rmdir(TEST_LOGPATH); +} + void flb_test_file_dynamic_path_traversal_fallback(void) { int ret; From 46742815d595c8c48814ddd0d929504583807205 Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Wed, 28 Oct 2026 16:52:46 +0900 Subject: [PATCH 3/3] tests: integration: verify file destination limits across workers Signed-off-by: Hiroshi Hatake --- .../out_file/tests/test_destination_limit.py | 118 ++++++++++++++++++ 1 file changed, 118 insertions(+) create mode 100644 tests/integration/scenarios/out_file/tests/test_destination_limit.py diff --git a/tests/integration/scenarios/out_file/tests/test_destination_limit.py b/tests/integration/scenarios/out_file/tests/test_destination_limit.py new file mode 100644 index 00000000000..b99616c2e83 --- /dev/null +++ b/tests/integration/scenarios/out_file/tests/test_destination_limit.py @@ -0,0 +1,118 @@ +"""Destination admission is bounded across workers, including failed opens.""" + +import json +from pathlib import Path +import time + +import pytest +import yaml + +from utils.fluent_bit_manager import FluentBitManager + + +DESTINATIONS = 40 +LIMIT = 8 + + +def create_file_output(tmp_path, mode, limit, action, failed_opens=False): + output_dir = tmp_path / "output" + output_dir.mkdir() + inputs = [] + for index in range(DESTINATIONS): + destination = f"dest{index}" + if failed_opens: + destination += "x" * 300 + elif mode == "tag": + destination += "/events" + inputs.append({ + "name": "dummy", "tag": destination, + "samples": 1 if failed_opens or action != "fallback" else 2, + "dummy": json.dumps({"id": index, "dest": destination}), + }) + output = { + "name": "file", "match": "*", "path": str(output_dir), + "format": "plain", "mkdir": True, "rotate": True, "workers": 4, + "on_limit_reached": action, "fallback_path": str(output_dir), + "fallback_file": "overflow.log", + } + if limit is not None: + output["max_dynamic_files"] = limit + if mode == "accessor": + output["file"] = "$dest" + config = tmp_path / "fluent-bit.yaml" + config.write_text(yaml.safe_dump({ + "service": {"flush": 0.2, "grace": 1, "http_server": True, + "http_port": "${FLUENT_BIT_HTTP_MONITORING_PORT}"}, + "pipeline": {"inputs": inputs, "outputs": [output]}, + })) + manager = FluentBitManager(str(config)) + return manager, output_dir + + +def wait_until(predicate, manager): + deadline = time.monotonic() + 30 + while time.monotonic() < deadline: + if predicate(): + return + time.sleep(0.1) + pytest.fail(Path(manager.log_file).read_text()) + + +def record_count(directory): + return sum(path.read_bytes().count(b"\n") for path in directory.rglob("*") if path.is_file()) + + +@pytest.mark.parametrize("mode,limit,action", [ + ("tag", LIMIT, "fallback"), + ("tag", LIMIT, "error"), + ("tag", LIMIT, "drop"), + ("tag", 0, "fallback"), + ("tag", None, "fallback"), + ("accessor", LIMIT, "fallback"), +]) +def test_destination_limit_workers(tmp_path, mode, limit, action): + manager, output_dir = create_file_output(tmp_path, mode, limit, action) + capped = limit == LIMIT + destination_count = LIMIT if capped else DESTINATIONS + samples = 2 if action == "fallback" else 1 + expected_records = samples * (destination_count if action != "fallback" else DESTINATIONS) + try: + manager.start() + wait_until(lambda: record_count(output_dir) == expected_records, manager) + if capped and action != "fallback": + wait_until(lambda: Path(manager.log_file).read_text().count( + "max_dynamic_files limit reached") == DESTINATIONS - LIMIT, manager) + finally: + manager.stop() + + files = [path for path in output_dir.rglob("*") if path.is_file()] + destinations = [path for path in files if path.name != "overflow.log"] + assert len(destinations) == destination_count + # Reusing admitted destinations remains possible after the cap is reached. + assert all(len(path.read_text().splitlines()) == samples for path in destinations) + if mode == "tag": + assert len(list(output_dir.iterdir())) == len(files) + overflow = output_dir / "overflow.log" + if capped and action == "fallback": + assert len(overflow.read_text().splitlines()) == 2 * (DESTINATIONS - LIMIT) + else: + assert not overflow.exists() + if action == "fallback": + ids = sorted(json.loads(line)["id"] for path in files for line in path.read_text().splitlines()) + assert ids == sorted(list(range(DESTINATIONS)) * 2) + + +@pytest.mark.parametrize("mode", ["tag", "accessor"]) +def test_failed_opens_consume_destination_slots(tmp_path, mode): + manager, output_dir = create_file_output(tmp_path, mode, LIMIT, "error", failed_opens=True) + try: + manager.start() + wait_until(lambda: Path(manager.log_file).read_text().count("error opening:") + + Path(manager.log_file).read_text().count("max_dynamic_files limit reached") + == DESTINATIONS, manager) + finally: + manager.stop() + log = Path(manager.log_file).read_text() + assert log.count("error opening:") == LIMIT + assert log.count("max_dynamic_files limit reached") == DESTINATIONS - LIMIT + assert not list(output_dir.iterdir())