From 535dc631ec96ea04d19cf7f4d736b27c068b2c0a Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 10 Oct 2023 12:21:19 +0200 Subject: [PATCH 01/19] Bugfix: standby cluster switchover (#2900) 1. Enforce `_load_cluster()` after acquisition for the leader lock in ZooKeeper. Sometimes the notification from ZooKeeper was arriving too late and Patroni wasn't setting the `role=standby_leader`. 2. The `_get_node_to_follow()` method was falsely assuming that we still own the leader lock and returning the remote node instead of the new standby leader. While not a big issue per se, because the next HA loop usually fixes it, such behavior was causing flakiness of behave tests with Postgres 12 and older, where restart is required to update `primary_conninfo` GUC. --- patroni/dcs/zookeeper.py | 1 + patroni/ha.py | 4 +++- 2 files changed, 4 insertions(+), 1 deletion(-) diff --git a/patroni/dcs/zookeeper.py b/patroni/dcs/zookeeper.py index 863d7b40..649c1ac5 100644 --- a/patroni/dcs/zookeeper.py +++ b/patroni/dcs/zookeeper.py @@ -334,6 +334,7 @@ class ZooKeeper(AbstractDCS): try: self._client.retry(self._client.create, self.leader_path, self._name.encode('utf-8'), makepath=True, ephemeral=True) + self.cluster_watcher(None) # the next _load_cluster() call must read from ZooKeeper. return True except (ConnectionClosedError, RetryFailedError) as e: raise ZooKeeperError(e) diff --git a/patroni/ha.py b/patroni/ha.py index 3b936e4b..877e24ef 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -584,7 +584,9 @@ class Ha(object): """ # 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()): + if self.is_standby_cluster() \ + and (cluster.leader and cluster.leader.name and cluster.leader.name == self.state_handler.name + or 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: From fb367cd73e65b536dc721f83c63ce08dfc0d703b Mon Sep 17 00:00:00 2001 From: Polina Bungina <27892524+hughcapet@users.noreply.github.com> Date: Tue, 10 Oct 2023 13:49:52 +0200 Subject: [PATCH 02/19] Change cb checks in standby cluster behave test (#2899) fix and extend callback content checks --- features/standby_cluster.feature | 13 +++++-------- 1 file changed, 5 insertions(+), 8 deletions(-) diff --git a/features/standby_cluster.feature b/features/standby_cluster.feature index ac6bbb85..a9f00c01 100644 --- a/features/standby_cluster.feature +++ b/features/standby_cluster.feature @@ -54,21 +54,18 @@ Feature: standby cluster And postgres1 does not have a logical replication slot named test_logical Scenario: check switchover - When I run patronictl.py switchover batman1 --force - And I issue a GET request to http://127.0.0.1:8010/standby_leader - Then I receive a response code 200 - And I receive a response role standby_leader + Given I run patronictl.py switchover batman1 --force + Then Status code on GET http://127.0.0.1:8010/standby_leader is 200 after 10 seconds And postgres1 is replicating from postgres2 after 32 seconds + And there is a postgres2_cb.log with "on_start replica batman1\non_role_change standby_leader batman1" in postgres2 data directory Scenario: check failover When I kill postgres2 And I kill postmaster on postgres2 Then postgres1 is replicating from postgres0 after 32 seconds + And Status code on GET http://127.0.0.1:8009/standby_leader is 200 after 10 seconds When I issue a GET request to http://127.0.0.1:8009/primary Then I receive a response code 503 - And I sleep for 3 seconds - When I issue a GET request to http://127.0.0.1:8009/standby_leader - Then I receive a response code 200 And I receive a response role standby_leader And replication works from postgres0 to postgres1 after 15 seconds - And there is a postgres1_cb.log with "on_start replica batman1\non_role_change standby_leader batman1" in postgres1 data directory + And there is a postgres1_cb.log with "on_role_change replica batman1\non_role_change standby_leader batman1" in postgres1 data directory From 588df5da05d9631d6df5465fc90033c792916e70 Mon Sep 17 00:00:00 2001 From: Chris Bandy Date: Wed, 11 Oct 2023 01:41:11 -0500 Subject: [PATCH 03/19] Refine the documentation about custom_conf (#2901) some back icks in this section needed to be balanced. --- docs/patroni_configuration.rst | 17 +++++++++-------- 1 file changed, 9 insertions(+), 8 deletions(-) diff --git a/docs/patroni_configuration.rst b/docs/patroni_configuration.rst index 3978e48d..e18c118e 100644 --- a/docs/patroni_configuration.rst +++ b/docs/patroni_configuration.rst @@ -70,15 +70,16 @@ There also are some parameters like **postgresql.listen**, **postgresql.data_dir When applying the local or dynamic configuration options, the following actions are taken: -- The node first checks if there is a `postgresql.base.conf` or if the ``custom_conf`` parameter is set. -- If the ``custom_conf`` parameter is set, it will take the file specified on it as a base configuration, ignoring `postgresql.base.conf` and `postgresql.conf`. -- If the ``custom_conf`` parameter is not set and `postgresql.base.conf` exists, it contains the renamed "original" configuration and it will be used as a base configuration. -- If there is no ``custom_conf``` nor `postgresql.base.conf`, the original `postgresql.conf`` is taken and renamed to postgresql.base.conf. -- The dynamic options (with the exceptions above) are dumped into the `postgresql.conf`` and an include is set in - postgresql.conf to the used base configuration (either `postgresql.base.conf` or what is on ``custom_conf``). Therefore, we would be able to apply new options without re-reading the configuration file to check if the include is present not. +- The node first checks if there is a `postgresql.base.conf` file or if the ``custom_conf`` parameter is set. +- If the ``custom_conf`` parameter is set, the file it specifies is used as the base configuration, ignoring `postgresql.base.conf` and `postgresql.conf`. +- If the ``custom_conf`` parameter is not set and `postgresql.base.conf` exists, it contains the renamed "original" configuration and is used as the base configuration. +- If there is no ``custom_conf`` nor `postgresql.base.conf`, the original `postgresql.conf` is renamed to `postgresql.base.conf` and used as the base configuration. +- The dynamic options (with the exceptions above) are dumped into the `postgresql.conf` and an include is set in + `postgresql.conf` to the base configuration (either `postgresql.base.conf` or the file at ``custom_conf``). + Therefore, we would be able to apply new options without re-reading the configuration file to check if the include is present or not. - Some parameters that are essential for Patroni to manage the cluster are overridden using the command line. -- If some of the options that require restart are changed (we should look at the context in pg_settings and at the actual - values of those options), a pending_restart flag of a given node is set. This flag is reset on any restart. +- If an option that requires restart is changed (we should look at the context in pg_settings and at the actual + values of those options), a pending_restart flag is set on that node. This flag is reset on any restart. The parameters would be applied in the following order (run-time are given the highest priority): From 6f4c2fe132afae67fbc7989b360b8809f4eb4dac Mon Sep 17 00:00:00 2001 From: zhjwpku Date: Wed, 11 Oct 2023 19:17:18 +0800 Subject: [PATCH 04/19] %s/iter_dcs_modules/iter_dcs_classes/g (#2905) --- patroni/dcs/__init__.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index 08112651..d71d41ad 100644 --- a/patroni/dcs/__init__.py +++ b/patroni/dcs/__init__.py @@ -158,7 +158,7 @@ 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 + Using the list of available DCS classes returned by :func:`iter_dcs_classes` attempt to dynamically 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 From 42976df86fa5938225c50b92d2b6ba8dbb11a572 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Mon, 16 Oct 2023 08:55:07 +0200 Subject: [PATCH 05/19] Make it easier to debug callbacks (#2902) 1. Introduce DEBUG logs for callbacks 2. Configure log format in behave tests to include filename, line, and method name that triggered the callback and enable DEBUG logs for `patroni.postgresql.callback_executor` module. P.S. unfortunately it works only starting from python 3.8, but it should be good enough for debug purpose because 3.7 is already EOL. --- features/environment.py | 4 ++++ patroni/postgresql/callback_executor.py | 5 ++++- 2 files changed, 8 insertions(+), 1 deletion(-) diff --git a/features/environment.py b/features/environment.py index 79db5d16..1f36eb0c 100644 --- a/features/environment.py +++ b/features/environment.py @@ -245,6 +245,10 @@ class PatroniController(AbstractController): self.recursive_update(config, custom_config) self.recursive_update(config, { + 'log': { + 'format': '%(asctime)s %(levelname)s [%(pathname)s:%(lineno)d - %(funcName)s]: %(message)s', + 'loggers': {'patroni.postgresql.callback_executor': 'DEBUG'} + }, 'bootstrap': { 'dcs': { 'loop_wait': 2, diff --git a/patroni/postgresql/callback_executor.py b/patroni/postgresql/callback_executor.py index fa645b86..06b9f353 100644 --- a/patroni/postgresql/callback_executor.py +++ b/patroni/postgresql/callback_executor.py @@ -1,8 +1,9 @@ import logging +import sys from enum import Enum from threading import Condition, Thread -from typing import List +from typing import Any, Dict, List from .cancellable import CancellableExecutor, CancellableSubprocess @@ -53,6 +54,8 @@ class CallbackExecutor(CancellableExecutor, Thread): If it couldn't be killed we wait until it finishes. :param cmd: command to be executed""" + kwargs: Dict[str, Any] = {'stacklevel': 3} if sys.version_info >= (3, 8) else {} + logger.debug('CallbackExecutor.call(%s)', cmd, **kwargs) if cmd[-3] == CallbackAction.ON_RELOAD: return self._on_reload_executor.call_nowait(cmd) From d93db20baa75eb0ed3c0668e96b02b0359e65f86 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Mon, 16 Oct 2023 10:21:50 +0200 Subject: [PATCH 06/19] Set citus.local_hostname (#2903) There are cases when Citus wants to have a connection to the local postgres. By default it uses `localhost` for that, which is not alwasy available. To solve it we will set `citus.local_hostname` GUC to custom value, which is the same as Patroni uses to connect to Postgres. --- docs/citus.rst | 12 ++++++++---- patroni/postgresql/citus.py | 3 +++ tests/__init__.py | 2 +- tests/test_bootstrap.py | 2 +- tests/test_citus.py | 2 +- 5 files changed, 14 insertions(+), 7 deletions(-) diff --git a/docs/citus.rst b/docs/citus.rst index 084931df..62cbdd1b 100644 --- a/docs/citus.rst +++ b/docs/citus.rst @@ -38,14 +38,18 @@ After that you just need to start Patroni and it will handle the rest: 2. If ``max_prepared_transactions`` isn't explicitly set in the global :ref:`dynamic configuration ` Patroni will automatically set it to ``2*max_connections``. -3. The ``citus.database`` will be automatically created followed by ``CREATE EXTENSION citus``. -4. Current superuser :ref:`credentials ` will be added to the ``pg_dist_authinfo`` +3. The ``citus.local_hostname`` GUC value will be adjusted from ``localhost`` to the + value that Patroni is using in order to connect to the local PostgreSQL + instance. The value sometimes should be different from the ``localhost`` + because PostgreSQL might be not listening on it. +4. The ``citus.database`` will be automatically created followed by ``CREATE EXTENSION citus``. +5. Current superuser :ref:`credentials ` will be added to the ``pg_dist_authinfo`` table to allow cross-node communication. Don't forget to update them if later you decide to change superuser username/password/sslcert/sslkey! -5. The coordinator primary node will automatically discover worker primary +6. The coordinator primary node will automatically discover worker primary nodes and add them to the ``pg_dist_node`` table using the ``citus_add_node()`` function. -6. Patroni will also maintain ``pg_dist_node`` in case failover/switchover +7. Patroni will also maintain ``pg_dist_node`` in case failover/switchover on the coordinator or worker clusters occurs. patronictl diff --git a/patroni/postgresql/citus.py b/patroni/postgresql/citus.py index 8ca2790e..26923f37 100644 --- a/patroni/postgresql/citus.py +++ b/patroni/postgresql/citus.py @@ -403,6 +403,9 @@ class CitusHandler(Thread): # Resharding in Citus implemented using logical replication parameters['wal_level'] = 'logical' + # Sometimes Citus needs to connect to the local postgres. We will do it the same way as Patroni does. + parameters['citus.local_hostname'] = self._postgresql.connection_pool.conn_kwargs.get('host', 'localhost') + def ignore_replication_slot(self, slot: Dict[str, str]) -> bool: if isinstance(self._config, dict) and self._postgresql.is_primary() and\ slot['type'] == 'logical' and slot['database'] == self._config['database']: diff --git a/tests/__init__.py b/tests/__init__.py index 32172926..bd70ba3d 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -241,7 +241,7 @@ class PostgresInit(unittest.TestCase): 'replication': {'username': '', 'password': 'rep-pass'}, 'rewind': {'username': 'rewind', 'password': 'test'}}, 'remove_data_directory_on_rewind_failure': True, - 'use_pg_rewind': True, 'pg_ctl_timeout': 'bla', + 'use_pg_rewind': True, 'pg_ctl_timeout': 'bla', 'use_unix_socket': True, 'parameters': self._PARAMETERS, 'recovery_conf': {'foo': 'bar'}, 'pg_hba': ['host all all 0.0.0.0/0 md5'], diff --git a/tests/test_bootstrap.py b/tests/test_bootstrap.py index 9ac6aa68..4c2d1c98 100644 --- a/tests/test_bootstrap.py +++ b/tests/test_bootstrap.py @@ -256,7 +256,7 @@ class TestBootstrap(BaseTestPostgresql): mock_cancellable_subprocess_call.assert_called() args, kwargs = mock_cancellable_subprocess_call.call_args self.assertTrue('PGPASSFILE' in kwargs['env']) - self.assertEqual(args[0], ['/bin/false', 'dbname=postgres host=127.0.0.2 port=5432']) + self.assertEqual(args[0], ['/bin/false', 'dbname=postgres host=/tmp port=5432']) mock_cancellable_subprocess_call.reset_mock() self.p.connection_pool._conn_kwargs.pop('host') diff --git a/tests/test_citus.py b/tests/test_citus.py index 9849a069..7279893e 100644 --- a/tests/test_citus.py +++ b/tests/test_citus.py @@ -13,7 +13,6 @@ class TestCitus(BaseTestPostgresql): def setUp(self): super(TestCitus, self).setUp() self.c = self.p.citus_handler - self.p.connection_pool.conn_kwargs = {'host': 'localhost', 'dbname': 'postgres'} self.cluster = get_cluster_initialized_with_leader() self.cluster.workers[1] = self.cluster @@ -139,6 +138,7 @@ class TestCitus(BaseTestPostgresql): self.assertEqual(parameters['max_prepared_transactions'], 202) self.assertEqual(parameters['shared_preload_libraries'], 'citus,foo,bar') self.assertEqual(parameters['wal_level'], 'logical') + self.assertEqual(parameters['citus.local_hostname'], '/tmp') def test_bootstrap(self): self.c._config = None From 88b35252c39d7c65cd9e8dbdfe3ba5bb6f777557 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andr=C3=A9=20Litfin?= Date: Mon, 16 Oct 2023 15:18:25 +0200 Subject: [PATCH 07/19] Update README.md to reflect changes in etcd v3 (#2912) In etcdctl v3 the ls command isn't present anymore, it has to be changed to etcdctl get --keys-only --prefix --- docker/README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docker/README.md b/docker/README.md index f2b30ab6..da87842f 100644 --- a/docker/README.md +++ b/docker/README.md @@ -78,7 +78,7 @@ Example session: | demo | patroni3 | 172.22.0.4 | | running | 1 | 0 | +---------+----------+------------+--------+---------+----+-----------+ - postgres@patroni1:~$ etcdctl ls --recursive --sort -p /service/demo + postgres@patroni1:~$ etcdctl get --keys-only --prefix /service/demo /service/demo/config /service/demo/initialize /service/demo/leader From c96e35c807fcf1f8598612761f2aaa4acb644590 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Mon, 16 Oct 2023 16:05:27 +0200 Subject: [PATCH 08/19] Enable Citus behave tests for Postgres v16 (#2914) and reduce flakiness --- .github/workflows/install_deps.py | 4 ++-- features/citus.feature | 2 +- features/steps/citus.py | 18 ++++++++++++------ 3 files changed, 15 insertions(+), 9 deletions(-) diff --git a/.github/workflows/install_deps.py b/.github/workflows/install_deps.py index ea0c9d1b..6480f66a 100644 --- a/.github/workflows/install_deps.py +++ b/.github/workflows/install_deps.py @@ -45,8 +45,8 @@ def install_packages(what): packages['exhibitor'] = packages['zookeeper'] packages = packages.get(what, []) ver = versions.get(what) - if float(ver) == 15: - packages += ['postgresql-{0}-citus-12.0'.format(ver)] + if float(ver) >= 15: + packages += ['postgresql-{0}-citus-12.1'.format(ver)] subprocess.call(['sudo', 'apt-get', 'update', '-y']) return subprocess.call(['sudo', 'apt-get', 'install', '-y', 'postgresql-' + ver, 'expect-dev'] + packages) diff --git a/features/citus.feature b/features/citus.feature index 35ccebbe..b23eb2b2 100644 --- a/features/citus.feature +++ b/features/citus.feature @@ -68,6 +68,6 @@ Feature: citus And I receive a response output "+ttl: 20" Then postgres4 is registered in the postgres2 as the primary in group 2 after 5 seconds When I shut down postgres4 - Then There is a transaction in progress on postgres0 changing pg_dist_node + Then there is a transaction in progress on postgres0 changing pg_dist_node after 5 seconds When I run patronictl.py restart batman postgres2 --group 1 --force Then a transaction finishes in 20 seconds diff --git a/features/steps/citus.py b/features/steps/citus.py index 644219c7..4dd2ffa6 100644 --- a/features/steps/citus.py +++ b/features/steps/citus.py @@ -115,12 +115,18 @@ def count_rows(context, name): assert rows == context.insert_counter, "Distributed table doesn't have expected amount of rows" -@step("There is a transaction in progress on {name:w} changing pg_dist_node") -def check_transaction(context, name): - cur = context.pctl.query(name, "SELECT xact_start FROM pg_stat_activity WHERE pid <> pg_backend_pid()" - " AND state = 'idle in transaction' AND query ~ 'citus_update_node'") - assert cur.rowcount == 1, "There is no idle in transaction updating pg_dist_node" - context.xact_start = cur.fetchone()[0] +@step("there is a transaction in progress on {name:w} changing pg_dist_node after {time_limit:d} seconds") +def check_transaction(context, name, time_limit): + time_limit *= context.timeout_multiplier + max_time = time.time() + int(time_limit) + while time.time() < max_time: + cur = context.pctl.query(name, "SELECT xact_start FROM pg_stat_activity WHERE pid <> pg_backend_pid()" + " AND state = 'idle in transaction' AND query ~ 'citus_update_node'") + if cur.rowcount == 1: + context.xact_start = cur.fetchone()[0] + return + time.sleep(1) + assert False, f"There is no idle in transaction on {name} updating pg_dist_node after {time_limit} seconds" @step("a transaction finishes in {timeout:d} seconds") From aa3ebe0af8bb1966230680857fd9d230e4d3c99c Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 17 Oct 2023 08:56:31 +0200 Subject: [PATCH 09/19] Don't cache anything in Zookeeper implementation (#2909) Cache creates a lot of problems and prevents implementing a feature of automatic retention of physical replication slots for members with configurable retention policy. Just read the entire cluster from Zookeeper instead and use watchers only for the `/leader` and `/config` keys. --- features/environment.py | 2 +- patroni/api.py | 6 +- patroni/dcs/__init__.py | 18 ++---- patroni/dcs/zookeeper.py | 117 +++++++++++---------------------------- tests/test_zookeeper.py | 22 +++----- 5 files changed, 47 insertions(+), 118 deletions(-) diff --git a/features/environment.py b/features/environment.py index 1f36eb0c..3e0ad3b1 100644 --- a/features/environment.py +++ b/features/environment.py @@ -692,7 +692,7 @@ class ZooKeeperController(AbstractExternalDcsController): self._client = kazoo.client.KazooClient() def process_name(self): - return "zookeeper" + return "java .*zookeeper" def query(self, key, scope='batman', group=None): import kazoo.exceptions diff --git a/patroni/api.py b/patroni/api.py index 911ea574..e1b71337 100644 --- a/patroni/api.py +++ b/patroni/api.py @@ -451,7 +451,7 @@ class RestApiHandler(BaseHTTPRequestHandler): Write an HTTP response with JSON content based on the output of :func:`~patroni.utils.cluster_as_json`, with HTTP status ``200`` and the JSON representation of the cluster topology. """ - cluster = self.server.patroni.dcs.get_cluster(True) + cluster = self.server.patroni.dcs.get_cluster() global_config = self.server.patroni.config.get_global_config(cluster) response = cluster_as_json(cluster, global_config) @@ -690,7 +690,7 @@ class RestApiHandler(BaseHTTPRequestHandler): """ request = self._read_json_content() if request: - cluster = self.server.patroni.dcs.get_cluster(True) + cluster = self.server.patroni.dcs.get_cluster() if not (cluster.config and cluster.config.modify_version): return self.send_error(503) data = cluster.config.data.copy() @@ -1166,7 +1166,7 @@ class RestApiHandler(BaseHTTPRequestHandler): patroni = self.server.patroni if patroni.postgresql.citus_handler.is_coordinator() and patroni.ha.is_leader(): - cluster = patroni.dcs.get_cluster(True) + cluster = patroni.dcs.get_cluster() patroni.postgresql.citus_handler.handle_event(cluster, request) self.write_response(200, 'OK') diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index d71d41ad..1a516cca 100644 --- a/patroni/dcs/__init__.py +++ b/patroni/dcs/__init__.py @@ -1548,9 +1548,6 @@ class AbstractDCS(abc.ABC): primary and exception raised, instance would be demoted. """ - def _bypass_caches(self) -> None: - """Used only in Zookeeper.""" - def __get_patroni_cluster(self, path: Optional[str] = None) -> Cluster: """Low level method to load a :class:`Cluster` object from DCS. @@ -1593,14 +1590,13 @@ class AbstractDCS(abc.ABC): dict. """ groups = self._load_cluster(self._base_path + '/', self._citus_cluster_loader) - if isinstance(groups, Cluster): # Zookeeper could return a cached version - cluster = groups - else: - cluster = groups.pop(CITUS_COORDINATOR_GROUP_ID, Cluster.empty()) - cluster.workers.update(groups) + if TYPE_CHECKING: # pragma: no cover + assert isinstance(groups, dict) + cluster = groups.pop(CITUS_COORDINATOR_GROUP_ID, Cluster.empty()) + cluster.workers.update(groups) return cluster - def get_cluster(self, force: bool = False) -> Cluster: + def get_cluster(self) -> Cluster: """Retrieve an appropriate cached or fresh view of DCS. .. note:: @@ -1609,12 +1605,8 @@ class AbstractDCS(abc.ABC): Returns either a Citus or Patroni implementation of :class:`Cluster` depending on availability. - :param force: a value of ``True`` will override Zookeeper caching features. - :returns: """ - if force: - self._bypass_caches() try: cluster = self._get_citus_cluster() if self.is_citus_coordinator() else self.__get_patroni_cluster() except Exception: diff --git a/patroni/dcs/zookeeper.py b/patroni/dcs/zookeeper.py index 649c1ac5..3093ba06 100644 --- a/patroni/dcs/zookeeper.py +++ b/patroni/dcs/zookeeper.py @@ -116,10 +116,7 @@ class ZooKeeper(AbstractDCS): timeout=config['ttl'], connection_retry=KazooRetry(max_delay=1, max_tries=-1, sleep_func=time.sleep), command_retry=KazooRetry(max_delay=1, max_tries=-1, deadline=config['retry_timeout'], sleep_func=time.sleep), **kwargs) - self._client.add_listener(self.session_listener) - self._fetch_cluster: bool = True - self._fetch_status: bool = True self.__last_member_data: Optional[Dict[str, Any]] = None self._orig_kazoo_connect = self._client._connection._connect @@ -142,18 +139,9 @@ class ZooKeeper(AbstractDCS): ret = self._orig_kazoo_connect(*args) return max(self.loop_wait - 2, 2) * 1000, ret[1] - def session_listener(self, state: str) -> None: - if state in [KazooState.SUSPENDED, KazooState.LOST]: - self.cluster_watcher(None) - - def status_watcher(self, event: Optional[WatchedEvent]) -> None: - self._fetch_status = True - self.event.set() - - def cluster_watcher(self, event: Optional[WatchedEvent]) -> None: - self._fetch_cluster = True - if not event or event.state != KazooState.CONNECTED or event.path.startswith(self.client_path('')): - self.status_watcher(event) + def _watcher(self, event: WatchedEvent) -> None: + if event.state != KazooState.CONNECTED or event.path.startswith(self.client_path('')): + self.event.set() def reload_config(self, config: Union['Config', Dict[str, Any]]) -> None: self.set_retry_timeout(config['retry_timeout']) @@ -202,75 +190,66 @@ class ZooKeeper(AbstractDCS): return None def get_status(self, path: str, leader: Optional[Leader]) -> Status: - watch = self.status_watcher if not leader or leader.name != self._name else None - - status = self.get_node(path + self._STATUS, watch) + status = self.get_node(path + self._STATUS) if not status: - status = self.get_node(path + self._LEADER_OPTIME, watch) - if status: - self._fetch_status = False + status = self.get_node(path + self._LEADER_OPTIME) return Status.from_node(status and status[0]) @staticmethod def member(name: str, value: str, znode: ZnodeStat) -> Member: return Member.from_node(znode.version, name, znode.ephemeralOwner, value) - def get_children(self, key: str, watch: Optional[Callable[[WatchedEvent], None]] = None) -> List[str]: + def get_children(self, key: str) -> List[str]: try: - return self._client.get_children(key, watch) + return self._client.get_children(key) except NoNodeError: return [] def load_members(self, path: str) -> List[Member]: members: List[Member] = [] - for member in self.get_children(path + self._MEMBERS, self.cluster_watcher): + for member in self.get_children(path + self._MEMBERS): data = self.get_node(path + self._MEMBERS + member) if data is not None: members.append(self.member(member, *data)) return members def _cluster_loader(self, path: str) -> Cluster: - self._fetch_cluster = False - self.event.clear() - nodes = set(self.get_children(path, self.cluster_watcher)) - if not nodes: - self._fetch_cluster = True + nodes = set(self.get_children(path)) # get initialize flag initialize = (self.get_node(path + self._INITIALIZE) or [None])[0] if self._INITIALIZE in nodes else None # get global dynamic configuration - config = self.get_node(path + self._CONFIG, watch=self.cluster_watcher) if self._CONFIG in nodes else None + config = self.get_node(path + self._CONFIG, watch=self._watcher) if self._CONFIG in nodes else None config = config and ClusterConfig.from_node(config[1].version, config[0], config[1].mzxid) # get timeline history - history = self.get_node(path + self._HISTORY, watch=self.cluster_watcher) if self._HISTORY in nodes else None + history = self.get_node(path + self._HISTORY) if self._HISTORY in nodes else None history = history and TimelineHistory.from_node(history[1].mzxid, history[0]) # get synchronization state - sync = self.get_node(path + self._SYNC, watch=self.cluster_watcher) if self._SYNC in nodes else None + sync = self.get_node(path + self._SYNC) if self._SYNC in nodes else None sync = SyncState.from_node(sync and sync[1].version, sync and sync[0]) # get list of members members = self.load_members(path) if self._MEMBERS[:-1] in nodes else [] # get leader - leader = self.get_node(path + self._LEADER) if self._LEADER in nodes else None + leader = self.get_node(path + self._LEADER, watch=self._watcher) if self._LEADER in nodes else None if leader: member = Member(-1, leader[0], None, {}) member = ([m for m in members if m.name == leader[0]] or [member])[0] leader = Leader(leader[1].version, leader[1].ephemeralOwner, member) - self._fetch_cluster = member.version == -1 # get last known leader lsn and slots status = self.get_status(path, leader) # failover key - failover = self.get_node(path + self._FAILOVER, watch=self.cluster_watcher) if self._FAILOVER in nodes else None + failover = self.get_node(path + self._FAILOVER) if self._FAILOVER in nodes else None failover = failover and Failover.from_node(failover[1].version, failover[0]) # get failsafe topology - failsafe = self.get_node(path + self._FAILSAFE, watch=self.cluster_watcher) if self._FAILSAFE in nodes else None + failsafe = self.get_node(path + self._FAILSAFE) if self._FAILSAFE in nodes else None try: failsafe = json.loads(failsafe[0]) if failsafe else None except Exception: @@ -279,45 +258,20 @@ class ZooKeeper(AbstractDCS): return Cluster(initialize, config, leader, status, members, failover, sync, history, failsafe) def _citus_cluster_loader(self, path: str) -> Dict[int, Cluster]: - fetch_cluster = False ret: Dict[int, Cluster] = {} - for node in self.get_children(path, self.cluster_watcher): + for node in self.get_children(path): if citus_group_re.match(node): ret[int(node)] = self._cluster_loader(path + node + '/') - fetch_cluster = fetch_cluster or self._fetch_cluster - self._fetch_cluster = fetch_cluster return ret def _load_cluster( self, path: str, loader: Callable[[str], Union[Cluster, Dict[int, Cluster]]] ) -> Union[Cluster, Dict[int, Cluster]]: - cluster = self.cluster if path == self._base_path + '/' else None - if self._fetch_cluster or cluster is None: - try: - cluster = self._client.retry(loader, path) - except Exception: - logger.exception('get_cluster') - self.cluster_watcher(None) - raise ZooKeeperError('ZooKeeper in not responding properly') - # The /status ZNode was updated or doesn't exist - elif self._fetch_status and not self._fetch_cluster or not cluster.last_lsn \ - or cluster.has_permanent_slots(self._name) and not cluster.slots: - # If current node is the leader just clear the event without fetching anything (we are updating the /status) - if cluster.leader and cluster.leader.name == self._name: - self.event.clear() - else: - try: - status = self.get_status(self.client_path(''), cluster.leader) - self.event.clear() - new_cluster: List[Any] = list(cluster) - new_cluster[3] = status - cluster = Cluster(*new_cluster) - except Exception: - pass - return cluster - - def _bypass_caches(self) -> None: - self._fetch_cluster = True + try: + return self._client.retry(loader, path) + except Exception: + logger.exception('get_cluster') + raise ZooKeeperError('ZooKeeper in not responding properly') def _create(self, path: str, value: bytes, retry: bool = False, ephemeral: bool = False) -> bool: try: @@ -334,7 +288,6 @@ class ZooKeeper(AbstractDCS): try: self._client.retry(self._client.create, self.leader_path, self._name.encode('utf-8'), makepath=True, ephemeral=True) - self.cluster_watcher(None) # the next _load_cluster() call must read from ZooKeeper. return True except (ConnectionClosedError, RetryFailedError) as e: raise ZooKeeperError(e) @@ -381,20 +334,9 @@ class ZooKeeper(AbstractDCS): member = cluster and cluster.get_member(self._name, fallback_to_leader=False) member_data = self.__last_member_data or member and member.data if member and member_data: - is_leader = data.get('role') in ('master', 'primary', 'standby_leader') - checkpoint_after_promote_changed = member_data.get('checkpoint_after_promote') \ - != data.get('checkpoint_after_promote') - state_running_changed = member_data.get('state') != data.get('state') \ - and 'running' in (member_data.get('state'), data.get('state')) - tags_changed = not deep_compare(member_data.get('tags', {}), data.get('tags', {})) - - # We want delete the member ZNode if: - # - our session doesn't match with session id on our member key; or - # - we want to notify leader if some important fields in the member key changed; or - # - if we are the leader and want to notify replicas about checkpoint_after_promote; - if self._client.client_id is not None and member.session != self._client.client_id[0] \ - or is_leader and checkpoint_after_promote_changed \ - or not is_leader and (state_running_changed or tags_changed): + # We want delete the member ZNode if our session doesn't match with session id on our member key + if self._client.client_id is not None and member.session != self._client.client_id[0]: + logger.warning('Recreating the member ZNode due to ownership mismatch') try: self._client.delete_async(self.member_path).get(timeout=1) except NoNodeError: @@ -493,7 +435,10 @@ class ZooKeeper(AbstractDCS): return self.set_sync_state_value("{}", version) is not False def watch(self, leader_version: Optional[int], timeout: float) -> bool: - ret = super(ZooKeeper, self).watch(leader_version, timeout + 0.5) - if ret and not self._fetch_status: - self._fetch_cluster = True - return ret or self._fetch_cluster + if leader_version: + timeout += 0.5 + + try: + return super(ZooKeeper, self).watch(leader_version, timeout) + finally: + self.event.clear() diff --git a/tests/test_zookeeper.py b/tests/test_zookeeper.py index 184fda05..3ce3ea75 100644 --- a/tests/test_zookeeper.py +++ b/tests/test_zookeeper.py @@ -1,13 +1,13 @@ import select import unittest -from kazoo.client import KazooClient, KazooState +from kazoo.client import KazooClient from kazoo.exceptions import NoNodeError, NodeExistsError from kazoo.handlers.threading import SequentialThreadingHandler -from kazoo.protocol.states import KeeperState, ZnodeStat +from kazoo.protocol.states import KeeperState, WatchedEvent, 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, PatroniKazooClient, \ PatroniSequentialThreadingHandler, ZooKeeper, ZooKeeperError @@ -152,9 +152,6 @@ class TestZooKeeper(unittest.TestCase): 'name': 'foo', 'ttl': 30, 'retry_timeout': 10, 'loop_wait': 10, 'set_acls': {'CN=principal2': ['ALL']}}) - def test_session_listener(self): - self.zk.session_listener(KazooState.SUSPENDED) - def test_reload_config(self): self.zk.reload_config({'ttl': 20, 'retry_timeout': 10, 'loop_wait': 10}) self.zk.reload_config({'ttl': 20, 'retry_timeout': 10, 'loop_wait': 5}) @@ -176,15 +173,6 @@ class TestZooKeeper(unittest.TestCase): self.zk._cluster_loader(self.zk.client_path('')) def test_get_cluster(self): - cluster = self.zk.get_cluster(True) - self.assertIsInstance(cluster.leader, Leader) - self.zk.status_watcher(None) - self.zk.get_cluster() - self.zk.touch_member({'foo': 'foo'}) - self.zk._name = 'bar' - self.zk.status_watcher(None) - with patch.object(ZooKeeper, 'get_node', Mock(side_effect=Exception)): - self.zk.get_cluster() cluster = self.zk.get_cluster() self.assertEqual(cluster.last_lsn, 500) @@ -295,3 +283,7 @@ class TestZooKeeper(unittest.TestCase): def test_set_history_value(self): self.zk.set_history_value('{}') + + def test_watcher(self): + self.zk._watcher(WatchedEvent('', '', '')) + self.assertTrue(self.zk.watch(1, 1)) From e513f7f12777f5902e534a44bebd6899ee20ca49 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 17 Oct 2023 11:27:41 +0200 Subject: [PATCH 10/19] Attempt to reduce flakiness for recovery behave test on K8s (#2917) wait until Postgres is properly started after the first crash before changing `primary_start_timeout` and killing it once again. --- features/recovery.feature | 2 ++ 1 file changed, 2 insertions(+) diff --git a/features/recovery.feature b/features/recovery.feature index 809f7fb9..7839e26f 100644 --- a/features/recovery.feature +++ b/features/recovery.feature @@ -14,6 +14,8 @@ Feature: recovery Then I receive a response code 200 And I receive a response role master And I receive a response timeline 1 + And "members/postgres0" key in DCS has state=running after 12 seconds + And replication works from postgres0 to postgres1 after 15 seconds Scenario: check immediate failover when master_start_timeout=0 Given I issue a PATCH request to http://127.0.0.1:8008/config with {"master_start_timeout": 0} From 60d8bc3a70e95a183da9bba1917062642a0171f8 Mon Sep 17 00:00:00 2001 From: GuanqunYang193 <144971563+GuanqunYang193@users.noreply.github.com> Date: Tue, 17 Oct 2023 07:04:59 -0400 Subject: [PATCH 11/19] Add warning of removing user creation (#2893) --- patroni/postgresql/bootstrap.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/patroni/postgresql/bootstrap.py b/patroni/postgresql/bootstrap.py index 26025e43..8a9b0fba 100644 --- a/patroni/postgresql/bootstrap.py +++ b/patroni/postgresql/bootstrap.py @@ -400,6 +400,9 @@ BEGIN END;$$""".format(f, quote_ident(rewind['username'], postgresql.connection())) postgresql.query(sql) + if config.get('users'): + logger.warning('User creation via "bootstrap.users" will be removed in v4.0.0') + for name, value in (config.get('users') or {}).items(): if all(name != a.get('username') for a in (superuser, replication, rewind)): self.create_or_update_role(name, value.get('password'), value.get('options', [])) From fc67ba73f0f29c41c6c1e93825a8648b1509ab16 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 17 Oct 2023 14:46:15 +0200 Subject: [PATCH 12/19] Allow to specify psycopg* in extras and switch to `build` (#2907) * remove check_psycopg() call from the setup.py, when installing from wheel it doesn't work anyway. * call check_psycopg() function before process_arguments(), because the last one is trying to import psycopg and fails with the stacktrace, while the first one shows a nice human-readable error message. * add psycopg2, psycopg2-binary, and psycopg3 extras, that will install psycopg2>=2.5.4, psycopg2-binary, or psycopg[binary]>=3.0.0 modules respectively. * move check_psycopg() function to the __main__.py. * introduce the new extra called `all`, it will allow to install all dependencies at once (except psycopg related). * use the `build` module in order to create sdist bdist_wheel packages. * update the documentation regarding psycopg and extras (dependencies). --- .github/workflows/release.yaml | 5 ++- README.rst | 29 +++++++--------- docs/installation.rst | 29 +++++++--------- patroni/__init__.py | 60 ++++------------------------------ patroni/__main__.py | 46 +++++++++++++++++++++++--- setup.py | 47 +++++++++++++------------- tests/test_patroni.py | 15 ++++++--- 7 files changed, 112 insertions(+), 119 deletions(-) diff --git a/.github/workflows/release.yaml b/.github/workflows/release.yaml index 77a4df5b..7a102379 100644 --- a/.github/workflows/release.yaml +++ b/.github/workflows/release.yaml @@ -24,8 +24,11 @@ jobs: - name: Run tests and flake8 run: python .github/workflows/run_tests.py + - name: Install Python packaging build frontend + run: python -m pip install build + - name: Build a binary wheel and a source tarball - run: python setup.py sdist bdist_wheel + run: python -m build - name: Publish distribution to Test PyPI if: github.event_name == 'push' diff --git a/README.rst b/README.rst index decc868f..c2fc165e 100644 --- a/README.rst +++ b/README.rst @@ -77,23 +77,8 @@ There are a few options available: sudo apt-get install python3-psycopg2 # install psycopg2 module on Debian/Ubuntu sudo yum install python3-psycopg2 # install psycopg2 on RedHat/Fedora/CentOS -2. Install psycopg2 from the binary package +2. Specify one of `psycopg`, `psycopg2`, or `psycopg2-binary` in the list of dependencies when installing Patroni with pip (see below). -:: - - pip install psycopg2-binary - -3. Install psycopg2 from source - -:: - - pip install psycopg2>=2.5.4 - -4. Use psycopg 3.0 instead of psycopg2 - -:: - - pip install psycopg[binary]>=3.0.0 **General installation for pip** @@ -119,12 +104,20 @@ raft `pysyncobj` module in order to use python Raft implementation as DCS aws `boto3` in order to use AWS callbacks +all + all of the above (except psycopg family) +psycopg3 + `psycopg[binary]>=3.0.0` module +psycopg2 + `psycopg2>=2.5.4` module +psycopg2-binary + `psycopg2-binary` module -For example, the command in order to install Patroni together with dependencies for Etcd as a DCS and AWS callbacks is: +For example, the command in order to install Patroni together with psycopg3, dependencies for Etcd as a DCS, and AWS callbacks is: :: - pip install patroni[etcd,aws] + pip install patroni[psycopg3,etcd3,aws] Note that external tools to call in the replica creation or custom bootstrap scripts (i.e. WAL-E) should be installed independently of Patroni. diff --git a/docs/installation.rst b/docs/installation.rst index 6c7a6029..3c9dddfa 100644 --- a/docs/installation.rst +++ b/docs/installation.rst @@ -30,23 +30,10 @@ There are a few options available: sudo apt-get install python3-psycopg2 # install psycopg2 module on Debian/Ubuntu sudo yum install python3-psycopg2 # install psycopg2 on RedHat/Fedora/CentOS -2. Install psycopg2 from the binary package +2. Specify one of `psycopg`, `psycopg2`, or `psycopg2-binary` in the :ref:`list of dependencies ` when installing Patroni with pip. -.. code-block:: shell - pip install psycopg2-binary - -3. Install psycopg2 from source - -.. code-block:: shell - - pip install psycopg2>=2.5.4 - -4. Use psycopg 3.0 instead of psycopg2 - -.. code-block:: shell - - pip install psycopg[binary]>=3.0.0 +.. _extras: General installation for pip ---------------------------- @@ -73,12 +60,20 @@ raft `pysyncobj` module in order to use python Raft implementation as DCS aws `boto3` in order to use AWS callbacks +all + all of the above (except psycopg family) +psycopg + `psycopg[binary]>=3.0.0` module +psycopg2 + `psycopg2>=2.5.4` module +psycopg2-binary + `psycopg2-binary` module -For example, the command in order to install Patroni together with dependencies for Etcd as a DCS and AWS callbacks is: +For example, the command in order to install Patroni together with psycopg3, dependencies for Etcd as a DCS, and AWS callbacks is: .. code-block:: shell - pip install patroni[etcd,aws] + pip install patroni[psycopg3,etcd3,aws] Note that external tools to call in the replica creation or custom bootstrap scripts (i.e. WAL-E) should be installed independently of Patroni. diff --git a/patroni/__init__.py b/patroni/__init__.py index 7f7035c2..7e67e299 100644 --- a/patroni/__init__.py +++ b/patroni/__init__.py @@ -3,23 +3,14 @@ :var PATRONI_ENV_PREFIX: prefix for Patroni related configuration environment variables. :var KUBERNETES_ENV_PREFIX: prefix for Kubernetes related configuration environment variables. :var MIN_PSYCOPG2: minimum version of :mod:`psycopg2` required by Patroni to work. +:var MIN_PSYCOPG3: minimum version of :mod:`psycopg` required by Patroni to work. """ - -import sys - -from typing import Any, Callable, Iterator, Tuple +from typing import Iterator, Tuple PATRONI_ENV_PREFIX = 'PATRONI_' KUBERNETES_ENV_PREFIX = 'KUBERNETES_' MIN_PSYCOPG2 = (2, 5, 4) - - -def fatal(string: str, *args: Any) -> None: - """Write a fatal message to stderr and exit with code ``1``. - - :param string: message to be written before exiting. - """ - sys.exit('FATAL: ' + string.format(*args)) +MIN_PSYCOPG3 = (3, 0, 0) def parse_version(version: str) -> Tuple[int, ...]: @@ -28,25 +19,25 @@ def parse_version(version: str) -> Tuple[int, ...]: .. note:: Designed for easy comparison of software versions in Python. - :param version: human-readable software version, e.g. ``2.5.4``. + :param version: human-readable software version, e.g. ``2.5.4.dev1 (dt dec pq3 ext lo64)``. :returns: tuple of *version* parts, each part as an integer. :Example: - >>> parse_version('2.5.4') + >>> parse_version('2.5.4.dev1 (dt dec pq3 ext lo64)') (2, 5, 4) """ def _parse_version(version: str) -> Iterator[int]: """Yield each part of a human-readable version string as an integer. - :param version: human-readable software version, e.g. ``2.5.4``. + :param version: human-readable software version, e.g. ``2.5.4.dev1``. :yields: each part of *version* as an integer. :Example: - >>> tuple(_parse_version('2.5.4')) + >>> tuple(_parse_version('2.5.4.dev1')) (2, 5, 4) """ for e in version.split('.'): @@ -55,40 +46,3 @@ def parse_version(version: str) -> Tuple[int, ...]: except ValueError: break return tuple(_parse_version(version.split(' ')[0])) - - -def check_psycopg(_min_psycopg2: Tuple[int, ...] = MIN_PSYCOPG2, - _parse_version: Callable[[str], Tuple[int, ...]] = parse_version) -> None: - """Ensure at least one among :mod:`psycopg2` or :mod:`psycopg` libraries are available in the environment. - - .. note:: - We pass ``MIN_PSYCOPG2`` and :func:`parse_version` as arguments to simplify usage of :func:`check_psycopg` from - the ``setup.py``. - - .. note:: - Patroni chooses :mod:`psycopg2` over :mod:`psycopg`, if possible. - - If nothing meeting the requirements is found, then exit with a fatal message. - - :param _min_psycopg2: minimum required version in case :mod:`psycopg2` is chosen. - :param _parse_version: function used to parse :mod:`psycopg2`/:mod:`psycopg` version into a comparable object. - """ - min_psycopg2_str = '.'.join(map(str, _min_psycopg2)) - - # try psycopg2 - try: - from psycopg2 import __version__ - if _parse_version(__version__) >= _min_psycopg2: - return - version_str = __version__.split(' ')[0] - except ImportError: - version_str = None - - # try psycopg3 - try: - from psycopg import __version__ - except ImportError: - error = 'Patroni requires psycopg2>={0}, psycopg2-binary, or psycopg>=3.0'.format(min_psycopg2_str) - if version_str is not None: - error += ', but only psycopg2=={0} is available'.format(version_str) - fatal(error) diff --git a/patroni/__main__.py b/patroni/__main__.py index 7d56172b..02ba56da 100644 --- a/patroni/__main__.py +++ b/patroni/__main__.py @@ -10,8 +10,9 @@ import sys import time from argparse import Namespace -from typing import Any, Dict, Optional, TYPE_CHECKING +from typing import Any, Dict, List, Optional, TYPE_CHECKING +from patroni import MIN_PSYCOPG2, MIN_PSYCOPG3, parse_version from patroni.daemon import AbstractPatroniDaemon, abstract_main, get_base_arg_parser from patroni.tags import Tags @@ -286,6 +287,45 @@ def process_arguments() -> Namespace: return args +def check_psycopg() -> None: + """Ensure at least one among :mod:`psycopg2` or :mod:`psycopg` libraries are available in the environment. + + .. note:: + Patroni chooses :mod:`psycopg2` over :mod:`psycopg`, if possible. + + If nothing meeting the requirements is found, then exit with a fatal message. + """ + min_psycopg2_str = '.'.join(map(str, MIN_PSYCOPG2)) + min_psycopg3_str = '.'.join(map(str, MIN_PSYCOPG3)) + + available_versions: List[str] = [] + + # try psycopg2 + try: + from psycopg2 import __version__ + if parse_version(__version__) >= MIN_PSYCOPG2: + return + available_versions.append('psycopg2=={0}'.format(__version__.split(' ')[0])) + except ImportError: + logger.debug('psycopg2 module is not available') + + # try psycopg3 + try: + from psycopg import __version__ + if parse_version(__version__) >= MIN_PSYCOPG3: + return + available_versions.append('psycopg=={0}'.format(__version__.split(' ')[0])) + except ImportError: + logger.debug('psycopg module is not available') + + error = f'FATAL: Patroni requires psycopg2>={min_psycopg2_str}, psycopg2-binary, or psycopg>={min_psycopg3_str}' + if available_versions: + error += ', but only {0} {1} available'.format( + ' and '.join(available_versions), + 'is' if len(available_versions) == 1 else 'are') + sys.exit(error) + + def main() -> None: """Main entrypoint of :mod:`patroni.__main__`. @@ -297,12 +337,10 @@ def main() -> None: ``patroni`` daemon as another process. In that case relevant signals received by the main process and forwarded to ``patroni`` daemon process. """ - from patroni import check_psycopg + check_psycopg() args = process_arguments() - check_psycopg() - if os.getpid() != 1: return patroni_main(args.configfile) diff --git a/setup.py b/setup.py index d61eab6e..4c1c25b3 100644 --- a/setup.py +++ b/setup.py @@ -26,7 +26,6 @@ KEYWORDS = 'etcd governor patroni postgresql postgres ha haproxy confd' +\ EXTRAS_REQUIRE = {'aws': ['boto3'], 'etcd': ['python-etcd'], 'etcd3': ['python-etcd'], 'consul': ['python-consul'], 'exhibitor': ['kazoo'], 'zookeeper': ['kazoo'], 'kubernetes': [], 'raft': ['pysyncobj', 'cryptography']} -COVERAGE_XML = True # Add here all kinds of additional classifiers as defined under # https://pypi.python.org/pypi?%3Aaction=list_classifiers @@ -120,14 +119,21 @@ def read(fname): return fd.read() -def setup_package(version): +def get_versions(): + old_modules = sys.modules.copy() + try: + from patroni import MIN_PSYCOPG2, MIN_PSYCOPG3 + from patroni.version import __version__ + return __version__, MIN_PSYCOPG2, MIN_PSYCOPG3 + finally: + sys.modules.clear() + sys.modules.update(old_modules) + + +def main(): logging.basicConfig(format='%(message)s', level=os.getenv('LOGLEVEL', logging.WARNING)) - # Assemble additional setup commands - cmdclass = {'test': PyTest, 'flake8': Flake8} - install_requires = [] - for r in read('requirements.txt').split('\n'): r = r.strip() if r == '': @@ -139,15 +145,22 @@ def setup_package(version): deps[i] = r EXTRAS_REQUIRE[e] = deps extra = True - break - if extra: - break if not extra: install_requires.append(r) + # Just for convenience, if someone wants to install dependencies for all extras + EXTRAS_REQUIRE['all'] = list({e for extras in EXTRAS_REQUIRE.values() for e in extras}) + + patroni_version, min_psycopg2, min_psycopg3 = get_versions() + + # Make it possible to specify psycopg dependency as extra + for name, version in {'psycopg[binary]': min_psycopg3, 'psycopg2': min_psycopg2, 'psycopg2-binary': None}.items(): + EXTRAS_REQUIRE[name] = [name + ('>=' + '.'.join(map(str, version)) if version else '')] + EXTRAS_REQUIRE['psycopg3'] = EXTRAS_REQUIRE.pop('psycopg[binary]') + setup( name=NAME, - version=version, + version=patroni_version, url=URL, author=AUTHOR, author_email=AUTHOR_EMAIL, @@ -163,20 +176,10 @@ def setup_package(version): ]}, install_requires=install_requires, extras_require=EXTRAS_REQUIRE, - cmdclass=cmdclass, + cmdclass={'test': PyTest, 'flake8': Flake8}, entry_points={'console_scripts': CONSOLE_SCRIPTS}, ) if __name__ == '__main__': - old_modules = sys.modules.copy() - try: - from patroni import check_psycopg - from patroni.version import __version__ - finally: - sys.modules.clear() - sys.modules.update(old_modules) - - check_psycopg() - - setup_package(__version__) + main() diff --git a/tests/test_patroni.py b/tests/test_patroni.py index df59677d..137157e5 100644 --- a/tests/test_patroni.py +++ b/tests/test_patroni.py @@ -15,8 +15,7 @@ from patroni.dcs.etcd import AbstractEtcdClientWithFailover from patroni.exceptions import DCSError from patroni.postgresql import Postgresql from patroni.postgresql.config import ConfigHandler -from patroni import check_psycopg -from patroni.__main__ import Patroni, main as _main +from patroni.__main__ import check_psycopg, Patroni, main as _main from threading import Thread from . import psycopg_connect, SleepException @@ -25,10 +24,16 @@ from .test_postgresql import MockPostmaster def mock_import(*args, **kwargs): - if args[0] == 'psycopg': + ret = Mock() + ret.__version__ = '2.5.3.dev1 a b c' if args[0] == 'psycopg2' else '3.1.0' + return ret + + +def mock_import2(*args, **kwargs): + if args[0] == 'psycopg2': raise ImportError ret = Mock() - ret.__version__ = '2.5.3.dev1 a b c' + ret.__version__ = '0.1.2' return ret @@ -205,6 +210,8 @@ class TestPatroni(unittest.TestCase): with patch('builtins.__import__', Mock(side_effect=ImportError)): self.assertRaises(SystemExit, check_psycopg) with patch('builtins.__import__', mock_import): + self.assertIsNone(check_psycopg()) + with patch('builtins.__import__', mock_import2): self.assertRaises(SystemExit, check_psycopg) def test_ensure_unique_name(self): From 260ab36f2ea4dbe783a251a989f1b90a00b3274f Mon Sep 17 00:00:00 2001 From: zhjwpku Date: Wed, 18 Oct 2023 01:53:19 +0800 Subject: [PATCH 13/19] mock getaddrinfo in case test failure (#2918) Close #2915 --- tests/test_validator.py | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/tests/test_validator.py b/tests/test_validator.py index 24379c6b..b02383ea 100644 --- a/tests/test_validator.py +++ b/tests/test_validator.py @@ -134,6 +134,23 @@ def connect_side_effect(host_port): raise socket.gaierror() +def mock_getaddrinfo(host, port, *args): + if port is None or port == "": + port = 0 + port = int(port) + if port not in range(0, 65536): + raise socket.gaierror() + + if host == "127.0.0.1" or host == "" or host is None: + return [(socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP, '', ('127.0.0.1', port))] + elif host == "127.0.0.2": + return [(socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP, '', ('127.0.0.2', port))] + elif host == "::1": + return [(socket.AF_INET6, socket.SOCK_STREAM, socket.IPPROTO_TCP, '', ('::1', port, 0, 0))] + else: + raise socket.gaierror() + + def parse_output(output): result = [] for s in output.split("\n"): @@ -145,6 +162,7 @@ def parse_output(output): @patch('socket.socket.connect_ex', Mock(side_effect=connect_side_effect)) +@patch('socket.getaddrinfo', Mock(side_effect=mock_getaddrinfo)) @patch('os.path.exists', Mock(side_effect=exists_side_effect)) @patch('os.path.isdir', Mock(side_effect=isdir_side_effect)) @patch('os.path.isfile', Mock(side_effect=isfile_side_effect)) From cb5f34b72133960d72b748f6076bfc55e9db98d5 Mon Sep 17 00:00:00 2001 From: zhjwpku Date: Mon, 23 Oct 2023 14:17:53 +0800 Subject: [PATCH 14/19] add some guide to run tests in different scopes (#2921) Introduce ways to run tests in different scopes which should be helpful for beginners. --- docs/contributing_guidelines.rst | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/docs/contributing_guidelines.rst b/docs/contributing_guidelines.rst index bfa1f4c5..805f2fa8 100644 --- a/docs/contributing_guidelines.rst +++ b/docs/contributing_guidelines.rst @@ -43,6 +43,13 @@ After you have all dependencies installed, you can run the various test suites: # Run the pytest suite in tests/: python setup.py test + # Moreover, you may want to run tests in different scopes for debugging purposes, + # the -s option include print output during test execution. + # Tests in pytest typically follow the pattern: FILEPATH::CLASSNAME::TESTNAME. + pytest -s tests/test_api.py + pytest -s tests/test_api.py::TestRestApiHandler + pytest -s tests/test_api.py::TestRestApiHandler::test_do_GET + # Run the behave (https://behave.readthedocs.io/en/latest/) test suite in features/; # modify DCS as desired (raft has no dependencies so is the easiest to start with): DCS=raft python -m behave From c5fffb3c976a9ce0e2529ab7d12fa705d5ec43ce Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Mon, 23 Oct 2023 08:24:28 +0200 Subject: [PATCH 15/19] Further work on permanent physical slots (#2891) - Fixed issues with has_permanent_slots() method. It didn't took into account the case of permanent physical slots for members, falsely concluding that there are no permanent slots. - Write to the status key only LSNs for permanent slots (not just for slots that exist on the primary). - Include pg_current_wal_flush_lsn() to slots feedback, so that slots on standby nodes could be advanced - Improved behave tests: - Verify that permanent slots are properly created on standby nodes - Verify that permanent slots are properly advanced, including DCS failsafe mode - Verify that only permanent slots are written to the `/status` --- docs/dynamic_configuration.rst | 7 ++- features/dcs_failsafe_mode.feature | 44 ++++++++++++-- features/environment.py | 5 +- features/ignored_slots.feature | 2 +- features/permanent_slots.feature | 55 ++++++++++++++--- features/standby_cluster.feature | 2 +- features/steps/basic_replication.py | 2 +- features/steps/slots.py | 18 +++++- patroni/dcs/__init__.py | 93 +++++++++++++++++++++-------- patroni/ha.py | 53 ++++++++++++++-- patroni/postgresql/__init__.py | 16 +++-- patroni/postgresql/slots.py | 15 ++--- patroni/utils.py | 2 +- tests/test_ha.py | 2 + tests/test_slots.py | 1 + 15 files changed, 250 insertions(+), 67 deletions(-) diff --git a/docs/dynamic_configuration.rst b/docs/dynamic_configuration.rst index a16a2a40..f38a8b39 100644 --- a/docs/dynamic_configuration.rst +++ b/docs/dynamic_configuration.rst @@ -57,7 +57,7 @@ In order to change the dynamic configuration you can use either :ref:`patronictl - **slots**: define permanent replication slots. These slots will be preserved during switchover/failover. Permanent slots that don't exist will be created by Patroni. With PostgreSQL 11 onwards permanent physical slots are created on all nodes and their position is advanced every **loop_wait** seconds. For PostgreSQL versions older than 11 permanent physical replication slots are maintained only on 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 replication slots requires **postgresql.use_slots** to be set to ``true``. If there are permanent logical replication slots defined Patroni will 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 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. + - **my\_slot\_name**: the name of the permanent replication slot. If the permanent slot name matches with the name of the current node it will not be created on this node. If you add a permanent physical replication slot which name matches the name of a Patroni member, Patroni will ensure that the slot that was created is not removed even if the corresponding 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 you want replication slots used by members to persist during temporary failures or 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, when the slot is no longer required, 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. @@ -103,3 +103,8 @@ Note: if cluster topology is static (fixed number of nodes that never change the node_name3: type: physical ... + + +.. warning:: + Permanent replication slots are synchronized only from the ``primary``/``standby_leader`` to replica nodes. That means, applications are supposed to be using them only from the leader node. Using them on replica nodes will cause indefinite growth of ``pg_wal`` on all other nodes in the cluster. + An exception to that rule are permanent physical slots that match the Patroni member names, if you happen to configure any. Those will be synchronized among all nodes as they are used for replication among them. diff --git a/features/dcs_failsafe_mode.feature b/features/dcs_failsafe_mode.feature index 8345d8ea..0489db39 100644 --- a/features/dcs_failsafe_mode.feature +++ b/features/dcs_failsafe_mode.feature @@ -11,7 +11,7 @@ Feature: dcs failsafe mode When I issue a GET request to http://127.0.0.1:8008/failsafe Then I receive a response code 200 And I receive a response postgres0 http://127.0.0.1:8008/patroni - When I issue a PATCH request to http://127.0.0.1:8008/config with {"postgresql": {"parameters": {"wal_level": "logical"}}} + When I issue a PATCH request to http://127.0.0.1:8008/config with {"postgresql": {"parameters": {"wal_level": "logical"}},"slots":{"dcs_slot_1": null,"postgres0":null}} Then I receive a response code 200 When I issue a PATCH request to http://127.0.0.1:8008/config with {"slots": {"dcs_slot_0": {"type": "logical", "database": "postgres", "plugin": "test_decoding"}}} Then I receive a response code 200 @@ -44,14 +44,18 @@ Feature: dcs failsafe mode @dcs-failsafe @slot-advance Scenario: check leader and replica are functioning while DCS is down - Given logical slot dcs_slot_0 is in sync between postgres0 and postgres1 after 10 seconds + Given I get all changes from physical slot dcs_slot_1 on postgres0 + Then physical slot dcs_slot_1 is in sync between postgres0 and postgres1 after 10 seconds + And logical slot dcs_slot_0 is in sync between postgres0 and postgres1 after 10 seconds And DCS is down Then Response on GET http://127.0.0.1:8008/primary contains failsafe_mode_is_active after 12 seconds Then postgres0 role is the primary after 10 seconds And postgres1 role is the replica after 2 seconds And replication works from postgres0 to postgres1 after 10 seconds - And I get all changes from logical slot dcs_slot_0 on postgres0 - And logical slot dcs_slot_0 is in sync between postgres0 and postgres1 after 20 seconds + When I get all changes from logical slot dcs_slot_0 on postgres0 + And I get all changes from physical slot dcs_slot_1 on postgres0 + Then logical slot dcs_slot_0 is in sync between postgres0 and postgres1 after 20 seconds + And physical slot dcs_slot_1 is in sync between postgres0 and postgres1 after 10 seconds @dcs-failsafe Scenario: check primary is demoted when one replica is shut down and DCS is down @@ -70,15 +74,43 @@ Feature: dcs failsafe mode And postgres1 role is the primary after 25 seconds @dcs-failsafe - Scenario: check three-node cluster is functioning while DCS is down + Scenario: scale to three-node cluster Given I start postgres0 And I start postgres2 Then "members/postgres2" key in DCS has state=running after 10 seconds And "members/postgres0" key in DCS has state=running after 20 seconds And Response on GET http://127.0.0.1:8008/failsafe contains postgres2 after 10 seconds And replication works from postgres1 to postgres0 after 10 seconds + And replication works from postgres1 to postgres2 after 10 seconds + + @dcs-failsafe + @slot-advance + Scenario: make sure permanent slots exist on replicas + Given I issue a PATCH request to http://127.0.0.1:8009/config with {"slots":{"dcs_slot_0":null,"dcs_slot_2":{"type":"logical","database":"postgres","plugin":"test_decoding"}}} + Then logical slot dcs_slot_2 is in sync between postgres1 and postgres0 after 20 seconds + And logical slot dcs_slot_2 is in sync between postgres1 and postgres2 after 20 seconds + When I get all changes from physical slot dcs_slot_1 on postgres1 + Then physical slot dcs_slot_1 is in sync between postgres1 and postgres0 after 10 seconds + And physical slot dcs_slot_1 is in sync between postgres1 and postgres2 after 10 seconds + And physical slot postgres0 is in sync between postgres1 and postgres2 after 10 seconds + + @dcs-failsafe + Scenario: check three-node cluster is functioning while DCS is down Given DCS is down - Then Response on GET http://127.0.0.1:8008/primary contains failsafe_mode_is_active after 12 seconds + Then Response on GET http://127.0.0.1:8009/primary contains failsafe_mode_is_active after 12 seconds Then postgres1 role is the primary after 10 seconds And postgres0 role is the replica after 2 seconds And postgres2 role is the replica after 2 seconds + + @dcs-failsafe + @slot-advance + Scenario: check that permanent slots are in sync between nodes while DCS is down + Given replication works from postgres1 to postgres0 after 10 seconds + And replication works from postgres1 to postgres2 after 10 seconds + When I get all changes from logical slot dcs_slot_2 on postgres1 + And I get all changes from physical slot dcs_slot_1 on postgres1 + Then logical slot dcs_slot_2 is in sync between postgres1 and postgres0 after 20 seconds + And logical slot dcs_slot_2 is in sync between postgres1 and postgres2 after 20 seconds + And physical slot dcs_slot_1 is in sync between postgres1 and postgres0 after 10 seconds + And physical slot dcs_slot_1 is in sync between postgres1 and postgres2 after 10 seconds + And physical slot postgres0 is in sync between postgres1 and postgres2 after 10 seconds diff --git a/features/environment.py b/features/environment.py index 3e0ad3b1..1c3e654b 100644 --- a/features/environment.py +++ b/features/environment.py @@ -654,9 +654,10 @@ class KubernetesController(AbstractExternalDcsController): try: if group is not None: scope = '{0}-{1}'.format(scope, group) - ep = scope + {'leader': '', 'history': '-config', 'initialize': '-config'}.get(key, '-' + key) + rkey = 'leader' if key in ('status', 'failsafe') else key + ep = scope + {'leader': '', 'history': '-config', 'initialize': '-config'}.get(rkey, '-' + rkey) e = self._api.read_namespaced_endpoints(ep, self._namespace) - if key != 'sync': + if key not in ('sync', 'status', 'failsafe'): return e.metadata.annotations[key] else: return json.dumps(e.metadata.annotations) diff --git a/features/ignored_slots.feature b/features/ignored_slots.feature index e0c53ea3..4e83570d 100644 --- a/features/ignored_slots.feature +++ b/features/ignored_slots.feature @@ -50,7 +50,7 @@ Feature: ignored slots And postgres1 has a logical replication slot named unmanaged_slot_1 with the test_decoding plugin after 2 seconds And postgres1 has a logical replication slot named unmanaged_slot_2 with the test_decoding plugin after 2 seconds And postgres1 has a logical replication slot named unmanaged_slot_3 with the test_decoding plugin after 2 seconds - And postgres1 does not have a logical replication slot named dummy_slot + And postgres1 does not have a replication slot named dummy_slot # 3. After a failover the server (now a primary) still has the slot. When I shut down postgres0 diff --git a/features/permanent_slots.feature b/features/permanent_slots.feature index 656e6ade..2928e829 100644 --- a/features/permanent_slots.feature +++ b/features/permanent_slots.feature @@ -3,32 +3,73 @@ Feature: permanent slots Given I start postgres0 Then postgres0 is a leader after 10 seconds And there is a non empty initialize key in DCS after 15 seconds - When I issue a PATCH request to http://127.0.0.1:8008/config with {"slots": {"test_physical": {"type": "physical"}}, "postgresql": {"parameters": {"wal_level": "logical"}}} + When I issue a PATCH request to http://127.0.0.1:8008/config with {"slots":{"test_physical":0,"postgres0":0,"postgres1":0,"postgres3":0},"postgresql":{"parameters":{"wal_level":"logical"}}} Then I receive a response code 200 And Response on GET http://127.0.0.1:8008/config contains slots after 10 seconds + When I start postgres1 + And I start postgres2 + And I configure and start postgres3 with a tag replicatefrom postgres2 Then postgres0 has a physical replication slot named test_physical after 10 seconds - And I start postgres1 + And postgres0 has a physical replication slot named postgres1 after 10 seconds + And postgres0 has a physical replication slot named postgres2 after 10 seconds + And postgres2 has a physical replication slot named postgres3 after 10 seconds @slot-advance Scenario: check that logical permanent slots are created Given I run patronictl.py restart batman postgres0 --force - And I issue a PATCH request to http://127.0.0.1:8008/config with {"slots": {"test_logical": {"type": "logical", "database": "postgres", "plugin": "test_decoding"}}} + And I issue a PATCH request to http://127.0.0.1:8008/config with {"slots":{"test_logical":{"type":"logical","database":"postgres","plugin":"test_decoding"}}} Then postgres0 has a logical replication slot named test_logical with the test_decoding plugin after 10 seconds @slot-advance - Scenario: check that permanent slots are created on the replica + Scenario: check that permanent slots are created on replicas Given postgres1 has a logical replication slot named test_logical with the test_decoding plugin after 10 seconds Then Logical slot test_logical is in sync between postgres0 and postgres1 after 10 seconds + And Logical slot test_logical is in sync between postgres0 and postgres2 after 10 seconds + And Logical slot test_logical is in sync between postgres0 and postgres3 after 10 seconds And postgres1 has a physical replication slot named test_physical after 2 seconds + And postgres2 has a physical replication slot named test_physical after 2 seconds + And postgres3 has a physical replication slot named test_physical after 2 seconds @slot-advance - Scenario: check that permanent slots are advanced on the replica + Scenario: check permanent physical slots that match with member names + Given postgres0 has a physical replication slot named postgres3 after 2 seconds + And postgres1 has a physical replication slot named postgres0 after 2 seconds + And postgres1 has a physical replication slot named postgres3 after 2 seconds + And postgres2 has a physical replication slot named postgres0 after 2 seconds + And postgres2 has a physical replication slot named postgres3 after 2 seconds + And postgres2 has a physical replication slot named postgres1 after 2 seconds + And postgres1 does not have a replication slot named postgres2 + And postgres3 does not have a replication slot named postgres2 + + @slot-advance + Scenario: check that permanent slots are advanced on replicas Given I add the table replicate_me to postgres0 - And I get all changes from physical slot test_physical on postgres0 When I get all changes from logical slot test_logical on postgres0 + And I get all changes from physical slot test_physical on postgres0 Then Logical slot test_logical is in sync between postgres0 and postgres1 after 10 seconds And Physical slot test_physical is in sync between postgres0 and postgres1 after 10 seconds + And Logical slot test_logical is in sync between postgres0 and postgres2 after 10 seconds + And Physical slot test_physical is in sync between postgres0 and postgres2 after 10 seconds + And Logical slot test_logical is in sync between postgres0 and postgres3 after 10 seconds + And Physical slot test_physical is in sync between postgres0 and postgres3 after 10 seconds + And Physical slot postgres1 is in sync between postgres0 and postgres2 after 10 seconds + And Physical slot postgres3 is in sync between postgres2 and postgres0 after 20 seconds + And Physical slot postgres3 is in sync between postgres2 and postgres1 after 10 seconds + And postgres1 does not have a replication slot named postgres2 + And postgres3 does not have a replication slot named postgres2 + + @slot-advance + Scenario: check that only permanent slots are written to the /status key + Given "status" key in DCS has test_physical in slots + And "status" key in DCS has postgres0 in slots + And "status" key in DCS has postgres1 in slots + And "status" key in DCS does not have postgres2 in slots + And "status" key in DCS has postgres3 in slots Scenario: check permanent physical replication slot after failover - Given I shut down postgres0 + Given I shut down postgres3 + And I shut down postgres2 + And I shut down postgres0 Then postgres1 has a physical replication slot named test_physical after 10 seconds + And postgres1 has a physical replication slot named postgres0 after 10 seconds + And postgres1 has a physical replication slot named postgres3 after 10 seconds diff --git a/features/standby_cluster.feature b/features/standby_cluster.feature index a9f00c01..97c27203 100644 --- a/features/standby_cluster.feature +++ b/features/standby_cluster.feature @@ -51,7 +51,7 @@ Feature: standby cluster When I issue a GET request to http://127.0.0.1:8010/patroni Then I receive a response code 200 And I receive a response replication_state streaming - And postgres1 does not have a logical replication slot named test_logical + And postgres1 does not have a replication slot named test_logical Scenario: check switchover Given I run patronictl.py switchover batman1 --force diff --git a/features/steps/basic_replication.py b/features/steps/basic_replication.py index 5977eb61..7a687e28 100644 --- a/features/steps/basic_replication.py +++ b/features/steps/basic_replication.py @@ -110,7 +110,7 @@ def replication_works(context, primary, replica, time_limit): context.execute_steps(u""" When I add the table test_{0} to {1} Then table test_{0} is present on {2} after {3} seconds - """.format(int(time()), primary, replica, time_limit)) + """.format(str(time()).replace('.', '_').replace(',', '_'), primary, replica, time_limit)) @then('there is a "{message}" {level:w} in the {node} patroni log') diff --git a/features/steps/slots.py b/features/steps/slots.py index f4a3cfa5..182aa87c 100644 --- a/features/steps/slots.py +++ b/features/steps/slots.py @@ -1,3 +1,4 @@ +import json import time from behave import step, then @@ -36,8 +37,9 @@ def has_logical_replication_slot(context, pg_name, slot_name, plugin, time_limit assert False, f"Error looking for slot {slot_name} on {pg_name} with plugin {plugin}" -@then('{pg_name:w} does not have a logical replication slot named {slot_name}') -def does_not_have_logical_replication_slot(context, pg_name, slot_name): +@step('{pg_name:w} does not have a replication slot named {slot_name:w}') +@then('{pg_name:w} does not have a replication slot named {slot_name:w}') +def does_not_have_replication_slot(context, pg_name, slot_name): try: row = context.pctl.query(pg_name, ("SELECT 1 FROM pg_replication_slots" " WHERE slot_name = '{0}'").format(slot_name)).fetchone() @@ -89,3 +91,15 @@ def has_physical_replication_slot(context, pg_name, slot_name, time_limit): pass time.sleep(1) assert False, f"Physical slot {slot_name} doesn't exist after {time_limit} seconds" + + +@step('"{name}" key in DCS has {subkey:w} in {key:w}') +def dcs_key_contains(context, name, subkey, key): + response = json.loads(context.dcs_ctl.query(name)) + assert key in response and subkey in response[key], f"{name} key in DCS doesn't have {subkey} in {key}" + + +@step('"{name}" key in DCS does not have {subkey:w} in {key:w}') +def dcs_key_does_not_contain(context, name, subkey, key): + response = json.loads(context.dcs_ctl.query(name)) + assert key not in response or subkey not in response[key], f"{name} key in DCS has {subkey} in {key}" diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index 1a516cca..bc74bbe0 100644 --- a/patroni/dcs/__init__.py +++ b/patroni/dcs/__init__.py @@ -24,6 +24,7 @@ import dateutil.parser from ..exceptions import PatroniFatalException from ..utils import deep_compare, uri from ..tags import Tags +from ..utils import parse_int if TYPE_CHECKING: # pragma: no cover from ..config import Config @@ -354,7 +355,7 @@ class Member(Tags, NamedTuple('Member', @property def lsn(self) -> Optional[int]: """Current LSN (receive/flush/replay).""" - return self.data.get('xlog_location') + return parse_int(self.data.get('xlog_location')) class RemoteMember(Member): @@ -974,29 +975,41 @@ class Cluster(NamedTuple('Cluster', def is_physical_slot(value: Union[Any, Dict[str, Any]]) -> bool: """Check whether provided configuration is for permanent physical replication slot. - :returns: ``True`` if this is a physical replication slot, otherwise ``False``. + :param value: configuration of the permanent replication slot. + + :returns: ``True`` if *value* is a physical replication slot, otherwise ``False``. """ return not value or isinstance(value, dict) and value.get('type', 'physical') == 'physical' + @staticmethod + def is_logical_slot(value: Union[Any, Dict[str, Any]]) -> bool: + """Check whether provided configuration is for permanent logical replication slot. + + :param value: configuration of the permanent replication slot. + + :returns: ``True`` if *value* is a logical replication slot, otherwise ``False``. + """ + return isinstance(value, dict) \ + and value.get('type', 'logical') == 'logical' \ + and bool(value.get('database') and value.get('plugin')) + @property def __permanent_slots(self) -> Dict[str, Union[Dict[str, Any], Any]]: """Dictionary of permanent replication slots with their known LSN.""" - leader = self.leader and self.leader.member - leader_name = slot_name_from_member_name(leader.name) if leader and leader.lsn else None - - slots = self.slots or {} ret: Dict[str, Union[Dict[str, Any], Any]] = deepcopy(self.config.permanent_slots if self.config else {}) + members: Dict[str, int] = {slot_name_from_member_name(m.name): m.lsn or 0 for m in self.members} + slots: Dict[str, int] = {k: parse_int(v) or 0 for k, v in (self.slots or {}).items()} for name, value in list(ret.items()): if not value: value = ret[name] = {} if isinstance(value, dict): - if name in slots: - # If primary reported flush LSN for permanent slots we want to enrich our structure with it - value['lsn'] = slots[name] - elif self.is_physical_slot(value) and name == leader_name and leader and leader.lsn: - # there is no slot on the leader for itself, use `lsn` from the member key. - value['lsn'] = leader.lsn + # for permanent physical slots we want to get MAX LSN from the `Cluster.slots` and from the + # member with the matching name. It is necessary because we may have the replication slot on + # the primary that is streaming from the other standby node using the `replicatefrom` tag. + lsn = max(members.get(name, 0) if self.is_physical_slot(value) else 0, slots.get(name, 0)) + if lsn: + value['lsn'] = lsn else: # Don't let anyone set 'lsn' in the global configuration :) value.pop('lsn', None) @@ -1010,8 +1023,7 @@ class Cluster(NamedTuple('Cluster', @property def __permanent_logical_slots(self) -> Dict[str, Any]: """Dictionary of permanent ``logical`` replication slots.""" - return {name: value for name, value in self.__permanent_slots.items() if isinstance(value, dict) - and value.get('type', 'logical') == 'logical' and value.get('database') and value.get('plugin')} + return {name: value for name, value in self.__permanent_slots.items() if self.is_logical_slot(value)} @property def use_slots(self) -> bool: @@ -1037,7 +1049,9 @@ class Cluster(NamedTuple('Cluster', :returns: final dictionary of slot names, after merging with permanent slots and performing sanity checks. """ slots: Dict[str, Dict[str, str]] = self._get_members_slots(my_name, role) - permanent_slots: Dict[str, Any] = self._get_permanent_slots(is_standby_cluster, role, nofailover, major_version) + permanent_slots: Dict[str, Any] = self._get_permanent_slots(is_standby_cluster=is_standby_cluster, + role=role, nofailover=nofailover, + major_version=major_version) disabled_permanent_logical_slots: List[str] = self._merge_permanent_slots( slots, permanent_slots, my_name, major_version) @@ -1048,8 +1062,7 @@ class Cluster(NamedTuple('Cluster', return slots - @staticmethod - def _merge_permanent_slots(slots: Dict[str, Dict[str, str]], permanent_slots: Dict[str, Any], my_name: str, + def _merge_permanent_slots(self, 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*. @@ -1084,7 +1097,7 @@ class Cluster(NamedTuple('Cluster', slots[name] = value continue - if value['type'] == 'logical' and value.get('database') and value.get('plugin'): + if self.is_logical_slot(value): if major_version < SLOT_ADVANCE_AVAILABLE_VERSION: disabled_permanent_logical_slots.append(name) elif name in slots: @@ -1097,7 +1110,7 @@ class Cluster(NamedTuple('Cluster', logger.error("Bad value for slot '%s' in permanent_slots: %s", name, permanent_slots[name]) return disabled_permanent_logical_slots - def _get_permanent_slots(self, is_standby_cluster: bool, role: str, + def _get_permanent_slots(self, *, is_standby_cluster: bool, role: str, nofailover: bool, major_version: int) -> Dict[str, Any]: """Get configured permanent replication slots. @@ -1171,20 +1184,50 @@ class Cluster(NamedTuple('Cluster', for k, v in slot_conflicts.items() if len(v) > 1)) return slots - def has_permanent_slots(self, my_name: str, nofailover: bool = False) -> bool: + def has_permanent_slots(self, my_name: str, *, is_standby_cluster: bool = False, nofailover: bool = False, + major_version: int = SLOT_ADVANCE_AVAILABLE_VERSION) -> bool: """Check if the given member node has permanent replication slots configured. :param my_name: name of the member node to check. + :param is_standby_cluster: ``True`` if it is known that this is a standby cluster. We pass the value from + the outside because we want to protect from the ``/config`` key removal. :param nofailover: ``True`` if this node is tagged to not be a failover candidate. + :param major_version: postgresql major version. :returns: ``True`` if there are permanent replication slots configured, otherwise ``False``. """ - members_slots: Dict[str, Dict[str, str]] = self._get_members_slots(my_name, 'replica') - permanent_slots: Dict[str, Any] = self._get_permanent_slots(nofailover, 'replica', False, - SLOT_ADVANCE_AVAILABLE_VERSION) + role = 'replica' + members_slots: Dict[str, Dict[str, str]] = self._get_members_slots(my_name, role) + permanent_slots: Dict[str, Any] = self._get_permanent_slots(is_standby_cluster=is_standby_cluster, + role=role, nofailover=nofailover, + major_version=major_version) slots = deepcopy(members_slots) - self._merge_permanent_slots(slots, permanent_slots, my_name, SLOT_ADVANCE_AVAILABLE_VERSION) - return len(slots) > len(members_slots) + self._merge_permanent_slots(slots, permanent_slots, my_name, major_version) + return len(slots) > len(members_slots) or any(self.is_physical_slot(v) for v in permanent_slots.values()) + + def filter_permanent_slots(self, slots: Dict[str, int], is_standby_cluster: bool, + major_version: int) -> Dict[str, int]: + """Filter out all non-permanent slots from provided *slots* dict. + + :param slots: slot names with LSN values + :param is_standby_cluster: ``True`` if it is known that this is a standby cluster. We pass the value from + the outside because we want to protect from the ``/config`` key removal. + :param major_version: postgresql major version. + + :returns: a :class:`dict` object that contains only slots that are known to be permanent. + """ + if major_version < SLOT_ADVANCE_AVAILABLE_VERSION: + return {} # for legacy PostgreSQL we don't support permanent slots on standby nodes + + permanent_slots: Dict[str, Any] = self._get_permanent_slots(is_standby_cluster=is_standby_cluster, + role='replica', + nofailover=False, + major_version=major_version) + members_slots = {slot_name_from_member_name(m.name) for m in self.members} + + return {name: value for name, value in slots.items() if name in permanent_slots + and (self.is_physical_slot(permanent_slots[name]) + or self.is_logical_slot(permanent_slots[name]) and name not in members_slots)} def _has_permanent_logical_slots(self, my_name: str, nofailover: bool) -> bool: """Check if the given member node has permanent ``logical`` replication slots configured. diff --git a/patroni/ha.py b/patroni/ha.py index 877e24ef..3b00beb9 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -14,7 +14,7 @@ from . import psycopg from .__main__ import Patroni from .async_executor import AsyncExecutor, CriticalTask from .collections import CaseInsensitiveSet -from .dcs import AbstractDCS, Cluster, Leader, Member, RemoteMember, Status +from .dcs import AbstractDCS, Cluster, Leader, Member, RemoteMember, Status, slot_name_from_member_name from .exceptions import DCSError, PostgresConnectionException, PatroniFatalException from .postgresql.callback_executor import CallbackAction from .postgresql.misc import postgres_version_to_int @@ -272,12 +272,32 @@ class Ha(object): ret[self.state_handler.name] = self.patroni.api.connection_string return ret - def update_lock(self, write_leader_optime: bool = False) -> bool: + def update_lock(self, update_status: bool = False) -> bool: + """Update the leader lock in DCS. + + .. note:: + After successful update of the leader key the :meth:`AbstractDCS.update_leader` method could also + optionally update the ``/status`` and ``/failsafe`` keys. + + The ``/status`` key contains the last known LSN on the leader node and the last known state + of permanent replication slots including permanent physical replication slot for the leader. + + Last, but not least, this method calls a :meth:`Watchdog.keepalive` method after the leader key + was successfully updated. + + :param update_status: ``True`` if we also need to update the ``/status`` key in DCS, otherwise ``False``. + + :returns: ``True`` if the leader key was successfully updated and we can continue to run postgres + as a ``primary`` or as a ``standby_leader``, otherwise ``False``. + """ last_lsn = slots = None - if write_leader_optime: + if update_status: try: last_lsn = self.state_handler.last_operation() - slots = self.state_handler.slots() + slots = self.cluster.filter_permanent_slots( + {**self.state_handler.slots(), slot_name_from_member_name(self.state_handler.name): last_lsn}, + self.is_standby_cluster(), + self.state_handler.major_version) except Exception: logger.exception('Exception when called state_handler.last_operation()') if TYPE_CHECKING: # pragma: no cover @@ -900,6 +920,26 @@ class Ha(object): return False def check_failsafe_topology(self) -> bool: + """Check whether we could continue to run as a primary by calling all members from the failsafe topology. + + .. note:: + If the ``/failsafe`` key contains invalid data or if the ``name`` of our node is missing in + the ``/failsafe`` key, we immediately give up and return ``False``. + + We send the JSON document in the POST request with the following fields: + + * ``name`` - the name of our node; + * ``conn_url`` - connection URL to the postgres, which is reachable from other nodes; + * ``api_url`` - connection URL to Patroni REST API on this node reachable from other nodes; + * ``slots`` - a :class:`dict` with replication slots that exist on the leader node, including the primary + itself with the last known LSN, because there could be a permanent physical slot on standby nodes. + + Standby nodes are using information from the ``slots`` dict to advance position of permanent + replication slots while DCS is not accessible in order to avoid indefinite growth of ``pg_wal``. + + :returns: ``True`` if all members from the ``/failsafe`` topology agree that this node could continue to + run as a ``primary``, or ``False`` if some of standby nodes are not accessible or don't agree. + """ failsafe = self.dcs.failsafe if not isinstance(failsafe, dict) or self.state_handler.name not in failsafe: return False @@ -909,7 +949,10 @@ class Ha(object): 'api_url': self.patroni.api.connection_string, } try: - data['slots'] = self.state_handler.slots() + data['slots'] = { + **self.state_handler.slots(), + slot_name_from_member_name(self.state_handler.name): self.state_handler.last_operation() + } except Exception: logger.exception('Exception when called state_handler.slots()') members = [RemoteMember(name, {'api_url': url}) diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py index cab33e8f..7a3d4065 100644 --- a/patroni/postgresql/__init__.py +++ b/patroni/postgresql/__init__.py @@ -451,15 +451,21 @@ class Postgresql(object): return if self._global_config.is_standby_cluster: - self._has_permanent_slots = False # Standby cluster can't have logical replication slots, and we don't need to enforce hot_standby_feedback self.set_enforce_hot_standby_feedback(False) - elif cluster and cluster.config and cluster.config.modify_version: - self._has_permanent_slots = cluster.has_permanent_slots(self.name, nofailover) + + if cluster and cluster.config and cluster.config.modify_version: # We want to enable hot_standby_feedback if the replica is supposed # to have a logical slot or in case if it is the cascading replica. - self.set_enforce_hot_standby_feedback( - self.can_advance_slots and cluster.should_enforce_hot_standby_feedback(self.name, nofailover)) + self.set_enforce_hot_standby_feedback(not self._global_config.is_standby_cluster and self.can_advance_slots + and cluster.should_enforce_hot_standby_feedback(self.name, + nofailover)) + + self._has_permanent_slots = cluster.has_permanent_slots( + my_name=self.name, + is_standby_cluster=self._global_config.is_standby_cluster, + nofailover=nofailover, + major_version=self.major_version) def _cluster_info_state_get(self, name: str) -> Optional[Any]: if not self._cluster_info_state: diff --git a/patroni/postgresql/slots.py b/patroni/postgresql/slots.py index 29bbb130..48b275e4 100644 --- a/patroni/postgresql/slots.py +++ b/patroni/postgresql/slots.py @@ -16,7 +16,6 @@ from .misc import format_lsn, fsync_dir from ..dcs import Cluster, Leader from ..file_perm import pg_perm from ..psycopg import OperationalError -from ..utils import parse_int if TYPE_CHECKING: # pragma: no cover from psycopg import Cursor @@ -378,10 +377,9 @@ class SlotsHandler: except Exception: logger.exception("Failed to create physical replication slot '%s'", name) self._schedule_load_slots = True - elif not self._postgresql.is_primary() and self._postgresql.can_advance_slots \ - and self._replication_slots[name]['type'] == 'physical': + elif self._postgresql.can_advance_slots and self._replication_slots[name]['type'] == 'physical': value['restart_lsn'] = self._replication_slots[name]['restart_lsn'] - lsn = parse_int(value.get('lsn')) + lsn = value.get('lsn') if lsn and lsn > value['restart_lsn']: # The slot has feedback in DCS and needs to be advanced try: lsn = format_lsn(lsn) @@ -477,12 +475,9 @@ class SlotsHandler: # If the logical already exists, copy some information about it into the original structure if name in self._replication_slots and compare_slots(value, self._replication_slots[name]): self._copy_items(self._replication_slots[name], value) - if 'lsn' in value: # The slot has feedback in DCS - try: # Skip slots that don't need to be advanced - if value['confirmed_flush_lsn'] < int(value['lsn']): - advance_slots[value['database']][name] = int(value['lsn']) - except Exception as e: - logger.error('Failed to parse "%s": %r', value['lsn'], e) + if 'lsn' in value and value['confirmed_flush_lsn'] < value['lsn']: # The slot has feedback in DCS + # Skip slots that don't need to be advanced + advance_slots[value['database']][name] = value['lsn'] elif name not in self._replication_slots and 'lsn' in value: # We want to copy only slots with feedback in a DCS create_slots.append(name) diff --git a/patroni/utils.py b/patroni/utils.py index be468d2e..6957369f 100644 --- a/patroni/utils.py +++ b/patroni/utils.py @@ -819,7 +819,7 @@ def cluster_as_json(cluster: 'Cluster', global_config: Optional['GlobalConfig'] member.update({n: m.data[n] for n in optional_attributes if n in m.data}) if m.name != leader_name: - lsn = m.data.get('xlog_location') + lsn = m.lsn if lsn is None: member['lag'] = 'unknown' elif cluster_lsn >= lsn: diff --git a/tests/test_ha.py b/tests/test_ha.py index ea8fa7c0..eeea68b6 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -167,6 +167,7 @@ def run_async(self, func, args=()): @patch.object(Postgresql, 'is_primary', Mock(return_value=True)) @patch.object(Postgresql, 'timeline_wal_position', Mock(return_value=(1, 10, 1))) @patch.object(Postgresql, '_cluster_info_state_get', Mock(return_value=10)) +@patch.object(Postgresql, 'slots', Mock(return_value={'l': 100})) @patch.object(Postgresql, 'data_directory_empty', Mock(return_value=False)) @patch.object(Postgresql, 'controldata', Mock(return_value={ 'Database system identifier': SYSID, @@ -1582,6 +1583,7 @@ class TestHa(PostgresInit): @patch('patroni.psycopg.connect', psycopg_connect) def test_permanent_logical_slots_after_promote(self): + self.p._major_version = 110000 config = ClusterConfig(1, {'slots': {'l': {'database': 'postgres', 'plugin': 'test_decoding'}}}, 1) self.p.name = 'other' self.ha.cluster = get_cluster_initialized_without_leader(cluster_config=config) diff --git a/tests/test_slots.py b/tests/test_slots.py index d1b8d458..83087c4e 100644 --- a/tests/test_slots.py +++ b/tests/test_slots.py @@ -128,6 +128,7 @@ class TestSlotsHandler(BaseTestPostgresql): self.cluster.slots['ls'] = 'a' self.assertEqual(self.s.sync_replication_slots(self.cluster, False), []) self.cluster.config.data['slots']['ls']['database'] = 'b' + self.cluster.slots['ls'] = '500' with patch.object(MockCursor, 'rowcount', PropertyMock(return_value=1), create=True): self.assertEqual(self.s.sync_replication_slots(self.cluster, False), ['ls']) From ce187bec3881f84d08d786a924d7b020f1c0cdb5 Mon Sep 17 00:00:00 2001 From: GuanqunYang193 <144971563+GuanqunYang193@users.noreply.github.com> Date: Mon, 23 Oct 2023 02:29:09 -0400 Subject: [PATCH 16/19] Remove user creation related docs (#2920) * Remove user creation related docs * remove template --- docs/ENVIRONMENT.rst | 9 --------- docs/yaml_configuration.rst | 16 ---------------- postgres0.yml | 8 -------- postgres1.yml | 8 -------- postgres2.yml | 8 -------- 5 files changed, 49 deletions(-) diff --git a/docs/ENVIRONMENT.rst b/docs/ENVIRONMENT.rst index f859649d..bd928af9 100644 --- a/docs/ENVIRONMENT.rst +++ b/docs/ENVIRONMENT.rst @@ -24,15 +24,6 @@ Log - **PATRONI\_LOG\_FILE\_SIZE**: Size of patroni.log file (in bytes) that triggers a log rolling. - **PATRONI\_LOG\_LOGGERS**: Redefine logging level per python module. Example ``PATRONI_LOG_LOGGERS="{patroni.postmaster: WARNING, urllib3: DEBUG}"`` -Bootstrap configuration ------------------------ -It is possible to create new database users right after the successful initialization of a new cluster. This process is defined by the following variables: - -- **PATRONI\_\_PASSWORD=''** -- **PATRONI\_\_OPTIONS='list,of,options'** - -Example: defining ``PATRONI_admin_PASSWORD=strongpasswd`` and ``PATRONI_admin_OPTIONS='createrole,createdb'`` will cause creation of the user **admin** with the password **strongpasswd** that is allowed to create other users and databases. - Citus ----- Enables integration Patroni with `Citus `__. If configured, Patroni will take care of registering Citus worker nodes on the coordinator. You can find more information about Citus support :ref:`here `. diff --git a/docs/yaml_configuration.rst b/docs/yaml_configuration.rst index 4c2ed7c0..75992217 100644 --- a/docs/yaml_configuration.rst +++ b/docs/yaml_configuration.rst @@ -49,24 +49,8 @@ Bootstrap configuration - **- data-checksums**: Must be enabled when pg_rewind is needed on 9.3. - **- encoding: UTF8**: default encoding for new databases. - **- locale: UTF8**: default locale for new databases. - - **users**: Some additional users which need to be created after initializing new cluster, see :ref:`Bootstrap users configuration ` below. - **post\_bootstrap** or **post\_init**: An additional script that will be executed after initializing the cluster. The script receives a connection string URL (with the cluster superuser as a user name). The PGPASSFILE variable is set to the location of pgpass file. -.. _bootstrap_users_configuration: - -Bootstrap users configuration -============================= - -Users which need to be created after initializing the cluster: - -- **admin**: the name of user - - - **password**: (optional) password for the user - - **options**: list of options for CREATE USER statement - - - **- createrole** - - **- createdb** - .. _citus_settings: Citus diff --git a/postgres0.yml b/postgres0.yml index 0605c322..ee69b2e8 100644 --- a/postgres0.yml +++ b/postgres0.yml @@ -93,14 +93,6 @@ 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 which needs to be created after initializing new cluster - users: - admin: - password: admin% - options: - - createrole - - createdb - postgresql: listen: 127.0.0.1:5432 connect_address: 127.0.0.1:5432 diff --git a/postgres1.yml b/postgres1.yml index 89dcca1a..860e364e 100644 --- a/postgres1.yml +++ b/postgres1.yml @@ -87,14 +87,6 @@ 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 which needs to be created after initializing new cluster - users: - admin: - password: admin% - options: - - createrole - - createdb - postgresql: listen: 127.0.0.1:5433 connect_address: 127.0.0.1:5433 diff --git a/postgres2.yml b/postgres2.yml index 581fa719..3a06d912 100644 --- a/postgres2.yml +++ b/postgres2.yml @@ -84,14 +84,6 @@ bootstrap: - encoding: UTF8 - data-checksums - # Some additional users which needs to be created after initializing new cluster - users: - admin: - password: admin% - options: - - createrole - - createdb - postgresql: listen: 127.0.0.1:5434 connect_address: 127.0.0.1:5434 From 6cfd90401ef2ffd1d7fcf9a7b6e8298ea547634d Mon Sep 17 00:00:00 2001 From: zhjwpku Date: Mon, 23 Oct 2023 14:30:13 +0800 Subject: [PATCH 17/19] get rid of stale comment of get_cluster (#2922) PR #2909 remove the cache in Zookeeper implementation of DCS, so the comment of get_cluster should be changed to 'Retrieve a fresh view of DCS' since every implementation does so. Signed-off-by: Zhao Junwang --- patroni/dcs/__init__.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index bc74bbe0..389cae86 100644 --- a/patroni/dcs/__init__.py +++ b/patroni/dcs/__init__.py @@ -1640,7 +1640,7 @@ class AbstractDCS(abc.ABC): return cluster def get_cluster(self) -> Cluster: - """Retrieve an appropriate cached or fresh view of DCS. + """Retrieve a fresh view of DCS. .. note:: Stores copy of time, status and failsafe values for comparison in DCS update decisions. From 6d98944e733c452f18567c8b66ccd538c7751296 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Mon, 23 Oct 2023 10:03:18 +0200 Subject: [PATCH 18/19] Add warning to the sample config about bootstrap section (#2925) often people are trying to change it and coming with the questions why it doesn't work. --- postgres0.yml | 8 ++++++-- postgres1.yml | 8 ++++++-- postgres2.yml | 8 ++++++-- 3 files changed, 18 insertions(+), 6 deletions(-) diff --git a/postgres0.yml b/postgres0.yml index ee69b2e8..8a975156 100644 --- a/postgres0.yml +++ b/postgres0.yml @@ -43,9 +43,13 @@ etcd: # - 127.0.0.1:2223 # - 127.0.0.1:2224 +# The bootstrap configuration. Works only when the cluster is not yet initialized. +# If the cluster is already initialized, all changes in the `bootstrap` section are ignored! bootstrap: - # this section will be written into Etcd:///config after initializing new cluster - # and all other cluster members will use it as a `global configuration` + # This section will be written into Etcd:///config after initializing new cluster + # and all other cluster members will use it as a `global configuration`. + # WARNING! If you want to change any of the parameters that were set up + # via `bootstrap.dcs` section, please use `patronictl edit-config`! dcs: ttl: 30 loop_wait: 10 diff --git a/postgres1.yml b/postgres1.yml index 860e364e..6ca2aa64 100644 --- a/postgres1.yml +++ b/postgres1.yml @@ -43,9 +43,13 @@ etcd: # - 127.0.0.1:2222 # - 127.0.0.1:2224 +# The bootstrap configuration. Works only when the cluster is not yet initialized. +# If the cluster is already initialized, all changes in the `bootstrap` section are ignored! bootstrap: - # this section will be written into Etcd:///config after initializing new cluster - # and all other cluster members will use it as a `global configuration` + # This section will be written into Etcd:///config after initializing new cluster + # and all other cluster members will use it as a `global configuration`. + # WARNING! If you want to change any of the parameters that were set up + # via `bootstrap.dcs` section, please use `patronictl edit-config`! dcs: ttl: 30 loop_wait: 10 diff --git a/postgres2.yml b/postgres2.yml index 3a06d912..ee61a023 100644 --- a/postgres2.yml +++ b/postgres2.yml @@ -43,9 +43,13 @@ etcd: # - 127.0.0.1:2222 # - 127.0.0.1:2223 +# The bootstrap configuration. Works only when the cluster is not yet initialized. +# If the cluster is already initialized, all changes in the `bootstrap` section are ignored! bootstrap: - # this section will be written into Etcd:///config after initializing new cluster - # and all other cluster members will use it as a `global configuration` + # This section will be written into Etcd:///config after initializing new cluster + # and all other cluster members will use it as a `global configuration`. + # WARNING! If you want to change any of the parameters that were set up + # via `bootstrap.dcs` section, please use `patronictl edit-config`! dcs: ttl: 30 loop_wait: 10 From d471f1156d795c65f4cd7bae1f6e87501972d977 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Mon, 23 Oct 2023 14:00:37 +0200 Subject: [PATCH 19/19] Handle AuthOldRevision error (#2913) The error is raised if Etcd is configured to use JWT auth tokens and when the user database in Etcd is updated, because the update invalidates all tokens. If retries are requested - try to get a new new token and repeat the request. Repeat it in a loop until request is successfully executed or until `retry_timeout` is exhausted. This is the only way of solving a race condition, because between authentication and executing the request yet another modification of the user database in Etcd might happen. In case if the request doesn't have to be immediately retried - set a flag that the next API request should perform the authentication first and let Patroni to naturally repeat the request on the next heartbeat loop. Co-authored-by: Kenny Do Ref: https://github.com/zalando/patroni/pull/2911 --- patroni/dcs/etcd3.py | 106 +++++++++++++++++++++++++++---------------- tests/test_etcd3.py | 15 ++++-- 2 files changed, 78 insertions(+), 43 deletions(-) diff --git a/patroni/dcs/etcd3.py b/patroni/dcs/etcd3.py index 5cbd813f..ea7e52f2 100644 --- a/patroni/dcs/etcd3.py +++ b/patroni/dcs/etcd3.py @@ -124,6 +124,10 @@ class AuthFailed(InvalidArgument): error = "etcdserver: authentication failed, invalid user ID or password" +class AuthOldRevision(InvalidArgument): + error = "etcdserver: revision of auth store is old" + + class PermissionDenied(Etcd3ClientError): code = GRPCCode.PermissionDenied error = "etcdserver: permission denied" @@ -193,6 +197,12 @@ def build_range_request(key: str, range_end: Union[bytes, str, None] = None) -> return fields +class ReAuthenticateMode(IntEnum): + NOT_REQUIRED = 0 + REQUIRED = 1 + WITHOUT_WATCHER_RESTART = 2 + + def _handle_auth_errors(func: Callable[..., Any]) -> Any: def wrapper(self: 'Etcd3Client', *args: Any, **kwargs: Any) -> Any: return self.handle_auth_errors(func, *args, **kwargs) @@ -204,6 +214,7 @@ class Etcd3Client(AbstractEtcdClientWithFailover): ERROR_CLS = Etcd3Error def __init__(self, config: Dict[str, Any], dns_resolver: DnsCachingResolver, cache_ttl: int = 300) -> None: + self._reauthenticate_reason = ReAuthenticateMode.NOT_REQUIRED self._token = None self._cluster_version: Tuple[int, ...] = tuple() super(Etcd3Client, self).__init__({**config, 'version_prefix': '/v3beta'}, dns_resolver, cache_ttl) @@ -282,7 +293,7 @@ class Etcd3Client(AbstractEtcdClientWithFailover): fields['retry'] = retry return self.api_execute(self.version_prefix + method, self._MPOST, fields) - def authenticate(self) -> bool: + def authenticate(self, *, restart_watcher: bool = True, retry: Optional[Retry] = None) -> bool: if self._use_proxies and not self._cluster_version: kwargs = self._prepare_common_parameters(1) self._ensure_version_prefix(self._base_uri, **kwargs) @@ -291,7 +302,7 @@ class Etcd3Client(AbstractEtcdClientWithFailover): logger.info('Trying to authenticate on Etcd...') old_token, self._token = self._token, None try: - response = self.call_rpc('/auth/authenticate', {'name': self.username, 'password': self.password}) + response = self.call_rpc('/auth/authenticate', {'name': self.username, 'password': self.password}, retry) except AuthNotEnabled: logger.info('Etcd authentication is not enabled') self._token = None @@ -302,48 +313,65 @@ class Etcd3Client(AbstractEtcdClientWithFailover): self._token = response.get('token') return old_token != self._token - def handle_auth_errors(self: 'Etcd3Client', func: Callable[..., Any], *args: Any, **kwargs: Any) -> Any: - def retry(ex: Exception) -> Any: - if self.username and self.password: - self.authenticate() - return func(self, *args, **kwargs) - else: - logger.fatal('Username or password not set, authentication is not possible') - raise ex + def handle_auth_errors(self: 'Etcd3Client', func: Callable[..., Any], *args: Any, + retry: Optional[Retry] = None, **kwargs: Any) -> Any: + exc = None + while True: + if self._reauthenticate_reason: + if self.username and self.password: + self.authenticate( + restart_watcher=self._reauthenticate_reason != ReAuthenticateMode.WITHOUT_WATCHER_RESTART, + retry=retry) + self._reauthenticate_reason = ReAuthenticateMode.NOT_REQUIRED + if retry: + retry.ensure_deadline(0) + else: + msg = 'Username or password not set, authentication is not possible' + logger.fatal(msg) + raise exc or Etcd3Exception(msg) - try: - return func(self, *args, **kwargs) - except (UserEmpty, PermissionDenied) as e: # no token provided - # PermissionDenied is raised on 3.0 and 3.1 - if self._cluster_version < (3, 3) and (not isinstance(e, PermissionDenied) - or self._cluster_version < (3, 2)): - raise UnsupportedEtcdVersion('Authentication is required by Etcd cluster but not ' - 'supported on version lower than 3.3.0. Cluster version: ' - '{0}'.format('.'.join(map(str, self._cluster_version)))) - return retry(e) - except InvalidAuthToken as e: - logger.error('Invalid auth token: %s', self._token) - return retry(e) + try: + return func(self, *args, retry=retry, **kwargs) + except (UserEmpty, PermissionDenied) as e: # no token provided + # PermissionDenied is raised on 3.0 and 3.1 + if self._cluster_version < (3, 3) and (not isinstance(e, PermissionDenied) + or self._cluster_version < (3, 2)): + raise UnsupportedEtcdVersion('Authentication is required by Etcd cluster but not ' + 'supported on version lower than 3.3.0. Cluster version: ' + '{0}'.format('.'.join(map(str, self._cluster_version)))) + exc = e + except InvalidAuthToken as e: + logger.error('Invalid auth token: %s', self._token) + exc = e + except AuthOldRevision as e: + logger.error('Auth token is for old revision of auth store') + exc = e + self._reauthenticate_reason = ReAuthenticateMode.WITHOUT_WATCHER_RESTART \ + if isinstance(exc, AuthOldRevision) else ReAuthenticateMode.REQUIRED + if not retry: + raise exc + retry.ensure_deadline(0.5, exc) @_handle_auth_errors def range(self, key: str, range_end: Union[bytes, str, None] = None, serializable: bool = True, - retry: Optional[Retry] = None) -> Dict[str, Any]: + *, retry: Optional[Retry] = None) -> Dict[str, Any]: params = build_range_request(key, range_end) params['serializable'] = serializable # For better performance. We can tolerate stale reads return self.call_rpc('/kv/range', params, retry) - def prefix(self, key: str, serializable: bool = True, retry: Optional[Retry] = None) -> Dict[str, Any]: - return self.range(key, prefix_range_end(key), serializable, retry) + def prefix(self, key: str, serializable: bool = True, *, retry: Optional[Retry] = None) -> Dict[str, Any]: + return self.range(key, prefix_range_end(key), serializable, retry=retry) @_handle_auth_errors - def lease_grant(self, ttl: int, retry: Optional[Retry] = None) -> str: + def lease_grant(self, ttl: int, *, retry: Optional[Retry] = None) -> str: return self.call_rpc('/lease/grant', {'TTL': ttl}, retry)['ID'] - def lease_keepalive(self, ID: str, retry: Optional[Retry] = None) -> Optional[str]: + def lease_keepalive(self, ID: str, *, retry: Optional[Retry] = None) -> Optional[str]: return self.call_rpc('/lease/keepalive', {'ID': ID}, retry).get('result', {}).get('TTL') + @_handle_auth_errors def txn(self, compare: Dict[str, Any], success: Dict[str, Any], - failure: Optional[Dict[str, Any]] = None, retry: Optional[Retry] = None) -> Dict[str, Any]: + failure: Optional[Dict[str, Any]] = None, *, retry: Optional[Retry] = None) -> Dict[str, Any]: fields = {'compare': [compare], 'success': [success]} if failure: fields['failure'] = [failure] @@ -352,7 +380,7 @@ class Etcd3Client(AbstractEtcdClientWithFailover): @_handle_auth_errors def put(self, key: str, value: str, lease: Optional[str] = None, create_revision: Optional[str] = None, - mod_revision: Optional[str] = None, retry: Optional[Retry] = None) -> Dict[str, Any]: + mod_revision: Optional[str] = None, *, retry: Optional[Retry] = None) -> Dict[str, Any]: fields = {'key': base64_encode(key), 'value': base64_encode(value)} if lease: fields['lease'] = lease @@ -367,14 +395,14 @@ class Etcd3Client(AbstractEtcdClientWithFailover): @_handle_auth_errors def deleterange(self, key: str, range_end: Union[bytes, str, None] = None, - mod_revision: Optional[str] = None, retry: Optional[Retry] = None) -> Dict[str, Any]: + mod_revision: Optional[str] = None, *, retry: Optional[Retry] = None) -> Dict[str, Any]: fields = build_range_request(key, range_end) if mod_revision is None: return self.call_rpc('/kv/deleterange', fields, retry) compare = {'target': 'MOD', 'mod_revision': mod_revision, 'key': fields['key']} return self.txn(compare, {'request_delete_range': fields}, retry=retry) - def deleteprefix(self, key: str, retry: Optional[Retry] = None) -> Dict[str, Any]: + def deleteprefix(self, key: str, *, retry: Optional[Retry] = None) -> Dict[str, Any]: return self.deleterange(key, prefix_range_end(key), retry=retry) def watchrange(self, key: str, range_end: Union[bytes, str, None] = None, @@ -574,9 +602,9 @@ class PatroniEtcd3Client(Etcd3Client): super(PatroniEtcd3Client, self).set_base_uri(value) self._restart_watcher() - def authenticate(self) -> bool: - ret = super(PatroniEtcd3Client, self).authenticate() - if ret: + def authenticate(self, *, restart_watcher: bool = True, retry: Optional[Retry] = None) -> bool: + ret = super(PatroniEtcd3Client, self).authenticate(restart_watcher=restart_watcher, retry=retry) + if ret and restart_watcher: self._restart_watcher() return ret @@ -631,8 +659,8 @@ 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) + failure: Optional[Dict[str, Any]] = None, *, retry: Optional[Retry] = None) -> Dict[str, Any]: + ret = super(PatroniEtcd3Client, self).txn(compare, success, failure, retry=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. @@ -676,12 +704,12 @@ class Etcd3(AbstractEtcd): if not force and self._lease and self._last_lease_refresh + self._loop_wait > time.time(): return False - if self._lease and not self._client.lease_keepalive(self._lease, retry): + if self._lease and not self._client.lease_keepalive(self._lease, retry=retry): self._lease = None ret = not self._lease if ret: - self._lease = self._client.lease_grant(self._ttl, retry) + self._lease = self._client.lease_grant(self._ttl, retry=retry) self._last_lease_refresh = time.time() return ret diff --git a/tests/test_etcd3.py b/tests/test_etcd3.py index 9aed7eb1..10ab1ea5 100644 --- a/tests/test_etcd3.py +++ b/tests/test_etcd3.py @@ -6,8 +6,8 @@ import urllib3 from mock import Mock, PropertyMock, patch from patroni.dcs.etcd import DnsCachingResolver from patroni.dcs.etcd3 import PatroniEtcd3Client, Cluster, Etcd3, Etcd3Client, \ - Etcd3Error, Etcd3ClientError, RetryFailedError, InvalidAuthToken, Unavailable, \ - Unknown, UnsupportedEtcdVersion, UserEmpty, AuthFailed, base64_encode + Etcd3Error, Etcd3ClientError, ReAuthenticateMode, RetryFailedError, InvalidAuthToken, Unavailable, \ + Unknown, UnsupportedEtcdVersion, UserEmpty, AuthFailed, AuthOldRevision, base64_encode from threading import Thread from . import SleepException, MockResponse @@ -161,9 +161,16 @@ class TestPatroniEtcd3Client(BaseTestEtcd3): mock_urlopen.return_value.content = '{"code":16,"error":"etcdserver: invalid auth token"}' self.assertRaises(InvalidAuthToken, self.client.deleteprefix, 'foo') with patch.object(PatroniEtcd3Client, 'authenticate', Mock(return_value=True)): - self.assertRaises(InvalidAuthToken, self.client.deleteprefix, 'foo') + retry = self.etcd3._retry.copy() + with patch('time.time', Mock(side_effect=[0, 10, 20, 30, 40])): + self.assertRaises(InvalidAuthToken, retry, self.client.deleteprefix, 'foo', retry=retry) self.client.username = None - self.assertRaises(InvalidAuthToken, self.client.deleteprefix, 'foo') + self.client._reauthenticate_reason = ReAuthenticateMode.NOT_REQUIRED + retry = self.etcd3._retry.copy() + self.assertRaises(InvalidAuthToken, retry, self.client.deleteprefix, 'foo', retry=retry) + mock_urlopen.return_value.content = '{"code":3,"error":"etcdserver: revision of auth store is old"}' + self.client._reauthenticate_reason = ReAuthenticateMode.NOT_REQUIRED + self.assertRaises(AuthOldRevision, retry, self.client.deleteprefix, 'foo', retry=retry) def test__handle_server_response(self): response = MockResponse()