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