DominoClient.py

   1 #!/usr/bin/env python
   2
   3 #Copyright 2015 Open Platform for NFV Project, Inc. and its contributors
   4 #   Licensed under the Apache License, Version 2.0 (the "License");
   5 #   you may not use this file except in compliance with the License.
   6 #   You may obtain a copy of the License at
   7 #       http://www.apache.org/licenses/LICENSE-2.0
   8 #   Unless required by applicable law or agreed to in writing, software
   9 #   distributed under the License is distributed on an "AS IS" BASIS,
  10 #   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  11 #   See the License for the specific language governing permissions and
  12 #   limitations under the License.
  13
  14
  15 import sys, os, glob, threading
  16 import getopt, socket
  17 import logging, errno
  18
  19 #sys.path.append('gen-py')
  20 #sys.path.insert(0, glob.glob('./lib/py/build/lib.*')[0])
  21 sys.path.insert(0, glob.glob('./lib')[0])
  22
  23 from dominoRPC import Communication
  24 from dominoRPC.ttypes import *
  25 from dominoRPC.constants import *
  26
  27 from dominoCLI import DominoClientCLI
  28 from dominoCLI.ttypes import *
  29 from dominoCLI.constants import *
  30
  31 from thrift import Thrift
  32 from thrift.transport import TSocket
  33 from thrift.transport import TTransport
  34 from thrift.protocol import TBinaryProtocol
  35 from thrift.server import TServer
  36
  37 from util import *
  38
  39 #Load configuration parameters
  40 from domino_conf import *
  41
  42 class CommunicationHandler:
  43   def __init__(self):
  44     self.log = {}
  45
  46   def __init__(self, dominoclient):
  47     self.log = {}
  48     self.dominoClient = dominoclient
  49     try:
  50       # Make socket
  51       transport = TSocket.TSocket(DOMINO_SERVER_IP, DOMINO_SERVER_PORT)
  52       transport.setTimeout(THRIFT_RPC_TIMEOUT_MS)
  53       # Add buffering to compensate for slow raw sockets
  54       self.transport = TTransport.TBufferedTransport(transport)
  55       # Wrap in a protocol
  56       self.protocol = TBinaryProtocol.TBinaryProtocol(self.transport)
  57       # Create a client to use the protocol encoder
  58       self.sender = Communication.Client(self.protocol)
  59     except Thrift.TException, tx:
  60       logging.error('%s' , tx.message)
  61
  62   # Template Push from Domino Server is received
  63   # Actions:
  64   #       - Depending on Controller Domain, call API
  65   #       - Respond Back with Push Response
  66   def d_push(self, push_msg):
  67     logging.info('%d Received Template File', self.dominoClient.UDID)
  68     # Retrieve the template file
  69     try:
  70       os.makedirs(TOSCA_RX_DIR+str(self.dominoClient.UDID))
  71     except OSError as exception:
  72       if exception.errno == errno.EEXIST:
  73         logging.debug('IGNORING error: ERRNO %d; %s exists.', exception.errno, TOSCA_RX_DIR+str(self.dominoClient.UDID))
  74       else:
  75         logging.error('IGNORING error in creating %s. Err no: %d', exception.errno)
  76
  77     try:
  78       miscutil.write_templatefile(TOSCA_RX_DIR+str(self.dominoClient.UDID)+'/'+str(push_msg.seq_no)+'.yaml' , push_msg.template)
  79     except:
  80       logging.error('FAILED to write the pushed file: %s', sys.exc_info()[0])
  81       push_r = PushResponseMessage()
  82       # Fill response message fields
  83       push_r.domino_udid = self.dominoClient.UDID
  84       push_r.seq_no = self.dominoClient.seqno
  85       push_r.responseCode = FAILED
  86       self.dominoClient.seqno = self.dominoClient.seqno + 1
  87       return push_r# Any inspection code goes here
  88
  89     ## End of inspection
  90
  91     # Call NB API
  92     # If heat client, call heat command
  93
  94     # If ONOS client, run as shell script
  95
  96
  97     ## End of NB API call
  98
  99     # Marshall the response message for the Domino Server Fill
 100     push_r = PushResponseMessage()
 101     # Fill response message fields
 102     push_r.domino_udid = self.dominoClient.UDID
 103     push_r.seq_no = self.dominoClient.seqno
 104     push_r.responseCode = SUCCESS
 105     ## End of filling
 106
 107     self.dominoClient.seqno = self.dominoClient.seqno + 1
 108
 109     return push_r
 110
 111
 112   def openconnection(self):
 113     self.transport.open()
 114
 115   def closeconnection():
 116     self.transport.close()
 117
 118 class CLIHandler:
 119   def __init__(self):
 120     self.log = {}
 121
 122   def __init__(self, dominoclient, CLIservice):
 123     self.log = {}
 124     self.dominoClient = dominoclient
 125     self.CLIservice = CLIservice
 126
 127   def d_CLI(self, msg):
 128     logging.info('Received CLI %s', msg.CLI_input)
 129
 130     self.CLIservice.process_input(msg.CLI_input)
 131
 132     CLIrespmsg = CLIResponse()
 133     CLIrespmsg.CLI_response = "Testing..."
 134     return CLIrespmsg
 135
 136
 137 class DominoClientCLIService(threading.Thread):
 138   def __init__(self, dominoclient, communicationhandler, interactive):
 139     threading.Thread.__init__(self)
 140     self.dominoclient = dominoclient
 141     self.communicationhandler = communicationhandler
 142     self.interactive = interactive
 143
 144   def process_input(self, args):
 145     try:
 146       if args[0] == 'heartbeat':
 147         self.dominoclient.heartbeat()
 148
 149       elif args[0] == 'publish':
 150         opts, args = getopt.getopt(args[1:],"t:",["tosca-file="])
 151         if len(opts) == 0:
 152           print '\nUsage: publish -t <toscafile>'
 153           return
 154
 155         for opt, arg in opts:
 156           if opt in ('-t', '--tosca-file'):
 157             toscafile = arg
 158
 159         self.dominoclient.publish(toscafile)
 160
 161       elif args[0] == 'subscribe':
 162         labels = []
 163         templateTypes = []
 164         labelop = APPEND
 165         templateop = APPEND
 166         opts, args = getopt.getopt(args[1:],"l:t:",["labels=","ttype=","lop=","top="])
 167         for opt, arg in opts:
 168           if opt in ('-l', '--labels'):
 169             labels = labels + arg.split(',')
 170           elif opt in ('-t', '--ttype'):
 171             templateTypes = templateTypes + arg.split(',')
 172           elif opt in ('--lop'):
 173             try:
 174               labelop = str2enum[arg.upper()]
 175             except KeyError as ex:
 176               print '\nInvalid label option, pick one of: APPEND, OVERWRITE, DELETE'
 177               return
 178           elif opt in ('--top'):
 179             try:
 180               templateop = str2enum[arg.upper()]
 181             except KeyError as ex:
 182               print '\nInvalid label option, pick one of: APPEND, OVERWRITE, DELETE'
 183               return
 184
 185         #check if labels or supported templates are nonempty
 186         if labels != [] or templateTypes != []:
 187           self.dominoclient.subscribe(labels, templateTypes, labelop, templateop)
 188
 189       elif args[0] == 'register':
 190         self.dominoclient.start()
 191
 192     except getopt.GetoptError:
 193       print 'Command is misentered or not supported!'
 194
 195
 196   def run(self):
 197     global DEFAULT_TOSCA_PUBFILE
 198     if self.interactive == "TRUE":
 199       flag = True
 200     else:
 201       flag = False
 202
 203     if flag: #interactive CLI, loop in while until killed
 204       while True:
 205          sys.stdout.write('>>')
 206          input_string = raw_input()
 207          args = input_string.split()
 208          if len(args) == 0:
 209            continue
 210
 211          sys.stdout.write('>>')
 212          #process input arguments
 213          self.process_input(args)
 214     else: #domino cli-client is used, listen for the CLI rpc calls
 215       cliHandler = CLIHandler(self.dominoclient, self)
 216       processor = DominoClientCLI.Processor(cliHandler)
 217       transport = TSocket.TServerSocket(port=self.dominoclient.CLIport)
 218       tfactory = TTransport.TBufferedTransportFactory()
 219       pfactory = TBinaryProtocol.TBinaryProtocolFactory()
 220       #Use TThreadedServer or TThreadPoolServer for a multithreaded server
 221       CLIServer = TServer.TSimpleServer(processor, transport, tfactory, pfactory)
 222       logging.debug('RPC service for CLI is starting...')
 223       CLIServer.serve()
 224
 225 class DominoClient:
 226   def __init__(self):
 227     self.communicationHandler = CommunicationHandler(self)
 228     self.processor = None
 229     self.transport = None
 230     self.tfactory = None
 231     self.pfactory = None
 232     self.communicationServer = None
 233
 234     self.CLIservice = None
 235
 236     self.serviceport = 9091
 237     self.dominoserver_IP = 'localhost'
 238
 239     self.CLIport = DOMINO_CLI_PORT
 240
 241     #Start from UNREGISTERED STATE
 242     #TO BE DONE: initialize from a saved state
 243     self.state = 'UNREGISTERED'
 244     self.seqno = 0
 245     self.UDID = 1
 246
 247   def start_communicationService(self):
 248     self.processor = Communication.Processor(self.communicationHandler)
 249     self.transport = TSocket.TServerSocket(port=int(self.serviceport))
 250     self.tfactory = TTransport.TBufferedTransportFactory()
 251     self.pfactory = TBinaryProtocol.TBinaryProtocolFactory()
 252     #Use TThreadedServer or TThreadPoolServer for a multithreaded server
 253     #self.communicationServer = TServer.TThreadedServer(self.processor, self.transport, self.tfactory, self.pfactory)
 254     self.communicationServer = TServer.TThreadPoolServer(self.processor, self.transport, self.tfactory, self.pfactory)
 255
 256     self.communicationServer.serve()
 257
 258   def start(self):
 259     try:
 260       self.communicationHandler.openconnection()
 261       self.register()
 262     except Thrift.TException, tx:
 263       print '%s' % (tx.message)
 264
 265   def register(self):
 266     if self.state == 'UNREGISTERED':
 267       logging.info('%d Sending Registration', self.UDID)
 268       #prepare registration message
 269       reg_msg = RegisterMessage()
 270       reg_msg.domino_udid_desired = UDID_DESIRED
 271       reg_msg.seq_no = self.seqno
 272       reg_msg.ipaddr = netutil.get_ip()
 273       reg_msg.tcpport = self.serviceport
 274       reg_msg.supported_templates = LIST_SUPPORTED_TEMPLATES
 275
 276       try:
 277         reg_msg_r = self.sender().d_register(reg_msg)
 278         logging.info('Registration Response: Response Code: %d'  , reg_msg_r.responseCode)
 279         if reg_msg_r.comments:
 280           logging.debug('Response Comments: %s' ,  reg_msg_r.comments)
 281
 282         if reg_msg_r.responseCode == SUCCESS:
 283           self.state = 'REGISTERED'
 284           self.UDID = reg_msg_r.domino_udid_assigned
 285         else:
 286           #Handle registration failure here (possibly based on reponse comments)
 287           pass
 288       except (Thrift.TException, TSocket.TTransportException) as tx:
 289         logging.error('%s' , tx.message)
 290       except (socket.timeout) as tx:
 291         self.dominoclient.handle_RPC_timeout(pub_msg)
 292       except (socket.error) as tx:
 293         logging.error('%s' , tx.message)
 294       self.seqno = self.seqno + 1
 295
 296   def heartbeat(self):
 297     if self.state == 'UNREGISTERED':
 298       self.start()
 299
 300     logging.info('%d Sending heartbeat', self.UDID)
 301     hbm = HeartBeatMessage()
 302     hbm.domino_udid = self.UDID
 303     hbm.seq_no = self.seqno
 304
 305     try:
 306       hbm_r = self.sender().d_heartbeat(hbm)
 307       logging.info('heart beat received from: %d ,sequence number: %d' , hbm_r.domino_udid, hbm_r.seq_no)
 308     except (Thrift.TException, TSocket.TTransportException) as tx:
 309       logging.error('%s' , tx.message)
 310     except (socket.timeout) as tx:
 311       self.handle_RPC_timeout(hbm)
 312     except:
 313       logging.error('Unexpected error: %s', sys.exc_info()[0])
 314
 315     self.seqno = self.seqno + 1
 316
 317   def publish(self, toscafile):
 318     if self.state == 'UNREGISTERED':
 319       self.start()
 320
 321     logging.info('Publishing the template file: ' + toscafile)
 322     pub_msg = PublishMessage()
 323     pub_msg.domino_udid = self.UDID
 324     pub_msg.seq_no = self.seqno
 325     pub_msg.template_type = 'tosca-nfv-v1.0'
 326
 327     try:
 328       pub_msg.template = miscutil.read_templatefile(toscafile)
 329     except IOError as e:
 330       logging.error('I/O error(%d): %s' , e.errno, e.strerror)
 331       return
 332     try:
 333       pub_msg_r = self.sender().d_publish(pub_msg)
 334       logging.info('Publish Response is received from: %d ,sequence number: %d Op. Status: %d', pub_msg_r.domino_udid, pub_msg_r.seq_no, pub_msg_r.responseCode)
 335     except (Thrift.TException, TSocket.TTransportException) as tx:
 336       print '%s' % (tx.message)
 337     except (socket.timeout) as tx:
 338       self.handle_RPC_timeout(pub_msg)
 339
 340     self.seqno = self.seqno + 1
 341
 342   def subscribe(self, labels, templateTypes, label_op, template_op):
 343      if self.state == 'UNREGISTERED':
 344        self.start()
 345
 346      logging.info('subscribing labels %s and templates %s', labels, templateTypes)
 347      #send subscription message
 348      sub_msg = SubscribeMessage()
 349      sub_msg.domino_udid = self.UDID
 350      sub_msg.seq_no = self.seqno
 351      sub_msg.template_op = template_op
 352      sub_msg.supported_template_types = templateTypes
 353      sub_msg.label_op = label_op
 354      sub_msg.labels = labels
 355      try:
 356        sub_msg_r = self.sender().d_subscribe(sub_msg)
 357        logging.info('Subscribe Response is received from: %d ,sequence number: %d', sub_msg_r.domino_udid,sub_msg_r.seq_no)
 358      except (Thrift.TException, TSocket.TTransportException) as tx:
 359        logging.error('%s' , tx.message)
 360      except (socket.timeout) as tx:
 361        self.handle_RPC_timeout(sub_msg)
 362
 363      self.seqno = self.seqno + 1
 364
 365   def stop(self):
 366     try:
 367       self.communicationHandler.closeconnection()
 368     except Thrift.TException, tx:
 369       logging.error('%s' , tx.message)
 370
 371   def sender(self):
 372     return self.communicationHandler.sender
 373
 374   def startCLI(self, interactive):
 375     self.CLIservice = DominoClientCLIService(self, self.communicationHandler, interactive)
 376     logging.info('CLI Service is starting')
 377     self.CLIservice.start()
 378     #to wait until CLI service is finished
 379     #self.CLIservice.join()
 380
 381   def set_serviceport(self, port):
 382     self.serviceport = port
 383
 384   def set_CLIport(self, cliport):
 385     self.CLIport = cliport
 386
 387   def set_dominoserver_ipaddr(self, ipaddr):
 388     self.dominoserver_IP = ipaddr
 389
 390   def handle_RPC_timeout(self, RPCmessage):
 391     # TBD: handle each RPC timeout separately
 392     if RPCmessage.messageType == HEART_BEAT:
 393       logging.debug('RPC Timeout for message type: HEART_BEAT')
 394     elif RPCmessage.messageType == PUBLISH:
 395       logging.debug('RPC Timeout for message type: PUBLISH')
 396     elif RPCmessage.messageType == SUBSCRIBE:
 397       logging.debug('RPC Timeout for message type: SUBSCRIBE')
 398     elif RPCmessage.messageType == REGISTER:
 399       logging.debug('RPC Timeout for message type: REGISTER')
 400     elif RPCmessage.messageType == QUERY:
 401       logging.debug('RPC Timeout for message type: QUERY')
 402
 403 def main(argv):
 404   client = DominoClient()
 405   loglevel = LOGLEVEL
 406   interactive = INTERACTIVE
 407   #process input arguments
 408   try:
 409       opts, args = getopt.getopt(argv,"hc:p:i:l:",["conf=","port=","ipaddr=","log=","iac=","cliport="])
 410   except getopt.GetoptError:
 411       print 'DominoClient.py -c/--conf <configfile> -p/--port <socketport> -i/--ipaddr <IPaddr> -l/--log <loglevel> --iac=true/false'
 412       sys.exit(2)
 413   for opt, arg in opts:
 414       if opt == '-h':
 415          print 'DominoClient.py -c/--conf <configfile> -p/--port <socketport> -i/--ipaddr <IPaddr> -l/--log <loglevel> --iac=true/false'
 416          sys.exit()
 417       elif opt in ("-c", "--conf"):
 418          configfile = arg
 419       elif opt in ("-p", "--port"):
 420          client.set_serviceport(int(arg))
 421       elif opt in ("-i", "--ipaddr"):
 422          client.set_dominoserver_ipaddr(arg)
 423       elif opt in ("-l", "--log"):
 424          loglevel = arg
 425       elif opt in ("--iac"):
 426          interactive = arg.upper()
 427       elif opt in ("--cliport"):
 428          client.set_CLIport(int(arg))
 429
 430   #Set logging level
 431   numeric_level = getattr(logging, loglevel.upper(), None)
 432   try:
 433     if not isinstance(numeric_level, int):
 434       raise ValueError('Invalid log level: %s' % loglevel)
 435     logging.basicConfig(filename=logfile,level=numeric_level, format='%(asctime)s %(message)s', datefmt='%m/%d/%Y %I:%M:%S %p')
 436   except ValueError, ex:
 437     print ex.message
 438     exit()
 439
 440   #The client is starting
 441   logging.debug('Domino Client Starting...')
 442   client.start()
 443   client.startCLI(interactive)
 444   client.start_communicationService()
 445
 446 if __name__ == "__main__":
 447    main(sys.argv[1:])
 448