mirror of
https://github.com/outbackdingo/patroni.git
synced 2026-08-25 14:53:37 +00:00
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`
This commit is contained in:
Alexander Kukushkin
@@ -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 <existing_data>` 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 <existing_data>` 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.
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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')
|
||||
|
||||
+16
-2
@@ -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}"
|
||||
|
||||
+68
-25
@@ -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.
|
||||
|
||||
+48
-5
@@ -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})
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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)
|
||||
|
||||
+1
-1
@@ -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:
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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'])
|
||||
|
||||
|
||||
Reference in New Issue
Block a user