arm64 compatibility

Close #2331

.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
+        python-version: '3.10'
-    - 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
     - 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

Dockerfile

@@ -1,6 +1,6 @@
 ## This Dockerfile is meant to aid in the building and debugging patroni whilst developing on your local machine
 ## It has all the necessary components to play/debug with a single node appliance, running etcd
-ARG PG_MAJOR=10
+ARG PG_MAJOR=14
 ARG COMPRESS=false
 ARG PGHOME=/home/postgres
 ARG PGDATA=$PGHOME/data
@@ -14,7 +14,7 @@ ARG PGDATA
 ARG LC_ALL
 ARG LANG
 
-ENV ETCDVERSION=3.3.13 CONFDVERSION=0.16.0
+ENV ETCDVERSION=3.5.4 CONFDVERSION=0.16.0
 
 RUN set -ex \
     && export DEBIAN_FRONTEND=noninteractive \
@@ -50,11 +50,11 @@ RUN set -ex \
     && chown -R postgres:postgres /var/log \
 \
     # Download etcd
-    && curl -sL https://github.com/coreos/etcd/releases/download/v${ETCDVERSION}/etcd-v${ETCDVERSION}-linux-amd64.tar.gz \
+    && curl -L https://github.com/coreos/etcd/releases/download/v${ETCDVERSION}/etcd-v${ETCDVERSION}-linux-$(dpkg --print-architecture).tar.gz \
             | tar xz -C /usr/local/bin --strip=1 --wildcards --no-anchored etcd etcdctl \
 \
     # Download confd
-    && curl -sL https://github.com/kelseyhightower/confd/releases/download/v${CONFDVERSION}/confd-${CONFDVERSION}-linux-amd64 \
+    && curl -sL https://github.com/kelseyhightower/confd/releases/download/v${CONFDVERSION}/confd-${CONFDVERSION}-linux-$(dpkg --print-architecture) \
             > /usr/local/bin/confd && chmod +x /usr/local/bin/confd \
 \
     # Clean up all useless packages and some files
@@ -89,30 +89,7 @@ RUN set -ex \
 #        /var/lib/dpkg/info/* \
     && find /usr/bin -xtype l -delete \
     && find /var/log -type f -exec truncate --size 0 {} \; \
-    && find /usr/lib/python3/dist-packages -name '*test*' | xargs rm -fr \
+    && find /usr/lib/python3/dist-packages -name '*test*' | xargs rm -fr
-    && find /lib/x86_64-linux-gnu/security -type f ! -name pam_env.so ! -name pam_permit.so ! -name pam_unix.so -delete
-
-# perform compression if it is necessary
-ARG COMPRESS
-RUN if [ "$COMPRESS" = "true" ]; then \
-        set -ex \
-        # Allow certain sudo commands from postgres
-        && echo 'postgres ALL=(ALL) NOPASSWD: /bin/tar xpJf /a.tar.xz -C /, /bin/rm /a.tar.xz, /bin/ln -snf dash /bin/sh' >> /etc/sudoers \
-        && ln -snf busybox /bin/sh \
-        && files="/bin/sh /usr/bin/sudo /usr/lib/sudo/sudoers.so /lib/x86_64-linux-gnu/security/pam_*.so" \
-        && libs="$(ldd $files | awk '{print $3;}' | grep '^/' | sort -u) /lib/x86_64-linux-gnu/ld-linux-x86-64.so.* /lib/x86_64-linux-gnu/libnsl.so.* /lib/x86_64-linux-gnu/libnss_compat.so.*" \
-        && (echo /var/run $files $libs | tr ' ' '\n' && realpath $files $libs) | sort -u | sed 's/^\///' > /exclude \
-        && find /etc/alternatives -xtype l -delete \
-        && save_dirs="usr lib var bin sbin etc/ssl etc/init.d etc/alternatives etc/apt" \
-        && XZ_OPT=-e9v tar -X /exclude -cpJf a.tar.xz $save_dirs \
-        # we call "cat /exclude" to avoid including files from the $save_dirs that are also among
-        # the exceptions listed in the /exclude, as "uniq -u" eliminates all non-unique lines.
-        # By calling "cat /exclude" a second time we guarantee that there will be at least two lines
-        # for each exception and therefore they will be excluded from the output passed to 'rm'.
-        && /bin/busybox sh -c "(find $save_dirs -not -type d && cat /exclude /exclude && echo exclude) | sort | uniq -u | xargs /bin/busybox rm" \
-        && /bin/busybox --install -s \
-        && /bin/busybox sh -c "find $save_dirs -type d -depth -exec rmdir -p {} \; 2> /dev/null"; \
-    fi
 
 FROM scratch
 COPY --from=builder / /

docker-compose.yml

@@ -15,6 +15,7 @@ services:
             ETCD_INITIAL_CLUSTER: etcd1=http://etcd1:2380,etcd2=http://etcd2:2380,etcd3=http://etcd3:2380
             ETCD_INITIAL_CLUSTER_STATE: new
             ETCD_INITIAL_CLUSTER_TOKEN: tutorial
+            ETCD_UNSUPPORTED_ARCH: arm64
         container_name: demo-etcd1
         hostname: etcd1
         command: etcd -name etcd1 -initial-advertise-peer-urls http://etcd1:2380

docs/ENVIRONMENT.rst

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

docs/README.rst

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

docs/SETTINGS.rst

@@ -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:
 
@@ -223,7 +224,7 @@ Raft
 
   - Q: How to list all the nodes providing consensus?
 
-    A: ``syncobj_admin -conn host:port`` -status where the host:port is the address of one of the cluster nodes
+    A: ``syncobj_admin -conn host:port -status`` where the host:port is the address of one of the cluster nodes
 
   - Q: Node that was a part of consensus and has gone and I can't reuse the same IP for other node. How to remove this node from the consensus?
 
@@ -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
 --------

docs/releases.rst

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

docs/rest_api.rst

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

features/environment.py

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

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

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

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'])
@@ -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',
+                              'REGISTER_SERVICE', 'SERVICE_CHECK_INTERVAL', 'SERVICE_CHECK_TLS_SERVER_NAME',
-                              'USE_ENDPOINTS', 'SCOPE_LABEL', 'ROLE_LABEL', 'POD_IP', 'PORTS', 'LABELS',
+                              'NAMESPACE', 'CONTEXT', 'USE_ENDPOINTS', 'SCOPE_LABEL', 'ROLE_LABEL', 'POD_IP',
-                              'BYPASS_API_SERVICE', 'KEY_PASSWORD', 'USE_SSL', 'SET_ACLS') and name:
+                              '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)

patroni/dcs/__init__.py

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

patroni/dcs/consul.py

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

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,6 +395,7 @@ class AbstractEtcdClientWithFailover(etcd.Client):
         self._machines_cache_updated = time.time()
 
     def set_base_uri(self, value):
+        if self._base_uri != value:
             logger.info('Selected new etcd server %s', value)
             self._base_uri = value
 

patroni/dcs/etcd3.py

@@ -315,6 +315,7 @@ class Etcd3Client(AbstractEtcdClientWithFailover):
     def prefix(self, key, retry=None):
         return self.range(key, prefix_range_end(key), retry)
 
+    @_handle_auth_errors
     def lease_grant(self, ttl, retry=None):
         return self.call_rpc('/lease/grant', {'TTL': ttl}, retry)['ID']
 

patroni/dcs/kubernetes.py

@@ -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,13 +82,8 @@ class K8sConfig(object):
             if not f.read():
                 raise self.ConfigException('Cert file exists but empty.')
         self.pool_config['ca_certs'] = ca_certs
-
+        self._token_refresh_interval = token_refresh_interval
-        if not os.path.isfile(SERVICE_TOKEN_FILENAME):
+        token = self._read_token_file()
-            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._server = uri('https', (os.environ[SERVICE_HOST_ENV_NAME], os.environ[SERVICE_PORT_ENV_NAME]))
 
@@ -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,23 +560,9 @@ class ObjectCache(Thread):
         with self._object_cache_lock:
             return self._object_cache.get(name)
 
-    def _build_cache(self):
+    def _process_event(self, event):
-        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']
+        obj = event['object']
         name = obj['metadata']['name']
 
         if ev_type in ('ADDED', 'MODIFIED'):
@@ -571,10 +574,9 @@ class ObjectCache(Thread):
             success, old_value = self.delete(name, obj['metadata']['resourceVersion'])
             new_value = None
         else:
-                    logger.warning('Unexpected event type: %s', ev_type)
+            return logger.warning('Unexpected event type: %s', ev_type)
-                    continue
 
-                if success and return_type != 'V1Pod':
+        if success and obj.get('kind') != 'Pod':
             if old_value:
                 old_value = (old_value.metadata.annotations or {}).get(self._annotations_map.get(name))
 
@@ -587,19 +589,70 @@ class ObjectCache(Thread):
             # 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()
+        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()
+
+        try:
+            self._do_watch(objects.metadata.resource_version)
         finally:
             with self._condition:
                 self._is_ready = False
-            response.close()
+            with self._response_lock:
-            response.release_conn()
+                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
+        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
@@ -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):

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()
@@ -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 self._rewind.should_remove_data_directory_on_diverged_timelines and not self.is_standby_cluster():
-        if 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') \
+        if data.get('Database cluster state') in ('in production', 'shutting down', 'in crash recovery'):
-                and not self._crash_recovery_executed and \
+            msg = self._handle_crash_recovery()
-                (self.cluster.is_unlocked() or self._rewind.can_rewind):
+            if msg:
-            self._crash_recovery_executed = True
+                return msg
-            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
 
         self.load_cluster_from_dcs()
 
@@ -413,6 +423,7 @@ class Ha(object):
                 self.state_handler.get_history(self._leader_timeline + 1):
             self._rewind.trigger_check_diverged_lsn()
 
+        if not self.state_handler.is_starting():
             msg = self._handle_rewind_or_reinitialize()
             if msg:
                 return msg
@@ -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,6 +1023,7 @@ class Ha(object):
                 if self.cluster.failover and self.cluster.failover.candidate == self.state_handler.name:
                     return 'waiting to become master after promote...'
 
+                if not self.is_standby_cluster():
                     self._delete_leader()
                     return 'removed leader lock because postgres is not running as master'
 
@@ -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

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

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
@@ -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,16 +628,16 @@ 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()
+                    wanted_lines = self._pgpass_line(wanted_primary_conninfo).splitlines()
-                    for raw_line in f:
+                    file_lines = f.read().splitlines()
-                        if raw_line.strip() == wanted_line:
+                    if set(wanted_lines) == set(file_lines):
                         self._passfile = passfile
                         self._passfile_mtime = passfile_mtime
                         return True
@@ -643,6 +653,7 @@ class ConfigHandler(object):
         elif not primary_conninfo:
             return False
 
+        if not self._postgresql.is_starting():
             wal_receiver_primary_conninfo = self._postgresql.primary_conninfo()
             if wal_receiver_primary_conninfo:
                 wal_receiver_primary_conninfo = parse_dsn(wal_receiver_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,6 +708,7 @@ class ConfigHandler(object):
             else:  # empty string, primary_conninfo is not in the config
                 primary_conninfo[0] = {}
 
+        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():
@@ -706,7 +718,9 @@ class ConfigHandler(object):
 
         # 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

patroni/postgresql/rewind.py

@@ -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(
+        # We want to use replication credentials when connecting to the "postgres" database in case if
-                self._conn_kwargs(leader, self._postgresql.config.rewind_credentials)):
+        # `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
+        if checkpoint_status is None:  # we are the standby-cluster leader or master still runs the old Patroni
-            leader_status = self._postgresql.checkpoint(self._conn_kwargs(leader, self._postgresql.config.superuser))
+            # superuser credentials match rewind_credentials if the latter are not provided or we run 10 or older
-            if leader_status:
+            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):
+        else:
+            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
-            else:
-                self._state = REWIND_STATUS.FAILED
         return False
 
     def reset_state(self):

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__)
 
@@ -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()"
+                    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 = %s", (slot_name,))
+                                " WHERE NOT pg_catalog.pg_is_in_recovery() AND slot_name = ANY(%s)",
-                    if cur.rowcount < 1:
+                                ([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']:
+            # The logical slot on a replica is safe to use when the physical replica slot on the primary:
-                self._unready_logical_slots.remove(name)
+            # 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()"
+                            " FROM pg_catalog.pg_get_replication_slots() JOIN pg_catalog.pg_database ON datoid = oid"
-                            " AND slot_name = ANY(%s)", (slots,))
+                            " WHERE NOT pg_catalog.pg_is_in_recovery() AND slot_name = ANY(%s)", (create_slots,))
-                slots = {r[0]: {'catalog_xmin': r[1], 'confirmed_flush_lsn': r[2], 'data': r[3]} for r in cur}
+
+                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))

patroni/postgresql/validator.py

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

patroni/psycopg.py

@@ -1,5 +1,4 @@
-__all__ = ['connect', 'quote_ident', 'quote_literal', 'DatabaseError',
+__all__ = ['connect', 'quote_ident', 'quote_literal', 'DatabaseError', 'Error', 'OperationalError', 'ProgrammingError']
-           'Error', 'OperationalError', 'ProgrammingError', 'UndefinedFile']
 
 _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/scripts/aws.py

@@ -3,10 +3,12 @@
 import json
 import logging
 import sys
-import boto.ec2
+import boto3
 
-from patroni.utils import Retry, RetryFailedError
+from ..utils import Retry, RetryFailedError
-from patroni.request import get as requests_get
+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}
+        tags = [{'Key': 'Name', 'Value': 'spilo_' + self.cluster_name},
-        volumes = conn.get_all_volumes(filters={'attachment.instance-id': self.instance_id})
+                {'Key': 'Role', 'Value': role},
-        conn.create_tags([v.id for v in volumes], tags)
+                {'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}
+        tags = [{'Key': 'Role', 'Value': role}]
-        conn.create_tags([self.instance_id], tags)
+        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:

patroni/utils.py

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

patroni/version.py

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

patroni/watchdog/linux.py

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

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

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

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

requirements.txt

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

setup.py

@@ -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
+        pass
-
-        self.flake8 = application.Application()
-        self.flake8.initialize([])
 
     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())
+        from flake8.main import application
-        self.flake8.formatter.start()
+
-        self.flake8.report_errors()
+        logging.getLogger().setLevel(logging.ERROR)
-        self.flake8.report_statistics()
+        flake8 = application.Application()
-        self.flake8.report_benchmarks()
+        flake8.run(self.targets())
-        self.flake8.formatter.stop()
+        flake8.exit()
-        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
 
 
-class PyTest(Command):
+class PyTest(_Command):
 
-    user_options = [('cov=', None, 'Run coverage'), ('cov-xml=', None, 'Generate junit xml report'),
+    def run(self):
-                    ('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):
         try:
             import pytest
         except Exception:
             raise RuntimeError('py.test is not installed, run: pip install pytest')
 
-        import logging
+        logging.getLogger().setLevel(logging.WARNING)
-        silence = logging.WARNING
-        logging.basicConfig(format='%(asctime)s %(levelname)s: %(message)s', level=os.getenv('LOGLEVEL', silence))
 
         args = ['--verbose', 'tests', '--doctest-modules', MAIN_PACKAGE] +\
-            ['-s' if logging.getLogger().getEffectiveLevel() < silence else '--capture=fd']
+            ['-s' if logging.getLogger().getEffectiveLevel() < logging.WARNING else '--capture=fd'] +\
-        if self.cov:
+            ['--cov', MAIN_PACKAGE, '--cov-report', 'term-missing', '--cov-report', 'xml']
-            args += self.cov
 
         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

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

tests/test_api.py

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

tests/test_aws.py

@@ -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):
+    def create_tags(Resources, **kwargs):
-        if len(objects) == 0:
+        if len(Resources) == 0:
-            raise boto.exception.BotoServerError(503, 'Service Unavailable', 'Request limit exceeded')
+            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(
+    @patch('patroni.scripts.aws.requests_get', Mock(return_value=urllib3.HTTPResponse(status=503, body=b'Error')))
-        status=200, body=b'{"instanceId": "012345", "region": "eu-west-1"}')))
     @patch('sys.exit', Mock())
     def test_main(self):
         self.assertIsNone(_main())

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.children.side_effect = psutil.NoSuchProcess(123)
-        self.c._process.suspend.side_effect = psutil.Error()
+        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_consul.py

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

tests/test_ha.py

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

tests/test_kubernetes.py

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

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

@@ -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='\
+            mock_get_pg_settings.return_value['primary_conninfo'][1] = 'host=1 target_session_attrs=read-write'\
-                + re.sub(r'([\'\\ ])', r'\\\1', self.p.config._pgpass)
+                + ' 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'))

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

@@ -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,6 +146,9 @@ 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.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()

tests/test_slots.py

@@ -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)
+        self.cluster.slots['ls'] = 12346
-        cluster = Cluster(True, config, self.leader, 0,
+        with patch.object(SlotsHandler, 'check_logical_slots_readiness', Mock()):
-                          [self.me, self.other, self.leadermem], None, None, None, {'ls': 12346})
+            self.assertEqual(self.s.sync_replication_slots(self.cluster, False), [])
-        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)),\
-            self.assertEqual(self.s.sync_replication_slots(cluster, False), ['ls'])
+                patch.object(psycopg.OperationalError, 'diag') as mock_diag:
-        cluster.slots['ls'] = 'a'
+            type(mock_diag).sqlstate = PropertyMock(return_value='58P01')
-        self.assertEqual(self.s.sync_replication_slots(cluster, False), [])
+            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'])
+        self.s.copy_logical_slots(self.cluster, ['ls'])
-        config = ClusterConfig(1, {'slots': {'ls': {'database': 'a', 'plugin': 'b'}}}, 1)
+        with patch.object(MockCursor, '__iter__', Mock(return_value=iter([('postgresql0', None)]))),\
-        cluster = Cluster(True, config, self.leader, 0,
+                patch.object(MockCursor, 'fetchone', Mock(side_effect=Exception)):
-                          [self.me, self.other, self.leadermem], None, None, None, {'ls': 12345})
+            self.assertIsNone(self.s.check_logical_slots_readiness(self.cluster, False, None))
-        self.assertEqual(self.s.sync_replication_slots(cluster, False), [])
+        with patch.object(MockCursor, '__iter__', Mock(return_value=iter([('postgresql0', None)]))),\
-        with patch.object(MockCursor, 'rowcount', PropertyMock(return_value=1), create=True):
+                patch.object(MockCursor, 'fetchone', Mock(return_value=(False,))):
-            self.s.check_logical_slots_readiness(cluster, False, None)
+            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")

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)