diff --git a/README.md b/README.md index 2582eb37..345c12c3 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, 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* @@ -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/helpers/postgresql.py b/helpers/postgresql.py index 8c9fd3c9..e451d844 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" @@ -50,6 +46,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.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,8 +253,8 @@ 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)", - (self.xlog_position(), )) + "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() @@ -305,10 +302,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)) @@ -361,26 +357,30 @@ primary_conninfo = '{}' self.admin['username']), self.admin['password']) 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] + 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): - 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) - # 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): diff --git a/helpers/utils.py b/helpers/utils.py index f0555f48..c2047918 100644 --- a/helpers/utils.py +++ b/helpers/utils.py @@ -45,32 +45,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/postgres0.yml b/postgres0.yml index dd8f8356..ce90da14 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 c2cb5ee6..763444e8 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 diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 08ba0b33..c027c52c 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -24,7 +24,6 @@ class MockCursor: def __init__(self): self.closed = False - self.current = 0 self.results = [] def execute(self, sql, *params): @@ -43,7 +42,7 @@ class MockCursor: self.results = [(True, -1)] else: self.results = [(False, 0)] - elif sql.startswith('SELECT CASE WHEN pg_is_in_recovery()'): + elif sql.startswith('SELECT pg_xlog_location_diff'): self.results = [(0,)] elif sql.startswith('SELECT pg_is_in_recovery()'): self.results = [(False, )] @@ -119,7 +118,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) @@ -189,7 +188,7 @@ class TestPostgresql(unittest.TestCase): 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):