gitops/patroni
1
0
Fork 0
mirror of https://github.com/outbackdingo/patroni.git synced 2026-08-26 15:40:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'master' of github.com:zalando/patroni into feature/citus-secondaries

Browse Source
This commit is contained in:
Alexander Kukushkin
2023-08-23 09:36:34 +02:00
parent 36bb077964 3333e78500
commit d5063bd3d7
15 changed files with 364 additions and 130 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+5
View File
@@ -48,10 +48,12 @@ pgpass
scm-source.json
# Sphinx-generated documentation
docs/_build/
docs/build/
docs/source/_static/
docs/source/_templates/
docs/modules/
docs/pdf/
# Pycharm IDE
.idea/
@@ -64,3 +66,6 @@ venv*/
# Default test data directory
data/
# macOS
**/.DS_Store
docs/.DS_Store
Vendored
BIN
View File
Binary file not shown.
docs/rest_api.rst
+185 -20
View File
@@ -92,26 +92,184 @@ Monitoring endpoint
The ``GET /patroni`` is used by Patroni during the leader race. It also could be used by your monitoring system. The JSON document produced by this endpoint has the same structure as the JSON produced by the health check endpoints.
**Example:** A healthy cluster
.. code-block:: bash
$ curl -s http://localhost:8008/patroni | jq .
{
"state": "running",
"postmaster_start_time": "2019-09-24 09:22:32.555 CEST",
"postmaster_start_time": "2023-08-18 11:03:37.966359+00:00",
"role": "master",
"server_version": 110005,
"cluster_unlocked": false,
"server_version": 150004,
"xlog": {
"location": 25624640
"location": 67395656
},
"timeline": 3,
"database_system_identifier": "6739877027151648096",
"timeline": 1,
"replication": [
{
"usename": "replicator",
"application_name": "patroni2",
"client_addr": "10.89.0.6",
"state": "streaming",
"sync_state": "async",
"sync_priority": 0
},
{
"usename": "replicator",
"application_name": "patroni3",
"client_addr": "10.89.0.2",
"state": "streaming",
"sync_state": "async",
"sync_priority": 0
}
],
"dcs_last_seen": 1692356718,
"tags": {
"clonefrom": true
},
"database_system_identifier": "7268616322854375442",
"patroni": {
"version": "1.6.0",
"scope": "batman"
"version": "3.1.0",
"scope": "demo"
}
}
**Example:** An unlocked cluster
.. code-block:: bash
$ curl -s http://localhost:8008/patroni | jq .
{
"state": "running",
"postmaster_start_time": "2023-08-18 11:09:08.615242+00:00",
"role": "replica",
"server_version": 150004,
"xlog": {
"received_location": 67419744,
"replayed_location": 67419744,
"replayed_timestamp": null,
"paused": false
},
"timeline": 1,
"replication": [
{
"usename": "replicator",
"application_name": "patroni2",
"client_addr": "10.89.0.6",
"state": "streaming",
"sync_state": "async",
"sync_priority": 0
},
{
"usename": "replicator",
"application_name": "patroni3",
"client_addr": "10.89.0.2",
"state": "streaming",
"sync_state": "async",
"sync_priority": 0
}
],
"cluster_unlocked": true,
"dcs_last_seen": 1692356928,
"tags": {
"clonefrom": true
},
"database_system_identifier": "7268616322854375442",
"patroni": {
"version": "3.1.0",
"scope": "demo"
}
}
**Example:** An unlocked cluster with :ref:`DCS failsafe mode <dcs_failsafe_mode>` enabled
.. code-block:: bash
$ curl -s http://localhost:8008/patroni | jq .
{
"state": "running",
"postmaster_start_time": "2023-08-18 11:09:08.615242+00:00",
"role": "replica",
"server_version": 150004,
"xlog": {
"location": 67420024
},
"timeline": 1,
"replication": [
{
"usename": "replicator",
"application_name": "patroni2",
"client_addr": "10.89.0.6",
"state": "streaming",
"sync_state": "async",
"sync_priority": 0
},
{
"usename": "replicator",
"application_name": "patroni3",
"client_addr": "10.89.0.2",
"state": "streaming",
"sync_state": "async",
"sync_priority": 0
}
],
"cluster_unlocked": true,
"failsafe_mode_is_active": true,
"dcs_last_seen": 1692356928,
"tags": {
"clonefrom": true
},
"database_system_identifier": "7268616322854375442",
"patroni": {
"version": "3.1.0",
"scope": "demo"
}
}
**Example:** A cluster with the :ref:`pause mode <pause>` enabled
.. code-block:: bash
$ curl -s http://localhost:8008/patroni | jq .
{
"state": "running",
"postmaster_start_time": "2023-08-18 11:09:08.615242+00:00",
"role": "replica",
"server_version": 150004,
"xlog": {
"location": 67420024
},
"timeline": 1,
"replication": [
{
"usename": "replicator",
"application_name": "patroni2",
"client_addr": "10.89.0.6",
"state": "streaming",
"sync_state": "async",
"sync_priority": 0
},
{
"usename": "replicator",
"application_name": "patroni3",
"client_addr": "10.89.0.2",
"state": "streaming",
"sync_state": "async",
"sync_priority": 0
}
],
"pause": true,
"dcs_last_seen": 1692356928,
"tags": {
"clonefrom": true
},
"database_system_identifier": "7268616322854375442",
"patroni": {
"version": "3.1.0",
"scope": "demo"
}
}
Retrieve the Patroni metrics in Prometheus format through the ``GET /metrics`` endpoint.
@@ -131,6 +289,9 @@ Retrieve the Patroni metrics in Prometheus format through the ``GET /metrics`` e
# HELP patroni_master Value is 1 if this node is the leader, 0 otherwise.
# TYPE patroni_master gauge
patroni_master{scope="batman"} 1
# HELP patroni_primary Value is 1 if this node is the leader, 0 otherwise.
# TYPE patroni_primary gauge
patroni_primary{scope="batman"} 1
# HELP patroni_xlog_location Current location of the Postgres transaction log, 0 if this node is not the leader.
# TYPE patroni_xlog_location counter
patroni_xlog_location{scope="batman"} 22320573386952
@@ -169,6 +330,9 @@ Retrieve the Patroni metrics in Prometheus format through the ``GET /metrics`` e
patroni_cluster_unlocked{scope="batman"} 0
# HELP patroni_postgres_timeline Postgres timeline of this node (if running), 0 otherwise.
# TYPE patroni_postgres_timeline counter
patroni_failsafe_mode_is_active{scope="batman"} 0
# HELP patroni_postgres_timeline Postgres timeline of this node (if running), 0 otherwise.
# TYPE patroni_postgres_timeline counter
patroni_postgres_timeline{scope="batman"} 24
# HELP patroni_dcs_last_seen Epoch timestamp when DCS was last contacted successfully by Patroni.
# TYPE patroni_dcs_last_seen gauge
@@ -192,24 +356,24 @@ Cluster status endpoints
{
"members": [
{
"name": "postgresql0",
"host": "127.0.0.1",
"port": 5432,
"name": "patroni1",
"role": "leader",
"state": "running",
"api_url": "http://127.0.0.1:8008/patroni",
"api_url": "http://10.89.0.4:8008/patroni",
"host": "10.89.0.4",
"port": 5432,
"timeline": 5,
"tags": {
"clonefrom": true
}
},
{
"name": "postgresql1",
"host": "127.0.0.1",
"port": 5433,
"name": "patroni2",
"role": "replica",
"state": "running",
"api_url": "http://127.0.0.1:8009/patroni",
"state": "streaming",
"api_url": "http://10.89.0.6:8008/patroni",
"host": "10.89.0.6",
"port": 5433,
"timeline": 5,
"tags": {
"clonefrom": true
@@ -218,8 +382,9 @@ Cluster status endpoints
}
],
"scheduled_switchover": {
"at": "2019-09-24T10:36:00+02:00",
"from": "postgresql0"
"at": "2023-09-24T10:36:00+02:00",
"from": "patroni1",
"to": "patroni3"
}
}
@@ -264,7 +429,7 @@ Config endpoint
.. code-block:: bash
$ curl -s localhost:8008/config | jq .
$ curl -s http://localhost:8008/config | jq .
{
"ttl": 30,
"loop_wait": 10,
patroni/__main__.py
+10 -36
View File
@@ -13,6 +13,7 @@ from argparse import Namespace
from typing import Any, Dict, Optional, TYPE_CHECKING
from patroni.daemon import AbstractPatroniDaemon, abstract_main, get_base_arg_parser
from patroni.tags import Tags
if TYPE_CHECKING: # pragma: no cover
from .config import Config
@@ -20,7 +21,7 @@ if TYPE_CHECKING: # pragma: no cover
logger = logging.getLogger(__name__)
class Patroni(AbstractPatroniDaemon):
class Patroni(AbstractPatroniDaemon, Tags):
"""Implement ``patroni`` command daemon.
:ivar version: Patroni version.
@@ -30,7 +31,6 @@ class Patroni(AbstractPatroniDaemon):
:ivar api: REST API server instance of this node.
:ivar request: wrapper for performing HTTP requests.
:ivar ha: HA handler.
:ivar tags: cache of custom tags configured for this node.
:ivar next_run: time when to run the next HA loop cycle.
:ivar scheduled_restart: when a restart has been scheduled to occur, if any. In that case, should contain two keys:
* ``schedule``: timestamp when restart should occur;
@@ -71,7 +71,7 @@ class Patroni(AbstractPatroniDaemon):
self.api = RestApiServer(self, self.config['restapi'])
self.ha = Ha(self)
self.tags = self.get_tags()
self._tags = self._get_tags()
self.next_run = time.time()
self.scheduled_restart: Dict[str, Any] = {}
@@ -121,33 +121,12 @@ class Patroni(AbstractPatroniDaemon):
except Exception:
return
def get_tags(self) -> Dict[str, Any]:
def _get_tags(self) -> Dict[str, Any]:
"""Get tags configured for this node, if any.
Handle both predefined Patroni tags and custom defined tags.
.. note::
A custom tag is any tag added to the configuration ``tags`` section that is not one of ``clonefrom``,
``nofailover``, ``noloadbalance`` or ``nosync``.
For the Patroni predefined tags, the returning object will only contain them if they are enabled as they
all are boolean values that default to disabled.
:returns: a dictionary of tags set for this node. The key is the tag name, and the value is the corresponding
tag value.
:returns: a dictionary of tags set for this node.
"""
return {tag: value for tag, value in self.config.get('tags', {}).items()
if tag not in ('clonefrom', 'nofailover', 'noloadbalance', 'nosync') or value}
@property
def nofailover(self) -> bool:
"""``True`` if ``tags.nofailover`` configuration is enabled for this node, else ``False``."""
return bool(self.tags.get('nofailover', False))
@property
def nosync(self) -> bool:
"""``True`` if ``tags.nosync`` configuration is enabled for this node, else ``False``."""
return bool(self.tags.get('nosync', False))
return self._filter_tags(self.config.get('tags', {}))
def reload_config(self, sighup: bool = False, local: Optional[bool] = False) -> None:
"""Apply new configuration values for ``patroni`` daemon.
@@ -166,7 +145,7 @@ class Patroni(AbstractPatroniDaemon):
try:
super(Patroni, self).reload_config(sighup, local)
if local:
self.tags = self.get_tags()
self._tags = self._get_tags()
self.request.reload_config(self.config)
if local or sighup and self.api.reload_local_certificate():
self.api.reload_config(self.config['restapi'])
@@ -177,14 +156,9 @@ class Patroni(AbstractPatroniDaemon):
logger.exception('Failed to reload config_file=%s', self.config.config_file)
@property
def replicatefrom(self) -> Optional[str]:
"""Value of ``tags.replicatefrom`` configuration, if any."""
return self.tags.get('replicatefrom')
@property
def noloadbalance(self) -> bool:
"""``True`` if ``tags.noloadbalance`` configuration is enabled for this node, else ``False``."""
return bool(self.tags.get('noloadbalance', False))
def tags(self) -> Dict[str, Any]:
"""Tags configured for this node, if any."""
return self._tags
def schedule_next_run(self) -> None:
"""Schedule the next run of the ``patroni`` daemon main loop.
patroni/dcs/__init__.py
+32 -37
View File
@@ -23,6 +23,7 @@ import dateutil.parser
from ..exceptions import PatroniFatalException
from ..utils import deep_compare, uri
from ..tags import Tags
if TYPE_CHECKING: # pragma: no cover
from ..config import Config
@@ -191,11 +192,11 @@ _Version = Union[int, str]
_Session = Union[int, float, str, None]
class Member(NamedTuple('Member',
[('version', _Version),
('name', str),
('session', _Session),
('data', Dict[str, Any])])):
class Member(Tags, NamedTuple('Member',
[('version', _Version),
('name', str),
('session', _Session),
('data', Dict[str, Any])])):
"""Immutable object (namedtuple) which represents single member of PostgreSQL cluster.
.. note::
@@ -316,20 +317,10 @@ class Member(NamedTuple('Member',
"""The ``tags`` value from :attr:`~Member.data` if defined, otherwise an empty dictionary."""
return self.data.get('tags', {})
@property
def nofailover(self) -> bool:
"""The value for ``nofailover`` in :attr:`Member`.tags`` if defined, otherwise ``False``."""
return self.tags.get('nofailover', False)
@property
def replicatefrom(self) -> Optional[str]:
"""The value for ``replicatefrom`` in :attr:`Member`.tags`` if defined."""
return self.tags.get('replicatefrom')
@property
def clonefrom(self) -> bool:
"""``True`` if both ``clonefrom`` tag is ``True`` and a connection URL is defined."""
return self.tags.get('clonefrom', False) and bool(self.conn_url)
return super().clonefrom and bool(self.conn_url)
@property
def state(self) -> str:
@@ -932,8 +923,8 @@ class Cluster(NamedTuple('Cluster',
"""``True`` if cluster is configured to use replication slots."""
return bool(self.config and (self.config.data.get('postgresql') or {}).get('use_slots', True))
def get_replication_slots(self, my_name: str, role: str, nofailover: bool,
major_version: int, show_error: bool = False) -> Dict[str, Dict[str, Any]]:
def get_replication_slots(self, my_name: str, role: str, nofailover: bool, major_version: int, *,
is_standby_cluster: bool = False, show_error: bool = False) -> Dict[str, Dict[str, Any]]:
"""Lookup configured slot names in the DCS, report issues found and merge with permanent slots.
Will log an error if:
@@ -945,11 +936,13 @@ class Cluster(NamedTuple('Cluster',
:param role: role of this node.
:param nofailover: ``True`` if this node is tagged to not be a failover candidate.
:param major_version: postgresql major version.
: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 show_error: if ``True`` report error if any disabled logical slots or conflicting slot names are found.
:returns: final dictionary of slot names, after merging with permanent slots and performing sanity checks.
"""
slot_members: List[str] = self._get_slot_members(my_name, role) if self.use_slots else []
slot_members: List[str] = self._get_slot_members(my_name, role)
slots: Dict[str, Dict[str, str]] = {slot_name_from_member_name(name): {'type': 'physical'}
for name in slot_members}
@@ -963,7 +956,7 @@ class Cluster(NamedTuple('Cluster',
"; ".join(f"{', '.join(v)} map to {k}"
for k, v in slot_conflicts.items() if len(v) > 1))
permanent_slots: dict[str, Any] = self._get_permanent_slots(role, nofailover) if self.use_slots else {}
permanent_slots: Dict[str, Any] = self._get_permanent_slots(is_standby_cluster, role, nofailover)
disabled_permanent_logical_slots: List[str] = self._merge_permanent_slots(
slots, permanent_slots, my_name, major_version)
@@ -1022,33 +1015,32 @@ 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, role: str, nofailover: bool) -> Dict[str, Any]:
def _get_permanent_slots(self, is_standby_cluster: bool, role: str, nofailover: bool) -> Dict[str, Any]:
"""Get configured permanent slot names.
.. note::
Permanent logical replication slots are only considered if ``use_slots`` configuration is enabled. Also,
only considered if *role* is ``primary`` or if it is a promotable ``replica`` -- what excludes a
``standby_leader`` or ``replica`` with ``nofailover`` tag enabled. That combination is used for failing
over logical replication slots, and the latter nodes are not eligible for such task.
Permanent replication slots are only considered if ``use_slots`` configuration is enabled.
A node that is not supposed to become a leader (*nofailover*) will not have permanent replication slots.
Permanent physical slots are only considered if *role* is ``primary`` or ``standby_leader``, independently
if ``use_slots`` is enabled or not. That is done that way because even if Patroni itself is not using slots
to replicate among its members when ``use_slots`` is disabled, the user may still have configured Patroni to
keep permanent physical slots used out of Patroni.
In a standby cluster we only support physical replication slots.
The returned dictionary for a non-standby cluster always contains permanent logical replication slots in
order to show a warning if they are not supported by PostgreSQL before v11.
: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 role: role of this node -- ``primary``, ``standby_leader`` or ``replica``.
or logical slots being consumed.
:param nofailover: ``True`` if this node is tagged to not be a failover candidate.
:returns: dictionary of permanent slot names mapped to attributes.
"""
if role in ('master', 'primary', 'standby_leader'):
permanent_slots = (self.__permanent_slots
if role in ('master', 'primary')
else self.__permanent_physical_slots)
else:
permanent_slots = self.__permanent_logical_slots if not nofailover else {}
return permanent_slots
if not self.use_slots or nofailover:
return {}
if is_standby_cluster:
return self.__permanent_physical_slots if role == 'standby_leader' else {}
return self.__permanent_slots if role in ('master', 'primary') else self.__permanent_logical_slots
def _get_slot_members(self, my_name: str, role: str) -> List[str]:
"""Get a list of member names that have replication slots sourcing from this node.
@@ -1065,6 +1057,9 @@ class Cluster(NamedTuple('Cluster',
:returns: list of member names.
"""
if not self.use_slots:
return []
if role in ('master', 'primary', 'standby_leader'):
slot_members = [m.name for m in self.members
if m.name != my_name
patroni/ha.py
+34 -24
View File
@@ -20,31 +20,28 @@ from .postgresql.callback_executor import CallbackAction
from .postgresql.misc import postgres_version_to_int
from .postgresql.postmaster import PostmasterProcess
from .postgresql.rewind import Rewind
from .tags import Tags
from .utils import polling_loop, tzutc
logger = logging.getLogger(__name__)
class _MemberStatus(NamedTuple):
"""Node status distilled from API response:
class _MemberStatus(Tags, NamedTuple('_MemberStatus',
[('member', Member),
('reachable', bool),
('in_recovery', Optional[bool]),
('wal_position', int),
('data', Dict[str, Any])])):
"""Node status distilled from API response.
member - dcs.Member object of the node
reachable - `!False` if the node is not reachable or is not responding with correct JSON
in_recovery - `!True` if pg_is_in_recovery() == true
dcs_last_seen - timestamp from JSON of last succesful communication with DCS
timeline - timeline value from JSON
wal_position - maximum value of `replayed_location` or `received_location` from JSON
tags - dictionary with values of different tags (i.e. nofailover)
watchdog_failed - indicates that watchdog is required by configuration but not available or failed
Consists of the following fields:
:ivar member: :class:`~patroni.dcs.Member` object of the node.
:ivar reachable: ``False`` if the node is not reachable or is not responding with correct JSON.
:ivar in_recovery: ``False`` if the node is running as a primary (`if pg_is_in_recovery() == true`).
:ivar wal_position: maximum value of ``replayed_location`` or ``received_location`` from JSON.
:ivar data: the whole JSON response for future usage.
"""
member: Member
reachable: bool
in_recovery: Optional[bool]
dcs_last_seen: int
timeline: int
wal_position: int
tags: Dict[str, Any]
watchdog_failed: bool
@classmethod
def from_api_response(cls, member: Member, json: Dict[str, Any]) -> '_MemberStatus':
@@ -57,21 +54,34 @@ class _MemberStatus(NamedTuple):
wal: Dict[str, Any] = json.get('wal') or json['xlog']
# abuse difference in primary/replica response format
in_recovery = not (bool(wal.get('location')) or json.get('role') in ('master', 'primary'))
timeline = json.get('timeline', 0)
dcs_last_seen = json.get('dcs_last_seen', 0)
lsn = int(in_recovery and max(wal.get('received_location', 0), wal.get('replayed_location', 0)))
return cls(member, True, in_recovery, dcs_last_seen, timeline, lsn,
json.get('tags', {}), json.get('watchdog_failed', False))
return cls(member, True, in_recovery, lsn, json)
@property
def tags(self) -> Dict[str, Any]:
"""Dictionary with values of different tags (i.e. nofailover)."""
return self.data.get('tags', {})
@property
def timeline(self) -> int:
"""Timeline value from JSON."""
return self.data.get('timeline', 0)
@property
def watchdog_failed(self) -> bool:
"""Indicates that watchdog is required by configuration but not available or failed."""
return self.data.get('watchdog_failed', False)
@classmethod
def unknown(cls, member: Member) -> '_MemberStatus':
return cls(member, False, None, 0, 0, 0, {}, False)
"""Create a new class instance with empty or null values."""
return cls(member, False, None, 0, {})
def failover_limitation(self) -> Optional[str]:
"""Returns reason why this node can't promote or None if everything is ok."""
if not self.reachable:
return 'not reachable'
if self.tags.get('nofailover', False):
if self.nofailover:
return 'not allowed to promote'
if self.watchdog_failed:
return 'not watchdog capable'
patroni/postgresql/__init__.py
+12 -4
View File
@@ -429,7 +429,18 @@ class Postgresql(object):
:param global_config: last known :class:`GlobalConfig` object
"""
self._cluster_info_state = {}
if cluster and cluster.config and cluster.config.modify_version:
if global_config:
self._global_config = global_config
if not self._global_config:
return
if self._global_config.is_standby_cluster:
# Standby cluster can't have logical replication slots, and we don't need to enforce hot_standby_feedback
self._has_permanent_logical_slots = False
self.set_enforce_hot_standby_feedback(False)
elif cluster and cluster.config and cluster.config.modify_version:
self._has_permanent_logical_slots =\
cluster.has_permanent_logical_slots(self.name, nofailover, self.major_version)
@@ -439,9 +450,6 @@ class Postgresql(object):
self._has_permanent_logical_slots
or cluster.should_enforce_hot_standby_feedback(self.name, nofailover, self.major_version))
if global_config:
self._global_config = global_config
def _cluster_info_state_get(self, name: str) -> Optional[Any]:
if not self._cluster_info_state:
try:
patroni/postgresql/slots.py
+9 -3
View File
@@ -471,6 +471,11 @@ class SlotsHandler:
elif cluster.slots and name in cluster.slots: # We want to copy only slots with feedback in a DCS
create_slots.append(name)
# Slots to be copied from the primary should be removed from the *slots* structure,
# otherwise Patroni falsely assumes that they already exist.
for name in create_slots:
slots.pop(name)
error, copy_slots = self.schedule_advance_slots(advance_slots)
if error:
self._schedule_load_slots = True
@@ -493,12 +498,13 @@ class SlotsHandler:
:returns: list of logical replication slots names that should be copied from the primary.
"""
ret = []
if self._postgresql.major_version >= 90400 and cluster.config:
if self._postgresql.major_version >= 90400 and self._postgresql.global_config and cluster.config:
try:
self.load_replication_slots()
slots = cluster.get_replication_slots(self._postgresql.name, self._postgresql.role,
nofailover, self._postgresql.major_version, True)
slots = cluster.get_replication_slots(
self._postgresql.name, self._postgresql.role, nofailover, self._postgresql.major_version,
is_standby_cluster=self._postgresql.global_config.is_standby_cluster, show_error=True)
self._drop_incorrect_slots(cluster, slots, paused)
patroni/postgresql/sync.py
+1 -1
View File
@@ -209,7 +209,7 @@ class _ReplicaList(List[_Replica]):
# 2. can be mapped to a ``Member`` of the ``Cluster``:
# a. ``Member`` doesn't have ``nosync`` tag set;
# b. PostgreSQL on the member is known to be running and accepting client connections.
if member and row[sort_col] is not None and member.is_running and not member.tags.get('nosync', False):
if member and row[sort_col] is not None and member.is_running and not member.nosync:
self.append(_Replica(row['pid'], row['application_name'],
row['sync_state'], row[sort_col], bool(member.nofailover)))
patroni/tags.py
+64
View File
@@ -0,0 +1,64 @@
"""Tags handling."""
import abc
from typing import Any, Dict, Optional
class Tags(abc.ABC):
"""An abstract class that encapsulates all the ``tags`` logic.
Child classes that want to use provided facilities must implement ``tags`` abstract property.
"""
@staticmethod
def _filter_tags(tags: Dict[str, Any]) -> Dict[str, Any]:
"""Get tags configured for this node, if any.
Handle both predefined Patroni tags and custom defined tags.
.. note::
A custom tag is any tag added to the configuration ``tags`` section that is not one of ``clonefrom``,
``nofailover``, ``noloadbalance`` or ``nosync``.
For the Patroni predefined tags, the returning object will only contain them if they are enabled as they
all are boolean values that default to disabled.
:returns: a dictionary of tags set for this node. The key is the tag name, and the value is the corresponding
tag value.
"""
return {tag: value for tag, value in tags.items()
if tag not in ('clonefrom', 'nofailover', 'noloadbalance', 'nosync') or value}
@property
@abc.abstractmethod
def tags(self) -> Dict[str, Any]:
"""Configured tags.
Must be implemented in a child class.
"""
raise NotImplementedError # pragma: no cover
@property
def clonefrom(self) -> bool:
"""``True`` if ``clonefrom`` tag is ``True``, else ``False``."""
return self.tags.get('clonefrom', False)
@property
def nofailover(self) -> bool:
"""``True`` if ``nofailover`` is ``True``, else ``False``."""
return bool(self.tags.get('nofailover', False))
@property
def noloadbalance(self) -> bool:
"""``True`` if ``noloadbalance`` is ``True``, else ``False``."""
return bool(self.tags.get('noloadbalance', False))
@property
def nosync(self) -> bool:
"""``True`` if ``nosync`` is ``True``, else ``False``."""
return bool(self.tags.get('nosync', False))
@property
def replicatefrom(self) -> Optional[str]:
"""Value of ``replicatefrom`` tag, if any."""
return self.tags.get('replicatefrom')
requirements.docs.txt
+1 -1
View File
@@ -1,5 +1,5 @@
sphinx>=4
sphinx_rtd_theme
sphinx_rtd_theme>1
sphinxcontrib-apidoc
sphinx-github-style
pyyaml
tests/test_api.py
+1 -1
View File
@@ -100,7 +100,7 @@ class MockHa(object):
@staticmethod
def fetch_nodes_statuses(members):
return [_MemberStatus(None, True, None, 0, 0, None, {}, False)]
return [_MemberStatus(None, True, None, 0, {})]
@staticmethod
def schedule_future_restart(data):
tests/test_ha.py
+6 -2
View File
@@ -99,7 +99,9 @@ def get_node_status(reachable=True, in_recovery=True, dcs_last_seen=0,
tags = {}
if nofailover:
tags['nofailover'] = True
return _MemberStatus(e, reachable, in_recovery, dcs_last_seen, timeline, wal_position, tags, watchdog_failed)
return _MemberStatus(e, reachable, in_recovery, wal_position,
{'tags': tags, 'watchdog_failed': watchdog_failed,
'dcs_last_seen': dcs_last_seen, 'timeline': timeline})
return fetch_node_status
@@ -1294,14 +1296,16 @@ class TestHa(PostgresInit):
mock_restart.assert_called_once()
self.ha.dcs.get_cluster.assert_not_called()
@patch.object(Cluster, 'is_unlocked', Mock(return_value=False))
def test_enable_synchronous_mode(self):
self.ha.is_synchronous_mode = true
self.ha.has_lock = true
self.p.name = 'leader'
self.p.sync_handler.current_state = Mock(return_value=(CaseInsensitiveSet(), CaseInsensitiveSet()))
self.ha.dcs.write_sync_state = Mock(return_value=SyncState.empty())
with patch('patroni.ha.logger.info') as mock_logger:
self.ha.run_cycle()
self.assertEqual(mock_logger.call_args[0][0], 'Enabled synchronous replication')
self.assertEqual(mock_logger.call_args_list[0][0][0], 'Enabled synchronous replication')
self.ha.dcs.write_sync_state = Mock(return_value=None)
with patch('patroni.ha.logger.warning') as mock_logger:
self.ha.run_cycle()
tests/test_patroni.py
+1 -1
View File
@@ -185,7 +185,7 @@ class TestPatroni(unittest.TestCase):
def test_reload_config(self):
self.p.reload_config()
self.p.get_tags = Mock(side_effect=Exception)
self.p._get_tags = Mock(side_effect=Exception)
self.p.reload_config(local=True)
def test_nosync(self):
tests/test_slots.py
+3
View File
@@ -7,6 +7,7 @@ from mock import Mock, PropertyMock, patch
from threading import Thread
from patroni import psycopg
from patroni.config import GlobalConfig
from patroni.dcs import Cluster, ClusterConfig, Member, SyncState
from patroni.postgresql import Postgresql
from patroni.postgresql.misc import fsync_dir
@@ -28,6 +29,7 @@ class TestSlotsHandler(BaseTestPostgresql):
@patch.object(Postgresql, 'is_running', Mock(return_value=True))
def setUp(self):
super(TestSlotsHandler, self).setUp()
self.p._global_config = GlobalConfig({})
self.s = self.p.slots_handler
self.p.start()
config = ClusterConfig(1, {'slots': {'ls': {'database': 'a', 'plugin': 'b'}}}, 1)
@@ -44,6 +46,7 @@ class TestSlotsHandler(BaseTestPostgresql):
self.s.sync_replication_slots(cluster, False)
self.p.set_role('standby_leader')
with patch.object(SlotsHandler, 'drop_replication_slot', Mock(return_value=(True, False))), \
patch.object(GlobalConfig, 'is_standby_cluster', PropertyMock(return_value=True)), \
patch('patroni.postgresql.slots.logger.debug') as mock_debug:
self.s.sync_replication_slots(cluster, False)
mock_debug.assert_called_once()