From 893e460695faf8fed5154d852ad8c4a5d7336d67 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Thu, 20 Jul 2023 08:00:52 +0200 Subject: [PATCH] Address review feedback --- docs/rest_api.rst | 2 ++ features/steps/citus.py | 3 +-- features/steps/quorum_commit.py | 27 ++++++++++----------- patroni/api.py | 12 ++++++++- patroni/ha.py | 6 ++--- patroni/postgresql/sync.py | 43 +++++++++++++++++++++++++++------ tests/test_api.py | 2 ++ 7 files changed, 68 insertions(+), 27 deletions(-) diff --git a/docs/rest_api.rst b/docs/rest_api.rst index d7a6824d..5310f73e 100644 --- a/docs/rest_api.rst +++ b/docs/rest_api.rst @@ -47,6 +47,8 @@ For all health check ``GET`` requests Patroni returns a JSON document with the s - ``GET /quorum``: returns HTTP status code **200** only when this Patroni node is listed as a quorum node in ``synchronous_standby_names`` on the primary. +- ``GET /read-only-quorum``: like the above endpoint, but also includes the primary. + - ``GET /asynchronous`` or ``GET /async``: returns HTTP status code **200** only when the Patroni node is running as an asynchronous standby. diff --git a/features/steps/citus.py b/features/steps/citus.py index 84186e8f..1af70a30 100644 --- a/features/steps/citus.py +++ b/features/steps/citus.py @@ -61,8 +61,7 @@ def check_registration(context, name1, name2, role, group, time_limit): except Exception: pass time.sleep(1) - assert False, "Worker {0} is not registered in pg_dist_node on the coordinator {1} after {2} seconds"\ - .format(name1, name2, time_limit) + assert False, "Node {0} is not registered in pg_dist_node on the node {1}".format(name1, name2) @step('I create a distributed table on {name:w}') diff --git a/features/steps/quorum_commit.py b/features/steps/quorum_commit.py index 7cedc0d5..19a887a7 100644 --- a/features/steps/quorum_commit.py +++ b/features/steps/quorum_commit.py @@ -25,6 +25,17 @@ def check_sync(context, key, value, time_limit): dcs_value, time_limit) +def _parse_synchronous_standby_names(value): + if '(' in value: + m = re.match(r'.*(\d+) \(([^)]+)\)', value) + expected_value = set(m.group(2).split()) + expected_num = m.group(1) + else: + expected_value = set([value]) + expected_num = '1' + return expected_num, expected_value + + @then('synchronous_standby_names on {name:2} is set to "{value}" after {time_limit:d} seconds') def check_synchronous_standby_names(context, name, value, time_limit): time_limit *= context.timeout_multiplier @@ -33,24 +44,12 @@ def check_synchronous_standby_names(context, name, value, time_limit): if value == '_empty_str_': value = '' - if '(' in value: - m = re.match(r'.*(\d+) \(([^)]+)\)', value) - expected_value = set(m.group(2).split()) - expected_num = m.group(1) - else: - expected_value = set([value]) - expected_num = '1' + expected_num, expected_value = _parse_synchronous_standby_names(value) while time.time() < max_time: try: ssn = context.pctl.query(name, "SHOW synchronous_standby_names").fetchone()[0] - if '(' in ssn: - m = re.match(r'.*(\d+) \(([^)]+)\)', ssn) - db_value = set(m.group(2).split()) - db_num = m.group(1) - else: - db_value = set([ssn]) - db_num = '1' + db_num, db_value = _parse_synchronous_standby_names(ssn) if expected_value == db_value and expected_num == db_num: return except Exception: diff --git a/patroni/api.py b/patroni/api.py index fe03b7f4..5da815d5 100644 --- a/patroni/api.py +++ b/patroni/api.py @@ -219,6 +219,10 @@ class RestApiHandler(BaseHTTPRequestHandler): * HTTP status ``200``: if up and running as a standby and without ``noloadbalance`` tag. * ``/read-only``: * HTTP status ``200``: if up and running and without ``noloadbalance`` tag. + * ``/quorum``: + * HTTP status ``200``: if up and running as a quorum synchronous standby. + * ``/read-only-quorum``: + * HTTP status ``200``: if up and running as a quorum synchronous standby or primary. * ``/synchronous`` or ``/sync``: * HTTP status ``200``: if up and running as a synchronous standby. * ``/read-only-sync``: @@ -290,7 +294,7 @@ class RestApiHandler(BaseHTTPRequestHandler): ignore_tags = True elif 'replica' in path: status_code = replica_status_code - elif 'read-only' in path and 'sync' not in path: + elif 'read-only' in path and 'sync' not in path and 'quorum' not in path: status_code = 200 if 200 in (primary_status_code, standby_leader_status_code) else replica_status_code elif 'health' in path: status_code = 200 if response.get('state') == 'running' else 503 @@ -303,6 +307,11 @@ class RestApiHandler(BaseHTTPRequestHandler): status_code = replica_status_code elif path in ('/async', '/asynchronous') and not is_synchronous and not is_quorum: status_code = replica_status_code + elif path == '/read-only-quorum': + if 200 in (primary_status_code, standby_leader_status_code): + status_code = 200 + elif is_quorum: + status_code = replica_status_code elif path in ('/read-only-sync', '/read-only-synchronous'): if 200 in (primary_status_code, standby_leader_status_code): status_code = 200 @@ -455,6 +464,7 @@ class RestApiHandler(BaseHTTPRequestHandler): * ``patroni_standby_leader``: ``1`` if standby leader node, else ``0``; * ``patroni_replica``: ``1`` if a replica, else ``0``; * ``patroni_sync_standby``: ``1`` if a sync replica, else ``0``; + * ``patroni_quorum_standby``: ``1`` if a quorum sync replica, else ``0``; * ``patroni_xlog_received_location``: ``pg_wal_lsn_diff(pg_last_wal_receive_lsn(), '0/0')``; * ``patroni_xlog_replayed_location``: ``pg_wal_lsn_diff(pg_last_wal_replay_lsn(), '0/0)``; * ``patroni_xlog_replayed_timestamp``: ``pg_last_xact_replay_timestamp``; diff --git a/patroni/ha.py b/patroni/ha.py index 9975e2dd..d7afc086 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -922,8 +922,8 @@ class Ha(object): def fetch_node_status(self, member: Member) -> _MemberStatus: """Perform http get request on member.api_url to fetch its status. - Usually this happens during the leader race and we can't afford to wait an indefinite time - for a response, therefore the request timeout is hardcoded to 2 seconds, which seems to be a + Usually this happens during the leader race and we can't afford to wait an indefinite time + for a response, therefore the request timeout is hardcoded to 2 seconds, which seems to be a good compromise. The node which is slow to respond is most likely unhealthy. :returns: :class:`_MemberStatus` object @@ -1002,7 +1002,7 @@ class Ha(object): :param check_replication_lag: whether to take the replication lag into account. If the lag exceeds configured threshold the node disqualifies itself. :returns: `True` if the node is eligible to become the new leader. Since this method is executed - on multiple nodes independently it is possible that multiple nodes could count + on multiple nodes independently it is possible that multiple nodes could count themselves as the healthiest because they received/replayed up to the same LSN, but this is totally fine. """ diff --git a/patroni/postgresql/sync.py b/patroni/postgresql/sync.py index 921b4765..1b9fda10 100644 --- a/patroni/postgresql/sync.py +++ b/patroni/postgresql/sync.py @@ -154,6 +154,18 @@ def parse_sync_standby_names(value: str) -> _SSN: class _SyncState(NamedTuple): + """Class representing the current synchronous state. + + :ivar sync_type: possible values: 'off', 'priority', 'quorum' + :ivar numsync: how many nodes are required to be synchronous (according to ``synchronous_standby_names``). + Is ``0`` in case if synchronous_standby_names value is invalid or has ``*``. + :ivar numsync_confirmed: how many nodes are known to be synchronous according to the ``pg_stat_replication`` + view. Only nodes that caught up with the ``SyncHandler._primary_flush_lsn` are counted. + :ivar sync: collection of synchronous node names. In case of quorum commit all nodes listed + in ``synchronous_standby_names`` or nodes that are confirmed to be synchronous according + to the `pg_stat_replication` view. + :ivar active: collection of node names that are streaming and have no restrictions to become synchronous. + """ sync_type: str numsync: int numsync_confirmed: int @@ -210,7 +222,22 @@ END;$$""") self._postgresql.reset_cluster_info_state(None) # Reset internal cache to query fresh values def _get_replica_list(self, cluster: Cluster) -> Iterator[Tuple[int, str, str, int, bool]]: - """Yields candidates based on higher replay/remote_write/flush lsn.""" + """Yields candidates based on higher replay/remote_write/flush lsn. + + .. note:: + Tuples are reverse ordered by sync_state and LSN fields so nodes that already synchronous or having + higher LSN values are preferred. + + :param cluster: current cluster topology from DCS. + + :yields: tuples composed of: + + * pid - a PID of walsender process + * member name - matches with the application_name + * sync_state - one of ("async", "potential", "quorum", "sync") + * LSN - write_lsn, flush_lsn, or replica_lsn, depending on the value of ``synchronous_commit`` GUC + * nofailover - whether the member has ``nofailover`` tag set + """ # What column from pg_stat_replication we want to sort on? Choose based on ``synchronous_commit`` value. sort_col = { @@ -259,14 +286,16 @@ END;$$""") synchronous standby any longer. Standbys are selected based on values from the global configuration: - - `maximum_lag_on_syncnode`: would help swapping unhealthy sync replica in case if it stops - responding (or hung). Please set the value high enough so it won't unncessarily swap sync - standbys during high loads. Any value less or equal of 0 keeps the behavior backward compatible. - Please note that it will not also swap sync standbys in case where all replicas are hung. - - `synchronous_node_count`: controlls how many nodes should be set as synchronous. + + - `maximum_lag_on_syncnode`: would help swapping unhealthy sync replica in case if it stops + responding (or hung). Please set the value high enough so it won't unncessarily swap sync + standbys during high loads. Any value less or equal of 0 keeps the behavior backward compatible. + Please note that it will not also swap sync standbys in case where all replicas are hung. + + - `synchronous_node_count`: controlls how many nodes should be set as synchronous. :param cluster: current cluster topology from DCS - + :returns: current synchronous replication state as a :class:`_SyncState` object """ self._handle_synchronous_standby_names_change() diff --git a/tests/test_api.py b/tests/test_api.py index 075e2029..ba62dd53 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -212,11 +212,13 @@ class TestRestApiHandler(unittest.TestCase): with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={'role': 'replica', 'quorum_standby': True})): MockRestApiServer(RestApiHandler, 'GET /quorum') + MockRestApiServer(RestApiHandler, 'GET /read-only-quorum') with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={'role': 'replica'})): MockRestApiServer(RestApiHandler, 'GET /asynchronous') with patch.object(MockHa, 'is_leader', Mock(return_value=True)): MockRestApiServer(RestApiHandler, 'GET /replica') MockRestApiServer(RestApiHandler, 'GET /read-only-sync') + MockRestApiServer(RestApiHandler, 'GET /read-only-quorum') with patch.object(GlobalConfig, 'is_standby_cluster', Mock(return_value=True)): MockRestApiServer(RestApiHandler, 'GET /standby_leader') MockPatroni.dcs.cluster = None