X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=osm_mon%2Fevaluator%2Fevaluator.py;h=24e8e43f0adbda6793cdbae672ee2cfd684800fe;hb=refs%2Fchanges%2F00%2F7400%2F2;hp=8f49a66fcb3c2c83a40e2d2b013430f931a5ca32;hpb=274a6e9fa4268db2a87525a27bc574c46eaa80dc;p=osm%2FMON.git diff --git a/osm_mon/evaluator/evaluator.py b/osm_mon/evaluator/evaluator.py index 8f49a66..24e8e43 100644 --- a/osm_mon/evaluator/evaluator.py +++ b/osm_mon/evaluator/evaluator.py @@ -20,31 +20,45 @@ # For those usages not covered by the Apache License, Version 2.0 please # contact: bdiaz@whitestack.com or glavado@whitestack.com ## -import json +import asyncio import logging import multiprocessing import time +from enum import Enum +import peewee import requests from osm_common.dbbase import DbException from osm_mon.collector.backends.prometheus import OSM_METRIC_PREFIX from osm_mon.core.common_db import CommonDbClient +from osm_mon.core.config import Config from osm_mon.core.database import DatabaseManager, Alarm -from osm_mon.core.message_bus.producer import Producer +from osm_mon.core.message_bus_client import MessageBusClient from osm_mon.core.response import ResponseBuilder -from osm_mon.core.settings import Config log = logging.getLogger(__name__) +class AlarmStatus(Enum): + ALARM = 'alarm' + OK = 'ok' + INSUFFICIENT = 'insufficient-data' + + class Evaluator: - def __init__(self): - self.common_db = CommonDbClient() + + def __init__(self, config: Config, loop=None): + self.conf = config + if not loop: + loop = asyncio.get_event_loop() + self.loop = loop + self.common_db = CommonDbClient(self.conf) self.plugins = [] - self.database_manager = DatabaseManager() + self.database_manager = DatabaseManager(self.conf) self.database_manager.create_tables() self.queue = multiprocessing.Queue() + self.msg_bus = MessageBusClient(config) def _evaluate_metric(self, nsr_id: str, @@ -54,27 +68,32 @@ class Evaluator: alarm: Alarm): log.debug("_evaluate_metric") # TODO: Refactor to fit backend plugin model - cfg = Config.instance() query_section = "query={0}{{ns_id=\"{1}\",vdu_name=\"{2}\",vnf_member_index=\"{3}\"}}".format( OSM_METRIC_PREFIX + metric_name, nsr_id, vdur_name, vnf_member_index) - request_url = cfg.OSMMON_PROMETHEUS_URL + "/api/v1/query?" + query_section + request_url = self.conf.get('prometheus', 'url') + "/api/v1/query?" + query_section log.info("Querying Prometheus: %s", request_url) - r = requests.get(request_url, timeout=cfg.OSMMON_REQUEST_TIMEOUT) + r = requests.get(request_url, timeout=int(self.conf.get('global', 'request_timeout'))) if r.status_code == 200: json_response = r.json() if json_response['status'] == 'success': result = json_response['data']['result'] - if len(result): + if result: metric_value = float(result[0]['value'][1]) log.info("Metric value: %s", metric_value) if alarm.operation.upper() == 'GT': if metric_value > alarm.threshold: - self.queue.put(alarm) + self.queue.put((alarm, AlarmStatus.ALARM)) + else: + self.queue.put((alarm, AlarmStatus.OK)) elif alarm.operation.upper() == 'LT': if metric_value < alarm.threshold: - self.queue.put(alarm) + self.queue.put((alarm, AlarmStatus.ALARM)) + else: + self.queue.put((alarm, AlarmStatus.OK)) else: log.warning("No metric result for alarm %s", alarm.id) + self.queue.put((alarm, AlarmStatus.INSUFFICIENT)) + else: log.warning("Prometheus response is not success. Got status %s", json_response['status']) else: @@ -82,11 +101,13 @@ class Evaluator: def evaluate_forever(self): log.debug('evaluate_forever') - cfg = Config.instance() while True: try: self.evaluate() - time.sleep(cfg.OSMMON_EVALUATOR_INTERVAL) + 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") @@ -147,22 +168,25 @@ class Evaluator: p.start() for process in processes: - process.join() - triggered_alarms = [] + process.join(timeout=10) + alarms_tuples = [] while not self.queue.empty(): - triggered_alarms.append(self.queue.get()) - for alarm in triggered_alarms: - self.notify_alarm(alarm) + alarms_tuples.append(self.queue.get()) + for alarm, status in alarms_tuples: p = multiprocessing.Process(target=self.notify_alarm, - args=(alarm,)) + args=(alarm, status)) p.start() - def notify_alarm(self, alarm: Alarm): + def notify_alarm(self, alarm: Alarm, status: AlarmStatus): log.debug("notify_alarm") + resp_message = self._build_alarm_response(alarm, status) + log.info("Sent alarm notification: %s", resp_message) + self.loop.run_until_complete(self.msg_bus.aiowrite('alarm_response', 'notify_alarm', resp_message)) + + def _build_alarm_response(self, alarm: Alarm, status: AlarmStatus): response = ResponseBuilder() now = time.strftime("%d-%m-%Y") + " " + time.strftime("%X") - # Generate and send response - resp_message = response.generate_response( + return response.generate_response( 'notify_alarm', alarm_id=alarm.uuid, vdu_name=alarm.vdur_name, @@ -172,9 +196,5 @@ class Evaluator: operation=alarm.operation, threshold_value=alarm.threshold, sev=alarm.severity, - status='alarm', + status=status.value, date=now) - producer = Producer() - producer.send(topic='alarm_response', key='notify_alarm', value=json.dumps(resp_message)) - producer.flush() - log.info("Sent alarm notification: %s", resp_message)