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..ffe93be3 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -50,6 +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.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')) @@ -255,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_last_xlog_replay_location() - '0/0000000'::pg_lsn)", - (self.xlog_position(), )) + member_cursor.execute("""SELECT pg_is_in_recovery(), + %s - pg_xlog_location_diff(pg_last_xlog_replay_location(),'0/00000')""", + (self.xlog_position(), )) row = member_cursor.fetchone() member_cursor.close() member_conn.close() @@ -305,10 +306,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)) @@ -362,25 +362,28 @@ 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): - 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/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 040d1c66..30278f87 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -111,6 +111,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'},