sfa/managers/slice_manager_pl.py

   1 ### $Id: slices.py 15842 2009-11-22 09:56:13Z anil $
   2 ### $URL: https://svn.planet-lab.org/svn/sfa/trunk/sfa/plc/slices.py $
   3
   4 import datetime
   5 import time
   6 import traceback
   7 import sys
   8 from copy import deepcopy
   9 from lxml import etree
  10 from StringIO import StringIO
  11 from types import StringTypes
  12 from sfa.util.rspecHelper import merge_rspecs
  13 from sfa.util.namespace import *
  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.util.policy import Policy
  19 from sfa.util.prefixTree import prefixTree
  20 from sfa.util.sfaticket import *
  21 from sfa.trust.credential import Credential
  22 from sfa.util.threadmanager import ThreadManager
  23 import sfa.util.xmlrpcprotocol as xmlrpcprotocol
  24 import sfa.plc.peers as peers
  25 from copy import copy
  26
  27 def get_version():
  28     version = {}
  29     version['geni_api'] = 1
  30     version['sfa'] = 1
  31     return version
  32
  33 def slice_status(api, slice_xrn, creds ):
  34     result = {}
  35     result['geni_urn'] = slice_xrn
  36     result['geni_status'] = 'unknown'
  37     result['geni_resources'] = {}
  38     return result
  39
  40 def create_slice(api, xrn, creds, rspec, users):
  41     hrn, type = urn_to_hrn(xrn)
  42
  43     # Validate the RSpec against PlanetLab's schema --disabled for now
  44     # The schema used here needs to aggregate the PL and VINI schemas
  45     # schema = "/var/www/html/schemas/pl.rng"
  46     schema = None
  47     if schema:
  48         try:
  49             tree = etree.parse(StringIO(rspec))
  50         except etree.XMLSyntaxError:
  51             message = str(sys.exc_info()[1])
  52             raise InvalidRSpec(message)
  53
  54         relaxng_doc = etree.parse(schema)
  55         relaxng = etree.RelaxNG(relaxng_doc)
  56
  57         if not relaxng(tree):
  58             error = relaxng.error_log.last_error
  59             message = "%s (line %s)" % (error.message, error.line)
  60             raise InvalidRSpec(message)
  61
  62     # get the callers hrn
  63     valid_cred = api.auth.checkCredentials(creds, 'createsliver', hrn)[0]
  64     caller_hrn = Credential(string=valid_cred).get_gid_caller().get_hrn()
  65
  66     # attempt to use delegated credential first
  67     credential = api.getDelegatedCredential(creds)
  68     if not credential:
  69         credential = api.getCredential()
  70     threads = ThreadManager()
  71     for aggregate in api.aggregates:
  72         # prevent infinite loop. Dont send request back to caller
  73         # unless the caller is the aggregate's SM
  74         if caller_hrn == aggregate and aggregate != api.hrn:
  75             continue
  76
  77         # Just send entire RSpec to each aggregate
  78         server = api.aggregates[aggregate]
  79         threads.run(server.CreateSliver, xrn, credential, rspec, users)
  80
  81     results = threads.get_results()
  82     merged_rspec = merge_rspecs(results)
  83     return merged_rspec
  84
  85 def renew_slice(api, xrn, creds, expiration_time):
  86     # get the callers hrn
  87     valid_cred = api.auth.checkCredentials(creds, 'renewesliver', hrn)[0]
  88     caller_hrn = Credential(string=valid_cred).get_gid_caller().get_hrn()
  89
  90     # attempt to use delegated credential first
  91     credential = api.getDelegatedCredential(creds)
  92     if not credential:
  93         credential = api.getCredential()
  94     threads = ThreadManager()
  95     for aggregate in api.aggregates:
  96         # prevent infinite loop. Dont send request back to caller
  97         # unless the caller is the aggregate's SM
  98         if caller_hrn == aggregate and aggregate != api.hrn:
  99             continue
 100
 101         server = api.aggregates[aggregate]
 102         threads.run(server.RenewSliver, xrn, credential, expiration_time)
 103     threads.get_results()
 104     return 1
 105
 106 def get_ticket(api, xrn, creds, rspec, users):
 107     slice_hrn, type = urn_to_hrn(xrn)
 108     # get the netspecs contained within the clients rspec
 109     aggregate_rspecs = {}
 110     tree= etree.parse(StringIO(rspec))
 111     elements = tree.findall('./network')
 112     for element in elements:
 113         aggregate_hrn = element.values()[0]
 114         aggregate_rspecs[aggregate_hrn] = rspec
 115
 116     # get the callers hrn
 117     valid_cred = api.auth.checkCredentials(creds, 'getticket', hrn)[0]
 118     caller_hrn = Credential(string=valid_cred).get_gid_caller().get_hrn()
 119
 120     # attempt to use delegated credential first
 121     credential = api.getDelegatedCredential(creds)
 122     if not credential:
 123         credential = api.getCredential()
 124     threads = ThreadManager()
 125     for aggregate, aggregate_rspec in aggregate_rspecs.items():
 126         # prevent infinite loop. Dont send request back to caller
 127         # unless the caller is the aggregate's SM
 128         if caller_hrn == aggregate and aggregate != api.hrn:
 129             continue
 130         server = None
 131         if aggregate in api.aggregates:
 132             server = api.aggregates[aggregate]
 133         else:
 134             net_urn = hrn_to_urn(aggregate, 'authority')
 135             # we may have a peer that knows about this aggregate
 136             for agg in api.aggregates:
 137                 target_aggs = api.aggregates[agg].get_aggregates(credential, net_urn)
 138                 if not target_aggs or not 'hrn' in target_aggs[0]:
 139                     continue
 140                 # send the request to this address
 141                 url = target_aggs[0]['url']
 142                 server = xmlrpcprotocol.get_server(url, api.key_file, api.cert_file)
 143                 # aggregate found, no need to keep looping
 144                 break
 145         if server is None:
 146             continue
 147         threads.run(server.GetTicket, xrn, credential, aggregate_rspec, users)
 148
 149     results = threads.get_results()
 150
 151     # gather information from each ticket
 152     rspecs = []
 153     initscripts = []
 154     slivers = []
 155     object_gid = None
 156     for result in results:
 157         agg_ticket = SfaTicket(string=result)
 158         attrs = agg_ticket.get_attributes()
 159         if not object_gid:
 160             object_gid = agg_ticket.get_gid_object()
 161         rspecs.append(agg_ticket.get_rspec())
 162         initscripts.extend(attrs.get('initscripts', []))
 163         slivers.extend(attrs.get('slivers', []))
 164
 165     # merge info
 166     attributes = {'initscripts': initscripts,
 167                  'slivers': slivers}
 168     merged_rspec = merge_rspecs(rspecs)
 169
 170     # create a new ticket
 171     ticket = SfaTicket(subject = slice_hrn)
 172     ticket.set_gid_caller(api.auth.client_gid)
 173     ticket.set_issuer(key=api.key, subject=api.hrn)
 174     ticket.set_gid_object(object_gid)
 175     ticket.set_pubkey(object_gid.get_pubkey())
 176     #new_ticket.set_parent(api.auth.hierarchy.get_auth_ticket(auth_hrn))
 177     ticket.set_attributes(attributes)
 178     ticket.set_rspec(merged_rspec)
 179     ticket.encode()
 180     ticket.sign()
 181     return ticket.save_to_string(save_parents=True)
 182
 183
 184 def delete_slice(api, xrn, creds):
 185     # get the callers hrn
 186     valid_cred = api.auth.checkCredentials(creds, 'deletesliver', hrn)[0]
 187     caller_hrn = Credential(string=valid_cred).get_gid_caller().get_hrn()
 188
 189     # attempt to use delegated credential first
 190     credential = api.getDelegatedCredential(creds)
 191     if not credential:
 192         credential = api.getCredential()
 193     threads = ThreadManager()
 194     for aggregate in api.aggregates:
 195         # prevent infinite loop. Dont send request back to caller
 196         # unless the caller is the aggregate's SM
 197         if caller_hrn == aggregate and aggregate != api.hrn:
 198             continue
 199         server = api.aggregates[aggregate]
 200         threads.run(server.DeleteSliver, xrn, credential)
 201     threads.get_results()
 202     return 1
 203
 204 def start_slice(api, xrn, creds):
 205     # get the callers hrn
 206     valid_cred = api.auth.checkCredentials(creds, 'startslice', hrn)[0]
 207     caller_hrn = Credential(string=valid_cred).get_gid_caller().get_hrn()
 208
 209     # attempt to use delegated credential first
 210     credential = api.getDelegatedCredential(creds)
 211     if not credential:
 212         credential = api.getCredential()
 213     threads = ThreadManager()
 214     for aggregate in api.aggregates:
 215         # prevent infinite loop. Dont send request back to caller
 216         # unless the caller is the aggregate's SM
 217         if caller_hrn == aggregate and aggregate != api.hrn:
 218             continue
 219         server = api.aggregates[aggregate]
 220         threads.run(server.Start, xrn, credential)
 221     threads.get_results()
 222     return 1
 223
 224 def stop_slice(api, xrn, creds):
 225     # get the callers hrn
 226     valid_cred = api.auth.checkCredentials(creds, 'stopslice', hrn)[0]
 227     caller_hrn = Credential(string=valid_cred).get_gid_caller().get_hrn()
 228
 229     # attempt to use delegated credential first
 230     credential = api.getDelegatedCredential(creds)
 231     if not credential:
 232         credential = api.getCredential()
 233     threads = ThreadManager()
 234     for aggregate in api.aggregates:
 235         # prevent infinite loop. Dont send request back to caller
 236         # unless the caller is the aggregate's SM
 237         if caller_hrn == aggregate and aggregate != api.hrn:
 238             continue
 239         server = api.aggregates[aggregate]
 240         threads.run(server.Stop, xrn, credential)
 241     threads.get_results()
 242     return 1
 243
 244 def reset_slice(api, xrn):
 245     """
 246     Not implemented
 247     """
 248     return 1
 249
 250 def shutdown(api, xrn, creds):
 251     """
 252     Not implemented
 253     """
 254     return 1
 255
 256 def status(api, xrn, creds):
 257     """
 258     Not implemented
 259     """
 260     return 1
 261
 262 def get_slices(api, creds):
 263
 264     # look in cache first
 265     if api.cache:
 266         slices = api.cache.get('slices')
 267         if slices:
 268             return slices
 269
 270     # get the callers hrn
 271     valid_cred = api.auth.checkCredentials(creds, 'listslices', hrn)[0]
 272     caller_hrn = Credential(string=valid_cred).get_gid_caller().get_hrn()
 273
 274     # attempt to use delegated credential first
 275     credential = api.getDelegatedCredential(creds)
 276     if not credential:
 277         credential = api.getCredential()
 278     threads = ThreadManager()
 279     # fetch from aggregates
 280     for aggregate in api.aggregates:
 281         # prevent infinite loop. Dont send request back to caller
 282         # unless the caller is the aggregate's SM
 283         if caller_hrn == aggregate and aggregate != api.hrn:
 284             continue
 285         server = api.aggregates[aggregate]
 286         threads.run(server.ListSlices, credential)
 287
 288     # combime results
 289     results = threads.get_results()
 290     slices = []
 291     for result in results:
 292         slices.extend(result)
 293
 294     # cache the result
 295     if api.cache:
 296         api.cache.add('slices', slices)
 297
 298     return slices
 299
 300 def get_rspec(api, creds, options):
 301
 302     # get slice's hrn from options
 303     xrn = options.get('geni_slice_urn', None)
 304     hrn, type = urn_to_hrn(xrn)
 305
 306     # get hrn of the original caller
 307     origin_hrn = options.get('origin_hrn', None)
 308     if not origin_hrn:
 309         origin_hrn = Credential(string=creds[0]).get_gid_caller().get_hrn()
 310
 311     # look in cache first
 312     if api.cache and not xrn:
 313         rspec =  api.cache.get('nodes')
 314         if rspec:
 315             return rspec
 316
 317     hrn, type = urn_to_hrn(xrn)
 318     rspec = None
 319
 320     # get the callers hrn
 321     valid_cred = api.auth.checkCredentials(creds, 'listnodes', hrn)[0]
 322     caller_hrn = Credential(string=valid_cred).get_gid_caller().get_hrn()
 323
 324     # attempt to use delegated credential first
 325     credential = api.getDelegatedCredential(creds)
 326     if not credential:
 327         credential = api.getCredential()
 328     threads = ThreadManager()
 329     for aggregate in api.aggregates:
 330         # prevent infinite loop. Dont send request back to caller
 331         # unless the caller is the aggregate's SM
 332         if caller_hrn == aggregate and aggregate != api.hrn:
 333             continue
 334         # get the rspec from the aggregate
 335         server = api.aggregates[aggregate]
 336         my_opts = copy(options)
 337         my_opts['geni_compressed'] = False
 338         threads.run(server.ListResources, credential, my_opts)
 339         #threads.run(server.get_resources, cred, xrn, origin_hrn)
 340
 341     results = threads.get_results()
 342     # combine the rspecs into a single rspec
 343     for agg_rspec in results:
 344         try:
 345             tree = etree.parse(StringIO(agg_rspec))
 346         except etree.XMLSyntaxError:
 347             message = str(agg_rspec) + ": " + str(sys.exc_info()[1])
 348             raise InvalidRSpec(message)
 349
 350         root = tree.getroot()
 351         if root.get("type") in ["SFA"]:
 352             if rspec == None:
 353                 rspec = root
 354             else:
 355                 for network in root.iterfind("./network"):
 356                     rspec.append(deepcopy(network))
 357                 for request in root.iterfind("./request"):
 358                     rspec.append(deepcopy(request))
 359
 360     rspec =  etree.tostring(rspec, xml_declaration=True, pretty_print=True)
 361     # cache the result
 362     if api.cache and not xrn:
 363         api.cache.add('nodes', rspec)
 364
 365     return rspec
 366
 367 def main():
 368     r = RSpec()
 369     r.parseFile(sys.argv[1])
 370     rspec = r.toDict()
 371     create_slice(None,'plc.princeton.tmacktestslice',rspec)
 372
 373 if __name__ == "__main__":
 374     main()
 375