From 4a4d9b85e5a919e783e7c57c72b54630a10826e7 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Fri, 22 May 2015 11:23:11 +0200 Subject: [PATCH 1/6] Make sure that there is the new line at the end of pg_hba.con --- helpers/postgresql.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index 72030e6d..f8d91fd6 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -165,7 +165,7 @@ class Postgresql: def write_pg_hba(self): with open(os.path.join(self.data_dir, 'pg_hba.conf'), 'a') as f: - f.write('host replication {username} {network} md5'.format(**self.replication)) + f.write('\nhost replication {username} {network} md5\n'.format(**self.replication)) @staticmethod def primary_conninfo(leader_url): From bb884bac0077ea27c232454396aa1128e7aa0a32 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Fri, 22 May 2015 11:45:09 +0200 Subject: [PATCH 2/6] Move is_unlocked method from Ha into Cluster class --- helpers/etcd.py | 11 +++++++---- helpers/ha.py | 5 +---- tests/test_ha.py | 19 ++++++++++++------- 3 files changed, 20 insertions(+), 15 deletions(-) diff --git a/helpers/etcd.py b/helpers/etcd.py index 4ea9ba12..9a6d4a6e 100644 --- a/helpers/etcd.py +++ b/helpers/etcd.py @@ -9,7 +9,12 @@ logger = logging.getLogger(__name__) Member = namedtuple('Member', 'hostname,address,ttl') -Cluster = namedtuple('Cluster', 'leader,last_leader_operation,members') + + +class Cluster(namedtuple('Cluster', 'leader,last_leader_operation,members')): + + def is_unlocked(self): + return not (self.leader and self.leader.hostname) class Etcd: @@ -115,9 +120,7 @@ class Etcd: def current_leader(self): try: cluster = self.get_cluster() - if not cluster.leader or not cluster.leader.address: - return None - return cluster.leader + return None if cluster.is_unlocked() else cluster.leader except: raise CurrentLeaderError("Etcd is not responding properly") diff --git a/helpers/ha.py b/helpers/ha.py index 89467bf4..b3e97d82 100644 --- a/helpers/ha.py +++ b/helpers/ha.py @@ -22,9 +22,6 @@ class Ha: def update_lock(self): return self.etcd.update_leader(self.state_handler) - def is_unlocked(self): - return not (self.cluster.leader and self.cluster.leader.hostname) - def has_lock(self): lock_owner = self.cluster.leader and self.cluster.leader.hostname logger.info('Lock owner: %s; I am %s', lock_owner, self.state_handler.name) @@ -48,7 +45,7 @@ class Ha: logging.info('started as readonly because i had the session lock') self.load_cluster_from_etcd() - if self.is_unlocked(): + if self.cluster.is_unlocked(): if self.state_handler.is_healthiest_node(self.cluster): if self.acquire_lock(): if not self.state_handler.is_leader(): diff --git a/tests/test_ha.py b/tests/test_ha.py index 741b674f..c262c9ee 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -1,7 +1,7 @@ import unittest import requests -from helpers.etcd import Etcd +from helpers.etcd import Cluster, Etcd from helpers.ha import Ha from test_etcd import requests_get, requests_put, requests_delete @@ -50,6 +50,10 @@ class MockPostgresql: return 0 +def nop(*args, **kwargs): + pass + + class TestHa(unittest.TestCase): def __init__(self, method_name='runTest'): @@ -63,59 +67,60 @@ class TestHa(unittest.TestCase): self.p = MockPostgresql() self.e = Etcd({'ttl': 30, 'host': 'remotehost', 'scope': 'test'}) self.ha = Ha(self.p, self.e) + self.ha.cluster = Cluster(None, None, []) + self.ha.load_cluster_from_etcd = nop def test_start_as_slave(self): self.p.is_healthy = false self.assertEquals(self.ha.run_cycle(), 'started as a secondary') def test_start_as_readonly(self): + self.ha.cluster.is_unlocked = false self.p.is_leader = self.p.is_healthy = false self.ha.has_lock = true self.assertEquals(self.ha.run_cycle(), 'promoted self to leader because i had the session lock') def test_acquire_lock_as_master(self): - self.ha.is_unlocked = true self.assertEquals(self.ha.run_cycle(), 'acquired session lock as a leader') def test_promoted_by_acquiring_lock(self): - self.ha.is_unlocked = true self.p.is_leader = false self.assertEquals(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock') def test_demote_after_failing_to_obtain_lock(self): - self.ha.is_unlocked = true self.ha.acquire_lock = false self.assertEquals(self.ha.run_cycle(), 'demoted self due after trying and failing to obtain lock') def test_follow_new_leader_after_failing_to_obtain_lock(self): - self.ha.is_unlocked = true self.ha.acquire_lock = false self.p.is_leader = false self.assertEquals(self.ha.run_cycle(), 'following new leader after trying and failing to obtain lock') def test_demote_because_not_healthiest(self): - self.ha.is_unlocked = true self.p.is_healthiest_node = false self.assertEquals(self.ha.run_cycle(), 'demoting self because i am not the healthiest node') def test_follow_new_leader_because_not_healthiest(self): - self.ha.is_unlocked = true self.p.is_healthiest_node = false self.p.is_leader = false self.assertEquals(self.ha.run_cycle(), 'following a different leader because i am not the healthiest node') def test_promote_because_have_lock(self): + self.ha.cluster.is_unlocked = false self.ha.has_lock = true self.p.is_leader = false self.assertEquals(self.ha.run_cycle(), 'promoted self to leader because i had the session lock') def test_leader_with_lock(self): + self.ha.cluster.is_unlocked = false self.ha.has_lock = true self.assertEquals(self.ha.run_cycle(), 'no action. i am the leader with the lock') def test_demote_because_not_having_lock(self): + self.ha.cluster.is_unlocked = false self.assertEquals(self.ha.run_cycle(), 'demoting self because i do not have the lock and i was a leader') def test_follow_the_leader(self): + self.ha.cluster.is_unlocked = false self.p.is_leader = false self.assertEquals(self.ha.run_cycle(), 'no action. i am a secondary and i am following a leader') From 096e6e4556f2260d0bcc6b7efd986137dfa59246 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Fri, 22 May 2015 11:55:20 +0200 Subject: [PATCH 3/6] Eventually member will expire in etcd --- governor.py | 1 - 1 file changed, 1 deletion(-) diff --git a/governor.py b/governor.py index ab72d261..d12c7093 100755 --- a/governor.py +++ b/governor.py @@ -84,7 +84,6 @@ def main(): governor.run() finally: governor.postgresql.stop() - governor.etcd.delete_member(governor.postgresql.name) governor.etcd.delete_leader(governor.postgresql.name) From 62d17d68de2042029952bc3145f9f7dcc6db5256 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Fri, 22 May 2015 12:00:07 +0200 Subject: [PATCH 4/6] Use "member_ttl" with default value=3600 for members --- helpers/etcd.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/helpers/etcd.py b/helpers/etcd.py index 9a6d4a6e..e808f8d4 100644 --- a/helpers/etcd.py +++ b/helpers/etcd.py @@ -21,6 +21,7 @@ class Etcd: def __init__(self, config): self.ttl = config['ttl'] + self.member_ttl = config.get('member_ttl', 3600) self.base_client_url = 'http://{host}/v2/keys/service/{scope}'.format(**config) self.postgres_cluster = None @@ -125,7 +126,7 @@ class Etcd: raise CurrentLeaderError("Etcd is not responding properly") def touch_member(self, member, connection_string): - return self.put_client_path('/members/' + member, value=connection_string, ttl=self.ttl) + return self.put_client_path('/members/' + member, value=connection_string, ttl=self.member_ttl) def take_leader(self, value): return self.put_client_path('/leader', value=value, ttl=self.ttl) From 09163ac4548008e02dcdcbb04365b0f9dace467e Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Fri, 22 May 2015 12:03:43 +0200 Subject: [PATCH 5/6] If there is no connect_address defined in yml file use address from listen as a fallback --- helpers/postgresql.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index f8d91fd6..8d32d65d 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -37,8 +37,9 @@ class Postgresql: self.config = config + connect_address = config.get('connect_address', config['listen']) or config['listen'] self.connection_string = 'postgres://{username}:{password}@{connect_address}/postgres'.format( - connect_address=self.config['connect_address'], **self.replication) + connect_address=connect_address, **self.replication) self.conn = None self.cursor_holder = None From 34f9b666cfdd2acafa032e61938a03a23a3538b8 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Fri, 22 May 2015 12:26:19 +0200 Subject: [PATCH 6/6] Demote master when etcd is not accessible --- helpers/ha.py | 3 +++ tests/test_ha.py | 9 +++++++++ 2 files changed, 12 insertions(+) diff --git a/helpers/ha.py b/helpers/ha.py index b3e97d82..74c6a892 100644 --- a/helpers/ha.py +++ b/helpers/ha.py @@ -89,6 +89,9 @@ class Ha: return 'no action. i am a secondary and i am following a leader' except EtcdError: logger.error('Error communicating with Etcd') + if self.state_handler.is_leader(): + self.state_handler.demote(None) + return 'demoted self because etcd is not accessible and i was a leader' except OperationalError: logger.error('Error communicating with Postgresql. Will try again') except HealthiestMemberError: diff --git a/tests/test_ha.py b/tests/test_ha.py index c262c9ee..cabdf7d5 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -1,6 +1,7 @@ import unittest import requests +from helpers.errors import EtcdError from helpers.etcd import Cluster, Etcd from helpers.ha import Ha from test_etcd import requests_get, requests_put, requests_delete @@ -54,6 +55,10 @@ def nop(*args, **kwargs): pass +def dead_etcd(): + raise EtcdError('Etcd is not responding properly') + + class TestHa(unittest.TestCase): def __init__(self, method_name='runTest'): @@ -124,3 +129,7 @@ class TestHa(unittest.TestCase): self.ha.cluster.is_unlocked = false self.p.is_leader = false self.assertEquals(self.ha.run_cycle(), 'no action. i am a secondary and i am following a leader') + + def test_no_etcd_connection_master_demote(self): + self.ha.load_cluster_from_etcd = dead_etcd + self.assertEquals(self.ha.run_cycle(), 'demoted self because etcd is not accessible and i was a leader')