Secondary calculation should consider retry_max_times. fix #1449

lib/fluent/plugin_helper/retry_state.rb

@@ -64,7 +64,12 @@ def initialize(title, wait, timeout, forever, max_steps, randomize, randomize_wi
           @secondary_threshold = secondary_threshold
           if @secondary
             raise "BUG: secondary_transition_threshold MUST be between 0 and 1" if @secondary_threshold <= 0 || @secondary_threshold >= 1
-            @secondary_transition_at = @start + timeout * @secondary_threshold
+            max_retry_timeout = timeout
+            if max_steps
+              timeout_by_max_steps = calc_max_retry_timeout(max_steps)
+              max_retry_timeout = timeout_by_max_steps if timeout_by_max_steps < max_retry_timeout
+            end
+            @secondary_transition_at = @start + max_retry_timeout * @secondary_threshold
             @secondary_transition_steps = nil
           end
         end
@@ -137,40 +142,59 @@ def limit?
 
       class ExponentialBackOffRetry < RetryStateMachine
         def initialize(title, wait, timeout, forever, max_steps, randomize, randomize_width, backoff_base, max_interval, secondary, secondary_threathold)
-          super(title, wait, timeout, forever, max_steps, randomize, randomize_width, secondary, secondary_threathold)
           @constant_factor = wait
           @backoff_base = backoff_base
           @max_interval = max_interval
 
+          super(title, wait, timeout, forever, max_steps, randomize, randomize_width, secondary, secondary_threathold)
+
           @next_time = @start + @constant_factor
         end
 
         def naive_next_time(retry_next_times)
-          # make it infinite if calculated "interval" is too big
-          interval = @constant_factor.to_f * ( @backoff_base ** ( retry_next_times - 1 ) )
-          intr = if interval.finite?
-                   if @max_interval && interval > @max_interval
-                     @max_interval
-                   else
-                     interval
-                   end
-                 else
-                   interval
-                 end
+          intr = calc_interval(retry_next_times)
           current_time + randomize(intr)
         end
+
+        def calc_max_retry_timeout(max_steps)
+          result = 0
+          max_steps.times { |i|
+            result += calc_interval(i)
+          }
+          result
+        end
+
+        def calc_interval(num)
+          # make it infinite if calculated "interval" is too big
+          interval = @constant_factor.to_f * (@backoff_base ** (num - 1))
+          if interval.finite?
+            if @max_interval && interval > @max_interval
+              @max_interval
+            else
+              interval
+            end
+          else
+            interval
+          end
+        end
       end
 
       class PeriodicRetry < RetryStateMachine
         def initialize(title, wait, timeout, forever, max_steps, randomize, randomize_width, secondary, secondary_threathold)
-          super(title, wait, timeout, forever, max_steps, randomize, randomize_width, secondary, secondary_threathold)
           @retry_wait = wait
+
+          super(title, wait, timeout, forever, max_steps, randomize, randomize_width, secondary, secondary_threathold)
+
           @next_time = @start + @retry_wait
         end
 
         def naive_next_time(retry_next_times)
           current_time + randomize(@retry_wait)
         end
+
+        def calc_max_retry_timeout(max_steps)
+          @retry_wait * max_steps
+        end
       end
     end
   end

test/plugin_helper/test_retry_state.rb

@@ -178,6 +178,16 @@ class Dummy < Fluent::Plugin::TestBase
     assert_equal (dummy_current_time + 100 * 0.75), s.secondary_transition_at
   end
 
+  test 'periodic retries with secondary and max_steps' do
+    s = @d.retry_state_create(:t3, :periodic, 3, 100, max_steps: 5, randomize: false, secondary: true)
+    dummy_current_time = s.start
+    override_current_time(s, dummy_current_time)
+
+    assert_equal dummy_current_time, s.current_time
+    assert_equal (dummy_current_time + 100), s.timeout_at
+    assert_equal (dummy_current_time + 3 * 5 * 0.8), s.secondary_transition_at
+  end
+
   test 'exponential backoff forever without randomization' do
     s = @d.retry_state_create(:t11, :exponential_backoff, 0.1, 300, randomize: false, forever: true, backoff_base: 2)
     dummy_current_time = s.start
@@ -396,4 +406,19 @@ class Dummy < Fluent::Plugin::TestBase
     assert_equal (dummy_current_time + 100), s.timeout_at
     assert_equal (dummy_current_time + 100 * 0.75), s.secondary_transition_at
   end
+
+  test 'exponential backoff retries with secondary and max_steps' do
+    s = @d.retry_state_create(:t15, :exponential_backoff, 1, 100, randomize: false, max_steps: 5, backoff_base: 2, secondary: true) # threshold 0.8
+    dummy_current_time = s.start
+    override_current_time(s, dummy_current_time)
+
+    timeout = 0
+    s.instance_variable_get(:@max_steps).times { |i|
+      timeout += 1.0 * (2 ** (i - 1))
+    }
+
+    assert_equal dummy_current_time, s.current_time
+    assert_equal (dummy_current_time + 100), s.timeout_at
+    assert_equal (dummy_current_time + timeout * 0.8), s.secondary_transition_at
+  end
 end