Repository: zalando/patroni
Pull request #1957 (2021-06-11 13:00:10)

Track /status key updates
The #1820 introduced a new key in DCS, named /status, which contains the leader optime and maybe flush LSN of logical slots.

In case of etcd3 and raft we should use updates of this key for syncing HA loop across nodes (before we were using /optime/leader).


Patch:
diff --git a/.github/workflows/install_deps.py b/.github/workflows/install_deps.py
index fb5194cc..187d634f 100644
--- a/.github/workflows/install_deps.py
+++ b/.github/workflows/install_deps.py
@@ -33,13 +33,17 @@ def install_requirements(what):
 
 
 def install_packages(what):
+    from mapping import versions
+
     packages = {
         'zookeeper': ['zookeeper', 'zookeeper-bin', 'zookeeperd'],
         'consul': ['consul'],
     }
     packages['exhibitor'] = packages['zookeeper']
     packages = packages.get(what, [])
-    ver = str({'etcd': '9.6', 'etcd3': '13', 'consul': 12, 'exhibitor': 11, 'kubernetes': 13, 'raft': 12}.get(what))
+    ver = versions.get(what)
+    subprocess.call(['sudo', 'sed', '-i', 's/pgdg main.*$/pgdg main {0}/'.format(ver),
+                     '/etc/apt/sources.list.d/pgdg.list'])
     subprocess.call(['sudo', 'apt-get', 'update', '-y'])
     return subprocess.call(['sudo', 'apt-get', 'install', '-y', 'postgresql-' + ver, 'expect-dev', 'wget'] + packages)
 
diff --git a/.github/workflows/mapping.py b/.github/workflows/mapping.py
new file mode 100644
index 00000000..abe30a39
--- /dev/null
+++ b/.github/workflows/mapping.py
@@ -0,0 +1,1 @@
+versions = {'etcd': '9.6', 'etcd3': '14', 'consul': '13', 'exhibitor': '12', 'raft': '11', 'kubernetes': '14'}
diff --git a/docs/SETTINGS.rst b/docs/SETTINGS.rst
index 75531dd9..664a62c8 100644
--- a/docs/SETTINGS.rst
+++ b/docs/SETTINGS.rst
@@ -373,3 +373,12 @@ Tags
 - **noloadbalance**: ``true`` or ``false``. If set to ``true`` the node will return HTTP Status Code 503 for the ``GET /replica`` REST API health-check and therefore will be excluded from the load-balancing. Defaults to ``false``.
 - **replicatefrom**: The IP address/hostname of another replica. Used to support cascading replication.
 - **nosync**: ``true`` or ``false``. If set to ``true`` the node will never be selected as a synchronous replica.
+
+In addition to these predefined tags, you can also add your own ones:
+
+- **key1**: ``true``
+- **key2**: ``false``
+- **key3**: ``1.4``
+- **key4**: ``"RandomString"``
+
+Tags are visible in the :ref:`REST API <rest_api>` and ``patronictl list`` You can also check for an instance health using these tags. If the tag isn't defined for an instance, or if the respective value doesn't match the querying value, it will return HTTP Status Code 503.
diff --git a/docs/rest_api.rst b/docs/rest_api.rst
index 74003540..05d579c9 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.
@@ -26,9 +29,18 @@ For all health check ``GET`` requests Patroni returns a JSON document with the s
   - ``GET /replica?lag=10MB``
   - ``GET /replica?lag=1GB``
 
-- ``GET /read-only``: like the above endpoint, but also includes the primary.
+- ``GET /replica?tag_key1=value1&tag_key2=value2``: replica check endpoint. In addition, It will also check for user defined tags ``key1`` and ``key2`` and their respective values in the **tags** section of the yaml configuration management. If the tag isn't defined for an instance, or if the value in the yaml configuration doesn't match the querying value, it will return HTTP Status Code 503.
 
-- ``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>`.
+ In the following requests, since we are checking for the leader or standby-leader status, Patroni doesn't apply any of the user defined tags and they will be ignored.
+  - ``GET /?tag_key1=value1&tag_key2=value2``
+  - ``GET /master?tag_key1=value1&tag_key2=value2``
+  - ``GET /leader?tag_key1=value1&tag_key2=value2``
+  - ``GET /primary?tag_key1=value1&tag_key2=value2``
+  - ``GET /read-write?tag_key1=value1&tag_key2=value2``
+  - ``GET /standby_leader?tag_key1=value1&tag_key2=value2``
+  - ``GET /standby-leader?tag_key1=value1&tag_key2=value2``
+
+- ``GET /read-only``: like the above endpoint, but also includes the primary.
 
 - ``GET /synchronous`` or ``GET /sync``: returns HTTP status code **200** only when the Patroni node is running as a synchronous standby.
 
diff --git a/features/basic_replication.feature b/features/basic_replication.feature
index 30891889..3aaeada6 100644
--- a/features/basic_replication.feature
+++ b/features/basic_replication.feature
@@ -29,7 +29,7 @@ Feature: basic replication
     Then I receive a response code 200
 
   Scenario: check stuck sync replica
-    Given I issue a PATCH request to http://127.0.0.1:8008/config with {"maximum_lag_on_syncnode": 15000000, "postgresql": {"parameters": {"synchronous_commit": "remote_apply"}}}
+    Given I issue a PATCH request to http://127.0.0.1:8008/config with {"pause": true, "maximum_lag_on_syncnode": 15000000, "postgresql": {"parameters": {"synchronous_commit": "remote_apply"}}}
     Then I receive a response code 200
     And I create table on postgres0
     And table mytest is present on postgres1 after 2 seconds
@@ -43,7 +43,7 @@ Feature: basic replication
     Then I receive a response code 200
     When I issue a GET request to http://127.0.0.1:8010/async
     Then I receive a response code 200
-    When I issue a PATCH request to http://127.0.0.1:8008/config with {"maximum_lag_on_syncnode": -1, "postgresql": {"parameters": {"synchronous_commit": "on"}}}
+    When I issue a PATCH request to http://127.0.0.1:8008/config with {"pause": null, "maximum_lag_on_syncnode": -1, "postgresql": {"parameters": {"synchronous_commit": "on"}}}
     Then I receive a response code 200
     And I drop table on postgres0
 
diff --git a/patroni/api.py b/patroni/api.py
index 4fbc19e5..2a34c242 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,14 +120,20 @@ 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
 
+        ignore_tags = False
         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:
+            ignore_tags = True
+        elif 'leader' in path:
+            status_code = leader_status_code
+            ignore_tags = True
+        elif 'master' in path or 'primary' in path or 'read-write' in path:
             status_code = primary_status_code
+            ignore_tags = True
         elif 'replica' in path:
             status_code = replica_status_code
         elif 'read-only' in path:
@@ -140,6 +148,25 @@ class RestApiHandler(BaseHTTPRequestHandler):
             elif path in ('/async', '/asynchronous') and not is_synchronous:
                 status_code = replica_status_code
 
+        # check for user defined tags in query params
+        if not ignore_tags and status_code == 200:
+            qs_tag_prefix = "tag_"
+            for qs_key, qs_value in self.path_query.items():
+                if not qs_key.startswith(qs_tag_prefix):
+                    continue
+                qs_key = qs_key[len(qs_tag_prefix):]
+                qs_value = qs_value[0]
+                instance_tag_value = patroni.tags.get(qs_key)
+                # tag not registered for instance
+                if instance_tag_value is None:
+                    status_code = 503
+                    break
+                if not isinstance(instance_tag_value, six.string_types):
+                    instance_tag_value = str(instance_tag_value).lower()
+                if instance_tag_value != qs_value:
+                    status_code = 503
+                    break
+
         if write_status_code_only:  # when haproxy sends OPTIONS request it reads only status code and nothing more
             self._write_status_code_only(status_code)
         else:
@@ -187,73 +214,74 @@ class RestApiHandler(BaseHTTPRequestHandler):
 
         metrics = []
 
+        scope_label = '{{scope="{0}"}}'.format(patroni.postgresql.scope)
         metrics.append("# HELP patroni_version Patroni semver without periods.")
         metrics.append("# TYPE patroni_version gauge")
         padded_semver = ''.join([x.zfill(2) for x in patroni.version.split('.')])  # 2.0.2 => 020002
-        metrics.append("patroni_version {0}".format(padded_semver))
+        metrics.append("patroni_version{0} {1}".format(scope_label, padded_semver))
 
         metrics.append("# HELP patroni_postgres_running Value is 1 if Postgres is running, 0 otherwise.")
         metrics.append("# TYPE patroni_postgres_running gauge")
-        metrics.append("patroni_postgres_running {0}".format(int(postgres['state'] == 'running')))
+        metrics.append("patroni_postgres_running{0} {1}".format(scope_label, int(postgres['state'] == 'running')))
 
         metrics.append("# HELP patroni_postmaster_start_time Epoch seconds since Postgres started.")
         metrics.append("# TYPE patroni_postmaster_start_time gauge")
         postmaster_start_time = postgres.get('postmaster_start_time')
         postmaster_start_time = (postmaster_start_time - epoch).total_seconds() if postmaster_start_time else 0
-        metrics.append("patroni_postmaster_start_time {0}".format(postmaster_start_time))
+        metrics.append("patroni_postmaster_start_time{0} {1}".format(scope_label, postmaster_start_time))
 
         metrics.append("# HELP patroni_master Value is 1 if this node is the leader, 0 otherwise.")
         metrics.append("# TYPE patroni_master gauge")
-        metrics.append("patroni_master {0}".format(int(postgres['role'] == 'master')))
+        metrics.append("patroni_master{0} {1}".format(scope_label, int(postgres['role'] == 'master')))
 
         metrics.append("# HELP patroni_xlog_location Current location of the Postgres"
                        " transaction log, 0 if this node is not the leader.")
         metrics.append("# TYPE patroni_xlog_location counter")
-        metrics.append("patroni_xlog_location {0}".format(postgres.get('xlog', {}).get('location', 0)))
+        metrics.append("patroni_xlog_location{0} {1}".format(scope_label, postgres.get('xlog', {}).get('location', 0)))
 
         metrics.append("# HELP patroni_standby_leader Value is 1 if this node is the standby_leader, 0 otherwise.")
         metrics.append("# TYPE patroni_standby_leader gauge")
-        metrics.append("patroni_standby_leader {0}".format(int(postgres['role'] == 'standby_leader')))
+        metrics.append("patroni_standby_leader{0} {1}".format(scope_label, int(postgres['role'] == 'standby_leader')))
 
         metrics.append("# HELP patroni_replica Value is 1 if this node is a replica, 0 otherwise.")
         metrics.append("# TYPE patroni_replica gauge")
-        metrics.append("patroni_replica {0}".format(int(postgres['role'] == 'replica')))
+        metrics.append("patroni_replica{0} {1}".format(scope_label, int(postgres['role'] == 'replica')))
 
         metrics.append("# HELP patroni_xlog_received_location Current location of the received"
                        " Postgres transaction log, 0 if this node is not a replica.")
         metrics.append("# TYPE patroni_xlog_received_location counter")
-        metrics.append("patroni_xlog_received_location {0}".format(
-                        postgres.get('xlog', {}).get('received_location', 0)))
+        metrics.append("patroni_xlog_received_location{0} {1}"
+                       .format(scope_label, postgres.get('xlog', {}).get('received_location', 0)))
 
         metrics.append("# HELP patroni_xlog_replayed_location Current location of the replayed"
                        " Postgres transaction log, 0 if this node is not a replica.")
         metrics.append("# TYPE patroni_xlog_replayed_location counter")
-        metrics.append("patroni_xlog_replayed_location {0}".format(
-                        postgres.get('xlog', {}).get('replayed_location', 0)))
+        metrics.append("patroni_xlog_replayed_location{0} {1}"
+                       .format(scope_label, postgres.get('xlog', {}).get('replayed_location', 0)))
 
         metrics.append("# HELP patroni_xlog_replayed_timestamp Current timestamp of the replayed"
                        " Postgres transaction log, 0 if null.")
         metrics.append("# TYPE patroni_xlog_replayed_timestamp gauge")
         replayed_timestamp = postgres.get('xlog', {}).get('replayed_timestamp')
         replayed_timestamp = (replayed_timestamp - epoch).total_seconds() if replayed_timestamp else 0
-        metrics.append("patroni_xlog_replayed_timestamp {0}".format(replayed_timestamp))
+        metrics.append("patroni_xlog_replayed_timestamp{0} {1}".format(scope_label, replayed_timestamp))
 
         metrics.append("# HELP patroni_xlog_paused Value is 1 if the Postgres xlog is paused, 0 otherwise.")
         metrics.append("# TYPE patroni_xlog_paused gauge")
-        metrics.append("patroni_xlog_paused {0}".format(
-                        int(postgres.get('xlog', {}).get('paused', False) is True)))
+        metrics.append("patroni_xlog_paused{0} {1}"
+                       .format(scope_label, int(postgres.get('xlog', {}).get('paused', False) is True)))
 
         metrics.append("# HELP patroni_postgres_server_version Version of Postgres (if running), 0 otherwise.")
         metrics.append("# TYPE patroni_postgres_server_version gauge")
-        metrics.append("patroni_postgres_server_version {0}".format(postgres.get('server_version', 0)))
+        metrics.append("patroni_postgres_server_version {0} {1}".format(scope_label, postgres.get('server_version', 0)))
 
         metrics.append("# HELP patroni_cluster_unlocked Value is 1 if the cluster is unlocked, 0 if locked.")
         metrics.append("# TYPE patroni_cluster_unlocked gauge")
-        metrics.append("patroni_cluster_unlocked {0}".format(int(postgres['cluster_unlocked'])))
+        metrics.append("patroni_cluster_unlocked{0} {1}".format(scope_label, int(postgres['cluster_unlocked'])))
 
         metrics.append("# HELP patroni_postgres_timeline Postgres timeline of this node (if running), 0 otherwise.")
         metrics.append("# TYPE patroni_postgres_timeline counter")
-        metrics.append("patroni_postgres_timeline {0}".format(postgres.get('timeline', 0)))
+        metrics.append("patroni_postgres_timeline{0} {1}".format(scope_label, postgres.get('timeline', 0)))
 
         self._write_response(200, '\n'.join(metrics)+'\n', content_type='text/plain')
 
diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py
index 10c2d5f8..9e0e225c 100644
--- a/patroni/dcs/__init__.py
+++ b/patroni/dcs/__init__.py
@@ -68,7 +68,11 @@ def dcs_modules():
 
     if getattr(sys, 'frozen', False):
         toc = set()
-        for importer in pkgutil.iter_importers(dcs_dirname):
+        # dcs_dirname may contain a dot, which causes pkgutil.iter_importers()
+        # to misinterpret the path as a package name. This can be avoided
+        # altogether by not passing a path at all, because PyInstaller's
+        # FrozenImporter is a singleton and registered as top-level finder.
+        for importer in pkgutil.iter_importers():
             if hasattr(importer, 'toc'):
                 toc |= importer.toc
         return [module for module in toc if module.startswith(module_prefix) and module.count('.') == 2]
diff --git a/patroni/dcs/etcd3.py b/patroni/dcs/etcd3.py
index 51786676..1e738100 100644
--- a/patroni/dcs/etcd3.py
+++ b/patroni/dcs/etcd3.py
@@ -262,6 +262,9 @@ class Etcd3Client(AbstractEtcdClientWithFailover):
         return self.api_execute(self.version_prefix + method, self._MPOST, fields)
 
     def authenticate(self):
+        if self._use_proxies and self._cluster_version is None:
+            kwargs = self._prepare_common_parameters(1)
+            self._ensure_version_prefix(self._base_uri, **kwargs)
         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
@@ -372,6 +375,7 @@ class KVCache(Thread):
         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._status_key = base64_encode(dcs.status_path)
         self._name = base64_encode(dcs._name)
         self._is_ready = False
         self._response = None
@@ -418,14 +422,15 @@ class KVCache(Thread):
             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._leader_key or key in (self._optime_key, self._status_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):
+            # We also want to wake up HA loop on replicas if leader optime (or status key) was updated
+            if value_changed and (key not in (self._optime_key, self._status_key) or
+                                  (self.get(self._leader_key) or {}).get('value') != self._name):
                 self._dcs.event.set()
 
     def _process_message(self, message):
diff --git a/patroni/dcs/raft.py b/patroni/dcs/raft.py
index 76c65efb..fed6d5df 100644
--- a/patroni/dcs/raft.py
+++ b/patroni/dcs/raft.py
@@ -5,8 +5,10 @@ import threading
 import time
 
 from pysyncobj import SyncObj, SyncObjConf, replicated, FAIL_REASON
+from pysyncobj.dns_resolver import globalDnsResolver
+from pysyncobj.node import TCPNode
 from pysyncobj.transport import TCPTransport, CONNECTION_STATE
-from pysyncobj.utility import TcpUtility, UtilityException
+from pysyncobj.utility import TcpUtility
 
 from . import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, SyncState, TimelineHistory
 from ..utils import validate_directory
@@ -20,6 +22,20 @@ class _TCPTransport(TCPTransport):
         super(_TCPTransport, self).__init__(syncObj, selfNode, otherNodes)
         self.setOnUtilityMessageCallback('members', syncObj.getMembers)
 
+    def _connectIfNecessarySingle(self, node):
+        try:
+            return super(_TCPTransport, self)._connectIfNecessarySingle(node)
+        except Exception as e:
+            logger.debug('Connection to %s failed: %r', node, e)
+            return False
+
+
+def resolve_host(self):
+    return globalDnsResolver().resolve(self.host)
+
+
+setattr(TCPNode, 'ip', property(resolve_host))
+
 
 class SyncObjUtility(object):
 
@@ -30,7 +46,7 @@ class SyncObjUtility(object):
     def executeCommand(self, command):
         try:
             return self._utility.executeCommand(self.__node, command)
-        except UtilityException:
+        except Exception:
             return None
 
     def getMembers(self):
@@ -99,7 +115,8 @@ class KVStoreTTL(DynMemberSyncObj):
         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'),
+                           bindAddress=config.get('bind_addr'), dnsFailCacheTime=(config.get('loop_wait') or 10),
+                           dnsCacheTime=(config.get('ttl') or 30), 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)
@@ -264,7 +281,8 @@ class Raft(AbstractDCS):
         leader = (self._sync_obj.get(self.leader_path) or {}).get('value')
         if key == value['created'] == value['updated'] and \
                 (key.startswith(self.members_path) or key == self.leader_path and leader != self._name) or \
-                key == self.leader_optime_path and leader != self._name or key in (self.config_path, self.sync_path):
+                key in (self.leader_optime_path, self.status_path) and leader != self._name or \
+                key in (self.config_path, self.sync_path):
             self.event.set()
 
     def _on_delete(self, key):
@@ -281,6 +299,10 @@ class Raft(AbstractDCS):
     def set_retry_timeout(self, retry_timeout):
         self._sync_obj.set_retry_timeout(retry_timeout)
 
+    def reload_config(self, config):
+        super(Raft, self).reload_config(config)
+        globalDnsResolver().setTimeouts(self.ttl, self.loop_wait)
+
     @staticmethod
     def member(key, value):
         return Member.from_node(value['index'], os.path.basename(key), None, value['value'])
diff --git a/patroni/dcs/zookeeper.py b/patroni/dcs/zookeeper.py
index ac0167c9..f3b9ac38 100644
--- a/patroni/dcs/zookeeper.py
+++ b/patroni/dcs/zookeeper.py
@@ -4,8 +4,9 @@ import select
 import time
 
 from kazoo.client import KazooClient, KazooState, KazooRetry
-from kazoo.exceptions import NoNodeError, NodeExistsError
+from kazoo.exceptions import NoNodeError, NodeExistsError, SessionExpiredError
 from kazoo.handlers.threading import SequentialThreadingHandler
+from kazoo.protocol.states import KeeperState
 
 from . import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, SyncState, TimelineHistory
 from ..exceptions import DCSError
@@ -55,6 +56,20 @@ class PatroniSequentialThreadingHandler(SequentialThreadingHandler):
             raise select.error(9, str(e))
 
 
+class PatroniKazooClient(KazooClient):
+
+    def _call(self, request, async_object):
+        # Before kazoo==2.7.0 it wasn't possible to send requests to zookeeper if
+        # the connection is in the SUSPENDED state and Patroni was strongly relying on it.
+        # The https://github.com/python-zk/kazoo/pull/588 changed it, and now such requests are queued.
+        # We override the `_call()` method in order to keep the old behavior.
+
+        if self._state == KeeperState.CONNECTING:
+            async_object.set_exception(SessionExpiredError())
+            return False
+        return super(PatroniKazooClient, self)._call(request, async_object)
+
+
 class ZooKeeper(AbstractDCS):
 
     def __init__(self, config):
@@ -68,10 +83,10 @@ class ZooKeeper(AbstractDCS):
                    'cert': 'certfile', 'key': 'keyfile', 'key_password': 'keyfile_password'}
         kwargs = {v: config[k] for k, v in mapping.items() if k in config}
 
-        self._client = KazooClient(hosts, handler=PatroniSequentialThreadingHandler(config['retry_timeout']),
-                                   timeout=config['ttl'], connection_retry=KazooRetry(max_delay=1, max_tries=-1,
-                                   sleep_func=time.sleep), command_retry=KazooRetry(deadline=config['retry_timeout'],
-                                   max_delay=1, max_tries=-1, sleep_func=time.sleep), **kwargs)
+        self._client = PatroniKazooClient(hosts, handler=PatroniSequentialThreadingHandler(config['retry_timeout']),
+                                          timeout=config['ttl'], connection_retry=KazooRetry(max_delay=1, max_tries=-1,
+                                          sleep_func=time.sleep), command_retry=KazooRetry(max_delay=1, max_tries=-1,
+                                          deadline=config['retry_timeout'], sleep_func=time.sleep), **kwargs)
         self._client.add_listener(self.session_listener)
 
         self._fetch_cluster = True
diff --git a/patroni/ha.py b/patroni/ha.py
index 8d94a94d..42237d9b 100644
--- a/patroni/ha.py
+++ b/patroni/ha.py
@@ -404,7 +404,7 @@ class Ha(object):
                     self.state_handler.set_role('replica')
 
                 if not node_to_follow:
-                    return 'no action'
+                    return 'no action. I am ({0})'.format(self.state_handler.name)
         elif is_leader:
             self.demote('immediate-nolock')
             return demote_reason
@@ -417,6 +417,9 @@ class Ha(object):
         if msg:
             return msg
 
+        if not self.is_paused():
+            self.state_handler.handle_parameter_change()
+
         role = 'standby_leader' if isinstance(node_to_follow, RemoteMember) and self.has_lock(False) else 'replica'
         # It might happen that leader key in the standby cluster references non-exiting member.
         # In this case it is safe to continue running without changing recovery.conf
@@ -992,14 +995,14 @@ class Ha(object):
                     # in case of standby cluster we don't really need to
                     # enforce anything, since the leader is not a master.
                     # So just remind the role.
-                    msg = 'no action.  i am the standby leader with the lock' \
+                    msg = 'no action. I am ({0}) the standby leader with the lock'.format(self.state_handler.name) \
                           if self.state_handler.role == 'standby_leader' else \
                           'promoted self to a standby leader because i had the session lock'
                     return self.enforce_follow_remote_master(msg)
                 else:
                     return self.enforce_master_role(
-                        'no action.  i am the leader with the lock',
-                        'promoted self to leader because i had the session lock'
+                        'no action. I am ({0}) the leader with the lock'.format(self.state_handler.name),
+                        'promoted self to leader because I had the session lock'
                     )
             else:
                 # Either there is no connection to DCS or someone else acquired the lock
@@ -1012,12 +1015,15 @@ class Ha(object):
                 else:
                     return 'not promoting because failed to update leader lock in DCS'
         else:
-            logger.info('does not have lock')
+            logger.debug('does not have lock')
+        lock_owner = self.cluster.leader and self.cluster.leader.name
         if self.is_standby_cluster():
-            return self.follow('cannot be a real master in standby cluster',
-                               'no action.  i am a secondary and i am following a standby leader', refresh=False)
-        return self.follow('demoting self because i do not have the lock and i was a leader',
-                           'no action.  i am a secondary and i am following a leader', refresh=False)
+            return self.follow('cannot be a real primary in a standby cluster',
+                               'no action. I am a secondary ({0}) and following a standby leader ({1})'.format(
+                                    self.state_handler.name, lock_owner), refresh=False)
+        return self.follow('demoting self because I do not have the lock and I was a leader',
+                           'no action. I am a secondary ({0}) and following a leader ({1})'.format(
+                                self.state_handler.name, lock_owner), refresh=False)
 
     def evaluate_scheduled_restart(self):
         if self._async_executor.busy:  # Restart already in progress
@@ -1213,9 +1219,6 @@ class Ha(object):
                 self._delete_leader()
                 return 'removed leader key after trying and failing to start postgres'
             return 'failed to start postgres'
-        self._crash_recovery_executed = False
-        if self._rewind.executed and not self._rewind.failed:
-            self._rewind.reset_state()
         return None
 
     def cancel_initialization(self):
@@ -1302,8 +1305,12 @@ class Ha(object):
     def _run_cycle(self):
         dcs_failed = False
         try:
-            self.load_cluster_from_dcs()
-            self.state_handler.reset_cluster_info_state(self.cluster, self.patroni.nofailover)
+            try:
+                self.load_cluster_from_dcs()
+                self.state_handler.reset_cluster_info_state(self.cluster, self.patroni.nofailover)
+            except Exception:
+                self.state_handler.reset_cluster_info_state(None, self.patroni.nofailover)
+                raise
 
             if self.is_paused():
                 self.watchdog.disable()
@@ -1335,12 +1342,24 @@ class Ha(object):
             if self.state_handler.bootstrapping:
                 return self.post_bootstrap()
 
-            if self.recovering and not self._rewind.is_needed:
+            if self.recovering:
                 self.recovering = False
-                # Check if we tried to recover and failed
-                msg = self.post_recover()
-                if msg is not None:
-                    return msg
+
+                if not self._rewind.is_needed:
+                    # Check if we tried to recover from postgres crash and failed
+                    msg = self.post_recover()
+                    if msg is not None:
+                        return msg
+
+                # Reset some states after postgres successfully started up
+                self._crash_recovery_executed = False
+                if self._rewind.executed and not self._rewind.failed:
+                    self._rewind.reset_state()
+
+                # The Raft cluster without a quorum takes a bit of time to stabilize.
+                # Therefore we want to postpone the leader race if we just started up.
+                if self.cluster.is_unlocked() and self.dcs.__class__.__name__ == 'Raft':
+                    return 'started as a secondary'
 
             # is data directory empty?
             if self.state_handler.data_directory_empty():
diff --git a/patroni/log.py b/patroni/log.py
index ce808bb1..1231f1dd 100644
--- a/patroni/log.py
+++ b/patroni/log.py
@@ -166,6 +166,8 @@ class PatroniLogger(Thread):
             self._root_logger.addHandler(self._queue_handler)
             self._root_logger.removeHandler(self._proxy_handler)
 
+        prev_record = None
+
         while True:
             self._close_old_handlers()
 
@@ -173,7 +175,18 @@ class PatroniLogger(Thread):
             if record is None:
                 break
 
-            self.log_handler.handle(record)
+            if self._root_logger.level == logging.INFO:
+                if record.msg.startswith('Lock owner: '):
+                    prev_record, record = record, None
+                else:
+                    if prev_record and prev_record.thread == record.thread:
+                        if not (record.msg.startswith('no action. ') or record.msg.startswith('PAUSE: no action')):
+                            self.log_handler.handle(prev_record)
+                        prev_record = None
+
+            if record:
+                self.log_handler.handle(record)
+
             self._queue_handler.queue.task_done()
 
     def shutdown(self):
diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py
index b33fff8b..b7059d5b 100644
--- a/patroni/postgresql/__init__.py
+++ b/patroni/postgresql/__init__.py
@@ -1,20 +1,22 @@
 import logging
 import os
 import psycopg2
+import re
 import shlex
 import shutil
+import six
 import subprocess
 import time
 
 from contextlib import contextmanager
 from copy import deepcopy
-from dateutil import tz
 from datetime import datetime
+from dateutil import tz
 from psutil import TimeoutExpired
 from threading import current_thread, Lock
 
-from .callback_executor import CallbackExecutor
 from .bootstrap import Bootstrap
+from .callback_executor import CallbackExecutor
 from .cancellable import CancellableSubprocess
 from .config import ConfigHandler, mtime
 from .connection import Connection, get_connection_cursor
@@ -109,10 +111,12 @@ class Postgresql(object):
         # Last known running process
         self._postmaster_proc = None
 
-        if self.is_running():
+        if self.is_running():  # we are "joining" already running postgres
             self.set_state('running')
             self.set_role('master' if self.is_leader() else 'replica')
-            self.config.write_postgresql_conf()  # we are "joining" already running postgres
+            # postpone writing postgresql.conf for 12+ because recovery parameters are not yet known
+            if self.major_version < 120000 or self.is_leader():
+                self.config.write_postgresql_conf()
             hba_saved = self.config.replace_pg_hba()
             ident_saved = self.config.replace_pg_ident()
             if hba_saved or ident_saved:
@@ -367,22 +371,58 @@ class Postgresql(object):
     def is_leader(self):
         return bool(self._cluster_info_state_get('timeline'))
 
+    def replay_paused(self):
+        return self._cluster_info_state_get('replay_paused')
+
+    def resume_wal_replay(self):
+        self._query('SELECT pg_catalog.pg_{0}_replay_resume()'.format(self.wal_name))
+
+    def handle_parameter_change(self):
+        if self.major_version >= 140000 and self.replay_paused():
+            logger.info('Resuming paused WAL replay for PostgreSQL 14+')
+            self.resume_wal_replay()
+
     def pg_control_timeline(self):
         try:
             return int(self.controldata().get("Latest checkpoint's TimeLineID"))
         except (TypeError, ValueError):
             logger.exception('Failed to parse timeline from pg_controldata output')
 
+    def parse_wal_record(self, timeline, lsn):
+        out, err = self.waldump(timeline, lsn, 1)
+        if out and not err:
+            match = re.match(r'^rmgr:\s+(.+?)\s+len \(rec/tot\):\s+\d+/\s+\d+, tx:\s+\d+, '
+                             r'lsn: ([0-9A-Fa-f]+/[0-9A-Fa-f]+), prev ([0-9A-Fa-f]+/[0-9A-Fa-f]+), '
+                             r'.*?desc: (.+)', out.decode('utf-8'))
+            if match:
+                return match.groups()
+        return None, None, None, None
+
     def latest_checkpoint_location(self):
-        """Returns checkpoint location for the cleanly shut down primary"""
+        """Returns checkpoint location for the cleanly shut down primary.
+           But, if we know that the checkpoint was written to the new WAL
+           due to the archive_mode=on, we will return the LSN of prev wal record (SWITCH)."""
 
         data = self.controldata()
-        lsn = data.get('Latest checkpoint location')
-        if data.get('Database cluster state') == 'shut down' and lsn:
+        timeline = data.get("Latest checkpoint's TimeLineID")
+        lsn = checkpoint_lsn = data.get('Latest checkpoint location')
+        if data.get('Database cluster state') == 'shut down' and lsn and timeline:
             try:
-                return str(parse_lsn(lsn))
-            except (IndexError, ValueError) as e:
-                logger.error('Exception when parsing lsn %s: %r', lsn, e)
+                checkpoint_lsn = parse_lsn(checkpoint_lsn)
+                rm_name, lsn, prev, desc = self.parse_wal_record(timeline, lsn)
+                desc = desc.strip().lower()
+                if rm_name == 'XLOG' and parse_lsn(lsn) == checkpoint_lsn and prev and\
+                        desc.startswith('checkpoint') and desc.endswith('shutdown'):
+                    _, lsn, _, desc = self.parse_wal_record(timeline, prev)
+                    prev = parse_lsn(prev)
+                    # If the cluster is shutdown with archive_mode=on, WAL is switched before writing the checkpoint.
+                    # In this case we want to take the LSN of previous record (switch) as the last known WAL location.
+                    if parse_lsn(lsn) == prev and desc.strip() in ('xlog switch', 'SWITCH'):
+                        return str(prev)
+            except Exception as e:
+                logger.error('Exception when parsing WAL pg_%sdump output: %r', self.wal_name, e)
+            if isinstance(checkpoint_lsn, six.integer_types):
+                return str(checkpoint_lsn)
 
     def is_running(self):
         """Returns PostmasterProcess if one is running on the data directory or None. If most recently seen process
@@ -754,6 +794,20 @@ class Postgresql(object):
                 logger.exception("Error when calling pg_controldata")
         return {}
 
+    def waldump(self, timeline, lsn, limit):
+        cmd = self.pgcommand('pg_{0}dump'.format(self.wal_name))
+        env = os.environ.copy()
+        env.update(LANG='C', LC_ALL='C', PGDATA=self._data_dir)
+        try:
+            waldump = subprocess.Popen([cmd, '-t', str(timeline), '-s', str(lsn), '-n', str(limit)],
+                                       stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env)
+            out, err = waldump.communicate()
+            waldump.wait()
+            return out, err
+        except Exception as e:
+            logger.error('Failed to execute `%s -t %s -s %s -n %s`: %r', cmd, timeline, lsn, limit, e)
+            return None, None
+
     @contextmanager
     def get_replication_connection_cursor(self, host='localhost', port=5432, **kwargs):
         conn_kwargs = self.config.replication.copy()
diff --git a/patroni/postgresql/config.py b/patroni/postgresql/config.py
index d0e5fc5b..c61a7695 100644
--- a/patroni/postgresql/config.py
+++ b/patroni/postgresql/config.py
@@ -702,7 +702,9 @@ class ConfigHandler(object):
         if wal_receiver_primary_slot_name is not None:
             self._current_recovery_params['primary_slot_name'][0] = wal_receiver_primary_slot_name
 
-        required = {'restart': 0, 'reload': 0}
+        # Increment the 'reload' to enforce write of postgresql.conf when joining the running postgres
+        required = {'restart': 0,
+                    'reload': int(not self._postgresql.cb_called and self._postgresql.major_version >= 120000)}
 
         def record_missmatch(mtype):
             required['restart' if mtype else 'reload'] += 1
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/rewind.py b/patroni/postgresql/rewind.py
index 519b0b24..03784428 100644
--- a/patroni/postgresql/rewind.py
+++ b/patroni/postgresql/rewind.py
@@ -72,19 +72,10 @@ class Rewind(object):
         `pg_waldump: fatal: error in WAL record at 0/182E220: invalid record length at 0/182E298: wanted 24, got 0`
         The error message contains information about LSN of the next record, which is exactly where checkpoint ends."""
 
-        cmd = self._postgresql.pgcommand('pg_{0}dump'.format(self._postgresql.wal_name))
         lsn8 = format_lsn(lsn, True)
         lsn = format_lsn(lsn)
-        env = os.environ.copy()
-        env.update(LANG='C', LC_ALL='C', PGDATA=self._postgresql.data_dir)
-        try:
-            waldump = subprocess.Popen([cmd, '-t', str(timeline), '-s', lsn, '-n', '2'],
-                                       stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env)
-            out, err = waldump.communicate()
-            waldump.wait()
-        except Exception as e:
-            logger.error('Failed to execute `%s -t %s -s %s -n 2`: %r', cmd, timeline, lsn, e)
-        else:
+        out, err = self._postgresql.waldump(timeline, lsn, 2)
+        if out is not None and err is not None:
             out = out.decode('utf-8').rstrip().split('\n')
             err = err.decode('utf-8').rstrip().split('\n')
             pattern = 'error in WAL record at {0}: invalid record length at '.format(lsn)
@@ -97,7 +88,7 @@ class Rewind(object):
                         return parse_lsn(err[0][i:j])
                     except Exception as e:
                         logger.error('Failed to parse lsn %s: %r', err[0][i:j], e)
-            logger.error('Failed to parse `%s -t %s -s %s -n 2` output', cmd, timeline, lsn)
+            logger.error('Failed to parse pg_%sdump output', self._postgresql.wal_name)
             logger.error(' stdout=%s', '\n'.join(out))
             logger.error(' stderr=%s', '\n'.join(err))
 
@@ -189,7 +180,9 @@ class Rewind(object):
                     need_rewind = False
                 elif master_timeline > 1:
                     cur.execute('TIMELINE_HISTORY %s', (master_timeline,))
-                    history = bytes(cur.fetchone()[1]).decode('utf-8')
+                    history = cur.fetchone()[1]
+                    if not isinstance(history, six.string_types):
+                        history = bytes(history).decode('utf-8')
                     logger.debug('master: history=%s', history)
         except Exception:
             return logger.exception('Exception when working with master via replication connection')
diff --git a/patroni/postgresql/validator.py b/patroni/postgresql/validator.py
index 93cdce54..945d5afd 100644
--- a/patroni/postgresql/validator.py
+++ b/patroni/postgresql/validator.py
@@ -108,7 +108,7 @@ parameters = CaseInsensitiveDict({
     ),
     'archive_timeout': Integer(90300, None, 0, 1073741823, 's'),
     'array_nulls': Bool(90300, None),
-    'authentication_timeout': Integer(90300, None,  1, 600, 's'),
+    'authentication_timeout': Integer(90300, None, 1, 600, 's'),
     'autovacuum': Bool(90300, None),
     'autovacuum_analyze_scale_factor': Real(90300, None, 0, 100, None),
     'autovacuum_analyze_threshold': Integer(90300, None, 0, 2147483647, None),
@@ -151,12 +151,14 @@ parameters = CaseInsensitiveDict({
         Integer(90600, None, 30, 86400, 's')
     ),
     'checkpoint_warning': Integer(90300, None, 0, 2147483647, 's'),
+    'client_connection_check_interval': Integer(140000, None, '0', '2147483647', 'ms'),
     'client_encoding': String(90300, None),
     'client_min_messages': Enum(90300, None, ('debug5', 'debug4', 'debug3', 'debug2',
                                               'debug1', 'log', 'notice', 'warning', 'error')),
     'cluster_name': String(90500, None),
     'commit_delay': Integer(90300, None, 0, 100000, None),
     'commit_siblings': Integer(90300, None, 0, 1000, None),
+    'compute_query_id': EnumBool(140000, None, ('auto',)),
     'config_file': String(90300, None),
     'constraint_exclusion': EnumBool(90300, None, ('partition',)),
     'cpu_index_tuple_cost': Real(90300, None, 0, 1.79769e+308, None),
@@ -168,6 +170,7 @@ parameters = CaseInsensitiveDict({
     'DateStyle': String(90300, None),
     'db_user_namespace': Bool(90300, None),
     'deadlock_timeout': Integer(90300, None, 1, 2147483647, 'ms'),
+    'debug_invalidate_system_caches_always': Integer(140000, None, '0', '0', None),
     'debug_pretty_print': Bool(90300, None),
     'debug_print_parse': Bool(90300, None),
     'debug_print_plan': Bool(90300, None),
@@ -176,6 +179,7 @@ parameters = CaseInsensitiveDict({
     'default_table_access_method': String(120000, None),
     'default_tablespace': String(90300, None),
     'default_text_search_config': String(90300, None),
+    'default_toast_compression': Enum(140000, None, ('pglz', 'lz4')),
     'default_transaction_deferrable': Bool(90300, None),
     'default_transaction_isolation': Enum(90300, None, ('serializable', 'repeatable read',
                                                         'read committed', 'read uncommitted')),
@@ -188,6 +192,7 @@ parameters = CaseInsensitiveDict({
     ),
     'effective_cache_size': Integer(90300, None, 1, 2147483647, '8kB'),
     'effective_io_concurrency': Integer(90300, None, 0, 1000, None),
+    'enable_async_append': Bool(140000, None),
     'enable_bitmapscan': Bool(90300, None),
     'enable_gathermerge': Bool(100000, None),
     'enable_hashagg': Bool(90300, None),
@@ -203,12 +208,14 @@ parameters = CaseInsensitiveDict({
     'enable_partition_pruning': Bool(110000, None),
     'enable_partitionwise_aggregate': Bool(110000, None),
     'enable_partitionwise_join': Bool(110000, None),
+    'enable_resultcache': Bool(140000, None),
     'enable_seqscan': Bool(90300, None),
     'enable_sort': Bool(90300, None),
     'enable_tidscan': Bool(90300, None),
     'escape_string_warning': Bool(90300, None),
     'event_source': String(90300, None),
     'exit_on_error': Bool(90300, None),
+    'extension_destdir': String(140000, None),
     'external_pid_file': String(90300, None),
     'extra_float_digits': Integer(90300, None, -15, 3, None),
     'force_parallel_mode': EnumBool(90600, None, ('regress',)),
@@ -229,8 +236,10 @@ parameters = CaseInsensitiveDict({
     'hot_standby': Bool(90300, None),
     'hot_standby_feedback': Bool(90300, None),
     'huge_pages': EnumBool(90400, None, ('try',)),
+    'huge_page_size': Integer(140000, None, '0', '2147483647', 'kB'),
     'ident_file': String(90300, None),
     'idle_in_transaction_session_timeout': Integer(90600, None, 0, 2147483647, 'ms'),
+    'idle_session_timeout': Integer(140000, None, '0', '2147483647', 'ms'),
     'ignore_checksum_failure': Bool(90300, None),
     'ignore_invalid_pages': Bool(130000, None),
     'ignore_system_indexes': Bool(90300, None),
@@ -283,6 +292,7 @@ parameters = CaseInsensitiveDict({
     'log_parameter_max_length_on_error': Integer(130000, None, -1, 1073741823, 'B'),
     'log_parser_stats': Bool(90300, None),
     'log_planner_stats': Bool(90300, None),
+    'log_recovery_conflict_waits': Bool(140000, None),
     'log_replication_commands': Bool(90500, None),
     'log_rotation_age': Integer(90300, None, 0, 35791394, 'min'),
     'log_rotation_size': Integer(90300, None, 0, 2097151, 'kB'),
@@ -336,6 +346,7 @@ parameters = CaseInsensitiveDict({
         Integer(90400, 90600, 1, 8388607, None),
         Integer(90600, None, 0, 262143, None)
     ),
+    'min_dynamic_shared_memory': Integer(140000, None, '0', '2147483647', 'MB'),
     'min_parallel_index_scan_size': Integer(100000, None, 0, 715827882, '8kB'),
     'min_parallel_relation_size': Integer(90600, 100000, 0, 715827882, '8kB'),
     'min_parallel_table_scan_size': Integer(100000, None, 0, 715827882, '8kB'),
@@ -344,7 +355,7 @@ parameters = CaseInsensitiveDict({
         Integer(100000, None, 2, 2147483647, 'MB')
     ),
     'old_snapshot_threshold': Integer(90600, None, -1, 86400, 'min'),
-    'operator_precedence_warning': Bool(90500, None),
+    'operator_precedence_warning': Bool(90500, 140000),
     'parallel_leader_participation': Bool(110000, None),
     'parallel_setup_cost': Real(90600, None, 0, 1.79769e+308, None),
     'parallel_tuple_cost': Real(90600, None, 0, 1.79769e+308, None),
@@ -358,6 +369,8 @@ parameters = CaseInsensitiveDict({
     'pre_auth_delay': Integer(90300, None, 0, 60, 's'),
     'quote_all_identifiers': Bool(90300, None),
     'random_page_cost': Real(90300, None, 0, 1.79769e+308, None),
+    'recovery_init_sync_method': Enum(140000, None, ('fsync', 'syncfs')),
+    'remove_temp_files_after_crash': Bool(140000, None),
     'replacement_sort_tuples': Integer(90600, 110000, 0, 2147483647, None),
     'restart_after_crash': Bool(90300, None),
     'row_security': Bool(90500, None),
@@ -373,6 +386,7 @@ parameters = CaseInsensitiveDict({
     'ssl_ca_file': String(90300, None),
     'ssl_cert_file': String(90300, None),
     'ssl_ciphers': String(90300, None),
+    'ssl_crl_dir': String(140000, None),
     'ssl_crl_file': String(90300, None),
     'ssl_dh_params_file': String(100000, None),
     'ssl_ecdh_curve': String(90400, None),
@@ -388,7 +402,7 @@ parameters = CaseInsensitiveDict({
     'stats_temp_directory': String(90300, None),
     'superuser_reserved_connections': (
         Integer(90300, 90600, 0, 8388607, None),
-        Integer(90600, None, 0, 262143, None),
+        Integer(90600, None, 0, 262143, None)
     ),
     'synchronize_seqscans': Bool(90300, None),
     'synchronous_commit': (
@@ -424,6 +438,7 @@ parameters = CaseInsensitiveDict({
     'track_counts': Bool(90300, None),
     'track_functions': Enum(90300, None, ('none', 'pl', 'all')),
     'track_io_timing': Bool(90300, None),
+    'track_wal_io_timing': Bool(140000, None),
     'transaction_deferrable': Bool(90300, None),
     'transaction_isolation': Enum(90300, None, ('serializable', 'repeatable read',
                                                 'read committed', 'read uncommitted')),
@@ -433,7 +448,7 @@ parameters = CaseInsensitiveDict({
     'unix_socket_group': String(90300, None),
     'unix_socket_permissions': Integer(90300, None, 0, 511, None),
     'update_process_title': Bool(90300, None),
-    'vacuum_cleanup_index_scale_factor': Real(110000, None, 0, 1e+10, None),
+    'vacuum_cleanup_index_scale_factor': Real(110000, 140000, 0, 1e+10, None),
     'vacuum_cost_delay': (
         Integer(90300, 120000, 0, 100, 'ms'),
         Real(120000, None, 0, 100, 'ms')
@@ -443,8 +458,10 @@ parameters = CaseInsensitiveDict({
     'vacuum_cost_page_hit': Integer(90300, None, 0, 10000, None),
     'vacuum_cost_page_miss': Integer(90300, None, 0, 10000, None),
     'vacuum_defer_cleanup_age': Integer(90300, None, 0, 1000000, None),
+    'vacuum_failsafe_age': Integer(140000, None, '0', '2100000000', None),
     'vacuum_freeze_min_age': Integer(90300, None, 0, 1000000000, None),
     'vacuum_freeze_table_age': Integer(90300, None, 0, 2000000000, None),
+    'vacuum_multixact_failsafe_age': Integer(140000, None, '0', '2100000000', None),
     'vacuum_multixact_freeze_min_age': Integer(90300, None, 0, 1000000000, None),
     'vacuum_multixact_freeze_table_age': Integer(90300, None, 0, 2000000000, None),
     'wal_buffers': Integer(90300, None, -1, 262143, '8kB'),


Test patch:
diff --git a/.github/workflows/run_tests.py b/.github/workflows/run_tests.py
index fbda3999..89a9b222 100644
--- a/.github/workflows/run_tests.py
+++ b/.github/workflows/run_tests.py
@@ -23,7 +23,9 @@ def main():
 
     env = os.environ.copy()
     if sys.platform.startswith('linux'):
-        version = {'etcd': '9.6', 'etcd3': '13', 'consul': 12, 'exhibitor': 11, 'kubernetes': 13, 'raft': 12}.get(what)
+        from mapping import versions
+
+        version = versions.get(what)
         path = '/usr/lib/postgresql/{0}/bin:.'.format(version)
         unbuffer = ['timeout', '600', 'unbuffer']
         args = ['--tags=-skip'] if what == 'etcd' else []
diff --git a/tests/test_api.py b/tests/test_api.py
index 079753f4..4b4d8ebf 100644
--- a/tests/test_api.py
+++ b/tests/test_api.py
@@ -118,7 +118,7 @@ class MockPatroni(object):
     postgresql = ha.state_handler
     dcs = Mock()
     logger = MockLogger()
-    tags = {}
+    tags = {"key1": True, "key2": False, "key3": 1, "key4": 1.4, "key5": "RandomTag"}
     version = '0.00'
     noloadbalance = PropertyMock(return_value=False)
     scheduled_restart = {'schedule': future_restart_time,
@@ -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'})):
@@ -197,6 +197,89 @@ class TestRestApiHandler(unittest.TestCase):
         with patch.object(MockHa, 'is_standby_cluster', Mock(return_value=True)):
             MockRestApiServer(RestApiHandler, 'GET /standby_leader')
 
+        # test tags
+        #
+        MockRestApiServer(RestApiHandler, 'GET /master?lag=1M&'
+                                          'tag_key1=true&tag_key2=false&'
+                                          'tag_key3=1&tag_key4=1.4&tag_key5=RandomTag')
+        MockRestApiServer(RestApiHandler, 'GET /master?lag=1M&'
+                                          'tag_key1=true&tag_key2=False&'
+                                          'tag_key3=1&tag_key4=1.4&tag_key5=RandomTag')
+        MockRestApiServer(RestApiHandler, 'GET /master?lag=1M&'
+                                          'tag_key1=true&tag_key2=false&'
+                                          'tag_key3=1.0&tag_key4=1.4&tag_key5=RandomTag')
+        MockRestApiServer(RestApiHandler, 'GET /master?lag=1M&'
+                                          'tag_key1=true&tag_key2=false&'
+                                          'tag_key3=1&tag_key4=1.4&tag_key5=RandomTag&tag_key6=RandomTag2')
+        #
+        with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={'role': 'master'})):
+            MockRestApiServer(RestApiHandler, 'GET /master?lag=1M&'
+                                              'tag_key1=true&tag_key2=false&'
+                                              'tag_key3=1&tag_key4=1.4&tag_key5=RandomTag')
+            MockRestApiServer(RestApiHandler, 'GET /master?lag=1M&'
+                                              'tag_key1=true&tag_key2=False&'
+                                              'tag_key3=1&tag_key4=1.4&tag_key5=RandomTag')
+            MockRestApiServer(RestApiHandler, 'GET /master?lag=1M&'
+                                              'tag_key1=true&tag_key2=false&'
+                                              'tag_key3=1.0&tag_key4=1.4&tag_key5=RandomTag')
+            MockRestApiServer(RestApiHandler, 'GET /master?lag=1M&'
+                                              'tag_key1=true&tag_key2=false&'
+                                              'tag_key3=1&tag_key4=1.4&tag_key5=RandomTag&tag_key6=RandomTag2')
+        #
+        with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={'role': 'standby_leader'})):
+            MockRestApiServer(RestApiHandler, 'GET /standby_leader?lag=1M&'
+                                              'tag_key1=true&tag_key2=false&'
+                                              'tag_key3=1&tag_key4=1.4&tag_key5=RandomTag')
+            MockRestApiServer(RestApiHandler, 'GET /standby_leader?lag=1M&'
+                                              'tag_key1=true&tag_key2=False&'
+                                              'tag_key3=1&tag_key4=1.4&tag_key5=RandomTag')
+            MockRestApiServer(RestApiHandler, 'GET /standby_leader?lag=1M&'
+                                              'tag_key1=true&tag_key2=false&'
+                                              'tag_key3=1.0&tag_key4=1.4&tag_key5=RandomTag')
+            MockRestApiServer(RestApiHandler, 'GET /standby_leader?lag=1M&'
+                                              'tag_key1=true&tag_key2=false&'
+                                              'tag_key3=1&tag_key4=1.4&tag_key5=RandomTag&tag_key6=RandomTag2')
+        #
+        MockRestApiServer(RestApiHandler, 'GET /replica?lag=1M&'
+                                          'tag_key1=true&tag_key2=false&'
+                                          'tag_key3=1&tag_key4=1.4&tag_key5=RandomTag')
+        MockRestApiServer(RestApiHandler, 'GET /replica?lag=1M&'
+                                          'tag_key1=true&tag_key2=False&'
+                                          'tag_key3=1&tag_key4=1.4&tag_key5=RandomTag')
+        MockRestApiServer(RestApiHandler, 'GET /replica?lag=1M&'
+                                          'tag_key1=true&tag_key2=false&'
+                                          'tag_key3=1.0&tag_key4=1.4&tag_key5=RandomTag')
+        MockRestApiServer(RestApiHandler, 'GET /replica?lag=1M&'
+                                          'tag_key1=true&tag_key2=false&'
+                                          'tag_key3=1&tag_key4=1.4&tag_key5=RandomTag&tag_key6=RandomTag2')
+        #
+        with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={'role': 'master'})):
+            MockRestApiServer(RestApiHandler, 'GET /replica?lag=1M&'
+                                              'tag_key1=true&tag_key2=false&'
+                                              'tag_key3=1&tag_key4=1.4&tag_key5=RandomTag')
+            MockRestApiServer(RestApiHandler, 'GET /replica?lag=1M&'
+                                              'tag_key1=true&tag_key2=False&'
+                                              'tag_key3=1&tag_key4=1.4&tag_key5=RandomTag')
+            MockRestApiServer(RestApiHandler, 'GET /replica?lag=1M&'
+                                              'tag_key1=true&tag_key2=false&'
+                                              'tag_key3=1.0&tag_key4=1.4&tag_key5=RandomTag')
+            MockRestApiServer(RestApiHandler, 'GET /replica?lag=1M&'
+                                              'tag_key1=true&tag_key2=false&'
+                                              'tag_key3=1&tag_key4=1.4&tag_key5=RandomTag&tag_key6=RandomTag2')
+        #
+        MockRestApiServer(RestApiHandler, 'GET /read-write?lag=1M&'
+                                          'tag_key1=true&tag_key2=false&'
+                                          'tag_key3=1&tag_key4=1.4&tag_key5=RandomTag')
+        MockRestApiServer(RestApiHandler, 'GET /read-write?lag=1M&'
+                                          'tag_key1=true&tag_key2=False&'
+                                          'tag_key3=1&tag_key4=1.4&tag_key5=RandomTag')
+        MockRestApiServer(RestApiHandler, 'GET /read-write?lag=1M&'
+                                          'tag_key1=true&tag_key2=false&'
+                                          'tag_key3=1.0&tag_key4=1.4&tag_key5=RandomTag')
+        MockRestApiServer(RestApiHandler, 'GET /read-write?lag=1M&'
+                                          'tag_key1=true&tag_key2=false&'
+                                          'tag_key3=1&tag_key4=1.4&tag_key5=RandomTag&tag_key6=RandomTag2')
+
     def test_do_OPTIONS(self):
         self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'OPTIONS / HTTP/1.0'))
 
@@ -507,3 +590,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_etcd3.py b/tests/test_etcd3.py
index abc593af..b0ae1d05 100644
--- a/tests/test_etcd3.py
+++ b/tests/test_etcd3.py
@@ -4,8 +4,9 @@ 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, AuthFailed, base64_encode
+from patroni.dcs.etcd import DnsCachingResolver
+from patroni.dcs.etcd3 import PatroniEtcd3Client, Cluster, Etcd3Client, Etcd3Error, Etcd3ClientError, RetryFailedError,\
+        InvalidAuthToken, Unavailable, Unknown, UnsupportedEtcdVersion, UserEmpty, AuthFailed, base64_encode, Etcd3
 from threading import Thread
 
 from . import SleepException, MockResponse
@@ -57,6 +58,16 @@ def mock_urlopen(self, method, url, **kwargs):
     return ret
 
 
+class TestEtcd3Client(unittest.TestCase):
+
+    @patch.object(Thread, 'start', Mock())
+    @patch.object(urllib3.PoolManager, 'urlopen', mock_urlopen)
+    def test_authenticate(self):
+        etcd3 = Etcd3Client({'host': '127.0.0.1', 'port': 2379, 'use_proxies': True, 'retry_timeout': 10},
+                            DnsCachingResolver())
+        self.assertIsNotNone(etcd3._cluster_version)
+
+
 class BaseTestEtcd3(unittest.TestCase):
 
     @patch.object(Thread, 'start', Mock())
diff --git a/tests/test_exhibitor.py b/tests/test_exhibitor.py
index 9d4d460d..fdd3e993 100644
--- a/tests/test_exhibitor.py
+++ b/tests/test_exhibitor.py
@@ -24,7 +24,7 @@ class TestExhibitor(unittest.TestCase):
 
     @patch('urllib3.PoolManager.request', Mock(return_value=urllib3.HTTPResponse(
         status=200, body=b'{"servers":["127.0.0.1","127.0.0.2","127.0.0.3"],"port":2181}')))
-    @patch('patroni.dcs.zookeeper.KazooClient', MockKazooClient)
+    @patch('patroni.dcs.zookeeper.PatroniKazooClient', MockKazooClient)
     def setUp(self):
         self.e = Exhibitor({'hosts': ['localhost', 'exhibitor'], 'port': 8181, 'scope': 'test',
                             'name': 'foo', 'ttl': 30, 'retry_timeout': 10})
diff --git a/tests/test_ha.py b/tests/test_ha.py
index ed534b27..ac520c9c 100644
--- a/tests/test_ha.py
+++ b/tests/test_ha.py
@@ -157,7 +157,8 @@ def run_async(self, func, args=()):
 @patch.object(Postgresql, 'controldata', Mock(return_value={
     'Database system identifier': SYSID,
     'Database cluster state': 'shut down',
-    'Latest checkpoint location': '0/12345678'}))
+    'Latest checkpoint location': '0/12345678',
+    "Latest checkpoint's TimeLineID": '2'}))
 @patch.object(SlotsHandler, 'load_replication_slots', Mock(side_effect=Exception))
 @patch.object(ConfigHandler, 'append_pg_hba', Mock())
 @patch.object(ConfigHandler, 'write_pgpass', Mock(return_value={}))
@@ -250,6 +251,15 @@ class TestHa(PostgresInit):
         self.assertEqual(self.ha.run_cycle(), 'starting as a secondary')
         self.assertEqual(self.ha.run_cycle(), 'failed to start postgres')
 
+    def test_recover_raft(self):
+        self.p.controldata = lambda: {'Database cluster state': 'in recovery', 'Database system identifier': SYSID}
+        self.p.is_running = false
+        self.p.follow = true
+        self.assertEqual(self.ha.run_cycle(), 'starting as a secondary')
+        self.p.is_running = true
+        self.ha.dcs.__class__.__name__ = 'Raft'
+        self.assertEqual(self.ha.run_cycle(), 'started as a secondary')
+
     def test_recover_former_master(self):
         self.p.follow = false
         self.p.is_running = false
@@ -278,7 +288,17 @@ class TestHa(PostgresInit):
     def test_recover_with_rewind(self):
         self.p.is_running = false
         self.ha.cluster = get_cluster_initialized_with_leader()
-        self.assertEqual(self.ha.run_cycle(), 'running pg_rewind from leader')
+        self.ha.cluster.leader.member.data.update(version='2.0.2', role='master')
+        self.ha._rewind.pg_rewind = true
+        self.ha._rewind.check_leader_is_not_in_recovery = true
+        with patch.object(Rewind, 'rewind_or_reinitialize_needed_and_possible', Mock(return_value=True)):
+            self.assertEqual(self.ha.run_cycle(), 'running pg_rewind from leader')
+        with patch.object(Rewind, 'rewind_or_reinitialize_needed_and_possible', Mock(return_value=False)):
+            self.p.follow = true
+            self.assertEqual(self.ha.run_cycle(), 'starting as a secondary')
+            self.p.is_running = true
+            self.ha.follow = Mock(return_value='fake')
+            self.assertEqual(self.ha.run_cycle(), 'fake')
 
     @patch.object(Rewind, 'rewind_or_reinitialize_needed_and_possible', Mock(return_value=True))
     @patch.object(Bootstrap, 'create_replica', Mock(return_value=1))
@@ -300,7 +320,7 @@ class TestHa(PostgresInit):
         self.p.is_healthy = true
         self.ha.has_lock = true
         self.p.controldata = lambda: {'Database cluster state': 'in production', 'Database system identifier': SYSID}
-        self.assertEqual(self.ha.run_cycle(), 'promoted self to leader because i had the session lock')
+        self.assertEqual(self.ha.run_cycle(), 'promoted self to leader because I had the session lock')
 
     @patch('psycopg2.connect', psycopg2_connect)
     def test_acquire_lock_as_master(self):
@@ -329,7 +349,7 @@ class TestHa(PostgresInit):
         self.ha.has_lock = true
         self.p.is_leader = false
         self.p.set_role('master')
-        self.assertEqual(self.ha.run_cycle(), 'no action.  i am the leader with the lock')
+        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0) the leader with the lock')
 
     def test_demote_after_failing_to_obtain_lock(self):
         self.ha.acquire_lock = false
@@ -354,7 +374,7 @@ class TestHa(PostgresInit):
         self.ha.cluster.is_unlocked = false
         self.ha.has_lock = true
         self.p.is_leader = false
-        self.assertEqual(self.ha.run_cycle(), 'promoted self to leader because i had the session lock')
+        self.assertEqual(self.ha.run_cycle(), 'promoted self to leader because I had the session lock')
 
     def test_promote_without_watchdog(self):
         self.ha.cluster.is_unlocked = false
@@ -369,12 +389,12 @@ class TestHa(PostgresInit):
         self.ha.cluster = get_cluster_initialized_with_leader()
         self.ha.cluster.is_unlocked = false
         self.ha.has_lock = true
-        self.assertEqual(self.ha.run_cycle(), 'no action.  i am the leader with the lock')
+        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0) the leader with the lock')
 
     def test_demote_because_not_having_lock(self):
         self.ha.cluster.is_unlocked = false
         with patch.object(Watchdog, 'is_running', PropertyMock(return_value=True)):
-            self.assertEqual(self.ha.run_cycle(), 'demoting self because i do not have the lock and i was a leader')
+            self.assertEqual(self.ha.run_cycle(), 'demoting self because I do not have the lock and I was a leader')
 
     def test_demote_because_update_lock_failed(self):
         self.ha.cluster.is_unlocked = false
@@ -388,16 +408,16 @@ class TestHa(PostgresInit):
     def test_follow(self):
         self.ha.cluster.is_unlocked = false
         self.p.is_leader = false
-        self.assertEqual(self.ha.run_cycle(), 'no action.  i am a secondary and i am following a leader')
+        self.assertEqual(self.ha.run_cycle(), 'no action. I am a secondary (postgresql0) and following a leader ()')
         self.ha.patroni.replicatefrom = "foo"
         self.p.config.check_recovery_conf = Mock(return_value=(True, False))
         self.ha.cluster.config.data.update({'slots': {'l': {'database': 'a', 'plugin': 'b'}}})
         self.ha.cluster.members[1].data['tags']['replicatefrom'] = 'postgresql0'
         self.ha.patroni.nofailover = True
-        self.assertEqual(self.ha.run_cycle(), 'no action.  i am a secondary and i am following a leader')
+        self.assertEqual(self.ha.run_cycle(), 'no action. I am a secondary (postgresql0) and following a leader ()')
         del self.ha.cluster.config.data['slots']
         self.ha.cluster.config.data.update({'postgresql': {'use_slots': False}})
-        self.assertEqual(self.ha.run_cycle(), 'no action.  i am a secondary and i am following a leader')
+        self.assertEqual(self.ha.run_cycle(), 'no action. I am a secondary (postgresql0) and following a leader ()')
         del self.ha.cluster.config.data['postgresql']['use_slots']
 
     def test_follow_in_pause(self):
@@ -405,7 +425,7 @@ class TestHa(PostgresInit):
         self.ha.is_paused = true
         self.assertEqual(self.ha.run_cycle(), 'PAUSE: continue to run as master without lock')
         self.p.is_leader = false
-        self.assertEqual(self.ha.run_cycle(), 'PAUSE: no action')
+        self.assertEqual(self.ha.run_cycle(), 'PAUSE: no action. I am (postgresql0)')
 
     @patch.object(Rewind, 'rewind_or_reinitialize_needed_and_possible', Mock(return_value=True))
     @patch.object(Rewind, 'can_rewind', PropertyMock(return_value=True))
@@ -516,27 +536,27 @@ class TestHa(PostgresInit):
         self.ha.fetch_node_status = get_node_status()
         self.ha.has_lock = true
         self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', '', None))
-        self.assertEqual(self.ha.run_cycle(), 'no action.  i am the leader with the lock')
+        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0) the leader with the lock')
         self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, '', self.p.name, None))
-        self.assertEqual(self.ha.run_cycle(), 'no action.  i am the leader with the lock')
+        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0) the leader with the lock')
         self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, '', 'blabla', None))
-        self.assertEqual(self.ha.run_cycle(), 'no action.  i am the leader with the lock')
+        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0) the leader with the lock')
         f = Failover(0, self.p.name, '', None)
         self.ha.cluster = get_cluster_initialized_with_leader(f)
         self.assertEqual(self.ha.run_cycle(), 'manual failover: demoting myself')
         self.ha._rewind.rewind_or_reinitialize_needed_and_possible = true
         self.assertEqual(self.ha.run_cycle(), 'manual failover: demoting myself')
         self.ha.fetch_node_status = get_node_status(nofailover=True)
-        self.assertEqual(self.ha.run_cycle(), 'no action.  i am the leader with the lock')
+        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0) the leader with the lock')
         self.ha.fetch_node_status = get_node_status(watchdog_failed=True)
-        self.assertEqual(self.ha.run_cycle(), 'no action.  i am the leader with the lock')
+        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0) the leader with the lock')
         self.ha.fetch_node_status = get_node_status(timeline=1)
-        self.assertEqual(self.ha.run_cycle(), 'no action.  i am the leader with the lock')
+        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0) the leader with the lock')
         self.ha.fetch_node_status = get_node_status(wal_position=1)
-        self.assertEqual(self.ha.run_cycle(), 'no action.  i am the leader with the lock')
+        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0) the leader with the lock')
         # manual failover from the previous leader to us won't happen if we hold the nofailover flag
         self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', self.p.name, None))
-        self.assertEqual(self.ha.run_cycle(), 'no action.  i am the leader with the lock')
+        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0) the leader with the lock')
 
         # Failover scheduled time must include timezone
         scheduled = datetime.datetime.now()
@@ -545,28 +565,28 @@ class TestHa(PostgresInit):
 
         scheduled = datetime.datetime.utcnow().replace(tzinfo=tzutc)
         self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', self.p.name, scheduled))
-        self.assertEqual('no action.  i am the leader with the lock', self.ha.run_cycle())
+        self.assertEqual('no action. I am (postgresql0) the leader with the lock', self.ha.run_cycle())
 
         scheduled = scheduled + datetime.timedelta(seconds=30)
         self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', self.p.name, scheduled))
-        self.assertEqual('no action.  i am the leader with the lock', self.ha.run_cycle())
+        self.assertEqual('no action. I am (postgresql0) the leader with the lock', self.ha.run_cycle())
 
         scheduled = scheduled + datetime.timedelta(seconds=-600)
         self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', self.p.name, scheduled))
-        self.assertEqual('no action.  i am the leader with the lock', self.ha.run_cycle())
+        self.assertEqual('no action. I am (postgresql0) the leader with the lock', self.ha.run_cycle())
 
         scheduled = None
         self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', self.p.name, scheduled))
-        self.assertEqual('no action.  i am the leader with the lock', self.ha.run_cycle())
+        self.assertEqual('no action. I am (postgresql0) the leader with the lock', self.ha.run_cycle())
 
     def test_manual_failover_from_leader_in_pause(self):
         self.ha.has_lock = true
         self.ha.is_paused = true
         scheduled = datetime.datetime.now()
         self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', self.p.name, scheduled))
-        self.assertEqual('PAUSE: no action.  i am the leader with the lock', self.ha.run_cycle())
+        self.assertEqual('PAUSE: no action. I am (postgresql0) the leader with the lock', self.ha.run_cycle())
         self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, self.p.name, '', None))
-        self.assertEqual('PAUSE: no action.  i am the leader with the lock', self.ha.run_cycle())
+        self.assertEqual('PAUSE: no action. I am (postgresql0) the leader with the lock', self.ha.run_cycle())
 
     def test_manual_failover_from_leader_in_synchronous_mode(self):
         self.p.is_leader = true
@@ -575,7 +595,7 @@ class TestHa(PostgresInit):
         self.ha.is_failover_possible = false
         self.ha.process_sync_replication = Mock()
         self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, self.p.name, 'a', None), (self.p.name, None))
-        self.assertEqual('no action.  i am the leader with the lock', self.ha.run_cycle())
+        self.assertEqual('no action. I am (postgresql0) the leader with the lock', self.ha.run_cycle())
         self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, self.p.name, 'a', None), (self.p.name, 'a'))
         self.ha.is_failover_possible = true
         self.assertEqual('manual failover: demoting myself', self.ha.run_cycle())
@@ -736,7 +756,7 @@ class TestHa(PostgresInit):
         self.p.config.check_recovery_conf = Mock(return_value=(False, False))
         self.ha._leader_timeline = 1
         self.assertEqual(self.ha.run_cycle(), 'promoted self to a standby leader because i had the session lock')
-        self.assertEqual(self.ha.run_cycle(), 'no action.  i am the standby leader with the lock')
+        self.assertEqual(self.ha.run_cycle(), 'no action. I am (leader) the standby leader with the lock')
         self.p.set_role('replica')
         self.p.config.check_recovery_conf = Mock(return_value=(True, False))
         self.assertEqual(self.ha.run_cycle(), 'promoted self to a standby leader because i had the session lock')
@@ -745,7 +765,8 @@ class TestHa(PostgresInit):
         self.p.is_leader = false
         self.p.name = 'replica'
         self.ha.cluster = get_standby_cluster_initialized_with_only_leader()
-        self.assertEqual(self.ha.run_cycle(), 'no action.  i am a secondary and i am following a standby leader')
+        self.assertEqual(self.ha.run_cycle(),
+                         'no action. I am a secondary (replica) and following a standby leader (leader)')
         with patch.object(Leader, 'conn_url', PropertyMock(return_value='')):
             self.assertEqual(self.ha.run_cycle(), 'continue following the old known standby leader')
 
@@ -838,7 +859,8 @@ class TestHa(PostgresInit):
 
         self.ha.has_lock = false
         self.p.is_leader = false
-        self.assertEqual(self.ha.run_cycle(), 'no action.  i am a secondary and i am following a leader')
+        self.assertEqual(self.ha.run_cycle(),
+                         'no action. I am a secondary (postgresql0) and following a leader (leader)')
         check_calls([(update_lock, False), (demote, False)])
 
     def test_manual_failover_while_starting(self):
@@ -1063,7 +1085,7 @@ class TestHa(PostgresInit):
         self.ha.cluster.config.data.clear()
         self.ha.has_lock = true
         self.ha.cluster.is_unlocked = false
-        self.assertEqual(self.ha.run_cycle(), 'no action.  i am the leader with the lock')
+        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0) the leader with the lock')
 
     def test_watch(self):
         self.ha.cluster = get_cluster_initialized_with_leader()
@@ -1098,7 +1120,7 @@ class TestHa(PostgresInit):
         self.ha.cluster.is_unlocked = false
         for tl in (1, 3):
             self.p.get_master_timeline = Mock(return_value=tl)
-            self.assertEqual(self.ha.run_cycle(), 'no action.  i am the leader with the lock')
+            self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0) the leader with the lock')
 
     @patch('sys.exit', return_value=1)
     def test_abort_join(self, exit_mock):
@@ -1111,9 +1133,9 @@ class TestHa(PostgresInit):
         self.ha.has_lock = true
         self.ha.cluster.is_unlocked = false
         self.ha.is_paused = true
-        self.assertEqual(self.ha.run_cycle(), 'PAUSE: no action.  i am the leader with the lock')
+        self.assertEqual(self.ha.run_cycle(), 'PAUSE: no action. I am (postgresql0) the leader with the lock')
         self.ha.is_paused = false
-        self.assertEqual(self.ha.run_cycle(), 'no action.  i am the leader with the lock')
+        self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0) the leader with the lock')
 
     @patch('psycopg2.connect', psycopg2_connect)
     def test_permanent_logical_slots_after_promote(self):
@@ -1123,7 +1145,7 @@ class TestHa(PostgresInit):
         self.assertEqual(self.ha.run_cycle(), 'acquired session lock as a leader')
         self.ha.cluster = get_cluster_initialized_without_leader(leader=True, cluster_config=config)
         self.ha.has_lock = true
-        self.assertEqual(self.ha.run_cycle(), 'no action.  i am the leader with the lock')
+        self.assertEqual(self.ha.run_cycle(), 'no action. I am (other) the leader with the lock')
 
     @patch.object(Cluster, 'has_member', true)
     def test_run_cycle(self):
diff --git a/tests/test_log.py b/tests/test_log.py
index 96dc0a3f..e5482c67 100644
--- a/tests/test_log.py
+++ b/tests/test_log.py
@@ -63,3 +63,12 @@ class TestPatroniLogger(unittest.TestCase):
             self.assertRaises(Exception, logger.shutdown)
         self.assertLessEqual(logger.queue_size, 2)  # "Failed to close the old log handler" could be still in the queue
         self.assertEqual(logger.records_lost, 0)
+
+    def test_interceptor(self):
+        logger = PatroniLogger()
+        logger.reload_config({'level': 'INFO'})
+        logger.start()
+        _LOG.info('Lock owner: ')
+        _LOG.info('blabla')
+        logger.shutdown()
+        self.assertEqual(logger.records_lost, 0)
diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py
index 42d01e6b..96f0b5bc 100644
--- a/tests/test_postgresql.py
+++ b/tests/test_postgresql.py
@@ -224,6 +224,7 @@ class TestPostgresql(BaseTestPostgresql):
     @patch('patroni.postgresql.config.mtime', mock_mtime)
     @patch('patroni.postgresql.config.ConfigHandler._get_pg_settings')
     def test_check_recovery_conf(self, mock_get_pg_settings):
+        self.p.call_nowait('on_start')
         mock_get_pg_settings.return_value = {
             'primary_conninfo': ['primary_conninfo', 'foo=', None, 'string', 'postmaster', self.p.config._auto_conf],
             'recovery_min_apply_delay': ['recovery_min_apply_delay', '0', 'ms', 'integer', 'sighup', 'foo']
@@ -259,6 +260,7 @@ class TestPostgresql(BaseTestPostgresql):
     @patch.object(MockPostmaster, 'create_time', Mock(return_value=1234567), create=True)
     @patch('patroni.postgresql.config.ConfigHandler._get_pg_settings')
     def test__read_recovery_params(self, mock_get_pg_settings):
+        self.p.call_nowait('on_start')
         mock_get_pg_settings.return_value = {'primary_conninfo': ['primary_conninfo', '', None, 'string',
                                                                   'postmaster', self.p.config._postgresql_conf]}
         self.p.config.write_recovery_conf({'standby_mode': 'on', 'primary_conninfo': {'password': 'foo'}})
@@ -320,10 +322,37 @@ class TestPostgresql(BaseTestPostgresql):
         with patch.object(Postgresql, '_query', Mock(side_effect=RetryFailedError(''))):
             self.assertRaises(PostgresConnectionException, self.p.is_leader)
 
-    @patch.object(Postgresql, 'controldata',
-                  Mock(return_value={'Database cluster state': 'shut down', 'Latest checkpoint location': 'X/678'}))
-    def test_latest_checkpoint_location(self):
-        self.assertIsNone(self.p.latest_checkpoint_location())
+    @patch.object(Postgresql, 'controldata', Mock(return_value={'Database cluster state': 'shut down',
+                                                                'Latest checkpoint location': '0/1ADBC18',
+                                                                "Latest checkpoint's TimeLineID": '1'}))
+    @patch('subprocess.Popen')
+    def test_latest_checkpoint_location(self, mock_popen):
+        mock_popen.return_value.communicate.return_value = (None, None)
+        self.assertEqual(self.p.latest_checkpoint_location(), '28163096')
+        # 9.3 and 9.4 format
+        mock_popen.return_value.communicate.side_effect = [
+            (b'rmgr: XLOG        len (rec/tot):     72/   104, tx:          0, lsn: 0/01ADBC18, prev 0/01ADBBB8, ' +
+             b'bkp: 0000, desc: checkpoint: redo 0/1ADBC18; tli 1; prev tli 1; fpw true; xid 0/727; oid 16386; multi' +
+             b' 1; offset 0; oldest xid 715 in DB 1; oldest multi 1 in DB 1; oldest running xid 0; shutdown', None),
+            (b'rmgr: Transaction len (rec/tot):     64/    96, tx:        726, lsn: 0/01ADBBB8, prev 0/01ADBB70, ' +
+             b'bkp: 0000, desc: commit: 2021-02-26 11:19:37.900918 CET; inval msgs: catcache 11 catcache 10', None)]
+        self.assertEqual(self.p.latest_checkpoint_location(), '28163096')
+        mock_popen.return_value.communicate.side_effect = [
+            (b'rmgr: XLOG        len (rec/tot):     72/   104, tx:          0, lsn: 0/01ADBC18, prev 0/01ADBBB8, ' +
+             b'bkp: 0000, desc: checkpoint: redo 0/1ADBC18; tli 1; prev tli 1; fpw true; xid 0/727; oid 16386; multi' +
+             b' 1; offset 0; oldest xid 715 in DB 1; oldest multi 1 in DB 1; oldest running xid 0; shutdown', None),
+            (b'rmgr: XLOG        len (rec/tot):      0/    32, tx:          0, lsn: 0/01ADBBB8, prev 0/01ADBBA0, ' +
+             b'bkp: 0000, desc: xlog switch ', None)]
+        self.assertEqual(self.p.latest_checkpoint_location(), '28163000')
+        # 9.5+ format
+        mock_popen.return_value.communicate.side_effect = [
+            (b'rmgr: XLOG        len (rec/tot):    114/   114, tx:          0, lsn: 0/01ADBC18, prev 0/018260F8, ' +
+             b'desc: CHECKPOINT_SHUTDOWN redo 0/1825ED8; tli 1; prev tli 1; fpw true; xid 0:494; oid 16387; multi 1' +
+             b'; offset 0; oldest xid 479 in DB 1; oldest multi 1 in DB 1; oldest/newest commit timestamp xid: 0/0;' +
+             b' oldest running xid 0; shutdown', None),
+            (b'rmgr: XLOG        len (rec/tot):     24/    24, tx:          0, lsn: 0/018260F8, prev 0/01826080, ' +
+             b'desc: SWITCH ', None)]
+        self.assertEqual(self.p.latest_checkpoint_location(), '25321720')
 
     def test_reload(self):
         self.assertTrue(self.p.reload())
@@ -725,3 +754,8 @@ class TestPostgresql(BaseTestPostgresql):
     @patch.object(Postgresql, 'is_running', Mock(return_value=True))
     def test_set_enforce_hot_standby_feedback(self):
         self.p.set_enforce_hot_standby_feedback(True)
+
+    @patch.object(Postgresql, 'major_version', PropertyMock(return_value=140000))
+    @patch.object(Postgresql, '_cluster_info_state_get', Mock(return_value=True))
+    def test_handle_parameter_change(self):
+        self.p.handle_parameter_change()
diff --git a/tests/test_raft.py b/tests/test_raft.py
index 716295da..a80a668d 100644
--- a/tests/test_raft.py
+++ b/tests/test_raft.py
@@ -4,7 +4,7 @@ import tempfile
 import time
 
 from mock import Mock, PropertyMock, patch
-from patroni.dcs.raft import DynMemberSyncObj, KVStoreTTL, Raft, SyncObjUtility
+from patroni.dcs.raft import DynMemberSyncObj, KVStoreTTL, Raft, SyncObjUtility, TCPTransport, _TCPTransport
 from pysyncobj import SyncObjConf, FAIL_REASON
 
 
@@ -23,6 +23,16 @@ def remove_files(prefix):
                     time.sleep(1.0)
 
 
+class TestTCPTransport(unittest.TestCase):
+
+    @patch.object(TCPTransport, '__init__', Mock())
+    @patch.object(TCPTransport, 'setOnUtilityMessageCallback', Mock())
+    @patch.object(TCPTransport, '_connectIfNecessarySingle', Mock(side_effect=Exception))
+    def test__connectIfNecessarySingle(self):
+        t = _TCPTransport(Mock(), None, [])
+        self.assertFalse(t._connectIfNecessarySingle(None))
+
+
 @patch('pysyncobj.tcp_server.TcpServer.bind', Mock())
 class TestDynMemberSyncObj(unittest.TestCase):
 
@@ -114,8 +124,7 @@ class TestRaft(unittest.TestCase):
     def test_raft(self):
         raft = Raft({'ttl': 30, 'scope': 'test', 'name': 'pg', 'self_addr': '127.0.0.1:1234',
                      'retry_timeout': 10, 'data_dir': self._TMP})
-        raft.set_retry_timeout(20)
-        raft.set_ttl(60)
+        raft.reload_config({'retry_timeout': 20, 'ttl': 60, 'loop_wait': 10})
         self.assertTrue(raft._sync_obj.set(raft.members_path + 'legacy', '{"version":"2.0.0"}'))
         self.assertTrue(raft.touch_member(''))
         self.assertTrue(raft.initialize())
diff --git a/tests/test_zookeeper.py b/tests/test_zookeeper.py
index ef368826..ed718a21 100644
--- a/tests/test_zookeeper.py
+++ b/tests/test_zookeeper.py
@@ -2,12 +2,13 @@ import select
 import six
 import unittest
 
-from kazoo.client import KazooState
+from kazoo.client import KazooClient, KazooState
 from kazoo.exceptions import NoNodeError, NodeExistsError
 from kazoo.handlers.threading import SequentialThreadingHandler
-from kazoo.protocol.states import ZnodeStat
+from kazoo.protocol.states import KeeperState, ZnodeStat
 from mock import Mock, patch
-from patroni.dcs.zookeeper import Leader, PatroniSequentialThreadingHandler, ZooKeeper, ZooKeeperError
+from patroni.dcs.zookeeper import Leader, PatroniKazooClient,\
+        PatroniSequentialThreadingHandler, ZooKeeper, ZooKeeperError
 
 
 class MockKazooClient(Mock):
@@ -128,9 +129,19 @@ class TestPatroniSequentialThreadingHandler(unittest.TestCase):
         self.assertRaises(select.error, self.handler.select)
 
 
+class TestPatroniKazooClient(unittest.TestCase):
+
+    def test__call(self):
+        c = PatroniKazooClient()
+        with patch.object(KazooClient, '_call', Mock()):
+            self.assertIsNotNone(c._call(None, Mock()))
+        c._state = KeeperState.CONNECTING
+        self.assertFalse(c._call(None, Mock()))
+
+
 class TestZooKeeper(unittest.TestCase):
 
-    @patch('patroni.dcs.zookeeper.KazooClient', MockKazooClient)
+    @patch('patroni.dcs.zookeeper.PatroniKazooClient', MockKazooClient)
     def setUp(self):
         self.zk = ZooKeeper({'hosts': ['localhost:2181'], 'scope': 'test',
                              'name': 'foo', 'ttl': 30, 'retry_timeout': 10, 'loop_wait': 10})