Repository: zalando/patroni
Pull request #1917 (2021-04-23 06:10:08)

The /leader endpoint returns 200 if node holds the lock
Promoting the standby cluster requires updating load-balancer health checks, which is not very convenient and easy to forget.
In order to solve it, we change the behavior of the `/leader` health-check endpoint. It will return 200 without taking into account whether PostgreSQL is running as the primary or the standby_leader.


Patch:
diff --git a/.github/workflows/install_deps.py b/.github/workflows/install_deps.py
index 1a4477ea..fb5194cc 100644
--- a/.github/workflows/install_deps.py
+++ b/.github/workflows/install_deps.py
@@ -22,7 +22,7 @@ def install_requirements(what):
     for r in read('requirements.txt').split('\n'):
         r = r.strip()
         if r != '':
-            extras = {e for e, v in EXTRAS_REQUIRE.items() if v and r.startswith(v[0])}
+            extras = {e for e, v in EXTRAS_REQUIRE.items() if v and any(r.startswith(x) for x in v)}
             if not extras or what == 'all' or what in extras:
                 requirements.append(r)
 
diff --git a/docs/rest_api.rst b/docs/rest_api.rst
index 74003540..2bd2147f 100644
--- a/docs/rest_api.rst
+++ b/docs/rest_api.rst
@@ -9,14 +9,17 @@ Health check endpoints
 ----------------------
 For all health check ``GET`` requests Patroni returns a JSON document with the status of the node, along with the HTTP status code. If you don't want or don't need the JSON document, you might consider using the ``OPTIONS`` method instead of ``GET``.
 
-- The following requests to Patroni REST API will return HTTP status code **200** only when the Patroni node is running as the leader:
+- The following requests to Patroni REST API will return HTTP status code **200** only when the Patroni node is running as the primary with leader lock:
 
   - ``GET /``
   - ``GET /master``
-  - ``GET /leader``
   - ``GET /primary``
   - ``GET /read-write``
 
+- ``GET /standby-leader``: returns HTTP status code **200** only when the Patroni node is running as the leader in a :ref:`standby cluster <standby_cluster>`.
+
+- ``GET /leader``: returns HTTP status code **200** when the Patroni node has the leader lock. The major difference from the two previous endpoints is that it doesn't take into account whether PostgreSQL is running as the ``primary`` or the ``standby_leader``.
+
 - ``GET /replica``: replica health check endpoint. It returns HTTP status code **200** only when the Patroni node is in the state ``running``, the role is ``replica`` and ``noloadbalance`` tag is not set.
 
 - ``GET /replica?lag=<max-lag>``: replica check endpoint. In addition to checks from ``replica``, it also checks replication latency and returns status code **200** only when it is below specified value. The key cluster.last_leader_operation from DCS is used for Leader wal position and compute latency on replica for performance reasons. max-lag can be specified in bytes (integer) or in human readable values, for e.g. 16kB, 64MB, 1GB.
@@ -28,8 +31,6 @@ For all health check ``GET`` requests Patroni returns a JSON document with the s
 
 - ``GET /read-only``: like the above endpoint, but also includes the primary.
 
-- ``GET /standby-leader``: returns HTTP status code **200** only when the Patroni node is running as the leader in a :ref:`standby cluster <standby_cluster>`.
-
 - ``GET /synchronous`` or ``GET /sync``: returns HTTP status code **200** only when the Patroni node is running as a synchronous standby.
 
 - ``GET /asynchronous`` or ``GET /async``: returns HTTP status code **200** only when the Patroni node is running as an asynchronous standby.
diff --git a/features/environment.py b/features/environment.py
index d457ae12..130a5bf5 100644
--- a/features/environment.py
+++ b/features/environment.py
@@ -591,10 +591,12 @@ class ExhibitorController(ZooKeeperController):
 class RaftController(AbstractDcsController):
 
     CONTROLLER_ADDR = 'localhost:1234'
+    PASSWORD = '12345'
 
     def __init__(self, context):
         super(RaftController, self).__init__(context)
-        os.environ.update(PATRONI_RAFT_PARTNER_ADDRS="'" + self.CONTROLLER_ADDR + "'", RAFT_PORT='1234')
+        os.environ.update(PATRONI_RAFT_PARTNER_ADDRS="'" + self.CONTROLLER_ADDR + "'",
+                          PATRONI_RAFT_PASSWORD=self.PASSWORD, RAFT_PORT='1234')
         self._raft = None
 
     def _start(self):
@@ -615,18 +617,16 @@ class RaftController(AbstractDcsController):
 
     def cleanup_service_tree(self):
         from patroni.dcs.raft import KVStoreTTL
-        from pysyncobj import SyncObjConf
 
         if self._raft:
             self._raft.destroy()
-            self._raft._SyncObj__thread.join()
             self.stop()
             os.makedirs(self._work_directory)
             self.start()
 
         ready_event = threading.Event()
-        conf = SyncObjConf(appendEntriesUseBatch=False, dynamicMembershipChange=True, onReady=ready_event.set)
-        self._raft = KVStoreTTL(None, [self.CONTROLLER_ADDR], conf)
+        self._raft = KVStoreTTL(ready_event.set, None, None, partner_addrs=[self.CONTROLLER_ADDR], password=self.PASSWORD)
+        self._raft.startAutoTick()
         ready_event.wait()
 
 
diff --git a/patroni/api.py b/patroni/api.py
index 4fbc19e5..df06ec8e 100644
--- a/patroni/api.py
+++ b/patroni/api.py
@@ -108,9 +108,11 @@ class RestApiHandler(BaseHTTPRequestHandler):
             response.get('role') == 'replica' and response.get('state') == 'running' else 503
 
         if not cluster and patroni.ha.is_paused():
+            leader_status_code = 200 if response.get('role') in ('master', 'standby_leader') else 503
             primary_status_code = 200 if response.get('role') == 'master' else 503
             standby_leader_status_code = 200 if response.get('role') == 'standby_leader' else 503
         elif patroni.ha.is_leader():
+            leader_status_code = 200
             if patroni.ha.is_standby_cluster():
                 primary_status_code = replica_status_code = 503
                 standby_leader_status_code = 200 if response.get('role') in ('replica', 'standby_leader') else 503
@@ -118,13 +120,15 @@ class RestApiHandler(BaseHTTPRequestHandler):
                 primary_status_code = 200
                 standby_leader_status_code = 503
         else:
-            primary_status_code = standby_leader_status_code = 503
+            leader_status_code = primary_status_code = standby_leader_status_code = 503
 
         status_code = 503
 
         if 'standby_leader' in path or 'standby-leader' in path:
             status_code = standby_leader_status_code
-        elif 'master' in path or 'leader' in path or 'primary' in path or 'read-write' in path:
+        elif 'leader' in path:
+            status_code = leader_status_code
+        elif 'master' in path or 'primary' in path or 'read-write' in path:
             status_code = primary_status_code
         elif 'replica' in path:
             status_code = replica_status_code
diff --git a/patroni/dcs/raft.py b/patroni/dcs/raft.py
index f94e6e64..76c65efb 100644
--- a/patroni/dcs/raft.py
+++ b/patroni/dcs/raft.py
@@ -5,7 +5,8 @@ import threading
 import time
 
 from pysyncobj import SyncObj, SyncObjConf, replicated, FAIL_REASON
-from pysyncobj.transport import Node, TCPTransport, CONNECTION_STATE
+from pysyncobj.transport import TCPTransport, CONNECTION_STATE
+from pysyncobj.utility import TcpUtility, UtilityException
 
 from . import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, SyncState, TimelineHistory
 from ..utils import validate_directory
@@ -13,122 +14,97 @@ from ..utils import validate_directory
 logger = logging.getLogger(__name__)
 
 
-class MessageNode(Node):
-
-    def __init__(self, address):
-        self.address = address
-
-
-class UtilityTransport(TCPTransport):
+class _TCPTransport(TCPTransport):
 
     def __init__(self, syncObj, selfNode, otherNodes):
-        super(UtilityTransport, self).__init__(syncObj, selfNode, otherNodes)
-        self._selfIsReadonlyNode = False
-
-    def _connectIfNecessarySingle(self, node):
-        pass
-
-    def connectionState(self, node):
-        return self._connections[node].state
+        super(_TCPTransport, self).__init__(syncObj, selfNode, otherNodes)
+        self.setOnUtilityMessageCallback('members', syncObj.getMembers)
 
-    def isDisconnected(self, node):
-        return self.connectionState(node) == CONNECTION_STATE.DISCONNECTED
 
-    def connectIfRequiredSingle(self, node):
-        if self.isDisconnected(node):
-            return self._connections[node].connect(node.ip, node.port)
-
-    def disconnectSingle(self, node):
-        self._connections[node].disconnect()
-
-
-class SyncObjUtility(SyncObj):
+class SyncObjUtility(object):
 
     def __init__(self, otherNodes, conf):
-        autoTick = conf.autoTick
-        conf.autoTick = False
-        super(SyncObjUtility, self).__init__(None, otherNodes, conf, transportClass=UtilityTransport)
-        conf.autoTick = autoTick
-        self._SyncObj__transport.setOnMessageReceivedCallback(self._onMessageReceived)
-        self.__result = None
-
-    def setPartnerNode(self, partner):
-        self.__node = partner
+        self._nodes = otherNodes
+        self._utility = TcpUtility(conf.password)
 
-    def sendMessage(self, message):
-        # Abuse the fact that node address is send as a first message
-        self._SyncObj__transport._selfNode = MessageNode(message)
-        self._SyncObj__transport.connectIfRequiredSingle(self.__node)
-        while not self._SyncObj__transport.isDisconnected(self.__node):
-            self._poller.poll(0.5)
-        return self.__result
+    def executeCommand(self, command):
+        try:
+            return self._utility.executeCommand(self.__node, command)
+        except UtilityException:
+            return None
 
-    def _onMessageReceived(self, _, message):
-        self.__result = message
-        self._SyncObj__transport.disconnectSingle(self.__node)
+    def getMembers(self):
+        for self.__node in self._nodes:
+            response = self.executeCommand(['members'])
+            if response:
+                return [member['addr'] for member in response]
 
 
-class MyTCPTransport(TCPTransport):
+class DynMemberSyncObj(SyncObj):
 
-    def _onIncomingMessageReceived(self, conn, message):
-        if self._syncObj.encryptor and not conn.sendRandKey:
-            conn.sendRandKey = message
-            conn.recvRandKey = os.urandom(32)
-            conn.send(conn.recvRandKey)
-            return
+    def __init__(self, selfAddress, partnerAddrs, conf):
+        self.__early_apply_local_log = selfAddress is not None
+        self.applied_local_log = False
 
-        # Utility messages
-        if isinstance(message, list) and message[0] == 'members':
-            conn.send(self._syncObj._get_members())
-            return True
+        utility = SyncObjUtility(partnerAddrs, conf)
+        members = utility.getMembers()
+        add_self = members and selfAddress not in members
 
-        return super(MyTCPTransport, self)._onIncomingMessageReceived(conn, message)
+        partnerAddrs = [member for member in (members or partnerAddrs) if member != selfAddress]
 
+        super(DynMemberSyncObj, self).__init__(selfAddress, partnerAddrs, conf, transportClass=_TCPTransport)
 
-class DynMemberSyncObj(SyncObj):
+        if add_self:
+            thread = threading.Thread(target=utility.executeCommand, args=(['add', selfAddress],))
+            thread.daemon = True
+            thread.start()
 
-    def __init__(self, selfAddress, partnerAddrs, conf):
-        add_self = False
-        utility = SyncObjUtility(partnerAddrs, conf)
-        for node in utility._SyncObj__otherNodes:
-            utility.setPartnerNode(node)
-            response = utility.sendMessage(['members'])
-            if response:
-                partnerAddrs = [member['addr'] for member in response if member['addr'] != selfAddress]
-                add_self = selfAddress and len(partnerAddrs) == len(response)
-                break
+    def getMembers(self, args, callback):
+        callback([{'addr': node.id, 'leader': node == self._getLeader(), 'status': CONNECTION_STATE.CONNECTED
+                   if self.isNodeConnected(node) else CONNECTION_STATE.DISCONNECTED} for node in self.otherNodes] +
+                 [{'addr': self.selfNode.id, 'leader': self._isLeader(), 'status': CONNECTION_STATE.CONNECTED}], None)
 
-        super(DynMemberSyncObj, self).__init__(selfAddress, partnerAddrs, conf, transportClass=MyTCPTransport)
-        if add_self:
-            threading.Thread(target=utility.sendMessage, args=(['add', selfAddress],)).start()
-
-    def _get_members(self):
-        ret = [{'addr': node.id, 'leader': node == self._getLeader(),
-                'status': CONNECTION_STATE.CONNECTED if node in self._SyncObj__connectedNodes
-                else CONNECTION_STATE.DISCONNECTED} for node in self._SyncObj__otherNodes]
-        ret.append({'addr': self._SyncObj__selfNode.id, 'leader': self._isLeader(),
-                    'status': CONNECTION_STATE.CONNECTED})
-        return ret
+    def _onTick(self, timeToWait=0.0):
+        super(DynMemberSyncObj, self)._onTick(timeToWait)
 
-    def _SyncObj__doChangeCluster(self, request, reverse=False):
-        ret = False
-        if not self._SyncObj__selfNode or request[0] != 'add' or reverse or request[1] != self._SyncObj__selfNode.id:
-            ret = super(DynMemberSyncObj, self)._SyncObj__doChangeCluster(request, reverse)
-            if ret:
-                self.forceLogCompaction()
-        return ret
+        # The SyncObj calls onReady callback only when cluster got the leader and is ready for writes.
+        # In some cases for us it is safe to "signal" the Raft object when the local log is fully applied.
+        # We are using the `applied_local_log` property for that, but not calling the callback function.
+        if self.__early_apply_local_log and not self.applied_local_log and self.raftLastApplied == self.raftCommitIndex:
+            self.applied_local_log = True
 
 
 class KVStoreTTL(DynMemberSyncObj):
 
-    def __init__(self, selfAddress, partnerAddrs, conf, on_set=None, on_delete=None):
+    def __init__(self, on_ready, on_set, on_delete, **config):
+        self.__thread = None
         self.__on_set = on_set
         self.__on_delete = on_delete
         self.__limb = {}
         self.__retry_timeout = None
-        self.__early_apply_local_log = selfAddress is not None
-        self.applied_local_log = False
-        super(KVStoreTTL, self).__init__(selfAddress, partnerAddrs, conf)
+
+        self_addr = config.get('self_addr')
+        partner_addrs = set(config.get('partner_addrs', []))
+        if config.get('patronictl'):
+            if self_addr:
+                partner_addrs.add(self_addr)
+            self_addr = None
+
+        # Create raft data_dir if necessary
+        raft_data_dir = config.get('data_dir', '')
+        if raft_data_dir != '':
+            validate_directory(raft_data_dir)
+
+        file_template = (self_addr or '')
+        file_template = file_template.replace(':', '_') if os.name == 'nt' else file_template
+        file_template = os.path.join(raft_data_dir, file_template)
+        conf = SyncObjConf(password=config.get('password'), autoTick=False, appendEntriesUseBatch=False,
+                           bindAddress=config.get('bind_addr'), commandsWaitLeader=config.get('commandsWaitLeader'),
+                           fullDumpFile=(file_template + '.dump' if self_addr else None),
+                           journalFile=(file_template + '.journal' if self_addr else None),
+                           onReady=on_ready, dynamicMembershipChange=True)
+
+        super(KVStoreTTL, self).__init__(self_addr, partner_addrs, conf)
         self.__data = {}
 
     @staticmethod
@@ -175,7 +151,7 @@ class KVStoreTTL(DynMemberSyncObj):
 
         if old_value and old_value['created'] != value['created']:
             value['created'] = value['updated']
-        value['index'] = self._SyncObj__raftLastApplied + 1
+        value['index'] = self.raftLastApplied + 1
 
         self.__data[key] = value
         if self.__on_set:
@@ -242,27 +218,29 @@ class KVStoreTTL(DynMemberSyncObj):
         return {k: v for k, v in self.__data.items() if k.startswith(key)}
 
     def _onTick(self, timeToWait=0.0):
-        # The SyncObj starts applying the local log only when there is at least one node connected.
-        # We want to change this behavior and apply the local log even when there is nobody except us.
-        # It gives us at least some picture about the last known cluster state.
-        if self.__early_apply_local_log and not self.applied_local_log and self._SyncObj__needLoadDumpFile:
-            self._SyncObj__raftCommitIndex = self._SyncObj__getCurrentLogIndex()
-            self._SyncObj__raftCurrentTerm = self._SyncObj__getCurrentLogTerm()
-
         super(KVStoreTTL, self)._onTick(timeToWait)
 
-        # The SyncObj calls onReady callback only when cluster got the leader and is ready for writes.
-        # In some cases for us it is safe to "signal" the Raft object when the local log is fully applied.
-        # We are using the `applied_local_log` property for that, but not calling the callback function.
-        if self.__early_apply_local_log and not self.applied_local_log and self._SyncObj__raftCommitIndex != 1 and \
-                self._SyncObj__raftLastApplied == self._SyncObj__raftCommitIndex:
-            self.applied_local_log = True
-
         if self._isLeader():
             self.__expire_keys()
         else:
             self.__limb.clear()
 
+    def _autoTickThread(self):
+        self.__destroying = False
+        while not self.__destroying:
+            self.doTick(self.conf.autoTickPeriod)
+
+    def startAutoTick(self):
+        self.__thread = threading.Thread(target=self._autoTickThread)
+        self.__thread.daemon = True
+        self.__thread.start()
+
+    def destroy(self):
+        if self.__thread:
+            self.__destroying = True
+            self.__thread.join()
+        super(KVStoreTTL, self).destroy()
+
 
 class Raft(AbstractDCS):
 
@@ -270,34 +248,16 @@ class Raft(AbstractDCS):
         super(Raft, self).__init__(config)
         self._ttl = int(config.get('ttl') or 30)
 
-        self_addr = config.get('self_addr')
-        partner_addrs = config.get('partner_addrs', [])
-        if self._ctl:
-            if self_addr:
-                partner_addrs.append(self_addr)
-            self_addr = None
-
-        # Create raft data_dir if necessary
-        raft_data_dir = config.get('data_dir', '')
-        if raft_data_dir != '':
-            validate_directory(raft_data_dir)
-
         ready_event = threading.Event()
-        file_template = os.path.join(config.get('data_dir', ''), (self_addr or ''))
-        conf = SyncObjConf(password=config.get('password'), appendEntriesUseBatch=False,
-                           bindAddress=config.get('bind_addr'), commandsWaitLeader=False,
-                           fullDumpFile=(file_template + '.dump' if self_addr else None),
-                           journalFile=(file_template + '.journal' if self_addr else None),
-                           onReady=ready_event.set, dynamicMembershipChange=True)
+        self._sync_obj = KVStoreTTL(ready_event.set, self._on_set, self._on_delete, commandsWaitLeader=False, **config)
+        self._sync_obj.startAutoTick()
 
-        self._sync_obj = KVStoreTTL(self_addr, partner_addrs, conf, self._on_set, self._on_delete)
         while True:
             ready_event.wait(5)
             if ready_event.isSet() or self._sync_obj.applied_local_log:
                 break
             else:
                 logger.info('waiting on raft')
-        self._sync_obj.forceLogCompaction()
         self.set_retry_timeout(int(config.get('retry_timeout') or 10))
 
     def _on_set(self, key, value):
diff --git a/patroni/postgresql/config.py b/patroni/postgresql/config.py
index 234b2523..d0e5fc5b 100644
--- a/patroni/postgresql/config.py
+++ b/patroni/postgresql/config.py
@@ -879,13 +879,13 @@ class ConfigHandler(object):
         tcp_local_address = self._get_tcp_local_address()
         netloc = self._config.get('connect_address') or tcp_local_address + ':' + port
 
+        unix_local_address = {'port': port}
         unix_socket_directories = self._server_parameters.get('unix_socket_directories')
-        # fallback to tcp if unix_socket_directories is set, but there are no sutable values
-        unix_local_address = unix_socket_directories and\
-            self._get_unix_local_address(unix_socket_directories) or tcp_local_address
+        if unix_socket_directories is not None:
+            # fallback to tcp if unix_socket_directories is set, but there are no sutable values
+            unix_local_address['host'] = self._get_unix_local_address(unix_socket_directories) or tcp_local_address
 
         tcp_local_address = {'host': tcp_local_address, 'port': port}
-        unix_local_address = {'host': unix_local_address, 'port': port}
 
         self._local_address = unix_local_address if self._config.get('use_unix_socket') else tcp_local_address
         self.local_replication_address = unix_local_address\
diff --git a/patroni/postgresql/connection.py b/patroni/postgresql/connection.py
index 933ee89c..434bda59 100644
--- a/patroni/postgresql/connection.py
+++ b/patroni/postgresql/connection.py
@@ -40,7 +40,8 @@ class Connection(object):
 
 @contextmanager
 def get_connection_cursor(**kwargs):
-    with psycopg2.connect(**kwargs) as conn:
-        conn.autocommit = True
-        with conn.cursor() as cur:
-            yield cur
+    conn = psycopg2.connect(**kwargs)
+    conn.autocommit = True
+    with conn.cursor() as cur:
+        yield cur
+    conn.close()
diff --git a/patroni/postgresql/slots.py b/patroni/postgresql/slots.py
index cee319d9..43b7e30c 100644
--- a/patroni/postgresql/slots.py
+++ b/patroni/postgresql/slots.py
@@ -5,6 +5,7 @@ import shutil
 
 from collections import defaultdict
 from contextlib import contextmanager
+from psycopg2.errors import UndefinedFile
 
 from .connection import get_connection_cursor
 from .misc import format_lsn
@@ -198,7 +199,9 @@ class SlotsHandler(object):
                         cur.execute("SELECT pg_catalog.pg_replication_slot_advance(%s, %s)",
                                     (name, format_lsn(int(cluster.slots[name]))))
                     except Exception as e:
-                        logger.exception("Failed to advance logical replication slot '%s': %r", name, e)
+                        logger.error("Failed to advance logical replication slot '%s': %r", name, e)
+                        if isinstance(e, UndefinedFile):
+                            create_slots.append(name)
                         self._schedule_load_slots = True
         return create_slots
 
@@ -280,6 +283,8 @@ class SlotsHandler(object):
                     f.write(value['data'])
                     f.flush()
                     os.fsync(f.fileno())
+                if os.path.exists(slot_dir):
+                    shutil.rmtree(slot_dir)
                 os.rename(slot_tmp_dir, slot_dir)
                 fsync_dir(slot_dir)
                 self._unready_logical_slots.add(name)
diff --git a/patroni/raft_controller.py b/patroni/raft_controller.py
index 5a9ae704..1baa9786 100644
--- a/patroni/raft_controller.py
+++ b/patroni/raft_controller.py
@@ -1,9 +1,7 @@
 import logging
-import os
 
-from patroni.daemon import AbstractPatroniDaemon, abstract_main
-from patroni.dcs.raft import KVStoreTTL
-from pysyncobj import SyncObjConf
+from .daemon import AbstractPatroniDaemon, abstract_main
+from .dcs.raft import KVStoreTTL
 
 logger = logging.getLogger(__name__)
 
@@ -13,16 +11,13 @@ class RaftController(AbstractPatroniDaemon):
     def __init__(self, config):
         super(RaftController, self).__init__(config)
 
-        raft_config = self.config.get('raft')
-        self_addr = raft_config['self_addr']
-        template = os.path.join(raft_config.get('data_dir', ''), self_addr)
-        self._syncobj_config = SyncObjConf(autoTick=False, appendEntriesUseBatch=False, dynamicMembershipChange=True,
-                                           journalFile=template + '.journal', fullDumpFile=template + '.dump')
-        self._raft = KVStoreTTL(self_addr, raft_config.get('partner_addrs', []), self._syncobj_config)
+        config = self.config.get('raft')
+        assert 'self_addr' in config
+        self._raft = KVStoreTTL(None, None, None, **config)
 
     def _run_cycle(self):
         try:
-            self._raft.doTick(self._syncobj_config.autoTickPeriod)
+            self._raft.doTick(self._raft.conf.autoTickPeriod)
         except Exception:
             logger.exception('doTick')
 
diff --git a/requirements.txt b/requirements.txt
index 9fc6f4b4..c7438d3d 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -9,6 +9,7 @@ python-consul>=0.7.1
 click>=4.1
 prettytable>=0.7
 python-dateutil
-pysyncobj>=0.3.7
+pysyncobj>=0.3.8
+cryptography>=1.4
 psutil>=2.0.0
 ydiff>=1.2.0
diff --git a/setup.py b/setup.py
index 17dd696b..7302858a 100644
--- a/setup.py
+++ b/setup.py
@@ -22,8 +22,9 @@ AUTHOR_EMAIL = 'alexander.kukushkin@zalando.de, dmitrii.dolgov@zalando.de, alexk
 KEYWORDS = 'etcd governor patroni postgresql postgres ha haproxy confd' +\
     ' zookeeper exhibitor consul streaming replication kubernetes k8s'
 
-EXTRAS_REQUIRE = {'aws': ['boto'], 'etcd': ['python-etcd'], 'etcd3': ['python-etcd'], 'consul': ['python-consul'],
-                  'exhibitor': ['kazoo'], 'zookeeper': ['kazoo'], 'kubernetes': ['ipaddress'], 'raft': ['pysyncobj']}
+EXTRAS_REQUIRE = {'aws': ['boto'], 'etcd': ['python-etcd'], 'etcd3': ['python-etcd'],
+                  'consul': ['python-consul'], 'exhibitor': ['kazoo'], 'zookeeper': ['kazoo'],
+                  'kubernetes': ['ipaddress'], 'raft': ['pysyncobj', 'cryptography']}
 COVERAGE_XML = True
 COVERAGE_HTML = False
 
@@ -171,10 +172,16 @@ def setup_package(version):
         if r == '':
             continue
         extra = False
-        for e, v in EXTRAS_REQUIRE.items():
-            if v and r.startswith(v[0]):
-                EXTRAS_REQUIRE[e] = [r] if e != 'kubernetes' or sys.version_info < (3, 0, 0) else []
-                extra = True
+        for e, deps in EXTRAS_REQUIRE.items():
+            for i, v in enumerate(deps):
+                if r.startswith(v):
+                    if e != 'kubernetes' or sys.version_info < (3, 0, 0):
+                        deps[i] = r
+                    else:
+                        deps = []
+                    EXTRAS_REQUIRE[e] = deps
+                    extra = True
+                    break
         if not extra:
             install_requires.append(r)
 


Test patch:
diff --git a/tests/test_api.py b/tests/test_api.py
index 079753f4..038d1f99 100644
--- a/tests/test_api.py
+++ b/tests/test_api.py
@@ -174,7 +174,7 @@ class TestRestApiHandler(unittest.TestCase):
             MockRestApiServer(RestApiHandler, 'GET /replica')
         with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={'state': 'running'})):
             MockRestApiServer(RestApiHandler, 'GET /health')
-        MockRestApiServer(RestApiHandler, 'GET /master')
+        MockRestApiServer(RestApiHandler, 'GET /leader')
         MockPatroni.dcs.cluster.sync.members = [MockPostgresql.name]
         MockPatroni.dcs.cluster.is_synchronous_mode = Mock(return_value=True)
         with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={'role': 'replica'})):
@@ -507,3 +507,7 @@ class TestRestApiServer(unittest.TestCase):
                 Mock(return_value=(mock_request, mock_address))
             ):
                 self.srv._handle_request_noblock()
+
+    @patch('ssl._ssl._test_decode_cert', Mock())
+    def test_reload_local_certificate(self):
+        self.assertTrue(self.srv.reload_local_certificate())
diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py
index a13dd73a..42d01e6b 100644
--- a/tests/test_postgresql.py
+++ b/tests/test_postgresql.py
@@ -492,6 +492,14 @@ class TestPostgresql(BaseTestPostgresql):
         self.p.reload_config(config)
         self.p.config.resolve_connection_addresses()
 
+    def test_resolve_connection_addresses(self):
+        self.p.config._config['use_unix_socket'] = self.p.config._config['use_unix_socket_repl'] = True
+        self.p.config.resolve_connection_addresses()
+        self.assertEqual(self.p.config.local_replication_address, {'host': '/tmp', 'port': '5432'})
+        self.p.config._server_parameters.pop('unix_socket_directories')
+        self.p.config.resolve_connection_addresses()
+        self.assertEqual(self.p.config._local_address, {'port': '5432'})
+
     @patch.object(Postgresql, '_version_file_exists', Mock(return_value=True))
     def test_get_major_version(self):
         with patch.object(builtins, 'open', mock_open(read_data='9.4')):
diff --git a/tests/test_raft.py b/tests/test_raft.py
index 54197df8..716295da 100644
--- a/tests/test_raft.py
+++ b/tests/test_raft.py
@@ -3,13 +3,13 @@ import unittest
 import tempfile
 import time
 
-from mock import Mock, patch
+from mock import Mock, PropertyMock, patch
 from patroni.dcs.raft import DynMemberSyncObj, KVStoreTTL, Raft, SyncObjUtility
 from pysyncobj import SyncObjConf, FAIL_REASON
 
 
 def remove_files(prefix):
-    for f in ('journal', 'dump'):
+    for f in ('journal', 'journal.meta', 'dump'):
         f = prefix + f
         if os.path.isfile(f):
             for i in range(0, 15):
@@ -31,50 +31,38 @@ class TestDynMemberSyncObj(unittest.TestCase):
         self.conf = SyncObjConf(appendEntriesUseBatch=False, dynamicMembershipChange=True, autoTick=False)
         self.so = DynMemberSyncObj('127.0.0.1:1234', ['127.0.0.1:1235'], self.conf)
 
-    @patch.object(SyncObjUtility, 'sendMessage')
-    def test_add_member(self, mock_send_message):
-        mock_send_message.return_value = [{'addr': '127.0.0.1:1235'}, {'addr': '127.0.0.1:1236'}]
-        mock_send_message.ver = 0
+    @patch.object(SyncObjUtility, 'executeCommand')
+    def test_add_member(self, mock_execute_command):
+        mock_execute_command.return_value = [{'addr': '127.0.0.1:1235'}, {'addr': '127.0.0.1:1236'}]
+        mock_execute_command.ver = 0
         DynMemberSyncObj('127.0.0.1:1234', ['127.0.0.1:1235'], self.conf)
         self.conf.dynamicMembershipChange = False
         DynMemberSyncObj('127.0.0.1:1234', ['127.0.0.1:1235'], self.conf)
 
-    def test___onUtilityMessage(self):
-        self.so._SyncObj__encryptor = Mock()
+    def test_getMembers(self):
         mock_conn = Mock()
-        mock_conn.sendRandKey = None
-        self.so._SyncObj__transport._onIncomingMessageReceived(mock_conn, 'randkey')
         self.so._SyncObj__transport._onIncomingMessageReceived(mock_conn, ['members'])
-        self.so._SyncObj__transport._onIncomingMessageReceived(mock_conn, ['status'])
 
     def test__SyncObj__doChangeCluster(self):
         self.so._SyncObj__doChangeCluster(['add', '127.0.0.1:1236'])
 
-    def test_utility(self):
-        utility = SyncObjUtility(['127.0.0.1:1235'], self.conf)
-        utility.setPartnerNode(list(utility._SyncObj__otherNodes)[0])
-        utility.sendMessage(['members'])
-        utility._onMessageReceived(0, '')
-
 
+@patch.object(SyncObjConf, 'fullDumpFile', PropertyMock(return_value=None), create=True)
+@patch.object(SyncObjConf, 'journalFile', PropertyMock(return_value=None), create=True)
 class TestKVStoreTTL(unittest.TestCase):
 
+    @patch.object(SyncObjConf, 'fullDumpFile', PropertyMock(return_value=None), create=True)
+    @patch.object(SyncObjConf, 'journalFile', PropertyMock(return_value=None), create=True)
     def setUp(self):
-        self.conf = SyncObjConf(appendEntriesUseBatch=False, appendEntriesPeriod=0.001,
-                                raftMinTimeout=0.004, raftMaxTimeout=0.005, autoTickPeriod=0.001)
         callback = Mock()
         callback.replicated = False
-        self.so = KVStoreTTL('127.0.0.1:1234', [], self.conf, on_set=callback, on_delete=callback)
+        self.so = KVStoreTTL(None, callback, callback, self_addr='127.0.0.1:1234')
+        self.so.startAutoTick()
         self.so.set_retry_timeout(10)
 
-    @staticmethod
-    def destroy(so):
-        so.destroy()
-        so._SyncObj__thread.join()
-
     def tearDown(self):
         if self.so:
-            self.destroy(self.so)
+            self.so.destroy()
 
     def test_set(self):
         self.assertTrue(self.so.set('foo', 'bar', prevExist=False, ttl=30))
@@ -83,7 +71,7 @@ class TestKVStoreTTL(unittest.TestCase):
         self.assertTrue(self.so.retry(self.so._set, 'foo', {'value': 'buz', 'created': 1, 'updated': 1}))
 
     def test_delete(self):
-        self.conf.autoTickPeriod = 0.1
+        self.so.autoTickPeriod = 0.2
         self.so.set('foo', 'bar')
         self.so.set('fooo', 'bar')
         self.assertFalse(self.so.delete('foo', prevValue='buz'))
@@ -111,11 +99,10 @@ class TestKVStoreTTL(unittest.TestCase):
 
     def test_on_ready_override(self):
         self.assertTrue(self.so.set('foo', 'bar'))
-        self.destroy(self.so)
+        self.so.destroy()
         self.so = None
-        self.conf.onReady = Mock()
-        self.conf.autoTick = False
-        so = KVStoreTTL('127.0.0.1:1234', ['127.0.0.1:1235'], self.conf)
+        so = KVStoreTTL(Mock(), None, None, self_addr='127.0.0.1:1234',
+                        partner_addrs=['127.0.0.1:1235'], patronictl=True)
         so.doTick(0)
         so.destroy()
 
@@ -150,7 +137,6 @@ class TestRaft(unittest.TestCase):
         self.assertTrue(raft.take_leader())
         raft.watch(None, 0.001)
         raft._sync_obj.destroy()
-        raft._sync_obj._SyncObj__thread.join()
 
     def tearDown(self):
         remove_files(os.path.join(self._TMP, '127.0.0.1:1234.'))
diff --git a/tests/test_slots.py b/tests/test_slots.py
index e17c4779..8c281ce7 100644
--- a/tests/test_slots.py
+++ b/tests/test_slots.py
@@ -86,8 +86,8 @@ class TestSlotsHandler(BaseTestPostgresql):
                           [self.me, self.other, self.leadermem], None, None, None, {'ls': 12346})
         self.assertEqual(self.s.sync_replication_slots(cluster, False), [])
         self.s._schedule_load_slots = False
-        with patch.object(MockCursor, 'execute', Mock(side_effect=psycopg2.OperationalError)):
-            self.assertEqual(self.s.sync_replication_slots(cluster, False), [])
+        with patch.object(MockCursor, 'execute', Mock(side_effect=psycopg2.errors.UndefinedFile)):
+            self.assertEqual(self.s.sync_replication_slots(cluster, False), ['ls'])
         cluster.slots['ls'] = 'a'
         self.assertEqual(self.s.sync_replication_slots(cluster, False), [])
         with patch.object(MockCursor, 'rowcount', PropertyMock(return_value=1), create=True):