3PAR: Adding volume checks to manage snapshot API

Updated the checks in the HPE 3PAR manage snapshot API to use
the volume object for comparison. Also added new checks to
prevent un/managing of snapshots from/to volumes that have a
status of 'failed-over' for replication status.

This is a follow-up patch for:
https://review.openstack.org/#/c/253760/

Implements: blueprint 3par-manage-unmanage-snapshot
Change-Id: Ie187e67d9eec2121d6ccad5667c6d9fc73f98ce3

diff --git a/cinder/tests/unit/test_hpe3par.py b/cinder/tests/unit/test_hpe3par.py
index ced41c7152e3449eed64edd4cc46714e1635d0b3..a6f033ac0d2643900d44f5a3747fd92f0f2b0d9c 100644 (file)
--- a/cinder/tests/unit/test_hpe3par.py
+++ b/cinder/tests/unit/test_hpe3par.py
@@ -3459,14 +3459,17 @@ class HPE3PARBaseDriver(object):
 
         new_comment = Comment({
             "display_name": "snap",
-            "volume_name": "volume-007dbfce-7579-40bc-8f90-a20b3902283e",
-            "volume_id": "007dbfce-7579-40bc-8f90-a20b3902283e",
+            "volume_name": self.VOLUME_NAME,
+            "volume_id": self.VOLUME_ID,
             "description": "",
         })
+
+        volume = {'id': self.VOLUME_ID}
+
         snapshot = {
             'display_name': None,
-            'id': '007dbfce-7579-40bc-8f90-a20b3902283e',
-            'volume_id': self.VOLUME_ID,
+            'id': self.SNAPSHOT_ID,
+            'volume': volume,
         }
 
         mock_client.getVolume.return_value = {
@@ -3492,7 +3495,6 @@ class HPE3PARBaseDriver(object):
                                        {'newName': oss_matcher,
                                         'comment': new_comment}),
             ]
-
             mock_client.assert_has_calls(
                 self.standard_login +
                 expected +
@@ -3502,10 +3504,12 @@ class HPE3PARBaseDriver(object):
     def test_manage_existing_snapshot_invalid_parent(self):
         mock_client = self.setup_driver()
 
+        volume = {'id': self.VOLUME_ID}
+
         snapshot = {
             'display_name': None,
             'id': '007dbfce-7579-40bc-8f90-a20b3902283e',
-            'volume_id': self.VOLUME_ID,
+            'volume': volume,
         }
 
         mock_client.getVolume.return_value = {
@@ -3535,6 +3539,38 @@ class HPE3PARBaseDriver(object):
                 expected +
                 self.standard_logout)
 
+    def test_manage_existing_snapshot_failed_over_volume(self):
+        mock_client = self.setup_driver()
+
+        volume = {
+            'id': self.VOLUME_ID,
+            'replication_status': 'failed-over',
+        }
+
+        snapshot = {
+            'display_name': None,
+            'id': '007dbfce-7579-40bc-8f90-a20b3902283e',
+            'volume': volume,
+        }
+
+        mock_client.getVolume.return_value = {
+            "comment": "{'display_name': 'snap'}",
+            'copyOf': self.VOLUME_NAME_3PAR,
+        }
+
+        with mock.patch.object(hpecommon.HPE3PARCommon,
+                               '_create_client') as mock_create_client:
+            mock_create_client.return_value = mock_client
+            common = self.driver._login()
+
+            ums_matcher = common._get_3par_ums_name(snapshot['id'])
+            existing_ref = {'source-name': ums_matcher}
+
+            self.assertRaises(exception.InvalidInput,
+                              self.driver.manage_existing_snapshot,
+                              snapshot=snapshot,
+                              existing_ref=existing_ref)
+
     def test_manage_existing_get_size(self):
         mock_client = self.setup_driver()
         mock_client.getVolume.return_value = {'sizeMiB': 2048}
@@ -3730,6 +3766,22 @@ class HPE3PARBaseDriver(object):
                 expected +
                 self.standard_logout)
 
+    def test_unmanage_snapshot_failed_over_volume(self):
+        mock_client = self.setup_driver()
+
+        volume = {'replication_status': 'failed-over', }
+        snapshot = {'id': self.SNAPSHOT_ID,
+                    'display_name': 'fake_snap',
+                    'volume': volume, }
+
+        with mock.patch.object(hpecommon.HPE3PARCommon,
+                               '_create_client') as mock_create_client:
+            mock_create_client.return_value = mock_client
+
+            self.assertRaises(exception.SnapshotIsBusy,
+                              self.driver.unmanage_snapshot,
+                              snapshot=snapshot)
+
     def test__safe_hostname(self):
         long_hostname = "abc123abc123abc123abc123abc123abc123"
         fixed_hostname = "abc123abc123abc123abc123abc123a"

diff --git a/cinder/volume/drivers/hpe/hpe_3par_common.py b/cinder/volume/drivers/hpe/hpe_3par_common.py
index 05f1f0c591e8f7caabe2bb7f4e027e8c861a9c01..1bd179cb2db7c9afd1b6115b4d4ee5a97fe68683 100644 (file)
--- a/cinder/volume/drivers/hpe/hpe_3par_common.py
+++ b/cinder/volume/drivers/hpe/hpe_3par_common.py
@@ -222,10 +222,11 @@ class HPE3PARCommon(object):
         3.0.7 - Enable standard capabilities based on 3PAR licenses
         3.0.8 - Optimize array ID retrieval
         3.0.9 - Bump minimum API version for volume replication
+        3.0.10 - Added additional volumes checks to the manage snapshot API
 
     """
 
-    VERSION = "3.0.9"
+    VERSION = "3.0.10"
 
     stats = {}
 
@@ -792,6 +793,15 @@ class HPE3PARCommon(object):
         existing_ref is a dictionary of the form:
         {'source-name': <name of the snapshot>}
         """
+        # Potential parent volume for the snapshot
+        volume = snapshot['volume']
+
+        # Do not allow for managing of snapshots for 'failed-over' volumes.
+        if volume.get('replication_status') == 'failed-over':
+            err = (_("Managing of snapshots to failed-over volumes is "
+                     "not allowed."))
+            raise exception.InvalidInput(reason=err)
+
         target_snap_name = self._get_existing_volume_ref_name(existing_ref,
                                                               is_snapshot=True)
 
@@ -805,7 +815,7 @@ class HPE3PARCommon(object):
             raise exception.InvalidInput(reason=err)
 
         # Make sure the snapshot is being associated with the correct volume.
-        parent_vol_name = self._get_3par_vol_name(snapshot['volume_id'])
+        parent_vol_name = self._get_3par_vol_name(volume['id'])
         if parent_vol_name != snap['copyOf']:
             err = (_("The provided snapshot '%s' is not a snapshot of "
                      "the provided volume.") % target_snap_name)
@@ -829,8 +839,8 @@ class HPE3PARCommon(object):
 
         # Generate the new snapshot information based on the new ID.
         new_snap_name = self._get_3par_snap_name(snapshot['id'])
-        new_comment['volume_id'] = snapshot['id']
-        new_comment['volume_name'] = 'volume-' + snapshot['id']
+        new_comment['volume_id'] = volume['id']
+        new_comment['volume_name'] = 'volume-' + volume['id']
         if snapshot.get('display_description', None):
             new_comment['description'] = snapshot['display_description']
         else:
@@ -922,6 +932,18 @@ class HPE3PARCommon(object):
 
     def unmanage_snapshot(self, snapshot):
         """Removes the specified snapshot from Cinder management."""
+        # Parent volume for the snapshot
+        volume = snapshot['volume']
+
+        # Do not allow unmanaging of snapshots from 'failed-over' volumes.
+        if volume.get('replication_status') == 'failed-over':
+            err = (_("Unmanaging of snapshots from failed-over volumes is "
+                     "not allowed."))
+            LOG.error(err)
+            # TODO(leeantho) Change this exception to Invalid when the volume
+            # manager supports handling that.
+            raise exception.SnapshotIsBusy(snapshot_name=snapshot['id'])
+
         # Rename the snapshots's name to ums-* format so that it can be
         # easily found later.
         snap_name = self._get_3par_snap_name(snapshot['id'])