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