From 82f5f8a224d2c93cc7c350a551361f476ff1bdac Mon Sep 17 00:00:00 2001 From: Shizuo Fujita Date: Tue, 15 Sep 2026 17:37:06 +0900 Subject: [PATCH] in_tail: fix log duplication when a rotated file is appended during rotate_wait with follow_inodes With follow_inodes, a TailWatcher detached on rotation keeps reading its inode until rotate_wait elapses, but refresh_watchers only looked at @tails and started a second TailWatcher for the same inode. Skip inodes still read by a watcher waiting for rotate_wait, and do not register a watcher in @tails_rotate_wait when open_on_every_update already detached it immediately. Fixes https://github.com/fluent/fluentd/issues/4243 Signed-off-by: Shizuo Fujita --- lib/fluent/plugin/in_tail.rb | 18 +++--- test/plugin/test_in_tail.rb | 118 ++++++++++++++++++++++++++++++----- 2 files changed, 114 insertions(+), 22 deletions(-) diff --git a/lib/fluent/plugin/in_tail.rb b/lib/fluent/plugin/in_tail.rb index 3d32d6a057..1df8b7f978 100644 --- a/lib/fluent/plugin/in_tail.rb +++ b/lib/fluent/plugin/in_tail.rb @@ -454,6 +454,15 @@ def refresh_watchers removed_hash = existence_paths_hash.reject {|key, value| target_paths_hash.key?(key)} added_hash = target_paths_hash.reject {|key, value| existence_paths_hash.key?(key)} + if @follow_inodes + # A TailWatcher waiting for `rotate_wait` is not in `@tails` but still reads its inode. + # Starting another TailWatcher for that inode would collect the same logs twice. + @tails_rotate_wait.each_value do |v| + target = added_hash.delete(v[:ino]) + log.debug { "skip #{target.path} (inode: #{target.ino}) because a watcher waiting rotate_wait still reads it" } if target + end + end + # If an existing TailWatcher already follows a target path with the different inode, # it means that the TailWatcher following the rotated file still exists. In this case, # `refresh_watcher` can't start the new TailWatcher for the new current file. So, we @@ -591,14 +600,6 @@ def update_watcher(tail_watcher, pe, new_inode) if @follow_inodes && new_inode.nil? # nil inode means the file disappeared, so we only need to stop it. @tails.delete(tail_watcher.path) - # https://github.com/fluent/fluentd/pull/4237#issuecomment-1633358632 - # Because of this problem, log duplication can occur during `rotate_wait`. - # Need to set `rotate_wait 0` for a workaround. - # Duplication will occur if `refresh_watcher` is called during the `rotate_wait`. - # In that case, `refresh_watcher` will add the new TailWatcher to tail the same target, - # and it causes the log duplication. - # (Other `detach_watcher_after_rotate_wait` may have the same problem. - # We need the mechanism not to add duplicated TailWatcher with detaching TailWatcher.) detach_watcher_after_rotate_wait(tail_watcher, pe.read_inode) return end @@ -665,6 +666,7 @@ def detach_watcher_after_rotate_wait(tw, ino) if @open_on_every_update # Detach now because it's already closed, waiting it doesn't make sense. detach_watcher(tw, ino) + return end return if @tails_rotate_wait[tw] diff --git a/test/plugin/test_in_tail.rb b/test/plugin/test_in_tail.rb index 0819708eee..aa259a7b9a 100644 --- a/test/plugin/test_in_tail.rb +++ b/test/plugin/test_in_tail.rb @@ -2951,6 +2951,7 @@ def test_updateTW_before_refreshTW_and_detach_after_refreshTW Fluent::FileWrapper.open("#{@tmp_dir}/tail.txt", "wb") {|f| f.puts "file1 log1"} + tail_watcher_count_after_first_refresh = nil d.run(expect_records: 4, timeout: 10) do # Rotate (If the timing is bad, `TailWatcher::on_notify` might be called between mv and new-file-creation) Fluent::FileWrapper.open("#{@tmp_dir}/tail.txt", "ab") {|f| f.puts "file1 log2"} @@ -2963,11 +2964,16 @@ def test_updateTW_before_refreshTW_and_detach_after_refreshTW # This reproduces the following situation: # Rotation => update_watcher => refresh_watchers - # This adds a new TailWatcher: TailWatcher(path: "tail.txt1", inode: inode_0) + # This does NOT add TailWatcher(path: "tail.txt1", inode: inode_0) because the old + # TailWatcher waiting `rotate_wait` still reads inode_0. d.instance.refresh_watchers + tail_watcher_count_after_first_refresh = tail_watchers.size - # The old TailWatcher is detached here since `rotate_wait` is `4s`. - sleep 3 + # The old TailWatcher is detached and closed after `rotate_wait` (`4s`). + waiting(10) { sleep 0.1 until tail_watchers[0].instance_variable_get(:@io_handler).nil? } + + # This adds a new TailWatcher: TailWatcher(path: "tail.txt1", inode: inode_0) + d.instance.refresh_watchers # Append to the new current log file. Fluent::FileWrapper.open("#{@tmp_dir}/tail.txt", "ab") {|f| f.puts "file2 log2"} @@ -2987,6 +2993,7 @@ def test_updateTW_before_refreshTW_and_detach_after_refreshTW assert_equal( { record_values: ["file1 log1", "file1 log2", "file2 log1", "file2 log2"], + tail_watcher_count_after_first_refresh: 2, tail_watcher_paths: ["#{@tmp_dir}/tail.txt", "#{@tmp_dir}/tail.txt", "#{@tmp_dir}/tail.txt1"], tail_watcher_inodes: [inode_0, inode_1, inode_0], tail_watcher_io_handler_opened_statuses: [false, false, false], @@ -2998,6 +3005,7 @@ def test_updateTW_before_refreshTW_and_detach_after_refreshTW }, { record_values: record_values, + tail_watcher_count_after_first_refresh: tail_watcher_count_after_first_refresh, tail_watcher_paths: tail_watchers.collect { |tw| tw.path }, tail_watcher_inodes: tail_watchers.collect { |tw| tw.ino }, tail_watcher_io_handler_opened_statuses: tail_watchers.collect { |tw| tw.instance_variable_get(:@io_handler)&.opened? || false }, @@ -3111,10 +3119,8 @@ def test_path_resurrection # In order to reproduce the same condition stably, ensure that `refresh_watchers` is not # called by a timer. "refresh_interval" => "1h", - # https://github.com/fluent/fluentd/pull/4237#issuecomment-1633358632 - # Because of this problem, log duplication can occur during `rotate_wait`. - # Need to set `rotate_wait 0` for a workaround. - "rotate_wait" => "0s", + # In order to keep the first TailWatcher reading the rotated file until the end of the test. + "rotate_wait" => "10s", } ) d = create_driver(config, false) @@ -3143,9 +3149,10 @@ def test_path_resurrection sleep 2 # On Windows and macOS, StatWatcher doesn't work, so need enough interval for TimeTrigger. Fluent::FileWrapper.open("#{@tmp_dir}/tail.txt", "wb") {|f| f.puts "file2 log1"} - # Add new TailWatchers + # Add a new TailWatcher # tail.txt: TailWatcher(path: "tail.txt", inode: inode_1) - # tail.txt: TailWatcher(path: "tail.txt1", inode: inode_0) + # TailWatcher(path: "tail.txt1", inode: inode_0) is not added because the first TailWatcher + # waiting `rotate_wait` still reads inode_0. # NOTE: If not discarding the first TailWatcher on notify, this makes it a orphan because # this overwrites the `@tails[tail.txt]` by adding TailWatcher(path: "tail.txt", inode: inode_1) d.instance.refresh_watchers @@ -3189,11 +3196,6 @@ def test_path_resurrection inode: inode_1, io_handler_opened_status: false, }, - { - path: "#{@tmp_dir}/tail.txt1", - inode: inode_0, - io_handler_opened_status: false, - }, ], position_entries: [ ["#{@tmp_dir}/tail.txt", "0000000000000021", inode_0], @@ -3214,6 +3216,94 @@ def test_path_resurrection ) end + # https://github.com/fluent/fluentd/issues/4243 + def test_no_duplication_when_appending_to_rotated_file_during_rotate_wait + config = config_element( + "ROOT", + "", + { + "path" => "#{@tmp_dir}/tail.txt*", + "pos_file" => "#{@tmp_dir}/tail.pos", + "tag" => "t1", + "format" => "none", + "read_from_head" => "true", + "follow_inodes" => "true", + # In order to keep the old watcher alive while `refresh_watchers` runs. + "rotate_wait" => "4s", + # In order to reproduce the same condition stably, ensure that `refresh_watchers` is not + # called by a timer. + "refresh_interval" => "1h", + # stat_watcher often calls `TailWatcher::on_notify` faster than creating a new log file, + # so disable it in order to reproduce the same condition stably. + "enable_stat_watcher" => "false", + } + ) + d = create_driver(config, false) + + tail_watchers = [] + stub.proxy(d.instance).setup_watcher do |tw| + tail_watchers.append(tw) + tw + end + + Fluent::FileWrapper.open("#{@tmp_dir}/tail.txt", "wb") {|f| f.puts "file1 log1"} + + d.run(expect_records: 6, timeout: 15) do + # Rotate + Fluent::FileWrapper.open("#{@tmp_dir}/tail.txt", "ab") {|f| f.puts "file1 log2"} + FileUtils.move("#{@tmp_dir}/tail.txt", "#{@tmp_dir}/tail.txt" + "1") + Fluent::FileWrapper.open("#{@tmp_dir}/tail.txt", "wb") {|f| f.puts "file2 log1"} + + # `watch_timer` calls `TailWatcher::on_notify`, and then `update_watcher` updates the TailWatcher: + # TailWatcher(path: "tail.txt", inode: inode_0) => TailWatcher(path: "tail.txt", inode: inode_1) + # The old TailWatcher keeps reading inode_0 during `rotate_wait`. + sleep 2 + + # `refresh_watchers` must not add another TailWatcher for inode_0 while the old one is + # still reading it. Otherwise the following append is collected twice. + d.instance.refresh_watchers + Fluent::FileWrapper.open("#{@tmp_dir}/tail.txt1", "ab") {|f| f.puts "file1 log3"} + + # The old TailWatcher is detached and closed after `rotate_wait` (`4s`). + waiting(10) { sleep 0.1 until tail_watchers[0].instance_variable_get(:@io_handler).nil? } + + # Now the rotated file is followed again from the recorded position. + # TailWatcher(path: "tail.txt1", inode: inode_0) + d.instance.refresh_watchers + Fluent::FileWrapper.open("#{@tmp_dir}/tail.txt1", "ab") {|f| f.puts "file1 log4"} + Fluent::FileWrapper.open("#{@tmp_dir}/tail.txt", "ab") {|f| f.puts "file2 log2"} + end + + inode_0 = tail_watchers[0].ino + inode_1 = tail_watchers[1].ino + record_values = d.events.collect { |event| event[2]["message"] }.sort + position_entries = [] + Fluent::FileWrapper.open("#{@tmp_dir}/tail.pos", "r") do |f| + f.readlines(chomp: true).each do |line| + values = line.split("\t") + position_entries.append([values[0], values[1], values[2].to_i(16)]) + end + end + + assert_equal( + { + record_values: ["file1 log1", "file1 log2", "file1 log3", "file1 log4", "file2 log1", "file2 log2"], + tail_watcher_paths: ["#{@tmp_dir}/tail.txt", "#{@tmp_dir}/tail.txt", "#{@tmp_dir}/tail.txt1"], + tail_watcher_inodes: [inode_0, inode_1, inode_0], + position_entries: [ + ["#{@tmp_dir}/tail.txt", "000000000000002c", inode_0], + ["#{@tmp_dir}/tail.txt", "0000000000000016", inode_1], + ], + }, + { + record_values: record_values, + tail_watcher_paths: tail_watchers.collect { |tw| tw.path }, + tail_watcher_inodes: tail_watchers.collect { |tw| tw.ino }, + position_entries: position_entries + }, + ) + end + def test_next_rotation_occurs_very_fast_while_old_TW_still_waiting_rotate_wait config = config_element( "ROOT",