Multiple endpoints

.travis.yml

@@ -36,6 +36,8 @@ before_install:
   - sudo docker cp tmp:/usr/local/bin/etcdctl /usr/bin/etcdctl && sudo chmod 755 /usr/bin/etcdctl
   - sudo docker rm tmp
   - which etcdctl
+  - sudo mkdir /tmp/shared
+  - sudo chmod 777 /tmp/shared
 
 # command to install dependencies, e.g. pip install -r requirements.txt --use-mirrors
 install: pip install -U tox-travis codecov

etcd3/aio_client.py

@@ -9,7 +9,9 @@
 import aiohttp
 import six
 from aiohttp.client import _RequestContextManager
+from asyncio import Lock
 
+from .utils import retry_all_hosts
 from .baseclient import BaseClient
 from .baseclient import BaseModelizedStreamResponse
 from .baseclient import DEFAULT_VERSION
@@ -73,7 +75,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 +113,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 +135,18 @@ 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)
+        self.current_endpoint_lock = Lock()
+        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:
@@ -165,7 +169,8 @@ def __init__(self, host='127.0.0.1', port=2379, protocol='http',
                 warnings.warn(Etcd3Warning("the openssl version of your python is too old to support TLSv1.1+,"
                                            "please upgrade you python"))
             ssl_context.verify_mode = cert_reqs
-            ssl_context.load_verify_locations(cafile=cafile)
+            if cafile:
+                ssl_context.load_verify_locations(cafile=cafile)
             ssl_context.load_cert_chain(*self.cert)
         connector = aiohttp.TCPConnector(limit=pool_size, ssl=self.ssl_context)
         self.session = aiohttp.ClientSession(connector=connector)
@@ -225,6 +230,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

@@ -25,6 +25,7 @@
 from .swaggerdefs import get_spec
 from .utils import Etcd3Warning
 from .utils import log
+from .utils import EtcdEndpoint
 from .version import __version__
 
 
@@ -50,13 +51,18 @@ def __iter__(self):
 
 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,
+    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
+                 server_version=DEFAULT_VERSION, cluster_version=DEFAULT_VERSION,
+                 failover_whitelist=None):
+        if failover_whitelist is None:
+            failover_whitelist = ["range", "watch", "status"]
+        if host is not None:
+            self.endpoints = ([EtcdEndpoint(host, port)])
+        else:
+            self.endpoints = endpoints
+        self.current_endpoint = self.endpoints[0]
         self.cert = cert
         self.protocol = protocol
         if cert:
@@ -74,6 +80,7 @@ def __init__(self, host='127.0.0.1', port=2379, protocol='http',
         self.cluster_version = cluster_version
         self.api_spec = None
         self.api_prefix = '/v3alpha'
+        self.failover_whitelist = failover_whitelist
         self._retrieve_version()
         self._verify_version()
         self._get_prefix()
@@ -123,7 +130,9 @@ def baseurl(self):
         """
         :return: baseurl from protocol, host, self
         """
-        return '{}://{}:{}'.format(self.protocol, self.host, self.port)
+        return '{}://{}:{}'.format(self.protocol,
+                                   self.current_endpoint.host,
+                                   self.current_endpoint.port)
 
     def _prefix(self, method):
         return self.api_prefix + method

etcd3/client.py

@@ -6,8 +6,10 @@
 
 import requests
 import six
+from threading import Lock
 
 from .baseclient import BaseClient
+from .utils import retry_all_hosts
 from .baseclient import BaseModelizedStreamResponse
 from .baseclient import DEFAULT_VERSION
 from .errors import Etcd3Exception
@@ -86,8 +88,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 +102,13 @@ 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)
+        self.current_endpoint_lock = Lock()
+        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 +168,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

@@ -7,9 +7,18 @@
 import sys
 import time
 import warnings
+import copy
+import inspect
+import socket
 from collections import namedtuple, OrderedDict, Hashable
 from subprocess import Popen, PIPE
 from threading import Lock
+from requests.exceptions import ChunkedEncodingError, ConnectTimeout
+from urllib3.exceptions import MaxRetryError
+try:
+    from httplib import IncompleteRead
+except ImportError:
+    from http.client import IncompleteRead
 
 try:  # pragma: no cover
     from inspect import getfullargspec as getargspec
@@ -382,3 +391,70 @@ def find_executable(executable, path=None):  # pragma: no cover
             f = os.path.join(p, execname)
             if os.path.isfile(f):
                 return f
+
+failover_exceptions = (ChunkedEncodingError,
+                       IncompleteRead,
+                       ConnectTimeout,
+                       MaxRetryError,
+                       socket.timeout,
+                       )
+
+def retry_all_hosts(func):
+    def current_caller_name():
+        previous_frame = inspect.currentframe().f_back.f_back
+        return inspect.getframeinfo(previous_frame).function
+    def wrapper(self, *args, **kwargs):
+        calling_function = current_caller_name()
+        if not calling_function in self.failover_whitelist:
+            log.debug('%s not in failover waitlist(%s)' %
+                      (calling_function, self.failover_whitelist))
+        try:
+            return func(self, *args, **kwargs)
+        except failover_exceptions as e:
+            if not calling_function in self.failover_whitelist:
+                log.debug('%s not in failover waitlist(%s)' %
+                          (calling_function, self.failover_whitelist))
+                raise e
+            errors = []
+            got_result = False
+            with self.current_endpoint_lock:
+                # to exclude the current endpoint it should be saved
+                # before the call in the outmost `try`, and the
+                # whole wrapper should depend on the lock
+                for endpoint in self.endpoints:
+                    self.current_endpoint = endpoint
+                    try:
+                        self.current_endpoint = endpoint
+                        ret = func(self, *args, **kwargs)
+                        got_result = True
+                        break
+                    except failover_exceptions as e:
+                        errors.append(e)
+                        log.warning('Failed to call %s(args: %s, kwargs: %s) on'
+                                    ' endpoint %s (%s)' %
+                                    (func.__name__, args, kwargs, endpoint, e))
+                    except Exception as e:
+                        log.debug('received exception %s, not in '
+                                  'failover_exceptions(%s)' %
+                                  (e, failover_exceptions))
+                if not got_result:
+                    log.error('Failed to call %s(args: %s, kwargs: %s) on all '
+                              'endpoints: %s. Got errors: %s' %
+                              (func.__name__, args, kwargs,
+                               call_endpoints, errors))
+                    exception_types = [x.__class__ for x in errors]
+                    if len(set(exception_types)) == 1:
+                        raise errors[0]
+                    else:
+                        raise Etcd3Exception('Failed failover')
+            return ret
+    return wrapper
+
+
+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,76 @@
+import six
+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()
+    cluster.wait_ready()
+
+    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()
+    cluster.wait_ready()
+
+    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():
+        if six.PY3:  # pragma: no cover
+            i = six.text_type(i, encoding='utf-8')
+        etcd_cluster.etcdctl('role delete %s' % i)
+    for i in (etcd_cluster.etcdctl('user list') or '').splitlines():
+        if six.PY3:  # pragma: no cover
+            i = six.text_type(i, encoding='utf-8')
+        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')