Multiple endpoints

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
@@ -73,7 +74,7 @@ async def __aenter__(self):
     async def __aexit__(self, exc_type, exc_val, exc_tb):
         self.close()
 
-    async def __aiter__(self):
+    def __aiter__(self):
         return self
 
     async def __anext__(self):
@@ -111,7 +112,7 @@ def __init__(self, resp):
         self.left_chunk = b''
         self.i = 0
 
-    async def __aiter__(self):
+    def __aiter__(self):
         return self
 
     async def next(self):
@@ -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,56 @@ def __iter__(self):
 DEFAULT_VERSION = '3.3.0'
 
 
+def retry_all_hosts(func):
+    def wrapper(self, *args, **kwargs):
+        errors = []
+        got_result = False
+        retries = len(self.endpoints)
+        while retries > 0:
+            retries -= 1
+            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 +123,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)

requirements_dev.txt

@@ -19,3 +19,4 @@ m2r==0.2.1
 codecov>=1.4.0
 codacy-coverage==1.3.11
 twine==1.13.0
+docker==3.7.0

tests/conftest.py

@@ -0,0 +1,69 @@
+from etcd3.client import Client
+import pytest
+from .etcd_cluster import EtcdTestCluster
+
+
+@pytest.fixture(scope='session')
+def etcd_cluster(request):
+    # function_name = request.function.__name__
+    # function_name = re.sub(r"[^a-zA-Z0-9]+", "", function_name)
+    cluster = EtcdTestCluster(ident='cleartext', size=3)
+
+    def fin():
+        cluster.down()
+    request.addfinalizer(fin)
+    cluster.up()
+
+    return cluster
+
+
+@pytest.fixture(scope='session')
+def etcd_cluster_ssl(request):
+    # function_name = request.function.__name__
+    # function_name = re.sub(r"[^a-zA-Z0-9]+", "", function_name)
+    cluster = EtcdTestCluster(ident='ssl', size=3, ssl=True)
+
+    def fin():
+        cluster.down()
+    request.addfinalizer(fin)
+    cluster.up()
+
+    return cluster
+
+
+@pytest.fixture(scope='module')
+def client(etcd_cluster):
+    """
+    init Etcd3Client, close its connection-pool when teardown
+    """
+    # _, p, _ = docker_run_etcd_main()
+    c = Client(endpoints=etcd_cluster.get_endpoints(),
+               protocol='https' if etcd_cluster.ssl else 'http')
+    yield c
+    c.close()
+
+
+@pytest.fixture
+def clear(etcd_cluster):
+    def _clear():
+        etcd_cluster.etcdctl('del --from-key ""')
+    return _clear
+
+
+def teardown_auth(etcd_cluster):  # pragma: no cover
+    """
+    disable auth, delete all users and roles
+    """
+    etcd_cluster.etcdctl('--user root:root auth disable')
+    etcd_cluster.etcdctl('--user root:changed auth disable')
+    for i in (etcd_cluster.etcdctl('role list') or '').splitlines():
+        etcd_cluster.etcdctl('role delete %s' % i)
+    for i in (etcd_cluster.etcdctl('user list') or '').splitlines():
+        etcd_cluster.etcdctl('user delete %s' % i)
+
+
+def enable_auth(etcd_cluster):  # pragma: no cover
+    etcd_cluster.etcdctl('user add root:root')
+    etcd_cluster.etcdctl('role add root')
+    etcd_cluster.etcdctl('user grant root root')
+    etcd_cluster.etcdctl('auth enable')