From 330f9023eecd0f5153a749a8208ac5324f2943df Mon Sep 17 00:00:00 2001 From: Josh Berkus Date: Thu, 3 Sep 2015 17:46:41 -0700 Subject: [PATCH 01/16] Interim commit to make replication slots optional for 9.3 users. --- helpers/postgresql.py | 30 +++++++++++++++++------------- 1 file changed, 17 insertions(+), 13 deletions(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index ce8ca18f..a8d749be 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -43,6 +43,7 @@ class Postgresql: def __init__(self, config): self.config = config self.name = config['name'] + self.scope = config['scope'] self.listen_addresses, self.port = config['listen'].split(':') self.data_dir = config['data_dir'] @@ -50,6 +51,7 @@ class Postgresql: self.superuser = config['superuser'] self.admin = config['admin'] self.callback = config.get('callbacks', {}) + self.use_slots = config['use_slots'] self.recovery_conf = os.path.join(self.data_dir, 'recovery.conf') self.configuration_to_save = (os.path.join(self.data_dir, 'pg_hba.conf'), os.path.join(self.data_dir, 'postgresql.conf')) @@ -365,22 +367,24 @@ primary_conninfo = '{}' ELSE pg_current_xlog_location() - '0/00000'::pg_lsn END""").fetchone()[0] def load_replication_slots(self): - cursor = self.query("SELECT slot_name FROM pg_replication_slots WHERE slot_type='physical'") - self.members = [r[0] for r in cursor] + if self.use_slots: + cursor = self.query("SELECT slot_name FROM pg_replication_slots WHERE slot_type='physical'") + self.members = [r[0] for r in cursor] def sync_replication_slots(self, members): - # drop unused slots - for slot in set(self.members) - set(members): - self.query("""SELECT pg_drop_replication_slot(%s) - WHERE EXISTS(SELECT 1 FROM pg_replication_slots - WHERE slot_name = %s)""", slot, slot) + if self.use_slots: + # drop unused slots + for slot in set(self.members) - set(members): + self.query("""SELECT pg_drop_replication_slot(%s) + WHERE EXISTS(SELECT 1 FROM pg_replication_slots + WHERE slot_name = %s)""", slot, slot) - # create new slots - for slot in set(members) - set(self.members): - self.query("""SELECT pg_create_physical_replication_slot(%s) - WHERE NOT EXISTS (SELECT 1 FROM pg_replication_slots - WHERE slot_name = %s)""", slot, slot) - self.members = members + # create new slots + for slot in set(members) - set(self.members): + self.query("""SELECT pg_create_physical_replication_slot(%s) + WHERE NOT EXISTS (SELECT 1 FROM pg_replication_slots + WHERE slot_name = %s)""", slot, slot) + self.members = members def create_replication_slots(self, cluster): self.sync_replication_slots([m.name for m in cluster.members if m.name != self.name]) From 2b801a3cccc5ebc634889366bd5565a7f3be3604 Mon Sep 17 00:00:00 2001 From: Josh Berkus Date: Thu, 3 Sep 2015 18:03:25 -0700 Subject: [PATCH 02/16] Next 9.3 compatibility commit. Removed pg_lsn, since it's not available in 9.3. --- helpers/postgresql.py | 13 ++++++------- 1 file changed, 6 insertions(+), 7 deletions(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index a8d749be..78a77174 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -258,7 +258,7 @@ class Postgresql: member_conn.autocommit = True member_cursor = member_conn.cursor() member_cursor.execute( - "SELECT pg_is_in_recovery(), %s - (pg_last_xlog_replay_location() - '0/0000000'::pg_lsn)", + "SELECT pg_is_in_recovery(), %s - pg_xlog_location_diff(pg_last_xlog_replay_location(),'0/0000000')", (self.xlog_position(), )) row = member_cursor.fetchone() member_cursor.close() @@ -307,10 +307,9 @@ class Postgresql: recovery_target_timeline = 'latest' """) if leader and leader.conn_url: - f.write(""" -primary_slot_name = '{}' -primary_conninfo = '{}' -""".format(self.name, self.primary_conninfo(leader.conn_url))) + f.write("""primary_conninfo = '{}'\n""".format(self.primary_conninfo(leader.conn_url))) + if self.use_slots: + f.write("""primary_slot_name = '{}'\n""".format(self.name)) for name, value in self.config.get('recovery_conf', {}).items(): f.write("{} = '{}'\n".format(name, value)) @@ -363,8 +362,8 @@ primary_conninfo = '{}' def xlog_position(self): return self.query("""SELECT CASE WHEN pg_is_in_recovery() - THEN pg_last_xlog_replay_location() - '0/0000000'::pg_lsn - ELSE pg_current_xlog_location() - '0/00000'::pg_lsn END""").fetchone()[0] + THEN pg_xlog_location_diff(pg_last_xlog_replay_location(),'0/0000000') + ELSE pg_xlog_location_diff(pg_current_xlog_location(),'0/00000') END""").fetchone()[0] def load_replication_slots(self): if self.use_slots: From f2338e074c006942c73b0f3f6b280895c6f0b65f Mon Sep 17 00:00:00 2001 From: Josh Berkus Date: Thu, 3 Sep 2015 18:16:44 -0700 Subject: [PATCH 03/16] Added documentation, sample config for optional replication slots. --- README.md | 5 +++-- postgres0.yml | 1 + postgres1.yml | 1 + 3 files changed, 5 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 2582eb37..cfa141f5 100644 --- a/README.md +++ b/README.md @@ -67,6 +67,7 @@ For an example file, see `postgres0.yml`. Below is an explanation of settings: * *connect_address*: ip address + port through which Postgres is accessible from other nodes and applications. * *data_dir*: file path to initialize and store Postgres data files * *maximum_lag_on_failover*: the maximum bytes a follower may lag before it is not eligible become leader + * *use_slots*: whether or not to use replication_slots. Must be False for PostgreSQL 9.3. * *pg_hba*: list of lines which should be added to pg_hba.conf * *- host all all 0.0.0.0/0 md5* * *replication* @@ -84,8 +85,8 @@ For an example file, see `postgres0.yml`. Below is an explanation of settings: * *admin*: * *username*: admin username, user will be created during initialization. It would have CREATEDB and CREATEROLE privileges * *password*: admin password, user will be created during initialization. - * *recovery_conf*: configuration settings written to recovery.conf when configuring follower - * *parameters*: list of configuration settings for Postgres + * *recovery_conf*: additional configuration settings written to recovery.conf when configuring follower + * *parameters*: list of configuration settings for Postgres. Many of these are required for replication to work. ## Replication choices diff --git a/postgres0.yml b/postgres0.yml index a2a7ce44..15dff82c 100644 --- a/postgres0.yml +++ b/postgres0.yml @@ -30,6 +30,7 @@ postgresql: connect_address: 127.0.0.1:5432 data_dir: data/postgresql0 maximum_lag_on_failover: 1048576 # 1 megabyte in bytes + use_slots: True pg_hba: - host all all 0.0.0.0/0 md5 - hostssl all all 0.0.0.0/0 md5 diff --git a/postgres1.yml b/postgres1.yml index 6ef6b1c9..90731fe0 100644 --- a/postgres1.yml +++ b/postgres1.yml @@ -30,6 +30,7 @@ postgresql: connect_address: 127.0.0.1:5433 data_dir: data/postgresql1 maximum_lag_on_failover: 1048576 # 1 megabyte in bytes + use_slots: True pg_hba: - host all all 0.0.0.0/0 md5 - hostssl all all 0.0.0.0/0 md5 From f5627a498e4e46f5480b38020f8a23194314285e Mon Sep 17 00:00:00 2001 From: Josh Berkus Date: Thu, 3 Sep 2015 18:41:30 -0700 Subject: [PATCH 04/16] Fixed test_postgresql.py to include use_slots. --- README.md | 2 +- tests/test_postgresql.py | 1 + 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index cfa141f5..345c12c3 100644 --- a/README.md +++ b/README.md @@ -67,7 +67,7 @@ For an example file, see `postgres0.yml`. Below is an explanation of settings: * *connect_address*: ip address + port through which Postgres is accessible from other nodes and applications. * *data_dir*: file path to initialize and store Postgres data files * *maximum_lag_on_failover*: the maximum bytes a follower may lag before it is not eligible become leader - * *use_slots*: whether or not to use replication_slots. Must be False for PostgreSQL 9.3. + * *use_slots*: whether or not to use replication_slots. Must be False for PostgreSQL 9.3, and you should comment out max_replication_slots. * *pg_hba*: list of lines which should be added to pg_hba.conf * *- host all all 0.0.0.0/0 md5* * *replication* diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index c4a3d6df..882886df 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -114,6 +114,7 @@ class TestPostgresql(unittest.TestCase): 'pg_hba': ['hostssl all all 0.0.0.0/0 md5', 'host all all 0.0.0.0/0 md5'], 'superuser': {'password': ''}, 'admin': {'username': 'admin', 'password': 'admin'}, + 'use_slots' : True, 'replication': {'username': 'replicator', 'password': 'rep-pass', 'network': '127.0.0.1/32'}, From 5f4a9ffabb77499f1565a1fa8e45eb2532d033dc Mon Sep 17 00:00:00 2001 From: Josh Berkus Date: Thu, 3 Sep 2015 21:04:42 -0700 Subject: [PATCH 05/16] Text spacing changes in an attempt to get flake8 to stop complaining. --- helpers/postgresql.py | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index 78a77174..2a6af912 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -43,7 +43,6 @@ class Postgresql: def __init__(self, config): self.config = config self.name = config['name'] - self.scope = config['scope'] self.listen_addresses, self.port = config['listen'].split(':') self.data_dir = config['data_dir'] @@ -257,8 +256,8 @@ 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/0000000')", + member_cursor.execute("""SELECT pg_is_in_recovery(), + %s - pg_xlog_location_diff(pg_last_xlog_replay_location(),'0/0000000')""", (self.xlog_position(), )) row = member_cursor.fetchone() member_cursor.close() @@ -362,8 +361,8 @@ recovery_target_timeline = 'latest' 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/0000000') - ELSE pg_xlog_location_diff(pg_current_xlog_location(),'0/00000') END""").fetchone()[0] + THEN pg_xlog_location_diff(pg_last_xlog_replay_location(),'0/0000000') + ELSE pg_xlog_location_diff(pg_current_xlog_location(),'0/00000') END""").fetchone()[0] def load_replication_slots(self): if self.use_slots: From bdb1454e3505031ac609a26c228fa179288613fb Mon Sep 17 00:00:00 2001 From: Josh Berkus Date: Thu, 3 Sep 2015 21:10:30 -0700 Subject: [PATCH 06/16] Another commit because flake8 is a huge waste of time. --- helpers/postgresql.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index 2a6af912..24f903e9 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -256,9 +256,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(), - %s - pg_xlog_location_diff(pg_last_xlog_replay_location(),'0/0000000')""", - (self.xlog_position(), )) + member_cursor.execute("""SELECT pg_is_in_recovery(), + %s - pg_xlog_location_diff(pg_last_xlog_replay_location(),'0/0000000')""", + (self.xlog_position(), )) row = member_cursor.fetchone() member_cursor.close() member_conn.close() From 2b62adae210f487ddf667f6787a81679a9b55a8c Mon Sep 17 00:00:00 2001 From: Josh Berkus Date: Thu, 3 Sep 2015 21:22:59 -0700 Subject: [PATCH 07/16] Fixed inherited bug in is_healthiest_node() with LSN position. --- helpers/postgresql.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index 24f903e9..62c7bb7e 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -257,7 +257,7 @@ class Postgresql: 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/0000000')""", + %s - pg_xlog_location_diff(pg_last_xlog_replay_location(),'0/00000')""", (self.xlog_position(), )) row = member_cursor.fetchone() member_cursor.close() From 6df56fc6ccfe68259bd0d1556843e72d7e9b65e6 Mon Sep 17 00:00:00 2001 From: Josh Berkus Date: Thu, 3 Sep 2015 21:41:33 -0700 Subject: [PATCH 08/16] Another try at the is_healthiest_node bugfix. --- helpers/postgresql.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index 62c7bb7e..bc6aed5b 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -382,7 +382,8 @@ recovery_target_timeline = 'latest' self.query("""SELECT pg_create_physical_replication_slot(%s) WHERE NOT EXISTS (SELECT 1 FROM pg_replication_slots WHERE slot_name = %s)""", slot, slot) - self.members = members + + self.members = members def create_replication_slots(self, cluster): self.sync_replication_slots([m.name for m in cluster.members if m.name != self.name]) From aeea7196bd1eebac4411c414786cf2175bad61da Mon Sep 17 00:00:00 2001 From: Josh Berkus Date: Thu, 3 Sep 2015 21:46:06 -0700 Subject: [PATCH 09/16] Fixing flake8 issue. --- helpers/postgresql.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index bc6aed5b..dbdecb8f 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -382,8 +382,8 @@ recovery_target_timeline = 'latest' self.query("""SELECT pg_create_physical_replication_slot(%s) WHERE NOT EXISTS (SELECT 1 FROM pg_replication_slots WHERE slot_name = %s)""", slot, slot) - - self.members = members + + self.members = members def create_replication_slots(self, cluster): self.sync_replication_slots([m.name for m in cluster.members if m.name != self.name]) From 5612cd0280ec016831efa0e461f7825d1bac3400 Mon Sep 17 00:00:00 2001 From: Josh Berkus Date: Thu, 3 Sep 2015 22:20:27 -0700 Subject: [PATCH 10/16] Changed use_slots to be backwards compatible by using config.get() per PR feedback. --- helpers/postgresql.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index dbdecb8f..03e4f4ef 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -50,7 +50,7 @@ class Postgresql: self.superuser = config['superuser'] self.admin = config['admin'] self.callback = config.get('callbacks', {}) - self.use_slots = config['use_slots'] + self.use_slots = config.get('use_slots',True) self.recovery_conf = os.path.join(self.data_dir, 'recovery.conf') self.configuration_to_save = (os.path.join(self.data_dir, 'pg_hba.conf'), os.path.join(self.data_dir, 'postgresql.conf')) From 936173272260288023dade904324067dbcc7e311 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Fri, 4 Sep 2015 08:29:16 +0200 Subject: [PATCH 11/16] Fix pep8 formatting and unit-tests --- helpers/postgresql.py | 20 ++++++++++---------- tests/test_postgresql.py | 3 +-- 2 files changed, 11 insertions(+), 12 deletions(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index ffe93be3..db4d3cd8 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -50,7 +50,7 @@ class Postgresql: self.superuser = config['superuser'] self.admin = config['admin'] self.callback = config.get('callbacks', {}) - self.use_slots = config.get('use_slots',True) + self.use_slots = config.get('use_slots', True) self.recovery_conf = os.path.join(self.data_dir, 'recovery.conf') self.configuration_to_save = (os.path.join(self.data_dir, 'pg_hba.conf'), os.path.join(self.data_dir, 'postgresql.conf')) @@ -256,9 +256,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(), - %s - pg_xlog_location_diff(pg_last_xlog_replay_location(),'0/00000')""", - (self.xlog_position(), )) + 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() @@ -362,8 +362,8 @@ recovery_target_timeline = 'latest' 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/0000000') - ELSE pg_xlog_location_diff(pg_current_xlog_location(),'0/00000') END""").fetchone()[0] + 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] def load_replication_slots(self): if self.use_slots: @@ -375,14 +375,14 @@ recovery_target_timeline = 'latest' # drop unused slots for slot in set(self.members) - set(members): self.query("""SELECT pg_drop_replication_slot(%s) - WHERE EXISTS(SELECT 1 FROM pg_replication_slots - WHERE slot_name = %s)""", slot, slot) + WHERE EXISTS(SELECT 1 FROM pg_replication_slots + WHERE slot_name = %s)""", slot, slot) # create new slots for slot in set(members) - set(self.members): self.query("""SELECT pg_create_physical_replication_slot(%s) - WHERE NOT EXISTS (SELECT 1 FROM pg_replication_slots - WHERE slot_name = %s)""", slot, slot) + WHERE NOT EXISTS (SELECT 1 FROM pg_replication_slots + WHERE slot_name = %s)""", slot, slot) self.members = members diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 30278f87..ade68d3a 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -111,7 +111,6 @@ class TestPostgresql(unittest.TestCase): 'pg_hba': ['hostssl all all 0.0.0.0/0 md5', 'host all all 0.0.0.0/0 md5'], 'superuser': {'password': ''}, 'admin': {'username': 'admin', 'password': 'admin'}, - 'use_slots' : True, 'replication': {'username': 'replicator', 'password': 'rep-pass', 'network': '127.0.0.1/32'}, @@ -120,7 +119,7 @@ class TestPostgresql(unittest.TestCase): 'on_restart': 'true', 'on_role_change': 'true', 'on_reload': 'true' }, - 'restore': '/usr/bin/true'}) + 'restore': 'true'}) psycopg2.connect = psycopg2_connect if not os.path.exists(self.p.data_dir): os.makedirs(self.p.data_dir) From 8b9e99090fc2cf2c8808fea711c9e8f9db9e588e Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Fri, 4 Sep 2015 10:41:59 +0200 Subject: [PATCH 12/16] Move lsn_to_bytes and bytes_to_lsn into Postgresql class and make their behavior version specific --- helpers/postgresql.py | 33 +++++++++++++++++++++++++++++---- helpers/utils.py | 26 -------------------------- tests/test_postgresql.py | 1 + 3 files changed, 30 insertions(+), 30 deletions(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index db4d3cd8..f84f87a0 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -4,14 +4,10 @@ import psycopg2 import shlex import shutil import subprocess -import six from helpers.utils import sleep from six.moves.urllib_parse import urlparse -if six.PY3: - long = int - logger = logging.getLogger(__name__) ACTION_ON_START = "on_start" @@ -39,6 +35,7 @@ def parseurl(url): class Postgresql: + _SERVER_VERSION = 90400 def __init__(self, config): self.config = config @@ -85,6 +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 return self._connection def _cursor(self): @@ -394,3 +392,30 @@ recovery_target_timeline = 'latest' def last_operation(self): return str(self.xlog_position()) + + @staticmethod + def lsn_to_bytes(value): + """ + >>> Postgresql.lsn_to_bytes('1/66000060') + 6006243424 + >>> Postgresql.lsn_to_bytes('j/66000060') + 0 + """ + try: + multiplier = 0xFF000000 if Postgresql._SERVER_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): + """ + >>> Postgresql.bytes_to_lsn(6006243424) + '1/66000060' + """ + divider = 0xFF000000 if Postgresql._SERVER_VERSION < 90300 else 0x100000000 + segment = value / divider + offset = value % divider + return '%X/%X' % (segment, offset) diff --git a/helpers/utils.py b/helpers/utils.py index c725e6d4..c0553374 100644 --- a/helpers/utils.py +++ b/helpers/utils.py @@ -42,32 +42,6 @@ def calculate_ttl(expiration): return int((expiration - now).total_seconds()) -def lsn_to_bytes(value): - """ - >>> lsn_to_bytes('1/66000060') - 6006243424 - >>> lsn_to_bytes('j/66000060') - 0 - """ - try: - e = value.split('/') - if len(e) == 2 and len(e[0]) > 0 and len(e[1]) > 0: - return (int(e[0], 16) << 32) | int(e[1], 16) - except ValueError: - pass - return 0 - - -def bytes_to_lsn(value): - """ - >>> bytes_to_lsn(6006243424) - '1/66000060' - """ - id = value >> 32 - off = value & 0xffffffff - return '%x/%x' % (id, off) - - def sigterm_handler(signo, stack_frame): sys.exit() diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index ade68d3a..f8187c9f 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -79,6 +79,7 @@ class MockConnect: def __init__(self): self.autocommit = False self.closed = 0 + self.server_version = 90400 def cursor(self): return MockCursor() From c913c8ad9270ee3ae100ac3912d06d2ace6d395b Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Fri, 4 Sep 2015 12:11:55 +0200 Subject: [PATCH 13/16] 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 | 27 +++++++++++++++------------ tests/test_postgresql.py | 19 +++++++++++-------- 2 files changed, 26 insertions(+), 20 deletions(-) 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): From a69565fc5f6a002c97faea8e6e544ecb17aeae73 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Fri, 4 Sep 2015 13:22:44 +0200 Subject: [PATCH 14/16] Set Postgresql._SERVER_VERSION only after real connect. --- helpers/postgresql.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index 642c48f4..3fc7983a 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 - Postgresql._SERVER_VERSION = self._connection.server_version + Postgresql._SERVER_VERSION = self._connection.server_version return self._connection def _cursor(self): @@ -358,8 +358,8 @@ recovery_target_timeline = 'latest' 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] + 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): From 7cce02ae958a499aca707c87131db051710bb269 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Sat, 5 Sep 2015 15:19:17 +0200 Subject: [PATCH 15/16] 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 | 47 +++++++--------------------------------- tests/test_postgresql.py | 18 ++++++--------- 2 files changed, 15 insertions(+), 50 deletions(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index 3fc7983a..e451d844 100644 --- a/helpers/postgresql.py +++ b/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) diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index a63c240b..ca33f3e5 100644 --- a/tests/test_postgresql.py +++ b/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 From ac2740eeb91813ebd9d470a79d79c165aa5b5078 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Sat, 5 Sep 2015 15:22:38 +0200 Subject: [PATCH 16/16] Drop unused variable server_version --- tests/test_postgresql.py | 1 - 1 file changed, 1 deletion(-) diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index ca33f3e5..b3c91f73 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -78,7 +78,6 @@ class MockConnect: def __init__(self): self.autocommit = False self.closed = 0 - self.server_version = 90400 def cursor(self): return MockCursor()