Track list of already existing physical replication slots

Drop replication slot when it was removed from etcd.
Execute pg_create_physical_replication_slot only when something new appeared in etcd.

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"])

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