mirror of
https://github.com/outbackdingo/patroni.git
synced 2026-08-25 14:53:37 +00:00
Fix bug with slot for former leader not retained on failover (#3261)
the problem existed because _build_retain_slots() method was falsely relying on members being present in DCS, while on failover the member key for the former leader is expiring exactly at the same time.
This commit is contained in:
@@ -154,7 +154,7 @@ jobs:
|
|||||||
with:
|
with:
|
||||||
name: behave-${{ matrix.os }}-${{ matrix.dcs }}-${{ matrix.python-version }}-logs
|
name: behave-${{ matrix.os }}-${{ matrix.dcs }}-${{ matrix.python-version }}-logs
|
||||||
path: |
|
path: |
|
||||||
features/output/*_failed/*postgres?.*
|
features/output/*_failed/*postgres-?.*
|
||||||
features/output/*.log
|
features/output/*.log
|
||||||
if-no-files-found: error
|
if-no-files-found: error
|
||||||
retention-days: 5
|
retention-days: 5
|
||||||
|
|||||||
@@ -81,4 +81,6 @@ Feature: permanent slots
|
|||||||
When I start postgres-0
|
When I start postgres-0
|
||||||
Then postgres-0 role is the replica after 20 seconds
|
Then postgres-0 role is the replica after 20 seconds
|
||||||
And physical replication slot named postgres_1 on postgres-0 has no xmin value after 10 seconds
|
And physical replication slot named postgres_1 on postgres-0 has no xmin value after 10 seconds
|
||||||
And physical replication slot named postgres_2 on postgres-0 has no xmin value after 10 seconds
|
# postgres_2 and postgres_3 slots are retained, but postgres_2 will still have xmin value :(
|
||||||
|
And postgres-0 has a physical replication slot named postgres_2 after 10 seconds
|
||||||
|
And postgres-0 has a physical replication slot named postgres_3 after 10 seconds
|
||||||
|
|||||||
@@ -99,8 +99,8 @@ def has_physical_replication_slot(context, pg_name, slot_name, time_limit):
|
|||||||
def physical_slot_no_xmin(context, pg_name, slot_name, time_limit):
|
def physical_slot_no_xmin(context, pg_name, slot_name, time_limit):
|
||||||
time_limit *= context.timeout_multiplier
|
time_limit *= context.timeout_multiplier
|
||||||
max_time = time.time() + int(time_limit)
|
max_time = time.time() + int(time_limit)
|
||||||
query = "SELECT xmin FROM pg_catalog.pg_replication_slots WHERE slot_type = 'physical'"
|
query = "SELECT xmin FROM pg_catalog.pg_replication_slots WHERE slot_type = 'physical'" +\
|
||||||
f" AND slot_name = '{slot_name}'"
|
f" AND slot_name = '{slot_name}'"
|
||||||
exists = False
|
exists = False
|
||||||
while time.time() < max_time:
|
while time.time() < max_time:
|
||||||
try:
|
try:
|
||||||
|
|||||||
+17
-4
@@ -1841,10 +1841,6 @@ class AbstractDCS(abc.ABC):
|
|||||||
"""
|
"""
|
||||||
timestamp = time.time()
|
timestamp = time.time()
|
||||||
|
|
||||||
# DCS is a source of truth, therefore we take missing values from there
|
|
||||||
self._last_retain_slots.update({name: timestamp for name in self._last_status['retain_slots']
|
|
||||||
if (not slots or name not in slots) and name not in self._last_retain_slots})
|
|
||||||
|
|
||||||
if slots: # if slots is not empty it implies we are running v11+
|
if slots: # if slots is not empty it implies we are running v11+
|
||||||
members: Set[str] = set()
|
members: Set[str] = set()
|
||||||
found_self = False
|
found_self = False
|
||||||
@@ -1929,6 +1925,23 @@ class AbstractDCS(abc.ABC):
|
|||||||
:returns: ``True`` if key has been created successfully.
|
:returns: ``True`` if key has been created successfully.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
|
def acquire_leader_lock(self) -> bool:
|
||||||
|
"""Attempt to acquire leader lock.
|
||||||
|
|
||||||
|
.. note::
|
||||||
|
This method wraps :meth:`~AbstractDCS.attempt_to_acquire_leader`: and is
|
||||||
|
used to reset retention time of physical replication slots that representing
|
||||||
|
members of the cluster when current node is to be promoted to the leader.
|
||||||
|
|
||||||
|
:returns: ``True`` if the leader key has been created successfully.
|
||||||
|
"""
|
||||||
|
ret = self.attempt_to_acquire_leader()
|
||||||
|
if ret:
|
||||||
|
timestamp = time.time()
|
||||||
|
# every time we promote we need to reset retention time for slots recorded in the /status key
|
||||||
|
self._last_retain_slots = {name: timestamp for name in self._last_status['retain_slots']}
|
||||||
|
return ret
|
||||||
|
|
||||||
@abc.abstractmethod
|
@abc.abstractmethod
|
||||||
def set_failover_value(self, value: str, version: Optional[Any] = None) -> bool:
|
def set_failover_value(self, value: str, version: Optional[Any] = None) -> bool:
|
||||||
"""Create or update ``/failover`` key.
|
"""Create or update ``/failover`` key.
|
||||||
|
|||||||
+2
-2
@@ -345,11 +345,11 @@ class Ha(object):
|
|||||||
|
|
||||||
def acquire_lock(self) -> bool:
|
def acquire_lock(self) -> bool:
|
||||||
try:
|
try:
|
||||||
ret = self.dcs.attempt_to_acquire_leader()
|
ret = self.dcs.acquire_leader_lock()
|
||||||
except DCSError:
|
except DCSError:
|
||||||
raise
|
raise
|
||||||
except Exception:
|
except Exception:
|
||||||
logger.exception('Unexpected exception raised from attempt_to_acquire_leader, please report it as a BUG')
|
logger.exception('Unexpected exception raised from acquire_leader_lock, please report it as a BUG')
|
||||||
ret = False
|
ret = False
|
||||||
self.set_is_leader(ret)
|
self.set_is_leader(ret)
|
||||||
return ret
|
return ret
|
||||||
|
|||||||
Reference in New Issue
Block a user