examples/xapp_test.py

   1 #!/usr/bin/env python3
   2 # ==================================================================================
   3 #       Copyright (c) 2022 Nokia
   4 #
   5 #   Licensed under the Apache License, Version 2.0 (the "License");
   6 #   you may not use this file except in compliance with the License.
   7 #   You may obtain a copy of the License at
   8 #
   9 #          http://www.apache.org/licenses/LICENSE-2.0
  10 #
  11 #   Unless required by applicable law or agreed to in writing, software
  12 #   distributed under the License is distributed on an "AS IS" BASIS,
  13 #   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14 #   See the License for the specific language governing permissions and
  15 #   limitations under the License.
  16 # ==================================================================================
  17 import os
  18 import sys
  19 import time
  20 import json
  21 import logging
  22 import datetime
  23 import argparse
  24 import threading
  25 import http.server
  26 import signal
  27 import struct
  28 import socket
  29 import urllib.parse
  30 from io import open
  31 from time import gmtime, strftime
  32
  33 #sys.path.insert(0, os.path.abspath("./"))
  34 #sys.path.insert(0, os.path.abspath("./ricxappframe"))
  35 sys.path.append(os.getcwd())
  36 from ricxappframe.xapp_frame import RMRXapp, rmr
  37 from ricxappframe.xapp_sdl import SDLWrapper
  38 from ricxappframe.xapp_symptomdata import Symptomdata
  39 import ricxappframe.xapp_subscribe as subscribe
  40 import ricxappframe.xapp_rest as ricrest
  41 from mdclogpy import Logger, Level
  42
  43 class Config(object):
  44     def __init__(self, xapp_name, config_file):
  45         self.config_file = config_file
  46         self.xapp_name = xapp_name
  47         self.cfg = None
  48         self.keys = dict()
  49         self.config()
  50
  51     def config(self):
  52         with open(self.config_file, 'r') as file:
  53             cfg = file.read()
  54             if cfg != None:
  55                 self.cfg = json.loads(cfg)
  56                 if self.cfg is not None:
  57                     self.controls = self.cfg['controls']
  58
  59     def get_key_item(self, key):
  60         data = None
  61         if self.keys.get(key) is not None:
  62             data = self.keys[key]
  63         return data
  64
  65     def get_config(self):
  66         data = None
  67         with open(self.config_file, 'r') as file:
  68             cfg = file.read()
  69             if cfg != None:
  70                 self.cfg = json.loads(cfg)
  71                 # following is required by the appmgr -  don't know why.
  72                 cfgescaped = cfg.replace('"', '\\"').replace('\n', '\\n')
  73                 data = '[{ "config": "' + cfgescaped + '", "metadata":{"configType":"json","xappName":"' + self.xapp_name + '"}}]'
  74         if data == None:
  75             logging.error("Config file %s empty or does not exists" % (self.config_file))
  76         return data
  77
  78 def read_file(filename):
  79     try:
  80         with open(filename, 'r') as f:
  81             data = f.read()
  82             if len(data) == 0:
  83                 return None
  84             return data
  85     except IOError as error:
  86         return None
  87
  88
  89 class MyXapp(object):
  90
  91     def __init__(self, name, service, address, port, config, loglevel):
  92         signal.signal(signal.SIGQUIT, self.signal_handler)
  93         signal.signal(signal.SIGTERM, self.signal_handler)
  94         signal.signal(signal.SIGINT, self.signal_handler)
  95         self.running = False
  96
  97         self.configData = Config(name, config)
  98
  99         if self.configData.controls.get('logger'):
 100             level = self.configData.controls['logger'].get('level')
 101
 102         if port == None and self.configData.cfg['messaging'].get('ports'):
 103             for item in self.configData.cfg['messaging'].get('ports'):
 104                 if item['name'] == http:
 105                     port = int(item['port'])
 106         if address == None and self.configData.cfg.get('name'):
 107             address = self.configData.cfg.get('name')
 108
 109         # save the listen address and port for later use
 110         self.port = port
 111         self.address = address
 112
 113         self.logger = Logger(name, loglevel)
 114         # setup the symptomdata
 115         symptomCfg = self.GetSymptomConfig()
 116         self.symptomHndl = Symptomdata(service, name, "/tmp/", symptomCfg['url'], symptomCfg['timeout'])
 117
 118         # create the thread HTTP server and set the uri handler callbacks
 119         self.server = ricrest.ThreadedHTTPServer(address, port)
 120         # trick to get the own handler with defined
 121         self.server.handler.add_handler(self.server.handler, "GET", "config", "/ric/v1/config", self.configGetHandler)
 122         self.server.handler.add_handler(self.server.handler, "GET", "healthAlive", "/ric/v1/health/alive", self.healthyGetAliveHandler)
 123         self.server.handler.add_handler(self.server.handler, "GET", "healthReady", "/ric/v1/health/ready", self.healthyGetReadyHandler)
 124         self.server.handler.add_handler(self.server.handler, "GET", "symptomdata", "/ric/v1/symptomdata", self.symptomdataGetHandler)
 125         # start rest server
 126         self.server.start()
 127         # start RMR
 128         self.startRMR(service, 4)
 129         self.running = True
 130         # now we can subscribe
 131         self.Subscribe()
 132
 133     def startRMR(self, service, level):
 134         # handle the RMR_SEED_RT and RMR_RTG_SVC which is different in mcxapp
 135         data = None
 136         os.environ["RMR_SRC_ID"] = service
 137         os.environ["RMR_LOG_VLEVEL"] = str(level)
 138         os.environ["RMR_RTG_SVC"] = "4561"
 139         rmrseed = os.environ.get('RMR_SEED_RT')
 140         if rmrseed is not None:
 141             data = read_file(rmrseed)
 142             if data is None:
 143                 self.logger.warning("RMR seed file %s does not exists or is empty" % (rmrseed))
 144         else:
 145             self.logger.info("RMR_SEED_RT seed file not set in environment")
 146             data = read_file('uta-rtg.rt')
 147             if data is not None:
 148                 os.environ['RMR_SEED_RT'] = "./uta-rtg.rt"
 149                 self.logger.info("Setting the default RMR_SEED_RT=uta-rtg.rt - content:")
 150             else:
 151                 self.logger.info("Try to export the RMR_SEED_RT file if your RMR is not getting ready")
 152         self.rmrInit(b"4560")
 153
 154     def signal_handler(self, sig, frame):
 155         if self.running is True:
 156             self.server.stop()
 157             rmr.rmr_close(self.rmr_mrc)
 158         self.running = False
 159         sys.exit(0)
 160
 161     def rmrInit(self, initbind):
 162         # Init rmr
 163         self.rmr_mrc = rmr.rmr_init(initbind, rmr.RMR_MAX_RCV_BYTES, 0x00)
 164         while rmr.rmr_ready(self.rmr_mrc) == 0:
 165             time.sleep(1)
 166             self.logger.info('RMR not yet ready')
 167         rmr.rmr_set_stimeout(self.rmr_mrc, 1)
 168         rmr.rmr_set_vlevel(5)
 169         self.logger.info('RMR ready')
 170
 171     def GetSymptomConfig(self):
 172         if self.configData.cfg['controls'].get('symptomdata').get('lwsd'):
 173             return self.configData.cfg['controls'].get('symptomdata').get('lwsd')
 174
 175     def GetSubsConfig(self):
 176         if self.configData.cfg['controls'].get('subscription'):
 177             return self.configData.cfg['controls'].get('subscription')
 178
 179     def Subscribe(self):
 180         self.subsCfgDetail = self.GetSubsConfig()
 181         if self.subsCfgDetail != None:
 182             # this is example subscription, for your use case fill the attributes according to your needs
 183             self.subscriber = subscribe.NewSubscriber(self.subsCfgDetail['url'] + 'ric/v1')
 184             # add as well the own subscription response callback handler
 185             if self.subscriber.ResponseHandler(self.subsResponseCB, self.server) is not True:
 186                 self.logger.error("Error when trying to set the subscription reponse callback")
 187             # setup the subscription data
 188             subEndPoint = self.subscriber.SubscriptionParamsClientEndpoint(self.subsCfgDetail['clientEndpoint'], self.port, 4061)
 189             subsDirective = self.subscriber.SubscriptionParamsE2SubscriptionDirectives(10, 2, False)
 190             subsequentAction = self.subscriber.SubsequentAction("continue", "w10ms")
 191             actionDefinitionList = self.subscriber.ActionToBeSetup(1, "policy", (11,12,13,14,15), subsequentAction)
 192             subsDetail = self.subscriber.SubscriptionDetail(12110, (1,2,3,4,5), actionDefinitionList)
 193             # subscription data ready, make the subscription
 194             subObj = self.subscriber.SubscriptionParams("sub10", subEndPoint,"gnb123456",1231, subsDirective, subsDetail)
 195             self.logger.info("Sending the subscription to %s" %(self.subsCfgDetail['url'] + 'ric/v1'))
 196             self.logger.info(subObj.to_dict())
 197             # subscribe
 198             data, reason, status  = self.subscriber.Subscribe(subObj)
 199             # returns the json data, make it dictionary
 200             self.logger.info("Getting the subscription reponse")
 201             self.logger.info(json.loads(data))
 202
 203     def Unsubscribe(self):
 204         reason, status  = self.subscriber.UnSubscribe("ygwefwebw")
 205
 206     def QuerySubscribtions(self):
 207         data, reason, status  = self.subscriber.QuerySubscriptions()
 208
 209     def healthyGetReadyHandler(self, name, path, data, ctype):
 210         response = server.initResponse()
 211         response['payload'] = ("{'status': 'ready'}")
 212         return response
 213
 214     def healthyGetAliveHandler(self, name, path, data, ctype):
 215         response = server.initResponse()
 216         response['payload'] = ("{'status': 'alive'}")
 217         return response
 218
 219     def subsResponseCB(self, name, path, data, ctype):
 220         response = server.initResponse()
 221         response['payload'] = ("{}")
 222         return response
 223
 224     def getSymptomData(self, uriparams):
 225         paramlist = urllib.parse.parse_qs(uriparams)
 226         [x.upper() for x in paramlist]
 227         fromtime = 0
 228         totime = 0
 229         print(paramlist)
 230         if paramlist.get('fromTime'):
 231             fromtime = getSeconds(paramlist.get('fromTime')[0])
 232         if paramlist.get('toTime'):
 233             totime = getSeconds(paramlist.get('toTime')[0])
 234         zipfile = self.symptomHndl.collect("symptomdata"+'-%Y-%m-%d-%H-%M-%S.zip', ('examples/.*.py',), fromtime, totime)
 235         if zipfile != None:
 236             (zipfile, size, data) = self.symptomHndl.read()
 237             return (zipfile, size, data)
 238         return (None, 0, None)
 239
 240     def symptomdataGetHandler(self, name, path, data, ctype):
 241         reponse = ricrest.initResponse()
 242         (zipfile, size, filedata) = self.getSymptomData(self.path[20:])
 243         if filedata != None:
 244             reponse['payload'] = filedata
 245             reponse['ctype'] = 'application/zip'
 246             reponse['attachment'] = "symptomdata.zip"
 247             reponse['mode'] = 'binary'
 248             return reponse
 249         logging.error("Symptom data does not exists")
 250         reponse['response'] = 'System error - symptomdata does not exists'
 251         reponse['status'] = 500
 252         return reponse
 253
 254     def configGetHandler(self, name, path, data, ctype):
 255         response = server.initResponse()
 256         response['payload'] = (self.configData.get_config())
 257         return response
 258
 259 def removeEnvVar(evar):
 260     if evar in os.environ:
 261         del os.environ[evar]
 262
 263 def main():
 264     parser = argparse.ArgumentParser()
 265     parser.add_argument('-port', dest='port', help='HTTP server listen port, default 8088', required=False, type=int)
 266     parser.add_argument('-address', dest='address', help='IP listen address, default all interfaces', required=False, type=str)
 267     parser.add_argument('-config', dest='config', help='config file path name, default /opt/ric/config/config.json', required=False, type=str)
 268     parser.add_argument('-xapp', dest='xapp', help='xapp name', required=True, type=str)
 269     parser.add_argument('-service', dest='service', help='xapp service name (same as pod host name)', required=True, type=str)
 270     parser.add_argument('-verbose', dest='verbose', help='verbose logging level', required=False, type=int)
 271
 272     args = parser.parse_args()
 273
 274     if args.port is None:
 275         args.port = 8088
 276     if args.address is None:
 277         args.address = "0.0.0.0"
 278     if args.config is None:
 279         args.config = '/opt/ric/config/config.json'
 280
 281     # remove proxy so that it won't impact to rest calls
 282     removeEnvVar('HTTPS_PROXY')
 283     removeEnvVar('HTTP_PROXY')
 284     removeEnvVar('https_proxy')
 285     removeEnvVar('http_proxy')
 286
 287     # starting argument option will overwrite the config settings
 288     if args.verbose is None:
 289         args.verbose = 2
 290
 291     loglevel = Level.INFO
 292     if args.verbose == 0:
 293         loglevel = Level.ERROR
 294     if args.verbose == 1:
 295         loglevel = Level.WARNING
 296     elif args.verbose == 2:
 297         loglevel = Level.INFO
 298     elif args.verbose >= 3:
 299         loglevel = Level.DEBUG
 300
 301     myxapp = MyXapp(args.xapp, args.service, args.address, args.port, args.config, loglevel)
 302
 303     while True:
 304         print("Waiting for a message, will timeout after 10s")
 305         rmr_sbuf = rmr.rmr_torcv_msg(myxapp.rmr_mrc, None, 10000)
 306         summary = rmr.message_summary(rmr_sbuf)
 307         if summary[rmr.RMR_MS_MSG_STATE] == 12:
 308             print("Nothing received")
 309         else:
 310             print("Message received!: {}".format(summary))
 311             data = rmr.get_payload(rmr_sbuf)
 312         rmr.rmr_free_msg(rmr_sbuf)
 313
 314 if __name__ == '__main__':
 315     main()
 316
 317