From fe16c3610e9d81b1c893f4ef00815b57e1eb3dcb Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 26 Sep 2023 08:29:35 +0200 Subject: [PATCH 1/5] Silence annoying warnings when checking for node uniqueness (#2878) WARNING messages are produced by `urllib3` if Patroni is quickly restarted. Instead we will check that the node is listen on a given port. This fact is actually enough to detect names clashes, while HTTP request could raise an exception is a few other cases, what might case false negatives. Close https://github.com/zalando/patroni/issues/2881 --- patroni/__main__.py | 11 ++++++++--- tests/test_patroni.py | 9 +++++---- 2 files changed, 13 insertions(+), 7 deletions(-) diff --git a/patroni/__main__.py b/patroni/__main__.py index 2b318a67..7d56172b 100644 --- a/patroni/__main__.py +++ b/patroni/__main__.py @@ -106,6 +106,8 @@ class Patroni(AbstractPatroniDaemon, Tags): def ensure_unique_name(self) -> None: """A helper method to prevent splitbrain from operator naming error.""" + from urllib.parse import urlparse + from urllib3.connection import HTTPConnection from patroni.dcs import Member cluster = self.dcs.get_cluster() @@ -115,9 +117,12 @@ class Patroni(AbstractPatroniDaemon, Tags): if not isinstance(member, Member): return try: - _ = self.request(member, endpoint="/liveness", timeout=3) - logger.fatal("Can't start; there is already a node named '%s' running", self.config['name']) - sys.exit(1) + parts = urlparse(member.api_url) + if isinstance(parts.hostname, str): + connection = HTTPConnection(parts.hostname, port=parts.port or 80, timeout=3) + connection.connect() + logger.fatal("Can't start; there is already a node named '%s' running", self.config['name']) + sys.exit(1) except Exception: return diff --git a/tests/test_patroni.py b/tests/test_patroni.py index 0385731c..df59677d 100644 --- a/tests/test_patroni.py +++ b/tests/test_patroni.py @@ -40,7 +40,7 @@ class MockFrozenImporter(object): @patch('time.sleep', Mock()) @patch('subprocess.call', Mock(return_value=0)) @patch('patroni.psycopg.connect', psycopg_connect) -@patch('urllib3.PoolManager.request', Mock(side_effect=Exception)) +@patch('urllib3.connection.HTTPConnection.connect', Mock(side_effect=Exception)) @patch.object(ConfigHandler, 'append_pg_hba', Mock()) @patch.object(ConfigHandler, 'write_postgresql_conf', Mock()) @patch.object(ConfigHandler, 'write_recovery_conf', Mock()) @@ -64,7 +64,7 @@ class TestPatroni(unittest.TestCase): self.assertRaises(SystemExit, _main) @patch('pkgutil.iter_importers', Mock(return_value=[MockFrozenImporter()])) - @patch('urllib3.PoolManager.request', Mock(side_effect=Exception)) + @patch('urllib3.connection.HTTPConnection.connect', Mock(side_effect=Exception)) @patch('sys.frozen', Mock(return_value=True), create=True) @patch.object(HTTPServer, '__init__', Mock()) @patch.object(etcd.Client, 'read', etcd_read) @@ -108,6 +108,7 @@ class TestPatroni(unittest.TestCase): @patch('os.getpid') @patch('multiprocessing.Process') @patch('patroni.__main__.patroni_main', Mock()) + @patch('sys.argv', ['patroni.py', 'postgres0.yml']) def test_patroni_main(self, mock_process, mock_getpid): mock_getpid.return_value = 2 _main() @@ -233,8 +234,8 @@ class TestPatroni(unittest.TestCase): ) with patch('patroni.dcs.AbstractDCS.get_cluster', Mock(return_value=bad_cluster)): # If the api of the running node cannot be reached, this implies unique name - with patch.object(self.p, 'request', Mock(side_effect=ConnectionError)): + with patch('urllib3.connection.HTTPConnection.connect', Mock(side_effect=ConnectionError)): self.assertIsNone(self.p.ensure_unique_name()) # Only if the api of the running node is reachable do we throw an error - with patch.object(self.p, 'request', Mock()): + with patch('urllib3.connection.HTTPConnection.connect', Mock()): self.assertRaises(SystemExit, self.p.ensure_unique_name) From 2bd821a7688fc2840d0ec3725b82aceb52a5470c Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 26 Sep 2023 08:31:55 +0200 Subject: [PATCH 2/5] Bugfix for GUC's values with units (#2883) Despite being validated by `IntValidator` some GUC's couldn't be casted directly to `int` because they include suffix. Example: `128MB`. Close https://github.com/zalando/patroni/issues/2879 --- patroni/config.py | 3 ++- tests/test_config.py | 1 + 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/patroni/config.py b/patroni/config.py index 1650934d..fee2147a 100644 --- a/patroni/config.py +++ b/patroni/config.py @@ -490,7 +490,8 @@ class Config(object): elif not is_local: validator = ConfigHandler.CMDLINE_OPTIONS[name][1] if validator(value): - pg_params[name] = int(value) if isinstance(validator, IntValidator) else value + int_val = parse_int(value) if isinstance(validator, IntValidator) else None + pg_params[name] = int_val if isinstance(int_val, int) else value else: logger.warning("postgresql parameter %s=%s failed validation, defaulting to %s", name, value, ConfigHandler.CMDLINE_OPTIONS[name][0]) diff --git a/tests/test_config.py b/tests/test_config.py index bd8d0a90..f0a780bc 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -155,6 +155,7 @@ class TestConfig(unittest.TestCase): expected_params = { 'f.oo': 'bar', # not in ConfigHandler.CMDLINE_OPTIONS 'max_connections': 100, # IntValidator + 'wal_keep_size': '128MB', # IntValidator 'wal_level': 'hot_standby', # EnumValidator } input_params = deepcopy(expected_params) From 48514db84b631a69e6191642832be378a0c63a4f Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 26 Sep 2023 09:12:31 +0200 Subject: [PATCH 3/5] Take into account current role when deciding on removal of member ZNode (#2884) Patroni doesn't watch on all changes of member keys in order to not create too much load on ZooKeeper, but only subscribes to changes (ZNodes added or deleted) in the `/member` directory. Therefore when some important fields in the value are updated we remove and recreate ZNode in order to notify the leader or other members. The leader should remove the member key only when the `checkpoint_after_promote` value is changed and replicas when the `state` is changed to/from `running`. We don't care about the `version` field, because Patroni version can't be changed without restart, what will case ZooKeeper `session_id` to change it anyway. This fix hopefully will reduce failures of behave tests on GH Actions. --- features/patroni_api.feature | 3 ++- patroni/dcs/zookeeper.py | 37 +++++++++++++++++++++--------------- tests/test_zookeeper.py | 1 + 3 files changed, 25 insertions(+), 16 deletions(-) diff --git a/features/patroni_api.feature b/features/patroni_api.feature index 624d3271..ff06f3c9 100644 --- a/features/patroni_api.feature +++ b/features/patroni_api.feature @@ -68,6 +68,7 @@ Scenario: check API requests for the primary-replica pair in the pause mode When I kill postmaster on postgres1 And I issue a GET request to http://127.0.0.1:8009/replica Then I receive a response code 503 + And "members/postgres1" key in DCS has state=stopped after 10 seconds When I run patronictl.py restart batman postgres1 --force Then I receive a response returncode 0 Then replication works from postgres0 to postgres1 after 20 seconds @@ -76,7 +77,7 @@ Scenario: check API requests for the primary-replica pair in the pause mode Then I receive a response code 200 And I receive a response state running And I receive a response role replica - When I run patronictl.py reinit batman postgres1 --force + When I run patronictl.py reinit batman postgres1 --force --wait Then I receive a response returncode 0 And I receive a response output "Success: reinitialize for member postgres1" And postgres1 role is the secondary after 30 seconds diff --git a/patroni/dcs/zookeeper.py b/patroni/dcs/zookeeper.py index 6390aa2e..863d7b40 100644 --- a/patroni/dcs/zookeeper.py +++ b/patroni/dcs/zookeeper.py @@ -379,21 +379,28 @@ class ZooKeeper(AbstractDCS): cluster = self.cluster member = cluster and cluster.get_member(self._name, fallback_to_leader=False) member_data = self.__last_member_data or member and member.data - # We want to notify leader if some important fields in the member key changed by removing ZNode - if member and (self._client.client_id is not None and member.session != self._client.client_id[0] - or not (member_data and deep_compare(member_data.get('tags', {}), data.get('tags', {})) - and (member_data.get('state') == data.get('state') - or 'running' not in (member_data.get('state'), data.get('state'))) - and member_data.get('version') == data.get('version') - and member_data.get('checkpoint_after_promote') - == data.get('checkpoint_after_promote'))): - try: - self._client.delete_async(self.member_path).get(timeout=1) - except NoNodeError: - pass - except Exception: - return False - member = None + if member and member_data: + is_leader = data.get('role') in ('master', 'primary', 'standby_leader') + checkpoint_after_promote_changed = member_data.get('checkpoint_after_promote') \ + != data.get('checkpoint_after_promote') + state_running_changed = member_data.get('state') != data.get('state') \ + and 'running' in (member_data.get('state'), data.get('state')) + tags_changed = not deep_compare(member_data.get('tags', {}), data.get('tags', {})) + + # We want delete the member ZNode if: + # - our session doesn't match with session id on our member key; or + # - we want to notify leader if some important fields in the member key changed; or + # - if we are the leader and want to notify replicas about checkpoint_after_promote; + if self._client.client_id is not None and member.session != self._client.client_id[0] \ + or is_leader and checkpoint_after_promote_changed \ + or not is_leader and (state_running_changed or tags_changed): + try: + self._client.delete_async(self.member_path).get(timeout=1) + except NoNodeError: + pass + except Exception: + return False + member = None encoded_data = json.dumps(data, separators=(',', ':')).encode('utf-8') if member and member_data: diff --git a/tests/test_zookeeper.py b/tests/test_zookeeper.py index 45aeac10..184fda05 100644 --- a/tests/test_zookeeper.py +++ b/tests/test_zookeeper.py @@ -276,6 +276,7 @@ class TestZooKeeper(unittest.TestCase): self.assertTrue(self.zk.delete_cluster()) def test_watch(self): + self.zk.event.wait = Mock() self.zk.watch(None, 0) self.zk.event.is_set = Mock(return_value=True) self.zk._fetch_status = False From 4c1c804cfd1becc7076c601bb7c5e28320ce568d Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 26 Sep 2023 10:40:51 +0200 Subject: [PATCH 4/5] Read GUC's values when joining running Postgres (#2876) If restarted in pause Patroni was discarding `synchronous_standby_names` from `postgresql.conf` because in the internal cache this values was set to `None`. As a result synchronous replication transitioned to a broken state, with no synchronous replicas according to the `synchronous_standby_names` and Patroni not selecting/setting the new synchronous replicas (another bug). To solve the problem of broken initial state and to avoid similar issues with other GUC's we will read GUC's value if Patroni is joining running Postgres. --- patroni/config_generator.py | 2 +- patroni/postgresql/__init__.py | 25 ++++++++++++++++++------- patroni/postgresql/config.py | 20 ++++++++++++++------ tests/__init__.py | 8 +++++++- tests/test_postgresql.py | 2 ++ 5 files changed, 42 insertions(+), 15 deletions(-) diff --git a/patroni/config_generator.py b/patroni/config_generator.py index 956e18d2..0269c49a 100644 --- a/patroni/config_generator.py +++ b/patroni/config_generator.py @@ -287,7 +287,7 @@ class RunningClusterConfigGenerator(AbstractConfigGenerator): :param cur: connection cursor to use. """ - cur.execute("SELECT name, current_setting(name) FROM pg_settings " + cur.execute("SELECT name, pg_catalog.current_setting(name) FROM pg_catalog.pg_settings " "WHERE context <> 'internal' " "AND source IN ('configuration file', 'command line', 'environment variable') " "AND category <> 'Write-Ahead Log / Recovery Target' " diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py index af48886f..cab33e8f 100644 --- a/patroni/postgresql/__init__.py +++ b/patroni/postgresql/__init__.py @@ -119,17 +119,28 @@ class Postgresql(object): # Last known running process self._postmaster_proc = None - if self.is_running(): # we are "joining" already running postgres - self.set_state('running') + if self.is_running(): + # If we found postmaster process we need to figure out whether postgres is accepting connections + self.set_state('starting') + self.check_startup_state_changed() + + if self.state == 'running': # we are "joining" already running postgres + # we know that PostgreSQL is accepting connections and can read some GUC's from pg_settings + self.config.load_current_server_parameters() + self.set_role('master' if self.is_primary() else 'replica') - # postpone writing postgresql.conf for 12+ because recovery parameters are not yet known - if self.major_version < 120000 or self.is_primary(): - self.config.write_postgresql_conf() + hba_saved = self.config.replace_pg_hba() ident_saved = self.config.replace_pg_ident() - if hba_saved or ident_saved: + + if self.major_version < 120000 or self.role in ('master', 'primary'): + # If PostgreSQL is running as a primary or we run PostgreSQL that is older than 12 we can + # call reload_config() once again (the first call happened in the ConfigHandler constructor), + # so that it can figure out if config files should be updated and pg_ctl reload executed. + self.config.reload_config(config, sighup=bool(hba_saved or ident_saved)) + elif hba_saved or ident_saved: self.reload() - elif self.role in ('master', 'primary'): + elif not self.is_running() and self.role in ('master', 'primary'): self.set_role('demoted') @property diff --git a/patroni/postgresql/config.py b/patroni/postgresql/config.py index a918a2cf..40d87f35 100644 --- a/patroni/postgresql/config.py +++ b/patroni/postgresql/config.py @@ -326,14 +326,22 @@ class ConfigHandler(object): .format(self._pgpass)) self._passfile = None self._passfile_mtime = None - self._synchronous_standby_names = None self._postmaster_ctime = None self._current_recovery_params: Optional[CaseInsensitiveDict] = None self._config = {} self._recovery_params = CaseInsensitiveDict() - self._server_parameters: CaseInsensitiveDict + self._server_parameters: CaseInsensitiveDict = CaseInsensitiveDict() self.reload_config(config) + def load_current_server_parameters(self) -> None: + """Read GUC's values from ``pg_settings`` when Patroni is joining the the postgres that is already running.""" + exclude = [name.lower() for name, value in self.CMDLINE_OPTIONS.items() if value[1] == _false_validator] \ + + [name.lower() for name in self._RECOVERY_PARAMETERS] + self._server_parameters = CaseInsensitiveDict({r[0]: r[1] for r in self._postgresql.query( + "SELECT name, pg_catalog.current_setting(name) FROM pg_catalog.pg_settings" + " WHERE (source IN ('command line', 'environment variable') OR sourcefile = %s)" + " AND pg_catalog.lower(name) != ALL(%s)", self._postgresql_conf, exclude)}) + def setup_server_parameters(self) -> None: self._server_parameters = self.get_server_parameters(self._config) self._adjust_recovery_parameters() @@ -922,14 +930,15 @@ class ConfigHandler(object): listen_addresses, port = split_host_port(config['listen'], 5432) parameters.update(cluster_name=self._postgresql.scope, listen_addresses=listen_addresses, port=str(port)) if not self._postgresql.global_config or self._postgresql.global_config.is_synchronous_mode: - if self._synchronous_standby_names is None: + synchronous_standby_names = self._server_parameters.get('synchronous_standby_names') + if synchronous_standby_names is None: if self._postgresql.global_config and self._postgresql.global_config.is_synchronous_mode_strict\ and self._postgresql.role in ('master', 'primary', 'promoted'): parameters['synchronous_standby_names'] = '*' else: parameters.pop('synchronous_standby_names', None) else: - parameters['synchronous_standby_names'] = self._synchronous_standby_names + parameters['synchronous_standby_names'] = synchronous_standby_names # Handle hot_standby <-> replica rename if parameters.get('wal_level') == ('hot_standby' if self._postgresql.major_version >= 90600 else 'replica'): @@ -1150,12 +1159,11 @@ class ConfigHandler(object): def set_synchronous_standby_names(self, value: Optional[str]) -> Optional[bool]: """Updates synchronous_standby_names and reloads if necessary. :returns: True if value was updated.""" - if value != self._synchronous_standby_names: + if value != self._server_parameters.get('synchronous_standby_names'): if value is None: self._server_parameters.pop('synchronous_standby_names', None) else: self._server_parameters['synchronous_standby_names'] = value - self._synchronous_standby_names = value if self._postgresql.state == 'running': self.write_postgresql_conf() self._postgresql.reload() diff --git a/tests/__init__.py b/tests/__init__.py index 5240cec6..32172926 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -117,7 +117,7 @@ class MockCursor(object): self.results = [(False, 2)] elif sql.startswith('SELECT pg_catalog.pg_postmaster_start_time'): self.results = [(datetime.datetime.now(tzutc),)] - elif sql.startswith('SELECT name, current_setting(name) FROM pg_settings'): + elif sql.startswith('SELECT name, pg_catalog.current_setting(name) FROM pg_catalog.pg_settings'): self.results = [('data_directory', 'data'), ('hba_file', os.path.join('data', 'pg_hba.conf')), ('ident_file', os.path.join('data', 'pg_ident.conf')), @@ -137,6 +137,11 @@ class MockCursor(object): ('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'), @@ -248,6 +253,7 @@ class PostgresInit(unittest.TestCase): class BaseTestPostgresql(PostgresInit): + @patch('time.sleep', Mock()) def setUp(self): super(BaseTestPostgresql, self).setUp() diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index e2fad6ba..6ae69e85 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -721,6 +721,7 @@ class TestPostgresql(BaseTestPostgresql): self.assertEqual(self.p.get_primary_timeline(), 1) @patch.object(Postgresql, 'get_postgres_role_from_data_directory', Mock(return_value='replica')) + @patch.object(Postgresql, 'is_running', Mock(return_value=False)) @patch.object(Bootstrap, 'running_custom_bootstrap', PropertyMock(return_value=True)) @patch.object(Postgresql, 'controldata', Mock(return_value={'max_connections setting': '200', 'max_worker_processes setting': '20', @@ -964,6 +965,7 @@ class TestPostgresql2(BaseTestPostgresql): @patch('patroni.postgresql.CallbackExecutor', Mock()) @patch.object(Postgresql, 'get_major_version', Mock(return_value=140000)) @patch.object(Postgresql, 'is_running', Mock(return_value=True)) + @patch.object(Postgresql, 'is_primary', Mock(return_value=False)) def setUp(self): super(TestPostgresql2, self).setUp() From c855b0bff937e81c6ed0565b5ceb3955a8f38e27 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 26 Sep 2023 11:14:20 +0200 Subject: [PATCH 5/5] Detect and solve inconsistency between /sync and actual sync nodes (#2877) Patroni is changing `synchronous_standby_names` and the `/sync` key in a very specific order, first we add nodes to `synchronous_standby_names` and only after, when they are recognized as synchronous they are added to the `/sync` key. When removing nodes the order is different: they are first removed from the `/sync` key and only after that from the `synchronous_standby_names`. As a result Patroni expects that either actual synchronous nodes will match with the nodes listed in the `/sync` key or that new candidates to synchronous nodes will not match with nodes listed in the `/sync` key. In case if `synchronous_standby_names` was removed from the `postgresql.conf`, manually, or due the the bug (#2876), the state becomes inconsistent because of the wrong order of updates. To solve inconsistent state we introduce additional checks and will update the `/sync` key with actual names of synchronous nodes (usually empty set). --- patroni/ha.py | 8 ++++++++ tests/test_ha.py | 18 ++++++++++++++++++ 2 files changed, 26 insertions(+) diff --git a/patroni/ha.py b/patroni/ha.py index d5d6e16b..3b936e4b 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -691,6 +691,14 @@ class Ha(object): current = CaseInsensitiveSet(sync.members) picked, allow_promote = self.state_handler.sync_handler.current_state(self.cluster) + if picked == current and current != allow_promote: + logger.warning('Inconsistent state between synchronous_standby_names = %s and /sync = %s key ' + 'detected, updating synchronous replication key...', list(allow_promote), list(current)) + sync = self.dcs.write_sync_state(self.state_handler.name, allow_promote, version=sync.version) + if not sync: + return logger.warning("Updating sync state failed") + current = CaseInsensitiveSet(sync.members) + if picked != current: # update synchronous standby list in dcs temporarily to point to common nodes in current and picked sync_common = current & allow_promote diff --git a/tests/test_ha.py b/tests/test_ha.py index 9b286f4f..ea8fa7c0 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -1478,6 +1478,24 @@ class TestHa(PostgresInit): self.ha.run_cycle() self.assertEqual(mock_logger.call_args[0][0], 'Updating sync state failed') + @patch.object(Cluster, 'is_unlocked', Mock(return_value=False)) + def test_inconsistent_synchronous_state(self): + self.ha.is_synchronous_mode = true + self.ha.has_lock = true + self.p.name = 'leader' + self.ha.cluster = get_cluster_initialized_without_leader(sync=('leader', 'a')) + self.p.sync_handler.current_state = Mock(return_value=(CaseInsensitiveSet('a'), CaseInsensitiveSet())) + self.ha.dcs.write_sync_state = Mock(return_value=SyncState.empty()) + mock_set_sync = self.p.sync_handler.set_synchronous_standby_names = Mock() + with patch('patroni.ha.logger.warning') as mock_logger: + self.ha.run_cycle() + mock_set_sync.assert_called_once() + self.assertTrue(mock_logger.call_args_list[0][0][0].startswith('Inconsistent state between ')) + self.ha.dcs.write_sync_state = Mock(return_value=None) + with patch('patroni.ha.logger.warning') as mock_logger: + self.ha.run_cycle() + self.assertEqual(mock_logger.call_args[0][0], 'Updating sync state failed') + def test_effective_tags(self): self.ha._disable_sync = True self.assertEqual(self.ha.get_effective_tags(), {'foo': 'bar', 'nosync': True})