From 480b8dbf95b77d221fc6446a2bb65c96234927bf Mon Sep 17 00:00:00 2001 From: Stan Bogatkin Date: Mon, 17 Jul 2023 15:55:06 +0300 Subject: [PATCH 01/26] Fix typo in yml files (#2760) Users statement was mentioned twice in templates - fix this simple typo by removing duplicates. --- postgres0.yml | 2 +- postgres1.yml | 2 +- postgres2.yml | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/postgres0.yml b/postgres0.yml index 2e83c7e1..0605c322 100644 --- a/postgres0.yml +++ b/postgres0.yml @@ -93,7 +93,7 @@ bootstrap: # Additional script to be launched after initial cluster creation (will be passed the connection URL as parameter) # post_init: /usr/local/bin/setup_cluster.sh - # Some additional users users which needs to be created after initializing new cluster + # Some additional users which needs to be created after initializing new cluster users: admin: password: admin% diff --git a/postgres1.yml b/postgres1.yml index e8b2806d..89dcca1a 100644 --- a/postgres1.yml +++ b/postgres1.yml @@ -87,7 +87,7 @@ bootstrap: # Additional script to be launched after initial cluster creation (will be passed the connection URL as parameter) # post_init: /usr/local/bin/setup_cluster.sh - # Some additional users users which needs to be created after initializing new cluster + # Some additional users which needs to be created after initializing new cluster users: admin: password: admin% diff --git a/postgres2.yml b/postgres2.yml index 8272e3ba..c77e734a 100644 --- a/postgres2.yml +++ b/postgres2.yml @@ -84,7 +84,7 @@ bootstrap: - encoding: UTF8 - data-checksums - # Some additional users users which needs to be created after initializing new cluster + # Some additional users which needs to be created after initializing new cluster users: admin: password: admin% From 0c5bf3c4cdda30ca6a0a7801e7eb1921ff5d1c10 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 19 Jul 2023 12:42:14 +0200 Subject: [PATCH 02/26] Validate more parameters in the config file (#2761) - parameters for different DCS - more bootstrap.dcs parameters - ctl, restapi, and watchdog parameters --- patroni/validator.py | 136 +++++++++++++++++++++++++++++++++++++--- tests/test_validator.py | 3 +- 2 files changed, 131 insertions(+), 8 deletions(-) diff --git a/patroni/validator.py b/patroni/validator.py index fc6cdde5..e9afcc13 100644 --- a/patroni/validator.py +++ b/patroni/validator.py @@ -11,11 +11,12 @@ import shutil import socket import subprocess -from typing import Any, Dict, Union, Iterator, List, Optional as OptionalType, TYPE_CHECKING +from typing import Any, Dict, Union, Iterator, List, Optional as OptionalType, Tuple, TYPE_CHECKING -from .utils import parse_int, split_host_port, data_directory_is_empty +from .collections import CaseInsensitiveSet from .dcs import dcs_modules from .exceptions import ConfigParseError +from .utils import parse_int, split_host_port, data_directory_is_empty def data_directory_empty(data_dir: str) -> bool: @@ -802,6 +803,39 @@ class IntValidator(object): return ret +class EnumValidator(object): + """Validate enum setting + + :ivar allowed_values: a ``set`` or ``CaseInsensitiveSet`` object with allowed enum values. + :ivar raise_assert: if an ``assert`` call should be performed regarding expected type and valid range. + """ + + def __init__(self, allowed_values: Tuple[str, ...], + case_sensitive: bool = False, raise_assert: bool = False) -> None: + """Create an :class:`EnumValidator` object with given allowed values. + + :param allowed_values: a tuple with allowed enum values + :param case_sensitive: set to ``True`` to do case sensitive comparisons + :param raise_assert: if an ``assert`` call should be performed regarding expected values. + """ + self.allowed_values = set(allowed_values) if case_sensitive else CaseInsensitiveSet(allowed_values) + self.raise_assert = raise_assert + + def __call__(self, value: str) -> bool: + """Check if provided *value* could be found within *allowed_values*. + + .. note:: + If ``raise_assert`` is ``True`` and *value* is not valid, then an ``AssertionError`` will be triggered. + :param value: value to be checked. + :returns: ``True`` if *value* could be found within *allowed_values*. + """ + ret = value in self.allowed_values + + if self.raise_assert: + assert_(ret) + return ret + + def validate_watchdog_mode(value: Any) -> None: """Validate ``watchdog.mode`` configuration option. @@ -827,15 +861,44 @@ validate_etcd = { "srv": str, "srv_suffix": str, "url": str, - "proxy": str}) + "proxy": str + }), + Optional("protocol"): str, + Optional("username"): str, + Optional("password"): str, + Optional("cacert"): str, + Optional("cert"): str, + Optional("key"): str } schema = Schema({ "name": str, "scope": str, + Optional("ctl"): { + Optional("insecure"): bool, + Optional("cacert"): str, + Optional("certfile"): str, + Optional("keyfile"): str, + Optional("keyfile_password"): str + }, "restapi": { "listen": validate_host_port_listen, "connect_address": validate_connect_address, + Optional("authentication"): { + "username": str, + "password": str + }, + Optional("certfile"): str, + Optional("keyfile"): str, + Optional("keyfile_password"): str, + Optional("cafile"): str, + Optional("ciphers"): str, + Optional("verify_client"): EnumValidator(("none", "optional", "required"), + case_sensitive=True, raise_assert=True), + Optional("allowlist"): [str], + Optional("allowlist_include_members"): bool, + Optional("http_extra_headers"): dict, + Optional("https_extra_headers"): dict, Optional("request_queue_size"): IntValidator(min=0, max=4096, raise_assert=True) }, Optional("bootstrap"): { @@ -843,15 +906,64 @@ schema = Schema({ Optional("ttl"): int, Optional("loop_wait"): int, Optional("retry_timeout"): int, - Optional("maximum_lag_on_failover"): int + Optional("maximum_lag_on_failover"): int, + Optional("maximum_lag_on_syncnode"): int, + Optional("postgresql"): { + Optional("parameters"): { + Optional("max_connections"): int, + Optional("max_locks_per_transaction"): int, + Optional("max_prepared_transactions"): int, + Optional("max_replication_slots"): int, + Optional("max_wal_senders"): int, + Optional("max_worker_processes"): int + }, + Optional("use_pg_rewind"): bool, + Optional("pg_hba"): [str], + Optional("pg_ident"): [str], + Optional("pg_ctl_timeout"): int, + Optional("use_slots"): bool, + }, + Optional("primary_start_timeout"): int, + Optional("primary_stop_timeout"): int, + Optional("standby_cluster"): { + Or("host", "port", "restore_command"): Case({ + "host": str, + "port": int, + "restore_command": str + }), + Optional("primary_slot_name"): str, + Optional("create_replica_methods"): [str], + Optional("archive_cleanup_command"): str, + Optional("recovery_min_apply_delay"): str + }, + Optional("synchronous_mode"): bool, + Optional("synchronous_mode_strict"): bool, + Optional("synchronous_node_count"): int }, - Optional("initdb"): [Or(str, dict)] + Optional("initdb"): [Or(str, dict)], + Optional("method"): str }, Or(*available_dcs): Case({ "consul": { Or("host", "url"): Case({ "host": validate_host_port, - "url": str}) + "url": str + }), + Optional("port"): int, + Optional("scheme"): str, + Optional("token"): str, + Optional("verify"): bool, + Optional("cacert"): str, + Optional("cert"): str, + Optional("key"): str, + Optional("dc"): str, + Optional("checks"): [str], + Optional("register_service"): bool, + Optional("service_tags"): [str], + Optional("service_check_interval"): str, + Optional("service_check_tls_server_name"): str, + Optional("consistency"): EnumValidator(('default', 'consistent', 'stale'), + case_sensitive=True, raise_assert=True) }, "etcd": validate_etcd, "etcd3": validate_etcd, @@ -869,15 +981,24 @@ schema = Schema({ }, "zookeeper": { "hosts": Or(comma_separated_host_port, [validate_host_port]), + Optional("use_ssl"): bool, + Optional("cacert"): str, + Optional("cert"): str, + Optional("key"): str, + Optional("key_password"): str, + Optional("verify"): bool, + Optional("set_acls"): dict }, "kubernetes": { "labels": {}, + Optional("bypass_api_service"): bool, Optional("namespace"): str, Optional("scope_label"): str, Optional("role_label"): str, Optional("use_endpoints"): bool, Optional("pod_ip"): Or(is_ipv4_address, is_ipv6_address), Optional("ports"): [{"name": str, "port": int}], + Optional("cacert"): str, Optional("retriable_http_codes"): Or(int, [int]), }, }), @@ -915,7 +1036,8 @@ schema = Schema({ }, Optional("watchdog"): { Optional("mode"): validate_watchdog_mode, - Optional("device"): str + Optional("device"): str, + Optional("safety_margin"): int }, Optional("tags"): { Optional("nofailover"): bool, diff --git a/tests/test_validator.py b/tests/test_validator.py index 4c195a0e..24379c6b 100644 --- a/tests/test_validator.py +++ b/tests/test_validator.py @@ -15,7 +15,8 @@ config = { "scope": "string", "restapi": { "listen": "127.0.0.2:800", - "connect_address": "127.0.0.2:800" + "connect_address": "127.0.0.2:800", + "verify_client": 'none' }, "bootstrap": { "dcs": { From 4830e36e2b1002daa4c182ed1b4c07882268bbe5 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Thu, 20 Jul 2023 13:23:55 +0200 Subject: [PATCH 03/26] Start primary back when it crashed with "in crash recovery" state (#2763) If one of backends crashed postmaster stops all backend and does crash recovery because shared memory could be corrupted. If something happens during this phase postgres may completely crash leaving pg_control with "in crash recovery" state. Followup on #2726 --- patroni/ha.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/patroni/ha.py b/patroni/ha.py index 1148802a..f704500e 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -473,7 +473,8 @@ class Ha(object): # timeout > 0 indicates that we still have the leader lock, and it was just updated if timeout\ - and data.get('Database cluster state') in ('in production', 'shutting down', 'shut down')\ + and data.get('Database cluster state') in ('in production', 'in crash recovery', + 'shutting down', 'shut down')\ and self.state_handler.state == 'crashed'\ and self.state_handler.role in ('primary', 'master')\ and not self.state_handler.config.recovery_conf_exists(): From cb9998ade67b69adebf31b09364591c3230df4bb Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Thu, 20 Jul 2023 13:24:17 +0200 Subject: [PATCH 04/26] Update docstring in do_GET_metrics() (#2765) followup #2733 --- patroni/api.py | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/patroni/api.py b/patroni/api.py index 836a8673..8d991d5f 100644 --- a/patroni/api.py +++ b/patroni/api.py @@ -463,6 +463,12 @@ class RestApiHandler(BaseHTTPRequestHandler): * ``patroni_dcs_last_seen``: epoch timestamp when DCS was last contacted successfully; * ``patroni_pending_restart``: ``1`` if this PostgreSQL node is pending a restart, else ``0``; * ``patroni_is_paused``: ``1`` if Patroni is in maintenance node, else ``0``. + + For PostgreSQL v9.6+ the response will also have the following: + + * ``patroni_postgres_streaming``: 1 if Postgres is streaming from another node, else ``0``; + * ``patroni_postgres_in_archive_recovery``: ``1`` if Postgres isn't streaming and + there is ``restore_command`` available, else ``0``. """ postgres = self.get_postgresql_status(True) patroni = self.server.patroni From 84c574e1ecbdec11f860a11eed675f6789a698db Mon Sep 17 00:00:00 2001 From: Waynerv Date: Fri, 21 Jul 2023 20:30:22 +0800 Subject: [PATCH 05/26] Run archive_command through shell (#2766) Close #2764 --- patroni/postgresql/rewind.py | 2 +- tests/test_rewind.py | 6 +++++- 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/patroni/postgresql/rewind.py b/patroni/postgresql/rewind.py index f70ad065..ff6fc751 100644 --- a/patroni/postgresql/rewind.py +++ b/patroni/postgresql/rewind.py @@ -370,7 +370,7 @@ class Rewind(object): # it is the author of archive_command, who is responsible # for not overriding the WALs already present in archive logger.info('Trying to archive %s: %s', wal, cmd) - if self._postgresql.cancellable.call(shlex.split(cmd)) == 0: + if self._postgresql.cancellable.call([cmd], shell=True) == 0: new_name = os.path.join(status_dir, wal + '.done') try: shutil.move(old_name, new_name) diff --git a/tests/test_rewind.py b/tests/test_rewind.py index b5858d8e..a4fafcff 100644 --- a/tests/test_rewind.py +++ b/tests/test_rewind.py @@ -239,7 +239,7 @@ class TestRewind(BaseTestPostgresql): with patch('os.listdir', Mock(return_value=['000000000000000000000000.ready'])): # successful archive_command call - with patch.object(CancellableSubprocess, 'call', Mock(return_value=0)): + with patch.object(CancellableSubprocess, 'call', Mock(return_value=0)) as mock_subprocess_call: get_guc_value_res = [ 'on', 'command %f', 'always', 'command %f', @@ -252,6 +252,10 @@ class TestRewind(BaseTestPostgresql): '000000000000000000000000', 'command 000000000000000000000000'), 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) + mock_subprocess_call.reset_mock() # failed archive_command call with patch.object(CancellableSubprocess, 'call', Mock(return_value=1)): From ffd1ad97d23eeb9f9cf7c2bda66dd85df8225bd7 Mon Sep 17 00:00:00 2001 From: Polina Bungina <27892524+hughcapet@users.noreply.github.com> Date: Fri, 21 Jul 2023 15:10:13 +0200 Subject: [PATCH 06/26] Fix Dockerfile_s (#2770) * Install dumb-init using apt * Remove python 2.7 packages purge --- Dockerfile | 5 ++--- Dockerfile.citus | 5 ++--- 2 files changed, 4 insertions(+), 6 deletions(-) diff --git a/Dockerfile b/Dockerfile index 4aa0ce5c..c5b927ee 100644 --- a/Dockerfile +++ b/Dockerfile @@ -25,8 +25,7 @@ RUN set -ex \ | grep -Ev '^python3-(sphinx|etcd|consul|kazoo|kubernetes)' \ | xargs apt-get install -y vim curl less jq locales haproxy sudo \ python3-etcd python3-kazoo python3-pip busybox \ - net-tools iputils-ping --fix-missing \ - && pip3 install dumb-init \ + net-tools iputils-ping dumb-init --fix-missing \ \ # Cleanup all locales but en_US.UTF-8 && find /usr/share/i18n/charmaps/ -type f ! -name UTF-8.gz -delete \ @@ -71,7 +70,7 @@ RUN set -ex \ # Clean up all useless packages and some files && apt-get purge -y --allow-remove-essential python3-pip gzip bzip2 util-linux e2fsprogs \ libmagic1 bsdmainutils login ncurses-bin libmagic-mgc e2fslibs bsdutils \ - exim4-config gnupg-agent dirmngr libpython2.7-stdlib libpython2.7-minimal \ + exim4-config gnupg-agent dirmngr \ git make \ && apt-get autoremove -y \ && apt-get clean -y \ diff --git a/Dockerfile.citus b/Dockerfile.citus index 8693b504..7e6ec18c 100644 --- a/Dockerfile.citus +++ b/Dockerfile.citus @@ -25,7 +25,7 @@ RUN set -ex \ | grep -Ev '^python3-(sphinx|etcd|consul|kazoo|kubernetes)' \ | xargs apt-get install -y vim curl less jq locales haproxy sudo \ python3-etcd python3-kazoo python3-pip busybox \ - net-tools iputils-ping lsb-release --fix-missing \ + net-tools iputils-ping lsb-release dumb-init --fix-missing \ && if [ $(dpkg --print-architecture) = 'arm64' ]; then \ apt-get install -y postgresql-server-dev-$PG_MAJOR \ git gcc make autoconf \ @@ -42,7 +42,6 @@ RUN set -ex \ && apt-get update -y \ && apt-get -y install postgresql-$PG_MAJOR-citus-11.3; \ fi \ - && pip3 install dumb-init \ \ # Cleanup all locales but en_US.UTF-8 && find /usr/share/i18n/charmaps/ -type f ! -name UTF-8.gz -delete \ @@ -88,7 +87,7 @@ RUN set -ex \ # Clean up all useless packages and some files && apt-get purge -y --allow-remove-essential python3-pip gzip bzip2 util-linux e2fsprogs \ libmagic1 bsdmainutils login ncurses-bin libmagic-mgc e2fslibs bsdutils \ - exim4-config gnupg-agent dirmngr libpython2.7-stdlib libpython2.7-minimal \ + exim4-config gnupg-agent dirmngr \ postgresql-server-dev-$PG_MAJOR git gcc make autoconf \ libc6-dev flex libicu-dev libkrb5-dev liblz4-dev \ libpam0g-dev libreadline-dev libselinux1-dev libssl-dev libxslt1-dev libzstd-dev uuid-dev \ From 0a8fb0860ec5e9cb5f9b65a4c81b123e4a3d77ba Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Fri, 21 Jul 2023 16:09:34 +0200 Subject: [PATCH 07/26] Skip flaky scenario when running with Raft (#2771) Sometimes Patroni doesn't see the latest Raft data on start. --- features/basic_replication.feature | 1 + features/environment.py | 2 ++ 2 files changed, 3 insertions(+) diff --git a/features/basic_replication.feature b/features/basic_replication.feature index 0e2e8c4b..b58520bd 100644 --- a/features/basic_replication.feature +++ b/features/basic_replication.feature @@ -79,6 +79,7 @@ Feature: basic replication When I add the table buz to postgres2 Then table buz is present on postgres0 after 20 seconds + @reject-duplicate-name Scenario: check graceful rejection when two nodes have the same name Given I start duplicate postgres0 on port 8011 Then there is a "Can't start; there is already a node named 'postgres0' running" CRITICAL in the dup-postgres0 patroni log diff --git a/features/environment.py b/features/environment.py index f7690ad2..5906becd 100644 --- a/features/environment.py +++ b/features/environment.py @@ -1144,3 +1144,5 @@ def before_scenario(context, scenario): break if 'dcs-failsafe' in scenario.effective_tags and not context.dcs_ctl._handle: scenario.skip('it is not possible to control state of {0} from tests'.format(context.dcs_ctl.name())) + if 'reject-duplicate-name' in scenario.effective_tags and context.dcs_ctl.name() == 'raft': + scenario.skip('Flaky test with Raft') From 48164774c227a102f881e289751c4b3c379eb2c1 Mon Sep 17 00:00:00 2001 From: Matt Baker <93600443+matthbakeredb@users.noreply.github.com> Date: Mon, 24 Jul 2023 13:57:34 +0100 Subject: [PATCH 08/26] Refactor get replication slots (#2746) Reduce complexity of single method and allow for documentation of distinct parts. No functional changes have been introduced. --- patroni/dcs/__init__.py | 127 ++++++++++++++++++++++++++++++++-------- 1 file changed, 103 insertions(+), 24 deletions(-) diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index 2f270dc0..7bcc7f14 100644 --- a/patroni/dcs/__init__.py +++ b/patroni/dcs/__init__.py @@ -586,24 +586,25 @@ class Cluster(NamedTuple): def get_replication_slots(self, my_name: str, role: str, nofailover: bool, major_version: int, show_error: bool = False) -> Dict[str, Dict[str, Any]]: - # if the replicatefrom tag is set on the member - we should not create the replication slot for it on - # the current primary, because that member would replicate from elsewhere. We still create the slot if - # the replicatefrom destination member is currently not a member of the cluster (fallback to the - # primary), or if replicatefrom destination member happens to be the current primary - use_slots = self.use_slots - if role in ('master', 'primary', 'standby_leader'): - slot_members = [m.name for m in self.members if use_slots and m.name != my_name - and (m.replicatefrom is None or m.replicatefrom == my_name - or not self.has_member(m.replicatefrom))] - permanent_slots = self.__permanent_slots if use_slots and \ - role in ('master', 'primary') else self.__permanent_physical_slots - else: - # only manage slots for replicas that replicate from this one, except for the leader among them - slot_members = [m.name for m in self.members if use_slots - and m.replicatefrom == my_name and m.name != self.leader_name] - permanent_slots = self.__permanent_logical_slots if use_slots and not nofailover else {} + """Lookup configured slot names in the DCS, report issues found and merge with permanent slots. - slots = {slot_name_from_member_name(name): {'type': 'physical'} for name in slot_members} + Will log an error if: + + * Conflicting slot names between members are found + * Any logical slots are disabled, due to version compatibility, and *show_error* is ``True``. + + :param my_name: name of this node. + :param role: role of this node. + :param nofailover: ``True`` if this node is tagged to not be a failover candidate. + :param major_version: postgresql major version. + :param show_error: if ``True`` report error if any disabled logical slots or conflicting slot names are found. + + :returns: final dictionary of slot names, after merging with permanent slots and performing sanity checks. + """ + slot_members: List[str] = self._get_slot_members(my_name, role) if self.use_slots else [] + + slots: Dict[str, Dict[str, str]] = {slot_name_from_member_name(name): {'type': 'physical'} + for name in slot_members} if len(slots) < len(slot_members): # Find which names are conflicting for a nicer error message @@ -611,11 +612,38 @@ class Cluster(NamedTuple): for name in slot_members: slot_conflicts[slot_name_from_member_name(name)].append(name) logger.error("Following cluster members share a replication slot name: %s", - "; ".join("{} map to {}".format(", ".join(v), k) + "; ".join(f"{', '.join(v)} map to {k}" for k, v in slot_conflicts.items() if len(v) > 1)) - # "merge" replication slots for members with permanent_replication_slots + permanent_slots: dict[str, Any] = self._get_permanent_slots(role, nofailover) if self.use_slots else {} + disabled_permanent_logical_slots: List[str] = self._merge_permanent_slots( + slots, permanent_slots, my_name, major_version) + + if disabled_permanent_logical_slots and show_error: + logger.error("Permanent logical replication slots supported by Patroni only starting from PostgreSQL 11. " + "Following slots will not be created: %s.", disabled_permanent_logical_slots) + + return slots + + @staticmethod + def _merge_permanent_slots(slots: Dict[str, Dict[str, str]], permanent_slots: Dict[str, Any], my_name: str, + major_version: int) -> List[str]: + """Merge replication *slots* for members with *permanent_slots*. + + Perform validation of configured permanent slot name, skipping invalid names. + + Will update *slots* in-line based on ``type`` of slot, ``physical`` or ``logical``, and name of node. + Type is assumed to be ``physical`` if there are no attributes stored as the slot value. + + :param slots: Slot names with existing attributes if known. + :param my_name: name of this node. + :param permanent_slots: dictionary containing slot name key and slot information values. + :param major_version: postgresql major version. + + :returns: List of disabled permanent, logical slot names, if postgresql version < 11. + """ disabled_permanent_logical_slots: List[str] = [] + for name, value in permanent_slots.items(): if not slot_name_re.match(name): logger.error("Invalid permanent replication slot name '%s'", name) @@ -632,7 +660,8 @@ class Cluster(NamedTuple): if name != slot_name_from_member_name(my_name): slots[name] = value continue - elif value['type'] == 'logical' and value.get('database') and value.get('plugin'): + + if value['type'] == 'logical' and value.get('database') and value.get('plugin'): if major_version < 110000: disabled_permanent_logical_slots.append(name) elif name in slots: @@ -643,12 +672,62 @@ class Cluster(NamedTuple): continue logger.error("Bad value for slot '%s' in permanent_slots: %s", name, permanent_slots[name]) + return disabled_permanent_logical_slots - if disabled_permanent_logical_slots and show_error: - logger.error("Permanent logical replication slots supported by Patroni only starting from PostgreSQL 11. " - "Following slots will not be created: %s.", disabled_permanent_logical_slots) + def _get_permanent_slots(self, role: str, nofailover: bool) -> Dict[str, Any]: + """Get configured permanent slot names. - return slots + .. note:: + Permanent logical replication slots are only considered if ``use_slots`` configuration is enabled. Also, + only considered if *role* is ``primary`` or if it is a promotable ``replica`` -- what excludes a + ``standby_leader`` or ``replica`` with ``nofailover`` tag enabled. That combination is used for failing + over logical replication slots, and the latter nodes are not eligible for such task. + + Permanent physical slots are only considered if *role* is ``primary`` or ``standby_leader``, independently + if ``use_slots`` is enabled or not. That is done that way because even if Patroni itself is not using slots + to replicate among its members when ``use_slots`` is disabled, the user may still have configured Patroni to + keep permanent physical slots used out of Patroni. + + :param role: role of this node -- ``primary``, ``standby_leader`` or ``replica``. + or logical slots being consumed. + :param nofailover: ``True`` if this node is tagged to not be a failover candidate. + + :returns: dictionary of permanent slot names mapped to attributes. + """ + if role in ('master', 'primary', 'standby_leader'): + permanent_slots = (self.__permanent_slots + if role in ('master', 'primary') + else self.__permanent_physical_slots) + else: + permanent_slots = self.__permanent_logical_slots if not nofailover else {} + return permanent_slots + + def _get_slot_members(self, my_name: str, role: str) -> List[str]: + """Get a list of member names that have replication slots sourcing from this node. + + If the ``replicatefrom`` tag is set on the member - we should not create the replication slot for it on + the current primary, because that member would replicate from elsewhere. We still create the slot if + the ``replicatefrom`` destination member is currently not a member of the cluster (fallback to the + primary), or if ``replicatefrom`` destination member happens to be the current primary. + + :param my_name: name of this node. + :param role: role of this node, if this is a ``primary`` or ``standby_leader`` return list of members + replicating from this node. If not then return a list of members replicating as cascaded + replicas from this node. + + :returns: list of member names. + """ + if role in ('master', 'primary', 'standby_leader'): + slot_members = [m.name for m in self.members + if m.name != my_name + and (m.replicatefrom is None + or m.replicatefrom == my_name + or not self.has_member(m.replicatefrom))] + else: + # only manage slots for replicas that replicate from this one, except for the leader among them + slot_members = [m.name for m in self.members + if m.replicatefrom == my_name and m.name != self.leader_name] + return slot_members def has_permanent_logical_slots(self, my_name: str, nofailover: bool, major_version: int = 110000) -> bool: if major_version < 110000: From e860cac348e5d33c64c4835dc09d9616758474f9 Mon Sep 17 00:00:00 2001 From: Polina Bungina <27892524+hughcapet@users.noreply.github.com> Date: Mon, 24 Jul 2023 16:23:18 +0200 Subject: [PATCH 09/26] Fix manual failover/switchover checks (#2769) In case of manual failover/switchover failover possibility should be checked only against the candidate --- patroni/ha.py | 38 ++++++++++++++++++++++++++++++-------- 1 file changed, 30 insertions(+), 8 deletions(-) diff --git a/patroni/ha.py b/patroni/ha.py index f704500e..5894824c 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -1085,7 +1085,9 @@ 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.cluster.members, cluster_lsn=checkpoint_location): + + # for a manual failover/switchover with a candidate, we should check the requested candidate only + if self.is_failover_possible(self.get_failover_candidates(), cluster_lsn=checkpoint_location): self.state_handler.set_role('demoted') with self._async_executor: self.release_leader_key_voluntarily(checkpoint_location) @@ -1189,15 +1191,12 @@ class Ha(object): logger.warning('Failover is possible only to a specific candidate in a paused state') else: if self.is_synchronous_mode(): - if failover.candidate and not self.cluster.sync.matches(failover.candidate): + members = self.get_failover_candidates(check_sync=True) + if failover.candidate and not members: logger.warning('Failover candidate=%s does not match with sync_standbys=%s', failover.candidate, self.cluster.sync.sync_standby) - members = [] - else: - members = [m for m in self.cluster.members if self.cluster.sync.matches(m.name)] else: - members = [m for m in self.cluster.members - if not failover.candidate or m.name == failover.candidate] + members = self.get_failover_candidates() if self.is_failover_possible(members, False): # check that there are healthy members ret = self._async_executor.try_run_async('manual failover: demote', self.demote, ('graceful',)) return ret or 'manual failover: demoting myself' @@ -1849,7 +1848,9 @@ 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.cluster.members, cluster_lsn=checkpoint_location): + + # for a manual failover/switchover with a candidate, we should check the requested candidate only + if self.is_failover_possible(self.get_failover_candidates(), cluster_lsn=checkpoint_location): self.dcs.delete_leader(checkpoint_location) status['deleted'] = True else: @@ -1909,3 +1910,24 @@ class Ha(object): name = member.name if member else 'remote_member:{}'.format(uuid.uuid1()) return RemoteMember.from_name_and_data(name, data) + + def get_failover_candidates(self, check_sync: bool = False) -> List[Member]: + """Return list of candidates for either manual or automatic failover. + + Mainly used to later be passed to ``Ha.is_failover_possible()``. + + :param check_sync: if ``True``, also check against the sync key members + + :returns: a list of ``Member`` ojects or an empty list if there is no candidate available + """ + failover = self.cluster.failover + if check_sync: + # TODO: allow manual failover (=no leader specified) to async node + # every sync_standby or the candidate specified if is in sync_standbys + 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)] + else: + # every member or the candidate specified + return [m for m in self.cluster.members + if not failover or not failover.candidate or m.name == failover.candidate] From c5a4befdc45809f43b1fece2ede6600e81f1784b Mon Sep 17 00:00:00 2001 From: Matt Baker <93600443+matthbakeredb@users.noreply.github.com> Date: Tue, 25 Jul 2023 07:00:59 +0100 Subject: [PATCH 10/26] Refactor `check_logical_slots_readiness` split to reduce complexity (#2749) Includes: * renaming of `_unready_logical_slots` to better represent that it is a processing queue which is emptied on successful completion. * ensuring that return type is consistent. * made logic variable names explicit to help explain how the decision of whether a slot is "ready" is made. --- patroni/postgresql/slots.py | 129 +++++++++++++++++++++++++----------- tests/test_slots.py | 8 +-- 2 files changed, 95 insertions(+), 42 deletions(-) diff --git a/patroni/postgresql/slots.py b/patroni/postgresql/slots.py index d08a09c3..fde81a7a 100644 --- a/patroni/postgresql/slots.py +++ b/patroni/postgresql/slots.py @@ -130,7 +130,7 @@ class SlotsHandler(object): self._postgresql = postgresql self._advance = None self._replication_slots: Dict[str, Dict[str, Any]] = {} # already existing replication slots - self._unready_logical_slots: Dict[str, Optional[int]] = {} + self._logical_slots_processing_queue: Dict[str, Optional[int]] = {} self.pg_replslot_dir = os.path.join(self._postgresql.data_dir, 'pg_replslot') self.schedule() @@ -190,7 +190,8 @@ class SlotsHandler(object): self._replication_slots = replication_slots self._schedule_load_slots = False if self._force_readiness_check: - self._unready_logical_slots = {n: None for n, v in replication_slots.items() if v['type'] == 'logical'} + self._logical_slots_processing_queue = {n: None for n, v in replication_slots.items() + if v['type'] == 'logical'} self._force_readiness_check = False def ignore_replication_slot(self, cluster: Cluster, name: str) -> bool: @@ -327,10 +328,10 @@ class SlotsHandler(object): self._ensure_physical_slots(slots) if self._postgresql.is_leader(): - self._unready_logical_slots.clear() + self._logical_slots_processing_queue.clear() self._ensure_logical_slots_primary(slots) elif cluster.slots and slots: - self.check_logical_slots_readiness(cluster, nofailover, replicatefrom) + self.check_logical_slots_readiness(cluster, replicatefrom) ret = self._ensure_logical_slots_replica(cluster, slots) @@ -347,48 +348,100 @@ class SlotsHandler(object): with get_connection_cursor(connect_timeout=3, options="-c statement_timeout=2000", **conn_kwargs) as cur: yield cur - def check_logical_slots_readiness(self, cluster: Cluster, nofailover: bool, replicatefrom: Optional[str]) -> None: + 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. + + :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. + """ catalog_xmin = None - if self._unready_logical_slots and cluster.leader: + if self._logical_slots_processing_queue and cluster.leader: slot_name = cluster.get_my_slot_name_on_primary(self._postgresql.name, replicatefrom) try: with self._get_leader_connection_cursor(cluster.leader) as cur: cur.execute("SELECT slot_name, catalog_xmin FROM pg_catalog.pg_get_replication_slots()" " WHERE NOT pg_catalog.pg_is_in_recovery() AND slot_name = ANY(%s)", - ([n for n, v in self._unready_logical_slots.items() if v is None] + [slot_name],)) + ([n for n, v in self._logical_slots_processing_queue.items() + if v is None] + [slot_name],)) slots = {row[0]: row[1] for row in cur} if slot_name not in slots: - return logger.warning('Physical slot %s does not exist on the primary', slot_name) + logger.warning('Physical slot %s does not exist on the primary', slot_name) + return False catalog_xmin = slots.pop(slot_name) except Exception as e: - return logger.error("Failed to check %s physical slot on the primary: %r", slot_name, e) - # Remember catalog_xmin of logical slots on the primary when catalog_xmin of - # the physical slot became valid. Logical slots on replica will be safe to use after - # promote when catalog_xmin of the physical slot overtakes these values. - if catalog_xmin is not None: - for name, value in slots.items(): - self._unready_logical_slots[name] = value - else: # Replica isn't streaming or the hot_standby_feedback isn't enabled - try: - cur = self._query("SELECT pg_catalog.current_setting('hot_standby_feedback')::boolean") - row = cur.fetchone() - if row and not row[0]: - logger.error('Logical slot failover requires "hot_standby_feedback".' - ' Please check postgresql.auto.conf') - except Exception as e: - logger.error('Failed to check the hot_standby_feedback setting: %r', e) - return # since `catalog_xmin` isn't valid further checks don't make any sense + logger.error("Failed to check %s physical slot on the primary: %r", slot_name, e) + return False - for name in list(self._unready_logical_slots): - value = self._replication_slots.get(name) - # The logical slot on a replica is safe to use when the physical replica slot on the primary: - # 1. has a nonzero/non-null catalog_xmin - # 2. has a catalog_xmin that is not newer (greater) than the catalog_xmin of any slot on the standby - # 3. overtook the catalog_xmin of remembered values of logical slots on the primary. - if not value or catalog_xmin is not None and\ - self._unready_logical_slots[name] <= catalog_xmin <= value['catalog_xmin']: - del self._unready_logical_slots[name] - if value: + if not self._update_pending_logical_slot_primary(slots, catalog_xmin): + return False # since `catalog_xmin` isn't valid further checks don't make any sense + + self._ready_logical_slots(catalog_xmin) + return True + + def _update_pending_logical_slot_primary(self, slots: Dict[str, Any], catalog_xmin: Optional[int] = None) -> bool: + """Store pending logical slot information for ``catalog_xmin`` on the primary. + + Remember ``catalog_xmin`` of logical slots on the primary when ``catalog_xmin`` of the physical slot became + valid. Logical slots on replica will be safe to use after promote when ``catalog_xmin`` of the physical slot + overtakes these values. + + :param slots: dictionary of slot information from the primary + :param catalog_xmin: ``catalog_xmin`` of the physical slot used by this replica to stream changes from primary. + + :returns: ``False`` if any issue was faced while processing, ``True`` otherwise. + """ + if catalog_xmin is not None: + for name, value in slots.items(): + self._logical_slots_processing_queue[name] = value + return True + + # Replica isn't streaming or the hot_standby_feedback isn't enabled + try: + cur = self._query("SELECT pg_catalog.current_setting('hot_standby_feedback')::boolean") + row = cur.fetchone() + if row and not row[0]: + logger.error('Logical slot failover requires "hot_standby_feedback".' + ' Please check postgresql.auto.conf') + except Exception as e: + logger.error('Failed to check the hot_standby_feedback setting: %r', e) + return False + + def _ready_logical_slots(self, primary_physical_catalog_xmin: Optional[int] = None) -> None: + """Ready logical slots by comparing primary physical slot ``catalog_xmin`` to logical ``catalog_xmin``. + + The logical slot on a replica is safe to use when the physical replica slot on the primary: + + 1. has a nonzero/non-null ``catalog_xmin`` represented by ``primary_physical_xmin``. + 2. has a ``catalog_xmin`` that is not newer (greater) than the ``catalog_xmin`` of any slot on the standby + 3. overtook the ``catalog_xmin`` of remembered values of logical slots on the primary. + + :param primary_physical_catalog_xmin: is the value retrieved from ``pg_catalog.pg_get_replication_slots()`` for + the physical replication slot on the primary. + """ + # Make a copy of processing queue keys as a list as the queue dictionary is modified inside the loop. + for name in list(self._logical_slots_processing_queue): + primary_logical_catalog_xmin = self._logical_slots_processing_queue[name] + standby_logical_slot = self._replication_slots.get(name, {}) + standby_logical_catalog_xmin = standby_logical_slot.get('catalog_xmin', 0) + if TYPE_CHECKING: # pragma: no cover + assert primary_logical_catalog_xmin is not None + + if ( + not standby_logical_slot + or primary_physical_catalog_xmin is not None + and primary_logical_catalog_xmin <= primary_physical_catalog_xmin <= standby_logical_catalog_xmin + ): + + del self._logical_slots_processing_queue[name] + + if standby_logical_slot: 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: @@ -433,7 +486,7 @@ class SlotsHandler(object): shutil.rmtree(slot_dir) os.rename(slot_tmp_dir, slot_dir) fsync_dir(slot_dir) - self._unready_logical_slots[name] = None + self._logical_slots_processing_queue[name] = None fsync_dir(self._postgresql.slots_handler.pg_replslot_dir) self._postgresql.start() @@ -446,6 +499,6 @@ class SlotsHandler(object): if self._advance: self._advance.on_promote() - if self._unready_logical_slots: + if self._logical_slots_processing_queue: logger.warning('Logical replication slots that might be unsafe to use after promote: %s', - set(self._unready_logical_slots)) + set(self._logical_slots_processing_queue)) diff --git a/tests/test_slots.py b/tests/test_slots.py index a962dbe3..3f21998f 100644 --- a/tests/test_slots.py +++ b/tests/test_slots.py @@ -93,7 +93,7 @@ class TestSlotsHandler(BaseTestPostgresql): def test__ensure_logical_slots_replica(self): self.p.set_role('replica') self.cluster.slots['ls'] = 12346 - with patch.object(SlotsHandler, 'check_logical_slots_readiness', Mock()): + with patch.object(SlotsHandler, 'check_logical_slots_readiness', Mock(return_value=False)): self.assertEqual(self.s.sync_replication_slots(self.cluster, False), []) self.s._schedule_load_slots = False with patch.object(MockCursor, 'execute', Mock(side_effect=psycopg.OperationalError)),\ @@ -121,12 +121,12 @@ class TestSlotsHandler(BaseTestPostgresql): self.s.copy_logical_slots(self.cluster, ['ls']) with patch.object(MockCursor, '__iter__', Mock(return_value=iter([('postgresql0', None)]))),\ patch.object(MockCursor, 'fetchone', Mock(side_effect=Exception)): - self.assertIsNone(self.s.check_logical_slots_readiness(self.cluster, False, None)) + self.assertFalse(self.s.check_logical_slots_readiness(self.cluster, None)) with patch.object(MockCursor, '__iter__', Mock(return_value=iter([('postgresql0', None)]))),\ patch.object(MockCursor, 'fetchone', Mock(return_value=(False,))): - self.assertIsNone(self.s.check_logical_slots_readiness(self.cluster, False, None)) + self.assertFalse(self.s.check_logical_slots_readiness(self.cluster, None)) with patch.object(MockCursor, '__iter__', Mock(return_value=iter([('ls', 100)]))): - self.s.check_logical_slots_readiness(self.cluster, False, None) + self.s.check_logical_slots_readiness(self.cluster, None) @patch.object(Postgresql, 'stop', Mock(return_value=True)) @patch.object(Postgresql, 'start', Mock(return_value=True)) From 817f39ad6d8e16f95a39060564c9f8a1065e770f Mon Sep 17 00:00:00 2001 From: Matt Baker <93600443+matthbakeredb@users.noreply.github.com> Date: Tue, 25 Jul 2023 07:01:35 +0100 Subject: [PATCH 11/26] Refactor get_dcs (#2747) Now uses generators instead of for loops and implements importing modules once. --- patroni/dcs/__init__.py | 107 +++++++++++++++++++++++++++++----------- tests/test_ha.py | 2 + 2 files changed, 79 insertions(+), 30 deletions(-) diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index 7bcc7f14..f07ce6cd 100644 --- a/patroni/dcs/__init__.py +++ b/patroni/dcs/__init__.py @@ -15,7 +15,9 @@ from collections import defaultdict from copy import deepcopy from random import randint from threading import Event, Lock -from typing import Any, Callable, Collection, Dict, List, NamedTuple, Optional, Set, Tuple, Union, TYPE_CHECKING +from types import ModuleType +from typing import Any, Callable, Collection, Dict, List, NamedTuple, Optional, Set, Tuple, Union, TYPE_CHECKING, \ + Type, Iterator from urllib.parse import urlparse, urlunparse, parse_qsl from ..exceptions import PatroniFatalException @@ -85,38 +87,83 @@ def dcs_modules() -> List[str]: return [module_prefix + name for _, name, is_pkg in pkgutil.iter_modules([dcs_dirname]) if not is_pkg] -def get_dcs(config: Union['Config', Dict[str, Any]]) -> 'AbstractDCS': - modules = dcs_modules() +def iter_dcs_classes( + config: Optional[Union['Config', Dict[str, Any]]] = None +) -> Iterator[Tuple[str, Type['AbstractDCS']]]: + """Attempt to import DCS modules that are present in the given configuration. + + .. note:: + If a module successfully imports we can assume that all its requirements are installed. + + :param config: configuration information with possible DCS names as keys. If given, only attempt to import DCS + modules defined in the configuration. Else, if ``None``, attempt to import any supported DCS module. + + :yields: a tuple containing the module ``name`` and the imported DCS class object. + """ + for mod_name in dcs_modules(): + name = mod_name.rpartition('.')[2] + if config is None or name in config: - for module_name in modules: - name = module_name.split('.')[-1] - if name in config: # we will try to import only modules which have configuration section in the config file try: - module = importlib.import_module(module_name) - for key, item in module.__dict__.items(): # iterate through the module content - # try to find implementation of AbstractDCS interface, class name must match with module_name - if key.lower() == name and inspect.isclass(item) and issubclass(item, AbstractDCS): - # propagate some parameters - config[name].update({p: config[p] for p in ('namespace', 'name', 'scope', 'loop_wait', - 'patronictl', 'ttl', 'retry_timeout') if p in config}) - # From citus section we only need "group" parameter, but will propagate everything just in case. - if isinstance(config.get('citus'), dict): - config[name].update(config['citus']) - return item(config[name]) - except ImportError: - logger.debug('Failed to import %s', module_name) + module = importlib.import_module(mod_name) + dcs_module = find_dcs_class_in_module(module) + if dcs_module: + yield name, dcs_module - available_implementations: List[str] = [] - for module_name in modules: - name = module_name.split('.')[-1] - try: - module = importlib.import_module(module_name) - available_implementations.extend(name for key, item in module.__dict__.items() if key.lower() == name - and inspect.isclass(item) and issubclass(item, AbstractDCS)) - except ImportError: - logger.info('Failed to import %s', module_name) - raise PatroniFatalException("""Can not find suitable configuration of distributed configuration store -Available implementations: """ + ', '.join(sorted(set(available_implementations)))) + except ImportError: + logger.log(logging.DEBUG if config is not None else logging.INFO, + 'Failed to import %s', mod_name) + + +def find_dcs_class_in_module(module: ModuleType) -> Optional[Type['AbstractDCS']]: + """Try to find the implementation of :class:`AbstractDCS` interface in *module* matching the *module* name. + + :param module: Imported DCS module. + + :returns: class with a name matching the name of *module* that implements :class:`AbstractDCS` or ``None`` if not + found. + """ + module_name = module.__name__.rpartition('.')[2] + return next( + (obj for obj_name, obj in module.__dict__.items() + if (obj_name.lower() == module_name + and inspect.isclass(obj) and issubclass(obj, AbstractDCS))), + None) + + +def get_dcs(config: Union['Config', Dict[str, Any]]) -> 'AbstractDCS': + """Attempt to load a Distributed Configuration Store from known available implementations. + + .. note:: + Using the list of available DCS modules returned by :func:`iter_dcs_modules` attempt to dynamically import and + instantiate the class that implements a DCS using the abstract class :class:`AbstractDCS`. + + Basic top-level configuration parameters retrieved from *config* are propagated to the DCS specific config + before being passed to the module DCS class. + + If no module is found to satisfy configuration then report and log an error. This will cause Patroni to exit. + + :raises :exc:`PatroniFatalException`: if a load of all available DCS modules have been tried and none succeeded. + + :param config: object or dictionary with Patroni configuration. This is normally a representation of the main + Patroni + + :returns: The first successfully loaded DCS module which is an implementation of :class:`AbstractDCS`. + """ + for name, dcs_class in iter_dcs_classes(config): + # Propagate some parameters from top level of config if defined to the DCS specific config section. + config[name].update({ + p: config[p] for p in ('namespace', 'name', 'scope', 'loop_wait', + 'patronictl', 'ttl', 'retry_timeout') + if p in config}) + # From citus section we only need "group" parameter, but will propagate everything just in case. + if isinstance(config.get('citus'), dict): + config[name].update(config['citus']) + return dcs_class(config[name]) + + raise PatroniFatalException( + f"Can not find suitable configuration of distributed configuration store\n" + f"Available implementations: {', '.join(sorted([n for n, _ in iter_dcs_classes()]))}") _Version = Union[int, str] diff --git a/tests/test_ha.py b/tests/test_ha.py index dad9562f..005e922e 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -278,8 +278,10 @@ class TestHa(PostgresInit): self.p.follow = true self.assertEqual(self.ha.run_cycle(), 'starting as a secondary') self.p.is_running = true + ha_dcs_orig_name = self.ha.dcs.__class__.__name__ self.ha.dcs.__class__.__name__ = 'Raft' self.assertEqual(self.ha.run_cycle(), 'started as a secondary') + self.ha.dcs.__class__.__name__ = ha_dcs_orig_name def test_recover_former_primary(self): self.p.follow = false From 06db2966122cc3bf0291169b3cd2a2e93a522445 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 25 Jul 2023 08:48:18 +0200 Subject: [PATCH 12/26] Fixes in `patroni.request` (#2768) 1. Take client certificates only from the `ctl` section. Motivation: sometimes there are server-only certificates that can't be used as client certificates. As a result neither Patroni not patronictl work correctly even if `--insecure` option is used. 2. Document that if `restapi.verify_client` is set to `required` then client certificates **must** be provided in the `ctl` section. 3. Add support for `ctl.authentication` and prefer to use it over `restapi.authentication`. 4. Silence annoying InsecureRequestWarning when `patronictl -k` is used, so that behavior becomes is similar to `curl -k`. --- docs/ENVIRONMENT.rst | 19 ++++++-- docs/yaml_configuration.rst | 16 +++++-- features/environment.py | 4 +- patroni/config.py | 14 +++--- patroni/ctl.py | 3 +- patroni/request.py | 76 +++++++++++++++++++----------- tests/test_ctl.py | 4 +- typings/urllib3/__init__.pyi | 3 +- typings/urllib3/connectionpool.pyi | 2 + 9 files changed, 94 insertions(+), 47 deletions(-) create mode 100644 typings/urllib3/connectionpool.pyi diff --git a/docs/ENVIRONMENT.rst b/docs/ENVIRONMENT.rst index 91816184..6fbfe4e0 100644 --- a/docs/ENVIRONMENT.rst +++ b/docs/ENVIRONMENT.rst @@ -196,10 +196,19 @@ REST API - **PATRONI\_RESTAPI\_HTTPS\_EXTRA\_HEADERS**: (optional) HTTPS headers let the REST API server pass additional information with an HTTP response when TLS is enabled. This will also pass additional information set in ``http_extra_headers``. - **PATRONI\_RESTAPI\_REQUEST\_QUEUE\_SIZE**: (optional): Sets request queue size for TCP socket used by Patroni REST API. Once the queue is full, further requests get a "Connection denied" error. The default value is 5. +.. warning:: + + - The ``PATRONI_RESTAPI_CONNECT_ADDRESS`` must be accessible from all nodes of a given Patroni cluster. Internally Patroni is using it during the leader race to find nodes with minimal replication lag. + - If you enabled client certificates validation (``PATRONI_RESTAPI_VERIFY_CLIENT`` is set to ``required``), you also **must** provide **valid client certificates** in the ``PATRONI_CTL_CERTFILE``, ``PATRONI_CTL_KEYFILE``, ``PATRONI_CTL_KEYFILE_PASSWORD``. If not provided, Patroni will not work correctly. + + CTL --- -- **PATRONICTL\_CONFIG\_FILE**: location of the configuration file. -- **PATRONI\_CTL\_INSECURE**: Allow connections to REST API without verifying SSL certs. -- **PATRONI\_CTL\_CACERT**: Specifies the file with the CA_BUNDLE file or directory with certificates of trusted CAs to use while verifying REST API SSL certs. If not provided patronictl will use the value provided for REST API "cafile" parameter. -- **PATRONI\_CTL\_CERTFILE**: Specifies the file with the client certificate in the PEM format. If not provided patronictl will use the value provided for REST API "certfile" parameter. -- **PATRONI\_CTL\_KEYFILE**: Specifies the file with the client secret key in the PEM format. If not provided patronictl will use the value provided for REST API "keyfile" parameter. +- **PATRONICTL\_CONFIG\_FILE**: (optional) location of the configuration file. +- **PATRONI\_CTL\_USERNAME**: (optional) Basic-auth username for accessing protected REST API endpoints. If not provided patronictl will use the value provided for REST API "username" parameter. +- **PATRONI\_CTL\_PASSWORD**: (optional) Basic-auth password for accessing protected REST API endpoints. If not provided patronictl will use the value provided for REST API "password" parameter. +- **PATRONI\_CTL\_INSECURE**: (optional) Allow connections to REST API without verifying SSL certs. +- **PATRONI\_CTL\_CACERT**: (optional) Specifies the file with the CA_BUNDLE file or directory with certificates of trusted CAs to use while verifying REST API SSL certs. If not provided patronictl will use the value provided for REST API "cafile" parameter. +- **PATRONI\_CTL\_CERTFILE**: (optional) Specifies the file with the client certificate in the PEM format. +- **PATRONI\_CTL\_KEYFILE**: (optional) Specifies the file with the client secret key in the PEM format. +- **PATRONI\_CTL\_KEYFILE\_PASSWORD**: (optional) Specifies a password for decrypting the client keyfile. diff --git a/docs/yaml_configuration.rst b/docs/yaml_configuration.rst index f58c9b0a..fbb2de68 100644 --- a/docs/yaml_configuration.rst +++ b/docs/yaml_configuration.rst @@ -335,17 +335,27 @@ Here is an example of both **http_extra_headers** and **https_extra_headers**: https_extra_headers: 'Strict-Transport-Security': 'max-age=31536000; includeSubDomains' +.. warning:: + + - The ``restapi.connect_address`` must be accessible from all nodes of a given Patroni cluster. Internally Patroni is using it during the leader race to find nodes with minimal replication lag. + - 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. + + .. _patronictl_settings: CTL --- - **ctl**: (optional) + - **authentication**: + + - **username**: Basic-auth username for accessing protected REST API endpoints. If not provided patronictl will use the value provided for REST API "username" parameter. + - **password**: Basic-auth password for accessing protected REST API endpoints. If not provided patronictl will use the value provided for REST API "password" parameter. - **insecure**: Allow connections to REST API without verifying SSL certs. - **cacert**: Specifies the file with the CA_BUNDLE file or directory with certificates of trusted CAs to use while verifying REST API SSL certs. If not provided patronictl will use the value provided for REST API "cafile" parameter. - - **certfile**: Specifies the file with the client certificate in the PEM format. If not provided patronictl will use the value provided for REST API "certfile" parameter. - - **keyfile**: Specifies the file with the client secret key in the PEM format. If not provided patronictl will use the value provided for REST API "keyfile" parameter. - - **keyfile\_password**: Specifies a password for decrypting the keyfile. If not provided patronictl will use the value provided for REST API "keyfile\_password" parameter. + - **certfile**: Specifies the file with the client certificate in the PEM format. + - **keyfile**: Specifies the file with the client secret key in the PEM format. + - **keyfile\_password**: Specifies a password for decrypting the client keyfile. Watchdog -------- diff --git a/features/environment.py b/features/environment.py index 5906becd..ec503bb5 100644 --- a/features/environment.py +++ b/features/environment.py @@ -1082,7 +1082,9 @@ def before_all(context): 'PATRONI_RESTAPI_CERTFILE': context.certfile, 'PATRONI_RESTAPI_KEYFILE': context.keyfile, 'PATRONI_RESTAPI_VERIFY_CLIENT': 'required', - 'PATRONI_CTL_INSECURE': 'on'}) + 'PATRONI_CTL_INSECURE': 'on', + 'PATRONI_CTL_CERTFILE': context.certfile, + 'PATRONI_CTL_KEYFILE': context.keyfile}) ctl.update({'cacert': context.certfile, 'certfile': context.certfile, 'keyfile': context.keyfile}) context.request_executor = PatroniRequest({'ctl': ctl}, True) context.dcs_ctl = context.pctl.known_dcs[context.pctl.dcs](context) diff --git a/patroni/config.py b/patroni/config.py index aa71d821..365abc24 100644 --- a/patroni/config.py +++ b/patroni/config.py @@ -452,9 +452,10 @@ class Config(object): ret[param] = value return ret - restapi_auth = _get_auth('restapi') - if restapi_auth: - ret['restapi']['authentication'] = restapi_auth + for section in ('ctl', 'restapi'): + auth = _get_auth(section) + if auth: + ret[section]['authentication'] = auth authentication = {} for user_type in ('replication', 'superuser', 'rewind'): @@ -531,9 +532,10 @@ class Config(object): elif name not in config or name in ['watchdog']: config[name] = deepcopy(value) if value else {} - # restapi server expects to get restapi.auth = 'username:password' - if 'restapi' in config and 'authentication' in config['restapi']: - config['restapi']['auth'] = '{username}:{password}'.format(**config['restapi']['authentication']) + # restapi server expects to get restapi.auth = 'username:password' and similarly for `ctl` + for section in ('ctl', 'restapi'): + if section in config and 'authentication' in config[section]: + config[section]['auth'] = '{username}:{password}'.format(**config[section]['authentication']) # special treatment for old config diff --git a/patroni/ctl.py b/patroni/ctl.py index 55a054c1..bb6b4b1c 100644 --- a/patroni/ctl.py +++ b/patroni/ctl.py @@ -254,7 +254,6 @@ arg_cluster_name = click.argument('cluster_name', required=False, option_default_citus_group = click.option('--group', required=False, type=int, help='Citus group', default=lambda: click.get_current_context().obj.get('citus', {}).get('group')) option_citus_group = click.option('--group', required=False, type=int, help='Citus group') -option_insecure = click.option('-k', '--insecure', is_flag=True, help='Allow connections to SSL sites without certs') role_choice = click.Choice(['leader', 'primary', 'standby-leader', 'replica', 'standby', 'any', 'master']) @@ -262,7 +261,7 @@ role_choice = click.Choice(['leader', 'primary', 'standby-leader', 'replica', 's @click.option('--config-file', '-c', help='Configuration file', envvar='PATRONICTL_CONFIG_FILE', default=CONFIG_FILE_PATH) @click.option('--dcs-url', '--dcs', '-d', 'dcs_url', help='The DCS connect url', envvar='DCS_URL') -@option_insecure +@click.option('-k', '--insecure', is_flag=True, help='Allow connections to SSL sites without certs') @click.pass_context def ctl(ctx: click.Context, config_file: str, dcs_url: Optional[str], insecure: bool) -> None: """Entry point of ``patronictl`` utility. diff --git a/patroni/request.py b/patroni/request.py index d063310d..0421bf86 100644 --- a/patroni/request.py +++ b/patroni/request.py @@ -11,6 +11,19 @@ from .dcs import Member from .utils import USER_AGENT +class HTTPSConnectionPool(urllib3.HTTPSConnectionPool): + + def _validate_conn(self, *args: Any, **kwargs: Any) -> None: + """Override parent method to silence warnings about requests without certificate verification enabled.""" + + +class PatroniPoolManager(urllib3.PoolManager): + + def __init__(self, *args: Any, **kwargs: Any) -> None: + super(PatroniPoolManager, self).__init__(*args, **kwargs) + self.pool_classes_by_scheme = {'http': urllib3.HTTPConnectionPool, 'https': HTTPSConnectionPool} + + class PatroniRequest(object): """Wrapper for performing requests to Patroni's REST API. @@ -28,22 +41,30 @@ class PatroniRequest(object): * If none of the above applies, then it falls back to ``False``. """ self._insecure = insecure - self._pool = urllib3.PoolManager(num_pools=10, maxsize=10) + self._pool = PatroniPoolManager(num_pools=10, maxsize=10) self.reload_config(config) @staticmethod - def _get_cfg_value(config: Union[Config, Dict[str, Any]], name: str) -> Union[Any, None]: - """Get value of *name* setting in *config*. - - .. note:: - *name* key will be searched only under ``ctl`` and ``restapi`` sections, in that order. + def _get_ctl_value(config: Union[Config, Dict[str, Any]], name: str, default: Any = None) -> Optional[Any]: + """Get value of *name* setting from the ``ctl`` section of the *config*. :param config: Patroni YAML configuration. :param name: name of the setting value to be retrieved. - :returns: value of ``ctl -> *name*`` or ``restapi -> *name*``, if either is present, ``None`` otherwise. + :returns: value of ``ctl -> *name*`` if present, ``None`` otherwise. """ - return config.get('ctl', {}).get(name) or config.get('restapi', {}).get(name) + return config.get('ctl', {}).get(name, default) + + @staticmethod + def _get_restapi_value(config: Union[Config, Dict[str, Any]], name: str) -> Optional[Any]: + """Get value of *name* setting from the ``restapi`` section of the *config*. + + :param config: Patroni YAML configuration. + :param name: name of the setting value to be retrieved. + + :returns: value of ``restapi -> *name*`` if present, ``None`` otherwise. + """ + return config.get('restapi', {}).get(name) def _apply_pool_param(self, param: str, value: Any) -> None: """Configure *param* as *value* in the request manager. @@ -65,12 +86,11 @@ class PatroniRequest(object): * ``cert``: gets translated to ``certfile`` * ``key``: gets translated to ``keyfile`` - Will attempt to fetch the requested key first from ``ctl`` section, and fall back to ``restapi`` section - if the former is missing. + Will attempt to fetch the requested key first from ``ctl`` section. - :returns: value of ``ctl -> *name*file`` or ``restapi -> *name*file`` if either is present, ``None`` otherwise. + :returns: value of ``ctl -> *name*file`` if present, ``None`` otherwise. """ - value = self._get_cfg_value(config, name + 'file') + value = self._get_ctl_value(config, name + 'file') self._apply_pool_param(name + '_file', value) return value @@ -79,37 +99,39 @@ class PatroniRequest(object): Configure these HTTP headers for requests: - * ``authorization``: based on Patroni' REST API authentication config; + * ``authorization``: based on Patroni' CTL or REST API authentication config; * ``user-agent``: based on `patroni.utils.USER_AGENT`. Also configure SSL related settings for requests: * ``ca_certs`` is configured if ``ctl -> cacert`` or ``restapi -> cafile`` is available; - * ``cert``, ``key`` and ``key_password`` are configured if ``ctl -> certile`` or ``restapi -> certfile`` is - available. + * ``cert``, ``key`` and ``key_password`` are configured if ``ctl -> certfile`` is available. :param config: Patroni YAML configuration. """ - # ``restapi -> auth`` is equivalent to ``restapi -> authentication -> username`` + ``:`` + - # ``restapi -> authentication -> password`` - self._pool.headers = urllib3.make_headers(basic_auth=self._get_cfg_value(config, 'auth'), user_agent=USER_AGENT) + # ``ctl -> auth`` is equivalent to ``ctl -> authentication -> username`` + ``:`` + + # ``ctl -> authentication -> password``. And the same for ``restapi -> auth`` + basic_auth = self._get_ctl_value(config, 'auth') or self._get_restapi_value(config, 'auth') + self._pool.headers = urllib3.make_headers(basic_auth=basic_auth, user_agent=USER_AGENT) + self._pool.connection_pool_kw['cert_reqs'] = 'CERT_REQUIRED' + + insecure = self._insecure if isinstance(self._insecure, bool)\ + else self._get_ctl_value(config, 'insecure', False) - insecure = self._insecure if isinstance(self._insecure, bool) else config.get('ctl', {}).get('insecure', False) if self._apply_ssl_file_param(config, 'cert'): - # With client certificate the cert_reqs must be set to CERT_REQUIRED even if insecure option is used - self._pool.connection_pool_kw['cert_reqs'] = 'CERT_REQUIRED' - # The assert_hostname = False helps to silence warnings - self._pool.connection_pool_kw['assert_hostname'] = False if insecure else None + if insecure: # The assert_hostname = False helps to silence warnings + self._pool.connection_pool_kw['assert_hostname'] = False self._apply_ssl_file_param(config, 'key') - - password = self._get_cfg_value(config, 'keyfile_password') + password = self._get_ctl_value(config, 'keyfile_password') self._apply_pool_param('key_password', password) else: - self._pool.connection_pool_kw['cert_reqs'] = 'CERT_NONE' if insecure else 'CERT_REQUIRED' + if insecure: # Disable server certificate validation if requested + self._pool.connection_pool_kw['cert_reqs'] = 'CERT_NONE' + self._pool.connection_pool_kw.pop('assert_hostname', None) self._pool.connection_pool_kw.pop('key_file', None) - cacert = config.get('ctl', {}).get('cacert') or config.get('restapi', {}).get('cafile') + cacert = self._get_ctl_value(config, 'cacert') or self._get_restapi_value(config, 'cafile') self._apply_pool_param('ca_certs', cacert) def request(self, method: str, url: str, body: Optional[Any] = None, diff --git a/tests/test_ctl.py b/tests/test_ctl.py index f7d3eec5..b30cecc1 100644 --- a/tests/test_ctl.py +++ b/tests/test_ctl.py @@ -22,7 +22,7 @@ from .test_ha import get_cluster_initialized_without_leader, get_cluster_initial @patch('patroni.ctl.load_config', Mock(return_value={ - 'scope': 'alpha', 'restapi': {'listen': '::', 'certfile': 'a'}, + 'scope': 'alpha', 'restapi': {'listen': '::', 'certfile': 'a'}, 'ctl': {'certfile': 'a'}, 'etcd': {'host': 'localhost:2379'}, 'citus': {'database': 'citus', 'group': 0}, 'postgresql': {'data_dir': '.', 'pgpass': './pgpass', 'parameters': {}, 'retry_timeout': 5}})) class TestCtl(unittest.TestCase): @@ -451,7 +451,7 @@ class TestCtl(unittest.TestCase): mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader for role in self.TEST_ROLES: - result = self.runner.invoke(ctl, ['flush', 'dummy', 'restart', '-r', role], input='y') + result = self.runner.invoke(ctl, ['-k', 'flush', 'dummy', 'restart', '-r', role], input='y') assert 'No scheduled restart' in result.output result = self.runner.invoke(ctl, ['flush', 'dummy', 'restart', '--force']) diff --git a/typings/urllib3/__init__.pyi b/typings/urllib3/__init__.pyi index ff573518..61b5b66c 100644 --- a/typings/urllib3/__init__.pyi +++ b/typings/urllib3/__init__.pyi @@ -1,6 +1,7 @@ +from .connectionpool import HTTPConnectionPool, HTTPSConnectionPool from .poolmanager import PoolManager from .response import HTTPResponse from .util.request import make_headers from .util.timeout import Timeout -__all__ = ['HTTPResponse', 'PoolManager', 'Timeout', 'make_headers'] +__all__ = ['HTTPResponse', 'HTTPConnectionPool', 'HTTPSConnectionPool', 'PoolManager', 'Timeout', 'make_headers'] diff --git a/typings/urllib3/connectionpool.pyi b/typings/urllib3/connectionpool.pyi new file mode 100644 index 00000000..b2e8d3db --- /dev/null +++ b/typings/urllib3/connectionpool.pyi @@ -0,0 +1,2 @@ +class HTTPConnectionPool: ... +class HTTPSConnectionPool(HTTPConnectionPool): ... From 0e19e3e98e89f1aba0584e55cc53489d98553b30 Mon Sep 17 00:00:00 2001 From: Waynerv Date: Tue, 25 Jul 2023 16:29:04 +0800 Subject: [PATCH 13/26] Make pod role label configurable (#2659) Close #2495 --- docs/ENVIRONMENT.rst | 6 ++++- docs/kubernetes.rst | 49 +++++++++++++++++++++++++++++++++++++ docs/yaml_configuration.rst | 6 ++++- patroni/config.py | 3 ++- patroni/dcs/kubernetes.py | 23 ++++++++++++++--- patroni/validator.py | 4 +++ tests/test_kubernetes.py | 22 +++++++++++++++++ 7 files changed, 106 insertions(+), 7 deletions(-) diff --git a/docs/ENVIRONMENT.rst b/docs/ENVIRONMENT.rst index 6fbfe4e0..00595cf7 100644 --- a/docs/ENVIRONMENT.rst +++ b/docs/ENVIRONMENT.rst @@ -112,7 +112,11 @@ Kubernetes - **PATRONI\_KUBERNETES\_NAMESPACE**: (optional) Kubernetes namespace where the Patroni pod is running. Default value is `default`. - **PATRONI\_KUBERNETES\_LABELS**: Labels in format ``{label1: value1, label2: value2}``. These labels will be used to find existing objects (Pods and either Endpoints or ConfigMaps) associated with the current cluster. Also Patroni will set them on every object (Endpoint or ConfigMap) it creates. - **PATRONI\_KUBERNETES\_SCOPE\_LABEL**: (optional) name of the label containing cluster name. Default value is `cluster-name`. -- **PATRONI\_KUBERNETES\_ROLE\_LABEL**: (optional) name of the label containing Postgres role (`master` or `replica`). Patroni will set this label on the pod it is running in. Default value is `role`. +- **PATRONI\_KUBERNETES\_ROLE\_LABEL**: (optional) name of the label containing role (master or replica or other custom value). Patroni will set this label on the pod it runs in. Default value is ``role``. +- **PATRONI\_KUBERNETES\_LEADER\_LABEL\_VALUE**: (optional) value of the pod label when Postgres role is `master`. Default value is `master`. +- **PATRONI\_KUBERNETES\_FOLLOWER\_LABEL\_VALUE**: (optional) value of the pod label when Postgres role is `replica`. Default value is `replica`. +- **PATRONI\_KUBERNETES\_STANDBY\_LEADER\_LABEL\_VALUE**: (optional) value of the pod label when Postgres role is ``standby-leader``. Default value is ``standby-leader``. +- **PATRONI\_KUBERNETES\_TMP\_ROLE\_LABEL**: (optional) name of the temporary label containing role (master or replica). Value of this label will always use the default of corresponding role. Set only when necessary. - **PATRONI\_KUBERNETES\_USE\_ENDPOINTS**: (optional) if set to true, Patroni will use Endpoints instead of ConfigMaps to run leader elections and keep cluster state. - **PATRONI\_KUBERNETES\_POD\_IP**: (optional) IP address of the pod Patroni is running in. This value is required when `PATRONI_KUBERNETES_USE_ENDPOINTS` is enabled and is used to populate the leader endpoint subsets when the pod's PostgreSQL is promoted. - **PATRONI\_KUBERNETES\_PORTS**: (optional) if the Service object has the name for the port, the same name must appear in the Endpoint object, otherwise service won't work. For example, if your service is defined as ``{Kind: Service, spec: {ports: [{name: postgresql, port: 5432, targetPort: 5432}]}}``, then you have to set ``PATRONI_KUBERNETES_PORTS='[{"name": "postgresql", "port": 5432}]'`` and Patroni will use it for updating subsets of the leader Endpoint. This parameter is used only if `PATRONI_KUBERNETES_USE_ENDPOINTS` is set. diff --git a/docs/kubernetes.rst b/docs/kubernetes.rst index 699321fa..ca7313bf 100644 --- a/docs/kubernetes.rst +++ b/docs/kubernetes.rst @@ -32,6 +32,55 @@ Configuration Patroni Kubernetes :ref:`settings ` and :ref:`environment variables ` are described in the general chapters of the documentation. +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`. + +Note that if you migrate from default role labels to custom ones, you can reduce downtime by following migration steps: + +1. Add a temporary label using original role value for the pod with `kubernetes.tmp_role_label` (like ``tmp_role``). Once pods are restarted they will get following labels set by Patroni: + + .. code:: YAML + + labels: + cluster-name: foo + role: master + tmp_role: master + +2. After all pods have been updated, modify the service selector to select the temporary label. + + .. code:: YAML + + selector: + cluster-name: foo + tmp_role: master + +3. Add your custom role label (e.g., set `kubernetes.leader_label_value=primary`). Once pods are restarted they will get following new labels set by Patroni: + + .. code:: YAML + + labels: + cluster-name: foo + role: primary + tmp_role: master + +4. After all pods have been updated again, modify the service selector to use new role value. + + .. code:: YAML + + selector: + cluster-name: foo + role: primary + +5. Finally, remove the temporary label from your configuration and update all pods. + + .. code:: YAML + + labels: + cluster-name: foo + role: primary + Examples -------- diff --git a/docs/yaml_configuration.rst b/docs/yaml_configuration.rst index fbb2de68..ab5b5587 100644 --- a/docs/yaml_configuration.rst +++ b/docs/yaml_configuration.rst @@ -155,7 +155,11 @@ Kubernetes - **namespace**: (optional) Kubernetes namespace where Patroni pod is running. Default value is `default`. - **labels**: Labels in format ``{label1: value1, label2: value2}``. These labels will be used to find existing objects (Pods and either Endpoints or ConfigMaps) associated with the current cluster. Also Patroni will set them on every object (Endpoint or ConfigMap) it creates. - **scope\_label**: (optional) name of the label containing cluster name. Default value is `cluster-name`. -- **role\_label**: (optional) name of the label containing role (master or replica). Patroni will set this label on the pod it runs in. Default value is ``role``. +- **role\_label**: (optional) name of the label containing role (master or replica or other custom value). Patroni will set this label on the pod it runs in. Default value is ``role``. +- **leader\_label\_value**: (optional) value of the pod label when Postgres role is ``master``. Default value is ``master``. +- **follower\_label\_value**: (optional) value of the pod label when Postgres role is ``replica``. Default value is ``replica``. +- **standby\_leader\_label\_value**: (optional) value of the pod label when Postgres role is ``standby-leader``. Default value is ``standby-leader``. +- **tmp_\role\_label**: (optional) name of the temporary label containing role (master or replica). Value of this label will always use the default of corresponding role. Set only when necessary. - **use\_endpoints**: (optional) if set to true, Patroni will use Endpoints instead of ConfigMaps to run leader elections and keep cluster state. - **pod\_ip**: (optional) IP address of the pod Patroni is running in. This value is required when `use_endpoints` is enabled and is used to populate the leader endpoint subsets when the pod's PostgreSQL is promoted. - **ports**: (optional) if the Service object has the name for the port, the same name must appear in the Endpoint object, otherwise service won't work. For example, if your service is defined as ``{Kind: Service, spec: {ports: [{name: postgresql, port: 5432, targetPort: 5432}]}}``, then you have to set ``kubernetes.ports: [{"name": "postgresql", "port": 5432}]`` and Patroni will use it for updating subsets of the leader Endpoint. This parameter is used only if `kubernetes.use_endpoints` is set. diff --git a/patroni/config.py b/patroni/config.py index 365abc24..66ffd891 100644 --- a/patroni/config.py +++ b/patroni/config.py @@ -475,7 +475,8 @@ class Config(object): 'REGISTER_SERVICE', 'SERVICE_CHECK_INTERVAL', 'SERVICE_CHECK_TLS_SERVER_NAME', 'SERVICE_TAGS', 'NAMESPACE', 'CONTEXT', 'USE_ENDPOINTS', 'SCOPE_LABEL', 'ROLE_LABEL', 'POD_IP', 'PORTS', 'LABELS', 'BYPASS_API_SERVICE', 'RETRIABLE_HTTP_CODES', 'KEY_PASSWORD', - 'USE_SSL', 'SET_ACLS', 'GROUP', 'DATABASE') and name: + 'USE_SSL', 'SET_ACLS', 'GROUP', 'DATABASE', 'LEADER_LABEL_VALUE', 'FOLLOWER_LABEL_VALUE', + 'STANDBY_LEADER_LABEL_VALUE', 'TMP_ROLE_LABEL') and name: value = os.environ.pop(param) if name == 'CITUS': if suffix == 'GROUP': diff --git a/patroni/dcs/kubernetes.py b/patroni/dcs/kubernetes.py index f770ac64..bda4bdc8 100644 --- a/patroni/dcs/kubernetes.py +++ b/patroni/dcs/kubernetes.py @@ -752,6 +752,10 @@ class Kubernetes(AbstractDCS): self._label_selector = ','.join('{0}={1}'.format(k, v) for k, v in self._labels.items()) self._namespace = config.get('namespace') or 'default' self._role_label = config.get('role_label', 'role') + self._leader_label_value = config.get('leader_label_value', 'master') + self._follower_label_value = config.get('follower_label_value', 'replica') + self._standby_leader_label_value = config.get('standby_leader_label_value', 'standby-leader') + self._tmp_role_label = config.get('tmp_role_label') self._ca_certs = os.environ.get('PATRONI_KUBERNETES_CACERT', config.get('cacert')) or SERVICE_CERT_FILENAME super(Kubernetes, self).__init__({**config, 'namespace': ''}) if self._citus_group: @@ -1263,19 +1267,30 @@ class Kubernetes(AbstractDCS): def touch_member(self, data: Dict[str, Any]) -> bool: cluster = self.cluster if cluster and cluster.leader and cluster.leader.name == self._name: - role = 'master' + role = self._leader_label_value + tmp_role = 'master' elif data['state'] == 'running' and data['role'] not in ('master', 'primary'): - role = data['role'] + role = { + 'replica': self._follower_label_value, + 'standby-leader': self._standby_leader_label_value, + }.get(data['role'], data['role']) + tmp_role = data['role'] else: role = None + tmp_role = None + + role_labels = {self._role_label: role} + if self._tmp_role_label: + role_labels[self._tmp_role_label] = tmp_role member = cluster and cluster.get_member(self._name, fallback_to_leader=False) pod_labels = member and member.data.pop('pod_labels', None) ret = member and pod_labels is not None\ - and pod_labels.get(self._role_label) == role and deep_compare(data, member.data) + and all(pod_labels.get(k) == v for k, v in role_labels.items())\ + and deep_compare(data, member.data) if not ret: - metadata = {'namespace': self._namespace, 'name': self._name, 'labels': {self._role_label: role}, + metadata = {'namespace': self._namespace, 'name': self._name, 'labels': role_labels, 'annotations': {'status': json.dumps(data, separators=(',', ':'))}} body = k8s_client.V1Pod(metadata=k8s_client.V1ObjectMeta(**metadata)) ret = self._api.patch_namespaced_pod(self._name, self._namespace, body) diff --git a/patroni/validator.py b/patroni/validator.py index e9afcc13..b68f9654 100644 --- a/patroni/validator.py +++ b/patroni/validator.py @@ -995,6 +995,10 @@ schema = Schema({ Optional("namespace"): str, Optional("scope_label"): str, Optional("role_label"): str, + Optional("leader_label_value"): str, + Optional("follower_label_value"): str, + Optional("standby_leader_label_value"): str, + Optional("tmp_role_label"): str, Optional("use_endpoints"): bool, Optional("pod_ip"): Or(is_ipv4_address, is_ipv6_address), Optional("ports"): [{"name": str, "port": int}], diff --git a/tests/test_kubernetes.py b/tests/test_kubernetes.py index cfed1559..f79539a7 100644 --- a/tests/test_kubernetes.py +++ b/tests/test_kubernetes.py @@ -298,6 +298,28 @@ class TestKubernetesConfigMaps(BaseTestKubernetes): self.k.touch_member({'state': 'running', 'role': 'replica'}) self.k.touch_member({'state': 'stopped', 'role': 'primary'}) + self.k._role_label = 'isMaster' + self.k._leader_label_value = 'true' + self.k._follower_label_value = 'false' + self.k._standby_leader_label_value = 'false' + self.k._tmp_role_label = 'tmp_role' + + 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.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.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') + def test_initialize(self): self.k.initialize() From ae2bbd28ae62142f4f473f30693d12ced3b06aff Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 25 Jul 2023 11:50:40 +0200 Subject: [PATCH 14/26] Fix `in_recovery` check (#2773) The primary that is still alive wasn't properly recognized. Regression was introduced in #2652 --- patroni/ha.py | 2 +- tests/test_ha.py | 5 ++++- 2 files changed, 5 insertions(+), 2 deletions(-) diff --git a/patroni/ha.py b/patroni/ha.py index 5894824c..59c57565 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -56,7 +56,7 @@ class _MemberStatus(NamedTuple): # If one of those is not in a response we want to count the node as not healthy/reachable wal: Dict[str, Any] = json.get('wal') or json['xlog'] # abuse difference in primary/replica response format - in_recovery = not bool(wal.get('location')) or json.get('role') in ('master', 'primary') + in_recovery = not (bool(wal.get('location')) or json.get('role') in ('master', 'primary')) timeline = json.get('timeline', 0) dcs_last_seen = json.get('dcs_last_seen', 0) lsn = int(in_recovery and max(wal.get('received_location', 0), wal.get('replayed_location', 0))) diff --git a/tests/test_ha.py b/tests/test_ha.py index 005e922e..7cdb00e1 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -885,7 +885,10 @@ class TestHa(PostgresInit): member = Member(0, 'test', 1, {'api_url': 'http://127.0.0.1:8011/patroni'}) self.ha.fetch_node_status(member) member = Member(0, 'test', 1, {'api_url': 'http://localhost:8011/patroni'}) - self.ha.fetch_node_status(member) + self.ha.patroni.request = Mock() + self.ha.patroni.request.return_value.data = b'{"wal":{"location":1},"role":"primary"}' + ret = self.ha.fetch_node_status(member) + self.assertFalse(ret.in_recovery) @patch.object(Rewind, 'pg_rewind', true) @patch.object(Rewind, 'check_leader_is_not_in_recovery', true) From 238aba39568b80417e0746262805985c34476d52 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 26 Jul 2023 12:33:17 +0200 Subject: [PATCH 15/26] Fix patronictl list (#2775) the `Cluster` name field was missing in tsv, json, and yaml formats The bug was introduced in #2652 --- patroni/ctl.py | 2 +- tests/test_ctl.py | 6 +++++- 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/patroni/ctl.py b/patroni/ctl.py index bb6b4b1c..312c0cec 100644 --- a/patroni/ctl.py +++ b/patroni/ctl.py @@ -1543,7 +1543,7 @@ def output_members(obj: Dict[str, Any], cluster: Cluster, name: str, logging.debug(member) lag = member.get('lag', '') - member.update(c=name, member=member['name'], group=g, + member.update(cluster=name, member=member['name'], group=g, host=member.get('host', ''), tl=member.get('timeline', ''), role=member['role'].replace('_', ' ').title(), lag_in_mb=round(lag / 1024 / 1024) if isinstance(lag, int) else lag, diff --git a/tests/test_ctl.py b/tests/test_ctl.py index b30cecc1..f4f08296 100644 --- a/tests/test_ctl.py +++ b/tests/test_ctl.py @@ -83,9 +83,13 @@ class TestCtl(unittest.TestCase): scheduled_at = datetime.now(tzutc) + timedelta(seconds=600) cluster = get_cluster_initialized_with_leader(Failover(1, 'foo', 'bar', scheduled_at)) del cluster.members[1].data['conn_url'] - for fmt in ('pretty', 'json', 'yaml', 'tsv', 'topology'): + for fmt in ('pretty', 'json', 'yaml', 'topology'): self.assertIsNone(output_members({}, cluster, name='abc', fmt=fmt)) + with patch('click.echo') as mock_echo: + self.assertIsNone(output_members({}, cluster, name='abc', fmt='tsv')) + self.assertEqual(mock_echo.call_args[0][0], 'abc\tother\t\tReplica\trunning\t\tunknown') + @patch('patroni.ctl.get_dcs') @patch.object(PoolManager, 'request', Mock(return_value=MockResponse())) def test_switchover(self, mock_get_dcs): From 384a2a4d8f579d5151cb5bb2d03f96644279275d Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Thu, 27 Jul 2023 13:38:24 +0200 Subject: [PATCH 16/26] Avoid unnecessary updates of /status key (#2782) When we don't have permanent logical slots Patroni was updating the `/status` on every heart-beat loop even when LSN on the primary isn't moving forward. The issue was introduced in the #2485 --- patroni/dcs/__init__.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index f07ce6cd..29a4d766 100644 --- a/patroni/dcs/__init__.py +++ b/patroni/dcs/__init__.py @@ -1033,7 +1033,9 @@ class AbstractDCS(abc.ABC): raise self._last_seen = int(time.time()) - self._last_status = {self._OPTIME: cluster.last_lsn, 'slots': cluster.slots} + self._last_status = {self._OPTIME: cluster.last_lsn} + if cluster.slots: + self._last_status['slots'] = cluster.slots self._last_failsafe = cluster.failsafe with self._cluster_thread_lock: From 2735c937fd1e75f372937f18c1c1825ceb34fca6 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Thu, 27 Jul 2023 13:39:28 +0200 Subject: [PATCH 17/26] Fix pg_rewind behaviour after pause (#2776) On Slack user reported that Patroni didn't run pg_rewind on one of the nodes after coming out of maintenance mode. Steps that were executed: 0. The initial state: node1 - primary, node2 - replica 1. `patronictl pause` 2. On node2: pg_ctl promote 3. On node1: pg_ctl stop 4. Patroni on node1 notice that Postgres isn't running and removes the leader lock 5. Patroni on node2 notice that Postgres is running as a primary and takes the leader lock. 6. `patronictl resume`. After that node1 started saying in logs: `Waiting for checkpoint on node2 before rewind`. Repeating this steps may not necessarily reproduce the problem, because presumably pg_rewind failed earlier on node1. Such situation was possible because promote wasn't executed by Patroni and therefore `Rewind._state` wasn't explicitly reset and the code that ensures that CHECKPOINT after promote was run wasn't triggered. As a mitigation following changes have been made: 1. retrigger pg_rewind checks after coming out of maintenance mode 2. run ensure CHECKPOINT after promote checks using `Rewind._state != REWIND_STATUS.CHECKPOINT` condition. It allowed to remove useless hook from `Postgresql.promote()`. --- patroni/ha.py | 12 ++++++------ patroni/postgresql/__init__.py | 6 ++---- patroni/postgresql/rewind.py | 2 +- 3 files changed, 9 insertions(+), 11 deletions(-) diff --git a/patroni/ha.py b/patroni/ha.py index 59c57565..95ca6e05 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -497,6 +497,7 @@ class Ha(object): role = 'replica' if self.has_lock() and not self.is_standby_cluster(): + self._rewind.reset_state() # we want to later trigger CHECKPOINT after promote msg = "starting as readonly because i had the session lock" node_to_follow = None else: @@ -769,18 +770,14 @@ 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'): - def on_success(): - self._rewind.reset_state() - logger.info("cleared rewind state after becoming the leader") - def before_promote(): self.notify_citus_coordinator('before_promote') with self._async_response: self._async_response.reset() + self._async_executor.try_run_async('promote', self.state_handler.promote, - args=(self.dcs.loop_wait, self._async_response, - before_promote, on_success)) + args=(self.dcs.loop_wait, self._async_response, before_promote)) return promote_message def fetch_node_status(self, member: Member) -> _MemberStatus: @@ -1614,6 +1611,9 @@ class Ha(object): else: if self._was_paused: self.state_handler.schedule_sanity_checks_after_pause() + # during pause people could manually do something with Postgres, therefore we want + # to double check rewind conditions on replicas and maybe run CHECKPOINT on the primary + self._rewind.reset_state() self._was_paused = False if not self.cluster.has_member(self.state_handler.name): diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py index 632cbee7..04fab0a2 100644 --- a/patroni/postgresql/__init__.py +++ b/patroni/postgresql/__init__.py @@ -1125,8 +1125,8 @@ class Postgresql(object): except Exception as e: logger.error('Exception when calling `%s`: %r', cmd, e) - def promote(self, wait_seconds: int, task: CriticalTask, before_promote: Optional[Callable[..., Any]] = None, - on_success: Optional[Callable[..., Any]] = None) -> Optional[bool]: + def promote(self, wait_seconds: int, task: CriticalTask, + before_promote: Optional[Callable[..., Any]] = None) -> Optional[bool]: if self.role in ('promoted', 'master', 'primary'): return True @@ -1152,8 +1152,6 @@ class Postgresql(object): ret = self.pg_ctl('promote', '-W') if ret: self.set_role('promoted') - if on_success is not None: - on_success() self.call_nowait(CallbackAction.ON_ROLE_CHANGE) ret = self._wait_promote(wait_seconds) return ret diff --git a/patroni/postgresql/rewind.py b/patroni/postgresql/rewind.py index ff6fc751..270d629c 100644 --- a/patroni/postgresql/rewind.py +++ b/patroni/postgresql/rewind.py @@ -280,7 +280,7 @@ class Rewind(object): """After promote issue a CHECKPOINT from a new thread and asynchronously check the result. In case if CHECKPOINT failed, just check that timeline in pg_control was updated.""" - if self._state == REWIND_STATUS.INITIAL and self._postgresql.is_leader(): + if self._state != REWIND_STATUS.CHECKPOINT and self._postgresql.is_leader(): with self._checkpoint_task_lock: if self._checkpoint_task: with self._checkpoint_task: From 7e89583ec7348bf0be177c9a76b51be4ea82bc12 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Mon, 31 Jul 2023 09:08:46 +0200 Subject: [PATCH 18/26] Please new flake8 (#2789) it stopped liking lack of space character between `,` and `\` ```python foo,\ bar ``` --- features/environment.py | 2 +- features/steps/basic_replication.py | 8 ++++---- features/steps/citus.py | 2 +- features/steps/patroni_api.py | 4 ++-- patroni/dcs/consul.py | 2 +- patroni/dcs/etcd.py | 2 +- patroni/dcs/etcd3.py | 2 +- patroni/dcs/kubernetes.py | 4 ++-- tests/test_api.py | 4 ++-- tests/test_bootstrap.py | 18 +++++++++--------- tests/test_citus.py | 4 ++-- tests/test_etcd.py | 4 ++-- tests/test_etcd3.py | 7 ++++--- tests/test_ha.py | 10 +++++----- tests/test_kubernetes.py | 20 ++++++++++---------- tests/test_log.py | 2 +- tests/test_postgresql.py | 6 +++--- tests/test_raft.py | 2 +- tests/test_rewind.py | 6 +++--- tests/test_slots.py | 14 +++++++------- tests/test_zookeeper.py | 2 +- 21 files changed, 63 insertions(+), 62 deletions(-) diff --git a/features/environment.py b/features/environment.py index ec503bb5..e3c21252 100644 --- a/features/environment.py +++ b/features/environment.py @@ -59,7 +59,7 @@ class AbstractController(abc.ABC): break time.sleep(1) else: - assert False,\ + assert False, \ "{0} instance is not available for queries after {1} seconds".format(self._name, max_wait_limit) def stop(self, kill=False, timeout=15, _=False): diff --git a/features/steps/basic_replication.py b/features/steps/basic_replication.py index 6499a03f..6718cb01 100644 --- a/features/steps/basic_replication.py +++ b/features/steps/basic_replication.py @@ -21,7 +21,7 @@ def start_duplicate_patroni(context, name, port): context.pctl.start('dup-' + name, custom_config=config) assert False, "Process was expected to fail" except AssertionError as e: - assert 'is not running after being started' in str(e),\ + assert 'is not running after being started' in str(e), \ "No error was raised by duplicate start of {0} ".format(name) @@ -88,14 +88,14 @@ def table_is_present_on(context, table_name, pg_name, max_replication_delay): break sleep(1) else: - assert False,\ + assert False, \ "Table {0} is not present on {1} after {2} seconds".format(table_name, pg_name, max_replication_delay) @then('{pg_name:w} role is the {pg_role:w} after {max_promotion_timeout:d} seconds') def check_role(context, pg_name, pg_role, max_promotion_timeout): max_promotion_timeout *= context.timeout_multiplier - assert context.pctl.check_role_has_changed_to(pg_name, pg_role, timeout=int(max_promotion_timeout)),\ + assert context.pctl.check_role_has_changed_to(pg_name, pg_role, timeout=int(max_promotion_timeout)), \ "{0} role didn't change to {1} after {2} seconds".format(pg_name, pg_role, max_promotion_timeout) @@ -111,5 +111,5 @@ def replication_works(context, primary, replica, time_limit): @then('there is a "{message}" {level:w} in the {node} patroni log') def check_patroni_log(context, message, level, node): messsages_of_level = context.pctl.read_patroni_log(node, level) - assert any(message in line for line in messsages_of_level),\ + assert any(message in line for line in messsages_of_level), \ "There was no {0} {1} in the {2} patroni log".format(message, level, node) diff --git a/features/steps/citus.py b/features/steps/citus.py index 1af70a30..644219c7 100644 --- a/features/steps/citus.py +++ b/features/steps/citus.py @@ -125,5 +125,5 @@ def check_transaction(context, name): @step("a transaction finishes in {timeout:d} seconds") def check_transaction_timeout(context, timeout): - assert (datetime.now(tzutc) - context.xact_start).seconds > timeout,\ + assert (datetime.now(tzutc) - context.xact_start).seconds > timeout, \ "a transaction finished earlier than in {0} seconds".format(timeout) diff --git a/features/steps/patroni_api.py b/features/steps/patroni_api.py index a745b192..2c76d32d 100644 --- a/features/steps/patroni_api.py +++ b/features/steps/patroni_api.py @@ -98,7 +98,7 @@ def do_run(context, cmd): @then('I receive a response {component:w} {data}') def check_response(context, component, data): if component == 'code': - assert context.status_code == int(data),\ + assert context.status_code == int(data), \ "status code {0} != {1}, response: {2}".format(context.status_code, data, context.response) elif component == 'returncode': assert context.status_code == int(data), "return code {0} != {1}, {2}".format(context.status_code, @@ -158,7 +158,7 @@ def check_http_response(context, url, value, timeout, negate=False): break time.sleep(1) else: - assert False,\ + assert False, \ "Value {0} is {1} present in response after {2} seconds".format(value, "not" if not negate else "", timeout) diff --git a/patroni/dcs/consul.py b/patroni/dcs/consul.py index 73755a41..7b310d15 100644 --- a/patroni/dcs/consul.py +++ b/patroni/dcs/consul.py @@ -15,7 +15,7 @@ from urllib3.exceptions import HTTPError from urllib.parse import urlencode, urlparse, quote from typing import Any, Callable, Dict, List, Mapping, NamedTuple, Optional, Union, Tuple, TYPE_CHECKING -from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, SyncState,\ +from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, SyncState, \ TimelineHistory, ReturnFalseException, catch_return_false_exception, citus_group_re from ..exceptions import DCSError from ..utils import deep_compare, parse_bool, Retry, RetryFailedError, split_host_port, uri, USER_AGENT diff --git a/patroni/dcs/etcd.py b/patroni/dcs/etcd.py index 94e546ac..447c7dab 100644 --- a/patroni/dcs/etcd.py +++ b/patroni/dcs/etcd.py @@ -21,7 +21,7 @@ from urllib.parse import urlparse from urllib3 import Timeout from urllib3.exceptions import HTTPError, ReadTimeoutError, ProtocolError -from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, SyncState,\ +from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, SyncState, \ TimelineHistory, ReturnFalseException, catch_return_false_exception, citus_group_re from ..exceptions import DCSError from ..request import get as requests_get diff --git a/patroni/dcs/etcd3.py b/patroni/dcs/etcd3.py index 32d00359..0bfc4fae 100644 --- a/patroni/dcs/etcd3.py +++ b/patroni/dcs/etcd3.py @@ -15,7 +15,7 @@ from urllib3.exceptions import ReadTimeoutError, ProtocolError from threading import Condition, Lock, Thread from typing import Any, Callable, Collection, Dict, Iterator, List, Optional, Tuple, Type, TYPE_CHECKING, Union -from . import ClusterConfig, Cluster, Failover, Leader, Member, SyncState,\ +from . import ClusterConfig, Cluster, Failover, Leader, Member, SyncState, \ TimelineHistory, catch_return_false_exception, citus_group_re from .etcd import AbstractEtcdClientWithFailover, AbstractEtcd, catch_etcd_errors, DnsCachingResolver, Retry from ..exceptions import DCSError, PatroniException diff --git a/patroni/dcs/kubernetes.py b/patroni/dcs/kubernetes.py index bda4bdc8..96097338 100644 --- a/patroni/dcs/kubernetes.py +++ b/patroni/dcs/kubernetes.py @@ -19,10 +19,10 @@ from urllib3.exceptions import HTTPError from threading import Condition, Lock, Thread from typing import Any, Callable, Collection, Dict, List, Optional, Tuple, Type, Union, TYPE_CHECKING -from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, SyncState,\ +from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, SyncState, \ TimelineHistory, CITUS_COORDINATOR_GROUP_ID, citus_group_re from ..exceptions import DCSError -from ..utils import deep_compare, iter_response_objects, keepalive_socket_options,\ +from ..utils import deep_compare, iter_response_objects, keepalive_socket_options, \ Retry, RetryFailedError, tzutc, uri, USER_AGENT if TYPE_CHECKING: # pragma: no cover from ..config import Config diff --git a/tests/test_api.py b/tests/test_api.py index f9d0b1c1..eb19ebca 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -226,7 +226,7 @@ class TestRestApiHandler(unittest.TestCase): self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'GET /primary')) with patch.object(RestApiServer, 'query', Mock(return_value=[('', 1, '', '', '', '', False, None, None, '')])): self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'GET /patroni')) - with patch.object(GlobalConfig, 'is_standby_cluster', Mock(return_value=True)),\ + with patch.object(GlobalConfig, 'is_standby_cluster', Mock(return_value=True)), \ patch.object(GlobalConfig, 'is_paused', Mock(return_value=True)): MockRestApiServer(RestApiHandler, 'GET /standby_leader') @@ -559,7 +559,7 @@ class TestRestApiHandler(unittest.TestCase): request = post + '103\n\n{"leader": "postgresql1", "member": "postgresql2",' +\ ' "scheduled_at": "6016-02-15T18:13:30.568224+01:00"}' MockRestApiServer(RestApiHandler, request) - with patch.object(GlobalConfig, 'is_paused', PropertyMock(return_value=True)),\ + with patch.object(GlobalConfig, 'is_paused', PropertyMock(return_value=True)), \ patch.object(MockPatroni, 'dcs') as d: d.manual_failover.return_value = False MockRestApiServer(RestApiHandler, request) diff --git a/tests/test_bootstrap.py b/tests/test_bootstrap.py index 4c7c0fcc..9f98fecb 100644 --- a/tests/test_bootstrap.py +++ b/tests/test_bootstrap.py @@ -155,9 +155,9 @@ class TestBootstrap(BaseTestPostgresql): config = {'users': {'replicator': {'password': 'rep-pass', 'options': ['replication']}}} - with patch.object(Postgresql, 'is_running', Mock(return_value=False)),\ - patch.object(Postgresql, 'get_major_version', Mock(return_value=140000)),\ - patch('multiprocessing.Process', Mock(side_effect=Exception)),\ + with patch.object(Postgresql, 'is_running', Mock(return_value=False)), \ + patch.object(Postgresql, 'get_major_version', Mock(return_value=140000)), \ + patch('multiprocessing.Process', Mock(side_effect=Exception)), \ patch('multiprocessing.get_context', Mock(side_effect=Exception), create=True): self.assertRaises(Exception, self.b.bootstrap, config) with open(os.path.join(self.p.data_dir, 'pg_hba.conf')) as f: @@ -185,12 +185,12 @@ class TestBootstrap(BaseTestPostgresql): self.assertFalse(self.b.bootstrap(config)) mock_cancellable_subprocess_call.return_value = 0 - with patch('multiprocessing.Process', Mock(side_effect=Exception("42"))),\ - patch('multiprocessing.get_context', Mock(side_effect=Exception("42")), create=True),\ - patch('os.path.isfile', Mock(return_value=True)),\ - patch('os.unlink', Mock()),\ - patch.object(ConfigHandler, 'save_configuration_files', Mock()),\ - patch.object(ConfigHandler, 'restore_configuration_files', Mock()),\ + with patch('multiprocessing.Process', Mock(side_effect=Exception("42"))), \ + patch('multiprocessing.get_context', Mock(side_effect=Exception("42")), create=True), \ + patch('os.path.isfile', Mock(return_value=True)), \ + patch('os.unlink', Mock()), \ + patch.object(ConfigHandler, 'save_configuration_files', Mock()), \ + patch.object(ConfigHandler, 'restore_configuration_files', Mock()), \ patch.object(ConfigHandler, 'write_recovery_conf', Mock()): with self.assertRaises(Exception) as e: self.b.bootstrap(config) diff --git a/tests/test_citus.py b/tests/test_citus.py index 40d4df8a..7c2d63bb 100644 --- a/tests/test_citus.py +++ b/tests/test_citus.py @@ -52,7 +52,7 @@ class TestCitus(BaseTestPostgresql): 'leader': 'leader', 'timeout': 30, 'cooldown': 10}) def test_add_task(self): - with patch('patroni.postgresql.citus.logger.error') as mock_logger,\ + with patch('patroni.postgresql.citus.logger.error') as mock_logger, \ patch('patroni.postgresql.citus.urlparse', Mock(side_effect=Exception)): self.c.add_task('', 1, None) mock_logger.assert_called_once() @@ -107,7 +107,7 @@ class TestCitus(BaseTestPostgresql): self.c.process_tasks() self.c.add_task('after_promote', 0, 'postgres://host3:5432/postgres') - with patch('patroni.postgresql.citus.logger.error') as mock_logger,\ + with patch('patroni.postgresql.citus.logger.error') as mock_logger, \ patch.object(CitusHandler, 'query', Mock(side_effect=Exception)): self.c.process_tasks() mock_logger.assert_called_once() diff --git a/tests/test_etcd.py b/tests/test_etcd.py index b5add201..f9e07144 100644 --- a/tests/test_etcd.py +++ b/tests/test_etcd.py @@ -172,12 +172,12 @@ class TestClient(unittest.TestCase): self.assertRaises(etcd.EtcdWatchTimedOut, self.client.api_execute, '/timeout', 'POST', params={'wait': 'true'}) self.assertRaises(etcd.EtcdWatchTimedOut, self.client.api_execute, '/timeout', 'POST', params={'wait': 'true'}) - with patch.object(EtcdClient, '_calculate_timeouts', Mock(side_effect=[(1, 1, 0), (1, 1, 0), (0, 1, 0)])),\ + with patch.object(EtcdClient, '_calculate_timeouts', Mock(side_effect=[(1, 1, 0), (1, 1, 0), (0, 1, 0)])), \ patch.object(EtcdClient, '_load_machines_cache', Mock(side_effect=Exception)): self.client.http.request = Mock(side_effect=socket.error) self.assertRaises(etcd.EtcdException, rtry, self.client.api_execute, '/', 'GET', params={'retry': rtry}) - with patch.object(EtcdClient, '_calculate_timeouts', Mock(side_effect=[(1, 1, 0), (1, 1, 0), (0, 1, 0)])),\ + with patch.object(EtcdClient, '_calculate_timeouts', Mock(side_effect=[(1, 1, 0), (1, 1, 0), (0, 1, 0)])), \ patch.object(EtcdClient, '_load_machines_cache', Mock(return_value=True)): self.assertRaises(etcd.EtcdException, rtry, self.client.api_execute, '/', 'GET', params={'retry': rtry}) diff --git a/tests/test_etcd3.py b/tests/test_etcd3.py index a737f199..ace59a62 100644 --- a/tests/test_etcd3.py +++ b/tests/test_etcd3.py @@ -5,8 +5,9 @@ import urllib3 from mock import Mock, PropertyMock, patch from patroni.dcs.etcd import DnsCachingResolver -from patroni.dcs.etcd3 import PatroniEtcd3Client, Cluster, Etcd3Client, Etcd3Error, Etcd3ClientError, RetryFailedError,\ - InvalidAuthToken, Unavailable, Unknown, UnsupportedEtcdVersion, UserEmpty, AuthFailed, base64_encode, Etcd3 +from patroni.dcs.etcd3 import PatroniEtcd3Client, Cluster, Etcd3, Etcd3Client, \ + Etcd3Error, Etcd3ClientError, RetryFailedError, InvalidAuthToken, Unavailable, \ + Unknown, UnsupportedEtcdVersion, UserEmpty, AuthFailed, base64_encode from threading import Thread from . import SleepException, MockResponse @@ -241,7 +242,7 @@ class TestEtcd3(BaseTestEtcd3): self.etcd3.update_leader(leader, '123', failsafe={'foo': 'bar'}) self.etcd3._last_lease_refresh = 0 self.etcd3.update_leader(leader, '124') - with patch.object(PatroniEtcd3Client, 'lease_keepalive', Mock(return_value=True)),\ + with patch.object(PatroniEtcd3Client, 'lease_keepalive', Mock(return_value=True)), \ patch('time.time', Mock(side_effect=[0, 100, 200, 300])): self.assertRaises(Etcd3Error, self.etcd3.update_leader, leader, '126') self.etcd3._lease = leader.session diff --git a/tests/test_ha.py b/tests/test_ha.py index 7cdb00e1..1f5f903d 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -307,7 +307,7 @@ class TestHa(PostgresInit): self.p.is_running = false self.p.controldata = lambda: {'Database cluster state': 'in production', 'Database system identifier': SYSID} self.assertEqual(self.ha.run_cycle(), 'doing crash recovery in a single user mode') - with patch('patroni.async_executor.AsyncExecutor.busy', PropertyMock(return_value=True)),\ + with patch('patroni.async_executor.AsyncExecutor.busy', PropertyMock(return_value=True)), \ patch.object(Ha, 'check_timeline', Mock(return_value=False)): self.ha._async_executor.schedule('doing crash recovery in a single user mode') self.ha.state_handler.cancellable._process = Mock() @@ -340,7 +340,7 @@ class TestHa(PostgresInit): self.ha._rewind.check_leader_is_not_in_recovery = true with patch.object(Rewind, 'rewind_or_reinitialize_needed_and_possible', Mock(return_value=True)): self.assertEqual(self.ha.run_cycle(), 'running pg_rewind from leader') - with patch.object(Rewind, 'rewind_or_reinitialize_needed_and_possible', Mock(return_value=False)),\ + with patch.object(Rewind, 'rewind_or_reinitialize_needed_and_possible', Mock(return_value=False)), \ patch.object(Ha, 'is_synchronous_mode', Mock(return_value=True)): self.p.follow = true self.assertEqual(self.ha.run_cycle(), 'starting as a secondary') @@ -608,7 +608,7 @@ class TestHa(PostgresInit): self.e.initialize = true self.ha.bootstrap() self.p.is_leader = true - with patch.object(Watchdog, 'activate', Mock(return_value=False)),\ + with patch.object(Watchdog, 'activate', Mock(return_value=False)), \ patch('patroni.ha.logger.error') as mock_logger: self.assertEqual(self.ha.post_bootstrap(), 'running post_bootstrap') self.assertRaises(PatroniFatalException, self.ha.post_bootstrap) @@ -669,9 +669,9 @@ class TestHa(PostgresInit): self.ha.update_lock = false self.p.set_role('primary') - with patch('patroni.async_executor.CriticalTask.cancel', Mock(return_value=False)),\ + with patch('patroni.async_executor.CriticalTask.cancel', Mock(return_value=False)), \ patch('patroni.async_executor.CriticalTask.result', - PropertyMock(return_value=PostmasterProcess(os.getpid())), create=True),\ + PropertyMock(return_value=PostmasterProcess(os.getpid())), create=True), \ patch('patroni.postgresql.Postgresql.terminate_starting_postmaster') as mock_terminate: self.assertEqual(self.ha.run_cycle(), 'lost leader lock during restart') mock_terminate.assert_called() diff --git a/tests/test_kubernetes.py b/tests/test_kubernetes.py index f79539a7..662e22a8 100644 --- a/tests/test_kubernetes.py +++ b/tests/test_kubernetes.py @@ -8,8 +8,8 @@ import unittest import urllib3 from mock import Mock, PropertyMock, mock_open, patch -from patroni.dcs.kubernetes import Cluster, k8s_client, k8s_config, K8sConfig, K8sConnectionFailed,\ - K8sException, K8sObject, Kubernetes, KubernetesError, KubernetesRetriableException,\ +from patroni.dcs.kubernetes import Cluster, k8s_client, k8s_config, K8sConfig, K8sConnectionFailed, \ + K8sException, K8sObject, Kubernetes, KubernetesError, KubernetesRetriableException, \ Retry, RetryFailedError, SERVICE_HOST_ENV_NAME, SERVICE_PORT_ENV_NAME from threading import Thread from . import MockResponse, SleepException @@ -86,8 +86,8 @@ class TestK8sConfig(unittest.TestCase): with patch('os.environ', env): self.assertRaises(k8s_config.ConfigException, k8s_config.load_incluster_config) - with patch('os.environ', {SERVICE_HOST_ENV_NAME: 'a', SERVICE_PORT_ENV_NAME: '1'}),\ - patch('os.path.isfile', Mock(side_effect=[False, True, True, False, True, True, True, True])),\ + with patch('os.environ', {SERVICE_HOST_ENV_NAME: 'a', SERVICE_PORT_ENV_NAME: '1'}), \ + patch('os.path.isfile', Mock(side_effect=[False, True, True, False, True, True, True, True])), \ patch('builtins.open', Mock(side_effect=[ mock_open()(), mock_open(read_data='a')(), mock_open(read_data='a')(), mock_open()(), mock_open(read_data='a')(), mock_open(read_data='a')()])): @@ -98,8 +98,8 @@ class TestK8sConfig(unittest.TestCase): self.assertEqual(k8s_config.headers.get('authorization'), 'Bearer a') def test_refresh_token(self): - with patch('os.environ', {SERVICE_HOST_ENV_NAME: 'a', SERVICE_PORT_ENV_NAME: '1'}),\ - patch('os.path.isfile', Mock(side_effect=[True, True, False, True, True, True])),\ + with patch('os.environ', {SERVICE_HOST_ENV_NAME: 'a', SERVICE_PORT_ENV_NAME: '1'}), \ + patch('os.path.isfile', Mock(side_effect=[True, True, False, True, True, True])), \ patch('builtins.open', Mock(side_effect=[ mock_open(read_data='cert')(), mock_open(read_data='a')(), mock_open()(), mock_open(read_data='b')(), mock_open(read_data='c')()])): @@ -138,10 +138,10 @@ class TestK8sConfig(unittest.TestCase): config["users"][0]["user"]["client-key-data"] = base64.b64encode(b'foobar').decode('utf-8') config["clusters"][0]["cluster"]["certificate-authority-data"] = base64.b64encode(b'foobar').decode('utf-8') - with patch('builtins.open', mock_open(read_data=json.dumps(config))),\ - patch('os.write', Mock()), patch('os.close', Mock()),\ - patch('os.remove') as mock_remove,\ - patch('atexit.register') as mock_atexit,\ + with patch('builtins.open', mock_open(read_data=json.dumps(config))), \ + patch('os.write', Mock()), patch('os.close', Mock()), \ + patch('os.remove') as mock_remove, \ + patch('atexit.register') as mock_atexit, \ patch('tempfile.mkstemp') as mock_mkstemp: mock_mkstemp.side_effect = [(3, '1.tmp'), (4, '2.tmp')] k8s_config.load_kube_config() diff --git a/tests/test_log.py b/tests/test_log.py index 1a383908..ebdb4945 100644 --- a/tests/test_log.py +++ b/tests/test_log.py @@ -43,7 +43,7 @@ class TestPatroniLogger(unittest.TestCase): _LOG.exception('test') logger.start() - with patch.object(logging.Handler, 'format', Mock(side_effect=Exception)),\ + with patch.object(logging.Handler, 'format', Mock(side_effect=Exception)), \ patch('_pytest.logging.LogCaptureHandler.emit', Mock()): logging.error('test') diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index c759cde0..036e225b 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -333,7 +333,7 @@ class TestPostgresql(BaseTestPostgresql): mock_read_auto = mock_open(read_data=read_data) mock_read_auto.return_value.__iter__ = lambda o: iter(o.readline, '') - with patch('builtins.open', Mock(side_effect=[mock_open()(), mock_read_auto(), IOError])),\ + with patch('builtins.open', Mock(side_effect=[mock_open()(), mock_read_auto(), IOError])), \ patch('os.chmod', Mock()): self.p.config.write_postgresql_conf() @@ -496,8 +496,8 @@ class TestPostgresql(BaseTestPostgresql): self.p.remove_data_directory() with patch('os.path.isfile', Mock(return_value=True)): self.p.remove_data_directory() - with patch('os.path.islink', Mock(side_effect=[False, False, True, True])),\ - patch('os.listdir', Mock(return_value=['12345'])),\ + with patch('os.path.islink', Mock(side_effect=[False, False, True, True])), \ + patch('os.listdir', Mock(return_value=['12345'])), \ patch('os.path.realpath', Mock(side_effect=['../foo', '../foo_tsp'])): self.p.remove_data_directory() diff --git a/tests/test_raft.py b/tests/test_raft.py index ab68c2e6..1fe8733c 100644 --- a/tests/test_raft.py +++ b/tests/test_raft.py @@ -4,7 +4,7 @@ import tempfile import time from mock import Mock, PropertyMock, patch -from patroni.dcs.raft import Cluster, DynMemberSyncObj, KVStoreTTL,\ +from patroni.dcs.raft import Cluster, DynMemberSyncObj, KVStoreTTL, \ Raft, RaftError, SyncObjUtility, TCPTransport, _TCPTransport from pysyncobj import SyncObjConf, FAIL_REASON diff --git a/tests/test_rewind.py b/tests/test_rewind.py index a4fafcff..22181266 100644 --- a/tests/test_rewind.py +++ b/tests/test_rewind.py @@ -65,14 +65,14 @@ class TestRewind(BaseTestPostgresql): def test_pg_rewind(self): r = {'user': '', 'host': '', 'port': '', 'database': '', 'password': ''} - with patch.object(Postgresql, 'major_version', PropertyMock(return_value=150000)),\ + with patch.object(Postgresql, 'major_version', PropertyMock(return_value=150000)), \ patch.object(CancellableSubprocess, 'call', Mock(return_value=None)): with patch('subprocess.check_output', Mock(return_value=b'boo')): self.assertFalse(self.r.pg_rewind(r)) with patch('subprocess.check_output', Mock(side_effect=Exception)): self.assertFalse(self.r.pg_rewind(r)) - with patch.object(Postgresql, 'major_version', PropertyMock(return_value=120000)),\ + with patch.object(Postgresql, 'major_version', PropertyMock(return_value=120000)), \ patch('subprocess.check_output', Mock(return_value=b'foo %f %p %r %% % %')): with patch.object(CancellableSubprocess, 'call', mock_cancellable_call): self.assertFalse(self.r.pg_rewind(r)) @@ -91,7 +91,7 @@ class TestRewind(BaseTestPostgresql): 'Latest checkpoint location': '0/'})): self.r.rewind_or_reinitialize_needed_and_possible(self.leader) - with patch.object(Postgresql, 'is_running', Mock(return_value=True)),\ + with patch.object(Postgresql, 'is_running', Mock(return_value=True)), \ patch.object(MockCursor, 'fetchone', Mock(side_effect=[(0, 0, 1, 1, 0, 0, 0, 0, 0, None, None, None), Exception])): self.r.rewind_or_reinitialize_needed_and_possible(self.leader) diff --git a/tests/test_slots.py b/tests/test_slots.py index 3f21998f..6d3f17d3 100644 --- a/tests/test_slots.py +++ b/tests/test_slots.py @@ -43,12 +43,12 @@ class TestSlotsHandler(BaseTestPostgresql): with mock.patch('patroni.postgresql.Postgresql._query', Mock(side_effect=psycopg.OperationalError)): self.s.sync_replication_slots(cluster, False) self.p.set_role('standby_leader') - with patch.object(SlotsHandler, 'drop_replication_slot', Mock(return_value=(True, False))),\ + with patch.object(SlotsHandler, 'drop_replication_slot', Mock(return_value=(True, False))), \ patch('patroni.postgresql.slots.logger.debug') as mock_debug: self.s.sync_replication_slots(cluster, False) mock_debug.assert_called_once() self.p.set_role('replica') - with patch.object(Postgresql, 'is_leader', Mock(return_value=False)),\ + with patch.object(Postgresql, 'is_leader', Mock(return_value=False)), \ patch.object(SlotsHandler, 'drop_replication_slot') as mock_drop: self.s.sync_replication_slots(cluster, False, paused=True) mock_drop.assert_not_called() @@ -96,8 +96,8 @@ class TestSlotsHandler(BaseTestPostgresql): with patch.object(SlotsHandler, 'check_logical_slots_readiness', Mock(return_value=False)): self.assertEqual(self.s.sync_replication_slots(self.cluster, False), []) self.s._schedule_load_slots = False - with patch.object(MockCursor, 'execute', Mock(side_effect=psycopg.OperationalError)),\ - patch.object(SlotsAdvanceThread, 'schedule', Mock(return_value=(True, ['ls']))),\ + with patch.object(MockCursor, 'execute', Mock(side_effect=psycopg.OperationalError)), \ + patch.object(SlotsAdvanceThread, 'schedule', Mock(return_value=(True, ['ls']))), \ patch.object(psycopg.OperationalError, 'diag') as mock_diag: type(mock_diag).sqlstate = PropertyMock(return_value='58P01') self.assertEqual(self.s.sync_replication_slots(self.cluster, False), ['ls']) @@ -119,10 +119,10 @@ class TestSlotsHandler(BaseTestPostgresql): @patch.object(Postgresql, 'is_leader', Mock(return_value=False)) def test_check_logical_slots_readiness(self): self.s.copy_logical_slots(self.cluster, ['ls']) - with patch.object(MockCursor, '__iter__', Mock(return_value=iter([('postgresql0', None)]))),\ + with patch.object(MockCursor, '__iter__', Mock(return_value=iter([('postgresql0', None)]))), \ patch.object(MockCursor, 'fetchone', Mock(side_effect=Exception)): self.assertFalse(self.s.check_logical_slots_readiness(self.cluster, None)) - with patch.object(MockCursor, '__iter__', Mock(return_value=iter([('postgresql0', None)]))),\ + with patch.object(MockCursor, '__iter__', Mock(return_value=iter([('postgresql0', None)]))), \ patch.object(MockCursor, 'fetchone', Mock(return_value=(False,))): self.assertFalse(self.s.check_logical_slots_readiness(self.cluster, None)) with patch.object(MockCursor, '__iter__', Mock(return_value=iter([('ls', 100)]))): @@ -144,7 +144,7 @@ class TestSlotsHandler(BaseTestPostgresql): self.assertRaises(OSError, fsync_dir, 'foo') def test_slots_advance_thread(self): - with patch.object(MockCursor, 'execute', Mock(side_effect=psycopg.OperationalError)),\ + with patch.object(MockCursor, 'execute', Mock(side_effect=psycopg.OperationalError)), \ patch.object(psycopg.OperationalError, 'diag') as mock_diag: type(mock_diag).sqlstate = PropertyMock(return_value='58P01') self.s.schedule_advance_slots({'foo': {'bar': 100}}) diff --git a/tests/test_zookeeper.py b/tests/test_zookeeper.py index baf9ad10..c72fefe9 100644 --- a/tests/test_zookeeper.py +++ b/tests/test_zookeeper.py @@ -7,7 +7,7 @@ from kazoo.handlers.threading import SequentialThreadingHandler from kazoo.protocol.states import KeeperState, ZnodeStat from kazoo.retry import RetryFailedError from mock import Mock, PropertyMock, patch -from patroni.dcs.zookeeper import Cluster, Leader, PatroniKazooClient,\ +from patroni.dcs.zookeeper import Cluster, Leader, PatroniKazooClient, \ PatroniSequentialThreadingHandler, ZooKeeper, ZooKeeperError From 8f3ed0088643543123fa1373b3748ae2170a1814 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Mon, 31 Jul 2023 10:16:19 +0200 Subject: [PATCH 19/26] Invalidate cache if txn failed due to revision mismatch (#2783) It was reported in #2779 that the primary was constantly logging messages like `Synchronous replication key updated by someone else`. It happened after Patroni was stuck due to resource starvation. Key updates are performed using create_revision/mod_revision field, which value is taken from the internal cached. Hence, it is a clear symptom of stale cache. Similar issues in K8s implementation were addressed by invalidating the cache and restarting watcher connections every time when update failed due to resource_version mismatch, so we do the same for Etcd3. --- patroni/dcs/etcd3.py | 10 ++++++++++ tests/test_etcd3.py | 8 +++++++- 2 files changed, 17 insertions(+), 1 deletion(-) diff --git a/patroni/dcs/etcd3.py b/patroni/dcs/etcd3.py index 0bfc4fae..e28da844 100644 --- a/patroni/dcs/etcd3.py +++ b/patroni/dcs/etcd3.py @@ -630,6 +630,16 @@ class PatroniEtcd3Client(Etcd3Client): return ret + def txn(self, compare: Dict[str, Any], success: Dict[str, Any], + failure: Optional[Dict[str, Any]] = None, retry: Optional[Retry] = None) -> Dict[str, Any]: + ret = super(PatroniEtcd3Client, self).txn(compare, success, failure, retry) + # Here we abuse the fact that the `failure` is only set in the call from update_leader(). + # In all other cases the txn() call failure may be an indicator of a stale cache, + # and therefore we want to restart watcher. + if not failure and not ret: + self._restart_watcher() + return ret + class Etcd3(AbstractEtcd): diff --git a/tests/test_etcd3.py b/tests/test_etcd3.py index ace59a62..2e3ed59d 100644 --- a/tests/test_etcd3.py +++ b/tests/test_etcd3.py @@ -127,10 +127,16 @@ class TestPatroniEtcd3Client(BaseTestEtcd3): request = {'key': base64_encode('/patroni/test/leader')} mock_urlopen.return_value = MockResponse() mock_urlopen.return_value.content = '{"succeeded":true,"header":{"revision":"1"}}' - self.client.call_rpc('/kv/txn', {'success': [{'request_delete_range': request}]}) self.client.call_rpc('/kv/put', request) self.client.call_rpc('/kv/deleterange', request) + @patch.object(urllib3.PoolManager, 'urlopen') + def test_txn(self, mock_urlopen): + mock_urlopen.return_value = MockResponse() + mock_urlopen.return_value.content = '{"header":{"revision":"1"}}' + self.client.txn({'target': 'MOD', 'mod_revision': '1'}, + {'request_delete_range': {'key': base64_encode('/patroni/test/leader')}}) + @patch('time.time', Mock(side_effect=[1, 10.9, 100])) def test__wait_cache(self): with self.kv_cache.condition: From 01976ec10bca5c1869f44ae710124d3447123f44 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Mon, 31 Jul 2023 11:22:18 +0200 Subject: [PATCH 20/26] Don't allow stale primary to win the leader race (#2787) Consider a following situation: 1. node1 is stressed so much that Patroni heart-beat can't run regularly and the leader lock expires. 2. node2 notice that there is no leader, gets the lock, promotes, and gets to a situation like it is described in 1. 3. Patroni on node1 finally wakes up, notice that Postgres is running as a primary, but without a leader lock and "happily" acquires the lock. That is, node1 discarded promoting of node2, and the node2 after that it will not be possible to join the node2 back to the cluster, because pg_rewind is not possible when two nodes are on the same timeline. To partially mitigate the problem we introduce an additional timeline check. If postgres is running as primary Patroni will consider it as a perfect candidate only if timeline isn't behind the last known cluster timeline recorder in the `/history` key. If postgres timeline is behind the cluster timeline postgres will be demoted to read-only. Further behavior would depend on `maximum_lag_on_failover` and `check_timeline` settings. Since the `/history` key isn't updated instantly after promotion, there is still a short period of time when the issue could happen, but it seems that it is close to impossible to make it more reliable. Close https://github.com/zalando/patroni/issues/2779 --- patroni/ha.py | 19 ++++++++++++++++--- tests/test_ha.py | 6 ++++++ 2 files changed, 22 insertions(+), 3 deletions(-) diff --git a/patroni/ha.py b/patroni/ha.py index 95ca6e05..2445aa02 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -996,9 +996,22 @@ class Ha(object): return ret if self.state_handler.is_leader(): - # in pause leader is the healthiest only when no initialize or sysid matches with initialize! - return not self.is_paused() or not self.cluster.initialize\ - or self.state_handler.sysid == self.cluster.initialize + if self.is_paused(): + # in pause leader is the healthiest only when no initialize or sysid matches with initialize! + return not self.cluster.initialize or self.state_handler.sysid == self.cluster.initialize + + # We want to protect from the following scenario: + # 1. node1 is stressed so much that heart-beat isn't running regularly and the leader lock expires. + # 2. node2 promotes, gets heavy load and the situation described in 1 repeats. + # 3. Patroni on node1 comes back, notices that Postgres is running as primary but there is + # no leader key and "happily" acquires the leader lock. + # That is, node1 discarded promotion of node2. To avoid it we want to detect timeline change. + my_timeline = self.state_handler.get_primary_timeline() + if my_timeline < self.cluster.timeline: + logger.warning('My timeline %s is behind last known cluster timeline %s', + my_timeline, self.cluster.timeline) + return False + return True if self.is_paused(): return False diff --git a/tests/test_ha.py b/tests/test_ha.py index 1f5f903d..ae6be3e2 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -375,6 +375,12 @@ class TestHa(PostgresInit): def test_acquire_lock_as_primary(self): self.assertEqual(self.ha.run_cycle(), 'acquired session lock as a leader') + def test_leader_race_stale_primary(self): + with patch.object(Postgresql, 'get_primary_timeline', Mock(return_value=1)), \ + patch('patroni.ha.logger.warning') as mock_logger: + self.assertEqual(self.ha.run_cycle(), 'demoting self because i am not the healthiest node') + self.assertEqual(mock_logger.call_args[0][0], 'My timeline %s is behind last known cluster timeline %s') + def test_promoted_by_acquiring_lock(self): self.ha.is_healthiest_node = true self.p.is_leader = false From 94bfea1a8179d121873ad17d790d968a06adbc34 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Mon, 31 Jul 2023 11:35:30 +0200 Subject: [PATCH 21/26] Do not fail validation for a value that is fine (#2791) In issue #2735 it was discussed that there should be some warning around PostgreSQL parameters that do not pass validation. This commit ensures something is logged for parameters that fail validation and therefore fall back to default values. Close #2735 Close #2740 --- patroni/config.py | 16 +++++++++++++--- patroni/postgresql/config.py | 16 ++++++++-------- patroni/validator.py | 6 +++--- tests/test_config.py | 3 ++- 4 files changed, 26 insertions(+), 15 deletions(-) diff --git a/patroni/config.py b/patroni/config.py index 66ffd891..facf16fd 100644 --- a/patroni/config.py +++ b/patroni/config.py @@ -329,9 +329,19 @@ class Config(object): @staticmethod def _process_postgresql_parameters(parameters: Dict[str, Any], is_local: bool = False) -> Dict[str, Any]: - return {name: value for name, value in (parameters or {}).items() - if name not in ConfigHandler.CMDLINE_OPTIONS - or not is_local and ConfigHandler.CMDLINE_OPTIONS[name][1](value)} + pg_params: Dict[str, Any] = {} + + for name, value in (parameters or {}).items(): + if name not in ConfigHandler.CMDLINE_OPTIONS: + pg_params[name] = value + elif not is_local: + if ConfigHandler.CMDLINE_OPTIONS[name][1](value): + pg_params[name] = value + else: + logging.warning("postgresql parameter %s=%s failed validation, defaulting to %s", + name, value, ConfigHandler.CMDLINE_OPTIONS[name][0]) + + return pg_params def _safe_copy_dynamic_configuration(self, dynamic_configuration: Dict[str, Any]) -> Dict[str, Any]: config = deepcopy(self.__DEFAULT_CONFIG) diff --git a/patroni/postgresql/config.py b/patroni/postgresql/config.py index 9461714f..3a61a31d 100644 --- a/patroni/postgresql/config.py +++ b/patroni/postgresql/config.py @@ -15,7 +15,7 @@ from ..collections import CaseInsensitiveDict, CaseInsensitiveSet from ..dcs import Leader, Member, RemoteMember, slot_name_from_member_name from ..exceptions import PatroniFatalException from ..utils import compare_values, parse_bool, parse_int, split_host_port, uri, validate_directory, is_subpath -from ..validator import IntValidator +from ..validator import IntValidator, EnumValidator if TYPE_CHECKING: # pragma: no cover from . import Postgresql @@ -258,14 +258,14 @@ def _false_validator(value: Any) -> bool: return False -def _wal_level_validator(value: Any) -> bool: - return str(value).lower() in ('hot_standby', 'replica', 'logical') - - def _bool_validator(value: Any) -> bool: return parse_bool(value) is not None +def _bool_is_true_validator(value: Any) -> bool: + return parse_bool(value) is True + + class ConfigHandler(object): # List of parameters which must be always passed to postmaster as command line options @@ -286,8 +286,8 @@ class ConfigHandler(object): 'listen_addresses': (None, _false_validator, 90100), 'port': (None, _false_validator, 90100), 'cluster_name': (None, _false_validator, 90500), - 'wal_level': ('hot_standby', _wal_level_validator, 90100), - 'hot_standby': ('on', _false_validator, 90100), + 'wal_level': ('hot_standby', EnumValidator(('hot_standby', 'replica', 'logical')), 90100), + 'hot_standby': ('on', _bool_is_true_validator, 90100), 'max_connections': (100, IntValidator(min=25), 90100), 'max_wal_senders': (10, IntValidator(min=3), 90100), 'wal_keep_segments': (8, IntValidator(min=1), 90100), @@ -297,7 +297,7 @@ class ConfigHandler(object): 'track_commit_timestamp': ('off', _bool_validator, 90500), 'max_replication_slots': (10, IntValidator(min=4), 90400), 'max_worker_processes': (8, IntValidator(min=2), 90400), - 'wal_log_hints': ('on', _false_validator, 90400) + 'wal_log_hints': ('on', _bool_is_true_validator, 90400) }) _RECOVERY_PARAMETERS = CaseInsensitiveSet(recovery_parameters.keys()) diff --git a/patroni/validator.py b/patroni/validator.py index b68f9654..a8ada1f5 100644 --- a/patroni/validator.py +++ b/patroni/validator.py @@ -785,7 +785,7 @@ class IntValidator(object): self.base_unit = base_unit self.raise_assert = raise_assert - def __call__(self, value: Union[int, str]) -> bool: + def __call__(self, value: Any) -> bool: """Check if *value* is a valid integer and within the expected range. .. note:: @@ -821,7 +821,7 @@ class EnumValidator(object): self.allowed_values = set(allowed_values) if case_sensitive else CaseInsensitiveSet(allowed_values) self.raise_assert = raise_assert - def __call__(self, value: str) -> bool: + def __call__(self, value: Any) -> bool: """Check if provided *value* could be found within *allowed_values*. .. note:: @@ -829,7 +829,7 @@ class EnumValidator(object): :param value: value to be checked. :returns: ``True`` if *value* could be found within *allowed_values*. """ - ret = value in self.allowed_values + ret = isinstance(value, str) and value in self.allowed_values if self.raise_assert: assert_(ret) diff --git a/tests/test_config.py b/tests/test_config.py index 57a717f5..3f7e4049 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -21,7 +21,8 @@ class TestConfig(unittest.TestCase): with patch.object(Config, '_build_effective_configuration', Mock(side_effect=Exception)): self.assertFalse(self.config.set_dynamic_configuration({'foo': 'bar'})) self.assertTrue(self.config.set_dynamic_configuration({'standby_cluster': {}, 'postgresql': { - 'parameters': {'cluster_name': 1, 'wal_keep_size': 1, 'track_commit_timestamp': 1, 'wal_level': 1}}})) + 'parameters': {'cluster_name': 1, 'hot_standby': 1, 'wal_keep_size': 1, + 'track_commit_timestamp': 1, 'wal_level': 1}}})) def test_reload_local_configuration(self): os.environ.update({ From a26e46cf76f2e80ae6c53bff4b97d59a10c59e6a Mon Sep 17 00:00:00 2001 From: Israel Date: Mon, 31 Jul 2023 10:41:14 -0300 Subject: [PATCH 22/26] Fix `replicatefrom` tag in `postgres2.yml` (#2788) The node `postgresql2` in the repository is apparently supposed to be a cascading standby. However, if that is the case, there is a typo in the name of its upstream node. This commit fixes that issue. --- postgres2.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 From e4703d4f74e518c15adfe3b081c5bc869fd38812 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Mon, 31 Jul 2023 15:52:43 +0200 Subject: [PATCH 23/26] Refactor replica_list (#2790) As suggested in https://github.com/zalando/patroni/pull/2668/files#r1276115738, introduce a couple of classes that represent a single replica and collection of replicas. --- patroni/postgresql/sync.py | 124 ++++++++++++++++++++++++++----------- 1 file changed, 88 insertions(+), 36 deletions(-) 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 From ec61aede858292b70a0cbf621bbcbf28596df660 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 1 Aug 2023 13:57:46 +0200 Subject: [PATCH 24/26] Fix bug in the Cluster class (#2794) The `workers` attribute when not passed explicitly was set to the same mutable `dict` object every time. Problem was introduced in #2652 --- patroni/dcs/__init__.py | 64 +++++++++++++++++++++++++---------------- 1 file changed, 39 insertions(+), 25 deletions(-) diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index 29a4d766..4bcac3ce 100644 --- a/patroni/dcs/__init__.py +++ b/patroni/dcs/__init__.py @@ -555,36 +555,50 @@ 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': + """Produce an empty :class:`Cluster` instance.""" return Cluster(None, None, None, 0, [], None, SyncState.empty(), None, None, None) def is_empty(self): From b7caf3b7f23924711766ef24250d09c4c198b4a1 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 1 Aug 2023 14:09:46 +0200 Subject: [PATCH 25/26] Fix behaviour of replicas in standby cluster in pause (#2795) When the leader key expires replicas should not follow the remote node but keep `primary_conninfo` as it is. --- patroni/ha.py | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/patroni/ha.py b/patroni/ha.py index 2445aa02..994db00b 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -527,10 +527,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: From 018a2f4dd9d633f7985d2a8a01bf498ef7ac833e Mon Sep 17 00:00:00 2001 From: Israel Date: Tue, 1 Aug 2023 10:40:07 -0300 Subject: [PATCH 26/26] Enhance docs of `slots` dynamic configuration (#2797) The docs of `slots` configuration used to have this mention: ``` 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. ``` However that is not true in the sense that Patroni does not check for clashes between `my_slot_name` and the name of replication slots created for replicating changes among members. If you specify a slot name that clashes with the name of a replication slot used by a member, it turns out Patroni will make the slot permanent in the primary even if the member key expire from the DCS. Through this commit we also enhance the docs in terms of explaining that physical permanent slots are maintained only in the primary, while logical replication slots are copied from primary to standbys. Signed-off-by: Israel Barth Rubio --- docs/dynamic_configuration.rst | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) 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.