input_chunk: in_storage_backlog: Display evicted logs on dropping chunks

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);
 

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);

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

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}

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))