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