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 #4243

Signed-off-by: Shizuo Fujita <fujita@clear-code.com>

Author: Watson1978

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]

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",