DominoServer.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 import sys, os, glob, random, errno
  15 import getopt, socket
  16 import logging, json
  17 import sqlite3
  18 #sys.path.append('gen-py')
  19 #sys.path.insert(0, glob.glob('./lib/py/build/lib.*')[0])
  20 sys.path.insert(0, glob.glob('./lib')[0])
  21
  22
  23 from dominoRPC import Communication
  24 from dominoRPC.ttypes import *
  25 from dominoRPC.constants import *
  26
  27 from thrift import Thrift
  28 from thrift.transport import TSocket
  29 from thrift.transport import TTransport
  30 from thrift.protocol import TBinaryProtocol
  31 from thrift.server import TServer
  32
  33 from toscaparser.tosca_template import ToscaTemplate
  34 #from toscaparser.utils.gettextutils import _
  35 #import toscaparser.utils.urlutils
  36
  37 from mapper import *
  38 from partitioner import *
  39 from util import miscutil
  40
  41 #Load configuration parameters
  42 from domino_conf import *
  43
  44
  45 class CommunicationHandler:
  46   def __init__(self):
  47     self.log = {}
  48
  49   def __init__(self, dominoserver):
  50     self.log = {}
  51     self.dominoServer = dominoserver
  52     self.seqno = 0;
  53
  54   def openconnection(self, ipaddr, tcpport):
  55     try:
  56       # Make socket
  57       transport = TSocket.TSocket(ipaddr, tcpport)
  58       transport.setTimeout(THRIFT_RPC_TIMEOUT_MS)
  59       # Add buffering to compensate for slow raw sockets
  60       self.transport = TTransport.TBufferedTransport(transport)
  61       # Wrap in a protocol
  62       self.protocol = TBinaryProtocol.TBinaryProtocol(self.transport)
  63       # Create a client to use the protocol encoder
  64       self.sender = Communication.Client(self.protocol)
  65       self.transport.open()
  66     except Thrift.TException, tx:
  67       logging.error('%s' , tx.message)
  68
  69
  70
  71   def closeconnection(self):
  72     self.transport.close()
  73
  74   def push_template(self,template,ipaddr,tcpport):
  75     self.openconnection(ipaddr,tcpport)
  76     pushm = PushMessage()
  77     pushm.domino_udid = SERVER_UDID
  78     pushm.seq_no = self.seqno
  79     pushm.template_type = 'tosca-nfv-v1.0'
  80     pushm.template = template
  81     try:
  82       push_r = self.sender.d_push(pushm)
  83       logging.info('Push Response received from %d' , push_r.domino_udid)
  84     except (Thrift.TException, TSocket.TTransportException) as tx:
  85       logging.error('%s' , tx.message)
  86     except (socket.timeout) as tx:
  87       self.dominoServer.handle_RPC_timeout(pushm)
  88     except:
  89       logging.error('Unexpected error: %s', sys.exc_info()[0])
  90
  91     self.seqno = self.seqno + 1
  92
  93     self.closeconnection()
  94
  95   #Heartbeat from Domino Client is received
  96   #Actions:
  97   #     - Respond Back with a heartbeat
  98
  99   def d_heartbeat(self, hb_msg):
 100     global SERVER_UDID
 101     logging.info('heartbeat received from %d' , hb_msg.domino_udid)
 102
 103     hb_r = HeartBeatMessage()
 104     hb_r.domino_udid = SERVER_UDID
 105     hb_r.seq_no = self.seqno
 106
 107     self.seqno = self.seqno + 1
 108
 109     return hb_r
 110
 111   #Registration from Domino Client is received
 112   #Actions:
 113   #
 114   #       - Respond Back with Registration Response
 115   def d_register(self, reg_msg):
 116     global SERVER_UDID
 117
 118     #Prepare and send Registration Response
 119     reg_r = RegisterResponseMessage()
 120     logging.info('Registration Request received for UDID %d from IP: %s port: %d', reg_msg.domino_udid_desired, reg_msg.ipaddr, reg_msg.tcpport)
 121
 122
 123     reg_r.domino_udid_assigned = self.dominoServer.assign_udid(reg_msg.domino_udid_desired)
 124     reg_r.seq_no = self.seqno
 125     reg_r.domino_udid = SERVER_UDID
 126     #return unconditional success
 127     #To be implemented:
 128     #Define conditions for unsuccessful registration (e.g., unsupported mapping)
 129     reg_r.responseCode = SUCCESS
 130     #no need to send comments
 131     #To be implemented:
 132     #Logic for a new UDID assignment
 133
 134     self.seqno = self.seqno + 1
 135
 136     #commit to the database
 137     dbconn = sqlite3.connect(SERVER_DBFILE)
 138     c = dbconn.cursor()
 139     try:
 140       newrow = [(reg_r.domino_udid_assigned, reg_msg.ipaddr, reg_msg.tcpport, ','.join(reg_msg.supported_templates), reg_msg.seq_no),]
 141       c.executemany('INSERT INTO clients VALUES (?,?,?,?,?)',newrow)
 142     except sqlite3.OperationalError as ex:
 143       logging.error('Could not add the new registration record into %s for Domino Client %d :  %s', SERVER_DBFILE, reg_r.domino_udid_assigned, ex.message)
 144     except:
 145       logging.error('Could not add the new registration record into %s for Domino Client %d', SERVER_DBFILE, reg_r.domino_udid_assigned)
 146       logging.error('Unexpected error: %s', sys.exc_info()[0])
 147
 148     dbconn.commit()
 149     dbconn.close()
 150
 151     return reg_r
 152
 153
 154   #Subscription from Domino Client is received
 155   #Actions:
 156   #       - Save the templates  & labels
 157   #       - Respond Back with Subscription Response
 158   def d_subscribe(self, sub_msg):
 159     global SERVER_UDID, SERVER_SEQNO
 160     logging.info('Subscribe Request received from %d' , sub_msg.domino_udid)
 161
 162     if sub_msg.template_op == APPEND:
 163       if self.dominoServer.subscribed_templateformats.has_key(sub_msg.domino_udid):
 164         self.dominoServer.subscribed_templateformats[sub_msg.domino_udid].update(set(sub_msg.supported_template_types))
 165       else:
 166         self.dominoServer.subscribed_templateformats[sub_msg.domino_udid] = set(sub_msg.supported_template_types)
 167     elif sub_msg.template_op == OVERWRITE:
 168       self.dominoServer.subscribed_templateformats[sub_msg.domino_udid] = set(sub_msg.supported_template_types)
 169     elif sub_msg.template_op == DELETE:
 170       self.dominoServer.subscribed_templateformats[sub_msg.domino_udid].difference_update(set(sub_msg.supported_template_types))
 171
 172     if sub_msg.labels != []:
 173       if sub_msg.label_op == APPEND:
 174         logging.debug('APPENDING Labels...')
 175         if self.dominoServer.subscribed_labels.has_key(sub_msg.domino_udid):
 176           self.dominoServer.subscribed_labels[sub_msg.domino_udid].update(set(sub_msg.labels))
 177         else:
 178           self.dominoServer.subscribed_labels[sub_msg.domino_udid] = set(sub_msg.labels)
 179       elif sub_msg.label_op == OVERWRITE:
 180         logging.debug('OVERWRITING Labels...')
 181         self.dominoServer.subscribed_labels[sub_msg.domino_udid] = set(sub_msg.labels)
 182       elif sub_msg.label_op == DELETE:
 183         logging.debug('DELETING Labels...')
 184         self.dominoServer.subscribed_labels[sub_msg.domino_udid].difference_update(set(sub_msg.labels))
 185
 186     logging.debug('Supported Template: %s Supported Labels: %s' , self.dominoServer.subscribed_templateformats[sub_msg.domino_udid] , self.dominoServer.subscribed_labels[sub_msg.domino_udid])
 187
 188     #commit to the database
 189     dbconn = sqlite3.connect(SERVER_DBFILE)
 190     c = dbconn.cursor()
 191     newlabelset = self.dominoServer.subscribed_labels[sub_msg.domino_udid]
 192     try:
 193       c.execute("REPLACE INTO labels (udid, label_list) VALUES ({udid}, '{newvalue}')".\
 194                format(udid=sub_msg.domino_udid, newvalue=','.join(list(newlabelset)) ))
 195     except sqlite3.OperationalError as ex1:
 196       logging.error('Could not add the new labels to %s for Domino Client %d :  %s', SERVER_DBFILE, sub_msg.domino_udid, ex1.message)
 197     except:
 198       logging.error('Could not add the new labels to %s for Domino Client %d', SERVER_DBFILE, sub_msg.domino_udid)
 199       logging.error('Unexpected error: %s', sys.exc_info()[0])
 200
 201     dbconn.commit()
 202     dbconn.close()
 203
 204
 205     #Fill in the details
 206     sub_r = SubscribeResponseMessage()
 207     sub_r.domino_udid = SERVER_UDID
 208     sub_r.seq_no = self.seqno
 209     sub_r.responseCode = SUCCESS
 210     self.seqno = self.seqno + 1
 211
 212     return sub_r
 213
 214   #Template Publication from Domino Client is received
 215   #Actions:
 216   #       - Parse the template, perform mapping, partition the template
 217   #       - Launch Push service
 218   #       - Respond Back with Publication Response
 219   def d_publish(self, pub_msg):
 220     global SERVER_UDID, SERVER_SEQNO, TOSCADIR, TOSCA_DEFAULT_FNAME
 221     logging.info('Publish Request received from %d' , pub_msg.domino_udid)
 222     logging.debug(pub_msg.template)
 223
 224     # Save as file
 225     try:
 226       os.makedirs(TOSCADIR)
 227     except OSError as exception:
 228       if exception.errno == errno.EEXIST:
 229         logging.debug('ERRNO %d; %s exists. Creating: %s', exception.errno, TOSCADIR,  TOSCADIR+TOSCA_DEFAULT_FNAME)
 230       else:
 231         logging.error('Error occurred in creating %s. Err no: %d', exception.errno)
 232
 233     #Risking a race condition if another process is attempting to write to same file
 234     f = open(TOSCADIR+TOSCA_DEFAULT_FNAME, 'w')
 235     for item in pub_msg.template:
 236       print>>f, item
 237     f.close()
 238
 239     # Load tosca object from file into memory
 240     tosca = ToscaTemplate( TOSCADIR+TOSCA_DEFAULT_FNAME )
 241
 242     # Extract Labels
 243     node_labels = label.extract_labels( tosca )
 244     logging.debug('Node Labels: %s', node_labels)
 245
 246     # Map nodes in the template to resource domains
 247     site_map = label.map_nodes( self.dominoServer.subscribed_labels , node_labels )
 248     logging.debug('Site Maps: %s' , site_map)
 249
 250     # Select a site for each VNF
 251     node_site = label.select_site( site_map )
 252     logging.debug('Selected Sites: %s', node_site)
 253
 254     # Create per-domain Tosca files
 255     file_paths = partitioner.partition_tosca('./toscafiles/template1.yaml',node_site,tosca.tpl)
 256
 257     # Create list of translated template files
 258
 259     # Create work-flow
 260
 261     # Send domain templates to each domain agent/client
 262     # FOR NOW: send untranslated but partitioned tosca files to scheduled sites
 263     # TBD: read from work-flow
 264     for site in file_paths:
 265       domino_client_ip = self.dominoServer.registration_record[site].ipaddr
 266       domino_client_port = self.dominoServer.registration_record[site].tcpport
 267       self.push_template(miscutil.read_templatefile(file_paths[site]), domino_client_ip, domino_client_port)
 268
 269     #Fill in the details
 270     pub_r = PublishResponseMessage()
 271     pub_r.domino_udid = SERVER_UDID
 272     pub_r.seq_no = self.seqno
 273     pub_r.responseCode = SUCCESS
 274     self.seqno = self.seqno + 1
 275     return pub_r
 276
 277   #Query from Domino Client is received
 278   #Actions:
 279   #
 280   #       - Respond Back with Query Response
 281   def d_query(self, qu_msg):
 282     #Fill in the details
 283     qu_r = QueryResponseMessage()
 284
 285     return qu_r
 286
 287
 288 class DominoServer:
 289    def __init__(self):
 290      self.assignedUUIDs = list()
 291      self.subscribed_labels = dict()
 292      self.subscribed_templateformats = dict()
 293      self.registration_record = dict()
 294      self.communicationHandler = CommunicationHandler(self)
 295      self.processor = Communication.Processor(self.communicationHandler)
 296      self.transport = TSocket.TServerSocket(port=DOMINO_SERVER_PORT)
 297      self.tfactory = TTransport.TBufferedTransportFactory()
 298      self.pfactory = TBinaryProtocol.TBinaryProtocolFactory()
 299      #Use TThreadedServer or TThreadPoolServer for a multithreaded server
 300      #self.communicationServer = TServer.TThreadedServer(self.processor, self.transport, self.tfactory, self.pfactory)
 301      self.communicationServer = TServer.TThreadPoolServer(self.processor, self.transport, self.tfactory, self.pfactory)
 302
 303
 304    def start_communicationService(self):
 305      self.communicationServer.serve()
 306
 307    #For now assign the desired UDID
 308    #To be implemented:
 309    #Check if ID is already assigned and in use
 310    #If not assigned, assign it
 311    #If assigned, offer a new random id
 312    def assign_udid(self, udid_desired):
 313      if udid_desired in self.assignedUUIDs:
 314        new_udid = random.getrandbits(63)
 315        while new_udid in self.assignedUUIDs:
 316          new_udid = random.getrandbits(63)
 317
 318        self.assignedUUIDs.append(new_udid)
 319        return new_udid
 320      else:
 321        self.assignedUUIDs.append(udid_desired)
 322        return udid_desired
 323
 324    def handle_RPC_timeout(self, RPCmessage):
 325      if RPCmessage.messageType == PUSH:
 326       logging.debug('RPC Timeout for message type: PUSH')
 327       # TBD: handle each RPC timeout separately
 328
 329 def main(argv):
 330   server = DominoServer()
 331   loglevel = 'WARNING'
 332   #process input arguments
 333   try:
 334       opts, args = getopt.getopt(argv,"hc:l:",["conf=","log="])
 335   except getopt.GetoptError:
 336       print 'DominoServer.py -c/--conf <configfile> -l/--log <loglevel>'
 337       sys.exit(2)
 338   for opt, arg in opts:
 339       if opt == '-h':
 340          print 'DominoClient.py -c/--conf <configfile> -p/--port <socketport> -i/--ipaddr <IPaddr> -l/--log <loglevel>'
 341          sys.exit()
 342       elif opt in ("-c", "--conf"):
 343          configfile = arg
 344       elif opt in ("-l", "--log"):
 345          loglevel= arg
 346   #Set logging level
 347   numeric_level = getattr(logging, loglevel.upper(), None)
 348   try:
 349     if not isinstance(numeric_level, int):
 350       raise ValueError('Invalid log level: %s' % loglevel)
 351     logging.basicConfig(filename=logfile,level=numeric_level, format='%(asctime)s %(message)s', datefmt='%m/%d/%Y %I:%M:%S %p')
 352   except ValueError, ex:
 353     print ex.message
 354     sys.exit(2)
 355
 356   #start the database with schemas
 357   dbconn = sqlite3.connect(SERVER_DBFILE)
 358   c = dbconn.cursor()
 359   try:
 360     c.execute('''CREATE TABLE labels (udid INTEGER PRIMARY KEY, label_list TEXT)''')
 361   except sqlite3.OperationalError as ex:
 362     logging.debug('In database file %s, no table is created as %s', SERVER_DBFILE, ex.message)
 363
 364   try:
 365     c.execute('''CREATE TABLE clients (udid INTEGER PRIMARY KEY, ipaddr TEXT, tcpport INTEGER, templatetypes TEXT, seqno INTEGER)''')
 366   except sqlite3.OperationalError as ex:
 367     logging.debug('In database file %s, no table is created as %s', SERVER_DBFILE, ex.message)
 368
 369   dbconn.commit()
 370   dbconn.close()
 371
 372   logging.debug('Domino Server Starting...')
 373   server.start_communicationService()
 374   print 'done.'
 375
 376 if __name__ == "__main__":
 377    main(sys.argv[1:])