From 5a77cbb08751a58e1f8a2e9473ef56a89974c1c0 Mon Sep 17 00:00:00 2001 From: Ali Mehraji Date: Thu, 30 Nov 2023 12:15:07 +0330 Subject: [PATCH 01/18] Update: etcd flags in command in docker-compose.yml and docker-compose-citus.yml (#2966) --- docker-compose-citus.yml | 6 +++--- docker-compose.yml | 6 +++--- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/docker-compose-citus.yml b/docker-compose-citus.yml index cd63d583..da71c50a 100644 --- a/docker-compose-citus.yml +++ b/docker-compose-citus.yml @@ -27,19 +27,19 @@ services: ETCD_UNSUPPORTED_ARCH: arm64 container_name: demo-etcd1 hostname: etcd1 - command: etcd -name etcd1 -initial-advertise-peer-urls http://etcd1:2380 + command: etcd --name etcd1 --initial-advertise-peer-urls http://etcd1:2380 etcd2: <<: *etcd container_name: demo-etcd2 hostname: etcd2 - command: etcd -name etcd2 -initial-advertise-peer-urls http://etcd2:2380 + command: etcd --name etcd2 --initial-advertise-peer-urls http://etcd2:2380 etcd3: <<: *etcd container_name: demo-etcd3 hostname: etcd3 - command: etcd -name etcd3 -initial-advertise-peer-urls http://etcd3:2380 + command: etcd --name etcd3 --initial-advertise-peer-urls http://etcd3:2380 haproxy: image: ${PATRONI_TEST_IMAGE:-patroni-citus} diff --git a/docker-compose.yml b/docker-compose.yml index 996c2c82..6b7d7a92 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -25,19 +25,19 @@ services: ETCD_UNSUPPORTED_ARCH: arm64 container_name: demo-etcd1 hostname: etcd1 - command: etcd -name etcd1 -initial-advertise-peer-urls http://etcd1:2380 + command: etcd --name etcd1 --initial-advertise-peer-urls http://etcd1:2380 etcd2: <<: *etcd container_name: demo-etcd2 hostname: etcd2 - command: etcd -name etcd2 -initial-advertise-peer-urls http://etcd2:2380 + command: etcd --name etcd2 --initial-advertise-peer-urls http://etcd2:2380 etcd3: <<: *etcd container_name: demo-etcd3 hostname: etcd3 - command: etcd -name etcd3 -initial-advertise-peer-urls http://etcd3:2380 + command: etcd --name etcd3 --initial-advertise-peer-urls http://etcd3:2380 haproxy: image: ${PATRONI_TEST_IMAGE:-patroni} From 47cadc9f63e087ebdfd598f18b02d831a8238ead Mon Sep 17 00:00:00 2001 From: Sophia Ruan <104968314+XiuhuaRuan@users.noreply.github.com> Date: Thu, 30 Nov 2023 17:02:19 +0800 Subject: [PATCH 02/18] Fix the issue that REST API returns unknown after postgres restart (#2956) Close #2955 --- patroni/postgresql/connection.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/patroni/postgresql/connection.py b/patroni/postgresql/connection.py index 2a50dbb5..040dcf78 100644 --- a/patroni/postgresql/connection.py +++ b/patroni/postgresql/connection.py @@ -147,7 +147,8 @@ class ConnectionPool: def close(self) -> None: """Close all named connections from Patroni to PostgreSQL registered in the pool.""" with self._lock: - if any(conn.close(True) for conn in self._connections.values()): + closed_connections = [conn.close(True) for conn in self._connections.values()] + if any(closed_connections): logger.info("closed patroni connections to postgres") From ef5f320602ccf05415444ee19c1497ed6b07de73 Mon Sep 17 00:00:00 2001 From: Waynerv Date: Thu, 30 Nov 2023 19:02:42 +0800 Subject: [PATCH 03/18] Cache `postgres --describe-config` output results (#2967) We don't expect GUCs list to change for the same major version and don't expect major version to change while Patroni is running. --- patroni/postgresql/__init__.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py index e373bd3c..c87c29fa 100644 --- a/patroni/postgresql/__init__.py +++ b/patroni/postgresql/__init__.py @@ -118,6 +118,8 @@ class Postgresql(object): # Last known running process self._postmaster_proc = None + self._available_gucs = None + if self.is_running(): # If we found postmaster process we need to figure out whether postgres is accepting connections self.set_state('starting') @@ -240,7 +242,9 @@ class Postgresql(object): @property def available_gucs(self) -> CaseInsensitiveSet: """GUCs available in this Postgres server.""" - return self._get_gucs() + if not self._available_gucs: + self._available_gucs = self._get_gucs() + return self._available_gucs def _version_file_exists(self) -> bool: return not self.data_directory_empty() and os.path.isfile(self._version_file) From 6976939f09fae574e3e7d251d61f1ba8cb0c49a5 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Thu, 30 Nov 2023 16:50:42 +0100 Subject: [PATCH 04/18] Release/v3.2.1 (#2968) - bump version - bump pyright - update release notes --- .github/workflows/tests.yaml | 2 +- docs/releases.rst | 38 ++++++++++++++++++++++++++++++++++++ patroni/api.py | 2 +- patroni/ctl.py | 4 ++-- patroni/dcs/consul.py | 2 +- patroni/postgresql/citus.py | 3 ++- patroni/version.py | 2 +- 7 files changed, 46 insertions(+), 7 deletions(-) diff --git a/.github/workflows/tests.yaml b/.github/workflows/tests.yaml index df911e54..aafe6b56 100644 --- a/.github/workflows/tests.yaml +++ b/.github/workflows/tests.yaml @@ -174,7 +174,7 @@ jobs: - uses: jakebailey/pyright-action@v1 with: - version: 1.1.336 + version: 1.1.338 docs: runs-on: ubuntu-latest diff --git a/docs/releases.rst b/docs/releases.rst index cd2714b6..c68977cd 100644 --- a/docs/releases.rst +++ b/docs/releases.rst @@ -3,6 +3,44 @@ Release notes ============= +Version 3.2.1 +------------- + +**Bugfixes** + +- Limit accepted values for ``--format`` argument in ``patronictl`` (Alexander Kukushkin) + + It used to accept any arbitrary string and produce no output if the value wasn't recognized. + +- Verify that replica nodes received checkpoint LSN on shutdown before releasing the leader key (Alexander Kukushkin) + + Previously in some cases, we were using LSN of the SWITCH record that is followed by CHECKPOINT (if archiving mode is enabled). As a result the former primary sometimes had to do ``pg_rewind``, but there would be no data loss involved. + +- Do a real HTTP request when performing node name uniqueness check (Alexander Kukushkin) + + When running Patroni in containers it is possible that the traffic is routed using ``docker-proxy``, which listens on the port and accepts incoming connections. It was causing false positives. + +- Fixed Citus support with Etcd v2 (Alexander Kukushkin) + + Patroni was failing to deploy a new Citus cluster with Etcd v2. + +- Fixed ``pg_rewind`` behavior with Postgres v16+ (Alexander Kukushkin) + + The error message format of ``pg_waldump`` changed in v16 which caused ``pg_rewind`` to be called by Patroni even when it was not necessary. + +- Fixed bug with custom bootstrap (Alexander Kukushkin) + + Patroni was falsely applying ``--command`` argument, which is a bootstrap command itself. + +- Fixed the issue with REST API health check endpoints (Sophia Ruan) + + There were chances that after Postgres restart it could return ``unknown`` state for Postgres because connections were not properly closed. + +- Cache ``postgres --describe-config`` output results (Waynerv) + + They are used to figure out which GUCs are available to validate PostgreSQL configuration and we don't expect this list to change while Patroni is running. + + Version 3.2.0 ------------- diff --git a/patroni/api.py b/patroni/api.py index 2adc5f98..5761d359 100644 --- a/patroni/api.py +++ b/patroni/api.py @@ -37,7 +37,7 @@ from .utils import deep_compare, enable_keepalive, parse_bool, patch_config, Ret logger = logging.getLogger(__name__) -def check_access(func: Callable[['RestApiHandler'], None]) -> Callable[..., None]: +def check_access(func: Callable[..., None]) -> Callable[..., None]: """Check the source ip, authorization header, or client certificates. .. note:: diff --git a/patroni/ctl.py b/patroni/ctl.py index 9345d18e..3e981b45 100644 --- a/patroni/ctl.py +++ b/patroni/ctl.py @@ -819,7 +819,7 @@ def query( raise PatroniCtlException('You need to specify either --command or --file') sql = command - connect_parameters = {} + connect_parameters: Dict[str, str] = {} if username: connect_parameters['username'] = username if password: @@ -1093,7 +1093,7 @@ def restart(cluster_name: str, group: Optional[int], member_names: List[str], version = click.prompt('Restart if the PostgreSQL version is less than provided (e.g. 9.5.2) ', type=str, default='') - content = {} + content: Dict[str, Any] = {} if pending: content['restart_pending'] = True diff --git a/patroni/dcs/consul.py b/patroni/dcs/consul.py index 1324e606..27cab778 100644 --- a/patroni/dcs/consul.py +++ b/patroni/dcs/consul.py @@ -423,7 +423,7 @@ class Consul(AbstractDCS): _, results = self.retry(self._client.kv.get, path, recurse=True, consistency=self._consistency) if results is None: return Cluster.empty() - nodes = {} + nodes: Dict[str, Dict[str, Any]] = {} for node in results: node['Value'] = (node['Value'] or b'').decode('utf-8') nodes[node['Key'][len(path):]] = node diff --git a/patroni/postgresql/citus.py b/patroni/postgresql/citus.py index 26923f37..b50dc1d0 100644 --- a/patroni/postgresql/citus.py +++ b/patroni/postgresql/citus.py @@ -100,7 +100,8 @@ class CitusHandler(Thread): def on_demote(self) -> None: with self._condition: self._pg_dist_node.clear() - self._tasks[:] = [] + empty_tasks: List[PgDistNode] = [] + self._tasks[:] = empty_tasks self._in_flight = None def query(self, sql: str, *params: Any) -> List[Tuple[Any, ...]]: diff --git a/patroni/version.py b/patroni/version.py index e5bcac2d..96592c7d 100644 --- a/patroni/version.py +++ b/patroni/version.py @@ -2,4 +2,4 @@ :var __version__: the current Patroni version. """ -__version__ = '3.2.0' +__version__ = '3.2.1' From 0e6a2ff3a9e004396d56103a86e278423e392140 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 5 Dec 2023 08:30:20 +0100 Subject: [PATCH 05/18] Don't let replica restore initialize key when DCS was wiped (#2970) It was happening from the branch where Patroni was supposed to be complain about converting standalone PG cluster to be governed by Patroni and exit. --- patroni/ha.py | 5 ++--- tests/test_ha.py | 5 +++++ 2 files changed, 7 insertions(+), 3 deletions(-) diff --git a/patroni/ha.py b/patroni/ha.py index ab1bc433..0d3e05a4 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -1849,10 +1849,9 @@ class Ha(object): logger.fatal('system ID mismatch, node %s belongs to a different cluster: %s != %s', self.state_handler.name, self.cluster.initialize, data_sysid) sys.exit(1) - elif self.cluster.is_unlocked() and not self.is_paused(): + elif self.cluster.is_unlocked() and not self.is_paused() and not self.state_handler.cb_called: # "bootstrap", but data directory is not empty - if not self.state_handler.cb_called and self.state_handler.is_running() \ - and not self.state_handler.is_primary(): + if self.state_handler.is_running() and not self.state_handler.is_primary(): self._join_aborted = True logger.error('No initialize key in DCS and PostgreSQL is running as replica, aborting start') logger.error('Please first start Patroni on the node running as primary') diff --git a/tests/test_ha.py b/tests/test_ha.py index 40063018..5b1d4562 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -1585,6 +1585,11 @@ class TestHa(PostgresInit): self.p.is_primary = false self.ha.run_cycle() exit_mock.assert_called_once_with(1) + self.p.set_role('replica') + self.ha.dcs.initialize = Mock() + with patch.object(Postgresql, 'cb_called', PropertyMock(return_value=True)): + self.assertEqual(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock') + self.ha.dcs.initialize.assert_not_called() @patch.object(Cluster, 'is_unlocked', Mock(return_value=False)) def test_after_pause(self): From a4e0a2220dd8dceffc38e23dc44e3ee12a048a8a Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 6 Dec 2023 15:28:03 +0100 Subject: [PATCH 06/18] Disable SSL for MacOS GH action runners (#2976) Latest runners release (20231127.1) somehow broke our tests. Connections to postgres somehow failing with strange error: ``` could not accept SSL connection: Socket operation on non-socket ``` --- features/environment.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/features/environment.py b/features/environment.py index a0367657..db10c93d 100644 --- a/features/environment.py +++ b/features/environment.py @@ -1073,6 +1073,8 @@ def before_all(context): context.keyfile = os.path.join(context.pctl.output_dir, 'patroni.key') context.certfile = os.path.join(context.pctl.output_dir, 'patroni.crt') try: + if sys.platform == 'darwin' and 'GITHUB_ACTIONS' in os.environ: + raise Exception with open(os.devnull, 'w') as null: ret = subprocess.call(['openssl', 'req', '-nodes', '-new', '-x509', '-subj', '/CN=batman.patroni', '-addext', 'subjectAltName=IP:127.0.0.1', '-keyout', context.keyfile, From bbddca6a76bac41ccd5fe142bc198fa7d42fa1ed Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 6 Dec 2023 15:55:51 +0100 Subject: [PATCH 07/18] Use consistent read when fetching just updated sync key (#2974) Consul doesn't provide any interface to immediately get `ModifyIndex` for the key that we just updated, therefore we have to perform an explicit read operation. By default stale reads are allowed and sometimes we may read stale data. As a result write_sync_state() call was considered as failed. To mitigate the problem we switch to `consistent` reads when that executed after update of the `/sync` key. Close #2972 --- patroni/dcs/consul.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/patroni/dcs/consul.py b/patroni/dcs/consul.py index 27cab778..fe66c6d8 100644 --- a/patroni/dcs/consul.py +++ b/patroni/dcs/consul.py @@ -666,7 +666,7 @@ class Consul(AbstractDCS): if ret: # We have no other choise, only read after write :( if not retry.ensure_deadline(0.5): return False - _, ret = self.retry(self._client.kv.get, self.sync_path) + _, ret = self.retry(self._client.kv.get, self.sync_path, consistency='consistent') if ret and (ret.get('Value') or b'').decode('utf-8') == value: return ret['ModifyIndex'] return False From efdedc7049527117b35849dc5c01525d7f1133e8 Mon Sep 17 00:00:00 2001 From: Polina Bungina <27892524+hughcapet@users.noreply.github.com> Date: Wed, 6 Dec 2023 15:57:05 +0100 Subject: [PATCH 08/18] Reload postgres config if a server param was reset (#2975) Fix the case when a parameter value was changed and then reset back to the initial value without restart - before this fix, the second change was not reflected in the Postgres config. This commit also includes the related unit test refactoring. --- patroni/postgresql/config.py | 6 ++ patroni/utils.py | 21 ++++--- tests/__init__.py | 56 +++++++++++------ tests/test_postgresql.py | 116 ++++++++++++++++++++++++++++------- 4 files changed, 150 insertions(+), 49 deletions(-) diff --git a/patroni/postgresql/config.py b/patroni/postgresql/config.py index 271bbdfe..7dfa8934 100644 --- a/patroni/postgresql/config.py +++ b/patroni/postgresql/config.py @@ -1098,6 +1098,12 @@ class ConfigHandler(object): local_connection_address_changed = True else: logger.info('Changed %s from %s to %s', r[0], r[1], new_value) + elif r[0] in self._server_parameters \ + and not compare_values(r[3], r[2], r[1], self._server_parameters[r[0]]): + # Check if any parameter was set back to the current pg_settings value + # We can use pg_settings value here, as it is proved to be equal to new_value + logger.info('Changed %s from %s to %s', r[0], self._server_parameters[r[0]], r[1]) + conf_changed = True for param, value in changes.items(): if '.' in param: # Check that user-defined-paramters have changed (parameters with period in name) diff --git a/patroni/utils.py b/patroni/utils.py index 23f419e5..2b1ded5b 100644 --- a/patroni/utils.py +++ b/patroni/utils.py @@ -400,22 +400,23 @@ def parse_real(value: Any, base_unit: Optional[str] = None) -> Optional[float]: return convert_to_base_unit(val, unit, base_unit) -def compare_values(vartype: str, unit: Optional[str], old_value: Any, new_value: Any) -> bool: - """Check if *old_value* and *new_value* are equivalent after parsing them as *vartype*. +def compare_values(vartype: str, unit: Optional[str], settings_value: Any, config_value: Any) -> bool: + """Check if the value from ``pg_settings`` and from Patroni config are equivalent after parsing them as *vartype*. - :param vartpe: the target type to parse *old_value* and *new_value* before comparing them. Accepts any among of the - following (case sensitive): + :param vartype: the target type to parse *settings_value* and *config_value* before comparing them. + Accepts any among of the following (case sensitive): * ``bool``: parse values using :func:`parse_bool`; or * ``integer``: parse values using :func:`parse_int`; or * ``real``: parse values using :func:`parse_real`; or * ``enum``: parse values as lowercase strings; or * ``string``: parse values as strings. This one is used by default if no valid value is passed as *vartype*. - :param unit: base unit to be used as argument when calling :func:`parse_int` or :func:`parse_real` for *new_value*. - :param old_value: value to be compared with *new_value*. - :param new_value: value to be compared with *old_value*. + :param unit: base unit to be used as argument when calling :func:`parse_int` or :func:`parse_real` + for *config_value*. + :param settings_value: value to be compared with *config_value*. + :param config_value: value to be compared with *settings_value*. - :returns: ``True`` if *old_value* is equivalent to *new_value* when both are parsed as *vartype*. + :returns: ``True`` if *settings_value* is equivalent to *config_value* when both are parsed as *vartype*. :Example: @@ -455,8 +456,8 @@ def compare_values(vartype: str, unit: Optional[str], old_value: Any, new_value: } converter = converters.get(vartype) or converters['string'] - old_converted = converter(old_value, None) - new_converted = converter(new_value, unit) + old_converted = converter(settings_value, None) + new_converted = converter(config_value, unit) return old_converted is not None and new_converted is not None and old_converted == new_converted diff --git a/tests/__init__.py b/tests/__init__.py index bd70ba3d..b013e4e1 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -25,8 +25,41 @@ mock_available_gucs = PropertyMock(return_value={ 'max_wal_senders', 'max_worker_processes', 'port', 'search_path', 'shared_preload_libraries', 'stats_temp_directory', 'synchronous_standby_names', 'track_commit_timestamp', 'unix_socket_directories', 'vacuum_cost_delay', 'vacuum_cost_limit', 'wal_keep_size', 'wal_level', 'wal_log_hints', 'zero_damaged_pages', + 'autovacuum', 'wal_segment_size', 'wal_block_size', 'shared_buffers', 'wal_buffers', }) +GET_PG_SETTINGS_RESULT = [ + ('wal_segment_size', '2048', '8kB', 'integer', 'internal'), + ('wal_block_size', '8192', None, 'integer', 'internal'), + ('shared_buffers', '16384', '8kB', 'integer', 'postmaster'), + ('wal_buffers', '-1', '8kB', 'integer', 'postmaster'), + ('max_connections', '100', None, 'integer', 'postmaster'), + ('max_prepared_transactions', '200', None, 'integer', 'postmaster'), + ('max_worker_processes', '8', None, 'integer', 'postmaster'), + ('max_locks_per_transaction', '64', None, 'integer', 'postmaster'), + ('max_wal_senders', '5', None, 'integer', 'postmaster'), + ('search_path', 'public', None, 'string', 'user'), + ('port', '5432', None, 'integer', 'postmaster'), + ('listen_addresses', '127.0.0.2, 127.0.0.3', None, 'string', 'postmaster'), + ('autovacuum', 'on', None, 'bool', 'sighup'), + ('unix_socket_directories', '/tmp', None, 'string', 'postmaster'), + ('shared_preload_libraries', 'citus', None, 'string', 'postmaster'), + ('wal_keep_size', '128', 'MB', 'integer', 'sighup'), + ('cluster_name', 'batman', None, 'string', 'postmaster'), + ('vacuum_cost_delay', '200', 'ms', 'real', 'user'), + ('vacuum_cost_limit', '-1', None, 'integer', 'user'), + ('max_stack_depth', '2048', 'kB', 'integer', 'superuser'), + ('constraint_exclusion', '', None, 'enum', 'user'), + ('force_parallel_mode', '1', None, 'enum', 'user'), + ('zero_damaged_pages', 'off', None, 'bool', 'superuser'), + ('stats_temp_directory', '/tmp', None, 'string', 'sighup'), + ('track_commit_timestamp', 'off', None, 'bool', 'postmaster'), + ('wal_log_hints', 'on', None, 'bool', 'superuser'), + ('hot_standby', 'on', None, 'bool', 'superuser'), + ('max_replication_slots', '5', None, 'integer', 'superuser'), + ('wal_level', 'logical', None, 'enum', 'superuser'), +] + class MockResponse(object): @@ -133,22 +166,9 @@ class MockCursor(object): ('archive_command', 'my archive command'), ('cluster_name', 'my_cluster')] elif sql.startswith('SELECT name, setting'): - self.results = [('wal_segment_size', '2048', '8kB', 'integer', 'internal'), - ('wal_block_size', '8192', None, 'integer', 'internal'), - ('shared_buffers', '16384', '8kB', 'integer', 'postmaster'), - ('wal_buffers', '-1', '8kB', 'integer', 'postmaster'), - ('max_connections', '100', None, 'integer', 'postmaster'), - ('max_prepared_transactions', '0', None, 'integer', 'postmaster'), - ('max_worker_processes', '8', None, 'integer', 'postmaster'), - ('max_locks_per_transaction', '64', None, 'integer', 'postmaster'), - ('max_wal_senders', '5', None, 'integer', 'postmaster'), - ('search_path', 'public', None, 'string', 'user'), - ('port', '5433', None, 'integer', 'postmaster'), - ('listen_addresses', '*', None, 'string', 'postmaster'), - ('autovacuum', 'on', None, 'bool', 'sighup'), - ('unix_socket_directories', '/tmp', None, 'string', 'postmaster')] + self.results = GET_PG_SETTINGS_RESULT elif sql.startswith('SELECT COUNT(*) FROM pg_catalog.pg_settings'): - self.results = [(1,)] + self.results = [(0,)] elif sql.startswith('IDENTIFY_SYSTEM'): self.results = [('1', 3, '0/402EEC0', '')] elif sql.startswith('TIMELINE_HISTORY '): @@ -218,11 +238,11 @@ class PostgresInit(unittest.TestCase): _PARAMETERS = {'wal_level': 'hot_standby', 'max_replication_slots': 5, 'f.oo': 'bar', 'search_path': 'public', 'hot_standby': 'on', 'max_wal_senders': 5, 'wal_keep_segments': 8, 'wal_log_hints': 'on', 'max_locks_per_transaction': 64, - 'max_worker_processes': 8, 'max_connections': 100, 'max_prepared_transactions': 0, + 'max_worker_processes': 8, 'max_connections': 100, 'max_prepared_transactions': 200, 'track_commit_timestamp': 'off', 'unix_socket_directories': '/tmp', - 'trigger_file': 'bla', 'stats_temp_directory': '/tmp', 'zero_damaged_pages': '', + 'trigger_file': 'bla', 'stats_temp_directory': '/tmp', 'zero_damaged_pages': 'off', 'force_parallel_mode': '1', 'constraint_exclusion': '', - 'max_stack_depth': 'Z', 'vacuum_cost_limit': -1, 'vacuum_cost_delay': 200} + 'max_stack_depth': 2048, 'vacuum_cost_limit': -1, 'vacuum_cost_delay': 200} @patch('patroni.psycopg._connect', psycopg_connect) @patch('patroni.postgresql.CallbackExecutor', Mock()) diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 31454479..8a5c491b 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -5,6 +5,7 @@ import re import subprocess import time +from copy import deepcopy from mock import Mock, MagicMock, PropertyMock, patch, mock_open import patroni.psycopg as psycopg @@ -25,7 +26,8 @@ from patroni.postgresql.validator import (ValidatorFactoryNoType, ValidatorFacto from patroni.utils import RetryFailedError from threading import Thread, current_thread -from . import BaseTestPostgresql, MockCursor, MockPostmaster, psycopg_connect, mock_available_gucs +from . import (BaseTestPostgresql, MockCursor, MockPostmaster, psycopg_connect, mock_available_gucs, + GET_PG_SETTINGS_RESULT) mtime_ret = {} @@ -559,31 +561,103 @@ class TestPostgresql(BaseTestPostgresql): @patch('time.sleep', Mock()) @patch.object(Postgresql, 'is_running', Mock(return_value=True)) - def test_reload_config(self): - parameters = self._PARAMETERS.copy() - parameters.pop('f.oo') - parameters['wal_buffers'] = '512' - config = {'pg_hba': [''], 'pg_ident': [''], 'use_unix_socket': True, 'use_unix_socket_repl': True, - 'authentication': {}, - 'retry_timeout': 10, 'listen': '*', 'krbsrvname': 'postgres', 'parameters': parameters} + @patch('patroni.postgresql.config.logger.info') + @patch('patroni.postgresql.config.logger.warning') + def test_reload_config(self, mock_warning, mock_info): + config = deepcopy(self.p.config._config) + + # Nothing changed self.p.reload_config(config) - parameters['b.ar'] = 'bar' - with patch.object(MockCursor, 'fetchall', - Mock(side_effect=[[('wal_block_size', '8191', None, 'integer', 'internal'), - ('wal_segment_size', '2048', '8kB', 'integer', 'internal'), - ('shared_buffers', '16384', '8kB', 'integer', 'postmaster'), - ('wal_buffers', '-1', '8kB', 'integer', 'postmaster'), - ('port', '5433', None, 'integer', 'postmaster')], Exception])): + mock_info.assert_called_once_with('No PostgreSQL configuration items changed, nothing to reload.') + mock_warning.assert_not_called() + self.assertEqual(self.p.pending_restart, False) + + mock_info.reset_mock() + + # Handle wal_buffers + self.p.config._config['parameters']['wal_buffers'] = '512' + self.p.reload_config(config) + mock_info.assert_called_once_with('No PostgreSQL configuration items changed, nothing to reload.') + self.assertEqual(self.p.pending_restart, False) + + mock_info.reset_mock() + config = deepcopy(self.p.config._config) + + # hba/ident_changed + config['pg_hba'] = [''] + config['pg_ident'] = [''] + self.p.reload_config(config) + mock_info.assert_called_once_with('Reloading PostgreSQL configuration.') + self.assertEqual(self.p.pending_restart, False) + + mock_info.reset_mock() + + # Postmaster parameter change (pending_restart) + init_max_worker_processes = config['parameters']['max_worker_processes'] + config['parameters']['max_worker_processes'] *= 2 + with patch('patroni.postgresql.Postgresql._query', Mock(side_effect=[GET_PG_SETTINGS_RESULT, [(1,)]])): self.p.reload_config(config) - parameters['autovacuum'] = 'on' + self.assertEqual(mock_info.call_args_list[0][0], ('Changed %s from %s to %s (restart might be required)', + 'max_worker_processes', str(init_max_worker_processes), + config['parameters']['max_worker_processes'])) + self.assertEqual(mock_info.call_args_list[1][0], ('Reloading PostgreSQL configuration.',)) + self.assertEqual(self.p.pending_restart, True) + + mock_info.reset_mock() + + # Reset to the initial value without restart + config['parameters']['max_worker_processes'] = init_max_worker_processes self.p.reload_config(config) - parameters['autovacuum'] = 'off' - parameters.pop('search_path') - config['listen'] = '*:5433' + self.assertEqual(mock_info.call_args_list[0][0], ('Changed %s from %s to %s', 'max_worker_processes', + init_max_worker_processes * 2, + str(config['parameters']['max_worker_processes']))) + self.assertEqual(mock_info.call_args_list[1][0], ('Reloading PostgreSQL configuration.',)) + self.assertEqual(self.p.pending_restart, False) + + mock_info.reset_mock() + + # User-defined parameter changed (removed) + config['parameters'].pop('f.oo') self.p.reload_config(config) - parameters['unix_socket_directories'] = '.' + self.assertEqual(mock_info.call_args_list[0][0], ('Changed %s from %s to %s', 'f.oo', 'bar', None)) + self.assertEqual(mock_info.call_args_list[1][0], ('Reloading PostgreSQL configuration.',)) + self.assertEqual(self.p.pending_restart, False) + + mock_info.reset_mock() + + # Non-postmaster parameter change + config['parameters']['autovacuum'] = 'off' self.p.reload_config(config) - self.p.config.resolve_connection_addresses() + self.assertEqual(mock_info.call_args_list[0][0], ("Changed %s from %s to %s", 'autovacuum', 'on', 'off')) + self.assertEqual(mock_info.call_args_list[1][0], ('Reloading PostgreSQL configuration.',)) + self.assertEqual(self.p.pending_restart, False) + + config['parameters']['autovacuum'] = 'on' + mock_info.reset_mock() + + # Remove invalid parameter + config['parameters']['invalid'] = 'value' + self.p.reload_config(config) + self.assertEqual(mock_warning.call_args_list[0][0], + ('Removing invalid parameter `%s` from postgresql.parameters', 'invalid')) + config['parameters'].pop('invalid') + + mock_warning.reset_mock() + mock_info.reset_mock() + + # Non-empty result (outside changes) and exception while querying pending_restart parameters + with patch('patroni.postgresql.Postgresql._query', + Mock(side_effect=[GET_PG_SETTINGS_RESULT, [(1,)], GET_PG_SETTINGS_RESULT, Exception])): + self.p.reload_config(config, True) + self.assertEqual(mock_info.call_args_list[0][0], ('Reloading PostgreSQL configuration.',)) + self.assertEqual(self.p.pending_restart, True) + + # Invalid values, just to increase silly coverage in postgresql.validator. + # One day we will have proper tests there. + config['parameters']['autovacuum'] = 'of' # Bool.transform() + config['parameters']['vacuum_cost_limit'] = 'smth' # Number.transform() + self.p.reload_config(config, True) + self.assertEqual(mock_warning.call_args_list[-1][0][0], 'Exception %r when running query') def test_resolve_connection_addresses(self): self.p.config._config['use_unix_socket'] = self.p.config._config['use_unix_socket_repl'] = True From f0719d148c54dfc6a73ef848025e9d162d9c2d39 Mon Sep 17 00:00:00 2001 From: Polina Bungina <27892524+hughcapet@users.noreply.github.com> Date: Wed, 13 Dec 2023 08:40:47 +0100 Subject: [PATCH 09/18] Actually allow failover to an async candidate in sync mode (#2980) --- patroni/ctl.py | 2 +- tests/test_ctl.py | 10 ++++++---- 2 files changed, 7 insertions(+), 5 deletions(-) diff --git a/patroni/ctl.py b/patroni/ctl.py index 3e981b45..3a0d2a17 100644 --- a/patroni/ctl.py +++ b/patroni/ctl.py @@ -1272,7 +1272,7 @@ def _do_failover_or_switchover(action: str, cluster_name: str, group: Optional[i config.is_synchronous_mode, not cluster.sync.is_empty, not cluster.sync.matches(candidate, True))): - if click.confirm(f'Are you sure you want to failover to the asynchronous node {candidate}'): + if not click.confirm(f'Are you sure you want to failover to the asynchronous node {candidate}?'): raise PatroniCtlException('Aborting ' + action) scheduled_at_str = None diff --git a/tests/test_ctl.py b/tests/test_ctl.py index a174b03d..badbeca1 100644 --- a/tests/test_ctl.py +++ b/tests/test_ctl.py @@ -226,18 +226,20 @@ class TestCtl(unittest.TestCase): self.assertIn('Supplying a leader name using this command is deprecated', result.output) failover_func_mock.assert_called_once_with('switchover', 'dummy', None, 'leader', None, False) - # Failover to an async member in sync mode (confirm) cluster = get_cluster_initialized_with_leader(sync=('leader', 'other')) cluster.members.append(Member(0, 'async', 28, {'api_url': 'http://127.0.0.1:8012/patroni'})) cluster.config.data['synchronous_mode'] = True with patch('patroni.dcs.AbstractDCS.get_cluster', Mock(return_value=cluster)): + # Failover to an async member in sync mode (confirm) result = self.runner.invoke(ctl, ['failover', 'dummy', '--group', '0', '--candidate', 'async'], input='y\ny') self.assertIn('Are you sure you want to failover to the asynchronous node async', result.output) + self.assertEqual(result.exit_code, 0) - # Failover to an async member in sync mode (abort) - result = self.runner.invoke(ctl, ['failover', 'dummy', '--group', '0', '--candidate', 'async'], input='N') - self.assertEqual(result.exit_code, 1) + # Failover to an async member in sync mode (abort) + result = self.runner.invoke(ctl, ['failover', 'dummy', '--group', '0', '--candidate', 'async'], input='N') + self.assertEqual(result.exit_code, 1) + self.assertIn('Aborting failover', result.output) @patch('patroni.dynamic_loader.iter_modules', Mock(return_value=['patroni.dcs.dummy', 'patroni.dcs.etcd'])) def test_get_dcs(self): From c1ee99d81da8b35e370350a433c5f5c4889baff5 Mon Sep 17 00:00:00 2001 From: Polina Bungina <27892524+hughcapet@users.noreply.github.com> Date: Mon, 18 Dec 2023 10:44:05 +0100 Subject: [PATCH 10/18] Update PG version in a couple of places (#2986) * All dockerfiles to use PG16 by default * PGVERSION env in the test pipelines to 16.1-1 by default * 11->14 in the dcs-pg mapping for test pipelines * Code comments fixes --- .github/workflows/install_deps.py | 2 +- .github/workflows/mapping.py | 2 +- .github/workflows/run_tests.py | 2 +- .github/workflows/tests.yaml | 2 +- Dockerfile | 2 +- Dockerfile.citus | 4 ++-- kubernetes/Dockerfile | 2 +- kubernetes/Dockerfile.citus | 8 ++++---- patroni/postgresql/postmaster.py | 2 +- 9 files changed, 13 insertions(+), 13 deletions(-) diff --git a/.github/workflows/install_deps.py b/.github/workflows/install_deps.py index 6480f66a..b089bde0 100644 --- a/.github/workflows/install_deps.py +++ b/.github/workflows/install_deps.py @@ -110,7 +110,7 @@ def install_etcd(): def install_postgres(): - version = os.environ.get('PGVERSION', '15.1-1') + version = os.environ.get('PGVERSION', '16.1-1') platform = {'darwin': 'osx', 'win32': 'windows-x64', 'cygwin': 'windows-x64'}[sys.platform] if platform == 'osx': return subprocess.call(['brew', 'install', 'expect', 'postgresql@{0}'.format(version.split('.')[0])]) diff --git a/.github/workflows/mapping.py b/.github/workflows/mapping.py index f75efec4..279438b0 100644 --- a/.github/workflows/mapping.py +++ b/.github/workflows/mapping.py @@ -1 +1 @@ -versions = {'etcd': '9.6', 'etcd3': '16', 'consul': '13', 'exhibitor': '12', 'raft': '11', 'kubernetes': '15'} +versions = {'etcd': '9.6', 'etcd3': '16', 'consul': '13', 'exhibitor': '12', 'raft': '14', 'kubernetes': '15'} diff --git a/.github/workflows/run_tests.py b/.github/workflows/run_tests.py index 9a078e4f..cece186f 100644 --- a/.github/workflows/run_tests.py +++ b/.github/workflows/run_tests.py @@ -30,7 +30,7 @@ def main(): unbuffer = ['timeout', '900', 'unbuffer'] else: if sys.platform == 'darwin': - version = os.environ.get('PGVERSION', '15.1-1') + version = os.environ.get('PGVERSION', '16.1-1') path = '/usr/local/opt/postgresql@{0}/bin:.'.format(version.split('.')[0]) unbuffer = ['unbuffer'] else: diff --git a/.github/workflows/tests.yaml b/.github/workflows/tests.yaml index aafe6b56..f1f55e2a 100644 --- a/.github/workflows/tests.yaml +++ b/.github/workflows/tests.yaml @@ -85,7 +85,7 @@ jobs: env: DCS: ${{ matrix.dcs }} ETCDVERSION: 3.4.23 - PGVERSION: 15.1-1 # for windows and macos + PGVERSION: 16.1-1 # for windows and macos strategy: fail-fast: false matrix: diff --git a/Dockerfile b/Dockerfile index b74cdf24..d4fddfea 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,6 +1,6 @@ ## This Dockerfile is meant to aid in the building and debugging patroni whilst developing on your local machine ## It has all the necessary components to play/debug with a single node appliance, running etcd -ARG PG_MAJOR=15 +ARG PG_MAJOR=16 ARG COMPRESS=false ARG PGHOME=/home/postgres ARG PGDATA=$PGHOME/data diff --git a/Dockerfile.citus b/Dockerfile.citus index 5f0164b4..6f02215b 100644 --- a/Dockerfile.citus +++ b/Dockerfile.citus @@ -1,6 +1,6 @@ ## This Dockerfile is meant to aid in the building and debugging patroni whilst developing on your local machine ## It has all the necessary components to play/debug with a single node appliance, running etcd -ARG PG_MAJOR=15 +ARG PG_MAJOR=16 ARG COMPRESS=false ARG PGHOME=/home/postgres ARG PGDATA=$PGHOME/data @@ -40,7 +40,7 @@ RUN set -ex \ echo "deb [signed-by=/etc/apt/trusted.gpg.d/citusdata_community.gpg] https://packagecloud.io/citusdata/community/debian/ $(lsb_release -cs) main" > /etc/apt/sources.list.d/citusdata_community.list \ && curl -sL https://packagecloud.io/citusdata/community/gpgkey | gpg --dearmor > /etc/apt/trusted.gpg.d/citusdata_community.gpg \ && apt-get update -y \ - && apt-get -y install postgresql-$PG_MAJOR-citus-11.3; \ + && apt-get -y install postgresql-$PG_MAJOR-citus-12.1; \ fi \ \ # Cleanup all locales but en_US.UTF-8 diff --git a/kubernetes/Dockerfile b/kubernetes/Dockerfile index 29a683bd..e41bf1cd 100644 --- a/kubernetes/Dockerfile +++ b/kubernetes/Dockerfile @@ -1,4 +1,4 @@ -FROM postgres:15 +FROM postgres:16 LABEL maintainer="Alexander Kukushkin " RUN export DEBIAN_FRONTEND=noninteractive \ diff --git a/kubernetes/Dockerfile.citus b/kubernetes/Dockerfile.citus index f9564521..7af9e5ae 100644 --- a/kubernetes/Dockerfile.citus +++ b/kubernetes/Dockerfile.citus @@ -1,4 +1,4 @@ -FROM postgres:15 +FROM postgres:16 LABEL maintainer="Alexander Kukushkin " RUN export DEBIAN_FRONTEND=noninteractive \ @@ -11,7 +11,7 @@ RUN export DEBIAN_FRONTEND=noninteractive \ ## Make sure we have a en_US.UTF-8 locale available && localedef -i en_US -c -f UTF-8 -A /usr/share/locale/locale.alias en_US.UTF-8 \ && if [ $(dpkg --print-architecture) = 'arm64' ]; then \ - apt-get install -y postgresql-server-dev-15 \ + apt-get install -y postgresql-server-dev-16 \ gcc make autoconf \ libc6-dev flex libcurl4-gnutls-dev \ libicu-dev libkrb5-dev liblz4-dev \ @@ -24,7 +24,7 @@ RUN export DEBIAN_FRONTEND=noninteractive \ echo "deb [signed-by=/etc/apt/trusted.gpg.d/citusdata_community.gpg] https://packagecloud.io/citusdata/community/debian/ $(lsb_release -cs) main" > /etc/apt/sources.list.d/citusdata_community.list \ && curl -sL https://packagecloud.io/citusdata/community/gpgkey | gpg --dearmor > /etc/apt/trusted.gpg.d/citusdata_community.gpg \ && apt-get update -y \ - && apt-get -y install postgresql-15-citus-12.0; \ + && apt-get -y install postgresql-16-citus-12.1; \ fi \ && pip3 install --break-system-packages setuptools \ && pip3 install --break-system-packages 'git+https://github.com/zalando/patroni.git#egg=patroni[kubernetes]' \ @@ -38,7 +38,7 @@ RUN export DEBIAN_FRONTEND=noninteractive \ && chmod 664 /etc/passwd \ # Clean up && apt-get remove -y git python3-pip python3-wheel \ - postgresql-server-dev-15 gcc make autoconf \ + postgresql-server-dev-16 gcc make autoconf \ libc6-dev flex libicu-dev libkrb5-dev liblz4-dev \ libpam0g-dev libreadline-dev libselinux1-dev libssl-dev libxslt1-dev libzstd-dev uuid-dev \ && apt-get autoremove -y \ diff --git a/patroni/postgresql/postmaster.py b/patroni/postgresql/postmaster.py index 4505e7f7..97eb10e4 100644 --- a/patroni/postgresql/postmaster.py +++ b/patroni/postgresql/postmaster.py @@ -176,7 +176,7 @@ class PostmasterProcess(psutil.Process): return not self.is_running() def wait_for_user_backends_to_close(self, stop_timeout: Optional[float]) -> None: - # These regexps are cross checked against versions PostgreSQL 9.1 .. 15 + # These regexps are cross checked against versions PostgreSQL 9.1 .. 16 aux_proc_re = re.compile("(?:postgres:)( .*:)? (?:(?:archiver|startup|autovacuum launcher|autovacuum worker|" "checkpointer|logger|stats collector|wal receiver|wal writer|writer)(?: process )?|" "walreceiver|wal sender process|walsender|walwriter|background writer|" From 206ee91b07eed4c405cfabb724105df974038a99 Mon Sep 17 00:00:00 2001 From: Polina Bungina <27892524+hughcapet@users.noreply.github.com> Date: Wed, 20 Dec 2023 09:54:04 +0100 Subject: [PATCH 11/18] Exclude leader from failover candidates in ctl (#2983) Exclude actual leader (not the passed leader argument) from the candidates list in the `patronictl failover` prompt. Abort `patronictl failover` execution if candidate specified is the same as the current cluster leader --- patroni/ctl.py | 36 +++++++++++++++++++----------------- tests/test_ctl.py | 8 +++++++- 2 files changed, 26 insertions(+), 18 deletions(-) diff --git a/patroni/ctl.py b/patroni/ctl.py index 3a0d2a17..de1fe892 100644 --- a/patroni/ctl.py +++ b/patroni/ctl.py @@ -1185,7 +1185,7 @@ def reinit(cluster_name: str, group: Optional[int], member_names: List[str], for def _do_failover_or_switchover(action: str, cluster_name: str, group: Optional[int], - leader: Optional[str], candidate: Optional[str], + switchover_leader: Optional[str], candidate: Optional[str], force: bool, scheduled: Optional[str] = None) -> None: """Perform a failover or a switchover operation in the cluster. @@ -1199,7 +1199,7 @@ def _do_failover_or_switchover(action: str, cluster_name: str, group: Optional[i :param cluster_name: name of the Patroni cluster. :param group: filter Citus group within we should perform a failover or switchover. If ``None``, user will be prompted for filling it -- unless *force* is ``True``, in which case an exception is raised. - :param leader: name of the current leader member. + :param switchover_leader: name of the leader member passed as switchover option. :param candidate: name of a standby member to be promoted. Nodes that are tagged with ``nofailover`` cannot be used. :param force: perform the failover or switchover without asking for confirmations. :param scheduled: timestamp when the switchover should be scheduled to occur. If ``now`` perform immediately. @@ -1208,10 +1208,11 @@ def _do_failover_or_switchover(action: str, cluster_name: str, group: Optional[i :class:`PatroniCtlException`: if: * Patroni is running on a Citus cluster, but no *group* was specified; or * a switchover was requested by the cluster has no leader; or - * *leader* does not match the current leader of the cluster; or + * *switchover_leader* does not match the current leader of the cluster; or * cluster has no candidates available for the operation; or * no *candidate* is given for a failover operation; or - * *leader* and *candidate* are the same; or + * current leader and *candidate* are the same; or + * *candidate* is tagged as nofailover; or * *candidate* is not a member of the cluster; or * trying to schedule a switchover in a cluster that is in maintenance mode; or * user aborts the operation. @@ -1231,23 +1232,24 @@ def _do_failover_or_switchover(action: str, cluster_name: str, group: Optional[i config = global_config.from_cluster(cluster) + cluster_leader = cluster.leader and cluster.leader.name # leader has to be be defined for switchover only if action == 'switchover': - if cluster.leader is None or not cluster.leader.name: + if not cluster_leader: raise PatroniCtlException('This cluster has no leader') - if leader is None: + if switchover_leader is None: if force: - leader = cluster.leader.name + switchover_leader = cluster_leader else: prompt = 'Standby Leader' if config.is_standby_cluster else 'Primary' - leader = click.prompt(prompt, type=str, default=(cluster.leader and cluster.leader.name)) + switchover_leader = click.prompt(prompt, type=str, default=cluster_leader) - if cluster.leader.name != leader: - raise PatroniCtlException(f'Member {leader} is not the leader of cluster {cluster_name}') + if cluster_leader != switchover_leader: + raise PatroniCtlException(f'Member {switchover_leader} is not the leader of cluster {cluster_name}') # excluding members with nofailover tag - candidate_names = [str(m.name) for m in cluster.members if m.name != leader and not m.nofailover] + candidate_names = [str(m.name) for m in cluster.members if m.name != cluster_leader and not m.nofailover] # We sort the names for consistent output to the client candidate_names.sort() @@ -1260,10 +1262,10 @@ def _do_failover_or_switchover(action: str, cluster_name: str, group: Optional[i if action == 'failover' and not candidate: raise PatroniCtlException('Failover could be performed only to a specific candidate') - if candidate == leader: - raise PatroniCtlException(action.title() + ' target and source are the same.') - if candidate and candidate not in candidate_names: + if candidate == cluster_leader: + raise PatroniCtlException( + f'Member {candidate} is already the leader of cluster {cluster_name}') raise PatroniCtlException( f'Member {candidate} does not exist in cluster {cluster_name} or is tagged as nofailover') @@ -1292,7 +1294,7 @@ def _do_failover_or_switchover(action: str, cluster_name: str, group: Optional[i failover_value = {'candidate': candidate} if action == 'switchover': - failover_value['leader'] = leader + failover_value['leader'] = switchover_leader if scheduled_at_str: failover_value['scheduled_at'] = scheduled_at_str @@ -1300,7 +1302,7 @@ def _do_failover_or_switchover(action: str, cluster_name: str, group: Optional[i # By now we have established that the leader exists and the candidate exists if not force: - demote_msg = f', demoting current leader {cluster.leader.name}' if cluster.leader else '' + demote_msg = f', demoting current leader {cluster_leader}' if cluster_leader else '' if scheduled_at_str: # only switchover can be scheduled if not click.confirm(f'Are you sure you want to schedule switchover of cluster ' @@ -1334,7 +1336,7 @@ def _do_failover_or_switchover(action: str, cluster_name: str, group: Optional[i logging.exception(r) logging.warning('Failing over to DCS') click.echo('{0} Could not {1} using Patroni api, falling back to DCS'.format(timestamp(), action)) - dcs.manual_failover(leader, candidate, scheduled_at=scheduled_at) + dcs.manual_failover(switchover_leader, candidate, scheduled_at=scheduled_at) output_members(cluster, cluster_name, group=group) diff --git a/tests/test_ctl.py b/tests/test_ctl.py index badbeca1..f9ee62ce 100644 --- a/tests/test_ctl.py +++ b/tests/test_ctl.py @@ -157,7 +157,8 @@ class TestCtl(unittest.TestCase): # Target and source are equal result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='leader\nleader\n\ny') self.assertEqual(result.exit_code, 1) - self.assertIn('Switchover target and source are the same', result.output) + self.assertIn("Candidate ['other']", result.output) + self.assertIn('Member leader is already the leader of cluster dummy', result.output) # Candidate is not a member of the cluster result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='leader\nReality\n\ny') @@ -220,6 +221,11 @@ class TestCtl(unittest.TestCase): result = self.runner.invoke(ctl, ['failover', 'dummy'], input='0\n') self.assertIn('Failover could be performed only to a specific candidate', result.output) + # Candidate is the same as the leader + result = self.runner.invoke(ctl, ['failover', 'dummy', '--group', '0'], input='leader\n') + self.assertIn("Candidate ['other']", result.output) + self.assertIn('Member leader is already the leader of cluster dummy', result.output) + # Temp test to check a fallback to switchover if leader is specified with patch('patroni.ctl._do_failover_or_switchover') as failover_func_mock: result = self.runner.invoke(ctl, ['failover', '--leader', 'leader', 'dummy'], input='0\n') From 5c3e1a693e2217e990ae42cd27c1d9795df38d68 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 20 Dec 2023 10:49:33 +0100 Subject: [PATCH 12/18] Implement validation of the `log` section (#2989) Somehow it was always forgotten. --- patroni/validator.py | 12 ++++++++++++ tests/test_validator.py | 14 ++++++++++++++ 2 files changed, 26 insertions(+) diff --git a/patroni/validator.py b/patroni/validator.py index d0b168be..bd69cc0e 100644 --- a/patroni/validator.py +++ b/patroni/validator.py @@ -937,6 +937,18 @@ validate_etcd = { schema = Schema({ "name": str, "scope": str, + Optional("log"): { + Optional("level"): EnumValidator(('DEBUG', 'INFO', 'WARN', 'WARNING', 'ERROR', 'FATAL', 'CRITICAL'), + case_sensitive=True, raise_assert=True), + Optional("traceback_level"): EnumValidator(('DEBUG', 'ERROR'), raise_assert=True), + Optional("format"): str, + Optional("dateformat"): str, + Optional("max_queue_size"): int, + Optional("dir"): str, + Optional("file_num"): int, + Optional("file_size"): int, + Optional("loggers"): dict + }, Optional("ctl"): { Optional("insecure"): bool, Optional("cacert"): str, diff --git a/tests/test_validator.py b/tests/test_validator.py index 1f647dbe..d5e82992 100644 --- a/tests/test_validator.py +++ b/tests/test_validator.py @@ -13,6 +13,20 @@ available_dcs = [m.split(".")[-1] for m in dcs_modules()] config = { "name": "string", "scope": "string", + "log": { + "level": "DEBUG", + "traceback_level": "DEBUG", + "format": "%(asctime)s %(levelname)s: %(message)s", + "dateformat": "%Y-%m-%d %H:%M:%S", + "max_queue_size": 100, + "dir": "/tmp", + "file_num": 10, + "file_size": 1000000, + "loggers": { + "patroni.postmaster": "WARNING", + "urllib3": "DEBUG" + } + }, "restapi": { "listen": "127.0.0.2:800", "connect_address": "127.0.0.2:800", From bcfd8438a50a1f36b65be76ee2d17d1da3b903e3 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Thu, 21 Dec 2023 08:58:26 +0100 Subject: [PATCH 13/18] Abstract CitusHandler and decouple it from configuration (#2950) the main issue was that the configuration for Citus handler and for DCS existed in two places, while ideally AbstractDCS should not know many details about what kind of MPP is in use. To solve the problem we first dynamically create an object implementing AbstractMPP interfaces, which is a configuration for DCS. Later this object is used to instantiate the class implementing AbstractMPPHandler interface. This is just a starting point, which does some heavy lifting. As a next steps all kind of variables named after Citus in files different from patroni/postgres/mpp/citus.py should be renamed. In other words this commit takes over the most complex part of #2940, which was never implemented. Co-authored-by: zhjwpku --- patroni/__main__.py | 2 +- patroni/api.py | 4 +- patroni/ctl.py | 8 +- patroni/dcs/__init__.py | 33 +-- patroni/dcs/consul.py | 9 +- patroni/dcs/etcd.py | 13 +- patroni/dcs/etcd3.py | 10 +- patroni/dcs/exhibitor.py | 5 +- patroni/dcs/kubernetes.py | 20 +- patroni/dcs/raft.py | 10 +- patroni/dcs/zookeeper.py | 10 +- patroni/ha.py | 4 +- patroni/postgresql/__init__.py | 6 +- patroni/postgresql/mpp/__init__.py | 296 ++++++++++++++++++++++++++ patroni/postgresql/{ => mpp}/citus.py | 103 ++++++--- patroni/postgresql/slots.py | 2 +- tests/__init__.py | 36 ++-- tests/test_citus.py | 34 ++- tests/test_consul.py | 22 +- tests/test_ctl.py | 11 +- tests/test_etcd.py | 11 +- tests/test_etcd3.py | 10 +- tests/test_exhibitor.py | 6 +- tests/test_ha.py | 22 +- tests/test_kubernetes.py | 17 +- tests/test_mpp.py | 52 +++++ tests/test_raft.py | 21 +- tests/test_zookeeper.py | 10 +- 28 files changed, 602 insertions(+), 185 deletions(-) create mode 100644 patroni/postgresql/mpp/__init__.py rename patroni/postgresql/{ => mpp}/citus.py (84%) create mode 100644 tests/test_mpp.py diff --git a/patroni/__main__.py b/patroni/__main__.py index 229ccfb9..2c253da4 100644 --- a/patroni/__main__.py +++ b/patroni/__main__.py @@ -68,7 +68,7 @@ class Patroni(AbstractPatroniDaemon, Tags): self.watchdog = Watchdog(self.config) self.load_dynamic_configuration() - self.postgresql = Postgresql(self.config['postgresql']) + self.postgresql = Postgresql(self.config['postgresql'], self.dcs.mpp) self.api = RestApiServer(self, self.config['restapi']) self.ha = Ha(self) diff --git a/patroni/api.py b/patroni/api.py index 5761d359..5c9d6603 100644 --- a/patroni/api.py +++ b/patroni/api.py @@ -1153,8 +1153,8 @@ class RestApiHandler(BaseHTTPRequestHandler): def do_POST_citus(self) -> None: """Handle a ``POST`` request to ``/citus`` path. - Call :func:`~patroni.postgresql.CitusHandler.handle_event` to handle the request, then write a response with - HTTP status code ``200``. + Call :func:`~patroni.postgresql.mpp.AbstractMPPHandler.handle_event` to handle the request, + then write a response with HTTP status code ``200``. .. note:: If unable to parse the request body, then the request is silently discarded. diff --git a/patroni/ctl.py b/patroni/ctl.py index de1fe892..6a5fe0e6 100644 --- a/patroni/ctl.py +++ b/patroni/ctl.py @@ -51,6 +51,7 @@ from .config import Config from .dcs import get_dcs as _get_dcs, AbstractDCS, Cluster, Member from .exceptions import PatroniException from .postgresql.misc import postgres_version_to_int +from .postgresql.mpp import get_mpp from .utils import cluster_as_json, patch_config, polling_loop from .request import PatroniRequest from .version import __version__ @@ -313,7 +314,7 @@ def ctl(ctx: click.Context, config_file: str, dcs_url: Optional[str], insecure: config = load_config(config_file, dcs_url) # backward compatibility for configuration file where ctl section is not defined config.setdefault('ctl', {})['insecure'] = config.get('ctl', {}).get('insecure') or insecure - ctx.obj = {'__config': config} + ctx.obj = {'__config': config, '__mpp': get_mpp(config)} def is_citus_cluster() -> bool: @@ -321,7 +322,7 @@ def is_citus_cluster() -> bool: :returns: ``True`` if configuration has ``citus`` section, otherwise ``False``. """ - return bool(_get_configuration().get('citus')) + return click.get_current_context().obj['__mpp'].is_enabled() def get_dcs(scope: str, group: Optional[int]) -> AbstractDCS: @@ -340,12 +341,13 @@ def get_dcs(scope: str, group: Optional[int]) -> AbstractDCS: config = _get_configuration() config.update({'scope': scope, 'patronictl': True}) if group is not None: - config['citus'] = {'group': group} + config['citus'] = {'group': group, 'database': 'postgres'} config.setdefault('name', scope) try: dcs = _get_dcs(config) if is_citus_cluster() and group is None: dcs.is_citus_coordinator = lambda: True + click.get_current_context().obj['__mpp'] = dcs.mpp return dcs except PatroniException as e: raise PatroniCtlException(str(e)) diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index 28c3734f..a210615b 100644 --- a/patroni/dcs/__init__.py +++ b/patroni/dcs/__init__.py @@ -25,10 +25,9 @@ from ..utils import parse_int if TYPE_CHECKING: # pragma: no cover from ..config import Config from ..postgresql import Postgresql + from ..postgresql.mpp import AbstractMPP SLOT_ADVANCE_AVAILABLE_VERSION = 110000 -CITUS_COORDINATOR_GROUP_ID = 0 -citus_group_re = re.compile('^(0|[1-9][0-9]*)$') slot_name_re = re.compile('^[a-z0-9_]{1,63}$') logger = logging.getLogger(__name__) @@ -130,10 +129,9 @@ def get_dcs(config: Union['Config', Dict[str, Any]]) -> 'AbstractDCS': p: config[p] for p in ('namespace', 'name', 'scope', 'loop_wait', 'patronictl', 'ttl', 'retry_timeout') if p in config}) - # From citus section we only need "group" parameter, but will propagate everything just in case. - if isinstance(config.get('citus'), dict): - config[name].update(config['citus']) - return dcs_class(config[name]) + + from patroni.postgresql.mpp import get_mpp + return dcs_class(config[name], get_mpp(config)) available_implementations = ', '.join(sorted([n for n, _ in iter_dcs_classes()])) raise PatroniFatalException("Can not find suitable configuration of distributed configuration store\n" @@ -1338,15 +1336,15 @@ class AbstractDCS(abc.ABC): _SYNC = 'sync' _FAILSAFE = 'failsafe' - def __init__(self, config: Dict[str, Any]) -> None: + def __init__(self, config: Dict[str, Any], mpp: 'AbstractMPP') -> None: """Prepare DCS paths, Citus group ID, initial values for state information and processing dependencies. :ivar config: :class:`dict`, reference to config section of selected DCS. i.e.: ``zookeeper`` for zookeeper, ``etcd`` for etcd, etc... """ + self._mpp = mpp self._name = config['name'] self._base_path = re.sub('/+', '/', '/'.join(['', config.get('namespace', 'service'), config['scope']])) - self._citus_group = str(config['group']) if isinstance(config.get('group'), int) else None self._set_loop_wait(config.get('loop_wait', 10)) self._ctl = bool(config.get('patronictl', False)) @@ -1359,6 +1357,11 @@ class AbstractDCS(abc.ABC): self._last_failsafe: Optional[Dict[str, str]] = {} self.event = Event() + @property + def mpp(self) -> 'AbstractMPP': + """Get the effective underlying MPP, if any has been configured.""" + return self._mpp + def client_path(self, path: str) -> str: """Construct the absolute key name from appropriate parts for the DCS type. @@ -1367,8 +1370,8 @@ class AbstractDCS(abc.ABC): :returns: absolute key name for the current Patroni cluster. """ components = [self._base_path] - if self._citus_group: - components.append(self._citus_group) + if self._mpp.is_enabled(): + components.append(str(self._mpp.group)) components.append(path.lstrip('/')) return '/'.join(components) @@ -1522,9 +1525,9 @@ class AbstractDCS(abc.ABC): def is_citus_coordinator(self) -> bool: """:class:`Cluster` instance has a Citus Coordinator group ID. - :returns: ``True`` if the given node is running as Citus Coordinator (``group=0``). + :returns: ``True`` if the given node is running as the MPP Coordinator. """ - return self._citus_group == str(CITUS_COORDINATOR_GROUP_ID) + return self._mpp.is_coordinator() def get_citus_coordinator(self) -> Optional[Cluster]: """Load the Patroni cluster for the Citus Coordinator. @@ -1532,10 +1535,10 @@ class AbstractDCS(abc.ABC): .. note:: This method is only executed on the worker nodes (``group!=0``) to find the coordinator. - :returns: Select :class:`Cluster` instance associated with the Citus Coordinator group ID. + :returns: Select :class:`Cluster` instance associated with the MPP Coordinator group ID. """ try: - return self.__get_patroni_cluster(f'{self._base_path}/{CITUS_COORDINATOR_GROUP_ID}/') + return self.__get_patroni_cluster(f'{self._base_path}/{self._mpp.coordinator_group_id}/') except Exception as e: logger.error('Failed to load Citus coordinator cluster from %s: %r', self.__class__.__name__, e) return None @@ -1549,7 +1552,7 @@ class AbstractDCS(abc.ABC): groups = self._load_cluster(self._base_path + '/', self._citus_cluster_loader) if TYPE_CHECKING: # pragma: no cover assert isinstance(groups, dict) - cluster = groups.pop(CITUS_COORDINATOR_GROUP_ID, Cluster.empty()) + cluster = groups.pop(self._mpp.coordinator_group_id, Cluster.empty()) cluster.workers.update(groups) return cluster diff --git a/patroni/dcs/consul.py b/patroni/dcs/consul.py index fe66c6d8..19d65306 100644 --- a/patroni/dcs/consul.py +++ b/patroni/dcs/consul.py @@ -16,8 +16,9 @@ from urllib.parse import urlencode, urlparse, quote from typing import Any, Callable, Dict, List, Mapping, NamedTuple, Optional, Union, Tuple, TYPE_CHECKING from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, Status, SyncState, \ - TimelineHistory, ReturnFalseException, catch_return_false_exception, citus_group_re + TimelineHistory, ReturnFalseException, catch_return_false_exception from ..exceptions import DCSError +from ..postgresql.mpp import AbstractMPP from ..utils import deep_compare, parse_bool, Retry, RetryFailedError, split_host_port, uri, USER_AGENT if TYPE_CHECKING: # pragma: no cover from ..config import Config @@ -232,8 +233,8 @@ def service_name_from_scope_name(scope_name: str) -> str: class Consul(AbstractDCS): - def __init__(self, config: Dict[str, Any]) -> None: - super(Consul, self).__init__(config) + def __init__(self, config: Dict[str, Any], mpp: AbstractMPP) -> None: + super(Consul, self).__init__(config, mpp) self._base_path = self._base_path[1:] self._scope = config['scope'] self._session = None @@ -435,7 +436,7 @@ class Consul(AbstractDCS): clusters: Dict[int, Dict[str, Cluster]] = defaultdict(dict) for node in results or []: key = node['Key'][len(path):].split('/', 1) - if len(key) == 2 and citus_group_re.match(key[0]): + if len(key) == 2 and self._mpp.group_re.match(key[0]): node['Value'] = (node['Value'] or b'').decode('utf-8') clusters[int(key[0])][key[1]] = node return {group: self._cluster_from_nodes(nodes) for group, nodes in clusters.items()} diff --git a/patroni/dcs/etcd.py b/patroni/dcs/etcd.py index 3be699a6..b9d3e0ab 100644 --- a/patroni/dcs/etcd.py +++ b/patroni/dcs/etcd.py @@ -22,8 +22,9 @@ from urllib3 import Timeout from urllib3.exceptions import HTTPError, ReadTimeoutError, ProtocolError from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, Status, SyncState, \ - TimelineHistory, ReturnFalseException, catch_return_false_exception, citus_group_re + TimelineHistory, ReturnFalseException, catch_return_false_exception from ..exceptions import DCSError +from ..postgresql.mpp import AbstractMPP from ..request import get as requests_get from ..utils import Retry, RetryFailedError, split_host_port, uri, USER_AGENT if TYPE_CHECKING: # pragma: no cover @@ -470,9 +471,9 @@ class EtcdClient(AbstractEtcdClientWithFailover): class AbstractEtcd(AbstractDCS): - def __init__(self, config: Dict[str, Any], client_cls: Type[AbstractEtcdClientWithFailover], + def __init__(self, config: Dict[str, Any], mpp: AbstractMPP, client_cls: Type[AbstractEtcdClientWithFailover], retry_errors_cls: Union[Type[Exception], Tuple[Type[Exception], ...]]) -> None: - super(AbstractEtcd, self).__init__(config) + super(AbstractEtcd, self).__init__(config, mpp) self._retry = Retry(deadline=config['retry_timeout'], max_delay=1, max_tries=-1, retry_exceptions=retry_errors_cls) self._ttl = int(config.get('ttl') or 30) @@ -645,8 +646,8 @@ def catch_etcd_errors(func: Callable[..., Any]) -> Any: class Etcd(AbstractEtcd): - def __init__(self, config: Dict[str, Any]) -> None: - super(Etcd, self).__init__(config, EtcdClient, (etcd.EtcdLeaderElectionInProgress, EtcdRaftInternal)) + def __init__(self, config: Dict[str, Any], mpp: AbstractMPP) -> None: + super(Etcd, self).__init__(config, mpp, EtcdClient, (etcd.EtcdLeaderElectionInProgress, EtcdRaftInternal)) self.__do_not_watch = False @property @@ -726,7 +727,7 @@ class Etcd(AbstractEtcd): clusters: Dict[int, Dict[str, etcd.EtcdResult]] = defaultdict(dict) for node in result.leaves: key = node.key[len(result.key):].lstrip('/').split('/', 1) - if len(key) == 2 and citus_group_re.match(key[0]): + if len(key) == 2 and self._mpp.group_re.match(key[0]): clusters[int(key[0])][key[1]] = node return {group: self._cluster_from_nodes(result.etcd_index, nodes) for group, nodes in clusters.items()} diff --git a/patroni/dcs/etcd3.py b/patroni/dcs/etcd3.py index ea7e52f2..7cc2a115 100644 --- a/patroni/dcs/etcd3.py +++ b/patroni/dcs/etcd3.py @@ -16,9 +16,10 @@ from threading import Condition, Lock, Thread from typing import Any, Callable, Collection, Dict, Iterator, List, Optional, Tuple, Type, TYPE_CHECKING, Union from . import ClusterConfig, Cluster, Failover, Leader, Member, Status, SyncState, \ - TimelineHistory, catch_return_false_exception, citus_group_re + TimelineHistory, catch_return_false_exception from .etcd import AbstractEtcdClientWithFailover, AbstractEtcd, catch_etcd_errors, DnsCachingResolver, Retry from ..exceptions import DCSError, PatroniException +from ..postgresql.mpp import AbstractMPP from ..utils import deep_compare, enable_keepalive, iter_response_objects, RetryFailedError, USER_AGENT logger = logging.getLogger(__name__) @@ -671,8 +672,9 @@ class PatroniEtcd3Client(Etcd3Client): class Etcd3(AbstractEtcd): - def __init__(self, config: Dict[str, Any]) -> None: - super(Etcd3, self).__init__(config, PatroniEtcd3Client, (DeadlineExceeded, Unavailable, FailedPrecondition)) + def __init__(self, config: Dict[str, Any], mpp: AbstractMPP) -> None: + super(Etcd3, self).__init__(config, mpp, PatroniEtcd3Client, + (DeadlineExceeded, Unavailable, FailedPrecondition)) self.__do_not_watch = False self._lease = None self._last_lease_refresh = 0 @@ -796,7 +798,7 @@ class Etcd3(AbstractEtcd): path = self._base_path + '/' for node in self._client.get_cluster(path): key = node['key'][len(path):].split('/', 1) - if len(key) == 2 and citus_group_re.match(key[0]): + if len(key) == 2 and self._mpp.group_re.match(key[0]): clusters[int(key[0])][key[1]] = node return {group: self._cluster_from_nodes(nodes) for group, nodes in clusters.items()} diff --git a/patroni/dcs/exhibitor.py b/patroni/dcs/exhibitor.py index 2b06073b..03d23575 100644 --- a/patroni/dcs/exhibitor.py +++ b/patroni/dcs/exhibitor.py @@ -7,6 +7,7 @@ from typing import Any, Callable, Dict, List, Union from . import Cluster from .zookeeper import ZooKeeper +from ..postgresql.mpp import AbstractMPP from ..request import get as requests_get from ..utils import uri @@ -66,10 +67,10 @@ class ExhibitorEnsembleProvider(object): class Exhibitor(ZooKeeper): - def __init__(self, config: Dict[str, Any]) -> None: + def __init__(self, config: Dict[str, Any], mpp: AbstractMPP) -> None: interval = config.get('poll_interval', 300) self._ensemble_provider = ExhibitorEnsembleProvider(config['hosts'], config['port'], poll_interval=interval) - super(Exhibitor, self).__init__({**config, 'hosts': self._ensemble_provider.zookeeper_hosts}) + super(Exhibitor, self).__init__({**config, 'hosts': self._ensemble_provider.zookeeper_hosts}, mpp) def _load_cluster( self, path: str, loader: Callable[[str], Union[Cluster, Dict[int, Cluster]]] diff --git a/patroni/dcs/kubernetes.py b/patroni/dcs/kubernetes.py index aee87bd3..343b496c 100644 --- a/patroni/dcs/kubernetes.py +++ b/patroni/dcs/kubernetes.py @@ -19,9 +19,9 @@ from urllib3.exceptions import HTTPError from threading import Condition, Lock, Thread from typing import Any, Callable, Collection, Dict, List, Optional, Tuple, Type, Union, TYPE_CHECKING -from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, Status, SyncState, \ - TimelineHistory, CITUS_COORDINATOR_GROUP_ID, citus_group_re +from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, Status, SyncState, TimelineHistory from ..exceptions import DCSError +from ..postgresql.mpp import AbstractMPP from ..utils import deep_compare, iter_response_objects, keepalive_socket_options, \ Retry, RetryFailedError, tzutc, uri, USER_AGENT if TYPE_CHECKING: # pragma: no cover @@ -748,7 +748,7 @@ class Kubernetes(AbstractDCS): _CITUS_LABEL = 'citus-group' - def __init__(self, config: Dict[str, Any]) -> None: + def __init__(self, config: Dict[str, Any], mpp: AbstractMPP) -> None: self._labels = deepcopy(config['labels']) self._labels[config.get('scope_label', 'cluster-name')] = config['scope'] self._label_selector = ','.join('{0}={1}'.format(k, v) for k, v in self._labels.items()) @@ -759,9 +759,9 @@ class Kubernetes(AbstractDCS): self._standby_leader_label_value = config.get('standby_leader_label_value', 'master') self._tmp_role_label = config.get('tmp_role_label') self._ca_certs = os.environ.get('PATRONI_KUBERNETES_CACERT', config.get('cacert')) or SERVICE_CERT_FILENAME - super(Kubernetes, self).__init__({**config, 'namespace': ''}) - if self._citus_group: - self._labels[self._CITUS_LABEL] = self._citus_group + super(Kubernetes, self).__init__({**config, 'namespace': ''}, mpp) + if self._mpp.is_enabled(): + self._labels[self._CITUS_LABEL] = str(self._mpp.group) self._retry = Retry(deadline=config['retry_timeout'], max_delay=1, max_tries=-1, retry_exceptions=KubernetesRetriableException) @@ -944,12 +944,12 @@ class Kubernetes(AbstractDCS): for name, pod in path['pods'].items(): group = pod.metadata.labels.get(self._CITUS_LABEL) - if group and citus_group_re.match(group): + if group and self._mpp.group_re.match(group): clusters[group]['pods'][name] = pod for name, kind in path['nodes'].items(): group = kind.metadata.labels.get(self._CITUS_LABEL) - if group and citus_group_re.match(group): + if group and self._mpp.group_re.match(group): clusters[group]['nodes'][name] = kind return {int(group): self._cluster_from_nodes(group, value['nodes'], value['pods'].values()) for group, value in clusters.items()} @@ -976,12 +976,12 @@ class Kubernetes(AbstractDCS): def _load_cluster( self, path: str, loader: Callable[[Any], Union[Cluster, Dict[int, Cluster]]] ) -> Union[Cluster, Dict[int, Cluster]]: - group = self._citus_group if path == self.client_path('') else None + group = str(self._mpp.group) if self._mpp.is_enabled() and path == self.client_path('') else None return self.__load_cluster(group, loader) def get_citus_coordinator(self) -> Optional[Cluster]: try: - ret = self.__load_cluster(str(CITUS_COORDINATOR_GROUP_ID), self._cluster_loader) + ret = self.__load_cluster(str(self._mpp.coordinator_group_id), self._cluster_loader) if TYPE_CHECKING: # pragma: no cover assert isinstance(ret, Cluster) return ret diff --git a/patroni/dcs/raft.py b/patroni/dcs/raft.py index 98c48f44..0528cfb0 100644 --- a/patroni/dcs/raft.py +++ b/patroni/dcs/raft.py @@ -12,9 +12,9 @@ from pysyncobj.transport import TCPTransport, CONNECTION_STATE from pysyncobj.utility import TcpUtility from typing import Any, Callable, Collection, Dict, List, Optional, Set, Union, TYPE_CHECKING -from . import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, Status, SyncState, \ - TimelineHistory, citus_group_re +from . import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, Status, SyncState, TimelineHistory from ..exceptions import DCSError +from ..postgresql.mpp import AbstractMPP from ..utils import validate_directory if TYPE_CHECKING: # pragma: no cover from ..config import Config @@ -285,8 +285,8 @@ class KVStoreTTL(DynMemberSyncObj): class Raft(AbstractDCS): - def __init__(self, config: Dict[str, Any]) -> None: - super(Raft, self).__init__(config) + def __init__(self, config: Dict[str, Any], mpp: AbstractMPP) -> None: + super(Raft, self).__init__(config, mpp) self._ttl = int(config.get('ttl') or 30) ready_event = threading.Event() @@ -387,7 +387,7 @@ class Raft(AbstractDCS): response = self._sync_obj.get(path, recursive=True) for key, value in (response or {}).items(): key = key[len(path):].split('/', 1) - if len(key) == 2 and citus_group_re.match(key[0]): + if len(key) == 2 and self._mpp.group_re.match(key[0]): clusters[int(key[0])][key[1]] = value return {group: self._cluster_from_nodes(nodes) for group, nodes in clusters.items()} diff --git a/patroni/dcs/zookeeper.py b/patroni/dcs/zookeeper.py index 3704b579..6bf77ae4 100644 --- a/patroni/dcs/zookeeper.py +++ b/patroni/dcs/zookeeper.py @@ -12,9 +12,9 @@ from kazoo.retry import RetryFailedError from kazoo.security import ACL, make_acl from typing import Any, Callable, Dict, List, Optional, Union, Tuple, TYPE_CHECKING -from . import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, Status, SyncState, \ - TimelineHistory, citus_group_re +from . import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, Status, SyncState, TimelineHistory from ..exceptions import DCSError +from ..postgresql.mpp import AbstractMPP from ..utils import deep_compare if TYPE_CHECKING: # pragma: no cover from ..config import Config @@ -87,8 +87,8 @@ class PatroniKazooClient(KazooClient): class ZooKeeper(AbstractDCS): - def __init__(self, config: Dict[str, Any]) -> None: - super(ZooKeeper, self).__init__(config) + def __init__(self, config: Dict[str, Any], mpp: AbstractMPP) -> None: + super(ZooKeeper, self).__init__(config, mpp) hosts: Union[str, List[str]] = config.get('hosts', []) if isinstance(hosts, list): @@ -261,7 +261,7 @@ class ZooKeeper(AbstractDCS): def _citus_cluster_loader(self, path: str) -> Dict[int, Cluster]: ret: Dict[int, Cluster] = {} for node in self.get_children(path): - if citus_group_re.match(node): + if self._mpp.group_re.match(node): ret[int(node)] = self._cluster_loader(path + node + '/') return ret diff --git a/patroni/ha.py b/patroni/ha.py index 0d3e05a4..dea78163 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -332,7 +332,7 @@ class Ha(object): if coordinator and coordinator.leader and coordinator.leader.conn_url: try: data = {'type': event, - 'group': self.state_handler.citus_handler.group(), + 'group': self.state_handler.citus_handler.group, 'leader': self.state_handler.name, 'timeout': self.dcs.ttl, 'cooldown': self.patroni.config['retry_timeout']} @@ -847,7 +847,7 @@ class Ha(object): self.state_handler.set_role('master') self.process_sync_replication() self.update_cluster_history() - self.state_handler.citus_handler.sync_pg_dist_node(self.cluster) + self.state_handler.citus_handler.sync_meta_data(self.cluster) return message elif self.state_handler.role in ('master', 'promoted', 'primary'): self.process_sync_replication() diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py index c87c29fa..f6e3ab54 100644 --- a/patroni/postgresql/__init__.py +++ b/patroni/postgresql/__init__.py @@ -19,8 +19,8 @@ from .callback_executor import CallbackAction, CallbackExecutor from .cancellable import CancellableSubprocess from .config import ConfigHandler, mtime from .connection import ConnectionPool, get_connection_cursor -from .citus import CitusHandler from .misc import parse_history, parse_lsn, postgres_major_version_to_int +from .mpp import AbstractMPP from .postmaster import PostmasterProcess from .slots import SlotsHandler from .sync import SyncHandler @@ -63,7 +63,7 @@ class Postgresql(object): "pg_catalog.pg_{0}_{1}_diff(COALESCE(pg_catalog.pg_last_{0}_receive_{1}(), '0/0'), '0/0')::bigint, " "pg_catalog.pg_is_in_recovery() AND pg_catalog.pg_is_{0}_replay_paused()") - def __init__(self, config: Dict[str, Any]) -> None: + def __init__(self, config: Dict[str, Any], mpp: AbstractMPP) -> None: self.name: str = config['name'] self.scope: str = config['scope'] self._data_dir: str = config['data_dir'] @@ -80,7 +80,7 @@ class Postgresql(object): self._pending_restart = False self.connection_pool = ConnectionPool() self._connection = self.connection_pool.get('heartbeat') - self.citus_handler = CitusHandler(self, config.get('citus')) + self.citus_handler = mpp.get_handler_impl(self) self.config = ConfigHandler(self, config) self.config.check_directories() diff --git a/patroni/postgresql/mpp/__init__.py b/patroni/postgresql/mpp/__init__.py new file mode 100644 index 00000000..3494793b --- /dev/null +++ b/patroni/postgresql/mpp/__init__.py @@ -0,0 +1,296 @@ +"""Abstract classes for MPP handler. + +MPP stands for Massively Parallel Processing, and Citus belongs to this architecture. Currently, Citus is the only +supported MPP cluster. However, we may consider adapting other databases such as TimescaleDB, GPDB, etc. into Patroni. +""" +import abc + +from typing import Any, Dict, Iterator, Optional, Union, Tuple, Type, TYPE_CHECKING + +from ...dcs import Cluster +from ...dynamic_loader import iter_classes +from ...exceptions import PatroniException + +if TYPE_CHECKING: # pragma: no cover + from .. import Postgresql + from ...config import Config + + +class AbstractMPP(abc.ABC): + """An abstract class which should be passed to :class:`AbstractDCS`. + + .. note:: + We create :class:`AbstractMPP` and :class:`AbstractMPPHandler` to solve the chicken-egg initialization problem. + When initializing DCS, we dynamically create an object implementing :class:`AbstractMPP`, later this object is + used to instantiate an object implementing :class:`AbstractMPPHandler`. + """ + + group_re: Any # re.Pattern[str] + + def __init__(self, config: Dict[str, Union[str, int]]) -> None: + """Init method for :class:`AbstractMPP`. + + :param config: configuration of MPP section. + """ + self._config = config + + def is_enabled(self) -> bool: + """Check if MPP is enabled for a given MPP. + + .. note:: + We just check that the :attr:`_config` object isn't empty and expect + it to be empty only in case of :class:`Null`. + + :returns: ``True`` if MPP is enabled, otherwise ``False``. + """ + return bool(self._config) + + @staticmethod + @abc.abstractmethod + def validate_config(config: Any) -> bool: + """Check whether provided config is good for a given MPP. + + :param config: configuration of MPP section. + + :returns: ``True`` is config passes validation, otherwise ``False``. + """ + + @property + @abc.abstractmethod + def group(self) -> Any: + """The group for a given MPP implementation.""" + + @property + @abc.abstractmethod + def coordinator_group_id(self) -> Any: + """The group id of the coordinator PostgreSQL cluster.""" + + def is_coordinator(self) -> bool: + """Check whether this node is running in the coordinator PostgreSQL cluster. + + :returns: ``True`` if MPP is enabled and the group id of this node + matches with the :attr:`coordinator_group_id`, otherwise ``False``. + """ + return self.is_enabled() and self.group == self.coordinator_group_id + + def is_worker(self) -> bool: + """Check whether this node is running as a MPP worker PostgreSQL cluster. + + :returns: ``True`` if MPP is enabled and this node is known to be not running + as the coordinator PostgreSQL cluster, otherwise ``False``. + """ + return self.is_enabled() and not self.is_coordinator() + + def _get_handler_cls(self) -> Iterator[Type['AbstractMPPHandler']]: + """Find Handler classes inherited from a class type of this object. + + :yields: handler classes for this object. + """ + for cls in self.__class__.__subclasses__(): + if issubclass(cls, AbstractMPPHandler) and cls.__name__.startswith(self.__class__.__name__): + yield cls + + def get_handler_impl(self, postgresql: 'Postgresql') -> 'AbstractMPPHandler': + """Find and instantiate Handler implementation of this object. + + :param postgresql: a reference to :class:`Postgresql` object. + + :raises: + :exc:`PatroniException`: if the Handler class haven't been found. + + :returns: an instantiated class that implements Handler for this object. + """ + for cls in self._get_handler_cls(): + return cls(postgresql, self._config) + raise PatroniException(f'Failed to initialize {self.__class__.__name__}Handler object') + + +class AbstractMPPHandler(AbstractMPP): + """An abstract class which defines interfaces that should be implemented by real handlers.""" + + def __init__(self, postgresql: 'Postgresql', config: Dict[str, Union[str, int]]) -> None: + """Init method for :class:`AbstractMPPHandler`. + + :param postgresql: a reference to :class:`Postgresql` object. + :param config: configuration of MPP section. + """ + super().__init__(config) + self._postgresql = postgresql + + @abc.abstractmethod + def handle_event(self, cluster: Cluster, event: Dict[str, Any]) -> None: + """Handle an event sent from a worker node. + + :param cluster: the currently known cluster state from DCS. + :param event: the event to be handled. + """ + + @abc.abstractmethod + def sync_meta_data(self, cluster: Cluster) -> None: + """Sync meta data on the coordinator. + + :param cluster: the currently known cluster state from DCS. + """ + + @abc.abstractmethod + def on_demote(self) -> None: + """On demote handler. + + Is called when the primary was demoted. + """ + + @abc.abstractmethod + def schedule_cache_rebuild(self) -> None: + """Cache rebuild handler. + + Is called to notify handler that it has to refresh its metadata cache from the database. + """ + + @abc.abstractmethod + def bootstrap(self) -> None: + """Bootstrap handler. + + Is called when the new cluster is initialized (through ``initdb`` or a custom bootstrap method). + """ + + @abc.abstractmethod + def adjust_postgres_gucs(self, parameters: Dict[str, Any]) -> None: + """Adjust GUCs in the current PostgreSQL configuration. + + :param parameters: dictionary of GUCs, with key as GUC name and the corresponding value as current GUC value. + """ + + @abc.abstractmethod + def ignore_replication_slot(self, slot: Dict[str, str]) -> bool: + """Check whether provided replication *slot* existing in the database should not be removed. + + .. note:: + MPP database may create replication slots for its own use, for example to migrate data between workers + using logical replication, and we don't want to suddenly drop them. + + :param slot: dictionary containing the replication slot settings, like ``name``, ``database``, ``type``, and + ``plugin``. + + :returns: ``True`` if the replication slots should not be removed, otherwise ``False``. + """ + + +class Null(AbstractMPP): + """Dummy implementation of :class:`AbstractMPP`.""" + + def __init__(self) -> None: + """Init method for :class:`Null`.""" + super().__init__({}) + + @staticmethod + def validate_config(config: Any) -> bool: + """Check whether provided config is good for :class:`Null`. + + :returns: always ``True``. + """ + return True + + @property + def group(self) -> None: + """The group for :class:`Null`. + + :returns: always ``None``. + """ + return None + + @property + def coordinator_group_id(self) -> None: + """The group id of the coordinator PostgreSQL cluster. + + :returns: always ``None``. + """ + return None + + +class NullHandler(Null, AbstractMPPHandler): + """Dummy implementation of :class:`AbstractMPPHandler`.""" + + def __init__(self, postgresql: 'Postgresql', config: Dict[str, Union[str, int]]) -> None: + """Init method for :class:`NullHandler`. + + :param postgresql: a reference to :class:`Postgresql` object. + :param config: configuration of MPP section. + """ + AbstractMPPHandler.__init__(self, postgresql, config) + + def handle_event(self, cluster: Cluster, event: Dict[str, Any]) -> None: + """Handle an event sent from a worker node. + + :param cluster: the currently known cluster state from DCS. + :param event: the event to be handled. + """ + + def sync_meta_data(self, cluster: Cluster) -> None: + """Sync meta data on the coordinator. + + :param cluster: the currently known cluster state from DCS. + """ + + def on_demote(self) -> None: + """On demote handler. + + Is called when the primary was demoted. + """ + + def schedule_cache_rebuild(self) -> None: + """Cache rebuild handler. + + Is called to notify handler that it has to refresh its metadata cache from the database. + """ + + def bootstrap(self) -> None: + """Bootstrap handler. + + Is called when the new cluster is initialized (through ``initdb`` or a custom bootstrap method). + """ + + def adjust_postgres_gucs(self, parameters: Dict[str, Any]) -> None: + """Adjust GUCs in the current PostgreSQL configuration. + + :param parameters: dictionary of GUCs, with key as GUC name and corresponding value as current GUC value. + """ + + def ignore_replication_slot(self, slot: Dict[str, str]) -> bool: + """Check whether provided replication *slot* existing in the database should not be removed. + + .. note:: + MPP database may create replication slots for its own use, for example to migrate data between workers + using logical replication, and we don't want to suddenly drop them. + + :param slot: dictionary containing the replication slot settings, like ``name``, ``database``, ``type``, and + ``plugin``. + + :returns: always ``False``. + """ + return False + + +def iter_mpp_classes( + config: Optional[Union['Config', Dict[str, Any]]] = None +) -> Iterator[Tuple[str, Type[AbstractMPP]]]: + """Attempt to import MPP modules that are present in the given configuration. + + :param config: configuration information with possible MPP names as keys. If given, only attempt to import MPP + modules defined in the configuration. Else, if ``None``, attempt to import any supported MPP module. + + :yields: tuples, each containing the module ``name`` and the imported MPP class object. + """ + yield from iter_classes(__package__, AbstractMPP, config) + + +def get_mpp(config: Union['Config', Dict[str, Any]]) -> AbstractMPP: + """Attempt to load and instantiate a MPP module from known available implementations. + + :param config: object or dictionary with Patroni configuration. + + :returns: The successfully loaded MPP or fallback to :class:`Null`. + """ + for name, mpp_class in iter_mpp_classes(config): + if mpp_class.validate_config(config[name]): + return mpp_class(config[name]) + return Null() diff --git a/patroni/postgresql/citus.py b/patroni/postgresql/mpp/citus.py similarity index 84% rename from patroni/postgresql/citus.py rename to patroni/postgresql/mpp/citus.py index b50dc1d0..a17d1435 100644 --- a/patroni/postgresql/citus.py +++ b/patroni/postgresql/mpp/citus.py @@ -6,12 +6,15 @@ from threading import Condition, Event, Thread from urllib.parse import urlparse from typing import Any, Dict, List, Optional, Union, Tuple, TYPE_CHECKING -from ..dcs import CITUS_COORDINATOR_GROUP_ID, Cluster -from ..psycopg import connect, quote_ident +from . import AbstractMPP, AbstractMPPHandler +from ...dcs import Cluster +from ...psycopg import connect, quote_ident +from ...utils import parse_int if TYPE_CHECKING: # pragma: no cover - from . import Postgresql + from .. import Postgresql +CITUS_COORDINATOR_GROUP_ID = 0 CITUS_SLOT_NAME_RE = re.compile(r'^citus_shard_(move|split)_slot(_[1-9][0-9]*){2,3}$') logger = logging.getLogger(__name__) @@ -63,13 +66,45 @@ class PgDistNode(object): return str(self) -class CitusHandler(Thread): +class Citus(AbstractMPP): - def __init__(self, postgresql: 'Postgresql', config: Optional[Dict[str, Union[str, int]]]) -> None: - super(CitusHandler, self).__init__() + group_re = re.compile('^(0|[1-9][0-9]*)$') + + @staticmethod + def validate_config(config: Union[Any, Dict[str, Union[str, int]]]) -> bool: + """Check whether provided config is good for a given MPP. + + :param config: configuration of ``citus`` MPP section. + + :returns: ``True`` is config passes validation, otherwise ``False``. + """ + return isinstance(config, dict) \ + and isinstance(config.get('database'), str) \ + and parse_int(config.get('group')) is not None + + @property + def group(self) -> int: + """The group of this Citus node.""" + return int(self._config['group']) + + @property + def coordinator_group_id(self) -> int: + """The group id of the Citus coordinator PostgreSQL cluster.""" + return CITUS_COORDINATOR_GROUP_ID + + +class CitusHandler(Citus, AbstractMPPHandler, Thread): + """Define the interfaces for handling an underlying Citus cluster.""" + + def __init__(self, postgresql: 'Postgresql', config: Dict[str, Union[str, int]]) -> None: + """"Initialize a new instance of :class:`CitusHandler`. + + :param postgresql: the Postgres node. + :param config: the ``citus`` MPP config section. + """ + Thread.__init__(self) + AbstractMPPHandler.__init__(self, postgresql, config) self.daemon = True - self._postgresql = postgresql - self._config = config if config: self._connection = postgresql.connection_pool.get( 'citus', {'dbname': config['database'], @@ -81,19 +116,11 @@ class CitusHandler(Thread): self._condition = Condition() # protects _pg_dist_node, _tasks, _in_flight, and _schedule_load_pg_dist_node self.schedule_cache_rebuild() - def is_enabled(self) -> bool: - return isinstance(self._config, dict) - - def group(self) -> Optional[int]: - return int(self._config['group']) if isinstance(self._config, dict) else None - - def is_coordinator(self) -> bool: - return self.is_enabled() and self.group() == CITUS_COORDINATOR_GROUP_ID - - def is_worker(self) -> bool: - return self.is_enabled() and not self.is_coordinator() - def schedule_cache_rebuild(self) -> None: + """Cache rebuild handler. + + Is called to notify handler that it has to refresh its metadata cache from the database. + """ with self._condition: self._schedule_load_pg_dist_node = True @@ -134,8 +161,8 @@ class CitusHandler(Thread): self._pg_dist_node = {r[1]: PgDistNode(r[1], r[2], r[3], 'after_promote', r[0]) for r in rows} return True - def sync_pg_dist_node(self, cluster: Cluster) -> None: - """Maintain the `pg_dist_node` from the coordinator leader every heartbeat loop. + def sync_meta_data(self, cluster: Cluster) -> None: + """Maintain the ``pg_dist_node`` from the coordinator leader every heartbeat loop. We can't always rely on REST API calls from worker nodes in order to maintain `pg_dist_node`, therefore at least once per heartbeat @@ -296,16 +323,16 @@ class CitusHandler(Thread): with self._condition: i = self.find_task_by_group(task.group) - # The `PgDistNode.timeout` == None is an indicator that it was scheduled from the sync_pg_dist_node(). + # The `PgDistNode.timeout` == None is an indicator that it was scheduled from the sync_meta_data(). if task.timeout is None: # We don't want to override the already existing task created from REST API. if i is not None and self._tasks[i].timeout is not None: return False # There is a little race condition with tasks created from REST API - the call made "before" the member - # key is updated in DCS. Therefore it is possible that :func:`sync_pg_dist_node` will try to create a - # task based on the outdated values of "state"/"role". To solve it we introduce an artificial timeout. - # Only when the timeout is reached new tasks could be scheduled from sync_pg_dist_node() + # key is updated in DCS. Therefore it is possible that :func:`sync_meta_data` will try to create a task + # based on the outdated values of "state"/"role". To solve it we introduce an artificial timeout. + # Only when the timeout is reached new tasks could be scheduled from sync_meta_data() if self._in_flight and self._in_flight.group == task.group and self._in_flight.timeout is not None\ and self._in_flight.deadline > time.time(): return False @@ -353,9 +380,10 @@ class CitusHandler(Thread): task.wait() def bootstrap(self) -> None: - if not isinstance(self._config, dict): # self.is_enabled() - return + """Bootstrap handler. + Is called when the new cluster is initialized (through ``initdb`` or a custom bootstrap method). + """ conn_kwargs = {**self._postgresql.connection_pool.conn_kwargs, 'options': '-c synchronous_commit=local -c statement_timeout=0'} if self._config['database'] != self._postgresql.database: @@ -388,9 +416,10 @@ class CitusHandler(Thread): conn.close() def adjust_postgres_gucs(self, parameters: Dict[str, Any]) -> None: - if not self.is_enabled(): - return + """Adjust GUCs in the current PostgreSQL configuration. + :param parameters: dictionary of GUCs, with key as GUC name and the corresponding value as current GUC value. + """ # citus extension must be on the first place in shared_preload_libraries shared_preload_libraries = list(filter( lambda el: el and el != 'citus', @@ -408,8 +437,18 @@ class CitusHandler(Thread): 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']: + """Check whether provided replication *slot* existing in the database should not be removed. + + .. note:: + MPP database may create replication slots for its own use, for example to migrate data between workers + using logical replication, and we don't want to suddenly drop them. + + :param slot: dictionary containing the replication slot settings, like ``name``, ``database``, ``type``, and + ``plugin``. + + :returns: ``True`` if the replication slots should not be removed, otherwise ``False``. + """ + if self._postgresql.is_primary() and slot['type'] == 'logical' and slot['database'] == self._config['database']: m = CITUS_SLOT_NAME_RE.match(slot['name']) return bool(m and {'move': 'pgoutput', 'split': 'citus'}.get(m.group(1)) == slot['plugin']) return False diff --git a/patroni/postgresql/slots.py b/patroni/postgresql/slots.py index fb9448cd..7f7fd294 100644 --- a/patroni/postgresql/slots.py +++ b/patroni/postgresql/slots.py @@ -291,7 +291,7 @@ class SlotsHandler: :param name: name of the slot to ignore :returns: ``True`` if slot *name* matches any slot specified in ``ignore_slots`` configuration, - otherwise will pass through and return result of :meth:`CitusHandler.ignore_replication_slot`. + otherwise will pass through and return result of :meth:`AbstractMPPHandler.ignore_replication_slot`. """ slot = self._replication_slots[name] if cluster.config: diff --git a/tests/__init__.py b/tests/__init__.py index b013e4e1..986bd88b 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -12,6 +12,7 @@ import patroni.psycopg as psycopg from patroni.dcs import Leader, Member from patroni.postgresql import Postgresql from patroni.postgresql.config import ConfigHandler +from patroni.postgresql.mpp import get_mpp from patroni.utils import RetryFailedError, tzutc @@ -252,23 +253,24 @@ class PostgresInit(unittest.TestCase): @patch.object(Postgresql, 'get_postgres_role_from_data_directory', Mock(return_value='primary')) def setUp(self): data_dir = os.path.join('data', 'test0') - self.p = Postgresql({'name': 'postgresql0', 'scope': 'batman', 'data_dir': data_dir, - 'config_dir': data_dir, 'retry_timeout': 10, - 'krbsrvname': 'postgres', 'pgpass': os.path.join(data_dir, 'pgpass0'), - 'listen': '127.0.0.2, 127.0.0.3:5432', - 'connect_address': '127.0.0.2:5432', 'proxy_address': '127.0.0.2:5433', - 'authentication': {'superuser': {'username': 'foo', 'password': 'test'}, - '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_unix_socket': True, - 'parameters': self._PARAMETERS, - 'recovery_conf': {'foo': 'bar'}, - 'pg_hba': ['host all all 0.0.0.0/0 md5'], - 'pg_ident': ['krb realm postgres'], - 'callbacks': {'on_start': 'true', 'on_stop': 'true', 'on_reload': 'true', - 'on_restart': 'true', 'on_role_change': 'true'}, - 'citus': {'group': 0, 'database': 'citus'}}) + config = {'name': 'postgresql0', 'scope': 'batman', 'data_dir': data_dir, + 'config_dir': data_dir, 'retry_timeout': 10, + 'krbsrvname': 'postgres', 'pgpass': os.path.join(data_dir, 'pgpass0'), + 'listen': '127.0.0.2, 127.0.0.3:5432', + 'connect_address': '127.0.0.2:5432', 'proxy_address': '127.0.0.2:5433', + 'authentication': {'superuser': {'username': 'foo', 'password': 'test'}, + '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_unix_socket': True, + 'parameters': self._PARAMETERS, + 'recovery_conf': {'foo': 'bar'}, + 'pg_hba': ['host all all 0.0.0.0/0 md5'], + 'pg_ident': ['krb realm postgres'], + 'callbacks': {'on_start': 'true', 'on_stop': 'true', 'on_reload': 'true', + 'on_restart': 'true', 'on_role_change': 'true'}, + 'citus': {'group': 0, 'database': 'citus'}} + self.p = Postgresql(config, get_mpp(config)) class BaseTestPostgresql(PostgresInit): diff --git a/tests/test_citus.py b/tests/test_citus.py index 7279893e..dbf6d9cf 100644 --- a/tests/test_citus.py +++ b/tests/test_citus.py @@ -1,12 +1,12 @@ import time from mock import Mock, patch -from patroni.postgresql.citus import CitusHandler +from patroni.postgresql.mpp.citus import CitusHandler from . import BaseTestPostgresql, MockCursor, psycopg_connect, SleepException from .test_ha import get_cluster_initialized_with_leader -@patch('patroni.postgresql.citus.Thread', Mock()) +@patch('patroni.postgresql.mpp.citus.Thread', Mock()) @patch('patroni.psycopg.connect', psycopg_connect) class TestCitus(BaseTestPostgresql): @@ -17,9 +17,9 @@ class TestCitus(BaseTestPostgresql): self.cluster.workers[1] = self.cluster @patch('time.time', Mock(side_effect=[100, 130, 160, 190, 220, 250, 280, 310, 340, 370])) - @patch('patroni.postgresql.citus.logger.exception', Mock(side_effect=SleepException)) - @patch('patroni.postgresql.citus.logger.warning') - @patch('patroni.postgresql.citus.PgDistNode.wait', Mock()) + @patch('patroni.postgresql.mpp.citus.logger.exception', Mock(side_effect=SleepException)) + @patch('patroni.postgresql.mpp.citus.logger.warning') + @patch('patroni.postgresql.mpp.citus.PgDistNode.wait', Mock()) @patch.object(CitusHandler, 'is_alive', Mock(return_value=True)) def test_run(self, mock_logger_warning): # `before_demote` or `before_promote` REST API calls starting a @@ -39,10 +39,10 @@ class TestCitus(BaseTestPostgresql): @patch.object(CitusHandler, 'is_alive', Mock(return_value=False)) @patch.object(CitusHandler, 'start', Mock()) - def test_sync_pg_dist_node(self): + def test_sync_meta_data(self): with patch.object(CitusHandler, 'is_enabled', Mock(return_value=False)): - self.c.sync_pg_dist_node(self.cluster) - self.c.sync_pg_dist_node(self.cluster) + self.c.sync_meta_data(self.cluster) + self.c.sync_meta_data(self.cluster) def test_handle_event(self): self.c.handle_event(self.cluster, {}) @@ -51,22 +51,22 @@ class TestCitus(BaseTestPostgresql): 'leader': 'leader', 'timeout': 30, 'cooldown': 10}) def test_add_task(self): - with patch('patroni.postgresql.citus.logger.error') as mock_logger, \ - patch('patroni.postgresql.citus.urlparse', Mock(side_effect=Exception)): + with patch('patroni.postgresql.mpp.citus.logger.error') as mock_logger, \ + patch('patroni.postgresql.mpp.citus.urlparse', Mock(side_effect=Exception)): self.c.add_task('', 1, None) mock_logger.assert_called_once() - with patch('patroni.postgresql.citus.logger.debug') as mock_logger: + with patch('patroni.postgresql.mpp.citus.logger.debug') as mock_logger: self.c.add_task('before_demote', 1, 'postgres://host:5432/postgres', 30) mock_logger.assert_called_once() self.assertTrue(mock_logger.call_args[0][0].startswith('Adding the new task:')) - with patch('patroni.postgresql.citus.logger.debug') as mock_logger: + with patch('patroni.postgresql.mpp.citus.logger.debug') as mock_logger: self.c.add_task('before_promote', 1, 'postgres://host:5432/postgres', 30) mock_logger.assert_called_once() self.assertTrue(mock_logger.call_args[0][0].startswith('Overriding existing task:')) - # add_task called from sync_pg_dist_node should not override already scheduled or in flight task until deadline + # add_task called from sync_meta_data should not override already scheduled or in flight task until deadline self.assertIsNotNone(self.c.add_task('after_promote', 1, 'postgres://host:5432/postgres', 30)) self.assertIsNone(self.c.add_task('after_promote', 1, 'postgres://host:5432/postgres')) self.c._in_flight = self.c._tasks.pop() @@ -106,7 +106,7 @@ class TestCitus(BaseTestPostgresql): self.c.process_tasks() self.c.add_task('after_promote', 0, 'postgres://host3:5432/postgres') - with patch('patroni.postgresql.citus.logger.error') as mock_logger, \ + with patch('patroni.postgresql.mpp.citus.logger.error') as mock_logger, \ patch.object(CitusHandler, 'query', Mock(side_effect=Exception)): self.c.process_tasks() mock_logger.assert_called_once() @@ -115,7 +115,7 @@ class TestCitus(BaseTestPostgresql): def test_on_demote(self): self.c.on_demote() - @patch('patroni.postgresql.citus.logger.error') + @patch('patroni.postgresql.mpp.citus.logger.error') @patch.object(MockCursor, 'execute', Mock(side_effect=Exception)) def test_load_pg_dist_node(self, mock_logger): # load_pg_dist_node() triggers, query fails and exception is property handled @@ -140,10 +140,6 @@ class TestCitus(BaseTestPostgresql): self.assertEqual(parameters['wal_level'], 'logical') self.assertEqual(parameters['citus.local_hostname'], '/tmp') - def test_bootstrap(self): - self.c._config = None - self.c.bootstrap() - def test_ignore_replication_slot(self): self.assertFalse(self.c.ignore_replication_slot({'name': 'foo', 'type': 'physical', 'database': 'bar', 'plugin': 'wal2json'})) diff --git a/tests/test_consul.py b/tests/test_consul.py index 83ee67d8..494d1126 100644 --- a/tests/test_consul.py +++ b/tests/test_consul.py @@ -3,8 +3,10 @@ import unittest from consul import ConsulException, NotFound from mock import Mock, PropertyMock, patch +from patroni.dcs import get_dcs from patroni.dcs.consul import AbstractDCS, Cluster, Consul, ConsulInternalError, \ ConsulError, ConsulClient, HTTPClient, InvalidSessionTTL, InvalidSession, RetryFailedError +from patroni.postgresql.mpp import get_mpp from . import SleepException @@ -91,13 +93,17 @@ class TestConsul(unittest.TestCase): @patch.object(consul.Consul.KV, 'get', kv_get) @patch.object(consul.Consul.KV, 'delete', Mock()) def setUp(self): - Consul({'ttl': 30, 'scope': 't', 'name': 'p', 'url': 'https://l:1', 'retry_timeout': 10, - 'verify': 'on', 'key': 'foo', 'cert': 'bar', 'cacert': 'buz', 'token': 'asd', 'dc': 'dc1', - 'register_service': True}) - Consul({'ttl': 30, 'scope': 't_', 'name': 'p', 'url': 'https://l:1', 'retry_timeout': 10, - 'verify': 'on', 'cert': 'bar', 'cacert': 'buz', 'register_service': True}) - self.c = Consul({'ttl': 30, 'scope': 'test', 'name': 'postgresql1', 'host': 'localhost:1', 'retry_timeout': 10, - 'register_service': True, 'service_check_tls_server_name': True}) + self.assertIsInstance(get_dcs({'ttl': 30, 'scope': 't', 'name': 'p', 'retry_timeout': 10, + 'consul': {'url': 'https://l:1', 'verify': 'on', + 'key': 'foo', 'cert': 'bar', 'cacert': 'buz', + 'token': 'asd', 'dc': 'dc1', 'register_service': True}}), Consul) + self.assertIsInstance(get_dcs({'ttl': 30, 'scope': 't_', 'name': 'p', 'retry_timeout': 10, + 'consul': {'url': 'https://l:1', 'verify': 'on', + 'cert': 'bar', 'cacert': 'buz', 'register_service': True}}), Consul) + self.c = get_dcs({'ttl': 30, 'scope': 'test', 'name': 'postgresql1', 'retry_timeout': 10, + 'consul': {'host': 'localhost:1', 'register_service': True, + 'service_check_tls_server_name': True}}) + self.assertIsInstance(self.c, Consul) self.c._base_path = 'service/good' self.c.get_cluster() @@ -130,7 +136,7 @@ class TestConsul(unittest.TestCase): self.assertIsInstance(self.c.get_cluster(), Cluster) def test__get_citus_cluster(self): - self.c._citus_group = '0' + self.c._mpp = get_mpp({'citus': {'group': 0, 'database': 'postgres'}}) cluster = self.c.get_cluster() self.assertIsInstance(cluster, Cluster) self.assertIsInstance(cluster.workers[1], Cluster) diff --git a/tests/test_ctl.py b/tests/test_ctl.py index f9ee62ce..f5341f88 100644 --- a/tests/test_ctl.py +++ b/tests/test_ctl.py @@ -12,6 +12,7 @@ from patroni.ctl import ctl, load_config, output_members, get_dcs, parse_dcs, \ get_all_members, get_any_member, get_cursor, query_member, PatroniCtlException, apply_config_changes, \ format_config_for_editing, show_diff, invoke_editor, format_pg_version, CONFIG_FILE_PATH, PatronictlPrettyTable from patroni.dcs import Cluster, Failover +from patroni.postgresql.mpp import get_mpp from patroni.psycopg import OperationalError from patroni.utils import tzutc from prettytable import PrettyTable, ALL @@ -69,7 +70,7 @@ class TestCtl(unittest.TestCase): @patch('patroni.psycopg.connect', psycopg_connect) def test_get_cursor(self): with click.Context(click.Command('query')) as ctx: - ctx.obj = {'__config': {}} + ctx.obj = {'__config': {}, '__mpp': get_mpp({})} for role in self.TEST_ROLES: self.assertIsNone(get_cursor(get_cluster_initialized_without_leader(), None, {}, role=role)) self.assertIsNotNone(get_cursor(get_cluster_initialized_with_leader(), None, {}, role=role)) @@ -107,7 +108,7 @@ class TestCtl(unittest.TestCase): def test_output_members(self): with click.Context(click.Command('list')) as ctx: - ctx.obj = {'__config': {}} + ctx.obj = {'__config': {}, '__mpp': get_mpp({})} scheduled_at = datetime.now(tzutc) + timedelta(seconds=600) cluster = get_cluster_initialized_with_leader(Failover(1, 'foo', 'bar', scheduled_at)) del cluster.members[1].data['conn_url'] @@ -250,7 +251,7 @@ class TestCtl(unittest.TestCase): @patch('patroni.dynamic_loader.iter_modules', Mock(return_value=['patroni.dcs.dummy', 'patroni.dcs.etcd'])) def test_get_dcs(self): with click.Context(click.Command('list')) as ctx: - ctx.obj = {'__config': {'dummy': {}}} + ctx.obj = {'__config': {'dummy': {}}, '__mpp': get_mpp({})} self.assertRaises(PatroniCtlException, get_dcs, 'dummy', 0) @patch('patroni.psycopg.connect', psycopg_connect) @@ -439,7 +440,7 @@ class TestCtl(unittest.TestCase): def test_get_any_member(self): with click.Context(click.Command('list')) as ctx: - ctx.obj = {'__config': {}} + ctx.obj = {'__config': {}, '__mpp': get_mpp({})} for role in self.TEST_ROLES: self.assertIsNone(get_any_member(get_cluster_initialized_without_leader(), None, role=role)) @@ -448,7 +449,7 @@ class TestCtl(unittest.TestCase): def test_get_all_members(self): with click.Context(click.Command('list')) as ctx: - ctx.obj = {'__config': {}} + ctx.obj = {'__config': {}, '__mpp': get_mpp({})} for role in self.TEST_ROLES: self.assertEqual(list(get_all_members(get_cluster_initialized_without_leader(), None, role=role)), []) diff --git a/tests/test_etcd.py b/tests/test_etcd.py index 874aac5c..d7a423be 100644 --- a/tests/test_etcd.py +++ b/tests/test_etcd.py @@ -5,8 +5,10 @@ import unittest from dns.exception import DNSException from mock import Mock, PropertyMock, patch +from patroni.dcs import get_dcs from patroni.dcs.etcd import AbstractDCS, EtcdClient, Cluster, Etcd, EtcdError, DnsCachingResolver from patroni.exceptions import DCSError +from patroni.postgresql.mpp import get_mpp from patroni.utils import Retry from urllib3.exceptions import ReadTimeoutError @@ -138,8 +140,9 @@ class TestClient(unittest.TestCase): @patch.object(EtcdClient, '_get_machines_list', Mock(return_value=['http://localhost:2379', 'http://localhost:4001'])) def setUp(self): - self.etcd = Etcd({'namespace': '/patroni/', 'ttl': 30, 'retry_timeout': 3, - 'srv': 'test', 'scope': 'test', 'name': 'foo'}) + self.etcd = get_dcs({'namespace': '/patroni/', 'ttl': 30, 'retry_timeout': 3, + 'etcd': {'srv': 'test'}, 'scope': 'test', 'name': 'foo'}) + self.assertIsInstance(self.etcd, Etcd) self.client = self.etcd._client self.client.http.request = http_request self.client.http.request_encode_body = http_request @@ -235,7 +238,7 @@ class TestEtcd(unittest.TestCase): Mock(return_value=['http://localhost:2379', 'http://localhost:4001'])) def setUp(self): self.etcd = Etcd({'namespace': '/patroni/', 'ttl': 30, 'retry_timeout': 10, - 'host': 'localhost:2379', 'scope': 'test', 'name': 'foo'}) + 'host': 'localhost:2379', 'scope': 'test', 'name': 'foo'}, get_mpp({})) def test_base_path(self): self.assertEqual(self.etcd._base_path, '/patroni/test') @@ -270,7 +273,7 @@ class TestEtcd(unittest.TestCase): self.assertRaises(EtcdError, self.etcd.get_cluster) def test__get_citus_cluster(self): - self.etcd._citus_group = '0' + self.etcd._mpp = get_mpp({'citus': {'group': 0, 'database': 'postgres'}}) cluster = self.etcd.get_cluster() self.assertIsInstance(cluster, Cluster) self.assertIsInstance(cluster.workers[1], Cluster) diff --git a/tests/test_etcd3.py b/tests/test_etcd3.py index 10ab1ea5..fcfd4e4b 100644 --- a/tests/test_etcd3.py +++ b/tests/test_etcd3.py @@ -4,10 +4,12 @@ import unittest import urllib3 from mock import Mock, PropertyMock, patch +from patroni.dcs import get_dcs from patroni.dcs.etcd import DnsCachingResolver from patroni.dcs.etcd3 import PatroniEtcd3Client, Cluster, Etcd3, Etcd3Client, \ Etcd3Error, Etcd3ClientError, ReAuthenticateMode, RetryFailedError, InvalidAuthToken, Unavailable, \ Unknown, UnsupportedEtcdVersion, UserEmpty, AuthFailed, AuthOldRevision, base64_encode +from patroni.postgresql.mpp import get_mpp from threading import Thread from . import SleepException, MockResponse @@ -80,9 +82,9 @@ class BaseTestEtcd3(unittest.TestCase): @patch.object(Thread, 'start', Mock()) @patch.object(urllib3.PoolManager, 'urlopen', mock_urlopen) def setUp(self): - self.etcd3 = Etcd3({'namespace': '/patroni/', 'ttl': 30, 'retry_timeout': 10, - 'host': 'localhost:2378', 'scope': 'test', 'name': 'foo', - 'username': 'etcduser', 'password': 'etcdpassword'}) + self.etcd3 = get_dcs({'namespace': '/patroni/', 'ttl': 30, 'retry_timeout': 10, 'name': 'foo', 'scope': 'test', + 'etcd3': {'host': 'localhost:2378', 'username': 'etcduser', 'password': 'etcdpassword'}}) + self.assertIsInstance(self.etcd3, Etcd3) self.client = self.etcd3._client self.kv_cache = self.client._kv_cache @@ -236,7 +238,7 @@ class TestEtcd3(BaseTestEtcd3): self.assertRaises(Etcd3Error, self.etcd3.get_cluster) def test__get_citus_cluster(self): - self.etcd3._citus_group = '0' + self.etcd3._mpp = get_mpp({'citus': {'group': 0, 'database': 'postgres'}}) cluster = self.etcd3.get_cluster() self.assertIsInstance(cluster, Cluster) self.assertIsInstance(cluster.workers[1], Cluster) diff --git a/tests/test_exhibitor.py b/tests/test_exhibitor.py index a908e1fc..5a72eb21 100644 --- a/tests/test_exhibitor.py +++ b/tests/test_exhibitor.py @@ -2,6 +2,7 @@ import unittest import urllib3 from mock import Mock, patch +from patroni.dcs import get_dcs from patroni.dcs.exhibitor import ExhibitorEnsembleProvider, Exhibitor from patroni.dcs.zookeeper import ZooKeeperError @@ -26,8 +27,9 @@ class TestExhibitor(unittest.TestCase): status=200, body=b'{"servers":["127.0.0.1","127.0.0.2","127.0.0.3"],"port":2181}'))) @patch('patroni.dcs.zookeeper.PatroniKazooClient', MockKazooClient) def setUp(self): - self.e = Exhibitor({'hosts': ['localhost', 'exhibitor'], 'port': 8181, 'scope': 'test', - 'name': 'foo', 'ttl': 30, 'retry_timeout': 10}) + self.e = get_dcs({'exhibitor': {'hosts': ['localhost', 'exhibitor'], 'port': 8181}, + 'scope': 'test', 'name': 'foo', 'ttl': 30, 'retry_timeout': 10}) + self.assertIsInstance(self.e, Exhibitor) @patch.object(ExhibitorEnsembleProvider, 'poll', Mock(return_value=True)) @patch.object(MockKazooClient, 'get_children', Mock(side_effect=Exception)) diff --git a/tests/test_ha.py b/tests/test_ha.py index 5b1d4562..45f64164 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -197,7 +197,7 @@ def run_async(self, func, args=()): @patch('patroni.async_executor.AsyncExecutor.busy', PropertyMock(return_value=False)) @patch('patroni.async_executor.AsyncExecutor.run_async', run_async) @patch('patroni.postgresql.rewind.Thread', Mock()) -@patch('patroni.postgresql.citus.CitusHandler.start', Mock()) +@patch('patroni.postgresql.mpp.citus.CitusHandler.start', Mock()) @patch('subprocess.call', Mock(return_value=0)) @patch('time.sleep', Mock()) class TestHa(PostgresInit): @@ -593,8 +593,8 @@ class TestHa(PostgresInit): self.assertEqual(self.ha.bootstrap(), 'failed to acquire initialize lock') @patch('patroni.psycopg.connect', psycopg_connect) - @patch('patroni.postgresql.citus.connect', psycopg_connect) - @patch('patroni.postgresql.citus.quote_ident', Mock()) + @patch('patroni.postgresql.mpp.citus.connect', psycopg_connect) + @patch('patroni.postgresql.mpp.citus.quote_ident', Mock()) @patch.object(Postgresql, 'connection', Mock(return_value=None)) def test_bootstrap_initialized_new_cluster(self): self.ha.cluster = get_cluster_not_initialized_without_leader() @@ -615,8 +615,8 @@ class TestHa(PostgresInit): self.assertRaises(PatroniFatalException, self.ha.post_bootstrap) @patch('patroni.psycopg.connect', psycopg_connect) - @patch('patroni.postgresql.citus.connect', psycopg_connect) - @patch('patroni.postgresql.citus.quote_ident', Mock()) + @patch('patroni.postgresql.mpp.citus.connect', psycopg_connect) + @patch('patroni.postgresql.mpp.citus.quote_ident', Mock()) @patch.object(Postgresql, 'connection', Mock(return_value=None)) def test_bootstrap_release_initialize_key_on_watchdog_failure(self): self.ha.cluster = get_cluster_not_initialized_without_leader() @@ -659,7 +659,7 @@ class TestHa(PostgresInit): @patch.object(ConfigHandler, 'replace_pg_hba', Mock()) @patch.object(ConfigHandler, 'replace_pg_ident', Mock()) @patch.object(PostmasterProcess, 'start', Mock(return_value=MockPostmaster())) - @patch('patroni.postgresql.citus.CitusHandler.is_coordinator', Mock(return_value=False)) + @patch('patroni.postgresql.mpp.AbstractMPPHandler.is_coordinator', Mock(return_value=False)) def test_worker_restart(self): self.ha.has_lock = true self.ha.patroni.request = Mock() @@ -694,7 +694,7 @@ class TestHa(PostgresInit): self.ha.is_paused = true self.assertEqual(self.ha.run_cycle(), 'PAUSE: restart in progress') - @patch('patroni.postgresql.citus.CitusHandler.is_coordinator', Mock(return_value=False)) + @patch('patroni.postgresql.mpp.AbstractMPPHandler.is_coordinator', Mock(return_value=False)) def test_manual_failover_from_leader(self): self.ha.has_lock = true # I am the leader @@ -733,7 +733,7 @@ class TestHa(PostgresInit): ('Member %s exceeds maximum replication lag', 'b')) self.ha.cluster.members.pop() - @patch('patroni.postgresql.citus.CitusHandler.is_coordinator', Mock(return_value=False)) + @patch('patroni.postgresql.mpp.AbstractMPPHandler.is_coordinator', Mock(return_value=False)) def test_manual_switchover_from_leader(self): self.ha.has_lock = true # I am the leader @@ -774,7 +774,7 @@ class TestHa(PostgresInit): self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock') self.assertEqual(mock_info.call_args_list[0][0], ('Member %s exceeds maximum replication lag', 'leader')) - @patch('patroni.postgresql.citus.CitusHandler.is_coordinator', Mock(return_value=False)) + @patch('patroni.postgresql.mpp.AbstractMPPHandler.is_coordinator', Mock(return_value=False)) def test_scheduled_switchover_from_leader(self): self.ha.has_lock = true # I am the leader @@ -1544,7 +1544,7 @@ class TestHa(PostgresInit): self.ha.is_failover_possible = true self.ha.shutdown() - @patch('patroni.postgresql.citus.CitusHandler.is_coordinator', Mock(return_value=False)) + @patch('patroni.postgresql.mpp.AbstractMPPHandler.is_coordinator', Mock(return_value=False)) def test_shutdown_citus_worker(self): self.ha.is_leader = true self.p.is_running = Mock(side_effect=[Mock(), False]) @@ -1656,7 +1656,7 @@ class TestHa(PostgresInit): self.assertRaises(DCSError, self.ha.acquire_lock) self.assertFalse(self.ha.acquire_lock()) - @patch('patroni.postgresql.citus.CitusHandler.is_coordinator', Mock(return_value=False)) + @patch('patroni.postgresql.mpp.AbstractMPPHandler.is_coordinator', Mock(return_value=False)) def test_notify_citus_coordinator(self): self.ha.patroni.request = Mock() self.ha.notify_citus_coordinator('before_demote') diff --git a/tests/test_kubernetes.py b/tests/test_kubernetes.py index b6db7fb3..c493d799 100644 --- a/tests/test_kubernetes.py +++ b/tests/test_kubernetes.py @@ -8,9 +8,11 @@ import unittest import urllib3 from mock import Mock, PropertyMock, mock_open, patch +from patroni.dcs import get_dcs from patroni.dcs.kubernetes import Cluster, k8s_client, k8s_config, K8sConfig, K8sConnectionFailed, \ K8sException, K8sObject, Kubernetes, KubernetesError, KubernetesRetriableException, \ Retry, RetryFailedError, SERVICE_HOST_ENV_NAME, SERVICE_PORT_ENV_NAME +from patroni.postgresql.mpp import get_mpp from threading import Thread from . import MockResponse, SleepException @@ -225,11 +227,12 @@ class BaseTestKubernetes(unittest.TestCase): @patch.object(k8s_client.CoreV1Api, 'list_namespaced_pod', mock_list_namespaced_pod, create=True) @patch.object(k8s_client.CoreV1Api, 'list_namespaced_config_map', mock_list_namespaced_config_map, create=True) def setUp(self, config=None): - config = config or {} - config.update(ttl=30, scope='test', name='p-0', loop_wait=10, group=0, - retry_timeout=10, labels={'f': 'b'}, bypass_api_service=True) - self.k = Kubernetes(config) - self.k._citus_group = None + config = {'ttl': 30, 'scope': 'test', 'name': 'p-0', 'loop_wait': 10, 'retry_timeout': 10, + 'kubernetes': {'labels': {'f': 'b'}, 'bypass_api_service': True, **(config or {})}, + 'citus': {'group': 0, 'database': 'postgres'}} + self.k = get_dcs(config) + self.assertIsInstance(self.k, Kubernetes) + self.k._mpp = get_mpp({}) self.assertRaises(AttributeError, self.k._pods._build_cache) self.k._pods._is_ready = True self.assertRaises(TypeError, self.k._kinds._build_cache) @@ -254,7 +257,7 @@ class TestKubernetesConfigMaps(BaseTestKubernetes): self.assertRaises(KubernetesError, self.k.get_cluster) def test__get_citus_cluster(self): - self.k._citus_group = '0' + self.k._mpp = get_mpp({'citus': {'group': 0, 'database': 'postgres'}}) cluster = self.k.get_cluster() self.assertIsInstance(cluster, Cluster) self.assertIsInstance(cluster.workers[1], Cluster) @@ -466,7 +469,7 @@ class TestCacheBuilder(BaseTestKubernetes): @patch('patroni.dcs.kubernetes.ObjectCache._watch', mock_watch) @patch.object(urllib3.HTTPResponse, 'read_chunked') def test__build_cache(self, mock_read_chunked): - self.k._citus_group = '0' + self.k._mpp = get_mpp({'citus': {'group': 0, 'database': 'postgres'}}) mock_read_chunked.return_value = [json.dumps( {'type': 'MODIFIED', 'object': {'metadata': { 'name': self.k.config_path, 'resourceVersion': '2', 'annotations': {self.k._CONFIG: 'foo'}}}} diff --git a/tests/test_mpp.py b/tests/test_mpp.py new file mode 100644 index 00000000..9eb87633 --- /dev/null +++ b/tests/test_mpp.py @@ -0,0 +1,52 @@ +from typing import Any +from patroni.exceptions import PatroniException +from patroni.postgresql.mpp import AbstractMPP, get_mpp, Null + +from . import BaseTestPostgresql +from .test_ha import get_cluster_initialized_with_leader + + +class TestMPP(BaseTestPostgresql): + + def setUp(self): + super(TestMPP, self).setUp() + self.cluster = get_cluster_initialized_with_leader() + + def test_get_handler_impl_exception(self): + class DummyMPP(AbstractMPP): + def __init__(self) -> None: + super().__init__({}) + + @staticmethod + def validate_config(config: Any) -> bool: + return True + + @property + def group(self) -> None: + return None + + @property + def coordinator_group_id(self) -> None: + return None + + @property + def type(self) -> str: + return "dummy" + + mpp = DummyMPP() + self.assertRaises(PatroniException, mpp.get_handler_impl, self.p) + + def test_null_handler(self): + config = {} + mpp = get_mpp(config) + self.assertIsInstance(mpp, Null) + self.assertIsNone(mpp.group) + self.assertTrue(mpp.validate_config(config)) + nullHandler = mpp.get_handler_impl(self.p) + self.assertIsNone(nullHandler.handle_event(self.cluster, {})) + self.assertIsNone(nullHandler.sync_meta_data(self.cluster)) + self.assertIsNone(nullHandler.on_demote()) + self.assertIsNone(nullHandler.schedule_cache_rebuild()) + self.assertIsNone(nullHandler.bootstrap()) + self.assertIsNone(nullHandler.adjust_postgres_gucs({})) + self.assertFalse(nullHandler.ignore_replication_slot({})) diff --git a/tests/test_raft.py b/tests/test_raft.py index 9bb109e9..387a5a56 100644 --- a/tests/test_raft.py +++ b/tests/test_raft.py @@ -4,8 +4,10 @@ import tempfile import time from mock import Mock, PropertyMock, patch +from patroni.dcs import get_dcs from patroni.dcs.raft import Cluster, DynMemberSyncObj, KVStoreTTL, \ Raft, RaftError, SyncObjUtility, TCPTransport, _TCPTransport +from patroni.postgresql.mpp import get_mpp from pysyncobj import SyncObjConf, FAIL_REASON @@ -128,9 +130,10 @@ class TestRaft(unittest.TestCase): _TMP = tempfile.gettempdir() def test_raft(self): - raft = Raft({'ttl': 30, 'scope': 'test', 'name': 'pg', 'self_addr': '127.0.0.1:1234', - 'retry_timeout': 10, 'data_dir': self._TMP, - 'database': 'citus', 'group': 0}) + raft = get_dcs({'ttl': 30, 'scope': 'test', 'name': 'pg', 'retry_timeout': 10, + 'raft': {'self_addr': '127.0.0.1:1234', 'data_dir': self._TMP}, + 'citus': {'group': 0, 'database': 'postgres'}}) + self.assertIsInstance(raft, Raft) raft.reload_config({'retry_timeout': 20, 'ttl': 60, 'loop_wait': 10}) self.assertTrue(raft._sync_obj.set(raft.members_path + 'legacy', '{"version":"2.0.0"}')) self.assertTrue(raft.touch_member('')) @@ -139,9 +142,9 @@ class TestRaft(unittest.TestCase): self.assertTrue(raft.set_config_value('{}')) self.assertTrue(raft.write_sync_state('foo', 'bar')) self.assertFalse(raft.write_sync_state('foo', 'bar', 1)) - raft._citus_group = '1' + raft._mpp = get_mpp({'citus': {'group': 1, 'database': 'postgres'}}) self.assertTrue(raft.manual_failover('foo', 'bar')) - raft._citus_group = '0' + raft._mpp = get_mpp({'citus': {'group': 0, 'database': 'postgres'}}) self.assertTrue(raft.take_leader()) cluster = raft.get_cluster() self.assertIsInstance(cluster, Cluster) @@ -157,9 +160,9 @@ class TestRaft(unittest.TestCase): self.assertTrue(raft.delete_sync_state()) self.assertTrue(raft.set_history_value('')) self.assertTrue(raft.delete_cluster()) - raft._citus_group = '1' + raft._mpp = get_mpp({'citus': {'group': 1, 'database': 'postgres'}}) self.assertTrue(raft.delete_cluster()) - raft._citus_group = None + raft._mpp = get_mpp({}) raft.get_cluster() raft.watch(None, 0.001) raft._sync_obj.destroy() @@ -175,5 +178,5 @@ class TestRaft(unittest.TestCase): def test_init(self, mock_event, mock_kvstore): mock_kvstore.return_value.applied_local_log = False mock_event.return_value.is_set.side_effect = [False, True] - self.assertIsNotNone(Raft({'ttl': 30, 'scope': 'test', 'name': 'pg', 'patronictl': True, - 'self_addr': '1', 'data_dir': self._TMP})) + self.assertIsInstance(get_dcs({'ttl': 30, 'scope': 'test', 'name': 'pg', 'patronictl': True, + 'raft': {'self_addr': '1', 'data_dir': self._TMP}}), Raft) diff --git a/tests/test_zookeeper.py b/tests/test_zookeeper.py index 3ce3ea75..3cd03467 100644 --- a/tests/test_zookeeper.py +++ b/tests/test_zookeeper.py @@ -7,8 +7,10 @@ from kazoo.handlers.threading import SequentialThreadingHandler from kazoo.protocol.states import KeeperState, WatchedEvent, ZnodeStat from kazoo.retry import RetryFailedError from mock import Mock, PropertyMock, patch +from patroni.dcs import get_dcs from patroni.dcs.zookeeper import Cluster, PatroniKazooClient, \ PatroniSequentialThreadingHandler, ZooKeeper, ZooKeeperError +from patroni.postgresql.mpp import get_mpp class MockKazooClient(Mock): @@ -148,9 +150,9 @@ class TestZooKeeper(unittest.TestCase): @patch('patroni.dcs.zookeeper.PatroniKazooClient', MockKazooClient) def setUp(self): - self.zk = ZooKeeper({'hosts': ['localhost:2181'], 'scope': 'test', - 'name': 'foo', 'ttl': 30, 'retry_timeout': 10, 'loop_wait': 10, - 'set_acls': {'CN=principal2': ['ALL']}}) + self.zk = get_dcs({'scope': 'test', 'name': 'foo', 'ttl': 30, 'retry_timeout': 10, 'loop_wait': 10, + 'zookeeper': {'hosts': ['localhost:2181'], 'set_acls': {'CN=principal2': ['ALL']}}}) + self.assertIsInstance(self.zk, ZooKeeper) def test_reload_config(self): self.zk.reload_config({'ttl': 20, 'retry_timeout': 10, 'loop_wait': 10}) @@ -177,7 +179,7 @@ class TestZooKeeper(unittest.TestCase): self.assertEqual(cluster.last_lsn, 500) def test__get_citus_cluster(self): - self.zk._citus_group = '0' + self.zk._mpp = get_mpp({'citus': {'group': 0, 'database': 'postgres'}}) for _ in range(0, 2): cluster = self.zk.get_cluster() self.assertIsInstance(cluster, Cluster) From dd548c49645c78d0b85cc723bf3bec500528c3da Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Thu, 21 Dec 2023 09:25:51 +0100 Subject: [PATCH 14/18] Create citus database and extension idempotently (#2990) Consider a task: we want to create an extension _before_ citus in a database. Currently `post_bootstrab` script is executed before `CitusHandler.bootstrap()` method, which seems to allow doing that, but in fact `CitusHandler.bootstrap()` will fail to create already existing database and as a result the whole bootstrap will fail. Changing the order of execution of `post_bootstrab` hook and `CitusHandler.bootstrap()` seems to be useless, because it will not allow creating another extension _before_ citus. Therefore the only way of solving it is making CREATE DATABASE and CREATE EXTENSION idempotent. It will allow to create citus database and all dependencies from the `post_bootstrab` hook. --- patroni/postgresql/mpp/citus.py | 15 +++++++++++---- 1 file changed, 11 insertions(+), 4 deletions(-) diff --git a/patroni/postgresql/mpp/citus.py b/patroni/postgresql/mpp/citus.py index a17d1435..b8c205ce 100644 --- a/patroni/postgresql/mpp/citus.py +++ b/patroni/postgresql/mpp/citus.py @@ -8,7 +8,7 @@ from typing import Any, Dict, List, Optional, Union, Tuple, TYPE_CHECKING from . import AbstractMPP, AbstractMPPHandler from ...dcs import Cluster -from ...psycopg import connect, quote_ident +from ...psycopg import connect, quote_ident, quote_literal from ...utils import parse_int if TYPE_CHECKING: # pragma: no cover @@ -389,9 +389,16 @@ class CitusHandler(Citus, AbstractMPPHandler, Thread): if self._config['database'] != self._postgresql.database: conn = connect(**conn_kwargs) try: + database = self._config['database'] + sql = """DO $$ +BEGIN + PERFORM * FROM pg_catalog.pg_database WHERE datname = {0}; + IF NOT FOUND THEN + CREATE DATABASE {1}; + END IF; +END;$$""".format(quote_literal(database), quote_ident(database, conn)) with conn.cursor() as cur: - cur.execute('CREATE DATABASE {0}'.format( - quote_ident(self._config['database'], conn)).encode('utf-8')) + cur.execute(sql.encode('utf-8')) finally: conn.close() @@ -399,7 +406,7 @@ class CitusHandler(Citus, AbstractMPPHandler, Thread): conn = connect(**conn_kwargs) try: with conn.cursor() as cur: - cur.execute('CREATE EXTENSION citus') + cur.execute('CREATE EXTENSION IF NOT EXISTS citus') superuser = self._postgresql.config.superuser params = {k: superuser[k] for k in ('password', 'sslcert', 'sslkey') if k in superuser} From 8acefefc4238b9d32cda2e8670ba53e212837c7c Mon Sep 17 00:00:00 2001 From: zhjwpku Date: Fri, 29 Dec 2023 16:01:46 +0800 Subject: [PATCH 15/18] Fix Citus bootstrap - CREATE DATABASE cannot be executed from a function (#2994) This was introduced by #2990: pod cannot be started and show the following logs: ``` 2023-12-26 03:29:25.569 UTC [47] CONTEXT: SQL statement "CREATE DATABASE "citus"" PL/pgSQL function inline_code_block line 5 at SQL statement 2023-12-26 03:29:25.569 UTC [47] STATEMENT: DO $$ BEGIN PERFORM * FROM pg_catalog.pg_database WHERE datname = 'citus'; IF NOT FOUND THEN CREATE DATABASE "citus"; END IF; END;$$ 2023-12-26 03:29:25,570 ERROR: post_bootstrap Traceback (most recent call last): File "/usr/local/lib/python3.11/dist-packages/patroni/postgresql/bootstrap.py", line 474, in post_bootstrap self._postgresql.citus_handler.bootstrap() File "/usr/local/lib/python3.11/dist-packages/patroni/postgresql/mpp/citus.py", line 401, in bootstrap cur.execute(sql.encode('utf-8')) psycopg2.errors.ActiveSqlTransaction: CREATE DATABASE cannot be executed from a function CONTEXT: SQL statement "CREATE DATABASE "citus"" PL/pgSQL function inline_code_block line 5 at SQL statement ``` --------- Signed-off-by: Zhao Junwang --- patroni/postgresql/mpp/citus.py | 15 +++++---------- patroni/psycopg.py | 5 ++++- tests/__init__.py | 2 ++ tests/test_citus.py | 8 ++++++++ 4 files changed, 19 insertions(+), 11 deletions(-) diff --git a/patroni/postgresql/mpp/citus.py b/patroni/postgresql/mpp/citus.py index b8c205ce..f3d6394c 100644 --- a/patroni/postgresql/mpp/citus.py +++ b/patroni/postgresql/mpp/citus.py @@ -8,7 +8,7 @@ from typing import Any, Dict, List, Optional, Union, Tuple, TYPE_CHECKING from . import AbstractMPP, AbstractMPPHandler from ...dcs import Cluster -from ...psycopg import connect, quote_ident, quote_literal +from ...psycopg import connect, quote_ident, DuplicateDatabase from ...utils import parse_int if TYPE_CHECKING: # pragma: no cover @@ -389,16 +389,11 @@ class CitusHandler(Citus, AbstractMPPHandler, Thread): if self._config['database'] != self._postgresql.database: conn = connect(**conn_kwargs) try: - database = self._config['database'] - sql = """DO $$ -BEGIN - PERFORM * FROM pg_catalog.pg_database WHERE datname = {0}; - IF NOT FOUND THEN - CREATE DATABASE {1}; - END IF; -END;$$""".format(quote_literal(database), quote_ident(database, conn)) with conn.cursor() as cur: - cur.execute(sql.encode('utf-8')) + cur.execute('CREATE DATABASE {0}'.format( + quote_ident(self._config['database'], conn)).encode('utf-8')) + except DuplicateDatabase as e: + logger.debug('Exception when creating database: %r', e) finally: conn.close() diff --git a/patroni/psycopg.py b/patroni/psycopg.py index 4a92047c..5d47ad5c 100644 --- a/patroni/psycopg.py +++ b/patroni/psycopg.py @@ -9,7 +9,8 @@ if TYPE_CHECKING: # pragma: no cover from psycopg import Connection from psycopg2 import connection, cursor -__all__ = ['connect', 'quote_ident', 'quote_literal', 'DatabaseError', 'Error', 'OperationalError', 'ProgrammingError'] +__all__ = ['connect', 'quote_ident', 'quote_literal', 'DatabaseError', 'Error', 'OperationalError', 'ProgrammingError', + 'DuplicateDatabase'] _legacy = False try: @@ -18,6 +19,7 @@ try: if parse_version(__version__) < MIN_PSYCOPG2: raise ImportError from psycopg2 import connect as _connect, Error, DatabaseError, OperationalError, ProgrammingError + from psycopg2.errors import DuplicateDatabase from psycopg2.extensions import adapt try: @@ -43,6 +45,7 @@ try: return value.getquoted().decode('utf-8') except ImportError: from psycopg import connect as __connect, sql, Error, DatabaseError, OperationalError, ProgrammingError + from psycopg.errors import DuplicateDatabase def _connect(dsn: Optional[str] = None, **kwargs: Any) -> 'Connection[Any]': """Call :func:`psycopg.connect` with *dsn* and ``**kwargs``. diff --git a/tests/__init__.py b/tests/__init__.py index 986bd88b..2f3730f6 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -129,6 +129,8 @@ class MockCursor(object): sql = sql.decode('utf-8') if sql.startswith('blabla'): raise psycopg.ProgrammingError() + if sql.startswith('CREATE DATABASE'): + raise psycopg.DuplicateDatabase() elif sql == 'CHECKPOINT' or sql.startswith('SELECT pg_catalog.pg_create_'): raise psycopg.OperationalError() elif sql.startswith('RetryFailedError'): diff --git a/tests/test_citus.py b/tests/test_citus.py index dbf6d9cf..f1d8a020 100644 --- a/tests/test_citus.py +++ b/tests/test_citus.py @@ -157,3 +157,11 @@ class TestCitus(BaseTestPostgresql): 'type': 'logical', 'database': 'citus', 'plugin': 'pgoutput'})) self.assertTrue(self.c.ignore_replication_slot({'name': 'citus_shard_split_slot_1_2_3', 'type': 'logical', 'database': 'citus', 'plugin': 'citus'})) + + @patch('patroni.postgresql.mpp.citus.logger.debug') + @patch('patroni.postgresql.mpp.citus.connect', psycopg_connect) + @patch('patroni.postgresql.mpp.citus.quote_ident', Mock()) + def test_bootstrap_duplicate_database(self, mock_logger): + self.c.bootstrap() + mock_logger.assert_called_once() + self.assertTrue(mock_logger.call_args[0][0].startswith('Exception when creating database')) From 71ccf91e3672be7879f5e5d7317d13b58f87220f Mon Sep 17 00:00:00 2001 From: Polina Bungina <27892524+hughcapet@users.noreply.github.com> Date: Tue, 2 Jan 2024 11:30:18 +0300 Subject: [PATCH 16/18] Don't filter out contradictory nofailover tag (#2992) * Ensure that nofailover will always be used if both nofailover and failover_priority tags are provided * Call _validate_failover_tags from reload_local_configuration() as well * Properly check values in the _validate_failover_tags(): nofailover value should be casted to boolean like it is done when accessed in other places --- features/priority_failover.feature | 18 ++++++++ features/steps/basic_replication.py | 2 +- features/steps/patroni_api.py | 6 +++ patroni/config.py | 13 +++--- patroni/tags.py | 10 +++-- postgres0.yml | 2 +- postgres1.yml | 2 +- postgres2.yml | 2 +- tests/test_config.py | 68 +++++++++++++---------------- tests/test_patroni.py | 16 +++++++ 10 files changed, 89 insertions(+), 50 deletions(-) diff --git a/features/priority_failover.feature b/features/priority_failover.feature index b33dd045..acb1cb5a 100644 --- a/features/priority_failover.feature +++ b/features/priority_failover.feature @@ -21,3 +21,21 @@ Feature: priority replication And I sleep for 5 seconds Then postgres3 role is the primary after 10 seconds And there is one of ["postgres3 has equally tolerable WAL position and priority 2, while this node has priority 1","Wal position of postgres3 is ahead of my wal position"] INFO in the postgres2 patroni log after 5 seconds + + Scenario: check conflicting configuration handling + When I set nofailover tag in postgres2 config + And I issue an empty POST request to http://127.0.0.1:8010/reload + Then I receive a response code 202 + And there is one of ["Conflicting configuration between nofailover: True and failover_priority: 1. Defaulting to nofailover: True"] WARNING in the postgres2 patroni log after 5 seconds + And "members/postgres2" key in DCS has tags={'failover_priority': '1', 'nofailover': True} after 10 seconds + When I issue a POST request to http://127.0.0.1:8010/failover with {"candidate": "postgres2"} + Then I receive a response code 412 + And I receive a response text "failover is not possible: no good candidates have been found" + When I reset nofailover tag in postgres1 config + And I issue an empty POST request to http://127.0.0.1:8009/reload + Then I receive a response code 202 + And there is one of ["Conflicting configuration between nofailover: False and failover_priority: 0. Defaulting to nofailover: False"] WARNING in the postgres1 patroni log after 5 seconds + And "members/postgres1" key in DCS has tags={'failover_priority': '0', 'nofailover': False} after 10 seconds + And I issue a POST request to http://127.0.0.1:8010/failover with {"candidate": "postgres1"} + Then I receive a response code 200 + And postgres1 role is the primary after 10 seconds diff --git a/features/steps/basic_replication.py b/features/steps/basic_replication.py index d70c6d0e..f2db7110 100644 --- a/features/steps/basic_replication.py +++ b/features/steps/basic_replication.py @@ -123,6 +123,6 @@ def check_patroni_log(context, message_list, level, node, timeout): messsages_of_level = context.pctl.read_patroni_log(node, level) if any(any(message in line for line in messsages_of_level) for message in message_list): break - time.sleep(1) + sleep(1) else: assert False, f"There were none of {message_list} {level} in the {node} patroni log after {timeout} seconds" diff --git a/features/steps/patroni_api.py b/features/steps/patroni_api.py index 2c76d32d..74a7c0da 100644 --- a/features/steps/patroni_api.py +++ b/features/steps/patroni_api.py @@ -128,6 +128,12 @@ def scheduled_restart(context, url, in_seconds, data): context.execute_steps(u"""Given I issue a POST request to {0}/restart with {1}""".format(url, json.dumps(data))) +@step('I {action:w} {tag:w} tag in {pg_name:w} config') +def add_bool_tag_to_config(context, action, tag, pg_name): + value = action == 'set' + context.pctl.add_tag_to_config(pg_name, tag, value) + + @step('I add tag {tag:w} {value:w} to {pg_name:w} config') def add_tag_to_config(context, tag, value, pg_name): context.pctl.add_tag_to_config(pg_name, tag, value) diff --git a/patroni/config.py b/patroni/config.py index e523bc08..f7d648d5 100644 --- a/patroni/config.py +++ b/patroni/config.py @@ -142,10 +142,10 @@ class Config(object): self.__effective_configuration = self._build_effective_configuration({}, self._local_configuration) self._data_dir = self.__effective_configuration.get('postgresql', {}).get('data_dir', "") self._cache_file = os.path.join(self._data_dir, self.__CACHE_FILENAME) - if validator: # patronictl uses validator=None and we don't want to load anything from local cache in this case - self._load_cache() + if validator: # patronictl uses validator=None + self._load_cache() # we don't want to load anything from local cache for ctl + self._validate_failover_tags() # irrelevant for ctl self._cache_needs_saving = False - self._validate_failover_tags() @property def config_file(self) -> Optional[str]: @@ -356,6 +356,7 @@ class Config(object): new_configuration = self._build_effective_configuration(self._dynamic_configuration, configuration) self._local_configuration = configuration self.__effective_configuration = new_configuration + self._validate_failover_tags() return True else: logger.info('No local configuration items changed.') @@ -814,10 +815,12 @@ class Config(object): bedrock source of truth) """ tags = self.get('tags', {}) + if 'nofailover' not in tags: + return nofailover_tag = tags.get('nofailover') failover_priority_tag = parse_int(tags.get('failover_priority')) if failover_priority_tag is not None \ - and (nofailover_tag is True and failover_priority_tag > 0 - or nofailover_tag is False and failover_priority_tag <= 0): + and (bool(nofailover_tag) is True and failover_priority_tag > 0 + or bool(nofailover_tag) is False and failover_priority_tag <= 0): logger.warning('Conflicting configuration between nofailover: %s and failover_priority: %s. ' 'Defaulting to nofailover: %s', nofailover_tag, failover_priority_tag, nofailover_tag) diff --git a/patroni/tags.py b/patroni/tags.py index 998ff693..eedc9674 100644 --- a/patroni/tags.py +++ b/patroni/tags.py @@ -22,14 +22,18 @@ class Tags(abc.ABC): A custom tag is any tag added to the configuration ``tags`` section that is not one of ``clonefrom``, ``nofailover``, ``noloadbalance`` or ``nosync``. - For the Patroni predefined tags, the returning object will only contain them if they are enabled as they - all are boolean values that default to disabled. + For most of the Patroni predefined tags, the returning object will only contain them if they are enabled as + they all are boolean values that default to disabled. + However ``nofailover`` tag is always returned if ``failover_priority`` tag is defined. In this case, we need + both values to see if they are contradictory and the ``nofailover`` value should be used. :returns: a dictionary of tags set for this node. The key is the tag name, and the value is the corresponding tag value. """ return {tag: value for tag, value in tags.items() - if tag not in ('clonefrom', 'nofailover', 'noloadbalance', 'nosync') or value} + if any((tag not in ('clonefrom', 'nofailover', 'noloadbalance', 'nosync'), + value, + tag == 'nofailover' and 'failover_priority' in tags))} @property @abc.abstractmethod diff --git a/postgres0.yml b/postgres0.yml index 8a975156..84796a46 100644 --- a/postgres0.yml +++ b/postgres0.yml @@ -132,7 +132,7 @@ postgresql: # safety_margin: 5 tags: - nofailover: false + # failover_priority: 1 noloadbalance: false clonefrom: false nosync: false diff --git a/postgres1.yml b/postgres1.yml index 6ca2aa64..c86e8790 100644 --- a/postgres1.yml +++ b/postgres1.yml @@ -124,6 +124,6 @@ postgresql: #pre_promote: /path/to/pre_promote.sh tags: - nofailover: false + # failover_priority: 1 noloadbalance: false clonefrom: false diff --git a/postgres2.yml b/postgres2.yml index ee61a023..7384568e 100644 --- a/postgres2.yml +++ b/postgres2.yml @@ -114,7 +114,7 @@ postgresql: # krb_server_keyfile: /var/spool/keytabs/postgres unix_socket_directories: '..' # parent directory of data_dir tags: - nofailover: false + # failover_priority: 1 noloadbalance: false clonefrom: false # replicatefrom: postgresql1 diff --git a/tests/test_config.py b/tests/test_config.py index 7bf01f56..a02a33fd 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -159,48 +159,40 @@ class TestConfig(unittest.TestCase): @patch('patroni.config.logger') def test__validate_failover_tags(self, mock_logger, mock_get): """Ensures that only one of `nofailover` or `failover_priority` can be provided""" - mock_logger.warning.reset_mock() config = Config("postgres0.yml") + # Providing one of `nofailover` or `failover_priority` is fine - just_nofailover = {"nofailover": True} - mock_get.side_effect = [just_nofailover] * 2 - self.assertIsNone(config._validate_failover_tags()) - mock_logger.warning.assert_not_called() - just_failover_priority = {"failover_priority": 1} - mock_get.side_effect = [just_failover_priority] * 2 - self.assertIsNone(config._validate_failover_tags()) - mock_logger.warning.assert_not_called() + for single_param in ({"nofailover": True}, {"failover_priority": 1}, {"failover_priority": 0}): + mock_get.side_effect = [single_param] * 2 + self.assertIsNone(config._validate_failover_tags()) + mock_logger.warning.assert_not_called() + # Providing both `nofailover` and `failover_priority` is fine if consistent - consistent_false = {"nofailover": False, "failover_priority": 1} - mock_get.side_effect = [consistent_false] * 2 - self.assertIsNone(config._validate_failover_tags()) - mock_logger.warning.assert_not_called() - consistent_true = {"nofailover": True, "failover_priority": 0} - mock_get.side_effect = [consistent_true] * 2 - self.assertIsNone(config._validate_failover_tags()) - mock_logger.warning.assert_not_called() + for consistent_state in ( + {"nofailover": False, "failover_priority": 1}, + {"nofailover": True, "failover_priority": 0}, + {"nofailover": "False", "failover_priority": 0} + ): + mock_get.side_effect = [consistent_state] * 2 + self.assertIsNone(config._validate_failover_tags()) + mock_logger.warning.assert_not_called() + # Providing both inconsistently should log a warning - inconsistent_false = {"nofailover": False, "failover_priority": 0} - mock_get.side_effect = [inconsistent_false] * 2 - self.assertIsNone(config._validate_failover_tags()) - mock_logger.warning.assert_called_once_with( - 'Conflicting configuration between nofailover: %s and failover_priority: %s.' - + ' Defaulting to nofailover: %s', - False, - 0, - False - ) - mock_logger.warning.reset_mock() - inconsistent_true = {"nofailover": True, "failover_priority": 1} - mock_get.side_effect = [inconsistent_true] * 2 - self.assertIsNone(config._validate_failover_tags()) - mock_logger.warning.assert_called_once_with( - 'Conflicting configuration between nofailover: %s and failover_priority: %s.' - + ' Defaulting to nofailover: %s', - True, - 1, - True - ) + for inconsistent_state in ( + {"nofailover": False, "failover_priority": 0}, + {"nofailover": True, "failover_priority": 1}, + {"nofailover": "False", "failover_priority": 1}, + {"nofailover": "", "failover_priority": 0} + ): + mock_get.side_effect = [inconsistent_state] * 2 + self.assertIsNone(config._validate_failover_tags()) + mock_logger.warning.assert_called_once_with( + 'Conflicting configuration between nofailover: %s and failover_priority: %s.' + + ' Defaulting to nofailover: %s', + inconsistent_state['nofailover'], + inconsistent_state['failover_priority'], + inconsistent_state['nofailover']) + mock_logger.warning.reset_mock() def test__process_postgresql_parameters(self): expected_params = { diff --git a/tests/test_patroni.py b/tests/test_patroni.py index 8e08d406..2f8428b1 100644 --- a/tests/test_patroni.py +++ b/tests/test_patroni.py @@ -175,6 +175,20 @@ class TestPatroni(unittest.TestCase): self.p.next_run = time.time() - self.p.dcs.loop_wait - 1 self.p.schedule_next_run() + def test__filter_tags(self): + tags = {'noloadbalance': False, 'clonefrom': False, 'nosync': False, 'smth': 'random'} + self.assertEqual(self.p._filter_tags(tags), {'smth': 'random'}) + + tags['clonefrom'] = True + tags['smth'] = False + self.assertEqual(self.p._filter_tags(tags), {'clonefrom': True, 'smth': False}) + + tags = {'nofailover': False, 'failover_priority': 0} + self.assertEqual(self.p._filter_tags(tags), tags) + + tags = {'nofailover': True, 'failover_priority': 1} + self.assertEqual(self.p._filter_tags(tags), tags) + def test_noloadbalance(self): self.p.tags['noloadbalance'] = True self.assertTrue(self.p.noloadbalance) @@ -186,9 +200,11 @@ class TestPatroni(unittest.TestCase): # Setting `nofailover: True` has precedence (True, 0, True), (True, 1, True), + ('False', 1, True), # because we use bool() for the value # Similarly, setting `nofailover: False` has precedence (False, 0, False), (False, 1, False), + ('', 0, False), # Only when we have `nofailover: None` should we got based on priority (None, 0, True), (None, 1, False), From 3390ee9dea84ad00eed7fba901922dc4e25d9b53 Mon Sep 17 00:00:00 2001 From: Sophia Ruan <104968314+XiuhuaRuan@users.noreply.github.com> Date: Thu, 4 Jan 2024 19:30:03 +0800 Subject: [PATCH 17/18] call freeze_support in main module to solve pyinstaller frozen issue (#2996) Close #2995 --- patroni/__main__.py | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/patroni/__main__.py b/patroni/__main__.py index 2c253da4..c4c9a497 100644 --- a/patroni/__main__.py +++ b/patroni/__main__.py @@ -229,11 +229,6 @@ def patroni_main(configfile: str) -> None: :param configfile: path to Patroni configuration file. """ - from multiprocessing import freeze_support - - # Windows executables created by PyInstaller are frozen, thus we need to enable frozen support for - # :mod:`multiprocessing` to avoid :class:`RuntimeError` exceptions. - freeze_support() abstract_main(Patroni, configfile) @@ -335,6 +330,12 @@ 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 multiprocessing import freeze_support + + # Executables created by PyInstaller are frozen, thus we need to enable frozen support for + # :mod:`multiprocessing` to avoid :class:`RuntimeError` exceptions. + freeze_support() + check_psycopg() args = process_arguments() From 4e5b2ee2491847ae8787c17f7d20bd3e7aacfad3 Mon Sep 17 00:00:00 2001 From: Israel Date: Thu, 4 Jan 2024 08:30:28 -0300 Subject: [PATCH 18/18] Close the doors for a possible future bug in the config generator (#3000) The `AbstractConfigGenerator._format_config` method was missing a comma in the declaration of a tuple. As a consequence it was concatenating the strings `ctl` and `citus` instead of creating two separate items in the tuple. There is currently no observed bug from that issue in the code because the template configuration created by the method `AbstractConfigGenerator.get_template_config` doesn't include either of `ctl` or `citus` keys. However, it is still important that we close the doors for possible future bugs that would come up if we ever attempt to use either of those keys in the template, for example. References: PAT-231. --- patroni/config_generator.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/patroni/config_generator.py b/patroni/config_generator.py index c2b133c6..3593bea3 100644 --- a/patroni/config_generator.py +++ b/patroni/config_generator.py @@ -178,7 +178,7 @@ class AbstractConfigGenerator(abc.ABC): :yields: formatted lines or blocks that represent a text output of the YAML document. """ - for name in ('scope', 'namespace', 'name', 'log', 'restapi', 'ctl' 'citus', + for name in ('scope', 'namespace', 'name', 'log', 'restapi', 'ctl', 'citus', 'consul', 'etcd', 'etcd3', 'exhibitor', 'kubernetes', 'raft', 'zookeeper'): yield from self._format_config_section(name)