sfa/managers/aggregate_manager.py

   1 import datetime
   2 import time
   3 import sys
   4
   5 from sfa.util.faults import RecordNotFound, SliverDoesNotExist
   6 from sfa.util.xrn import get_authority, hrn_to_urn, urn_to_hrn, Xrn, urn_to_sliver_id
   7 from sfa.util.plxrn import slicename_to_hrn, hrn_to_pl_slicename
   8 from sfa.util.version import version_core
   9 from sfa.util.sfatime import utcparse
  10 from sfa.util.callids import Callids
  11
  12 from sfa.trust.sfaticket import SfaTicket
  13 from sfa.trust.credential import Credential
  14 from sfa.rspecs.version_manager import VersionManager
  15 from sfa.rspecs.rspec import RSpec
  16
  17 import sfa.plc.peers as peers
  18 from sfa.plc.plcsfaapi import PlcSfaApi
  19 from sfa.plc.aggregate import Aggregate
  20 from sfa.plc.slices import Slices
  21
  22 def GetVersion(api):
  23
  24     version_manager = VersionManager()
  25     ad_rspec_versions = []
  26     request_rspec_versions = []
  27     for rspec_version in version_manager.versions:
  28         if rspec_version.content_type in ['*', 'ad']:
  29             ad_rspec_versions.append(rspec_version.to_dict())
  30         if rspec_version.content_type in ['*', 'request']:
  31             request_rspec_versions.append(rspec_version.to_dict())
  32     default_rspec_version = version_manager.get_version("sfa 1").to_dict()
  33     xrn=Xrn(api.hrn)
  34     version_more = {'interface':'aggregate',
  35                     'testbed':'myplc',
  36                     'hrn':xrn.get_hrn(),
  37                     'request_rspec_versions': request_rspec_versions,
  38                     'ad_rspec_versions': ad_rspec_versions,
  39                     'default_ad_rspec': default_rspec_version
  40                     }
  41     return version_core(version_more)
  42
  43 def __get_registry_objects(slice_xrn, creds, users):
  44     """
  45
  46     """
  47     hrn, _ = urn_to_hrn(slice_xrn)
  48
  49     hrn_auth = get_authority(hrn)
  50
  51     # Build up objects that an SFA registry would return if SFA
  52     # could contact the slice's registry directly
  53     reg_objects = None
  54
  55     if users:
  56         # dont allow special characters in the site login base
  57         #only_alphanumeric = re.compile('[^a-zA-Z0-9]+')
  58         #login_base = only_alphanumeric.sub('', hrn_auth[:20]).lower()
  59         slicename = hrn_to_pl_slicename(hrn)
  60         login_base = slicename.split('_')[0]
  61         reg_objects = {}
  62         site = {}
  63         site['site_id'] = 0
  64         site['name'] = 'geni.%s' % login_base
  65         site['enabled'] = True
  66         site['max_slices'] = 100
  67
  68         # Note:
  69         # Is it okay if this login base is the same as one already at this myplc site?
  70         # Do we need uniqueness?  Should use hrn_auth instead of just the leaf perhaps?
  71         site['login_base'] = login_base
  72         site['abbreviated_name'] = login_base
  73         site['max_slivers'] = 1000
  74         reg_objects['site'] = site
  75
  76         slice = {}
  77
  78         # get_expiration always returns a normalized datetime - no need to utcparse
  79         extime = Credential(string=creds[0]).get_expiration()
  80         # If the expiration time is > 60 days from now, set the expiration time to 60 days from now
  81         if extime > datetime.datetime.utcnow() + datetime.timedelta(days=60):
  82             extime = datetime.datetime.utcnow() + datetime.timedelta(days=60)
  83         slice['expires'] = int(time.mktime(extime.timetuple()))
  84         slice['hrn'] = hrn
  85         slice['name'] = hrn_to_pl_slicename(hrn)
  86         slice['url'] = hrn
  87         slice['description'] = hrn
  88         slice['pointer'] = 0
  89         reg_objects['slice_record'] = slice
  90
  91         reg_objects['users'] = {}
  92         for user in users:
  93             user['key_ids'] = []
  94             hrn, _ = urn_to_hrn(user['urn'])
  95             user['email'] = hrn_to_pl_slicename(hrn) + "@geni.net"
  96             user['first_name'] = hrn
  97             user['last_name'] = hrn
  98             reg_objects['users'][user['email']] = user
  99
 100         return reg_objects
 101
 102 def __get_hostnames(nodes):
 103     hostnames = []
 104     for node in nodes:
 105         hostnames.append(node.hostname)
 106     return hostnames
 107
 108 def SliverStatus(api, slice_xrn, creds, call_id):
 109     if Callids().already_handled(call_id): return {}
 110
 111     (hrn, _) = urn_to_hrn(slice_xrn)
 112     # find out where this slice is currently running
 113     slicename = hrn_to_pl_slicename(hrn)
 114
 115     slices = api.plshell.GetSlices(api.plauth, [slicename], ['slice_id', 'node_ids','person_ids','name','expires'])
 116     if len(slices) == 0:
 117         raise Exception("Slice %s not found (used %s as slicename internally)" % (slice_xrn, slicename))
 118     slice = slices[0]
 119
 120     # report about the local nodes only
 121     nodes = api.plshell.GetNodes(api.plauth, {'node_id':slice['node_ids'],'peer_id':None},
 122                                  ['node_id', 'hostname', 'site_id', 'boot_state', 'last_contact'])
 123     site_ids = [node['site_id'] for node in nodes]
 124
 125     result = {}
 126     top_level_status = 'unknown'
 127     if nodes:
 128         top_level_status = 'ready'
 129     slice_urn = Xrn(slice_xrn, 'slice').get_urn()
 130     result['geni_urn'] = slice_urn
 131     result['pl_login'] = slice['name']
 132     result['pl_expires'] = datetime.datetime.fromtimestamp(slice['expires']).ctime()
 133
 134     resources = []
 135     for node in nodes:
 136         res = {}
 137         res['pl_hostname'] = node['hostname']
 138         res['pl_boot_state'] = node['boot_state']
 139         res['pl_last_contact'] = node['last_contact']
 140         if node['last_contact'] is not None:
 141             res['pl_last_contact'] = datetime.datetime.fromtimestamp(node['last_contact']).ctime()
 142         sliver_id = urn_to_sliver_id(slice_urn, slice['slice_id'], node['node_id'])
 143         res['geni_urn'] = sliver_id
 144         if node['boot_state'] == 'boot':
 145             res['geni_status'] = 'ready'
 146         else:
 147             res['geni_status'] = 'failed'
 148             top_level_status = 'failed'
 149
 150         res['geni_error'] = ''
 151
 152         resources.append(res)
 153
 154     result['geni_status'] = top_level_status
 155     result['geni_resources'] = resources
 156     return result
 157
 158 def CreateSliver(api, slice_xrn, creds, rspec_string, users, call_id):
 159     """
 160     Create the sliver[s] (slice) at this aggregate.
 161     Verify HRN and initialize the slice record in PLC if necessary.
 162     """
 163     if Callids().already_handled(call_id): return ""
 164
 165     aggregate = Aggregate(api)
 166     slices = Slices(api)
 167     (hrn, _) = urn_to_hrn(slice_xrn)
 168     peer = slices.get_peer(hrn)
 169     sfa_peer = slices.get_sfa_peer(hrn)
 170     slice_record=None
 171     if users:
 172         slice_record = users[0].get('slice_record', {})
 173
 174     # parse rspec
 175     rspec = RSpec(rspec_string)
 176     requested_attributes = rspec.version.get_slice_attributes()
 177
 178     # ensure site record exists
 179     site = slices.verify_site(hrn, slice_record, peer, sfa_peer)
 180     # ensure slice record exists
 181     slice = slices.verify_slice(hrn, slice_record, peer, sfa_peer)
 182     # ensure person records exists
 183     persons = slices.verify_persons(hrn, slice, users, peer, sfa_peer)
 184     # ensure slice attributes exists
 185     slices.verify_slice_attributes(slice, requested_attributes)
 186
 187     # add/remove slice from nodes
 188     requested_slivers = [str(host) for host in rspec.version.get_nodes_with_slivers()]
 189     slices.verify_slice_nodes(slice, requested_slivers, peer)
 190
 191     slices.verify_slice_links(slice, rspec.version.get_link_requests(), aggregate)
 192
 193     # hanlde MyPLC peer association.
 194     # only used by plc and ple.
 195     slices.handle_peer(site, slice, persons, peer)
 196
 197     return aggregate.get_rspec(slice_xrn=slice_xrn, version=rspec.version)
 198
 199
 200 def RenewSliver(api, xrn, creds, expiration_time, call_id):
 201     if Callids().already_handled(call_id): return True
 202     (hrn, _) = urn_to_hrn(xrn)
 203     slicename = hrn_to_pl_slicename(hrn)
 204     slices = api.plshell.GetSlices(api.plauth, {'name': slicename}, ['slice_id'])
 205     if not slices:
 206         raise RecordNotFound(hrn)
 207     slice = slices[0]
 208     requested_time = utcparse(expiration_time)
 209     record = {'expires': int(time.mktime(requested_time.timetuple()))}
 210     try:
 211         api.plshell.UpdateSlice(api.plauth, slice['slice_id'], record)
 212         return True
 213     except:
 214         return False
 215
 216 def start_slice(api, xrn, creds):
 217     (hrn, _) = urn_to_hrn(xrn)
 218     slicename = hrn_to_pl_slicename(hrn)
 219     slices = api.plshell.GetSlices(api.plauth, {'name': slicename}, ['slice_id'])
 220     if not slices:
 221         raise RecordNotFound(hrn)
 222     slice_id = slices[0]['slice_id']
 223     slice_tags = api.plshell.GetSliceTags(api.plauth, {'slice_id': slice_id, 'tagname': 'enabled'}, ['slice_tag_id'])
 224     # just remove the tag if it exists
 225     if slice_tags:
 226         api.plshell.DeleteSliceTag(api.plauth, slice_tags[0]['slice_tag_id'])
 227
 228     return 1
 229
 230 def stop_slice(api, xrn, creds):
 231     hrn, _ = urn_to_hrn(xrn)
 232     slicename = hrn_to_pl_slicename(hrn)
 233     slices = api.plshell.GetSlices(api.plauth, {'name': slicename}, ['slice_id'])
 234     if not slices:
 235         raise RecordNotFound(hrn)
 236     slice_id = slices[0]['slice_id']
 237     slice_tags = api.plshell.GetSliceTags(api.plauth, {'slice_id': slice_id, 'tagname': 'enabled'})
 238     if not slice_tags:
 239         api.plshell.AddSliceTag(api.plauth, slice_id, 'enabled', '0')
 240     elif slice_tags[0]['value'] != "0":
 241         tag_id = slice_tags[0]['slice_tag_id']
 242         api.plshell.UpdateSliceTag(api.plauth, tag_id, '0')
 243     return 1
 244
 245 def reset_slice(api, xrn):
 246     # XX not implemented at this interface
 247     return 1
 248
 249 def DeleteSliver(api, xrn, creds, call_id):
 250     if Callids().already_handled(call_id): return ""
 251     (hrn, _) = urn_to_hrn(xrn)
 252     slicename = hrn_to_pl_slicename(hrn)
 253     slices = api.plshell.GetSlices(api.plauth, {'name': slicename})
 254     if not slices:
 255         return 1
 256     slice = slices[0]
 257
 258     # determine if this is a peer slice
 259     peer = peers.get_peer(api, hrn)
 260     try:
 261         if peer:
 262             api.plshell.UnBindObjectFromPeer(api.plauth, 'slice', slice['slice_id'], peer)
 263         api.plshell.DeleteSliceFromNodes(api.plauth, slicename, slice['node_ids'])
 264     finally:
 265         if peer:
 266             api.plshell.BindObjectToPeer(api.plauth, 'slice', slice['slice_id'], peer, slice['peer_slice_id'])
 267     return 1
 268
 269 # xxx Thierry : caching at the aggregate level sounds wrong...
 270 #caching=True
 271 caching=False
 272 def ListSlices(api, creds, call_id):
 273     if Callids().already_handled(call_id): return []
 274     # look in cache first
 275     if caching and api.cache:
 276         slices = api.cache.get('slices')
 277         if slices:
 278             return slices
 279
 280     # get data from db
 281     slices = api.plshell.GetSlices(api.plauth, {'peer_id': None}, ['name'])
 282     slice_hrns = [slicename_to_hrn(api.hrn, slice['name']) for slice in slices]
 283     slice_urns = [hrn_to_urn(slice_hrn, 'slice') for slice_hrn in slice_hrns]
 284
 285     # cache the result
 286     if caching and api.cache:
 287         api.cache.add('slices', slice_urns)
 288
 289     return slice_urns
 290
 291 def ListResources(api, creds, options, call_id):
 292     if Callids().already_handled(call_id): return ""
 293     # get slice's hrn from options
 294     xrn = options.get('geni_slice_urn', None)
 295     (hrn, _) = urn_to_hrn(xrn)
 296
 297     version_manager = VersionManager()
 298     # get the rspec's return format from options
 299     rspec_version = version_manager.get_version(options.get('rspec_version'))
 300     version_string = "rspec_%s" % (rspec_version.to_string())
 301
 302     #panos adding the info option to the caching key (can be improved)
 303     if options.get('info'):
 304         version_string = version_string + "_"+options.get('info', 'default')
 305
 306     # look in cache first
 307     if caching and api.cache and not xrn:
 308         rspec = api.cache.get(version_string)
 309         if rspec:
 310             api.logger.info("aggregate.ListResources: returning cached value for hrn %s"%hrn)
 311             return rspec
 312
 313     #panos: passing user-defined options
 314     #print "manager options = ",options
 315     aggregate = Aggregate(api, options)
 316     rspec =  aggregate.get_rspec(slice_xrn=xrn, version=rspec_version)
 317
 318     # cache the result
 319     if caching and api.cache and not xrn:
 320         api.cache.add(version_string, rspec)
 321
 322     return rspec
 323
 324
 325 def get_ticket(api, xrn, creds, rspec, users):
 326
 327     (slice_hrn, _) = urn_to_hrn(xrn)
 328     slices = Slices(api)
 329     peer = slices.get_peer(slice_hrn)
 330     sfa_peer = slices.get_sfa_peer(slice_hrn)
 331
 332     # get the slice record
 333     credential = api.getCredential()
 334     interface = api.registries[api.hrn]
 335     registry = api.get_server(interface, credential)
 336     records = registry.Resolve(xrn, credential)
 337
 338     # make sure we get a local slice record
 339     record = None
 340     for tmp_record in records:
 341         if tmp_record['type'] == 'slice' and \
 342            not tmp_record['peer_authority']:
 343 #Error (E0602, get_ticket): Undefined variable 'SliceRecord'
 344             record = SliceRecord(dict=tmp_record)
 345     if not record:
 346         raise RecordNotFound(slice_hrn)
 347
 348     # similar to CreateSliver, we must verify that the required records exist
 349     # at this aggregate before we can issue a ticket
 350     # parse rspec
 351     rspec = RSpec(rspec_string)
 352     requested_attributes = rspec.version.get_slice_attributes()
 353
 354     # ensure site record exists
 355     site = slices.verify_site(hrn, slice_record, peer, sfa_peer)
 356     # ensure slice record exists
 357     slice = slices.verify_slice(hrn, slice_record, peer, sfa_peer)
 358     # ensure person records exists
 359     persons = slices.verify_persons(hrn, slice, users, peer, sfa_peer)
 360     # ensure slice attributes exists
 361     slices.verify_slice_attributes(slice, requested_attributes)
 362
 363     # get sliver info
 364     slivers = slices.get_slivers(slice_hrn)
 365
 366     if not slivers:
 367         raise SliverDoesNotExist(slice_hrn)
 368
 369     # get initscripts
 370     initscripts = []
 371     data = {
 372         'timestamp': int(time.time()),
 373         'initscripts': initscripts,
 374         'slivers': slivers
 375     }
 376
 377     # create the ticket
 378     object_gid = record.get_gid_object()
 379     new_ticket = SfaTicket(subject = object_gid.get_subject())
 380     new_ticket.set_gid_caller(api.auth.client_gid)
 381     new_ticket.set_gid_object(object_gid)
 382     new_ticket.set_issuer(key=api.key, subject=api.hrn)
 383     new_ticket.set_pubkey(object_gid.get_pubkey())
 384     new_ticket.set_attributes(data)
 385     new_ticket.set_rspec(rspec)
 386     #new_ticket.set_parent(api.auth.hierarchy.get_auth_ticket(auth_hrn))
 387     new_ticket.encode()
 388     new_ticket.sign()
 389
 390     return new_ticket.save_to_string(save_parents=True)
 391
 392
 393
 394 def main():
 395     """
 396     rspec = ListResources(api, "plc.princeton.sapan", None, 'pl_test_sapan')
 397     #rspec = ListResources(api, "plc.princeton.coblitz", None, 'pl_test_coblitz')
 398     #rspec = ListResources(api, "plc.pl.sirius", None, 'pl_test_sirius')
 399     print rspec
 400     """
 401     api = PlcSfaApi()
 402     f = open(sys.argv[1])
 403     xml = f.read()
 404     f.close()
 405 #Error (E1120, main): No value passed for parameter 'users' in function call
 406 #Error (E1120, main): No value passed for parameter 'call_id' in function call
 407     CreateSliver(api, "plc.princeton.sapan", xml, 'CreateSliver_sapan')
 408
 409 if __name__ == "__main__":
 410     main()