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/__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/patroni/config.py b/patroni/config.py index 82f5f1f6..590e765e 100644 --- a/patroni/config.py +++ b/patroni/config.py @@ -495,7 +495,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/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/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/patroni/ha.py b/patroni/ha.py index cb45725a..c2fc915a 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -789,6 +789,14 @@ class Ha(object): allow_promote = current_state.sync voters = CaseInsensitiveSet(sync.voters) + if picked == voters and voters != allow_promote: + logger.warning('Inconsistent state between synchronous_standby_names = %s and /sync = %s key ' + 'detected, updating synchronous replication key...', list(allow_promote), list(voters)) + sync = self.dcs.write_sync_state(self.state_handler.name, allow_promote, 0, version=sync.version) + if not sync: + return logger.warning("Updating sync state failed") + voters = CaseInsensitiveSet(sync.voters) + if picked == voters: return diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py index e7ea3acb..c5e970c1 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_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) diff --git a/tests/test_ha.py b/tests/test_ha.py index 2a35e9f4..69b715f7 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -1486,6 +1486,25 @@ 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=_SyncState('priority', 0, 0, + CaseInsensitiveSet(), CaseInsensitiveSet('a'))) + 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}) 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) 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() 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