mirror of
https://github.com/outbackdingo/patroni.git
synced 2026-08-25 14:53:37 +00:00
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
This commit is contained in:
Alexander Kukushkin
@@ -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:
|
||||
|
||||
+6
-2
@@ -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)
|
||||
|
||||
@@ -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]):
|
||||
|
||||
+5
-3
@@ -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:[email protected]:5435/postgres'})
|
||||
self.leadermem = Member(0, 'leader', 28, {
|
||||
'state': 'running', 'conn_url': 'postgres://replicator:[email protected]:5435/postgres'})
|
||||
self.leader = Leader(-1, 28, self.leadermem)
|
||||
self.other = Member(0, 'test-1', 28, {'conn_url': 'postgres://replicator:[email protected]:5433/postgres',
|
||||
'tags': {'replicatefrom': 'leader'}})
|
||||
self.me = Member(0, 'test0', 28, {'conn_url': 'postgres://replicator:[email protected]:5434/postgres'})
|
||||
'state': 'running', 'tags': {'replicatefrom': 'leader'}})
|
||||
self.me = Member(0, 'test0', 28, {
|
||||
'state': 'running', 'conn_url': 'postgres://replicator:[email protected]:5434/postgres'})
|
||||
|
||||
def tearDown(self):
|
||||
if os.path.exists(self.p.data_dir):
|
||||
|
||||
+2
-1
@@ -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
|
||||
|
||||
+22
-22
@@ -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), ([], []))
|
||||
|
||||
|
||||
Reference in New Issue
Block a user