diff --git a/cinder/backup/api.py b/cinder/backup/api.py
index 31b4f3fba86..b6cc0192bde 100644
--- a/cinder/backup/api.py
+++ b/cinder/backup/api.py
@@ -279,7 +279,10 @@ class API(base.Base):
                 raise exception.InvalidBackup(reason=msg)
 
         parent_id = None
+        parent = None
+
         if latest_backup:
+            parent = latest_backup
             parent_id = latest_backup.id
             if latest_backup['status'] != fields.BackupStatus.AVAILABLE:
                 QUOTAS.rollback(context, reservations)
@@ -315,6 +318,7 @@ class API(base.Base):
                 'availability_zone': availability_zone,
                 'snapshot_id': snapshot_id,
                 'data_timestamp': data_timestamp,
+                'parent': parent,
                 'metadata': metadata or {}
             }
             backup = objects.Backup(context=context, **kwargs)
diff --git a/cinder/backup/drivers/ceph.py b/cinder/backup/drivers/ceph.py
index d0686a4a346..7bf4e846f18 100644
--- a/cinder/backup/drivers/ceph.py
+++ b/cinder/backup/drivers/ceph.py
@@ -43,6 +43,7 @@ restore to a new volume (default).
 """
 
 import fcntl
+import json
 import os
 import re
 import subprocess
@@ -314,22 +315,39 @@ class CephBackupDriver(driver.BackupDriver):
         ioctx.close()
         client.shutdown()
 
-    def _get_backup_base_name(self, volume_id, backup_id=None,
-                              diff_format=False):
+    def _format_base_name(self, service_metadata):
+        base_name = json.loads(service_metadata)["base"]
+        return utils.convert_str(base_name)
+
+    def _get_backup_base_name(self, volume_id, backup=None):
         """Return name of base image used for backup.
 
         Incremental backups use a new base name so we support old and new style
         format.
         """
         # Ensure no unicode
-        if diff_format:
+        if not backup:
             return utils.convert_str("volume-%s.backup.base" % volume_id)
-        else:
-            if backup_id is None:
-                msg = _("Backup id required")
-                raise exception.InvalidParameterValue(msg)
-            return utils.convert_str("volume-%s.backup.%s"
-                                     % (volume_id, backup_id))
+
+        if backup.service_metadata:
+            return self._format_base_name(backup.service_metadata)
+
+        # 'parent' field will only be present in incremental backups. This is
+        # filled by cinder-api
+        if backup.parent:
+            # Old backups don't have the base name in the service_metadata,
+            # so we use the default RBD backup base
+            if backup.parent.service_metadata:
+                service_metadata = backup.parent.service_metadata
+                base_name = self._format_base_name(service_metadata)
+            else:
+                base_name = utils.convert_str("volume-%s.backup.base"
+                                              % volume_id)
+
+            return base_name
+
+        return utils.convert_str("volume-%s.backup.%s"
+                                 % (volume_id, backup.id))
 
     def _discard_bytes(self, volume, offset, length):
         """Trim length bytes from offset.
@@ -479,7 +497,7 @@ class CephBackupDriver(driver.BackupDriver):
         if base_name is None:
             try_diff_format = True
 
-            base_name = self._get_backup_base_name(volume_id, backup.id)
+            base_name = self._get_backup_base_name(volume_id, backup=backup)
             LOG.debug("Trying diff format basename='%(basename)s' for "
                       "backup base image of volume %(volume)s.",
                       {'basename': base_name, 'volume': volume_id})
@@ -630,7 +648,7 @@ class CephBackupDriver(driver.BackupDriver):
         if name not in rbds:
             LOG.debug("Image '%s' not found - trying diff format name", name)
             if try_diff_format:
-                name = self._get_backup_base_name(volume_id, diff_format=True)
+                name = self._get_backup_base_name(volume_id)
                 if name not in rbds:
                     LOG.debug("Diff format image '%s' not found", name)
                     return False, name
@@ -657,50 +675,79 @@ class CephBackupDriver(driver.BackupDriver):
 
         return False
 
+    def _full_rbd_backup(self, container, base_name, length):
+        """Create the base_image for a full RBD backup."""
+        with eventlet.tpool.Proxy(rbd_driver.RADOSClient(self,
+                                  container)) as client:
+            self._create_base_image(base_name, length, client)
+        # Now we just need to return from_snap=None and image_created=True, if
+        # there is some exception in making backup snapshot, will clean up the
+        # base image.
+        return None, True
+
+    def _incremental_rbd_backup(self, backup, base_name, length,
+                                source_rbd_image, volume_id):
+        """Select the last snapshot for a RBD incremental backup."""
+
+        container = backup.container
+        last_incr = backup.parent_id
+        LOG.debug("Trying to perform an incremental backup with container: "
+                  "%(container)s, base_name: %(base)s, source RBD image: "
+                  "%(source)s, volume ID %(volume)s and last incremental "
+                  "backup ID: %(incr)s.",
+                  {'container': container,
+                   'base': base_name,
+                   'source': source_rbd_image,
+                   'volume': volume_id,
+                   'incr': last_incr,
+                   })
+
+        with eventlet.tpool.Proxy(rbd_driver.RADOSClient(self,
+                                  container)) as client:
+            base_rbd = eventlet.tpool.Proxy(self.rbd.Image(client.ioctx,
+                                                           base_name,
+                                                           read_only=True))
+            try:
+                from_snap = self._get_backup_snap_name(base_rbd,
+                                                       base_name,
+                                                       last_incr)
+                if from_snap is None:
+                    msg = (_(
+                        "Can't find snapshot from parent %(incr)s and "
+                        "base name image %(base)s.") %
+                        {'incr': last_incr, 'base': base_name})
+                    LOG.error(msg)
+                    raise exception.BackupRBDOperationFailed(msg)
+            finally:
+                base_rbd.close()
+
+        return from_snap, False
+
     def _backup_rbd(self, backup, volume_file, volume_name, length):
-        """Create an incremental backup from an RBD image."""
+        """Create an incremental or full backup from an RBD image."""
         rbd_user = volume_file.rbd_user
         rbd_pool = volume_file.rbd_pool
         rbd_conf = volume_file.rbd_conf
         source_rbd_image = eventlet.tpool.Proxy(volume_file.rbd_image)
         volume_id = backup.volume_id
-        updates = {}
-        base_name = self._get_backup_base_name(volume_id, diff_format=True)
-        image_created = False
-        with eventlet.tpool.Proxy(rbd_driver.RADOSClient(self,
-                                  backup.container)) as client:
-            # If from_snap does not exist at the destination (and the
-            # destination exists), this implies a previous backup has failed.
-            # In this case we will force a full backup.
-            #
-            # TODO(dosaboy): find a way to repair the broken backup
-            #
-            if base_name not in eventlet.tpool.Proxy(self.rbd.RBD()).list(
-                    ioctx=client.ioctx):
-                src_vol_snapshots = self.get_backup_snaps(source_rbd_image)
-                if src_vol_snapshots:
-                    # If there are source volume snapshots but base does not
-                    # exist then we delete it and set from_snap to None
-                    LOG.debug("Volume '%(volume)s' has stale source "
-                              "snapshots so deleting them.",
-                              {'volume': volume_id})
-                for snap in src_vol_snapshots:
-                    from_snap = snap['name']
-                    source_rbd_image.remove_snap(from_snap)
-                from_snap = None
+        base_name = None
 
-                # Create new base image
-                self._create_base_image(base_name, length, client)
-                image_created = True
-            else:
-                # If a from_snap is defined and is present in the source volume
-                # image but does not exist in the backup base then we look down
-                # the list of source volume snapshots and find the latest one
-                # for which a backup snapshot exist in the backup base. Until
-                # that snapshot is reached, we delete all the other snapshots
-                # for which backup snapshot does not exist.
-                from_snap = self._get_most_recent_snap(source_rbd_image,
-                                                       base_name, client)
+        # If backup.parent_id is None performs full RBD backup
+        if backup.parent_id is None:
+            base_name = self._get_backup_base_name(volume_id, backup=backup)
+            from_snap, image_created = self._full_rbd_backup(backup.container,
+                                                             base_name,
+                                                             length)
+        # Otherwise performs incremental rbd backup
+        else:
+            # Find the base name from the parent backup's service_metadata
+            base_name = self._get_backup_base_name(volume_id, backup=backup)
+            rbd_img = source_rbd_image
+            from_snap, image_created = self._incremental_rbd_backup(backup,
+                                                                    base_name,
+                                                                    length,
+                                                                    rbd_img,
+                                                                    volume_id)
 
         LOG.debug("Using --from-snap '%(snap)s' for incremental backup of "
                   "volume %(volume)s.",
@@ -744,14 +791,8 @@ class CephBackupDriver(driver.BackupDriver):
                           "source volume='%(volume)s'.",
                           {'snapshot': new_snap, 'volume': volume_id})
                 source_rbd_image.remove_snap(new_snap)
-        # We update the parent_id here. The from_snap is of the format:
-        # backup.BACKUP_ID.snap.TIMESTAMP. So we need to extract the
-        # backup_id of the parent only from from_snap and set it as
-        # parent_id
-        if from_snap:
-            parent_id = from_snap.split('.')
-            updates = {'parent_id': parent_id[1]}
-        return updates
+
+        return {'service_metadata': '{"base": "%s"}' % base_name}
 
     def _file_is_rbd(self, volume_file):
         """Returns True if the volume_file is actually an RBD image."""
@@ -765,7 +806,7 @@ class CephBackupDriver(driver.BackupDriver):
         image.
         """
         volume_id = backup.volume_id
-        backup_name = self._get_backup_base_name(volume_id, backup.id)
+        backup_name = self._get_backup_base_name(volume_id, backup=backup)
 
         with eventlet.tpool.Proxy(rbd_driver.RADOSClient(self,
                                   backup.container)) as client:
@@ -868,23 +909,6 @@ class CephBackupDriver(driver.BackupDriver):
         LOG.debug("Found snapshot '%s'", snaps[0])
         return snaps[0]
 
-    def _get_most_recent_snap(self, rbd_image, base_name, client):
-        """Get the most recent backup snapshot of the provided image.
-
-        Returns name of most recent backup snapshot or None if there are no
-        backup snapshots.
-        """
-        src_vol_backup_snaps = self.get_backup_snaps(rbd_image, sort=True)
-        from_snap = None
-
-        for snap in src_vol_backup_snaps:
-            if self._snap_exists(base_name, snap['name'], client):
-                from_snap = snap['name']
-                break
-            rbd_image.remove_snap(snap['name'])
-
-        return from_snap
-
     def _get_volume_size_gb(self, volume):
         """Return the size in gigabytes of the given volume.
 
@@ -938,17 +962,23 @@ class CephBackupDriver(driver.BackupDriver):
         volume_file.seek(0)
         length = self._get_volume_size_gb(volume)
 
-        do_full_backup = False
-        if self._file_is_rbd(volume_file):
-            # If volume an RBD, attempt incremental backup.
-            LOG.debug("Volume file is RBD: attempting incremental backup.")
+        if backup.snapshot_id:
+            do_full_backup = True
+        elif self._file_is_rbd(volume_file):
+            # If volume an RBD, attempt incremental or full backup.
+            do_full_backup = False
+            LOG.debug("Volume file is RBD: attempting optimized backup")
             try:
-                updates = self._backup_rbd(backup, volume_file,
-                                           volume.name, length)
+                updates = self._backup_rbd(backup, volume_file, volume.name,
+                                           length)
             except exception.BackupRBDOperationFailed:
-                LOG.debug("Forcing full backup of volume %s.", volume.id)
-                do_full_backup = True
+                with excutils.save_and_reraise_exception():
+                    self.delete_backup(backup)
         else:
+            if backup.parent_id:
+                LOG.debug("Volume file is NOT RBD: can't perform"
+                          "incremental backup.")
+                raise exception.BackupRBDOperationFailed
             LOG.debug("Volume file is NOT RBD: will do full backup.")
             do_full_backup = True
 
@@ -970,11 +1000,6 @@ class CephBackupDriver(driver.BackupDriver):
         LOG.debug("Backup '%(backup_id)s' of volume %(volume_id)s finished.",
                   {'backup_id': backup.id, 'volume_id': volume.id})
 
-        # If updates is empty then set parent_id to None. This will
-        # take care if --incremental flag is used in CLI but a full
-        # backup is performed instead
-        if not updates and backup.parent_id:
-            updates = {'parent_id': None}
         return updates
 
     def _full_restore(self, backup, dest_file, dest_name, length,
@@ -989,13 +1014,10 @@ class CephBackupDriver(driver.BackupDriver):
             # If a source snapshot is provided we assume the base is diff
             # format.
             if src_snap:
-                diff_format = True
+                backup_name = self._get_backup_base_name(backup.volume_id,
+                                                         backup=backup)
             else:
-                diff_format = False
-
-            backup_name = self._get_backup_base_name(backup.volume_id,
-                                                     backup_id=backup.id,
-                                                     diff_format=diff_format)
+                backup_name = self._get_backup_base_name(backup.volume_id)
 
             # Retrieve backup volume
             src_rbd = eventlet.tpool.Proxy(self.rbd.Image(client.ioctx,
@@ -1022,7 +1044,7 @@ class CephBackupDriver(driver.BackupDriver):
         post-process and resize it back to its expected size.
         """
         backup_base = self._get_backup_base_name(backup.volume_id,
-                                                 diff_format=True)
+                                                 backup=backup)
 
         with eventlet.tpool.Proxy(rbd_driver.RADOSClient(self,
                                   backup.container)) as client:
@@ -1047,7 +1069,7 @@ class CephBackupDriver(driver.BackupDriver):
         rbd_pool = restore_file.rbd_pool
         rbd_conf = restore_file.rbd_conf
         base_name = self._get_backup_base_name(backup.volume_id,
-                                               diff_format=True)
+                                               backup=backup)
 
         LOG.debug("Attempting incremental restore from base='%(base)s' "
                   "snap='%(snap)s'",
@@ -1179,8 +1201,10 @@ class CephBackupDriver(driver.BackupDriver):
         """
         length = int(volume.size) * units.Gi
 
-        base_name = self._get_backup_base_name(backup.volume_id,
-                                               diff_format=True)
+        if backup.service_metadata:
+            base_name = self._get_backup_base_name(backup.volume_id, backup)
+        else:
+            base_name = self._get_backup_base_name(backup.volume_id)
 
         with eventlet.tpool.Proxy(rbd_driver.RADOSClient(
                                   self, backup.container)) as client:
diff --git a/cinder/objects/backup.py b/cinder/objects/backup.py
index e1fa7b48b40..514a25315e2 100644
--- a/cinder/objects/backup.py
+++ b/cinder/objects/backup.py
@@ -40,7 +40,8 @@ class Backup(base.CinderPersistentObject, base.CinderObject,
     # Version 1.4: Add restore_volume_id
     # Version 1.5: Add metadata
     # Version 1.6: Add encryption_key_id
-    VERSION = '1.6'
+    # Version 1.7: Add parent
+    VERSION = '1.7'
 
     OPTIONAL_FIELDS = ('metadata',)
 
@@ -55,6 +56,7 @@ class Backup(base.CinderPersistentObject, base.CinderObject,
         'availability_zone': fields.StringField(nullable=True),
         'container': fields.StringField(nullable=True),
         'parent_id': fields.StringField(nullable=True),
+        'parent': fields.ObjectField('Backup', nullable=True),
         'status': c_fields.BackupStatusField(nullable=True),
         'fail_reason': fields.StringField(nullable=True),
         'size': fields.IntegerField(nullable=True),
@@ -110,8 +112,14 @@ class Backup(base.CinderPersistentObject, base.CinderObject,
 
     def obj_make_compatible(self, primitive, target_version):
         """Make an object representation compatible with a target version."""
+        added_fields = (((1, 7), ('parent',)),)
+
         super(Backup, self).obj_make_compatible(primitive, target_version)
         target_version = versionutils.convert_version_to_tuple(target_version)
+        for version, remove_fields in added_fields:
+            if target_version < version:
+                for obj_field in remove_fields:
+                    primitive.pop(obj_field, None)
 
     @classmethod
     def _from_db_object(cls, context, backup, db_backup, expected_attrs=None):
@@ -174,6 +182,7 @@ class Backup(base.CinderPersistentObject, base.CinderObject,
                 self.metadata = db.backup_metadata_update(self._context,
                                                           self.id, metadata,
                                                           True)
+            updates.pop('parent', None)
             db.backup_update(self._context, self.id, updates)
 
         self.obj_reset_changes()
diff --git a/cinder/objects/base.py b/cinder/objects/base.py
index 1efb281d1d0..3dc68e7a69a 100644
--- a/cinder/objects/base.py
+++ b/cinder/objects/base.py
@@ -150,6 +150,7 @@ OBJ_VERSIONS.add('1.34', {'VolumeAttachment': '1.3'})
 OBJ_VERSIONS.add('1.35', {'Backup': '1.6', 'BackupImport': '1.6'})
 OBJ_VERSIONS.add('1.36', {'RequestSpec': '1.4'})
 OBJ_VERSIONS.add('1.37', {'RequestSpec': '1.5'})
+OBJ_VERSIONS.add('1.38', {'Backup': '1.7', 'BackupImport': '1.7'})
 
 
 class CinderObjectRegistry(base.VersionedObjectRegistry):
diff --git a/cinder/tests/unit/backup/drivers/test_backup_ceph.py b/cinder/tests/unit/backup/drivers/test_backup_ceph.py
index 1f4b6f229be..048ea4d4edb 100644
--- a/cinder/tests/unit/backup/drivers/test_backup_ceph.py
+++ b/cinder/tests/unit/backup/drivers/test_backup_ceph.py
@@ -15,6 +15,7 @@
 """ Tests for Ceph backup service."""
 
 import hashlib
+import json
 import os
 import tempfile
 import threading
@@ -39,6 +40,7 @@ from cinder.i18n import _
 from cinder import objects
 from cinder import test
 from cinder.tests.unit import fake_constants as fake
+import cinder.volume.drivers.rbd as rbd_driver
 
 # This is used to collect raised exceptions so that tests may check what was
 # raised.
@@ -119,6 +121,14 @@ class BackupCephTestCase(test.TestCase):
                   'user_id': userid, 'project_id': projectid}
         return db.backup_create(self.ctxt, backup)['id']
 
+    def _create_parent_backup_object(self):
+        tmp_backup_id = fake.BACKUP3_ID
+        self._create_backup_db_entry(tmp_backup_id, self.volume_id,
+                                     self.volume_size)
+        tmp_backup = objects.Backup.get_by_id(self.ctxt, tmp_backup_id)
+        tmp_backup.service_metadata = 'mock_base_name'
+        return tmp_backup
+
     def time_inc(self):
         self.counter += 1
         return self.counter
@@ -170,6 +180,22 @@ class BackupCephTestCase(test.TestCase):
         self.backup = objects.Backup.get_by_id(self.ctxt, self.backup_id)
         self.backup.container = "backups"
 
+        # Create parent backup of volume
+        self.parent_backup = self._create_parent_backup_object()
+
+        # Create alternate backup with parent
+        self.alt_backup_id = fake.BACKUP2_ID
+        self._create_backup_db_entry(self.alt_backup_id, self.volume_id,
+                                     self.volume_size)
+
+        self.alt_backup = objects.Backup.get_by_id(self.ctxt,
+                                                   self.alt_backup_id)
+
+        base_name = "volume-%s.backup.%s" % (self.volume_id, self.backup_id)
+        self.alt_backup.container = "backups"
+        self.alt_backup.parent = self.backup
+        self.alt_backup.parent.service_metadata = '{"base": "%s"}' % base_name
+
         # Create alternate volume.
         self.alt_volume_id = str(uuid.uuid4())
         self._create_volume_db_entry(self.alt_volume_id, self.volume_size)
@@ -255,24 +281,6 @@ class BackupCephTestCase(test.TestCase):
         self.assertFalse(oldformat)
         self.assertEqual(1 | 2 | 4 | 64, features)
 
-    @common_mocks
-    def test_get_most_recent_snap(self):
-        last = 'backup.%s.snap.9824923.1212' % (uuid.uuid4())
-
-        image = self.mock_rbd.Image.return_value
-        with mock.patch.object(self.service, '_snap_exists') as \
-                mock_snap_exists:
-            mock_snap_exists.return_value = True
-            image.list_snaps.return_value = \
-                [{'name': 'backup.%s.snap.6423868.2342' % (uuid.uuid4())},
-                 {'name': 'backup.%s.snap.1321319.3235' % (uuid.uuid4())},
-                 {'name': last},
-                 {'name': 'backup.%s.snap.3824923.1412' % (uuid.uuid4())}]
-            base_name = "mock_base"
-            client = mock.Mock()
-            snap = self.service._get_most_recent_snap(image, base_name, client)
-        self.assertEqual(last, snap)
-
     @common_mocks
     def test_get_backup_snap_name(self):
         snap_name = 'backup.%s.snap.3824923.1412' % (uuid.uuid4())
@@ -415,7 +423,7 @@ class BackupCephTestCase(test.TestCase):
         with mock.patch.object(self.service, '_backup_metadata'):
             with mock.patch.object(self.service, '_discard_bytes'):
                 with tempfile.NamedTemporaryFile() as test_file:
-                    self.service.backup(self.backup, self.volume_file)
+                    self.service.backup(self.alt_backup, self.volume_file)
 
                     # Ensure the files are equal
                     self.assertEqual(checksum.digest(), self.checksum.digest())
@@ -424,25 +432,34 @@ class BackupCephTestCase(test.TestCase):
         self.assertNotEqual(threading.current_thread(), thread_dict['thread'])
 
     @common_mocks
-    def test_get_backup_base_name(self):
-        name = self.service._get_backup_base_name(self.volume_id,
-                                                  diff_format=True)
+    def test_get_backup_base_name_without_backup_param(self):
+        """Test _get_backup_base_name without backup."""
+        name = self.service._get_backup_base_name(self.volume_id)
         self.assertEqual("volume-%s.backup.base" % (self.volume_id), name)
 
-        self.assertRaises(exception.InvalidParameterValue,
-                          self.service._get_backup_base_name,
-                          self.volume_id)
+    @common_mocks
+    def test_get_backup_base_name_w_backup_and_no_parent(self):
+        """Test _get_backup_base_name with backup and no parent."""
+        name = self.service._get_backup_base_name(self.volume_id,
+                                                  self.backup)
+        self.assertEqual("volume-%s.backup.%s" %
+                         (self.volume_id, self.backup.id), name)
 
-        name = self.service._get_backup_base_name(self.volume_id, '1234')
-        self.assertEqual("volume-%s.backup.%s" % (self.volume_id, '1234'),
-                         name)
+    @common_mocks
+    def test_get_backup_base_name_w_backup_and_parent(self):
+        """Test _get_backup_base_name with backup and parent."""
+        name = self.service._get_backup_base_name(self.volume_id,
+                                                  self.alt_backup)
+        base_name = json.loads(self.alt_backup.parent.service_metadata)
+        self.assertEqual(base_name["base"], name)
 
     @common_mocks
     @mock.patch('fcntl.fcntl', spec=True)
     @mock.patch('subprocess.Popen', spec=True)
     def test_backup_volume_from_rbd(self, mock_popen, mock_fnctl):
+        """Test full RBD backup generated successfully."""
         backup_name = self.service._get_backup_base_name(self.volume_id,
-                                                         diff_format=True)
+                                                         self.alt_backup)
 
         def mock_write_data():
             self.volume_file.seek(0)
@@ -483,8 +500,11 @@ class BackupCephTestCase(test.TestCase):
                                  {'name': 'backup.mock.snap.15341241.90'},
                                  {'name': 'backup.mock.snap.199994362.10'}])
 
-                            output = self.service.backup(self.backup, rbdio)
-                            self.assertDictEqual({}, output)
+                            output = self.service.backup(self.alt_backup,
+                                                         rbdio)
+                            base_name = '{"base": "%s"}' % backup_name
+                            service_meta = {'service_metadata': base_name}
+                            self.assertDictEqual(service_meta, output)
 
                             self.assertEqual(['popen_init',
                                               'read',
@@ -494,7 +514,7 @@ class BackupCephTestCase(test.TestCase):
                                               'communicate'], self.callstack)
 
                             self.assertFalse(mock_full_backup.called)
-                            self.assertTrue(mock_get_backup_snaps.called)
+                            self.assertFalse(mock_get_backup_snaps.called)
 
                             # Ensure the files are equal
                             self.assertEqual(checksum.digest(),
@@ -505,7 +525,7 @@ class BackupCephTestCase(test.TestCase):
         with mock.patch.object(self.service, '_backup_rbd') as \
                 mock_backup_rbd, mock.patch.object(self.service,
                                                    '_backup_metadata'):
-            mock_backup_rbd.return_value = {'parent_id': 'mock'}
+            mock_backup_rbd.return_value = {'service_metadata': 'base_name'}
             image = self.service.rbd.Image()
             meta = linuxrbd.RBDImageMetadata(image,
                                              'pool_foo',
@@ -513,15 +533,14 @@ class BackupCephTestCase(test.TestCase):
                                              'conf_foo')
             rbdio = linuxrbd.RBDVolumeIOWrapper(meta)
             output = self.service.backup(self.backup, rbdio)
-            self.assertDictEqual({'parent_id': 'mock'}, output)
+            self.assertDictEqual({'service_metadata': 'base_name'}, output)
 
     @common_mocks
-    def test_backup_volume_from_rbd_set_parent_id_none(self):
-        backup_name = self.service._get_backup_base_name(
-            self.volume_id, diff_format=True)
+    def test_backup_volume_from_rbd_got_exception(self):
+        base_name = self.service._get_backup_base_name(self.volume_id,
+                                                       self.alt_backup)
 
-        self.mock_rbd.RBD().list.return_value = [backup_name]
-        self.backup.parent_id = 'mock_parent_id'
+        self.mock_rbd.RBD().list.return_value = [base_name]
 
         with mock.patch.object(self.service, 'get_backup_snaps'), \
                 mock.patch.object(self.service, '_rbd_diff_transfer') as \
@@ -551,29 +570,54 @@ class BackupCephTestCase(test.TestCase):
                                                          'conf_foo')
                         rbdio = linuxrbd.RBDVolumeIOWrapper(meta)
                         mock_get_backup_snaps.return_value = (
-                            [{'name': 'backup.mock.snap.153464362.12'},
-                             {'name': 'backup.mock.snap.199994362.10'}])
-                        output = self.service.backup(self.backup, rbdio)
-                        self.assertIsNone(output['parent_id'])
+                            [{'name': 'backup.mock.snap.153464362.12',
+                              'backup_id': 'mock_parent_id'},
+                             {'name': 'backup.mock.snap.199994362.10',
+                              'backup_id': 'mock'}])
+                        self.assertRaises(exception.BackupRBDOperationFailed,
+                                          self.service.backup,
+                                          self.alt_backup, rbdio)
 
     @common_mocks
     def test_backup_rbd_set_parent_id(self):
-        backup_name = self.service._get_backup_base_name(
-            self.volume_id, diff_format=True)
+        base_name = self.service._get_backup_base_name(self.volume_id,
+                                                       self.alt_backup)
         vol_name = self.volume.name
         vol_length = self.volume.size
 
-        self.mock_rbd.RBD().list.return_value = [backup_name]
+        self.mock_rbd.RBD().list.return_value = [base_name]
 
         with mock.patch.object(self.service, '_snap_exists'), \
-                mock.patch.object(self.service, '_get_backup_base_name') as \
-                mock_get_backup_base_name, \
-                mock.patch.object(self.service, '_get_most_recent_snap') as \
-                mock_get_most_recent_snap, \
+                mock.patch.object(self.service, '_get_backup_snap_name') as \
+                mock_get_backup_snap_name, \
                 mock.patch.object(self.service, '_rbd_diff_transfer'):
-            mock_get_backup_base_name.return_value = backup_name
-            mock_get_most_recent_snap.return_value = (
-                'backup.mock.snap.153464362.12')
+            image = self.service.rbd.Image()
+            mock_get_backup_snap_name.return_value = 'mock_snap_name'
+            meta = linuxrbd.RBDImageMetadata(image,
+                                             'pool_foo',
+                                             'user_foo',
+                                             'conf_foo')
+            rbdio = linuxrbd.RBDVolumeIOWrapper(meta)
+            rbdio.seek(0)
+            output = self.service._backup_rbd(self.alt_backup, rbdio,
+                                              vol_name, vol_length)
+            base_name = '{"base": "%s"}' % base_name
+            self.assertEqual({'service_metadata': base_name}, output)
+            self.backup.parent_id = None
+
+    @common_mocks
+    def test_backup_rbd_without_parent_id(self):
+        full_backup_name = self.service._get_backup_base_name(self.volume_id,
+                                                              self.alt_backup)
+        vol_name = self.volume.name
+        vol_length = self.volume.size
+
+        with mock.patch.object(self.service, '_rbd_diff_transfer'), \
+                mock.patch.object(self.service, '_create_base_image') as \
+                mock_create_base_image, mock.patch.object(
+                rbd_driver, 'RADOSClient') as mock_rados_client:
+            client = mock.Mock()
+            mock_rados_client.return_value.__enter__.return_value = client
             image = self.service.rbd.Image()
             meta = linuxrbd.RBDImageMetadata(image,
                                              'pool_foo',
@@ -581,9 +625,12 @@ class BackupCephTestCase(test.TestCase):
                                              'conf_foo')
             rbdio = linuxrbd.RBDVolumeIOWrapper(meta)
             rbdio.seek(0)
-            output = self.service._backup_rbd(self.backup, rbdio,
+            output = self.service._backup_rbd(self.alt_backup, rbdio,
                                               vol_name, vol_length)
-            self.assertDictEqual({'parent_id': 'mock'}, output)
+            mock_create_base_image.assert_called_with(full_backup_name,
+                                                      vol_length, client)
+            base_name = '{"base": "%s"}' % full_backup_name
+            self.assertEqual({'service_metadata': base_name}, output)
 
     @common_mocks
     @mock.patch('fcntl.fcntl', spec=True)
@@ -597,7 +644,7 @@ class BackupCephTestCase(test.TestCase):
         self._try_delete_base_image().
         """
         backup_name = self.service._get_backup_base_name(self.volume_id,
-                                                         diff_format=True)
+                                                         self.alt_backup)
 
         def mock_write_data():
             self.volume_file.seek(0)
@@ -662,7 +709,7 @@ class BackupCephTestCase(test.TestCase):
                             self.assertRaises(
                                 self.service.rbd.ImageNotFound,
                                 self.service.backup,
-                                self.backup, rbdio)
+                                self.alt_backup, rbdio)
 
     @common_mocks
     @mock.patch('fcntl.fcntl', spec=True)
@@ -675,7 +722,7 @@ class BackupCephTestCase(test.TestCase):
         second exception occurs in self.delete_backup().
         """
         backup_name = self.service._get_backup_base_name(self.volume_id,
-                                                         diff_format=True)
+                                                         self.alt_backup)
 
         def mock_write_data():
             self.volume_file.seek(0)
@@ -733,12 +780,11 @@ class BackupCephTestCase(test.TestCase):
                         self.assertRaises(
                             self.service.rbd.ImageBusy,
                             self.service.backup,
-                            self.backup, rbdio)
+                            self.alt_backup, rbdio)
 
     @common_mocks
     def test_backup_rbd_from_snap(self):
-        backup_name = self.service._get_backup_base_name(self.volume_id,
-                                                         diff_format=True)
+        backup_name = self.service._get_backup_base_name(self.volume_id)
         vol_name = self.volume['name']
         vol_length = self.service._get_volume_size_gb(self.volume)
 
@@ -780,44 +826,36 @@ class BackupCephTestCase(test.TestCase):
 
     @common_mocks
     def test_backup_rbd_from_snap2(self):
-        backup_name = self.service._get_backup_base_name(self.volume_id,
-                                                         diff_format=True)
+        base_name = self.service._get_backup_base_name(self.volume_id,
+                                                       self.alt_backup)
         vol_name = self.volume['name']
         vol_length = self.service._get_volume_size_gb(self.volume)
 
         self.mock_rbd.RBD().list = mock.Mock()
-        self.mock_rbd.RBD().list.return_value = [backup_name]
+        self.mock_rbd.RBD().list.return_value = [base_name]
 
-        with mock.patch.object(self.service, '_get_most_recent_snap') as \
-                mock_get_most_recent_snap:
-            with mock.patch.object(self.service, '_get_backup_base_name') as \
-                    mock_get_backup_base_name:
-                with mock.patch.object(self.service, '_rbd_diff_transfer') as \
-                        mock_rbd_diff_transfer:
-                    with mock.patch.object(self.service,
-                                           '_get_new_snap_name') as \
-                            mock_get_new_snap_name:
-                        mock_get_backup_base_name.return_value = (
-                            backup_name)
-                        mock_get_most_recent_snap.return_value = (
-                            'backup.mock.snap.153464362.12')
-                        mock_get_new_snap_name.return_value = 'new_snap'
-                        image = self.service.rbd.Image()
-                        meta = linuxrbd.RBDImageMetadata(image,
-                                                         'pool_foo',
-                                                         'user_foo',
-                                                         'conf_foo')
-                        rbdio = linuxrbd.RBDVolumeIOWrapper(meta)
-                        rbdio.seek(0)
-                        self.service._backup_rbd(self.backup, rbdio,
-                                                 vol_name, vol_length)
-                        mock_rbd_diff_transfer.assert_called_with(
-                            vol_name, 'pool_foo', backup_name,
-                            self.backup.container, src_user='user_foo',
-                            src_conf='conf_foo',
-                            dest_conf='/etc/ceph/ceph.conf',
-                            dest_user='cinder', src_snap='new_snap',
-                            from_snap='backup.mock.snap.153464362.12')
+        with mock.patch.object(self.service, '_get_backup_base_name') as \
+                mock_get_backup_base_name:
+            with mock.patch.object(self.service, '_rbd_diff_transfer') as \
+                    mock_rbd_diff_transfer:
+                with mock.patch.object(self.service, '_get_new_snap_name') as \
+                        mock_get_new_snap_name:
+                    mock_get_backup_base_name.return_value = base_name
+                    mock_get_new_snap_name.return_value = 'new_snap'
+                    image = self.service.rbd.Image()
+                    meta = linuxrbd.RBDImageMetadata(image, 'pool_foo',
+                                                     'user_foo', 'conf_foo')
+                    rbdio = linuxrbd.RBDVolumeIOWrapper(meta)
+                    rbdio.seek(0)
+                    self.service._backup_rbd(self.alt_backup, rbdio, vol_name,
+                                             vol_length)
+                    mock_rbd_diff_transfer.assert_called_with(
+                        vol_name, 'pool_foo', base_name,
+                        self.backup.container, src_user='user_foo',
+                        src_conf='conf_foo',
+                        dest_conf='/etc/ceph/ceph.conf',
+                        dest_user='cinder', src_snap='new_snap',
+                        from_snap=None)
 
     @common_mocks
     def test_backup_vol_length_0(self):
@@ -848,7 +886,7 @@ class BackupCephTestCase(test.TestCase):
     @common_mocks
     def test_restore(self):
         backup_name = self.service._get_backup_base_name(self.volume_id,
-                                                         diff_format=True)
+                                                         self.alt_backup)
 
         self.mock_rbd.RBD.return_value.list.return_value = [backup_name]
 
@@ -870,7 +908,7 @@ class BackupCephTestCase(test.TestCase):
                 with tempfile.NamedTemporaryFile() as test_file:
                     self.volume_file.seek(0)
 
-                    self.service.restore(self.backup, self.volume_id,
+                    self.service.restore(self.alt_backup, self.volume_id,
                                          test_file)
 
                     checksum = hashlib.sha256()
@@ -965,8 +1003,7 @@ class BackupCephTestCase(test.TestCase):
     @common_mocks
     def test_delete_backup_snapshot(self):
         snap_name = 'backup.%s.snap.3824923.1412' % (uuid.uuid4())
-        base_name = self.service._get_backup_base_name(self.volume_id,
-                                                       diff_format=True)
+        base_name = self.service._get_backup_base_name(self.volume_id)
         self.mock_rbd.RBD.remove_snap = mock.Mock()
         thread_dict = {}
 
@@ -996,16 +1033,16 @@ class BackupCephTestCase(test.TestCase):
     @mock.patch('cinder.backup.drivers.ceph.VolumeMetadataBackup', spec=True)
     def test_try_delete_base_image_diff_format(self, mock_meta_backup):
         backup_name = self.service._get_backup_base_name(self.volume_id,
-                                                         diff_format=True)
+                                                         self.alt_backup)
 
         self.mock_rbd.RBD.return_value.list.return_value = [backup_name]
 
         with mock.patch.object(self.service, '_delete_backup_snapshot') as \
                 mock_del_backup_snap:
-            snap_name = self.service._get_new_snap_name(self.backup_id)
+            snap_name = self.service._get_new_snap_name(self.alt_backup_id)
             mock_del_backup_snap.return_value = (snap_name, 0)
 
-            self.service.delete_backup(self.backup)
+            self.service.delete_backup(self.alt_backup)
             self.assertTrue(mock_del_backup_snap.called)
 
         self.assertTrue(self.mock_rbd.RBD.return_value.list.called)
@@ -1015,7 +1052,7 @@ class BackupCephTestCase(test.TestCase):
     @mock.patch('cinder.backup.drivers.ceph.VolumeMetadataBackup', spec=True)
     def test_try_delete_base_image(self, mock_meta_backup):
         backup_name = self.service._get_backup_base_name(self.volume_id,
-                                                         self.backup_id)
+                                                         self.alt_backup)
         thread_dict = {}
 
         def mock_side_effect(ioctx, base_name):
@@ -1024,7 +1061,7 @@ class BackupCephTestCase(test.TestCase):
         self.mock_rbd.RBD.return_value.list.return_value = [backup_name]
         self.mock_rbd.RBD.return_value.remove.side_effect = mock_side_effect
         with mock.patch.object(self.service, 'get_backup_snaps'):
-            self.service.delete_backup(self.backup)
+            self.service.delete_backup(self.alt_backup)
             self.assertTrue(self.mock_rbd.RBD.return_value.remove.called)
             self.assertNotEqual(threading.current_thread(),
                                 thread_dict['thread'])
@@ -1033,7 +1070,7 @@ class BackupCephTestCase(test.TestCase):
     def test_try_delete_base_image_busy(self):
         """This should induce retries then raise rbd.ImageBusy."""
         backup_name = self.service._get_backup_base_name(self.volume_id,
-                                                         self.backup_id)
+                                                         self.alt_backup)
 
         rbd = self.mock_rbd.RBD.return_value
         rbd.list.return_value = [backup_name]
@@ -1043,7 +1080,7 @@ class BackupCephTestCase(test.TestCase):
                 mock_get_backup_snaps:
             self.assertRaises(self.mock_rbd.ImageBusy,
                               self.service._try_delete_base_image,
-                              self.backup)
+                              self.alt_backup)
             self.assertTrue(mock_get_backup_snaps.called)
 
         self.assertTrue(rbd.list.called)
diff --git a/cinder/tests/unit/objects/test_objects.py b/cinder/tests/unit/objects/test_objects.py
index 6d5b78c6548..3b18be4deb4 100644
--- a/cinder/tests/unit/objects/test_objects.py
+++ b/cinder/tests/unit/objects/test_objects.py
@@ -23,9 +23,9 @@ from cinder import test
 # NOTE: The hashes in this list should only be changed if they come with a
 # corresponding version bump in the affected objects.
 object_data = {
-    'Backup': '1.6-c7ede487ba6fbcdd2a4711343cd972be',
+    'Backup': '1.7-fffdbcd5da3c30750916fa2cc0e8ffb5',
     'BackupDeviceInfo': '1.0-74b3950676c690538f4bc6796bd0042e',
-    'BackupImport': '1.6-c7ede487ba6fbcdd2a4711343cd972be',
+    'BackupImport': '1.7-fffdbcd5da3c30750916fa2cc0e8ffb5',
     'BackupList': '1.0-15ecf022a68ddbb8c2a6739cfc9f8f5e',
     'CleanupRequest': '1.0-e7c688b893e1d5537ccf65cc3eb10a28',
     'Cluster': '1.1-e2c533eb8cdd8d229b6c45c6cf3a9e2c',
diff --git a/releasenotes/notes/support-incremental-backup-completion-in-rbd-1f2165fefcc470d1.yaml b/releasenotes/notes/support-incremental-backup-completion-in-rbd-1f2165fefcc470d1.yaml
new file mode 100644
index 00000000000..d8be52f1cc4
--- /dev/null
+++ b/releasenotes/notes/support-incremental-backup-completion-in-rbd-1f2165fefcc470d1.yaml
@@ -0,0 +1,5 @@
+---
+fixes:
+  - Fixed issue where all Ceph RBD backups would be incremental after the
+    first one. The driver now honors whether ``--incremental`` is specified or
+    not.