Merge "Ceph: Fix restore backups to diff backend"

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.