Repository: zalando/patroni
Pull request #1959 (2021-06-14 06:09:19)

Implemented allowlist for REST API
If configured, only IPs that matching rules would be allowed to call unsafe endpoints.
In addition to that, it is possible to automatically include IPs of members of the cluster to the list.
If neither of the above is configured the old behavior is retained.

Partially address https://github.com/zalando/patroni/issues/1734


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/ENVIRONMENT.rst b/docs/ENVIRONMENT.rst
index 618497aa..7c837a6b 100644
--- a/docs/ENVIRONMENT.rst
+++ b/docs/ENVIRONMENT.rst
@@ -166,6 +166,8 @@ REST API
 -  **PATRONI\_RESTAPI\_CAFILE**: Specifies the file with the CA_BUNDLE with certificates of trusted CAs to use while verifying client certs.
 -  **PATRONI\_RESTAPI\_CIPHERS**: (optional) Specifies the permitted cipher suites (e.g. "ECDHE-RSA-AES256-GCM-SHA384:DHE-RSA-AES256-GCM-SHA384:ECDHE-RSA-AES128-GCM-SHA256:DHE-RSA-AES128-GCM-SHA256:!SSLv1:!SSLv2:!SSLv3:!TLSv1:!TLSv1.1")
 -  **PATRONI\_RESTAPI\_VERIFY\_CLIENT**: ``none`` (default), ``optional`` or ``required``. When ``none`` REST API will not check client certificates. When ``required`` client certificates are required for all REST API calls. When ``optional`` client certificates are required for all unsafe REST API endpoints. When ``required`` is used, then client authentication succeeds, if the certificate signature verification succeeds. For ``optional`` the client cert will only be checked for ``PUT``, ``POST``, ``PATCH``, and ``DELETE`` requests.
+-  **PATRONI\_RESTAPI\_ALLOWLIST**: (optional): Specifies the set of hosts that are allowed to call unsafe REST API endpoints. The single element could be a host name, an IP address or a network address using CIDR notation. By default ``allow all`` is used. In case if ``allowlist`` or ``allowlist_include_members`` are set, anything that is not included is rejected.
+-  **PATRONI\_RESTAPI\_ALLOWLIST\_INCLUDE\_MEMBERS**: (optional): If set to ``true`` it allows accessing unsafe REST API endpoints from other cluster members registered in DCS (IP address or hostname is taken from the members ``api_url``). Be careful, it might happen that OS will use a different IP for outgoing connections.
 -  **PATRONI\_RESTAPI\_HTTP\_EXTRA\_HEADERS**: (optional) HTTP headers let the REST API server pass additional information with an HTTP response.
 -  **PATRONI\_RESTAPI\_HTTPS\_EXTRA\_HEADERS**: (optional) HTTPS headers let the REST API server pass additional information with an HTTP response when TLS is enabled. This will also pass additional information set in ``http_extra_headers``.
 
diff --git a/docs/SETTINGS.rst b/docs/SETTINGS.rst
index 75531dd9..64428f38 100644
--- a/docs/SETTINGS.rst
+++ b/docs/SETTINGS.rst
@@ -327,6 +327,8 @@ REST API
         -  **cafile**: (optional): Specifies the file with the CA_BUNDLE with certificates of trusted CAs to use while verifying client certs.
         -  **ciphers**: (optional): Specifies the permitted cipher suites (e.g. "ECDHE-RSA-AES256-GCM-SHA384:DHE-RSA-AES256-GCM-SHA384:ECDHE-RSA-AES128-GCM-SHA256:DHE-RSA-AES128-GCM-SHA256:!SSLv1:!SSLv2:!SSLv3:!TLSv1:!TLSv1.1")
         -  **verify\_client**: (optional): ``none`` (default), ``optional`` or ``required``. When ``none`` REST API will not check client certificates. When ``required`` client certificates are required for all REST API calls. When ``optional`` client certificates are required for all unsafe REST API endpoints. When ``required`` is used, then client authentication succeeds, if the certificate signature verification succeeds.  For ``optional`` the client cert will only be checked for ``PUT``, ``POST``, ``PATCH``, and ``DELETE`` requests.
+        -  **allowlist**: (optional): Specifies the set of hosts that are allowed to call unsafe REST API endpoints. The single element could be a host name, an IP address or a network address using CIDR notation. By default ``allow all`` is used. In case if ``allowlist`` or ``allowlist_include_members`` are set, anything that is not included is rejected.
+        -  **allowlist\_include\_members**: (optional): If set to ``true`` it allows accessing unsafe REST API endpoints from other cluster members registered in DCS (IP address or hostname is taken from the members ``api_url``). Be careful, it might happen that OS will use a different IP for outgoing connections.
         -  **http\_extra\_headers**: (optional): HTTP headers let the REST API server pass additional information with an HTTP response.
         -  **https\_extra\_headers**: (optional): HTTPS headers let the REST API server pass additional information with an HTTP response when TLS is enabled. This will also pass additional information set in ``http_extra_headers``.
 
@@ -373,3 +375,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..e5e2baa9 100644
--- a/patroni/api.py
+++ b/patroni/api.py
@@ -12,6 +12,7 @@ import six
 import socket
 import sys
 
+from ipaddress import ip_address, ip_network as _ip_network
 from six.moves.BaseHTTPServer import BaseHTTPRequestHandler, HTTPServer
 from six.moves.socketserver import ThreadingMixIn
 from six.moves.urllib_parse import urlparse, parse_qs
@@ -25,6 +26,10 @@ from .utils import deep_compare, enable_keepalive, parse_bool, patch_config, Ret
 logger = logging.getLogger(__name__)
 
 
+def ip_network(value):
+    return _ip_network(value.decode('utf-8') if six.PY2 else value, False)
+
+
 class RestApiHandler(BaseHTTPRequestHandler):
 
     def _write_status_code_only(self, status_code):
@@ -47,17 +52,17 @@ class RestApiHandler(BaseHTTPRequestHandler):
     def _write_json_response(self, status_code, response):
         self._write_response(status_code, json.dumps(response, default=str), content_type='application/json')
 
-    def check_auth(func):
-        """Decorator function to check authorization header or client certificates
+    def check_access(func):
+        """Decorator function to check the source ip, authorization header. or client certificates
 
         Usage example:
-        @check_auth
+        @check_access
         def do_PUT_foo():
             pass
         """
 
         def wrapper(self, *args, **kwargs):
-            if self.server.check_auth(self):
+            if self.server.check_access(self):
                 return func(self, *args, **kwargs)
 
         return wrapper
@@ -108,9 +113,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 +125,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 +153,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 +219,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')
 
@@ -271,7 +304,7 @@ class RestApiHandler(BaseHTTPRequestHandler):
             logger.exception('Bad request')
         self.send_error(400)
 
-    @check_auth
+    @check_access
     def do_PATCH_config(self):
         request = self._read_json_content()
         if request:
@@ -286,7 +319,7 @@ class RestApiHandler(BaseHTTPRequestHandler):
             self.server.patroni.ha.wakeup()
             self._write_json_response(200, data)
 
-    @check_auth
+    @check_access
     def do_PUT_config(self):
         request = self._read_json_content()
         if request:
@@ -297,7 +330,7 @@ class RestApiHandler(BaseHTTPRequestHandler):
                     return self.send_error(502)
             self._write_json_response(200, request)
 
-    @check_auth
+    @check_access
     def do_POST_reload(self):
         self.server.patroni.sighup_handler()
         self._write_response(202, 'reload scheduled')
@@ -323,7 +356,7 @@ class RestApiHandler(BaseHTTPRequestHandler):
             status_code = 422
         return (status_code, error, scheduled_at)
 
-    @check_auth
+    @check_access
     def do_POST_restart(self):
         status_code = 500
         data = 'restart failed'
@@ -384,7 +417,7 @@ class RestApiHandler(BaseHTTPRequestHandler):
                     status_code = 409
         self._write_response(status_code, data)
 
-    @check_auth
+    @check_access
     def do_DELETE_restart(self):
         if self.server.patroni.ha.delete_future_restart():
             data = "scheduled restart deleted"
@@ -394,7 +427,7 @@ class RestApiHandler(BaseHTTPRequestHandler):
             code = 404
         self._write_response(code, data)
 
-    @check_auth
+    @check_access
     def do_DELETE_switchover(self):
         failover = self.server.patroni.dcs.get_cluster().failover
         if failover and failover.scheduled_at:
@@ -408,7 +441,7 @@ class RestApiHandler(BaseHTTPRequestHandler):
             code = 404
         self._write_response(code, data)
 
-    @check_auth
+    @check_access
     def do_POST_reinitialize(self):
         request = self._read_json_content(body_is_optional=True)
 
@@ -465,7 +498,7 @@ class RestApiHandler(BaseHTTPRequestHandler):
                 return None
         return action + ' is not possible: no good candidates have been found'
 
-    @check_auth
+    @check_access
     def do_POST_failover(self, action='failover'):
         request = self._read_json_content()
         (status_code, data) = (400, '')
@@ -613,7 +646,6 @@ class RestApiServer(ThreadingMixIn, HTTPServer, Thread):
         self.patroni = patroni
         self.__listen = None
         self.__ssl_options = None
-        self.http_extra_headers = {}
         self.reload_config(config)
         self.daemon = True
         self.__ssl_serial_number = None
@@ -647,7 +679,35 @@ class RestApiServer(ThreadingMixIn, HTTPServer, Thread):
             if not auth_header.startswith('Basic ') or not self.check_basic_auth_key(auth_header[6:]):
                 return 'not authenticated'
 
-    def check_auth(self, rh):
+    @staticmethod
+    def __resolve_ips(host, port):
+        try:
+            for _, _, _, _, sa in socket.getaddrinfo(host, port, 0, socket.SOCK_STREAM, socket.IPPROTO_TCP):
+                yield ip_network(sa[0])
+        except Exception as e:
+            logger.error('Failed to resolve %s: %r', host, e)
+
+    def __members_ips(self):
+        cluster = self.patroni.dcs.cluster
+        if self.__allowlist_include_members and cluster:
+            for member in cluster.members:
+                if member.api_url:
+                    try:
+                        r = urlparse(member.api_url)
+                        host = r.hostname
+                        port = r.port or (443 if r.scheme == 'https' else 80)
+                        for ip in self.__resolve_ips(host, port):
+                            yield ip
+                    except Exception as e:
+                        logger.debug('Failed to parse url %s: %r', member.api_url, e)
+
+    def check_access(self, rh):
+        if self.__allowlist or self.__allowlist_include_members:
+            incoming_ip = rh.client_address[0]
+            incoming_ip = ip_address(incoming_ip.decode('utf-8') if six.PY2 else incoming_ip)
+            if not any(incoming_ip in net for net in self.__allowlist + tuple(self.__members_ips())):
+                return rh._write_response(403, 'Access is denied')
+
         if not hasattr(rh.request, 'getpeercert') or not rh.request.getpeercert():  # valid client cert isn't present
             if self.__protocol == 'https' and self.__ssl_options.get('verify_client') in ('required', 'optional'):
                 return rh._write_response(403, 'client certificate required')
@@ -771,10 +831,25 @@ class RestApiServer(ThreadingMixIn, HTTPServer, Thread):
                 self.__ssl_serial_number = on_disk_cert_serial_number
                 return True
 
+    def _build_allowlist(self, value):
+        if isinstance(value, list):
+            for v in value:
+                if '/' in v:  # netmask
+                    try:
+                        yield ip_network(v)
+                    except Exception as e:
+                        logger.error('Invalid value "%s" in the allowlist: %r', v, e)
+                else:  # ip or hostname, try to resolve it
+                    for ip in self.__resolve_ips(v, 8080):
+                        yield ip
+
     def reload_config(self, config):
         if 'listen' not in config:  # changing config in runtime
             raise ValueError('Can not find "restapi.listen" config')
 
+        self.__allowlist = tuple(self._build_allowlist(config.get('allowlist')))
+        self.__allowlist_include_members = config.get('allowlist_include_members')
+
         ssl_options = {n: config[n] for n in ('certfile', 'keyfile', 'keyfile_password',
                                               'cafile', 'ciphers') if n in config}
 
diff --git a/patroni/config.py b/patroni/config.py
index 6c6c73b0..ce5b840e 100644
--- a/patroni/config.py
+++ b/patroni/config.py
@@ -268,11 +268,42 @@ class Config(object):
 
         _set_section_values('restapi', ['listen', 'connect_address', 'certfile', 'keyfile', 'keyfile_password',
                                         'cafile', 'ciphers', 'verify_client', 'http_extra_headers',
-                                        'https_extra_headers'])
+                                        'https_extra_headers', 'allowlist', 'allowlist_include_members'])
         _set_section_values('ctl', ['insecure', 'cacert', 'certfile', 'keyfile'])
         _set_section_values('postgresql', ['listen', 'connect_address', 'config_dir', 'data_dir', 'pgpass', 'bin_dir'])
         _set_section_values('log', ['level', 'traceback_level', 'format', 'dateformat', 'max_queue_size',
                                     'dir', 'file_size', 'file_num', 'loggers'])
+        _set_section_values('raft', ['data_dir', 'self_addr', 'partner_addrs', 'password', 'bind_addr'])
+
+        for first, second in (('restapi', 'allowlist_include_members'), ('ctl', 'insecure')):
+            value = ret.get(first, {}).pop(second, None)
+            if value:
+                value = parse_bool(value)
+                if value is not None:
+                    ret[first][second] = value
+
+        for second in ('max_queue_size', 'file_size', 'file_num'):
+            value = ret.get('log', {}).pop(second, None)
+            if value:
+                value = parse_int(value)
+                if value is not None:
+                    ret['log'][second] = value
+
+        def _parse_list(value):
+            if not (value.strip().startswith('-') or '[' in value):
+                value = '[{0}]'.format(value)
+            try:
+                return yaml.safe_load(value)
+            except Exception:
+                logger.exception('Exception when parsing list %s', value)
+                return None
+
+        for first, second in (('raft', 'partner_addrs'), ('restapi', 'allowlist')):
+            value = ret.get(first, {}).pop(second, None)
+            if value:
+                value = _parse_list(value)
+                if value:
+                    ret[first][second] = value
 
         def _parse_dict(value):
             if not value.strip().startswith('{'):
@@ -283,11 +314,13 @@ class Config(object):
                 logger.exception('Exception when parsing dict %s', value)
                 return None
 
-        value = ret.get('log', {}).pop('loggers', None)
-        if value:
-            value = _parse_dict(value)
-            if value:
-                ret['log']['loggers'] = value
+        for first, params in (('restapi', ('http_extra_headers', 'https_extra_headers')), ('log', ('loggers',))):
+            for second in params:
+                value = ret.get(first, {}).pop(second, None)
+                if value:
+                    value = _parse_dict(value)
+                    if value:
+                        ret[first][second] = value
 
         def _get_auth(name, params=None):
             ret = {}
@@ -310,19 +343,6 @@ class Config(object):
         if authentication:
             ret['postgresql']['authentication'] = authentication
 
-        def _parse_list(value):
-            if not (value.strip().startswith('-') or '[' in value):
-                value = '[{0}]'.format(value)
-            try:
-                return yaml.safe_load(value)
-            except Exception:
-                logger.exception('Exception when parsing list %s', value)
-                return None
-
-        _set_section_values('raft', ['data_dir', 'self_addr', 'partner_addrs', 'password', 'bind_addr'])
-        if 'raft' in ret and 'partner_addrs' in ret['raft']:
-            ret['raft']['partner_addrs'] = _parse_list(ret['raft']['partner_addrs'])
-
         for param in list(os.environ.keys()):
             if param.startswith(PATRONI_ENV_PREFIX):
                 # PATRONI_(ETCD|CONSUL|ZOOKEEPER|EXHIBITOR|...)_(HOSTS?|PORT|..)
@@ -339,7 +359,7 @@ class Config(object):
                         value = value and _parse_list(value)
                     elif suffix == 'LABELS':
                         value = _parse_dict(value)
-                    elif suffix in ('USE_PROXIES', 'REGISTER_SERVICE', 'USE_ENDPOINTS', 'BYPASS_API_SERVICE'):
+                    elif suffix in ('USE_PROXIES', 'REGISTER_SERVICE', 'USE_ENDPOINTS', 'BYPASS_API_SERVICE', 'VERIFY'):
                         value = parse_bool(value)
                     if value:
                         ret[name.lower()][suffix.lower()] = value
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'),
diff --git a/setup.py b/setup.py
index 7302858a..8cc5940a 100644
--- a/setup.py
+++ b/setup.py
@@ -24,7 +24,7 @@ KEYWORDS = 'etcd governor patroni postgresql postgres ha haproxy confd' +\
 
 EXTRAS_REQUIRE = {'aws': ['boto'], 'etcd': ['python-etcd'], 'etcd3': ['python-etcd'],
                   'consul': ['python-consul'], 'exhibitor': ['kazoo'], 'zookeeper': ['kazoo'],
-                  'kubernetes': ['ipaddress'], 'raft': ['pysyncobj', 'cryptography']}
+                  'kubernetes': [], 'raft': ['pysyncobj', 'cryptography']}
 COVERAGE_XML = True
 COVERAGE_HTML = False
 
@@ -175,13 +175,12 @@ def setup_package(version):
         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 = []
+                    deps[i] = r
                     EXTRAS_REQUIRE[e] = deps
                     extra = True
                     break
+            if extra:
+                break
         if not extra:
             install_requires.append(r)
 


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..34a6224e 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'))
 
@@ -455,7 +538,9 @@ class TestRestApiServer(unittest.TestCase):
     @patch.object(BaseHTTPServer.HTTPServer, '__init__', Mock())
     def setUp(self):
         self.srv = MockRestApiServer(Mock(), '', {'listen': '*:8008', 'certfile': 'a', 'verify_client': 'required',
-                                                  'ciphers': '!SSLv1:!SSLv2:!SSLv3:!TLSv1:!TLSv1.1'})
+                                                  'ciphers': '!SSLv1:!SSLv2:!SSLv3:!TLSv1:!TLSv1.1',
+                                                  'allowlist': ['127.0.0.1', '::1/128', '::1/zxc'],
+                                                  'allowlist_include_members': True})
 
     @patch.object(BaseHTTPServer.HTTPServer, '__init__', Mock())
     def test_reload_config(self):
@@ -466,10 +551,17 @@ class TestRestApiServer(unittest.TestCase):
         with patch.object(socket.socket, 'setsockopt', Mock(side_effect=socket.error)):
             self.srv.reload_config({'listen': ':8008'})
 
-    def test_check_auth(self):
+    @patch.object(MockPatroni, 'dcs')
+    def test_check_access(self, mock_dcs):
+        mock_dcs.cluster = get_cluster_initialized_without_leader()
+        mock_dcs.cluster.members[1].data['api_url'] = 'http://127.0.0.1z:8011/patroni'
+        mock_dcs.cluster.members.append(Member(0, 'bad-api-url', 30, {'api_url': 123}))
         mock_rh = Mock()
+        mock_rh.client_address = ('127.0.0.2',)
+        self.assertIsNot(self.srv.check_access(mock_rh), True)
+        mock_rh.client_address = ('127.0.0.1',)
         mock_rh.request.getpeercert.return_value = None
-        self.assertIsNot(self.srv.check_auth(mock_rh), True)
+        self.assertIsNot(self.srv.check_access(mock_rh), True)
 
     def test_handle_error(self):
         try:
@@ -507,3 +599,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_config.py b/tests/test_config.py
index b656da26..dddae174 100644
--- a/tests/test_config.py
+++ b/tests/test_config.py
@@ -30,12 +30,14 @@ class TestConfig(unittest.TestCase):
             'PATRONI_SCOPE': 'batman2',
             'PATRONI_LOGLEVEL': 'ERROR',
             'PATRONI_LOG_LOGGERS': 'patroni.postmaster: WARNING, urllib3: DEBUG',
+            'PATRONI_LOG_FILE_NUM': '5',
             'PATRONI_RESTAPI_USERNAME': 'username',
             'PATRONI_RESTAPI_PASSWORD': 'password',
             'PATRONI_RESTAPI_LISTEN': '0.0.0.0:8008',
             'PATRONI_RESTAPI_CONNECT_ADDRESS': '127.0.0.1:8008',
             'PATRONI_RESTAPI_CERTFILE': '/certfile',
             'PATRONI_RESTAPI_KEYFILE': '/keyfile',
+            'PATRONI_RESTAPI_ALLOWLIST_INCLUDE_MEMBERS': 'on',
             'PATRONI_POSTGRESQL_LISTEN': '0.0.0.0:5432',
             'PATRONI_POSTGRESQL_CONNECT_ADDRESS': '127.0.0.1:5432',
             'PATRONI_POSTGRESQL_DATA_DIR': 'data/postgres0',
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})