merged trunk

6 files changed, 199 insertions, 148 deletions

diff --git a/nova/auth/manager.py b/nova/auth/manager.py
index d44ed52b2..6d71a7ad6 100644
--- a/nova/auth/manager.py
+++ b/nova/auth/manager.py
@@ -29,15 +29,13 @@ import uuid
 import zipfile
 
 from nova import crypto
-from nova import datastore
 from nova import exception
 from nova import flags
-from nova import objectstore # for flags
 from nova import utils
-from nova.auth import ldapdriver # for flags
 from nova.auth import signer
 from nova.network import vpn
 
+
 FLAGS = flags.FLAGS
 
 # NOTE(vish): a user with one of these roles will be a superuser and
@@ -99,6 +97,7 @@ class AuthBase(object):
 class User(AuthBase):
     """Object representing a user"""
     def __init__(self, id, name, access, secret, admin):
+        AuthBase.__init__(self)
         self.id = id
         self.name = name
         self.access = access
@@ -159,6 +158,7 @@ class KeyPair(AuthBase):
     fingerprint is stored. The user's private key is not saved.
     """
     def __init__(self, id, name, owner_id, public_key, fingerprint):
+        AuthBase.__init__(self)
         self.id = id
         self.name = name
         self.owner_id = owner_id
@@ -176,6 +176,7 @@ class KeyPair(AuthBase):
 class Project(AuthBase):
     """Represents a Project returned from the datastore"""
     def __init__(self, id, name, project_manager_id, description, member_ids):
+        AuthBase.__init__(self)
         self.id = id
         self.name = name
         self.project_manager_id = project_manager_id
@@ -234,7 +235,7 @@ class AuthManager(object):
     AuthManager also manages associated data related to Auth objects that
     need to be more accessible, such as vpn ips and ports.
     """
-    _instance=None
+    _instance = None
     def __new__(cls, *args, **kwargs):
         """Returns the AuthManager singleton"""
         if not cls._instance:
@@ -248,7 +249,7 @@ class AuthManager(object):
         reset the driver if it is not set or a new driver is specified.
         """
         if driver or not getattr(self, 'driver', None):
-           self.driver = utils.import_class(driver or FLAGS.auth_driver)
+            self.driver = utils.import_class(driver or FLAGS.auth_driver)
 
     def authenticate(self, access, signature, params, verb='GET',
                      server_string='127.0.0.1:8773', path='/',
diff --git a/nova/endpoint/cloud.py b/nova/endpoint/cloud.py
index 878d54a15..ad9188ff3 100644
--- a/nova/endpoint/cloud.py
+++ b/nova/endpoint/cloud.py
@@ -47,6 +47,7 @@ FLAGS = flags.FLAGS
 
 flags.DEFINE_string('cloud_topic', 'cloud', 'the topic clouds listen on')
 
+
 def _gen_key(user_id, key_name):
     """ Tuck this into AuthManager """
     try:
@@ -102,15 +103,16 @@ class CloudController(object):
         result = {}
         for instance in self.instdir.all:
             if instance['project_id'] == project_id:
-                line = '%s slots=%d' % (instance['private_dns_name'], INSTANCE_TYPES[instance['instance_type']]['vcpus'])
+                line = '%s slots=%d' % (instance['private_dns_name'], 
+                    INSTANCE_TYPES[instance['instance_type']]['vcpus'])
                 if instance['key_name'] in result:
                     result[instance['key_name']].append(line)
                 else:
                     result[instance['key_name']] = [line]
         return result
 
-    def get_metadata(self, ip):
-        i = self.get_instance_by_ip(ip)
+    def get_metadata(self, ipaddress):
+        i = self.get_instance_by_ip(ipaddress)
         if i is None:
             return None
         mpi = self._get_mpi_data(i['project_id'])
@@ -147,7 +149,7 @@ class CloudController(object):
                 },
                 'public-hostname': i.get('dns_name', ''),
                 'public-ipv4': i.get('dns_name', ''), # TODO: switch to IP
-                'public-keys' : keys,
+                'public-keys': keys,
                 'ramdisk-id': i.get('ramdisk_id', ''),
                 'reservation-id': i['reservation_id'],
                 'security-groups': i.get('groups', ''),
@@ -203,26 +205,22 @@ class CloudController(object):
                     'keyFingerprint': key_pair.fingerprint,
                 })
 
-        return { 'keypairsSet': result }
+        return {'keypairsSet': result}
 
     @rbac.allow('all')
     def create_key_pair(self, context, key_name, **kwargs):
-        try:
-            d = defer.Deferred()
-            p = context.handler.application.settings.get('pool')
-            def _complete(kwargs):
-                if 'exception' in kwargs:
-                    d.errback(kwargs['exception'])
-                    return
-                d.callback({'keyName': key_name,
-                    'keyFingerprint': kwargs['fingerprint'],
-                    'keyMaterial': kwargs['private_key']})
-            p.apply_async(_gen_key, [context.user.id, key_name],
-                callback=_complete)
-            return d
-
-        except manager.UserError as e:
-            raise
+        dcall = defer.Deferred()
+        pool = context.handler.application.settings.get('pool')
+        def _complete(kwargs):
+            if 'exception' in kwargs:
+                dcall.errback(kwargs['exception'])
+                return
+            dcall.callback({'keyName': key_name,
+                'keyFingerprint': kwargs['fingerprint'],
+                'keyMaterial': kwargs['private_key']})
+        pool.apply_async(_gen_key, [context.user.id, key_name],
+            callback=_complete)
+        return dcall
 
     @rbac.allow('all')
     def delete_key_pair(self, context, key_name, **kwargs):
@@ -232,7 +230,7 @@ class CloudController(object):
 
     @rbac.allow('all')
     def describe_security_groups(self, context, group_names, **kwargs):
-        groups = { 'securityGroupSet': [] }
+        groups = {'securityGroupSet': []}
 
         # Stubbed for now to unblock other things.
         return groups
@@ -251,7 +249,7 @@ class CloudController(object):
         instance = self._get_instance(context, instance_id[0])
         return rpc.call('%s.%s' % (FLAGS.compute_topic, instance['node_name']),
             {"method": "get_console_output",
-             "args" : {"instance_id": instance_id[0]}})
+             "args": {"instance_id": instance_id[0]}})
 
     def _get_user_id(self, context):
         if context and context.user:
@@ -285,10 +283,10 @@ class CloudController(object):
         if volume['attach_status'] == 'attached':
             v['attachmentSet'] = [{'attachTime': volume['attach_time'],
                                    'deleteOnTermination': volume['delete_on_termination'],
-                                   'device' : volume['mountpoint'],
-                                   'instanceId' : volume['instance_id'],
-                                   'status' : 'attached',
-                                   'volume_id' : volume['volume_id']}]
+                                   'device': volume['mountpoint'],
+                                   'instanceId': volume['instance_id'],
+                                   'status': 'attached',
+                                   'volume_id': volume['volume_id']}]
         else:
             v['attachmentSet'] = [{}]
         return v
@@ -298,7 +296,7 @@ class CloudController(object):
     def create_volume(self, context, size, **kwargs):
         # TODO(vish): refactor this to create the volume object here and tell service to create it
         result = yield rpc.call(FLAGS.volume_topic, {"method": "create_volume",
-                                 "args" : {"size": size,
+                                 "args": {"size": size,
                                            "user_id": context.user.id,
                                            "project_id": context.project.id}})
         # NOTE(vish): rpc returned value is in the result key in the dictionary
@@ -348,15 +346,15 @@ class CloudController(object):
         compute_node = instance['node_name']
         rpc.cast('%s.%s' % (FLAGS.compute_topic, compute_node),
                                 {"method": "attach_volume",
-                                 "args" : {"volume_id": volume_id,
-                                           "instance_id" : instance_id,
-                                           "mountpoint" : device}})
-        return defer.succeed({'attachTime' : volume['attach_time'],
-                              'device' : volume['mountpoint'],
-                              'instanceId' : instance_id,
-                              'requestId' : context.request_id,
-                              'status' : volume['attach_status'],
-                              'volumeId' : volume_id})
+                                 "args": {"volume_id": volume_id,
+                                           "instance_id": instance_id,
+                                           "mountpoint": device}})
+        return defer.succeed({'attachTime': volume['attach_time'],
+                              'device': volume['mountpoint'],
+                              'instanceId': instance_id,
+                              'requestId': context.request_id,
+                              'status': volume['attach_status'],
+                              'volumeId': volume_id})
 
 
     @rbac.allow('projectmanager', 'sysadmin')
@@ -372,18 +370,18 @@ class CloudController(object):
             instance = self._get_instance(context, instance_id)
             rpc.cast('%s.%s' % (FLAGS.compute_topic, instance['node_name']),
                                 {"method": "detach_volume",
-                                 "args" : {"instance_id": instance_id,
+                                 "args": {"instance_id": instance_id,
                                            "volume_id": volume_id}})
         except exception.NotFound:
             # If the instance doesn't exist anymore,
             # then we need to call detach blind
             volume.finish_detach()
-        return defer.succeed({'attachTime' : volume['attach_time'],
-                              'device' : volume['mountpoint'],
-                              'instanceId' : instance_id,
-                              'requestId' : context.request_id,
-                              'status' : volume['attach_status'],
-                              'volumeId' : volume_id})
+        return defer.succeed({'attachTime': volume['attach_time'],
+                              'device': volume['mountpoint'],
+                              'instanceId': instance_id,
+                              'requestId': context.request_id,
+                              'status': volume['attach_status'],
+                              'volumeId': volume_id})
 
     def _convert_to_set(self, lst, label):
         if lst == None or lst == []:
@@ -425,7 +423,8 @@ class CloudController(object):
             i['key_name'] = instance.get('key_name', None)
             if context.user.is_admin():
                 i['key_name'] = '%s (%s, %s)' % (i['key_name'],
-                    instance.get('project_id', None), instance.get('node_name',''))
+                    instance.get('project_id', None), 
+                    instance.get('node_name', ''))
             i['product_codes_set'] = self._convert_to_set(
                 instance.get('product_codes', None), 'product_code')
             i['instance_type'] = instance.get('instance_type', None)
@@ -442,7 +441,7 @@ class CloudController(object):
                 reservations[res_id] = r
             reservations[res_id]['instances_set'].append(i)
 
-        instance_response = {'reservationSet' : list(reservations.values()) }
+        instance_response = {'reservationSet': list(reservations.values())}
         return instance_response
 
     @rbac.allow('all')
@@ -457,7 +456,7 @@ class CloudController(object):
                 address['project_id'] == context.project.id):
                 address_rv = {
                     'public_ip': address['address'],
-                    'instance_id' : address.get('instance_id', 'free')
+                    'instance_id': address.get('instance_id', 'free')
                 }
                 if context.user.is_admin():
                     address_rv['instance_id'] = "%s (%s, %s)" % (
@@ -477,7 +476,7 @@ class CloudController(object):
                           "args": {"user_id": context.user.id,
                                    "project_id": context.project.id}})
         public_ip = alloc_result['result']
-        defer.returnValue({'addressSet': [{'publicIp' : public_ip}]})
+        defer.returnValue({'addressSet': [{'publicIp': public_ip}]})
 
     @rbac.allow('netadmin')
     @defer.inlineCallbacks
@@ -591,7 +590,7 @@ class CloudController(object):
             inst.save()
             rpc.cast(FLAGS.compute_topic,
                  {"method": "run_instance",
-                  "args": {"instance_id" : inst.instance_id}})
+                  "args": {"instance_id": inst.instance_id}})
             logging.debug("Casting to node for %s's instance with IP of %s" %
                       (context.user.name, inst['private_dns_name']))
         # TODO: Make Network figure out the network name from ip.
@@ -646,7 +645,7 @@ class CloudController(object):
             instance = self._get_instance(context, i)
             rpc.cast('%s.%s' % (FLAGS.compute_topic, instance['node_name']),
                              {"method": "reboot_instance",
-                              "args" : {"instance_id": i}})
+                              "args": {"instance_id": i}})
         return defer.succeed(True)
 
     @rbac.allow('projectmanager', 'sysadmin')
@@ -656,7 +655,7 @@ class CloudController(object):
         volume_node = volume['node_name']
         rpc.cast('%s.%s' % (FLAGS.volume_topic, volume_node),
                             {"method": "delete_volume",
-                             "args" : {"volume_id": volume_id}})
+                             "args": {"volume_id": volume_id}})
         return defer.succeed(True)
 
     @rbac.allow('all')
@@ -689,9 +688,9 @@ class CloudController(object):
             image = images.list(context, image_id)[0]
         except IndexError:
             raise exception.ApiError('invalid id: %s' % image_id)
-        result = { 'image_id': image_id, 'launchPermission': [] }
+        result = {'image_id': image_id, 'launchPermission': []}
         if image['isPublic']:
-            result['launchPermission'].append({ 'group': 'all' })
+            result['launchPermission'].append({'group': 'all'})
         return defer.succeed(result)
 
     @rbac.allow('projectmanager', 'sysadmin')
diff --git a/nova/network/model.py b/nova/network/model.py
index 1de5d17d3..2074a6d46 100644
--- a/nova/network/model.py
+++ b/nova/network/model.py
@@ -97,11 +97,11 @@ class Vlan(datastore.BasicModel):
     def dict_by_vlan(cls):
         """a hash of vlan:project"""
         set_name = cls._redis_set_name(cls.__name__)
-        rv = {}
-        h = datastore.Redis.instance().hgetall(set_name)
-        for v in h.keys():
-            rv[h[v]] = v
-        return rv
+        retvals = {}
+        hashset = datastore.Redis.instance().hgetall(set_name)
+        for val in hashset.keys():
+            retvals[hashset[val]] = val
+        return retvals
 
     @classmethod
     @datastore.absorb_connection_error
@@ -136,7 +136,8 @@ class Vlan(datastore.BasicModel):
 
 # CLEANUP:
 # TODO(ja): Save the IPs at the top of each subnet for cloudpipe vpn clients
-# TODO(ja): does vlanpool "keeper" need to know the min/max - shouldn't FLAGS always win?
+# TODO(ja): does vlanpool "keeper" need to know the min/max -
+#           shouldn't FLAGS always win?
 # TODO(joshua): Save the IPs at the top of each subnet for cloudpipe vpn clients
 
 class BaseNetwork(datastore.BasicModel):
@@ -352,8 +353,9 @@ class DHCPNetwork(BridgedNetwork):
         private_ip = str(self.network[2])
         linux_net.confirm_rule("FORWARD -d %s -p udp --dport 1194 -j ACCEPT"
                                % (private_ip, ))
-        linux_net.confirm_rule("PREROUTING -t nat -d %s -p udp --dport %s -j DNAT --to %s:1194"
-                               % (self.project.vpn_ip, self.project.vpn_port, private_ip))
+        linux_net.confirm_rule(
+            "PREROUTING -t nat -d %s -p udp --dport %s -j DNAT --to %s:1194"
+            % (self.project.vpn_ip, self.project.vpn_port, private_ip))
 
     def deexpress(self, address=None):
         # if this is the last address, stop dns
@@ -388,13 +390,14 @@ class PublicAddress(datastore.BasicModel):
         return addr
 
 
-DEFAULT_PORTS = [("tcp",80), ("tcp",22), ("udp",1194), ("tcp",443)]
+DEFAULT_PORTS = [("tcp", 80), ("tcp", 22), ("udp", 1194), ("tcp", 443)]
 class PublicNetworkController(BaseNetwork):
     override_type = 'network'
 
     def __init__(self, *args, **kwargs):
         network_id = "public:default"
-        super(PublicNetworkController, self).__init__(network_id, FLAGS.public_range)
+        super(PublicNetworkController, self).__init__(network_id,
+            FLAGS.public_range)
         self['user_id'] = "public"
         self['project_id'] = "public"
         self["create_time"] = time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime())
@@ -468,8 +471,9 @@ class PublicNetworkController(BaseNetwork):
             linux_net.confirm_rule("FORWARD -d %s -p icmp -j ACCEPT"
                                    % (private_ip))
             for (protocol, port) in DEFAULT_PORTS:
-                linux_net.confirm_rule("FORWARD -d %s -p %s --dport %s -j ACCEPT"
-                                       % (private_ip, protocol, port))
+                linux_net.confirm_rule(
+                    "FORWARD -d %s -p %s --dport %s -j ACCEPT"
+                    % (private_ip, protocol, port))
 
     def deexpress(self, address=None):
         addr = self.get_host(address)
diff --git a/nova/tests/volume_unittest.py b/nova/tests/volume_unittest.py
index 0f4f0e34d..2a07afe69 100644
--- a/nova/tests/volume_unittest.py
+++ b/nova/tests/volume_unittest.py
@@ -17,6 +17,10 @@
 #    under the License.
 
 import logging
+import shutil
+import tempfile
+
+from twisted.internet import defer
 
 from nova import compute
 from nova import exception
@@ -34,10 +38,16 @@ class VolumeTestCase(test.TrialTestCase):
         super(VolumeTestCase, self).setUp()
         self.compute = compute.service.ComputeService()
         self.volume = None
+        self.tempdir = tempfile.mkdtemp()
         self.flags(connection_type='fake',
-                   fake_storage=True)
+                   fake_storage=True,
+                   aoe_export_dir=self.tempdir)
         self.volume = volume_service.VolumeService()
 
+    def tearDown(self):
+        shutil.rmtree(self.tempdir)
+
+    @defer.inlineCallbacks
     def test_run_create_volume(self):
         vol_size = '0'
         user_id = 'fake'
@@ -48,34 +58,40 @@ class VolumeTestCase(test.TrialTestCase):
                          volume_service.get_volume(volume_id)['volume_id'])
 
         rv = self.volume.delete_volume(volume_id)
-        self.assertFailure(volume_service.get_volume(volume_id),
-                           exception.Error)
+        self.assertRaises(exception.Error, volume_service.get_volume, volume_id)
 
+    @defer.inlineCallbacks
     def test_too_big_volume(self):
         vol_size = '1001'
         user_id = 'fake'
         project_id = 'fake'
-        self.assertRaises(TypeError,
-                          self.volume.create_volume,
-                          vol_size, user_id, project_id)
+        try:
+            yield self.volume.create_volume(vol_size, user_id, project_id)
+            self.fail("Should have thrown TypeError")
+        except TypeError:
+            pass
 
+    @defer.inlineCallbacks
     def test_too_many_volumes(self):
         vol_size = '1'
         user_id = 'fake'
         project_id = 'fake'
         num_shelves = FLAGS.last_shelf_id - FLAGS.first_shelf_id + 1
-        total_slots = FLAGS.slots_per_shelf * num_shelves
+        total_slots = FLAGS.blades_per_shelf * num_shelves
         vols = []
+        from nova import datastore
+        redis = datastore.Redis.instance()
         for i in xrange(total_slots):
             vid = yield self.volume.create_volume(vol_size, user_id, project_id)
             vols.append(vid)
         self.assertFailure(self.volume.create_volume(vol_size,
                                                      user_id,
                                                      project_id),
-                           volume_service.NoMoreVolumes)
+                           volume_service.NoMoreBlades)
         for id in vols:
             yield self.volume.delete_volume(id)
 
+    @defer.inlineCallbacks
     def test_run_attach_detach_volume(self):
         # Create one volume and one compute to test with
         instance_id = "storage-test"
@@ -84,22 +100,26 @@ class VolumeTestCase(test.TrialTestCase):
         project_id = 'fake'
         mountpoint = "/dev/sdf"
         volume_id = yield self.volume.create_volume(vol_size, user_id, project_id)
-
         volume_obj = volume_service.get_volume(volume_id)
         volume_obj.start_attach(instance_id, mountpoint)
-        rv = yield self.compute.attach_volume(volume_id,
-                                          instance_id,
-                                          mountpoint)
+        if FLAGS.fake_tests:
+            volume_obj.finish_attach()
+        else:
+            rv = yield self.compute.attach_volume(instance_id,
+                                                  volume_id,
+                                                  mountpoint)
         self.assertEqual(volume_obj['status'], "in-use")
-        self.assertEqual(volume_obj['attachStatus'], "attached")
+        self.assertEqual(volume_obj['attach_status'], "attached")
         self.assertEqual(volume_obj['instance_id'], instance_id)
         self.assertEqual(volume_obj['mountpoint'], mountpoint)
 
-        self.assertRaises(exception.Error,
-                          self.volume.delete_volume,
-                          volume_id)
-
-        rv = yield self.volume.detach_volume(volume_id)
+        self.assertFailure(self.volume.delete_volume(volume_id), exception.Error)
+        volume_obj.start_detach()
+        if FLAGS.fake_tests:
+            volume_obj.finish_detach()
+        else:
+            rv = yield self.volume.detach_volume(instance_id,
+                                                 volume_id)
         volume_obj = volume_service.get_volume(volume_id)
         self.assertEqual(volume_obj['status'], "available")
 
@@ -108,6 +128,27 @@ class VolumeTestCase(test.TrialTestCase):
                           volume_service.get_volume,
                           volume_id)
 
+    @defer.inlineCallbacks
+    def test_multiple_volume_race_condition(self):
+        vol_size = "5"
+        user_id = "fake"
+        project_id = 'fake'
+        shelf_blades = []
+        def _check(volume_id):
+            vol = volume_service.get_volume(volume_id)
+            shelf_blade = '%s.%s' % (vol['shelf_id'], vol['blade_id'])
+            self.assert_(shelf_blade not in shelf_blades)
+            shelf_blades.append(shelf_blade)
+            logging.debug("got %s" % shelf_blade)
+            vol.destroy()
+        deferreds = []
+        for i in range(5):
+            d = self.volume.create_volume(vol_size, user_id, project_id)
+            d.addCallback(_check)
+            d.addErrback(self.fail)
+            deferreds.append(d)
+        yield defer.DeferredList(deferreds)
+
     def test_multi_node(self):
         # TODO(termie): Figure out how to test with two nodes,
         # each of them having a different FLAG for storage_node
diff --git a/nova/virt/libvirt_conn.py b/nova/virt/libvirt_conn.py
index 551ba6e54..13305be0f 100644
--- a/nova/virt/libvirt_conn.py
+++ b/nova/virt/libvirt_conn.py
@@ -114,7 +114,8 @@ class LibvirtConnection(object):
     def _cleanup(self, instance):
         target = os.path.abspath(instance.datamodel['basepath'])
         logging.info("Deleting instance files at %s", target)
-        shutil.rmtree(target)
+        if os.path.exists(target):
+            shutil.rmtree(target)
 
 
     @defer.inlineCallbacks
diff --git a/nova/volume/service.py b/nova/volume/service.py
index e12f675a7..66163a812 100644
--- a/nova/volume/service.py
+++ b/nova/volume/service.py
@@ -22,12 +22,8 @@ destroying persistent storage volumes, ala EBS.
 Currently uses Ata-over-Ethernet.
 """
 
-import glob
 import logging
 import os
-import shutil
-import socket
-import tempfile
 
 from twisted.internet import defer
 
@@ -47,9 +43,6 @@ flags.DEFINE_string('volume_group', 'nova-volumes',
                     'Name for the VG that will contain exported volumes')
 flags.DEFINE_string('aoe_eth_dev', 'eth0',
                     'Which device to export the volumes on')
-flags.DEFINE_string('storage_name',
-                    socket.gethostname(),
-                    'name of this service')
 flags.DEFINE_integer('first_shelf_id',
                     utils.last_octet(utils.get_my_ip()) * 10,
                     'AoE starting shelf_id for this service')
@@ -59,9 +52,9 @@ flags.DEFINE_integer('last_shelf_id',
 flags.DEFINE_string('aoe_export_dir',
                     '/var/lib/vblade-persist/vblades',
                     'AoE directory where exports are created')
-flags.DEFINE_integer('slots_per_shelf',
+flags.DEFINE_integer('blades_per_shelf',
                     16,
-                    'Number of AoE slots per shelf')
+                    'Number of AoE blades per shelf')
 flags.DEFINE_string('storage_availability_zone',
                     'nova',
                     'availability zone of this service')
@@ -69,7 +62,7 @@ flags.DEFINE_boolean('fake_storage', False,
                      'Should we make real storage volumes to attach?')
 
 
-class NoMoreVolumes(exception.Error):
+class NoMoreBlades(exception.Error):
     pass
 
 def get_volume(volume_id):
@@ -77,8 +70,9 @@ def get_volume(volume_id):
     volume_class = Volume
     if FLAGS.fake_storage:
         volume_class = FakeVolume
-    if datastore.Redis.instance().sismember('volumes', volume_id):
-        return volume_class(volume_id=volume_id)
+    vol = volume_class.lookup(volume_id)
+    if vol:
+        return vol
     raise exception.Error("Volume does not exist")
 
 class VolumeService(service.Service):
@@ -91,18 +85,9 @@ class VolumeService(service.Service):
         super(VolumeService, self).__init__()
         self.volume_class = Volume
         if FLAGS.fake_storage:
-            FLAGS.aoe_export_dir = tempfile.mkdtemp()
             self.volume_class = FakeVolume
         self._init_volume_group()
 
-    def __del__(self):
-        # TODO(josh): Get rid of this destructor, volumes destroy themselves
-        if FLAGS.fake_storage:
-            try:
-                shutil.rmtree(FLAGS.aoe_export_dir)
-            except Exception, err:
-                pass
-
     @defer.inlineCallbacks
     @validate.rangetest(size=(0, 1000))
     def create_volume(self, size, user_id, project_id):
@@ -113,8 +98,6 @@ class VolumeService(service.Service):
         """
         logging.debug("Creating volume of size: %s" % (size))
         vol = yield self.volume_class.create(size, user_id, project_id)
-        datastore.Redis.instance().sadd('volumes', vol['volume_id'])
-        datastore.Redis.instance().sadd('volumes:%s' % (FLAGS.storage_name), vol['volume_id'])
         logging.debug("restarting exports")
         yield self._restart_exports()
         defer.returnValue(vol['volume_id'])
@@ -134,21 +117,19 @@ class VolumeService(service.Service):
     def delete_volume(self, volume_id):
         logging.debug("Deleting volume with id of: %s" % (volume_id))
         vol = get_volume(volume_id)
-        if vol['status'] == "attached":
+        if vol['attach_status'] == "attached":
             raise exception.Error("Volume is still attached")
-        if vol['node_name'] != FLAGS.storage_name:
+        if vol['node_name'] != FLAGS.node_name:
             raise exception.Error("Volume is not local to this node")
         yield vol.destroy()
-        datastore.Redis.instance().srem('volumes', vol['volume_id'])
-        datastore.Redis.instance().srem('volumes:%s' % (FLAGS.storage_name), vol['volume_id'])
         defer.returnValue(True)
 
     @defer.inlineCallbacks
     def _restart_exports(self):
         if FLAGS.fake_storage:
             return
-        yield process.simple_execute("sudo vblade-persist auto all")
-        # NOTE(vish): this command sometimes sends output to stderr for warnings
+        # NOTE(vish): these commands sometimes sends output to stderr for warnings
+        yield process.simple_execute("sudo vblade-persist auto all", error_ok=1)
         yield process.simple_execute("sudo vblade-persist start all", error_ok=1)
 
     @defer.inlineCallbacks
@@ -172,14 +153,15 @@ class Volume(datastore.BasicModel):
         return self.volume_id
 
     def default_state(self):
-        return {"volume_id": self.volume_id}
+        return {"volume_id": self.volume_id,
+                "node_name": "unassigned"}
 
     @classmethod
     @defer.inlineCallbacks
     def create(cls, size, user_id, project_id):
         volume_id = utils.generate_uid('vol')
         vol = cls(volume_id)
-        vol['node_name'] = FLAGS.storage_name
+        vol['node_name'] = FLAGS.node_name
         vol['size'] = size
         vol['user_id'] = user_id
         vol['project_id'] = project_id
@@ -225,14 +207,31 @@ class Volume(datastore.BasicModel):
         self['attach_status'] = "detached"
         self.save()
 
+    def save(self):
+        is_new = self.is_new_record()
+        super(Volume, self).save()
+        if is_new:
+            redis = datastore.Redis.instance()
+            key = self.__devices_key
+            # TODO(vish): these should be added by admin commands
+            more = redis.scard(self._redis_association_name("node",
+                                                            self['node_name']))
+            if (not redis.exists(key) and not more):
+                for shelf_id in range(FLAGS.first_shelf_id,
+                                      FLAGS.last_shelf_id + 1):
+                    for blade_id in range(FLAGS.blades_per_shelf):
+                        redis.sadd(key, "%s.%s" % (shelf_id, blade_id))
+            self.associate_with("node", self['node_name'])
+
     @defer.inlineCallbacks
     def destroy(self):
-        try:
-            yield self._remove_export()
-        except Exception as ex:
-            logging.debug("Ingnoring failure to remove export %s" % ex)
-            pass
+        yield self._remove_export()
         yield self._delete_lv()
+        self.unassociate_with("node", self['node_name'])
+        if self.get('shelf_id', None) and self.get('blade_id', None):
+            redis = datastore.Redis.instance()
+            key = self.__devices_key
+            redis.sadd(key, "%s.%s" % (self['shelf_id'], self['blade_id']))
         super(Volume, self).destroy()
 
     @defer.inlineCallbacks
@@ -244,66 +243,72 @@ class Volume(datastore.BasicModel):
         yield process.simple_execute(
                 "sudo lvcreate -L %s -n %s %s" % (sizestr,
                                                   self['volume_id'],
-                                                  FLAGS.volume_group))
+                                                  FLAGS.volume_group),
+                error_ok=1)
 
     @defer.inlineCallbacks
     def _delete_lv(self):
         yield process.simple_execute(
                 "sudo lvremove -f %s/%s" % (FLAGS.volume_group,
-                                            self['volume_id']))
+                                            self['volume_id']), error_ok=1)
+
+    @property
+    def __devices_key(self):
+        return 'volume_devices:%s' % FLAGS.node_name
 
     @defer.inlineCallbacks
     def _setup_export(self):
-        (shelf_id, blade_id) = get_next_aoe_numbers()
+        redis = datastore.Redis.instance()
+        key = self.__devices_key
+        device = redis.spop(key)
+        if not device:
+            raise NoMoreBlades()
+        (shelf_id, blade_id) = device.split('.')
         self['aoe_device'] = "e%s.%s" % (shelf_id, blade_id)
         self['shelf_id'] = shelf_id
         self['blade_id'] = blade_id
         self.save()
-        yield self._exec_export()
+        yield self._exec_setup_export()
 
     @defer.inlineCallbacks
-    def _exec_export(self):
+    def _exec_setup_export(self):
         yield process.simple_execute(
                 "sudo vblade-persist setup %s %s %s /dev/%s/%s" %
                 (self['shelf_id'],
                  self['blade_id'],
                  FLAGS.aoe_eth_dev,
                  FLAGS.volume_group,
-                 self['volume_id']))
+                 self['volume_id']), error_ok=1)
 
     @defer.inlineCallbacks
     def _remove_export(self):
+        if not self.get('shelf_id', None) or not self.get('blade_id', None):
+            defer.returnValue(False)
+        yield self._exec_remove_export()
+        defer.returnValue(True)
+
+    @defer.inlineCallbacks
+    def _exec_remove_export(self):
         yield process.simple_execute(
                 "sudo vblade-persist stop %s %s" % (self['shelf_id'],
-                                                    self['blade_id']))
+                                                    self['blade_id']), error_ok=1)
         yield process.simple_execute(
                 "sudo vblade-persist destroy %s %s" % (self['shelf_id'],
-                                                       self['blade_id']))
+                                                       self['blade_id']), error_ok=1)
+
 
 
 class FakeVolume(Volume):
     def _create_lv(self):
         pass
 
-    def _exec_export(self):
+    def _exec_setup_export(self):
         fname = os.path.join(FLAGS.aoe_export_dir, self['aoe_device'])
         f = file(fname, "w")
         f.close()
 
-    def _remove_export(self):
-        pass
+    def _exec_remove_export(self):
+        os.unlink(os.path.join(FLAGS.aoe_export_dir, self['aoe_device']))
 
     def _delete_lv(self):
         pass
-
-def get_next_aoe_numbers():
-    for shelf_id in xrange(FLAGS.first_shelf_id, FLAGS.last_shelf_id + 1):
-        aoes = glob.glob("%s/e%s.*" % (FLAGS.aoe_export_dir, shelf_id))
-        if not aoes:
-            blade_id = 0
-        else:
-            blade_id = int(max([int(a.rpartition('.')[2]) for a in aoes])) + 1
-        if blade_id < FLAGS.slots_per_shelf:
-            logging.debug("Next shelf.blade is %s.%s", shelf_id, blade_id)
-            return (shelf_id, blade_id)
-    raise NoMoreVolumes()