diff --git a/patroni/ha.py b/patroni/ha.py index f41ce505..1e825d4f 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -67,11 +67,13 @@ class Ha(object): self.dcs.touch_member(json.dumps(data, separators=(',', ':'))) def clone(self, clone_member=None, msg='(without leader)'): - if self.state_handler.bootstrap(cluster_initialized=True, clone_member=clone_member): + if self.state_handler.clone(clone_member): logger.info('bootstrapped %s', msg) + cluster = self.dcs.get_cluster() + node_to_follow = self._get_node_to_follow(cluster) + self.state_handler.follow(node_to_follow, True) else: logger.error('failed to bootstrap %s', msg) - self.state_handler.stop('immediate') self.state_handler.remove_data_directory() def bootstrap(self): @@ -109,22 +111,27 @@ class Ha(object): self.recovering = True return self.follow("starting as readonly because i had the session lock", "starting as a secondary", True, True) + def _get_node_to_follow(self, cluster): + # determine the node to follow. If replicatefrom tag is set, + # try to follow the node mentioned there, otherwise, follow the leader. + if not self.patroni.replicatefrom or self.patroni.replicatefrom == self.state_handler.name: + node_to_follow = cluster.leader + else: + node_to_follow = cluster.get_member(self.patroni.replicatefrom) + + return node_to_follow if node_to_follow and node_to_follow.name != self.state_handler.name else None + def follow(self, demote_reason, follow_reason, refresh=True, recovery=False): if refresh: self.load_cluster_from_dcs() - ret = demote_reason if not recovery and self.state_handler.is_leader() else follow_reason - - # 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 = self.cluster.get_member(self.patroni.replicatefrom, fallback_to_leader=True) + if recovery: + ret = demote_reason if self.has_lock() else follow_reason else: - node_to_follow = self.cluster.leader - if node_to_follow and node_to_follow.name == self.state_handler.name: - ret = demote_reason - node_to_follow = None + ret = demote_reason if self.state_handler.is_leader() else follow_reason + + node_to_follow = self._get_node_to_follow(self.cluster) + if not self.state_handler.check_recovery_conf(node_to_follow) or recovery: self._async_executor.schedule('changing primary_conninfo and restarting') self._async_executor.run_async(self.state_handler.follow, (node_to_follow, recovery)) diff --git a/patroni/postgresql.py b/patroni/postgresql.py index 64fab95f..c162c1ec 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -126,7 +126,12 @@ class Postgresql(object): return local_address + ':' + self.port def get_postgres_role_from_data_directory(self): - return 'replica' if os.path.exists(self.recovery_conf) else 'master' + if self.data_directory_empty(): + return 'uninitialized' + elif os.path.exists(self.recovery_conf): + return 'replica' + else: + return 'master' @property def _connect_kwargs(self): @@ -233,14 +238,6 @@ class Postgresql(object): env['PGPASSFILE'] = self.pgpass return env - 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 - @staticmethod def build_connstring(conn): """ @@ -257,28 +254,39 @@ class Postgresql(object): that does not require a working replication connection. """ replica_methods = self.config.get('create_replica_method', []) - return any(self.replica_method_can_work_without_replication_connection(replica_method) - for replica_method in replica_methods) + return any(self.replica_method_can_work_without_replication_connection(method) for method in replica_methods) + + def create_replica(self, clone_member): + """ + 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 + """ - def create_replica(self, clone_member, env): self.set_state('creating replica') self._sysid = None - # 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 = 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 source, leave only replica methods that work without it - replica_methods = replica_methods if clone_member else \ - [r for r in replica_methods if self.replica_method_can_work_without_replication_connection(r)] + + if clone_member: + connstring = clone_member.conn_url + # add the credentials to connect to the replica origin to pgpass. + env = self.write_pgpass(parseurl(clone_member.conn_url)) + else: + connstring = '' + env = os.environ.copy() + # 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)] + # 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(clone_member, env) + ret = self.basebackup(connstring, env) if ret == 0: logger.info("replica has been created using basebackup") # if basebackup succeeds, exit with success @@ -304,10 +312,10 @@ class Postgresql(object): ret = subprocess.call(shlex.split(cmd) + params, env=env) # if we succeeded, stop if ret == 0: - logger.info("replica has been created using {0}".format(replica_method)) + logger.info('replica has been created using %s', replica_method) break - except Exception as e: - logger.exception('Error creating replica using method {0}: {1}'.format(replica_method, str(e))) + except Exception: + logger.exception('Error creating replica using method %s', replica_method) ret = 1 self.set_state('stopped') @@ -698,42 +706,28 @@ $$""".format(name, options), name, password, password) def last_operation(self): return str(self.xlog_position()) - def bootstrap(self, cluster_initialized=False, clone_member=None): + def clone(self, clone_member): """ - Populate PostgreSQL data directory by doing one of the following: - - create with initdb if there is no master. - initialize the replica from an existing member (master or replica) - initialize the replica using the replica creation method that 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 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). - - Failure during initdb always leads to an exception, since there is - no point in continuing if initdb fails. For the rest of the cases, - the function returns False in order to inidicate a failed attempt - that should be retried in the future. """ - ret = False - if not (cluster_initialized or clone_member): - ret = self.initialize() and self.start() - if ret: - self.create_replication_user() - self.create_connection_user() - else: - raise PostgresException("Could not bootstrap master PostgreSQL") - else: - if self.sync_replica(clone_member): - self.restore_configuration_files() - self.write_recovery_conf(clone_member) - ret = self.start() + + ret = self.create_replica(clone_member) == 0 + if ret: + self.delete_trigger_file() + self.restore_configuration_files() return ret + def bootstrap(self): + """ Initialize a new node from scratch and start it. """ + if self.initialize() and self.start(): + self.create_replication_user() + self.create_connection_user() + else: + raise PostgresException("Could not bootstrap master PostgreSQL") + def move_data_directory(self): if os.path.isdir(self.data_dir) and not self.is_running(): try: @@ -758,18 +752,17 @@ $$""".format(name, options), name, password, password) logger.exception('Could not remove data directory %s', self.data_dir) self.move_data_directory() - def basebackup(self, clone_member, env): + def basebackup(self, conn_url, 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 = clone_member.conn_url maxfailures = 2 ret = 1 for bbfailures in range(0, maxfailures): try: ret = subprocess.call(['pg_basebackup', '--pgdata=' + self.data_dir, - '--xlog-method=stream', "--dbname=" + master_connection], env=env) + '--xlog-method=stream', "--dbname=" + conn_url], env=env) if ret == 0: break diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 43d830ea..763a59db 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -239,12 +239,6 @@ class TestPostgresql(unittest.TestCase): def test_write_pgpass(self): self.p.write_pgpass({'host': 'localhost', 'port': '5432', 'user': 'foo', 'password': 'bar'}) - @patch('patroni.postgresql.Postgresql.write_pgpass', MagicMock(return_value=dict())) - def test_sync_replica(self): - self.assertTrue(self.p.sync_replica(self.leader)) - self.p.create_replica = Mock(return_value=1) - self.assertFalse(self.p.sync_replica(self.leader)) - @patch('subprocess.call', side_effect=OSError) @patch('patroni.postgresql.Postgresql.write_pgpass', MagicMock(return_value=dict())) def test_pg_rewind(self, mock_call): @@ -294,19 +288,19 @@ class TestPostgresql(unittest.TestCase): def test_create_replica(self): self.p.delete_trigger_file = Mock(side_effect=OSError) with patch('subprocess.call', Mock(side_effect=[1, 0])): - self.assertEquals(self.p.create_replica(self.leader, ''), 0) + self.assertEquals(self.p.create_replica(self.leader), 0) with patch('subprocess.call', Mock(side_effect=[Exception(), 0])): - self.assertEquals(self.p.create_replica(self.leader, ''), 0) + self.assertEquals(self.p.create_replica(self.leader), 0) self.p.config['create_replica_method'] = ['wale', 'basebackup'] self.p.config['wale'] = {'command': 'foo'} with patch('subprocess.call', Mock(return_value=0)): - self.assertEquals(self.p.create_replica(self.leader, ''), 0) + self.assertEquals(self.p.create_replica(self.leader), 0) del self.p.config['wale'] - self.assertEquals(self.p.create_replica(self.leader, ''), 0) + self.assertEquals(self.p.create_replica(self.leader), 0) with patch('subprocess.call', Mock(side_effect=Exception("foo"))): - self.assertEquals(self.p.create_replica(self.leader, ''), 1) + self.assertEquals(self.p.create_replica(self.leader), 1) def test_sync_replication_slots(self): self.p.start() @@ -372,13 +366,19 @@ class TestPostgresql(unittest.TestCase): with patch('os.rename', Mock(side_effect=OSError)): self.p.move_data_directory() - @patch('patroni.postgresql.Postgresql.write_pgpass', MagicMock(return_value=dict())) def test_bootstrap(self): with patch('subprocess.call', Mock(return_value=1)): self.assertRaises(PostgresException, self.p.bootstrap) self.p.bootstrap() - with patch('patroni.postgresql.Postgresql.sync_replica', MagicMock(return_value=True)): - self.p.bootstrap(self.leader) + + @patch('patroni.postgresql.Postgresql.create_replica', Mock(return_value=0)) + def test_clone(self): + self.p.clone(self.leader) + + @patch('os.listdir', Mock(return_value=['recovery.conf'])) + @patch('os.path.exists', Mock(return_value=True)) + def test_get_postgres_role_from_data_directory(self): + self.assertEquals(self.p.get_postgres_role_from_data_directory(), 'replica') def test_remove_data_directory(self): self.p.data_dir = 'data_dir'