Take a max from xlog_receive and xlog_replay (#363)

patroni/ha.py

@@ -30,8 +30,8 @@ class _MemberStatus(namedtuple('_MemberStatus', 'member,reachable,in_recovery,xl
     @classmethod
     def from_api_response(cls, member, json):
         is_master = json['role'] == 'master'
-        xlog_location = None if is_master else json['xlog']['received_location']
-        return cls(member, True, not is_master, xlog_location, json.get('tags', {}))
+        xlog = not is_master and max(json['xlog'].get('received_location', 0), json['xlog'].get('replayed_location', 0))
+        return cls(member, True, not is_master, xlog, json.get('tags', {}))
 
     @classmethod
     def unknown(cls, member):

patroni/postgresql.py

@@ -1111,11 +1111,12 @@ END;
 $$""".format(name, ' '.join(options)), name, password, password)
 
     def xlog_position(self, retry=True):
-        stmt = """SELECT pg_xlog_location_diff(CASE WHEN pg_is_in_recovery()
-                                                    THEN COALESCE(pg_last_xlog_receive_location(),
-                                                                  pg_last_xlog_replay_location())
-                                                    ELSE pg_current_xlog_location()
-                                               END, '0/0')::bigint"""
+        stmt = """SELECT CASE WHEN pg_is_in_recovery()
+                              THEN GREATEST(pg_xlog_location_diff(COALESCE(pg_last_xlog_receive_location(), '0/0'),
+                                                                  '0/0')::bigint,
+                                            pg_xlog_location_diff(pg_last_xlog_replay_location(), '0/0')::bigint)
+                              ELSE pg_xlog_location_diff(pg_current_xlog_location(), '0/0')::bigint
+                          END"""
 
         # 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.

patroni/scripts/wale_restore.py

@@ -33,9 +33,6 @@ import sys
 import time
 import argparse
 
-if sys.hexversion >= 0x3000000:
-    long = int
-
 logger = logging.getLogger(__name__)
 
 RETRY_SLEEP_INTERVAL = 1
@@ -122,12 +119,12 @@ class WALERestore(object):
 
         lsn_segment = backup_start_segment[8:16]
         # first 2 characters of the result are 0x and the last one is L
-        lsn_offset = hex((long(backup_start_segment[16:32], 16) << 24) + long(backup_start_offset))[2:-1]
+        lsn_offset = hex((int(backup_start_segment[16:32], 16) << 24) + int(backup_start_offset))[2:-1]
 
         # construct the LSN from the segment and offset
         backup_start_lsn = '{0}/{1}'.format(lsn_segment, lsn_offset)
 
-        diff_in_bytes = long(backup_size)
+        diff_in_bytes = int(backup_size)
         attempts_no = 0
         while True:
             if self.master_connection:
@@ -138,7 +135,7 @@ class WALERestore(object):
                         with con.cursor() as cur:
                             cur.execute("SELECT pg_xlog_location_diff(pg_current_xlog_location(), %s)",
                                         (backup_start_lsn,))
-                            diff_in_bytes = long(cur.fetchone()[0])
+                            diff_in_bytes = int(cur.fetchone()[0])
                 except psycopg2.Error:
                     logger.exception('could not determine difference with the master location')
                     if attempts_no < self.retries:  # retry in case of a temporarily connection issue
@@ -158,8 +155,8 @@ class WALERestore(object):
 
         # if the size of the accumulated WAL segments is more than a certan percentage of the backup size
         # or exceeds the pre-determined size - pg_basebackup is chosen instead.
-        return (diff_in_bytes < long(threshold_megabytes) * 1048576) and\
-            (diff_in_bytes < long(backup_size) * float(threshold_backup_size_percentage) / 100)
+        return (diff_in_bytes < int(threshold_megabytes) * 1048576) and\
+            (diff_in_bytes < int(backup_size) * float(threshold_backup_size_percentage) / 100)
 
     def fix_subdirectory_path_if_broken(self, dirname):
         # in case it is a symlink pointing to a non-existing location, remove it and create the actual directory

patroni/utils.py

@@ -1,14 +1,10 @@
 import random
-import sys
 import time
 import re
 
 from dateutil import tz
 from patroni.exceptions import PatroniException
 
-if sys.hexversion >= 0x3000000:
-    long = int
-
 tzutc = tz.tzutc()
 
 
@@ -121,7 +117,7 @@ def strtol(value, strict=True):
         while i <= l:
             try:  # try to find maximally long number
                 i += 1  # by giving to `int` longer and longer strings
-                ret = long(value[:i], base)
+                ret = int(value[:i], base)
             except ValueError:  # until we will not get an exception or end of the string
                 i -= 1
                 break

tests/test_postgresql.py

@@ -28,7 +28,7 @@ class MockCursor(object):
             raise RetryFailedError('retry')
         elif sql.startswith('SELECT slot_name'):
             self.results = [('blabla',), ('foobar',)]
-        elif sql.startswith('SELECT pg_xlog_location_diff'):
+        elif sql.startswith('SELECT CASE WHEN pg_is_in_recovery()'):
             self.results = [(0,)]
         elif sql == 'SELECT pg_is_in_recovery()':
             self.results = [(False, )]

tests/test_wale_restore.py

@@ -24,7 +24,8 @@ wale_output = b'name last_modified expanded_size_bytes wal_segment_backup_start
 class TestWALERestore(unittest.TestCase):
 
     def setUp(self):
-        self.wale_restore = WALERestore("batman", "/data", "host=batman port=5432 user=batman", "/etc", 100, 100, 1, 0, 1)
+        self.wale_restore = WALERestore("batman", "/data", "host=batman port=5432 user=batman",
+                                        "/etc", 100, 100, 1, 0, 1)
 
     def test_should_use_s3_to_create_replica(self):
         self.assertTrue(self.wale_restore.should_use_s3_to_create_replica())