Repository: zalando/patroni
Pull request #2283 (2022-04-28 10:39:14)

K8s: reset watchers if PATCH fails with 409
High CPU load on Etcd nodes and K8s API servers created a very strange situation. A few clusters were running without a leader and the pod which is ahead of others was failing to take a leader lock because updates were failing with HTTP response code `409` (`resource_version` mismatch).

Effectively that means that TCP connections to K8s master nodes were alive (in the opposite case tcp keepalives would have resolved it), but no `UPDATE` events were arriving via these connections, resulting in the stale cache of the cluster in memory.

The only good way to prevent this situation is to intercept 409 HTTP responses and terminate existing TCP connections used for watches.

Now a few words about implementation. Unfortunately, watch threads are waiting in the read() call most of the time and there is no good way to interrupt them. But, the `socket.shutdown()` seems to do this job. We already used this trick in the Etcd3 implementation.

This approach will help to mitigate the issue of not having a leader, but at the same time replicas might still end up with the stale cluster state cached and in the worst case will not stream from the leader. Non-streaming replicas are less dangerous and could be covered by monitoring and partially mitigated by correctly configured `archive_command` and `restore_command`.


Patch:
diff --git a/patroni/dcs/kubernetes.py b/patroni/dcs/kubernetes.py
index 25969baf..3f1c1e51 100644
--- a/patroni/dcs/kubernetes.py
+++ b/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 _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):
+    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()
 
diff --git a/patroni/ha.py b/patroni/ha.py
index 7ec0d6ce..9f84613f 100644
--- a/patroni/ha.py
+++ b/patroni/ha.py
@@ -307,7 +307,8 @@ class Ha(object):
 
         if self._rewind.can_rewind:
             # rewind is required, but postgres wasn't shut down cleanly.
-            if self.state_handler.controldata().get('Database cluster state') == 'in archive recovery':
+            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
@@ -315,8 +316,7 @@ class Ha(object):
             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
 
@@ -423,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()
diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py
index 3b9f2507..c52c397b 100644
--- a/patroni/postgresql/__init__.py
+++ b/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()
 
diff --git a/patroni/postgresql/config.py b/patroni/postgresql/config.py
index 6afb9e77..93143152 100644
--- a/patroni/postgresql/config.py
+++ b/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
diff --git a/patroni/postgresql/rewind.py b/patroni/postgresql/rewind.py
index 5a2e4a39..db7055df 100644
--- a/patroni/postgresql/rewind.py
+++ b/patroni/postgresql/rewind.py
@@ -46,9 +46,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:
@@ -374,19 +378,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):
diff --git a/patroni/postgresql/slots.py b/patroni/postgresql/slots.py
index ab47944c..32cbc9b8 100644
--- a/patroni/postgresql/slots.py
+++ b/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,17 +245,38 @@ 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)
 
@@ -300,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()
 
@@ -312,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))


Test patch:
diff --git a/tests/__init__.py b/tests/__init__.py
index 7c1a23cf..9a81108f 100644
--- a/tests/__init__.py
+++ b/tests/__init__.py
@@ -91,8 +91,8 @@ 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 = [('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'):
diff --git a/tests/test_ha.py b/tests/test_ha.py
index 56277f97..5ed4503d 100644
--- a/tests/test_ha.py
+++ b/tests/test_ha.py
@@ -314,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
diff --git a/tests/test_kubernetes.py b/tests/test_kubernetes.py
index 92339b5a..fbdd9b67 100644
--- a/tests/test_kubernetes.py
+++ b/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()
diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py
index 523072c2..3f9bd738 100644
--- a/tests/test_postgresql.py
+++ b/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'))
diff --git a/tests/test_slots.py b/tests/test_slots.py
index df8fe9c3..454fab8d 100644
--- a/tests/test_slots.py
+++ b/tests/test_slots.py
@@ -85,7 +85,8 @@ class TestSlotsHandler(BaseTestPostgresql):
     def test__ensure_logical_slots_replica(self):
         self.p.set_role('replica')
         self.cluster.slots['ls'] = 12346
-        self.assertEqual(self.s.sync_replication_slots(self.cluster, False), [])
+        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:
@@ -107,8 +108,13 @@ class TestSlotsHandler(BaseTestPostgresql):
     @patch.object(Postgresql, 'is_leader', Mock(return_value=False))
     def test_check_logical_slots_readiness(self):
         self.s.copy_logical_slots(self.cluster, ['ls'])
-        self.assertEqual(self.s.sync_replication_slots(self.cluster, False), [])
-        with patch.object(MockCursor, 'rowcount', PropertyMock(return_value=1), create=True):
+        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))