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

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)
 

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