From eb100fd5869d2bcf020a50f9b0b643c61f27c49f Mon Sep 17 00:00:00 2001 From: Matt Baker <93600443+matthbakeredb@users.noreply.github.com> Date: Tue, 8 Aug 2023 07:49:12 +0100 Subject: [PATCH] Add docs to `patroni.dcs.__init__.py` (#2777) Also, made some small code changes to satisfy formatting and pylint. --- patroni/dcs/__init__.py | 1005 ++++++++++++++++++++++++++++++--------- 1 file changed, 786 insertions(+), 219 deletions(-) diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index 75ae32c3..e536e9dc 100644 --- a/patroni/dcs/__init__.py +++ b/patroni/dcs/__init__.py @@ -1,5 +1,5 @@ +"""Abstract classes for Distributed Configuration Store.""" import abc -import dateutil.parser import datetime import importlib import inspect @@ -10,7 +10,6 @@ import pkgutil import re import sys import time - from collections import defaultdict from copy import deepcopy from random import randint @@ -20,8 +19,11 @@ from typing import Any, Callable, Collection, Dict, List, NamedTuple, Optional, Type, Iterator from urllib.parse import urlparse, urlunparse, parse_qsl +import dateutil.parser + from ..exceptions import PatroniFatalException from ..utils import deep_compare, uri + if TYPE_CHECKING: # pragma: no cover from ..config import Config @@ -34,30 +36,45 @@ logger = logging.getLogger(__name__) def slot_name_from_member_name(member_name: str) -> str: """Translate member name to valid PostgreSQL slot name. - PostgreSQL replication slot names must be valid PostgreSQL names. This function maps the wider space of - member names to valid PostgreSQL names. Names are lowercased, dashes and periods common in hostnames - are replaced with underscores, other characters are encoded as their unicode codepoint. Name is truncated - to 64 characters. Multiple different member names may map to a single slot name.""" + .. note:: + PostgreSQL's replication slot names must be valid PostgreSQL names. This function maps the wider space of + member names to valid PostgreSQL names. Names have their case lowered, dashes and periods common in hostnames + are replaced with underscores, other characters are encoded as their unicode codepoint. Name is truncated + to 64 characters. Multiple different member names may map to a single slot name. + + :param member_name: The string to convert to a slot name. + + :returns: The string converted using the rules described above. + """ def replace_char(match: Any) -> str: c = match.group(0) - return '_' if c in '-.' else "u{:04d}".format(ord(c)) + return '_' if c in '-.' else f"u{ord(c):04d}" slot_name = re.sub('[^a-z0-9_]', replace_char, member_name.lower()) return slot_name[0:63] def parse_connection_string(value: str) -> Tuple[str, Union[str, None]]: - """Original Governor stores connection strings for each cluster members if a following format: - postgres://{username}:{password}@{connect_address}/postgres - Since each of our patroni instances provides own REST API endpoint it's good to store this information - in DCS among with postgresql connection string. In order to not introduce new keys and be compatible with - original Governor we decided to extend original connection string in a following way: - postgres://{username}:{password}@{connect_address}/postgres?application_name={api_url} - This way original Governor could use such connection string as it is, because of feature of `libpq` library. + """Split and rejoin a URL string into a connection URL and an API URL. - This method is able to split connection string stored in DCS into two parts, `conn_url` and `api_url`""" + .. note:: + Original Governor stores connection strings for each cluster members in a following format: + postgres://{username}:{password}@{connect_address}/postgres + + Since each of our patroni instances provides their own REST API endpoint, it's good to store this information + in DCS along with PostgreSQL connection string. In order to not introduce new keys and be compatible with + original Governor we decided to extend original connection string in a following way: + + postgres://{username}:{password}@{connect_address}/postgres?application_name={api_url} + + This way original Governor could use such connection string as it is, because of feature of ``libpq`` library. + + :param value: The URL string to split. + + :returns: the connection string stored in DCS split into two parts, ``conn_url`` and ``api_url``. + """ scheme, netloc, path, params, query, fragment = urlparse(value) conn_url = urlunparse((scheme, netloc, path, params, '', fragment)) api_url = ([v for n, v in parse_qsl(query) if n == 'application_name'] or [None])[0] @@ -65,11 +82,15 @@ def parse_connection_string(value: str) -> Tuple[str, Union[str, None]]: def dcs_modules() -> List[str]: - """Get names of DCS modules, depending on execution environment. If being packaged with PyInstaller, - modules aren't discoverable dynamically by scanning source directory because `FrozenImporter` doesn't - implement `iter_modules` method. But it is still possible to find all potential DCS modules by - iterating through `toc`, which contains list of all "frozen" resources.""" + """Get names of DCS modules, depending on execution environment. + .. note:: + If being packaged with PyInstaller, modules aren't discoverable dynamically by scanning source directory because + :class:`importlib.machinery.FrozenImporter` doesn't implement :func:`iter_modules`. But it is still possible to + find all potential DCS modules by iterating through ``toc``, which contains list of all "frozen" resources. + + :returns: list of known module names with absolute python module path namespace, e.g. ``patroni.dcs.etcd``. + """ dcs_dirname = os.path.dirname(__file__) module_prefix = __package__ + '.' @@ -83,8 +104,8 @@ def dcs_modules() -> List[str]: if hasattr(importer, 'toc'): toc |= getattr(importer, 'toc') return [module for module in toc if module.startswith(module_prefix) and module.count('.') == 2] - else: - return [module_prefix + name for _, name, is_pkg in pkgutil.iter_modules([dcs_dirname]) if not is_pkg] + + return [module_prefix + name for _, name, is_pkg in pkgutil.iter_modules([dcs_dirname]) if not is_pkg] def iter_dcs_classes( @@ -199,11 +220,24 @@ class Member(NamedTuple('Member', @staticmethod def from_node(version: _Version, name: str, session: _Session, value: str) -> 'Member': - """ - >>> Member.from_node(-1, '', '', '{"conn_url": "postgres://foo@bar/postgres"}') is not None - True - >>> Member.from_node(-1, '', '', '{') - Member(version=-1, name='', session='', data={}) + """Factory method for instantiating :class:`Member` from a JSON serialised string or object. + + :param version: modification version of a given member key in a Configuration Store. + :param name: name of PostgreSQL cluster member. + :param session: either session id or just ttl in seconds. + :param value: JSON encoded string containing arbitrary data i.e. ``conn_url``, ``api_url``, + ``xlog_location``, ``state``, ``role``, ``tags``, etc. OR a connection URL + starting with ``postgres://``. + + :returns: an :class:`Member` instance built with the given arguments. + + :Example: + + >>> Member.from_node(-1, '', '', '{"conn_url": "postgres://foo@bar/postgres"}') is not None + True + + >>> Member.from_node(-1, '', '', '{') + Member(version=-1, name='', session='', data={}) """ if value.startswith('postgres'): conn_url, api_url = parse_connection_string(value) @@ -218,6 +252,7 @@ class Member(NamedTuple('Member', @property def conn_url(self) -> Optional[str]: + """The ``conn_url`` value from :attr:`~Member.data` if defined or constructed from ``conn_kwargs``.""" conn_url = self.data.get('conn_url') if conn_url: return conn_url @@ -228,7 +263,21 @@ class Member(NamedTuple('Member', self.data['conn_url'] = conn_url return conn_url + return None + def conn_kwargs(self, auth: Union[Any, Dict[str, Any], None] = None) -> Dict[str, Any]: + """Give keyword arguments used for PostgreSQL connection settings. + + :param auth: Authentication properties - can be defined as anything supported by the ``psycopg2`` or + ``psycopg`` modules. + Converts a key of ``username`` to ``user`` if supplied. + + :returns: A dictionary containing a merge of default parameter keys ``host``, ``port`` and ``dbname``, with + the contents of :attr:`~Member.data` ``conn_kwargs`` key. If those are not defined will + parse and reform connection parameters from :attr:`~Member.conn_url`. One of these two attributes + needs to have data defined to construct the output dictionary. Finally, *auth* parameters are merged + with the dictionary before returned. + """ defaults = { "host": None, "port": None, @@ -259,40 +308,56 @@ class Member(NamedTuple('Member', @property def api_url(self) -> Optional[str]: + """The ``api_url`` value from :attr:`~Member.data` if defined.""" return self.data.get('api_url') @property def tags(self) -> Dict[str, Any]: + """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) @property def state(self) -> str: + """The ``state`` value of :attr:`~Member.data`.""" return self.data.get('state', 'unknown') @property def is_running(self) -> bool: + """``True`` if the member :attr:`~Member.state` is ``running``.""" return self.state == 'running' @property def patroni_version(self) -> Optional[Tuple[int, ...]]: + """The ``version`` string value from :attr:`~Member.data` converted to tuple. + + :Example: + + >>> Member.from_node(1, '', '', '{"version":"1.2.3"}').patroni_version + (1, 2, 3) + + """ version = self.data.get('version') if version: try: return tuple(map(int, version.split('.'))) except Exception: logger.debug('Failed to parse Patroni version %s', version) + return None class RemoteMember(Member): @@ -314,81 +379,125 @@ class RemoteMember(Member): """Factory method to construct instance from given *name* and *data*. :param name: name of the remote member. - :param data: dictionary of member information. + :param data: dictionary of member information, which can contain keys from :const:`~RemoteMember.ALLOWED_KEYS` + but also member connection information ``api_url`` and ``conn_kwargs``, and slot information. :returns: constructed instance using supplied parameters. """ return super(RemoteMember, cls).__new__(cls, -1, name, None, data) def __getattr__(self, name: str) -> Any: - if name in RemoteMember.ALLOWED_KEYS: - return self.data.get(name) + """Dictionary style key lookup. + + :param name: key to lookup. + + :returns: value of *name* key in :attr:`~RemoteMember.data` if key *name* is in + :cvar:`~RemoteMember.ALLOWED_KEYS`, else ``None``. + """ + return self.data.get(name) if name in RemoteMember.ALLOWED_KEYS else None class Leader(NamedTuple): """Immutable object (namedtuple) which represents leader key. Consists of the following fields: - :param version: modification version of a leader key in a Configuration Store - :param session: either session id or just ttl in seconds - :param member: reference to a `Member` object which represents current leader (see `Cluster.members`) + + :ivar version: modification version of a leader key in a Configuration Store + :ivar session: either session id or just ttl in seconds + :ivar member: reference to a :class:`Member` object which represents current leader (see :attr:`Cluster.members`) """ + version: _Version session: _Session member: Member @property def name(self) -> str: + """The leader "member" name.""" return self.member.name def conn_kwargs(self, auth: Optional[Dict[str, str]] = None) -> Dict[str, str]: + """Connection keyword arguments. + + :param auth: an optional dictionary containing authentication information. + + :returns: the result of the called :meth:`Member.conn_kwargs` method. + """ return self.member.conn_kwargs(auth) @property def conn_url(self) -> Optional[str]: + """Connection URL value of the :class:`Member` instance.""" return self.member.conn_url @property def data(self) -> Dict[str, Any]: + """Data value of the :class:`Member` instance.""" return self.member.data @property def timeline(self) -> Optional[int]: + """Timeline value of :attr:`~Member.data`.""" return self.data.get('timeline') @property def checkpoint_after_promote(self) -> Optional[bool]: - """ - >>> Leader(1, '', Member.from_node(1, '', '', '{"version":"z"}')).checkpoint_after_promote + """Determine whether a checkpoint has occurred for this leader after promotion. + :returns: ``True`` if the role is ``master`` or ``primary`` and ``checkpoint_after_promote`` is not set, + ``False`` if not a ``master`` or ``primary`` or if the checkpoint hasn't occurred. + If the version of Patroni is older than 1.5.6, return ``None``. + + :Example: + + >>> Leader(1, '', Member.from_node(1, '', '', '{"version":"z"}')).checkpoint_after_promote """ version = self.member.patroni_version # 1.5.6 is the last version which doesn't expose checkpoint_after_promote: false if version and version > (1, 5, 6): return self.data.get('role') in ('master', 'primary') and 'checkpoint_after_promote' not in self.data + return None class Failover(NamedTuple): + """Immutable object (namedtuple) representing configuration information required for failover/switchover capability. + :ivar version: version of the object. + :ivar leader: name of the leader. If value isn't empty we treat it as a switchover from the specified node. + :ivar candidate: the name of the member node to be considered as a failover candidate. + :ivar scheduled_at: in the case of a switchover the :class:`~datetime.datetime` object to perform the scheduled + switchover. + + :Example: + + >>> 'Failover' in str(Failover.from_node(1, '{"leader": "cluster_leader"}')) + True + + >>> 'Failover' in str(Failover.from_node(1, {"leader": "cluster_leader"})) + True + + >>> 'Failover' in str(Failover.from_node(1, '{"leader": "cluster_leader", "member": "cluster_candidate"}')) + True + + >>> Failover.from_node(1, 'null') is None + False + + >>> n = '''{"leader": "cluster_leader", "member": "cluster_candidate", + ... "scheduled_at": "2016-01-14T10:09:57.1394Z"}''' + + >>> 'tzinfo=' in str(Failover.from_node(1, n)) + True + + >>> Failover.from_node(1, None) is None + False + + >>> Failover.from_node(1, '{}') is None + False + + >>> 'abc' in Failover.from_node(1, 'abc:def') + True """ - >>> 'Failover' in str(Failover.from_node(1, '{"leader": "cluster_leader"}')) - True - >>> 'Failover' in str(Failover.from_node(1, {"leader": "cluster_leader"})) - True - >>> 'Failover' in str(Failover.from_node(1, '{"leader": "cluster_leader", "member": "cluster_candidate"}')) - True - >>> Failover.from_node(1, 'null') is None - False - >>> n = '{"leader": "cluster_leader", "member": "cluster_candidate", "scheduled_at": "2016-01-14T10:09:57.1394Z"}' - >>> 'tzinfo=' in str(Failover.from_node(1, n)) - True - >>> Failover.from_node(1, None) is None - False - >>> Failover.from_node(1, '{}') is None - False - >>> 'abc' in Failover.from_node(1, 'abc:def') - True - """ + version: _Version leader: Optional[str] candidate: Optional[str] @@ -396,6 +505,15 @@ class Failover(NamedTuple): @staticmethod def from_node(version: _Version, value: Union[str, Dict[str, str]]) -> 'Failover': + """Factory method to parse *value* as failover configuration. + + :param version: version number for the object. + :param value: JSON serialized data or a dictionary of configuration. + Can also be a colon ``:`` delimited list of leader, followed by candidate name (legacy format). + If ``scheduled_at`` key is defined the value will be parsed by :func:`dateutil.parser.parse`. + + :returns: constructed :class:`Failover` information object + """ if isinstance(value, dict): data: Dict[str, Any] = value elif value: @@ -418,21 +536,54 @@ class Failover(NamedTuple): return Failover(version, data.get('leader'), data.get('member'), data.get('scheduled_at')) def __len__(self) -> int: + """Implement ``len`` function capability. + + .. note:: + This magic method aids in the evaluation of "emptiness" of a :class:`Failover` instance. For example: + + >>> failover = Failover.from_node(1, None) + >>> len(failover) + 0 + >>> assert bool(failover) is False + + >>> failover = Failover.from_node(1, {"leader": "cluster_leader"}) + >>> len(failover) + 1 + >>> assert bool(failover) is True + + This makes it easier to write ``if cluster.failover`` rather than the longer statement. + + """ return int(bool(self.leader)) + int(bool(self.candidate)) class ClusterConfig(NamedTuple): + """Immutable object (namedtuple) which represents cluster configuration. + + :ivar version: version number for the object. + :ivar data: dictionary of configuration information. + :ivar modify_version: modified version number. + """ + version: _Version data: Dict[str, Any] modify_version: _Version @staticmethod def from_node(version: _Version, value: str, modify_version: Optional[_Version] = None) -> 'ClusterConfig': - """ - >>> ClusterConfig.from_node(1, '{') is None - False - """ + """Factory method to parse *value* as configuration information. + :param version: version number for object. + :param value: raw JSON serialized data, if not parsable replaced with an empty dictionary. + :param modify_version: optional modify version number, use *version* if not provided. + + :returns: constructed :class:`ClusterConfig` instance. + + :Example: + + >>> ClusterConfig.from_node(1, '{') is None + False + """ try: data = json.loads(value) assert isinstance(data, dict) @@ -443,44 +594,63 @@ class ClusterConfig(NamedTuple): @property def permanent_slots(self) -> Dict[str, Any]: - return self.data.get('permanent_replication_slots')\ - or self.data.get('permanent_slots') or self.data.get('slots') or {} + """Dictionary of permanent slots information looked up from :attr:`~ClusterConfig.data`.""" + return (self.data.get('permanent_replication_slots') + or self.data.get('permanent_slots') + or self.data.get('slots') + or {}) @property def ignore_slots_matchers(self) -> List[Dict[str, Any]]: + """The value for ``ignore_slots`` from :attr:`~ClusterConfig.data` if defined or an empty list.""" return self.data.get('ignore_slots') or [] @property def max_timelines_history(self) -> int: + """The value for ``max_timelines_history`` from :attr:`~ClusterConfig.data` if defined or ``0``.""" return self.data.get('max_timelines_history', 0) class SyncState(NamedTuple): - """Immutable object (namedtuple) which represents last observed synhcronous replication state + """Immutable object (namedtuple) which represents last observed synchronous replication state. - :param version: modification version of a synchronization key in a Configuration Store - :param leader: reference to member that was leader - :param sync_standby: synchronous standby list (comma delimited) which are last synchronized to leader + :ivar version: modification version of a synchronization key in a Configuration Store. + :ivar leader: reference to member that was leader. + :ivar sync_standby: synchronous standby list (comma delimited) which are last synchronized to leader. """ + version: Optional[_Version] leader: Optional[str] sync_standby: Optional[str] @staticmethod def from_node(version: Optional[_Version], value: Union[str, Dict[str, Any], None]) -> 'SyncState': - """ - >>> SyncState.from_node(1, None).leader is None - True - >>> SyncState.from_node(1, '{}').leader is None - True - >>> SyncState.from_node(1, '{').leader is None - True - >>> SyncState.from_node(1, '[]').leader is None - True - >>> SyncState.from_node(1, '{"leader": "leader"}').leader == "leader" - True - >>> SyncState.from_node(1, {"leader": "leader"}).leader == "leader" - True + """Factory method to parse *value* as synchronisation state information. + + :param version: optional *version* number for the object. + :param value: (optionally JSON serialised) sychronisation state information + + :returns: constructed :class:`SyncState` object. + + :Example: + + >>> SyncState.from_node(1, None).leader is None + True + + >>> SyncState.from_node(1, '{}').leader is None + True + + >>> SyncState.from_node(1, '{').leader is None + True + + >>> SyncState.from_node(1, '[]').leader is None + True + + >>> SyncState.from_node(1, '{"leader": "leader"}').leader == "leader" + True + + >>> SyncState.from_node(1, {"leader": "leader"}).leader == "leader" + True """ try: if value and isinstance(value, str): @@ -492,49 +662,69 @@ class SyncState(NamedTuple): @staticmethod def empty(version: Optional[_Version] = None) -> 'SyncState': + """Construct an empty :class:`SyncState` instance. + + :param version: optional version number. + + :returns: empty synchronisation state object. + """ return SyncState(version, None, None) @property def is_empty(self) -> bool: - """:returns: True if /sync key is not valid (doesn't have a leader).""" + """``True`` if ``/sync`` key is not valid (doesn't have a leader).""" return not self.leader @staticmethod def _str_to_list(value: str) -> List[str]: """Splits a string by comma and returns list of strings. - :param value: a comma separated string - :returns: list of non-empty strings after splitting an input value by comma + :param value: a comma separated string. + + :returns: list of non-empty strings after splitting an input value by comma. """ return list(filter(lambda a: a, [s.strip() for s in value.split(',')])) @property def members(self) -> List[str]: - """:returns: sync_standby as list.""" + """:attr:`~SyncState.sync_standby` as list or an empty list if undefined or object considered ``empty``.""" return self._str_to_list(self.sync_standby) if not self.is_empty and self.sync_standby else [] def matches(self, name: Optional[str], check_leader: bool = False) -> bool: """Checks if node is presented in the /sync state. Since PostgreSQL does case-insensitive checks for synchronous_standby_name we do it also. - :param name: name of the node - :param check_leader: by default the name is searched in members, check_leader=True will include leader to list - :returns: `True` if the /sync key not :func:`is_empty` and a given name is among presented in the sync state - >>> s = SyncState(1, 'foo', 'bar,zoo') - >>> s.matches('foo') - False - >>> s.matches('fOo', True) - True - >>> s.matches('Bar') - True - >>> s.matches('zoO') - True - >>> s.matches('baz') - False - >>> s.matches(None) - False - >>> SyncState.empty(1).matches('foo') - False + + :param name: name of the node. + :param check_leader: by default the *name* is searched for only in members, a value of ``True`` will include the + leader to list. + + :returns: ``True`` if the ``/sync`` key not :func:`is_empty` and the given *name* is among those presented in + the sync state. + + :Example: + >>> s = SyncState(1, 'foo', 'bar,zoo') + + >>> s.matches('foo') + False + + >>> s.matches('fOo', True) + True + + >>> s.matches('Bar') + True + + >>> s.matches('zoO') + True + + >>> s.matches('baz') + False + + >>> s.matches(None) + False + + >>> SyncState.empty(1).matches('foo') + False """ ret = False if name and not self.is_empty: @@ -543,7 +733,10 @@ class SyncState(NamedTuple): return ret def leader_matches(self, name: Optional[str]) -> bool: - """:returns: `True` if name is matching the `SyncState.leader` value.""" + """Compare the given *name* to stored leader value. + + :returns: ``True`` if *name* is matching the :attr:`~SyncState.leader` value. + """ return bool(name and not self.is_empty and name.lower() == (self.leader or '').lower()) @@ -551,17 +744,40 @@ _HistoryTuple = Union[Tuple[int, int, str], Tuple[int, int, str, str], Tuple[int class TimelineHistory(NamedTuple): - """Object representing timeline history file""" + """Object representing timeline history file. + + .. note:: + The content held in *lines* deserialized from *value* are lines parsed from PostgreSQL timeline history files, + consisting of the timeline number, the LSN where the timeline split and any other string held in the file. + The files are parsed by :func:`~patroni.postgresql.misc.parse_history`. + + :ivar version: version number of the file. + :ivar value: raw JSON serialised data consisting of parsed lines from history files. + :ivar lines: ``List`` of ``Tuple`` parsed lines from history files. + """ + version: _Version value: Any lines: List[_HistoryTuple] @staticmethod def from_node(version: _Version, value: str) -> 'TimelineHistory': - """ - >>> h = TimelineHistory.from_node(1, 2) - >>> h.lines - [] + """Parse the given JSON serialized string as a list of timeline history lines. + + :param version: version number + :param value: JSON serialized string, consisting of parsed lines of PostgreSQL timeline history files, + see :class:`TimelineHistory`. + + :returns: composed timeline history object using parsed lines. + + :Example: + + If the passed *value* argument is not parsed an empty list of lines is returned: + + >>> h = TimelineHistory.from_node(1, 2) + + >>> h.lines + [] """ try: lines = json.loads(value) @@ -615,50 +831,105 @@ class Cluster(NamedTuple('Cluster', @staticmethod def empty() -> 'Cluster': """Produce an empty :class:`Cluster` instance.""" - return Cluster(None, None, None, 0, [], None, SyncState.empty(), None, None, None) + return Cluster(None, None, None, 0, [], None, SyncState.empty(), None, None, None, {}) def is_empty(self): - return self.initialize is None and self.config is None and self.leader is None and self.last_lsn == 0\ - and self.members == [] and self.failover is None and self.sync.version is None\ - and self.history is None and self.slots is None and self.failsafe is None and self.workers == {} + """Validate definition of all attributes of this :class:`Cluster` instance. + + :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, + 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 == {})) def __len__(self) -> int: + """Implement ``len`` function capability. + + .. note:: + This magic method aids in the evaluation of "emptiness" of a ``Cluster`` instance. For example: + + >>> cluster = Cluster.empty() + >>> len(cluster) + 0 + + >>> assert bool(cluster) is False + + >>> cluster = Cluster(None, None, None, 0, [1, 2, 3], None, SyncState.empty(), None, None, None, {}) + >>> len(cluster) + 1 + + >>> assert bool(cluster) is True + + This makes it easier to write ``if cluster`` rather than the longer statement. + + """ return int(not self.is_empty()) @property def leader_name(self) -> Optional[str]: + """The name of the leader if defined otherwise ``None``.""" return self.leader and self.leader.name def is_unlocked(self) -> bool: + """Check if the cluster does not have the leader. + + :returns: ``True`` if a leader name is not defined. + """ return not self.leader_name def has_member(self, member_name: str) -> bool: + """Check if the given member name is present in the cluster. + + :param member_name: name to look up in the :attr:`~Cluster.members`. + + :returns: ``True`` if the member name is found. + """ return any(m for m in self.members if m.name == member_name) def get_member(self, member_name: str, fallback_to_leader: bool = True) -> Union[Member, Leader, None]: - return ([m for m in self.members if m.name == member_name] or [self.leader if fallback_to_leader else None])[0] + """Get :class:`Member` object by name or the :class:`Leader`. + + :param member_name: name of the member to retrieve. + :param fallback_to_leader: if ``True`` return the :class:`Leader` instead if the member cannot be found. + + :returns: the :class:`Member` if found or :class:`Leader` object. + """ + return next((m for m in self.members if m.name == member_name), + self.leader if fallback_to_leader else None) def get_clone_member(self, exclude_name: str) -> Union[Member, Leader, None]: + """Get member or leader object to use as clone source. + + :param exclude_name: name of a member name to exclude. + + :returns: a randomly selected candidate member from available running members that are configured to as viable + sources for cloning (has tag ``clonefrom`` in configuration). If no member is appropriate the current + leader is used. + """ exclude = [exclude_name] + ([self.leader.name] if self.leader else []) candidates = [m for m in self.members if m.clonefrom and m.is_running and m.name not in exclude] return candidates[randint(0, len(candidates) - 1)] if candidates else self.leader @property def __permanent_slots(self) -> Dict[str, Union[Dict[str, Any], Any]]: + """Dictionary of permanent replication slots.""" return self.config and self.config.permanent_slots or {} @property def __permanent_physical_slots(self) -> Dict[str, Any]: + """Dictionary of permanent ``physical`` replication slots.""" return {name: value for name, value in self.__permanent_slots.items() if not value or isinstance(value, dict) and value.get('type', 'physical') == 'physical'} @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')} @property def use_slots(self) -> bool: + """``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, @@ -807,17 +1078,32 @@ class Cluster(NamedTuple('Cluster', return slot_members def has_permanent_logical_slots(self, my_name: str, nofailover: bool, major_version: int = 110000) -> bool: + """Check if the given member node has permanent ``logical`` replication slots configured. + + :param my_name: name of the member node to check. + :param nofailover: ``True`` if this node is tagged to not be a failover candidate. + :param major_version: the PostgreSQL major version number. + + :returns: ``False`` if PostgreSQL is < 11, ``True`` if any detected replications slots are ``logical``. + """ if major_version < 110000: return False slots = self.get_replication_slots(my_name, 'replica', nofailover, major_version).values() return any(v for v in slots if v.get("type") == "logical") def should_enforce_hot_standby_feedback(self, my_name: str, nofailover: bool, major_version: int) -> bool: - """ - The hot_standby_feedback must be enabled if the current replica has logical slots - or it is working as a cascading replica for the other node that has logical slots. - """ + """Determine whether ``hot_standby_feedback`` should be enabled for the given member. + The ``hot_standby_feedback`` must be enabled if the current replica has ``logical`` slots, + or it is working as a cascading replica for the other node that has ``logical`` slots. + + :param my_name: name of the member node to check. + :param nofailover: ``True`` if this node is tagged to not be a failover candidate. + :param major_version: PostgreSQL major version number. + + :returns: ``True`` if this node or any member replicating from this node has permanent logical slots. + ``False`` if PostgreSQL major version is < 11. + """ if major_version < 110000: return False @@ -830,25 +1116,48 @@ class Cluster(NamedTuple('Cluster', return False def get_my_slot_name_on_primary(self, my_name: str, replicatefrom: Optional[str]) -> str: - """ - P <-- I <-- L - In case of cascading replication we have to check not our physical slot, - but slot of the replica that connects us to the primary. - """ + """Canonical slot name for physical replication. + .. note:: + P <-- I <-- L + + In case of cascading replication we have to check not our physical slot, but slot of the replica that + connects us to the primary. + + :param my_name: the member node name that is replicating. + :param replicatefrom: the Intermediate member name that is configured to replicate for cascading replication. + + :returns: The slot name that is in use for physical replication on this no`de. + """ m = self.get_member(replicatefrom, False) if replicatefrom else None - return self.get_my_slot_name_on_primary(m.name, m.replicatefrom)\ + return self.get_my_slot_name_on_primary(m.name, m.replicatefrom) \ if isinstance(m, Member) else slot_name_from_member_name(my_name) @property def timeline(self) -> int: - """ - >>> Cluster(0, 0, 0, 0, 0, 0, 0, 0, 0, None).timeline - 0 - >>> Cluster(0, 0, 0, 0, 0, 0, 0, TimelineHistory.from_node(1, '[]'), 0, None).timeline - 1 - >>> Cluster(0, 0, 0, 0, 0, 0, 0, TimelineHistory.from_node(1, '[["a"]]'), 0, None).timeline - 0 + """Get the cluster history index from the :attr:`~Cluster.history`. + + :returns: If the recorded history is empty assume timeline is ``1``, if it is not defined or the stored history + is not formatted as expected ``0`` is returned and an error will be logged. + Otherwise, the last number stored incremented by 1 is returned. + + :Example: + + No history provided: + >>> Cluster(0, 0, 0, 0, 0, 0, 0, 0, 0, None, {}).timeline + 0 + + Empty history assume timeline is ``1``: + >>> Cluster(0, 0, 0, 0, 0, 0, 0, TimelineHistory.from_node(1, '[]'), 0, None, {}).timeline + 1 + + Invalid history format, a string of ``a``, returns ``0``: + >>> Cluster(0, 0, 0, 0, 0, 0, 0, TimelineHistory.from_node(1, '[["a"]]'), 0, None, {}).timeline + 0 + + History as a list of strings: + >>> Cluster(0, 0, 0, 0, 0, 0, 0, TimelineHistory.from_node(1, '[["3", "2", "1"]]'), 0, None, {}).timeline + 4 """ if self.history: if self.history.lines: @@ -862,14 +1171,22 @@ class Cluster(NamedTuple('Cluster', @property def min_version(self) -> Optional[Tuple[int, ...]]: + """Lowest Patroni software version found in known members of the cluster.""" return next(iter(sorted(m.patroni_version for m in self.members if m.patroni_version)), None) class ReturnFalseException(Exception): - pass + """Exception to be caught by the :func:`catch_return_false_exception` decorator.""" def catch_return_false_exception(func: Callable[..., Any]) -> Any: + """Decorator function for catching functions raising :exc:`ReturnFalseException`. + + :param func: function to be wrapped. + + :returns: wrapped function. + """ + def wrapper(*args: Any, **kwargs: Any): try: return func(*args, **kwargs) @@ -880,6 +1197,74 @@ def catch_return_false_exception(func: Callable[..., Any]) -> Any: class AbstractDCS(abc.ABC): + """Abstract representation of DCS modules. + + Implementations of a concrete DCS class, using appropriate backend client interfaces, must include the following + methods and properties. + + Functional methods that are critical in their timing, required to complete within ``retry_timeout`` period in order + to prevent the DCS considered inaccessible, each perform construction of complex data objects: + + * :meth:`~AbstractDCS._cluster_loader`: + method which processes the structure of data stored in the DCS used to build the :class:`Cluster` object + with all relevant associated data. + * :meth:`~AbstractDCS._citus_cluster_loader`: + Similar to above but specifically representing Citus group and workers information. + * :meth:`~AbstractDCS._load_cluster`: + main method for calling specific ``loader`` method to build the :class:`Cluster` object representing the + state and topology of the cluster. + + Functional methods that are critical in their timing and must be written with ACID transaction properties in mind: + + * :meth:`~AbstractDCS.attempt_to_acquire_leader`: + method used in the leader race to attempt to acquire the leader lock by creating the leader key in the DCS, + if it does not exist. + * :meth:`~AbstractDCS._update_leader`: + method to update ``leader`` key in DCS. Relies on Compare-And-Set to ensure the Primary lock key is updated. + If this fails to update within the ``retry_timeout`` window the Primary will be demoted. + + Functional method that relies on Compare-And-Create to ensure only one member creates the relevant key: + + * :meth:`~AbstractDCS.initialize`: + method used in the race for cluster initialization which creates the ``initialize`` key in the DCS. + + DCS backend getter and setter methods and properties: + + * :meth:`~AbstractDCS.take_leader`: method to create a new leader key in the DCS. + * :meth:`~AbstractDCS.set_ttl`: method for setting TTL value in DCS. + * :meth:`~AbstractDCS.ttl`: property which returns the current TTL. + * :meth:`~AbstractDCS.set_retry_timeout`: method for setting ``retry_timeout`` in DCS backend. + * :meth:`~AbstractDCS._write_leader_optime`: compatibility method to write WAL LSN to DCS. + * :meth:`~AbstractDCS._write_status`: method to write WAL LSN for slots to the DCS. + * :meth:`~AbstractDCS._write_failsafe`: method to write cluster topology to the DCS, used by failsafe mechanism. + * :meth:`~AbstractDCS.touch_member`: method to update individual member key in the DCS. + * :meth:`~AbstractDCS.set_history_value`: method to set the ``history`` key in the DCS. + + DCS setter methods using Compare-And-Set which although important are less critical if they fail, attempts can be + retried or may result in warning log messages: + + * :meth:`~AbstractDCS.set_failover_value`: method to create and/or update the ``failover`` key in the DCS. + * :meth:`~AbstractDCS.set_config_value`: method to create and/or update the ``failover`` key in the DCS. + * :meth:`~AbstractDCS.set_sync_state_value`: method to set the synchronous state ``sync`` key in the DCS. + + DCS data and key removal methods: + + * :meth:`~AbstractDCS.delete_sync_state`: + likewise, a method to remove synchronous state ``sync`` key from the DCS. + * :meth:`~AbstractDCS.delete_cluster`: + method which will remove cluster information from the DCS. Used only from `patronictl`. + * :meth:`~AbstractDCS._delete_leader`: + method relies on CAS, used by a member that is the current leader, to remove the ``leader`` key in the DCS. + * :meth:`~AbstractDCS.cancel_initialization`: + method to remove the ``initialize`` key for the cluster from the DCS. + + If either of the `sync_state` set or delete methods fail, although not critical, this may result in + ``Synchronous replication key updated by someone else`` messages being logged. + + Care should be taken to consult each abstract method for any additional information and requirements such as + expected exceptions that should be raised in certain conditions and the object types for arguments and return from + methods and properties. + """ _INITIALIZE = 'initialize' _CONFIG = 'config' @@ -894,9 +1279,10 @@ class AbstractDCS(abc.ABC): _FAILSAFE = 'failsafe' def __init__(self, config: Dict[str, Any]) -> None: - """ - :param config: dict, reference to config section of selected DCS. - i.e.: `zookeeper` for zookeeper, `etcd` for etcd, etc... + """Prepare DCS paths, Citus group ID, initial values for state information and processing dependencies. + + :ivar config: :class:`dict`, reference to config section of selected DCS. + i.e.: ``zookeeper`` for zookeeper, ``etcd`` for etcd, etc... """ self._name = config['name'] self._base_path = re.sub('/+', '/', '/'.join(['', config.get('namespace', 'service'), config['scope']])) @@ -914,6 +1300,12 @@ class AbstractDCS(abc.ABC): self.event = Event() def client_path(self, path: str) -> str: + """Construct the absolute key name from appropriate parts for the DCS type. + + :param path: The key name within the current Patroni cluster. + + :returns: absolute key name for the current Patroni cluster. + """ components = [self._base_path] if self._citus_group: components.append(self._citus_group) @@ -922,112 +1314,147 @@ class AbstractDCS(abc.ABC): @property def initialize_path(self) -> str: + """Get the client path for ``initialize``.""" return self.client_path(self._INITIALIZE) @property def config_path(self) -> str: + """Get the client path for ``config``.""" return self.client_path(self._CONFIG) @property def members_path(self) -> str: + """Get the client path for ``members``.""" return self.client_path(self._MEMBERS) @property def member_path(self) -> str: + """Get the client path for ``member`` representing this node.""" return self.client_path(self._MEMBERS + self._name) @property def leader_path(self) -> str: + """Get the client path for ``leader``.""" return self.client_path(self._LEADER) @property def failover_path(self) -> str: + """Get the client path for ``failover``.""" return self.client_path(self._FAILOVER) @property def history_path(self) -> str: + """Get the client path for ``history``.""" return self.client_path(self._HISTORY) @property def status_path(self) -> str: + """Get the client path for ``status``.""" return self.client_path(self._STATUS) @property def leader_optime_path(self) -> str: + """Get the client path for ``optime/leader`` (legacy key, superseded by ``status``).""" return self.client_path(self._LEADER_OPTIME) @property def sync_path(self) -> str: + """Get the client path for ``sync``.""" return self.client_path(self._SYNC) @property def failsafe_path(self) -> str: + """Get the client path for ``failsafe``.""" return self.client_path(self._FAILSAFE) @abc.abstractmethod def set_ttl(self, ttl: int) -> Optional[bool]: - """Set the new ttl value for leader key""" + """Set the new *ttl* value for DCS keys.""" @property @abc.abstractmethod def ttl(self) -> int: - """Get new ttl value""" + """Get current ``ttl`` value.""" @abc.abstractmethod def set_retry_timeout(self, retry_timeout: int) -> None: - """Set the new value for retry_timeout""" + """Set the new value for *retry_timeout*.""" def _set_loop_wait(self, loop_wait: int) -> None: + """Set new *loop_wait* value. + + :param loop_wait: value to set. + """ self._loop_wait = loop_wait def reload_config(self, config: Union['Config', Dict[str, Any]]) -> None: + """Load and set relevant values from configuration. + + Sets ``loop_wait``, ``ttl`` and ``retry_timeout`` properties. + + :param config: Loaded configuration information object or dictionary of key value pairs. + """ self._set_loop_wait(config['loop_wait']) self.set_ttl(config['ttl']) self.set_retry_timeout(config['retry_timeout']) @property def loop_wait(self) -> int: + """The recorded value for cluster HA loop wait time in seconds.""" return self._loop_wait @property def last_seen(self) -> int: + """The time recorded when the DCS was last reachable.""" return self._last_seen @abc.abstractmethod def _cluster_loader(self, path: Any) -> Cluster: - """Load and build the `Cluster` object from DCS, which - represents a single Patroni cluster. + """Load and build the :class:`Cluster` object from DCS, which represents a single Patroni or Citus cluster. :param path: the path in DCS where to load Cluster(s) from. - :returns: `Cluster`""" + + :returns: :class:`Cluster` instance. + """ @abc.abstractmethod def _citus_cluster_loader(self, path: Any) -> Union[Cluster, Dict[int, Cluster]]: - """Load and build `Cluster` onjects from DCS that represent all - Patroni clusters from a single Citus cluster. + """Load and build all Patroni clusters from a single Citus cluster. :param path: the path in DCS where to load Cluster(s) from. - :returns: all Citus groups as `dict`, with group ids as keys""" + + :returns: all Citus groups as :class:`dict`, with group IDs as keys and :class:`Cluster` objects as values or a + :class:`Cluster` object representing the coordinator with filled `Cluster.workers` attribute. + """ @abc.abstractmethod def _load_cluster( self, path: str, loader: Callable[[Any], Union[Cluster, Dict[int, Cluster]]] ) -> Union[Cluster, Dict[int, Cluster]]: - """Internally this method should call the `loader` method that - will build `Cluster` object which represents current state and - topology of the cluster in DCS. This method supposed to be - called only by `get_cluster` method. + """Main abstract method that implements the loading of :class:`Cluster` instance. + + .. note:: + Internally this method should call the *loader* method that will build :class:`Cluster` object which + represents current state and topology of the cluster in DCS. This method supposed to be called only by + the :meth:`~AbstractDCS.get_cluster` method. :param path: the path in DCS where to load Cluster(s) from. - :param loader: one of `_cluster_loader` or `_citus_cluster_loader` - :raise: `~DCSError` in case of communication problems with DCS. - If the current node was running as a primary and exception - raised, instance would be demoted.""" + :param loader: one of :meth:`~AbstractDCS._cluster_loader` or :meth:`~AbstractDCS._citus_cluster_loader`. + + :raise: :exc:`~DCSError` in case of communication problems with DCS. If the current node was running as a + primary and exception raised, instance would be demoted. + """ def _bypass_caches(self) -> None: - """Used only in zookeeper""" + """Used only in Zookeeper.""" def __get_patroni_cluster(self, path: Optional[str] = None) -> Cluster: + """Low level method to load a :class:`Cluster` object from DCS. + + :param path: optional client path in DCS backend to load from. + + :returns: a loaded :class:`Cluster` instance. + """ if path is None: path = self.client_path('') cluster = self._load_cluster(path, self._cluster_loader) @@ -1036,15 +1463,32 @@ class AbstractDCS(abc.ABC): return cluster def is_citus_coordinator(self) -> bool: + """:class:`Cluster` instance has a Citus Coordinator group ID. + + :returns: ``True`` if the given node is running as Citus Coordinator (``group=0``). + """ return self._citus_group == str(CITUS_COORDINATOR_GROUP_ID) def get_citus_coordinator(self) -> Optional[Cluster]: + """Load the Patroni cluster for the Citus Coordinator. + + .. note:: + This method is only executed on the worker nodes (``group!=0``) to find the coordinator. + + :returns: Select :class:`Cluster` instance associated with the Citus Coordinator group ID. + """ try: - return self.__get_patroni_cluster('{0}/{1}/'.format(self._base_path, CITUS_COORDINATOR_GROUP_ID)) + return self.__get_patroni_cluster(f'{self._base_path}/{CITUS_COORDINATOR_GROUP_ID}/') except Exception as e: logger.error('Failed to load Citus coordinator cluster from %s: %r', self.__class__.__name__, e) + return None def _get_citus_cluster(self) -> Cluster: + """Load Citus cluster from DCS. + + :returns: A Citus :class:`Cluster` instance for the coordinator with workers clusters in the `Cluster.workers` + dict. + """ groups = self._load_cluster(self._base_path + '/', self._citus_cluster_loader) if isinstance(groups, Cluster): # Zookeeper could return a cached version cluster = groups @@ -1054,6 +1498,18 @@ class AbstractDCS(abc.ABC): return cluster def get_cluster(self, force: bool = False) -> Cluster: + """Retrieve an appropriate cached or fresh view of DCS. + + .. note:: + Stores copy of time, status and failsafe values for comparison in DCS update decisions. + Caching is required to avoid overhead placed upon the REST API. + + Returns either a Citus or Patroni implementation of :class:`Cluster` depending on availability. + + :param force: a value of ``True`` will override Zookeeper caching features. + + :returns: + """ if force: self._bypass_caches() try: @@ -1075,32 +1531,57 @@ class AbstractDCS(abc.ABC): @property def cluster(self) -> Optional[Cluster]: + """Cached DCS cluster information that has not yet expired.""" with self._cluster_thread_lock: return self._cluster if self._cluster_valid_till > time.time() else None def reset_cluster(self) -> None: + """Clear cached state of DCS.""" with self._cluster_thread_lock: self._cluster = None self._cluster_valid_till = 0 @abc.abstractmethod def _write_leader_optime(self, last_lsn: str) -> bool: - """write current WAL LSN into `/optime/leader` key in DCS + """Write current WAL LSN into ``/optime/leader`` key in DCS. - :param last_lsn: absolute WAL LSN in bytes - :returns: `!True` on success.""" + :param last_lsn: absolute WAL LSN in bytes. + + :returns: ``True`` if successfully committed to DCS. + """ def write_leader_optime(self, last_lsn: int) -> None: + """Write value for WAL LSN to ``optime/leader`` key in DCS. + + .. note:: + This method abstracts away the required data structure of :meth:`~Cluster.write_status`, so it + is not needed in the caller. However, the ``optime/leader`` is only written in + :meth:`~Cluster.write_status` when the cluster has members with a Patroni version that + is old enough to require it (i.e. the old Patroni version doesn't understand the new format). + + :param last_lsn: absolute WAL LSN in bytes. + """ self.write_status({self._OPTIME: last_lsn}) @abc.abstractmethod def _write_status(self, value: str) -> bool: - """write current WAL LSN and confirmed_flush_lsn of permanent slots into the `/status` key in DCS + """Write current WAL LSN and ``confirmed_flush_lsn`` of permanent slots into the ``/status`` key in DCS. - :param value: status serialized in JSON forman - :returns: `!True` on success.""" + :param value: status serialized in JSON format. + + :returns: ``True`` if successfully committed to DCS. + """ def write_status(self, value: Dict[str, Any]) -> None: + """Write cluster status to DCS if changed. + + .. note:: + The DCS key ``/status`` was introduced in Patroni version 2.1.0. Previous to this the position of last known + leader LSN was stored in ``optime/leader``. This method has detection for backwards compatibility of members + with a version older than this. + + :param value: JSON serializable dictionary with current WAL LSN and ``confirmed_flush_lsn`` of permanent slots. + """ if not deep_compare(self._last_status, value) and self._write_status(json.dumps(value, separators=(',', ':'))): self._last_status = value cluster = self.cluster @@ -1113,38 +1594,55 @@ class AbstractDCS(abc.ABC): def _write_failsafe(self, value: str) -> bool: """Write current cluster topology to DCS that will be used by failsafe mechanism (if enabled). - :param value: failsafe topology serialized in JSON format - :returns: `!True` on success.""" + :param value: failsafe topology serialized in JSON format. + + :returns: ``True`` if successfully committed to DCS. + """ def write_failsafe(self, value: Dict[str, str]) -> None: - if not (isinstance(self._last_failsafe, dict) and deep_compare(self._last_failsafe, value))\ + """Write the ``/failsafe`` key in DCS. + + :param value: dictionary value to set, consisting of the ``name`` and ``api_url`` of members. + """ + if not (isinstance(self._last_failsafe, dict) and deep_compare(self._last_failsafe, value)) \ and self._write_failsafe(json.dumps(value, separators=(',', ':'))): self._last_failsafe = value @property def failsafe(self) -> Optional[Dict[str, str]]: + """Stored value of :attr:`~AbstractDCS._last_failsafe`.""" return self._last_failsafe @abc.abstractmethod def _update_leader(self, leader: Leader) -> bool: - """Update leader key (or session) ttl + """Update ``leader`` key (or session) ttl. - :param leader: a reference to a current leader key object - :returns: `!True` if leader key (or session) has been updated successfully + .. note:: + You have to use CAS (Compare And Swap) operation in order to update leader key, for example for etcd + ``prevValue`` parameter must be used. - You have to use CAS (Compare And Swap) operation in order to update leader key, - for example for etcd `prevValue` parameter must be used. - If update fails due to DCS not being accessible or because it is not able to - process requests (hopefuly temporary), the ~DCSError exception should be raised.""" + If update fails due to DCS not being accessible or because it is not able to process requests (hopefully + temporary), the :exc:`DCSError` exception should be raised. - def update_leader(self, leader: Leader, last_lsn: Optional[int], - slots: Optional[Dict[str, int]] = None, failsafe: Optional[Dict[str, str]] = None) -> bool: - """Update leader key (or session) ttl and optime/leader + :param leader: a reference to a current :class:`leader` object. - :param last_lsn: absolute WAL LSN in bytes - :param slots: dict with permanent slots confirmed_flush_lsn - :returns: `!True` if leader key (or session) has been updated successfully.""" + :returns: ``True`` if ``leader`` key (or session) has been updated successfully. + """ + def update_leader(self, + leader: Leader, + last_lsn: Optional[int], + slots: Optional[Dict[str, int]] = None, + failsafe: Optional[Dict[str, str]] = None) -> bool: + """Update ``leader`` key (or session) ttl and optime/leader. + + :param leader: :class:`Leader` object with information about the leader. + :param last_lsn: absolute WAL LSN in bytes. + :param slots: dictionary with permanent slots ``confirmed_flush_lsn``. + :param failsafe: if defined dictionary passed to :meth:`~AbstractDCS.write_failsafe`. + + :returns: ``True`` if ``leader`` key (or session) has been updated successfully. + """ ret = self._update_leader(leader) if ret and last_lsn: status: Dict[str, Any] = {self._OPTIME: last_lsn} @@ -1159,22 +1657,41 @@ class AbstractDCS(abc.ABC): @abc.abstractmethod def attempt_to_acquire_leader(self) -> bool: - """Attempt to acquire leader lock - This method should create `/leader` key with value=`~self._name` - :returns: `!True` if key has been created successfully. + """Attempt to acquire leader lock. - Key must be created atomically. In case if key already exists it should not be - overwritten and `!False` must be returned. + .. note:: + This method should create ``/leader`` key with the value :attr:`~AbstractDCS._name`. - If key creation fails due to DCS not being accessible or because it is not able to - process requests (hopefuly temporary), the ~DCSError exception should be raised""" + The key must be created atomically. In case the key already exists it should not be + overwritten and ``False`` must be returned. + + If key creation fails due to DCS not being accessible or because it is not able to + process requests (hopefully temporary), the :exc:`DCSError` exception should be raised. + + :returns: ``True`` if key has been created successfully. + """ @abc.abstractmethod def set_failover_value(self, value: str, version: Optional[Any] = None) -> bool: - """Create or update `/failover` key""" + """Create or update ``/failover`` key. + + :param value: value to set. + :param version: for conditional update of the key/object. + + :returns: ``True`` if successfully committed to DCS. + """ def manual_failover(self, leader: Optional[str], candidate: Optional[str], scheduled_at: Optional[datetime.datetime] = None, version: Optional[Any] = None) -> bool: + """Prepare dictionary with given values and set ``/failover`` key in DCS. + + :param leader: value to set for ``leader``. + :param candidate: value to set for ``member``. + :param scheduled_at: value converted to ISO date format for ``scheduled_at``. + :param version: for conditional update of the key/object. + + :returns: ``True`` if successfully committed to DCS. + """ failover_value = {} if leader: failover_value['leader'] = leader @@ -1188,106 +1705,156 @@ class AbstractDCS(abc.ABC): @abc.abstractmethod def set_config_value(self, value: str, version: Optional[Any] = None) -> bool: - """Create or update `/config` key""" + """Create or update ``/config`` key in DCS. + + :param value: new value to set in the ``config`` key. + :param version: for conditional update of the key/object. + + :returns: ``True`` if successfully committed to DCS. + """ @abc.abstractmethod def touch_member(self, data: Dict[str, Any]) -> bool: """Update member key in DCS. - This method should create or update key with the name = '/members/' + `~self._name` - and value = data in a given DCS. - :param data: information about instance (including connection strings) - :param ttl: ttl for member key, optional parameter. If it is None `~self.member_ttl will be used` - :returns: `!True` on success otherwise `!False` + .. note:: + This method should create or update key with the name with ``/members/`` + :attr:`~AbstractDCS._name` + and the value of *data* in a given DCS. + + :param data: information about an instance (including connection strings). + + :returns: ``True`` if successfully committed to DCS. """ @abc.abstractmethod def take_leader(self) -> bool: - """This method should create leader key with value = `~self._name` and ttl=`~self.ttl` - Since it could be called only on initial cluster bootstrap it could create this key regardless, - overwriting the key if necessary.""" + """Establish a new leader in DCS. + + .. note:: + This method should create leader key with value of :attr:`~AbstractDCS._name` and ``ttl`` of + :attr:`~AbstractDCS.ttl`. + + Since it could be called only on initial cluster bootstrap it could create this key regardless, + overwriting the key if necessary. + + :returns: ``True`` if successfully committed to DCS. + """ @abc.abstractmethod def initialize(self, create_new: bool = True, sysid: str = "") -> bool: """Race for cluster initialization. - :param create_new: False if the key should already exist (in the case we are setting the system_id) - :param sysid: PostgreSQL cluster system identifier, if specified, is written to the key - :returns: `!True` if key has been created successfully. + This method should atomically create ``initialize`` key and return ``True``, + otherwise it should return ``False``. - this method should create atomically initialize key and return `!True` - otherwise it should return `!False`""" + :param create_new: ``False`` if the key should already exist (in the case we are setting the system_id). + :param sysid: PostgreSQL cluster system identifier, if specified, is written to the key. + + :returns: ``True`` if key has been created successfully. + """ @abc.abstractmethod def _delete_leader(self) -> bool: """Remove leader key from DCS. - This method should remove leader key if current instance is the leader""" + + This method should remove leader key if current instance is the leader. + + :returns: ``True`` if successfully committed to DCS. + """ def delete_leader(self, last_lsn: Optional[int] = None) -> bool: - """Update optime/leader and voluntarily remove leader key from DCS. - This method should remove leader key if current instance is the leader. - :param last_lsn: latest checkpoint location in bytes""" + """Update ``optime/leader`` and voluntarily remove leader key from DCS. + This method should remove leader key if current instance is the leader. + + :param last_lsn: latest checkpoint location in bytes. + + :returns: boolean result of called abstract :meth:`~AbstractDCS._delete_leader`. + """ if last_lsn: self.write_status({self._OPTIME: last_lsn}) return self._delete_leader() @abc.abstractmethod def cancel_initialization(self) -> bool: - """ Removes the initialize key for a cluster """ + """Removes the ``initialize`` key for a cluster. + + :returns: ``True`` if successfully committed to DCS. + """ @abc.abstractmethod def delete_cluster(self) -> bool: - """Delete cluster from DCS""" + """Delete cluster from DCS. + + :returns: ``True`` if successfully committed to DCS. + """ @staticmethod def sync_state(leader: Optional[str], sync_standby: Optional[Collection[str]]) -> Dict[str, Any]: - """Build sync_state dict. - The sync_standby key being kept for backward compatibility. - :param leader: name of the leader node that manages /sync key - :param sync_standby: collection of currently known synchronous standby node names - :returns: dictionary that later could be serialized to JSON or saved directly to DCS + """Build ``sync_state`` dictionary. + + :param leader: name of the leader node that manages ``/sync`` key. + :param sync_standby: collection of currently known synchronous standby node names. + + :returns: dictionary that later could be serialized to JSON or saved directly to DCS. """ return {'leader': leader, 'sync_standby': ','.join(sorted(sync_standby)) if sync_standby else None} def write_sync_state(self, leader: Optional[str], sync_standby: Optional[Collection[str]], version: Optional[Any] = None) -> Optional[SyncState]: """Write the new synchronous state to DCS. - Calls :func:`sync_state` method to build a dict and than calls DCS specific :func:`set_sync_state_value` method. - :param leader: name of the leader node that manages /sync key - :param sync_standby: collection of currently known synchronous standby node names - :param version: for conditional update of the key/object - :returns: the new :class:`SyncState` object or None + + Calls :meth:`~AbstractDCS.sync_state` to build a dictionary and then calls DCS specific + :meth:`~AbstractDCS.set_sync_state_value`. + + :param leader: name of the leader node that manages ``/sync`` key. + :param sync_standby: collection of currently known synchronous standby node names. + :param version: for conditional update of the key/object. + + :returns: the new :class:`SyncState` object or ``None``. """ sync_value = self.sync_state(leader, sync_standby) ret = self.set_sync_state_value(json.dumps(sync_value, separators=(',', ':')), version) if not isinstance(ret, bool): return SyncState.from_node(ret, sync_value) + return None @abc.abstractmethod def set_history_value(self, value: str) -> bool: - """""" + """Set value for ``history`` in DCS. + + :param value: new value of ``history`` key/object. + + :returns: ``True`` if successfully committed to DCS. + """ @abc.abstractmethod def set_sync_state_value(self, value: str, version: Optional[Any] = None) -> Union[Any, bool]: - """Set synchronous state in DCS, should be implemented in the child class. + """Set synchronous state in DCS. - :param value: the new value of /sync key - :param version: for conditional update of the key/object - :returns: version of the new object or `False` in case of error + :param value: the new value of ``/sync`` key. + :param version: for conditional update of the key/object. + + :returns: *version* of the new object or ``False`` in case of error. """ @abc.abstractmethod def delete_sync_state(self, version: Optional[Any] = None) -> bool: - """""" + """Delete the synchronous state from DCS. + + :param version: for conditional deletion of the key/object. + + :returns: ``True`` if delete successful. + """ def watch(self, leader_version: Optional[Any], timeout: float) -> bool: - """If the current node is a leader it should just sleep. - Any other node should watch for changes of leader key with a given timeout + """Sleep if the current node is a leader, otherwise, watch for changes of leader key with a given *timeout*. - :param leader_version: version of a leader key - :param timeout: timeout in seconds - :returns: `!True` if you would like to reschedule the next run of ha cycle""" + :param leader_version: version of a leader key. + :param timeout: timeout in seconds. + :returns: if ``True`` this will reschedule the next run of the HA cycle. + """ + _ = leader_version self.event.wait(timeout) return self.event.is_set()