From cb6abac8bf7111d38ed77f7f136afa33eeaa8734 Mon Sep 17 00:00:00 2001 From: Atul Agarwal Date: Fri, 26 Mar 2021 11:14:25 +0000 Subject: [PATCH] Removed process pool from MON evaluator Change-Id: I01cd02ca542586491304f706ad28197bfa04c8d8 Signed-off-by: Atul Agarwal --- osm_mon/core/mon.yaml | 2 - osm_mon/evaluator/evaluator.py | 71 +++++-------------- .../tests/unit/evaluator/test_evaluator.py | 8 +-- 3 files changed, 20 insertions(+), 61 deletions(-) diff --git a/osm_mon/core/mon.yaml b/osm_mon/core/mon.yaml index 03b5897..d396934 100644 --- a/osm_mon/core/mon.yaml +++ b/osm_mon/core/mon.yaml @@ -47,8 +47,6 @@ collector: evaluator: interval: 30 backend: prometheus - process_pool_size: 10 - process_timeout: 50 dashboarder: interval: 30 diff --git a/osm_mon/evaluator/evaluator.py b/osm_mon/evaluator/evaluator.py index 6ca0dc5..0b4efde 100644 --- a/osm_mon/evaluator/evaluator.py +++ b/osm_mon/evaluator/evaluator.py @@ -22,8 +22,8 @@ ## import asyncio import logging +import multiprocessing import time -import concurrent.futures from osm_mon.core.config import Config from osm_mon.core.message_bus_client import MessageBusClient @@ -55,64 +55,25 @@ class Evaluator: def evaluate(self): log.debug('evaluate') + log.info('Starting alarm evaluation') alarms_tuples = self.service.evaluate_alarms() - # Starting evaluate executor pool with pool size process_pool_size. Default process_pool_size is 20 - with concurrent.futures.ProcessPoolExecutor(self.conf.get('evaluator', 'process_pool_size')) as executor: - log.info('Started evaluate process pool with pool size %s' % (self.conf.get('evaluator', - 'process_pool_size'))) - evaluate_futures = [] - for alarm, status in alarms_tuples: - evaluate_futures.append(executor.submit(Evaluator._notify_alarm, self.conf, alarm, status)) - - try: - # Wait for future calls to complete till process_timeout. Default is 50 seconds - for evaluate_future in concurrent.futures.as_completed(evaluate_futures, - self.conf.get('evaluator', 'process_timeout')): - result = evaluate_future.result(timeout=int(self.conf.get('evaluator', - 'process_timeout'))) - log.debug('result = %s' % (result)) - except concurrent.futures.TimeoutError as e: - # Some processes have not completed due to timeout error - log.info('Some processes have not finished due to TimeoutError exception') - log.debug('concurrent.futures.TimeoutError exception %s' % (e)) - - # Shutting down process pool executor - Evaluator._stop_process_pool(executor) - - @staticmethod - def _stop_process_pool(executor): - log.debug("_stop_process_pool") - log.info('Shutting down process pool') - try: - log.debug('Stopping residual processes in the process pool') - for pid, process in executor._processes.items(): - if process.is_alive(): - process.terminate() - except Exception as e: - log.info("Exception during process termination") - log.debug("Exception %s" % (e)) - - try: - # Shutting down executor - log.debug('Shutting down process pool executor') - executor.shutdown() - except RuntimeError as e: - log.info('RuntimeError in shutting down executer') - log.debug('RuntimeError %s' % (e)) - return - - @staticmethod - def _notify_alarm(conf: Config, alarm: Alarm, status: AlarmStatus): + processes = [] + for alarm, status in alarms_tuples: + p = multiprocessing.Process(target=self.notify_alarm, + args=(alarm, status)) + p.start() + processes.append(p) + for process in processes: + process.join(timeout=10) + log.info('Alarm evaluation is complete') + + def notify_alarm(self, alarm: Alarm, status: AlarmStatus): log.debug("_notify_alarm") - resp_message = Evaluator._build_alarm_response(alarm, status) - msg_bus = MessageBusClient(conf) - loop = asyncio.get_event_loop() + resp_message = self._build_alarm_response(alarm, status) log.info("Sent alarm notification: %s", resp_message) - loop.run_until_complete(msg_bus.aiowrite('alarm_response', 'notify_alarm', resp_message)) - return + self.loop.run_until_complete(self.msg_bus.aiowrite('alarm_response', 'notify_alarm', resp_message)) - @staticmethod - def _build_alarm_response(alarm: Alarm, status: AlarmStatus): + def _build_alarm_response(self, alarm: Alarm, status: AlarmStatus): log.debug("_build_alarm_response") response = ResponseBuilder() tags = {} diff --git a/osm_mon/tests/unit/evaluator/test_evaluator.py b/osm_mon/tests/unit/evaluator/test_evaluator.py index e7b9162..43cec96 100644 --- a/osm_mon/tests/unit/evaluator/test_evaluator.py +++ b/osm_mon/tests/unit/evaluator/test_evaluator.py @@ -36,10 +36,10 @@ class EvaluatorTest(TestCase): super().setUp() self.config = Config() - @mock.patch('concurrent.futures.ProcessPoolExecutor.submit') - @mock.patch.object(Evaluator, "_notify_alarm") + @mock.patch('multiprocessing.Process') + @mock.patch.object(Evaluator, "notify_alarm") @mock.patch.object(EvaluatorService, "evaluate_alarms") - def test_evaluate(self, evaluate_alarms, _notify_alarm, futures): + def test_evaluate(self, evaluate_alarms, notify_alarm, process): mock_alarm = mock.Mock() mock_alarm.operation = 'gt' mock_alarm.threshold = 50.0 @@ -48,4 +48,4 @@ class EvaluatorTest(TestCase): evaluator = Evaluator(self.config) evaluator.evaluate() - futures.assert_called_with(Evaluator._notify_alarm, self.config, mock_alarm, AlarmStatus.ALARM) + process.assert_called_with(target=notify_alarm, args=(mock_alarm, AlarmStatus.ALARM)) -- 2.17.1