geni/slicemgr.py

   1 import os
   2 import sys
   3 import datetime
   4 import time
   5
   6 from geni.util.geniserver import *
   7 from geni.util.geniclient import *
   8 from geni.util.cert import *
   9 from geni.util.credential import Credential
  10 from geni.util.trustedroot import *
  11 from geni.util.excep import *
  12 from geni.util.misc import *
  13 from geni.util.config import *
  14 from geni.util.rspec import Rspec
  15 from geni.util.specdict import *
  16 from geni.util.storage import SimpleStorage, XmlStorage
  17
  18 class SliceMgr(GeniServer):
  19
  20     hrn = None
  21     nodes_ttl = None
  22     nodes = None
  23     slices = None
  24     policy = None
  25     aggregates = None
  26     timestamp = None
  27     threshold = None
  28     shell = None
  29     registry = None
  30     key_file = None
  31     cert_file = None
  32     credential = None
  33
  34     ##
  35     # Create a new slice manager object.
  36     #
  37     # @param ip the ip address to listen on
  38     # @param port the port to listen on
  39     # @param key_file private key filename of registry
  40     # @param cert_file certificate filename containing public key (could be a GID file)
  41
  42     def __init__(self, ip, port, key_file, cert_file, config = "/usr/share/geniwrapper/geni/util/geni_config"):
  43         GeniServer.__init__(self, ip, port, key_file, cert_file)
  44         self.key_file = key_file
  45         self.cert_file = cert_file
  46         self.config = Config(config)
  47         self.basedir = self.config.GENI_BASE_DIR + os.sep
  48         self.server_basedir = self.basedir + os.sep + "geni" + os.sep
  49         self.hrn = self.config.GENI_INTERFACE_HRN
  50         self.time_format = "%Y-%m-%d %H:%M:%S"
  51
  52         # Get list of aggregates this sm talks to
  53         aggregates_file = self.server_basedir + os.sep + 'aggregates.xml'
  54         self.aggregate_info = XmlStorage(aggregates_file, {'aggregates': {'aggregate': []}} )
  55         self.aggregate_info.load()
  56
  57         # Get cached list of nodes (rspec)
  58         nodes_file = os.sep.join([self.server_basedir, 'smgr.' + self.hrn + '.components'])
  59         self.nodes = SimpleStorage(nodes_file)
  60         self.nodes.load()
  61
  62         # Get cacheds slice states
  63         slices_file = os.sep.join([self.server_basedir, 'smgr.' + self.hrn + '.slices'])
  64         self.slices = SimpleStorage(slices_file)
  65         self.slices.load()
  66
  67         # Get the policy
  68         policy_file = os.sep.join([self.server_basedir, 'smgr.' + self.hrn + '.policy'])
  69         self.policy = SimpleStorage(policy_file, {'whitelist': [], 'blacklist': []})
  70         self.policy.load()
  71
  72         # How long before we refresh nodes cache
  73         self.nodes_ttl = 1
  74
  75         self.connectRegistry()
  76         self.loadCredential()
  77         self.connectAggregates()
  78
  79
  80     def loadCredential(self):
  81         """
  82         Attempt to load credential from file if it exists. If it doesnt get
  83         credential from registry.
  84         """
  85
  86         # see if this file exists
  87         ma_cred_filename = self.server_basedir + os.sep + "smgr." + self.hrn + ".sa.cred"
  88         try:
  89             self.credential = Credential(filename = ma_cred_filename)
  90         except IOError:
  91             self.credential = self.getCredentialFromRegistry()
  92
  93
  94     def getCredentialFromRegistry(self):
  95         """
  96         Get our current credential from the registry.
  97         """
  98         # get self credential
  99         self_cred_filename = self.server_basedir + os.sep + "smgr." + self.hrn + ".cred"
 100         self_cred = self.registry.get_credential(None, 'ma', self.hrn)
 101         self_cred.save_to_file(self_cred_filename, save_parents=True)
 102
 103         # get ma credential
 104         ma_cred_filename = self.server_basedir + os.sep + "smgr." + self.hrn + ".sa.cred"
 105         ma_cred = self.registry.get_credential(self_cred, 'sa', self.hrn)
 106         ma_cred.save_to_file(ma_cred_filename, save_parents=True)
 107         return ma_cred
 108
 109     def connectRegistry(self):
 110         """
 111         Connect to the registry
 112         """
 113         # connect to registry using GeniClient
 114         address = self.config.GENI_REGISTRY_HOSTNAME
 115         port = self.config.GENI_REGISTRY_PORT
 116         url = 'http://%(address)s:%(port)s' % locals()
 117         self.registry = GeniClient(url, self.key_file, self.cert_file)
 118
 119     def connectAggregates(self):
 120         """
 121         Get connection details for the trusted peer aggregates from file and
 122         create a GeniClient connection to each.
 123         """
 124         self.aggregates = {}
 125         aggregates = self.aggregate_info['aggregates']['aggregate']
 126         if isinstance(aggregates, dict):
 127             aggregates = [aggregates]
 128         if isinstance(aggregates, list):
 129             for aggregate in aggregates:
 130                 # create xmlrpc connection using GeniClient
 131                 hrn, address, port = aggregate['hrn'], aggregate['addr'], aggregate['port']
 132                 url = 'http://%(address)s:%(port)s' % locals()
 133                 self.aggregates[hrn] = GeniClient(url, self.key_file, self.cert_file)
 134
 135     def item_hrns(self, items):
 136         """
 137         Take a list of items (components or slices) and return a dictionary where
 138         the key is the authoritative hrn and the value is a list of items at that
 139         hrn.
 140         """
 141         item_hrns = {}
 142         agg_hrns = self.aggregates.keys()
 143         for agg_hrn in agg_hrns:
 144             item_hrns[agg_hrn] = []
 145         for item in items:
 146             for agg_hrn in agg_hrns:
 147                 if item.startswith(agg_hrn):
 148                     item_hrns[agg_hrn] = item
 149
 150         return item_hrns
 151
 152
 153     def hostname_to_hrn(self, login_base, hostname):
 154         """
 155         Convert hrn to plantelab name.
 156         """
 157         genihostname = "_".join(hostname.split("."))
 158         return ".".join([self.hrn, login_base, genihostname])
 159
 160     def slicename_to_hrn(self, slicename):
 161         """
 162         Convert hrn to planetlab name.
 163         """
 164         slicename = slicename.replace("_", ".")
 165         return ".".join([self.hrn, slicename])
 166
 167     def refresh_components(self):
 168         """
 169         Update the cached list of nodes.
 170         """
 171
 172         # convert and threshold to ints
 173         if self.nodes.has_key('timestamp') and self.nodes['timestamp']:
 174             hr_timestamp = self.nodes['timestamp']
 175             timestamp = datetime.datetime.fromtimestamp(time.mktime(time.strptime(hr_timestamp, self.time_format)))
 176             hr_threshold = self.nodes['threshold']
 177             threshold = datetime.datetime.fromtimestamp(time.mktime(time.strptime(hr_threshold, self.time_format)))
 178         else:
 179             timestamp = datetime.datetime.now()
 180             hr_timestamp = timestamp.strftime(self.time_format)
 181             delta = datetime.timedelta(hours=self.nodes_ttl)
 182             threshold = timestamp + delta
 183             hr_threshold = threshold.strftime(self.time_format)
 184
 185         start_time = int(timestamp.strftime("%s"))
 186         end_time = int(threshold.strftime("%s"))
 187         duration = end_time - start_time
 188
 189         aggregates = self.aggregates.keys()
 190         rspecs = {}
 191         networks = []
 192         rspec = Rspec()
 193         for aggregate in aggregates:
 194             try:
 195                 # get the rspec from the aggregate
 196                 agg_rspec = self.aggregates[aggregate].list_nodes(self.credential)
 197                 # extract the netspec from each aggregates rspec
 198                 rspec.parseString(agg_rspec)
 199                 networks.extend([{'NetSpec': rspec.getDictsByTagName('NetSpec')}])
 200             except:
 201                 # XX print out to some error log
 202                 print "Error calling list nodes at aggregate %s" % aggregate
 203                 raise
 204
 205         # create the rspec dict
 206         resources = {'networks': networks, 'start_time': start_time, 'duration': duration}
 207         resourceDict = {'Rspec': resources}
 208         # convert rspec dict to xml
 209         rspec.parseDict(resourceDict)
 210
 211         # filter according to policy
 212         rspec.filter('NodeSpec', 'name', blacklist=self.policy['blacklist'], whitelist=self.policy['whitelist'])
 213
 214         # update timestamp and threshold
 215         timestamp = datetime.datetime.now()
 216         hr_timestamp = timestamp.strftime(self.time_format)
 217         delta = datetime.timedelta(hours=self.nodes_ttl)
 218         threshold = timestamp + delta
 219         hr_threshold = threshold.strftime(self.time_format)
 220
 221         nodedict = {'rspec': rspec.toxml(),
 222                     'timestamp': hr_timestamp,
 223                     'threshold':  hr_threshold}
 224
 225         self.nodes = SimpleStorage(self.nodes.db_filename, nodedict)
 226         self.nodes.write()
 227
 228     def load_policy(self):
 229         """
 230         Read the list of blacklisted and whitelisted nodes.
 231         """
 232         self.policy.load()
 233
 234     def load_slices(self):
 235         """
 236         Read current slice instantiation states.
 237         """
 238         self.slices.load()
 239
 240
 241     def getNodes(self, format = 'rspec'):
 242         """
 243         Return a list of components managed by this slice manager.
 244         """
 245         # Reload components list
 246         if not self.nodes.has_key('threshold') or not self.nodes['threshold'] or not self.nodes.has_key('timestamp') or not self.nodes['timestamp']:
 247             self.refresh_components()
 248         else:
 249             now = datetime.datetime.now()
 250             threshold = datetime.datetime.fromtimestamp(time.mktime(time.strptime(self.nodes['threshold'], self.time_format)))
 251             if  now > threshold:
 252                 self.refresh_components()
 253         return self.nodes[format]
 254
 255
 256     def getSlices(self):
 257         """
 258         Return a list of instnatiated managed by this slice manager.
 259         """
 260         slice_hrns = []
 261         for aggregate in self.aggregates:
 262             try:
 263                 slices = self.aggregates[aggregate].list_slices(self.credential)
 264                 slice_hrns.extend(slices)
 265             except:
 266                 raise
 267                 # print to some error log
 268                 pass
 269
 270         return slice_hrns
 271
 272     def getResources(self, slice_hrn):
 273         """
 274         Return the current rspec for the specified slice.
 275         """
 276
 277         # request this slices state from all known aggregates
 278         rspec = Rspec()
 279         rspecdicts = []
 280         networks = []
 281         for hrn in self.aggregates.keys():
 282             # check if the slice has resources at this hrn
 283             slice_resources = self.aggregates[hrn].get_resources(self.credential, slice_hrn)
 284             rspec.parseString(slice_resources)
 285             networks.extend({'NetSpec': rspec.getDictsByTagName('NetSpec')})
 286
 287         # merge all these rspecs into one
 288         start_time = int(datetime.datetime.now().strftime("%s"))
 289         end_time = start_time
 290         duration = end_time - start_time
 291
 292         resources = {'networks': networks, 'start_time': start_time, 'duration': duration}
 293         resourceDict = {'Rspec': resources}
 294         # convert rspec dict to xml
 295         rspec.parseDict(resourceDict)
 296         # save this slices resources
 297         #self.slices[slice_hrn] = rspec.toxml()
 298         #self.slices.write()
 299
 300         return rspec.toxml()
 301
 302     def createSlice(self, cred, slice_hrn, rspec):
 303         """
 304         Instantiate the specified slice according to whats defined in the rspec.
 305         """
 306
 307         # save slice state locally
 308         # we can assume that spec object has been validated so its safer to
 309         # save this instead of the unvalidated rspec the user gave us
 310         rspec = Rspec()
 311         tempspec = Rspec()
 312         rspec.parseString(rspec)
 313
 314         self.slices[slice_hrn] = rspec.toxml()
 315         self.slices.write()
 316
 317         # extract network list from the rspec and create a separate
 318         # rspec for each network
 319         slicename = self.hrn_to_plcslicename(slice_hrn)
 320         specDict = rspec.toDict()
 321         start_time = specDict['start_time']
 322         end_time = specDict['end_time']
 323
 324         rspecs = {}
 325         # only attempt to extract information about the aggregates we know about
 326         for hrn in self.aggregates.keys():
 327             netspec = spec.getDictByTagNameValue('NetSpec', hrn)
 328             if netspec:
 329                 # creat a plc dict
 330                 resources = {'start_time': star_time, 'end_time': end_time, 'networks': netspec}
 331                 resourceDict = {'Rspec': resources}
 332                 tempspec.parseDict(resourceDict)
 333                 rspecs[hrn] = tempspec.toxml()
 334
 335         # notify the aggregates
 336         for hrn in self.rspecs.keys():
 337             self.aggregates[hrn].createSlice(self.credential, rspecs[hrn])
 338
 339         return 1
 340
 341     def updateSlice(self, slice_hrn, rspec, attributes = []):
 342         """
 343         Update the specifed slice
 344         """
 345         self.create_slice(slice_hrn, rspec, attributes)
 346
 347     def deleteSlice_(self, slice_hrn):
 348         """
 349         Remove this slice from all components it was previouly associated with and
 350         free up the resources it was using.
 351         """
 352         # XX need to get the correct credential
 353         cred = self.credential
 354
 355         if self.slices.has_key(slice_hrn):
 356             self.slices.pop(slice_hrn)
 357             self.slices.write()
 358
 359         for hrn in self.aggregates.keys():
 360             self.aggregates[hrn].deleteSlice(cred, slice_hrn)
 361
 362         return 1
 363
 364     def startSlice(self, slice_hrn):
 365         """
 366         Stop the slice at plc.
 367         """
 368         cred = self.credential
 369
 370         for hrn in self.aggregates.keys():
 371             self.aggregates[hrn].startSlice(cred, slice_hrn)
 372         return 1
 373
 374     def stopSlice(self, slice_hrn):
 375         """
 376         Stop the slice at plc
 377         """
 378         cred = self.credential
 379         for hrn in self.aggregates.keys():
 380             self.aggregates[hrn].startSlice(cred, slice_hrn)
 381         return 1
 382
 383     def resetSlice(self, slice_hrn):
 384         """
 385         Reset the slice
 386         """
 387         # XX not yet implemented
 388         return 1
 389
 390     def getPolicy(self):
 391         """
 392         Return the policy of this slice manager.
 393         """
 394
 395         return self.policy
 396
 397
 398
 399 ##############################
 400 ## Server methods here for now
 401 ##############################
 402
 403     def list_nodes(self, cred):
 404         self.decode_authentication(cred, 'listnodes')
 405         return self.getNodes()
 406
 407     def list_slices(self, cred):
 408         self.decode_authentication(cred, 'listslices')
 409         return self.getSlices()
 410
 411     def get_resources(self, cred, hrn):
 412         self.decode_authentication(cred, 'listnodes')
 413         return self.getResources(hrn)
 414
 415     def get_ticket(self, cred, hrn, rspec):
 416         self.decode_authentication(cred, 'getticket')
 417         return self.getTicket(hrn, rspec)
 418
 419     def get_policy(self, cred):
 420         self.decode_authentication(cred, 'getpolicy')
 421         return self.getPolicy()
 422
 423     def create_slice(self, cred, hrn, rspec):
 424         self.decode_authentication(cred, 'creatslice')
 425         return self.createSlice(cred, hrn, rspec)
 426
 427     def delete_slice(self, cred, hrn):
 428         self.decode_authentication(cred, 'deleteslice')
 429         return self.deleteSlice(hrn)
 430
 431     def start_slice(self, cred, hrn):
 432         self.decode_authentication(cred, 'startslice')
 433         return self.startSlice(hrn)
 434
 435     def stop_slice(self, cred, hrn):
 436         self.decode_authentication(cred, 'stopslice')
 437         return self.stopSlice(hrn)
 438
 439     def reset_slice(self, cred, hrn):
 440         self.decode_authentication(cred, 'resetslice')
 441         return self.resetSlice(hrn)
 442
 443     def register_functions(self):
 444         GeniServer.register_functions(self)
 445
 446         # Aggregate interface methods
 447         self.server.register_function(self.list_nodes)
 448         self.server.register_function(self.list_slices)
 449         self.server.register_function(self.get_resources)
 450         self.server.register_function(self.get_policy)
 451         self.server.register_function(self.create_slice)
 452         self.server.register_function(self.delete_slice)
 453         self.server.register_function(self.start_slice)
 454         self.server.register_function(self.stop_slice)
 455         self.server.register_function(self.reset_slice)
 456