sfa/openstack/osaggregate.py

   1
   2 import os
   3 import socket
   4 import base64
   5 import string
   6 import random
   7 from collections import defaultdict
   8 from nova.exception import ImageNotFound
   9 from nova.api.ec2.cloud import CloudController
  10 from sfa.util.faults import SfaAPIError
  11 from sfa.rspecs.rspec import RSpec
  12 from sfa.rspecs.elements.hardware_type import HardwareType
  13 from sfa.rspecs.elements.node import Node
  14 from sfa.rspecs.elements.sliver import Sliver
  15 from sfa.rspecs.elements.login import Login
  16 from sfa.rspecs.elements.disk_image import DiskImage
  17 from sfa.rspecs.elements.services import Services
  18 from sfa.rspecs.elements.interface import Interface
  19 from sfa.util.xrn import Xrn
  20 from sfa.planetlab.plxrn import PlXrn
  21 from sfa.openstack.osxrn import OSXrn, hrn_to_os_slicename
  22 from sfa.rspecs.version_manager import VersionManager
  23 from sfa.openstack.image import ImageManager
  24 from sfa.openstack.security_group import SecurityGroup
  25 from sfa.util.sfalogging import logger
  26
  27 def instance_to_sliver(instance, slice_xrn=None):
  28     # should include?
  29     # * instance.image_ref
  30     # * instance.kernel_id
  31     # * instance.ramdisk_id
  32     import nova.db.sqlalchemy.models
  33     name=None
  34     type=None
  35     sliver_id = None
  36     if isinstance(instance, dict):
  37         # this is an isntance type dict
  38         name = instance['name']
  39         type = instance['name']
  40     elif isinstance(instance, nova.db.sqlalchemy.models.Instance):
  41         # this is an object that describes a running instance
  42         name = instance.display_name
  43         type = instance.instance_type.name
  44     else:
  45         raise SfaAPIError("instnace must be an instance_type dict or" + \
  46                            " a nova.db.sqlalchemy.models.Instance object")
  47     if slice_xrn:
  48         xrn = Xrn(slice_xrn, 'slice')
  49         sliver_id = xrn.get_sliver_id(instance.project_id, instance.hostname, instance.id)
  50
  51     sliver = Sliver({'slice_id': sliver_id,
  52                      'name': name,
  53                      'type':  type,
  54                      'tags': []})
  55     return sliver
  56
  57
  58 def ec2_id(id=None, type=None):
  59     ec2_id = None
  60     if type == 'ovf':
  61         type = 'ami'
  62     if id and type:
  63         ec2_id = CloudController.image_ec2_id(id, type)
  64     return ec2_id
  65
  66
  67 class OSAggregate:
  68
  69     def __init__(self, driver):
  70         self.driver = driver
  71
  72     def get_rspec(self, slice_xrn=None, version=None, options={}):
  73         version_manager = VersionManager()
  74         version = version_manager.get_version(version)
  75         if not slice_xrn:
  76             rspec_version = version_manager._get_version(version.type, version.version, 'ad')
  77             nodes = self.get_aggregate_nodes()
  78         else:
  79             rspec_version = version_manager._get_version(version.type, version.version, 'manifest')
  80             nodes = self.get_slice_nodes(slice_xrn)
  81         rspec = RSpec(version=rspec_version, user_options=options)
  82         rspec.version.add_nodes(nodes)
  83         return rspec.toxml()
  84
  85     def get_availability_zones(self):
  86         try:
  87             # pre essex releases
  88             zones = self.driver.shell.db.zone_get_all()
  89         except:
  90             # essex release
  91             zones = self.driver.shell.db.dnsdomain_list()
  92
  93         if not zones:
  94             zones = ['cloud']
  95         else:
  96             zones = [zone.name for zone in zones]
  97         return zones
  98
  99     def get_slice_nodes(self, slice_xrn):
 100         image_manager = ImageManager(self.driver)
 101
 102         zones = self.get_availability_zones()
 103         name = hrn_to_os_slicename(slice_xrn)
 104         instances = self.driver.shell.db.instance_get_all_by_project(name)
 105         rspec_nodes = []
 106         for instance in instances:
 107             rspec_node = Node()
 108             interfaces = []
 109             for fixed_ip in instance.fixed_ips:
 110                 if_xrn = PlXrn(auth=self.driver.hrn,
 111                                interface='node%s:eth0' % (instance.hostname))
 112                 interface = Interface({'component_id': if_xrn.urn})
 113                 interface['ips'] =  [{'address': fixed_ip['address'],
 114                                      'netmask': fixed_ip['network'].netmask,
 115                                      'type': 'ipv4'}]
 116                 interface['floating_ips'] = []
 117                 for floating_ip in fixed_ip.floating_ips:
 118                     interface['floating_ips'].append(floating_ip.address)
 119                 interfaces.append(interface)
 120             if instance.availability_zone:
 121                 node_xrn = OSXrn(instance.availability_zone, 'node')
 122             else:
 123                 node_xrn = OSXrn('cloud', 'node')
 124
 125             rspec_node['component_id'] = node_xrn.urn
 126             rspec_node['component_name'] = node_xrn.name
 127             rspec_node['component_manager_id'] = Xrn(self.driver.hrn, 'authority+cm').get_urn()
 128             sliver = instance_to_sliver(instance)
 129             disk_image = image_manager.get_disk_image(instance.image_ref)
 130             sliver['disk_image'] = [disk_image.to_rspec_object()]
 131             rspec_node['slivers'] = [sliver]
 132             rspec_node['interfaces'] = interfaces
 133             # slivers always provide the ssh service
 134             rspec_node['services'] = []
 135             for interface in interfaces:
 136                 if 'floating_ips' in interface:
 137                     for hostname in interface['floating_ips']:
 138                         login = Login({'authentication': 'ssh-keys',
 139                                        'hostname': hostname,
 140                                        'port':'22', 'username': 'root'})
 141                         service = Services({'login': login})
 142                         rspec_node['services'].append(service)
 143             rspec_nodes.append(rspec_node)
 144         return rspec_nodes
 145
 146     def get_aggregate_nodes(self):
 147         zones = self.get_availability_zones()
 148         # available sliver/instance/vm types
 149         instances = self.driver.shell.db.instance_type_get_all()
 150         if isinstance(instances, dict):
 151             instances = instances.values()
 152         # available images
 153         image_manager = ImageManager(self.driver)
 154         disk_images = image_manager.get_available_disk_images()
 155         disk_image_objects = [image.to_rspec_object() \
 156                                for image in disk_images]
 157         rspec_nodes = []
 158         for zone in zones:
 159             rspec_node = Node()
 160             xrn = OSXrn(zone, 'node')
 161             rspec_node['component_id'] = xrn.urn
 162             rspec_node['component_name'] = xrn.name
 163             rspec_node['component_manager_id'] = Xrn(self.driver.hrn, 'authority+cm').get_urn()
 164             rspec_node['exclusive'] = 'false'
 165             rspec_node['hardware_types'] = [HardwareType({'name': 'plos-pc'}),
 166                                                 HardwareType({'name': 'pc'})]
 167             slivers = []
 168             for instance in instances:
 169                 sliver = instance_to_sliver(instance)
 170                 sliver['disk_image'] = disk_image_objects
 171                 slivers.append(sliver)
 172
 173             rspec_node['slivers'] = slivers
 174             rspec_nodes.append(rspec_node)
 175
 176         return rspec_nodes
 177
 178
 179     def create_project(self, slicename, users, options={}):
 180         """
 181         Create the slice if it doesn't alredy exist. Create user
 182         accounts that don't already exist
 183         """
 184         from nova.exception import ProjectNotFound, UserNotFound
 185         for user in users:
 186             username = Xrn(user['urn']).get_leaf()
 187             try:
 188                 self.driver.shell.auth_manager.get_user(username)
 189             except UserNotFound:
 190                 self.driver.shell.auth_manager.create_user(username)
 191             self.verify_user_keys(username, user['keys'], options)
 192
 193         try:
 194             slice = self.driver.shell.auth_manager.get_project(slicename)
 195         except ProjectNotFound:
 196             # assume that the first user is the project manager
 197             proj_manager = Xrn(users[0]['urn']).get_leaf()
 198             self.driver.shell.auth_manager.create_project(slicename, proj_manager)
 199
 200     def verify_user_keys(self, username, keys, options={}):
 201         """
 202         Add requested keys.
 203         """
 204         append = options.get('append', True)
 205         existing_keys = self.driver.shell.db.key_pair_get_all_by_user(username)
 206         existing_pub_keys = [key.public_key for key in existing_keys]
 207         removed_pub_keys = set(existing_pub_keys).difference(keys)
 208         added_pub_keys = set(keys).difference(existing_pub_keys)
 209         pubkeys = []
 210         # add new keys
 211         for public_key in added_pub_keys:
 212             key = {}
 213             key['user_id'] = username
 214             key['name'] =  username
 215             key['public_key'] = public_key
 216             self.driver.shell.db.key_pair_create(key)
 217
 218         # remove old keys
 219         if not append:
 220             for key in existing_keys:
 221                 if key.public_key in removed_pub_keys:
 222                     self.driver.shell.db.key_pair_destroy(username, key.name)
 223
 224
 225     def create_security_group(self, slicename, fw_rules=[]):
 226         # use default group by default
 227         group_name = 'default'
 228         if isinstance(fw_rules, list) and fw_rules:
 229             # Each sliver get's its own security group.
 230             # Keep security group names unique by appending some random
 231             # characters on end.
 232             random_name = "".join([random.choice(string.letters+string.digits)
 233                                            for i in xrange(6)])
 234             group_name = slicename + random_name
 235             security_group = SecurityGroup(self.driver)
 236             security_group.create_security_group(group_name)
 237             for rule in fw_rules:
 238                 security_group.add_rule_to_group(group_name,
 239                                              protocol = rule.get('protocol'),
 240                                              cidr_ip = rule.get('cidr_ip'),
 241                                              port_range = rule.get('port_range'),
 242                                              icmp_type_code = rule.get('icmp_type_code'))
 243         return group_name
 244
 245     def add_rule_to_security_group(self, group_name, **kwds):
 246         security_group = SecurityGroup(self.driver)
 247         security_group.add_rule_to_group(group_name=group_name,
 248                                          protocol=kwds.get('protocol'),
 249                                          cidr_ip =kwds.get('cidr_ip'),
 250                                          icmp_type_code = kwds.get('icmp_type_code'))
 251
 252
 253     def reserve_instance(self, image_id, kernel_id, ramdisk_id, \
 254                          instance_type, key_name, user_data, group_name):
 255         conn  = self.driver.euca_shell.get_euca_connection()
 256         logger.info('Reserving an instance: image: %s, kernel: ' \
 257                     '%s, ramdisk: %s, type: %s, key: %s' % \
 258                     (image_id, kernel_id, ramdisk_id,
 259                     instance_type, key_name))
 260         try:
 261             reservation = conn.run_instances(image_id=image_id,
 262                                              kernel_id=kernel_id,
 263                                              ramdisk_id=ramdisk_id,
 264                                              instance_type=instance_type,
 265                                              key_name=key_name,
 266                                              user_data = user_data,
 267                                              security_groups=[group_name])
 268                                              #placement=zone,
 269                                              #min_count=min_count,
 270                                              #max_count=max_count,
 271
 272         except Exception, err:
 273             logger.log_exc(err)
 274
 275
 276     def run_instances(self, slicename, rspec, keyname, pubkeys):
 277         """
 278         Create the security groups and instances.
 279         """
 280         # the default image to use for instnaces that dont
 281         # explicitly request an image.
 282         # Just choose the first available image for now.
 283         image_manager = ImageManager(self.driver)
 284         available_images = image_manager.get_available_disk_images()
 285         default_image_id = None
 286         default_aki_id  = None
 287         default_ari_id = None
 288         default_image = available_images[0]
 289         default_image_id = ec2_id(default_image.id, default_image.container_format)
 290         default_aki_id = ec2_id(default_image.kernel_id, 'aki')
 291         default_ari_id = ec2_id(default_image.ramdisk_id, 'ari')
 292
 293         # get requested slivers
 294         rspec = RSpec(rspec)
 295         user_data = "\n".join(pubkeys)
 296         requested_instances = defaultdict(list)
 297         # iterate over clouds/zones/nodes
 298         for node in rspec.version.get_nodes_with_slivers():
 299             instance_types = node.get('slivers', [])
 300             if isinstance(instance_types, list):
 301                 # iterate over sliver/instance types
 302                 for instance_type in instance_types:
 303                     fw_rules = instance_type.get('fw_rules', [])
 304                     group_name = self.create_security_group(slicename, fw_rules)
 305                     ami_id = default_image_id
 306                     aki_id = default_aki_id
 307                     ari_id = default_ari_id
 308                     req_image = instance_type.get('disk_image')
 309                     if req_image and isinstance(req_image, list):
 310                         req_image_name = req_image[0]['name']
 311                         disk_image = image_manager.get_disk_image(name=req_image_name)
 312                         if disk_image:
 313                             ami_id = ec2_id(disk_image.id, disk_image.container_format)
 314                             aki_id = ec2_id(disk_image.kernel_id, 'aki')
 315                             ari_id = ec2_id(disk_image.ramdisk_id, 'ari')
 316                     # start the instance
 317                     self.reserve_instance(image_id=ami_id,
 318                                           kernel_id=aki_id,
 319                                           ramdisk_id=ari_id,
 320                                           instance_type=instance_type['name'],
 321                                           key_name=keyname,
 322                                           user_data=user_data,
 323                                           group_name=group_name)
 324
 325
 326     def delete_instances(self, project_name):
 327         instances = self.driver.shell.db.instance_get_all_by_project(project_name)
 328         security_group_manager = SecurityGroup(self.driver)
 329         for instance in instances:
 330             # deleate this instance's security groups
 331             for security_group in instance.security_groups:
 332                 # dont delete the default security group
 333                 if security_group.name != 'default':
 334                     security_group_manager.delete_security_group(security_group.name)
 335             # destroy instance
 336             self.driver.shell.db.instance_destroy(instance.id)
 337         return 1
 338
 339     def stop_instances(self, project_name):
 340         instances = self.driver.shell.db.instance_get_all_by_project(project_name)
 341         for instance in instances:
 342             self.driver.shell.db.instance_stop(instance.id)
 343         return 1
 344
 345     def update_instances(self, project_name):
 346         pass