voltha/extensions/alarms/device_alarms.py

#
# Copyright 2019 the original author or authors.
#
# 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.

import structlog
log = structlog.get_logger()


class DeviceAlarms:
    """
    Class for managing device alarms
    """
    def __init__(self, data_model):
        """
        Adapter alarm manager initializer

        :param data_model: data_model
        """
        self.log = structlog.get_logger()
        self.data_model = data_model
        self.lc = None

    def format_id(self, alarm):
        """
        Format the Unique Alarm ID for this alarm.  This is provided in the
        alarms 'id' field

        :param alarm: (str) The name of the alarm such as 'Discover' or 'LOS'

        :return: (str) Alarm ID
        """
        return 'voltha.{}.{}.{}'.format(self.data_model._adapter_name(),
                                        self.data_model._device_id(),
                                        alarm)

    def format_description(self, _object, alarm, status):
        """
        Format the textual description field of this alarm

        :param _object: ()
        :param alarm: (str) The name of the alarm such as 'Discover' or 'LOS'
        :param status: (bool) If True, the alarm is active (it is being raised)

        :return: (str) Alarm description
        """
        return '{} Alarm - {} - {}'.format(_object.upper(),
                                           alarm.upper(),
                                           'Raised' if status else 'Cleared')

    def send_alarm(self, context_data, alarm_data):
        """
        Send the alarm to the event bus

        :param context_data: (dict) Alarm specific context data
        :param alarm_data: (dict) Common Alarm information dictionary
        """
        try:
            current_context = {}
            if isinstance(context_data, dict):
                for key, value in context_data.iteritems():
                    current_context[key] = str(value)
            # FIXME - The entity that is raising the alarm needs to
            # provide the onu and olt serial numbers.
            alarm_event = self.data_model.create_alarm(
                id=alarm_data.get('id', 'voltha.{}.{}.olt'.format(
                    self.data_model._adapter_name(),
                    self.data_model._device_id())),
                resource_id=str(alarm_data.get('resource_id',
                                               self.data_model.device.id)),
                description="{}.{} - {}".format(self.data_model._adapter_name(),
                                                self.data_model._device_id(),
                                                alarm_data.get('description')),
                type=alarm_data.get('type'),
                category=alarm_data.get('category'),
                severity=alarm_data.get('severity'),
                state=alarm_data.get('state'),
                raised_ts=alarm_data.get('ts', 0),
                context=current_context,
                alarm_type_name=alarm_data.get('alarm_type_name')
            )
            self.data_model.submit_alarm(alarm_event)

        except Exception as e:
            self.log.exception('failed-to-send-alarm', e=e)
            raise