gitops/patroni
1
0
Fork 0
mirror of https://github.com/outbackdingo/patroni.git synced 2026-08-25 14:53:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

Correct implementation of 'clonefrom' feature

Browse Source 
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.
This commit is contained in:
Alexander Kukushkin
2016-03-30 11:30:05 +02:00
parent 454c9dcb33
commit db5999a639
6 changed files with 42 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
features/cascading_replication.feature
+2 -2
View File
@@ -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
patroni/__init__.py
-4
View File
@@ -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:
patroni/dcs.py
+17 -4
View File
@@ -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):
patroni/ha.py
+19 -20
View File
@@ -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():
patroni/postgresql.py
+2 -3
View File
@@ -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:
tests/test_ha.py
+2 -8
View File
@@ -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:[email protected]: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()