--- /dev/null
+# -*- 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 json
+import logging
+import multiprocessing
+import time
+
+from osm_common.dbbase import DbException
+
+from osm_mon.collector.collector import VIM_COLLECTORS
+from osm_mon.collector.collectors.juju import VCACollector
+from osm_mon.core.common_db import CommonDbClient
+from osm_mon.core.database import DatabaseManager, Alarm
+from osm_mon.core.message_bus.producer import Producer
+from osm_mon.core.response import ResponseBuilder
+from osm_mon.core.settings import Config
+
+log = logging.getLogger(__name__)
+
+
+class Evaluator:
+ def __init__(self):
+ self.common_db = CommonDbClient()
+ self.plugins = []
+ self.database_manager = DatabaseManager()
+ self.database_manager.create_tables()
+ self.queue = multiprocessing.Queue()
+
+ def _evaluate_vim_metric(self,
+ nsr_id: str,
+ vnf_member_index: int,
+ vdur_name: str,
+ nfvi_metric_name: str,
+ vim_account_id: str,
+ alarm: Alarm):
+ vim_type = self.database_manager.get_vim_type(vim_account_id)
+ if vim_type in VIM_COLLECTORS:
+ collector = VIM_COLLECTORS[vim_type](vim_account_id)
+ metric = collector.collect_one(nsr_id, vnf_member_index, vdur_name, nfvi_metric_name)
+ if alarm.operation.upper() == 'GT':
+ if metric.value > alarm.threshold:
+ self.queue.put(alarm)
+ elif alarm.operation.upper() == 'LT':
+ if metric.value < alarm.threshold:
+ self.queue.put(alarm)
+
+ else:
+ log.debug("vimtype %s is not supported.", vim_type)
+
+ def _evaluate_vca_metric(self,
+ nsr_id: str,
+ vnf_member_index: int,
+ vdur_name: str,
+ vnf_metric_name: str,
+ alarm: Alarm):
+ collector = VCACollector()
+ metric = collector.collect_one(nsr_id, vnf_member_index, vdur_name, vnf_metric_name)
+ if alarm.operation.upper() == 'GT':
+ if metric.value > alarm.threshold:
+ self.queue.put(alarm)
+ elif alarm.operation.upper() == 'LT':
+ if metric.value < alarm.threshold:
+ self.queue.put(alarm)
+
+ def evaluate_forever(self):
+ log.debug('collect_forever')
+ cfg = Config.instance()
+ while True:
+ try:
+ self.evaluate()
+ time.sleep(cfg.OSMMON_EVALUATOR_INTERVAL)
+ except Exception:
+ log.exception("Error evaluating alarms")
+
+ def evaluate(self):
+ 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']
+
+ vim_account_id = self.common_db.get_vim_account_id(nsr_id, vnf_member_index)
+ p = multiprocessing.Process(target=self._evaluate_vim_metric,
+ args=(nsr_id,
+ vnf_member_index,
+ vdur_name,
+ nfvi_metric,
+ vim_account_id,
+ 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_vca_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:
+ log.warning("vnf-metric is not currently supported.")
+ continue
+
+ for process in processes:
+ process.join()
+ triggered_alarms = []
+ while not self.queue.empty():
+ triggered_alarms.append(self.queue.get())
+ for alarm in triggered_alarms:
+ self.notify_alarm(alarm)
+ p = multiprocessing.Process(target=self.notify_alarm,
+ args=(alarm,))
+ p.start()
+
+ def notify_alarm(self, alarm: Alarm):
+ response = ResponseBuilder()
+ now = time.strftime("%d-%m-%Y") + " " + time.strftime("%X")
+ # Generate and send response
+ resp_message = response.generate_response(
+ 'notify_alarm',
+ alarm_id=alarm.id,
+ vdu_name=alarm.vdur_name,
+ vnf_member_index=alarm.vnf_member_index,
+ ns_id=alarm.nsr_id,
+ metric_name=alarm.monitoring_param,
+ operation=alarm.operation,
+ threshold_value=alarm.threshold,
+ sev=alarm.severity,
+ status='alarm',
+ 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)
projects / osm / MON.git / blobdiff