diff --git a/ironic/common/raid.py b/ironic/common/raid.py
index feb6316f69..54a3419c09 100644
--- a/ironic/common/raid.py
+++ b/ironic/common/raid.py
@@ -105,7 +105,6 @@ def update_raid_info(node, raid_config):
     current = raid_config.copy()
     current['last_updated'] = str(datetime.datetime.utcnow())
     node.raid_config = current
-    node.target_raid_config = None
 
     # Current RAID configuration can have 0 or 1 root volumes. If there
     # are > 1 root volumes, then it's invalid.  We check for this condition
diff --git a/ironic/conductor/manager.py b/ironic/conductor/manager.py
index e45a8c949f..931515d53a 100644
--- a/ironic/conductor/manager.py
+++ b/ironic/conductor/manager.py
@@ -199,9 +199,10 @@ CLEANING_INTERFACE_PRIORITY = {
     # by which interface is implementing the clean step. The clean step of the
     # interface with the highest value here, will be executed first in that
     # case.
-    'power': 3,
-    'management': 2,
-    'deploy': 1
+    'power': 4,
+    'management': 3,
+    'deploy': 2,
+    'raid': 1,
 }
 SYNC_EXCLUDED_STATES = (states.DEPLOYWAIT, states.CLEANWAIT, states.ENROLL)
 
diff --git a/ironic/drivers/agent.py b/ironic/drivers/agent.py
index 39e4270429..8f95f098e8 100644
--- a/ironic/drivers/agent.py
+++ b/ironic/drivers/agent.py
@@ -54,6 +54,7 @@ class AgentAndIPMIToolDriver(base.BaseDriver):
         self.vendor = utils.MixinVendorInterface(
             self.mapping,
             driver_passthru_mapping=self.driver_passthru_mapping)
+        self.raid = agent.AgentRAID()
 
 
 class AgentAndIPMINativeDriver(base.BaseDriver):
@@ -75,6 +76,7 @@ class AgentAndIPMINativeDriver(base.BaseDriver):
         self.management = ipminative.NativeIPMIManagement()
         self.console = ipminative.NativeIPMIShellinaboxConsole()
         self.vendor = agent.AgentVendorInterface()
+        self.raid = agent.AgentRAID()
 
 
 class AgentAndSSHDriver(base.BaseDriver):
@@ -96,6 +98,7 @@ class AgentAndSSHDriver(base.BaseDriver):
         self.deploy = agent.AgentDeploy()
         self.management = ssh.SSHManagement()
         self.vendor = agent.AgentVendorInterface()
+        self.raid = agent.AgentRAID()
 
 
 class AgentAndVirtualBoxDriver(base.BaseDriver):
@@ -121,6 +124,7 @@ class AgentAndVirtualBoxDriver(base.BaseDriver):
         self.deploy = agent.AgentDeploy()
         self.management = virtualbox.VirtualBoxManagement()
         self.vendor = agent.AgentVendorInterface()
+        self.raid = agent.AgentRAID()
 
 
 class AgentAndUcsDriver(base.BaseDriver):
diff --git a/ironic/drivers/fake.py b/ironic/drivers/fake.py
index 657cee0ef8..da4644935d 100644
--- a/ironic/drivers/fake.py
+++ b/ironic/drivers/fake.py
@@ -139,6 +139,7 @@ class FakeAgentDriver(base.BaseDriver):
         self.boot = pxe.PXEBoot()
         self.deploy = agent.AgentDeploy()
         self.vendor = agent.AgentVendorInterface()
+        self.raid = agent.AgentRAID()
 
 
 class FakeIBootDriver(base.BaseDriver):
diff --git a/ironic/drivers/ilo.py b/ironic/drivers/ilo.py
index 524e463ebf..5e91fdacdc 100644
--- a/ironic/drivers/ilo.py
+++ b/ironic/drivers/ilo.py
@@ -20,6 +20,7 @@ from oslo_utils import importutils
 from ironic.common import exception
 from ironic.common.i18n import _
 from ironic.drivers import base
+from ironic.drivers.modules import agent
 from ironic.drivers.modules.ilo import deploy
 from ironic.drivers.modules.ilo import inspect
 from ironic.drivers.modules.ilo import management
@@ -72,3 +73,4 @@ class IloVirtualMediaAgentDriver(base.BaseDriver):
         self.management = management.IloManagement()
         self.vendor = deploy.IloVirtualMediaAgentVendorInterface()
         self.inspect = inspect.IloInspect()
+        self.raid = agent.AgentRAID()
diff --git a/ironic/drivers/modules/agent.py b/ironic/drivers/modules/agent.py
index 7474459683..7821cc01cd 100644
--- a/ironic/drivers/modules/agent.py
+++ b/ironic/drivers/modules/agent.py
@@ -27,6 +27,7 @@ from ironic.common.i18n import _LI
 from ironic.common import image_service
 from ironic.common import keystone
 from ironic.common import paths
+from ironic.common import raid
 from ironic.common import states
 from ironic.conductor import task_manager
 from ironic.conductor import utils as manager_utils
@@ -466,3 +467,138 @@ class AgentVendorInterface(agent_base_vendor.BaseAgentVendor):
         # check once all in-tree drivers have a boot interface.
         if task.driver.boot:
             task.driver.boot.clean_up_ramdisk(task)
+
+
+class AgentRAID(base.RAIDInterface):
+    """Implementation of RAIDInterface which uses agent ramdisk."""
+
+    def get_properties(self):
+        """Return the properties of the interface."""
+        return {}
+
+    @base.clean_step(priority=0)
+    def create_configuration(self, task,
+                             create_root_volume=True,
+                             create_nonroot_volumes=True):
+        """Create a RAID configuration on a bare metal using agent ramdisk.
+
+        This method creates a RAID configuration on the given node.
+
+        :param task: a TaskManager instance.
+        :param create_root_volume: If True, a root volume is created
+            during RAID configuration. Otherwise, no root volume is
+            created. Default is True.
+        :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 operation was successfully invoked.
+        :raises: MissingParameterValue, if node.target_raid_config is missing
+            or was found to be empty after skipping root volume and/or non-root
+            volumes.
+        """
+        node = task.node
+        LOG.debug("Agent RAID create_configuration invoked for node %(node)s "
+                  "with create_root_volume=%(create_root_volume)s and "
+                  "create_nonroot_volumes=%(create_nonroot_volumes)s with the "
+                  "following target_raid_config: %(target_raid_config)s.",
+                  {'node': node.uuid,
+                   'create_root_volume': create_root_volume,
+                   'create_nonroot_volumes': create_nonroot_volumes,
+                   'target_raid_config': node.target_raid_config})
+
+        if not node.target_raid_config:
+            raise exception.MissingParameterValue(
+                _("Node %s has no target RAID configuration.") % node.uuid)
+
+        target_raid_config = node.target_raid_config.copy()
+        error_msg = None
+
+        error_msg_list = []
+        if not create_root_volume:
+            target_raid_config['logical_disks'] = [
+                x for x in target_raid_config['logical_disks']
+                if x.get('is_root_volume', False) is False]
+            error_msg_list.append(_("skipping root volume"))
+
+        if not create_nonroot_volumes:
+            error_msg_list.append(_("skipping non-root volumes"))
+
+            target_raid_config['logical_disks'] = [
+                x for x in target_raid_config['logical_disks']
+                if x.get('is_root_volume', False) is True]
+
+        if not target_raid_config['logical_disks']:
+            error_msg = _(' and ').join(error_msg_list)
+            raise exception.MissingParameterValue(
+                _("Node %(node)s has empty target RAID configuration "
+                  "after %(msg)s.") % {'node': node.uuid, 'msg': error_msg})
+
+        # Rewrite it back to the node object, but no need to save it as
+        # we need to just send this to the agent ramdisk.
+        node.driver_internal_info['target_raid_config'] = target_raid_config
+
+        LOG.debug("Calling agent RAID create_configuration for node %(node)s "
+                  "with the following target RAID configuration: %(target)s",
+                  {'node': node.uuid, 'target': target_raid_config})
+        step = node.clean_step
+        return deploy_utils.agent_execute_clean_step(task, step)
+
+    @staticmethod
+    @agent_base_vendor.post_clean_step_hook(
+        interface='raid', step='create_configuration')
+    def _create_configuration_final(task, command):
+        """Clean step hook after a RAID configuration was created.
+
+        This method is invoked as a post clean step hook by the Ironic
+        conductor once a create raid configuration is completed successfully.
+        The node (properties, capabilities, RAID information) will be updated
+        to reflect the actual RAID configuration that was created.
+
+        :param task: a TaskManager instance.
+        :param command: A command result structure of the RAID operation
+            returned from agent ramdisk on query of the status of command(s).
+        :raises: InvalidParameterValue, if 'current_raid_config' has more than
+            one root volume or if node.properties['capabilities'] is malformed.
+        :raises: IronicException, if clean_result couldn't be found within
+            the 'command' argument passed.
+        """
+        try:
+            clean_result = command['command_result']['clean_result']
+        except KeyError:
+            raise exception.IronicException(
+                _("Agent ramdisk didn't return a proper command result while "
+                  "cleaning %(node)s. It returned '%(result)s' after command "
+                  "execution.") % {'node': task.node.uuid,
+                                   'result': command})
+
+        raid.update_raid_info(task.node, clean_result)
+
+    @base.clean_step(priority=0)
+    def delete_configuration(self, task):
+        """Deletes RAID configuration on the given node.
+
+        :param task: a TaskManager instance.
+        :returns: states.CLEANWAIT if operation was successfully invoked
+        """
+        LOG.debug("Agent RAID delete_configuration invoked for node %s.",
+                  task.node.uuid)
+        step = task.node.clean_step
+        return deploy_utils.agent_execute_clean_step(task, step)
+
+    @staticmethod
+    @agent_base_vendor.post_clean_step_hook(
+        interface='raid', step='delete_configuration')
+    def _delete_configuration_final(task, command):
+        """Clean step hook after RAID configuration was deleted.
+
+        This method is invoked as a post clean step hook by the Ironic
+        conductor once a delete raid configuration is completed successfully.
+        It sets node.raid_config to empty dictionary.
+
+        :param task: a TaskManager instance.
+        :param command: A command result structure of the RAID operation
+            returned from agent ramdisk on query of the status of command(s).
+        :returns: None
+        """
+        task.node.raid_config = {}
+        task.node.save()
diff --git a/ironic/tests/common/test_raid.py b/ironic/tests/common/test_raid.py
index 6e76ef0751..893c7d2d4b 100644
--- a/ironic/tests/common/test_raid.py
+++ b/ironic/tests/common/test_raid.py
@@ -182,7 +182,9 @@ class RaidPublicMethodsTestCase(db_base.DbTestCase):
             properties['capabilities'] = capabilities
             del properties['local_gb']
             node.properties = properties
-            node.save()
+        target_raid_config = json.loads(raid_constants.RAID_CONFIG_OKAY)
+        node.target_raid_config = target_raid_config
+        node.save()
         raid.update_raid_info(node, current_config)
         properties = node.properties
         current = node.raid_config
@@ -202,7 +204,8 @@ class RaidPublicMethodsTestCase(db_base.DbTestCase):
             if capabilities:
                 self.assertNotIn('raid_level:1', properties['capabilities'])
 
-        self.assertEqual({}, target)
+        # Verify node.target_raid_config is preserved.
+        self.assertEqual(target_raid_config, target)
 
     def test_update_raid_info_okay(self):
         current_config = json.loads(raid_constants.CURRENT_RAID_CONFIG)
diff --git a/ironic/tests/drivers/test_agent.py b/ironic/tests/drivers/test_agent.py
index 2546d4f49f..7266b185b7 100644
--- a/ironic/tests/drivers/test_agent.py
+++ b/ironic/tests/drivers/test_agent.py
@@ -20,11 +20,14 @@ from oslo_config import cfg
 from ironic.common import exception
 from ironic.common import image_service
 from ironic.common import keystone
+from ironic.common import raid
 from ironic.common import states
 from ironic.conductor import task_manager
 from ironic.conductor import utils as manager_utils
 from ironic.drivers.modules import agent
+from ironic.drivers.modules import agent_base_vendor
 from ironic.drivers.modules import agent_client
+from ironic.drivers.modules import deploy_utils
 from ironic.drivers.modules import fake
 from ironic.drivers.modules import pxe
 from ironic.tests.conductor import utils as mgr_utils
@@ -636,3 +639,183 @@ class TestAgentVendor(db_base.DbTestCase):
             mock_get_cmd.return_value = [{'command_name': 'prepare_image',
                                           'command_status': 'RUNNING'}]
             self.assertFalse(self.passthru.deploy_is_done(task))
+
+
+class AgentRAIDTestCase(db_base.DbTestCase):
+
+    def setUp(self):
+        super(AgentRAIDTestCase, self).setUp()
+        mgr_utils.mock_the_extension_manager(driver="fake_agent")
+        self.passthru = agent.AgentVendorInterface()
+        self.target_raid_config = {
+            "logical_disks": [
+                {'size_gb': 200, 'raid_level': 0, 'is_root_volume': True},
+                {'size_gb': 200, 'raid_level': 5}
+            ]}
+        self.clean_step = {'step': 'create_configuration',
+                           'interface': 'raid'}
+        n = {
+            'driver': 'fake_agent',
+            'instance_info': INSTANCE_INFO,
+            'driver_info': DRIVER_INFO,
+            'driver_internal_info': DRIVER_INTERNAL_INFO,
+            'target_raid_config': self.target_raid_config,
+            'clean_step': self.clean_step,
+        }
+        self.node = object_utils.create_test_node(self.context, **n)
+
+    @mock.patch.object(deploy_utils, 'agent_get_clean_steps', autospec=True)
+    def test_get_clean_steps(self, get_steps_mock):
+        get_steps_mock.return_value = [
+            {'step': 'create_configuration', 'interface': 'raid',
+             'priority': 1},
+            {'step': 'delete_configuration', 'interface': 'raid',
+             'priority': 2}]
+
+        with task_manager.acquire(self.context, self.node.uuid) as task:
+            ret = task.driver.raid.get_clean_steps(task)
+
+        self.assertEqual(0, ret[0]['priority'])
+        self.assertEqual(0, ret[1]['priority'])
+
+    @mock.patch.object(deploy_utils, 'agent_execute_clean_step',
+                       autospec=True)
+    def test_create_configuration(self, execute_mock):
+        with task_manager.acquire(self.context, self.node.uuid) as task:
+            execute_mock.return_value = states.CLEANWAIT
+
+            return_value = task.driver.raid.create_configuration(task)
+
+            self.assertEqual(states.CLEANWAIT, return_value)
+            self.assertEqual(
+                self.target_raid_config,
+                task.node.driver_internal_info['target_raid_config'])
+            execute_mock.assert_called_once_with(task, self.clean_step)
+
+    @mock.patch.object(deploy_utils, 'agent_execute_clean_step',
+                       autospec=True)
+    def test_create_configuration_skip_root(self, execute_mock):
+        with task_manager.acquire(self.context, self.node.uuid) as task:
+            execute_mock.return_value = states.CLEANWAIT
+
+            return_value = task.driver.raid.create_configuration(
+                task, create_root_volume=False)
+
+            self.assertEqual(states.CLEANWAIT, return_value)
+            execute_mock.assert_called_once_with(task, self.clean_step)
+            exp_target_raid_config = {
+                "logical_disks": [
+                    {'size_gb': 200, 'raid_level': 5}
+                ]}
+            self.assertEqual(
+                exp_target_raid_config,
+                task.node.driver_internal_info['target_raid_config'])
+
+    @mock.patch.object(deploy_utils, 'agent_execute_clean_step',
+                       autospec=True)
+    def test_create_configuration_skip_nonroot(self, execute_mock):
+        with task_manager.acquire(self.context, self.node.uuid) as task:
+            execute_mock.return_value = states.CLEANWAIT
+
+            return_value = task.driver.raid.create_configuration(
+                task, create_nonroot_volumes=False)
+
+            self.assertEqual(states.CLEANWAIT, return_value)
+            execute_mock.assert_called_once_with(task, self.clean_step)
+            exp_target_raid_config = {
+                "logical_disks": [
+                    {'size_gb': 200, 'raid_level': 0, 'is_root_volume': True},
+                ]}
+            self.assertEqual(
+                exp_target_raid_config,
+                task.node.driver_internal_info['target_raid_config'])
+
+    @mock.patch.object(deploy_utils, 'agent_execute_clean_step',
+                       autospec=True)
+    def test_create_configuration_no_target_raid_config_after_skipping(
+            self, execute_mock):
+        with task_manager.acquire(self.context, self.node.uuid) as task:
+            self.assertRaises(
+                exception.MissingParameterValue,
+                task.driver.raid.create_configuration,
+                task, create_root_volume=False,
+                create_nonroot_volumes=False)
+
+            self.assertFalse(execute_mock.called)
+
+    @mock.patch.object(deploy_utils, 'agent_execute_clean_step',
+                       autospec=True)
+    def test_create_configuration_empty_target_raid_config(
+            self, execute_mock):
+        execute_mock.return_value = states.CLEANING
+        self.node.target_raid_config = {}
+        self.node.save()
+        with task_manager.acquire(self.context, self.node.uuid) as task:
+            self.assertRaises(exception.MissingParameterValue,
+                              task.driver.raid.create_configuration,
+                              task)
+            self.assertFalse(execute_mock.called)
+
+    @mock.patch.object(raid, 'update_raid_info', autospec=True)
+    def test__create_configuration_final(
+            self, update_raid_info_mock):
+        command = {'command_result': {'clean_result': 'foo'}}
+        with task_manager.acquire(self.context, self.node.uuid) as task:
+            raid_mgmt = agent.AgentRAID
+            raid_mgmt._create_configuration_final(task, command)
+            update_raid_info_mock.assert_called_once_with(task.node, 'foo')
+
+    @mock.patch.object(raid, 'update_raid_info', autospec=True)
+    def test__create_configuration_final_registered(
+            self, update_raid_info_mock):
+        self.node.clean_step = {'interface': 'raid',
+                                'step': 'create_configuration'}
+        command = {'command_result': {'clean_result': 'foo'}}
+        create_hook = agent_base_vendor._get_post_clean_step_hook(self.node)
+        with task_manager.acquire(self.context, self.node.uuid) as task:
+            create_hook(task, command)
+            update_raid_info_mock.assert_called_once_with(task.node, 'foo')
+
+    @mock.patch.object(raid, 'update_raid_info', autospec=True)
+    def test__create_configuration_final_bad_command_result(
+            self, update_raid_info_mock):
+        command = {}
+        with task_manager.acquire(self.context, self.node.uuid) as task:
+            raid_mgmt = agent.AgentRAID
+            self.assertRaises(exception.IronicException,
+                              raid_mgmt._create_configuration_final,
+                              task, command)
+            self.assertFalse(update_raid_info_mock.called)
+
+    @mock.patch.object(deploy_utils, 'agent_execute_clean_step',
+                       autospec=True)
+    def test_delete_configuration(self, execute_mock):
+        execute_mock.return_value = states.CLEANING
+        with task_manager.acquire(self.context, self.node.uuid) as task:
+            return_value = task.driver.raid.delete_configuration(task)
+
+            execute_mock.assert_called_once_with(task, self.clean_step)
+            self.assertEqual(states.CLEANING, return_value)
+
+    def test__delete_configuration_final(self):
+        command = {'command_result': {'clean_result': 'foo'}}
+        with task_manager.acquire(self.context, self.node.uuid) as task:
+            task.node.raid_config = {'foo': 'bar'}
+            raid_mgmt = agent.AgentRAID
+            raid_mgmt._delete_configuration_final(task, command)
+
+        self.node.refresh()
+        self.assertEqual({}, self.node.raid_config)
+
+    def test__delete_configuration_final_registered(
+            self):
+        self.node.clean_step = {'interface': 'raid',
+                                'step': 'delete_configuration'}
+        self.node.raid_config = {'foo': 'bar'}
+        command = {'command_result': {'clean_result': 'foo'}}
+        delete_hook = agent_base_vendor._get_post_clean_step_hook(self.node)
+        with task_manager.acquire(self.context, self.node.uuid) as task:
+            delete_hook(task, command)
+
+        self.node.refresh()
+        self.assertEqual({}, self.node.raid_config)
diff --git a/ironic/tests/drivers/test_deploy_utils.py b/ironic/tests/drivers/test_deploy_utils.py
index aa16854643..7b7f22f97f 100644
--- a/ironic/tests/drivers/test_deploy_utils.py
+++ b/ironic/tests/drivers/test_deploy_utils.py
@@ -1901,7 +1901,7 @@ class AgentCleaningTestCase(db_base.DbTestCase):
                      'step': 'update_firmware',
                      'priority': 30},
                     {'interface': 'raid',
-                     'step': 'create_raid',
+                     'step': 'create_configuration',
                      'priority': 10},
                 ]
             }