Add support for multiple endpoints

This commits adds a new `endpoints` parameter for the `BaseClient` class.
This parameter, which must be used alternatively to `host` and `port`,
can be set to a list of EtcdEndpoint, which is a simple data class with
`address` and `port` attributes.

A new decorator is  present: `retry all hosts`. This decorator if
applied to a function of a class descendant from `BaseClient` has the
function calls retried against all the `endpoints` if an exception
occours.
If one of the tries returns with no errors the return value is propagated;
if all the tries throws exceptions of the same type the first one is
propagated; if they throw different exceptions an Etcd3Exception is
thrown. All the failed tries are logged.

Author: dariko

etcd3/aio_client.py

@@ -10,6 +10,7 @@
 import six
 from aiohttp.client import _RequestContextManager
 
+from .baseclient import retry_all_hosts
 from .baseclient import BaseClient
 from .baseclient import BaseModelizedStreamResponse
 from .baseclient import DEFAULT_VERSION
@@ -133,16 +134,17 @@ async def next(self):
 
 
 class AioClient(BaseClient):
-    def __init__(self, host='127.0.0.1', port=2379, protocol='http',
+    def __init__(self, host=None, port=None, endpoints=None, protocol='http',
                  cert=(), verify=None,
                  timeout=None, headers=None, user_agent=None, pool_size=30,
                  username=None, password=None, token=None,
                  server_version=DEFAULT_VERSION, cluster_version=DEFAULT_VERSION):
-        super(AioClient, self).__init__(host=host, port=port, protocol=protocol,
-                                        cert=cert, verify=verify,
-                                        timeout=timeout, headers=headers, user_agent=user_agent, pool_size=pool_size,
-                                        username=username, password=password, token=token,
-                                        server_version=server_version, cluster_version=cluster_version)
+        super(AioClient, self).__init__(
+            host=host, port=port, endpoints=endpoints, protocol=protocol,
+            cert=cert, verify=verify, timeout=timeout, headers=headers,
+            user_agent=user_agent, pool_size=pool_size,
+            username=username, password=password, token=token,
+            server_version=server_version, cluster_version=cluster_version)
         self.ssl_context = None
         if self.cert:
             if verify is False:
@@ -225,6 +227,7 @@ async def _raise_for_status(resp):
             code = data.get('code')
         raise get_client_error(error, code, status, resp)
 
+    @retry_all_hosts
     def call_rpc(self, method, data=None, stream=False, encode=True, raw=False, **kwargs):
         """
         call ETCDv3 RPC and return response object

etcd3/baseclient.py

@@ -17,6 +17,7 @@
 from .apis import MaintenanceAPI
 from .apis import WatchAPI
 from .errors import UnsupportedServerVersion
+from .errors import Etcd3Exception
 from .stateful import Lease
 from .stateful import Lock
 from .stateful import Txn
@@ -25,6 +26,8 @@
 from .swaggerdefs import get_spec
 from .utils import Etcd3Warning
 from .utils import log
+from .utils import check_param
+from .utils import EtcdEndpoint
 from .version import __version__
 
 
@@ -48,15 +51,54 @@ def __iter__(self):
 DEFAULT_VERSION = '3.3.0'
 
 
+def retry_all_hosts(func):
+    def wrapper(self, *args, **kwargs):
+        errors = []
+        got_result = False
+        for i in range(len(self.endpoints)):
+            endpoint = self.endpoints.pop(0)
+            self.endpoints.append(endpoint)
+            self.host = endpoint.host
+            self.port = endpoint.port
+            try:
+                ret = func(self, *args, **kwargs)
+                got_result = True
+                break
+            except Exception as e:
+                errors.append(e)
+                log.warning('Failed to call %s(args: %s, kwargs: %s) on '
+                            'endpoint %s' %
+                            (func.__name__, args, kwargs, endpoint))
+        if not got_result:
+            exception_types = [x.__class__ for x in errors]
+            if len(set(exception_types)) == 1:
+                log.error('Failed to call %s(args: %s, kwargs: %s) on all '
+                          'endpoints: %s. Got errors: %s' %
+                          (func.__name__, args, kwargs, self.endpoints, errors))
+                raise errors[0]
+            else:
+                raise Etcd3Exception(
+                    'Failed to call %s(args: %s, kwargs: %s) on all '
+                    'endpoints: %s. Got errors: %s' %
+                    (func.__name__, args, kwargs, self.endpoints, errors))
+        # elif len(errors) > 0:
+            # log.warning('Got errors %s, retried successfully')
+        return ret
+    return wrapper
+
+
 class BaseClient(AuthAPI, ClusterAPI, KVAPI, LeaseAPI, MaintenanceAPI,
                  WatchAPI, ExtraAPI, LockAPI):
-    def __init__(self, host='127.0.0.1', port=2379, protocol='http',
-                 cert=(), verify=None,
-                 timeout=None, headers=None, user_agent=None, pool_size=30,
+    @check_param(at_most_one_of=['port', 'endpoints'], at_least_one_of=['port', 'endpoints'])
+    @check_param(at_most_one_of=['host', 'endpoints'], at_least_one_of=['host', 'endpoints'])
+    def __init__(self, host=None, port=None, endpoints=None, protocol='http', cert=(),
+                 verify=None, timeout=None, headers=None, user_agent=None, pool_size=30,
                  username=None, password=None, token=None,
                  server_version=DEFAULT_VERSION, cluster_version=DEFAULT_VERSION):
-        self.host = host
-        self.port = port
+        if host is not None:
+            self.endpoints = ([EtcdEndpoint(host, port)])
+        else:
+            self.endpoints = endpoints
         self.cert = cert
         self.protocol = protocol
         if cert:
@@ -79,6 +121,7 @@ def __init__(self, host='127.0.0.1', port=2379, protocol='http',
         self._get_prefix()
         self.api_spec = SwaggerSpec(get_spec(self.server_version))
 
+    @retry_all_hosts
     def _retrieve_version(self):  # pragma: no cover
         try:
             import requests

etcd3/client.py

@@ -8,6 +8,7 @@
 import six
 
 from .baseclient import BaseClient
+from .baseclient import retry_all_hosts
 from .baseclient import BaseModelizedStreamResponse
 from .baseclient import DEFAULT_VERSION
 from .errors import Etcd3Exception
@@ -86,8 +87,8 @@ def iter_response(resp):
 
 
 class Client(BaseClient):
-    def __init__(self, host='127.0.0.1', port=2379, protocol='http',
-                 cert=(), verify=None,
+    def __init__(self, host=None, port=None, endpoints=None,
+                 protocol='http', cert=(), verify=None,
                  timeout=None, headers=None, user_agent=None, pool_size=30,
                  username=None, password=None, token=None, max_retries=0,
                  server_version=DEFAULT_VERSION, cluster_version=DEFAULT_VERSION):
@@ -100,11 +101,12 @@ def __init__(self, host='127.0.0.1', port=2379, protocol='http',
             which we retry a request, import urllib3's ``Retry`` class and pass
             that instead.
         """
-        super(Client, self).__init__(host=host, port=port, protocol=protocol,
-                                     cert=cert, verify=verify,
-                                     timeout=timeout, headers=headers, user_agent=user_agent, pool_size=pool_size,
-                                     username=username, password=password, token=token,
-                                     server_version=server_version, cluster_version=cluster_version)
+        super(Client, self).__init__(
+            host=host, port=port, endpoints=endpoints, protocol=protocol,
+            cert=cert, verify=verify, timeout=timeout, headers=headers,
+            user_agent=user_agent, pool_size=pool_size,
+            username=username, password=password, token=token,
+            server_version=server_version, cluster_version=cluster_version)
         self._session = requests.session()
         self._session.cert = self.cert
         self._session.verify = self.verify
@@ -164,6 +166,7 @@ def _post(self, url, data=None, json=None, **kwargs):
         """
         return self._session.post(url, data=data, json=json, **kwargs)
 
+    @retry_all_hosts
     def call_rpc(self, method, data=None, stream=False, encode=True, raw=False, **kwargs):  # TODO: add modelize param
         """
         call ETCDv3 RPC and return response object

etcd3/utils.py

@@ -382,3 +382,12 @@ def find_executable(executable, path=None):  # pragma: no cover
             f = os.path.join(p, execname)
             if os.path.isfile(f):
                 return f
+
+
+class EtcdEndpoint():
+    def __init__(self, host='127.0.0.1', port=2379):
+        self.host = host
+        self.port = port
+
+    def __repr__(self):
+        return "EtcdEndpoint(host=%s, port=%s)" % (self.host, self.port)