sfa/openstack/nova_driver.py

   1 import time
   2 import datetime
   3
   4 from sfa.util.faults import MissingSfaInfo, UnknownSfaType, \
   5     RecordNotFound, SfaNotImplemented, SfaInvalidArgument, UnsupportedOperation
   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
  11 from sfa.openstack.osxrn import OSXrn, hrn_to_os_slicename, hrn_to_os_tenant_name
  12 from sfa.util.cache import Cache
  13 from sfa.trust.credential import Credential
  14 # used to be used in get_ticket
  15 #from sfa.trust.sfaticket import SfaTicket
  16 from sfa.rspecs.version_manager import VersionManager
  17 from sfa.rspecs.rspec import RSpec
  18 from sfa.storage.model import RegRecord, SliverAllocation
  19
  20 # the driver interface, mostly provides default behaviours
  21 from sfa.managers.driver import Driver
  22 from sfa.openstack.shell import Shell
  23 from sfa.openstack.osaggregate import OSAggregate
  24 from sfa.planetlab.plslices import PlSlices
  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
  40
  41 class NovaDriver(Driver):
  42
  43     # the cache instance is a class member so it survives across incoming
  44     # requests
  45     cache = None
  46
  47     def __init__(self, api):
  48         Driver.__init__(self, api)
  49         config = api.config
  50         self.shell = Shell(config=config)
  51         self.cache = None
  52         if config.SFA_AGGREGATE_CACHING:
  53             if NovaDriver.cache is None:
  54                 NovaDriver.cache = Cache()
  55             self.cache = NovaDriver.cache
  56
  57     def sliver_to_slice_xrn(self, xrn):
  58         sliver_id_parts = Xrn(xrn).get_sliver_id_parts()
  59         slice = self.shell.auth_manager.tenants.find(id=sliver_id_parts[0])
  60         if not slice:
  61             raise Forbidden(
  62                 "Unable to locate slice record for sliver:  %s" % xrn)
  63         slice_xrn = OSXrn(name=slice.name, type='slice')
  64         return slice_xrn
  65
  66     def check_sliver_credentials(self, creds, urns):
  67         # build list of cred object hrns
  68         slice_cred_names = []
  69         for cred in creds:
  70             slice_cred_hrn = Credential(cred=cred).get_gid_object().get_hrn()
  71             slice_cred_names.append(OSXrn(xrn=slice_cred_hrn).get_slicename())
  72
  73         # look up slice name of slivers listed in urns arg
  74         slice_ids = []
  75         for urn in urns:
  76             sliver_id_parts = Xrn(xrn=urn).get_sliver_id_parts()
  77             slice_ids.append(sliver_id_parts[0])
  78
  79         if not slice_ids:
  80             raise Forbidden("sliver urn not provided")
  81
  82         sliver_names = []
  83         for slice_id in slice_ids:
  84             slice = self.shell.auth_manager.tenants.find(slice_id)
  85             sliver_names.append(slice['name'])
  86
  87         # make sure we have a credential for every specified sliver ierd
  88         for sliver_name in sliver_names:
  89             if sliver_name not in slice_cred_names:
  90                 msg = "Valid credential not found for target: %s" % sliver_name
  91                 raise Forbidden(msg)
  92
  93     ########################################
  94     # registry oriented
  95     ########################################
  96
  97     # disabled users
  98     def is_enabled(self, record):
  99         # all records are enabled
 100         return True
 101
 102     def augment_records_with_testbed_info(self, sfa_records):
 103         return self.fill_record_info(sfa_records)
 104
 105     ##########
 106     def register(self, sfa_record, hrn, pub_key):
 107
 108         if sfa_record['type'] == 'slice':
 109             record = self.register_slice(sfa_record, hrn)
 110         elif sfa_record['type'] == 'user':
 111             record = self.register_user(sfa_record, hrn, pub_key)
 112         elif sfa_record['type'].startswith('authority'):
 113             record = self.register_authority(sfa_record, hrn)
 114         # We should be returning the records id as a pointer but
 115         # this is a string and the records table expects this to be an
 116         # int.
 117         # return record.id
 118         return -1
 119
 120     def register_slice(self, sfa_record, hrn):
 121         # add slice description, name, researchers, PI
 122         name = hrn_to_os_tenant_name(hrn)
 123         description = sfa_record.get('description', None)
 124         self.shell.auth_manager.tenants.create(name, description)
 125         tenant = self.shell.auth_manager.tenants.find(name=name)
 126         auth_hrn = OSXrn(xrn=hrn, type='slice').get_authority_hrn()
 127         parent_tenant_name = OSXrn(
 128             xrn=auth_hrn, type='slice').get_tenant_name()
 129         parent_tenant = self.shell.auth_manager.tenants.find(
 130             name=parent_tenant_name)
 131         researchers = sfa_record.get('researchers', [])
 132         for researcher in researchers:
 133             name = Xrn(researcher).get_leaf()
 134             user = self.shell.auth_manager.users.find(name=name)
 135             self.shell.auth_manager.roles.add_user_role(user, 'Member', tenant)
 136             self.shell.auth_manager.roles.add_user_role(user, 'user', tenant)
 137
 138         pis = sfa_record.get('pis', [])
 139         for pi in pis:
 140             name = Xrn(pi).get_leaf()
 141             user = self.shell.auth_manager.users.find(name=name)
 142             self.shell.auth_manager.roles.add_user_role(user, 'pi', tenant)
 143             self.shell.auth_manager.roles.add_user_role(
 144                 user, 'pi', parent_tenant)
 145
 146         return tenant
 147
 148     def register_user(self, sfa_record, hrn, pub_key):
 149         # add person roles, projects and keys
 150         email = sfa_record.get('email', None)
 151         xrn = Xrn(hrn)
 152         name = xrn.get_leaf()
 153         auth_hrn = xrn.get_authority_hrn()
 154         tenant_name = OSXrn(xrn=auth_hrn, type='authority').get_tenant_name()
 155         tenant = self.shell.auth_manager.tenants.find(name=tenant_name)
 156         self.shell.auth_manager.users.create(
 157             name, email=email, tenant_id=tenant.id)
 158         user = self.shell.auth_manager.users.find(name=name)
 159         slices = sfa_records.get('slices', [])
 160         for slice in projects:
 161             slice_tenant_name = OSXrn(
 162                 xrn=slice, type='slice').get_tenant_name()
 163             slice_tenant = self.shell.auth_manager.tenants.find(
 164                 name=slice_tenant_name)
 165             self.shell.auth_manager.roles.add_user_role(
 166                 user, slice_tenant, 'user')
 167         keys = sfa_records.get('keys', [])
 168         for key in keys:
 169             keyname = OSXrn(xrn=hrn, type='user').get_slicename()
 170             self.shell.nova_client.keypairs.create(keyname, key)
 171         return user
 172
 173     def register_authority(self, sfa_record, hrn):
 174         name = OSXrn(xrn=hrn, type='authority').get_tenant_name()
 175         self.shell.auth_manager.tenants.create(
 176             name, sfa_record.get('description', ''))
 177         tenant = self.shell.auth_manager.tenants.find(name=name)
 178         return tenant
 179
 180     ##########
 181     # xxx actually old_sfa_record comes filled with plc stuff as well in the
 182     # original code
 183     def update(self, old_sfa_record, new_sfa_record, hrn, new_key):
 184         type = new_sfa_record['type']
 185
 186         # new_key implemented for users only
 187         if new_key and type not in ['user']:
 188             raise UnknownSfaType(type)
 189
 190         elif type == "slice":
 191             # can update project manager and description
 192             name = hrn_to_os_slicename(hrn)
 193             researchers = sfa_record.get('researchers', [])
 194             pis = sfa_record.get('pis', [])
 195             project_manager = None
 196             description = sfa_record.get('description', None)
 197             if pis:
 198                 project_manager = Xrn(pis[0], 'user').get_leaf()
 199             elif researchers:
 200                 project_manager = Xrn(researchers[0], 'user').get_leaf()
 201             self.shell.auth_manager.modify_project(
 202                 name, project_manager, description)
 203
 204         elif type == "user":
 205             # can techinally update access_key and secret_key,
 206             # but that is not in our scope, so we do nothing.
 207             pass
 208         return True
 209
 210     ##########
 211     def remove(self, sfa_record):
 212         type = sfa_record['type']
 213         if type == 'user':
 214             name = Xrn(sfa_record['hrn']).get_leaf()
 215             if self.shell.auth_manager.get_user(name):
 216                 self.shell.auth_manager.delete_user(name)
 217         elif type == 'slice':
 218             name = hrn_to_os_slicename(sfa_record['hrn'])
 219             if self.shell.auth_manager.get_project(name):
 220                 self.shell.auth_manager.delete_project(name)
 221         return True
 222
 223     ####################
 224     def fill_record_info(self, records):
 225         """
 226         Given a (list of) SFA record, fill in the PLC specific
 227         and SFA specific fields in the record.
 228         """
 229         if not isinstance(records, list):
 230             records = [records]
 231
 232         for record in records:
 233             if record['type'] == 'user':
 234                 record = self.fill_user_record_info(record)
 235             elif record['type'] == 'slice':
 236                 record = self.fill_slice_record_info(record)
 237             elif record['type'].startswith('authority'):
 238                 record = self.fill_auth_record_info(record)
 239             else:
 240                 continue
 241             record['geni_urn'] = hrn_to_urn(record['hrn'], record['type'])
 242             record['geni_certificate'] = record['gid']
 243             # if os_record.created_at is not None:
 244             #    record['date_created'] = datetime_to_string(utcparse(os_record.created_at))
 245             # if os_record.updated_at is not None:
 246             #    record['last_updated'] = datetime_to_string(utcparse(os_record.updated_at))
 247
 248         return records
 249
 250     def fill_user_record_info(self, record):
 251         xrn = Xrn(record['hrn'])
 252         name = xrn.get_leaf()
 253         record['name'] = name
 254         user = self.shell.auth_manager.users.find(name=name)
 255         record['email'] = user.email
 256         tenant = self.shell.auth_manager.tenants.find(id=user.tenantId)
 257         slices = []
 258         all_tenants = self.shell.auth_manager.tenants.list()
 259         for tmp_tenant in all_tenants:
 260             if tmp_tenant.name.startswith(tenant.name + "."):
 261                 for tmp_user in tmp_tenant.list_users():
 262                     if tmp_user.name == user.name:
 263                         slice_hrn = ".".join([self.hrn, tmp_tenant.name])
 264                         slices.append(slice_hrn)
 265         record['slices'] = slices
 266         roles = self.shell.auth_manager.roles.roles_for_user(user, tenant)
 267         record['roles'] = [role.name for role in roles]
 268         keys = self.shell.nova_manager.keypairs.findall(name=record['hrn'])
 269         record['keys'] = [key.public_key for key in keys]
 270         return record
 271
 272     def fill_slice_record_info(self, record):
 273         tenant_name = hrn_to_os_tenant_name(record['hrn'])
 274         tenant = self.shell.auth_manager.tenants.find(name=tenant_name)
 275         parent_tenant_name = OSXrn(xrn=tenant_name).get_authority_hrn()
 276         parent_tenant = self.shell.auth_manager.tenants.find(
 277             name=parent_tenant_name)
 278         researchers = []
 279         pis = []
 280
 281         # look for users and pis in slice tenant
 282         for user in tenant.list_users():
 283             for role in self.shell.auth_manager.roles.roles_for_user(user, tenant):
 284                 if role.name.lower() == 'pi':
 285                     user_tenant = self.shell.auth_manager.tenants.find(
 286                         id=user.tenantId)
 287                     hrn = ".".join([self.hrn, user_tenant.name, user.name])
 288                     pis.append(hrn)
 289                 elif role.name.lower() in ['user', 'member']:
 290                     user_tenant = self.shell.auth_manager.tenants.find(
 291                         id=user.tenantId)
 292                     hrn = ".".join([self.hrn, user_tenant.name, user.name])
 293                     researchers.append(hrn)
 294
 295         # look for pis in the slice's parent (site/organization) tenant
 296         for user in parent_tenant.list_users():
 297             for role in self.shell.auth_manager.roles.roles_for_user(user, parent_tenant):
 298                 if role.name.lower() == 'pi':
 299                     user_tenant = self.shell.auth_manager.tenants.find(
 300                         id=user.tenantId)
 301                     hrn = ".".join([self.hrn, user_tenant.name, user.name])
 302                     pis.append(hrn)
 303         record['name'] = tenant_name
 304         record['description'] = tenant.description
 305         record['PI'] = pis
 306         if pis:
 307             record['geni_creator'] = pis[0]
 308         else:
 309             record['geni_creator'] = None
 310         record['researcher'] = researchers
 311         return record
 312
 313     def fill_auth_record_info(self, record):
 314         tenant_name = hrn_to_os_tenant_name(record['hrn'])
 315         tenant = self.shell.auth_manager.tenants.find(name=tenant_name)
 316         researchers = []
 317         pis = []
 318
 319         # look for users and pis in slice tenant
 320         for user in tenant.list_users():
 321             for role in self.shell.auth_manager.roles.roles_for_user(user, tenant):
 322                 hrn = ".".join([self.hrn, tenant.name, user.name])
 323                 if role.name.lower() == 'pi':
 324                     pis.append(hrn)
 325                 elif role.name.lower() in ['user', 'member']:
 326                     researchers.append(hrn)
 327
 328         # look for slices
 329         slices = []
 330         all_tenants = self.shell.auth_manager.tenants.list()
 331         for tmp_tenant in all_tenants:
 332             if tmp_tenant.name.startswith(tenant.name + "."):
 333                 slices.append(".".join([self.hrn, tmp_tenant.name]))
 334
 335         record['name'] = tenant_name
 336         record['description'] = tenant.description
 337         record['PI'] = pis
 338         record['enabled'] = tenant.enabled
 339         record['researchers'] = researchers
 340         record['slices'] = slices
 341         return record
 342
 343     ####################
 344     # plcapi works by changes, compute what needs to be added/deleted
 345     def update_relation(self, subject_type, target_type, subject_id, target_ids):
 346         # hard-wire the code for slice/user for now, could be smarter if needed
 347         if subject_type == 'slice' and target_type == 'user':
 348             subject = self.shell.project_get(subject_id)[0]
 349             current_target_ids = [user.name for user in subject.members]
 350             add_target_ids = list(
 351                 set(target_ids).difference(current_target_ids))
 352             del_target_ids = list(
 353                 set(current_target_ids).difference(target_ids))
 354             logger.debug("subject_id = %s (type=%s)" %
 355                          (subject_id, type(subject_id)))
 356             for target_id in add_target_ids:
 357                 self.shell.project_add_member(target_id, subject_id)
 358                 logger.debug("add_target_id = %s (type=%s)" %
 359                              (target_id, type(target_id)))
 360             for target_id in del_target_ids:
 361                 logger.debug("del_target_id = %s (type=%s)" %
 362                              (target_id, type(target_id)))
 363                 self.shell.project_remove_member(target_id, subject_id)
 364         else:
 365             logger.info('unexpected relation to maintain, %s -> %s' %
 366                         (subject_type, target_type))
 367
 368     ########################################
 369     # aggregate oriented
 370     ########################################
 371
 372     def testbed_name(self): return "openstack"
 373
 374     def aggregate_version(self):
 375         return {}
 376
 377     # first 2 args are None in case of resource discovery
 378     def list_resources(self, version=None, options=None):
 379         if options is None:
 380             options = {}
 381         aggregate = OSAggregate(self)
 382         rspec = aggregate.list_resources(version=version, options=options)
 383         return rspec
 384
 385     def describe(self, urns, version=None, options=None):
 386         if options is None:
 387             options = {}
 388         aggregate = OSAggregate(self)
 389         return aggregate.describe(urns, version=version, options=options)
 390
 391     def status(self, urns, options=None):
 392         if options is None:
 393             options = {}
 394         aggregate = OSAggregate(self)
 395         desc = aggregate.describe(urns)
 396         status = {'geni_urn': desc['geni_urn'],
 397                   'geni_slivers': desc['geni_slivers']}
 398         return status
 399
 400     def allocate(self, urn, rspec_string, expiration, options=None):
 401         if options is None:
 402             options = {}
 403         xrn = Xrn(urn)
 404         aggregate = OSAggregate(self)
 405
 406         # assume first user is the caller and use their context
 407         # for the ec2/euca api connection. Also, use the first users
 408         # key as the project key.
 409         key_name = None
 410         if len(users) > 1:
 411             key_name = aggregate.create_instance_key(xrn.get_hrn(), users[0])
 412
 413         # collect public keys
 414         users = options.get('geni_users', [])
 415         pubkeys = []
 416         for user in users:
 417             pubkeys.extend(user['keys'])
 418
 419         rspec = RSpec(rspec_string)
 420         instance_name = hrn_to_os_slicename(slice_hrn)
 421         tenant_name = OSXrn(xrn=slice_hrn, type='slice').get_tenant_name()
 422         slivers = aggregate.run_instances(instance_name, tenant_name,
 423                                           rspec_string, key_name, pubkeys)
 424
 425         # update all sliver allocation states setting then to geni_allocated
 426         sliver_ids = [sliver.id for sliver in slivers]
 427         dbsession = self.api.dbsession()
 428         SliverAllocation.set_allocations(
 429             sliver_ids, 'geni_provisioned', dbsession)
 430
 431         return aggregate.describe(urns=[urn], version=rspec.version)
 432
 433     def provision(self, urns, options=None):
 434         if options is None:
 435             options = {}
 436         # update sliver allocation states and set them to geni_provisioned
 437         aggregate = OSAggregate(self)
 438         instances = aggregate.get_instances(urns)
 439         sliver_ids = []
 440         for instance in instances:
 441             sliver_hrn = "%s.%s" % (self.driver.hrn, instance.id)
 442             sliver_ids.append(Xrn(sliver_hrn, type='sliver').urn)
 443         dbsession = self.api.dbsession()
 444         SliverAllocation.set_allocations(
 445             sliver_ids, 'geni_provisioned', dbsession)
 446         version_manager = VersionManager()
 447         rspec_version = version_manager.get_version(
 448             options['geni_rspec_version'])
 449         return self.describe(urns, rspec_version, options=options)
 450
 451     def delete(self, urns, options=None):
 452         if options is None:
 453             options = {}
 454         # collect sliver ids so we can update sliver allocation states after
 455         # we remove the slivers.
 456         aggregate = OSAggregate(self)
 457         instances = aggregate.get_instances(urns)
 458         sliver_ids = []
 459         for instance in instances:
 460             sliver_hrn = "%s.%s" % (self.driver.hrn, instance.id)
 461             sliver_ids.append(Xrn(sliver_hrn, type='sliver').urn)
 462
 463             # delete the instance
 464             aggregate.delete_instance(instance)
 465
 466         # delete sliver allocation states
 467         dbsession = self.api.dbsession()
 468         SliverAllocation.delete_allocations(sliver_ids, dbsession)
 469
 470         # return geni_slivers
 471         geni_slivers = []
 472         for sliver_id in sliver_ids:
 473             geni_slivers.append(
 474                 {'geni_sliver_urn': sliver['sliver_id'],
 475                  'geni_allocation_status': 'geni_unallocated',
 476                  'geni_expires': None})
 477         return geni_slivers
 478
 479     def renew(self, urns, expiration_time, options=None):
 480         if options is None:
 481             options = {}
 482         description = self.describe(urns, None, options)
 483         return description['geni_slivers']
 484
 485     def perform_operational_action(self, urns, action, options=None):
 486         if options is None:
 487             options = {}
 488         aggregate = OSAggregate(self)
 489         action = action.lower()
 490         if action == 'geni_start':
 491             action_method = aggregate.start_instances
 492         elif action == 'geni_stop':
 493             action_method = aggregate.stop_instances
 494         elif action == 'geni_restart':
 495             action_method = aggreate.restart_instances
 496         else:
 497             raise UnsupportedOperation(action)
 498
 499          # fault if sliver is not full allocated (operational status is
 500          # geni_pending_allocation)
 501         description = self.describe(urns, None, options)
 502         for sliver in description['geni_slivers']:
 503             if sliver['geni_operational_status'] == 'geni_pending_allocation':
 504                 raise UnsupportedOperation(
 505                     action, "Sliver must be fully allocated (operational status is not geni_pending_allocation)")
 506         #
 507         # Perform Operational Action Here
 508         #
 509
 510         instances = aggregate.get_instances(urns)
 511         for instance in instances:
 512             tenant_name = self.driver.shell.auth_manager.client.tenant_name
 513             action_method(tenant_name, instance.name, instance.id)
 514         description = self.describe(urns)
 515         geni_slivers = self.describe(urns, None, options)['geni_slivers']
 516         return geni_slivers
 517
 518     def shutdown(self, xrn, options=None):
 519         if options is None:
 520             options = {}
 521         xrn = OSXrn(xrn=xrn, type='slice')
 522         tenant_name = xrn.get_tenant_name()
 523         name = xrn.get_slicename()
 524         self.driver.shell.nova_manager.connect(tenant=tenant_name)
 525         instances = self.driver.shell.nova_manager.servers.findall(name=name)
 526         for instance in instances:
 527             self.driver.shell.nova_manager.servers.shutdown(instance)
 528         return True