diff --git a/lib/fluent/plugin/in_tail.rb b/lib/fluent/plugin/in_tail.rb index 7498df1e5c..525f7d563d 100644 --- a/lib/fluent/plugin/in_tail.rb +++ b/lib/fluent/plugin/in_tail.rb @@ -32,6 +32,7 @@ require 'fluent/plugin/in_tail/tail_watcher' require 'fluent/plugin/in_tail/line_feeder' require 'fluent/plugin/in_tail/compatibility' +require 'fluent/plugin/in_tail/worker_pool' require 'fluent/file_wrapper' module Fluent::Plugin @@ -98,6 +99,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 threads used to parse tailed files. Set to 1 to keep parsing synchronous; 2 or more enables worker parsing. The default is 1.' + config_param :num_threads, :integer, default: 1 desc 'The number of reading bytes per second' config_param :read_bytes_limit_per_second, :size, default: -1 desc 'The interval of flushing the buffer for multiline format' @@ -162,6 +165,7 @@ def configure(conf) if !@enable_watch_timer && !@enable_stat_watcher raise Fluent::ConfigError, "either of enable_watch_timer or enable_stat_watcher must be true" end + raise Fluent::ConfigError, "'num_threads' must be greater than 0" unless @num_threads.positive? if @enable_dir_watcher && !@enable_stat_watcher log.info( @@ -290,6 +294,7 @@ def start super @line_feeder = build_line_feeder + configure_worker_mode if @num_threads > 1 if @pos_file pos_file_dir = File.dirname(@pos_file) @@ -337,19 +342,23 @@ def shutdown detach_refresh_timer # 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], false) + @tails_rotate_wait.dup.each do |tw, rotation| + event_loop_detach(rotation[:timer]) if @worker_pool + detach_watcher(tw, rotation[:ino], false) end - @pf_file.close if @pf_file + shutdown_worker_pool + @pf_file.close if @pf_file && !@worker_pool super end def close super + @worker_completion_watcher&.close # close file handles after all threads stopped (in #close of thread plugin helper) # It may be because we need to wait IOHandler.ready_to_shutdown() close_watcher_handles + @pf_file.close if @pf_file && !@pf_file.closed? end def have_read_capability? @@ -471,6 +480,11 @@ def stop_watchers(targets_info, immediate: false, unwatched: false, remove_watch end def close_watcher_handles + if @worker_pool + close_worker_watcher_handles + return + end + @tails.keys.each do |path| tw = @tails.delete(path) if tw @@ -546,6 +560,12 @@ def detach_watcher(tw, ino, close_io = true) end tw.detach(@shutdown_start_time) + # The final read in detach may itself submit a worker batch. + defer_worker_completion = tw.pending? || @worker_waiting_watchers&.include?(tw) + if close_io && defer_worker_completion + tw.defer_close + @worker_detached_watchers << tw unless @worker_detached_watchers.include?(tw) + end tw.close if close_io # A watcher waiting for `rotate_wait` cannot read once its path leaves the group watcher. @@ -553,7 +573,11 @@ def detach_watcher(tw, ino, close_io = true) if @pf && tw.unwatched && (@follow_inodes || !@tails[tw.path]) target_info = TargetInfo.new(tw.path, ino) - @pf.unwatch(target_info) + if defer_worker_completion + @worker_deferred_unwatch[tw] = target_info + else + @pf.unwatch(target_info) + end end end @@ -618,6 +642,13 @@ def statistics 'tracked_file_count' => @metrics.tracked.get, }) } + if @worker_pool + stats['input'].merge!( + 'worker_pending_batch_count' => @worker_metrics[:pending_batches].get, + 'worker_pending_bytes' => @worker_metrics[:pending_bytes].get, + 'worker_parse_error_count' => @worker_metrics[:parse_errors].get, + ) + end stats end @@ -649,4 +680,238 @@ def io_handler(watcher, path) ) end end + + # Reopen TailInput to keep worker-mode setup and lifecycle helpers grouped + # separately from the synchronous file-watching implementation above. + class TailInput + WORKER_COMPATIBILITY = WorkerCompatibility.new(self) + WORKER_PARSER_TYPES = %w[ + none regexp json csv tsv ltsv msgpack apache apache2 apache_error nginx syslog + ].freeze + WORKER_QUEUE_BYTES = 64 * 1024 * 1024 + WorkerBatch = Struct.new(:watcher, :lines) + + private + + def configure_worker_mode + incompatibility_reasons = worker_incompatibility_reasons + @worker_compatible = incompatibility_reasons.empty? + if @worker_compatible + setup_worker_pool + log.info "in_tail worker parsing enabled with #{@num_threads} threads" + else + log.warn "in_tail worker parsing is unavailable; using synchronous parsing (#{incompatibility_reasons.join(', ')})" + end + end + + def worker_incompatibility_reasons + reasons = WORKER_COMPATIBILITY.incompatible_hooks(self).map { |hook| "overridden hook: #{hook}" } + reasons << 'multiline parsing is enabled' if @multiline_mode + reasons << 'open_on_every_update is enabled' if @open_on_every_update + parser_type = @parser_configs.first[:@type] + reasons << "unsupported parser: #{parser_type}" unless WORKER_PARSER_TYPES.include?(parser_type) + parser_usage = @parser_configs.first.usage + reasons << 'the configured parser instance was replaced' unless @parser.equal?(@_parsers[parser_usage]) + reasons + end + + def create_worker_parsers(num_threads) + return [] unless @worker_compatible + raise ArgumentError, 'worker count must be positive' unless num_threads.positive? + + parser_config = @parser_configs.first.corresponding_config_element + Array.new(num_threads) do |index| + usage = "__in_tail_worker_#{object_id}_#{index}" + parser_create(usage: usage, conf: parser_config) + end + end + + def setup_worker_pool + @worker_metrics = { + pending_batches: metrics_create(namespace: 'fluentd', subsystem: 'input', name: 'worker_pending_batches', + help_text: 'Number of worker batches awaiting completion or acknowledgment', + prefer_gauge: true), + pending_bytes: metrics_create(namespace: 'fluentd', subsystem: 'input', name: 'worker_pending_bytes', + help_text: 'Estimated bytes accounted for by pending worker batches', + prefer_gauge: true), + parse_errors: metrics_create(namespace: 'fluentd', subsystem: 'input', name: 'worker_parse_errors_total', + help_text: 'Total number of worker parse errors'), + } + @worker_parsers = create_worker_parsers(@num_threads) + @worker_line_feeders = @worker_parsers.map { |parser| build_worker_line_feeder(parser) } + @worker_waiting_watchers = [] + @worker_deferred_unwatch = {} + @worker_detached_watchers = [] + @worker_completion_mutex = Mutex.new + @worker_completions_stopped = false + @worker_completion_watcher = WorkerCompletionWatcher.new(log: log) { process_worker_completions } + event_loop_attach(@worker_completion_watcher) + @worker_pool = WorkerPool.new( + num_threads: @num_threads, + task_limit: @num_threads * 2, + byte_limit: WORKER_QUEUE_BYTES, + thread_create: ->(title, &block) { Thread.new { Thread.current.name = title.to_s if Thread.current.respond_to?(:name=); block.call } }, + notify: @worker_completion_watcher.method(:signal) + ) do |index, batch| + @worker_line_feeders[index].parse(batch.lines.map(&:dup), batch.watcher) + end + @worker_pool.start + rescue + event_loop_detach(@worker_completion_watcher) if @worker_completion_watcher&.attached? + @worker_completion_watcher&.close + raise + end + + def build_worker_line_feeder(parser) + LineFeeder.new( + parser: parser, + router_provider: method(:router), + log: log, + tag: @tag, + tag_prefix: @tag_prefix, + tag_suffix: @tag_suffix, + path_key: @path_key, + emit_unmatched_lines: @emit_unmatched_lines, + multiline_mode: false + ) + end + + def async_receive_lines(lines, watcher) + return @line_feeder.feed_lines(lines, watcher) unless @worker_pool.statistics[:state] == :running + + bytes = lines.sum(&:bytesize) * 3 + batch = WorkerBatch.new(watcher, lines) + if @worker_pool.submit(watcher, batch, bytes: bytes) + update_worker_metrics + @worker_waiting_watchers.delete(watcher) + TailWatcher::IOHandler::ASYNC_PENDING + else + @worker_waiting_watchers << watcher unless @worker_waiting_watchers.include?(watcher) + false + end + end + + def process_worker_completions + @worker_completion_mutex.synchronize do + drain_worker_completions(resume: true) unless @worker_completions_stopped + end + end + + def drain_worker_completions(resume:) + while (result = @worker_pool.next_result) + begin + complete_worker_result(result, resume: resume) + rescue => error + log.error 'unexpected error while completing in_tail worker batch', error: error + log.error_backtrace + end + end + + return unless resume + + waiting = @worker_waiting_watchers + @worker_waiting_watchers = [] + waiting.each do |watcher| + begin + watcher.detached? ? watcher.read_more : watcher.on_notify + rescue => error + log.error 'unexpected error while retrying in_tail worker batch', error: error + log.error_backtrace + ensure + finish_worker_deferred_close(watcher) + end + end + end + + def complete_worker_result(result, resume:) + batch = result.task.payload + emitted = false + begin + if result.error + @worker_metrics[:parse_errors].inc + if resume + log.warn 'worker parsing failed; retrying synchronously', path: batch.watcher.path, error: result.error + emitted = @line_feeder.feed_lines(batch.lines, batch.watcher) + else + log.warn 'worker parsing failed during shutdown; leaving position unchanged', path: batch.watcher.path, error: result.error + end + else + emitted = @line_feeder.emit(result.value, batch.watcher) + end + ensure + begin + @worker_pool.acknowledge(result) + update_worker_metrics + ensure + begin + batch.watcher.complete_async(emitted, resume: resume) + ensure + finish_worker_deferred_close(batch.watcher) + end + end + end + end + + def finish_worker_deferred_close(watcher) + return if watcher.pending? || @worker_waiting_watchers.include?(watcher) + + # Match synchronous detach on failed emit or a throttled final read: + # close the handle, leaving uncommitted data for a future watch. + watcher.close_if_ready(force: watcher.detached?) + @worker_detached_watchers.delete(watcher) + + target_info = @worker_deferred_unwatch.delete(watcher) + @pf.unwatch(target_info) if target_info && @pf + end + + def update_worker_metrics + statistics = @worker_pool.statistics + @worker_metrics[:pending_batches].set(statistics[:tasks]) + @worker_metrics[:pending_bytes].set(statistics[:bytes]) + end + + def close_worker_watcher_handles + watchers = (@tails.values + @tails_rotate_wait.keys + @worker_detached_watchers + @worker_deferred_unwatch.keys).uniq + watchers.each(&:force_close) + @tails.clear + @worker_detached_watchers.clear + @worker_deferred_unwatch.each_value do |target_info| + @pf.unwatch(target_info) if @pf + end + @worker_deferred_unwatch.clear + end + + def shutdown_worker_pool + return unless @worker_pool + + deadline = Fluent::Clock.now + TailWatcher::IOHandler::SHUTDOWN_TIMEOUT + while worker_work_pending? + break if Fluent::Clock.now >= deadline + sleep 0.01 + end + if worker_work_pending? + log.warn 'in_tail worker pool did not drain before shutdown timeout' + end + event_loop_detach(@worker_completion_watcher) + # Detach alone does not wait for a callback already running. Serialize + # the handover so every result is handled before output shutdown starts. + @worker_completion_mutex.synchronize do + @worker_completions_stopped = true + @worker_pool.stop + end + unless @worker_pool.join(timeout: TailWatcher::IOHandler::SHUTDOWN_TIMEOUT) + log.warn 'in_tail worker threads did not stop before timeout' + @worker_pool.kill + end + @worker_completion_mutex.synchronize do + drain_worker_completions(resume: false) + end + end + + def worker_work_pending? + @worker_completion_mutex.synchronize do + @worker_pool.statistics[:tasks].positive? || !@worker_waiting_watchers.empty? + end + end + end end diff --git a/lib/fluent/plugin/in_tail/compatibility.rb b/lib/fluent/plugin/in_tail/compatibility.rb index be5b985f7f..fcd30c1784 100644 --- a/lib/fluent/plugin/in_tail/compatibility.rb +++ b/lib/fluent/plugin/in_tail/compatibility.rb @@ -33,6 +33,8 @@ def flush_buffer(tw, buf) # # @return true if no error or unrecoverable error happens in emit action. false if got BufferOverflowError def receive_lines(lines, tail_watcher) + return async_receive_lines(lines, tail_watcher) if @worker_pool + file_feed = tail_watcher.file_feed return file_feed.feed_lines(lines, tail_watcher) if file_feed.respond_to?(:feed_lines) diff --git a/lib/fluent/plugin/in_tail/dir_watcher.rb b/lib/fluent/plugin/in_tail/dir_watcher.rb index 87be457c0c..89c089c8ec 100644 --- a/lib/fluent/plugin/in_tail/dir_watcher.rb +++ b/lib/fluent/plugin/in_tail/dir_watcher.rb @@ -237,12 +237,18 @@ def refresh_watchers_raw # After `update_watcher` detaches an old TailWatcher, the inode is lost from the `@tails`. # So that inode can't be contained in `removed_hash`, and can't be unwatched by `stop_watchers`. # - # Inodes still read by a watcher waiting for `rotate_wait` are kept. Compaction only updates - # the entries that remain in the position file, so such a watcher would otherwise write its - # position into the line of another file. + # Keep inodes still read by watchers waiting for rotate_wait or worker completion. + # Compaction only updates entries remaining in the position file, so removing one + # early would let its watcher write into another file's line. def unwatch_removed_inodes(target_paths_hash) return unless @pf draining_hash = @tails_rotate_wait.to_h { |tw, v| [v[:ino], TargetInfo.new(tw.path, v[:ino])] } + @worker_detached_watchers&.each do |tw| + draining_hash[tw.ino] = TargetInfo.new(tw.path, tw.ino) + end + @worker_deferred_unwatch&.each_value do |target| + draining_hash[target.ino] = target + end @pf.unwatch_removed_targets(draining_hash.merge(target_paths_hash)) end diff --git a/lib/fluent/plugin/in_tail/io_handler.rb b/lib/fluent/plugin/in_tail/io_handler.rb index a1f36a185b..acb36a22ab 100644 --- a/lib/fluent/plugin/in_tail/io_handler.rb +++ b/lib/fluent/plugin/in_tail/io_handler.rb @@ -108,6 +108,7 @@ def reading_bytesize class IOHandler BYTES_TO_READ = 64 * 1024 SHUTDOWN_TIMEOUT = 5 + ASYNC_PENDING = :async_pending attr_accessor :shutdown_timeout @@ -131,6 +132,10 @@ def initialize(watcher, path:, read_lines_limit:, read_bytes_limit_per_second:, @shutdown_mutex = Mutex.new @eof = false @metrics = metrics + @async_pending = false + @notify_pending = false + @pending_position = nil + @pending_read_more = false @log.info "following tail of #{@path}" end @@ -166,6 +171,40 @@ def eof? @eof end + def drained? + @eof && @lines.empty? && !@async_pending + end + + def pending? + @async_pending + end + + # Completes a receive callback that returned ASYNC_PENDING. The caller + # must invoke TailWatcher#on_notify when this returns true. + def complete_async(success) + @notify_mutex.synchronize do + raise 'no asynchronous receive is pending' unless @async_pending + + begin + if success + @watcher.pe.update_pos(@pending_position) + @lines.clear + end + should_notify = (success && @pending_read_more) || @notify_pending + @pending_read_more = false if success + should_notify + rescue => error + @log.error error.to_s + @log.error_backtrace + @notify_pending + ensure + @async_pending = false + @notify_pending = false + @pending_position = nil + end + end + end + private def limit_bytes_per_second_reached? @@ -195,6 +234,14 @@ def should_shutdown_now? end def handle_notify + if @async_pending + @notify_pending = true + return + end + + # A notification may arrive while a rejected batch is waiting for capacity. + @pending_read_more = true unless @lines.empty? + if limit_bytes_per_second_reached? || group_watcher&.limit_lines_reached?(@path) @metrics.throttled.inc return @@ -247,10 +294,19 @@ def handle_notify if @lines.empty? @watcher.pe.update_pos(io.pos - @fifo.reading_bytesize) if has_skipped_line else - if @receive_lines.call(@lines, @watcher) + received = @receive_lines.call(@lines, @watcher) + if received == ASYNC_PENDING + @pending_position = io.pos - @fifo.reading_bytesize + @pending_read_more ||= read_more + @async_pending = true + read_more = false + elsif received @watcher.pe.update_pos(io.pos - @fifo.reading_bytesize) @lines.clear + @pending_read_more = false else + # A rejected worker submission must retain the continuation for retry. + @pending_read_more ||= read_more read_more = false end end @@ -321,6 +377,14 @@ def opened? def eof? true end + + def drained? + true + end + + def pending? + false + end end end end diff --git a/lib/fluent/plugin/in_tail/tail_watcher.rb b/lib/fluent/plugin/in_tail/tail_watcher.rb index 857b9e6fa6..73143c008b 100644 --- a/lib/fluent/plugin/in_tail/tail_watcher.rb +++ b/lib/fluent/plugin/in_tail/tail_watcher.rb @@ -36,9 +36,15 @@ def initialize(target_info, pe, log, read_from_head, follow_inodes, update_watch @io_handler_build = io_handler_build @metrics = metrics @watchers = [] + @detached = false + @close_pending = false end attr_reader :path, :ino + + def detached? + @detached + end attr_reader :pe attr_reader :file_feed attr_accessor :unwatched # This is used for removing position entry from PositionFile @@ -54,6 +60,7 @@ def register_watcher(watcher) end def detach(shutdown_start_time = nil) + @detached = true if @io_handler ready_to_shutdown(shutdown_start_time) @io_handler.on_notify @@ -67,17 +74,52 @@ def ready_to_shutdown(shutdown_start_time) end def close + if @io_handler&.pending? || @close_pending + @close_pending = true + return + end + + force_close + end + + def force_close + @close_pending = false if @io_handler @io_handler.close @io_handler = nil end end + def defer_close + @close_pending = true + end + + def close_if_ready(force: false) + return false unless @close_pending && @io_handler && !@io_handler.pending? + return false unless force || @io_handler.drained? + + force_close + true + end + def eof? @io_handler.nil? || @io_handler.eof? end + def drained? + @io_handler.nil? || @io_handler.drained? + end + + def pending? + @io_handler&.pending? || false + end + def on_notify + if @io_handler&.pending? + @io_handler.on_notify + return + end + begin stat = Fluent::FileWrapper.stat(@path) rescue Errno::ENOENT, Errno::EACCES @@ -89,6 +131,19 @@ def on_notify read_more end + def complete_async(success, resume: true) + return false unless @io_handler + + should_notify = @io_handler.complete_async(success) + if @detached + read_more if resume && (should_notify || success) + close_if_ready + else + on_notify if resume && should_notify + end + should_notify + end + def read_more @file_feed&.on_notify(self) @io_handler.on_notify if @io_handler diff --git a/lib/fluent/plugin/in_tail/worker_pool.rb b/lib/fluent/plugin/in_tail/worker_pool.rb new file mode 100644 index 0000000000..a303ef3787 --- /dev/null +++ b/lib/fluent/plugin/in_tail/worker_pool.rb @@ -0,0 +1,298 @@ +# +# Fluentd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +require 'cool.io' +require 'socket' +require 'fluent/env' +require 'fluent/plugin/input' + +module Fluent::Plugin + class TailInput < Fluent::Plugin::Input + class WorkerPool + class WorkerStopped < StandardError; end + Task = Struct.new(:key, :bytes, :payload) + Result = Struct.new(:task, :value, :error) + + def initialize(num_threads:, task_limit:, byte_limit:, thread_create:, notify:, &process) + unless num_threads > 0 && task_limit > 0 && byte_limit > 0 + raise ArgumentError, 'worker and queue limits must be positive' + end + raise ArgumentError, 'processor is required' unless process + + @num_threads = num_threads + @task_limit = task_limit + @byte_limit = byte_limit + @thread_create = thread_create + @notify = notify + @process = process + @mutex = Mutex.new + @jobs = Queue.new + @results = Queue.new + @active = {} + @bytes = 0 + @regular_bytes = 0 + @oversized_task = nil + @threads = [] + @state = :new + @failure = nil + end + + def start + @mutex.synchronize do + raise 'worker pool already started' unless @state == :new + @state = :starting + end + begin + @num_threads.times do |index| + @threads << @thread_create.call(:"in_tail_worker_#{index}") { run(index) } + end + @mutex.synchronize do + raise(@failure || WorkerStopped.new('pool stopped during start')) unless @state == :starting + @state = :running + end + self + rescue + stop + join + raise + end + end + + # Capacity remains reserved until emit and position handling acknowledge + # the result, including while a parsed batch waits in the result queue. + def submit(key, payload, bytes:) + raise ArgumentError, 'batch bytes must be nonnegative' if bytes < 0 + + @mutex.synchronize do + return false unless @state == :running + return false if @active.key?(key) || @active.size >= @task_limit + # A single over-limit batch may use an extra reservation alongside the + # bounded regular budget so it does not force parsing on the event loop. + if bytes > @byte_limit + return false if @oversized_task + else + return false if @regular_bytes + bytes > @byte_limit + end + + task = Task.new(key, bytes, payload).freeze + @active[key] = task + @bytes += bytes + if bytes > @byte_limit + @oversized_task = task + else + @regular_bytes += bytes + end + @jobs << task + true + end + end + + def next_result + @results.pop(true) + rescue ThreadError + nil + end + + def acknowledge(result) + @mutex.synchronize do + task = result.task + unless @active[task.key].equal?(task) + raise ArgumentError, 'result is no longer active' + end + + @active.delete(task.key) + @bytes -= task.bytes + if task.bytes > @byte_limit + @oversized_task = nil + else + @regular_bytes -= task.bytes + end + end + end + + def statistics + @mutex.synchronize { { tasks: @active.size, bytes: @bytes, state: @state }.freeze } + end + + def failure + @mutex.synchronize { @failure } + end + + def stop + @mutex.synchronize do + @state = :stopping unless @state == :failed + @jobs.close + end + end + + # The source joins after draining results; the event loop never joins. + def join(timeout: nil) + deadline = timeout && Process.clock_gettime(Process::CLOCK_MONOTONIC) + timeout + @threads.each do |thread| + remaining = deadline && [deadline - Process.clock_gettime(Process::CLOCK_MONOTONIC), 0].max + return false unless thread.join(remaining) + end + true + end + + def kill + @threads.each(&:kill) + @threads.each(&:join) + end + + private + + def run(index) + normal_exit = false + while (task = @jobs.pop) + value = nil + error = nil + completed = false + begin + value = @process.call(index, task.payload) + completed = true + rescue Exception => e + # Deliver task failures to the owner instead of losing an active + # file or terminating a helper thread with abort_on_exception. + error = e + completed = true + ensure + error = WorkerStopped.new('worker stopped during a task') unless completed + publish(Result.new(task, value, error).freeze) + end + end + normal_exit = true + ensure + fail_pool(WorkerStopped.new('worker exited unexpectedly')) unless normal_exit + end + + def publish(result) + @results << result + begin + @notify.call + rescue Exception => error + fail_pool(error) + end + end + + def fail_pool(error) + @mutex.synchronize do + @failure ||= error + @state = :failed + @jobs.close + end + begin + while (task = @jobs.pop(true)) + @results << Result.new(task, nil, error).freeze + end + rescue ThreadError + # All queued jobs have been accounted for. + end + begin + @notify.call + rescue Exception => notification_error + @mutex.synchronize { @failure ||= notification_error } + end + end + end + + # Captures the built-in hook implementations once and checks whether a + # TailInput instance resolves each hook to the same method body. A mismatch + # means worker execution must fall back to the synchronous compatibility + # path; this deliberately does not try to infer safety from method owners. + class WorkerCompatibility + HOOKS = %i[ + receive_lines flush_buffer convert_line_to_event parse_singleline + parse_multilines setup_watcher io_handler build_line_feeder + ].freeze + + def initialize(base_class) + @hooks = HOOKS.to_h { |name| [name, base_class.instance_method(name)] }.freeze + end + + def compatible?(plugin) + incompatible_hooks(plugin).empty? + end + + def incompatible_hooks(plugin) + @hooks.filter_map do |name, built_in| + name unless plugin.singleton_class.instance_method(name) == built_in + rescue NameError + name + end + end + end + + class WorkerCompletionWatcher < Coolio::IOWatcher + def initialize(log: nil, &on_completion) + @log = log + if Fluent.windows? + # Winsock select cannot monitor IO.pipe handles. Use connected + # loopback sockets, as libev does for its own Windows wakeups. + begin + listener = TCPServer.new('127.0.0.1', 0) + @writer = TCPSocket.new('127.0.0.1', listener.addr[1]) + @reader = listener.accept + rescue + @reader&.close + @writer&.close + raise + ensure + listener&.close + end + else + @reader, @writer = IO.pipe + end + super(@reader) + @on_completion = on_completion + end + + def signal + @writer.write_nonblock("\0") + true + rescue IO::WaitWritable + # A full pipe already has a notification pending on the event loop. + false + end + + def on_signal + @on_completion.call + end + + def on_readable + notified = false + begin + loop do + @reader.read_nonblock(4096) + notified = true + end + rescue IO::WaitReadable + return unless notified + end + on_signal + rescue => error + @log&.error 'unexpected error while processing in_tail worker completions', error: error + @log&.error_backtrace + end + + def close + detach if attached? + @reader.close unless @reader.closed? + @writer.close unless @writer.closed? + end + end + end +end diff --git a/tasks/benchmark/in_tail_fairness.rb b/tasks/benchmark/in_tail_fairness.rb new file mode 100644 index 0000000000..85ff5bbc5f --- /dev/null +++ b/tasks/benchmark/in_tail_fairness.rb @@ -0,0 +1,184 @@ +# Run with: bundle exec ruby -Ilib tasks/benchmark/in_tail_fairness.rb +# Generates real JSONL files and measures how late the event loop dispatches a +# second-file update while a large file is being drained. This targets watcher +# latency, not total throughput; it isolates read/parse scheduling from output. + +require 'cool.io' +require 'json' +require 'tmpdir' +require 'timeout' +require 'fluent/plugin/in_tail/worker_pool' + +module TailFairnessBenchmark + LARGE_MB = Integer(ENV.fetch('LARGE_MB', '100')) + SHORT_MB = Integer(ENV.fetch('SHORT_MB', '1')) + WORKERS = Integer(ENV.fetch('WORKERS', '4')) + BATCH_LINES = Integer(ENV.fetch('BATCH_LINES', '1000')) + REPEATS = Integer(ENV.fetch('REPEATS', '3')) + UPDATE_AFTER = Float(ENV.fetch('UPDATE_AFTER', '0.05')) + LINE = JSON.generate('message' => 'x' * 80) + "\n" + + class OneShotTimer < Coolio::TimerWatcher + def initialize(delay, &callback) + @callback = callback + super(delay, false) + end + + def on_timer + @callback.call + end + end + + def self.create_file(path, megabytes) + target_bytes = megabytes * 1024 * 1024 + block = LINE * 4096 + File.open(path, 'wb') do |file| + written = 0 + while written < target_bytes + file.write(block) + written += block.bytesize + end + end + end + + def self.read_batch(io) + lines = [] + BATCH_LINES.times do + line = io.gets + break unless line + lines << line + end + lines.empty? ? nil : lines + end + + def self.parse(lines) + lines.each { |line| JSON.parse(line).fetch('message') } + lines.size + end + + def self.measure(mode, large_path, short_path) + loop = Coolio::Loop.new + large_io = File.open(large_path, 'rb') + short_io = File.open(short_path, 'rb') + started = Process.clock_gettime(Process::CLOCK_MONOTONIC) + update_due = started + UPDATE_AFTER + dispatch_latency = nil + first_short_latency = nil + large_records = 0 + short_records = 0 + pool = nil + completion_watcher = nil + submit_large = nil + submit_short = nil + + update_timer = OneShotTimer.new(UPDATE_AFTER) do + dispatch_latency = Process.clock_gettime(Process::CLOCK_MONOTONIC) - update_due + lines = read_batch(short_io) + if mode == :sync + short_records += parse(lines) if lines + first_short_latency = Process.clock_gettime(Process::CLOCK_MONOTONIC) - started + loop.stop + elsif lines + submit_short.call(lines) + else + first_short_latency = Process.clock_gettime(Process::CLOCK_MONOTONIC) - started + loop.stop + end + end + update_timer.attach(loop) + + kickoff = OneShotTimer.new(0.001) do + if mode == :sync + while (lines = read_batch(large_io)) + large_records += parse(lines) + end + else + submit_large.call + end + end + kickoff.attach(loop) + + if mode == :workers + completion_watcher = Fluent::Plugin::TailInput::WorkerCompletionWatcher.new do + while (result = pool.next_result) + raise result.error if result.error + if result.task.key == :large + large_records += result.value + pool.acknowledge(result) + submit_large.call + else + short_records += result.value + pool.acknowledge(result) + first_short_latency = Process.clock_gettime(Process::CLOCK_MONOTONIC) - started + loop.stop if loop.instance_variable_get(:@running) + end + end + end + completion_watcher.attach(loop) + pool = Fluent::Plugin::TailInput::WorkerPool.new( + num_threads: WORKERS, task_limit: WORKERS, byte_limit: WORKERS * BATCH_LINES * LINE.bytesize, + thread_create: ->(title, &block) { Thread.new { Thread.current.name = title.to_s; block.call } }, + notify: completion_watcher.method(:signal) + ) { |_worker, lines| parse(lines) } + pool.start + submit_large = lambda do + lines = read_batch(large_io) + if lines + raise 'could not enqueue large-file batch' unless pool.submit(:large, lines, bytes: lines.sum(&:bytesize)) + end + end + submit_short = lambda do |lines| + raise 'could not enqueue short-file batch' unless pool.submit(:short, lines, bytes: lines.sum(&:bytesize)) + end + end + + Timeout.timeout(600) { loop.run } + { + dispatch: dispatch_latency, + first_short: first_short_latency, + large_records: large_records, + short_records: short_records, + elapsed: Process.clock_gettime(Process::CLOCK_MONOTONIC) - started, + } + ensure + pool&.stop + pool&.join + completion_watcher&.close + large_io&.close unless large_io&.closed? + short_io&.close unless short_io&.closed? + end + + def self.main + unless LARGE_MB.positive? && SHORT_MB.positive? && WORKERS.positive? && BATCH_LINES.positive? && + REPEATS.positive? && UPDATE_AFTER.positive? + raise 'file sizes, worker count, batch size, and update delay must be positive' + end + + Dir.mktmpdir('in-tail-fairness') do |directory| + large_path = File.join(directory, 'large.jsonl') + short_path = File.join(directory, 'updated.jsonl') + puts "Generating #{LARGE_MB} MiB large file and #{SHORT_MB} MiB updated file..." + create_file(large_path, LARGE_MB) + create_file(short_path, SHORT_MB) + puts RUBY_DESCRIPTION + puts "large=#{LARGE_MB}MiB, updated=#{SHORT_MB}MiB, workers=#{WORKERS}, batch_lines=#{BATCH_LINES}, update_after=#{UPDATE_AFTER}s, repeats=#{REPEATS}" + puts 'mode,update_dispatch_lag_seconds,first_short_batch_seconds,large_records_before_update,elapsed_seconds' + + samples = { sync: [], workers: [] } + REPEATS.times do |index| + (index.even? ? %i[sync workers] : %i[workers sync]).each do |mode| + samples[mode] << measure(mode, large_path, short_path) + end + end + samples.each do |mode, results| + median = lambda do |key| + results.map { |result| result[key] }.sort[REPEATS / 2] + end + puts format('%s,%.4f,%.4f,%d,%.4f', mode, median.call(:dispatch), median.call(:first_short), + median.call(:large_records), median.call(:elapsed)) + end + end + end +end + +TailFairnessBenchmark.main diff --git a/tasks/benchmark/in_tail_parse_threads.rb b/tasks/benchmark/in_tail_parse_threads.rb new file mode 100644 index 0000000000..bebd24bdda --- /dev/null +++ b/tasks/benchmark/in_tail_parse_threads.rb @@ -0,0 +1,185 @@ +# Run with: bundle exec ruby -Ilib tasks/benchmark/in_tail_parse_threads.rb +# MODE=feed also offloads emit. EMIT_WAIT adds a synthetic blocking collector. +# This does not measure file IO or real downstream plugins. + +require 'json' +require 'stringio' +require 'serverengine' +require 'fluent/engine' +require 'fluent/log' +require 'fluent/plugin/in_tail/line_feeder' +require 'fluent/plugin/in_tail/worker_pool' +require 'fluent/plugin/parser_none' +require 'fluent/plugin/parser_json' +require 'fluent/plugin/parser_regexp' + +$log = Fluent::Log.new(ServerEngine::DaemonLogger.new(StringIO.new)) +Fluent::Engine.init(Fluent::SystemConfig.new) + +module TailParseThreadsBenchmark + Watcher = Struct.new(:path, :tag, :file_feed) + BATCH_SIZE = 1000 + BATCHES = Integer(ENV.fetch('BATCHES', '200')) + REPEATS = Integer(ENV.fetch('REPEATS', '3')) + FILES = Integer(ENV.fetch('FILES', '4')) + MODE = ENV.fetch('MODE', 'parse') + EMIT_WAIT = Float(ENV.fetch('EMIT_WAIT', '0')) + + class Router + attr_reader :count + + def initialize + @count = 0 + @thread = Thread.current + @mutex = Mutex.new + end + + def emit_stream(tag, events) + if MODE == 'parse' && Thread.current != @thread + raise 'emit moved off the caller thread' + end + + sleep EMIT_WAIT if EMIT_WAIT > 0 + @mutex.synchronize { @count += events.size } + end + end + + def self.create_feeder(type, router) + params = { '@type' => type } + params['expression'] = '^(?.+)$' if type == 'regexp' + parser = Fluent::Plugin.new_parser(type) + parser.configure(Fluent::Config::Element.new('parse', '', params, [])) + parser.start + parser.after_start + feeder = Fluent::Plugin::TailInput::LineFeeder.new( + parser: parser, router_provider: -> { router }, log: $log, + tag: 'tail', tag_prefix: nil, tag_suffix: nil, path_key: nil, + emit_unmatched_lines: false, multiline_mode: false + ) + [feeder, parser] + end + + def self.close_parser(parser) + %i[stop before_shutdown shutdown after_shutdown close terminate].each do |method| + parser.public_send(method) + end + end + + def self.run(type, line, num_threads) + router = Router.new + resources = Array.new(num_threads) { create_feeder(type, router) } + feeders = resources.map(&:first) + batches = Array.new(BATCHES) { Array.new(BATCH_SIZE) { line.dup } } + file_batches = Array.new(FILES) { [] } + batches.each_with_index { |batch, index| file_batches[index % FILES] << batch } + watchers = Array.new(FILES) do |index| + Watcher.new("benchmark-#{index}.log", "benchmark.#{index}", feeders.first.new_file_feed) + end + ready = Queue.new + notifications = Queue.new + pool = nil + + if num_threads > 1 + pool = Fluent::Plugin::TailInput::WorkerPool.new( + num_threads: num_threads, task_limit: num_threads * 2, + byte_limit: num_threads * 2 * BATCH_SIZE * line.bytesize, + thread_create: ->(_title, &block) { + Thread.new { ready << true; block.call } + }, notify: -> { notifications << true } + ) do |index, payload| + file, batch = payload + if MODE == 'parse' + feeders[index].parse(batch, watchers[file]) + else + feeders[index].feed_lines(batch, watchers[file]) + end + end + pool.start + num_threads.times { ready.pop } + end + + GC.start + started = Process.clock_gettime(Process::CLOCK_MONOTONIC) + if num_threads == 1 + file_batches.each_with_index do |inputs, file| + inputs.each { |batch| feeders.first.feed_lines(batch, watchers[file]) } + end + else + completed = 0 + busy = Array.new(FILES, false) + positions = Array.new(FILES, 0) + while completed < batches.size + FILES.times do |file| + next if busy[file] || positions[file] >= file_batches[file].size + + batch = file_batches[file][positions[file]] + if pool.submit(file, [file, batch], bytes: batch.sum(&:bytesize)) + positions[file] += 1 + busy[file] = true + end + end + completion = pool.next_result + unless completion + notifications.pop + next + end + file = completion.task.key + result = completion.value + raise completion.error if completion.error + raise 'completion without an active file' unless busy[file] + + if MODE == 'parse' + feeders.first.emit(result, watchers[file]) + else + raise 'emit failed' unless result + end + pool.acknowledge(completion) + busy[file] = false + completed += 1 + end + end + elapsed = Process.clock_gettime(Process::CLOCK_MONOTONIC) - started + raise "missing events: #{router.count}" unless router.count == BATCHES * BATCH_SIZE + + elapsed + ensure + pool&.stop + pool&.join + resources&.each { |_, parser| close_parser(parser) } + end + + def self.main + raise 'MODE must be parse or feed' unless %w[parse feed].include?(MODE) + unless BATCHES > 0 && REPEATS > 0 && FILES > 0 + raise 'BATCHES, REPEATS and FILES must be positive' + end + raise 'EMIT_WAIT must be nonnegative' if EMIT_WAIT < 0 + + workloads = { + 'none' => "#{'message ' * 32}\n", + 'regexp' => "#{'message ' * 32}\n", + 'json' => "#{JSON.generate('time' => 1234567890, 'message' => 'message ' * 32, 'values' => (1..32).to_a)}\n", + } + puts RUBY_DESCRIPTION + puts "#{FILES} files, #{BATCHES} batches x #{BATCH_SIZE} lines, #{REPEATS} repeats (median)" + puts "mode=#{MODE}, synthetic collector wait=#{EMIT_WAIT}s/batch" + puts 'parser,threads,seconds,records_per_second,speedup' + workloads.each do |type, line| + samples = { 1 => [], 4 => [] } + REPEATS.times do |index| + # Alternate run order to reduce warm-up/order bias. + (index.even? ? [1, 4] : [4, 1]).each do |threads| + samples[threads] << run(type, line, threads) + end + end + baseline = samples[1].sort[REPEATS / 2] + samples.each do |threads, times| + elapsed = times.sort[REPEATS / 2] + puts format('%s,%d,%.4f,%.0f,%.3f', type, threads, elapsed, + BATCHES * BATCH_SIZE / elapsed, baseline / elapsed) + end + end + end +end + +TailParseThreadsBenchmark.main diff --git a/test/plugin/in_tail/test_compatibility.rb b/test/plugin/in_tail/test_compatibility.rb index 1a166b6550..d4c6c68297 100644 --- a/test/plugin/in_tail/test_compatibility.rb +++ b/test/plugin/in_tail/test_compatibility.rb @@ -27,6 +27,10 @@ def configure(conf) end end + class WorkerParserBuilder < RecordingInput + attr_reader :worker_parsers + end + class ReceiveLinesOverride < RecordingInput def receive_lines(lines, tail_watcher) @calls << [:receive_lines, lines] @@ -279,6 +283,47 @@ def create_driver(conf = SINGLE_LINE_CONFIG, use_common_conf = true, klass: Flue Fluent::Test::Driver::Input.new(klass).configure(config) end + sub_test_case 'worker compatibility classification' do + test 'the built-in implementation is eligible' do + config = SINGLE_LINE_CONFIG + config_element('', '', { 'num_threads' => 2 }) + driver = create_driver(config) + driver.run {} + + assert_true Fluent::Plugin::TailInput::WORKER_COMPATIBILITY.compatible?(driver.instance) + end + + test 'an inherited line-processing override uses the compatibility path' do + config = SINGLE_LINE_CONFIG + config_element('', '', { 'num_threads' => 2 }) + driver = create_driver(config, true, klass: TailInputSubclasses::ParseSinglelineOverrideWithSuper) + driver.run {} + + assert_false Fluent::Plugin::TailInput::WORKER_COMPATIBILITY.compatible?(driver.instance) + end + + test 'an inherited override stays synchronous when workers are requested' do + File.write("#{@tmp_dir}/tail.txt", "one\ntwo\n") + config = config_element('', '', { 'read_from_head' => true }) + SINGLE_LINE_CONFIG + config_element('', '', { 'num_threads' => 2 }) + driver = create_driver(config, true, klass: TailInputSubclasses::ParseSinglelineOverrideWithSuper) + driver.run(expect_emits: 2, timeout: 5) {} + + assert_nil driver.instance.instance_variable_get(:@worker_pool) + assert_equal %w[one two], driver.events.map { |event| event[2]['message'] } + assert_true driver.logs.any? { |line| line.include?('overridden hook: parse_singleline') } + end + + test 'worker parsers are recreated from the configured parser and joined to parser lifecycle' do + config = SINGLE_LINE_CONFIG + config_element('', '', { 'num_threads' => 2 }) + driver = create_driver(config, true, klass: TailInputSubclasses::WorkerParserBuilder) + driver.run {} + + plugin = driver.instance + assert_equal 2, plugin.worker_parsers.size + assert_not_same plugin.instance_variable_get(:@parser), plugin.worker_parsers[0] + assert_not_same plugin.worker_parsers[0], plugin.worker_parsers[1] + assert_true plugin.worker_parsers.all? { |parser| parser.class == plugin.instance_variable_get(:@parser).class } + end + end + sub_test_case "receive_lines compatibility" do # Watcher used with a FileFeed. DummyWatcher = Struct.new("DummyWatcher", :tag, :file_feed) diff --git a/test/plugin/in_tail/test_io_handler.rb b/test/plugin/in_tail/test_io_handler.rb index 65ae45b688..7a1456dfd1 100644 --- a/test/plugin/in_tail/test_io_handler.rb +++ b/test/plugin/in_tail/test_io_handler.rb @@ -62,6 +62,230 @@ def create_watcher assert_equal text, returned_lines end + test 'an asynchronous receive commits position only after successful completion' do + text = "first\nsecond\n" + @file.write(text) + @file.close + watcher = create_watcher + position = 0 + stub(watcher).pe do + pe = 'position_file' + stub(pe).read_pos { position } + stub(pe).update_pos { |value| position = value } + pe + end + returned_lines = nil + handler = Fluent::Plugin::TailInput::TailWatcher::IOHandler.new( + watcher, path: @file.path, read_lines_limit: 1, read_bytes_limit_per_second: -1, + log: $log, open_on_every_update: false, metrics: @metrics + ) do |lines, _tail_watcher| + returned_lines = lines.dup + Fluent::Plugin::TailInput::TailWatcher::IOHandler::ASYNC_PENDING + end + + handler.on_notify + assert_true handler.pending? + assert_equal 0, position + assert_equal ["first\n", "second\n"], returned_lines + assert_true handler.complete_async(true) + assert_equal text.bytesize, position + assert_false handler.pending? + end + + test 'a batch rejected by a full queue resumes reading after asynchronous retry' do + text = Array.new(2000) { |index| "#{index} #{'x' * 100}\n" }.join + @file.write(text) + @file.close + watcher = create_watcher + calls = 0 + received = [] + handler = Fluent::Plugin::TailInput::TailWatcher::IOHandler.new( + watcher, path: @file.path, read_lines_limit: 1, read_bytes_limit_per_second: -1, + log: $log, open_on_every_update: false, metrics: @metrics + ) do |lines, _tail_watcher| + calls += 1 + if calls == 1 + false + else + received.concat(lines) + Fluent::Plugin::TailInput::TailWatcher::IOHandler::ASYNC_PENDING + end + end + + handler.on_notify + assert_false handler.pending? + handler.on_notify + assert_true handler.pending? + assert_true handler.complete_async(true) + + until handler.drained? + handler.on_notify + handler.complete_async(true) if handler.pending? + end + assert_equal text.lines, received + assert_equal text.bytesize, watcher.pe.read_pos + ensure + handler&.close + end + + test 'a notification while an EOF batch is parked resumes reads after completion' do + @file.write("first\n") + @file.close + watcher = create_watcher + calls = 0 + received = [] + handler = Fluent::Plugin::TailInput::TailWatcher::IOHandler.new( + watcher, path: @file.path, read_lines_limit: 100, read_bytes_limit_per_second: -1, + log: $log, open_on_every_update: false, metrics: @metrics + ) do |lines, _tail_watcher| + calls += 1 + if calls <= 2 + false + else + received.concat(lines) + Fluent::Plugin::TailInput::TailWatcher::IOHandler::ASYNC_PENDING + end + end + + handler.on_notify + assert_true handler.eof? + assert_false handler.pending? + File.open(@file.path, 'ab') { |file| file.puts('last') } + handler.on_notify + assert_false handler.pending? + handler.on_notify + assert_true handler.pending? + assert_true handler.complete_async(true) + + handler.on_notify + assert_true handler.pending? + assert_false handler.complete_async(true) + assert_true handler.drained? + assert_equal ["first\n", "last\n"], received + assert_equal 11, watcher.pe.read_pos + ensure + handler&.close + end + + test 'synchronous retry preserves the existing per-notification read behavior' do + text = "#{'x' * 1000}\n" * 200 + @file.write(text) + @file.close + watcher = create_watcher + calls = 0 + handler = Fluent::Plugin::TailInput::TailWatcher::IOHandler.new( + watcher, path: @file.path, read_lines_limit: 1, read_bytes_limit_per_second: -1, + log: $log, open_on_every_update: false, metrics: @metrics + ) do + calls += 1 + calls > 1 + end + + handler.on_notify + assert_equal 0, watcher.pe.read_pos + handler.on_notify + assert_operator watcher.pe.read_pos, :>, 0 + assert_operator watcher.pe.read_pos, :<, text.bytesize + assert_equal 2, calls + ensure + handler&.close + end + + test 'a failed asynchronous position write clears pending state and retains the batch' do + @file.write("line\n") + @file.close + watcher = create_watcher + position_entry = watcher.pe + writes = 0 + original_update = position_entry.method(:update_pos) + position_entry.define_singleton_method(:update_pos) do |position| + writes += 1 + raise Errno::ENOSPC if writes == 1 + original_update.call(position) + end + batches = [] + handler = Fluent::Plugin::TailInput::TailWatcher::IOHandler.new( + watcher, path: @file.path, read_lines_limit: 1, read_bytes_limit_per_second: -1, + log: $log, open_on_every_update: false, metrics: @metrics + ) do |lines, _tail_watcher| + batches << lines.dup + Fluent::Plugin::TailInput::TailWatcher::IOHandler::ASYNC_PENDING + end + + handler.on_notify + assert_nothing_raised { handler.complete_async(true) } + assert_false handler.pending? + assert_equal 0, position_entry.read_pos + handler.on_notify + assert_true handler.pending? + handler.complete_async(true) + assert_false handler.pending? + assert_equal 5, position_entry.read_pos + assert_equal [["line\n"], ["line\n"]], batches + ensure + handler&.close + end + + test 'a final force-close releases a file even while its batch is pending' do + @file.write("line\n") + @file.close + watcher = create_watcher + handler = Fluent::Plugin::TailInput::TailWatcher::IOHandler.new( + watcher, path: @file.path, read_lines_limit: 1, read_bytes_limit_per_second: -1, + log: $log, open_on_every_update: false, metrics: @metrics + ) do + Fluent::Plugin::TailInput::TailWatcher::IOHandler::ASYNC_PENDING + end + watcher.instance_variable_set(:@io_handler, handler) + handler.on_notify + watcher.close + assert_true handler.opened? + + watcher.force_close + assert_false handler.opened? + assert_false watcher.pending? + ensure + handler&.close + end + + test 'a null handler is drained without pending work' do + handler = Fluent::Plugin::TailInput::TailWatcher::NullIOHandler.new + assert_true handler.drained? + assert_false handler.pending? + end + + test 'a failed asynchronous receive keeps the batch and position for retry' do + text = "line\n" + @file.write(text) + @file.close + watcher = create_watcher + position = 0 + stub(watcher).pe do + pe = 'position_file' + stub(pe).read_pos { position } + stub(pe).update_pos { |value| position = value } + pe + end + calls = 0 + handler = Fluent::Plugin::TailInput::TailWatcher::IOHandler.new( + watcher, path: @file.path, read_lines_limit: 10, read_bytes_limit_per_second: -1, + log: $log, open_on_every_update: false, metrics: @metrics + ) do + calls += 1 + calls == 1 ? Fluent::Plugin::TailInput::TailWatcher::IOHandler::ASYNC_PENDING : true + end + + handler.on_notify + handler.on_notify + assert_true handler.complete_async(false) + assert_equal 0, position + assert_false handler.pending? + handler.on_notify + + assert_equal 2, calls + assert_equal text.bytesize, position + end + sub_test_case 'when open_on_every_update is true and read_pos returns always 0' do test 'open new IO and change pos to 0 and read it' do text = "this line is test\ntest line is test\n" diff --git a/test/plugin/in_tail/test_worker_pool.rb b/test/plugin/in_tail/test_worker_pool.rb new file mode 100644 index 0000000000..858915fa12 --- /dev/null +++ b/test/plugin/in_tail/test_worker_pool.rb @@ -0,0 +1,353 @@ +require_relative '../../helper' +require 'fluent/plugin/in_tail/worker_pool' +require 'timeout' + +class TailInputWorkerPoolTest < Test::Unit::TestCase + def setup + @pools = [] + @watchers = [] + @release = Queue.new + end + + def teardown + 10.times { @release << true } + @pools.each do |pool| + pool.stop + assert_true pool.join(timeout: 5) + end + @watchers.each(&:close) + super + end + + def create_pool(task_limit: 2, byte_limit: 10, num_threads: 1, &process) + pool = Fluent::Plugin::TailInput::WorkerPool.new( + num_threads: num_threads, task_limit: task_limit, byte_limit: byte_limit, + thread_create: ->(_title, &block) { Thread.new(&block) }, + notify: -> {}, &process + ) + @pools << pool + pool.start + end + + def result_from(pool) + Timeout.timeout(5) do + loop do + result = pool.next_result + return result if result + Thread.pass + end + end + end + + test 'file and capacity remain reserved until the owner acknowledges emit' do + pool = create_pool(task_limit: 1) { |_, lines| lines.upcase } + assert_true pool.submit(:file, 'first', bytes: 5) + result = result_from(pool) + + assert_equal 'FIRST', result.value + assert_false pool.submit(:file, 'second', bytes: 1) + assert_false pool.submit(:other_file, 'other', bytes: 1) + assert_equal({ tasks: 1, bytes: 5, state: :running }, pool.statistics) + + pool.acknowledge(result) + assert_true pool.submit(:file, 'second', bytes: 1) + assert_raise(ArgumentError) { pool.acknowledge(result) } + end + + test 'worker compatibility requires every hook to resolve to the captured implementation' do + base = Class.new do + Fluent::Plugin::TailInput::WorkerCompatibility::HOOKS.each do |name| + define_method(name) {} + end + private :io_handler + end + compatibility = Fluent::Plugin::TailInput::WorkerCompatibility.new(base) + plugin = Class.new(base).new + assert_true compatibility.compatible?(plugin) + + plugin_class = Class.new(base) { def receive_lines; end } + assert_false compatibility.compatible?(plugin_class.new) + + private_override = Class.new(base) { private; def io_handler; end } + assert_false compatibility.compatible?(private_override.new) + + prepended = Class.new(base) + prepended.prepend(Module.new { def parse_singleline; end }) + assert_false compatibility.compatible?(prepended.new) + + singleton_override = Class.new(base).new + def singleton_override.build_line_feeder; end + assert_false compatibility.compatible?(singleton_override) + end + + test 'byte capacity includes queued and completed batches' do + pool = create_pool { |_, payload| @release.pop; payload } + assert_true pool.submit(:first, 'first', bytes: 7) + assert_false pool.submit(:second, 'second', bytes: 4) + assert_true pool.submit(:second, 'second', bytes: 3) + assert_false pool.submit(:third, 'third', bytes: 0) + 2.times { @release << true } + + first, second = result_from(pool), result_from(pool) + assert_equal 10, pool.statistics[:bytes] + pool.acknowledge(first) + pool.acknowledge(second) + assert_equal 0, pool.statistics[:bytes] + assert_true pool.submit(:too_large, 'oversized', bytes: 11) + @release << true + pool.acknowledge(result_from(pool)) + end + + test 'one oversized batch uses a separate reservation beside regular capacity' do + started = Queue.new + pool = create_pool(task_limit: 3, byte_limit: 10, num_threads: 2) do |_, payload| + started << payload + @release.pop + payload + end + assert_true pool.submit(:large, :oversized, bytes: 11) + assert_equal :oversized, Timeout.timeout(5) { started.pop } + assert_true pool.submit(:small, :regular, bytes: 10) + assert_equal :regular, Timeout.timeout(5) { started.pop } + assert_false pool.submit(:another_large, :oversized, bytes: 12) + assert_false pool.submit(:another_small, :regular, bytes: 1) + assert_equal({ tasks: 2, bytes: 21, state: :running }, pool.statistics) + + 2.times { @release << true } + results = Array.new(2) { result_from(pool) } + assert_equal %i[oversized regular], results.map(&:value).sort + results.each { |result| pool.acknowledge(result) } + assert_equal({ tasks: 0, bytes: 0, state: :running }, pool.statistics) + end + + test 'a failed task produces an error result and the worker handles the next file' do + pool = create_pool { |_, payload| raise Interrupt, 'task failed' if payload == :fail; payload } + assert_true pool.submit(:file, :fail, bytes: 1) + result = result_from(pool) + assert_kind_of Interrupt, result.error + pool.acknowledge(result) + + assert_true pool.submit(:file, :next, bytes: 1) + result = result_from(pool) + assert_nil result.error + assert_equal :next, result.value + pool.acknowledge(result) + end + + test 'stop drains accepted work and rejects new batches' do + pool = create_pool { |_, payload| @release.pop; payload } + assert_true pool.submit(:first, 'first', bytes: 1) + assert_true pool.submit(:second, 'second', bytes: 1) + pool.stop + assert_false pool.submit(:third, 'third', bytes: 1) + assert_false pool.join(timeout: 0) + + 2.times { @release << true } + 2.times { pool.acknowledge(result_from(pool)) } + assert_true pool.join(timeout: 5) + assert_equal 0, pool.statistics[:tasks] + end + + test 'killing a blocked worker publishes failures for active and queued tasks' do + started = Queue.new + pool = create_pool do |_, payload| + started << true + @release.pop + payload + end + assert_true pool.submit(:active, 'active', bytes: 1) + Timeout.timeout(5) { started.pop } + assert_true pool.submit(:queued, 'queued', bytes: 1) + pool.stop + pool.kill + + results = Array.new(2) { result_from(pool) } + assert_equal %i[active queued], results.map { |result| result.task.key }.sort + results.each do |result| + assert_kind_of Fluent::Plugin::TailInput::WorkerPool::WorkerStopped, result.error + pool.acknowledge(result) + end + assert_true pool.join(timeout: 0) + assert_equal 0, pool.statistics[:tasks] + end + + test 'completion notifications are drained before invoking the callback' do + completions = 0 + watcher = Fluent::Plugin::TailInput::WorkerCompletionWatcher.new { completions += 1 } + @watchers << watcher + 100.times { watcher.signal } + + watcher.on_readable + assert_equal 1, completions + watcher.on_readable + assert_equal 1, completions + end + + test 'a completion callback error is logged without escaping the watcher' do + completions = 0 + watcher = Fluent::Plugin::TailInput::WorkerCompletionWatcher.new(log: $log) do + completions += 1 + raise Errno::ENOSPC if completions == 1 + end + @watchers << watcher + + watcher.signal + assert_nothing_raised { watcher.on_readable } + watcher.signal + watcher.on_readable + assert_equal 2, completions + end + + test 'completion notifications never block when the pipe is full' do + watcher = Fluent::Plugin::TailInput::WorkerCompletionWatcher.new {} + @watchers << watcher + Timeout.timeout(5) do + loop { break unless watcher.signal } + end + assert_false watcher.signal + end + + test 'completion watcher handles repeated wakeups and closes both endpoints' do + event_loop = Coolio::Loop.new + completions = 0 + watcher = Fluent::Plugin::TailInput::WorkerCompletionWatcher.new do + completions += 1 + event_loop.stop + end + @watchers << watcher + watcher.attach(event_loop) + + 3.times do + thread = Thread.new { watcher.signal } + Timeout.timeout(5) { event_loop.run } + thread.join + end + assert_equal 3, completions + + watcher.close + assert_false watcher.attached? + assert_true watcher.instance_variable_get(:@reader).closed? + assert_true watcher.instance_variable_get(:@writer).closed? + assert_nothing_raised { watcher.close } + end + + test 'different files run concurrently while each file stays reserved' do + large_started = Queue.new + large_finished = Queue.new + short_completed = Queue.new + pool = create_pool(task_limit: 3, num_threads: 2) do |_, payload| + if payload == :large + large_started << true + @release.pop + large_finished << true + else + short_completed << payload + end + payload + end + assert_true pool.submit(:large_file, :large, bytes: 1) + Timeout.timeout(5) { large_started.pop } + assert_true pool.submit(:short_file_a, :short_a, bytes: 1) + assert_true pool.submit(:short_file_b, :short_b, bytes: 1) + short_records = Timeout.timeout(5) { [short_completed.pop, short_completed.pop] } + + assert_equal [:short_a, :short_b], short_records.sort + assert_true large_finished.empty? + assert_equal 3, pool.statistics[:tasks] + assert_false pool.submit(:large_file, :later_batch, bytes: 1) + + @release << true + results = Array.new(3) { result_from(pool) } + assert_equal [:large, :short_a, :short_b], results.map(&:value).sort_by(&:to_s) + results.each { |result| pool.acknowledge(result) } + end + + test 'repeated start does not stop the running pool' do + pool = create_pool { |_, payload| payload } + assert_raise(RuntimeError) { pool.start } + assert_true pool.submit(:file, 'lines', bytes: 1) + pool.acknowledge(result_from(pool)) + end + + test 'failed thread creation cleans up workers already created' do + threads = [] + pool = Fluent::Plugin::TailInput::WorkerPool.new( + num_threads: 2, task_limit: 2, byte_limit: 10, + thread_create: ->(_title, &block) { + raise 'failed to start' unless threads.empty? + threads << Thread.new(&block) + threads.last + }, notify: -> {} + ) { |_, payload| payload } + @pools << pool + assert_raise(RuntimeError) { pool.start } + assert_true threads.all? { |thread| !thread.alive? } + assert_false pool.submit(:file, 'lines', bytes: 1) + end + + test 'unexpected worker termination reports active and queued work as failed' do + threads = Queue.new + pool = create_pool do |_, payload| + threads << Thread.current + @release.pop + payload + end + pool.submit(:active, 'active', bytes: 1) + pool.submit(:queued, 'queued', bytes: 1) + worker = Timeout.timeout(5) { threads.pop } + worker.kill + worker.join + results = [result_from(pool), result_from(pool)] + + assert_true results.all? { |result| result.error.is_a?(Fluent::Plugin::TailInput::WorkerPool::WorkerStopped) } + assert_equal :failed, pool.statistics[:state] + assert_not_nil pool.failure + assert_false pool.submit(:next, 'next', bytes: 1) + results.each { |result| pool.acknowledge(result) } + end + + test 'notification failure retains results and shuts down admissions' do + pool = Fluent::Plugin::TailInput::WorkerPool.new( + num_threads: 1, task_limit: 1, byte_limit: 10, + thread_create: ->(_title, &block) { Thread.new(&block) }, + notify: -> { raise IOError, 'notification channel closed' } + ) { |_, payload| payload } + @pools << pool + pool.start + pool.submit(:file, 'lines', bytes: 1) + result = result_from(pool) + assert_true pool.join(timeout: 5) + assert_kind_of IOError, pool.failure + assert_equal 'lines', result.value + assert_false pool.submit(:other, 'other', bytes: 1) + pool.acknowledge(result) + end + + test 'a worker completion wakes the event loop on its own thread' do + loop = Coolio::Loop.new + result = nil + callback_thread = nil + pool = nil + watcher = Fluent::Plugin::TailInput::WorkerCompletionWatcher.new do + result = pool.next_result + callback_thread = Thread.current + loop.stop + end + @watchers << watcher + watcher.attach(loop) + pool = Fluent::Plugin::TailInput::WorkerPool.new( + num_threads: 1, task_limit: 1, byte_limit: 10, + thread_create: ->(_title, &block) { Thread.new(&block) }, + notify: watcher.method(:signal) + ) { |_, payload| [Thread.current, payload] } + @pools << pool + pool.start + pool.submit(:file, 'lines', bytes: 5) + Timeout.timeout(5) { loop.run } + + assert_equal Thread.current, callback_thread + assert_not_equal callback_thread, result.value[0] + assert_equal 'lines', result.value[1] + pool.acknowledge(result) + end +end diff --git a/test/plugin/test_in_tail.rb b/test/plugin/test_in_tail.rb index 3380a741e4..f1f87b03a0 100644 --- a/test/plugin/test_in_tail.rb +++ b/test/plugin/test_in_tail.rb @@ -170,6 +170,14 @@ def create_driver(conf = SINGLE_LINE_CONFIG, use_common_conf = true, klass: Flue Fluent::Test::Driver::Input.new(klass).configure(config) end + def wrap_worker_pool_process(pool, &before_process) + original_process = pool.instance_variable_get(:@process) + pool.instance_variable_set(:@process, lambda do |index, payload| + before_process.call(payload) + original_process.call(index, payload) + end) + end + sub_test_case "configure" do test "plain single line" do d = create_driver @@ -185,6 +193,13 @@ def create_driver(conf = SINGLE_LINE_CONFIG, use_common_conf = true, klass: Flue end end + test "num_threads must be positive" do + [0, -1].each do |num_threads| + conf = SINGLE_LINE_CONFIG + config_element('', '', { 'num_threads' => num_threads }) + assert_raise(Fluent::ConfigError) { create_driver(conf) } + end + end + data('without *': { tag: 'tail', tag_prefix: nil, tag_suffix: nil }, 'only *': { tag: '*', tag_prefix: '', tag_suffix: '' }, 'prefix': { tag: 'pre.*', tag_prefix: 'pre.', tag_suffix: '' }, @@ -799,6 +814,785 @@ def test_emit_with_read_from_head(data) assert_equal({"message" => "test4"}, events[3][2]) end + def test_emit_with_worker_threads + records = Array.new(2000) { |index| format('worker%04d %s', index, 'x' * 90) } + Fluent::FileWrapper.open("#{@tmp_dir}/tail.txt", "wb") { |file| records.each { |record| file.puts(record) } } + config = CONFIG_READ_FROM_HEAD + SINGLE_LINE_CONFIG + config_element("", "", { "num_threads" => 2, "read_lines_limit" => 100 }) + driver = create_driver(config) + + driver.run(expect_records: records.size, timeout: 5) {} + + assert_equal records, driver.events.map { |event| event[2]["message"] } + assert_equal 2, driver.instance.instance_variable_get(:@worker_parsers).size + assert_equal 0, driver.instance.instance_variable_get(:@worker_pool).statistics[:tasks] + assert_true driver.instance.instance_variable_get(:@tails).values.all?(&:eof?) + assert_true driver.logs.any? { |line| line.include?('worker parsing enabled with 2 threads') } + assert_equal 0, driver.instance.statistics['input']['worker_pending_batch_count'] + assert_equal 0, driver.instance.statistics['input']['worker_pending_bytes'] + assert_equal 0, driver.instance.statistics['input']['worker_parse_error_count'] + end + + def test_worker_completion_detects_rotation_and_drains_both_files + path = "#{@tmp_dir}/tail.txt" + Fluent::FileWrapper.open(path, 'wb') {} + config = CONFIG_READ_FROM_HEAD + SINGLE_LINE_CONFIG + config_element('', '', { + 'num_threads' => 2, + 'read_lines_limit' => 1, + 'rotate_wait' => '1s', + 'refresh_interval' => '1h', + 'enable_stat_watcher' => false, + }) + driver = create_driver(config) + parse_started = Queue.new + continue_parse = Queue.new + + driver.run(expect_records: 2, timeout: 10) do + pool = driver.instance.instance_variable_get(:@worker_pool) + wrap_worker_pool_process(pool) do |batch| + if batch.lines.any? { |line| line.include?('old-file') } + parse_started << true + continue_parse.pop + end + end + + Fluent::FileWrapper.open(path, 'ab') { |file| file.puts('old-file') } + waiting(5) { sleep 0.01 until parse_started.size == 1 } + FileUtils.move(path, "#{path}.1") + Fluent::FileWrapper.open(path, 'wb') { |file| file.puts('new-file') } + continue_parse << true + end + + assert_equal %w[new-file old-file], driver.events.map { |event| event[2]['message'] }.sort + end + + def test_worker_completion_drains_file_during_unwatch + path = "#{@tmp_dir}/tail.txt" + Fluent::FileWrapper.open(path, 'wb') {} + config = CONFIG_READ_FROM_HEAD + SINGLE_LINE_CONFIG + config_element('', '', { + 'num_threads' => 2, + 'read_lines_limit' => 1, + 'enable_stat_watcher' => true, + 'enable_watch_timer' => false, + }) + driver = create_driver(config) + parse_started = Queue.new + continue_parse = Queue.new + position_entry = nil + + driver.run(expect_records: 1, timeout: 10) do + plugin = driver.instance + pool = plugin.instance_variable_get(:@worker_pool) + wrap_worker_pool_process(pool) do |batch| + if batch.lines.any? { |line| line.include?('unwatch-record') } + parse_started << true + continue_parse.pop + end + end + Fluent::FileWrapper.open(path, 'ab') { |file| file.puts('unwatch-record') } + tail_watcher = plugin.instance_variable_get(:@tails)[path] + # StatWatcher polls about every five seconds without inotify. Trigger + # only the initial read; completion must drain without a watch timer. + plugin.timer_execute(:test_worker_initial_read, 0.01, repeat: false) { tail_watcher.on_notify } + waiting(5) { sleep 0.01 until parse_started.size == 1 } + waiting(5) { sleep 0.01 until tail_watcher.pending? } + target = Fluent::Plugin::TailInput::TargetInfo.new(path, Fluent::FileWrapper.stat(path).ino) + assert_true tail_watcher.pending? + position_entry = plugin.instance_variable_get(:@pf)[target] + plugin.send(:stop_watchers, { path => target }, immediate: true, unwatched: true) + assert_true plugin.instance_variable_get(:@worker_deferred_unwatch).key?(tail_watcher) + 1.times { continue_parse << true } + ensure + continue_parse.close + end + + assert_equal ['unwatch-record'], driver.events.map { |event| event[2]['message'] } + assert_equal 0, driver.instance.instance_variable_get(:@worker_pool).statistics[:tasks] + assert_equal Fluent::Plugin::TailInput::PositionFile::UNWATCHED_POSITION, position_entry.read_pos + end + + def test_worker_batch_submitted_by_detach_keeps_its_position_entry_until_completion + paths = %w[tail-a.txt tail-b.txt].map { |name| "#{@tmp_dir}/#{name}" } + paths.each { |path| File.binwrite(path, '') } + config = SINGLE_LINE_CONFIG + config_element('', '', { + 'path' => "#{@tmp_dir}/tail-*.txt", + 'pos_file' => "#{@tmp_dir}/tail.pos", + 'tag' => 't1', + 'num_threads' => 2, + 'read_from_head' => true, + 'follow_inodes' => true, + 'enable_stat_watcher' => false, + }) + driver = create_driver(config, false) + started = Queue.new + release = Queue.new + driver.run(expect_records: 1, timeout: 5) do + plugin = driver.instance + tails = plugin.instance_variable_get(:@tails) + tails.each_value do |watcher| + watcher.watchers.each { |trigger| plugin.event_loop_detach(trigger) } + end + first, second = paths.map { |path| tails[path] } + second.pe.update_pos(130) + pool = plugin.instance_variable_get(:@worker_pool) + wrap_worker_pool_process(pool) { started << true; release.pop } + File.binwrite(paths.first, "detached-batch\n") + assert_false first.pending? + first.unwatched = true + tails.delete(first.path) + plugin.detach_watcher(first, first.ino) + waiting(5) { sleep 0.01 until started.size == 1 } + + assert_true plugin.instance_variable_get(:@worker_deferred_unwatch).key?(first) + pf = plugin.instance_variable_get(:@pf) + pf.try_compact + release.close + waiting(5) do + sleep 0.01 until plugin.instance_variable_get(:@worker_deferred_unwatch).empty? + end + pf.try_compact + entries = File.binread("#{@tmp_dir}/tail.pos").lines.map { |line| line.split("\t") } + assert_equal [paths.last], entries.map(&:first) + assert_equal 130, entries.first[1].to_i(16) + ensure + release.close + end + end + + data(rotate_wait: false, deferred_unwatch: true) + def test_follow_inodes_refresh_preserves_worker_draining_position_entries(immediate_unwatch) + paths = %w[tail-a.txt tail-b.txt].map { |name| "#{@tmp_dir}/#{name}" } + File.binwrite(paths.first, "a1\n") + File.binwrite(paths.last, "#{'b' * 129}\n") + config = SINGLE_LINE_CONFIG + config_element('', '', { + 'path' => "#{@tmp_dir}/tail-*.txt", + 'pos_file' => "#{@tmp_dir}/tail.pos", + 'tag' => 't1', + 'num_threads' => 2, + 'read_from_head' => true, + 'follow_inodes' => true, + 'enable_stat_watcher' => false, + 'refresh_interval' => '1h', + 'rotate_wait' => '1h', + }) + driver = create_driver(config, false) + started = Queue.new + release = Queue.new + driver.run(expect_records: 3, timeout: 5) do + plugin = driver.instance + waiting(5) { sleep 0.01 until driver.events.size == 2 } + plugin.send(:process_worker_completions) + tails = plugin.instance_variable_get(:@tails) + tails.each_value do |watcher| + watcher.watchers.each { |trigger| plugin.event_loop_detach(trigger) } + end + first = tails[paths.first] + pool = plugin.instance_variable_get(:@worker_pool) + wrap_worker_pool_process(pool) { started << true; release.pop } + File.open(paths.first, 'ab') { |file| file.puts('late-record') } + File.rename(paths.first, "#{@tmp_dir}/rotated.log") + + if immediate_unwatch + tails.delete(first.path) + first.unwatched = true + plugin.detach_watcher(first, first.ino, false) + assert_true plugin.instance_variable_get(:@worker_deferred_unwatch).key?(first) + assert_not_include plugin.instance_variable_get(:@worker_detached_watchers), first + else + plugin.refresh_watchers + rotation = plugin.instance_variable_get(:@tails_rotate_wait).fetch(first) + plugin.event_loop_detach(rotation[:timer]) + rotation[:timer].on_timer + assert_include plugin.instance_variable_get(:@worker_detached_watchers), first + assert_empty plugin.instance_variable_get(:@worker_deferred_unwatch) + end + waiting(5) { sleep 0.01 until started.size == 1 } + assert_true first.pending? + assert_false plugin.instance_variable_get(:@tails_rotate_wait).key?(first) + + pf = plugin.instance_variable_get(:@pf) + plugin.refresh_watchers + pf.try_compact + entries = File.binread("#{@tmp_dir}/tail.pos").lines.map { |line| line.split("\t") } + assert_equal paths.sort, entries.map(&:first).sort + + release.close + waiting(5) { sleep 0.01 until pool.statistics[:tasks].zero? && !first.pending? } + plugin.send(:process_worker_completions) + entries = File.binread("#{@tmp_dir}/tail.pos").lines.map { |line| line.split("\t") } + assert_equal 130, entries.find { |entry| entry.first == paths.last }[1].to_i(16) + + plugin.refresh_watchers + pf.try_compact + entries = File.binread("#{@tmp_dir}/tail.pos").lines.map { |line| line.split("\t") } + assert_equal [paths.last], entries.map(&:first) + assert_equal 130, entries.first[1].to_i(16) + ensure + release.close + first.force_close if first && !first.pending? + end + assert_equal ['a1', 'b' * 129, 'late-record'].sort, driver.events.map { |event| event[2]['message'] }.sort + end + + def test_worker_position_write_error_does_not_stop_completion_processing + paths = %w[tail-a.txt tail-b.txt].map { |name| "#{@tmp_dir}/#{name}" } + paths.each { |path| File.binwrite(path, '') } + config = SINGLE_LINE_CONFIG + config_element('', '', { + 'path' => "#{@tmp_dir}/tail-*.txt", + 'pos_file' => "#{@tmp_dir}/tail.pos", + 'tag' => 't1', + 'num_threads' => 2, + 'enable_stat_watcher' => false, + }) + driver = create_driver(config, false) + driver.run(expect_records: 3, timeout: 5) do + plugin = driver.instance + watchers = paths.map { |path| plugin.instance_variable_get(:@tails)[path] } + watchers.each do |watcher| + watcher.watchers.each { |trigger| plugin.event_loop_detach(trigger) } + end + pe = watchers.first.pe + original_update = pe.method(:update_pos) + failed = false + pe.define_singleton_method(:update_pos) do |position| + unless failed + failed = true + raise Errno::ENOSPC, 'controlled position write failure' + end + original_update.call(position) + end + paths.each { |path| File.binwrite(path, "line\n") } + watchers.each(&:on_notify) + pool = plugin.instance_variable_get(:@worker_pool) + waiting(5) { sleep 0.01 until driver.events.size == 2 && pool.statistics[:tasks].zero? && watchers.none?(&:pending?) } + assert_equal 0, watchers.first.pe.read_pos + assert_equal 5, watchers.last.pe.read_pos + watchers.first.on_notify + waiting(5) { sleep 0.01 until watchers.first.pe.read_pos == 5 } + end + assert_equal ['line', 'line', 'line'], driver.events.map { |event| event[2]['message'] } + assert_true driver.logs.any? { |line| line.include?('controlled position write failure') } + end + + def test_worker_completion_handles_truncation_during_parse + path = "#{@tmp_dir}/tail.txt" + Fluent::FileWrapper.open(path, 'wb') { |file| file.puts('before-truncate') } + config = CONFIG_READ_FROM_HEAD + SINGLE_LINE_CONFIG + config_element('', '', { + 'num_threads' => 2, + 'read_lines_limit' => 1, + 'enable_watch_timer' => false, + 'enable_stat_watcher' => true, + }) + driver = create_driver(config) + parse_started = Queue.new + continue_parse = Queue.new + + driver.run(expect_records: 3, timeout: 10) do + pool = driver.instance.instance_variable_get(:@worker_pool) + wrap_worker_pool_process(pool) do |batch| + if batch.lines.any? { |line| line.include?('blocked-before-truncate') } + parse_started << true + continue_parse.pop + end + end + waiting(5) { sleep 0.01 until driver.events.size == 1 } + Fluent::FileWrapper.open(path, 'ab') { |file| file.puts('blocked-before-truncate') } + tail_watcher = driver.instance.instance_variable_get(:@tails)[path] + # Avoid depending on the platform's stat polling interval. No further + # timer notification should be needed to detect truncation on completion. + driver.instance.timer_execute(:test_worker_initial_read, 0.01, repeat: false) { tail_watcher.on_notify } + waiting(5) { sleep 0.01 until parse_started.size == 1 } + Fluent::FileWrapper.open(path, 'wb') { |file| file.puts('after-truncate') } + continue_parse << true + ensure + continue_parse.close + end + + assert_equal %w[before-truncate blocked-before-truncate after-truncate], driver.events.map { |event| event[2]['message'] } + assert_equal 0, driver.instance.instance_variable_get(:@worker_pool).statistics[:tasks] + end + + data(forward: :each, reverse: :reverse_each) + def test_worker_queue_saturation_reschedules_waiting_files(notification_order) + paths = %w[tail-a.txt tail-b.txt tail-c.txt tail-d.txt tail-e.txt tail-f.txt].map { |name| "#{@tmp_dir}/#{name}" } + notification_paths = paths.public_send(notification_order).to_a + parked_path = notification_paths.last + records = paths.each_index.map do |index| + Array.new(200) { |line| "file-#{index}-#{line} #{'x' * 1000}" } + end + paths.each { |path| Fluent::FileWrapper.open(path, 'wb') {} } + config = config_element('ROOT', '', { + 'path' => "#{@tmp_dir}/tail-*.txt", + 'pos_file' => "#{@tmp_dir}/tail.pos", + 'tag' => 't1', + 'read_from_head' => true, + 'format' => '/(?.*)/', + 'num_threads' => 2, + 'read_lines_limit' => 50, + 'enable_stat_watcher' => false, + }) + driver = create_driver(config, false) + parse_started = Queue.new + continue_parse = Queue.new + + driver.run(expect_records: records.flatten.size, timeout: 10) do + plugin = driver.instance + plugin.instance_variable_get(:@tails).each_value do |watcher| + watcher.watchers.each { |trigger| plugin.event_loop_detach(trigger) } + end + pool = driver.instance.instance_variable_get(:@worker_pool) + # Block whichever files arrive first so notification order cannot + # allow completed batches to free capacity before the queue fills. + wrap_worker_pool_process(pool) do |_batch| + parse_started << true + continue_parse.pop + end + paths.each_with_index do |path, index| + lines = path == parked_path ? records[index].first(26) : records[index] + Fluent::FileWrapper.open(path, 'ab') { |file| file.puts(lines) } + end + notification_paths.each do |path| + driver.instance.instance_variable_get(:@tails)[path].on_notify + end + waiting(5) { sleep 0.01 until parse_started.size == 2 } + waiting(5) do + until pool.statistics[:tasks] == 4 && driver.instance.instance_variable_get(:@worker_waiting_watchers).any? + sleep 0.01 + end + end + watcher = plugin.instance_variable_get(:@tails)[parked_path] + assert_include plugin.instance_variable_get(:@worker_waiting_watchers), watcher + assert_true watcher.eof? + Fluent::FileWrapper.open(parked_path, 'ab') { |file| file.puts(records[paths.index(parked_path)].drop(26)) } + watcher.on_notify + continue_parse.close + waiting(5) { sleep 0.01 until driver.events.size == records.flatten.size } + ensure + continue_parse.close + end + + assert_equal records.flatten.sort, driver.events.map { |event| event[2]['message'] }.sort + assert_equal 0, driver.instance.instance_variable_get(:@worker_pool).statistics[:tasks] + end + + data(forward: :each, reverse: :reverse_each) + def test_worker_queue_saturation_drains_detached_waiting_file(notification_order) + paths = %w[tail-a.txt tail-b.txt tail-c.txt tail-d.txt tail-e.txt tail-f.txt].map { |name| "#{@tmp_dir}/#{name}" } + paths.each { |path| Fluent::FileWrapper.open(path, 'wb') {} } + config = config_element('ROOT', '', { + 'path' => "#{@tmp_dir}/tail-*.txt", + 'pos_file' => "#{@tmp_dir}/tail.pos", + 'tag' => 't1', + 'read_from_head' => true, + 'format' => '/(?.*)/', + 'num_threads' => 2, + 'read_lines_limit' => 1, + 'enable_stat_watcher' => false, + }) + driver = create_driver(config, false) + parse_started = Queue.new + continue_parse = Queue.new + position_entry = nil + + driver.run(expect_records: paths.size, timeout: 10) do + plugin = driver.instance + pool = plugin.instance_variable_get(:@worker_pool) + wrap_worker_pool_process(pool) do |_batch| + parse_started << true + continue_parse.pop + end + paths.each_with_index do |path, index| + Fluent::FileWrapper.open(path, 'ab') { |file| file.puts("file-#{index}") } + end + paths.public_send(notification_order) do |path| + plugin.instance_variable_get(:@tails)[path].on_notify + end + waiting(5) { sleep 0.01 until parse_started.size == 2 } + waiting(5) do + until pool.statistics[:tasks] == 4 && plugin.instance_variable_get(:@worker_waiting_watchers).any? + sleep 0.01 + end + end + + watcher = plugin.instance_variable_get(:@worker_waiting_watchers).first + target = Fluent::Plugin::TailInput::TargetInfo.new(watcher.path, watcher.ino) + position_entry = plugin.instance_variable_get(:@pf)[target] + plugin.send(:stop_watchers, { watcher.path => target }, immediate: true, unwatched: true) + assert_true plugin.instance_variable_get(:@worker_deferred_unwatch).key?(watcher) + assert_false watcher.eof? + continue_parse.close + ensure + continue_parse.close + end + + assert_equal (0...paths.size).map { |index| "file-#{index}" }, driver.events.map { |event| event[2]['message'] }.sort + assert_equal 0, driver.instance.instance_variable_get(:@worker_pool).statistics[:tasks] + assert_equal Fluent::Plugin::TailInput::PositionFile::UNWATCHED_POSITION, position_entry.read_pos + assert_empty driver.instance.instance_variable_get(:@worker_deferred_unwatch) + end + + def test_oversized_worker_batch_does_not_block_other_files + tail_input = Fluent::Plugin::TailInput + original_byte_limit = tail_input.const_get(:WORKER_QUEUE_BYTES) + tail_input.send(:remove_const, :WORKER_QUEUE_BYTES) + tail_input.const_set(:WORKER_QUEUE_BYTES, 6) + large_path = "#{@tmp_dir}/tail-large.txt" + short_path = "#{@tmp_dir}/tail-short.txt" + [large_path, short_path].each { |path| Fluent::FileWrapper.open(path, 'wb') {} } + config = config_element('ROOT', '', { + 'path' => "#{@tmp_dir}/tail-*.txt", + 'pos_file' => "#{@tmp_dir}/tail.pos", + 'tag' => 't1', + 'read_from_head' => true, + 'format' => '/(?.*)/', + 'num_threads' => 2, + 'read_lines_limit' => 1, + 'enable_stat_watcher' => false, + }) + driver = create_driver(config, false) + large_started = Queue.new + continue_large = Queue.new + released = false + + driver.run(expect_records: 2, timeout: 10) do + pool = driver.instance.instance_variable_get(:@worker_pool) + wrap_worker_pool_process(pool) do |batch| + if batch.lines.any? { |line| line.include?('oversized-record') } + large_started << true + continue_large.pop + end + end + Fluent::FileWrapper.open(large_path, 'ab') { |file| file.puts('oversized-record') } + Fluent::FileWrapper.open(short_path, 'ab') { |file| file.puts('x') } + waiting(5) { sleep 0.01 until large_started.size == 1 } + waiting(5) { sleep 0.01 until driver.events.any? { |event| event[2]['message'] == 'x' } } + continue_large << true + released = true + end + + assert_equal %w[oversized-record x], driver.events.map { |event| event[2]['message'] }.sort + ensure + continue_large << true if continue_large && !released + tail_input.send(:remove_const, :WORKER_QUEUE_BYTES) if tail_input.const_defined?(:WORKER_QUEUE_BYTES, false) + tail_input.const_set(:WORKER_QUEUE_BYTES, original_byte_limit) if original_byte_limit + end + + data(forward: :each, reverse: :reverse_each) + def test_worker_queue_saturation_drains_rotated_file(notification_order) + paths = %w[tail-a.txt tail-b.txt tail-c.txt tail-d.txt tail-e.txt tail-f.txt].map { |name| "#{@tmp_dir}/#{name}" } + paths.each { |path| Fluent::FileWrapper.open(path, 'wb') {} } + config = config_element('ROOT', '', { + 'path' => "#{@tmp_dir}/tail-*.txt", + 'pos_file' => "#{@tmp_dir}/tail.pos", + 'tag' => 't1', + 'read_from_head' => true, + 'format' => '/(?.*)/', + 'num_threads' => 2, + 'read_lines_limit' => 1, + 'refresh_interval' => '0.05s', + 'rotate_wait' => '0.1s', + 'enable_stat_watcher' => false, + }) + driver = create_driver(config, false) + parse_started = Queue.new + continue_parse = Queue.new + + driver.run(expect_records: paths.size + 1, timeout: 10) do + pool = driver.instance.instance_variable_get(:@worker_pool) + wrap_worker_pool_process(pool) do |_batch| + parse_started << true + continue_parse.pop + end + paths.each_with_index do |path, index| + Fluent::FileWrapper.open(path, 'ab') { |file| file.puts("file-#{index}") } + end + paths.public_send(notification_order) do |path| + driver.instance.instance_variable_get(:@tails)[path].on_notify + end + waiting(5) { sleep 0.01 until parse_started.size == 2 } + waiting(5) do + until pool.statistics[:tasks] == 4 && driver.instance.instance_variable_get(:@worker_waiting_watchers).any? + sleep 0.01 + end + end + + FileUtils.move(paths.first, "#{paths.first}.1") + Fluent::FileWrapper.open(paths.first, 'wb') { |file| file.puts('replacement-file') } + continue_parse.close + ensure + continue_parse.close + end + + messages = driver.events.map { |event| event[2]['message'] }.sort + assert_equal (0...paths.size).map { |index| "file-#{index}" } + ['replacement-file'], messages + assert_equal 0, driver.instance.instance_variable_get(:@worker_pool).statistics[:tasks] + assert_empty driver.instance.instance_variable_get(:@worker_waiting_watchers) + end + + def test_shutdown_drains_a_worker_batch_before_stopping + path = "#{@tmp_dir}/tail.txt" + Fluent::FileWrapper.open(path, 'wb') {} + config = CONFIG_READ_FROM_HEAD + SINGLE_LINE_CONFIG + config_element('', '', { + 'num_threads' => 2, + 'read_lines_limit' => 1, + 'enable_stat_watcher' => false, + }) + driver = create_driver(config) + parse_started = Queue.new + continue_parse = Queue.new + pool = nil + + driver.run(shutdown: false, timeout: 5) do + pool = driver.instance.instance_variable_get(:@worker_pool) + wrap_worker_pool_process(pool) do |_batch| + parse_started << true + continue_parse.pop + end + Fluent::FileWrapper.open(path, 'ab') { |file| file.puts('shutdown-record') } + waiting(5) { sleep 0.01 until parse_started.size == 1 } + + shutdown_thread = Thread.new { driver.instance_shutdown } + sleep 0.1 + continue_parse << true + shutdown_thread.value + end + + assert_equal ['shutdown-record'], driver.events.map { |event| event[2]['message'] } + assert_equal 0, pool.statistics[:tasks] + assert_true pool.instance_variable_get(:@threads).none?(&:alive?) + end + + def test_worker_shutdown_does_not_wait_for_a_throttled_backlog_to_reach_eof + path = "#{@tmp_dir}/tail.txt" + File.binwrite(path, '') + config = CONFIG_READ_FROM_HEAD + SINGLE_LINE_CONFIG + config_element('', '', { + 'num_threads' => 2, + 'read_bytes_limit_per_second' => 65536, + 'enable_stat_watcher' => false, + }) + driver = create_driver(config) + driver.run(shutdown: false) do + plugin = driver.instance + watcher = plugin.instance_variable_get(:@tails)[path] + File.binwrite(path, "#{'x' * 1000}\n" * 1200) + watcher.on_notify + waiting(5) { sleep 0.01 until driver.events.any? } + assert_false watcher.eof? + handler = watcher.instance_variable_get(:@io_handler) + + started = Fluent::Clock.now + driver.instance_shutdown + assert_operator Fluent::Clock.now - started, :<, Fluent::Plugin::TailInput::TailWatcher::IOHandler::SHUTDOWN_TIMEOUT + assert_false handler.opened? + assert_true plugin.instance_variable_get(:@pf_file).closed? + assert_empty driver.logs.grep(/worker pool did not drain/) + end + ensure + driver&.instance_shutdown unless driver&.instance&.terminated? + end + + data(completes_during_join: true, blocked_until_killed: false) + def test_worker_shutdown_handles_results_published_during_join(finish_parse) + io_handler_class = Fluent::Plugin::TailInput::TailWatcher::IOHandler + original_timeout = io_handler_class::SHUTDOWN_TIMEOUT + io_handler_class.send(:remove_const, :SHUTDOWN_TIMEOUT) + io_handler_class.const_set(:SHUTDOWN_TIMEOUT, 0.05) + path = "#{@tmp_dir}/tail.txt" + File.binwrite(path, '') + config = CONFIG_READ_FROM_HEAD + SINGLE_LINE_CONFIG + config_element('', '', { + 'num_threads' => 2, + 'enable_stat_watcher' => false, + }) + driver = create_driver(config) + started = Queue.new + release = Queue.new + + driver.run(shutdown: false) do + plugin = driver.instance + pool = plugin.instance_variable_get(:@worker_pool) + wrap_worker_pool_process(pool) { started << true; release.pop } + watcher = plugin.instance_variable_get(:@tails)[path] + File.binwrite(path, "last-batch\n") + watcher.on_notify + waiting(5) { sleep 0.01 until started.size == 1 } + handler = watcher.instance_variable_get(:@io_handler) + rotation_timer = plugin.timer_execute(:test_rotation_during_shutdown, 60, repeat: false) {} + plugin.instance_variable_get(:@tails).delete(path) + plugin.instance_variable_get(:@tails_rotate_wait)[watcher] = { ino: watcher.ino, timer: rotation_timer } + original_stop = pool.method(:stop) + rotation_timer_attached_at_stop = nil + pool.define_singleton_method(:stop) do + rotation_timer_attached_at_stop = rotation_timer.attached? + original_stop.call + end + plugin.event_loop_detach(plugin.instance_variable_get(:@worker_completion_watcher)) + if finish_parse + original_join = pool.method(:join) + pool.define_singleton_method(:join) do |timeout: nil| + release << true + original_join.call(timeout: 5) + end + end + + plugin.stop + plugin.before_shutdown + plugin.shutdown + assert_equal(finish_parse ? ['last-batch'] : [], driver.events.map { |event| event[2]['message'] }) + assert_equal(finish_parse ? 11 : 0, watcher.pe.read_pos) + assert_equal 0, pool.statistics[:tasks] + assert_false rotation_timer_attached_at_stop + assert_false handler.pending? + assert_true pool.instance_variable_get(:@threads).none?(&:alive?) + + driver.instance_shutdown + assert_false handler.opened? + assert_true plugin.instance_variable_get(:@pf_file).closed? + end + ensure + release&.close + driver&.instance_shutdown unless driver&.instance&.terminated? + if original_timeout + io_handler_class.send(:remove_const, :SHUTDOWN_TIMEOUT) + io_handler_class.const_set(:SHUTDOWN_TIMEOUT, original_timeout) + end + end + + def test_worker_shutdown_waits_for_a_completion_after_pool_acknowledgment + io_handler_class = Fluent::Plugin::TailInput::TailWatcher::IOHandler + original_timeout = io_handler_class::SHUTDOWN_TIMEOUT + io_handler_class.send(:remove_const, :SHUTDOWN_TIMEOUT) + io_handler_class.const_set(:SHUTDOWN_TIMEOUT, 0.05) + path = "#{@tmp_dir}/tail.txt" + File.binwrite(path, '') + config = SINGLE_LINE_CONFIG + config_element('', '', { + 'num_threads' => 2, + 'enable_stat_watcher' => false, + }) + driver = create_driver(config) + acknowledged = Queue.new + release = Queue.new + shutdown_thread = nil + driver.run(shutdown: false) do + plugin = driver.instance + pool = plugin.instance_variable_get(:@worker_pool) + original_acknowledge = pool.method(:acknowledge) + pool.define_singleton_method(:acknowledge) do |result| + original_acknowledge.call(result) + acknowledged << true + release.pop + end + watcher = plugin.instance_variable_get(:@tails)[path] + File.binwrite(path, "line\n") + watcher.on_notify + waiting(5) { sleep 0.01 until acknowledged.size == 1 } + assert_equal 0, pool.statistics[:tasks] + + shutdown_thread = Thread.new { driver.instance_shutdown } + assert_nil shutdown_thread.join(0.2) + release.close + assert_not_nil shutdown_thread.join(5) + assert_equal 5, watcher.pe.read_pos + assert_false watcher.pending? + assert_true plugin.instance_variable_get(:@pf_file).closed? + end + ensure + release&.close + shutdown_thread&.join(5) + driver&.instance_shutdown unless driver&.instance&.terminated? + if original_timeout + io_handler_class.send(:remove_const, :SHUTDOWN_TIMEOUT) + io_handler_class.const_set(:SHUTDOWN_TIMEOUT, original_timeout) + end + end + + def test_worker_detached_emit_failure_closes_the_file_without_a_position_file + path = "#{@tmp_dir}/tail.txt" + File.binwrite(path, '') + config = SINGLE_LINE_CONFIG + config_element('', '', { + 'path' => path, + 'tag' => 't1', + 'num_threads' => 2, + 'enable_stat_watcher' => false, + }) + driver = create_driver(config, false) + started = Queue.new + release = Queue.new + driver.run(shutdown: false) do + plugin = driver.instance + pool = plugin.instance_variable_get(:@worker_pool) + wrap_worker_pool_process(pool) { started << true; release.pop } + plugin.instance_variable_get(:@line_feeder).define_singleton_method(:emit) { |_events, _watcher| false } + watcher = plugin.instance_variable_get(:@tails)[path] + File.binwrite(path, "blocked-emit\n") + watcher.on_notify + waiting(5) { sleep 0.01 until started.size == 1 } + handler = watcher.instance_variable_get(:@io_handler) + plugin.instance_variable_get(:@tails).delete(path) + plugin.detach_watcher(watcher, watcher.ino) + release.close + + waiting(5) { sleep 0.01 until pool.statistics[:tasks].zero? && !handler.opened? } + assert_false handler.pending? + assert_equal 0, watcher.pe.read_pos + assert_empty plugin.instance_variable_get(:@worker_detached_watchers) + assert_empty driver.events + end + ensure + release&.close + driver&.instance_shutdown unless driver&.instance&.terminated? + end + + def test_worker_threads_do_not_leak_across_plugin_restarts + path = "#{@tmp_dir}/tail.txt" + Fluent::FileWrapper.open(path, 'wb') {} + config = CONFIG_READ_FROM_HEAD + SINGLE_LINE_CONFIG + config_element('', '', { + 'num_threads' => 2, + 'read_lines_limit' => 1, + 'enable_stat_watcher' => false, + }) + + 3.times do |index| + Fluent::FileWrapper.open(path, 'ab') { |file| file.puts("restart-#{index}") } + driver = create_driver(config) + driver.run(expect_records: 1, timeout: 5) {} + + pool = driver.instance.instance_variable_get(:@worker_pool) + assert_equal ["restart-#{index}"], driver.events.map { |event| event[2]['message'] } + assert_true pool.instance_variable_get(:@threads).none?(&:alive?) + end + + assert_empty Thread.list.select { |thread| thread.name&.start_with?('in_tail_worker_') } + end + + def test_worker_parse_error_retries_synchronously + path = "#{@tmp_dir}/tail.txt" + Fluent::FileWrapper.open(path, 'wb') {} + config = CONFIG_READ_FROM_HEAD + SINGLE_LINE_CONFIG + config_element('', '', { + 'num_threads' => 2, + 'read_lines_limit' => 1, + }) + driver = create_driver(config) + + driver.run(expect_records: 1, timeout: 5) do + pool = driver.instance.instance_variable_get(:@worker_pool) + original_process = pool.instance_variable_get(:@process) + failed_once = false + pool.instance_variable_set(:@process, lambda do |index, payload| + unless failed_once + failed_once = true + raise 'controlled worker parse error' + end + original_process.call(index, payload) + end) + Fluent::FileWrapper.open(path, 'ab') { |file| file.puts('retry-record') } + driver.instance.instance_variable_get(:@tails)[path].on_notify + end + + assert_equal ['retry-record'], driver.events.map { |event| event[2]['message'] } + assert_true driver.logs.any? { |line| line.include?('worker parsing failed; retrying synchronously') } + assert_equal 0, driver.instance.instance_variable_get(:@worker_pool).statistics[:tasks] + assert_equal 1, driver.instance.statistics['input']['worker_parse_error_count'] + end + data(flat: CONFIG_DISABLE_WATCH_TIMER + SINGLE_LINE_CONFIG, parse: CONFIG_DISABLE_WATCH_TIMER + PARSE_SINGLE_LINE_CONFIG) def test_emit_without_watch_timer(data) @@ -2146,6 +2940,61 @@ def test_detach_watcher_unwatches_pos_entry_of_rotated_inode_with_follow_inodes d.instance_shutdown end + def test_detach_worker_waiting_watcher_defers_close_and_position_unwatch + config = config_element("ROOT", "", { + "path" => "#{@tmp_dir}/not-created*", + "pos_file" => "#{@tmp_dir}/tail.pos", + "tag" => "t1", + "format" => "/(?.*)/", + "num_threads" => 2, + "read_from_head" => "true", + "follow_inodes" => "true", + "enable_stat_watcher" => "false", + }) + d = create_driver(config, false) + d.run(shutdown: false) do + plugin = d.instance + pf = plugin.instance_variable_get(:@pf) + ino = 0xffffffffffffff02 + target_info = Fluent::Plugin::TailInput::TargetInfo.new("#{@tmp_dir}/rotated.txt", ino) + pe = pf[target_info] + pe.update(ino, 12) + metrics = plugin.instance_variable_get(:@metrics) + tw = Fluent::Plugin::TailInput::TailWatcher.new( + target_info, pe, $log, true, true, nil, nil, nil, metrics + ) + tw.unwatched = true + + closed = false + at_eof = false + io_handler = Object.new + io_handler.define_singleton_method(:pending?) { false } + io_handler.define_singleton_method(:eof?) { at_eof } + io_handler.define_singleton_method(:drained?) { at_eof } + io_handler.define_singleton_method(:ready_to_shutdown) { |_shutdown_start_time| } + io_handler.define_singleton_method(:on_notify) {} + io_handler.define_singleton_method(:close) { closed = true } + tw.instance_variable_set(:@io_handler, io_handler) + plugin.instance_variable_get(:@worker_waiting_watchers) << tw + + plugin.detach_watcher(tw, ino) + + assert_false closed + assert_equal 12, pe.read_pos + assert_false tw.eof? + + at_eof = true + plugin.send(:process_worker_completions) + + assert_true closed + assert_true tw.eof? + assert_equal Fluent::Plugin::TailInput::PositionFile::UNWATCHED_POSITION, pe.read_pos + assert_false pf.instance_variable_get(:@map).key?(ino) + end + + d.instance_shutdown + end + # With `follow_inodes false`, position entries are keyed by path, so the # entry must not be unwatched while the path is still followed by a watcher. # (See https://github.com/fluent/fluentd/pull/4327)