Ceph: Fix restore backups to diff backend

The current implementation of the `_full_restore` method in the Ceph
backup driver is causing restoration to fail when restoring to another
volume type. This is because the method generates an incorrect image
name in the Ceph pool, resulting in the error message:

```
rbd.ImageNotFound: [errno 2] error opening image
b'volume-a91c11af-1147-4ac7-a5ce-61676736e076.backup.base' at snapshot
None.
```
Originally, the format `volume-<vol-uuid>.backup-<backup-uuid>` was used
for every backup created on the Ceph pool. However, after the change introduced in commit Ia08c252d747148e624f8d9e8b0e43f94773421e0, the
base name format was altered to volume-<uuid>.backup.base when
creating backups from Cinder snapshots.

This patch modifies the current if statement in _full_restore() to check
for the presence of a snapshot_id attribute in the backup. If the
attribute exists, the old base name format is used; otherwise, the new
format is applied. The if block statement in _full_restore() should now
match the format generated by the _full_backup() method.

Additionally, src_snap is activated when the destination volume is RBD
or if the backup is incremental. Currently, the Ceph backup driver only
supports incremental RBD backups, so incremental backups from non-RBD
volumes are not permitted.

NOTE: This patch primarily mirrors the modifications in
Ia08c252d747148e624f8d9e8b0e43f94773421e0, addressing it as a partial
bug fix. There's a consequential patch that further strengthens the
backup restore operations, which should be considered to fully resolve the bug. These two should be squashed and backported together for
optimal implementation.

Partial-Bug: #1895035
Change-Id: Iff8e1e90ab3c7b519819577ec3aafff838e6934f

cinder/backup/drivers/ceph.py

@@ -1087,16 +1087,27 @@ class CephBackupDriver(driver.BackupDriver):
 
         This will result in all extents being copied from source to
         destination.
+
+        :param backup: Backup object describing the backup to be restored.
+        :param dest_file: File object of the destination volume.
+        :param dest_name: Name of the destination volume.
+        :param length: Size of the destination volume in bytes.
+        :param volume_is_new: True if the destination volume is new.
+        :param src_snap: A string, the name of the restore point snapshot,
+        optional, used for incremental backups or RBD backup.
         """
         with eventlet.tpool.Proxy(rbd_driver.RADOSClient(self,
                                   backup.container)) as client:
-            # If a source snapshot is provided we assume the base is diff
-            # format.
-            if src_snap:
+            # In case of snapshot_id, the old base name format is used:
+            # volume-<vol-uuid>.backup.base
+            # Otherwise, the new base name format is used:
+            # volume-<vol-uuid>.backup-<backup-uuid>
+            # Should match the base name format in _full_backup()
+            if backup.snapshot_id:
+                backup_name = self._get_backup_base_name(backup.volume_id)
+            else:
                 backup_name = self._get_backup_base_name(backup.volume_id,
                                                          backup=backup)
-            else:
-                backup_name = self._get_backup_base_name(backup.volume_id)
 
             # Retrieve backup volume
             src_rbd = eventlet.tpool.Proxy(self.rbd.Image(client.ioctx,

cinder/tests/unit/backup/drivers/test_backup_ceph.py

@@ -933,6 +933,71 @@ class BackupCephTestCase(test.TestCase):
         self.assertTrue(self.service.rbd.Image.return_value.read.called)
         self.assertNotEqual(threading.current_thread(), thread_dict['thread'])
 
+    @common_mocks
+    def test_full_restore_without_snapshot_id_nor_src_snap(self):
+        length = 1024
+        volume_is_new = True
+        src_snap = ''
+        with tempfile.NamedTemporaryFile() as dest_file:
+            with mock.patch.object(self.service,
+                                   '_transfer_data') as mock_transfer_data,\
+                mock.patch.object(self.service,
+                                  '_get_backup_base_name') as mock_getbasename:
+
+                self.service._full_restore(self.backup, dest_file, length,
+                                           volume_is_new, src_snap)
+
+                mock_getbasename.assert_called_once_with(self.volume_id,
+                                                         backup=self.backup)
+                mock_transfer_data.assert_called_once()
+
+    @common_mocks
+    def test_full_restore_without_snapshot_id_w_src_snap(self):
+        length = 1024
+        volume_is_new = True
+        src_snap = 'random_snap'
+        with tempfile.NamedTemporaryFile() as dest_file:
+            with mock.patch.object(self.service,
+                                   '_transfer_data') as mock_transfer_data,\
+                mock.patch.object(self.service,
+                                  '_get_backup_base_name') as mock_getbasename:
+
+                self.service._full_restore(self.backup, dest_file, length,
+                                           volume_is_new, src_snap)
+
+                mock_getbasename.assert_called_once_with(self.volume_id,
+                                                         backup=self.backup)
+                mock_transfer_data.assert_called_once()
+
+    @common_mocks
+    def test_full_restore_with_snapshot_id(self):
+        length = 1024
+        volume_is_new = True
+        src_snap = ''
+
+        # Create alternate backup with snapshot_id
+        backup_id = fake.BACKUP4_ID
+        self._create_backup_db_entry(backup_id, self.volume_id,
+                                     self.volume_size)
+
+        backup = objects.Backup.get_by_id(self.ctxt, backup_id)
+        backup.snapshot_id = 'random_snap_id'
+        backup.container = "backups"
+        backup.parent = self.backup
+        backup.parent.service_metadata = '{"base": "random"}'
+
+        with tempfile.NamedTemporaryFile() as dest_file:
+            with mock.patch.object(self.service,
+                                   '_transfer_data') as mock_transfer_data,\
+                mock.patch.object(self.service,
+                                  '_get_backup_base_name') as mock_getbasename:
+
+                self.service._full_restore(backup, dest_file, length,
+                                           volume_is_new, src_snap)
+
+                mock_getbasename.assert_called_once_with(self.volume_id)
+                mock_transfer_data.assert_called_once()
+
     @common_mocks
     def test_discard_bytes(self):
         # Lower the chunksize to a memory manageable number

releasenotes/notes/bug-1895035-rbd-restore-0cd94ccd467ae1e3.yaml

@@ -0,0 +1,6 @@
+---
+fixes:
+  - |
+    Ceph backup driver `Bug #1895035
+    <https://bugs.launchpad.net/cinder/+bug/1895035>`_: Fixed restore full
+    backups to non RBD volumes.