x/python-ganttclient
Code Issues Proposed changes

Remove Virtual Storage Array (VSA) code

Browse Source 
* Remove core vsa code (api, manager, drivers)
* Remove nova-vsa executable
* Remove OpenStack Compute API v2 vsa extension
* Remove vsa scheduler
* Remove vsa db api methods
* Remove Zadara volume driver
* Do not migrate out any existing data
* Fixes bug 954490

Change-Id: Idab3d60796d5edbc23ef9f0887fcc1af558c6215
This commit is contained in:
Brian Waldon
2012-03-14 17:24:07 -07:00
committed by Chris Behrens
parent 37cb0bbe16
commit e83b24bf3a
15 changed files with 1 additions and 2744 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
bin/nova-all
View File

@@ -70,7 +70,7 @@ if __name__ == '__main__':
LOG.exception(_('Failed to load %s') % mod.__name__)
for binary in ['nova-compute', 'nova-volume',
'nova-network', 'nova-scheduler', 'nova-vsa', 'nova-cert']:
'nova-network', 'nova-scheduler', 'nova-cert']:
try:
servers.append(service.Service.create(binary=binary))
except (Exception, SystemExit):

475
bin/nova-manage
View File

@@ -87,7 +87,6 @@ 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.compute import instance_types
@@ -1116,478 +1115,6 @@ 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['vm_state'],
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=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")
logging.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 Exception:
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
try:
volume_types.create(self.context, name, extra_specs)
result = volume_types.get_volume_type_by_name(self.context, name)
self._list({name: result})
except exception.VolumeTypeExists:
print
print "Volume Type Exists"
print "Please ensure volume_type name is unique."
print "Currently defined volume types:"
print
self.list()
@args('--name', dest='name', metavar="<name>", help='Drive name')
def delete(self, name):
"""Marks volume types as deleted"""
try:
volume_types.destroy(self.context, name)
except exception.InvalidVolumeType:
print "Valid volume type name is required"
sys.exit(1)
except exception.DBError, e:
print "DB Error: %s" % e
sys.exit(2)
except Exception:
sys.exit(3)
else:
print "%s deleted" % name
@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 not all:
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"""
@@ -2067,7 +1594,6 @@ CATEGORIES = [
('agent', AgentBuildCommands),
('config', ConfigCommands),
('db', DbCommands),
('drive', VsaDriveTypeCommands),
('export', ExportCommands),
('fixed', FixedIpCommands),
('flavor', InstanceTypeCommands),
@@ -2085,7 +1611,6 @@ CATEGORIES = [
('vm', VmCommands),
('volume', VolumeCommands),
('vpn', VpnCommands),
('vsa', VsaCommands),
('logs', GetLogCommands)]

49
bin/nova-vsa
View File

@@ -1,49 +0,0 @@
#!/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()

18
nova/flags.py
View File

@@ -185,9 +185,6 @@ global_opts = [
cfg.StrOpt('network_topic',
default='network',
help='the topic network nodes listen on'),
cfg.StrOpt('vsa_topic',
default='vsa',
help='the topic that nova-vsa service listens on'),
cfg.StrOpt('rabbit_host',
default='localhost',
help='the RabbitMQ host'),
@@ -351,21 +348,6 @@ global_opts = [
cfg.StrOpt('scheduler_manager',
default='nova.scheduler.manager.SchedulerManager',
help='full class name for the Manager for scheduler'),
cfg.StrOpt('vsa_manager',
default='nova.vsa.manager.VsaManager',
help='full class name for the Manager for VSA'),
cfg.StrOpt('vc_image_name',
default='vc_image',
help='the VC image ID (for a VC image that exists in Glance)'),
cfg.StrOpt('default_vsa_instance_type',
default='m1.small',
help='default instance type for VSA instances'),
cfg.IntOpt('max_vcs_in_vsa',
default=32,
help='maxinum VCs in a VSA'),
cfg.IntOpt('vsa_part_size_gb',
default=100,
help='default partition size for shared capacity'),
cfg.StrOpt('firewall_driver',
default='nova.virt.firewall.IptablesFirewallDriver',
help='Firewall driver (defaults to iptables)'),

3
nova/rootwrap/volume.py
View File

@@ -42,7 +42,4 @@ filterlist = [
# nova/volume/driver.py: 'iscsiadm', '-m', 'discovery', '-t',...
# nova/volume/driver.py: 'iscsiadm', '-m', 'node', '-T', ...
filters.CommandFilter("/sbin/iscsiadm", "root"),
# nova/volume/driver.py:'/var/lib/zadara/bin/zadara_sncfg', *
# sudoers does not allow zadara_sncfg yet
]

532
nova/scheduler/vsa.py
View File

@@ -1,532 +0,0 @@
# 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 exception
from nova import flags
from nova import log as logging
from nova.openstack.common import cfg
from nova import rpc
from nova import utils
from nova.scheduler import driver
from nova.scheduler import simple
from nova.volume import volume_types
from nova.vsa import api as vsa_api
LOG = logging.getLogger(__name__)
vsa_scheduler_opts = [
cfg.IntOpt('drive_type_approx_capacity_percent',
default=10,
help='The percentage range for capacity comparison'),
cfg.IntOpt('vsa_unique_hosts_per_alloc',
default=10,
help='The number of unique hosts per storage allocation'),
cfg.BoolOpt('vsa_select_unique_drives',
default=True,
help='Allow selection of same host for multiple drives'),
]
FLAGS = flags.FLAGS
FLAGS.register_opts(vsa_scheduler_opts)
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.host_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 not unique 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 exception.NoValidHost(reason="")
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),
}
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.elevated(), options)
driver.cast_to_volume_host(context, vol['host'],
'create_volume', 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 service['disabled'] or not utils.service_is_up(service):
raise exception.WillNotSchedule(host=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 Exception:
LOG.exception(_("Error creating volumes"))
if vsa_id:
db.vsa_update(context, vsa_id,
dict(status=vsa_api.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:
driver.cast_to_volume_host(context, host, 'create_volume',
volume_id=volume_id, **_kwargs)
return None
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 Exception:
LOG.exception(_("Error creating volume"))
if volume_ref['to_vsa_id']:
db.vsa_update(context, volume_ref['to_vsa_id'],
dict(status=vsa_api.VsaState.FAILED))
raise
if host:
driver.cast_to_volume_host(context, host, 'create_volume',
volume_id=volume_id, **_kwargs)
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:
_fp = qos_values['PartitionDrive']['NumFreePartitions']
_fd = qos_values['FullDrive']['NumFreeDrives']
if (qos_values['AvailableCapacity'] >= size and
(_fp > 0 or _fd > 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)

626
nova/tests/scheduler/test_vsa_scheduler.py
View File

@@ -1,626 +0,0 @@
# 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.
from nova import db
from nova import exception
from nova import flags
from nova import log as logging
from nova import rpc
from nova.scheduler import vsa as vsa_sched
from nova.tests.scheduler import test_scheduler
from nova import utils
from nova.volume import volume_types
FLAGS = flags.FLAGS
LOG = logging.getLogger(__name__)
scheduled_volumes = []
scheduled_volume = {}
global_volume = {}
def fake_rpc_cast(*args, **kwargs):
pass
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_scheduler.SchedulerTestCase):
driver_cls = FakeVsaLeastUsedScheduler
def setUp(self):
super(VsaSchedulerTestCase, self).setUp()
self.host_num = 10
self.drive_type_num = 5
self.stubs.Set(rpc, 'cast', fake_rpc_cast)
self.stubs.Set(self.driver,
'_get_service_states', self._fake_get_service_states)
self.stubs.Set(self.driver,
'_provision_volume', self._fake_provision_volume)
self.stubs.Set(db, 'vsa_update', self._fake_vsa_update)
self.stubs.Set(db, 'volume_get', self._fake_volume_get)
self.stubs.Set(db, 'volume_update', self._fake_volume_update)
self.created_types_lst = []
def tearDown(self):
for name in self.created_types_lst:
volume_types.destroy(self.context.elevated(), name)
super(VsaSchedulerTestCase, self).tearDown()
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.elevated(), 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.VolumeTypeExists:
# 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())
def _fake_vsa_update(self, context, vsa_id, values):
LOG.debug(_("Test: VSA update request: vsa_id=%(vsa_id)s "
"values=%(values)s"), locals())
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']}
def _fake_service_get_by_args(self, context, host, binary):
return {'host': 'fake_host', 'disabled': False}
def _fake_service_is_up_True(self, service):
return True
def _fake_service_is_up_False(self, service):
return False
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.driver.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(exception.NoValidHost,
self.driver.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(exception.NoValidHost,
self.driver.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(exception.NoValidHost,
self.driver.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.driver,
'_get_service_states', self._fake_get_service_states)
self.stubs.Set(db, 'volume_create', self._fake_volume_create)
self.stubs.Set(db, 'volume_update', self._fake_volume_update)
self.stubs.Set(rpc, 'cast', fake_rpc_cast)
self.driver.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.driver.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.driver.schedule_create_volumes(self.context,
request_spec,
availability_zone=None)
self._print_service_states()
self.assertRaises(exception.NoValidHost,
self.driver.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.driver.schedule_create_volumes,
self.context.elevated(),
request_spec,
availability_zone="nova:host_5")
self.stubs.Set(db,
'service_get_by_args', self._fake_service_get_by_args)
self.stubs.Set(utils,
'service_is_up', self._fake_service_is_up_False)
self.assertRaises(exception.WillNotSchedule,
self.driver.schedule_create_volumes,
self.context.elevated(),
request_spec,
availability_zone="nova:host_5")
self.stubs.Set(utils,
'service_is_up', self._fake_service_is_up_True)
self.driver.schedule_create_volumes(self.context.elevated(),
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.driver.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(db, 'volume_get', _fake_volume_get_az)
self.stubs.Set(db, 'service_get_by_args',
self._fake_service_get_by_args)
self.stubs.Set(utils,
'service_is_up', self._fake_service_is_up_True)
self.driver.schedule_create_volume(self.context.elevated(),
123, availability_zone=None)
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(exception.NoValidHost,
self.driver.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.elevated(), 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
self.driver.schedule_create_volume(self.context,
123, availability_zone=None)
self.assertEqual(scheduled_volume['id'], 123)
self.assertEqual(scheduled_volume['host'], 'host_2')
class VsaSchedulerTestCaseMostAvail(VsaSchedulerTestCase):
driver_cls = FakeVsaMostAvailCapacityScheduler
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.elevated(), 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
self.driver.schedule_create_volume(self.context,
123, availability_zone=None)
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.driver.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.driver.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)

171
nova/tests/test_vsa.py
View File

@@ -1,171 +0,0 @@
# 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
from xml.etree import ElementTree
from nova import context
from nova import exception
from nova import flags
from nova import log as logging
from nova import test
from nova import vsa
from nova.volume import volume_types
from nova.vsa import utils as vsa_utils
import nova.image.fake
FLAGS = flags.FLAGS
LOG = logging.getLogger(__name__)
class VsaTestCase(test.TestCase):
def setUp(self):
super(VsaTestCase, self).setUp()
self.vsa_api = vsa.api.API()
self.flags(quota_volumes=100, 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 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.ImageNotFound,
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, 'vsa_create', fake_vsa_create)
self.assertRaises(exception.Error,
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.InvalidVolumeType,
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.InvalidVolumeType,
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"""
self.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):
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'])

133
nova/tests/test_vsa_volumes.py
View File

@@ -1,133 +0,0 @@
# 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.
from nova import exception
from nova import flags
from nova.vsa import api as vsa_api
from nova import volume
from nova import context
from nova import test
from nova import log as logging
import nova.image.fake
FLAGS = flags.FLAGS
LOG = logging.getLogger(__name__)
class VsaVolumesTestCase(test.TestCase):
def setUp(self):
super(VsaVolumesTestCase, self).setUp()
self.vsa_api = vsa_api.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)
super(VsaVolumesTestCase, self).tearDown()
def _default_volume_param(self):
return {
'size': 1,
'snapshot': 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,
{'status': 'available', 'host': 'fake'})
volume_ref = self.volume_api.get(self.context, volume_ref['id'])
self.volume_api.delete(self.context, volume_ref)
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 deletion 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,
{'status': 'in-use', 'host': 'fake'})
volume_ref = self.volume_api.get(self.context, volume_ref['id'])
self.assertRaises(exception.InvalidVolume,
self.volume_api.delete,
self.context, volume_ref)
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))

16
nova/vsa/__init__.py
View File

@@ -1,16 +0,0 @@
# 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.

412
nova/vsa/api.py
View File

@@ -1,412 +0,0 @@
# 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
"""
from nova import compute
from nova import exception
from nova import flags
from nova import log as logging
from nova.openstack.common import cfg
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
vsa_opts = [
cfg.StrOpt('vsa_ec2_access_key',
default=None,
help='EC2 access key used by VSA for accessing nova'),
cfg.StrOpt('vsa_ec2_user_id',
default=None,
help='User ID used by VSA for accessing nova'),
cfg.BoolOpt('vsa_multi_vol_creation',
default=True,
help='Ask scheduler to create multiple volumes in one call'),
cfg.StrOpt('vsa_volume_type_name',
default='VSA volume type',
help='Name of volume type associated with FE VSA volumes'),
]
FLAGS = flags.FLAGS
FLAGS.register_opts(vsa_opts)
LOG = logging.getLogger(__name__)
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') is None or
vol_type['extra_specs'].get('type') != 'vsa_drive' or
vol_type['extra_specs'].get('drive_type') is None or
vol_type['extra_specs'].get('drive_size') is None):
msg = _("invalid drive data")
raise exception.InvalidVolumeType(reason=msg)
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:
msg = _("drive_name not defined")
raise exception.InvalidVolumeType(reason=msg)
try:
vol_type = volume_types.get_volume_type_by_name(context, name)
except exception.NotFound:
msg = _("invalid drive type name %s")
raise exception.InvalidVolumeType(reason=msg % 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 compute instances and 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 not shared 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
image_service = self.compute_api.image_service
vc_image = image_service.show_by_name(context, image_name)
vc_image_href = vc_image['id']
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
vsa_ref = self.db.vsa_create(context, options)
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.InvalidVolumeType:
self.db.vsa_destroy(context, vsa_id)
raise
# 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,
vol_name,
vol['description'],
None,
volume_type=vol_type,
metadata=dict(to_vsa_id=str(vsa_id)),
availability_zone=availability_zone)
except Exception:
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(vladimir.p):
# 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(vladimir.p): 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(vladimir.p): 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)
except exception.InvalidVolume:
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))})

25
nova/vsa/connection.py
View File

@@ -1,25 +0,0 @@
# 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()

22
nova/vsa/fake.py
View File

@@ -1,22 +0,0 @@
# 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

181
nova/vsa/manager.py
View File

@@ -1,181 +0,0 @@
# 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.openstack.common import cfg
from nova import volume
from nova import utils
from nova.compute import instance_types
from nova.vsa import utils as vsa_utils
from nova.vsa import api as vsa_api
vsa_driver_opt = cfg.StrOpt('vsa_driver',
default='nova.vsa.connection.get_connection',
help='Driver to use for controlling VSAs')
FLAGS = flags.FLAGS
FLAGS.register_opt(vsa_driver_opt)
LOG = logging.getLogger(__name__)
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.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'] == vsa_api.VsaState.CREATING:
self.vsa_api.update_vsa_status(context, vsa_id,
vsa_api.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,
# 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,
vsa_api.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,
vsa_api.VsaState.CREATED)

80
nova/vsa/utils.py
View File

@@ -1,80 +0,0 @@
# 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)