From db5999a6395a86a5765a01b9f90c2999e39b0a10 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 30 Mar 2016 11:30:05 +0200 Subject: [PATCH 1/4] Correct implementation of 'clonefrom' feature According to https://github.com/zalando/patroni/issues/48 'clonefrom' tag should be boolean and it should be used to mark node as a suitable for creation of a new replica from. If there are more then one such node in the cluster (with tag clonefrom=true), one of them will be chosed randomly. --- features/cascading_replication.feature | 4 +-- patroni/__init__.py | 4 --- patroni/dcs.py | 21 +++++++++++--- patroni/ha.py | 39 +++++++++++++------------- patroni/postgresql.py | 5 ++-- tests/test_ha.py | 10 ++----- 6 files changed, 42 insertions(+), 41 deletions(-) diff --git a/features/cascading_replication.feature b/features/cascading_replication.feature index 8a3d2f80..44b23386 100644 --- a/features/cascading_replication.feature +++ b/features/cascading_replication.feature @@ -4,10 +4,10 @@ Feature: cascading replication Scenario: check a base backup from the replica Given I start postgres0 And postgres0 is a leader after 10 seconds - And I start postgres1 + And I configure and start postgres1 with a tag clonefrom true And replication works from postgres0 to postgres1 after 15 seconds And I create label with "postgres0" in postgres0 data directory And I create label with "postgres1" in postgres1 data directory - And I configure and start postgres2 with a tag clonefrom postgres1 + And I configure and start postgres2 with a tag replicatefrom postgres1 Then replication works from postgres0 to postgres2 after 30 seconds And there is a label with "postgres1" in postgres2 data directory diff --git a/patroni/__init__.py b/patroni/__init__.py index 85b6c5df..36c8feac 100644 --- a/patroni/__init__.py +++ b/patroni/__init__.py @@ -35,10 +35,6 @@ class Patroni(object): def replicatefrom(self): return self.tags.get('replicatefrom') - @property - def clonefrom(self): - return self.tags.get('clonefrom') - @staticmethod def get_dcs(name, config): if 'etcd' in config: diff --git a/patroni/dcs.py b/patroni/dcs.py index 3cb76cdb..36b4c4ea 100644 --- a/patroni/dcs.py +++ b/patroni/dcs.py @@ -3,6 +3,7 @@ import json import dateutil from collections import namedtuple +from random import randint from six.moves.urllib_parse import urlparse, urlunparse, parse_qsl from threading import Event, Lock @@ -63,13 +64,21 @@ class Member(namedtuple('Member', 'index,name,session,data')): def api_url(self): return self.data.get('api_url') + @property + def tags(self): + return self.data.get('tags', {}) + @property def nofailover(self): - return self.data.get('tags', {}).get('nofailover', False) + return self.tags.get('nofailover', False) @property def replicatefrom(self): - return self.data.get('tags', {}).get('replicatefrom') + return self.tags.get('replicatefrom') + + @property + def clonefrom(self): + return self.tags.get('clonefrom', False) class Leader(namedtuple('Leader', 'index,session,member')): @@ -146,8 +155,12 @@ class Cluster(namedtuple('Cluster', 'initialize,leader,last_leader_operation,mem def has_member(self, member_name): return any(m for m in self.members if m.name == member_name) - def get_member(self, member_name): - return ([m for m in self.members if m.name == member_name] or [None])[0] + def get_member(self, member_name, fallback_to_leader=True): + return ([m for m in self.members if m.name == member_name] or [self.leader if fallback_to_leader else None])[0] + + def get_clone_member(self): + candidates = [m for m in self.members if m.clonefrom and (not self.leader or m.name != self.leader.name)] + return candidates[randint(0, len(candidates) - 1)] if candidates else self.leader class AbstractDCS(object): diff --git a/patroni/ha.py b/patroni/ha.py index d9916da6..f62e37af 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -65,25 +65,22 @@ class Ha(object): pass self.dcs.touch_member(json.dumps(data, separators=(',', ':'))) - def clone(self, clone_member, clone_member_name="leader"): + def clone(self, clone_member=None, msg='(without leader)'): if self.state_handler.bootstrap(cluster_initialized=True, clone_member=clone_member): - logger.info('bootstrapped from {0}'.format(clone_member_name) - if clone_member else 'bootstrapped without leader') + logger.info('bootstrapped %s', msg) else: + logger.error('failed to bootstrap %s', msg) self.state_handler.stop('immediate') self.state_handler.remove_data_directory() - logger.error('failed to bootstrap from {0}'.format(clone_member_name) - if clone_member else 'failed to bootstrap (without leader)') def bootstrap(self): if not self.cluster.is_unlocked(): # cluster already has leader - clonefrom = self.patroni.clonefrom - clone_member = self.cluster.get_member(clonefrom)\ - if self.cluster.has_member(clonefrom) else self.cluster.leader - clone_member_name = 'leader' if clone_member == self.cluster.leader else 'replica \'{0}\''.format(clonefrom) - self._async_executor.schedule('bootstrap from {0}'.format(clone_member_name)) - self._async_executor.run_async(self.clone, args=(clone_member, clone_member_name)) - return 'trying to bootstrap from {0}'.format(clone_member_name) + clone_member = self.cluster.get_clone_member() + member_role = 'leader' if clone_member == self.cluster.leader else 'replica' + msg = "from {0} '{1}'".format(member_role, clone_member.name) + self._async_executor.schedule('bootstrap {0}'.format(msg)) + self._async_executor.run_async(self.clone, args=(clone_member, msg)) + return 'trying to bootstrap {0}'.format(msg) elif not self.cluster.initialize and not self.patroni.nofailover: # no initialize key if self.dcs.initialize(create_new=True): # race for initialization try: @@ -103,8 +100,8 @@ class Ha(object): return 'failed to acquire initialize lock' else: if self.state_handler.can_create_replica_without_replication_connection(): - self._async_executor.run_async(self.clone, args=(None, )) - return "trying to bootstrap without leader" + self._async_executor.run_async(self.clone) + return "trying to bootstrap (without leader)" return 'waiting for leader to bootstrap' def recover(self): @@ -131,8 +128,7 @@ class Ha(object): # determine the node to follow. If replicatefrom tag is set, # try to follow the node mentioned there, otherwise, follow the leader. if self.patroni.replicatefrom: - node_to_follow = [m for m in self.cluster.members if m.name == self.patroni.replicatefrom] - node_to_follow = node_to_follow[0] if node_to_follow else self.cluster.leader + node_to_follow = self.cluster.get_member(self.patroni.replicatefrom, fallback_to_leader=True) else: node_to_follow = self.cluster.leader node_to_follow = None if node_to_follow and node_to_follow.name == self.state_handler.name else node_to_follow @@ -227,9 +223,9 @@ class Ha(object): return True # find specific node and check that it is healthy - members = [m for m in self.cluster.members if m.name == failover.candidate] - if members: - member, reachable, _, _, tags = self.fetch_node_status(members[0]) + member = self.cluster.get_member(failover.candidate, fallback_to_leader=False) + if member: + member, reachable, _, _, tags = self.fetch_node_status(member) if reachable and not tags.get('nofailover', False): # node is healthy logger.info('manual failover: to %s, i am %s', member.name, self.state_handler.name) return False @@ -389,7 +385,10 @@ class Ha(object): def reinitialize(self, cluster): self.state_handler.stop('immediate') self.state_handler.remove_data_directory() - self.clone(cluster.leader) + + clone_member = cluster.get_clone_member() + member_role = 'leader' if clone_member == cluster.leader else 'replica' + self.clone(clone_member, "from {0} '{1}'".format(member_role, clone_member.name)) def process_scheduled_action(self): if self.reinitialize_scheduled(): diff --git a/patroni/postgresql.py b/patroni/postgresql.py index df17200d..50859d17 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -269,9 +269,8 @@ class Postgresql(object): # If there is no configuration key, or no value is specified, use basebackup replica_methods = self.config.get('create_replica_method') or ['basebackup'] # if we don't have any source, leave only replica methods that work without it - replica_methods = \ - [r for r in replica_methods if self.replica_method_can_work_without_replication_connection(r)]\ - if not clone_member else replica_methods + replica_methods = replica_methods if clone_member else \ + [r for r in replica_methods if self.replica_method_can_work_without_replication_connection(r)] # go through them in priority order ret = 1 for replica_method in replica_methods: diff --git a/tests/test_ha.py b/tests/test_ha.py index 3ab7442b..c38c4cc1 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -33,7 +33,7 @@ def get_cluster_initialized_without_leader(leader=False, failover=None): 'api_url': 'http://127.0.0.1:8008/patroni', 'xlog_location': 4}) l = Leader(0, 0, m1) if leader else None m2 = Member(0, 'other', 28, {'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5436/postgres', - 'api_url': 'http://127.0.0.1:8011/patroni'}) + 'api_url': 'http://127.0.0.1:8011/patroni', 'tags': {'clonefrom': True}}) return get_cluster(True, l, [m1, m2], failover) @@ -202,14 +202,8 @@ class TestHa(unittest.TestCase): self.ha.load_cluster_from_dcs = Mock(side_effect=DCSError('Etcd is not responding properly')) self.assertEquals(self.ha.run_cycle(), 'demoted self because DCS is not accessible and i was a leader') - def test_bootstrap_from_leader(self): - self.ha.cluster = get_cluster_initialized_with_leader() - self.p.bootstrap = false - self.assertEquals(self.ha.bootstrap(), 'trying to bootstrap from leader') - def test_bootstrap_from_another_member(self): self.ha.cluster = get_cluster_initialized_with_leader() - self.ha.patroni.clonefrom = 'other' self.assertEquals(self.ha.bootstrap(), 'trying to bootstrap from replica \'other\'') def test_bootstrap_waiting_for_leader(self): @@ -219,7 +213,7 @@ class TestHa(unittest.TestCase): def test_bootstrap_without_leader(self): self.ha.cluster = get_cluster_initialized_without_leader() self.p.can_create_replica_without_replication_connection = MagicMock(return_value=True) - self.assertEquals(self.ha.bootstrap(), "trying to bootstrap without leader") + self.assertEquals(self.ha.bootstrap(), 'trying to bootstrap (without leader)') def test_bootstrap_initialize_lock_failed(self): self.ha.cluster = get_cluster_not_initialized_without_leader() From 24cc93ac70cc29aa641afe09172faa696c8aa9ff Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 30 Mar 2016 15:00:25 +0200 Subject: [PATCH 2/4] Don't expose known boolean tags with value not True --- patroni/__init__.py | 3 ++- patroni/api.py | 7 ++----- 2 files changed, 4 insertions(+), 6 deletions(-) diff --git a/patroni/__init__.py b/patroni/__init__.py index 36c8feac..21fdbea3 100644 --- a/patroni/__init__.py +++ b/patroni/__init__.py @@ -19,7 +19,8 @@ class Patroni(object): def __init__(self, config): self.nap_time = config['loop_wait'] - self.tags = config.get('tags', dict()) + self.tags = {tag: value for tag, value in config.get('tags', {}).items() + if tag not in ('clonefrom', 'nofailover', 'noloadbalance') or value} self.postgresql = Postgresql(config['postgresql']) self.dcs = self.get_dcs(self.postgresql.name, config) self.version = __version__ diff --git a/patroni/api.py b/patroni/api.py index cb7b2d1f..3f51059b 100644 --- a/patroni/api.py +++ b/patroni/api.py @@ -63,7 +63,7 @@ class RestApiHandler(BaseHTTPRequestHandler): path = '/master' if self.path == '/' else self.path response = self.get_postgresql_status() - response.update(self.get_tags()) + response['tags'] = self.server.patroni.tags patroni = self.server.patroni cluster = patroni.dcs.cluster @@ -94,7 +94,7 @@ class RestApiHandler(BaseHTTPRequestHandler): def do_GET_patroni(self): response = self.get_postgresql_status(True) - response.update(self.get_tags()) + response['tags'] = self.server.patroni.tags response['patroni'] = {'version': self.server.patroni.version, 'scope': self.server.patroni.postgresql.scope} self.send_response(200) @@ -289,9 +289,6 @@ class RestApiHandler(BaseHTTPRequestHandler): state = 'unknown' return {'state': state} - def get_tags(self): - return {'tags': self.server.patroni.tags} - def log_message(self, fmt, *args): logger.debug("API thread: %s - - [%s] %s", self.client_address[0], self.log_date_time_string(), fmt % args) From 7c683336fbdae3f3c57e739e685707981205b94b Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 30 Mar 2016 15:16:57 +0200 Subject: [PATCH 3/4] Don't expose empty `tags` dict in API and DCS --- patroni/api.py | 7 +++++-- patroni/ha.py | 5 +++-- tests/test_ha.py | 2 +- 3 files changed, 9 insertions(+), 5 deletions(-) diff --git a/patroni/api.py b/patroni/api.py index 3f51059b..2ae66e40 100644 --- a/patroni/api.py +++ b/patroni/api.py @@ -63,7 +63,7 @@ class RestApiHandler(BaseHTTPRequestHandler): path = '/master' if self.path == '/' else self.path response = self.get_postgresql_status() - response['tags'] = self.server.patroni.tags + response.update(self.get_tags()) patroni = self.server.patroni cluster = patroni.dcs.cluster @@ -94,7 +94,7 @@ class RestApiHandler(BaseHTTPRequestHandler): def do_GET_patroni(self): response = self.get_postgresql_status(True) - response['tags'] = self.server.patroni.tags + response.update(self.get_tags()) response['patroni'] = {'version': self.server.patroni.version, 'scope': self.server.patroni.postgresql.scope} self.send_response(200) @@ -289,6 +289,9 @@ class RestApiHandler(BaseHTTPRequestHandler): state = 'unknown' return {'state': state} + def get_tags(self): + return {'tags': self.server.patroni.tags} if self.server.patroni.tags else {} + def log_message(self, fmt, *args): logger.debug("API thread: %s - - [%s] %s", self.client_address[0], self.log_date_time_string(), fmt % args) diff --git a/patroni/ha.py b/patroni/ha.py index f62e37af..e25a76df 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -55,9 +55,10 @@ class Ha(object): 'conn_url': self.state_handler.connection_string, 'api_url': self.patroni.api.connection_string, 'state': self.state_handler.state, - 'role': self.state_handler.role, - 'tags': self.patroni.tags + 'role': self.state_handler.role } + if self.patroni.tags: + data['tags'] = self.patroni.tags if data['state'] in ['running', 'restarting', 'starting']: try: data['xlog_location'] = self.state_handler.xlog_position() diff --git a/tests/test_ha.py b/tests/test_ha.py index c38c4cc1..02c97f5a 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -52,7 +52,7 @@ class MockPatroni(object): self.postgresql = p self.dcs = d self.api = Mock() - self.tags = {} + self.tags = {'foo': 'bar'} self.nofailover = None self.nap_time = 10 self.replicatefrom = None From ada50e418c13f4e69f7e7066c352b4a60addac83 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Thu, 31 Mar 2016 17:13:29 +0200 Subject: [PATCH 4/4] Update scenario description --- features/cascading_replication.feature | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/features/cascading_replication.feature b/features/cascading_replication.feature index 44b23386..27f72eba 100644 --- a/features/cascading_replication.feature +++ b/features/cascading_replication.feature @@ -1,7 +1,7 @@ Feature: cascading replication We should check that patroni can do base backup and streaming from the replica -Scenario: check a base backup from the replica +Scenario: check a base backup and streaming replication from a replica Given I start postgres0 And postgres0 is a leader after 10 seconds And I configure and start postgres1 with a tag clonefrom true