--- /dev/null
+# ==================================================================================
+# Copyright (c) 2020 AT&T Intellectual Property.
+# Copyright (c) 2020 Nokia
+#
+# 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.
+# ==================================================================================
+"""
+Provides classes and methods to define, raise, reraise and clear alarms.
+All actions are implemented by sending RMR messages to the Alarm Adapter
+that comply with the JSON schema in file alarm-schema.json.
+"""
+
+from ctypes import c_void_p
+from enum import Enum, auto
+import json
+import time
+from mdclogpy import Logger
+from ricxappframe.rmr import rmr
+
+##############
+# PRIVATE API
+##############
+
+mdc_logger = Logger(name=__name__)
+RETRIES = 4
+
+##############
+# PUBLIC API
+##############
+
+# constants
+RIC_ALARM_UPDATE = 13111
+# RIC_ALARM_QUERY = 13112 # TBD
+
+# Publish dict keys as constants for convenience of client code.
+# Mixed lower/upper casing to comply with the Adapter JSON requirements.
+KEY_ALARM = "alarm"
+KEY_MANAGED_OBJECT_ID = "managedObjectId"
+KEY_APPLICATION_ID = "applicationId"
+KEY_SPECIFIC_PROBLEM = "specificProblem"
+KEY_PERCEIVED_SEVERITY = "perceivedSeverity"
+KEY_ADDITIONAL_INFO = "additionalInfo"
+KEY_IDENTIFYING_INFO = "identifyingInfo"
+KEY_ALARM_ACTION = "AlarmAction"
+KEY_ALARM_TIME = "AlarmTime"
+
+
+class AlarmAction(Enum):
+ """
+ Action to perform at the Alarm Adapter
+ """
+ RAISE = auto()
+ CLEAR = auto()
+ CLEARALL = auto()
+
+
+class AlarmSeverity(Enum):
+ """
+ Severity of an alarm
+ """
+ UNSPECIFIED = auto()
+ CRITICAL = auto()
+ MAJOR = auto()
+ MINOR = auto()
+ WARNING = auto()
+ CLEARED = auto()
+ DEFAULT = auto()
+
+
+class AlarmDetail(dict):
+ """
+ An alarm that can be raised or cleared.
+
+ Parameters
+ ----------
+ managed_object_id: str
+ The name of the managed object that is the cause of the fault (required)
+
+ application_id: str
+ The name of the process that raised the alarm (required)
+
+ specific_problem: int
+ The problem that is the cause of the alarm
+
+ perceived_severity: AlarmSeverity
+ The severity of the alarm, a value from the enum.
+
+ identifying_info: str
+ Identifying additional information, which is part of alarm identity
+
+ additional_info: str
+ Additional information given by the application (optional)
+ """
+ # pylint: disable=too-many-arguments
+ def __init__(self,
+ managed_object_id: str,
+ application_id: str,
+ specific_problem: int,
+ perceived_severity: AlarmSeverity,
+ identifying_info: str,
+ additional_info: str = ""):
+ """
+ Creates an object with the specified items.
+ """
+ dict.__init__(self)
+ self[KEY_MANAGED_OBJECT_ID] = managed_object_id
+ self[KEY_APPLICATION_ID] = application_id
+ self[KEY_SPECIFIC_PROBLEM] = specific_problem
+ self[KEY_PERCEIVED_SEVERITY] = perceived_severity.name
+ self[KEY_IDENTIFYING_INFO] = identifying_info
+ self[KEY_ADDITIONAL_INFO] = additional_info
+
+
+class AlarmManager:
+ """
+ Provides an API for an Xapp to raise and clear alarms by sending messages
+ via RMR, which should route the messages to an Alarm Adapter.
+
+ Parameters
+ ----------
+ vctx: ctypes c_void_p
+ Pointer to RMR context obtained by initializing RMR.
+ The context is used to allocate space and send messages.
+ The RMR routing table must have a destination for message
+ type RIC_ALARM_UPDATE as defined in this module.
+
+ managed_object_id: str
+ The name of the managed object that raises alarms
+
+ application_id: str
+ The name of the process that raises alarms
+ """
+ def __init__(self,
+ vctx: c_void_p,
+ managed_object_id: str,
+ application_id: str):
+ """
+ Creates an alarm manager.
+ """
+ self.vctx = vctx
+ self.managed_object_id = managed_object_id
+ self.application_id = application_id
+
+ def create_alarm(self,
+ specific_problem: int,
+ perceived_severity: AlarmSeverity,
+ identifying_info: str,
+ additional_info: str = ""):
+ """
+ Convenience method that creates an alarm instance, an AlarmDetail object,
+ using cached values for managed object ID and application ID.
+
+ Parameters
+ ----------
+ specific_problem: int
+ The problem that is the cause of the alarm
+
+ perceived_severity: AlarmSeverity
+ The severity of the alarm, a value from the enum.
+
+ identifying_info: str
+ Identifying additional information, which is part of alarm identity
+
+ additional_info: str
+ Additional information given by the application (optional)
+
+ Returns
+ -------
+ AlarmDetail
+ """
+ return AlarmDetail(managed_object_id=self.managed_object_id,
+ application_id=self.application_id,
+ specific_problem=specific_problem, perceived_severity=perceived_severity,
+ identifying_info=identifying_info, additional_info=additional_info)
+
+ @staticmethod
+ def _create_alarm_message(alarm: AlarmDetail, action: AlarmAction):
+ """
+ Creates a dict with the specified alarm detail plus action and time.
+ Uses the current system time in milliseconds since the Epoch.
+
+ Parameters
+ ----------
+ detail: AlarmDetail
+ The alarm details.
+
+ action: AlarmAction
+ The action to perform at the Alarm Adapter on this alarm.
+ """
+ return {
+ **alarm,
+ KEY_ALARM_ACTION: action.name,
+ KEY_ALARM_TIME: int(round(time.time() * 1000))
+ }
+
+ def _rmr_send_alarm(self, msg: dict):
+ """
+ Serializes the dict and sends the result via RMR using a predefined message type.
+
+ Parameters
+ ----------
+ msg: dict
+ Dictionary with alarm message to encode and send
+
+ Returns
+ -------
+ bool
+ True if the send succeeded (possibly with retries), False otherwise
+ """
+ payload = json.dumps(msg).encode()
+ mdc_logger.debug("_rmr_send_alarm: payload is {}".format(payload))
+ sbuf = rmr.rmr_alloc_msg(vctx=self.vctx, size=len(payload), payload=payload,
+ mtype=RIC_ALARM_UPDATE, gen_transaction_id=True)
+
+ for _ in range(0, RETRIES):
+ sbuf = rmr.rmr_send_msg(self.vctx, sbuf)
+ post_send_summary = rmr.message_summary(sbuf)
+ mdc_logger.debug("_rmr_send_alarm: try {0} result is {1}".format(_, post_send_summary[rmr.RMR_MS_MSG_STATE]))
+ # stop trying if RMR does not indicate retry
+ if post_send_summary[rmr.RMR_MS_MSG_STATE] != rmr.RMR_ERR_RETRY:
+ break
+
+ rmr.rmr_free_msg(sbuf)
+ if post_send_summary[rmr.RMR_MS_MSG_STATE] != rmr.RMR_OK:
+ mdc_logger.warning("_rmr_send_alarm: failed after {} retries".format(RETRIES))
+ return False
+
+ return True
+
+ def raise_alarm(self, detail: AlarmDetail):
+ """
+ Builds and sends a message to the AlarmAdapter to raise an alarm
+ with the specified detail.
+
+ Parameters
+ ----------
+ detail: AlarmDetail
+ Alarm to raise
+
+ Returns
+ -------
+ bool
+ True if the send succeeded (possibly with retries), False otherwise
+ """
+ msg = self._create_alarm_message(detail, AlarmAction.RAISE)
+ return self._rmr_send_alarm(msg)
+
+ def clear_alarm(self, detail: AlarmDetail):
+ """
+ Builds and sends a message to the AlarmAdapter to clear the alarm
+ with the specified detail.
+
+ Parameters
+ ----------
+ detail: AlarmDetail
+ Alarm to clear
+
+ Returns
+ -------
+ bool
+ True if the send succeeded (possibly with retries), False otherwise
+ """
+ msg = self._create_alarm_message(detail, AlarmAction.CLEAR)
+ return self._rmr_send_alarm(msg)
+
+ def reraise_alarm(self, detail: AlarmDetail):
+ """
+ Builds and sends a message to the AlarmAdapter to clear the alarm with the
+ the specified detail, then builds and sends a message to raise the alarm again.
+
+ Parameters
+ ----------
+ detail: AlarmDetail
+ Alarm to clear and raise again.
+
+ Returns
+ -------
+ bool
+ True if the send succeeded (possibly with retries), False otherwise
+ """
+ success = self.clear_alarm(detail)
+ if success:
+ success = self.raise_alarm(detail)
+ return success
+
+ def clear_all_alarms(self):
+ """
+ Builds and sends a message to the AlarmAdapter to clear all alarms.
+
+ Returns
+ -------
+ bool
+ True if the send succeeded (possibly with retries), False otherwise
+ """
+ detail = self.create_alarm(0, AlarmSeverity.DEFAULT, "", "")
+ msg = self._create_alarm_message(detail, AlarmAction.CLEARALL)
+ return self._rmr_send_alarm(msg)
Code Review / ric-plt / xapp-frame-py.git / blobdiff