geni/plc/slicemgr.py

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