Calculate xlog bytes with using lsn_to_bytes method

This method behaves differently depending on server version and will
allow to use patroni with postgres older than 9.3

helpers/postgresql.py

@@ -82,7 +82,7 @@ class Postgresql:
             r = parseurl('postgres://{}/postgres'.format(self.local_address))
             self._connection = psycopg2.connect(**r)
             self._connection.autocommit = True
-        self._SERVER_VERSION = self._connection.server_version
+        Postgresql._SERVER_VERSION = self._connection.server_version
         return self._connection
 
     def _cursor(self):
@@ -254,9 +254,7 @@ class Postgresql:
                 member_conn = psycopg2.connect(**r)
                 member_conn.autocommit = True
                 member_cursor = member_conn.cursor()
-                member_cursor.execute(
-                    "SELECT pg_is_in_recovery(), %s - pg_xlog_location_diff(pg_last_xlog_replay_location(),'0/0')",
-                    (self.xlog_position(), ))
+                member_cursor.execute("SELECT pg_is_in_recovery(), COALESCE(pg_last_xlog_replay_location(), '0/0')")
                 row = member_cursor.fetchone()
                 member_cursor.close()
                 member_conn.close()
@@ -264,7 +262,7 @@ class Postgresql:
                 if not row[0]:
                     logger.warning('Master (%s) is still alive', member.name)
                     return False
-                if row[1] < 0:
+                if self.xlog_position() < self.lsn_to_bytes(row[1], member_conn.server_version):
                     return False
             except psycopg2.Error:
                 continue
@@ -359,9 +357,10 @@ recovery_target_timeline = 'latest'
                 self.admin['username']), self.admin['password'])
 
     def xlog_position(self):
-        return self.query("""SELECT CASE WHEN pg_is_in_recovery()
-                               THEN pg_xlog_location_diff(pg_last_xlog_replay_location(),'0/0')
-                               ELSE pg_xlog_location_diff(pg_current_xlog_location(),'0/0') END""").fetchone()[0]
+        lsn = self.query("""SELECT CASE WHEN pg_is_in_recovery()
+                               THEN pg_last_xlog_replay_location()
+                               ELSE pg_current_xlog_location() END""").fetchone()[0]
+        return self.lsn_to_bytes(lsn)
 
     def load_replication_slots(self):
         if self.use_slots:
@@ -394,15 +393,17 @@ recovery_target_timeline = 'latest'
         return str(self.xlog_position())
 
     @staticmethod
-    def lsn_to_bytes(value):
+    def lsn_to_bytes(value, version=None):
         """
         >>> Postgresql.lsn_to_bytes('1/66000060')
         6006243424
         >>> Postgresql.lsn_to_bytes('j/66000060')
         0
         """
+        if version is None:
+            version = Postgresql._SERVER_VERSION
         try:
-            multiplier = 0xFF000000 if Postgresql._SERVER_VERSION < 90300 else 0x100000000
+            multiplier = 0xFF000000 if version < 90300 else 0x100000000
             e = value.split('/')
             if len(e) == 2 and len(e[0]) > 0 and len(e[1]) > 0:
                 return int(e[0], 16) * multiplier + int(e[1], 16)
@@ -410,12 +411,14 @@ recovery_target_timeline = 'latest'
             return 0
 
     @staticmethod
-    def bytes_to_lsn(value):
+    def bytes_to_lsn(value, version=None):
         """
         >>> Postgresql.bytes_to_lsn(6006243424)
         '1/66000060'
         """
-        divider = 0xFF000000 if Postgresql._SERVER_VERSION < 90300 else 0x100000000
+        if version is None:
+            version = Postgresql._SERVER_VERSION
+        divider = 0xFF000000 if version < 90300 else 0x100000000
         segment = value / divider
         offset = value % divider
         return '%X/%X' % (segment, offset)

tests/test_postgresql.py

@@ -21,13 +21,14 @@ def false(*args, **kwargs):
 
 
 class MockCursor:
+    _count = 0
 
     def __init__(self):
         self.closed = False
-        self.current = 0
         self.results = []
 
     def execute(self, sql, *params):
+        MockCursor._count += 1
         if sql.startswith('blabla'):
             raise psycopg2.OperationalError()
         elif sql.startswith('InterfaceError'):
@@ -36,15 +37,15 @@ class MockCursor:
             self.results = [('blabla',), ('foobar',)]
         elif sql.startswith('SELECT pg_current_xlog_location()'):
             self.results = [(0,)]
-        elif sql.startswith('SELECT pg_is_in_recovery(), %s'):
-            if params[0][0] == 1:
+        elif sql.startswith('SELECT pg_is_in_recovery(), COALESCE'):
+            if MockCursor._count == 1:
                 raise psycopg2.OperationalError()
-            elif params[0][0] == 2:
-                self.results = [(True, -1)]
+            elif MockCursor._count == 2:
+                self.results = [(True, '0/1')]
             else:
-                self.results = [(False, 0)]
+                self.results = [(False, '0/1')]
         elif sql.startswith('SELECT CASE WHEN pg_is_in_recovery()'):
-            self.results = [(0,)]
+            self.results = [('0/0', )]
         elif sql.startswith('SELECT pg_is_in_recovery()'):
             self.results = [(False, )]
         elif sql.startswith('SELECT to_char(pg_postmaster_start_time'):
@@ -184,12 +185,14 @@ class TestPostgresql(unittest.TestCase):
         cluster = Cluster(True, self.leader, 0, [self.me, self.other, self.leader])
         self.assertTrue(self.p.is_healthiest_node(cluster))
         self.p.is_leader = false
+        MockCursor._count = 0
         self.assertFalse(self.p.is_healthiest_node(cluster))
+        MockCursor._count = 0
         self.p.xlog_position = lambda: 1
         self.assertTrue(self.p.is_healthiest_node(cluster))
         self.p.xlog_position = lambda: 2
         self.assertFalse(self.p.is_healthiest_node(cluster))
-        self.p.config['maximum_lag_on_failover'] = -2
+        self.p.config['maximum_lag_on_failover'] = -3
         self.assertFalse(self.p.is_healthiest_node(cluster))
 
     def test_is_leader(self):