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         zones = self.driver.shell.db.zone_get_all()
  87         if not zones:
  88             zones = ['cloud']
  89         else:
  90             zones = [zone.name for zone in zones]
  91         return zones
  92
  93     def get_slice_nodes(self, slice_xrn):
  94         image_manager = ImageManager(self.driver)
  95
  96         zones = self.get_availability_zones()
  97         name = hrn_to_os_slicename(slice_xrn)
  98         instances = self.driver.shell.db.instance_get_all_by_project(name)
  99         rspec_nodes = []
 100         for instance in instances:
 101             rspec_node = Node()
 102             interfaces = []
 103             for fixed_ip in instance.fixed_ips:
 104                 if_xrn = PlXrn(auth=self.driver.hrn,
 105                                interface='node%s:eth0' % (instance.hostname))
 106                 interface = Interface({'component_id': if_xrn.urn})
 107                 interface['ips'] =  [{'address': fixed_ip['address'],
 108                                      'netmask': fixed_ip['network'].netmask,
 109                                      'type': 'ipv4'}]
 110                 interfaces.append(interface)
 111             if instance.availability_zone:
 112                 node_xrn = OSXrn(instance.availability_zone, 'node')
 113             else:
 114                 node_xrn = OSXrn('cloud', 'node')
 115
 116             rspec_node['component_id'] = node_xrn.urn
 117             rspec_node['component_name'] = node_xrn.name
 118             rspec_node['component_manager_id'] = Xrn(self.driver.hrn, 'authority+cm').get_urn()
 119             sliver = instance_to_sliver(instance)
 120             disk_image = image_manager.get_disk_image(instance.image_ref)
 121             sliver['disk_image'] = [disk_image.to_rspec_object()]
 122             rspec_node['slivers'] = [sliver]
 123             rspec_node['interfaces'] = interfaces
 124             # slivers always provide the ssh service
 125             hostname = None
 126             for interface in interfaces:
 127                 if 'ips' in interface and interface['ips'] and \
 128                 isinstance(interface['ips'], list):
 129                     if interface['ips'][0].get('address'):
 130                         hostname = interface['ips'][0].get('address')
 131                         break
 132             login = Login({'authentication': 'ssh-keys',
 133                            'hostname': hostname,
 134                            'port':'22', 'username': 'root'})
 135             service = Services({'login': login})
 136             rspec_node['services'] = [service]
 137             rspec_nodes.append(rspec_node)
 138         return rspec_nodes
 139
 140     def get_aggregate_nodes(self):
 141         zones = self.get_availability_zones()
 142         # available sliver/instance/vm types
 143         instances = self.driver.shell.db.instance_type_get_all().values()
 144         # available images
 145         image_manager = ImageManager(self.driver)
 146         disk_images = image_manager.get_available_disk_images()
 147         disk_image_objects = [image.to_rspec_object() \
 148                                for image in disk_images]
 149         rspec_nodes = []
 150         for zone in zones:
 151             rspec_node = Node()
 152             xrn = OSXrn(zone, 'node')
 153             rspec_node['component_id'] = xrn.urn
 154             rspec_node['component_name'] = xrn.name
 155             rspec_node['component_manager_id'] = Xrn(self.driver.hrn, 'authority+cm').get_urn()
 156             rspec_node['exclusive'] = 'false'
 157             rspec_node['hardware_types'] = [HardwareType({'name': 'plos-pc'}),
 158                                                 HardwareType({'name': 'pc'})]
 159             slivers = []
 160             for instance in instances:
 161                 sliver = instance_to_sliver(instance)
 162                 sliver['disk_image'] = disk_image_objects
 163                 slivers.append(sliver)
 164
 165             rspec_node['slivers'] = slivers
 166             rspec_nodes.append(rspec_node)
 167
 168         return rspec_nodes
 169
 170
 171     def create_project(self, slicename, users, options={}):
 172         """
 173         Create the slice if it doesn't alredy exist. Create user
 174         accounts that don't already exist
 175         """
 176         from nova.exception import ProjectNotFound, UserNotFound
 177         for user in users:
 178             username = Xrn(user['urn']).get_leaf()
 179             try:
 180                 self.driver.shell.auth_manager.get_user(username)
 181             except UserNotFound:
 182                 self.driver.shell.auth_manager.create_user(username)
 183             self.verify_user_keys(username, user['keys'], options)
 184
 185         try:
 186             slice = self.driver.shell.auth_manager.get_project(slicename)
 187         except ProjectNotFound:
 188             # assume that the first user is the project manager
 189             proj_manager = Xrn(users[0]['urn']).get_leaf()
 190             self.driver.shell.auth_manager.create_project(slicename, proj_manager)
 191
 192     def verify_user_keys(self, username, keys, options={}):
 193         """
 194         Add requested keys.
 195         """
 196         append = options.get('append', True)
 197         existing_keys = self.driver.shell.db.key_pair_get_all_by_user(username)
 198         existing_pub_keys = [key.public_key for key in existing_keys]
 199         removed_pub_keys = set(existing_pub_keys).difference(keys)
 200         added_pub_keys = set(keys).difference(existing_pub_keys)
 201         pubkeys = []
 202         # add new keys
 203         for public_key in added_pub_keys:
 204             key = {}
 205             key['user_id'] = username
 206             key['name'] =  username
 207             key['public_key'] = public_key
 208             self.driver.shell.db.key_pair_create(key)
 209
 210         # remove old keys
 211         if not append:
 212             for key in existing_keys:
 213                 if key.public_key in removed_pub_keys:
 214                     self.driver.shell.db.key_pair_destroy(username, key.name)
 215
 216
 217     def create_security_group(self, slicename, fw_rules=[]):
 218         # use default group by default
 219         group_name = 'default'
 220         if isinstance(fw_rules, list) and fw_rules:
 221             # Each sliver get's its own security group.
 222             # Keep security group names unique by appending some random
 223             # characters on end.
 224             random_name = "".join([random.choice(string.letters+string.digits)
 225                                            for i in xrange(6)])
 226             group_name = slicename + random_name
 227             security_group = SecurityGroup(self.driver)
 228             security_group.create_security_group(group_name)
 229             for rule in fw_rules:
 230                 security_group.add_rule_to_group(group_name,
 231                                              protocol = rule.get('protocol'),
 232                                              cidr_ip = rule.get('cidr_ip'),
 233                                              port_range = rule.get('port_range'),
 234                                              icmp_type_code = rule.get('icmp_type_code'))
 235         return group_name
 236
 237     def add_rule_to_security_group(self, group_name, **kwds):
 238         security_group = SecurityGroup(self.driver)
 239         security_group.add_rule_to_group(group_name=group_name,
 240                                          protocol=kwds.get('protocol'),
 241                                          cidr_ip =kwds.get('cidr_ip'),
 242                                          icmp_type_code = kwds.get('icmp_type_code'))
 243
 244
 245     def reserve_instance(self, image_id, kernel_id, ramdisk_id, \
 246                          instance_type, key_name, user_data, group_name):
 247         conn  = self.driver.euca_shell.get_euca_connection()
 248         logger.info('Reserving an instance: image: %s, kernel: ' \
 249                     '%s, ramdisk: %s, type: %s, key: %s' % \
 250                     (image_id, kernel_id, ramdisk_id,
 251                     instance_type, key_name))
 252         try:
 253             reservation = conn.run_instances(image_id=image_id,
 254                                              kernel_id=kernel_id,
 255                                              ramdisk_id=ramdisk_id,
 256                                              instance_type=instance_type,
 257                                              key_name=key_name,
 258                                              user_data = user_data,
 259                                              security_groups=[group_name])
 260                                              #placement=zone,
 261                                              #min_count=min_count,
 262                                              #max_count=max_count,
 263
 264         except Exception, err:
 265             logger.log_exc(err)
 266
 267
 268     def run_instances(self, slicename, rspec, keyname, pubkeys):
 269         """
 270         Create the security groups and instances.
 271         """
 272         # the default image to use for instnaces that dont
 273         # explicitly request an image.
 274         # Just choose the first available image for now.
 275         image_manager = ImageManager(self.driver)
 276         available_images = image_manager.get_available_disk_images()
 277         default_image_id = None
 278         default_aki_id  = None
 279         default_ari_id = None
 280         default_image = available_images[0]
 281         default_image_id = ec2_id(default_image.id, default_image.container_format)
 282         default_aki_id = ec2_id(default_image.kernel_id, 'aki')
 283         default_ari_id = ec2_id(default_image.ramdisk_id, 'ari')
 284
 285         # get requested slivers
 286         rspec = RSpec(rspec)
 287         user_data = "\n".join(pubkeys)
 288         requested_instances = defaultdict(list)
 289         # iterate over clouds/zones/nodes
 290         for node in rspec.version.get_nodes_with_slivers():
 291             instance_types = node.get('slivers', [])
 292             if isinstance(instance_types, list):
 293                 # iterate over sliver/instance types
 294                 for instance_type in instance_types:
 295                     fw_rules = instance_type.get('fw_rules', [])
 296                     group_name = self.create_security_group(slicename, fw_rules)
 297                     ami_id = default_image_id
 298                     aki_id = default_aki_id
 299                     ari_id = default_ari_id
 300                     req_image = instance_type.get('disk_image')
 301                     if req_image and isinstance(req_image, list):
 302                         req_image_name = req_image[0]['name']
 303                         disk_image = image_manager.get_disk_image(name=req_image_name)
 304                         if disk_image:
 305                             ami_id = ec2_id(disk_image.id, disk_image.container_format)
 306                             aki_id = ec2_id(disk_image.kernel_id, 'aki')
 307                             ari_id = ec2_id(disk_image.ramdisk_id, 'ari')
 308                     # start the instance
 309                     self.reserve_instance(image_id=ami_id,
 310                                           kernel_id=aki_id,
 311                                           ramdisk_id=ari_id,
 312                                           instance_type=instance_type['name'],
 313                                           key_name=keyname,
 314                                           user_data=user_data,
 315                                           group_name=group_name)
 316
 317
 318     def delete_instances(self, project_name):
 319         instances = self.driver.shell.db.instance_get_all_by_project(project_name)
 320         security_group_manager = SecurityGroup(self.driver)
 321         for instance in instances:
 322             # deleate this instance's security groups
 323             for security_group in instance.security_groups:
 324                 # dont delete the default security group
 325                 if security_group.name != 'default':
 326                     security_group_manager.delete_security_group(security_group.name)
 327             # destroy instance
 328             self.driver.shell.db.instance_destroy(instance.id)
 329         return 1
 330
 331     def stop_instances(self, project_name):
 332         instances = self.driver.shell.db.instance_get_all_by_project(project_name)
 333         for instance in instances:
 334             self.driver.shell.db.instance_stop(instance.id)
 335         return 1
 336
 337     def update_instances(self, project_name):
 338         pass