File introduce-process_count_max-minion-configuration-par.patch of Package salt.openSUSE_Leap_42.3_Update
From dd3490c7e56a0aca17f73dde1684d469fe5582df Mon Sep 17 00:00:00 2001
From: Silvio Moioli <smoioli@suse.de>
Date: Wed, 20 Sep 2017 14:33:33 +0200
Subject: [PATCH] Introduce process_count_max minion configuration
parameter
This allows users to limit the number of processes or threads a minion
will start in response to published messages, prevents resource
exhaustion in case a high number of concurrent jobs is scheduled in a
short time.
process_count_max: add defaults and documentation
process_count_max: adapt existing unit tests
process_count_max: add unit test
process_count_max: disable by default
---
conf/minion | 6 +++++
doc/ref/configuration/minion.rst | 17 +++++++++++++
salt/config/__init__.py | 4 +++
salt/minion.py | 10 ++++++++
tests/unit/test_minion.py | 53 +++++++++++++++++++++++++++++++++++++---
5 files changed, 87 insertions(+), 3 deletions(-)
diff --git a/conf/minion b/conf/minion
index b1122c9e52..6d2b43da1b 100644
--- a/conf/minion
+++ b/conf/minion
@@ -671,6 +671,12 @@
# for a full explanation.
#multiprocessing: True
+# Limit the maximum amount of processes or threads created by salt-minion.
+# This is useful to avoid resource exhaustion in case the minion receives more
+# publications than it is able to handle, as it limits the number of spawned
+# processes or threads. -1 is the default and disables the limit.
+#process_count_max: -1
+
##### Logging settings #####
##########################################
diff --git a/doc/ref/configuration/minion.rst b/doc/ref/configuration/minion.rst
index 5cc72f1daf..b935f86656 100644
--- a/doc/ref/configuration/minion.rst
+++ b/doc/ref/configuration/minion.rst
@@ -2179,6 +2179,23 @@ executed in a thread.
multiprocessing: True
+.. conf_minion:: process_count_max
+
+``process_count_max``
+-------
+
+.. versionadded:: Oxygen
+
+Default: ``-1``
+
+Limit the maximum amount of processes or threads created by ``salt-minion``.
+This is useful to avoid resource exhaustion in case the minion receives more
+publications than it is able to handle, as it limits the number of spawned
+processes or threads. ``-1`` is the default and disables the limit.
+
+.. code-block:: yaml
+
+ process_count_max: -1
.. _minion-logging-settings:
diff --git a/salt/config/__init__.py b/salt/config/__init__.py
index b5b7f2a1f7..e843d8c79d 100644
--- a/salt/config/__init__.py
+++ b/salt/config/__init__.py
@@ -328,6 +328,9 @@ VALID_OPTS = {
# Whether or not processes should be forked when needed. The alternative is to use threading.
'multiprocessing': bool,
+ # Maximum number of concurrently active processes at any given point in time
+ 'process_count_max': int,
+
# Whether or not the salt minion should run scheduled mine updates
'mine_enabled': bool,
@@ -1193,6 +1196,7 @@ DEFAULT_MINION_OPTS = {
'auto_accept': True,
'autosign_timeout': 120,
'multiprocessing': True,
+ 'process_count_max': -1,
'mine_enabled': True,
'mine_return_job': False,
'mine_interval': 60,
diff --git a/salt/minion.py b/salt/minion.py
index 3c5046ee93..9e7301aafa 100644
--- a/salt/minion.py
+++ b/salt/minion.py
@@ -1288,6 +1288,7 @@ class Minion(MinionBase):
self._send_req_async(load, timeout, callback=lambda f: None) # pylint: disable=unexpected-keyword-arg
return True
+ @tornado.gen.coroutine
def _handle_decoded_payload(self, data):
'''
Override this method if you wish to handle the decoded data
@@ -1319,6 +1320,15 @@ class Minion(MinionBase):
self.functions, self.returners, self.function_errors, self.executors = self._load_modules()
self.schedule.functions = self.functions
self.schedule.returners = self.returners
+
+ process_count_max = self.opts.get('process_count_max')
+ if process_count_max > 0:
+ process_count = len(salt.utils.minion.running(self.opts))
+ while process_count >= process_count_max:
+ log.warn("Maximum number of processes reached while executing jid {0}, waiting...".format(data['jid']))
+ yield tornado.gen.sleep(10)
+ process_count = len(salt.utils.minion.running(self.opts))
+
# We stash an instance references to allow for the socket
# communication in Windows. You can't pickle functions, and thus
# python needs to be able to reconstruct the reference on the other
diff --git a/tests/unit/test_minion.py b/tests/unit/test_minion.py
index 535dfeedfc..6c9dca13cd 100644
--- a/tests/unit/test_minion.py
+++ b/tests/unit/test_minion.py
@@ -18,6 +18,7 @@ from salt.utils import event
from salt.exceptions import SaltSystemExit
import salt.syspaths
import tornado
+from salt.ext.six.moves import range
__opts__ = {}
@@ -69,7 +70,7 @@ class MinionTestCase(TestCase):
mock_jid_queue = [123]
try:
minion = salt.minion.Minion(mock_opts, jid_queue=copy.copy(mock_jid_queue), io_loop=tornado.ioloop.IOLoop())
- ret = minion._handle_decoded_payload(mock_data)
+ ret = minion._handle_decoded_payload(mock_data).result()
self.assertEqual(minion.jid_queue, mock_jid_queue)
self.assertIsNone(ret)
finally:
@@ -98,7 +99,7 @@ class MinionTestCase(TestCase):
# Call the _handle_decoded_payload function and update the mock_jid_queue to include the new
# mock_jid. The mock_jid should have been added to the jid_queue since the mock_jid wasn't
# previously included. The minion's jid_queue attribute and the mock_jid_queue should be equal.
- minion._handle_decoded_payload(mock_data)
+ minion._handle_decoded_payload(mock_data).result()
mock_jid_queue.append(mock_jid)
self.assertEqual(minion.jid_queue, mock_jid_queue)
finally:
@@ -126,8 +127,54 @@ class MinionTestCase(TestCase):
# Call the _handle_decoded_payload function and check that the queue is smaller by one item
# and contains the new jid
- minion._handle_decoded_payload(mock_data)
+ minion._handle_decoded_payload(mock_data).result()
self.assertEqual(len(minion.jid_queue), 2)
self.assertEqual(minion.jid_queue, [456, 789])
finally:
minion.destroy()
+
+ def test_process_count_max(self):
+ '''
+ Tests that the _handle_decoded_payload function does not spawn more than the configured amount of processes,
+ as per process_count_max.
+ '''
+ with patch('salt.minion.Minion.ctx', MagicMock(return_value={})), \
+ patch('salt.utils.process.SignalHandlingMultiprocessingProcess.start', MagicMock(return_value=True)), \
+ patch('salt.utils.process.SignalHandlingMultiprocessingProcess.join', MagicMock(return_value=True)), \
+ patch('salt.utils.minion.running', MagicMock(return_value=[])), \
+ patch('tornado.gen.sleep', MagicMock(return_value=tornado.concurrent.Future())):
+ process_count_max = 10
+ mock_opts = salt.config.DEFAULT_MINION_OPTS
+ mock_opts['minion_jid_queue_hwm'] = 100
+ mock_opts["process_count_max"] = process_count_max
+
+ try:
+ io_loop = tornado.ioloop.IOLoop()
+ minion = salt.minion.Minion(mock_opts, jid_queue=[], io_loop=io_loop)
+
+ # mock gen.sleep to throw a special Exception when called, so that we detect it
+ class SleepCalledEception(Exception):
+ """Thrown when sleep is called"""
+ pass
+ tornado.gen.sleep.return_value.set_exception(SleepCalledEception())
+
+ # up until process_count_max: gen.sleep does not get called, processes are started normally
+ for i in range(process_count_max):
+ mock_data = {'fun': 'foo.bar',
+ 'jid': i}
+ io_loop.run_sync(lambda data=mock_data: minion._handle_decoded_payload(data))
+ self.assertEqual(salt.utils.process.SignalHandlingMultiprocessingProcess.start.call_count, i + 1)
+ self.assertEqual(len(minion.jid_queue), i + 1)
+ salt.utils.minion.running.return_value += [i]
+
+ # above process_count_max: gen.sleep does get called, JIDs are created but no new processes are started
+ mock_data = {'fun': 'foo.bar',
+ 'jid': process_count_max + 1}
+
+ self.assertRaises(SleepCalledEception,
+ lambda: io_loop.run_sync(lambda: minion._handle_decoded_payload(mock_data)))
+ self.assertEqual(salt.utils.process.SignalHandlingMultiprocessingProcess.start.call_count,
+ process_count_max)
+ self.assertEqual(len(minion.jid_queue), process_count_max + 1)
+ finally:
+ minion.destroy()
--
2.14.2
