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