Release 2.1.4 (#2322 )

- bump version - update release notes - implement missing unit-tests
Add /read-only-sync endpoint (#2305 ) (#2311 )
2026-08-26 23:50:23 +00:00 · 2022-06-01 16:00:56 +02:00 · 2022-05-30 17:09:43 +02:00 · 2022-05-20 10:24:49 +02:00 · 2022-05-20 10:24:25 +02:00 · 2022-05-19 15:36:09 +02:00
53 changed files with 942 additions and 547 deletions
@@ -110,7 +110,7 @@ def install_etcd():


 def install_postgres():
-    version = os.environ.get('PGVERSION', '12.1-1')
+    version = os.environ.get('PGVERSION', '14.1-1')
    platform = {'darwin': 'osx', 'win32': 'windows-x64', 'cygwin': 'windows-x64'}[sys.platform]
    name = 'postgresql-{0}-{1}-binaries.zip'.format(version, platform)
    get_file('http://get.enterprisedb.com/postgresql/' + name, name)
@@ -27,7 +27,7 @@ def main():

        version = versions.get(what)
        path = '/usr/lib/postgresql/{0}/bin:.'.format(version)
-        unbuffer = ['timeout', '600', 'unbuffer']
+        unbuffer = ['timeout', '900', 'unbuffer']
        args = ['--tags=-skip'] if what == 'etcd' else []
    else:
        path = os.path.abspath(os.path.join('pgsql', 'bin'))
@@ -30,15 +30,6 @@ jobs:
      run: python .github/workflows/run_tests.py
      if: matrix.os != 'windows'

-    - name: Set up Python 3.5
-      uses: actions/setup-python@v2
-      with:
-        python-version: 3.5
-    - name: Install dependencies
-      run: python .github/workflows/install_deps.py
-    - name: Run tests and flake8
-      run: python .github/workflows/run_tests.py
-
    - name: Set up Python 3.6
      uses: actions/setup-python@v2
      with:
@@ -75,6 +66,15 @@ jobs:
    - name: Run tests and flake8
      run: python .github/workflows/run_tests.py

+    - name: Set up Python 3.10
+      uses: actions/setup-python@v2
+      with:
+        python-version: '3.10'
+    - name: Install dependencies
+      run: python .github/workflows/install_deps.py
+    - name: Run tests and flake8
+      run: python .github/workflows/run_tests.py
+
    - name: Combine coverage
      run: python .github/workflows/run_tests.py combine

@@ -88,26 +88,31 @@ jobs:
        GITHUB_TOKEN: ${{ secrets.github_token }}
      run: python -m coveralls --service=github

-    - name: Run codacy-coverage-reporter
-      uses: codacy/codacy-coverage-reporter-action@master
-      env:
-        SECRETS_AVAILABLE: ${{ secrets.CODACY_PROJECT_TOKEN != '' }}
-      with:
-        project-token: ${{ secrets.CODACY_PROJECT_TOKEN }}
-        coverage-reports: coverage.xml
-      if: ${{ matrix.os == 'ubuntu' && env.SECRETS_AVAILABLE == 'true' }}
-
  behave:
    runs-on: ${{ matrix.os }}-latest
    env:
      DCS: ${{ matrix.dcs }}
      ETCDVERSION: 3.3.13
+      PGVERSION: 12.1-1  # for windows and macos
    strategy:
      fail-fast: false
      matrix:
        os: [ubuntu]
-        python-version: [2.7, 3.5, 3.8]
+        python-version: [2.7, 3.6, 3.9]
        dcs: [etcd, etcd3, consul, exhibitor, kubernetes, raft]
+        exclude:
+        - dcs: kubernetes
+          python-version: 2.7
+        include:
+        - os: macos
+          python-version: 3.7
+          dcs: raft
+        - os: macos
+          python-version: 3.8
+          dcs: etcd
+        - os: macos
+          python-version: '3.10'
+          dcs: etcd3

    steps:
    - uses: actions/checkout@v1
@@ -117,45 +122,14 @@ jobs:
        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'
+      if: matrix.os == 'ubuntu'
    - name: Install dependencies
      run: python .github/workflows/install_deps.py
    - name: Run behave tests
      run: python .github/workflows/run_tests.py
    - uses: actions/setup-python@v2
      with:
-        python-version: 3.9
-    - name: Install coveralls
-      run: python -m pip install coveralls
-    - name: Upload Coverage
-      env:
-        COVERALLS_FLAG_NAME: behave-${{ matrix.os }}-${{ matrix.dcs }}-${{ matrix.python-version }}
-        COVERALLS_PARALLEL: 'true'
-        GITHUB_TOKEN: ${{ secrets.github_token }}
-      run: python -m coveralls --service=github
-
-  behavem:
-    runs-on: ${{ matrix.os }}-latest
-    env:
-      DCS: ${{ matrix.dcs }}
-      ETCDVERSION: 3.3.13
-      PGVERSION: 12.1-1  # for windows and macos
-    strategy:
-      fail-fast: false
-      matrix:
-        os: [macos]  #, windows]
-        python-version: [3.7]
-        dcs: [etcd, etcd3, raft]
-
-    steps:
-    - uses: actions/checkout@v1
-    - name: Set up Python
-      uses: actions/setup-python@v2
-      with:
-        python-version: ${{ matrix.python-version }}
-    - name: Install dependencies
-      run: python .github/workflows/install_deps.py
-    - name: Run behave tests
-      run: python .github/workflows/run_tests.py
+        python-version: '3.10'
    - name: Install coveralls
      run: python -m pip install coveralls
    - name: Upload Coverage
@@ -167,7 +141,7 @@ jobs:

  coveralls-finish:
    name: Finalize coveralls.io
-    needs: [unit, behave, behavem]
+    needs: [unit, behave]
    runs-on: ubuntu-latest
    steps:
    - uses: actions/setup-python@v2
@@ -49,6 +49,7 @@ Consul
 -  **PATRONI\_CONSUL\_CHECKS**: (optional) list of Consul health checks used for the session. By default an empty list is used.
 -  **PATRONI\_CONSUL\_REGISTER\_SERVICE**: (optional) whether or not to register a service with the name defined by the scope parameter and the tag master, replica or standby-leader depending on the node's role. Defaults to **false**
 -  **PATRONI\_CONSUL\_SERVICE\_CHECK\_INTERVAL**: (optional) how often to perform health check against registered url
+-  **PATRONI\_CONSUL\_SERVICE\_CHECK\_TLS\_SERVER\_NAME**: (optional) overide SNI host when connecting via TLS, see also `consul agent check API reference <https://www.consul.io/api-docs/agent/check#tlsservername>`__.

 Etcd
 ----
@@ -79,7 +79,7 @@ Patroni can be installed with pip:
 where dependencies can be either empty, or consist of one or more of the following:

 etcd or etcd3
-    `python-etcd` module in order to use Etcd as DCS
+    `python-etcd` module in order to use Etcd as Distributed Configuration Store (DCS)
 consul
    `python-consul` module in order to use Consul as DCS
 zookeeper
@@ -132,6 +132,7 @@ Most of the parameters are optional, but you have to specify one of the **host**
 -  **register\_service**: (optional) whether or not to register a service with the name defined by the scope parameter and the tag master, replica or standby-leader depending on the node's role. Defaults to **false**.
 -  **service\_tags**: (optional) additional static tags to add to the Consul service apart from the role (``master``/``replica``/``standby-leader``).  By default an empty list is used.
 -  **service\_check\_interval**: (optional) how often to perform health check against registered url.
+-  **service\_check\_tls\_server\_name**: (optional) overide SNI host when connecting via TLS, see also `consul agent check API reference <https://www.consul.io/api-docs/agent/check#tlsservername>`__.

 The ``token`` needs to have the following ACL permissions:

@@ -313,7 +314,7 @@ PostgreSQL
    -  **pg\_ctl\_timeout**: How long should pg_ctl wait when doing ``start``, ``stop`` or ``restart``. Default value is 60 seconds.
    -  **use\_pg\_rewind**: try to use pg\_rewind on the former leader when it joins cluster as a replica.
    -  **remove\_data\_directory\_on\_rewind\_failure**: If this option is enabled, Patroni will remove the PostgreSQL data directory and recreate the replica. Otherwise it will try to follow the new leader. Default value is **false**.
-    -  **remove\_data\_directory\_on\_diverged\_timelines**: Patroni will remove the PostgreSQL data directory and recreate the replica if it notices that timelines are diverging and the former master can not start streaming from the new master. This option is useful when ``pg_rewind`` can not be used. Default value is **false**.
+    -  **remove\_data\_directory\_on\_diverged\_timelines**: Patroni will remove the PostgreSQL data directory and recreate the replica if it notices that timelines are diverging and the former master can not start streaming from the new master. This option is useful when ``pg_rewind`` can not be used. While performing timelines divergence check on PostgreSQL v10 and older Patroni will try to connect with replication credential to the "postgres" database. Hence, such access should be allowed in the  pg_hba.conf. Default value is **false**.
    -  **replica\_method**: for each create_replica_methods other than basebackup, you would add a configuration section of the same name. At a minimum, this should include "command" with a full path to the actual script to be executed. Other configuration parameters will be passed along to the script in the form "parameter=value".
    -  **pre\_promote**: a fencing script that executes during a failover after acquiring the leader lock but before promoting the replica. If the script exits with a non-zero code, Patroni does not promote the replica and removes the leader key from DCS.

@@ -329,7 +330,7 @@ REST API
            -  **password**: Basic-auth password to protect unsafe REST API endpoints.
        -  **certfile**: (optional): Specifies the file with the certificate in the PEM format. If the certfile is not specified or is left empty, the API server will work without SSL.
        -  **keyfile**: (optional): Specifies the file with the secret key in the PEM format.
-        -  **keyfile_password**: (optional): Specifies a password for decrypting the keyfile.
+        -  **keyfile\_password**: (optional): Specifies a password for decrypting the keyfile.
        -  **cafile**: (optional): Specifies the file with the CA_BUNDLE with certificates of trusted CAs to use while verifying client certs.
        -  **ciphers**: (optional): Specifies the permitted cipher suites (e.g. "ECDHE-RSA-AES256-GCM-SHA384:DHE-RSA-AES256-GCM-SHA384:ECDHE-RSA-AES128-GCM-SHA256:DHE-RSA-AES128-GCM-SHA256:!SSLv1:!SSLv2:!SSLv3:!TLSv1:!TLSv1.1")
        -  **verify\_client**: (optional): ``none`` (default), ``optional`` or ``required``. When ``none`` REST API will not check client certificates. When ``required`` client certificates are required for all REST API calls. When ``optional`` client certificates are required for all unsafe REST API endpoints. When ``required`` is used, then client authentication succeeds, if the certificate signature verification succeeds.  For ``optional`` the client cert will only be checked for ``PUT``, ``POST``, ``PATCH``, and ``DELETE`` requests.
@@ -367,6 +368,7 @@ CTL
    -  **cacert**: Specifies the file with the CA_BUNDLE file or directory with certificates of trusted CAs to use while verifying REST API SSL certs. If not provided patronictl will use the value provided for REST API "cafile" parameter.
    -  **certfile**: Specifies the file with the client certificate in the PEM format. If not provided patronictl will use the value provided for REST API "certfile" parameter.
    -  **keyfile**: Specifies the file with the client secret key in the PEM format. If not provided patronictl will use the value provided for REST API "keyfile" parameter.
+    -  **keyfile\_password**: Specifies a password for decrypting the keyfile. If not provided patronictl will use the value provided for REST API "keyfile\_password" parameter.

 Watchdog
 --------
@@ -3,6 +3,144 @@
 Release notes
 =============

+Version 2.1.4
+-------------
+
+**New features**
+
+- Improve ``pg_rewind`` behavior on typical Debian/Ubuntu systems (Gunnar "Nick" Bluth)
+
+  On Postgres setups that keep `postgresql.conf` outside of the data directory (e.g. Ubuntu/Debian packages), ``pg_rewind --restore-target-wal``  fails to figure out the value of the ``restore_command``.
+
+- Allow setting ``TLSServerName`` on Consul service checks (Michael Gmelin)
+
+  Useful when checks are performed by IP and the Consul ``node_name`` is not a FQDN.
+
+- Added ``ppc64le`` support in watchdog (Jean-Michel Scheiwiler)
+
+  And fixed watchdog support on some non-x86 platforms.
+
+- Switched aws.py callback from ``boto`` to ``boto3`` (Alexander Kukushkin)
+
+ ``boto``  2.x is abandoned since 2018 and fails with python 3.9.
+
+- Periodically refresh service account token on K8s (Haitao Li)
+
+  Since Kubernetes v1.21 service account tokens expire in 1 hour.
+
+- Added ``/read-only-sync`` monitoring endpoint (Dennis4b)
+
+  It is similar to the ``/read-only`` but includes only synchronous replicas.
+
+
+**Stability improvements**
+
+- Don't copy the logical replication slot to a replica if there is a configuration mismatch in the logical decoding setup with the primary (Alexander)
+
+  A replica won't copy a logical replication slot from the primary anymore if the slot doesn't match the ``plugin`` or ``database`` configuration options. Previously, the check for whether the slot matches those configuration options was not performed until after the replica copied the slot and started with it, resulting in unnecessary and repeated restarts.
+
+- Special handling of recovery configuration parameters for PostgreSQL v12+ (Alexander)
+
+  While starting as replica Patroni should be able to update ``postgresql.conf`` and restart/reload if the leader address has changed by caching current parameters values instead of querying them from ``pg_settings``.
+
+- Better handling of IPv6 addresses in the ``postgresql.listen`` parameters (Alexander)
+
+  Since the ``listen`` parameter has a port, people try to put IPv6 addresses into square brackets, which were not correctly stripped when there is more than one IP in the list.
+
+- Use ``replication`` credentials when performing divergence check only on PostgreSQL v10 and older (Alexander)
+
+  If ``rewind`` is enabled, Patroni will again use either ``superuser`` or ``rewind`` credentials on newer Postgres versions.
+
+
+**Bugfixes**
+
+- Fixed missing import of ``dateutil.parser`` (Wesley Mendes)
+
+  Tests weren't failing only because it was also imported from other modules.
+
+- Ensure that ``optime`` annotation is a string (Sebastian Hasler)
+
+  In certain cases Patroni was trying to pass it as numeric.
+
+- Better handling of failed ``pg_rewind`` attempt (Alexander)
+
+  If the primary becomes unavailable during ``pg_rewind``, ``$PGDATA`` will be left in a broken state. Following that,  Patroni will remove the data directory even if this is not allowed by the configuration.
+
+- Don't remove ``slots`` annotations from the leader ``ConfigMap``/``Endpoint`` when PostgreSQL isn't ready (Alexander)
+
+  If ``slots`` value isn't passed the annotation will keep the current value.
+
+- Handle concurrency problem with K8s API watchers (Alexander)
+
+  Under certain (unknown) conditions watchers might become stale; as a result, ``attempt_to_acquire_leader()`` method could fail due to the HTTP status code 409. In that case we reset watchers connections and restart from scratch.
+
+
+Version 2.1.3
+-------------
+
+**New features**
+
+- Added support for encrypted TLS keys for ``patronictl`` (Alexander Kukushkin)
+
+  It could be configured via ``ctl.keyfile_password`` or the ``PATRONI_CTL_KEYFILE_PASSWORD`` environment variable.
+
+- Added more metrics to the /metrics endpoint (Alexandre Pereira)
+
+  Specifically, ``patroni_pending_restart`` and ``patroni_is_paused``.
+
+- Make it possible to specify multiple hosts in the standby cluster configuration (Michael Banck)
+
+  If the standby cluster is replicating from the Patroni cluster it might be nice to rely on client-side failover which is available in ``libpq`` since PostgreSQL v10. That is, the ``primary_conninfo`` on the standby leader and ``pg_rewind`` setting ``target_session_attrs=read-write`` in the connection string. The ``pgpass`` file will be generated with multiple lines (one line per host), and instead of calling ``CHECKPOINT`` on the primary cluster nodes the standby cluster will wait for ``pg_control`` to be updated.
+
+**Stability improvements**
+
+- Compatibility with legacy ``psycopg2`` (Alexander)
+
+  For example, the ``psycopg2`` installed from Ubuntu 18.04 packages doesn't have the ``UndefinedFile`` exception yet.
+
+- Restart ``etcd3`` watcher if all Etcd nodes don't respond (Alexander)
+
+  If the watcher is alive the ``get_cluster()`` method continues returning stale information even if all Etcd nodes are failing.
+
+- Don't remove the leader lock in the standby cluster while paused (Alexander)
+
+  Previously the lock was maintained only by the node that was running as a primary and not a standby leader.
+
+**Bugfixes**
+
+- Fixed bug in the standby-leader bootstrap (Alexander)
+
+  Patroni was considering bootstrap as failed if Postgres didn't start accepting connections after 60 seconds. The bug was introduced in the 2.1.2 release.
+
+- Fixed bug with failover to a cascading standby (Alexander)
+
+  When figuring out which slots should be created on cascading standby we forgot to take into account that the leader might be absent.
+
+- Fixed small issues in Postgres config validator (Alexander)
+
+  Integer parameters introduced in PostgreSQL v14 were failing to validate because min and max values were quoted in the validator.py
+
+- Use replication credentials when checking leader status (Alexander)
+
+  It could be that the ``remove_data_directory_on_diverged_timelines`` is set, but there is no ``rewind_credentials`` defined and superuser access between nodes is not allowed.
+
+- Fixed "port in use" error on REST API certificate replacement (Ants Aasma)
+
+  When switching certificates there was a race condition with a concurrent API request. If there is one active during the replacement period then the replacement will error out with a port in use error and Patroni gets stuck in a state without an active API server.
+
+- Fixed a bug in cluster bootstrap if passwords contain ``%`` characters (Bastien Wirtz)
+
+  The bootstrap method executes the ``DO`` block, with all parameters properly quoted, but the ``cursor.execute()`` method didn't like an empty list with parameters passed.
+
+- Fixed the "AttributeError: no attribute 'leader'" exception (Hrvoje Milković)
+
+  It could happen if the synchronous mode is enabled and the DCS content was wiped out.
+
+- Fix bug in divergence timeline check (Alexander)
+
+  Patroni was falsely assuming that timelines have diverged. For pg_rewind it didn't create any problem, but if pg_rewind is not allowed and the ``remove_data_directory_on_diverged_timelines`` is set, it resulted in reinitializing the former leader.
+
+
 Version 2.1.2
 -------------

@@ -44,8 +44,11 @@ For all health check ``GET`` requests Patroni returns a JSON document with the s

 - ``GET /synchronous`` or ``GET /sync``: returns HTTP status code **200** only when the Patroni node is running as a synchronous standby.

+- ``GET /read-only-sync``: like the above endpoint, but also includes the primary.
+
 - ``GET /asynchronous`` or ``GET /async``: returns HTTP status code **200** only when the Patroni node is running as an asynchronous standby.

+
 - ``GET /asynchronous?lag=<max-lag>`` or ``GET /async?lag=<max-lag>``: asynchronous standby check endpoint. In addition to checks from ``asynchronous`` or ``async``, 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.

  - ``GET /async?lag=1048576``
@@ -186,7 +186,7 @@ class PatroniController(AbstractController):
        config['postgresql']['parameters'].update({
            'logging_collector': 'on', 'log_destination': 'csvlog', 'log_directory': self._output_dir,
            'log_filename': name + '.log', 'log_statement': 'all', 'log_min_messages': 'debug1',
-            'unix_socket_directories': self._data_dir})
+            'unix_socket_directories': tempfile.gettempdir()})

        if 'bootstrap' in config:
            config['bootstrap']['post_bootstrap'] = 'psql -w -c "SELECT 1"'
@@ -660,7 +660,7 @@ class PatroniPoolController(object):
    def output_dir(self):
        return self._output_dir

-    def start(self, name, max_wait_limit=20, custom_config=None):
+    def start(self, name, max_wait_limit=40, custom_config=None):
        if name not in self._processes:
            self._processes[name] = PatroniController(self._context, name, self.patroni_path,
                                                      self._output_dir, custom_config)
@@ -51,11 +51,11 @@ Scenario: check the scheduled restart
 	Given I issue a PATCH request to http://127.0.0.1:8008/config with {"postgresql": {"parameters": {"superuser_reserved_connections": "6"}}}
 	Then I receive a response code 200
 		And Response on GET http://127.0.0.1:8008/patroni contains pending_restart after 5 seconds
-	Given I issue a scheduled restart at http://127.0.0.1:8008 in 3 seconds with {"role": "replica"}
+	Given I issue a scheduled restart at http://127.0.0.1:8008 in 5 seconds with {"role": "replica"}
 	Then I receive a response code 202
-		And I sleep for 4 seconds
+		And I sleep for 8 seconds
 		And Response on GET http://127.0.0.1:8008/patroni contains pending_restart after 10 seconds
-	Given I issue a scheduled restart at http://127.0.0.1:8008 in 3 seconds with {"restart_pending": "True"}
+	Given I issue a scheduled restart at http://127.0.0.1:8008 in 5 seconds with {"restart_pending": "True"}
 	Then I receive a response code 202
 		And Response on GET http://127.0.0.1:8008/patroni does not contain pending_restart after 10 seconds
 		And postgres0 role is the primary after 10 seconds
@@ -104,12 +104,12 @@ Scenario: check the switchover via the API in the pause mode
 	Then I receive a response code 503

 Scenario: check the scheduled switchover
-	Given I issue a scheduled switchover from postgres1 to postgres0 in 3 seconds
+	Given I issue a scheduled switchover from postgres1 to postgres0 in 10 seconds
 	Then I receive a response returncode 1
 	And I receive a response output "Can't schedule switchover in the paused state"
 	When I run patronictl.py resume batman
 	Then I receive a response returncode 0
-	Given I issue a scheduled switchover from postgres1 to postgres0 in 3 seconds
+	Given I issue a scheduled switchover from postgres1 to postgres0 in 5 seconds
 	Then I receive a response returncode 0
 	And postgres0 is a leader after 20 seconds
 	And postgres0 role is the primary after 10 seconds
@@ -76,6 +76,8 @@ def do_request(context, request_method, url, data):
    data = data and json.loads(data)
    try:
        r = request_executor.request(request_method, url, data)
+        if request_method == 'PATCH' and r.status == 409:
+            r = request_executor.request(request_method, url, data)
    except Exception:
        context.status_code = context.response = None
    else:
@@ -57,7 +57,7 @@ def start_patroni_standby_cluster(context, name, cluster_name, name2):

@step('{pg_name1:w} is replicating from {pg_name2:w} after {timeout:d} seconds')
 def check_replication_status(context, pg_name1, pg_name2, timeout):
-    bound_time = time.time() + timeout
+    bound_time = time.time() + timeout * context.timeout_multiplier

    while time.time() < bound_time:
        cur = context.pctl.query(
@@ -1,5 +1,5 @@
 #!/usr/bin/env python
-from patroni import main
+from patroni.__main__ import main


 if __name__ == '__main__':
@@ -1,145 +1,10 @@
-import logging
-import os
-import signal
 import sys
-import time
-
-from .daemon import AbstractPatroniDaemon, abstract_main
-from .version import __version__
-
-logger = logging.getLogger(__name__)

 PATRONI_ENV_PREFIX = 'PATRONI_'
 KUBERNETES_ENV_PREFIX = 'KUBERNETES_'
 MIN_PSYCOPG2 = (2, 5, 4)


-class Patroni(AbstractPatroniDaemon):
-
-    def __init__(self, config):
-        from patroni.api import RestApiServer
-        from patroni.dcs import get_dcs
-        from patroni.ha import Ha
-        from patroni.postgresql import Postgresql
-        from patroni.request import PatroniRequest
-        from patroni.watchdog import Watchdog
-
-        super(Patroni, self).__init__(config)
-
-        self.version = __version__
-        self.dcs = get_dcs(self.config)
-        self.watchdog = Watchdog(self.config)
-        self.load_dynamic_configuration()
-
-        self.postgresql = Postgresql(self.config['postgresql'])
-        self.api = RestApiServer(self, self.config['restapi'])
-        self.request = PatroniRequest(self.config, True)
-        self.ha = Ha(self)
-
-        self.tags = self.get_tags()
-        self.next_run = time.time()
-        self.scheduled_restart = {}
-
-    def load_dynamic_configuration(self):
-        from patroni.exceptions import DCSError
-        while True:
-            try:
-                cluster = self.dcs.get_cluster()
-                if cluster and cluster.config and cluster.config.data:
-                    if self.config.set_dynamic_configuration(cluster.config):
-                        self.dcs.reload_config(self.config)
-                        self.watchdog.reload_config(self.config)
-                elif not self.config.dynamic_configuration and 'bootstrap' in self.config:
-                    if self.config.set_dynamic_configuration(self.config['bootstrap']['dcs']):
-                        self.dcs.reload_config(self.config)
-                break
-            except DCSError:
-                logger.warning('Can not get cluster from dcs')
-                time.sleep(5)
-
-    def get_tags(self):
-        return {tag: value for tag, value in self.config.get('tags', {}).items()
-                if tag not in ('clonefrom', 'nofailover', 'noloadbalance', 'nosync') or value}
-
-    @property
-    def nofailover(self):
-        return bool(self.tags.get('nofailover', False))
-
-    @property
-    def nosync(self):
-        return bool(self.tags.get('nosync', False))
-
-    def reload_config(self, sighup=False, local=False):
-        try:
-            super(Patroni, self).reload_config(sighup, local)
-            if local:
-                self.tags = self.get_tags()
-                self.request.reload_config(self.config)
-            if local or sighup and 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)
-            self.dcs.reload_config(self.config)
-        except Exception:
-            logger.exception('Failed to reload config_file=%s', self.config.config_file)
-
-    @property
-    def replicatefrom(self):
-        return self.tags.get('replicatefrom')
-
-    @property
-    def noloadbalance(self):
-        return bool(self.tags.get('noloadbalance', False))
-
-    def schedule_next_run(self):
-        self.next_run += self.dcs.loop_wait
-        current_time = time.time()
-        nap_time = self.next_run - current_time
-        if nap_time <= 0:
-            self.next_run = current_time
-            # Release the GIL so we don't starve anyone waiting on async_executor lock
-            time.sleep(0.001)
-            # Warn user that Patroni is not keeping up
-            logger.warning("Loop time exceeded, rescheduling immediately.")
-        elif self.ha.watch(nap_time):
-            self.next_run = time.time()
-
-    def run(self):
-        self.api.start()
-        self.next_run = time.time()
-        super(Patroni, self).run()
-
-    def _run_cycle(self):
-        logger.info(self.ha.run_cycle())
-
-        if self.dcs.cluster and self.dcs.cluster.config and self.dcs.cluster.config.data \
-                and self.config.set_dynamic_configuration(self.dcs.cluster.config):
-            self.reload_config()
-
-        if self.postgresql.role != 'uninitialized':
-            self.config.save_cache()
-
-        self.schedule_next_run()
-
-    def _shutdown(self):
-        try:
-            self.api.shutdown()
-        except Exception:
-            logger.exception('Exception during RestApi.shutdown')
-        try:
-            self.ha.shutdown()
-        except Exception:
-            logger.exception('Exception during Ha.shutdown')
-
-
-def patroni_main():
-    from multiprocessing import freeze_support
-    from patroni.validator import schema
-
-    freeze_support()
-    abstract_main(Patroni, schema)
-
-
 def fatal(string, *args):
    sys.stderr.write('FATAL: ' + string.format(*args) + '\n')
    sys.exit(1)
@@ -174,44 +39,3 @@ def check_psycopg(_min_psycopg2=MIN_PSYCOPG2, _parse_version=parse_version):
        if version_str:
            error += ', but only psycopg2=={0} is available'.format(version_str)
        fatal(error)
-
-
-def main():
-    if os.getpid() != 1:
-        check_psycopg()
-        return patroni_main()
-
-    # Patroni started with PID=1, it looks like we are in the container
-    pid = 0
-
-    # Looks like we are in a docker, so we will act like init
-    def sigchld_handler(signo, stack_frame):
-        try:
-            while True:
-                ret = os.waitpid(-1, os.WNOHANG)
-                if ret == (0, 0):
-                    break
-                elif ret[0] != pid:
-                    logger.info('Reaped pid=%s, exit status=%s', *ret)
-        except OSError:
-            pass
-
-    def passtochild(signo, stack_frame):
-        if pid:
-            os.kill(pid, signo)
-
-    if os.name != 'nt':
-        signal.signal(signal.SIGCHLD, sigchld_handler)
-        signal.signal(signal.SIGHUP, passtochild)
-        signal.signal(signal.SIGQUIT, passtochild)
-        signal.signal(signal.SIGUSR1, passtochild)
-        signal.signal(signal.SIGUSR2, passtochild)
-    signal.signal(signal.SIGINT, passtochild)
-    signal.signal(signal.SIGABRT, passtochild)
-    signal.signal(signal.SIGTERM, passtochild)
-
-    import multiprocessing
-    patroni = multiprocessing.Process(target=patroni_main)
-    patroni.start()
-    pid = patroni.pid
-    patroni.join()
@@ -1,4 +1,181 @@
-from patroni import main
+import logging
+import os
+import signal
+import time
+
+from .daemon import AbstractPatroniDaemon, abstract_main
+
+logger = logging.getLogger(__name__)
+
+
+class Patroni(AbstractPatroniDaemon):
+
+    def __init__(self, config):
+        from .api import RestApiServer
+        from .dcs import get_dcs
+        from .ha import Ha
+        from .postgresql import Postgresql
+        from .request import PatroniRequest
+        from .version import __version__
+        from .watchdog import Watchdog
+
+        super(Patroni, self).__init__(config)
+
+        self.version = __version__
+        self.dcs = get_dcs(self.config)
+        self.watchdog = Watchdog(self.config)
+        self.load_dynamic_configuration()
+
+        self.postgresql = Postgresql(self.config['postgresql'])
+        self.api = RestApiServer(self, self.config['restapi'])
+        self.request = PatroniRequest(self.config, True)
+        self.ha = Ha(self)
+
+        self.tags = self.get_tags()
+        self.next_run = time.time()
+        self.scheduled_restart = {}
+
+    def load_dynamic_configuration(self):
+        from patroni.exceptions import DCSError
+        while True:
+            try:
+                cluster = self.dcs.get_cluster()
+                if cluster and cluster.config and cluster.config.data:
+                    if self.config.set_dynamic_configuration(cluster.config):
+                        self.dcs.reload_config(self.config)
+                        self.watchdog.reload_config(self.config)
+                elif not self.config.dynamic_configuration and 'bootstrap' in self.config:
+                    if self.config.set_dynamic_configuration(self.config['bootstrap']['dcs']):
+                        self.dcs.reload_config(self.config)
+                break
+            except DCSError:
+                logger.warning('Can not get cluster from dcs')
+                time.sleep(5)
+
+    def get_tags(self):
+        return {tag: value for tag, value in self.config.get('tags', {}).items()
+                if tag not in ('clonefrom', 'nofailover', 'noloadbalance', 'nosync') or value}
+
+    @property
+    def nofailover(self):
+        return bool(self.tags.get('nofailover', False))
+
+    @property
+    def nosync(self):
+        return bool(self.tags.get('nosync', False))
+
+    def reload_config(self, sighup=False, local=False):
+        try:
+            super(Patroni, self).reload_config(sighup, local)
+            if local:
+                self.tags = self.get_tags()
+                self.request.reload_config(self.config)
+            if local or sighup and 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)
+            self.dcs.reload_config(self.config)
+        except Exception:
+            logger.exception('Failed to reload config_file=%s', self.config.config_file)
+
+    @property
+    def replicatefrom(self):
+        return self.tags.get('replicatefrom')
+
+    @property
+    def noloadbalance(self):
+        return bool(self.tags.get('noloadbalance', False))
+
+    def schedule_next_run(self):
+        self.next_run += self.dcs.loop_wait
+        current_time = time.time()
+        nap_time = self.next_run - current_time
+        if nap_time <= 0:
+            self.next_run = current_time
+            # Release the GIL so we don't starve anyone waiting on async_executor lock
+            time.sleep(0.001)
+            # Warn user that Patroni is not keeping up
+            logger.warning("Loop time exceeded, rescheduling immediately.")
+        elif self.ha.watch(nap_time):
+            self.next_run = time.time()
+
+    def run(self):
+        self.api.start()
+        self.next_run = time.time()
+        super(Patroni, self).run()
+
+    def _run_cycle(self):
+        logger.info(self.ha.run_cycle())
+
+        if self.dcs.cluster and self.dcs.cluster.config and self.dcs.cluster.config.data \
+                and self.config.set_dynamic_configuration(self.dcs.cluster.config):
+            self.reload_config()
+
+        if self.postgresql.role != 'uninitialized':
+            self.config.save_cache()
+
+        self.schedule_next_run()
+
+    def _shutdown(self):
+        try:
+            self.api.shutdown()
+        except Exception:
+            logger.exception('Exception during RestApi.shutdown')
+        try:
+            self.ha.shutdown()
+        except Exception:
+            logger.exception('Exception during Ha.shutdown')
+
+
+def patroni_main():
+    from multiprocessing import freeze_support
+    from patroni.validator import schema
+
+    freeze_support()
+    abstract_main(Patroni, schema)
+
+
+def main():
+    if os.getpid() != 1:
+        from . import check_psycopg
+
+        check_psycopg()
+        return patroni_main()
+
+    # Patroni started with PID=1, it looks like we are in the container
+    pid = 0
+
+    # Looks like we are in a docker, so we will act like init
+    def sigchld_handler(signo, stack_frame):
+        try:
+            while True:
+                ret = os.waitpid(-1, os.WNOHANG)
+                if ret == (0, 0):
+                    break
+                elif ret[0] != pid:
+                    logger.info('Reaped pid=%s, exit status=%s', *ret)
+        except OSError:
+            pass
+
+    def passtochild(signo, stack_frame):
+        if pid:
+            os.kill(pid, signo)
+
+    if os.name != 'nt':
+        signal.signal(signal.SIGCHLD, sigchld_handler)
+        signal.signal(signal.SIGHUP, passtochild)
+        signal.signal(signal.SIGQUIT, passtochild)
+        signal.signal(signal.SIGUSR1, passtochild)
+        signal.signal(signal.SIGUSR2, passtochild)
+    signal.signal(signal.SIGINT, passtochild)
+    signal.signal(signal.SIGABRT, passtochild)
+    signal.signal(signal.SIGTERM, passtochild)
+
+    import multiprocessing
+    patroni = multiprocessing.Process(target=patroni_main)
+    patroni.start()
+    pid = patroni.pid
+    patroni.join()


 if __name__ == '__main__':
@@ -152,6 +152,11 @@ class RestApiHandler(BaseHTTPRequestHandler):
                status_code = replica_status_code
            elif path in ('/async', '/asynchronous') and not is_synchronous:
                status_code = replica_status_code
+            elif path in ('/read-only-sync', '/read-only-synchronous'):
+                if 200 in (primary_status_code, standby_leader_status_code):
+                    status_code = 200
+                elif is_synchronous:
+                    status_code = replica_status_code

        # check for user defined tags in query params
        if not ignore_tags and status_code == 200:
@@ -293,6 +298,16 @@ class RestApiHandler(BaseHTTPRequestHandler):
        metrics.append("# TYPE patroni_dcs_last_seen gauge")
        metrics.append("patroni_dcs_last_seen{0} {1}".format(scope_label, postgres.get('dcs_last_seen', 0)))

+        metrics.append("# HELP patroni_pending_restart Value is 1 if the node needs a restart, 0 otherwise.")
+        metrics.append("# TYPE patroni_pending_restart gauge")
+        metrics.append("patroni_pending_restart{0} {1}"
+                       .format(scope_label, int(patroni.postgresql.pending_restart)))
+
+        metrics.append("# HELP patroni_is_paused Value is 1 if auto failover is disabled, 0 otherwise.")
+        metrics.append("# TYPE patroni_is_paused gauge")
+        metrics.append("patroni_is_paused{0} {1}"
+                       .format(scope_label, int(patroni.ha.is_paused())))
+
        self._write_response(200, '\n'.join(metrics)+'\n', content_type='text/plain')

    def _read_json_content(self, body_is_optional=False):
@@ -768,6 +783,8 @@ class RestApiServer(ThreadingMixIn, HTTPServer, Thread):
        reloading_config = self.__listen is not None  # changing config in runtime
        if reloading_config:
            self.shutdown()
+            # Rely on ThreadingMixIn.server_close() to have all requests terminate before we continue
+            self.server_close()

        self.__listen = listen
        self.__ssl_options = ssl_options
@@ -270,7 +270,7 @@ class Config(object):
        _set_section_values('restapi', ['listen', 'connect_address', 'certfile', 'keyfile', 'keyfile_password',
                                        'cafile', 'ciphers', 'verify_client', 'http_extra_headers',
                                        'https_extra_headers', 'allowlist', 'allowlist_include_members'])
-        _set_section_values('ctl', ['insecure', 'cacert', 'certfile', 'keyfile'])
+        _set_section_values('ctl', ['insecure', 'cacert', 'certfile', 'keyfile', 'keyfile_password'])
        _set_section_values('postgresql', ['listen', 'connect_address', 'config_dir', 'data_dir', 'pgpass', 'bin_dir'])
        _set_section_values('log', ['level', 'traceback_level', 'format', 'dateformat', 'max_queue_size',
                                    'dir', 'file_size', 'file_num', 'loggers'])
@@ -350,9 +350,9 @@ class Config(object):
                name, suffix = (param[8:].split('_', 1) + [''])[:2]
                if suffix in ('HOST', 'HOSTS', 'PORT', 'USE_PROXIES', 'PROTOCOL', 'SRV', 'SRV_SUFFIX', 'URL', 'PROXY',
                              'CACERT', 'CERT', 'KEY', 'VERIFY', 'TOKEN', 'CHECKS', 'DC', 'CONSISTENCY',
-                              'REGISTER_SERVICE', 'SERVICE_CHECK_INTERVAL', 'NAMESPACE', 'CONTEXT',
-                              'USE_ENDPOINTS', 'SCOPE_LABEL', 'ROLE_LABEL', 'POD_IP', 'PORTS', 'LABELS',
-                              'BYPASS_API_SERVICE', 'KEY_PASSWORD', 'USE_SSL', 'SET_ACLS') and name:
+                              'REGISTER_SERVICE', 'SERVICE_CHECK_INTERVAL', 'SERVICE_CHECK_TLS_SERVER_NAME',
+                              'NAMESPACE', 'CONTEXT', 'USE_ENDPOINTS', 'SCOPE_LABEL', 'ROLE_LABEL', 'POD_IP',
+                              'PORTS', 'LABELS', 'BYPASS_API_SERVICE', 'KEY_PASSWORD', 'USE_SSL', 'SET_ACLS') and name:
                    value = os.environ.pop(param)
                    if suffix == 'PORT':
                        value = value and parse_int(value)
@@ -1,5 +1,5 @@
 import abc
-import dateutil
+import dateutil.parser
 import importlib
 import inspect
 import json
@@ -460,8 +460,12 @@ class Cluster(namedtuple('Cluster', 'initialize,config,leader,last_lsn,members,f
    :param slots: state of permanent logical replication slots on the primary in the format: {"slot_name": int}
    """

+    @property
+    def leader_name(self):
+        return self.leader and self.leader.name
+
    def is_unlocked(self):
-        return not (self.leader and self.leader.name)
+        return not self.leader_name

    def has_member(self, member_name):
        return any(m for m in self.members if m.name == member_name)
@@ -516,7 +520,7 @@ class Cluster(namedtuple('Cluster', 'initialize,config,leader,last_lsn,members,f
        else:
            # only manage slots for replicas that replicate from this one, except for the leader among them
            slot_members = [m.name for m in self.members if use_slots and
-                            m.replicatefrom == my_name and m.name != self.leader.name]
+                            m.replicatefrom == my_name and m.name != self.leader_name]
            permanent_slots = self.__permanent_logical_slots if use_slots and not nofailover else {}

        slots = {slot_name_from_member_name(name): {'type': 'physical'} for name in slot_members}
@@ -585,7 +589,7 @@ class Cluster(namedtuple('Cluster', 'initialize,config,leader,last_lsn,members,f
            return True

        if self.use_slots:
-            members = [m for m in self.members if m.replicatefrom == my_name and m.name != self.leader.name]
+            members = [m for m in self.members if m.replicatefrom == my_name and m.name != self.leader_name]
            return any(self.should_enforce_hot_standby_feedback(m.name, m.nofailover, major_version) for m in members)
        return False

@@ -933,4 +937,4 @@ class AbstractDCS(object):
        :returns: `!True` if you would like to reschedule the next run of ha cycle"""

        self.event.wait(timeout)
-        return self.event.isSet()
+        return self.event.is_set()
@@ -233,6 +233,7 @@ class Consul(AbstractDCS):
        if self._register_service:
            self._set_service_name()
        self._service_check_interval = config.get('service_check_interval', '5s')
+        self._service_check_tls_server_name = config.get('service_check_tls_server_name', None)
        if not self._ctl:
            self.create_session()

@@ -458,6 +459,8 @@ class Consul(AbstractDCS):
        conn_parts = urlparse(data['conn_url'])
        check = base.Check.http(api_parts.geturl(), self._service_check_interval,
                                deregister='{0}s'.format(self._client.http.ttl * 10))
+        if self._service_check_tls_server_name is not None:
+            check['TLSServerName'] = self._service_check_tls_server_name
        tags = self._service_tags[:]
        tags.append(role)
        self._previous_loop_service_tags = self._service_tags
@@ -269,6 +269,7 @@ class AbstractEtcdClientWithFailover(etcd.Client):
                nodes, timeout, retries = self._calculate_timeouts(etcd_nodes, remaining_time)
                if nodes == 0:
                    self._update_machines_cache = True
+                    self.set_base_uri(self._base_uri)  # trigger Etcd3 watcher restart
                    raise ex
                retry.sleep_func(sleeptime)
                retry.update_delay()
@@ -394,8 +395,9 @@ class AbstractEtcdClientWithFailover(etcd.Client):
        self._machines_cache_updated = time.time()

    def set_base_uri(self, value):
-        logger.info('Selected new etcd server %s', value)
-        self._base_uri = value
+        if self._base_uri != value:
+            logger.info('Selected new etcd server %s', value)
+            self._base_uri = value


 class EtcdClient(AbstractEtcdClientWithFailover):
@@ -48,14 +48,29 @@ class K8sConfig(object):

    def __init__(self):
        self.pool_config = {'maxsize': 10, 'num_pools': 10}  # configuration for urllib3.PoolManager
+        self._token_expires_at = datetime.datetime.max
        self._make_headers()

+    def _set_token(self, token):
+        self._headers['authorization'] = 'Bearer ' + token
+
    def _make_headers(self, token=None, **kwargs):
        self._headers = urllib3.make_headers(user_agent=USER_AGENT, **kwargs)
        if token:
-            self._headers['authorization'] = 'Bearer ' + token
+            self._set_token(token)

-    def load_incluster_config(self, ca_certs=SERVICE_CERT_FILENAME):
+    def _read_token_file(self):
+        if not os.path.isfile(SERVICE_TOKEN_FILENAME):
+            raise self.ConfigException('Service token file does not exists.')
+        with open(SERVICE_TOKEN_FILENAME) as f:
+            token = f.read()
+            if not token:
+                raise self.ConfigException('Token file exists but empty.')
+            self._token_expires_at = datetime.datetime.now() + self._token_refresh_interval
+            return token
+
+    def load_incluster_config(self, ca_certs=SERVICE_CERT_FILENAME,
+                              token_refresh_interval=datetime.timedelta(minutes=1)):
        if SERVICE_HOST_ENV_NAME not in os.environ or SERVICE_PORT_ENV_NAME not in os.environ:
            raise self.ConfigException('Service host/port is not set.')
        if not os.environ[SERVICE_HOST_ENV_NAME] or not os.environ[SERVICE_PORT_ENV_NAME]:
@@ -67,14 +82,9 @@ class K8sConfig(object):
            if not f.read():
                raise self.ConfigException('Cert file exists but empty.')
        self.pool_config['ca_certs'] = ca_certs
-
-        if not os.path.isfile(SERVICE_TOKEN_FILENAME):
-            raise self.ConfigException('Service token file does not exists.')
-        with open(SERVICE_TOKEN_FILENAME) as f:
-            token = f.read()
-            if not token:
-                raise self.ConfigException('Token file exists but empty.')
-            self._make_headers(token=token)
+        self._token_refresh_interval = token_refresh_interval
+        token = self._read_token_file()
+        self._make_headers(token=token)
        self._server = uri('https', (os.environ[SERVICE_HOST_ENV_NAME], os.environ[SERVICE_PORT_ENV_NAME]))

    @staticmethod
@@ -109,6 +119,11 @@ class K8sConfig(object):

    @property
    def headers(self):
+        if self._token_expires_at <= datetime.datetime.now():
+            try:
+                self._set_token(self._read_token_file())
+            except Exception as e:
+                logger.error('Failed to refresh service account token: %r', e)
        return self._headers.copy()


@@ -503,6 +518,8 @@ class ObjectCache(Thread):
        self._condition = condition
        self._name = name  # name of this pod
        self._is_ready = False
+        self._response = None  # needs to be accessible from the `kill_stream()` method
+        self._response_lock = Lock()  # protect the `self._response` from concurrent access
        self._object_cache = {}
        self._object_cache_lock = Lock()
        self._annotations_map = {self._dcs.leader_path: self._dcs._LEADER, self._dcs.config_path: self._dcs._CONFIG}
@@ -543,63 +560,99 @@ class ObjectCache(Thread):
        with self._object_cache_lock:
            return self._object_cache.get(name)

+    def _process_event(self, event):
+        ev_type = event['type']
+        obj = event['object']
+        name = obj['metadata']['name']
+
+        if ev_type in ('ADDED', 'MODIFIED'):
+            obj = K8sObject(obj)
+            success, old_value = self.set(name, obj)
+            if success:
+                new_value = (obj.metadata.annotations or {}).get(self._annotations_map.get(name))
+        elif ev_type == 'DELETED':
+            success, old_value = self.delete(name, obj['metadata']['resourceVersion'])
+            new_value = None
+        else:
+            return logger.warning('Unexpected event type: %s', ev_type)
+
+        if success and obj.get('kind') != 'Pod':
+            if old_value:
+                old_value = (old_value.metadata.annotations or {}).get(self._annotations_map.get(name))
+
+            value_changed = old_value != new_value and \
+                (name != self._dcs.config_path or old_value is not None and new_value is not None)
+
+            if value_changed:
+                logger.debug('%s changed from %s to %s', name, old_value, new_value)
+
+            # Do not wake up HA loop if we run as leader and received leader object update event
+            if value_changed or name == self._dcs.leader_path and self._name != new_value:
+                self._dcs.event.set()
+
+    @staticmethod
+    def _finish_response(response):
+        try:
+            response.close()
+        finally:
+            response.release_conn()
+
+    def _do_watch(self, resource_version):
+        with self._response_lock:
+            self._response = None
+        response = self._watch(resource_version)
+        with self._response_lock:
+            if self._response is None:
+                self._response = response
+
+        if not self._response:
+            return self._finish_response(response)
+
+        for event in iter_response_objects(response):
+            if event['object'].get('code') == 410:
+                break
+            self._process_event(event)
+
    def _build_cache(self):
        objects = self._list()
-        return_type = 'V1' + objects.kind[:-4]
        with self._object_cache_lock:
            self._object_cache = {item.metadata.name: item for item in objects.items}
        with self._condition:
            self._is_ready = True
            self._condition.notify()

-        response = self._watch(objects.metadata.resource_version)
        try:
-            for event in iter_response_objects(response):
-                obj = event['object']
-                if obj.get('code') == 410:
-                    break
-
-                ev_type = event['type']
-                name = obj['metadata']['name']
-
-                if ev_type in ('ADDED', 'MODIFIED'):
-                    obj = K8sObject(obj)
-                    success, old_value = self.set(name, obj)
-                    if success:
-                        new_value = (obj.metadata.annotations or {}).get(self._annotations_map.get(name))
-                elif ev_type == 'DELETED':
-                    success, old_value = self.delete(name, obj['metadata']['resourceVersion'])
-                    new_value = None
-                else:
-                    logger.warning('Unexpected event type: %s', ev_type)
-                    continue
-
-                if success and return_type != 'V1Pod':
-                    if old_value:
-                        old_value = (old_value.metadata.annotations or {}).get(self._annotations_map.get(name))
-
-                    value_changed = old_value != new_value and \
-                        (name != self._dcs.config_path or old_value is not None and new_value is not None)
-
-                    if value_changed:
-                        logger.debug('%s changed from %s to %s', name, old_value, new_value)
-
-                    # Do not wake up HA loop if we run as leader and received leader object update event
-                    if value_changed or name == self._dcs.leader_path and self._name != new_value:
-                        self._dcs.event.set()
+            self._do_watch(objects.metadata.resource_version)
        finally:
            with self._condition:
                self._is_ready = False
-            response.close()
-            response.release_conn()
+            with self._response_lock:
+                response, self._response = self._response, None
+            if response:
+                self._finish_response(response)
+
+    def kill_stream(self):
+        sock = None
+        with self._response_lock:
+            if self._response:
+                try:
+                    sock = self._response.connection.sock
+                except Exception:
+                    sock = None
+            else:
+                self._response = False
+        if sock:
+            try:
+                sock.shutdown(socket.SHUT_RDWR)
+                sock.close()
+            except Exception as e:
+                logger.debug('Error on socket.shutdown: %r', e)

    def run(self):
        while True:
            try:
                self._build_cache()
            except Exception as e:
-                with self._condition:
-                    self._is_ready = False
                logger.error('ObjectCache.run %r', e)

    def is_ready(self):
@@ -874,7 +927,14 @@ class Kubernetes(AbstractDCS):
    def patch_or_create(self, name, annotations, resource_version=None, patch=False, retry=True, ips=None):
        if retry is True:
            retry = self.retry
-        return self._patch_or_create(name, annotations, resource_version, patch, retry, ips)
+        try:
+            return self._patch_or_create(name, annotations, resource_version, patch, retry, ips)
+        except k8s_client.rest.ApiException as e:
+            if e.status == 409 and resource_version:  # Conflict in resource_version
+                # Terminate watchers, it could be a sign that K8s API is in a failed state
+                self._kinds.kill_stream()
+                self._pods.kill_stream()
+            raise e

    def patch_or_create_config(self, annotations, resource_version=None, patch=False, retry=True):
        # SCOPE-config endpoint requires corresponding service otherwise it might be "cleaned" by k8s master
@@ -961,7 +1021,7 @@ class Kubernetes(AbstractDCS):
                       'transitions': leader_observed_record.get('transitions') or '0'}
        if last_lsn:
            annotations[self._OPTIME] = str(last_lsn)
-        annotations['slots'] = json.dumps(slots) if slots else None
+            annotations['slots'] = json.dumps(slots) if slots else None

        resource_version = kind and kind.metadata.resource_version
        return self._update_leader_with_retry(annotations, resource_version, self.__ips)
@@ -1045,12 +1105,12 @@ class Kubernetes(AbstractDCS):
        if kind and (kind.metadata.annotations or {}).get(self._LEADER) == self._name:
            annotations = {self._LEADER: None}
            if last_lsn:
-                annotations[self._OPTIME] = last_lsn
+                annotations[self._OPTIME] = str(last_lsn)
            self.patch_or_create(self.leader_path, annotations, kind.metadata.resource_version, True, False, [])
            self.reset_cluster()

    def cancel_initialization(self):
-        self.patch_or_create_config({self._INITIALIZE: None}, self._config_resource_version, True)
+        return self.patch_or_create_config({self._INITIALIZE: None}, None, True)

    @catch_kubernetes_errors
    def delete_cluster(self):
@@ -271,7 +271,7 @@ class Raft(AbstractDCS):

        while True:
            ready_event.wait(5)
-            if ready_event.isSet() or self._sync_obj.applied_local_log:
+            if ready_event.is_set() or self._sync_obj.applied_local_log:
                break
            else:
                logger.info('waiting on raft')
@@ -241,7 +241,7 @@ class Ha(object):
            logger.info('bootstrapped %s', msg)
            cluster = self.dcs.get_cluster()
            node_to_follow = self._get_node_to_follow(cluster)
-            return self.state_handler.follow(node_to_follow)
+            return self.state_handler.follow(node_to_follow) is not False
        else:
            logger.error('failed to bootstrap %s', msg)
            self.state_handler.remove_data_directory()
@@ -293,17 +293,30 @@ class Ha(object):

        return result

+    def _handle_crash_recovery(self):
+        if not self._crash_recovery_executed and (self.cluster.is_unlocked() or self._rewind.can_rewind):
+            self._crash_recovery_executed = True
+            self._crash_recovery_started = time.time()
+            msg = 'doing crash recovery in a single user mode'
+            return self._async_executor.try_run_async(msg, self._rewind.ensure_clean_shutdown) or msg
+
    def _handle_rewind_or_reinitialize(self):
        leader = self.get_remote_master() if self.is_standby_cluster() else self.cluster.leader
        if not self._rewind.rewind_or_reinitialize_needed_and_possible(leader):
            return None

        if self._rewind.can_rewind:
+            # rewind is required, but postgres wasn't shut down cleanly.
+            if not self.state_handler.is_running() and \
+                    self.state_handler.controldata().get('Database cluster state') == 'in archive recovery':
+                msg = self._handle_crash_recovery()
+                if msg:
+                    return msg
+
            msg = 'running pg_rewind from ' + leader.name
            return self._async_executor.try_run_async(msg, self._rewind.execute, args=(leader,)) or msg

-        # remove_data_directory_on_diverged_timelines is set
-        if not self.is_standby_cluster():
+        if self._rewind.should_remove_data_directory_on_diverged_timelines and not self.is_standby_cluster():
            msg = 'reinitializing due to diverged timelines'
            return self._async_executor.try_run_async(msg, self._do_reinitialize, args=(self.cluster,)) or msg

@@ -325,13 +338,10 @@ class Ha(object):

        data = self.state_handler.controldata()
        logger.info('pg_controldata:\n%s\n', '\n'.join('  {0}: {1}'.format(k, v) for k, v in data.items()))
-        if data.get('Database cluster state') in ('in production', 'shutting down', 'in crash recovery') \
-                and not self._crash_recovery_executed and \
-                (self.cluster.is_unlocked() or self._rewind.can_rewind):
-            self._crash_recovery_executed = True
-            self._crash_recovery_started = time.time()
-            msg = 'doing crash recovery in a single user mode'
-            return self._async_executor.try_run_async(msg, self._rewind.ensure_clean_shutdown) or msg
+        if data.get('Database cluster state') in ('in production', 'shutting down', 'in crash recovery'):
+            msg = self._handle_crash_recovery()
+            if msg:
+                return msg

        self.load_cluster_from_dcs()

@@ -413,9 +423,10 @@ class Ha(object):
                self.state_handler.get_history(self._leader_timeline + 1):
            self._rewind.trigger_check_diverged_lsn()

-        msg = self._handle_rewind_or_reinitialize()
-        if msg:
-            return msg
+        if not self.state_handler.is_starting():
+            msg = self._handle_rewind_or_reinitialize()
+            if msg:
+                return msg

        if not self.is_paused():
            self.state_handler.handle_parameter_change()
@@ -796,7 +807,7 @@ class Ha(object):

        # When in sync mode, only last known master and sync standby are allowed to promote automatically.
        all_known_members = self.cluster.members + self.old_cluster.members
-        if self.is_synchronous_mode() and self.cluster.sync.leader:
+        if self.is_synchronous_mode() and self.cluster.sync and self.cluster.sync.leader:
            if not self.cluster.sync.matches(self.state_handler.name):
                return False
            # pick between synchronous candidates so we minimize unnecessary failovers/demotions
@@ -1012,8 +1023,9 @@ class Ha(object):
                if self.cluster.failover and self.cluster.failover.candidate == self.state_handler.name:
                    return 'waiting to become master after promote...'

-                self._delete_leader()
-                return 'removed leader lock because postgres is not running as master'
+                if not self.is_standby_cluster():
+                    self._delete_leader()
+                    return 'removed leader lock because postgres is not running as master'

            if self.update_lock(True):
                msg = self.process_manual_failover_from_leader()
@@ -1479,7 +1491,7 @@ class Ha(object):
                    if create_slots and self.cluster.leader:
                        err = self._async_executor.try_run_async('copy_logical_slots',
                                                                 self.state_handler.slots_handler.copy_logical_slots,
-                                                                 args=(self.cluster.leader, create_slots))
+                                                                 args=(self.cluster, create_slots))
                        if not err:
                            ret = 'Copying logical slots {0} from the primary'.format(create_slots)
            return ret
@@ -264,7 +264,7 @@ class Postgresql(object):
        cursor = None
        try:
            cursor = self._connection.cursor()
-            cursor.execute(sql, params)
+            cursor.execute(sql, params or None)
            return cursor
        except psycopg.Error as e:
            if cursor and cursor.connection.closed == 0:
@@ -387,7 +387,7 @@ class Postgresql(object):
        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():
+        if self.major_version >= 140000 and not self.is_starting() and self.replay_paused():
            logger.info('Resuming paused WAL replay for PostgreSQL 14+')
            self.resume_wal_replay()

@@ -798,7 +798,8 @@ class Postgresql(object):
        return True

    def get_guc_value(self, name):
-        cmd = [self.pgcommand('postgres'), '-D', self._data_dir, '-C', name]
+        cmd = [self.pgcommand('postgres'), '-D', self._data_dir, '-C', name,
+               '--config-file={}'.format(self.config.postgresql_conf)]
        try:
            data = subprocess.check_output(cmd)
            if data:
@@ -486,7 +486,8 @@ class ConfigHandler(object):
        # A list of keywords that can be found in a conninfo string. Follows what is acceptable by libpq
        keywords = ('dbname', 'user', 'passfile' if params.get('passfile') else 'password', 'host', 'port',
                    'sslmode', 'sslcompression', 'sslcert', 'sslkey', 'sslpassword', 'sslrootcert', 'sslcrl',
-                    'sslcrldir', 'application_name', 'krbsrvname', 'gssencmode', 'channel_binding')
+                    'sslcrldir', 'application_name', 'krbsrvname', 'gssencmode', 'channel_binding',
+                    'target_session_attrs')
        if include_dbname:
            params = params.copy()
            if 'dbname' not in params:
@@ -542,6 +543,12 @@ class ConfigHandler(object):
            if use_slots and not (is_remote_master and member.no_replication_slot):
                primary_slot_name = member.primary_slot_name if is_remote_master else self._postgresql.name
                recovery_params['primary_slot_name'] = slot_name_from_member_name(primary_slot_name)
+                # We are a standby leader and are using a replication slot. Make sure we connect to
+                # the leader of the main cluster (in case more than one host is specified in the
+                # connstr) by adding 'target_session_attrs=read-write' to primary_conninfo.
+                if is_remote_master and 'target_sesions_attrs' not in primary_conninfo and\
+                        self._postgresql.major_version >= 100000:
+                    primary_conninfo['target_session_attrs'] = 'read-write'
            recovery_params['primary_conninfo'] = primary_conninfo

        # standby_cluster config might have different parameters, we want to override them
@@ -570,6 +577,9 @@ class ConfigHandler(object):
        return self._RECOVERY_PARAMETERS - skip_params

    def _read_recovery_params(self):
+        if self._postgresql.is_starting():
+            return None, False
+
        pg_conf_mtime = mtime(self._postgresql_conf)
        auto_conf_mtime = mtime(self._auto_conf)
        passfile_mtime = mtime(self._passfile) if self._passfile else False
@@ -618,19 +628,19 @@ class ConfigHandler(object):

    def _check_passfile(self, passfile, wanted_primary_conninfo):
        # If there is a passfile in the primary_conninfo try to figure out that
-        # the passfile contains the line allowing connection to the given node.
+        # the passfile contains the line(s) allowing connection to the given node.
        # We assume that the passfile was created by Patroni and therefore doing
        # the full match and not covering cases when host, port or user are set to '*'
        passfile_mtime = mtime(passfile)
        if passfile_mtime:
            try:
                with open(passfile) as f:
-                    wanted_line = self._pgpass_line(wanted_primary_conninfo).strip()
-                    for raw_line in f:
-                        if raw_line.strip() == wanted_line:
-                            self._passfile = passfile
-                            self._passfile_mtime = passfile_mtime
-                            return True
+                    wanted_lines = self._pgpass_line(wanted_primary_conninfo).splitlines()
+                    file_lines = f.read().splitlines()
+                    if set(wanted_lines) == set(file_lines):
+                        self._passfile = passfile
+                        self._passfile_mtime = passfile_mtime
+                        return True
            except Exception:
                logger.info('Failed to read %s', passfile)
        return False
@@ -643,16 +653,17 @@ class ConfigHandler(object):
        elif not primary_conninfo:
            return False

-        wal_receiver_primary_conninfo = self._postgresql.primary_conninfo()
-        if wal_receiver_primary_conninfo:
-            wal_receiver_primary_conninfo = parse_dsn(wal_receiver_primary_conninfo)
-            # when wal receiver is alive use primary_conninfo from pg_stat_wal_receiver for comparison
+        if not self._postgresql.is_starting():
+            wal_receiver_primary_conninfo = self._postgresql.primary_conninfo()
            if wal_receiver_primary_conninfo:
-                primary_conninfo = wal_receiver_primary_conninfo
-                # There could be no password in the primary_conninfo or it is masked.
-                # Just copy the "desired" value in order to make comparison succeed.
-                if 'password' in wanted_primary_conninfo:
-                    primary_conninfo['password'] = wanted_primary_conninfo['password']
+                wal_receiver_primary_conninfo = parse_dsn(wal_receiver_primary_conninfo)
+                # when wal receiver is alive use primary_conninfo from pg_stat_wal_receiver for comparison
+                if wal_receiver_primary_conninfo:
+                    primary_conninfo = wal_receiver_primary_conninfo
+                    # There could be no password in the primary_conninfo or it is masked.
+                    # Just copy the "desired" value in order to make comparison succeed.
+                    if 'password' in wanted_primary_conninfo:
+                        primary_conninfo['password'] = wanted_primary_conninfo['password']

        if 'passfile' in primary_conninfo and 'password' not in primary_conninfo \
                and 'password' in wanted_primary_conninfo:
@@ -661,7 +672,7 @@ class ConfigHandler(object):
            else:
                return False

-        return all(primary_conninfo.get(p) == str(v) for p, v in wanted_primary_conninfo.items() if v is not None)
+        return all(str(primary_conninfo.get(p)) == str(v) for p, v in wanted_primary_conninfo.items() if v is not None)

    def check_recovery_conf(self, member):
        """Returns a tuple. The first boolean element indicates that recovery params don't match
@@ -697,16 +708,19 @@ class ConfigHandler(object):
            else:  # empty string, primary_conninfo is not in the config
                primary_conninfo[0] = {}

-        # when wal receiver is alive take primary_slot_name from pg_stat_wal_receiver
-        wal_receiver_primary_slot_name = self._postgresql.primary_slot_name()
-        if not wal_receiver_primary_slot_name and self._postgresql.primary_conninfo():
-            wal_receiver_primary_slot_name = ''
-        if wal_receiver_primary_slot_name is not None:
-            self._current_recovery_params['primary_slot_name'][0] = wal_receiver_primary_slot_name
+        if not self._postgresql.is_starting():
+            # when wal receiver is alive take primary_slot_name from pg_stat_wal_receiver
+            wal_receiver_primary_slot_name = self._postgresql.primary_slot_name()
+            if not wal_receiver_primary_slot_name and self._postgresql.primary_conninfo():
+                wal_receiver_primary_slot_name = ''
+            if wal_receiver_primary_slot_name is not None:
+                self._current_recovery_params['primary_slot_name'][0] = wal_receiver_primary_slot_name

        # 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)}
+                    'reload': int(self._postgresql.major_version >= 120000
+                                  and not self._postgresql.cb_called
+                                  and not self._postgresql.is_starting())}

        def record_missmatch(mtype):
            required['restart' if mtype else 'reload'] += 1
@@ -745,7 +759,12 @@ class ConfigHandler(object):
                return re.sub(r'([:\\])', r'\\\1', str(value))

            record = {n: escape(record.get(n) or '*') for n in ('host', 'port', 'user', 'password')}
-            return '{host}:{port}:*:{user}:{password}'.format(**record)
+            # 'host' could be several comma-separated hostnames, in this case
+            # we need to write on pgpass line per host
+            line = ''
+            for hostname in record.get('host').split(','):
+                line += hostname + ':{port}:*:{user}:{password}'.format(**record) + '\n'
+            return line.rstrip()

    def write_pgpass(self, record):
        line = self._pgpass_line(record)
@@ -768,6 +787,15 @@ class ConfigHandler(object):
            else:
                self._remove_file_if_exists(self._standby_signal)
                open(self._recovery_signal, 'w').close()
+
+            def restart_required(name):
+                if self._postgresql.major_version >= 140000:
+                    return False
+                return name == 'restore_command' or (self._postgresql.major_version < 130000
+                                                     and name in ('primary_conninfo', 'primary_slot_name'))
+
+            self._current_recovery_params = {n: [v, restart_required(n), self._postgresql_conf]
+                                             for n, v in recovery_params.items()}
        else:
            with ConfigWriter(self._recovery_conf) as f:
                os.chmod(self._recovery_conf, stat.S_IWRITE | stat.S_IREAD)
@@ -777,6 +805,7 @@ class ConfigHandler(object):
        for name in (self._recovery_conf, self._standby_signal, self._recovery_signal):
            self._remove_file_if_exists(name)
        self._recovery_params = {}
+        self._current_recovery_params = None

    def _sanitize_auto_conf(self):
        overwrite = False
@@ -28,13 +28,17 @@ class Rewind(object):
    def configuration_allows_rewind(data):
        return data.get('wal_log_hints setting', 'off') == 'on' or data.get('Data page checksum version', '0') != '0'

+    @property
+    def enabled(self):
+        return self._postgresql.config.get('use_pg_rewind')
+
    @property
    def can_rewind(self):
        """ check if pg_rewind executable is there and that pg_controldata indicates
            we have either wal_log_hints or checksums turned on
        """
        # low-hanging fruit: check if pg_rewind configuration is there
-        if not self._postgresql.config.get('use_pg_rewind'):
+        if not self.enabled:
            return False

        cmd = [self._postgresql.pgcommand('pg_rewind'), '--help']
@@ -46,9 +50,13 @@ class Rewind(object):
            return False
        return self.configuration_allows_rewind(self._postgresql.controldata())

+    @property
+    def should_remove_data_directory_on_diverged_timelines(self):
+        return self._postgresql.config.get('remove_data_directory_on_diverged_timelines')
+
    @property
    def can_rewind_or_reinitialize_allowed(self):
-        return self._postgresql.config.get('remove_data_directory_on_diverged_timelines') or self.can_rewind
+        return self.should_remove_data_directory_on_diverged_timelines or self.can_rewind

    def trigger_check_diverged_lsn(self):
        if self.can_rewind_or_reinitialize_allowed and self._state != REWIND_STATUS.NEED:
@@ -65,6 +73,20 @@ class Rewind(object):
        except Exception:
            return logger.exception('Exception when working with leader')

+    @staticmethod
+    def check_leader_has_run_checkpoint(conn_kwargs):
+        try:
+            with get_connection_cursor(connect_timeout=3, options='-c statement_timeout=2000', **conn_kwargs) as cur:
+                cur.execute("SELECT NOT pg_catalog.pg_is_in_recovery()" +
+                            " AND ('x' || pg_catalog.substr(pg_catalog.pg_walfile_name(" +
+                            " pg_catalog.pg_current_wal_lsn()), 1, 8))::bit(32)::int = timeline_id" +
+                            " FROM pg_catalog.pg_control_checkpoint()")
+                if not cur.fetchone()[0]:
+                    return 'leader has not run a checkpoint yet'
+        except Exception:
+            logger.exception('Exception when working with leader')
+            return 'not accessible or not healty'
+
    def _get_checkpoint_end(self, timeline, lsn):
        """The checkpoint record size in WAL depends on postgres major version and platform (memory alignment).
        Hence, the only reliable way to figure out where it ends, read the record from file with the help of pg_waldump
@@ -98,7 +120,7 @@ class Rewind(object):
        in_recovery = timeline = lsn = None
        data = self._postgresql.controldata()
        try:
-            if data.get('Database cluster state') == 'shut down in recovery':
+            if data.get('Database cluster state') in ('shut down in recovery', 'in archive recovery'):
                in_recovery = True
                lsn = data.get('Minimum recovery ending location')
                timeline = int(data.get("Min recovery ending loc's timeline"))
@@ -154,6 +176,10 @@ class Rewind(object):
        ret = member.conn_kwargs(auth)
        if not ret.get('dbname'):
            ret['dbname'] = self._postgresql.database
+        # Add target_session_attrs in case more than one hostname is specified
+        # (libpq client-side failover) making sure we hit the primary
+        if 'target_session_attrs' not in ret and self._postgresql.major_version >= 100000:
+            ret['target_session_attrs'] = 'read-write'
        return ret

    def _check_timeline_and_lsn(self, leader):
@@ -164,8 +190,14 @@ class Rewind(object):
        if isinstance(leader, Leader) and leader.member.data.get('role') != 'master':
            return

-        if not self.check_leader_is_not_in_recovery(
-                self._conn_kwargs(leader, self._postgresql.config.rewind_credentials)):
+        # We want to use replication credentials when connecting to the "postgres" database in case if
+        # `use_pg_rewind` isn't enabled and only `remove_data_directory_on_diverged_timelines` is set
+        # for Postgresql older than v11 (where Patroni can't use a dedicated user for rewind).
+        # In all other cases we will use rewind or superuser credentials.
+        check_credentials = self._postgresql.config.replication if not self.enabled and\
+            self.should_remove_data_directory_on_diverged_timelines and\
+            self._postgresql.major_version < 110000 else self._postgresql.config.rewind_credentials
+        if not self.check_leader_is_not_in_recovery(self._conn_kwargs(leader, check_credentials)):
            return

        history = need_rewind = None
@@ -200,6 +232,7 @@ class Rewind(object):
                        need_rewind = True
                    else:
                        need_rewind = switchpoint != self._get_checkpoint_end(local_timeline, local_lsn)
+                    break
                elif parent_timeline > local_timeline:
                    need_rewind = True
                    break
@@ -292,9 +325,18 @@ class Rewind(object):
        restore_command = self._postgresql.config.get('recovery_conf', {}).get('restore_command') \
            if self._postgresql.major_version < 120000 else self._postgresql.get_guc_value('restore_command')

+        # Until v15 pg_rewind expected postgresql.conf to be inside $PGDATA, which is not the case on e.g. Debian
+        pg_rewind_can_restore = restore_command and (self._postgresql.major_version >= 150000 or
+                                                     (self._postgresql.major_version >= 130000 and
+                                                      self._postgresql.config._config_dir == self._postgresql.data_dir))
+
        cmd = [self._postgresql.pgcommand('pg_rewind')]
-        if self._postgresql.major_version >= 130000 and restore_command:
+        if pg_rewind_can_restore:
            cmd.append('--restore-target-wal')
+            if self._postgresql.major_version >= 150000 and\
+                    self._postgresql.config._config_dir != self._postgresql.data_dir:
+                cmd.append('--config-file={0}'.format(self._postgresql.config.postgresql_conf))
+
        cmd.extend(['-D', self._postgresql.data_dir, '--source-server', dsn])

        while True:
@@ -310,7 +352,7 @@ class Rewind(object):
            if ret == 0:
                return True

-            if not restore_command or self._postgresql.major_version >= 130000:
+            if not restore_command or pg_rewind_can_restore:
                return False

            missing_wal = self._find_missing_wal(results['stderr']) or self._find_missing_wal(results['stdout'])
@@ -333,9 +375,14 @@ class Rewind(object):
        #   running a checkpoint or
        #   waiting until Patroni on the master will expose checkpoint_after_promote=True
        checkpoint_status = leader.checkpoint_after_promote if isinstance(leader, Leader) else None
-        if checkpoint_status is None:  # master still runs the old Patroni
-            leader_status = self._postgresql.checkpoint(self._conn_kwargs(leader, self._postgresql.config.superuser))
-            if leader_status:
+        if checkpoint_status is None:  # we are the standby-cluster leader or master still runs the old Patroni
+            # superuser credentials match rewind_credentials if the latter are not provided or we run 10 or older
+            if self._postgresql.config.superuser == self._postgresql.config.rewind_credentials:
+                leader_status = self._postgresql.checkpoint(
+                        self._conn_kwargs(leader, self._postgresql.config.superuser))
+            else:  # we run 11+ and have a dedicated pg_rewind user
+                leader_status = self.check_leader_has_run_checkpoint(r)
+            if leader_status:  # we tried to run/check for a checkpoint on the remote leader, but it failed
                return logger.warning('Can not use %s for rewind: %s', leader.name, leader_status)
        elif not checkpoint_status:
            return logger.info('Waiting for checkpoint on %s before rewind', leader.name)
@@ -344,19 +391,22 @@ class Rewind(object):

        if self.pg_rewind(r):
            self._state = REWIND_STATUS.SUCCESS
-        elif not self.check_leader_is_not_in_recovery(r):
-            logger.warning('Failed to rewind because master %s become unreachable', leader.name)
        else:
-            logger.error('Failed to rewind from healty master: %s', leader.name)
-
-            for name in ('remove_data_directory_on_rewind_failure', 'remove_data_directory_on_diverged_timelines'):
-                if self._postgresql.config.get(name):
-                    logger.warning('%s is set. removing...', name)
-                    self._postgresql.remove_data_directory()
-                    self._state = REWIND_STATUS.INITIAL
-                    break
+            if not self.check_leader_is_not_in_recovery(r):
+                logger.warning('Failed to rewind because master %s become unreachable', leader.name)
+                if not self.can_rewind:  # It is possible that the previous attempt damaged pg_control file!
+                    self._state = REWIND_STATUS.FAILED
            else:
+                logger.error('Failed to rewind from healty master: %s', leader.name)
                self._state = REWIND_STATUS.FAILED
+
+            if self.failed:
+                for name in ('remove_data_directory_on_rewind_failure', 'remove_data_directory_on_diverged_timelines'):
+                    if self._postgresql.config.get(name):
+                        logger.warning('%s is set. removing...', name)
+                        self._postgresql.remove_data_directory()
+                        self._state = REWIND_STATUS.INITIAL
+                        break
        return False

    def reset_state(self):
@@ -8,7 +8,7 @@ from contextlib import contextmanager

 from .connection import get_connection_cursor
 from .misc import format_lsn
-from ..psycopg import UndefinedFile
+from ..psycopg import OperationalError

 logger = logging.getLogger(__name__)

@@ -36,7 +36,7 @@ class SlotsHandler(object):
    def __init__(self, postgresql):
        self._postgresql = postgresql
        self._replication_slots = {}  # already existing replication slots
-        self._unready_logical_slots = set()
+        self._unready_logical_slots = {}
        self.schedule()

    def _query(self, sql, *params):
@@ -95,7 +95,7 @@ class SlotsHandler(object):
            self._replication_slots = replication_slots
            self._schedule_load_slots = False
            if self._force_readiness_check:
-                self._unready_logical_slots = set(n for n, v in replication_slots.items() if v['type'] == 'logical')
+                self._unready_logical_slots = {n: None for n, v in replication_slots.items() if v['type'] == 'logical'}
                self._force_readiness_check = False

    def ignore_replication_slot(self, cluster, name):
@@ -201,7 +201,7 @@ class SlotsHandler(object):
                                    (name, format_lsn(int(cluster.slots[name]))))
                    except Exception as e:
                        logger.error("Failed to advance logical replication slot '%s': %r", name, e)
-                        if isinstance(e, UndefinedFile):
+                        if isinstance(e, OperationalError) and e.diag.sqlstate == '58P01':  # WAL file is gone
                            create_slots.append(name)
                        self._schedule_load_slots = True
        return create_slots
@@ -245,35 +245,68 @@ class SlotsHandler(object):
            slot_name = cluster.get_my_slot_name_on_primary(self._postgresql.name, replicatefrom)
            try:
                with self._get_leader_connection_cursor(cluster.leader) as cur:
-                    cur.execute("SELECT catalog_xmin FROM pg_catalog.pg_get_replication_slots()"
-                                " WHERE NOT pg_catalog.pg_is_in_recovery() AND slot_name = %s", (slot_name,))
-                    if cur.rowcount < 1:
+                    cur.execute("SELECT slot_name, catalog_xmin FROM pg_catalog.pg_get_replication_slots()"
+                                " WHERE NOT pg_catalog.pg_is_in_recovery() AND slot_name = ANY(%s)",
+                                ([n for n, v in self._unready_logical_slots.items() if v is None] + [slot_name],))
+                    slots = {row[0]: row[1] for row in cur}
+                    if slot_name not in slots:
                        return logger.warning('Physical slot %s does not exist on the primary', slot_name)
-                    catalog_xmin = cur.fetchone()[0]
+                    catalog_xmin = slots.pop(slot_name)
            except Exception as e:
                return logger.error("Failed to check %s physical slot on the primary: %r", slot_name, e)
+            # Remember catalog_xmin of logical slots on the primary when catalog_xmin of
+            # the physical slot became valid. Logical slots on replica will be safe to use after
+            # promote when catalog_xmin of the physical slot overtakes these values.
+            if catalog_xmin:
+                for name, value in slots.items():
+                    self._unready_logical_slots[name] = value
+            else:  # Replica isn't streaming or the hot_standby_feedback isn't enabled
+                try:
+                    cur = self._query("SELECT pg_catalog.current_setting('hot_standby_feedback')::boolean")
+                    if not cur.fetchone()[0]:
+                        return logger.error('Logical slot failover requires "hot_standby_feedback".'
+                                            ' Please check postgresql.auto.conf')
+                except Exception as e:
+                    return logger.error('Failed to check the hot_standby_feedback setting: %r', e)
+
        for name in list(self._unready_logical_slots):
            value = self._replication_slots.get(name)
-            if not value or catalog_xmin <= value['catalog_xmin']:
-                self._unready_logical_slots.remove(name)
+            # The logical slot on a replica is safe to use when the physical replica slot on the primary:
+            # 1. has a nonzero/non-null catalog_xmin
+            # 2. has a catalog_xmin that is not newer (greater) than the catalog_xmin of any slot on the standby
+            # 3. overtook the catalog_xmin of remembered values of logical slots on the primary.
+            if not value or self._unready_logical_slots[name] <= catalog_xmin <= value['catalog_xmin']:
+                del self._unready_logical_slots[name]
                if value:
                    logger.info('Logical slot %s is safe to be used after a failover', name)

-    def copy_logical_slots(self, leader, slots):
+    def copy_logical_slots(self, cluster, create_slots):
+        leader = cluster.leader
+        slots = cluster.get_replication_slots(self._postgresql.name, 'replica', False, self._postgresql.major_version)
        with self._get_leader_connection_cursor(leader) as cur:
            try:
-                cur.execute("SELECT slot_name, catalog_xmin, "
+                cur.execute("SELECT slot_name, slot_type, datname, plugin, catalog_xmin, "
                            "pg_catalog.pg_wal_lsn_diff(confirmed_flush_lsn, '0/0')::bigint, "
                            "pg_catalog.pg_read_binary_file('pg_replslot/' || slot_name || '/state')"
-                            " FROM pg_catalog.pg_get_replication_slots() WHERE NOT pg_catalog.pg_is_in_recovery()"
-                            " AND slot_name = ANY(%s)", (slots,))
-                slots = {r[0]: {'catalog_xmin': r[1], 'confirmed_flush_lsn': r[2], 'data': r[3]} for r in cur}
+                            " FROM pg_catalog.pg_get_replication_slots() JOIN pg_catalog.pg_database ON datoid = oid"
+                            " WHERE NOT pg_catalog.pg_is_in_recovery() AND slot_name = ANY(%s)", (create_slots,))
+
+                create_slots = {}
+                for r in cur:
+                    if r[0] in slots:  # slot_name is defined in the global configuration
+                        slot = {'type': r[1], 'database': r[2], 'plugin': r[3],
+                                'catalog_xmin': r[4], 'confirmed_flush_lsn': r[5], 'data': r[6]}
+                        if compare_slots(slot, slots[r[0]]):
+                            create_slots[r[0]] = slot
+                        else:
+                            logger.warning('Will not copy the logical slot "%s" due to the configuration mismatch: ' +
+                                           'configuration=%s, slot on the primary=%s', r[0], slots[r[0]], slot)
            except Exception as e:
                logger.error("Failed to copy logical slots from the %s via postgresql connection: %r", leader.name, e)

-        if isinstance(slots, dict) and self._postgresql.stop():
+        if isinstance(create_slots, dict) and create_slots and self._postgresql.stop():
            pg_replslot_dir = os.path.join(self._postgresql.data_dir, 'pg_replslot')
-            for name, value in slots.items():
+            for name, value in create_slots.items():
                slot_dir = os.path.join(pg_replslot_dir, name)
                slot_tmp_dir = slot_dir + '.tmp'
                if os.path.exists(slot_tmp_dir):
@@ -288,7 +321,7 @@ class SlotsHandler(object):
                    shutil.rmtree(slot_dir)
                os.rename(slot_tmp_dir, slot_dir)
                fsync_dir(slot_dir)
-                self._unready_logical_slots.add(name)
+                self._unready_logical_slots[name] = None
            fsync_dir(pg_replslot_dir)
            self._postgresql.start()

@@ -300,4 +333,4 @@ class SlotsHandler(object):
    def on_promote(self):
        if self._unready_logical_slots:
            logger.warning('Logical replication slots that might be unsafe to use after promote: %s',
-                           self._unready_logical_slots)
+                           set(self._unready_logical_slots))
@@ -99,9 +99,11 @@ class String(namedtuple('String', 'version_from,version_till')):
 #  key - parameter name
 #  value - tuple or multiple tuples if something was changing in GUC across postgres versions
 parameters = CaseInsensitiveDict({
+    'allow_in_place_tablespaces': Bool(150000, None),
    'allow_system_table_mods': Bool(90300, None),
    'application_name': String(90300, None),
    'archive_command': String(90300, None),
+    'archive_library': String(150000, None),
    'archive_mode': (
        Bool(90300, 90500),
        EnumBool(90500, None, ('always',))
@@ -151,14 +153,17 @@ 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_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',)),
+    'compute_query_id': (
+        EnumBool(140000, 150000, ('auto',)),
+        EnumBool(150000, None, ('auto', 'regress'))
+    ),
    'config_file': String(90300, None),
    'constraint_exclusion': EnumBool(90300, None, ('partition',)),
    'cpu_index_tuple_cost': Real(90300, None, 0, 1.79769e+308, None),
@@ -170,6 +175,7 @@ parameters = CaseInsensitiveDict({
    'DateStyle': String(90300, None),
    'db_user_namespace': Bool(90300, None),
    'deadlock_timeout': Integer(90300, None, 1, 2147483647, 'ms'),
+    'debug_discard_caches': Integer(150000, None, 0, 0, None),
    'debug_pretty_print': Bool(90300, None),
    'debug_print_parse': Bool(90300, None),
    'debug_print_plan': Bool(90300, None),
@@ -194,12 +200,14 @@ parameters = CaseInsensitiveDict({
    'enable_async_append': Bool(140000, None),
    'enable_bitmapscan': Bool(90300, None),
    'enable_gathermerge': Bool(100000, None),
+    'enable_group_by_reordering': Bool(150000, None),
    'enable_hashagg': Bool(90300, None),
    'enable_hashjoin': Bool(90300, None),
    'enable_incremental_sort': Bool(130000, None),
    'enable_indexonlyscan': Bool(90300, None),
    'enable_indexscan': Bool(90300, None),
    'enable_material': Bool(90300, None),
+    'enable_memoize': Bool(150000, None),
    'enable_mergejoin': Bool(90300, None),
    'enable_nestloop': Bool(90300, None),
    'enable_parallel_append': Bool(110000, None),
@@ -234,10 +242,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'),
+    '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'),
+    '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),
@@ -294,6 +302,7 @@ parameters = CaseInsensitiveDict({
    'log_replication_commands': Bool(90500, None),
    'log_rotation_age': Integer(90300, None, 0, 35791394, 'min'),
    'log_rotation_size': Integer(90300, None, 0, 2097151, 'kB'),
+    'log_startup_progress_interval': Integer(150000, None, 0, 2147483647, 'ms'),
    'log_statement': Enum(90300, None, ('none', 'ddl', 'mod', 'all')),
    'log_statement_sample_rate': Real(130000, None, 0, 1, None),
    'log_statement_stats': Bool(90300, None),
@@ -344,7 +353,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_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'),
@@ -368,6 +377,8 @@ parameters = CaseInsensitiveDict({
    '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')),
+    'recovery_prefetch': EnumBool(150000, None, ('try',)),
+    'recursive_worktable_factor': Real(150000, None, 0.001, 1e+06, None),
    'remove_temp_files_after_crash': Bool(140000, None),
    'replacement_sort_tuples': Integer(90600, 110000, 0, 2147483647, None),
    'restart_after_crash': Bool(90300, None),
@@ -397,7 +408,8 @@ parameters = CaseInsensitiveDict({
    'ssl_renegotiation_limit': Integer(90300, 90500, 0, 2147483647, 'kB'),
    'standard_conforming_strings': Bool(90300, None),
    'statement_timeout': Integer(90300, None, 0, 2147483647, 'ms'),
-    'stats_temp_directory': String(90300, None),
+    'stats_fetch_consistency': Enum(150000, None, ('none', 'cache', 'snapshot')),
+    'stats_temp_directory': String(90300, 150000),
    'superuser_reserved_connections': (
        Integer(90300, 90600, 0, 8388607, None),
        Integer(90600, None, 0, 262143, None)
@@ -456,15 +468,19 @@ 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_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_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'),
-    'wal_compression': Bool(90500, None),
+    'wal_compression': (
+        Bool(90500, 150000),
+        EnumBool(150000, None, ('pglz', 'lz4', 'zstd'))
+    ),
    'wal_consistency_checking': String(100000, None),
+    'wal_decode_buffer_size': Integer(150000, None, 65536, 1073741823, 'B'),
    'wal_init_zero': Bool(120000, None),
    'wal_keep_segments': Integer(90300, 130000, 0, 2147483647, None),
    'wal_keep_size': Integer(130000, None, 0, 2147483647, 'MB'),
@@ -1,5 +1,4 @@
-__all__ = ['connect', 'quote_ident', 'quote_literal', 'DatabaseError',
-           'Error', 'OperationalError', 'ProgrammingError', 'UndefinedFile']
+__all__ = ['connect', 'quote_ident', 'quote_literal', 'DatabaseError', 'Error', 'OperationalError', 'ProgrammingError']

 _legacy = False
 try:
@@ -8,7 +7,6 @@ try:
    if parse_version(__version__) < MIN_PSYCOPG2:
        raise ImportError
    from psycopg2 import connect, Error, DatabaseError, OperationalError, ProgrammingError
-    from psycopg2.errors import UndefinedFile
    from psycopg2.extensions import adapt

    try:
@@ -23,7 +21,6 @@ try:
        return value.getquoted().decode('utf-8')
 except ImportError:
    from psycopg import connect as _connect, sql, Error, DatabaseError, OperationalError, ProgrammingError
-    from psycopg.errors import UndefinedFile

    def connect(*args, **kwargs):
        ret = _connect(*args, **kwargs)
@@ -34,6 +34,9 @@ class PatroniRequest(object):

        if self._apply_ssl_file_param(config, 'cert'):
            self._apply_ssl_file_param(config, 'key')
+
+            password = self._get_cfg_value(config, 'keyfile_password')
+            self._apply_pool_param('key_password', password)
        else:
            self._pool.connection_pool_kw.pop('key_file', None)

@@ -3,10 +3,12 @@
 import json
 import logging
 import sys
-import boto.ec2
+import boto3

-from patroni.utils import Retry, RetryFailedError
-from patroni.request import get as requests_get
+from ..utils import Retry, RetryFailedError
+from ..request import get as requests_get
+
+from botocore.exceptions import ClientError

 logger = logging.getLogger(__name__)

@@ -16,7 +18,7 @@ class AWSConnection(object):
    def __init__(self, cluster_name):
        self.available = False
        self.cluster_name = cluster_name if cluster_name is not None else 'unknown'
-        self._retry = Retry(deadline=300, max_delay=30, max_tries=-1, retry_exceptions=(boto.exception.StandardError,))
+        self._retry = Retry(deadline=300, max_delay=30, max_tries=-1, retry_exceptions=(ClientError,))
        try:
            # get the instance id
            r = requests_get('http://169.254.169.254/latest/dynamic/instance-identity/document', timeout=2.1)
@@ -42,20 +44,22 @@ class AWSConnection(object):

    def _tag_ebs(self, conn, role):
        """ set tags, carrying the cluster name, instance role and instance id for the EBS storage """
-        tags = {'Name': 'spilo_' + self.cluster_name, 'Role': role, 'Instance': self.instance_id}
-        volumes = conn.get_all_volumes(filters={'attachment.instance-id': self.instance_id})
-        conn.create_tags([v.id for v in volumes], tags)
+        tags = [{'Key': 'Name', 'Value': 'spilo_' + self.cluster_name},
+                {'Key': 'Role', 'Value': role},
+                {'Key': 'Instance', 'Value': self.instance_id}]
+        volumes = conn.volumes.filter(Filters=[{'Name': 'attachment.instance-id', 'Values': [self.instance_id]}])
+        conn.create_tags(Resources=[v.id for v in volumes], Tags=tags)

    def _tag_ec2(self, conn, role):
        """ tag the current EC2 instance with a cluster role """
-        tags = {'Role': role}
-        conn.create_tags([self.instance_id], tags)
+        tags = [{'Key': 'Role', 'Value': role}]
+        conn.create_tags(Resources=[self.instance_id], Tags=tags)

    def on_role_change(self, new_role):
        if not self.available:
            return False
        try:
-            conn = self.retry(boto.ec2.connect_to_region, self.region)
+            conn = boto3.resource('ec2', region_name=self.region)
            self.retry(self._tag_ec2, conn, new_role)
            self.retry(self._tag_ebs, conn, new_role)
        except RetryFailedError:
@@ -362,7 +362,7 @@ def polling_loop(timeout, interval=1):
 def split_host_port(value, default_port):
    t = value.rsplit(':', 1)
    if ':' in t[0]:
-        t[0] = t[0].strip('[]')
+        t[0] = ','.join([h.strip().strip('[]') for h in t[0].split(',')])
    t.append(default_port)
    return t[0], int(t[1])

@@ -1 +1 @@
-__version__ = '2.1.2'
+__version__ = '2.1.4'
@@ -16,10 +16,10 @@ IOC_DIRBITS = 2

 # Non-generic platform special cases
 machine = platform.machine()
-if machine in ['mips', 'sparc', 'powerpc', 'ppc64']:  # pragma: no cover
+if machine in ['mips', 'sparc', 'powerpc', 'ppc64', 'ppc64le']:  # pragma: no cover
    IOC_SIZEBITS = 13
    IOC_DIRBITS = 3
-    IOC_NONE, IOC_WRITE, IOC_READ = 1, 2, 4
+    IOC_NONE, IOC_WRITE, IOC_READ = 1, 4, 2
 elif machine == 'parisc':  # pragma: no cover
    IOC_WRITE, IOC_READ = 2, 1

@@ -91,7 +91,7 @@ bootstrap:
  # Some additional users users which needs to be created after initializing new cluster
  users:
    admin:
-      password: admin
+      password: admin%
      options:
        - createrole
        - createdb
@@ -119,7 +119,7 @@ postgresql:
    # Fully qualified kerberos ticket file for the running user
    # same as KRB5CCNAME used by the GSS
 #   krb_server_keyfile: /var/spool/keytabs/postgres
-    unix_socket_directories: '.'
+    unix_socket_directories: '..'  # parent directory of data_dir
  # Additional fencing script executed after acquiring the leader lock but before promoting the replica
  #pre_promote: /path/to/pre_promote.sh

@@ -85,7 +85,7 @@ bootstrap:
  # Some additional users users which needs to be created after initializing new cluster
  users:
    admin:
-      password: admin
+      password: admin%
      options:
        - createrole
        - createdb
@@ -113,7 +113,7 @@ postgresql:
    # Fully qualified kerberos ticket file for the running user
    # same as KRB5CCNAME used by the GSS
 #    krb_server_keyfile: /var/spool/keytabs/postgres
-    unix_socket_directories: '.'
+    unix_socket_directories: '..'  # parent directory of data_dir
  basebackup:
      - verbose
      - max-rate: 100M
@@ -82,7 +82,7 @@ bootstrap:
  # Some additional users users which needs to be created after initializing new cluster
  users:
    admin:
-      password: admin
+      password: admin%
      options:
        - createrole
        - createdb
@@ -110,7 +110,7 @@ postgresql:
    # Fully qualified kerberos ticket file for the running user
    # same as KRB5CCNAME used by the GSS
 #    krb_server_keyfile: /var/spool/keytabs/postgres
-    unix_socket_directories: '.'
+    unix_socket_directories: '..'  # parent directory of data_dir
 tags:
    nofailover: false
    noloadbalance: false
@@ -1,6 +1,6 @@
 urllib3>=1.19.1,!=1.21
 ipaddress; python_version=="2.7"
-boto
+boto3
 PyYAML
 six >= 1.7
 kazoo>=1.3.1
@@ -5,6 +5,7 @@
 """

 import inspect
+import logging
 import os
 import sys

@@ -22,11 +23,10 @@ AUTHOR_EMAIL = '[email protected], [email protected], alexk
 KEYWORDS = 'etcd governor patroni postgresql postgres ha haproxy confd' +\
    ' zookeeper exhibitor consul streaming replication kubernetes k8s'

-EXTRAS_REQUIRE = {'aws': ['boto'], 'etcd': ['python-etcd'], 'etcd3': ['python-etcd'],
+EXTRAS_REQUIRE = {'aws': ['boto3'], 'etcd': ['python-etcd'], 'etcd3': ['python-etcd'],
                  'consul': ['python-consul'], 'exhibitor': ['kazoo'], 'zookeeper': ['kazoo'],
                  'kubernetes': [], 'raft': ['pysyncobj', 'cryptography']}
 COVERAGE_XML = True
-COVERAGE_HTML = False

 # Add here all kinds of additional classifiers as defined under
 # https://pypi.python.org/pypi?%3Aaction=list_classifiers
@@ -49,29 +49,29 @@ CLASSIFIERS = [
    'Programming Language :: Python :: 3.7',
    'Programming Language :: Python :: 3.8',
    'Programming Language :: Python :: 3.9',
+    'Programming Language :: Python :: 3.10',
    'Programming Language :: Python :: Implementation :: CPython',
 ]

-CONSOLE_SCRIPTS = ['patroni = patroni:main',
+CONSOLE_SCRIPTS = ['patroni = patroni.__main__:main',
                   'patronictl = patroni.ctl:ctl',
                   'patroni_raft_controller = patroni.raft_controller:main',
                   "patroni_wale_restore = patroni.scripts.wale_restore:main",
                   "patroni_aws = patroni.scripts.aws:main"]


-class Flake8(Command):
-
+class _Command(Command):
    user_options = []

    def initialize_options(self):
-        from flake8.main import application
-
-        self.flake8 = application.Application()
-        self.flake8.initialize([])
+        pass

    def finalize_options(self):
        pass

+
+class Flake8(_Command):
+
    def package_files(self):
        seen_package_directories = ()
        directories = self.distribution.package_dir or {}
@@ -93,68 +93,31 @@ class Flake8(Command):
        return [package for package in self.package_files()] + ['tests', 'setup.py']

    def run(self):
-        self.flake8.run_checks(self.targets())
-        self.flake8.formatter.start()
-        self.flake8.report_errors()
-        self.flake8.report_statistics()
-        self.flake8.report_benchmarks()
-        self.flake8.formatter.stop()
-        try:
-            self.flake8.exit()
-        except SystemExit as e:
-            # Cause system exit only if exit code is not zero (terminates
-            # other possibly remaining/pending setuptools commands).
-            if e.code:
-                raise
+        from flake8.main import application
+
+        logging.getLogger().setLevel(logging.ERROR)
+        flake8 = application.Application()
+        flake8.run(self.targets())
+        flake8.exit()


-class PyTest(Command):
+class PyTest(_Command):

-    user_options = [('cov=', None, 'Run coverage'), ('cov-xml=', None, 'Generate junit xml report'),
-                    ('cov-html=', None, 'Generate junit html report')]
-
-    def initialize_options(self):
-        self.cov = []
-        self.cov_xml = False
-        self.cov_html = False
-
-    def finalize_options(self):
-        if self.cov_xml or self.cov_html:
-            self.cov = ['--cov', MAIN_PACKAGE, '--cov-report', 'term-missing']
-            if self.cov_xml:
-                self.cov.extend(['--cov-report', 'xml'])
-            if self.cov_html:
-                self.cov.extend(['--cov-report', 'html'])
-
-    def run_tests(self):
+    def run(self):
        try:
            import pytest
        except Exception:
            raise RuntimeError('py.test is not installed, run: pip install pytest')

-        import logging
-        silence = logging.WARNING
-        logging.basicConfig(format='%(asctime)s %(levelname)s: %(message)s', level=os.getenv('LOGLEVEL', silence))
+        logging.getLogger().setLevel(logging.WARNING)

        args = ['--verbose', 'tests', '--doctest-modules', MAIN_PACKAGE] +\
-            ['-s' if logging.getLogger().getEffectiveLevel() < silence else '--capture=fd']
-        if self.cov:
-            args += self.cov
+            ['-s' if logging.getLogger().getEffectiveLevel() < logging.WARNING else '--capture=fd'] +\
+            ['--cov', MAIN_PACKAGE, '--cov-report', 'term-missing', '--cov-report', 'xml']

        errno = pytest.main(args=args)
        sys.exit(errno)

-    def run(self):
-        from pkg_resources import evaluate_marker
-
-        requirements = set(self.distribution.install_requires + ['mock>=2.0.0', 'pytest-cov', 'pytest'])
-        for k, v in self.distribution.extras_require.items():
-            if not k.startswith(':') or evaluate_marker(k[1:]):
-                requirements.update(v)
-
-        self.distribution.fetch_build_eggs(list(requirements))
-        self.run_tests()
-

 def read(fname):
    with open(os.path.join(__location__, fname)) as fd:
@@ -162,6 +125,8 @@ def read(fname):


 def setup_package(version):
+    logging.basicConfig(format='%(message)s', level=os.getenv('LOGLEVEL', logging.WARNING))
+
    # Assemble additional setup commands
    cmdclass = {'test': PyTest, 'flake8': Flake8}

@@ -184,12 +149,6 @@ def setup_package(version):
        if not extra:
            install_requires.append(r)

-    command_options = {'test': {}}
-    if COVERAGE_XML:
-        command_options['test']['cov_xml'] = 'setup.py', True
-    if COVERAGE_HTML:
-        command_options['test']['cov_html'] = 'setup.py', True
-
    setup(
        name=NAME,
        version=version,
@@ -206,9 +165,7 @@ def setup_package(version):
        python_requires='>=2.7',
        install_requires=install_requires,
        extras_require=EXTRAS_REQUIRE,
-        setup_requires='flake8',
        cmdclass=cmdclass,
-        command_options=command_options,
        entry_points={'console_scripts': CONSOLE_SCRIPTS},
    )

@@ -216,7 +173,8 @@ def setup_package(version):
 if __name__ == '__main__':
    old_modules = sys.modules.copy()
    try:
-        from patroni import check_psycopg, fatal, __version__
+        from patroni import check_psycopg, fatal
+        from patroni.version import __version__
    finally:
        sys.modules.clear()
        sys.modules.update(old_modules)
@@ -91,10 +91,10 @@ class MockCursor(object):
            raise psycopg.OperationalError()
        elif sql.startswith('RetryFailedError'):
            raise RetryFailedError('retry')
-        elif sql.startswith('SELECT catalog_xmin'):
-            self.results = [(100, 501)]
        elif sql.startswith('SELECT slot_name, catalog_xmin'):
-            self.results = [('ls', 100, 500, b'123456')]
+            self.results = [('postgresql0', 100), ('ls', 100)]
+        elif sql.startswith('SELECT slot_name, slot_type, datname, plugin, catalog_xmin'):
+            self.results = [('ls', 'logical', 'a', 'b', 100, 500, b'123456')]
        elif sql.startswith('SELECT slot_name'):
            self.results = [('blabla', 'physical'), ('foobar', 'physical'), ('ls', 'logical', 'a', 'b', 5, 100, 500)]
        elif sql.startswith('SELECT CASE WHEN pg_catalog.pg_is_in_recovery()'):
@@ -190,7 +190,8 @@ class PostgresInit(unittest.TestCase):
                             'krbsrvname': 'postgres', 'pgpass': os.path.join(data_dir, 'pgpass0'),
                             'listen': '127.0.0.2, 127.0.0.3:5432', 'connect_address': '127.0.0.2:5432',
                             'authentication': {'superuser': {'username': 'foo', 'password': 'test'},
-                                                'replication': {'username': '', 'password': 'rep-pass'}},
+                                                'replication': {'username': '', 'password': 'rep-pass'},
+                                                'rewind': {'username': 'rewind', 'password': 'test'}},
                             'remove_data_directory_on_rewind_failure': True,
                             'use_pg_rewind': True, 'pg_ctl_timeout': 'bla',
                             'parameters': self._PARAMETERS,
@@ -169,6 +169,7 @@ class TestRestApiHandler(unittest.TestCase):
        MockRestApiServer(RestApiHandler, 'GET /replica?lag=10MB')
        MockRestApiServer(RestApiHandler, 'GET /replica?lag=10485760')
        MockRestApiServer(RestApiHandler, 'GET /read-only')
+        MockRestApiServer(RestApiHandler, 'GET /read-only-sync')
        with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={})):
            MockRestApiServer(RestApiHandler, 'GET /replica')
        with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={'role': 'master'})):
@@ -180,6 +181,8 @@ class TestRestApiHandler(unittest.TestCase):
        MockPatroni.dcs.cluster.is_synchronous_mode = Mock(return_value=True)
        with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={'role': 'replica'})):
            MockRestApiServer(RestApiHandler, 'GET /synchronous')
+        with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={'role': 'replica'})):
+            MockRestApiServer(RestApiHandler, 'GET /read-only-sync')
        with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={'role': 'replica'})):
            MockPatroni.dcs.cluster.sync.members = []
            MockRestApiServer(RestApiHandler, 'GET /asynchronous')
@@ -549,7 +552,8 @@ class TestRestApiServer(unittest.TestCase):
        self.assertRaises(ValueError, MockRestApiServer, None, '', bad_config)
        self.assertRaises(ValueError, self.srv.reload_config, bad_config)
        self.assertRaises(ValueError, self.srv.reload_config, {})
-        with patch.object(socket.socket, 'setsockopt', Mock(side_effect=socket.error)):
+        with patch.object(socket.socket, 'setsockopt', Mock(side_effect=socket.error)), \
+                patch.object(MockRestApiServer, 'server_close', Mock()):
            self.srv.reload_config({'listen': ':8008'})

    @patch.object(MockPatroni, 'dcs')
@@ -1,4 +1,4 @@
-import boto.ec2
+import botocore
 import sys
 import unittest
 import urllib3
@@ -8,21 +8,27 @@ from collections import namedtuple
 from patroni.scripts.aws import AWSConnection, main as _main


-class MockEc2Connection(object):
+class MockVolumes(object):

    @staticmethod
-    def get_all_volumes(*args, **kwargs):
+    def filter(*args, **kwargs):
        oid = namedtuple('Volume', 'id')
        return [oid(id='a'), oid(id='b')]

+
+class MockEc2Connection(object):
+
+    volumes = MockVolumes()
+
    @staticmethod
-    def create_tags(objects, *args, **kwargs):
-        if len(objects) == 0:
-            raise boto.exception.BotoServerError(503, 'Service Unavailable', 'Request limit exceeded')
+    def create_tags(Resources, **kwargs):
+        if len(Resources) == 0:
+            raise botocore.exceptions.ClientError({'Error': {'Code': 503, 'Message': 'Request limit exceeded'}},
+                                                  'create_tags')
        return True


-@patch('boto.ec2.connect_to_region', Mock(return_value=MockEc2Connection()))
+@patch('boto3.resource', Mock(return_value=MockEc2Connection()))
 class TestAWSConnection(unittest.TestCase):

    @patch('patroni.scripts.aws.requests_get', Mock(return_value=urllib3.HTTPResponse(
@@ -32,7 +38,7 @@ class TestAWSConnection(unittest.TestCase):

    def test_on_role_change(self):
        self.assertTrue(self.conn.on_role_change('master'))
-        with patch.object(MockEc2Connection, 'get_all_volumes', Mock(return_value=[])):
+        with patch.object(MockVolumes, 'filter', Mock(return_value=[])):
            self.conn._retry.max_tries = 1
            self.assertFalse(self.conn.on_role_change('master'))

@@ -46,8 +52,7 @@ class TestAWSConnection(unittest.TestCase):
        conn = AWSConnection('test')
        self.assertFalse(conn.aws_available())

-    @patch('patroni.scripts.aws.requests_get', Mock(return_value=urllib3.HTTPResponse(
-        status=200, body=b'{"instanceId": "012345", "region": "eu-west-1"}')))
+    @patch('patroni.scripts.aws.requests_get', Mock(return_value=urllib3.HTTPResponse(status=503, body=b'Error')))
    @patch('sys.exit', Mock())
    def test_main(self):
        self.assertIsNone(_main())
@@ -27,8 +27,8 @@ class TestCancellableSubprocess(unittest.TestCase):
    def test_cancel(self):
        self.c._process = Mock()
        self.c._process.is_running.return_value = True
-        self.c._process.children.side_effect = psutil.Error()
-        self.c._process.suspend.side_effect = psutil.Error()
+        self.c._process.children.side_effect = psutil.NoSuchProcess(123)
+        self.c._process.suspend.side_effect = psutil.AccessDenied()
        self.c.cancel()
        self.c._process.is_running.side_effect = [True, False]
        self.c.cancel()
@@ -91,7 +91,7 @@ class TestConsul(unittest.TestCase):
        Consul({'ttl': 30, 'scope': 't_', 'name': 'p', 'url': 'https://l:1', 'retry_timeout': 10,
                'verify': 'on', 'cert': 'bar', 'cacert': 'buz', 'register_service': True})
        self.c = Consul({'ttl': 30, 'scope': 'test', 'name': 'postgresql1', 'host': 'localhost:1', 'retry_timeout': 10,
-                         'register_service': True})
+                         'register_service': True, 'service_check_tls_server_name': True})
        self.c._base_path = '/service/good'
        self.c.get_cluster()

@@ -284,6 +284,18 @@ class TestHa(PostgresInit):
            self.ha.patroni.config.set_dynamic_configuration({'maximum_lag_on_failover': 10})
            self.assertEqual(self.ha.run_cycle(), 'terminated crash recovery because of startup timeout')

+    @patch.object(Rewind, 'ensure_clean_shutdown', Mock())
+    @patch.object(Rewind, 'rewind_or_reinitialize_needed_and_possible', Mock(return_value=True))
+    @patch.object(Rewind, 'can_rewind', PropertyMock(return_value=True))
+    def test_crash_recovery_before_rewind(self):
+        self.p.is_leader = false
+        self.p.is_running = false
+        self.p.controldata = lambda: {'Database cluster state': 'in archive recovery',
+                                      'Database system identifier': SYSID}
+        self.ha._rewind.trigger_check_diverged_lsn()
+        self.ha.cluster = get_cluster_initialized_with_leader()
+        self.assertEqual(self.ha.run_cycle(), 'doing crash recovery in a single user mode')
+
    @patch.object(Rewind, 'rewind_or_reinitialize_needed_and_possible', Mock(return_value=True))
    @patch.object(Rewind, 'can_rewind', PropertyMock(return_value=True))
    def test_recover_with_rewind(self):
@@ -302,6 +314,7 @@ class TestHa(PostgresInit):
            self.assertEqual(self.ha.run_cycle(), 'fake')

    @patch.object(Rewind, 'rewind_or_reinitialize_needed_and_possible', Mock(return_value=True))
+    @patch.object(Rewind, 'should_remove_data_directory_on_diverged_timelines', PropertyMock(return_value=True))
    @patch.object(Bootstrap, 'create_replica', Mock(return_value=1))
    def test_recover_with_reinitialize(self):
        self.p.is_running = false
@@ -1199,9 +1212,12 @@ class TestHa(PostgresInit):
    @patch('os.rename', Mock())
    @patch('patroni.postgresql.Postgresql.is_starting', Mock(return_value=False))
    @patch.object(builtins, 'open', mock_open())
-    @patch.object(SlotsHandler, 'sync_replication_slots', Mock(return_value=['foo']))
+    @patch.object(ConfigHandler, 'check_recovery_conf', Mock(return_value=(False, False)))
+    @patch.object(Postgresql, 'major_version', PropertyMock(return_value=130000))
+    @patch.object(SlotsHandler, 'sync_replication_slots', Mock(return_value=['ls']))
    def test_follow_copy(self):
        self.ha.cluster.is_unlocked = false
+        self.ha.cluster.config.data['slots'] = {'ls': {'database': 'a', 'plugin': 'b'}}
        self.p.is_leader = false
        self.assertTrue(self.ha.run_cycle().startswith('Copying logical slots'))

@@ -1,9 +1,10 @@
+import datetime
 import json
 import socket
 import time
 import unittest

-from mock import Mock, mock_open, patch
+from mock import Mock, PropertyMock, mock_open, patch
 from patroni.dcs.kubernetes import k8s_client, k8s_config, K8sConfig, K8sConnectionFailed,\
        K8sException, K8sObject, Kubernetes, KubernetesError, KubernetesRetriableException,\
        Retry, RetryFailedError, SERVICE_HOST_ENV_NAME, SERVICE_PORT_ENV_NAME
@@ -79,6 +80,28 @@ class TestK8sConfig(unittest.TestCase):
                self.assertRaises(k8s_config.ConfigException, k8s_config.load_incluster_config)
            k8s_config.load_incluster_config()
            self.assertEqual(k8s_config.server, 'https://a:1')
+            self.assertEqual(k8s_config.headers.get('authorization'), 'Bearer a')
+
+    def test_refresh_token(self):
+        with patch('os.environ', {SERVICE_HOST_ENV_NAME: 'a', SERVICE_PORT_ENV_NAME: '1'}),\
+                patch('os.path.isfile', Mock(side_effect=[True, True, False, True, True, True])),\
+                patch.object(builtins, 'open', Mock(side_effect=[
+                    mock_open(read_data='cert')(), mock_open(read_data='a')(),
+                    mock_open()(), mock_open(read_data='b')(), mock_open(read_data='c')()])):
+            k8s_config.load_incluster_config(token_refresh_interval=datetime.timedelta(milliseconds=100))
+            self.assertEqual(k8s_config.headers.get('authorization'), 'Bearer a')
+            time.sleep(0.1)
+            # token file doesn't exist
+            self.assertEqual(k8s_config.headers.get('authorization'), 'Bearer a')
+            # token file is empty
+            self.assertEqual(k8s_config.headers.get('authorization'), 'Bearer a')
+            # token refreshed
+            self.assertEqual(k8s_config.headers.get('authorization'), 'Bearer b')
+            time.sleep(0.1)
+            # token refreshed
+            self.assertEqual(k8s_config.headers.get('authorization'), 'Bearer c')
+            # no need to refresh token
+            self.assertEqual(k8s_config.headers.get('authorization'), 'Bearer c')

    def test_load_kube_config(self):
        config = {
@@ -212,7 +235,9 @@ class TestKubernetesConfigMaps(BaseTestKubernetes):
            self.k.manual_failover('foo', 'bar')

    def test_set_config_value(self):
-        self.k.set_config_value('{}')
+        with patch.object(k8s_client.CoreV1Api, 'patch_namespaced_config_map',
+                          Mock(side_effect=k8s_client.rest.ApiException(409, '')), create=True):
+            self.k.set_config_value('{}', 1)

    @patch.object(k8s_client.CoreV1Api, 'patch_namespaced_pod', create=True)
    def test_touch_member(self, mock_patch_namespaced_pod):
@@ -322,3 +347,18 @@ class TestCacheBuilder(BaseTestKubernetes):
    def test__list(self):
        self.k._pods._func = Mock(side_effect=Exception)
        self.assertRaises(Exception, self.k._pods._list)
+
+    @patch('patroni.dcs.kubernetes.ObjectCache._watch', Mock(return_value=None))
+    def test__do_watch(self):
+        self.assertRaises(AttributeError, self.k._kinds._do_watch, '1')
+
+    @patch.object(k8s_client.CoreV1Api, 'list_namespaced_config_map', mock_list_namespaced_config_map, create=True)
+    @patch('patroni.dcs.kubernetes.ObjectCache._watch')
+    def test_kill_stream(self, mock_watch):
+        self.k._kinds.kill_stream()
+        mock_watch.return_value.read_chunked.return_value = []
+        mock_watch.return_value.connection.sock.close.side_effect = Exception
+        self.k._kinds._do_watch('1')
+        self.k._kinds.kill_stream()
+        type(mock_watch.return_value).connection = PropertyMock(side_effect=Exception)
+        self.k._kinds.kill_stream()
@@ -13,7 +13,8 @@ from patroni.dcs.etcd import AbstractEtcdClientWithFailover
 from patroni.exceptions import DCSError
 from patroni.postgresql import Postgresql
 from patroni.postgresql.config import ConfigHandler
-from patroni import Patroni, main as _main, patroni_main, check_psycopg
+from patroni import check_psycopg
+from patroni.__main__ import Patroni, main as _main, patroni_main
 from six.moves import BaseHTTPServer, builtins
 from threading import Thread

@@ -97,7 +98,7 @@ class TestPatroni(unittest.TestCase):

    @patch('os.getpid')
    @patch('multiprocessing.Process')
-    @patch('patroni.patroni_main', Mock())
+    @patch('patroni.__main__.patroni_main', Mock())
    def test_patroni_main(self, mock_process, mock_getpid):
        mock_getpid.return_value = 2
        _main()
@@ -95,7 +95,7 @@ class TestPostgresql(BaseTestPostgresql):
    @patch('subprocess.call', Mock(return_value=0))
    @patch('os.rename', Mock())
    @patch('patroni.postgresql.CallbackExecutor', Mock())
-    @patch.object(Postgresql, 'get_major_version', Mock(return_value=130000))
+    @patch.object(Postgresql, 'get_major_version', Mock(return_value=140000))
    @patch.object(Postgresql, 'is_running', Mock(return_value=True))
    def setUp(self):
        super(TestPostgresql, self).setUp()
@@ -260,8 +260,8 @@ class TestPostgresql(BaseTestPostgresql):
        with patch('patroni.postgresql.config.ConfigHandler.primary_conninfo_params', Mock(return_value=conninfo)):
            mock_get_pg_settings.return_value['recovery_min_apply_delay'][1] = '1'
            self.assertEqual(self.p.config.check_recovery_conf(None), (True, True))
-            mock_get_pg_settings.return_value['primary_conninfo'][1] = 'host=1 passfile='\
-                + re.sub(r'([\'\\ ])', r'\\\1', self.p.config._pgpass)
+            mock_get_pg_settings.return_value['primary_conninfo'][1] = 'host=1 target_session_attrs=read-write'\
+                + ' passfile=' + re.sub(r'([\'\\ ])', r'\\\1', self.p.config._pgpass)
            mock_get_pg_settings.return_value['recovery_min_apply_delay'][1] = '0'
            self.assertEqual(self.p.config.check_recovery_conf(None), (True, True))
            self.p.config.write_recovery_conf({'standby_mode': 'on', 'primary_conninfo': conninfo.copy()})
@@ -287,6 +287,8 @@ class TestPostgresql(BaseTestPostgresql):
        mock_get_pg_settings.side_effect = Exception
        with patch('patroni.postgresql.config.mtime', mock_mtime):
            self.assertEqual(self.p.config.check_recovery_conf(None), (True, True))
+        with patch.object(Postgresql, 'is_starting', Mock(return_value=True)):
+            self.assertEqual(self.p.config.check_recovery_conf(None), (False, False))

    @patch.object(Postgresql, 'major_version', PropertyMock(return_value=100000))
    @patch.object(Postgresql, 'primary_conninfo', Mock(return_value='host=1'))
@@ -73,7 +73,7 @@ class TestPostmasterProcess(unittest.TestCase):

        # all processes successfully stopped
        mock_children.return_value = [Mock()]
-        mock_children.return_value[0].kill.side_effect = psutil.Error
+        mock_children.return_value[0].kill.side_effect = psutil.NoSuchProcess(123)
        self.assertTrue(proc.signal_kill())

        # postmaster has gone before suspend
@@ -81,17 +81,17 @@ class TestPostmasterProcess(unittest.TestCase):
        self.assertTrue(proc.signal_kill())

        # postmaster has gone before we got a list of children
-        mock_suspend.side_effect = psutil.Error()
+        mock_suspend.side_effect = psutil.AccessDenied()
        mock_children.side_effect = psutil.NoSuchProcess(123)
        self.assertTrue(proc.signal_kill())

        # postmaster has gone after we got a list of children
-        mock_children.side_effect = psutil.Error()
+        mock_children.side_effect = psutil.AccessDenied()
        mock_kill.side_effect = psutil.NoSuchProcess(123)
        self.assertTrue(proc.signal_kill())

        # failed to kill postmaster
-        mock_kill.side_effect = psutil.AccessDenied(123)
+        mock_kill.side_effect = psutil.AccessDenied()
        self.assertFalse(proc.signal_kill())

    @patch('psutil.Process.__init__', Mock())
@@ -157,6 +157,6 @@ class TestRaft(unittest.TestCase):
    @patch('threading.Event')
    def test_init(self, mock_event, mock_kvstore):
        mock_kvstore.return_value.applied_local_log = False
-        mock_event.return_value.isSet.side_effect = [False, True]
+        mock_event.return_value.is_set.side_effect = [False, True]
        self.assertIsNotNone(Raft({'ttl': 30, 'scope': 'test', 'name': 'pg', 'patronictl': True,
                                   'self_addr': '1', 'data_dir': self._TMP}))
@@ -66,7 +66,7 @@ class TestRewind(BaseTestPostgresql):

    def test_pg_rewind(self):
        r = {'user': '', 'host': '', 'port': '', 'database': '', 'password': ''}
-        with patch.object(Postgresql, 'major_version', PropertyMock(return_value=130000)),\
+        with patch.object(Postgresql, 'major_version', PropertyMock(return_value=150000)),\
                patch.object(CancellableSubprocess, 'call', Mock(return_value=None)):
            with patch('subprocess.check_output', Mock(return_value=b'boo')):
                self.assertFalse(self.r.pg_rewind(r))
@@ -102,6 +102,11 @@ class TestRewind(BaseTestPostgresql):
    @patch.object(Postgresql, 'start', Mock())
    def test_execute(self, mock_checkpoint):
        self.r.execute(self.leader)
+        with patch.object(Postgresql, 'major_version', PropertyMock(return_value=130000)):
+            self.r.execute(self.leader)
+            with patch.object(MockCursor, 'fetchone', Mock(side_effect=Exception)):
+                self.r.execute(self.leader)
+
        with patch.object(Rewind, 'pg_rewind', Mock(return_value=False)):
            mock_checkpoint.side_effect = ['1', '', '', '']
            self.r.execute(self.leader)
@@ -141,7 +146,10 @@ class TestRewind(BaseTestPostgresql):
        self.leader = self.leader.member
        self.assertFalse(self.r.rewind_or_reinitialize_needed_and_possible(self.leader))
        mock_check_leader_is_not_in_recovery.return_value = True
-        self.assertTrue(self.r.rewind_or_reinitialize_needed_and_possible(self.leader))
+        self.assertFalse(self.r.rewind_or_reinitialize_needed_and_possible(self.leader))
+        self.r.trigger_check_diverged_lsn()
+        with patch.object(MockCursor, 'fetchone', Mock(side_effect=[('', 3, '0/0'), ('', b'4\t0/40159C0\tn\n')])):
+            self.assertTrue(self.r.rewind_or_reinitialize_needed_and_possible(self.leader))
        self.r.reset_state()
        self.r.trigger_check_diverged_lsn()
        with patch('patroni.psycopg.connect', Mock(side_effect=Exception)):
@@ -27,6 +27,9 @@ class TestSlotsHandler(BaseTestPostgresql):
        super(TestSlotsHandler, self).setUp()
        self.s = self.p.slots_handler
        self.p.start()
+        config = ClusterConfig(1, {'slots': {'ls': {'database': 'a', 'plugin': 'b'}}}, 1)
+        self.cluster = Cluster(True, config, self.leader, 0,
+                               [self.me, self.other, self.leadermem], None, None, None, {'ls': 12345})

    def test_sync_replication_slots(self):
        config = ClusterConfig(1, {'slots': {'test_3': {'database': 'a', 'plugin': 'b'},
@@ -81,39 +84,44 @@ class TestSlotsHandler(BaseTestPostgresql):
    @patch.object(Postgresql, 'is_leader', Mock(return_value=False))
    def test__ensure_logical_slots_replica(self):
        self.p.set_role('replica')
-        config = ClusterConfig(1, {'slots': {'ls': {'database': 'a', 'plugin': 'b'}}}, 1)
-        cluster = Cluster(True, config, self.leader, 0,
-                          [self.me, self.other, self.leadermem], None, None, None, {'ls': 12346})
-        self.assertEqual(self.s.sync_replication_slots(cluster, False), [])
+        self.cluster.slots['ls'] = 12346
+        with patch.object(SlotsHandler, 'check_logical_slots_readiness', Mock()):
+            self.assertEqual(self.s.sync_replication_slots(self.cluster, False), [])
        self.s._schedule_load_slots = False
-        with patch.object(MockCursor, 'execute', Mock(side_effect=psycopg.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, 'execute', Mock(side_effect=psycopg.OperationalError)),\
+                patch.object(psycopg.OperationalError, 'diag') as mock_diag:
+            type(mock_diag).sqlstate = PropertyMock(return_value='58P01')
+            self.assertEqual(self.s.sync_replication_slots(self.cluster, False), ['ls'])
+        self.cluster.slots['ls'] = 'a'
+        self.assertEqual(self.s.sync_replication_slots(self.cluster, False), [])
        with patch.object(MockCursor, 'rowcount', PropertyMock(return_value=1), create=True):
-            self.assertEqual(self.s.sync_replication_slots(cluster, False), ['ls'])
+            self.assertEqual(self.s.sync_replication_slots(self.cluster, False), ['ls'])

-    @patch.object(MockCursor, 'execute', Mock(side_effect=psycopg.OperationalError))
    def test_copy_logical_slots(self):
-        self.s.copy_logical_slots(self.leader, ['foo'])
+        self.cluster.config.data['slots']['ls']['database'] = 'b'
+        self.s.copy_logical_slots(self.cluster, ['ls'])
+        with patch.object(MockCursor, 'execute', Mock(side_effect=psycopg.OperationalError)):
+            self.s.copy_logical_slots(self.cluster, ['foo'])

    @patch.object(Postgresql, 'stop', Mock(return_value=True))
    @patch.object(Postgresql, 'start', Mock(return_value=True))
    @patch.object(Postgresql, 'is_leader', Mock(return_value=False))
    def test_check_logical_slots_readiness(self):
-        self.s.copy_logical_slots(self.leader, ['ls'])
-        config = ClusterConfig(1, {'slots': {'ls': {'database': 'a', 'plugin': 'b'}}}, 1)
-        cluster = Cluster(True, config, self.leader, 0,
-                          [self.me, self.other, self.leadermem], None, None, None, {'ls': 12345})
-        self.assertEqual(self.s.sync_replication_slots(cluster, False), [])
-        with patch.object(MockCursor, 'rowcount', PropertyMock(return_value=1), create=True):
-            self.s.check_logical_slots_readiness(cluster, False, None)
+        self.s.copy_logical_slots(self.cluster, ['ls'])
+        with patch.object(MockCursor, '__iter__', Mock(return_value=iter([('postgresql0', None)]))),\
+                patch.object(MockCursor, 'fetchone', Mock(side_effect=Exception)):
+            self.assertIsNone(self.s.check_logical_slots_readiness(self.cluster, False, None))
+        with patch.object(MockCursor, '__iter__', Mock(return_value=iter([('postgresql0', None)]))),\
+                patch.object(MockCursor, 'fetchone', Mock(return_value=(False,))):
+            self.assertIsNone(self.s.check_logical_slots_readiness(self.cluster, False, None))
+        with patch.object(MockCursor, '__iter__', Mock(return_value=iter([('ls', 100)]))):
+            self.s.check_logical_slots_readiness(self.cluster, False, None)

    @patch.object(Postgresql, 'stop', Mock(return_value=True))
    @patch.object(Postgresql, 'start', Mock(return_value=True))
    @patch.object(Postgresql, 'is_leader', Mock(return_value=False))
    def test_on_promote(self):
-        self.s.copy_logical_slots(self.leader, ['ls'])
+        self.s.copy_logical_slots(self.cluster, ['ls'])
        self.s.on_promote()

    @unittest.skipIf(os.name == 'nt', "Windows not supported")
@@ -245,7 +245,7 @@ class TestZooKeeper(unittest.TestCase):

    def test_watch(self):
        self.zk.watch(None, 0)
-        self.zk.event.isSet = Mock(return_value=True)
+        self.zk.event.is_set = Mock(return_value=True)
        self.zk._fetch_status = False
        self.zk.watch(None, 0)
Author	SHA1	Message	Date
Alexander KukushkinandGitHub	fb06af9adb	Release 2.1.4 (#2322 ) - bump version - update release notes - implement missing unit-tests	2022-06-01 16:00:56 +02:00
Dennis4bandGitHub	b42550aad4	Add /read-only-sync endpoint (#2305 ) (#2311 ) `/read-only-sync` mirrors `/read-only`, but only returns `200` on a replica if this replica is a synchronous standby.	2022-05-30 17:09:43 +02:00
Alexander KukushkinandGitHub	f67174d7cc	Use replication credential in divergance check only with v10 and older (#2308 ) and document in which case pg_hba.conf should allow access to "postgres" database with replication credentials. Close #2261	2022-05-20 10:24:49 +02:00
Alexander KukushkinandGitHub	4c14b302f6	Remove [] characters from IPv6 hosts when splitting to host and port (#2309 ) Close #2269	2022-05-20 10:24:25 +02:00
Alexander KukushkinandGitHub	729f1dddc8	Compatibility with PostgreSQL 15 beta1 (#2299 ) * update postgresql/validator.py * pg_rewind doesn't like if there are unix sockets in PGDATA * pg_rewind now supports --config-file option	2022-05-19 15:36:09 +02:00
Alexander KukushkinandGitHub	ad3d953410	K8s: reset watchers if PATCH fails with 409 (#2283 ) High CPU load on Etcd nodes and K8s API servers created a very strange situation. A few clusters were running without a leader and the pod which is ahead of others was failing to take a leader lock because updates were failing with HTTP response code `409` (`resource_version` mismatch). Effectively that means that TCP connections to K8s master nodes were alive (in the opposite case tcp keepalives would have resolved it), but no `UPDATE` events were arriving via these connections, resulting in the stale cache of the cluster in memory. The only good way to prevent this situation is to intercept 409 HTTP responses and terminate existing TCP connections used for watches. Now a few words about implementation. Unfortunately, watch threads are waiting in the read() call most of the time and there is no good way to interrupt them. But, the `socket.shutdown()` seems to do this job. We already used this trick in the Etcd3 implementation. This approach will help to mitigate the issue of not having a leader, but at the same time replicas might still end up with the stale cluster state cached and in the worst case will not stream from the leader. Non-streaming replicas are less dangerous and could be covered by monitoring and partially mitigated by correctly configured `archive_command` and `restore_command`.	2022-05-19 15:24:20 +02:00
Alexander KukushkinandGitHub	ef3401c17f	Don't reset slots annotation if postgres isn't ready (#2306 ) The current state of permanent logical replication slots on the primary is queried together with `pg_current_wal_lsn()` and hence they "fail" simultaneously if Postgres isn't yet ready for accepting connections and in this case we want to avoid updating the `/status` key altogether. On K8s we don't use a dedicated object for the `/status` key, but use the same object (Endpoint or ConfigMap) as for the leader. If the `last_lsn` isn't set we avoid patching the corresponding annotation, but, the `slots` annotation was reset due to the oversight.	2022-05-19 15:06:59 +02:00
Alexander KukushkinandGitHub	496d14e6ca	Better handling of failed pg_rewind attempt (#2304 ) Close #2302	2022-05-19 14:52:26 +02:00
Alexander KukushkinandGitHub	6e8b2ce0a4	Don't try to run crash recovery if postgres is running (#2298 ) It was a minor oversight of #2252 and didn't get to a release	2022-05-19 13:46:49 +02:00
Alexander KukushkinandGitHub	96b75fa7cb	Special handling of check_recovery_conf for v12+ (#2292 ) When starting as a replica it may take some time before Postgres starts accepting new connections, but meanwhile, it could happen that the leader transitioned to a different member and the `primary_conninfo` must be updated. On pre v12 Patroni regularly checks `recovery.conf` in order to check that recovery parameters match the expectation. Starting from v12 recovery parameters were converted to GUC's and Patroni gets current values from the `pg_settings` view. The last one creates a problem when it takes more than a minute for Postgres to start accepting new connections. Since Patroni attempts to execute at least `pg_is_in_recovery()` every HA loop, and it is raising at exception, the `check_recovery_conf()` effectively wasn't reachable until recovery is finished, but it changed when #2082 was introduced. As a result of #2082 we got the following behavior: 1. Up to v12 (not including) everything was working as expected 2. v12 and v13 - Patroni restarting Postgres after 1m of recovery 3. v14+ - the `check_recovery_conf()` is not executed because the `replay_paused()` method raising an exception. In order to properly handle changes of recovery parameters or leader transitioned to a different node on v12+, we will rely on the cached values of recovery parameters until Postgres becomes ready to execute queries. Close https://github.com/zalando/patroni/issues/2289	2022-05-12 07:45:49 +02:00
Alexander KukushkinandGitHub	b901e62ad0	Enhanced checks of replica logical slots safety (#2285 ) The logical slot on a replica is safe to use when the physical replica slot on the primary: 1. has a nonzero/non-null `catalog_xmin` 2. has a `catalog_xmin` that is not newer (greater) than the `catalog_xmin` of any slot on the standby 3. the `catalog_xmin` is known to overtake `catalog_xmin` of logical slots on the primary observed during `1` In case if `1` doesn't take place, Patroni will run an additional check whether the `hot_standby_feedback` is actually in effect and shows the warning in case it is not.	2022-05-10 12:24:47 +02:00
haslersnandGitHub	7a2d6dc3c0	Ensure that optime annotation is a string (#2291 ) Fixes #2290	2022-05-10 09:20:39 +02:00
Haitao LiandGitHub	aa0cd48060	k8s: Support refreshing service account tokens (#2287 ) Since Kubernetes v1.21, with projected service account token feature, service account tokens expire in 1 hour. Kubernetes clients are expected to reread the token file to refresh the token. This patch re-reads the token file very minute for in-cluster config. Fixes #2286 Signed-off-by: Haitao Li <[email protected]>	2022-05-05 17:35:06 +02:00
Alexander KukushkinandGitHub	5f6197aaad	Don't copy logical slot if there is mismatch with the config (#2274 ) A couple of times we have seen in the wild that the database for the permanent logical slots was changed in the Patroni config. It resulted in the below situation. On the primary: 1. The slot must be dropped before creating it in a different DB. 2. Patroni fails to drop it because the slot is in use. Replica: 1. Patroni notice that the slot exists in the wrong DB and successfully dropping it. 2. Patroni copying the existing slot from the primary by its name with Postgres restart. And the loop repeats while the "wrong" slot exists on the primary. Basically, replicas are continuously restarting, which badly affects availability. In order to solve the problem, we will perform additional checks while copying replication slot files from the primary and discard them if `slot_type`, `database`, or `plugin` don't match our expectations.	2022-04-14 12:10:37 +02:00
Alexander KukushkinandGitHub	aea0589404	Switch to boto3 (#2275 ) Close https://github.com/zalando/patroni/issues/2237	2022-04-14 10:47:16 +02:00
zejeanmiandGitHub	40b5db4b85	Add ppc64le and fix inverted IOC_READ/WRITE vars (#2271 ) Close #2265	2022-04-14 10:46:01 +02:00
Wesley MendesandGitHub	e491edd1bf	Fixes missing import of dateutil.parser (#2259 ) Close #2258	2022-04-14 10:45:00 +02:00
James StroudandGitHub	0057f9018b	Spell out DCS (#2228 )	2022-03-24 13:58:10 +01:00
gremboandGitHub	c4e208ec50	Allow setting TLSServerName on consul service checks (#2231 ) See also https://www.consul.io/api-docs/agent/check#tlsservername Useful in case checks are done by IP and the consul `node_name` is not an FQDN.	2022-03-24 13:57:17 +01:00
Gunnar "Nick" BluthandGitHub	7626b5fef8	Fix pg_rewind on typical Debian/Ubuntu systems (#2225 ) On Debian/Ubuntu systems it is common to keep Postgres config files outside of the data directory. It created a couple of problems for pg_rewind support in Patroni. 1. The `--config_file` argument must be supplied while figuring out the `restore_command` GUC value on Postgres v12+ 2. With Postgres v13+ pg_rewind by itself can't find postgresql.conf in order to figure out `restore_command` and therefore we have to use Patroni as a fallback for fetching missing WAL's that are required for rewind. This commit addresses both problems.	2022-03-24 13:56:16 +01:00
Alexander KukushkinandGitHub	81912c9cae	Handle rewind when demoted node was shut down (#2252 ) In case of DCS unavailability Patroni restarts Postgres in read-only. It will cause pg_control to be updated with the `Database cluster state: in archive recovery` and also could set the `MinRecoveryPoint`. When the next time Patroni is started it will assume that Postgres was running as a replica and rewind isn't required and will try to start the Postgres up. In this situation there is the chance that the start will be aborted with the FATAL error message that looks like `requested timeline 2 does not contain minimum recovery point 0/501E8B8 on timeline 1`. On the next heart-beat Patroni will again notice that Postgres isn't running which would lead to another start-fail attempt. This loop is endless. In order to mitigate the problem we do the following: 1. While figuring out whether the rewind is required we consider `in archive recovery` along with `shut down in recovery`. 2. If pg_rewind is required and the cluster state is `in archive recovery` we also perform recovery in a single-user mode. Close https://github.com/zalando/patroni/issues/2242	2022-03-24 13:51:59 +01:00
Alexander KukushkinandGitHub	333d41d9f0	Release 2.1.3 (#2219 ) * Implement missing unit-tests * Bump version * Update release notes	2022-02-18 14:16:15 +01:00
Alexander KukushkinandGitHub	aa91557a80	Fix bug in divergence timeline check (#2221 ) Patroni was falsely assuming that timelines have diverged. For pg_rewind it didn't create any problem, but if pg_rewind is not allowed and the `remove_data_directory_on_diverged_timelines` is set, it resulted in reinitializing the former leader. Close https://github.com/zalando/patroni/issues/2220	2022-02-17 15:53:13 +01:00
Hrvoje MilkovićandGitHub	075918d447	Fixed AttributeError no attribute 'leader' (#2217 ) Close https://github.com/zalando/patroni/issues/2218	2022-02-16 10:20:15 +01:00
Michael BanckandGitHub	c4535ae208	Avoid running CHECKPOINT on remote master if credentials are missing (#2195 ) Close #2194	2022-02-14 15:21:51 +01:00
Bastien WirtzandGitHub	38d84b1d15	Make sure no substitution attemps is made when params is empty. (#2212 ) Close #2209	2022-02-14 15:20:38 +01:00
Michael BanckandGitHub	2d15e0dae6	Add target_session_attrs=read-write to standby_leader primary_conninfo (#2193 ) This allows to have multiple hosts in a standby_cluster and ensures that the standby leader follows the main cluster's new leader after a switchover. Partially addresses #2189	2022-02-10 15:50:14 +01:00
Michael BanckandGitHub	48d8c13e6b	Write pgpass line per host if more than one is specified in connstr (#2192 ) Partly addresses #2189	2022-02-10 15:40:24 +01:00
Alexander KukushkinandGitHub	d3e3b4e16f	Minor tuning of tests (#2201 ) - Reduce verbosity for unit tests - Refactor GH actions config and try again macos behave tests	2022-02-10 15:38:16 +01:00
Alexandre PereiraandGitHub	afab392ead	Add metrics (#2199 ) This PR adds metrics for additional information : - If a node or cluster is pending restart, - If the cluster management is paused. This may be useful for Prometheus/Grafana monitoring. Close #2198	2022-02-10 15:37:14 +01:00
Alexander KukushkinandGitHub	291754eeb0	Don't remove the leader lock while paused (#2187 ) Close https://github.com/zalando/patroni/issues/2179	2022-02-10 15:36:25 +01:00
Alexander KukushkinandGitHub	cdc80a1d89	Restart etcd3 watcher if all etcd nodes don't respond (#2186 ) Close https://github.com/zalando/patroni/issues/2180	2022-02-10 15:32:29 +01:00
Alexander KukushkinandGitHub	04c6f58b2b	Make Kubernetes.cancel_initialization() method similar to other DCS (#2210 ) I.e., do delete unconditionally and return the success	2022-02-10 15:29:29 +01:00
Ants AasmaandGitHub	0980838cb3	Fix port in use error on certificate replacement (#2185 ) When switching certificates there is a race condition with a concurrent API request. If there is one active during the replacement period then the replacement will error out with a port in use error and Patroni gets stuck in a state without an active API server. Fix is to call server_close after shutdown which will wait for already running requests to complete before returning. Close #2184	2022-01-26 13:52:25 +01:00
Alexander KukushkinandGitHub	3e1076a574	Use replication credentials when checking leader status (#2165 ) It could be that `remove_data_directory_on_diverged_timelines` is set, but there is no `rewind_credentials` defined and superuser access between nodes is not allowed. Close https://github.com/zalando/patroni/issues/2162	2022-01-11 16:23:13 +01:00
Alexander KukushkinandGitHub	cb3071adfb	Annual cleanup (#2159 ) - Simplify setup.py: remove unneeded features and get rid of deprecation warnings - Compatibility with Python 3.10: handle `threading.Event.isSet()` deprecation - Make sure setup.py could run without `six`: move Patroni class and main function to the `__main__.py`. The `__init__.py` will have only a few functions used by the Patroni class and from the setup.py	2022-01-06 10:20:31 +01:00
Alexander KukushkinandGitHub	bf354aeebd	Compatibility with legacy psycopg2 (#2158 ) For example, psycopg2 installed from Ubuntu 18.04 packages doesn't have `UndefinedFile` exception yet.	2022-01-06 10:14:50 +01:00
Alexander KukushkinandGitHub	01d40a4a13	Compatibility with latest psutil and setuptools (#2155 ) Issues don't affect Patroni code, only unit-tests	2022-01-05 09:53:33 +01:00
Alexander KukushkinandGitHub	3cc14cc059	Unquote integers in validator (#2154 ) Close https://github.com/zalando/patroni/issues/2150	2022-01-04 10:47:02 +01:00
Alexander KukushkinandGitHub	a015e0e271	Fix bug with failover to cascading standby (#2138 ) When figuring out which slots should be created on cascading standby we forgot to take into account that the leader might be absent. Close: https://github.com/zalando/patroni/issues/2137	2021-12-21 11:20:35 +01:00
Alexander KukushkinandGitHub	d2b681b07e	Fix bug in the bootstrap standby-leader (#2144 ) When starting postgres after bootstrap of the standby-leader the `follow()` method is used to always return `True`. This behavior was changed in the #2054 in order to avoid hammering logs if postgres is failing to start. Since now the method returns `None` if postgres didn't start accepting connections after 60s, the change broke the standby-leader bootstrap code. As the solution, we will assume that the clone was successful if the `follow()` method returned anything different from `False`.	2021-12-21 11:20:06 +01:00
Alexander KukushkinandGitHub	63586f0477	Add ctl.keyfile_password support (#2145 ) It compliments restapi.keyfile_password added in the #1825	2021-12-21 11:19:39 +01:00
Alexander KukushkinandGitHub	4215565cb4	Rearrange tests (#2146 ) - remove codacy steps: they removed legacy organizations and there seems to be no easy way of installing codacy app to the Zalando GH. - Don't run behave on MacOS: recently worker became way to slow - Disable behave for combination of kubernetes and python 2.7 - Remove python 3.5 (it will be removed by GH from workers in January) and add 3.10 - Run behave with 3.6 and 3.9 instead of 3.5 and 3.8	2021-12-21 09:36:22 +01:00