From 36d187ee1f5864bb26d5c587aa359416a5448cf4 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Mon, 9 May 2016 14:29:22 +0200 Subject: [PATCH] Remove data directory only if replica creation failed And follow the right node after replica creation (it was following the same node from which it took the backup) --- patroni/ha.py | 33 ++++++++------ patroni/postgresql.py | 95 +++++++++++++++++++--------------------- tests/test_postgresql.py | 28 ++++++------ 3 files changed, 80 insertions(+), 76 deletions(-) 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 e771bcfb..df06760f 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,26 +254,36 @@ 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): - # 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 @@ -302,10 +309,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 return ret @@ -693,41 +700,32 @@ $$""".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). + ret = self.create_replica(clone_member) == 0 + if ret: + self.delete_trigger_file() + self.restore_configuration_files() + return ret + def bootstrap(self): + """ 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() - return ret + if self.initialize() and self.start(): + self.create_replication_user() + self.create_connection_user() + return True + raise PostgresException("Could not bootstrap master PostgreSQL") def move_data_directory(self): if os.path.isdir(self.data_dir) and not self.is_running(): @@ -753,18 +751,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'