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     aggregate.prepare_nodes({'hostname': requested_slivers})
 192     aggregate.prepare_interfaces({'node_id': aggregate.nodes.keys()})
 193     slices.verify_slice_links(slice, rspec.version.get_link_requests(), aggregate)
 194
 195     # handle MyPLC peer association.
 196     # only used by plc and ple.
 197     slices.handle_peer(site, slice, persons, peer)
 198
 199     return aggregate.get_rspec(slice_xrn=slice_xrn, version=rspec.version)
 200
 201
 202 def RenewSliver(api, xrn, creds, expiration_time, call_id):
 203     if Callids().already_handled(call_id): return True
 204     (hrn, _) = urn_to_hrn(xrn)
 205     slicename = hrn_to_pl_slicename(hrn)
 206     slices = api.plshell.GetSlices(api.plauth, {'name': slicename}, ['slice_id'])
 207     if not slices:
 208         raise RecordNotFound(hrn)
 209     slice = slices[0]
 210     requested_time = utcparse(expiration_time)
 211     record = {'expires': int(time.mktime(requested_time.timetuple()))}
 212     try:
 213         api.plshell.UpdateSlice(api.plauth, slice['slice_id'], record)
 214         return True
 215     except:
 216         return False
 217
 218 def start_slice(api, xrn, creds):
 219     (hrn, _) = urn_to_hrn(xrn)
 220     slicename = hrn_to_pl_slicename(hrn)
 221     slices = api.plshell.GetSlices(api.plauth, {'name': slicename}, ['slice_id'])
 222     if not slices:
 223         raise RecordNotFound(hrn)
 224     slice_id = slices[0]['slice_id']
 225     slice_tags = api.plshell.GetSliceTags(api.plauth, {'slice_id': slice_id, 'tagname': 'enabled'}, ['slice_tag_id'])
 226     # just remove the tag if it exists
 227     if slice_tags:
 228         api.plshell.DeleteSliceTag(api.plauth, slice_tags[0]['slice_tag_id'])
 229
 230     return 1
 231
 232 def stop_slice(api, xrn, creds):
 233     hrn, _ = urn_to_hrn(xrn)
 234     slicename = hrn_to_pl_slicename(hrn)
 235     slices = api.plshell.GetSlices(api.plauth, {'name': slicename}, ['slice_id'])
 236     if not slices:
 237         raise RecordNotFound(hrn)
 238     slice_id = slices[0]['slice_id']
 239     slice_tags = api.plshell.GetSliceTags(api.plauth, {'slice_id': slice_id, 'tagname': 'enabled'})
 240     if not slice_tags:
 241         api.plshell.AddSliceTag(api.plauth, slice_id, 'enabled', '0')
 242     elif slice_tags[0]['value'] != "0":
 243         tag_id = slice_tags[0]['slice_tag_id']
 244         api.plshell.UpdateSliceTag(api.plauth, tag_id, '0')
 245     return 1
 246
 247 def reset_slice(api, xrn):
 248     # XX not implemented at this interface
 249     return 1
 250
 251 def DeleteSliver(api, xrn, creds, call_id):
 252     if Callids().already_handled(call_id): return ""
 253     (hrn, _) = urn_to_hrn(xrn)
 254     slicename = hrn_to_pl_slicename(hrn)
 255     slices = api.plshell.GetSlices(api.plauth, {'name': slicename})
 256     if not slices:
 257         return 1
 258     slice = slices[0]
 259
 260     # determine if this is a peer slice
 261     peer = peers.get_peer(api, hrn)
 262     try:
 263         if peer:
 264             api.plshell.UnBindObjectFromPeer(api.plauth, 'slice', slice['slice_id'], peer)
 265         api.plshell.DeleteSliceFromNodes(api.plauth, slicename, slice['node_ids'])
 266     finally:
 267         if peer:
 268             api.plshell.BindObjectToPeer(api.plauth, 'slice', slice['slice_id'], peer, slice['peer_slice_id'])
 269     return 1
 270
 271 # xxx Thierry : caching at the aggregate level sounds wrong...
 272 #caching=True
 273 caching=False
 274 def ListSlices(api, creds, call_id):
 275     if Callids().already_handled(call_id): return []
 276     # look in cache first
 277     if caching and api.cache:
 278         slices = api.cache.get('slices')
 279         if slices:
 280             return slices
 281
 282     # get data from db
 283     slices = api.plshell.GetSlices(api.plauth, {'peer_id': None}, ['name'])
 284     slice_hrns = [slicename_to_hrn(api.hrn, slice['name']) for slice in slices]
 285     slice_urns = [hrn_to_urn(slice_hrn, 'slice') for slice_hrn in slice_hrns]
 286
 287     # cache the result
 288     if caching and api.cache:
 289         api.cache.add('slices', slice_urns)
 290
 291     return slice_urns
 292
 293 def ListResources(api, creds, options, call_id):
 294     if Callids().already_handled(call_id): return ""
 295     # get slice's hrn from options
 296     xrn = options.get('geni_slice_urn', None)
 297     (hrn, _) = urn_to_hrn(xrn)
 298
 299     version_manager = VersionManager()
 300     # get the rspec's return format from options
 301     rspec_version = version_manager.get_version(options.get('rspec_version'))
 302     version_string = "rspec_%s" % (rspec_version.to_string())
 303
 304     #panos adding the info option to the caching key (can be improved)
 305     if options.get('info'):
 306         version_string = version_string + "_"+options.get('info', 'default')
 307
 308     # look in cache first
 309     if caching and api.cache and not xrn:
 310         rspec = api.cache.get(version_string)
 311         if rspec:
 312             api.logger.info("aggregate.ListResources: returning cached value for hrn %s"%hrn)
 313             return rspec
 314
 315     #panos: passing user-defined options
 316     #print "manager options = ",options
 317     aggregate = Aggregate(api, options)
 318     rspec =  aggregate.get_rspec(slice_xrn=xrn, version=rspec_version)
 319
 320     # cache the result
 321     if caching and api.cache and not xrn:
 322         api.cache.add(version_string, rspec)
 323
 324     return rspec
 325
 326
 327 def get_ticket(api, xrn, creds, rspec, users):
 328
 329     (slice_hrn, _) = urn_to_hrn(xrn)
 330     slices = Slices(api)
 331     peer = slices.get_peer(slice_hrn)
 332     sfa_peer = slices.get_sfa_peer(slice_hrn)
 333
 334     # get the slice record
 335     credential = api.getCredential()
 336     interface = api.registries[api.hrn]
 337     registry = api.get_server(interface, credential)
 338     records = registry.Resolve(xrn, credential)
 339
 340     # make sure we get a local slice record
 341     record = None
 342     for tmp_record in records:
 343         if tmp_record['type'] == 'slice' and \
 344            not tmp_record['peer_authority']:
 345 #Error (E0602, get_ticket): Undefined variable 'SliceRecord'
 346             record = SliceRecord(dict=tmp_record)
 347     if not record:
 348         raise RecordNotFound(slice_hrn)
 349
 350     # similar to CreateSliver, we must verify that the required records exist
 351     # at this aggregate before we can issue a ticket
 352     # parse rspec
 353     rspec = RSpec(rspec_string)
 354     requested_attributes = rspec.version.get_slice_attributes()
 355
 356     # ensure site record exists
 357     site = slices.verify_site(hrn, slice_record, peer, sfa_peer)
 358     # ensure slice record exists
 359     slice = slices.verify_slice(hrn, slice_record, peer, sfa_peer)
 360     # ensure person records exists
 361     persons = slices.verify_persons(hrn, slice, users, peer, sfa_peer)
 362     # ensure slice attributes exists
 363     slices.verify_slice_attributes(slice, requested_attributes)
 364
 365     # get sliver info
 366     slivers = slices.get_slivers(slice_hrn)
 367
 368     if not slivers:
 369         raise SliverDoesNotExist(slice_hrn)
 370
 371     # get initscripts
 372     initscripts = []
 373     data = {
 374         'timestamp': int(time.time()),
 375         'initscripts': initscripts,
 376         'slivers': slivers
 377     }
 378
 379     # create the ticket
 380     object_gid = record.get_gid_object()
 381     new_ticket = SfaTicket(subject = object_gid.get_subject())
 382     new_ticket.set_gid_caller(api.auth.client_gid)
 383     new_ticket.set_gid_object(object_gid)
 384     new_ticket.set_issuer(key=api.key, subject=api.hrn)
 385     new_ticket.set_pubkey(object_gid.get_pubkey())
 386     new_ticket.set_attributes(data)
 387     new_ticket.set_rspec(rspec)
 388     #new_ticket.set_parent(api.auth.hierarchy.get_auth_ticket(auth_hrn))
 389     new_ticket.encode()
 390     new_ticket.sign()
 391
 392     return new_ticket.save_to_string(save_parents=True)
 393
 394
 395
 396 def main():
 397     """
 398     rspec = ListResources(api, "plc.princeton.sapan", None, 'pl_test_sapan')
 399     #rspec = ListResources(api, "plc.princeton.coblitz", None, 'pl_test_coblitz')
 400     #rspec = ListResources(api, "plc.pl.sirius", None, 'pl_test_sirius')
 401     print rspec
 402     """
 403     api = PlcSfaApi()
 404     f = open(sys.argv[1])
 405     xml = f.read()
 406     f.close()
 407 #Error (E1120, main): No value passed for parameter 'users' in function call
 408 #Error (E1120, main): No value passed for parameter 'call_id' in function call
 409     CreateSliver(api, "plc.princeton.sapan", xml, 'CreateSliver_sapan')
 410
 411 if __name__ == "__main__":
 412     main()