Fix handling of replication slots on the master.

Master shouldn't keep a replication slot for the members that
replicate from other members instead of the master (replicatefrom).
Otherwise, the master will keep collecting WAL segments that won't
be requested ever. Of course, if the destination of replicatefrom
is not part of the cluster, master should create the slot.

patroni/dcs.py

@@ -111,6 +111,9 @@ class Cluster(namedtuple('Cluster', 'initialize,leader,last_leader_operation,mem
     def is_unlocked(self):
         return not (self.leader and self.leader.name)
 
+    def has_member(self, member_name):
+        return len([m for m in self.members if m.name == member_name]) > 0
+
 
 class AbstractDCS:
 

patroni/postgresql.py

@@ -651,11 +651,18 @@ $$""".format(name, options), name, password, password)
         if self.use_slots:
             try:
                 self.load_replication_slots()
+                # if the replicatefrom tag is set on the member - we should not create the replication slot for it on
+                # the current master, because that member would replicate from elsewhere. We still create the slot if
+                # the replicatefrom destination member is currently not a member of the cluster (fallback to the
+                # master), or if replicatefrom destination member happens to be the current master
                 if self.role == 'master':
-                    slots = [m.name for m in cluster.members if m.name != self.name]
+                    slots = [m.name for m in cluster.members if m.name != self.name and
+                             (not cluster.has_member(m.replicatefrom)
+                              if m.replicatefrom and m.replicatefrom != self.name else True)]
                 else:
                     # only manage slots for replicas that want to replicate from this one
                     slots = [m.name for m in cluster.members if m.replicatefrom == self.name]
+                logger.info("setting replication slots for members {0}".format(slots))
                 # drop unused slots
                 for slot in set(self.replication_slots) - set(slots):
                     self.query("""SELECT pg_drop_replication_slot(%s)
@@ -675,7 +682,7 @@ $$""".format(name, options), name, password, password)
     def last_operation(self):
         return str(self.xlog_position())
 
-    def bootstrap(self, initialize=False, current_leader=None):
+    def bootstrap(self, cluster_initialized=False, current_leader=None):
         """
             Populate PostgreSQL data directory by doing one of the following:
              - create with initdb if there is no master.
@@ -696,7 +703,7 @@ $$""".format(name, options), name, password, password)
             that should be retried in the future.
         """
         ret = False
-        if not (initialize or current_leader):
+        if not (cluster_initialized or current_leader):
             ret = self.initialize() and self.start()
             if ret:
                 self.create_replication_user()

tests/test_postgresql.py

@@ -181,7 +181,8 @@ class TestPostgresql(unittest.TestCase):
             os.makedirs(self.p.data_dir)
         self.leadermem = Member(0, 'leader', 28, {'conn_url': 'postgres://replicator:[email protected]:5435/postgres'})
         self.leader = Leader(-1, 28, self.leadermem)
-        self.other = Member(0, 'test1', 28, {'conn_url': 'postgres://replicator:[email protected]:5433/postgres'})
+        self.other = Member(0, 'test1', 28, {'conn_url': 'postgres://replicator:[email protected]:5433/postgres',
+                            'tags': {'replicatefrom': 'leader'}})
         self.me = Member(0, 'test0', 28, {'conn_url': 'postgres://replicator:[email protected]:5434/postgres'})
 
     def tearDown(self):