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); diff --git a/src/flb_input_chunk.c b/src/flb_input_chunk.c index 2c11af562a3..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, @@ -513,6 +528,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 +570,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 +601,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, + get_input_chunk_source_name(old_input_chunk), + 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), + get_input_chunk_source_name(old_input_chunk), + 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); 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 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))