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