From 6a75b1591b3b115de6949cb645851eba0d7d30c2 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 15 Aug 2023 09:01:37 +0200 Subject: [PATCH] Use pg_current_wal_flush_lsn() starting from 9.6 (#2813) Due to historical reasons (not available before 9.6) we used `pg_current_wal_lsn()`/`pg_current_xlog_location()` functions to get current WAL LSN on the primary. But, this LSN is not necessarily synced to disk, and could be lost if the primary node crashed. --- patroni/api.py | 8 ++++---- patroni/postgresql/__init__.py | 11 ++++++++--- patroni/postgresql/sync.py | 2 +- tests/test_api.py | 1 + tests/test_postgresql.py | 7 +++++++ 5 files changed, 21 insertions(+), 8 deletions(-) diff --git a/patroni/api.py b/patroni/api.py index 8d991d5f..3397d8c3 100644 --- a/patroni/api.py +++ b/patroni/api.py @@ -448,7 +448,7 @@ class RestApiHandler(BaseHTTPRequestHandler): * ``patroni_postmaster_start_time``: epoch timestamp since Postmaster was started; * ``patroni_master``: ``1`` if this node holds the leader lock, else ``0``; * ``patroni_primary``: same as ``patroni_master``; - * ``patroni_xlog_location``: ``pg_wal_lsn_diff(pg_current_wal_lsn(), '0/0')`` if leader, else ``0``; + * ``patroni_xlog_location``: ``pg_wal_lsn_diff(pg_current_wal_flush_lsn(), '0/0')`` if leader, else ``0``; * ``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``; @@ -1141,7 +1141,7 @@ class RestApiHandler(BaseHTTPRequestHandler): * ``server_version``: Postgres version without periods, e.g. ``150002`` for Postgres ``15.2``; * ``xlog``: dictionary. Its structure depends on ``role``: * If ``master``: - * ``location``: ``pg_current_wal_lsn()`` + * ``location``: ``pg_current_wal_flush_lsn()`` * If ``replica``: * ``received_location``: ``pg_wal_lsn_diff(pg_last_wal_receive_lsn(), '0/0')``; * ``replayed_location``: ``pg_wal_lsn_diff(pg_last_wal_replay_lsn(), '0/0)``; @@ -1179,8 +1179,8 @@ class RestApiHandler(BaseHTTPRequestHandler): " application_name, client_addr, w.state, sync_state, sync_priority" " FROM pg_catalog.pg_stat_get_wal_senders() w, pg_catalog.pg_stat_get_activity(pid)) AS ri") - row = self.query(stmt.format(postgresql.wal_name, postgresql.lsn_name), retry=retry)[0] - + row = self.query(stmt.format(postgresql.wal_name, postgresql.lsn_name, + postgresql.wal_flush), retry=retry)[0] result = { 'state': postgresql.state, 'postmaster_start_time': row[0], diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py index 172e575e..943a5832 100644 --- a/patroni/postgresql/__init__.py +++ b/patroni/postgresql/__init__.py @@ -57,8 +57,8 @@ class Postgresql(object): TL_LSN = ("CASE WHEN pg_catalog.pg_is_in_recovery() THEN 0 " "ELSE ('x' || pg_catalog.substr(pg_catalog.pg_{0}file_name(" "pg_catalog.pg_current_{0}_{1}()), 1, 8))::bit(32)::int END, " # primary timeline - "CASE WHEN pg_catalog.pg_is_in_recovery() THEN 0 " - "ELSE pg_catalog.pg_{0}_{1}_diff(pg_catalog.pg_current_{0}_{1}(), '0/0')::bigint END, " # write_lsn + "CASE WHEN pg_catalog.pg_is_in_recovery() THEN 0 ELSE " + "pg_catalog.pg_{0}_{1}_diff(pg_catalog.pg_current_{0}{2}_{1}(), '0/0')::bigint END, " # wal(_flush)?_lsn "pg_catalog.pg_{0}_{1}_diff(pg_catalog.pg_last_{0}_replay_{1}(), '0/0')::bigint, " "pg_catalog.pg_{0}_{1}_diff(COALESCE(pg_catalog.pg_last_{0}_receive_{1}(), '0/0'), '0/0')::bigint, " "pg_catalog.pg_is_in_recovery() AND pg_catalog.pg_is_{0}_replay_paused()") @@ -159,6 +159,11 @@ class Postgresql(object): def wal_name(self) -> str: return 'wal' if self._major_version >= 100000 else 'xlog' + @property + def wal_flush(self) -> str: + """For PostgreSQL 9.6 onwards we want to use pg_current_wal_flush_lsn()/pg_current_xlog_flush_location().""" + return '_flush' if self._major_version >= 90600 else '' + @property def lsn_name(self) -> str: return 'lsn' if self._major_version >= 100000 else 'location' @@ -211,7 +216,7 @@ class Postgresql(object): else: extra = "0, NULL, NULL, NULL, NULL, NULL, NULL" + extra - return ("SELECT " + self.TL_LSN + ", {2}").format(self.wal_name, self.lsn_name, extra) + return ("SELECT " + self.TL_LSN + ", {3}").format(self.wal_name, self.lsn_name, self.wal_flush, extra) @property def available_gucs(self) -> CaseInsensitiveSet: diff --git a/patroni/postgresql/sync.py b/patroni/postgresql/sync.py index 3193a7e7..97a6fd46 100644 --- a/patroni/postgresql/sync.py +++ b/patroni/postgresql/sync.py @@ -182,7 +182,7 @@ class _ReplicaList(List[_Replica]): swapping, but only if lag on this member is exceeding a threshold (``maximum_lag_on_syncnode``). :ivar max_lsn: maximum value of ``_Replica.lsn`` among all values. In case if there is just one - element in the list we take value of ``pg_current_wal_lsn()``. + element in the list we take value of ``pg_current_wal_flush_lsn()``. """ def __init__(self, postgresql: 'Postgresql', cluster: Cluster) -> None: diff --git a/tests/test_api.py b/tests/test_api.py index eb19ebca..b5aeff8a 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -36,6 +36,7 @@ class MockPostgresql(object): pending_restart = True wal_name = 'wal' lsn_name = 'lsn' + wal_flush = '_flush' POSTMASTER_START_TIME = 'pg_catalog.pg_postmaster_start_time()' TL_LSN = 'CASE WHEN pg_catalog.pg_is_in_recovery()' citus_handler = Mock() diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index a703f235..4f1e36da 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -955,3 +955,10 @@ class TestPostgresql2(BaseTestPostgresql): gucs = self.p.available_gucs self.assertIsInstance(gucs, CaseInsensitiveSet) self.assertEqual(gucs, mock_available_gucs.return_value) + + def test_cluster_info_query(self): + self.assertIn('diff(pg_catalog.pg_current_wal_flush_lsn(', self.p.cluster_info_query) + self.p._major_version = 90600 + self.assertIn('diff(pg_catalog.pg_current_xlog_flush_location(', self.p.cluster_info_query) + self.p._major_version = 90500 + self.assertIn('diff(pg_catalog.pg_current_xlog_location(', self.p.cluster_info_query)