From e6fbbd36a1c6513de8b07002c56417185846f484 Mon Sep 17 00:00:00 2001 From: Mehrdad Biukian Naeini Date: Fri, 21 Aug 2026 21:18:46 +0400 Subject: [PATCH 1/3] fix(buffer): decrement queue_size in ensure so failed purge does not leak (#5468) When chunk.purge raises (e.g. unlink/close failure on the buffer path), the rescue swallows the error but @queue_size_metrics.sub was skipped. The chunk is already gone from @dequeued, so it is never retried and the queued byte counter ratchets toward total_limit_size, making storable? permanently false and causing spurious BufferOverflowError on a near-empty buffer. Moving the sub into an ensure block keeps the counter correct regardless of purge success. Co-Authored-By: Mehrdad Biukian Signed-off-by: Mehrdad Biukian Naeini --- lib/fluent/plugin/buffer.rb | 10 ++++++++-- test/plugin/test_buffer.rb | 26 ++++++++++++++++++++++++++ 2 files changed, 34 insertions(+), 2 deletions(-) diff --git a/lib/fluent/plugin/buffer.rb b/lib/fluent/plugin/buffer.rb index ea50eb3ecb..243065be38 100644 --- a/lib/fluent/plugin/buffer.rb +++ b/lib/fluent/plugin/buffer.rb @@ -601,13 +601,19 @@ def purge_chunk(chunk_id) metadata = chunk.metadata log.on_trace { log.trace "purging a chunk", instance: self.object_id, chunk_id: dump_unique_id_hex(chunk_id), metadata: metadata } + bytesize = chunk.bytesize begin - bytesize = chunk.bytesize chunk.purge - @queue_size_metrics.sub(bytesize) rescue => e log.error "failed to purge buffer chunk", chunk_id: dump_unique_id_hex(chunk_id), error_class: e.class, error: e log.error_backtrace + ensure + # Always release the queued byte counter, even when purge raises + # (e.g. unlink/close failure on the buffer path). Otherwise + # @queue_size is never decremented for a dequeued chunk that is + # gone from both @queue and @dequeued, so the leak ratchets toward + # total_limit_size and storable? becomes permanently false (#5468). + @queue_size_metrics.sub(bytesize) end @dequeued_num[chunk.metadata] -= 1 diff --git a/test/plugin/test_buffer.rb b/test/plugin/test_buffer.rb index 9cb803f2f7..1318cdd4a6 100644 --- a/test/plugin/test_buffer.rb +++ b/test/plugin/test_buffer.rb @@ -435,6 +435,32 @@ def create_chunk_es(metadata, es) assert_equal({}, @p.dequeued) end + test '#purge_chunk releases queue_size even when chunk.purge raises (#5468)' do + # Simulate a physical purge failure (unlink/close on the buffer path) + # so chunk.purge raises after the chunk is dequeued. + failing_purge = Module.new do + def purge + raise IOError, 'simulated purge failure (unlink EIO)' + end + end + @p.buffer_class::Chunk.include(failing_purge) + + m1 = @p.dequeue_chunk + assert_equal [@dm0, @dm1, @dm1], @p.queue.map(&:metadata) + assert_equal({m1.unique_id => m1}, @p.dequeued) + + queued_before = @p.queue_size + assert queued_before > 0 + + # purge_chunk swallows the error but must still decrement queue_size + @p.purge_chunk(m1.unique_id) + + assert m1.purged + # queue_size must return to its pre-dequeue value, not leak upward + assert_equal queued_before - m1.bytesize, @p.queue_size + assert @p.storable? + end + test '#takeback_chunk returns false if specified chunk_id is already purged' do assert_equal [@dm0,@dm1,@dm1], @p.queue.map(&:metadata) assert_equal({}, @p.dequeued) From 5fc6fec23116f660c2daf5abced1e575bed6619a Mon Sep 17 00:00:00 2001 From: Mehrdad Biukian Naeini Date: Sat, 22 Aug 2026 22:58:21 +0400 Subject: [PATCH 2/3] fix(out_file): create symlink for every latest metadata sharing a timekey Bulk input whose events share a timekey but differ in a chunk-key variable (e.g. myid=1 and myid=2) stages one chunk per metadata, but SymlinkBufferMixin kept only a single \@latest_metadata. Each new metadata with an equal timekey overwrote the previous one, so only the chunk staged last ever received a symlink; earlier chunks were left without one until their timekey advanced. Track all metadatas of the newest timekey in a set and create the symlink for any chunk generated from one of them. Fixes #5099 Signed-off-by: Mehrdad Biukian Naeini --- lib/fluent/plugin/out_file.rb | 14 ++++++++++++-- test/plugin/test_out_file.rb | 32 ++++++++++++++++++++++++++++++++ 2 files changed, 44 insertions(+), 2 deletions(-) diff --git a/lib/fluent/plugin/out_file.rb b/lib/fluent/plugin/out_file.rb index 13f6c695cf..c50c5c684f 100644 --- a/lib/fluent/plugin/out_file.rb +++ b/lib/fluent/plugin/out_file.rb @@ -77,7 +77,16 @@ def metadata(timekey: nil, tag: nil, variables: nil) @latest_metadata ||= new_metadata(timekey: 0) if metadata.timekey && (metadata.timekey >= @latest_metadata.timekey) - @latest_metadata = metadata + if metadata.timekey == @latest_metadata.timekey + # Same timekey, different variables (e.g. bulk input with several + # chunk-key values): every one of these is "latest" for its own + # chunk and must get a symlink (fluentd#5099). + (@latest_metadatas ||= {})[metadata] = true + @latest_metadata = metadata + else + @latest_metadata = metadata + @latest_metadatas = { metadata => true } + end end metadata @@ -93,11 +102,12 @@ def symlink_path=(path) def generate_chunk(metadata) chunk = super + latest_set = @latest_metadatas || { @latest_metadata => true } # "symlink" feature is to link from symlink_path to the latest file chunk. Records with latest # timekey will be appended into that file chunk. On the other side, resumed file chunks might NOT # have timekey, especially in the cases that resumed file chunks are generated by Fluentd v0.12. # These chunks will be enqueued immediately, and will be flushed soon. - if chunk.metadata == @latest_metadata + if latest_set.key?(chunk.metadata) || chunk.metadata == @latest_metadata sym_path = @_output_plugin_for_symlink.extract_placeholders(@_symlink_path, chunk) FileUtils.mkdir_p(File.dirname(sym_path), mode: @_output_plugin_for_symlink.dir_perm) if @_output_plugin_for_symlink.symlink_path_use_relative diff --git a/test/plugin/test_out_file.rb b/test/plugin/test_out_file.rb index 79bbeec1d9..41d195fa44 100644 --- a/test/plugin/test_out_file.rb +++ b/test/plugin/test_out_file.rb @@ -779,6 +779,38 @@ def parse_system(text) end end + test 'symlink with placeholders and variable chunk keys creates a symlink per metadata (bulk input)' do + omit "Windows doesn't support symlink" if Fluent.windows? + conf = %[ + path #{TMP_DIR}/${tag}/out_file_${myid}_%Y%m%d.log + symlink_path #{SYMLINK_PATH}/foo/${tag}_${myid}.log + + + ] + + d = create_driver(conf) + begin + d.run(default_tag: 'tag') do + # Bulk input: two events with different `myid` values but the SAME + # timekey arrive in one stream. Each metadata must get its own + # staged chunk and its own symlink (fluentd#5099). + es = Fluent::MultiEventStream.new + t = event_time("2011-01-02 13:14:15 UTC") + es.add(t, {"a" => 1, "myid" => "1"}) + es.add(t, {"a" => 2, "myid" => "2"}) + d.feed(es) + end + + assert File.symlink?("#{SYMLINK_PATH}/foo/tag_1.log"), + "missing symlink for myid=1" + assert File.symlink?("#{SYMLINK_PATH}/foo/tag_2.log"), + "missing symlink for myid=2" + ensure + FileUtils.rm_f("#{SYMLINK_PATH}/foo/tag_1.log") + FileUtils.rm_f("#{SYMLINK_PATH}/foo/tag_2.log") + end + end + test 'relative symlink' do omit "Windows doesn't support symlinks" if Fluent.windows? From fed914e1da379f48cb9ccb7028fffc50acc11d41 Mon Sep 17 00:00:00 2001 From: Mehrdad Biukian Naeini Date: Wed, 26 Aug 2026 15:43:04 +0400 Subject: [PATCH 3/3] out_file: fix symlink tracking race + O(n^2) + unbounded growth - Add Mutex to protect @latest_metadata/@latest_metadatas mutations (called from Output#metadata, reachable concurrently from input threads). - Replace Metadata-object Hash keys (which collide because Metadata#hash only hashes timekey) with compound [timekey, tag, variables] keys that distinguish chunk-key variants sharing a timekey. - Evict the old window's tracking when timekey advances, preventing unbounded growth within a single timekey window. - Remove dead-code fallback (unreachable because @latest_metadata is always in @latest_metadatas when that Hash is non-nil, and the single-entry fallback reconstructs it inline). - All changes within SymlinkBufferMixin; no external API changes. Signed-off-by: Mehrdad Biukian Naeini --- lib/fluent/plugin/out_file.rb | 59 +++++++++++++++++++++++++---------- 1 file changed, 42 insertions(+), 17 deletions(-) diff --git a/lib/fluent/plugin/out_file.rb b/lib/fluent/plugin/out_file.rb index c50c5c684f..0254ad904f 100644 --- a/lib/fluent/plugin/out_file.rb +++ b/lib/fluent/plugin/out_file.rb @@ -18,6 +18,7 @@ require 'zlib' require 'time' require 'pathname' +require 'thread' require 'fluent/plugin/output' require 'fluent/config/error' @@ -75,17 +76,29 @@ module SymlinkBufferMixin def metadata(timekey: nil, tag: nil, variables: nil) metadata = super - @latest_metadata ||= new_metadata(timekey: 0) - if metadata.timekey && (metadata.timekey >= @latest_metadata.timekey) - if metadata.timekey == @latest_metadata.timekey - # Same timekey, different variables (e.g. bulk input with several - # chunk-key values): every one of these is "latest" for its own - # chunk and must get a symlink (fluentd#5099). - (@latest_metadatas ||= {})[metadata] = true - @latest_metadata = metadata - else - @latest_metadata = metadata - @latest_metadatas = { metadata => true } + @symlink_mutex ||= Mutex.new + @symlink_mutex.synchronize do + @latest_metadata ||= new_metadata(timekey: 0) + if metadata.timekey && (metadata.timekey >= @latest_metadata.timekey) + if metadata.timekey == @latest_metadata.timekey + # Same timekey, different variables (e.g. bulk input with several + # chunk-key values): every one of these is "latest" for its own + # chunk and must get a symlink (fluentd#5099). + # + # NOTE: Metadata#hash only hashes timekey (by design, for perf), so + # keying a Hash on Metadata objects makes every distinct metadata + # sharing a timekey collide into one bucket -> O(n^2). Use a + # compound key [timekey, tag, variables] that uniquely identifies + # the metadata's chunk-key identity instead. + key = [metadata.timekey, metadata.tag, metadata.variables] + (@latest_metadatas ||= {})[key] = metadata + @latest_metadata = metadata + else + # Timekey advanced: drop the previous window's tracking entirely + # so memory does not grow without bound within the new timekey. + @latest_metadata = metadata + @latest_metadatas = { [metadata.timekey, metadata.tag, metadata.variables] => metadata } + end end end @@ -102,12 +115,24 @@ def symlink_path=(path) def generate_chunk(metadata) chunk = super - latest_set = @latest_metadatas || { @latest_metadata => true } - # "symlink" feature is to link from symlink_path to the latest file chunk. Records with latest - # timekey will be appended into that file chunk. On the other side, resumed file chunks might NOT - # have timekey, especially in the cases that resumed file chunks are generated by Fluentd v0.12. - # These chunks will be enqueued immediately, and will be flushed soon. - if latest_set.key?(chunk.metadata) || chunk.metadata == @latest_metadata + # Snapshot the tracked set under the same lock used by #metadata so we + # never read a half-updated Hash (lost-update race flagged in review). + latest_set = nil + @symlink_mutex ||= Mutex.new + @symlink_mutex.synchronize do + latest_set = @latest_metadatas || { [@latest_metadata.timekey, @latest_metadata.tag, @latest_metadata.variables] => @latest_metadata } + end + # "symlink" feature links symlink_path to the latest file chunk. Records + # with the latest timekey are appended into that chunk. Resumed file + # chunks (e.g. from Fluentd v0.12) may have no timekey and are flushed + # immediately, so they are intentionally excluded here. + # + # The former `chunk.metadata == @latest_metadata` fallback was dead code: + # whenever @latest_metadatas is non-nil, @latest_metadata is always one + # of its values (assigned together above), and when it is nil the + # fallback builds the same single-entry set inline. + key = [chunk.metadata.timekey, chunk.metadata.tag, chunk.metadata.variables] + if latest_set.key?(key) sym_path = @_output_plugin_for_symlink.extract_placeholders(@_symlink_path, chunk) FileUtils.mkdir_p(File.dirname(sym_path), mode: @_output_plugin_for_symlink.dir_perm) if @_output_plugin_for_symlink.symlink_path_use_relative