X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=ricxappframe%2Falarm%2Falarm.py;fp=ricxappframe%2Falarm%2Falarm.py;h=9a3ba2d0779099733aaeb0d989af9886d706ae77;hb=81084bc31ea1d5cde6616dd2267ea01f49a1d6d1;hp=0000000000000000000000000000000000000000;hpb=884192bf7ab8d637e8007760fbe50dbcdccd4671;p=ric-plt%2Fxapp-frame-py.git

diff --git a/ricxappframe/alarm/alarm.py b/ricxappframe/alarm/alarm.py
new file mode 100644
index 0000000..9a3ba2d
--- /dev/null
+++ b/ricxappframe/alarm/alarm.py
@@ -0,0 +1,307 @@
+# ==================================================================================
+#       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)