sfa/managers/slice_manager.py

   1 import sys
   2 import time
   3 from StringIO import StringIO
   4 from copy import copy
   5 from lxml import etree
   6
   7 from sfa.trust.sfaticket import SfaTicket
   8 from sfa.trust.credential import Credential
   9
  10 from sfa.util.sfalogging import logger
  11 from sfa.util.xrn import Xrn, urn_to_hrn
  12 from sfa.util.threadmanager import ThreadManager
  13 from sfa.util.version import version_core
  14 from sfa.util.callids import Callids
  15
  16 from sfa.rspecs.rspec_converter import RSpecConverter
  17 from sfa.rspecs.version_manager import VersionManager
  18 from sfa.rspecs.rspec import RSpec
  19 from sfa.client.client_helper import sfa_to_pg_users_arg
  20
  21 def _call_id_supported(api, server):
  22     """
  23     Returns true if server support the optional call_id arg, false otherwise.
  24     """
  25     server_version = api.get_cached_server_version(server)
  26
  27     if 'sfa' in server_version:
  28         code_tag = server_version['code_tag']
  29         code_tag_parts = code_tag.split("-")
  30
  31         version_parts = code_tag_parts[0].split(".")
  32         major, minor = version_parts[0:2]
  33         rev = code_tag_parts[1]
  34         if int(major) > 1:
  35             if int(minor) > 0 or int(rev) > 20:
  36                 return True
  37     return False
  38
  39 # we have specialized xmlrpclib.ServerProxy to remember the input url
  40 # OTOH it's not clear if we're only dealing with XMLRPCServerProxy instances
  41 def get_serverproxy_url (server):
  42     try:
  43         return server.get_url()
  44     except:
  45         logger.warning("GetVersion, falling back to xmlrpclib.ServerProxy internals")
  46         return server._ServerProxy__host + server._ServerProxy__handler
  47
  48 def GetVersion(api):
  49     # peers explicitly in aggregates.xml
  50     peers =dict ([ (peername,get_serverproxy_url(v)) for (peername,v) in api.aggregates.iteritems()
  51                    if peername != api.hrn])
  52     version_manager = VersionManager()
  53     ad_rspec_versions = []
  54     request_rspec_versions = []
  55     for rspec_version in version_manager.versions:
  56         if rspec_version.content_type in ['*', 'ad']:
  57             ad_rspec_versions.append(rspec_version.to_dict())
  58         if rspec_version.content_type in ['*', 'request']:
  59             request_rspec_versions.append(rspec_version.to_dict())
  60     default_rspec_version = version_manager.get_version("sfa 1").to_dict()
  61     xrn=Xrn(api.hrn, 'authority+sa')
  62     version_more = {'interface':'slicemgr',
  63                     'hrn' : xrn.get_hrn(),
  64                     'urn' : xrn.get_urn(),
  65                     'peers': peers,
  66                     'request_rspec_versions': request_rspec_versions,
  67                     'ad_rspec_versions': ad_rspec_versions,
  68                     'default_ad_rspec': default_rspec_version
  69                     }
  70     sm_version=version_core(version_more)
  71     # local aggregate if present needs to have localhost resolved
  72     if api.hrn in api.aggregates:
  73         local_am_url=get_serverproxy_url(api.aggregates[api.hrn])
  74         sm_version['peers'][api.hrn]=local_am_url.replace('localhost',sm_version['hostname'])
  75     return sm_version
  76
  77 def drop_slicemgr_stats(rspec):
  78     try:
  79         stats_elements = rspec.xml.xpath('//statistics')
  80         for node in stats_elements:
  81             node.getparent().remove(node)
  82     except Exception, e:
  83         logger.warn("drop_slicemgr_stats failed: %s " % (str(e)))
  84
  85 def add_slicemgr_stat(rspec, callname, aggname, elapsed, status):
  86     try:
  87         stats_tags = rspec.xml.xpath('//statistics[@call="%s"]' % callname)
  88         if stats_tags:
  89             stats_tag = stats_tags[0]
  90         else:
  91             stats_tag = etree.SubElement(rspec.xml.root, "statistics", call=callname)
  92
  93         etree.SubElement(stats_tag, "aggregate", name=str(aggname), elapsed=str(elapsed), status=str(status))
  94     except Exception, e:
  95         logger.warn("add_slicemgr_stat failed on  %s: %s" %(aggname, str(e)))
  96
  97 def ListResources(api, creds, options, call_id):
  98     version_manager = VersionManager()
  99     def _ListResources(aggregate, server, credential, opts, call_id):
 100
 101         my_opts = copy(opts)
 102         args = [credential, my_opts]
 103         tStart = time.time()
 104         try:
 105             if _call_id_supported(api, server):
 106                 args.append(call_id)
 107             version = api.get_cached_server_version(server)
 108             # force ProtoGENI aggregates to give us a v2 RSpec
 109             if 'sfa' not in version.keys():
 110                 my_opts['rspec_version'] = version_manager.get_version('ProtoGENI 2').to_dict()
 111             rspec = server.ListResources(*args)
 112             return {"aggregate": aggregate, "rspec": rspec, "elapsed": time.time()-tStart, "status": "success"}
 113         except Exception, e:
 114             api.logger.log_exc("ListResources failed at %s" %(server.url))
 115             return {"aggregate": aggregate, "elapsed": time.time()-tStart, "status": "exception"}
 116
 117     if Callids().already_handled(call_id): return ""
 118
 119     # get slice's hrn from options
 120     xrn = options.get('geni_slice_urn', '')
 121     (hrn, type) = urn_to_hrn(xrn)
 122     if 'geni_compressed' in options:
 123         del(options['geni_compressed'])
 124
 125     # get the rspec's return format from options
 126     rspec_version = version_manager.get_version(options.get('rspec_version'))
 127     version_string = "rspec_%s" % (rspec_version.to_string())
 128
 129     # look in cache first
 130     if caching and api.cache and not xrn:
 131         rspec =  api.cache.get(version_string)
 132         if rspec:
 133             return rspec
 134
 135     # get the callers hrn
 136     valid_cred = api.auth.checkCredentials(creds, 'listnodes', hrn)[0]
 137     caller_hrn = Credential(string=valid_cred).get_gid_caller().get_hrn()
 138
 139     # attempt to use delegated credential first
 140     cred = api.getDelegatedCredential(creds)
 141     if not cred:
 142         cred = api.getCredential()
 143     threads = ThreadManager()
 144     for aggregate in api.aggregates:
 145         # prevent infinite loop. Dont send request back to caller
 146         # unless the caller is the aggregate's SM
 147         if caller_hrn == aggregate and aggregate != api.hrn:
 148             continue
 149
 150         # get the rspec from the aggregate
 151         interface = api.aggregates[aggregate]
 152         server = api.get_server(interface, cred)
 153         threads.run(_ListResources, aggregate, server, [cred], options, call_id)
 154
 155
 156     results = threads.get_results()
 157     rspec_version = version_manager.get_version(options.get('rspec_version'))
 158     if xrn:
 159         result_version = version_manager._get_version(rspec_version.type, rspec_version.version, 'manifest')
 160     else:
 161         result_version = version_manager._get_version(rspec_version.type, rspec_version.version, 'ad')
 162     rspec = RSpec(version=result_version)
 163     for result in results:
 164         add_slicemgr_stat(rspec, "ListResources", result["aggregate"], result["elapsed"], result["status"])
 165         if result["status"]=="success":
 166             try:
 167                 rspec.version.merge(result["rspec"])
 168             except:
 169                 api.logger.log_exc("SM.ListResources: Failed to merge aggregate rspec")
 170
 171     # cache the result
 172     if caching and api.cache and not xrn:
 173         api.cache.add(version_string, rspec.toxml())
 174
 175     return rspec.toxml()
 176
 177
 178 def CreateSliver(api, xrn, creds, rspec_str, users, call_id):
 179
 180     version_manager = VersionManager()
 181     def _CreateSliver(aggregate, server, xrn, credential, rspec, users, call_id):
 182         tStart = time.time()
 183         try:
 184             # Need to call GetVersion at an aggregate to determine the supported
 185             # rspec type/format beofre calling CreateSliver at an Aggregate.
 186             server_version = api.get_cached_server_version(server)
 187             requested_users = users
 188             if 'sfa' not in server_version and 'geni_api' in server_version:
 189                 # sfa aggregtes support both sfa and pg rspecs, no need to convert
 190                 # if aggregate supports sfa rspecs. otherwise convert to pg rspec
 191                 rspec = RSpec(RSpecConverter.to_pg_rspec(rspec, 'request'))
 192                 filter = {'component_manager_id': server_version['urn']}
 193                 rspec.filter(filter)
 194                 rspec = rspec.toxml()
 195                 requested_users = sfa_to_pg_users_arg(users)
 196             args = [xrn, credential, rspec, requested_users]
 197             if _call_id_supported(api, server):
 198                 args.append(call_id)
 199             rspec = server.CreateSliver(*args)
 200             return {"aggregate": aggregate, "rspec": rspec, "elapsed": time.time()-tStart, "status": "success"}
 201         except:
 202             logger.log_exc('Something wrong in _CreateSliver with URL %s'%server.url)
 203             return {"aggregate": aggregate, "elapsed": time.time()-tStart, "status": "exception"}
 204
 205     if Callids().already_handled(call_id): return ""
 206     # Validate the RSpec against PlanetLab's schema --disabled for now
 207     # The schema used here needs to aggregate the PL and VINI schemas
 208     # schema = "/var/www/html/schemas/pl.rng"
 209     rspec = RSpec(rspec_str)
 210 #    schema = None
 211 #    if schema:
 212 #        rspec.validate(schema)
 213
 214     # if there is a <statistics> section, the aggregates don't care about it,
 215     # so delete it.
 216     drop_slicemgr_stats(rspec)
 217
 218     # attempt to use delegated credential first
 219     cred = api.getDelegatedCredential(creds)
 220     if not cred:
 221         cred = api.getCredential()
 222
 223     # get the callers hrn
 224     hrn, type = urn_to_hrn(xrn)
 225     valid_cred = api.auth.checkCredentials(creds, 'createsliver', hrn)[0]
 226     caller_hrn = Credential(string=valid_cred).get_gid_caller().get_hrn()
 227     threads = ThreadManager()
 228     for aggregate in api.aggregates:
 229         # prevent infinite loop. Dont send request back to caller
 230         # unless the caller is the aggregate's SM
 231         if caller_hrn == aggregate and aggregate != api.hrn:
 232             continue
 233         interface = api.aggregates[aggregate]
 234         server = api.get_server(interface, cred)
 235         # Just send entire RSpec to each aggregate
 236         threads.run(_CreateSliver, aggregate, server, xrn, [cred], rspec.toxml(), users, call_id)
 237
 238     results = threads.get_results()
 239     manifest_version = version_manager._get_version(rspec.version.type, rspec.version.version, 'manifest')
 240     result_rspec = RSpec(version=manifest_version)
 241     for result in results:
 242         add_slicemgr_stat(result_rspec, "CreateSliver", result["aggregate"], result["elapsed"], result["status"])
 243         if result["status"]=="success":
 244             try:
 245                 result_rspec.version.merge(result["rspec"])
 246             except:
 247                 api.logger.log_exc("SM.CreateSliver: Failed to merge aggregate rspec")
 248     return result_rspec.toxml()
 249
 250 def RenewSliver(api, xrn, creds, expiration_time, call_id):
 251     def _RenewSliver(server, xrn, creds, expiration_time, call_id):
 252         server_version = api.get_cached_server_version(server)
 253         args =  [xrn, creds, expiration_time, call_id]
 254         if _call_id_supported(api, server):
 255             args.append(call_id)
 256         return server.RenewSliver(*args)
 257
 258     if Callids().already_handled(call_id): return True
 259
 260     (hrn, type) = urn_to_hrn(xrn)
 261     # get the callers hrn
 262     valid_cred = api.auth.checkCredentials(creds, 'renewsliver', hrn)[0]
 263     caller_hrn = Credential(string=valid_cred).get_gid_caller().get_hrn()
 264
 265     # attempt to use delegated credential first
 266     cred = api.getDelegatedCredential(creds)
 267     if not cred:
 268         cred = api.getCredential()
 269     threads = ThreadManager()
 270     for aggregate in api.aggregates:
 271         # prevent infinite loop. Dont send request back to caller
 272         # unless the caller is the aggregate's SM
 273         if caller_hrn == aggregate and aggregate != api.hrn:
 274             continue
 275         interface = api.aggregates[aggregate]
 276         server = api.get_server(interface, cred)
 277         threads.run(_RenewSliver, server, xrn, [cred], expiration_time, call_id)
 278     # 'and' the results
 279     return reduce (lambda x,y: x and y, threads.get_results() , True)
 280
 281 def DeleteSliver(api, xrn, creds, call_id):
 282     def _DeleteSliver(server, xrn, creds, call_id):
 283         server_version = api.get_cached_server_version(server)
 284         args =  [xrn, creds]
 285         if _call_id_supported(api, server):
 286             args.append(call_id)
 287         return server.DeleteSliver(*args)
 288
 289     if Callids().already_handled(call_id): return ""
 290     (hrn, type) = urn_to_hrn(xrn)
 291     # get the callers hrn
 292     valid_cred = api.auth.checkCredentials(creds, 'deletesliver', hrn)[0]
 293     caller_hrn = Credential(string=valid_cred).get_gid_caller().get_hrn()
 294
 295     # attempt to use delegated credential first
 296     cred = api.getDelegatedCredential(creds)
 297     if not cred:
 298         cred = api.getCredential()
 299     threads = ThreadManager()
 300     for aggregate in api.aggregates:
 301         # prevent infinite loop. Dont send request back to caller
 302         # unless the caller is the aggregate's SM
 303         if caller_hrn == aggregate and aggregate != api.hrn:
 304             continue
 305         interface = api.aggregates[aggregate]
 306         server = api.get_server(interface, cred)
 307         threads.run(_DeleteSliver, server, xrn, [cred], call_id)
 308     threads.get_results()
 309     return 1
 310
 311
 312 # first draft at a merging SliverStatus
 313 def SliverStatus(api, slice_xrn, creds, call_id):
 314     def _SliverStatus(server, xrn, creds, call_id):
 315         server_version = api.get_cached_server_version(server)
 316         args =  [xrn, creds]
 317         if _call_id_supported(api, server):
 318             args.append(call_id)
 319         return server.SliverStatus(*args)
 320
 321     if Callids().already_handled(call_id): return {}
 322     # attempt to use delegated credential first
 323     cred = api.getDelegatedCredential(creds)
 324     if not cred:
 325         cred = api.getCredential()
 326     threads = ThreadManager()
 327     for aggregate in api.aggregates:
 328         interface = api.aggregates[aggregate]
 329         server = api.get_server(interface, cred)
 330         threads.run (_SliverStatus, server, slice_xrn, [cred], call_id)
 331     results = threads.get_results()
 332
 333     # get rid of any void result - e.g. when call_id was hit where by convention we return {}
 334     results = [ result for result in results if result and result['geni_resources']]
 335
 336     # do not try to combine if there's no result
 337     if not results : return {}
 338
 339     # otherwise let's merge stuff
 340     overall = {}
 341
 342     # mmh, it is expected that all results carry the same urn
 343     overall['geni_urn'] = results[0]['geni_urn']
 344     overall['pl_login'] = results[0]['pl_login']
 345     # append all geni_resources
 346     overall['geni_resources'] = \
 347         reduce (lambda x,y: x+y, [ result['geni_resources'] for result in results] , [])
 348     overall['status'] = 'unknown'
 349     if overall['geni_resources']:
 350         overall['status'] = 'ready'
 351
 352     return overall
 353
 354 caching=True
 355 #caching=False
 356 def ListSlices(api, creds, call_id):
 357     def _ListSlices(server, creds, call_id):
 358         server_version = api.get_cached_server_version(server)
 359         args =  [creds]
 360         if _call_id_supported(api, server):
 361             args.append(call_id)
 362         return server.ListSlices(*args)
 363
 364     if Callids().already_handled(call_id): return []
 365
 366     # look in cache first
 367     if caching and api.cache:
 368         slices = api.cache.get('slices')
 369         if slices:
 370             return slices
 371
 372     # get the callers hrn
 373     valid_cred = api.auth.checkCredentials(creds, 'listslices', None)[0]
 374     caller_hrn = Credential(string=valid_cred).get_gid_caller().get_hrn()
 375
 376     # attempt to use delegated credential first
 377     cred= api.getDelegatedCredential(creds)
 378     if not cred:
 379         cred = api.getCredential()
 380     threads = ThreadManager()
 381     # fetch from aggregates
 382     for aggregate in api.aggregates:
 383         # prevent infinite loop. Dont send request back to caller
 384         # unless the caller is the aggregate's SM
 385         if caller_hrn == aggregate and aggregate != api.hrn:
 386             continue
 387         interface = api.aggregates[aggregate]
 388         server = api.get_server(interface, cred)
 389         threads.run(_ListSlices, server, [cred], call_id)
 390
 391     # combime results
 392     results = threads.get_results()
 393     slices = []
 394     for result in results:
 395         slices.extend(result)
 396
 397     # cache the result
 398     if caching and api.cache:
 399         api.cache.add('slices', slices)
 400
 401     return slices
 402
 403
 404 def get_ticket(api, xrn, creds, rspec, users):
 405     slice_hrn, type = urn_to_hrn(xrn)
 406     # get the netspecs contained within the clients rspec
 407     aggregate_rspecs = {}
 408     tree= etree.parse(StringIO(rspec))
 409     elements = tree.findall('./network')
 410     for element in elements:
 411         aggregate_hrn = element.values()[0]
 412         aggregate_rspecs[aggregate_hrn] = rspec
 413
 414     # get the callers hrn
 415     valid_cred = api.auth.checkCredentials(creds, 'getticket', slice_hrn)[0]
 416     caller_hrn = Credential(string=valid_cred).get_gid_caller().get_hrn()
 417
 418     # attempt to use delegated credential first
 419     cred = api.getDelegatedCredential(creds)
 420     if not cred:
 421         cred = api.getCredential()
 422     threads = ThreadManager()
 423     for (aggregate, aggregate_rspec) in aggregate_rspecs.iteritems():
 424         # prevent infinite loop. Dont send request back to caller
 425         # unless the caller is the aggregate's SM
 426         if caller_hrn == aggregate and aggregate != api.hrn:
 427             continue
 428
 429         interface = api.aggregates[aggregate]
 430         server = api.get_server(interface, cred)
 431         threads.run(server.GetTicket, xrn, [cred], aggregate_rspec, users)
 432
 433     results = threads.get_results()
 434
 435     # gather information from each ticket
 436     rspec = None
 437     initscripts = []
 438     slivers = []
 439     object_gid = None
 440     for result in results:
 441         agg_ticket = SfaTicket(string=result)
 442         attrs = agg_ticket.get_attributes()
 443         if not object_gid:
 444             object_gid = agg_ticket.get_gid_object()
 445         if not rspec:
 446             rspec = RSpec(agg_ticket.get_rspec())
 447         else:
 448             rspec.version.merge(agg_ticket.get_rspec())
 449         initscripts.extend(attrs.get('initscripts', []))
 450         slivers.extend(attrs.get('slivers', []))
 451
 452     # merge info
 453     attributes = {'initscripts': initscripts,
 454                  'slivers': slivers}
 455
 456     # create a new ticket
 457     ticket = SfaTicket(subject = slice_hrn)
 458     ticket.set_gid_caller(api.auth.client_gid)
 459     ticket.set_issuer(key=api.key, subject=api.hrn)
 460     ticket.set_gid_object(object_gid)
 461     ticket.set_pubkey(object_gid.get_pubkey())
 462     #new_ticket.set_parent(api.auth.hierarchy.get_auth_ticket(auth_hrn))
 463     ticket.set_attributes(attributes)
 464     ticket.set_rspec(rspec.toxml())
 465     ticket.encode()
 466     ticket.sign()
 467     return ticket.save_to_string(save_parents=True)
 468
 469 def start_slice(api, xrn, creds):
 470     hrn, type = urn_to_hrn(xrn)
 471
 472     # get the callers hrn
 473     valid_cred = api.auth.checkCredentials(creds, 'startslice', hrn)[0]
 474     caller_hrn = Credential(string=valid_cred).get_gid_caller().get_hrn()
 475
 476     # attempt to use delegated credential first
 477     cred = api.getDelegatedCredential(creds)
 478     if not cred:
 479         cred = api.getCredential()
 480     threads = ThreadManager()
 481     for aggregate in api.aggregates:
 482         # prevent infinite loop. Dont send request back to caller
 483         # unless the caller is the aggregate's SM
 484         if caller_hrn == aggregate and aggregate != api.hrn:
 485             continue
 486         interface = api.aggregates[aggregate]
 487         server = api.get_server(interface, cred)
 488         threads.run(server.Start, xrn, cred)
 489     threads.get_results()
 490     return 1
 491
 492 def stop_slice(api, xrn, creds):
 493     hrn, type = urn_to_hrn(xrn)
 494
 495     # get the callers hrn
 496     valid_cred = api.auth.checkCredentials(creds, 'stopslice', hrn)[0]
 497     caller_hrn = Credential(string=valid_cred).get_gid_caller().get_hrn()
 498
 499     # attempt to use delegated credential first
 500     cred = api.getDelegatedCredential(creds)
 501     if not cred:
 502         cred = api.getCredential()
 503     threads = ThreadManager()
 504     for aggregate in api.aggregates:
 505         # prevent infinite loop. Dont send request back to caller
 506         # unless the caller is the aggregate's SM
 507         if caller_hrn == aggregate and aggregate != api.hrn:
 508             continue
 509         interface = api.aggregates[aggregate]
 510         server = api.get_server(interface, cred)
 511         threads.run(server.Stop, xrn, cred)
 512     threads.get_results()
 513     return 1
 514
 515 def reset_slice(api, xrn):
 516     """
 517     Not implemented
 518     """
 519     return 1
 520
 521 def shutdown(api, xrn, creds):
 522     """
 523     Not implemented
 524     """
 525     return 1
 526
 527 def status(api, xrn, creds):
 528     """
 529     Not implemented
 530     """
 531     return 1
 532
 533 # this is plain broken
 534 #def main():
 535 #    r = RSpec()
 536 #    r.parseFile(sys.argv[1])
 537 #    rspec = r.toDict()
 538 #    CreateSliver(None,'plc.princeton.tmacktestslice',rspec,'create-slice-tmacktestslice')
 539
 540 if __name__ == "__main__":
 541     main()
 542