ricxappframe/xapp_frame.py

# ==================================================================================
#       Copyright (c) 2020 Nokia
#       Copyright (c) 2020 AT&T Intellectual Property.
#
#   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.
# ==================================================================================
"""
This framework for Python Xapps provides classes that Xapp writers should
instantiate and/or subclass depending on their needs.
"""

import json
import os
import queue
from threading import Thread
import inotify_simple
from mdclogpy import Logger
from ricxappframe import xapp_rmr
from ricxappframe.rmr import rmr
from ricxappframe.xapp_sdl import SDLWrapper

# message-type constants
RIC_HEALTH_CHECK_REQ = 100
RIC_HEALTH_CHECK_RESP = 101

# environment variable with path to configuration file
CONFIG_FILE_ENV = "CONFIG_FILE"

# Private base class; not for direct client use


class _BaseXapp:
    """
    This base class initializes RMR by starting a thread that checks for
    incoming messages, and provisions an SDL object.

    If environment variable CONFIG_FILE_ENV is defined, and that value is a
    path to an existing file, a watcher is defined to monitor modifications
    (writes) to that file using the Linux kernel's inotify feature, and the
    configuration-change handler function is invoked.  The watcher can be
    polled by calling method config_check().

    Parameters
    ----------
    rmr_port: int
        port to listen on

    rmr_wait_for_ready: bool (optional)
        If this is True, then init waits until rmr is ready to send, which
        includes having a valid routing file. This can be set to
        False if the client only wants to *receive only*.

    use_fake_sdl: bool (optional)
        if this is True, it uses the dbaas "fake dict backend" instead
        of Redis or other backends. Set this to true when developing
        an xapp or during unit testing to eliminate the need for DBAAS.

    post_init: function (optional)
        Runs this user-provided function after the base xapp is
        initialized; its signature should be post_init(self)
    """

    def __init__(self, rmr_port=4562, rmr_wait_for_ready=True, use_fake_sdl=False, post_init=None):
        """
        Documented in the class comment.
        """
        # PUBLIC, can be used by xapps using self.(name):
        self.logger = Logger(name=__name__)

        # Start rmr rcv thread
        self._rmr_loop = xapp_rmr.RmrLoop(port=rmr_port, wait_for_ready=rmr_wait_for_ready)
        self._mrc = self._rmr_loop.mrc  # for convenience

        # SDL
        self._sdl = SDLWrapper(use_fake_sdl)

        # Config
        # The environment variable specifies the path to the Xapp config file
        self._config_path = os.environ.get(CONFIG_FILE_ENV, None)
        if self._config_path and os.path.isfile(self._config_path):
            self._inotify = inotify_simple.INotify()
            self._inotify.add_watch(self._config_path, inotify_simple.flags.MODIFY)
            self.logger.debug("__init__: watching config file {}".format(self._config_path))
        else:
            self._inotify = None
            self.logger.warning("__init__: NOT watching any config file")

        # run the optionally provided user post init
        if post_init:
            post_init(self)

    # Public rmr methods

    def rmr_get_messages(self):
        """
        Returns a generator iterable over all items in the queue that
        have not yet been read by the client xapp. Each item is a tuple
        (S, sbuf) where S is a message summary dict and sbuf is the raw
        message. The caller MUST call rmr.rmr_free_msg(sbuf) when
        finished with each sbuf to prevent memory leaks!
        """
        while not self._rmr_loop.rcv_queue.empty():
            (summary, sbuf) = self._rmr_loop.rcv_queue.get()
            yield (summary, sbuf)

    def rmr_send(self, payload, mtype, retries=100):
        """
        Allocates a buffer, sets payload and mtype, and sends

        Parameters
        ----------
        payload: bytes
            payload to set
        mtype: int
            message type
        retries: int (optional)
            Number of times to retry at the application level before excepting RMRFailure

        Returns
        -------
        bool
            whether or not the send worked after retries attempts
        """
        sbuf = rmr.rmr_alloc_msg(vctx=self._mrc, size=len(payload), payload=payload, gen_transaction_id=True, mtype=mtype)

        for _ in range(retries):
            sbuf = rmr.rmr_send_msg(self._mrc, sbuf)
            if sbuf.contents.state == 0:
                self.rmr_free(sbuf)
                return True

        self.rmr_free(sbuf)
        return False

    def rmr_rts(self, sbuf, new_payload=None, new_mtype=None, retries=100):
        """
        Allows the xapp to return to sender, possibly adjusting the
        payload and message type before doing so.  This does NOT free
        the sbuf for the caller as the caller may wish to perform
        multiple rts per buffer. The client needs to free.

        Parameters
        ----------
        sbuf: ctypes c_void_p
             Pointer to an rmr message buffer
        new_payload: bytes (optional)
            New payload to set
        new_mtype: int (optional)
            New message type (replaces the received message)
        retries: int (optional)
            Number of times to retry at the application level before
            throwing exception RMRFailure

        Returns
        -------
        bool
            whether or not the send worked after retries attempts
        """
        for _ in range(retries):
            sbuf = rmr.rmr_rts_msg(self._mrc, sbuf, payload=new_payload, mtype=new_mtype)
            if sbuf.contents.state == 0:
                return True

        self.logger.warning("RTS Failed! Summary: {}".format(rmr.message_summary(sbuf)))
        return False

    def rmr_free(self, sbuf):
        """
        Frees an rmr message buffer after use

        Note: this does not need to be a class method, self is not
        used. However if we break it out as a function we need a home
        for it.

        Parameters
        ----------
        sbuf: ctypes c_void_p
             Pointer to an rmr message buffer
        """
        rmr.rmr_free_msg(sbuf)

    # Convenience (pass-thru) function for invoking SDL.

    def sdl_set(self, ns, key, value, usemsgpack=True):
        """
        Stores a key-value pair to SDL, optionally serializing the value
        to bytes using msgpack.

        Parameters
        ----------
        ns: string
            SDL namespace
        key: string
            SDL key
        value:
            Object or byte array to store.  See the `usemsgpack` parameter.
        usemsgpack: boolean (optional, default is True)
            Determines whether the value is serialized using msgpack before storing.
            If usemsgpack is True, the msgpack function `packb` is invoked
            on the value to yield a byte array that is then sent to SDL.
            Stated differently, if usemsgpack is True, the value can be anything
            that is serializable by msgpack.
            If usemsgpack is False, the value must be bytes.
        """
        self._sdl.set(ns, key, value, usemsgpack)

    def sdl_get(self, ns, key, usemsgpack=True):
        """
        Gets the value for the specified namespace and key from SDL,
        optionally deserializing stored bytes using msgpack.

        Parameters
        ----------
        ns: string
            SDL namespace
        key: string
            SDL key
        usemsgpack: boolean (optional, default is True)
            If usemsgpack is True, the byte array stored by SDL is deserialized
            using msgpack to yield the original object that was stored.
            If usemsgpack is False, the byte array stored by SDL is returned
            without further processing.

        Returns
        -------
        Value
            See the usemsgpack parameter for an explanation of the returned value type.
            Answers None if the key is not found.
        """
        return self._sdl.get(ns, key, usemsgpack)

    def sdl_find_and_get(self, ns, prefix, usemsgpack=True):
        """
        Gets all key-value pairs in the specified namespace
        with keys that start with the specified prefix,
        optionally deserializing stored bytes using msgpack.

        Parameters
        ----------
        ns: string
           SDL namespace
        prefix: string
            the key prefix
        usemsgpack: boolean (optional, default is True)
            If usemsgpack is True, the byte array stored by SDL is deserialized
            using msgpack to yield the original value that was stored.
            If usemsgpack is False, the byte array stored by SDL is returned
            without further processing.

        Returns
        -------
        Dictionary of key-value pairs
            Each key has the specified prefix.
            The value object (its type) depends on the usemsgpack parameter,
            but is either a Python object or raw bytes as discussed above.
            Answers an empty dictionary if no keys matched the prefix.
        """
        return self._sdl.find_and_get(ns, prefix, usemsgpack)

    def sdl_delete(self, ns, key):
        """
        Deletes the key-value pair with the specified key in the specified namespace.

        Parameters
        ----------
        ns: string
           SDL namespace
        key: string
            SDL key
        """
        self._sdl.delete(ns, key)

    # Health

    def healthcheck(self):
        """
        this needs to be understood how this is supposed to work
        """
        return self._rmr_loop.healthcheck() and self._sdl.healthcheck()

    # Convenience function for discovering config change events

    def config_check(self, timeout=0):
        """
        Checks the watcher for configuration-file events. The watcher
        prerequisites and event mask are documented in __init__().

        Parameters
        ----------
        timeout: int (optional)
            Number of seconds to wait for a configuration-file event, default 0.

        Returns
        -------
        List of Events, possibly empty
            An event is a tuple with objects wd, mask, cookie and name.
            For example::

                Event(wd=1, mask=1073742080, cookie=0, name='foo')

        """
        if not self._inotify:
            return []
        events = self._inotify.read(timeout=timeout)
        return list(events)

    def stop(self):
        """
        cleans up and stops the xapp rmr thread (currently). This is
        critical for unit testing as pytest will never return if the
        thread is running.

        TODO: can we register a ctrl-c handler so this gets called on
        ctrl-c? Because currently two ctrl-c are needed to stop.
        """
        self._rmr_loop.stop()


# Public classes that Xapp writers should instantiate or subclass
# to implement an Xapp.


class RMRXapp(_BaseXapp):
    """
    Represents an Xapp that reacts only to RMR messages; i.e., when
    messages are received, the Xapp does something. When run is called,
    the xapp framework waits for RMR messages, and calls the appropriate
    client-registered consume callback on each.

    If environment variable CONFIG_FILE_ENV is defined, and that value is a
    path to an existing file, the configuration-change handler is invoked at
    startup and on each configuration-file write event. If no handler is
    supplied, this class defines a default handler that logs each invocation.

    Parameters
    ----------
    default_handler: function
        A function with the signature (summary, sbuf) to be called
        when a message type is received for which no other handler is registered.
    default_handler argument summary: dict
        The RMR message summary, a dict of key-value pairs
    default_handler argument sbuf: ctypes c_void_p
        Pointer to an RMR message buffer. The user must call free on this when done.
    config_handler: function (optional, default is documented above)
        A function with the signature (json) to be called at startup and each time
        a configuration-file change event is detected. The JSON object is read from
        the configuration file, if the prerequisites are met.
    config_handler argument json: dict
        The contents of the configuration file, parsed as JSON.
    rmr_port: integer (optional, default is 4562)
        Initialize RMR to listen on this port
    rmr_wait_for_ready: boolean (optional, default is True)
        Wait for RMR to signal ready before starting the dispatch loop
    use_fake_sdl: boolean (optional, default is False)
        Use an in-memory store instead of the real SDL service
    post_init: function (optional, default None)
        Run this function after the app initializes and before the dispatch loop starts;
        its signature should be post_init(self)
    """

    def __init__(self, default_handler, config_handler=None, rmr_port=4562, rmr_wait_for_ready=True, use_fake_sdl=False, post_init=None):
        """
        Also see _BaseXapp
        """
        # init base
        super().__init__(
            rmr_port=rmr_port, rmr_wait_for_ready=rmr_wait_for_ready, use_fake_sdl=use_fake_sdl, post_init=post_init
        )

        # setup callbacks
        self._default_handler = default_handler
        self._config_handler = config_handler
        self._dispatch = {}

        # used for thread control
        self._keep_going = True

        # register a default healthcheck handler
        # this default checks that rmr is working and SDL is working
        # the user can override this and register their own handler
        # if they wish since the "last registered callback wins".
        def handle_healthcheck(self, summary, sbuf):
            ok = self.healthcheck()
            payload = b"OK\n" if ok else b"ERROR [RMR or SDL is unhealthy]\n"
            self.rmr_rts(sbuf, new_payload=payload, new_mtype=RIC_HEALTH_CHECK_RESP)
            self.rmr_free(sbuf)

        self.register_callback(handle_healthcheck, RIC_HEALTH_CHECK_REQ)

        # define a default configuration-change handler if none was provided.
        if not config_handler:
            def handle_config_change(self, config):
                self.logger.debug("xapp_frame: default config handler invoked")
            self._config_handler = handle_config_change

        # call the config handler at startup if prereqs were met
        if self._inotify:
            with open(self._config_path) as json_file:
                data = json.load(json_file)
            self.logger.debug("run: invoking config handler at start")
            self._config_handler(self, data)

    def register_callback(self, handler, message_type):
        """
        registers this xapp to call handler(summary, buf) when an rmr message is received of type message_type

        Parameters
        ----------
        handler: function
            a function with the signature (summary, sbuf) to be called
            when a message of type message_type is received
        summary: dict
            the rmr message summary
        sbuf: ctypes c_void_p
            Pointer to an rmr message buffer. The user must call free on this when done.

        message:type: int
            the message type to look for

        Note if this method is called multiple times for a single message type, the "last one wins".
        """
        self._dispatch[message_type] = handler

    def run(self, thread=False, rmr_timeout=5, inotify_timeout=0):
        """
        This function should be called when the reactive Xapp is ready to start.
        After start, the Xapp's handlers will be called on received messages.

        Parameters
        ----------
        thread: bool (optional, default is False)
            If False, execution is not returned and the framework loops forever.
            If True, a thread is started to run the queue read/dispatch loop
            and execution is returned to caller; the thread can be stopped
            by calling the .stop() method.

        rmr_timeout: integer (optional, default is 5 seconds)
            Length of time to wait for an RMR message to arrive.

        inotify_timeout: integer (optional, default is 0 seconds)
            Length of time to wait for an inotify event to arrive.
        """

        def loop():
            while self._keep_going:

                # poll RMR
                try:
                    (summary, sbuf) = self._rmr_loop.rcv_queue.get(block=True, timeout=rmr_timeout)
                    # dispatch
                    func = self._dispatch.get(summary[rmr.RMR_MS_MSG_TYPE], None)
                    if not func:
                        func = self._default_handler
                    self.logger.debug("run: invoking msg handler on type {}".format(summary[rmr.RMR_MS_MSG_TYPE]))
                    func(self, summary, sbuf)
                except queue.Empty:
                    # the get timed out
                    pass

                # poll configuration file watcher
                try:
                    events = self.config_check(timeout=inotify_timeout)
                    for event in events:
                        with open(self._config_path) as json_file:
                            data = json.load(json_file)
                        self.logger.debug("run: invoking config handler on change event {}".format(event))
                        self._config_handler(self, data)
                except Exception as error:
                    self.logger.error("run: configuration handler failed: {}".format(error))

        if thread:
            Thread(target=loop).start()
        else:
            loop()

    def stop(self):
        """
        Sets the flag to end the dispatch loop.
        """
        super().stop()
        self.logger.debug("Setting flag to end framework work loop.")
        self._keep_going = False


class Xapp(_BaseXapp):
    """
    Represents a generic Xapp where the client provides a single function
    for the framework to call at startup time (instead of providing callback
    functions by message type). The Xapp writer must implement and provide a
    function with a loop-forever construct similar to the `run` function in
    the `RMRXapp` class.  That function should poll to retrieve RMR messages
    and dispatch them appropriately, poll for configuration changes, etc.

    Parameters
    ----------
    entrypoint: function
        This function is called when the Xapp class's run method is invoked.
        The function signature must be just function(self)
    rmr_port: integer (optional, default is 4562)
        Initialize RMR to listen on this port
    rmr_wait_for_ready: boolean (optional, default is True)
        Wait for RMR to signal ready before starting the dispatch loop
    use_fake_sdl: boolean (optional, default is False)
        Use an in-memory store instead of the real SDL service
    """

    def __init__(self, entrypoint, rmr_port=4562, rmr_wait_for_ready=True, use_fake_sdl=False):
        """
        Parameters
        ----------

        For the other parameters, see class _BaseXapp.
        """
        # init base
        super().__init__(rmr_port=rmr_port, rmr_wait_for_ready=rmr_wait_for_ready, use_fake_sdl=use_fake_sdl)
        self._entrypoint = entrypoint

    def run(self):
        """
        This function should be called when the general Xapp is ready to start.
        """
        self._entrypoint(self)

    # there is no need for stop currently here (base has, and nothing
    # special to do here)