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.trustedroot import *
  10 from geni.util.excep import *
  11 from geni.util.misc import *
  12 from geni.util.config import Config
  13 from geni.util.rspec import Rspec
  14 from geni.util.specdict import *
  15 from geni.util.storage import SimpleStorage
  16
  17 class SliceMgr(GeniServer):
  18
  19     hrn = None
  20     key_file = None
  21     cert_file = None
  22     nodes = {}
  23     slices = {}
  24     policy = {}
  25     aggregates = {}
  26     timestamp = None
  27     threshold = None
  28     shell = None
  29     registry = None
  30
  31
  32     ##
  33     # Create a new slice manager object.
  34     #
  35     # @param ip the ip address to listen on
  36     # @param port the port to listen on
  37     # @param key_file private key filename of registry
  38     # @param cert_file certificate filename containing public key (could be a GID file)
  39
  40     def __init__(self, ip, port, key_file, cert_file, config = "/usr/share/geniwrapper/util/geni_config"):
  41         GeniServer.__init__(ip, port, key_file, cert_file)
  42         self.key_file = key_file
  43         self.cert_file = cert_file
  44         self.conf = Config(config)
  45         basedir = self.conf.GENI_BASE_DIR + os.sep
  46         server_basedir = basedir + os.sep + "geni" + os.sep
  47         self.hrn = conf.GENI_INTERFACE_HRN
  48
  49         # Get list of aggregates this sm talks to
  50         aggregates_file = server_basedir + os.sep + 'aggregates'
  51         self.aggregates = SimpleStorage(aggregates_file)
  52         self.load_aggregates(aggregates_file)
  53
  54         components_file = os.sep.join([server_basedir, 'components', 'slicemgr.' + hrn + '.comp'])
  55
  56         self.slices_file = os.sep.join([server_basedir, 'components', 'slicemgr' + hrn + '.slices'])
  57         self.timestamp_file = os.sep.join([server_basedir, 'components', 'slicemgr' + hrn + '.timestamp'])
  58         self.components_ttl = components_ttl
  59         self.policy['whitelist'] = []
  60         self.policy['blacklist'] = []
  61         self.connect()
  62
  63     def load_aggregates(self, aggregates_file):
  64         """
  65         Get info about the aggregates available to us from file and create
  66         an xmlrpc connection to each. If any info is invalid, skip it.
  67         """
  68         lines = []
  69         try:
  70             f = open(aggregates_file, 'r')
  71             lines = f.readlines()
  72             f.close()
  73         except: raise
  74
  75         for line in lines:
  76             # Skip comments
  77             if line.strip.startswith("#"):
  78                 continue
  79             agg_info = line.split("\t").split(" ")
  80
  81             # skip invalid info
  82             if len(agg_info) != 3:
  83                 continue
  84
  85             # create xmlrpc connection using GeniClient
  86             hrn, address, port = agg_info[0], agg_info[1], agg_info[2]
  87             url = 'https://%(address)s:%(port)s' % locals()
  88             self.aggregates[hrn] = GeniClient(url, self.key_file, self.cert_file)
  89
  90
  91     def item_hrns(self, items):
  92         """
  93         Take a list of items (components or slices) and return a dictionary where
  94         the key is the authoritative hrn and the value is a list of items at that
  95         hrn.
  96         """
  97         item_hrns = {}
  98         agg_hrns = self.aggregates.keys()
  99         for agg_hrn in agg_hrns:
 100             item_hrns[agg_hrn] = []
 101         for item in items:
 102             for agg_hrn in agg_hrns:
 103                 if item.startswith(agg_hrn):
 104                     item_hrns[agg_hrn] = item
 105
 106         return item_hrns
 107
 108
 109     def hostname_to_hrn(self, login_base, hostname):
 110         """
 111         Convert hrn to plantelab name.
 112         """
 113         genihostname = "_".join(hostname.split("."))
 114         return ".".join([self.hrn, login_base, genihostname])
 115
 116     def slicename_to_hrn(self, slicename):
 117         """
 118         Convert hrn to planetlab name.
 119         """
 120         slicename = slicename.replace("_", ".")
 121         return ".".join([self.hrn, slicename])
 122
 123     def refresh_components(self):
 124         """
 125         Update the cached list of nodes.
 126         """
 127         print "refreshing"
 128
 129         aggregates = self.aggregates.keys()
 130         all_nodes = []
 131         all_slices = []
 132         for aggregate in aggregates:
 133             try:
 134                 # resolve components hostnames
 135                 nodes = self.aggregates[aggregate].get_components()
 136                 all_nodes.extend(nodes)
 137                 # update timestamp and threshold
 138                 self.timestamp = datetime.datetime.now()
 139                 delta = datetime.timedelta(hours=self.components_ttl)
 140                 self.threshold = self.timestamp + delta
 141             except:
 142                 # XX print out to some error log
 143                 pass
 144
 145         self.components = all_nodes
 146         f = open(self.components_file, 'w')
 147         f.write(str(self.components))
 148         f.close()
 149         f = open(self.timestamp_file, 'w')
 150         f.write(str(self.threshold))
 151         f.close()
 152
 153     def load_components(self):
 154         """
 155         Read cached list of nodes and slices.
 156         """
 157         print "loading nodes"
 158         # Read component list from cached file
 159         if os.path.exists(self.components_file):
 160             f = open(self.components_file, 'r')
 161             self.components = eval(f.read())
 162             f.close()
 163
 164         time_format = "%Y-%m-%d %H:%M:%S"
 165         if os.path.exists(self.timestamp_file):
 166             f = open(self.timestamp_file, 'r')
 167             timestamp = str(f.read()).split(".")[0]
 168             self.timestamp = datetime.datetime.fromtimestamp(time.mktime(time.strptime(timestamp, time_format)))
 169             delta = datetime.timedelta(hours=self.components_ttl)
 170             self.threshold = self.timestamp + delta
 171             f.close()
 172
 173     def load_policy(self):
 174         """
 175         Read the list of blacklisted and whitelisted nodes.
 176         """
 177         whitelist = []
 178         blacklist = []
 179         if os.path.exists(self.whitelist_file):
 180             f = open(self.whitelist_file, 'r')
 181             lines = f.readlines()
 182             f.close()
 183             for line in lines:
 184                 line = line.strip().replace(" ", "").replace("\n", "")
 185                 whitelist.extend(line.split(","))
 186
 187
 188         if os.path.exists(self.blacklist_file):
 189             f = open(self.blacklist_file, 'r')
 190             lines = f.readlines()
 191             f.close()
 192             for line in lines:
 193                 line = line.strip().replace(" ", "").replace("\n", "")
 194                 blacklist.extend(line.split(","))
 195
 196         self.policy['whitelist'] = whitelist
 197         self.policy['blacklist'] = blacklist
 198
 199     def load_slices(self):
 200         """
 201          Read current slice instantiation states.
 202         """
 203         print "loading slices"
 204         if os.path.exists(self.slices_file):
 205             f = open(self.components_file, 'r')
 206             self.slices = eval(f.read())
 207             f.close()
 208
 209     def write_slices(self):
 210         """
 211         Write current slice instantiations to file.
 212         """
 213         print "writing slices"
 214         f = open(self.slices_file, 'w')
 215         f.write(str(self.slices))
 216         f.close()
 217
 218
 219     def get_components(self):
 220         """
 221         Return a list of components managed by this slice manager.
 222         """
 223         # Reload components list
 224         now = datetime.datetime.now()
 225         #self.load_components()
 226         if not self.threshold or not self.timestamp or now > self.threshold:
 227             self.refresh_components()
 228         elif now < self.threshold and not self.components:
 229             self.load_components()
 230         return self.components
 231
 232
 233     def get_slices(self):
 234         """
 235         Return a list of instnatiated managed by this slice manager.
 236         """
 237         now = datetime.datetime.now()
 238         #self.load_components()
 239         if not self.threshold or not self.timestamp or now > self.threshold:
 240             self.refresh_components()
 241         elif now < self.threshold and not self.slices:
 242             self.load_components()
 243         return self.slices
 244
 245     def get_slivers(self, hrn):
 246         """
 247         Return the list of slices instantiated at the specified component.
 248         """
 249
 250         # hrn is assumed to be a component hrn
 251         if hrn not in self.slices:
 252             raise RecordNotFound(hrn)
 253
 254         return self.slices[hrn]
 255
 256     def get_rspec(self, hrn, type):
 257         #rspec = Rspec()
 258         if type in ['node']:
 259             nodes = self.shell.GetNodes(self.auth)
 260         elif type in ['slice']:
 261             slices = self.shell.GetSlices(self.auth)
 262         elif type in ['aggregate']:
 263             pass
 264
 265     def get_resources(self, slice_hrn):
 266         """
 267         Return the current rspec for the specified slice.
 268         """
 269         slicename = hrn_to_plcslicename(slice_hrn)
 270         rspec = self.get_rspec(slicenamem, 'slice' )
 271
 272         return rspec
 273
 274     def create_slice(self, slice_hrn, rspec, attributes):
 275         """
 276         Instantiate the specified slice according to whats defined in the rspec.
 277         """
 278         slicename = self.hrn_to_plcslicename(slice_hrn)
 279         #spec = Rspec(rspec)
 280         node_hrns = []
 281         #for netspec in spec['networks]:
 282         #    networkname = netspec['name']
 283         #    nodespec = spec['networks']['nodes']
 284         #    nodes = [nspec['name'] for nspec in nodespec]
 285         #    node_hrns = [networkname + node for node in nodes]
 286         #
 287         self.db.AddSliceToNodes(slice_hrn, node_hrns)
 288         return 1
 289
 290     def delete_slice_(self, slice_hrn):
 291         """
 292         Remove this slice from all components it was previouly associated with and
 293         free up the resources it was using.
 294         """
 295         self.db.DeleteSliceFromNodes(self.auth, slicename, self.components)
 296         return 1
 297
 298     def start_slice(self, slice_hrn):
 299         """
 300         Stop the slice at plc.
 301         """
 302         slicename = hrn_to_plcslicename(slice_hrn)
 303         slices = self.shell.GetSlices(self.auth, {'name': slicename}, ['slice_id'])
 304         if not slices:
 305             raise RecordNotFound(slice_hrn)
 306         slice_id = slices[0]
 307         atrribtes = self.shell.GetSliceAttributes({'slice_id': slice_id, 'name': 'enabled'}, ['slice_attribute_id'])
 308         attribute_id = attreibutes[0]
 309         self.shell.UpdateSliceAttribute(self.auth, attribute_id, "1" )
 310         return 1
 311
 312     def stop_slice(self, slice_hrn):
 313         """
 314         Stop the slice at plc
 315         """
 316         slicename = hrn_to_plcslicename(slice_hrn)
 317         slices = self.shell.GetSlices(self.auth, {'name': slicename}, ['slice_id'])
 318         if not slices:
 319             raise RecordNotFound(slice_hrn)
 320         slice_id = slices[0]
 321         atrribtes = self.shell.GetSliceAttributes({'slice_id': slice_id, 'name': 'enabled'}, ['slice_attribute_id'])
 322         attribute_id = attreibutes[0]
 323         self.shell.UpdateSliceAttribute(self.auth, attribute_id, "0")
 324         return 1
 325
 326     def reset_slice(self, slice_hrn):
 327         """
 328         Reset the slice
 329         """
 330         slicename = self.hrn_to_plcslicename(slice_hrn)
 331         return 1
 332
 333     def get_policy(self):
 334         """
 335         Return the policy of this slice manager.
 336         """
 337
 338         return self.policy
 339
 340
 341
 342 ##############################
 343 ## Server methods here for now
 344 ##############################
 345
 346     def nodes(self):
 347         return self.get_components()
 348
 349     def slices(self):
 350         return self.get_slices()
 351
 352     def resources(self, cred, hrn):
 353         self.decode_authentication(cred, 'info')
 354         self.verify_object_belongs_to_me(hrn)
 355
 356         return self.get_resources(hrn)
 357
 358     def create(self, cred, hrn, rspec):
 359         self.decode_authentication(cred, 'embed')
 360         self.verify_object_belongs_to_me(hrn, rspec)
 361         return self.create(hrn)
 362
 363     def delete(self, cred, hrn):
 364         self.decode_authentication(cred, 'embed')
 365         self.verify_object_belongs_to_me(hrn)
 366         return self.delete_slice(hrn)
 367
 368     def start(self, cred, hrn):
 369         self.decode_authentication(cred, 'control')
 370         return self.start(hrn)
 371
 372     def stop(self, cred, hrn):
 373         self.decode_authentication(cred, 'control')
 374         return self.stop(hrn)
 375
 376     def reset(self, cred, hrn):
 377         self.decode_authentication(cred, 'control')
 378         return self.reset(hrn)
 379
 380     def policy(self, cred):
 381         self.decode_authentication(cred, 'info')
 382         return self.get_policy()
 383
 384     def register_functions(self):
 385         GeniServer.register_functions(self)
 386
 387         # Aggregate interface methods
 388         self.server.register_function(self.components)
 389         self.server.register_function(self.slices)
 390         self.server.register_function(self.resources)
 391         self.server.register_function(self.create)
 392         self.server.register_function(self.delete)
 393         self.server.register_function(self.start)
 394         self.server.register_function(self.stop)
 395         self.server.register_function(self.reset)
 396         self.server.register_function(self.policy)
 397