Merge pull request #188 from Shopify/half-open-resource-timeout

circuit: introduce half_open_resource_timeout

README.md

@@ -431,6 +431,7 @@ There are three configuration parameters for circuit breakers in Semian:
   again.
 * **success_threshold**. The amount of successes on the circuit until closing it
   again, that is to start accepting all requests to the circuit.
+* **half_open_resource_timeout**. Timeout for the resource when the circuit is half-open (only supported for MySQL).
 
 ### Bulkheading
 

lib/semian.rb

@@ -244,6 +244,7 @@ def create_circuit_breaker(name, **options)
       error_threshold: options[:error_threshold],
       error_timeout: options[:error_timeout],
       exceptions: Array(exceptions) + [::Semian::BaseError],
+      half_open_resource_timeout: options[:half_open_resource_timeout],
       implementation: implementation,
     )
   end

lib/semian/adapter.rb

@@ -31,7 +31,7 @@ def clear_semian_resource
 
     def acquire_semian_resource(scope:, adapter:, &block)
       return yield if resource_already_acquired?
-      semian_resource.acquire(scope: scope, adapter: adapter) do
+      semian_resource.acquire(scope: scope, adapter: adapter, resource: self) do
         mark_resource_as_acquired(&block)
       end
     rescue ::Semian::OpenCircuitError => error

lib/semian/circuit_breaker.rb

@@ -4,21 +4,22 @@ class CircuitBreaker #:nodoc:
 
     def_delegators :@state, :closed?, :open?, :half_open?
 
-    attr_reader :name
+    attr_reader :name, :half_open_resource_timeout
 
-    def initialize(name, exceptions:, success_threshold:, error_threshold:, error_timeout:, implementation:)
+    def initialize(name, exceptions:, success_threshold:, error_threshold:, error_timeout:, implementation:, half_open_resource_timeout: nil)
       @name = name.to_sym
       @success_count_threshold = success_threshold
       @error_count_threshold = error_threshold
       @error_timeout = error_timeout
       @exceptions = exceptions
+      @half_open_resource_timeout = half_open_resource_timeout
 
       @errors = implementation::SlidingWindow.new(max_size: @error_count_threshold)
       @successes = implementation::Integer.new
       @state = implementation::State.new
     end
 
-    def acquire
+    def acquire(resource = nil, &block)
       return yield if disabled?
 
       half_open if open? && error_timeout_expired?
@@ -27,7 +28,7 @@ def acquire
 
       result = nil
       begin
-        result = yield
+        result = maybe_with_half_open_resource_timeout(resource, &block)
       rescue *@exceptions => error
         mark_failed(error)
         raise error
@@ -122,5 +123,19 @@ def log_state_transition(new_state)
     def disabled?
       ENV['SEMIAN_CIRCUIT_BREAKER_DISABLED'] || ENV['SEMIAN_DISABLED']
     end
+
+    def maybe_with_half_open_resource_timeout(resource, &block)
+      result = nil
+
+      if half_open? && @half_open_resource_timeout && resource.respond_to?(:with_resource_timeout)
+        resource.with_resource_timeout(@half_open_resource_timeout) do
+          result = block.call
+        end
+      else
+        result = block.call
+      end
+
+      result
+    end
   end
 end

lib/semian/mysql2.rb

@@ -83,6 +83,22 @@ def query(*args)
       end
     end
 
+    # TODO: write_timeout and connect_timeout can't be configured currently
+    # dynamically, await https://github.com/brianmario/mysql2/pull/955
+    def with_resource_timeout(temp_timeout)
+      prev_read_timeout = @read_timeout
+
+      begin
+        # C-ext reads this directly, writer method will configure
+        # properly on the client but based on my read--this is good enough
+        # until we get https://github.com/brianmario/mysql2/pull/955 in
+        @read_timeout = temp_timeout
+        yield
+      ensure
+        @read_timeout = prev_read_timeout
+      end
+    end
+
     private
 
     def query_whitelisted?(sql, *)

lib/semian/protected_resource.rb

@@ -19,8 +19,8 @@ def destroy
       @circuit_breaker.destroy unless @circuit_breaker.nil?
     end
 
-    def acquire(timeout: nil, scope: nil, adapter: nil)
-      acquire_circuit_breaker(scope, adapter) do
+    def acquire(timeout: nil, scope: nil, adapter: nil, resource: nil)
+      acquire_circuit_breaker(scope, adapter, resource) do
         acquire_bulkhead(timeout, scope, adapter) do
           yield self
         end
@@ -29,11 +29,11 @@ def acquire(timeout: nil, scope: nil, adapter: nil)
 
     private
 
-    def acquire_circuit_breaker(scope, adapter)
+    def acquire_circuit_breaker(scope, adapter, resource)
       if @circuit_breaker.nil?
         yield self
       else
-        @circuit_breaker.acquire do
+        @circuit_breaker.acquire(resource) do
           yield self
         end
       end

semian.gemspec

@@ -22,6 +22,7 @@ Gem::Specification.new do |s|
   s.add_development_dependency 'rake', '< 11.0'
   s.add_development_dependency 'timecop'
   s.add_development_dependency 'minitest'
+  s.add_development_dependency 'byebug'
   s.add_development_dependency 'mysql2'
   s.add_development_dependency 'redis'
   s.add_development_dependency 'thin', '~> 1.6.4'

test/circuit_breaker_test.rb

@@ -113,4 +113,72 @@ def test_semian_wide_env_var_disables_circuit_breaker
   ensure
     ENV.delete('SEMIAN_DISABLED')
   end
+
+  class RawResource
+    def timeout
+      @timeout || 2
+    end
+
+    def with_resource_timeout(timeout)
+      prev_timeout = @timeout
+      @timeout = timeout
+      yield
+    ensure
+      @timeout = prev_timeout
+    end
+  end
+
+  def test_changes_resource_timeout_when_configured
+    Semian.register(:resource_timeout, tickets: 1, exceptions: [SomeError],
+                                       error_threshold: 2, error_timeout: 5, success_threshold: 1,
+                                       half_open_resource_timeout: 0.123)
+    resource = Semian[:resource_timeout]
+
+    half_open_cicuit!(resource)
+
+    raw_resource = RawResource.new
+
+    triggered = false
+    resource.acquire(resource: raw_resource) do
+      triggered = true
+      assert_equal 0.123, raw_resource.timeout
+    end
+
+    assert triggered
+    assert_equal 2, raw_resource.timeout
+  end
+
+  def test_doesnt_change_resource_timeout_when_closed
+    Semian.register(:resource_timeout, tickets: 1, exceptions: [SomeError],
+                                       error_threshold: 2, error_timeout: 5, success_threshold: 1,
+                                       half_open_resource_timeout: 0.123)
+    resource = Semian[:resource_timeout]
+
+    raw_resource = RawResource.new
+
+    triggered = false
+    resource.acquire(resource: raw_resource) do
+      triggered = true
+      assert_equal 2, raw_resource.timeout
+    end
+
+    assert triggered
+    assert_equal 2, raw_resource.timeout
+  end
+
+  def test_doesnt_blow_up_when_configured_half_open_timeout_but_adapter_doesnt_support
+    Semian.register(:resource_timeout, tickets: 1, exceptions: [SomeError],
+                                       error_threshold: 2, error_timeout: 5, success_threshold: 1,
+                                       half_open_resource_timeout: 0.123)
+    resource = Semian[:resource_timeout]
+
+    raw_resource = Object.new
+
+    triggered = false
+    resource.acquire(resource: raw_resource) do
+      triggered = true
+    end
+
+    assert triggered
+  end
 end

test/mysql2_test.rb

@@ -34,7 +34,7 @@ def test_semian_can_be_disabled
   end
 
   def test_connection_errors_open_the_circuit
-    @proxy.downstream(:latency, latency: 1200).apply do
+    @proxy.downstream(:latency, latency: 2200).apply do
       ERROR_THRESHOLD.times do
         assert_raises ::Mysql2::Error do
           connect_to_mysql!
@@ -233,7 +233,7 @@ def test_circuit_breaker_on_query
     client = connect_to_mysql!
     client2 = connect_to_mysql!
 
-    @proxy.downstream(:latency, latency: 1000).apply do
+    @proxy.downstream(:latency, latency: 2200).apply do
       background { client2.query('SELECT 1 + 1;') }
 
       ERROR_THRESHOLD.times do
@@ -272,7 +272,7 @@ def client.raw_ping
       super
     end
 
-    @proxy.downstream(:latency, latency: 1200).apply do
+    @proxy.downstream(:latency, latency: 2200).apply do
       ERROR_THRESHOLD.times do
         client.ping
       end
@@ -283,12 +283,52 @@ def client.raw_ping
     assert_equal ERROR_THRESHOLD, client.instance_variable_get(:@real_pings)
   end
 
+  def test_changes_timeout_when_half_open_and_configured
+    client = connect_to_mysql!(half_open_resource_timeout: 1)
+
+    @proxy.downstream(:latency, latency: 3000).apply do
+      (ERROR_THRESHOLD * 2).times do
+        assert_raises Mysql2::Error do
+          client.query('SELECT 1 + 1;')
+        end
+      end
+    end
+
+    assert_raises Mysql2::CircuitOpenError do
+      client.query('SELECT 1 + 1;')
+    end
+
+    Timecop.travel(ERROR_TIMEOUT + 1) do
+      @proxy.downstream(:latency, latency: 1500).apply do
+        assert_raises Mysql2::Error do
+          client.query('SELECT 1 + 1;')
+        end
+      end
+    end
+
+    Timecop.travel(ERROR_TIMEOUT * 2 + 1) do
+      client.query('SELECT 1 + 1;')
+      client.query('SELECT 1 + 1;')
+
+      # Timeout has reset to the normal 2 seconds now that Circuit is closed
+      @proxy.downstream(:latency, latency: 1500).apply do
+        client.query('SELECT 1 + 1;')
+      end
+    end
+
+    assert_equal 2, client.query_options[:connect_timeout]
+    assert_equal 2, client.query_options[:read_timeout]
+    assert_equal 2, client.query_options[:write_timeout]
+  end
+
   private
 
   def connect_to_mysql!(semian_options = {})
     Mysql2::Client.new(
-      connect_timeout: 1,
-      read_timeout: 1,
+      connect_timeout: 2,
+      read_timeout: 2,
+      write_timeout: 2,
+      reconnect: true,
       host: SemianConfig['toxiproxy_upstream_host'],
       port: SemianConfig['mysql_toxiproxy_port'],
       semian: SEMIAN_OPTIONS.merge(semian_options),

test/test_helper.rb

@@ -6,6 +6,7 @@
 require 'timecop'
 require 'tempfile'
 require 'fileutils'
+require 'byebug'
 
 require 'helpers/background_helper'
 require 'helpers/circuit_breaker_helper'