snapshot_name = "fakeshapshot"
snapshot_id = 3
snapshot = {
+ 'id': snapshot_id,
'name': snapshot_name,
+ 'display_name': 'fakesnap',
'volume_name': volume_name,
'volume': volume}
mock.call.getVolumeByName(self.volume_name),
mock.call.logout()])
+ def test_manage_existing_snapshot(self):
+ mock_client = self.setup_driver()
+
+ self.driver.api_version = "1.1"
+
+ volume = {
+ 'id': '111',
+ }
+ snapshot = {
+ 'display_name': 'Foo Snap',
+ 'id': '12345',
+ 'volume': volume,
+ 'volume_id': '111',
+ }
+
+ with mock.patch.object(hpe_lefthand_iscsi.HPELeftHandISCSIDriver,
+ '_create_client') as mock_do_setup:
+ mock_do_setup.return_value = mock_client
+ mock_client.getSnapshotByName.return_value = {
+ 'id': self.snapshot_id
+ }
+ mock_client.getSnapshotParentVolume.return_value = {
+ 'name': 'volume-111'
+ }
+
+ existing_ref = {'source-name': self.snapshot_name}
+ expected_obj = {'display_name': 'Foo Snap'}
+
+ obj = self.driver.manage_existing_snapshot(snapshot, existing_ref)
+
+ mock_client.assert_has_calls(
+ self.driver_startup_call_stack + [
+ mock.call.getSnapshotByName(self.snapshot_name),
+ mock.call.getSnapshotParentVolume(self.snapshot_name),
+ mock.call.modifySnapshot(self.snapshot_id,
+ {'name': 'snapshot-12345'}),
+ mock.call.logout()])
+ self.assertEqual(expected_obj, obj)
+
+ def test_manage_existing_snapshot_failed_over_volume(self):
+ mock_client = self.setup_driver()
+
+ self.driver.api_version = "1.1"
+
+ volume = {
+ 'id': self.volume_id,
+ 'replication_status': 'failed-over',
+ }
+ snapshot = {
+ 'display_name': 'Foo Snap',
+ 'id': '12345',
+ 'volume': volume,
+ }
+ existing_ref = {'source-name': self.snapshot_name}
+
+ with mock.patch.object(hpe_lefthand_iscsi.HPELeftHandISCSIDriver,
+ '_create_client') as mock_do_setup:
+ mock_do_setup.return_value = mock_client
+
+ 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.getVolumeByName.return_value = {'size': 2147483648}
self.driver_startup_call_stack +
expected)
+ def test_manage_existing_snapshot_get_size(self):
+ mock_client = self.setup_driver()
+ mock_client.getSnapshotByName.return_value = {'size': 2147483648}
+
+ self.driver.api_version = "1.1"
+
+ with mock.patch.object(hpe_lefthand_iscsi.HPELeftHandISCSIDriver,
+ '_create_client') as mock_do_setup:
+ mock_do_setup.return_value = mock_client
+
+ snapshot = {}
+ existing_ref = {'source-name': self.snapshot_name}
+
+ size = self.driver.manage_existing_snapshot_get_size(snapshot,
+ existing_ref)
+
+ expected_size = 2
+ expected = [mock.call.getSnapshotByName(
+ existing_ref['source-name']),
+ mock.call.logout()]
+
+ mock_client.assert_has_calls(
+ self.driver_startup_call_stack +
+ expected)
+ self.assertEqual(expected_size, size)
+
+ def test_manage_existing_snapshot_get_size_invalid_reference(self):
+ mock_client = self.setup_driver()
+ mock_client.getSnapshotByName.return_value = {'size': 2147483648}
+
+ self.driver.api_version = "1.1"
+
+ with mock.patch.object(hpe_lefthand_iscsi.HPELeftHandISCSIDriver,
+ '_create_client') as mock_do_setup:
+ mock_do_setup.return_value = mock_client
+
+ snapshot = {}
+ existing_ref = {'source-name': "snapshot-12345"}
+
+ self.assertRaises(exception.ManageExistingInvalidReference,
+ self.driver.manage_existing_snapshot_get_size,
+ snapshot=snapshot,
+ existing_ref=existing_ref)
+
+ mock_client.assert_has_calls([])
+
+ existing_ref = {}
+
+ self.assertRaises(exception.ManageExistingInvalidReference,
+ self.driver.manage_existing_snapshot_get_size,
+ snapshot=snapshot,
+ existing_ref=existing_ref)
+
+ mock_client.assert_has_calls([])
+
+ def test_manage_existing_snapshot_get_size_invalid_input(self):
+ mock_client = self.setup_driver()
+ mock_client.getSnapshotByName.side_effect = (
+ hpeexceptions.HTTPNotFound('fake'))
+
+ self.driver.api_version = "1.1"
+
+ with mock.patch.object(hpe_lefthand_iscsi.HPELeftHandISCSIDriver,
+ '_create_client') as mock_do_setup:
+ mock_do_setup.return_value = mock_client
+
+ snapshot = {}
+ existing_ref = {'source-name': self.snapshot_name}
+
+ self.assertRaises(exception.InvalidInput,
+ self.driver.manage_existing_snapshot_get_size,
+ snapshot=snapshot,
+ existing_ref=existing_ref)
+
+ expected = [mock.call.getSnapshotByName(
+ existing_ref['source-name'])]
+
+ mock_client.assert_has_calls(
+ self.driver_startup_call_stack +
+ expected)
+
def test_unmanage(self):
mock_client = self.setup_driver()
mock_client.getVolumeByName.return_value = {'id': self.volume_id}
self.driver_startup_call_stack +
expected)
+ def test_unmanage_snapshot(self):
+ mock_client = self.setup_driver()
+ volume = {
+ 'id': self.volume_id,
+ }
+ snapshot = {
+ 'name': self.snapshot_name,
+ 'display_name': 'Foo Snap',
+ 'volume': volume,
+ 'id': self.snapshot_id,
+ }
+ mock_client.getSnapshotByName.return_value = {'id': self.snapshot_id, }
+
+ self.driver.api_version = "1.1"
+
+ with mock.patch.object(hpe_lefthand_iscsi.HPELeftHandISCSIDriver,
+ '_create_client') as mock_do_setup:
+ mock_do_setup.return_value = mock_client
+ self.driver.unmanage_snapshot(snapshot)
+
+ new_name = 'ums-' + str(self.snapshot_id)
+
+ expected = [
+ mock.call.getSnapshotByName(snapshot['name']),
+ mock.call.modifySnapshot(self.snapshot_id, {'name': new_name}),
+ mock.call.logout()
+ ]
+
+ mock_client.assert_has_calls(
+ self.driver_startup_call_stack +
+ expected)
+
+ def test_unmanage_snapshot_failed_over_volume(self):
+ mock_client = self.setup_driver()
+ volume = {
+ 'id': self.volume_id,
+ 'replication_status': 'failed-over',
+ }
+ snapshot = {
+ 'name': self.snapshot_name,
+ 'display_name': 'Foo Snap',
+ 'volume': volume,
+ 'id': self.snapshot_id,
+ }
+ mock_client.getSnapshotByName.return_value = {'id': self.snapshot_id, }
+
+ self.driver.api_version = "1.1"
+
+ with mock.patch.object(hpe_lefthand_iscsi.HPELeftHandISCSIDriver,
+ '_create_client') as mock_do_setup:
+ mock_do_setup.return_value = mock_client
+
+ self.assertRaises(exception.SnapshotIsBusy,
+ self.driver.unmanage_snapshot,
+ snapshot=snapshot)
+
def test_api_version(self):
self.setup_driver()
self.driver.api_version = "1.1"
2.0.1 - Remove db access for consistency groups
2.0.2 - Adds v2 managed replication support
2.0.3 - Adds v2 unmanaged replication support
+ 2.0.4 - Add manage/unmanage snapshot support
"""
- VERSION = "2.0.3"
+ VERSION = "2.0.4"
device_stats = {}
# any model updates from retype.
return updates
+ def manage_existing_snapshot(self, snapshot, existing_ref):
+ """Manage an existing LeftHand snapshot.
+
+ existing_ref is a dictionary of the form:
+ {'source-name': <name of the snapshot>}
+ """
+ # Check API Version
+ self._check_api_version()
+
+ # Potential parent volume for the snapshot
+ volume = snapshot['volume']
+
+ 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)
+
+ # Check for the existence of the virtual volume.
+ client = self._login()
+ try:
+ updates = self._manage_snapshot(client,
+ volume,
+ snapshot,
+ target_snap_name,
+ existing_ref)
+ finally:
+ self._logout(client)
+
+ # Return display name to update the name displayed in the GUI and
+ # any model updates from retype.
+ return updates
+
+ def _manage_snapshot(self, client, volume, snapshot, target_snap_name,
+ existing_ref):
+ # Check for the existence of the virtual volume.
+ try:
+ snapshot_info = client.getSnapshotByName(target_snap_name)
+ except hpeexceptions.HTTPNotFound:
+ err = (_("Snapshot '%s' doesn't exist on array.") %
+ target_snap_name)
+ LOG.error(err)
+ raise exception.InvalidInput(reason=err)
+
+ # Make sure the snapshot is being associated with the correct volume.
+ try:
+ parent_vol = client.getSnapshotParentVolume(target_snap_name)
+ except hpeexceptions.HTTPNotFound:
+ err = (_("Could not find the parent volume for Snapshot '%s' on "
+ "array.") % target_snap_name)
+ LOG.error(err)
+ raise exception.InvalidInput(reason=err)
+
+ parent_vol_name = 'volume-' + snapshot['volume_id']
+ if parent_vol_name != parent_vol['name']:
+ err = (_("The provided snapshot '%s' is not a snapshot of "
+ "the provided volume.") % target_snap_name)
+ LOG.error(err)
+ raise exception.InvalidInput(reason=err)
+
+ # Generate the new snapshot information based on the new ID.
+ new_snap_name = 'snapshot-' + snapshot['id']
+
+ new_vals = {"name": new_snap_name}
+
+ try:
+ # Update the existing snapshot with the new name.
+ client.modifySnapshot(snapshot_info['id'], new_vals)
+ except hpeexceptions.HTTPServerError:
+ err = (_("An error occured while attempting to modify"
+ "Snapshot '%s'.") % snapshot_info['id'])
+ LOG.error(err)
+
+ LOG.info(_LI("Snapshot '%(ref)s' renamed to '%(new)s'."),
+ {'ref': existing_ref['source-name'], 'new': new_snap_name})
+
+ display_name = None
+ if snapshot['display_name']:
+ display_name = snapshot['display_name']
+
+ updates = {'display_name': display_name}
+
+ LOG.info(_LI("Snapshot %(disp)s '%(new)s' is "
+ "now being managed."),
+ {'disp': display_name, 'new': new_snap_name})
+
+ return updates
+
def manage_existing_get_size(self, volume, existing_ref):
"""Return size of volume to be managed by manage_existing.
return int(math.ceil(float(volume_info['size']) / units.Gi))
+ def manage_existing_snapshot_get_size(self, snapshot, existing_ref):
+ """Return size of volume to be managed by manage_existing.
+
+ existing_ref is a dictionary of the form:
+ {'source-name': <name of the virtual volume>}
+ """
+ # Check API version.
+ self._check_api_version()
+
+ target_snap_name = self._get_existing_volume_ref_name(existing_ref)
+
+ # Make sure the reference is not in use.
+ if re.match('volume-*|snapshot-*|unm-*', target_snap_name):
+ reason = _("Reference must be the name of an unmanaged "
+ "snapshot.")
+ raise exception.ManageExistingInvalidReference(
+ existing_ref=target_snap_name,
+ reason=reason)
+
+ # Check for the existence of the virtual volume.
+ client = self._login()
+ try:
+ snapshot_info = client.getSnapshotByName(target_snap_name)
+ except hpeexceptions.HTTPNotFound:
+ err = (_("Snapshot '%s' doesn't exist on array.") %
+ target_snap_name)
+ LOG.error(err)
+ raise exception.InvalidInput(reason=err)
+ finally:
+ self._logout(client)
+
+ return int(math.ceil(float(snapshot_info['size']) / units.Gi))
+
def unmanage(self, volume):
"""Removes the specified volume from Cinder management."""
# Check API version.
'vol': volume['name'],
'new': new_vol_name})
+ def unmanage_snapshot(self, snapshot):
+ """Removes the specified snapshot from Cinder management."""
+ # Check API version.
+ self._check_api_version()
+
+ # Potential parent volume for the snapshot
+ volume = snapshot['volume']
+
+ 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.
+ client = self._login()
+ try:
+ snapshot_info = client.getSnapshotByName(snapshot['name'])
+ new_snap_name = 'ums-' + six.text_type(snapshot['id'])
+ options = {'name': new_snap_name}
+ client.modifySnapshot(snapshot_info['id'], options)
+ LOG.info(_LI("Snapshot %(disp)s '%(vol)s' is no longer managed. "
+ "Snapshot renamed to '%(new)s'."),
+ {'disp': snapshot['display_name'],
+ 'vol': snapshot['name'],
+ 'new': new_snap_name})
+ finally:
+ self._logout(client)
+
def _get_existing_volume_ref_name(self, existing_ref):
"""Returns the volume name of an existing reference.
Code Review / openstack-build / cinder-build.git / commitdiff