Fix wrong calcuration of retry interval and detecting retry limit

lib/fluent/plugin/output.rb

@@ -1275,37 +1275,32 @@ def update_retry_state(chunk_id, using_secondary, error = nil)
 
           unless @retry
             @retry = retry_state(@buffer_config.retry_randomize)
+
             if @retry.limit?
-              # @retry_max_times == 0, fail imediately by the following block
+              handle_limit_reached(error)
             else
               if error
                 log.warn "failed to flush the buffer.", retry_times: @retry.steps, next_retry_time: @retry.next_time.round, chunk: chunk_id_hex, error: error
                 log.warn_backtrace error.backtrace
               end
-              return
             end
+
+            return
           end
 
           # @retry exists
 
+          # Ensure that the current time is greater than or equal to @retry.next_time to avoid the situation when
+          # @retry.step is called almost as many times as the number of flush threads in a short time.
+          if Time.now >= @retry.next_time
+            @retry.step
+          else
+            @retry.recalc_next_time # to prevent all flush threads from retrying at the same time
+          end
+
           if @retry.limit?
-            if error
-              records = @buffer.queued_records
-              msg = "failed to flush the buffer, and hit limit for retries. dropping all chunks in the buffer queue."
-              log.error msg, retry_times: @retry.steps, records: records, error: error
-              log.error_backtrace error.backtrace
-            end
-            @buffer.clear_queue!
-            log.debug "buffer queue cleared"
-            @retry = nil
+            handle_limit_reached(error)
           else
-            # Ensure that the current time is greater than or equal to @retry.next_time to avoid the situation when
-            # @retry.step is called almost as many times as the number of flush threads in a short time.
-            if Time.now >= @retry.next_time
-              @retry.step
-            else
-              @retry.recalc_next_time # to prevent all flush threads from retrying at the same time
-            end
             if error
               if using_secondary
                 msg = "failed to flush the buffer with secondary output."
@@ -1321,6 +1316,18 @@ def update_retry_state(chunk_id, using_secondary, error = nil)
         end
       end
 
+      def handle_limit_reached(error)
+        if error
+          records = @buffer.queued_records
+          msg = "Hit limit for retries. dropping all chunks in the buffer queue."
+          log.error msg, retry_times: @retry.steps, records: records, error: error
+          log.error_backtrace error.backtrace
+        end
+        @buffer.clear_queue!
+        log.debug "buffer queue cleared"
+        @retry = nil
+      end
+
       def retry_state(randomize)
         if @secondary
           retry_state_create(

lib/fluent/plugin_helper/retry_state.rb

@@ -44,6 +44,8 @@ def initialize(title, wait, timeout, forever, max_steps, randomize, randomize_wi
 
           @timeout = timeout
           @timeout_at = @start + timeout
+          @has_reached_timeout = false
+          @has_timed_out = false
           @current = :primary
 
           if randomize_width < 0 || randomize_width > 0.5
@@ -98,7 +100,7 @@ def calc_next_time
               naive
             end
           elsif @current == :secondary
-            naive = naive_next_time(@steps - @secondary_transition_steps + 1)
+            naive = naive_next_time(@steps - @secondary_transition_steps)
             if naive >= @timeout_at
               @timeout_at
             else
@@ -123,7 +125,15 @@ def step
             @current = :secondary
             @secondary_transition_steps = @steps
           end
+
           @next_time = calc_next_time
+
+          if @has_reached_timeout
+            @has_timed_out = @next_time >= @timeout_at
+          else
+            @has_reached_timeout = @next_time >= @timeout_at
+          end
+
           nil
         end
 
@@ -135,7 +145,7 @@ def limit?
           if @forever
             false
           else
-            @next_time >= @timeout_at || !!(@max_steps && @steps >= @max_steps)
+            @has_timed_out || !!(@max_steps && @steps >= @max_steps)
           end
         end
       end
@@ -159,13 +169,13 @@ def naive_next_time(retry_next_times)
         def calc_max_retry_timeout(max_steps)
           result = 0
           max_steps.times { |i|
-            result += calc_interval(i + 1)
+            result += calc_interval(i)
           }
           result
         end
 
         def calc_interval(num)
-          interval = raw_interval(num - 1)
+          interval = raw_interval(num)
           if @max_interval && interval > @max_interval
             @max_interval
           else
@@ -175,7 +185,7 @@ def calc_interval(num)
               # Calculate previous finite value to avoid inf related errors. If this re-computing is heavy, use cache.
               until interval.finite?
                 num -= 1
-                interval = raw_interval(num - 1)
+                interval = raw_interval(num)
               end
               interval
             end

test/plugin/test_output_as_buffered_retries.rb

@@ -140,13 +140,13 @@ def get_log_time(msg, logs)
 
       retry_state = @i.retry_state( @i.buffer_config.retry_randomize )
       retry_state.step
-      assert_equal 1, (retry_state.next_time - now)
-      retry_state.step
       assert_equal (1 * (2 ** 1)), (retry_state.next_time - now)
       retry_state.step
       assert_equal (1 * (2 ** 2)), (retry_state.next_time - now)
       retry_state.step
       assert_equal (1 * (2 ** 3)), (retry_state.next_time - now)
+      retry_state.step
+      assert_equal (1 * (2 ** 4)), (retry_state.next_time - now)
     end
 
     test 'does retries correctly when #write fails' do
@@ -332,7 +332,7 @@ def get_log_time(msg, logs)
       @i.emit_events("test.tag.3", dummy_event_stream())
 
       logs = @i.log.out.logs
-      assert{ logs.any?{|l| l.include?("[error]: failed to flush the buffer, and hit limit for retries. dropping all chunks in the buffer queue.") } }
+      assert{ logs.any?{|l| l.include?("[error]: Hit limit for retries. dropping all chunks in the buffer queue.") } }
     end
 
     test 'output plugin give retries up by retry_max_times, and clear queue in buffer' do
@@ -409,7 +409,7 @@ def get_log_time(msg, logs)
       @i.emit_events("test.tag.3", dummy_event_stream())
 
       logs = @i.log.out.logs
-      assert{ logs.any?{|l| l.include?("[error]: failed to flush the buffer, and hit limit for retries. dropping all chunks in the buffer queue.") && l.include?("retry_times=10") } }
+      assert{ logs.any?{|l| l.include?("[error]: Hit limit for retries. dropping all chunks in the buffer queue.") && l.include?("retry_times=10") } }
 
       assert{ @i.buffer.queue.size == 0 }
       assert{ @i.buffer.stage.size == 1 }
@@ -607,7 +607,7 @@ def get_log_time(msg, logs)
       logs = @i.log.out.logs
 
       target_time = Time.parse("2016-04-13 18:35:31 -0700")
-      target_msg = "[error]: failed to flush the buffer, and hit limit for retries. dropping all chunks in the buffer queue."
+      target_msg = "[error]: Hit limit for retries. dropping all chunks in the buffer queue."
       assert{ logs.any?{|l| l.include?(target_msg) } }
 
       log_time = get_log_time(target_msg, logs)
@@ -695,7 +695,7 @@ def get_log_time(msg, logs)
       @i.emit_events("test.tag.3", dummy_event_stream())
 
       logs = @i.log.out.logs
-      assert{ logs.any?{|l| l.include?("[error]: failed to flush the buffer, and hit limit for retries. dropping all chunks in the buffer queue.") && l.include?("retry_times=10") } }
+      assert{ logs.any?{|l| l.include?("[error]: Hit limit for retries. dropping all chunks in the buffer queue.") && l.include?("retry_times=10") } }
 
       assert{ @i.buffer.queue.size == 0 }
       assert{ @i.buffer.stage.size == 1 }
@@ -743,7 +743,7 @@ def get_log_time(msg, logs)
 
       assert(@i.write_count == 1)
       assert(@i.num_errors == 1)
-      assert(@i.log.out.logs.any?{|l| l.include?("[error]: failed to flush the buffer, and hit limit for retries. dropping all chunks in the buffer queue.") && l.include?("retry_times=0") })
+      assert(@i.log.out.logs.any?{|l| l.include?("[error]: Hit limit for retries. dropping all chunks in the buffer queue.") && l.include?("retry_times=0") })
       assert(@i.buffer.queue.size == 0)
       assert(@i.buffer.stage.size == 1)
       assert(@i.buffer.queue.all?{|c| c.empty? })

test/plugin/test_output_as_buffered_secondary.rb

@@ -874,7 +874,7 @@ def dummy_event_stream
       end
 
       logs = @i.log.out.logs
-      assert{ logs.any?{|l| l.include?("[error]: failed to flush the buffer, and hit limit for retries. dropping all chunks in the buffer queue.") } }
+      assert{ logs.any?{|l| l.include?("[error]: Hit limit for retries. dropping all chunks in the buffer queue.") } }
 
       assert{ now >= first_failure + 60 }
     end