Always use pg_xlog_location_diff to calculate bytes written to xlog

Drop unused functionality (lsn_to_bytes, bytes_to_lsn)
Revert some changes.

helpers/postgresql.py

@@ -35,7 +35,6 @@ def parseurl(url):
 
 
 class Postgresql:
-    _SERVER_VERSION = 90400
 
     def __init__(self, config):
         self.config = config
@@ -82,7 +81,6 @@ class Postgresql:
             r = parseurl('postgres://{}/postgres'.format(self.local_address))
             self._connection = psycopg2.connect(**r)
             self._connection.autocommit = True
-            Postgresql._SERVER_VERSION = self._connection.server_version
         return self._connection
 
     def _cursor(self):
@@ -254,7 +252,9 @@ class Postgresql:
                 member_conn = psycopg2.connect(**r)
                 member_conn.autocommit = True
                 member_cursor = member_conn.cursor()
-                member_cursor.execute("SELECT pg_is_in_recovery(), COALESCE(pg_last_xlog_replay_location(), '0/0')")
+                member_cursor.execute(
+                    "SELECT pg_is_in_recovery(), %s - pg_xlog_location_diff(pg_last_xlog_replay_location(), '0/0')",
+                    (self.xlog_position(),))
                 row = member_cursor.fetchone()
                 member_cursor.close()
                 member_conn.close()
@@ -262,7 +262,7 @@ class Postgresql:
                 if not row[0]:
                     logger.warning('Master (%s) is still alive', member.name)
                     return False
-                if self.xlog_position() < self.lsn_to_bytes(row[1], member_conn.server_version):
+                if row[1] < 0:
                     return False
             except psycopg2.Error:
                 continue
@@ -357,10 +357,10 @@ recovery_target_timeline = 'latest'
                 self.admin['username']), self.admin['password'])
 
     def xlog_position(self):
-        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)
+        return self.query("""SELECT pg_xlog_location_diff(CASE WHEN pg_is_in_recovery()
+                                                               THEN pg_last_xlog_replay_location()
+                                                               ELSE pg_current_xlog_location()
+                                                          END, '0/0')""").fetchone()[0]
 
     def load_replication_slots(self):
         if self.use_slots:
@@ -391,34 +391,3 @@ recovery_target_timeline = 'latest'
 
     def last_operation(self):
         return str(self.xlog_position())
-
-    @staticmethod
-    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 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)
-        except ValueError:
-            return 0
-
-    @staticmethod
-    def bytes_to_lsn(value, version=None):
-        """
-        >>> Postgresql.bytes_to_lsn(6006243424)
-        '1/66000060'
-        """
-        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,14 +21,12 @@ def false(*args, **kwargs):
 
 
 class MockCursor:
-    _count = 0
 
     def __init__(self):
         self.closed = False
         self.results = []
 
     def execute(self, sql, *params):
-        MockCursor._count += 1
         if sql.startswith('blabla'):
             raise psycopg2.OperationalError()
         elif sql.startswith('InterfaceError'):
@@ -37,15 +35,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(), COALESCE'):
-            if MockCursor._count == 1:
+        elif sql.startswith('SELECT pg_is_in_recovery(), %s'):
+            if params[0][0] == 1:
                 raise psycopg2.OperationalError()
-            elif MockCursor._count == 2:
-                self.results = [(True, '0/1')]
+            elif params[0][0] == 2:
+                self.results = [(True, -1)]
             else:
-                self.results = [(False, '0/1')]
-        elif sql.startswith('SELECT CASE WHEN pg_is_in_recovery()'):
-            self.results = [('0/0', )]
+                self.results = [(False, 0)]
+        elif sql.startswith('SELECT pg_xlog_location_diff'):
+            self.results = [(0,)]
         elif sql.startswith('SELECT pg_is_in_recovery()'):
             self.results = [(False, )]
         elif sql.startswith('SELECT to_char(pg_postmaster_start_time'):
@@ -185,9 +183,7 @@ 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