remove inactive replication slots from known governor nodes

governor.py

@@ -85,12 +85,17 @@ def run(config):
             # create replication slots
             if postgresql.is_leader():
                 logging.info("Governor Running: I am the Leader")
-                for node in etcd.get_client_path("/members?recursive=true")["node"]["nodes"]:
-                    member = node["key"].split('/')[-1]
-                    if member != postgresql.name:
+            for node in etcd.members():
+                member = node["hostname"]
+                if member != postgresql.name:
+                    if postgresql.is_leader():
                         postgresql.ensure_replication_slot(
                             postgresql.replication_slot_name(member)
                         )
+                    else:
+                        postgresql.drop_replication_slot(
+                            postgresql.replication_slot_name(member)
+                        )
             etcd.touch_member(postgresql.name, postgresql.connection_string)
 
             time.sleep(config["loop_wait"])

helpers/postgresql.py

@@ -200,6 +200,17 @@ def ensure_replication_slot(self, slot_name):
             "END IF; "
             "END$$;" % {"slot": slot_name})
 
+    def drop_replication_slot(self, slot_name):
+        self.query(
+            "DO LANGUAGE plpgsql $$DECLARE somevar VARCHAR; "
+            "BEGIN "
+            "SELECT slot_name INTO somevar FROM pg_replication_slots "
+            "WHERE slot_name = '%(slot)s' AND active = 'f' LIMIT 1; "
+            "IF FOUND THEN PERFORM "
+            "pg_drop_replication_slot('%(slot)s'); "
+            "END IF;"
+            "END$$;" % {"slot": slot_name})
+
     def write_pg_hba(self):
         f = open("%s/pg_hba.conf" % self.data_dir, "a")
         f.write("host replication %(username)s %(network)s md5" %