Merge "NFS driver: Set volume format for snapshot operations"

cinder/image/image_utils.py

@@ -221,6 +221,20 @@ def qemu_img_info(
     return info
 
 
+def get_image_format(path: str,
+                     allow_qcow2_backing_file: Optional[bool] = False
+                     ) -> Optional[str]:
+    """Inspect image file and return its format if safe.
+
+    :param path: Path to the image file to inspect
+    :param allow_qcow2_backing_file: Allow qcow2 images with backing files
+    :returns: Format name ('raw', 'qcow2', etc.) if safe, None if unsafe
+    """
+    return cinder.privsep.format_inspector.get_format_if_safe(
+        path=path,
+        allow_qcow2_backing_file=allow_qcow2_backing_file)
+
+
 def get_qemu_img_version() -> Optional[list[int]]:
     """The qemu-img version will be cached until the process is restarted."""
 

cinder/tests/unit/test_image_utils.py

@@ -2409,6 +2409,23 @@ class TestTemporaryFileContextManager(test.TestCase):
         mock_delete.assert_called_once_with(mock.sentinel.temporary_file)
 
 
+@ddt.ddt
+class TestGetImageFormat(test.TestCase):
+    @ddt.data(False, True)
+    @mock.patch('cinder.privsep.format_inspector.get_format_if_safe')
+    def test_get_image_format(self, allow_qcow2_backing_file,
+                              mock_get_format_if_safe):
+        fake_path = mock.sentinel.fake_path
+        mock_get_format_if_safe.return_value = 'qcow2'
+
+        result = image_utils.get_image_format(
+            fake_path, allow_qcow2_backing_file=allow_qcow2_backing_file)
+
+        self.assertEqual('qcow2', result)
+        mock_get_format_if_safe.assert_called_once_with(
+            path=fake_path, allow_qcow2_backing_file=allow_qcow2_backing_file)
+
+
 class TestImageUtils(test.TestCase):
     def test_get_virtual_size(self):
         image_id = fake.IMAGE_ID

cinder/tests/unit/volume/drivers/test_nfs.py

@@ -1561,6 +1561,7 @@ class NfsDriverTestCase(test.TestCase):
         self._set_driver()
         drv = self._driver
         volume = self._simple_volume()
+        volume.admin_metadata = {'format': 'raw'}
         self.override_config('nfs_snapshot_support', True)
         fake_snap = fake_snapshot.fake_snapshot_obj(self.context)
         fake_snap.volume = volume
@@ -1570,6 +1571,7 @@ class NfsDriverTestCase(test.TestCase):
         snap_path = os.path.join(vol_dir, snap_file)
         info_path = os.path.join(vol_dir, volume['name']) + '.info'
 
+        mock_vol_save = self.mock_object(fake_snap.volume, 'save')
         with mock.patch.object(drv, '_local_path_volume_info',
                                return_value=info_path), \
                 mock.patch.object(drv, '_read_info_file', return_value={}), \
@@ -1582,12 +1584,15 @@ class NfsDriverTestCase(test.TestCase):
                 mock.patch.object(drv, 'get_active_image_from_info',
                                   return_value=volume['name']), \
                 mock.patch.object(drv, '_get_new_snap_path',
-                                  return_value=snap_path):
+                                  return_value=snap_path), \
+                mock.patch.object(fake_snap.volume, 'save') \
+                as mock_vol_save:
             self._driver.create_snapshot(fake_snap)
 
         mock_check_support.assert_called_once()
         mock_do_create_snapshot.assert_called_with(fake_snap, volume['name'],
                                                    snap_path)
+        mock_vol_save.assert_called_once()
         mock_write_info_file.assert_called_with(
             info_path, {'active': snap_file, fake_snap.id: snap_file})
 

cinder/tests/unit/volume/drivers/test_quobyte.py

@@ -733,10 +733,12 @@ class QuobyteDriverTestCase(test.TestCase):
 
         drv._create_sparsed_file = mock.Mock()
         drv._set_rw_permissions_for_all = mock.Mock()
+        mock_volume_save = self.mock_object(volume, 'save')
 
         drv._do_create_volume(volume)
         drv._create_sparsed_file.assert_called_once_with(mock.ANY, mock.ANY)
         drv._set_rw_permissions_for_all.assert_called_once_with(mock.ANY)
+        mock_volume_save.assert_called_once()
 
     def test_create_nonsparsed_volume(self):
         drv = self._driver
@@ -747,10 +749,12 @@ class QuobyteDriverTestCase(test.TestCase):
 
         drv._create_regular_file = mock.Mock()
         drv._set_rw_permissions_for_all = mock.Mock()
+        mock_volume_save = self.mock_object(volume, 'save')
 
         drv._do_create_volume(volume)
         drv._create_regular_file.assert_called_once_with(mock.ANY, mock.ANY)
         drv._set_rw_permissions_for_all.assert_called_once_with(mock.ANY)
+        mock_volume_save.assert_called_once()
 
         self._configuration.quobyte_sparsed_volumes = old_value
 
@@ -761,6 +765,7 @@ class QuobyteDriverTestCase(test.TestCase):
         old_value = self._configuration.quobyte_qcow2_volumes
         self._configuration.quobyte_qcow2_volumes = True
 
+        mock_volume_save = self.mock_object(volume, 'save')
         drv._execute = mock.Mock()
 
         hashed = drv._get_hash_str(volume['provider_location'])
@@ -777,6 +782,7 @@ class QuobyteDriverTestCase(test.TestCase):
                         mock.call('chmod', 'ugo+rw', path,
                                   run_as_root=self._driver._execute_as_root)]
         drv._execute.assert_has_calls(assert_calls)
+        mock_volume_save.assert_called_once()
 
         self._configuration.quobyte_qcow2_volumes = old_value
 

cinder/tests/unit/volume/drivers/test_remotefs.py

@@ -98,7 +98,10 @@ class RemoteFsSnapDriverTestCase(test.TestCase):
                               stale_snapshot=False,
                               is_active_image=True,
                               is_tmp_snap=False,
-                              encryption=False):
+                              encryption=False,
+                              base_format_exists=True,
+                              base_file_exists=True,
+                              base_file_inspect_format=None):
         # If the snapshot is not the active image, it is guaranteed that
         # another snapshot exists having it as backing file.
 
@@ -135,6 +138,7 @@ class RemoteFsSnapDriverTestCase(test.TestCase):
         else:
             fake_snap_img_info.backing_file = volume_name
         fake_snap_img_info.file_format = 'qcow2'
+        fake_snap_img_info.backing_file_format = 'raw'
         fake_base_img_info.backing_file = None
         fake_base_img_info.file_format = 'raw'
 
@@ -156,6 +160,12 @@ class RemoteFsSnapDriverTestCase(test.TestCase):
         exp_acceptable_states = ['available', 'in-use', 'backing-up',
                                  'deleting', 'downloading']
 
+        if base_format_exists:
+            snapshot.volume.admin_metadata = {'format': 'qcow2',
+                                              'base_format': 'raw'}
+        else:
+            snapshot.volume.admin_metadata = {'format': 'qcow2'}
+
         if volume_in_use:
             snapshot.volume.status = 'backing-up'
             snapshot.volume.attach_status = 'attached'
@@ -183,9 +193,61 @@ class RemoteFsSnapDriverTestCase(test.TestCase):
 
         elif is_active_image:
             self._driver._read_info_file.return_value = fake_info
+            mock_vol_save = self.mock_object(snapshot.volume, 'save')
+            mock_exists = self.mock_object(os.path, 'exists')
+            mock_get_image_format = self.mock_object(
+                image_utils, 'get_image_format')
+
+            mock_exists.return_value = base_file_exists
+            mock_get_image_format.return_value = base_file_inspect_format
 
             self._driver._delete_snapshot(snapshot)
 
+            if not base_format_exists:
+                # When base_format does not exist, we expect the function to
+                # try recoverying the information based on the state of
+                # volume's base file.
+                base_path = os.path.join(self._FAKE_MNT_POINT, volume_name)
+                mock_exists.assert_called_once_with(base_path)
+
+                if base_file_exists:
+                    # If base file exists, get_image_format is expected to be
+                    # called for detecting the base file format.
+                    mock_get_image_format.assert_called_once_with(
+                        path=base_path, allow_qcow2_backing_file=True)
+
+                    if base_file_inspect_format == 'raw':
+                        # When format inspection returns 'raw', the
+                        # base_format should be set to 'raw'.
+                        self.assertEqual(
+                            'raw',
+                            snapshot.volume.admin_metadata['base_format'])
+                        self.assertEqual(
+                            'raw',
+                            snapshot.volume.admin_metadata['format'])
+                        mock_vol_save.assert_called_once()
+                    else:
+                        # If inspection fails or returns non-raw (likely
+                        # 'qcow2'), it isn't possible to reliably
+                        # determine the original format. For this reason,
+                        # metadata should NOT be updated.
+                        self.assertNotIn(
+                            'base_format', snapshot.volume.admin_metadata)
+                        # Format should still be qcow2 as before
+                        self.assertEqual(
+                            'qcow2', snapshot.volume.admin_metadata['format'])
+                        # "save" should NOT be called since metadata wasn't
+                        # suposed to be updated.
+                        mock_vol_save.assert_not_called()
+                else:
+                    # When base_format is missing and base file doesn't exist,
+                    # verify base_format and format were set to 'qcow2'.
+                    self.assertEqual(
+                        'qcow2', snapshot.volume.admin_metadata['base_format'])
+                    self.assertEqual(
+                        'qcow2', snapshot.volume.admin_metadata['format'])
+                    mock_vol_save.assert_called_once()
+
             self._driver._img_commit.assert_called_once_with(
                 snapshot_path)
             self.assertNotIn(snapshot.id, fake_info)
@@ -217,9 +279,10 @@ class RemoteFsSnapDriverTestCase(test.TestCase):
 
             self._driver._img_commit.assert_called_once_with(
                 snapshot_path)
-            self._driver._rebase_img.assert_called_once_with(
-                fake_upper_snap_path, volume_name,
-                fake_base_img_info.file_format)
+            if volume_in_use:
+                self._driver._rebase_img.assert_called_once_with(
+                    fake_upper_snap_path, volume_name,
+                    fake_base_img_info.file_format)
             self._driver._write_info_file.assert_called_once_with(
                 mock.sentinel.fake_info_path, expected_info)
 
@@ -245,6 +308,45 @@ class RemoteFsSnapDriverTestCase(test.TestCase):
         self._test_delete_snapshot(is_active_image=False,
                                    encryption=encryption)
 
+    @ddt.data({'encryption': True}, {'encryption': False})
+    def test_delete_snapshot_base_format_recovery_base_not_exists(self,
+                                                                  encryption):
+        self._test_delete_snapshot(
+            is_active_image=True,
+            encryption=encryption,
+            base_format_exists=False,
+            base_file_exists=False)
+
+    @ddt.data({'encryption': True}, {'encryption': False})
+    def test_delete_snapshot_base_format_recovery_inspect_raw(self,
+                                                              encryption):
+        self._test_delete_snapshot(
+            is_active_image=True,
+            encryption=encryption,
+            base_format_exists=False,
+            base_file_exists=True,
+            base_file_inspect_format='raw')
+
+    @ddt.data({'encryption': True}, {'encryption': False})
+    def test_delete_snapshot_base_format_recovery_inspect_qcow2(self,
+                                                                encryption):
+        self._test_delete_snapshot(
+            is_active_image=True,
+            encryption=encryption,
+            base_format_exists=False,
+            base_file_exists=True,
+            base_file_inspect_format='qcow2')
+
+    @ddt.data({'encryption': True}, {'encryption': False})
+    def test_delete_snapshot_base_format_recovery_inspect_none(self,
+                                                               encryption):
+        self._test_delete_snapshot(
+            is_active_image=True,
+            encryption=encryption,
+            base_format_exists=False,
+            base_file_exists=True,
+            base_file_inspect_format=None)
+
     @ddt.data({'encryption': True}, {'encryption': False})
     def test_delete_stale_snapshot(self, encryption):
         if encryption:
@@ -278,6 +380,41 @@ class RemoteFsSnapDriverTestCase(test.TestCase):
         self._driver._write_info_file.assert_called_once_with(
             mock.sentinel.fake_info_path, expected_info)
 
+    @ddt.data({'snap_info': {'active': 'snap2', 'snap1': 'snap1',
+                             'snap2': 'snap2'},
+               'expected_snapshot_ids': ['snap1', 'snap2']},
+              {'snap_info': {},
+               'expected_snapshot_ids': []})
+    @ddt.unpack
+    def test_get_snapshots_from_snap_info(self, snap_info,
+                                          expected_snapshot_ids):
+        result = self._driver._get_snapshots_from_snap_info(snap_info)
+        self.assertEqual(expected_snapshot_ids, result)
+
+    @ddt.data({'read_info': {'active': 'snap2', 'snap1': 'snap1',
+                             'snap2': 'snap2'},
+               'expected_snapshot_ids': ['snap1', 'snap2']},
+              {'read_info': {},
+               'expected_snapshot_ids': []})
+    @ddt.unpack
+    @mock.patch.object(remotefs.RemoteFSSnapDriver, '_read_info_file')
+    @mock.patch.object(remotefs.RemoteFSSnapDriver, '_local_path_volume_info')
+    def test_get_volume_snapshots(self, mock_local_path_info,
+                                  mock_read_info_file, read_info,
+                                  expected_snapshot_ids):
+
+        mock_volume = mock.sentinel.volume
+        mock_local_path_info.return_value = mock.sentinel.fake_info_path
+        mock_read_info_file.return_value = read_info
+
+        result = self._driver._get_volume_snapshots(volume=mock_volume)
+
+        self.assertEqual(expected_snapshot_ids, result)
+
+        mock_local_path_info.assert_called_once_with(mock_volume)
+        mock_read_info_file.assert_called_once_with(
+            mock.sentinel.fake_info_path, empty_if_missing=True)
+
     @mock.patch.object(remotefs.RemoteFSDriver,
                        'secure_file_operations_enabled',
                        return_value=True)
@@ -352,6 +489,8 @@ class RemoteFsSnapDriverTestCase(test.TestCase):
             exp_acceptable_states.append('downloading')
             self._fake_snapshot.volume.status = 'downloading'
 
+        mock_vol_save = self.mock_object(snapshot.volume, 'save')
+        snapshot.volume.admin_metadata = {'format': 'raw'}
         if volume_in_use:
             snapshot.volume.status = 'backing-up'
             snapshot.volume.attach_status = 'attached'
@@ -363,7 +502,6 @@ class RemoteFsSnapDriverTestCase(test.TestCase):
                 self.context, connection_info=conn_info)
             snapshot.volume.volume_attachment.objects.append(attachment)
             mock_save = self.mock_object(attachment, 'save')
-            mock_vol_save = self.mock_object(snapshot.volume, 'save')
 
             # After the snapshot the connection info should change the name of
             # the file
@@ -389,9 +527,9 @@ class RemoteFsSnapDriverTestCase(test.TestCase):
         if volume_in_use:
             mock_save.assert_called_once()
             # We should have updated the volume format after the snapshot
-            mock_vol_save.assert_called_once()
             changed_fields = attachment.cinder_obj_get_changes()
             self.assertEqual(expected, changed_fields['connection_info'])
+        mock_vol_save.assert_called_once()
 
     @ddt.data({'encryption': True}, {'encryption': False})
     def test_create_snapshot_volume_available(self, encryption):

cinder/volume/drivers/quobyte.py

@@ -587,6 +587,13 @@ class QuobyteDriver(remotefs_drv.RemoteFSSnapDriverDistributed):
             else:
                 self._create_regular_file(volume_path, volume_size)
 
+        # It is required to store the volume format so the generic Snapshots
+        # implementation can revert to the volume's original format on snapshot
+        # deletion
+        volume.admin_metadata['format'] = self.format
+        with volume.obj_as_admin():
+            volume.save()
+
         self._set_rw_permissions_for_all(volume_path)
 
     def _load_shares_config(self, share_file=None):

cinder/volume/drivers/remotefs.py

@@ -793,6 +793,14 @@ class RemoteFSSnapDriverBase(RemoteFSDriver):
 
         self._nova = compute.API()
 
+        # Set the expected volume format based on driver configuration.
+        # This is used as a heuristic for base_format recovery during
+        # upgrade scenarios when metadata is missing.
+        self.format = 'raw'
+        if getattr(self.configuration, self.driver_prefix + '_qcow2_volumes',
+                   False):
+            self.format = 'qcow2'
+
     def snapshot_revert_use_temp_snapshot(self) -> bool:
         # Considering that RemoteFS based drivers use COW images
         # for storing snapshots, having chains of such images,
@@ -1324,30 +1332,34 @@ class RemoteFSSnapDriverBase(RemoteFSDriver):
             # info file instead of blocking
             return self._delete_stale_snapshot(snapshot)
 
-        base_path = os.path.join(vol_path, base_file)
-        base_file_img_info = self._qemu_img_info(base_path,
-                                                 snapshot.volume.name)
+        base_id = None
+        snapshots = [(snap_id, snap_file) for snap_id, snap_file
+                     in snap_info.items()
+                     if snap_id != 'active']
+        for snap_id, snap_file in snapshots:
+            if utils.paths_normcase_equal(snap_file, base_file):
+                base_id = snap_id
+                break
+
+        is_last_snapshot = False
+        if base_id is None:
+            # This snapshot is backed directly by the base volume file,
+            # meaning it's the last remaining snapshot in the chain
+            is_last_snapshot = True
 
         # Find what file has this as its backing file
         active_file = self.get_active_image_from_info(snapshot.volume)
 
+        base_path = os.path.join(vol_path, base_file)
+        base_file_img_info = self._qemu_img_info(base_path,
+                                                 snapshot.volume.name)
+
         if self._is_volume_attached(snapshot.volume):
             # Online delete
             context = snapshot._context
 
             new_base_file = base_file_img_info.backing_file
 
-            base_id = None
-            for key, value in snap_info.items():
-                if utils.paths_normcase_equal(value,
-                                              base_file) and key != 'active':
-                    base_id = key
-                    break
-            if base_id is None:
-                # This means we are deleting the oldest snapshot
-                LOG.debug('No %(base_id)s found for %(file)s',
-                          {'base_id': 'base_id', 'file': snapshot_file})
-
             online_delete_info = {
                 'active_file': active_file,
                 'snapshot_file': snapshot_file,
@@ -1389,6 +1401,89 @@ class RemoteFSSnapDriverBase(RemoteFSDriver):
                 self._img_commit(snapshot_path)
             # Active file has changed
             snap_info['active'] = base_file
+
+            # Restore original format when deleting last snapshot
+            if is_last_snapshot:
+                update_metadata = False
+                with snapshot.volume.obj_as_admin():
+                    admin_metadata = snapshot.volume.admin_metadata
+
+                    if 'base_format' in admin_metadata:
+                        # If base_format exists, update the volume's format
+                        # back to its original format
+                        base_format = admin_metadata['base_format']
+                        admin_metadata['format'] = base_format
+                        update_metadata = True
+                    else:
+                        # Volume has a snapshot but base_format doesn't exist,
+                        # this is likely because Cinder was upgraded from a
+                        # previous version that did not save the base_format.
+                        #
+                        # Even though the base file may still exist and
+                        # could be inspected to determine its format, the
+                        # result might be unreliable when the output differs
+                        # from 'raw' format. In case of a 'raw'
+                        # volume containing a 'qcow2' image, inspection will
+                        # report the inner format 'qcow2' instead of the
+                        # expected outer format 'raw'. This would lead to
+                        # the base format being set incorrectly in the
+                        # database.
+                        #
+                        # There are two possible ways to recover the original
+                        # volume format:
+                        #
+                        # 1. If the base file still exists, inspect it:
+                        #    a) If format is 'raw': No qcow2 header found,
+                        #       we can reliably assume 'raw' format.
+                        #    b) If format is NOT 'raw' (likely qcow2):
+                        #       Don't recover since we can't reliably
+                        #       determine the image format.
+                        # 2. If the base file no longer exists, this means
+                        #    the volume has been merged (originally raw or
+                        #    qcow2) into the active snapshot on a previous
+                        #    online snapshot deletion. We can assume 'qcow2'
+                        #    is the base format.
+                        if os.path.exists(base_path):
+                            # Base file exists, inspect it to determine format
+                            inspected_format = (
+                                image_utils.get_image_format(
+                                    path=base_path,
+                                    # Snapshot operations requires 'qcow2'
+                                    # backing file to be allowed
+                                    allow_qcow2_backing_file=True))
+
+                            if inspected_format == 'raw':
+                                # No qcow2 header found, assuming raw format
+                                admin_metadata['base_format'] = 'raw'
+                                admin_metadata['format'] = 'raw'
+                                msg = _("Recovered base_format as 'raw' for "
+                                        "volume %(volume_id)s from format "
+                                        "inspection.")
+                                LOG.info(
+                                    msg, {'volume_id': snapshot.volume.id})
+                                update_metadata = True
+                            else:
+                                # Format inspection failed or non-raw format
+                                # detected (e.g. qcow2)
+                                msg = _("Unable to recover base_format "
+                                        "for volume %(volume_id)s: can't "
+                                        "determine original format from "
+                                        "format inspection.")
+                                LOG.warning(
+                                    msg, {'volume_id': snapshot.volume.id})
+
+                        else:
+                            # Volume file no longer exists, this means the
+                            # volume has been merged into the active snapshot
+                            # during a previous online snapshot deletion and
+                            # qcow2 can be assumed as the new base format.
+                            admin_metadata['base_format'] = 'qcow2'
+                            admin_metadata['format'] = 'qcow2'
+                            update_metadata = True
+
+                    if update_metadata:
+                        snapshot.volume.save()
+
         else:
             #      T0        |      T1         |     T2         |      T3
             #     base       |  snapshot_file  |  higher_file   | highest_file
@@ -1587,6 +1682,30 @@ class RemoteFSSnapDriverBase(RemoteFSDriver):
                        new_snap_path]
             self._execute(*command, run_as_root=self._execute_as_root)
 
+    def _get_snapshots_from_snap_info(self, snap_info: dict) -> List[str]:
+        """Get list of snapshot IDs from snap_info dictionary.
+
+        Returns snapshot IDs from the snap_info dictionary, excluding the
+        'active' key.
+
+        :param snap_info: Snapshot info dictionary
+        :returns: List of snapshot IDs
+        """
+        return [k for k in snap_info.keys() if k != 'active']
+
+    def _get_volume_snapshots(self, volume: objects.Volume) -> List[str]:
+        """Get list of existing snapshots for a volume.
+
+        Reads the volume's snapshot info file and returns the list of
+        snapshot IDs.
+
+        :param volume: Volume object
+        :returns: List of snapshot IDs
+        """
+        info_path = self._local_path_volume_info(volume)
+        snap_info = self._read_info_file(info_path, empty_if_missing=True)
+        return self._get_snapshots_from_snap_info(snap_info)
+
     def _create_snapshot(self, snapshot: objects.Snapshot) -> None:
         """Create a snapshot.
 
@@ -1715,23 +1834,19 @@ class RemoteFSSnapDriverBase(RemoteFSDriver):
             snapshot.volume)
         new_snap_path = self._get_new_snap_path(snapshot)
         active = os.path.basename(new_snap_path)
+        active_format = 'qcow2'
 
         if self._is_volume_attached(snapshot.volume):
             self._create_snapshot_online(snapshot,
                                          backing_filename,
                                          new_snap_path)
-            # Update the format for the volume and the connection_info. The
-            # connection_info needs to reflect the current volume format in
+            # The connection_info needs to reflect the current volume format in
             # order for Nova to create the disk device correctly whenever the
-            # instance is stopped/started or rebooted.
-            new_format = 'qcow2'
-            snapshot.volume.admin_metadata['format'] = new_format
-            with snapshot.volume.obj_as_admin():
-                snapshot.volume.save()
-            # Update reference in the only attachment (no multi-attach support)
+            # instance is stopped/started or rebooted. Update reference in the
+            # only attachment (no multi-attach support)
             attachment = snapshot.volume.volume_attachment[0]
             attachment.connection_info['name'] = active
-            attachment.connection_info['format'] = new_format
+            attachment.connection_info['format'] = active_format
             # Let OVO know it has been updated
             attachment.connection_info = attachment.connection_info
             attachment.save()
@@ -1740,6 +1855,30 @@ class RemoteFSSnapDriverBase(RemoteFSDriver):
                                      backing_filename,
                                      new_snap_path)
 
+        # Update volume format metadata to match active snapshot
+        with snapshot.volume.obj_as_admin():
+            admin_metadata = snapshot.volume.admin_metadata
+            update_metadata = False
+            volume_has_snapshots = (
+                len(self._get_snapshots_from_snap_info(snap_info)) > 0)
+
+            if 'base_format' not in admin_metadata:
+                if not volume_has_snapshots:
+                    # This is the first snapshot, we need to save the original
+                    # volume format so we can later restore it if all snapshots
+                    # are eventually deleted offline.
+                    admin_metadata['base_format'] = admin_metadata['format']
+                    update_metadata = True
+
+            # No need to update the format metadata if it already matches the
+            # active snapshot.
+            if admin_metadata['format'] != active_format:
+                admin_metadata['format'] = active_format
+                update_metadata = True
+
+            if update_metadata:
+                snapshot.volume.save()
+
         snap_info['active'] = active
         snap_info[snapshot.id] = active
         self._write_info_file(info_path, snap_info)
@@ -1824,7 +1963,6 @@ class RemoteFSSnapDriverBase(RemoteFSDriver):
         # active file never changes
         info_path = self._local_path_volume_info(snapshot.volume)
         snap_info = self._read_info_file(info_path)
-        update_format = False
 
         if utils.paths_normcase_equal(info['active_file'],
                                       info['snapshot_file']):
@@ -1846,7 +1984,19 @@ class RemoteFSSnapDriverBase(RemoteFSDriver):
                            'volume_id': snapshot.volume.id}
 
             del snap_info[snapshot.id]
-            update_format = True
+
+            # If base_id is None, it means we are deleting the oldest snapshot
+            if info['base_id'] is None:
+                with snapshot.volume.obj_as_admin():
+                    admin_metadata = snapshot.volume.admin_metadata
+                    # When deleting the oldest snapshot, The previous base file
+                    # is merged into the active file, which then becomes the
+                    # new base. Consequently, the base_format must be updated
+                    # to match the active format.
+                    if 'format' in admin_metadata:
+                        admin_metadata['base_format'] = (
+                            admin_metadata['format'])
+                        snapshot.volume.save()
         else:
             # blockCommit snapshot into base
             # info['base'] <= snapshot_file
@@ -1862,11 +2012,6 @@ class RemoteFSSnapDriverBase(RemoteFSDriver):
 
         self._nova_assisted_vol_snap_delete(context, snapshot, delete_info)
 
-        if update_format:
-            snapshot.volume.admin_metadata['format'] = 'qcow2'
-            with snapshot.volume.obj_as_admin():
-                snapshot.volume.save()
-
         # Write info file updated above
         self._write_info_file(info_path, snap_info)