Release 2.1.3 (#2219)

* Implement missing unit-tests
* Bump version
* Update release notes

.github/workflows/install_deps.py

@@ -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)

.github/workflows/run_tests.py

@@ -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'))

.github/workflows/tests.yaml

@@ -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

docs/SETTINGS.rst

@@ -329,7 +329,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 +367,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
 --------

docs/releases.rst

@@ -3,6 +3,72 @@
 Release notes
 =============
 
+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
 -------------
 

features/environment.py

@@ -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)

features/patroni_api.feature

@@ -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

features/steps/patroni_api.py

@@ -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:

features/steps/standby_cluster.py

@@ -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(

patroni.py

@@ -1,5 +1,5 @@
 #!/usr/bin/env python
-from patroni import main
+from patroni.__main__ import main
 
 
 if __name__ == '__main__':

patroni/__init__.py

@@ -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()

patroni/__main__.py

@@ -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__':

patroni/api.py

@@ -293,6 +293,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 +778,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

patroni/config.py

@@ -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'])

patroni/dcs/__init__.py

@@ -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()

patroni/dcs/etcd.py

@@ -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):

patroni/dcs/kubernetes.py

@@ -1050,7 +1050,7 @@ class Kubernetes(AbstractDCS):
             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):

patroni/dcs/raft.py

@@ -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')

patroni/ha.py

@@ -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()
@@ -796,7 +796,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 +1012,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()

patroni/postgresql/__init__.py

@@ -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:

patroni/postgresql/config.py

@@ -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
@@ -618,19 +625,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
@@ -745,7 +752,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)

patroni/postgresql/rewind.py

@@ -65,6 +65,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
@@ -154,6 +168,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):
@@ -165,7 +183,7 @@ class Rewind(object):
             return
 
         if not self.check_leader_is_not_in_recovery(
-                self._conn_kwargs(leader, self._postgresql.config.rewind_credentials)):
+                self._conn_kwargs(leader, self._postgresql.config.replication)):
             return
 
         history = need_rewind = None
@@ -200,6 +218,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
@@ -333,9 +352,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)

patroni/postgresql/slots.py

@@ -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__)
 
@@ -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

patroni/postgresql/validator.py

@@ -151,7 +151,7 @@ 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')),
@@ -234,10 +234,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),
@@ -344,7 +344,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'),
@@ -456,10 +456,10 @@ parameters = CaseInsensitiveDict({
     'vacuum_cost_page_hit': Integer(90300, None, 0, 10000, None),
     'vacuum_cost_page_miss': Integer(90300, None, 0, 10000, None),
     'vacuum_defer_cleanup_age': Integer(90300, None, 0, 1000000, None),
-    'vacuum_failsafe_age': Integer(140000, None, '0', '2100000000', None),
+    'vacuum_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'),

patroni/psycopg.py

@@ -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)

patroni/request.py

@@ -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)
 

patroni/version.py

@@ -1 +1 @@
-__version__ = '2.1.2'
+__version__ = '2.1.3'

postgres0.yml

@@ -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

postgres1.yml

@@ -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

postgres2.yml

@@ -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

setup.py

@@ -5,6 +5,7 @@
 """
 
 import inspect
+import logging
 import os
 import sys
 
@@ -26,7 +27,6 @@ EXTRAS_REQUIRE = {'aws': ['boto'], 'etcd': ['python-etcd'], 'etcd3': ['python-et
                   '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)

tests/__init__.py

@@ -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,

tests/test_api.py

@@ -549,7 +549,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')

tests/test_cancellable.py

@@ -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()

tests/test_patroni.py

@@ -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()

tests/test_postgresql.py

@@ -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()})

tests/test_postmaster.py

@@ -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())

tests/test_raft.py

@@ -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}))

tests/test_rewind.py

@@ -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)):

tests/test_slots.py

@@ -86,7 +86,9 @@ class TestSlotsHandler(BaseTestPostgresql):
                           [self.me, self.other, self.leadermem], None, None, None, {'ls': 12346})
         self.assertEqual(self.s.sync_replication_slots(cluster, False), [])
         self.s._schedule_load_slots = False
-        with patch.object(MockCursor, 'execute', Mock(side_effect=psycopg.UndefinedFile)):
+        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(cluster, False), ['ls'])
         cluster.slots['ls'] = 'a'
         self.assertEqual(self.s.sync_replication_slots(cluster, False), [])

tests/test_zookeeper.py

@@ -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)