NetApp ONTAP: Fix FlexGroup replication

Currently driver's implementation use the `volume-create` ZAPI call
to create a secondary volume on destination and enable replication.
That call only supports the creation of FlexVols, and must be
replaced by the async call `volume-create-async` when dealing with
FlexGroups.

Change-Id: I07fe2e62a046f7276b6dcd78fd25c89412045d70
This commit is contained in:
Fernando Ferraz 2021-03-16 02:50:59 +00:00 committed by Fernando Ferraz Silva
parent 4fcc583813
commit 1fba4a9e93
8 changed files with 624 additions and 70 deletions

View File

@ -663,6 +663,31 @@ AGGR_GET_ITER_CAPACITY_RESPONSE = etree.XML("""
'total_size': AGGR_SIZE_TOTAL, 'total_size': AGGR_SIZE_TOTAL,
}) })
VOLUME_STATE_ONLINE = 'online'
VOLUME_GET_ITER_STATE_ATTR_STR = """
<volume-attributes>
<volume-id-attributes>
<style-extended>flexgroup</style-extended>
</volume-id-attributes>
<volume-state-attributes>
<state>%(state)s</state>
</volume-state-attributes>
</volume-attributes>
""" % {
'state': VOLUME_STATE_ONLINE
}
VOLUME_GET_ITER_STATE_ATTR = etree.XML(VOLUME_GET_ITER_STATE_ATTR_STR)
VOLUME_GET_ITER_STATE_RESPONSE = etree.XML("""
<results status="passed">
<num-records>1</num-records>
<attributes-list> %(volume)s </attributes-list>
</results>
""" % {
'volume': VOLUME_GET_ITER_STATE_ATTR_STR,
})
VOLUME_SIZE_TOTAL = 19922944 VOLUME_SIZE_TOTAL = 19922944
VOLUME_SIZE_AVAILABLE = 19791872 VOLUME_SIZE_AVAILABLE = 19791872
VOLUME_GET_ITER_CAPACITY_ATTR_STR = """ VOLUME_GET_ITER_CAPACITY_ATTR_STR = """

View File

@ -22,6 +22,7 @@ import uuid
import ddt import ddt
from lxml import etree from lxml import etree
from oslo_utils import units
import paramiko import paramiko
import six import six
@ -1717,6 +1718,51 @@ class NetAppCmodeClientTestCase(test.TestCase):
self.assertEqual(expected_result, address_list) self.assertEqual(expected_result, address_list)
@ddt.data({'junction_path': '/fake/vol'},
{'name': 'fake_volume'},
{'junction_path': '/fake/vol', 'name': 'fake_volume'})
def test_get_volume_state(self, kwargs):
api_response = netapp_api.NaElement(
fake_client.VOLUME_GET_ITER_STATE_RESPONSE)
mock_send_iter_request = self.mock_object(
self.client, 'send_iter_request', return_value=api_response)
volume_response = netapp_api.NaElement(
fake_client.VOLUME_GET_ITER_STATE_ATTR)
mock_get_unique_vol = self.mock_object(
self.client, 'get_unique_volume', return_value=volume_response)
state = self.client.get_volume_state(**kwargs)
volume_id_attributes = {}
if 'junction_path' in kwargs:
volume_id_attributes['junction-path'] = kwargs['junction_path']
if 'name' in kwargs:
volume_id_attributes['name'] = kwargs['name']
volume_get_iter_args = {
'query': {
'volume-attributes': {
'volume-id-attributes': volume_id_attributes,
}
},
'desired-attributes': {
'volume-attributes': {
'volume-id-attributes': {
'style-extended': None,
},
'volume-state-attributes': {
'state': None
}
}
},
}
mock_send_iter_request.assert_called_once_with(
'volume-get-iter', volume_get_iter_args)
mock_get_unique_vol.assert_called_once_with(api_response)
self.assertEqual(fake_client.VOLUME_STATE_ONLINE, state)
@ddt.data({'flexvol_path': '/fake/vol'}, @ddt.data({'flexvol_path': '/fake/vol'},
{'flexvol_name': 'fake_volume'}, {'flexvol_name': 'fake_volume'},
{'flexvol_path': '/fake/vol', 'flexvol_name': 'fake_volume'}) {'flexvol_path': '/fake/vol', 'flexvol_name': 'fake_volume'})
@ -1961,6 +2007,51 @@ class NetAppCmodeClientTestCase(test.TestCase):
self.client.enable_flexvol_compression.assert_called_once_with( self.client.enable_flexvol_compression.assert_called_once_with(
fake_client.VOLUME_NAME) fake_client.VOLUME_NAME)
def test_create_volume_async(self):
self.mock_object(self.client.connection, 'send_request')
self.client.create_volume_async(
fake_client.VOLUME_NAME, [fake_client.VOLUME_AGGREGATE_NAME], 100,
volume_type='dp')
volume_create_args = {
'aggr-list': [{'aggr-name': fake_client.VOLUME_AGGREGATE_NAME}],
'size': 100 * units.Gi,
'volume-name': fake_client.VOLUME_NAME,
'volume-type': 'dp'
}
self.client.connection.send_request.assert_called_once_with(
'volume-create-async', volume_create_args)
@ddt.data('dp', 'rw', None)
def test_create_volume_async_with_extra_specs(self, volume_type):
self.mock_object(self.client.connection, 'send_request')
self.client.create_volume_async(
fake_client.VOLUME_NAME, [fake_client.VOLUME_AGGREGATE_NAME], 100,
space_guarantee_type='volume', language='en-US',
snapshot_policy='default', snapshot_reserve=15,
volume_type=volume_type)
volume_create_args = {
'aggr-list': [{'aggr-name': fake_client.VOLUME_AGGREGATE_NAME}],
'size': 100 * units.Gi,
'volume-name': fake_client.VOLUME_NAME,
'space-reserve': 'volume',
'language-code': 'en-US',
'volume-type': volume_type,
'percentage-snapshot-reserve': '15',
}
if volume_type != 'dp':
volume_create_args['snapshot-policy'] = 'default'
volume_create_args['junction-path'] = ('/%s' %
fake_client.VOLUME_NAME)
self.client.connection.send_request.assert_called_with(
'volume-create-async', volume_create_args)
def test_flexvol_exists(self): def test_flexvol_exists(self):
api_response = netapp_api.NaElement( api_response = netapp_api.NaElement(
@ -2045,6 +2136,53 @@ class NetAppCmodeClientTestCase(test.TestCase):
self.client.connection.send_request.assert_has_calls([ self.client.connection.send_request.assert_has_calls([
mock.call('volume-mount', volume_mount_args)]) mock.call('volume-mount', volume_mount_args)])
def test_enable_volume_dedupe_async(self):
self.mock_object(self.client.connection, 'send_request')
self.client.enable_volume_dedupe_async(fake_client.VOLUME_NAME)
sis_enable_args = {'volume-name': fake_client.VOLUME_NAME}
self.client.connection.send_request.assert_called_once_with(
'sis-enable-async', sis_enable_args)
def test_disable_volume_dedupe_async(self):
self.mock_object(self.client.connection, 'send_request')
self.client.disable_volume_dedupe_async(fake_client.VOLUME_NAME)
sis_enable_args = {'volume-name': fake_client.VOLUME_NAME}
self.client.connection.send_request.assert_called_once_with(
'sis-disable-async', sis_enable_args)
def test_enable_volume_compression_async(self):
self.mock_object(self.client.connection, 'send_request')
self.client.enable_volume_compression_async(fake_client.VOLUME_NAME)
sis_set_config_args = {
'volume-name': fake_client.VOLUME_NAME,
'enable-compression': 'true'
}
self.client.connection.send_request.assert_called_once_with(
'sis-set-config-async', sis_set_config_args)
def test_disable_volume_compression_async(self):
self.mock_object(self.client.connection, 'send_request')
self.client.disable_volume_compression_async(fake_client.VOLUME_NAME)
sis_set_config_args = {
'volume-name': fake_client.VOLUME_NAME,
'enable-compression': 'false'
}
self.client.connection.send_request.assert_called_once_with(
'sis-set-config-async', sis_set_config_args)
def test_enable_flexvol_dedupe(self): def test_enable_flexvol_dedupe(self):
self.mock_object(self.client.connection, 'send_request') self.mock_object(self.client.connection, 'send_request')

View File

@ -135,12 +135,12 @@ SSC_AGGREGATE_INFO = {
} }
PROVISIONING_OPTS = { PROVISIONING_OPTS = {
'aggregate': 'fake_aggregate', 'aggregate': ['fake_aggregate'],
'thin_provisioned': True, 'thin_provisioned': True,
'snapshot_policy': None, 'snapshot_policy': None,
'language': 'en_US', 'language': 'en_US',
'dedupe_enabled': False, 'dedupe_enabled': True,
'compression_enabled': False, 'compression_enabled': True,
'snapshot_reserve': '12', 'snapshot_reserve': '12',
'volume_type': 'rw', 'volume_type': 'rw',
'size': 20, 'size': 20,
@ -148,7 +148,7 @@ PROVISIONING_OPTS = {
} }
ENCRYPTED_PROVISIONING_OPTS = { ENCRYPTED_PROVISIONING_OPTS = {
'aggregate': 'fake_aggregate', 'aggregate': ['fake_aggregate'],
'thin_provisioned': True, 'thin_provisioned': True,
'snapshot_policy': None, 'snapshot_policy': None,
'language': 'en_US', 'language': 'en_US',

View File

@ -166,9 +166,15 @@ class NetAppCDOTDataMotionMixinTestCase(test.TestCase):
self.assertDictEqual({'aggr1': 'aggr10'}, aggr_map) self.assertDictEqual({'aggr1': 'aggr10'}, aggr_map)
@ddt.data(True, False) @ddt.data({'dest_exists': True, 'is_flexgroup': False},
def test_create_snapmirror_dest_flexvol_exists(self, dest_exists): {'dest_exists': True, 'is_flexgroup': True},
{'dest_exists': False, 'is_flexgroup': False},
{'dest_exists': False, 'is_flexgroup': True})
@ddt.unpack
def test_create_snapmirror_dest_flexvol_exists(self, dest_exists,
is_flexgroup):
mock_dest_client = mock.Mock() mock_dest_client = mock.Mock()
mock_src_client = mock.Mock()
self.mock_object(mock_dest_client, 'flexvol_exists', self.mock_object(mock_dest_client, 'flexvol_exists',
return_value=dest_exists) return_value=dest_exists)
self.mock_object(mock_dest_client, 'get_snapmirrors', self.mock_object(mock_dest_client, 'get_snapmirrors',
@ -176,7 +182,15 @@ class NetAppCDOTDataMotionMixinTestCase(test.TestCase):
create_destination_flexvol = self.mock_object( create_destination_flexvol = self.mock_object(
self.dm_mixin, 'create_destination_flexvol') self.dm_mixin, 'create_destination_flexvol')
self.mock_object(utils, 'get_client_for_backend', self.mock_object(utils, 'get_client_for_backend',
return_value=mock_dest_client) side_effect=[mock_dest_client,
mock_src_client])
mock_provisioning_options = mock.Mock()
mock_provisioning_options.get.return_value = is_flexgroup
self.mock_object(mock_src_client,
'get_provisioning_options_from_flexvol',
return_value=mock_provisioning_options)
self.dm_mixin.create_snapmirror(self.src_backend, self.dm_mixin.create_snapmirror(self.src_backend,
self.dest_backend, self.dest_backend,
@ -186,16 +200,72 @@ class NetAppCDOTDataMotionMixinTestCase(test.TestCase):
if not dest_exists: if not dest_exists:
create_destination_flexvol.assert_called_once_with( create_destination_flexvol.assert_called_once_with(
self.src_backend, self.dest_backend, self.src_flexvol_name, self.src_backend, self.dest_backend, self.src_flexvol_name,
self.dest_flexvol_name) self.dest_flexvol_name, pool_is_flexgroup=is_flexgroup)
else: else:
self.assertFalse(create_destination_flexvol.called) self.assertFalse(create_destination_flexvol.called)
mock_dest_client.create_snapmirror.assert_called_once_with( mock_dest_client.create_snapmirror.assert_called_once_with(
self.src_vserver, self.src_flexvol_name, self.dest_vserver, self.src_vserver, self.src_flexvol_name, self.dest_vserver,
self.dest_flexvol_name, schedule='hourly') self.dest_flexvol_name,
schedule='hourly',
relationship_type=('extended_data_protection'
if is_flexgroup
else 'data_protection'))
mock_dest_client.initialize_snapmirror.assert_called_once_with( mock_dest_client.initialize_snapmirror.assert_called_once_with(
self.src_vserver, self.src_flexvol_name, self.dest_vserver, self.src_vserver, self.src_flexvol_name, self.dest_vserver,
self.dest_flexvol_name) self.dest_flexvol_name)
def test_create_snapmirror_cleanup_on_geometry_has_changed(self):
mock_dest_client = mock.Mock()
mock_src_client = mock.Mock()
self.mock_object(mock_dest_client, 'flexvol_exists',
return_value=True)
self.mock_object(mock_dest_client, 'get_snapmirrors',
return_value=None)
create_destination_flexvol = self.mock_object(
self.dm_mixin, 'create_destination_flexvol')
mock_delete_snapshot = self.mock_object(
self.dm_mixin, 'delete_snapmirror'
)
self.mock_object(utils, 'get_client_for_backend',
side_effect=[mock_dest_client,
mock_src_client])
geometry_exception_message = ("Geometry of the destination FlexGroup "
"has been changed since the SnapMirror "
"relationship was created.")
mock_dest_client.initialize_snapmirror.side_effect = [
netapp_api.NaApiError(code=netapp_api.EAPIERROR,
message=geometry_exception_message),
]
mock_provisioning_options = mock.Mock()
mock_provisioning_options.get.return_value = False
self.mock_object(mock_src_client,
'get_provisioning_options_from_flexvol',
return_value=mock_provisioning_options)
self.assertRaises(na_utils.GeometryHasChangedOnDestination,
self.dm_mixin.create_snapmirror,
self.src_backend,
self.dest_backend,
self.src_flexvol_name,
self.dest_flexvol_name)
self.assertFalse(create_destination_flexvol.called)
mock_dest_client.create_snapmirror.assert_called_once_with(
self.src_vserver, self.src_flexvol_name, self.dest_vserver,
self.dest_flexvol_name, schedule='hourly',
relationship_type='data_protection')
mock_dest_client.initialize_snapmirror.assert_called_once_with(
self.src_vserver, self.src_flexvol_name, self.dest_vserver,
self.dest_flexvol_name)
mock_delete_snapshot.assert_called_once_with(
self.src_backend, self.dest_backend, self.src_flexvol_name,
self.dest_flexvol_name)
@ddt.data('uninitialized', 'broken-off', 'snapmirrored') @ddt.data('uninitialized', 'broken-off', 'snapmirrored')
def test_create_snapmirror_snapmirror_exists_state(self, mirror_state): def test_create_snapmirror_snapmirror_exists_state(self, mirror_state):
mock_dest_client = mock.Mock() mock_dest_client = mock.Mock()
@ -223,7 +293,7 @@ class NetAppCDOTDataMotionMixinTestCase(test.TestCase):
mock_dest_client.resume_snapmirror.assert_called_once_with( mock_dest_client.resume_snapmirror.assert_called_once_with(
self.src_vserver, self.src_flexvol_name, self.src_vserver, self.src_flexvol_name,
self.dest_vserver, self.dest_flexvol_name) self.dest_vserver, self.dest_flexvol_name)
mock_dest_client.resume_snapmirror.assert_called_once_with( mock_dest_client.resync_snapmirror.assert_called_once_with(
self.src_vserver, self.src_flexvol_name, self.src_vserver, self.src_flexvol_name,
self.dest_vserver, self.dest_flexvol_name) self.dest_vserver, self.dest_flexvol_name)
@ -254,9 +324,10 @@ class NetAppCDOTDataMotionMixinTestCase(test.TestCase):
mock_dest_client.resume_snapmirror.assert_called_once_with( mock_dest_client.resume_snapmirror.assert_called_once_with(
self.src_vserver, self.src_flexvol_name, self.src_vserver, self.src_flexvol_name,
self.dest_vserver, self.dest_flexvol_name) self.dest_vserver, self.dest_flexvol_name)
mock_dest_client.resume_snapmirror.assert_called_once_with( if failed_call == 'resync_snapmirror':
self.src_vserver, self.src_flexvol_name, mock_dest_client.resync_snapmirror.assert_called_once_with(
self.dest_vserver, self.dest_flexvol_name) self.src_vserver, self.src_flexvol_name,
self.dest_vserver, self.dest_flexvol_name)
self.assertEqual(1, mock_exception_log.call_count) self.assertEqual(1, mock_exception_log.call_count)
def test_delete_snapmirror(self): def test_delete_snapmirror(self):
@ -528,6 +599,7 @@ class NetAppCDOTDataMotionMixinTestCase(test.TestCase):
self.dm_mixin.create_destination_flexvol, self.dm_mixin.create_destination_flexvol,
self.src_backend, self.dest_backend, self.src_backend, self.dest_backend,
self.src_flexvol_name, self.dest_flexvol_name) self.src_flexvol_name, self.dest_flexvol_name)
if size and is_flexgroup is False: if size and is_flexgroup is False:
self.dm_mixin._get_replication_aggregate_map.\ self.dm_mixin._get_replication_aggregate_map.\
assert_called_once_with(self.src_backend, self.dest_backend) assert_called_once_with(self.src_backend, self.dest_backend)
@ -536,9 +608,63 @@ class NetAppCDOTDataMotionMixinTestCase(test.TestCase):
self.dm_mixin._get_replication_aggregate_map.called) self.dm_mixin._get_replication_aggregate_map.called)
self.assertFalse(mock_client_call.called) self.assertFalse(mock_client_call.called)
def test_create_destination_flexvol(self): @ddt.data('mixed', None)
def test_create_destination_flexgroup_online_timeout(self, volume_state):
aggr_map = { aggr_map = {
fakes.PROVISIONING_OPTS['aggregate']: 'aggr01', fakes.PROVISIONING_OPTS['aggregate'][0]: 'aggr01',
'aggr20': 'aggr02',
}
provisioning_opts = copy.deepcopy(fakes.PROVISIONING_OPTS)
expected_prov_opts = copy.deepcopy(fakes.PROVISIONING_OPTS)
expected_prov_opts.pop('volume_type', None)
expected_prov_opts.pop('size', None)
expected_prov_opts.pop('aggregate', None)
expected_prov_opts.pop('is_flexgroup', None)
self.mock_object(
self.mock_src_client, 'get_provisioning_options_from_flexvol',
return_value=provisioning_opts)
self.mock_object(self.dm_mixin, '_get_replication_aggregate_map',
return_value=aggr_map)
self.mock_object(self.dm_mixin,
'_get_replication_volume_online_timeout',
return_value=2)
mock_create_volume_async = self.mock_object(self.mock_dest_client,
'create_volume_async')
mock_volume_state = self.mock_object(self.mock_dest_client,
'get_volume_state',
return_value=volume_state)
self.mock_object(self.mock_src_client, 'is_flexvol_encrypted',
return_value=False)
mock_dedupe_enabled = self.mock_object(
self.mock_dest_client, 'enable_volume_dedupe_async')
mock_compression_enabled = self.mock_object(
self.mock_dest_client, 'enable_volume_compression_async')
self.assertRaises(na_utils.NetAppDriverException,
self.dm_mixin.create_destination_flexvol,
self.src_backend, self.dest_backend,
self.src_flexvol_name, self.dest_flexvol_name,
pool_is_flexgroup=True)
expected_prov_opts.pop('dedupe_enabled')
expected_prov_opts.pop('compression_enabled')
mock_create_volume_async.assert_called_once_with(
self.dest_flexvol_name,
['aggr01'],
fakes.PROVISIONING_OPTS['size'],
volume_type='dp', **expected_prov_opts)
mock_volume_state.assert_called_with(
flexvol_name=self.dest_flexvol_name)
mock_dedupe_enabled.assert_not_called()
mock_compression_enabled.assert_not_called()
@ddt.data('flexvol', 'flexgroup')
def test_create_destination_flexvol(self, volume_style):
aggr_map = {
fakes.PROVISIONING_OPTS['aggregate'][0]: 'aggr01',
'aggr20': 'aggr02', 'aggr20': 'aggr02',
} }
provisioning_opts = copy.deepcopy(fakes.PROVISIONING_OPTS) provisioning_opts = copy.deepcopy(fakes.PROVISIONING_OPTS)
@ -555,27 +681,64 @@ class NetAppCDOTDataMotionMixinTestCase(test.TestCase):
return_value=False) return_value=False)
self.mock_object(self.dm_mixin, '_get_replication_aggregate_map', self.mock_object(self.dm_mixin, '_get_replication_aggregate_map',
return_value=aggr_map) return_value=aggr_map)
mock_client_call = self.mock_object(
self.mock_dest_client, 'create_flexvol') pool_is_flexgroup = False
if volume_style == 'flexgroup':
pool_is_flexgroup = True
self.mock_object(self.dm_mixin,
'_get_replication_volume_online_timeout',
return_value=2)
mock_create_volume_async = self.mock_object(self.mock_dest_client,
'create_volume_async')
mock_volume_state = self.mock_object(self.mock_dest_client,
'get_volume_state',
return_value='online')
mock_dedupe_enabled = self.mock_object(
self.mock_dest_client, 'enable_volume_dedupe_async')
mock_compression_enabled = self.mock_object(
self.mock_dest_client, 'enable_volume_compression_async')
else:
mock_create_flexvol = self.mock_object(self.mock_dest_client,
'create_flexvol')
retval = self.dm_mixin.create_destination_flexvol( retval = self.dm_mixin.create_destination_flexvol(
self.src_backend, self.dest_backend, self.src_backend, self.dest_backend,
self.src_flexvol_name, self.dest_flexvol_name) self.src_flexvol_name, self.dest_flexvol_name,
pool_is_flexgroup=pool_is_flexgroup)
self.assertIsNone(retval) self.assertIsNone(retval)
mock_get_provisioning_opts_call.assert_called_once_with( mock_get_provisioning_opts_call.assert_called_once_with(
self.src_flexvol_name) self.src_flexvol_name)
self.dm_mixin._get_replication_aggregate_map.assert_called_once_with( self.dm_mixin._get_replication_aggregate_map.assert_called_once_with(
self.src_backend, self.dest_backend) self.src_backend, self.dest_backend)
mock_client_call.assert_called_once_with(
self.dest_flexvol_name, 'aggr01', fakes.PROVISIONING_OPTS['size'], if volume_style == 'flexgroup':
volume_type='dp', **expected_prov_opts) expected_prov_opts.pop('dedupe_enabled')
expected_prov_opts.pop('compression_enabled')
mock_create_volume_async.assert_called_once_with(
self.dest_flexvol_name,
['aggr01'],
fakes.PROVISIONING_OPTS['size'],
volume_type='dp', **expected_prov_opts)
mock_volume_state.assert_called_once_with(
flexvol_name=self.dest_flexvol_name)
mock_dedupe_enabled.assert_called_once_with(
self.dest_flexvol_name)
mock_compression_enabled.assert_called_once_with(
self.dest_flexvol_name)
else:
mock_create_flexvol.assert_called_once_with(
self.dest_flexvol_name,
'aggr01',
fakes.PROVISIONING_OPTS['size'],
volume_type='dp', **expected_prov_opts)
mock_is_flexvol_encrypted.assert_called_once_with( mock_is_flexvol_encrypted.assert_called_once_with(
self.src_flexvol_name, self.src_vserver) self.src_flexvol_name, self.src_vserver)
def test_create_encrypted_destination_flexvol(self): def test_create_encrypted_destination_flexvol(self):
aggr_map = { aggr_map = {
fakes.ENCRYPTED_PROVISIONING_OPTS['aggregate']: 'aggr01', fakes.ENCRYPTED_PROVISIONING_OPTS['aggregate'][0]: 'aggr01',
'aggr20': 'aggr02', 'aggr20': 'aggr02',
} }
provisioning_opts = copy.deepcopy(fakes.ENCRYPTED_PROVISIONING_OPTS) provisioning_opts = copy.deepcopy(fakes.ENCRYPTED_PROVISIONING_OPTS)
@ -637,6 +800,33 @@ class NetAppCDOTDataMotionMixinTestCase(test.TestCase):
self.mock_src_config) self.mock_src_config)
self.dm_mixin.create_snapmirror.assert_has_calls(expected_calls) self.dm_mixin.create_snapmirror.assert_has_calls(expected_calls)
def test_ensure_snapmirrors_number_of_tries_exceeded(self):
flexvols = ['nvol1']
replication_backends = ['fallback1']
mock_error_log = self.mock_object(data_motion.LOG, 'error')
self.mock_object(self.dm_mixin, 'get_replication_backend_names',
return_value=replication_backends)
self.mock_object(self.dm_mixin, 'create_snapmirror',
side_effect=na_utils.GeometryHasChangedOnDestination)
self.assertRaises(na_utils.GeometryHasChangedOnDestination,
self.dm_mixin.ensure_snapmirrors,
self.mock_src_config,
self.src_backend,
flexvols)
self.dm_mixin.get_replication_backend_names.assert_called_once_with(
self.mock_src_config)
excepted_call = mock.call(
self.src_backend, replication_backends[0],
flexvols[0], flexvols[0])
self.dm_mixin.create_snapmirror.assert_has_calls([
excepted_call, excepted_call, excepted_call
])
mock_error_log.assert_called()
def test_break_snapmirrors(self): def test_break_snapmirrors(self):
flexvols = ['nvol1', 'nvol2'] flexvols = ['nvol1', 'nvol2']
replication_backends = ['fallback1', 'fallback2'] replication_backends = ['fallback1', 'fallback2']

View File

@ -1068,7 +1068,7 @@ class Client(client_base.Client):
volume_list.append(volume_attributes) volume_list.append(volume_attributes)
if len(volume_list) != 1: if len(volume_list) != 1:
msg = _('Could not find unique volume. Volumes foud: %(vol)s.') msg = _('Could not find unique volume. Volumes found: %(vol)s.')
msg_args = {'vol': volume_list} msg_args = {'vol': volume_list}
raise exception.VolumeBackendAPIException(data=msg % msg_args) raise exception.VolumeBackendAPIException(data=msg % msg_args)
@ -1118,6 +1118,43 @@ class Client(client_base.Client):
return volumes return volumes
def get_volume_state(self, junction_path=None, name=None):
"""Returns volume state for a given name or junction path"""
volume_id_attributes = {}
if junction_path:
volume_id_attributes['junction-path'] = junction_path
if name:
volume_id_attributes['name'] = name
api_args = {
'query': {
'volume-attributes': {
'volume-id-attributes': volume_id_attributes
}
},
'desired-attributes': {
'volume-attributes': {
'volume-id-attributes': {
'style-extended': None
},
'volume-state-attributes': {
'state': None
}
}
}
}
result = self.send_iter_request('volume-get-iter', api_args)
try:
volume_attributes = self.get_unique_volume(result)
except exception.VolumeBackendAPIException:
return None
volume_state_attributes = volume_attributes.get_child_by_name(
'volume-state-attributes') or netapp_api.NaElement('none')
volume_state = volume_state_attributes.get_child_content('state')
return volume_state
def get_flexvol(self, flexvol_path=None, flexvol_name=None): def get_flexvol(self, flexvol_path=None, flexvol_name=None):
"""Get flexvol attributes needed for the storage service catalog.""" """Get flexvol attributes needed for the storage service catalog."""
@ -1396,6 +1433,41 @@ class Client(client_base.Client):
qos_min_name = na_utils.qos_min_feature_name(is_nfs, node_name) qos_min_name = na_utils.qos_min_feature_name(is_nfs, node_name)
return getattr(self.features, qos_min_name, False).__bool__() return getattr(self.features, qos_min_name, False).__bool__()
def create_volume_async(self, name, aggregate_list, size_gb,
space_guarantee_type=None, snapshot_policy=None,
language=None, snapshot_reserve=None,
volume_type='rw'):
"""Creates a FlexGroup volume asynchronously."""
api_args = {
'aggr-list': [{'aggr-name': aggr} for aggr in aggregate_list],
'size': size_gb * units.Gi,
'volume-name': name,
'volume-type': volume_type,
}
if volume_type == 'dp':
snapshot_policy = None
else:
api_args['junction-path'] = '/%s' % name
if snapshot_policy is not None:
api_args['snapshot-policy'] = snapshot_policy
if space_guarantee_type:
api_args['space-reserve'] = space_guarantee_type
if language is not None:
api_args['language-code'] = language
if snapshot_reserve is not None:
api_args['percentage-snapshot-reserve'] = six.text_type(
snapshot_reserve)
result = self.connection.send_request('volume-create-async', api_args)
job_info = {
'status': result.get_child_content('result-status'),
'jobid': result.get_child_content('result-jobid'),
'error-code': result.get_child_content('result-error-code'),
'error-message': result.get_child_content('result-error-message')
}
return job_info
def create_flexvol(self, flexvol_name, aggregate_name, size_gb, def create_flexvol(self, flexvol_name, aggregate_name, size_gb,
space_guarantee_type=None, snapshot_policy=None, space_guarantee_type=None, snapshot_policy=None,
language=None, dedupe_enabled=False, language=None, dedupe_enabled=False,
@ -1496,6 +1568,32 @@ class Client(client_base.Client):
} }
self.connection.send_request('sis-set-config', api_args) self.connection.send_request('sis-set-config', api_args)
def enable_volume_dedupe_async(self, volume_name):
"""Enable deduplication on FlexVol/FlexGroup volume asynchronously."""
api_args = {'volume-name': volume_name}
self.connection.send_request('sis-enable-async', api_args)
def disable_volume_dedupe_async(self, volume_name):
"""Disable deduplication on FlexVol/FlexGroup volume asynchronously."""
api_args = {'volume-name': volume_name}
self.connection.send_request('sis-disable-async', api_args)
def enable_volume_compression_async(self, volume_name):
"""Enable compression on FlexVol/FlexGroup volume asynchronously."""
api_args = {
'volume-name': volume_name,
'enable-compression': 'true'
}
self.connection.send_request('sis-set-config-async', api_args)
def disable_volume_compression_async(self, volume_name):
"""Disable compression on FlexVol/FlexGroup volume asynchronously."""
api_args = {
'volume-name': volume_name,
'enable-compression': 'false'
}
self.connection.send_request('sis-set-config-async', api_args)
@volume_utils.trace_method @volume_utils.trace_method
def delete_file(self, path_to_file): def delete_file(self, path_to_file):
"""Delete file at path.""" """Delete file at path."""
@ -2140,6 +2238,7 @@ class Client(client_base.Client):
'destination-vserver': destination_vserver, 'destination-vserver': destination_vserver,
'relationship-type': relationship_type, 'relationship-type': relationship_type,
} }
if schedule: if schedule:
api_args['schedule'] = schedule api_args['schedule'] = schedule
if policy: if policy:

View File

@ -21,6 +21,7 @@ SnapMirror, and copy-offload as improvements to brute force data transfer.
""" """
from oslo_log import log from oslo_log import log
from oslo_service import loopingcall
from oslo_utils import excutils from oslo_utils import excutils
from cinder import exception from cinder import exception
@ -34,6 +35,9 @@ from cinder.volume import volume_utils
LOG = log.getLogger(__name__) LOG = log.getLogger(__name__)
ENTRY_DOES_NOT_EXIST = "(entry doesn't exist)" ENTRY_DOES_NOT_EXIST = "(entry doesn't exist)"
GEOMETRY_HAS_BEEN_CHANGED = (
"Geometry of the destination", # This intends to be a Tuple
"has been changed since the SnapMirror relationship was created")
QUIESCE_RETRY_INTERVAL = 5 QUIESCE_RETRY_INTERVAL = 5
@ -128,22 +132,33 @@ class DataMotionMixin(object):
If a SnapMirror relationship already exists and is broken off or If a SnapMirror relationship already exists and is broken off or
quiesced, resume and re-sync the mirror. quiesced, resume and re-sync the mirror.
""" """
dest_backend_config = config_utils.get_backend_configuration( dest_backend_config = config_utils.get_backend_configuration(
dest_backend_name) dest_backend_name)
dest_vserver = dest_backend_config.netapp_vserver dest_vserver = dest_backend_config.netapp_vserver
dest_client = config_utils.get_client_for_backend(
dest_backend_name, vserver_name=dest_vserver)
source_backend_config = config_utils.get_backend_configuration( source_backend_config = config_utils.get_backend_configuration(
src_backend_name) src_backend_name)
src_vserver = source_backend_config.netapp_vserver src_vserver = source_backend_config.netapp_vserver
dest_client = config_utils.get_client_for_backend(
dest_backend_name, vserver_name=dest_vserver)
src_client = config_utils.get_client_for_backend(
src_backend_name, vserver_name=src_vserver)
provisioning_options = (
src_client.get_provisioning_options_from_flexvol(
src_flexvol_name)
)
pool_is_flexgroup = provisioning_options.get('is_flexgroup', False)
# 1. Create destination 'dp' FlexVol if it doesn't exist # 1. Create destination 'dp' FlexVol if it doesn't exist
if not dest_client.flexvol_exists(dest_flexvol_name): if not dest_client.flexvol_exists(dest_flexvol_name):
self.create_destination_flexvol(src_backend_name, self.create_destination_flexvol(
dest_backend_name, src_backend_name,
src_flexvol_name, dest_backend_name,
dest_flexvol_name) src_flexvol_name,
dest_flexvol_name,
pool_is_flexgroup=pool_is_flexgroup)
# 2. Check if SnapMirror relationship exists # 2. Check if SnapMirror relationship exists
existing_mirrors = dest_client.get_snapmirrors( existing_mirrors = dest_client.get_snapmirrors(
@ -158,28 +173,48 @@ class DataMotionMixin(object):
# 3. Create and initialize SnapMirror if it doesn't already exist # 3. Create and initialize SnapMirror if it doesn't already exist
if not existing_mirrors: if not existing_mirrors:
# TODO(gouthamr): Change the schedule from hourly to a config value # TODO(gouthamr): Change the schedule from hourly to a config value
msg = ("Creating a SnapMirror relationship between " msg = ("Creating a SnapMirror relationship between "
"%(src_vserver)s:%(src_volume)s and %(dest_vserver)s:" "%(src_vserver)s:%(src_volume)s and %(dest_vserver)s:"
"%(dest_volume)s.") "%(dest_volume)s.")
LOG.debug(msg, msg_payload) LOG.debug(msg, msg_payload)
dest_client.create_snapmirror(src_vserver, try:
src_flexvol_name, dest_client.create_snapmirror(
dest_vserver, src_vserver,
dest_flexvol_name, src_flexvol_name,
schedule='hourly') dest_vserver,
dest_flexvol_name,
schedule='hourly',
relationship_type=('extended_data_protection'
if pool_is_flexgroup
else 'data_protection'))
msg = ("Initializing SnapMirror transfers between " msg = ("Initializing SnapMirror transfers between "
"%(src_vserver)s:%(src_volume)s and %(dest_vserver)s:" "%(src_vserver)s:%(src_volume)s and %(dest_vserver)s:"
"%(dest_volume)s.") "%(dest_volume)s.")
LOG.debug(msg, msg_payload) LOG.debug(msg, msg_payload)
# Initialize async transfer of the initial data # Initialize async transfer of the initial data
dest_client.initialize_snapmirror(src_vserver, dest_client.initialize_snapmirror(src_vserver,
src_flexvol_name, src_flexvol_name,
dest_vserver, dest_vserver,
dest_flexvol_name) dest_flexvol_name)
except netapp_api.NaApiError as e:
with excutils.save_and_reraise_exception() as raise_ctxt:
if (e.code == netapp_api.EAPIERROR and
all(substr in e.message for
substr in GEOMETRY_HAS_BEEN_CHANGED)):
msg = _("Error creating SnapMirror. Geometry has "
"changed on destination volume.")
LOG.error(msg)
self.delete_snapmirror(src_backend_name,
dest_backend_name,
src_flexvol_name,
dest_flexvol_name)
raise_ctxt.reraise = False
raise na_utils.GeometryHasChangedOnDestination(msg)
# 4. Try to repair SnapMirror if existing # 4. Try to repair SnapMirror if existing
else: else:
@ -191,6 +226,7 @@ class DataMotionMixin(object):
"'%(state)s' state. Attempting to repair it.") "'%(state)s' state. Attempting to repair it.")
msg_payload['state'] = snapmirror.get('mirror-state') msg_payload['state'] = snapmirror.get('mirror-state')
LOG.debug(msg, msg_payload) LOG.debug(msg, msg_payload)
dest_client.resume_snapmirror(src_vserver, dest_client.resume_snapmirror(src_vserver,
src_flexvol_name, src_flexvol_name,
dest_vserver, dest_vserver,
@ -399,7 +435,8 @@ class DataMotionMixin(object):
dest_flexvol_name) dest_flexvol_name)
def create_destination_flexvol(self, src_backend_name, dest_backend_name, def create_destination_flexvol(self, src_backend_name, dest_backend_name,
src_flexvol_name, dest_flexvol_name): src_flexvol_name, dest_flexvol_name,
pool_is_flexgroup=False):
"""Create a SnapMirror mirror target FlexVol for a given source.""" """Create a SnapMirror mirror target FlexVol for a given source."""
dest_backend_config = config_utils.get_backend_configuration( dest_backend_config = config_utils.get_backend_configuration(
dest_backend_name) dest_backend_name)
@ -417,11 +454,7 @@ class DataMotionMixin(object):
src_client.get_provisioning_options_from_flexvol( src_client.get_provisioning_options_from_flexvol(
src_flexvol_name) src_flexvol_name)
) )
provisioning_options.pop('is_flexgroup')
if provisioning_options.pop('is_flexgroup', False):
msg = _("Destination volume cannot be created as FlexGroup for "
"replication, it must already exist there.")
raise na_utils.NetAppDriverException(msg)
# If the source is encrypted then the destination needs to be # If the source is encrypted then the destination needs to be
# encrypted too. Using is_flexvol_encrypted because it includes # encrypted too. Using is_flexvol_encrypted because it includes
@ -441,23 +474,65 @@ class DataMotionMixin(object):
aggregate_map = self._get_replication_aggregate_map( aggregate_map = self._get_replication_aggregate_map(
src_backend_name, dest_backend_name) src_backend_name, dest_backend_name)
if not aggregate_map.get(source_aggregate): destination_aggregate = []
msg = _("Unable to find configuration matching the source " for src_aggr in source_aggregate:
"aggregate (%s) and the destination aggregate. Option " dst_aggr = aggregate_map.get(src_aggr, None)
"netapp_replication_aggregate_map may be incorrect.") if dst_aggr:
raise na_utils.NetAppDriverException( destination_aggregate.append(dst_aggr)
message=msg % source_aggregate) else:
msg = _("Unable to find configuration matching the source "
destination_aggregate = aggregate_map[source_aggregate] "aggregate and the destination aggregate. Option "
"netapp_replication_aggregate_map may be incorrect.")
raise na_utils.NetAppDriverException(message=msg)
# NOTE(gouthamr): The volume is intentionally created as a Data # NOTE(gouthamr): The volume is intentionally created as a Data
# Protection volume; junction-path will be added on breaking # Protection volume; junction-path will be added on breaking
# the mirror. # the mirror.
provisioning_options['volume_type'] = 'dp' provisioning_options['volume_type'] = 'dp'
dest_client.create_flexvol(dest_flexvol_name,
destination_aggregate, if pool_is_flexgroup:
size, compression_enabled = provisioning_options.pop(
**provisioning_options) 'compression_enabled', False)
# cDOT compression requires that deduplication be enabled.
dedupe_enabled = provisioning_options.pop(
'dedupe_enabled', False) or compression_enabled
dest_client.create_volume_async(
dest_flexvol_name,
destination_aggregate,
size,
**provisioning_options)
timeout = self._get_replication_volume_online_timeout()
def _wait_volume_is_online():
volume_state = dest_client.get_volume_state(
flexvol_name=dest_flexvol_name)
if volume_state and volume_state == 'online':
raise loopingcall.LoopingCallDone()
try:
wait_call = loopingcall.FixedIntervalWithTimeoutLoopingCall(
_wait_volume_is_online)
wait_call.start(interval=5, timeout=timeout).wait()
if dedupe_enabled:
dest_client.enable_volume_dedupe_async(
dest_flexvol_name)
if compression_enabled:
dest_client.enable_volume_compression_async(
dest_flexvol_name)
except loopingcall.LoopingCallTimeOut:
msg = _("Timeout waiting destination FlexGroup to to come "
"online.")
raise na_utils.NetAppDriverException(msg)
else:
dest_client.create_flexvol(dest_flexvol_name,
destination_aggregate[0],
size,
**provisioning_options)
def ensure_snapmirrors(self, config, src_backend_name, src_flexvol_names): def ensure_snapmirrors(self, config, src_backend_name, src_flexvol_names):
"""Ensure all the SnapMirrors needed for whole-backend replication.""" """Ensure all the SnapMirrors needed for whole-backend replication."""
@ -467,10 +542,24 @@ class DataMotionMixin(object):
dest_flexvol_name = src_flexvol_name dest_flexvol_name = src_flexvol_name
self.create_snapmirror(src_backend_name, retry_exceptions = (
dest_backend_name, na_utils.GeometryHasChangedOnDestination,
src_flexvol_name, )
dest_flexvol_name)
@utils.retry(retry_exceptions,
interval=30, retries=6, backoff_rate=1)
def _try_create_snapmirror():
self.create_snapmirror(src_backend_name,
dest_backend_name,
src_flexvol_name,
dest_flexvol_name)
try:
_try_create_snapmirror()
except na_utils.NetAppDriverException as e:
with excutils.save_and_reraise_exception():
if isinstance(e, retry_exceptions):
LOG.error("Number of tries exceeded "
"while trying to create SnapMirror.")
def break_snapmirrors(self, config, src_backend_name, src_flexvol_names, def break_snapmirrors(self, config, src_backend_name, src_flexvol_names,
chosen_target): chosen_target):
@ -648,3 +737,6 @@ class DataMotionMixin(object):
self.failed_over_backend_name = active_backend_name self.failed_over_backend_name = active_backend_name
return active_backend_name, volume_updates, [] return active_backend_name, volume_updates, []
def _get_replication_volume_online_timeout(self):
return self.configuration.netapp_replication_volume_online_timeout

View File

@ -150,9 +150,10 @@ netapp_replication_opts = [
"mapping between source and destination back ends when " "mapping between source and destination back ends when "
"using whole back end replication. For every " "using whole back end replication. For every "
"source aggregate associated with a cinder pool (NetApp " "source aggregate associated with a cinder pool (NetApp "
"FlexVol), you would need to specify the destination " "FlexVol/FlexGroup), you would need to specify the "
"aggregate on the replication target device. A " "destination aggregate on the replication target "
"replication target device is configured with the " "device. "
"A replication target device is configured with the "
"configuration option replication_device. Specify this " "configuration option replication_device. Specify this "
"option as many times as you have replication devices. " "option as many times as you have replication devices. "
"Each entry takes the standard dict config form: " "Each entry takes the standard dict config form: "
@ -165,7 +166,12 @@ netapp_replication_opts = [
default=3600, # One Hour default=3600, # One Hour
help='The maximum time in seconds to wait for existing ' help='The maximum time in seconds to wait for existing '
'SnapMirror transfers to complete before aborting ' 'SnapMirror transfers to complete before aborting '
'during a failover.'), ] 'during a failover.'),
cfg.IntOpt('netapp_replication_volume_online_timeout',
min=60,
default=360, # Default to six minutes
help='Sets time in seconds to wait for a replication volume '
'create to complete and go online.')]
netapp_support_opts = [ netapp_support_opts = [
cfg.StrOpt('netapp_api_trace_pattern', cfg.StrOpt('netapp_api_trace_pattern',

View File

@ -89,6 +89,10 @@ class NetAppDriverException(exception.VolumeDriverException):
message = _("NetApp Cinder Driver exception.") message = _("NetApp Cinder Driver exception.")
class GeometryHasChangedOnDestination(NetAppDriverException):
message = _("Geometry has changed on destination volume.")
def validate_instantiation(**kwargs): def validate_instantiation(**kwargs):
"""Checks if a driver is instantiated other than by the unified driver. """Checks if a driver is instantiated other than by the unified driver.