osm_mon/evaluator/evaluator.py

# 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 asyncio
import logging
import multiprocessing
import time

from osm_mon.core.config import Config
from osm_mon.core.message_bus_client import MessageBusClient
from osm_mon.core.models import Alarm
from osm_mon.core.response import ResponseBuilder
from osm_mon.evaluator.service import EvaluatorService, AlarmStatus

log = logging.getLogger(__name__)


class Evaluator:
    def __init__(self, config: Config):
        self.conf = config
        self.service = EvaluatorService(config)
        self.msg_bus = MessageBusClient(config)

    def evaluate_forever(self):
        log.debug("evaluate_forever")
        while True:
            try:
                self.evaluate()
                time.sleep(int(self.conf.get("evaluator", "interval")))
            except Exception:
                log.exception("Error evaluating alarms")

    def evaluate(self):
        log.debug("evaluate")
        log.info("Starting alarm evaluation")
        alarms_tuples = self.service.evaluate_alarms()
        processes = []
        for alarm, status in alarms_tuples:
            p = multiprocessing.Process(target=self.notify_alarm, args=(alarm, status))
            p.start()
            processes.append(p)
        for process in processes:
            process.join(timeout=10)
        log.info("Alarm evaluation is complete")

    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)
        asyncio.run(
            self.msg_bus.aiowrite("alarm_response", "notify_alarm", resp_message)
        )
        evaluator_service = EvaluatorService(self.conf)
        evaluator_service.update_alarm_status(status.value, alarm.uuid)
        evaluator_service.update_alarm_extra_labels(alarm.extra_labels, alarm.uuid)
        return

    def _build_alarm_response(self, alarm: Alarm, status: AlarmStatus):
        log.debug("_build_alarm_response")
        response = ResponseBuilder()
        tags = {}
        for name, value in alarm.tags.items():
            tags[name] = value
        now = time.strftime("%d-%m-%Y") + " " + time.strftime("%X")
        return response.generate_response(
            "notify_alarm",
            alarm_id=alarm.uuid,
            metric_name=alarm.metric,
            operation=alarm.operation,
            threshold_value=alarm.threshold,
            sev=alarm.severity,
            status=status.value,
            date=now,
            tags=tags,
            extra_tabels=alarm.extra_labels,
        )