mirror of
https://github.com/ansible-collections/community.general.git
synced 2024-09-14 20:13:21 +02:00
Add a function to check for killed processes in all strategies (#16684)
* Add a function to check for killed processes so that if any threads are sigkilled or sigtermed, the entire playbook execution is aborted.
This commit is contained in:
parent
ae0c1bbc47
commit
238c6461f6
3 changed files with 22 additions and 1 deletions
|
@ -316,6 +316,18 @@ class TaskQueueManager:
|
|||
def terminate(self):
|
||||
self._terminated = True
|
||||
|
||||
def has_dead_workers(self):
|
||||
|
||||
# [<WorkerProcess(WorkerProcess-2, stopped[SIGKILL])>,
|
||||
# <WorkerProcess(WorkerProcess-2, stopped[SIGTERM])>
|
||||
|
||||
defunct = False
|
||||
for idx,x in enumerate(self._workers):
|
||||
if hasattr(x[0], 'exitcode'):
|
||||
if x[0].exitcode in [-9, -15]:
|
||||
defunct = True
|
||||
return defunct
|
||||
|
||||
def send_callback(self, method_name, *args, **kwargs):
|
||||
for callback_plugin in [self._stdout_callback] + self._callback_plugins:
|
||||
# a plugin that set self.disabled to True will not be called
|
||||
|
|
|
@ -454,9 +454,14 @@ class StrategyBase:
|
|||
|
||||
display.debug("waiting for pending results...")
|
||||
while self._pending_results > 0 and not self._tqm._terminated:
|
||||
|
||||
if self._tqm.has_dead_workers():
|
||||
raise AnsibleError("A worker was found in a dead state")
|
||||
|
||||
results = self._process_pending_results(iterator)
|
||||
ret_results.extend(results)
|
||||
time.sleep(0.005)
|
||||
|
||||
display.debug("no more pending results, returning what we have")
|
||||
|
||||
return ret_results
|
||||
|
|
|
@ -172,7 +172,7 @@ class TestStrategyBase(unittest.TestCase):
|
|||
raise Queue.Empty
|
||||
else:
|
||||
return queue_items.pop()
|
||||
|
||||
|
||||
mock_queue = MagicMock()
|
||||
mock_queue.empty.side_effect = _queue_empty
|
||||
mock_queue.get.side_effect = _queue_get
|
||||
|
@ -238,6 +238,10 @@ class TestStrategyBase(unittest.TestCase):
|
|||
strategy_base._variable_manager = mock_var_mgr
|
||||
strategy_base._blocked_hosts = dict()
|
||||
|
||||
def _has_dead_workers():
|
||||
return False
|
||||
|
||||
strategy_base._tqm.has_dead_workers = _has_dead_workers
|
||||
results = strategy_base._wait_on_pending_results(iterator=mock_iterator)
|
||||
self.assertEqual(len(results), 0)
|
||||
|
||||
|
|
Loading…
Reference in a new issue