openstack
/
cinder
Code Issues Proposed changes

nimble storage: support for force detach 

nimble storage driver doesn't handle force detach with empty connector
information in terminate connection. The fix takes care of empty connectors
and removing all ACL's on the volume if connector information is absent

Change-Id: I74a5b48c414db2b5b9e27d986d776d7c2ae3f383
Closes-Bug: #1795070
Signed-off-by: Raunak Kumar <rkumar@nimblestorage.com>
(cherry picked from commit b2d0ac0c2e)
This commit is contained in:
Raunak Kumar 2018-10-02 09:48:31 -07:00
parent ecb06ef6fc
commit 3ca19b0dd0
2 changed files with 54 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
cinder/tests/unit/volume/drivers/test_nimble.py
View File

@ -1365,6 +1365,27 @@ class NimbleDriverConnectionTestCase(NimbleDriverBaseTestCase):
self.mock_client_service.method_calls,
expected_calls)
@mock.patch(NIMBLE_URLLIB2)
@mock.patch(NIMBLE_CLIENT)
@mock.patch.object(obj_volume.VolumeList, 'get_all_by_host',
mock.Mock(return_value=[]))
@NimbleDriverBaseTestCase.client_mock_decorator(create_configuration(
'nimble', 'nimble_pass', '10.18.108.55', 'default', '*'))
def test_terminate_connection_without_connector(self):
self.mock_client_service.get_initiator_grp_list.return_value = (
FAKE_IGROUP_LIST_RESPONSE)
self.driver.terminate_connection(
{'name': 'test-volume',
'provider_location': '12 13',
'id': 12},
None)
expected_calls = [mock.call._get_igroupname_for_initiator(
'test-initiator1'),
mock.call.remove_all_acls({'name': 'test-volume'})]
self.mock_client_service.assert_has_calls(
self.mock_client_service.method_calls,
expected_calls)
@mock.patch(NIMBLE_URLLIB2)
@mock.patch(NIMBLE_CLIENT)
@mock.patch.object(obj_volume.VolumeList, 'get_all_by_host',

33
cinder/volume/drivers/nimble.py
View File

@ -785,6 +785,13 @@ class NimbleISCSIDriver(NimbleBaseVolumeDriver, san.SanISCSIDriver):
{'vol': volume['name'],
'conn': connector,
'loc': volume['provider_location']})
if connector is None:
LOG.warning("Removing ALL host connections for volume %s",
volume)
self.APIExecutor.remove_all_acls(volume)
return
initiator_name = connector['initiator']
initiator_group_name = self._get_igroupname_for_initiator(
initiator_name)
@ -977,6 +984,12 @@ class NimbleFCDriver(NimbleBaseVolumeDriver, driver.FibreChannelDriver):
'conn': connector,
'loc': volume['provider_location']})
wwpns = []
if connector is None:
LOG.warning("Removing ALL host connections for volume %s",
volume)
self.APIExecutor.remove_all_acls(volume)
return
initiator_name = connector['initiator']
for wwpn in connector['wwpns']:
wwpns.append(wwpn)
@ -1504,6 +1517,26 @@ class NimbleRestAPIExecutor(object):
'igroup': initiator_group_id})
return r.json()['data'][0]
def get_volume_acl_records(self, volume_id):
api = "volumes/" + six.text_type(volume_id)
r = self.get(api)
if not r.json()['data']:
raise NimbleAPIException(_("Unable to retrieve information for "
"volume: %s") % volume_id)
return r.json()['data']['access_control_records']
def remove_all_acls(self, volume):
LOG.info("removing all access control list from volume=%(vol)s",
{"vol": volume['name']})
volume_id = self.get_volume_id_by_name(volume['name'])
acl_records = self.get_volume_acl_records(volume_id)
if acl_records is not None:
for acl_record in acl_records:
LOG.info("removing acl=%(acl)s with igroup=%(igroup)s",
{"acl": acl_record['id'],
"igroup": acl_record['initiator_group_name']})
self.remove_acl(volume, acl_record['initiator_group_name'])
def remove_acl(self, volume, initiator_group_name):
LOG.info("removing ACL from volume=%(vol)s"
"and %(igroup)s",