sfa/managers/aggregate_manager_slab.py

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