From 5f6197aaad3ab0fb6c7eea9b8dd3cf6f141a616e Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Thu, 14 Apr 2022 12:10:37 +0200 Subject: [PATCH] Don't copy logical slot if there is mismatch with the config (#2274) A couple of times we have seen in the wild that the database for the permanent logical slots was changed in the Patroni config. It resulted in the below situation. On the primary: 1. The slot must be dropped before creating it in a different DB. 2. Patroni fails to drop it because the slot is in use. Replica: 1. Patroni notice that the slot exists in the wrong DB and successfully dropping it. 2. Patroni copying the existing slot from the primary by its name with Postgres restart. And the loop repeats while the "wrong" slot exists on the primary. Basically, replicas are continuously restarting, which badly affects availability. In order to solve the problem, we will perform additional checks while copying replication slot files from the primary and discard them if `slot_type`, `database`, or `plugin` don't match our expectations. --- patroni/ha.py | 2 +- patroni/postgresql/slots.py | 26 +++++++++++++++++++------- tests/__init__.py | 4 ++-- tests/test_ha.py | 5 ++++- tests/test_slots.py | 34 +++++++++++++++++----------------- 5 files changed, 43 insertions(+), 28 deletions(-) diff --git a/patroni/ha.py b/patroni/ha.py index d5f47bd6..7ec0d6ce 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -1490,7 +1490,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 diff --git a/patroni/postgresql/slots.py b/patroni/postgresql/slots.py index 8e6a2d87..ab47944c 100644 --- a/patroni/postgresql/slots.py +++ b/patroni/postgresql/slots.py @@ -259,21 +259,33 @@ class SlotsHandler(object): 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): diff --git a/tests/__init__.py b/tests/__init__.py index 2e867b57..7c1a23cf 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -93,8 +93,8 @@ class MockCursor(object): 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')] + 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()'): diff --git a/tests/test_ha.py b/tests/test_ha.py index e2ef7b6f..56277f97 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -1211,9 +1211,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')) diff --git a/tests/test_slots.py b/tests/test_slots.py index 6ab66a82..df8fe9c3 100644 --- a/tests/test_slots.py +++ b/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,38 @@ 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 + 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), []) + 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): - self.s.check_logical_slots_readiness(cluster, False, None) + 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")