Merge "NetApp ONTAP: Add volume replication functions on REST client"

cinder/tests/unit/volume/drivers/netapp/dataontap/client/fakes.py

@@ -2529,10 +2529,10 @@ FAKE_FORMATTED_HTTP_QUERY = '?type=fake_type'
 
 JOB_RESPONSE_REST = {
     "job": {
-        "uuid": "uuid-12345",
+        "uuid": FAKE_UUID,
         "_links": {
             "self": {
-                "href": "/api/cluster/jobs/uuid-12345"
+                "href": f"/api/cluster/jobs/{FAKE_UUID}"
             }
         }
     }
@@ -2843,6 +2843,32 @@ GET_LUN_MAPS = {
     "num_records": 1,
 }
 
+SNAPMIRROR_GET_ITER_RESPONSE_REST = {
+    "records": [
+        {
+            "uuid": FAKE_UUID,
+            "source": {
+                "path": SM_SOURCE_VSERVER + ':' + SM_SOURCE_VOLUME,
+                "svm": {
+                    "name": SM_SOURCE_VSERVER
+                }
+            },
+            "destination": {
+                "path": SM_DEST_VSERVER + ':' + SM_DEST_VOLUME,
+                "svm": {
+                    "name": SM_DEST_VSERVER
+                }
+            },
+            "policy": {
+                "type": "async"
+            },
+            "state": "snapmirrored",
+            "healthy": True
+        }
+    ],
+    "num_records": 1,
+}
+
 GET_LUN_MAPS_NO_MAPS = {
     "records": [
         {
@@ -2921,3 +2947,50 @@ VOLUME_GET_ITER_CAPACITY_RESPONSE_REST = {
     ],
     "num_records": 1,
 }
+
+REST_GET_SNAPMIRRORS_RESPONSE = [{
+    'destination-volume': SM_DEST_VOLUME,
+    'destination-vserver': SM_DEST_VSERVER,
+    'is-healthy': True,
+    'lag-time': None,
+    'last-transfer-end-timestamp': None,
+    'mirror-state': 'snapmirrored',
+    'relationship-status': 'snapmirrored',
+    'source-volume': SM_SOURCE_VOLUME,
+    'source-vserver': SM_SOURCE_VSERVER,
+    'uuid': FAKE_UUID,
+}]
+
+TRANSFERS_GET_ITER_REST = {
+    "records": [
+        {
+            "uuid": FAKE_UUID,
+            "state": "transferring"
+        },
+        {
+            "uuid": FAKE_UUID,
+            "state": "failed"
+        }
+    ],
+    "num_records": 2,
+}
+
+JOB_SUCCESSFUL_REST = {
+    "uuid": FAKE_UUID,
+    "description": "Fake description",
+    "state": "success",
+    "message": "success",
+    "code": 0,
+    "start_time": "2022-02-18T20:08:03+00:00",
+    "end_time": "2022-02-18T20:08:04+00:00",
+}
+
+JOB_ERROR_REST = {
+    "uuid": FAKE_UUID,
+    "description": "Fake description",
+    "state": "failure",
+    "message": "failure",
+    "code": -1,
+    "start_time": "2022-02-18T20:08:03+00:00",
+    "end_time": "2022-02-18T20:08:04+00:00",
+}

cinder/tests/unit/volume/drivers/netapp/dataontap/client/test_client_cmode_rest.py

@@ -20,6 +20,7 @@ from unittest import mock
 import uuid
 
 import ddt
+from oslo_utils import units
 import six
 
 from cinder import exception
@@ -995,10 +996,10 @@ class NetAppRestCmodeClientTestCase(test.TestCase):
     def test_get_operational_lif_addresses(self):
         expected_result = ['1.2.3.4', '99.98.97.96']
         api_response = fake_client.GET_OPERATIONAL_LIF_ADDRESSES_RESPONSE_REST
+
         mock_send_request = self.mock_object(self.client,
                                              'send_request',
                                              return_value=api_response)
-
         address_list = self.client.get_operational_lif_addresses()
 
         query = {
@@ -2503,3 +2504,879 @@ class NetAppRestCmodeClientTestCase(test.TestCase):
         result = self.client.check_cluster_api(endpoint_api)
 
         self.assertFalse(result)
+
+    def test_get_provisioning_options_from_flexvol(self):
+
+        self.mock_object(self.client, 'get_flexvol',
+                         return_value=fake_client.VOLUME_INFO_SSC)
+        self.mock_object(self.client, 'get_flexvol_dedupe_info',
+                         return_value=fake_client.VOLUME_DEDUPE_INFO_SSC)
+
+        expected_prov_opts = {
+            'aggregate': 'fake_aggr1',
+            'compression_enabled': False,
+            'dedupe_enabled': True,
+            'language': 'c.utf_8',
+            'size': 1,
+            'snapshot_policy': 'default',
+            'snapshot_reserve': '5',
+            'space_guarantee_type': 'none',
+            'volume_type': 'rw',
+            'is_flexgroup': False,
+        }
+
+        actual_prov_opts = self.client.get_provisioning_options_from_flexvol(
+            fake_client.VOLUME_NAME)
+
+        self.assertEqual(expected_prov_opts, actual_prov_opts)
+
+    def test_flexvol_exists(self):
+
+        api_response = fake_client.GET_NUM_RECORDS_RESPONSE_REST
+        mock_send_request = self.mock_object(self.client,
+                                             'send_request',
+                                             return_value=api_response)
+
+        result = self.client.flexvol_exists(fake_client.VOLUME_NAME)
+
+        query = {
+            'name': fake_client.VOLUME_NAME,
+            'return_records': 'false'
+        }
+
+        mock_send_request.assert_has_calls([
+            mock.call('/storage/volumes/', 'get', query=query)])
+        self.assertTrue(result)
+
+    def test_flexvol_exists_not_found(self):
+
+        api_response = fake_client.NO_RECORDS_RESPONSE_REST
+        self.mock_object(self.client,
+                         'send_request',
+                         return_value=api_response)
+
+        self.assertFalse(self.client.flexvol_exists(fake_client.VOLUME_NAME))
+
+    @ddt.data(fake_client.VOLUME_AGGREGATE_NAME,
+              [fake_client.VOLUME_AGGREGATE_NAME],
+              [fake_client.VOLUME_AGGREGATE_NAMES[0],
+               fake_client.VOLUME_AGGREGATE_NAMES[1]])
+    def test_create_volume_async(self, aggregates):
+        self.mock_object(self.client, 'send_request')
+
+        self.client.create_volume_async(
+            fake_client.VOLUME_NAME, aggregates, 100, volume_type='dp')
+
+        body = {
+            'name': fake_client.VOLUME_NAME,
+            'size': 100 * units.Gi,
+            'type': 'dp'
+        }
+
+        if isinstance(aggregates, list):
+            body['style'] = 'flexgroup'
+            body['aggregates'] = [{'name': aggr} for aggr in aggregates]
+        else:
+            body['style'] = 'flexvol'
+            body['aggregates'] = [{'name': aggregates}]
+
+        self.client.send_request.assert_called_once_with(
+            '/storage/volumes/', 'post', body=body, wait_on_accepted=False)
+
+    @ddt.data('dp', 'rw', None)
+    def test_create_volume_async_with_extra_specs(self, volume_type):
+        self.mock_object(self.client, 'send_request')
+
+        aggregates = [fake_client.VOLUME_AGGREGATE_NAME]
+        snapshot_policy = 'default'
+        size = 100
+        space_guarantee_type = 'volume'
+        language = 'en-US'
+        snapshot_reserve = 15
+
+        self.client.create_volume_async(
+            fake_client.VOLUME_NAME, aggregates, size,
+            space_guarantee_type=space_guarantee_type, language=language,
+            snapshot_policy=snapshot_policy, snapshot_reserve=snapshot_reserve,
+            volume_type=volume_type)
+
+        body = {
+            'name': fake_client.VOLUME_NAME,
+            'size': size * units.Gi,
+            'type': volume_type,
+            'guarantee': {'type': space_guarantee_type},
+            'space': {'snapshot': {'reserve_percent': str(snapshot_reserve)}},
+            'language': language,
+        }
+
+        if isinstance(aggregates, list):
+            body['style'] = 'flexgroup'
+            body['aggregates'] = [{'name': aggr} for aggr in aggregates]
+        else:
+            body['style'] = 'flexvol'
+            body['aggregates'] = [{'name': aggregates}]
+
+        if volume_type == 'dp':
+            snapshot_policy = None
+        else:
+            body['nas'] = {'path': '/%s' % fake_client.VOLUME_NAME}
+
+        if snapshot_policy is not None:
+            body['snapshot_policy'] = {'name': snapshot_policy}
+
+        self.client.send_request.assert_called_once_with(
+            '/storage/volumes/', 'post', body=body, wait_on_accepted=False)
+
+    def test_create_flexvol(self):
+        aggregates = [fake_client.VOLUME_AGGREGATE_NAME]
+        size = 100
+
+        mock_response = {
+            'job': {
+                'uuid': fake.JOB_UUID,
+            }
+        }
+
+        self.mock_object(self.client, 'send_request',
+                         return_value=mock_response)
+
+        expected_response = {
+            'status': None,
+            'jobid': fake.JOB_UUID,
+            'error-code': None,
+            'error-message': None
+        }
+
+        response = self.client.create_volume_async(fake_client.VOLUME_NAME,
+                                                   aggregates, size_gb = size)
+        self.assertEqual(expected_response, response)
+
+    def test_enable_volume_dedupe_async(self):
+        query = {
+            'name': fake_client.VOLUME_NAME,
+            'fields': 'uuid,style',
+        }
+
+        # This is needed because the first calling to send_request inside
+        # enable_volume_dedupe_async must return a valid uuid for the given
+        # volume name.
+        mock_response = {
+            'records': [
+                {
+                    'uuid': fake.JOB_UUID,
+                    'name': fake_client.VOLUME_NAME,
+                    "style": 'flexgroup',
+                }
+            ],
+            "num_records": 1,
+        }
+
+        body = {
+            'efficiency': {'dedupe': 'background'}
+        }
+
+        mock_send_request = self.mock_object(self.client, 'send_request',
+                                             return_value=mock_response)
+
+        call_list = [mock.call('/storage/volumes/',
+                               'patch', body=body, query=query,
+                               wait_on_accepted=False)]
+
+        self.client.enable_volume_dedupe_async(fake_client.VOLUME_NAME)
+        mock_send_request.assert_has_calls(call_list)
+
+    def test_enable_volume_compression_async(self):
+        query = {
+            'name': fake_client.VOLUME_NAME,
+        }
+
+        # This is needed because the first calling to send_request inside
+        # enable_volume_compression_async must return a valid uuid for the
+        # given volume name.
+        mock_response = {
+            'records': [
+                {
+                    'uuid': fake.JOB_UUID,
+                    'name': fake_client.VOLUME_NAME,
+                    "style": 'flexgroup',
+                }
+            ],
+            "num_records": 1,
+        }
+
+        body = {
+            'efficiency': {'compression': 'background'}
+        }
+
+        mock_send_request = self.mock_object(self.client, 'send_request',
+                                             return_value=mock_response)
+
+        call_list = [mock.call('/storage/volumes/',
+                               'patch', body=body, query=query,
+                               wait_on_accepted=False)]
+
+        self.client.enable_volume_compression_async(fake_client.VOLUME_NAME)
+        mock_send_request.assert_has_calls(call_list)
+
+    def test__get_snapmirrors(self):
+
+        api_response = fake_client.SNAPMIRROR_GET_ITER_RESPONSE_REST
+        mock_send_request = self.mock_object(self.client,
+                                             'send_request',
+                                             return_value=api_response)
+
+        result = self.client._get_snapmirrors(
+            fake_client.SM_SOURCE_VSERVER, fake_client.SM_SOURCE_VOLUME,
+            fake_client.SM_DEST_VSERVER, fake_client.SM_DEST_VOLUME)
+
+        query = {
+            'source.path': (fake_client.SM_SOURCE_VSERVER + ':' +
+                            fake_client.SM_SOURCE_VOLUME),
+            'destination.path': (fake_client.SM_DEST_VSERVER +
+                                 ':' + fake_client.SM_DEST_VOLUME),
+            'fields': 'state,source.svm.name,source.path,destination.svm.name,'
+                      'destination.path,transfer.end_time,lag_time,healthy,'
+                      'uuid'
+        }
+
+        mock_send_request.assert_called_once_with('/snapmirror/relationships',
+                                                  'get', query=query)
+        self.assertEqual(1, len(result))
+
+    def test__get_snapmirrors_not_found(self):
+
+        api_response = fake_client.NO_RECORDS_RESPONSE_REST
+        mock_send_request = self.mock_object(self.client,
+                                             'send_request',
+                                             return_value=api_response)
+
+        result = self.client._get_snapmirrors(
+            fake_client.SM_SOURCE_VSERVER, fake_client.SM_SOURCE_VOLUME,
+            fake_client.SM_DEST_VSERVER, fake_client.SM_DEST_VOLUME)
+
+        query = {
+            'source.path': (fake_client.SM_SOURCE_VSERVER + ':' +
+                            fake_client.SM_SOURCE_VOLUME),
+            'destination.path': (fake_client.SM_DEST_VSERVER +
+                                 ':' + fake_client.SM_DEST_VOLUME),
+            'fields': 'state,source.svm.name,source.path,destination.svm.name,'
+                      'destination.path,transfer.end_time,lag_time,healthy,'
+                      'uuid'
+        }
+
+        mock_send_request.assert_called_once_with('/snapmirror/relationships',
+                                                  'get', query=query)
+        self.assertEqual([], result)
+
+    def test_get_snapmirrors(self):
+
+        api_response = fake_client.SNAPMIRROR_GET_ITER_RESPONSE_REST
+        mock_send_request = self.mock_object(self.client,
+                                             'send_request',
+                                             return_value=api_response)
+
+        result = self.client.get_snapmirrors(
+            fake_client.SM_SOURCE_VSERVER, fake_client.SM_SOURCE_VOLUME,
+            fake_client.SM_DEST_VSERVER, fake_client.SM_DEST_VOLUME)
+
+        expected = fake_client.REST_GET_SNAPMIRRORS_RESPONSE
+
+        query = {
+            'source.path': (fake_client.SM_SOURCE_VSERVER + ':' +
+                            fake_client.SM_SOURCE_VOLUME),
+            'destination.path': (fake_client.SM_DEST_VSERVER +
+                                 ':' + fake_client.SM_DEST_VOLUME),
+            'fields': 'state,source.svm.name,source.path,destination.svm.name,'
+                      'destination.path,transfer.end_time,lag_time,healthy,'
+                      'uuid'
+        }
+
+        mock_send_request.assert_called_once_with('/snapmirror/relationships',
+                                                  'get', query=query)
+        self.assertEqual(expected, result)
+
+    @ddt.data({'policy': 'fake_policy'},
+              {'policy': None})
+    @ddt.unpack
+    def test_create_snapmirror(self, policy):
+        api_responses = [
+            {
+                "job": {
+                    "uuid": fake_client.FAKE_UUID,
+                },
+            },
+        ]
+        self.mock_object(self.client, 'send_request',
+                         side_effect = copy.deepcopy(api_responses))
+        self.client.create_snapmirror(
+            fake_client.SM_SOURCE_VSERVER, fake_client.SM_SOURCE_VOLUME,
+            fake_client.SM_DEST_VSERVER, fake_client.SM_DEST_VOLUME,
+            policy=policy)
+
+        body = {
+            'source': {
+                'path': (fake_client.SM_SOURCE_VSERVER + ':' +
+                         fake_client.SM_SOURCE_VOLUME),
+            },
+            'destination': {
+                'path': (fake_client.SM_DEST_VSERVER + ':' +
+                         fake_client.SM_DEST_VOLUME)
+            }
+        }
+
+        if policy:
+            body['policy'] = {'name': policy}
+        self.client.send_request.assert_has_calls([
+            mock.call('/snapmirror/relationships/', 'post', body=body)])
+
+    def test_create_snapmirror_already_exists(self):
+        api_responses = netapp_api.NaApiError(
+            code=netapp_api.REST_ERELATION_EXISTS)
+        self.mock_object(self.client, 'send_request',
+                         side_effect=api_responses)
+
+        response = self.client.create_snapmirror(
+            fake_client.SM_SOURCE_VSERVER,
+            fake_client.SM_SOURCE_VOLUME,
+            fake_client.SM_DEST_VSERVER,
+            fake_client.SM_DEST_VOLUME,
+            schedule=None,
+            policy=None,
+            relationship_type='data_protection')
+        self.assertIsNone(response)
+        self.assertTrue(self.client.send_request.called)
+
+    def test_create_snapmirror_error(self):
+        self.mock_object(self.client, 'send_request',
+                         side_effect=netapp_api.NaApiError(code=123))
+
+        self.assertRaises(netapp_api.NaApiError,
+                          self.client.create_snapmirror,
+                          fake_client.SM_SOURCE_VSERVER,
+                          fake_client.SM_SOURCE_VOLUME,
+                          fake_client.SM_DEST_VSERVER,
+                          fake_client.SM_DEST_VOLUME,
+                          schedule=None,
+                          policy=None,
+                          relationship_type='data_protection')
+        self.assertTrue(self.client.send_request.called)
+
+    def test__set_snapmirror_state(self):
+
+        api_responses = [
+            fake_client.SNAPMIRROR_GET_ITER_RESPONSE_REST,
+            {
+                "job":
+                {
+                    "uuid": fake_client.FAKE_UUID
+                },
+                "num_records": 1
+            }
+        ]
+
+        expected_body = {'state': 'snapmirrored'}
+        self.mock_object(self.client,
+                         'send_request',
+                         side_effect=copy.deepcopy(api_responses))
+
+        result = self.client._set_snapmirror_state(
+            'snapmirrored',
+            fake_client.SM_SOURCE_VSERVER, fake_client.SM_SOURCE_VOLUME,
+            fake_client.SM_DEST_VSERVER, fake_client.SM_DEST_VOLUME)
+
+        self.client.send_request.assert_has_calls([
+            mock.call('/snapmirror/relationships/' + fake_client.FAKE_UUID,
+                      'patch', body=expected_body, wait_on_accepted=True)])
+
+        expected = {
+            'operation-id': None,
+            'status': None,
+            'jobid': fake_client.FAKE_UUID,
+            'error-code': None,
+            'error-message': None,
+            'relationship-uuid': fake_client.FAKE_UUID
+        }
+        self.assertEqual(expected, result)
+
+    def test_initialize_snapmirror(self):
+
+        expected_job = {
+            'operation-id': None,
+            'status': None,
+            'jobid': fake_client.FAKE_UUID,
+            'error-code': None,
+            'error-message': None,
+        }
+
+        mock_set_snapmirror_state = self.mock_object(
+            self.client,
+            '_set_snapmirror_state',
+            return_value=expected_job)
+
+        result = self.client.initialize_snapmirror(
+            fake_client.SM_SOURCE_VSERVER, fake_client.SM_SOURCE_VOLUME,
+            fake_client.SM_DEST_VSERVER, fake_client.SM_DEST_VOLUME)
+
+        mock_set_snapmirror_state.assert_called_once_with(
+            'snapmirrored',
+            fake_client.SM_SOURCE_VSERVER, fake_client.SM_SOURCE_VOLUME,
+            fake_client.SM_DEST_VSERVER, fake_client.SM_DEST_VOLUME,
+            wait_result=False)
+
+        self.assertEqual(expected_job, result)
+
+    @ddt.data(True, False)
+    def test_abort_snapmirror(self, clear_checkpoint):
+
+        self.mock_object(
+            self.client, 'get_snapmirrors',
+            return_value=fake_client.REST_GET_SNAPMIRRORS_RESPONSE)
+        responses = [fake_client.TRANSFERS_GET_ITER_REST, None, None]
+        self.mock_object(self.client, 'send_request',
+                         side_effect=copy.deepcopy(responses))
+
+        self.client.abort_snapmirror(
+            fake_client.SM_SOURCE_VSERVER, fake_client.SM_SOURCE_VOLUME,
+            fake_client.SM_DEST_VSERVER, fake_client.SM_DEST_VOLUME,
+            clear_checkpoint=clear_checkpoint)
+
+        body = {'state': 'hard_aborted' if clear_checkpoint else 'aborted'}
+        query = {'state': 'transferring'}
+        self.client.send_request.assert_has_calls([
+            mock.call('/snapmirror/relationships/' +
+                      fake_client.FAKE_UUID + '/transfers/', 'get',
+                      query=query),
+            mock.call('/snapmirror/relationships/' +
+                      fake_client.FAKE_UUID + '/transfers/' +
+                      fake_client.FAKE_UUID, 'patch', body=body)])
+        self.client.get_snapmirrors.assert_called_once_with(
+            fake_client.SM_SOURCE_VSERVER, fake_client.SM_SOURCE_VOLUME,
+            fake_client.SM_DEST_VSERVER, fake_client.SM_DEST_VOLUME)
+
+    def test_abort_snapmirror_no_transfer_in_progress(self):
+
+        self.mock_object(self.client, 'send_request',
+                         return_value=fake_client.NO_RECORDS_RESPONSE_REST)
+        self.mock_object(
+            self.client, 'get_snapmirrors',
+            return_value=fake_client.REST_GET_SNAPMIRRORS_RESPONSE)
+
+        self.assertRaises(netapp_api.NaApiError,
+                          self.client.abort_snapmirror,
+                          fake_client.SM_SOURCE_VSERVER,
+                          fake_client.SM_SOURCE_VOLUME,
+                          fake_client.SM_DEST_VSERVER,
+                          fake_client.SM_DEST_VOLUME,
+                          clear_checkpoint=True)
+
+        query = {'state': 'transferring'}
+        self.client.send_request.assert_has_calls([
+            mock.call('/snapmirror/relationships/' + fake_client.FAKE_UUID +
+                      '/transfers/', 'get', query=query)])
+
+    def test_delete_snapmirror(self):
+
+        response_list = [fake_client.SNAPMIRROR_GET_ITER_RESPONSE_REST,
+                         fake_client.JOB_RESPONSE_REST,
+                         fake_client.JOB_SUCCESSFUL_REST]
+
+        self.mock_object(self.client, 'send_request',
+                         side_effect=copy.deepcopy(response_list))
+
+        self.client.delete_snapmirror(
+            fake_client.SM_SOURCE_VSERVER, fake_client.SM_SOURCE_VOLUME,
+            fake_client.SM_DEST_VSERVER, fake_client.SM_DEST_VOLUME)
+
+        query_uuid = {}
+        query_uuid['source.path'] = (fake_client.SM_SOURCE_VSERVER + ':' +
+                                     fake_client.SM_SOURCE_VOLUME)
+        query_uuid['destination.path'] = (fake_client.SM_DEST_VSERVER + ':' +
+                                          fake_client.SM_DEST_VOLUME)
+        query_uuid['fields'] = 'uuid'
+
+        query_delete = {"destination_only": "true"}
+        self.client.send_request.assert_has_calls([
+            mock.call('/snapmirror/relationships/', 'get', query=query_uuid),
+            mock.call('/snapmirror/relationships/' + fake_client.FAKE_UUID,
+                      'delete', query=query_delete)])
+
+    def test_delete_snapmirror_timeout(self):
+        # when a timeout happens, an exception is thrown by send_request
+        api_error = netapp_api.NaRetryableError()
+        self.mock_object(self.client, 'send_request',
+                         side_effect=api_error)
+
+        self.assertRaises(netapp_api.NaRetryableError,
+                          self.client.delete_snapmirror,
+                          fake_client.SM_SOURCE_VSERVER,
+                          fake_client.SM_SOURCE_VOLUME,
+                          fake_client.SM_DEST_VSERVER,
+                          fake_client.SM_DEST_VOLUME)
+
+    @ddt.data('async', 'sync')
+    def test_resume_snapmirror(self, snapmirror_policy):
+        snapmirror_response = copy.deepcopy(
+            fake_client.SNAPMIRROR_GET_ITER_RESPONSE_REST)
+        snapmirror_response['records'][0]['policy'] = {
+            'type': snapmirror_policy}
+
+        if snapmirror_policy == 'async':
+            snapmirror_response['state'] = 'snapmirrored'
+        elif snapmirror_policy == 'sync':
+            snapmirror_response['state'] = 'in_sync'
+
+        response_list = [snapmirror_response,
+                         fake_client.JOB_RESPONSE_REST,
+                         snapmirror_response]
+
+        self.mock_object(self.client, 'send_request',
+                         side_effect=copy.deepcopy(response_list))
+
+        self.client.resync_snapmirror(
+            fake_client.SM_SOURCE_VSERVER, fake_client.SM_SOURCE_VOLUME,
+            fake_client.SM_DEST_VSERVER, fake_client.SM_DEST_VOLUME)
+
+        query_uuid = {}
+        query_uuid['source.path'] = (fake_client.SM_SOURCE_VSERVER + ':' +
+                                     fake_client.SM_SOURCE_VOLUME)
+        query_uuid['destination.path'] = (fake_client.SM_DEST_VSERVER + ':' +
+                                          fake_client.SM_DEST_VOLUME)
+        query_uuid['fields'] = 'uuid,policy.type'
+
+        body_resync = {}
+        if snapmirror_policy == 'async':
+            body_resync['state'] = 'snapmirrored'
+        elif snapmirror_policy == 'sync':
+            body_resync['state'] = 'in_sync'
+
+        self.client.send_request.assert_has_calls([
+            mock.call('/snapmirror/relationships/', 'get', query=query_uuid),
+            mock.call('/snapmirror/relationships/' + fake_client.FAKE_UUID,
+                      'patch', body=body_resync)])
+
+    def test_resume_snapmirror_not_found(self):
+        query_uuid = {}
+        query_uuid['source.path'] = (fake_client.SM_SOURCE_VSERVER + ':' +
+                                     fake_client.SM_SOURCE_VOLUME)
+        query_uuid['destination.path'] = (fake_client.SM_DEST_VSERVER + ':' +
+                                          fake_client.SM_DEST_VOLUME)
+        query_uuid['fields'] = 'uuid,policy.type'
+
+        self.mock_object(
+            self.client, 'send_request',
+            return_value={'records': []})
+
+        self.assertRaises(
+            netapp_api.NaApiError,
+            self.client.resume_snapmirror,
+            fake_client.SM_SOURCE_VSERVER,
+            fake_client.SM_SOURCE_VOLUME,
+            fake_client.SM_DEST_VSERVER,
+            fake_client.SM_DEST_VOLUME)
+
+        self.client.send_request.assert_called_once_with(
+            '/snapmirror/relationships/', 'get', query=query_uuid)
+
+    def test_resume_snapmirror_api_error(self):
+        query_resume = {}
+        query_resume['source.path'] = (fake_client.SM_SOURCE_VSERVER + ':' +
+                                       fake_client.SM_SOURCE_VOLUME)
+        query_resume['destination.path'] = (fake_client.SM_DEST_VSERVER + ':' +
+                                            fake_client.SM_DEST_VOLUME)
+
+        query_uuid = copy.deepcopy(query_resume)
+        query_uuid['fields'] = 'uuid,policy.type'
+
+        api_error = netapp_api.NaApiError(code=0)
+        self.mock_object(
+            self.client, 'send_request',
+            side_effect=[fake_client.SNAPMIRROR_GET_ITER_RESPONSE_REST,
+                         api_error])
+
+        self.assertRaises(netapp_api.NaApiError,
+                          self.client.resume_snapmirror,
+                          fake_client.SM_SOURCE_VSERVER,
+                          fake_client.SM_SOURCE_VOLUME,
+                          fake_client.SM_DEST_VSERVER,
+                          fake_client.SM_DEST_VOLUME)
+
+    @ddt.data(True, False)
+    def test_release_snapmirror(self, relationship_info_only):
+
+        response_list = [fake_client.SNAPMIRROR_GET_ITER_RESPONSE_REST,
+                         fake_client.JOB_RESPONSE_REST,
+                         fake_client.JOB_SUCCESSFUL_REST]
+
+        self.mock_object(self.client, 'send_request',
+                         side_effect=copy.deepcopy(response_list))
+
+        self.client.release_snapmirror(
+            fake_client.SM_SOURCE_VSERVER, fake_client.SM_SOURCE_VOLUME,
+            fake_client.SM_DEST_VSERVER, fake_client.SM_DEST_VOLUME,
+            relationship_info_only)
+
+        query_uuid = {}
+        query_uuid['list_destinations_only'] = 'true'
+        query_uuid['source.path'] = (fake_client.SM_SOURCE_VSERVER + ':' +
+                                     fake_client.SM_SOURCE_VOLUME)
+        query_uuid['destination.path'] = (fake_client.SM_DEST_VSERVER + ':' +
+                                          fake_client.SM_DEST_VOLUME)
+        query_uuid['fields'] = 'uuid'
+
+        query_release = {}
+        if relationship_info_only:
+            # release WITHOUT removing related snapshots
+            query_release['source_info_only'] = 'true'
+        else:
+            # release and REMOVING all related snapshots
+            query_release['source_only'] = 'true'
+
+        self.client.send_request.assert_has_calls([
+            mock.call('/snapmirror/relationships/', 'get', query=query_uuid),
+            mock.call('/snapmirror/relationships/' + fake_client.FAKE_UUID,
+                      'delete', query=query_release)])
+
+    def test_release_snapmirror_timeout(self):
+        # when a timeout happens, an exception is thrown by send_request
+        api_error = netapp_api.NaRetryableError()
+        self.mock_object(self.client, 'send_request',
+                         side_effect=api_error)
+
+        self.assertRaises(netapp_api.NaRetryableError,
+                          self.client.release_snapmirror,
+                          fake_client.SM_SOURCE_VSERVER,
+                          fake_client.SM_SOURCE_VOLUME,
+                          fake_client.SM_DEST_VSERVER,
+                          fake_client.SM_DEST_VOLUME)
+
+    @ddt.data('async', 'sync')
+    def test_resync_snapmirror(self, snapmirror_policy):
+
+        snapmirror_response = copy.deepcopy(
+            fake_client.SNAPMIRROR_GET_ITER_RESPONSE_REST)
+        snapmirror_response['records'][0]['policy'] = {
+            'type': snapmirror_policy}
+
+        if snapmirror_policy == 'async':
+            snapmirror_response['state'] = 'snapmirrored'
+        elif snapmirror_policy == 'sync':
+            snapmirror_response['state'] = 'in_sync'
+
+        response_list = [snapmirror_response,
+                         fake_client.JOB_RESPONSE_REST,
+                         snapmirror_response]
+
+        self.mock_object(self.client, 'send_request',
+                         side_effect=copy.deepcopy(response_list))
+
+        self.client.resync_snapmirror(
+            fake_client.SM_SOURCE_VSERVER, fake_client.SM_SOURCE_VOLUME,
+            fake_client.SM_DEST_VSERVER, fake_client.SM_DEST_VOLUME)
+
+        query_uuid = {}
+        query_uuid['source.path'] = (fake_client.SM_SOURCE_VSERVER + ':' +
+                                     fake_client.SM_SOURCE_VOLUME)
+        query_uuid['destination.path'] = (fake_client.SM_DEST_VSERVER + ':' +
+                                          fake_client.SM_DEST_VOLUME)
+        query_uuid['fields'] = 'uuid,policy.type'
+
+        body_resync = {}
+        if snapmirror_policy == 'async':
+            body_resync['state'] = 'snapmirrored'
+        elif snapmirror_policy == 'sync':
+            body_resync['state'] = 'in_sync'
+
+        self.client.send_request.assert_has_calls([
+            mock.call('/snapmirror/relationships/', 'get', query=query_uuid),
+            mock.call('/snapmirror/relationships/' + fake_client.FAKE_UUID,
+                      'patch', body=body_resync)])
+
+    def test_resync_snapmirror_timeout(self):
+        api_error = netapp_api.NaRetryableError()
+        self.mock_object(self.client, 'resume_snapmirror',
+                         side_effect=api_error)
+
+        self.assertRaises(netapp_api.NaRetryableError,
+                          self.client.resync_snapmirror,
+                          fake_client.SM_SOURCE_VSERVER,
+                          fake_client.SM_SOURCE_VOLUME,
+                          fake_client.SM_DEST_VSERVER,
+                          fake_client.SM_DEST_VOLUME)
+
+    def test_quiesce_snapmirror(self):
+
+        expected_job = {
+            'operation-id': None,
+            'status': None,
+            'jobid': fake_client.FAKE_UUID,
+            'error-code': None,
+            'error-message': None,
+            'relationship-uuid': fake_client.FAKE_UUID,
+        }
+
+        mock_set_snapmirror_state = self.mock_object(
+            self.client,
+            '_set_snapmirror_state',
+            return_value=expected_job)
+
+        result = self.client.quiesce_snapmirror(
+            fake_client.SM_SOURCE_VSERVER, fake_client.SM_SOURCE_VOLUME,
+            fake_client.SM_DEST_VSERVER, fake_client.SM_DEST_VOLUME)
+
+        mock_set_snapmirror_state.assert_called_once_with(
+            'paused',
+            fake_client.SM_SOURCE_VSERVER, fake_client.SM_SOURCE_VOLUME,
+            fake_client.SM_DEST_VSERVER, fake_client.SM_DEST_VOLUME)
+
+        self.assertEqual(expected_job, result)
+
+    def test_break_snapmirror(self):
+        snapmirror_response = copy.deepcopy(
+            fake_client.SNAPMIRROR_GET_ITER_RESPONSE_REST)
+
+        snapmirror_response['state'] = 'broken_off'
+        response_list = [snapmirror_response]
+
+        self.mock_object(self.client, 'send_request',
+                         side_effect=copy.deepcopy(response_list))
+
+        expected_job = {
+            'operation-id': None,
+            'status': None,
+            'jobid': fake_client.FAKE_UUID,
+            'error-code': None,
+            'error-message': None,
+            'relationship-uuid': fake_client.FAKE_UUID,
+        }
+
+        mock_set_snapmirror_state = self.mock_object(
+            self.client,
+            '_set_snapmirror_state',
+            return_value=expected_job)
+
+        self.client.break_snapmirror(
+            fake_client.SM_SOURCE_VSERVER, fake_client.SM_SOURCE_VOLUME,
+            fake_client.SM_DEST_VSERVER, fake_client.SM_DEST_VOLUME)
+
+        mock_set_snapmirror_state.assert_called_once_with(
+            'broken-off',
+            fake_client.SM_SOURCE_VSERVER, fake_client.SM_SOURCE_VOLUME,
+            fake_client.SM_DEST_VSERVER, fake_client.SM_DEST_VOLUME)
+
+    def test_break_snapmirror_not_found(self):
+        self.mock_object(
+            self.client, 'send_request',
+            return_value={'records': []})
+
+        self.assertRaises(
+            netapp_utils.NetAppDriverException,
+            self.client.break_snapmirror,
+            fake_client.SM_SOURCE_VSERVER,
+            fake_client.SM_SOURCE_VOLUME,
+            fake_client.SM_DEST_VSERVER,
+            fake_client.SM_DEST_VOLUME)
+
+    def test_break_snapmirror_timeout(self):
+        # when a timeout happens, an exception is thrown by send_request
+        api_error = netapp_api.NaRetryableError()
+        self.mock_object(self.client, 'send_request',
+                         side_effect=api_error)
+
+        self.assertRaises(netapp_api.NaRetryableError,
+                          self.client.break_snapmirror,
+                          fake_client.SM_SOURCE_VSERVER,
+                          fake_client.SM_SOURCE_VOLUME,
+                          fake_client.SM_DEST_VSERVER,
+                          fake_client.SM_DEST_VOLUME)
+
+    def test_update_snapmirror(self):
+
+        snapmirrors = fake_client.REST_GET_SNAPMIRRORS_RESPONSE
+        self.mock_object(self.client, 'send_request')
+        self.mock_object(self.client, 'get_snapmirrors',
+                         return_value=snapmirrors)
+
+        self.client.update_snapmirror(
+            fake_client.SM_SOURCE_VSERVER, fake_client.SM_SOURCE_VOLUME,
+            fake_client.SM_DEST_VSERVER, fake_client.SM_DEST_VOLUME)
+
+        self.client.send_request.assert_has_calls([
+            mock.call('/snapmirror/relationships/' +
+                      snapmirrors[0]['uuid'] + '/transfers/', 'post',
+                      wait_on_accepted=False)])
+
+    def test_update_snapmirror_no_records(self):
+
+        self.mock_object(self.client, 'send_request')
+        self.mock_object(self.client, 'get_snapmirrors',
+                         return_value=[])
+
+        self.assertRaises(netapp_utils.NetAppDriverException,
+                          self.client.update_snapmirror,
+                          fake_client.SM_SOURCE_VSERVER,
+                          fake_client.SM_SOURCE_VOLUME,
+                          fake_client.SM_DEST_VSERVER,
+                          fake_client.SM_DEST_VOLUME)
+
+        self.client.send_request.assert_not_called()
+
+    def test_update_snapmirror_exception(self):
+
+        snapmirrors = fake_client.REST_GET_SNAPMIRRORS_RESPONSE
+        api_error = netapp_api.NaApiError(
+            code=netapp_api.REST_UPDATE_SNAPMIRROR_FAILED)
+        self.mock_object(self.client, 'send_request',
+                         side_effect=api_error)
+        self.mock_object(self.client, 'get_snapmirrors',
+                         return_value=snapmirrors)
+
+        self.assertRaises(netapp_api.NaApiError,
+                          self.client.update_snapmirror,
+                          fake_client.SM_SOURCE_VSERVER,
+                          fake_client.SM_SOURCE_VOLUME,
+                          fake_client.SM_DEST_VSERVER,
+                          fake_client.SM_DEST_VOLUME)
+
+        self.client.send_request.assert_has_calls([
+            mock.call('/snapmirror/relationships/' +
+                      snapmirrors[0]['uuid'] + '/transfers/', 'post',
+                      wait_on_accepted=False)])
+
+    def test_mount_flexvol(self):
+        volumes = fake_client.VOLUME_GET_ITER_SSC_RESPONSE_REST
+        self.mock_object(self.client, 'send_request',
+                         side_effect=[volumes, None])
+
+        fake_path = '/fake_path'
+        fake_vol_name = volumes['records'][0]['name']
+
+        body = {
+            'nas.path': fake_path
+        }
+        query = {
+            'name': fake_vol_name
+        }
+
+        self.client.mount_flexvol(fake_client.VOLUME_NAME,
+                                  junction_path=fake_path)
+
+        self.client.send_request.assert_has_calls([
+            mock.call('/storage/volumes', 'patch', body=body, query=query)])
+
+    def test_mount_flexvol_default_junction_path(self):
+        volumes = fake_client.VOLUME_GET_ITER_SSC_RESPONSE_REST
+        self.mock_object(self.client, 'send_request',
+                         side_effect=[volumes, None])
+
+        fake_vol_name = volumes['records'][0]['name']
+        body = {
+            'nas.path': '/' + fake_client.VOLUME_NAME
+        }
+        query = {
+            'name': fake_vol_name
+        }
+
+        self.client.mount_flexvol(fake_client.VOLUME_NAME)
+
+        self.client.send_request.assert_has_calls([
+            mock.call('/storage/volumes', 'patch', body=body, query=query)])

cinder/tests/unit/volume/drivers/netapp/dataontap/utils/test_data_motion.py

@@ -675,6 +675,12 @@ class NetAppCDOTDataMotionMixinTestCase(test.TestCase):
                           'is_flexgroup': is_flexgroup})
         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)
+        self.mock_object(self.mock_dest_client,
+                         'get_volume_state',
+                         return_value='online')
         mock_client_call = self.mock_object(
             self.mock_dest_client, 'create_flexvol')
 
@@ -766,18 +772,18 @@ class NetAppCDOTDataMotionMixinTestCase(test.TestCase):
             return_value=False)
         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_volume_state = self.mock_object(self.mock_dest_client,
+                                             'get_volume_state',
+                                             return_value='online')
 
         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(
@@ -840,6 +846,12 @@ class NetAppCDOTDataMotionMixinTestCase(test.TestCase):
             return_value=True)
         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)
+        self.mock_object(self.mock_dest_client,
+                         'get_volume_state',
+                         return_value='online')
         mock_client_call = self.mock_object(
             self.mock_dest_client, 'create_flexvol')
 

cinder/volume/drivers/netapp/dataontap/client/client_cmode.py

@@ -1741,7 +1741,8 @@ class Client(client_base.Client):
 
     def create_volume_async(self, name, aggregate_list, size_gb,
                             space_guarantee_type=None, snapshot_policy=None,
-                            language=None, snapshot_reserve=None,
+                            language=None, dedupe_enabled=False,
+                            compression_enabled=False, snapshot_reserve=None,
                             volume_type='rw'):
         """Creates a FlexGroup volume asynchronously."""
 

cinder/volume/drivers/netapp/dataontap/client/client_cmode_rest.py

@@ -13,11 +13,14 @@
 #    under the License.
 
 import copy
+from datetime import datetime
+from datetime import timedelta
 import math
 from time import time
 
 from oslo_log import log as logging
 from oslo_utils import excutils
+from oslo_utils import units
 import six
 
 from cinder import exception
@@ -35,6 +38,7 @@ ONTAP_C190 = 'C190'
 HTTP_ACCEPTED = 202
 DELETED_PREFIX = 'deleted_cinder_'
 DEFAULT_TIMEOUT = 15
+REST_SYNC_TIMEOUT = 15
 
 # Keys in this map are REST API's endpoints that the user shall have permission
 # in order to enable extra specs reported to Cinder's scheduler.
@@ -1787,3 +1791,499 @@ class RestClient(object):
             msg = _('Volume %s not found.')
             msg_args = flexvol_path or flexvol_name
             raise na_utils.NetAppDriverException(msg % msg_args)
+
+    def get_provisioning_options_from_flexvol(self, flexvol_name):
+        """Get a dict of provisioning options matching existing flexvol."""
+
+        flexvol_info = self.get_flexvol(flexvol_name=flexvol_name)
+        dedupe_info = self.get_flexvol_dedupe_info(flexvol_name)
+
+        provisioning_opts = {
+            'aggregate': flexvol_info['aggregate'],
+            # space-guarantee can be 'none', 'file', 'volume'
+            'space_guarantee_type': flexvol_info.get('space-guarantee'),
+            'snapshot_policy': flexvol_info['snapshot-policy'],
+            'language': flexvol_info['language'],
+            'dedupe_enabled': dedupe_info['dedupe'],
+            'compression_enabled': dedupe_info['compression'],
+            'snapshot_reserve': flexvol_info['percentage-snapshot-reserve'],
+            'volume_type': flexvol_info['type'],
+            'size': int(math.ceil(float(flexvol_info['size']) / units.Gi)),
+            'is_flexgroup': flexvol_info['style-extended'] == 'flexgroup',
+        }
+
+        return provisioning_opts
+
+    def flexvol_exists(self, volume_name):
+        """Checks if a flexvol exists on the storage array."""
+        LOG.debug('Checking if volume %s exists', volume_name)
+
+        query = {
+            'name': volume_name,
+            'return_records': 'false'
+        }
+
+        response = self.send_request('/storage/volumes/', 'get', query=query)
+
+        return response['num_records'] > 0
+
+    def create_volume_async(self, name, aggregate_list, size_gb,
+                            space_guarantee_type=None, snapshot_policy=None,
+                            language=None, dedupe_enabled=False,
+                            compression_enabled=False, snapshot_reserve=None,
+                            volume_type='rw'):
+        """Creates a volume asynchronously."""
+
+        body = {
+            'name': name,
+            'size': size_gb * units.Gi,
+            'type': volume_type,
+        }
+
+        if isinstance(aggregate_list, list):
+            body['style'] = 'flexgroup'
+            body['aggregates'] = [{'name': aggr} for aggr in aggregate_list]
+        else:
+            body['style'] = 'flexvol'
+            body['aggregates'] = [{'name': aggregate_list}]
+
+        if volume_type == 'dp':
+            snapshot_policy = None
+        else:
+            body['nas'] = {'path': '/%s' % name}
+
+        if snapshot_policy is not None:
+            body['snapshot_policy'] = {'name': snapshot_policy}
+
+        if space_guarantee_type:
+            body['guarantee'] = {'type': space_guarantee_type}
+
+        if language is not None:
+            body['language'] = language
+
+        if snapshot_reserve is not None:
+            body['space'] = {
+                'snapshot': {
+                    'reserve_percent': str(snapshot_reserve)
+                }
+            }
+
+        # cDOT compression requires that deduplication be enabled.
+        if dedupe_enabled or compression_enabled:
+            body['efficiency'] = {'dedupe': 'background'}
+
+        if compression_enabled:
+            body['efficiency']['compression'] = 'background'
+
+        response = self.send_request('/storage/volumes/', 'post', body=body,
+                                     wait_on_accepted=False)
+
+        job_info = {
+            'status': None,
+            'jobid': response["job"]["uuid"],
+            'error-code': None,
+            'error-message': None,
+        }
+
+        return job_info
+
+    def create_flexvol(self, flexvol_name, aggregate_name, size_gb,
+                       space_guarantee_type=None, snapshot_policy=None,
+                       language=None, dedupe_enabled=False,
+                       compression_enabled=False, snapshot_reserve=None,
+                       volume_type='rw'):
+        """Creates a flexvol asynchronously and return the job info."""
+
+        return self.create_volume_async(
+            flexvol_name, aggregate_name, size_gb,
+            space_guarantee_type=space_guarantee_type,
+            snapshot_policy=snapshot_policy, language=language,
+            dedupe_enabled=dedupe_enabled,
+            compression_enabled=compression_enabled,
+            snapshot_reserve=snapshot_reserve, volume_type=volume_type)
+
+    def enable_volume_dedupe_async(self, volume_name):
+        """Enable deduplication on FlexVol/FlexGroup volume asynchronously."""
+
+        query = {
+            'name': volume_name,
+            'fields': 'uuid,style',
+        }
+        body = {
+            'efficiency': {'dedupe': 'background'}
+        }
+        self.send_request('/storage/volumes/', 'patch', body=body, query=query,
+                          wait_on_accepted=False)
+
+    def enable_volume_compression_async(self, volume_name):
+        """Enable compression on FlexVol/FlexGroup volume asynchronously."""
+        query = {
+            'name': volume_name
+        }
+        body = {
+            'efficiency': {'compression': 'background'}
+        }
+        self.send_request('/storage/volumes/', 'patch', body=body, query=query,
+                          wait_on_accepted=False)
+
+    def _parse_lagtime(self, time_str):
+        """Parse lagtime string (ISO 8601) into a number of seconds."""
+
+        fmt_str = 'PT'
+        if 'H' in time_str:
+            fmt_str += '%HH'
+        if 'M' in time_str:
+            fmt_str += '%MM'
+        if 'S' in time_str:
+            fmt_str += '%SS'
+
+        t = None
+        try:
+            t = datetime.strptime(time_str, fmt_str)
+        except Exception:
+            LOG.debug("Failed to parse lagtime: %s", time_str)
+            raise
+
+        # convert to timedelta to get the total seconds
+        td = timedelta(hours=t.hour, minutes=t.minute, seconds=t.second)
+        return td.total_seconds()
+
+    def _get_snapmirrors(self, source_vserver=None, source_volume=None,
+                         destination_vserver=None, destination_volume=None):
+
+        fields = ['state', 'source.svm.name', 'source.path',
+                  'destination.svm.name', 'destination.path',
+                  'transfer.end_time', 'lag_time', 'healthy', 'uuid']
+
+        query = {}
+        query['fields'] = '{}'.format(','.join(f for f in fields))
+
+        query_src_vol = source_volume if source_volume else '*'
+        query_src_vserver = source_vserver if source_vserver else '*'
+        query['source.path'] = query_src_vserver + ':' + query_src_vol
+
+        query_dst_vol = destination_volume if destination_volume else '*'
+        query_dst_vserver = destination_vserver if destination_vserver else '*'
+        query['destination.path'] = query_dst_vserver + ':' + query_dst_vol
+
+        response = self.send_request(
+            '/snapmirror/relationships', 'get', query=query)
+
+        snapmirrors = []
+        for record in response.get('records', []):
+            snapmirrors.append({
+                'relationship-status': record.get('state'),
+                'mirror-state': record['state'],
+                'source-vserver': record['source']['svm']['name'],
+                'source-volume': (record['source']['path'].split(':')[1] if
+                                  record.get('source') else None),
+                'destination-vserver': record['destination']['svm']['name'],
+                'destination-volume': (
+                    record['destination']['path'].split(':')[1]
+                    if record.get('destination') else None),
+                'last-transfer-end-timestamp':
+                    (record['transfer']['end_time'] if
+                     record.get('transfer', {}).get('end_time') else None),
+                'lag-time': (self._parse_lagtime(record['lag_time']) if
+                             record.get('lag_time') else None),
+                'is-healthy': record['healthy'],
+                'uuid': record['uuid']
+            })
+
+        return snapmirrors
+
+    def get_snapmirrors(self, source_vserver, source_volume,
+                        destination_vserver, destination_volume,
+                        desired_attributes=None):
+        """Gets one or more SnapMirror relationships.
+
+        Either the source or destination info may be omitted.
+        Desired attributes exists only to keep consistent with ZAPI client
+        signature and has no effect in the output.
+        """
+
+        snapmirrors = self._get_snapmirrors(
+            source_vserver=source_vserver,
+            source_volume=source_volume,
+            destination_vserver=destination_vserver,
+            destination_volume=destination_volume)
+
+        return snapmirrors
+
+    def create_snapmirror(self, source_vserver, source_volume,
+                          destination_vserver, destination_volume,
+                          schedule=None, policy=None,
+                          relationship_type='data_protection'):
+        """Creates a SnapMirror relationship.
+
+        The schedule and relationship type is kept to avoid breaking
+        the API used by data_motion, but are not used on the REST API.
+
+        The schedule is part of the policy associated the relationship and the
+        relationship_type will be ignored because XDP is the only type
+        supported through REST API.
+        """
+
+        body = {
+            'source': {
+                'path': source_vserver + ':' + source_volume
+            },
+            'destination': {
+                'path': destination_vserver + ':' + destination_volume
+            }
+        }
+
+        if policy:
+            body['policy'] = {'name': policy}
+
+        try:
+            self.send_request('/snapmirror/relationships/', 'post', body=body)
+        except netapp_api.NaApiError as e:
+            if e.code != netapp_api.REST_ERELATION_EXISTS:
+                raise e
+
+    def _set_snapmirror_state(self, state, source_vserver, source_volume,
+                              destination_vserver, destination_volume,
+                              wait_result=True):
+        """Change the snapmirror state between two volumes."""
+
+        snapmirror = self.get_snapmirrors(source_vserver, source_volume,
+                                          destination_vserver,
+                                          destination_volume)
+
+        if not snapmirror:
+            msg = _('Failed to get information about relationship between '
+                    'source %(src_vserver)s:%(src_volume)s and '
+                    'destination %(dst_vserver)s:%(dst_volume)s.') % {
+                'src_vserver': source_vserver,
+                'src_volume': source_volume,
+                'dst_vserver': destination_vserver,
+                'dst_volume': destination_volume}
+            raise na_utils.NetAppDriverException(msg)
+
+        uuid = snapmirror[0]['uuid']
+        body = {'state': state}
+        result = self.send_request('/snapmirror/relationships/' + uuid,
+                                   'patch', body=body,
+                                   wait_on_accepted=wait_result)
+        job = result['job']
+        job_info = {
+            'operation-id': None,
+            'status': None,
+            'jobid': job.get('uuid'),
+            'error-code': None,
+            'error-message': None,
+            'relationship-uuid': uuid,
+        }
+
+        return job_info
+
+    def initialize_snapmirror(self, source_vserver, source_volume,
+                              destination_vserver, destination_volume,
+                              source_snapshot=None, transfer_priority=None):
+        """Initializes a SnapMirror relationship."""
+
+        # TODO: Trigger a geometry exception to be caught by data_motion.
+        # This error is raised when using ZAPI with different volume component
+        # numbers, but in REST, the job must be checked sometimes before that
+        # error occurs.
+
+        return self._set_snapmirror_state(
+            'snapmirrored', source_vserver, source_volume,
+            destination_vserver, destination_volume, wait_result=False)
+
+    def abort_snapmirror(self, source_vserver, source_volume,
+                         destination_vserver, destination_volume,
+                         clear_checkpoint=False):
+        """Stops ongoing transfers for a SnapMirror relationship."""
+
+        snapmirror = self.get_snapmirrors(source_vserver, source_volume,
+                                          destination_vserver,
+                                          destination_volume)
+        if not snapmirror:
+            msg = _('Failed to get information about relationship between '
+                    'source %(src_vserver)s:%(src_volume)s and '
+                    'destination %(dst_vserver)s:%(dst_volume)s.') % {
+                'src_vserver': source_vserver,
+                'src_volume': source_volume,
+                'dst_vserver': destination_vserver,
+                'dst_volume': destination_volume}
+            raise na_utils.NetAppDriverException(msg)
+
+        snapmirror_uuid = snapmirror[0]['uuid']
+
+        query = {'state': 'transferring'}
+        transfers = self.send_request('/snapmirror/relationships/' +
+                                      snapmirror_uuid + '/transfers/', 'get',
+                                      query=query)
+
+        if not transfers.get('records'):
+            raise netapp_api.NaApiError(
+                code=netapp_api.ENOTRANSFER_IN_PROGRESS)
+
+        body = {'state': 'hard_aborted' if clear_checkpoint else 'aborted'}
+
+        for transfer in transfers['records']:
+            transfer_uuid = transfer['uuid']
+            self.send_request('/snapmirror/relationships/' +
+                              snapmirror_uuid + '/transfers/' +
+                              transfer_uuid, 'patch', body=body)
+
+    def delete_snapmirror(self, source_vserver, source_volume,
+                          destination_vserver, destination_volume):
+
+        """Deletes an SnapMirror relationship on destination."""
+
+        query_uuid = {}
+        query_uuid['source.path'] = source_vserver + ':' + source_volume
+        query_uuid['destination.path'] = (destination_vserver + ':' +
+                                          destination_volume)
+        query_uuid['fields'] = 'uuid'
+
+        response = self.send_request('/snapmirror/relationships/', 'get',
+                                     query=query_uuid)
+
+        records = response.get('records')
+        if not records:
+            raise netapp_api.NaApiError(code=netapp_api.EOBJECTNOTFOUND)
+
+        # 'destination_only' deletes the snapmirror on destination but does not
+        # release it on source.
+        query_delete = {"destination_only": "true"}
+
+        snapmirror_uuid = records[0].get('uuid')
+        self.send_request('/snapmirror/relationships/' +
+                          snapmirror_uuid, 'delete',
+                          query=query_delete)
+
+    def resume_snapmirror(self, source_vserver, source_volume,
+                          destination_vserver, destination_volume):
+
+        """Resume a SnapMirror relationship."""
+
+        query_uuid = {}
+        query_uuid['source.path'] = source_vserver + ':' + source_volume
+        query_uuid['destination.path'] = (destination_vserver + ':' +
+                                          destination_volume)
+        query_uuid['fields'] = 'uuid,policy.type'
+
+        response_snapmirrors = self.send_request('/snapmirror/relationships/',
+                                                 'get', query=query_uuid)
+
+        records = response_snapmirrors.get('records')
+        if not records:
+            raise netapp_api.NaApiError(code=netapp_api.EOBJECTNOTFOUND)
+
+        snapmirror_uuid = records[0]['uuid']
+        snapmirror_policy = records[0]['policy']['type']
+
+        body_resync = {}
+        if snapmirror_policy == 'async':
+            body_resync['state'] = 'snapmirrored'
+        elif snapmirror_policy == 'sync':
+            body_resync['state'] = 'in_sync'
+
+        self.send_request('/snapmirror/relationships/' +
+                          snapmirror_uuid, 'patch',
+                          body=body_resync)
+
+    def release_snapmirror(self, source_vserver, source_volume,
+                           destination_vserver, destination_volume,
+                           relationship_info_only=False):
+        """Removes a SnapMirror relationship on the source endpoint."""
+
+        query_uuid = {}
+        query_uuid['list_destinations_only'] = 'true'
+        query_uuid['source.path'] = source_vserver + ':' + source_volume
+        query_uuid['destination.path'] = (destination_vserver + ':' +
+                                          destination_volume)
+        query_uuid['fields'] = 'uuid'
+
+        response_snapmirrors = self.send_request('/snapmirror/relationships/',
+                                                 'get', query=query_uuid)
+
+        records = response_snapmirrors.get('records')
+        if not records:
+            raise netapp_api.NaApiError(code=netapp_api.EOBJECTNOTFOUND)
+
+        query_release = {}
+        if relationship_info_only:
+            # release without removing related snapshots
+            query_release['source_info_only'] = 'true'
+        else:
+            # release and removing all related snapshots
+            query_release['source_only'] = 'true'
+
+        snapmirror_uuid = records[0].get('uuid')
+        self.send_request('/snapmirror/relationships/' +
+                          snapmirror_uuid, 'delete',
+                          query=query_release)
+
+    def resync_snapmirror(self, source_vserver, source_volume,
+                          destination_vserver, destination_volume):
+        """Resync a SnapMirror relationship."""
+
+        # We reuse the resume operation for resync since both are handled in
+        # the same way in the REST API, by setting the snapmirror relationship
+        # to the snapmirrored state.
+        self.resume_snapmirror(source_vserver,
+                               source_volume,
+                               destination_vserver,
+                               destination_volume)
+
+    def quiesce_snapmirror(self, source_vserver, source_volume,
+                           destination_vserver, destination_volume):
+        """Disables future transfers to a SnapMirror destination."""
+
+        return self._set_snapmirror_state(
+            'paused', source_vserver, source_volume,
+            destination_vserver, destination_volume)
+
+    def break_snapmirror(self, source_vserver, source_volume,
+                         destination_vserver, destination_volume):
+        """Breaks a data protection SnapMirror relationship."""
+
+        self._set_snapmirror_state(
+            'broken-off', source_vserver, source_volume,
+            destination_vserver, destination_volume)
+
+    def update_snapmirror(self, source_vserver, source_volume,
+                          destination_vserver, destination_volume):
+        """Schedules a SnapMirror update."""
+
+        snapmirror = self.get_snapmirrors(source_vserver, source_volume,
+                                          destination_vserver,
+                                          destination_volume)
+        if not snapmirror:
+            msg = _('Failed to get information about relationship between '
+                    'source %(src_vserver)s:%(src_volume)s and '
+                    'destination %(dst_vserver)s:%(dst_volume)s.') % {
+                'src_vserver': source_vserver,
+                'src_volume': source_volume,
+                'dst_vserver': destination_vserver,
+                'dst_volume': destination_volume}
+
+            raise na_utils.NetAppDriverException(msg)
+
+        snapmirror_uuid = snapmirror[0]['uuid']
+
+        # NOTE(nahimsouza): A POST with an empty body starts the update
+        # snapmirror operation.
+        try:
+            self.send_request('/snapmirror/relationships/' +
+                              snapmirror_uuid + '/transfers/', 'post',
+                              wait_on_accepted=False)
+        except netapp_api.NaApiError as e:
+            if (e.code != netapp_api.REST_UPDATE_SNAPMIRROR_FAILED):
+                LOG.warning('Unexpected failure during snapmirror update.'
+                            'Code: %(code)s, Message: %(message)s',
+                            {'code': e.code, 'message': e.message})
+            raise
+
+    def mount_flexvol(self, flexvol_name, junction_path=None):
+        """Mounts a volume on a junction path."""
+
+        query = {'name': flexvol_name}
+        body = {'nas.path': (
+            junction_path if junction_path else '/%s' % flexvol_name)}
+        self.send_request('/storage/volumes', 'patch', query=query, body=body)

cinder/volume/drivers/netapp/dataontap/utils/data_motion.py

@@ -353,7 +353,8 @@ class DataMotionMixin(object):
                 src_vserver, src_flexvol_name, dest_vserver,
                 dest_flexvol_name,
                 desired_attributes=['relationship-status', 'mirror-state'])[0]
-            if snapmirror.get('relationship-status') != 'quiesced':
+            if (snapmirror.get('relationship-status') not in ['quiesced',
+                                                              'paused']):
                 msg = _("SnapMirror relationship is not quiesced.")
                 raise na_utils.NetAppDriverException(msg)
 
@@ -524,8 +525,8 @@ class DataMotionMixin(object):
                         dest_flexvol_name)
 
             except loopingcall.LoopingCallTimeOut:
-                msg = _("Timeout waiting destination FlexGroup to to come "
+                msg = _("Timeout waiting destination FlexGroup "
-                        "online.")
+                        "to come online.")
                 raise na_utils.NetAppDriverException(msg)
 
         else:
@@ -534,6 +535,24 @@ class DataMotionMixin(object):
                                        size,
                                        **provisioning_options)
 
+            timeout = self._get_replication_volume_online_timeout()
+
+            def _wait_volume_is_online():
+                volume_state = dest_client.get_volume_state(
+                    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()
+
+            except loopingcall.LoopingCallTimeOut:
+                msg = _("Timeout waiting destination FlexVol to to come "
+                        "online.")
+                raise na_utils.NetAppDriverException(msg)
+
     def ensure_snapmirrors(self, config, src_backend_name, src_flexvol_names):
         """Ensure all the SnapMirrors needed for whole-backend replication."""
         backend_names = self.get_replication_backend_names(config)