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