sfa/managers/slice_manager_pl.py

   1 #
   2 import sys
   3 import time,datetime
   4 from StringIO import StringIO
   5 from types import StringTypes
   6 from copy import deepcopy
   7 from copy import copy
   8 from lxml import etree
   9
  10 from sfa.util.sfalogging import sfa_logger
  11 from sfa.util.rspecHelper import merge_rspecs
  12 from sfa.util.xrn import Xrn, urn_to_hrn, hrn_to_urn
  13 from sfa.util.plxrn import hrn_to_pl_slicename
  14 from sfa.util.rspec import *
  15 from sfa.util.specdict import *
  16 from sfa.util.faults import *
  17 from sfa.util.record import SfaRecord
  18 from sfa.rspecs.pg_rspec import PGRSpec
  19 from sfa.rspecs.sfa_rspec import SfaRSpec
  20 from sfa.rspecs.pg_rspec_converter import PGRSpecConverter
  21 from sfa.rspecs.rspec_parser import parse_rspec
  22 from sfa.util.policy import Policy
  23 from sfa.util.prefixTree import prefixTree
  24 from sfa.util.sfaticket import *
  25 from sfa.trust.credential import Credential
  26 from sfa.util.threadmanager import ThreadManager
  27 import sfa.util.xmlrpcprotocol as xmlrpcprotocol
  28 import sfa.plc.peers as peers
  29 from sfa.util.version import version_core
  30 from sfa.rspecs.rspec_version import RSpecVersion
  31 from sfa.util.callids import Callids
  32
  33 # we have specialized xmlrpclib.ServerProxy to remember the input url
  34 # OTOH it's not clear if we're only dealing with XMLRPCServerProxy instances
  35 def get_serverproxy_url (server):
  36     try:
  37         return server.url
  38     except:
  39         sfa_logger().warning("GetVersion, falling back to xmlrpclib.ServerProxy internals")
  40         return server._ServerProxy__host + server._ServerProxy__handler
  41
  42 def GetVersion(api):
  43     # peers explicitly in aggregates.xml
  44     peers =dict ([ (peername,get_serverproxy_url(v)) for (peername,v) in api.aggregates.iteritems()
  45                    if peername != api.hrn])
  46     xrn=Xrn (api.hrn)
  47     sm_version=version_core({'interface':'slicemgr',
  48                              'hrn' : xrn.get_hrn(),
  49                              'urn' : xrn.get_urn(),
  50                              'peers': peers,
  51                              })
  52     # local aggregate if present needs to have localhost resolved
  53     if api.hrn in api.aggregates:
  54         local_am_url=get_serverproxy_url(api.aggregates[api.hrn])
  55         sm_version['peers'][api.hrn]=local_am_url.replace('localhost',sm_version['hostname'])
  56     return sm_version
  57
  58 def CreateSliver(api, xrn, creds, rspec, users, call_id):
  59
  60     if Callids().already_handled(call_id): return ""
  61
  62     hrn, type = urn_to_hrn(xrn)
  63
  64     # Validate the RSpec against PlanetLab's schema --disabled for now
  65     # The schema used here needs to aggregate the PL and VINI schemas
  66     # schema = "/var/www/html/schemas/pl.rng"
  67     schema = None
  68     if schema:
  69         try:
  70             tree = etree.parse(StringIO(rspec))
  71         except etree.XMLSyntaxError:
  72             message = str(sys.exc_info()[1])
  73             raise InvalidRSpec(message)
  74
  75         relaxng_doc = etree.parse(schema)
  76         relaxng = etree.RelaxNG(relaxng_doc)
  77
  78         if not relaxng(tree):
  79             error = relaxng.error_log.last_error
  80             message = "%s (line %s)" % (error.message, error.line)
  81             raise InvalidRSpec(message)
  82
  83     # get the callers hrn
  84     valid_cred = api.auth.checkCredentials(creds, 'createsliver', hrn)[0]
  85     caller_hrn = Credential(string=valid_cred).get_gid_caller().get_hrn()
  86
  87     # attempt to use delegated credential first
  88     credential = api.getDelegatedCredential(creds)
  89     if not credential:
  90         credential = api.getCredential()
  91     threads = ThreadManager()
  92     for aggregate in api.aggregates:
  93         # prevent infinite loop. Dont send request back to caller
  94         # unless the caller is the aggregate's SM
  95         if caller_hrn == aggregate and aggregate != api.hrn:
  96             continue
  97
  98         # Just send entire RSpec to each aggregate
  99         server = api.aggregates[aggregate]
 100         threads.run(server.CreateSliver, xrn, credential, rspec, users, call_id)
 101
 102     results = threads.get_results()
 103     rspec = SfaRSpec()
 104     for result in results:
 105         rspec.merge(result)
 106     return rspec.toxml()
 107
 108 def RenewSliver(api, xrn, creds, expiration_time, call_id):
 109     if Callids().already_handled(call_id): return True
 110
 111     (hrn, type) = urn_to_hrn(xrn)
 112     # get the callers hrn
 113     valid_cred = api.auth.checkCredentials(creds, 'renewsliver', hrn)[0]
 114     caller_hrn = Credential(string=valid_cred).get_gid_caller().get_hrn()
 115
 116     # attempt to use delegated credential first
 117     credential = api.getDelegatedCredential(creds)
 118     if not credential:
 119         credential = api.getCredential()
 120     threads = ThreadManager()
 121     for aggregate in api.aggregates:
 122         # prevent infinite loop. Dont send request back to caller
 123         # unless the caller is the aggregate's SM
 124         if caller_hrn == aggregate and aggregate != api.hrn:
 125             continue
 126
 127         server = api.aggregates[aggregate]
 128         threads.run(server.RenewSliver, xrn, [credential], expiration_time, call_id)
 129     # 'and' the results
 130     return reduce (lambda x,y: x and y, threads.get_results() , True)
 131
 132 def get_ticket(api, xrn, creds, rspec, users):
 133     slice_hrn, type = urn_to_hrn(xrn)
 134     # get the netspecs contained within the clients rspec
 135     aggregate_rspecs = {}
 136     tree= etree.parse(StringIO(rspec))
 137     elements = tree.findall('./network')
 138     for element in elements:
 139         aggregate_hrn = element.values()[0]
 140         aggregate_rspecs[aggregate_hrn] = rspec
 141
 142     # get the callers hrn
 143     valid_cred = api.auth.checkCredentials(creds, 'getticket', slice_hrn)[0]
 144     caller_hrn = Credential(string=valid_cred).get_gid_caller().get_hrn()
 145
 146     # attempt to use delegated credential first
 147     credential = api.getDelegatedCredential(creds)
 148     if not credential:
 149         credential = api.getCredential()
 150     threads = ThreadManager()
 151     for (aggregate, aggregate_rspec) in aggregate_rspecs.iteritems():
 152         # prevent infinite loop. Dont send request back to caller
 153         # unless the caller is the aggregate's SM
 154         if caller_hrn == aggregate and aggregate != api.hrn:
 155             continue
 156         server = None
 157         if aggregate in api.aggregates:
 158             server = api.aggregates[aggregate]
 159         else:
 160             net_urn = hrn_to_urn(aggregate, 'authority')
 161             # we may have a peer that knows about this aggregate
 162             for agg in api.aggregates:
 163                 target_aggs = api.aggregates[agg].get_aggregates(credential, net_urn)
 164                 if not target_aggs or not 'hrn' in target_aggs[0]:
 165                     continue
 166                 # send the request to this address
 167                 url = target_aggs[0]['url']
 168                 server = xmlrpcprotocol.get_server(url, api.key_file, api.cert_file)
 169                 # aggregate found, no need to keep looping
 170                 break
 171         if server is None:
 172             continue
 173         threads.run(server.GetTicket, xrn, credential, aggregate_rspec, users)
 174
 175     results = threads.get_results()
 176
 177     # gather information from each ticket
 178     rspecs = []
 179     initscripts = []
 180     slivers = []
 181     object_gid = None
 182     for result in results:
 183         agg_ticket = SfaTicket(string=result)
 184         attrs = agg_ticket.get_attributes()
 185         if not object_gid:
 186             object_gid = agg_ticket.get_gid_object()
 187         rspecs.append(agg_ticket.get_rspec())
 188         initscripts.extend(attrs.get('initscripts', []))
 189         slivers.extend(attrs.get('slivers', []))
 190
 191     # merge info
 192     attributes = {'initscripts': initscripts,
 193                  'slivers': slivers}
 194     merged_rspec = merge_rspecs(rspecs)
 195
 196     # create a new ticket
 197     ticket = SfaTicket(subject = slice_hrn)
 198     ticket.set_gid_caller(api.auth.client_gid)
 199     ticket.set_issuer(key=api.key, subject=api.hrn)
 200     ticket.set_gid_object(object_gid)
 201     ticket.set_pubkey(object_gid.get_pubkey())
 202     #new_ticket.set_parent(api.auth.hierarchy.get_auth_ticket(auth_hrn))
 203     ticket.set_attributes(attributes)
 204     ticket.set_rspec(merged_rspec)
 205     ticket.encode()
 206     ticket.sign()
 207     return ticket.save_to_string(save_parents=True)
 208
 209
 210 def DeleteSliver(api, xrn, creds, call_id):
 211     if Callids().already_handled(call_id): return ""
 212     (hrn, type) = urn_to_hrn(xrn)
 213     # get the callers hrn
 214     valid_cred = api.auth.checkCredentials(creds, 'deletesliver', hrn)[0]
 215     caller_hrn = Credential(string=valid_cred).get_gid_caller().get_hrn()
 216
 217     # attempt to use delegated credential first
 218     credential = api.getDelegatedCredential(creds)
 219     if not credential:
 220         credential = api.getCredential()
 221     threads = ThreadManager()
 222     for aggregate in api.aggregates:
 223         # prevent infinite loop. Dont send request back to caller
 224         # unless the caller is the aggregate's SM
 225         if caller_hrn == aggregate and aggregate != api.hrn:
 226             continue
 227         server = api.aggregates[aggregate]
 228         threads.run(server.DeleteSliver, xrn, credential, call_id)
 229     threads.get_results()
 230     return 1
 231
 232 def start_slice(api, xrn, creds):
 233     hrn, type = urn_to_hrn(xrn)
 234
 235     # get the callers hrn
 236     valid_cred = api.auth.checkCredentials(creds, 'startslice', hrn)[0]
 237     caller_hrn = Credential(string=valid_cred).get_gid_caller().get_hrn()
 238
 239     # attempt to use delegated credential first
 240     credential = api.getDelegatedCredential(creds)
 241     if not credential:
 242         credential = api.getCredential()
 243     threads = ThreadManager()
 244     for aggregate in api.aggregates:
 245         # prevent infinite loop. Dont send request back to caller
 246         # unless the caller is the aggregate's SM
 247         if caller_hrn == aggregate and aggregate != api.hrn:
 248             continue
 249         server = api.aggregates[aggregate]
 250         threads.run(server.Start, xrn, credential)
 251     threads.get_results()
 252     return 1
 253
 254 def stop_slice(api, xrn, creds):
 255     hrn, type = urn_to_hrn(xrn)
 256
 257     # get the callers hrn
 258     valid_cred = api.auth.checkCredentials(creds, 'stopslice', hrn)[0]
 259     caller_hrn = Credential(string=valid_cred).get_gid_caller().get_hrn()
 260
 261     # attempt to use delegated credential first
 262     credential = api.getDelegatedCredential(creds)
 263     if not credential:
 264         credential = api.getCredential()
 265     threads = ThreadManager()
 266     for aggregate in api.aggregates:
 267         # prevent infinite loop. Dont send request back to caller
 268         # unless the caller is the aggregate's SM
 269         if caller_hrn == aggregate and aggregate != api.hrn:
 270             continue
 271         server = api.aggregates[aggregate]
 272         threads.run(server.Stop, xrn, credential)
 273     threads.get_results()
 274     return 1
 275
 276 def reset_slice(api, xrn):
 277     """
 278     Not implemented
 279     """
 280     return 1
 281
 282 def shutdown(api, xrn, creds):
 283     """
 284     Not implemented
 285     """
 286     return 1
 287
 288 def status(api, xrn, creds):
 289     """
 290     Not implemented
 291     """
 292     return 1
 293
 294 # Thierry : caching at the slicemgr level makes sense to some extent
 295 caching=True
 296 #caching=False
 297 def ListSlices(api, creds, call_id):
 298
 299     if Callids().already_handled(call_id): return []
 300
 301     # look in cache first
 302     if caching and api.cache:
 303         slices = api.cache.get('slices')
 304         if slices:
 305             return slices
 306
 307     # get the callers hrn
 308     valid_cred = api.auth.checkCredentials(creds, 'listslices', None)[0]
 309     caller_hrn = Credential(string=valid_cred).get_gid_caller().get_hrn()
 310
 311     # attempt to use delegated credential first
 312     credential = api.getDelegatedCredential(creds)
 313     if not credential:
 314         credential = api.getCredential()
 315     threads = ThreadManager()
 316     # fetch from aggregates
 317     for aggregate in api.aggregates:
 318         # prevent infinite loop. Dont send request back to caller
 319         # unless the caller is the aggregate's SM
 320         if caller_hrn == aggregate and aggregate != api.hrn:
 321             continue
 322         server = api.aggregates[aggregate]
 323         threads.run(server.ListSlices, credential, call_id)
 324
 325     # combime results
 326     results = threads.get_results()
 327     slices = []
 328     for result in results:
 329         slices.extend(result)
 330
 331     # cache the result
 332     if caching and api.cache:
 333         api.cache.add('slices', slices)
 334
 335     return slices
 336
 337
 338 def ListResources(api, creds, options, call_id):
 339
 340     if Callids().already_handled(call_id): return ""
 341
 342     # get slice's hrn from options
 343     xrn = options.get('geni_slice_urn', '')
 344     (hrn, type) = urn_to_hrn(xrn)
 345
 346     # get the rspec's return format from options
 347     rspec_version = RSpecVersion(options.get('rspec_version', 'SFA 1'))
 348     version_string = "rspec_%s_%s" % (rspec_version.format, rspec_version.version)
 349
 350     # get hrn of the original caller
 351     origin_hrn = options.get('origin_hrn', None)
 352     if not origin_hrn:
 353         if isinstance(creds, list):
 354             origin_hrn = Credential(string=creds[0]).get_gid_caller().get_hrn()
 355         else:
 356             origin_hrn = Credential(string=creds).get_gid_caller().get_hrn()
 357
 358     # look in cache first
 359     if caching and api.cache and not xrn:
 360         rspec =  api.cache.get(version_string)
 361         if rspec:
 362             return rspec
 363
 364     # get the callers hrn
 365     valid_cred = api.auth.checkCredentials(creds, 'listnodes', hrn)[0]
 366     caller_hrn = Credential(string=valid_cred).get_gid_caller().get_hrn()
 367
 368     # attempt to use delegated credential first
 369     credential = api.getDelegatedCredential(creds)
 370     if not credential:
 371         credential = api.getCredential()
 372     threads = ThreadManager()
 373     for aggregate in api.aggregates:
 374         # prevent infinite loop. Dont send request back to caller
 375         # unless the caller is the aggregate's SM
 376         if caller_hrn == aggregate and aggregate != api.hrn:
 377             continue
 378         # get the rspec from the aggregate
 379         server = api.aggregates[aggregate]
 380         my_opts = copy(options)
 381         my_opts['geni_compressed'] = False
 382         threads.run(server.ListResources, credential, my_opts, call_id)
 383         #threads.run(server.get_resources, cred, xrn, origin_hrn)
 384
 385     results = threads.get_results()
 386     #results.append(open('/root/protogeni.rspec', 'r').read())
 387     rspec = SfaRSpec()
 388     for result in results:
 389         try:
 390             tmp_rspec = parse_rspec(result)
 391             if isinstance(tmp_rspec, SfaRSpec):
 392                 rspec.merge(result)
 393             elif isinstance(tmp_rspec, PGRSpec):
 394                 rspec.merge(PGRSpecConverter.to_sfa_rspec(result))
 395             else:
 396                 api.logger.info("SM.ListResources: invalid aggregate rspec")
 397         except:
 398             api.logger.info("SM.ListResources: Failed to merge aggregate rspec")
 399
 400     # cache the result
 401     if caching and api.cache and not xrn:
 402         api.cache.add(version_string, rspec.toxml())
 403
 404     return rspec.toxml()
 405
 406 # first draft at a merging SliverStatus
 407 def SliverStatus(api, slice_xrn, creds, call_id):
 408     if Callids().already_handled(call_id): return {}
 409     # attempt to use delegated credential first
 410     credential = api.getDelegatedCredential(creds)
 411     if not credential:
 412         credential = api.getCredential()
 413     threads = ThreadManager()
 414     for aggregate in api.aggregates:
 415         server = api.aggregates[aggregate]
 416         threads.run (server.SliverStatus, slice_xrn, credential, call_id)
 417     results = threads.get_results()
 418
 419     # get rid of any void result - e.g. when call_id was hit where by convention we return {}
 420     results = [ result for result in results if result and result['geni_resources']]
 421
 422     # do not try to combine if there's no result
 423     if not results : return {}
 424
 425     # otherwise let's merge stuff
 426     overall = {}
 427
 428     # mmh, it is expected that all results carry the same urn
 429     overall['geni_urn'] = results[0]['geni_urn']
 430
 431     # consolidate geni_status - simple model using max on a total order
 432     states = [ 'ready', 'configuring', 'failed', 'unknown' ]
 433     # hash name to index
 434     shash = dict ( zip ( states, range(len(states)) ) )
 435     def combine_status (x,y):
 436         return shash [ max (shash(x),shash(y)) ]
 437     overall['geni_status'] = reduce (combine_status, [ result['geni_status'] for result in results], 'ready' )
 438
 439     # {'ready':0,'configuring':1,'failed':2,'unknown':3}
 440     # append all geni_resources
 441     overall['geni_resources'] = \
 442         reduce (lambda x,y: x+y, [ result['geni_resources'] for result in results] , [])
 443
 444     return overall
 445
 446 def main():
 447     r = RSpec()
 448     r.parseFile(sys.argv[1])
 449     rspec = r.toDict()
 450     CreateSliver(None,'plc.princeton.tmacktestslice',rspec,'create-slice-tmacktestslice')
 451
 452 if __name__ == "__main__":
 453     main()
 454