Skip to content

in_tail: fix log duplication when a rotated file is appended during rotate_wait with follow_inodes #5499

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
Watson1978 wants to merge 1 commit into
base: master
Choose a base branch
from
Draft

in_tail: fix log duplication when a rotated file is appended during rotate_wait with follow_inodes #5499

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 10 additions & 8 deletions lib/fluent/plugin/in_tail.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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]
Expand Down
118 changes: 104 additions & 14 deletions test/plugin/test_in_tail.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"}
Expand All @@ -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"}
Expand All @@ -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],
Expand All @@ -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 },
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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],
Expand All @@ -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",
Expand Down
Loading