Removed process pool from MON evaluator
[osm/MON.git] / osm_mon / evaluator / evaluator.py
1 # -*- coding: utf-8 -*-
2
3 # Copyright 2018 Whitestack, LLC
4 # *************************************************************
5
6 # This file is part of OSM Monitoring module
7 # All Rights Reserved to Whitestack, LLC
8
9 # Licensed under the Apache License, Version 2.0 (the "License"); you may
10 # not use this file except in compliance with the License. You may obtain
11 # a copy of the License at
12
13 # http://www.apache.org/licenses/LICENSE-2.0
14
15 # Unless required by applicable law or agreed to in writing, software
16 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
17 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
18 # License for the specific language governing permissions and limitations
19 # under the License.
20 # For those usages not covered by the Apache License, Version 2.0 please
21 # contact: bdiaz@whitestack.com or glavado@whitestack.com
22 ##
23 import asyncio
24 import logging
25 import multiprocessing
26 import time
27
28 from osm_mon.core.config import Config
29 from osm_mon.core.message_bus_client import MessageBusClient
30 from osm_mon.core.models import Alarm
31 from osm_mon.core.response import ResponseBuilder
32 from osm_mon.evaluator.service import EvaluatorService, AlarmStatus
33
34 log = logging.getLogger(__name__)
35
36
37 class Evaluator:
38
39 def __init__(self, config: Config, loop=None):
40 self.conf = config
41 if not loop:
42 loop = asyncio.get_event_loop()
43 self.loop = loop
44 self.service = EvaluatorService(config)
45 self.msg_bus = MessageBusClient(config)
46
47 def evaluate_forever(self):
48 log.debug('evaluate_forever')
49 while True:
50 try:
51 self.evaluate()
52 time.sleep(int(self.conf.get('evaluator', 'interval')))
53 except Exception:
54 log.exception("Error evaluating alarms")
55
56 def evaluate(self):
57 log.debug('evaluate')
58 log.info('Starting alarm evaluation')
59 alarms_tuples = self.service.evaluate_alarms()
60 processes = []
61 for alarm, status in alarms_tuples:
62 p = multiprocessing.Process(target=self.notify_alarm,
63 args=(alarm, status))
64 p.start()
65 processes.append(p)
66 for process in processes:
67 process.join(timeout=10)
68 log.info('Alarm evaluation is complete')
69
70 def notify_alarm(self, alarm: Alarm, status: AlarmStatus):
71 log.debug("_notify_alarm")
72 resp_message = self._build_alarm_response(alarm, status)
73 log.info("Sent alarm notification: %s", resp_message)
74 self.loop.run_until_complete(self.msg_bus.aiowrite('alarm_response', 'notify_alarm', resp_message))
75
76 def _build_alarm_response(self, alarm: Alarm, status: AlarmStatus):
77 log.debug("_build_alarm_response")
78 response = ResponseBuilder()
79 tags = {}
80 for name, value in alarm.tags.items():
81 tags[name] = value
82 now = time.strftime("%d-%m-%Y") + " " + time.strftime("%X")
83 return response.generate_response(
84 'notify_alarm',
85 alarm_id=alarm.uuid,
86 metric_name=alarm.metric,
87 operation=alarm.operation,
88 threshold_value=alarm.threshold,
89 sev=alarm.severity,
90 status=status.value,
91 date=now,
92 tags=tags)