geni/aggregate.py

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