From 9f402c1ecfafe1c3783a2bb383dc57d5ae625564 Mon Sep 17 00:00:00 2001 From: Anthony Comtois Date: Tue, 29 Oct 2019 10:06:49 +0000 Subject: [PATCH 1/5] Add log throttling per file Remove debug log Signed-off-by: Anthony Comtois --- lib/fluent/plugin/in_tail.rb | 18 ++++++++++++++---- test/plugin/test_in_tail.rb | 34 ++++++++++++++++++++++++++++++++-- 2 files changed, 46 insertions(+), 6 deletions(-) diff --git a/lib/fluent/plugin/in_tail.rb b/lib/fluent/plugin/in_tail.rb index 6a226a544d..1b645639a1 100644 --- a/lib/fluent/plugin/in_tail.rb +++ b/lib/fluent/plugin/in_tail.rb @@ -74,6 +74,8 @@ def initialize config_param :refresh_interval, :time, default: 60 desc 'The number of reading lines at each IO.' config_param :read_lines_limit, :integer, default: 1000 + desc 'The number of reading lines per notify' + config_param :read_lines_limit_per_notify, :integer, default: -1 desc 'The interval of flushing the buffer for multiline format' config_param :multiline_flush_interval, :time, default: nil desc 'Enable the option to emit unmatched lines.' @@ -283,7 +285,7 @@ def refresh_watchers def setup_watcher(path, pe) line_buffer_timer_flusher = (@multiline_mode && @multiline_flush_interval) ? TailWatcher::LineBufferTimerFlusher.new(log, @multiline_flush_interval, &method(:flush_buffer)) : nil - tw = TailWatcher.new(path, @rotate_wait, pe, log, @read_from_head, @enable_watch_timer, @enable_stat_watcher, @read_lines_limit, method(:update_watcher), line_buffer_timer_flusher, @from_encoding, @encoding, open_on_every_update, &method(:receive_lines)) + tw = TailWatcher.new(path, @rotate_wait, pe, log, @read_from_head, @enable_watch_timer, @enable_stat_watcher, @read_lines_limit, @read_lines_limit_per_notify, method(:update_watcher), line_buffer_timer_flusher, @from_encoding, @encoding, open_on_every_update, &method(:receive_lines)) tw.attach do |watcher| event_loop_attach(watcher.timer_trigger) if watcher.timer_trigger event_loop_attach(watcher.stat_trigger) if watcher.stat_trigger @@ -494,7 +496,7 @@ def parse_multilines(lines, tail_watcher) end class TailWatcher - def initialize(path, rotate_wait, pe, log, read_from_head, enable_watch_timer, enable_stat_watcher, read_lines_limit, update_watcher, line_buffer_timer_flusher, from_encoding, encoding, open_on_every_update, &receive_lines) + def initialize(path, rotate_wait, pe, log, read_from_head, enable_watch_timer, enable_stat_watcher, read_lines_limit, read_lines_limit_per_notify, update_watcher, line_buffer_timer_flusher, from_encoding, encoding, open_on_every_update, &receive_lines) @path = path @rotate_wait = rotate_wait @pe = pe || MemoryPositionEntry.new @@ -502,6 +504,7 @@ def initialize(path, rotate_wait, pe, log, read_from_head, enable_watch_timer, e @enable_watch_timer = enable_watch_timer @enable_stat_watcher = enable_stat_watcher @read_lines_limit = read_lines_limit + @read_lines_limit_per_notify = read_lines_limit_per_notify @receive_lines = receive_lines @update_watcher = update_watcher @@ -519,7 +522,7 @@ def initialize(path, rotate_wait, pe, log, read_from_head, enable_watch_timer, e end attr_reader :path - attr_reader :log, :pe, :read_lines_limit, :open_on_every_update + attr_reader :log, :pe, :read_lines_limit, :read_lines_limit_per_notify, :open_on_every_update attr_reader :from_encoding, :encoding attr_reader :stat_trigger, :enable_watch_timer, :enable_stat_watcher attr_accessor :timer_trigger @@ -754,11 +757,18 @@ def handle_notify while true @fifo << io.readpartial(8192, @iobuf) @fifo.read_lines(@lines) - if @lines.size >= @watcher.read_lines_limit + limit_per_notify = @lines.size >= @watcher.read_lines_limit_per_notify and @watcher.read_lines_limit_per_notify > 0 + if limit_per_notify + # stop reading files when we reach the read lines limit per notify, to throttle the log ingestion + read_more = false + elsif @lines.size >= @watcher.read_lines_limit # not to use too much memory in case the file is very large read_more = true + end + if @lines.size >= @watcher.read_lines_limit or limit_per_notify break end + end rescue EOFError end diff --git a/test/plugin/test_in_tail.rb b/test/plugin/test_in_tail.rb index d67cbdce3c..e634fe7095 100644 --- a/test/plugin/test_in_tail.rb +++ b/test/plugin/test_in_tail.rb @@ -73,6 +73,7 @@ def create_driver(conf = SINGLE_LINE_CONFIG, use_common_conf = true) assert_equal 2, d.instance.rotate_wait assert_equal "#{TMP_DIR}/tail.pos", d.instance.pos_file assert_equal 1000, d.instance.read_lines_limit + assert_equal -1, d.instance.read_lines_limit_per_notify assert_equal false, d.instance.ignore_repeated_permission_error end @@ -212,6 +213,35 @@ def test_emit_with_read_lines_limit(data) assert_equal(num_events, d.emit_count) end + data('flat 1' => [:flat, 100, 1, 10], + 'flat 10' => [:flat, 100, 10, 20], + 'parse 1' => [:parse, 100, 3, 10], + 'parse 10' => [:parse, 100, 10, 20]) + def test_emit_with_read_lines_limit_per_notify(data) + config_style, limit, limit_per_notify, num_events = data + case config_style + when :flat + config = CONFIG_READ_FROM_HEAD + SINGLE_LINE_CONFIG + config_element("", "", { "read_lines_limit" => limit, "read_lines_limit_per_notify" => limit_per_notify }) + when :parse + config = CONFIG_READ_FROM_HEAD + config_element("", "", { "read_lines_limit" => limit, "read_lines_limit_per_notify" => limit_per_notify }) + PARSE_SINGLE_LINE_CONFIG + end + d = create_driver(config) + msg = 'test' * 2000 # in_tail reads 8192 bytes at once. + + d.run(expect_emits: 1) do + File.open("#{TMP_DIR}/tail.txt", "ab") {|f| + for x in 0..20 + f.puts msg + end + } + end + + events = d.events + assert_equal(true, events.length <= num_events) + assert_equal({"message" => msg}, events[0][2]) + assert_equal({"message" => msg}, events[1][2]) + end + data(flat: CONFIG_READ_FROM_HEAD + SINGLE_LINE_CONFIG, parse: CONFIG_READ_FROM_HEAD + PARSE_SINGLE_LINE_CONFIG) def test_emit_with_read_from_head(data) @@ -1061,7 +1091,7 @@ def test_z_refresh_watchers Timecop.freeze(2010, 1, 2, 3, 4, 5) do flexstub(Fluent::Plugin::TailInput::TailWatcher) do |watcherclass| EX_PATHS.each do |path| - watcherclass.should_receive(:new).with(path, EX_ROTATE_WAIT, Fluent::Plugin::TailInput::FilePositionEntry, any, true, true, true, 1000, any, any, any, any, any, any).once.and_return do + watcherclass.should_receive(:new).with(path, EX_ROTATE_WAIT, Fluent::Plugin::TailInput::FilePositionEntry, any, true, true, true, 1000, -1, any, any, any, any, any, any).once.and_return do flexmock('TailWatcher') { |watcher| watcher.should_receive(:attach).once watcher.should_receive(:unwatched=).zero_or_more_times @@ -1079,7 +1109,7 @@ def test_z_refresh_watchers Timecop.freeze(2010, 1, 2, 3, 4, 6) do flexstub(Fluent::Plugin::TailInput::TailWatcher) do |watcherclass| - watcherclass.should_receive(:new).with('test/plugin/data/2010/01/20100102-030406.log', EX_ROTATE_WAIT, Fluent::Plugin::TailInput::FilePositionEntry, any, true, true, true, 1000, any, any, any, any, any, any).once.and_return do + watcherclass.should_receive(:new).with('test/plugin/data/2010/01/20100102-030406.log', EX_ROTATE_WAIT, Fluent::Plugin::TailInput::FilePositionEntry, any, true, true, true, 1000, -1, any, any, any, any, any, any).once.and_return do flexmock('TailWatcher') do |watcher| watcher.should_receive(:attach).once watcher.should_receive(:unwatched=).zero_or_more_times From 2ce05b1aa46737409e95e9d9725386fd1e385804 Mon Sep 17 00:00:00 2001 From: Anthony Comtois Date: Mon, 18 Nov 2019 20:57:02 +0000 Subject: [PATCH 2/5] Update log throttling read bytes per second Signed-off-by: Anthony Comtois --- lib/fluent/plugin/in_tail.rb | 25 ++++++++++++++----------- test/plugin/test_in_tail.rb | 18 +++++++++--------- 2 files changed, 23 insertions(+), 20 deletions(-) diff --git a/lib/fluent/plugin/in_tail.rb b/lib/fluent/plugin/in_tail.rb index 1b645639a1..feea22fc00 100644 --- a/lib/fluent/plugin/in_tail.rb +++ b/lib/fluent/plugin/in_tail.rb @@ -74,8 +74,8 @@ def initialize config_param :refresh_interval, :time, default: 60 desc 'The number of reading lines at each IO.' config_param :read_lines_limit, :integer, default: 1000 - desc 'The number of reading lines per notify' - config_param :read_lines_limit_per_notify, :integer, default: -1 + desc 'The number of reading bytes per second' + config_param :read_bytes_limit_per_second, :integer, default: -1 desc 'The interval of flushing the buffer for multiline format' config_param :multiline_flush_interval, :time, default: nil desc 'Enable the option to emit unmatched lines.' @@ -285,7 +285,7 @@ def refresh_watchers def setup_watcher(path, pe) line_buffer_timer_flusher = (@multiline_mode && @multiline_flush_interval) ? TailWatcher::LineBufferTimerFlusher.new(log, @multiline_flush_interval, &method(:flush_buffer)) : nil - tw = TailWatcher.new(path, @rotate_wait, pe, log, @read_from_head, @enable_watch_timer, @enable_stat_watcher, @read_lines_limit, @read_lines_limit_per_notify, method(:update_watcher), line_buffer_timer_flusher, @from_encoding, @encoding, open_on_every_update, &method(:receive_lines)) + tw = TailWatcher.new(path, @rotate_wait, pe, log, @read_from_head, @enable_watch_timer, @enable_stat_watcher, @read_lines_limit, @read_bytes_limit_per_second, method(:update_watcher), line_buffer_timer_flusher, @from_encoding, @encoding, open_on_every_update, &method(:receive_lines)) tw.attach do |watcher| event_loop_attach(watcher.timer_trigger) if watcher.timer_trigger event_loop_attach(watcher.stat_trigger) if watcher.stat_trigger @@ -496,7 +496,7 @@ def parse_multilines(lines, tail_watcher) end class TailWatcher - def initialize(path, rotate_wait, pe, log, read_from_head, enable_watch_timer, enable_stat_watcher, read_lines_limit, read_lines_limit_per_notify, update_watcher, line_buffer_timer_flusher, from_encoding, encoding, open_on_every_update, &receive_lines) + def initialize(path, rotate_wait, pe, log, read_from_head, enable_watch_timer, enable_stat_watcher, read_lines_limit, read_bytes_limit_per_second, update_watcher, line_buffer_timer_flusher, from_encoding, encoding, open_on_every_update, &receive_lines) @path = path @rotate_wait = rotate_wait @pe = pe || MemoryPositionEntry.new @@ -504,7 +504,7 @@ def initialize(path, rotate_wait, pe, log, read_from_head, enable_watch_timer, e @enable_watch_timer = enable_watch_timer @enable_stat_watcher = enable_stat_watcher @read_lines_limit = read_lines_limit - @read_lines_limit_per_notify = read_lines_limit_per_notify + @read_bytes_limit_per_second = read_bytes_limit_per_second @receive_lines = receive_lines @update_watcher = update_watcher @@ -522,7 +522,7 @@ def initialize(path, rotate_wait, pe, log, read_from_head, enable_watch_timer, e end attr_reader :path - attr_reader :log, :pe, :read_lines_limit, :read_lines_limit_per_notify, :open_on_every_update + attr_reader :log, :pe, :read_lines_limit, :read_bytes_limit_per_second, :open_on_every_update attr_reader :from_encoding, :encoding attr_reader :stat_trigger, :enable_watch_timer, :enable_stat_watcher attr_accessor :timer_trigger @@ -750,22 +750,25 @@ def on_notify def handle_notify with_io do |io| begin + bytes_to_read = 8192 + number_bytes_read = 0 read_more = false if !io.nil? && @lines.empty? begin while true - @fifo << io.readpartial(8192, @iobuf) + @fifo << io.readpartial(bytes_to_read, @iobuf) @fifo.read_lines(@lines) - limit_per_notify = @lines.size >= @watcher.read_lines_limit_per_notify and @watcher.read_lines_limit_per_notify > 0 - if limit_per_notify - # stop reading files when we reach the read lines limit per notify, to throttle the log ingestion + number_bytes_read += bytes_to_read + limit_bytes_per_second_reached = (@lines.size >= @watcher.read_bytes_limit_per_second and @watcher.read_bytes_limit_per_second > 0) + if limit_bytes_per_second_reached + # stop reading files when we reach the read bytes limit per second, to throttle the log ingestion read_more = false elsif @lines.size >= @watcher.read_lines_limit # not to use too much memory in case the file is very large read_more = true end - if @lines.size >= @watcher.read_lines_limit or limit_per_notify + if @lines.size >= @watcher.read_lines_limit or limit_bytes_per_second_reached break end diff --git a/test/plugin/test_in_tail.rb b/test/plugin/test_in_tail.rb index e634fe7095..29ea7a9c91 100644 --- a/test/plugin/test_in_tail.rb +++ b/test/plugin/test_in_tail.rb @@ -73,7 +73,7 @@ def create_driver(conf = SINGLE_LINE_CONFIG, use_common_conf = true) assert_equal 2, d.instance.rotate_wait assert_equal "#{TMP_DIR}/tail.pos", d.instance.pos_file assert_equal 1000, d.instance.read_lines_limit - assert_equal -1, d.instance.read_lines_limit_per_notify + assert_equal -1, d.instance.read_bytes_limit_per_second assert_equal false, d.instance.ignore_repeated_permission_error end @@ -213,17 +213,17 @@ def test_emit_with_read_lines_limit(data) assert_equal(num_events, d.emit_count) end - data('flat 1' => [:flat, 100, 1, 10], - 'flat 10' => [:flat, 100, 10, 20], - 'parse 1' => [:parse, 100, 3, 10], - 'parse 10' => [:parse, 100, 10, 20]) - def test_emit_with_read_lines_limit_per_notify(data) - config_style, limit, limit_per_notify, num_events = data + data('flat 1' => [:flat, 100, 8192, 10], + 'flat 10' => [:flat, 100, (8192 * 10), 20], + 'parse 1' => [:parse, 100, (8192 * 3), 10], + 'parse 10' => [:parse, 100, (8192 * 10), 20]) + def test_emit_with_read_bytes_limit_per_second(data) + config_style, limit, limit_bytes, num_events = data case config_style when :flat - config = CONFIG_READ_FROM_HEAD + SINGLE_LINE_CONFIG + config_element("", "", { "read_lines_limit" => limit, "read_lines_limit_per_notify" => limit_per_notify }) + config = CONFIG_READ_FROM_HEAD + SINGLE_LINE_CONFIG + config_element("", "", { "read_lines_limit" => limit, "read_bytes_limit_per_second" => limit_bytes }) when :parse - config = CONFIG_READ_FROM_HEAD + config_element("", "", { "read_lines_limit" => limit, "read_lines_limit_per_notify" => limit_per_notify }) + PARSE_SINGLE_LINE_CONFIG + config = CONFIG_READ_FROM_HEAD + config_element("", "", { "read_lines_limit" => limit, "read_bytes_limit_per_second" => limit_bytes }) + PARSE_SINGLE_LINE_CONFIG end d = create_driver(config) msg = 'test' * 2000 # in_tail reads 8192 bytes at once. From 7ff13658c6171c4bff05001bb6e24e8350626269 Mon Sep 17 00:00:00 2001 From: Anthony Comtois Date: Sat, 16 Nov 2019 20:55:14 +0000 Subject: [PATCH 3/5] Update log throttling based on number of bytes and compatible with inotify Signed-off-by: Anthony Comtois --- lib/fluent/plugin/in_tail.rb | 24 ++++++++++++++++++------ 1 file changed, 18 insertions(+), 6 deletions(-) diff --git a/lib/fluent/plugin/in_tail.rb b/lib/fluent/plugin/in_tail.rb index feea22fc00..ef75cfdbb8 100644 --- a/lib/fluent/plugin/in_tail.rb +++ b/lib/fluent/plugin/in_tail.rb @@ -750,23 +750,35 @@ def on_notify def handle_notify with_io do |io| begin - bytes_to_read = 8192 number_bytes_read = 0 + start_reading = Time.new read_more = false if !io.nil? && @lines.empty? begin while true + bytes_to_read = 8192 @fifo << io.readpartial(bytes_to_read, @iobuf) @fifo.read_lines(@lines) + number_bytes_read += bytes_to_read - limit_bytes_per_second_reached = (@lines.size >= @watcher.read_bytes_limit_per_second and @watcher.read_bytes_limit_per_second > 0) - if limit_bytes_per_second_reached - # stop reading files when we reach the read bytes limit per second, to throttle the log ingestion - read_more = false - elsif @lines.size >= @watcher.read_lines_limit + limit_bytes_per_second_reached = number_bytes_read >= @watcher.read_bytes_limit_per_second and @watcher.read_bytes_limit_per_second > 0 + + if @lines.size >= @watcher.read_lines_limit or limit_bytes_per_second_reached # not to use too much memory in case the file is very large read_more = true + + if limit_bytes_per_second_reached + # sleep to stop reading files when we reach the read lines byte per second, to throttle the log ingestion + time_spent_reading = Time.new - start_reading + @watcher.log.debug("time_spent_reading: #{time_spent_reading}") + if (time_spent_reading < 1) + debug_time = 1 - time_spent_reading + @watcher.log.debug("sleep: #{debug_time}") + sleep(1 - time_spent_reading) + end + start_reading = Time.new + end end if @lines.size >= @watcher.read_lines_limit or limit_bytes_per_second_reached break From 39370c459f90db9310147b094a9aa79cabe4d743 Mon Sep 17 00:00:00 2001 From: Anthony Comtois Date: Sun, 17 Nov 2019 14:00:31 +0000 Subject: [PATCH 4/5] Add tail concurrency with Thread for TailWatcher Signed-off-by: Anthony Comtois --- lib/fluent/plugin/in_tail.rb | 59 ++++++++++++++++++++++-------------- 1 file changed, 37 insertions(+), 22 deletions(-) diff --git a/lib/fluent/plugin/in_tail.rb b/lib/fluent/plugin/in_tail.rb index ef75cfdbb8..19739a08a5 100644 --- a/lib/fluent/plugin/in_tail.rb +++ b/lib/fluent/plugin/in_tail.rb @@ -49,6 +49,7 @@ def to_s def initialize super @paths = [] + @threads = {} @tails = {} @pf_file = nil @pf = nil @@ -203,7 +204,14 @@ def start end refresh_watchers unless @skip_refresh_on_startup - timer_execute(:in_tail_refresh_watchers, @refresh_interval, &method(:refresh_watchers)) + + @threads['in_tail_refresh_watchers'] = Thread.new(@refresh_interval) do |refresh_interval| + timer_execute(:in_tail_refresh_watchers, @refresh_interval, &method(:refresh_watchers)) + end + + @threads.each { |thr| + thr.join + } end def stop @@ -304,31 +312,38 @@ def setup_watcher(path, pe) def start_watchers(paths) paths.each { |path| - pe = nil - if @pf - pe = @pf[path] - if @read_from_head && pe.read_inode.zero? - begin - pe.update(Fluent::FileWrapper.stat(path).ino, 0) - rescue Errno::ENOENT - $log.warn "#{path} not found. Continuing without tailing it." + @threads[path] = Thread.new(path) do |path| + pe = nil + if @pf + pe = @pf[path] + if @read_from_head && pe.read_inode.zero? + begin + pe.update(Fluent::FileWrapper.stat(path).ino, 0) + rescue Errno::ENOENT + $log.warn "#{path} not found. Continuing without tailing it." + end end end - end - begin - tw = setup_watcher(path, pe) - rescue WatcherSetupError => e - log.warn "Skip #{path} because unexpected setup error happens: #{e}" - next + begin + tw = setup_watcher(path, pe) + rescue WatcherSetupError => e + log.warn "Skip #{path} because unexpected setup error happens: #{e}" + next + end + @tails[path] = tw end - @tails[path] = tw } end def stop_watchers(paths, immediate: false, unwatched: false, remove_watcher: true) paths.each { |path| tw = remove_watcher ? @tails.delete(path) : @tails[path] + if remove_watcher + @threads[path].exit + @threads.delete(path) + end + if tw tw.unwatched = unwatched if immediate @@ -342,6 +357,8 @@ def stop_watchers(paths, immediate: false, unwatched: false, remove_watcher: tru def close_watcher_handles @tails.keys.each do |path| + @threads[path].exit + @threads.delete(path) tw = @tails.delete(path) if tw tw.close @@ -750,6 +767,7 @@ def on_notify def handle_notify with_io do |io| begin + bytes_to_read = 8192 number_bytes_read = 0 start_reading = Time.new read_more = false @@ -757,19 +775,18 @@ def handle_notify if !io.nil? && @lines.empty? begin while true - bytes_to_read = 8192 @fifo << io.readpartial(bytes_to_read, @iobuf) @fifo.read_lines(@lines) number_bytes_read += bytes_to_read - limit_bytes_per_second_reached = number_bytes_read >= @watcher.read_bytes_limit_per_second and @watcher.read_bytes_limit_per_second > 0 + limit_bytes_per_second_reached = (number_bytes_read >= @watcher.read_bytes_limit_per_second and @watcher.read_bytes_limit_per_second > 0) if @lines.size >= @watcher.read_lines_limit or limit_bytes_per_second_reached # not to use too much memory in case the file is very large read_more = true if limit_bytes_per_second_reached - # sleep to stop reading files when we reach the read lines byte per second, to throttle the log ingestion + # sleep to stop reading files when we reach the read bytes per second limit, to throttle the log ingestion time_spent_reading = Time.new - start_reading @watcher.log.debug("time_spent_reading: #{time_spent_reading}") if (time_spent_reading < 1) @@ -779,11 +796,9 @@ def handle_notify end start_reading = Time.new end - end - if @lines.size >= @watcher.read_lines_limit or limit_bytes_per_second_reached + break end - end rescue EOFError end From 416693c23719e3071b6cf191b186947370d6ddf0 Mon Sep 17 00:00:00 2001 From: Anthony Comtois Date: Mon, 18 Nov 2019 20:41:01 +0000 Subject: [PATCH 5/5] implement thread based tailwatcher Signed-off-by: Anthony Comtois --- lib/fluent/plugin/in_tail.rb | 53 ++++++++++++++++++++++++------------ 1 file changed, 36 insertions(+), 17 deletions(-) diff --git a/lib/fluent/plugin/in_tail.rb b/lib/fluent/plugin/in_tail.rb index 19739a08a5..2ba4b5292f 100644 --- a/lib/fluent/plugin/in_tail.rb +++ b/lib/fluent/plugin/in_tail.rb @@ -208,6 +208,7 @@ def start @threads['in_tail_refresh_watchers'] = Thread.new(@refresh_interval) do |refresh_interval| timer_execute(:in_tail_refresh_watchers, @refresh_interval, &method(:refresh_watchers)) end + @threads['in_tail_refresh_watchers'].priority = 10 # Default is zero; higher-priority threads will run before lower-priority threads. @threads.each { |thr| thr.join @@ -289,6 +290,11 @@ def refresh_watchers stop_watchers(unwatched, immediate: false, unwatched: true) unless unwatched.empty? start_watchers(added) unless added.empty? + + log.debug "Thread refresh_watchers" + @threads.each { |thr| + log.debug "Thread #{thr[0]} #{thr[1].status}" + } end def setup_watcher(path, pe) @@ -312,26 +318,37 @@ def setup_watcher(path, pe) def start_watchers(paths) paths.each { |path| - @threads[path] = Thread.new(path) do |path| - pe = nil - if @pf - pe = @pf[path] - if @read_from_head && pe.read_inode.zero? - begin - pe.update(Fluent::FileWrapper.stat(path).ino, 0) - rescue Errno::ENOENT - $log.warn "#{path} not found. Continuing without tailing it." + unless @threads[path].nil? + log.debug "Check Thread #{path} #{@threads[path].status}" + if @threads[path].status != "sleep" and @threads[path].status != "run" + log.debug "Stopping Thread #{path} #{@threads[path].status}" + @threads[path].exit + @threads.delete(path) + end + end + if @threads[path].nil? + log.debug "Add Thread #{path}" + @threads[path] = Thread.new(path) do |path| + pe = nil + if @pf + pe = @pf[path] + if @read_from_head && pe.read_inode.zero? + begin + pe.update(Fluent::FileWrapper.stat(path).ino, 0) + rescue Errno::ENOENT + $log.warn "#{path} not found. Continuing without tailing it." + end end end - end - begin - tw = setup_watcher(path, pe) - rescue WatcherSetupError => e - log.warn "Skip #{path} because unexpected setup error happens: #{e}" - next + begin + tw = setup_watcher(path, pe) + rescue WatcherSetupError => e + log.warn "Skip #{path} because unexpected setup error happens: #{e}" + next + end + @tails[path] = tw end - @tails[path] = tw end } end @@ -375,6 +392,7 @@ def update_watcher(path, pe) end end rotated_tw = @tails[path] + @tails[path] = setup_watcher(path, pe) detach_watcher_after_rotate_wait(rotated_tw) if rotated_tw end @@ -781,6 +799,7 @@ def handle_notify number_bytes_read += bytes_to_read limit_bytes_per_second_reached = (number_bytes_read >= @watcher.read_bytes_limit_per_second and @watcher.read_bytes_limit_per_second > 0) + @watcher.log.debug("reading file: #{ @watcher.path}") if @lines.size >= @watcher.read_lines_limit or limit_bytes_per_second_reached # not to use too much memory in case the file is very large read_more = true @@ -788,7 +807,7 @@ def handle_notify if limit_bytes_per_second_reached # sleep to stop reading files when we reach the read bytes per second limit, to throttle the log ingestion time_spent_reading = Time.new - start_reading - @watcher.log.debug("time_spent_reading: #{time_spent_reading}") + @watcher.log.debug("time_spent_reading: #{time_spent_reading} #{ @watcher.path}") if (time_spent_reading < 1) debug_time = 1 - time_spent_reading @watcher.log.debug("sleep: #{debug_time}")