diff --git a/lib/fluent/plugin/buffer.rb b/lib/fluent/plugin/buffer.rb index ea50eb3ecb..2df85b57af 100644 --- a/lib/fluent/plugin/buffer.rb +++ b/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. # diff --git a/test/plugin/test_buffer.rb b/test/plugin/test_buffer.rb index 9cb803f2f7..7dc959e41c 100644 --- a/test/plugin/test_buffer.rb +++ b/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