From 863aed314ba78f0938f25a2ee18d641b670d830c Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Fri, 11 Oct 2019 10:17:02 +0200 Subject: [PATCH] Fix race conditions in async actions (#1215) Specifically, there was a chance that `patronictl reinit --force` was overwritten by recover and we end up in a situation when Patroni was trying to start the postgres while basebackup still running. --- patroni/async_executor.py | 6 +++ patroni/ha.py | 80 +++++++++++++++++---------------------- tests/test_ha.py | 12 ++++-- 3 files changed, 48 insertions(+), 50 deletions(-) diff --git a/patroni/async_executor.py b/patroni/async_executor.py index fec5506a..46194fed 100644 --- a/patroni/async_executor.py +++ b/patroni/async_executor.py @@ -110,6 +110,12 @@ class AsyncExecutor(object): def run_async(self, func, args=()): Thread(target=self.run, args=(func, args)).start() + def try_run_async(self, action, func, args=()): + prev = self.schedule(action) + if prev is None: + return self.run_async(func, args) + return 'Failed to run {0}, {1} is already in progress'.format(action, prev) + def cancel(self): with self: with self._scheduled_action_lock: diff --git a/patroni/ha.py b/patroni/ha.py index 5033f56a..e2a3828a 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -230,9 +230,8 @@ class Ha(object): clone_member = self.cluster.get_clone_member(self.state_handler.name) 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) + ret = self._async_executor.try_run_async('bootstrap {0}'.format(msg), self.clone, args=(clone_member, msg)) + return ret or 'trying to bootstrap {0}'.format(msg) # no initialize key and node is allowed to be master and has 'bootstrap' section in a configuration file elif self.cluster.initialize is None and not self.patroni.nofailover and 'bootstrap' in self.patroni.config: @@ -241,16 +240,12 @@ class Ha(object): self._post_bootstrap_task = CriticalTask() if self.is_standby_cluster(): - self._async_executor.schedule('bootstrap_standby_leader') - self._async_executor.run_async(self.bootstrap_standby_leader) - return 'trying to bootstrap a new standby leader' + ret = self._async_executor.try_run_async('bootstrap_standby_leader', self.bootstrap_standby_leader) + return ret or 'trying to bootstrap a new standby leader' else: - self._async_executor.schedule('bootstrap') - self._async_executor.run_async( - self.state_handler.bootstrap.bootstrap, - args=(self.patroni.config['bootstrap'],) - ) - return 'trying to bootstrap a new cluster' + ret = self._async_executor.try_run_async('bootstrap', self.state_handler.bootstrap.bootstrap, + args=(self.patroni.config['bootstrap'],)) + return ret or 'trying to bootstrap a new cluster' else: return 'failed to acquire initialize lock' else: @@ -258,9 +253,7 @@ class Ha(object): if self.is_standby_cluster() else None if self.state_handler.can_create_replica_without_replication_connection(create_replica_methods): msg = 'bootstrap (without leader)' - self._async_executor.schedule(msg) - self._async_executor.run_async(self.clone) - return 'trying to ' + msg + return self._async_executor.try_run_async(msg, self.clone) or 'trying to ' + msg return 'waiting for {0}leader to bootstrap'.format('standby_' if self.is_standby_cluster() else '') def bootstrap_standby_leader(self): @@ -283,15 +276,13 @@ class Ha(object): return None if self._rewind.can_rewind: - self._async_executor.schedule('running pg_rewind from ' + leader.name) - self._async_executor.run_async(self._rewind.execute, (leader,)) - return True + msg = 'running pg_rewind from ' + leader.name + return self._async_executor.try_run_async(msg, self._rewind.execute, args=(leader,)) or msg # remove_data_directory_on_diverged_timelines is set if not self.is_standby_cluster(): - self._async_executor.schedule('reinitializing due to diverged timelines') - self._async_executor.run_async(self._do_reinitialize, args=(self.cluster, )) - return True + msg = 'reinitializing due to diverged timelines' + return self._async_executor.try_run_async(msg, self._do_reinitialize, args=(self.cluster,)) or msg def recover(self): # Postgres is not running and we will restart in standby mode. Watchdog is not needed until we promote. @@ -318,9 +309,8 @@ class Ha(object): and not self._crash_recovery_executed and \ (self.cluster.is_unlocked() or self._rewind.can_rewind): self._crash_recovery_executed = True - self._async_executor.schedule('doing crash recovery in a single user mode') - self._async_executor.run_async(self.state_handler.fix_cluster_state) - return self._async_executor.scheduled_action + msg = 'doing crash recovery in a single user mode' + return self._async_executor.try_run_async(msg, self.state_handler.fix_cluster_state) or msg self.load_cluster_from_dcs() @@ -328,8 +318,9 @@ class Ha(object): if self.is_standby_cluster() or not self.has_lock(): if not self._rewind.executed: self._rewind.trigger_check_diverged_lsn() - if self._handle_rewind_or_reinitialize(): - return self._async_executor.scheduled_action + msg = self._handle_rewind_or_reinitialize() + if msg: + return msg if self.has_lock(): # in standby cluster msg = "starting as a standby leader because i had the session lock" @@ -345,10 +336,9 @@ class Ha(object): msg = "starting as readonly because i had the session lock" node_to_follow = None - self.recovering = True - - self._async_executor.schedule('restarting after failure') - self._async_executor.run_async(self.state_handler.follow, (node_to_follow, role, timeout)) + if self._async_executor.try_run_async('restarting after failure', self.state_handler.follow, + args=(node_to_follow, role, timeout)) is None: + self.recovering = True return msg def _get_node_to_follow(self, cluster): @@ -383,8 +373,9 @@ class Ha(object): self.demote('immediate-nolock') return demote_reason - if self._handle_rewind_or_reinitialize(): - return self._async_executor.scheduled_action + msg = self._handle_rewind_or_reinitialize() + if msg: + return msg role = 'standby_leader' if isinstance(node_to_follow, RemoteMember) and self.has_lock(False) else 'replica' # It might happen that leader key in the standby cluster references non-exiting member. @@ -392,8 +383,8 @@ class Ha(object): if self.is_standby_cluster() and role == 'replica' and not (node_to_follow and node_to_follow.conn_url): return 'continue following the old known standby leader' elif not self.state_handler.config.check_recovery_conf(node_to_follow): - self._async_executor.schedule('changing primary_conninfo and restarting') - self._async_executor.run_async(self.state_handler.follow, (node_to_follow, role)) + self._async_executor.try_run_async('changing primary_conninfo and restarting', + self.state_handler.follow, args=(node_to_follow, role)) elif role == 'standby_leader' and self.state_handler.role != role: self.state_handler.set_role(role) self.state_handler.call_nowait(ACTION_ON_ROLE_CHANGE) @@ -552,9 +543,9 @@ class Ha(object): self._rewind.reset_state() logger.info("cleared rewind state after becoming the leader") - self._async_executor.schedule('promote') - self._async_executor.run_async(self.state_handler.promote, - args=(self.dcs.loop_wait, on_success, self._leader_access_is_restricted)) + self._async_executor.try_run_async('promote', self.state_handler.promote, + args=(self.dcs.loop_wait, on_success, + self._leader_access_is_restricted)) return promote_message def fetch_node_status(self, member): @@ -772,8 +763,7 @@ class Ha(object): # there could be an async action already running, calling follow from here will lead # to racy state handler state updates. if mode_control['async_req']: - self._async_executor.schedule('starting after demotion') - self._async_executor.run_async(self.state_handler.follow, (node_to_follow,)) + self._async_executor.try_run_async('starting after demotion', self.state_handler.follow, (node_to_follow,)) else: if self.is_synchronous_mode(): self.state_handler.config.set_synchronous_standby(None) @@ -846,9 +836,8 @@ class Ha(object): members = [m for m in self.cluster.members if not failover.candidate or m.name == failover.candidate] if self.is_failover_possible(members): # check that there are healthy members - self._async_executor.schedule('manual failover: demote') - self._async_executor.run_async(self.demote, ('graceful',)) - return 'manual failover: demoting myself' + ret = self._async_executor.try_run_async('manual failover: demote', self.demote, ('graceful',)) + return ret or 'manual failover: demoting myself' else: logger.warning('manual failover: no healthy members found, failover is not possible') else: @@ -1151,10 +1140,9 @@ class Ha(object): return 'waiting for end of recovery after bootstrap' self.state_handler.set_role('master') - self._async_executor.schedule('post_bootstrap') - self._async_executor.run_async(self.state_handler.bootstrap.post_bootstrap, - args=(self.patroni.config['bootstrap'], self._post_bootstrap_task)) - return 'running post_bootstrap' + ret = self._async_executor.try_run_async('post_bootstrap', self.state_handler.bootstrap.post_bootstrap, + args=(self.patroni.config['bootstrap'], self._post_bootstrap_task)) + return ret or 'running post_bootstrap' self.state_handler.bootstrapping = False self.dcs.set_config_value(json.dumps(self.patroni.config.dynamic_configuration, separators=(',', ':'))) diff --git a/tests/test_ha.py b/tests/test_ha.py index 4586f6e1..599dc24a 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -141,7 +141,11 @@ zookeeper: def run_async(self, func, args=()): - return func(*args) if args else func() + self.reset_scheduled_action() + if args: + func(*args) + else: + func() @patch.object(Postgresql, 'is_running', Mock(return_value=MockPostmaster())) @@ -426,7 +430,7 @@ class TestHa(PostgresInit): self.ha.cluster = get_cluster_initialized_with_leader() self.assertIsNone(self.ha.reinitialize(True)) - + self.ha._async_executor.schedule('reinitialize') self.assertIsNotNone(self.ha.reinitialize()) self.ha.state_handler.name = self.ha.cluster.leader.name @@ -440,7 +444,7 @@ class TestHa(PostgresInit): self.p.restart = false self.assertEqual(self.ha.restart({}), (False, 'restart failed')) self.ha.cluster = get_cluster_initialized_with_leader() - self.ha.reinitialize() + self.ha._async_executor.schedule('reinitialize') self.assertEqual(self.ha.restart({}), (False, 'reinitialize already in progress')) with patch.object(self.ha, "restart_matches", return_value=False): self.assertEqual(self.ha.restart({'foo': 'bar'}), (False, "restart conditions are not satisfied")) @@ -448,7 +452,7 @@ class TestHa(PostgresInit): @patch('os.kill', Mock()) def test_restart_in_progress(self): with patch('patroni.async_executor.AsyncExecutor.busy', PropertyMock(return_value=True)): - self.ha.restart({}, run_async=True) + self.ha._async_executor.schedule('restart') self.assertTrue(self.ha.restart_scheduled()) self.assertEqual(self.ha.run_cycle(), 'restart in progress')