diff --git a/.github/workflows/tests.yaml b/.github/workflows/tests.yaml index 355ea927..0540cca0 100644 --- a/.github/workflows/tests.yaml +++ b/.github/workflows/tests.yaml @@ -173,4 +173,4 @@ jobs: - uses: jakebailey/pyright-action@v1 with: - version: 1.1.317 + version: 1.1.320 diff --git a/docs/dynamic_configuration.rst b/docs/dynamic_configuration.rst index 5486bf6e..f5bb4ee5 100644 --- a/docs/dynamic_configuration.rst +++ b/docs/dynamic_configuration.rst @@ -46,9 +46,9 @@ In order to change the dynamic configuration you can use either ``patronictl edi - **archive\_cleanup\_command**: cleanup command for standby leader - **recovery\_min\_apply\_delay**: how long to wait before actually apply WAL records on a standby leader -- **slots**: define permanent replication slots. These slots will be preserved during switchover/failover. The logical slots are copied from the primary to a standby with restart, and after that their position advanced every **loop_wait** seconds (if necessary). Copying logical slot files performed via ``libpq`` connection and using either rewind or superuser credentials (see **postgresql.authentication** section). There is always a chance that the logical slot position on the replica is a bit behind the former primary, therefore application should be prepared that some messages could be received the second time after the failover. The easiest way of doing so - tracking ``confirmed_flush_lsn``. Enabling permanent logical replication slots requires **postgresql.use_slots** to be set and will also automatically enable the ``hot_standby_feedback``. Since the failover of logical replication slots is unsafe on PostgreSQL 9.6 and older and PostgreSQL version 10 is missing some important functions, the feature only works with PostgreSQL 11+. +- **slots**: define permanent replication slots. These slots will be preserved during switchover/failover. Permanent slots that don't exist will be created by Patroni. The physical slots are maintained only in the current primary. The logical slots are copied from the primary to a standby with restart, and after that their position advanced every **loop_wait** seconds (if necessary). Copying logical slot files performed via ``libpq`` connection and using either rewind or superuser credentials (see **postgresql.authentication** section). There is always a chance that the logical slot position on the replica is a bit behind the former primary, therefore application should be prepared that some messages could be received the second time after the failover. The easiest way of doing so - tracking ``confirmed_flush_lsn``. Enabling permanent logical replication slots requires **postgresql.use_slots** to be set and will also automatically enable the ``hot_standby_feedback``. Since the failover of logical replication slots is unsafe on PostgreSQL 9.6 and older and PostgreSQL version 10 is missing some important functions, the feature only works with PostgreSQL 11+. - - **my\_slot\_name**: the name of replication slot. If the permanent slot name matches with the name of the current primary it will not be created. Everything else is the responsibility of the operator to make sure that there are no clashes in names between replication slots automatically created by Patroni for members and permanent replication slots. + - **my\_slot\_name**: the name of the permanent replication slot. If the permanent slot name matches with the name of the current leader it will not be created. Please note that Patroni does not make checks for permanent slot names added to this configuration matching those that Patroni creates automatically for members. If those names are added, Patroni will ensure that any slots that were created are not removed even if the member becomes unresponsive, situation which would normally result in the slot's removal by Patroni. Although this can be useful in some situations, such as when importing existing members to a new Patroni cluster (see :ref:`Convert a Standalone to a Patroni Cluster ` for details), caution should be exercised by the operator that these clashes in names are not persisted in the DCS due to its effect on normal functioning of Patroni. - **type**: slot type. Could be ``physical`` or ``logical``. If the slot is logical, you have to additionally define ``database`` and ``plugin``. - **database**: the database name where logical slots should be created. diff --git a/docs/existing_data.rst b/docs/existing_data.rst index 7c78f6e3..cb07bfa9 100644 --- a/docs/existing_data.rst +++ b/docs/existing_data.rst @@ -10,18 +10,58 @@ To deploy a Patroni cluster without using a pre-existing PostgreSQL instance, se Procedure --------- -A Patroni cluster can be started with a data directory from a single-node PostgreSQL database. This is achieved by following closely these steps: +You can find below an overview of steps for converting an existing Postgres cluster to a Patroni managed cluster. In the steps we assume all nodes that are part of the existing cluster are currently up and running, and that you *do not* intend to change Postgres configuration while the migration is ongoing. The steps: -1. Manually start PostgreSQL daemon -2. Create Patroni superuser and replication users as defined in the :ref:`authentication ` section of the Patroni configuration. If this user is created in SQL, the following queries achieve this: +#. Create the Postgres users as explained for :ref:`authentication ` section of the Patroni configuration. You can find sample SQL commands to create the users in the code block below, in which you need to replace the usernames and passwords as per your environment. If you already have the relevant users, then you can skip this step. -.. code-block:: sql + .. code-block:: sql - CREATE USER $PATRONI_SUPERUSER_USERNAME WITH SUPERUSER ENCRYPTED PASSWORD '$PATRONI_SUPERUSER_PASSWORD'; - CREATE USER $PATRONI_REPLICATION_USERNAME WITH REPLICATION ENCRYPTED PASSWORD '$PATRONI_REPLICATION_PASSWORD'; + -- Patroni superuser + -- Replace PATRONI_SUPERUSER_USERNAME and PATRONI_SUPERUSER_PASSWORD accordingly + CREATE USER PATRONI_SUPERUSER_USERNAME WITH SUPERUSER ENCRYPTED PASSWORD 'PATRONI_SUPERUSER_PASSWORD'; -3. Start Patroni (e.g. ``patroni /etc/patroni/patroni.yml``). It automatically detects that PostgreSQL daemon is already running but its configuration might be out-of-date. -4. Ask Patroni to restart the node with ``patronictl restart cluster-name node-name``. This step is only required if PostgreSQL configuration is out-of-date. + -- Patroni replication user + -- Replace PATRONI_REPLICATION_USERNAME and PATRONI_REPLICATION_PASSWORD accordingly + CREATE USER PATRONI_REPLICATION_USERNAME WITH REPLICATION ENCRYPTED PASSWORD 'PATRONI_REPLICATION_PASSWORD'; + + -- Patroni rewind user, if you intend to enable use_pg_rewind in your Patroni configuration + -- Replace PATRONI_REWIND_USERNAME and PATRONI_REWIND_PASSWORD accordingly + CREATE USER PATRONI_REWIND_USERNAME WITH ENCRYPTED PASSWORD 'PATRONI_REWIND_PASSWORD'; + GRANT EXECUTE ON function pg_catalog.pg_ls_dir(text, boolean, boolean) TO PATRONI_REWIND_USERNAME; + GRANT EXECUTE ON function pg_catalog.pg_stat_file(text, boolean) TO PATRONI_REWIND_USERNAME; + GRANT EXECUTE ON function pg_catalog.pg_read_binary_file(text) TO PATRONI_REWIND_USERNAME; + GRANT EXECUTE ON function pg_catalog.pg_read_binary_file(text, bigint, bigint, boolean) TO PATRONI_REWIND_USERNAME; + +#. Perform the following steps on all Postgres nodes. Perform all steps on one node before proceeding with the next node. Start with the primary node, then proceed with each standby node: + + #. If you are running Postgres through systemd, then disable the Postgres systemd unit. This is performed as Patroni manages starting and stopping the Postgres daemon. + + #. Create a YAML configuration file for Patroni. + + * **Note (specific for the primary node):** If you have replication slots being used for replication between cluster members, then it is recommended that you enable ``use_slots`` and configure the existing replication slots as permanent via the ``slots`` configuration item. Be aware that Patroni automatically creates replication slots for replication between members, and drops replication slots that it does not recognize, when ``use_slots`` is enabled. The idea of using permanent slots here is to allow your existing slots to persist while the migration to Patroni is in progress. See :ref:`YAML Configuration Settings ` for details. + + #. Start Patroni using the ``patroni`` systemd service unit. It automatically detects that Postgres is already running and starts monitoring the instance. + +#. Hand over Postgres "start up procedure" to Patroni. In order to do that you need to restart the cluster members through ``patronictl restart cluster-name member-name`` command. For minimal downtime you might want to split this step into: + + #. Immediate restart of the standby nodes. + #. Scheduled restart of the primary node within a maintenance window. + +#. If you configured permanent slots in step ``1.2.``, then you should remove them from ``slots`` configuration through ``patronictl edit-config cluster-name member-name`` command once the ``restart_lsn`` of the slots created by Patroni is able to catch up with the ``restart_lsn`` of the original slots for the corresponding members. By removing the slots from ``slots`` configuration you will allow Patroni to drop the original slots from your cluster once they are not needed anymore. You can find below an example query to check the ``restart_lsn`` of a couple slots, so you can compare them: + + .. code-block:: sql + + -- Assume original_slot_for_member_x is the name of the slot in your original + -- cluster for replicating changes to member X, and slot_for_member_x is the + -- slot created by Patroni for that purpose. You need restart_lsn of + -- slot_for_member_x to be >= restart_lsn of original_slot_for_member_x + SELECT slot_name, + restart_lsn + FROM pg_replication_slots + WHERE slot_name IN ( + 'original_slot_for_member_x', + 'slot_for_member_x' + ) .. _major_upgrade: @@ -30,14 +70,14 @@ Major Upgrade of PostgreSQL Version The only possible way to do a major upgrade currently is: -1. Stop Patroni -2. Upgrade PostgreSQL binaries and perform `pg_upgrade `_ on the primary node -3. Update patroni.yml -4. Remove the initialize key from DCS or wipe complete cluster state from DCS. The second one could be achieved by running ``patronictl remove ``. It is necessary because pg_upgrade runs initdb which actually creates a new database with a new PostgreSQL system identifier. -5. If you wiped the cluster state in the previous step, you may wish to copy patroni.dynamic.json from old data dir to the new one. It will help you to retain some PostgreSQL parameters you had set before. -6. Start Patroni on the primary node. -7. Upgrade PostgreSQL binaries, update patroni.yml and wipe the data_dir on standby nodes. -8. Start Patroni on the standby nodes and wait for the replication to complete. +#. Stop Patroni +#. Upgrade PostgreSQL binaries and perform `pg_upgrade `_ on the primary node +#. Update patroni.yml +#. Remove the initialize key from DCS or wipe complete cluster state from DCS. The second one could be achieved by running ``patronictl remove ``. It is necessary because pg_upgrade runs initdb which actually creates a new database with a new PostgreSQL system identifier. +#. If you wiped the cluster state in the previous step, you may wish to copy patroni.dynamic.json from old data dir to the new one. It will help you to retain some PostgreSQL parameters you had set before. +#. Start Patroni on the primary node. +#. Upgrade PostgreSQL binaries, update patroni.yml and wipe the data_dir on standby nodes. +#. Start Patroni on the standby nodes and wait for the replication to complete. Running pg_upgrade on standby nodes is not supported by PostgreSQL. If you know what you are doing, you can try the rsync procedure described in https://www.postgresql.org/docs/current/pgupgrade.html instead of wiping data_dir on standby nodes. The safest way is however to let Patroni replicate the data for you. diff --git a/docs/kubernetes.rst b/docs/kubernetes.rst index ca7313bf..b9456f2a 100644 --- a/docs/kubernetes.rst +++ b/docs/kubernetes.rst @@ -32,8 +32,11 @@ Configuration Patroni Kubernetes :ref:`settings ` and :ref:`environment variables ` are described in the general chapters of the documentation. +.. _kubernetes_role_values: + Customize role label ^^^^^^^^^^^^^^^^^^^^ + By default, Patroni will set corresponding labels on the pod it runs in based on node's role, such as ``role=master``. The key and value of label can be customized by `kubernetes.role_label`, `kubernetes.leader_label_value`, `kubernetes.follower_label_value` and `kubernetes.standby_leader_label_value`. diff --git a/docs/releases.rst b/docs/releases.rst index 48cad683..7a05bf7e 100644 --- a/docs/releases.rst +++ b/docs/releases.rst @@ -3,6 +3,84 @@ Release notes ============= +Version 3.1.0 +------------- + +**Breaking changes** + +- Changed semantic of ``restapi.keyfile`` and ``restapi.certfile`` (Alexander Kukushkin) + + Previously Patroni was using ``restapi.keyfile`` and ``restapi.certfile`` as client certificates as a fallback if there were no respective configuration parameters in the ``ctl`` section. + +.. warning:: + If you enabled client certificates validation (``restapi.verify_client`` is set to ``required``), you also **must** provide **valid client certificates** in the ``ctl.certfile``, ``ctl.keyfile``, ``ctl.keyfile_password``. If not provided, Patroni will not work correctly. + + +**New features** + +- Make Pod role label configurable (Waynerv) + + Values could be customized using ``kubernetes.leader_label_value``, ``kubernetes.follower_label_value`` and ``kubernetes.standby_leader_label_value`` parameters. This feature will be very useful when we change the ``master`` role to the ``primary``. You can read more about the feature and migration steps :ref:`here `. + + +**Improvements** + +- Various improvements of ``patroni --validate-config`` (Alexander Kukushkin) + + Improved parameter validation for different DCS, ``bootstrap.dcs`` , ``ctl``, ``restapi``, and ``watchdog`` sections. + +- Start Postgres not in recovery if it crashed during recovery while Patroni is running (Alexander Kukushkin) + + It may reduce recovery time and will help to prevent unnecessary timeline increments. + +- Avoid unnecessary updates of ``/status`` key (Alexander Kukushkin) + + When there are no permanent logical slots Patroni was updating the ``/status`` on every heartbeat loop even when LSN on the primary didn't move forward. + +- Don't allow stale primary to win the leader race (Alexander Kukushkin) + + If Patroni was hanging during a significant time due to lack of resources it will additionally check that no other nodes promoted Postgres before acquiring the leader lock. + +- Implemented visibility of certain PostgreSQL parameters validation (Alexander Kukushkin, Feike Steenbergen) + + If validation of ``max_connections``, ``max_wal_senders``, ``max_prepared_transactions``, ``max_locks_per_transaction``, ``max_replication_slots``, or ``max_worker_processes`` failed Patroni was using some sane default value. Now in addition to that it will also show a warning. + +- Set permissions for files and directories created in ``PGDATA`` (Alexander Kukushkin) + + All files created by Patroni had only owner read/write permissions. This behaviour was breaking backup tools that run under a different user and relying on group read permissions. Now Patroni honors permissions on ``PGDATA`` and correctly sets permissions on all directories and files it creates inside ``PGDATA``. + + +**Bugfixes** + +- Run ``archive_command`` through shell (Waynerv) + + Patroni might archive some WAL segments before doing crash recovery in a single-user mode or before ``pg_rewind``. If the archive_command contains some shell operators, like ``&&`` it didn't work with Patroni. + +- Fixed "on switchover" shutdown checks (Polina Bungina) + + It was possible that specified candidate is still streaming and didn't received shut down checking but the leader key was removed because some other nodes were healthy. + +- Fixed "is primary" check (Alexander Kukushkin) + + During the leader race replicas were not able to recognize that Postgres on the old leader is still running as a primary. + +- Fixed ``patronictl list`` (Alexander Kukushkin) + + The Cluster name field was missing in ``tsv``, ``json``, and ``yaml`` output formats. + +- Fixed ``pg_rewind`` behaviour after pause (Alexander Kukushkin) + + Under certain conditions, Patroni wasn't able to join the false primary back to the cluster with ``pg_rewind`` after coming out of maintenance mode. + +- Fixed bug in Etcd v3 implementation (Alexander Kukushkin) + + Invalidate internal KV cache if key update performed using ``create_revision``/``mod_revision`` field due to revision mismatch. + +- Fixed behaviour of replicas in standby cluster in pause (Alexander Kukushkin) + + When the leader key expires replicas in standby cluster will not follow the remote node but keep ``primary_conninfo`` as it is. + + Version 3.0.4 ------------- diff --git a/patroni/__main__.py b/patroni/__main__.py index b365bfe2..42f539d1 100644 --- a/patroni/__main__.py +++ b/patroni/__main__.py @@ -74,7 +74,7 @@ class Patroni(AbstractPatroniDaemon): if not isinstance(member, Member): return try: - _ = self.request(member, endpoint="/liveness") + _ = self.request(member, endpoint="/liveness", timeout=3) logger.fatal("Can't start; there is already a node named '%s' running", self.config['name']) sys.exit(1) except Exception: diff --git a/patroni/config.py b/patroni/config.py index facf16fd..d2770517 100644 --- a/patroni/config.py +++ b/patroni/config.py @@ -13,6 +13,7 @@ from . import PATRONI_ENV_PREFIX from .collections import CaseInsensitiveDict from .dcs import ClusterConfig, Cluster from .exceptions import ConfigParseError +from .file_perm import pg_perm from .postgresql.config import ConfigHandler from .utils import deep_compare, parse_bool, parse_int, patch_config @@ -275,11 +276,13 @@ class Config(object): if self._cache_needs_saving: tmpfile = fd = None try: + pg_perm.set_permissions_from_data_directory(self._data_dir) (fd, tmpfile) = tempfile.mkstemp(prefix=self.__CACHE_FILENAME, dir=self._data_dir) with os.fdopen(fd, 'w') as f: fd = None json.dump(self.dynamic_configuration, f) tmpfile = shutil.move(tmpfile, self._cache_file) + os.chmod(self._cache_file, pg_perm.file_create_mode) self._cache_needs_saving = False except Exception: logger.exception('Exception when saving file: %s', self._cache_file) diff --git a/patroni/ctl.py b/patroni/ctl.py index e7739965..fabb4077 100644 --- a/patroni/ctl.py +++ b/patroni/ctl.py @@ -36,7 +36,7 @@ from collections import defaultdict from contextlib import contextmanager from prettytable import ALL, FRAME, PrettyTable from urllib.parse import urlparse -from typing import Any, Dict, Generator, Iterator, List, Optional, Union, Tuple, TYPE_CHECKING +from typing import Any, Dict, Iterator, List, Optional, Union, Tuple, TYPE_CHECKING if TYPE_CHECKING: # pragma: no cover from psycopg import Cursor from psycopg2 import cursor @@ -1824,7 +1824,7 @@ def resume(obj: Dict[str, Any], cluster_name: str, group: Optional[int], wait: b @contextmanager -def temporary_file(contents: bytes, suffix: str = '', prefix: str = 'tmp') -> Generator[str, None, None]: +def temporary_file(contents: bytes, suffix: str = '', prefix: str = 'tmp') -> Iterator[str]: """Create a temporary file with specified contents that persists for the context. :param contents: binary string that will be written to the file. diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index 29a4d766..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( @@ -170,30 +191,53 @@ _Version = Union[int, str] _Session = Union[int, float, str, None] -class Member(NamedTuple): +class Member(NamedTuple('Member', + [('version', _Version), + ('name', str), + ('session', _Session), + ('data', Dict[str, Any])])): """Immutable object (namedtuple) which represents single member of PostgreSQL cluster. - Consists of the following fields: - :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 data: arbitrary data i.e. conn_url, api_url, xlog location, state, role, tags, etc... - There are two mandatory keys in a data: - conn_url: connection string containing host, user and password which could be used to access this member. - api_url: REST API url of patroni instance + .. note:: + We are using an old-style attribute declaration here because otherwise it is not possible to override + ``__new__`` method in the :class:`RemoteMember` class. + + .. note:: + These two keys in data are always written to the DCS, but care is taken to maintain consistency and resilience + from data that is read: + + ``conn_url``: connection string containing host, user and password which could be used to access this member. + ``api_url``: REST API url of patroni instance + + Consists of the following fields: + + :ivar version: modification version of a given member key in a Configuration Store. + :ivar name: name of PostgreSQL cluster member. + :ivar session: either session id or just ttl in seconds. + :ivar data: dictionary containing arbitrary data i.e. ``conn_url``, ``api_url``, ``xlog_location``, ``state``, + ``role``, ``tags``, etc... """ - version: _Version - name: str - session: _Session - data: Dict[str, Any] @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) @@ -208,6 +252,7 @@ class Member(NamedTuple): @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 @@ -218,7 +263,21 @@ class Member(NamedTuple): 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, @@ -249,40 +308,56 @@ class Member(NamedTuple): @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): @@ -300,79 +375,129 @@ class RemoteMember(Member): 'no_replication_slot' ) - @classmethod - def from_name_and_data(cls, name: str, data: Dict[str, Any]) -> 'RemoteMember': + def __new__(cls, name: str, data: Dict[str, Any]) -> 'RemoteMember': + """Factory method to construct instance from given *name* and *data*. + + :param name: name of the remote member. + :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] @@ -380,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: @@ -402,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) @@ -427,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): @@ -476,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: @@ -527,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()) @@ -535,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) @@ -555,80 +787,149 @@ class TimelineHistory(NamedTuple): return TimelineHistory(version, value, lines) -class Cluster(NamedTuple): - """Immutable object (namedtuple) which represents PostgreSQL cluster. +class Cluster(NamedTuple('Cluster', + [('initialize', Optional[str]), + ('config', Optional[ClusterConfig]), + ('leader', Optional[Leader]), + ('last_lsn', int), + ('members', List[Member]), + ('failover', Optional[Failover]), + ('sync', SyncState), + ('history', Optional[TimelineHistory]), + ('slots', Optional[Dict[str, int]]), + ('failsafe', Optional[Dict[str, str]]), + ('workers', Dict[int, 'Cluster'])])): + """Immutable object (namedtuple) which represents PostgreSQL or Citus cluster. + + .. note:: + We are using an old-style attribute declaration here because otherwise it is not possible to override `__new__` + method. Without it the *workers* by default gets always the same :class:`dict` object that could be mutated. + Consists of the following fields: - :param initialize: shows whether this cluster has initialization key stored in DC or not. - :param config: global dynamic configuration, reference to `ClusterConfig` object - :param leader: `Leader` object which represents current leader of the cluster - :param last_lsn: int or long object containing position of last known leader LSN. - This value is stored in the `/status` key or `/optime/leader` (legacy) key - :param members: list of Member object, all PostgreSQL cluster members including leader - :param failover: reference to `Failover` object - :param sync: reference to `SyncState` object, last observed synchronous replication state. - :param history: reference to `TimelineHistory` object - :param slots: state of permanent logical replication slots on the primary in the format: {"slot_name": int} - :param failsafe: failsafe topology. Node is allowed to become the leader only if its name is found in this list. - :param workers: workers of the Citus cluster, optional. Format: {int(group): Cluster()} + + :ivar initialize: shows whether this cluster has initialization key stored in DC or not. + :ivar config: global dynamic configuration, reference to `ClusterConfig` object. + :ivar leader: :class:`Leader` object which represents current leader of the cluster. + :ivar last_lsn: :class:int object containing position of last known leader LSN. + This value is stored in the `/status` key or `/optime/leader` (legacy) key. + :ivar members: list of:class:` Member` objects, all PostgreSQL cluster members including leader + :ivar failover: reference to :class:`Failover` object. + :ivar sync: reference to :class:`SyncState` object, last observed synchronous replication state. + :ivar history: reference to `TimelineHistory` object. + :ivar slots: state of permanent logical replication slots on the primary in the format: {"slot_name": int}. + :ivar failsafe: failsafe topology. Node is allowed to become the leader only if its name is found in this list. + :ivar workers: dictionary of workers of the Citus cluster, optional. Each key is an :class:`int` representing + the group, and the corresponding value is a :class:`Cluster` instance. """ - initialize: Optional[str] - config: Optional[ClusterConfig] - leader: Optional[Leader] - last_lsn: int - members: List[Member] - failover: Optional[Failover] - sync: SyncState - history: Optional[TimelineHistory] - slots: Optional[Dict[str, int]] - failsafe: Optional[Dict[str, str]] - workers: Dict[int, 'Cluster'] = {} + + def __new__(cls, *args: Any, **kwargs: Any): + """Make workers argument optional and set it to an empty dict object.""" + if len(args) < len(cls._fields) and 'workers' not in kwargs: + kwargs['workers'] = {} + return super(Cluster, cls).__new__(cls, *args, **kwargs) @staticmethod def empty() -> 'Cluster': - return Cluster(None, None, None, 0, [], None, SyncState.empty(), None, None, None) + """Produce an empty :class:`Cluster` instance.""" + 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, @@ -777,17 +1078,32 @@ class Cluster(NamedTuple): 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 @@ -800,25 +1116,48 @@ class Cluster(NamedTuple): 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: @@ -832,14 +1171,22 @@ class Cluster(NamedTuple): @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) @@ -850,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' @@ -864,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']])) @@ -884,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) @@ -892,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) @@ -1006,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 @@ -1024,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: @@ -1045,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 @@ -1083,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} @@ -1129,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 @@ -1158,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() diff --git a/patroni/dcs/etcd3.py b/patroni/dcs/etcd3.py index e28da844..e5e069c5 100644 --- a/patroni/dcs/etcd3.py +++ b/patroni/dcs/etcd3.py @@ -228,7 +228,7 @@ class Etcd3Client(AbstractEtcdClientWithFailover): return self.http.urlopen def _handle_server_response(self, response: urllib3.response.HTTPResponse) -> Dict[str, Any]: - data: Union[bytes, str] = response.data + data = response.data try: data = data.decode('utf-8') ret: Dict[str, Any] = json.loads(data) diff --git a/patroni/dcs/kubernetes.py b/patroni/dcs/kubernetes.py index 96097338..4a66df71 100644 --- a/patroni/dcs/kubernetes.py +++ b/patroni/dcs/kubernetes.py @@ -134,6 +134,8 @@ class K8sConfig(object): config: Dict[str, Any] = yaml.safe_load(f) context = context or config['current-context'] + if TYPE_CHECKING: # pragma: no cover + assert isinstance(context, str) context_value = self._get_by_name(config, 'context', context) if TYPE_CHECKING: # pragma: no cover assert isinstance(context_value, dict) diff --git a/patroni/file_perm.py b/patroni/file_perm.py new file mode 100644 index 00000000..ed9c4e67 --- /dev/null +++ b/patroni/file_perm.py @@ -0,0 +1,95 @@ +"""Helper object that helps with figuring out file and directory permissions based on permissions of PGDATA. + +:var logger: logger of this module. +:var pg_perm: instance of the :class:`__FilePermissions` object. +""" +import logging +import os +import stat + +logger = logging.getLogger(__name__) + + +class __FilePermissions: + """Helper class for managing permissions of directories and files under PGDATA. + + Execute :meth:`set_permissions_from_data_directory` to figure out which permissions should be used for files and + directories under PGDATA based on permissions of PGDATA root directory. + """ + + # Mode mask for data directory permissions that only allows the owner to + # read/write directories and files -- mask 077. + __PG_MODE_MASK_OWNER = stat.S_IRWXG | stat.S_IRWXO + + # Mode mask for data directory permissions that also allows group read/execute -- mask 027. + __PG_MODE_MASK_GROUP = stat.S_IWGRP | stat.S_IRWXO + + # Default mode for creating directories -- mode 700. + __PG_DIR_MODE_OWNER = stat.S_IRWXU + + # Mode for creating directories that allows group read/execute -- mode 750. + __PG_DIR_MODE_GROUP = stat.S_IRWXU | stat.S_IRGRP | stat.S_IXGRP + + # Default mode for creating files -- mode 600. + __PG_FILE_MODE_OWNER = stat.S_IRUSR | stat.S_IWUSR + + # Mode for creating files that allows group read -- mode 640. + __PG_FILE_MODE_GROUP = stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP + + def __init__(self) -> None: + """Create a :class:`__FilePermissions` object and set default permissions.""" + self.__set_owner_permissions() + self.__set_umask() + + def __set_umask(self) -> None: + """Set umask value based on calculations. + + .. note:: + Should only be called once either :meth:`__set_owner_permissions` + or :meth:`__set_group_permissions` has been executed. + """ + try: + os.umask(self.__pg_mode_mask) + except Exception as e: + logger.error('Can not set umask to %03o: %r', self.__pg_mode_mask, e) + + def __set_owner_permissions(self) -> None: + """Make directories/files accessible only by the owner.""" + self.__pg_dir_create_mode = self.__PG_DIR_MODE_OWNER + self.__pg_file_create_mode = self.__PG_FILE_MODE_OWNER + self.__pg_mode_mask = self.__PG_MODE_MASK_OWNER + + def __set_group_permissions(self) -> None: + """Make directories/files accessible by the owner and readable by group.""" + self.__pg_dir_create_mode = self.__PG_DIR_MODE_GROUP + self.__pg_file_create_mode = self.__PG_FILE_MODE_GROUP + self.__pg_mode_mask = self.__PG_MODE_MASK_GROUP + + def set_permissions_from_data_directory(self, data_dir: str) -> None: + """Set new permissions based on provided *data_dir*. + + :param data_dir: reference to PGDATA to calculate permissions from. + """ + try: + st = os.stat(data_dir) + if (st.st_mode & self.__PG_DIR_MODE_GROUP) == self.__PG_DIR_MODE_GROUP: + self.__set_group_permissions() + else: + self.__set_owner_permissions() + except Exception as e: + logger.error('Can not check permissions on %s: %r', data_dir, e) + else: + self.__set_umask() + + @property + def dir_create_mode(self) -> int: + """Directory permissions.""" + return self.__pg_dir_create_mode + + @property + def file_create_mode(self) -> int: + """File permissions.""" + return self.__pg_file_create_mode + + +pg_perm = __FilePermissions() diff --git a/patroni/ha.py b/patroni/ha.py index 115118d5..3cea58d4 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -83,11 +83,7 @@ class Failsafe(object): def __init__(self, dcs: AbstractDCS) -> None: self._lock = RLock() self._dcs = dcs - self._last_update = 0 - self._name = None - self._conn_url = None - self._api_url = None - self._slots = None + self._reset_state() def update(self, data: Dict[str, Any]) -> None: with self._lock: @@ -97,13 +93,20 @@ class Failsafe(object): self._api_url = data['api_url'] self._slots = data.get('slots') + def _reset_state(self) -> None: + self._last_update = 0 + self._name = None + self._conn_url = None + self._api_url = None + self._slots = None + @property def leader(self) -> Optional[Leader]: with self._lock: if self._last_update + self._dcs.ttl > time.time() and self._name: - return Leader('', '', RemoteMember.from_name_and_data(self._name, {'api_url': self._api_url, - 'conn_url': self._conn_url, - 'slots': self._slots})) + return Leader('', '', RemoteMember(self._name, {'api_url': self._api_url, + 'conn_url': self._conn_url, + 'slots': self._slots})) def update_cluster(self, cluster: Cluster) -> Cluster: # Enreach cluster with the real leader if there was a ping from it @@ -130,6 +133,8 @@ class Failsafe(object): def set_is_active(self, value: float) -> None: with self._lock: self._last_update = value + if not value: + self._reset_state() class Ha(object): @@ -195,6 +200,13 @@ class Ha(object): with self._is_leader_lock: self._is_leader = time.time() + self.dcs.ttl if value else 0 + def sync_mode_is_active(self) -> bool: + """Check whether synchronous replication is requested and already active. + + :returns: ``True`` if the primary already put its name into the ``/sync`` in DCS. + """ + return self.is_synchronous_mode() and not self.cluster.sync.is_empty + def load_cluster_from_dcs(self) -> None: cluster = self.dcs.get_cluster() @@ -460,7 +472,7 @@ class Ha(object): if timeout == 0: # We are requested to prefer failing over to restarting primary. But see first if there # is anyone to fail over to. - if self.is_failover_possible(self.get_failover_candidates()): + if self.is_failover_possible(): self.watchdog.disable() logger.info("Primary crashed. Failing over.") self.demote('immediate') @@ -527,10 +539,17 @@ class Ha(object): return msg def _get_node_to_follow(self, cluster: Cluster) -> Union[Leader, Member, None]: - # determine the node to follow. If replicatefrom tag is set, - # try to follow the node mentioned there, otherwise, follow the leader. - if self.is_standby_cluster() and (self.cluster.is_unlocked() or self.has_lock(False)): + """Determine the node to follow. + + :param cluster: the currently known cluster state from DCS. + + :returns: the node which we should be replicating from. + """ + # The standby leader or when there is no standby leader we want to follow + # the remote member, except when there is no standby leader in pause. + if self.is_standby_cluster() and (self.has_lock(False) or self.cluster.is_unlocked() and not self.is_paused()): node_to_follow = self.get_remote_member() + # If replicatefrom tag is set, try to follow the node mentioned there, otherwise, follow the leader. elif self.patroni.replicatefrom and self.patroni.replicatefrom != self.state_handler.name: node_to_follow = cluster.get_member(self.patroni.replicatefrom) else: @@ -770,6 +789,9 @@ class Ha(object): self.state_handler.sync_handler.set_synchronous_standby_names( CaseInsensitiveSet('*') if self.global_config.is_synchronous_mode_strict else CaseInsensitiveSet()) if self.state_handler.role not in ('master', 'promoted', 'primary'): + # reset failsafe state when promote + self._failsafe.set_is_active(0) + def before_promote(): self.notify_citus_coordinator('before_promote') @@ -795,6 +817,8 @@ class Ha(object): return _MemberStatus.unknown(member) def fetch_nodes_statuses(self, members: List[Member]) -> List[_MemberStatus]: + if not members: + return [] pool = ThreadPool(len(members)) results = pool.map(self.fetch_node_status, members) # Run API calls on members in parallel pool.close() @@ -832,7 +856,7 @@ class Ha(object): data['slots'] = self.state_handler.slots() except Exception: logger.exception('Exception when called state_handler.slots()') - members = [RemoteMember.from_name_and_data(name, {'api_url': url}) + members = [RemoteMember(name, {'api_url': url}) for name, url in failsafe.items() if name != self.state_handler.name] if not members: # A sinlge node cluster return True @@ -873,48 +897,52 @@ class Ha(object): # Prepare list of nodes to run check against members = [m for m in members if m.name != self.state_handler.name and not m.nofailover and m.api_url] - if members: - for st in self.fetch_nodes_statuses(members): - if st.failover_limitation() is None: - if st.in_recovery is False: - logger.warning('Primary (%s) is still alive', st.member.name) + for st in self.fetch_nodes_statuses(members): + if st.failover_limitation() is None: + if st.in_recovery is False: + logger.warning('Primary (%s) is still alive', st.member.name) + return False + if my_wal_position < st.wal_position: + logger.info('Wal position of %s is ahead of my wal position', st.member.name) + # In synchronous mode the former leader might be still accessible and even be ahead of us. + # We should not disqualify himself from the leader race in such a situation. + if not self.sync_mode_is_active() or not self.cluster.sync.leader_matches(st.member.name): return False - if my_wal_position < st.wal_position: - logger.info('Wal position of %s is ahead of my wal position', st.member.name) - # In synchronous mode the former leader might be still accessible and even be ahead of us. - # We should not disqualify himself from the leader race in such a situation. - if not self.is_synchronous_mode() or self.cluster.sync.is_empty\ - or not self.cluster.sync.leader_matches(st.member.name): - return False - logger.info('Ignoring the former leader being ahead of us') + logger.info('Ignoring the former leader being ahead of us') return True - def is_failover_possible(self, members: List[Member], cluster_lsn: Optional[int] = 0) -> bool: - """Checks whether one of the members from the list is healthy enough and is allowed to promote. + def is_failover_possible(self, *, cluster_lsn: int = 0, exclude_failover_candidate: bool = False) -> bool: + """Checks whether any of the cluster members is allowed to promote and is healthy enough for that. - :param members: list of members to check - :param cluster_lsn: to calculate replication lag and exclude member if it is laggin - :returns: `True` if there are members eligible to be the new leader + :param cluster_lsn: to calculate replication lag and exclude member if it is lagging. + :param exclude_failover_candidate: if ``True``, exclude :attr:`failover.candidate` from the members + list against which the failover possibility checks are run. + :returns: `True` if there are members eligible to become the new leader. """ + candidates = self.get_failover_candidates(exclude_failover_candidate) + + action = 'switchover' if self.cluster.failover and self.cluster.failover.leader else 'failover' + if self.is_synchronous_mode() and self.cluster.failover and self.cluster.failover.candidate and not candidates: + logger.warning('%s candidate=%s does not match with sync_standbys=%s', + action.title(), self.cluster.failover.candidate, self.cluster.sync.sync_standby) + elif not candidates: + logger.warning('%s%s: candidates list is empty', '' if not self.cluster.failover else 'manual ', action) + logger.warning('manual failover: candidates list is empty') + ret = False cluster_timeline = self.cluster.timeline - members = [m for m in members if not m.nofailover and m.api_url] - if members: - for st in self.fetch_nodes_statuses(members): - not_allowed_reason = st.failover_limitation() - if not_allowed_reason: - logger.info('Member %s is %s', st.member.name, not_allowed_reason) - elif cluster_lsn and st.wal_position < cluster_lsn or\ - not cluster_lsn and self.is_lagging(st.wal_position): - logger.info('Member %s exceeds maximum replication lag', st.member.name) - elif self.check_timeline() and (not st.timeline or st.timeline < cluster_timeline): - logger.info('Timeline %s of member %s is behind the cluster timeline %s', - st.timeline, st.member.name, cluster_timeline) - else: - ret = True - else: - action = 'switchover' if self.cluster.failover and self.cluster.failover.leader else 'failover' - logger.warning('%s%s: members list is empty', '' if not self.cluster.failover else 'manual ', action) + for st in self.fetch_nodes_statuses(candidates): + not_allowed_reason = st.failover_limitation() + if not_allowed_reason: + logger.info('Member %s is %s', st.member.name, not_allowed_reason) + elif cluster_lsn and st.wal_position < cluster_lsn or \ + not cluster_lsn and self.is_lagging(st.wal_position): + logger.info('Member %s exceeds maximum replication lag', st.member.name) + elif self.check_timeline() and (not st.timeline or st.timeline < cluster_timeline): + logger.info('Timeline %s of member %s is behind the cluster timeline %s', + st.timeline, st.member.name, cluster_timeline) + else: + ret = True return ret def manual_failover_process_no_leader(self) -> Optional[bool]: @@ -966,9 +994,8 @@ class Ha(object): # try to pick some other members to switchover and check that they are healthy if failover.leader: if self.state_handler.name == failover.leader: # I was the leader - # exclude me and desired member which is unhealthy (failover.candidate can be None) - members = [m for m in self.cluster.members if m.name not in (failover.candidate, failover.leader)] - if self.is_failover_possible(members): # check that there are healthy members + # exclude desired member which is unhealthy if it was specified + if self.is_failover_possible(exclude_failover_candidate=bool(failover.candidate)): return False else: # I was the leader and it looks like currently I am the only healthy member return True @@ -1021,8 +1048,8 @@ class Ha(object): if self.cluster.failover: # When doing a switchover in synchronous mode only synchronous nodes and former leader are allowed to race - if self.cluster.failover.leader and self.is_synchronous_mode() and\ - not self.cluster.sync.is_empty and not self.cluster.sync.matches(self.state_handler.name, True): + if self.cluster.failover.leader and self.sync_mode_is_active() \ + and not self.cluster.sync.matches(self.state_handler.name, True): return False return self.manual_failover_process_no_leader() or False @@ -1039,12 +1066,11 @@ class Ha(object): if failsafe_members and self.state_handler.name not in failsafe_members: return False # Race among not only existing cluster members, but also all known members from the failsafe config - all_known_members += [RemoteMember.from_name_and_data(name, {'api_url': url}) - for name, url in failsafe_members.items()] + all_known_members += [RemoteMember(name, {'api_url': url}) for name, url in failsafe_members.items()] all_known_members += self.cluster.members # When in sync mode, only last known primary and sync standby are allowed to promote automatically. - if self.is_synchronous_mode() and not self.cluster.sync.is_empty: + if self.sync_mode_is_active(): if not self.cluster.sync.matches(self.state_handler.name, True): return False # pick between synchronous candidates so we minimize unnecessary failovers/demotions @@ -1095,8 +1121,7 @@ class Ha(object): # It could happen if Postgres is still archiving the backlog of WAL files. # If we know that there are replicas that received the shutdown checkpoint # location, we can remove the leader key and allow them to start leader race. - if self.is_failover_possible(self.get_failover_candidates(check_sync=False), - cluster_lsn=checkpoint_location): + if self.is_failover_possible(cluster_lsn=checkpoint_location): self.state_handler.set_role('demoted') with self._async_executor: self.release_leader_key_voluntarily(checkpoint_location) @@ -1204,16 +1229,11 @@ class Ha(object): if not failover.candidate or failover.candidate != self.state_handler.name: if not failover.candidate and self.is_paused(): logger.warning('%s is possible only to a specific candidate in a paused state', action.title()) + elif self.is_failover_possible(): + ret = self._async_executor.try_run_async(f'manual {action}: demote', self.demote, ('graceful',)) + return ret or f'manual {action}: demoting myself' else: - members = self.get_failover_candidates(check_sync=self.is_synchronous_mode()) - if failover.candidate and not members: - logger.warning('%s candidate=%s does not match with sync_standbys=%s', - action.title(), failover.candidate, self.cluster.sync.sync_standby) - if self.is_failover_possible(members): # check that there are healthy members - ret = self._async_executor.try_run_async(f'manual {action}: demote', self.demote, ('graceful',)) - return ret or f'manual {action}: demoting myself' - else: - logger.warning('manual %s: no healthy members found, %s is not possible', action, action) + logger.warning('manual %s: no healthy members found, %s is not possible', action, action) else: logger.warning('manual %s: I am already the leader, no need to %s', action, action) else: @@ -1475,7 +1495,7 @@ class Ha(object): if self.has_lock() and self.update_lock(): if self._async_executor.scheduled_action == 'doing crash recovery in a single user mode': time_left = self.global_config.primary_start_timeout - (time.time() - self._crash_recovery_started) - if time_left <= 0 and self.is_failover_possible(self.get_failover_candidates()): + if time_left <= 0 and self.is_failover_possible(): logger.info("Demoting self because crash recovery is taking too long") self.state_handler.cancellable.cancel(True) self.demote('immediate') @@ -1587,7 +1607,7 @@ class Ha(object): time_left = timeout - self.state_handler.time_in_state() if time_left <= 0: - if self.is_failover_possible(self.get_failover_candidates()): + if self.is_failover_possible(): logger.info("Demoting self because primary startup is taking too long") self.demote('immediate') return 'stopped PostgreSQL because of startup timeout' @@ -1865,9 +1885,7 @@ class Ha(object): # If we know that there are replicas that received the shutdown checkpoint # location, we can remove the leader key and allow them to start leader race. - # for a manual failover/switchover with a candidate, we should check the requested candidate only - if self.is_failover_possible(self.get_failover_candidates(check_sync=False), - cluster_lsn=checkpoint_location): + if self.is_failover_possible(cluster_lsn=checkpoint_location): self.dcs.delete_leader(checkpoint_location) status['deleted'] = True else: @@ -1926,31 +1944,33 @@ class Ha(object): data['conn_kwargs'] = conn_kwargs name = member.name if member else 'remote_member:{}'.format(uuid.uuid1()) - return RemoteMember.from_name_and_data(name, data) + return RemoteMember(name, data) - def get_failover_candidates(self, check_sync: bool = True) -> List[Member]: - """Return list of candidates (except me) for either manual or automatic failover. + def get_failover_candidates(self, exclude_failover_candidate: bool) -> List[Member]: + """Return a list of candidates for either manual or automatic failover. - The result is later passed to ``Ha.is_failover_possible()`` to check if any member - is actually healthy enough and is allowed to poromote. + Exclude non-sync members when in synchronous mode, the current node (its checks are always performed earlier) + and the candidate if required. If failover candidate exclusion is not requested and a candidate is specified + in the /failover key, return the candidate only. + The result is further evaluated in the caller :func:`Ha.is_failover_possible` to check if any member is actually + healthy enough and is allowed to poromote. - :param check_sync: if ``True``, also check against the sync key members + :param exclude_failover_candidate: if ``True``, exclude :attr:`failover.candidate` from the candidates. - :returns: a list of ``Member`` ojects or an empty list if there is no candidate available. - Never includes the current node, as its checks are always performed earlier. + :returns: a list of :class:`Member` ojects or an empty list if there is no candidate available. """ failover = self.cluster.failover - if check_sync and self.is_synchronous_mode() and not self.cluster.sync.is_empty: - if failover and not failover.leader: - # manual *failover*, only check the candidate (even if not in sync members) - return [m for m in self.cluster.members if m.name == failover.candidate - and m.name != self.state_handler.name] - else: - # the candidate if is in /sync members for a candidate failover, every /sync member otherwise - return [m for m in self.cluster.members if self.cluster.sync.matches(m.name) - and (not failover or not failover.candidate or m.name == failover.candidate) - and m.name != self.state_handler.name] - # the candidate for a candidate failover, every cluster member otherwise - return [m for m in self.cluster.members - if (not failover or not failover.candidate or m.name == failover.candidate) - and m.name != self.state_handler.name] + exclude = [self.state_handler.name] + ([failover.candidate] if failover and exclude_failover_candidate else []) + + def is_eligible(node: Member) -> bool: + # in synchronous mode we allow failover (not switchover!) to async node + if self.sync_mode_is_active() and not self.cluster.sync.matches(node.name)\ + and not (failover and not failover.leader): + return False + # Don't spend time on "nofailover" nodes checking. + # We also don't need nodes which we can't query with the api in the list. + return node.name not in exclude and \ + not node.nofailover and bool(node.api_url) and \ + (not failover or not failover.candidate or node.name == failover.candidate) + + return list(filter(is_eligible, self.cluster.members)) diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py index 04fab0a2..e9844710 100644 --- a/patroni/postgresql/__init__.py +++ b/patroni/postgresql/__init__.py @@ -12,7 +12,7 @@ from datetime import datetime from dateutil import tz from psutil import TimeoutExpired from threading import current_thread, Lock -from typing import Any, Callable, Dict, Generator, List, Optional, Union, Tuple, TYPE_CHECKING +from typing import Any, Callable, Dict, Iterator, List, Optional, Union, Tuple, TYPE_CHECKING from .bootstrap import Bootstrap from .callback_executor import CallbackAction, CallbackExecutor @@ -999,7 +999,7 @@ class Postgresql(object): @contextmanager def get_replication_connection_cursor(self, host: Optional[str] = None, port: int = 5432, - **kwargs: Any) -> Generator[Union['cursor', 'Cursor[Any]'], None, None]: + **kwargs: Any) -> Iterator[Union['cursor', 'Cursor[Any]']]: conn_kwargs = self.config.replication.copy() conn_kwargs.update(host=host, port=int(port) if port else None, user=conn_kwargs.pop('username'), connect_timeout=3, replication=1, options='-c statement_timeout=2000') diff --git a/patroni/postgresql/config.py b/patroni/postgresql/config.py index 3a61a31d..6d1fe384 100644 --- a/patroni/postgresql/config.py +++ b/patroni/postgresql/config.py @@ -6,14 +6,16 @@ import socket import stat import time +from contextlib import contextmanager from urllib.parse import urlparse, parse_qsl, unquote from types import TracebackType -from typing import Any, Collection, Dict, List, Optional, Union, Tuple, Type, TYPE_CHECKING +from typing import Any, Collection, Dict, Iterator, List, Optional, Union, Tuple, Type, TYPE_CHECKING from .validator import recovery_parameters, transform_postgresql_parameter_value, transform_recovery_parameter_value from ..collections import CaseInsensitiveDict, CaseInsensitiveSet from ..dcs import Leader, Member, RemoteMember, slot_name_from_member_name from ..exceptions import PatroniFatalException +from ..file_perm import pg_perm from ..utils import compare_values, parse_bool, parse_int, split_host_port, uri, validate_directory, is_subpath from ..validator import IntValidator, EnumValidator @@ -367,6 +369,30 @@ class ConfigHandler(object): configuration.append('pg_ident.conf') return configuration + def set_file_permissions(self, filename: str) -> None: + """Set permissions of file *filename* according to the expected permissions if it resides under PGDATA. + + .. note:: + Do nothing if the file is not under PGDATA. + + :param filename: path to a file which permissions might need to be adjusted. + """ + if is_subpath(self._postgresql.data_dir, filename): + pg_perm.set_permissions_from_data_directory(self._postgresql.data_dir) + os.chmod(filename, pg_perm.file_create_mode) + + @contextmanager + def config_writer(self, filename: str) -> Iterator[ConfigWriter]: + """Create :class:`ConfigWriter` object and set permissions on a *filename*. + + :param filename: path to a config file. + + :yields: :class:`ConfigWriter` object. + """ + with ConfigWriter(filename) as writer: + yield writer + self.set_file_permissions(filename) + def save_configuration_files(self, check_custom_bootstrap: bool = False) -> bool: """ copy postgresql.conf to postgresql.conf.backup to be able to retrieve configuration files @@ -380,6 +406,7 @@ class ConfigHandler(object): backup_file = os.path.join(self._postgresql.data_dir, f + '.backup') if os.path.isfile(config_file): shutil.copy(config_file, backup_file) + self.set_file_permissions(backup_file) except IOError: logger.exception('unable to create backup copies of configuration files') return True @@ -393,9 +420,11 @@ class ConfigHandler(object): if not os.path.isfile(config_file): if os.path.isfile(backup_file): shutil.copy(backup_file, config_file) + self.set_file_permissions(config_file) # Previously we didn't backup pg_ident.conf, if file is missing just create empty elif f == 'pg_ident.conf': open(config_file, 'w').close() + self.set_file_permissions(config_file) except IOError: logger.exception('unable to restore configuration files from backup') @@ -409,7 +438,7 @@ class ConfigHandler(object): if self._postgresql.enforce_hot_standby_feedback: configuration['hot_standby_feedback'] = 'on' - with ConfigWriter(self._postgresql_conf) as f: + with self.config_writer(self._postgresql_conf) as f: include = self._config.get('custom_conf') or self._postgresql_base_conf_name f.writeline("include '{0}'\n".format(ConfigWriter.escape(include))) for name, value in sorted((configuration).items()): @@ -439,6 +468,7 @@ class ConfigHandler(object): if not self.hba_file and not self._config.get('pg_hba'): with open(self._pg_hba_conf, 'a') as f: f.write('\n{}\n'.format('\n'.join(config))) + self.set_file_permissions(self._pg_hba_conf) return True def replace_pg_hba(self) -> Optional[bool]: @@ -458,14 +488,14 @@ class ConfigHandler(object): self.local_replication_address['host'], self.local_replication_address['port'], 0, socket.SOCK_STREAM, socket.IPPROTO_TCP)}) - with ConfigWriter(self._pg_hba_conf) as f: + with self.config_writer(self._pg_hba_conf) as f: for address, t in addresses.items(): f.writeline(( '{0}\treplication\t{1}\t{3}\ttrust\n' '{0}\tall\t{2}\t{3}\ttrust' ).format(t, self.replication['username'], self._superuser.get('username') or 'all', address)) elif not self.hba_file and self._config.get('pg_hba'): - with ConfigWriter(self._pg_hba_conf) as f: + with self.config_writer(self._pg_hba_conf) as f: f.writelines(self._config['pg_hba']) return True @@ -478,7 +508,7 @@ class ConfigHandler(object): """ if not self.ident_file and self._config.get('pg_ident'): - with ConfigWriter(self._pg_ident_conf) as f: + with self.config_writer(self._pg_ident_conf) as f: f.writelines(self._config['pg_ident']) return True @@ -800,9 +830,11 @@ class ConfigHandler(object): if self._postgresql.major_version >= 120000: if parse_bool(recovery_params.pop('standby_mode', None)): open(self._standby_signal, 'w').close() + self.set_file_permissions(self._standby_signal) else: self._remove_file_if_exists(self._standby_signal) open(self._recovery_signal, 'w').close() + self.set_file_permissions(self._recovery_signal) def restart_required(name: str) -> bool: if self._postgresql.major_version >= 140000: @@ -813,8 +845,7 @@ class ConfigHandler(object): self._current_recovery_params = CaseInsensitiveDict({n: [v, restart_required(n), self._postgresql_conf] for n, v in recovery_params.items()}) else: - with ConfigWriter(self._recovery_conf) as f: - os.chmod(self._recovery_conf, stat.S_IWRITE | stat.S_IREAD) + with self.config_writer(self._recovery_conf) as f: self._write_recovery_params(f, recovery_params) def remove_recovery_conf(self) -> None: @@ -843,6 +874,7 @@ class ConfigHandler(object): if overwrite: try: with open(self._auto_conf, 'w') as f: + self.set_file_permissions(self._auto_conf) for raw_line in lines: f.write(raw_line) except Exception: diff --git a/patroni/postgresql/connection.py b/patroni/postgresql/connection.py index 6a556983..277a4889 100644 --- a/patroni/postgresql/connection.py +++ b/patroni/postgresql/connection.py @@ -2,7 +2,7 @@ import logging from contextlib import contextmanager from threading import Lock -from typing import Any, Dict, Generator, Union, TYPE_CHECKING +from typing import Any, Dict, Iterator, Union, TYPE_CHECKING if TYPE_CHECKING: # pragma: no cover from psycopg import Connection as Connection3, Cursor from psycopg2 import connection, cursor @@ -44,7 +44,7 @@ class Connection(object): @contextmanager -def get_connection_cursor(**kwargs: Any) -> Generator[Union['cursor', 'Cursor[Any]'], None, None]: +def get_connection_cursor(**kwargs: Any) -> Iterator[Union['cursor', 'Cursor[Any]']]: conn = psycopg.connect(**kwargs) with conn.cursor() as cur: yield cur diff --git a/patroni/postgresql/slots.py b/patroni/postgresql/slots.py index fde81a7a..4abda4b1 100644 --- a/patroni/postgresql/slots.py +++ b/patroni/postgresql/slots.py @@ -1,15 +1,20 @@ +"""Replication slot handling. + +Provides classes for the creation, monitoring, management and synchronisation of PostgreSQL replication slots. +""" + import logging import os import shutil - from collections import defaultdict from contextlib import contextmanager from threading import Condition, Thread -from typing import Any, Dict, Generator, List, Optional, Union, Tuple, TYPE_CHECKING +from typing import Any, Dict, Iterator, List, Optional, Union, Tuple, TYPE_CHECKING, Collection from .connection import get_connection_cursor from .misc import format_lsn, fsync_dir from ..dcs import Cluster, Leader +from ..file_perm import pg_perm from ..psycopg import OperationalError if TYPE_CHECKING: # pragma: no cover @@ -42,9 +47,17 @@ def compare_slots(s1: Dict[str, Any], s2: Dict[str, Any], dbid: str = 'database' class SlotsAdvanceThread(Thread): + """Daemon process :class:``Thread`` object for advancing logical replication slots on replicas. + + This ensures that slot advancing queries sent to postgres do not block the main loop. + """ def __init__(self, slots_handler: 'SlotsHandler') -> None: - super(SlotsAdvanceThread, self).__init__() + """Create and start a new thread for handling slot advance queries. + + :param slots_handler: The calling class instance for reference to slot information attributes. + """ + super().__init__() self.daemon = True self._slots_handler = slots_handler @@ -58,6 +71,13 @@ class SlotsAdvanceThread(Thread): self.start() def sync_slot(self, cur: Union['cursor', 'Cursor[Any]'], database: str, slot: str, lsn: int) -> None: + """Execute a ``pg_replication_slot_advance`` query and store success for scheduled synchronisation task. + + :param cur: database connection cursor. + :param database: name of the database associated with the slot. + :param slot: name of the slot to be synchronised. + :param lsn: last known LSN position + """ failed = copy = False try: cur.execute("SELECT pg_catalog.pg_replication_slot_advance(%s, %s)", (slot, format_lsn(lsn))) @@ -79,6 +99,11 @@ class SlotsAdvanceThread(Thread): self._scheduled.pop(database) def sync_slots_in_database(self, database: str, slots: List[str]) -> None: + """Synchronise slots for a single database. + + :param database: name of the database. + :param slots: list of slot names to synchronise. + """ with self._slots_handler.get_local_connection_cursor(dbname=database, options='-c statement_timeout=0') as cur: for slot in slots: with self._condition: @@ -87,6 +112,7 @@ class SlotsAdvanceThread(Thread): self.sync_slot(cur, database, slot, lsn) def sync_slots(self) -> None: + """Synchronise slots for all scheduled databases.""" with self._condition: databases = list(self._scheduled.keys()) for database in databases: @@ -99,6 +125,12 @@ class SlotsAdvanceThread(Thread): logger.error('Failed to advance replication slots in database %s: %r', database, e) def run(self) -> None: + """Thread main loop entrypoint. + + .. note:: + Thread will wait until a sync is scheduled from outside, normally triggered during the HA loop or a wakeup + call. + """ while True: with self._condition: if not self._scheduled: @@ -107,6 +139,14 @@ class SlotsAdvanceThread(Thread): self.sync_slots() def schedule(self, advance_slots: Dict[str, Dict[str, int]]) -> Tuple[bool, List[str]]: + """Trigger a synchronisation of slots. + + This is the main entrypoint for Patroni HA loop wakeup call. + + :param advance_slots: dictionary containing slots that need to be advanced + + :return: tuple of failure status and a list of slots to be copied + """ with self._condition: for database, values in advance_slots.items(): self._scheduled[database].update(values) @@ -118,15 +158,27 @@ class SlotsAdvanceThread(Thread): return ret def on_promote(self) -> None: + """Reset state of the daemon.""" with self._condition: self._scheduled.clear() self._failed = False self._copy_slots = [] -class SlotsHandler(object): +class SlotsHandler: + """Handler for managing and storing information on replication slots in PostgreSQL. + + :ivar pg_replslot_dir: system location path of the PostgreSQL replication slots. + :ivar _logical_slots_processing_queue: yet to be processed logical replication slots on the primary + """ def __init__(self, postgresql: 'Postgresql') -> None: + """Create an instance with storage attributes for replication slots and schedule the first synchronisation. + + :param postgresql: Calling class instance providing interface to PostgreSQL. + """ + self._force_readiness_check = False + self._schedule_load_slots = False self._postgresql = postgresql self._advance = None self._replication_slots: Dict[str, Dict[str, Any]] = {} # already existing replication slots @@ -135,23 +187,46 @@ class SlotsHandler(object): self.schedule() def _query(self, sql: str, *params: Any) -> Union['cursor', 'Cursor[Any]']: + """Helper method for :meth:`Postgresql.query`. + + :param sql: SQL statement to execute. + :param params: parameters to pass through to :meth:`Postgresql.query`. + + :returns: query response. + """ return self._postgresql.query(sql, *params, retry=False) @staticmethod - def _copy_items(src: Dict[str, Any], dst: Dict[str, Any], keys: Optional[List[str]] = None) -> None: + def _copy_items(src: Dict[str, Any], dst: Dict[str, Any], keys: Optional[Collection[str]] = None) -> None: + """Select values from *src* dictionary to update in *dst* dictionary for optional supplied *keys*. + + :param src: source dictionary that *keys* will be looked up from. + :param dst: destination dictionary to be updated. + :param keys: optional list of keys to be looked up in the source dictionary. + """ dst.update({key: src[key] for key in keys or ('datoid', 'catalog_xmin', 'confirmed_flush_lsn')}) def process_permanent_slots(self, slots: List[Dict[str, Any]]) -> Dict[str, int]: - """This methods solves three problems at once (I know, it is weird). + """Process replication slot information from the host and prepare information used in subsequent cluster tasks. + + .. note:: + This methods solves three problems. + + The ``cluster_info_query`` from :class:``Postgresql`` is executed every HA loop and returns information + about all replication slots that exists on the current host. + + Based on this information perform the following actions: + + 1. For the primary we want to expose to DCS permanent logical slots, therefore build (and return) a dict + that maps permanent logical slot names to ``confirmed_flush_lsn``. + 2. detect if one of the previously known permanent slots is missing and schedule resync. + 3. Update the local cache with the fresh ``catalog_xmin`` and ``confirmed_flush_lsn`` for every known slot. - The cluster_info_query from `Postgresql` is executed every HA loop and returns - information about all replication slots that exists on the current host. - Based on this information we perform the following actions: - 1. For the primary we want to expose to DCS permanent logical slots, therefore the method - builds (and returns) a dict, that maps permanent logical slot names and confirmed_flush_lsns. - 2. This method also detects if one of the previously known permanent slots got missing and schedules resync. - 3. Updates the local cache with the fresh catalog_xmin and confirmed_flush_lsn for every known slot. This info is used when performing the check of logical slot readiness on standbys. + + :param slots: replication slot information that exists on the current host. + + :return: dictionary of logical slot names to ``confirmed_flush_lsn``. """ ret: Dict[str, int] = {} @@ -173,13 +248,23 @@ class SlotsHandler(object): return ret def load_replication_slots(self) -> None: + """Query replication slot information from the database and store it for processing by other tasks. + + .. note:: + Only supported from PostgreSQL version 9.4 onwards. + + Store replication slot ``name``, ``type``, ``plugin``, ``database`` and ``datoid``. + If PostgreSQL version is 10 or newer also store ``catalog_xmin`` and ``confirmed_flush_lsn``. + + When using logical slots, store information separately for slot synchronisation on replica nodes. + """ if self._postgresql.major_version >= 90400 and self._schedule_load_slots: replication_slots: Dict[str, Dict[str, Any]] = {} - extra = ", catalog_xmin, pg_catalog.pg_wal_lsn_diff(confirmed_flush_lsn, '0/0')::bigint"\ + extra = ", catalog_xmin, pg_catalog.pg_wal_lsn_diff(confirmed_flush_lsn, '0/0')::bigint" \ if self._postgresql.major_version >= 100000 else "" skip_temp_slots = ' WHERE NOT temporary' if self._postgresql.major_version >= 100000 else '' - cursor = self._query('SELECT slot_name, slot_type, plugin, database, datoid' - '{0} FROM pg_catalog.pg_replication_slots{1}'.format(extra, skip_temp_slots)) + cursor = self._query(f'SELECT slot_name, slot_type, plugin, database, datoid' + f'{extra} FROM pg_catalog.pg_replication_slots{skip_temp_slots}') for r in cursor: value = {'type': r[1]} if r[1] == 'logical': @@ -195,16 +280,34 @@ class SlotsHandler(object): self._force_readiness_check = False def ignore_replication_slot(self, cluster: Cluster, name: str) -> bool: + """Check if slot *name* should not be managed by Patroni. + + :param cluster: cluster state information object. + :param name: name of the slot to ignore + + :returns: ``True`` if slot *name* matches any slot specified in ``ignore_slots`` configuration, + otherwise will pass through and return result of :meth:`CitusHandler.ignore_replication_slot`. + """ slot = self._replication_slots[name] if cluster.config: for matcher in cluster.config.ignore_slots_matchers: - if ((matcher.get("name") is None or matcher["name"] == name) - and all(not matcher.get(a) or matcher[a] == slot.get(a) for a in ('database', 'plugin', 'type'))): + if ( + (matcher.get("name") is None or matcher["name"] == name) + and all(not matcher.get(a) or matcher[a] == slot.get(a) + for a in ('database', 'plugin', 'type')) + ): return True return self._postgresql.citus_handler.ignore_replication_slot(slot) def drop_replication_slot(self, name: str) -> Tuple[bool, bool]: - """Returns a tuple(active, dropped)""" + """Drop a named slot from Postgres. + + :param name: name of the slot to be dropped. + + :returns: a tuple of ``active`` and ``dropped``. ``active`` is ``True`` if the slot is active, + ``dropped`` is ``True`` if the slot was successfully dropped. If the slot was not found return + ``False`` for both. + """ cursor = self._query(('WITH slots AS (SELECT slot_name, active' ' FROM pg_catalog.pg_replication_slots WHERE slot_name = %s),' ' dropped AS (SELECT pg_catalog.pg_drop_replication_slot(slot_name),' @@ -217,7 +320,19 @@ class SlotsHandler(object): return row def _drop_incorrect_slots(self, cluster: Cluster, slots: Dict[str, Any], paused: bool) -> None: - # drop old replication slots which are not presented in desired slots + """Compare required slots and configured as permanent slots with those found, dropping extraneous ones. + + .. note:: + Slots that are not contained in *slots* will be dropped. + Slots can be filtered out with ``ignore_slots`` configuration. + + Slots that have matching names but do not match attributes in *slots* will also be dropped. + + :param cluster: cluster state information object. + :param slots: dictionary of desired slot names as keys with slot attributes as a dictionary value, if known. + :param paused: ``True`` if the patroni cluster is currently in a paused state. + """ + # drop old replication slots which are not presented in desired slots. for name in set(self._replication_slots) - set(slots): if not paused and not self.ignore_replication_slot(cluster, name): active, dropped = self.drop_replication_slot(name) @@ -229,6 +344,8 @@ class SlotsHandler(object): logger.debug("Unable to drop unknown replication slot '%s', slot is still active", name) else: logger.error("Failed to drop replication slot '%s'", name) + + # drop slots with matching names but attributes that do not match, e.g. `plugin` or `database`. for name, value in slots.items(): if name in self._replication_slots and not compare_slots(value, self._replication_slots[name]): logger.info("Trying to drop replication slot '%s' because value is changing from %s to %s", @@ -240,31 +357,57 @@ class SlotsHandler(object): self._schedule_load_slots = True def _ensure_physical_slots(self, slots: Dict[str, Any]) -> None: + """Create any missing physical replication *slots*. + + Any failures are logged and do not interrupt creation of all *slots*. + + :param slots: A dictionary mapping slot name to slot attributes. This method only considers a slot + if the value is a dictionary with the key ``type`` and a value of ``physical``. + """ immediately_reserve = ', true' if self._postgresql.major_version >= 90600 else '' for name, value in slots.items(): if name not in self._replication_slots and value['type'] == 'physical': try: - self._query(("SELECT pg_catalog.pg_create_physical_replication_slot(%s{0})" - " WHERE NOT EXISTS (SELECT 1 FROM pg_catalog.pg_replication_slots" - " WHERE slot_type = 'physical' AND slot_name = %s)").format( - immediately_reserve), name, name) + self._query(f"SELECT pg_catalog.pg_create_physical_replication_slot(%s{immediately_reserve})" + f" WHERE NOT EXISTS (SELECT 1 FROM pg_catalog.pg_replication_slots" + f" WHERE slot_type = 'physical' AND slot_name = %s)", + name, name) except Exception: logger.exception("Failed to create physical replication slot '%s'", name) self._schedule_load_slots = True @contextmanager - def get_local_connection_cursor(self, **kwargs: Any) -> Generator[Union['cursor', 'Cursor[Any]'], None, None]: + def get_local_connection_cursor(self, **kwargs: Any) -> Iterator[Union['cursor', 'Cursor[Any]']]: + """Create a new database connection to local server. + + Create a non-blocking connection cursor to avoid the situation where an execution of the query of + ``pg_replication_slot_advance`` takes longer than the timeout on a HA loop, which could cause a false + failure state. + + :param kwargs: Any keyword arguments to pass to :func:`psycopg.connect`. + + :yields: connection cursor object, note implementation varies depending on version of :mod:`psycopg`. + """ conn_kwargs = self._postgresql.config.local_connect_kwargs conn_kwargs.update(kwargs) with get_connection_cursor(**conn_kwargs) as cur: yield cur def _ensure_logical_slots_primary(self, slots: Dict[str, Any]) -> None: + """Create any missing logical replication *slots* on the primary. + + If the logical slot already exists, copy state information into the replication slots structure stored in the + class instance. + + :param slots: Slots that should exist are supplied in a dictionary, mapping slot name to any attributes. + The method will only consider slots that have a value that is a dictionary with a key ``type`` + with a value that is ``logical``. + + """ # Group logical slots to be created by database name logical_slots: Dict[str, Dict[str, Dict[str, Any]]] = defaultdict(dict) for name, value in slots.items(): if value['type'] == 'logical': - # If the logical already exists, copy some information about it into the original structure if self._replication_slots.get(name, {}).get('datoid'): self._copy_items(self._replication_slots[name], value) else: @@ -286,27 +429,51 @@ class SlotsHandler(object): self._schedule_load_slots = True def schedule_advance_slots(self, slots: Dict[str, Dict[str, int]]) -> Tuple[bool, List[str]]: + """Wrapper to ensure slots advance daemon thread is started if not already. + + :param slots: dictionary containing slot information. + + :return: tuple with the result of the scheduling of slot advancement: ``failed`` and list of slots to copy. + """ if not self._advance: self._advance = SlotsAdvanceThread(self) return self._advance.schedule(slots) def _ensure_logical_slots_replica(self, cluster: Cluster, slots: Dict[str, Any]) -> List[str]: + """Update logical *slots* on replicas. + + If the logical slot already exists, copy state information into the replication slots structure stored in the + class instance. Slots that exist are also advanced if their ``confirmed_flush_lsn`` is greater than the stored + state of the slot. + + As logical slots can only be created when the primary is available, pass the list of slots that need to be + copied back to the caller. They will be created on replicas with :meth:`SlotsHandler.copy_logical_slots`. + + :param cluster: object containing stateful information for the cluster. + :param slots: A dictionary mapping slot name to slot attributes. This method only considers a slot + if the value is a dictionary with the key ``type`` and a value of ``logical``. + + :returns: list of slots to be copied from the primary. + """ # Group logical slots to be advanced by database name advance_slots: Dict[str, Dict[str, int]] = defaultdict(dict) - create_slots: List[str] = [] # And collect logical slots to be created on the replica + create_slots: List[str] = [] # Collect logical slots to be created on the replica + for name, value in slots.items(): - if value['type'] == 'logical': - # If the logical already exists, copy some information about it into the original structure - if self._replication_slots.get(name, {}).get('datoid'): - self._copy_items(self._replication_slots[name], value) - if cluster.slots and name in cluster.slots: - try: # Skip slots that doesn't need to be advanced - if value['confirmed_flush_lsn'] < int(cluster.slots[name]): - advance_slots[value['database']][name] = int(cluster.slots[name]) - except Exception as e: - logger.error('Failed to parse "%s": %r', cluster.slots[name], e) - elif cluster.slots and name in cluster.slots: # We want to copy only slots with feedback in a DCS - create_slots.append(name) + if value['type'] != 'logical': + continue + + # If the logical already exists, copy some information about it into the original structure + if self._replication_slots.get(name, {}).get('datoid'): + self._copy_items(self._replication_slots[name], value) + if cluster.slots and name in cluster.slots: + try: # Skip slots that don't need to be advanced + if value['confirmed_flush_lsn'] < int(cluster.slots[name]): + advance_slots[value['database']][name] = int(cluster.slots[name]) + except Exception as e: + logger.error('Failed to parse "%s": %r', cluster.slots[name], e) + elif cluster.slots and name in cluster.slots: # We want to copy only slots with feedback in a DCS + create_slots.append(name) error, copy_slots = self.schedule_advance_slots(advance_slots) if error: @@ -315,6 +482,20 @@ class SlotsHandler(object): def sync_replication_slots(self, cluster: Cluster, nofailover: bool, replicatefrom: Optional[str] = None, paused: bool = False) -> List[str]: + """During the HA loop read, check and alter replication slots found in the cluster. + + Read physical and logical slots found on the primary, then compare to those configured in the DCS. + Drop any slots that do not match those required by configuration and are not configured as permanent. + Create any missing physical slots. If we are the leader then logical slots too, otherwise if logical slots + are known and active create them on replica nodes. + + :param cluster: object containing stateful information for the cluster. + :param nofailover: ``True`` if this node has been tagged to not be a failover candidate. + :param replicatefrom: the tag containing the node to replicate from. + :param paused: ``True`` if the cluster is in maintenance mode. + + :returns: list of logical replication slots names that should be copied from the primary. + """ ret = [] if self._postgresql.major_version >= 90400 and cluster.config: try: @@ -342,7 +523,17 @@ class SlotsHandler(object): return ret @contextmanager - def _get_leader_connection_cursor(self, leader: Leader) -> Generator[Union['cursor', 'Cursor[Any]'], None, None]: + def _get_leader_connection_cursor(self, leader: Leader) -> Iterator[Union['cursor', 'Cursor[Any]']]: + """Create a new database connection to the leader. + + .. note:: + Uses rewind user credentials because it has enough permissions to read files from PGDATA. + Sets the options ``connect_timeout`` to ``3`` and ``statement_timeout`` to ``2000``. + + :param leader: object with information on the leader + + :yields: connection cursor object, note implementation varies depending on version of ``psycopg``. + """ conn_kwargs = leader.conn_kwargs(self._postgresql.config.rewind_credentials) conn_kwargs['dbname'] = self._postgresql.database with get_connection_cursor(connect_timeout=3, options="-c statement_timeout=2000", **conn_kwargs) as cur: @@ -351,16 +542,16 @@ class SlotsHandler(object): def check_logical_slots_readiness(self, cluster: Cluster, replicatefrom: Optional[str]) -> bool: """Determine whether all known logical slots are synchronised from the leader. - 1) Retrieve the current ``catalog_xmin`` value for the physical slot from the cluster leader, and - 2) using previously stored list of "unready" logical slots, those which have yet to be checked hence have no - stored slot attributes, - 3) store logical slot ``catalog_xmin`` when the physical slot ``catalog_xmin`` becomes valid. + 1) Retrieve the current ``catalog_xmin`` value for the physical slot from the cluster leader, and + 2) using previously stored list of "unready" logical slots, those which have yet to be checked hence have no + stored slot attributes, + 3) store logical slot ``catalog_xmin`` when the physical slot ``catalog_xmin`` becomes valid. - :param cluster: object containing stateful information for the cluster. - :param replicatefrom: name of the member that should be used to replicate from. + :param cluster: object containing stateful information for the cluster. + :param replicatefrom: name of the member that should be used to replicate from. - :returns: ``False`` if any issue while checking logical slots readiness, ``True`` otherwise. - """ + :returns: ``False`` if any issue while checking logical slots readiness, ``True`` otherwise. + """ catalog_xmin = None if self._logical_slots_processing_queue and cluster.leader: slot_name = cluster.get_my_slot_name_on_primary(self._postgresql.name, replicatefrom) @@ -445,6 +636,11 @@ class SlotsHandler(object): logger.info('Logical slot %s is safe to be used after a failover', name) def copy_logical_slots(self, cluster: Cluster, create_slots: List[str]) -> None: + """Create logical replication slots on standby nodes. + + :param cluster: object containing stateful information for the cluster. + :param create_slots: list of slot names to copy from the primary. + """ leader = cluster.leader if not leader: return @@ -471,31 +667,48 @@ class SlotsHandler(object): logger.error("Failed to copy logical slots from the %s via postgresql connection: %r", leader.name, e) if copy_slots and self._postgresql.stop(): + pg_perm.set_permissions_from_data_directory(self._postgresql.data_dir) for name, value in copy_slots.items(): - slot_dir = os.path.join(self._postgresql.slots_handler.pg_replslot_dir, name) + slot_dir = os.path.join(self.pg_replslot_dir, name) slot_tmp_dir = slot_dir + '.tmp' if os.path.exists(slot_tmp_dir): shutil.rmtree(slot_tmp_dir) os.makedirs(slot_tmp_dir) + os.chmod(slot_tmp_dir, pg_perm.dir_create_mode) fsync_dir(slot_tmp_dir) - with open(os.path.join(slot_tmp_dir, 'state'), 'wb') as f: + slot_filename = os.path.join(slot_tmp_dir, 'state') + with open(slot_filename, 'wb') as f: + os.chmod(slot_filename, pg_perm.file_create_mode) f.write(value['data']) f.flush() os.fsync(f.fileno()) if os.path.exists(slot_dir): shutil.rmtree(slot_dir) os.rename(slot_tmp_dir, slot_dir) + os.chmod(slot_dir, pg_perm.dir_create_mode) fsync_dir(slot_dir) self._logical_slots_processing_queue[name] = None - fsync_dir(self._postgresql.slots_handler.pg_replslot_dir) + fsync_dir(self.pg_replslot_dir) self._postgresql.start() def schedule(self, value: Optional[bool] = None) -> None: + """Schedule the loading of slot information from the database. + + :param value: the optional value can be used to unschedule if set to ``False`` or force it to be ``True``. + If it is omitted the value will be ``True`` if this PostgreSQL node supports slot replication. + """ if value is None: value = self._postgresql.major_version >= 90400 self._schedule_load_slots = self._force_readiness_check = value def on_promote(self) -> None: + """Entry point from HA cycle used when a standby node is to be promoted to primary. + + .. note:: + If logical replication slot synchronisation is enabled then slot advancement will be triggered. + If any logical slots that were copied are yet to be confirmed as ready a warning message will be logged. + + """ if self._advance: self._advance.on_promote() diff --git a/patroni/postgresql/sync.py b/patroni/postgresql/sync.py index d0f28586..2280a68c 100644 --- a/patroni/postgresql/sync.py +++ b/patroni/postgresql/sync.py @@ -153,6 +153,72 @@ def parse_sync_standby_names(value: str) -> _SSN: return _SSN(sync_type, has_star, num, members) +class _Replica(NamedTuple): + """Class representing a single replica that is eligible to be synchronous. + + Attributes are taken from ``pg_stat_replication`` view and respective ``Cluster.members``. + + :ivar pid: PID of walsender process. + :ivar application_name: matches with the ``Member.name``. + :ivar sync_state: possible values are: ``async``, ``potential``, ``quorum``, and ``sync``. + :ivar lsn: ``write_lsn``, ``flush_lsn``, or ``replay_lsn``, depending on the value of ``synchronous_commit`` GUC. + :ivar nofailover: whether the corresponding member has ``nofailover`` tag set to ``True``. + """ + pid: int + application_name: str + sync_state: str + lsn: int + nofailover: bool + + +class _ReplicaList(List[_Replica]): + """A collection of :class:``_Replica`` objects. + + Values are reverse ordered by ``_Replica.sync_state`` and ``_Replica.lsn``. + That is, first there will be replicas that have ``sync_state`` == ``sync``, even if they are not + the most up-to-date in term of write/flush/replay LSN. It helps to keep the result of chosing new + synchronous nodes consistent in case if a synchronous standby member is slowed down OR async node + is receiving changes faster than the sync member. Such cases would trigger sync standby member + swapping, but only if lag on this member is exceeding a threshold (``maximum_lag_on_syncnode``). + + :ivar max_lsn: maximum value of ``_Replica.lsn`` among all values. In case if there is just one + element in the list we take value of ``pg_current_wal_lsn()``. + """ + + def __init__(self, postgresql: 'Postgresql', cluster: Cluster) -> None: + """Create :class:``_ReplicaList`` object. + + :param postgresql: reference to :class:``Postgresql`` object. + :param cluster: currently known cluster state from DCS. + """ + super().__init__() + + # We want to prioritize candidates based on `write_lsn``, ``flush_lsn``, or ``replay_lsn``. + # Which column exactly to pick depends on the values of ``synchronous_commit`` GUC. + sort_col = { + 'remote_apply': 'replay', + 'remote_write': 'write' + }.get(postgresql.synchronous_commit(), 'flush') + '_lsn' + + members = CaseInsensitiveDict({m.name: m for m in cluster.members}) + for row in postgresql.pg_stat_replication(): + member = members.get(row['application_name']) + + # We want to consider only rows from ``pg_stat_replication` that: + # 1. are known to be streaming (write/flush/replay LSN are not NULL). + # 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): + self.append(_Replica(row['pid'], row['application_name'], + row['sync_state'], row[sort_col], bool(member.nofailover))) + + # Prefer replicas that are in state ``sync`` and with higher values of ``write``/``flush``/``replay`` LSN. + self.sort(key=lambda r: (r.sync_state, r.lsn), reverse=True) + + self.max_lsn = max(self, key=lambda x: x.lsn).lsn if len(self) > 1 else postgresql.last_operation() + + class SyncHandler(object): """Class responsible for working with the `synchronous_standby_names`. @@ -201,6 +267,21 @@ BEGIN END;$$""") self._postgresql.reset_cluster_info_state(None) # Reset internal cache to query fresh values + def _process_replica_readiness(self, cluster: Cluster, replica_list: _ReplicaList) -> None: + """Flags replicas as truly "synchronous" when they have caught up with ``_primary_flush_lsn``. + + :param cluster: current cluster topology from DCS + :param replica_list: collection of replicas that we want to evaluate. + """ + for replica in replica_list: + # if standby name is listed in the /sync key we can count it as synchronous, otherwise + # it becomes really synchronous when sync_state = 'sync' and it is known that it managed to catch up + if replica.application_name not in self._ready_replicas\ + and replica.application_name in self._ssn_data.members\ + and (cluster.sync.matches(replica.application_name) + or replica.sync_state == 'sync' and replica.lsn >= self._primary_flush_lsn): + self._ready_replicas[replica.application_name] = replica.pid + def current_state(self, cluster: Cluster) -> Tuple[CaseInsensitiveSet, CaseInsensitiveSet]: """Finds best candidates to be the synchronous standbys. @@ -218,31 +299,8 @@ END;$$""") """ self._handle_synchronous_standby_names_change() - # Pick candidates based on who has higher replay/remote_write/flush lsn. - sort_col = { - 'remote_apply': 'replay', - 'remote_write': 'write' - }.get(self._postgresql.synchronous_commit(), 'flush') + '_lsn' - - pg_stat_replication = [(r['pid'], r['application_name'], r['sync_state'], r[sort_col]) - for r in self._postgresql.pg_stat_replication() - if r[sort_col] is not None] - - members = CaseInsensitiveDict({m.name: m for m in cluster.members}) - replica_list: List[Tuple[int, str, str, int, bool]] = [] - # pg_stat_replication.sync_state has 4 possible states - async, potential, quorum, sync. - # That is, alphabetically they are in the reversed order of priority. - # Since we are doing reversed sort on (sync_state, lsn) tuples, it helps to keep the result - # consistent in case if a synchronous standby member is slowed down OR async node receiving - # changes faster than the sync member (very rare but possible). - # Such cases would trigger sync standby member swapping, but only if lag on a sync node exceeding a threshold. - for pid, app_name, sync_state, replica_lsn in sorted(pg_stat_replication, key=lambda r: r[2:4], reverse=True): - member = members.get(app_name) - if member and member.is_running and not member.tags.get('nosync', False): - replica_list.append((pid, member.name, sync_state, replica_lsn, bool(member.nofailover))) - - max_lsn = max(replica_list, key=lambda x: x[3])[3]\ - if len(replica_list) > 1 else self._postgresql.last_operation() + replica_list = _ReplicaList(self._postgresql, cluster) + self._process_replica_readiness(cluster, replica_list) if TYPE_CHECKING: # pragma: no cover assert self._postgresql.global_config is not None @@ -253,17 +311,11 @@ END;$$""") candidates = CaseInsensitiveSet() sync_nodes = CaseInsensitiveSet() # Prefer members without nofailover tag. We are relying on the fact that sorts are guaranteed to be stable. - for pid, app_name, sync_state, replica_lsn, _ in sorted(replica_list, key=lambda x: x[4]): - # if standby name is listed in the /sync key we can count it as synchronous, otherwice - # it becomes really synchronous when sync_state = 'sync' and it is known that it managed to catch up - if app_name not in self._ready_replicas and app_name in self._ssn_data.members and\ - (cluster.sync.matches(app_name) or sync_state == 'sync' and replica_lsn >= self._primary_flush_lsn): - self._ready_replicas[app_name] = pid - - if sync_node_maxlag <= 0 or max_lsn - replica_lsn <= sync_node_maxlag: - candidates.add(app_name) - if sync_state == 'sync' and app_name in self._ready_replicas: - sync_nodes.add(app_name) + for replica in sorted(replica_list, key=lambda x: x.nofailover): + if sync_node_maxlag <= 0 or replica_list.max_lsn - replica.lsn <= sync_node_maxlag: + candidates.add(replica.application_name) + if replica.sync_state == 'sync' and replica.application_name in self._ready_replicas: + sync_nodes.add(replica.application_name) if len(candidates) >= sync_node_count: break diff --git a/patroni/version.py b/patroni/version.py index 8653a3bc..87eff52e 100644 --- a/patroni/version.py +++ b/patroni/version.py @@ -2,4 +2,4 @@ :var __version__: the current Patroni version. """ -__version__ = '3.0.4' +__version__ = '3.1.0' diff --git a/postgres2.yml b/postgres2.yml index c77e734a..581fa719 100644 --- a/postgres2.yml +++ b/postgres2.yml @@ -121,4 +121,4 @@ tags: nofailover: false noloadbalance: false clonefrom: false - replicatefrom: postgres1 +# replicatefrom: postgresql1 diff --git a/tests/test_config.py b/tests/test_config.py index 3f7e4049..cf798d00 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -85,6 +85,7 @@ class TestConfig(unittest.TestCase): @patch('os.path.exists', Mock(return_value=True)) @patch('os.remove', Mock(side_effect=IOError)) @patch('os.close', Mock(side_effect=IOError)) + @patch('os.chmod', Mock()) @patch('shutil.move', Mock(return_value=None)) @patch('json.dump', Mock()) def test_save_cache(self): diff --git a/tests/test_file_perm.py b/tests/test_file_perm.py new file mode 100644 index 00000000..e85b3c65 --- /dev/null +++ b/tests/test_file_perm.py @@ -0,0 +1,33 @@ +import unittest +import stat + +from mock import Mock, patch + +from patroni.file_perm import pg_perm + + +class TestFilePermissions(unittest.TestCase): + + @patch('os.stat') + @patch('os.umask') + @patch('patroni.file_perm.logger.error') + def test_set_umask(self, mock_logger, mock_umask, mock_stat): + mock_umask.side_effect = Exception + mock_stat.return_value.st_mode = stat.S_IRWXU | stat.S_IRGRP | stat.S_IXGRP + pg_perm.set_permissions_from_data_directory('test') + + # umask is called with PG_MODE_MASK_GROUP + self.assertEqual(mock_umask.call_args[0][0], stat.S_IWGRP | stat.S_IRWXO) + self.assertEqual(mock_logger.call_args[0][0], 'Can not set umask to %03o: %r') + + mock_umask.reset_mock() + mock_stat.return_value.st_mode = stat.S_IRWXU + pg_perm.set_permissions_from_data_directory('test') + # umask is called with PG_MODE_MASK_OWNER (permissions changed from group to owner) + self.assertEqual(mock_umask.call_args[0][0], stat.S_IRWXG | stat.S_IRWXO) + + @patch('os.stat', Mock(side_effect=FileNotFoundError)) + @patch('patroni.file_perm.logger.error') + def test_set_permissions_from_data_directory(self, mock_logger): + pg_perm.set_permissions_from_data_directory('test') + self.assertEqual(mock_logger.call_args[0][0], 'Can not check permissions on %s: %r') diff --git a/tests/test_ha.py b/tests/test_ha.py index 06b72fba..b6b21640 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -1584,6 +1584,7 @@ class TestHa(PostgresInit): @patch('os.open', Mock()) @patch('os.fsync', Mock()) @patch('os.close', Mock()) + @patch('os.chmod', Mock()) @patch('os.rename', Mock()) @patch('patroni.postgresql.Postgresql.is_starting', Mock(return_value=False)) @patch('builtins.open', mock_open()) diff --git a/tests/test_kubernetes.py b/tests/test_kubernetes.py index 662e22a8..9eaa173a 100644 --- a/tests/test_kubernetes.py +++ b/tests/test_kubernetes.py @@ -306,19 +306,19 @@ class TestKubernetesConfigMaps(BaseTestKubernetes): self.k.touch_member({'state': 'running', 'role': 'replica'}) mock_patch_namespaced_pod.assert_called() - self.assertEqual(mock_patch_namespaced_pod.call_args.args[2].metadata.labels['isMaster'], 'false') - self.assertEqual(mock_patch_namespaced_pod.call_args.args[2].metadata.labels['tmp_role'], 'replica') + self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['isMaster'], 'false') + self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['tmp_role'], 'replica') self.k.touch_member({'state': 'running', 'role': 'standby-leader'}) mock_patch_namespaced_pod.assert_called() - self.assertEqual(mock_patch_namespaced_pod.call_args.args[2].metadata.labels['isMaster'], 'false') - self.assertEqual(mock_patch_namespaced_pod.call_args.args[2].metadata.labels['tmp_role'], 'standby-leader') + self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['isMaster'], 'false') + self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['tmp_role'], 'standby-leader') self.k._name = 'p-0' self.k.touch_member({'role': 'primary'}) mock_patch_namespaced_pod.assert_called() - self.assertEqual(mock_patch_namespaced_pod.call_args.args[2].metadata.labels['isMaster'], 'true') - self.assertEqual(mock_patch_namespaced_pod.call_args.args[2].metadata.labels['tmp_role'], 'master') + self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['isMaster'], 'true') + self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['tmp_role'], 'master') def test_initialize(self): self.k.initialize() diff --git a/tests/test_patroni.py b/tests/test_patroni.py index e5785231..605b3227 100644 --- a/tests/test_patroni.py +++ b/tests/test_patroni.py @@ -40,6 +40,7 @@ class MockFrozenImporter(object): @patch('time.sleep', Mock()) @patch('subprocess.call', Mock(return_value=0)) @patch('patroni.psycopg.connect', psycopg_connect) +@patch('urllib3.PoolManager.request', Mock(side_effect=Exception)) @patch.object(ConfigHandler, 'append_pg_hba', Mock()) @patch.object(ConfigHandler, 'write_postgresql_conf', Mock()) @patch.object(ConfigHandler, 'write_recovery_conf', Mock()) @@ -63,6 +64,7 @@ class TestPatroni(unittest.TestCase): self.assertRaises(SystemExit, _main) @patch('pkgutil.iter_importers', Mock(return_value=[MockFrozenImporter()])) + @patch('urllib3.PoolManager.request', Mock(side_effect=Exception)) @patch('sys.frozen', Mock(return_value=True), create=True) @patch.object(HTTPServer, '__init__', Mock()) @patch.object(etcd.Client, 'read', etcd_read) diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 036e225b..a3475ef5 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -346,8 +346,7 @@ class TestPostgresql(BaseTestPostgresql): @patch.object(Postgresql, 'start', Mock()) def test_follow(self): self.p.call_nowait(CallbackAction.ON_START) - m = RemoteMember.from_name_and_data('1', {'restore_command': '2', 'primary_slot_name': 'foo', - 'conn_kwargs': {'host': 'bar'}}) + m = RemoteMember('1', {'restore_command': '2', 'primary_slot_name': 'foo', 'conn_kwargs': {'host': 'bar'}}) self.p.follow(m) with patch.object(Postgresql, 'ensure_major_version_is_known', Mock(return_value=False)): self.assertIsNone(self.p.follow(m)) @@ -523,8 +522,9 @@ class TestPostgresql(BaseTestPostgresql): def test_save_configuration_files(self): self.p.config.save_configuration_files() - @patch('os.path.isfile', Mock(side_effect=[False, True])) - @patch('shutil.copy', Mock(side_effect=IOError)) + @patch('os.path.isfile', Mock(side_effect=[False, True, False, True])) + @patch('shutil.copy', Mock(side_effect=[None, IOError])) + @patch('os.chmod', Mock()) def test_restore_configuration_files(self): self.p.config.restore_configuration_files() diff --git a/tests/test_rewind.py b/tests/test_rewind.py index 22181266..f40c46af 100644 --- a/tests/test_rewind.py +++ b/tests/test_rewind.py @@ -253,8 +253,8 @@ class TestRewind(BaseTestPostgresql): mock_logger_info.call_args[0]) mock_logger_info.reset_mock() mock_subprocess_call.assert_called_once() - self.assertEqual(mock_subprocess_call.call_args.args[0], ['command 000000000000000000000000']) - self.assertEqual(mock_subprocess_call.call_args.kwargs['shell'], True) + self.assertEqual(mock_subprocess_call.call_args[0][0], ['command 000000000000000000000000']) + self.assertEqual(mock_subprocess_call.call_args[1]['shell'], True) mock_subprocess_call.reset_mock() # failed archive_command call diff --git a/tests/test_slots.py b/tests/test_slots.py index 6d3f17d3..a5be0d1d 100644 --- a/tests/test_slots.py +++ b/tests/test_slots.py @@ -67,6 +67,23 @@ class TestSlotsHandler(BaseTestPostgresql): with patch.object(Postgresql, 'major_version', PropertyMock(return_value=90618)): self.s.sync_replication_slots(cluster, False) + def test_cascading_replica_sync_replication_slots(self): + """Test sync with a cascading replica so physical slots are present on a replica.""" + config = ClusterConfig(1, {'slots': {'ls': {'database': 'a', 'plugin': 'b'}}}, 1) + cascading_replica = Member(0, 'test-2', 28, { + 'state': 'running', 'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5436/postgres', + 'tags': {'replicatefrom': 'postgresql0'} + }) + cluster = Cluster(True, config, self.leader, 0, + [self.me, self.other, self.leadermem, cascading_replica], + None, SyncState.empty(), None, {'ls': 10}, None) + self.p.set_role('replica') + with patch.object(Postgresql, '_query') as mock_query, \ + patch.object(Postgresql, 'is_leader', Mock(return_value=False)): + mock_query.return_value = [('ls', 'logical', 'b', 'a', 5, 12345, 105)] + ret = self.s.sync_replication_slots(cluster, False) + self.assertEqual(ret, []) + def test_process_permanent_slots(self): config = ClusterConfig(1, {'slots': {'ls': {'database': 'a', 'plugin': 'b'}}, 'ignore_slots': [{'name': 'blabla'}]}, 1)