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

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

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