From 78a46b9ebc710daa620f223cb48c70909a1ccaba Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 17 Sep 2024 12:02:12 +0200 Subject: [PATCH] Follow up on #3148 (#3167) the original fix didn't address same problem with with permanent slots, but only the part with member slots being retained due to `member_slots_ttl`. --- patroni/dcs/__init__.py | 23 +++++++++++++++++------ tests/test_slots.py | 26 ++++++++++++++++++++++++++ 2 files changed, 43 insertions(+), 6 deletions(-) diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index 53a839c2..7eb67373 100644 --- a/patroni/dcs/__init__.py +++ b/patroni/dcs/__init__.py @@ -1028,7 +1028,7 @@ class Cluster(NamedTuple('Cluster', permanent_slots: Dict[str, Any] = self._get_permanent_slots(postgresql, member, role) disabled_permanent_logical_slots: List[str] = self._merge_permanent_slots( - slots, permanent_slots, name, postgresql.can_advance_slots) + slots, permanent_slots, name, role, postgresql.can_advance_slots) if disabled_permanent_logical_slots and show_error: logger.error("Permanent logical replication slots supported by Patroni only starting from PostgreSQL 11. " @@ -1036,8 +1036,8 @@ class Cluster(NamedTuple('Cluster', return slots - def _merge_permanent_slots(self, slots: Dict[str, Dict[str, str]], permanent_slots: Dict[str, Any], name: str, - can_advance_slots: bool) -> List[str]: + def _merge_permanent_slots(self, slots: Dict[str, Dict[str, Any]], permanent_slots: Dict[str, Any], + name: str, role: str, can_advance_slots: bool) -> List[str]: """Merge replication *slots* for members with *permanent_slots*. Perform validation of configured permanent slot name, skipping invalid names. @@ -1047,12 +1047,17 @@ class Cluster(NamedTuple('Cluster', :param slots: Slot names with existing attributes if known. :param name: name of this node. + :param role: role of the node -- ``primary``, ``standby_leader`` or ``replica``. :param permanent_slots: dictionary containing slot name key and slot information values. :param can_advance_slots: ``True`` if ``pg_replication_slot_advance()`` function is available, ``False`` otherwise. :returns: List of disabled permanent, logical slot names, if postgresql version < 11. """ + name = slot_name_from_member_name(name) + topology = {slot_name_from_member_name(m.name): m.replicatefrom and slot_name_from_member_name(m.replicatefrom) + for m in self.members} + disabled_permanent_logical_slots: List[str] = [] for slot_name, value in permanent_slots.items(): @@ -1068,8 +1073,14 @@ class Cluster(NamedTuple('Cluster', if value['type'] == 'physical': # Don't try to create permanent physical replication slot for yourself - if slot_name not in slots and slot_name != slot_name_from_member_name(name): - slots[slot_name] = value + if slot_name not in slots and slot_name != name: + # On the leader we expected to have permanent slots active, except the case when it is a slot + # for a cascading replica. Lets consider a configuration with C being a permanent slot. In this + # case we should have the following: A(B: active, C: inactive) <- B (C: active) <- C + # We don't consider the same situation on node B, because if node C doesn't exists, we will not + # be able to know its `replicatefrom` tag value. + expected_active = not topology.get(slot_name) and role in ('primary', 'standby_leader') + slots[slot_name] = {**value, 'expected_active': expected_active} continue if self.is_logical_slot(value): @@ -1228,7 +1239,7 @@ class Cluster(NamedTuple('Cluster', postgresql.can_advance_slots) permanent_slots: Dict[str, Any] = self._get_permanent_slots(postgresql, member, role) slots = deepcopy(members_slots) - self._merge_permanent_slots(slots, permanent_slots, postgresql.name, postgresql.can_advance_slots) + self._merge_permanent_slots(slots, permanent_slots, postgresql.name, role, postgresql.can_advance_slots) return len(slots) > len(members_slots) or any(self.is_physical_slot(v) for v in permanent_slots.values()) def maybe_filter_permanent_slots(self, postgresql: 'Postgresql', slots: Dict[str, int]) -> Dict[str, int]: diff --git a/tests/test_slots.py b/tests/test_slots.py index fa2acc09..d003a0bb 100644 --- a/tests/test_slots.py +++ b/tests/test_slots.py @@ -295,7 +295,33 @@ class TestSlotsHandler(BaseTestPostgresql): self.s.schedule_advance_slots({'foo': {'bar': 100}}) self.s._advance.sync_slots() + def test_advance_physical_primary(self): + self.p.name = self.me.name + config = ClusterConfig(1, {'member_slots_ttl': 0, 'slots': {'test_1': {'type': 'physical'}}}, 1) + cluster = Cluster(True, config, self.leader, Status(0, {}, []), + [self.me, self.other, self.leadermem], None, SyncState.empty(), None, None) + self.other.data['xlog_location'] = 12346 + global_config.update(cluster) + + # Should advance permanent physical slot on the primary for a node that is cascading from the other node + with patch.object(SlotsHandler, '_query', Mock(side_effect=[[('test_1', 'physical', None, 12345, None, None, + None, None, None)], Exception])) as mock_query, \ + patch('patroni.postgresql.slots.logger.error') as mock_error: + self.s.sync_replication_slots(cluster, self.tags) + self.assertEqual(mock_query.call_args[0], + ("SELECT pg_catalog.pg_replication_slot_advance(%s, %s)", "test_1", '0/303A')) + self.assertEqual(mock_error.call_args[0][0], + "Error while advancing replication slot %s to position '%s': %r") + + # Should drop permanent physical slot on the primary for a node + # that is cascading from the other node if given slot has xmin set + with patch.object(SlotsHandler, '_query', Mock(side_effect=[[('test_1', 'physical', 1, 12345, None, None, + None, None, None)], Exception])) as mock_query: + self.s.sync_replication_slots(cluster, self.tags) + self.assertTrue(mock_query.call_args[0][0].startswith('WITH slots AS (SELECT slot_name, active')) + @patch.object(Postgresql, 'is_primary', Mock(return_value=False)) + @patch.object(Postgresql, 'role', PropertyMock(return_value='replica')) def test_advance_physical_slots(self): config = ClusterConfig(1, {'slots': {'blabla': {'type': 'physical'}, 'leader': None}}, 1) cluster = Cluster(True, config, self.leader, Status(0, {'blabla': 12346}, []),