From 0ba085928c75f2fc27fb03eaa3aaeff6618e8875 Mon Sep 17 00:00:00 2001 From: MORITA Kazutaka Date: Tue, 19 Apr 2011 20:48:26 +0900 Subject: Add support for creating a snapshot of a nova volume with euca-create-snapshot. --- nova/api/ec2/__init__.py | 6 ++ nova/api/ec2/cloud.py | 52 ++++++++++++--- nova/db/api.py | 39 +++++++++++ nova/db/sqlalchemy/api.py | 77 ++++++++++++++++++++++ .../versions/015_add_volume_snapshot_support.py | 71 ++++++++++++++++++++ nova/db/sqlalchemy/models.py | 24 +++++++ nova/exception.py | 50 ++++++++++++++ nova/volume/api.py | 44 +++++++++++++ nova/volume/driver.py | 8 +++ nova/volume/manager.py | 42 ++++++++++++ 10 files changed, 405 insertions(+), 8 deletions(-) create mode 100644 nova/db/sqlalchemy/migrate_repo/versions/015_add_volume_snapshot_support.py diff --git a/nova/api/ec2/__init__.py b/nova/api/ec2/__init__.py index cd59340bd..4a49a5a6b 100644 --- a/nova/api/ec2/__init__.py +++ b/nova/api/ec2/__init__.py @@ -327,6 +327,12 @@ class Executor(wsgi.Application): ec2_id = ec2utils.id_to_ec2_id(ex.volume_id, 'vol-%08x') message = _('Volume %s not found') % ec2_id return self._error(req, context, type(ex).__name__, message) + except exception.SnapshotNotFound as ex: + LOG.info(_('SnapshotNotFound raised: %s'), unicode(ex), + context=context) + ec2_id = ec2utils.id_to_ec2_id(ex.snapshot_id, 'snap-%08x') + message = _('Snapshot %s not found') % ec2_id + return self._error(req, context, type(ex).__name__, message) except exception.NotFound as ex: LOG.info(_('NotFound raised: %s'), unicode(ex), context=context) return self._error(req, context, type(ex).__name__, unicode(ex)) diff --git a/nova/api/ec2/cloud.py b/nova/api/ec2/cloud.py index 092b80fa2..f5360af0b 100644 --- a/nova/api/ec2/cloud.py +++ b/nova/api/ec2/cloud.py @@ -280,14 +280,46 @@ class CloudController(object): owner=None, restorable_by=None, **kwargs): - return {'snapshotSet': [{'snapshotId': 'fixme', - 'volumeId': 'fixme', - 'status': 'fixme', - 'startTime': 'fixme', - 'progress': 'fixme', - 'ownerId': 'fixme', - 'volumeSize': 0, - 'description': 'fixme'}]} + if snapshot_id: + snapshots = [] + for ec2_id in snapshot_id: + internal_id = ec2utils.ec2_id_to_id(ec2_id) + snapshot = self.volume_api.get_snapshot(context, snapshot_id=internal_id) + snapshots.append(snapshot) + else: + snapshots = self.volume_api.get_all_snapshots(context) + snapshots = [self._format_snapshot(context, s) for s in snapshots] + return {'snapshotSet': snapshots} + + def _format_snapshot(self, context, snapshot): + s = {} + s['snapshotId'] = ec2utils.id_to_ec2_id(snapshot['id'], 'snap-%08x') + s['volumeId'] = ec2utils.id_to_ec2_id(snapshot['volume_id'], 'vol-%08x') + s['status'] = snapshot['status'] + s['startTime'] = snapshot['created_at'] + s['progress'] = snapshot['progress'] + s['ownerId'] = snapshot['project_id'] + s['volumeSize'] = snapshot['volume_size'] + s['description'] = snapshot['display_description'] + + s['display_name'] = snapshot['display_name'] + s['display_description'] = snapshot['display_description'] + return s + + def create_snapshot(self, context, volume_id, **kwargs): + LOG.audit(_("Create snapshot of volume %s"), volume_id, context=context) + volume_id = ec2utils.ec2_id_to_id(volume_id) + snapshot = self.volume_api.create_snapshot( + context, + volume_id=volume_id, + name=kwargs.get('display_name'), + description=kwargs.get('display_description')) + return {'snapshotSet': [self._format_snapshot(context, snapshot)]} + + def delete_snapshot(self, context, snapshot_id, **kwargs): + snapshot_id = ec2utils.ec2_id_to_id(snapshot_id) + self.volume_api.delete_snapshot(context, snapshot_id=snapshot_id) + return True def describe_key_pairs(self, context, key_name=None, **kwargs): key_pairs = db.key_pair_get_all_by_user(context, context.user_id) @@ -595,6 +627,10 @@ class CloudController(object): 'volumeId': v['volumeId']}] else: v['attachmentSet'] = [{}] + if volume.get('snapshot_id') != None: + v['snapshotId'] = ec2utils.id_to_ec2_id(volume['snapshot_id'], 'snap-%08x') + else: + v['snapshotId'] = None v['display_name'] = volume['display_name'] v['display_description'] = volume['display_description'] diff --git a/nova/db/api.py b/nova/db/api.py index f9a4b5b4b..57e585a9c 100644 --- a/nova/db/api.py +++ b/nova/db/api.py @@ -47,6 +47,8 @@ flags.DEFINE_string('instance_name_template', 'instance-%08x', 'Template string to be used to generate instance names') flags.DEFINE_string('volume_name_template', 'volume-%08x', 'Template string to be used to generate instance names') +flags.DEFINE_string('snapshot_name_template', 'snapshot-%08x', + 'Template string to be used to generate instance names') IMPL = utils.LazyPluggable(FLAGS['db_backend'], @@ -871,6 +873,43 @@ def volume_update(context, volume_id, values): #################### +def snapshot_create(context, values): + """Create a volume from the values dictionary.""" + return IMPL.snapshot_create(context, values) + + +def snapshot_destroy(context, snapshot_id): + """Create a volume from the values dictionary.""" + return IMPL.snapshot_destroy(context, snapshot_id) + + +def snapshot_get(context, snapshot_id): + """Get a volume or raise if it does not exist.""" + return IMPL.snapshot_get(context, snapshot_id) + + +def snapshot_get_all(context): + """Get all volumes.""" + return IMPL.snapshot_get_all(context) + + +def snapshot_get_all_by_project(context, project_id): + """Get all volumes belonging to a project.""" + return IMPL.snapshot_get_all_by_project(context, project_id) + + +def snapshot_update(context, snapshot_id, values): + """Set the given properties on an snapshot and update it. + + Raises NotFound if snapshot does not exist. + + """ + return IMPL.snapshot_update(context, snapshot_id, values) + + +#################### + + def security_group_get_all(context): """Get all security groups.""" return IMPL.security_group_get_all(context) diff --git a/nova/db/sqlalchemy/api.py b/nova/db/sqlalchemy/api.py index 285b22a04..ebdb2ad5c 100644 --- a/nova/db/sqlalchemy/api.py +++ b/nova/db/sqlalchemy/api.py @@ -1758,6 +1758,83 @@ def volume_update(context, volume_id, values): ################### +@require_context +def snapshot_create(context, values): + snapshot_ref = models.Snapshot() + snapshot_ref.update(values) + + session = get_session() + with session.begin(): + snapshot_ref.save(session=session) + return snapshot_ref + + +@require_admin_context +def snapshot_destroy(context, snapshot_id): + session = get_session() + with session.begin(): + session.query(models.Snapshot).\ + filter_by(id=snapshot_id).\ + update({'deleted': 1, + 'deleted_at': datetime.datetime.utcnow(), + 'updated_at': literal_column('updated_at')}) + + +@require_context +def snapshot_get(context, snapshot_id, session=None): + if not session: + session = get_session() + result = None + + if is_admin_context(context): + result = session.query(models.Snapshot).\ + filter_by(id=snapshot_id).\ + filter_by(deleted=can_read_deleted(context)).\ + first() + elif is_user_context(context): + result = session.query(models.Snapshot).\ + filter_by(project_id=context.project_id).\ + filter_by(id=snapshot_id).\ + filter_by(deleted=False).\ + first() + if not result: + raise exception.SnapshotNotFound(_('Snapshot %s not found') % snapshot_id, + snapshot_id) + + return result + + +@require_admin_context +def snapshot_get_all(context): + session = get_session() + return session.query(models.Snapshot).\ + filter_by(deleted=can_read_deleted(context)).\ + all() + + +@require_context +def snapshot_get_all_by_project(context, project_id): + authorize_project_context(context, project_id) + + session = get_session() + return session.query(models.Snapshot).\ + filter_by(project_id=project_id).\ + filter_by(deleted=can_read_deleted(context)).\ + all() + + +@require_context +def snapshot_update(context, snapshot_id, values): + session = get_session() + with session.begin(): + snapshot_ref = snapshot_get(context, snapshot_id, session=session) + snapshot_ref.update(values) + snapshot_ref.save(session=session) + + +################### + + @require_context def security_group_get_all(context): session = get_session() diff --git a/nova/db/sqlalchemy/migrate_repo/versions/015_add_volume_snapshot_support.py b/nova/db/sqlalchemy/migrate_repo/versions/015_add_volume_snapshot_support.py new file mode 100644 index 000000000..288f63e72 --- /dev/null +++ b/nova/db/sqlalchemy/migrate_repo/versions/015_add_volume_snapshot_support.py @@ -0,0 +1,71 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# Copyright 2011 MORITA Kazutaka. +# 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 sqlalchemy import * +from migrate import * + +from nova import log as logging + +meta = MetaData() + +snapshots = Table('snapshots', meta, + Column('created_at', DateTime(timezone=False)), + Column('updated_at', DateTime(timezone=False)), + Column('deleted_at', DateTime(timezone=False)), + Column('deleted', Boolean(create_constraint=True, name=None)), + Column('id', Integer(), primary_key=True, nullable=False), + Column('volume_id', Integer(), nullable=False), + Column('user_id', + String(length=255, convert_unicode=False, assert_unicode=None, + unicode_error=None, _warn_on_bytestring=False)), + Column('project_id', + String(length=255, convert_unicode=False, assert_unicode=None, + unicode_error=None, _warn_on_bytestring=False)), + Column('status', + String(length=255, convert_unicode=False, assert_unicode=None, + unicode_error=None, _warn_on_bytestring=False)), + Column('progress', + String(length=255, convert_unicode=False, assert_unicode=None, + unicode_error=None, _warn_on_bytestring=False)), + Column('volume_size', Integer()), + Column('scheduled_at', DateTime(timezone=False)), + Column('display_name', + String(length=255, convert_unicode=False, assert_unicode=None, + unicode_error=None, _warn_on_bytestring=False)), + Column('display_description', + String(length=255, convert_unicode=False, assert_unicode=None, + unicode_error=None, _warn_on_bytestring=False)) + ) + + +def upgrade(migrate_engine): + # Upgrade operations go here. Don't create your own engine; + # bind migrate_engine to your metadata + meta.bind = migrate_engine + + try: + snapshots.create() + except Exception: + logging.info(repr(snapshots)) + logging.exception('Exception while creating table') + meta.drop_all(tables=[snapshots]) + raise + + +def downgrade(migrate_engine): + # Operations to reverse the above upgrade go here. + snapshots.drop() diff --git a/nova/db/sqlalchemy/models.py b/nova/db/sqlalchemy/models.py index 36a084a1d..2e0ead5f9 100644 --- a/nova/db/sqlalchemy/models.py +++ b/nova/db/sqlalchemy/models.py @@ -327,6 +327,30 @@ class Quota(BASE, NovaBase): metadata_items = Column(Integer) +class Snapshot(BASE, NovaBase): + """Represents a block storage device that can be attached to a vm.""" + __tablename__ = 'snapshots' + id = Column(Integer, primary_key=True, autoincrement=True) + + @property + def name(self): + return FLAGS.snapshot_name_template % self.id + + @property + def volume_name(self): + return FLAGS.volume_name_template % self.volume_id + + user_id = Column(String(255)) + project_id = Column(String(255)) + + volume_id = Column(Integer) + status = Column(String(255)) + progress = Column(String(255)) + volume_size = Column(Integer) + + display_name = Column(String(255)) + display_description = Column(String(255)) + class ExportDevice(BASE, NovaBase): """Represates a shelf and blade that a volume can be exported on.""" __tablename__ = 'export_devices' diff --git a/nova/exception.py b/nova/exception.py index 9905fb19b..2dffeb795 100644 --- a/nova/exception.py +++ b/nova/exception.py @@ -60,6 +60,56 @@ class ApiError(Error): class BuildInProgress(Error): + super(ApiError, self).__init__('%s: %s' % (code, message)) + + +class NotFound(Error): + pass + + +class InstanceNotFound(NotFound): + def __init__(self, message, instance_id): + self.instance_id = instance_id + super(InstanceNotFound, self).__init__(message) + + +class VolumeNotFound(NotFound): + def __init__(self, message, volume_id): + self.volume_id = volume_id + super(VolumeNotFound, self).__init__(message) + + +class SnapshotNotFound(NotFound): + def __init__(self, message, snapshot_id): + self.snapshot_id = snapshot_id + super(SnapshotNotFound, self).__init__(message) + + +class Duplicate(Error): + pass + + +class NotAuthorized(Error): + pass + + +class NotEmpty(Error): + pass + + +class Invalid(Error): + pass + + +class InvalidInputException(Error): + pass + + +class InvalidContentType(Error): + pass + + +class TimeoutException(Error): pass diff --git a/nova/volume/api.py b/nova/volume/api.py index 09befb647..c1af30de0 100644 --- a/nova/volume/api.py +++ b/nova/volume/api.py @@ -90,6 +90,15 @@ class API(base.Base): return self.db.volume_get_all(context) return self.db.volume_get_all_by_project(context, context.project_id) + def get_snapshot(self, context, snapshot_id): + rv = self.db.snapshot_get(context, snapshot_id) + return dict(rv.iteritems()) + + def get_all_snapshots(self, context): + if context.is_admin: + return self.db.snapshot_get_all(context) + return self.db.snapshot_get_all_by_project(context, context.project_id) + def check_attach(self, context, volume_id): volume = self.get(context, volume_id) # TODO(vish): abstract status checking? @@ -110,3 +119,38 @@ class API(base.Base): self.db.queue_get_for(context, FLAGS.compute_topic, host), {"method": "remove_volume", "args": {'volume_id': volume_id}}) + + def create_snapshot(self, context, volume_id, name, description): + volume = self.get(context, volume_id) + if volume['status'] != "available": + raise exception.ApiError(_("Volume status must be available")) + + options = { + 'volume_id': volume_id, + 'user_id': context.user_id, + 'project_id': context.project_id, + 'status': "creating", + 'progress': '0%', + 'volume_size': volume['size'], + 'display_name': name, + 'display_description': description} + + snapshot = self.db.snapshot_create(context, options) + rpc.cast(context, + FLAGS.scheduler_topic, + {"method": "create_snapshot", + "args": {"topic": FLAGS.volume_topic, + "volume_id": volume_id, + "snapshot_id": snapshot['id']}}) + return snapshot + + def delete_snapshot(self, context, snapshot_id): + snapshot = self.get_snapshot(context, snapshot_id) + if snapshot['status'] != "available": + raise exception.ApiError(_("Snapshot status must be available")) + self.db.snapshot_update(context, snapshot_id, {'status': 'deleting'}) + rpc.cast(context, + FLAGS.scheduler_topic, + {"method": "delete_snapshot", + "args": {"topic": FLAGS.volume_topic, + "snapshot_id": snapshot_id}}) diff --git a/nova/volume/driver.py b/nova/volume/driver.py index 55307ad9b..31998e307 100644 --- a/nova/volume/driver.py +++ b/nova/volume/driver.py @@ -122,6 +122,14 @@ class VolumeDriver(object): (FLAGS.volume_group, volume['name'])) + def create_snapshot(self, snapshot): + """Creates a snapshot.""" + raise NotImplementedError() + + def delete_snapshot(self, snapshot): + """Deletes a snapshot.""" + raise NotImplementedError() + def local_path(self, volume): # NOTE(vish): stops deprecation warning escaped_group = FLAGS.volume_group.replace('-', '--') diff --git a/nova/volume/manager.py b/nova/volume/manager.py index 2178389ce..87fd3bf17 100644 --- a/nova/volume/manager.py +++ b/nova/volume/manager.py @@ -152,6 +152,48 @@ class VolumeManager(manager.SchedulerDependentManager): LOG.debug(_("volume %s: deleted successfully"), volume_ref['name']) return True + def create_snapshot(self, context, volume_id, snapshot_id): + """Creates and exports the snapshot.""" + context = context.elevated() + snapshot_ref = self.db.snapshot_get(context, snapshot_id) + LOG.info(_("snapshot %s: creating"), snapshot_ref['name']) + + try: + snap_name = snapshot_ref['name'] + LOG.debug(_("snapshot %(snap_name)s: creating") % locals()) + model_update = self.driver.create_snapshot(snapshot_ref) + if model_update: + self.db.snapshot_update(context, snapshot_ref['id'], model_update) + + except Exception: + self.db.snapshot_update(context, + snapshot_ref['id'], {'status': 'error'}) + raise + + self.db.snapshot_update(context, + snapshot_ref['id'], {'status': 'available', + 'progress': '100%'}) + LOG.debug(_("snapshot %s: created successfully"), snapshot_ref['name']) + return snapshot_id + + def delete_snapshot(self, context, snapshot_id): + """Deletes and unexports snapshot.""" + context = context.elevated() + snapshot_ref = self.db.snapshot_get(context, snapshot_id) + + try: + LOG.debug(_("snapshot %s: deleting"), snapshot_ref['name']) + self.driver.delete_snapshot(snapshot_ref) + except Exception: + self.db.snapshot_update(context, + snapshot_ref['id'], + {'status': 'error_deleting'}) + raise + + self.db.snapshot_destroy(context, snapshot_id) + LOG.debug(_("snapshot %s: deleted successfully"), snapshot_ref['name']) + return True + def setup_compute_volume(self, context, volume_id): """Setup remote volume on compute host. -- cgit From dcda6be23c3797872c406f58578b05befd378c97 Mon Sep 17 00:00:00 2001 From: MORITA Kazutaka Date: Tue, 19 Apr 2011 20:48:26 +0900 Subject: Add support for creating a snapshot of a nova volume with euca-create-snapshot. --- nova/api/ec2/__init__.py | 6 ++ nova/api/ec2/cloud.py | 52 ++++++++++++--- nova/db/api.py | 39 +++++++++++ nova/db/sqlalchemy/api.py | 77 ++++++++++++++++++++++ .../versions/015_add_volume_snapshot_support.py | 71 ++++++++++++++++++++ nova/db/sqlalchemy/models.py | 24 +++++++ nova/exception.py | 6 ++ nova/volume/api.py | 44 +++++++++++++ nova/volume/driver.py | 8 +++ nova/volume/manager.py | 42 ++++++++++++ 10 files changed, 361 insertions(+), 8 deletions(-) create mode 100644 nova/db/sqlalchemy/migrate_repo/versions/015_add_volume_snapshot_support.py diff --git a/nova/api/ec2/__init__.py b/nova/api/ec2/__init__.py index a3c3b25a1..a89d65a38 100644 --- a/nova/api/ec2/__init__.py +++ b/nova/api/ec2/__init__.py @@ -331,6 +331,12 @@ class Executor(wsgi.Application): ec2_id = ec2utils.id_to_ec2_id(ex.volume_id, 'vol-%08x') message = _('Volume %s not found') % ec2_id return self._error(req, context, type(ex).__name__, message) + except exception.SnapshotNotFound as ex: + LOG.info(_('SnapshotNotFound raised: %s'), unicode(ex), + context=context) + ec2_id = ec2utils.id_to_ec2_id(ex.snapshot_id, 'snap-%08x') + message = _('Snapshot %s not found') % ec2_id + return self._error(req, context, type(ex).__name__, message) except exception.NotFound as ex: LOG.info(_('NotFound raised: %s'), unicode(ex), context=context) return self._error(req, context, type(ex).__name__, unicode(ex)) diff --git a/nova/api/ec2/cloud.py b/nova/api/ec2/cloud.py index bd4c9dcd4..6daf299b9 100644 --- a/nova/api/ec2/cloud.py +++ b/nova/api/ec2/cloud.py @@ -283,14 +283,46 @@ class CloudController(object): owner=None, restorable_by=None, **kwargs): - return {'snapshotSet': [{'snapshotId': 'fixme', - 'volumeId': 'fixme', - 'status': 'fixme', - 'startTime': 'fixme', - 'progress': 'fixme', - 'ownerId': 'fixme', - 'volumeSize': 0, - 'description': 'fixme'}]} + if snapshot_id: + snapshots = [] + for ec2_id in snapshot_id: + internal_id = ec2utils.ec2_id_to_id(ec2_id) + snapshot = self.volume_api.get_snapshot(context, snapshot_id=internal_id) + snapshots.append(snapshot) + else: + snapshots = self.volume_api.get_all_snapshots(context) + snapshots = [self._format_snapshot(context, s) for s in snapshots] + return {'snapshotSet': snapshots} + + def _format_snapshot(self, context, snapshot): + s = {} + s['snapshotId'] = ec2utils.id_to_ec2_id(snapshot['id'], 'snap-%08x') + s['volumeId'] = ec2utils.id_to_ec2_id(snapshot['volume_id'], 'vol-%08x') + s['status'] = snapshot['status'] + s['startTime'] = snapshot['created_at'] + s['progress'] = snapshot['progress'] + s['ownerId'] = snapshot['project_id'] + s['volumeSize'] = snapshot['volume_size'] + s['description'] = snapshot['display_description'] + + s['display_name'] = snapshot['display_name'] + s['display_description'] = snapshot['display_description'] + return s + + def create_snapshot(self, context, volume_id, **kwargs): + LOG.audit(_("Create snapshot of volume %s"), volume_id, context=context) + volume_id = ec2utils.ec2_id_to_id(volume_id) + snapshot = self.volume_api.create_snapshot( + context, + volume_id=volume_id, + name=kwargs.get('display_name'), + description=kwargs.get('display_description')) + return {'snapshotSet': [self._format_snapshot(context, snapshot)]} + + def delete_snapshot(self, context, snapshot_id, **kwargs): + snapshot_id = ec2utils.ec2_id_to_id(snapshot_id) + self.volume_api.delete_snapshot(context, snapshot_id=snapshot_id) + return True def describe_key_pairs(self, context, key_name=None, **kwargs): key_pairs = db.key_pair_get_all_by_user(context, context.user_id) @@ -598,6 +630,10 @@ class CloudController(object): 'volumeId': v['volumeId']}] else: v['attachmentSet'] = [{}] + if volume.get('snapshot_id') != None: + v['snapshotId'] = ec2utils.id_to_ec2_id(volume['snapshot_id'], 'snap-%08x') + else: + v['snapshotId'] = None v['display_name'] = volume['display_name'] v['display_description'] = volume['display_description'] diff --git a/nova/db/api.py b/nova/db/api.py index 63901e94d..9fc4b8c0a 100644 --- a/nova/db/api.py +++ b/nova/db/api.py @@ -46,6 +46,8 @@ flags.DEFINE_string('instance_name_template', 'instance-%08x', 'Template string to be used to generate instance names') flags.DEFINE_string('volume_name_template', 'volume-%08x', 'Template string to be used to generate instance names') +flags.DEFINE_string('snapshot_name_template', 'snapshot-%08x', + 'Template string to be used to generate instance names') IMPL = utils.LazyPluggable(FLAGS['db_backend'], @@ -867,6 +869,43 @@ def volume_update(context, volume_id, values): #################### +def snapshot_create(context, values): + """Create a volume from the values dictionary.""" + return IMPL.snapshot_create(context, values) + + +def snapshot_destroy(context, snapshot_id): + """Create a volume from the values dictionary.""" + return IMPL.snapshot_destroy(context, snapshot_id) + + +def snapshot_get(context, snapshot_id): + """Get a volume or raise if it does not exist.""" + return IMPL.snapshot_get(context, snapshot_id) + + +def snapshot_get_all(context): + """Get all volumes.""" + return IMPL.snapshot_get_all(context) + + +def snapshot_get_all_by_project(context, project_id): + """Get all volumes belonging to a project.""" + return IMPL.snapshot_get_all_by_project(context, project_id) + + +def snapshot_update(context, snapshot_id, values): + """Set the given properties on an snapshot and update it. + + Raises NotFound if snapshot does not exist. + + """ + return IMPL.snapshot_update(context, snapshot_id, values) + + +#################### + + def security_group_get_all(context): """Get all security groups.""" return IMPL.security_group_get_all(context) diff --git a/nova/db/sqlalchemy/api.py b/nova/db/sqlalchemy/api.py index 646675a45..059a22cb9 100644 --- a/nova/db/sqlalchemy/api.py +++ b/nova/db/sqlalchemy/api.py @@ -1760,6 +1760,83 @@ def volume_update(context, volume_id, values): ################### +@require_context +def snapshot_create(context, values): + snapshot_ref = models.Snapshot() + snapshot_ref.update(values) + + session = get_session() + with session.begin(): + snapshot_ref.save(session=session) + return snapshot_ref + + +@require_admin_context +def snapshot_destroy(context, snapshot_id): + session = get_session() + with session.begin(): + session.query(models.Snapshot).\ + filter_by(id=snapshot_id).\ + update({'deleted': 1, + 'deleted_at': datetime.datetime.utcnow(), + 'updated_at': literal_column('updated_at')}) + + +@require_context +def snapshot_get(context, snapshot_id, session=None): + if not session: + session = get_session() + result = None + + if is_admin_context(context): + result = session.query(models.Snapshot).\ + filter_by(id=snapshot_id).\ + filter_by(deleted=can_read_deleted(context)).\ + first() + elif is_user_context(context): + result = session.query(models.Snapshot).\ + filter_by(project_id=context.project_id).\ + filter_by(id=snapshot_id).\ + filter_by(deleted=False).\ + first() + if not result: + raise exception.SnapshotNotFound(_('Snapshot %s not found') % snapshot_id, + snapshot_id) + + return result + + +@require_admin_context +def snapshot_get_all(context): + session = get_session() + return session.query(models.Snapshot).\ + filter_by(deleted=can_read_deleted(context)).\ + all() + + +@require_context +def snapshot_get_all_by_project(context, project_id): + authorize_project_context(context, project_id) + + session = get_session() + return session.query(models.Snapshot).\ + filter_by(project_id=project_id).\ + filter_by(deleted=can_read_deleted(context)).\ + all() + + +@require_context +def snapshot_update(context, snapshot_id, values): + session = get_session() + with session.begin(): + snapshot_ref = snapshot_get(context, snapshot_id, session=session) + snapshot_ref.update(values) + snapshot_ref.save(session=session) + + +################### + + @require_context def security_group_get_all(context): session = get_session() diff --git a/nova/db/sqlalchemy/migrate_repo/versions/015_add_volume_snapshot_support.py b/nova/db/sqlalchemy/migrate_repo/versions/015_add_volume_snapshot_support.py new file mode 100644 index 000000000..288f63e72 --- /dev/null +++ b/nova/db/sqlalchemy/migrate_repo/versions/015_add_volume_snapshot_support.py @@ -0,0 +1,71 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# Copyright 2011 MORITA Kazutaka. +# 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 sqlalchemy import * +from migrate import * + +from nova import log as logging + +meta = MetaData() + +snapshots = Table('snapshots', meta, + Column('created_at', DateTime(timezone=False)), + Column('updated_at', DateTime(timezone=False)), + Column('deleted_at', DateTime(timezone=False)), + Column('deleted', Boolean(create_constraint=True, name=None)), + Column('id', Integer(), primary_key=True, nullable=False), + Column('volume_id', Integer(), nullable=False), + Column('user_id', + String(length=255, convert_unicode=False, assert_unicode=None, + unicode_error=None, _warn_on_bytestring=False)), + Column('project_id', + String(length=255, convert_unicode=False, assert_unicode=None, + unicode_error=None, _warn_on_bytestring=False)), + Column('status', + String(length=255, convert_unicode=False, assert_unicode=None, + unicode_error=None, _warn_on_bytestring=False)), + Column('progress', + String(length=255, convert_unicode=False, assert_unicode=None, + unicode_error=None, _warn_on_bytestring=False)), + Column('volume_size', Integer()), + Column('scheduled_at', DateTime(timezone=False)), + Column('display_name', + String(length=255, convert_unicode=False, assert_unicode=None, + unicode_error=None, _warn_on_bytestring=False)), + Column('display_description', + String(length=255, convert_unicode=False, assert_unicode=None, + unicode_error=None, _warn_on_bytestring=False)) + ) + + +def upgrade(migrate_engine): + # Upgrade operations go here. Don't create your own engine; + # bind migrate_engine to your metadata + meta.bind = migrate_engine + + try: + snapshots.create() + except Exception: + logging.info(repr(snapshots)) + logging.exception('Exception while creating table') + meta.drop_all(tables=[snapshots]) + raise + + +def downgrade(migrate_engine): + # Operations to reverse the above upgrade go here. + snapshots.drop() diff --git a/nova/db/sqlalchemy/models.py b/nova/db/sqlalchemy/models.py index f79d0f16c..9abe4d9ae 100644 --- a/nova/db/sqlalchemy/models.py +++ b/nova/db/sqlalchemy/models.py @@ -327,6 +327,30 @@ class Quota(BASE, NovaBase): metadata_items = Column(Integer) +class Snapshot(BASE, NovaBase): + """Represents a block storage device that can be attached to a vm.""" + __tablename__ = 'snapshots' + id = Column(Integer, primary_key=True, autoincrement=True) + + @property + def name(self): + return FLAGS.snapshot_name_template % self.id + + @property + def volume_name(self): + return FLAGS.volume_name_template % self.volume_id + + user_id = Column(String(255)) + project_id = Column(String(255)) + + volume_id = Column(Integer) + status = Column(String(255)) + progress = Column(String(255)) + volume_size = Column(Integer) + + display_name = Column(String(255)) + display_description = Column(String(255)) + class ExportDevice(BASE, NovaBase): """Represates a shelf and blade that a volume can be exported on.""" __tablename__ = 'export_devices' diff --git a/nova/exception.py b/nova/exception.py index 4e2bbdbaf..7adc3d007 100644 --- a/nova/exception.py +++ b/nova/exception.py @@ -68,6 +68,12 @@ class VolumeNotFound(NotFound): super(VolumeNotFound, self).__init__(message) +class SnapshotNotFound(NotFound): + def __init__(self, message, snapshot_id): + self.snapshot_id = snapshot_id + super(SnapshotNotFound, self).__init__(message) + + class Duplicate(Error): pass diff --git a/nova/volume/api.py b/nova/volume/api.py index 4b4bb9dc5..f5285f31f 100644 --- a/nova/volume/api.py +++ b/nova/volume/api.py @@ -90,6 +90,15 @@ class API(base.Base): return self.db.volume_get_all(context) return self.db.volume_get_all_by_project(context, context.project_id) + def get_snapshot(self, context, snapshot_id): + rv = self.db.snapshot_get(context, snapshot_id) + return dict(rv.iteritems()) + + def get_all_snapshots(self, context): + if context.is_admin: + return self.db.snapshot_get_all(context) + return self.db.snapshot_get_all_by_project(context, context.project_id) + def check_attach(self, context, volume_id): volume = self.get(context, volume_id) # TODO(vish): abstract status checking? @@ -103,3 +112,38 @@ class API(base.Base): # TODO(vish): abstract status checking? if volume['status'] == "available": raise exception.ApiError(_("Volume is already detached")) + + def create_snapshot(self, context, volume_id, name, description): + volume = self.get(context, volume_id) + if volume['status'] != "available": + raise exception.ApiError(_("Volume status must be available")) + + options = { + 'volume_id': volume_id, + 'user_id': context.user_id, + 'project_id': context.project_id, + 'status': "creating", + 'progress': '0%', + 'volume_size': volume['size'], + 'display_name': name, + 'display_description': description} + + snapshot = self.db.snapshot_create(context, options) + rpc.cast(context, + FLAGS.scheduler_topic, + {"method": "create_snapshot", + "args": {"topic": FLAGS.volume_topic, + "volume_id": volume_id, + "snapshot_id": snapshot['id']}}) + return snapshot + + def delete_snapshot(self, context, snapshot_id): + snapshot = self.get_snapshot(context, snapshot_id) + if snapshot['status'] != "available": + raise exception.ApiError(_("Snapshot status must be available")) + self.db.snapshot_update(context, snapshot_id, {'status': 'deleting'}) + rpc.cast(context, + FLAGS.scheduler_topic, + {"method": "delete_snapshot", + "args": {"topic": FLAGS.volume_topic, + "snapshot_id": snapshot_id}}) diff --git a/nova/volume/driver.py b/nova/volume/driver.py index 55307ad9b..31998e307 100644 --- a/nova/volume/driver.py +++ b/nova/volume/driver.py @@ -122,6 +122,14 @@ class VolumeDriver(object): (FLAGS.volume_group, volume['name'])) + def create_snapshot(self, snapshot): + """Creates a snapshot.""" + raise NotImplementedError() + + def delete_snapshot(self, snapshot): + """Deletes a snapshot.""" + raise NotImplementedError() + def local_path(self, volume): # NOTE(vish): stops deprecation warning escaped_group = FLAGS.volume_group.replace('-', '--') diff --git a/nova/volume/manager.py b/nova/volume/manager.py index 2178389ce..87fd3bf17 100644 --- a/nova/volume/manager.py +++ b/nova/volume/manager.py @@ -152,6 +152,48 @@ class VolumeManager(manager.SchedulerDependentManager): LOG.debug(_("volume %s: deleted successfully"), volume_ref['name']) return True + def create_snapshot(self, context, volume_id, snapshot_id): + """Creates and exports the snapshot.""" + context = context.elevated() + snapshot_ref = self.db.snapshot_get(context, snapshot_id) + LOG.info(_("snapshot %s: creating"), snapshot_ref['name']) + + try: + snap_name = snapshot_ref['name'] + LOG.debug(_("snapshot %(snap_name)s: creating") % locals()) + model_update = self.driver.create_snapshot(snapshot_ref) + if model_update: + self.db.snapshot_update(context, snapshot_ref['id'], model_update) + + except Exception: + self.db.snapshot_update(context, + snapshot_ref['id'], {'status': 'error'}) + raise + + self.db.snapshot_update(context, + snapshot_ref['id'], {'status': 'available', + 'progress': '100%'}) + LOG.debug(_("snapshot %s: created successfully"), snapshot_ref['name']) + return snapshot_id + + def delete_snapshot(self, context, snapshot_id): + """Deletes and unexports snapshot.""" + context = context.elevated() + snapshot_ref = self.db.snapshot_get(context, snapshot_id) + + try: + LOG.debug(_("snapshot %s: deleting"), snapshot_ref['name']) + self.driver.delete_snapshot(snapshot_ref) + except Exception: + self.db.snapshot_update(context, + snapshot_ref['id'], + {'status': 'error_deleting'}) + raise + + self.db.snapshot_destroy(context, snapshot_id) + LOG.debug(_("snapshot %s: deleted successfully"), snapshot_ref['name']) + return True + def setup_compute_volume(self, context, volume_id): """Setup remote volume on compute host. -- cgit From f76f2ee50f2407155a0aaefac3224e6af14e7d26 Mon Sep 17 00:00:00 2001 From: MORITA Kazutaka Date: Tue, 19 Apr 2011 20:50:10 +0900 Subject: Add support for creating a Sheepdog snapshot. --- nova/volume/driver.py | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/nova/volume/driver.py b/nova/volume/driver.py index 31998e307..ba0a7efef 100644 --- a/nova/volume/driver.py +++ b/nova/volume/driver.py @@ -620,6 +620,16 @@ class SheepdogDriver(VolumeDriver): """Deletes a logical volume""" self._try_execute('collie', 'vdi', 'delete', volume['name']) + def create_snapshot(self, snapshot): + """Creates a sheepdog snapshot""" + self._try_execute('qemu-img', 'snapshot', '-c', snapshot['name'], + "sheepdog:%s" % snapshot['volume_name']) + + def delete_snapshot(self, snapshot): + """Deletes a sheepdog snapshot""" + self._try_execute('collie', 'vdi', 'delete', snapshot['volume_name'], + '-s', snapshot['name']) + def local_path(self, volume): return "sheepdog:%s" % volume['name'] -- cgit From aad857a18153792d96f300732c3bb5bb16aa02c3 Mon Sep 17 00:00:00 2001 From: MORITA Kazutaka Date: Tue, 19 Apr 2011 20:50:10 +0900 Subject: Add support for creating a Sheepdog snapshot. --- nova/volume/driver.py | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/nova/volume/driver.py b/nova/volume/driver.py index 31998e307..ba0a7efef 100644 --- a/nova/volume/driver.py +++ b/nova/volume/driver.py @@ -620,6 +620,16 @@ class SheepdogDriver(VolumeDriver): """Deletes a logical volume""" self._try_execute('collie', 'vdi', 'delete', volume['name']) + def create_snapshot(self, snapshot): + """Creates a sheepdog snapshot""" + self._try_execute('qemu-img', 'snapshot', '-c', snapshot['name'], + "sheepdog:%s" % snapshot['volume_name']) + + def delete_snapshot(self, snapshot): + """Deletes a sheepdog snapshot""" + self._try_execute('collie', 'vdi', 'delete', snapshot['volume_name'], + '-s', snapshot['name']) + def local_path(self, volume): return "sheepdog:%s" % volume['name'] -- cgit From 2f3819628b6d3dea13a56ea6e93e02992b2e1f5f Mon Sep 17 00:00:00 2001 From: MORITA Kazutaka Date: Tue, 19 Apr 2011 21:01:25 +0900 Subject: Add support for creating a new volume from a existing snapshot with EC2 API. --- nova/api/ec2/cloud.py | 12 +++++- .../versions/016_add_snapshot_id_to_volumes.py | 48 ++++++++++++++++++++++ nova/db/sqlalchemy/models.py | 2 + nova/volume/api.py | 12 +++++- nova/volume/driver.py | 4 ++ nova/volume/manager.py | 9 +++- 6 files changed, 81 insertions(+), 6 deletions(-) create mode 100644 nova/db/sqlalchemy/migrate_repo/versions/016_add_snapshot_id_to_volumes.py diff --git a/nova/api/ec2/cloud.py b/nova/api/ec2/cloud.py index f5360af0b..aa15539ac 100644 --- a/nova/api/ec2/cloud.py +++ b/nova/api/ec2/cloud.py @@ -636,11 +636,19 @@ class CloudController(object): v['display_description'] = volume['display_description'] return v - def create_volume(self, context, size, **kwargs): - LOG.audit(_("Create volume of %s GB"), size, context=context) + def create_volume(self, context, **kwargs): + size = kwargs.get('size'); + if kwargs.get('snapshot_id') != None: + snapshot_id = ec2utils.ec2_id_to_id(kwargs['snapshot_id']) + LOG.audit(_("Create volume from snapshot %s"), snapshot_id, context=context) + else: + snapshot_id = None + LOG.audit(_("Create volume of %s GB"), size, context=context) + volume = self.volume_api.create( context, size=size, + snapshot_id=snapshot_id, name=kwargs.get('display_name'), description=kwargs.get('display_description')) # TODO(vish): Instance should be None at db layer instead of diff --git a/nova/db/sqlalchemy/migrate_repo/versions/016_add_snapshot_id_to_volumes.py b/nova/db/sqlalchemy/migrate_repo/versions/016_add_snapshot_id_to_volumes.py new file mode 100644 index 000000000..0a50123bf --- /dev/null +++ b/nova/db/sqlalchemy/migrate_repo/versions/016_add_snapshot_id_to_volumes.py @@ -0,0 +1,48 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# Copyright 2011 MORITA Kazutaka. +# 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 sqlalchemy import * +from migrate import * + +from nova import log as logging + + +meta = MetaData() + + +# Table stub-definitions +# Just for the ForeignKey and column creation to succeed, these are not the +# actual definitions of instances or services. +# +volumes = Table('volumes', meta, + Column('id', Integer(), primary_key=True, nullable=False), + ) + +# +# New Column +# + +snapshot_id = Column('snapshot_id', Integer()) + + +def upgrade(migrate_engine): + # Upgrade operations go here. Don't create your own engine; + # bind migrate_engine to your metadata + meta.bind = migrate_engine + + # Add columns to existing tables + volumes.create_column(snapshot_id) diff --git a/nova/db/sqlalchemy/models.py b/nova/db/sqlalchemy/models.py index 2e0ead5f9..ca762ca9f 100644 --- a/nova/db/sqlalchemy/models.py +++ b/nova/db/sqlalchemy/models.py @@ -287,6 +287,8 @@ class Volume(BASE, NovaBase): user_id = Column(String(255)) project_id = Column(String(255)) + snapshot_id = Column(String(255)) + host = Column(String(255)) # , ForeignKey('hosts.id')) size = Column(Integer) availability_zone = Column(String(255)) # TODO(vish): foreign key? diff --git a/nova/volume/api.py b/nova/volume/api.py index c1af30de0..7fa80383b 100644 --- a/nova/volume/api.py +++ b/nova/volume/api.py @@ -39,7 +39,13 @@ LOG = logging.getLogger('nova.volume') class API(base.Base): """API for interacting with the volume manager.""" - def create(self, context, size, name, description): + def create(self, context, size, snapshot_id, name, description): + if snapshot_id != None: + snapshot = self.get_snapshot(context, snapshot_id) + if snapshot['status'] != "available": + raise exception.ApiError(_("Snapshot status must be available")) + size = snapshot['volume_size'] + if quota.allowed_volumes(context, 1, size) < 1: pid = context.project_id LOG.warn(_("Quota exceeeded for %(pid)s, tried to create" @@ -51,6 +57,7 @@ class API(base.Base): 'size': size, 'user_id': context.user_id, 'project_id': context.project_id, + 'snapshot_id': snapshot_id, 'availability_zone': FLAGS.storage_availability_zone, 'status': "creating", 'attach_status': "detached", @@ -62,7 +69,8 @@ class API(base.Base): FLAGS.scheduler_topic, {"method": "create_volume", "args": {"topic": FLAGS.volume_topic, - "volume_id": volume['id']}}) + "volume_id": volume['id'], + "snapshot_id": snapshot_id}}) return volume def delete(self, context, volume_id): diff --git a/nova/volume/driver.py b/nova/volume/driver.py index ba0a7efef..02b0d50f4 100644 --- a/nova/volume/driver.py +++ b/nova/volume/driver.py @@ -101,6 +101,10 @@ class VolumeDriver(object): volume['name'], FLAGS.volume_group) + def create_volume_from_snapshot(self, volume, snapshot): + """Creates a volume from a snapshot.""" + raise NotImplementedError() + def delete_volume(self, volume): """Deletes a logical volume.""" try: diff --git a/nova/volume/manager.py b/nova/volume/manager.py index 87fd3bf17..7d47fc191 100644 --- a/nova/volume/manager.py +++ b/nova/volume/manager.py @@ -90,7 +90,7 @@ class VolumeManager(manager.SchedulerDependentManager): else: LOG.info(_("volume %s: skipping export"), volume['name']) - def create_volume(self, context, volume_id): + def create_volume(self, context, volume_id, snapshot_id): """Creates and exports the volume.""" context = context.elevated() volume_ref = self.db.volume_get(context, volume_id) @@ -108,7 +108,12 @@ class VolumeManager(manager.SchedulerDependentManager): vol_size = volume_ref['size'] LOG.debug(_("volume %(vol_name)s: creating lv of" " size %(vol_size)sG") % locals()) - model_update = self.driver.create_volume(volume_ref) + if snapshot_id == None: + model_update = self.driver.create_volume(volume_ref) + else: + snapshot_ref = self.db.snapshot_get(context, snapshot_id) + model_update = self.driver.create_volume_from_snapshot(volume_ref, + snapshot_ref) if model_update: self.db.volume_update(context, volume_ref['id'], model_update) -- cgit From 1018a60e3194e7e283cd89af28efd689623058a8 Mon Sep 17 00:00:00 2001 From: MORITA Kazutaka Date: Tue, 19 Apr 2011 21:01:25 +0900 Subject: Add support for creating a new volume from a existing snapshot with EC2 API. --- nova/api/ec2/cloud.py | 12 +++++- .../versions/016_add_snapshot_id_to_volumes.py | 48 ++++++++++++++++++++++ nova/db/sqlalchemy/models.py | 2 + nova/volume/api.py | 12 +++++- nova/volume/driver.py | 4 ++ nova/volume/manager.py | 9 +++- 6 files changed, 81 insertions(+), 6 deletions(-) create mode 100644 nova/db/sqlalchemy/migrate_repo/versions/016_add_snapshot_id_to_volumes.py diff --git a/nova/api/ec2/cloud.py b/nova/api/ec2/cloud.py index 6daf299b9..5d4d2ad27 100644 --- a/nova/api/ec2/cloud.py +++ b/nova/api/ec2/cloud.py @@ -639,11 +639,19 @@ class CloudController(object): v['display_description'] = volume['display_description'] return v - def create_volume(self, context, size, **kwargs): - LOG.audit(_("Create volume of %s GB"), size, context=context) + def create_volume(self, context, **kwargs): + size = kwargs.get('size'); + if kwargs.get('snapshot_id') != None: + snapshot_id = ec2utils.ec2_id_to_id(kwargs['snapshot_id']) + LOG.audit(_("Create volume from snapshot %s"), snapshot_id, context=context) + else: + snapshot_id = None + LOG.audit(_("Create volume of %s GB"), size, context=context) + volume = self.volume_api.create( context, size=size, + snapshot_id=snapshot_id, name=kwargs.get('display_name'), description=kwargs.get('display_description')) # TODO(vish): Instance should be None at db layer instead of diff --git a/nova/db/sqlalchemy/migrate_repo/versions/016_add_snapshot_id_to_volumes.py b/nova/db/sqlalchemy/migrate_repo/versions/016_add_snapshot_id_to_volumes.py new file mode 100644 index 000000000..0a50123bf --- /dev/null +++ b/nova/db/sqlalchemy/migrate_repo/versions/016_add_snapshot_id_to_volumes.py @@ -0,0 +1,48 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# Copyright 2011 MORITA Kazutaka. +# 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 sqlalchemy import * +from migrate import * + +from nova import log as logging + + +meta = MetaData() + + +# Table stub-definitions +# Just for the ForeignKey and column creation to succeed, these are not the +# actual definitions of instances or services. +# +volumes = Table('volumes', meta, + Column('id', Integer(), primary_key=True, nullable=False), + ) + +# +# New Column +# + +snapshot_id = Column('snapshot_id', Integer()) + + +def upgrade(migrate_engine): + # Upgrade operations go here. Don't create your own engine; + # bind migrate_engine to your metadata + meta.bind = migrate_engine + + # Add columns to existing tables + volumes.create_column(snapshot_id) diff --git a/nova/db/sqlalchemy/models.py b/nova/db/sqlalchemy/models.py index 9abe4d9ae..afc2ea4e4 100644 --- a/nova/db/sqlalchemy/models.py +++ b/nova/db/sqlalchemy/models.py @@ -287,6 +287,8 @@ class Volume(BASE, NovaBase): user_id = Column(String(255)) project_id = Column(String(255)) + snapshot_id = Column(String(255)) + host = Column(String(255)) # , ForeignKey('hosts.id')) size = Column(Integer) availability_zone = Column(String(255)) # TODO(vish): foreign key? diff --git a/nova/volume/api.py b/nova/volume/api.py index f5285f31f..bd073964d 100644 --- a/nova/volume/api.py +++ b/nova/volume/api.py @@ -39,7 +39,13 @@ LOG = logging.getLogger('nova.volume') class API(base.Base): """API for interacting with the volume manager.""" - def create(self, context, size, name, description): + def create(self, context, size, snapshot_id, name, description): + if snapshot_id != None: + snapshot = self.get_snapshot(context, snapshot_id) + if snapshot['status'] != "available": + raise exception.ApiError(_("Snapshot status must be available")) + size = snapshot['volume_size'] + if quota.allowed_volumes(context, 1, size) < 1: pid = context.project_id LOG.warn(_("Quota exceeeded for %(pid)s, tried to create" @@ -51,6 +57,7 @@ class API(base.Base): 'size': size, 'user_id': context.user_id, 'project_id': context.project_id, + 'snapshot_id': snapshot_id, 'availability_zone': FLAGS.storage_availability_zone, 'status': "creating", 'attach_status': "detached", @@ -62,7 +69,8 @@ class API(base.Base): FLAGS.scheduler_topic, {"method": "create_volume", "args": {"topic": FLAGS.volume_topic, - "volume_id": volume['id']}}) + "volume_id": volume['id'], + "snapshot_id": snapshot_id}}) return volume def delete(self, context, volume_id): diff --git a/nova/volume/driver.py b/nova/volume/driver.py index ba0a7efef..02b0d50f4 100644 --- a/nova/volume/driver.py +++ b/nova/volume/driver.py @@ -101,6 +101,10 @@ class VolumeDriver(object): volume['name'], FLAGS.volume_group) + def create_volume_from_snapshot(self, volume, snapshot): + """Creates a volume from a snapshot.""" + raise NotImplementedError() + def delete_volume(self, volume): """Deletes a logical volume.""" try: diff --git a/nova/volume/manager.py b/nova/volume/manager.py index 87fd3bf17..7d47fc191 100644 --- a/nova/volume/manager.py +++ b/nova/volume/manager.py @@ -90,7 +90,7 @@ class VolumeManager(manager.SchedulerDependentManager): else: LOG.info(_("volume %s: skipping export"), volume['name']) - def create_volume(self, context, volume_id): + def create_volume(self, context, volume_id, snapshot_id): """Creates and exports the volume.""" context = context.elevated() volume_ref = self.db.volume_get(context, volume_id) @@ -108,7 +108,12 @@ class VolumeManager(manager.SchedulerDependentManager): vol_size = volume_ref['size'] LOG.debug(_("volume %(vol_name)s: creating lv of" " size %(vol_size)sG") % locals()) - model_update = self.driver.create_volume(volume_ref) + if snapshot_id == None: + model_update = self.driver.create_volume(volume_ref) + else: + snapshot_ref = self.db.snapshot_get(context, snapshot_id) + model_update = self.driver.create_volume_from_snapshot(volume_ref, + snapshot_ref) if model_update: self.db.volume_update(context, volume_ref['id'], model_update) -- cgit From 1c7c53a9f40a88eb9def7ab9d706e7399ad5e65b Mon Sep 17 00:00:00 2001 From: MORITA Kazutaka Date: Tue, 19 Apr 2011 21:02:00 +0900 Subject: Add support for cloning a Sheepdog volume. --- nova/volume/driver.py | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/nova/volume/driver.py b/nova/volume/driver.py index 02b0d50f4..3f3caf37a 100644 --- a/nova/volume/driver.py +++ b/nova/volume/driver.py @@ -620,6 +620,13 @@ class SheepdogDriver(VolumeDriver): "sheepdog:%s" % volume['name'], sizestr) + def create_volume_from_snapshot(self, volume, snapshot): + """Creates a sheepdog volume from a snapshot.""" + self._try_execute('qemu-img', 'create', '-b', + "sheepdog:%s:%s" % (snapshot['volume_name'], snapshot['name']), + "sheepdog:%s" % volume['name']) + + def delete_volume(self, volume): """Deletes a logical volume""" self._try_execute('collie', 'vdi', 'delete', volume['name']) -- cgit From 5b670fe9bca9103642967bce609853704d0d1b88 Mon Sep 17 00:00:00 2001 From: MORITA Kazutaka Date: Tue, 19 Apr 2011 21:02:00 +0900 Subject: Add support for cloning a Sheepdog volume. --- nova/volume/driver.py | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/nova/volume/driver.py b/nova/volume/driver.py index 02b0d50f4..3f3caf37a 100644 --- a/nova/volume/driver.py +++ b/nova/volume/driver.py @@ -620,6 +620,13 @@ class SheepdogDriver(VolumeDriver): "sheepdog:%s" % volume['name'], sizestr) + def create_volume_from_snapshot(self, volume, snapshot): + """Creates a sheepdog volume from a snapshot.""" + self._try_execute('qemu-img', 'create', '-b', + "sheepdog:%s:%s" % (snapshot['volume_name'], snapshot['name']), + "sheepdog:%s" % volume['name']) + + def delete_volume(self, volume): """Deletes a logical volume""" self._try_execute('collie', 'vdi', 'delete', volume['name']) -- cgit From 389f7c79199d5ad908a72375a7377a1122f36707 Mon Sep 17 00:00:00 2001 From: Isaku Yamahata Date: Mon, 9 May 2011 17:52:26 +0900 Subject: volume/driver: factor out lvm opration Factor out lvm operation for implementing basic snapshot later. --- nova/volume/driver.py | 62 ++++++++++++++++++++++++++++----------------------- 1 file changed, 34 insertions(+), 28 deletions(-) diff --git a/nova/volume/driver.py b/nova/volume/driver.py index 3f3caf37a..9591c93d0 100644 --- a/nova/volume/driver.py +++ b/nova/volume/driver.py @@ -90,16 +90,40 @@ class VolumeDriver(object): raise exception.Error(_("volume group %s doesn't exist") % FLAGS.volume_group) + def _create_volume(self, volume_name, sizestr): + self._try_execute('sudo', 'lvcreate', '-L', sizestr, '-n', + volume_name, FLAGS.volume_group) + + def _copy_volume(self, srcstr, deststr, size_in_g): + self._execute('sudo', 'dd', 'if=%s' % srcstr, 'of=%s' % deststr, + 'count=%d' % (size_in_g * 1024), 'bs=1M') + + def _volume_not_present(self, volume_name): + path_name = '%s/%s' % (FLAGS.volume_group, volume_name) + try: + self._try_execute('sudo', 'lvdisplay', path_name) + except Exception as e: + # If the volume isn't present + return True + return False + + def _delete_volume(self, volume, size_in_g): + """Deletes a logical volume.""" + # zero out old volumes to prevent data leaking between users + # TODO(ja): reclaiming space should be done lazy and low priority + self._copy_volume('/dev/zero', self.local_path(volume), size_in_g) + self._try_execute('sudo', 'lvremove', '-f', "%s/%s" % + (FLAGS.volume_group, volume['name'])) + + def _sizestr(self, size_in_g): + if int(size_in_g) == 0: + return '100M' + return '%sG' % size_in_g + def create_volume(self, volume): """Creates a logical volume. Can optionally return a Dictionary of changes to the volume object to be persisted.""" - if int(volume['size']) == 0: - sizestr = '100M' - else: - sizestr = '%sG' % volume['size'] - self._try_execute('sudo', 'lvcreate', '-L', sizestr, '-n', - volume['name'], - FLAGS.volume_group) + self._create_volume(volume['name'], self._sizestr(volume['size'])) def create_volume_from_snapshot(self, volume, snapshot): """Creates a volume from a snapshot.""" @@ -107,24 +131,10 @@ class VolumeDriver(object): def delete_volume(self, volume): """Deletes a logical volume.""" - try: - self._try_execute('sudo', 'lvdisplay', - '%s/%s' % - (FLAGS.volume_group, - volume['name'])) - except Exception as e: + if self._volume_not_present(volume['name']): # If the volume isn't present, then don't attempt to delete return True - - # zero out old volumes to prevent data leaking between users - # TODO(ja): reclaiming space should be done lazy and low priority - self._execute('sudo', 'dd', 'if=/dev/zero', - 'of=%s' % self.local_path(volume), - 'count=%d' % (volume['size'] * 1024), - 'bs=1M') - self._try_execute('sudo', 'lvremove', '-f', "%s/%s" % - (FLAGS.volume_group, - volume['name'])) + self._delete_volume(volume, volume['size']) def create_snapshot(self, snapshot): """Creates a snapshot.""" @@ -612,13 +622,9 @@ class SheepdogDriver(VolumeDriver): def create_volume(self, volume): """Creates a sheepdog volume""" - if int(volume['size']) == 0: - sizestr = '100M' - else: - sizestr = '%sG' % volume['size'] self._try_execute('qemu-img', 'create', "sheepdog:%s" % volume['name'], - sizestr) + self._sizestr(volume['size'])) def create_volume_from_snapshot(self, volume, snapshot): """Creates a sheepdog volume from a snapshot.""" -- cgit From 03c735bb186a44d80a9d595e00e9c06fd8f709cc Mon Sep 17 00:00:00 2001 From: Isaku Yamahata Date: Mon, 9 May 2011 17:53:25 +0900 Subject: volume/driver: implement basic snapshot/clone added basic support for snapshot/clone to VolumeDriver. The implementation is not effective, but works. The effective implementation should be done by drived driver class. --- Authors | 1 + nova/exception.py | 6 ++++++ nova/volume/driver.py | 42 +++++++++++++++++++++++++++++++++++++----- nova/volume/manager.py | 6 ++++++ 4 files changed, 50 insertions(+), 5 deletions(-) diff --git a/Authors b/Authors index 60e1d2dad..9eae53e9b 100644 --- a/Authors +++ b/Authors @@ -28,6 +28,7 @@ Gabe Westmaas Hisaharu Ishii Hisaki Ohara Ilya Alekseyev +Isaku Yamahata Jason Koelker Jay Pipes Jesse Andrews diff --git a/nova/exception.py b/nova/exception.py index 2dffeb795..6748ef265 100644 --- a/nova/exception.py +++ b/nova/exception.py @@ -79,6 +79,12 @@ class VolumeNotFound(NotFound): super(VolumeNotFound, self).__init__(message) +class VolumeIsBusy(Error): + def __init__(self, message, volume_id): + self.volume_id = volume_id + super(Error, self).__init__(message) + + class SnapshotNotFound(NotFound): def __init__(self, message, snapshot_id): self.snapshot_id = snapshot_id diff --git a/nova/volume/driver.py b/nova/volume/driver.py index 9591c93d0..457a1c9e6 100644 --- a/nova/volume/driver.py +++ b/nova/volume/driver.py @@ -113,13 +113,21 @@ class VolumeDriver(object): # TODO(ja): reclaiming space should be done lazy and low priority self._copy_volume('/dev/zero', self.local_path(volume), size_in_g) self._try_execute('sudo', 'lvremove', '-f', "%s/%s" % - (FLAGS.volume_group, volume['name'])) + (FLAGS.volume_group, + self._escape_snapshot(volume['name']))) def _sizestr(self, size_in_g): if int(size_in_g) == 0: return '100M' return '%sG' % size_in_g + # Linux LVM reserves name that starts with snapshot, so that + # such volume name can't be created. Mangle it. + def _escape_snapshot(self, snapshot_name): + if not snapshot_name.startswith('snapshot'): + return snapshot_name + return '_' + snapshot_name + def create_volume(self, volume): """Creates a logical volume. Can optionally return a Dictionary of changes to the volume object to be persisted.""" @@ -127,27 +135,51 @@ class VolumeDriver(object): def create_volume_from_snapshot(self, volume, snapshot): """Creates a volume from a snapshot.""" - raise NotImplementedError() + self._create_volume(volume['name'], self._sizestr(volume['size'])) + self._copy_volume(self.local_path(snapshot), self.local_path(volume), + snapshot['volume_size']) def delete_volume(self, volume): """Deletes a logical volume.""" if self._volume_not_present(volume['name']): # If the volume isn't present, then don't attempt to delete return True + + # TODO(yamahata): lvm can't delete origin volume only without + # deleting derived snapshots. Can we do something fancy? + out, err = self._execute('sudo', 'lvdisplay', '--noheading', + '-C', '-o', 'Attr', + '%s/%s' % (FLAGS.volume_group, + volume['name'])) + out = out.strip() + if (out[0] == 'o') or (out[0] == 'O'): + raise exception.VolumeIsBusy( + _('deleting volume %s that has snapshot'), volume['name']) + self._delete_volume(volume, volume['size']) def create_snapshot(self, snapshot): """Creates a snapshot.""" - raise NotImplementedError() + orig_lv_name = "%s/%s" % (FLAGS.volume_group, snapshot['volume_name']) + self._try_execute('sudo', 'lvcreate', '-L', + self._sizestr(snapshot['volume_size']), + '--name', self._escape_snapshot(snapshot['name']), + '--snapshot', orig_lv_name) def delete_snapshot(self, snapshot): """Deletes a snapshot.""" - raise NotImplementedError() + if self._volume_not_present(self._escape_snapshot(snapshot['name'])): + # If the snapshot isn't present, then don't attempt to delete + return True + + # TODO(yamahata): zeroing out the whole snapshot triggers COW. + # it's quite slow. + self._delete_volume(snapshot, snapshot['volume_size']) def local_path(self, volume): # NOTE(vish): stops deprecation warning escaped_group = FLAGS.volume_group.replace('-', '--') - escaped_name = volume['name'].replace('-', '--') + escaped_name = self._escape_snapshot(volume['name']).replace('-', '--') return "/dev/mapper/%s-%s" % (escaped_group, escaped_name) def ensure_export(self, context, volume): diff --git a/nova/volume/manager.py b/nova/volume/manager.py index 7d47fc191..84085fbd8 100644 --- a/nova/volume/manager.py +++ b/nova/volume/manager.py @@ -147,6 +147,12 @@ class VolumeManager(manager.SchedulerDependentManager): self.driver.remove_export(context, volume_ref) LOG.debug(_("volume %s: deleting"), volume_ref['name']) self.driver.delete_volume(volume_ref) + except exception.VolumeIsBusy, e: + LOG.debug(_("volume %s: volume is busy"), volume_ref['name']) + self.driver.ensure_export(context, volume_ref) + self.db.volume_update(context, volume_ref['id'], + {'status': 'available'}) + return True except Exception: self.db.volume_update(context, volume_ref['id'], -- cgit From db148f108dfc4829e1302a54fe4f57ab81212786 Mon Sep 17 00:00:00 2001 From: Isaku Yamahata Date: Mon, 9 May 2011 19:25:02 +0900 Subject: fix mismerge by 1059 --- nova/db/sqlalchemy/api.py | 3 +-- nova/exception.py | 65 ++++++----------------------------------------- nova/volume/driver.py | 3 +-- 3 files changed, 10 insertions(+), 61 deletions(-) diff --git a/nova/db/sqlalchemy/api.py b/nova/db/sqlalchemy/api.py index ebdb2ad5c..7302f25b0 100644 --- a/nova/db/sqlalchemy/api.py +++ b/nova/db/sqlalchemy/api.py @@ -1798,8 +1798,7 @@ def snapshot_get(context, snapshot_id, session=None): filter_by(deleted=False).\ first() if not result: - raise exception.SnapshotNotFound(_('Snapshot %s not found') % snapshot_id, - snapshot_id) + raise exception.SnapshotNotFound(snapshot_id=snapshot_id) return result diff --git a/nova/exception.py b/nova/exception.py index 6748ef265..b16ea6810 100644 --- a/nova/exception.py +++ b/nova/exception.py @@ -60,65 +60,8 @@ class ApiError(Error): class BuildInProgress(Error): - super(ApiError, self).__init__('%s: %s' % (code, message)) - - -class NotFound(Error): - pass - - -class InstanceNotFound(NotFound): - def __init__(self, message, instance_id): - self.instance_id = instance_id - super(InstanceNotFound, self).__init__(message) - - -class VolumeNotFound(NotFound): - def __init__(self, message, volume_id): - self.volume_id = volume_id - super(VolumeNotFound, self).__init__(message) - - -class VolumeIsBusy(Error): - def __init__(self, message, volume_id): - self.volume_id = volume_id - super(Error, self).__init__(message) - - -class SnapshotNotFound(NotFound): - def __init__(self, message, snapshot_id): - self.snapshot_id = snapshot_id - super(SnapshotNotFound, self).__init__(message) - - -class Duplicate(Error): pass - -class NotAuthorized(Error): - pass - - -class NotEmpty(Error): - pass - - -class Invalid(Error): - pass - - -class InvalidInputException(Error): - pass - - -class InvalidContentType(Error): - pass - - -class TimeoutException(Error): - pass - - class DBError(Error): """Wraps an implementation specific exception.""" def __init__(self, inner_exception): @@ -319,6 +262,14 @@ class VolumeNotFoundForInstance(VolumeNotFound): message = _("Volume not found for instance %(instance_id)s.") +class SnapshotNotFound(NotFound): + message = _("Snapshot %(snapshot_id)s not found") + + +class VolumeIsBusy(Error): + message = _("deleting volume %(volume_name)s that has snapshot") + + class ExportDeviceNotFoundForVolume(NotFound): message = _("No export device found for volume %(volume_id)s.") diff --git a/nova/volume/driver.py b/nova/volume/driver.py index 457a1c9e6..e783d3a5a 100644 --- a/nova/volume/driver.py +++ b/nova/volume/driver.py @@ -153,8 +153,7 @@ class VolumeDriver(object): volume['name'])) out = out.strip() if (out[0] == 'o') or (out[0] == 'O'): - raise exception.VolumeIsBusy( - _('deleting volume %s that has snapshot'), volume['name']) + raise exception.VolumeIsBusy(volume_name=volume['name']) self._delete_volume(volume, volume['size']) -- cgit From c5dbee818b1a06bf5358c32197c8e15ecf0f660d Mon Sep 17 00:00:00 2001 From: Isaku Yamahata Date: Mon, 9 May 2011 20:19:35 +0900 Subject: db: fix db versioning --- .../versions/015_add_volume_snapshot_support.py | 71 ---------------------- .../versions/016_add_snapshot_id_to_volumes.py | 48 --------------- .../versions/016_add_volume_snapshot_support.py | 71 ++++++++++++++++++++++ .../versions/017_add_snapshot_id_to_volumes.py | 48 +++++++++++++++ 4 files changed, 119 insertions(+), 119 deletions(-) delete mode 100644 nova/db/sqlalchemy/migrate_repo/versions/015_add_volume_snapshot_support.py delete mode 100644 nova/db/sqlalchemy/migrate_repo/versions/016_add_snapshot_id_to_volumes.py create mode 100644 nova/db/sqlalchemy/migrate_repo/versions/016_add_volume_snapshot_support.py create mode 100644 nova/db/sqlalchemy/migrate_repo/versions/017_add_snapshot_id_to_volumes.py diff --git a/nova/db/sqlalchemy/migrate_repo/versions/015_add_volume_snapshot_support.py b/nova/db/sqlalchemy/migrate_repo/versions/015_add_volume_snapshot_support.py deleted file mode 100644 index 288f63e72..000000000 --- a/nova/db/sqlalchemy/migrate_repo/versions/015_add_volume_snapshot_support.py +++ /dev/null @@ -1,71 +0,0 @@ -# vim: tabstop=4 shiftwidth=4 softtabstop=4 - -# Copyright 2011 MORITA Kazutaka. -# 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 sqlalchemy import * -from migrate import * - -from nova import log as logging - -meta = MetaData() - -snapshots = Table('snapshots', meta, - Column('created_at', DateTime(timezone=False)), - Column('updated_at', DateTime(timezone=False)), - Column('deleted_at', DateTime(timezone=False)), - Column('deleted', Boolean(create_constraint=True, name=None)), - Column('id', Integer(), primary_key=True, nullable=False), - Column('volume_id', Integer(), nullable=False), - Column('user_id', - String(length=255, convert_unicode=False, assert_unicode=None, - unicode_error=None, _warn_on_bytestring=False)), - Column('project_id', - String(length=255, convert_unicode=False, assert_unicode=None, - unicode_error=None, _warn_on_bytestring=False)), - Column('status', - String(length=255, convert_unicode=False, assert_unicode=None, - unicode_error=None, _warn_on_bytestring=False)), - Column('progress', - String(length=255, convert_unicode=False, assert_unicode=None, - unicode_error=None, _warn_on_bytestring=False)), - Column('volume_size', Integer()), - Column('scheduled_at', DateTime(timezone=False)), - Column('display_name', - String(length=255, convert_unicode=False, assert_unicode=None, - unicode_error=None, _warn_on_bytestring=False)), - Column('display_description', - String(length=255, convert_unicode=False, assert_unicode=None, - unicode_error=None, _warn_on_bytestring=False)) - ) - - -def upgrade(migrate_engine): - # Upgrade operations go here. Don't create your own engine; - # bind migrate_engine to your metadata - meta.bind = migrate_engine - - try: - snapshots.create() - except Exception: - logging.info(repr(snapshots)) - logging.exception('Exception while creating table') - meta.drop_all(tables=[snapshots]) - raise - - -def downgrade(migrate_engine): - # Operations to reverse the above upgrade go here. - snapshots.drop() diff --git a/nova/db/sqlalchemy/migrate_repo/versions/016_add_snapshot_id_to_volumes.py b/nova/db/sqlalchemy/migrate_repo/versions/016_add_snapshot_id_to_volumes.py deleted file mode 100644 index 0a50123bf..000000000 --- a/nova/db/sqlalchemy/migrate_repo/versions/016_add_snapshot_id_to_volumes.py +++ /dev/null @@ -1,48 +0,0 @@ -# vim: tabstop=4 shiftwidth=4 softtabstop=4 - -# Copyright 2011 MORITA Kazutaka. -# 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 sqlalchemy import * -from migrate import * - -from nova import log as logging - - -meta = MetaData() - - -# Table stub-definitions -# Just for the ForeignKey and column creation to succeed, these are not the -# actual definitions of instances or services. -# -volumes = Table('volumes', meta, - Column('id', Integer(), primary_key=True, nullable=False), - ) - -# -# New Column -# - -snapshot_id = Column('snapshot_id', Integer()) - - -def upgrade(migrate_engine): - # Upgrade operations go here. Don't create your own engine; - # bind migrate_engine to your metadata - meta.bind = migrate_engine - - # Add columns to existing tables - volumes.create_column(snapshot_id) diff --git a/nova/db/sqlalchemy/migrate_repo/versions/016_add_volume_snapshot_support.py b/nova/db/sqlalchemy/migrate_repo/versions/016_add_volume_snapshot_support.py new file mode 100644 index 000000000..288f63e72 --- /dev/null +++ b/nova/db/sqlalchemy/migrate_repo/versions/016_add_volume_snapshot_support.py @@ -0,0 +1,71 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# Copyright 2011 MORITA Kazutaka. +# 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 sqlalchemy import * +from migrate import * + +from nova import log as logging + +meta = MetaData() + +snapshots = Table('snapshots', meta, + Column('created_at', DateTime(timezone=False)), + Column('updated_at', DateTime(timezone=False)), + Column('deleted_at', DateTime(timezone=False)), + Column('deleted', Boolean(create_constraint=True, name=None)), + Column('id', Integer(), primary_key=True, nullable=False), + Column('volume_id', Integer(), nullable=False), + Column('user_id', + String(length=255, convert_unicode=False, assert_unicode=None, + unicode_error=None, _warn_on_bytestring=False)), + Column('project_id', + String(length=255, convert_unicode=False, assert_unicode=None, + unicode_error=None, _warn_on_bytestring=False)), + Column('status', + String(length=255, convert_unicode=False, assert_unicode=None, + unicode_error=None, _warn_on_bytestring=False)), + Column('progress', + String(length=255, convert_unicode=False, assert_unicode=None, + unicode_error=None, _warn_on_bytestring=False)), + Column('volume_size', Integer()), + Column('scheduled_at', DateTime(timezone=False)), + Column('display_name', + String(length=255, convert_unicode=False, assert_unicode=None, + unicode_error=None, _warn_on_bytestring=False)), + Column('display_description', + String(length=255, convert_unicode=False, assert_unicode=None, + unicode_error=None, _warn_on_bytestring=False)) + ) + + +def upgrade(migrate_engine): + # Upgrade operations go here. Don't create your own engine; + # bind migrate_engine to your metadata + meta.bind = migrate_engine + + try: + snapshots.create() + except Exception: + logging.info(repr(snapshots)) + logging.exception('Exception while creating table') + meta.drop_all(tables=[snapshots]) + raise + + +def downgrade(migrate_engine): + # Operations to reverse the above upgrade go here. + snapshots.drop() diff --git a/nova/db/sqlalchemy/migrate_repo/versions/017_add_snapshot_id_to_volumes.py b/nova/db/sqlalchemy/migrate_repo/versions/017_add_snapshot_id_to_volumes.py new file mode 100644 index 000000000..0a50123bf --- /dev/null +++ b/nova/db/sqlalchemy/migrate_repo/versions/017_add_snapshot_id_to_volumes.py @@ -0,0 +1,48 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# Copyright 2011 MORITA Kazutaka. +# 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 sqlalchemy import * +from migrate import * + +from nova import log as logging + + +meta = MetaData() + + +# Table stub-definitions +# Just for the ForeignKey and column creation to succeed, these are not the +# actual definitions of instances or services. +# +volumes = Table('volumes', meta, + Column('id', Integer(), primary_key=True, nullable=False), + ) + +# +# New Column +# + +snapshot_id = Column('snapshot_id', Integer()) + + +def upgrade(migrate_engine): + # Upgrade operations go here. Don't create your own engine; + # bind migrate_engine to your metadata + meta.bind = migrate_engine + + # Add columns to existing tables + volumes.create_column(snapshot_id) -- cgit From 321d724df5f0c4ca008da2a08ef279ca18df0733 Mon Sep 17 00:00:00 2001 From: Isaku Yamahata Date: Fri, 13 May 2011 23:07:34 +0900 Subject: Authors: add myself to Authers file add myself to Authers file for later commit. --- Authors | 1 + 1 file changed, 1 insertion(+) diff --git a/Authors b/Authors index 8b54240c1..baf8fde4c 100644 --- a/Authors +++ b/Authors @@ -28,6 +28,7 @@ Gabe Westmaas Hisaharu Ishii Hisaki Ohara Ilya Alekseyev +Isaku Yamahata Jason Koelker Jay Pipes Jesse Andrews -- cgit From bbbea57cf6ab28c3ad1081041275e0d6d2bbd308 Mon Sep 17 00:00:00 2001 From: Isaku Yamahata Date: Fri, 13 May 2011 23:08:57 +0900 Subject: volume/driver: factor out lvm opration Factor out lvm operation for implementing basic snapshot later. --- nova/volume/driver.py | 62 ++++++++++++++++++++++++++++----------------------- 1 file changed, 34 insertions(+), 28 deletions(-) diff --git a/nova/volume/driver.py b/nova/volume/driver.py index ba0a7efef..ec7be37bf 100644 --- a/nova/volume/driver.py +++ b/nova/volume/driver.py @@ -90,37 +90,47 @@ class VolumeDriver(object): raise exception.Error(_("volume group %s doesn't exist") % FLAGS.volume_group) - def create_volume(self, volume): - """Creates a logical volume. Can optionally return a Dictionary of - changes to the volume object to be persisted.""" - if int(volume['size']) == 0: - sizestr = '100M' - else: - sizestr = '%sG' % volume['size'] + def _create_volume(self, volume_name, sizestr): self._try_execute('sudo', 'lvcreate', '-L', sizestr, '-n', - volume['name'], - FLAGS.volume_group) - - def delete_volume(self, volume): - """Deletes a logical volume.""" + volume_name, FLAGS.volume_group) + + def _copy_volume(self, srcstr, deststr, size_in_g): + self._execute('sudo', 'dd', 'if=%s' % srcstr, 'of=%s' % deststr, + 'count=%d' % (size_in_g * 1024), 'bs=1M') + + def _volume_not_present(self, volume_name): + path_name = '%s/%s' % (FLAGS.volume_group, volume_name) try: - self._try_execute('sudo', 'lvdisplay', - '%s/%s' % - (FLAGS.volume_group, - volume['name'])) + self._try_execute('sudo', 'lvdisplay', path_name) except Exception as e: - # If the volume isn't present, then don't attempt to delete + # If the volume isn't present return True + return False + def _delete_volume(self, volume, size_in_g): + """Deletes a logical volume.""" # zero out old volumes to prevent data leaking between users # TODO(ja): reclaiming space should be done lazy and low priority - self._execute('sudo', 'dd', 'if=/dev/zero', - 'of=%s' % self.local_path(volume), - 'count=%d' % (volume['size'] * 1024), - 'bs=1M') + self._copy_volume('/dev/zero', self.local_path(volume), size_in_g) self._try_execute('sudo', 'lvremove', '-f', "%s/%s" % - (FLAGS.volume_group, - volume['name'])) + (FLAGS.volume_group, volume['name'])) + + def _sizestr(self, size_in_g): + if int(size_in_g) == 0: + return '100M' + return '%sG' % size_in_g + + def create_volume(self, volume): + """Creates a logical volume. Can optionally return a Dictionary of + changes to the volume object to be persisted.""" + self._create_volume(volume['name'], self._sizestr(volume['size'])) + + def delete_volume(self, volume): + """Deletes a logical volume.""" + if self._volume_not_present(volume['name']): + # If the volume isn't present, then don't attempt to delete + return True + self._delete_volume(volume, volume['size']) def create_snapshot(self, snapshot): """Creates a snapshot.""" @@ -608,13 +618,9 @@ class SheepdogDriver(VolumeDriver): def create_volume(self, volume): """Creates a sheepdog volume""" - if int(volume['size']) == 0: - sizestr = '100M' - else: - sizestr = '%sG' % volume['size'] self._try_execute('qemu-img', 'create', "sheepdog:%s" % volume['name'], - sizestr) + self._sizestr(volume['size'])) def delete_volume(self, volume): """Deletes a logical volume""" -- cgit From 4f7cfba4a00f04b7c30c61da2946f183241a7c7f Mon Sep 17 00:00:00 2001 From: Isaku Yamahata Date: Fri, 13 May 2011 23:27:35 +0900 Subject: volume/driver: implement basic snapshot added basic support for snapshot to VolumeDriver base class. The implementation is not effective, but works. The effective implementation should be done by drived driver class. --- nova/exception.py | 4 ++++ nova/volume/driver.py | 37 +++++++++++++++++++++++++++++++++---- nova/volume/manager.py | 6 ++++++ 3 files changed, 43 insertions(+), 4 deletions(-) diff --git a/nova/exception.py b/nova/exception.py index 39620ccc1..bd04435ed 100644 --- a/nova/exception.py +++ b/nova/exception.py @@ -271,6 +271,10 @@ class SnapshotNotFound(NotFound): message = _("Snapshot %(snapshot_id)s could not be found.") +class VolumeIsBusy(Error): + message = _("deleting volume %(volume_name)s that has snapshot") + + class ExportDeviceNotFoundForVolume(NotFound): message = _("No export device found for volume %(volume_id)s.") diff --git a/nova/volume/driver.py b/nova/volume/driver.py index ec7be37bf..a6cf2cb46 100644 --- a/nova/volume/driver.py +++ b/nova/volume/driver.py @@ -113,13 +113,21 @@ class VolumeDriver(object): # TODO(ja): reclaiming space should be done lazy and low priority self._copy_volume('/dev/zero', self.local_path(volume), size_in_g) self._try_execute('sudo', 'lvremove', '-f', "%s/%s" % - (FLAGS.volume_group, volume['name'])) + (FLAGS.volume_group, + self._escape_snapshot(volume['name']))) def _sizestr(self, size_in_g): if int(size_in_g) == 0: return '100M' return '%sG' % size_in_g + # Linux LVM reserves name that starts with snapshot, so that + # such volume name can't be created. Mangle it. + def _escape_snapshot(self, snapshot_name): + if not snapshot_name.startswith('snapshot'): + return snapshot_name + return '_' + snapshot_name + def create_volume(self, volume): """Creates a logical volume. Can optionally return a Dictionary of changes to the volume object to be persisted.""" @@ -130,20 +138,41 @@ class VolumeDriver(object): if self._volume_not_present(volume['name']): # If the volume isn't present, then don't attempt to delete return True + + # TODO(yamahata): lvm can't delete origin volume only without + # deleting derived snapshots. Can we do something fancy? + out, err = self._execute('sudo', 'lvdisplay', '--noheading', + '-C', '-o', 'Attr', + '%s/%s' % (FLAGS.volume_group, + volume['name'])) + out = out.strip() + if (out[0] == 'o') or (out[0] == 'O'): + raise exception.VolumeIsBusy(volume_name=volume['name']) + self._delete_volume(volume, volume['size']) def create_snapshot(self, snapshot): """Creates a snapshot.""" - raise NotImplementedError() + orig_lv_name = "%s/%s" % (FLAGS.volume_group, snapshot['volume_name']) + self._try_execute('sudo', 'lvcreate', '-L', + self._sizestr(snapshot['volume_size']), + '--name', self._escape_snapshot(snapshot['name']), + '--snapshot', orig_lv_name) def delete_snapshot(self, snapshot): """Deletes a snapshot.""" - raise NotImplementedError() + if self._volume_not_present(self._escape_snapshot(snapshot['name'])): + # If the snapshot isn't present, then don't attempt to delete + return True + + # TODO(yamahata): zeroing out the whole snapshot triggers COW. + # it's quite slow. + self._delete_volume(snapshot, snapshot['volume_size']) def local_path(self, volume): # NOTE(vish): stops deprecation warning escaped_group = FLAGS.volume_group.replace('-', '--') - escaped_name = volume['name'].replace('-', '--') + escaped_name = self._escape_snapshot(volume['name']).replace('-', '--') return "/dev/mapper/%s-%s" % (escaped_group, escaped_name) def ensure_export(self, context, volume): diff --git a/nova/volume/manager.py b/nova/volume/manager.py index 87fd3bf17..fd889633d 100644 --- a/nova/volume/manager.py +++ b/nova/volume/manager.py @@ -142,6 +142,12 @@ class VolumeManager(manager.SchedulerDependentManager): self.driver.remove_export(context, volume_ref) LOG.debug(_("volume %s: deleting"), volume_ref['name']) self.driver.delete_volume(volume_ref) + except exception.VolumeIsBusy, e: + LOG.debug(_("volume %s: volume is busy"), volume_ref['name']) + self.driver.ensure_export(context, volume_ref) + self.db.volume_update(context, volume_ref['id'], + {'status': 'available'}) + return True except Exception: self.db.volume_update(context, volume_ref['id'], -- cgit From aaec8400be701c674bbf89badd59ee9468827ed9 Mon Sep 17 00:00:00 2001 From: Isaku Yamahata Date: Sat, 14 May 2011 01:42:26 +0900 Subject: volume/driver: make unit test, test_volume, pass fake command executer doesn't return command result. Which return None instead of string. So add None check to make unit test pass. --- nova/volume/driver.py | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/nova/volume/driver.py b/nova/volume/driver.py index a6cf2cb46..0807ff476 100644 --- a/nova/volume/driver.py +++ b/nova/volume/driver.py @@ -145,9 +145,11 @@ class VolumeDriver(object): '-C', '-o', 'Attr', '%s/%s' % (FLAGS.volume_group, volume['name'])) - out = out.strip() - if (out[0] == 'o') or (out[0] == 'O'): - raise exception.VolumeIsBusy(volume_name=volume['name']) + # fake_execute returns None resulting unit test error + if out: + out = out.strip() + if (out[0] == 'o') or (out[0] == 'O'): + raise exception.VolumeIsBusy(volume_name=volume['name']) self._delete_volume(volume, volume['size']) -- cgit From 8b86fb3a4d9ee3e328232c0051b9daff6838d00d Mon Sep 17 00:00:00 2001 From: Josh Durgin Date: Fri, 13 May 2011 10:26:13 -0700 Subject: Add support for rbd snapshots. --- nova/volume/driver.py | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/nova/volume/driver.py b/nova/volume/driver.py index 0807ff476..e0e18b9bf 100644 --- a/nova/volume/driver.py +++ b/nova/volume/driver.py @@ -608,6 +608,18 @@ class RBDDriver(VolumeDriver): self._try_execute('rbd', '--pool', FLAGS.rbd_pool, 'rm', volume['name']) + def create_snapshot(self, snapshot): + """Creates an rbd snapshot""" + self._try_execute('rbd', '--pool', FLAGS.rbd_pool, + 'snap', 'create', '--snap', snapshot['name'], + snapshot['volume_name']) + + def delete_snapshot(self, snapshot): + """Deletes an rbd snapshot""" + self._try_execute('rbd', '--pool', FLAGS.rbd_pool, + 'snap', 'rm', '--snap', snapshot['name'], + snapshot['volume_name']) + def local_path(self, volume): """Returns the path of the rbd volume.""" # This is the same as the remote path -- cgit From 5eb57c6191ac7c8d98539eb3967ceb00f7c55daf Mon Sep 17 00:00:00 2001 From: MORITA Kazutaka Date: Mon, 16 May 2011 16:29:21 +0900 Subject: Add a unit test for snapshot_volume. --- nova/tests/test_volume.py | 27 +++++++++++++++++++++++++++ 1 file changed, 27 insertions(+) diff --git a/nova/tests/test_volume.py b/nova/tests/test_volume.py index 236d12434..c66b66959 100644 --- a/nova/tests/test_volume.py +++ b/nova/tests/test_volume.py @@ -176,6 +176,33 @@ class VolumeTestCase(test.TestCase): # This will allow us to test cross-node interactions pass + @staticmethod + def _create_snapshot(volume_id, size='0'): + """Create a snapshot object.""" + snap = {} + snap['volume_size'] = size + snap['user_id'] = 'fake' + snap['project_id'] = 'fake' + snap['volume_id'] = volume_id + snap['status'] = "creating" + return db.snapshot_create(context.get_admin_context(), snap)['id'] + + def test_create_delete_snapshot(self): + """Test snapshot can be created and deleted.""" + volume_id = self._create_volume() + self.volume.create_volume(self.context, volume_id) + snapshot_id = self._create_snapshot(volume_id) + self.volume.create_snapshot(self.context, volume_id, snapshot_id) + self.assertEqual(snapshot_id, db.snapshot_get(context.get_admin_context(), + snapshot_id).id) + + self.volume.delete_snapshot(self.context, snapshot_id) + self.assertRaises(exception.NotFound, + db.snapshot_get, + self.context, + snapshot_id) + self.volume.delete_volume(self.context, volume_id) + class DriverTestCase(test.TestCase): """Base Test class for Drivers.""" -- cgit From d44299be90bbfcac5f8de1e1264b81fbb0bfa5e2 Mon Sep 17 00:00:00 2001 From: Masanori Itoh Date: Tue, 17 May 2011 01:00:16 +0900 Subject: Add vnc_keymap flag and enable setting keymap for vnc console. --- nova/virt/libvirt.xml.template | 2 +- nova/virt/libvirt_conn.py | 1 + nova/vnc/__init__.py | 2 ++ 3 files changed, 4 insertions(+), 1 deletion(-) diff --git a/nova/virt/libvirt.xml.template b/nova/virt/libvirt.xml.template index de2497a76..20986d4d5 100644 --- a/nova/virt/libvirt.xml.template +++ b/nova/virt/libvirt.xml.template @@ -116,7 +116,7 @@ #if $getVar('vncserver_host', False) - + #end if diff --git a/nova/virt/libvirt_conn.py b/nova/virt/libvirt_conn.py index 555e44ce2..7552c9488 100644 --- a/nova/virt/libvirt_conn.py +++ b/nova/virt/libvirt_conn.py @@ -1022,6 +1022,7 @@ class LibvirtConnection(driver.ComputeDriver): if FLAGS.vnc_enabled: if FLAGS.libvirt_type != 'lxc': xml_info['vncserver_host'] = FLAGS.vncserver_host + xml_info['vnc_keymap'] = FLAGS.vnc_keymap if not rescue: if instance['kernel_id']: xml_info['kernel'] = xml_info['basepath'] + "/kernel" diff --git a/nova/vnc/__init__.py b/nova/vnc/__init__.py index b5b00e44e..859bfd65f 100644 --- a/nova/vnc/__init__.py +++ b/nova/vnc/__init__.py @@ -32,3 +32,5 @@ flags.DEFINE_string('vncserver_host', '0.0.0.0', 'the host interface on which vnc server should listen') flags.DEFINE_bool('vnc_enabled', True, 'enable vnc related features') +flags.DEFINE_string('vnc_keymap', 'en-us', + 'keymap for vnc') -- cgit From 0b698186b56af6580633dedd7916df2897945f29 Mon Sep 17 00:00:00 2001 From: MORITA Kazutaka Date: Thu, 19 May 2011 21:31:14 +0900 Subject: Avoid wildcard import. --- .../migrate_repo/versions/019_add_volume_snapshot_support.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/nova/db/sqlalchemy/migrate_repo/versions/019_add_volume_snapshot_support.py b/nova/db/sqlalchemy/migrate_repo/versions/019_add_volume_snapshot_support.py index 288f63e72..5a44bac16 100644 --- a/nova/db/sqlalchemy/migrate_repo/versions/019_add_volume_snapshot_support.py +++ b/nova/db/sqlalchemy/migrate_repo/versions/019_add_volume_snapshot_support.py @@ -15,8 +15,8 @@ # License for the specific language governing permissions and limitations # under the License. -from sqlalchemy import * -from migrate import * +from sqlalchemy import Column, Table, MetaData +from sqlalchemy import Integer, DateTime, Boolean, String from nova import log as logging -- cgit From a4cc51b78ae5e08227bef7a4be52953776a3e947 Mon Sep 17 00:00:00 2001 From: MORITA Kazutaka Date: Thu, 19 May 2011 21:49:15 +0900 Subject: Add a unitest to test EC2 snapshot APIs. --- nova/tests/test_cloud.py | 46 ++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 46 insertions(+) diff --git a/nova/tests/test_cloud.py b/nova/tests/test_cloud.py index c8559615a..d9169a646 100644 --- a/nova/tests/test_cloud.py +++ b/nova/tests/test_cloud.py @@ -188,6 +188,52 @@ class CloudTestCase(test.TestCase): db.service_destroy(self.context, service1['id']) db.service_destroy(self.context, service2['id']) + def test_describe_snapshots(self): + """Makes sure describe_snapshots works and filters results.""" + vol = db.volume_create(self.context, {}) + snap1 = db.snapshot_create(self.context, {'volume_id': vol['id']}) + snap2 = db.snapshot_create(self.context, {'volume_id': vol['id']}) + result = self.cloud.describe_snapshots(self.context) + self.assertEqual(len(result['snapshotSet']), 2) + snapshot_id = ec2utils.id_to_ec2_id(snap2['id'], 'snap-%08x') + result = self.cloud.describe_snapshots(self.context, + snapshot_id=[snapshot_id]) + self.assertEqual(len(result['snapshotSet']), 1) + self.assertEqual( + ec2utils.ec2_id_to_id(result['snapshotSet'][0]['snapshotId']), + snap2['id']) + db.snapshot_destroy(self.context, snap1['id']) + db.snapshot_destroy(self.context, snap2['id']) + db.volume_destroy(self.context, vol['id']) + + def test_create_snapshot(self): + """Makes sure create_snapshot works.""" + vol = db.volume_create(self.context, {'status': "available"}) + volume_id = ec2utils.id_to_ec2_id(vol['id'], 'vol-%08x') + + result = self.cloud.create_snapshot(self.context, + volume_id=volume_id) + snapshot_id = result['snapshotId'] + result = self.cloud.describe_snapshots(self.context) + self.assertEqual(len(result['snapshotSet']), 1) + self.assertEqual(result['snapshotSet'][0]['snapshotId'], snapshot_id) + + db.snapshot_destroy(self.context, ec2utils.ec2_id_to_id(snapshot_id)) + db.volume_destroy(self.context, vol['id']) + + def test_delete_snapshot(self): + """Makes sure delete_snapshot works.""" + vol = db.volume_create(self.context, {'status': "available"}) + snap = db.snapshot_create(self.context, {'volume_id': vol['id'], + 'status': "available"}) + snapshot_id = ec2utils.id_to_ec2_id(snap['id'], 'snap-%08x') + + result = self.cloud.delete_snapshot(self.context, + snapshot_id=snapshot_id) + self.assertTrue(result) + + db.volume_destroy(self.context, vol['id']) + def test_describe_instances(self): """Makes sure describe_instances works and filters results.""" inst1 = db.instance_create(self.context, {'reservation_id': 'a', -- cgit From c04a59fefbcbd0e5e21cbc8c70eb3147785cf22d Mon Sep 17 00:00:00 2001 From: MORITA Kazutaka Date: Thu, 19 May 2011 22:06:18 +0900 Subject: Fix comments. --- nova/db/api.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/nova/db/api.py b/nova/db/api.py index 1ef82b461..3597732b9 100644 --- a/nova/db/api.py +++ b/nova/db/api.py @@ -884,27 +884,27 @@ def volume_update(context, volume_id, values): def snapshot_create(context, values): - """Create a volume from the values dictionary.""" + """Create a snapshot from the values dictionary.""" return IMPL.snapshot_create(context, values) def snapshot_destroy(context, snapshot_id): - """Create a volume from the values dictionary.""" + """Destroy the snapshot or raise if it does not exist.""" return IMPL.snapshot_destroy(context, snapshot_id) def snapshot_get(context, snapshot_id): - """Get a volume or raise if it does not exist.""" + """Get a snapshot or raise if it does not exist.""" return IMPL.snapshot_get(context, snapshot_id) def snapshot_get_all(context): - """Get all volumes.""" + """Get all snapshots.""" return IMPL.snapshot_get_all(context) def snapshot_get_all_by_project(context, project_id): - """Get all volumes belonging to a project.""" + """Get all snapshots belonging to a project.""" return IMPL.snapshot_get_all_by_project(context, project_id) -- cgit From ffac2aa8162ba5111a01b495d9dd7e43bfda4af4 Mon Sep 17 00:00:00 2001 From: Chris Behrens Date: Mon, 23 May 2011 14:38:37 -0500 Subject: initial fudging in of swap disk --- nova/tests/xenapi/stubs.py | 2 +- nova/virt/xenapi/vm_utils.py | 18 ++++++++++++------ nova/virt/xenapi/vmops.py | 21 +++++++++++++-------- plugins/xenserver/xenapi/etc/xapi.d/plugins/glance | 12 +++++++++--- 4 files changed, 35 insertions(+), 18 deletions(-) diff --git a/nova/tests/xenapi/stubs.py b/nova/tests/xenapi/stubs.py index 4833ccb07..d9306900d 100644 --- a/nova/tests/xenapi/stubs.py +++ b/nova/tests/xenapi/stubs.py @@ -37,7 +37,7 @@ def stubout_instance_snapshot(stubs): sr_ref=sr_ref, sharable=False) vdi_rec = session.get_xenapi().VDI.get_record(vdi_ref) vdi_uuid = vdi_rec['uuid'] - return vdi_uuid + return dict(primary_vdi_uuid=vdi_uuid, swap_vdi_uuid=None) stubs.Set(vm_utils.VMHelper, 'fetch_image', fake_fetch_image) diff --git a/nova/virt/xenapi/vm_utils.py b/nova/virt/xenapi/vm_utils.py index 9f6cd608c..c24fc7ba6 100644 --- a/nova/virt/xenapi/vm_utils.py +++ b/nova/virt/xenapi/vm_utils.py @@ -408,18 +408,24 @@ class VMHelper(HelperBase): kwargs = {'params': pickle.dumps(params)} task = session.async_call_plugin('glance', 'download_vhd', kwargs) - vdi_uuid = session.wait_for_task(task, instance_id) + vdi_uuids = session.wait_for_task(task, instance_id) + primary_vdi_uuid = vdi_uuids.get('primary_vdi_uuid') + swap_vdi_uuid = vdi_uuids.get('swap_vdi_uuid') cls.scan_sr(session, instance_id, sr_ref) # Set the name-label to ease debugging - vdi_ref = session.get_xenapi().VDI.get_by_uuid(vdi_uuid) - name_label = get_name_label_for_image(image) - session.get_xenapi().VDI.set_name_label(vdi_ref, name_label) + primary_vdi_ref = session.get_xenapi().VDI.get_by_uuid(primary_vdi_uuid) + primary_name_label = get_name_label_for_image(image) + session.get_xenapi().VDI.set_name_label(primary_vdi_ref, primary_name_label) - LOG.debug(_("xapi 'download_vhd' returned VDI UUID %(vdi_uuid)s") + LOG.debug(_("xapi 'download_vhd' returned VDI UUID %(primary_vdi_uuid)s") % locals()) - return vdi_uuid + + LOG.debug("=" * 100) + LOG.debug(rimary_vdi_uuid) + LOG.debug(swap_vdi_uuid) + return (primary_vdi_uuid, swap_vdi_uuid) @classmethod def _fetch_image_glance_disk(cls, session, instance_id, image, access, diff --git a/nova/virt/xenapi/vmops.py b/nova/virt/xenapi/vmops.py index 0074444f8..4a01cac29 100644 --- a/nova/virt/xenapi/vmops.py +++ b/nova/virt/xenapi/vmops.py @@ -109,20 +109,20 @@ class VMOps(object): user = AuthManager().get_user(instance.user_id) project = AuthManager().get_project(instance.project_id) disk_image_type = VMHelper.determine_disk_image_type(instance) - vdi_uuid = VMHelper.fetch_image(self._session, instance.id, - instance.image_id, user, project, disk_image_type) - return vdi_uuid + (primary_vdi_uuid, swap_vdi_uuid) = VMHelper.fetch_image(self._session, + instance.id, instance.image_id, user, project, disk_image_type) + return (primary_vdi_uuid, swap_vdi_uuid) def spawn(self, instance, network_info=None): - vdi_uuid = self._create_disk(instance) - vm_ref = self._create_vm(instance, vdi_uuid, network_info) + vdi_uuid, swap_uuid = self._create_disk(instance) + vm_ref = self._create_vm(instance, vdi_uuid, swap_uuid, network_info) self._spawn(instance, vm_ref) def spawn_rescue(self, instance): """Spawn a rescue instance.""" self.spawn(instance) - def _create_vm(self, instance, vdi_uuid, network_info=None): + def _create_vm(self, instance, vdi_uuid, swap_vdi_uuid=None, network_info=None): """Create VM instance.""" instance_name = instance.name vm_ref = VMHelper.lookup(self._session, instance_name) @@ -143,18 +143,20 @@ class VMOps(object): # Are we building from a pre-existing disk? vdi_ref = self._session.call_xenapi('VDI.get_by_uuid', vdi_uuid) + if swap_vdi_uuid: + swap_vdi_ref = self._session.call_xenapi('VDI.get_by_uuid', swap_vdi_uuid) disk_image_type = VMHelper.determine_disk_image_type(instance) kernel = None if instance.kernel_id: kernel = VMHelper.fetch_image(self._session, instance.id, - instance.kernel_id, user, project, ImageType.KERNEL_RAMDISK) + instance.kernel_id, user, project, ImageType.KERNEL_RAMDISK)[0] ramdisk = None if instance.ramdisk_id: ramdisk = VMHelper.fetch_image(self._session, instance.id, - instance.ramdisk_id, user, project, ImageType.KERNEL_RAMDISK) + instance.ramdisk_id, user, project, ImageType.KERNEL_RAMDISK)[0] use_pv_kernel = VMHelper.determine_is_pv(self._session, instance.id, vdi_ref, disk_image_type, instance.os_type) @@ -163,6 +165,9 @@ class VMOps(object): VMHelper.create_vbd(session=self._session, vm_ref=vm_ref, vdi_ref=vdi_ref, userdevice=0, bootable=True) + if swap_vdi_uuid: + VMHelper.create_vbd(session=self._session, vm_ref=vm_ref, + vdi_ref=swap_vdi_ref, userdevice=0, bootable=False) # TODO(tr3buchet) - check to make sure we have network info, otherwise # create it now. This goes away once nova-multi-nic hits. diff --git a/plugins/xenserver/xenapi/etc/xapi.d/plugins/glance b/plugins/xenserver/xenapi/etc/xapi.d/plugins/glance index 4b45671ae..9d6ee78ab 100644 --- a/plugins/xenserver/xenapi/etc/xapi.d/plugins/glance +++ b/plugins/xenserver/xenapi/etc/xapi.d/plugins/glance @@ -177,8 +177,15 @@ def _fixup_vhds(sr_path, staging_path, uuid_stack): else: assert_vhd_not_hidden(base_copy_path) + # If we find a swap.vhd, go ahead and copy it into the SR + swap_uuid = None + orig_swap_path = os.path.join(staging_path, 'swap.vhd') + if os.path.exists(orig_swap_path): + swap_path, swap_uuid = rename_with_uuid(orig_swap_path) + move_into_sr(swap_path) + move_into_sr(base_copy_path) - return vdi_uuid + return dict(primary_vdi_uuid=vdi_uuid, swap_vdi_uuid=swap_uuid) def _prepare_staging_area_for_upload(sr_path, staging_path, vdi_uuids): @@ -324,8 +331,7 @@ def download_vhd(session, args): try: _download_tarball(sr_path, staging_path, image_id, glance_host, glance_port) - vdi_uuid = _fixup_vhds(sr_path, staging_path, uuid_stack) - return vdi_uuid + return _fixup_vhds(sr_path, staging_path, uuid_stack) finally: _cleanup_staging_area(staging_path) -- cgit From 94766fac0f5fdb3c7847b1129a8f05948a97f887 Mon Sep 17 00:00:00 2001 From: Chris Behrens Date: Mon, 23 May 2011 20:42:54 +0000 Subject: cleanup and fixes --- nova/virt/xenapi/vm_utils.py | 18 +++++----- nova/virt/xenapi/vmops.py | 40 +++++++++++++--------- plugins/xenserver/xenapi/etc/xapi.d/plugins/glance | 12 ++++--- 3 files changed, 42 insertions(+), 28 deletions(-) diff --git a/nova/virt/xenapi/vm_utils.py b/nova/virt/xenapi/vm_utils.py index c24fc7ba6..f1f7b8249 100644 --- a/nova/virt/xenapi/vm_utils.py +++ b/nova/virt/xenapi/vm_utils.py @@ -410,7 +410,7 @@ class VMHelper(HelperBase): task = session.async_call_plugin('glance', 'download_vhd', kwargs) vdi_uuids = session.wait_for_task(task, instance_id) primary_vdi_uuid = vdi_uuids.get('primary_vdi_uuid') - swap_vdi_uuid = vdi_uuids.get('swap_vdi_uuid') + swap_vdi_uuid = vdi_uuids.get('swap_vdi_uuid', None) cls.scan_sr(session, instance_id, sr_ref) @@ -419,13 +419,14 @@ class VMHelper(HelperBase): primary_name_label = get_name_label_for_image(image) session.get_xenapi().VDI.set_name_label(primary_vdi_ref, primary_name_label) - LOG.debug(_("xapi 'download_vhd' returned VDI UUID %(primary_vdi_uuid)s") - % locals()) + LOG.debug(_("xapi 'download_vhd' returned VDI UUID " + "%(primary_vdi_uuid)s") % locals()) + if swap_vdi_uuid: + LOG.debug(_("xapi 'download_vhd' returned SWAP VDI UUID " + "%(swap_vdi_uuid)s") % locals()) - LOG.debug("=" * 100) - LOG.debug(rimary_vdi_uuid) - LOG.debug(swap_vdi_uuid) - return (primary_vdi_uuid, swap_vdi_uuid) + LOG.debug("=" * 100) + return vdi_uuids @classmethod def _fetch_image_glance_disk(cls, session, instance_id, image, access, @@ -482,7 +483,8 @@ class VMHelper(HelperBase): LOG.debug(_("Kernel/Ramdisk VDI %s destroyed"), vdi_ref) return filename else: - return session.get_xenapi().VDI.get_uuid(vdi_ref) + vdi_uuid = session.get_xenapi().VDI.get_uuid(vdi_ref) + return {'primary_vdi_uuid': vdi_uuid} @classmethod def determine_disk_image_type(cls, instance): diff --git a/nova/virt/xenapi/vmops.py b/nova/virt/xenapi/vmops.py index 4a01cac29..0c30ad4cb 100644 --- a/nova/virt/xenapi/vmops.py +++ b/nova/virt/xenapi/vmops.py @@ -109,20 +109,21 @@ class VMOps(object): user = AuthManager().get_user(instance.user_id) project = AuthManager().get_project(instance.project_id) disk_image_type = VMHelper.determine_disk_image_type(instance) - (primary_vdi_uuid, swap_vdi_uuid) = VMHelper.fetch_image(self._session, - instance.id, instance.image_id, user, project, disk_image_type) - return (primary_vdi_uuid, swap_vdi_uuid) + vdi_uuids = VMHelper.fetch_image(self._session, + instance.id, instance.image_id, user, project, + disk_image_type) + return vdi_uuids def spawn(self, instance, network_info=None): - vdi_uuid, swap_uuid = self._create_disk(instance) - vm_ref = self._create_vm(instance, vdi_uuid, swap_uuid, network_info) + vdi_uuids = self._create_disk(instance) + vm_ref = self._create_vm(instance, vdi_uuids, network_info) self._spawn(instance, vm_ref) def spawn_rescue(self, instance): """Spawn a rescue instance.""" self.spawn(instance) - def _create_vm(self, instance, vdi_uuid, swap_vdi_uuid=None, network_info=None): + def _create_vm(self, instance, vdi_uuids, network_info=None): """Create VM instance.""" instance_name = instance.name vm_ref = VMHelper.lookup(self._session, instance_name) @@ -142,30 +143,37 @@ class VMOps(object): project = AuthManager().get_project(instance.project_id) # Are we building from a pre-existing disk? - vdi_ref = self._session.call_xenapi('VDI.get_by_uuid', vdi_uuid) + primary_vdi_ref = self._session.call_xenapi('VDI.get_by_uuid', + vdi_uuids['primary_vdi_uuid']) + swap_vdi_uuid = vdi_uuids.get('swap_vdi_uuid', None) if swap_vdi_uuid: swap_vdi_ref = self._session.call_xenapi('VDI.get_by_uuid', swap_vdi_uuid) + else: + swap_vdi_ref = None disk_image_type = VMHelper.determine_disk_image_type(instance) kernel = None if instance.kernel_id: kernel = VMHelper.fetch_image(self._session, instance.id, - instance.kernel_id, user, project, ImageType.KERNEL_RAMDISK)[0] + instance.kernel_id, user, project, + ImageType.KERNEL_RAMDISK) ramdisk = None if instance.ramdisk_id: ramdisk = VMHelper.fetch_image(self._session, instance.id, - instance.ramdisk_id, user, project, ImageType.KERNEL_RAMDISK)[0] + instance.ramdisk_id, user, project, + ImageType.KERNEL_RAMDISK) - use_pv_kernel = VMHelper.determine_is_pv(self._session, instance.id, - vdi_ref, disk_image_type, instance.os_type) - vm_ref = VMHelper.create_vm(self._session, instance, kernel, ramdisk, - use_pv_kernel) + use_pv_kernel = VMHelper.determine_is_pv(self._session, + instance.id, primary_vdi_ref, disk_image_type, + instance.os_type) + vm_ref = VMHelper.create_vm(self._session, instance, kernel, + ramdisk, use_pv_kernel) VMHelper.create_vbd(session=self._session, vm_ref=vm_ref, - vdi_ref=vdi_ref, userdevice=0, bootable=True) - if swap_vdi_uuid: + vdi_ref=primary_vdi_ref, userdevice=0, bootable=True) + if swap_vdi_ref: VMHelper.create_vbd(session=self._session, vm_ref=vm_ref, vdi_ref=swap_vdi_ref, userdevice=0, bootable=False) @@ -177,7 +185,7 @@ class VMOps(object): # Alter the image before VM start for, e.g. network injection if FLAGS.xenapi_inject_image: VMHelper.preconfigure_instance(self._session, instance, - vdi_ref, network_info) + primary_vdi_ref, network_info) self.create_vifs(vm_ref, network_info) self.inject_network_info(instance, network_info, vm_ref) diff --git a/plugins/xenserver/xenapi/etc/xapi.d/plugins/glance b/plugins/xenserver/xenapi/etc/xapi.d/plugins/glance index 9d6ee78ab..6cc7617e0 100644 --- a/plugins/xenserver/xenapi/etc/xapi.d/plugins/glance +++ b/plugins/xenserver/xenapi/etc/xapi.d/plugins/glance @@ -178,15 +178,19 @@ def _fixup_vhds(sr_path, staging_path, uuid_stack): assert_vhd_not_hidden(base_copy_path) # If we find a swap.vhd, go ahead and copy it into the SR - swap_uuid = None + swap_vdi_uuid = None orig_swap_path = os.path.join(staging_path, 'swap.vhd') if os.path.exists(orig_swap_path): - swap_path, swap_uuid = rename_with_uuid(orig_swap_path) + swap_path, swap_vdi_uuid = rename_with_uuid(orig_swap_path) move_into_sr(swap_path) - move_into_sr(base_copy_path) - return dict(primary_vdi_uuid=vdi_uuid, swap_vdi_uuid=swap_uuid) + vdi_uuids = {} + vdi_uuids['primary_vdi_uuid'] = vdi_uuid + if swap_vdi_uuid: + vdi_uuids['swap_vdi_uuid'] = swap_vdi_uuid + move_into_sr(base_copy_path) + return vdi_uuids def _prepare_staging_area_for_upload(sr_path, staging_path, vdi_uuids): """Hard-link VHDs into staging area with appropriate filename -- cgit From 42c209d90f491d19b3aabc70f8dafc33b76cf20d Mon Sep 17 00:00:00 2001 From: Chris Behrens Date: Mon, 23 May 2011 16:51:28 -0500 Subject: fix tests, have glance plugin return json encoded string of vdi uuids --- nova/tests/xenapi/stubs.py | 11 +++++++++-- nova/virt/xenapi/vm_utils.py | 6 +++++- nova/virt/xenapi/vmops.py | 4 ++-- plugins/xenserver/xenapi/etc/xapi.d/plugins/glance | 6 +++++- 4 files changed, 21 insertions(+), 6 deletions(-) diff --git a/nova/tests/xenapi/stubs.py b/nova/tests/xenapi/stubs.py index d9306900d..9f6f64318 100644 --- a/nova/tests/xenapi/stubs.py +++ b/nova/tests/xenapi/stubs.py @@ -17,6 +17,7 @@ """Stubouts, mocks and fixtures for the test suite""" import eventlet +import json from nova.virt import xenapi_conn from nova.virt.xenapi import fake from nova.virt.xenapi import volume_utils @@ -37,7 +38,7 @@ def stubout_instance_snapshot(stubs): sr_ref=sr_ref, sharable=False) vdi_rec = session.get_xenapi().VDI.get_record(vdi_ref) vdi_uuid = vdi_rec['uuid'] - return dict(primary_vdi_uuid=vdi_uuid, swap_vdi_uuid=None) + return {'primary_vdi_uuid': vdi_uuid} stubs.Set(vm_utils.VMHelper, 'fetch_image', fake_fetch_image) @@ -132,10 +133,16 @@ class FakeSessionForVMTests(fake.SessionBase): def __init__(self, uri): super(FakeSessionForVMTests, self).__init__(uri) - def host_call_plugin(self, _1, _2, _3, _4, _5): + def host_call_plugin(self, _1, _2, plugin, method, _5): sr_ref = fake.get_all('SR')[0] vdi_ref = fake.create_vdi('', False, sr_ref, False) vdi_rec = fake.get_record('VDI', vdi_ref) + if plugin == "glance" and method == "download_vhd": + swap_vdi_ref = fake.create_vdi('', False, sr_ref, False) + swap_vdi_rec = fake.get_record('VDI', swap_vdi_ref) + return '%s' % json.dumps( + {'primary_vdi_uuid': vdi_rec['uuid'], + 'swap_vdi_uuid': swap_vdi_rec['uuid']}) return '%s' % vdi_rec['uuid'] def VM_start(self, _1, ref, _2, _3): diff --git a/nova/virt/xenapi/vm_utils.py b/nova/virt/xenapi/vm_utils.py index f1f7b8249..3d980013a 100644 --- a/nova/virt/xenapi/vm_utils.py +++ b/nova/virt/xenapi/vm_utils.py @@ -19,6 +19,7 @@ Helper methods for operations related to the management of VM records and their attributes like VDIs, VIFs, as well as their lookup functions. """ +import json import os import pickle import re @@ -408,7 +409,8 @@ class VMHelper(HelperBase): kwargs = {'params': pickle.dumps(params)} task = session.async_call_plugin('glance', 'download_vhd', kwargs) - vdi_uuids = session.wait_for_task(task, instance_id) + result = session.wait_for_task(task, instance_id) + vdi_uuids = json.loads(result) primary_vdi_uuid = vdi_uuids.get('primary_vdi_uuid') swap_vdi_uuid = vdi_uuids.get('swap_vdi_uuid', None) @@ -571,6 +573,8 @@ class VMHelper(HelperBase): args['raw'] = 'true' task = session.async_call_plugin('objectstore', fn, args) uuid = session.wait_for_task(task, instance_id) + if image_type != ImageType.KERNEL_RAMDISK: + return {'primary_vdi_uuid': uuid} return uuid @classmethod diff --git a/nova/virt/xenapi/vmops.py b/nova/virt/xenapi/vmops.py index 0c30ad4cb..0d7ef5fac 100644 --- a/nova/virt/xenapi/vmops.py +++ b/nova/virt/xenapi/vmops.py @@ -91,7 +91,7 @@ class VMOps(object): def finish_resize(self, instance, disk_info): vdi_uuid = self.link_disks(instance, disk_info['base_copy'], disk_info['cow']) - vm_ref = self._create_vm(instance, vdi_uuid) + vm_ref = self._create_vm(instance, {'primary_vdi_uuid': vdi_uuid}) self.resize_instance(instance, vdi_uuid) self._spawn(instance, vm_ref) @@ -144,7 +144,7 @@ class VMOps(object): # Are we building from a pre-existing disk? primary_vdi_ref = self._session.call_xenapi('VDI.get_by_uuid', - vdi_uuids['primary_vdi_uuid']) + vdi_uuids.get('primary_vdi_uuid')) swap_vdi_uuid = vdi_uuids.get('swap_vdi_uuid', None) if swap_vdi_uuid: swap_vdi_ref = self._session.call_xenapi('VDI.get_by_uuid', swap_vdi_uuid) diff --git a/plugins/xenserver/xenapi/etc/xapi.d/plugins/glance b/plugins/xenserver/xenapi/etc/xapi.d/plugins/glance index 6cc7617e0..0d02adfe9 100644 --- a/plugins/xenserver/xenapi/etc/xapi.d/plugins/glance +++ b/plugins/xenserver/xenapi/etc/xapi.d/plugins/glance @@ -22,6 +22,10 @@ # import httplib +try: + import json +except ImportError: + import simplejson as json import os import os.path import pickle @@ -335,7 +339,7 @@ def download_vhd(session, args): try: _download_tarball(sr_path, staging_path, image_id, glance_host, glance_port) - return _fixup_vhds(sr_path, staging_path, uuid_stack) + return json.dumps(_fixup_vhds(sr_path, staging_path, uuid_stack)) finally: _cleanup_staging_area(staging_path) -- cgit From 038ce7e16ee7ee1afc86ded260c1aa0d40d1e1ad Mon Sep 17 00:00:00 2001 From: Chris Behrens Date: Mon, 23 May 2011 22:52:56 +0000 Subject: swap should use device 1 and rescue use device 2 --- nova/virt/xenapi/vmops.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/nova/virt/xenapi/vmops.py b/nova/virt/xenapi/vmops.py index 0d7ef5fac..6ff8fd6a4 100644 --- a/nova/virt/xenapi/vmops.py +++ b/nova/virt/xenapi/vmops.py @@ -175,7 +175,7 @@ class VMOps(object): vdi_ref=primary_vdi_ref, userdevice=0, bootable=True) if swap_vdi_ref: VMHelper.create_vbd(session=self._session, vm_ref=vm_ref, - vdi_ref=swap_vdi_ref, userdevice=0, bootable=False) + vdi_ref=swap_vdi_ref, userdevice=1, bootable=False) # TODO(tr3buchet) - check to make sure we have network info, otherwise # create it now. This goes away once nova-multi-nic hits. @@ -711,7 +711,7 @@ class VMOps(object): vbd_ref = self._session.get_xenapi().VM.get_VBDs(vm_ref)[0] vdi_ref = self._session.get_xenapi().VBD.get_record(vbd_ref)["VDI"] rescue_vbd_ref = VMHelper.create_vbd(self._session, rescue_vm_ref, - vdi_ref, 1, False) + vdi_ref, 2, False) self._session.call_xenapi("Async.VBD.plug", rescue_vbd_ref) -- cgit From f488576ae27f8eb96a04022d0ecd11a28bd15116 Mon Sep 17 00:00:00 2001 From: Alex Meade Date: Tue, 24 May 2011 16:44:28 -0400 Subject: Added filtering on image properties --- nova/api/openstack/images.py | 23 +++++++++++++++++++++-- nova/image/fake.py | 4 ++-- nova/image/glance.py | 8 ++++---- nova/tests/api/openstack/fakes.py | 4 ++-- nova/tests/api/openstack/test_images.py | 9 +++++++++ nova/tests/image/test_glance.py | 2 +- 6 files changed, 39 insertions(+), 11 deletions(-) diff --git a/nova/api/openstack/images.py b/nova/api/openstack/images.py index 34d4c27fc..755ce8ead 100644 --- a/nova/api/openstack/images.py +++ b/nova/api/openstack/images.py @@ -28,6 +28,9 @@ from nova.api.openstack.views import images as images_view LOG = log.getLogger('nova.api.openstack.images') FLAGS = flags.FLAGS +SUPPORTED_FILTERS = ['name', 'status', 'container_format', 'disk_format', + 'size_min', 'size_max'] + class Controller(common.OpenstackController): """Base `wsgi.Controller` for retrieving/displaying images.""" @@ -59,7 +62,8 @@ class Controller(common.OpenstackController): :param req: `wsgi.Request` object """ context = req.environ['nova.context'] - images = self._image_service.index(context) + filters = self._get_filters(req) + images = self._image_service.index(context, filters) images = common.limited(images, req) builder = self.get_builder(req).build return dict(images=[builder(image, detail=False) for image in images]) @@ -70,11 +74,26 @@ class Controller(common.OpenstackController): :param req: `wsgi.Request` object. """ context = req.environ['nova.context'] - images = self._image_service.detail(context) + filters = self._get_filters(req) + images = self._image_service.detail(context, filters) images = common.limited(images, req) builder = self.get_builder(req).build return dict(images=[builder(image, detail=True) for image in images]) + def _get_filters(self, req): + """ + Return a dictionary of query param filters from the request + + :param req: the Request object coming from the wsgi layer + :retval a dict of key/value filters + """ + filters = {} + for param in req.str_params: + if param in SUPPORTED_FILTERS or param.startswith('property-'): + filters[param] = req.str_params.get(param) + + return filters + def show(self, req, id): """Return detailed information about a specific image. diff --git a/nova/image/fake.py b/nova/image/fake.py index b400b2adb..8e84c8597 100644 --- a/nova/image/fake.py +++ b/nova/image/fake.py @@ -52,11 +52,11 @@ class FakeImageService(service.BaseImageService): self.create(None, image) super(FakeImageService, self).__init__() - def index(self, context): + def index(self, context, filters=None): """Returns list of images.""" return copy.deepcopy(self.images.values()) - def detail(self, context): + def detail(self, context, filters=None): """Return list of detailed image information.""" return copy.deepcopy(self.images.values()) diff --git a/nova/image/glance.py b/nova/image/glance.py index 193e37273..dec797619 100644 --- a/nova/image/glance.py +++ b/nova/image/glance.py @@ -58,23 +58,23 @@ class GlanceImageService(service.BaseImageService): else: self.client = client - def index(self, context): + def index(self, context, filters=None): """Calls out to Glance for a list of images available.""" # NOTE(sirp): We need to use `get_images_detailed` and not # `get_images` here because we need `is_public` and `properties` # included so we can filter by user filtered = [] - image_metas = self.client.get_images_detailed() + image_metas = self.client.get_images_detailed(filters=filters) for image_meta in image_metas: if self._is_image_available(context, image_meta): meta_subset = utils.subset_dict(image_meta, ('id', 'name')) filtered.append(meta_subset) return filtered - def detail(self, context): + def detail(self, context, filters=None): """Calls out to Glance for a list of detailed image information.""" filtered = [] - image_metas = self.client.get_images_detailed() + image_metas = self.client.get_images_detailed(filters=filters) for image_meta in image_metas: if self._is_image_available(context, image_meta): base_image_meta = self._translate_to_base(image_meta) diff --git a/nova/tests/api/openstack/fakes.py b/nova/tests/api/openstack/fakes.py index bf51239e6..8e0156afa 100644 --- a/nova/tests/api/openstack/fakes.py +++ b/nova/tests/api/openstack/fakes.py @@ -166,11 +166,11 @@ def stub_out_glance(stubs, initial_fixtures=None): def __init__(self, initial_fixtures): self.fixtures = initial_fixtures or [] - def fake_get_images(self): + def fake_get_images(self, filters=None): return [dict(id=f['id'], name=f['name']) for f in self.fixtures] - def fake_get_images_detailed(self): + def fake_get_images_detailed(self, filters=None): return copy.deepcopy(self.fixtures) def fake_get_image_meta(self, image_id): diff --git a/nova/tests/api/openstack/test_images.py b/nova/tests/api/openstack/test_images.py index 2c329f920..76d4e2f56 100644 --- a/nova/tests/api/openstack/test_images.py +++ b/nova/tests/api/openstack/test_images.py @@ -708,6 +708,15 @@ class ImageControllerWithGlanceServiceTest(test.TestCase): self.assertDictListMatch(expected, response_list) + def test_get_image_request_filters(self): + request =\ + webob.Request.blank('/v1.1/images/detail?status=ACTIVE&name=testname') + filters = images.Controller()._get_filters(request) + expected = {'status': 'ACTIVE', + 'name': 'testname', + } + self.assertDictMatch(expected, filters) + def test_get_image_found(self): req = webob.Request.blank('/v1.0/images/123') res = req.get_response(fakes.wsgi_app()) diff --git a/nova/tests/image/test_glance.py b/nova/tests/image/test_glance.py index 109905ded..6d108d494 100644 --- a/nova/tests/image/test_glance.py +++ b/nova/tests/image/test_glance.py @@ -34,7 +34,7 @@ class StubGlanceClient(object): def get_image_meta(self, image_id): return self.images[image_id] - def get_images_detailed(self): + def get_images_detailed(self, filters=None): return self.images.itervalues() def get_image(self, image_id): -- cgit From 26842cba90bd5637bd6aa185b300102ff257d9f1 Mon Sep 17 00:00:00 2001 From: Chris Behrens Date: Tue, 24 May 2011 22:39:16 +0000 Subject: move devices back --- nova/virt/xenapi/vmops.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/nova/virt/xenapi/vmops.py b/nova/virt/xenapi/vmops.py index 6ff8fd6a4..6fff1d494 100644 --- a/nova/virt/xenapi/vmops.py +++ b/nova/virt/xenapi/vmops.py @@ -175,7 +175,7 @@ class VMOps(object): vdi_ref=primary_vdi_ref, userdevice=0, bootable=True) if swap_vdi_ref: VMHelper.create_vbd(session=self._session, vm_ref=vm_ref, - vdi_ref=swap_vdi_ref, userdevice=1, bootable=False) + vdi_ref=swap_vdi_ref, userdevice=2, bootable=False) # TODO(tr3buchet) - check to make sure we have network info, otherwise # create it now. This goes away once nova-multi-nic hits. @@ -711,7 +711,7 @@ class VMOps(object): vbd_ref = self._session.get_xenapi().VM.get_VBDs(vm_ref)[0] vdi_ref = self._session.get_xenapi().VBD.get_record(vbd_ref)["VDI"] rescue_vbd_ref = VMHelper.create_vbd(self._session, rescue_vm_ref, - vdi_ref, 2, False) + vdi_ref, 1, False) self._session.call_xenapi("Async.VBD.plug", rescue_vbd_ref) -- cgit From 17abaeafaf3fed2847e4377a16b47771eb663304 Mon Sep 17 00:00:00 2001 From: MORITA Kazutaka Date: Wed, 25 May 2011 16:27:28 +0900 Subject: Fix wrong call of the volume api create() --- nova/api/openstack/contrib/volumes.py | 2 +- nova/tests/test_quota.py | 2 ++ 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/nova/api/openstack/contrib/volumes.py b/nova/api/openstack/contrib/volumes.py index 18de2ec71..b22bd2846 100644 --- a/nova/api/openstack/contrib/volumes.py +++ b/nova/api/openstack/contrib/volumes.py @@ -135,7 +135,7 @@ class VolumeController(wsgi.Controller): vol = env['volume'] size = vol['size'] LOG.audit(_("Create volume of %s GB"), size, context=context) - new_volume = self.volume_api.create(context, size, + new_volume = self.volume_api.create(context, size, None, vol.get('display_name'), vol.get('display_description')) diff --git a/nova/tests/test_quota.py b/nova/tests/test_quota.py index 7ace2ad7d..990068fae 100644 --- a/nova/tests/test_quota.py +++ b/nova/tests/test_quota.py @@ -228,6 +228,7 @@ class QuotaTestCase(test.TestCase): volume.API().create, self.context, size=10, + snapshot_id=None, name='', description='') for volume_id in volume_ids: @@ -241,6 +242,7 @@ class QuotaTestCase(test.TestCase): volume.API().create, self.context, size=10, + snapshot_id=None, name='', description='') for volume_id in volume_ids: -- cgit From 7139cf1f0cfe9241a1710e5b7c621db569a2fc2d Mon Sep 17 00:00:00 2001 From: MORITA Kazutaka Date: Wed, 25 May 2011 16:37:52 +0900 Subject: Make snapshot_id=None a default value in VolumeManager:create_volume(). It is not a regular case to create a volume from a snapshot. --- nova/volume/manager.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/nova/volume/manager.py b/nova/volume/manager.py index 84085fbd8..b6f0f5eeb 100644 --- a/nova/volume/manager.py +++ b/nova/volume/manager.py @@ -90,7 +90,7 @@ class VolumeManager(manager.SchedulerDependentManager): else: LOG.info(_("volume %s: skipping export"), volume['name']) - def create_volume(self, context, volume_id, snapshot_id): + def create_volume(self, context, volume_id, snapshot_id=None): """Creates and exports the volume.""" context = context.elevated() volume_ref = self.db.volume_get(context, volume_id) -- cgit From f3125b3012da7b6429e4e551060498e665c4596e Mon Sep 17 00:00:00 2001 From: MORITA Kazutaka Date: Wed, 25 May 2011 17:51:30 +0900 Subject: Add unittests for cloning volumes. --- nova/tests/test_cloud.py | 19 +++++++++++++++++++ nova/tests/test_volume.py | 20 +++++++++++++++++++- 2 files changed, 38 insertions(+), 1 deletion(-) diff --git a/nova/tests/test_cloud.py b/nova/tests/test_cloud.py index d9169a646..8c7520fe8 100644 --- a/nova/tests/test_cloud.py +++ b/nova/tests/test_cloud.py @@ -171,6 +171,25 @@ class CloudTestCase(test.TestCase): db.volume_destroy(self.context, vol1['id']) db.volume_destroy(self.context, vol2['id']) + def test_create_volume_from_snapshot(self): + """Makes sure create_volume works when we specify a snapshot.""" + vol = db.volume_create(self.context, {'size': 1}) + snap = db.snapshot_create(self.context, {'volume_id': vol['id'], + 'volume_size': vol['size'], + 'status': "available"}) + snapshot_id = ec2utils.id_to_ec2_id(snap['id'], 'snap-%08x') + + result = self.cloud.create_volume(self.context, + snapshot_id=snapshot_id) + volume_id = result['volumeId'] + result = self.cloud.describe_volumes(self.context) + self.assertEqual(len(result['volumeSet']), 2) + self.assertEqual(result['volumeSet'][1]['volumeId'], volume_id) + + db.volume_destroy(self.context, ec2utils.ec2_id_to_id(volume_id)) + db.snapshot_destroy(self.context, snap['id']) + db.volume_destroy(self.context, vol['id']) + def test_describe_availability_zones(self): """Makes sure describe_availability_zones works and filters results.""" service1 = db.service_create(self.context, {'host': 'host1_zones', diff --git a/nova/tests/test_volume.py b/nova/tests/test_volume.py index c66b66959..1c25d601a 100644 --- a/nova/tests/test_volume.py +++ b/nova/tests/test_volume.py @@ -45,10 +45,11 @@ class VolumeTestCase(test.TestCase): self.context = context.get_admin_context() @staticmethod - def _create_volume(size='0'): + def _create_volume(size='0', snapshot_id=None): """Create a volume object.""" vol = {} vol['size'] = size + vol['snapshot_id'] = snapshot_id vol['user_id'] = 'fake' vol['project_id'] = 'fake' vol['availability_zone'] = FLAGS.storage_availability_zone @@ -69,6 +70,23 @@ class VolumeTestCase(test.TestCase): self.context, volume_id) + def test_create_volume_from_snapshot(self): + """Test volume can be created from a snapshot.""" + volume_src_id = self._create_volume() + self.volume.create_volume(self.context, volume_src_id) + snapshot_id = self._create_snapshot(volume_src_id) + self.volume.create_snapshot(self.context, volume_src_id, snapshot_id) + volume_dst_id = self._create_volume(0, snapshot_id) + self.volume.create_volume(self.context, volume_dst_id, snapshot_id) + self.assertEqual(volume_dst_id, db.volume_get(context.get_admin_context(), + volume_dst_id).id) + self.assertEqual(snapshot_id, db.volume_get(context.get_admin_context(), + volume_dst_id).snapshot_id) + + self.volume.delete_volume(self.context, volume_dst_id) + self.volume.delete_snapshot(self.context, snapshot_id) + self.volume.delete_volume(self.context, volume_src_id) + def test_too_big_volume(self): """Ensure failure if a too large of a volume is requested.""" # FIXME(vish): validation needs to move into the data layer in -- cgit From d380729b162c8d6120279db74327e61a4942e28f Mon Sep 17 00:00:00 2001 From: MORITA Kazutaka Date: Wed, 25 May 2011 18:02:07 +0900 Subject: Avoid wildcard import. --- .../sqlalchemy/migrate_repo/versions/020_add_snapshot_id_to_volumes.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/nova/db/sqlalchemy/migrate_repo/versions/020_add_snapshot_id_to_volumes.py b/nova/db/sqlalchemy/migrate_repo/versions/020_add_snapshot_id_to_volumes.py index 0a50123bf..10bd9d5c9 100644 --- a/nova/db/sqlalchemy/migrate_repo/versions/020_add_snapshot_id_to_volumes.py +++ b/nova/db/sqlalchemy/migrate_repo/versions/020_add_snapshot_id_to_volumes.py @@ -15,8 +15,7 @@ # License for the specific language governing permissions and limitations # under the License. -from sqlalchemy import * -from migrate import * +from sqlalchemy import Column, Table, MetaData, Integer from nova import log as logging -- cgit From 3d9569147cee2eaa94fc49c55b40f70a72171ebe Mon Sep 17 00:00:00 2001 From: Alex Meade Date: Wed, 25 May 2011 09:33:51 -0400 Subject: Added test --- nova/tests/api/openstack/test_images.py | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/nova/tests/api/openstack/test_images.py b/nova/tests/api/openstack/test_images.py index 76d4e2f56..233419c6d 100644 --- a/nova/tests/api/openstack/test_images.py +++ b/nova/tests/api/openstack/test_images.py @@ -709,11 +709,20 @@ class ImageControllerWithGlanceServiceTest(test.TestCase): self.assertDictListMatch(expected, response_list) def test_get_image_request_filters(self): - request =\ - webob.Request.blank('/v1.1/images/detail?status=ACTIVE&name=testname') + request = webob.Request.blank( + '/v1.1/images/detail?status=ACTIVE&name=testname&property-test=3') filters = images.Controller()._get_filters(request) expected = {'status': 'ACTIVE', 'name': 'testname', + 'property-test': '3', + } + self.assertDictMatch(expected, filters) + + def test_get_image_request_filters_not_supported(self): + request = webob.Request.blank( + '/v1.1/images/detail?status=ACTIVE&UNSUPPORTEDFILTER=testname') + filters = images.Controller()._get_filters(request) + expected = {'status': 'ACTIVE', } self.assertDictMatch(expected, filters) -- cgit From 537c5aea298a6c09b3329185c2d0eed77a0a21bd Mon Sep 17 00:00:00 2001 From: Mark Washenberger Date: Wed, 25 May 2011 12:09:53 -0400 Subject: try out mox for testing image request filters --- nova/tests/api/openstack/test_images.py | 19 +++++++++++++------ 1 file changed, 13 insertions(+), 6 deletions(-) diff --git a/nova/tests/api/openstack/test_images.py b/nova/tests/api/openstack/test_images.py index 233419c6d..e25334732 100644 --- a/nova/tests/api/openstack/test_images.py +++ b/nova/tests/api/openstack/test_images.py @@ -28,6 +28,7 @@ import shutil import tempfile import xml.dom.minidom as minidom +import mox import stubout import webob @@ -709,14 +710,20 @@ class ImageControllerWithGlanceServiceTest(test.TestCase): self.assertDictListMatch(expected, response_list) def test_get_image_request_filters(self): + mocker = mox.Mox() + image_service = mocker.CreateMockAnything() + context = object() + filters = {'status': 'ACTIVE', + 'name': 'testname', + 'property-test': '3'} + image_service.detail(context, filters).AndReturn([]) + mocker.ReplayAll() request = webob.Request.blank( '/v1.1/images/detail?status=ACTIVE&name=testname&property-test=3') - filters = images.Controller()._get_filters(request) - expected = {'status': 'ACTIVE', - 'name': 'testname', - 'property-test': '3', - } - self.assertDictMatch(expected, filters) + request.environ['nova.context'] = context + controller = images.ControllerV11(image_service=image_service) + controller.detail(request) + mocker.VerifyAll() def test_get_image_request_filters_not_supported(self): request = webob.Request.blank( -- cgit From e4bf97ba29e8e5858f37cedb34e20ccd8e210bae Mon Sep 17 00:00:00 2001 From: Alex Meade Date: Wed, 25 May 2011 12:24:27 -0400 Subject: Updated tests to use mox pep8 --- nova/api/openstack/images.py | 2 +- nova/tests/api/openstack/test_images.py | 14 ++++++++++---- 2 files changed, 11 insertions(+), 5 deletions(-) diff --git a/nova/api/openstack/images.py b/nova/api/openstack/images.py index 755ce8ead..553566d58 100644 --- a/nova/api/openstack/images.py +++ b/nova/api/openstack/images.py @@ -93,7 +93,7 @@ class Controller(common.OpenstackController): filters[param] = req.str_params.get(param) return filters - + def show(self, req, id): """Return detailed information about a specific image. diff --git a/nova/tests/api/openstack/test_images.py b/nova/tests/api/openstack/test_images.py index e25334732..f3f0217d6 100644 --- a/nova/tests/api/openstack/test_images.py +++ b/nova/tests/api/openstack/test_images.py @@ -726,12 +726,18 @@ class ImageControllerWithGlanceServiceTest(test.TestCase): mocker.VerifyAll() def test_get_image_request_filters_not_supported(self): + mocker = mox.Mox() + image_service = mocker.CreateMockAnything() + context = object() + filters = {'status': 'ACTIVE'} + image_service.detail(context, filters).AndReturn([]) + mocker.ReplayAll() request = webob.Request.blank( '/v1.1/images/detail?status=ACTIVE&UNSUPPORTEDFILTER=testname') - filters = images.Controller()._get_filters(request) - expected = {'status': 'ACTIVE', - } - self.assertDictMatch(expected, filters) + request.environ['nova.context'] = context + controller = images.ControllerV11(image_service=image_service) + controller.detail(request) + mocker.VerifyAll() def test_get_image_found(self): req = webob.Request.blank('/v1.0/images/123') -- cgit From c440aecaaacf3caa8683234022bc10836d232971 Mon Sep 17 00:00:00 2001 From: Alex Meade Date: Wed, 25 May 2011 17:28:10 -0400 Subject: Added params to local and base image service --- nova/image/local.py | 4 ++-- nova/image/service.py | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/nova/image/local.py b/nova/image/local.py index 918180bae..677d5302b 100644 --- a/nova/image/local.py +++ b/nova/image/local.py @@ -63,7 +63,7 @@ class LocalImageService(service.BaseImageService): images.append(unhexed_image_id) return images - def index(self, context): + def index(self, context, *args, **kwargs): filtered = [] image_metas = self.detail(context) for image_meta in image_metas: @@ -71,7 +71,7 @@ class LocalImageService(service.BaseImageService): filtered.append(meta) return filtered - def detail(self, context): + def detail(self, context, *args, **kwargs): images = [] for image_id in self._ids(): try: diff --git a/nova/image/service.py b/nova/image/service.py index ab6749049..5361cfc89 100644 --- a/nova/image/service.py +++ b/nova/image/service.py @@ -46,7 +46,7 @@ class BaseImageService(object): # the ImageService subclass SERVICE_IMAGE_ATTRS = [] - def index(self, context): + def index(self, context, *args, **kwargs): """List images. :returns: a sequence of mappings with the following signature @@ -55,7 +55,7 @@ class BaseImageService(object): """ raise NotImplementedError - def detail(self, context): + def detail(self, context, *args, **kwargs): """Detailed information about an images. :returns: a sequence of mappings with the following signature -- cgit From fdd27860724cd57db6df059a97e98289f88ce6ac Mon Sep 17 00:00:00 2001 From: termie Date: Wed, 25 May 2011 15:42:24 -0700 Subject: add support to rpc for multicall --- nova/rpc.py | 99 +++++++++++++++++++++++++++++++++++++------------- nova/tests/test_rpc.py | 17 +++++++++ 2 files changed, 90 insertions(+), 26 deletions(-) diff --git a/nova/rpc.py b/nova/rpc.py index 2116f22c3..04198a4a6 100644 --- a/nova/rpc.py +++ b/nova/rpc.py @@ -32,8 +32,11 @@ import uuid from carrot import connection as carrot_connection from carrot import messaging +import eventlet from eventlet import greenpool from eventlet import greenthread +from eventlet import queue + from nova import context from nova import exception @@ -131,7 +134,8 @@ class Consumer(messaging.Consumer): self.connection = Connection.recreate() self.backend = self.connection.create_backend() self.declare() - super(Consumer, self).fetch(no_ack, auto_ack, enable_callbacks) + return super(Consumer, self).fetch( + no_ack, auto_ack, enable_callbacks) if self.failed_connection: LOG.error(_('Reconnected to queue')) self.failed_connection = False @@ -347,8 +351,9 @@ def _unpack_context(msg): if key.startswith('_context_'): value = msg.pop(key) context_dict[key[9:]] = value + context_dict['msg_id'] = msg.pop('_msg_id', None) LOG.debug(_('unpacked context: %s'), context_dict) - return context.RequestContext.from_dict(context_dict) + return RpcContext.from_dict(context_dict) def _pack_context(msg, context): @@ -365,26 +370,27 @@ def _pack_context(msg, context): msg.update(context) -def call(context, topic, msg): - """Sends a message on a topic and wait for a response.""" +class RpcContext(context.RequestContext): + def __init__(self, *args, **kwargs): + msg_id = kwargs.pop('msg_id', None) + self.msg_id = msg_id + super(RpcContext, self).__init__(*args, **kwargs) + + def reply(self, *args, **kwargs): + msg_reply(self.msg_id, *args, **kwargs) + + +def multicall(context, topic, msg): + """Make a call that returns multiple times.""" LOG.debug(_('Making asynchronous call on %s ...'), topic) msg_id = uuid.uuid4().hex msg.update({'_msg_id': msg_id}) LOG.debug(_('MSG_ID is %s') % (msg_id)) _pack_context(msg, context) - class WaitMessage(object): - def __call__(self, data, message): - """Acks message and sets result.""" - message.ack() - if data['failure']: - self.result = RemoteError(*data['failure']) - else: - self.result = data['result'] - - wait_msg = WaitMessage() conn = Connection.instance() consumer = DirectConsumer(connection=conn, msg_id=msg_id) + wait_msg = MulticallWaiter(consumer) consumer.register_callback(wait_msg) conn = Connection.instance() @@ -392,18 +398,59 @@ def call(context, topic, msg): publisher.send(msg) publisher.close() - try: - consumer.wait(limit=1) - except StopIteration: - pass - consumer.close() - # NOTE(termie): this is a little bit of a change from the original - # non-eventlet code where returning a Failure - # instance from a deferred call is very similar to - # raising an exception - if isinstance(wait_msg.result, Exception): - raise wait_msg.result - return wait_msg.result + return wait_msg + + +class MulticallWaiter(object): + def __init__(self, consumer): + self._consumer = consumer + self._results = queue.Queue() + self._closed = False + + def close(self): + self._closed = True + self._consumer.close() + + def __call__(self, data, message): + """Acks message and sets result.""" + message.ack() + if data['failure']: + self._results.put(RemoteError(*data['failure'])) + else: + self._results.put(data['result']) + + def __iter__(self): + return self.wait() + + def wait(self): + # TODO(termie): This is probably really a much simpler issue but am + # trying to solve the problem quickly. This works but + # I'd prefer to dig in and do it the best way later on. + + def _waiter(): + while not self._closed: + try: + self._consumer.wait(limit=1) + except StopIteration: + pass + eventlet.spawn(_waiter) + + while True: + result = self._results.get() + if isinstance(result, Exception): + raise result + if result == None: + self.close() + raise StopIteration + yield result + + +def call(context, topic, msg): + """Sends a message on a topic and wait for a response.""" + rv = multicall(context, topic, msg) + for x in rv: + rv.close() + return x def cast(context, topic, msg): diff --git a/nova/tests/test_rpc.py b/nova/tests/test_rpc.py index 44d7c91eb..92ddfcffc 100644 --- a/nova/tests/test_rpc.py +++ b/nova/tests/test_rpc.py @@ -49,6 +49,17 @@ class RpcTestCase(test.TestCase): "args": {"value": value}}) self.assertEqual(value, result) + def test_multicall_succeed_three_times(self): + """Get a value through rpc call""" + value = 42 + result = rpc.multicall(self.context, + 'test', + {"method": "echo_three_times", + "args": {"value": value}}) + + for x in result: + self.assertEqual(value, x) + def test_context_passed(self): """Makes sure a context is passed through rpc call""" value = 42 @@ -126,6 +137,12 @@ class TestReceiver(object): LOG.debug(_("Received %s"), context) return context.to_dict() + @staticmethod + def echo_three_times(context, value): + context.reply(value) + context.reply(value) + context.reply(value) + @staticmethod def fail(context, value): """Raises an exception with the value sent in""" -- cgit From d46c9fffe4fab8f55483c73d3e6ef12116de9bc5 Mon Sep 17 00:00:00 2001 From: termie Date: Wed, 25 May 2011 15:42:24 -0700 Subject: make the test more expicit --- nova/tests/test_rpc.py | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/nova/tests/test_rpc.py b/nova/tests/test_rpc.py index 92ddfcffc..acab3e758 100644 --- a/nova/tests/test_rpc.py +++ b/nova/tests/test_rpc.py @@ -56,9 +56,10 @@ class RpcTestCase(test.TestCase): 'test', {"method": "echo_three_times", "args": {"value": value}}) - + i = 0 for x in result: - self.assertEqual(value, x) + self.assertEqual(value + i, x) + i += 1 def test_context_passed(self): """Makes sure a context is passed through rpc call""" @@ -140,8 +141,8 @@ class TestReceiver(object): @staticmethod def echo_three_times(context, value): context.reply(value) - context.reply(value) - context.reply(value) + context.reply(value + 1) + context.reply(value + 2) @staticmethod def fail(context, value): -- cgit From 7622e854ef68fbdbfc531690cf74916301956c8e Mon Sep 17 00:00:00 2001 From: termie Date: Wed, 25 May 2011 15:42:24 -0700 Subject: add commented out unworking code for yield-based returns --- nova/rpc.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/nova/rpc.py b/nova/rpc.py index 04198a4a6..f43291c4b 100644 --- a/nova/rpc.py +++ b/nova/rpc.py @@ -201,6 +201,11 @@ class AdapterConsumer(Consumer): try: rval = node_func(context=ctxt, **node_args) if msg_id: + # TODO(termie): re-enable when fix the yielding issue + #if hasattr(rval, 'send'): + # logging.error('rval! %s', rval) + # for x in rval: + # msg_reply(msg_id, x, None) msg_reply(msg_id, rval, None) except Exception as e: logging.exception('Exception during message handling') -- cgit From b44c1fe9561ee8754137d2700bab295f20a4032b Mon Sep 17 00:00:00 2001 From: Chris Behrens Date: Wed, 25 May 2011 15:42:24 -0700 Subject: Add a connection pool for rpc cast/call Use the same rabbit connection for all topic listening and wait to be notified vs doing a 0.1 second poll for each. --- nova/rpc.py | 96 ++++++++++++++++++++++++++++++++++++++++++--------------- nova/service.py | 21 +++++++------ 2 files changed, 84 insertions(+), 33 deletions(-) diff --git a/nova/rpc.py b/nova/rpc.py index f43291c4b..62590ca92 100644 --- a/nova/rpc.py +++ b/nova/rpc.py @@ -35,9 +35,9 @@ from carrot import messaging import eventlet from eventlet import greenpool from eventlet import greenthread +from eventlet import pools from eventlet import queue - from nova import context from nova import exception from nova import fakerabbit @@ -92,6 +92,11 @@ class Connection(carrot_connection.BrokerConnection): pass return cls.instance() +class Pool(pools.Pool): + def create(self): + return Connection.instance(new=True) + +ConnectionPool = Pool(max_size=20) class Consumer(messaging.Consumer): """Consumer base class. @@ -163,21 +168,9 @@ class AdapterConsumer(Consumer): self.pool = greenpool.GreenPool(FLAGS.rpc_thread_pool_size) super(AdapterConsumer, self).__init__(connection=connection, topic=topic) + self.register_callback(self.process_data) - def receive(self, *args, **kwargs): - self.pool.spawn_n(self._receive, *args, **kwargs) - - @exception.wrap_exception - def _receive(self, message_data, message): - """Magically looks for a method on the proxy object and calls it. - - Message data should be a dictionary with two keys: - method: string representing the method to call - args: dictionary of arg: value - - Example: {'method': 'echo', 'args': {'value': 42}} - - """ + def process_data(self, message_data, message): LOG.debug(_('received %s') % message_data) msg_id = message_data.pop('_msg_id', None) @@ -194,6 +187,19 @@ class AdapterConsumer(Consumer): LOG.warn(_('no method for message: %s') % message_data) msg_reply(msg_id, _('No method for message: %s') % message_data) return + self.pool.spawn_n(self._process_data, msg_id, ctxt, method, args) + + @exception.wrap_exception + def _process_data(self, msg_id, ctxt, method, args): + """Magically looks for a method on the proxy object and calls it. + + Message data should be a dictionary with two keys: + method: string representing the method to call + args: dictionary of arg: value + + Example: {'method': 'echo', 'args': {'value': 42}} + + """ node_func = getattr(self.proxy, str(method)) node_args = dict((str(k), v) for k, v in args.iteritems()) @@ -214,11 +220,6 @@ class AdapterConsumer(Consumer): return -class Publisher(messaging.Publisher): - """Publisher base class.""" - pass - - class TopicAdapterConsumer(AdapterConsumer): """Consumes messages on a specific topic.""" @@ -251,6 +252,50 @@ class FanoutAdapterConsumer(AdapterConsumer): topic=topic, proxy=proxy) +class ConsumerSet(object): + """Groups consumers to listen on together on a single connection""" + + def __init__(self, conn, consumer_list): + self.consumer_list = set(consumer_list) + self.consumer_set = None + self.init(conn) + + def init(self, conn): + if not conn: + conn = Connection.instance(new=True) + if self.consumer_set: + self.consumer_set.close() + self.consumer_set = messaging.ConsumerSet(conn) + for consumer in self.consumer_list: + consumer.connection = conn + # consumer.backend is set for us + self.consumer_set.add_consumer(consumer) + + def reconnect(self): + self.init(None) + + def wait(self, limit=None): + while True: + it = self.consumer_set.iterconsume(limit=limit) + while True: + try: + it.next() + except StopIteration: + return + except Exception as e: + LOG.error(_("Received exception %s " % str(e) + \ + "while processing consumer")) + fuck + self.reconnect() + # Break to outer loop + break + + +class Publisher(messaging.Publisher): + """Publisher base class.""" + pass + + class TopicPublisher(Publisher): """Publishes messages on a specific topic.""" @@ -315,7 +360,7 @@ def msg_reply(msg_id, reply=None, failure=None): LOG.error(_("Returning exception %s to caller"), message) LOG.error(tb) failure = (failure[0].__name__, str(failure[1]), tb) - conn = Connection.instance() + conn = ConnectionPool.get() publisher = DirectPublisher(connection=conn, msg_id=msg_id) try: publisher.send({'result': reply, 'failure': failure}) @@ -324,7 +369,9 @@ def msg_reply(msg_id, reply=None, failure=None): {'result': dict((k, repr(v)) for k, v in reply.__dict__.iteritems()), 'failure': failure}) + publisher.close() + ConnectionPool.put(conn) class RemoteError(exception.Error): @@ -393,12 +440,11 @@ def multicall(context, topic, msg): LOG.debug(_('MSG_ID is %s') % (msg_id)) _pack_context(msg, context) - conn = Connection.instance() + conn = ConnectionPool.get() consumer = DirectConsumer(connection=conn, msg_id=msg_id) wait_msg = MulticallWaiter(consumer) consumer.register_callback(wait_msg) - conn = Connection.instance() publisher = TopicPublisher(connection=conn, topic=topic) publisher.send(msg) publisher.close() @@ -462,10 +508,11 @@ def cast(context, topic, msg): """Sends a message on a topic without waiting for a response.""" LOG.debug(_('Making asynchronous cast on %s...'), topic) _pack_context(msg, context) - conn = Connection.instance() + conn = ConnectionPool.get() publisher = TopicPublisher(connection=conn, topic=topic) publisher.send(msg) publisher.close() + ConnectionPool.put(conn) def fanout_cast(context, topic, msg): @@ -511,6 +558,7 @@ def send_message(topic, message, wait=True): if wait: consumer.wait() + consumer.close() if __name__ == '__main__': diff --git a/nova/service.py b/nova/service.py index ab1238c3b..7761cfef5 100644 --- a/nova/service.py +++ b/nova/service.py @@ -91,26 +91,29 @@ class Service(object): if 'nova-compute' == self.binary: self.manager.update_available_resource(ctxt) - conn1 = rpc.Connection.instance(new=True) - conn2 = rpc.Connection.instance(new=True) - conn3 = rpc.Connection.instance(new=True) + if self.report_interval: + conn = rpc.Connection.instance(new=True) + + # Share this same connection for these Consumers consumer_all = rpc.TopicAdapterConsumer( - connection=conn1, + connection=conn, topic=self.topic, proxy=self) consumer_node = rpc.TopicAdapterConsumer( - connection=conn2, + connection=conn, topic='%s.%s' % (self.topic, self.host), proxy=self) fanout = rpc.FanoutAdapterConsumer( - connection=conn3, + connection=conn, topic=self.topic, proxy=self) - self.timers.append(consumer_all.attach_to_eventlet()) - self.timers.append(consumer_node.attach_to_eventlet()) - self.timers.append(fanout.attach_to_eventlet()) + cset = rpc.ConsumerSet(conn, [consumer_all, + consumer_node, + fanout]) + # Wait forever, processing these consumers + greenthread.spawn_n(cset.wait) pulse = utils.LoopingCall(self.report_state) pulse.start(interval=self.report_interval, now=False) -- cgit From e1a47584cc63136280cf3ca9ef02da3efc1dff7f Mon Sep 17 00:00:00 2001 From: Chris Behrens Date: Wed, 25 May 2011 15:42:24 -0700 Subject: pep8 and comment fixes --- nova/rpc.py | 25 ++++++++++++++++--------- nova/service.py | 1 - 2 files changed, 16 insertions(+), 10 deletions(-) diff --git a/nova/rpc.py b/nova/rpc.py index 62590ca92..db5aec826 100644 --- a/nova/rpc.py +++ b/nova/rpc.py @@ -92,12 +92,16 @@ class Connection(carrot_connection.BrokerConnection): pass return cls.instance() + class Pool(pools.Pool): + """Class that implements a Pool of Connections""" + def create(self): return Connection.instance(new=True) ConnectionPool = Pool(max_size=20) + class Consumer(messaging.Consumer): """Consumer base class. @@ -171,6 +175,16 @@ class AdapterConsumer(Consumer): self.register_callback(self.process_data) def process_data(self, message_data, message): + """Consumer callback that parses the message for validity and + fires off a thread to call the proxy object method. + + Message data should be a dictionary with two keys: + method: string representing the method to call + args: dictionary of arg: value + + Example: {'method': 'echo', 'args': {'value': 42}} + + """ LOG.debug(_('received %s') % message_data) msg_id = message_data.pop('_msg_id', None) @@ -191,14 +205,8 @@ class AdapterConsumer(Consumer): @exception.wrap_exception def _process_data(self, msg_id, ctxt, method, args): - """Magically looks for a method on the proxy object and calls it. - - Message data should be a dictionary with two keys: - method: string representing the method to call - args: dictionary of arg: value - - Example: {'method': 'echo', 'args': {'value': 42}} - + """Thread that maigcally looks for a method on the proxy + object and calls it. """ node_func = getattr(self.proxy, str(method)) @@ -285,7 +293,6 @@ class ConsumerSet(object): except Exception as e: LOG.error(_("Received exception %s " % str(e) + \ "while processing consumer")) - fuck self.reconnect() # Break to outer loop break diff --git a/nova/service.py b/nova/service.py index 7761cfef5..c51c9b066 100644 --- a/nova/service.py +++ b/nova/service.py @@ -91,7 +91,6 @@ class Service(object): if 'nova-compute' == self.binary: self.manager.update_available_resource(ctxt) - if self.report_interval: conn = rpc.Connection.instance(new=True) -- cgit From d0be426d4e7bbfb1ecb3f078c71c1e176da441a5 Mon Sep 17 00:00:00 2001 From: Chris Behrens Date: Wed, 25 May 2011 15:42:24 -0700 Subject: convert fanout_cast to ConnectionPool --- nova/rpc.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/nova/rpc.py b/nova/rpc.py index db5aec826..fdb228695 100644 --- a/nova/rpc.py +++ b/nova/rpc.py @@ -526,10 +526,11 @@ def fanout_cast(context, topic, msg): """Sends a message on a fanout exchange without waiting for a response.""" LOG.debug(_('Making asynchronous fanout cast...')) _pack_context(msg, context) - conn = Connection.instance() + conn = ConnectionPool.get() publisher = FanoutPublisher(topic, connection=conn) publisher.send(msg) publisher.close() + ConnectionPool.put(conn) def generic_response(message_data, message): -- cgit From f2c2a593c828fc86e298d3eb31672a09b498c41f Mon Sep 17 00:00:00 2001 From: Chris Behrens Date: Wed, 25 May 2011 15:42:24 -0700 Subject: fakerabbit's declare_consumer should support more than 1 consumer. also: make fakerabbit Backend.consume be an iterator like it should be.. --- nova/fakerabbit.py | 22 +++++++++++++++------- 1 file changed, 15 insertions(+), 7 deletions(-) diff --git a/nova/fakerabbit.py b/nova/fakerabbit.py index a7dee8caf..a29ba9d86 100644 --- a/nova/fakerabbit.py +++ b/nova/fakerabbit.py @@ -77,6 +77,10 @@ class Queue(object): class Backend(base.BaseBackend): + def __init__(self, connection, **kwargs): + super(Backend, self).__init__(connection, **kwargs) + self.consumers = [] + def queue_declare(self, queue, **kwargs): global QUEUES if queue not in QUEUES: @@ -97,16 +101,20 @@ class Backend(base.BaseBackend): EXCHANGES[exchange].bind(QUEUES[queue].push, routing_key) def declare_consumer(self, queue, callback, *args, **kwargs): - self.current_queue = queue - self.current_callback = callback + self.consumers.append((queue, callback)) def consume(self, limit=None): + num = 0 while True: - item = self.get(self.current_queue) - if item: - self.current_callback(item) - raise StopIteration() - greenthread.sleep(0) + for (queue, callback) in self.consumers: + item = self.get(queue) + if item: + callback(item) + num += 1 + yield + if limit and num == limit: + raise StopIteration() + greenthread.sleep(0.1) def get(self, queue, no_ack=False): global QUEUES -- cgit From 90e30806a2e0c235612eb09792656cd861997f84 Mon Sep 17 00:00:00 2001 From: Vishvananda Ishaya Date: Wed, 25 May 2011 15:42:24 -0700 Subject: fix consumers to actually be deleted and clean up cloud test --- nova/fakerabbit.py | 13 +++++++++---- nova/rpc.py | 13 ++++++++++--- nova/service.py | 8 +++----- nova/tests/test_cloud.py | 26 ++++++++++---------------- 4 files changed, 32 insertions(+), 28 deletions(-) diff --git a/nova/fakerabbit.py b/nova/fakerabbit.py index a29ba9d86..5f3e75c48 100644 --- a/nova/fakerabbit.py +++ b/nova/fakerabbit.py @@ -79,7 +79,7 @@ class Queue(object): class Backend(base.BaseBackend): def __init__(self, connection, **kwargs): super(Backend, self).__init__(connection, **kwargs) - self.consumers = [] + self.consumers = {} def queue_declare(self, queue, **kwargs): global QUEUES @@ -100,13 +100,18 @@ class Backend(base.BaseBackend): ' key %(routing_key)s') % locals()) EXCHANGES[exchange].bind(QUEUES[queue].push, routing_key) - def declare_consumer(self, queue, callback, *args, **kwargs): - self.consumers.append((queue, callback)) + def declare_consumer(self, queue, callback, consumer_tag, *args, **kwargs): + LOG.debug("Adding consumer %s", consumer_tag) + self.consumers[consumer_tag] = (queue, callback) + + def cancel(self, consumer_tag): + LOG.debug("Removing consumer %s", consumer_tag) + del self.consumers[consumer_tag] def consume(self, limit=None): num = 0 while True: - for (queue, callback) in self.consumers: + for (queue, callback) in self.consumers.itervalues(): item = self.get(queue) if item: callback(item) diff --git a/nova/rpc.py b/nova/rpc.py index fdb228695..e2e962fcc 100644 --- a/nova/rpc.py +++ b/nova/rpc.py @@ -30,11 +30,11 @@ import time import traceback import uuid +import greenlet from carrot import connection as carrot_connection from carrot import messaging import eventlet from eventlet import greenpool -from eventlet import greenthread from eventlet import pools from eventlet import queue @@ -266,6 +266,7 @@ class ConsumerSet(object): def __init__(self, conn, consumer_list): self.consumer_list = set(consumer_list) self.consumer_set = None + self.enabled = True self.init(conn) def init(self, conn): @@ -283,15 +284,21 @@ class ConsumerSet(object): self.init(None) def wait(self, limit=None): - while True: + running = True + while running: it = self.consumer_set.iterconsume(limit=limit) + if not it: + break while True: try: it.next() except StopIteration: return + except greenlet.GreenletExit: + running = False + break except Exception as e: - LOG.error(_("Received exception %s " % str(e) + \ + LOG.error(_("Received exception %s " % type(e) + \ "while processing consumer")) self.reconnect() # Break to outer loop diff --git a/nova/service.py b/nova/service.py index c51c9b066..a0ff7c9f3 100644 --- a/nova/service.py +++ b/nova/service.py @@ -21,12 +21,8 @@ import inspect import os -import sys -import time -from eventlet import event from eventlet import greenthread -from eventlet import greenpool from nova import context from nova import db @@ -112,7 +108,7 @@ class Service(object): consumer_node, fanout]) # Wait forever, processing these consumers - greenthread.spawn_n(cset.wait) + self.csetthread = greenthread.spawn(cset.wait) pulse = utils.LoopingCall(self.report_state) pulse.start(interval=self.report_interval, now=False) @@ -169,6 +165,8 @@ class Service(object): def kill(self): """Destroy the service object in the datastore.""" + self.csetthread.kill() + self.csetthread.wait() self.stop() try: db.service_destroy(context.get_admin_context(), self.service_id) diff --git a/nova/tests/test_cloud.py b/nova/tests/test_cloud.py index 54c0454de..1e14c327c 100644 --- a/nova/tests/test_cloud.py +++ b/nova/tests/test_cloud.py @@ -17,13 +17,8 @@ # under the License. from base64 import b64decode -import json from M2Crypto import BIO from M2Crypto import RSA -import os -import shutil -import tempfile -import time from eventlet import greenthread @@ -33,12 +28,10 @@ from nova import db from nova import flags from nova import log as logging from nova import rpc -from nova import service from nova import test from nova import utils from nova import exception from nova.auth import manager -from nova.compute import power_state from nova.api.ec2 import cloud from nova.api.ec2 import ec2utils from nova.image import local @@ -79,6 +72,15 @@ class CloudTestCase(test.TestCase): self.stubs.Set(local.LocalImageService, 'show', fake_show) self.stubs.Set(local.LocalImageService, 'show_by_name', fake_show) + # NOTE(vish): set up a manual wait so rpc.cast has a chance to finish + rpc_cast = rpc.cast + + def finish_cast(*args, **kwargs): + rpc_cast(*args, **kwargs) + greenthread.sleep(0.2) + + self.stubs.Set(rpc, 'cast', finish_cast) + def tearDown(self): network_ref = db.project_get_network(self.context, self.project.id) @@ -113,7 +115,6 @@ class CloudTestCase(test.TestCase): self.cloud.describe_addresses(self.context) self.cloud.release_address(self.context, public_ip=address) - greenthread.sleep(0.3) db.floating_ip_destroy(self.context, address) def test_associate_disassociate_address(self): @@ -129,12 +130,10 @@ class CloudTestCase(test.TestCase): self.cloud.associate_address(self.context, instance_id=ec2_id, public_ip=address) - greenthread.sleep(0.3) self.cloud.disassociate_address(self.context, public_ip=address) self.cloud.release_address(self.context, public_ip=address) - greenthread.sleep(0.3) self.network.deallocate_fixed_ip(self.context, fixed) db.instance_destroy(self.context, inst['id']) db.floating_ip_destroy(self.context, address) @@ -306,31 +305,26 @@ class CloudTestCase(test.TestCase): 'instance_type': instance_type, 'max_count': max_count} rv = self.cloud.run_instances(self.context, **kwargs) - greenthread.sleep(0.3) instance_id = rv['instancesSet'][0]['instanceId'] output = self.cloud.get_console_output(context=self.context, instance_id=[instance_id]) self.assertEquals(b64decode(output['output']), 'FAKE CONSOLE?OUTPUT') # TODO(soren): We need this until we can stop polling in the rpc code # for unit tests. - greenthread.sleep(0.3) rv = self.cloud.terminate_instances(self.context, [instance_id]) - greenthread.sleep(0.3) def test_ajax_console(self): + kwargs = {'image_id': 'ami-1'} rv = self.cloud.run_instances(self.context, **kwargs) instance_id = rv['instancesSet'][0]['instanceId'] - greenthread.sleep(0.3) output = self.cloud.get_ajax_console(context=self.context, instance_id=[instance_id]) self.assertEquals(output['url'], '%s/?token=FAKETOKEN' % FLAGS.ajax_console_proxy_url) # TODO(soren): We need this until we can stop polling in the rpc code # for unit tests. - greenthread.sleep(0.3) rv = self.cloud.terminate_instances(self.context, [instance_id]) - greenthread.sleep(0.3) def test_key_generation(self): result = self._create_key('test') -- cgit From 8f2557dcd3e3d88c0eabb63bcce90ced79347ae4 Mon Sep 17 00:00:00 2001 From: Chris Behrens Date: Wed, 25 May 2011 15:42:24 -0700 Subject: catch greenlet.GreenletExit when shutting service down --- nova/rpc.py | 2 +- nova/service.py | 6 +++++- 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/nova/rpc.py b/nova/rpc.py index e2e962fcc..02052ecf5 100644 --- a/nova/rpc.py +++ b/nova/rpc.py @@ -24,13 +24,13 @@ No fan-out support yet. """ +import greenlet import json import sys import time import traceback import uuid -import greenlet from carrot import connection as carrot_connection from carrot import messaging import eventlet diff --git a/nova/service.py b/nova/service.py index a0ff7c9f3..c7e48544c 100644 --- a/nova/service.py +++ b/nova/service.py @@ -19,6 +19,7 @@ """Generic Node baseclass for all workers that run on hosts.""" +import greenlet import inspect import os @@ -166,7 +167,10 @@ class Service(object): def kill(self): """Destroy the service object in the datastore.""" self.csetthread.kill() - self.csetthread.wait() + try: + self.csetthread.wait() + except greenlet.GreenletExit: + pass self.stop() try: db.service_destroy(context.get_admin_context(), self.service_id) -- cgit From 5f3adfc3110ed8095cdac43cc651aa46087c5490 Mon Sep 17 00:00:00 2001 From: Chris Behrens Date: Wed, 25 May 2011 15:42:24 -0700 Subject: Always create Service consumers no matter if report_interval is 0 Fix tests to handle how Service loads Consumers now --- nova/service.py | 46 +++++++++++++++++++------------------ nova/tests/test_service.py | 57 +++++++++++++++++++++++++++++++++++++++++----- 2 files changed, 75 insertions(+), 28 deletions(-) diff --git a/nova/service.py b/nova/service.py index c7e48544c..3a364b6c6 100644 --- a/nova/service.py +++ b/nova/service.py @@ -88,29 +88,31 @@ class Service(object): if 'nova-compute' == self.binary: self.manager.update_available_resource(ctxt) - if self.report_interval: - conn = rpc.Connection.instance(new=True) - - # Share this same connection for these Consumers - consumer_all = rpc.TopicAdapterConsumer( - connection=conn, - topic=self.topic, - proxy=self) - consumer_node = rpc.TopicAdapterConsumer( - connection=conn, - topic='%s.%s' % (self.topic, self.host), - proxy=self) - fanout = rpc.FanoutAdapterConsumer( - connection=conn, - topic=self.topic, - proxy=self) - - cset = rpc.ConsumerSet(conn, [consumer_all, - consumer_node, - fanout]) - # Wait forever, processing these consumers - self.csetthread = greenthread.spawn(cset.wait) + conn = rpc.Connection.instance(new=True) + logging.debug("Creating Consumer connection for Service %s" % \ + self.topic) + + # Share this same connection for these Consumers + consumer_all = rpc.TopicAdapterConsumer( + connection=conn, + topic=self.topic, + proxy=self) + consumer_node = rpc.TopicAdapterConsumer( + connection=conn, + topic='%s.%s' % (self.topic, self.host), + proxy=self) + fanout = rpc.FanoutAdapterConsumer( + connection=conn, + topic=self.topic, + proxy=self) + + cset = rpc.ConsumerSet(conn, [consumer_all, + consumer_node, + fanout]) + # Wait forever, processing these consumers + self.csetthread = greenthread.spawn(cset.wait) + if self.report_interval: pulse = utils.LoopingCall(self.report_state) pulse.start(interval=self.report_interval, now=False) self.timers.append(pulse) diff --git a/nova/tests/test_service.py b/nova/tests/test_service.py index d48de2057..0bba01d92 100644 --- a/nova/tests/test_service.py +++ b/nova/tests/test_service.py @@ -106,7 +106,10 @@ class ServiceTestCase(test.TestCase): # NOTE(vish): Create was moved out of mox replay to make sure that # the looping calls are created in StartService. - app = service.Service.create(host=host, binary=binary) + app = service.Service.create(host=host, binary=binary, topic=topic) + + self.mox.StubOutWithMock(service.rpc.Connection, 'instance') + service.rpc.Connection.instance(new=mox.IgnoreArg()) self.mox.StubOutWithMock(rpc, 'TopicAdapterConsumer', @@ -114,6 +117,11 @@ class ServiceTestCase(test.TestCase): self.mox.StubOutWithMock(rpc, 'FanoutAdapterConsumer', use_mock_anything=True) + + self.mox.StubOutWithMock(rpc, + 'ConsumerSet', + use_mock_anything=True) + rpc.TopicAdapterConsumer(connection=mox.IgnoreArg(), topic=topic, proxy=mox.IsA(service.Service)).AndReturn( @@ -129,9 +137,13 @@ class ServiceTestCase(test.TestCase): proxy=mox.IsA(service.Service)).AndReturn( rpc.FanoutAdapterConsumer) - rpc.TopicAdapterConsumer.attach_to_eventlet() - rpc.TopicAdapterConsumer.attach_to_eventlet() - rpc.FanoutAdapterConsumer.attach_to_eventlet() + def wait_func(self, limit=None): + return None + + mock_cset = self.mox.CreateMock(rpc.ConsumerSet, + {'wait': wait_func}) + rpc.ConsumerSet(mox.IgnoreArg(), mox.IsA(list)).AndReturn(mock_cset) + wait_func(mox.IgnoreArg()) service_create = {'host': host, 'binary': binary, @@ -287,8 +299,41 @@ class ServiceTestCase(test.TestCase): # Creating mocks self.mox.StubOutWithMock(service.rpc.Connection, 'instance') service.rpc.Connection.instance(new=mox.IgnoreArg()) - service.rpc.Connection.instance(new=mox.IgnoreArg()) - service.rpc.Connection.instance(new=mox.IgnoreArg()) + + self.mox.StubOutWithMock(rpc, + 'TopicAdapterConsumer', + use_mock_anything=True) + self.mox.StubOutWithMock(rpc, + 'FanoutAdapterConsumer', + use_mock_anything=True) + + self.mox.StubOutWithMock(rpc, + 'ConsumerSet', + use_mock_anything=True) + + rpc.TopicAdapterConsumer(connection=mox.IgnoreArg(), + topic=topic, + proxy=mox.IsA(service.Service)).AndReturn( + rpc.TopicAdapterConsumer) + + rpc.TopicAdapterConsumer(connection=mox.IgnoreArg(), + topic='%s.%s' % (topic, host), + proxy=mox.IsA(service.Service)).AndReturn( + rpc.TopicAdapterConsumer) + + rpc.FanoutAdapterConsumer(connection=mox.IgnoreArg(), + topic=topic, + proxy=mox.IsA(service.Service)).AndReturn( + rpc.FanoutAdapterConsumer) + + def wait_func(self, limit=None): + return None + + mock_cset = self.mox.CreateMock(rpc.ConsumerSet, + {'wait': wait_func}) + rpc.ConsumerSet(mox.IgnoreArg(), mox.IsA(list)).AndReturn(mock_cset) + wait_func(mox.IgnoreArg()) + self.mox.StubOutWithMock(serv.manager.driver, 'update_available_resource') serv.manager.driver.update_available_resource(mox.IgnoreArg(), host) -- cgit From 11d3672ad655c39265e5d2477a30db3a12adc65c Mon Sep 17 00:00:00 2001 From: Chris Behrens Date: Wed, 25 May 2011 15:42:24 -0700 Subject: Add rpc_conn_pool_size flag for the new connection pool --- nova/rpc.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/nova/rpc.py b/nova/rpc.py index 02052ecf5..82869fc46 100644 --- a/nova/rpc.py +++ b/nova/rpc.py @@ -50,7 +50,10 @@ LOG = logging.getLogger('nova.rpc') FLAGS = flags.FLAGS -flags.DEFINE_integer('rpc_thread_pool_size', 1024, 'Size of RPC thread pool') +flags.DEFINE_integer('rpc_thread_pool_size', 1024, + 'Size of RPC thread pool') +flags.DEFINE_integer('rpc_conn_pool_size', 30, + 'Size of RPC connection pool') class Connection(carrot_connection.BrokerConnection): @@ -99,7 +102,7 @@ class Pool(pools.Pool): def create(self): return Connection.instance(new=True) -ConnectionPool = Pool(max_size=20) +ConnectionPool = Pool(max_size=FLAGS.rpc_conn_pool_size) class Consumer(messaging.Consumer): -- cgit From b193b97054f11664a72cd53547f355d1c9044f88 Mon Sep 17 00:00:00 2001 From: Chris Behrens Date: Wed, 25 May 2011 15:42:24 -0700 Subject: connection pool tests and make the pool LIFO --- nova/rpc.py | 8 +++++++- nova/tests/test_rpc.py | 42 ++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 49 insertions(+), 1 deletion(-) diff --git a/nova/rpc.py b/nova/rpc.py index 82869fc46..3cc0dadd4 100644 --- a/nova/rpc.py +++ b/nova/rpc.py @@ -99,10 +99,16 @@ class Connection(carrot_connection.BrokerConnection): class Pool(pools.Pool): """Class that implements a Pool of Connections""" + # TODO(comstud): Timeout connections not used in a while def create(self): return Connection.instance(new=True) -ConnectionPool = Pool(max_size=FLAGS.rpc_conn_pool_size) +# Create a ConnectionPool to use for RPC calls. We'll order the +# pool as a stack (LIFO), so that we can potentially loop through and +# timeout old unused connections at some point +ConnectionPool = Pool( + max_size=FLAGS.rpc_conn_pool_size, + order_as_stack=True) class Consumer(messaging.Consumer): diff --git a/nova/tests/test_rpc.py b/nova/tests/test_rpc.py index acab3e758..f64209596 100644 --- a/nova/tests/test_rpc.py +++ b/nova/tests/test_rpc.py @@ -120,6 +120,48 @@ class RpcTestCase(test.TestCase): "value": value}}) self.assertEqual(value, result) + def test_connectionpool_single(self): + """Test that ConnectionPool recycles a single connection""" + + conn1 = rpc.ConnectionPool.get() + rpc.ConnectionPool.put(conn1) + conn2 = rpc.ConnectionPool.get() + rpc.ConnectionPool.put(conn2) + self.assertEqual(conn1, conn2) + + def test_connectionpool_double(self): + """Test that ConnectionPool returns 2 separate connections + when called consecutively and the pool returns connections LIFO + """ + + conn1 = rpc.ConnectionPool.get() + conn2 = rpc.ConnectionPool.get() + + self.assertNotEqual(conn1, conn2) + rpc.ConnectionPool.put(conn1) + rpc.ConnectionPool.put(conn2) + + conn3 = rpc.ConnectionPool.get() + conn4 = rpc.ConnectionPool.get() + self.assertEqual(conn2, conn3) + self.assertEqual(conn1, conn4) + + def test_connectionpool_limit(self): + """Test connection pool limit and verify all connections + are unique + """ + + max_size = FLAGS.rpc_conn_pool_size + conns = [] + + for i in xrange(max_size): + conns.append(rpc.ConnectionPool.get()) + + self.assertFalse(rpc.ConnectionPool.free_items) + self.assertEqual(rpc.ConnectionPool.current_size, + rpc.ConnectionPool.max_size) + self.assertEqual(len(set(conns)), max_size) + class TestReceiver(object): """Simple Proxy class so the consumer has methods to call -- cgit From 51e8eeb9b3a23f811bcbf52d9700d94c5c8b15e4 Mon Sep 17 00:00:00 2001 From: termie Date: Wed, 25 May 2011 15:42:24 -0700 Subject: bring back commits lost in merge --- nova/rpc.py | 107 +++++++++++++++++++++++++++++-------------------- nova/tests/test_rpc.py | 19 +++++++++ 2 files changed, 82 insertions(+), 44 deletions(-) diff --git a/nova/rpc.py b/nova/rpc.py index 3cc0dadd4..d7d7bb014 100644 --- a/nova/rpc.py +++ b/nova/rpc.py @@ -35,6 +35,7 @@ from carrot import connection as carrot_connection from carrot import messaging import eventlet from eventlet import greenpool +from eventlet import greenthread from eventlet import pools from eventlet import queue @@ -140,30 +141,30 @@ class Consumer(messaging.Consumer): FLAGS.rabbit_max_retries) sys.exit(1) - def fetch(self, no_ack=None, auto_ack=None, enable_callbacks=False): - """Wraps the parent fetch with some logic for failed connection.""" - # TODO(vish): the logic for failed connections and logging should be - # refactored into some sort of connection manager object - try: - if self.failed_connection: - # NOTE(vish): connection is defined in the parent class, we can - # recreate it as long as we create the backend too - # pylint: disable=W0201 - self.connection = Connection.recreate() - self.backend = self.connection.create_backend() - self.declare() - return super(Consumer, self).fetch( - no_ack, auto_ack, enable_callbacks) - if self.failed_connection: - LOG.error(_('Reconnected to queue')) - self.failed_connection = False - # NOTE(vish): This is catching all errors because we really don't - # want exceptions to be logged 10 times a second if some - # persistent failure occurs. - except Exception, e: # pylint: disable=W0703 - if not self.failed_connection: - LOG.exception(_('Failed to fetch message from queue: %s' % e)) - self.failed_connection = True + #def fetch(self, no_ack=None, auto_ack=None, enable_callbacks=False): + # """Wraps the parent fetch with some logic for failed connection.""" + # # TODO(vish): the logic for failed connections and logging should be + # # refactored into some sort of connection manager object + # try: + # if self.failed_connection: + # # NOTE(vish): connection is defined in the parent class, we can + # # recreate it as long as we create the backend too + # # pylint: disable=W0201 + # self.connection = Connection.recreate() + # self.backend = self.connection.create_backend() + # self.declare() + # return super(Consumer, self).fetch( + # no_ack, auto_ack, enable_callbacks) + # if self.failed_connection: + # LOG.error(_('Reconnected to queue')) + # self.failed_connection = False + # # NOTE(vish): This is catching all errors because we really don't + # # want exceptions to be logged 10 times a second if some + # # persistent failure occurs. + # except Exception, e: # pylint: disable=W0703 + # if not self.failed_connection: + # LOG.exception(_('Failed to fetch message from queue: %s' % e)) + # self.failed_connection = True def attach_to_eventlet(self): """Only needed for unit tests!""" @@ -195,7 +196,7 @@ class AdapterConsumer(Consumer): """ LOG.debug(_('received %s') % message_data) - msg_id = message_data.pop('_msg_id', None) + msg_id = message_data.get('_msg_id', None) ctxt = _unpack_context(message_data) @@ -225,11 +226,14 @@ class AdapterConsumer(Consumer): rval = node_func(context=ctxt, **node_args) if msg_id: # TODO(termie): re-enable when fix the yielding issue - #if hasattr(rval, 'send'): - # logging.error('rval! %s', rval) - # for x in rval: - # msg_reply(msg_id, x, None) - msg_reply(msg_id, rval, None) + if hasattr(rval, 'send'): + logging.error('rval! %s', rval) + for x in rval: + msg_reply(msg_id, x, None) + msg_reply(msg_id, None, None) + else: + msg_reply(msg_id, rval, None) + #msg_reply(msg_id, rval, None) except Exception as e: logging.exception('Exception during message handling') if msg_id: @@ -355,7 +359,7 @@ class DirectConsumer(Consumer): self.routing_key = msg_id self.exchange = msg_id self.auto_delete = True - self.exclusive = True + self.exclusive = False super(DirectConsumer, self).__init__(connection=connection) @@ -387,7 +391,9 @@ def msg_reply(msg_id, reply=None, failure=None): publisher = DirectPublisher(connection=conn, msg_id=msg_id) try: publisher.send({'result': reply, 'failure': failure}) + LOG.error('MSG REPLY SUCCESS') except TypeError: + LOG.error('MSG REPLY FAILURE') publisher.send( {'result': dict((k, repr(v)) for k, v in reply.__dict__.iteritems()), @@ -440,9 +446,9 @@ def _pack_context(msg, context): for args at some point. """ - context = dict([('_context_%s' % key, value) - for (key, value) in context.to_dict().iteritems()]) - msg.update(context) + context_d = dict([('_context_%s' % key, value) + for (key, value) in context.to_dict().iteritems()]) + msg.update(context_d) class RpcContext(context.RequestContext): @@ -463,12 +469,13 @@ def multicall(context, topic, msg): LOG.debug(_('MSG_ID is %s') % (msg_id)) _pack_context(msg, context) - conn = ConnectionPool.get() - consumer = DirectConsumer(connection=conn, msg_id=msg_id) + con_conn = ConnectionPool.get() + consumer = DirectConsumer(connection=con_conn, msg_id=msg_id) wait_msg = MulticallWaiter(consumer) consumer.register_callback(wait_msg) - publisher = TopicPublisher(connection=conn, topic=topic) + pub_conn = ConnectionPool.get() + publisher = TopicPublisher(connection=pub_conn, topic=topic) publisher.send(msg) publisher.close() @@ -484,6 +491,7 @@ class MulticallWaiter(object): def close(self): self._closed = True self._consumer.close() + ConnectionPool.put(self._consumer.connection) def __call__(self, data, message): """Acks message and sets result.""" @@ -501,15 +509,26 @@ class MulticallWaiter(object): # trying to solve the problem quickly. This works but # I'd prefer to dig in and do it the best way later on. - def _waiter(): - while not self._closed: - try: - self._consumer.wait(limit=1) - except StopIteration: - pass - eventlet.spawn(_waiter) + #def _waiter(): + # i = 0 + # while not self._closed: + # LOG.error('Iteration #%s (%s)', i, self._consumer.consumer_tag) + # i += 1 + # try: + # self._consumer.wait(limit=1) + # except StopIteration: + # pass + # self._consumer.close() + # ConnectionPool.put(self._consumer.connection) + #eventlet.spawn(_waiter) while True: + rv = None + while rv is None and not self._closed: + rv = self._consumer.fetch(enable_callbacks=True) + time.sleep(0.01) + + LOG.error('RV %s', rv) result = self._results.get() if isinstance(result, Exception): raise result diff --git a/nova/tests/test_rpc.py b/nova/tests/test_rpc.py index f64209596..e5d99474d 100644 --- a/nova/tests/test_rpc.py +++ b/nova/tests/test_rpc.py @@ -61,6 +61,18 @@ class RpcTestCase(test.TestCase): self.assertEqual(value + i, x) i += 1 + def test_multicall_succeed_three_times_yield(self): + """Get a value through rpc call""" + value = 42 + result = rpc.multicall(self.context, + 'test', + {"method": "echo_three_times_yield", + "args": {"value": value}}) + i = 0 + for x in result: + self.assertEqual(value + i, x) + i += 1 + def test_context_passed(self): """Makes sure a context is passed through rpc call""" value = 42 @@ -83,6 +95,7 @@ class RpcTestCase(test.TestCase): 'test', {"method": "fail", "args": {"value": value}}) + LOG.error('INNNNNNN BETTTWWWWWWWWWWEEEEEEEEEEN') try: rpc.call(self.context, 'test', @@ -186,6 +199,12 @@ class TestReceiver(object): context.reply(value + 1) context.reply(value + 2) + @staticmethod + def echo_three_times_yield(context, value): + yield value + yield value + 1 + yield value + 2 + @staticmethod def fail(context, value): """Raises an exception with the value sent in""" -- cgit From 64b13a2aad676d2310947e3bf8b9e3dde6b763e7 Mon Sep 17 00:00:00 2001 From: termie Date: Wed, 25 May 2011 15:42:24 -0700 Subject: almost everything working with fake_rabbit --- nova/rpc.py | 16 +++++++++++++++- nova/service.py | 22 ++++++++++++++++------ nova/test.py | 1 + nova/tests/test_cloud.py | 4 ++-- run_tests.py | 1 + 5 files changed, 35 insertions(+), 9 deletions(-) diff --git a/nova/rpc.py b/nova/rpc.py index d7d7bb014..e1f594a99 100644 --- a/nova/rpc.py +++ b/nova/rpc.py @@ -102,6 +102,7 @@ class Pool(pools.Pool): # TODO(comstud): Timeout connections not used in a while def create(self): + LOG.debug('Creating new connection') return Connection.instance(new=True) # Create a ConnectionPool to use for RPC calls. We'll order the @@ -166,6 +167,10 @@ class Consumer(messaging.Consumer): # LOG.exception(_('Failed to fetch message from queue: %s' % e)) # self.failed_connection = True + def close(self, *args, **kwargs): + LOG.debug('Closing consumer %s', self.consumer_tag) + return super(Consumer, self).close(*args, **kwargs) + def attach_to_eventlet(self): """Only needed for unit tests!""" timer = utils.LoopingCall(self.fetch, enable_callbacks=True) @@ -317,6 +322,8 @@ class ConsumerSet(object): # Break to outer loop break + def close(self): + self.consumer_set.close() class Publisher(messaging.Publisher): """Publisher base class.""" @@ -525,12 +532,19 @@ class MulticallWaiter(object): while True: rv = None while rv is None and not self._closed: - rv = self._consumer.fetch(enable_callbacks=True) + try: + rv = self._consumer.fetch(enable_callbacks=True) + except Exception: + self.close() + raise + #rv = self._consumer.fetch(enable_callbacks=True) time.sleep(0.01) LOG.error('RV %s', rv) result = self._results.get() + LOG.error('RESULT %s', result) if isinstance(result, Exception): + self.close() raise result if result == None: self.close() diff --git a/nova/service.py b/nova/service.py index 3a364b6c6..2da510140 100644 --- a/nova/service.py +++ b/nova/service.py @@ -88,29 +88,39 @@ class Service(object): if 'nova-compute' == self.binary: self.manager.update_available_resource(ctxt) - conn = rpc.Connection.instance(new=True) + conn1 = rpc.Connection.instance(new=True) + conn2 = rpc.Connection.instance(new=True) + conn3 = rpc.Connection.instance(new=True) logging.debug("Creating Consumer connection for Service %s" % \ self.topic) # Share this same connection for these Consumers consumer_all = rpc.TopicAdapterConsumer( - connection=conn, + connection=conn1, topic=self.topic, proxy=self) consumer_node = rpc.TopicAdapterConsumer( - connection=conn, + connection=conn1, topic='%s.%s' % (self.topic, self.host), proxy=self) fanout = rpc.FanoutAdapterConsumer( - connection=conn, + connection=conn1, topic=self.topic, proxy=self) - cset = rpc.ConsumerSet(conn, [consumer_all, + cset = rpc.ConsumerSet(conn1, [consumer_all, consumer_node, fanout]) # Wait forever, processing these consumers - self.csetthread = greenthread.spawn(cset.wait) + def _wait(): + cset.wait() + cset.close() + + self.csetthread = greenthread.spawn(_wait) + + #self.timers.append(consumer_all.attach_to_eventlet()) + #self.timers.append(consumer_node.attach_to_eventlet()) + #self.timers.append(fanout.attach_to_eventlet()) if self.report_interval: pulse = utils.LoopingCall(self.report_state) diff --git a/nova/test.py b/nova/test.py index 4deb2a175..7b2cf94b6 100644 --- a/nova/test.py +++ b/nova/test.py @@ -85,6 +85,7 @@ class TestCase(unittest.TestCase): self._monkey_patch_attach() self._monkey_patch_wsgi() self._original_flags = FLAGS.FlagValuesDict() + rpc.ConnectionPool = rpc.Pool(max_size=30) def tearDown(self): """Runs after each test method to tear down test environment.""" diff --git a/nova/tests/test_cloud.py b/nova/tests/test_cloud.py index 1e14c327c..a838dd530 100644 --- a/nova/tests/test_cloud.py +++ b/nova/tests/test_cloud.py @@ -87,8 +87,8 @@ class CloudTestCase(test.TestCase): db.network_disassociate(self.context, network_ref['id']) self.manager.delete_project(self.project) self.manager.delete_user(self.user) - self.compute.kill() - self.network.kill() + #self.compute.kill() + #self.network.kill() super(CloudTestCase, self).tearDown() def _create_key(self, name): diff --git a/run_tests.py b/run_tests.py index d5d8acd16..509a60caa 100644 --- a/run_tests.py +++ b/run_tests.py @@ -285,6 +285,7 @@ if __name__ == '__main__': # If any argument looks like a test name but doesn't have "nova.tests" in # front of it, automatically add that so we don't have to type as much argv = [] + logging.getLogger('amqplib').setLevel(logging.DEBUG) for x in sys.argv: if x.startswith('test_'): argv.append('nova.tests.%s' % x) -- cgit From e3a88390fd62308cde3d4c597d653c8dc245bed4 Mon Sep 17 00:00:00 2001 From: termie Date: Wed, 25 May 2011 15:42:24 -0700 Subject: don't need to use a separate connection --- nova/rpc.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/nova/rpc.py b/nova/rpc.py index e1f594a99..a212383fd 100644 --- a/nova/rpc.py +++ b/nova/rpc.py @@ -481,8 +481,7 @@ def multicall(context, topic, msg): wait_msg = MulticallWaiter(consumer) consumer.register_callback(wait_msg) - pub_conn = ConnectionPool.get() - publisher = TopicPublisher(connection=pub_conn, topic=topic) + publisher = TopicPublisher(connection=con_conn, topic=topic) publisher.send(msg) publisher.close() -- cgit From c9b21b0619891c069251c568e4d89be791af56c3 Mon Sep 17 00:00:00 2001 From: Vishvananda Ishaya Date: Wed, 25 May 2011 15:42:24 -0700 Subject: lots of fixes for rpc and extra imports --- nova/fakerabbit.py | 12 +++-- nova/rpc.py | 68 +++++++++++------------------ nova/service.py | 9 ++-- nova/test.py | 8 ++-- nova/tests/integrated/integrated_helpers.py | 5 +-- 5 files changed, 45 insertions(+), 57 deletions(-) diff --git a/nova/fakerabbit.py b/nova/fakerabbit.py index 5f3e75c48..ff993e29a 100644 --- a/nova/fakerabbit.py +++ b/nova/fakerabbit.py @@ -31,6 +31,7 @@ LOG = logging.getLogger("nova.fakerabbit") EXCHANGES = {} QUEUES = {} +CONSUMERS = {} class Message(base.BaseMessage): @@ -101,17 +102,20 @@ class Backend(base.BaseBackend): EXCHANGES[exchange].bind(QUEUES[queue].push, routing_key) def declare_consumer(self, queue, callback, consumer_tag, *args, **kwargs): + global CONSUMERS LOG.debug("Adding consumer %s", consumer_tag) - self.consumers[consumer_tag] = (queue, callback) + CONSUMERS[consumer_tag] = (queue, callback) def cancel(self, consumer_tag): + global CONSUMERS LOG.debug("Removing consumer %s", consumer_tag) - del self.consumers[consumer_tag] + del CONSUMERS[consumer_tag] def consume(self, limit=None): + global CONSUMERS num = 0 while True: - for (queue, callback) in self.consumers.itervalues(): + for (queue, callback) in CONSUMERS.itervalues(): item = self.get(queue) if item: callback(item) @@ -147,5 +151,7 @@ class Backend(base.BaseBackend): def reset_all(): global EXCHANGES global QUEUES + global CONSUMERS EXCHANGES = {} QUEUES = {} + CONSUMERS = {} diff --git a/nova/rpc.py b/nova/rpc.py index a212383fd..7faed4d3a 100644 --- a/nova/rpc.py +++ b/nova/rpc.py @@ -33,9 +33,7 @@ import uuid from carrot import connection as carrot_connection from carrot import messaging -import eventlet from eventlet import greenpool -from eventlet import greenthread from eventlet import pools from eventlet import queue @@ -142,30 +140,30 @@ class Consumer(messaging.Consumer): FLAGS.rabbit_max_retries) sys.exit(1) - #def fetch(self, no_ack=None, auto_ack=None, enable_callbacks=False): - # """Wraps the parent fetch with some logic for failed connection.""" - # # TODO(vish): the logic for failed connections and logging should be - # # refactored into some sort of connection manager object - # try: - # if self.failed_connection: - # # NOTE(vish): connection is defined in the parent class, we can - # # recreate it as long as we create the backend too - # # pylint: disable=W0201 - # self.connection = Connection.recreate() - # self.backend = self.connection.create_backend() - # self.declare() - # return super(Consumer, self).fetch( - # no_ack, auto_ack, enable_callbacks) - # if self.failed_connection: - # LOG.error(_('Reconnected to queue')) - # self.failed_connection = False - # # NOTE(vish): This is catching all errors because we really don't - # # want exceptions to be logged 10 times a second if some - # # persistent failure occurs. - # except Exception, e: # pylint: disable=W0703 - # if not self.failed_connection: - # LOG.exception(_('Failed to fetch message from queue: %s' % e)) - # self.failed_connection = True + def fetch(self, no_ack=None, auto_ack=None, enable_callbacks=False): + """Wraps the parent fetch with some logic for failed connection.""" + # TODO(vish): the logic for failed connections and logging should be + # refactored into some sort of connection manager object + try: + if self.failed_connection: + # NOTE(vish): connection is defined in the parent class, we can + # recreate it as long as we create the backend too + # pylint: disable=W0201 + self.connection = Connection.recreate() + self.backend = self.connection.create_backend() + self.declare() + return super(Consumer, self).fetch( + no_ack, auto_ack, enable_callbacks) + if self.failed_connection: + LOG.error(_('Reconnected to queue')) + self.failed_connection = False + # NOTE(vish): This is catching all errors because we really don't + # want exceptions to be logged 10 times a second if some + # persistent failure occurs. + except Exception, e: # pylint: disable=W0703 + if not self.failed_connection: + LOG.exception(_('Failed to fetch message from queue: %s' % e)) + self.failed_connection = True def close(self, *args, **kwargs): LOG.debug('Closing consumer %s', self.consumer_tag) @@ -325,6 +323,7 @@ class ConsumerSet(object): def close(self): self.consumer_set.close() + class Publisher(messaging.Publisher): """Publisher base class.""" pass @@ -511,23 +510,6 @@ class MulticallWaiter(object): return self.wait() def wait(self): - # TODO(termie): This is probably really a much simpler issue but am - # trying to solve the problem quickly. This works but - # I'd prefer to dig in and do it the best way later on. - - #def _waiter(): - # i = 0 - # while not self._closed: - # LOG.error('Iteration #%s (%s)', i, self._consumer.consumer_tag) - # i += 1 - # try: - # self._consumer.wait(limit=1) - # except StopIteration: - # pass - # self._consumer.close() - # ConnectionPool.put(self._consumer.connection) - #eventlet.spawn(_waiter) - while True: rv = None while rv is None and not self._closed: diff --git a/nova/service.py b/nova/service.py index 2da510140..2626c49ae 100644 --- a/nova/service.py +++ b/nova/service.py @@ -89,8 +89,6 @@ class Service(object): self.manager.update_available_resource(ctxt) conn1 = rpc.Connection.instance(new=True) - conn2 = rpc.Connection.instance(new=True) - conn3 = rpc.Connection.instance(new=True) logging.debug("Creating Consumer connection for Service %s" % \ self.topic) @@ -111,10 +109,13 @@ class Service(object): cset = rpc.ConsumerSet(conn1, [consumer_all, consumer_node, fanout]) + # Wait forever, processing these consumers def _wait(): - cset.wait() - cset.close() + try: + cset.wait() + finally: + cset.close() self.csetthread = greenthread.spawn(_wait) diff --git a/nova/test.py b/nova/test.py index 7b2cf94b6..df48afbb1 100644 --- a/nova/test.py +++ b/nova/test.py @@ -31,17 +31,15 @@ import uuid import unittest import mox -import shutil import stubout from eventlet import greenthread -from nova import context -from nova import db from nova import fakerabbit from nova import flags from nova import rpc from nova import service from nova import wsgi +from nova.virt import fake FLAGS = flags.FLAGS @@ -100,6 +98,10 @@ class TestCase(unittest.TestCase): if FLAGS.fake_rabbit: fakerabbit.reset_all() + if FLAGS.connection_type == 'fake': + if hasattr(fake.FakeConnection, '_instance'): + del fake.FakeConnection._instance + # Reset any overriden flags self.reset_flags() diff --git a/nova/tests/integrated/integrated_helpers.py b/nova/tests/integrated/integrated_helpers.py index bc98921f0..7f590441e 100644 --- a/nova/tests/integrated/integrated_helpers.py +++ b/nova/tests/integrated/integrated_helpers.py @@ -154,10 +154,7 @@ class _IntegratedTestBase(test.TestCase): # set up services self.start_service('compute') self.start_service('volume') - # NOTE(justinsb): There's a bug here which is eluding me... - # If we start the network_service, all is good, but then subsequent - # tests fail: CloudTestCase.test_ajax_console in particular. - #self.start_service('network') + self.start_service('network') self.start_service('scheduler') self._start_api_service() -- cgit From 9334d41c6fe638a3119327702094695cfbd38271 Mon Sep 17 00:00:00 2001 From: termie Date: Wed, 25 May 2011 15:42:25 -0700 Subject: make sure that using multicall on a call with a single result still functions --- nova/rpc.py | 4 ++-- nova/tests/test_rpc.py | 29 +++++++++++++++++++++++++++++ 2 files changed, 31 insertions(+), 2 deletions(-) diff --git a/nova/rpc.py b/nova/rpc.py index 7faed4d3a..84493271f 100644 --- a/nova/rpc.py +++ b/nova/rpc.py @@ -233,10 +233,10 @@ class AdapterConsumer(Consumer): logging.error('rval! %s', rval) for x in rval: msg_reply(msg_id, x, None) - msg_reply(msg_id, None, None) else: msg_reply(msg_id, rval, None) - #msg_reply(msg_id, rval, None) + # This final None tells multicall that it is done. + msg_reply(msg_id, None, None) except Exception as e: logging.exception('Exception during message handling') if msg_id: diff --git a/nova/tests/test_rpc.py b/nova/tests/test_rpc.py index e5d99474d..c1ef60ff6 100644 --- a/nova/tests/test_rpc.py +++ b/nova/tests/test_rpc.py @@ -49,6 +49,35 @@ class RpcTestCase(test.TestCase): "args": {"value": value}}) self.assertEqual(value, result) + def test_call_succeed_despite_multiple_returns(self): + """Get a value through rpc call""" + value = 42 + result = rpc.call(self.context, 'test', {"method": "echo_three_times", + "args": {"value": value}}) + self.assertEqual(value, result) + + def test_call_succeed_despite_multiple_returns_yield(self): + """Get a value through rpc call""" + value = 42 + result = rpc.call(self.context, 'test', + {"method": "echo_three_times_yield", + "args": {"value": value}}) + self.assertEqual(value, result) + + def test_multicall_succeed_once(self): + """Get a value through rpc call""" + value = 42 + result = rpc.multicall(self.context, + 'test', + {"method": "echo", + "args": {"value": value}}) + i = 0 + for x in result: + if i > 0: + self.fail('should only receive one response') + self.assertEqual(value + i, x) + i += 1 + def test_multicall_succeed_three_times(self): """Get a value through rpc call""" value = 42 -- cgit From c7fe7e5e28b9f4bb999c8309f56953f6609cbc57 Mon Sep 17 00:00:00 2001 From: termie Date: Wed, 25 May 2011 15:42:49 -0700 Subject: cleanup the code for merging --- nova/fakerabbit.py | 4 --- nova/rpc.py | 78 ++++++++++++++++++++++-------------------------- nova/service.py | 23 +++++++------- nova/test.py | 2 +- nova/tests/test_cloud.py | 3 -- nova/tests/test_rpc.py | 1 - run_tests.py | 1 - 7 files changed, 46 insertions(+), 66 deletions(-) diff --git a/nova/fakerabbit.py b/nova/fakerabbit.py index ff993e29a..e7e9dab77 100644 --- a/nova/fakerabbit.py +++ b/nova/fakerabbit.py @@ -78,10 +78,6 @@ class Queue(object): class Backend(base.BaseBackend): - def __init__(self, connection, **kwargs): - super(Backend, self).__init__(connection, **kwargs) - self.consumers = {} - def queue_declare(self, queue, **kwargs): global QUEUES if queue not in QUEUES: diff --git a/nova/rpc.py b/nova/rpc.py index 84493271f..8d14494f0 100644 --- a/nova/rpc.py +++ b/nova/rpc.py @@ -24,7 +24,6 @@ No fan-out support yet. """ -import greenlet import json import sys import time @@ -36,6 +35,7 @@ from carrot import messaging from eventlet import greenpool from eventlet import pools from eventlet import queue +import greenlet from nova import context from nova import exception @@ -50,9 +50,9 @@ LOG = logging.getLogger('nova.rpc') FLAGS = flags.FLAGS flags.DEFINE_integer('rpc_thread_pool_size', 1024, - 'Size of RPC thread pool') + 'Size of RPC thread pool') flags.DEFINE_integer('rpc_conn_pool_size', 30, - 'Size of RPC connection pool') + 'Size of RPC connection pool') class Connection(carrot_connection.BrokerConnection): @@ -96,7 +96,7 @@ class Connection(carrot_connection.BrokerConnection): class Pool(pools.Pool): - """Class that implements a Pool of Connections""" + """Class that implements a Pool of Connections.""" # TODO(comstud): Timeout connections not used in a while def create(self): @@ -152,8 +152,9 @@ class Consumer(messaging.Consumer): self.connection = Connection.recreate() self.backend = self.connection.create_backend() self.declare() - return super(Consumer, self).fetch( - no_ack, auto_ack, enable_callbacks) + return super(Consumer, self).fetch(no_ack, + auto_ack, + enable_callbacks) if self.failed_connection: LOG.error(_('Reconnected to queue')) self.failed_connection = False @@ -165,10 +166,6 @@ class Consumer(messaging.Consumer): LOG.exception(_('Failed to fetch message from queue: %s' % e)) self.failed_connection = True - def close(self, *args, **kwargs): - LOG.debug('Closing consumer %s', self.consumer_tag) - return super(Consumer, self).close(*args, **kwargs) - def attach_to_eventlet(self): """Only needed for unit tests!""" timer = utils.LoopingCall(self.fetch, enable_callbacks=True) @@ -188,8 +185,10 @@ class AdapterConsumer(Consumer): self.register_callback(self.process_data) def process_data(self, message_data, message): - """Consumer callback that parses the message for validity and - fires off a thread to call the proxy object method. + """Consumer callback to call a method on a proxy object. + + Parses the message for validity and fires off a thread to call the + proxy object method. Message data should be a dictionary with two keys: method: string representing the method to call @@ -199,8 +198,8 @@ class AdapterConsumer(Consumer): """ LOG.debug(_('received %s') % message_data) + # This will be popped off in _unpack_context msg_id = message_data.get('_msg_id', None) - ctxt = _unpack_context(message_data) method = message_data.get('method') @@ -228,13 +227,13 @@ class AdapterConsumer(Consumer): try: rval = node_func(context=ctxt, **node_args) if msg_id: - # TODO(termie): re-enable when fix the yielding issue + # Check if the result was a generator if hasattr(rval, 'send'): - logging.error('rval! %s', rval) for x in rval: msg_reply(msg_id, x, None) else: msg_reply(msg_id, rval, None) + # This final None tells multicall that it is done. msg_reply(msg_id, None, None) except Exception as e: @@ -277,7 +276,7 @@ class FanoutAdapterConsumer(AdapterConsumer): class ConsumerSet(object): - """Groups consumers to listen on together on a single connection""" + """Groups consumers to listen on together on a single connection.""" def __init__(self, conn, consumer_list): self.consumer_list = set(consumer_list) @@ -365,7 +364,7 @@ class DirectConsumer(Consumer): self.routing_key = msg_id self.exchange = msg_id self.auto_delete = True - self.exclusive = False + self.exclusive = True super(DirectConsumer, self).__init__(connection=connection) @@ -393,20 +392,18 @@ def msg_reply(msg_id, reply=None, failure=None): LOG.error(_("Returning exception %s to caller"), message) LOG.error(tb) failure = (failure[0].__name__, str(failure[1]), tb) - conn = ConnectionPool.get() - publisher = DirectPublisher(connection=conn, msg_id=msg_id) - try: - publisher.send({'result': reply, 'failure': failure}) - LOG.error('MSG REPLY SUCCESS') - except TypeError: - LOG.error('MSG REPLY FAILURE') - publisher.send( - {'result': dict((k, repr(v)) - for k, v in reply.__dict__.iteritems()), - 'failure': failure}) - publisher.close() - ConnectionPool.put(conn) + with ConnectionPool.item() as conn: + publisher = DirectPublisher(connection=conn, msg_id=msg_id) + try: + publisher.send({'result': reply, 'failure': failure}) + except TypeError: + publisher.send( + {'result': dict((k, repr(v)) + for k, v in reply.__dict__.iteritems()), + 'failure': failure}) + + publisher.close() class RemoteError(exception.Error): @@ -518,12 +515,9 @@ class MulticallWaiter(object): except Exception: self.close() raise - #rv = self._consumer.fetch(enable_callbacks=True) time.sleep(0.01) - LOG.error('RV %s', rv) result = self._results.get() - LOG.error('RESULT %s', result) if isinstance(result, Exception): self.close() raise result @@ -545,22 +539,20 @@ def cast(context, topic, msg): """Sends a message on a topic without waiting for a response.""" LOG.debug(_('Making asynchronous cast on %s...'), topic) _pack_context(msg, context) - conn = ConnectionPool.get() - publisher = TopicPublisher(connection=conn, topic=topic) - publisher.send(msg) - publisher.close() - ConnectionPool.put(conn) + with ConnectionPool.item() as conn: + publisher = TopicPublisher(connection=conn, topic=topic) + publisher.send(msg) + publisher.close() def fanout_cast(context, topic, msg): """Sends a message on a fanout exchange without waiting for a response.""" LOG.debug(_('Making asynchronous fanout cast...')) _pack_context(msg, context) - conn = ConnectionPool.get() - publisher = FanoutPublisher(topic, connection=conn) - publisher.send(msg) - publisher.close() - ConnectionPool.put(conn) + with ConnectionPool.item() as conn: + publisher = FanoutPublisher(topic, connection=conn) + publisher.send(msg) + publisher.close() def generic_response(message_data, message): diff --git a/nova/service.py b/nova/service.py index 2626c49ae..94afd5f78 100644 --- a/nova/service.py +++ b/nova/service.py @@ -88,27 +88,27 @@ class Service(object): if 'nova-compute' == self.binary: self.manager.update_available_resource(ctxt) - conn1 = rpc.Connection.instance(new=True) - logging.debug("Creating Consumer connection for Service %s" % \ - self.topic) + self.conn = rpc.Connection.instance(new=True) + logging.debug("Creating Consumer connection for Service %s" % + self.topic) # Share this same connection for these Consumers consumer_all = rpc.TopicAdapterConsumer( - connection=conn1, + connection=self.conn, topic=self.topic, proxy=self) consumer_node = rpc.TopicAdapterConsumer( - connection=conn1, + connection=self.conn, topic='%s.%s' % (self.topic, self.host), proxy=self) fanout = rpc.FanoutAdapterConsumer( - connection=conn1, + connection=self.conn, topic=self.topic, proxy=self) - cset = rpc.ConsumerSet(conn1, [consumer_all, - consumer_node, - fanout]) + cset = rpc.ConsumerSet(self.conn, [consumer_all, + consumer_node, + fanout]) # Wait forever, processing these consumers def _wait(): @@ -119,10 +119,6 @@ class Service(object): self.csetthread = greenthread.spawn(_wait) - #self.timers.append(consumer_all.attach_to_eventlet()) - #self.timers.append(consumer_node.attach_to_eventlet()) - #self.timers.append(fanout.attach_to_eventlet()) - if self.report_interval: pulse = utils.LoopingCall(self.report_state) pulse.start(interval=self.report_interval, now=False) @@ -185,6 +181,7 @@ class Service(object): except greenlet.GreenletExit: pass self.stop() + rpc.ConnectionPool.put(self.conn) try: db.service_destroy(context.get_admin_context(), self.service_id) except exception.NotFound: diff --git a/nova/test.py b/nova/test.py index df48afbb1..80b2d0a74 100644 --- a/nova/test.py +++ b/nova/test.py @@ -83,7 +83,7 @@ class TestCase(unittest.TestCase): self._monkey_patch_attach() self._monkey_patch_wsgi() self._original_flags = FLAGS.FlagValuesDict() - rpc.ConnectionPool = rpc.Pool(max_size=30) + rpc.ConnectionPool = rpc.Pool(max_size=FLAGS.rpc_conn_pool_size) def tearDown(self): """Runs after each test method to tear down test environment.""" diff --git a/nova/tests/test_cloud.py b/nova/tests/test_cloud.py index a838dd530..ca3ef7ffe 100644 --- a/nova/tests/test_cloud.py +++ b/nova/tests/test_cloud.py @@ -87,8 +87,6 @@ class CloudTestCase(test.TestCase): db.network_disassociate(self.context, network_ref['id']) self.manager.delete_project(self.project) self.manager.delete_user(self.user) - #self.compute.kill() - #self.network.kill() super(CloudTestCase, self).tearDown() def _create_key(self, name): @@ -314,7 +312,6 @@ class CloudTestCase(test.TestCase): rv = self.cloud.terminate_instances(self.context, [instance_id]) def test_ajax_console(self): - kwargs = {'image_id': 'ami-1'} rv = self.cloud.run_instances(self.context, **kwargs) instance_id = rv['instancesSet'][0]['instanceId'] diff --git a/nova/tests/test_rpc.py b/nova/tests/test_rpc.py index c1ef60ff6..fcecfb352 100644 --- a/nova/tests/test_rpc.py +++ b/nova/tests/test_rpc.py @@ -124,7 +124,6 @@ class RpcTestCase(test.TestCase): 'test', {"method": "fail", "args": {"value": value}}) - LOG.error('INNNNNNN BETTTWWWWWWWWWWEEEEEEEEEEN') try: rpc.call(self.context, 'test', diff --git a/run_tests.py b/run_tests.py index 509a60caa..d5d8acd16 100644 --- a/run_tests.py +++ b/run_tests.py @@ -285,7 +285,6 @@ if __name__ == '__main__': # If any argument looks like a test name but doesn't have "nova.tests" in # front of it, automatically add that so we don't have to type as much argv = [] - logging.getLogger('amqplib').setLevel(logging.DEBUG) for x in sys.argv: if x.startswith('test_'): argv.append('nova.tests.%s' % x) -- cgit From 7755bbfc7b16248dab23bfab479d09501519290f Mon Sep 17 00:00:00 2001 From: termie Date: Wed, 25 May 2011 15:43:04 -0700 Subject: cleanups --- nova/tests/test_rpc.py | 47 +++++++++++++++++++++-------------------------- 1 file changed, 21 insertions(+), 26 deletions(-) diff --git a/nova/tests/test_rpc.py b/nova/tests/test_rpc.py index fcecfb352..8523b409c 100644 --- a/nova/tests/test_rpc.py +++ b/nova/tests/test_rpc.py @@ -31,7 +31,6 @@ LOG = logging.getLogger('nova.tests.rpc') class RpcTestCase(test.TestCase): - """Test cases for rpc""" def setUp(self): super(RpcTestCase, self).setUp() self.conn = rpc.Connection.instance(True) @@ -43,21 +42,18 @@ class RpcTestCase(test.TestCase): self.context = context.get_admin_context() def test_call_succeed(self): - """Get a value through rpc call""" value = 42 result = rpc.call(self.context, 'test', {"method": "echo", "args": {"value": value}}) self.assertEqual(value, result) def test_call_succeed_despite_multiple_returns(self): - """Get a value through rpc call""" value = 42 result = rpc.call(self.context, 'test', {"method": "echo_three_times", "args": {"value": value}}) self.assertEqual(value, result) def test_call_succeed_despite_multiple_returns_yield(self): - """Get a value through rpc call""" value = 42 result = rpc.call(self.context, 'test', {"method": "echo_three_times_yield", @@ -65,7 +61,6 @@ class RpcTestCase(test.TestCase): self.assertEqual(value, result) def test_multicall_succeed_once(self): - """Get a value through rpc call""" value = 42 result = rpc.multicall(self.context, 'test', @@ -79,7 +74,6 @@ class RpcTestCase(test.TestCase): i += 1 def test_multicall_succeed_three_times(self): - """Get a value through rpc call""" value = 42 result = rpc.multicall(self.context, 'test', @@ -91,7 +85,6 @@ class RpcTestCase(test.TestCase): i += 1 def test_multicall_succeed_three_times_yield(self): - """Get a value through rpc call""" value = 42 result = rpc.multicall(self.context, 'test', @@ -103,7 +96,7 @@ class RpcTestCase(test.TestCase): i += 1 def test_context_passed(self): - """Makes sure a context is passed through rpc call""" + """Makes sure a context is passed through rpc call.""" value = 42 result = rpc.call(self.context, 'test', {"method": "context", @@ -111,11 +104,12 @@ class RpcTestCase(test.TestCase): self.assertEqual(self.context.to_dict(), result) def test_call_exception(self): - """Test that exception gets passed back properly + """Test that exception gets passed back properly. rpc.call returns a RemoteError object. The value of the exception is converted to a string, so we convert it back to an int in the test. + """ value = 42 self.assertRaises(rpc.RemoteError, @@ -134,7 +128,7 @@ class RpcTestCase(test.TestCase): self.assertEqual(int(exc.value), value) def test_nested_calls(self): - """Test that we can do an rpc.call inside another call""" + """Test that we can do an rpc.call inside another call.""" class Nested(object): @staticmethod def echo(context, queue, value): @@ -162,8 +156,7 @@ class RpcTestCase(test.TestCase): self.assertEqual(value, result) def test_connectionpool_single(self): - """Test that ConnectionPool recycles a single connection""" - + """Test that ConnectionPool recycles a single connection.""" conn1 = rpc.ConnectionPool.get() rpc.ConnectionPool.put(conn1) conn2 = rpc.ConnectionPool.get() @@ -171,10 +164,13 @@ class RpcTestCase(test.TestCase): self.assertEqual(conn1, conn2) def test_connectionpool_double(self): - """Test that ConnectionPool returns 2 separate connections - when called consecutively and the pool returns connections LIFO - """ + """Test that ConnectionPool returns and reuses separate connections. + + When called consecutively we should get separate connections and upon + returning them those connections should be reused for future calls + before generating a new connection. + """ conn1 = rpc.ConnectionPool.get() conn2 = rpc.ConnectionPool.get() @@ -184,14 +180,11 @@ class RpcTestCase(test.TestCase): conn3 = rpc.ConnectionPool.get() conn4 = rpc.ConnectionPool.get() - self.assertEqual(conn2, conn3) - self.assertEqual(conn1, conn4) + self.assertEqual(conn1, conn3) + self.assertEqual(conn2, conn4) def test_connectionpool_limit(self): - """Test connection pool limit and verify all connections - are unique - """ - + """Test connection pool limit and connection uniqueness.""" max_size = FLAGS.rpc_conn_pool_size conns = [] @@ -205,19 +198,21 @@ class RpcTestCase(test.TestCase): class TestReceiver(object): - """Simple Proxy class so the consumer has methods to call + """Simple Proxy class so the consumer has methods to call. + + Uses static methods because we aren't actually storing any state. - Uses static methods because we aren't actually storing any state""" + """ @staticmethod def echo(context, value): - """Simply returns whatever value is sent in""" + """Simply returns whatever value is sent in.""" LOG.debug(_("Received %s"), value) return value @staticmethod def context(context, value): - """Returns dictionary version of context""" + """Returns dictionary version of context.""" LOG.debug(_("Received %s"), context) return context.to_dict() @@ -235,5 +230,5 @@ class TestReceiver(object): @staticmethod def fail(context, value): - """Raises an exception with the value sent in""" + """Raises an exception with the value sent in.""" raise Exception(value) -- cgit From f56df190ee888ae731740e7e949fb6f0c012d687 Mon Sep 17 00:00:00 2001 From: termie Date: Wed, 25 May 2011 15:43:04 -0700 Subject: replace removed import --- nova/tests/test_cloud.py | 1 + 1 file changed, 1 insertion(+) diff --git a/nova/tests/test_cloud.py b/nova/tests/test_cloud.py index ca3ef7ffe..b64be662e 100644 --- a/nova/tests/test_cloud.py +++ b/nova/tests/test_cloud.py @@ -19,6 +19,7 @@ from base64 import b64decode from M2Crypto import BIO from M2Crypto import RSA +import os from eventlet import greenthread -- cgit From b3506a471bbce063d72aead211f45d693bda7853 Mon Sep 17 00:00:00 2001 From: termie Date: Wed, 25 May 2011 15:43:04 -0700 Subject: don't put connection back in pool --- nova/service.py | 1 - 1 file changed, 1 deletion(-) diff --git a/nova/service.py b/nova/service.py index 94afd5f78..141fd4253 100644 --- a/nova/service.py +++ b/nova/service.py @@ -181,7 +181,6 @@ class Service(object): except greenlet.GreenletExit: pass self.stop() - rpc.ConnectionPool.put(self.conn) try: db.service_destroy(context.get_admin_context(), self.service_id) except exception.NotFound: -- cgit From a05e8e7587e42633e8459fd050eee3a4da247330 Mon Sep 17 00:00:00 2001 From: termie Date: Wed, 25 May 2011 15:43:04 -0700 Subject: move consumerset killing into stop --- nova/service.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/nova/service.py b/nova/service.py index 141fd4253..782183322 100644 --- a/nova/service.py +++ b/nova/service.py @@ -175,11 +175,6 @@ class Service(object): def kill(self): """Destroy the service object in the datastore.""" - self.csetthread.kill() - try: - self.csetthread.wait() - except greenlet.GreenletExit: - pass self.stop() try: db.service_destroy(context.get_admin_context(), self.service_id) @@ -187,6 +182,11 @@ class Service(object): logging.warn(_('Service killed that has no database entry')) def stop(self): + self.csetthread.kill() + try: + self.csetthread.wait() + except greenlet.GreenletExit: + pass for x in self.timers: try: x.stop() -- cgit From feb04f0117450bcd6e8f4966f4487575073be41c Mon Sep 17 00:00:00 2001 From: termie Date: Wed, 25 May 2011 15:43:04 -0700 Subject: change the behavior of calling a multicall --- nova/rpc.py | 8 +++++--- nova/tests/test_rpc.py | 4 ++-- 2 files changed, 7 insertions(+), 5 deletions(-) diff --git a/nova/rpc.py b/nova/rpc.py index 8d14494f0..493978e57 100644 --- a/nova/rpc.py +++ b/nova/rpc.py @@ -236,6 +236,9 @@ class AdapterConsumer(Consumer): # This final None tells multicall that it is done. msg_reply(msg_id, None, None) + elif hasattr(rval, 'send'): + # NOTE(vish): this iterates through the generator + list(rval) except Exception as e: logging.exception('Exception during message handling') if msg_id: @@ -530,9 +533,8 @@ class MulticallWaiter(object): def call(context, topic, msg): """Sends a message on a topic and wait for a response.""" rv = multicall(context, topic, msg) - for x in rv: - rv.close() - return x + # NOTE(vish): return the last result from the multicall + return list(rv)[-1] def cast(context, topic, msg): diff --git a/nova/tests/test_rpc.py b/nova/tests/test_rpc.py index 8523b409c..35f4a64d9 100644 --- a/nova/tests/test_rpc.py +++ b/nova/tests/test_rpc.py @@ -51,14 +51,14 @@ class RpcTestCase(test.TestCase): value = 42 result = rpc.call(self.context, 'test', {"method": "echo_three_times", "args": {"value": value}}) - self.assertEqual(value, result) + self.assertEqual(value + 2, result) def test_call_succeed_despite_multiple_returns_yield(self): value = 42 result = rpc.call(self.context, 'test', {"method": "echo_three_times_yield", "args": {"value": value}}) - self.assertEqual(value, result) + self.assertEqual(value + 2, result) def test_multicall_succeed_once(self): value = 42 -- cgit From 61a4dd17be5d89e8aac62d6783310cb5ddb6ee60 Mon Sep 17 00:00:00 2001 From: sateesh Date: Thu, 26 May 2011 10:36:52 +0530 Subject: Instead of redefining the flag 'vlan_interface', just setting a default value (vmnic0) in vmwareapi_net.py --- nova/network/vmwareapi_net.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/nova/network/vmwareapi_net.py b/nova/network/vmwareapi_net.py index 373060add..04210c011 100644 --- a/nova/network/vmwareapi_net.py +++ b/nova/network/vmwareapi_net.py @@ -30,9 +30,7 @@ LOG = logging.getLogger("nova.network.vmwareapi_net") FLAGS = flags.FLAGS -flags.DEFINE_string('vlan_interface', 'vmnic0', - 'Physical network adapter name in VMware ESX host for ' - 'vlan networking') +FLAGS['vlan_interface'].SetDefault('vmnic0') def ensure_vlan_bridge(vlan_num, bridge, net_attrs=None): -- cgit From f37d94428dd0b56632958d5d3a6930531a51cd44 Mon Sep 17 00:00:00 2001 From: Alex Meade Date: Thu, 26 May 2011 10:54:46 -0400 Subject: Restricted image filtering by name and status only --- nova/api/openstack/images.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/nova/api/openstack/images.py b/nova/api/openstack/images.py index 553566d58..2e779da79 100644 --- a/nova/api/openstack/images.py +++ b/nova/api/openstack/images.py @@ -28,8 +28,7 @@ from nova.api.openstack.views import images as images_view LOG = log.getLogger('nova.api.openstack.images') FLAGS = flags.FLAGS -SUPPORTED_FILTERS = ['name', 'status', 'container_format', 'disk_format', - 'size_min', 'size_max'] +SUPPORTED_FILTERS = ['name', 'status'] class Controller(common.OpenstackController): -- cgit From b9b16ca71d4bbb9782482bdf5d848bb5b787732f Mon Sep 17 00:00:00 2001 From: Alex Meade Date: Thu, 26 May 2011 13:59:25 -0400 Subject: Expanded tests --- nova/tests/api/openstack/test_images.py | 122 ++++++++++++++++++++++++++++++-- 1 file changed, 116 insertions(+), 6 deletions(-) diff --git a/nova/tests/api/openstack/test_images.py b/nova/tests/api/openstack/test_images.py index f3f0217d6..9f1f28611 100644 --- a/nova/tests/api/openstack/test_images.py +++ b/nova/tests/api/openstack/test_images.py @@ -709,23 +709,119 @@ class ImageControllerWithGlanceServiceTest(test.TestCase): self.assertDictListMatch(expected, response_list) - def test_get_image_request_filters(self): + def test_image_filter_with_name(self): mocker = mox.Mox() image_service = mocker.CreateMockAnything() context = object() - filters = {'status': 'ACTIVE', - 'name': 'testname', - 'property-test': '3'} + filters = {'name': 'testname'} + image_service.index(context, filters).AndReturn([]) + mocker.ReplayAll() + request = webob.Request.blank( + '/v1.1/images?name=testname') + request.environ['nova.context'] = context + controller = images.ControllerV11(image_service=image_service) + controller.index(request) + mocker.VerifyAll() + + def test_image_filter_with_status(self): + mocker = mox.Mox() + image_service = mocker.CreateMockAnything() + context = object() + filters = {'status': 'ACTIVE'} + image_service.index(context, filters).AndReturn([]) + mocker.ReplayAll() + request = webob.Request.blank( + '/v1.1/images?status=ACTIVE') + request.environ['nova.context'] = context + controller = images.ControllerV11(image_service=image_service) + controller.index(request) + mocker.VerifyAll() + + def test_image_filter_with_property(self): + mocker = mox.Mox() + image_service = mocker.CreateMockAnything() + context = object() + filters = {'property-test': '3'} + image_service.index(context, filters).AndReturn([]) + mocker.ReplayAll() + request = webob.Request.blank( + '/v1.1/images?property-test=3') + request.environ['nova.context'] = context + controller = images.ControllerV11(image_service=image_service) + controller.index(request) + mocker.VerifyAll() + + def test_image_filter_not_supported(self): + mocker = mox.Mox() + image_service = mocker.CreateMockAnything() + context = object() + filters = {'status': 'ACTIVE'} + image_service.index(context, filters).AndReturn([]) + mocker.ReplayAll() + request = webob.Request.blank( + '/v1.1/images?status=ACTIVE&UNSUPPORTEDFILTER=testname') + request.environ['nova.context'] = context + controller = images.ControllerV11(image_service=image_service) + controller.index(request) + mocker.VerifyAll() + + def test_image_no_filters(self): + mocker = mox.Mox() + image_service = mocker.CreateMockAnything() + context = object() + filters = {} + image_service.index(context, filters).AndReturn([]) + mocker.ReplayAll() + request = webob.Request.blank( + '/v1.1/images') + request.environ['nova.context'] = context + controller = images.ControllerV11(image_service=image_service) + controller.index(request) + mocker.VerifyAll() + + def test_image_detail_filter_with_name(self): + mocker = mox.Mox() + image_service = mocker.CreateMockAnything() + context = object() + filters = {'name': 'testname'} + image_service.detail(context, filters).AndReturn([]) + mocker.ReplayAll() + request = webob.Request.blank( + '/v1.1/images/detail?name=testname') + request.environ['nova.context'] = context + controller = images.ControllerV11(image_service=image_service) + controller.detail(request) + mocker.VerifyAll() + + def test_image_detail_filter_with_status(self): + mocker = mox.Mox() + image_service = mocker.CreateMockAnything() + context = object() + filters = {'status': 'ACTIVE'} + image_service.detail(context, filters).AndReturn([]) + mocker.ReplayAll() + request = webob.Request.blank( + '/v1.1/images/detail?status=ACTIVE') + request.environ['nova.context'] = context + controller = images.ControllerV11(image_service=image_service) + controller.detail(request) + mocker.VerifyAll() + + def test_image_detail_filter_with_property(self): + mocker = mox.Mox() + image_service = mocker.CreateMockAnything() + context = object() + filters = {'property-test': '3'} image_service.detail(context, filters).AndReturn([]) mocker.ReplayAll() request = webob.Request.blank( - '/v1.1/images/detail?status=ACTIVE&name=testname&property-test=3') + '/v1.1/images/detail?property-test=3') request.environ['nova.context'] = context controller = images.ControllerV11(image_service=image_service) controller.detail(request) mocker.VerifyAll() - def test_get_image_request_filters_not_supported(self): + def test_image_detail_filter_not_supported(self): mocker = mox.Mox() image_service = mocker.CreateMockAnything() context = object() @@ -739,6 +835,20 @@ class ImageControllerWithGlanceServiceTest(test.TestCase): controller.detail(request) mocker.VerifyAll() + def test_image_detail_no_filters(self): + mocker = mox.Mox() + image_service = mocker.CreateMockAnything() + context = object() + filters = {} + image_service.detail(context, filters).AndReturn([]) + mocker.ReplayAll() + request = webob.Request.blank( + '/v1.1/images/detail') + request.environ['nova.context'] = context + controller = images.ControllerV11(image_service=image_service) + controller.detail(request) + mocker.VerifyAll() + def test_get_image_found(self): req = webob.Request.blank('/v1.0/images/123') res = req.get_response(fakes.wsgi_app()) -- cgit From 899642030dd60541153ccee810d082816f92dd49 Mon Sep 17 00:00:00 2001 From: Chris Behrens Date: Thu, 26 May 2011 19:27:27 +0000 Subject: Change the return from glance to be a list of dictionaries describing VDIs Fix the rest of the code to account for this Add a test for swap --- nova/tests/test_xenapi.py | 23 +++++ nova/tests/xenapi/stubs.py | 23 +++-- nova/virt/xenapi/fake.py | 5 +- nova/virt/xenapi/vm_utils.py | 49 +++++++---- nova/virt/xenapi/vmops.py | 52 ++++++------ plugins/xenserver/xenapi/etc/xapi.d/plugins/glance | 97 ++++++++++++++-------- 6 files changed, 168 insertions(+), 81 deletions(-) diff --git a/nova/tests/test_xenapi.py b/nova/tests/test_xenapi.py index be1e35697..18a267896 100644 --- a/nova/tests/test_xenapi.py +++ b/nova/tests/test_xenapi.py @@ -395,6 +395,29 @@ class XenAPIVMTestCase(test.TestCase): os_type="linux") self.check_vm_params_for_linux() + def test_spawn_vhd_glance_swapdisk(self): + # Change the default host_call_plugin to one that'll return + # a swap disk + orig_func = stubs.FakeSessionForVMTests.host_call_plugin + + stubs.FakeSessionForVMTests.host_call_plugin = \ + stubs.FakeSessionForVMTests.host_call_plugin_swap + + try: + # We'll steal the above glance linux test + self.test_spawn_vhd_glance_linux() + finally: + # Make sure to put this back + stubs.FakeSessionForVMTests.host_call_plugin = orig_func + + # We should have 2 VBDs. + self.assertEqual(len(self.vm['VBDs']), 2) + # Now test that we have 1. + self.tearDown() + self.setUp() + self.test_spawn_vhd_glance_linux() + self.assertEqual(len(self.vm['VBDs']), 1) + def test_spawn_vhd_glance_windows(self): FLAGS.xenapi_image_service = 'glance' self._test_spawn(glance_stubs.FakeGlance.IMAGE_VHD, None, None, diff --git a/nova/tests/xenapi/stubs.py b/nova/tests/xenapi/stubs.py index 9f6f64318..35308d95f 100644 --- a/nova/tests/xenapi/stubs.py +++ b/nova/tests/xenapi/stubs.py @@ -38,7 +38,7 @@ def stubout_instance_snapshot(stubs): sr_ref=sr_ref, sharable=False) vdi_rec = session.get_xenapi().VDI.get_record(vdi_ref) vdi_uuid = vdi_rec['uuid'] - return {'primary_vdi_uuid': vdi_uuid} + return [dict(vdi_type='os', vdi_uuid=vdi_uuid)] stubs.Set(vm_utils.VMHelper, 'fetch_image', fake_fetch_image) @@ -134,16 +134,29 @@ class FakeSessionForVMTests(fake.SessionBase): super(FakeSessionForVMTests, self).__init__(uri) def host_call_plugin(self, _1, _2, plugin, method, _5): + sr_ref = fake.get_all('SR')[0] + vdi_ref = fake.create_vdi('', False, sr_ref, False) + vdi_rec = fake.get_record('VDI', vdi_ref) + if plugin == "glance" and method == "download_vhd": + ret_str = json.dumps([dict(vdi_type='os', + vdi_uuid=vdi_rec['uuid'])]) + else: + ret_str = vdi_rec['uuid'] + return '%s' % ret_str + + def host_call_plugin_swap(self, _1, _2, plugin, method, _5): sr_ref = fake.get_all('SR')[0] vdi_ref = fake.create_vdi('', False, sr_ref, False) vdi_rec = fake.get_record('VDI', vdi_ref) if plugin == "glance" and method == "download_vhd": swap_vdi_ref = fake.create_vdi('', False, sr_ref, False) swap_vdi_rec = fake.get_record('VDI', swap_vdi_ref) - return '%s' % json.dumps( - {'primary_vdi_uuid': vdi_rec['uuid'], - 'swap_vdi_uuid': swap_vdi_rec['uuid']}) - return '%s' % vdi_rec['uuid'] + ret_str = json.dumps( + [dict(vdi_type='os', vdi_uuid=vdi_rec['uuid']), + dict(vdi_type='swap', vdi_uuid=swap_vdi_rec['uuid'])]) + else: + ret_str = vdi_rec['uuid'] + return '%s' % ret_str def VM_start(self, _1, ref, _2, _3): vm = fake.get_record('VM', ref) diff --git a/nova/virt/xenapi/fake.py b/nova/virt/xenapi/fake.py index e36ef3288..76988b172 100644 --- a/nova/virt/xenapi/fake.py +++ b/nova/virt/xenapi/fake.py @@ -159,7 +159,10 @@ def after_VBD_create(vbd_ref, vbd_rec): vbd_rec['device'] = '' vm_ref = vbd_rec['VM'] vm_rec = _db_content['VM'][vm_ref] - vm_rec['VBDs'] = [vbd_ref] + if vm_rec.get('VBDs', None): + vm_rec['VBDs'].append(vbd_ref) + else: + vm_rec['VBDs'] = [vbd_ref] vm_name_label = _db_content['VM'][vm_ref]['name_label'] vbd_rec['vm_name_label'] = vm_name_label diff --git a/nova/virt/xenapi/vm_utils.py b/nova/virt/xenapi/vm_utils.py index 3d980013a..bee9742a4 100644 --- a/nova/virt/xenapi/vm_utils.py +++ b/nova/virt/xenapi/vm_utils.py @@ -377,6 +377,9 @@ class VMHelper(HelperBase): xenapi_image_service = ['glance', 'objectstore'] glance_address = 'address for glance services' glance_port = 'port for glance services' + + Returns: A single filename if image_type is KERNEL_RAMDISK + A list of dictionaries that describe VDIs, otherwise """ access = AuthManager().get_access_key(user, project) @@ -391,6 +394,10 @@ class VMHelper(HelperBase): @classmethod def _fetch_image_glance_vhd(cls, session, instance_id, image, access, image_type): + """Tell glance to download an image and put the VHDs into the SR + + Returns: A list of dictionaries that describe VDIs + """ LOG.debug(_("Asking xapi to fetch vhd image %(image)s") % locals()) @@ -410,25 +417,21 @@ class VMHelper(HelperBase): kwargs = {'params': pickle.dumps(params)} task = session.async_call_plugin('glance', 'download_vhd', kwargs) result = session.wait_for_task(task, instance_id) - vdi_uuids = json.loads(result) - primary_vdi_uuid = vdi_uuids.get('primary_vdi_uuid') - swap_vdi_uuid = vdi_uuids.get('swap_vdi_uuid', None) + vdis = json.loads(result) + for vdi in vdis: + LOG.debug(_("xapi 'download_vhd' returned VDI of " + "type '%(vdi_type)s' with UUID '%(vdi_uuid)s'" % vdi)) cls.scan_sr(session, instance_id, sr_ref) + # Pull out the UUID of the first VDI + vdi_uuid = vdis[0]['vdi_uuid'] # Set the name-label to ease debugging - primary_vdi_ref = session.get_xenapi().VDI.get_by_uuid(primary_vdi_uuid) + vdi_ref = session.get_xenapi().VDI.get_by_uuid(vdi_uuid) primary_name_label = get_name_label_for_image(image) - session.get_xenapi().VDI.set_name_label(primary_vdi_ref, primary_name_label) - - LOG.debug(_("xapi 'download_vhd' returned VDI UUID " - "%(primary_vdi_uuid)s") % locals()) - if swap_vdi_uuid: - LOG.debug(_("xapi 'download_vhd' returned SWAP VDI UUID " - "%(swap_vdi_uuid)s") % locals()) + session.get_xenapi().VDI.set_name_label(vdi_ref, primary_name_label) - LOG.debug("=" * 100) - return vdi_uuids + return vdis @classmethod def _fetch_image_glance_disk(cls, session, instance_id, image, access, @@ -440,6 +443,8 @@ class VMHelper(HelperBase): plugin; instead, it streams the disks through domU to the VDI directly. + Returns: A single filename if image_type is KERNEL_RAMDISK + A list of dictionaries that describe VDIs, otherwise """ # FIXME(sirp): Since the Glance plugin seems to be required for the # VHD disk, it may be worth using the plugin for both VHD and RAW and @@ -486,7 +491,7 @@ class VMHelper(HelperBase): return filename else: vdi_uuid = session.get_xenapi().VDI.get_uuid(vdi_ref) - return {'primary_vdi_uuid': vdi_uuid} + return [dict(vdi_type='os', vdi_uuid=vdi_uuid)] @classmethod def determine_disk_image_type(cls, instance): @@ -545,6 +550,11 @@ class VMHelper(HelperBase): @classmethod def _fetch_image_glance(cls, session, instance_id, image, access, image_type): + """Fetch image from glance based on image type. + + Returns: A single filename if image_type is KERNEL_RAMDISK + A list of dictionaries that describe VDIs, otherwise + """ if image_type == ImageType.DISK_VHD: return cls._fetch_image_glance_vhd( session, instance_id, image, access, image_type) @@ -555,6 +565,11 @@ class VMHelper(HelperBase): @classmethod def _fetch_image_objectstore(cls, session, instance_id, image, access, secret, image_type): + """Fetch an image from objectstore. + + Returns: A single filename if image_type is KERNEL_RAMDISK + A list of dictionaries that describe VDIs, otherwise + """ url = images.image_url(image) LOG.debug(_("Asking xapi to fetch %(url)s as %(access)s") % locals()) if image_type == ImageType.KERNEL_RAMDISK: @@ -572,10 +587,10 @@ class VMHelper(HelperBase): if image_type == ImageType.DISK_RAW: args['raw'] = 'true' task = session.async_call_plugin('objectstore', fn, args) - uuid = session.wait_for_task(task, instance_id) + uuid_or_fn = session.wait_for_task(task, instance_id) if image_type != ImageType.KERNEL_RAMDISK: - return {'primary_vdi_uuid': uuid} - return uuid + return [dict(vdi_type='os', vdi_uuid=uuid_or_fn)] + return uuid_or_fn @classmethod def determine_is_pv(cls, session, instance_id, vdi_ref, disk_image_type, diff --git a/nova/virt/xenapi/vmops.py b/nova/virt/xenapi/vmops.py index 2a8d97a9d..02e140dcc 100644 --- a/nova/virt/xenapi/vmops.py +++ b/nova/virt/xenapi/vmops.py @@ -91,7 +91,8 @@ class VMOps(object): def finish_resize(self, instance, disk_info): vdi_uuid = self.link_disks(instance, disk_info['base_copy'], disk_info['cow']) - vm_ref = self._create_vm(instance, {'primary_vdi_uuid': vdi_uuid}) + vm_ref = self._create_vm(instance, + [dict(vdi_type='os', vdi_uuid=vdi_uuid)]) self.resize_instance(instance, vdi_uuid) self._spawn(instance, vm_ref) @@ -105,25 +106,25 @@ class VMOps(object): LOG.debug(_("Starting instance %s"), instance.name) self._session.call_xenapi('VM.start', vm_ref, False, False) - def _create_disk(self, instance): + def _create_disks(self, instance): user = AuthManager().get_user(instance.user_id) project = AuthManager().get_project(instance.project_id) disk_image_type = VMHelper.determine_disk_image_type(instance) - vdi_uuids = VMHelper.fetch_image(self._session, + vdis = VMHelper.fetch_image(self._session, instance.id, instance.image_id, user, project, disk_image_type) - return vdi_uuids + return vdis def spawn(self, instance, network_info=None): - vdi_uuids = self._create_disk(instance) - vm_ref = self._create_vm(instance, vdi_uuids, network_info) + vdis = self._create_disks(instance) + vm_ref = self._create_vm(instance, vdis, network_info) self._spawn(instance, vm_ref) def spawn_rescue(self, instance): """Spawn a rescue instance.""" self.spawn(instance) - def _create_vm(self, instance, vdi_uuids, network_info=None): + def _create_vm(self, instance, vdis, network_info=None): """Create VM instance.""" instance_name = instance.name vm_ref = VMHelper.lookup(self._session, instance_name) @@ -142,15 +143,6 @@ class VMOps(object): user = AuthManager().get_user(instance.user_id) project = AuthManager().get_project(instance.project_id) - # Are we building from a pre-existing disk? - primary_vdi_ref = self._session.call_xenapi('VDI.get_by_uuid', - vdi_uuids.get('primary_vdi_uuid')) - swap_vdi_uuid = vdi_uuids.get('swap_vdi_uuid', None) - if swap_vdi_uuid: - swap_vdi_ref = self._session.call_xenapi('VDI.get_by_uuid', swap_vdi_uuid) - else: - swap_vdi_ref = None - disk_image_type = VMHelper.determine_disk_image_type(instance) kernel = None @@ -165,17 +157,29 @@ class VMOps(object): instance.ramdisk_id, user, project, ImageType.KERNEL_RAMDISK) + # Create the VM ref and attach the first disk + first_vdi_ref = self._session.call_xenapi('VDI.get_by_uuid', + vdis[0]['vdi_uuid']) use_pv_kernel = VMHelper.determine_is_pv(self._session, - instance.id, primary_vdi_ref, disk_image_type, + instance.id, first_vdi_ref, disk_image_type, instance.os_type) - vm_ref = VMHelper.create_vm(self._session, instance, kernel, - ramdisk, use_pv_kernel) - + vm_ref = VMHelper.create_vm(self._session, instance, + kernel, ramdisk, use_pv_kernel) VMHelper.create_vbd(session=self._session, vm_ref=vm_ref, - vdi_ref=primary_vdi_ref, userdevice=0, bootable=True) - if swap_vdi_ref: + vdi_ref=first_vdi_ref, userdevice=0, bootable=True) + + # Attach any other disks + # userdevice 1 is reserved for rescue + userdevice = 2 + for vdi in vdis[1:]: + # vdi['vdi_type'] is either 'os' or 'swap', but we don't + # really care what it is right here. + vdi_ref = self._session.call_xenapi('VDI.get_by_uuid', + vdi['vdi_uuid']) VMHelper.create_vbd(session=self._session, vm_ref=vm_ref, - vdi_ref=swap_vdi_ref, userdevice=2, bootable=False) + vdi_ref=vdi_ref, userdevice=userdevice, + bootable=False) + userdevice += 1 # TODO(tr3buchet) - check to make sure we have network info, otherwise # create it now. This goes away once nova-multi-nic hits. @@ -185,7 +189,7 @@ class VMOps(object): # Alter the image before VM start for, e.g. network injection if FLAGS.xenapi_inject_image: VMHelper.preconfigure_instance(self._session, instance, - primary_vdi_ref, network_info) + first_vdi_ref, network_info) self.create_vifs(vm_ref, network_info) self.inject_network_info(instance, network_info, vm_ref) diff --git a/plugins/xenserver/xenapi/etc/xapi.d/plugins/glance b/plugins/xenserver/xenapi/etc/xapi.d/plugins/glance index 0d02adfe9..039d1b8f6 100644 --- a/plugins/xenserver/xenapi/etc/xapi.d/plugins/glance +++ b/plugins/xenserver/xenapi/etc/xapi.d/plugins/glance @@ -91,8 +91,8 @@ def _download_tarball(sr_path, staging_path, image_id, glance_host, conn.close() -def _fixup_vhds(sr_path, staging_path, uuid_stack): - """Fixup the downloaded VHDs before we move them into the SR. +def _import_vhds(sr_path, staging_path, uuid_stack): + """Import the VHDs found in the staging path. We cannot extract VHDs directly into the SR since they don't yet have UUIDs, aren't properly associated with each other, and would be subject to @@ -102,16 +102,25 @@ def _fixup_vhds(sr_path, staging_path, uuid_stack): To avoid these we problems, we use a staging area to fixup the VHDs before moving them into the SR. The steps involved are: - 1. Extracting tarball into staging area + 1. Extracting tarball into staging area (done prior to this call) 2. Renaming VHDs to use UUIDs ('snap.vhd' -> 'ffff-aaaa-...vhd') - 3. Linking the two VHDs together + 3. Linking VHDs together if there's a snap.vhd 4. Pseudo-atomically moving the images into the SR. (It's not really - atomic because it takes place as two os.rename operations; however, - the chances of an SR.scan occuring between the two rename() + atomic because it takes place as multiple os.rename operations; + however, the chances of an SR.scan occuring between the rename()s invocations is so small that we can safely ignore it) + + Returns: A list of VDIs. Each list element is a dictionary containing + information about the VHD. Dictionary keys are: + 1. "vdi_type" - The type of VDI. Currently they can be "os_disk" or + "swap" + 2. "vdi_uuid" - The UUID of the VDI + + Example return: [{"vdi_type": "os_disk","vdi_uuid": "ffff-aaa..vhd"}, + {"vdi_type": "swap","vdi_uuid": "ffff-bbb..vhd"}] """ def rename_with_uuid(orig_path): """Rename VHD using UUID so that it will be recognized by SR on a @@ -162,39 +171,57 @@ def _fixup_vhds(sr_path, staging_path, uuid_stack): "VHD %(path)s is marked as hidden without child" % locals()) - orig_base_copy_path = os.path.join(staging_path, 'image.vhd') - if not os.path.exists(orig_base_copy_path): + def prepare_if_exists(staging_path, vhd_name, parent_path=None): + """ + Check for existance of a particular VHD in the staging path and + preparing it for moving into the SR. + + Returns: Tuple of (Path to move into the SR, VDI_UUID) + None, if the vhd_name doesn't exist in the staging path + + If the VHD exists, we will do the following: + 1. Rename it with a UUID. + 2. If parent_path exists, we'll link up the VHDs. + """ + orig_path = os.path.join(staging_path, vhd_name) + if not os.path.exists(orig_path): + return None + new_path, vdi_uuid = rename_with_uuid(orig_path) + if parent_path: + # NOTE(sirp): this step is necessary so that an SR scan won't + # delete the base_copy out from under us (since it would be + # orphaned) + link_vhds(new_path, parent_path) + return (new_path, vdi_uuid) + + vdi_return_list = [] + paths_to_move = [] + + image_info = prepare_if_exists(staging_path, 'image.vhd') + if not image_info: raise Exception("Invalid image: image.vhd not present") - base_copy_path, base_copy_uuid = rename_with_uuid(orig_base_copy_path) - - vdi_uuid = base_copy_uuid - orig_snap_path = os.path.join(staging_path, 'snap.vhd') - if os.path.exists(orig_snap_path): - snap_path, snap_uuid = rename_with_uuid(orig_snap_path) - vdi_uuid = snap_uuid - # NOTE(sirp): this step is necessary so that an SR scan won't - # delete the base_copy out from under us (since it would be - # orphaned) - link_vhds(snap_path, base_copy_path) - move_into_sr(snap_path) + paths_to_move.append(image_info[0]) + + snap_info = prepare_if_exists(staging_path, 'snap.vhd', image_path) + if snap_info: + paths_to_move.append(snap_info[0]) + # We return this snap as the VDI instead of image.vhd + vdi_return_list.append(dict(vdi_type="os", vdi_uuid=snap_info[1])) else: - assert_vhd_not_hidden(base_copy_path) + # If there's no snap, we return the image.vhd UUID + vdi_return_list.append(dict(vdi_type="os", vdi_uuid=image_info[1])) + + swap_info = prepare_if_exists(staging_path, 'swap.vhd') + if swap_info: + paths_to_move.append(swap_info[0]) + vdi_return_list.append(dict(vdi_type="swap", vdi_uuid=swap_info[1])) - # If we find a swap.vhd, go ahead and copy it into the SR - swap_vdi_uuid = None - orig_swap_path = os.path.join(staging_path, 'swap.vhd') - if os.path.exists(orig_swap_path): - swap_path, swap_vdi_uuid = rename_with_uuid(orig_swap_path) - move_into_sr(swap_path) + for path in paths_to_move: + move_into_sr(path) - vdi_uuids = {} - vdi_uuids['primary_vdi_uuid'] = vdi_uuid - if swap_vdi_uuid: - vdi_uuids['swap_vdi_uuid'] = swap_vdi_uuid + return vdi_return_list - move_into_sr(base_copy_path) - return vdi_uuids def _prepare_staging_area_for_upload(sr_path, staging_path, vdi_uuids): """Hard-link VHDs into staging area with appropriate filename @@ -339,7 +366,9 @@ def download_vhd(session, args): try: _download_tarball(sr_path, staging_path, image_id, glance_host, glance_port) - return json.dumps(_fixup_vhds(sr_path, staging_path, uuid_stack)) + # Right now, it's easier to return a single string via XenAPI, + # so we'll json encode the list of VHDs. + return json.dumps(_import_vhds(sr_path, staging_path, uuid_stack)) finally: _cleanup_staging_area(staging_path) -- cgit From 04785db717492c8ba7c2d184924b3773ec944f4c Mon Sep 17 00:00:00 2001 From: Chris Behrens Date: Thu, 26 May 2011 19:37:51 +0000 Subject: fix image_path in glance plugin --- plugins/xenserver/xenapi/etc/xapi.d/plugins/glance | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/plugins/xenserver/xenapi/etc/xapi.d/plugins/glance b/plugins/xenserver/xenapi/etc/xapi.d/plugins/glance index 039d1b8f6..fef31a9ff 100644 --- a/plugins/xenserver/xenapi/etc/xapi.d/plugins/glance +++ b/plugins/xenserver/xenapi/etc/xapi.d/plugins/glance @@ -203,7 +203,8 @@ def _import_vhds(sr_path, staging_path, uuid_stack): paths_to_move.append(image_info[0]) - snap_info = prepare_if_exists(staging_path, 'snap.vhd', image_path) + snap_info = prepare_if_exists(staging_path, 'snap.vhd', + image_info[0]) if snap_info: paths_to_move.append(snap_info[0]) # We return this snap as the VDI instead of image.vhd -- cgit From 9e22f51c80cc5f7f5ea60b5b8bb779779a19667c Mon Sep 17 00:00:00 2001 From: Chris Behrens Date: Thu, 26 May 2011 20:01:09 +0000 Subject: put back the hidden assert check i accidentally removed from glance plugin --- plugins/xenserver/xenapi/etc/xapi.d/plugins/glance | 2 ++ 1 file changed, 2 insertions(+) diff --git a/plugins/xenserver/xenapi/etc/xapi.d/plugins/glance b/plugins/xenserver/xenapi/etc/xapi.d/plugins/glance index fef31a9ff..0c00d168b 100644 --- a/plugins/xenserver/xenapi/etc/xapi.d/plugins/glance +++ b/plugins/xenserver/xenapi/etc/xapi.d/plugins/glance @@ -210,11 +210,13 @@ def _import_vhds(sr_path, staging_path, uuid_stack): # We return this snap as the VDI instead of image.vhd vdi_return_list.append(dict(vdi_type="os", vdi_uuid=snap_info[1])) else: + assert_vhd_not_hidden(image_info[0]) # If there's no snap, we return the image.vhd UUID vdi_return_list.append(dict(vdi_type="os", vdi_uuid=image_info[1])) swap_info = prepare_if_exists(staging_path, 'swap.vhd') if swap_info: + assert_vhd_not_hidden(swap_info[0]) paths_to_move.append(swap_info[0]) vdi_return_list.append(dict(vdi_type="swap", vdi_uuid=swap_info[1])) -- cgit From fc27a0ac4f907282a669e2c9f3e128890907f236 Mon Sep 17 00:00:00 2001 From: Chris Behrens Date: Thu, 26 May 2011 20:21:40 +0000 Subject: add a comment when calling glance:download_vhd so it's clear what is returned --- nova/virt/xenapi/vm_utils.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/nova/virt/xenapi/vm_utils.py b/nova/virt/xenapi/vm_utils.py index bee9742a4..06ee8ee9b 100644 --- a/nova/virt/xenapi/vm_utils.py +++ b/nova/virt/xenapi/vm_utils.py @@ -417,6 +417,10 @@ class VMHelper(HelperBase): kwargs = {'params': pickle.dumps(params)} task = session.async_call_plugin('glance', 'download_vhd', kwargs) result = session.wait_for_task(task, instance_id) + # 'download_vhd' will return a json encoded string containing + # a list of dictionaries describing VDIs. The dictionary will + # contain 'vdi_type' and 'vdi_uuid' keys. 'vdi_type' can be + # 'os' or 'swap' right now. vdis = json.loads(result) for vdi in vdis: LOG.debug(_("xapi 'download_vhd' returned VDI of " -- cgit From d7e0b45a9bc415e87beee32f10c8d6bdff9819ed Mon Sep 17 00:00:00 2001 From: termie Date: Thu, 26 May 2011 15:08:53 -0700 Subject: changes per review --- nova/rpc.py | 17 ++++++++++------- nova/service.py | 17 ++++++++--------- nova/tests/test_rpc.py | 12 +++--------- nova/tests/test_service.py | 6 ++++-- 4 files changed, 25 insertions(+), 27 deletions(-) diff --git a/nova/rpc.py b/nova/rpc.py index 493978e57..1ec495bc8 100644 --- a/nova/rpc.py +++ b/nova/rpc.py @@ -28,6 +28,7 @@ import json import sys import time import traceback +import types import uuid from carrot import connection as carrot_connection @@ -228,7 +229,7 @@ class AdapterConsumer(Consumer): rval = node_func(context=ctxt, **node_args) if msg_id: # Check if the result was a generator - if hasattr(rval, 'send'): + if isinstance(rval, types.GeneratorType): for x in rval: msg_reply(msg_id, x, None) else: @@ -236,7 +237,7 @@ class AdapterConsumer(Consumer): # This final None tells multicall that it is done. msg_reply(msg_id, None, None) - elif hasattr(rval, 'send'): + elif isinstance(rval, types.GeneratorType): # NOTE(vish): this iterates through the generator list(rval) except Exception as e: @@ -281,11 +282,11 @@ class FanoutAdapterConsumer(AdapterConsumer): class ConsumerSet(object): """Groups consumers to listen on together on a single connection.""" - def __init__(self, conn, consumer_list): + def __init__(self, connection, consumer_list): self.consumer_list = set(consumer_list) self.consumer_set = None self.enabled = True - self.init(conn) + self.init(connection) def init(self, conn): if not conn: @@ -316,8 +317,7 @@ class ConsumerSet(object): running = False break except Exception as e: - LOG.error(_("Received exception %s " % type(e) + \ - "while processing consumer")) + LOG.exception(_("Exception while processing consumer")) self.reconnect() # Break to outer loop break @@ -534,7 +534,10 @@ def call(context, topic, msg): """Sends a message on a topic and wait for a response.""" rv = multicall(context, topic, msg) # NOTE(vish): return the last result from the multicall - return list(rv)[-1] + rv = list(rv) + if not rv: + return + return rv[-1] def cast(context, topic, msg): diff --git a/nova/service.py b/nova/service.py index 782183322..74f9f04d8 100644 --- a/nova/service.py +++ b/nova/service.py @@ -105,19 +105,18 @@ class Service(object): connection=self.conn, topic=self.topic, proxy=self) - - cset = rpc.ConsumerSet(self.conn, [consumer_all, - consumer_node, - fanout]) + consumer_set = rpc.ConsumerSet( + connection=self.conn, + consumer_list=[consumer_all, consumer_node, fanout]) # Wait forever, processing these consumers def _wait(): try: - cset.wait() + consumer_set.wait() finally: - cset.close() + consumer_set.close() - self.csetthread = greenthread.spawn(_wait) + self.consumer_set_thread = greenthread.spawn(_wait) if self.report_interval: pulse = utils.LoopingCall(self.report_state) @@ -182,9 +181,9 @@ class Service(object): logging.warn(_('Service killed that has no database entry')) def stop(self): - self.csetthread.kill() + self.consumer_set_thread.kill() try: - self.csetthread.wait() + self.consumer_set_thread.wait() except greenlet.GreenletExit: pass for x in self.timers: diff --git a/nova/tests/test_rpc.py b/nova/tests/test_rpc.py index 35f4a64d9..ffd748efe 100644 --- a/nova/tests/test_rpc.py +++ b/nova/tests/test_rpc.py @@ -66,12 +66,10 @@ class RpcTestCase(test.TestCase): 'test', {"method": "echo", "args": {"value": value}}) - i = 0 - for x in result: + for i, x in enumerate(result): if i > 0: self.fail('should only receive one response') self.assertEqual(value + i, x) - i += 1 def test_multicall_succeed_three_times(self): value = 42 @@ -79,10 +77,8 @@ class RpcTestCase(test.TestCase): 'test', {"method": "echo_three_times", "args": {"value": value}}) - i = 0 - for x in result: + for i, x in enumerate(result): self.assertEqual(value + i, x) - i += 1 def test_multicall_succeed_three_times_yield(self): value = 42 @@ -90,10 +86,8 @@ class RpcTestCase(test.TestCase): 'test', {"method": "echo_three_times_yield", "args": {"value": value}}) - i = 0 - for x in result: + for i, x in enumerate(result): self.assertEqual(value + i, x) - i += 1 def test_context_passed(self): """Makes sure a context is passed through rpc call.""" diff --git a/nova/tests/test_service.py b/nova/tests/test_service.py index 0bba01d92..d1cc8bd61 100644 --- a/nova/tests/test_service.py +++ b/nova/tests/test_service.py @@ -142,7 +142,8 @@ class ServiceTestCase(test.TestCase): mock_cset = self.mox.CreateMock(rpc.ConsumerSet, {'wait': wait_func}) - rpc.ConsumerSet(mox.IgnoreArg(), mox.IsA(list)).AndReturn(mock_cset) + rpc.ConsumerSet(connection=mox.IgnoreArg(), + consumer_list=mox.IsA(list)).AndReturn(mock_cset) wait_func(mox.IgnoreArg()) service_create = {'host': host, @@ -331,7 +332,8 @@ class ServiceTestCase(test.TestCase): mock_cset = self.mox.CreateMock(rpc.ConsumerSet, {'wait': wait_func}) - rpc.ConsumerSet(mox.IgnoreArg(), mox.IsA(list)).AndReturn(mock_cset) + rpc.ConsumerSet(connection=mox.IgnoreArg(), + consumer_list=mox.IsA(list)).AndReturn(mock_cset) wait_func(mox.IgnoreArg()) self.mox.StubOutWithMock(serv.manager.driver, -- cgit From 103bcae9f172dfee64e7b9235807bcfe1a8aefb3 Mon Sep 17 00:00:00 2001 From: termie Date: Thu, 26 May 2011 17:06:52 -0700 Subject: fix a minor bug unrelated to this change --- nova/rpc.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/nova/rpc.py b/nova/rpc.py index 1ec495bc8..c5277c6a9 100644 --- a/nova/rpc.py +++ b/nova/rpc.py @@ -212,7 +212,9 @@ class AdapterConsumer(Consumer): # we just log the message and send an error string # back to the caller LOG.warn(_('no method for message: %s') % message_data) - msg_reply(msg_id, _('No method for message: %s') % message_data) + if msg_id: + msg_reply(msg_id, + _('No method for message: %s') % message_data) return self.pool.spawn_n(self._process_data, msg_id, ctxt, method, args) -- cgit From 6b0ed0cb61838d01b15df26fc32df0de90f1cfbe Mon Sep 17 00:00:00 2001 From: MORITA Kazutaka Date: Fri, 27 May 2011 13:20:45 +0900 Subject: Fix a description of 'snapshot_name_template'. --- nova/db/api.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/nova/db/api.py b/nova/db/api.py index 3597732b9..e85ce9f16 100644 --- a/nova/db/api.py +++ b/nova/db/api.py @@ -48,7 +48,7 @@ flags.DEFINE_string('instance_name_template', 'instance-%08x', flags.DEFINE_string('volume_name_template', 'volume-%08x', 'Template string to be used to generate instance names') flags.DEFINE_string('snapshot_name_template', 'snapshot-%08x', - 'Template string to be used to generate instance names') + 'Template string to be used to generate snapshot names') IMPL = utils.LazyPluggable(FLAGS['db_backend'], -- cgit From 8b4c91b9f2c28e4809659f199affddbd66482dbb Mon Sep 17 00:00:00 2001 From: MORITA Kazutaka Date: Fri, 27 May 2011 13:36:59 +0900 Subject: Fix pep8 violations. --- nova/api/ec2/cloud.py | 13 +++++++++---- .../versions/019_add_volume_snapshot_support.py | 3 +-- nova/db/sqlalchemy/models.py | 1 + nova/tests/test_volume.py | 5 +++-- nova/volume/driver.py | 6 +++--- nova/volume/manager.py | 3 ++- 6 files changed, 19 insertions(+), 12 deletions(-) diff --git a/nova/api/ec2/cloud.py b/nova/api/ec2/cloud.py index 6927d6774..403b7ab40 100644 --- a/nova/api/ec2/cloud.py +++ b/nova/api/ec2/cloud.py @@ -285,7 +285,9 @@ class CloudController(object): snapshots = [] for ec2_id in snapshot_id: internal_id = ec2utils.ec2_id_to_id(ec2_id) - snapshot = self.volume_api.get_snapshot(context, snapshot_id=internal_id) + snapshot = self.volume_api.get_snapshot( + context, + snapshot_id=internal_id) snapshots.append(snapshot) else: snapshots = self.volume_api.get_all_snapshots(context) @@ -295,7 +297,8 @@ class CloudController(object): def _format_snapshot(self, context, snapshot): s = {} s['snapshotId'] = ec2utils.id_to_ec2_id(snapshot['id'], 'snap-%08x') - s['volumeId'] = ec2utils.id_to_ec2_id(snapshot['volume_id'], 'vol-%08x') + s['volumeId'] = ec2utils.id_to_ec2_id(snapshot['volume_id'], + 'vol-%08x') s['status'] = snapshot['status'] s['startTime'] = snapshot['created_at'] s['progress'] = snapshot['progress'] @@ -308,7 +311,8 @@ class CloudController(object): return s def create_snapshot(self, context, volume_id, **kwargs): - LOG.audit(_("Create snapshot of volume %s"), volume_id, context=context) + LOG.audit(_("Create snapshot of volume %s"), volume_id, + context=context) volume_id = ec2utils.ec2_id_to_id(volume_id) snapshot = self.volume_api.create_snapshot( context, @@ -629,7 +633,8 @@ class CloudController(object): else: v['attachmentSet'] = [{}] if volume.get('snapshot_id') != None: - v['snapshotId'] = ec2utils.id_to_ec2_id(volume['snapshot_id'], 'snap-%08x') + v['snapshotId'] = ec2utils.id_to_ec2_id(volume['snapshot_id'], + 'snap-%08x') else: v['snapshotId'] = None diff --git a/nova/db/sqlalchemy/migrate_repo/versions/019_add_volume_snapshot_support.py b/nova/db/sqlalchemy/migrate_repo/versions/019_add_volume_snapshot_support.py index 5a44bac16..f16d6db56 100644 --- a/nova/db/sqlalchemy/migrate_repo/versions/019_add_volume_snapshot_support.py +++ b/nova/db/sqlalchemy/migrate_repo/versions/019_add_volume_snapshot_support.py @@ -48,8 +48,7 @@ snapshots = Table('snapshots', meta, unicode_error=None, _warn_on_bytestring=False)), Column('display_description', String(length=255, convert_unicode=False, assert_unicode=None, - unicode_error=None, _warn_on_bytestring=False)) - ) + unicode_error=None, _warn_on_bytestring=False))) def upgrade(migrate_engine): diff --git a/nova/db/sqlalchemy/models.py b/nova/db/sqlalchemy/models.py index b887c5bad..480f62399 100644 --- a/nova/db/sqlalchemy/models.py +++ b/nova/db/sqlalchemy/models.py @@ -353,6 +353,7 @@ class Snapshot(BASE, NovaBase): display_name = Column(String(255)) display_description = Column(String(255)) + class ExportDevice(BASE, NovaBase): """Represates a shelf and blade that a volume can be exported on.""" __tablename__ = 'export_devices' diff --git a/nova/tests/test_volume.py b/nova/tests/test_volume.py index c66b66959..3472b1f59 100644 --- a/nova/tests/test_volume.py +++ b/nova/tests/test_volume.py @@ -193,8 +193,9 @@ class VolumeTestCase(test.TestCase): self.volume.create_volume(self.context, volume_id) snapshot_id = self._create_snapshot(volume_id) self.volume.create_snapshot(self.context, volume_id, snapshot_id) - self.assertEqual(snapshot_id, db.snapshot_get(context.get_admin_context(), - snapshot_id).id) + self.assertEqual(snapshot_id, + db.snapshot_get(context.get_admin_context(), + snapshot_id).id) self.volume.delete_snapshot(self.context, snapshot_id) self.assertRaises(exception.NotFound, diff --git a/nova/volume/driver.py b/nova/volume/driver.py index e0e18b9bf..21cc228c9 100644 --- a/nova/volume/driver.py +++ b/nova/volume/driver.py @@ -97,7 +97,7 @@ class VolumeDriver(object): def _copy_volume(self, srcstr, deststr, size_in_g): self._execute('sudo', 'dd', 'if=%s' % srcstr, 'of=%s' % deststr, 'count=%d' % (size_in_g * 1024), 'bs=1M') - + def _volume_not_present(self, volume_name): path_name = '%s/%s' % (FLAGS.volume_group, volume_name) try: @@ -115,7 +115,7 @@ class VolumeDriver(object): self._try_execute('sudo', 'lvremove', '-f', "%s/%s" % (FLAGS.volume_group, self._escape_snapshot(volume['name']))) - + def _sizestr(self, size_in_g): if int(size_in_g) == 0: return '100M' @@ -150,7 +150,7 @@ class VolumeDriver(object): out = out.strip() if (out[0] == 'o') or (out[0] == 'O'): raise exception.VolumeIsBusy(volume_name=volume['name']) - + self._delete_volume(volume, volume['size']) def create_snapshot(self, snapshot): diff --git a/nova/volume/manager.py b/nova/volume/manager.py index fd889633d..40a104d35 100644 --- a/nova/volume/manager.py +++ b/nova/volume/manager.py @@ -169,7 +169,8 @@ class VolumeManager(manager.SchedulerDependentManager): LOG.debug(_("snapshot %(snap_name)s: creating") % locals()) model_update = self.driver.create_snapshot(snapshot_ref) if model_update: - self.db.snapshot_update(context, snapshot_ref['id'], model_update) + self.db.snapshot_update(context, snapshot_ref['id'], + model_update) except Exception: self.db.snapshot_update(context, -- cgit From c229d6e32f5275b2eb10e760f89a52dc31635c47 Mon Sep 17 00:00:00 2001 From: MORITA Kazutaka Date: Fri, 27 May 2011 14:13:17 +0900 Subject: Fix pep8 errors. --- nova/api/ec2/cloud.py | 7 ++++--- nova/tests/test_volume.py | 10 ++++++---- nova/volume/api.py | 3 ++- nova/volume/driver.py | 4 ++-- nova/volume/manager.py | 5 +++-- 5 files changed, 17 insertions(+), 12 deletions(-) diff --git a/nova/api/ec2/cloud.py b/nova/api/ec2/cloud.py index b717a10c0..79cc3b3bf 100644 --- a/nova/api/ec2/cloud.py +++ b/nova/api/ec2/cloud.py @@ -666,14 +666,15 @@ class CloudController(object): return v def create_volume(self, context, **kwargs): - size = kwargs.get('size'); + size = kwargs.get('size') if kwargs.get('snapshot_id') != None: snapshot_id = ec2utils.ec2_id_to_id(kwargs['snapshot_id']) - LOG.audit(_("Create volume from snapshot %s"), snapshot_id, context=context) + LOG.audit(_("Create volume from snapshot %s"), snapshot_id, + context=context) else: snapshot_id = None LOG.audit(_("Create volume of %s GB"), size, context=context) - + volume = self.volume_api.create( context, size=size, diff --git a/nova/tests/test_volume.py b/nova/tests/test_volume.py index 8d58b3135..4f10ee6af 100644 --- a/nova/tests/test_volume.py +++ b/nova/tests/test_volume.py @@ -78,10 +78,12 @@ class VolumeTestCase(test.TestCase): self.volume.create_snapshot(self.context, volume_src_id, snapshot_id) volume_dst_id = self._create_volume(0, snapshot_id) self.volume.create_volume(self.context, volume_dst_id, snapshot_id) - self.assertEqual(volume_dst_id, db.volume_get(context.get_admin_context(), - volume_dst_id).id) - self.assertEqual(snapshot_id, db.volume_get(context.get_admin_context(), - volume_dst_id).snapshot_id) + self.assertEqual(volume_dst_id, db.volume_get( + context.get_admin_context(), + volume_dst_id).id) + self.assertEqual(snapshot_id, db.volume_get( + context.get_admin_context(), + volume_dst_id).snapshot_id) self.volume.delete_volume(self.context, volume_dst_id) self.volume.delete_snapshot(self.context, snapshot_id) diff --git a/nova/volume/api.py b/nova/volume/api.py index 7fa80383b..5804955f7 100644 --- a/nova/volume/api.py +++ b/nova/volume/api.py @@ -43,7 +43,8 @@ class API(base.Base): if snapshot_id != None: snapshot = self.get_snapshot(context, snapshot_id) if snapshot['status'] != "available": - raise exception.ApiError(_("Snapshot status must be available")) + raise exception.ApiError( + _("Snapshot status must be available")) size = snapshot['volume_size'] if quota.allowed_volumes(context, 1, size) < 1: diff --git a/nova/volume/driver.py b/nova/volume/driver.py index df9767a79..87e13277f 100644 --- a/nova/volume/driver.py +++ b/nova/volume/driver.py @@ -674,10 +674,10 @@ class SheepdogDriver(VolumeDriver): def create_volume_from_snapshot(self, volume, snapshot): """Creates a sheepdog volume from a snapshot.""" self._try_execute('qemu-img', 'create', '-b', - "sheepdog:%s:%s" % (snapshot['volume_name'], snapshot['name']), + "sheepdog:%s:%s" % (snapshot['volume_name'], + snapshot['name']), "sheepdog:%s" % volume['name']) - def delete_volume(self, volume): """Deletes a logical volume""" self._try_execute('collie', 'vdi', 'delete', volume['name']) diff --git a/nova/volume/manager.py b/nova/volume/manager.py index 673771aa7..ff53f0701 100644 --- a/nova/volume/manager.py +++ b/nova/volume/manager.py @@ -112,8 +112,9 @@ class VolumeManager(manager.SchedulerDependentManager): model_update = self.driver.create_volume(volume_ref) else: snapshot_ref = self.db.snapshot_get(context, snapshot_id) - model_update = self.driver.create_volume_from_snapshot(volume_ref, - snapshot_ref) + model_update = self.driver.create_volume_from_snapshot( + volume_ref, + snapshot_ref) if model_update: self.db.volume_update(context, volume_ref['id'], model_update) -- cgit From a92f2bcbbaa40458e81bad3f6cb21288161322f9 Mon Sep 17 00:00:00 2001 From: Chris Behrens Date: Fri, 27 May 2011 06:56:50 +0000 Subject: fix calls to openssl properly now. Only append \n to stdin when decoding. Updated the test slightly, also. --- nova/tests/test_xenapi.py | 1 + nova/virt/xenapi/vmops.py | 6 +++++- 2 files changed, 6 insertions(+), 1 deletion(-) diff --git a/nova/tests/test_xenapi.py b/nova/tests/test_xenapi.py index 18a267896..3ba37a762 100644 --- a/nova/tests/test_xenapi.py +++ b/nova/tests/test_xenapi.py @@ -595,6 +595,7 @@ class XenAPIDiffieHellmanTestCase(test.TestCase): def test_encryption(self): msg = "This is a top-secret message" enc = self.alice.encrypt(msg) + self.assertFalse(enc.endswith('\n')) dec = self.bob.decrypt(enc) self.assertEquals(dec, msg) diff --git a/nova/virt/xenapi/vmops.py b/nova/virt/xenapi/vmops.py index 6d516ddbc..1d8678ce2 100644 --- a/nova/virt/xenapi/vmops.py +++ b/nova/virt/xenapi/vmops.py @@ -1195,12 +1195,16 @@ class SimpleDH(object): '-nosalt %(dec_flag)s') if which.lower()[0] == 'd': dec_flag = ' -d' + # When decoding base64, we need to make sure there's a + # single '\n' at the end of the base64 encoded data. + # It's kinda dumb that openssl wants to see a newline + text = text.strip('\n') + '\n' else: dec_flag = '' shared = self._shared cmd = base_cmd % locals() proc = _runproc(cmd) - proc.stdin.write(text + '\n') + proc.stdin.write(text) proc.stdin.close() proc.wait() err = proc.stderr.read() -- cgit From 107b15d2dd7d554d9cca177343ab45c51029d484 Mon Sep 17 00:00:00 2001 From: Mark Washenberger Date: Fri, 27 May 2011 10:15:33 -0400 Subject: fix encryption handling of newlines again and restructure the code a bit --- nova/tests/test_xenapi.py | 22 ++++++++++++++++++---- nova/virt/xenapi/vmops.py | 24 +++++++----------------- 2 files changed, 25 insertions(+), 21 deletions(-) diff --git a/nova/tests/test_xenapi.py b/nova/tests/test_xenapi.py index 3ba37a762..0632d05a5 100644 --- a/nova/tests/test_xenapi.py +++ b/nova/tests/test_xenapi.py @@ -592,12 +592,26 @@ class XenAPIDiffieHellmanTestCase(test.TestCase): bob_shared = self.bob.compute_shared(alice_pub) self.assertEquals(alice_shared, bob_shared) - def test_encryption(self): - msg = "This is a top-secret message" - enc = self.alice.encrypt(msg) + def _test_encryption(self, message): + enc = self.alice.encrypt(message) self.assertFalse(enc.endswith('\n')) dec = self.bob.decrypt(enc) - self.assertEquals(dec, msg) + self.assertEquals(dec, message) + + def test_encrypt_simple_message(self): + self._test_encryption('This is a simple message.') + + def test_encrypt_message_with_newlines_at_end(self): + self._test_encryption('This message has a newline at the end.\n') + + def test_encrypt_many_newlines_at_end(self): + self._test_encryption('Message with lotsa newlines.\n\n\n') + + def test_encrypt_newlines_inside_message(self): + self._test_encryption('Message\nwith\ninterior\nnewlines.') + + def test_encrypt_with_leading_newlines(self): + self._test_encryption('\n\nMessage with leading newlines.') def tearDown(self): super(XenAPIDiffieHellmanTestCase, self).tearDown() diff --git a/nova/virt/xenapi/vmops.py b/nova/virt/xenapi/vmops.py index 1d8678ce2..ce84c8652 100644 --- a/nova/virt/xenapi/vmops.py +++ b/nova/virt/xenapi/vmops.py @@ -1190,30 +1190,20 @@ class SimpleDH(object): mpi = M2Crypto.m2.bn_to_mpi(bn) return mpi - def _run_ssl(self, text, which): - base_cmd = ('openssl enc -aes-128-cbc -a -pass pass:%(shared)s ' - '-nosalt %(dec_flag)s') - if which.lower()[0] == 'd': - dec_flag = ' -d' - # When decoding base64, we need to make sure there's a - # single '\n' at the end of the base64 encoded data. - # It's kinda dumb that openssl wants to see a newline - text = text.strip('\n') + '\n' - else: - dec_flag = '' - shared = self._shared - cmd = base_cmd % locals() - proc = _runproc(cmd) + def _run_ssl(self, subcommand, text): + proc = _runproc('openssl %s' % subcommand) proc.stdin.write(text) proc.stdin.close() proc.wait() err = proc.stderr.read() if err: raise RuntimeError(_('OpenSSL error: %s') % err) - return proc.stdout.read().strip('\n') + return proc.stdout.read() def encrypt(self, text): - return self._run_ssl(text, 'enc') + cmd = 'enc -aes-128-cbc -a -pass pass:%s -nosalt' % self._shared + return self._run_ssl(cmd, text).strip('\n') def decrypt(self, text): - return self._run_ssl(text, 'dec') + cmd = 'enc -aes-128-cbc -a -A -pass pass:%s -nosalt -d' % self._shared + return self._run_ssl(cmd, text) -- cgit From 28b19b9e20100236f98e04cc43bcf106768ff2bb Mon Sep 17 00:00:00 2001 From: "Dave Walker (Daviey)" Date: Fri, 27 May 2011 15:28:10 +0100 Subject: nova/auth/novarc.template: Changed NOVA_KEY_DIR to allow symlink support --- nova/auth/novarc.template | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/nova/auth/novarc.template b/nova/auth/novarc.template index cda2ecc28..8170fcafe 100644 --- a/nova/auth/novarc.template +++ b/nova/auth/novarc.template @@ -1,4 +1,4 @@ -NOVA_KEY_DIR=$(pushd $(dirname $BASH_SOURCE)>/dev/null; pwd; popd>/dev/null) +NOVA_KEY_DIR=$(dirname $(readlink -f ${BASH_SOURCE})) export EC2_ACCESS_KEY="%(access)s:%(project)s" export EC2_SECRET_KEY="%(secret)s" export EC2_URL="%(ec2)s" -- cgit From f6d847cd867c09319f9fc451c09dc7322542e26b Mon Sep 17 00:00:00 2001 From: Mark Washenberger Date: Fri, 27 May 2011 10:40:50 -0400 Subject: prevent encryption from adding newlines on long messages --- nova/tests/test_xenapi.py | 4 ++++ nova/virt/xenapi/vmops.py | 2 +- 2 files changed, 5 insertions(+), 1 deletion(-) diff --git a/nova/tests/test_xenapi.py b/nova/tests/test_xenapi.py index 0632d05a5..fe37f0ebe 100644 --- a/nova/tests/test_xenapi.py +++ b/nova/tests/test_xenapi.py @@ -594,6 +594,7 @@ class XenAPIDiffieHellmanTestCase(test.TestCase): def _test_encryption(self, message): enc = self.alice.encrypt(message) + print enc self.assertFalse(enc.endswith('\n')) dec = self.bob.decrypt(enc) self.assertEquals(dec, message) @@ -613,6 +614,9 @@ class XenAPIDiffieHellmanTestCase(test.TestCase): def test_encrypt_with_leading_newlines(self): self._test_encryption('\n\nMessage with leading newlines.') + def test_encrypt_really_long_message(self): + self._test_encryption(''.join(['abcd' for i in xrange(1024)])) + def tearDown(self): super(XenAPIDiffieHellmanTestCase, self).tearDown() diff --git a/nova/virt/xenapi/vmops.py b/nova/virt/xenapi/vmops.py index ce84c8652..1fcaaeede 100644 --- a/nova/virt/xenapi/vmops.py +++ b/nova/virt/xenapi/vmops.py @@ -1201,7 +1201,7 @@ class SimpleDH(object): return proc.stdout.read() def encrypt(self, text): - cmd = 'enc -aes-128-cbc -a -pass pass:%s -nosalt' % self._shared + cmd = 'enc -aes-128-cbc -a -A -pass pass:%s -nosalt' % self._shared return self._run_ssl(cmd, text).strip('\n') def decrypt(self, text): -- cgit From 60a291747eeded09ade608088eae47fdb300a56b Mon Sep 17 00:00:00 2001 From: Mark Washenberger Date: Fri, 27 May 2011 10:41:12 -0400 Subject: remove errant print statement --- nova/tests/test_xenapi.py | 1 - 1 file changed, 1 deletion(-) diff --git a/nova/tests/test_xenapi.py b/nova/tests/test_xenapi.py index fe37f0ebe..9d56c1644 100644 --- a/nova/tests/test_xenapi.py +++ b/nova/tests/test_xenapi.py @@ -594,7 +594,6 @@ class XenAPIDiffieHellmanTestCase(test.TestCase): def _test_encryption(self, message): enc = self.alice.encrypt(message) - print enc self.assertFalse(enc.endswith('\n')) dec = self.bob.decrypt(enc) self.assertEquals(dec, message) -- cgit From 1af3ac5f60bb9a4ad201f0bd84a355235be2f354 Mon Sep 17 00:00:00 2001 From: Chris Behrens Date: Fri, 27 May 2011 19:50:57 +0000 Subject: fixed so all the new encryption tests pass.. including data with newlines and so forth --- nova/virt/xenapi/vmops.py | 16 ++++++++++------ 1 file changed, 10 insertions(+), 6 deletions(-) diff --git a/nova/virt/xenapi/vmops.py b/nova/virt/xenapi/vmops.py index 1fcaaeede..e116ef2d1 100644 --- a/nova/virt/xenapi/vmops.py +++ b/nova/virt/xenapi/vmops.py @@ -1190,8 +1190,12 @@ class SimpleDH(object): mpi = M2Crypto.m2.bn_to_mpi(bn) return mpi - def _run_ssl(self, subcommand, text): - proc = _runproc('openssl %s' % subcommand) + def _run_ssl(self, text, extra_args=None): + if not extra_args: + extra_args = '' + cmd = 'enc -aes-128-cbc -a -pass pass:%s -nosalt %s' % ( + self._shared, extra_args) + proc = _runproc('openssl %s' % cmd) proc.stdin.write(text) proc.stdin.close() proc.wait() @@ -1201,9 +1205,9 @@ class SimpleDH(object): return proc.stdout.read() def encrypt(self, text): - cmd = 'enc -aes-128-cbc -a -A -pass pass:%s -nosalt' % self._shared - return self._run_ssl(cmd, text).strip('\n') + return self._run_ssl(text).strip('\n') def decrypt(self, text): - cmd = 'enc -aes-128-cbc -a -A -pass pass:%s -nosalt -d' % self._shared - return self._run_ssl(cmd, text) + if text[len(text)-1:] != '\n': + text = text + '\n' + return self._run_ssl(text, '-d') -- cgit From cb42d3ec2c358a1666fde06d4252d1d76baeffff Mon Sep 17 00:00:00 2001 From: Chris Behrens Date: Fri, 27 May 2011 20:29:48 +0000 Subject: added -A back in to pass to openssl --- nova/virt/xenapi/vmops.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/nova/virt/xenapi/vmops.py b/nova/virt/xenapi/vmops.py index e116ef2d1..389c27598 100644 --- a/nova/virt/xenapi/vmops.py +++ b/nova/virt/xenapi/vmops.py @@ -1193,7 +1193,7 @@ class SimpleDH(object): def _run_ssl(self, text, extra_args=None): if not extra_args: extra_args = '' - cmd = 'enc -aes-128-cbc -a -pass pass:%s -nosalt %s' % ( + cmd = 'enc -aes-128-cbc -A -a -pass pass:%s -nosalt %s' % ( self._shared, extra_args) proc = _runproc('openssl %s' % cmd) proc.stdin.write(text) -- cgit From 2e4fca0b2a8dc4295d14a337ffa2771fab857420 Mon Sep 17 00:00:00 2001 From: Brian Waldon Date: Fri, 27 May 2011 16:31:18 -0400 Subject: now pip-requires mox version 0.5.3 --- tools/pip-requires | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/tools/pip-requires b/tools/pip-requires index 8f8018765..f1c5b2003 100644 --- a/tools/pip-requires +++ b/tools/pip-requires @@ -17,8 +17,7 @@ redis==2.0.0 routes==1.12.3 WebOb==0.9.8 wsgiref==0.1.2 -mox==0.5.0 --f http://pymox.googlecode.com/files/mox-0.5.0.tar.gz +mox==0.5.3 greenlet==0.3.1 nose bzr -- cgit From 132d0579a11b5f3b0be930e5a9369205cb282e35 Mon Sep 17 00:00:00 2001 From: Chris Behrens Date: Fri, 27 May 2011 20:48:57 +0000 Subject: added \n is not needed with -A --- nova/virt/xenapi/vmops.py | 2 -- 1 file changed, 2 deletions(-) diff --git a/nova/virt/xenapi/vmops.py b/nova/virt/xenapi/vmops.py index 389c27598..2b3fb6a39 100644 --- a/nova/virt/xenapi/vmops.py +++ b/nova/virt/xenapi/vmops.py @@ -1208,6 +1208,4 @@ class SimpleDH(object): return self._run_ssl(text).strip('\n') def decrypt(self, text): - if text[len(text)-1:] != '\n': - text = text + '\n' return self._run_ssl(text, '-d') -- cgit From a9278909cbb6d5ea9283231dbd6efc67b812abff Mon Sep 17 00:00:00 2001 From: Dan Prince Date: Sat, 28 May 2011 23:10:42 -0400 Subject: Update the rebuild_instance function in the compute manager so that it accepts the arguments that our current compute API sends. --- nova/compute/manager.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/nova/compute/manager.py b/nova/compute/manager.py index d1e01f275..3897b3a9e 100644 --- a/nova/compute/manager.py +++ b/nova/compute/manager.py @@ -331,7 +331,7 @@ class ComputeManager(manager.SchedulerDependentManager): @exception.wrap_exception @checks_instance_lock - def rebuild_instance(self, context, instance_id, image_id): + def rebuild_instance(self, context, instance_id, **kwargs): """Destroy and re-make this instance. A 'rebuild' effectively purges all existing data from the system and @@ -349,7 +349,8 @@ class ComputeManager(manager.SchedulerDependentManager): self._update_state(context, instance_id, power_state.BUILDING) self.driver.destroy(instance_ref) - instance_ref.image_id = image_id + instance_ref.image_id = kwargs.get('image_id') + instance_ref.injected_files = kwargs.get('injected_files', []) self.driver.spawn(instance_ref) self._update_image_id(context, instance_id, image_id) -- cgit From ccf522daaca0d4136c072c1905dd9fbaa1dfb2e9 Mon Sep 17 00:00:00 2001 From: Dan Prince Date: Sat, 28 May 2011 23:12:07 -0400 Subject: Fixes to the SQLAlchmeny API such that metadata is saved on an instance_update. Added integration test to test that instance metadata is updated on a rebuild. --- nova/db/sqlalchemy/api.py | 22 +++++++++++++--------- nova/tests/integrated/api/client.py | 10 ++++++++-- nova/tests/integrated/test_servers.py | 34 ++++++++++++++++++++++++++++++++++ 3 files changed, 55 insertions(+), 11 deletions(-) diff --git a/nova/db/sqlalchemy/api.py b/nova/db/sqlalchemy/api.py index e4dda5c12..1a7cae6e9 100644 --- a/nova/db/sqlalchemy/api.py +++ b/nova/db/sqlalchemy/api.py @@ -771,6 +771,15 @@ def fixed_ip_update(context, address, values): ################### +def _metadata_refs(metadata_dict): + metadata_refs = [] + if metadata_dict: + for k, v in metadata_dict.iteritems(): + metadata_ref = models.InstanceMetadata() + metadata_ref['key'] = k + metadata_ref['value'] = v + metadata_refs.append(metadata_ref) + return metadata_refs @require_context @@ -780,15 +789,7 @@ def instance_create(context, values): context - request context object values - dict containing column values. """ - metadata = values.get('metadata') - metadata_refs = [] - if metadata: - for k, v in metadata.iteritems(): - metadata_ref = models.InstanceMetadata() - metadata_ref['key'] = k - metadata_ref['value'] = v - metadata_refs.append(metadata_ref) - values['metadata'] = metadata_refs + values['metadata'] = _metadata_refs(values.get('metadata')) instance_ref = models.Instance() instance_ref.update(values) @@ -1010,6 +1011,9 @@ def instance_set_state(context, instance_id, state, description=None): @require_context def instance_update(context, instance_id, values): session = get_session() + metadata = values.get('metadata') + if metadata: + values['metadata'] = _metadata_refs(values.get('metadata')) with session.begin(): instance_ref = instance_get(context, instance_id, session=session) instance_ref.update(values) diff --git a/nova/tests/integrated/api/client.py b/nova/tests/integrated/api/client.py index 7e20c9b00..eb9a3056e 100644 --- a/nova/tests/integrated/api/client.py +++ b/nova/tests/integrated/api/client.py @@ -152,7 +152,10 @@ class TestOpenStackClient(object): def _decode_json(self, response): body = response.read() LOG.debug(_("Decoding JSON: %s") % (body)) - return json.loads(body) + if body: + return json.loads(body) + else: + return "" def api_get(self, relative_uri, **kwargs): kwargs.setdefault('check_response_status', [200]) @@ -166,7 +169,7 @@ class TestOpenStackClient(object): headers['Content-Type'] = 'application/json' kwargs['body'] = json.dumps(body) - kwargs.setdefault('check_response_status', [200]) + kwargs.setdefault('check_response_status', [200, 202]) response = self.api_request(relative_uri, **kwargs) return self._decode_json(response) @@ -185,6 +188,9 @@ class TestOpenStackClient(object): def post_server(self, server): return self.api_post('/servers', server)['server'] + def post_server_action(self, server_id, data): + return self.api_post('/servers/%s/action' % server_id, data) + def delete_server(self, server_id): return self.api_delete('/servers/%s' % server_id) diff --git a/nova/tests/integrated/test_servers.py b/nova/tests/integrated/test_servers.py index e89d0100a..604faf59f 100644 --- a/nova/tests/integrated/test_servers.py +++ b/nova/tests/integrated/test_servers.py @@ -179,6 +179,40 @@ class ServersTest(integrated_helpers._IntegratedTestBase): # Cleanup self._delete_server(created_server_id) + def test_create_and_rebuild_server_with_metadata(self): + """Rebuild a server with metadata.""" + + # create a server with initially has no metadata + server = self._build_minimal_create_server_request() + server_post = {'server': server} + created_server = self.api.post_server(server_post) + LOG.debug("created_server: %s" % created_server) + self.assertTrue(created_server['id']) + created_server_id = created_server['id'] + + # rebuild the server with metadata + post = {} + post['rebuild'] = { + "imageRef": "https://localhost/v1.1/32278/images/2", + "name": "blah" + } + + metadata = {} + for i in range(30): + metadata['key_%s' % i] = 'value_%s' % i + + post['rebuild']['metadata'] = metadata + + self.api.post_server_action(created_server_id, post) + LOG.debug("rebuilt server: %s" % created_server) + self.assertTrue(created_server['id']) + + found_server = self.api.get_server(created_server_id) + self.assertEqual(created_server_id, found_server['id']) + self.assertEqual(metadata, found_server.get('metadata')) + + # Cleanup + self._delete_server(created_server_id) if __name__ == "__main__": unittest.main() -- cgit From 833481d796db557dddde6b4b9e75b7cf518b88fa Mon Sep 17 00:00:00 2001 From: Dan Prince Date: Sun, 29 May 2011 07:51:44 -0400 Subject: Use metadata variable when calling _metadata_refs. --- nova/db/sqlalchemy/api.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/nova/db/sqlalchemy/api.py b/nova/db/sqlalchemy/api.py index 1a7cae6e9..a678ebedd 100644 --- a/nova/db/sqlalchemy/api.py +++ b/nova/db/sqlalchemy/api.py @@ -1013,7 +1013,7 @@ def instance_update(context, instance_id, values): session = get_session() metadata = values.get('metadata') if metadata: - values['metadata'] = _metadata_refs(values.get('metadata')) + values['metadata'] = _metadata_refs(metadata) with session.begin(): instance_ref = instance_get(context, instance_id, session=session) instance_ref.update(values) -- cgit From 2155f2b1ab22c6183ab5266e16a675f1469fca50 Mon Sep 17 00:00:00 2001 From: Dan Prince Date: Mon, 30 May 2011 11:29:55 -0400 Subject: Updates so that 'name' can be updated when doing a OS API v1.1 rebuild. Fixed issue where metadata wasn't getting deleted when an empty dict was POST'd on a rebuild. --- nova/api/openstack/servers.py | 10 +++-- nova/compute/api.py | 13 ++++--- nova/db/sqlalchemy/api.py | 17 +++++++-- nova/tests/integrated/test_servers.py | 72 +++++++++++++++++++++++++++++++++++ 4 files changed, 100 insertions(+), 12 deletions(-) diff --git a/nova/api/openstack/servers.py b/nova/api/openstack/servers.py index 5c10fc916..8e191c232 100644 --- a/nova/api/openstack/servers.py +++ b/nova/api/openstack/servers.py @@ -708,14 +708,16 @@ class ControllerV11(Controller): image_id = common.get_id_from_href(image_ref) personalities = info["rebuild"].get("personality", []) - metadata = info["rebuild"].get("metadata", {}) + metadata = info["rebuild"].get("metadata") + name = info["rebuild"].get("name") - self._validate_metadata(metadata) + if metadata: + self._validate_metadata(metadata) self._decode_personalities(personalities) try: - self.compute_api.rebuild(context, instance_id, image_id, metadata, - personalities) + self.compute_api.rebuild(context, instance_id, image_id, name, + metadata, personalities) except exception.BuildInProgress: msg = _("Instance %d is currently being rebuilt.") % instance_id LOG.debug(msg) diff --git a/nova/compute/api.py b/nova/compute/api.py index 4f2363387..151679521 100644 --- a/nova/compute/api.py +++ b/nova/compute/api.py @@ -530,7 +530,7 @@ class API(base.Base): """Reboot the given instance.""" self._cast_compute_message('reboot_instance', context, instance_id) - def rebuild(self, context, instance_id, image_id, metadata=None, + def rebuild(self, context, instance_id, image_id, name=None, metadata=None, files_to_inject=None): """Rebuild the given instance with the provided metadata.""" instance = db.api.instance_get(context, instance_id) @@ -539,13 +539,16 @@ class API(base.Base): msg = _("Instance already building") raise exception.BuildInProgress(msg) - metadata = metadata or {} - self._check_metadata_properties_quota(context, metadata) - files_to_inject = files_to_inject or [] self._check_injected_file_quota(context, files_to_inject) - self.db.instance_update(context, instance_id, {"metadata": metadata}) + values = {} + if metadata is not None: + self._check_metadata_properties_quota(context, metadata) + values['metadata'] = metadata + if name is not None: + values['display_name'] = name + self.db.instance_update(context, instance_id, values) rebuild_params = { "image_id": image_id, diff --git a/nova/db/sqlalchemy/api.py b/nova/db/sqlalchemy/api.py index a678ebedd..ea84e96e7 100644 --- a/nova/db/sqlalchemy/api.py +++ b/nova/db/sqlalchemy/api.py @@ -1012,8 +1012,9 @@ def instance_set_state(context, instance_id, state, description=None): def instance_update(context, instance_id, values): session = get_session() metadata = values.get('metadata') - if metadata: - values['metadata'] = _metadata_refs(metadata) + if metadata is not None: + instance_metadata_update_or_create(context, instance_id, + values.pop('metadata'), True) with session.begin(): instance_ref = instance_get(context, instance_id, session=session) instance_ref.update(values) @@ -2570,8 +2571,12 @@ def instance_metadata_get_item(context, instance_id, key): @require_context -def instance_metadata_update_or_create(context, instance_id, metadata): +def instance_metadata_update_or_create(context, instance_id, metadata, + purge=False): session = get_session() + + original_metadata = instance_metadata_get(context, instance_id) + meta_ref = None for key, value in metadata.iteritems(): try: @@ -2583,4 +2588,10 @@ def instance_metadata_update_or_create(context, instance_id, metadata): "instance_id": instance_id, "deleted": 0}) meta_ref.save(session=session) + + if purge: + for key in original_metadata.keys(): + if not key in metadata.keys(): + instance_metadata_delete(context, instance_id, key) + return metadata diff --git a/nova/tests/integrated/test_servers.py b/nova/tests/integrated/test_servers.py index 604faf59f..a67fa1bb5 100644 --- a/nova/tests/integrated/test_servers.py +++ b/nova/tests/integrated/test_servers.py @@ -179,6 +179,36 @@ class ServersTest(integrated_helpers._IntegratedTestBase): # Cleanup self._delete_server(created_server_id) + def test_create_and_rebuild_server(self): + """Rebuild a server.""" + + # create a server with initially has no metadata + server = self._build_minimal_create_server_request() + server_post = {'server': server} + created_server = self.api.post_server(server_post) + LOG.debug("created_server: %s" % created_server) + self.assertTrue(created_server['id']) + created_server_id = created_server['id'] + + # rebuild the server with metadata + post = {} + post['rebuild'] = { + "imageRef": "https://localhost/v1.1/32278/images/2", + "name": "blah" + } + + self.api.post_server_action(created_server_id, post) + LOG.debug("rebuilt server: %s" % created_server) + self.assertTrue(created_server['id']) + + found_server = self.api.get_server(created_server_id) + self.assertEqual(created_server_id, found_server['id']) + self.assertEqual({}, found_server.get('metadata')) + self.assertEqual('blah', found_server.get('name')) + + # Cleanup + self._delete_server(created_server_id) + def test_create_and_rebuild_server_with_metadata(self): """Rebuild a server with metadata.""" @@ -210,9 +240,51 @@ class ServersTest(integrated_helpers._IntegratedTestBase): found_server = self.api.get_server(created_server_id) self.assertEqual(created_server_id, found_server['id']) self.assertEqual(metadata, found_server.get('metadata')) + self.assertEqual('blah', found_server.get('name')) + + # Cleanup + self._delete_server(created_server_id) + + def test_create_and_rebuild_server_with_metadata_removal(self): + """Rebuild a server with metadata.""" + + # create a server with initially has no metadata + server = self._build_minimal_create_server_request() + server_post = {'server': server} + + metadata = {} + for i in range(30): + metadata['key_%s' % i] = 'value_%s' % i + + server_post['server']['metadata'] = metadata + + created_server = self.api.post_server(server_post) + LOG.debug("created_server: %s" % created_server) + self.assertTrue(created_server['id']) + created_server_id = created_server['id'] + + # rebuild the server with metadata + post = {} + post['rebuild'] = { + "imageRef": "https://localhost/v1.1/32278/images/2", + "name": "blah" + } + + metadata = {} + post['rebuild']['metadata'] = metadata + + self.api.post_server_action(created_server_id, post) + LOG.debug("rebuilt server: %s" % created_server) + self.assertTrue(created_server['id']) + + found_server = self.api.get_server(created_server_id) + self.assertEqual(created_server_id, found_server['id']) + self.assertEqual(metadata, found_server.get('metadata')) + self.assertEqual('blah', found_server.get('name')) # Cleanup self._delete_server(created_server_id) + if __name__ == "__main__": unittest.main() -- cgit From be9113bc5c08cbafb7af9f83bd61f318d1ba6145 Mon Sep 17 00:00:00 2001 From: "Vivek YS vivek.ys@gmail.com" <> Date: Tue, 31 May 2011 09:49:06 +0530 Subject: Fixed the typo of APIError with ApiError --- nova/virt/vmwareapi/vmops.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/nova/virt/vmwareapi/vmops.py b/nova/virt/vmwareapi/vmops.py index c3e79a92f..6d7149841 100644 --- a/nova/virt/vmwareapi/vmops.py +++ b/nova/virt/vmwareapi/vmops.py @@ -590,11 +590,11 @@ class VMWareVMOps(object): def pause(self, instance, callback): """Pause a VM instance.""" - raise exception.APIError("pause not supported for vmwareapi") + raise exception.ApiError("pause not supported for vmwareapi") def unpause(self, instance, callback): """Un-Pause a VM instance.""" - raise exception.APIError("unpause not supported for vmwareapi") + raise exception.ApiError("unpause not supported for vmwareapi") def suspend(self, instance, callback): """Suspend the specified instance.""" @@ -673,7 +673,7 @@ class VMWareVMOps(object): def get_diagnostics(self, instance): """Return data about VM diagnostics.""" - raise exception.APIError("get_diagnostics not implemented for " + raise exception.ApiError("get_diagnostics not implemented for " "vmwareapi") def get_console_output(self, instance): -- cgit From 05c21ed52da0a9505000505120c70dffd6e730e2 Mon Sep 17 00:00:00 2001 From: "Vivek YS vivek.ys@gmail.com" <> Date: Tue, 31 May 2011 21:04:01 +0530 Subject: Added myself to Authors file --- Authors | 1 + 1 file changed, 1 insertion(+) diff --git a/Authors b/Authors index 50f4680a9..8dbdbfdb6 100644 --- a/Authors +++ b/Authors @@ -83,6 +83,7 @@ Trey Morris Tushar Patil Vasiliy Shlykov Vishvananda Ishaya +Vivek Y S William Wolf Yoshiaki Tamura Youcef Laribi -- cgit From 7beafb1aafac97e6dfc28108062785465cc8f577 Mon Sep 17 00:00:00 2001 From: Dan Prince Date: Tue, 31 May 2011 14:38:12 -0400 Subject: Use a new instance_metadata_delete_all DB api call to delete existing metadata when updating a server. --- nova/db/sqlalchemy/api.py | 22 ++++++++++++++-------- 1 file changed, 14 insertions(+), 8 deletions(-) diff --git a/nova/db/sqlalchemy/api.py b/nova/db/sqlalchemy/api.py index ea84e96e7..8df96cbf4 100644 --- a/nova/db/sqlalchemy/api.py +++ b/nova/db/sqlalchemy/api.py @@ -1013,8 +1013,9 @@ def instance_update(context, instance_id, values): session = get_session() metadata = values.get('metadata') if metadata is not None: + instance_metadata_delete_all(context, instance_id) instance_metadata_update_or_create(context, instance_id, - values.pop('metadata'), True) + values.pop('metadata')) with session.begin(): instance_ref = instance_get(context, instance_id, session=session) instance_ref.update(values) @@ -2554,6 +2555,17 @@ def instance_metadata_delete(context, instance_id, key): 'updated_at': literal_column('updated_at')}) +@require_context +def instance_metadata_delete_all(context, instance_id): + session = get_session() + session.query(models.InstanceMetadata).\ + filter_by(instance_id=instance_id).\ + filter_by(deleted=False).\ + update({'deleted': True, + 'deleted_at': datetime.datetime.utcnow(), + 'updated_at': literal_column('updated_at')}) + + @require_context def instance_metadata_get_item(context, instance_id, key): session = get_session() @@ -2571,8 +2583,7 @@ def instance_metadata_get_item(context, instance_id, key): @require_context -def instance_metadata_update_or_create(context, instance_id, metadata, - purge=False): +def instance_metadata_update_or_create(context, instance_id, metadata): session = get_session() original_metadata = instance_metadata_get(context, instance_id) @@ -2589,9 +2600,4 @@ def instance_metadata_update_or_create(context, instance_id, metadata, "deleted": 0}) meta_ref.save(session=session) - if purge: - for key in original_metadata.keys(): - if not key in metadata.keys(): - instance_metadata_delete(context, instance_id, key) - return metadata -- cgit