diff --git a/images/cinder/Dockerfile b/images/cinder/Dockerfile
index f954b3a..4d5addf 100644
--- a/images/cinder/Dockerfile
+++ b/images/cinder/Dockerfile
@@ -18,6 +18,8 @@
 ARG CINDER_GIT_REF=f74e2729554bee01b0a3e631a8001bb39e540433
 ADD --keep-git-dir=true https://opendev.org/openstack/cinder.git#${CINDER_GIT_REF} /src/cinder
 RUN git -C /src/cinder fetch --unshallow
+COPY patches/cinder /patches/cinder
+RUN git -C /src/cinder apply --verbose /patches/cinder/*
 RUN --mount=type=cache,mode=0755,target=/root/.cache/pip,sharing=private <<EOF bash -xe
 pip3 install \
     --constraint /upper-constraints.txt \
diff --git a/images/cinder/patches/cinder/0001-Create-encrypted-volumes-directly-to-RBD.patch b/images/cinder/patches/cinder/0001-Create-encrypted-volumes-directly-to-RBD.patch
new file mode 100644
index 0000000..fd8370a
--- /dev/null
+++ b/images/cinder/patches/cinder/0001-Create-encrypted-volumes-directly-to-RBD.patch
@@ -0,0 +1,146 @@
+From 96cca9076fb95b2fae7dbf254d49f492ce02b69a Mon Sep 17 00:00:00 2001
+From: ricolin <ricolin@ricolky.com>
+Date: Fri, 1 Mar 2024 13:50:13 +0800
+Subject: [PATCH 1/3] Create encrypted volumes directly to RBD
+
+This fix slow on create encrypted volumes with temp file import.
+Encrypted volume create is now directly upload to RBD with qemu-img
+command without temprory image file generated.
+
+Closes-Bug: #2055517
+Change-Id: If7a72a4acd5600de1350289a9d9c38017d42659e
+---
+ cinder/tests/unit/volume/drivers/test_rbd.py  |  9 +--
+ cinder/volume/drivers/rbd.py                  | 62 +++++++++----------
+ ...ate-encrypted-volume-c1bb6b44b85c0242.yaml |  7 +++
+ 3 files changed, 40 insertions(+), 38 deletions(-)
+ create mode 100644 releasenotes/notes/improve-create-encrypted-volume-c1bb6b44b85c0242.yaml
+
+diff --git a/cinder/tests/unit/volume/drivers/test_rbd.py b/cinder/tests/unit/volume/drivers/test_rbd.py
+index 57dae3af1..38ceaf27f 100644
+--- a/cinder/tests/unit/volume/drivers/test_rbd.py
++++ b/cinder/tests/unit/volume/drivers/test_rbd.py
+@@ -3247,7 +3247,6 @@ class RBDTestCase(test.TestCase):
+                 self.__dict__ = d
+ 
+         mock_temp_file.return_value.__enter__.side_effect = [
+-            DictObj({'name': '/imgfile'}),
+             DictObj({'name': '/passfile'})]
+ 
+         key_mgr = fake_keymgr.fake_api()
+@@ -3268,15 +3267,13 @@ class RBDTestCase(test.TestCase):
+                                                  self.context)
+             mock_open.assert_called_with('/passfile', 'w')
+ 
+-            mock_exec.assert_any_call(
++            mock_exec.assert_called_with(
+                 'qemu-img', 'create', '-f', 'luks', '-o',
+                 'cipher-alg=aes-256,cipher-mode=xts,ivgen-alg=essiv',
+                 '--object',
+                 'secret,id=luks_sec,format=raw,file=/passfile',
+-                '-o', 'key-secret=luks_sec', '/imgfile', '12288M')
+-            mock_exec.assert_any_call(
+-                'rbd', 'import', '--dest-pool', 'rbd', '--order', 22,
+-                '/imgfile', self.volume_c.name)
++                '-o', 'key-secret=luks_sec', 'rbd:rbd/%s' % self.volume_c.name,
++                '12288M')
+ 
+     @mock.patch('cinder.objects.Volume.get_by_id')
+     @mock.patch('cinder.db.volume_glance_metadata_get', return_value={})
+diff --git a/cinder/volume/drivers/rbd.py b/cinder/volume/drivers/rbd.py
+index 6cc86c2c5..b883ee47a 100644
+--- a/cinder/volume/drivers/rbd.py
++++ b/cinder/volume/drivers/rbd.py
+@@ -1087,8 +1087,8 @@ class RBDDriver(driver.CloneableImageVD, driver.MigrateVD,
+                                  context: context.RequestContext) -> None:
+         """Create an encrypted volume.
+ 
+-        This works by creating an encrypted image locally,
+-        and then uploading it to the volume.
++        This works by creating an encrypted image and
++        then uploading it to the volume directly.
+         """
+         encryption = volume_utils.check_encryption_provider(volume, context)
+ 
+@@ -1100,37 +1100,35 @@ class RBDDriver(driver.CloneableImageVD, driver.MigrateVD,
+         # create a file
+         tmp_dir = volume_utils.image_conversion_dir()
+ 
+-        with tempfile.NamedTemporaryFile(dir=tmp_dir) as tmp_image:
+-            with tempfile.NamedTemporaryFile(dir=tmp_dir) as tmp_key:
+-                with open(tmp_key.name, 'w') as f:
+-                    f.write(passphrase)
+-
+-                cipher_spec = image_utils.decode_cipher(encryption['cipher'],
+-                                                        encryption['key_size'])
+-
+-                create_cmd = (
+-                    'qemu-img', 'create', '-f', 'luks',
+-                    '-o', 'cipher-alg=%(cipher_alg)s,'
+-                    'cipher-mode=%(cipher_mode)s,'
+-                    'ivgen-alg=%(ivgen_alg)s' % cipher_spec,
+-                    '--object', 'secret,id=luks_sec,'
+-                    'format=raw,file=%(passfile)s' % {'passfile':
+-                                                      tmp_key.name},
+-                    '-o', 'key-secret=luks_sec',
+-                    tmp_image.name,
+-                    '%sM' % (volume.size * 1024))
+-                self._execute(*create_cmd)
+-
+-            # Copy image into RBD
+-            chunk_size = self.configuration.rbd_store_chunk_size * units.Mi
+-            order = int(math.log(chunk_size, 2))
++        with tempfile.NamedTemporaryFile(dir=tmp_dir) as tmp_key:
++            with open(tmp_key.name, 'w') as f:
++                f.write(passphrase)
+ 
+-            cmd = ['rbd', 'import',
+-                   '--dest-pool', self.configuration.rbd_pool,
+-                   '--order', order,
+-                   tmp_image.name, volume.name]
+-            cmd.extend(self._ceph_args())
+-            self._execute(*cmd)
++            cipher_spec = image_utils.decode_cipher(encryption['cipher'],
++                                                    encryption['key_size'])
++
++            _, conf, user_id, _ = self._get_config_tuple()
++            rbd_options = ''
++            if user_id:
++                rbd_options += ':id=%(user_id)s' % {'user_id': user_id}
++            if conf:
++                rbd_options += ':conf=%(conf)s' % {'conf': conf}
++            create_cmd = (
++                'qemu-img', 'create', '-f', 'luks',
++                '-o', 'cipher-alg=%(cipher_alg)s,'
++                'cipher-mode=%(cipher_mode)s,'
++                'ivgen-alg=%(ivgen_alg)s' % cipher_spec,
++                '--object', 'secret,id=luks_sec,'
++                'format=raw,file=%(passfile)s' % {'passfile':
++                                                  tmp_key.name},
++                '-o', 'key-secret=luks_sec',
++                'rbd:%(pool_name)s/%(image_name)s%(rbd_options)s' % {
++                    'pool_name': self.configuration.rbd_pool,
++                    'image_name': volume.name,
++                    'rbd_options': rbd_options
++                },
++                '%sM' % (volume.size * 1024))
++            self._execute(*create_cmd)
+ 
+     def create_volume(self, volume: Volume) -> dict[str, Any]:
+         """Creates a logical volume."""
+diff --git a/releasenotes/notes/improve-create-encrypted-volume-c1bb6b44b85c0242.yaml b/releasenotes/notes/improve-create-encrypted-volume-c1bb6b44b85c0242.yaml
+new file mode 100644
+index 000000000..8bdff6746
+--- /dev/null
++++ b/releasenotes/notes/improve-create-encrypted-volume-c1bb6b44b85c0242.yaml
+@@ -0,0 +1,7 @@
++---
++fixes:
++  - |
++    [Bug 255517](https://bugs.launchpad.net/cinder/+bug/2055517): Fix slow
++    on create encrypted volumes with temp file import. Encrypted volume create
++    is now directly upload to rbd with qemu-img command without temprory image
++    file generated.
+-- 
+2.25.1
+
diff --git a/images/cinder/patches/cinder/0002-Allow-clone-encrypted-image-to-encrypted-volume.patch b/images/cinder/patches/cinder/0002-Allow-clone-encrypted-image-to-encrypted-volume.patch
new file mode 100644
index 0000000..60c75d1
--- /dev/null
+++ b/images/cinder/patches/cinder/0002-Allow-clone-encrypted-image-to-encrypted-volume.patch
@@ -0,0 +1,130 @@
+From 410d45106cfc20c064380c3914ff8c2cd41a1a5e Mon Sep 17 00:00:00 2001
+From: ricolin <rlin@vexxhost.com>
+Date: Sat, 16 Mar 2024 00:35:12 +0800
+Subject: [PATCH 2/3] Allow clone encrypted image to encrypted volume
+
+Exactly like what we did in copy-and-import image when create encrypted
+volume from encrypted image. If the image is encrypted, we will copy
+`cinder_encryption_key_id` from image metadata to volume. That means we
+should be safe to try directly clone from encrypted image.
+
+Related-Bug: #2055517
+Change-Id: Id6a1452c2c197a58677bf181470f54565fbd263b
+---
+ .../volume/flows/test_create_volume_flow.py   | 46 +++++++++++++++++++
+ cinder/volume/flows/manager/create_volume.py  |  9 +++-
+ ...clone-encryped-image-6961ca1439825dc4.yaml |  8 ++++
+ 3 files changed, 61 insertions(+), 2 deletions(-)
+ create mode 100644 releasenotes/notes/allow-clone-encryped-image-6961ca1439825dc4.yaml
+
+diff --git a/cinder/tests/unit/volume/flows/test_create_volume_flow.py b/cinder/tests/unit/volume/flows/test_create_volume_flow.py
+index 5b4ddb35f..7e08f8e78 100644
+--- a/cinder/tests/unit/volume/flows/test_create_volume_flow.py
++++ b/cinder/tests/unit/volume/flows/test_create_volume_flow.py
+@@ -1202,6 +1202,7 @@ class CreateVolumeFlowManagerTestCase(test.TestCase):
+             encryption_key_id=fakes.ENCRYPTION_KEY_ID,
+             host='host@backend#pool')
+ 
++        fake_driver.clone_image.return_value = (None, False)
+         fake_image_service = fake_image.FakeImageService()
+         image_meta = {}
+         image_id = fakes.IMAGE_ID
+@@ -1218,6 +1219,7 @@ class CreateVolumeFlowManagerTestCase(test.TestCase):
+                                         image_meta, fake_image_service)
+ 
+         fake_driver.create_volume.assert_called_once_with(volume)
++        fake_driver.clone_image.assert_called_once()
+         fake_driver.copy_image_to_encrypted_volume.assert_not_called()
+         fake_driver.copy_image_to_volume.assert_called_once_with(
+             self.ctxt, volume, fake_image_service, image_id)
+@@ -1226,6 +1228,50 @@ class CreateVolumeFlowManagerTestCase(test.TestCase):
+                                                      image_meta=image_meta)
+         mock_cleanup_cg.assert_called_once_with(volume)
+ 
++    @mock.patch('cinder.volume.flows.manager.create_volume.'
++                'CreateVolumeFromSpecTask.'
++                '_handle_bootable_volume_glance_meta')
++    @mock.patch('cinder.image.image_utils.TemporaryImages.fetch')
++    @mock.patch('cinder.image.image_utils.qemu_img_info')
++    @mock.patch('cinder.image.image_utils.check_virtual_size')
++    def test_create_encrypted_volume_from_enc_image_clone(
++        self, mock_check_size, mock_qemu_img,
++        mock_fetch_img, mock_handle_bootable
++    ):
++        fake_db = mock.MagicMock()
++        fake_driver = mock.MagicMock()
++        fake_volume_manager = mock.MagicMock()
++        fake_manager = create_volume_manager.CreateVolumeFromSpecTask(
++            fake_volume_manager, fake_db, fake_driver)
++        volume = fake_volume.fake_volume_obj(
++            self.ctxt,
++            encryption_key_id=fakes.ENCRYPTION_KEY_ID,
++            host='host@backend#pool')
++
++        fake_driver.clone_image.return_value = (None, True)
++        fake_image_service = fake_image.FakeImageService()
++        image_meta = {}
++        image_id = fakes.IMAGE_ID
++        image_meta['id'] = image_id
++        image_meta['status'] = 'active'
++        image_meta['size'] = 1
++        image_meta['cinder_encryption_key_id'] = \
++            '00000000-0000-0000-0000-000000000000'
++        image_location = 'abc'
++
++        fake_db.volume_update.return_value = volume
++        fake_manager._create_from_image(self.ctxt, volume,
++                                        image_location, image_id,
++                                        image_meta, fake_image_service)
++
++        fake_driver.create_volume.assert_not_called()
++        fake_driver.clone_image.assert_called_once()
++        fake_driver.copy_image_to_encrypted_volume.assert_not_called()
++        fake_driver.copy_image_to_volume.assert_not_called()
++        mock_handle_bootable.assert_called_once_with(self.ctxt, volume,
++                                                     image_id=image_id,
++                                                     image_meta=image_meta)
++
+     @ddt.data({'driver_error': True},
+               {'driver_error': False})
+     @mock.patch('cinder.backup.api.API.get_available_backup_service_host')
+diff --git a/cinder/volume/flows/manager/create_volume.py b/cinder/volume/flows/manager/create_volume.py
+index 0ae3cb59d..9d0590d9c 100644
+--- a/cinder/volume/flows/manager/create_volume.py
++++ b/cinder/volume/flows/manager/create_volume.py
+@@ -1087,11 +1087,16 @@ class CreateVolumeFromSpecTask(flow_utils.CinderTask):
+         # dict containing provider_location for cloned volume
+         # and clone status.
+         # NOTE (lixiaoy1): Currently all images are raw data, we can't
+-        # use clone_image to copy data if new volume is encrypted.
++        # use clone_image to copy data if new volume is encrypted
++        # NOTE (ricolin):  If the image provided an encryption key, we have
++        # already cloned it to the volume's key in
++        # _get_encryption_key_id, so we can do a direct clone.
++        image_encryption_key = image_meta.get('cinder_encryption_key_id')
+         volume_is_encrypted = volume.encryption_key_id is not None
+         cloned = False
+         model_update = None
+-        if not volume_is_encrypted:
++        if not volume_is_encrypted or (
++                volume_is_encrypted and image_encryption_key):
+             model_update, cloned = self.driver.clone_image(context,
+                                                            volume,
+                                                            image_location,
+diff --git a/releasenotes/notes/allow-clone-encryped-image-6961ca1439825dc4.yaml b/releasenotes/notes/allow-clone-encryped-image-6961ca1439825dc4.yaml
+new file mode 100644
+index 000000000..d6c7e8eb8
+--- /dev/null
++++ b/releasenotes/notes/allow-clone-encryped-image-6961ca1439825dc4.yaml
+@@ -0,0 +1,8 @@
++---
++features:
++  - |
++    Allow clone encrypted image when create encrypted volume from image.
++    Exactly like what we did in copy-and-import image when create encrypted
++    volume from encrypted image. If the image is encrypted, we will copy
++    `cinder_encryption_key_id` from image metadata to volume. That means we
++    should be safe to try directly clone from encrypted image.
+-- 
+2.25.1
+
diff --git a/images/cinder/patches/cinder/0003-Allow-encrypted-volume-clone-from-Glance-image.patch b/images/cinder/patches/cinder/0003-Allow-encrypted-volume-clone-from-Glance-image.patch
new file mode 100644
index 0000000..49ad536
--- /dev/null
+++ b/images/cinder/patches/cinder/0003-Allow-encrypted-volume-clone-from-Glance-image.patch
@@ -0,0 +1,320 @@
+From e72198cdf5b63fba70acb3ff43ec0b7531ff76da Mon Sep 17 00:00:00 2001
+From: ricolin <rlin@vexxhost.com>
+Date: Fri, 15 Mar 2024 23:26:14 +0800
+Subject: [PATCH 3/3] Allow encrypted volume clone from Glance image
+
+Allow clone image when creating encrypted volume from Glance image if both
+stored in RBD.
+Previously, Glance image clone is not supported for encrypted volume
+creation. The old process is to download image to local disk, encrypt the
+local file, and import it back to RBD. This not just slow, but also
+protentially take large amount of local disk space from hosts that runs
+Cinder volume service.
+The new process is to try and clone from Glance image (if it's also stored
+in RBD), flatten it, and encrypting new image in RBD for volume. And If
+Glance image source is not clonable, will continue with copy-and-import
+method as previous flow.
+In above flow, If clone from Glance image is appliable. Even it still
+requires to clone and flatten RBD image might took some time, but should
+still be a lot faster than copy-and-import. And also no local disk will
+be used to store raw image in this case.
+This also introduced driver method `clone_image_and_encrypt` for drivers
+that seperate the clone process from non-encrypted volume so the create
+flow won't be affected.
+
+Related-Bug: #2055517
+Change-Id: Ia023646d8bc9468bf5cc8955f7013299b2a3a460
+---
+ .../volume/flows/test_create_volume_flow.py   | 49 ++++++++++
+ cinder/volume/driver.py                       | 11 +++
+ cinder/volume/drivers/rbd.py                  | 95 ++++++++++++++++---
+ cinder/volume/flows/manager/create_volume.py  |  8 +-
+ ...for-encrypted-volume-de477647e9016b8b.yaml | 21 ++++
+ 5 files changed, 167 insertions(+), 17 deletions(-)
+ create mode 100644 releasenotes/notes/allow-clone-image-for-encrypted-volume-de477647e9016b8b.yaml
+
+diff --git a/cinder/tests/unit/volume/flows/test_create_volume_flow.py b/cinder/tests/unit/volume/flows/test_create_volume_flow.py
+index 7e08f8e78..04d9048ea 100644
+--- a/cinder/tests/unit/volume/flows/test_create_volume_flow.py
++++ b/cinder/tests/unit/volume/flows/test_create_volume_flow.py
+@@ -1164,6 +1164,7 @@ class CreateVolumeFlowManagerTestCase(test.TestCase):
+         image_location = 'abc'
+ 
+         fake_db.volume_update.return_value = volume
++        fake_driver.clone_image_and_encrypt.return_value = (None, False)
+         fake_manager._create_from_image(self.ctxt, volume,
+                                         image_location, image_id,
+                                         image_meta, fake_image_service)
+@@ -1177,6 +1178,54 @@ class CreateVolumeFlowManagerTestCase(test.TestCase):
+                                                      image_meta=image_meta)
+         mock_cleanup_cg.assert_called_once_with(volume)
+ 
++    @mock.patch('cinder.volume.flows.manager.create_volume.'
++                'CreateVolumeFromSpecTask.'
++                '_prepare_image_cache_entry')
++    @mock.patch('cinder.volume.flows.manager.create_volume.'
++                'CreateVolumeFromSpecTask.'
++                '_handle_bootable_volume_glance_meta')
++    @mock.patch('cinder.image.image_utils.TemporaryImages.fetch')
++    @mock.patch('cinder.image.image_utils.qemu_img_info')
++    @mock.patch('cinder.image.image_utils.check_virtual_size')
++    def test_create_encrypted_volume_from_image_clone(
++        self, mock_check_size, mock_qemu_img, mock_fetch_img,
++        mock_handle_bootable, mock_prepare_image_cache
++    ):
++        fake_db = mock.MagicMock()
++        fake_driver = mock.MagicMock()
++        fake_volume_manager = mock.MagicMock()
++        fake_cache = mock.MagicMock()
++        fake_manager = create_volume_manager.CreateVolumeFromSpecTask(
++            fake_volume_manager, fake_db, fake_driver, fake_cache)
++        volume = fake_volume.fake_volume_obj(
++            self.ctxt,
++            encryption_key_id=fakes.ENCRYPTION_KEY_ID,
++            host='host@backend#pool')
++
++        fake_image_service = fake_image.FakeImageService()
++        image_meta = {}
++        image_id = fakes.IMAGE_ID
++        image_meta['id'] = image_id
++        image_meta['status'] = 'active'
++        image_meta['size'] = 1
++        image_location = 'abc'
++
++        fake_db.volume_update.return_value = volume
++        fake_driver.clone_image_and_encrypt.return_value = (None, True)
++        fake_manager._create_from_image(self.ctxt, volume,
++                                        image_location, image_id,
++                                        image_meta, fake_image_service)
++
++        mock_prepare_image_cache.assert_not_called()
++        fake_driver.create_volume.assert_not_called()
++        fake_driver.clone_image.assert_not_called()
++        fake_driver.clone_image_and_encrypt.assert_called_once()
++        fake_driver.copy_image_to_encrypted_volume.assert_not_called()
++        fake_driver.copy_image_to_volume.assert_not_called()
++        mock_handle_bootable.assert_called_once_with(self.ctxt, volume,
++                                                     image_id=image_id,
++                                                     image_meta=image_meta)
++
+     @mock.patch('cinder.volume.flows.manager.create_volume.'
+                 'CreateVolumeFromSpecTask.'
+                 '_cleanup_cg_in_volume')
+diff --git a/cinder/volume/driver.py b/cinder/volume/driver.py
+index 0208b8be6..b0b669019 100644
+--- a/cinder/volume/driver.py
++++ b/cinder/volume/driver.py
+@@ -1167,6 +1167,17 @@ class BaseVD(object, metaclass=abc.ABCMeta):
+         """
+         return None, False
+ 
++    def clone_image_and_encrypt(
++        self, context, volume, image_location, image_meta, image_service
++    ):
++        """Create and encrypt a volume efficiently from an existing image.
++
++        Refer to
++        :obj:`cinder.interface.volume_driver.VolumeDriverCore.clone_image`
++        for additional information.
++        """
++        return None, False
++
+     def backup_use_temp_snapshot(self):
+         """Get the configured setting for backup from snapshot.
+ 
+diff --git a/cinder/volume/drivers/rbd.py b/cinder/volume/drivers/rbd.py
+index b883ee47a..855f32896 100644
+--- a/cinder/volume/drivers/rbd.py
++++ b/cinder/volume/drivers/rbd.py
+@@ -141,6 +141,13 @@ CONF.register_opts(RBD_OPTS, group=configuration.SHARED_CONF_GROUP)
+ EXTRA_SPECS_REPL_ENABLED = "replication_enabled"
+ EXTRA_SPECS_MULTIATTACH = "multiattach"
+ 
++# Note(ricolin): Reference ceph site for more information:
++# https://github.com/ceph/ceph/blob/main/src/include/rbd/librbd.h
++RBD_ENCRYPTION_ALG = {
++    'aes-128': 0,
++    'aes-256': 1
++}
++
+ QOS_KEY_MAP = {
+     'total_iops_sec': {
+         'ceph_key': 'rbd_qos_iops_limit',
+@@ -1188,6 +1195,20 @@ class RBDDriver(driver.CloneableImageVD, driver.MigrateVD,
+ 
+         return max(image_stripe_unit, default_stripe_unit)
+ 
++    def _encrypt_volume(self,
++                        context: context.RequestContext,
++                        volume: Volume,
++                        passphrase: str,
++                        cipher_spec: dict
++                        ) -> None:
++        LOG.debug("Encrypting volume $s", volume.name)
++        with RBDVolumeProxy(self, volume.name) as vol:
++            vol.encryption_format(
++                0,
++                passphrase,
++                RBD_ENCRYPTION_ALG[cipher_spec['cipher_alg']]
++            )
++
+     def _clone(self,
+                volume: Volume,
+                src_pool: str,
+@@ -1871,6 +1892,37 @@ class RBDDriver(driver.CloneableImageVD, driver.MigrateVD,
+                     image_location: Optional[list],
+                     image_meta: dict,
+                     image_service) -> tuple[dict, bool]:
++        return self._clone_image(context, volume, image_location,
++                                 image_meta, image_service)
++
++    def clone_image_and_encrypt(
++        self,
++        context: context.RequestContext,
++        volume: Volume,
++        image_location: Optional[list],
++        image_meta: dict,
++        image_service
++    ) -> tuple[dict, bool]:
++
++        # Note(ricolin): method `encryption_format` added after Ceph Pacific
++        # release (>=16.1.0).
++        if self.rbd and hasattr(
++            self.rbd.Image, 'encryption_format') and callable(
++                self.rbd.Image.encryption_format):
++            return self._clone_image(
++                context, volume, image_location,
++                image_meta, image_service, is_encrypt=True)
++        else:
++            return {}, False
++
++    def _clone_image(self,
++                     context: context.RequestContext,
++                     volume: Volume,
++                     image_location: Optional[list],
++                     image_meta: dict,
++                     image_service,
++                     is_encrypt: Optional[bool] = False
++                     ) -> tuple[dict, bool]:
+         if image_location:
+             # Note: image_location[0] is glance image direct_url.
+             # image_location[1] contains the list of all locations (including
+@@ -1888,12 +1940,41 @@ class RBDDriver(driver.CloneableImageVD, driver.MigrateVD,
+                         url_location, image_meta):
+                     _prefix, pool, image, snapshot = \
+                         self._parse_location(url_location)
++                    if is_encrypt:
++                        passphrase, cipher_spec = self._fetch_encryption_info(
++                            context, volume)
++                        if cipher_spec['cipher_alg'] not in RBD_ENCRYPTION_ALG:
++                            LOG.debug(
++                                "Skip clone. Cipher spec: %s not supported "
++                                "for encrypt volume directly from RBD.",
++                                cipher_spec)
++                            return ({}, False)
+                     volume_update = self._clone(volume, pool, image, snapshot)
++                    if is_encrypt:
++                        self._flatten(self.configuration.rbd_pool, volume.name)
++                        self._encrypt_volume(
++                            context, volume, passphrase, cipher_spec)
+                     volume_update['provider_location'] = None
+                     self._resize(volume)
+                     return volume_update, True
+         return ({}, False)
+ 
++    def _fetch_encryption_info(self,
++                               context: context.RequestContext,
++                               volume: Volume) -> tuple[str, dict]:
++        encryption = volume_utils.check_encryption_provider(
++            volume,
++            context)
++        # Fetch the key associated with the volume and decode the passphrase
++        keymgr = key_manager.API(CONF)
++        key = keymgr.get(context, encryption['encryption_key_id'])
++        passphrase = binascii.hexlify(key.get_encoded()).decode('utf-8')
++
++        # Decode the dm-crypt style cipher spec into something qemu-img can use
++        cipher_spec = image_utils.decode_cipher(encryption['cipher'],
++                                                encryption['key_size'])
++        return passphrase, cipher_spec
++
+     def copy_image_to_encrypted_volume(self,
+                                        context: context.RequestContext,
+                                        volume: Volume,
+@@ -1914,18 +1995,8 @@ class RBDDriver(driver.CloneableImageVD, driver.MigrateVD,
+                        volume: Volume,
+                        tmp_dir: str,
+                        src_image_path: Any) -> None:
+-        encryption = volume_utils.check_encryption_provider(
+-            volume,
+-            context)
+-
+-        # Fetch the key associated with the volume and decode the passphrase
+-        keymgr = key_manager.API(CONF)
+-        key = keymgr.get(context, encryption['encryption_key_id'])
+-        passphrase = binascii.hexlify(key.get_encoded()).decode('utf-8')
+-
+-        # Decode the dm-crypt style cipher spec into something qemu-img can use
+-        cipher_spec = image_utils.decode_cipher(encryption['cipher'],
+-                                                encryption['key_size'])
++        passphrase, cipher_spec = self._fetch_encryption_info(
++            context, volume)
+ 
+         tmp_dir = volume_utils.image_conversion_dir()
+ 
+diff --git a/cinder/volume/flows/manager/create_volume.py b/cinder/volume/flows/manager/create_volume.py
+index 9d0590d9c..10eaa4b92 100644
+--- a/cinder/volume/flows/manager/create_volume.py
++++ b/cinder/volume/flows/manager/create_volume.py
+@@ -1086,11 +1086,6 @@ class CreateVolumeFromSpecTask(flow_utils.CinderTask):
+         # NOTE (singn): two params need to be returned
+         # dict containing provider_location for cloned volume
+         # and clone status.
+-        # NOTE (lixiaoy1): Currently all images are raw data, we can't
+-        # use clone_image to copy data if new volume is encrypted
+-        # NOTE (ricolin):  If the image provided an encryption key, we have
+-        # already cloned it to the volume's key in
+-        # _get_encryption_key_id, so we can do a direct clone.
+         image_encryption_key = image_meta.get('cinder_encryption_key_id')
+         volume_is_encrypted = volume.encryption_key_id is not None
+         cloned = False
+@@ -1102,6 +1097,9 @@ class CreateVolumeFromSpecTask(flow_utils.CinderTask):
+                                                            image_location,
+                                                            image_meta,
+                                                            image_service)
++        else:
++            model_update, cloned = self.driver.clone_image_and_encrypt(
++                context, volume, image_location, image_meta, image_service)
+ 
+         # Try and clone the image if we have it set as a glance location.
+         if not cloned and 'cinder' in CONF.allowed_direct_url_schemes:
+diff --git a/releasenotes/notes/allow-clone-image-for-encrypted-volume-de477647e9016b8b.yaml b/releasenotes/notes/allow-clone-image-for-encrypted-volume-de477647e9016b8b.yaml
+new file mode 100644
+index 000000000..63d1f38cd
+--- /dev/null
++++ b/releasenotes/notes/allow-clone-image-for-encrypted-volume-de477647e9016b8b.yaml
+@@ -0,0 +1,21 @@
++---
++features:
++  - |
++    Allow clone image when creating encrypted volume from Glance image if both
++    stored in RBD.
++    Previously, Glance image clone is not supported for encrypted volume
++    creation. The old process is to download image to local disk, encrypt the
++    local file, and import it back to RBD. This not just slow, but also
++    protentially take large amount of local disk space from hosts that runs
++    Cinder volume service.
++    The new process is to try and clone from Glance image (if it's also stored
++    in RBD), flatten it, and encrypting new image in RBD for volume. And If
++    Glance image source is not clonable, will continue with copy-and-import
++    method as previous flow.
++    In above flow, If clone from Glance image is appliable. Even it still
++    requires to clone and flatten RBD image might took some time, but should
++    still be a lot faster than copy-and-import. And also no local disk will
++    be used to store raw image in this case.
++    This also introduced driver method `clone_image_and_encrypt` for drivers
++    that seperate the clone process from non-encrypted volume so the create
++    flow won't be affected.
+-- 
+2.25.1
+