diff --git a/README.md b/README.md index 37050983..6f6560b9 100644 --- a/README.md +++ b/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. diff --git a/governor.py b/governor.py index d3eb5334..b752feb3 100755 --- a/governor.py +++ b/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() diff --git a/haproxy.cfg b/haproxy.cfg index 2f794acd..c0a959d9 100644 --- a/haproxy.cfg +++ b/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 diff --git a/helpers/ha.py b/helpers/ha.py index 0453686e..4f0e61bc 100644 --- a/helpers/ha.py +++ b/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(): diff --git a/helpers/postgresql.py b/helpers/postgresql.py index b872ab3a..12262456 100644 --- a/helpers/postgresql.py +++ b/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() diff --git a/tests/test_governor.py b/tests/test_governor.py index ad74e834..940fa968 100644 --- a/tests/test_governor.py +++ b/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