Verify volume is replication capable

The V1 implementation of replication added a replication_status
field to the Volume object and set it to disabled by default on
volume creation.

This is problematic, because there's no way for the API or any
caller to know if the status is disabled because it was
explicitly set that way, or if it is because the back end or
volume-type do not support replication.

This results in cases like enable replication (which does a status
check on disabled) to be called inappropriately on devices that
don't support replication.

This patch adds a decorator to the new replication methods which will
check that the volume is of type with replication_enabled=True before
attempting any replication related operations.

Change-Id: I943be2aef3b7c32026278f311dae9f82194372fe
Closes-Bug: #1503439

diff --git a/cinder/tests/unit/test_volume.py b/cinder/tests/unit/test_volume.py
index 266858c5300b9b35303ebe409dac40d2140610c3..884d46990c228253d04091ea8cf0e7545fec29aa 100644 (file)
--- a/cinder/tests/unit/test_volume.py
+++ b/cinder/tests/unit/test_volume.py
@@ -6292,6 +6292,24 @@ class GenericVolumeDriverTestCase(DriverTestCase):
                           volume_api.enable_replication,
                           ctxt, volume)
 
+    def test_enable_replication_invalid_type(self):
+        volume_api = cinder.volume.api.API()
+        ctxt = context.get_admin_context()
+
+        volume = tests_utils.create_volume(self.context,
+                                           size=1,
+                                           host=CONF.host,
+                                           replication_status='disabled')
+        volume['volume_type_id'] = 'dab02f01-b50f-4ed6-8d42-2b5b9680996e'
+        fake_specs = {}
+        with mock.patch.object(volume_types,
+                               'get_volume_type_extra_specs',
+                               return_value = fake_specs):
+            self.assertRaises(exception.InvalidVolume,
+                              volume_api.enable_replication,
+                              ctxt,
+                              volume)
+
     def test_enable_replication(self):
         volume_api = cinder.volume.api.API()
         ctxt = context.get_admin_context()
@@ -6300,8 +6318,14 @@ class GenericVolumeDriverTestCase(DriverTestCase):
                                            size=1,
                                            host=CONF.host,
                                            replication_status='disabled')
+        volume['volume_type_id'] = 'dab02f01-b50f-4ed6-8d42-2b5b9680996e'
+        fake_specs = {'replication_enabled': '<is> True'}
         with mock.patch.object(volume_rpcapi.VolumeAPI,
-                               'enable_replication') as mock_enable_rep:
+                               'enable_replication') as mock_enable_rep,\
+            mock.patch.object(volume_types,
+                              'get_volume_type_extra_specs',
+                              return_value = fake_specs):
+
             volume_api.enable_replication(ctxt, volume)
             self.assertTrue(mock_enable_rep.called)
 
@@ -6326,8 +6350,13 @@ class GenericVolumeDriverTestCase(DriverTestCase):
                                            host=CONF.host,
                                            replication_status='disabled')
 
+        volume['volume_type_id'] = 'dab02f01-b50f-4ed6-8d42-2b5b9680996e'
+        fake_specs = {'replication_enabled': '<is> True'}
         with mock.patch.object(volume_rpcapi.VolumeAPI,
-                               'disable_replication') as mock_disable_rep:
+                               'disable_replication') as mock_disable_rep,\
+                mock.patch.object(volume_types,
+                                  'get_volume_type_extra_specs',
+                                  return_value = fake_specs):
             volume_api.disable_replication(ctxt, volume)
             self.assertTrue(mock_disable_rep.called)
 

diff --git a/cinder/volume/api.py b/cinder/volume/api.py
index fe8c48312bc8e75b26e749f1098ec25ed3731a84..91034cd003e1e21688d604854ec7be33d8585ee5 100644 (file)
--- a/cinder/volume/api.py
+++ b/cinder/volume/api.py
@@ -115,6 +115,29 @@ def check_policy(context, action, target_obj=None):
     cinder.policy.enforce(context, _action, target)
 
 
+def valid_replication_volume(func):
+    """Check that the volume is capable of replication.
+
+    This decorator requires the first 3 args of the wrapped function
+    to be (self, context, volume)
+    """
+    @functools.wraps(func)
+    def wrapped(self, context, volume, *args, **kwargs):
+        rep_capable = False
+        if volume.get('volume_type_id', None):
+            extra_specs = volume_types.get_volume_type_extra_specs(
+                volume.get('volume_type_id'))
+            rep_capable = extra_specs.get('replication_enabled',
+                                          False) == "<is> True"
+        if not rep_capable:
+            msg = _("Volume is not a replication enabled volume, "
+                    "replication operations can only be performed "
+                    "on volumes that are of type replication_enabled.")
+            raise exception.InvalidVolume(reason=msg)
+        return func(self, context, volume, *args, **kwargs)
+    return wrapped
+
+
 class API(base.Base):
     """API for interacting with the volume manager."""
 
@@ -1590,8 +1613,8 @@ class API(base.Base):
     # now they're a special resource, so no.
 
     @wrap_check_policy
+    @valid_replication_volume
     def enable_replication(self, ctxt, volume):
-
         # NOTE(jdg): details like sync vs async
         # and replica count are to be set via the
         # volume-type and config files.
@@ -1627,6 +1650,7 @@ class API(base.Base):
         self.volume_rpcapi.enable_replication(ctxt, vref)
 
     @wrap_check_policy
+    @valid_replication_volume
     def disable_replication(self, ctxt, volume):
 
         valid_disable_status = ['disabled', 'enabled']
@@ -1652,6 +1676,7 @@ class API(base.Base):
         self.volume_rpcapi.disable_replication(ctxt, vref)
 
     @wrap_check_policy
+    @valid_replication_volume
     def failover_replication(self,
                              ctxt,
                              volume,
@@ -1684,6 +1709,7 @@ class API(base.Base):
                                                 secondary)
 
     @wrap_check_policy
+    @valid_replication_volume
     def list_replication_targets(self, ctxt, volume):
 
         # NOTE(jdg): This collects info for the specified volume