Refactors code and adds unit tests
Change-Id: Ia14034679af8fc7e9ac91a1bfb766e4b99ec9e39
Signed-off-by: Benjamin Diaz <bdiaz@whitestack.com>
diff --git a/osm_mon/evaluator/backends/__init__.py b/osm_mon/evaluator/backends/__init__.py
new file mode 100644
index 0000000..4450364
--- /dev/null
+++ b/osm_mon/evaluator/backends/__init__.py
@@ -0,0 +1,22 @@
+# -*- coding: utf-8 -*-
+
+# Copyright 2018 Whitestack, LLC
+# *************************************************************
+
+# This file is part of OSM Monitoring module
+# All Rights Reserved to Whitestack, LLC
+
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+
+# http://www.apache.org/licenses/LICENSE-2.0
+
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+# For those usages not covered by the Apache License, Version 2.0 please
+# contact: bdiaz@whitestack.com or glavado@whitestack.com
+##
diff --git a/osm_mon/evaluator/backends/base.py b/osm_mon/evaluator/backends/base.py
new file mode 100644
index 0000000..0e9fc0d
--- /dev/null
+++ b/osm_mon/evaluator/backends/base.py
@@ -0,0 +1,30 @@
+# Copyright 2018 Whitestack, LLC
+# *************************************************************
+
+# This file is part of OSM Monitoring module
+# All Rights Reserved to Whitestack, LLC
+
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+
+# http://www.apache.org/licenses/LICENSE-2.0
+
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+
+# For those usages not covered by the Apache License, Version 2.0 please
+# contact: bdiaz@whitestack.com or glavado@whitestack.com
+##
+from osm_mon.core.config import Config
+
+
+class BaseBackend:
+ def __init__(self, config: Config):
+ pass
+
+ def get_metric_value(self, metric_name, nsr_id, vdur_name, vnf_member_index):
+ pass
diff --git a/osm_mon/evaluator/backends/prometheus.py b/osm_mon/evaluator/backends/prometheus.py
new file mode 100644
index 0000000..9ff50d6
--- /dev/null
+++ b/osm_mon/evaluator/backends/prometheus.py
@@ -0,0 +1,61 @@
+# -*- coding: utf-8 -*-
+
+# Copyright 2018 Whitestack, LLC
+# *************************************************************
+
+# This file is part of OSM Monitoring module
+# All Rights Reserved to Whitestack, LLC
+
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+
+# http://www.apache.org/licenses/LICENSE-2.0
+
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+# For those usages not covered by the Apache License, Version 2.0 please
+# contact: bdiaz@whitestack.com or glavado@whitestack.com
+##
+import logging
+
+import requests
+
+from osm_mon.core.config import Config
+from osm_mon.evaluator.backends.base import BaseBackend
+
+log = logging.getLogger(__name__)
+
+OSM_METRIC_PREFIX = 'osm_'
+
+
+class PrometheusBackend(BaseBackend):
+
+ def __init__(self, config: Config):
+ super().__init__(config)
+ self.conf = config
+
+ def get_metric_value(self, metric_name, nsr_id, vdur_name, vnf_member_index):
+ 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 = self.conf.get('prometheus', 'url') + "/api/v1/query?" + query_section
+ log.info("Querying Prometheus: %s", request_url)
+ 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):
+ metric_value = float(result[0]['value'][1])
+ log.info("Metric value: %s", metric_value)
+ return metric_value
+ else:
+ return None
+ else:
+ log.warning("Prometheus response is not success. Got status %s", json_response['status'])
+ else:
+ log.warning("Error contacting Prometheus. Got status code %s: %s", r.status_code, r.text)
+ return None
diff --git a/osm_mon/evaluator/evaluator.py b/osm_mon/evaluator/evaluator.py
index 24e8e43..d3fdfd5 100644
--- a/osm_mon/evaluator/evaluator.py
+++ b/osm_mon/evaluator/evaluator.py
@@ -24,28 +24,18 @@
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.database import Alarm
from osm_mon.core.message_bus_client import MessageBusClient
from osm_mon.core.response import ResponseBuilder
+from osm_mon.evaluator.service import EvaluatorService, AlarmStatus
log = logging.getLogger(__name__)
-class AlarmStatus(Enum):
- ALARM = 'alarm'
- OK = 'ok'
- INSUFFICIENT = 'insufficient-data'
-
-
class Evaluator:
def __init__(self, config: Config, loop=None):
@@ -53,52 +43,9 @@
if not loop:
loop = asyncio.get_event_loop()
self.loop = loop
- self.common_db = CommonDbClient(self.conf)
- self.plugins = []
- self.database_manager = DatabaseManager(self.conf)
- self.database_manager.create_tables()
- self.queue = multiprocessing.Queue()
+ self.service = EvaluatorService(config)
self.msg_bus = MessageBusClient(config)
- def _evaluate_metric(self,
- nsr_id: str,
- vnf_member_index: int,
- vdur_name: str,
- metric_name: str,
- alarm: Alarm):
- log.debug("_evaluate_metric")
- # TODO: Refactor to fit backend plugin model
- 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 = self.conf.get('prometheus', 'url') + "/api/v1/query?" + query_section
- log.info("Querying Prometheus: %s", request_url)
- 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 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, AlarmStatus.ALARM))
- else:
- self.queue.put((alarm, AlarmStatus.OK))
- elif alarm.operation.upper() == 'LT':
- if metric_value < alarm.threshold:
- 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:
- log.warning("Error contacting Prometheus. Got status code %s: %s", r.status_code, r.text)
-
def evaluate_forever(self):
log.debug('evaluate_forever')
while True:
@@ -113,65 +60,7 @@
def evaluate(self):
log.debug('evaluate')
- processes = []
- for alarm in Alarm.select():
- try:
- vnfr = self.common_db.get_vnfr(alarm.nsr_id, alarm.vnf_member_index)
- except DbException:
- log.exception("Error getting vnfr: ")
- continue
- vnfd = self.common_db.get_vnfd(vnfr['vnfd-id'])
- try:
- vdur = next(filter(lambda vdur: vdur['name'] == alarm.vdur_name, vnfr['vdur']))
- except StopIteration:
- log.warning("No vdur found with name %s for alarm %s", alarm.vdur_name, alarm.id)
- continue
- vdu = next(filter(lambda vdu: vdu['id'] == vdur['vdu-id-ref'], vnfd['vdu']))
- vnf_monitoring_param = next(
- filter(lambda param: param['id'] == alarm.monitoring_param, vnfd['monitoring-param']))
- nsr_id = vnfr['nsr-id-ref']
- vnf_member_index = vnfr['member-vnf-index-ref']
- vdur_name = vdur['name']
- if 'vdu-monitoring-param' in vnf_monitoring_param:
- vdu_monitoring_param = next(filter(
- lambda param: param['id'] == vnf_monitoring_param['vdu-monitoring-param'][
- 'vdu-monitoring-param-ref'], vdu['monitoring-param']))
- nfvi_metric = vdu_monitoring_param['nfvi-metric']
-
- p = multiprocessing.Process(target=self._evaluate_metric,
- args=(nsr_id,
- vnf_member_index,
- vdur_name,
- nfvi_metric,
- alarm))
- processes.append(p)
- p.start()
- if 'vdu-metric' in vnf_monitoring_param:
- vnf_metric_name = vnf_monitoring_param['vdu-metric']['vdu-metric-name-ref']
- p = multiprocessing.Process(target=self._evaluate_metric,
- args=(nsr_id,
- vnf_member_index,
- vdur_name,
- vnf_metric_name,
- alarm))
- processes.append(p)
- p.start()
- if 'vnf-metric' in vnf_monitoring_param:
- vnf_metric_name = vnf_monitoring_param['vnf-metric']['vnf-metric-name-ref']
- p = multiprocessing.Process(target=self._evaluate_metric,
- args=(nsr_id,
- vnf_member_index,
- '',
- vnf_metric_name,
- alarm))
- processes.append(p)
- p.start()
-
- for process in processes:
- process.join(timeout=10)
- alarms_tuples = []
- while not self.queue.empty():
- alarms_tuples.append(self.queue.get())
+ alarms_tuples = self.service.evaluate_alarms()
for alarm, status in alarms_tuples:
p = multiprocessing.Process(target=self.notify_alarm,
args=(alarm, status))
diff --git a/osm_mon/evaluator/service.py b/osm_mon/evaluator/service.py
new file mode 100644
index 0000000..0868e32
--- /dev/null
+++ b/osm_mon/evaluator/service.py
@@ -0,0 +1,153 @@
+# -*- coding: utf-8 -*-
+
+# Copyright 2018 Whitestack, LLC
+# *************************************************************
+
+# This file is part of OSM Monitoring module
+# All Rights Reserved to Whitestack, LLC
+
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+
+# http://www.apache.org/licenses/LICENSE-2.0
+
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+# For those usages not covered by the Apache License, Version 2.0 please
+# contact: bdiaz@whitestack.com or glavado@whitestack.com
+##
+import logging
+import multiprocessing
+from enum import Enum
+from typing import Tuple, List
+
+from osm_common.dbbase import DbException
+
+from osm_mon.core import database
+from osm_mon.core.common_db import CommonDbClient
+from osm_mon.core.config import Config
+from osm_mon.core.database import Alarm, AlarmRepository
+from osm_mon.evaluator.backends.prometheus import PrometheusBackend
+
+log = logging.getLogger(__name__)
+
+BACKENDS = {
+ 'prometheus': PrometheusBackend
+}
+
+
+class AlarmStatus(Enum):
+ ALARM = 'alarm'
+ OK = 'ok'
+ INSUFFICIENT = 'insufficient-data'
+
+
+class EvaluatorService:
+
+ def __init__(self, config: Config):
+ self.conf = config
+ self.common_db = CommonDbClient(self.conf)
+ self.queue = multiprocessing.Queue()
+
+ def _get_metric_value(self,
+ nsr_id: str,
+ vnf_member_index: int,
+ vdur_name: str,
+ metric_name: str):
+ return BACKENDS[self.conf.get('evaluator', 'backend')]().get_metric_value(metric_name, nsr_id, vdur_name,
+ vnf_member_index)
+
+ def _evaluate_metric(self,
+ nsr_id: str,
+ vnf_member_index: int,
+ vdur_name: str,
+ metric_name: str,
+ alarm: Alarm):
+ log.debug("_evaluate_metric")
+ metric_value = self._get_metric_value(nsr_id, vnf_member_index, vdur_name, metric_name)
+ if not metric_value:
+ log.warning("No metric result for alarm %s", alarm.id)
+ self.queue.put((alarm, AlarmStatus.INSUFFICIENT))
+ else:
+ if alarm.operation.upper() == 'GT':
+ if metric_value > alarm.threshold:
+ 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, AlarmStatus.ALARM))
+ else:
+ self.queue.put((alarm, AlarmStatus.OK))
+
+ def evaluate_alarms(self) -> List[Tuple[Alarm, AlarmStatus]]:
+ log.debug('evaluate_alarms')
+ processes = []
+ database.db.connect()
+ try:
+ with database.db.atomic():
+ for alarm in AlarmRepository.list():
+ try:
+ vnfr = self.common_db.get_vnfr(alarm.nsr_id, alarm.vnf_member_index)
+ except DbException:
+ log.exception("Error getting vnfr: ")
+ continue
+ vnfd = self.common_db.get_vnfd(vnfr['vnfd-id'])
+ try:
+ vdur = next(filter(lambda vdur: vdur['name'] == alarm.vdur_name, vnfr['vdur']))
+ except StopIteration:
+ log.warning("No vdur found with name %s for alarm %s", alarm.vdur_name, alarm.id)
+ continue
+ vdu = next(filter(lambda vdu: vdu['id'] == vdur['vdu-id-ref'], vnfd['vdu']))
+ vnf_monitoring_param = next(
+ filter(lambda param: param['id'] == alarm.monitoring_param, vnfd['monitoring-param']))
+ nsr_id = vnfr['nsr-id-ref']
+ vnf_member_index = vnfr['member-vnf-index-ref']
+ vdur_name = vdur['name']
+ if 'vdu-monitoring-param' in vnf_monitoring_param:
+ vdu_monitoring_param = next(filter(
+ lambda param: param['id'] == vnf_monitoring_param['vdu-monitoring-param'][
+ 'vdu-monitoring-param-ref'], vdu['monitoring-param']))
+ nfvi_metric = vdu_monitoring_param['nfvi-metric']
+
+ p = multiprocessing.Process(target=self._evaluate_metric,
+ args=(nsr_id,
+ vnf_member_index,
+ vdur_name,
+ nfvi_metric,
+ alarm))
+ processes.append(p)
+ p.start()
+ if 'vdu-metric' in vnf_monitoring_param:
+ vnf_metric_name = vnf_monitoring_param['vdu-metric']['vdu-metric-name-ref']
+ p = multiprocessing.Process(target=self._evaluate_metric,
+ args=(nsr_id,
+ vnf_member_index,
+ vdur_name,
+ vnf_metric_name,
+ alarm))
+ processes.append(p)
+ p.start()
+ if 'vnf-metric' in vnf_monitoring_param:
+ vnf_metric_name = vnf_monitoring_param['vnf-metric']['vnf-metric-name-ref']
+ p = multiprocessing.Process(target=self._evaluate_metric,
+ args=(nsr_id,
+ vnf_member_index,
+ '',
+ vnf_metric_name,
+ alarm))
+ processes.append(p)
+ p.start()
+
+ for process in processes:
+ process.join(timeout=10)
+ alarms_tuples = []
+ while not self.queue.empty():
+ alarms_tuples.append(self.queue.get())
+ return alarms_tuples
+ finally:
+ database.db.close()