sfa/openstack/nova_driver.py

   1 import time
   2 import datetime
   3
   4 from sfa.util.faults import MissingSfaInfo, UnknownSfaType, \
   5     RecordNotFound, SfaNotImplemented, SliverDoesNotExist, \
   6     SfaInvalidArgument
   7
   8 from sfa.util.sfalogging import logger
   9 from sfa.util.defaultdict import defaultdict
  10 from sfa.util.sfatime import utcparse, datetime_to_string, datetime_to_epoch
  11 from sfa.util.xrn import Xrn, hrn_to_urn, get_leaf, urn_to_sliver_id
  12 from sfa.openstack.osxrn import OSXrn, hrn_to_os_slicename, hrn_to_os_tenant_name
  13 from sfa.util.cache import Cache
  14 from sfa.trust.credential import Credential
  15 # used to be used in get_ticket
  16 #from sfa.trust.sfaticket import SfaTicket
  17
  18 from sfa.rspecs.version_manager import VersionManager
  19 from sfa.rspecs.rspec import RSpec
  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)
  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     ########################################
  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
  68         if sfa_record['type'] == 'slice':
  69             record = self.register_slice(sfa_record, hrn)
  70         elif sfa_record['type'] == 'user':
  71             record = self.register_user(sfa_record, hrn, pub_key)
  72         elif sfa_record['type'].startswith('authority'):
  73             record = self.register_authority(sfa_record, hrn)
  74         # We should be returning the records id as a pointer but
  75         # this is a string and the records table expects this to be an
  76         # int.
  77         #return record.id
  78         return -1
  79
  80     def register_slice(self, sfa_record, hrn):
  81         # add slice description, name, researchers, PI
  82         name = hrn_to_os_tenant_name(hrn)
  83         description = sfa_record.get('description', None)
  84         self.shell.auth_manager.tenants.create(name, description)
  85         tenant = self.shell.auth_manager.tenants.find(name=name)
  86         auth_hrn = OSXrn(xrn=hrn, type='slice').get_authority_hrn()
  87         parent_tenant_name = OSXrn(xrn=auth_hrn, type='slice').get_tenant_name()
  88         parent_tenant = self.shell.auth_manager.tenants.find(name=parent_tenant_name)
  89         researchers = sfa_record.get('researchers', [])
  90         for researcher in researchers:
  91             name = Xrn(researcher).get_leaf()
  92             user = self.shell.auth_manager.users.find(name=name)
  93             self.shell.auth_manager.roles.add_user_role(user, 'user', tenant)
  94
  95         pis = sfa_record.get('pis', [])
  96         for pi in pis:
  97             name = Xrn(pi).get_leaf()
  98             user = self.shell.auth_manager.users.find(name=name)
  99             self.shell.auth_manager.roles.add_user_role(user, 'pi', tenant)
 100             self.shell.auth_manager.roles.add_user_role(user, 'pi', parent_tenant)
 101
 102         return tenant
 103
 104     def register_user(self, sfa_record, hrn, pub_key):
 105         # add person roles, projects and keys
 106         email = sfa_record.get('email', None)
 107         xrn = Xrn(hrn)
 108         name = xrn.get_leaf()
 109         auth_hrn = xrn.get_authority_hrn()
 110         tenant_name = OSXrn(xrn=auth_hrn, type='authority').get_tenant_name()
 111         tenant = self.shell.auth_manager.tenants.find(name=tenant_name)
 112         self.shell.auth_manager.users.create(name, email=email, tenant_id=tenant.id)
 113         user = self.shell.auth_manager.users.find(name=name)
 114         slices = sfa_records.get('slices', [])
 115         for slice in projects:
 116             slice_tenant_name = OSXrn(xrn=slice, type='slice').get_tenant_name()
 117             slice_tenant = self.shell.auth_manager.tenants.find(name=slice_tenant_name)
 118             self.shell.auth_manager.roles.add_user_role(user, slice_tenant, 'user')
 119         keys = sfa_records.get('keys', [])
 120         for key in keys:
 121             keyname = OSXrn(xrn=hrn, type='user').get_slicename()
 122             self.shell.nova_client.keypairs.create(keyname, key)
 123         return user
 124
 125     def register_authority(self, sfa_record, hrn):
 126         name = OSXrn(xrn=hrn, type='authority').get_tenant_name()
 127         self.shell.auth_manager.tenants.create(name, sfa_record.get('description', ''))
 128         tenant = self.shell.auth_manager.tenants.find(name=name)
 129         return tenant
 130
 131
 132     ##########
 133     # xxx actually old_sfa_record comes filled with plc stuff as well in the original code
 134     def update (self, old_sfa_record, new_sfa_record, hrn, new_key):
 135         type = new_sfa_record['type']
 136
 137         # new_key implemented for users only
 138         if new_key and type not in [ 'user' ]:
 139             raise UnknownSfaType(type)
 140
 141         elif type == "slice":
 142             # can update project manager and description
 143             name = hrn_to_os_slicename(hrn)
 144             researchers = sfa_record.get('researchers', [])
 145             pis = sfa_record.get('pis', [])
 146             project_manager = None
 147             description = sfa_record.get('description', None)
 148             if pis:
 149                 project_manager = Xrn(pis[0], 'user').get_leaf()
 150             elif researchers:
 151                 project_manager = Xrn(researchers[0], 'user').get_leaf()
 152             self.shell.auth_manager.modify_project(name, project_manager, description)
 153
 154         elif type == "user":
 155             # can techinally update access_key and secret_key,
 156             # but that is not in our scope, so we do nothing.
 157             pass
 158         return True
 159
 160
 161     ##########
 162     def remove (self, sfa_record):
 163         type=sfa_record['type']
 164         if type == 'user':
 165             name = Xrn(sfa_record['hrn']).get_leaf()
 166             if self.shell.auth_manager.get_user(name):
 167                 self.shell.auth_manager.delete_user(name)
 168         elif type == 'slice':
 169             name = hrn_to_os_slicename(sfa_record['hrn'])
 170             if self.shell.auth_manager.get_project(name):
 171                 self.shell.auth_manager.delete_project(name)
 172         return True
 173
 174
 175     ####################
 176     def fill_record_info(self, records):
 177         """
 178         Given a (list of) SFA record, fill in the PLC specific
 179         and SFA specific fields in the record.
 180         """
 181         if not isinstance(records, list):
 182             records = [records]
 183
 184         for record in records:
 185             if record['type'] == 'user':
 186                 record = self.fill_user_record_info(record)
 187             elif record['type'] == 'slice':
 188                 record = self.fill_slice_record_info(record)
 189             elif record['type'].startswith('authority'):
 190                 record = self.fill_auth_record_info(record)
 191             else:
 192                 continue
 193             record['geni_urn'] = hrn_to_urn(record['hrn'], record['type'])
 194             record['geni_certificate'] = record['gid']
 195             #if os_record.created_at is not None:
 196             #    record['date_created'] = datetime_to_string(utcparse(os_record.created_at))
 197             #if os_record.updated_at is not None:
 198             #    record['last_updated'] = datetime_to_string(utcparse(os_record.updated_at))
 199
 200         return records
 201
 202     def fill_user_record_info(self, record):
 203         xrn = Xrn(record['hrn'])
 204         name = xrn.get_leaf()
 205         record['name'] = name
 206         user = self.shell.auth_manager.users.find(name=name)
 207         record['email'] = user.email
 208         tenant = self.shell.auth_manager.tenants.find(id=user.tenantId)
 209         slices = []
 210         all_tenants = self.shell.auth_manager.tenants.list()
 211         for tmp_tenant in all_tenants:
 212             if tmp_tenant.name.startswith(tenant.name +"."):
 213                 for tmp_user in tmp_tenant.list_users():
 214                     if tmp_user.name == user.name:
 215                         slice_hrn = ".".join([self.hrn, tmp_tenant.name])
 216                         slices.append(slice_hrn)
 217         record['slices'] = slices
 218         roles = self.shell.auth_manager.roles.roles_for_user(user, tenant)
 219         record['roles'] = [role.name for role in roles]
 220         keys = self.shell.nova_manager.keypairs.findall(name=record['hrn'])
 221         record['keys'] = [key.public_key for key in keys]
 222         return record
 223
 224     def fill_slice_record_info(self, record):
 225         tenant_name = hrn_to_os_tenant_name(record['hrn'])
 226         tenant = self.shell.auth_manager.tenants.find(name=tenant_name)
 227         parent_tenant_name = OSXrn(xrn=tenant_name).get_authority_hrn()
 228         parent_tenant = self.shell.auth_manager.tenants.find(name=parent_tenant_name)
 229         researchers = []
 230         pis = []
 231
 232         # look for users and pis in slice tenant
 233         for user in tenant.list_users():
 234             for role in self.shell.auth_manager.roles.roles_for_user(user, tenant):
 235                 if role.name.lower() == 'pi':
 236                     user_tenant = self.shell.auth_manager.tenants.find(id=user.tenantId)
 237                     hrn = ".".join([self.hrn, user_tenant.name, user.name])
 238                     pis.append(hrn)
 239                 elif role.name.lower() in ['user', 'member']:
 240                     user_tenant = self.shell.auth_manager.tenants.find(id=user.tenantId)
 241                     hrn = ".".join([self.hrn, user_tenant.name, user.name])
 242                     researchers.append(hrn)
 243
 244         # look for pis in the slice's parent (site/organization) tenant
 245         for user in parent_tenant.list_users():
 246             for role in self.shell.auth_manager.roles.roles_for_user(user, parent_tenant):
 247                 if role.name.lower() == 'pi':
 248                     user_tenant = self.shell.auth_manager.tenants.find(id=user.tenantId)
 249                     hrn = ".".join([self.hrn, user_tenant.name, user.name])
 250                     pis.append(hrn)
 251         record['name'] = tenant_name
 252         record['description'] = tenant.description
 253         record['PI'] = pis
 254         if pis:
 255             record['geni_creator'] = pis[0]
 256         else:
 257             record['geni_creator'] = None
 258         record['researcher'] = researchers
 259         return record
 260
 261     def fill_auth_record_info(self, record):
 262         tenant_name = hrn_to_os_tenant_name(record['hrn'])
 263         tenant = self.shell.auth_manager.tenants.find(name=tenant_name)
 264         researchers = []
 265         pis = []
 266
 267         # look for users and pis in slice tenant
 268         for user in tenant.list_users():
 269             for role in self.shell.auth_manager.roles.roles_for_user(user, tenant):
 270                 hrn = ".".join([self.hrn, tenant.name, user.name])
 271                 if role.name.lower() == 'pi':
 272                     pis.append(hrn)
 273                 elif role.name.lower() in ['user', 'member']:
 274                     researchers.append(hrn)
 275
 276         # look for slices
 277         slices = []
 278         all_tenants = self.shell.auth_manager.tenants.list()
 279         for tmp_tenant in all_tenants:
 280             if tmp_tenant.name.startswith(tenant.name+"."):
 281                 slices.append(".".join([self.hrn, tmp_tenant.name]))
 282
 283         record['name'] = tenant_name
 284         record['description'] = tenant.description
 285         record['PI'] = pis
 286         record['enabled'] = tenant.enabled
 287         record['researchers'] = researchers
 288         record['slices'] = slices
 289         return record
 290
 291     ####################
 292     # plcapi works by changes, compute what needs to be added/deleted
 293     def update_relation (self, subject_type, target_type, subject_id, target_ids):
 294         # hard-wire the code for slice/user for now, could be smarter if needed
 295         if subject_type =='slice' and target_type == 'user':
 296             subject=self.shell.project_get(subject_id)[0]
 297             current_target_ids = [user.name for user in subject.members]
 298             add_target_ids = list ( set (target_ids).difference(current_target_ids))
 299             del_target_ids = list ( set (current_target_ids).difference(target_ids))
 300             logger.debug ("subject_id = %s (type=%s)"%(subject_id,type(subject_id)))
 301             for target_id in add_target_ids:
 302                 self.shell.project_add_member(target_id,subject_id)
 303                 logger.debug ("add_target_id = %s (type=%s)"%(target_id,type(target_id)))
 304             for target_id in del_target_ids:
 305                 logger.debug ("del_target_id = %s (type=%s)"%(target_id,type(target_id)))
 306                 self.shell.project_remove_member(target_id, subject_id)
 307         else:
 308             logger.info('unexpected relation to maintain, %s -> %s'%(subject_type,target_type))
 309
 310
 311     ########################################
 312     ########## aggregate oriented
 313     ########################################
 314
 315     def testbed_name (self): return "openstack"
 316
 317     # 'geni_request_rspec_versions' and 'geni_ad_rspec_versions' are mandatory
 318     def aggregate_version (self):
 319         version_manager = VersionManager()
 320         ad_rspec_versions = []
 321         request_rspec_versions = []
 322         for rspec_version in version_manager.versions:
 323             if rspec_version.content_type in ['*', 'ad']:
 324                 ad_rspec_versions.append(rspec_version.to_dict())
 325             if rspec_version.content_type in ['*', 'request']:
 326                 request_rspec_versions.append(rspec_version.to_dict())
 327         return {
 328             'testbed':self.testbed_name(),
 329             'geni_request_rspec_versions': request_rspec_versions,
 330             'geni_ad_rspec_versions': ad_rspec_versions,
 331             }
 332
 333     def list_slices (self, creds, options):
 334         # look in cache first
 335         if self.cache:
 336             slices = self.cache.get('slices')
 337             if slices:
 338                 logger.debug("OpenStackDriver.list_slices returns from cache")
 339                 return slices
 340
 341         # get data from db
 342         projs = self.shell.auth_manager.get_projects()
 343         slice_urns = [OSXrn(proj.name, 'slice').urn for proj in projs]
 344
 345         # cache the result
 346         if self.cache:
 347             logger.debug ("OpenStackDriver.list_slices stores value in cache")
 348             self.cache.add('slices', slice_urns)
 349
 350         return slice_urns
 351
 352     # first 2 args are None in case of resource discovery
 353     def list_resources (self, slice_urn, slice_hrn, creds, options):
 354         cached_requested = options.get('cached', True)
 355
 356         version_manager = VersionManager()
 357         # get the rspec's return format from options
 358         rspec_version = version_manager.get_version(options.get('geni_rspec_version'))
 359         version_string = "rspec_%s" % (rspec_version)
 360
 361         #panos adding the info option to the caching key (can be improved)
 362         if options.get('info'):
 363             version_string = version_string + "_"+options.get('info', 'default')
 364
 365         # look in cache first
 366         if cached_requested and self.cache and not slice_hrn:
 367             rspec = self.cache.get(version_string)
 368             if rspec:
 369                 logger.debug("OpenStackDriver.ListResources: returning cached advertisement")
 370                 return rspec
 371
 372         #panos: passing user-defined options
 373         #print "manager options = ",options
 374         aggregate = OSAggregate(self)
 375         rspec =  aggregate.get_rspec(slice_xrn=slice_urn, version=rspec_version,
 376                                      options=options)
 377
 378         # cache the result
 379         if self.cache and not slice_hrn:
 380             logger.debug("OpenStackDriver.ListResources: stores advertisement in cache")
 381             self.cache.add(version_string, rspec)
 382
 383         return rspec
 384
 385     def sliver_status (self, slice_urn, slice_hrn):
 386         # find out where this slice is currently running
 387         project_name = hrn_to_os_slicename(slice_hrn)
 388         project = self.shell.auth_manager.get_project(project_name)
 389         instances = self.shell.db.instance_get_all_by_project(project_name)
 390         if len(instances) == 0:
 391             raise SliverDoesNotExist("You have not allocated any slivers here")
 392
 393         result = {}
 394         top_level_status = 'unknown'
 395         if instances:
 396             top_level_status = 'ready'
 397         result['geni_urn'] = slice_urn
 398         result['plos_login'] = 'root'
 399         result['plos_expires'] = None
 400
 401         resources = []
 402         for instance in instances:
 403             res = {}
 404             # instances are accessed by ip, not hostname. We need to report the ip
 405             # somewhere so users know where to ssh to.
 406             res['plos_hostname'] = instance.hostname
 407             res['plos_created_at'] = datetime_to_string(utcparse(instance.created_at))
 408             res['plos_boot_state'] = instance.vm_state
 409             res['plos_sliver_type'] = instance.instance_type.name
 410             sliver_id =  Xrn(slice_urn).get_sliver_id(instance.project_id, \
 411                                                       instance.hostname, instance.id)
 412             res['geni_urn'] = sliver_id
 413
 414             if instance.vm_state == 'running':
 415                 res['boot_state'] = 'ready'
 416                 res['geni_status'] = 'ready'
 417             else:
 418                 res['boot_state'] = 'unknown'
 419                 res['geni_status'] = 'unknown'
 420             resources.append(res)
 421
 422         result['geni_status'] = top_level_status
 423         result['geni_resources'] = resources
 424         return result
 425
 426     def create_sliver (self, slice_urn, slice_hrn, creds, rspec_string, users, options):
 427
 428         aggregate = OSAggregate(self)
 429         rspec = RSpec(rspec_string)
 430         instance_name = hrn_to_os_slicename(slice_hrn)
 431
 432         # assume first user is the caller and use their context
 433         # for the ec2/euca api connection. Also, use the first users
 434         # key as the project key.
 435         key_name = None
 436         if len(users) > 1:
 437             key_name = aggregate.create_instance_key(slice_hrn, users[0])
 438
 439         # collect public keys
 440         pubkeys = []
 441         for user in users:
 442             pubkeys.extend(user['keys'])
 443
 444         aggregate.run_instances(instance_name, rspec_string, key_name, pubkeys)
 445
 446         return aggregate.get_rspec(slice_xrn=slice_urn, version=rspec.version)
 447
 448     def delete_sliver (self, slice_urn, slice_hrn, creds, options):
 449         aggregate = OSAggregate(self)
 450         project_name = hrn_to_os_slicename(slice_hrn)
 451         return aggregate.delete_instances(project_name)
 452
 453     def update_sliver(self, slice_urn, slice_hrn, rspec, creds, options):
 454         name = hrn_to_os_slicename(slice_hrn)
 455         aggregate = OSAggregate(self)
 456         return aggregate.update_instances(name)
 457
 458     def renew_sliver (self, slice_urn, slice_hrn, creds, expiration_time, options):
 459         return True
 460
 461     def start_slice (self, slice_urn, slice_hrn, creds):
 462         return 1
 463
 464     def stop_slice (self, slice_urn, slice_hrn, creds):
 465         name = OSXrn(xrn=slice_urn).name
 466         aggregate = OSAggregate(self)
 467         return aggregate.stop_instances(name)
 468
 469     def reset_slice (self, slice_urn, slice_hrn, creds):
 470         raise SfaNotImplemented ("reset_slice not available at this interface")
 471
 472     # xxx this code is quite old and has not run for ages
 473     # it is obviously totally broken and needs a rewrite
 474     def get_ticket (self, slice_urn, slice_hrn, creds, rspec_string, options):
 475         raise SfaNotImplemented,"OpenStackDriver.get_ticket needs a rewrite"
 476 # please keep this code for future reference
 477 #        slices = PlSlices(self)
 478 #        peer = slices.get_peer(slice_hrn)
 479 #        sfa_peer = slices.get_sfa_peer(slice_hrn)
 480 #
 481 #        # get the slice record
 482 #        credential = api.getCredential()
 483 #        interface = api.registries[api.hrn]
 484 #        registry = api.server_proxy(interface, credential)
 485 #        records = registry.Resolve(xrn, credential)
 486 #
 487 #        # make sure we get a local slice record
 488 #        record = None
 489 #        for tmp_record in records:
 490 #            if tmp_record['type'] == 'slice' and \
 491 #               not tmp_record['peer_authority']:
 492 #    #Error (E0602, GetTicket): Undefined variable 'SliceRecord'
 493 #                slice_record = SliceRecord(dict=tmp_record)
 494 #        if not record:
 495 #            raise RecordNotFound(slice_hrn)
 496 #
 497 #        # similar to CreateSliver, we must verify that the required records exist
 498 #        # at this aggregate before we can issue a ticket
 499 #        # parse rspec
 500 #        rspec = RSpec(rspec_string)
 501 #        requested_attributes = rspec.version.get_slice_attributes()
 502 #
 503 #        # ensure site record exists
 504 #        site = slices.verify_site(slice_hrn, slice_record, peer, sfa_peer)
 505 #        # ensure slice record exists
 506 #        slice = slices.verify_slice(slice_hrn, slice_record, peer, sfa_peer)
 507 #        # ensure person records exists
 508 #    # xxx users is undefined in this context
 509 #        persons = slices.verify_persons(slice_hrn, slice, users, peer, sfa_peer)
 510 #        # ensure slice attributes exists
 511 #        slices.verify_slice_attributes(slice, requested_attributes)
 512 #
 513 #        # get sliver info
 514 #        slivers = slices.get_slivers(slice_hrn)
 515 #
 516 #        if not slivers:
 517 #            raise SliverDoesNotExist(slice_hrn)
 518 #
 519 #        # get initscripts
 520 #        initscripts = []
 521 #        data = {
 522 #            'timestamp': int(time.time()),
 523 #            'initscripts': initscripts,
 524 #            'slivers': slivers
 525 #        }
 526 #
 527 #        # create the ticket
 528 #        object_gid = record.get_gid_object()
 529 #        new_ticket = SfaTicket(subject = object_gid.get_subject())
 530 #        new_ticket.set_gid_caller(api.auth.client_gid)
 531 #        new_ticket.set_gid_object(object_gid)
 532 #        new_ticket.set_issuer(key=api.key, subject=self.hrn)
 533 #        new_ticket.set_pubkey(object_gid.get_pubkey())
 534 #        new_ticket.set_attributes(data)
 535 #        new_ticket.set_rspec(rspec)
 536 #        #new_ticket.set_parent(api.auth.hierarchy.get_auth_ticket(auth_hrn))
 537 #        new_ticket.encode()
 538 #        new_ticket.sign()
 539 #
 540 #        return new_ticket.save_to_string(save_parents=True)