gitops/patroni
1
0
Fork 0
mirror of https://github.com/outbackdingo/patroni.git synced 2026-08-25 14:53:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

Follow up on #3148 (#3167)

Browse Source 
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`.
This commit is contained in:
Alexander Kukushkin
2024-09-17 12:02:12 +02:00
committed by GitHub
parent d7e172c20a
commit 78a46b9ebc
2 changed files with 43 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
patroni/dcs/__init__.py
+17 -6
View File
@@ -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]:
tests/test_slots.py
+26
View File
@@ -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}, []),