From db5999a6395a86a5765a01b9f90c2999e39b0a10 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 30 Mar 2016 11:30:05 +0200 Subject: [PATCH] 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()