From c5fffb3c976a9ce0e2529ab7d12fa705d5ec43ce Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Mon, 23 Oct 2023 08:24:28 +0200 Subject: [PATCH] Further work on permanent physical slots (#2891) - Fixed issues with has_permanent_slots() method. It didn't took into account the case of permanent physical slots for members, falsely concluding that there are no permanent slots. - Write to the status key only LSNs for permanent slots (not just for slots that exist on the primary). - Include pg_current_wal_flush_lsn() to slots feedback, so that slots on standby nodes could be advanced - Improved behave tests: - Verify that permanent slots are properly created on standby nodes - Verify that permanent slots are properly advanced, including DCS failsafe mode - Verify that only permanent slots are written to the `/status` --- docs/dynamic_configuration.rst | 7 ++- features/dcs_failsafe_mode.feature | 44 ++++++++++++-- features/environment.py | 5 +- features/ignored_slots.feature | 2 +- features/permanent_slots.feature | 55 ++++++++++++++--- features/standby_cluster.feature | 2 +- features/steps/basic_replication.py | 2 +- features/steps/slots.py | 18 +++++- patroni/dcs/__init__.py | 93 +++++++++++++++++++++-------- patroni/ha.py | 53 ++++++++++++++-- patroni/postgresql/__init__.py | 16 +++-- patroni/postgresql/slots.py | 15 ++--- patroni/utils.py | 2 +- tests/test_ha.py | 2 + tests/test_slots.py | 1 + 15 files changed, 250 insertions(+), 67 deletions(-) diff --git a/docs/dynamic_configuration.rst b/docs/dynamic_configuration.rst index a16a2a40..f38a8b39 100644 --- a/docs/dynamic_configuration.rst +++ b/docs/dynamic_configuration.rst @@ -57,7 +57,7 @@ In order to change the dynamic configuration you can use either :ref:`patronictl - **slots**: define permanent replication slots. These slots will be preserved during switchover/failover. Permanent slots that don't exist will be created by Patroni. With PostgreSQL 11 onwards permanent physical slots are created on all nodes and their position is advanced every **loop_wait** seconds. For PostgreSQL versions older than 11 permanent physical replication slots are maintained only on the current primary. The logical slots are copied from the primary to a standby with restart, and after that their position advanced every **loop_wait** seconds (if necessary). Copying logical slot files performed via ``libpq`` connection and using either rewind or superuser credentials (see **postgresql.authentication** section). There is always a chance that the logical slot position on the replica is a bit behind the former primary, therefore application should be prepared that some messages could be received the second time after the failover. The easiest way of doing so - tracking ``confirmed_flush_lsn``. Enabling permanent replication slots requires **postgresql.use_slots** to be set to ``true``. If there are permanent logical replication slots defined Patroni will automatically enable the ``hot_standby_feedback``. Since the failover of logical replication slots is unsafe on PostgreSQL 9.6 and older and PostgreSQL version 10 is missing some important functions, the feature only works with PostgreSQL 11+. - - **my\_slot\_name**: the name of the permanent replication slot. If the permanent slot name matches with the name of the current leader it will not be created. Please note that Patroni does not make checks for permanent slot names added to this configuration matching those that Patroni creates automatically for members. If those names are added, Patroni will ensure that any slots that were created are not removed even if the member becomes unresponsive, situation which would normally result in the slot's removal by Patroni. Although this can be useful in some situations, such as when importing existing members to a new Patroni cluster (see :ref:`Convert a Standalone to a Patroni Cluster ` for details), caution should be exercised by the operator that these clashes in names are not persisted in the DCS due to its effect on normal functioning of Patroni. + - **my\_slot\_name**: the name of the permanent replication slot. If the permanent slot name matches with the name of the current node it will not be created on this node. If you add a permanent physical replication slot which name matches the name of a Patroni member, Patroni will ensure that the slot that was created is not removed even if the corresponding member becomes unresponsive, situation which would normally result in the slot's removal by Patroni. Although this can be useful in some situations, such as when you want replication slots used by members to persist during temporary failures or when importing existing members to a new Patroni cluster (see :ref:`Convert a Standalone to a Patroni Cluster ` for details), caution should be exercised by the operator that these clashes in names are not persisted in the DCS, when the slot is no longer required, due to its effect on normal functioning of Patroni. - **type**: slot type. Could be ``physical`` or ``logical``. If the slot is logical, you have to additionally define ``database`` and ``plugin``. - **database**: the database name where logical slots should be created. @@ -103,3 +103,8 @@ Note: if cluster topology is static (fixed number of nodes that never change the node_name3: type: physical ... + + +.. warning:: + Permanent replication slots are synchronized only from the ``primary``/``standby_leader`` to replica nodes. That means, applications are supposed to be using them only from the leader node. Using them on replica nodes will cause indefinite growth of ``pg_wal`` on all other nodes in the cluster. + An exception to that rule are permanent physical slots that match the Patroni member names, if you happen to configure any. Those will be synchronized among all nodes as they are used for replication among them. diff --git a/features/dcs_failsafe_mode.feature b/features/dcs_failsafe_mode.feature index 8345d8ea..0489db39 100644 --- a/features/dcs_failsafe_mode.feature +++ b/features/dcs_failsafe_mode.feature @@ -11,7 +11,7 @@ Feature: dcs failsafe mode When I issue a GET request to http://127.0.0.1:8008/failsafe Then I receive a response code 200 And I receive a response postgres0 http://127.0.0.1:8008/patroni - When I issue a PATCH request to http://127.0.0.1:8008/config with {"postgresql": {"parameters": {"wal_level": "logical"}}} + When I issue a PATCH request to http://127.0.0.1:8008/config with {"postgresql": {"parameters": {"wal_level": "logical"}},"slots":{"dcs_slot_1": null,"postgres0":null}} Then I receive a response code 200 When I issue a PATCH request to http://127.0.0.1:8008/config with {"slots": {"dcs_slot_0": {"type": "logical", "database": "postgres", "plugin": "test_decoding"}}} Then I receive a response code 200 @@ -44,14 +44,18 @@ Feature: dcs failsafe mode @dcs-failsafe @slot-advance Scenario: check leader and replica are functioning while DCS is down - Given logical slot dcs_slot_0 is in sync between postgres0 and postgres1 after 10 seconds + Given I get all changes from physical slot dcs_slot_1 on postgres0 + Then physical slot dcs_slot_1 is in sync between postgres0 and postgres1 after 10 seconds + And logical slot dcs_slot_0 is in sync between postgres0 and postgres1 after 10 seconds And DCS is down Then Response on GET http://127.0.0.1:8008/primary contains failsafe_mode_is_active after 12 seconds Then postgres0 role is the primary after 10 seconds And postgres1 role is the replica after 2 seconds And replication works from postgres0 to postgres1 after 10 seconds - And I get all changes from logical slot dcs_slot_0 on postgres0 - And logical slot dcs_slot_0 is in sync between postgres0 and postgres1 after 20 seconds + When I get all changes from logical slot dcs_slot_0 on postgres0 + And I get all changes from physical slot dcs_slot_1 on postgres0 + Then logical slot dcs_slot_0 is in sync between postgres0 and postgres1 after 20 seconds + And physical slot dcs_slot_1 is in sync between postgres0 and postgres1 after 10 seconds @dcs-failsafe Scenario: check primary is demoted when one replica is shut down and DCS is down @@ -70,15 +74,43 @@ Feature: dcs failsafe mode And postgres1 role is the primary after 25 seconds @dcs-failsafe - Scenario: check three-node cluster is functioning while DCS is down + Scenario: scale to three-node cluster Given I start postgres0 And I start postgres2 Then "members/postgres2" key in DCS has state=running after 10 seconds And "members/postgres0" key in DCS has state=running after 20 seconds And Response on GET http://127.0.0.1:8008/failsafe contains postgres2 after 10 seconds And replication works from postgres1 to postgres0 after 10 seconds + And replication works from postgres1 to postgres2 after 10 seconds + + @dcs-failsafe + @slot-advance + Scenario: make sure permanent slots exist on replicas + Given I issue a PATCH request to http://127.0.0.1:8009/config with {"slots":{"dcs_slot_0":null,"dcs_slot_2":{"type":"logical","database":"postgres","plugin":"test_decoding"}}} + Then logical slot dcs_slot_2 is in sync between postgres1 and postgres0 after 20 seconds + And logical slot dcs_slot_2 is in sync between postgres1 and postgres2 after 20 seconds + When I get all changes from physical slot dcs_slot_1 on postgres1 + Then physical slot dcs_slot_1 is in sync between postgres1 and postgres0 after 10 seconds + And physical slot dcs_slot_1 is in sync between postgres1 and postgres2 after 10 seconds + And physical slot postgres0 is in sync between postgres1 and postgres2 after 10 seconds + + @dcs-failsafe + Scenario: check three-node cluster is functioning while DCS is down Given DCS is down - Then Response on GET http://127.0.0.1:8008/primary contains failsafe_mode_is_active after 12 seconds + Then Response on GET http://127.0.0.1:8009/primary contains failsafe_mode_is_active after 12 seconds Then postgres1 role is the primary after 10 seconds And postgres0 role is the replica after 2 seconds And postgres2 role is the replica after 2 seconds + + @dcs-failsafe + @slot-advance + Scenario: check that permanent slots are in sync between nodes while DCS is down + Given replication works from postgres1 to postgres0 after 10 seconds + And replication works from postgres1 to postgres2 after 10 seconds + When I get all changes from logical slot dcs_slot_2 on postgres1 + And I get all changes from physical slot dcs_slot_1 on postgres1 + Then logical slot dcs_slot_2 is in sync between postgres1 and postgres0 after 20 seconds + And logical slot dcs_slot_2 is in sync between postgres1 and postgres2 after 20 seconds + And physical slot dcs_slot_1 is in sync between postgres1 and postgres0 after 10 seconds + And physical slot dcs_slot_1 is in sync between postgres1 and postgres2 after 10 seconds + And physical slot postgres0 is in sync between postgres1 and postgres2 after 10 seconds diff --git a/features/environment.py b/features/environment.py index 3e0ad3b1..1c3e654b 100644 --- a/features/environment.py +++ b/features/environment.py @@ -654,9 +654,10 @@ class KubernetesController(AbstractExternalDcsController): try: if group is not None: scope = '{0}-{1}'.format(scope, group) - ep = scope + {'leader': '', 'history': '-config', 'initialize': '-config'}.get(key, '-' + key) + rkey = 'leader' if key in ('status', 'failsafe') else key + ep = scope + {'leader': '', 'history': '-config', 'initialize': '-config'}.get(rkey, '-' + rkey) e = self._api.read_namespaced_endpoints(ep, self._namespace) - if key != 'sync': + if key not in ('sync', 'status', 'failsafe'): return e.metadata.annotations[key] else: return json.dumps(e.metadata.annotations) diff --git a/features/ignored_slots.feature b/features/ignored_slots.feature index e0c53ea3..4e83570d 100644 --- a/features/ignored_slots.feature +++ b/features/ignored_slots.feature @@ -50,7 +50,7 @@ Feature: ignored slots And postgres1 has a logical replication slot named unmanaged_slot_1 with the test_decoding plugin after 2 seconds And postgres1 has a logical replication slot named unmanaged_slot_2 with the test_decoding plugin after 2 seconds And postgres1 has a logical replication slot named unmanaged_slot_3 with the test_decoding plugin after 2 seconds - And postgres1 does not have a logical replication slot named dummy_slot + And postgres1 does not have a replication slot named dummy_slot # 3. After a failover the server (now a primary) still has the slot. When I shut down postgres0 diff --git a/features/permanent_slots.feature b/features/permanent_slots.feature index 656e6ade..2928e829 100644 --- a/features/permanent_slots.feature +++ b/features/permanent_slots.feature @@ -3,32 +3,73 @@ Feature: permanent slots Given I start postgres0 Then postgres0 is a leader after 10 seconds And there is a non empty initialize key in DCS after 15 seconds - When I issue a PATCH request to http://127.0.0.1:8008/config with {"slots": {"test_physical": {"type": "physical"}}, "postgresql": {"parameters": {"wal_level": "logical"}}} + When I issue a PATCH request to http://127.0.0.1:8008/config with {"slots":{"test_physical":0,"postgres0":0,"postgres1":0,"postgres3":0},"postgresql":{"parameters":{"wal_level":"logical"}}} Then I receive a response code 200 And Response on GET http://127.0.0.1:8008/config contains slots after 10 seconds + When I start postgres1 + And I start postgres2 + And I configure and start postgres3 with a tag replicatefrom postgres2 Then postgres0 has a physical replication slot named test_physical after 10 seconds - And I start postgres1 + And postgres0 has a physical replication slot named postgres1 after 10 seconds + And postgres0 has a physical replication slot named postgres2 after 10 seconds + And postgres2 has a physical replication slot named postgres3 after 10 seconds @slot-advance Scenario: check that logical permanent slots are created Given I run patronictl.py restart batman postgres0 --force - And I issue a PATCH request to http://127.0.0.1:8008/config with {"slots": {"test_logical": {"type": "logical", "database": "postgres", "plugin": "test_decoding"}}} + And I issue a PATCH request to http://127.0.0.1:8008/config with {"slots":{"test_logical":{"type":"logical","database":"postgres","plugin":"test_decoding"}}} Then postgres0 has a logical replication slot named test_logical with the test_decoding plugin after 10 seconds @slot-advance - Scenario: check that permanent slots are created on the replica + Scenario: check that permanent slots are created on replicas Given postgres1 has a logical replication slot named test_logical with the test_decoding plugin after 10 seconds Then Logical slot test_logical is in sync between postgres0 and postgres1 after 10 seconds + And Logical slot test_logical is in sync between postgres0 and postgres2 after 10 seconds + And Logical slot test_logical is in sync between postgres0 and postgres3 after 10 seconds And postgres1 has a physical replication slot named test_physical after 2 seconds + And postgres2 has a physical replication slot named test_physical after 2 seconds + And postgres3 has a physical replication slot named test_physical after 2 seconds @slot-advance - Scenario: check that permanent slots are advanced on the replica + Scenario: check permanent physical slots that match with member names + Given postgres0 has a physical replication slot named postgres3 after 2 seconds + And postgres1 has a physical replication slot named postgres0 after 2 seconds + And postgres1 has a physical replication slot named postgres3 after 2 seconds + And postgres2 has a physical replication slot named postgres0 after 2 seconds + And postgres2 has a physical replication slot named postgres3 after 2 seconds + And postgres2 has a physical replication slot named postgres1 after 2 seconds + And postgres1 does not have a replication slot named postgres2 + And postgres3 does not have a replication slot named postgres2 + + @slot-advance + Scenario: check that permanent slots are advanced on replicas Given I add the table replicate_me to postgres0 - And I get all changes from physical slot test_physical on postgres0 When I get all changes from logical slot test_logical on postgres0 + And I get all changes from physical slot test_physical on postgres0 Then Logical slot test_logical is in sync between postgres0 and postgres1 after 10 seconds And Physical slot test_physical is in sync between postgres0 and postgres1 after 10 seconds + And Logical slot test_logical is in sync between postgres0 and postgres2 after 10 seconds + And Physical slot test_physical is in sync between postgres0 and postgres2 after 10 seconds + And Logical slot test_logical is in sync between postgres0 and postgres3 after 10 seconds + And Physical slot test_physical is in sync between postgres0 and postgres3 after 10 seconds + And Physical slot postgres1 is in sync between postgres0 and postgres2 after 10 seconds + And Physical slot postgres3 is in sync between postgres2 and postgres0 after 20 seconds + And Physical slot postgres3 is in sync between postgres2 and postgres1 after 10 seconds + And postgres1 does not have a replication slot named postgres2 + And postgres3 does not have a replication slot named postgres2 + + @slot-advance + Scenario: check that only permanent slots are written to the /status key + Given "status" key in DCS has test_physical in slots + And "status" key in DCS has postgres0 in slots + And "status" key in DCS has postgres1 in slots + And "status" key in DCS does not have postgres2 in slots + And "status" key in DCS has postgres3 in slots Scenario: check permanent physical replication slot after failover - Given I shut down postgres0 + Given I shut down postgres3 + And I shut down postgres2 + And I shut down postgres0 Then postgres1 has a physical replication slot named test_physical after 10 seconds + And postgres1 has a physical replication slot named postgres0 after 10 seconds + And postgres1 has a physical replication slot named postgres3 after 10 seconds diff --git a/features/standby_cluster.feature b/features/standby_cluster.feature index a9f00c01..97c27203 100644 --- a/features/standby_cluster.feature +++ b/features/standby_cluster.feature @@ -51,7 +51,7 @@ Feature: standby cluster When I issue a GET request to http://127.0.0.1:8010/patroni Then I receive a response code 200 And I receive a response replication_state streaming - And postgres1 does not have a logical replication slot named test_logical + And postgres1 does not have a replication slot named test_logical Scenario: check switchover Given I run patronictl.py switchover batman1 --force diff --git a/features/steps/basic_replication.py b/features/steps/basic_replication.py index 5977eb61..7a687e28 100644 --- a/features/steps/basic_replication.py +++ b/features/steps/basic_replication.py @@ -110,7 +110,7 @@ def replication_works(context, primary, replica, time_limit): context.execute_steps(u""" When I add the table test_{0} to {1} Then table test_{0} is present on {2} after {3} seconds - """.format(int(time()), primary, replica, time_limit)) + """.format(str(time()).replace('.', '_').replace(',', '_'), primary, replica, time_limit)) @then('there is a "{message}" {level:w} in the {node} patroni log') diff --git a/features/steps/slots.py b/features/steps/slots.py index f4a3cfa5..182aa87c 100644 --- a/features/steps/slots.py +++ b/features/steps/slots.py @@ -1,3 +1,4 @@ +import json import time from behave import step, then @@ -36,8 +37,9 @@ def has_logical_replication_slot(context, pg_name, slot_name, plugin, time_limit assert False, f"Error looking for slot {slot_name} on {pg_name} with plugin {plugin}" -@then('{pg_name:w} does not have a logical replication slot named {slot_name}') -def does_not_have_logical_replication_slot(context, pg_name, slot_name): +@step('{pg_name:w} does not have a replication slot named {slot_name:w}') +@then('{pg_name:w} does not have a replication slot named {slot_name:w}') +def does_not_have_replication_slot(context, pg_name, slot_name): try: row = context.pctl.query(pg_name, ("SELECT 1 FROM pg_replication_slots" " WHERE slot_name = '{0}'").format(slot_name)).fetchone() @@ -89,3 +91,15 @@ def has_physical_replication_slot(context, pg_name, slot_name, time_limit): pass time.sleep(1) assert False, f"Physical slot {slot_name} doesn't exist after {time_limit} seconds" + + +@step('"{name}" key in DCS has {subkey:w} in {key:w}') +def dcs_key_contains(context, name, subkey, key): + response = json.loads(context.dcs_ctl.query(name)) + assert key in response and subkey in response[key], f"{name} key in DCS doesn't have {subkey} in {key}" + + +@step('"{name}" key in DCS does not have {subkey:w} in {key:w}') +def dcs_key_does_not_contain(context, name, subkey, key): + response = json.loads(context.dcs_ctl.query(name)) + assert key not in response or subkey not in response[key], f"{name} key in DCS has {subkey} in {key}" diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index 1a516cca..bc74bbe0 100644 --- a/patroni/dcs/__init__.py +++ b/patroni/dcs/__init__.py @@ -24,6 +24,7 @@ import dateutil.parser from ..exceptions import PatroniFatalException from ..utils import deep_compare, uri from ..tags import Tags +from ..utils import parse_int if TYPE_CHECKING: # pragma: no cover from ..config import Config @@ -354,7 +355,7 @@ class Member(Tags, NamedTuple('Member', @property def lsn(self) -> Optional[int]: """Current LSN (receive/flush/replay).""" - return self.data.get('xlog_location') + return parse_int(self.data.get('xlog_location')) class RemoteMember(Member): @@ -974,29 +975,41 @@ class Cluster(NamedTuple('Cluster', def is_physical_slot(value: Union[Any, Dict[str, Any]]) -> bool: """Check whether provided configuration is for permanent physical replication slot. - :returns: ``True`` if this is a physical replication slot, otherwise ``False``. + :param value: configuration of the permanent replication slot. + + :returns: ``True`` if *value* is a physical replication slot, otherwise ``False``. """ return not value or isinstance(value, dict) and value.get('type', 'physical') == 'physical' + @staticmethod + def is_logical_slot(value: Union[Any, Dict[str, Any]]) -> bool: + """Check whether provided configuration is for permanent logical replication slot. + + :param value: configuration of the permanent replication slot. + + :returns: ``True`` if *value* is a logical replication slot, otherwise ``False``. + """ + return isinstance(value, dict) \ + and value.get('type', 'logical') == 'logical' \ + and bool(value.get('database') and value.get('plugin')) + @property def __permanent_slots(self) -> Dict[str, Union[Dict[str, Any], Any]]: """Dictionary of permanent replication slots with their known LSN.""" - leader = self.leader and self.leader.member - leader_name = slot_name_from_member_name(leader.name) if leader and leader.lsn else None - - slots = self.slots or {} ret: Dict[str, Union[Dict[str, Any], Any]] = deepcopy(self.config.permanent_slots if self.config else {}) + members: Dict[str, int] = {slot_name_from_member_name(m.name): m.lsn or 0 for m in self.members} + slots: Dict[str, int] = {k: parse_int(v) or 0 for k, v in (self.slots or {}).items()} for name, value in list(ret.items()): if not value: value = ret[name] = {} if isinstance(value, dict): - if name in slots: - # If primary reported flush LSN for permanent slots we want to enrich our structure with it - value['lsn'] = slots[name] - elif self.is_physical_slot(value) and name == leader_name and leader and leader.lsn: - # there is no slot on the leader for itself, use `lsn` from the member key. - value['lsn'] = leader.lsn + # for permanent physical slots we want to get MAX LSN from the `Cluster.slots` and from the + # member with the matching name. It is necessary because we may have the replication slot on + # the primary that is streaming from the other standby node using the `replicatefrom` tag. + lsn = max(members.get(name, 0) if self.is_physical_slot(value) else 0, slots.get(name, 0)) + if lsn: + value['lsn'] = lsn else: # Don't let anyone set 'lsn' in the global configuration :) value.pop('lsn', None) @@ -1010,8 +1023,7 @@ class Cluster(NamedTuple('Cluster', @property def __permanent_logical_slots(self) -> Dict[str, Any]: """Dictionary of permanent ``logical`` replication slots.""" - return {name: value for name, value in self.__permanent_slots.items() if isinstance(value, dict) - and value.get('type', 'logical') == 'logical' and value.get('database') and value.get('plugin')} + return {name: value for name, value in self.__permanent_slots.items() if self.is_logical_slot(value)} @property def use_slots(self) -> bool: @@ -1037,7 +1049,9 @@ class Cluster(NamedTuple('Cluster', :returns: final dictionary of slot names, after merging with permanent slots and performing sanity checks. """ slots: Dict[str, Dict[str, str]] = self._get_members_slots(my_name, role) - permanent_slots: Dict[str, Any] = self._get_permanent_slots(is_standby_cluster, role, nofailover, major_version) + permanent_slots: Dict[str, Any] = self._get_permanent_slots(is_standby_cluster=is_standby_cluster, + role=role, nofailover=nofailover, + major_version=major_version) disabled_permanent_logical_slots: List[str] = self._merge_permanent_slots( slots, permanent_slots, my_name, major_version) @@ -1048,8 +1062,7 @@ class Cluster(NamedTuple('Cluster', return slots - @staticmethod - def _merge_permanent_slots(slots: Dict[str, Dict[str, str]], permanent_slots: Dict[str, Any], my_name: str, + def _merge_permanent_slots(self, slots: Dict[str, Dict[str, str]], permanent_slots: Dict[str, Any], my_name: str, major_version: int) -> List[str]: """Merge replication *slots* for members with *permanent_slots*. @@ -1084,7 +1097,7 @@ class Cluster(NamedTuple('Cluster', slots[name] = value continue - if value['type'] == 'logical' and value.get('database') and value.get('plugin'): + if self.is_logical_slot(value): if major_version < SLOT_ADVANCE_AVAILABLE_VERSION: disabled_permanent_logical_slots.append(name) elif name in slots: @@ -1097,7 +1110,7 @@ class Cluster(NamedTuple('Cluster', logger.error("Bad value for slot '%s' in permanent_slots: %s", name, permanent_slots[name]) return disabled_permanent_logical_slots - def _get_permanent_slots(self, is_standby_cluster: bool, role: str, + def _get_permanent_slots(self, *, is_standby_cluster: bool, role: str, nofailover: bool, major_version: int) -> Dict[str, Any]: """Get configured permanent replication slots. @@ -1171,20 +1184,50 @@ class Cluster(NamedTuple('Cluster', for k, v in slot_conflicts.items() if len(v) > 1)) return slots - def has_permanent_slots(self, my_name: str, nofailover: bool = False) -> bool: + def has_permanent_slots(self, my_name: str, *, is_standby_cluster: bool = False, nofailover: bool = False, + major_version: int = SLOT_ADVANCE_AVAILABLE_VERSION) -> bool: """Check if the given member node has permanent replication slots configured. :param my_name: name of the member node to check. + :param is_standby_cluster: ``True`` if it is known that this is a standby cluster. We pass the value from + the outside because we want to protect from the ``/config`` key removal. :param nofailover: ``True`` if this node is tagged to not be a failover candidate. + :param major_version: postgresql major version. :returns: ``True`` if there are permanent replication slots configured, otherwise ``False``. """ - members_slots: Dict[str, Dict[str, str]] = self._get_members_slots(my_name, 'replica') - permanent_slots: Dict[str, Any] = self._get_permanent_slots(nofailover, 'replica', False, - SLOT_ADVANCE_AVAILABLE_VERSION) + role = 'replica' + members_slots: Dict[str, Dict[str, str]] = self._get_members_slots(my_name, role) + permanent_slots: Dict[str, Any] = self._get_permanent_slots(is_standby_cluster=is_standby_cluster, + role=role, nofailover=nofailover, + major_version=major_version) slots = deepcopy(members_slots) - self._merge_permanent_slots(slots, permanent_slots, my_name, SLOT_ADVANCE_AVAILABLE_VERSION) - return len(slots) > len(members_slots) + self._merge_permanent_slots(slots, permanent_slots, my_name, major_version) + return len(slots) > len(members_slots) or any(self.is_physical_slot(v) for v in permanent_slots.values()) + + def filter_permanent_slots(self, slots: Dict[str, int], is_standby_cluster: bool, + major_version: int) -> Dict[str, int]: + """Filter out all non-permanent slots from provided *slots* dict. + + :param slots: slot names with LSN values + :param is_standby_cluster: ``True`` if it is known that this is a standby cluster. We pass the value from + the outside because we want to protect from the ``/config`` key removal. + :param major_version: postgresql major version. + + :returns: a :class:`dict` object that contains only slots that are known to be permanent. + """ + if major_version < SLOT_ADVANCE_AVAILABLE_VERSION: + return {} # for legacy PostgreSQL we don't support permanent slots on standby nodes + + permanent_slots: Dict[str, Any] = self._get_permanent_slots(is_standby_cluster=is_standby_cluster, + role='replica', + nofailover=False, + major_version=major_version) + members_slots = {slot_name_from_member_name(m.name) for m in self.members} + + return {name: value for name, value in slots.items() if name in permanent_slots + and (self.is_physical_slot(permanent_slots[name]) + or self.is_logical_slot(permanent_slots[name]) and name not in members_slots)} def _has_permanent_logical_slots(self, my_name: str, nofailover: bool) -> bool: """Check if the given member node has permanent ``logical`` replication slots configured. diff --git a/patroni/ha.py b/patroni/ha.py index 877e24ef..3b00beb9 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -14,7 +14,7 @@ from . import psycopg from .__main__ import Patroni from .async_executor import AsyncExecutor, CriticalTask from .collections import CaseInsensitiveSet -from .dcs import AbstractDCS, Cluster, Leader, Member, RemoteMember, Status +from .dcs import AbstractDCS, Cluster, Leader, Member, RemoteMember, Status, slot_name_from_member_name from .exceptions import DCSError, PostgresConnectionException, PatroniFatalException from .postgresql.callback_executor import CallbackAction from .postgresql.misc import postgres_version_to_int @@ -272,12 +272,32 @@ class Ha(object): ret[self.state_handler.name] = self.patroni.api.connection_string return ret - def update_lock(self, write_leader_optime: bool = False) -> bool: + def update_lock(self, update_status: bool = False) -> bool: + """Update the leader lock in DCS. + + .. note:: + After successful update of the leader key the :meth:`AbstractDCS.update_leader` method could also + optionally update the ``/status`` and ``/failsafe`` keys. + + The ``/status`` key contains the last known LSN on the leader node and the last known state + of permanent replication slots including permanent physical replication slot for the leader. + + Last, but not least, this method calls a :meth:`Watchdog.keepalive` method after the leader key + was successfully updated. + + :param update_status: ``True`` if we also need to update the ``/status`` key in DCS, otherwise ``False``. + + :returns: ``True`` if the leader key was successfully updated and we can continue to run postgres + as a ``primary`` or as a ``standby_leader``, otherwise ``False``. + """ last_lsn = slots = None - if write_leader_optime: + if update_status: try: last_lsn = self.state_handler.last_operation() - slots = self.state_handler.slots() + slots = self.cluster.filter_permanent_slots( + {**self.state_handler.slots(), slot_name_from_member_name(self.state_handler.name): last_lsn}, + self.is_standby_cluster(), + self.state_handler.major_version) except Exception: logger.exception('Exception when called state_handler.last_operation()') if TYPE_CHECKING: # pragma: no cover @@ -900,6 +920,26 @@ class Ha(object): return False def check_failsafe_topology(self) -> bool: + """Check whether we could continue to run as a primary by calling all members from the failsafe topology. + + .. note:: + If the ``/failsafe`` key contains invalid data or if the ``name`` of our node is missing in + the ``/failsafe`` key, we immediately give up and return ``False``. + + We send the JSON document in the POST request with the following fields: + + * ``name`` - the name of our node; + * ``conn_url`` - connection URL to the postgres, which is reachable from other nodes; + * ``api_url`` - connection URL to Patroni REST API on this node reachable from other nodes; + * ``slots`` - a :class:`dict` with replication slots that exist on the leader node, including the primary + itself with the last known LSN, because there could be a permanent physical slot on standby nodes. + + Standby nodes are using information from the ``slots`` dict to advance position of permanent + replication slots while DCS is not accessible in order to avoid indefinite growth of ``pg_wal``. + + :returns: ``True`` if all members from the ``/failsafe`` topology agree that this node could continue to + run as a ``primary``, or ``False`` if some of standby nodes are not accessible or don't agree. + """ failsafe = self.dcs.failsafe if not isinstance(failsafe, dict) or self.state_handler.name not in failsafe: return False @@ -909,7 +949,10 @@ class Ha(object): 'api_url': self.patroni.api.connection_string, } try: - data['slots'] = self.state_handler.slots() + data['slots'] = { + **self.state_handler.slots(), + slot_name_from_member_name(self.state_handler.name): self.state_handler.last_operation() + } except Exception: logger.exception('Exception when called state_handler.slots()') members = [RemoteMember(name, {'api_url': url}) diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py index cab33e8f..7a3d4065 100644 --- a/patroni/postgresql/__init__.py +++ b/patroni/postgresql/__init__.py @@ -451,15 +451,21 @@ class Postgresql(object): return if self._global_config.is_standby_cluster: - self._has_permanent_slots = False # Standby cluster can't have logical replication slots, and we don't need to enforce hot_standby_feedback self.set_enforce_hot_standby_feedback(False) - elif cluster and cluster.config and cluster.config.modify_version: - self._has_permanent_slots = cluster.has_permanent_slots(self.name, nofailover) + + if cluster and cluster.config and cluster.config.modify_version: # We want to enable hot_standby_feedback if the replica is supposed # to have a logical slot or in case if it is the cascading replica. - self.set_enforce_hot_standby_feedback( - self.can_advance_slots and cluster.should_enforce_hot_standby_feedback(self.name, nofailover)) + self.set_enforce_hot_standby_feedback(not self._global_config.is_standby_cluster and self.can_advance_slots + and cluster.should_enforce_hot_standby_feedback(self.name, + nofailover)) + + self._has_permanent_slots = cluster.has_permanent_slots( + my_name=self.name, + is_standby_cluster=self._global_config.is_standby_cluster, + nofailover=nofailover, + major_version=self.major_version) def _cluster_info_state_get(self, name: str) -> Optional[Any]: if not self._cluster_info_state: diff --git a/patroni/postgresql/slots.py b/patroni/postgresql/slots.py index 29bbb130..48b275e4 100644 --- a/patroni/postgresql/slots.py +++ b/patroni/postgresql/slots.py @@ -16,7 +16,6 @@ from .misc import format_lsn, fsync_dir from ..dcs import Cluster, Leader from ..file_perm import pg_perm from ..psycopg import OperationalError -from ..utils import parse_int if TYPE_CHECKING: # pragma: no cover from psycopg import Cursor @@ -378,10 +377,9 @@ class SlotsHandler: except Exception: logger.exception("Failed to create physical replication slot '%s'", name) self._schedule_load_slots = True - elif not self._postgresql.is_primary() and self._postgresql.can_advance_slots \ - and self._replication_slots[name]['type'] == 'physical': + elif self._postgresql.can_advance_slots and self._replication_slots[name]['type'] == 'physical': value['restart_lsn'] = self._replication_slots[name]['restart_lsn'] - lsn = parse_int(value.get('lsn')) + lsn = value.get('lsn') if lsn and lsn > value['restart_lsn']: # The slot has feedback in DCS and needs to be advanced try: lsn = format_lsn(lsn) @@ -477,12 +475,9 @@ class SlotsHandler: # If the logical already exists, copy some information about it into the original structure if name in self._replication_slots and compare_slots(value, self._replication_slots[name]): self._copy_items(self._replication_slots[name], value) - if 'lsn' in value: # The slot has feedback in DCS - try: # Skip slots that don't need to be advanced - if value['confirmed_flush_lsn'] < int(value['lsn']): - advance_slots[value['database']][name] = int(value['lsn']) - except Exception as e: - logger.error('Failed to parse "%s": %r', value['lsn'], e) + if 'lsn' in value and value['confirmed_flush_lsn'] < value['lsn']: # The slot has feedback in DCS + # Skip slots that don't need to be advanced + advance_slots[value['database']][name] = value['lsn'] elif name not in self._replication_slots and 'lsn' in value: # We want to copy only slots with feedback in a DCS create_slots.append(name) diff --git a/patroni/utils.py b/patroni/utils.py index be468d2e..6957369f 100644 --- a/patroni/utils.py +++ b/patroni/utils.py @@ -819,7 +819,7 @@ def cluster_as_json(cluster: 'Cluster', global_config: Optional['GlobalConfig'] member.update({n: m.data[n] for n in optional_attributes if n in m.data}) if m.name != leader_name: - lsn = m.data.get('xlog_location') + lsn = m.lsn if lsn is None: member['lag'] = 'unknown' elif cluster_lsn >= lsn: diff --git a/tests/test_ha.py b/tests/test_ha.py index ea8fa7c0..eeea68b6 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -167,6 +167,7 @@ def run_async(self, func, args=()): @patch.object(Postgresql, 'is_primary', Mock(return_value=True)) @patch.object(Postgresql, 'timeline_wal_position', Mock(return_value=(1, 10, 1))) @patch.object(Postgresql, '_cluster_info_state_get', Mock(return_value=10)) +@patch.object(Postgresql, 'slots', Mock(return_value={'l': 100})) @patch.object(Postgresql, 'data_directory_empty', Mock(return_value=False)) @patch.object(Postgresql, 'controldata', Mock(return_value={ 'Database system identifier': SYSID, @@ -1582,6 +1583,7 @@ class TestHa(PostgresInit): @patch('patroni.psycopg.connect', psycopg_connect) def test_permanent_logical_slots_after_promote(self): + self.p._major_version = 110000 config = ClusterConfig(1, {'slots': {'l': {'database': 'postgres', 'plugin': 'test_decoding'}}}, 1) self.p.name = 'other' self.ha.cluster = get_cluster_initialized_without_leader(cluster_config=config) diff --git a/tests/test_slots.py b/tests/test_slots.py index d1b8d458..83087c4e 100644 --- a/tests/test_slots.py +++ b/tests/test_slots.py @@ -128,6 +128,7 @@ class TestSlotsHandler(BaseTestPostgresql): self.cluster.slots['ls'] = 'a' self.assertEqual(self.s.sync_replication_slots(self.cluster, False), []) self.cluster.config.data['slots']['ls']['database'] = 'b' + self.cluster.slots['ls'] = '500' with patch.object(MockCursor, 'rowcount', PropertyMock(return_value=1), create=True): self.assertEqual(self.s.sync_replication_slots(self.cluster, False), ['ls'])