Re-Revert "Revert "Migrates alarms to MongoDB"" approved by TSC
[osm/MON.git] / osm_mon / evaluator / service.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 logging
24 import multiprocessing
25 from enum import Enum
26 from typing import Tuple, List
27
28 from osm_mon.core.common_db import CommonDbClient
29 from osm_mon.core.config import Config
30 from osm_mon.core.models import Alarm
31 from osm_mon.evaluator.backends.prometheus import PrometheusBackend
32
33 log = logging.getLogger(__name__)
34
35 BACKENDS = {
36 'prometheus': PrometheusBackend
37 }
38
39
40 class AlarmStatus(Enum):
41 ALARM = 'alarm'
42 OK = 'ok'
43 INSUFFICIENT = 'insufficient-data'
44
45
46 class EvaluatorService:
47
48 def __init__(self, config: Config):
49 self.conf = config
50 self.common_db = CommonDbClient(self.conf)
51 self.queue = multiprocessing.Queue()
52
53 def _get_metric_value(self,
54 metric_name: str,
55 tags: dict):
56 return BACKENDS[self.conf.get('evaluator', 'backend')](self.conf).get_metric_value(metric_name, tags)
57
58 def _evaluate_metric(self,
59 alarm: Alarm):
60 log.debug("_evaluate_metric")
61 metric_value = self._get_metric_value(alarm.metric, alarm.tags)
62 if metric_value is None:
63 log.warning("No metric result for alarm %s", alarm.uuid)
64 self.queue.put((alarm, AlarmStatus.INSUFFICIENT))
65 else:
66 if alarm.operation.upper() == 'GT':
67 if metric_value > alarm.threshold:
68 self.queue.put((alarm, AlarmStatus.ALARM))
69 else:
70 self.queue.put((alarm, AlarmStatus.OK))
71 elif alarm.operation.upper() == 'LT':
72 if metric_value < alarm.threshold:
73 self.queue.put((alarm, AlarmStatus.ALARM))
74 else:
75 self.queue.put((alarm, AlarmStatus.OK))
76
77 def evaluate_alarms(self) -> List[Tuple[Alarm, AlarmStatus]]:
78 log.debug('evaluate_alarms')
79 processes = []
80 for alarm in self.common_db.get_alarms():
81 p = multiprocessing.Process(target=self._evaluate_metric,
82 args=(alarm,))
83 processes.append(p)
84 p.start()
85
86 for process in processes:
87 process.join(timeout=10)
88 alarms_tuples = []
89 while not self.queue.empty():
90 alarms_tuples.append(self.queue.get())
91 return alarms_tuples