From dddcfad9f3394e1c55852e1d43524ba7e2f06220 Mon Sep 17 00:00:00 2001 From: Takuro Ashie Date: Mon, 16 Aug 2021 12:56:38 +0900 Subject: [PATCH 1/6] in_tail: Don't use TargetInfo as hash table key Signed-off-by: Takuro Ashie --- lib/fluent/plugin/in_tail.rb | 31 ++++++------- lib/fluent/plugin/in_tail/position_file.rb | 16 +------ test/plugin/in_tail/test_position_file.rb | 54 ---------------------- test/plugin/test_in_tail.rb | 21 ++++----- 4 files changed, 24 insertions(+), 98 deletions(-) diff --git a/lib/fluent/plugin/in_tail.rb b/lib/fluent/plugin/in_tail.rb index da9bba81c9..c6622751a5 100644 --- a/lib/fluent/plugin/in_tail.rb +++ b/lib/fluent/plugin/in_tail.rb @@ -357,11 +357,11 @@ def expand_paths def existence_path hash = {} - @tails.each_key {|target_info| + @tails.each {|path, tw| if @follow_inodes - hash[target_info.ino] = target_info + hash[tw.ino] = TargetInfo.new(tw.path, tw.ino) else - hash[target_info.path] = target_info + hash[tw.path] = TargetInfo.new(tw.path, tw.ino) end } hash @@ -446,8 +446,7 @@ def construct_watcher(target_info) begin target_info = TargetInfo.new(target_info.path, Fluent::FileWrapper.stat(target_info.path).ino) - @tails.delete(target_info) - @tails[target_info] = tw + @tails[target_info.path] = tw tw.on_notify rescue Errno::ENOENT, Errno::EACCES => e $log.warn "stat() for #{target_info.path} failed with #{e.class.name}. Drop tail watcher for now." @@ -469,9 +468,9 @@ def stop_watchers(targets_info, immediate: false, unwatched: false, remove_watch remove_path_from_group_watcher(target_info.path) if remove_watcher - tw = @tails.delete(target_info) + tw = @tails.delete(target_info.path) else - tw = @tails[target_info] + tw = @tails[target_info.path] end if tw tw.unwatched = unwatched @@ -485,8 +484,8 @@ def stop_watchers(targets_info, immediate: false, unwatched: false, remove_watch end def close_watcher_handles - @tails.keys.each do |target_info| - tw = @tails.delete(target_info) + @tails.keys.each do |path| + tw = @tails.delete(path) if tw tw.close end @@ -507,7 +506,7 @@ def update_watcher(target_info, pe) end rotated_target_info = TargetInfo.new(target_info.path, pe.read_inode) - rotated_tw = @tails[rotated_target_info] + rotated_tw = @tails[rotated_target_info.path] new_target_info = target_info.dup if @follow_inodes @@ -517,16 +516,12 @@ def update_watcher(target_info, pe) # When follow_inodes is true, it's not cleaned up by refresh_watcher. # So it should be unwatched here explicitly. rotated_tw.unwatched = true - # Make sure to delete old key, it has a different ino while the hash key is same. - @tails.delete(rotated_target_info) - @tails[new_target_info] = setup_watcher(new_target_info, new_position_entry) - @tails[new_target_info].on_notify + @tails[new_target_info.path] = setup_watcher(new_target_info, new_position_entry) + @tails[new_target_info.path].on_notify end else - # Make sure to delete old key, it has a different ino while the hash key is same. - @tails.delete(rotated_target_info) - @tails[new_target_info] = setup_watcher(new_target_info, pe) - @tails[new_target_info].on_notify + @tails[new_target_info.path] = setup_watcher(new_target_info, pe) + @tails[new_target_info.path].on_notify end detach_watcher_after_rotate_wait(rotated_tw, pe.read_inode) if rotated_tw end diff --git a/lib/fluent/plugin/in_tail/position_file.rb b/lib/fluent/plugin/in_tail/position_file.rb index 340de6cd88..ad1cb3ef11 100644 --- a/lib/fluent/plugin/in_tail/position_file.rb +++ b/lib/fluent/plugin/in_tail/position_file.rb @@ -250,20 +250,6 @@ def read_inode end end - TargetInfo = Struct.new(:path, :ino) do - def ==(other) - return false unless other.is_a?(TargetInfo) - self.path == other.path - end - - def hash - self.path.hash - end - - def eql?(other) - return false unless other.is_a?(TargetInfo) - self.path == other.path - end - end + TargetInfo = Struct.new(:path, :ino) end end diff --git a/test/plugin/in_tail/test_position_file.rb b/test/plugin/in_tail/test_position_file.rb index 9ccb772627..22ce4629f8 100644 --- a/test/plugin/in_tail/test_position_file.rb +++ b/test/plugin/in_tail/test_position_file.rb @@ -313,58 +313,4 @@ def build_files(file) assert_equal 2, f.read_inode end end - - sub_test_case "TargetInfo equality rules" do - sub_test_case "== operator" do - def test_equal - t1 = Fluent::Plugin::TailInput::TargetInfo.new("test", 1234) - t2 = Fluent::Plugin::TailInput::TargetInfo.new("test", 1235) - - assert_equal t1, t2 - end - - def test_not_equal - t1 = Fluent::Plugin::TailInput::TargetInfo.new("test", 1234) - t2 = Fluent::Plugin::TailInput::TargetInfo.new("test2", 1234) - - assert_not_equal t1, t2 - end - end - - sub_test_case "eql? method" do - def test_eql? - t1 = Fluent::Plugin::TailInput::TargetInfo.new("test", 1234) - t2 = Fluent::Plugin::TailInput::TargetInfo.new("test", 5321) - - assert do - t1.eql? t2 - end - end - - def test_not_eql? - t1 = Fluent::Plugin::TailInput::TargetInfo.new("test2", 1234) - t2 = Fluent::Plugin::TailInput::TargetInfo.new("test3", 1234) - - assert do - !t1.eql? t2 - end - end - end - - sub_test_case "hash" do - def test_equal - t1 = Fluent::Plugin::TailInput::TargetInfo.new("test", 1234) - t2 = Fluent::Plugin::TailInput::TargetInfo.new("test", 7321) - - assert_equal t1.hash, t2.hash - end - - def test_not_equal - t1 = Fluent::Plugin::TailInput::TargetInfo.new("test", 1234) - t2 = Fluent::Plugin::TailInput::TargetInfo.new("test2", 1234) - - assert_not_equal t1.hash, t2.hash - end - end - end end diff --git a/test/plugin/test_in_tail.rb b/test/plugin/test_in_tail.rb index 6433ae970a..9657440015 100644 --- a/test/plugin/test_in_tail.rb +++ b/test/plugin/test_in_tail.rb @@ -1777,8 +1777,7 @@ def test_z_refresh_watchers end path = 'test/plugin/data/2010/01/20100102-030405.log' - target_info = Fluent::Plugin::TailInput::TargetInfo.new(path, Fluent::FileWrapper.stat(path).ino) - mock.proxy(plugin).detach_watcher_after_rotate_wait(plugin.instance_variable_get(:@tails)[target_info], target_info.ino) + mock.proxy(plugin).detach_watcher_after_rotate_wait(plugin.instance_variable_get(:@tails)[path], Fluent::FileWrapper.stat(path).ino) Timecop.freeze(2010, 1, 2, 3, 4, 6) do path = "test/plugin/data/2010/01/20100102-030406.log" @@ -2179,13 +2178,13 @@ def test_should_close_watcher_after_rotate_wait target_info = create_target_info("#{TMP_DIR}/tail.txt") mock.proxy(Fluent::Plugin::TailInput::TailWatcher).new(target_info, anything, anything, true, true, anything, nil, anything, anything).once d.run(shutdown: false) - assert d.instance.instance_variable_get(:@tails)[target_info] + assert d.instance.instance_variable_get(:@tails)[target_info.path] Timecop.travel(now + 10) do d.instance.instance_eval do - sleep 0.1 until @tails[target_info] == nil + sleep 0.1 until @tails[target_info.path] == nil end - assert_nil d.instance.instance_variable_get(:@tails)[target_info] + assert_nil d.instance.instance_variable_get(:@tails)[target_info.path] end d.instance_shutdown end @@ -2216,8 +2215,8 @@ def test_should_create_new_watcher_for_new_file_with_same_name Timecop.travel(now + 10) do sleep 3 d.instance.instance_eval do - @tails[path_ino] == nil - @tails[new_path_ino] != nil + @tails[path_ino.path] == nil + @tails[new_path_ino.path] != nil end end @@ -2276,8 +2275,8 @@ def test_should_replace_target_info while d.events.size < 1 do sleep 0.1 end - inodes = d.instance.instance_variable_get(:@tails).keys.collect do |key| - key.ino + inodes = d.instance.instance_variable_get(:@tails).values.collect do |tw| + tw.ino end assert_equal([target_info.ino], inodes) @@ -2287,8 +2286,8 @@ def test_should_replace_target_info while d.events.size < 2 do sleep 0.1 end - inodes = d.instance.instance_variable_get(:@tails).keys.collect do |key| - key.ino + inodes = d.instance.instance_variable_get(:@tails).values.collect do |tw| + tw.ino end new_target_info = create_target_info("#{TMP_DIR}/tail.txt") assert_not_equal(target_info.ino, new_target_info.ino) From 548a42efe02d72972d47efc45bc0468e765084d4 Mon Sep 17 00:00:00 2001 From: Takuro Ashie Date: Wed, 25 Aug 2021 17:12:41 +0900 Subject: [PATCH 2/6] in_tail: Simplify using target_info.path Signed-off-by: Takuro Ashie --- lib/fluent/plugin/in_tail.rb | 31 ++++++++++++++++--------------- 1 file changed, 16 insertions(+), 15 deletions(-) diff --git a/lib/fluent/plugin/in_tail.rb b/lib/fluent/plugin/in_tail.rb index c6622751a5..e097c22214 100644 --- a/lib/fluent/plugin/in_tail.rb +++ b/lib/fluent/plugin/in_tail.rb @@ -425,14 +425,15 @@ def setup_watcher(target_info, pe) end def construct_watcher(target_info) + path = target_info.path pe = nil if @pf pe = @pf[target_info] if @read_from_head && pe.read_inode.zero? begin - pe.update(Fluent::FileWrapper.stat(target_info.path).ino, 0) + pe.update(Fluent::FileWrapper.stat(path).ino, 0) rescue Errno::ENOENT, Errno::EACCES - $log.warn "stat() for #{target_info.path} failed. Continuing without tailing it." + $log.warn "stat() for #{path} failed. Continuing without tailing it." end end end @@ -440,16 +441,16 @@ def construct_watcher(target_info) begin tw = setup_watcher(target_info, pe) rescue WatcherSetupError => e - log.warn "Skip #{target_info.path} because unexpected setup error happens: #{e}" + log.warn "Skip #{path} because unexpected setup error happens: #{e}" return end begin - target_info = TargetInfo.new(target_info.path, Fluent::FileWrapper.stat(target_info.path).ino) - @tails[target_info.path] = tw + target_info = TargetInfo.new(path, Fluent::FileWrapper.stat(path).ino) + @tails[path] = tw tw.on_notify rescue Errno::ENOENT, Errno::EACCES => e - $log.warn "stat() for #{target_info.path} failed with #{e.class.name}. Drop tail watcher for now." + $log.warn "stat() for #{path} failed with #{e.class.name}. Drop tail watcher for now." # explicitly detach and unwatch watcher `tw`. tw.unwatched = true detach_watcher(tw, target_info.ino, false) @@ -494,20 +495,20 @@ def close_watcher_handles # refresh_watchers calls @tails.keys so we don't use stop_watcher -> start_watcher sequence for safety. def update_watcher(target_info, pe) - log.info("detected rotation of #{target_info.path}; waiting #{@rotate_wait} seconds") + path = target_info.path + + log.info("detected rotation of #{path}; waiting #{@rotate_wait} seconds") if @pf pe_inode = pe.read_inode - target_info_from_position_entry = TargetInfo.new(target_info.path, pe_inode) + target_info_from_position_entry = TargetInfo.new(path, pe_inode) unless pe_inode == @pf[target_info_from_position_entry].read_inode log.debug "Skip update_watcher because watcher has been already updated by other inotify event" return end end - rotated_target_info = TargetInfo.new(target_info.path, pe.read_inode) - rotated_tw = @tails[rotated_target_info.path] - new_target_info = target_info.dup + rotated_tw = @tails[path] if @follow_inodes new_position_entry = @pf[target_info] @@ -516,12 +517,12 @@ def update_watcher(target_info, pe) # When follow_inodes is true, it's not cleaned up by refresh_watcher. # So it should be unwatched here explicitly. rotated_tw.unwatched = true - @tails[new_target_info.path] = setup_watcher(new_target_info, new_position_entry) - @tails[new_target_info.path].on_notify + @tails[path] = setup_watcher(target_info, new_position_entry) + @tails[path].on_notify end else - @tails[new_target_info.path] = setup_watcher(new_target_info, pe) - @tails[new_target_info.path].on_notify + @tails[path] = setup_watcher(target_info, pe) + @tails[path].on_notify end detach_watcher_after_rotate_wait(rotated_tw, pe.read_inode) if rotated_tw end From c1f348a7c4c88c172b6805c2794f4ef912f4e6a5 Mon Sep 17 00:00:00 2001 From: Takuro Ashie Date: Wed, 25 Aug 2021 17:54:56 +0900 Subject: [PATCH 3/6] in_tail: Simplify the error handling while contructing a watcher Signed-off-by: Takuro Ashie --- lib/fluent/plugin/in_tail.rb | 29 +++++++++++------------------ test/plugin/test_in_tail.rb | 18 ++++++++++-------- 2 files changed, 21 insertions(+), 26 deletions(-) diff --git a/lib/fluent/plugin/in_tail.rb b/lib/fluent/plugin/in_tail.rb index e097c22214..79b4a31b3c 100644 --- a/lib/fluent/plugin/in_tail.rb +++ b/lib/fluent/plugin/in_tail.rb @@ -426,35 +426,28 @@ def setup_watcher(target_info, pe) def construct_watcher(target_info) path = target_info.path + + begin + ino = Fluent::FileWrapper.stat(path).ino + rescue Errno::ENOENT, Errno::EACCES + $log.warn "stat() for #{path} failed. Continuing without tailing it." + return + end + pe = nil if @pf pe = @pf[target_info] - if @read_from_head && pe.read_inode.zero? - begin - pe.update(Fluent::FileWrapper.stat(path).ino, 0) - rescue Errno::ENOENT, Errno::EACCES - $log.warn "stat() for #{path} failed. Continuing without tailing it." - end - end + pe.update(ino, 0) if @read_from_head && pe.read_inode.zero? end begin tw = setup_watcher(target_info, pe) + @tails[path] = tw + tw.on_notify rescue WatcherSetupError => e log.warn "Skip #{path} because unexpected setup error happens: #{e}" return end - - begin - target_info = TargetInfo.new(path, Fluent::FileWrapper.stat(path).ino) - @tails[path] = tw - tw.on_notify - rescue Errno::ENOENT, Errno::EACCES => e - $log.warn "stat() for #{path} failed with #{e.class.name}. Drop tail watcher for now." - # explicitly detach and unwatch watcher `tw`. - tw.unwatched = true - detach_watcher(tw, target_info.ino, false) - end end def start_watchers(targets_info) diff --git a/test/plugin/test_in_tail.rb b/test/plugin/test_in_tail.rb index 9657440015..5c2ace8446 100644 --- a/test/plugin/test_in_tail.rb +++ b/test/plugin/test_in_tail.rb @@ -2457,14 +2457,15 @@ def test_ENOENT_error_after_setup_watcher 'format' => 'none', }) d = create_driver(config) - mock.proxy(d.instance).setup_watcher(anything, anything) do |tw| + mock.proxy(d.instance).existence_path do |hash| cleanup_file(path) - tw - end + hash + end.at_least(1) assert_nothing_raised do d.run(shutdown: false) {} end - assert($log.out.logs.any?{|log| log.include?("stat() for #{path} failed with Errno::ENOENT. Drop tail watcher for now.\n") }) + assert($log.out.logs.any?{|log| log.include?("stat() for #{path} failed. Continuing without tailing it.\n") }, + $log.out.logs.join("\n")) ensure d.instance_shutdown if d && d.instance end @@ -2482,14 +2483,15 @@ def test_EACCES_error_after_setup_watcher 'format' => 'none', }) d = create_driver(config, false) - mock.proxy(d.instance).setup_watcher(anything, anything) do |tw| + mock.proxy(d.instance).existence_path do |hash| FileUtils.chmod(0000, "#{TMP_DIR}/noaccess") - tw - end + hash + end.at_least(1) assert_nothing_raised do d.run(shutdown: false) {} end - assert($log.out.logs.any?{|log| log.include?("stat() for #{path} failed with Errno::EACCES. Drop tail watcher for now.\n") }) + assert($log.out.logs.any?{|log| log.include?("stat() for #{path} failed. Continuing without tailing it.\n") }, + $log.out.logs.join("\n")) end ensure d.instance_shutdown if d && d.instance From 1a676351d3f3cbcd2e300ded70511c0d19076970 Mon Sep 17 00:00:00 2001 From: Takuro Ashie Date: Wed, 25 Aug 2021 18:29:39 +0900 Subject: [PATCH 4/6] in_tail: Keep consistency more carefully at construct_watcher Signed-off-by: Takuro Ashie --- lib/fluent/plugin/in_tail.rb | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/lib/fluent/plugin/in_tail.rb b/lib/fluent/plugin/in_tail.rb index 79b4a31b3c..ada3f43ff0 100644 --- a/lib/fluent/plugin/in_tail.rb +++ b/lib/fluent/plugin/in_tail.rb @@ -427,8 +427,10 @@ def setup_watcher(target_info, pe) def construct_watcher(target_info) path = target_info.path + # The file might be rotated or removed after collecting paths, so check inode again here. begin ino = Fluent::FileWrapper.stat(path).ino + target_info.ino = ino rescue Errno::ENOENT, Errno::EACCES $log.warn "stat() for #{path} failed. Continuing without tailing it." return @@ -442,12 +444,13 @@ def construct_watcher(target_info) begin tw = setup_watcher(target_info, pe) - @tails[path] = tw - tw.on_notify rescue WatcherSetupError => e log.warn "Skip #{path} because unexpected setup error happens: #{e}" return end + + @tails[path] = tw + tw.on_notify end def start_watchers(targets_info) From b5d7a2553c3fd9675515d621cc1333efe54f8a03 Mon Sep 17 00:00:00 2001 From: Takuro Ashie Date: Thu, 26 Aug 2021 11:03:55 +0900 Subject: [PATCH 5/6] in_tail: Fix a broken test on Windows Signed-off-by: Takuro Ashie --- test/plugin/test_in_tail.rb | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/test/plugin/test_in_tail.rb b/test/plugin/test_in_tail.rb index 5c2ace8446..f9b23580a7 100644 --- a/test/plugin/test_in_tail.rb +++ b/test/plugin/test_in_tail.rb @@ -2457,10 +2457,14 @@ def test_ENOENT_error_after_setup_watcher 'format' => 'none', }) d = create_driver(config) + file_deleted = false mock.proxy(d.instance).existence_path do |hash| - cleanup_file(path) + unless file_deleted + cleanup_file(path) + file_deleted = true + end hash - end.at_least(1) + end.twice assert_nothing_raised do d.run(shutdown: false) {} end @@ -2486,7 +2490,7 @@ def test_EACCES_error_after_setup_watcher mock.proxy(d.instance).existence_path do |hash| FileUtils.chmod(0000, "#{TMP_DIR}/noaccess") hash - end.at_least(1) + end.twice assert_nothing_raised do d.run(shutdown: false) {} end From bd4782d8de1b523d73f4352d1be357f7630d158c Mon Sep 17 00:00:00 2001 From: Takuro Ashie Date: Thu, 26 Aug 2021 11:56:18 +0900 Subject: [PATCH 6/6] in_tail: Remove a needless local variable Signed-off-by: Takuro Ashie --- lib/fluent/plugin/in_tail.rb | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/lib/fluent/plugin/in_tail.rb b/lib/fluent/plugin/in_tail.rb index ada3f43ff0..b31d88e7a2 100644 --- a/lib/fluent/plugin/in_tail.rb +++ b/lib/fluent/plugin/in_tail.rb @@ -429,8 +429,7 @@ def construct_watcher(target_info) # The file might be rotated or removed after collecting paths, so check inode again here. begin - ino = Fluent::FileWrapper.stat(path).ino - target_info.ino = ino + target_info.ino = Fluent::FileWrapper.stat(path).ino rescue Errno::ENOENT, Errno::EACCES $log.warn "stat() for #{path} failed. Continuing without tailing it." return @@ -439,7 +438,7 @@ def construct_watcher(target_info) pe = nil if @pf pe = @pf[target_info] - pe.update(ino, 0) if @read_from_head && pe.read_inode.zero? + pe.update(target_info.ino, 0) if @read_from_head && pe.read_inode.zero? end begin