examples/xapp_symptomdata.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, '../ricxappframe/xapp-frame-py')
  34 from ricxappframe.xapp_frame import RMRXapp, rmr
  35 from ricxappframe.xapp_sdl import SDLWrapper
  36 from ricxappframe.xapp_symptomdata import Symptomdata
  37
  38 # rmr init mode - when set to port 4561 then will wait for the rtmgr to connect
  39 # otherwise will connect to rtmgr like set below
  40 RMR_INIT_SVC = b"4560"
  41 MRC = None
  42 xapp = None
  43
  44 def signal_handler(sig, frame):
  45     global server
  46     global MRC
  47
  48     server.stop()
  49     rmr.rmr_close(MRC)
  50     sys.exit(0)
  51
  52
  53 def RMR_init_xapp(initbind):
  54     global RMR_INIT_SVC
  55     # Init rmr
  56     MRC = mrc = rmr.rmr_init(initbind, rmr.RMR_MAX_RCV_BYTES, 0x00)
  57     while rmr.rmr_ready(mrc) == 0:
  58         time.sleep(1)
  59         print('[%d]::RMR not yet ready')
  60     rmr.rmr_set_stimeout(mrc, 1)
  61     sbuf = rmr.rmr_alloc_msg(mrc, 500)
  62     rmr.rmr_set_vlevel(5)
  63     print('[%d]::RMR ready')
  64     return mrc, sbuf
  65
  66 def read_file(filename):
  67     try:
  68         with open(filename, 'r') as f:
  69             data = f.read()
  70             if len(data) == 0:
  71                 return None
  72             return data
  73     except IOError as error:
  74         return None
  75
  76 def getSymptomData(symptomHndl, uriparams):
  77     paramlist = urllib.parse.parse_qs(uriparams)
  78     [x.upper() for x in paramlist]
  79     fromtime = 0
  80     totime = 0
  81     print(paramlist)
  82     if paramlist.get('fromTime'):
  83         fromtime = getSeconds(paramlist.get('fromTime')[0])
  84     if paramlist.get('toTime'):
  85         totime = getSeconds(paramlist.get('toTime')[0])
  86     zipfile = symptomHndl.collect("symptomdata"+'-%Y-%m-%d-%H-%M-%S.zip', ('examples/.*.py',), fromtime, totime)
  87     if zipfile != None:
  88         (zipfile, size, data) = symptomHndl.read()
  89         return (zipfile, size, data)
  90     return (None, 0, None)
  91
  92
  93 class RestHandler(http.server.BaseHTTPRequestHandler):
  94     # responds to http request according to the process status
  95     global symptomHndl
  96
  97     def _set_headers(self, status, length=0, ctype = 'application/json', attachment = None):
  98         self.send_response(status)
  99         self.send_header("Server-name", "XAPP REST SERVER 1.0")
 100         self.send_header('Content-type', ctype)
 101         if length != 0:
 102             self.send_header('Content-length', length)
 103         if attachment != None:
 104             self.send_header('Content-Disposition', "attachment; filename=" + attachment)
 105         self.end_headers()
 106
 107     def do_HEAD(self):
 108         self._set_headers()
 109
 110     def do_POST(self):
 111         try:
 112             logging.debug("POST %s" % (self.path))
 113         except (socket.error, IOError):
 114             pass
 115
 116     def do_DELETE(self):
 117         try:
 118             logging.debug("DELETE %s" % (self.path))
 119         except (socket.error, IOError):
 120             pass
 121
 122
 123     def do_GET(self):
 124         # default get handler
 125         try:
 126             data = None
 127             mode = 'plain'
 128             ctype = 'application/json'
 129             attachment = None
 130             if self.path == "/ric/v1/health/alive":
 131                 data = json.dumps({'status': 'alive'})
 132             elif self.path == "/ric/v1/health/ready":
 133                 data = json.dumps({'status': 'ready'})
 134             elif self.path.find("/ric/v1/symptomdata") >= 0:
 135                 (zipfile, size, data) = getSymptomData(symptomHndl, self.path[20:])
 136                 if data != None:
 137                     mode = 'binary'
 138                     ctype = 'application/zip'
 139                     attachment = "symptomdata.zip"
 140                 else:
 141                     logging.error("Symptom data does not exists")
 142                     self._set_headers(404, 0)
 143
 144             if data is not None:
 145                 length = len(data)
 146                 self._set_headers(200, length, ctype, attachment)
 147                 if mode == 'plain':
 148                     # ascii mode
 149                     self.wfile.write(data.encode('utf-8'))
 150                 else:
 151                     # binary mode
 152                     self.wfile.write(data)
 153             else:
 154                 logging.error("Unknown uri %s" % (self.path))
 155                 self._set_headers(404, 0)
 156         except (socket.error, IOError):
 157             pass
 158
 159 class ThreadedHTTPServer(object):
 160     handler = RestHandler
 161     server_class = http.server.HTTPServer
 162     def __init__(self, host, port):
 163         self.server = self.server_class((host, port), self.handler)
 164         self.server_thread = threading.Thread(target=self.server.serve_forever)
 165         self.server_thread.daemon = True
 166
 167     def start(self):
 168         self.server_thread.start()
 169
 170     def stop(self):
 171         self.server.socket.close()
 172         self.server.server_close()
 173         self.server.shutdown()
 174
 175 def main():
 176     global server
 177     global xapp
 178     global symptomHndl
 179
 180     # init the default values
 181     ADDRESS = "0.0.0.0"     # bind to all interfaces
 182     PORT = 8080             # web server listen port
 183
 184     parser = argparse.ArgumentParser()
 185     parser.add_argument('-port', dest='port', help='HTTP server listen port, default 3000', required=False, type=int)
 186     parser.add_argument('-address', dest='address', help='IP listen address, default all interfaces', required=False, type=str)
 187     parser.add_argument('-xapp', dest='xapp', help='xapp name', required=True, type=str)
 188     parser.add_argument('-service', dest='service', help='xapp service name (same as pod host name)', required=True, type=str)
 189     args = parser.parse_args()
 190
 191     if args.port is not None:
 192         PORT = args.port
 193     if args.address is not None:
 194         ADDRESS = args.address
 195
 196     # handle the RMR_SEED_RT and RMR_RTG_SVC which is different in mcxapp
 197     data = None
 198     os.environ["RMR_SRC_ID"] = args.service
 199     os.environ["RMR_LOG_VLEVEL"] = '4'
 200     os.environ["RMR_RTG_SVC"] = "4561"
 201     rmrseed = os.environ.get('RMR_SEED_RT')
 202     if rmrseed is not None:
 203         data = read_file(rmrseed)
 204         if data is None:
 205             print("RMR seed file %s does not exists or is empty" % (rmrseed))
 206     else:
 207         print("RMR_SEED_RT seed file not set in environment")
 208         data = read_file('uta-rtg.rt')
 209         if data is not None:
 210             os.environ['RMR_SEED_RT'] = "./uta-rtg.rt"
 211             print("Setting the default RMR_SEED_RT=uta-rtg.rt - content:")
 212             print(data)
 213         else:
 214             print("Try to export the RMR_SEED_RT file if your RMR is not getting ready")
 215
 216     symptomHndl = Symptomdata(args.service, args.xapp, "/tmp/", "http://service-ricplt-lwsd-http:8080/ric/v1/lwsd", 10)
 217
 218     # Start the threaded server, bind to address
 219     server = ThreadedHTTPServer(ADDRESS, PORT)
 220     server.start()
 221
 222     mrc, sbuf = RMR_init_xapp(b"4560")
 223
 224     while True:
 225         print("Waiting for a message, will timeout after 2000ms")
 226         sbuf = rmr.rmr_torcv_msg(mrc, None, 2000)
 227         summary = rmr.message_summary(sbuf)
 228         if summary[rmr.RMR_MS_MSG_STATE] == 12:
 229             print("Nothing received =(")
 230         else:
 231             print("Message received!: {}".format(summary))
 232             data = rmr.get_payload(sbuf)
 233         rmr.rmr_free_msg(sbuf)
 234
 235 if __name__ == '__main__':
 236     signal.signal(signal.SIGQUIT, signal_handler)
 237     signal.signal(signal.SIGTERM, signal_handler)
 238     signal.signal(signal.SIGINT, signal_handler)
 239     main()