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, glob, threading
  16 import getopt, socket
  17 import logging
  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
  70     ## End of retrieval
  71
  72     # Any inspection code goes here
  73
  74     ## End of inspection
  75
  76     # Call NB API
  77     # If heat client, call heat command
  78
  79     # If ONOS client, run as shell script
  80
  81
  82     ## End of NB API call
  83
  84     # Marshall the response message for the Domino Server Fill
  85     push_r = PushResponseMessage()
  86     # Fill response message fields
  87     push_r.domino_udid = self.dominoClient.UDID
  88     push_r.seq_no = self.dominoClient.seqno
  89     push_r.responseCode = SUCCESS
  90     ## End of filling
  91
  92     self.dominoClient.seqno = self.dominoClient.seqno + 1
  93
  94     return push_r
  95
  96
  97   def openconnection(self):
  98     self.transport.open()
  99
 100   def closeconnection():
 101     self.transport.close()
 102
 103 class CLIHandler:
 104   def __init__(self):
 105     self.log = {}
 106
 107   def __init__(self, dominoclient, CLIservice):
 108     self.log = {}
 109     self.dominoClient = dominoclient
 110     self.CLIservice = CLIservice
 111
 112   def d_CLI(self, msg):
 113     logging.info('Received CLI %s', msg.CLI_input)
 114
 115     self.CLIservice.process_input(msg.CLI_input)
 116
 117     CLIrespmsg = CLIResponse()
 118     CLIrespmsg.CLI_response = "Testing..."
 119     return CLIrespmsg
 120
 121 def read_templatefile(temp_filename):
 122   f = open(temp_filename, 'r')
 123   lines = f.read().splitlines()
 124
 125   return lines
 126
 127 class DominoClientCLIService(threading.Thread):
 128   def __init__(self, dominoclient, communicationhandler, interactive):
 129     threading.Thread.__init__(self)
 130     self.dominoclient = dominoclient
 131     self.communicationhandler = communicationhandler
 132     self.interactive = interactive
 133
 134   def process_input(self, args):
 135
 136     try:
 137       if args[0] == 'heartbeat':
 138         self.dominoclient.heartbeat()
 139
 140       elif args[0] == 'publish':
 141         opts, args = getopt.getopt(args[1:],"t:",["tosca-file="])
 142         if len(opts) == 0:
 143           print '\nUsage: publish -t <toscafile>'
 144           return
 145
 146         for opt, arg in opts:
 147           if opt in ('-t', '--tosca-file'):
 148             toscafile = arg
 149
 150         self.dominoclient.publish(toscafile)
 151
 152       elif args[0] == 'subscribe':
 153         labels = []
 154         templateTypes = []
 155         opts, args = getopt.getopt(args[1:],"l:t:",["labels=","ttype="])
 156         for opt, arg in opts:
 157           if opt in ('-l', '--labels'):
 158             labels = labels + arg.split(',')
 159           elif opt in ('-t', '--ttype'):
 160             templateTypes = templateTypes + arg.split(',')
 161
 162         #check if labels or supported templates are nonempty
 163         if labels != [] or templateTypes != []:
 164           self.dominoclient.subscribe(labels, templateTypes)
 165
 166       elif args[0] == 'register':
 167         self.dominoclient.start()
 168
 169     except getopt.GetoptError:
 170       print 'Command is misentered or not supported!'
 171
 172
 173   def run(self):
 174     global DEFAULT_TOSCA_PUBFILE
 175     if self.interactive == "TRUE":
 176       flag = True
 177     else:
 178       flag = False
 179
 180     if flag: #interactive CLI, loop in while until killed
 181       while True:
 182          sys.stdout.write('>>')
 183          input_string = raw_input()
 184          args = input_string.split()
 185          if len(args) == 0:
 186            continue
 187
 188          sys.stdout.write('>>')
 189          #process input arguments
 190          self.process_input(args)
 191     else: #domino cli-client is used, listen for the CLI rpc calls
 192       cliHandler = CLIHandler(self.dominoclient, self)
 193       processor = DominoClientCLI.Processor(cliHandler)
 194       transport = TSocket.TServerSocket(port=self.dominoclient.CLIport)
 195       tfactory = TTransport.TBufferedTransportFactory()
 196       pfactory = TBinaryProtocol.TBinaryProtocolFactory()
 197       #Use TThreadedServer or TThreadPoolServer for a multithreaded server
 198       CLIServer = TServer.TSimpleServer(processor, transport, tfactory, pfactory)
 199       logging.debug('RPC service for CLI is starting...')
 200       CLIServer.serve()
 201
 202 class DominoClient:
 203   def __init__(self):
 204     self.communicationHandler = CommunicationHandler(self)
 205     self.processor = None
 206     self.transport = None
 207     self.tfactory = None
 208     self.pfactory = None
 209     self.communicationServer = None
 210
 211     self.CLIservice = None
 212
 213     self.serviceport = 9091
 214     self.dominoserver_IP = 'localhost'
 215
 216     self.CLIport = DOMINO_CLI_PORT
 217
 218     #Start from UNREGISTERED STATE
 219     #TO BE DONE: initialize from a saved state
 220     self.state = 'UNREGISTERED'
 221     self.seqno = 0
 222     self.UDID = 1
 223
 224   def start_communicationService(self):
 225     self.processor = Communication.Processor(self.communicationHandler)
 226     self.transport = TSocket.TServerSocket(port=int(self.serviceport))
 227     self.tfactory = TTransport.TBufferedTransportFactory()
 228     self.pfactory = TBinaryProtocol.TBinaryProtocolFactory()
 229     #Use TThreadedServer or TThreadPoolServer for a multithreaded server
 230     #self.communicationServer = TServer.TThreadedServer(self.processor, self.transport, self.tfactory, self.pfactory)
 231     self.communicationServer = TServer.TThreadPoolServer(self.processor, self.transport, self.tfactory, self.pfactory)
 232
 233     self.communicationServer.serve()
 234
 235   def start(self):
 236     try:
 237       self.communicationHandler.openconnection()
 238       self.register()
 239     except Thrift.TException, tx:
 240       print '%s' % (tx.message)
 241
 242   def register(self):
 243     if self.state == 'UNREGISTERED':
 244       logging.info('%d Sending Registration', self.UDID)
 245       #prepare registration message
 246       reg_msg = RegisterMessage()
 247       reg_msg.domino_udid_desired = UDID_DESIRED
 248       reg_msg.seq_no = self.seqno
 249       reg_msg.ipaddr = netutil.get_ip()
 250       reg_msg.tcpport = self.serviceport
 251       reg_msg.supported_templates = LIST_SUPPORTED_TEMPLATES
 252
 253       try:
 254         reg_msg_r = self.sender().d_register(reg_msg)
 255         logging.info('Registration Response: Response Code: %d'  , reg_msg_r.responseCode)
 256         if reg_msg_r.comments:
 257           logging.debug('Response Comments: %s' ,  reg_msg_r.comments)
 258
 259         if reg_msg_r.responseCode == SUCCESS:
 260           self.state = 'REGISTERED'
 261           self.UDID = reg_msg_r.domino_udid_assigned
 262         else:
 263           #Handle registration failure here (possibly based on reponse comments)
 264           pass
 265       except (Thrift.TException, TSocket.TTransportException) as tx:
 266         logging.error('%s' , tx.message)
 267       except (socket.timeout) as tx:
 268         self.dominoclient.handle_RPC_timeout(pub_msg)
 269       except (socket.error) as tx:
 270         logging.error('%s' , tx.message)
 271       self.seqno = self.seqno + 1
 272
 273   def heartbeat(self):
 274     if self.state == 'UNREGISTERED':
 275       self.start()
 276
 277     logging.info('%d Sending heartbeat', self.UDID)
 278     hbm = HeartBeatMessage()
 279     hbm.domino_udid = self.UDID
 280     hbm.seq_no = self.seqno
 281
 282     try:
 283       hbm_r = self.sender().d_heartbeat(hbm)
 284       logging.info('heart beat received from: %d ,sequence number: %d' , hbm_r.domino_udid, hbm_r.seq_no)
 285     except (Thrift.TException, TSocket.TTransportException) as tx:
 286       logging.error('%s' , tx.message)
 287     except (socket.timeout) as tx:
 288       self.handle_RPC_timeout(hbm)
 289     except:
 290       logging.error('Unexpected error: %s', sys.exc_info()[0])
 291
 292     self.seqno = self.seqno + 1
 293
 294   def publish(self, toscafile):
 295     if self.state == 'UNREGISTERED':
 296       self.start()
 297
 298     logging.info('Publishing the template file: ' + toscafile)
 299     pub_msg = PublishMessage()
 300     pub_msg.domino_udid = self.UDID
 301     pub_msg.seq_no = self.seqno
 302     pub_msg.template_type = 'tosca-nfv-v1.0'
 303
 304     try:
 305       pub_msg.template = read_templatefile(toscafile)
 306     except IOError as e:
 307       logging.error('I/O error(%d): %s' , e.errno, e.strerror)
 308       return
 309     try:
 310       pub_msg_r = self.sender().d_publish(pub_msg)
 311       logging.info('Publish Response is received from: %d ,sequence number: %d', pub_msg_r.domino_udid, pub_msg_r.seq_no)
 312     except (Thrift.TException, TSocket.TTransportException) as tx:
 313       print '%s' % (tx.message)
 314     except (socket.timeout) as tx:
 315       self.handle_RPC_timeout(pub_msg)
 316
 317     self.seqno = self.seqno + 1
 318
 319   def subscribe(self, labels, templateTypes):
 320      if self.state == 'UNREGISTERED':
 321        self.start()
 322
 323      logging.info('subscribing labels %s and templates %s', labels, templateTypes)
 324      #send subscription message
 325      sub_msg = SubscribeMessage()
 326      sub_msg.domino_udid = self.UDID
 327      sub_msg.seq_no = self.seqno
 328      sub_msg.template_op = APPEND
 329      sub_msg.supported_template_types = templateTypes
 330      sub_msg.label_op = APPEND
 331      sub_msg.labels = labels
 332      try:
 333        sub_msg_r = self.sender().d_subscribe(sub_msg)
 334        logging.info('Subscribe Response is received from: %d ,sequence number: %d', sub_msg_r.domino_udid,sub_msg_r.seq_no)
 335      except (Thrift.TException, TSocket.TTransportException) as tx:
 336        logging.error('%s' , tx.message)
 337      except (socket.timeout) as tx:
 338        self.handle_RPC_timeout(sub_msg)
 339
 340      self.seqno = self.seqno + 1
 341
 342   def stop(self):
 343     try:
 344       self.communicationHandler.closeconnection()
 345     except Thrift.TException, tx:
 346       logging.error('%s' , tx.message)
 347
 348   def sender(self):
 349     return self.communicationHandler.sender
 350
 351   def startCLI(self, interactive):
 352     self.CLIservice = DominoClientCLIService(self, self.communicationHandler, interactive)
 353     logging.info('CLI Service is starting')
 354     self.CLIservice.start()
 355     #to wait until CLI service is finished
 356     #self.CLIservice.join()
 357
 358   def set_serviceport(self, port):
 359     self.serviceport = port
 360
 361   def set_CLIport(self, cliport):
 362     self.CLIport = cliport
 363
 364   def set_dominoserver_ipaddr(self, ipaddr):
 365     self.dominoserver_IP = ipaddr
 366
 367   def handle_RPC_timeout(self, RPCmessage):
 368     # TBD: handle each RPC timeout separately
 369     if RPCmessage.messageType == HEART_BEAT:
 370       logging.debug('RPC Timeout for message type: HEART_BEAT')
 371     elif RPCmessage.messageType == PUBLISH:
 372       logging.debug('RPC Timeout for message type: PUBLISH')
 373     elif RPCmessage.messageType == SUBSCRIBE:
 374       logging.debug('RPC Timeout for message type: SUBSCRIBE')
 375     elif RPCmessage.messageType == REGISTER:
 376       logging.debug('RPC Timeout for message type: REGISTER')
 377     elif RPCmessage.messageType == QUERY:
 378       logging.debug('RPC Timeout for message type: QUERY')
 379
 380 def main(argv):
 381   client = DominoClient()
 382   loglevel = 'WARNING'
 383   interactive = "FALSE"
 384   #process input arguments
 385   try:
 386       opts, args = getopt.getopt(argv,"hc:p:i:l:",["conf=","port=","ipaddr=","log=","iac=","cliport="])
 387   except getopt.GetoptError:
 388       print 'DominoClient.py -c/--conf <configfile> -p/--port <socketport> -i/--ipaddr <IPaddr> -l/--log <loglevel> --iac=true/false'
 389       sys.exit(2)
 390   for opt, arg in opts:
 391       if opt == '-h':
 392          print 'DominoClient.py -c/--conf <configfile> -p/--port <socketport> -i/--ipaddr <IPaddr> -l/--log <loglevel> --iac=true/false'
 393          sys.exit()
 394       elif opt in ("-c", "--conf"):
 395          configfile = arg
 396       elif opt in ("-p", "--port"):
 397          client.set_serviceport(int(arg))
 398       elif opt in ("-i", "--ipaddr"):
 399          client.set_dominoserver_ipaddr(arg)
 400       elif opt in ("-l", "--log"):
 401          loglevel = arg
 402       elif opt in ("--iac"):
 403          interactive = arg.upper()
 404       elif opt in ("--cliport"):
 405          client.set_CLIport(int(arg))
 406
 407   #Set logging level
 408   numeric_level = getattr(logging, loglevel.upper(), None)
 409   try:
 410     if not isinstance(numeric_level, int):
 411       raise ValueError('Invalid log level: %s' % loglevel)
 412     logging.basicConfig(filename=logfile,level=numeric_level, format='%(asctime)s %(message)s', datefmt='%m/%d/%Y %I:%M:%S %p')
 413   except ValueError, ex:
 414     print ex.message
 415     exit()
 416
 417   #The client is starting
 418   logging.debug('Domino Client Starting...')
 419   client.start()
 420   client.startCLI(interactive)
 421   client.start_communicationService()
 422
 423 if __name__ == "__main__":
 424    main(sys.argv[1:])
 425