Skip to content
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

in_tail: Manage tail watchers that are rorate_wait state too #4334

Merged
merged 6 commits into from
Mar 8, 2024
Merged

in_tail: Manage tail watchers that are rorate_wait state too #4334

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
ab81c16
in_tail: Manage tail watchers under `rorate_wait` state
ashie Oct 31, 2023
441b38b
in_tail: Add test for fast rotation
daipom Jan 25, 2024
341d8b8
in_tail: Make deletion be done on the timer callback side
daipom Jan 31, 2024
b8315c7
in_tail: Make closing io handler be done in closing phase
daipom Mar 8, 2024
13dd299
in_tail: Remove old comments
daipom Jan 31, 2024
d28f32e
in_tail: Make the new tests stable
daipom Mar 7, 2024
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
16 changes: 14 additions & 2 deletions lib/fluent/plugin/in_tail.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,7 @@ def initialize
super
@paths = []
@tails = {}
@tails_rotate_wait = {}
@pf_file = nil
@pf = nil
@ignore_list = []
Expand Down Expand Up @@ -267,6 +268,9 @@ def shutdown
@shutdown_start_time = Fluent::Clock.now
# during shutdown phase, don't close io. It should be done in close after all threads are stopped. See close.
stop_watchers(existence_path, immediate: true, remove_watcher: false)
@tails_rotate_wait.keys.each do |tw|
detach_watcher(tw, @tails_rotate_wait[tw][:ino])
end
daipom marked this conversation as resolved.
Show resolved Hide resolved
@pf_file.close if @pf_file

super
Expand Down Expand Up @@ -590,6 +594,8 @@ def detach_watcher(tw, ino, close_io = true)
target_info = TargetInfo.new(tw.path, ino)
@pf.unwatch(target_info)
end

@tails_rotate_wait.delete(tw)
daipom marked this conversation as resolved.
Show resolved Hide resolved
end

def throttling_is_enabled?(tw)
Expand All @@ -604,7 +610,11 @@ 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)
elsif throttling_is_enabled?(tw)
end

return if @tails_rotate_wait[tw]

if throttling_is_enabled?(tw)
# When the throttling feature is enabled, it might not reach EOF yet.
# Should ensure to read all contents before closing it, with keeping throttling.
start_time_to_wait = Fluent::Clock.now
Expand All @@ -615,11 +625,13 @@ def detach_watcher_after_rotate_wait(tw, ino)
detach_watcher(tw, ino)
end
end
@tails_rotate_wait[tw] = { ino: ino, timer: timer }
else
# when the throttling feature isn't enabled, just wait @rotate_wait
timer_execute(:in_tail_close_watcher, @rotate_wait, repeat: false) do
timer = timer_execute(:in_tail_close_watcher, @rotate_wait, repeat: false) do
detach_watcher(tw, ino)
end
@tails_rotate_wait[tw] = { ino: ino, timer: timer }
daipom marked this conversation as resolved.
Show resolved Hide resolved
end
end

Expand Down
3 changes: 0 additions & 3 deletions test/plugin/test_in_tail.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3084,9 +3084,6 @@ def test_refreshTW_during_rotation
sleep 3

Fluent::FileWrapper.open("#{@tmp_dir}/tail.txt0", "ab") {|f| f.puts "file3 log2"}

# Wait `rotate_wait` for file2 to make sure to close all IO handlers
sleep 3
end

inode_0 = tail_watchers[0]&.ino
Expand Down