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
 192     def get_rspec(self, hrn, type):
 193         #rspec = Rspec()
 194         if type in ['node']:
 195             nodes = self.shell.GetNodes(self.auth)
 196         elif type in ['slice']:
 197             slices = self.shell.GetSlices(self.auth)
 198         elif type in ['aggregate']:
 199             pass
 200
 201     def get_resources(self, slice_hrn):
 202         """
 203         Return the current rspec for the specified slice.
 204         """
 205         slicename = hrn_to_plcslicename(slice_hrn)
 206         rspec = self.get_rspec(slicenamem, 'slice' )
 207
 208         return rspec
 209
 210     def create_slice(self, slice_hrn, rspec):
 211         """
 212         Instantiate the specified slice according to whats defined in the rspec.
 213         """
 214         slicename = self.hrn_to_plcslicename(slice_hrn)
 215         #spec = Rspec(rspec)
 216         #components = spec.components()
 217         #shell.AddSliceToNodes(self.auth, slicename, components)
 218         return 1
 219
 220     def delete_slice_(self, slice_hrn):
 221         """
 222         Remove this slice from all components it was previouly associated with and
 223         free up the resources it was using.
 224         """
 225         slicename = self.hrn_to_plcslicename(slice_hrn)
 226         rspec = self.get_resources(slice_hrn)
 227         components = rspec.components()
 228         shell.DeleteSliceFromNodes(self.auth, slicename, components)
 229         return 1
 230
 231     def start_slice(self, slice_hrn):
 232         """
 233         Stop the slice at plc.
 234         """
 235         slicename = hrn_to_plcslicename(slice_hrn)
 236         slices = self.shell.GetSlices(self.auth, {'name': slicename}, ['slice_id'])
 237         if not slices:
 238             raise RecordNotFound(slice_hrn)
 239         slice_id = slices[0]
 240         atrribtes = self.shell.GetSliceAttributes({'slice_id': slice_id, 'name': 'enabled'}, ['slice_attribute_id'])
 241         attribute_id = attreibutes[0]
 242         self.shell.UpdateSliceAttribute(self.auth, attribute_id, "1" )
 243         return 1
 244
 245     def stop_slice(self, slice_hrn):
 246         """
 247         Stop the slice at plc
 248         """
 249         slicename = hrn_to_plcslicename(slice_hrn)
 250         slices = self.shell.GetSlices(self.auth, {'name': slicename}, ['slice_id'])
 251         if not slices:
 252             raise RecordNotFound(slice_hrn)
 253         slice_id = slices[0]
 254         atrribtes = self.shell.GetSliceAttributes({'slice_id': slice_id, 'name': 'enabled'}, ['slice_attribute_id'])
 255         attribute_id = attreibutes[0]
 256         self.shell.UpdateSliceAttribute(self.auth, attribute_id, "0")
 257         return 1
 258
 259     def reset_slice(self, slice_hrn):
 260         """
 261         Reset the slice
 262         """
 263         slicename = self.hrn_to_plcslicename(slice_hrn)
 264         return 1
 265
 266     def get_policy(self):
 267         """
 268         Return this aggregates policy.
 269         """
 270
 271         return self.policy
 272
 273
 274
 275 ##############################
 276 ## Server methods here for now
 277 ##############################
 278
 279     def nodes(self):
 280         return self..get_components()
 281
 282     #def slices(self):
 283     #    return self.get_slices()
 284
 285     def resources(self, cred, hrn):
 286         self.decode_authentication(cred, 'info')
 287         self.verify_object_belongs_to_me(hrn)
 288
 289         return self.get_resources(hrn)
 290
 291     def create(self, cred, hrn, rspec):
 292         self.decode_authentication(cred, 'embed')
 293         self.verify_object_belongs_to_me(hrn, rspec)
 294         return self.create(hrn)
 295
 296     def delete(self, cred, hrn):
 297         self.decode_authentication(cred, 'embed')
 298         self.verify_object_belongs_to_me(hrn)
 299         return self.delete_slice(hrn)
 300
 301     def start(self, cred, hrn):
 302         self.decode_authentication(cred, 'control')
 303         return self.start(hrn)
 304
 305     def stop(self, cred, hrn):
 306         self.decode_authentication(cred, 'control')
 307         return self.stop(hrn)
 308
 309     def reset(self, cred, hrn):
 310         self.decode_authentication(cred, 'control')
 311         return self.reset(hrn)
 312
 313     def policy(self, cred):
 314         self.decode_authentication(cred, 'info')
 315         return self.get_policy()
 316
 317     def register_functions(self):
 318         GeniServer.register_functions(self)
 319
 320         # Aggregate interface methods
 321         self.server.register_function(self.components)
 322         #self.server.register_function(self.slices)
 323         self.server.register_function(self.resources)
 324         self.server.register_function(self.create)
 325         self.server.register_function(self.delete)
 326         self.server.register_function(self.start)
 327         self.server.register_function(self.stop)
 328         self.server.register_function(self.reset)
 329         self.server.register_function(self.policy)
 330