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