Handle both of stat and timer watchers are enabled and steady growing

files case

In this case, timer watcher tells file contents every 1 seconds and file
contents changes is also notified from stats watcher.

In such circumstances, if some tailing files are steadily growing,
previous implementation does not prevent log ingestion for written
contents which is nodified by stat watcher.

This commit also prevents log ingestion in such cases.

Signed-off-by: Hiroshi Hatake <hatake@calyptia.com>

lib/fluent/plugin/in_tail.rb

@@ -908,6 +908,8 @@ def initialize(watcher, path:, read_lines_limit:, read_bytes_limit_per_second:,
           @io = nil
           @notify_mutex = Mutex.new
           @log = log
+          @last_read_time = 0
+          @number_bytes_read = 0
 
           @log.info "following tail of #{@path}"
         end
@@ -948,18 +950,27 @@ def limit_bytes_per_second_reached?(start_reading, number_bytes_read)
           false
         end
 
+        def refresh_bytes_read_counter
+          if Fluent::Clock.now - @last_read_time > 1
+            @number_bytes_read = 0
+          end
+
+          yield
+        end
+
         def handle_notify
           with_io do |io|
             begin
-              number_bytes_read = 0
               start_reading = Fluent::Clock.now
               read_more = false
 
               if !io.nil? && @lines.empty?
                 begin
                   while true
                     data = io.readpartial(BYTES_TO_READ, @iobuf)
-                    number_bytes_read += data.bytesize
+                    refresh_bytes_read_counter do
+                      @number_bytes_read += data.bytesize
+                    end
                     @fifo << data
                     @fifo.read_lines(@lines)
 
@@ -969,11 +980,12 @@ def handle_notify
                       read_more = true
                       break
                     end
-                    if limit_bytes_per_second_reached?(start_reading, number_bytes_read)
+                    if limit_bytes_per_second_reached?(start_reading, @number_bytes_read)
                       # Just get out from tailing loop.
                       read_more = false
                       break
                     end
+                    @last_read_time = Fluent::Clock.now
                   end
                 rescue EOFError
                 end