From b901e62ad0dc5f0d9596d5ce9ca9d04af5f7142a Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 10 May 2022 12:24:47 +0200 Subject: [PATCH] Enhanced checks of replica logical slots safety (#2285) The logical slot on a replica is safe to use when the physical replica slot on the primary: 1. has a nonzero/non-null `catalog_xmin` 2. has a `catalog_xmin` that is not newer (greater) than the `catalog_xmin` of any slot on the standby 3. the `catalog_xmin` is known to overtake `catalog_xmin` of logical slots on the primary observed during `1` In case if `1` doesn't take place, Patroni will run an additional check whether the `hot_standby_feedback` is actually in effect and shows the warning in case it is not. --- patroni/postgresql/slots.py | 41 ++++++++++++++++++++++++++++--------- tests/__init__.py | 4 ++-- tests/test_slots.py | 12 ++++++++--- 3 files changed, 42 insertions(+), 15 deletions(-) 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)) 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_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))