geni/aggregate.py

   1 import os
   2 import sys
   3 import datetime
   4 import time
   5 import xmlrpclib
   6
   7 from geni.util.geniserver import GeniServer
   8 from geni.util.geniclient import *
   9 from geni.util.cert import Keypair, Certificate
  10 from geni.util.trustedroot import TrustedRootList
  11 from geni.util.excep import *
  12 from geni.util.misc import *
  13 from geni.util.config import Config
  14 from geni.util.rspec import Rspec
  15 from geni.util.specdict import *
  16 from geni.util.storage import SimpleStorage
  17
  18 class Aggregate(GeniServer):
  19
  20     hrn = None
  21     nodes_ttl = None
  22     nodes = {}
  23     slices = {}
  24     policy = {}
  25     timestamp = None
  26     threshold = None
  27     shell = None
  28     registry = None
  29
  30     ##
  31     # Create a new aggregate object.
  32     #
  33     # @param ip the ip address to listen on
  34     # @param port the port to listen on
  35     # @param key_file private key filename of registry
  36     # @param cert_file certificate filename containing public key (could be a GID file)
  37
  38     def __init__(self, ip, port, key_file, cert_file, config = "/usr/share/geniwrapper/util/geni_config"):
  39         GeniServer.__init__(self, ip, port, key_file, cert_file)
  40         self.conf = Config(config)
  41         basedir = self.conf.GENI_BASE_DIR + os.sep
  42         server_basedir = basedir + os.sep + "geni" + os.sep
  43         self.hrn = self.conf.GENI_INTERFACE_HRN
  44
  45         nodes_file = os.sep.join([server_basedir, 'agg.' + self.hrn + '.components'])
  46         self.nodes = SimpleStorage(nodes_file)
  47
  48         node_slices_file = os.sep.join([server_basedir, 'agg.' + self.hrn + '.slices'])
  49         self.slices = SimpleStorage(node_slices_file)
  50         self.slices.load()
  51
  52         policy_file = os.sep.join([server_basedir, 'policy'])
  53         self.policy = SimpleStorage(policy_file, {'whitelist': [], 'blacklist': []})
  54
  55         timestamp_file = os.sep.join([server_basedir, 'components', self.hrn + '.timestamp'])
  56         self.timestamp = SimpleStorage(timestamp_file)
  57
  58         self.nodes_ttl = 1
  59         self.connectPLC()
  60         self.connectRegistry()
  61
  62     def connectRegistry(self):
  63         """
  64         Connect to the registry
  65         """
  66         pass
  67
  68     def connectPLC(self):
  69         """
  70         Connect to the plc api interface. First attempt to impor thte shell, if that fails
  71         try to connect to the xmlrpc server.
  72         """
  73         self.auth = {'Username': self.conf.GENI_PLC_USER,
  74                      'AuthMethod': 'password',
  75                      'AuthString': self.conf.GENI_PLC_PASSWORD}
  76
  77         try:
  78            # try to import PLC.Shell directly
  79             sys.path.append(self.conf.GENI_PLC_SHELL_PATH)
  80             import PLC.Shell
  81             self.shell = PLC.Shell.Shell(globals())
  82             self.shell.AuthCheck()
  83         except ImportError:
  84             # connect to plc api via xmlrpc
  85             plc_host = self.conf.GENI_PLC_HOST
  86             plc_port = self.conf.GENI_PLC_PORT
  87             plc_api_path = self.conf.GENI_PLC_API_PATH
  88             url = "https://%(plc_host)s:%(plc_port)s/%(plc_api_path)s/" % locals()
  89             self.auth = {'Username': self.conf.GENI_PLC_USER,
  90                  'AuthMethod': 'password',
  91                  'AuthString': self.conf.GENI_PLC_PASSWORD}
  92
  93             self.shell = xmlrpclib.Server(url, verbose = 0, allow_none = True)
  94             self.shell.AuthCheck(self.auth)
  95
  96     def hostname_to_hrn(self, login_base, hostname):
  97         """
  98         Convert hrn to plantelab name.
  99         """
 100         genihostname = "_".join(hostname.split("."))
 101         return ".".join([self.hrn, login_base, genihostname])
 102
 103     def slicename_to_hrn(self, slicename):
 104         """
 105         Convert hrn to planetlab name.
 106         """
 107         slicename = slicename.replace("_", ".")
 108         return ".".join([self.hrn, slicename])
 109
 110     def refresh_components(self):
 111         """
 112         Update the cached list of nodes.
 113         """
 114         # resolve component hostnames
 115         nodes = self.shell.GetNodes(self.auth, {}, ['hostname', 'site_id'])
 116
 117         # resolve site login_bases
 118         site_ids = [node['site_id'] for node in nodes]
 119         sites = self.shell.GetSites(self.auth, site_ids, ['site_id', 'login_base'])
 120         site_dict = {}
 121         for site in sites:
 122             site_dict[site['site_id']] = site['login_base']
 123
 124         # convert plc names to geni hrn
 125         nodedict = {}
 126         for node in nodes:
 127             node_hrn = self.hostname_to_hrn(site_dict[node['site_id']], node['hostname'])
 128             # apply policy.
 129             # Do not allow nodes found in blacklist, only allow nodes found in whitelist
 130             if self.polciy['whitelist'] and node_hrn not in self.polciy['whitelist']:
 131                 continue
 132             if self.polciy['blacklist'] and node_hrn in self.policy['blacklist']:
 133                 continue
 134             nodedict[node_hrn] = node['hostname']
 135
 136         self.nodes = SimpleStorage(self.nodes.db_filename, nodedict)
 137         self.nodes.write()
 138
 139         # update timestamp and threshold
 140         self.timestamp['timestamp'] =  datetime.datetime.now()
 141         delta = datetime.timedelta(hours=self.nodes_ttl)
 142         self.threshold = self.timestamp['timestamp'] + delta
 143         self.timestamp.write()
 144
 145     def load_components(self):
 146         """
 147         Read cached list of nodes.
 148         """
 149         # Read component list from cached file
 150         self.nodes.load()
 151         self.timestamp.load()
 152         time_format = "%Y-%m-%d %H:%M:%S"
 153         timestamp = self.timestamp['timestamp']
 154         self.timestamp['timestamp'] = datetime.datetime.fromtimestamp(time.mktime(time.strptime(timestamp, time_format)))
 155         delta = datetime.timedelta(hours=self.nodes_ttl)
 156         self.threshold = self.timestamp['timestamp'] + delta
 157
 158     def load_policy(self):
 159         """
 160         Read the list of blacklisted and whitelisted nodes.
 161         """
 162         self.policy.load()
 163
 164
 165     def get_components(self):
 166         """
 167         Return a list of components at this aggregate.
 168         """
 169         # Reload components list
 170         now = datetime.datetime.now()
 171         #self.load_components()
 172         if not self.threshold or not self.timestamp['timestamp'] or now > self.threshold:
 173             self.refresh_components()
 174         elif now < self.threshold and not self.nodes.keys():
 175             self.load_components()
 176         return self.nodes.keys()
 177
 178     def get_rspec(self, hrn, type):
 179
 180         # Get the required nodes
 181         if type in ['aggregate']:
 182             nodes = self.shell.GetNodes(self.auth)
 183         elif type in ['slice']:
 184             slicename = hrn_to_pl_slicename(hrn)
 185             slices = self.shell.GetSlices(self.auth, [slicename])
 186             node_ids = slices[0]['node_ids']
 187             nodes = self.shell.GetNodes(self.auth, node_ids)
 188
 189         # Get all network interfaces
 190         interface_ids = []
 191         for node in nodes:
 192             interface_ids.extend(node['nodenetwork_ids'])
 193         interfaces = self.shell.GetNodeNetworks(self.auth, interface_ids)
 194         interface_dict = {}
 195         for interface in interfaces:
 196             interface_dict[interface['nodenetwork_id']] = interface
 197
 198         # join nodes with thier interfaces
 199         for node in nodes:
 200             node['interfaces'] = []
 201             for nodenetwork_id in node['nodenetwork_ids']:
 202                 node['interfaces'].append(interface_dict[nodenetwork_id])
 203
 204         # convert and threshold to ints
 205         timestamp = self.timestamp['timestamp']
 206         start_time = int(self.timestamp['timestamp'].strftime("%s"))
 207         end_time = int(self.duration.strftime("%s"))
 208         duration = end_time - start_time
 209
 210         # create the plc dict
 211         networks = {'nodes': nodes, 'name': self.hrn, 'start_time': start_time, 'duration': duration}
 212         resources = {'networks': networks, 'start_time': start_time, 'duration': duration}
 213
 214         # convert the plc dict to an rspec dict
 215         resouceDict = RspecDict(resources)
 216
 217         # convert the rspec dict to xml
 218         rspec = Rspec()
 219         rspec.parseDict(resourceDict)
 220         return rspec.toxml()
 221
 222     def get_resources(self, slice_hrn):
 223         """
 224         Return the current rspec for the specified slice.
 225         """
 226         slicename = hrn_to_plcslicename(slice_hrn)
 227         rspec = self.get_rspec(slicenamem, 'slice')
 228
 229         return rspec
 230
 231     def create_slice(self, slice_hrn, rspec, attributes = []):
 232         """
 233         Instantiate the specified slice according to whats defined in the rspec.
 234         """
 235         slicename = self.hrn_to_plcslicename(slice_hrn)
 236
 237         # extract node list from rspec
 238         spec = Rspec(rspec)
 239         nodespecs = spec.getDictsByTagName('NodeSpec')
 240         nodes = [nodespec['name'] for nodespec in nodespecs]
 241
 242         # save slice state locally
 243         # we can assume that spec object has been validated so its safer to
 244         # save this instead of the unvalidated rspec the user gave us
 245         self.slices[slice_hrn] = spec.toxml()
 246         self.slices.write()
 247
 248         # add slice to nodes at plc
 249         self.shell.AddSliceToNodes(self.auth, slicename, nodes)
 250         for attribute in attributes:
 251             type, value, node, nodegroup = attribute['type'], attribute['value'], attribute['node'], attribute['nodegroup']
 252             shell.AddSliceAttribute(self.auth, slicename, type, value, node, nodegroup)
 253
 254         # XX contact the registry to get the list of users on this slice and
 255         # their keys.
 256         #slice_record = self.registry.resolve(slice_hrn)
 257         #person_records = slice_record['users']
 258         # for person in person_record:
 259         #    email = person['email']
 260         #    self.shell.AddPersonToSlice(self.auth, email, slicename)
 261
 262
 263         return 1
 264
 265     def update_slice(self, slice_hrn, rspec, attributes = []):
 266         """
 267         Update the specified slice.
 268         """
 269         # Get slice info
 270         slicename = self.hrn_to_plcslicename(slice_hrn)
 271         slices = self.shell.GetSlices(self.auth, [slicename], ['node_ids'])
 272         if not slice:
 273             raise RecordNotFound(slice_hrn)
 274         slice = slices[0]
 275
 276         # find out where this slice is currently running
 277         nodes = self.shell.GetNodes(self.auth, slice['node_ids'], ['hostname'])
 278         hostnames = [node['hostname'] for node in nodes]
 279
 280         # get netspec details
 281         spec = Rspec(rspec)
 282         nodespecs = spec.getDictsByTagName('NodeSpec')
 283         nodes = [nodespec['name'] for nodespec in nodespecs]
 284
 285         # save slice state locally
 286         # we can assume that spec object has been validated so its safer to
 287         # save this instead of the unvalidated rspec the user gave us
 288         self.slices[slice_hrn] = spec.toxml()
 289         self.slices.write()
 290
 291         # remove nodes not in rspec
 292         delete_nodes = set(hostnames).difference(nodes)
 293         # add nodes from rspec
 294         added_nodes = set(nodes).difference(hostnames)
 295
 296         shell.AddSliceToNodes(self.auth, slicename, added_nodes)
 297         shell.DeleteSliceFromNodes(self.auth, slicename, deleted_nodes)
 298
 299         for attribute in attributes:
 300             type, value, node, nodegroup = attribute['type'], attribute['value'], attribute['node'], attribute['nodegroup']
 301             shell.AddSliceAttribute(self.auth, slicename, type, value, node, nodegroup)
 302
 303         # contact registry to get slice users and add them to the slice
 304         # slice_record = self.registry.resolve(slice_hrn)
 305         # persons = slice_record['users']
 306
 307         #for person in persons:
 308         #    shell.AddPersonToSlice(person['email'], slice_name)
 309
 310
 311     def delete_slice_(self, slice_hrn):
 312         """
 313         Remove this slice from all components it was previouly associated with and
 314         free up the resources it was using.
 315         """
 316         if self.slices.has_key(slice_hrn):
 317             self.slices.pop(slice_hrn)
 318             self.slices.write()
 319
 320         slicename = self.hrn_to_plcslicename(slice_hrn)
 321         slices = shell.GetSlices(self.auth, [slicename])
 322         if not slice:
 323             raise RecordNotFound(slice_hrn)
 324         slice = slices[0]
 325
 326         shell.DeleteSliceFromNodes(self.auth, slicename, slice['node_ids'])
 327         return 1
 328
 329     def start_slice(self, slice_hrn):
 330         """
 331         Stop the slice at plc.
 332         """
 333         slicename = hrn_to_plcslicename(slice_hrn)
 334         slices = self.shell.GetSlices(self.auth, {'name': slicename}, ['slice_id'])
 335         if not slices:
 336             raise RecordNotFound(slice_hrn)
 337         slice_id = slices[0]
 338         atrribtes = self.shell.GetSliceAttributes({'slice_id': slice_id, 'name': 'enabled'}, ['slice_attribute_id'])
 339         attribute_id = attreibutes[0]
 340         self.shell.UpdateSliceAttribute(self.auth, attribute_id, "1" )
 341         return 1
 342
 343     def stop_slice(self, slice_hrn):
 344         """
 345         Stop the slice at plc
 346         """
 347         slicename = hrn_to_plcslicename(slice_hrn)
 348         slices = self.shell.GetSlices(self.auth, {'name': slicename}, ['slice_id'])
 349         if not slices:
 350             raise RecordNotFound(slice_hrn)
 351         slice_id = slices[0]
 352         atrribtes = self.shell.GetSliceAttributes({'slice_id': slice_id, 'name': 'enabled'}, ['slice_attribute_id'])
 353         attribute_id = attreibutes[0]
 354         self.shell.UpdateSliceAttribute(self.auth, attribute_id, "0")
 355         return 1
 356
 357
 358     def reset_slice(self, slice_hrn):
 359         """
 360         Reset the slice
 361         """
 362         slicename = self.hrn_to_plcslicename(slice_hrn)
 363         return 1
 364
 365     def get_policy(self):
 366         """
 367         Return this aggregates policy.
 368         """
 369
 370         return self.policy
 371
 372
 373
 374 ##############################
 375 ## Server methods here for now
 376 ##############################
 377
 378     def components(self):
 379         return self.get_components()
 380
 381     #def slices(self):
 382     #    return self.get_slices()
 383
 384     def resources(self, cred, hrn):
 385         self.decode_authentication(cred, 'info')
 386         self.verify_object_belongs_to_me(hrn)
 387
 388         return self.get_resources(hrn)
 389
 390     def createSlice(self, cred, hrn, rspec):
 391         self.decode_authentication(cred, 'embed')
 392         self.verify_object_belongs_to_me(hrn)
 393         return self.create_slice(hrn)
 394
 395     def updateSlice(self, cred, hrn, rspec):
 396         self.decode_authentication(cred, 'embed')
 397         self.verify_object_belongs_to_me(hrn)
 398         return self.update_slice(hrn)
 399
 400     def deleteSlice(self, cred, hrn):
 401         self.decode_authentication(cred, 'embed')
 402         self.verify_object_belongs_to_me(hrn)
 403         return self.delete_slice(hrn)
 404
 405     def startSlice(self, cred, hrn):
 406         self.decode_authentication(cred, 'control')
 407         return self.start_slice(hrn)
 408
 409     def stopSlice(self, cred, hrn):
 410         self.decode_authentication(cred, 'control')
 411         return self.stop(hrn)
 412
 413     def resetSlice(self, cred, hrn):
 414         self.decode_authentication(cred, 'control')
 415         return self.reset(hrn)
 416
 417     def policy(self, cred):
 418         self.decode_authentication(cred, 'info')
 419         return self.get_policy()
 420
 421     def register_functions(self):
 422         GeniServer.register_functions(self)
 423
 424         # Aggregate interface methods
 425         self.server.register_function(self.components)
 426         #self.server.register_function(self.slices)
 427         self.server.register_function(self.resources)
 428         self.server.register_function(self.createSlice)
 429         self.server.register_function(self.deleteSlice)
 430         self.server.register_function(self.startSlice)
 431         self.server.register_function(self.stopSlice)
 432         self.server.register_function(self.resetSlice)
 433         self.server.register_function(self.policy)
 434