--- /dev/null
+# 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)
Code Review / ric-plt / xapp-frame-py.git / blobdiff