Add Etcd v3 protocol support via api gRPC-gateway (#1162)

The only python-etcd3 client working directly via gRPC still supports only a single endpoint, which is not very nice for high-availability.

Since Patroni is already using a heavily hacked version of python-etcd with smart retries and auto-discovery out-of-the-box, I decided to enhance the existing code with limited support of v3 protocol via gRPC-gateway.

Unfortunately, watches via gRPC-gateway requires us to open and keep the second connection to the etcd.

Known limitations:
* The very minimal supported version is 3.0.4. On earlier versions transactions don't work due to bugs in grpc-gateway. Without transactions we can't do atomic operations, i.e. leader locks.
* Watches work only starting from 3.1.0
* Authentication works only starting from 3.3.0
* gRPC-gateway does not support authentication using TLS Common Name. This is because gRPC-proxy terminates TLS from its client so all the clients share a cert of the proxy: https://github.com/etcd-io/etcd/blob/master/Documentation/op-guide/authentication.md#using-tls-common-name

.travis.yml

@@ -7,16 +7,17 @@ addons:
       - expect-dev # for unbuffer
 env:
   global:
-  - ETCDVERSION=3.0.17 ZKVERSION=3.4.14 CONSULVERSION=0.7.4
-  - PYVERSIONS="2.7 3.5 3.6"
-  - EXCLUDE_BEHAVE="3.5"
+  - ETCDVERSION=3.3.13 ZKVERSION=3.4.14 CONSULVERSION=0.7.4
+  - PYVERSIONS="2.7 3.6"
   - BOTO_CONFIG=/doesnotexist
 matrix:
   include:
   - python: "3.5"
-    env: TEST_SUITE="python setup.py"
+    env: PYVERSIONS="2.7 3.5 3.6" TEST_SUITE="python setup.py"
   - python: "3.6"
     env: DCS="etcd" TEST_SUITE="behave"
+  - python: "3.6"
+    env: DCS="etcd3" TEST_SUITE="behave"
   - python: "3.6"
     env: DCS="exhibitor" TEST_SUITE="behave"
   - python: "3.6"
@@ -36,10 +37,8 @@ before_cache:
 - |
   rm -fr $HOME/mycache/python*
   for pv in $PYVERSIONS; do
-      if [[ $TEST_SUITE != "behave" || $pv != $EXCLUDE_BEHAVE ]]; then
-          fpv=$(basename $(readlink $HOME/virtualenv/python${pv}))
-          mv $HOME/virtualenv/${fpv} $HOME/mycache/${fpv}
-      fi
+      fpv=$(basename $(readlink $HOME/virtualenv/python${pv}))
+      mv $HOME/virtualenv/${fpv} $HOME/mycache/${fpv}
   done
 install:
   - |
@@ -60,6 +59,7 @@ install:
         function get_etcd() {
             EC=~/mycache/etcd_${ETCDVERSION}
             if [[ ! -x $EC ]]; then
+                rm -rf ~/mycache/etcd_*
                 curl -L https://github.com/coreos/etcd/releases/download/v${ETCDVERSION}/etcd-v${ETCDVERSION}-linux-amd64.tar.gz \
                     | tar xz -C . --strip=1 --wildcards --no-anchored etcd
                 [[ ${PIPESTATUS[0]} == 0 ]] || return 1
@@ -68,6 +68,10 @@ install:
             ln -s $EC etcd
         }
 
+        function get_etcd3() {
+            get_etcd
+        }
+
         function get_kubernetes() {
             wget -O localkube "https://storage.googleapis.com/minikube/k8sReleases/v1.7.0/localkube-linux-amd64"
             chmod +x localkube
@@ -117,41 +121,38 @@ install:
     fi
 
     for pv in $PYVERSIONS; do
-        if [[ $TEST_SUITE != "behave" || $pv != $EXCLUDE_BEHAVE ]]; then
-            fpv=$(basename $(readlink $HOME/virtualenv/python$pv))
-            if [[ -d ~/mycache/${fpv} ]]; then
-                mv ~/virtualenv/${fpv} ~/virtualenv/${fpv}.bckp
-                mv ~/mycache/${fpv} ~/virtualenv/${fpv}
-            fi
-            source ~/virtualenv/python${pv}/bin/activate
-            # explicitly install all needed python modules to cache them
-            for p in '-r requirements.txt' 'psycopg2-binary behave codacy-coverage coverage coveralls flake8 mock pytest-cov pytest setuptools'; do
-                pip install $p --upgrade
-            done
+        fpv=$(basename $(readlink $HOME/virtualenv/python$pv))
+        if [[ -d ~/mycache/${fpv} ]]; then
+            mv ~/virtualenv/${fpv} ~/virtualenv/${fpv}.bckp
+            mv ~/mycache/${fpv} ~/virtualenv/${fpv}
         fi
+        source ~/virtualenv/python${pv}/bin/activate
+        # explicitly install all needed python modules to cache them
+        for p in '-r requirements.txt' 'psycopg2-binary behave codacy-coverage coverage coveralls flake8 mock pytest-cov pytest setuptools'; do
+            pip install $p --upgrade
+        done
     done
 script:
   - |
     for pv in $PYVERSIONS; do
-        if [[ $TEST_SUITE == "behave" && $pv == $EXCLUDE_BEHAVE ]]; then
-            continue
-        fi
-
         source ~/virtualenv/python${pv}/bin/activate
 
-        if [[ $TEST_SUITE != "behave" ]]; then
-            echo Running unit tests using python${pv}
-            unbuffer $TEST_SUITE test
-            $TEST_SUITE flake8
-        elif [[ $pv != $EXCLUDE_BEHAVE ]]; then
-            echo Running acceptance tests using python${pv}
+        if [[ $TEST_SUITE = "behave" ]]; then
+            echo Running integration tests using python${pv}
             if ! PATH=.:/usr/lib/postgresql/9.6/bin:$PATH unbuffer $TEST_SUITE; then
                 # output all log files when tests are failing
                 grep . features/output/*_failed/*postgres?.*
                 exit 1
             fi
+        else
+            echo Running unit tests using python${pv}
+            unbuffer $TEST_SUITE test
+            $TEST_SUITE flake8
         fi
+        mv .coverage /tmp/.coverage.$pv
     done
+    mv /tmp/.coverage.* .
+    python -m coverage combine
 
     set +e
 after_success:

Dockerfile

@@ -14,7 +14,7 @@ ARG PGDATA
 ARG LC_ALL
 ARG LANG
 
-ENV ETCDVERSION=2.3.8 CONFDVERSION=0.16.0
+ENV ETCDVERSION=3.3.13 CONFDVERSION=0.16.0
 
 RUN set -ex \
     && export DEBIAN_FRONTEND=noninteractive \

README.rst

@@ -96,7 +96,7 @@ Patroni can be installed with pip:
 
 where dependencies can be either empty, or consist of one or more of the following:
 
-etcd
+etcd or etcd3
     `python-etcd` module in order to use Etcd as DCS
 consul
     `python-consul` module in order to use Consul as DCS
@@ -106,6 +106,8 @@ exhibitor
     `kazoo` module in order to use Exhibitor as DCS (same dependencies as for Zookeeper)
 kubernetes
     `kubernetes` module in order to use Kubernetes as DCS in Patroni
+raft
+    `pysyncobj` module in order to use python Raft implementation as DCS
 aws
     `boto` in order to use AWS callbacks
 

docker/entrypoint.sh

@@ -25,7 +25,7 @@ case "$1" in
             while ! etcdctl cluster-health 2> /dev/null; do
                 sleep 1
             done
-            exec dumb-init $CONFD etcd -node $(echo $ETCDCTL_ENDPOINTS | sed 's/,/ -node /g')
+            exec dumb-init $CONFD etcdv3 -node $(echo $ETCDCTL_ENDPOINTS | sed 's/,/ -node /g')
         fi
         ;;
     etcd)
@@ -37,7 +37,7 @@ case "$1" in
 esac
 
 ## We start an etcd
-if [ -z "$PATRONI_ETCD_HOSTS" ] && [ -z "$PATRONI_ZOOKEEPER_HOSTS" ]; then
+if [ -z "$PATRONI_ETCD3_HOSTS" ] && [ -z "$PATRONI_ZOOKEEPER_HOSTS" ]; then
     export PATRONI_ETCD_URL="http://127.0.0.1:2379"
     etcd --data-dir /tmp/etcd.data -advertise-client-urls=$PATRONI_ETCD_URL -listen-client-urls=http://0.0.0.0:2379 > /var/log/etcd.log 2> /var/log/etcd.err &
 fi

docker/patroni.env

@@ -1,5 +1,5 @@
 PATRONI_SCOPE=demo
-PATRONI_ETCD_HOSTS='etcd1:2379','etcd2:2379','etcd3:2379'
+PATRONI_ETCD3_HOSTS='etcd1:2379','etcd2:2379','etcd3:2379'
 
 PATRONI_RESTAPI_USERNAME=admin
 PATRONI_RESTAPI_PASSWORD=admin

docs/ENVIRONMENT.rst

@@ -66,6 +66,14 @@ Etcd
 -  **PATRONI\_ETCD\_CERT**: File with the client certificate.
 -  **PATRONI\_ETCD\_KEY**: File with the client key. Can be empty if the key is part of certificate.
 
+Etcdv3
+------
+Environment names for Etcdv3 are similar as for Etcd, you just need to use ``ETCD3`` instead of ``ETCD`` in the variable name. Example: ``PATRONI_ETCD3_HOST``, ``PATRONI_ETCD3_CACERT``, and so on.
+
+.. warning::
+    Keys created with protocol version 2 are not visible with protocol version 3 and the other way around, therefore it is not possible to switch from Etcd to Etcdv3 just by updating Patroni configuration.
+
+
 ZooKeeper
 ---------
 -  **PATRONI\_ZOOKEEPER\_HOSTS**: comma separated list of ZooKeeper cluster members: "'host1:port1','host2:port2','etc...'". It is important to quote every single entity!

docs/README.rst

@@ -72,7 +72,7 @@ Patroni can be installed with pip:
 
 where dependencies can be either empty, or consist of one or more of the following:
 
-etcd
+etcd or etcd3
     `python-etcd` module in order to use Etcd as DCS
 consul
     `python-consul` module in order to use Consul as DCS
@@ -82,6 +82,8 @@ exhibitor
     `kazoo` module in order to use Exhibitor as DCS (same dependencies as for Zookeeper)
 kubernetes
     `kubernetes` module in order to use Kubernetes as DCS in Patroni
+raft
+    `pysyncobj` module in order to use python Raft implementation as DCS
 aws
     `boto` in order to use AWS callbacks
 

docs/SETTINGS.rst

@@ -137,6 +137,14 @@ Most of the parameters are optional, but you have to specify one of the **host**
 -  **cert**: (optional) file with the client certificate.
 -  **key**: (optional) file with the client key. Can be empty if the key is part of **cert**.
 
+Etcdv3
+------
+If you want that Patroni works with Etcd cluster via protocol version 3, you need to use the ``etcd3`` section in the Patroni configuration file. All configuration parameters are the same as for ``etcd``.
+
+.. warning::
+    Keys created with protocol version 2 are not visible with protocol version 3 and the other way around, therefore it is not possible to switch from ``etcd`` to ``etcd3`` just by updating Patroni config file.
+
+
 ZooKeeper
 ----------
 -  **hosts**: list of ZooKeeper cluster members in format: ['host1:port1', 'host2:port2', 'etc...'].

features/environment.py

@@ -378,21 +378,36 @@ class ConsulController(AbstractDcsController):
         super(ConsulController, self).start(max_wait_limit)
 
 
-class EtcdController(AbstractDcsController):
+class AbstractEtcdController(AbstractDcsController):
 
     """ handles all etcd related tasks, used for the tests setup and cleanup """
 
-    def __init__(self, context):
-        super(EtcdController, self).__init__(context)
-        os.environ['PATRONI_ETCD_HOST'] = 'localhost:2379'
-
-        import etcd
-        self._client = etcd.Client(port=2379)
+    def __init__(self, context, client_cls):
+        super(AbstractEtcdController, self).__init__(context)
+        self._client_cls = client_cls
 
     def _start(self):
         return subprocess.Popen(["etcd", "--debug", "--data-dir", self._work_directory],
                                 stdout=self._log, stderr=subprocess.STDOUT)
 
+    def _is_running(self):
+        from patroni.dcs.etcd import DnsCachingResolver
+        # if etcd is running, but we didn't start it
+        try:
+            self._client = self._client_cls({'host': 'localhost', 'port': 2379, 'retry_timeout': 30,
+                                             'patronictl': 1}, DnsCachingResolver())
+            return True
+        except Exception:
+            return False
+
+
+class EtcdController(AbstractEtcdController):
+
+    def __init__(self, context):
+        from patroni.dcs.etcd import EtcdClient
+        super(EtcdController, self).__init__(context, EtcdClient)
+        os.environ['PATRONI_ETCD_HOST'] = 'localhost:2379'
+
     def query(self, key, scope='batman'):
         import etcd
         try:
@@ -409,12 +424,25 @@ class EtcdController(AbstractDcsController):
         except Exception as e:
             assert False, "exception when cleaning up etcd contents: {0}".format(e)
 
-    def _is_running(self):
-        # if etcd is running, but we didn't start it
+
+class Etcd3Controller(AbstractEtcdController):
+
+    def __init__(self, context):
+        from patroni.dcs.etcd3 import Etcd3Client
+        super(Etcd3Controller, self).__init__(context, Etcd3Client)
+        os.environ['PATRONI_ETCD3_HOST'] = 'localhost:2379'
+
+    def query(self, key, scope='batman'):
+        import base64
+        response = self._client.range(self.path(key, scope))
+        for k in response.get('kvs', []):
+            return base64.b64decode(k['value']).decode('utf-8') if 'value' in k else None
+
+    def cleanup_service_tree(self):
         try:
-            return bool(self._client.machines)
-        except Exception:
-            return False
+            self._client.deleteprefix(self.path(scope=''))
+        except Exception as e:
+            assert False, "exception when cleaning up etcd contents: {0}".format(e)
 
 
 class KubernetesController(AbstractDcsController):

patroni/config.py

@@ -316,8 +316,9 @@ class Config(object):
                         value = parse_bool(value)
                     if value:
                         ret[name.lower()][suffix.lower()] = value
-        if 'etcd' in ret:
-            ret['etcd'].update(_get_auth('etcd'))
+        for dcs in ('etcd', 'etcd3'):
+            if dcs in ret:
+                ret[dcs].update(_get_auth(dcs))
 
         users = {}
         for param in list(os.environ.keys()):

patroni/dcs/etcd.py

@@ -1,4 +1,5 @@
 from __future__ import absolute_import
+import abc
 import etcd
 import json
 import logging
@@ -82,7 +83,8 @@ class DnsCachingResolver(Thread):
             return []
 
 
-class Client(etcd.Client):
+@six.add_metaclass(abc.ABCMeta)
+class AbstractEtcdClientWithFailover(etcd.Client):
 
     def __init__(self, config, dns_resolver, cache_ttl=300):
         self._dns_resolver = dns_resolver
@@ -90,7 +92,7 @@ class Client(etcd.Client):
         self._machines_cache_updated = 0
         args = {p: config.get(p) for p in ('host', 'port', 'protocol', 'use_proxies', 'username', 'password',
                                            'cert', 'ca_cert') if config.get(p)}
-        super(Client, self).__init__(read_timeout=config['retry_timeout'], **args)
+        super(AbstractEtcdClientWithFailover, self).__init__(read_timeout=config['retry_timeout'], **args)
         # For some reason python3-etcd on debian and ubuntu are not based on the latest version
         # Workaround for the case when https://github.com/jplana/python-etcd/pull/196 is not applied
         self.http.connection_pool_kw.pop('ssl_version', None)
@@ -130,12 +132,16 @@ class Client(etcd.Client):
 
         return etcd_nodes, per_node_timeout, per_node_retries - 1
 
+    def reload_config(self, config):
+        self.username = config.get('username')
+        self.password = config.get('password')
+
     def _get_headers(self):
         basic_auth = ':'.join((self.username, self.password)) if self.username and self.password else None
         return urllib3.make_headers(basic_auth=basic_auth, user_agent=USER_AGENT)
 
-    def _build_request_parameters(self, etcd_nodes, timeout=None):
-        kwargs = {'headers': self._get_headers(), 'redirect': self.allow_redirect}
+    def _prepare_common_parameters(self, etcd_nodes, timeout=None):
+        kwargs = {'headers': self._get_headers(), 'redirect': self.allow_redirect, 'preload_content': False}
 
         if timeout is not None:
             kwargs.update(retries=0, timeout=timeout)
@@ -148,6 +154,14 @@ class Client(etcd.Client):
     def set_machines_cache_ttl(self, cache_ttl):
         self._machines_cache_ttl = cache_ttl
 
+    @abc.abstractmethod
+    def _prepare_get_members(self, etcd_nodes):
+        """returns: request parameters"""
+
+    @abc.abstractmethod
+    def _get_members(self, base_uri, **kwargs):
+        """returns: list of clientURLs"""
+
     @property
     def machines_cache(self):
         base_uri, cache = self._base_uri, self._machines_cache
@@ -166,13 +180,11 @@ class Client(etcd.Client):
         the original method was retrying 2 times with the `read_timeout` on each node."""
 
         machines_cache = self.machines_cache
-        kwargs = self._build_request_parameters(len(machines_cache))
+        kwargs = self._prepare_get_members(len(machines_cache))
 
         for base_uri in machines_cache:
             try:
-                response = self.http.request(self._MGET, base_uri + self.version_prefix + '/machines', **kwargs)
-                data = self._handle_server_response(response).data.decode('utf-8')
-                machines = [m.strip() for m in data.split(',') if m.strip()]
+                machines = list(self._get_members(base_uri, **kwargs))
                 logger.debug("Retrieved list of machines: %s", machines)
                 if machines:
                     random.shuffle(machines)
@@ -188,14 +200,15 @@ class Client(etcd.Client):
         self._read_timeout = timeout
 
     def _do_http_request(self, retry, machines_cache, request_executor, method, path, fields=None, **kwargs):
+        if fields is not None:
+            kwargs['fields'] = fields
         some_request_failed = False
         for i, base_uri in enumerate(machines_cache):
             if i > 0:
                 logger.info("Retrying on %s", base_uri)
             try:
-                response = request_executor(method, base_uri + path, fields=fields, **kwargs)
+                response = request_executor(method, base_uri + path, **kwargs)
                 response.data.decode('utf-8')
-                self._check_cluster_id(response)
                 if some_request_failed:
                     self.set_base_uri(base_uri)
                     self._refresh_machines_cache()
@@ -218,20 +231,12 @@ class Client(etcd.Client):
 
         raise etcd.EtcdConnectionFailed('No more machines in the cluster')
 
+    @abc.abstractmethod
+    def _prepare_request(self, kwargs, params=None, method=None):
+        """returns: request_executor"""
+
     def api_execute(self, path, method, params=None, timeout=None):
-        if not path.startswith('/'):
-            raise ValueError('Path does not start with /')
-
         retry = params.pop('retry', None) if isinstance(params, dict) else None
-        kwargs = {'fields': params, 'preload_content': False}
-
-        if method in [self._MGET, self._MDELETE]:
-            request_executor = self.http.request
-        elif method in [self._MPUT, self._MPOST]:
-            request_executor = self.http.request_encode_body
-            kwargs['encode_multipart'] = False
-        else:
-            raise etcd.EtcdException('HTTP method {0} not supported'.format(method))
 
         # Update machines_cache if previous attempt of update has failed
         if self._update_machines_cache:
@@ -241,7 +246,9 @@ class Client(etcd.Client):
 
         machines_cache = self.machines_cache
         etcd_nodes = len(machines_cache)
-        kwargs.update(self._build_request_parameters(etcd_nodes, timeout))
+
+        kwargs = self._prepare_common_parameters(etcd_nodes, timeout)
+        request_executor = self._prepare_request(kwargs, params, method)
 
         while True:
             try:
@@ -389,17 +396,47 @@ class Client(etcd.Client):
         self._base_uri = value
 
 
-class Etcd(AbstractDCS):
+class EtcdClient(AbstractEtcdClientWithFailover):
 
-    def __init__(self, config):
-        super(Etcd, self).__init__(config)
-        self._ttl = int(config.get('ttl') or 30)
+    ERROR_CLS = EtcdError
+
+    def __del__(self):
+        if self.http is not None:
+            try:
+                self.http.clear()
+            except (ReferenceError, TypeError):
+                pass
+
+    def _prepare_get_members(self, etcd_nodes):
+        return self._prepare_common_parameters(etcd_nodes)
+
+    def _get_members(self, base_uri, **kwargs):
+        response = self.http.request(self._MGET, base_uri + self.version_prefix + '/machines', **kwargs)
+        data = self._handle_server_response(response).data.decode('utf-8')
+        return [m.strip() for m in data.split(',') if m.strip()]
+
+    def _prepare_request(self, kwargs, params=None, method=None):
+        kwargs['fields'] = params
+        if method in (self._MPOST, self._MPUT):
+            kwargs['encode_multipart'] = False
+        return self.http.request
+
+
+class AbstractEtcd(AbstractDCS):
+
+    def __init__(self, config, client_cls, retry_errors_cls):
+        super(AbstractEtcd, self).__init__(config)
         self._retry = Retry(deadline=config['retry_timeout'], max_delay=1, max_tries=-1,
-                            retry_exceptions=(etcd.EtcdLeaderElectionInProgress, EtcdRaftInternal))
-        self._client = self.get_etcd_client(config)
+                            retry_exceptions=retry_errors_cls)
+        self._ttl = int(config.get('ttl') or 30)
+        self._client = self.get_etcd_client(config, client_cls)
         self.__do_not_watch = False
         self._has_failed = False
 
+    def reload_config(self, config):
+        super(AbstractEtcd, self).reload_config(config)
+        self._client.reload_config(config.get(self.__class__.__name__.lower(), {}))
+
     def retry(self, *args, **kwargs):
         retry = self._retry.copy()
         kwargs['retry'] = retry
@@ -416,22 +453,13 @@ class Etcd(AbstractDCS):
         if isinstance(raise_ex, Exception):
             raise raise_ex
 
-    def catch_etcd_errors(func):
-        def wrapper(self, *args, **kwargs):
-            try:
-                retval = func(self, *args, **kwargs) is not None
-                self._has_failed = False
-                return retval
-            except (RetryFailedError, etcd.EtcdException) as e:
-                self._handle_exception(e)
-                return False
-            except Exception as e:
-                self._handle_exception(e, raise_ex=EtcdError('unexpected error'))
-
-        return wrapper
-
     @staticmethod
-    def get_etcd_client(config):
+    def set_socket_options(sock, socket_options):
+        if socket_options:
+            for opt in socket_options:
+                sock.setsockopt(*opt)
+
+    def get_etcd_client(self, config, client_cls):
         if 'proxy' in config:
             config['use_proxies'] = True
             config['url'] = config['proxy']
@@ -480,9 +508,7 @@ class Etcd(AbstractDCS):
                 sock = None
                 try:
                     sock = socket.socket(af, socktype, proto)
-                    if socket_options:
-                        for opt in socket_options:
-                            sock.setsockopt(*opt)
+                    self.set_socket_options(sock, socket_options)
                     if timeout is not socket._GLOBAL_DEFAULT_TIMEOUT:
                         sock.settimeout(timeout)
                     if source_address:
@@ -506,7 +532,7 @@ class Etcd(AbstractDCS):
         client = None
         while not client:
             try:
-                client = Client(config, dns_resolver)
+                client = client_cls(config, dns_resolver)
                 if 'use_proxies' in config and not client.machines:
                     raise etcd.EtcdException
             except etcd.EtcdException:
@@ -516,9 +542,10 @@ class Etcd(AbstractDCS):
 
     def set_ttl(self, ttl):
         ttl = int(ttl)
-        self.__do_not_watch = self._ttl != ttl
+        ret = self._ttl != ttl
         self._ttl = ttl
         self._client.set_machines_cache_ttl(ttl*10)
+        return ret
 
     @property
     def ttl(self):
@@ -528,6 +555,31 @@ class Etcd(AbstractDCS):
         self._retry.deadline = retry_timeout
         self._client.set_read_timeout(retry_timeout)
 
+
+def catch_etcd_errors(func):
+    def wrapper(self, *args, **kwargs):
+        try:
+            retval = func(self, *args, **kwargs) is not None
+            self._has_failed = False
+            return retval
+        except (RetryFailedError, etcd.EtcdException) as e:
+            self._handle_exception(e)
+            return False
+        except Exception as e:
+            self._handle_exception(e, raise_ex=self._client.ERROR_CLS('unexpected error'))
+
+    return wrapper
+
+
+class Etcd(AbstractEtcd):
+
+    def __init__(self, config):
+        super(Etcd, self).__init__(config, EtcdClient, (etcd.EtcdLeaderElectionInProgress, EtcdRaftInternal))
+        self.__do_not_watch = False
+
+    def set_ttl(self, ttl):
+        self.__do_not_watch = super(Etcd, self).set_ttl(ttl)
+
     @staticmethod
     def member(node):
         return Member.from_node(node.modifiedIndex, os.path.basename(node.key), node.ttl, node.value)

patroni/dcs/etcd3.py

@@ -0,0 +1,795 @@
+from __future__ import absolute_import
+import base64
+import etcd
+import json
+import logging
+import os
+import six
+import socket
+import sys
+import time
+import urllib3
+
+from threading import Condition, Lock, Thread
+
+from . import ClusterConfig, Cluster, Failover, Leader, Member, SyncState, TimelineHistory
+from .etcd import AbstractEtcdClientWithFailover, AbstractEtcd, catch_etcd_errors
+from ..exceptions import DCSError, PatroniException
+from ..utils import deep_compare, enable_keepalive, iter_response_objects, RetryFailedError, USER_AGENT
+
+logger = logging.getLogger(__name__)
+
+
+class Etcd3Error(DCSError):
+    pass
+
+
+class UnsupportedEtcdVersion(PatroniException):
+    pass
+
+
+# google.golang.org/grpc/codes
+GRPCCode = type('Enum', (), {'OK': 0, 'Canceled': 1, 'Unknown': 2, 'InvalidArgument': 3, 'DeadlineExceeded': 4,
+                             'NotFound': 5, 'AlreadyExists': 6, 'PermissionDenied': 7, 'ResourceExhausted': 8,
+                             'FailedPrecondition': 9, 'Aborted': 10, 'OutOfRange': 11, 'Unimplemented': 12,
+                             'Internal': 13, 'Unavailable': 14, 'DataLoss': 15, 'Unauthenticated': 16})
+GRPCcodeToText = {v: k for k, v in GRPCCode.__dict__.items() if not k.startswith('__') and isinstance(v, int)}
+
+
+class Etcd3Exception(etcd.EtcdException):
+    pass
+
+
+class Etcd3ClientError(Etcd3Exception):
+
+    def __init__(self, code=None, error=None, status=None):
+        if not hasattr(self, 'error'):
+            self.error = error and error.strip()
+        self.codeText = GRPCcodeToText.get(code)
+        self.status = status
+
+    def __repr__(self):
+        return "<{0} error: '{1}', code: {2}>".format(self.__class__.__name__, self.error, self.code)
+
+    __str__ = __repr__
+
+    def as_dict(self):
+        return {'error': self.error, 'code': self.code, 'codeText': self.codeText, 'status': self.status}
+
+    @classmethod
+    def get_subclasses(cls):
+        for subclass in cls.__subclasses__():
+            for subsubclass in subclass.get_subclasses():
+                yield subsubclass
+            yield subclass
+
+
+class Unknown(Etcd3ClientError):
+    code = GRPCCode.Unknown
+
+
+class InvalidArgument(Etcd3ClientError):
+    code = GRPCCode.InvalidArgument
+
+
+class DeadlineExceeded(Etcd3ClientError):
+    code = GRPCCode.DeadlineExceeded
+    error = "context deadline exceeded"
+
+
+class NotFound(Etcd3ClientError):
+    code = GRPCCode.NotFound
+
+
+class FailedPrecondition(Etcd3ClientError):
+    code = GRPCCode.FailedPrecondition
+
+
+class Unavailable(Etcd3ClientError):
+    code = GRPCCode.Unavailable
+
+
+# https://github.com/etcd-io/etcd/blob/master/etcdserver/api/v3rpc/rpctypes/error.go
+class LeaseNotFound(NotFound):
+    error = "etcdserver: requested lease not found"
+
+
+class UserEmpty(InvalidArgument):
+    error = "etcdserver: user name is empty"
+
+
+class PermissionDenied(Etcd3ClientError):
+    code = GRPCCode.PermissionDenied
+    error = "etcdserver: permission denied"
+
+
+class AuthNotEnabled(FailedPrecondition):
+    error = "etcdserver: authentication is not enabled"
+
+
+class InvalidAuthToken(Etcd3ClientError):
+    code = GRPCCode.Unauthenticated
+    error = "etcdserver: invalid auth token"
+
+
+errStringToClientError = {s.error: s for s in Etcd3ClientError.get_subclasses() if hasattr(s, 'error')}
+errCodeToClientError = {s.code: s for s in Etcd3ClientError.__subclasses__()}
+
+
+def _raise_for_data(data, status_code=None):
+    try:
+        error = data.get('error') or data.get('Error')
+        if isinstance(error, dict):  # streaming response
+            status_code = error.get('http_code')
+            code = error['grpc_code']
+            error = error['message']
+        else:
+            code = data.get('code') or data.get('Code')
+    except Exception:
+        error = str(data)
+        code = GRPCCode.Unknown
+    err = errStringToClientError.get(error) or errCodeToClientError.get(code) or Unknown
+    raise err(code, error, status_code)
+
+
+def to_bytes(v):
+    return v if isinstance(v, bytes) else v.encode('utf-8')
+
+
+def prefix_range_end(v):
+    v = bytearray(to_bytes(v))
+    for i in range(len(v) - 1, -1, -1):
+        if v[i] < 0xff:
+            v[i] += 1
+            break
+    return bytes(v)
+
+
+def base64_encode(v):
+    return base64.b64encode(to_bytes(v)).decode('utf-8')
+
+
+def base64_decode(v):
+    return base64.b64decode(v).decode('utf-8')
+
+
+def build_range_request(key, range_end=None):
+    fields = {'key': base64_encode(key)}
+    if range_end:
+        fields['range_end'] = base64_encode(range_end)
+    return fields
+
+
+class Etcd3Client(AbstractEtcdClientWithFailover):
+
+    ERROR_CLS = Etcd3Error
+
+    def __init__(self, config, dns_resolver, cache_ttl=300):
+        self._token = None
+        self._cluster_version = None
+        self.version_prefix = '/v3beta'
+        super(Etcd3Client, self).__init__(config, dns_resolver, cache_ttl)
+
+        if six.PY2:  # pragma: no cover
+            # Old grpc-gateway sometimes sends double 'transfer-encoding: chunked' headers,
+            # what breaks the old (python2.7) httplib.HTTPConnection (it closes the socket).
+            def dedup_addheader(httpm, key, value):
+                prev = httpm.dict.get(key)
+                if prev is None:
+                    httpm.dict[key] = value
+                elif key != 'transfer-encoding' or prev != value:
+                    combined = ", ".join((prev, value))
+                    httpm.dict[key] = combined
+
+            import httplib
+            httplib.HTTPMessage.addheader = dedup_addheader
+
+        try:
+            self.authenticate()
+        except Exception as e:
+            logger.fatal('Etcd3 authentication failed: %r', e)
+            sys.exit(1)
+
+    def _get_headers(self):
+        headers = urllib3.make_headers(user_agent=USER_AGENT)
+        if self._token and self._cluster_version >= (3, 3, 0):
+            headers['authorization'] = self._token
+        return headers
+
+    def _prepare_request(self, kwargs, params=None, method=None):
+        if params is not None:
+            kwargs['body'] = json.dumps(params)
+            kwargs['headers']['Content-Type'] = 'application/json'
+        return self.http.urlopen
+
+    @staticmethod
+    def _handle_server_response(response):
+        data = response.data
+        try:
+            data = data.decode('utf-8')
+            data = json.loads(data)
+        except (TypeError, ValueError, UnicodeError) as e:
+            if response.status < 400:
+                raise etcd.EtcdException('Server response was not valid JSON: %r' % e)
+        if response.status < 400:
+            return data
+        _raise_for_data(data, response.status)
+
+    def _ensure_version_prefix(self, base_uri, **kwargs):
+        if self.version_prefix != '/v3':
+            response = self.http.urlopen(self._MGET, base_uri + '/version', **kwargs)
+            response = self._handle_server_response(response)
+
+            server_version_str = response['etcdserver']
+            server_version = tuple(int(x) for x in server_version_str.split('.'))
+            cluster_version_str = response['etcdcluster']
+            self._cluster_version = tuple(int(x) for x in cluster_version_str.split('.'))
+
+            if self._cluster_version < (3, 0) or server_version < (3, 0, 4):
+                raise UnsupportedEtcdVersion('Detected Etcd version {0} is lower than 3.0.4'.format(server_version_str))
+
+            if self._cluster_version < (3, 3):
+                if self.version_prefix != '/v3alpha':
+                    if self._cluster_version < (3, 1):
+                        logger.warning('Detected Etcd version %s is lower than 3.1.0, watches are not supported',
+                                       cluster_version_str)
+                    if self.username and self.password:
+                        logger.warning('Detected Etcd version %s is lower than 3.3.0, authentication is not supported',
+                                       cluster_version_str)
+                    self.version_prefix = '/v3alpha'
+            elif self._cluster_version < (3, 4):
+                self.version_prefix = '/v3beta'
+            else:
+                self.version_prefix = '/v3'
+
+    def _prepare_get_members(self, etcd_nodes):
+        kwargs = self._prepare_common_parameters(etcd_nodes)
+        self._prepare_request(kwargs, {})
+        return kwargs
+
+    def _get_members(self, base_uri, **kwargs):
+        self._ensure_version_prefix(base_uri, **kwargs)
+        resp = self.http.urlopen(self._MPOST, base_uri + self.version_prefix + '/cluster/member/list', **kwargs)
+        members = self._handle_server_response(resp)['members']
+        return set(url for member in members for url in member.get('clientURLs', []))
+
+    def call_rpc(self, method, fields, retry=None):
+        fields['retry'] = retry
+        return self.api_execute(self.version_prefix + method, self._MPOST, fields)
+
+    def authenticate(self):
+        if self._cluster_version >= (3, 3) and self.username and self.password:
+            logger.info('Trying to authenticate on Etcd...')
+            old_token, self._token = self._token, None
+            try:
+                response = self.call_rpc('/auth/authenticate', {'name': self.username, 'password': self.password})
+            except AuthNotEnabled:
+                logger.info('Etcd authentication is not enabled')
+                self._token = None
+            except Exception:
+                self._token = old_token
+                raise
+            else:
+                self._token = response.get('token')
+            return old_token != self._token
+
+    def _handle_auth_errors(func):
+        def wrapper(self, *args, **kwargs):
+            def retry(ex):
+                if self.username and self.password:
+                    self.authenticate()
+                    return func(self, *args, **kwargs)
+                else:
+                    logger.fatal('Username or password not set, authentication is not possible')
+                    raise ex
+
+            try:
+                return func(self, *args, **kwargs)
+            except (UserEmpty, PermissionDenied) as e:  # no token provided
+                # PermissionDenied is raised on 3.0 and 3.1
+                if self._cluster_version < (3, 3) and (not isinstance(e, PermissionDenied)
+                                                       or self._cluster_version < (3, 2)):
+                    raise UnsupportedEtcdVersion('Authentication is required by Etcd cluster but not '
+                                                 'supported on version lower than 3.3.0. Cluster version: '
+                                                 '{0}'.format('.'.join(map(str, self._cluster_version))))
+                return retry(e)
+            except InvalidAuthToken as e:
+                logger.error('Invalid auth token: %s', self._token)
+                return retry(e)
+
+        return wrapper
+
+    @_handle_auth_errors
+    def range(self, key, range_end=None, retry=None):
+        params = build_range_request(key, range_end)
+        params['serializable'] = True  # For better performance. We can tolerate stale reads.
+        return self.call_rpc('/kv/range', params, retry)
+
+    def prefix(self, key, retry=None):
+        return self.range(key, prefix_range_end(key), retry)
+
+    def lease_grant(self, ttl, retry=None):
+        return self.call_rpc('/lease/grant', {'TTL': ttl}, retry)['ID']
+
+    def lease_keepalive(self, ID, retry=None):
+        return self.call_rpc('/lease/keepalive', {'ID': ID}, retry).get('result', {}).get('TTL')
+
+    def txn(self, compare, success, retry=None):
+        return self.call_rpc('/kv/txn', {'compare': [compare], 'success': [success]}, retry).get('succeeded')
+
+    @_handle_auth_errors
+    def put(self, key, value, lease=None, create_revision=None, mod_revision=None, retry=None):
+        fields = {'key': base64_encode(key), 'value': base64_encode(value)}
+        if lease:
+            fields['lease'] = lease
+        if create_revision is not None:
+            compare = {'target': 'CREATE', 'create_revision': create_revision}
+        elif mod_revision is not None:
+            compare = {'target': 'MOD', 'mod_revision': mod_revision}
+        else:
+            return self.call_rpc('/kv/put', fields, retry)
+        compare['key'] = fields['key']
+        return self.txn(compare, {'request_put': fields}, retry)
+
+    @_handle_auth_errors
+    def deleterange(self, key, range_end=None, mod_revision=None, retry=None):
+        fields = build_range_request(key, range_end)
+        if mod_revision is None:
+            return self.call_rpc('/kv/deleterange', fields, retry)
+        compare = {'target': 'MOD', 'mod_revision': mod_revision, 'key': fields['key']}
+        return self.txn(compare, {'request_delete_range': fields}, retry)
+
+    def deleteprefix(self, key, retry=None):
+        return self.deleterange(key, prefix_range_end(key), retry=retry)
+
+    def watchrange(self, key, range_end=None, start_revision=None, filters=None):
+        """returns: response object"""
+        params = build_range_request(key, range_end)
+        if start_revision is not None:
+            params['start_revision'] = start_revision
+        params['filters'] = filters or []
+        kwargs = self._prepare_common_parameters(1, self.read_timeout)
+        request_executor = self._prepare_request(kwargs, {'create_request': params})
+        kwargs.update(timeout=urllib3.Timeout(connect=kwargs['timeout']), retries=0)
+        return request_executor(self._MPOST, self._base_uri + self.version_prefix + '/watch', **kwargs)
+
+    def watchprefix(self, key, start_revision=None, filters=None):
+        return self.watchrange(key, prefix_range_end(key), start_revision, filters)
+
+
+class KVCache(Thread):
+
+    def __init__(self, dcs, client):
+        Thread.__init__(self)
+        self.daemon = True
+        self._dcs = dcs
+        self._client = client
+        self.condition = Condition()
+        self._config_key = base64_encode(dcs.config_path)
+        self._leader_key = base64_encode(dcs.leader_path)
+        self._optime_key = base64_encode(dcs.leader_optime_path)
+        self._name = base64_encode(dcs._name)
+        self._is_ready = False
+        self._response = None
+        self._response_lock = Lock()
+        self._object_cache = {}
+        self._object_cache_lock = Lock()
+        self.start()
+
+    def set(self, value, overwrite=False):
+        with self._object_cache_lock:
+            name = value['key']
+            old_value = self._object_cache.get(name)
+            ret = not old_value or int(old_value['mod_revision']) < int(value['mod_revision'])
+            if ret or overwrite and old_value['mod_revision'] == value['mod_revision']:
+                self._object_cache[name] = value
+        return ret, old_value
+
+    def delete(self, name, mod_revision):
+        with self._object_cache_lock:
+            old_value = self._object_cache.get(name)
+            ret = old_value and int(old_value['mod_revision']) < int(mod_revision)
+            if ret:
+                del self._object_cache[name]
+        return not old_value or ret, old_value
+
+    def copy(self):
+        with self._object_cache_lock:
+            return [v.copy() for v in self._object_cache.values()]
+
+    def get(self, name):
+        with self._object_cache_lock:
+            return self._object_cache.get(name)
+
+    def _process_event(self, event):
+        kv = event['kv']
+        key = kv['key']
+        if event.get('type') == 'DELETE':
+            success, old_value = self.delete(key, kv['mod_revision'])
+        else:
+            success, old_value = self.set(kv, True)
+
+        if success:
+            old_value = old_value and old_value.get('value')
+            new_value = kv.get('value')
+
+            value_changed = old_value != new_value and \
+                (key == self._leader_key or key == self._optime_key and new_value is not None or
+                 key == self._config_key and old_value is not None and new_value is not None)
+
+            if value_changed:
+                logger.debug('%s changed from %s to %s', key, old_value, new_value)
+
+            # We also want to wake up HA loop on replicas if leader optime was updated
+            if value_changed and (key != self._optime_key or self.get(self._leader_key) != self._name):
+                self._dcs.event.set()
+
+    def _process_message(self, message):
+        logger.debug('Received message: %s', message)
+        if 'error' in message:
+            _raise_for_data(message)
+        for event in message.get('result', {}).get('events', []):
+            self._process_event(event)
+
+    @staticmethod
+    def _finish_response(response):
+        try:
+            response.close()
+        finally:
+            response.release_conn()
+
+    def _do_watch(self, revision):
+        with self._response_lock:
+            self._response = None
+        response = self._client.watchprefix(self._dcs.cluster_prefix, revision)
+        with self._response_lock:
+            if self._response is None:
+                self._response = response
+
+        if not self._response:
+            return self._finish_response(response)
+
+        for message in iter_response_objects(response):
+            self._process_message(message)
+
+    def _build_cache(self):
+        result = self._dcs.retry(self._client.prefix, self._dcs.cluster_prefix)
+        with self._object_cache_lock:
+            self._object_cache = {node['key']: node for node in result.get('kvs', [])}
+        with self.condition:
+            self._is_ready = True
+            self.condition.notify()
+
+        try:
+            self._do_watch(result['header']['revision'])
+        except Exception as e:
+            logger.error('watchprefix failed: %r', e)
+        finally:
+            with self.condition:
+                self._is_ready = False
+            with self._response_lock:
+                response, self._response = self._response, None
+            if response:
+                self._finish_response(response)
+
+    def run(self):
+        while True:
+            try:
+                self._build_cache()
+            except Exception as e:
+                logger.error('KVCache.run %r', e)
+                time.sleep(1)
+
+    def kill_stream(self):
+        sock = None
+        with self._response_lock:
+            if self._response:
+                try:
+                    sock = self._response.connection.sock
+                except Exception:
+                    sock = None
+            else:
+                self._response = False
+        if sock:
+            try:
+                sock.shutdown(socket.SHUT_RDWR)
+                sock.close()
+            except Exception as e:
+                logger.debug('Error on socket.shutdown: %r', e)
+
+    def is_ready(self):
+        """Must be called only when holding the lock on `condition`"""
+        return self._is_ready
+
+
+class PatroniEtcd3Client(Etcd3Client):
+
+    def __init__(self, *args, **kwargs):
+        self._kv_cache = None
+        super(PatroniEtcd3Client, self).__init__(*args, **kwargs)
+
+    def configure(self, etcd3):
+        self._etcd3 = etcd3
+
+    def start_watcher(self):
+        if self._cluster_version >= (3, 1):
+            self._kv_cache = KVCache(self._etcd3, self)
+
+    def _restart_watcher(self):
+        if self._kv_cache:
+            self._kv_cache.kill_stream()
+
+    def set_base_uri(self, value):
+        super(PatroniEtcd3Client, self).set_base_uri(value)
+        self._restart_watcher()
+
+    def authenticate(self):
+        ret = super(PatroniEtcd3Client, self).authenticate()
+        if ret:
+            self._restart_watcher()
+        return ret
+
+    def _wait_cache(self, timeout):
+        stop_time = time.time() + timeout
+        while not self._kv_cache.is_ready():
+            timeout = stop_time - time.time()
+            if timeout <= 0:
+                raise RetryFailedError('Exceeded retry deadline')
+            self._kv_cache.condition.wait(timeout)
+
+    def get_cluster(self):
+        if self._kv_cache:
+            with self._kv_cache.condition:
+                self._wait_cache(self._etcd3._retry.deadline)
+                return self._kv_cache.copy()
+        else:
+            return self._etcd3.retry(self.prefix, self._etcd3.cluster_prefix).get('kvs', [])
+
+    def call_rpc(self, method, fields, retry=None):
+        ret = super(PatroniEtcd3Client, self).call_rpc(method, fields, retry)
+
+        if self._kv_cache:
+            value = delete = None
+            if method == '/kv/txn' and ret.get('succeeded'):
+                on_success = fields['success'][0]
+                value = on_success.get('request_put')
+                delete = on_success.get('request_delete_range')
+            elif method == '/kv/put' and ret:
+                value = fields
+            elif method == '/kv/deleterange' and ret:
+                delete = fields
+
+            if value:
+                value['mod_revision'] = ret['header']['revision']
+                self._kv_cache.set(value)
+            elif delete and 'range_end' not in delete:
+                self._kv_cache.delete(delete['key'], ret['header']['revision'])
+
+        return ret
+
+
+class Etcd3(AbstractEtcd):
+
+    def __init__(self, config):
+        super(Etcd3, self).__init__(config, PatroniEtcd3Client, (DeadlineExceeded, Unavailable, FailedPrecondition))
+        self.__do_not_watch = False
+        self._lease = None
+        self._last_lease_refresh = 0
+
+        self._client.configure(self)
+        if not self._ctl:
+            self._client.start_watcher()
+            self.create_lease()
+
+    def set_socket_options(self, sock, socket_options):
+        enable_keepalive(sock, self.ttl, int(self.loop_wait + self._retry.deadline))
+
+    def set_ttl(self, ttl):
+        self.__do_not_watch = super(Etcd3, self).set_ttl(ttl)
+        if self.__do_not_watch:
+            self._lease = None
+
+    def _do_refresh_lease(self, retry=None):
+        if self._lease and self._last_lease_refresh + self._loop_wait > time.time():
+            return False
+
+        if self._lease and not self._client.lease_keepalive(self._lease, retry):
+            self._lease = None
+
+        ret = not self._lease
+        if ret:
+            self._lease = self._client.lease_grant(self._ttl, retry)
+
+        self._last_lease_refresh = time.time()
+        return ret
+
+    def refresh_lease(self):
+        try:
+            return self.retry(self._do_refresh_lease)
+        except (Etcd3ClientError, RetryFailedError):
+            logger.exception('refresh_lease')
+        raise Etcd3Error('Failed ro keepalive/grant lease')
+
+    def create_lease(self):
+        while not self._lease:
+            try:
+                self.refresh_lease()
+            except Etcd3Error:
+                logger.info('waiting on etcd')
+                time.sleep(5)
+
+    @property
+    def cluster_prefix(self):
+        return self.client_path('')
+
+    @staticmethod
+    def member(node):
+        return Member.from_node(node['mod_revision'], os.path.basename(node['key']), node['lease'], node['value'])
+
+    def _load_cluster(self):
+        cluster = None
+        try:
+            path_len = len(self.cluster_prefix)
+
+            nodes = {}
+            for node in self._client.get_cluster():
+                node['key'] = base64_decode(node['key'])
+                node['value'] = base64_decode(node.get('value', ''))
+                node['lease'] = node.get('lease')
+                nodes[node['key'][path_len:].lstrip('/')] = node
+
+            # get initialize flag
+            initialize = nodes.get(self._INITIALIZE)
+            initialize = initialize and initialize['value']
+
+            # get global dynamic configuration
+            config = nodes.get(self._CONFIG)
+            config = config and ClusterConfig.from_node(config['mod_revision'], config['value'])
+
+            # get timeline history
+            history = nodes.get(self._HISTORY)
+            history = history and TimelineHistory.from_node(history['mod_revision'], history['value'])
+
+            # get last leader operation
+            last_leader_operation = nodes.get(self._LEADER_OPTIME)
+            last_leader_operation = 0 if last_leader_operation is None else int(last_leader_operation['value'])
+
+            # get list of members
+            members = [self.member(n) for k, n in nodes.items() if k.startswith(self._MEMBERS) and k.count('/') == 1]
+
+            # get leader
+            leader = nodes.get(self._LEADER)
+            if not self._ctl and leader and leader['value'] == self._name and self._lease != leader.get('lease'):
+                logger.warning('I am the leader but not owner of the lease')
+
+            if leader:
+                member = Member(-1, leader['value'], None, {})
+                member = ([m for m in members if m.name == leader['value']] or [member])[0]
+                leader = Leader(leader['mod_revision'], leader['lease'], member)
+
+            # failover key
+            failover = nodes.get(self._FAILOVER)
+            if failover:
+                failover = Failover.from_node(failover['mod_revision'], failover['value'])
+
+            # get synchronization state
+            sync = nodes.get(self._SYNC)
+            sync = SyncState.from_node(sync and sync['mod_revision'], sync and sync['value'])
+
+            cluster = Cluster(initialize, config, leader, last_leader_operation, members, failover, sync, history)
+        except UnsupportedEtcdVersion:
+            raise
+        except Exception as e:
+            self._handle_exception(e, 'get_cluster', raise_ex=Etcd3Error('Etcd is not responding properly'))
+        self._has_failed = False
+        return cluster
+
+    @catch_etcd_errors
+    def touch_member(self, data, permanent=False):
+        if not permanent:
+            try:
+                self.refresh_lease()
+            except Etcd3Error:
+                return False
+
+        cluster = self.cluster
+        member = cluster and cluster.get_member(self._name, fallback_to_leader=False)
+
+        if member and member.session == self._lease and deep_compare(data, member.data):
+            return True
+
+        data = json.dumps(data, separators=(',', ':'))
+        try:
+            return self._client.put(self.member_path, data, None if permanent else self._lease)
+        except LeaseNotFound:
+            self._lease = None
+            logger.error('Our lease disappeared from Etcd, can not "touch_member"')
+
+    @catch_etcd_errors
+    def take_leader(self):
+        return self.retry(self._client.put, self.leader_path, self._name, self._lease)
+
+    @catch_etcd_errors
+    def _do_attempt_to_acquire_leader(self, permanent):
+        try:
+            return self.retry(self._client.put, self.leader_path, self._name, None if permanent else self._lease, 0)
+        except LeaseNotFound:
+            self._lease = None
+            logger.error('Our lease disappeared from Etcd. Will try to get a new one and retry attempt')
+            self.refresh_lease()
+            return self.retry(self._client.put, self.leader_path, self._name, None if permanent else self._lease, 0)
+
+    def attempt_to_acquire_leader(self, permanent=False):
+        if not self._lease and not permanent:
+            self.refresh_lease()
+
+        ret = self._do_attempt_to_acquire_leader(permanent)
+        if not ret:
+            logger.info('Could not take out TTL lock')
+        return ret
+
+    @catch_etcd_errors
+    def set_failover_value(self, value, index=None):
+        return self._client.put(self.failover_path, value, mod_revision=index)
+
+    @catch_etcd_errors
+    def set_config_value(self, value, index=None):
+        return self._client.put(self.config_path, value, mod_revision=index)
+
+    @catch_etcd_errors
+    def _write_leader_optime(self, last_operation):
+        return self._client.put(self.leader_optime_path, last_operation)
+
+    @catch_etcd_errors
+    def _update_leader(self):
+        if not self._lease:
+            self.refresh_lease()
+        elif self.retry(self._client.lease_keepalive, self._lease):
+            self._last_lease_refresh = time.time()
+
+        if self._lease:
+            cluster = self.cluster
+            leader_lease = cluster and isinstance(cluster.leader, Leader) and cluster.leader.session
+            if leader_lease != self._lease:
+                self.take_leader()
+        return bool(self._lease)
+
+    @catch_etcd_errors
+    def initialize(self, create_new=True, sysid=""):
+        return self.retry(self._client.put, self.initialize_path, sysid, None, 0 if create_new else None)
+
+    @catch_etcd_errors
+    def _delete_leader(self):
+        cluster = self.cluster
+        if cluster and isinstance(cluster.leader, Leader) and cluster.leader.name == self._name:
+            return self._client.deleterange(self.leader_path, mod_revision=cluster.leader.index)
+
+    @catch_etcd_errors
+    def cancel_initialization(self):
+        return self.retry(self._client.deleterange, self.initialize_path)
+
+    @catch_etcd_errors
+    def delete_cluster(self):
+        return self.retry(self._client.deleteprefix, self.cluster_prefix)
+
+    @catch_etcd_errors
+    def set_history_value(self, value):
+        return self._client.put(self.history_path, value)
+
+    @catch_etcd_errors
+    def set_sync_state_value(self, value, index=None):
+        return self.retry(self._client.put, self.sync_path, value, mod_revision=index)
+
+    @catch_etcd_errors
+    def delete_sync_state(self, index=None):
+        return self.retry(self._client.deleterange, self.sync_path, mod_revision=index)
+
+    def watch(self, leader_index, timeout):
+        if self.__do_not_watch:
+            self.__do_not_watch = False
+            return True
+
+        try:
+            return super(Etcd3, self).watch(None, timeout)
+        finally:
+            self.event.clear()

setup.py

@@ -22,7 +22,7 @@ AUTHOR_EMAIL = '[email protected], [email protected], alexk
 KEYWORDS = 'etcd governor patroni postgresql postgres ha haproxy confd' +\
     ' zookeeper exhibitor consul streaming replication kubernetes k8s'
 
-EXTRAS_REQUIRE = {'aws': ['boto'], 'etcd': ['python-etcd'], 'consul': ['python-consul'],
+EXTRAS_REQUIRE = {'aws': ['boto'], 'etcd': ['python-etcd'], 'etcd3': ['python-etcd'], 'consul': ['python-consul'],
                   'exhibitor': ['kazoo'], 'zookeeper': ['kazoo'], 'kubernetes': [], 'raft': ['pysyncobj']}
 COVERAGE_XML = True
 COVERAGE_HTML = False

tests/__init__.py

@@ -169,6 +169,7 @@ class PostgresInit(unittest.TestCase):
                    'stats_temp_directory': '/tmp'}
 
     @patch('psycopg2.connect', psycopg2_connect)
+    @patch('patroni.postgresql.CallbackExecutor', Mock())
     @patch.object(ConfigHandler, 'write_postgresql_conf', Mock())
     @patch.object(ConfigHandler, 'replace_pg_hba', Mock())
     @patch.object(ConfigHandler, 'replace_pg_ident', Mock())

tests/test_ctl.py

@@ -8,7 +8,7 @@ from mock import patch, Mock
 from patroni.ctl import ctl, store_config, load_config, output_members, get_dcs, parse_dcs, \
     get_all_members, get_any_member, get_cursor, query_member, configure, PatroniCtlException, apply_config_changes, \
     format_config_for_editing, show_diff, invoke_editor, format_pg_version, find_executable, CONFIG_FILE_PATH
-from patroni.dcs.etcd import Client, Failover
+from patroni.dcs.etcd import AbstractEtcdClientWithFailover, Failover
 from patroni.utils import tzutc
 from psycopg2 import OperationalError
 from urllib3 import PoolManager
@@ -37,7 +37,7 @@ class TestCtl(unittest.TestCase):
 
     @patch('socket.getaddrinfo', socket_getaddrinfo)
     def setUp(self):
-        with patch.object(Client, 'machines') as mock_machines:
+        with patch.object(AbstractEtcdClientWithFailover, 'machines') as mock_machines:
             mock_machines.__get__ = Mock(return_value=['http://remotehost:2379'])
             self.runner = CliRunner()
             self.e = get_dcs({'etcd': {'ttl': 30, 'host': 'ok:2379', 'retry_timeout': 10}}, 'foo')

tests/test_etcd.py

@@ -5,7 +5,7 @@ import unittest
 
 from dns.exception import DNSException
 from mock import Mock, patch
-from patroni.dcs.etcd import AbstractDCS, Client, Cluster, Etcd, EtcdError, DnsCachingResolver
+from patroni.dcs.etcd import AbstractDCS, EtcdClient, Cluster, Etcd, EtcdError, DnsCachingResolver
 from patroni.exceptions import DCSError
 from patroni.utils import Retry
 from urllib3.exceptions import ReadTimeoutError
@@ -123,9 +123,9 @@ class TestClient(unittest.TestCase):
     @patch('socket.getaddrinfo', socket_getaddrinfo)
     @patch('patroni.dcs.etcd.requests_get', requests_get)
     def setUp(self):
-        with patch.object(Client, 'machines') as mock_machines:
+        with patch.object(EtcdClient, 'machines') as mock_machines:
             mock_machines.__get__ = Mock(return_value=['http://localhost:2379', 'http://localhost:4001'])
-            self.client = Client({'srv': 'test', 'retry_timeout': 3}, DnsCachingResolver())
+            self.client = EtcdClient({'srv': 'test', 'retry_timeout': 3}, DnsCachingResolver())
             self.client.http.request = http_request
             self.client.http.request_encode_body = http_request
 
@@ -143,10 +143,9 @@ class TestClient(unittest.TestCase):
         except Exception:
             self.assertIsNone(machines)
 
-    @patch.object(Client, 'machines')
+    @patch.object(EtcdClient, 'machines')
     def test_api_execute(self, mock_machines):
         mock_machines.__get__ = Mock(return_value=['http://localhost:4001', 'http://localhost:2379'])
-        self.assertRaises(ValueError, self.client.api_execute, '', '')
         self.client._base_uri = 'http://localhost:4001'
         self.assertRaises(etcd.EtcdException, self.client.api_execute, '/', 'POST', timeout=0)
         self.client._base_uri = 'http://localhost:4001'
@@ -157,18 +156,17 @@ class TestClient(unittest.TestCase):
         self.client._machines_cache = [self.client._base_uri]
         self.assertRaises(etcd.EtcdWatchTimedOut, self.client.api_execute, '/timeout', 'POST', params={'wait': 'true'})
         self.assertRaises(etcd.EtcdWatchTimedOut, self.client.api_execute, '/timeout', 'POST', params={'wait': 'true'})
-        self.assertRaises(etcd.EtcdException, self.client.api_execute, '/', '')
 
-        with patch.object(Client, '_calculate_timeouts', Mock(side_effect=[(1, 1, 0), (1, 1, 0), (0, 1, 0)])),\
-                patch.object(Client, '_load_machines_cache', Mock(side_effect=Exception)):
+        with patch.object(EtcdClient, '_calculate_timeouts', Mock(side_effect=[(1, 1, 0), (1, 1, 0), (0, 1, 0)])),\
+                patch.object(EtcdClient, '_load_machines_cache', Mock(side_effect=Exception)):
             self.client.http.request = Mock(side_effect=socket.error)
             self.assertRaises(etcd.EtcdException, rtry, self.client.api_execute, '/', 'GET', params={'retry': rtry})
 
-        with patch.object(Client, '_calculate_timeouts', Mock(side_effect=[(1, 1, 0), (1, 1, 0), (0, 1, 0)])),\
-                patch.object(Client, '_load_machines_cache', Mock(return_value=True)):
+        with patch.object(EtcdClient, '_calculate_timeouts', Mock(side_effect=[(1, 1, 0), (1, 1, 0), (0, 1, 0)])),\
+                patch.object(EtcdClient, '_load_machines_cache', Mock(return_value=True)):
             self.assertRaises(etcd.EtcdException, rtry, self.client.api_execute, '/', 'GET', params={'retry': rtry})
 
-        with patch.object(Client, '_do_http_request', Mock(side_effect=etcd.EtcdException)):
+        with patch.object(EtcdClient, '_do_http_request', Mock(side_effect=etcd.EtcdException)):
             self.client._read_timeout = 0.01
             self.assertRaises(etcd.EtcdException, self.client.api_execute, '/', 'GET')
 
@@ -184,7 +182,7 @@ class TestClient(unittest.TestCase):
     def test__get_machines_cache_from_dns(self):
         self.client._get_machines_cache_from_dns('error', 2379)
 
-    @patch.object(Client, 'machines')
+    @patch.object(EtcdClient, 'machines')
     def test__refresh_machines_cache(self, mock_machines):
         mock_machines.__get__ = Mock(side_effect=etcd.EtcdConnectionFailed)
         self.assertIsNone(self.client._refresh_machines_cache())
@@ -205,6 +203,10 @@ class TestClient(unittest.TestCase):
                           timeout=1, source_address=('localhost', 53333),
                           socket_options=[(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)])
 
+    def test___del__(self):
+        self.client.http.clear = Mock(side_effect=TypeError)
+        del self.client
+
 
 @patch('patroni.dcs.etcd.requests_get', requests_get)
 @patch('socket.getaddrinfo', socket_getaddrinfo)
@@ -215,7 +217,7 @@ class TestEtcd(unittest.TestCase):
 
     @patch('socket.getaddrinfo', socket_getaddrinfo)
     def setUp(self):
-        with patch.object(Client, 'machines') as mock_machines:
+        with patch.object(EtcdClient, 'machines') as mock_machines:
             mock_machines.__get__ = Mock(return_value=['http://localhost:2379', 'http://localhost:4001'])
             self.etcd = Etcd({'namespace': '/patroni/', 'ttl': 30, 'retry_timeout': 10,
                               'host': 'localhost:2379', 'scope': 'test', 'name': 'foo'})
@@ -226,17 +228,18 @@ class TestEtcd(unittest.TestCase):
     @patch('dns.resolver.query', dns_query)
     def test_get_etcd_client(self):
         with patch('time.sleep', Mock(side_effect=SleepException)),\
-                patch.object(Client, 'machines') as mock_machines:
+                patch.object(EtcdClient, 'machines') as mock_machines:
             mock_machines.__get__ = Mock(side_effect=etcd.EtcdException)
             self.assertRaises(SleepException, self.etcd.get_etcd_client,
-                              {'discovery_srv': 'test', 'retry_timeout': 10, 'cacert': '1', 'key': '1', 'cert': 1})
+                              {'discovery_srv': 'test', 'retry_timeout': 10, 'cacert': '1', 'key': '1', 'cert': 1},
+                              EtcdClient)
             self.assertRaises(SleepException, self.etcd.get_etcd_client,
-                              {'url': 'https://test:2379', 'retry_timeout': 10})
+                              {'url': 'https://test:2379', 'retry_timeout': 10}, EtcdClient)
             self.assertRaises(SleepException, self.etcd.get_etcd_client,
-                              {'hosts': 'foo:4001,bar', 'retry_timeout': 10})
+                              {'hosts': 'foo:4001,bar', 'retry_timeout': 10}, EtcdClient)
             mock_machines.__get__ = Mock(return_value=[])
             self.assertRaises(SleepException, self.etcd.get_etcd_client,
-                              {'proxy': 'https://user:password@test:2379', 'retry_timeout': 10})
+                              {'proxy': 'https://user:password@test:2379', 'retry_timeout': 10}, EtcdClient)
 
     def test_get_cluster(self):
         cluster = self.etcd.get_cluster()

tests/test_etcd3.py

@@ -0,0 +1,246 @@
+import etcd
+import json
+import unittest
+import urllib3
+
+from mock import Mock, patch
+from patroni.dcs.etcd3 import PatroniEtcd3Client, Cluster, Etcd3, Etcd3Error, Etcd3ClientError, RetryFailedError,\
+        InvalidAuthToken, Unavailable, Unknown, UnsupportedEtcdVersion, UserEmpty, base64_encode
+from threading import Thread
+
+from . import SleepException, MockResponse
+
+
+def mock_urlopen(self, method, url, **kwargs):
+    ret = MockResponse()
+    if method == 'GET' and url.endswith('/version'):
+        ret.content = '{"etcdserver": "3.3.13", "etcdcluster": "3.3.0"}'
+    elif method != 'POST':
+        raise Exception('Unexpected request method: {0} {1} {2}'.format(method, url, kwargs))
+    elif url.endswith('/cluster/member/list'):
+        ret.content = '{"members":[{"clientURLs":["http://localhost:2379", "http://localhost:4001"]}]}'
+    elif url.endswith('/auth/authenticate'):
+        ret.content = '{"token":"authtoken"}'
+    elif url.endswith('/lease/grant'):
+        ret.content = '{"ID": "123"}'
+    elif url.endswith('/lease/keepalive'):
+        ret.content = '{"result":{"TTL":30}}'
+    elif url.endswith('/kv/range'):
+        ret.content = json.dumps({
+            "header": {"revision": "1"},
+            "kvs": [
+                {"key": base64_encode('/patroni/test/leader'),
+                 "value": base64_encode('foo'), "lease": "bla", "mod_revision": '1'},
+                {"key": base64_encode('/patroni/test/members/foo'),
+                 "value": base64_encode('{}'), "lease": "123", "mod_revision": '1'},
+                {"key": base64_encode('/patroni/test/failover'), "value": base64_encode('{}'), "mod_revision": '1'}
+            ]
+        })
+    elif url.endswith('/watch'):
+        key = base64_encode('/patroni/test/config')
+        ret.read_chunked = Mock(return_value=[json.dumps({
+            'result': {'events': [
+                {'kv': {'key': key, 'value': base64_encode('bar'), 'mod_revision': '2'}},
+                {'kv': {'key': key, 'value': base64_encode('buzz'), 'mod_revision': '3'}},
+                {'type': 'DELETE', 'kv': {'key': key, 'mod_revision': '4'}},
+                {'kv': {'key': base64_encode('/patroni/test/optime/leader'),
+                        'value': base64_encode('1234567'), 'mod_revision': '5'}},
+            ]}
+        })[:-1].encode('utf-8'), b'}{"error":{"grpc_code":14,"message":"","http_code":503}}'])
+    elif url.endswith('/kv/put') or url.endswith('/kv/txn'):
+        ret.status_code = 400
+        ret.content = '{"code":5,"error":"etcdserver: requested lease not found"}'
+    elif not url.endswith('/kv/deleterange'):
+        raise Exception('Unexpected url: {0} {1} {2}'.format(method, url, kwargs))
+    return ret
+
+
+class BaseTestEtcd3(unittest.TestCase):
+
+    @patch.object(Thread, 'start', Mock())
+    @patch.object(urllib3.PoolManager, 'urlopen', mock_urlopen)
+    def setUp(self):
+        self.etcd3 = Etcd3({'namespace': '/patroni/', 'ttl': 30, 'retry_timeout': 10,
+                            'host': 'localhost:2378', 'scope': 'test', 'name': 'foo',
+                            'username': 'etcduser', 'password': 'etcdpassword'})
+        self.client = self.etcd3._client
+        self.kv_cache = self.client._kv_cache
+
+
+class TestKVCache(BaseTestEtcd3):
+
+    def test__do_watch(self):
+        self.client.watchprefix = Mock(return_value=False)
+        self.assertRaises(AttributeError, self.kv_cache._do_watch, '1')
+
+    @patch('time.sleep', Mock(side_effect=SleepException))
+    @patch('patroni.dcs.etcd3.KVCache._build_cache', Mock(side_effect=Exception))
+    def test_run(self):
+        self.assertRaises(SleepException, self.kv_cache.run)
+
+    @patch.object(urllib3.PoolManager, 'urlopen', mock_urlopen)
+    def test_kill_stream(self):
+        self.assertRaises(Unavailable, self.kv_cache._do_watch, '1')
+        self.kv_cache.kill_stream()
+        with patch.object(MockResponse, 'connection', create=True) as mock_conn:
+            self.kv_cache.kill_stream()
+            mock_conn.sock.close.side_effect = Exception
+            self.kv_cache.kill_stream()
+
+
+class TestPatroniEtcd3Client(BaseTestEtcd3):
+
+    @patch('patroni.dcs.etcd3.Etcd3Client.authenticate', Mock(side_effect=Exception))
+    def test__init__(self):
+        self.assertRaises(SystemExit, self.setUp)
+
+    @patch.object(urllib3.PoolManager, 'urlopen')
+    def test_call_rpc(self, mock_urlopen):
+        request = {'key': base64_encode('/patroni/test/leader')}
+        mock_urlopen.return_value = MockResponse()
+        mock_urlopen.return_value.content = '{"succeeded":true,"header":{"revision":"1"}}'
+        self.client.call_rpc('/kv/txn', {'success': [{'request_delete_range': request}]})
+        self.client.call_rpc('/kv/put', request)
+        self.client.call_rpc('/kv/deleterange', request)
+
+    @patch('time.time', Mock(side_effect=[1, 10.9, 100]))
+    def test__wait_cache(self):
+        with self.kv_cache.condition:
+            self.assertRaises(RetryFailedError, self.client._wait_cache, 10)
+
+    @patch.object(urllib3.PoolManager, 'urlopen')
+    def test__restart_watcher(self, mock_urlopen):
+        mock_urlopen.return_value = MockResponse()
+        mock_urlopen.return_value.status_code = 400
+        mock_urlopen.return_value.content = '{"code":9,"error":"etcdserver: authentication is not enabled"}'
+        self.client.authenticate()
+
+    @patch.object(urllib3.PoolManager, 'urlopen')
+    def test__handle_auth_errors(self, mock_urlopen):
+        mock_urlopen.return_value = MockResponse()
+        mock_urlopen.return_value.content = '{"code":3,"error":"etcdserver: user name is empty"}'
+        mock_urlopen.return_value.status_code = 403
+        self.client._cluster_version = (3, 1, 5)
+        self.assertRaises(UnsupportedEtcdVersion, self.client.deleteprefix, 'foo')
+        self.client._cluster_version = (3, 3, 13)
+        self.assertRaises(UserEmpty, self.client.deleteprefix, 'foo')
+        mock_urlopen.return_value.content = '{"code":16,"error":"etcdserver: invalid auth token"}'
+        self.assertRaises(InvalidAuthToken, self.client.deleteprefix, 'foo')
+        with patch.object(PatroniEtcd3Client, 'authenticate', Mock(return_value=True)):
+            self.assertRaises(InvalidAuthToken, self.client.deleteprefix, 'foo')
+            self.client.username = None
+            self.assertRaises(InvalidAuthToken, self.client.deleteprefix, 'foo')
+
+    def test__handle_server_response(self):
+        response = MockResponse()
+        response.content = '{"code":0,"error":"'
+        self.assertRaises(etcd.EtcdException, self.client._handle_server_response, response)
+        response.status_code = 400
+        self.assertRaises(Unknown, self.client._handle_server_response, response)
+        response.content = '{"error":{"grpc_code":0,"message":"","http_code":400}}'
+        try:
+            self.client._handle_server_response(response)
+        except Unknown as e:
+            self.assertEqual(e.as_dict(), {'code': 2, 'codeText': 'OK', 'error': u'', 'status': 400})
+
+    @patch.object(urllib3.PoolManager, 'urlopen')
+    def test__ensure_version_prefix(self, mock_urlopen):
+        self.client.version_prefix = None
+        mock_urlopen.return_value = MockResponse()
+        mock_urlopen.return_value.content = '{"etcdserver": "3.0.3", "etcdcluster": "3.0.0"}'
+        self.assertRaises(UnsupportedEtcdVersion, self.client._ensure_version_prefix, '')
+        mock_urlopen.return_value.content = '{"etcdserver": "3.0.4", "etcdcluster": "3.0.0"}'
+        self.client._ensure_version_prefix('')
+        self.assertEqual(self.client.version_prefix, '/v3alpha')
+        mock_urlopen.return_value.content = '{"etcdserver": "3.4.4", "etcdcluster": "3.4.0"}'
+        self.client._ensure_version_prefix('')
+        self.assertEqual(self.client.version_prefix, '/v3')
+
+
+@patch.object(urllib3.PoolManager, 'urlopen', mock_urlopen)
+class TestEtcd3(BaseTestEtcd3):
+
+    @patch.object(Thread, 'start', Mock())
+    @patch.object(urllib3.PoolManager, 'urlopen', mock_urlopen)
+    def setUp(self):
+        super(TestEtcd3, self).setUp()
+        self.assertRaises(AttributeError, self.kv_cache._build_cache)
+        self.kv_cache._is_ready = True
+        self.etcd3.get_cluster()
+
+    def test_get_cluster(self):
+        self.assertIsInstance(self.etcd3.get_cluster(), Cluster)
+        self.client._kv_cache = None
+        with patch.object(urllib3.PoolManager, 'urlopen') as mock_urlopen:
+            mock_urlopen.side_effect = UnsupportedEtcdVersion('')
+            self.assertRaises(UnsupportedEtcdVersion, self.etcd3.get_cluster)
+            mock_urlopen.side_effect = SleepException()
+            self.assertRaises(Etcd3Error, self.etcd3.get_cluster)
+
+    def test_touch_member(self):
+        self.etcd3.touch_member({})
+        self.etcd3._lease = 'bla'
+        self.etcd3.touch_member({})
+        with patch.object(PatroniEtcd3Client, 'lease_grant', Mock(side_effect=Etcd3ClientError)):
+            self.etcd3.touch_member({})
+
+    def test__update_leader(self):
+        self.etcd3._lease = None
+        self.etcd3.update_leader('123')
+        self.etcd3.update_leader('124')
+
+    def test_attempt_to_acquire_leader(self):
+        self.etcd3._lease = None
+        self.assertFalse(self.etcd3.attempt_to_acquire_leader())
+
+    def test_set_ttl(self):
+        self.etcd3.set_ttl(20)
+
+    @patch.object(PatroniEtcd3Client, 'lease_keepalive', Mock(return_value=False))
+    def test_refresh_lease(self):
+        self.etcd3._last_lease_refresh = 0
+        self.etcd3.refresh_lease()
+
+    @patch('time.sleep', Mock(side_effect=SleepException))
+    @patch.object(PatroniEtcd3Client, 'lease_keepalive', Mock(return_value=False))
+    @patch.object(PatroniEtcd3Client, 'lease_grant', Mock(side_effect=Etcd3ClientError))
+    def test_create_lease(self):
+        self.etcd3._lease = None
+        self.etcd3._last_lease_refresh = 0
+        self.assertRaises(SleepException, self.etcd3.create_lease)
+
+    def test_set_failover_value(self):
+        self.etcd3.set_failover_value('', 1)
+
+    def test_set_config_value(self):
+        self.etcd3.set_config_value('')
+
+    def test_initialize(self):
+        self.etcd3.initialize()
+
+    def test_cancel_initialization(self):
+        self.etcd3.cancel_initialization()
+
+    def test_delete_leader(self):
+        self.etcd3.delete_leader()
+
+    def test_delete_cluster(self):
+        self.etcd3.delete_cluster()
+
+    def test_set_history_value(self):
+        self.etcd3.set_history_value('')
+
+    def test_set_sync_state_value(self):
+        self.etcd3.set_sync_state_value('')
+
+    def test_delete_sync_state(self):
+        self.etcd3.delete_sync_state()
+
+    def test_watch(self):
+        self.etcd3.set_ttl(10)
+        self.etcd3.watch(None, 0)
+        self.etcd3.watch(None, 0)
+
+    def test_set_socket_options(self):
+        with patch('socket.SIO_KEEPALIVE_VALS', 1, create=True):
+            self.etcd3.set_socket_options(Mock(), None)

tests/test_ha.py

@@ -6,7 +6,7 @@ import sys
 from mock import Mock, MagicMock, PropertyMock, patch, mock_open
 from patroni.config import Config
 from patroni.dcs import Cluster, ClusterConfig, Failover, Leader, Member, get_dcs, SyncState, TimelineHistory
-from patroni.dcs.etcd import Client
+from patroni.dcs.etcd import AbstractEtcdClientWithFailover
 from patroni.exceptions import DCSError, PostgresConnectionException, PatroniFatalException
 from patroni.ha import Ha, _MemberStatus
 from patroni.postgresql import Postgresql
@@ -185,7 +185,7 @@ class TestHa(PostgresInit):
     @patch.object(etcd.Client, 'read', etcd_read)
     def setUp(self):
         super(TestHa, self).setUp()
-        with patch.object(Client, 'machines') as mock_machines:
+        with patch.object(AbstractEtcdClientWithFailover, 'machines') as mock_machines:
             mock_machines.__get__ = Mock(return_value=['http://remotehost:2379'])
             self.p.set_state('running')
             self.p.set_role('replica')

tests/test_patroni.py

@@ -9,7 +9,7 @@ import patroni.config as config
 from mock import Mock, PropertyMock, patch
 from patroni.api import RestApiServer
 from patroni.async_executor import AsyncExecutor
-from patroni.dcs.etcd import Client
+from patroni.dcs.etcd import AbstractEtcdClientWithFailover
 from patroni.exceptions import DCSError
 from patroni.postgresql import Postgresql
 from patroni.postgresql.config import ConfigHandler
@@ -53,7 +53,7 @@ class TestPatroni(unittest.TestCase):
     @patch.object(BaseHTTPServer.HTTPServer, '__init__', Mock())
     @patch.object(etcd.Client, 'read', etcd_read)
     @patch.object(Thread, 'start', Mock())
-    @patch.object(Client, 'machines', PropertyMock(return_value=['http://remotehost:2379']))
+    @patch.object(AbstractEtcdClientWithFailover, 'machines', PropertyMock(return_value=['http://remotehost:2379']))
     def setUp(self):
         self._handlers = logging.getLogger().handlers[:]
         RestApiServer._BaseServer__is_shut_down = Mock()
@@ -75,7 +75,7 @@ class TestPatroni(unittest.TestCase):
     @patch('sys.argv', ['patroni.py', 'postgres0.yml'])
     @patch('time.sleep', Mock(side_effect=SleepException))
     @patch.object(etcd.Client, 'delete', Mock())
-    @patch.object(Client, 'machines', PropertyMock(return_value=['http://remotehost:2379']))
+    @patch.object(AbstractEtcdClientWithFailover, 'machines', PropertyMock(return_value=['http://remotehost:2379']))
     @patch.object(Thread, 'join', Mock())
     def test_patroni_patroni_main(self):
         with patch('subprocess.call', Mock(return_value=1)):

tests/test_postgresql.py

@@ -98,7 +98,6 @@ class TestPostgresql(BaseTestPostgresql):
     def setUp(self):
         super(TestPostgresql, self).setUp()
         self.p.config.write_postgresql_conf()
-        self.p._callback_executor = Mock()
 
     @patch('subprocess.Popen')
     @patch.object(Postgresql, 'wait_for_startup')