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

Finish refactoring of the Status class (#3103)

Browse Source 
The `Status` class was introduced in #2853, but we kept old properties in the `Cluster` object in order to have fewer changes in the rest of the code.

This PR is finishing the refactoring.
The following adjustments were made:
- Introduced `Status.is_empty()` method, which is used in the `Cluster.is_empty()` instead of checking actual values to simplify introduction of further fields to the Status object.
- Removed `Cluster.last_lsn` property
- Changed `Cluster.slots` property to always return dict and perform sanity checks on values.

Besides that, this PR addressing a couple of problems:
- the `AbstractDCS.get_cluster()` method some properties without holding a lock on `_cluster_thread_lock`.
- `Cluster.__permanent_slots` property was setting 'lsn' from all cluster members, while it should be doing that only for members with `replicatefrom` tag.
This commit is contained in:
Alexander Kukushkin
2024-07-16 09:47:20 +02:00
committed by GitHub
parent fbbd32a537
commit a5796a03f1
7 changed files with 33 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
patroni/api.py
+1 -1
View File
@@ -295,7 +295,7 @@ class RestApiHandler(BaseHTTPRequestHandler):
cluster = patroni.dcs.cluster
config = global_config.from_cluster(cluster)
leader_optime = cluster and cluster.last_lsn or 0
leader_optime = cluster and cluster.status.last_lsn
replayed_location = response.get('xlog', {}).get('replayed_location', 0)
max_replica_lag = parse_int(self.path_query.get('lag', [sys.maxsize])[0], 'B')
if max_replica_lag is None:
patroni/dcs/__init__.py
+25 -17
View File
@@ -738,6 +738,13 @@ class Status(NamedTuple):
"""
return Status(0, None)
def is_empty(self):
"""Validate definition of all attributes of this :class:`Status` instance.
:returns: ``True`` if all attributes of the current :class:`Status` are unpopulated.
"""
return self.last_lsn == 0 and self.slots is None
@staticmethod
def from_node(value: Union[str, Dict[str, Any], None]) -> 'Status':
"""Factory method to parse *value* as :class:`Status` object.
@@ -814,14 +821,13 @@ class Cluster(NamedTuple('Cluster',
return super(Cluster, cls).__new__(cls, *args, **kwargs)
@property
def last_lsn(self) -> int:
"""Last known leader LSN."""
return self.status.last_lsn
def slots(self) -> Dict[str, int]:
"""State of permanent replication slots on the primary in the format: ``{"slot_name": int}``.
@property
def slots(self) -> Optional[Dict[str, int]]:
"""State of permanent replication slots on the primary in the format: ``{"slot_name": int}``."""
return self.status.slots
.. note::
We are trying to be foolproof here and for values that can't be parsed to :class:`int` will return ``0``.
"""
return {k: parse_int(v) or 0 for k, v in (self.status.slots or {}).items()}
@staticmethod
def empty() -> 'Cluster':
@@ -833,9 +839,9 @@ class Cluster(NamedTuple('Cluster',
:returns: ``True`` if all attributes of the current :class:`Cluster` are unpopulated.
"""
return all((self.initialize is None, self.config is None, self.leader is None, self.last_lsn == 0,
return all((self.initialize is None, self.config is None, self.leader is None, self.status.is_empty(),
self.members == [], self.failover is None, self.sync.version is None,
self.history is None, self.slots is None, self.failsafe is None, self.workers == {}))
self.history is None, self.failsafe is None, self.workers == {}))
def __len__(self) -> int:
"""Implement ``len`` function capability.
@@ -934,8 +940,9 @@ class Cluster(NamedTuple('Cluster',
"""Dictionary of permanent replication slots with their known LSN."""
ret: Dict[str, Union[Dict[str, Any], Any]] = global_config.permanent_slots
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()}
members: Dict[str, int] = {slot_name_from_member_name(m.name): m.lsn or 0
for m in self.members if m.replicatefrom}
slots: Dict[str, int] = self.slots
for name, value in list(ret.items()):
if not value:
value = ret[name] = {}
@@ -1600,15 +1607,16 @@ class AbstractDCS(abc.ABC):
self.reset_cluster()
raise
self._last_seen = int(time.time())
self._last_status = {self._OPTIME: cluster.last_lsn}
if cluster.slots:
self._last_status['slots'] = cluster.slots
self._last_failsafe = cluster.failsafe
with self._cluster_thread_lock:
self._cluster = cluster
self._cluster_valid_till = time.time() + self.ttl
self._last_seen = int(time.time())
self._last_status = {self._OPTIME: cluster.status.last_lsn}
if cluster.status.slots:
self._last_status['slots'] = cluster.status.slots
self._last_failsafe = cluster.failsafe
return cluster
@property
patroni/ha.py
+1 -1
View File
@@ -993,7 +993,7 @@ class Ha(object):
:returns True when node is lagging
"""
lag = (self.cluster.last_lsn or 0) - wal_position
lag = self.cluster.status.last_lsn - wal_position
return lag > global_config.maximum_lag_on_failover
def _is_healthiest_node(self, members: Collection[Member], check_replication_lag: bool = True) -> bool:
patroni/utils.py
+1 -1
View File
@@ -946,7 +946,7 @@ def cluster_as_json(cluster: 'Cluster') -> Dict[str, Any]:
config = global_config.from_cluster(cluster)
leader_name = cluster.leader.name if cluster.leader else None
cluster_lsn = cluster.last_lsn or 0
cluster_lsn = cluster.status.last_lsn
ret: Dict[str, Any] = {'members': []}
for m in cluster.members:
tests/test_api.py
+1 -1
View File
@@ -204,7 +204,7 @@ class TestRestApiHandler(unittest.TestCase):
def test_do_GET(self):
MockPostgresql.pending_restart_reason = {'max_connections': get_param_diff('200', '100')}
MockPatroni.dcs.cluster.last_lsn = 20
MockPatroni.dcs.cluster.status.last_lsn = 20
MockPatroni.dcs.cluster.sync.members = [MockPostgresql.name]
with patch.object(global_config.__class__, 'is_synchronous_mode', PropertyMock(return_value=True)):
MockRestApiServer(RestApiHandler, 'GET /replica')
tests/test_slots.py
+3 -3
View File
@@ -213,7 +213,7 @@ class TestSlotsHandler(BaseTestPostgresql):
@patch.object(Postgresql, 'is_primary', Mock(return_value=False))
def test__ensure_logical_slots_replica(self):
self.p.set_role('replica')
self.cluster.slots['ls'] = 12346
self.cluster.status.slots['ls'] = 12346
with patch.object(SlotsHandler, 'check_logical_slots_readiness', Mock(return_value=False)):
self.assertEqual(self.s.sync_replication_slots(self.cluster, self.tags), [])
with patch.object(SlotsHandler, '_query', Mock(return_value=[('ls', 'logical', 499, 'b', 'a', 5, 100, 500)])), \
@@ -222,10 +222,10 @@ class TestSlotsHandler(BaseTestPostgresql):
patch.object(psycopg.OperationalError, 'diag') as mock_diag:
type(mock_diag).sqlstate = PropertyMock(return_value='58P01')
self.assertEqual(self.s.sync_replication_slots(self.cluster, self.tags), ['ls'])
self.cluster.slots['ls'] = 'a'
self.cluster.status.slots['ls'] = 'a'
self.assertEqual(self.s.sync_replication_slots(self.cluster, self.tags), [])
self.cluster.config.data['slots']['ls']['database'] = 'b'
self.cluster.slots['ls'] = '500'
self.cluster.status.slots['ls'] = '500'
with patch.object(MockCursor, 'rowcount', PropertyMock(return_value=1), create=True):
self.assertEqual(self.s.sync_replication_slots(self.cluster, self.tags), ['ls'])
tests/test_zookeeper.py
+1 -1
View File
@@ -176,7 +176,7 @@ class TestZooKeeper(unittest.TestCase):
def test_get_cluster(self):
cluster = self.zk.get_cluster()
self.assertEqual(cluster.last_lsn, 500)
self.assertEqual(cluster.status.last_lsn, 500)
def test__get_citus_cluster(self):
self.zk._mpp = get_mpp({'citus': {'group': 0, 'database': 'postgres'}})