From a6c22238e1021f51d0348e58402db4f56dbe539d Mon Sep 17 00:00:00 2001
From: Peter Wang <peter.wang13@emc.com>
Date: Mon, 13 Mar 2017 15:54:51 +0800
Subject: [PATCH] Unity: Use thin clone when cloning volume

Unity storage now supports cloning volume by taking its snapshot,
it's more efficient than previous implementation.

However, there is limitation on the count of thin clones inside one LUN
family sharing the same base LUN. Every time the limitation reaches, a
new volume will be cloned using `dd` copy. Then thin clone will be taken
from the new dd-copied volume.

DocImpact
Implements: blueprint unity-fast-clone

Change-Id: I5e4264ce9917e831e3efe27dc332641c9f94c07f
---
 .../drivers/dell_emc/unity/fake_exception.py  |   4 +
 .../drivers/dell_emc/unity/test_adapter.py    | 290 +++++++++++++++---
 .../drivers/dell_emc/unity/test_client.py     |  46 ++-
 .../drivers/dell_emc/unity/test_driver.py     |   8 +-
 .../drivers/dell_emc/unity/test_utils.py      |  16 +-
 .../volume/drivers/dell_emc/unity/adapter.py  | 239 ++++++++++++---
 .../volume/drivers/dell_emc/unity/client.py   |  14 +
 .../volume/drivers/dell_emc/unity/driver.py   |   3 +-
 .../unity-fast-clone-02ae88ba8fdef145.yaml    |   8 +
 9 files changed, 520 insertions(+), 108 deletions(-)
 create mode 100644 releasenotes/notes/unity-fast-clone-02ae88ba8fdef145.yaml

diff --git a/cinder/tests/unit/volume/drivers/dell_emc/unity/fake_exception.py b/cinder/tests/unit/volume/drivers/dell_emc/unity/fake_exception.py
index dec754b0fca..3f7f3fccee4 100644
--- a/cinder/tests/unit/volume/drivers/dell_emc/unity/fake_exception.py
+++ b/cinder/tests/unit/volume/drivers/dell_emc/unity/fake_exception.py
@@ -46,6 +46,10 @@ class UnityNothingToModifyError(StoropsException):
     pass
 
 
+class UnityThinCloneLimitExceededError(StoropsException):
+    pass
+
+
 class ExtendLunError(Exception):
     pass
 
diff --git a/cinder/tests/unit/volume/drivers/dell_emc/unity/test_adapter.py b/cinder/tests/unit/volume/drivers/dell_emc/unity/test_adapter.py
index d52eefeef55..e353c2cad0d 100644
--- a/cinder/tests/unit/volume/drivers/dell_emc/unity/test_adapter.py
+++ b/cinder/tests/unit/volume/drivers/dell_emc/unity/test_adapter.py
@@ -13,10 +13,12 @@
 # License for the specific language governing permissions and limitations
 # under the License.
 
+import contextlib
 import functools
 import unittest
 
 import mock
+from oslo_utils import units
 
 from cinder import exception
 from cinder.tests.unit.volume.drivers.dell_emc.unity \
@@ -75,17 +77,21 @@ class MockClient(object):
 
     @staticmethod
     def create_lun(name, size, pool, description=None, io_limit_policy=None):
-        return test_client.MockResource(_id='lun_3')
+        return test_client.MockResource(_id=name, name=name)
 
     @staticmethod
     def get_lun(name=None, lun_id=None):
         if lun_id is None:
             lun_id = 'lun_4'
+        if lun_id in ('lun_43',):  # for thin clone cases
+            return test_client.MockResource(_id=lun_id, name=name)
         if name == 'not_exists':
             ret = test_client.MockResource(name=lun_id)
             ret.existed = False
         else:
-            ret = test_client.MockResource(_id=lun_id)
+            if name is None:
+                name = lun_id
+            ret = test_client.MockResource(_id=lun_id, name=name)
         return ret
 
     @staticmethod
@@ -99,10 +105,15 @@ class MockClient(object):
 
     @staticmethod
     def create_snap(src_lun_id, name=None):
+        if src_lun_id in ('lun_53', 'lun_55'):  # for thin clone cases
+            return test_client.MockResource(
+                _id='snap_clone_{}'.format(src_lun_id))
         return test_client.MockResource(name=name, _id=src_lun_id)
 
     @staticmethod
     def get_snap(name=None):
+        if name in ('snap_50',):  # for thin clone cases
+            return name
         snap = test_client.MockResource(name=name, _id=name)
         if name is not None:
             ret = snap
@@ -169,6 +180,14 @@ class MockClient(object):
     def get_ethernet_ports():
         return test_client.MockResourceList(ids=['spa_eth0', 'spb_eth0'])
 
+    @staticmethod
+    def thin_clone(obj, name, io_limit_policy, description, new_size_gb):
+        if (obj.name, name) in (
+                ('snap_61', 'lun_60'), ('lun_63', 'lun_60')):
+            return test_client.MockResource(_id=name)
+        else:
+            raise ex.UnityThinCloneLimitExceededError
+
     @property
     def system(self):
         return self._system
@@ -187,11 +206,19 @@ class MockLookupService(object):
         }
 
 
+class MockOSResource(mock.Mock):
+    def __init__(self, *args, **kwargs):
+        super(MockOSResource, self).__init__(*args, **kwargs)
+        if 'name' in kwargs:
+            self.name = kwargs['name']
+
+
 def mock_adapter(driver_clz):
     ret = driver_clz()
     ret._client = MockClient()
     with mock.patch('cinder.volume.drivers.dell_emc.unity.adapter.'
-                    'CommonAdapter.validate_ports'):
+                    'CommonAdapter.validate_ports'), \
+            patch_storops():
         ret.do_setup(MockDriver(), MockConfig())
     ret.lookup_service = MockLookupService()
     return ret
@@ -205,14 +232,14 @@ def get_connector_properties():
     return {'host': 'host1', 'wwpns': 'abcdefg'}
 
 
-def copy_volume(from_path, to_path, size_in_m, block_size, sparse=True):
-    pass
-
-
 def get_lun_pl(name):
     return 'id^%s|system^CLIENT_SERIAL|type^lun|version^None' % name
 
 
+def get_snap_lun_pl(name):
+    return 'id^%s|system^CLIENT_SERIAL|type^snap_lun|version^None' % name
+
+
 def get_snap_pl(name):
     return 'id^%s|system^CLIENT_SERIAL|type^snapshot|version^None' % name
 
@@ -232,7 +259,6 @@ def patch_for_unity_adapter(func):
                 new=get_backend_qos_specs)
     @mock.patch('cinder.utils.brick_get_connector_properties',
                 new=get_connector_properties)
-    @mock.patch('cinder.volume.utils.copy_volume', new=copy_volume)
     def func_wrapper(*args, **kwargs):
         return func(*args, **kwargs)
 
@@ -261,11 +287,48 @@ patch_for_fc_adapter = patch_for_concrete_adapter(
     'cinder.volume.drivers.dell_emc.unity.adapter.FCAdapter')
 
 
+@contextlib.contextmanager
+def patch_thin_clone(cloned_lun):
+    with mock.patch.object(adapter.CommonAdapter, '_thin_clone') as tc:
+        tc.return_value = cloned_lun
+        yield tc
+
+
+@contextlib.contextmanager
+def patch_dd_copy(copied_lun):
+    with mock.patch.object(adapter.CommonAdapter, '_dd_copy') as dd:
+        dd.return_value = copied_lun
+        yield dd
+
+
+@contextlib.contextmanager
+def patch_copy_volume():
+    with mock.patch('cinder.volume.utils.copy_volume') as mocked:
+        yield mocked
+
+
+@contextlib.contextmanager
+def patch_storops():
+    with mock.patch.object(adapter, 'storops') as storops:
+        storops.ThinCloneActionEnum = mock.Mock(DD_COPY='DD_COPY')
+        yield storops
+
+
+class IdMatcher(object):
+    def __init__(self, obj):
+        self._obj = obj
+
+    def __eq__(self, other):
+        return self._obj._id == other._id
+
+
 ########################
 #
 #   Start of Tests
 #
 ########################
+
+@mock.patch.object(adapter, 'storops_ex', new=ex)
 class CommonAdapterTest(unittest.TestCase):
     def setUp(self):
         self.adapter = mock_adapter(adapter.CommonAdapter)
@@ -278,38 +341,35 @@ class CommonAdapterTest(unittest.TestCase):
 
     @patch_for_unity_adapter
     def test_create_volume(self):
-        volume = mock.Mock(size=5, host='unity#pool1')
+        volume = MockOSResource(name='lun_3', size=5, host='unity#pool1')
         ret = self.adapter.create_volume(volume)
         expected = get_lun_pl('lun_3')
         self.assertEqual(expected, ret['provider_location'])
 
     def test_create_snapshot(self):
-        volume = mock.Mock(provider_location='id^lun_43')
-        snap = mock.Mock(volume=volume)
-        snap.name = 'abc-def_snap'
+        volume = MockOSResource(provider_location='id^lun_43')
+        snap = MockOSResource(volume=volume, name='abc-def_snap')
         result = self.adapter.create_snapshot(snap)
         self.assertEqual(get_snap_pl('lun_43'), result['provider_location'])
         self.assertEqual('lun_43', result['provider_id'])
 
     def test_delete_snap(self):
         def f():
-            snap = mock.Mock()
-            snap.name = 'abc-def_snap'
-
+            snap = MockOSResource(name='abc-def_snap')
             self.adapter.delete_snapshot(snap)
 
         self.assertRaises(ex.SnapDeleteIsCalled, f)
 
     def test_get_lun_id_has_location(self):
-        volume = mock.Mock(provider_location='id^lun_43')
+        volume = MockOSResource(provider_location='id^lun_43')
         self.assertEqual('lun_43', self.adapter.get_lun_id(volume))
 
     def test_get_lun_id_no_location(self):
-        volume = mock.Mock(provider_location=None)
+        volume = MockOSResource(provider_location=None)
         self.assertEqual('lun_4', self.adapter.get_lun_id(volume))
 
     def test_delete_volume(self):
-        volume = mock.Mock(provider_location='id^lun_4')
+        volume = MockOSResource(provider_location='id^lun_4')
         self.adapter.delete_volume(volume)
 
     def test_get_pool_stats(self):
@@ -376,7 +436,7 @@ class CommonAdapterTest(unittest.TestCase):
 
     def test_terminate_connection_volume(self):
         def f():
-            volume = mock.Mock(provider_location='id^lun_43', id='id_43')
+            volume = MockOSResource(provider_location='id^lun_43', id='id_43')
             connector = {'host': 'host1'}
             self.adapter.terminate_connection(volume, connector)
 
@@ -385,22 +445,21 @@ class CommonAdapterTest(unittest.TestCase):
     def test_terminate_connection_snapshot(self):
         def f():
             connector = {'host': 'host1'}
-            snap = mock.Mock(id='snap_0', name='snap_0')
-            snap.name = 'snap_0'
+            snap = MockOSResource(name='snap_0', id='snap_0')
             self.adapter.terminate_connection_snapshot(snap, connector)
 
         self.assertRaises(ex.DetachIsCalled, f)
 
     def test_manage_existing_by_name(self):
         ref = {'source-id': 12}
-        volume = mock.Mock(name='lun1')
+        volume = MockOSResource(name='lun1')
         ret = self.adapter.manage_existing(volume, ref)
         expected = get_lun_pl('12')
         self.assertEqual(expected, ret['provider_location'])
 
     def test_manage_existing_by_id(self):
         ref = {'source-name': 'lunx'}
-        volume = mock.Mock(name='lun1')
+        volume = MockOSResource(name='lun1')
         ret = self.adapter.manage_existing(volume, ref)
         expected = get_lun_pl('lun_4')
         self.assertEqual(expected, ret['provider_location'])
@@ -408,7 +467,7 @@ class CommonAdapterTest(unittest.TestCase):
     def test_manage_existing_invalid_ref(self):
         def f():
             ref = {}
-            volume = mock.Mock(name='lun1')
+            volume = MockOSResource(name='lun1')
             self.adapter.manage_existing(volume, ref)
 
         self.assertRaises(exception.ManageExistingInvalidReference, f)
@@ -416,7 +475,7 @@ class CommonAdapterTest(unittest.TestCase):
     def test_manage_existing_lun_not_found(self):
         def f():
             ref = {'source-name': 'not_exists'}
-            volume = mock.Mock(name='lun1')
+            volume = MockOSResource(name='lun1')
             self.adapter.manage_existing(volume, ref)
 
         self.assertRaises(exception.ManageExistingInvalidReference, f)
@@ -424,8 +483,8 @@ class CommonAdapterTest(unittest.TestCase):
     @patch_for_unity_adapter
     def test_manage_existing_get_size_invalid_backend(self):
         def f():
-            volume = mock.Mock(volume_type_id='thin',
-                               host='host@backend#pool1')
+            volume = MockOSResource(volume_type_id='thin',
+                                    host='host@backend#pool1')
             ref = {'source-id': 12}
             self.adapter.manage_existing_get_size(volume, ref)
 
@@ -433,38 +492,175 @@ class CommonAdapterTest(unittest.TestCase):
 
     @patch_for_unity_adapter
     def test_manage_existing_get_size_success(self):
-        volume = mock.Mock(volume_type_id='thin', host='host@backend#pool0')
+        volume = MockOSResource(volume_type_id='thin',
+                                host='host@backend#pool0')
         ref = {'source-id': 12}
         volume_size = self.adapter.manage_existing_get_size(volume, ref)
         self.assertEqual(5, volume_size)
 
     @patch_for_unity_adapter
     def test_create_volume_from_snapshot(self):
-        volume = mock.Mock(id='id_44', host='unity#pool1',
-                           provider_location=get_lun_pl('12'))
-        snap = mock.Mock(name='snap_44')
-        ret = self.adapter.create_volume_from_snapshot(volume, snap)
-        self.assertEqual(get_lun_pl('lun_3'), ret['provider_location'])
+        lun_id = 'lun_50'
+        volume = MockOSResource(name=lun_id, id=lun_id, host='unity#pool1')
+        snap_id = 'snap_50'
+        snap = MockOSResource(name=snap_id)
+        with patch_thin_clone(test_client.MockResource(_id=lun_id)) as tc:
+            ret = self.adapter.create_volume_from_snapshot(volume, snap)
+            self.assertEqual(get_snap_lun_pl(lun_id),
+                             ret['provider_location'])
+            tc.assert_called_with(adapter.VolumeParams(self.adapter, volume),
+                                  snap_id)
 
     @patch_for_unity_adapter
-    def test_create_cloned_volume(self):
-        volume = mock.Mock(id='id_55', host='unity#pool1', size=3,
-                           provider_location=get_lun_pl('lun55'))
-        src_vref = mock.Mock(id='id_66', name='LUN 66',
-                             provider_location=get_lun_pl('lun66'))
-        ret = self.adapter.create_cloned_volume(volume, src_vref)
-        self.assertEqual(get_lun_pl('lun_3'), ret['provider_location'])
+    def test_create_cloned_volume_attached(self):
+        lun_id = 'lun_51'
+        src_lun_id = 'lun_53'
+        volume = MockOSResource(name=lun_id, id=lun_id, host='unity#pool1')
+        src_vref = MockOSResource(id=src_lun_id, name=src_lun_id,
+                                  provider_location=get_lun_pl(src_lun_id),
+                                  volume_attachment=['not_care'])
+        with patch_dd_copy(test_client.MockResource(_id=lun_id)) as dd:
+            ret = self.adapter.create_cloned_volume(volume, src_vref)
+            dd.assert_called_with(
+                adapter.VolumeParams(self.adapter, volume),
+                IdMatcher(test_client.MockResource(
+                    _id='snap_clone_{}'.format(src_lun_id))),
+                src_lun=IdMatcher(test_client.MockResource(_id=src_lun_id)))
+            self.assertEqual(get_lun_pl(lun_id), ret['provider_location'])
+
+    @patch_for_unity_adapter
+    def test_create_cloned_volume_available(self):
+        lun_id = 'lun_54'
+        src_lun_id = 'lun_55'
+        volume = MockOSResource(id=lun_id, host='unity#pool1', size=3,
+                                provider_location=get_lun_pl(lun_id))
+        src_vref = MockOSResource(id=src_lun_id, name=src_lun_id,
+                                  provider_location=get_lun_pl(src_lun_id),
+                                  volume_attachment=None)
+        with patch_thin_clone(test_client.MockResource(_id=lun_id)) as tc:
+            ret = self.adapter.create_cloned_volume(volume, src_vref)
+            tc.assert_called_with(
+                adapter.VolumeParams(self.adapter, volume),
+                IdMatcher(test_client.MockResource(
+                    _id='snap_clone_{}'.format(src_lun_id))),
+                src_lun=IdMatcher(test_client.MockResource(_id=src_lun_id)))
+            self.assertEqual(get_snap_lun_pl(lun_id), ret['provider_location'])
+
+    @patch_for_unity_adapter
+    def test_dd_copy_with_src_lun(self):
+        lun_id = 'lun_56'
+        src_lun_id = 'lun_57'
+        src_snap_id = 'snap_57'
+        volume = MockOSResource(name=lun_id, id=lun_id, host='unity#pool1',
+                                provider_location=get_lun_pl(lun_id))
+        src_snap = test_client.MockResource(name=src_snap_id, _id=src_snap_id)
+        src_lun = test_client.MockResource(name=src_lun_id, _id=src_lun_id)
+        src_lun.size_total = 6 * units.Gi
+        with patch_copy_volume() as copy_volume:
+            ret = self.adapter._dd_copy(
+                adapter.VolumeParams(self.adapter, volume), src_snap,
+                src_lun=src_lun)
+            copy_volume.assert_called_with('dev', 'dev', 6144, '1M',
+                                           sparse=True)
+            self.assertEqual(IdMatcher(test_client.MockResource(_id=lun_id)),
+                             ret)
+
+    @patch_for_unity_adapter
+    def test_dd_copy_wo_src_lun(self):
+        lun_id = 'lun_58'
+        src_lun_id = 'lun_59'
+        src_snap_id = 'snap_59'
+        volume = MockOSResource(name=lun_id, id=lun_id, host='unity#pool1',
+                                provider_location=get_lun_pl(lun_id))
+        src_snap = test_client.MockResource(name=src_snap_id, _id=src_snap_id)
+        src_snap.storage_resource = test_client.MockResource(name=src_lun_id,
+                                                             _id=src_lun_id)
+        with patch_copy_volume() as copy_volume:
+            ret = self.adapter._dd_copy(
+                adapter.VolumeParams(self.adapter, volume), src_snap)
+            copy_volume.assert_called_with('dev', 'dev', 5120, '1M',
+                                           sparse=True)
+            self.assertEqual(IdMatcher(test_client.MockResource(_id=lun_id)),
+                             ret)
+
+    @patch_for_unity_adapter
+    def test_dd_copy_raise(self):
+        lun_id = 'lun_58'
+        src_snap_id = 'snap_59'
+        volume = MockOSResource(name=lun_id, id=lun_id, host='unity#pool1',
+                                provider_location=get_lun_pl(lun_id))
+        src_snap = test_client.MockResource(name=src_snap_id, _id=src_snap_id)
+        with patch_copy_volume() as copy_volume:
+            copy_volume.side_effect = AttributeError
+            self.assertRaises(AttributeError,
+                              self.adapter._dd_copy, volume, src_snap)
+
+    @patch_for_unity_adapter
+    def test_thin_clone(self):
+        lun_id = 'lun_60'
+        src_snap_id = 'snap_61'
+        volume = MockOSResource(name=lun_id, id=lun_id, size=1,
+                                provider_location=get_snap_lun_pl(lun_id))
+        src_snap = test_client.MockResource(name=src_snap_id, _id=src_snap_id)
+        ret = self.adapter._thin_clone(volume, src_snap)
+        self.assertEqual(IdMatcher(test_client.MockResource(_id=lun_id)), ret)
+
+    @patch_for_unity_adapter
+    def test_thin_clone_downgraded_with_src_lun(self):
+        lun_id = 'lun_60'
+        src_snap_id = 'snap_62'
+        src_lun_id = 'lun_62'
+        volume = MockOSResource(name=lun_id, id=lun_id, size=1,
+                                provider_location=get_snap_lun_pl(lun_id))
+        src_snap = test_client.MockResource(name=src_snap_id, _id=src_snap_id)
+        src_lun = test_client.MockResource(name=src_lun_id, _id=src_lun_id)
+        new_dd_lun = test_client.MockResource(name='lun_63')
+        with patch_storops() as mocked_storops, \
+                patch_dd_copy(new_dd_lun) as dd:
+            ret = self.adapter._thin_clone(
+                adapter.VolumeParams(self.adapter, volume),
+                src_snap, src_lun=src_lun)
+            vol_params = adapter.VolumeParams(self.adapter, volume)
+            vol_params.name = 'hidden-{}'.format(volume.name)
+            vol_params.description = 'hidden-{}'.format(volume.description)
+            dd.assert_called_with(vol_params, src_snap, src_lun=src_lun)
+            mocked_storops.TCHelper.notify.assert_called_with(src_lun,
+                                                              'DD_COPY',
+                                                              new_dd_lun)
+        self.assertEqual(IdMatcher(test_client.MockResource(_id=lun_id)), ret)
+
+    @patch_for_unity_adapter
+    def test_thin_clone_downgraded_wo_src_lun(self):
+        lun_id = 'lun_60'
+        src_snap_id = 'snap_62'
+        volume = MockOSResource(name=lun_id, id=lun_id, size=1,
+                                provider_location=get_snap_lun_pl(lun_id))
+        src_snap = test_client.MockResource(name=src_snap_id, _id=src_snap_id)
+        new_dd_lun = test_client.MockResource(name='lun_63')
+        with patch_storops() as mocked_storops, \
+                patch_dd_copy(new_dd_lun) as dd:
+            ret = self.adapter._thin_clone(
+                adapter.VolumeParams(self.adapter, volume), src_snap)
+            vol_params = adapter.VolumeParams(self.adapter, volume)
+            vol_params.name = 'hidden-{}'.format(volume.name)
+            vol_params.description = 'hidden-{}'.format(volume.description)
+            dd.assert_called_with(vol_params, src_snap, src_lun=None)
+            mocked_storops.TCHelper.notify.assert_called_with(src_snap,
+                                                              'DD_COPY',
+                                                              new_dd_lun)
+        self.assertEqual(IdMatcher(test_client.MockResource(_id=lun_id)), ret)
 
     def test_extend_volume_error(self):
         def f():
-            volume = mock.Mock(id='l56', provider_location=get_lun_pl('lun56'))
+            volume = MockOSResource(id='l56',
+                                    provider_location=get_lun_pl('lun56'))
             self.adapter.extend_volume(volume, -1)
 
         self.assertRaises(ex.ExtendLunError, f)
 
     def test_extend_volume_no_id(self):
         def f():
-            volume = mock.Mock(provider_location='type^lun')
+            volume = MockOSResource(provider_location='type^lun')
             self.adapter.extend_volume(volume, 5)
 
         self.assertRaises(exception.VolumeBackendAPIException, f)
@@ -546,7 +742,7 @@ class FCAdapterTest(unittest.TestCase):
 
     @patch_for_fc_adapter
     def test_initialize_connection_volume(self):
-        volume = mock.Mock(provider_location='id^lun_43', id='id_43')
+        volume = MockOSResource(provider_location='id^lun_43', id='id_43')
         connector = {'host': 'host1'}
         conn_info = self.adapter.initialize_connection(volume, connector)
         self.assertEqual('fibre_channel', conn_info['driver_volume_type'])
@@ -555,7 +751,7 @@ class FCAdapterTest(unittest.TestCase):
 
     @patch_for_fc_adapter
     def test_initialize_connection_snapshot(self):
-        snap = mock.Mock(id='snap_1', name='snap_1')
+        snap = MockOSResource(id='snap_1', name='snap_1')
         connector = {'host': 'host1'}
         conn_info = self.adapter.initialize_connection_snapshot(
             snap, connector)
@@ -565,7 +761,7 @@ class FCAdapterTest(unittest.TestCase):
 
     def test_terminate_connection_auto_zone_enabled(self):
         connector = {'host': 'host1', 'wwpns': 'abcdefg'}
-        volume = mock.Mock(provider_location='id^lun_41', id='id_41')
+        volume = MockOSResource(provider_location='id^lun_41', id='id_41')
         ret = self.adapter.terminate_connection(volume, connector)
         self.assertEqual('fibre_channel', ret['driver_volume_type'])
         data = ret['data']
@@ -631,7 +827,7 @@ class ISCSIAdapterTest(unittest.TestCase):
 
     @patch_for_iscsi_adapter
     def test_initialize_connection_volume(self):
-        volume = mock.Mock(provider_location='id^lun_43', id='id_43')
+        volume = MockOSResource(provider_location='id^lun_43', id='id_43')
         connector = {'host': 'host1'}
         conn_info = self.adapter.initialize_connection(volume, connector)
         self.assertEqual('iscsi', conn_info['driver_volume_type'])
@@ -640,7 +836,7 @@ class ISCSIAdapterTest(unittest.TestCase):
 
     @patch_for_iscsi_adapter
     def test_initialize_connection_snapshot(self):
-        snap = mock.Mock(id='snap_1', name='snap_1')
+        snap = MockOSResource(id='snap_1', name='snap_1')
         connector = {'host': 'host1'}
         conn_info = self.adapter.initialize_connection_snapshot(
             snap, connector)
diff --git a/cinder/tests/unit/volume/drivers/dell_emc/unity/test_client.py b/cinder/tests/unit/volume/drivers/dell_emc/unity/test_client.py
index e69bbb61653..4094d2f83ac 100644
--- a/cinder/tests/unit/volume/drivers/dell_emc/unity/test_client.py
+++ b/cinder/tests/unit/volume/drivers/dell_emc/unity/test_client.py
@@ -47,6 +47,7 @@ class MockResource(object):
         self.max_iops = None
         self.max_kbps = None
         self.pool_name = 'Pool0'
+        self._storage_resource = None
 
     @property
     def id(self):
@@ -159,12 +160,23 @@ class MockResource(object):
 
     @property
     def storage_resource(self):
-        return MockResource(_id='sr_%s' % self._id,
-                            name='sr_%s' % self.name)
+        if self._storage_resource is None:
+            self._storage_resource = MockResource(_id='sr_%s' % self._id,
+                                                  name='sr_%s' % self.name)
+        return self._storage_resource
+
+    @storage_resource.setter
+    def storage_resource(self, value):
+        self._storage_resource = value
 
     def modify(self, name=None):
         self.name = name
 
+    def thin_clone(self, name, io_limit_policy=None, description=None):
+        if name == 'thin_clone_name_in_use':
+            raise ex.UnityLunNameInUseError
+        return MockResource(_id=name, name=name)
+
 
 class MockResourceList(object):
     def __init__(self, names=None, ids=None):
@@ -204,10 +216,20 @@ class MockSystem(object):
         self.serial_number = 'SYSTEM_SERIAL'
         self.system_version = '4.1.0'
 
+    @property
+    def info(self):
+        mocked_info = mock.Mock()
+        mocked_info.name = self.serial_number
+        return mocked_info
+
     @staticmethod
     def get_lun(_id=None, name=None):
         if _id == 'not_found':
             raise ex.UnityResourceNotFoundError()
+        if _id == 'tc_80':  # for thin clone with extending size
+            lun = MockResource(name=_id, _id=_id)
+            lun.total_size_gb = 7
+            return lun
         return MockResource(name, _id)
 
     @staticmethod
@@ -299,6 +321,26 @@ class ClientTest(unittest.TestCase):
         lun = self.client.create_lun('LUN 4', 6, pool, io_limit_policy=limit)
         self.assertEqual(100, lun.max_kbps)
 
+    def test_thin_clone_success(self):
+        name = 'tc_77'
+        src_lun = MockResource(_id='id_77')
+        lun = self.client.thin_clone(src_lun, name)
+        self.assertEqual(name, lun.name)
+
+    def test_thin_clone_name_in_used(self):
+        name = 'thin_clone_name_in_use'
+        src_lun = MockResource(_id='id_79')
+        lun = self.client.thin_clone(src_lun, name)
+        self.assertEqual(name, lun.name)
+
+    def test_thin_clone_extend_size(self):
+        name = 'tc_80'
+        src_lun = MockResource(_id='id_80')
+        lun = self.client.thin_clone(src_lun, name, io_limit_policy=None,
+                                     new_size_gb=7)
+        self.assertEqual(name, lun.name)
+        self.assertEqual(7, lun.total_size_gb)
+
     def test_delete_lun_normal(self):
         self.assertIsNone(self.client.delete_lun('lun3'))
 
diff --git a/cinder/tests/unit/volume/drivers/dell_emc/unity/test_driver.py b/cinder/tests/unit/volume/drivers/dell_emc/unity/test_driver.py
index b1ce7721bd1..e9cc913805b 100644
--- a/cinder/tests/unit/volume/drivers/dell_emc/unity/test_driver.py
+++ b/cinder/tests/unit/volume/drivers/dell_emc/unity/test_driver.py
@@ -15,10 +15,9 @@
 
 import unittest
 
-import mock
-
 from cinder.tests.unit.volume.drivers.dell_emc.unity \
     import fake_exception as ex
+from cinder.tests.unit.volume.drivers.dell_emc.unity import test_adapter
 from cinder.volume import configuration as conf
 from cinder.volume.drivers.dell_emc.unity import driver
 
@@ -107,11 +106,12 @@ class MockAdapter(object):
 class UnityDriverTest(unittest.TestCase):
     @staticmethod
     def get_volume():
-        return mock.Mock(provider_location='id^lun_43', id='id_43')
+        return test_adapter.MockOSResource(provider_location='id^lun_43',
+                                           id='id_43')
 
     @classmethod
     def get_snapshot(cls):
-        return mock.Mock(volume=cls.get_volume())
+        return test_adapter.MockOSResource(volume=cls.get_volume())
 
     @staticmethod
     def get_context():
diff --git a/cinder/tests/unit/volume/drivers/dell_emc/unity/test_utils.py b/cinder/tests/unit/volume/drivers/dell_emc/unity/test_utils.py
index abb10d7dd28..7eb0c2fcc39 100644
--- a/cinder/tests/unit/volume/drivers/dell_emc/unity/test_utils.py
+++ b/cinder/tests/unit/volume/drivers/dell_emc/unity/test_utils.py
@@ -20,6 +20,7 @@ import mock
 from oslo_utils import units
 
 from cinder import exception
+from cinder.tests.unit.volume.drivers.dell_emc.unity import test_adapter
 from cinder.volume.drivers.dell_emc.unity import utils
 
 
@@ -185,7 +186,7 @@ class UnityUtilsTest(unittest.TestCase):
         self.assertEqual(targets, mapping['200000051e55a121'])
 
     def test_get_pool_name(self):
-        volume = mock.Mock(host='host@backend#pool_name')
+        volume = test_adapter.MockOSResource(host='host@backend#pool_name')
         self.assertEqual('pool_name', utils.get_pool_name(volume))
 
     def test_ignore_exception(self):
@@ -217,38 +218,39 @@ class UnityUtilsTest(unittest.TestCase):
         self.assertEqual(9, data[0])
 
     def test_get_backend_qos_specs_type_none(self):
-        volume = mock.Mock(volume_type_id=None)
+        volume = test_adapter.MockOSResource(volume_type_id=None)
         ret = utils.get_backend_qos_specs(volume)
         self.assertIsNone(ret)
 
     @patch_volume_types
     def test_get_backend_qos_specs_none(self):
-        volume = mock.Mock(volume_type_id='no_qos')
+        volume = test_adapter.MockOSResource(volume_type_id='no_qos')
         ret = utils.get_backend_qos_specs(volume)
         self.assertIsNone(ret)
 
     @patch_volume_types
     def test_get_backend_qos_invalid_consumer(self):
-        volume = mock.Mock(volume_type_id='invalid_backend_qos_consumer')
+        volume = test_adapter.MockOSResource(
+            volume_type_id='invalid_backend_qos_consumer')
         ret = utils.get_backend_qos_specs(volume)
         self.assertIsNone(ret)
 
     @patch_volume_types
     def test_get_backend_qos_both_none(self):
-        volume = mock.Mock(volume_type_id='both_none')
+        volume = test_adapter.MockOSResource(volume_type_id='both_none')
         ret = utils.get_backend_qos_specs(volume)
         self.assertIsNone(ret)
 
     @patch_volume_types
     def test_get_backend_qos_iops(self):
-        volume = mock.Mock(volume_type_id='max_1000_iops')
+        volume = test_adapter.MockOSResource(volume_type_id='max_1000_iops')
         ret = utils.get_backend_qos_specs(volume)
         expected = {'maxBWS': None, 'id': 'max_1000_iops', 'maxIOPS': 1000}
         self.assertEqual(expected, ret)
 
     @patch_volume_types
     def test_get_backend_qos_mbps(self):
-        volume = mock.Mock(volume_type_id='max_2_mbps')
+        volume = test_adapter.MockOSResource(volume_type_id='max_2_mbps')
         ret = utils.get_backend_qos_specs(volume)
         expected = {'maxBWS': 2, 'id': 'max_2_mbps', 'maxIOPS': None}
         self.assertEqual(expected, ret)
diff --git a/cinder/volume/drivers/dell_emc/unity/adapter.py b/cinder/volume/drivers/dell_emc/unity/adapter.py
index 65d4938375f..14763c19827 100644
--- a/cinder/volume/drivers/dell_emc/unity/adapter.py
+++ b/cinder/volume/drivers/dell_emc/unity/adapter.py
@@ -14,11 +14,22 @@
 # under the License.
 
 import contextlib
+import copy
 import functools
+import os
 import random
 
+from oslo_config import cfg
 from oslo_log import log as logging
 from oslo_utils import excutils
+from oslo_utils import importutils
+
+storops = importutils.try_import('storops')
+if storops:
+    from storops import exception as storops_ex
+else:
+    # Set storops_ex to be None for unit test
+    storops_ex = None
 
 from cinder import exception
 from cinder.i18n import _
@@ -33,6 +44,77 @@ PROTOCOL_FC = 'FC'
 PROTOCOL_ISCSI = 'iSCSI'
 
 
+class VolumeParams(object):
+    def __init__(self, adapter, volume):
+        self._adapter = adapter
+        self._volume = volume
+
+        self._volume_id = volume.id
+        self._name = volume.name
+        self._size = volume.size
+        self._description = (volume.display_description
+                             if volume.display_description
+                             else volume.display_name)
+        self._pool = None
+        self._io_limit_policy = None
+
+    @property
+    def volume_id(self):
+        return self._volume_id
+
+    @property
+    def name(self):
+        return self._name
+
+    @name.setter
+    def name(self, value):
+        self._name = value
+
+    @property
+    def size(self):
+        return self._size
+
+    @size.setter
+    def size(self, value):
+        self._size = value
+
+    @property
+    def description(self):
+        return self._description
+
+    @description.setter
+    def description(self, value):
+        self._description = value
+
+    @property
+    def pool(self):
+        if self._pool is None:
+            self._pool = self._adapter._get_target_pool(self._volume)
+        return self._pool
+
+    @pool.setter
+    def pool(self, value):
+        self._pool = value
+
+    @property
+    def io_limit_policy(self):
+        if self._io_limit_policy is None:
+            qos_specs = utils.get_backend_qos_specs(self._volume)
+            self._io_limit_policy = self._adapter.client.get_io_limit_policy(
+                qos_specs)
+        return self._io_limit_policy
+
+    @io_limit_policy.setter
+    def io_limit_policy(self, value):
+        self._io_limit_policy = value
+
+    def __eq__(self, other):
+        return (self.volume_id == other.volume_id
+                and self.name == other.name
+                and self.size == other.size
+                and self.io_limit_policy == other.io_limit_policy)
+
+
 class CommonAdapter(object):
     protocol = 'unknown'
     driver_name = 'UnityAbstractDriver'
@@ -84,6 +166,13 @@ class CommonAdapter(object):
 
         self.allowed_ports = self.validate_ports(self.config.unity_io_ports)
 
+        group_name = (self.config.config_group if self.config.config_group
+                      else 'DEFAULT')
+        folder_name = '%(group)s.%(sys_name)s' % {
+            'group': group_name, 'sys_name': self.client.system.info.name}
+        persist_path = os.path.join(cfg.CONF.state_path, 'unity', folder_name)
+        storops.TCHelper.set_up(persist_path)
+
     def normalize_config(self, config):
         config.unity_storage_pool_names = utils.remove_empty(
             '%s.unity_storage_pool_names' % config.config_group,
@@ -92,6 +181,7 @@ class CommonAdapter(object):
         config.unity_io_ports = utils.remove_empty(
             '%s.unity_io_ports' % config.config_group,
             config.unity_io_ports)
+
         return config
 
     def get_all_ports(self):
@@ -159,36 +249,32 @@ class CommonAdapter(object):
         valid_names = utils.validate_pool_names(names, array_pools.name)
         return {p.name: p for p in array_pools if p.name in valid_names}
 
+    def makeup_model(self, lun, is_snap_lun=False):
+        lun_type = 'snap_lun' if is_snap_lun else 'lun'
+        location = self._build_provider_location(lun_id=lun.get_id(),
+                                                 lun_type=lun_type)
+        return {
+            'provider_location': location,
+            'provider_id': lun.get_id()
+        }
+
     def create_volume(self, volume):
         """Creates a volume.
 
         :param volume: volume information
         """
-        volume_size = volume.size
-        volume_name = volume.name
-        volume_description = (volume.display_description
-                              if volume.display_description
-                              else volume.display_name)
+        params = VolumeParams(self, volume)
 
-        pool = self._get_target_pool(volume)
-        qos_specs = utils.get_backend_qos_specs(volume)
-        limit_policy = self.client.get_io_limit_policy(qos_specs)
+        LOG.info('Create Volume: %(name)s, size: %(size)s, description: '
+                 '%(description)s, pool: %(pool)s, io limit policy: '
+                 '%(io_limit_policy)s.', params)
 
-        LOG.info('Create Volume: %(volume)s  Size: %(size)s '
-                 'Pool: %(pool)s Qos: %(qos)s.',
-                 {'volume': volume_name,
-                  'size': volume_size,
-                  'pool': pool.name,
-                  'qos': qos_specs})
-
-        lun = self.client.create_lun(
-            volume_name, volume_size, pool, description=volume_description,
-            io_limit_policy=limit_policy)
-        location = self._build_provider_location(
-            lun_type='lun',
-            lun_id=lun.get_id())
-        return {'provider_location': location,
-                'provider_id': lun.get_id()}
+        return self.makeup_model(
+            self.client.create_lun(name=params.name,
+                                   size=params.size,
+                                   pool=params.pool,
+                                   description=params.description,
+                                   io_limit_policy=params.io_limit_policy))
 
     def delete_volume(self, volume):
         lun_id = self.get_lun_id(volume)
@@ -364,10 +450,12 @@ class CommonAdapter(object):
         """
         lun = self._get_referenced_lun(existing_ref)
         lun.modify(name=volume.name)
-        return {'provider_location':
+        return {
+            'provider_location':
                 self._build_provider_location(lun_id=lun.get_id(),
                                               lun_type='lun'),
-                'provider_id': lun.get_id()}
+            'provider_id': lun.get_id()
+        }
 
     def manage_existing_get_size(self, volume, existing_ref):
         """Returns size of volume to be managed by `manage_existing`.
@@ -424,30 +512,32 @@ class CommonAdapter(object):
                                       True) as attach_info:
                 yield attach_info
 
-    def _create_volume_from_snap(self, volume, snap, size_in_m=None):
-        """Creates a volume from a Unity snapshot.
+    def _dd_copy(self, vol_params, src_snap, src_lun=None):
+        """Creates a volume via copying a Unity snapshot.
 
         It attaches the `volume` and `snap`, then use `dd` to copy the
         data from the Unity snapshot to the `volume`.
         """
-        model_update = self.create_volume(volume)
-        # Update `provider_location` and `provider_id` of `volume` explicitly.
-        volume.update(model_update)
-        src_id = snap.get_id()
-        dest_lun = self.client.get_lun(lun_id=self.get_lun_id(volume))
+        dest_lun = self.client.create_lun(
+            name=vol_params.name, size=vol_params.size, pool=vol_params.pool,
+            description=vol_params.description,
+            io_limit_policy=vol_params.io_limit_policy)
+        src_id = src_snap.get_id()
         try:
             conn_props = cinder_utils.brick_get_connector_properties()
 
             with self._connect_resource(dest_lun, conn_props,
-                                        volume.id) as dest_info, \
-                    self._connect_resource(snap, conn_props,
+                                        vol_params.volume_id) as dest_info, \
+                    self._connect_resource(src_snap, conn_props,
                                            src_id) as src_info:
-                if size_in_m is None:
+                if src_lun is None:
                     # If size is not specified, need to get the size from LUN
                     # of snapshot.
                     lun = self.client.get_lun(
-                        lun_id=snap.storage_resource.get_id())
+                        lun_id=src_snap.storage_resource.get_id())
                     size_in_m = utils.byte_to_mib(lun.size_total)
+                else:
+                    size_in_m = utils.byte_to_mib(src_lun.size_total)
                 vol_utils.copy_volume(
                     src_info['device']['path'],
                     dest_info['device']['path'],
@@ -456,35 +546,82 @@ class CommonAdapter(object):
                     sparse=True)
         except Exception:
             with excutils.save_and_reraise_exception():
-                utils.ignore_exception(self.delete_volume, volume)
+                utils.ignore_exception(self.client.delete_lun,
+                                       dest_lun.get_id())
                 LOG.error('Failed to create cloned volume: %(vol_id)s, '
                           'from source unity snapshot: %(snap_name)s.',
-                          {'vol_id': volume.id, 'snap_name': snap.name})
+                          {'vol_id': vol_params.volume_id,
+                           'snap_name': src_snap.name})
 
-        return model_update
+        return dest_lun
+
+    def _thin_clone(self, vol_params, src_snap, src_lun=None):
+        tc_src = src_snap if src_lun is None else src_lun
+        try:
+            LOG.debug('Try to thin clone from %s.', tc_src.name)
+            lun = self.client.thin_clone(
+                tc_src, vol_params.name,
+                description=vol_params.description,
+                io_limit_policy=vol_params.io_limit_policy,
+                new_size_gb=vol_params.size)
+        except storops_ex.UnityThinCloneLimitExceededError:
+            LOG.info('Number of thin clones of base LUN exceeds system '
+                     'limit, dd-copy a new one and thin clone from it.')
+            # Copy via dd if thin clone meets the system limit
+            hidden = copy.copy(vol_params)
+            hidden.name = 'hidden-%s' % vol_params.name
+            hidden.description = 'hidden-%s' % vol_params.description
+            copied_lun = self._dd_copy(hidden, src_snap, src_lun=src_lun)
+            LOG.debug('Notify storops the dd action of lun: %(src_name)s. And '
+                      'the newly copied lun is: %(copied)s.',
+                      {'src_name': tc_src.name, 'copied': copied_lun.name})
+            storops.TCHelper.notify(tc_src,
+                                    storops.ThinCloneActionEnum.DD_COPY,
+                                    copied_lun)
+            lun = self.client.thin_clone(
+                copied_lun, vol_params.name,
+                description=vol_params.description,
+                io_limit_policy=vol_params.io_limit_policy,
+                new_size_gb=vol_params.size)
+        except storops_ex.SystemAPINotSupported:
+            # Thin clone not support on array version before Merlin
+            lun = self._dd_copy(vol_params, src_snap, src_lun=src_lun)
+            LOG.debug(
+                'Volume copied via dd because array OE is too old to support '
+                'thin clone api. source snap: %(src_snap)s, lun: %(src_lun)s.',
+                {'src_snap': src_snap.name,
+                 'src_lun': 'Unknown' if src_lun is None else src_lun.name})
+        return lun
 
     def create_volume_from_snapshot(self, volume, snapshot):
         snap = self.client.get_snap(snapshot.name)
-        return self._create_volume_from_snap(volume, snap)
+        return self.makeup_model(
+            self._thin_clone(VolumeParams(self, volume), snap),
+            is_snap_lun=True)
 
     def create_cloned_volume(self, volume, src_vref):
         """Creates cloned volume.
 
         1. Take an internal snapshot of source volume, and attach it.
-        2. Create a new volume, and attach it.
-        3. Copy from attached snapshot of step 1 to the volume of step 2.
-        4. Delete the internal snapshot created in step 1.
+        2. Thin clone from the snapshot to a new volume.
+           Note: there are several cases the thin clone will downgrade to `dd`,
+           2.1 Source volume is attached (in-use).
+           2.2 Array OE version doesn't support thin clone.
+           2.3 The current LUN family reaches the thin clone limits.
+        3. Delete the internal snapshot created in step 1.
         """
 
         src_lun_id = self.get_lun_id(src_vref)
         if src_lun_id is None:
             raise exception.VolumeBackendAPIException(
-                data=_("LUN ID of source volume: %s not found.") %
-                src_vref.name)
+                data=_(
+                    "LUN ID of source volume: %s not found.") % src_vref.name)
+        src_lun = self.client.get_lun(lun_id=src_lun_id)
         src_snap_name = 'snap_clone_%s' % volume.id
 
         create_snap_func = functools.partial(self.client.create_snap,
                                              src_lun_id, src_snap_name)
+        vol_params = VolumeParams(self, volume)
         with utils.assure_cleanup(create_snap_func,
                                   self.client.delete_snap,
                                   True) as src_snap:
@@ -492,8 +629,16 @@ class CommonAdapter(object):
                       'name: %(name)s, id: %(id)s.',
                       {'name': src_snap_name,
                        'id': src_snap.get_id()})
-            return self._create_volume_from_snap(
-                volume, src_snap, size_in_m=utils.gib_to_mib(volume.size))
+            if src_vref.volume_attachment:
+                lun = self._dd_copy(vol_params, src_snap, src_lun=src_lun)
+                LOG.debug('Volume copied using dd because source volume: '
+                          '%(name)s is attached: %(attach)s.',
+                          {'name': src_vref.name,
+                           'attach': src_vref.volume_attachment})
+                return self.makeup_model(lun)
+            else:
+                lun = self._thin_clone(vol_params, src_snap, src_lun=src_lun)
+                return self.makeup_model(lun, is_snap_lun=True)
 
     def get_pool_name(self, volume):
         return self.client.get_pool_name(volume.name)
diff --git a/cinder/volume/drivers/dell_emc/unity/client.py b/cinder/volume/drivers/dell_emc/unity/client.py
index d0f12dcf8b3..c4ce0ff9284 100644
--- a/cinder/volume/drivers/dell_emc/unity/client.py
+++ b/cinder/volume/drivers/dell_emc/unity/client.py
@@ -76,6 +76,20 @@ class UnityClient(object):
             lun = self.system.get_lun(name=name)
         return lun
 
+    def thin_clone(self, lun_or_snap, name, io_limit_policy=None,
+                   description=None, new_size_gb=None):
+        try:
+            lun = lun_or_snap.thin_clone(
+                name=name, io_limit_policy=io_limit_policy,
+                description=description)
+        except storops_ex.UnityLunNameInUseError:
+            LOG.debug("LUN(thin clone) %s already exists. "
+                      "Return the existing one.", name)
+            lun = self.system.get_lun(name=name)
+        if new_size_gb is not None and new_size_gb > lun.total_size_gb:
+            lun = self.extend_lun(lun.get_id(), new_size_gb)
+        return lun
+
     def delete_lun(self, lun_id):
         """Deletes LUN on the Unity system.
 
diff --git a/cinder/volume/drivers/dell_emc/unity/driver.py b/cinder/volume/drivers/dell_emc/unity/driver.py
index 6cb35483406..0862a119ff2 100644
--- a/cinder/volume/drivers/dell_emc/unity/driver.py
+++ b/cinder/volume/drivers/dell_emc/unity/driver.py
@@ -49,9 +49,10 @@ class UnityDriver(driver.ManageableVD,
 
     Version history:
         1.0.0 - Initial version
+        2.0.0 - Add thin clone support
     """
 
-    VERSION = '01.00.00'
+    VERSION = '02.00.00'
     VENDOR = 'Dell EMC'
     # ThirdPartySystems wiki page
     CI_WIKI_NAME = "EMC_UNITY_CI"
diff --git a/releasenotes/notes/unity-fast-clone-02ae88ba8fdef145.yaml b/releasenotes/notes/unity-fast-clone-02ae88ba8fdef145.yaml
new file mode 100644
index 00000000000..85835b8f738
--- /dev/null
+++ b/releasenotes/notes/unity-fast-clone-02ae88ba8fdef145.yaml
@@ -0,0 +1,8 @@
+---
+features:
+  - Add thin clone support in the Unity driver. Unity storage supports the thin
+    clone of a LUN from OE version 4.2.0. It is more efficient than the dd
+    solution. However, there is a limit of thin clone inside
+    each LUN family. Every time the limit reaches, a new LUN family will be
+    created by a dd-copy, and then the volume clone afterward will use the
+    thin clone of the new LUN family.