merged trunk

bin/nova-manage

@@ -53,6 +53,7 @@
   CLI interface for nova management.
 """
 
+import ast
 import gettext
 import glob
 import json
@@ -85,11 +86,13 @@ from nova import quota
 from nova import rpc
 from nova import utils
 from nova import version
+from nova import vsa
 from nova.api.ec2 import ec2utils
 from nova.auth import manager
 from nova.cloudpipe import pipelib
 from nova.compute import instance_types
 from nova.db import migration
+from nova.volume import volume_types
 
 FLAGS = flags.FLAGS
 flags.DECLARE('fixed_range', 'nova.network.manager')
@@ -1097,6 +1100,477 @@ class VersionCommands(object):
         self.list()
 
 
+class VsaCommands(object):
+    """Methods for dealing with VSAs"""
+
+    def __init__(self, *args, **kwargs):
+        self.manager = manager.AuthManager()
+        self.vsa_api = vsa.API()
+        self.context = context.get_admin_context()
+
+        self._format_str_vsa = "%(id)-5s %(vsa_id)-15s %(name)-25s "\
+                    "%(type)-10s %(vcs)-6s %(drives)-9s %(stat)-10s "\
+                    "%(az)-10s %(time)-10s"
+        self._format_str_volume = "\t%(id)-4s %(name)-15s %(size)-5s "\
+                    "%(stat)-10s %(att)-20s %(time)s"
+        self._format_str_drive = "\t%(id)-4s %(name)-15s %(size)-5s "\
+                    "%(stat)-10s %(host)-20s %(type)-4s %(tname)-10s %(time)s"
+        self._format_str_instance = "\t%(id)-4s %(name)-10s %(dname)-20s "\
+                    "%(image)-12s %(type)-10s %(fl_ip)-15s %(fx_ip)-15s "\
+                    "%(stat)-10s %(host)-15s %(time)s"
+
+    def _print_vsa_header(self):
+        print self._format_str_vsa %\
+               dict(id=_('ID'),
+                    vsa_id=_('vsa_id'),
+                    name=_('displayName'),
+                    type=_('vc_type'),
+                    vcs=_('vc_cnt'),
+                    drives=_('drive_cnt'),
+                    stat=_('status'),
+                    az=_('AZ'),
+                    time=_('createTime'))
+
+    def _print_vsa(self, vsa):
+        print self._format_str_vsa %\
+                dict(id=vsa['id'],
+                     vsa_id=vsa['name'],
+                     name=vsa['display_name'],
+                     type=vsa['vsa_instance_type'].get('name', None),
+                     vcs=vsa['vc_count'],
+                     drives=vsa['vol_count'],
+                     stat=vsa['status'],
+                     az=vsa['availability_zone'],
+                     time=str(vsa['created_at']))
+
+    def _print_volume_header(self):
+        print _('      === Volumes ===')
+        print self._format_str_volume %\
+                dict(id=_('ID'),
+                     name=_('name'),
+                     size=_('size'),
+                     stat=_('status'),
+                     att=_('attachment'),
+                     time=_('createTime'))
+
+    def _print_volume(self, vol):
+        print self._format_str_volume %\
+                dict(id=vol['id'],
+                     name=vol['display_name'] or vol['name'],
+                     size=vol['size'],
+                     stat=vol['status'],
+                     att=vol['attach_status'],
+                     time=str(vol['created_at']))
+
+    def _print_drive_header(self):
+        print _('      === Drives ===')
+        print self._format_str_drive %\
+                dict(id=_('ID'),
+                    name=_('name'),
+                    size=_('size'),
+                    stat=_('status'),
+                    host=_('host'),
+                    type=_('type'),
+                    tname=_('typeName'),
+                    time=_('createTime'))
+
+    def _print_drive(self, drive):
+        if drive['volume_type_id'] is not None and drive.get('volume_type'):
+            drive_type_name = drive['volume_type'].get('name')
+        else:
+            drive_type_name = ''
+
+        print self._format_str_drive %\
+                dict(id=drive['id'],
+                    name=drive['display_name'],
+                    size=drive['size'],
+                    stat=drive['status'],
+                    host=drive['host'],
+                    type=drive['volume_type_id'],
+                    tname=drive_type_name,
+                    time=str(drive['created_at']))
+
+    def _print_instance_header(self):
+        print _('      === Instances ===')
+        print self._format_str_instance %\
+                dict(id=_('ID'),
+                    name=_('name'),
+                    dname=_('disp_name'),
+                    image=_('image'),
+                    type=_('type'),
+                    fl_ip=_('floating_IP'),
+                    fx_ip=_('fixed_IP'),
+                    stat=_('status'),
+                    host=_('host'),
+                    time=_('createTime'))
+
+    def _print_instance(self, vc):
+
+        fixed_addr = None
+        floating_addr = None
+        if vc['fixed_ips']:
+            fixed = vc['fixed_ips'][0]
+            fixed_addr = fixed['address']
+            if fixed['floating_ips']:
+                floating_addr = fixed['floating_ips'][0]['address']
+        floating_addr = floating_addr or fixed_addr
+
+        print self._format_str_instance %\
+                dict(id=vc['id'],
+                    name=ec2utils.id_to_ec2_id(vc['id']),
+                    dname=vc['display_name'],
+                    image=('ami-%08x' % int(vc['image_ref'])),
+                    type=vc['instance_type']['name'],
+                    fl_ip=floating_addr,
+                    fx_ip=fixed_addr,
+                    stat=vc['state_description'],
+                    host=vc['host'],
+                    time=str(vc['created_at']))
+
+    def _list(self, context, vsas, print_drives=False,
+                    print_volumes=False, print_instances=False):
+        if vsas:
+            self._print_vsa_header()
+
+        for vsa in vsas:
+            self._print_vsa(vsa)
+            vsa_id = vsa.get('id')
+
+            if print_instances:
+                instances = self.vsa_api.get_all_vsa_instances(context, vsa_id)
+                if instances:
+                    print
+                    self._print_instance_header()
+                    for instance in instances:
+                        self._print_instance(instance)
+                    print
+
+            if print_drives:
+                drives = self.vsa_api.get_all_vsa_drives(context, vsa_id)
+                if drives:
+                    self._print_drive_header()
+                    for drive in drives:
+                        self._print_drive(drive)
+                    print
+
+            if print_volumes:
+                volumes = self.vsa_api.get_all_vsa_volumes(context, vsa_id)
+                if volumes:
+                    self._print_volume_header()
+                    for volume in volumes:
+                        self._print_volume(volume)
+                    print
+
+    @args('--storage', dest='storage',
+        metavar="[{'drive_name': 'type', 'num_drives': N, 'size': M},..]",
+        help='Initial storage allocation for VSA')
+    @args('--name', dest='name', metavar="<name>", help='VSA name')
+    @args('--description', dest='description', metavar="<description>",
+        help='VSA description')
+    @args('--vc', dest='vc_count', metavar="<number>", help='Number of VCs')
+    @args('--instance_type', dest='instance_type_name', metavar="<name>",
+        help='Instance type name')
+    @args('--image', dest='image_name', metavar="<name>", help='Image name')
+    @args('--shared', dest='shared', action="store_true", default=False,
+        help='Use shared drives')
+    @args('--az', dest='az', metavar="<zone:host>", help='Availability zone')
+    @args('--user', dest="user_id", metavar='<User name>',
+            help='User name')
+    @args('--project', dest="project_id", metavar='<Project name>',
+            help='Project name')
+    def create(self, storage='[]', name=None, description=None, vc_count=1,
+                     instance_type_name=None, image_name=None, shared=None,
+                     az=None, user_id=None, project_id=None):
+        """Create a VSA."""
+
+        if project_id is None:
+            try:
+                project_id = os.getenv("EC2_ACCESS_KEY").split(':')[1]
+            except Exception as exc:
+                print _("Failed to retrieve project id: %(exc)s") % exc
+                raise
+
+        if user_id is None:
+            try:
+                project = self.manager.get_project(project_id)
+                user_id = project.project_manager_id
+            except Exception as exc:
+                print _("Failed to retrieve user info: %(exc)s") % exc
+                raise
+
+        is_admin = self.manager.is_admin(user_id)
+        ctxt = context.RequestContext(user_id, project_id, is_admin)
+        if not is_admin and \
+           not self.manager.is_project_member(user_id, project_id):
+            msg = _("%(user_id)s must be an admin or a "
+                    "member of %(project_id)s")
+            LOG.warn(msg % locals())
+            raise ValueError(msg % locals())
+
+        # Sanity check for storage string
+        storage_list = []
+        if storage is not None:
+            try:
+                storage_list = ast.literal_eval(storage)
+            except:
+                print _("Invalid string format %s") % storage
+                raise
+
+            for node in storage_list:
+                if ('drive_name' not in node) or ('num_drives' not in node):
+                    print (_("Invalid string format for element %s. " \
+                            "Expecting keys 'drive_name' & 'num_drives'"),
+                            str(node))
+                    raise KeyError
+
+        if instance_type_name == '':
+            instance_type_name = None
+        instance_type = instance_types.get_instance_type_by_name(
+                                instance_type_name)
+
+        if image_name == '':
+            image_name = None
+
+        if shared in [None, False, "--full_drives"]:
+            shared = False
+        elif shared in [True, "--shared"]:
+            shared = True
+        else:
+            raise ValueError(_('Shared parameter should be set either to "\
+                "--shared or --full_drives'))
+
+        values = {
+            'display_name': name,
+            'display_description': description,
+            'vc_count': int(vc_count),
+            'instance_type': instance_type,
+            'image_name': image_name,
+            'availability_zone': az,
+            'storage': storage_list,
+            'shared': shared,
+            }
+
+        result = self.vsa_api.create(ctxt, **values)
+        self._list(ctxt, [result])
+
+    @args('--id', dest='vsa_id', metavar="<vsa_id>", help='VSA ID')
+    @args('--name', dest='name', metavar="<name>", help='VSA name')
+    @args('--description', dest='description', metavar="<description>",
+        help='VSA description')
+    @args('--vc', dest='vc_count', metavar="<number>", help='Number of VCs')
+    def update(self, vsa_id, name=None, description=None, vc_count=None):
+        """Updates name/description of vsa and number of VCs."""
+
+        values = {}
+        if name is not None:
+            values['display_name'] = name
+        if description is not None:
+            values['display_description'] = description
+        if vc_count is not None:
+            values['vc_count'] = int(vc_count)
+
+        vsa_id = ec2utils.ec2_id_to_id(vsa_id)
+        result = self.vsa_api.update(self.context, vsa_id=vsa_id, **values)
+        self._list(self.context, [result])
+
+    @args('--id', dest='vsa_id', metavar="<vsa_id>", help='VSA ID')
+    def delete(self, vsa_id):
+        """Delete a VSA."""
+        vsa_id = ec2utils.ec2_id_to_id(vsa_id)
+        self.vsa_api.delete(self.context, vsa_id)
+
+    @args('--id', dest='vsa_id', metavar="<vsa_id>",
+        help='VSA ID (optional)')
+    @args('--all', dest='all', action="store_true", default=False,
+        help='Show all available details')
+    @args('--drives', dest='drives', action="store_true",
+        help='Include drive-level details')
+    @args('--volumes', dest='volumes', action="store_true",
+        help='Include volume-level details')
+    @args('--instances', dest='instances', action="store_true",
+        help='Include instance-level details')
+    def list(self, vsa_id=None, all=False,
+             drives=False, volumes=False, instances=False):
+        """Describe all available VSAs (or particular one)."""
+
+        vsas = []
+        if vsa_id is not None:
+            internal_id = ec2utils.ec2_id_to_id(vsa_id)
+            vsa = self.vsa_api.get(self.context, internal_id)
+            vsas.append(vsa)
+        else:
+            vsas = self.vsa_api.get_all(self.context)
+
+        if all:
+            drives = volumes = instances = True
+
+        self._list(self.context, vsas, drives, volumes, instances)
+
+    def update_capabilities(self):
+        """Forces updates capabilities on all nova-volume nodes."""
+
+        rpc.fanout_cast(context.get_admin_context(),
+                 FLAGS.volume_topic,
+                 {"method": "notification",
+                  "args": {"event": "startup"}})
+
+
+class VsaDriveTypeCommands(object):
+    """Methods for dealing with VSA drive types"""
+
+    def __init__(self, *args, **kwargs):
+        super(VsaDriveTypeCommands, self).__init__(*args, **kwargs)
+        self.context = context.get_admin_context()
+        self._drive_type_template = '%s_%sGB_%sRPM'
+
+    def _list(self, drives):
+        format_str = "%-5s %-30s %-10s %-10s %-10s %-20s %-10s %s"
+        if len(drives):
+            print format_str %\
+                   (_('ID'),
+                    _('name'),
+                    _('type'),
+                    _('size_gb'),
+                    _('rpm'),
+                    _('capabilities'),
+                    _('visible'),
+                    _('createTime'))
+
+        for name, vol_type in drives.iteritems():
+            drive = vol_type.get('extra_specs')
+            print format_str %\
+                (str(vol_type['id']),
+                drive['drive_name'],
+                drive['drive_type'],
+                drive['drive_size'],
+                drive['drive_rpm'],
+                drive.get('capabilities', ''),
+                str(drive.get('visible', '')),
+                str(vol_type['created_at']))
+
+    @args('--type', dest='type', metavar="<type>",
+        help='Drive type (SATA, SAS, SSD, etc.)')
+    @args('--size', dest='size_gb', metavar="<gb>", help='Drive size in GB')
+    @args('--rpm', dest='rpm', metavar="<rpm>", help='RPM')
+    @args('--capabilities', dest='capabilities', default=None,
+        metavar="<string>", help='Different capabilities')
+    @args('--hide', dest='hide', action="store_true", default=False,
+        help='Show or hide drive')
+    @args('--name', dest='name', metavar="<name>", help='Drive name')
+    def create(self, type, size_gb, rpm, capabilities=None,
+                     hide=False, name=None):
+        """Create drive type."""
+
+        hide = True if hide in [True, "True", "--hide", "hide"] else False
+
+        if name is None:
+            name = self._drive_type_template % (type, size_gb, rpm)
+
+        extra_specs = {'type': 'vsa_drive',
+                       'drive_name': name,
+                       'drive_type': type,
+                       'drive_size': size_gb,
+                       'drive_rpm': rpm,
+                       'visible': True,
+                       }
+        if hide:
+            extra_specs['visible'] = False
+
+        if capabilities is not None and capabilities != '':
+            extra_specs['capabilities'] = capabilities
+
+        volume_types.create(self.context, name, extra_specs)
+        result = volume_types.get_volume_type_by_name(self.context, name)
+        self._list({name: result})
+
+    @args('--name', dest='name', metavar="<name>", help='Drive name')
+    @args('--purge', action="store_true", dest='purge', default=False,
+            help='purge record from database')
+    def delete(self, name, purge):
+        """Marks instance types / flavors as deleted"""
+        try:
+            if purge:
+                volume_types.purge(self.context, name)
+                verb = "purged"
+            else:
+                volume_types.destroy(self.context, name)
+                verb = "deleted"
+        except exception.ApiError:
+            print "Valid volume type name is required"
+            sys.exit(1)
+        except exception.DBError, e:
+            print "DB Error: %s" % e
+            sys.exit(2)
+        except:
+            sys.exit(3)
+        else:
+            print "%s %s" % (name, verb)
+
+    @args('--all', dest='all', action="store_true", default=False,
+        help='Show all drives (including invisible)')
+    @args('--name', dest='name', metavar="<name>",
+        help='Show only specified drive')
+    def list(self, all=False, name=None):
+        """Describe all available VSA drive types (or particular one)."""
+
+        all = False if all in ["--all", False, "False"] else True
+
+        search_opts = {'extra_specs': {'type': 'vsa_drive'}}
+        if name is not None:
+            search_opts['extra_specs']['name'] = name
+
+        if all == False:
+            search_opts['extra_specs']['visible'] = '1'
+
+        drives = volume_types.get_all_types(self.context,
+                                            search_opts=search_opts)
+        self._list(drives)
+
+    @args('--name', dest='name', metavar="<name>", help='Drive name')
+    @args('--type', dest='type', metavar="<type>",
+        help='Drive type (SATA, SAS, SSD, etc.)')
+    @args('--size', dest='size_gb', metavar="<gb>", help='Drive size in GB')
+    @args('--rpm', dest='rpm', metavar="<rpm>", help='RPM')
+    @args('--capabilities', dest='capabilities', default=None,
+        metavar="<string>", help='Different capabilities')
+    @args('--visible', dest='visible',
+        metavar="<show|hide>", help='Show or hide drive')
+    def update(self, name, type=None, size_gb=None, rpm=None,
+                     capabilities=None, visible=None):
+        """Update drive type."""
+
+        volume_type = volume_types.get_volume_type_by_name(self.context, name)
+
+        extra_specs = {'type': 'vsa_drive'}
+
+        if type:
+            extra_specs['drive_type'] = type
+
+        if size_gb:
+            extra_specs['drive_size'] = size_gb
+
+        if rpm:
+            extra_specs['drive_rpm'] = rpm
+
+        if capabilities:
+            extra_specs['capabilities'] = capabilities
+
+        if visible is not None:
+            if visible in ["show", True, "True"]:
+                extra_specs['visible'] = True
+            elif visible in ["hide", False, "False"]:
+                extra_specs['visible'] = False
+            else:
+                raise ValueError(_('visible parameter should be set to '\
+                                   'show or hide'))
+
+        db.api.volume_type_extra_specs_update_or_create(self.context,
+                    volume_type['id'],
+                    extra_specs)
+        result = volume_types.get_volume_type_by_name(self.context, name)
+        self._list({name: result})
+
+
 class VolumeCommands(object):
     """Methods for dealing with a cloud in an odd state"""
 
@@ -1483,6 +1957,7 @@ CATEGORIES = [
     ('agent', AgentBuildCommands),
     ('config', ConfigCommands),
     ('db', DbCommands),
+    ('drive', VsaDriveTypeCommands),
     ('fixed', FixedIpCommands),
     ('flavor', InstanceTypeCommands),
     ('floating', FloatingIpCommands),
@@ -1498,7 +1973,8 @@ CATEGORIES = [
     ('version', VersionCommands),
     ('vm', VmCommands),
     ('volume', VolumeCommands),
-    ('vpn', VpnCommands)]
+    ('vpn', VpnCommands),
+    ('vsa', VsaCommands)]
 
 
 def lazy_match(name, key_value_tuples):

bin/nova-vsa

@@ -0,0 +1,49 @@
+#!/usr/bin/env python
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright (c) 2011 Zadara Storage Inc.
+# Copyright (c) 2011 OpenStack LLC.
+#
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+"""Starter script for Nova VSA."""
+
+import eventlet
+eventlet.monkey_patch()
+
+import os
+import sys
+
+# If ../nova/__init__.py exists, add ../ to Python search path, so that
+# it will override what happens to be installed in /usr/(local/)lib/python...
+possible_topdir = os.path.normpath(os.path.join(os.path.abspath(sys.argv[0]),
+                                   os.pardir,
+                                   os.pardir))
+if os.path.exists(os.path.join(possible_topdir, 'nova', '__init__.py')):
+    sys.path.insert(0, possible_topdir)
+
+
+from nova import flags
+from nova import log as logging
+from nova import service
+from nova import utils
+
+if __name__ == '__main__':
+    utils.default_flagfile()
+    flags.FLAGS(sys.argv)
+    logging.setup()
+    utils.monkey_patch()
+    server = service.Service.create(binary='nova-vsa')
+    service.serve(server)
+    service.wait()

nova/flags.py

@@ -292,6 +292,7 @@ DEFINE_string('ajax_console_proxy_url',
                in the form "http://127.0.0.1:8000"')
 DEFINE_string('ajax_console_proxy_port',
                8000, 'port that ajax_console_proxy binds')
+DEFINE_string('vsa_topic', 'vsa', 'the topic that nova-vsa service listens on')
 DEFINE_bool('verbose', False, 'show debug output')
 DEFINE_boolean('fake_rabbit', False, 'use a fake rabbit')
 DEFINE_bool('fake_network', False,
@@ -375,6 +376,17 @@ DEFINE_string('volume_manager', 'nova.volume.manager.VolumeManager',
               'Manager for volume')
 DEFINE_string('scheduler_manager', 'nova.scheduler.manager.SchedulerManager',
               'Manager for scheduler')
+DEFINE_string('vsa_manager', 'nova.vsa.manager.VsaManager',
+              'Manager for vsa')
+DEFINE_string('vc_image_name', 'vc_image',
+              'the VC image ID (for a VC image that exists in DB Glance)')
+# VSA constants and enums
+DEFINE_string('default_vsa_instance_type', 'm1.small',
+              'default instance type for VSA instances')
+DEFINE_integer('max_vcs_in_vsa', 32,
+               'maxinum VCs in a VSA')
+DEFINE_integer('vsa_part_size_gb', 100,
+               'default partition size for shared capacity')
 
 # The service to use for image search and retrieval
 DEFINE_string('image_service', 'nova.image.glance.GlanceImageService',

nova/scheduler/abstract_scheduler.py

@@ -180,7 +180,6 @@ class AbstractScheduler(driver.Scheduler):
         for zone_id, result in child_results:
             if not result:
                 continue
-            assert isinstance(zone_id, int)
 
             for zone_rec in zones:
                 if zone_rec['id'] != zone_id:

nova/scheduler/vsa.py

@@ -0,0 +1,535 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright (c) 2011 Zadara Storage Inc.
+# Copyright (c) 2011 OpenStack LLC.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+"""
+VSA Simple Scheduler
+"""
+
+from nova import context
+from nova import db
+from nova import flags
+from nova import log as logging
+from nova import rpc
+from nova import utils
+from nova.scheduler import driver
+from nova.scheduler import simple
+from nova.vsa.api import VsaState
+from nova.volume import volume_types
+
+LOG = logging.getLogger('nova.scheduler.vsa')
+
+FLAGS = flags.FLAGS
+flags.DEFINE_integer('drive_type_approx_capacity_percent', 10,
+                    'The percentage range for capacity comparison')
+flags.DEFINE_integer('vsa_unique_hosts_per_alloc', 10,
+                    'The number of unique hosts per storage allocation')
+flags.DEFINE_boolean('vsa_select_unique_drives', True,
+                     'Allow selection of same host for multiple drives')
+
+
+def BYTES_TO_GB(bytes):
+    return bytes >> 30
+
+
+def GB_TO_BYTES(gb):
+    return gb << 30
+
+
+class VsaScheduler(simple.SimpleScheduler):
+    """Implements Scheduler for volume placement."""
+
+    def __init__(self, *args, **kwargs):
+        super(VsaScheduler, self).__init__(*args, **kwargs)
+        self._notify_all_volume_hosts("startup")
+
+    def _notify_all_volume_hosts(self, event):
+        rpc.fanout_cast(context.get_admin_context(),
+                 FLAGS.volume_topic,
+                 {"method": "notification",
+                  "args": {"event": event}})
+
+    def _qosgrp_match(self, drive_type, qos_values):
+
+        def _compare_names(str1, str2):
+            return str1.lower() == str2.lower()
+
+        def _compare_sizes_approxim(cap_capacity, size):
+            cap_capacity = BYTES_TO_GB(int(cap_capacity))
+            size = int(size)
+            size_perc = size * \
+                FLAGS.drive_type_approx_capacity_percent / 100
+
+            return cap_capacity >= size - size_perc and \
+                   cap_capacity <= size + size_perc
+
+        # Add more entries for additional comparisons
+        compare_list = [{'cap1': 'DriveType',
+                         'cap2': 'type',
+                         'cmp_func': _compare_names},
+                        {'cap1': 'DriveCapacity',
+                         'cap2': 'size',
+                         'cmp_func': _compare_sizes_approxim}]
+
+        for cap in compare_list:
+            if cap['cap1'] in qos_values.keys() and \
+               cap['cap2'] in drive_type.keys() and \
+               cap['cmp_func'] is not None and \
+               cap['cmp_func'](qos_values[cap['cap1']],
+                               drive_type[cap['cap2']]):
+                pass
+            else:
+                return False
+        return True
+
+    def _get_service_states(self):
+        return self.zone_manager.service_states
+
+    def _filter_hosts(self, topic, request_spec, host_list=None):
+
+        LOG.debug(_("_filter_hosts: %(request_spec)s"), locals())
+
+        drive_type = request_spec['drive_type']
+        LOG.debug(_("Filter hosts for drive type %s"), drive_type['name'])
+
+        if host_list is None:
+            host_list = self._get_service_states().iteritems()
+
+        filtered_hosts = []     # returns list of (hostname, capability_dict)
+        for host, host_dict in host_list:
+            for service_name, service_dict in host_dict.iteritems():
+                if service_name != topic:
+                    continue
+
+                gos_info = service_dict.get('drive_qos_info', {})
+                for qosgrp, qos_values in gos_info.iteritems():
+                    if self._qosgrp_match(drive_type, qos_values):
+                        if qos_values['AvailableCapacity'] > 0:
+                            filtered_hosts.append((host, gos_info))
+                        else:
+                            LOG.debug(_("Host %s has no free capacity. Skip"),
+                                        host)
+                        break
+
+        host_names = [item[0] for item in filtered_hosts]
+        LOG.debug(_("Filter hosts: %s"), host_names)
+        return filtered_hosts
+
+    def _allowed_to_use_host(self, host, selected_hosts, unique):
+        if unique == False or \
+           host not in [item[0] for item in selected_hosts]:
+            return True
+        else:
+            return False
+
+    def _add_hostcap_to_list(self, selected_hosts, host, cap):
+        if host not in [item[0] for item in selected_hosts]:
+            selected_hosts.append((host, cap))
+
+    def host_selection_algorithm(self, request_spec, all_hosts,
+                                selected_hosts, unique):
+        """Must override this method for VSA scheduler to work."""
+        raise NotImplementedError(_("Must implement host selection mechanism"))
+
+    def _select_hosts(self, request_spec, all_hosts, selected_hosts=None):
+
+        if selected_hosts is None:
+            selected_hosts = []
+
+        host = None
+        if len(selected_hosts) >= FLAGS.vsa_unique_hosts_per_alloc:
+            # try to select from already selected hosts only
+            LOG.debug(_("Maximum number of hosts selected (%d)"),
+                        len(selected_hosts))
+            unique = False
+            (host, qos_cap) = self.host_selection_algorithm(request_spec,
+                                                            selected_hosts,
+                                                            selected_hosts,
+                                                            unique)
+
+            LOG.debug(_("Selected excessive host %(host)s"), locals())
+        else:
+            unique = FLAGS.vsa_select_unique_drives
+
+        if host is None:
+            # if we've not tried yet (# of sel hosts < max) - unique=True
+            # or failed to select from selected_hosts - unique=False
+            # select from all hosts
+            (host, qos_cap) = self.host_selection_algorithm(request_spec,
+                                                            all_hosts,
+                                                            selected_hosts,
+                                                            unique)
+        if host is None:
+            raise driver.WillNotSchedule(_("No available hosts"))
+
+        return (host, qos_cap)
+
+    def _provision_volume(self, context, vol, vsa_id, availability_zone):
+
+        if availability_zone is None:
+            availability_zone = FLAGS.storage_availability_zone
+
+        now = utils.utcnow()
+        options = {
+            'size': vol['size'],
+            'user_id': context.user_id,
+            'project_id': context.project_id,
+            'snapshot_id': None,
+            'availability_zone': availability_zone,
+            'status': "creating",
+            'attach_status': "detached",
+            'display_name': vol['name'],
+            'display_description': vol['description'],
+            'volume_type_id': vol['volume_type_id'],
+            'metadata': dict(to_vsa_id=vsa_id),
+            'host': vol['host'],
+            'scheduled_at': now
+            }
+
+        size = vol['size']
+        host = vol['host']
+        name = vol['name']
+        LOG.debug(_("Provision volume %(name)s of size %(size)s GB on "\
+                    "host %(host)s"), locals())
+
+        volume_ref = db.volume_create(context, options)
+        rpc.cast(context,
+                 db.queue_get_for(context, "volume", vol['host']),
+                 {"method": "create_volume",
+                  "args": {"volume_id": volume_ref['id'],
+                           "snapshot_id": None}})
+
+    def _check_host_enforcement(self, context, availability_zone):
+        if (availability_zone
+            and ':' in availability_zone
+            and context.is_admin):
+            zone, _x, host = availability_zone.partition(':')
+            service = db.service_get_by_args(context.elevated(), host,
+                                             'nova-volume')
+            if not self.service_is_up(service):
+                raise driver.WillNotSchedule(_("Host %s not available") % host)
+
+            return host
+        else:
+            return None
+
+    def _assign_hosts_to_volumes(self, context, volume_params, forced_host):
+
+        prev_volume_type_id = None
+        request_spec = {}
+        selected_hosts = []
+
+        LOG.debug(_("volume_params %(volume_params)s") % locals())
+
+        i = 1
+        for vol in volume_params:
+            name = vol['name']
+            LOG.debug(_("%(i)d: Volume %(name)s"), locals())
+            i += 1
+
+            if forced_host:
+                vol['host'] = forced_host
+                vol['capabilities'] = None
+                continue
+
+            volume_type_id = vol['volume_type_id']
+            request_spec['size'] = vol['size']
+
+            if prev_volume_type_id is None or\
+               prev_volume_type_id != volume_type_id:
+                # generate list of hosts for this drive type
+
+                volume_type = volume_types.get_volume_type(context,
+                                                volume_type_id)
+                drive_type = {
+                    'name': volume_type['extra_specs'].get('drive_name'),
+                    'type': volume_type['extra_specs'].get('drive_type'),
+                    'size': int(volume_type['extra_specs'].get('drive_size')),
+                    'rpm': volume_type['extra_specs'].get('drive_rpm'),
+                    }
+                request_spec['drive_type'] = drive_type
+
+                all_hosts = self._filter_hosts("volume", request_spec)
+                prev_volume_type_id = volume_type_id
+
+            (host, qos_cap) = self._select_hosts(request_spec,
+                                    all_hosts, selected_hosts)
+            vol['host'] = host
+            vol['capabilities'] = qos_cap
+            self._consume_resource(qos_cap, vol['size'], -1)
+
+    def schedule_create_volumes(self, context, request_spec,
+                                availability_zone=None, *_args, **_kwargs):
+        """Picks hosts for hosting multiple volumes."""
+
+        num_volumes = request_spec.get('num_volumes')
+        LOG.debug(_("Attempting to spawn %(num_volumes)d volume(s)") %
+                locals())
+
+        vsa_id = request_spec.get('vsa_id')
+        volume_params = request_spec.get('volumes')
+
+        host = self._check_host_enforcement(context, availability_zone)
+
+        try:
+            self._print_capabilities_info()
+
+            self._assign_hosts_to_volumes(context, volume_params, host)
+
+            for vol in volume_params:
+                self._provision_volume(context, vol, vsa_id, availability_zone)
+        except:
+            if vsa_id:
+                db.vsa_update(context, vsa_id, dict(status=VsaState.FAILED))
+
+            for vol in volume_params:
+                if 'capabilities' in vol:
+                    self._consume_resource(vol['capabilities'],
+                                           vol['size'], 1)
+            raise
+
+        return None
+
+    def schedule_create_volume(self, context, volume_id, *_args, **_kwargs):
+        """Picks the best host based on requested drive type capability."""
+        volume_ref = db.volume_get(context, volume_id)
+
+        host = self._check_host_enforcement(context,
+                                            volume_ref['availability_zone'])
+        if host:
+            now = utils.utcnow()
+            db.volume_update(context, volume_id, {'host': host,
+                                                  'scheduled_at': now})
+            return host
+
+        volume_type_id = volume_ref['volume_type_id']
+        if volume_type_id:
+            volume_type = volume_types.get_volume_type(context, volume_type_id)
+
+        if volume_type_id is None or\
+           volume_types.is_vsa_volume(volume_type_id, volume_type):
+
+            LOG.debug(_("Non-VSA volume %d"), volume_ref['id'])
+            return super(VsaScheduler, self).schedule_create_volume(context,
+                        volume_id, *_args, **_kwargs)
+
+        self._print_capabilities_info()
+
+        drive_type = {
+            'name': volume_type['extra_specs'].get('drive_name'),
+            'type': volume_type['extra_specs'].get('drive_type'),
+            'size': int(volume_type['extra_specs'].get('drive_size')),
+            'rpm': volume_type['extra_specs'].get('drive_rpm'),
+            }
+
+        LOG.debug(_("Spawning volume %(volume_id)s with drive type "\
+                    "%(drive_type)s"), locals())
+
+        request_spec = {'size': volume_ref['size'],
+                        'drive_type': drive_type}
+        hosts = self._filter_hosts("volume", request_spec)
+
+        try:
+            (host, qos_cap) = self._select_hosts(request_spec, all_hosts=hosts)
+        except:
+            if volume_ref['to_vsa_id']:
+                db.vsa_update(context, volume_ref['to_vsa_id'],
+                                dict(status=VsaState.FAILED))
+            raise
+
+        if host:
+            now = utils.utcnow()
+            db.volume_update(context, volume_id, {'host': host,
+                                                  'scheduled_at': now})
+            self._consume_resource(qos_cap, volume_ref['size'], -1)
+            return host
+
+    def _consume_full_drive(self, qos_values, direction):
+        qos_values['FullDrive']['NumFreeDrives'] += direction
+        qos_values['FullDrive']['NumOccupiedDrives'] -= direction
+
+    def _consume_partition(self, qos_values, size, direction):
+
+        if qos_values['PartitionDrive']['PartitionSize'] != 0:
+            partition_size = qos_values['PartitionDrive']['PartitionSize']
+        else:
+            partition_size = size
+        part_per_drive = qos_values['DriveCapacity'] / partition_size
+
+        if direction == -1 and \
+           qos_values['PartitionDrive']['NumFreePartitions'] == 0:
+
+            self._consume_full_drive(qos_values, direction)
+            qos_values['PartitionDrive']['NumFreePartitions'] += \
+                                                        part_per_drive
+
+        qos_values['PartitionDrive']['NumFreePartitions'] += direction
+        qos_values['PartitionDrive']['NumOccupiedPartitions'] -= direction
+
+        if direction == 1 and \
+           qos_values['PartitionDrive']['NumFreePartitions'] >= \
+                                                        part_per_drive:
+
+            self._consume_full_drive(qos_values, direction)
+            qos_values['PartitionDrive']['NumFreePartitions'] -= \
+                                                        part_per_drive
+
+    def _consume_resource(self, qos_values, size, direction):
+        if qos_values is None:
+            LOG.debug(_("No capability selected for volume of size %(size)s"),
+                        locals())
+            return
+
+        if size == 0:   # full drive match
+            qos_values['AvailableCapacity'] += direction * \
+                                        qos_values['DriveCapacity']
+            self._consume_full_drive(qos_values, direction)
+        else:
+            qos_values['AvailableCapacity'] += direction * GB_TO_BYTES(size)
+            self._consume_partition(qos_values, GB_TO_BYTES(size), direction)
+        return
+
+    def _print_capabilities_info(self):
+        host_list = self._get_service_states().iteritems()
+        for host, host_dict in host_list:
+            for service_name, service_dict in host_dict.iteritems():
+                if service_name != "volume":
+                    continue
+
+                LOG.info(_("Host %s:"), host)
+
+                gos_info = service_dict.get('drive_qos_info', {})
+                for qosgrp, qos_values in gos_info.iteritems():
+                    total = qos_values['TotalDrives']
+                    used = qos_values['FullDrive']['NumOccupiedDrives']
+                    free = qos_values['FullDrive']['NumFreeDrives']
+                    avail = BYTES_TO_GB(qos_values['AvailableCapacity'])
+
+                    LOG.info(_("\tDrive %(qosgrp)-25s: total %(total)2s, "\
+                               "used %(used)2s, free %(free)2s. Available "\
+                               "capacity %(avail)-5s"), locals())
+
+
+class VsaSchedulerLeastUsedHost(VsaScheduler):
+    """
+    Implements VSA scheduler to select the host with least used capacity
+    of particular type.
+    """
+
+    def __init__(self, *args, **kwargs):
+        super(VsaSchedulerLeastUsedHost, self).__init__(*args, **kwargs)
+
+    def host_selection_algorithm(self, request_spec, all_hosts,
+                                selected_hosts, unique):
+        size = request_spec['size']
+        drive_type = request_spec['drive_type']
+        best_host = None
+        best_qoscap = None
+        best_cap = None
+        min_used = 0
+
+        for (host, capabilities) in all_hosts:
+
+            has_enough_capacity = False
+            used_capacity = 0
+            for qosgrp, qos_values in capabilities.iteritems():
+
+                used_capacity = used_capacity + qos_values['TotalCapacity'] \
+                                - qos_values['AvailableCapacity']
+
+                if self._qosgrp_match(drive_type, qos_values):
+                    # we found required qosgroup
+
+                    if size == 0:   # full drive match
+                        if qos_values['FullDrive']['NumFreeDrives'] > 0:
+                            has_enough_capacity = True
+                            matched_qos = qos_values
+                        else:
+                            break
+                    else:
+                        if qos_values['AvailableCapacity'] >= size and \
+                           (qos_values['PartitionDrive'][
+                                            'NumFreePartitions'] > 0 or \
+                            qos_values['FullDrive']['NumFreeDrives'] > 0):
+                            has_enough_capacity = True
+                            matched_qos = qos_values
+                        else:
+                            break
+
+            if has_enough_capacity and \
+               self._allowed_to_use_host(host,
+                                         selected_hosts,
+                                         unique) and \
+               (best_host is None or used_capacity < min_used):
+
+                min_used = used_capacity
+                best_host = host
+                best_qoscap = matched_qos
+                best_cap = capabilities
+
+        if best_host:
+            self._add_hostcap_to_list(selected_hosts, best_host, best_cap)
+            min_used = BYTES_TO_GB(min_used)
+            LOG.debug(_("\t LeastUsedHost: Best host: %(best_host)s. "\
+                        "(used capacity %(min_used)s)"), locals())
+        return (best_host, best_qoscap)
+
+
+class VsaSchedulerMostAvailCapacity(VsaScheduler):
+    """
+    Implements VSA scheduler to select the host with most available capacity
+    of one particular type.
+    """
+
+    def __init__(self, *args, **kwargs):
+        super(VsaSchedulerMostAvailCapacity, self).__init__(*args, **kwargs)
+
+    def host_selection_algorithm(self, request_spec, all_hosts,
+                                selected_hosts, unique):
+        size = request_spec['size']
+        drive_type = request_spec['drive_type']
+        best_host = None
+        best_qoscap = None
+        best_cap = None
+        max_avail = 0
+
+        for (host, capabilities) in all_hosts:
+            for qosgrp, qos_values in capabilities.iteritems():
+                if self._qosgrp_match(drive_type, qos_values):
+                    # we found required qosgroup
+
+                    if size == 0:   # full drive match
+                        available = qos_values['FullDrive']['NumFreeDrives']
+                    else:
+                        available = qos_values['AvailableCapacity']
+
+                    if available > max_avail and \
+                       self._allowed_to_use_host(host,
+                                                 selected_hosts,
+                                                 unique):
+                        max_avail = available
+                        best_host = host
+                        best_qoscap = qos_values
+                        best_cap = capabilities
+                    break   # go to the next host
+
+        if best_host:
+            self._add_hostcap_to_list(selected_hosts, best_host, best_cap)
+            type_str = "drives" if size == 0 else "bytes"
+            LOG.debug(_("\t MostAvailCap: Best host: %(best_host)s. "\
+                        "(available %(max_avail)s %(type_str)s)"), locals())
+
+        return (best_host, best_qoscap)

nova/tests/scheduler/test_vsa_scheduler.py

@@ -0,0 +1,641 @@
+# Copyright 2011 OpenStack LLC.
+# All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+import stubout
+
+import nova
+
+from nova import context
+from nova import db
+from nova import exception
+from nova import flags
+from nova import log as logging
+from nova import test
+from nova import utils
+from nova.volume import volume_types
+
+from nova.scheduler import vsa as vsa_sched
+from nova.scheduler import driver
+
+FLAGS = flags.FLAGS
+LOG = logging.getLogger('nova.tests.scheduler.vsa')
+
+scheduled_volumes = []
+scheduled_volume = {}
+global_volume = {}
+
+
+class FakeVsaLeastUsedScheduler(
+                vsa_sched.VsaSchedulerLeastUsedHost):
+    # No need to stub anything at the moment
+    pass
+
+
+class FakeVsaMostAvailCapacityScheduler(
+                vsa_sched.VsaSchedulerMostAvailCapacity):
+    # No need to stub anything at the moment
+    pass
+
+
+class VsaSchedulerTestCase(test.TestCase):
+
+    def _get_vol_creation_request(self, num_vols, drive_ix, size=0):
+        volume_params = []
+        for i in range(num_vols):
+
+            name = 'name_' + str(i)
+            try:
+                volume_types.create(self.context, name,
+                            extra_specs={'type': 'vsa_drive',
+                                         'drive_name': name,
+                                         'drive_type': 'type_' + str(drive_ix),
+                                         'drive_size': 1 + 100 * (drive_ix)})
+                self.created_types_lst.append(name)
+            except exception.ApiError:
+                # type is already created
+                pass
+
+            volume_type = volume_types.get_volume_type_by_name(self.context,
+                                                                name)
+            volume = {'size': size,
+                      'snapshot_id': None,
+                      'name': 'vol_' + str(i),
+                      'description': None,
+                      'volume_type_id': volume_type['id']}
+            volume_params.append(volume)
+
+        return {'num_volumes': len(volume_params),
+                'vsa_id': 123,
+                'volumes': volume_params}
+
+    def _generate_default_service_states(self):
+        service_states = {}
+        for i in range(self.host_num):
+            host = {}
+            hostname = 'host_' + str(i)
+            if hostname in self.exclude_host_list:
+                continue
+
+            host['volume'] = {'timestamp': utils.utcnow(),
+                              'drive_qos_info': {}}
+
+            for j in range(self.drive_type_start_ix,
+                           self.drive_type_start_ix + self.drive_type_num):
+                dtype = {}
+                dtype['Name'] = 'name_' + str(j)
+                dtype['DriveType'] = 'type_' + str(j)
+                dtype['TotalDrives'] = 2 * (self.init_num_drives + i)
+                dtype['DriveCapacity'] = vsa_sched.GB_TO_BYTES(1 + 100 * j)
+                dtype['TotalCapacity'] = dtype['TotalDrives'] * \
+                                            dtype['DriveCapacity']
+                dtype['AvailableCapacity'] = (dtype['TotalDrives'] - i) * \
+                                            dtype['DriveCapacity']
+                dtype['DriveRpm'] = 7200
+                dtype['DifCapable'] = 0
+                dtype['SedCapable'] = 0
+                dtype['PartitionDrive'] = {
+                            'PartitionSize': 0,
+                            'NumOccupiedPartitions': 0,
+                            'NumFreePartitions': 0}
+                dtype['FullDrive'] = {
+                            'NumFreeDrives': dtype['TotalDrives'] - i,
+                            'NumOccupiedDrives': i}
+                host['volume']['drive_qos_info'][dtype['Name']] = dtype
+
+            service_states[hostname] = host
+
+        return service_states
+
+    def _print_service_states(self):
+        for host, host_val in self.service_states.iteritems():
+            LOG.info(_("Host %s"), host)
+            total_used = 0
+            total_available = 0
+            qos = host_val['volume']['drive_qos_info']
+
+            for k, d in qos.iteritems():
+                LOG.info("\t%s: type %s: drives (used %2d, total %2d) "\
+                    "size %3d, total %4d, used %4d, avail %d",
+                    k, d['DriveType'],
+                    d['FullDrive']['NumOccupiedDrives'], d['TotalDrives'],
+                    vsa_sched.BYTES_TO_GB(d['DriveCapacity']),
+                    vsa_sched.BYTES_TO_GB(d['TotalCapacity']),
+                    vsa_sched.BYTES_TO_GB(d['TotalCapacity'] - \
+                        d['AvailableCapacity']),
+                    vsa_sched.BYTES_TO_GB(d['AvailableCapacity']))
+
+                total_used += vsa_sched.BYTES_TO_GB(d['TotalCapacity'] - \
+                                    d['AvailableCapacity'])
+                total_available += vsa_sched.BYTES_TO_GB(
+                                        d['AvailableCapacity'])
+            LOG.info("Host %s: used %d, avail %d",
+                     host, total_used, total_available)
+
+    def _set_service_states(self, host_num,
+                            drive_type_start_ix, drive_type_num,
+                            init_num_drives=10,
+                            exclude_host_list=[]):
+        self.host_num = host_num
+        self.drive_type_start_ix = drive_type_start_ix
+        self.drive_type_num = drive_type_num
+        self.exclude_host_list = exclude_host_list
+        self.init_num_drives = init_num_drives
+        self.service_states = self._generate_default_service_states()
+
+    def _get_service_states(self):
+        return self.service_states
+
+    def _fake_get_service_states(self):
+        return self._get_service_states()
+
+    def _fake_provision_volume(self, context, vol, vsa_id, availability_zone):
+        global scheduled_volumes
+        scheduled_volumes.append(dict(vol=vol,
+                                      vsa_id=vsa_id,
+                                      az=availability_zone))
+        name = vol['name']
+        host = vol['host']
+        LOG.debug(_("Test: provision vol %(name)s on host %(host)s"),
+                    locals())
+        LOG.debug(_("\t vol=%(vol)s"), locals())
+        pass
+
+    def _fake_vsa_update(self, context, vsa_id, values):
+        LOG.debug(_("Test: VSA update request: vsa_id=%(vsa_id)s "\
+                    "values=%(values)s"), locals())
+        pass
+
+    def _fake_volume_create(self, context, options):
+        LOG.debug(_("Test: Volume create: %s"), options)
+        options['id'] = 123
+        global global_volume
+        global_volume = options
+        return options
+
+    def _fake_volume_get(self, context, volume_id):
+        LOG.debug(_("Test: Volume get request: id=%(volume_id)s"), locals())
+        global global_volume
+        global_volume['id'] = volume_id
+        global_volume['availability_zone'] = None
+        return global_volume
+
+    def _fake_volume_update(self, context, volume_id, values):
+        LOG.debug(_("Test: Volume update request: id=%(volume_id)s "\
+                    "values=%(values)s"), locals())
+        global scheduled_volume
+        scheduled_volume = {'id': volume_id, 'host': values['host']}
+        pass
+
+    def _fake_service_get_by_args(self, context, host, binary):
+        return "service"
+
+    def _fake_service_is_up_True(self, service):
+        return True
+
+    def _fake_service_is_up_False(self, service):
+        return False
+
+    def setUp(self, sched_class=None):
+        super(VsaSchedulerTestCase, self).setUp()
+        self.stubs = stubout.StubOutForTesting()
+        self.context = context.get_admin_context()
+
+        if sched_class is None:
+            self.sched = FakeVsaLeastUsedScheduler()
+        else:
+            self.sched = sched_class
+
+        self.host_num = 10
+        self.drive_type_num = 5
+
+        self.stubs.Set(self.sched,
+                        '_get_service_states', self._fake_get_service_states)
+        self.stubs.Set(self.sched,
+                        '_provision_volume', self._fake_provision_volume)
+        self.stubs.Set(nova.db, 'vsa_update', self._fake_vsa_update)
+
+        self.stubs.Set(nova.db, 'volume_get', self._fake_volume_get)
+        self.stubs.Set(nova.db, 'volume_update', self._fake_volume_update)
+
+        self.created_types_lst = []
+
+    def tearDown(self):
+        for name in self.created_types_lst:
+            volume_types.purge(self.context, name)
+
+        self.stubs.UnsetAll()
+        super(VsaSchedulerTestCase, self).tearDown()
+
+    def test_vsa_sched_create_volumes_simple(self):
+        global scheduled_volumes
+        scheduled_volumes = []
+        self._set_service_states(host_num=10,
+                                 drive_type_start_ix=0,
+                                 drive_type_num=5,
+                                 init_num_drives=10,
+                                 exclude_host_list=['host_1', 'host_3'])
+        prev = self._generate_default_service_states()
+        request_spec = self._get_vol_creation_request(num_vols=3, drive_ix=2)
+
+        self.sched.schedule_create_volumes(self.context,
+                                           request_spec,
+                                           availability_zone=None)
+
+        self.assertEqual(len(scheduled_volumes), 3)
+        self.assertEqual(scheduled_volumes[0]['vol']['host'], 'host_0')
+        self.assertEqual(scheduled_volumes[1]['vol']['host'], 'host_2')
+        self.assertEqual(scheduled_volumes[2]['vol']['host'], 'host_4')
+
+        cur = self._get_service_states()
+        for host in ['host_0', 'host_2', 'host_4']:
+            cur_dtype = cur[host]['volume']['drive_qos_info']['name_2']
+            prev_dtype = prev[host]['volume']['drive_qos_info']['name_2']
+            self.assertEqual(cur_dtype['DriveType'], prev_dtype['DriveType'])
+            self.assertEqual(cur_dtype['FullDrive']['NumFreeDrives'],
+                             prev_dtype['FullDrive']['NumFreeDrives'] - 1)
+            self.assertEqual(cur_dtype['FullDrive']['NumOccupiedDrives'],
+                             prev_dtype['FullDrive']['NumOccupiedDrives'] + 1)
+
+    def test_vsa_sched_no_drive_type(self):
+        self._set_service_states(host_num=10,
+                                 drive_type_start_ix=0,
+                                 drive_type_num=5,
+                                 init_num_drives=1)
+        request_spec = self._get_vol_creation_request(num_vols=1, drive_ix=6)
+        self.assertRaises(driver.WillNotSchedule,
+                          self.sched.schedule_create_volumes,
+                                self.context,
+                                request_spec,
+                                availability_zone=None)
+
+    def test_vsa_sched_no_enough_drives(self):
+        global scheduled_volumes
+        scheduled_volumes = []
+
+        self._set_service_states(host_num=3,
+                                 drive_type_start_ix=0,
+                                 drive_type_num=1,
+                                 init_num_drives=0)
+        prev = self._generate_default_service_states()
+        request_spec = self._get_vol_creation_request(num_vols=3, drive_ix=0)
+
+        self.assertRaises(driver.WillNotSchedule,
+                          self.sched.schedule_create_volumes,
+                                self.context,
+                                request_spec,
+                                availability_zone=None)
+
+        # check that everything was returned back
+        cur = self._get_service_states()
+        for k, v in prev.iteritems():
+            self.assertEqual(prev[k]['volume']['drive_qos_info'],
+                              cur[k]['volume']['drive_qos_info'])
+
+    def test_vsa_sched_wrong_topic(self):
+        self._set_service_states(host_num=1,
+                                 drive_type_start_ix=0,
+                                 drive_type_num=5,
+                                 init_num_drives=1)
+        states = self._get_service_states()
+        new_states = {}
+        new_states['host_0'] = {'compute': states['host_0']['volume']}
+        self.service_states = new_states
+        request_spec = self._get_vol_creation_request(num_vols=1, drive_ix=0)
+
+        self.assertRaises(driver.WillNotSchedule,
+                          self.sched.schedule_create_volumes,
+                                self.context,
+                                request_spec,
+                                availability_zone=None)
+
+    def test_vsa_sched_provision_volume(self):
+        global global_volume
+        global_volume = {}
+        self._set_service_states(host_num=1,
+                                 drive_type_start_ix=0,
+                                 drive_type_num=1,
+                                 init_num_drives=1)
+        request_spec = self._get_vol_creation_request(num_vols=1, drive_ix=0)
+
+        self.stubs.UnsetAll()
+        self.stubs.Set(self.sched,
+                        '_get_service_states', self._fake_get_service_states)
+        self.stubs.Set(nova.db, 'volume_create', self._fake_volume_create)
+
+        self.sched.schedule_create_volumes(self.context,
+                                           request_spec,
+                                           availability_zone=None)
+
+        self.assertEqual(request_spec['volumes'][0]['name'],
+                         global_volume['display_name'])
+
+    def test_vsa_sched_no_free_drives(self):
+        self._set_service_states(host_num=1,
+                                 drive_type_start_ix=0,
+                                 drive_type_num=1,
+                                 init_num_drives=1)
+        request_spec = self._get_vol_creation_request(num_vols=1, drive_ix=0)
+
+        self.sched.schedule_create_volumes(self.context,
+                                           request_spec,
+                                           availability_zone=None)
+
+        cur = self._get_service_states()
+        cur_dtype = cur['host_0']['volume']['drive_qos_info']['name_0']
+        self.assertEqual(cur_dtype['FullDrive']['NumFreeDrives'], 1)
+
+        new_request = self._get_vol_creation_request(num_vols=1, drive_ix=0)
+
+        self.sched.schedule_create_volumes(self.context,
+                                           request_spec,
+                                           availability_zone=None)
+        self._print_service_states()
+
+        self.assertRaises(driver.WillNotSchedule,
+                          self.sched.schedule_create_volumes,
+                                self.context,
+                                new_request,
+                                availability_zone=None)
+
+    def test_vsa_sched_forced_host(self):
+        global scheduled_volumes
+        scheduled_volumes = []
+
+        self._set_service_states(host_num=10,
+                                 drive_type_start_ix=0,
+                                 drive_type_num=5,
+                                 init_num_drives=10)
+
+        request_spec = self._get_vol_creation_request(num_vols=3, drive_ix=2)
+
+        self.assertRaises(exception.HostBinaryNotFound,
+                          self.sched.schedule_create_volumes,
+                                self.context,
+                                request_spec,
+                                availability_zone="nova:host_5")
+
+        self.stubs.Set(nova.db,
+                        'service_get_by_args', self._fake_service_get_by_args)
+        self.stubs.Set(self.sched,
+                        'service_is_up', self._fake_service_is_up_False)
+
+        self.assertRaises(driver.WillNotSchedule,
+                          self.sched.schedule_create_volumes,
+                                self.context,
+                                request_spec,
+                                availability_zone="nova:host_5")
+
+        self.stubs.Set(self.sched,
+                        'service_is_up', self._fake_service_is_up_True)
+
+        self.sched.schedule_create_volumes(self.context,
+                                           request_spec,
+                                           availability_zone="nova:host_5")
+
+        self.assertEqual(len(scheduled_volumes), 3)
+        self.assertEqual(scheduled_volumes[0]['vol']['host'], 'host_5')
+        self.assertEqual(scheduled_volumes[1]['vol']['host'], 'host_5')
+        self.assertEqual(scheduled_volumes[2]['vol']['host'], 'host_5')
+
+    def test_vsa_sched_create_volumes_partition(self):
+        global scheduled_volumes
+        scheduled_volumes = []
+        self._set_service_states(host_num=5,
+                                 drive_type_start_ix=0,
+                                 drive_type_num=5,
+                                 init_num_drives=1,
+                                 exclude_host_list=['host_0', 'host_2'])
+        prev = self._generate_default_service_states()
+        request_spec = self._get_vol_creation_request(num_vols=3,
+                                                      drive_ix=3,
+                                                      size=50)
+        self.sched.schedule_create_volumes(self.context,
+                                           request_spec,
+                                           availability_zone=None)
+
+        self.assertEqual(len(scheduled_volumes), 3)
+        self.assertEqual(scheduled_volumes[0]['vol']['host'], 'host_1')
+        self.assertEqual(scheduled_volumes[1]['vol']['host'], 'host_3')
+        self.assertEqual(scheduled_volumes[2]['vol']['host'], 'host_4')
+
+        cur = self._get_service_states()
+        for host in ['host_1', 'host_3', 'host_4']:
+            cur_dtype = cur[host]['volume']['drive_qos_info']['name_3']
+            prev_dtype = prev[host]['volume']['drive_qos_info']['name_3']
+
+            self.assertEqual(cur_dtype['DriveType'], prev_dtype['DriveType'])
+            self.assertEqual(cur_dtype['FullDrive']['NumFreeDrives'],
+                             prev_dtype['FullDrive']['NumFreeDrives'] - 1)
+            self.assertEqual(cur_dtype['FullDrive']['NumOccupiedDrives'],
+                             prev_dtype['FullDrive']['NumOccupiedDrives'] + 1)
+
+            self.assertEqual(prev_dtype['PartitionDrive']
+                                            ['NumOccupiedPartitions'], 0)
+            self.assertEqual(cur_dtype['PartitionDrive']
+                                            ['NumOccupiedPartitions'], 1)
+            self.assertEqual(cur_dtype['PartitionDrive']
+                                            ['NumFreePartitions'], 5)
+
+            self.assertEqual(prev_dtype['PartitionDrive']
+                                            ['NumFreePartitions'], 0)
+            self.assertEqual(prev_dtype['PartitionDrive']
+                                            ['PartitionSize'], 0)
+
+    def test_vsa_sched_create_single_volume_az(self):
+        global scheduled_volume
+        scheduled_volume = {}
+
+        def _fake_volume_get_az(context, volume_id):
+            LOG.debug(_("Test: Volume get: id=%(volume_id)s"), locals())
+            return {'id': volume_id, 'availability_zone': 'nova:host_3'}
+
+        self.stubs.Set(nova.db, 'volume_get', _fake_volume_get_az)
+        self.stubs.Set(nova.db,
+                        'service_get_by_args', self._fake_service_get_by_args)
+        self.stubs.Set(self.sched,
+                        'service_is_up', self._fake_service_is_up_True)
+
+        host = self.sched.schedule_create_volume(self.context,
+                                                 123, availability_zone=None)
+
+        self.assertEqual(host, 'host_3')
+        self.assertEqual(scheduled_volume['id'], 123)
+        self.assertEqual(scheduled_volume['host'], 'host_3')
+
+    def test_vsa_sched_create_single_non_vsa_volume(self):
+        global scheduled_volume
+        scheduled_volume = {}
+
+        global global_volume
+        global_volume = {}
+        global_volume['volume_type_id'] = None
+
+        self.assertRaises(driver.NoValidHost,
+                          self.sched.schedule_create_volume,
+                                self.context,
+                                123,
+                                availability_zone=None)
+
+    def test_vsa_sched_create_single_volume(self):
+        global scheduled_volume
+        scheduled_volume = {}
+        self._set_service_states(host_num=10,
+                                 drive_type_start_ix=0,
+                                 drive_type_num=5,
+                                 init_num_drives=10,
+                                 exclude_host_list=['host_0', 'host_1'])
+        prev = self._generate_default_service_states()
+
+        global global_volume
+        global_volume = {}
+
+        drive_ix = 2
+        name = 'name_' + str(drive_ix)
+        volume_types.create(self.context, name,
+                    extra_specs={'type': 'vsa_drive',
+                                 'drive_name': name,
+                                 'drive_type': 'type_' + str(drive_ix),
+                                 'drive_size': 1 + 100 * (drive_ix)})
+        self.created_types_lst.append(name)
+        volume_type = volume_types.get_volume_type_by_name(self.context, name)
+
+        global_volume['volume_type_id'] = volume_type['id']
+        global_volume['size'] = 0
+
+        host = self.sched.schedule_create_volume(self.context,
+                                                 123, availability_zone=None)
+
+        self.assertEqual(host, 'host_2')
+        self.assertEqual(scheduled_volume['id'], 123)
+        self.assertEqual(scheduled_volume['host'], 'host_2')
+
+
+class VsaSchedulerTestCaseMostAvail(VsaSchedulerTestCase):
+
+    def setUp(self):
+        super(VsaSchedulerTestCaseMostAvail, self).setUp(
+                    FakeVsaMostAvailCapacityScheduler())
+
+    def tearDown(self):
+        self.stubs.UnsetAll()
+        super(VsaSchedulerTestCaseMostAvail, self).tearDown()
+
+    def test_vsa_sched_create_single_volume(self):
+        global scheduled_volume
+        scheduled_volume = {}
+        self._set_service_states(host_num=10,
+                                 drive_type_start_ix=0,
+                                 drive_type_num=5,
+                                 init_num_drives=10,
+                                 exclude_host_list=['host_0', 'host_1'])
+        prev = self._generate_default_service_states()
+
+        global global_volume
+        global_volume = {}
+
+        drive_ix = 2
+        name = 'name_' + str(drive_ix)
+        volume_types.create(self.context, name,
+                    extra_specs={'type': 'vsa_drive',
+                                 'drive_name': name,
+                                 'drive_type': 'type_' + str(drive_ix),
+                                 'drive_size': 1 + 100 * (drive_ix)})
+        self.created_types_lst.append(name)
+        volume_type = volume_types.get_volume_type_by_name(self.context, name)
+
+        global_volume['volume_type_id'] = volume_type['id']
+        global_volume['size'] = 0
+
+        host = self.sched.schedule_create_volume(self.context,
+                                                 123, availability_zone=None)
+
+        self.assertEqual(host, 'host_9')
+        self.assertEqual(scheduled_volume['id'], 123)
+        self.assertEqual(scheduled_volume['host'], 'host_9')
+
+    def test_vsa_sched_create_volumes_simple(self):
+        global scheduled_volumes
+        scheduled_volumes = []
+        self._set_service_states(host_num=10,
+                                 drive_type_start_ix=0,
+                                 drive_type_num=5,
+                                 init_num_drives=10,
+                                 exclude_host_list=['host_1', 'host_3'])
+        prev = self._generate_default_service_states()
+        request_spec = self._get_vol_creation_request(num_vols=3, drive_ix=2)
+
+        self._print_service_states()
+
+        self.sched.schedule_create_volumes(self.context,
+                                           request_spec,
+                                           availability_zone=None)
+
+        self.assertEqual(len(scheduled_volumes), 3)
+        self.assertEqual(scheduled_volumes[0]['vol']['host'], 'host_9')
+        self.assertEqual(scheduled_volumes[1]['vol']['host'], 'host_8')
+        self.assertEqual(scheduled_volumes[2]['vol']['host'], 'host_7')
+
+        cur = self._get_service_states()
+        for host in ['host_9', 'host_8', 'host_7']:
+            cur_dtype = cur[host]['volume']['drive_qos_info']['name_2']
+            prev_dtype = prev[host]['volume']['drive_qos_info']['name_2']
+            self.assertEqual(cur_dtype['DriveType'], prev_dtype['DriveType'])
+            self.assertEqual(cur_dtype['FullDrive']['NumFreeDrives'],
+                             prev_dtype['FullDrive']['NumFreeDrives'] - 1)
+            self.assertEqual(cur_dtype['FullDrive']['NumOccupiedDrives'],
+                             prev_dtype['FullDrive']['NumOccupiedDrives'] + 1)
+
+    def test_vsa_sched_create_volumes_partition(self):
+        global scheduled_volumes
+        scheduled_volumes = []
+        self._set_service_states(host_num=5,
+                                 drive_type_start_ix=0,
+                                 drive_type_num=5,
+                                 init_num_drives=1,
+                                 exclude_host_list=['host_0', 'host_2'])
+        prev = self._generate_default_service_states()
+        request_spec = self._get_vol_creation_request(num_vols=3,
+                                                      drive_ix=3,
+                                                      size=50)
+        self.sched.schedule_create_volumes(self.context,
+                                           request_spec,
+                                           availability_zone=None)
+
+        self.assertEqual(len(scheduled_volumes), 3)
+        self.assertEqual(scheduled_volumes[0]['vol']['host'], 'host_4')
+        self.assertEqual(scheduled_volumes[1]['vol']['host'], 'host_3')
+        self.assertEqual(scheduled_volumes[2]['vol']['host'], 'host_1')
+
+        cur = self._get_service_states()
+        for host in ['host_1', 'host_3', 'host_4']:
+            cur_dtype = cur[host]['volume']['drive_qos_info']['name_3']
+            prev_dtype = prev[host]['volume']['drive_qos_info']['name_3']
+
+            self.assertEqual(cur_dtype['DriveType'], prev_dtype['DriveType'])
+            self.assertEqual(cur_dtype['FullDrive']['NumFreeDrives'],
+                             prev_dtype['FullDrive']['NumFreeDrives'] - 1)
+            self.assertEqual(cur_dtype['FullDrive']['NumOccupiedDrives'],
+                             prev_dtype['FullDrive']['NumOccupiedDrives'] + 1)
+
+            self.assertEqual(prev_dtype['PartitionDrive']
+                                            ['NumOccupiedPartitions'], 0)
+            self.assertEqual(cur_dtype['PartitionDrive']
+                                            ['NumOccupiedPartitions'], 1)
+            self.assertEqual(cur_dtype['PartitionDrive']
+                                            ['NumFreePartitions'], 5)
+            self.assertEqual(prev_dtype['PartitionDrive']
+                                            ['NumFreePartitions'], 0)
+            self.assertEqual(prev_dtype['PartitionDrive']
+                                            ['PartitionSize'], 0)

nova/tests/test_vsa.py

@@ -0,0 +1,182 @@
+# Copyright 2011 OpenStack LLC.
+# All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+import base64
+import stubout
+
+from xml.etree import ElementTree
+from xml.etree.ElementTree import Element, SubElement
+
+from nova import context
+from nova import db
+from nova import exception
+from nova import flags
+from nova import log as logging
+from nova import test
+from nova import vsa
+from nova import volume
+from nova.volume import volume_types
+from nova.vsa import utils as vsa_utils
+
+import nova.image.fake
+
+FLAGS = flags.FLAGS
+LOG = logging.getLogger('nova.tests.vsa')
+
+
+class VsaTestCase(test.TestCase):
+
+    def setUp(self):
+        super(VsaTestCase, self).setUp()
+        self.stubs = stubout.StubOutForTesting()
+        self.vsa_api = vsa.API()
+        self.volume_api = volume.API()
+
+        FLAGS.quota_volumes = 100
+        FLAGS.quota_gigabytes = 10000
+
+        self.context = context.get_admin_context()
+
+        volume_types.create(self.context,
+                            'SATA_500_7200',
+                            extra_specs={'type': 'vsa_drive',
+                                         'drive_name': 'SATA_500_7200',
+                                         'drive_type': 'SATA',
+                                         'drive_size': '500',
+                                         'drive_rpm': '7200'})
+
+        def fake_show_by_name(meh, context, name):
+            if name == 'wrong_image_name':
+                LOG.debug(_("Test: Emulate wrong VSA name. Raise"))
+                raise exception.ImageNotFound
+            return {'id': 1, 'properties': {'kernel_id': 1, 'ramdisk_id': 1}}
+
+        self.stubs.Set(nova.image.fake._FakeImageService,
+                        'show_by_name',
+                        fake_show_by_name)
+
+    def tearDown(self):
+        self.stubs.UnsetAll()
+        super(VsaTestCase, self).tearDown()
+
+    def test_vsa_create_delete_defaults(self):
+        param = {'display_name': 'VSA name test'}
+        vsa_ref = self.vsa_api.create(self.context, **param)
+        self.assertEqual(vsa_ref['display_name'], param['display_name'])
+        self.vsa_api.delete(self.context, vsa_ref['id'])
+
+    def test_vsa_create_delete_check_in_db(self):
+        vsa_list1 = self.vsa_api.get_all(self.context)
+        vsa_ref = self.vsa_api.create(self.context)
+        vsa_list2 = self.vsa_api.get_all(self.context)
+        self.assertEqual(len(vsa_list2), len(vsa_list1) + 1)
+
+        self.vsa_api.delete(self.context, vsa_ref['id'])
+        vsa_list3 = self.vsa_api.get_all(self.context)
+        self.assertEqual(len(vsa_list3), len(vsa_list2) - 1)
+
+    def test_vsa_create_delete_high_vc_count(self):
+        param = {'vc_count': FLAGS.max_vcs_in_vsa + 1}
+        vsa_ref = self.vsa_api.create(self.context, **param)
+        self.assertEqual(vsa_ref['vc_count'], FLAGS.max_vcs_in_vsa)
+        self.vsa_api.delete(self.context, vsa_ref['id'])
+
+    def test_vsa_create_wrong_image_name(self):
+        param = {'image_name': 'wrong_image_name'}
+        self.assertRaises(exception.ApiError,
+                          self.vsa_api.create, self.context, **param)
+
+    def test_vsa_create_db_error(self):
+
+        def fake_vsa_create(context, options):
+            LOG.debug(_("Test: Emulate DB error. Raise"))
+            raise exception.Error
+
+        self.stubs.Set(nova.db.api, 'vsa_create', fake_vsa_create)
+        self.assertRaises(exception.ApiError,
+                          self.vsa_api.create, self.context)
+
+    def test_vsa_create_wrong_storage_params(self):
+        vsa_list1 = self.vsa_api.get_all(self.context)
+        param = {'storage': [{'stub': 1}]}
+        self.assertRaises(exception.ApiError,
+                          self.vsa_api.create, self.context, **param)
+        vsa_list2 = self.vsa_api.get_all(self.context)
+        self.assertEqual(len(vsa_list2), len(vsa_list1))
+
+        param = {'storage': [{'drive_name': 'wrong name'}]}
+        self.assertRaises(exception.ApiError,
+                          self.vsa_api.create, self.context, **param)
+
+    def test_vsa_create_with_storage(self, multi_vol_creation=True):
+        """Test creation of VSA with BE storage"""
+
+        FLAGS.vsa_multi_vol_creation = multi_vol_creation
+
+        param = {'storage': [{'drive_name': 'SATA_500_7200',
+                              'num_drives': 3}]}
+        vsa_ref = self.vsa_api.create(self.context, **param)
+        self.assertEqual(vsa_ref['vol_count'], 3)
+        self.vsa_api.delete(self.context, vsa_ref['id'])
+
+        param = {'storage': [{'drive_name': 'SATA_500_7200',
+                              'num_drives': 3}],
+                 'shared': True}
+        vsa_ref = self.vsa_api.create(self.context, **param)
+        self.assertEqual(vsa_ref['vol_count'], 15)
+        self.vsa_api.delete(self.context, vsa_ref['id'])
+
+    def test_vsa_create_with_storage_single_volumes(self):
+        self.test_vsa_create_with_storage(multi_vol_creation=False)
+
+    def test_vsa_update(self):
+        vsa_ref = self.vsa_api.create(self.context)
+
+        param = {'vc_count': FLAGS.max_vcs_in_vsa + 1}
+        vsa_ref = self.vsa_api.update(self.context, vsa_ref['id'], **param)
+        self.assertEqual(vsa_ref['vc_count'], FLAGS.max_vcs_in_vsa)
+
+        param = {'vc_count': 2}
+        vsa_ref = self.vsa_api.update(self.context, vsa_ref['id'], **param)
+        self.assertEqual(vsa_ref['vc_count'], 2)
+
+        self.vsa_api.delete(self.context, vsa_ref['id'])
+
+    def test_vsa_generate_user_data(self):
+
+        FLAGS.vsa_multi_vol_creation = False
+        param = {'display_name': 'VSA name test',
+                 'display_description': 'VSA desc test',
+                 'vc_count': 2,
+                 'storage': [{'drive_name': 'SATA_500_7200',
+                              'num_drives': 3}]}
+        vsa_ref = self.vsa_api.create(self.context, **param)
+        volumes = self.vsa_api.get_all_vsa_drives(self.context,
+                                                  vsa_ref['id'])
+
+        user_data = vsa_utils.generate_user_data(vsa_ref, volumes)
+        user_data = base64.b64decode(user_data)
+
+        LOG.debug(_("Test: user_data = %s"), user_data)
+
+        elem = ElementTree.fromstring(user_data)
+        self.assertEqual(elem.findtext('name'),
+                         param['display_name'])
+        self.assertEqual(elem.findtext('description'),
+                         param['display_description'])
+        self.assertEqual(elem.findtext('vc_count'),
+                         str(param['vc_count']))
+
+        self.vsa_api.delete(self.context, vsa_ref['id'])

nova/tests/test_vsa_volumes.py

@@ -0,0 +1,136 @@
+# Copyright 2011 OpenStack LLC.
+# All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+import stubout
+
+from nova import exception
+from nova import flags
+from nova import vsa
+from nova import volume
+from nova import db
+from nova import context
+from nova import test
+from nova import log as logging
+import nova.image.fake
+
+FLAGS = flags.FLAGS
+LOG = logging.getLogger('nova.tests.vsa.volumes')
+
+
+class VsaVolumesTestCase(test.TestCase):
+
+    def setUp(self):
+        super(VsaVolumesTestCase, self).setUp()
+        self.stubs = stubout.StubOutForTesting()
+        self.vsa_api = vsa.API()
+        self.volume_api = volume.API()
+        self.context = context.get_admin_context()
+
+        self.default_vol_type = self.vsa_api.get_vsa_volume_type(self.context)
+
+        def fake_show_by_name(meh, context, name):
+            return {'id': 1, 'properties': {'kernel_id': 1, 'ramdisk_id': 1}}
+
+        self.stubs.Set(nova.image.fake._FakeImageService,
+                        'show_by_name',
+                        fake_show_by_name)
+
+        param = {'display_name': 'VSA name test'}
+        vsa_ref = self.vsa_api.create(self.context, **param)
+        self.vsa_id = vsa_ref['id']
+
+    def tearDown(self):
+        if self.vsa_id:
+            self.vsa_api.delete(self.context, self.vsa_id)
+        self.stubs.UnsetAll()
+        super(VsaVolumesTestCase, self).tearDown()
+
+    def _default_volume_param(self):
+        return {
+            'size': 1,
+            'snapshot_id': None,
+            'name': 'Test volume name',
+            'description': 'Test volume desc name',
+            'volume_type': self.default_vol_type,
+            'metadata': {'from_vsa_id': self.vsa_id}
+            }
+
+    def _get_all_volumes_by_vsa(self):
+        return self.volume_api.get_all(self.context,
+                search_opts={'metadata': {"from_vsa_id": str(self.vsa_id)}})
+
+    def test_vsa_volume_create_delete(self):
+        """ Check if volume properly created and deleted. """
+        volume_param = self._default_volume_param()
+        volume_ref = self.volume_api.create(self.context, **volume_param)
+
+        self.assertEqual(volume_ref['display_name'],
+                         volume_param['name'])
+        self.assertEqual(volume_ref['display_description'],
+                         volume_param['description'])
+        self.assertEqual(volume_ref['size'],
+                         volume_param['size'])
+        self.assertEqual(volume_ref['status'],
+                         'creating')
+
+        vols2 = self._get_all_volumes_by_vsa()
+        self.assertEqual(1, len(vols2))
+        volume_ref = vols2[0]
+
+        self.assertEqual(volume_ref['display_name'],
+                         volume_param['name'])
+        self.assertEqual(volume_ref['display_description'],
+                         volume_param['description'])
+        self.assertEqual(volume_ref['size'],
+                         volume_param['size'])
+        self.assertEqual(volume_ref['status'],
+                         'creating')
+
+        self.volume_api.update(self.context,
+                    volume_ref['id'], {'status': 'available'})
+        self.volume_api.delete(self.context, volume_ref['id'])
+
+        vols3 = self._get_all_volumes_by_vsa()
+        self.assertEqual(1, len(vols2))
+        volume_ref = vols3[0]
+        self.assertEqual(volume_ref['status'],
+                         'deleting')
+
+    def test_vsa_volume_delete_nonavail_volume(self):
+        """ Check volume deleton in different states. """
+        volume_param = self._default_volume_param()
+        volume_ref = self.volume_api.create(self.context, **volume_param)
+
+        self.volume_api.update(self.context,
+                            volume_ref['id'], {'status': 'in-use'})
+        self.assertRaises(exception.ApiError,
+                            self.volume_api.delete,
+                            self.context, volume_ref['id'])
+
+    def test_vsa_volume_delete_vsa_with_volumes(self):
+        """ Check volume deleton in different states. """
+
+        vols1 = self._get_all_volumes_by_vsa()
+        for i in range(3):
+            volume_param = self._default_volume_param()
+            volume_ref = self.volume_api.create(self.context, **volume_param)
+
+        vols2 = self._get_all_volumes_by_vsa()
+        self.assertEqual(len(vols1) + 3, len(vols2))
+
+        self.vsa_api.delete(self.context, self.vsa_id)
+
+        vols3 = self._get_all_volumes_by_vsa()
+        self.assertEqual(len(vols1), len(vols3))

nova/vsa/__init__.py

@@ -0,0 +1,18 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright (c) 2011 Zadara Storage Inc.
+# Copyright (c) 2011 OpenStack LLC.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+from nova.vsa.api import API

nova/vsa/api.py

@@ -0,0 +1,411 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright (c) 2011 Zadara Storage Inc.
+# Copyright (c) 2011 OpenStack LLC.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+"""
+Handles all requests relating to Virtual Storage Arrays (VSAs).
+
+Experimental code. Requires special VSA image.
+For assistance and guidelines pls contact
+    Zadara Storage Inc & Openstack community
+"""
+
+import sys
+
+from nova import compute
+from nova import db
+from nova import exception
+from nova import flags
+from nova import log as logging
+from nova import rpc
+from nova import volume
+from nova.compute import instance_types
+from nova.db import base
+from nova.volume import volume_types
+
+
+class VsaState:
+    CREATING = 'creating'       # VSA creating (not ready yet)
+    LAUNCHING = 'launching'     # Launching VCs (all BE volumes were created)
+    CREATED = 'created'         # VSA fully created and ready for use
+    PARTIAL = 'partial'         # Some BE drives were allocated
+    FAILED = 'failed'           # Some BE storage allocations failed
+    DELETING = 'deleting'       # VSA started the deletion procedure
+
+
+FLAGS = flags.FLAGS
+flags.DEFINE_string('vsa_ec2_access_key', None,
+                    'EC2 access key used by VSA for accessing nova')
+flags.DEFINE_string('vsa_ec2_user_id', None,
+                    'User ID used by VSA for accessing nova')
+flags.DEFINE_boolean('vsa_multi_vol_creation', True,
+                  'Ask scheduler to create multiple volumes in one call')
+flags.DEFINE_string('vsa_volume_type_name', 'VSA volume type',
+                    'Name of volume type associated with FE VSA volumes')
+
+LOG = logging.getLogger('nova.vsa')
+
+
+class API(base.Base):
+    """API for interacting with the VSA manager."""
+
+    def __init__(self, compute_api=None, volume_api=None, **kwargs):
+        self.compute_api = compute_api or compute.API()
+        self.volume_api = volume_api or volume.API()
+        super(API, self).__init__(**kwargs)
+
+    def _check_volume_type_correctness(self, vol_type):
+        if vol_type.get('extra_specs') == None or\
+           vol_type['extra_specs'].get('type') != 'vsa_drive' or\
+           vol_type['extra_specs'].get('drive_type') == None or\
+           vol_type['extra_specs'].get('drive_size') == None:
+
+            raise exception.ApiError(_("Invalid drive type %s")
+                                        % vol_type['name'])
+
+    def _get_default_vsa_instance_type(self):
+        return instance_types.get_instance_type_by_name(
+                                FLAGS.default_vsa_instance_type)
+
+    def _check_storage_parameters(self, context, vsa_name, storage,
+                                  shared, first_index=0):
+        """
+        Translates storage array of disks to the list of volumes
+        :param storage: List of dictionaries with following keys:
+                        disk_name, num_disks, size
+        :param shared: Specifies if storage is dedicated or shared.
+                       For shared storage disks split into partitions
+        """
+        volume_params = []
+        for node in storage:
+
+            name = node.get('drive_name', None)
+            num_disks = node.get('num_drives', 1)
+
+            if name is None:
+                raise exception.ApiError(_("No drive_name param found in %s")
+                                            % node)
+            try:
+                vol_type = volume_types.get_volume_type_by_name(context, name)
+            except exception.NotFound:
+                raise exception.ApiError(_("Invalid drive type name %s")
+                                            % name)
+
+            self._check_volume_type_correctness(vol_type)
+
+            # if size field present - override disk size specified in DB
+            size = int(node.get('size',
+                                vol_type['extra_specs'].get('drive_size')))
+
+            if shared:
+                part_size = FLAGS.vsa_part_size_gb
+                total_capacity = num_disks * size
+                num_volumes = total_capacity / part_size
+                size = part_size
+            else:
+                num_volumes = num_disks
+                size = 0    # special handling for full drives
+
+            for i in range(num_volumes):
+                volume_name = "drive-%03d" % first_index
+                first_index += 1
+                volume_desc = 'BE volume for VSA %s type %s' % \
+                              (vsa_name, name)
+                volume = {
+                    'size': size,
+                    'name': volume_name,
+                    'description': volume_desc,
+                    'volume_type_id': vol_type['id'],
+                    }
+                volume_params.append(volume)
+
+        return volume_params
+
+    def create(self, context, display_name='', display_description='',
+                vc_count=1, instance_type=None, image_name=None,
+                availability_zone=None, storage=[], shared=None):
+        """
+        Provision VSA instance with corresponding compute instances
+        and associated volumes
+        :param storage: List of dictionaries with following keys:
+                        disk_name, num_disks, size
+        :param shared: Specifies if storage is dedicated or shared.
+                       For shared storage disks split into partitions
+        """
+
+        LOG.info(_("*** Experimental VSA code ***"))
+
+        if vc_count > FLAGS.max_vcs_in_vsa:
+            LOG.warning(_("Requested number of VCs (%d) is too high."\
+                          " Setting to default"), vc_count)
+            vc_count = FLAGS.max_vcs_in_vsa
+
+        if instance_type is None:
+            instance_type = self._get_default_vsa_instance_type()
+
+        if availability_zone is None:
+            availability_zone = FLAGS.storage_availability_zone
+
+        if storage is None:
+            storage = []
+
+        if shared is None or shared == 'False' or shared == False:
+            shared = False
+        else:
+            shared = True
+
+        # check if image is ready before starting any work
+        if image_name is None:
+            image_name = FLAGS.vc_image_name
+        try:
+            image_service = self.compute_api.image_service
+            vc_image = image_service.show_by_name(context, image_name)
+            vc_image_href = vc_image['id']
+        except exception.ImageNotFound:
+            raise exception.ApiError(_("Failed to find configured image %s")
+                                        % image_name)
+
+        options = {
+            'display_name': display_name,
+            'display_description': display_description,
+            'project_id': context.project_id,
+            'availability_zone': availability_zone,
+            'instance_type_id': instance_type['id'],
+            'image_ref': vc_image_href,
+            'vc_count': vc_count,
+            'status': VsaState.CREATING,
+        }
+        LOG.info(_("Creating VSA: %s") % options)
+
+        # create DB entry for VSA instance
+        try:
+            vsa_ref = self.db.vsa_create(context, options)
+        except exception.Error:
+            raise exception.ApiError(_(sys.exc_info()[1]))
+        vsa_id = vsa_ref['id']
+        vsa_name = vsa_ref['name']
+
+        # check storage parameters
+        try:
+            volume_params = self._check_storage_parameters(context, vsa_name,
+                                                           storage, shared)
+        except exception.ApiError:
+            self.db.vsa_destroy(context, vsa_id)
+            raise exception.ApiError(_("Error in storage parameters: %s")
+                                        % storage)
+
+        # after creating DB entry, re-check and set some defaults
+        updates = {}
+        if (not hasattr(vsa_ref, 'display_name') or
+                vsa_ref.display_name is None or
+                vsa_ref.display_name == ''):
+            updates['display_name'] = display_name = vsa_name
+        updates['vol_count'] = len(volume_params)
+        vsa_ref = self.update(context, vsa_id, **updates)
+
+        # create volumes
+        if FLAGS.vsa_multi_vol_creation:
+            if len(volume_params) > 0:
+                request_spec = {
+                    'num_volumes': len(volume_params),
+                    'vsa_id': str(vsa_id),
+                    'volumes': volume_params,
+                }
+
+                rpc.cast(context,
+                         FLAGS.scheduler_topic,
+                         {"method": "create_volumes",
+                          "args": {"topic": FLAGS.volume_topic,
+                                   "request_spec": request_spec,
+                                   "availability_zone": availability_zone}})
+        else:
+            # create BE volumes one-by-one
+            for vol in volume_params:
+                try:
+                    vol_name = vol['name']
+                    vol_size = vol['size']
+                    vol_type_id = vol['volume_type_id']
+                    LOG.debug(_("VSA ID %(vsa_id)d %(vsa_name)s: Create "\
+                                "volume %(vol_name)s, %(vol_size)d GB, "\
+                                "type %(vol_type_id)s"), locals())
+
+                    vol_type = volume_types.get_volume_type(context,
+                                                vol['volume_type_id'])
+
+                    vol_ref = self.volume_api.create(context,
+                                    vol_size,
+                                    None,
+                                    vol_name,
+                                    vol['description'],
+                                    volume_type=vol_type,
+                                    metadata=dict(to_vsa_id=str(vsa_id)),
+                                    availability_zone=availability_zone)
+                except:
+                    self.update_vsa_status(context, vsa_id,
+                                           status=VsaState.PARTIAL)
+                    raise
+
+        if len(volume_params) == 0:
+            # No BE volumes - ask VSA manager to start VCs
+            rpc.cast(context,
+                     FLAGS.vsa_topic,
+                     {"method": "create_vsa",
+                      "args": {"vsa_id": str(vsa_id)}})
+
+        return vsa_ref
+
+    def update_vsa_status(self, context, vsa_id, status):
+        updates = dict(status=status)
+        LOG.info(_("VSA ID %(vsa_id)d: Update VSA status to %(status)s"),
+                    locals())
+        return self.update(context, vsa_id, **updates)
+
+    def update(self, context, vsa_id, **kwargs):
+        """Updates the VSA instance in the datastore.
+
+        :param context: The security context
+        :param vsa_id: ID of the VSA instance to update
+        :param kwargs: All additional keyword args are treated
+                       as data fields of the instance to be
+                       updated
+
+        :returns: None
+        """
+        LOG.info(_("VSA ID %(vsa_id)d: Update VSA call"), locals())
+
+        updatable_fields = ['status', 'vc_count', 'vol_count',
+                            'display_name', 'display_description']
+        changes = {}
+        for field in updatable_fields:
+            if field in kwargs:
+                changes[field] = kwargs[field]
+
+        vc_count = kwargs.get('vc_count', None)
+        if vc_count is not None:
+            # VP-TODO: This request may want to update number of VCs
+            # Get number of current VCs and add/delete VCs appropriately
+            vsa = self.get(context, vsa_id)
+            vc_count = int(vc_count)
+            if vc_count > FLAGS.max_vcs_in_vsa:
+                LOG.warning(_("Requested number of VCs (%d) is too high."\
+                              " Setting to default"), vc_count)
+                vc_count = FLAGS.max_vcs_in_vsa
+
+            if vsa['vc_count'] != vc_count:
+                self.update_num_vcs(context, vsa, vc_count)
+                changes['vc_count'] = vc_count
+
+        return self.db.vsa_update(context, vsa_id, changes)
+
+    def update_num_vcs(self, context, vsa, vc_count):
+        vsa_name = vsa['name']
+        old_vc_count = int(vsa['vc_count'])
+        if vc_count > old_vc_count:
+            add_cnt = vc_count - old_vc_count
+            LOG.debug(_("Adding %(add_cnt)s VCs to VSA %(vsa_name)s."),
+                        locals())
+            # VP-TODO: actual code for adding new VCs
+
+        elif vc_count < old_vc_count:
+            del_cnt = old_vc_count - vc_count
+            LOG.debug(_("Deleting %(del_cnt)s VCs from VSA %(vsa_name)s."),
+                        locals())
+            # VP-TODO: actual code for deleting extra VCs
+
+    def _force_volume_delete(self, ctxt, volume):
+        """Delete a volume, bypassing the check that it must be available."""
+        host = volume['host']
+        if not host:
+            # Deleting volume from database and skipping rpc.
+            self.db.volume_destroy(ctxt, volume['id'])
+            return
+
+        rpc.cast(ctxt,
+                 self.db.queue_get_for(ctxt, FLAGS.volume_topic, host),
+                 {"method": "delete_volume",
+                  "args": {"volume_id": volume['id']}})
+
+    def delete_vsa_volumes(self, context, vsa_id, direction,
+                           force_delete=True):
+        if direction == "FE":
+            volumes = self.get_all_vsa_volumes(context, vsa_id)
+        else:
+            volumes = self.get_all_vsa_drives(context, vsa_id)
+
+        for volume in volumes:
+            try:
+                vol_name = volume['name']
+                LOG.info(_("VSA ID %(vsa_id)s: Deleting %(direction)s "\
+                           "volume %(vol_name)s"), locals())
+                self.volume_api.delete(context, volume['id'])
+            except exception.ApiError:
+                LOG.info(_("Unable to delete volume %s"), volume['name'])
+                if force_delete:
+                    LOG.info(_("VSA ID %(vsa_id)s: Forced delete. "\
+                               "%(direction)s volume %(vol_name)s"), locals())
+                    self._force_volume_delete(context, volume)
+
+    def delete(self, context, vsa_id):
+        """Terminate a VSA instance."""
+        LOG.info(_("Going to try to terminate VSA ID %s"), vsa_id)
+
+        # Delete all FrontEnd and BackEnd volumes
+        self.delete_vsa_volumes(context, vsa_id, "FE", force_delete=True)
+        self.delete_vsa_volumes(context, vsa_id, "BE", force_delete=True)
+
+        # Delete all VC instances
+        instances = self.compute_api.get_all(context,
+                search_opts={'metadata': dict(vsa_id=str(vsa_id))})
+        for instance in instances:
+            name = instance['name']
+            LOG.debug(_("VSA ID %(vsa_id)s: Delete instance %(name)s"),
+                        locals())
+            self.compute_api.delete(context, instance['id'])
+
+        # Delete VSA instance
+        self.db.vsa_destroy(context, vsa_id)
+
+    def get(self, context, vsa_id):
+        rv = self.db.vsa_get(context, vsa_id)
+        return rv
+
+    def get_all(self, context):
+        if context.is_admin:
+            return self.db.vsa_get_all(context)
+        return self.db.vsa_get_all_by_project(context, context.project_id)
+
+    def get_vsa_volume_type(self, context):
+        name = FLAGS.vsa_volume_type_name
+        try:
+            vol_type = volume_types.get_volume_type_by_name(context, name)
+        except exception.NotFound:
+            volume_types.create(context, name,
+                                extra_specs=dict(type='vsa_volume'))
+            vol_type = volume_types.get_volume_type_by_name(context, name)
+
+        return vol_type
+
+    def get_all_vsa_instances(self, context, vsa_id):
+        return self.compute_api.get_all(context,
+                search_opts={'metadata': dict(vsa_id=str(vsa_id))})
+
+    def get_all_vsa_volumes(self, context, vsa_id):
+        return self.volume_api.get_all(context,
+                search_opts={'metadata': dict(from_vsa_id=str(vsa_id))})
+
+    def get_all_vsa_drives(self, context, vsa_id):
+        return self.volume_api.get_all(context,
+                search_opts={'metadata': dict(to_vsa_id=str(vsa_id))})

nova/vsa/connection.py

@@ -0,0 +1,25 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright (c) 2011 Zadara Storage Inc.
+# Copyright (c) 2011 OpenStack LLC.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+"""Abstraction of the underlying connection to VC."""
+
+from nova.vsa import fake
+
+
+def get_connection():
+    # Return an object that is able to talk to VCs
+    return fake.FakeVcConnection()

nova/vsa/fake.py

@@ -0,0 +1,22 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright (c) 2011 Zadara Storage Inc.
+# Copyright (c) 2011 OpenStack LLC.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+
+class FakeVcConnection(object):
+
+    def init_host(self, host):
+        pass

nova/vsa/manager.py

@@ -0,0 +1,179 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright (c) 2011 Zadara Storage Inc.
+# Copyright (c) 2011 OpenStack LLC.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+"""
+Handles all processes relating to Virtual Storage Arrays (VSA).
+
+**Related Flags**
+
+"""
+
+from nova import compute
+from nova import exception
+from nova import flags
+from nova import log as logging
+from nova import manager
+from nova import volume
+from nova import vsa
+from nova import utils
+from nova.compute import instance_types
+from nova.vsa import utils as vsa_utils
+from nova.vsa.api import VsaState
+
+FLAGS = flags.FLAGS
+flags.DEFINE_string('vsa_driver', 'nova.vsa.connection.get_connection',
+                    'Driver to use for controlling VSAs')
+
+LOG = logging.getLogger('nova.vsa.manager')
+
+
+class VsaManager(manager.SchedulerDependentManager):
+    """Manages Virtual Storage Arrays (VSAs)."""
+
+    def __init__(self, vsa_driver=None, *args, **kwargs):
+        if not vsa_driver:
+            vsa_driver = FLAGS.vsa_driver
+        self.driver = utils.import_object(vsa_driver)
+        self.compute_manager = utils.import_object(FLAGS.compute_manager)
+
+        self.compute_api = compute.API()
+        self.volume_api = volume.API()
+        self.vsa_api = vsa.API()
+
+        if FLAGS.vsa_ec2_user_id is None or \
+           FLAGS.vsa_ec2_access_key is None:
+            raise exception.VSANovaAccessParamNotFound()
+
+        super(VsaManager, self).__init__(*args, **kwargs)
+
+    def init_host(self):
+        self.driver.init_host(host=self.host)
+        super(VsaManager, self).init_host()
+
+    @exception.wrap_exception()
+    def create_vsa(self, context, vsa_id):
+        """Called by API if there were no BE volumes assigned"""
+        LOG.debug(_("Create call received for VSA %s"), vsa_id)
+
+        vsa_id = int(vsa_id)    # just in case
+
+        try:
+            vsa = self.vsa_api.get(context, vsa_id)
+        except Exception as ex:
+            msg = _("Failed to find VSA %(vsa_id)d") % locals()
+            LOG.exception(msg)
+            return
+
+        return self._start_vcs(context, vsa)
+
+    @exception.wrap_exception()
+    def vsa_volume_created(self, context, vol_id, vsa_id, status):
+        """Callback for volume creations"""
+        LOG.debug(_("VSA ID %(vsa_id)s: Drive %(vol_id)s created. "\
+                    "Status %(status)s"), locals())
+        vsa_id = int(vsa_id)    # just in case
+
+        # Get all volumes for this VSA
+        # check if any of them still in creating phase
+        drives = self.vsa_api.get_all_vsa_drives(context, vsa_id)
+        for drive in drives:
+            if drive['status'] == 'creating':
+                vol_name = drive['name']
+                vol_disp_name = drive['display_name']
+                LOG.debug(_("Drive %(vol_name)s (%(vol_disp_name)s) still "\
+                            "in creating phase - wait"), locals())
+                return
+
+        try:
+            vsa = self.vsa_api.get(context, vsa_id)
+        except Exception as ex:
+            msg = _("Failed to find VSA %(vsa_id)d") % locals()
+            LOG.exception(msg)
+            return
+
+        if len(drives) != vsa['vol_count']:
+            cvol_real = len(drives)
+            cvol_exp = vsa['vol_count']
+            LOG.debug(_("VSA ID %(vsa_id)d: Not all volumes are created "\
+                        "(%(cvol_real)d of %(cvol_exp)d)"), locals())
+            return
+
+        # all volumes created (successfully or not)
+        return self._start_vcs(context, vsa, drives)
+
+    def _start_vcs(self, context, vsa, drives=[]):
+        """Start VCs for VSA """
+
+        vsa_id = vsa['id']
+        if vsa['status'] == VsaState.CREATING:
+            self.vsa_api.update_vsa_status(context, vsa_id,
+                                           VsaState.LAUNCHING)
+        else:
+            return
+
+        # in _separate_ loop go over all volumes and mark as "attached"
+        has_failed_volumes = False
+        for drive in drives:
+            vol_name = drive['name']
+            vol_disp_name = drive['display_name']
+            status = drive['status']
+            LOG.info(_("VSA ID %(vsa_id)d: Drive %(vol_name)s "\
+                        "(%(vol_disp_name)s) is in %(status)s state"),
+                        locals())
+            if status == 'available':
+                try:
+                    # self.volume_api.update(context, volume['id'],
+                    #                   dict(attach_status="attached"))
+                    pass
+                except Exception as ex:
+                    msg = _("Failed to update attach status for volume "
+                            "%(vol_name)s. %(ex)s") % locals()
+                    LOG.exception(msg)
+            else:
+                has_failed_volumes = True
+
+        if has_failed_volumes:
+            LOG.info(_("VSA ID %(vsa_id)d: Delete all BE volumes"), locals())
+            self.vsa_api.delete_vsa_volumes(context, vsa_id, "BE", True)
+            self.vsa_api.update_vsa_status(context, vsa_id,
+                                           VsaState.FAILED)
+            return
+
+        # create user-data record for VC
+        storage_data = vsa_utils.generate_user_data(vsa, drives)
+
+        instance_type = instance_types.get_instance_type(
+                                            vsa['instance_type_id'])
+
+        # now start the VC instance
+
+        vc_count = vsa['vc_count']
+        LOG.info(_("VSA ID %(vsa_id)d: Start %(vc_count)d instances"),
+                    locals())
+        vc_instances = self.compute_api.create(context,
+                instance_type,      # vsa['vsa_instance_type'],
+                vsa['image_ref'],
+                min_count=1,
+                max_count=vc_count,
+                display_name='vc-' + vsa['display_name'],
+                display_description='VC for VSA ' + vsa['display_name'],
+                availability_zone=vsa['availability_zone'],
+                user_data=storage_data,
+                metadata=dict(vsa_id=str(vsa_id)))
+
+        self.vsa_api.update_vsa_status(context, vsa_id,
+                                       VsaState.CREATED)

nova/vsa/utils.py

@@ -0,0 +1,80 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright (c) 2011 Zadara Storage Inc.
+# Copyright (c) 2011 OpenStack LLC.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+import base64
+from xml.etree import ElementTree
+
+from nova import flags
+
+FLAGS = flags.FLAGS
+
+
+def generate_user_data(vsa, volumes):
+    SubElement = ElementTree.SubElement
+
+    e_vsa = ElementTree.Element("vsa")
+
+    e_vsa_detail = SubElement(e_vsa, "id")
+    e_vsa_detail.text = str(vsa['id'])
+    e_vsa_detail = SubElement(e_vsa, "name")
+    e_vsa_detail.text = vsa['display_name']
+    e_vsa_detail = SubElement(e_vsa, "description")
+    e_vsa_detail.text = vsa['display_description']
+    e_vsa_detail = SubElement(e_vsa, "vc_count")
+    e_vsa_detail.text = str(vsa['vc_count'])
+
+    e_vsa_detail = SubElement(e_vsa, "auth_user")
+    e_vsa_detail.text = FLAGS.vsa_ec2_user_id
+    e_vsa_detail = SubElement(e_vsa, "auth_access_key")
+    e_vsa_detail.text = FLAGS.vsa_ec2_access_key
+
+    e_volumes = SubElement(e_vsa, "volumes")
+    for volume in volumes:
+
+        loc = volume['provider_location']
+        if loc is None:
+            ip = ''
+            iscsi_iqn = ''
+            iscsi_portal = ''
+        else:
+            (iscsi_target, _sep, iscsi_iqn) = loc.partition(" ")
+            (ip, iscsi_portal) = iscsi_target.split(":", 1)
+
+        e_vol = SubElement(e_volumes, "volume")
+        e_vol_detail = SubElement(e_vol, "id")
+        e_vol_detail.text = str(volume['id'])
+        e_vol_detail = SubElement(e_vol, "name")
+        e_vol_detail.text = volume['name']
+        e_vol_detail = SubElement(e_vol, "display_name")
+        e_vol_detail.text = volume['display_name']
+        e_vol_detail = SubElement(e_vol, "size_gb")
+        e_vol_detail.text = str(volume['size'])
+        e_vol_detail = SubElement(e_vol, "status")
+        e_vol_detail.text = volume['status']
+        e_vol_detail = SubElement(e_vol, "ip")
+        e_vol_detail.text = ip
+        e_vol_detail = SubElement(e_vol, "iscsi_iqn")
+        e_vol_detail.text = iscsi_iqn
+        e_vol_detail = SubElement(e_vol, "iscsi_portal")
+        e_vol_detail.text = iscsi_portal
+        e_vol_detail = SubElement(e_vol, "lun")
+        e_vol_detail.text = '0'
+        e_vol_detail = SubElement(e_vol, "sn_host")
+        e_vol_detail.text = volume['host']
+
+    _xml = ElementTree.tostring(e_vsa)
+    return base64.b64encode(_xml)