sfa/openstack/openstack_driver.py

   1 import time
   2 import datetime
   3 #
   4 from sfa.util.faults import MissingSfaInfo, UnknownSfaType, \
   5     RecordNotFound, SfaNotImplemented, SliverDoesNotExist
   6
   7 from sfa.util.sfalogging import logger
   8 from sfa.util.defaultdict import defaultdict
   9 from sfa.util.sfatime import utcparse, datetime_to_string, datetime_to_epoch
  10 from sfa.util.xrn import Xrn, hrn_to_urn, get_leaf, urn_to_sliver_id
  11 from sfa.util.cache import Cache
  12 # used to be used in get_ticket
  13 #from sfa.trust.sfaticket import SfaTicket
  14
  15 from sfa.rspecs.version_manager import VersionManager
  16 from sfa.rspecs.rspec import RSpec
  17
  18 # the driver interface, mostly provides default behaviours
  19 from sfa.managers.driver import Driver
  20
  21 from sfa.openstack.openstack_shell import OpenstackShell
  22 from sfa.openstack.osaggregate import OSAggregate
  23 from sfa.plc.plslices import PlSlices
  24 from sfa.util.osxrn import OSXrn
  25
  26
  27 def list_to_dict(recs, key):
  28     """
  29     convert a list of dictionaries into a dictionary keyed on the
  30     specified dictionary key
  31     """
  32     return dict ( [ (rec[key],rec) for rec in recs ] )
  33
  34 #
  35 # PlShell is just an xmlrpc serverproxy where methods
  36 # can be sent as-is; it takes care of authentication
  37 # from the global config
  38 #
  39 class OpenstackDriver (Driver):
  40
  41     # the cache instance is a class member so it survives across incoming requests
  42     cache = None
  43
  44     def __init__ (self, config):
  45         Driver.__init__ (self, config)
  46         self.shell = OpenstackShell (config)
  47         self.cache=None
  48         if config.SFA_AGGREGATE_CACHING:
  49             if OpenstackDriver.cache is None:
  50                 OpenstackDriver.cache = Cache()
  51             self.cache = OpenstackDriver.cache
  52
  53     ########################################
  54     ########## registry oriented
  55     ########################################
  56
  57     ########## disabled users
  58     def is_enabled (self, record):
  59         # all records are enabled
  60         return True
  61
  62     def augment_records_with_testbed_info (self, sfa_records):
  63         return self.fill_record_info (sfa_records)
  64
  65     ##########
  66     def register (self, sfa_record, hrn, pub_key):
  67         type = sfa_record['type']
  68         pl_record = self.sfa_fields_to_pl_fields(type, hrn, sfa_record)
  69
  70         if type == 'slice':
  71             acceptable_fields=['url', 'instantiation', 'name', 'description']
  72             # add slice description, name, researchers, PI
  73             pass
  74
  75         elif type == 'user':
  76             # add person roles, projects and keys
  77             pass
  78         return pointer
  79
  80     ##########
  81     # xxx actually old_sfa_record comes filled with plc stuff as well in the original code
  82     def update (self, old_sfa_record, new_sfa_record, hrn, new_key):
  83         pointer = old_sfa_record['pointer']
  84         type = old_sfa_record['type']
  85
  86         # new_key implemented for users only
  87         if new_key and type not in [ 'user' ]:
  88             raise UnknownSfaType(type)
  89
  90         elif type == "slice":
  91             # can update description, researchers and PI
  92             pass
  93         elif type == "user":
  94             # can update  slices, keys and roles
  95             pass
  96         return True
  97
  98
  99     ##########
 100     def remove (self, sfa_record):
 101         type=sfa_record['type']
 102         name = Xrn(sfa_record['hrn']).get_leaf()
 103         if type == 'user':
 104             if self.shell.user_get(name):
 105                 self.shell.user_delete(name)
 106         elif type == 'slice':
 107             if self.shell.project_get(name):
 108                 self.shell.project_delete(name)
 109         return True
 110
 111
 112     ####################
 113     def fill_record_info(self, records):
 114         """
 115         Given a (list of) SFA record, fill in the PLC specific
 116         and SFA specific fields in the record.
 117         """
 118         if not isinstance(records, list):
 119             records = [records]
 120
 121         for record in records:
 122             name = Xrn(record['hrn']).get_leaf()
 123             os_record = None
 124             if record['type'] == 'user':
 125                 os_record = self.shell.user_get(name)
 126                 record['slices'] = [self.hrn + "." + proj.name for \
 127                                     proj in os_record.projects]
 128                 record['roles'] = [role for role in os_record.roles]
 129                 keys = self.shell.key_pair_get_all_by_user(name)
 130                 record['keys'] = [key.public_key for key in keys]
 131             elif record['type'] == 'slice':
 132                 os_record = self.shell.project_get(name)
 133                 record['description'] = os_record.description
 134                 record['PI'] = self.hrn + "." + os_record.project_manager
 135                 record['geni_creator'] = record['PI']
 136                 record['researcher'] = [self.hrn + "." + user.name for \
 137                                          user in os_record.members]
 138             else:
 139                 continue
 140             record['geni_urn'] = hrn_to_urn(record['hrn'], record['type'])
 141             record['geni_certificate'] = record['gid']
 142             record['name'] = os_record.name
 143             if os_record.created_at is not None:
 144                 record['date_created'] = datetime_to_string(utcparse(os_record.created_at))
 145             if os_record.updated_at is not None:
 146                 record['last_updated'] = datetime_to_string(utcparse(os_record.updated_at))
 147
 148         return records
 149
 150
 151     ####################
 152     # plcapi works by changes, compute what needs to be added/deleted
 153     def update_relation (self, subject_type, target_type, subject_id, target_ids):
 154         # hard-wire the code for slice/user for now, could be smarter if needed
 155         if subject_type =='slice' and target_type == 'user':
 156             subject=self.shell.project_get(subject_id)[0]
 157             current_target_ids = [user.name for user in subject.members]
 158             add_target_ids = list ( set (target_ids).difference(current_target_ids))
 159             del_target_ids = list ( set (current_target_ids).difference(target_ids))
 160             logger.debug ("subject_id = %s (type=%s)"%(subject_id,type(subject_id)))
 161             for target_id in add_target_ids:
 162                 self.shell.project_add_member(target_id,subject_id)
 163                 logger.debug ("add_target_id = %s (type=%s)"%(target_id,type(target_id)))
 164             for target_id in del_target_ids:
 165                 logger.debug ("del_target_id = %s (type=%s)"%(target_id,type(target_id)))
 166                 self.shell.project_remove_member(target_id, subject_id)
 167         else:
 168             logger.info('unexpected relation to maintain, %s -> %s'%(subject_type,target_type))
 169
 170
 171     ########################################
 172     ########## aggregate oriented
 173     ########################################
 174
 175     def testbed_name (self): return "openstack"
 176
 177     # 'geni_request_rspec_versions' and 'geni_ad_rspec_versions' are mandatory
 178     def aggregate_version (self):
 179         version_manager = VersionManager()
 180         ad_rspec_versions = []
 181         request_rspec_versions = []
 182         for rspec_version in version_manager.versions:
 183             if rspec_version.content_type in ['*', 'ad']:
 184                 ad_rspec_versions.append(rspec_version.to_dict())
 185             if rspec_version.content_type in ['*', 'request']:
 186                 request_rspec_versions.append(rspec_version.to_dict())
 187         return {
 188             'testbed':self.testbed_name(),
 189             'geni_request_rspec_versions': request_rspec_versions,
 190             'geni_ad_rspec_versions': ad_rspec_versions,
 191             }
 192
 193     def list_slices (self, creds, options):
 194         # look in cache first
 195         if self.cache:
 196             slices = self.cache.get('slices')
 197             if slices:
 198                 logger.debug("OpenStackDriver.list_slices returns from cache")
 199                 return slices
 200
 201         # get data from db
 202         slices = self.shell.project_get_all()
 203         slice_urns = [OSXrn(name, 'slice').urn for name in slice]
 204
 205         # cache the result
 206         if self.cache:
 207             logger.debug ("OpenStackDriver.list_slices stores value in cache")
 208             self.cache.add('slices', slice_urns)
 209
 210         return slice_urns
 211
 212     # first 2 args are None in case of resource discovery
 213     def list_resources (self, slice_urn, slice_hrn, creds, options):
 214         cached_requested = options.get('cached', True)
 215
 216         version_manager = VersionManager()
 217         # get the rspec's return format from options
 218         rspec_version = version_manager.get_version(options.get('geni_rspec_version'))
 219         version_string = "rspec_%s" % (rspec_version)
 220
 221         #panos adding the info option to the caching key (can be improved)
 222         if options.get('info'):
 223             version_string = version_string + "_"+options.get('info', 'default')
 224
 225         # look in cache first
 226         if cached_requested and self.cache and not slice_hrn:
 227             rspec = self.cache.get(version_string)
 228             if rspec:
 229                 logger.debug("OpenStackDriver.ListResources: returning cached advertisement")
 230                 return rspec
 231
 232         #panos: passing user-defined options
 233         #print "manager options = ",options
 234         aggregate = OSAggregate(self)
 235         rspec =  aggregate.get_rspec(slice_xrn=slice_urn, version=rspec_version,
 236                                      options=options)
 237
 238         # cache the result
 239         if self.cache and not slice_hrn:
 240             logger.debug("OpenStackDriver.ListResources: stores advertisement in cache")
 241             self.cache.add(version_string, rspec)
 242
 243         return rspec
 244
 245     def sliver_status (self, slice_urn, slice_hrn):
 246         # find out where this slice is currently running
 247         slicename = hrn_to_pl_slicename(slice_hrn)
 248
 249         slices = self.shell.GetSlices([slicename], ['slice_id', 'node_ids','person_ids','name','expires'])
 250         if len(slices) == 0:
 251             raise SliverDoesNotExist("%s (used %s as slicename internally)" % (slice_hrn, slicename))
 252         slice = slices[0]
 253
 254         # report about the local nodes only
 255         nodes = self.shell.GetNodes({'node_id':slice['node_ids'],'peer_id':None},
 256                               ['node_id', 'hostname', 'site_id', 'boot_state', 'last_contact'])
 257
 258         if len(nodes) == 0:
 259             raise SliverDoesNotExist("You have not allocated any slivers here")
 260
 261         site_ids = [node['site_id'] for node in nodes]
 262
 263         result = {}
 264         top_level_status = 'unknown'
 265         if nodes:
 266             top_level_status = 'ready'
 267         result['geni_urn'] = slice_urn
 268         result['pl_login'] = slice['name']
 269         result['pl_expires'] = datetime_to_string(utcparse(slice['expires']))
 270
 271         resources = []
 272         for node in nodes:
 273             res = {}
 274             res['pl_hostname'] = node['hostname']
 275             res['pl_boot_state'] = node['boot_state']
 276             res['pl_last_contact'] = node['last_contact']
 277             if node['last_contact'] is not None:
 278
 279                 res['pl_last_contact'] = datetime_to_string(utcparse(node['last_contact']))
 280             sliver_id = urn_to_sliver_id(slice_urn, slice['slice_id'], node['node_id'])
 281             res['geni_urn'] = sliver_id
 282             if node['boot_state'] == 'boot':
 283                 res['geni_status'] = 'ready'
 284             else:
 285                 res['geni_status'] = 'failed'
 286                 top_level_status = 'failed'
 287
 288             res['geni_error'] = ''
 289
 290             resources.append(res)
 291
 292         result['geni_status'] = top_level_status
 293         result['geni_resources'] = resources
 294         return result
 295
 296     def create_sliver (self, slice_urn, slice_hrn, creds, rspec_string, users, options):
 297
 298         aggregate = PlAggregate(self)
 299         slices = PlSlices(self)
 300         peer = slices.get_peer(slice_hrn)
 301         sfa_peer = slices.get_sfa_peer(slice_hrn)
 302         slice_record=None
 303         if users:
 304             slice_record = users[0].get('slice_record', {})
 305
 306         # parse rspec
 307         rspec = RSpec(rspec_string)
 308         requested_attributes = rspec.version.get_slice_attributes()
 309
 310         # ensure site record exists
 311         site = slices.verify_site(slice_hrn, slice_record, peer, sfa_peer, options=options)
 312         # ensure slice record exists
 313         slice = slices.verify_slice(slice_hrn, slice_record, peer, sfa_peer, options=options)
 314         # ensure person records exists
 315         persons = slices.verify_persons(slice_hrn, slice, users, peer, sfa_peer, options=options)
 316         # ensure slice attributes exists
 317         slices.verify_slice_attributes(slice, requested_attributes, options=options)
 318
 319         # add/remove slice from nodes
 320         requested_slivers = [node.get('component_name') for node in rspec.version.get_nodes_with_slivers()]
 321         nodes = slices.verify_slice_nodes(slice, requested_slivers, peer)
 322
 323         # add/remove links links
 324         slices.verify_slice_links(slice, rspec.version.get_link_requests(), nodes)
 325
 326         # handle MyPLC peer association.
 327         # only used by plc and ple.
 328         slices.handle_peer(site, slice, persons, peer)
 329
 330         return aggregate.get_rspec(slice_xrn=slice_urn, version=rspec.version)
 331
 332     def delete_sliver (self, slice_urn, slice_hrn, creds, options):
 333         name = OSXrn(xrn=slice_urn).name
 334         slice = self.shell.project_get(name)
 335         if not slice:
 336             return 1
 337
 338         self.shell.DeleteSliceFromNodes(slicename, slice['node_ids'])
 339         instances = self.shell.instance_get_all_by_project(name)
 340         for instance in instances:
 341             self.shell.instance_destroy(instance.instance_id)
 342         return 1
 343
 344     def renew_sliver (self, slice_urn, slice_hrn, creds, expiration_time, options):
 345         return True
 346
 347     def start_slice (self, slice_urn, slice_hrn, creds):
 348         return 1
 349
 350     def stop_slice (self, slice_urn, slice_hrn, creds):
 351         name = OSXrn(xrn=slice_urn).name
 352         slice = self.shell.project_get(name)
 353         if not slice:
 354             return 1
 355
 356         self.shell.DeleteSliceFromNodes(slicename, slice['node_ids'])
 357         instances = self.shell.instance_get_all_by_project(name)
 358         for instance in instances:
 359             self.shell.instance_stop(instance.instance_id)
 360         return 1
 361
 362     def reset_slice (self, slice_urn, slice_hrn, creds):
 363         raise SfaNotImplemented ("reset_slice not available at this interface")
 364
 365     # xxx this code is quite old and has not run for ages
 366     # it is obviously totally broken and needs a rewrite
 367     def get_ticket (self, slice_urn, slice_hrn, creds, rspec_string, options):
 368         raise SfaNotImplemented,"OpenStackDriver.get_ticket needs a rewrite"
 369 # please keep this code for future reference
 370 #        slices = PlSlices(self)
 371 #        peer = slices.get_peer(slice_hrn)
 372 #        sfa_peer = slices.get_sfa_peer(slice_hrn)
 373 #
 374 #        # get the slice record
 375 #        credential = api.getCredential()
 376 #        interface = api.registries[api.hrn]
 377 #        registry = api.server_proxy(interface, credential)
 378 #        records = registry.Resolve(xrn, credential)
 379 #
 380 #        # make sure we get a local slice record
 381 #        record = None
 382 #        for tmp_record in records:
 383 #            if tmp_record['type'] == 'slice' and \
 384 #               not tmp_record['peer_authority']:
 385 #    #Error (E0602, GetTicket): Undefined variable 'SliceRecord'
 386 #                slice_record = SliceRecord(dict=tmp_record)
 387 #        if not record:
 388 #            raise RecordNotFound(slice_hrn)
 389 #
 390 #        # similar to CreateSliver, we must verify that the required records exist
 391 #        # at this aggregate before we can issue a ticket
 392 #        # parse rspec
 393 #        rspec = RSpec(rspec_string)
 394 #        requested_attributes = rspec.version.get_slice_attributes()
 395 #
 396 #        # ensure site record exists
 397 #        site = slices.verify_site(slice_hrn, slice_record, peer, sfa_peer)
 398 #        # ensure slice record exists
 399 #        slice = slices.verify_slice(slice_hrn, slice_record, peer, sfa_peer)
 400 #        # ensure person records exists
 401 #    # xxx users is undefined in this context
 402 #        persons = slices.verify_persons(slice_hrn, slice, users, peer, sfa_peer)
 403 #        # ensure slice attributes exists
 404 #        slices.verify_slice_attributes(slice, requested_attributes)
 405 #
 406 #        # get sliver info
 407 #        slivers = slices.get_slivers(slice_hrn)
 408 #
 409 #        if not slivers:
 410 #            raise SliverDoesNotExist(slice_hrn)
 411 #
 412 #        # get initscripts
 413 #        initscripts = []
 414 #        data = {
 415 #            'timestamp': int(time.time()),
 416 #            'initscripts': initscripts,
 417 #            'slivers': slivers
 418 #        }
 419 #
 420 #        # create the ticket
 421 #        object_gid = record.get_gid_object()
 422 #        new_ticket = SfaTicket(subject = object_gid.get_subject())
 423 #        new_ticket.set_gid_caller(api.auth.client_gid)
 424 #        new_ticket.set_gid_object(object_gid)
 425 #        new_ticket.set_issuer(key=api.key, subject=self.hrn)
 426 #        new_ticket.set_pubkey(object_gid.get_pubkey())
 427 #        new_ticket.set_attributes(data)
 428 #        new_ticket.set_rspec(rspec)
 429 #        #new_ticket.set_parent(api.auth.hierarchy.get_auth_ticket(auth_hrn))
 430 #        new_ticket.encode()
 431 #        new_ticket.sign()
 432 #
 433 #        return new_ticket.save_to_string(save_parents=True)