Show primary position LSN and lag on standby leaders

Author: avandras

patroni/ctl.py

@@ -30,6 +30,8 @@
 from collections import defaultdict
 from contextlib import contextmanager
 from enum import Enum
+from functools import reduce
+from operator import ior
 from typing import Any, Dict, Iterator, List, Optional, Tuple, TYPE_CHECKING, Union
 from urllib.parse import urlparse
 
@@ -1517,7 +1519,7 @@ def _do_site_switchover(cluster_name: str, group: Optional[int],
 
     if scheduled is None and not force:
         next_hour = (datetime.datetime.now() + datetime.timedelta(hours=1)).strftime('%Y-%m-%dT%H:%M')
-        scheduled = click.prompt('When should the switchover take place (e.g. ' + next_hour + ' ) ',
+        scheduled = click.prompt('When should the switchover take place (e.g. ' + next_hour + ') ',
                                  type=str, default='now')
 
         scheduled_at = parse_scheduled(scheduled)
@@ -1746,6 +1748,18 @@ def get_cluster_service_info(cluster: Dict[str, Any]) -> List[str]:
             info += f" ({standby_config['host']}:{standby_config.get('port', 5432)})"
         service_info.append(info)
 
+    # latest_end_lsn is only registered on standby leaders - we just combine all the member dicts to find it
+    r = {}
+    if 'members' in cluster:
+        r = reduce(ior, cluster['members'], {})
+    if 'latest_end_lsn' in r:
+        lag_to_primary = r['lag_to_primary']
+        lag_to_primary = round(lag_to_primary / 1024 / 1024) if isinstance(lag_to_primary, int) \
+            else '' if lag_to_primary == 'unknown' else lag_to_primary
+        info = (f"The latest known LSN of the primary instance is {r['latest_end_lsn']}, "
+                f"the replication lag is {lag_to_primary} MB")
+        service_info.append(info)
+
     if cluster.get('pause'):
         service_info.append('Maintenance mode: on')
 

patroni/dcs/__init__.py

@@ -337,6 +337,10 @@ def receive_lsn(self) -> Optional[int]:
     def replay_lsn(self) -> Optional[int]:
         return parse_int(self.data.get('replay_lsn'))
 
+    @property
+    def latest_end_lsn(self) -> Optional[int]:
+        return parse_int(self.data.get('latest_end_lsn'))
+
     @property
     def multisite(self) -> Optional[Dict[str, Any]]:
         """The ``multisite`` dict of the member if multisite is on."""

patroni/ha.py

@@ -477,14 +477,16 @@ def touch_member(self) -> bool:
                     and data['state'] in [PostgresqlState.RUNNING, PostgresqlState.RESTARTING,
                                           PostgresqlState.STARTING]:
                 try:
-                    timeline, wal_position, pg_control_timeline, receive_lsn, replay_lsn =\
+                    timeline, wal_position, pg_control_timeline, receive_lsn, replay_lsn, latest_end_lsn =\
                         self.state_handler.timeline_wal_position()
                     data['xlog_location'] = self._last_wal_lsn = wal_position
                     if not timeline:  # running as a standby
                         if replay_lsn:
                             data['replay_lsn'] = replay_lsn
                         if receive_lsn:
                             data['receive_lsn'] = receive_lsn
+                        if latest_end_lsn:
+                            data['latest_end_lsn'] = latest_end_lsn
                         replication_state = self.state_handler.replication_state()
                         if replication_state:
                             data['replication_state'] = replication_state

patroni/postgresql/__init__.py

@@ -250,7 +250,8 @@ def cluster_info_query(self) -> str:
 
             written_lsn = ("pg_catalog.pg_wal_lsn_diff(written_lsn, '0/0')::bigint"
                            if self._major_version >= 130000 else "NULL")
-            extra = (", CASE WHEN latest_end_lsn IS NULL THEN NULL ELSE received_tli END, {0}, slot_name, "
+            extra = (", CASE WHEN latest_end_lsn IS NULL THEN NULL ELSE received_tli END, {0}, "
+                     "pg_catalog.pg_wal_lsn_diff(latest_end_lsn, '0/0')::bigint, slot_name, "
                      "conninfo, status, {1} FROM pg_catalog.pg_stat_get_wal_receiver()").format(written_lsn, extra)
             if self.role == PostgresqlRole.STANDBY_LEADER:
                 extra = "timeline_id" + extra + ", pg_catalog.pg_control_checkpoint()"
@@ -491,7 +492,7 @@ def _cluster_info_state_get(self, name: str) -> Optional[Any]:
                 result = self._is_leader_retry(self._query, self.cluster_info_query)[0]
                 cluster_info_state = dict(zip(['timeline', 'wal_position', 'replay_lsn',
                                                'receive_lsn', 'replay_paused', 'pg_control_timeline',
-                                               'received_tli', 'write_location', 'slot_name', 'conninfo',
+                                               'received_tli', 'write_location', 'latest_end_lsn', 'slot_name', 'conninfo',
                                                'receiver_state', 'restore_command', 'slots', 'synchronous_commit',
                                                'synchronous_standby_names', 'pg_stat_replication'], result))
                 if self._should_query_slots and self.can_advance_slots:
@@ -511,6 +512,9 @@ def _cluster_info_state_get(self, name: str) -> Optional[Any]:
     def replay_lsn(self) -> Optional[int]:
         return self._cluster_info_state_get('replay_lsn')
 
+    def latest_end_lsn(self) -> Optional[int]:
+        return self._cluster_info_state_get('latest_end_lsn')
+
     def receive_lsn(self) -> Optional[int]:
         write = self._cluster_info_state_get('write_location')
         received = self._cluster_info_state_get('receive_lsn')
@@ -1252,12 +1256,16 @@ def _wal_position(is_primary: bool, wal_position: int,
                       receive_lsn: Optional[int], replay_lsn: Optional[int]) -> int:
         return wal_position if is_primary else max(receive_lsn or 0, replay_lsn or 0)
 
-    def timeline_wal_position(self) -> Tuple[int, int, Optional[int], Optional[int], Optional[int]]:
+    def timeline_wal_position(self) -> Tuple[int, int, Optional[int], Optional[int], Optional[int], Optional[int]]:
         """Get timeline and various wal positions.
 
-        :returns: a tuple composed of 5 integers representing timeline, ``pg_current_wal_lsn()`` position
-            on primary/the biggest value among receive and replay LSN on replicas, ``pg_control_checkpoint()``
-            value on a standby leader, receive and replay LSN on replicas (if available).
+        :returns: a tuple composed of 6 integers representing
+            * timeline,
+            * ``pg_current_wal_lsn()`` position on the primary
+              or the biggest value among receive and replay LSN on replicas,
+            * ``pg_control_checkpoint()`` value on a standby leader,
+            * receive and replay LSN on replicas (if available),
+            * latest_end_lsn (the last known LSN on the primary)
         """
         # This method could be called from different threads (simultaneously with some other `_query` calls).
         # If it is called not from main thread we will create a new cursor to execute statement.
@@ -1266,14 +1274,16 @@ def timeline_wal_position(self) -> Tuple[int, int, Optional[int], Optional[int],
             wal_position = self._cluster_info_state_get('wal_position') or 0
             replay_lsn = self.replay_lsn()
             receive_lsn = self.receive_lsn()
+            latest_end_lsn = self.latest_end_lsn()
             pg_control_timeline = self._cluster_info_state_get('pg_control_timeline')
         else:
-            timeline, wal_position, replay_lsn, receive_lsn, _, pg_control_timeline, _, write_location = \
-                self._query(self.cluster_info_query)[0][:8]
+            timeline, wal_position, replay_lsn, receive_lsn, _, \
+            pg_control_timeline, _, write_location, latest_end_lsn = \
+                self._query(self.cluster_info_query)[0][:9]
             receive_lsn = max(receive_lsn or 0, write_location or 0)
 
         wal_position = self._wal_position(bool(timeline), wal_position, receive_lsn, replay_lsn)
-        return timeline, wal_position, pg_control_timeline, receive_lsn, replay_lsn
+        return timeline, wal_position, pg_control_timeline, receive_lsn, replay_lsn, latest_end_lsn
 
     def postmaster_start_time(self) -> Optional[str]:
         try:

patroni/utils.py

@@ -995,6 +995,11 @@ def cluster_as_json(cluster: 'Cluster') -> Dict[str, Any]:
                 else:
                     member[lag_type] = 0
                     member[lsn_type] = format_lsn(lsn)
+        elif m.name == leader_name and (config.is_standby_cluster or multisite_active):
+            latest_end_lsn = getattr(m, 'latest_end_lsn')
+            member['latest_end_lsn'] = format_lsn(latest_end_lsn)
+            if member['latest_end_lsn']:
+                member['lag_to_primary'] = latest_end_lsn - cluster_lsn
 
         ret['members'].append(member)