diff --git a/ironic/drivers/modules/drac/job.py b/ironic/drivers/modules/drac/job.py
index fd399d85d9..cab1af32e2 100644
--- a/ironic/drivers/modules/drac/job.py
+++ b/ironic/drivers/modules/drac/job.py
@@ -27,20 +27,26 @@ drac_exceptions = importutils.try_import('dracclient.exceptions')
 LOG = logging.getLogger(__name__)
 
 
-def validate_job_queue(node):
+def validate_job_queue(node, name_prefix=None):
     """Validates the job queue on the node.
 
     It raises an exception if an unfinished configuration job exists.
 
     :param node: an ironic node object.
+    :param name_prefix: A name prefix for jobs to validate.
     :raises: DracOperationError on an error from python-dracclient.
     """
 
     unfinished_jobs = list_unfinished_jobs(node)
-    if unfinished_jobs:
-        msg = _('Unfinished config jobs found: %(jobs)r. Make sure they are '
-                'completed before retrying.') % {'jobs': unfinished_jobs}
-        raise exception.DracOperationError(error=msg)
+    if name_prefix is not None:
+        # Filter out jobs that don't match the name prefix.
+        unfinished_jobs = [job for job in unfinished_jobs
+                           if job.name.startswith(name_prefix)]
+    if not unfinished_jobs:
+        return
+    msg = _('Unfinished config jobs found: %(jobs)r. Make sure they are '
+            'completed before retrying.') % {'jobs': unfinished_jobs}
+    raise exception.DracOperationError(error=msg)
 
 
 def get_job(node, job_id):
diff --git a/ironic/drivers/modules/drac/raid.py b/ironic/drivers/modules/drac/raid.py
index de5dd6af71..4ed40e62e6 100644
--- a/ironic/drivers/modules/drac/raid.py
+++ b/ironic/drivers/modules/drac/raid.py
@@ -26,11 +26,11 @@ from oslo_utils import units
 from ironic.common import exception
 from ironic.common.i18n import _
 from ironic.common import raid as raid_common
-from ironic.common import states
 from ironic.conductor import task_manager
 from ironic.conductor import utils as manager_utils
 from ironic.conf import CONF
 from ironic.drivers import base
+from ironic.drivers.modules import deploy_utils
 from ironic.drivers.modules.drac import common as drac_common
 from ironic.drivers.modules.drac import job as drac_job
 
@@ -163,6 +163,18 @@ def _is_raid_controller(node, raid_controller_fqdd, raid_controllers=None):
         raise exception.DracOperationError(error=exc)
 
 
+def _validate_job_queue(node, raid_controller=None):
+    """Validate that there are no pending jobs for this controller.
+
+    :param node: an ironic node object.
+    :param raid_controller: id of the RAID controller.
+    """
+    kwargs = {}
+    if raid_controller:
+        kwargs["name_prefix"] = "Config:RAID:%s" % raid_controller
+    drac_job.validate_job_queue(node, **kwargs)
+
+
 def create_virtual_disk(node, raid_controller, physical_disks, raid_level,
                         size_mb, disk_name=None, span_length=None,
                         span_depth=None):
@@ -185,7 +197,9 @@ def create_virtual_disk(node, raid_controller, physical_disks, raid_level,
               values to be applied.
     :raises: DracOperationError on an error from python-dracclient.
     """
-    drac_job.validate_job_queue(node)
+    # This causes config to fail, because the boot mode is set via a config
+    # job.
+    _validate_job_queue(node, raid_controller)
 
     client = drac_common.get_drac_client(node)
 
@@ -215,7 +229,8 @@ def delete_virtual_disk(node, virtual_disk):
               values to be applied.
     :raises: DracOperationError on an error from python-dracclient.
     """
-    drac_job.validate_job_queue(node)
+    # NOTE(mgoddard): Cannot specify raid_controller as we don't know it.
+    _validate_job_queue(node)
 
     client = drac_common.get_drac_client(node)
 
@@ -247,7 +262,7 @@ def _reset_raid_config(node, raid_controller):
     """
     try:
 
-        drac_job.validate_job_queue(node)
+        _validate_job_queue(node, raid_controller)
 
         client = drac_common.get_drac_client(node)
         return client.reset_raid_config(raid_controller)
@@ -279,7 +294,7 @@ def clear_foreign_config(node, raid_controller):
     """
     try:
 
-        drac_job.validate_job_queue(node)
+        _validate_job_queue(node, raid_controller)
 
         client = drac_common.get_drac_client(node)
         return client.clear_foreign_config(raid_controller)
@@ -480,13 +495,18 @@ def _volume_usage_per_disk_mb(logical_disk, physical_disks, spans_count=1,
     return int(stripes_per_disk * stripe_size_kb / units.Ki)
 
 
-def _find_configuration(logical_disks, physical_disks):
+def _find_configuration(logical_disks, physical_disks, pending_delete):
     """Find RAID configuration.
 
     This method transforms the RAID configuration defined in Ironic to a format
     that is required by dracclient. This includes matching the physical disks
     to RAID volumes when it's not pre-defined, or in general calculating
     missing properties.
+
+    :param logical_disks: list of logical disk definitions.
+    :param physical_disks: list of physical disk definitions.
+    :param pending_delete: Whether there is a pending deletion of virtual
+        disks that should be accounted for.
     """
 
     # shared physical disks of RAID volumes size_gb='MAX' should be
@@ -508,7 +528,7 @@ def _find_configuration(logical_disks, physical_disks):
     free_space_mb = {}
     for disk in physical_disks:
         # calculate free disk space
-        free_space_mb[disk] = disk.free_size_mb
+        free_space_mb[disk] = _get_disk_free_size_mb(disk, pending_delete)
 
         disk_type = (disk.controller, disk.media_type, disk.interface_type,
                      disk.size_mb)
@@ -550,7 +570,8 @@ def _find_configuration(logical_disks, physical_disks):
     if volumes_without_disks:
         result, free_space_mb = (
             _assign_disks_to_volume(volumes_without_disks,
-                                    physical_disks_by_type, free_space_mb))
+                                    physical_disks_by_type, free_space_mb,
+                                    pending_delete))
         if not result:
             # try again using the reserved physical disks in addition
             for disk_type, disks in physical_disks_by_type.items():
@@ -560,7 +581,8 @@ def _find_configuration(logical_disks, physical_disks):
             result, free_space_mb = (
                 _assign_disks_to_volume(volumes_without_disks,
                                         physical_disks_by_type,
-                                        free_space_mb))
+                                        free_space_mb,
+                                        pending_delete))
             if not result:
                 error_msg = _('failed to find matching physical disks for all '
                               'logical disks')
@@ -636,7 +658,7 @@ def _calculate_volume_props(logical_disk, physical_disks, free_space_mb):
 
 
 def _assign_disks_to_volume(logical_disks, physical_disks_by_type,
-                            free_space_mb):
+                            free_space_mb, pending_delete):
     logical_disk = logical_disks.pop(0)
     raid_level = logical_disk['raid_level']
 
@@ -664,8 +686,12 @@ def _assign_disks_to_volume(logical_disks, physical_disks_by_type,
         # filter out disks already in use if sharing is disabled
         if ('share_physical_disks' not in logical_disk
                 or not logical_disk['share_physical_disks']):
+            initial_free_size_mb = {
+                disk: _get_disk_free_size_mb(disk, pending_delete)
+                for disk in disks
+            }
             disks = [disk for disk in disks
-                     if disk.free_size_mb == free_space_mb[disk]]
+                     if initial_free_size_mb[disk] == free_space_mb[disk]]
 
         max_spans = _calculate_spans(raid_level, len(disks))
         min_spans = min([2, max_spans])
@@ -701,7 +727,8 @@ def _assign_disks_to_volume(logical_disks, physical_disks_by_type,
                 result, candidate_free_space_mb = (
                     _assign_disks_to_volume(logical_disks,
                                             physical_disks_by_type,
-                                            candidate_free_space_mb))
+                                            candidate_free_space_mb,
+                                            pending_delete))
                 if result:
                     logical_disks.append(candidate_volume)
                     return (True, candidate_free_space_mb)
@@ -742,11 +769,12 @@ def _commit_to_controllers(node, controllers, substep="completed"):
                         enumerated value indicating whether the server must
                         be rebooted only if raid controller does not support
                         realtime.
-    :param substep: contain sub cleaning step which executes any raid
-                    configuration job if set after cleaning step.
+    :param substep: contain sub cleaning or deploy step which executes any raid
+                    configuration job if set after cleaning or deploy step.
                     (default to completed)
-    :returns: states.CLEANWAIT if deletion is in progress asynchronously
-              or None if it is completed.
+    :returns: states.CLEANWAIT (cleaning) or states.DEPLOYWAIT (deployment) if
+              configuration is in progress asynchronously or None if it is
+              completed.
     """
     if not controllers:
         LOG.debug('No changes on any of the controllers on node %s',
@@ -796,9 +824,28 @@ def _commit_to_controllers(node, controllers, substep="completed"):
                 raid_controller)
 
     node.driver_internal_info = driver_internal_info
-    node.save()
 
-    return states.CLEANWAIT
+    # Signal whether the node has been rebooted, that we do not need to execute
+    # the step again, and that this completion of this step is triggered
+    # through async polling.
+    # NOTE(mgoddard): set_async_step_flags calls node.save().
+    deploy_utils.set_async_step_flags(
+        node,
+        reboot=not all_realtime,
+        skip_current_step=True,
+        polling=True)
+
+    return deploy_utils.get_async_step_return_state(node)
+
+
+def _get_disk_free_size_mb(disk, pending_delete):
+    """Return the size of free space on the disk in MB.
+
+    :param disk: a PhysicalDisk object.
+    :param pending_delete: Whether there is a pending deletion of all virtual
+        disks.
+    """
+    return disk.size_mb if pending_delete else disk.free_size_mb
 
 
 class DracWSManRAID(base.RAIDInterface):
@@ -807,6 +854,16 @@ class DracWSManRAID(base.RAIDInterface):
         """Return the properties of the interface."""
         return drac_common.COMMON_PROPERTIES
 
+    @base.deploy_step(priority=0,
+                      argsinfo=base.RAID_APPLY_CONFIGURATION_ARGSINFO)
+    def apply_configuration(self, task, raid_config, create_root_volume=True,
+                            create_nonroot_volumes=False,
+                            delete_existing=True):
+        return super(DracRAID, self).apply_configuration(
+            task, raid_config, create_root_volume=create_root_volume,
+            create_nonroot_volumes=create_nonroot_volumes,
+            delete_existing=delete_existing)
+
     @METRICS.timer('DracRAID.create_configuration')
     @base.clean_step(priority=0, abortable=False, argsinfo={
         'create_root_volume': {
@@ -822,11 +879,20 @@ class DracWSManRAID(base.RAIDInterface):
                 'Defaults to `True`.'
             ),
             'required': False
+        },
+        "delete_existing": {
+            "description": (
+                "Setting this to 'True' indicates to delete existing RAID "
+                "configuration prior to creating the new configuration. "
+                "Default value is 'False'."
+            ),
+            "required": False,
         }
     })
     def create_configuration(self, task,
                              create_root_volume=True,
-                             create_nonroot_volumes=True):
+                             create_nonroot_volumes=True,
+                             delete_existing=False):
         """Create the RAID configuration.
 
         This method creates the RAID configuration on the given node.
@@ -838,8 +904,12 @@ class DracWSManRAID(base.RAIDInterface):
         :param create_nonroot_volumes: If True, non-root volumes are
             created. If False, no non-root volumes are created. Default
             is True.
-        :returns: states.CLEANWAIT if creation is in progress asynchronously
-                  or None if it is completed.
+        :param delete_existing: Setting this to True indicates to delete RAID
+            configuration prior to creating the new configuration. Default is
+            False.
+        :returns: states.CLEANWAIT (cleaning) or states.DEPLOYWAIT (deployment)
+            if creation is in progress asynchronously or None if it is
+            completed.
         :raises: MissingParameterValue, if node.target_raid_config is missing
             or empty.
         :raises: DracOperationError on an error from python-dracclient.
@@ -864,13 +934,18 @@ class DracWSManRAID(base.RAIDInterface):
 
             del disk['size_gb']
 
+        if delete_existing:
+            controllers = self._delete_configuration_no_commit(task)
+        else:
+            controllers = list()
+
         physical_disks = list_physical_disks(node)
-        logical_disks = _find_configuration(logical_disks, physical_disks)
+        logical_disks = _find_configuration(logical_disks, physical_disks,
+                                            pending_delete=delete_existing)
 
         logical_disks_to_create = _filter_logical_disks(
             logical_disks, create_root_volume, create_nonroot_volumes)
 
-        controllers = list()
         for logical_disk in logical_disks_to_create:
             controller = dict()
             controller_cap = create_virtual_disk(
@@ -892,28 +967,19 @@ class DracWSManRAID(base.RAIDInterface):
 
     @METRICS.timer('DracRAID.delete_configuration')
     @base.clean_step(priority=0)
+    @base.deploy_step(priority=0)
     def delete_configuration(self, task):
         """Delete the RAID configuration.
 
         :param task: a TaskManager instance containing the node to act on.
-        :returns: states.CLEANWAIT if deletion is in progress asynchronously
-                  or None if it is completed.
+        :returns: states.CLEANWAIT (cleaning) or states.DEPLOYWAIT (deployment)
+            if deletion is in progress asynchronously or None if it is
+            completed.
         :raises: DracOperationError on an error from python-dracclient.
         """
-        node = task.node
 
-        controllers = list()
-        drac_raid_controllers = list_raid_controllers(node)
-        for cntrl in drac_raid_controllers:
-            if _is_raid_controller(node, cntrl.id, drac_raid_controllers):
-                controller = dict()
-                controller_cap = _reset_raid_config(node, cntrl.id)
-                controller["raid_controller"] = cntrl.id
-                controller["is_reboot_required"] = controller_cap[
-                    "is_reboot_required"]
-                controllers.append(controller)
-
-        return _commit_to_controllers(node, controllers,
+        controllers = self._delete_configuration_no_commit(task)
+        return _commit_to_controllers(task.node, controllers,
                                       substep="delete_foreign_config")
 
     @METRICS.timer('DracRAID.get_logical_disks')
@@ -1005,18 +1071,18 @@ class DracWSManRAID(base.RAIDInterface):
             if 'raid_config_substep' in node.driver_internal_info:
                 if node.driver_internal_info['raid_config_substep'] == \
                         'delete_foreign_config':
-                    self._execute_cleaning_foreign_drives(task, node)
+                    self._execute_foreign_drives(task, node)
                 elif node.driver_internal_info['raid_config_substep'] == \
                         'completed':
-                    self._complete_raid_cleaning_substep(task, node)
+                    self._complete_raid_substep(task, node)
             else:
-                self._complete_raid_cleaning_substep(task, node)
+                self._complete_raid_substep(task, node)
         else:
             self._clear_raid_substep(node)
             self._clear_raid_config_job_failure(node)
-            self._set_clean_failed(task, config_job)
+            self._set_failed(task, config_job)
 
-    def _execute_cleaning_foreign_drives(self, task, node):
+    def _execute_foreign_drives(self, task, node):
         controllers = list()
         jobs_required = False
         for controller_id in node.driver_internal_info[
@@ -1034,17 +1100,17 @@ class DracWSManRAID(base.RAIDInterface):
         if not jobs_required:
             LOG.info(
                 "No foreign drives detected, so "
-                "resume cleaning")
-            self._complete_raid_cleaning_substep(task, node)
+                "resume %s", "cleaning" if node.clean_step else "deployment")
+            self._complete_raid_substep(task, node)
         else:
             _commit_to_controllers(
                 node,
                 controllers,
                 substep='completed')
 
-    def _complete_raid_cleaning_substep(self, task, node):
+    def _complete_raid_substep(self, task, node):
         self._clear_raid_substep(node)
-        self._resume_cleaning(task)
+        self._resume(task)
 
     def _clear_raid_substep(self, node):
         driver_internal_info = node.driver_internal_info
@@ -1076,7 +1142,7 @@ class DracWSManRAID(base.RAIDInterface):
         node.driver_internal_info = driver_internal_info
         node.save()
 
-    def _set_clean_failed(self, task, config_job):
+    def _set_failed(self, task, config_job):
         LOG.error("RAID configuration job failed for node %(node)s. "
                   "Failed config job: %(config_job_id)s. "
                   "Message: '%(message)s'.",
@@ -1085,14 +1151,33 @@ class DracWSManRAID(base.RAIDInterface):
         task.node.last_error = config_job.message
         task.process_event('fail')
 
-    def _resume_cleaning(self, task):
+    def _resume(self, task):
         raid_common.update_raid_info(
             task.node, self.get_logical_disks(task))
-        driver_internal_info = task.node.driver_internal_info
-        driver_internal_info['cleaning_reboot'] = True
-        task.node.driver_internal_info = driver_internal_info
-        task.node.save()
-        manager_utils.notify_conductor_resume_clean(task)
+        if task.node.clean_step:
+            manager_utils.notify_conductor_resume_clean(task)
+        else:
+            manager_utils.notify_conductor_resume_deploy(task)
+
+    def _delete_configuration_no_commit(self, task):
+        """Delete existing RAID configuration without committing the change.
+
+        :param task: A TaskManager instance.
+        :returns: A set of names of RAID controllers which need RAID changes to
+            be committed.
+        """
+        node = task.node
+        controllers = list()
+        drac_raid_controllers = list_raid_controllers(node)
+        for cntrl in drac_raid_controllers:
+            if _is_raid_controller(node, cntrl.id, drac_raid_controllers):
+                controller = dict()
+                controller_cap = _reset_raid_config(node, cntrl.id)
+                controller["raid_controller"] = cntrl.id
+                controller["is_reboot_required"] = controller_cap[
+                    "is_reboot_required"]
+                controllers.append(controller)
+        return controllers
 
 
 class DracRAID(DracWSManRAID):
diff --git a/ironic/tests/unit/drivers/modules/drac/test_job.py b/ironic/tests/unit/drivers/modules/drac/test_job.py
index 195fc97602..5a68c8d5e3 100644
--- a/ironic/tests/unit/drivers/modules/drac/test_job.py
+++ b/ironic/tests/unit/drivers/modules/drac/test_job.py
@@ -111,6 +111,25 @@ class DracJobTestCase(test_utils.BaseDracTest):
         self.assertRaises(exception.DracOperationError,
                           drac_job.validate_job_queue, self.node)
 
+    def test_validate_job_queue_name_prefix(self, mock_get_drac_client):
+        mock_client = mock.Mock()
+        mock_get_drac_client.return_value = mock_client
+        mock_client.list_jobs.return_value = [self.job]
+
+        drac_job.validate_job_queue(self.node, name_prefix='Fake')
+
+        mock_client.list_jobs.assert_called_once_with(only_unfinished=True)
+
+    def test_validate_job_queue_name_prefix_invalid(self,
+                                                    mock_get_drac_client):
+        mock_client = mock.Mock()
+        mock_get_drac_client.return_value = mock_client
+        mock_client.list_jobs.return_value = [self.job]
+
+        self.assertRaises(exception.DracOperationError,
+                          drac_job.validate_job_queue, self.node,
+                          name_prefix='ConfigBIOS')
+
 
 @mock.patch.object(drac_common, 'get_drac_client', spec_set=True,
                    autospec=True)
diff --git a/ironic/tests/unit/drivers/modules/drac/test_periodic_task.py b/ironic/tests/unit/drivers/modules/drac/test_periodic_task.py
index e2546bb8db..fb0f53f51c 100644
--- a/ironic/tests/unit/drivers/modules/drac/test_periodic_task.py
+++ b/ironic/tests/unit/drivers/modules/drac/test_periodic_task.py
@@ -139,9 +139,8 @@ class DracPeriodicTaskTestCase(db_base.DbTestCase):
                        autospec=True)
     @mock.patch.object(drac_raid.DracRAID, 'get_logical_disks',
                        spec_set=True, autospec=True)
-    @mock.patch.object(manager_utils, 'notify_conductor_resume_clean')
-    def test__check_node_raid_jobs_with_completed_job(
-            self, mock_notify_conductor_resume_clean,
+    def _test__check_node_raid_jobs_with_completed_job(
+            self, mock_notify_conductor_resume,
             mock_get_logical_disks, mock_get_drac_client):
         expected_logical_disk = {'size_gb': 558,
                                  'raid_level': '1',
@@ -171,14 +170,29 @@ class DracPeriodicTaskTestCase(db_base.DbTestCase):
                          self.node.driver_internal_info['raid_config_job_ids'])
         self.assertEqual([expected_logical_disk],
                          self.node.raid_config['logical_disks'])
-        mock_notify_conductor_resume_clean.assert_called_once_with(task)
+        mock_notify_conductor_resume.assert_called_once_with(task)
+
+    @mock.patch.object(manager_utils, 'notify_conductor_resume_clean')
+    def test__check_node_raid_jobs_with_completed_job_in_clean(
+            self, mock_notify_conductor_resume):
+        self.node.clean_step = {'foo': 'bar'}
+        self.node.save()
+        self._test__check_node_raid_jobs_with_completed_job(
+            mock_notify_conductor_resume)
+
+    @mock.patch.object(manager_utils, 'notify_conductor_resume_deploy')
+    def test__check_node_raid_jobs_with_completed_job_in_deploy(
+            self, mock_notify_conductor_resume):
+        self._test__check_node_raid_jobs_with_completed_job(
+            mock_notify_conductor_resume)
 
     @mock.patch.object(drac_common, 'get_drac_client', spec_set=True,
                        autospec=True)
     def test__check_node_raid_jobs_with_failed_job(self, mock_get_drac_client):
-        # mock node.driver_internal_info
+        # mock node.driver_internal_info and node.clean_step
         driver_internal_info = {'raid_config_job_ids': ['42']}
         self.node.driver_internal_info = driver_internal_info
+        self.node.clean_step = {'foo': 'bar'}
         self.node.save()
         # mock task
         task = mock.Mock(node=self.node, context=self.context)
@@ -207,9 +221,8 @@ class DracPeriodicTaskTestCase(db_base.DbTestCase):
                        autospec=True)
     @mock.patch.object(drac_raid.DracRAID, 'get_logical_disks',
                        spec_set=True, autospec=True)
-    @mock.patch.object(manager_utils, 'notify_conductor_resume_clean')
-    def test__check_node_raid_jobs_with_completed_job_already_failed(
-            self, mock_notify_conductor_resume_clean,
+    def _test__check_node_raid_jobs_with_completed_job_already_failed(
+            self, mock_notify_conductor_resume,
             mock_get_logical_disks, mock_get_drac_client):
         expected_logical_disk = {'size_gb': 558,
                                  'raid_level': '1',
@@ -242,14 +255,28 @@ class DracPeriodicTaskTestCase(db_base.DbTestCase):
                          self.node.driver_internal_info)
         self.assertNotIn('logical_disks', self.node.raid_config)
         task.process_event.assert_called_once_with('fail')
+        self.assertFalse(mock_notify_conductor_resume.called)
+
+    @mock.patch.object(manager_utils, 'notify_conductor_resume_clean')
+    def test__check_node_raid_jobs_with_completed_job_already_failed_in_clean(
+            self, mock_notify_conductor_resume):
+        self.node.clean_step = {'foo': 'bar'}
+        self.node.save()
+        self._test__check_node_raid_jobs_with_completed_job_already_failed(
+            mock_notify_conductor_resume)
+
+    @mock.patch.object(manager_utils, 'notify_conductor_resume_deploy')
+    def test__check_node_raid_jobs_with_completed_job_already_failed_in_deploy(
+            self, mock_notify_conductor_resume):
+        self._test__check_node_raid_jobs_with_completed_job_already_failed(
+            mock_notify_conductor_resume)
 
     @mock.patch.object(drac_common, 'get_drac_client', spec_set=True,
                        autospec=True)
     @mock.patch.object(drac_raid.DracRAID, 'get_logical_disks',
                        spec_set=True, autospec=True)
-    @mock.patch.object(manager_utils, 'notify_conductor_resume_clean')
-    def test__check_node_raid_jobs_with_multiple_jobs_completed(
-            self, mock_notify_conductor_resume_clean,
+    def _test__check_node_raid_jobs_with_multiple_jobs_completed(
+            self, mock_notify_conductor_resume,
             mock_get_logical_disks, mock_get_drac_client):
         expected_logical_disk = {'size_gb': 558,
                                  'raid_level': '1',
@@ -282,15 +309,28 @@ class DracPeriodicTaskTestCase(db_base.DbTestCase):
                          self.node.driver_internal_info)
         self.assertEqual([expected_logical_disk],
                          self.node.raid_config['logical_disks'])
-        mock_notify_conductor_resume_clean.assert_called_once_with(task)
+        mock_notify_conductor_resume.assert_called_once_with(task)
+
+    @mock.patch.object(manager_utils, 'notify_conductor_resume_clean')
+    def test__check_node_raid_jobs_with_multiple_jobs_completed_in_clean(
+            self, mock_notify_conductor_resume):
+        self.node.clean_step = {'foo': 'bar'}
+        self.node.save()
+        self._test__check_node_raid_jobs_with_multiple_jobs_completed(
+            mock_notify_conductor_resume)
+
+    @mock.patch.object(manager_utils, 'notify_conductor_resume_deploy')
+    def test__check_node_raid_jobs_with_multiple_jobs_completed_in_deploy(
+            self, mock_notify_conductor_resume):
+        self._test__check_node_raid_jobs_with_multiple_jobs_completed(
+            mock_notify_conductor_resume)
 
     @mock.patch.object(drac_common, 'get_drac_client', spec_set=True,
                        autospec=True)
     @mock.patch.object(drac_raid.DracRAID, 'get_logical_disks',
                        spec_set=True, autospec=True)
-    @mock.patch.object(manager_utils, 'notify_conductor_resume_clean')
-    def test__check_node_raid_jobs_with_multiple_jobs_failed(
-            self, mock_notify_conductor_resume_clean,
+    def _test__check_node_raid_jobs_with_multiple_jobs_failed(
+            self, mock_notify_conductor_resume,
             mock_get_logical_disks, mock_get_drac_client):
         expected_logical_disk = {'size_gb': 558,
                                  'raid_level': '1',
@@ -327,3 +367,18 @@ class DracPeriodicTaskTestCase(db_base.DbTestCase):
                          self.node.driver_internal_info)
         self.assertNotIn('logical_disks', self.node.raid_config)
         task.process_event.assert_called_once_with('fail')
+        self.assertFalse(mock_notify_conductor_resume.called)
+
+    @mock.patch.object(manager_utils, 'notify_conductor_resume_clean')
+    def test__check_node_raid_jobs_with_multiple_jobs_failed_in_clean(
+            self, mock_notify_conductor_resume):
+        self.node.clean_step = {'foo': 'bar'}
+        self.node.save()
+        self._test__check_node_raid_jobs_with_multiple_jobs_failed(
+            mock_notify_conductor_resume)
+
+    @mock.patch.object(manager_utils, 'notify_conductor_resume_deploy')
+    def test__check_node_raid_jobs_with_multiple_jobs_failed_in_deploy(
+            self, mock_notify_conductor_resume):
+        self._test__check_node_raid_jobs_with_multiple_jobs_failed(
+            mock_notify_conductor_resume)
diff --git a/ironic/tests/unit/drivers/modules/drac/test_raid.py b/ironic/tests/unit/drivers/modules/drac/test_raid.py
index 387e3c0c74..f609858d53 100644
--- a/ironic/tests/unit/drivers/modules/drac/test_raid.py
+++ b/ironic/tests/unit/drivers/modules/drac/test_raid.py
@@ -166,7 +166,8 @@ class DracManageVirtualDisksTestCase(test_utils.BaseDracTest):
         drac_raid.create_virtual_disk(
             self.node, 'controller', ['disk1', 'disk2'], '1+0', 43008)
 
-        mock_validate_job_queue.assert_called_once_with(self.node)
+        mock_validate_job_queue.assert_called_once_with(
+            self.node, name_prefix='Config:RAID:controller')
         mock_client.create_virtual_disk.assert_called_once_with(
             'controller', ['disk1', 'disk2'], '1+0', 43008, None, None, None)
 
@@ -182,7 +183,8 @@ class DracManageVirtualDisksTestCase(test_utils.BaseDracTest):
             self.node, 'controller', ['disk1', 'disk2'], '1+0', 43008,
             disk_name='name', span_length=3, span_depth=2)
 
-        mock_validate_job_queue.assert_called_once_with(self.node)
+        mock_validate_job_queue.assert_called_once_with(
+            self.node, name_prefix='Config:RAID:controller')
         mock_client.create_virtual_disk.assert_called_once_with(
             'controller', ['disk1', 'disk2'], '1+0', 43008, 'name', 3, 2)
 
@@ -234,7 +236,8 @@ class DracManageVirtualDisksTestCase(test_utils.BaseDracTest):
         drac_raid._reset_raid_config(
             self.node, 'controller')
 
-        mock_validate_job_queue.assert_called_once_with(self.node)
+        mock_validate_job_queue.assert_called_once_with(
+            self.node, name_prefix='Config:RAID:controller')
         mock_client.reset_raid_config.assert_called_once_with(
             'controller')
 
@@ -262,7 +265,8 @@ class DracManageVirtualDisksTestCase(test_utils.BaseDracTest):
         drac_raid.clear_foreign_config(
             self.node, 'RAID.Integrated.1-1')
 
-        mock_validate_job_queue.assert_called_once_with(self.node)
+        mock_validate_job_queue.assert_called_once_with(
+            self.node, 'Config:RAID:RAID.Integrated.1-1')
         mock_client.clear_foreign_config.assert_called_once_with(
             'RAID.Integrated.1-1')
 
@@ -499,7 +503,8 @@ class DracCreateRaidConfigurationHelpersTestCase(test_utils.BaseDracTest):
             'Disk.Bay.2:Enclosure.Internal.0-1:RAID.Integrated.1-1']
 
         logical_disks = drac_raid._find_configuration(logical_disks,
-                                                      physical_disks)
+                                                      physical_disks,
+                                                      False)
 
         self.assertEqual(expected_contoller,
                          logical_disks[0]['controller'])
@@ -525,7 +530,8 @@ class DracCreateRaidConfigurationHelpersTestCase(test_utils.BaseDracTest):
             'Disk.Bay.6:Enclosure.Internal.0-1:RAID.Integrated.1-1']
 
         logical_disks = drac_raid._find_configuration(logical_disks,
-                                                      physical_disks)
+                                                      physical_disks,
+                                                      False)
 
         self.assertEqual(expected_contoller,
                          logical_disks[0]['controller'])
@@ -553,7 +559,8 @@ class DracCreateRaidConfigurationHelpersTestCase(test_utils.BaseDracTest):
         physical_disks = self._generate_physical_disks()
 
         logical_disks = drac_raid._find_configuration(logical_disks,
-                                                      physical_disks)
+                                                      physical_disks,
+                                                      False)
 
         self.assertEqual(3, len(logical_disks))
         # step 1
@@ -591,6 +598,33 @@ class DracCreateRaidConfigurationHelpersTestCase(test_utils.BaseDracTest):
                  'Disk.Bay.3:Enclosure.Internal.0-1:RAID.Integrated.1-1']},
             logical_disks)
 
+    def test__find_configuration_pending_delete(self):
+        logical_disks = [
+            {'size_mb': 102400,
+             'raid_level': '5',
+             'is_root_volume': True,
+             'disk_type': 'hdd'}
+        ]
+        physical_disks = self._generate_physical_disks()
+        # No free space, but deletion pending means they're still usable.
+        physical_disks = [disk._replace(free_size_mb=0)
+                          for disk in physical_disks]
+
+        expected_contoller = 'RAID.Integrated.1-1'
+        expected_physical_disk_ids = [
+            'Disk.Bay.0:Enclosure.Internal.0-1:RAID.Integrated.1-1',
+            'Disk.Bay.1:Enclosure.Internal.0-1:RAID.Integrated.1-1',
+            'Disk.Bay.2:Enclosure.Internal.0-1:RAID.Integrated.1-1']
+
+        logical_disks = drac_raid._find_configuration(logical_disks,
+                                                      physical_disks,
+                                                      True)
+
+        self.assertEqual(expected_contoller,
+                         logical_disks[0]['controller'])
+        self.assertEqual(expected_physical_disk_ids,
+                         logical_disks[0]['physical_disks'])
+
 
 class DracRaidInterfaceTestCase(test_utils.BaseDracTest):
 
@@ -652,6 +686,7 @@ class DracRaidInterfaceTestCase(test_utils.BaseDracTest):
             [self.root_logical_disk] + self.nonroot_logical_disks)
         self.target_raid_configuration = {'logical_disks': self.logical_disks}
         self.node.target_raid_config = self.target_raid_configuration
+        self.node.clean_step = {'foo': 'bar'}
         self.node.save()
 
     def _generate_physical_disks(self):
@@ -664,26 +699,43 @@ class DracRaidInterfaceTestCase(test_utils.BaseDracTest):
 
     @mock.patch.object(drac_common, 'get_drac_client', spec_set=True,
                        autospec=True)
+    @mock.patch.object(drac_raid, '_reset_raid_config', autospec=True)
     @mock.patch.object(drac_raid, 'list_physical_disks', autospec=True)
     @mock.patch.object(drac_job, 'validate_job_queue', spec_set=True,
                        autospec=True)
     @mock.patch.object(drac_raid, 'commit_config', spec_set=True,
                        autospec=True)
-    def test_create_configuration(
-            self, mock_commit_config, mock_validate_job_queue,
-            mock_list_physical_disks, mock_get_drac_client):
+    def _test_create_configuration(
+            self, expected_state, mock_commit_config, mock_validate_job_queue,
+            mock_list_physical_disks, mock__reset_raid_config,
+            mock_get_drac_client):
         mock_client = mock.Mock()
         mock_get_drac_client.return_value = mock_client
         physical_disks = self._generate_physical_disks()
         mock_list_physical_disks.return_value = physical_disks
         mock_commit_config.return_value = '42'
+        raid_controller_dict = {
+            'id': 'RAID.Integrated.1-1',
+            'description': 'Integrated RAID Controller 1',
+            'manufacturer': 'DELL',
+            'model': 'PERC H710 Mini',
+            'primary_status': 'ok',
+            'firmware_version': '21.3.0-0009',
+            'bus': '1',
+            'supports_realtime': True}
+        raid_controller = test_utils.make_raid_controller(
+            raid_controller_dict)
+        mock_client.list_raid_controllers.return_value = [raid_controller]
+        mock__reset_raid_config.return_value = {
+            'is_reboot_required': constants.RebootRequired.optional,
+            'is_commit_required': True}
         mock_client.create_virtual_disk.return_value = {
             'is_reboot_required': constants.RebootRequired.optional,
             'is_commit_required': True}
 
         with task_manager.acquire(self.context, self.node.uuid,
                                   shared=False) as task:
-            task.driver.raid.create_configuration(
+            return_value = task.driver.raid.create_configuration(
                 task, create_root_volume=True, create_nonroot_volumes=False)
 
             mock_client.create_virtual_disk.assert_called_once_with(
@@ -696,10 +748,19 @@ class DracRaidInterfaceTestCase(test_utils.BaseDracTest):
                 task.node, raid_controller='RAID.Integrated.1-1', reboot=False,
                 realtime=True)
 
+        self.assertEqual(expected_state, return_value)
         self.node.refresh()
         self.assertEqual(['42'],
                          self.node.driver_internal_info['raid_config_job_ids'])
 
+    def test_create_configuration_in_clean(self):
+        self._test_create_configuration(states.CLEANWAIT)
+
+    def test_create_configuration_in_deploy(self):
+        self.node.clean_step = None
+        self.node.save()
+        self._test_create_configuration(states.DEPLOYWAIT)
+
     @mock.patch.object(drac_common, 'get_drac_client', spec_set=True,
                        autospec=True)
     @mock.patch.object(drac_raid, 'list_physical_disks', autospec=True)
@@ -709,7 +770,8 @@ class DracRaidInterfaceTestCase(test_utils.BaseDracTest):
                        autospec=True)
     def test_create_configuration_no_change(
             self, mock_commit_config, mock_validate_job_queue,
-            mock_list_physical_disks, mock_get_drac_client):
+            mock_list_physical_disks,
+            mock_get_drac_client):
         mock_client = mock.Mock()
         mock_get_drac_client.return_value = mock_client
         physical_disks = self._generate_physical_disks()
@@ -721,7 +783,8 @@ class DracRaidInterfaceTestCase(test_utils.BaseDracTest):
         with task_manager.acquire(self.context, self.node.uuid,
                                   shared=False) as task:
             return_value = task.driver.raid.create_configuration(
-                task, create_root_volume=False, create_nonroot_volumes=False)
+                task, create_root_volume=False, create_nonroot_volumes=False,
+                delete_existing=False)
 
             self.assertEqual(0, mock_client.create_virtual_disk.call_count)
             self.assertEqual(0, mock_commit_config.call_count)
@@ -731,6 +794,72 @@ class DracRaidInterfaceTestCase(test_utils.BaseDracTest):
         self.node.refresh()
         self.assertNotIn('raid_config_job_ids', self.node.driver_internal_info)
 
+    @mock.patch.object(drac_common, 'get_drac_client', spec_set=True,
+                       autospec=True)
+    @mock.patch.object(drac_raid, '_reset_raid_config', autospec=True)
+    @mock.patch.object(drac_raid, 'list_virtual_disks', autospec=True)
+    @mock.patch.object(drac_raid, 'list_physical_disks', autospec=True)
+    @mock.patch.object(drac_job, 'validate_job_queue', spec_set=True,
+                       autospec=True)
+    @mock.patch.object(drac_raid, 'commit_config', spec_set=True,
+                       autospec=True)
+    def test_create_configuration_delete_existing(
+            self, mock_commit_config,
+            mock_validate_job_queue,
+            mock_list_physical_disks,
+            mock_list_virtual_disks,
+            mock__reset_raid_config,
+            mock_get_drac_client):
+        self.node.clean_step = None
+        self.node.save()
+        mock_client = mock.Mock()
+        mock_get_drac_client.return_value = mock_client
+
+        physical_disks = self._generate_physical_disks()
+        raid_controller_dict = {
+            'id': 'RAID.Integrated.1-1',
+            'description': 'Integrated RAID Controller 1',
+            'manufacturer': 'DELL',
+            'model': 'PERC H710 Mini',
+            'primary_status': 'ok',
+            'firmware_version': '21.3.0-0009',
+            'bus': '1',
+            'supports_realtime': True}
+        raid_controller = test_utils.make_raid_controller(
+            raid_controller_dict)
+        mock_list_physical_disks.return_value = physical_disks
+        mock_commit_config.return_value = '42'
+        mock_client.list_raid_controllers.return_value = [raid_controller]
+        mock__reset_raid_config.return_value = {
+            'is_reboot_required': constants.RebootRequired.optional,
+            'is_commit_required': True}
+
+        mock_client.create_virtual_disk.return_value = {
+            'is_reboot_required': constants.RebootRequired.optional,
+            'is_commit_required': True
+        }
+
+        with task_manager.acquire(self.context, self.node.uuid,
+                                  shared=False) as task:
+            return_value = task.driver.raid.create_configuration(
+                task, create_root_volume=True, create_nonroot_volumes=False,
+                delete_existing=True)
+            mock_client.create_virtual_disk.assert_called_once_with(
+                'RAID.Integrated.1-1',
+                ['Disk.Bay.0:Enclosure.Internal.0-1:RAID.Integrated.1-1',
+                 'Disk.Bay.1:Enclosure.Internal.0-1:RAID.Integrated.1-1'],
+                '1', 51200, None, 2, 1)
+            mock_commit_config.assert_called_once_with(
+                task.node, raid_controller='RAID.Integrated.1-1',
+                realtime=True, reboot=False)
+
+            self.assertEqual(1, mock_commit_config.call_count)
+
+        self.assertEqual(states.DEPLOYWAIT, return_value)
+        self.node.refresh()
+        self.assertEqual(['42'],
+                         self.node.driver_internal_info['raid_config_job_ids'])
+
     @mock.patch.object(drac_common, 'get_drac_client', spec_set=True,
                        autospec=True)
     @mock.patch.object(drac_raid, 'list_physical_disks', autospec=True)
@@ -765,7 +894,8 @@ class DracRaidInterfaceTestCase(test_utils.BaseDracTest):
         with task_manager.acquire(self.context, self.node.uuid,
                                   shared=False) as task:
             task.driver.raid.create_configuration(
-                task, create_root_volume=True, create_nonroot_volumes=True)
+                task, create_root_volume=True, create_nonroot_volumes=True,
+                delete_existing=False)
 
             mock_client.create_virtual_disk.assert_called_once_with(
                 'RAID.Integrated.1-1',
@@ -821,7 +951,8 @@ class DracRaidInterfaceTestCase(test_utils.BaseDracTest):
         with task_manager.acquire(self.context, self.node.uuid,
                                   shared=False) as task:
             task.driver.raid.create_configuration(
-                task, create_root_volume=True, create_nonroot_volumes=True)
+                task, create_root_volume=True, create_nonroot_volumes=True,
+                delete_existing=False)
 
             mock_client.create_virtual_disk.assert_called_once_with(
                 'RAID.Integrated.1-1',
@@ -873,7 +1004,8 @@ class DracRaidInterfaceTestCase(test_utils.BaseDracTest):
         with task_manager.acquire(self.context, self.node.uuid,
                                   shared=False) as task:
             task.driver.raid.create_configuration(
-                task, create_root_volume=True, create_nonroot_volumes=True)
+                task, create_root_volume=True, create_nonroot_volumes=True,
+                delete_existing=False)
             mock_client.create_virtual_disk.assert_has_calls(
                 [mock.call(
                     'controller-2',
@@ -941,7 +1073,8 @@ class DracRaidInterfaceTestCase(test_utils.BaseDracTest):
         with task_manager.acquire(self.context, self.node.uuid,
                                   shared=False) as task:
             task.driver.raid.create_configuration(
-                task, create_root_volume=True, create_nonroot_volumes=True)
+                task, create_root_volume=True, create_nonroot_volumes=True,
+                delete_existing=False)
             mock_client.create_virtual_disk.assert_has_calls(
                 [mock.call(
                     'RAID.Integrated.1-1',
@@ -1004,7 +1137,8 @@ class DracRaidInterfaceTestCase(test_utils.BaseDracTest):
         with task_manager.acquire(self.context, self.node.uuid,
                                   shared=False) as task:
             task.driver.raid.create_configuration(
-                task, create_root_volume=True, create_nonroot_volumes=True)
+                task, create_root_volume=True, create_nonroot_volumes=True,
+                delete_existing=False)
 
             mock_client.create_virtual_disk.assert_has_calls(
                 [mock.call(
@@ -1069,7 +1203,8 @@ class DracRaidInterfaceTestCase(test_utils.BaseDracTest):
         with task_manager.acquire(self.context, self.node.uuid,
                                   shared=False) as task:
             task.driver.raid.create_configuration(
-                task, create_root_volume=True, create_nonroot_volumes=True)
+                task, create_root_volume=True, create_nonroot_volumes=True,
+                delete_existing=False)
 
             mock_client.create_virtual_disk.assert_has_calls(
                 [mock.call(
@@ -1162,7 +1297,8 @@ class DracRaidInterfaceTestCase(test_utils.BaseDracTest):
         with task_manager.acquire(self.context, self.node.uuid,
                                   shared=False) as task:
             task.driver.raid.create_configuration(
-                task, create_root_volume=True, create_nonroot_volumes=True)
+                task, create_root_volume=True, create_nonroot_volumes=True,
+                delete_existing=False)
 
             mock_client.create_virtual_disk.assert_has_calls(
                 [mock.call(
@@ -1189,6 +1325,7 @@ class DracRaidInterfaceTestCase(test_utils.BaseDracTest):
 
     @mock.patch.object(drac_common, 'get_drac_client', spec_set=True,
                        autospec=True)
+    @mock.patch.object(drac_raid, '_reset_raid_config', autospec=True)
     @mock.patch.object(drac_raid, 'list_physical_disks', autospec=True)
     @mock.patch.object(drac_job, 'validate_job_queue', spec_set=True,
                        autospec=True)
@@ -1196,7 +1333,8 @@ class DracRaidInterfaceTestCase(test_utils.BaseDracTest):
                        autospec=True)
     def test_create_configuration_fails_with_sharing_disabled(
             self, mock_commit_config, mock_validate_job_queue,
-            mock_list_physical_disks, mock_get_drac_client):
+            mock_list_physical_disks, mock__reset_raid_config,
+            mock_get_drac_client):
         mock_client = mock.Mock()
         mock_get_drac_client.return_value = mock_client
 
@@ -1210,7 +1348,18 @@ class DracRaidInterfaceTestCase(test_utils.BaseDracTest):
         self.physical_disks = self.physical_disks[0:3]
         physical_disks = self._generate_physical_disks()
         mock_list_physical_disks.return_value = physical_disks
-
+        raid_controller_dict = {
+            'id': 'RAID.Integrated.1-1',
+            'description': 'Integrated RAID Controller 1',
+            'manufacturer': 'DELL',
+            'model': 'PERC H710 Mini',
+            'primary_status': 'ok',
+            'firmware_version': '21.3.0-0009',
+            'bus': '1',
+            'supports_realtime': True}
+        raid_controller = test_utils.make_raid_controller(
+            raid_controller_dict)
+        mock_client.list_raid_controllers.return_value = [raid_controller]
         mock_commit_config.return_value = '42'
         mock_client.create_virtual_disk.return_value = {
             'is_reboot_required': constants.RebootRequired.optional,
@@ -1262,7 +1411,8 @@ class DracRaidInterfaceTestCase(test_utils.BaseDracTest):
         with task_manager.acquire(self.context, self.node.uuid,
                                   shared=False) as task:
             task.driver.raid.create_configuration(
-                task, create_root_volume=True, create_nonroot_volumes=True)
+                task, create_root_volume=True, create_nonroot_volumes=True,
+                delete_existing=False)
 
             mock_client.create_virtual_disk.assert_has_calls(
                 [mock.call(
@@ -1332,10 +1482,12 @@ class DracRaidInterfaceTestCase(test_utils.BaseDracTest):
             self.assertRaises(
                 exception.DracOperationError,
                 task.driver.raid.create_configuration,
-                task, create_root_volume=True, create_nonroot_volumes=True)
+                task, create_root_volume=True, create_nonroot_volumes=True,
+                delete_existing=False)
 
     @mock.patch.object(drac_common, 'get_drac_client', spec_set=True,
                        autospec=True)
+    @mock.patch.object(drac_raid, '_reset_raid_config', autospec=True)
     @mock.patch.object(drac_raid, 'list_physical_disks', autospec=True)
     @mock.patch.object(drac_job, 'validate_job_queue', spec_set=True,
                        autospec=True)
@@ -1346,7 +1498,7 @@ class DracRaidInterfaceTestCase(test_utils.BaseDracTest):
     def test_create_configuration_fails_if_not_enough_space(
             self, mock_create_virtual_disk, mock_commit_config,
             mock_validate_job_queue, mock_list_physical_disks,
-            mock_get_drac_client):
+            mock__reset_raid_config, mock_get_drac_client):
         mock_client = mock.Mock()
         mock_get_drac_client.return_value = mock_client
 
@@ -1362,6 +1514,21 @@ class DracRaidInterfaceTestCase(test_utils.BaseDracTest):
         self.physical_disks = self.physical_disks[0:3]
         physical_disks = self._generate_physical_disks()
         mock_list_physical_disks.return_value = physical_disks
+        raid_controller_dict = {
+            'id': 'RAID.Integrated.1-1',
+            'description': 'Integrated RAID Controller 1',
+            'manufacturer': 'DELL',
+            'model': 'PERC H710 Mini',
+            'primary_status': 'ok',
+            'firmware_version': '21.3.0-0009',
+            'bus': '1',
+            'supports_realtime': True}
+        raid_controller = test_utils.make_raid_controller(
+            raid_controller_dict)
+        mock_client.list_raid_controllers.return_value = [raid_controller]
+        mock__reset_raid_config.return_value = {
+            'is_reboot_required': constants.RebootRequired.optional,
+            'is_commit_required': True}
 
         mock_commit_config.return_value = '42'
         mock_create_virtual_disk.return_value = {
@@ -1377,6 +1544,8 @@ class DracRaidInterfaceTestCase(test_utils.BaseDracTest):
 
     @mock.patch.object(drac_common, 'get_drac_client', spec_set=True,
                        autospec=True)
+    @mock.patch.object(drac_raid, '_reset_raid_config', spec_set=True,
+                       autospec=True)
     @mock.patch.object(drac_raid, 'list_physical_disks', autospec=True)
     @mock.patch.object(drac_job, 'validate_job_queue', spec_set=True,
                        autospec=True)
@@ -1387,7 +1556,7 @@ class DracRaidInterfaceTestCase(test_utils.BaseDracTest):
     def test_create_configuration_fails_if_disk_already_reserved(
             self, mock_create_virtual_disk, mock_commit_config,
             mock_validate_job_queue, mock_list_physical_disks,
-            mock_get_drac_client):
+            mock__reset_raid_config, mock_get_drac_client):
         mock_client = mock.Mock()
         mock_get_drac_client.return_value = mock_client
 
@@ -1406,7 +1575,24 @@ class DracRaidInterfaceTestCase(test_utils.BaseDracTest):
         physical_disks = self._generate_physical_disks()
         mock_list_physical_disks.return_value = physical_disks
 
+        raid_controller_dict = {
+            'id': 'RAID.Integrated.1-1',
+            'description': 'Integrated RAID Controller 1',
+            'manufacturer': 'DELL',
+            'model': 'PERC H710 Mini',
+            'primary_status': 'ok',
+            'firmware_version': '21.3.0-0009',
+            'bus': '1',
+            'supports_realtime': True}
+        raid_controller = test_utils.make_raid_controller(
+            raid_controller_dict)
+        mock_client.list_raid_controllers.return_value = [raid_controller]
+
         mock_commit_config.return_value = '42'
+        mock__reset_raid_config.return_value = {
+            'is_reboot_required': constants.RebootRequired.optional,
+            'is_commit_required': True}
+
         mock_create_virtual_disk.return_value = {
             'is_reboot_required': constants.RebootRequired.optional,
             'is_commit_required': True}
@@ -1420,18 +1606,18 @@ class DracRaidInterfaceTestCase(test_utils.BaseDracTest):
 
     @mock.patch.object(drac_common, 'get_drac_client', spec_set=True,
                        autospec=True)
+    @mock.patch.object(drac_raid, '_reset_raid_config', autospec=True)
     @mock.patch.object(drac_raid, 'list_raid_controllers', autospec=True)
     @mock.patch.object(drac_job, 'validate_job_queue', spec_set=True,
                        autospec=True)
     @mock.patch.object(drac_raid, 'commit_config', spec_set=True,
                        autospec=True)
-    @mock.patch.object(drac_raid, '_reset_raid_config', spec_set=True,
-                       autospec=True)
-    def test_delete_configuration(self, mock__reset_raid_config,
-                                  mock_commit_config,
-                                  mock_validate_job_queue,
-                                  mock_list_raid_controllers,
-                                  mock_get_drac_client):
+    def _test_delete_configuration(self, expected_state,
+                                   mock_commit_config,
+                                   mock_validate_job_queue,
+                                   mock_list_raid_controllers,
+                                   mock__reset_raid_config,
+                                   mock_get_drac_client):
         mock_client = mock.Mock()
         mock_get_drac_client.return_value = mock_client
         raid_controller_dict = {
@@ -1459,11 +1645,19 @@ class DracRaidInterfaceTestCase(test_utils.BaseDracTest):
                 task.node, raid_controller='RAID.Integrated.1-1', reboot=False,
                 realtime=True)
 
-        self.assertEqual(states.CLEANWAIT, return_value)
+        self.assertEqual(expected_state, return_value)
         self.node.refresh()
         self.assertEqual(['42'],
                          self.node.driver_internal_info['raid_config_job_ids'])
 
+    def test_delete_configuration_in_clean(self):
+        self._test_delete_configuration(states.CLEANWAIT)
+
+    def test_delete_configuration_in_deploy(self):
+        self.node.clean_step = None
+        self.node.save()
+        self._test_delete_configuration(states.DEPLOYWAIT)
+
     @mock.patch.object(drac_common, 'get_drac_client', spec_set=True,
                        autospec=True)
     @mock.patch.object(drac_raid, 'list_raid_controllers', autospec=True)
@@ -1568,7 +1762,7 @@ class DracRaidInterfaceTestCase(test_utils.BaseDracTest):
 
         with task_manager.acquire(self.context, self.node.uuid,
                                   shared=False) as task:
-            return_value = task.driver.raid._execute_cleaning_foreign_drives(
+            return_value = task.driver.raid._execute_foreign_drives(
                 task, self.node)
             mock_resume.assert_called_once_with(
                 task, 'cleaning', 'continue_node_clean')
diff --git a/releasenotes/notes/add-deploy-steps-drac-raid-interface-7023c03a96996265.yaml b/releasenotes/notes/add-deploy-steps-drac-raid-interface-7023c03a96996265.yaml
new file mode 100644
index 0000000000..c19dfc1a19
--- /dev/null
+++ b/releasenotes/notes/add-deploy-steps-drac-raid-interface-7023c03a96996265.yaml
@@ -0,0 +1,10 @@
+---
+features:
+  - |
+    Adds support for deploy steps to ``raid`` interface of ``idrac``
+    hardware type. The methods ``apply_configuration`` and
+    ``delete_configuration`` can be used as deploy steps.
+  - |
+    Adds a new ``delete_existing`` argument to the ``create_configuration``
+    clean step on the ``idrac`` ``raid`` interface which can be used to
+    delete existing virtual disks. The default for this argument is ``False``.