Repository: zalando/patroni
Pull request #1889 (2021-03-22 07:19:10)

Write prev LSN before the checkpoint to optime if achive_mode=on
The #1527 introduced a feature of updating `/optime/leader` with the location of the last checkpoint after the Postgres was shutdown cleanly.

If wal archiving is enabled, Postgres always switching the WAL file before writing the checkpoint shutdown record. Normally it is not an issue, but for databases without too much write activity it could lead to the situation that the visible replication lag becomes equal to the size of a single WAL file. In fact, the previous WAL file is mostly empty and contains only a few records.

Therefore it should be safe to report the LSN of the SWITCH record before the shutdown checkpoint.
In order to do that, Patroni first gets the output of the pg_controldata and based on it calls pg_waldump two times:
* The first call reads the checkpoint record (and verifies that this is really the shutdown checkpoint).
* The next call reads the previous record and in case if it is the 'xlog switch' (for 9.3 and 9.4) or 'SWITCH' (for 9.5+), the LSN
of the SWITCH record is written to the `/optime/leader`.

In case of any mismatch, failure to call pg_waldump or parse its output, the old behavior is retained, i.e. `Latest checkpoint location` from the pg_controldata is used.

Close https://github.com/zalando/patroni/issues/1860


Patch:
diff --git a/.github/workflows/install_deps.py b/.github/workflows/install_deps.py
index 1a4477ea..187d634f 100644
--- a/.github/workflows/install_deps.py
+++ b/.github/workflows/install_deps.py
@@ -22,7 +22,7 @@ def install_requirements(what):
     for r in read('requirements.txt').split('\n'):
         r = r.strip()
         if r != '':
-            extras = {e for e, v in EXTRAS_REQUIRE.items() if v and r.startswith(v[0])}
+            extras = {e for e, v in EXTRAS_REQUIRE.items() if v and any(r.startswith(x) for x in v)}
             if not extras or what == 'all' or what in extras:
                 requirements.append(r)
 
@@ -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/README.rst b/README.rst
index bf4322e0..b26cfafe 100644
--- a/README.rst
+++ b/README.rst
@@ -1,4 +1,4 @@
-|Build Status| |Coverage Status|
+|Tests Status| |Coverage Status|
 
 Patroni: A Template for PostgreSQL HA with ZooKeeper, etcd or Consul
 --------------------------------------------------------------------
@@ -171,7 +171,7 @@ Applications Should Not Use Superusers
 
 When connecting from an application, always use a non-superuser. Patroni requires access to the database to function properly. By using a superuser from an application, you can potentially use the entire connection pool, including the connections reserved for superusers, with the ``superuser_reserved_connections`` setting. If Patroni cannot access the Primary because the connection pool is full, behavior will be undesirable.
 
-.. |Build Status| image:: https://travis-ci.org/zalando/patroni.svg?branch=master
-   :target: https://travis-ci.org/zalando/patroni
+.. |Tests Status| image:: https://github.com/zalando/patroni/actions/workflows/tests.yaml/badge.svg
+   :target: https://github.com/zalando/patroni/actions/workflows/tests.yaml?query=branch%3Amaster
 .. |Coverage Status| image:: https://coveralls.io/repos/zalando/patroni/badge.svg?branch=master
-   :target: https://coveralls.io/r/zalando/patroni?branch=master
+   :target: https://coveralls.io/github/zalando/patroni?branch=master
diff --git a/docs/SETTINGS.rst b/docs/SETTINGS.rst
index cd951e68..664a62c8 100644
--- a/docs/SETTINGS.rst
+++ b/docs/SETTINGS.rst
@@ -293,6 +293,7 @@ PostgreSQL
     -  **bin\_dir**: Path to PostgreSQL binaries (pg_ctl, pg_rewind, pg_basebackup, postgres). The default value is an empty string meaning that PATH environment variable will be used to find the executables.
     -  **listen**: IP address + port that Postgres listens to; must be accessible from other nodes in the cluster, if you're using streaming replication. Multiple comma-separated addresses are permitted, as long as the port component is appended after to the last one with a colon, i.e. ``listen: 127.0.0.1,127.0.0.2:5432``. Patroni will use the first address from this list to establish local connections to the PostgreSQL node.
     -  **use\_unix\_socket**: specifies that Patroni should prefer to use unix sockets to connect to the cluster. Default value is ``false``. If ``unix_socket_directories`` is defined, Patroni will use the first suitable value from it to connect to the cluster and fallback to tcp if nothing is suitable. If ``unix_socket_directories`` is not specified in ``postgresql.parameters``, Patroni will assume that the default value should be used and omit ``host`` from the connection parameters.
+    -  **use\_unix\_socket\_repl**: specifies that Patroni should prefer to use unix sockets for replication user cluster connection. Default value is ``false``. If ``unix_socket_directories`` is defined, Patroni will use the first suitable value from it to connect to the cluster and fallback to tcp if nothing is suitable. If ``unix_socket_directories`` is not specified in ``postgresql.parameters``, Patroni will assume that the default value should be used and omit ``host`` from the connection parameters.
     -  **pgpass**: path to the `.pgpass <https://www.postgresql.org/docs/current/static/libpq-pgpass.html>`__ password file. Patroni creates this file before executing pg\_basebackup, the post_init script and under some other circumstances. The location must be writable by Patroni.
     -  **recovery\_conf**: additional configuration settings written to recovery.conf when configuring follower.
     -  **custom\_conf** : path to an optional custom ``postgresql.conf`` file, that will be used in place of ``postgresql.base.conf``. The file must exist on all cluster nodes, be readable by PostgreSQL and will be included from its location on the real ``postgresql.conf``. Note that Patroni will not monitor this file for changes, nor backup it. However, its settings can still be overridden by Patroni's own configuration facilities - see :ref:`dynamic configuration <dynamic_configuration>` for details.
@@ -372,3 +373,12 @@ Tags
 - **noloadbalance**: ``true`` or ``false``. If set to ``true`` the node will return HTTP Status Code 503 for the ``GET /replica`` REST API health-check and therefore will be excluded from the load-balancing. Defaults to ``false``.
 - **replicatefrom**: The IP address/hostname of another replica. Used to support cascading replication.
 - **nosync**: ``true`` or ``false``. If set to ``true`` the node will never be selected as a synchronous replica.
+
+In addition to these predefined tags, you can also add your own ones:
+
+- **key1**: ``true``
+- **key2**: ``false``
+- **key3**: ``1.4``
+- **key4**: ``"RandomString"``
+
+Tags are visible in the :ref:`REST API <rest_api>` and ``patronictl list`` You can also check for an instance health using these tags. If the tag isn't defined for an instance, or if the respective value doesn't match the querying value, it will return HTTP Status Code 503.
diff --git a/docs/rest_api.rst b/docs/rest_api.rst
index 74003540..05d579c9 100644
--- a/docs/rest_api.rst
+++ b/docs/rest_api.rst
@@ -9,14 +9,17 @@ Health check endpoints
 ----------------------
 For all health check ``GET`` requests Patroni returns a JSON document with the status of the node, along with the HTTP status code. If you don't want or don't need the JSON document, you might consider using the ``OPTIONS`` method instead of ``GET``.
 
-- The following requests to Patroni REST API will return HTTP status code **200** only when the Patroni node is running as the leader:
+- The following requests to Patroni REST API will return HTTP status code **200** only when the Patroni node is running as the primary with leader lock:
 
   - ``GET /``
   - ``GET /master``
-  - ``GET /leader``
   - ``GET /primary``
   - ``GET /read-write``
 
+- ``GET /standby-leader``: returns HTTP status code **200** only when the Patroni node is running as the leader in a :ref:`standby cluster <standby_cluster>`.
+
+- ``GET /leader``: returns HTTP status code **200** when the Patroni node has the leader lock. The major difference from the two previous endpoints is that it doesn't take into account whether PostgreSQL is running as the ``primary`` or the ``standby_leader``.
+
 - ``GET /replica``: replica health check endpoint. It returns HTTP status code **200** only when the Patroni node is in the state ``running``, the role is ``replica`` and ``noloadbalance`` tag is not set.
 
 - ``GET /replica?lag=<max-lag>``: replica check endpoint. In addition to checks from ``replica``, it also checks replication latency and returns status code **200** only when it is below specified value. The key cluster.last_leader_operation from DCS is used for Leader wal position and compute latency on replica for performance reasons. max-lag can be specified in bytes (integer) or in human readable values, for e.g. 16kB, 64MB, 1GB.
@@ -26,9 +29,18 @@ For all health check ``GET`` requests Patroni returns a JSON document with the s
   - ``GET /replica?lag=10MB``
   - ``GET /replica?lag=1GB``
 
-- ``GET /read-only``: like the above endpoint, but also includes the primary.
+- ``GET /replica?tag_key1=value1&tag_key2=value2``: replica check endpoint. In addition, It will also check for user defined tags ``key1`` and ``key2`` and their respective values in the **tags** section of the yaml configuration management. If the tag isn't defined for an instance, or if the value in the yaml configuration doesn't match the querying value, it will return HTTP Status Code 503.
 
-- ``GET /standby-leader``: returns HTTP status code **200** only when the Patroni node is running as the leader in a :ref:`standby cluster <standby_cluster>`.
+ In the following requests, since we are checking for the leader or standby-leader status, Patroni doesn't apply any of the user defined tags and they will be ignored.
+  - ``GET /?tag_key1=value1&tag_key2=value2``
+  - ``GET /master?tag_key1=value1&tag_key2=value2``
+  - ``GET /leader?tag_key1=value1&tag_key2=value2``
+  - ``GET /primary?tag_key1=value1&tag_key2=value2``
+  - ``GET /read-write?tag_key1=value1&tag_key2=value2``
+  - ``GET /standby_leader?tag_key1=value1&tag_key2=value2``
+  - ``GET /standby-leader?tag_key1=value1&tag_key2=value2``
+
+- ``GET /read-only``: like the above endpoint, but also includes the primary.
 
 - ``GET /synchronous`` or ``GET /sync``: returns HTTP status code **200** only when the Patroni node is running as a synchronous standby.
 
diff --git a/features/basic_replication.feature b/features/basic_replication.feature
index 30891889..3aaeada6 100644
--- a/features/basic_replication.feature
+++ b/features/basic_replication.feature
@@ -29,7 +29,7 @@ Feature: basic replication
     Then I receive a response code 200
 
   Scenario: check stuck sync replica
-    Given I issue a PATCH request to http://127.0.0.1:8008/config with {"maximum_lag_on_syncnode": 15000000, "postgresql": {"parameters": {"synchronous_commit": "remote_apply"}}}
+    Given I issue a PATCH request to http://127.0.0.1:8008/config with {"pause": true, "maximum_lag_on_syncnode": 15000000, "postgresql": {"parameters": {"synchronous_commit": "remote_apply"}}}
     Then I receive a response code 200
     And I create table on postgres0
     And table mytest is present on postgres1 after 2 seconds
@@ -43,7 +43,7 @@ Feature: basic replication
     Then I receive a response code 200
     When I issue a GET request to http://127.0.0.1:8010/async
     Then I receive a response code 200
-    When I issue a PATCH request to http://127.0.0.1:8008/config with {"maximum_lag_on_syncnode": -1, "postgresql": {"parameters": {"synchronous_commit": "on"}}}
+    When I issue a PATCH request to http://127.0.0.1:8008/config with {"pause": null, "maximum_lag_on_syncnode": -1, "postgresql": {"parameters": {"synchronous_commit": "on"}}}
     Then I receive a response code 200
     And I drop table on postgres0
 
diff --git a/features/environment.py b/features/environment.py
index 9a050923..130a5bf5 100644
--- a/features/environment.py
+++ b/features/environment.py
@@ -180,6 +180,7 @@ class PatroniController(AbstractController):
         config['postgresql']['data_dir'] = self._data_dir
         config['postgresql']['basebackup'] = [{'checkpoint': 'fast'}]
         config['postgresql']['use_unix_socket'] = os.name != 'nt'  # windows doesn't yet support unix-domain sockets
+        config['postgresql']['use_unix_socket_repl'] = os.name != 'nt'  # windows doesn't yet support unix-domain sockets
         config['postgresql']['pgpass'] = os.path.join(tempfile.gettempdir(), 'pgpass_' + name)
         config['postgresql']['parameters'].update({
             'logging_collector': 'on', 'log_destination': 'csvlog', 'log_directory': self._output_dir,
@@ -590,10 +591,12 @@ class ExhibitorController(ZooKeeperController):
 class RaftController(AbstractDcsController):
 
     CONTROLLER_ADDR = 'localhost:1234'
+    PASSWORD = '12345'
 
     def __init__(self, context):
         super(RaftController, self).__init__(context)
-        os.environ.update(PATRONI_RAFT_PARTNER_ADDRS="'" + self.CONTROLLER_ADDR + "'", RAFT_PORT='1234')
+        os.environ.update(PATRONI_RAFT_PARTNER_ADDRS="'" + self.CONTROLLER_ADDR + "'",
+                          PATRONI_RAFT_PASSWORD=self.PASSWORD, RAFT_PORT='1234')
         self._raft = None
 
     def _start(self):
@@ -614,18 +617,16 @@ class RaftController(AbstractDcsController):
 
     def cleanup_service_tree(self):
         from patroni.dcs.raft import KVStoreTTL
-        from pysyncobj import SyncObjConf
 
         if self._raft:
             self._raft.destroy()
-            self._raft._SyncObj__thread.join()
             self.stop()
             os.makedirs(self._work_directory)
             self.start()
 
         ready_event = threading.Event()
-        conf = SyncObjConf(appendEntriesUseBatch=False, dynamicMembershipChange=True, onReady=ready_event.set)
-        self._raft = KVStoreTTL(None, [self.CONTROLLER_ADDR], conf)
+        self._raft = KVStoreTTL(ready_event.set, None, None, partner_addrs=[self.CONTROLLER_ADDR], password=self.PASSWORD)
+        self._raft.startAutoTick()
         ready_event.wait()
 
 
diff --git a/patroni/__init__.py b/patroni/__init__.py
index 4d3463a5..dc5823fc 100644
--- a/patroni/__init__.py
+++ b/patroni/__init__.py
@@ -74,6 +74,7 @@ class Patroni(AbstractPatroniDaemon):
             if local:
                 self.tags = self.get_tags()
                 self.request.reload_config(self.config)
+            if local or self.api.reload_local_certificate():
                 self.api.reload_config(self.config['restapi'])
             self.watchdog.reload_config(self.config)
             self.postgresql.reload_config(self.config['postgresql'], sighup)
diff --git a/patroni/api.py b/patroni/api.py
index 7bc9fc63..2a34c242 100644
--- a/patroni/api.py
+++ b/patroni/api.py
@@ -108,9 +108,11 @@ class RestApiHandler(BaseHTTPRequestHandler):
             response.get('role') == 'replica' and response.get('state') == 'running' else 503
 
         if not cluster and patroni.ha.is_paused():
+            leader_status_code = 200 if response.get('role') in ('master', 'standby_leader') else 503
             primary_status_code = 200 if response.get('role') == 'master' else 503
             standby_leader_status_code = 200 if response.get('role') == 'standby_leader' else 503
         elif patroni.ha.is_leader():
+            leader_status_code = 200
             if patroni.ha.is_standby_cluster():
                 primary_status_code = replica_status_code = 503
                 standby_leader_status_code = 200 if response.get('role') in ('replica', 'standby_leader') else 503
@@ -118,14 +120,20 @@ class RestApiHandler(BaseHTTPRequestHandler):
                 primary_status_code = 200
                 standby_leader_status_code = 503
         else:
-            primary_status_code = standby_leader_status_code = 503
+            leader_status_code = primary_status_code = standby_leader_status_code = 503
 
         status_code = 503
 
+        ignore_tags = False
         if 'standby_leader' in path or 'standby-leader' in path:
             status_code = standby_leader_status_code
-        elif 'master' in path or 'leader' in path or 'primary' in path or 'read-write' in path:
+            ignore_tags = True
+        elif 'leader' in path:
+            status_code = leader_status_code
+            ignore_tags = True
+        elif 'master' in path or 'primary' in path or 'read-write' in path:
             status_code = primary_status_code
+            ignore_tags = True
         elif 'replica' in path:
             status_code = replica_status_code
         elif 'read-only' in path:
@@ -140,6 +148,25 @@ class RestApiHandler(BaseHTTPRequestHandler):
             elif path in ('/async', '/asynchronous') and not is_synchronous:
                 status_code = replica_status_code
 
+        # check for user defined tags in query params
+        if not ignore_tags and status_code == 200:
+            qs_tag_prefix = "tag_"
+            for qs_key, qs_value in self.path_query.items():
+                if not qs_key.startswith(qs_tag_prefix):
+                    continue
+                qs_key = qs_key[len(qs_tag_prefix):]
+                qs_value = qs_value[0]
+                instance_tag_value = patroni.tags.get(qs_key)
+                # tag not registered for instance
+                if instance_tag_value is None:
+                    status_code = 503
+                    break
+                if not isinstance(instance_tag_value, six.string_types):
+                    instance_tag_value = str(instance_tag_value).lower()
+                if instance_tag_value != qs_value:
+                    status_code = 503
+                    break
+
         if write_status_code_only:  # when haproxy sends OPTIONS request it reads only status code and nothing more
             self._write_status_code_only(status_code)
         else:
@@ -187,73 +214,74 @@ class RestApiHandler(BaseHTTPRequestHandler):
 
         metrics = []
 
+        scope_label = '{{scope="{0}"}}'.format(patroni.postgresql.scope)
         metrics.append("# HELP patroni_version Patroni semver without periods.")
         metrics.append("# TYPE patroni_version gauge")
         padded_semver = ''.join([x.zfill(2) for x in patroni.version.split('.')])  # 2.0.2 => 020002
-        metrics.append("patroni_version {0}".format(padded_semver))
+        metrics.append("patroni_version{0} {1}".format(scope_label, padded_semver))
 
         metrics.append("# HELP patroni_postgres_running Value is 1 if Postgres is running, 0 otherwise.")
         metrics.append("# TYPE patroni_postgres_running gauge")
-        metrics.append("patroni_postgres_running {0}".format(int(postgres['state'] == 'running')))
+        metrics.append("patroni_postgres_running{0} {1}".format(scope_label, int(postgres['state'] == 'running')))
 
         metrics.append("# HELP patroni_postmaster_start_time Epoch seconds since Postgres started.")
         metrics.append("# TYPE patroni_postmaster_start_time gauge")
         postmaster_start_time = postgres.get('postmaster_start_time')
         postmaster_start_time = (postmaster_start_time - epoch).total_seconds() if postmaster_start_time else 0
-        metrics.append("patroni_postmaster_start_time {0}".format(postmaster_start_time))
+        metrics.append("patroni_postmaster_start_time{0} {1}".format(scope_label, postmaster_start_time))
 
         metrics.append("# HELP patroni_master Value is 1 if this node is the leader, 0 otherwise.")
         metrics.append("# TYPE patroni_master gauge")
-        metrics.append("patroni_master {0}".format(int(postgres['role'] == 'master')))
+        metrics.append("patroni_master{0} {1}".format(scope_label, int(postgres['role'] == 'master')))
 
         metrics.append("# HELP patroni_xlog_location Current location of the Postgres"
                        " transaction log, 0 if this node is not the leader.")
         metrics.append("# TYPE patroni_xlog_location counter")
-        metrics.append("patroni_xlog_location {0}".format(postgres.get('xlog', {}).get('location', 0)))
+        metrics.append("patroni_xlog_location{0} {1}".format(scope_label, postgres.get('xlog', {}).get('location', 0)))
 
         metrics.append("# HELP patroni_standby_leader Value is 1 if this node is the standby_leader, 0 otherwise.")
         metrics.append("# TYPE patroni_standby_leader gauge")
-        metrics.append("patroni_standby_leader {0}".format(int(postgres['role'] == 'standby_leader')))
+        metrics.append("patroni_standby_leader{0} {1}".format(scope_label, int(postgres['role'] == 'standby_leader')))
 
         metrics.append("# HELP patroni_replica Value is 1 if this node is a replica, 0 otherwise.")
         metrics.append("# TYPE patroni_replica gauge")
-        metrics.append("patroni_replica {0}".format(int(postgres['role'] == 'replica')))
+        metrics.append("patroni_replica{0} {1}".format(scope_label, int(postgres['role'] == 'replica')))
 
         metrics.append("# HELP patroni_xlog_received_location Current location of the received"
                        " Postgres transaction log, 0 if this node is not a replica.")
         metrics.append("# TYPE patroni_xlog_received_location counter")
-        metrics.append("patroni_xlog_received_location {0}".format(
-                        postgres.get('xlog', {}).get('received_location', 0)))
+        metrics.append("patroni_xlog_received_location{0} {1}"
+                       .format(scope_label, postgres.get('xlog', {}).get('received_location', 0)))
 
         metrics.append("# HELP patroni_xlog_replayed_location Current location of the replayed"
                        " Postgres transaction log, 0 if this node is not a replica.")
         metrics.append("# TYPE patroni_xlog_replayed_location counter")
-        metrics.append("patroni_xlog_replayed_location {0}".format(
-                        postgres.get('xlog', {}).get('replayed_location', 0)))
+        metrics.append("patroni_xlog_replayed_location{0} {1}"
+                       .format(scope_label, postgres.get('xlog', {}).get('replayed_location', 0)))
 
         metrics.append("# HELP patroni_xlog_replayed_timestamp Current timestamp of the replayed"
                        " Postgres transaction log, 0 if null.")
         metrics.append("# TYPE patroni_xlog_replayed_timestamp gauge")
         replayed_timestamp = postgres.get('xlog', {}).get('replayed_timestamp')
         replayed_timestamp = (replayed_timestamp - epoch).total_seconds() if replayed_timestamp else 0
-        metrics.append("patroni_xlog_replayed_timestamp {0}".format(replayed_timestamp))
+        metrics.append("patroni_xlog_replayed_timestamp{0} {1}".format(scope_label, replayed_timestamp))
 
         metrics.append("# HELP patroni_xlog_paused Value is 1 if the Postgres xlog is paused, 0 otherwise.")
         metrics.append("# TYPE patroni_xlog_paused gauge")
-        metrics.append("patroni_xlog_paused {0}".format(
-                        int(postgres.get('xlog', {}).get('paused', False) is True)))
+        metrics.append("patroni_xlog_paused{0} {1}"
+                       .format(scope_label, int(postgres.get('xlog', {}).get('paused', False) is True)))
 
         metrics.append("# HELP patroni_postgres_server_version Version of Postgres (if running), 0 otherwise.")
         metrics.append("# TYPE patroni_postgres_server_version gauge")
-        metrics.append("patroni_postgres_server_version {0}".format(postgres.get('server_version', 0)))
+        metrics.append("patroni_postgres_server_version {0} {1}".format(scope_label, postgres.get('server_version', 0)))
 
         metrics.append("# HELP patroni_cluster_unlocked Value is 1 if the cluster is unlocked, 0 if locked.")
         metrics.append("# TYPE patroni_cluster_unlocked gauge")
-        metrics.append("patroni_cluster_unlocked {0}".format(int(postgres['cluster_unlocked'])))
+        metrics.append("patroni_cluster_unlocked{0} {1}".format(scope_label, int(postgres['cluster_unlocked'])))
 
         metrics.append("# HELP patroni_postgres_timeline Postgres timeline of this node (if running), 0 otherwise.")
         metrics.append("# TYPE patroni_postgres_timeline counter")
-        metrics.append("patroni_postgres_timeline {0}".format(postgres.get('timeline', 0)))
+        metrics.append("patroni_postgres_timeline{0} {1}".format(scope_label, postgres.get('timeline', 0)))
 
         self._write_response(200, '\n'.join(metrics)+'\n', content_type='text/plain')
 
@@ -616,6 +644,8 @@ class RestApiServer(ThreadingMixIn, HTTPServer, Thread):
         self.http_extra_headers = {}
         self.reload_config(config)
         self.daemon = True
+        self.__ssl_serial_number = None
+        self._received_new_cert = False
 
     def query(self, sql, *params):
         cursor = None
@@ -701,6 +731,7 @@ class RestApiServer(ThreadingMixIn, HTTPServer, Thread):
 
         self.__listen = listen
         self.__ssl_options = ssl_options
+        self._received_new_cert = False  # reset to False after reload_config()
 
         self.__httpserver_init(host, port)
         Thread.__init__(self, target=self.serve_forever)
@@ -723,6 +754,7 @@ class RestApiServer(ThreadingMixIn, HTTPServer, Thread):
                     ctx.verify_mode = modes[verify_client]
                 else:
                     logger.error('Bad value in the "restapi.verify_client": %s', verify_client)
+            self.__ssl_serial_number = self.get_certificate_serial_number()
             self.socket = ctx.wrap_socket(self.socket, server_side=True)
         if reloading_config:
             self.start()
@@ -750,6 +782,23 @@ class RestApiServer(ThreadingMixIn, HTTPServer, Thread):
             _, request = request  # SSLSocket
         return super(RestApiServer, self).shutdown_request(request)
 
+    def get_certificate_serial_number(self):
+        if self.__ssl_options.get('certfile'):
+            import ssl
+            try:
+                crt = ssl._ssl._test_decode_cert(self.__ssl_options['certfile'])
+                return crt.get('serialNumber')
+            except ssl.SSLError as e:
+                logger.error('Failed to get serial number from certificate %s: %r', self.__ssl_options['certfile'], e)
+
+    def reload_local_certificate(self):
+        if self.__protocol == 'https':
+            on_disk_cert_serial_number = self.get_certificate_serial_number()
+            if on_disk_cert_serial_number != self.__ssl_serial_number:
+                self._received_new_cert = True
+                self.__ssl_serial_number = on_disk_cert_serial_number
+                return True
+
     def reload_config(self, config):
         if 'listen' not in config:  # changing config in runtime
             raise ValueError('Can not find "restapi.listen" config')
@@ -763,7 +812,7 @@ class RestApiServer(ThreadingMixIn, HTTPServer, Thread):
         if isinstance(config.get('verify_client'), six.string_types):
             ssl_options['verify_client'] = config['verify_client'].lower()
 
-        if self.__listen != config['listen'] or self.__ssl_options != ssl_options:
+        if self.__listen != config['listen'] or self.__ssl_options != ssl_options or self._received_new_cert:
             self.__initialize(config['listen'], ssl_options)
 
         self.__auth_key = base64.b64encode(config['auth'].encode('utf-8')) if 'auth' in config else None
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..4abb998b 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
diff --git a/patroni/dcs/raft.py b/patroni/dcs/raft.py
index f94e6e64..76c65efb 100644
--- a/patroni/dcs/raft.py
+++ b/patroni/dcs/raft.py
@@ -5,7 +5,8 @@ import threading
 import time
 
 from pysyncobj import SyncObj, SyncObjConf, replicated, FAIL_REASON
-from pysyncobj.transport import Node, TCPTransport, CONNECTION_STATE
+from pysyncobj.transport import TCPTransport, CONNECTION_STATE
+from pysyncobj.utility import TcpUtility, UtilityException
 
 from . import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, SyncState, TimelineHistory
 from ..utils import validate_directory
@@ -13,122 +14,97 @@ from ..utils import validate_directory
 logger = logging.getLogger(__name__)
 
 
-class MessageNode(Node):
-
-    def __init__(self, address):
-        self.address = address
-
-
-class UtilityTransport(TCPTransport):
+class _TCPTransport(TCPTransport):
 
     def __init__(self, syncObj, selfNode, otherNodes):
-        super(UtilityTransport, self).__init__(syncObj, selfNode, otherNodes)
-        self._selfIsReadonlyNode = False
-
-    def _connectIfNecessarySingle(self, node):
-        pass
-
-    def connectionState(self, node):
-        return self._connections[node].state
+        super(_TCPTransport, self).__init__(syncObj, selfNode, otherNodes)
+        self.setOnUtilityMessageCallback('members', syncObj.getMembers)
 
-    def isDisconnected(self, node):
-        return self.connectionState(node) == CONNECTION_STATE.DISCONNECTED
 
-    def connectIfRequiredSingle(self, node):
-        if self.isDisconnected(node):
-            return self._connections[node].connect(node.ip, node.port)
-
-    def disconnectSingle(self, node):
-        self._connections[node].disconnect()
-
-
-class SyncObjUtility(SyncObj):
+class SyncObjUtility(object):
 
     def __init__(self, otherNodes, conf):
-        autoTick = conf.autoTick
-        conf.autoTick = False
-        super(SyncObjUtility, self).__init__(None, otherNodes, conf, transportClass=UtilityTransport)
-        conf.autoTick = autoTick
-        self._SyncObj__transport.setOnMessageReceivedCallback(self._onMessageReceived)
-        self.__result = None
-
-    def setPartnerNode(self, partner):
-        self.__node = partner
+        self._nodes = otherNodes
+        self._utility = TcpUtility(conf.password)
 
-    def sendMessage(self, message):
-        # Abuse the fact that node address is send as a first message
-        self._SyncObj__transport._selfNode = MessageNode(message)
-        self._SyncObj__transport.connectIfRequiredSingle(self.__node)
-        while not self._SyncObj__transport.isDisconnected(self.__node):
-            self._poller.poll(0.5)
-        return self.__result
+    def executeCommand(self, command):
+        try:
+            return self._utility.executeCommand(self.__node, command)
+        except UtilityException:
+            return None
 
-    def _onMessageReceived(self, _, message):
-        self.__result = message
-        self._SyncObj__transport.disconnectSingle(self.__node)
+    def getMembers(self):
+        for self.__node in self._nodes:
+            response = self.executeCommand(['members'])
+            if response:
+                return [member['addr'] for member in response]
 
 
-class MyTCPTransport(TCPTransport):
+class DynMemberSyncObj(SyncObj):
 
-    def _onIncomingMessageReceived(self, conn, message):
-        if self._syncObj.encryptor and not conn.sendRandKey:
-            conn.sendRandKey = message
-            conn.recvRandKey = os.urandom(32)
-            conn.send(conn.recvRandKey)
-            return
+    def __init__(self, selfAddress, partnerAddrs, conf):
+        self.__early_apply_local_log = selfAddress is not None
+        self.applied_local_log = False
 
-        # Utility messages
-        if isinstance(message, list) and message[0] == 'members':
-            conn.send(self._syncObj._get_members())
-            return True
+        utility = SyncObjUtility(partnerAddrs, conf)
+        members = utility.getMembers()
+        add_self = members and selfAddress not in members
 
-        return super(MyTCPTransport, self)._onIncomingMessageReceived(conn, message)
+        partnerAddrs = [member for member in (members or partnerAddrs) if member != selfAddress]
 
+        super(DynMemberSyncObj, self).__init__(selfAddress, partnerAddrs, conf, transportClass=_TCPTransport)
 
-class DynMemberSyncObj(SyncObj):
+        if add_self:
+            thread = threading.Thread(target=utility.executeCommand, args=(['add', selfAddress],))
+            thread.daemon = True
+            thread.start()
 
-    def __init__(self, selfAddress, partnerAddrs, conf):
-        add_self = False
-        utility = SyncObjUtility(partnerAddrs, conf)
-        for node in utility._SyncObj__otherNodes:
-            utility.setPartnerNode(node)
-            response = utility.sendMessage(['members'])
-            if response:
-                partnerAddrs = [member['addr'] for member in response if member['addr'] != selfAddress]
-                add_self = selfAddress and len(partnerAddrs) == len(response)
-                break
+    def getMembers(self, args, callback):
+        callback([{'addr': node.id, 'leader': node == self._getLeader(), 'status': CONNECTION_STATE.CONNECTED
+                   if self.isNodeConnected(node) else CONNECTION_STATE.DISCONNECTED} for node in self.otherNodes] +
+                 [{'addr': self.selfNode.id, 'leader': self._isLeader(), 'status': CONNECTION_STATE.CONNECTED}], None)
 
-        super(DynMemberSyncObj, self).__init__(selfAddress, partnerAddrs, conf, transportClass=MyTCPTransport)
-        if add_self:
-            threading.Thread(target=utility.sendMessage, args=(['add', selfAddress],)).start()
-
-    def _get_members(self):
-        ret = [{'addr': node.id, 'leader': node == self._getLeader(),
-                'status': CONNECTION_STATE.CONNECTED if node in self._SyncObj__connectedNodes
-                else CONNECTION_STATE.DISCONNECTED} for node in self._SyncObj__otherNodes]
-        ret.append({'addr': self._SyncObj__selfNode.id, 'leader': self._isLeader(),
-                    'status': CONNECTION_STATE.CONNECTED})
-        return ret
+    def _onTick(self, timeToWait=0.0):
+        super(DynMemberSyncObj, self)._onTick(timeToWait)
 
-    def _SyncObj__doChangeCluster(self, request, reverse=False):
-        ret = False
-        if not self._SyncObj__selfNode or request[0] != 'add' or reverse or request[1] != self._SyncObj__selfNode.id:
-            ret = super(DynMemberSyncObj, self)._SyncObj__doChangeCluster(request, reverse)
-            if ret:
-                self.forceLogCompaction()
-        return ret
+        # The SyncObj calls onReady callback only when cluster got the leader and is ready for writes.
+        # In some cases for us it is safe to "signal" the Raft object when the local log is fully applied.
+        # We are using the `applied_local_log` property for that, but not calling the callback function.
+        if self.__early_apply_local_log and not self.applied_local_log and self.raftLastApplied == self.raftCommitIndex:
+            self.applied_local_log = True
 
 
 class KVStoreTTL(DynMemberSyncObj):
 
-    def __init__(self, selfAddress, partnerAddrs, conf, on_set=None, on_delete=None):
+    def __init__(self, on_ready, on_set, on_delete, **config):
+        self.__thread = None
         self.__on_set = on_set
         self.__on_delete = on_delete
         self.__limb = {}
         self.__retry_timeout = None
-        self.__early_apply_local_log = selfAddress is not None
-        self.applied_local_log = False
-        super(KVStoreTTL, self).__init__(selfAddress, partnerAddrs, conf)
+
+        self_addr = config.get('self_addr')
+        partner_addrs = set(config.get('partner_addrs', []))
+        if config.get('patronictl'):
+            if self_addr:
+                partner_addrs.add(self_addr)
+            self_addr = None
+
+        # Create raft data_dir if necessary
+        raft_data_dir = config.get('data_dir', '')
+        if raft_data_dir != '':
+            validate_directory(raft_data_dir)
+
+        file_template = (self_addr or '')
+        file_template = file_template.replace(':', '_') if os.name == 'nt' else file_template
+        file_template = os.path.join(raft_data_dir, file_template)
+        conf = SyncObjConf(password=config.get('password'), autoTick=False, appendEntriesUseBatch=False,
+                           bindAddress=config.get('bind_addr'), commandsWaitLeader=config.get('commandsWaitLeader'),
+                           fullDumpFile=(file_template + '.dump' if self_addr else None),
+                           journalFile=(file_template + '.journal' if self_addr else None),
+                           onReady=on_ready, dynamicMembershipChange=True)
+
+        super(KVStoreTTL, self).__init__(self_addr, partner_addrs, conf)
         self.__data = {}
 
     @staticmethod
@@ -175,7 +151,7 @@ class KVStoreTTL(DynMemberSyncObj):
 
         if old_value and old_value['created'] != value['created']:
             value['created'] = value['updated']
-        value['index'] = self._SyncObj__raftLastApplied + 1
+        value['index'] = self.raftLastApplied + 1
 
         self.__data[key] = value
         if self.__on_set:
@@ -242,27 +218,29 @@ class KVStoreTTL(DynMemberSyncObj):
         return {k: v for k, v in self.__data.items() if k.startswith(key)}
 
     def _onTick(self, timeToWait=0.0):
-        # The SyncObj starts applying the local log only when there is at least one node connected.
-        # We want to change this behavior and apply the local log even when there is nobody except us.
-        # It gives us at least some picture about the last known cluster state.
-        if self.__early_apply_local_log and not self.applied_local_log and self._SyncObj__needLoadDumpFile:
-            self._SyncObj__raftCommitIndex = self._SyncObj__getCurrentLogIndex()
-            self._SyncObj__raftCurrentTerm = self._SyncObj__getCurrentLogTerm()
-
         super(KVStoreTTL, self)._onTick(timeToWait)
 
-        # The SyncObj calls onReady callback only when cluster got the leader and is ready for writes.
-        # In some cases for us it is safe to "signal" the Raft object when the local log is fully applied.
-        # We are using the `applied_local_log` property for that, but not calling the callback function.
-        if self.__early_apply_local_log and not self.applied_local_log and self._SyncObj__raftCommitIndex != 1 and \
-                self._SyncObj__raftLastApplied == self._SyncObj__raftCommitIndex:
-            self.applied_local_log = True
-
         if self._isLeader():
             self.__expire_keys()
         else:
             self.__limb.clear()
 
+    def _autoTickThread(self):
+        self.__destroying = False
+        while not self.__destroying:
+            self.doTick(self.conf.autoTickPeriod)
+
+    def startAutoTick(self):
+        self.__thread = threading.Thread(target=self._autoTickThread)
+        self.__thread.daemon = True
+        self.__thread.start()
+
+    def destroy(self):
+        if self.__thread:
+            self.__destroying = True
+            self.__thread.join()
+        super(KVStoreTTL, self).destroy()
+
 
 class Raft(AbstractDCS):
 
@@ -270,34 +248,16 @@ class Raft(AbstractDCS):
         super(Raft, self).__init__(config)
         self._ttl = int(config.get('ttl') or 30)
 
-        self_addr = config.get('self_addr')
-        partner_addrs = config.get('partner_addrs', [])
-        if self._ctl:
-            if self_addr:
-                partner_addrs.append(self_addr)
-            self_addr = None
-
-        # Create raft data_dir if necessary
-        raft_data_dir = config.get('data_dir', '')
-        if raft_data_dir != '':
-            validate_directory(raft_data_dir)
-
         ready_event = threading.Event()
-        file_template = os.path.join(config.get('data_dir', ''), (self_addr or ''))
-        conf = SyncObjConf(password=config.get('password'), appendEntriesUseBatch=False,
-                           bindAddress=config.get('bind_addr'), commandsWaitLeader=False,
-                           fullDumpFile=(file_template + '.dump' if self_addr else None),
-                           journalFile=(file_template + '.journal' if self_addr else None),
-                           onReady=ready_event.set, dynamicMembershipChange=True)
+        self._sync_obj = KVStoreTTL(ready_event.set, self._on_set, self._on_delete, commandsWaitLeader=False, **config)
+        self._sync_obj.startAutoTick()
 
-        self._sync_obj = KVStoreTTL(self_addr, partner_addrs, conf, self._on_set, self._on_delete)
         while True:
             ready_event.wait(5)
             if ready_event.isSet() or self._sync_obj.applied_local_log:
                 break
             else:
                 logger.info('waiting on raft')
-        self._sync_obj.forceLogCompaction()
         self.set_retry_timeout(int(config.get('retry_timeout') or 10))
 
     def _on_set(self, key, value):
diff --git a/patroni/dcs/zookeeper.py b/patroni/dcs/zookeeper.py
index 3be8adb4..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
@@ -186,11 +201,10 @@ class ZooKeeper(AbstractDCS):
         except NoNodeError:
             return []
 
-    def load_members(self, sync_standby):
+    def load_members(self):
         members = []
         for member in self.get_children(self.members_path, self.cluster_watcher):
-            watch = member in sync_standby and self.cluster_watcher or None
-            data = self.get_node(self.members_path + member, watch)
+            data = self.get_node(self.members_path + member)
             if data is not None:
                 members.append(self.member(member, *data))
         return members
@@ -218,8 +232,7 @@ class ZooKeeper(AbstractDCS):
         sync = SyncState.from_node(sync and sync[1].version, sync and sync[0])
 
         # get list of members
-        sync_standby = sync.leader == self._name and sync.members or []
-        members = self.load_members(sync_standby) if self._MEMBERS[:-1] in nodes else []
+        members = self.load_members() if self._MEMBERS[:-1] in nodes else []
 
         # get leader
         leader = self.get_node(self.leader_path) if self._LEADER in nodes else None
@@ -255,12 +268,13 @@ class ZooKeeper(AbstractDCS):
                 logger.exception('get_cluster')
                 self.cluster_watcher(None)
                 raise ZooKeeperError('ZooKeeper in not responding properly')
-        # Optime ZNode was updated or doesn't exist and we are not leader
+        # The /status ZNode was updated or doesn't exist and we are not leader
         elif (self._fetch_status and not self._fetch_cluster or not cluster.last_lsn
               or cluster.has_permanent_logical_slots(self._name, False) and not cluster.slots) and\
                 not (cluster.leader and cluster.leader.name == self._name):
             try:
                 last_lsn, slots = self.get_status(cluster.leader)
+                self.event.clear()
                 cluster = Cluster(cluster.initialize, cluster.config, cluster.leader, last_lsn,
                                   cluster.members, cluster.failover, cluster.sync, cluster.history, slots)
             except Exception:
@@ -396,7 +410,7 @@ class ZooKeeper(AbstractDCS):
         return self.set_sync_state_value("{}", index)
 
     def watch(self, leader_index, timeout):
-        ret = super(ZooKeeper, self).watch(leader_index, timeout)
+        ret = super(ZooKeeper, self).watch(leader_index, timeout + 0.5)
         if ret and not self._fetch_status:
             self._fetch_cluster = True
         return ret or self._fetch_cluster
diff --git a/patroni/ha.py b/patroni/ha.py
index 8d94a94d..ee1c78e8 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
@@ -1302,8 +1308,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()
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/bootstrap.py b/patroni/postgresql/bootstrap.py
index ca72fc59..2c328718 100644
--- a/patroni/postgresql/bootstrap.py
+++ b/patroni/postgresql/bootstrap.py
@@ -90,8 +90,8 @@ class Bootstrap(object):
         self._postgresql.configure_server_parameters()
 
         # make sure there is no trigger file or postgres will be automatically promoted
-        trigger_file = 'promote_trigger_file' if self._postgresql.major_version >= 120000 else 'trigger_file'
-        trigger_file = self._postgresql.config.get('recovery_conf', {}).get(trigger_file) or 'promote'
+        trigger_file = self._postgresql.config.triggerfile_good_name
+        trigger_file = (self._postgresql.config.get('recovery_conf') or {}).get(trigger_file) or 'promote'
         trigger_file = os.path.abspath(os.path.join(self._postgresql.data_dir, trigger_file))
         if os.path.exists(trigger_file):
             os.unlink(trigger_file)
diff --git a/patroni/postgresql/config.py b/patroni/postgresql/config.py
index 7ad794ef..c61a7695 100644
--- a/patroni/postgresql/config.py
+++ b/patroni/postgresql/config.py
@@ -336,8 +336,9 @@ class ConfigHandler(object):
         if "stats_temp_directory" in self._server_parameters:
             self.try_to_create_dir(self._server_parameters["stats_temp_directory"],
                                    "'{}' is defined in stats_temp_directory, {}")
-        self.try_to_create_dir(os.path.dirname(self._pgpass),
-                               "'{}' is defined in `postgresql.pgpass`, {}")
+        if not self._krbsrvname:
+            self.try_to_create_dir(os.path.dirname(self._pgpass),
+                                   "'{}' is defined in `postgresql.pgpass`, {}")
 
     @property
     def _configuration_to_save(self):
@@ -388,12 +389,6 @@ class ConfigHandler(object):
             os.rename(self._postgresql_conf, self._postgresql_base_conf)
 
         configuration = configuration or self._server_parameters.copy()
-
-        # In case we are using custom bootstrap from spilo image with PITR it fails if it contains increasing
-        # values like Max_connections. We disable hot_standby so it will accept increasing values.
-        if self._postgresql.bootstrap.running_custom_bootstrap:
-            configuration['hot_standby'] = 'off'
-
         # Due to the permanent logical replication slots configured we have to enable hot_standby_feedback
         if self._postgresql.enforce_hot_standby_feedback:
             configuration['hot_standby_feedback'] = 'on'
@@ -403,8 +398,8 @@ class ConfigHandler(object):
             f.writeline("include '{0}'\n".format(ConfigWriter.escape(include)))
             for name, value in sorted((configuration).items()):
                 value = transform_postgresql_parameter_value(self._postgresql.major_version, name, value)
-                if (not self._postgresql.bootstrap.running_custom_bootstrap or name != 'hba_file') \
-                        and name not in self._RECOVERY_PARAMETERS and value is not None:
+                if value is not None and\
+                        (name != 'hba_file' or not self._postgresql.bootstrap.running_custom_bootstrap):
                     f.write_param(name, value)
             # when we are doing custom bootstrap we assume that we don't know superuser password
             # and in order to be able to change it, we are opening trust access from a certain address
@@ -559,7 +554,7 @@ class ConfigHandler(object):
         return os.path.exists(self._recovery_conf)
 
     @property
-    def _triggerfile_good_name(self):
+    def triggerfile_good_name(self):
         return 'trigger_file' if self._postgresql.major_version < 120000 else 'promote_trigger_file'
 
     @property
@@ -707,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
@@ -762,13 +759,13 @@ class ConfigHandler(object):
         return env
 
     def write_recovery_conf(self, recovery_params):
+        self._recovery_params = recovery_params
         if self._postgresql.major_version >= 120000:
             if parse_bool(recovery_params.pop('standby_mode', None)):
                 open(self._standby_signal, 'w').close()
             else:
                 self._remove_file_if_exists(self._standby_signal)
                 open(self._recovery_signal, 'w').close()
-            self._recovery_params = recovery_params
         else:
             with ConfigWriter(self._recovery_conf) as f:
                 os.chmod(self._recovery_conf, stat.S_IWRITE | stat.S_IREAD)
@@ -812,8 +809,8 @@ class ConfigHandler(object):
 
         if self.get('recovery_conf'):
             value = self._config['recovery_conf'].pop(self._triggerfile_wrong_name, None)
-            if self._triggerfile_good_name not in self._config['recovery_conf'] and value:
-                self._config['recovery_conf'][self._triggerfile_good_name] = value
+            if self.triggerfile_good_name not in self._config['recovery_conf'] and value:
+                self._config['recovery_conf'][self.triggerfile_good_name] = value
 
     def get_server_parameters(self, config):
         parameters = config['parameters'].copy()
@@ -882,25 +879,21 @@ class ConfigHandler(object):
     def resolve_connection_addresses(self):
         port = self._server_parameters['port']
         tcp_local_address = self._get_tcp_local_address()
+        netloc = self._config.get('connect_address') or tcp_local_address + ':' + port
 
-        local_address = {'port': port}
-        if self._config.get('use_unix_socket'):
-            unix_socket_directories = self._server_parameters.get('unix_socket_directories')
-            if unix_socket_directories is not None:
-                # fallback to tcp if unix_socket_directories is set, but there are no sutable values
-                local_address['host'] = self._get_unix_local_address(unix_socket_directories) or tcp_local_address
+        unix_local_address = {'port': port}
+        unix_socket_directories = self._server_parameters.get('unix_socket_directories')
+        if unix_socket_directories is not None:
+            # fallback to tcp if unix_socket_directories is set, but there are no sutable values
+            unix_local_address['host'] = self._get_unix_local_address(unix_socket_directories) or tcp_local_address
 
-            # if unix_socket_directories is not specified, but use_unix_socket is set to true - do our best
-            # to use default value, i.e. don't specify a host neither in connection url nor arguments
-        else:
-            local_address['host'] = tcp_local_address
+        tcp_local_address = {'host': tcp_local_address, 'port': port}
 
-        self._local_address = local_address
-        self.local_replication_address = {'host': tcp_local_address, 'port': port}
+        self._local_address = unix_local_address if self._config.get('use_unix_socket') else tcp_local_address
+        self.local_replication_address = unix_local_address\
+            if self._config.get('use_unix_socket_repl') else tcp_local_address
 
-        netloc = self._config.get('connect_address') or tcp_local_address + ':' + port
         self._postgresql.connection_string = uri('postgres', netloc, self._postgresql.database)
-
         self._postgresql.set_connection_kwargs(self.local_connect_kwargs)
 
     def _get_pg_settings(self, names):
@@ -1073,6 +1066,14 @@ class ConfigHandler(object):
             if cvalue > value:
                 effective_configuration[name] = cvalue
                 self._postgresql.set_pending_restart(True)
+
+        # If we are using custom bootstrap with PITR it could fail when values
+        # like max_connections are increased, therefore we disable hot_standby.
+        if self._postgresql.bootstrap.running_custom_bootstrap and \
+                (self._postgresql.bootstrap.keep_existing_recovery_conf or self._recovery_conf):
+            effective_configuration['hot_standby'] = 'off'
+            self._postgresql.set_pending_restart(True)
+
         return effective_configuration
 
     @property
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/slots.py b/patroni/postgresql/slots.py
index cee319d9..43b7e30c 100644
--- a/patroni/postgresql/slots.py
+++ b/patroni/postgresql/slots.py
@@ -5,6 +5,7 @@ import shutil
 
 from collections import defaultdict
 from contextlib import contextmanager
+from psycopg2.errors import UndefinedFile
 
 from .connection import get_connection_cursor
 from .misc import format_lsn
@@ -198,7 +199,9 @@ class SlotsHandler(object):
                         cur.execute("SELECT pg_catalog.pg_replication_slot_advance(%s, %s)",
                                     (name, format_lsn(int(cluster.slots[name]))))
                     except Exception as e:
-                        logger.exception("Failed to advance logical replication slot '%s': %r", name, e)
+                        logger.error("Failed to advance logical replication slot '%s': %r", name, e)
+                        if isinstance(e, UndefinedFile):
+                            create_slots.append(name)
                         self._schedule_load_slots = True
         return create_slots
 
@@ -280,6 +283,8 @@ class SlotsHandler(object):
                     f.write(value['data'])
                     f.flush()
                     os.fsync(f.fileno())
+                if os.path.exists(slot_dir):
+                    shutil.rmtree(slot_dir)
                 os.rename(slot_tmp_dir, slot_dir)
                 fsync_dir(slot_dir)
                 self._unready_logical_slots.add(name)
diff --git a/patroni/postgresql/validator.py b/patroni/postgresql/validator.py
index 886beac9..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'),
@@ -511,6 +528,8 @@ def _transform_parameter_value(validators, version, name, value):
 def transform_postgresql_parameter_value(version, name, value):
     if '.' in name:
         return value
+    if name in recovery_parameters:
+        return None
     return _transform_parameter_value(parameters, version, name, value)
 
 
diff --git a/patroni/raft_controller.py b/patroni/raft_controller.py
index 5a9ae704..1baa9786 100644
--- a/patroni/raft_controller.py
+++ b/patroni/raft_controller.py
@@ -1,9 +1,7 @@
 import logging
-import os
 
-from patroni.daemon import AbstractPatroniDaemon, abstract_main
-from patroni.dcs.raft import KVStoreTTL
-from pysyncobj import SyncObjConf
+from .daemon import AbstractPatroniDaemon, abstract_main
+from .dcs.raft import KVStoreTTL
 
 logger = logging.getLogger(__name__)
 
@@ -13,16 +11,13 @@ class RaftController(AbstractPatroniDaemon):
     def __init__(self, config):
         super(RaftController, self).__init__(config)
 
-        raft_config = self.config.get('raft')
-        self_addr = raft_config['self_addr']
-        template = os.path.join(raft_config.get('data_dir', ''), self_addr)
-        self._syncobj_config = SyncObjConf(autoTick=False, appendEntriesUseBatch=False, dynamicMembershipChange=True,
-                                           journalFile=template + '.journal', fullDumpFile=template + '.dump')
-        self._raft = KVStoreTTL(self_addr, raft_config.get('partner_addrs', []), self._syncobj_config)
+        config = self.config.get('raft')
+        assert 'self_addr' in config
+        self._raft = KVStoreTTL(None, None, None, **config)
 
     def _run_cycle(self):
         try:
-            self._raft.doTick(self._syncobj_config.autoTickPeriod)
+            self._raft.doTick(self._raft.conf.autoTickPeriod)
         except Exception:
             logger.exception('doTick')
 
diff --git a/postgres0.yml b/postgres0.yml
index 6137cbf4..4e081de3 100644
--- a/postgres0.yml
+++ b/postgres0.yml
@@ -57,10 +57,15 @@ bootstrap:
       parameters:
 #        wal_level: hot_standby
 #        hot_standby: "on"
+#        max_connections: 100
+#        max_worker_processes: 8
 #        wal_keep_segments: 8
 #        max_wal_senders: 10
 #        max_replication_slots: 10
+#        max_prepared_transactions: 0
+#        max_locks_per_transaction: 64
 #        wal_log_hints: "on"
+#        track_commit_timestamp: "off"
 #        archive_mode: "on"
 #        archive_timeout: 1800s
 #        archive_command: mkdir -p ../wal_archive && test ! -f ../wal_archive/%f && cp %p ../wal_archive/%f
diff --git a/postgres1.yml b/postgres1.yml
index 49a0ed71..15a6897a 100644
--- a/postgres1.yml
+++ b/postgres1.yml
@@ -51,10 +51,15 @@ bootstrap:
       parameters:
 #        wal_level: hot_standby
 #        hot_standby: "on"
+#        max_connections: 100
+#        max_worker_processes: 8
 #        wal_keep_segments: 8
 #        max_wal_senders: 10
 #        max_replication_slots: 10
+#        max_prepared_transactions: 0
+#        max_locks_per_transaction: 64
 #        wal_log_hints: "on"
+#        track_commit_timestamp: "off"
 #        archive_mode: "on"
 #        archive_timeout: 1800s
 #        archive_command: mkdir -p ../wal_archive && test ! -f ../wal_archive/%f && cp %p ../wal_archive/%f
diff --git a/postgres2.yml b/postgres2.yml
index b5bb863f..829209ff 100644
--- a/postgres2.yml
+++ b/postgres2.yml
@@ -51,10 +51,15 @@ bootstrap:
       parameters:
 #        wal_level: hot_standby
 #        hot_standby: "on"
+#        max_connections: 100
+#        max_worker_processes: 8
 #        wal_keep_segments: 8
 #        max_wal_senders: 10
 #        max_replication_slots: 10
+#        max_prepared_transactions: 0
+#        max_locks_per_transaction: 64
 #        wal_log_hints: "on"
+#        track_commit_timestamp: "off"
 #        archive_mode: "on"
 #        archive_timeout: 1800s
 #        archive_command: mkdir -p ../wal_archive && test ! -f ../wal_archive/%f && cp %p ../wal_archive/%f
diff --git a/requirements.txt b/requirements.txt
index 9fc6f4b4..c7438d3d 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -9,6 +9,7 @@ python-consul>=0.7.1
 click>=4.1
 prettytable>=0.7
 python-dateutil
-pysyncobj>=0.3.7
+pysyncobj>=0.3.8
+cryptography>=1.4
 psutil>=2.0.0
 ydiff>=1.2.0
diff --git a/setup.py b/setup.py
index 17dd696b..7302858a 100644
--- a/setup.py
+++ b/setup.py
@@ -22,8 +22,9 @@ AUTHOR_EMAIL = 'alexander.kukushkin@zalando.de, dmitrii.dolgov@zalando.de, alexk
 KEYWORDS = 'etcd governor patroni postgresql postgres ha haproxy confd' +\
     ' zookeeper exhibitor consul streaming replication kubernetes k8s'
 
-EXTRAS_REQUIRE = {'aws': ['boto'], 'etcd': ['python-etcd'], 'etcd3': ['python-etcd'], 'consul': ['python-consul'],
-                  'exhibitor': ['kazoo'], 'zookeeper': ['kazoo'], 'kubernetes': ['ipaddress'], 'raft': ['pysyncobj']}
+EXTRAS_REQUIRE = {'aws': ['boto'], 'etcd': ['python-etcd'], 'etcd3': ['python-etcd'],
+                  'consul': ['python-consul'], 'exhibitor': ['kazoo'], 'zookeeper': ['kazoo'],
+                  'kubernetes': ['ipaddress'], 'raft': ['pysyncobj', 'cryptography']}
 COVERAGE_XML = True
 COVERAGE_HTML = False
 
@@ -171,10 +172,16 @@ def setup_package(version):
         if r == '':
             continue
         extra = False
-        for e, v in EXTRAS_REQUIRE.items():
-            if v and r.startswith(v[0]):
-                EXTRAS_REQUIRE[e] = [r] if e != 'kubernetes' or sys.version_info < (3, 0, 0) else []
-                extra = True
+        for e, deps in EXTRAS_REQUIRE.items():
+            for i, v in enumerate(deps):
+                if r.startswith(v):
+                    if e != 'kubernetes' or sys.version_info < (3, 0, 0):
+                        deps[i] = r
+                    else:
+                        deps = []
+                    EXTRAS_REQUIRE[e] = deps
+                    extra = True
+                    break
         if not extra:
             install_requires.append(r)
 


Test patch:
diff --git a/.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/.github/workflows/tests.yaml b/.github/workflows/tests.yaml
index ccc5f7ed..f83fddef 100644
--- a/.github/workflows/tests.yaml
+++ b/.github/workflows/tests.yaml
@@ -115,6 +115,8 @@ jobs:
       uses: actions/setup-python@v2
       with:
         python-version: ${{ matrix.python-version }}
+    - name: Add postgresql apt repo
+      run: sudo sh -c 'echo "deb http://apt.postgresql.org/pub/repos/apt $(lsb_release -cs)-pgdg main" > /etc/apt/sources.list.d/pgdg.list'
     - name: Install dependencies
       run: python .github/workflows/install_deps.py
     - name: Run behave tests
diff --git a/tests/test_api.py b/tests/test_api.py
index 079753f4..4b4d8ebf 100644
--- a/tests/test_api.py
+++ b/tests/test_api.py
@@ -118,7 +118,7 @@ class MockPatroni(object):
     postgresql = ha.state_handler
     dcs = Mock()
     logger = MockLogger()
-    tags = {}
+    tags = {"key1": True, "key2": False, "key3": 1, "key4": 1.4, "key5": "RandomTag"}
     version = '0.00'
     noloadbalance = PropertyMock(return_value=False)
     scheduled_restart = {'schedule': future_restart_time,
@@ -174,7 +174,7 @@ class TestRestApiHandler(unittest.TestCase):
             MockRestApiServer(RestApiHandler, 'GET /replica')
         with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={'state': 'running'})):
             MockRestApiServer(RestApiHandler, 'GET /health')
-        MockRestApiServer(RestApiHandler, 'GET /master')
+        MockRestApiServer(RestApiHandler, 'GET /leader')
         MockPatroni.dcs.cluster.sync.members = [MockPostgresql.name]
         MockPatroni.dcs.cluster.is_synchronous_mode = Mock(return_value=True)
         with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={'role': 'replica'})):
@@ -197,6 +197,89 @@ class TestRestApiHandler(unittest.TestCase):
         with patch.object(MockHa, 'is_standby_cluster', Mock(return_value=True)):
             MockRestApiServer(RestApiHandler, 'GET /standby_leader')
 
+        # test tags
+        #
+        MockRestApiServer(RestApiHandler, 'GET /master?lag=1M&'
+                                          'tag_key1=true&tag_key2=false&'
+                                          'tag_key3=1&tag_key4=1.4&tag_key5=RandomTag')
+        MockRestApiServer(RestApiHandler, 'GET /master?lag=1M&'
+                                          'tag_key1=true&tag_key2=False&'
+                                          'tag_key3=1&tag_key4=1.4&tag_key5=RandomTag')
+        MockRestApiServer(RestApiHandler, 'GET /master?lag=1M&'
+                                          'tag_key1=true&tag_key2=false&'
+                                          'tag_key3=1.0&tag_key4=1.4&tag_key5=RandomTag')
+        MockRestApiServer(RestApiHandler, 'GET /master?lag=1M&'
+                                          'tag_key1=true&tag_key2=false&'
+                                          'tag_key3=1&tag_key4=1.4&tag_key5=RandomTag&tag_key6=RandomTag2')
+        #
+        with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={'role': 'master'})):
+            MockRestApiServer(RestApiHandler, 'GET /master?lag=1M&'
+                                              'tag_key1=true&tag_key2=false&'
+                                              'tag_key3=1&tag_key4=1.4&tag_key5=RandomTag')
+            MockRestApiServer(RestApiHandler, 'GET /master?lag=1M&'
+                                              'tag_key1=true&tag_key2=False&'
+                                              'tag_key3=1&tag_key4=1.4&tag_key5=RandomTag')
+            MockRestApiServer(RestApiHandler, 'GET /master?lag=1M&'
+                                              'tag_key1=true&tag_key2=false&'
+                                              'tag_key3=1.0&tag_key4=1.4&tag_key5=RandomTag')
+            MockRestApiServer(RestApiHandler, 'GET /master?lag=1M&'
+                                              'tag_key1=true&tag_key2=false&'
+                                              'tag_key3=1&tag_key4=1.4&tag_key5=RandomTag&tag_key6=RandomTag2')
+        #
+        with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={'role': 'standby_leader'})):
+            MockRestApiServer(RestApiHandler, 'GET /standby_leader?lag=1M&'
+                                              'tag_key1=true&tag_key2=false&'
+                                              'tag_key3=1&tag_key4=1.4&tag_key5=RandomTag')
+            MockRestApiServer(RestApiHandler, 'GET /standby_leader?lag=1M&'
+                                              'tag_key1=true&tag_key2=False&'
+                                              'tag_key3=1&tag_key4=1.4&tag_key5=RandomTag')
+            MockRestApiServer(RestApiHandler, 'GET /standby_leader?lag=1M&'
+                                              'tag_key1=true&tag_key2=false&'
+                                              'tag_key3=1.0&tag_key4=1.4&tag_key5=RandomTag')
+            MockRestApiServer(RestApiHandler, 'GET /standby_leader?lag=1M&'
+                                              'tag_key1=true&tag_key2=false&'
+                                              'tag_key3=1&tag_key4=1.4&tag_key5=RandomTag&tag_key6=RandomTag2')
+        #
+        MockRestApiServer(RestApiHandler, 'GET /replica?lag=1M&'
+                                          'tag_key1=true&tag_key2=false&'
+                                          'tag_key3=1&tag_key4=1.4&tag_key5=RandomTag')
+        MockRestApiServer(RestApiHandler, 'GET /replica?lag=1M&'
+                                          'tag_key1=true&tag_key2=False&'
+                                          'tag_key3=1&tag_key4=1.4&tag_key5=RandomTag')
+        MockRestApiServer(RestApiHandler, 'GET /replica?lag=1M&'
+                                          'tag_key1=true&tag_key2=false&'
+                                          'tag_key3=1.0&tag_key4=1.4&tag_key5=RandomTag')
+        MockRestApiServer(RestApiHandler, 'GET /replica?lag=1M&'
+                                          'tag_key1=true&tag_key2=false&'
+                                          'tag_key3=1&tag_key4=1.4&tag_key5=RandomTag&tag_key6=RandomTag2')
+        #
+        with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={'role': 'master'})):
+            MockRestApiServer(RestApiHandler, 'GET /replica?lag=1M&'
+                                              'tag_key1=true&tag_key2=false&'
+                                              'tag_key3=1&tag_key4=1.4&tag_key5=RandomTag')
+            MockRestApiServer(RestApiHandler, 'GET /replica?lag=1M&'
+                                              'tag_key1=true&tag_key2=False&'
+                                              'tag_key3=1&tag_key4=1.4&tag_key5=RandomTag')
+            MockRestApiServer(RestApiHandler, 'GET /replica?lag=1M&'
+                                              'tag_key1=true&tag_key2=false&'
+                                              'tag_key3=1.0&tag_key4=1.4&tag_key5=RandomTag')
+            MockRestApiServer(RestApiHandler, 'GET /replica?lag=1M&'
+                                              'tag_key1=true&tag_key2=false&'
+                                              'tag_key3=1&tag_key4=1.4&tag_key5=RandomTag&tag_key6=RandomTag2')
+        #
+        MockRestApiServer(RestApiHandler, 'GET /read-write?lag=1M&'
+                                          'tag_key1=true&tag_key2=false&'
+                                          'tag_key3=1&tag_key4=1.4&tag_key5=RandomTag')
+        MockRestApiServer(RestApiHandler, 'GET /read-write?lag=1M&'
+                                          'tag_key1=true&tag_key2=False&'
+                                          'tag_key3=1&tag_key4=1.4&tag_key5=RandomTag')
+        MockRestApiServer(RestApiHandler, 'GET /read-write?lag=1M&'
+                                          'tag_key1=true&tag_key2=false&'
+                                          'tag_key3=1.0&tag_key4=1.4&tag_key5=RandomTag')
+        MockRestApiServer(RestApiHandler, 'GET /read-write?lag=1M&'
+                                          'tag_key1=true&tag_key2=false&'
+                                          'tag_key3=1&tag_key4=1.4&tag_key5=RandomTag&tag_key6=RandomTag2')
+
     def test_do_OPTIONS(self):
         self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'OPTIONS / HTTP/1.0'))
 
@@ -507,3 +590,7 @@ class TestRestApiServer(unittest.TestCase):
                 Mock(return_value=(mock_request, mock_address))
             ):
                 self.srv._handle_request_noblock()
+
+    @patch('ssl._ssl._test_decode_cert', Mock())
+    def test_reload_local_certificate(self):
+        self.assertTrue(self.srv.reload_local_certificate())
diff --git a/tests/test_etcd3.py b/tests/test_etcd3.py
index abc593af..b0ae1d05 100644
--- a/tests/test_etcd3.py
+++ b/tests/test_etcd3.py
@@ -4,8 +4,9 @@ import unittest
 import urllib3
 
 from mock import Mock, patch
-from patroni.dcs.etcd3 import PatroniEtcd3Client, Cluster, Etcd3, Etcd3Error, Etcd3ClientError, RetryFailedError,\
-        InvalidAuthToken, Unavailable, Unknown, UnsupportedEtcdVersion, UserEmpty, AuthFailed, base64_encode
+from patroni.dcs.etcd import DnsCachingResolver
+from patroni.dcs.etcd3 import PatroniEtcd3Client, Cluster, Etcd3Client, Etcd3Error, Etcd3ClientError, RetryFailedError,\
+        InvalidAuthToken, Unavailable, Unknown, UnsupportedEtcdVersion, UserEmpty, AuthFailed, base64_encode, Etcd3
 from threading import Thread
 
 from . import SleepException, MockResponse
@@ -57,6 +58,16 @@ def mock_urlopen(self, method, url, **kwargs):
     return ret
 
 
+class TestEtcd3Client(unittest.TestCase):
+
+    @patch.object(Thread, 'start', Mock())
+    @patch.object(urllib3.PoolManager, 'urlopen', mock_urlopen)
+    def test_authenticate(self):
+        etcd3 = Etcd3Client({'host': '127.0.0.1', 'port': 2379, 'use_proxies': True, 'retry_timeout': 10},
+                            DnsCachingResolver())
+        self.assertIsNotNone(etcd3._cluster_version)
+
+
 class BaseTestEtcd3(unittest.TestCase):
 
     @patch.object(Thread, 'start', Mock())
diff --git a/tests/test_exhibitor.py b/tests/test_exhibitor.py
index 9d4d460d..fdd3e993 100644
--- a/tests/test_exhibitor.py
+++ b/tests/test_exhibitor.py
@@ -24,7 +24,7 @@ class TestExhibitor(unittest.TestCase):
 
     @patch('urllib3.PoolManager.request', Mock(return_value=urllib3.HTTPResponse(
         status=200, body=b'{"servers":["127.0.0.1","127.0.0.2","127.0.0.3"],"port":2181}')))
-    @patch('patroni.dcs.zookeeper.KazooClient', MockKazooClient)
+    @patch('patroni.dcs.zookeeper.PatroniKazooClient', MockKazooClient)
     def setUp(self):
         self.e = Exhibitor({'hosts': ['localhost', 'exhibitor'], 'port': 8181, 'scope': 'test',
                             'name': 'foo', 'ttl': 30, 'retry_timeout': 10})
diff --git a/tests/test_ha.py b/tests/test_ha.py
index ed534b27..726addc3 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={}))
@@ -300,7 +301,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 +330,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 +355,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 +370,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 +389,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 +406,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 +517,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 +546,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 +576,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 +737,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 +746,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 +840,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 +1066,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 +1101,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 +1114,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 +1126,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 479554fe..96f0b5bc 100644
--- a/tests/test_postgresql.py
+++ b/tests/test_postgresql.py
@@ -11,6 +11,7 @@ from patroni.async_executor import CriticalTask
 from patroni.dcs import Cluster, RemoteMember, SyncState
 from patroni.exceptions import PostgresConnectionException, PatroniException
 from patroni.postgresql import Postgresql, STATE_REJECT, STATE_NO_RESPONSE
+from patroni.postgresql.bootstrap import Bootstrap
 from patroni.postgresql.postmaster import PostmasterProcess
 from patroni.utils import RetryFailedError
 from six.moves import builtins
@@ -223,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']
@@ -258,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'}})
@@ -319,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())
@@ -474,7 +504,8 @@ class TestPostgresql(BaseTestPostgresql):
         parameters = self._PARAMETERS.copy()
         parameters.pop('f.oo')
         parameters['wal_buffers'] = '512'
-        config = {'pg_hba': [''], 'pg_ident': [''], 'use_unix_socket': True, 'authentication': {},
+        config = {'pg_hba': [''], 'pg_ident': [''], 'use_unix_socket': True, 'use_unix_socket_repl': True,
+                  'authentication': {},
                   'retry_timeout': 10, 'listen': '*', 'krbsrvname': 'postgres', 'parameters': parameters}
         self.p.reload_config(config)
         mock_fetchone.side_effect = Exception
@@ -490,6 +521,14 @@ class TestPostgresql(BaseTestPostgresql):
         self.p.reload_config(config)
         self.p.config.resolve_connection_addresses()
 
+    def test_resolve_connection_addresses(self):
+        self.p.config._config['use_unix_socket'] = self.p.config._config['use_unix_socket_repl'] = True
+        self.p.config.resolve_connection_addresses()
+        self.assertEqual(self.p.config.local_replication_address, {'host': '/tmp', 'port': '5432'})
+        self.p.config._server_parameters.pop('unix_socket_directories')
+        self.p.config.resolve_connection_addresses()
+        self.assertEqual(self.p.config._local_address, {'port': '5432'})
+
     @patch.object(Postgresql, '_version_file_exists', Mock(return_value=True))
     def test_get_major_version(self):
         with patch.object(builtins, 'open', mock_open(read_data='9.4')):
@@ -684,6 +723,8 @@ class TestPostgresql(BaseTestPostgresql):
         self.assertEqual(self.p.get_master_timeline(), 1)
 
     @patch.object(Postgresql, 'get_postgres_role_from_data_directory', Mock(return_value='replica'))
+    @patch.object(Bootstrap, 'running_custom_bootstrap', PropertyMock(return_value=True))
+    @patch.object(Bootstrap, 'keep_existing_recovery_conf', PropertyMock(return_value=True))
     def test__build_effective_configuration(self):
         with patch.object(Postgresql, 'controldata',
                           Mock(return_value={'max_connections setting': '200',
@@ -713,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 54197df8..716295da 100644
--- a/tests/test_raft.py
+++ b/tests/test_raft.py
@@ -3,13 +3,13 @@ import unittest
 import tempfile
 import time
 
-from mock import Mock, patch
+from mock import Mock, PropertyMock, patch
 from patroni.dcs.raft import DynMemberSyncObj, KVStoreTTL, Raft, SyncObjUtility
 from pysyncobj import SyncObjConf, FAIL_REASON
 
 
 def remove_files(prefix):
-    for f in ('journal', 'dump'):
+    for f in ('journal', 'journal.meta', 'dump'):
         f = prefix + f
         if os.path.isfile(f):
             for i in range(0, 15):
@@ -31,50 +31,38 @@ class TestDynMemberSyncObj(unittest.TestCase):
         self.conf = SyncObjConf(appendEntriesUseBatch=False, dynamicMembershipChange=True, autoTick=False)
         self.so = DynMemberSyncObj('127.0.0.1:1234', ['127.0.0.1:1235'], self.conf)
 
-    @patch.object(SyncObjUtility, 'sendMessage')
-    def test_add_member(self, mock_send_message):
-        mock_send_message.return_value = [{'addr': '127.0.0.1:1235'}, {'addr': '127.0.0.1:1236'}]
-        mock_send_message.ver = 0
+    @patch.object(SyncObjUtility, 'executeCommand')
+    def test_add_member(self, mock_execute_command):
+        mock_execute_command.return_value = [{'addr': '127.0.0.1:1235'}, {'addr': '127.0.0.1:1236'}]
+        mock_execute_command.ver = 0
         DynMemberSyncObj('127.0.0.1:1234', ['127.0.0.1:1235'], self.conf)
         self.conf.dynamicMembershipChange = False
         DynMemberSyncObj('127.0.0.1:1234', ['127.0.0.1:1235'], self.conf)
 
-    def test___onUtilityMessage(self):
-        self.so._SyncObj__encryptor = Mock()
+    def test_getMembers(self):
         mock_conn = Mock()
-        mock_conn.sendRandKey = None
-        self.so._SyncObj__transport._onIncomingMessageReceived(mock_conn, 'randkey')
         self.so._SyncObj__transport._onIncomingMessageReceived(mock_conn, ['members'])
-        self.so._SyncObj__transport._onIncomingMessageReceived(mock_conn, ['status'])
 
     def test__SyncObj__doChangeCluster(self):
         self.so._SyncObj__doChangeCluster(['add', '127.0.0.1:1236'])
 
-    def test_utility(self):
-        utility = SyncObjUtility(['127.0.0.1:1235'], self.conf)
-        utility.setPartnerNode(list(utility._SyncObj__otherNodes)[0])
-        utility.sendMessage(['members'])
-        utility._onMessageReceived(0, '')
-
 
+@patch.object(SyncObjConf, 'fullDumpFile', PropertyMock(return_value=None), create=True)
+@patch.object(SyncObjConf, 'journalFile', PropertyMock(return_value=None), create=True)
 class TestKVStoreTTL(unittest.TestCase):
 
+    @patch.object(SyncObjConf, 'fullDumpFile', PropertyMock(return_value=None), create=True)
+    @patch.object(SyncObjConf, 'journalFile', PropertyMock(return_value=None), create=True)
     def setUp(self):
-        self.conf = SyncObjConf(appendEntriesUseBatch=False, appendEntriesPeriod=0.001,
-                                raftMinTimeout=0.004, raftMaxTimeout=0.005, autoTickPeriod=0.001)
         callback = Mock()
         callback.replicated = False
-        self.so = KVStoreTTL('127.0.0.1:1234', [], self.conf, on_set=callback, on_delete=callback)
+        self.so = KVStoreTTL(None, callback, callback, self_addr='127.0.0.1:1234')
+        self.so.startAutoTick()
         self.so.set_retry_timeout(10)
 
-    @staticmethod
-    def destroy(so):
-        so.destroy()
-        so._SyncObj__thread.join()
-
     def tearDown(self):
         if self.so:
-            self.destroy(self.so)
+            self.so.destroy()
 
     def test_set(self):
         self.assertTrue(self.so.set('foo', 'bar', prevExist=False, ttl=30))
@@ -83,7 +71,7 @@ class TestKVStoreTTL(unittest.TestCase):
         self.assertTrue(self.so.retry(self.so._set, 'foo', {'value': 'buz', 'created': 1, 'updated': 1}))
 
     def test_delete(self):
-        self.conf.autoTickPeriod = 0.1
+        self.so.autoTickPeriod = 0.2
         self.so.set('foo', 'bar')
         self.so.set('fooo', 'bar')
         self.assertFalse(self.so.delete('foo', prevValue='buz'))
@@ -111,11 +99,10 @@ class TestKVStoreTTL(unittest.TestCase):
 
     def test_on_ready_override(self):
         self.assertTrue(self.so.set('foo', 'bar'))
-        self.destroy(self.so)
+        self.so.destroy()
         self.so = None
-        self.conf.onReady = Mock()
-        self.conf.autoTick = False
-        so = KVStoreTTL('127.0.0.1:1234', ['127.0.0.1:1235'], self.conf)
+        so = KVStoreTTL(Mock(), None, None, self_addr='127.0.0.1:1234',
+                        partner_addrs=['127.0.0.1:1235'], patronictl=True)
         so.doTick(0)
         so.destroy()
 
@@ -150,7 +137,6 @@ class TestRaft(unittest.TestCase):
         self.assertTrue(raft.take_leader())
         raft.watch(None, 0.001)
         raft._sync_obj.destroy()
-        raft._sync_obj._SyncObj__thread.join()
 
     def tearDown(self):
         remove_files(os.path.join(self._TMP, '127.0.0.1:1234.'))
diff --git a/tests/test_slots.py b/tests/test_slots.py
index e17c4779..8c281ce7 100644
--- a/tests/test_slots.py
+++ b/tests/test_slots.py
@@ -86,8 +86,8 @@ class TestSlotsHandler(BaseTestPostgresql):
                           [self.me, self.other, self.leadermem], None, None, None, {'ls': 12346})
         self.assertEqual(self.s.sync_replication_slots(cluster, False), [])
         self.s._schedule_load_slots = False
-        with patch.object(MockCursor, 'execute', Mock(side_effect=psycopg2.OperationalError)):
-            self.assertEqual(self.s.sync_replication_slots(cluster, False), [])
+        with patch.object(MockCursor, 'execute', Mock(side_effect=psycopg2.errors.UndefinedFile)):
+            self.assertEqual(self.s.sync_replication_slots(cluster, False), ['ls'])
         cluster.slots['ls'] = 'a'
         self.assertEqual(self.s.sync_replication_slots(cluster, False), [])
         with patch.object(MockCursor, 'rowcount', PropertyMock(return_value=1), create=True):
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})