ricxappframe/xapp_symptomdata.py

#       Copyright (c) 2022 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.
#
# Symptomdata collection is triggered from the trblmgr ricplt pod. This subsystem provides for xapp interface to subscribe the
# symptomdata collection via lwsd pod. When the symptomdata collection is triggered then the xapp gets the callback to collect
# the symptomdata.
#
# If the dynamic registration is needed, then the xapp needs to use the Symptomdata.subscribe(...) method to indicate symptomdata
# collection. In case the xapp is set to trblmgr config file then the registration is not needed.
#
# If the xapp has the internal data for symptomdata collection REST call response, it can use the helper methods getFileList and collect
# to get the needed files or readymade zipped package for reponse.
#
import os
import re
import time
import requests
import json
from requests.exceptions import HTTPError
from zipfile import ZipFile
from threading import Timer
from datetime import datetime
from mdclogpy import Logger

logging = Logger(name=__name__)


class RepeatTimer(Timer):
    # timer class for housekeeping and file rotating
    def run(self):
        while not self.finished.wait(self.interval):
            self.function(*self.args, **self.kwargs)


class Symptomdata(object):
    # service is the local POD service id, path the temporal storage, host should be the trblmgr service name
    def __init__(self, service="", servicehost="", path="/tmp/", lwsduri=None, timeout=30):
        """
        init

        Parameters
        ----------
        service: string
            xapp service name
        servicehost: string
            xapp service host name
        path:
            temporal path where the symptomdata collection is stored
        lwsduri:
            lwsd uri for symptomdata dynamic registration
        timeout:
            timeout for subscription status polling
        """
        if not os.path.exists(path):
            os.mkdir(path)
        self.service = service
        self.servicehost = servicehost
        self.path = path
        self.lwsduri = lwsduri
        self.timeout = timeout
        # runtime attrs
        self.zipfilename = None
        logging.info("Symptomdata init service:%s path:%s lwsduri:%s timeout:%d" % (self.service, self.path, self.lwsduri, self.timeout))
        if self.lwsduri is not None:
            # do the subscription, set to True so that first the query is triggered
            self.lwsdok = True
            self.subscribe(args=("",))
            self.subscribetimer = RepeatTimer(self.timeout, self.subscribe, args=("",))
            self.subscribetimer.start()

    # make the symptomdata subscription query to lwsd - dynamic registration (needed if the static config in trblmgr does not have xapp service data)
    def subscribe(self, args):
        """
        subscribe
            internally used subscription function if the dynamic registration has been set
        """
        if self.lwsduri is not None:
            try:
                proxies = {"http": "", "https": ""}    # disable proxy usage
                headers = {'Content-type': 'application/json', 'Accept': 'application/json'}
                if self.lwsdok is False:
                    jsondata = json.dumps({'url': 'http://' + self.servicehost +
                                           ':8080/ric/v1/symptomdata', 'service': self.service, 'instance': '1'})
                    response = requests.post(self.lwsduri,
                                             data=jsondata,
                                             headers=headers,
                                             proxies=proxies)
                    logging.info("Symptomdata subscription success")
                    self.lwsdok = True
                elif self.lwsdok is True:
                    self.lwsdok = False
                    response = requests.get(self.lwsduri, headers=headers, proxies=proxies)
                    for item in response.json():
                        if item.get('service') == self.service:
                            logging.info("Symptomdata subscription request success")
                            self.lwsdok = True
                    if self.lwsdok is False:
                        logging.error("Symptomdata subscription missing")
                response.raise_for_status()
            except HTTPError as http_err:
                logging.error("Symptomdata subscription failed - http error : %s" % (http_err))
                self.lwsdok = False
            except Exception as err:
                logging.error("Symptomdata subscription failed - error : %s" % (err))
                self.lwsdok = False

    def stop(self):
        """
        stop
            stops the dynamic service registration/polling
        """
        if self.subscribetimer is not None:
            self.subscribetimer.cancel()

    def __del__(self):
        if self.subscribetimer is not None:
            self.subscribetimer.cancel()

    def getFileList(self, regex, fromtime, totime):
        """
        getFileList
            internal use only, get the matching files for collect method
        """
        fileList = []
        path, wc = regex.rsplit('/', 1)
        logging.info("Filtering path: %s using wildcard %s fromtime %d totime %d" % (path + '/', wc, fromtime, totime))
        try:
            for root, dirs, files in os.walk((path + '/')):
                for filename in files:
                    if re.match(wc, filename):
                        file_path = os.path.join(root, filename)
                        filest = os.stat(file_path)
                        if fromtime > 0:
                            logging.info("Filtering file time %d fromtime %d totime %d" % (filest.st_ctime, fromtime, totime))
                            if fromtime <= filest.st_ctime:
                                logging.info("Adding file time %d fromtime %d" % (filest.st_ctime, fromtime))
                                if totime > 0:
                                    if totime >= filest.st_ctime:
                                        fileList.append(file_path)
                                else:
                                    fileList.append(file_path)
                        elif totime > 0:
                            if totime >= filest.st_ctime:
                                logging.info("Filtering file time %d fromtime %d totime %d" % (filest.st_ctime, fromtime, totime))
                                fileList.append(file_path)
                        else:
                            fileList.append(file_path)

        except OSError as e:
            logging.error("System error %d" % (e.errno))
        return fileList

    def collect(self, zipfiletmpl, fileregexlist, fromtime, totime):
        """
        collect
            collects the symptomdata based on the file regular expression match and stored the symptomdata. Optionaly
            caller can use fromtime and totime to choose only files matching the access time

        Parameters
        ----------
        zipfiletmpl: string
            template for zip file name using the strftime format - ex: ``"symptomdata"+'-%Y-%m-%d-%H-%M-%S.zip'``
        fileregexlist: string array
            array for file matching - ex: ``('examples/*.csv',)``
        fromtime: integer
            time value seconds
        totime: integer
            time value seconds
        Returns
        -------
        string
            zipfile name
        """
        zipfilename = self.path + datetime.fromtimestamp(int(time.time())).strftime(zipfiletmpl)
        logging.info("Compressing files to symptomdata archive: %s" % (zipfilename))
        zipdata = ZipFile(zipfilename, "w")
        self.remove()
        self.zipfilename = None
        fileCnt = 0
        for fileregex in fileregexlist:
            logging.info("Compressing files using %s" % (fileregex))
            fileList = self.getFileList(fileregex, fromtime, totime)
            try:
                if len(fileList) > 0:
                    for file_path in fileList:
                        logging.info("Adding file %s to archive" % (file_path))
                        zipdata.write(file_path, file_path)
                        fileCnt += 1
            except OSError as e:
                logging.error("System error %d" % (e.errno))
        zipdata.close()
        if fileCnt > 0:
            self.zipfilename = zipfilename
        return self.zipfilename

    def read(self):
        """
        read
            reads the stored symptomdata file content

        Returns
        -------
        string
            zipfile name
        integer
            data lenght
        bytes
            bytes of the file data
        """
        data = None
        with open(self.zipfilename, 'rb') as file:
            data = file.read()
        return (self.zipfilename, len(data), data)

    def remove(self):
        if self.zipfilename is not None:
            os.remove(self.zipfilename)