From 4d72eef164232e5d52d4301aa0b105263ac52eb0 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 31 Aug 2016 12:38:02 +0200 Subject: [PATCH] Execute API restart outside of lock Otherwise it was blocking HA loop... --- patroni/async_executor.py | 7 +------ patroni/ha.py | 20 ++++++++++---------- tests/test_ha.py | 28 ++++++++++++++-------------- 3 files changed, 25 insertions(+), 30 deletions(-) diff --git a/patroni/async_executor.py b/patroni/async_executor.py index 640ef992..04b6a5a8 100644 --- a/patroni/async_executor.py +++ b/patroni/async_executor.py @@ -7,21 +7,19 @@ logger = logging.getLogger(__name__) class AsyncExecutor(object): def __init__(self): - self._busy = False self._thread_lock = RLock() self._scheduled_action = None self._scheduled_action_lock = RLock() @property def busy(self): - return self._busy + return self.scheduled_action is not None def schedule(self, action, immediately=False): with self._scheduled_action_lock: if self._scheduled_action is not None: return self._scheduled_action self._scheduled_action = action - self._busy = immediately return None @property @@ -32,7 +30,6 @@ class AsyncExecutor(object): def reset_scheduled_action(self): with self._scheduled_action_lock: self._scheduled_action = None - self._busy = False def run(self, func, args=()): try: @@ -41,11 +38,9 @@ class AsyncExecutor(object): logger.exception('Exception during execution of long running task %s', self.scheduled_action) finally: with self: - self._busy = False self.reset_scheduled_action() def run_async(self, func, args=()): - self._busy = True Thread(target=self.run, args=(func, args)).start() def __enter__(self): diff --git a/patroni/ha.py b/patroni/ha.py index 11f02308..cc9bca0f 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -505,17 +505,17 @@ class Ha(object): return (False, "restart conditions are not satisfied") with self._async_executor: - prev = self._async_executor.schedule('restart', not run_async) + prev = self._async_executor.schedule('restart') if prev is not None: return (False, prev + ' already in progress') - if not run_async: - if self._async_executor.run(self.state_handler.restart): - return (True, 'restarted successfully') - else: - return (False, 'restart failed') - else: - self._async_executor.run_async(self.state_handler.restart) - return (True, "restart initiated") + + if run_async: + self._async_executor.run_async(self.state_handler.restart) + return (True, 'restart initiated') + elif self._async_executor.run(self.state_handler.restart): + return (True, 'restarted successfully') + else: + return (False, 'restart failed') def _do_reinitialize(self, cluster): self.state_handler.stop('immediate') @@ -539,7 +539,7 @@ class Ha(object): if action is not None: return '{0} already in progress'.format(action) - self._async_executor.run_async(self._do_reinitialize, args=(self.cluster, )) + self._async_executor.run_async(self._do_reinitialize, args=(self.cluster, )) def handle_long_action_in_progress(self): if self.has_lock(): diff --git a/tests/test_ha.py b/tests/test_ha.py index cf4a84e5..754a5788 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -4,7 +4,7 @@ import os import pytz import unittest -from mock import Mock, MagicMock, patch +from mock import Mock, MagicMock, PropertyMock, patch from patroni.config import Config from patroni.dcs import Cluster, Failover, Leader, Member, get_dcs from patroni.dcs.etcd import Client @@ -90,7 +90,7 @@ zookeeper: 'postmaster_start_time': str(postmaster_start_time)} -def run_async(func, args=()): +def run_async(self, func, args=()): return func(*args) if args else func() @@ -109,6 +109,8 @@ def run_async(func, args=()): @patch.object(etcd.Client, 'write', etcd_write) @patch.object(etcd.Client, 'read', etcd_read) @patch.object(etcd.Client, 'delete', Mock(side_effect=etcd.EtcdException)) +@patch('patroni.async_executor.AsyncExecutor.busy', PropertyMock(return_value=False)) +@patch('patroni.async_executor.AsyncExecutor.run_async', run_async) @patch('subprocess.call', Mock(return_value=0)) class TestHa(unittest.TestCase): @@ -131,7 +133,6 @@ class TestHa(unittest.TestCase): self.e = get_dcs({'etcd': {'ttl': 30, 'host': 'ok:2379', 'scope': 'test', 'name': 'foo', 'retry_timeout': 10}}) self.ha = Ha(MockPatroni(self.p, self.e)) - self.ha._async_executor.run_async = run_async self.ha.old_cluster = self.e.get_cluster() self.ha.cluster = get_cluster_not_initialized_without_leader() self.ha.load_cluster_from_dcs = Mock() @@ -278,11 +279,9 @@ class TestHa(unittest.TestCase): def test_reinitialize(self): self.assertIsNotNone(self.ha.reinitialize()) - self.assertIsNone(self.ha._async_executor.scheduled_action) self.ha.cluster = get_cluster_initialized_with_leader() self.assertIsNone(self.ha.reinitialize()) - self.assertIsNotNone(self.ha._async_executor.scheduled_action) self.assertIsNotNone(self.ha.reinitialize()) @@ -300,18 +299,19 @@ class TestHa(unittest.TestCase): self.assertEquals(self.ha.restart({'foo': 'bar'}), (False, "restart conditions are not satisfied")) def test_restart_in_progress(self): - self.ha._async_executor.schedule('restart', True) - self.assertTrue(self.ha.restart_scheduled()) - self.assertEquals(self.ha.run_cycle(), 'not healthy enough for leader race') + with patch('patroni.async_executor.AsyncExecutor.busy', PropertyMock(return_value=True)): + self.ha.restart(run_async=True) + self.assertTrue(self.ha.restart_scheduled()) + self.assertEquals(self.ha.run_cycle(), 'not healthy enough for leader race') - self.ha.cluster = get_cluster_initialized_with_leader() - self.assertEquals(self.ha.run_cycle(), 'restart in progress') + self.ha.cluster = get_cluster_initialized_with_leader() + self.assertEquals(self.ha.run_cycle(), 'restart in progress') - self.ha.has_lock = true - self.assertEquals(self.ha.run_cycle(), 'updated leader lock during restart') + self.ha.has_lock = true + self.assertEquals(self.ha.run_cycle(), 'updated leader lock during restart') - self.ha.update_lock = false - self.assertEquals(self.ha.run_cycle(), 'failed to update leader lock during restart') + self.ha.update_lock = false + self.assertEquals(self.ha.run_cycle(), 'failed to update leader lock during restart') @patch('requests.get', requests_get) @patch('time.sleep', Mock())