diff --git a/api-ref/source/servers-actions.inc b/api-ref/source/servers-actions.inc
index 7a33990dbe2..29b3d24793d 100644
--- a/api-ref/source/servers-actions.inc
+++ b/api-ref/source/servers-actions.inc
@@ -841,7 +841,8 @@ automatically confirms the resize operation after the configured interval.
 .. _resize_confirm_window: https://docs.openstack.org/nova/latest/configuration/config.html#DEFAULT.resize_confirm_window
 
 .. note:: There is a `known limitation <https://bugs.launchpad.net/nova/+bug/1558880>`__
-          that ephemeral disks are not resized.
+          that ephemeral disks are not resized if they are smaller than the
+          flavors ephemeral size or if there are multiple ephemeral disks.
 
 Normal response codes: 202
 
diff --git a/nova/compute/manager.py b/nova/compute/manager.py
index fbe45df38fe..dffc8d1853b 100644
--- a/nova/compute/manager.py
+++ b/nova/compute/manager.py
@@ -4942,6 +4942,7 @@ def _delete_stashed_flavor_info(self, instance):
         instance.old_flavor = None
         instance.new_flavor = None
         instance.system_metadata.pop('old_vm_state', None)
+        instance.system_metadata.pop('old_disk_sizes', None)
         instance.save()
 
     @wrap_exception()
@@ -5486,6 +5487,13 @@ def _finish_revert_resize(
             # compatibility if old_vm_state is not set
             old_vm_state = instance.system_metadata.get(
                 'old_vm_state', vm_states.ACTIVE)
+            old_sizes = instance.system_metadata.get('old_disk_sizes')
+            if old_sizes:
+                self._revert_resize_ephemeral_bdms(context, instance,
+                                                   jsonutils.loads(old_sizes))
+                # Refresh bdms
+                bdms = objects.BlockDeviceMappingList.get_by_instance_uuid(
+                    context, instance.uuid)
 
             # Revert the flavor and host/node fields to their previous values
             self._revert_instance_flavor_host_node(instance, migration)
@@ -6271,6 +6279,176 @@ def _complete_volume_attachments(self, context, bdms):
                     self.volume_api.attachment_complete(
                         context, bdm.attachment_id)
 
+    def _update_ephemeral_bdms(self, context, instance, block_device_info,
+                               ephemeral_size, swap_size):
+        """Update ephemeral bdm objects in DB (for instance resizing).
+
+        Updates volume size of block device mapping object according to
+        specified new size. Works only for a single ephemeral and swap disks.
+        Stores the result into the DB.
+
+        Creates a new bdm if necessary.
+
+        :param context: Request context object.
+        :param instance: An instance which bdms are to be updated.
+        :param block_device_info: Instance block device mapping in driver
+                                  format.
+        :param ephemeral_size: New size in GB of the single ephemeral disk.
+                               Creates a new bdm if no bdm exists.
+                               If None - does nothing with ephemerals.
+                               If 0 - removes the ephemeral disk
+        :param swap_size: A size in MB of swap disk. Creates a new bdm if no
+                          swap bdm exists and non zero size is specified.
+                          Destroys swap bdm if zero size is specified.
+                          If None - does nothing with swap.
+        """
+        ephemerals = driver.block_device_info_get_ephemerals(block_device_info)
+        swap = driver.block_device_info_get_swap(block_device_info)
+        reassing_device_names = False
+
+        # Resize ephemeral disk
+        if ephemeral_size is not None:
+            if not ephemerals and ephemeral_size > 0:
+                LOG.debug(
+                    f'Creating ephemeral bdm: size={ephemeral_size}.',
+                    instance=instance
+                )
+                bdm = objects.BlockDeviceMapping(
+                    context=context, instance_uuid=instance.uuid,
+                    source_type='blank', destination_type='local',
+                    device_type='disk', delete_on_termination=True,
+                    boot_index=-1, volume_size=ephemeral_size)
+                bdm.create()
+                reassing_device_names = True
+            elif len(ephemerals) == 1 and ephemeral_size == 0:
+                LOG.debug('Deleting ephemeral bdm.', instance=instance)
+                ephemerals[0].destroy()
+                reassing_device_names = True
+            elif len(ephemerals) == 1:
+                LOG.debug(
+                    f'Resizing ephemeral bdm: new_size={ephemeral_size}, '
+                    f'old_size={ephemerals[0].volume_size}.',
+                    instance=instance,
+                )
+                ephemerals[0].volume_size = ephemeral_size
+                ephemerals[0].save()
+
+        # Resize swap
+        if swap_size is not None:
+            if swap['swap_size'] == 0 and swap_size > 0:
+                LOG.debug(
+                    f'Creating swap bdm: size={swap_size}.',
+                    instance=instance,
+                )
+                bdm = objects.BlockDeviceMapping(
+                    context=context, instance_uuid=instance.uuid,
+                    source_type='blank', destination_type='local',
+                    device_type='disk', guest_format='swap',
+                    boot_index=-1, volume_size=swap_size)
+                bdm.create()
+                reassing_device_names = True
+            elif swap['swap_size'] > 0 and swap_size == 0:
+                LOG.debug('Deleting swap bdm.', instance=instance)
+                swap.destroy()
+                reassing_device_names = True
+            elif swap['swap_size'] != swap_size:
+                LOG.debug(
+                    f'Resizing swap bdm: old_size={swap.volume_size} '
+                    f'new_size={swap_size}',
+                    instance=instance,
+                )
+                swap.volume_size = swap_size
+                swap.save()
+
+        if reassing_device_names:
+            bdms = objects.BlockDeviceMappingList.get_by_instance_uuid(
+                    context, instance.uuid)
+            self._default_block_device_names(instance, None, bdms)
+
+    def _resize_ephemeral_bdms(self, context, instance, bdms):
+        """Resize ephemeral bdm objects in DB (for instance resizing).
+
+        For an instance in resizing state (having actual old and new flavor
+        fields) this method resizes the instance's ephemeral and swap bdms
+        objects, persists them into DB, preparing them to be used by virt
+        driver to create/resize disks.
+
+        Supported features:
+        - add/remove/resize swap disk;
+        - add/increase single ephemeral disk;
+        - keep custom user specified ephemeral/swap size if it differs from
+            flavor specified size.
+
+        :param context: Request context object.
+        :param instance: An instance which bdms are to be resized.
+        :returns: A serializable object contains original bdm sizes.
+        """
+
+        block_device_info = self._get_instance_block_device_info(
+            context, instance, bdms=bdms)
+
+        # Handle ephemeral disk resizing
+        old_ephemeral_flavor = instance.old_flavor.ephemeral_gb
+        new_ephemeral_flavor = instance.new_flavor.ephemeral_gb
+        ephemerals = driver.block_device_info_get_ephemerals(block_device_info)
+        old_ephemeral_size = new_ephemeral_size = None
+
+        changed = old_ephemeral_flavor != new_ephemeral_flavor
+        single_ephemeral = len(ephemerals) == 1
+
+        # Current config uses the maximum ephemeral disk size of the old flavor
+        max_size = (
+            (len(ephemerals) == 0 and old_ephemeral_flavor == 0) or
+            (single_ephemeral and
+             old_ephemeral_flavor == ephemerals[0]['size'])
+        )
+
+        if changed and max_size:
+            old_ephemeral_size = (ephemerals[0]['size'] if len(ephemerals) > 0
+                                  else 0)
+            new_ephemeral_size = new_ephemeral_flavor
+
+        # Handle Swap resizing
+        swap = driver.block_device_info_get_swap(block_device_info)
+        old_swap_size = new_swap_size = None
+
+        if (instance.new_flavor.swap != instance.old_flavor.swap and
+            (swap['swap_size'] == instance.old_flavor.swap or
+             swap['swap_size'] > instance.new_flavor.swap)):
+            old_swap_size = swap['swap_size']
+            new_swap_size = instance.new_flavor.swap
+
+        if new_ephemeral_size is not None or new_swap_size is not None:
+            self._update_ephemeral_bdms(context, instance, block_device_info,
+                                        new_ephemeral_size, new_swap_size)
+
+        return old_ephemeral_size, old_swap_size
+
+    def _revert_resize_ephemeral_bdms(self, context, instance, old_disk_sizes):
+        """Revert resized ephemeral bdm objects in DB (for instance resizing).
+
+        Restore ephemeral and swap bdms previously resized with
+        _resize_ephemeral_bdms.
+
+        :param context: Request context object.
+        :param instance: An instance which bdms are to be reverted.
+        :param old_disk_sizes: An object created with _resize_ephemeral_bdms.
+        """
+        old_ephemeral_size, old_swap_size = old_disk_sizes
+        if old_ephemeral_size is None and old_swap_size is None:
+            return
+
+        LOG.debug(
+            f'Reverting ephemeral bdms: ephemeral_size={old_ephemeral_size}, '
+            f'swap_size={old_swap_size}.',
+            instance=instance,
+        )
+        bdms = objects.BlockDeviceMappingList.get_by_instance_uuid(
+                context, instance.uuid)
+        block_device_info = driver.get_block_device_info(instance, bdms)
+        self._update_ephemeral_bdms(context, instance, block_device_info,
+                                    old_ephemeral_size, old_swap_size)
+
     def _finish_resize(self, context, instance, migration, disk_info,
                        image_meta, bdms, request_spec):
         resize_instance = False  # indicates disks have been resized
@@ -6297,6 +6475,11 @@ def _finish_resize(self, context, instance, migration, disk_info,
                 if old_flavor[key] != new_flavor[key]:
                     resize_instance = True
                     break
+            if resize_instance:
+                old_sizes = self._resize_ephemeral_bdms(
+                    context, instance, bdms)
+                instance.system_metadata['old_disk_sizes'] = jsonutils.dumps(
+                    old_sizes)
         instance.apply_migration_context()
 
         # NOTE(tr3buchet): setup networks on destination host
diff --git a/nova/conf/libvirt.py b/nova/conf/libvirt.py
index abcefee777b..e3cd8ec31c4 100644
--- a/nova/conf/libvirt.py
+++ b/nova/conf/libvirt.py
@@ -959,6 +959,13 @@
 * images_volume_group
 * [workarounds]/ensure_libvirt_rbd_instance_dir_cleanup
 * compute.force_raw_images
+"""),
+    cfg.BoolOpt('remote_copy_compression',
+                default=False,
+                help="""
+Use SSH compression to copy ephemeral disks to a remote host. On fast
+local networks compression often reduces the copy bandwidth because the CPU
+cannot compress fast enough.
 """),
     cfg.StrOpt('images_volume_group',
                help="""
diff --git a/nova/test.py b/nova/test.py
index ee848cdab1a..894707dbd4b 100644
--- a/nova/test.py
+++ b/nova/test.py
@@ -89,13 +89,6 @@ class TestingException(Exception):
     pass
 
 
-# NOTE(claudiub): this needs to be called before any mock.patch calls are
-# being done, and especially before any other test classes load. This fixes
-# the mock.patch autospec issue:
-# https://github.com/testing-cabal/mock/issues/396
-mock_fixture.patch_mock_module()
-
-
 def _poison_unfair_compute_resource_semaphore_locking():
     """Ensure that every locking on COMPUTE_RESOURCE_SEMAPHORE is called with
     fair=True.
diff --git a/nova/tests/unit/compute/test_compute.py b/nova/tests/unit/compute/test_compute.py
index 6cddd2ca858..3e4c841622e 100644
--- a/nova/tests/unit/compute/test_compute.py
+++ b/nova/tests/unit/compute/test_compute.py
@@ -4835,10 +4835,16 @@ def _test_finish_resize(self, power_on, resize_instance=True):
         fake_bdms = objects.BlockDeviceMappingList(objects=[
             objects.BlockDeviceMapping(destination_type='volume',
                                        attachment_id=uuids.attachment_id,
-                                       device_name='/dev/vdb'),
+                                       device_name='/dev/vdb',
+                                       no_device=False,
+                                       source_type=None),
             objects.BlockDeviceMapping(destination_type='volume',
-                                       attachment_id=None),
-            objects.BlockDeviceMapping(destination_type='local')
+                                       attachment_id=None,
+                                       no_device=False,
+                                       source_type=None),
+            objects.BlockDeviceMapping(destination_type='local',
+                                       no_device=False,
+                                       source_type=None)
         ])
 
         with test.nested(
@@ -4857,11 +4863,13 @@ def _test_finish_resize(self, power_on, resize_instance=True):
             mock.patch.object(instance, 'save'),
             mock.patch.object(self.compute.driver, 'get_volume_connector'),
             mock.patch.object(self.compute.volume_api, 'attachment_update'),
-            mock.patch.object(self.compute.volume_api, 'attachment_complete')
+            mock.patch.object(self.compute.volume_api, 'attachment_complete'),
+            mock.patch('nova.virt.driver.block_device_info_get_ephemerals'),
+            mock.patch('nova.virt.driver.block_device_info_get_swap')
         ) as (mock_get_bdm, mock_setup, mock_net_mig, mock_get_nw, mock_notify,
               mock_notify_action, mock_virt_mig, mock_get_blk, mock_mig_save,
               mock_inst_save, mock_get_vol_connector, mock_attachment_update,
-              mock_attachment_complete):
+              mock_attachment_complete, mock_get_ephemerals, mock_get_swap):
             def _mig_save():
                 self.assertEqual(migration.status, 'finished')
                 self.assertEqual(vm_state, instance.vm_state)
@@ -4933,9 +4941,9 @@ def _instance_save1(expected_task_state=None):
                 instance, disk_info, 'fake-nwinfo1',
                 test.MatchType(objects.ImageMeta), resize_instance, mock.ANY,
                 'fake-bdminfo', power_on)
-            mock_get_blk.assert_called_once_with(self.context, instance,
-                                                 refresh_conn_info=True,
-                                                 bdms=fake_bdms)
+            mock_get_blk.assert_called_with(self.context, instance,
+                                            refresh_conn_info=True,
+                                            bdms=fake_bdms)
             mock_inst_save.assert_has_calls(inst_call_list)
             mock_mig_save.assert_called_once_with()
 
diff --git a/nova/tests/unit/compute/test_compute_mgr.py b/nova/tests/unit/compute/test_compute_mgr.py
index d26844c180d..c44fa49a709 100644
--- a/nova/tests/unit/compute/test_compute_mgr.py
+++ b/nova/tests/unit/compute/test_compute_mgr.py
@@ -13485,3 +13485,577 @@ def test_delete_swap_bdm(self, mock_bdm_destroy, mock_create_bdm):
 
         mock_bdm_destroy.assert_called_once()
         self.instance.get_bdms.assert_called_once()
+
+
+@mock.patch('nova.objects.BlockDeviceMappingList.get_by_instance_uuid')
+@mock.patch.object(objects.BlockDeviceMapping, 'create', autospec=True)
+@mock.patch.object(objects.BlockDeviceMapping, 'destroy', autospec=True)
+@mock.patch.object(objects.BlockDeviceMapping, 'save', autospec=True)
+class ComputeManagerBDMResizeTestCase(test.NoDBTestCase):
+
+    def setUp(self):
+        super().setUp()
+        self.flags(compute_driver='fake.SameHostColdMigrateDriver')
+        self.compute = manager.ComputeManager()
+        self.context = context.RequestContext(
+            fakes.FAKE_USER_ID,
+            fakes.FAKE_PROJECT_ID,
+        )
+
+    def _create_migration_instance(
+            self,
+            old_flavor_ephemeral=0,
+            old_flavor_swap=0,
+            new_flavor_ephemeral=0,
+            new_flavor_swap=0,
+            instance_ephemerals=None,
+            instance_swap=None,
+    ):
+        old_flavor = fake_flavor.fake_flavor_obj(
+            self.context,
+            name='old-flavor',
+            id=1,
+            flavorid='1',
+            ephemeral_gb=old_flavor_ephemeral,
+            swap=old_flavor_swap,
+        )
+
+        new_flavor = fake_flavor.fake_flavor_obj(
+            self.context,
+            name='new-flavor',
+            id=2,
+            flavorid='2',
+            ephemeral_gb=new_flavor_ephemeral,
+            swap=new_flavor_swap,
+        )
+
+        instance = fake_instance.fake_instance_obj(
+            self.context,
+            flavor=old_flavor,
+            vm_state=vm_states.ACTIVE,
+            expected_attrs=['metadata', 'system_metadata', 'info_cache'],
+        )
+
+        # This would be set in _finish_resize
+        instance.old_flavor = instance.flavor
+        # This would be set in _prep_resize
+        instance.new_flavor = new_flavor
+
+        # Root disk
+        bdm_objects = [
+            fake_block_device.fake_bdm_object(
+                self.context,
+                dict(source_type='image',
+                     destination_type='local',
+                     boot_index=0,
+                     image_id=uuids.image_id,
+                     device_name='/dev/vda',
+                     volume_size=1))
+        ]
+
+        # Ephemeral disks
+        if instance_ephemerals:
+            for i, size in enumerate(instance_ephemerals):
+                bdm_objects.append(
+                    fake_block_device.fake_bdm_object(
+                        self.context,
+                        dict(source_type='blank',
+                             destination_type='local',
+                             device_type='disk',
+                             boot_index=-1,
+                             # Ephemeral disks start at /dev/vdb
+                             device_name=f'/dev/vd{chr(ord("b") + i)}',
+                             volume_size=size)
+                    )
+                )
+
+        # Swap disks
+        if instance_swap:
+            bdm_objects.append(
+                fake_block_device.fake_bdm_object(
+                    self.context,
+                    dict(source_type='blank',
+                         destination_type='local',
+                         device_type='disk',
+                         guest_format='swap',
+                         boot_index=-1,
+                         volume_size=instance_swap),
+                )
+            )
+
+        bdm_list = objects.BlockDeviceMappingList(objects=bdm_objects)
+
+        return instance, bdm_list
+
+    def _setup_mocks(self, mock_bdm_get, mock_create, mock_destroy, bdm_list):
+
+        def create(_self):
+
+            # Set UUID and ID
+            _self.uuid = uuids.bdm
+            _self.id = 1
+
+            # Fill in any unset fields with default values
+            for field in _self.fields:
+                if field not in _self._changed_fields:
+                    if _self.fields[field].nullable:
+                        value = None
+                    else:
+                        value = _self.fields[field].default
+
+                    setattr(_self, field, value)
+
+            bdm_list.objects.append(_self)
+
+        def destroy(_self):
+            bdm_list.objects.remove(_self)
+
+        mock_create.side_effect = create
+        mock_destroy.side_effect = destroy
+        mock_bdm_get.return_value = bdm_list
+
+    def test_resize_ephemeral_bdms_single_full(
+            self,
+            mock_save,
+            mock_destroy,
+            mock_create,
+            mock_bdm_get
+    ):
+        """Test resizing a single ephemeral disk using the full allocation of
+        the flavor. The disk should be resized to the ephemeral size of the
+        new flavor.
+        """
+
+        instance, bdms = self._create_migration_instance(
+            old_flavor_ephemeral=1,
+            new_flavor_ephemeral=2,
+            instance_ephemerals=[1],
+        )
+
+        self._setup_mocks(mock_bdm_get, mock_create, mock_destroy, bdms)
+
+        old_sizes = self.compute._resize_ephemeral_bdms(
+            self.context,
+            instance,
+            bdms,
+        )
+
+        self.assertEqual(bdms[1].volume_size, 2)
+        self.assertEqual(old_sizes, (1, None))
+        mock_save.assert_called_once()
+
+        # Test reverting
+        mock_save.reset_mock()
+        mock_create.reset_mock()
+        mock_destroy.reset_mock()
+
+        self.compute._revert_resize_ephemeral_bdms(
+            self.context,
+            instance,
+            old_sizes,
+        )
+
+        mock_save.assert_called_once()
+        mock_create.assert_not_called()
+        mock_destroy.assert_not_called()
+        self.assertEqual(bdms[1].volume_size, 1)
+
+    def test_resize_ephemeral_bdms_single_ephemeral_smaller(
+            self,
+            mock_save,
+            mock_destroy,
+            mock_create,
+            mock_bdm_get
+    ):
+        """Test resizing a single ephemeral disk not using the full allocation
+        of the flavor. The disk should not be resized.
+        """
+        instance, bdms = self._create_migration_instance(
+            old_flavor_ephemeral=2,
+            new_flavor_ephemeral=3,
+            instance_ephemerals=[1],
+        )
+
+        self._setup_mocks(mock_bdm_get, mock_create, mock_destroy, bdms)
+
+        old_sizes = self.compute._resize_ephemeral_bdms(
+            self.context,
+            instance,
+            bdms,
+        )
+
+        self.assertEqual(bdms[1].volume_size, 1)
+        self.assertEqual(old_sizes, (None, None))
+        mock_save.assert_not_called()
+
+        # Test reverting
+        mock_save.reset_mock()
+        mock_create.reset_mock()
+        mock_destroy.reset_mock()
+
+        self.compute._revert_resize_ephemeral_bdms(
+            self.context,
+            instance,
+            old_sizes,
+        )
+
+        mock_save.assert_not_called()
+        mock_create.assert_not_called()
+        mock_destroy.assert_not_called()
+        self.assertEqual(bdms[1].volume_size, 1)
+
+    def test_resize_ephemeral_bdms_multiple_ephemerals(
+            self,
+            mock_save,
+            mock_destroy,
+            mock_create,
+            mock_bdm_get
+    ):
+        """Test resizing with multiple ephemeral disks. This should not change
+        the ephemeral disk sizes.
+        """
+        instance, bdms = self._create_migration_instance(
+            old_flavor_ephemeral=3,
+            new_flavor_ephemeral=4,
+            instance_ephemerals=[2, 1],
+        )
+
+        self._setup_mocks(mock_bdm_get, mock_create, mock_destroy, bdms)
+
+        old_sizes = self.compute._resize_ephemeral_bdms(
+            self.context,
+            instance,
+            bdms,
+        )
+
+        self.assertEqual(bdms[1].volume_size, 2)
+        self.assertEqual(bdms[2].volume_size, 1)
+        self.assertEqual(old_sizes, (None, None))
+        mock_save.assert_not_called()
+
+        # Test reverting
+        mock_save.reset_mock()
+        mock_create.reset_mock()
+        mock_destroy.reset_mock()
+
+        self.compute._revert_resize_ephemeral_bdms(
+            self.context,
+            instance,
+            old_sizes,
+        )
+
+        mock_save.assert_not_called()
+        mock_create.assert_not_called()
+        mock_destroy.assert_not_called()
+        self.assertEqual(bdms[1].volume_size, 2)
+        self.assertEqual(bdms[2].volume_size, 1)
+
+    def test_resize_ephemeral_bdms_to_single_ephemeral(
+            self,
+            mock_save,
+            mock_destroy,
+            mock_create,
+            mock_bdm_get
+    ):
+        """Test resizing an instance without an ephemeral disk to a flavor
+        with ephemeral storage. This should add an ephemeral disk.
+        """
+        instance, bdms = self._create_migration_instance(
+            old_flavor_ephemeral=0,
+            new_flavor_ephemeral=1,
+            instance_ephemerals=[],
+        )
+
+        self._setup_mocks(mock_bdm_get, mock_create, mock_destroy, bdms)
+
+        old_sizes = self.compute._resize_ephemeral_bdms(
+            self.context,
+            instance,
+            bdms,
+        )
+
+        mock_create.assert_called_once()
+        mock_save.assert_called_once()  # after assigning device name
+        mock_destroy.assert_not_called()
+        self.assertEqual(old_sizes, (0, None))
+        self.assertEqual(len(bdms), 2)
+        self.assertEqual(bdms[1].volume_size, 1)
+
+        # Test reverting
+        mock_save.reset_mock()
+        mock_create.reset_mock()
+        mock_destroy.reset_mock()
+
+        self.compute._revert_resize_ephemeral_bdms(
+            self.context,
+            instance,
+            old_sizes,
+        )
+
+        mock_save.assert_not_called()
+        mock_create.assert_not_called()
+        mock_destroy.assert_called_once()
+        self.assertEqual(len(bdms), 1)
+
+    def test_resize_ephemeral_bdms_swap_full_larger(
+            self,
+            mock_save,
+            mock_destroy,
+            mock_create,
+            mock_bdm_get
+    ):
+        """Test resizing to a flavor with more swap space if all swap space
+        allowed by the current flavor is used. Swap will be resized.
+        """
+        instance, bdms = self._create_migration_instance(
+            old_flavor_swap=1,
+            new_flavor_swap=2,
+            instance_swap=1,
+        )
+
+        self._setup_mocks(mock_bdm_get, mock_create, mock_destroy, bdms)
+
+        old_sizes = self.compute._resize_ephemeral_bdms(
+            self.context,
+            instance,
+            bdms,
+        )
+
+        self.assertEqual(bdms[1].volume_size, 2)
+        self.assertEqual(old_sizes, (None, 1))
+        mock_save.assert_called_once()
+
+        # Test reverting
+        mock_save.reset_mock()
+        mock_create.reset_mock()
+        mock_destroy.reset_mock()
+
+        self.compute._revert_resize_ephemeral_bdms(
+            self.context,
+            instance,
+            old_sizes,
+        )
+
+        mock_save.assert_called_once()
+        mock_create.assert_not_called()
+        mock_destroy.assert_not_called()
+        self.assertEqual(bdms[1].volume_size, 1)
+
+    def test_resize_ephemeral_bdms_swap_full_smaller(
+            self,
+            mock_save,
+            mock_destroy,
+            mock_create,
+            mock_bdm_get
+    ):
+        """Test resizing to a flavor with less swap space if all swap space
+        allowed by the current flavor is used. Swap will always be resized.
+        """
+        instance, bdms = self._create_migration_instance(
+            old_flavor_swap=2,
+            new_flavor_swap=1,
+            instance_swap=2,
+        )
+
+        self._setup_mocks(mock_bdm_get, mock_create, mock_destroy, bdms)
+
+        old_sizes = self.compute._resize_ephemeral_bdms(
+            self.context,
+            instance,
+            bdms,
+        )
+
+        self.assertEqual(bdms[1].volume_size, 1)
+        self.assertEqual(old_sizes, (None, 2))
+        mock_save.assert_called_once()
+
+        # Test reverting
+        mock_save.reset_mock()
+        mock_create.reset_mock()
+        mock_destroy.reset_mock()
+
+        self.compute._revert_resize_ephemeral_bdms(
+            self.context,
+            instance,
+            old_sizes,
+        )
+
+        mock_save.assert_called_once()
+        mock_create.assert_not_called()
+        mock_destroy.assert_not_called()
+        self.assertEqual(bdms[1].volume_size, 2)
+
+    def test_resize_ephemeral_bdms_swap_unused(
+            self,
+            mock_save,
+            mock_destroy,
+            mock_create,
+            mock_bdm_get
+    ):
+        """Test resizing to a flavor with more swap space if not all swap
+        space allowed by the old flavor is used. Swap will not be resized.
+        """
+        instance, bdms = self._create_migration_instance(
+            old_flavor_swap=2,
+            new_flavor_swap=3,
+            instance_swap=1,
+        )
+
+        self._setup_mocks(mock_bdm_get, mock_create, mock_destroy, bdms)
+
+        old_sizes = self.compute._resize_ephemeral_bdms(
+            self.context,
+            instance,
+            bdms,
+        )
+
+        self.assertEqual(bdms[1].volume_size, 1)
+        self.assertEqual(old_sizes, (None, None))
+        mock_save.assert_not_called()
+
+        # Test reverting
+        mock_save.reset_mock()
+        mock_create.reset_mock()
+        mock_destroy.reset_mock()
+
+        self.compute._revert_resize_ephemeral_bdms(
+            self.context,
+            instance,
+            old_sizes,
+        )
+
+        mock_save.assert_not_called()
+        mock_create.assert_not_called()
+        mock_destroy.assert_not_called()
+        self.assertEqual(bdms[1].volume_size, 1)
+
+    def test_resize_ephemeral_bdms_swap_too_large(
+            self,
+            mock_save,
+            mock_destroy,
+            mock_create,
+            mock_bdm_get
+    ):
+        """Test resizing to a flavor with less swap space than allowed by the
+        new flavor for an instance that does not use all swap space allowed
+        by the old flavor. Swap will be sized down to the new flavor.
+        """
+        instance, bdms = self._create_migration_instance(
+            old_flavor_swap=3,
+            new_flavor_swap=1,
+            instance_swap=2,
+        )
+
+        self._setup_mocks(mock_bdm_get, mock_create, mock_destroy, bdms)
+
+        old_sizes = self.compute._resize_ephemeral_bdms(
+            self.context,
+            instance,
+            bdms,
+        )
+
+        self.assertEqual(bdms[1].volume_size, 1)
+        self.assertEqual(old_sizes, (None, 2))
+        mock_save.assert_called_once()
+
+        # Test reverting
+        mock_save.reset_mock()
+        mock_create.reset_mock()
+        mock_destroy.reset_mock()
+
+        self.compute._revert_resize_ephemeral_bdms(
+            self.context,
+            instance,
+            old_sizes,
+        )
+
+        mock_save.assert_called_once()
+        mock_create.assert_not_called()
+        mock_destroy.assert_not_called()
+        self.assertEqual(bdms[1].volume_size, 2)
+
+    def test_resize_ephemeral_bdms_to_noswap(
+            self,
+            mock_save,
+            mock_destroy,
+            mock_create,
+            mock_bdm_get
+    ):
+        """Test resizing from a flavor with swap space to one without swap."""
+        instance, bdms = self._create_migration_instance(
+            old_flavor_swap=1,
+            new_flavor_swap=0,
+            instance_swap=1,
+        )
+
+        self._setup_mocks(mock_bdm_get, mock_create, mock_destroy, bdms)
+
+        old_sizes = self.compute._resize_ephemeral_bdms(
+            self.context,
+            instance,
+            bdms,
+        )
+
+        mock_destroy.assert_called_once()
+        mock_save.assert_not_called()
+        self.assertEqual(old_sizes, (None, 1))
+        self.assertEqual(len(bdms), 1)
+
+        # Test reverting
+        mock_save.reset_mock()
+        mock_create.reset_mock()
+        mock_destroy.reset_mock()
+
+        self.compute._revert_resize_ephemeral_bdms(
+            self.context,
+            instance,
+            old_sizes,
+        )
+
+        mock_save.assert_called_once()  # after assigning device name
+        mock_create.assert_called_once()
+        mock_destroy.assert_not_called()
+        self.assertEqual(bdms[1].volume_size, 1)
+
+    def test_resize_ephemeral_bdms_to_swap(
+            self,
+            mock_save,
+            mock_destroy,
+            mock_create,
+            mock_bdm_get
+    ):
+        """Test resizing from a flavor without swap space to one with swap."""
+        instance, bdms = self._create_migration_instance(
+            old_flavor_swap=0,
+            new_flavor_swap=1
+        )
+
+        self._setup_mocks(mock_bdm_get, mock_create, mock_destroy, bdms)
+
+        old_sizes = self.compute._resize_ephemeral_bdms(
+            self.context,
+            instance,
+            bdms,
+        )
+
+        mock_create.assert_called_once()
+        mock_save.assert_called_once()  # after assigning device name
+        self.assertEqual(old_sizes, (None, 0))
+        self.assertEqual(len(bdms), 2)
+        self.assertEqual(bdms[1].volume_size, 1)
+
+        # Test reverting
+        mock_save.reset_mock()
+        mock_create.reset_mock()
+        mock_destroy.reset_mock()
+
+        self.compute._revert_resize_ephemeral_bdms(
+            self.context,
+            instance,
+            old_sizes,
+        )
+
+        mock_save.assert_not_called()
+        mock_create.assert_not_called()
+        mock_destroy.assert_called_once()
+        self.assertEqual(len(bdms), 1)
diff --git a/nova/tests/unit/virt/libvirt/test_driver.py b/nova/tests/unit/virt/libvirt/test_driver.py
index cf824df3e98..56e26b39529 100644
--- a/nova/tests/unit/virt/libvirt/test_driver.py
+++ b/nova/tests/unit/virt/libvirt/test_driver.py
@@ -23098,13 +23098,14 @@ def _test_migrate_disk_and_power_off(
             self, ctxt, flavor_obj, mock_execute, mock_exists, mock_rename,
             mock_is_shared, mock_get_host_ip, mock_destroy,
             mock_get_disk_info, mock_vtpm, mock_unplug_vifs,
-            mock_cleanup, block_device_info=None, params_for_instance=None):
+            mock_cleanup, block_device_info=None, params_for_instance=None,
+            images_type='qcow2', assert_compression=False):
         """Test for nova.virt.libvirt.driver.LivirtConnection
         .migrate_disk_and_power_off.
         """
 
         instance = self._create_instance(params=params_for_instance)
-        disk_info = list(fake_disk_info_byname(instance).values())
+        disk_info = list(fake_disk_info_byname(instance, images_type).values())
         disk_info_text = jsonutils.dumps(disk_info)
         mock_get_disk_info.return_value = disk_info
 
@@ -23131,11 +23132,26 @@ def _test_migrate_disk_and_power_off(
             instance.uuid, mock.ANY, mock.ANY, '10.0.0.1', mock.ANY, mock.ANY)
         mock_unplug_vifs.assert_called_once()
 
+        assert_func = (
+            self.assertIn if assert_compression
+            else self.assertNotIn
+        )
+        for i, info in enumerate(disk_info):
+            self.assertIn('scp', mock_execute.call_args_list[i + 1].args)
+            assert_func('-C', mock_execute.call_args_list[i + 1].args)
+
     def test_migrate_disk_and_power_off(self):
         flavor = {'root_gb': 10, 'ephemeral_gb': 20}
         flavor_obj = objects.Flavor(**flavor)
         self._test_migrate_disk_and_power_off(self.context, flavor_obj)
 
+    def test_migrate_disk_and_power_off_compressed(self):
+        CONF.set_override('remote_copy_compression', True, group='libvirt')
+        flavor = {'root_gb': 10, 'ephemeral_gb': 20}
+        flavor_obj = objects.Flavor(**flavor)
+        self._test_migrate_disk_and_power_off(self.context, flavor_obj,
+                                              assert_compression=True)
+
     @mock.patch('nova.virt.libvirt.driver.LibvirtDriver._disconnect_volume')
     def test_migrate_disk_and_power_off_boot_from_volume(self,
                                                          disconnect_volume):
diff --git a/nova/tests/unit/virt/libvirt/test_imagebackend.py b/nova/tests/unit/virt/libvirt/test_imagebackend.py
index 29e5d55055c..b569b944862 100644
--- a/nova/tests/unit/virt/libvirt/test_imagebackend.py
+++ b/nova/tests/unit/virt/libvirt/test_imagebackend.py
@@ -197,13 +197,12 @@ def test_libvirt_info(self):
         self.assertEqual(30 * units.Mi, disk.disk_total_bytes_sec)
         self.assertEqual(3 * units.Ki, disk.disk_total_iops_sec)
 
-    @mock.patch('nova.virt.disk.api.get_disk_size')
-    def test_get_disk_size(self, get_disk_size):
+    def _test_get_disk_size(self, get_disk_size, size_call_attr='path'):
         get_disk_size.return_value = 2361393152
 
         image = self.image_class(self.INSTANCE, self.NAME)
         self.assertEqual(2361393152, image.get_disk_size(image.path))
-        get_disk_size.assert_called_once_with(image.path)
+        get_disk_size.assert_called_once_with(getattr(image, size_call_attr))
 
     def _test_libvirt_info_scsi_with_unit(self, disk_unit):
         disk_info = {
@@ -883,9 +882,11 @@ def fake_exists(path):
 
         mock_exists.side_effect = fake_exists
 
-        # Fake create_volume causes exists to return true for the volume
+        # Fake create_volume causes exists to return true and set the size for
+        # the volume
         def fake_create_volume(vg, lv, size, sparse=False):
             exists.add(os.path.join('/dev', vg, lv))
+            mock_lvm.get_volume_size.return_value = size
 
         mock_lvm.create_volume.side_effect = fake_create_volume
 
@@ -960,6 +961,10 @@ def test_create_image_generated_negative(self, mock_remove, mock_create):
         fn.assert_called_once_with(target=self.PATH, ephemeral_size=None)
         mock_remove.assert_called_once_with([self.PATH])
 
+    @mock.patch('nova.virt.libvirt.storage.lvm.get_volume_size')
+    def test_get_disk_size(self, get_disk_size):
+        self._test_get_disk_size(get_disk_size)
+
     def test_prealloc_image(self):
         CONF.set_override('preallocate_images', 'space')
 
@@ -1308,6 +1313,10 @@ def test_create_image_generated_encrypt_negative(self):
                 self.PATH.rpartition('/')[2])
             self.lvm.remove_volumes.assert_called_with([self.LV_PATH])
 
+    @mock.patch('nova.virt.libvirt.storage.lvm.get_volume_size')
+    def test_get_disk_size(self, get_disk_size):
+        self._test_get_disk_size(get_disk_size)
+
     def test_prealloc_image(self):
         self.flags(preallocate_images='space')
         fake_processutils.fake_execute_clear_log()
@@ -1569,13 +1578,9 @@ def test_image_path(self):
 
         self.assertEqual(image.path, rbd_path)
 
-    def test_get_disk_size(self):
-        image = self.image_class(self.INSTANCE, self.NAME)
-        with mock.patch.object(image.driver, 'size') as size_mock:
-            size_mock.return_value = 2361393152
-
-            self.assertEqual(2361393152, image.get_disk_size(image.path))
-            size_mock.assert_called_once_with(image.rbd_name)
+    @mock.patch('nova.storage.rbd_utils.RBDDriver.size')
+    def test_get_disk_size(self, get_disk_size):
+        self._test_get_disk_size(get_disk_size, 'rbd_name')
 
     @mock.patch.object(images, 'qemu_img_info',
                        return_value=imageutils.QemuImgInfo())
diff --git a/nova/tests/unit/virt/libvirt/test_utils.py b/nova/tests/unit/virt/libvirt/test_utils.py
index c47b2e947cb..7cf4e5b50ba 100644
--- a/nova/tests/unit/virt/libvirt/test_utils.py
+++ b/nova/tests/unit/virt/libvirt/test_utils.py
@@ -57,14 +57,14 @@ def test_copy_image_remote_ssh(self, mock_rem_fs_remove):
         self.flags(remote_filesystem_transport='ssh', group='libvirt')
         libvirt_utils.copy_image('src', 'dest', host='host')
         mock_rem_fs_remove.assert_called_once_with('src', 'host:dest',
-            on_completion=None, on_execute=None, compression=True)
+            on_completion=None, on_execute=None, compression=False)
 
     @mock.patch('nova.virt.libvirt.volume.remotefs.RsyncDriver.copy_file')
     def test_copy_image_remote_rsync(self, mock_rem_fs_remove):
         self.flags(remote_filesystem_transport='rsync', group='libvirt')
         libvirt_utils.copy_image('src', 'dest', host='host')
         mock_rem_fs_remove.assert_called_once_with('src', 'host:dest',
-            on_completion=None, on_execute=None, compression=True)
+            on_completion=None, on_execute=None, compression=False)
 
     @mock.patch('os.path.exists', return_value=True)
     def test_disk_type_from_path(self, mock_exists):
diff --git a/nova/virt/block_device.py b/nova/virt/block_device.py
index f7265117eb2..fc83a4ebdf8 100644
--- a/nova/virt/block_device.py
+++ b/nova/virt/block_device.py
@@ -216,6 +216,9 @@ def save(self):
                 setattr(self._bdm_obj, attr_name, self[lookup_name])
         self._bdm_obj.save()
 
+    def destroy(self):
+        self._bdm_obj.destroy()
+
 
 class DriverSwapBlockDevice(DriverBlockDevice):
     _fields = set([
@@ -227,6 +230,7 @@ class DriverSwapBlockDevice(DriverBlockDevice):
         'encryption_format',
         'encryption_options',
     ])
+    _proxy_as_attr_inherited = set(['volume_size'])
     _readonly_fields = set(['encrypted'])
     _update_on_save = {
         'disk_bus': None,
@@ -325,6 +329,7 @@ class DriverEphemeralBlockDevice(DriverBlockDevice):
         'encryption_format': None,
         'encryption_options': None,
     }
+    _proxy_as_attr_inherited = set(['volume_size'])
 
     def _transform(self):
         if not block_device.new_format_is_ephemeral(self._bdm_obj):
diff --git a/nova/virt/libvirt/driver.py b/nova/virt/libvirt/driver.py
index 64ebb9e8149..d895759f468 100644
--- a/nova/virt/libvirt/driver.py
+++ b/nova/virt/libvirt/driver.py
@@ -229,9 +229,6 @@ def repr_method(self):
 # Virtuozzo driver support
 MIN_VIRTUOZZO_VERSION = (7, 0, 0)
 
-# Names of the types that do not get compressed during migration
-NO_COMPRESSION_TYPES = ('qcow2',)
-
 # number of serial console limit
 QEMU_MAX_SERIAL_PORTS = 4
 # Qemu supports 4 serial consoles, we remove 1 because of the PTY one defined
@@ -12053,7 +12050,7 @@ def migrate_disk_and_power_off(self, context, instance, dest,
                 if fname == 'disk.swap':
                     continue
 
-                compression = info['type'] not in NO_COMPRESSION_TYPES
+                compression = CONF.libvirt.remote_copy_compression
                 libvirt_utils.copy_image(from_path, img_path, host=dest,
                                          on_execute=on_execute,
                                          on_completion=on_completion,
diff --git a/nova/virt/libvirt/imagebackend.py b/nova/virt/libvirt/imagebackend.py
index b1cfabf59b9..f058ffe73d3 100644
--- a/nova/virt/libvirt/imagebackend.py
+++ b/nova/virt/libvirt/imagebackend.py
@@ -304,7 +304,7 @@ def fetch_func_sync(target, *args, **kwargs):
         if size:
             # create_image() only creates the base image if needed, so
             # we cannot rely on it to exist here
-            if os.path.exists(base) and size > self.get_disk_size(base):
+            if os.path.exists(base) and size > self.get_disk_size(self.path):
                 self.resize_image(size)
 
             if (self.preallocate and self._can_fallocate() and
@@ -866,6 +866,9 @@ def create_lvm_image(base, size):
     def resize_image(self, size):
         pass
 
+    def get_disk_size(self, path):
+        return lvm.get_volume_size(path)
+
     @contextlib.contextmanager
     def remove_volume_on_error(self, path):
         try:
diff --git a/nova/virt/libvirt/migration.py b/nova/virt/libvirt/migration.py
index 25d271cac76..8e3cb22ac10 100644
--- a/nova/virt/libvirt/migration.py
+++ b/nova/virt/libvirt/migration.py
@@ -416,6 +416,10 @@ def _update_vif_xml(xml_doc, migrate_data, get_vif_config):
         # can't change during live migration.
         address = interface_dev.find('address')
         mtu = interface_dev.find('mtu')
+        # We always use the driver already found on the VM, as any changes to
+        # it have a high chance of causing libvirt to fail, due to mismatching
+        # virtio options.
+        driver = interface_dev.find('driver')
         # Now clear the interface's current elements and insert everything
         # from the destination vif config xml.
         interface_dev.clear()
@@ -428,7 +432,14 @@ def _update_vif_xml(xml_doc, migrate_data, get_vif_config):
             #               the live migration will crash.
             if dest_interface_subelem.tag == 'mtu' and mtu is None:
                 continue
+            # Never use the driver from the newly created XML
+            if dest_interface_subelem.tag == 'driver':
+                continue
             interface_dev.insert(index, dest_interface_subelem)
+        # Re-insert the driver.
+        if driver is not None:
+            interface_dev.insert(index + 1, driver)
+            index += 1
         # And finally re-insert the hw address.
         interface_dev.insert(index + 1, address)
 
diff --git a/nova/virt/libvirt/utils.py b/nova/virt/libvirt/utils.py
index 902852a7332..dce5ca1849c 100644
--- a/nova/virt/libvirt/utils.py
+++ b/nova/virt/libvirt/utils.py
@@ -310,7 +310,7 @@ def copy_image(
     receive: bool = False,
     on_execute: ty.Optional[ty.Callable] = None,
     on_completion: ty.Optional[ty.Callable] = None,
-    compression: bool = True,
+    compression: bool = False,
 ) -> None:
     """Copy a disk image to an existing directory
 
diff --git a/nova/virt/libvirt/volume/remotefs.py b/nova/virt/libvirt/volume/remotefs.py
index 8ddcaa1fccc..9d9085c0992 100644
--- a/nova/virt/libvirt/volume/remotefs.py
+++ b/nova/virt/libvirt/volume/remotefs.py
@@ -99,7 +99,7 @@ def remove_dir(self, host, dst_path, on_execute=None,
                                on_completion=on_completion)
 
     def copy_file(self, src, dst, on_execute=None,
-                    on_completion=None, compression=True):
+                    on_completion=None, compression=False):
         LOG.debug("Copying file %s to %s", src, dst)
         self.driver.copy_file(src, dst, on_execute=on_execute,
                               on_completion=on_completion,
diff --git a/releasenotes/notes/ssh-compression-config-4633a047e410983e.yaml b/releasenotes/notes/ssh-compression-config-4633a047e410983e.yaml
new file mode 100644
index 00000000000..1c9659722da
--- /dev/null
+++ b/releasenotes/notes/ssh-compression-config-4633a047e410983e.yaml
@@ -0,0 +1,12 @@
+---
+features:
+  - |
+    A new configuration option [libvirt]/remote_copy_compression has been
+    added. This option replaces the static list of disk formats which use
+    SSH compression for remote copy operations.
+
+    On fast local networks compression often reduces the copy bandwidth because
+    the CPU cannot compress fast enough.
+
+    By default compression is disabled. In previous releases compression was
+    for the raw, flat, lvm and ploop disks types.