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