diff --git a/patroni/__init__.py b/patroni/__init__.py index 36c8feac..85b6c5df 100644 --- a/patroni/__init__.py +++ b/patroni/__init__.py @@ -35,6 +35,10 @@ 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 8cda0620..2768b13e 100644 --- a/patroni/dcs.py +++ b/patroni/dcs.py @@ -146,6 +146,9 @@ 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] + class AbstractDCS(object): diff --git a/patroni/ha.py b/patroni/ha.py index 5a7e50a9..d273931a 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -65,19 +65,25 @@ class Ha(object): pass self.dcs.touch_member(json.dumps(data, separators=(',', ':'))) - def clone(self, leader): - if self.state_handler.bootstrap(cluster_initialized=True, current_leader=leader): - logger.info('bootstrapped from leader' if leader else 'bootstrapped without leader') + def clone(self, clone_member, clone_member_name="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') else: self.state_handler.stop('immediate') self.state_handler.remove_data_directory() - logger.error('failed to bootstrap from leader' if leader else 'failed to bootstrap (without leader)') + 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 - self._async_executor.schedule('bootstrap from leader') - self._async_executor.run_async(self.clone, args=(self.cluster.leader, )) - return 'trying to bootstrap from 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) elif not self.cluster.initialize and not self.patroni.nofailover: # no initialize key if self.dcs.initialize(create_new=True): # race for initialization try: @@ -96,7 +102,7 @@ class Ha(object): else: return 'failed to acquire initialize lock' else: - if self.state_handler.can_create_replica_without_leader(): + 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" return 'waiting for leader to bootstrap' diff --git a/patroni/postgresql.py b/patroni/postgresql.py index a1cedb58..d8f649c7 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -233,9 +233,10 @@ class Postgresql(object): env['PGPASSFILE'] = self.pgpass return env - def sync_replica(self, leader): - env = self.write_pgpass(parseurl(leader.conn_url)) if leader else os.environ.copy() - if self.create_replica(leader, env) == 0: + def sync_replica(self, clone_member): + # add the credentials to connect to the replica origin to pgpass. + env = self.write_pgpass(parseurl(clone_member.conn_url)) if clone_member else os.environ.copy() + if self.create_replica(clone_member, env) == 0: self.delete_trigger_file() return True return False @@ -248,33 +249,35 @@ class Postgresql(object): """ return ' '.join('{0}={1}'.format(param, val) for param, val in sorted(conn.items())) - def replica_method_can_work_without_leader(self, method): + def replica_method_can_work_without_replication_connection(self, method): return method != 'basebackup' and self.config and self.config.get(method, {}).get('no_master') - def can_create_replica_without_leader(self): + def can_create_replica_without_replication_connection(self): """ go through the replication methods to see if there are ones - that does not require a running leader to create the replica. + that does not require a working replication connection. """ replica_methods = self.config.get('create_replica_method', []) - return any(self.replica_method_can_work_without_leader(replica_method) for replica_method in replica_methods) + return any(self.replica_method_can_work_without_replication_connection(replica_method) + for replica_method in replica_methods) - def create_replica(self, leader, env): + def create_replica(self, clone_member, env): # create the replica according to the replica_method # defined by the user. this is a list, so we need to # loop through all methods the user supplies - connstring = leader.conn_url if leader else "" + connstring = clone_member.conn_url if clone_member else "" # get list of replica methods from config. # 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 leader, leave only replica methods that work without it - replica_methods = [r for r in replica_methods if self.replica_method_can_work_without_leader(r)] if not leader \ - else replica_methods + # 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 # go through them in priority order ret = 1 for replica_method in replica_methods: # if the method is basebackup, then use the built-in if replica_method == "basebackup": - ret = self.basebackup(leader, env) + ret = self.basebackup(clone_member, env) if ret == 0: logger.info("replica has been created using basebackup") # if basebackup succeeds, exit with success @@ -698,18 +701,19 @@ $$""".format(name, options), name, password, password) def last_operation(self): return str(self.xlog_position()) - def bootstrap(self, cluster_initialized=False, current_leader=None): + def bootstrap(self, cluster_initialized=False, clone_member=None): """ Populate PostgreSQL data directory by doing one of the following: - create with initdb if there is no master. - - initialize the replica from an existing master + - initialize the replica from an existing member (master or replica) - initialize the replica using the replica creation method that - works without the master (i.e. restore from on-disk base backup) + works without the replication connection (i.e. restore from on-disk + base backup) The choice between the last 2 is triggered by the initialize flag. We should never try to initdb an already initialized cluster, nor - try to bootstrap the cluster that lacks the initialize key from from - the master-less replica creation method (in the latter case, there is + try to bootstrap the cluster that lacks the initialize key using the + master-less replica creation method (in the latter case, there is no clear inidicator of the moment we should abandon our attempts and swich to initdb). @@ -719,7 +723,7 @@ $$""".format(name, options), name, password, password) that should be retried in the future. """ ret = False - if not (cluster_initialized or current_leader): + if not (cluster_initialized or clone_member): ret = self.initialize() and self.start() if ret: self.create_replication_user() @@ -727,9 +731,9 @@ $$""".format(name, options), name, password, password) else: raise PostgresException("Could not bootstrap master PostgreSQL") else: - if self.sync_replica(current_leader): + if self.sync_replica(clone_member): self.restore_configuration_files() - self.write_recovery_conf(current_leader, True) + self.write_recovery_conf(clone_member, True) ret = self.start() return ret @@ -757,12 +761,12 @@ $$""".format(name, options), name, password, password) logger.exception('Could not remove data directory %s', self.data_dir) self.move_data_directory() - def basebackup(self, leader, env): + def basebackup(self, clone_member, env): # creates a replica data dir using pg_basebackup. # this is the default, built-in create_replica_method # tries twice, then returns failure (as 1) # uses "stream" as the xlog-method to avoid sync issues - master_connection = leader.conn_url + master_connection = clone_member.conn_url maxfailures = 2 ret = 1 for bbfailures in range(0, maxfailures): diff --git a/tests/test_ha.py b/tests/test_ha.py index 98b1ad1c..2f9e5478 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -29,12 +29,12 @@ def get_cluster_not_initialized_without_leader(): def get_cluster_initialized_without_leader(leader=False, failover=None): - m = Member(0, 'leader', 28, {'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5435/postgres', - 'api_url': 'http://127.0.0.1:8008/patroni', 'xlog_location': 4}) - l = Leader(0, 0, m) if leader else None - o = 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'}) - return get_cluster(True, l, [m, o], failover) + m1 = Member(0, 'leader', 28, {'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5435/postgres', + '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'}) + return get_cluster(True, l, [m1, m2], failover) def get_cluster_initialized_with_leader(failover=None): @@ -57,6 +57,7 @@ class MockPatroni(object): self.nap_time = 10 self.replicatefrom = None self.api.connection_string = 'http://127.0.0.1:8008' + self.clonefrom = None def run_async(func, args=()): @@ -87,7 +88,7 @@ class TestHa(unittest.TestCase): 'replication': {'username': '', 'password': '', 'network': ''}}) self.p.set_state('running') self.p.check_replication_lag = true - self.p.can_create_replica_without_leader = MagicMock(return_value=False) + self.p.can_create_replica_without_replication_connection = MagicMock(return_value=False) self.e = Etcd('foo', {'ttl': 30, 'host': 'ok:2379', 'scope': 'test'}) self.e.client.read = etcd_read self.e.client.write = etcd_write @@ -205,13 +206,18 @@ class TestHa(unittest.TestCase): 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): self.ha.cluster = get_cluster_initialized_without_leader() self.assertEquals(self.ha.bootstrap(), 'waiting for leader to bootstrap') def test_bootstrap_without_leader(self): self.ha.cluster = get_cluster_initialized_without_leader() - self.p.can_create_replica_without_leader = MagicMock(return_value=True) + self.p.can_create_replica_without_replication_connection = MagicMock(return_value=True) self.assertEquals(self.ha.bootstrap(), "trying to bootstrap without leader") def test_bootstrap_initialize_lock_failed(self): diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 6924d144..fdd321df 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -471,17 +471,17 @@ class TestPostgresql(unittest.TestCase): def test_restore_configuration_files(self): self.p.restore_configuration_files() - def test_can_create_replica_without_leader(self): + def test_can_create_replica_without_replication_connection(self): self.p.config['create_replica_method'] = [] - self.assertFalse(self.p.can_create_replica_without_leader()) + self.assertFalse(self.p.can_create_replica_without_replication_connection()) self.p.config['create_replica_method'] = ['wale', 'basebackup'] self.p.config['wale'] = {'command': 'foo', 'no_master': 1} - self.assertTrue(self.p.can_create_replica_without_leader()) + self.assertTrue(self.p.can_create_replica_without_replication_connection()) - def test_replica_method_can_work_without_leader(self): - self.assertFalse(self.p.replica_method_can_work_without_leader('basebackup')) - self.assertFalse(self.p.replica_method_can_work_without_leader('foobar')) + def test_replica_method_can_work_without_replication_connection(self): + self.assertFalse(self.p.replica_method_can_work_without_replication_connection('basebackup')) + self.assertFalse(self.p.replica_method_can_work_without_replication_connection('foobar')) self.p.config['foo'] = {'command': 'bar', 'no_master': 1} - self.assertTrue(self.p.replica_method_can_work_without_leader('foo')) + self.assertTrue(self.p.replica_method_can_work_without_replication_connection('foo')) self.p.config['foo'] = {'command': 'bar'} - self.assertFalse(self.p.replica_method_can_work_without_leader('foo')) + self.assertFalse(self.p.replica_method_can_work_without_replication_connection('foo'))