From 0b70260cdf92949a8fa3b07a4e2f9b7d759befbe Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Fri, 14 Aug 2026 17:59:07 +0900 Subject: [PATCH 1/5] input_chunk: Add evicted logs Signed-off-by: Hiroshi Hatake --- src/flb_input_chunk.c | 30 ++++++++++++++++++++++++++++++ 1 file changed, 30 insertions(+) diff --git a/src/flb_input_chunk.c b/src/flb_input_chunk.c index 2c11af562a3..35f3a9b896c 100644 --- a/src/flb_input_chunk.c +++ b/src/flb_input_chunk.c @@ -513,6 +513,7 @@ static int flb_input_chunk_release_space( ssize_t released_space; int chunk_released; ssize_t chunk_size; + int task_id; released_space = 0; @@ -554,6 +555,11 @@ static int flb_input_chunk_release_space( chunk_released = FLB_FALSE; chunk_destroy_flag = FLB_FALSE; + task_id = -1; + + if (old_input_chunk->task != NULL) { + task_id = old_input_chunk->task->id; + } if (flb_input_chunk_drop_task_route(old_input_chunk->task, output_plugin, @@ -580,6 +586,30 @@ static int flb_input_chunk_release_space( chunk_destroy_flag = FLB_TRUE; } + if (task_id >= 0) { + flb_warn("[input chunk] chunk '%s' evicted from output queue to make room " + "under storage.total_limit_size: task_id=%d, input=%s > output=%s " + "(out_id=%d), bytes=%zd, limit=%zu", + flb_input_chunk_get_name(old_input_chunk), + task_id, + flb_input_name(old_input_chunk->in), + flb_output_name(output_plugin), + output_plugin->id, + chunk_size, + output_plugin->total_limit_size); + } + else { + flb_warn("[input chunk] chunk '%s' evicted from output queue to make room " + "under storage.total_limit_size: input=%s > output=%s " + "(out_id=%d), bytes=%zd, limit=%zu", + flb_input_chunk_get_name(old_input_chunk), + flb_input_name(old_input_chunk->in), + flb_output_name(output_plugin), + output_plugin->id, + chunk_size, + output_plugin->total_limit_size); + } + #ifdef FLB_HAVE_METRICS if (dropped_record_count < 0) { dropped_record_count = get_input_chunk_record_count(old_input_chunk); From 056050f706f691aa598e86b9fd461d9344317e0c Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Fri, 14 Aug 2026 17:59:40 +0900 Subject: [PATCH 2/5] in_storage_backlog: Add an evicted log line Signed-off-by: Hiroshi Hatake --- plugins/in_storage_backlog/sb.c | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/plugins/in_storage_backlog/sb.c b/plugins/in_storage_backlog/sb.c index 5cdf1ade202..4dae0f18fb0 100644 --- a/plugins/in_storage_backlog/sb.c +++ b/plugins/in_storage_backlog/sb.c @@ -534,6 +534,16 @@ int sb_release_output_queue_space(struct flb_output_instance *output_plugin, released_space += chunk->size; underlying_chunk = chunk->chunk; + flb_warn("[storage backlog] chunk '%s' evicted from output queue to make room " + "under storage.total_limit_size: input=%s > output=%s " + "(out_id=%d), bytes=%zu, limit=%zu", + underlying_chunk->name, + chunk->stream->name, + flb_output_name(output_plugin), + output_plugin->id, + chunk->size, + output_plugin->total_limit_size); + sb_remove_chunk_from_segregated_backlogs(underlying_chunk, context); cio_chunk_close(underlying_chunk, FLB_TRUE); From 9e5997fe79fde40f235ebf88b41c49667504196d Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Fri, 14 Aug 2026 18:00:03 +0900 Subject: [PATCH 3/5] tests: integration: Add assertions for evicted logs Signed-off-by: Hiroshi Hatake --- .../in_forward/tests/test_in_forward_001.py | 23 +++++++++++++++++++ 1 file changed, 23 insertions(+) diff --git a/tests/integration/scenarios/in_forward/tests/test_in_forward_001.py b/tests/integration/scenarios/in_forward/tests/test_in_forward_001.py index f1ca1311d3b..3c1c442e417 100644 --- a/tests/integration/scenarios/in_forward/tests/test_in_forward_001.py +++ b/tests/integration/scenarios/in_forward/tests/test_in_forward_001.py @@ -3,6 +3,7 @@ import json import mmap import os +import re import shutil import socket import ssl @@ -2141,6 +2142,12 @@ def single_output_eviction_snapshot(): "forward storage eviction preference is not supported by this Fluent Bit binary" ) raise + + service.wait_for_log_contains( + "evicted from output queue to make room under " + "storage.total_limit_size: input=forward.0 > output=http.0", + timeout=10, + ) finally: service.stop() @@ -2263,9 +2270,25 @@ def shared_success_eviction_snapshot(): interval=0.2, description="shared-output stale route eviction snapshot", ) + + log_text = service.wait_for_log_contains( + "evicted from output queue to make room under " + "storage.total_limit_size: task_id=", + timeout=timeout, + ) finally: service.stop() + failed_chunks = set(re.findall( + r"failed to flush chunk '([^']+)'.*output=non_working_endpoint", + log_text, + )) + evicted_chunks = set(re.findall( + r"chunk '([^']+)' evicted from output queue.*output=non_working_endpoint", + log_text, + )) + assert not any(b"shared-one" in content for content in chunk_contents) assert any(b"shared-two" in content for content in chunk_contents) assert any(b"shared-three" in content for content in chunk_contents) + assert failed_chunks & evicted_chunks From f412132fabae436a2ff110e83c7bf9c7fcf77c18 Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Fri, 14 Aug 2026 18:17:05 +0900 Subject: [PATCH 4/5] input_chunk: Use the original plugin name Signed-off-by: Hiroshi Hatake --- src/flb_input_chunk.c | 19 +++++++++++++++++-- 1 file changed, 17 insertions(+), 2 deletions(-) diff --git a/src/flb_input_chunk.c b/src/flb_input_chunk.c index 35f3a9b896c..af5b7c62bc0 100644 --- a/src/flb_input_chunk.c +++ b/src/flb_input_chunk.c @@ -494,6 +494,21 @@ static ssize_t get_input_chunk_record_count(struct flb_input_chunk *input_chunk) return record_count; } +static const char *get_input_chunk_source_name(struct flb_input_chunk *input_chunk) +{ + struct cio_chunk *chunk; + + if (input_chunk->fs_backlog == FLB_TRUE && input_chunk->chunk != NULL) { + chunk = (struct cio_chunk *) input_chunk->chunk; + + if (chunk->st != NULL && chunk->st->name != NULL) { + return chunk->st->name; + } + } + + return flb_input_name(input_chunk->in); +} + static int flb_input_chunk_release_space( struct flb_input_chunk *new_input_chunk, struct flb_input_instance *input_plugin, @@ -592,7 +607,7 @@ static int flb_input_chunk_release_space( "(out_id=%d), bytes=%zd, limit=%zu", flb_input_chunk_get_name(old_input_chunk), task_id, - flb_input_name(old_input_chunk->in), + get_input_chunk_source_name(old_input_chunk), flb_output_name(output_plugin), output_plugin->id, chunk_size, @@ -603,7 +618,7 @@ static int flb_input_chunk_release_space( "under storage.total_limit_size: input=%s > output=%s " "(out_id=%d), bytes=%zd, limit=%zu", flb_input_chunk_get_name(old_input_chunk), - flb_input_name(old_input_chunk->in), + get_input_chunk_source_name(old_input_chunk), flb_output_name(output_plugin), output_plugin->id, chunk_size, From 099d039f73674552b8af1f717cfeed94b2a5004c Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Fri, 14 Aug 2026 18:17:43 +0900 Subject: [PATCH 5/5] tests: integration: Add a reloading case for chunk evictions Signed-off-by: Hiroshi Hatake --- .../reload_with_storage_limit_eviction.yaml | 29 ++++++++++++++ .../tests/test_in_storage_backlog_001.py | 38 +++++++++++++++++++ 2 files changed, 67 insertions(+) create mode 100644 tests/integration/scenarios/in_storage_backlog/config/reload_with_storage_limit_eviction.yaml diff --git a/tests/integration/scenarios/in_storage_backlog/config/reload_with_storage_limit_eviction.yaml b/tests/integration/scenarios/in_storage_backlog/config/reload_with_storage_limit_eviction.yaml new file mode 100644 index 00000000000..4a062cf9f5d --- /dev/null +++ b/tests/integration/scenarios/in_storage_backlog/config/reload_with_storage_limit_eviction.yaml @@ -0,0 +1,29 @@ +service: + flush: 60 + grace: 1 + log_level: info + http_server: on + http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT} + storage.path: ${STORAGE_BACKLOG_PATH} + +pipeline: + inputs: + - name: tail + tag: live.one + path: ${STORAGE_BACKLOG_LIVE_ONE_PATH} + refresh_interval: 1 + storage.type: filesystem + + - name: tail + tag: live.two + path: ${STORAGE_BACKLOG_LIVE_TWO_PATH} + refresh_interval: 1 + storage.type: filesystem + + outputs: + - name: http + match: "*" + host: 127.0.0.1 + port: 9 + retry_limit: no_limits + storage.total_limit_size: ${STORAGE_BACKLOG_TOTAL_LIMIT_SIZE} diff --git a/tests/integration/scenarios/in_storage_backlog/tests/test_in_storage_backlog_001.py b/tests/integration/scenarios/in_storage_backlog/tests/test_in_storage_backlog_001.py index 608585d79ab..cf3ec27d465 100644 --- a/tests/integration/scenarios/in_storage_backlog/tests/test_in_storage_backlog_001.py +++ b/tests/integration/scenarios/in_storage_backlog/tests/test_in_storage_backlog_001.py @@ -1,3 +1,4 @@ +import mmap import os from pathlib import Path import signal @@ -14,9 +15,14 @@ SEED_CONFIG = CONFIG_DIR / "seed_unroutable_chunk.yaml" RELOAD_WITHOUT_ROUTE = CONFIG_DIR / "reload_without_stale_route.yaml" RELOAD_WITH_ROUTE = CONFIG_DIR / "reload_with_stale_route.yaml" +RELOAD_WITH_EVICTION = CONFIG_DIR / "reload_with_storage_limit_eviction.yaml" KEEP_ON_DISK_MESSAGE = "no matching route for dummy.0/" REGISTER_MESSAGE = "register dummy.0/" QUEUE_MESSAGE = "queueing dummy.0:" +EVICTION_MESSAGE = ( + "evicted from output queue to make room under " + "storage.total_limit_size: input=dummy.0 > output=http.0" +) def wait_for_persisted_chunk(storage_path, process, log_path, timeout=10): @@ -91,9 +97,14 @@ def seeded_chunk(tmp_path, monkeypatch): storage_path.mkdir() seed_log_path = tmp_path / "seed.log" missing_path = tmp_path / "does-not-exist.log" + live_one_path = tmp_path / "live-one.log" + live_two_path = tmp_path / "live-two.log" monkeypatch.setenv("STORAGE_BACKLOG_PATH", str(storage_path)) monkeypatch.setenv("STORAGE_BACKLOG_MISSING_PATH", str(missing_path)) + monkeypatch.setenv("STORAGE_BACKLOG_LIVE_ONE_PATH", str(live_one_path)) + monkeypatch.setenv("STORAGE_BACKLOG_LIVE_TWO_PATH", str(live_two_path)) + monkeypatch.setenv("STORAGE_BACKLOG_TOTAL_LIMIT_SIZE", str(mmap.PAGESIZE * 2)) persisted_chunk = seed_persisted_chunk(storage_path, seed_log_path) assert persisted_chunk.exists() @@ -119,6 +130,33 @@ def test_routable_persisted_chunk_is_replayed_on_startup(seeded_chunk): assert not seeded_chunk.exists() +def test_loaded_backlog_eviction_reports_original_input(seeded_chunk): + manager = FluentBitManager(str(RELOAD_WITH_EVICTION)) + live_one_path = Path(os.environ["STORAGE_BACKLOG_LIVE_ONE_PATH"]) + live_two_path = Path(os.environ["STORAGE_BACKLOG_LIVE_TWO_PATH"]) + + try: + manager.start() + wait_for_log_occurrences( + Path(manager.log_file), QUEUE_MESSAGE, expected=1 + ) + + live_one_path.write_text("live one\n", encoding="utf-8") + live_two_path.write_text("live two\n", encoding="utf-8") + + wait_for_log_occurrences( + Path(manager.log_file), EVICTION_MESSAGE, expected=1 + ) + wait_for_path_removal(seeded_chunk) + + log = Path(manager.log_file).read_text(encoding="utf-8", errors="replace") + finally: + manager.stop() + + assert "input=storage_backlog." not in log + assert not seeded_chunk.exists() + + @pytest.mark.skipif(not hasattr(signal, "SIGHUP"), reason="SIGHUP is unavailable") def test_unroutable_chunk_stays_on_disk_across_hot_reload(seeded_chunk): manager = FluentBitManager(str(RELOAD_WITHOUT_ROUTE))