Merge lp:~vishvananda/nova/no-db-messaging into lp:~hudson-openstack/nova/trunk

[Vish Ishaya]

This is an initial proposal for feedback. This branch is an attempt to start on this blueprint:
https://blueprints.launchpad.net/nova/+spec/no-db-messaging
which will allow for the implementation of this blueprint:
https://blueprints.launchpad.net/nova/+spec/separate-code-for-services
And will ultimately make it easy to replace our various services with external projects.

This prototype changes volume_create and volume_delete to pass data through the queue instead of writing information to the database and reading it on the other end. It attempts to make minimal changes. It includes:
 * a small change to model code to allow for conversion into dicts
 * scheduler uses muticall instead of cast
 * volume.api.create_volume and delete volume use multicall to communicate with volume.manager
 * volume.manager returns updates instead of modifying the database directly

Please note that this is an initial proposal. It is based on some changes made by termie to allow the driver to return multiple times for a single call. It works to create volumes, but I haven't modified the tests to pass yet.

To Do:
 * change the driver code so it doesn't store its extra data about volumes in the global db.
 * fix the tests
 * modify attach and detach to use the same methodology

I'm open to feedback about this approach. I tried a few other versions and this seems like the simplest change set to get what we want. If this looks good, I will modify the other volume commands to work the same way and propose a similar set of changes for compute.

[Dan Prince (dan-prince)]

Hey Vish,

I'm not able to boot any instances via the OS API w/ this branch. Looks to be related to the set_admin_password functionality which now polls for the compute host to be assigned:

(nova.api.openstack): TRACE: File "/usr/lib/pymodules/python2.6/nova/compute/api.py", line 501, in _set_admin_password
(nova.api.openstack): TRACE: host = self._find_host(context, instance_id)
(nova.api.openstack): TRACE: File "/usr/lib/pymodules/python2.6/nova/compute/api.py", line 497, in _find_host
(nova.api.openstack): TRACE: % instance_id)
(nova.api.openstack): TRACE: Error: Unable to find host for Instance 1
(nova.api.openstack): TRACE:

review: Needs Fixing

[Vish Ishaya (vishvananda)]

Thanks dan,

Turns out it was an error where if you cast to a generator (multicall uses generators), then it wouldn't actually execute the generator on the other end. I think casting should work now.

Vish

[OpenStack Infra (hudson-openstack)]

The prerequisite lp:~termie/nova/rpc_multicall has not yet been merged into lp:nova.

[Dan Prince (dan-prince)]

Hey Vish,

So you latest fix resolves the RPC issues. Thanks!

I'm still hitting an issue with things like instance metadata (aka a model attributes that aren't columns).

The following changes to NovaBase.update resolves it for me.

            if key == 'name' and key in columns:
                setattr(self, key, value)
            else:
                setattr(self, key, value)

review: Needs Fixing

[Dan Prince (dan-prince)]

Oops. This one should work:

      if key == 'name' and key in columns:
            setattr(self, key, value)
       elif key != 'name':
            setattr(self, key, value)

[Mark Washenberger (markwash)]

I'm having a little trouble grasping the multicall approach as shown here so I might be a bit confused.

It looks like the way this works is, when we schedule a volume to be created, we create an eventlet-based volume listener which will update the database as it hears back from the volume manager.

Maybe I'm mistaken about eventlet, but this implies that if for some reason the process terminates before the multicall has finished its last return, then the database won't be updated. Then, when the process is restarted, there will be no calling context remaining to handle the updates.

Would it be a more robust approach to create a permanently running VolumeListener that handles all volume update events?

Thanks!

[lp:~vishvananda/nova/no-db-messaging]

1113. By Vish Ishaya on 2011-05-25

merged rpc_multicall

1114. By Vish Ishaya on 2011-05-26

keep the database on the receiving end as well

1115. By Vish Ishaya on 2011-05-26

fix tests

1116. By Vish Ishaya on 2011-05-26

use strtime for passing datetimes back and forth through the queue

1117. By Vish Ishaya on 2011-05-26

lost some changes from rpc branch, bring them in manually

1118. By Vish Ishaya on 2011-06-01

merged trunk and removed conflicts

1119. By Vish Ishaya on 2011-06-02

make sure to handle VolumeIsBusy

1120. By Vish Ishaya on 2011-06-02

return not yield in scheduler shortcut

1121. By Vish Ishaya on 2011-06-02

fix snapshot test

[Dan Prince (dan-prince)]

Sorry for holding this up. I'm no multicall expert but I should probably remove my 'needs fixing' at least so more people check it out. The latest fix resolves my issue with model attributes that aren't columns.

review: Abstain

[Rick Harris (rconradharris)]

Ran into a test failure:

======================================================================
FAIL: test_create_and_delete_volume (nova.tests.integrated.test_volumes.VolumesTest)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/home/rick/openstack/nova/no-db-messaging/nova/tests/integrated/test_volumes.py", line 122, in test_create_and_delete_volume
    self.assertEquals(1, len(create_actions))
AssertionError: 1 != 2
-------------------- >> begin captured logging << --------------------

Other than that, I think this looks good.

review: Needs Fixing

[Dan Prince (dan-prince)]

Couple of conflicts now w/ a trunk merge too:

Text conflict in nova/volume/api.py
Text conflict in nova/volume/manager.py
2 conflicts encountered.

[Matt Dietz (cerberus)]

I really like this approach, but I have the same reservations that Mark does.

We could go with the Listener approach, but I will say that I've seen issues with bottlenecking in our current architecture trying to do something very similar.

It seems like this would make updating a running nova installation prohibitively difficult. If there was a way to recall the calling context/msg_id even after a worker bounce, then I'd feel a lot better.
We may need some kind of state dump mechanism for the workers. Perhaps we could pickle some pertinent data upon a worker receiving a HUP/KILL/whatever.

review: Needs Information

[lp:~vishvananda/nova/no-db-messaging]

1122. By Vish Ishaya on 2011-06-13

remove merge error calling failing test

1123. By Vish Ishaya on 2011-06-22

merged trunk

Unmerged revisions

1123. By Vish Ishaya on 2011-06-22

merged trunk

1122. By Vish Ishaya on 2011-06-13

remove merge error calling failing test

1121. By Vish Ishaya on 2011-06-02

fix snapshot test

1120. By Vish Ishaya on 2011-06-02

return not yield in scheduler shortcut

1119. By Vish Ishaya on 2011-06-02

make sure to handle VolumeIsBusy

1118. By Vish Ishaya on 2011-06-01

merged trunk and removed conflicts

1117. By Vish Ishaya on 2011-05-26

lost some changes from rpc branch, bring them in manually

1116. By Vish Ishaya on 2011-05-26

use strtime for passing datetimes back and forth through the queue

1115. By Vish Ishaya on 2011-05-26

fix tests

1114. By Vish Ishaya on 2011-05-26

keep the database on the receiving end as well

=== modified file 'nova/context.py'
--- nova/context.py	2011-06-02 21:23:05 +0000
+++ nova/context.py	2011-06-22 17:01:37 +0000
@@ -56,8 +56,8 @@
         self.remote_address = remote_address
         if not timestamp:
             timestamp = utils.utcnow()
-        if isinstance(timestamp, str) or isinstance(timestamp, unicode):
-            timestamp = utils.parse_isotime(timestamp)
+        if isinstance(timestamp, basestring):
+            timestamp = utils.parse_strtime(timestamp)
         self.timestamp = timestamp
         if not request_id:
             chars = 'ABCDEFGHIJKLMNOPQRSTUVWXYZ1234567890-'
@@ -95,7 +95,7 @@
                 'is_admin': self.is_admin,
                 'read_deleted': self.read_deleted,
                 'remote_address': self.remote_address,
-                'timestamp': utils.isotime(self.timestamp),
+                'timestamp': utils.strtime(self.timestamp),
                 'request_id': self.request_id}
 
     @classmethod
 
=== modified file 'nova/db/sqlalchemy/api.py'
--- nova/db/sqlalchemy/api.py	2011-06-20 20:55:16 +0000
+++ nova/db/sqlalchemy/api.py	2011-06-22 17:01:37 +0000
@@ -1689,7 +1689,7 @@
     return (result[0] or 0, result[1] or 0)
 
 
-@require_admin_context
+@require_context
 def volume_destroy(context, volume_id):
     session = get_session()
     with session.begin():
 
=== modified file 'nova/db/sqlalchemy/models.py'
--- nova/db/sqlalchemy/models.py	2011-06-20 20:55:16 +0000
+++ nova/db/sqlalchemy/models.py	2011-06-22 17:01:37 +0000
@@ -25,6 +25,7 @@
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.declarative import declarative_base
 from sqlalchemy.schema import ForeignKeyConstraint
+from sqlalchemy.types import DateTime as DTType
 
 from nova.db.sqlalchemy.session import get_session
 
@@ -77,17 +78,29 @@
         return getattr(self, key, default)
 
     def __iter__(self):
-        self._i = iter(object_mapper(self).columns)
+        # NOTE(vish): include name property in the iterator
+        columns = dict(object_mapper(self).columns).keys()
+        name = self.get('name')
+        if name:
+            columns.append('name')
+        self._i = iter(columns)
         return self
 
     def next(self):
-        n = self._i.next().name
+        n = self._i.next()
         return n, getattr(self, n)
 
     def update(self, values):
-        """Make the model object behave like a dict"""
-        for k, v in values.iteritems():
-            setattr(self, k, v)
+        """Make the model object behave like a dict and convert datetimes."""
+        columns = object_mapper(self).columns
+        for key, value in values.iteritems():
+            # NOTE(vish): don't update the 'name' property
+            if key != 'name' or key in columns:
+                if (key in columns and
+                    isinstance(value, basestring) and
+                    isinstance(columns[key].type, DTType)):
+                    value = utils.parse_strtime(value)
+                setattr(self, key, value)
 
     def iteritems(self):
         """Make the model object behave like a dict.
 
=== modified file 'nova/scheduler/manager.py'
--- nova/scheduler/manager.py	2011-06-09 23:16:55 +0000
+++ nova/scheduler/manager.py	2011-06-22 17:01:37 +0000
@@ -98,11 +98,13 @@
                         % locals())
             return
 
-        rpc.cast(context,
-                 db.queue_get_for(context, topic, host),
-                 {"method": method,
-                  "args": kwargs})
-        LOG.debug(_("Casted to %(topic)s %(host)s for %(method)s") % locals())
+        LOG.debug(_("Multicall %(topic)s %(host)s for %(method)s") % locals())
+        rvs = rpc.multicall(context,
+                            db.queue_get_for(context, topic, host),
+                            {"method": method,
+                             "args": kwargs})
+        for rv in rvs:
+            yield rv
 
     # NOTE (masumotok) : This method should be moved to nova.api.ec2.admin.
     #                    Based on bexar design summit discussion,
 
=== modified file 'nova/tests/scheduler/test_scheduler.py'
--- nova/tests/scheduler/test_scheduler.py	2011-06-17 23:53:30 +0000
+++ nova/tests/scheduler/test_scheduler.py	2011-06-22 17:01:37 +0000
@@ -98,7 +98,7 @@
 
     def test_fallback(self):
         scheduler = manager.SchedulerManager()
-        self.mox.StubOutWithMock(rpc, 'cast', use_mock_anything=True)
+        self.mox.StubOutWithMock(rpc, 'call', use_mock_anything=True)
         ctxt = context.get_admin_context()
         rpc.cast(ctxt,
                  'topic.fallback_host',
@@ -109,7 +109,7 @@
 
     def test_named_method(self):
         scheduler = manager.SchedulerManager()
-        self.mox.StubOutWithMock(rpc, 'cast', use_mock_anything=True)
+        self.mox.StubOutWithMock(rpc, 'call', use_mock_anything=True)
         ctxt = context.get_admin_context()
         rpc.cast(ctxt,
                  'topic.named_host',
@@ -225,17 +225,17 @@
         self.mox.StubOutWithMock(db, 'service_get_all_by_topic')
         arg = IgnoreArg()
         db.service_get_all_by_topic(arg, arg).AndReturn(service_list)
-        self.mox.StubOutWithMock(rpc, 'cast', use_mock_anything=True)
-        rpc.cast(ctxt,
-                 'compute.host1',
-                 {'method': 'run_instance',
-                  'args': {'instance_id': 'i-ffffffff',
-                           'availability_zone': 'zone1'}})
+        self.mox.StubOutWithMock(rpc, 'multicall', use_mock_anything=True)
+        rpc.multicall(ctxt,
+                      'compute.host1',
+                      {'method': 'run_instance',
+                       'args': {'instance_id': 'i-ffffffff',
+                                'availability_zone': 'zone1'}}).AndReturn([])
         self.mox.ReplayAll()
-        scheduler.run_instance(ctxt,
-                               'compute',
-                               instance_id='i-ffffffff',
-                               availability_zone='zone1')
+        list(scheduler.run_instance(ctxt,
+                                    'compute',
+                                    instance_id='i-ffffffff',
+                                    availability_zone='zone1'))
 
 
 class SimpleDriverTestCase(test.TestCase):
@@ -601,12 +601,13 @@
         volume1 = self.start_service('volume', host='host1')
         volume2 = self.start_service('volume', host='host2')
         volume_id1 = self._create_volume()
-        volume1.create_volume(self.context, volume_id1)
+        host1 = self.scheduler.driver.schedule_create_volume(self.context,
+                                                             volume_id1)
         volume_id2 = self._create_volume()
-        host = self.scheduler.driver.schedule_create_volume(self.context,
-                                                            volume_id2)
-        self.assertEqual(host, 'host2')
-        volume1.delete_volume(self.context, volume_id1)
+        host2 = self.scheduler.driver.schedule_create_volume(self.context,
+                                                             volume_id2)
+        self.assertNotEqual(host1, host2)
+        db.volume_destroy(self.context, volume_id1)
         db.volume_destroy(self.context, volume_id2)
         volume1.kill()
         volume2.kill()
@@ -619,10 +620,12 @@
         volume_ids2 = []
         for index in xrange(FLAGS.max_gigabytes):
             volume_id = self._create_volume()
-            volume1.create_volume(self.context, volume_id)
+            self.scheduler.driver.schedule_create_volume(self.context,
+                                                         volume_id)
             volume_ids1.append(volume_id)
             volume_id = self._create_volume()
-            volume2.create_volume(self.context, volume_id)
+            self.scheduler.driver.schedule_create_volume(self.context,
+                                                         volume_id)
             volume_ids2.append(volume_id)
         volume_id = self._create_volume()
         self.assertRaises(driver.NoValidHost,
@@ -658,16 +661,18 @@
         driver_i._live_migration_src_check(nocare, nocare)
         driver_i._live_migration_dest_check(nocare, nocare, i_ref['host'])
         driver_i._live_migration_common_check(nocare, nocare, i_ref['host'])
-        self.mox.StubOutWithMock(rpc, 'cast', use_mock_anything=True)
+        self.mox.StubOutWithMock(rpc, 'multicall', use_mock_anything=True)
         kwargs = {'instance_id': instance_id, 'dest': i_ref['host']}
-        rpc.cast(self.context,
-                 db.queue_get_for(nocare, FLAGS.compute_topic, i_ref['host']),
-                 {"method": 'live_migration', "args": kwargs})
-
+        rpc.multicall(self.context,
+                      db.queue_get_for(nocare,
+                                       FLAGS.compute_topic,
+                                       i_ref['host']),
+                      {"method": 'live_migration',
+                       "args": kwargs}).AndReturn([])
         self.mox.ReplayAll()
-        self.scheduler.live_migration(self.context, FLAGS.compute_topic,
-                                      instance_id=instance_id,
-                                      dest=i_ref['host'])
+        list(self.scheduler.live_migration(self.context, FLAGS.compute_topic,
+                                           instance_id=instance_id,
+                                           dest=i_ref['host']))
 
         i_ref = db.instance_get(self.context, instance_id)
         self.assertTrue(i_ref['state_description'] == 'migrating')
 
=== modified file 'nova/tests/test_volume.py'
--- nova/tests/test_volume.py	2011-05-27 05:13:17 +0000
+++ nova/tests/test_volume.py	2011-06-22 17:01:37 +0000
@@ -57,14 +57,24 @@
         vol['attach_status'] = "detached"
         return db.volume_create(context.get_admin_context(), vol)['id']
 
+    def _id_create_volume(self, context, volume_id):
+        """Version of create volume that uses id"""
+        volume_ref = utils.to_primitive(db.volume_get(context, volume_id))
+        return list(self.volume.create_volume(context, volume_ref))[-1]['id']
+
+    def _id_delete_volume(self, context, volume_id):
+        """Version of delete volume that uses id"""
+        volume_ref = utils.to_primitive(db.volume_get(context, volume_id))
+        return list(self.volume.delete_volume(context, volume_ref))[-1]
+
     def test_create_delete_volume(self):
         """Test volume can be created and deleted."""
         volume_id = self._create_volume()
-        self.volume.create_volume(self.context, volume_id)
+        self._id_create_volume(self.context, volume_id)
         self.assertEqual(volume_id, db.volume_get(context.get_admin_context(),
                          volume_id).id)
 
-        self.volume.delete_volume(self.context, volume_id)
+        self._id_delete_volume(self.context, volume_id)
         self.assertRaises(exception.NotFound,
                           db.volume_get,
                           self.context,
@@ -77,7 +87,7 @@
         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._id_create_volume(self.context, volume_dst_id)
         self.assertEqual(volume_dst_id, db.volume_get(
                 context.get_admin_context(),
                 volume_dst_id).id)
@@ -96,7 +106,7 @@
         return True
         try:
             volume_id = self._create_volume('1001')
-            self.volume.create_volume(self.context, volume_id)
+            self._id_create_volume(self.context, volume_id)
             self.fail("Should have thrown TypeError")
         except TypeError:
             pass
@@ -107,16 +117,16 @@
         total_slots = FLAGS.iscsi_num_targets
         for _index in xrange(total_slots):
             volume_id = self._create_volume()
-            self.volume.create_volume(self.context, volume_id)
+            self._id_create_volume(self.context, volume_id)
             vols.append(volume_id)
         volume_id = self._create_volume()
         self.assertRaises(db.NoMoreTargets,
-                          self.volume.create_volume,
+                          self._id_create_volume,
                           self.context,
                           volume_id)
         db.volume_destroy(context.get_admin_context(), volume_id)
         for volume_id in vols:
-            self.volume.delete_volume(self.context, volume_id)
+            self._id_delete_volume(self.context, volume_id)
 
     def test_run_attach_detach_volume(self):
         """Make sure volume can be attached and detached from instance."""
@@ -132,7 +142,7 @@
         instance_id = db.instance_create(self.context, inst)['id']
         mountpoint = "/dev/sdf"
         volume_id = self._create_volume()
-        self.volume.create_volume(self.context, volume_id)
+        self._id_create_volume(self.context, volume_id)
         if FLAGS.fake_tests:
             db.volume_attached(self.context, volume_id, instance_id,
                                mountpoint)
@@ -148,10 +158,6 @@
         instance_ref = db.volume_get_instance(self.context, volume_id)
         self.assertEqual(instance_ref['id'], instance_id)
 
-        self.assertRaises(exception.Error,
-                          self.volume.delete_volume,
-                          self.context,
-                          volume_id)
         if FLAGS.fake_tests:
             db.volume_detached(self.context, volume_id)
         else:
@@ -161,7 +167,7 @@
         vol = db.volume_get(self.context, volume_id)
         self.assertEqual(vol['status'], "available")
 
-        self.volume.delete_volume(self.context, volume_id)
+        self._id_delete_volume(self.context, volume_id)
         self.assertRaises(exception.VolumeNotFound,
                           db.volume_get,
                           self.context,
@@ -185,10 +191,10 @@
         total_slots = FLAGS.iscsi_num_targets
         for _index in xrange(total_slots):
             volume_id = self._create_volume()
-            d = self.volume.create_volume(self.context, volume_id)
+            d = self._id_create_volume(self.context, volume_id)
             _check(d)
         for volume_id in volume_ids:
-            self.volume.delete_volume(self.context, volume_id)
+            self._id_delete_volume(self.context, volume_id)
 
     def test_multi_node(self):
         # TODO(termie): Figure out how to test with two nodes,
@@ -253,6 +259,16 @@
     def tearDown(self):
         super(DriverTestCase, self).tearDown()
 
+    def _id_create_volume(self, context, volume_id):
+        """Version of create volume that uses id"""
+        volume_ref = utils.to_primitive(db.volume_get(context, volume_id))
+        return list(self.volume.create_volume(context, volume_ref))[-1]['id']
+
+    def _id_delete_volume(self, context, volume_id):
+        """Version of delete volume that uses id"""
+        volume_ref = utils.to_primitive(db.volume_get(context, volume_id))
+        return list(self.volume.delete_volume(context, volume_ref))[-1]
+
     def _attach_volume(self):
         """Attach volumes to an instance. This function also sets
            a fake log message."""
@@ -262,7 +278,7 @@
         """Detach volumes from an instance."""
         for volume_id in volume_id_list:
             db.volume_detached(self.context, volume_id)
-            self.volume.delete_volume(self.context, volume_id)
+            self._id_delete_volume(self.context, volume_id)
 
 
 class AOETestCase(DriverTestCase):
@@ -284,7 +300,7 @@
             vol['size'] = 0
             volume_id = db.volume_create(self.context,
                                          vol)['id']
-            self.volume.create_volume(self.context, volume_id)
+            self._id_create_volume(self.context, volume_id)
 
             # each volume has a different mountpoint
             mountpoint = "/dev/sd" + chr((ord('b') + index))
@@ -360,7 +376,7 @@
             vol = {}
             vol['size'] = 0
             vol_ref = db.volume_create(self.context, vol)
-            self.volume.create_volume(self.context, vol_ref['id'])
+            self._id_create_volume(self.context, vol_ref['id'])
             vol_ref = db.volume_get(self.context, vol_ref['id'])
 
             # each volume has a different mountpoint
 
=== modified file 'nova/utils.py'
--- nova/utils.py	2011-06-18 00:12:44 +0000
+++ nova/utils.py	2011-06-22 17:01:37 +0000
@@ -50,6 +50,7 @@
 
 LOG = logging.getLogger("nova.utils")
 TIME_FORMAT = "%Y-%m-%dT%H:%M:%SZ"
+PERFECT_TIME_FORMAT = "%Y-%m-%dT%H:%M:%S.%f"
 FLAGS = flags.FLAGS
 
 
@@ -362,6 +363,18 @@
     return datetime.datetime.strptime(timestr, TIME_FORMAT)
 
 
+def strtime(at=None):
+    """Returns iso formatted utcnow."""
+    if not at:
+        at = utcnow()
+    return at.strftime(PERFECT_TIME_FORMAT)
+
+
+def parse_strtime(timestr):
+    """Turn an iso formatted time back into a datetime."""
+    return datetime.datetime.strptime(timestr, PERFECT_TIME_FORMAT)
+
+
 def parse_mailmap(mailmap='.mailmap'):
     mapping = {}
     if os.path.exists(mailmap):
@@ -505,7 +518,7 @@
             o[k] = to_primitive(v)
         return o
     elif isinstance(value, datetime.datetime):
-        return str(value)
+        return strtime(value)
     elif hasattr(value, 'iteritems'):
         return to_primitive(dict(value.iteritems()))
     elif hasattr(value, '__iter__'):
 
=== modified file 'nova/volume/api.py'
--- nova/volume/api.py	2011-06-15 16:46:24 +0000
+++ nova/volume/api.py	2011-06-22 17:01:37 +0000
@@ -20,10 +20,8 @@
 Handles all requests relating to volumes.
 """
 
-
-from eventlet import greenthread
-
-from nova import db
+import eventlet
+
 from nova import exception
 from nova import flags
 from nova import log as logging
@@ -68,14 +66,27 @@
             'display_name': name,
             'display_description': description}
 
-        volume = self.db.volume_create(context, options)
-        rpc.cast(context,
-                 FLAGS.scheduler_topic,
-                 {"method": "create_volume",
-                  "args": {"topic": FLAGS.volume_topic,
-                           "volume_id": volume['id'],
-                           "snapshot_id": snapshot_id}})
-        return volume
+        volume_ref = self.db.volume_create(context, options)
+        volume_ref = utils.to_primitive(volume_ref)
+
+        def delayed_create(volume_ref):
+            vid = volume_ref['id']
+            try:
+                rvs = rpc.multicall(context,
+                                    FLAGS.scheduler_topic,
+                                    {"method": "create_volume",
+                                     "args": {"topic": FLAGS.volume_topic,
+                                              "volume_ref": volume_ref}})
+                for volume_ref in rvs:
+                    self.db.volume_update(context, vid, volume_ref)
+                volume_ref['launched_at'] = utils.utcnow()
+                self.db.volume_update(context, vid, volume_ref)
+
+            except rpc.RemoteError:
+                self.db.volume_update(context, vid, {'status': 'error'})
+
+        eventlet.spawn_n(delayed_create, volume_ref)
+        return volume_ref
 
     # TODO(yamahata): eliminate dumb polling
     def wait_creation(self, context, volume_id):
@@ -83,20 +94,44 @@
             volume = self.get(context, volume_id)
             if volume['status'] != 'creating':
                 return
-            greenthread.sleep(1)
+            eventlet.greenthread.sleep(1)
 
     def delete(self, context, volume_id):
-        volume = self.get(context, volume_id)
-        if volume['status'] != "available":
+        volume_ref = self.get(context, volume_id)
+        if volume_ref['status'] != "available":
             raise exception.ApiError(_("Volume status must be available"))
-        now = utils.utcnow()
-        self.db.volume_update(context, volume_id, {'status': 'deleting',
-                                                   'terminated_at': now})
-        host = volume['host']
-        rpc.cast(context,
-                 self.db.queue_get_for(context, FLAGS.volume_topic, host),
-                 {"method": "delete_volume",
-                  "args": {"volume_id": volume_id}})
+        if volume_ref['attach_status'] == "attached":
+            raise exception.Error(_("Volume is still attached"))
+
+        volume_ref['status'] = 'deleting'
+        volume_ref['terminated_at'] = utils.utcnow()
+        self.db.volume_update(context, volume_ref['id'], volume_ref)
+        volume_ref = utils.to_primitive(volume_ref)
+
+        def delayed_delete(volume_ref):
+            vid = volume_ref['id']
+            try:
+                topic = self.db.queue_get_for(context,
+                                              FLAGS.volume_topic,
+                                              volume_ref['host'])
+                rvs = rpc.multicall(context,
+                                    topic,
+                                    {"method": "delete_volume",
+                                     "args": {"volume_ref": volume_ref}})
+                for volume_ref in rvs:
+                    self.db.volume_update(context, vid, volume_ref)
+
+                # NOTE(vish): See TODO in manager.py.  This can be removed
+                #             if change to a better method for handling
+                #             deletes
+                if volume_ref['status'] != 'available':
+                    self.db.volume_destroy(context, vid)
+
+            except rpc.RemoteError:
+                self.db.volume_update(context, vid, {'status': 'err_delete'})
+
+        eventlet.spawn_n(delayed_delete, volume_ref)
+        return True
 
     def update(self, context, volume_id, fields):
         self.db.volume_update(context, volume_id, fields)
 
=== modified file 'nova/volume/manager.py'
--- nova/volume/manager.py	2011-06-02 21:23:05 +0000
+++ nova/volume/manager.py	2011-06-22 17:01:37 +0000
@@ -88,79 +88,72 @@
             else:
                 LOG.info(_("volume %s: skipping export"), volume['name'])
 
-    def create_volume(self, context, volume_id, snapshot_id=None):
+    def create_volume(self, context, volume_ref):
         """Creates and exports the volume."""
-        context = context.elevated()
-        volume_ref = self.db.volume_get(context, volume_id)
         LOG.info(_("volume %s: creating"), volume_ref['name'])
 
-        self.db.volume_update(context,
-                              volume_id,
-                              {'host': self.host})
-        # NOTE(vish): so we don't have to get volume from db again
-        #             before passing it to the driver.
+        @utils.synchronized(volume_ref['name'])
+        def safe_create(volume_ref):
+            try:
+                volume_ref = self.db.volume_get(context, volume_ref['id'])
+            except exception.VolumeNotFound:
+                volume_ref = self.db.volume_create(context, volume_ref)
+            return volume_ref
+
+        volume_ref = safe_create(volume_ref)
         volume_ref['host'] = self.host
-
-        try:
-            vol_name = volume_ref['name']
-            vol_size = volume_ref['size']
-            LOG.debug(_("volume %(vol_name)s: creating lv of"
-                    " size %(vol_size)sG") % locals())
-            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)
-
-            LOG.debug(_("volume %s: creating export"), volume_ref['name'])
-            model_update = self.driver.create_export(context, volume_ref)
-            if model_update:
-                self.db.volume_update(context, volume_ref['id'], model_update)
-        except Exception:
-            self.db.volume_update(context,
-                                  volume_ref['id'], {'status': 'error'})
-            raise
-
-        now = utils.utcnow()
-        self.db.volume_update(context,
-                              volume_ref['id'], {'status': 'available',
-                                                 'launched_at': now})
+        self.db.volume_update(context, volume_ref['id'], volume_ref)
+        yield volume_ref
+
+        vol_name = volume_ref['name']
+        vol_size = volume_ref['size']
+        LOG.debug(_("volume %(vol_name)s: creating lv of"
+                " size %(vol_size)sG") % locals())
+        snapshot_id = volume_ref['snapshot_id']
+        if snapshot_id is 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:
+            volume_ref.update(model_update)
+            self.db.volume_update(context, volume_ref['id'], model_update)
+            yield volume_ref
+
+        LOG.debug(_("volume %s: creating export"), volume_ref['name'])
+        model_update = self.driver.create_export(context, volume_ref)
+        if model_update:
+            volume_ref.update(model_update)
+            self.db.volume_update(context, volume_ref['id'], model_update)
+            yield volume_ref
+
         LOG.debug(_("volume %s: created successfully"), volume_ref['name'])
-        return volume_id
+        volume_ref['status'] = 'available'
+        self.db.volume_update(context, volume_ref['id'], volume_ref)
+        yield volume_ref
 
-    def delete_volume(self, context, volume_id):
+    def delete_volume(self, context, volume_ref):
         """Deletes and unexports volume."""
-        context = context.elevated()
-        volume_ref = self.db.volume_get(context, volume_id)
-        if volume_ref['attach_status'] == "attached":
-            raise exception.Error(_("Volume is still attached"))
-        if volume_ref['host'] != self.host:
-            raise exception.Error(_("Volume is not local to this node"))
-
+        LOG.debug(_("volume %s: removing export"), volume_ref['name'])
+        self.driver.remove_export(context, volume_ref)
         try:
-            LOG.debug(_("volume %s: removing export"), volume_ref['name'])
-            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:
+        # TODO(vish): This may not be the best way to handle a busy delete
+        #             but I'm leaving it because this is the current way
+        #             it is handled.
+        except exception.VolumeIsBusy:
             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'],
-                                  {'status': 'error_deleting'})
-            raise
-
-        self.db.volume_destroy(context, volume_id)
+            volume_ref['status'] = 'available'
+            self.db.volume_update(context, volume_ref['id'], volume_ref)
+            yield volume_ref
+            return
         LOG.debug(_("volume %s: deleted successfully"), volume_ref['name'])
-        return True
+        self.db.volume_destroy(context, volume_ref['id'])
+        yield volume_ref
 
     def create_snapshot(self, context, volume_id, snapshot_id):
         """Creates and exports the snapshot."""