arm64 compatibility

Close #2331

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 /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
+    && find /usr/lib/python3/dist-packages -name '*test*' | xargs rm -fr
 
 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.
 

docs/releases.rst

@@ -3,6 +3,78 @@
 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
 -------------
 

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"'

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:

patroni/config.py

@@ -350,9 +350,9 @@ class Config(object):
                 name, suffix = (param[8:].split('_', 1) + [''])[:2]
                 if suffix in ('HOST', 'HOSTS', 'PORT', 'USE_PROXIES', 'PROTOCOL', 'SRV', 'SRV_SUFFIX', 'URL', 'PROXY',
                               'CACERT', 'CERT', 'KEY', 'VERIFY', 'TOKEN', 'CHECKS', 'DC', 'CONSISTENCY',
-                              'REGISTER_SERVICE', 'SERVICE_CHECK_INTERVAL', 'NAMESPACE', 'CONTEXT',
-                              'USE_ENDPOINTS', 'SCOPE_LABEL', 'ROLE_LABEL', 'POD_IP', 'PORTS', 'LABELS',
-                              'BYPASS_API_SERVICE', 'KEY_PASSWORD', 'USE_SSL', 'SET_ACLS') and name:
+                              'REGISTER_SERVICE', 'SERVICE_CHECK_INTERVAL', 'SERVICE_CHECK_TLS_SERVER_NAME',
+                              'NAMESPACE', 'CONTEXT', 'USE_ENDPOINTS', 'SCOPE_LABEL', 'ROLE_LABEL', 'POD_IP',
+                              'PORTS', 'LABELS', 'BYPASS_API_SERVICE', 'KEY_PASSWORD', 'USE_SSL', 'SET_ACLS') and name:
                     value = os.environ.pop(param)
                     if suffix == 'PORT':
                         value = value and parse_int(value)

patroni/dcs/__init__.py

@@ -1,5 +1,5 @@
 import abc
-import dateutil
+import dateutil.parser
 import importlib
 import inspect
 import json

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/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,14 +82,9 @@ class K8sConfig(object):
             if not f.read():
                 raise self.ConfigException('Cert file exists but empty.')
         self.pool_config['ca_certs'] = ca_certs
-
-        if not os.path.isfile(SERVICE_TOKEN_FILENAME):
-            raise self.ConfigException('Service token file does not exists.')
-        with open(SERVICE_TOKEN_FILENAME) as f:
-            token = f.read()
-            if not token:
-                raise self.ConfigException('Token file exists but empty.')
-            self._make_headers(token=token)
+        self._token_refresh_interval = token_refresh_interval
+        token = self._read_token_file()
+        self._make_headers(token=token)
         self._server = uri('https', (os.environ[SERVICE_HOST_ENV_NAME], os.environ[SERVICE_PORT_ENV_NAME]))
 
     @staticmethod
@@ -109,6 +119,11 @@ class K8sConfig(object):
 
     @property
     def headers(self):
+        if self._token_expires_at <= datetime.datetime.now():
+            try:
+                self._set_token(self._read_token_file())
+            except Exception as e:
+                logger.error('Failed to refresh service account token: %r', e)
         return self._headers.copy()
 
 
@@ -503,6 +518,8 @@ class ObjectCache(Thread):
         self._condition = condition
         self._name = name  # name of this pod
         self._is_ready = False
+        self._response = None  # needs to be accessible from the `kill_stream()` method
+        self._response_lock = Lock()  # protect the `self._response` from concurrent access
         self._object_cache = {}
         self._object_cache_lock = Lock()
         self._annotations_map = {self._dcs.leader_path: self._dcs._LEADER, self._dcs.config_path: self._dcs._CONFIG}
@@ -543,63 +560,99 @@ class ObjectCache(Thread):
         with self._object_cache_lock:
             return self._object_cache.get(name)
 
+    def _process_event(self, event):
+        ev_type = event['type']
+        obj = event['object']
+        name = obj['metadata']['name']
+
+        if ev_type in ('ADDED', 'MODIFIED'):
+            obj = K8sObject(obj)
+            success, old_value = self.set(name, obj)
+            if success:
+                new_value = (obj.metadata.annotations or {}).get(self._annotations_map.get(name))
+        elif ev_type == 'DELETED':
+            success, old_value = self.delete(name, obj['metadata']['resourceVersion'])
+            new_value = None
+        else:
+            return logger.warning('Unexpected event type: %s', ev_type)
+
+        if success and obj.get('kind') != 'Pod':
+            if old_value:
+                old_value = (old_value.metadata.annotations or {}).get(self._annotations_map.get(name))
+
+            value_changed = old_value != new_value and \
+                (name != self._dcs.config_path or old_value is not None and new_value is not None)
+
+            if value_changed:
+                logger.debug('%s changed from %s to %s', name, old_value, new_value)
+
+            # Do not wake up HA loop if we run as leader and received leader object update event
+            if value_changed or name == self._dcs.leader_path and self._name != new_value:
+                self._dcs.event.set()
+
+    @staticmethod
+    def _finish_response(response):
+        try:
+            response.close()
+        finally:
+            response.release_conn()
+
+    def _do_watch(self, resource_version):
+        with self._response_lock:
+            self._response = None
+        response = self._watch(resource_version)
+        with self._response_lock:
+            if self._response is None:
+                self._response = response
+
+        if not self._response:
+            return self._finish_response(response)
+
+        for event in iter_response_objects(response):
+            if event['object'].get('code') == 410:
+                break
+            self._process_event(event)
+
     def _build_cache(self):
         objects = self._list()
-        return_type = 'V1' + objects.kind[:-4]
         with self._object_cache_lock:
             self._object_cache = {item.metadata.name: item for item in objects.items}
         with self._condition:
             self._is_ready = True
             self._condition.notify()
 
-        response = self._watch(objects.metadata.resource_version)
         try:
-            for event in iter_response_objects(response):
-                obj = event['object']
-                if obj.get('code') == 410:
-                    break
-
-                ev_type = event['type']
-                name = obj['metadata']['name']
-
-                if ev_type in ('ADDED', 'MODIFIED'):
-                    obj = K8sObject(obj)
-                    success, old_value = self.set(name, obj)
-                    if success:
-                        new_value = (obj.metadata.annotations or {}).get(self._annotations_map.get(name))
-                elif ev_type == 'DELETED':
-                    success, old_value = self.delete(name, obj['metadata']['resourceVersion'])
-                    new_value = None
-                else:
-                    logger.warning('Unexpected event type: %s', ev_type)
-                    continue
-
-                if success and return_type != 'V1Pod':
-                    if old_value:
-                        old_value = (old_value.metadata.annotations or {}).get(self._annotations_map.get(name))
-
-                    value_changed = old_value != new_value and \
-                        (name != self._dcs.config_path or old_value is not None and new_value is not None)
-
-                    if value_changed:
-                        logger.debug('%s changed from %s to %s', name, old_value, new_value)
-
-                    # Do not wake up HA loop if we run as leader and received leader object update event
-                    if value_changed or name == self._dcs.leader_path and self._name != new_value:
-                        self._dcs.event.set()
+            self._do_watch(objects.metadata.resource_version)
         finally:
             with self._condition:
                 self._is_ready = False
-            response.close()
-            response.release_conn()
+            with self._response_lock:
+                response, self._response = self._response, None
+            if response:
+                self._finish_response(response)
+
+    def kill_stream(self):
+        sock = None
+        with self._response_lock:
+            if self._response:
+                try:
+                    sock = self._response.connection.sock
+                except Exception:
+                    sock = None
+            else:
+                self._response = False
+        if sock:
+            try:
+                sock.shutdown(socket.SHUT_RDWR)
+                sock.close()
+            except Exception as e:
+                logger.debug('Error on socket.shutdown: %r', e)
 
     def run(self):
         while True:
             try:
                 self._build_cache()
             except Exception as e:
-                with self._condition:
-                    self._is_ready = False
                 logger.error('ObjectCache.run %r', e)
 
     def is_ready(self):
@@ -874,7 +927,14 @@ class Kubernetes(AbstractDCS):
     def patch_or_create(self, name, annotations, resource_version=None, patch=False, retry=True, ips=None):
         if retry is True:
             retry = self.retry
-        return self._patch_or_create(name, annotations, resource_version, patch, retry, ips)
+        try:
+            return self._patch_or_create(name, annotations, resource_version, patch, retry, ips)
+        except k8s_client.rest.ApiException as e:
+            if e.status == 409 and resource_version:  # Conflict in resource_version
+                # Terminate watchers, it could be a sign that K8s API is in a failed state
+                self._kinds.kill_stream()
+                self._pods.kill_stream()
+            raise e
 
     def patch_or_create_config(self, annotations, resource_version=None, patch=False, retry=True):
         # SCOPE-config endpoint requires corresponding service otherwise it might be "cleaned" by k8s master
@@ -961,7 +1021,7 @@ class Kubernetes(AbstractDCS):
                        'transitions': leader_observed_record.get('transitions') or '0'}
         if last_lsn:
             annotations[self._OPTIME] = str(last_lsn)
-        annotations['slots'] = json.dumps(slots) if slots else None
+            annotations['slots'] = json.dumps(slots) if slots else None
 
         resource_version = kind and kind.metadata.resource_version
         return self._update_leader_with_retry(annotations, resource_version, self.__ips)
@@ -1045,7 +1105,7 @@ 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()
 

patroni/ha.py

@@ -293,17 +293,30 @@ class Ha(object):
 
         return result
 
+    def _handle_crash_recovery(self):
+        if not self._crash_recovery_executed and (self.cluster.is_unlocked() or self._rewind.can_rewind):
+            self._crash_recovery_executed = True
+            self._crash_recovery_started = time.time()
+            msg = 'doing crash recovery in a single user mode'
+            return self._async_executor.try_run_async(msg, self._rewind.ensure_clean_shutdown) or msg
+
     def _handle_rewind_or_reinitialize(self):
         leader = self.get_remote_master() if self.is_standby_cluster() else self.cluster.leader
         if not self._rewind.rewind_or_reinitialize_needed_and_possible(leader):
             return None
 
         if self._rewind.can_rewind:
+            # rewind is required, but postgres wasn't shut down cleanly.
+            if not self.state_handler.is_running() and \
+                    self.state_handler.controldata().get('Database cluster state') == 'in archive recovery':
+                msg = self._handle_crash_recovery()
+                if msg:
+                    return msg
+
             msg = 'running pg_rewind from ' + leader.name
             return self._async_executor.try_run_async(msg, self._rewind.execute, args=(leader,)) or msg
 
-        # remove_data_directory_on_diverged_timelines is set
-        if not self.is_standby_cluster():
+        if self._rewind.should_remove_data_directory_on_diverged_timelines and not self.is_standby_cluster():
             msg = 'reinitializing due to diverged timelines'
             return self._async_executor.try_run_async(msg, self._do_reinitialize, args=(self.cluster,)) or msg
 
@@ -325,13 +338,10 @@ class Ha(object):
 
         data = self.state_handler.controldata()
         logger.info('pg_controldata:\n%s\n', '\n'.join('  {0}: {1}'.format(k, v) for k, v in data.items()))
-        if data.get('Database cluster state') in ('in production', 'shutting down', 'in crash recovery') \
-                and not self._crash_recovery_executed and \
-                (self.cluster.is_unlocked() or self._rewind.can_rewind):
-            self._crash_recovery_executed = True
-            self._crash_recovery_started = time.time()
-            msg = 'doing crash recovery in a single user mode'
-            return self._async_executor.try_run_async(msg, self._rewind.ensure_clean_shutdown) or msg
+        if data.get('Database cluster state') in ('in production', 'shutting down', 'in crash recovery'):
+            msg = self._handle_crash_recovery()
+            if msg:
+                return msg
 
         self.load_cluster_from_dcs()
 
@@ -413,9 +423,10 @@ class Ha(object):
                 self.state_handler.get_history(self._leader_timeline + 1):
             self._rewind.trigger_check_diverged_lsn()
 
-        msg = self._handle_rewind_or_reinitialize()
-        if msg:
-            return msg
+        if not self.state_handler.is_starting():
+            msg = self._handle_rewind_or_reinitialize()
+            if msg:
+                return msg
 
         if not self.is_paused():
             self.state_handler.handle_parameter_change()
@@ -1480,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

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

@@ -577,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
@@ -650,16 +653,17 @@ class ConfigHandler(object):
         elif not primary_conninfo:
             return False
 
-        wal_receiver_primary_conninfo = self._postgresql.primary_conninfo()
-        if wal_receiver_primary_conninfo:
-            wal_receiver_primary_conninfo = parse_dsn(wal_receiver_primary_conninfo)
-            # when wal receiver is alive use primary_conninfo from pg_stat_wal_receiver for comparison
+        if not self._postgresql.is_starting():
+            wal_receiver_primary_conninfo = self._postgresql.primary_conninfo()
             if wal_receiver_primary_conninfo:
-                primary_conninfo = wal_receiver_primary_conninfo
-                # There could be no password in the primary_conninfo or it is masked.
-                # Just copy the "desired" value in order to make comparison succeed.
-                if 'password' in wanted_primary_conninfo:
-                    primary_conninfo['password'] = wanted_primary_conninfo['password']
+                wal_receiver_primary_conninfo = parse_dsn(wal_receiver_primary_conninfo)
+                # when wal receiver is alive use primary_conninfo from pg_stat_wal_receiver for comparison
+                if wal_receiver_primary_conninfo:
+                    primary_conninfo = wal_receiver_primary_conninfo
+                    # There could be no password in the primary_conninfo or it is masked.
+                    # Just copy the "desired" value in order to make comparison succeed.
+                    if 'password' in wanted_primary_conninfo:
+                        primary_conninfo['password'] = wanted_primary_conninfo['password']
 
         if 'passfile' in primary_conninfo and 'password' not in primary_conninfo \
                 and 'password' in wanted_primary_conninfo:
@@ -668,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
@@ -704,16 +708,19 @@ class ConfigHandler(object):
             else:  # empty string, primary_conninfo is not in the config
                 primary_conninfo[0] = {}
 
-        # when wal receiver is alive take primary_slot_name from pg_stat_wal_receiver
-        wal_receiver_primary_slot_name = self._postgresql.primary_slot_name()
-        if not wal_receiver_primary_slot_name and self._postgresql.primary_conninfo():
-            wal_receiver_primary_slot_name = ''
-        if wal_receiver_primary_slot_name is not None:
-            self._current_recovery_params['primary_slot_name'][0] = wal_receiver_primary_slot_name
+        if not self._postgresql.is_starting():
+            # when wal receiver is alive take primary_slot_name from pg_stat_wal_receiver
+            wal_receiver_primary_slot_name = self._postgresql.primary_slot_name()
+            if not wal_receiver_primary_slot_name and self._postgresql.primary_conninfo():
+                wal_receiver_primary_slot_name = ''
+            if wal_receiver_primary_slot_name is not None:
+                self._current_recovery_params['primary_slot_name'][0] = wal_receiver_primary_slot_name
 
         # Increment the 'reload' to enforce write of postgresql.conf when joining the running postgres
         required = {'restart': 0,
-                    'reload': int(not self._postgresql.cb_called and self._postgresql.major_version >= 120000)}
+                    'reload': int(self._postgresql.major_version >= 120000
+                                  and not self._postgresql.cb_called
+                                  and not self._postgresql.is_starting())}
 
         def record_missmatch(mtype):
             required['restart' if mtype else 'reload'] += 1
@@ -780,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)
@@ -789,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:
@@ -112,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"))
@@ -182,8 +190,14 @@ class Rewind(object):
         if isinstance(leader, Leader) and leader.member.data.get('role') != 'master':
             return
 
-        if not self.check_leader_is_not_in_recovery(
-                self._conn_kwargs(leader, self._postgresql.config.replication)):
+        # We want to use replication credentials when connecting to the "postgres" database in case if
+        # `use_pg_rewind` isn't enabled and only `remove_data_directory_on_diverged_timelines` is set
+        # for Postgresql older than v11 (where Patroni can't use a dedicated user for rewind).
+        # In all other cases we will use rewind or superuser credentials.
+        check_credentials = self._postgresql.config.replication if not self.enabled and\
+            self.should_remove_data_directory_on_diverged_timelines and\
+            self._postgresql.major_version < 110000 else self._postgresql.config.rewind_credentials
+        if not self.check_leader_is_not_in_recovery(self._conn_kwargs(leader, check_credentials)):
             return
 
         history = need_rewind = None
@@ -311,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:
@@ -329,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'])
@@ -368,19 +391,22 @@ class Rewind(object):
 
         if self.pg_rewind(r):
             self._state = REWIND_STATUS.SUCCESS
-        elif not self.check_leader_is_not_in_recovery(r):
-            logger.warning('Failed to rewind because master %s become unreachable', leader.name)
         else:
-            logger.error('Failed to rewind from healty master: %s', leader.name)
-
-            for name in ('remove_data_directory_on_rewind_failure', 'remove_data_directory_on_diverged_timelines'):
-                if self._postgresql.config.get(name):
-                    logger.warning('%s is set. removing...', name)
-                    self._postgresql.remove_data_directory()
-                    self._state = REWIND_STATUS.INITIAL
-                    break
+            if not self.check_leader_is_not_in_recovery(r):
+                logger.warning('Failed to rewind because master %s become unreachable', leader.name)
+                if not self.can_rewind:  # It is possible that the previous attempt damaged pg_control file!
+                    self._state = REWIND_STATUS.FAILED
             else:
+                logger.error('Failed to rewind from healty master: %s', leader.name)
                 self._state = REWIND_STATUS.FAILED
+
+            if self.failed:
+                for name in ('remove_data_directory_on_rewind_failure', 'remove_data_directory_on_diverged_timelines'):
+                    if self._postgresql.config.get(name):
+                        logger.warning('%s is set. removing...', name)
+                        self._postgresql.remove_data_directory()
+                        self._state = REWIND_STATUS.INITIAL
+                        break
         return False
 
     def reset_state(self):

patroni/postgresql/slots.py

@@ -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):
@@ -245,35 +245,68 @@ class SlotsHandler(object):
             slot_name = cluster.get_my_slot_name_on_primary(self._postgresql.name, replicatefrom)
             try:
                 with self._get_leader_connection_cursor(cluster.leader) as cur:
-                    cur.execute("SELECT catalog_xmin FROM pg_catalog.pg_get_replication_slots()"
-                                " WHERE NOT pg_catalog.pg_is_in_recovery() AND slot_name = %s", (slot_name,))
-                    if cur.rowcount < 1:
+                    cur.execute("SELECT slot_name, catalog_xmin FROM pg_catalog.pg_get_replication_slots()"
+                                " WHERE NOT pg_catalog.pg_is_in_recovery() AND slot_name = ANY(%s)",
+                                ([n for n, v in self._unready_logical_slots.items() if v is None] + [slot_name],))
+                    slots = {row[0]: row[1] for row in cur}
+                    if slot_name not in slots:
                         return logger.warning('Physical slot %s does not exist on the primary', slot_name)
-                    catalog_xmin = cur.fetchone()[0]
+                    catalog_xmin = slots.pop(slot_name)
             except Exception as e:
                 return logger.error("Failed to check %s physical slot on the primary: %r", slot_name, e)
+            # Remember catalog_xmin of logical slots on the primary when catalog_xmin of
+            # the physical slot became valid. Logical slots on replica will be safe to use after
+            # promote when catalog_xmin of the physical slot overtakes these values.
+            if catalog_xmin:
+                for name, value in slots.items():
+                    self._unready_logical_slots[name] = value
+            else:  # Replica isn't streaming or the hot_standby_feedback isn't enabled
+                try:
+                    cur = self._query("SELECT pg_catalog.current_setting('hot_standby_feedback')::boolean")
+                    if not cur.fetchone()[0]:
+                        return logger.error('Logical slot failover requires "hot_standby_feedback".'
+                                            ' Please check postgresql.auto.conf')
+                except Exception as e:
+                    return logger.error('Failed to check the hot_standby_feedback setting: %r', e)
+
         for name in list(self._unready_logical_slots):
             value = self._replication_slots.get(name)
-            if not value or catalog_xmin <= value['catalog_xmin']:
-                self._unready_logical_slots.remove(name)
+            # The logical slot on a replica is safe to use when the physical replica slot on the primary:
+            # 1. has a nonzero/non-null catalog_xmin
+            # 2. has a catalog_xmin that is not newer (greater) than the catalog_xmin of any slot on the standby
+            # 3. overtook the catalog_xmin of remembered values of logical slots on the primary.
+            if not value or self._unready_logical_slots[name] <= catalog_xmin <= value['catalog_xmin']:
+                del self._unready_logical_slots[name]
                 if value:
                     logger.info('Logical slot %s is safe to be used after a failover', name)
 
-    def copy_logical_slots(self, leader, slots):
+    def copy_logical_slots(self, cluster, create_slots):
+        leader = cluster.leader
+        slots = cluster.get_replication_slots(self._postgresql.name, 'replica', False, self._postgresql.major_version)
         with self._get_leader_connection_cursor(leader) as cur:
             try:
-                cur.execute("SELECT slot_name, catalog_xmin, "
+                cur.execute("SELECT slot_name, slot_type, datname, plugin, catalog_xmin, "
                             "pg_catalog.pg_wal_lsn_diff(confirmed_flush_lsn, '0/0')::bigint, "
                             "pg_catalog.pg_read_binary_file('pg_replslot/' || slot_name || '/state')"
-                            " FROM pg_catalog.pg_get_replication_slots() WHERE NOT pg_catalog.pg_is_in_recovery()"
-                            " AND slot_name = ANY(%s)", (slots,))
-                slots = {r[0]: {'catalog_xmin': r[1], 'confirmed_flush_lsn': r[2], 'data': r[3]} for r in cur}
+                            " FROM pg_catalog.pg_get_replication_slots() JOIN pg_catalog.pg_database ON datoid = oid"
+                            " WHERE NOT pg_catalog.pg_is_in_recovery() AND slot_name = ANY(%s)", (create_slots,))
+
+                create_slots = {}
+                for r in cur:
+                    if r[0] in slots:  # slot_name is defined in the global configuration
+                        slot = {'type': r[1], 'database': r[2], 'plugin': r[3],
+                                'catalog_xmin': r[4], 'confirmed_flush_lsn': r[5], 'data': r[6]}
+                        if compare_slots(slot, slots[r[0]]):
+                            create_slots[r[0]] = slot
+                        else:
+                            logger.warning('Will not copy the logical slot "%s" due to the configuration mismatch: ' +
+                                           'configuration=%s, slot on the primary=%s', r[0], slots[r[0]], slot)
             except Exception as e:
                 logger.error("Failed to copy logical slots from the %s via postgresql connection: %r", leader.name, e)
 
-        if isinstance(slots, dict) and self._postgresql.stop():
+        if isinstance(create_slots, dict) and create_slots and self._postgresql.stop():
             pg_replslot_dir = os.path.join(self._postgresql.data_dir, 'pg_replslot')
-            for name, value in slots.items():
+            for name, value in create_slots.items():
                 slot_dir = os.path.join(pg_replslot_dir, name)
                 slot_tmp_dir = slot_dir + '.tmp'
                 if os.path.exists(slot_tmp_dir):
@@ -288,7 +321,7 @@ class SlotsHandler(object):
                     shutil.rmtree(slot_dir)
                 os.rename(slot_tmp_dir, slot_dir)
                 fsync_dir(slot_dir)
-                self._unready_logical_slots.add(name)
+                self._unready_logical_slots[name] = None
             fsync_dir(pg_replslot_dir)
             self._postgresql.start()
 
@@ -300,4 +333,4 @@ class SlotsHandler(object):
     def on_promote(self):
         if self._unready_logical_slots:
             logger.warning('Logical replication slots that might be unsafe to use after promote: %s',
-                           self._unready_logical_slots)
+                           set(self._unready_logical_slots))

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',))
@@ -158,7 +160,10 @@ parameters = CaseInsensitiveDict({
     '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),
@@ -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),
@@ -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)
@@ -463,8 +475,12 @@ parameters = CaseInsensitiveDict({
     '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/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 patroni.request import get as requests_get
+from ..utils import Retry, RetryFailedError
+from ..request import get as requests_get
+
+from botocore.exceptions import ClientError
 
 logger = logging.getLogger(__name__)
 
@@ -16,7 +18,7 @@ class AWSConnection(object):
     def __init__(self, cluster_name):
         self.available = False
         self.cluster_name = cluster_name if cluster_name is not None else 'unknown'
-        self._retry = Retry(deadline=300, max_delay=30, max_tries=-1, retry_exceptions=(boto.exception.StandardError,))
+        self._retry = Retry(deadline=300, max_delay=30, max_tries=-1, retry_exceptions=(ClientError,))
         try:
             # get the instance id
             r = requests_get('http://169.254.169.254/latest/dynamic/instance-identity/document', timeout=2.1)
@@ -42,20 +44,22 @@ class AWSConnection(object):
 
     def _tag_ebs(self, conn, role):
         """ set tags, carrying the cluster name, instance role and instance id for the EBS storage """
-        tags = {'Name': 'spilo_' + self.cluster_name, 'Role': role, 'Instance': self.instance_id}
-        volumes = conn.get_all_volumes(filters={'attachment.instance-id': self.instance_id})
-        conn.create_tags([v.id for v in volumes], tags)
+        tags = [{'Key': 'Name', 'Value': 'spilo_' + self.cluster_name},
+                {'Key': 'Role', 'Value': role},
+                {'Key': 'Instance', 'Value': self.instance_id}]
+        volumes = conn.volumes.filter(Filters=[{'Name': 'attachment.instance-id', 'Values': [self.instance_id]}])
+        conn.create_tags(Resources=[v.id for v in volumes], Tags=tags)
 
     def _tag_ec2(self, conn, role):
         """ tag the current EC2 instance with a cluster role """
-        tags = {'Role': role}
-        conn.create_tags([self.instance_id], tags)
+        tags = [{'Key': 'Role', 'Value': role}]
+        conn.create_tags(Resources=[self.instance_id], Tags=tags)
 
     def on_role_change(self, new_role):
         if not self.available:
             return False
         try:
-            conn = self.retry(boto.ec2.connect_to_region, self.region)
+            conn = boto3.resource('ec2', region_name=self.region)
             self.retry(self._tag_ec2, conn, new_role)
             self.retry(self._tag_ebs, conn, new_role)
         except RetryFailedError:

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.3'
+__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

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

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

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

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

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()'):

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

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):
-        if len(objects) == 0:
-            raise boto.exception.BotoServerError(503, 'Service Unavailable', 'Request limit exceeded')
+    def create_tags(Resources, **kwargs):
+        if len(Resources) == 0:
+            raise botocore.exceptions.ClientError({'Error': {'Code': 503, 'Message': 'Request limit exceeded'}},
+                                                  'create_tags')
         return True
 
 
-@patch('boto.ec2.connect_to_region', Mock(return_value=MockEc2Connection()))
+@patch('boto3.resource', Mock(return_value=MockEc2Connection()))
 class TestAWSConnection(unittest.TestCase):
 
     @patch('patroni.scripts.aws.requests_get', Mock(return_value=urllib3.HTTPResponse(
@@ -32,7 +38,7 @@ class TestAWSConnection(unittest.TestCase):
 
     def test_on_role_change(self):
         self.assertTrue(self.conn.on_role_change('master'))
-        with patch.object(MockEc2Connection, 'get_all_volumes', Mock(return_value=[])):
+        with patch.object(MockVolumes, 'filter', Mock(return_value=[])):
             self.conn._retry.max_tries = 1
             self.assertFalse(self.conn.on_role_change('master'))
 
@@ -46,8 +52,7 @@ class TestAWSConnection(unittest.TestCase):
         conn = AWSConnection('test')
         self.assertFalse(conn.aws_available())
 
-    @patch('patroni.scripts.aws.requests_get', Mock(return_value=urllib3.HTTPResponse(
-        status=200, body=b'{"instanceId": "012345", "region": "eu-west-1"}')))
+    @patch('patroni.scripts.aws.requests_get', Mock(return_value=urllib3.HTTPResponse(status=503, body=b'Error')))
     @patch('sys.exit', Mock())
     def test_main(self):
         self.assertIsNone(_main())

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

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,41 +84,44 @@ class TestSlotsHandler(BaseTestPostgresql):
     @patch.object(Postgresql, 'is_leader', Mock(return_value=False))
     def test__ensure_logical_slots_replica(self):
         self.p.set_role('replica')
-        config = ClusterConfig(1, {'slots': {'ls': {'database': 'a', 'plugin': 'b'}}}, 1)
-        cluster = Cluster(True, config, self.leader, 0,
-                          [self.me, self.other, self.leadermem], None, None, None, {'ls': 12346})
-        self.assertEqual(self.s.sync_replication_slots(cluster, False), [])
+        self.cluster.slots['ls'] = 12346
+        with patch.object(SlotsHandler, 'check_logical_slots_readiness', Mock()):
+            self.assertEqual(self.s.sync_replication_slots(self.cluster, False), [])
         self.s._schedule_load_slots = False
         with patch.object(MockCursor, 'execute', Mock(side_effect=psycopg.OperationalError)),\
                 patch.object(psycopg.OperationalError, 'diag') as mock_diag:
             type(mock_diag).sqlstate = PropertyMock(return_value='58P01')
-            self.assertEqual(self.s.sync_replication_slots(cluster, False), ['ls'])
-        cluster.slots['ls'] = 'a'
-        self.assertEqual(self.s.sync_replication_slots(cluster, False), [])
+            self.assertEqual(self.s.sync_replication_slots(self.cluster, False), ['ls'])
+        self.cluster.slots['ls'] = 'a'
+        self.assertEqual(self.s.sync_replication_slots(self.cluster, False), [])
         with patch.object(MockCursor, 'rowcount', PropertyMock(return_value=1), create=True):
-            self.assertEqual(self.s.sync_replication_slots(cluster, False), ['ls'])
+            self.assertEqual(self.s.sync_replication_slots(self.cluster, False), ['ls'])
 
-    @patch.object(MockCursor, 'execute', Mock(side_effect=psycopg.OperationalError))
     def test_copy_logical_slots(self):
-        self.s.copy_logical_slots(self.leader, ['foo'])
+        self.cluster.config.data['slots']['ls']['database'] = 'b'
+        self.s.copy_logical_slots(self.cluster, ['ls'])
+        with patch.object(MockCursor, 'execute', Mock(side_effect=psycopg.OperationalError)):
+            self.s.copy_logical_slots(self.cluster, ['foo'])
 
     @patch.object(Postgresql, 'stop', Mock(return_value=True))
     @patch.object(Postgresql, 'start', Mock(return_value=True))
     @patch.object(Postgresql, 'is_leader', Mock(return_value=False))
     def test_check_logical_slots_readiness(self):
-        self.s.copy_logical_slots(self.leader, ['ls'])
-        config = ClusterConfig(1, {'slots': {'ls': {'database': 'a', 'plugin': 'b'}}}, 1)
-        cluster = Cluster(True, config, self.leader, 0,
-                          [self.me, self.other, self.leadermem], None, None, None, {'ls': 12345})
-        self.assertEqual(self.s.sync_replication_slots(cluster, False), [])
-        with patch.object(MockCursor, 'rowcount', PropertyMock(return_value=1), create=True):
-            self.s.check_logical_slots_readiness(cluster, False, None)
+        self.s.copy_logical_slots(self.cluster, ['ls'])
+        with patch.object(MockCursor, '__iter__', Mock(return_value=iter([('postgresql0', None)]))),\
+                patch.object(MockCursor, 'fetchone', Mock(side_effect=Exception)):
+            self.assertIsNone(self.s.check_logical_slots_readiness(self.cluster, False, None))
+        with patch.object(MockCursor, '__iter__', Mock(return_value=iter([('postgresql0', None)]))),\
+                patch.object(MockCursor, 'fetchone', Mock(return_value=(False,))):
+            self.assertIsNone(self.s.check_logical_slots_readiness(self.cluster, False, None))
+        with patch.object(MockCursor, '__iter__', Mock(return_value=iter([('ls', 100)]))):
+            self.s.check_logical_slots_readiness(self.cluster, False, None)
 
     @patch.object(Postgresql, 'stop', Mock(return_value=True))
     @patch.object(Postgresql, 'start', Mock(return_value=True))
     @patch.object(Postgresql, 'is_leader', Mock(return_value=False))
     def test_on_promote(self):
-        self.s.copy_logical_slots(self.leader, ['ls'])
+        self.s.copy_logical_slots(self.cluster, ['ls'])
         self.s.on_promote()
 
     @unittest.skipIf(os.name == 'nt', "Windows not supported")