Merge pull request #27 from CyberDem0n/master

BUGFIX: keep list of replication slots on slaves empty

README.md

@@ -25,8 +25,6 @@ We provide a haproxy configuration, which will give your application a single en
 
 ```
 > haproxy -f haproxy.cfg
-> sh haproxy_status.sh 127.0.0.1 5432 15432
-> sh haproxy_status.sh 127.0.0.1 5433 15433
 ```
 
 ```
@@ -64,7 +62,7 @@ For an example file, see `postgres0.yml`.  Below is an explanation of settings:
 
 Governor uses Postgres' streaming replication.  By default, this replication is asynchronous.  For more information, see the [Postgres documentation on streaming replication](http://www.postgresql.org/docs/current/static/warm-standby.html#STREAMING-REPLICATION). 
 
-Governor's asynchronous replication configuration allows for `maximum_lag_on_failover` settings. This setting ensures replication will not occur if a follower is more than a certain number of bytes behind the follower.  This setting should be increased or decreased based on business requirements.
+Governor's asynchronous replication configuration allows for `maximum_lag_on_failover` settings. This setting ensures failover will not occur if a follower is more than a certain number of bytes behind the follower.  This setting should be increased or decreased based on business requirements.
 
 When asynchronous replication is not best for your use-case, investigate how Postgres's [synchronous replication](http://www.postgresql.org/docs/current/static/warm-standby.html#SYNCHRONOUS-REPLICATION) works.  Synchronous replication ensures consistency across a cluster by confirming that writes are written to a secondary before returning to the connecting client with a success.  The cost of synchronous replication will be reduced throughput on writes.  This throughput will be entirely based on network performance.  In hosted datacenter environments (like AWS, Rackspace, or any network you do not control), synchrous replication increases the variability of write performance significantly.  If followers become inaccessible from the leader, the leader will becomes effectively readonly.
 

governor.py

@@ -77,7 +77,13 @@ class Governor:
         while True:
             self.touch_member()
             logging.info(self.ha.run_cycle())
-
+            try:
+                if self.ha.state_handler.is_leader():
+                    self.ha.cluster and self.ha.state_handler.create_replication_slots(self.ha.cluster)
+                else:
+                    self.ha.state_handler.drop_replication_slots()
+            except:
+                logging.exception('Exception when changing replication slots')
             self.schedule_next_run()
 
 

haproxy.cfg

@@ -17,5 +17,5 @@ frontend ft_postgresql
 backend bk_db
 	option httpchk GET
 
-  server postgresql_127.0.0.1_5432 127.0.0.1:5432 maxconn 100 check port 15432
-  server postgresql_127.0.0.1_5433 127.0.0.1:5433 maxconn 100 check port 15433
+  server postgresql_127.0.0.1_5432 127.0.0.1:5432 maxconn 100 check port 8008
+  server postgresql_127.0.0.1_5433 127.0.0.1:5433 maxconn 100 check port 8009

helpers/ha.py

@@ -50,8 +50,9 @@ class Ha:
                     if self.acquire_lock():
                         if self.state_handler.is_leader() or self.state_handler.is_promoted:
                             return 'acquired session lock as a leader'
-                        self.state_handler.promote()
-                        return 'promoted self to leader by acquiring session lock'
+                        else:
+                            self.state_handler.promote()
+                            return 'promoted self to leader by acquiring session lock'
                     else:
                         self.load_cluster_from_etcd()
                         if self.state_handler.is_leader():
@@ -70,14 +71,11 @@ class Ha:
                         return 'following a different leader because i am not the healthiest node'
             else:
                 if self.has_lock() and self.update_lock():
-                    try:
-                        if self.state_handler.is_leader() or self.state_handler.is_promoted:
-                            return 'no action.  i am the leader with the lock'
+                    if self.state_handler.is_leader() or self.state_handler.is_promoted:
+                        return 'no action.  i am the leader with the lock'
+                    else:
                         self.state_handler.promote()
                         return 'promoted self to leader because i had the session lock'
-                    finally:
-                        # create replication slots
-                        self.state_handler.create_replication_slots(self.cluster)
                 else:
                     logger.info('does not have lock')
                     if self.state_handler.is_leader():

helpers/postgresql.py

@@ -401,8 +401,7 @@ primary_conninfo = '{}'
         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, cluster):
-        members = [m.name for m in cluster.members if m.name != self.name]
+    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)
@@ -416,5 +415,11 @@ primary_conninfo = '{}'
                            WHERE slot_name = %s)""", slot, slot)
         self.members = members
 
+    def create_replication_slots(self, cluster):
+        self.sync_replication_slots([m.name for m in cluster.members if m.name != self.name])
+
+    def drop_replication_slots(self):
+        self.sync_replication_slots([])
+
     def last_operation(self):
         return self.xlog_position()

tests/test_governor.py

@@ -23,7 +23,7 @@ def nop(*args, **kwargs):
     pass
 
 
-def time_sleep(_):
+def time_sleep(*args):
     raise Exception()
 
 
@@ -63,6 +63,12 @@ class TestGovernor(unittest.TestCase):
         time.sleep = time_sleep
         self.assertRaises(Exception, main)
 
+    def test_governor_run(self):
+        time.sleep = time_sleep
+        self.g.postgresql.is_leader = lambda: False
+        self.g.ha.state_handler.sync_replication_slots = time_sleep
+        self.assertRaises(Exception, self.g.run)
+
     def touch_member(self):
         if not self.touched:
             self.touched = True