From 27bb8057f4bfbfa1b2f93ff812f40db3d75fddc4 Mon Sep 17 00:00:00 2001
From: Hiroshi Hatake <hatake@calyptia.com>
Date: Tue, 11 May 2021 15:22:29 +0900
Subject: [PATCH] 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 notified 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 | 18 +++++++++++++++---
 1 file changed, 15 insertions(+), 3 deletions(-)

diff --git a/lib/fluent/plugin/in_tail.rb b/lib/fluent/plugin/in_tail.rb
index a07902a715..8a9a9d8a51 100644
--- a/lib/fluent/plugin/in_tail.rb
+++ b/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,10 +950,17 @@ 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
 
@@ -959,7 +968,9 @@ def handle_notify
                 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