diff --git a/helpers/postgresql.py b/helpers/postgresql.py index f84f87a0..642c48f4 100644 --- a/helpers/postgresql.py +++ b/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) diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index f8187c9f..a63c240b 100644 --- a/tests/test_postgresql.py +++ b/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):