Prevent double-attachment race in attachment_reserve

If multiple attachments are requested simultaneously,
a volume can be attached to an instance more than once.

This fixes this problem by detecting that a race has
occurred in _attachment_reserve, and backing out one of
the invalid attachment records.

This means that if the attachment API is called
repeatedly and quickly for the same volume, some requests
may fail, but this is much better than incorrectly
creating multiple attachments.

Closes-Bug: #1833736

Change-Id: Ic2463338b698c5cf805c0ae06d0229f54f64b3fc
(cherry picked from commit 7f3a77b66f)
(cherry picked from commit eadff0df7e)
Conflicts:
	cinder/tests/unit/attachments/test_attachments_api.py
(cherry picked from commit 5f9c93b194)

cinder/cmd/api.py

@@ -35,6 +35,7 @@ i18n.enable_lazy()
 
 # Need to register global_opts
 from cinder.common import config
+from cinder import coordination
 from cinder import rpc
 from cinder import service
 from cinder import utils
@@ -56,6 +57,8 @@ def main():
 
     gmr.TextGuruMeditation.setup_autorun(version, conf=CONF)
 
+    coordination.COORDINATOR.start()
+
     rpc.init(CONF)
     launcher = service.process_launcher()
     server = service.WSGIService('osapi_volume')

cinder/tests/unit/api/v3/test_attachments.py

@@ -19,12 +19,16 @@ Tests for attachments Api.
 
 import ddt
 import mock
+from oslo_policy import policy as oslo_policy
 
 from cinder.api import microversions as mv
 from cinder.api.v3 import attachments as v3_attachments
 from cinder import context
 from cinder import exception
 from cinder import objects
+from cinder.policies import attachments as attachments_policies
+from cinder.policies import base as base_policy
+from cinder import policy
 from cinder import test
 from cinder.tests.unit.api import fakes
 from cinder.tests.unit import fake_constants as fake
@@ -48,13 +52,17 @@ class AttachmentsAPITestCase(test.TestCase):
         self.volume2 = self._create_volume(display_name='fake_volume_2',
                                            project_id=fake.PROJECT2_ID)
         self.attachment1 = self._create_attachment(
-            volume_uuid=self.volume1.id, instance_uuid=fake.UUID1)
+            volume_uuid=self.volume1.id, instance_uuid=fake.UUID1,
+            host='host-a')
         self.attachment2 = self._create_attachment(
-            volume_uuid=self.volume1.id, instance_uuid=fake.UUID1)
+            volume_uuid=self.volume1.id, instance_uuid=fake.UUID1,
+            host='host-b')
         self.attachment3 = self._create_attachment(
-            volume_uuid=self.volume1.id, instance_uuid=fake.UUID2)
+            volume_uuid=self.volume1.id, instance_uuid=fake.UUID2,
+            host='host-c')
         self.attachment4 = self._create_attachment(
-            volume_uuid=self.volume2.id, instance_uuid=fake.UUID2)
+            volume_uuid=self.volume2.id, instance_uuid=fake.UUID2,
+            host='host-d')
         self.addCleanup(self._cleanup)
 
     def _cleanup(self):
@@ -98,7 +106,8 @@ class AttachmentsAPITestCase(test.TestCase):
 
     @mock.patch.object(volume_rpcapi.VolumeAPI, 'attachment_update')
     def test_update_attachment(self, mock_update):
-        fake_connector = {'fake_key': 'fake_value'}
+        fake_connector = {'fake_key': 'fake_value',
+                          'host': 'somehost'}
         mock_update.return_value = fake_connector
         req = fakes.HTTPRequest.blank('/v3/%s/attachments/%s' %
                                       (fake.PROJECT_ID, self.attachment1.id),
@@ -107,7 +116,9 @@ class AttachmentsAPITestCase(test.TestCase):
         body = {
             "attachment":
                 {
-                    "connector": {'fake_key': 'fake_value'},
+                    "connector": {'fake_key': 'fake_value',
+                                  'host': 'somehost',
+                                  'connection_info': 'a'},
                 },
         }
 
@@ -133,19 +144,26 @@ class AttachmentsAPITestCase(test.TestCase):
                           self.controller.update, req,
                           self.attachment1.id, body=body)
 
+    @mock.patch('cinder.coordination.synchronized')
     @mock.patch.object(objects.VolumeAttachment, 'get_by_id')
-    def test_attachment_operations_not_authorized(self, mock_get):
+    def test_attachment_operations_not_authorized(self, mock_get, mock_synch):
-        mock_get.return_value = {'project_id': fake.PROJECT2_ID}
+        mock_get.return_value = self.attachment1
         req = fakes.HTTPRequest.blank('/v3/%s/attachments/%s' %
-                                      (fake.PROJECT_ID, self.attachment1.id),
+                                      (fake.PROJECT2_ID, self.attachment1.id),
                                       version=mv.NEW_ATTACH,
                                       use_admin_context=False)
         body = {
             "attachment":
                 {
-                    "connector": {'fake_key': 'fake_value'},
+                    "connector": {'fake_key': 'fake_value',
+                                  'host': 'somehost'},
                 },
         }
+        rules = {attachments_policies.UPDATE_POLICY:
+                 base_policy.RULE_ADMIN_OR_OWNER}
+        policy.set_rules(oslo_policy.Rules.from_dict(rules))
+        self.addCleanup(policy.reset)
+
         self.assertRaises(exception.NotAuthorized,
                           self.controller.update, req,
                           self.attachment1.id, body=body)
@@ -198,7 +216,7 @@ class AttachmentsAPITestCase(test.TestCase):
     def _create_attachment(self, ctxt=None, volume_uuid=None,
                            instance_uuid=None, mountpoint=None,
                            attach_time=None, detach_time=None,
-                           attach_status=None, attach_mode=None):
+                           attach_status=None, attach_mode=None, host=None):
         """Create an attachment object."""
         ctxt = ctxt or self.ctxt
         attachment = objects.VolumeAttachment(ctxt)
@@ -209,6 +227,7 @@ class AttachmentsAPITestCase(test.TestCase):
         attachment.detach_time = detach_time
         attachment.attach_status = attach_status or 'reserved'
         attachment.attach_mode = attach_mode
+        attachment.connector = {'host': host}
         attachment.create()
         return attachment
 

cinder/tests/unit/attachments/test_attachments_api.py

@@ -126,7 +126,8 @@ class AttachmentManagerTestCase(test.TestCase):
         vref = objects.Volume.get_by_id(self.context,
                                         vref.id)
 
-        connector = {'fake': 'connector'}
+        connector = {'fake': 'connector',
+                     'host': 'somehost'}
         self.volume_api.attachment_update(self.context,
                                           aref,
                                           connector)
@@ -308,9 +309,51 @@ class AttachmentManagerTestCase(test.TestCase):
         self.assertEqual({}, aref.connection_info)
         vref.status = 'error'
         vref.save()
-        connector = {'fake': 'connector'}
+        connector = {'fake': 'connector',
+                     'host': 'somehost'}
         self.assertRaises(exception.InvalidVolume,
                           self.volume_api.attachment_update,
                           self.context,
                           aref,
                           connector)
+
+    @mock.patch('cinder.db.sqlalchemy.api.volume_attachment_update',
+                return_value={})
+    @mock.patch('cinder.volume.rpcapi.VolumeAPI.attachment_update',
+                return_value={})
+    def test_attachment_update_duplicate(self, mock_va_update, mock_db_upd):
+        volume_params = {'status': 'available'}
+
+        vref = tests_utils.create_volume(self.context,
+                                         deleted=0,
+                                         **volume_params)
+
+        tests_utils.attach_volume(self.context,
+                                  vref.id,
+                                  fake.UUID1,
+                                  'somehost',
+                                  'somemountpoint')
+
+        # Update volume with another attachment
+        tests_utils.attach_volume(self.context,
+                                  vref.id,
+                                  fake.UUID2,
+                                  'somehost2',
+                                  'somemountpoint2')
+        vref.refresh()
+
+        # This attachment will collide with the first
+        connector = {'host': 'somehost'}
+        vref.volume_attachment[0]['connector'] = {'host': 'somehost'}
+        vref.volume_attachment[0]['connection_info'] = {'c': 'd'}
+        with mock.patch('cinder.objects.Volume.get_by_id', return_value=vref):
+            with mock.patch.object(self.volume_api.volume_rpcapi,
+                                   'attachment_update') as m_au:
+                self.assertRaises(exception.InvalidVolume,
+                                  self.volume_api.attachment_update,
+                                  self.context,
+                                  vref.volume_attachment[1],
+                                  connector)
+                m_au.assert_not_called()
+        mock_va_update.assert_not_called()
+        mock_db_upd.assert_not_called()

cinder/volume/api.py

@@ -32,6 +32,7 @@ import six
 from cinder.api import common
 from cinder.common import constants
 from cinder import context
+from cinder import coordination
 from cinder import db
 from cinder.db import base
 from cinder import exception
@@ -2144,6 +2145,8 @@ class API(base.Base):
         attachment_ref.save()
         return attachment_ref
 
+    @coordination.synchronized(
+        '{f_name}-{attachment_ref.volume_id}-{connector[host]}')
     def attachment_update(self, ctxt, attachment_ref, connector):
         """Update an existing attachment record."""
         # Valid items to update (connector includes mode and mountpoint):
@@ -2152,6 +2155,10 @@ class API(base.Base):
         #     b. mountpoint (if None use value from attachment_ref)
         #     c. instance_uuid(if None use value from attachment_ref)
 
+        # This method has a synchronized() lock on the volume id
+        # because we have to prevent race conditions around checking
+        # for duplicate attachment requests to the same host.
+
         # We fetch the volume object and pass it to the rpc call because we
         # need to direct this to the correct host/backend
 
@@ -2166,6 +2173,29 @@ class API(base.Base):
                        'volume_status': volume_ref.status}
             LOG.error(msg)
             raise exception.InvalidVolume(reason=msg)
+
+        if (len(volume_ref.volume_attachment) > 1 and
+            not (volume_ref.multiattach or
+                 self._is_multiattach(volume_ref.volume_type))):
+            # Check whether all connection hosts are unique
+            # Multiple attachments to different hosts is permitted to
+            # support Nova instance migration.
+
+            # This particular check also does not prevent multiple attachments
+            # for a multiattach volume to the same instance.
+
+            connection_hosts = set(a.connector['host']
+                                   for a in volume_ref.volume_attachment
+                                   if a.connection_info)
+
+            if len(connection_hosts) > 0:
+                # We raced, and have more than one connection
+
+                msg = _('duplicate connectors detected on volume '
+                        '%(vol)s') % {'vol': volume_ref.id}
+
+                raise exception.InvalidVolume(reason=msg)
+
         connection_info = (
             self.volume_rpcapi.attachment_update(ctxt,
                                                  volume_ref,

cinder/wsgi/wsgi.py

@@ -30,6 +30,7 @@ i18n.enable_lazy()
 # Need to register global_opts
 from cinder.common import config
 from cinder.common import constants
+from cinder import coordination
 from cinder import rpc
 from cinder import service
 from cinder import version
@@ -44,6 +45,8 @@ def initialize_application():
     logging.setup(CONF, "cinder")
     config.set_middleware_defaults()
 
+    coordination.COORDINATOR.start()
+
     rpc.init(CONF)
     service.setup_profiler(constants.API_BINARY, CONF.host)