gitops/patroni
1
0
Fork 0
mirror of https://github.com/outbackdingo/patroni.git synced 2026-08-25 14:53:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

Execute API restart outside of lock

Browse Source 
Otherwise it was blocking HA loop...
This commit is contained in:
Alexander Kukushkin
2016-08-31 12:38:02 +02:00
parent c0fae1b2e9
commit 4d72eef164
3 changed files with 25 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
patroni/async_executor.py
+1 -6
View File
@@ -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):
patroni/ha.py
+10 -10
View File
@@ -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():
tests/test_ha.py
+14 -14
View File
@@ -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())