From c12fe4146d284560f25e0945995954021604332b Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 10 Jan 2023 10:44:17 +0100 Subject: [PATCH] Run only one query per HA loop (#2516) If the cluster is stable (no nodes are joining/leaving/lagging) we want to run at most one monitor query per every HA loop. So far it worker perfectly except when synchronous_mode is enabled, where we run two additional queries: 1. SHOW synchronous_mode 2. SELECT ... FROM pg_stat_replication In order to solve it, we will include these "queries" to the common monitoring query is synchronous_mode is enabled. In addition to that make sure that `synchronous_standby_names` is reset on replicas that used to be a primary and avoid using replicas which are not in the 'running' state. P.S.: in the monitoring query we also extract the current value of synchronous_standby_names, because it will be useful for the quorum commit feature. Close https://github.com/zalando/patroni/issues/2469 --- patroni/dcs/zookeeper.py | 3 ++ patroni/ha.py | 8 +++- patroni/postgresql/__init__.py | 74 +++++++++++++++++++++++----------- tests/__init__.py | 8 ++-- tests/test_ha.py | 3 +- tests/test_postgresql.py | 44 ++++++++++---------- 6 files changed, 88 insertions(+), 52 deletions(-) diff --git a/patroni/dcs/zookeeper.py b/patroni/dcs/zookeeper.py index c4f686a6..2125120f 100644 --- a/patroni/dcs/zookeeper.py +++ b/patroni/dcs/zookeeper.py @@ -371,8 +371,11 @@ 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 (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: diff --git a/patroni/ha.py b/patroni/ha.py index 6c4e3e97..53801107 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -386,6 +386,9 @@ class Ha(object): else: msg = "starting as a secondary" node_to_follow = self._get_node_to_follow(self.cluster) + + if self.is_synchronous_mode(): + self.state_handler.config.set_synchronous_standby([]) elif self.has_lock(): msg = "starting as readonly because i had the session lock" node_to_follow = None @@ -913,14 +916,15 @@ class Ha(object): except Exception: node_to_follow, leader = None, None + if self.is_synchronous_mode(): + self.state_handler.config.set_synchronous_standby([]) + # FIXME: with mode offline called from DCS exception handler and handle_long_action_in_progress # there could be an async action already running, calling follow from here will lead # to racy state handler state updates. if mode_control['async_req']: self._async_executor.try_run_async('starting after demotion', self.state_handler.follow, (node_to_follow,)) else: - if self.is_synchronous_mode(): - self.state_handler.config.set_synchronous_standby([]) if self._rewind.rewind_or_reinitialize_needed_and_possible(leader): return False # do not start postgres, but run pg_rewind on the next iteration self.state_handler.follow(node_to_follow) diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py index 3e2e1d38..f3eae728 100644 --- a/patroni/postgresql/__init__.py +++ b/patroni/postgresql/__init__.py @@ -22,6 +22,7 @@ from .connection import Connection, get_connection_cursor from .misc import parse_history, parse_lsn, postgres_major_version_to_int from .postmaster import PostmasterProcess from .slots import SlotsHandler +from .validator import CaseInsensitiveDict from .. import psycopg from ..exceptions import PostgresConnectionException from ..utils import Retry, RetryFailedError, polling_loop, data_directory_is_empty, parse_int @@ -106,6 +107,7 @@ class Postgresql(object): self._cluster_info_state = {} self._has_permanent_logical_slots = True self._enforce_hot_standby_feedback = False + self._is_synchronous_mode = True self._cached_replica_timeline = None # Last known running process @@ -158,11 +160,36 @@ class Postgresql(object): @property def cluster_info_query(self): + """Returns the monitoring query with a fixed number of fields. + + The query text is constructed based on current state in DCS and PostgreSQL version: + 1. function names depend on version. wal/lsn for v10+ and xlog/location for pre v10. + 2. for primary we query timeline_id (extracted from pg_walfile_name()) and pg_current_wal_lsn() + 3. for replicas we query pg_last_wal_receive_lsn(), pg_last_wal_replay_lsn(), and pg_is_wal_replay_paused() + 4. for v9.6+ we query primary_slot_name and primary_conninfo from pg_stat_get_wal_receiver() + 5. for v11+ with permanent logical slots we query from pg_replication_slots and aggregate the result + 6. for standby_leader node running v9.6+ we also query pg_control_checkpoint to fetch timeline_id + 7. if sync replication is enabled we query pg_stat_replication and aggregate the result. + In addition to that we get current values of synchronous_commit and synchronous_standby_names GUCs. + + If some conditions are not satisfied we simply put static values instead. E.g., NULL, 0, '', and so on.""" + + extra = ", " + (("pg_catalog.current_setting('synchronous_commit'), " + + "pg_catalog.current_setting('synchronous_standby_names'), " + "(SELECT pg_catalog.json_agg(r.*) FROM (SELECT application_name, sync_state," + + " pg_catalog.pg_{0}_{1}_diff(write_{1}, '0/0')::bigint AS write_lsn," + + " pg_catalog.pg_{0}_{1}_diff(flush_{1}, '0/0')::bigint AS flush_lsn," + + " pg_catalog.pg_{0}_{1}_diff(replay_{1}, '0/0')::bigint AS replay_lsn " + + "FROM pg_catalog.pg_stat_get_wal_senders() w," + + " pg_catalog.pg_stat_get_activity(pid)" + + " WHERE w.state = 'streaming') r)").format(self.wal_name, self.lsn_name) + if self._is_synchronous_mode and self.role == 'master' else "'on', '', NULL") + if self._major_version >= 90600: - extra = "(SELECT pg_catalog.json_agg(s.*) FROM (SELECT slot_name, slot_type as type, datoid::bigint, " +\ - "plugin, catalog_xmin, pg_catalog.pg_wal_lsn_diff(confirmed_flush_lsn, '0/0')::bigint" + \ - " AS confirmed_flush_lsn FROM pg_catalog.pg_get_replication_slots()) AS s)"\ - if self._has_permanent_logical_slots and self._major_version >= 110000 else "NULL" + extra = ("(SELECT pg_catalog.json_agg(s.*) FROM (SELECT slot_name, slot_type as type, datoid::bigint, " + + "plugin, catalog_xmin, pg_catalog.pg_wal_lsn_diff(confirmed_flush_lsn, '0/0')::bigint" + + " AS confirmed_flush_lsn FROM pg_catalog.pg_get_replication_slots()) AS s)" + if self._has_permanent_logical_slots and self._major_version >= 110000 else "NULL") + extra extra = (", CASE WHEN latest_end_lsn IS NULL THEN NULL ELSE received_tli END," " slot_name, conninfo, {0} FROM pg_catalog.pg_stat_get_wal_receiver()").format(extra) if self.role == 'standby_leader': @@ -170,7 +197,7 @@ class Postgresql(object): else: extra = "0" + extra else: - extra = "0, NULL, NULL, NULL, NULL" + extra = "0, NULL, NULL, NULL, NULL" + extra return ("SELECT " + self.TL_LSN + ", {2}").format(self.wal_name, self.lsn_name, extra) @@ -334,13 +361,16 @@ class Postgresql(object): self._has_permanent_logical_slots or cluster.should_enforce_hot_standby_feedback(self.name, nofailover, self.major_version)) + self._is_synchronous_mode = cluster.is_synchronous_mode() + def _cluster_info_state_get(self, name): if not self._cluster_info_state: try: result = self._is_leader_retry(self._query, self.cluster_info_query).fetchone() cluster_info_state = dict(zip(['timeline', 'wal_position', 'replayed_location', 'received_location', 'replay_paused', 'pg_control_timeline', - 'received_tli', 'slot_name', 'conninfo', 'slots'], result)) + 'received_tli', 'slot_name', 'conninfo', 'slots', 'synchronous_commit', + 'synchronous_standby_names', 'pg_stat_replication'], result)) if self._has_permanent_logical_slots: cluster_info_state['slots'] =\ self.slots_handler.process_permanent_slots(cluster_info_state['slots']) @@ -1076,9 +1106,6 @@ class Postgresql(object): logger.exception('Could not remove data directory %s', self._data_dir) self.move_data_directory() - def _get_synchronous_commit_param(self): - return self.query("SHOW synchronous_commit").fetchone()[0] - def pick_synchronous_standby(self, cluster, sync_node_count=1, sync_node_maxlag=-1): """Finds the best candidate to be the synchronous standby. @@ -1093,29 +1120,28 @@ class Postgresql(object): """ if self._major_version < 90600: sync_node_count = 1 - members = {m.name.lower(): m for m in cluster.members} + members = CaseInsensitiveDict({m.name: m for m in cluster.members}) candidates = [] sync_nodes = [] replica_list = [] # Pick candidates based on who has higher replay/remote_write/flush lsn. - sync_commit_par = self._get_synchronous_commit_param() - sort_col = {'remote_apply': 'replay', 'remote_write': 'write'}.get(sync_commit_par, 'flush') + synchronous_commit = self._cluster_info_state_get('synchronous_commit') + sort_col = {'remote_apply': 'replay', 'remote_write': 'write'}.get(synchronous_commit, 'flush') + '_lsn' + pg_stat_replication = [(r['application_name'], r['sync_state'], r[sort_col]) + for r in self._cluster_info_state_get('pg_stat_replication') or [] + if r[sort_col] is not None] # pg_stat_replication.sync_state has 4 possible states - async, potential, quorum, sync. - # Sort clause "ORDER BY sync_state DESC" is to get the result in required order and to keep - # the result consistent in case if a synchronous standby member is slowed down OR async node - # receiving changes faster than the sync member (very rare but possible). Such cases would - # trigger sync standby member swapping frequently and the sort on sync_state desc should - # help in keeping the query result consistent. - for app_name, sync_state, replica_lsn in self.query( - "SELECT pg_catalog.lower(application_name), sync_state, pg_{2}_{1}_diff({0}_{1}, '0/0')::bigint" - " FROM pg_catalog.pg_stat_replication" - " WHERE state = 'streaming' AND {0}_{1} IS NOT NULL" - " ORDER BY sync_state DESC, {0}_{1} DESC".format(sort_col, self.lsn_name, self.wal_name)): + # That is, alphabetically they are in the reversed order of priority. + # Since we are doing reversed sort on (sync_state, lsn) tuples, it helps to keep the result + # consistent in case if a synchronous standby member is slowed down OR async node receiving + # changes faster than the sync member (very rare but possible). + # Such cases would trigger sync standby member swapping, but only if lag on a sync node exceeding a threshold. + for app_name, sync_state, replica_lsn in sorted(pg_stat_replication, key=lambda r: (r[1], r[2]), reverse=True): member = members.get(app_name) - if member and not member.tags.get('nosync', False): + if member and member.is_running and not member.tags.get('nosync', False): replica_list.append((member.name, sync_state, replica_lsn, bool(member.nofailover))) - max_lsn = max(replica_list, key=lambda x: x[2])[2] if len(replica_list) > 1 else int(str(self.last_operation())) + max_lsn = max(replica_list, key=lambda x: x[2])[2] if len(replica_list) > 1 else self.last_operation() # Prefer members without nofailover tag. We are relying on the fact that sorts are guaranteed to be stable. for app_name, sync_state, replica_lsn, _ in sorted(replica_list, key=lambda x: x[3]): diff --git a/tests/__init__.py b/tests/__init__.py index 0c634828..a0fc8958 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -211,11 +211,13 @@ class BaseTestPostgresql(PostgresInit): if not os.path.exists(self.p.data_dir): os.makedirs(self.p.data_dir) - self.leadermem = Member(0, 'leader', 28, {'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5435/postgres'}) + self.leadermem = Member(0, 'leader', 28, { + 'state': 'running', 'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5435/postgres'}) self.leader = Leader(-1, 28, self.leadermem) self.other = Member(0, 'test-1', 28, {'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5433/postgres', - 'tags': {'replicatefrom': 'leader'}}) - self.me = Member(0, 'test0', 28, {'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5434/postgres'}) + 'state': 'running', 'tags': {'replicatefrom': 'leader'}}) + self.me = Member(0, 'test0', 28, { + 'state': 'running', 'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5434/postgres'}) def tearDown(self): if os.path.exists(self.p.data_dir): diff --git a/tests/test_ha.py b/tests/test_ha.py index 4ec0d7f7..92551982 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -311,7 +311,8 @@ class TestHa(PostgresInit): self.ha._rewind.check_leader_is_not_in_recovery = true with patch.object(Rewind, 'rewind_or_reinitialize_needed_and_possible', Mock(return_value=True)): self.assertEqual(self.ha.run_cycle(), 'running pg_rewind from leader') - with patch.object(Rewind, 'rewind_or_reinitialize_needed_and_possible', Mock(return_value=False)): + with patch.object(Rewind, 'rewind_or_reinitialize_needed_and_possible', Mock(return_value=False)),\ + patch.object(Ha, 'is_synchronous_mode', Mock(return_value=True)): self.p.follow = true self.assertEqual(self.ha.run_cycle(), 'starting as a secondary') self.p.is_running = true diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index df24ba55..d5524bad 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -638,44 +638,44 @@ class TestPostgresql(BaseTestPostgresql): self.p._state = 'starting' self.assertIsNone(self.p.wait_for_startup()) + @patch.object(Postgresql, 'last_operation', Mock(return_value=2)) def test_pick_sync_standby(self): cluster = Cluster(True, None, self.leader, 0, [self.me, self.other, self.leadermem], None, SyncState(0, self.me.name, self.leadermem.name), None, None, None) - mock_cursor = Mock() - mock_cursor.fetchone.return_value = ('remote_apply',) - with patch.object(Postgresql, "query", side_effect=[ - mock_cursor, - [(self.leadermem.name, 'sync', 1), - (self.me.name, 'async', 2), - (self.other.name, 'async', 2)] + with patch.object(Postgresql, "_cluster_info_state_get", side_effect=[ + 'on', + [{'application_name': self.leadermem.name, 'sync_state': 'sync', 'flush_lsn': 1}, + {'application_name': self.me.name, 'sync_state': 'async', 'flush_lsn': 2}, + {'application_name': self.other.name, 'sync_state': 'async', 'flush_lsn': 2}] ]): + self.assertEqual(self.p.pick_synchronous_standby(cluster), ([self.leadermem.name], [self.leadermem.name])) - with patch.object(Postgresql, "query", side_effect=[ - mock_cursor, - [(self.leadermem.name, 'potential', 1), - (self.me.name, 'async', 2), - (self.other.name, 'async', 2)] + with patch.object(Postgresql, "_cluster_info_state_get", side_effect=[ + 'remote_write', + [{'application_name': self.leadermem.name, 'sync_state': 'potential', 'write_lsn': 1}, + {'application_name': self.me.name, 'sync_state': 'async', 'write_lsn': 2}, + {'application_name': self.other.name, 'sync_state': 'async', 'write_lsn': 2}] ]): self.assertEqual(self.p.pick_synchronous_standby(cluster), ([self.leadermem.name], [])) - with patch.object(Postgresql, "query", side_effect=[ - mock_cursor, - [(self.me.name, 'async', 1), - (self.other.name, 'async', 2)] + with patch.object(Postgresql, "_cluster_info_state_get", side_effect=[ + 'remote_apply', + [{'application_name': self.me.name.upper(), 'sync_state': 'async', 'replay_lsn': 2}, + {'application_name': self.other.name, 'sync_state': 'async', 'replay_lsn': 1}] ]): self.assertEqual(self.p.pick_synchronous_standby(cluster), ([self.me.name], [])) - with patch.object(Postgresql, "query", side_effect=[ - mock_cursor, - [('missing', 'sync', 1), - (self.me.name, 'async', 2), - (self.other.name, 'async', 3)] + with patch.object(Postgresql, "_cluster_info_state_get", side_effect=[ + 'remote_apply', + [{'application_name': 'missing', 'sync_state': 'sync', 'replay_lsn': 3}, + {'application_name': self.me.name, 'sync_state': 'async', 'replay_lsn': 2}, + {'application_name': self.other.name, 'sync_state': 'async', 'replay_lsn': 1}] ]): self.assertEqual(self.p.pick_synchronous_standby(cluster), ([self.me.name], [])) - with patch.object(Postgresql, "query", side_effect=[mock_cursor, []]): + with patch.object(Postgresql, "_cluster_info_state_get", side_effect=['remote_apply', []]): self.p._major_version = 90400 self.assertEqual(self.p.pick_synchronous_standby(cluster), ([], []))