diff --git a/README.md b/README.md index c61b91f..73176eb 100644 --- a/README.md +++ b/README.md @@ -202,6 +202,7 @@ module Freno DEFAULT_WAIT_SECONDS = 0.5 DEFAULT_MAX_WAIT_SECONDS = 10 + DEFAULT_REQUIRED_CONSECUTIVE_SUCCESSES = 3 def initialize(client: nil, app: nil, @@ -209,7 +210,8 @@ module Freno instrumenter: Instrumenter::Noop, circuit_breaker: CircuitBreaker::Noop, wait_seconds: DEFAULT_WAIT_SECONDS, - max_wait_seconds: DEFAULT_MAX_WAIT_SECONDS) + max_wait_seconds: DEFAULT_MAX_WAIT_SECONDS, + required_consecutive_successes: DEFAULT_REQUIRED_CONSECUTIVE_SUCCESSES) @client = client @@ -238,6 +240,16 @@ You optionally provide the time you want the throttler to sleep in case the chec If replication lags badly, you can control until when you want to keep sleeping and retrying the check by setting `max_wait_seconds`. When that times out, the throttle will raise a `Freno::Throttler::WaitedTooLong` error. +An initially healthy check proceeds immediately. After any failed check, +`required_consecutive_successes` passing checks are required before the block +runs. Passing samples must be consecutive; another rejection resets the count. +This prevents an oscillating metric from releasing work on a single healthy +trough. + +`WaitedTooLong` represents normal sustained throttling and does not count as a +circuit-breaker failure. Freno transport and decision errors still fail the +circuit breaker. + #### Instrumenting the throttler You can also configure the throttler with an `instrumenter` collaborator to subscribe to events happening during the `throttle` call. diff --git a/lib/freno/client.rb b/lib/freno/client.rb index 36505e1..4674bb5 100644 --- a/lib/freno/client.rb +++ b/lib/freno/client.rb @@ -233,7 +233,7 @@ def decorated(request) outermost = to_decorate[0] current = outermost - (to_decorate[1..]).each do |decorator| + to_decorate[1..].each do |decorator| current.request = decorator current = current.request end diff --git a/lib/freno/client/errors.rb b/lib/freno/client/errors.rb index afbb591..fbbb7a1 100644 --- a/lib/freno/client/errors.rb +++ b/lib/freno/client/errors.rb @@ -1,5 +1,6 @@ # frozen_string_literal: true module Freno - Error = Class.new(StandardError) + class Error < StandardError + end end diff --git a/lib/freno/client/preconditions.rb b/lib/freno/client/preconditions.rb index a49174b..2c5ed55 100644 --- a/lib/freno/client/preconditions.rb +++ b/lib/freno/client/preconditions.rb @@ -5,7 +5,8 @@ class Client module Preconditions module_function - PreconditionNotMet = Class.new(ArgumentError) + class PreconditionNotMet < ArgumentError + end class Checker attr_reader :errors diff --git a/lib/freno/throttler.rb b/lib/freno/throttler.rb index aec22bb..78d0fda 100644 --- a/lib/freno/throttler.rb +++ b/lib/freno/throttler.rb @@ -35,6 +35,7 @@ module Freno class Throttler DEFAULT_WAIT_SECONDS = 0.5 DEFAULT_MAX_WAIT_SECONDS = 10 + DEFAULT_REQUIRED_CONSECUTIVE_SUCCESSES = 3 REQUIRED_ARGS = %i[ client app @@ -43,6 +44,7 @@ class Throttler circuit_breaker wait_seconds max_wait_seconds + required_consecutive_successes ].freeze attr_accessor :client, @@ -51,7 +53,8 @@ class Throttler :instrumenter, :circuit_breaker, :wait_seconds, - :max_wait_seconds + :max_wait_seconds, + :required_consecutive_successes # Initializes a new instance of the throttler # @@ -99,6 +102,10 @@ class Throttler # seconds the throttler will wait in total for replicas to catch-up # before raising a `WaitedTooLong` error. # + # - `:required_consecutive_successes`: The number of consecutive passing + # checks required after any failed check. An initially passing check + # still proceeds immediately. + # def initialize( client: nil, app: nil, @@ -106,7 +113,8 @@ def initialize( instrumenter: Instrumenter::Noop, circuit_breaker: CircuitBreaker::Noop, wait_seconds: DEFAULT_WAIT_SECONDS, - max_wait_seconds: DEFAULT_MAX_WAIT_SECONDS + max_wait_seconds: DEFAULT_MAX_WAIT_SECONDS, + required_consecutive_successes: DEFAULT_REQUIRED_CONSECUTIVE_SUCCESSES ) @client = client @app = app @@ -115,6 +123,7 @@ def initialize( @circuit_breaker = circuit_breaker @wait_seconds = wait_seconds @max_wait_seconds = max_wait_seconds + @required_consecutive_successes = required_consecutive_successes yield self if block_given? @@ -165,6 +174,8 @@ def throttle(context = nil, **options) store_names = mapper.call(context) instrument(:called, store_names: store_names) waited = 0 + throttled = false + consecutive_successes = 0 while true unless circuit_breaker.allow_request? @@ -173,19 +184,43 @@ def throttle(context = nil, **options) end if all_stores_ok?(store_names, **options) - instrument(:succeeded, store_names: store_names, waited: waited) - circuit_breaker.success - break + consecutive_successes += 1 + if !throttled || consecutive_successes >= required_consecutive_successes + instrument( + :succeeded, + store_names: store_names, + waited: waited, + consecutive_successes: consecutive_successes + ) + circuit_breaker.success + break + end + else + throttled = true + consecutive_successes = 0 end if waited + wait_seconds > max_wait_seconds - instrument(:waited_too_long, store_names: store_names, waited: waited, max: max_wait_seconds) - circuit_breaker.failure + instrument( + :waited_too_long, + store_names: store_names, + waited: waited, + max: max_wait_seconds, + consecutive_successes: consecutive_successes, + required_consecutive_successes: required_consecutive_successes + ) raise WaitedTooLong.new(waited_seconds: waited, max_wait_seconds: max_wait_seconds) else wait waited += wait_seconds - instrument(:waited, store_names: store_names, waited: waited, max: max_wait_seconds) + instrument( + :waited, + store_names: store_names, + waited: waited, + max: max_wait_seconds, + consecutive_successes: consecutive_successes, + required_consecutive_successes: required_consecutive_successes + ) end end @@ -204,6 +239,9 @@ def validate_args unless max_wait_seconds > wait_seconds errors << "max_wait_seconds (#{max_wait_seconds}) has to be greather than wait_seconds (#{wait_seconds})" end + unless required_consecutive_successes.is_a?(Integer) && required_consecutive_successes.positive? + errors << "required_consecutive_successes (#{required_consecutive_successes}) must be a positive integer" + end raise ArgumentError, errors.join("\n") if errors.any? end diff --git a/test/freno/throttler_test.rb b/test/freno/throttler_test.rb index 7053983..01e0e08 100644 --- a/test/freno/throttler_test.rb +++ b/test/freno/throttler_test.rb @@ -10,6 +10,15 @@ def test_validations assert_includes ex.message, "app must be provided" assert_includes ex.message, "client must be provided" assert_includes ex.message, "max_wait_seconds (0.5) has to be greather than wait_seconds (1)" + + ex = assert_raises(ArgumentError) do + Freno::Throttler.new( + client: sample_client, + app: :github, + required_consecutive_successes: 0 + ) + end + assert_includes ex.message, "required_consecutive_successes (0) must be a positive integer" end def test_using_the_default_identity_mapper @@ -80,9 +89,9 @@ def test_sleeps_when_a_check_fails_and_then_calls_the_block block_called = false stub = sample_client - stub.expects(:check?).times(2) + stub.expects(:check?).times(4) .with(app: :github, store_name: :mysqla, options: {}) - .returns(false).then.returns(true) + .returns(false).then.returns(true).then.returns(true).then.returns(true) throttler = Freno::Throttler.new do |t| t.client = stub @@ -90,7 +99,7 @@ def test_sleeps_when_a_check_fails_and_then_calls_the_block t.mapper = ->(_context) { [:mysqla] } t.instrumenter = MemoryInstrumenter.new end - throttler.expects(:wait).once + throttler.expects(:wait).times(3) throttler.throttle do block_called = true @@ -105,10 +114,12 @@ def test_sleeps_when_a_check_fails_and_then_calls_the_block waited_events = throttler.instrumenter.events_for("throttler.waited") - assert_equal 1, waited_events.count + assert_equal 3, waited_events.count assert_equal [:mysqla], waited_events.first[:store_names] assert_in_delta 0.5, waited_events.first[:waited], 0.01 assert_equal 10, waited_events.first[:max] + assert_equal 3, waited_events.last[:required_consecutive_successes] + assert_equal 2, waited_events.last[:consecutive_successes] assert_equal 0, throttler.instrumenter.count("throttler.waited_too_long") assert_equal 0, throttler.instrumenter.count("throttler.freno_errored") @@ -158,6 +169,65 @@ def test_raises_waited_too_long_if_freno_checks_failed_consistenly assert_equal 0, throttler.instrumenter.count("throttler.circuit_open") end + def test_recovery_requires_consecutive_passing_checks + block_called = false + + stub = sample_client + stub.expects(:check?).times(7) + .with(app: :github, store_name: :mysqla, options: {}) + .returns(false) + .then.returns(true) + .then.returns(false) + .then.returns(true) + .then.returns(true) + .then.returns(false) + .then.returns(true) + + throttler = Freno::Throttler.new do |t| + t.client = stub + t.app = :github + t.mapper = ->(_context) { [:mysqla] } + t.instrumenter = MemoryInstrumenter.new + t.wait_seconds = 1 + t.max_wait_seconds = 6 + end + throttler.expects(:wait).times(6) + + assert_raises(Freno::Throttler::WaitedTooLong) do + throttler.throttle do + block_called = true + end + end + + refute block_called + + event = throttler.instrumenter.events_for("throttler.waited_too_long").first + + assert_equal 1, event[:consecutive_successes] + assert_equal 3, event[:required_consecutive_successes] + end + + def test_waited_too_long_does_not_fail_the_circuit_breaker + client = sample_client + client.stubs(:check?).returns(false) + circuit_breaker = mock + circuit_breaker.stubs(:allow_request?).returns(true) + circuit_breaker.expects(:failure).never + + throttler = Freno::Throttler.new( + client: client, + app: :github, + circuit_breaker: circuit_breaker, + wait_seconds: 1, + max_wait_seconds: 2 + ) + throttler.expects(:wait).times(2) + + assert_raises(Freno::Throttler::WaitedTooLong) do + throttler.throttle(:mysqla) { flunk "throttled block should not run" } + end + end + def test_raises_a_specific_error_in_case_freno_itself_errored block_called = false