This adds a volume snapshot support with the EC2 api.

13 files changed, 535 insertions, 35 deletions

diff --git a/Authors b/Authors
index 50f4680a9..26e1281f4 100644
--- a/Authors
+++ b/Authors
@@ -30,6 +30,7 @@ Gabe Westmaas <gabe.westmaas@rackspace.com>
 Hisaharu Ishii <ishii.hisaharu@lab.ntt.co.jp>
 Hisaki Ohara <hisaki.ohara@intel.com>
 Ilya Alekseyev <ialekseev@griddynamics.com>
+Isaku Yamahata <yamahata@valinux.co.jp>
 Jason Koelker <jason@koelker.net>
 Jay Pipes <jaypipes@gmail.com>
 Jesse Andrews <anotherjesse@gmail.com>
diff --git a/nova/api/ec2/__init__.py b/nova/api/ec2/__init__.py
index c13993dd3..1915d007d 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 c35b6024e..d92838f38 100644
--- a/nova/api/ec2/cloud.py
+++ b/nova/api/ec2/cloud.py
@@ -283,14 +283,50 @@ 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 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)
@@ -619,6 +655,11 @@ 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 310c0bb09..4e0aa60a2 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 snapshot names')
 
 
 IMPL = utils.LazyPluggable(FLAGS['db_backend'],
@@ -881,6 +883,43 @@ def volume_update(context, volume_id, values):
 ####################
 
 
+def snapshot_create(context, values):
+    """Create a snapshot from the values dictionary."""
+    return IMPL.snapshot_create(context, values)
+
+
+def snapshot_destroy(context, snapshot_id):
+    """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 snapshot or raise if it does not exist."""
+    return IMPL.snapshot_get(context, snapshot_id)
+
+
+def snapshot_get_all(context):
+    """Get all snapshots."""
+    return IMPL.snapshot_get_all(context)
+
+
+def snapshot_get_all_by_project(context, project_id):
+    """Get all snapshots 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 e4dda5c12..d53b76f4a 100644
--- a/nova/db/sqlalchemy/api.py
+++ b/nova/db/sqlalchemy/api.py
@@ -1790,6 +1790,82 @@ 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_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()
     return session.query(models.SecurityGroup).\
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
new file mode 100644
index 000000000..f16d6db56
--- /dev/null
+++ b/nova/db/sqlalchemy/migrate_repo/versions/019_add_volume_snapshot_support.py
@@ -0,0 +1,70 @@
+# 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 Column, Table, MetaData
+from sqlalchemy import Integer, DateTime, Boolean, String
+
+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 1215448f8..480f62399 100644
--- a/nova/db/sqlalchemy/models.py
+++ b/nova/db/sqlalchemy/models.py
@@ -329,6 +329,31 @@ class Quota(BASE, NovaBase):
     hard_limit = Column(Integer, nullable=True)
 
 
+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 56c20d111..02c65fd64 100644
--- a/nova/exception.py
+++ b/nova/exception.py
@@ -271,6 +271,14 @@ class VolumeNotFoundForInstance(VolumeNotFound):
     message = _("Volume not found for instance %(instance_id)s.")
 
 
+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/tests/test_cloud.py b/nova/tests/test_cloud.py
index b64be662e..34a73ad1f 100644
--- a/nova/tests/test_cloud.py
+++ b/nova/tests/test_cloud.py
@@ -186,6 +186,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',
diff --git a/nova/tests/test_volume.py b/nova/tests/test_volume.py
index 236d12434..3472b1f59 100644
--- a/nova/tests/test_volume.py
+++ b/nova/tests/test_volume.py
@@ -176,6 +176,34 @@ 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."""
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..21cc228c9 100644
--- a/nova/volume/driver.py
+++ b/nova/volume/driver.py
@@ -90,42 +90,91 @@ 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)
+                          volume_name, FLAGS.volume_group)
 
-    def delete_volume(self, volume):
-        """Deletes a logical volume."""
+    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']))
+                           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."""
+        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
+
+        # 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']))
+        # 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'])
+
+    def create_snapshot(self, snapshot):
+        """Creates a snapshot."""
+        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."""
+        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):
@@ -559,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
@@ -600,18 +661,24 @@ 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"""
         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']
 
diff --git a/nova/volume/manager.py b/nova/volume/manager.py
index 2178389ce..40a104d35 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'],
@@ -152,6 +158,49 @@ 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.