Merge pull request #132 from graywolf-at-work/dont_use_recursion_for_reconnect

Remove automatic retries in case of errors

Author: davissp14

.gitignore

@@ -1,9 +1,10 @@
-Gemfile.lock
-Guardfile
-*.sh
 *.gem
-/tmp/
+*.sh
+*.swp
+/.bundle
 /.idea
 /coverage/*
 /doc/
-*.swp
+/tmp/
+Gemfile.lock
+Guardfile

README.md

@@ -32,8 +32,23 @@ conn = Etcdv3.new(endpoints: 'https://hostname:port', user: 'root', password: 'm
 ```
 **High Availability**
 
-In the event of a failure, the client will work to restore connectivity by cycling through the specified endpoints until a connection can be established.  With that being said, it is encouraged to specify multiple endpoints when available.
+In the event of a failure, the client will work to restore connectivity by cycling through the specified endpoints until a connection can be established.  With that being said, it is encouraged to specify multiple endpoints when available. That's the default
+behaviour.
 
+However, sometimes this is not what you want. If you need more control over
+failures, you can suppress this mechanism by using
+
+```ruby
+conn = Etcdv3.new(endpoints: 'https://hostname:port', allow_reconnect: false)
+```
+
+This will still rotate the endpoints, but it will raise an exception so you can
+handle the failure yourself. On next call new endpoint (since they were
+rotated) is tried. One thing you need to keep in mind if you are using etcd with
+authorization is that you need to take care of `GRPC::Unauthenticated` exceptions
+and manually re-authenticate when token expires. To reiterate, you are
+responsible for handling the errors, so some understanding of how this gem and
+etcd works is recommended.
 
 ## Adding, Fetching and Deleting Keys
 ```ruby

lib/etcdv3.rb

@@ -21,10 +21,14 @@ class Etcdv3
   attr_reader :conn, :credentials, :options
   DEFAULT_TIMEOUT = 120
 
-  def initialize(options = {})
+  def initialize(**options)
     @options = options
     @timeout = options[:command_timeout] || DEFAULT_TIMEOUT
-    @conn = ConnectionWrapper.new(@timeout, *sanitized_endpoints)
+    @conn = ConnectionWrapper.new(
+      @timeout,
+      *sanitized_endpoints,
+      @options.fetch(:allow_reconnect, true),
+    )
     warn "WARNING: `url` is deprecated. Please use `endpoints` instead." if @options.key?(:url)
     authenticate(@options[:user], @options[:password]) if @options.key?(:user)
   end

lib/etcdv3/connection_wrapper.rb

@@ -3,31 +3,40 @@ class ConnectionWrapper
 
     attr_accessor :connection, :endpoints, :user, :password, :token, :timeout
 
-    def initialize(timeout, *endpoints)
+    def initialize(timeout, *endpoints, allow_reconnect)
       @user, @password, @token = nil, nil, nil
       @timeout = timeout
       @endpoints = endpoints.map{|endpoint| Etcdv3::Connection.new(endpoint, @timeout) }
       @connection = @endpoints.first
+      @allow_reconnect = allow_reconnect
+    end
+
+    private def retry_or_raise(*args)
+      if @allow_reconnect
+        handle(*args)
+      else
+        raise
+      end
     end
 
     def handle(stub, method, method_args=[], retries: 1)
       @connection.call(stub, method, method_args)
 
-    rescue GRPC::Unavailable, GRPC::Core::CallError => exception
+    rescue GRPC::Unavailable, GRPC::Core::CallError
       $stderr.puts("Failed to connect to endpoint '#{@connection.hostname}'")
       if @endpoints.size > 1
         rotate_connection_endpoint
         $stderr.puts("Failover event triggered. Failing over to '#{@connection.hostname}'")
-        return handle(stub, method, method_args)
+        return retry_or_raise(stub, method, method_args)
       else
-        return handle(stub, method, method_args)
+        return retry_or_raise(stub, method, method_args)
       end
     rescue GRPC::Unauthenticated => exception
       # Regenerate token in the event it expires.
       if exception.details == 'etcdserver: invalid auth token'
         if retries > 0
           authenticate(@user, @password)
-          return handle(stub, method, method_args, retries: retries - 1)
+          return retry_or_raise(stub, method, method_args, retries: retries - 1)
         end
       end
       raise exception

spec/etcdv3/connection_wrapper_spec.rb

@@ -3,9 +3,9 @@
 describe Etcdv3::ConnectionWrapper do
   let(:conn) { local_connection }
   let(:endpoints) { ['http://localhost:2379', 'http://localhost:2389'] }
+  subject { Etcdv3::ConnectionWrapper.new(10, *endpoints, true) }
 
   describe '#initialize' do
-    subject { Etcdv3::ConnectionWrapper.new(10, *endpoints) }
     it { is_expected.to have_attributes(user: nil, password: nil, token: nil) }
     it 'sets hostnames in correct order' do
       expect(subject.endpoints.map(&:hostname)).to eq(['localhost:2379', 'localhost:2389'])
@@ -16,7 +16,6 @@
   end
 
   describe "#rotate_connection_endpoint" do
-    subject { Etcdv3::ConnectionWrapper.new(10, *endpoints) }
     before do
       subject.rotate_connection_endpoint
     end
@@ -29,11 +28,30 @@
   end
 
   describe "Failover Simulation" do
-    let(:modified_conn) { local_connection("http://localhost:2369, http://localhost:2379") }
+    let(:allow_reconnect) { true }
+    let(:modified_conn) {
+      local_connection(
+        "http://localhost:2369, http://localhost:2379",
+        allow_reconnect: allow_reconnect
+      )
+    }
+    subject { modified_conn.get('boom') }
+
     context 'without auth' do
       # Set primary endpoint to a non-existing etcd endpoint
-      subject { modified_conn.get('boom') }
-      it { is_expected.to be_an_instance_of(Etcdserverpb::RangeResponse) }
+      context 'with reconnect' do
+        it { is_expected.to be_an_instance_of(Etcdserverpb::RangeResponse) }
+      end
+      context 'without reconnect' do
+        let(:allow_reconnect) { false }
+        it { expect { subject }.to raise_error(GRPC::Unavailable) }
+      end
+      context 'without reconnect with single endpoint' do
+        let(:modified_conn) {
+          local_connection("http://localhost:2369",allow_reconnect: false)
+        }
+        it { expect { subject }.to raise_error(GRPC::Unavailable) }
+      end
     end
     context 'with auth' do
       before do
@@ -50,12 +68,19 @@
         modified_conn.auth_disable
         modified_conn.user_delete('root')
       end
-      subject { modified_conn.get('boom') }
-      it { is_expected.to be_an_instance_of(Etcdserverpb::RangeResponse) }
+      context 'with reconnect' do
+        it { is_expected.to be_an_instance_of(Etcdserverpb::RangeResponse) }
+      end
+      context 'without reconnect' do
+        let(:allow_reconnect) { false }
+        it { expect { subject }.to raise_error(GRPC::Unavailable) }
+      end
     end
   end
 
   describe "GRPC::Unauthenticated recovery" do
+    let(:allow_reconnect) { true }
+    let(:conn) { local_connection(allow_reconnect: allow_reconnect) }
     let(:wrapper) { conn.send(:conn) }
     let(:connection) { wrapper.connection }
     before do
@@ -71,6 +96,12 @@
       conn.user_delete('root')
     end
     subject { conn.user_get('root') }
-    it { is_expected.to be_an_instance_of(Etcdserverpb::AuthUserGetResponse) }
+    context 'with reconnect' do
+      it { is_expected.to be_an_instance_of(Etcdserverpb::AuthUserGetResponse) }
+    end
+    context 'without reconnect' do
+      let(:allow_reconnect) { false }
+      it { expect { subject }.to raise_error(GRPC::Unauthenticated) }
+    end
   end
 end

spec/helpers/connections.rb

@@ -5,8 +5,8 @@ def local_connection_with_auth(user, password)
       Etcdv3.new(endpoints: "http://#{local_url}", user: user, password: password)
     end
 
-    def local_connection(endpoints="http://#{local_url}")
-      Etcdv3.new(endpoints: endpoints)
+    def local_connection(endpoints="http://#{local_url}", allow_reconnect: true)
+      Etcdv3.new(endpoints: endpoints, allow_reconnect: allow_reconnect)
     end
 
     def local_connection_with_timeout(timeout)