Fix for bug 1450 high memory consumption

[osm/MON.git] / osm_mon / evaluator / evaluator.py

diff --git a/osm_mon/evaluator/evaluator.py b/osm_mon/evaluator/evaluator.py
index 2f22625..6ca0dc5 100644 (file)
--- a/osm_mon/evaluator/evaluator.py
+++ b/osm_mon/evaluator/evaluator.py
@@ -22,14 +22,12 @@
 ##
 import asyncio
 import logging
-import multiprocessing
 import time
-
-import peewee
+import concurrent.futures
 
 from osm_mon.core.config import Config
-from osm_mon.core.database import Alarm
 from osm_mon.core.message_bus_client import MessageBusClient
+from osm_mon.core.models import Alarm
 from osm_mon.core.response import ResponseBuilder
 from osm_mon.evaluator.service import EvaluatorService, AlarmStatus
 
@@ -52,35 +50,74 @@ class Evaluator:
             try:
                 self.evaluate()
                 time.sleep(int(self.conf.get('evaluator', 'interval')))
-            except peewee.PeeweeException:
-                log.exception("Database error evaluating alarms: ")
-                raise
             except Exception:
                 log.exception("Error evaluating alarms")
 
     def evaluate(self):
         log.debug('evaluate')
         alarms_tuples = self.service.evaluate_alarms()
-        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)
-
-    def notify_alarm(self, alarm: Alarm, status: AlarmStatus):
-        log.debug("notify_alarm")
-        resp_message = self._build_alarm_response(alarm, status)
+        # 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):
+        log.debug("_notify_alarm")
+        resp_message = Evaluator._build_alarm_response(alarm, status)
+        msg_bus = MessageBusClient(conf)
+        loop = asyncio.get_event_loop()
         log.info("Sent alarm notification: %s", resp_message)
-        self.loop.run_until_complete(self.msg_bus.aiowrite('alarm_response', 'notify_alarm', resp_message))
+        loop.run_until_complete(msg_bus.aiowrite('alarm_response', 'notify_alarm', resp_message))
+        return
 
-    def _build_alarm_response(self, alarm: Alarm, status: AlarmStatus):
+    @staticmethod
+    def _build_alarm_response(alarm: Alarm, status: AlarmStatus):
+        log.debug("_build_alarm_response")
         response = ResponseBuilder()
         tags = {}
-        for tag in alarm.tags:
-            tags[tag.name] = tag.value
+        for name, value in alarm.tags.items():
+            tags[name] = value
         now = time.strftime("%d-%m-%Y") + " " + time.strftime("%X")
         return response.generate_response(
             'notify_alarm',