diff --git a/nova/compute/api.py b/nova/compute/api.py
index 1969f59e7375..0d4ff573dcc5 100644
--- a/nova/compute/api.py
+++ b/nova/compute/api.py
@@ -4071,9 +4071,31 @@ class API(base.Base):
         return objects.Migration.get_by_id_and_instance(
                 context, migration_id, instance_uuid)
 
+    def _get_bdm_by_volume_id(self, context, volume_id, expected_attrs=None):
+        """Retrieve a BDM without knowing its cell.
+
+        .. note:: The context will be targeted to the cell in which the
+            BDM is found, if any.
+
+        :param context: The API request context.
+        :param volume_id: The ID of the volume.
+        :param expected_attrs: list of any additional attributes that should
+            be joined when the BDM is loaded from the database.
+        :raises: nova.exception.VolumeBDMNotFound if not found in any cell
+        """
+        load_cells()
+        for cell in CELLS:
+            nova_context.set_target_cell(context, cell)
+            try:
+                return objects.BlockDeviceMapping.get_by_volume(
+                    context, volume_id, expected_attrs=expected_attrs)
+            except exception.NotFound:
+                continue
+        raise exception.VolumeBDMNotFound(volume_id=volume_id)
+
     def volume_snapshot_create(self, context, volume_id, create_info):
-        bdm = objects.BlockDeviceMapping.get_by_volume(
-                context, volume_id, expected_attrs=['instance'])
+        bdm = self._get_bdm_by_volume_id(
+            context, volume_id, expected_attrs=['instance'])
 
         # We allow creating the snapshot in any vm_state as long as there is
         # no task being performed on the instance and it has a host.
@@ -4094,8 +4116,8 @@ class API(base.Base):
 
     def volume_snapshot_delete(self, context, volume_id, snapshot_id,
                                delete_info):
-        bdm = objects.BlockDeviceMapping.get_by_volume(
-                context, volume_id, expected_attrs=['instance'])
+        bdm = self._get_bdm_by_volume_id(
+            context, volume_id, expected_attrs=['instance'])
 
         # We allow deleting the snapshot in any vm_state as long as there is
         # no task being performed on the instance and it has a host.
diff --git a/nova/tests/unit/compute/test_compute_api.py b/nova/tests/unit/compute/test_compute_api.py
index 732670aae862..a5e887a91cfd 100644
--- a/nova/tests/unit/compute/test_compute_api.py
+++ b/nova/tests/unit/compute/test_compute_api.py
@@ -2789,6 +2789,46 @@ class _ComputeAPIUnitTestMixIn(object):
         self._test_snapshot_volume_backed(False, True,
                                           vm_state=vm_states.SUSPENDED)
 
+    @mock.patch.object(context, 'set_target_cell')
+    @mock.patch.object(objects.BlockDeviceMapping, 'get_by_volume')
+    def test_get_bdm_by_volume_id(self, mock_get_by_volume,
+                                  mock_target_cell):
+        fake_cells = [mock.sentinel.cell0, mock.sentinel.cell1]
+
+        mock_get_by_volume.side_effect = [
+            exception.VolumeBDMNotFound(volume_id=mock.sentinel.volume_id),
+            mock.sentinel.bdm]
+
+        with mock.patch.object(compute_api, 'CELLS', fake_cells):
+            bdm = self.compute_api._get_bdm_by_volume_id(
+                self.context, mock.sentinel.volume_id,
+                mock.sentinel.expected_attrs)
+
+        self.assertEqual(mock.sentinel.bdm, bdm)
+
+        mock_target_cell.assert_has_calls([
+            mock.call(self.context, cell) for cell in fake_cells])
+        mock_get_by_volume.assert_has_calls(
+            [mock.call(self.context,
+                       mock.sentinel.volume_id,
+                       expected_attrs=mock.sentinel.expected_attrs)] * 2)
+
+    @mock.patch.object(context, 'set_target_cell')
+    @mock.patch.object(objects.BlockDeviceMapping, 'get_by_volume')
+    def test_get_missing_bdm_by_volume_id(self, mock_get_by_volume,
+                                          mock_target_cell):
+        fake_cells = [mock.sentinel.cell0, mock.sentinel.cell1]
+
+        mock_get_by_volume.side_effect = exception.VolumeBDMNotFound(
+            volume_id=mock.sentinel.volume_id)
+
+        with mock.patch.object(compute_api, 'CELLS', fake_cells):
+            self.assertRaises(
+                exception.VolumeBDMNotFound,
+                self.compute_api._get_bdm_by_volume_id,
+                self.context, mock.sentinel.volume_id,
+                mock.sentinel.expected_attrs)
+
     def test_volume_snapshot_create(self):
         volume_id = '1'
         create_info = {'id': 'eyedee'}
@@ -2808,12 +2848,12 @@ class _ComputeAPIUnitTestMixIn(object):
                 self.context, objects.BlockDeviceMapping(),
                 fake_bdm, expected_attrs=['instance'])
 
-        self.mox.StubOutWithMock(objects.BlockDeviceMapping,
-                                 'get_by_volume')
+        self.mox.StubOutWithMock(self.compute_api,
+                                 '_get_bdm_by_volume_id')
         self.mox.StubOutWithMock(self.compute_api.compute_rpcapi,
                 'volume_snapshot_create')
 
-        objects.BlockDeviceMapping.get_by_volume(
+        self.compute_api._get_bdm_by_volume_id(
                 self.context, volume_id,
                 expected_attrs=['instance']).AndReturn(fake_bdm)
         self.compute_api.compute_rpcapi.volume_snapshot_create(self.context,
@@ -2883,12 +2923,12 @@ class _ComputeAPIUnitTestMixIn(object):
                 self.context, objects.BlockDeviceMapping(),
                 fake_bdm, expected_attrs=['instance'])
 
-        self.mox.StubOutWithMock(objects.BlockDeviceMapping,
-                                 'get_by_volume')
+        self.mox.StubOutWithMock(self.compute_api,
+                                 '_get_bdm_by_volume_id')
         self.mox.StubOutWithMock(self.compute_api.compute_rpcapi,
                 'volume_snapshot_delete')
 
-        objects.BlockDeviceMapping.get_by_volume(
+        self.compute_api._get_bdm_by_volume_id(
                 self.context, volume_id,
                 expected_attrs=['instance']).AndReturn(fake_bdm)
         self.compute_api.compute_rpcapi.volume_snapshot_delete(self.context,