Add evacuate_limit_size to bound chunk evacuation

lib/fluent/plugin/buffer.rb

@@ -72,6 +72,9 @@ class BufferChunkOverflowError < BufferError; end # A record size is larger than
       desc 'If true, chunks are thrown away when unrecoverable error happens'
       config_param :disable_chunk_backup, :bool, default: false
 
+      desc 'The total size limit for chunks evacuated on unrecoverable errors. Once evacuation reaches this size, remaining chunks are purged without being evacuated. Set 0 to disable evacuation. nil (default) means no limit.'
+      config_param :evacuate_limit_size, :size, default: nil
+
       Metadata = Struct.new(:timekey, :tag, :variables, :seq) do
         def initialize(timekey, tag, variables)
           super(timekey, tag, variables, 0)
@@ -625,11 +628,15 @@ def clear_queue!
         log.on_trace { log.trace "clearing queue", instance: self.object_id }
 
         synchronize do
+          evacuated_size = 0
           until @queue.empty?
             begin
               q = @queue.shift
-              evacuate_chunk(q)
-              log.trace("purging a chunk in queue"){ {id: dump_unique_id_hex(chunk.unique_id), bytesize: chunk.bytesize, size: chunk.size} }
+              if evacuate_chunk?(q.bytesize, evacuated_size)
+                evacuate_chunk(q)
+                evacuated_size += q.bytesize
+              end
+              log.trace("purging a chunk in queue"){ {id: dump_unique_id_hex(q.unique_id), bytesize: q.bytesize, size: q.size} }
               q.purge
             rescue => e
               log.error "unexpected error while clearing buffer queue", error_class: e.class, error: e
@@ -640,6 +647,17 @@ def clear_queue!
         end
       end
 
+      # Decide whether a chunk should be evacuated, honoring evacuate_limit_size.
+      # nil limit means no bound (evacuate everything), 0 disables evacuation, and
+      # a positive limit only evacuates a chunk if it fits within the remaining budget
+      # so the total evacuated size never exceeds the limit.
+      def evacuate_chunk?(chunk_bytesize, evacuated_size)
+        return true if @evacuate_limit_size.nil?
+        return false if @evacuate_limit_size <= 0
+
+        evacuated_size + chunk_bytesize <= @evacuate_limit_size
+      end
+
       def evacuate_chunk(chunk)
         # Overwrite this on demand.
         #

test/plugin/test_buffer.rb

@@ -475,6 +475,51 @@ def create_chunk_es(metadata, es)
       assert{ qchunks.all?{ |c| c.purged } }
     end
 
+    test '#clear_queue! evacuates every queued chunk by default (no evacuate_limit_size)' do
+      evacuated = []
+      @p.define_singleton_method(:evacuate_chunk) { |chunk| evacuated << chunk.bytesize }
+      qchunks = @p.queue.dup
+
+      assert_nil @p.evacuate_limit_size
+
+      @p.clear_queue!
+
+      assert_equal [100, 100, 3], evacuated
+      assert_equal [], @p.queue
+      assert{ qchunks.all?{ |c| c.purged } }
+    end
+
+    test '#clear_queue! skips evacuation entirely when evacuate_limit_size is 0, but still purges' do
+      p = create_buffer({'evacuate_limit_size' => 0})
+      evacuated = []
+      p.define_singleton_method(:evacuate_chunk) { |chunk| evacuated << chunk.bytesize }
+      p.start
+      qchunks = p.queue.dup
+
+      p.clear_queue!
+
+      assert_equal [], evacuated
+      assert_equal [], p.queue
+      assert{ qchunks.all?{ |c| c.purged } }
+    end
+
+    test '#clear_queue! bounds the total evacuated size to evacuate_limit_size' do
+      p = create_buffer({'evacuate_limit_size' => 200})
+      evacuated = []
+      p.define_singleton_method(:evacuate_chunk) { |chunk| evacuated << chunk.bytesize }
+      p.start
+      qchunks = p.queue.dup
+
+      p.clear_queue!
+
+      # queued chunks are 100, 100 and 3 bytes; the first two fill the 200 byte
+      # budget, so the last one is purged without being evacuated
+      assert_equal [100, 100], evacuated
+      assert_equal 200, evacuated.sum
+      assert_equal [], p.queue
+      assert{ qchunks.all?{ |c| c.purged } }
+    end
+
     test '#write returns immediately if argument data is empty array' do
       assert_equal [@dm0,@dm1,@dm1], @p.queue.map(&:metadata)
       assert_equal [@dm2,@dm3], @p.stage.keys