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/lib/fluent/plugin/out_file.rb b/lib/fluent/plugin/out_file.rb index 13f6c695cf..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,9 +76,30 @@ 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) - @latest_metadata = metadata + @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 metadata @@ -93,11 +115,24 @@ def symlink_path=(path) def generate_chunk(metadata) chunk = super - # "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 + # 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 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) 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?