diff --git a/governor.py b/governor.py index e976798c..e145531c 100755 --- a/governor.py +++ b/governor.py @@ -69,10 +69,7 @@ while True: # create replication slots if postgresql.is_leader(): - for member in etcd.members(): - if member['hostname'] != postgresql.name: - postgresql.query("""SELECT pg_create_physical_replication_slot(%s) - WHERE NOT EXISTS (SELECT 1 FROM pg_replication_slots - WHERE slot_name = %s)""", member['hostname'], member['hostname']) + members = [m['hostname'] for m in etcd.members() if m['hostname'] != postgresql.name] + postgresql.create_replication_slots(members) time.sleep(config["loop_wait"]) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index c21d6440..8d4f2739 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -35,6 +35,7 @@ class Postgresql: self.conn = None self.cursor_holder = None + self.members = [] # list of already existing replication slots def cursor(self): if not self.cursor_holder: @@ -100,6 +101,7 @@ class Postgresql: def start(self): if self.is_running(): + self.load_replication_slots() logger.error('Cannot start PostgreSQL because one is already running.') return False @@ -108,7 +110,9 @@ class Postgresql: os.remove(pid_path) logger.info('Removed %s', pid_path) - return os.system(self._pg_ctl + ' start -o "{}"'.format(self.server_options())) == 0 + ret = os.system(self._pg_ctl + ' start -o "{}"'.format(self.server_options())) == 0 + ret and self.load_replication_slots() + return ret def stop(self): return os.system(self._pg_ctl + ' stop') != 0 @@ -213,3 +217,21 @@ primary_conninfo = '{}' def xlog_position(self): return self.query('SELECT pg_last_xlog_replay_location()').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] + + def create_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) + + # 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