diff --git a/devstack/lib/ironic b/devstack/lib/ironic
index ba2f088ef4..74e038a66e 100644
--- a/devstack/lib/ironic
+++ b/devstack/lib/ironic
@@ -560,6 +560,8 @@ IRONIC_ANSIBLE_SSH_USER=${IRONIC_ANSIBLE_SSH_USER:-}
 # DevStack deployment, as we do not distribute this generated key to subnodes yet.
 IRONIC_ANSIBLE_SSH_KEY=${IRONIC_ANSIBLE_SSH_KEY:-$IRONIC_DATA_DIR/ansible_ssh_key}
 
+IRONIC_AGENT_IMAGE_DOWNLOAD_SOURCE=${IRONIC_AGENT_IMAGE_DOWNLOAD_SOURCE:-swift}
+
 # Functions
 # ---------
 
@@ -1105,6 +1107,8 @@ function configure_ironic {
     iniset $IRONIC_CONF_FILE agent deploy_logs_collect $IRONIC_DEPLOY_LOGS_COLLECT
     iniset $IRONIC_CONF_FILE agent deploy_logs_storage_backend $IRONIC_DEPLOY_LOGS_STORAGE_BACKEND
     iniset $IRONIC_CONF_FILE agent deploy_logs_local_path $IRONIC_DEPLOY_LOGS_LOCAL_PATH
+    # Set image_download_source for direct interface
+    iniset $IRONIC_CONF_FILE agent image_download_source $IRONIC_AGENT_IMAGE_DOWNLOAD_SOURCE
 
     # Configure Ironic conductor, if it was enabled.
     if is_service_enabled ir-cond; then
diff --git a/ironic/conf/agent.py b/ironic/conf/agent.py
index 3348c03419..57bdc72065 100644
--- a/ironic/conf/agent.py
+++ b/ironic/conf/agent.py
@@ -89,6 +89,19 @@ opts = [
                       'forever or until manually deleted. Used when the '
                       'deploy_logs_storage_backend is configured to '
                       '"swift".')),
+    cfg.StrOpt('image_download_source',
+               choices=[('swift', _('IPA ramdisk retrieves instance image '
+                                    'from the Object Storage service.')),
+                        ('http', _('IPA ramdisk retrieves instance image '
+                                   'from HTTP service served at conductor '
+                                   'nodes.'))],
+               default='swift',
+               help=_('Specifies whether direct deploy interface should try '
+                      'to use the image source directly or if ironic should '
+                      'cache the image on the conductor and serve it from '
+                      'ironic\'s own http server. This option takes effect '
+                      'only when instance image is provided from the Image '
+                      'service.')),
 ]
 
 
diff --git a/ironic/conf/deploy.py b/ironic/conf/deploy.py
index 780792bb5d..0c763fd751 100644
--- a/ironic/conf/deploy.py
+++ b/ironic/conf/deploy.py
@@ -96,6 +96,13 @@ opts = [
                 help=_('Whether to upload the config drive to object store. '
                        'Set this option to True to store config drive '
                        'in a swift endpoint.')),
+    cfg.StrOpt('http_image_subdir',
+               default='agent_images',
+               help=_('The name of subdirectory under ironic-conductor '
+                      'node\'s HTTP root path which is used to place instance '
+                      'images for the direct deploy interface, when local '
+                      'HTTP service is incorporated to provide instance image '
+                      'instead of swift tempurls.')),
 ]
 
 
diff --git a/ironic/drivers/modules/agent.py b/ironic/drivers/modules/agent.py
index a015f6305e..9a4421c05c 100644
--- a/ironic/drivers/modules/agent.py
+++ b/ironic/drivers/modules/agent.py
@@ -145,6 +145,27 @@ def validate_image_proxies(node):
         raise exception.InvalidParameterValue(msg)
 
 
+def validate_http_provisioning_configuration(node):
+    """Validate configuration options required to perform HTTP provisioning.
+
+    :param node: an ironic node object
+    :raises: MissingParameterValue if required option(s) is not set.
+    """
+    image_source = node.instance_info.get('image_source')
+    if (not service_utils.is_glance_image(image_source) or
+            CONF.agent.image_download_source != 'http'):
+        return
+
+    params = {
+        '[deploy]http_url': CONF.deploy.http_url,
+        '[deploy]http_root': CONF.deploy.http_root,
+        '[deploy]http_image_subdir': CONF.deploy.http_image_subdir
+    }
+    error_msg = _('Node %s failed to validate http provisoning. Some '
+                  'configuration options were missing') % node.uuid
+    deploy_utils.check_for_missing_params(params, error_msg)
+
+
 class AgentDeployMixin(agent_base_vendor.AgentDeployMixin):
 
     @METRICS.timer('AgentDeployMixin.deploy_has_started')
@@ -338,6 +359,10 @@ class AgentDeployMixin(agent_base_vendor.AgentDeployMixin):
         else:
             manager_utils.node_set_boot_device(task, 'disk', persistent=True)
 
+        # Remove symbolic link when deploy is done.
+        if CONF.agent.image_download_source == 'http':
+            deploy_utils.remove_http_instance_symlink(task.node.uuid)
+
         LOG.debug('Rebooting node %s to instance', node.uuid)
         self.reboot_and_finish_deploy(task)
 
@@ -397,6 +422,8 @@ class AgentDeploy(AgentDeployMixin, base.DeployInterface):
                     "image_source's image_checksum must be provided in "
                     "instance_info for node %s") % node.uuid)
 
+        validate_http_provisioning_configuration(node)
+
         check_image_size(task, image_source)
         # Validate the root device hints
         try:
@@ -562,6 +589,8 @@ class AgentDeploy(AgentDeployMixin, base.DeployInterface):
         task.driver.boot.clean_up_instance(task)
         provider = dhcp_factory.DHCPFactory()
         provider.clean_dhcp(task)
+        if CONF.agent.image_download_source == 'http':
+            deploy_utils.destroy_http_instance_images(task.node)
 
     def take_over(self, task):
         """Take over management of this node from a dead conductor.
diff --git a/ironic/drivers/modules/deploy_utils.py b/ironic/drivers/modules/deploy_utils.py
index b37309e2a8..9fcd17f369 100644
--- a/ironic/drivers/modules/deploy_utils.py
+++ b/ironic/drivers/modules/deploy_utils.py
@@ -22,9 +22,11 @@ import time
 
 from ironic_lib import disk_utils
 from ironic_lib import metrics_utils
+from ironic_lib import utils as il_utils
 from oslo_concurrency import processutils
 from oslo_log import log as logging
 from oslo_utils import excutils
+from oslo_utils import fileutils
 from oslo_utils import netutils
 from oslo_utils import strutils
 import six
@@ -1067,6 +1069,98 @@ def _check_disk_layout_unchanged(node, i_info):
                                               {'error_msg': error_msg})
 
 
+def _get_image_dir_path(node_uuid):
+    """Generate the dir for an instances disk."""
+    return os.path.join(CONF.pxe.images_path, node_uuid)
+
+
+def _get_image_file_path(node_uuid):
+    """Generate the full path for an instances disk."""
+    return os.path.join(_get_image_dir_path(node_uuid), 'disk')
+
+
+def _get_http_image_symlink_dir_path():
+    """Generate the dir for storing symlinks to cached instance images."""
+    return os.path.join(CONF.deploy.http_root, CONF.deploy.http_image_subdir)
+
+
+def _get_http_image_symlink_file_path(node_uuid):
+    """Generate the full path for the symlink to an cached instance image."""
+    return os.path.join(_get_http_image_symlink_dir_path(), node_uuid)
+
+
+def direct_deploy_should_convert_raw_image(node):
+    """Whether converts image to raw format for specified node.
+
+    :param node: ironic node object
+    :returns: Boolean, whether the direct deploy interface should convert
+        image to raw.
+    """
+    iwdi = node.driver_internal_info.get('is_whole_disk_image')
+    return CONF.force_raw_images and CONF.agent.stream_raw_images and iwdi
+
+
+@image_cache.cleanup(priority=50)
+class InstanceImageCache(image_cache.ImageCache):
+
+    def __init__(self):
+        super(self.__class__, self).__init__(
+            CONF.pxe.instance_master_path,
+            # MiB -> B
+            cache_size=CONF.pxe.image_cache_size * 1024 * 1024,
+            # min -> sec
+            cache_ttl=CONF.pxe.image_cache_ttl * 60)
+
+
+@METRICS.timer('cache_instance_image')
+def cache_instance_image(ctx, node, force_raw=CONF.force_raw_images):
+    """Fetch the instance's image from Glance
+
+    This method pulls the AMI and writes them to the appropriate place
+    on local disk.
+
+    :param ctx: context
+    :param node: an ironic node object
+    :param force_raw: whether convert image to raw format
+    :returns: a tuple containing the uuid of the image and the path in
+        the filesystem where image is cached.
+    """
+    i_info = parse_instance_info(node)
+    fileutils.ensure_tree(_get_image_dir_path(node.uuid))
+    image_path = _get_image_file_path(node.uuid)
+    uuid = i_info['image_source']
+
+    LOG.debug("Fetching image %(image)s for node %(uuid)s",
+              {'image': uuid, 'uuid': node.uuid})
+
+    fetch_images(ctx, InstanceImageCache(), [(uuid, image_path)],
+                 force_raw)
+
+    return (uuid, image_path)
+
+
+@METRICS.timer('destroy_images')
+def destroy_images(node_uuid):
+    """Delete instance's image file.
+
+    :param node_uuid: the uuid of the ironic node.
+    """
+    il_utils.unlink_without_raise(_get_image_file_path(node_uuid))
+    utils.rmtree_without_raise(_get_image_dir_path(node_uuid))
+    InstanceImageCache().clean_up()
+
+
+def remove_http_instance_symlink(node_uuid):
+    symlink_path = _get_http_image_symlink_file_path(node_uuid)
+    il_utils.unlink_without_raise(symlink_path)
+
+
+def destroy_http_instance_images(node):
+    """Delete instance image file and symbolic link refers to it."""
+    remove_http_instance_symlink(node.uuid)
+    destroy_images(node.uuid)
+
+
 @METRICS.timer('build_instance_info_for_deploy')
 def build_instance_info_for_deploy(task):
     """Build instance_info necessary for deploying to a node.
@@ -1098,17 +1192,55 @@ def build_instance_info_for_deploy(task):
     instance_info = node.instance_info
     iwdi = node.driver_internal_info.get('is_whole_disk_image')
     image_source = instance_info['image_source']
+
     if service_utils.is_glance_image(image_source):
         glance = image_service.GlanceImageService(version=2,
                                                   context=task.context)
         image_info = glance.show(image_source)
         LOG.debug('Got image info: %(info)s for node %(node)s.',
                   {'info': image_info, 'node': node.uuid})
-        swift_temp_url = glance.swift_temp_url(image_info)
-        validate_image_url(swift_temp_url, secret=True)
-        instance_info['image_url'] = swift_temp_url
-        instance_info['image_checksum'] = image_info['checksum']
-        instance_info['image_disk_format'] = image_info['disk_format']
+        if CONF.agent.image_download_source == 'swift':
+            swift_temp_url = glance.swift_temp_url(image_info)
+            validate_image_url(swift_temp_url, secret=True)
+            instance_info['image_url'] = swift_temp_url
+            instance_info['image_checksum'] = image_info['checksum']
+            instance_info['image_disk_format'] = image_info['disk_format']
+        else:
+            # Ironic cache and serve images from httpboot server
+            force_raw = direct_deploy_should_convert_raw_image(node)
+            _, image_path = cache_instance_image(task.context, node,
+                                                 force_raw=force_raw)
+            if force_raw:
+                time_start = time.time()
+                LOG.debug('Start calculating checksum for image %(image)s.',
+                          {'image': image_path})
+                checksum = fileutils.compute_file_checksum(image_path,
+                                                           algorithm='md5')
+                time_elapsed = time.time() - time_start
+                LOG.debug('Recalculated checksum for image %(image)s in '
+                          '%(delta).2f seconds, new checksum %(checksum)s ',
+                          {'image': image_path, 'delta': time_elapsed,
+                           'checksum': checksum})
+                instance_info['image_checksum'] = checksum
+                instance_info['image_disk_format'] = 'raw'
+            else:
+                instance_info['image_checksum'] = image_info['checksum']
+                instance_info['image_disk_format'] = image_info['disk_format']
+
+            # Create symlink and update image url
+            symlink_dir = _get_http_image_symlink_dir_path()
+            fileutils.ensure_tree(symlink_dir)
+            symlink_path = _get_http_image_symlink_file_path(node.uuid)
+            utils.create_link_without_raise(image_path, symlink_path)
+            base_url = CONF.deploy.http_url
+            if base_url.endswith('/'):
+                base_url = base_url[:-1]
+            http_image_url = '/'.join(
+                [base_url, CONF.deploy.http_image_subdir,
+                 node.uuid])
+            validate_image_url(http_image_url, secret=True)
+            instance_info['image_url'] = http_image_url
+
         instance_info['image_container_format'] = (
             image_info['container_format'])
         instance_info['image_tags'] = image_info.get('tags', [])
diff --git a/ironic/drivers/modules/image_cache.py b/ironic/drivers/modules/image_cache.py
index 76b55d51d4..0e1f7d35cb 100644
--- a/ironic/drivers/modules/image_cache.py
+++ b/ironic/drivers/modules/image_cache.py
@@ -99,6 +99,11 @@ class ImageCache(object):
             href_encoded = href.encode('utf-8') if six.PY2 else href
             master_file_name = str(uuid.uuid5(uuid.NAMESPACE_URL,
                                               href_encoded))
+        # NOTE(kaifeng) The ".converted" suffix acts as an indicator that the
+        # image cached has gone through the conversion logic.
+        if force_raw:
+            master_file_name = master_file_name + '.converted'
+
         master_path = os.path.join(self.master_dir, master_file_name)
 
         if CONF.parallel_image_downloads:
diff --git a/ironic/drivers/modules/iscsi_deploy.py b/ironic/drivers/modules/iscsi_deploy.py
index 90a18a9e0e..bfedd37499 100644
--- a/ironic/drivers/modules/iscsi_deploy.py
+++ b/ironic/drivers/modules/iscsi_deploy.py
@@ -13,21 +13,17 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
-import os
-
 from ironic_lib import disk_utils
 from ironic_lib import metrics_utils
 from ironic_lib import utils as il_utils
 from oslo_log import log as logging
 from oslo_utils import excutils
-from oslo_utils import fileutils
 from six.moves.urllib import parse
 
 from ironic.common import dhcp_factory
 from ironic.common import exception
 from ironic.common.i18n import _
 from ironic.common import states
-from ironic.common import utils
 from ironic.conductor import task_manager
 from ironic.conductor import utils as manager_utils
 from ironic.conf import CONF
@@ -35,7 +31,6 @@ from ironic.drivers import base
 from ironic.drivers.modules import agent_base_vendor
 from ironic.drivers.modules import boot_mode_utils
 from ironic.drivers.modules import deploy_utils
-from ironic.drivers.modules import image_cache
 
 LOG = logging.getLogger(__name__)
 
@@ -44,28 +39,6 @@ METRICS = metrics_utils.get_metrics_logger(__name__)
 DISK_LAYOUT_PARAMS = ('root_gb', 'swap_mb', 'ephemeral_gb')
 
 
-@image_cache.cleanup(priority=50)
-class InstanceImageCache(image_cache.ImageCache):
-
-    def __init__(self):
-        super(self.__class__, self).__init__(
-            CONF.pxe.instance_master_path,
-            # MiB -> B
-            cache_size=CONF.pxe.image_cache_size * 1024 * 1024,
-            # min -> sec
-            cache_ttl=CONF.pxe.image_cache_ttl * 60)
-
-
-def _get_image_dir_path(node_uuid):
-    """Generate the dir for an instances disk."""
-    return os.path.join(CONF.pxe.images_path, node_uuid)
-
-
-def _get_image_file_path(node_uuid):
-    """Generate the full path for an instances disk."""
-    return os.path.join(_get_image_dir_path(node_uuid), 'disk')
-
-
 def _save_disk_layout(node, i_info):
     """Saves the disk layout.
 
@@ -101,7 +74,7 @@ def check_image_size(task):
         return
 
     i_info = deploy_utils.parse_instance_info(task.node)
-    image_path = _get_image_file_path(task.node.uuid)
+    image_path = deploy_utils._get_image_file_path(task.node.uuid)
     image_mb = disk_utils.get_image_mb(image_path)
     root_mb = 1024 * int(i_info['root_gb'])
     if image_mb > root_mb:
@@ -111,43 +84,6 @@ def check_image_size(task):
         raise exception.InstanceDeployFailure(msg)
 
 
-@METRICS.timer('cache_instance_image')
-def cache_instance_image(ctx, node):
-    """Fetch the instance's image from Glance
-
-    This method pulls the AMI and writes them to the appropriate place
-    on local disk.
-
-    :param ctx: context
-    :param node: an ironic node object
-    :returns: a tuple containing the uuid of the image and the path in
-        the filesystem where image is cached.
-    """
-    i_info = deploy_utils.parse_instance_info(node)
-    fileutils.ensure_tree(_get_image_dir_path(node.uuid))
-    image_path = _get_image_file_path(node.uuid)
-    uuid = i_info['image_source']
-
-    LOG.debug("Fetching image %(ami)s for node %(uuid)s",
-              {'ami': uuid, 'uuid': node.uuid})
-
-    deploy_utils.fetch_images(ctx, InstanceImageCache(), [(uuid, image_path)],
-                              CONF.force_raw_images)
-
-    return (uuid, image_path)
-
-
-@METRICS.timer('destroy_images')
-def destroy_images(node_uuid):
-    """Delete instance's image file.
-
-    :param node_uuid: the uuid of the ironic node.
-    """
-    il_utils.unlink_without_raise(_get_image_file_path(node_uuid))
-    utils.rmtree_without_raise(_get_image_dir_path(node_uuid))
-    InstanceImageCache().clean_up()
-
-
 @METRICS.timer('get_deploy_info')
 def get_deploy_info(node, address, iqn, port=None, lun='1'):
     """Returns the information required for doing iSCSI deploy in a dictionary.
@@ -169,7 +105,7 @@ def get_deploy_info(node, address, iqn, port=None, lun='1'):
         'port': port or CONF.iscsi.portal_port,
         'iqn': iqn,
         'lun': lun,
-        'image_path': _get_image_file_path(node.uuid),
+        'image_path': deploy_utils._get_image_file_path(node.uuid),
         'node_uuid': node.uuid}
 
     is_whole_disk_image = node.driver_internal_info['is_whole_disk_image']
@@ -236,7 +172,7 @@ def continue_deploy(task, **kwargs):
                      'Error: %(error)s') %
                    {'instance': node.instance_uuid, 'error': msg})
         deploy_utils.set_failed_state(task, msg)
-        destroy_images(task.node.uuid)
+        deploy_utils.destroy_images(task.node.uuid)
         if raise_exception:
             raise exception.InstanceDeployFailure(msg)
 
@@ -283,7 +219,7 @@ def continue_deploy(task, **kwargs):
         # for any future rebuilds
         _save_disk_layout(node, deploy_utils.parse_instance_info(node))
 
-    destroy_images(node.uuid)
+    deploy_utils.destroy_images(node.uuid)
     return uuid_dict_returned
 
 
@@ -464,7 +400,7 @@ class ISCSIDeploy(AgentDeployMixin, base.DeployInterface):
         """
         node = task.node
         if task.driver.storage.should_write_image(task):
-            cache_instance_image(task.context, node)
+            deploy_utils.cache_instance_image(task.context, node)
             check_image_size(task)
             manager_utils.node_power_action(task, states.REBOOT)
 
@@ -561,7 +497,7 @@ class ISCSIDeploy(AgentDeployMixin, base.DeployInterface):
 
         :param task: a TaskManager instance containing the node to act on.
         """
-        destroy_images(task.node.uuid)
+        deploy_utils.destroy_images(task.node.uuid)
         task.driver.boot.clean_up_ramdisk(task)
         task.driver.boot.clean_up_instance(task)
         provider = dhcp_factory.DHCPFactory()
diff --git a/ironic/tests/unit/drivers/modules/test_agent.py b/ironic/tests/unit/drivers/modules/test_agent.py
index dd8a541708..005763eba5 100644
--- a/ironic/tests/unit/drivers/modules/test_agent.py
+++ b/ironic/tests/unit/drivers/modules/test_agent.py
@@ -138,6 +138,30 @@ class TestAgentMethods(db_base.DbTestCase):
                               task, 'fake-image')
             show_mock.assert_called_once_with(self.context, 'fake-image')
 
+    @mock.patch.object(deploy_utils, 'check_for_missing_params')
+    def test_validate_http_provisioning_not_glance(self, utils_mock):
+        agent.validate_http_provisioning_configuration(self.node)
+        utils_mock.assert_not_called()
+
+    @mock.patch.object(deploy_utils, 'check_for_missing_params')
+    def test_validate_http_provisioning_not_http(self, utils_mock):
+        i_info = self.node.instance_info
+        i_info['image_source'] = '0448fa34-4db1-407b-a051-6357d5f86c59'
+        self.node.instance_info = i_info
+        agent.validate_http_provisioning_configuration(self.node)
+        utils_mock.assert_not_called()
+
+    def test_validate_http_provisioning_missing_args(self):
+        CONF.set_override('image_download_source', 'http', group='agent')
+        CONF.set_override('http_url', None, group='deploy')
+        i_info = self.node.instance_info
+        i_info['image_source'] = '0448fa34-4db1-407b-a051-6357d5f86c59'
+        self.node.instance_info = i_info
+        self.assertRaisesRegex(exception.MissingParameterValue,
+                               'failed to validate http provisoning',
+                               agent.validate_http_provisioning_configuration,
+                               self.node)
+
 
 class TestAgentDeploy(db_base.DbTestCase):
     def setUp(self):
@@ -164,12 +188,14 @@ class TestAgentDeploy(db_base.DbTestCase):
         expected = agent.COMMON_PROPERTIES
         self.assertEqual(expected, self.driver.get_properties())
 
+    @mock.patch.object(agent, 'validate_http_provisioning_configuration',
+                       autospec=True)
     @mock.patch.object(deploy_utils, 'validate_capabilities',
                        spec_set=True, autospec=True)
     @mock.patch.object(images, 'image_show', autospec=True)
     @mock.patch.object(pxe.PXEBoot, 'validate', autospec=True)
     def test_validate(self, pxe_boot_validate_mock, show_mock,
-                      validate_capability_mock):
+                      validate_capability_mock, validate_http_mock):
         with task_manager.acquire(
                 self.context, self.node['uuid'], shared=False) as task:
             self.driver.validate(task)
@@ -177,14 +203,17 @@ class TestAgentDeploy(db_base.DbTestCase):
                 task.driver.boot, task)
             show_mock.assert_called_once_with(self.context, 'fake-image')
             validate_capability_mock.assert_called_once_with(task.node)
+            validate_http_mock.assert_called_once_with(task.node)
 
+    @mock.patch.object(agent, 'validate_http_provisioning_configuration',
+                       autospec=True)
     @mock.patch.object(deploy_utils, 'validate_capabilities',
                        spec_set=True, autospec=True)
     @mock.patch.object(images, 'image_show', autospec=True)
     @mock.patch.object(pxe.PXEBoot, 'validate', autospec=True)
     def test_validate_driver_info_manage_agent_boot_false(
             self, pxe_boot_validate_mock, show_mock,
-            validate_capability_mock):
+            validate_capability_mock, validate_http_mock):
 
         self.config(manage_agent_boot=False, group='agent')
         self.node.driver_info = {}
@@ -195,6 +224,7 @@ class TestAgentDeploy(db_base.DbTestCase):
             self.assertFalse(pxe_boot_validate_mock.called)
             show_mock.assert_called_once_with(self.context, 'fake-image')
             validate_capability_mock.assert_called_once_with(task.node)
+            validate_http_mock.assert_called_once_with(task.node)
 
     @mock.patch.object(pxe.PXEBoot, 'validate', autospec=True)
     def test_validate_instance_info_missing_params(
@@ -226,10 +256,12 @@ class TestAgentDeploy(db_base.DbTestCase):
             pxe_boot_validate_mock.assert_called_once_with(
                 task.driver.boot, task)
 
+    @mock.patch.object(agent, 'validate_http_provisioning_configuration',
+                       autospec=True)
     @mock.patch.object(images, 'image_show', autospec=True)
     @mock.patch.object(pxe.PXEBoot, 'validate', autospec=True)
     def test_validate_invalid_root_device_hints(
-            self, pxe_boot_validate_mock, show_mock):
+            self, pxe_boot_validate_mock, show_mock, validate_http_mock):
         with task_manager.acquire(self.context, self.node.uuid,
                                   shared=True) as task:
             task.node.properties['root_device'] = {'size': 'not-int'}
@@ -238,10 +270,14 @@ class TestAgentDeploy(db_base.DbTestCase):
             pxe_boot_validate_mock.assert_called_once_with(
                 task.driver.boot, task)
             show_mock.assert_called_once_with(self.context, 'fake-image')
+            validate_http_mock.assert_called_once_with(task.node)
 
+    @mock.patch.object(agent, 'validate_http_provisioning_configuration',
+                       autospec=True)
     @mock.patch.object(images, 'image_show', autospec=True)
     @mock.patch.object(pxe.PXEBoot, 'validate', autospec=True)
-    def test_validate_invalid_proxies(self, pxe_boot_validate_mock, show_mock):
+    def test_validate_invalid_proxies(self, pxe_boot_validate_mock, show_mock,
+                                      validate_http_mock):
         with task_manager.acquire(self.context, self.node.uuid,
                                   shared=True) as task:
             task.node.driver_info.update({
@@ -254,6 +290,7 @@ class TestAgentDeploy(db_base.DbTestCase):
             pxe_boot_validate_mock.assert_called_once_with(
                 task.driver.boot, task)
             show_mock.assert_called_once_with(self.context, 'fake-image')
+            validate_http_mock.assert_called_once_with(task.node)
 
     @mock.patch.object(pxe.PXEBoot, 'validate', autospec=True)
     @mock.patch.object(deploy_utils, 'check_for_missing_params',
@@ -948,6 +985,8 @@ class TestAgentDeploy(db_base.DbTestCase):
             self.assertEqual(states.ACTIVE,
                              task.node.target_provision_state)
 
+    @mock.patch.object(deploy_utils, 'remove_http_instance_symlink',
+                       autospec=True)
     @mock.patch.object(agent.LOG, 'warning', spec_set=True, autospec=True)
     @mock.patch.object(agent.AgentDeployMixin, '_get_uuid_from_result',
                        autospec=True)
@@ -963,8 +1002,9 @@ class TestAgentDeploy(db_base.DbTestCase):
     def test_reboot_to_instance(self, check_deploy_mock,
                                 prepare_instance_mock, power_off_mock,
                                 get_power_state_mock, node_power_action_mock,
-                                uuid_mock, log_mock):
+                                uuid_mock, log_mock, remove_symlink_mock):
         self.config(manage_agent_boot=True, group='agent')
+        self.config(image_download_source='http', group='agent')
         check_deploy_mock.return_value = None
         uuid_mock.return_value = None
         self.node.provision_state = states.DEPLOYWAIT
@@ -990,6 +1030,7 @@ class TestAgentDeploy(db_base.DbTestCase):
                 task, states.POWER_ON)
             self.assertEqual(states.ACTIVE, task.node.provision_state)
             self.assertEqual(states.NOSTATE, task.node.target_provision_state)
+            self.assertTrue(remove_symlink_mock.called)
 
     @mock.patch.object(agent.LOG, 'warning', spec_set=True, autospec=True)
     @mock.patch.object(manager_utils, 'node_set_boot_device', autospec=True)
diff --git a/ironic/tests/unit/drivers/modules/test_deploy_utils.py b/ironic/tests/unit/drivers/modules/test_deploy_utils.py
index 3272da549e..fa345cf03b 100644
--- a/ironic/tests/unit/drivers/modules/test_deploy_utils.py
+++ b/ironic/tests/unit/drivers/modules/test_deploy_utils.py
@@ -19,10 +19,12 @@ import tempfile
 import time
 import types
 
+import fixtures
 from ironic_lib import disk_utils
 import mock
 from oslo_concurrency import processutils
 from oslo_config import cfg
+from oslo_utils import fileutils
 from oslo_utils import uuidutils
 import testtools
 from testtools import matchers
@@ -1718,6 +1720,42 @@ class AgentMethodsTestCase(db_base.DbTestCase):
         self.assertEqual('https://api-url', options['ipa-api-url'])
         self.assertEqual(0, options['coreos.configdrive'])
 
+    def test_direct_deploy_should_convert_raw_image_true(self):
+        cfg.CONF.set_override('force_raw_images', True)
+        cfg.CONF.set_override('stream_raw_images', True, group='agent')
+        internal_info = self.node.driver_internal_info
+        internal_info['is_whole_disk_image'] = True
+        self.node.driver_internal_info = internal_info
+        self.assertTrue(
+            utils.direct_deploy_should_convert_raw_image(self.node))
+
+    def test_direct_deploy_should_convert_raw_image_no_force_raw(self):
+        cfg.CONF.set_override('force_raw_images', False)
+        cfg.CONF.set_override('stream_raw_images', True, group='agent')
+        internal_info = self.node.driver_internal_info
+        internal_info['is_whole_disk_image'] = True
+        self.node.driver_internal_info = internal_info
+        self.assertFalse(
+            utils.direct_deploy_should_convert_raw_image(self.node))
+
+    def test_direct_deploy_should_convert_raw_image_no_stream(self):
+        cfg.CONF.set_override('force_raw_images', True)
+        cfg.CONF.set_override('stream_raw_images', False, group='agent')
+        internal_info = self.node.driver_internal_info
+        internal_info['is_whole_disk_image'] = True
+        self.node.driver_internal_info = internal_info
+        self.assertFalse(
+            utils.direct_deploy_should_convert_raw_image(self.node))
+
+    def test_direct_deploy_should_convert_raw_image_partition(self):
+        cfg.CONF.set_override('force_raw_images', True)
+        cfg.CONF.set_override('stream_raw_images', True, group='agent')
+        internal_info = self.node.driver_internal_info
+        internal_info['is_whole_disk_image'] = False
+        self.node.driver_internal_info = internal_info
+        self.assertFalse(
+            utils.direct_deploy_should_convert_raw_image(self.node))
+
 
 @mock.patch.object(disk_utils, 'is_block_device', autospec=True)
 @mock.patch.object(utils, 'login_iscsi', lambda *_: None)
@@ -2383,6 +2421,118 @@ class TestBuildInstanceInfoForDeploy(db_base.DbTestCase):
                               utils.build_instance_info_for_deploy, task)
 
 
+class TestBuildInstanceInfoForHttpProvisioning(db_base.DbTestCase):
+    def setUp(self):
+        super(TestBuildInstanceInfoForHttpProvisioning, self).setUp()
+        self.node = obj_utils.create_test_node(self.context,
+                                               boot_interface='pxe',
+                                               deploy_interface='direct')
+        i_info = self.node.instance_info
+        i_info['image_source'] = '733d1c44-a2ea-414b-aca7-69decf20d810'
+        i_info['root_gb'] = 100
+        driver_internal_info = self.node.driver_internal_info
+        driver_internal_info['is_whole_disk_image'] = True
+        self.node.driver_internal_info = driver_internal_info
+        self.node.instance_info = i_info
+        self.node.save()
+
+        self.md5sum_mock = self.useFixture(fixtures.MockPatchObject(
+            fileutils, 'compute_file_checksum')).mock
+        self.md5sum_mock.return_value = 'fake md5'
+        self.cache_image_mock = self.useFixture(fixtures.MockPatchObject(
+            utils, 'cache_instance_image', autospec=True)).mock
+        self.cache_image_mock.return_value = (
+            '733d1c44-a2ea-414b-aca7-69decf20d810',
+            '/var/lib/ironic/images/{}/disk'.format(self.node.uuid))
+        self.ensure_tree_mock = self.useFixture(fixtures.MockPatchObject(
+            utils.fileutils, 'ensure_tree', autospec=True)).mock
+        self.create_link_mock = self.useFixture(fixtures.MockPatchObject(
+            common_utils, 'create_link_without_raise', autospec=True)).mock
+
+        cfg.CONF.set_override('http_url', 'http://172.172.24.10:8080',
+                              group='deploy')
+        cfg.CONF.set_override('image_download_source', 'http', group='agent')
+
+        self.expected_url = '/'.join([cfg.CONF.deploy.http_url,
+                                     cfg.CONF.deploy.http_image_subdir,
+                                     self.node.uuid])
+
+    @mock.patch.object(image_service.HttpImageService, 'validate_href',
+                       autospec=True)
+    @mock.patch.object(image_service, 'GlanceImageService', autospec=True)
+    def test_build_instance_info_no_force_raw(self, glance_mock,
+                                              validate_mock):
+        cfg.CONF.set_override('force_raw_images', False)
+
+        image_info = {'checksum': 'aa', 'disk_format': 'qcow2',
+                      'container_format': 'bare', 'properties': {}}
+        glance_mock.return_value.show = mock.MagicMock(spec_set=[],
+                                                       return_value=image_info)
+
+        with task_manager.acquire(
+                self.context, self.node.uuid, shared=False) as task:
+
+            instance_info = utils.build_instance_info_for_deploy(task)
+
+            glance_mock.assert_called_once_with(version=2,
+                                                context=task.context)
+            glance_mock.return_value.show.assert_called_once_with(
+                self.node.instance_info['image_source'])
+            self.cache_image_mock.assert_called_once_with(task.context,
+                                                          task.node,
+                                                          force_raw=False)
+            symlink_dir = utils._get_http_image_symlink_dir_path()
+            symlink_file = utils._get_http_image_symlink_file_path(
+                self.node.uuid)
+            image_path = utils._get_image_file_path(self.node.uuid)
+            self.ensure_tree_mock.assert_called_once_with(symlink_dir)
+            self.create_link_mock.assert_called_once_with(image_path,
+                                                          symlink_file)
+            self.assertEqual(instance_info['image_checksum'], 'aa')
+            self.assertEqual(instance_info['image_disk_format'], 'qcow2')
+            self.md5sum_mock.assert_not_called()
+            validate_mock.assert_called_once_with(mock.ANY, self.expected_url,
+                                                  secret=True)
+
+    @mock.patch.object(image_service.HttpImageService, 'validate_href',
+                       autospec=True)
+    @mock.patch.object(image_service, 'GlanceImageService', autospec=True)
+    def test_build_instance_info_force_raw(self, glance_mock,
+                                           validate_mock):
+        cfg.CONF.set_override('force_raw_images', True)
+
+        image_info = {'checksum': 'aa', 'disk_format': 'qcow2',
+                      'container_format': 'bare', 'properties': {}}
+        glance_mock.return_value.show = mock.MagicMock(spec_set=[],
+                                                       return_value=image_info)
+
+        with task_manager.acquire(
+                self.context, self.node.uuid, shared=False) as task:
+
+            instance_info = utils.build_instance_info_for_deploy(task)
+
+            glance_mock.assert_called_once_with(version=2,
+                                                context=task.context)
+            glance_mock.return_value.show.assert_called_once_with(
+                self.node.instance_info['image_source'])
+            self.cache_image_mock.assert_called_once_with(task.context,
+                                                          task.node,
+                                                          force_raw=True)
+            symlink_dir = utils._get_http_image_symlink_dir_path()
+            symlink_file = utils._get_http_image_symlink_file_path(
+                self.node.uuid)
+            image_path = utils._get_image_file_path(self.node.uuid)
+            self.ensure_tree_mock.assert_called_once_with(symlink_dir)
+            self.create_link_mock.assert_called_once_with(image_path,
+                                                          symlink_file)
+            self.assertEqual(instance_info['image_checksum'], 'fake md5')
+            self.assertEqual(instance_info['image_disk_format'], 'raw')
+            self.md5sum_mock.assert_called_once_with(image_path,
+                                                     algorithm='md5')
+            validate_mock.assert_called_once_with(mock.ANY, self.expected_url,
+                                                  secret=True)
+
+
 class TestStorageInterfaceUtils(db_base.DbTestCase):
     def setUp(self):
         super(TestStorageInterfaceUtils, self).setUp()
diff --git a/ironic/tests/unit/drivers/modules/test_image_cache.py b/ironic/tests/unit/drivers/modules/test_image_cache.py
index b50d3fe40d..68a1d95e01 100644
--- a/ironic/tests/unit/drivers/modules/test_image_cache.py
+++ b/ironic/tests/unit/drivers/modules/test_image_cache.py
@@ -47,7 +47,8 @@ class TestImageCacheFetch(base.TestCase):
         self.dest_dir = tempfile.mkdtemp()
         self.dest_path = os.path.join(self.dest_dir, 'dest')
         self.uuid = uuidutils.generate_uuid()
-        self.master_path = os.path.join(self.master_dir, self.uuid)
+        self.master_path = ''.join([os.path.join(self.master_dir, self.uuid),
+                                    '.converted'])
 
     @mock.patch.object(image_cache, '_fetch', autospec=True)
     @mock.patch.object(image_cache.ImageCache, 'clean_up', autospec=True)
@@ -81,6 +82,26 @@ class TestImageCacheFetch(base.TestCase):
         self.assertFalse(mock_download.called)
         self.assertFalse(mock_clean_up.called)
 
+    @mock.patch.object(image_cache.ImageCache, 'clean_up', autospec=True)
+    @mock.patch.object(image_cache.ImageCache, '_download_image',
+                       autospec=True)
+    @mock.patch.object(os, 'link', autospec=True)
+    @mock.patch.object(image_cache, '_delete_dest_path_if_stale',
+                       return_value=True, autospec=True)
+    @mock.patch.object(image_cache, '_delete_master_path_if_stale',
+                       return_value=True, autospec=True)
+    def test_fetch_image_dest_and_master_uptodate_no_force_raw(
+            self, mock_cache_upd, mock_dest_upd, mock_link, mock_download,
+            mock_clean_up):
+        master_path = os.path.join(self.master_dir, self.uuid)
+        self.cache.fetch_image(self.uuid, self.dest_path, force_raw=False)
+        mock_cache_upd.assert_called_once_with(master_path, self.uuid,
+                                               None)
+        mock_dest_upd.assert_called_once_with(master_path, self.dest_path)
+        self.assertFalse(mock_link.called)
+        self.assertFalse(mock_download.called)
+        self.assertFalse(mock_clean_up.called)
+
     @mock.patch.object(image_cache.ImageCache, 'clean_up', autospec=True)
     @mock.patch.object(image_cache.ImageCache, '_download_image',
                        autospec=True)
@@ -149,13 +170,29 @@ class TestImageCacheFetch(base.TestCase):
         href = u'http://abc.com/ubuntu.qcow2'
         href_encoded = href.encode('utf-8') if six.PY2 else href
         href_converted = str(uuid.uuid5(uuid.NAMESPACE_URL, href_encoded))
-        master_path = os.path.join(self.master_dir, href_converted)
+        master_path = ''.join([os.path.join(self.master_dir, href_converted),
+                               '.converted'])
         self.cache.fetch_image(href, self.dest_path)
         mock_download.assert_called_once_with(
             self.cache, href, master_path, self.dest_path,
             ctx=None, force_raw=True)
         self.assertTrue(mock_clean_up.called)
 
+    @mock.patch.object(image_cache.ImageCache, 'clean_up', autospec=True)
+    @mock.patch.object(image_cache.ImageCache, '_download_image',
+                       autospec=True)
+    def test_fetch_image_not_uuid_no_force_raw(self, mock_download,
+                                               mock_clean_up):
+        href = u'http://abc.com/ubuntu.qcow2'
+        href_encoded = href.encode('utf-8') if six.PY2 else href
+        href_converted = str(uuid.uuid5(uuid.NAMESPACE_URL, href_encoded))
+        master_path = os.path.join(self.master_dir, href_converted)
+        self.cache.fetch_image(href, self.dest_path, force_raw=False)
+        mock_download.assert_called_once_with(
+            self.cache, href, master_path, self.dest_path,
+            ctx=None, force_raw=False)
+        self.assertTrue(mock_clean_up.called)
+
     @mock.patch.object(image_cache, '_fetch', autospec=True)
     def test__download_image(self, mock_fetch):
         def _fake_fetch(ctx, uuid, tmp_path, *args):
diff --git a/ironic/tests/unit/drivers/modules/test_iscsi_deploy.py b/ironic/tests/unit/drivers/modules/test_iscsi_deploy.py
index f02c7e0859..b6d14f357a 100644
--- a/ironic/tests/unit/drivers/modules/test_iscsi_deploy.py
+++ b/ironic/tests/unit/drivers/modules/test_iscsi_deploy.py
@@ -83,13 +83,13 @@ class IscsiDeployPrivateMethodsTestCase(db_base.DbTestCase):
     def test__get_image_dir_path(self):
         self.assertEqual(os.path.join(CONF.pxe.images_path,
                                       self.node.uuid),
-                         iscsi_deploy._get_image_dir_path(self.node.uuid))
+                         deploy_utils._get_image_dir_path(self.node.uuid))
 
     def test__get_image_file_path(self):
         self.assertEqual(os.path.join(CONF.pxe.images_path,
                                       self.node.uuid,
                                       'disk'),
-                         iscsi_deploy._get_image_file_path(self.node.uuid))
+                         deploy_utils._get_image_file_path(self.node.uuid))
 
 
 class IscsiDeployMethodsTestCase(db_base.DbTestCase):
@@ -115,7 +115,7 @@ class IscsiDeployMethodsTestCase(db_base.DbTestCase):
             task.node.instance_info['root_gb'] = 1
             iscsi_deploy.check_image_size(task)
             get_image_mb_mock.assert_called_once_with(
-                iscsi_deploy._get_image_file_path(task.node.uuid))
+                deploy_utils._get_image_file_path(task.node.uuid))
 
     @mock.patch.object(disk_utils, 'get_image_mb', autospec=True)
     def test_check_image_size_whole_disk_image(self, get_image_mb_mock):
@@ -138,7 +138,7 @@ class IscsiDeployMethodsTestCase(db_base.DbTestCase):
                               iscsi_deploy.check_image_size,
                               task)
             get_image_mb_mock.assert_called_once_with(
-                iscsi_deploy._get_image_file_path(task.node.uuid))
+                deploy_utils._get_image_file_path(task.node.uuid))
 
     @mock.patch.object(deploy_utils, 'fetch_images', autospec=True)
     def test_cache_instance_images_master_path(self, mock_fetch_image):
@@ -149,7 +149,7 @@ class IscsiDeployMethodsTestCase(db_base.DbTestCase):
                     group='pxe')
         fileutils.ensure_tree(CONF.pxe.instance_master_path)
 
-        (uuid, image_path) = iscsi_deploy.cache_instance_image(None, self.node)
+        (uuid, image_path) = deploy_utils.cache_instance_image(None, self.node)
         mock_fetch_image.assert_called_once_with(None,
                                                  mock.ANY,
                                                  [(uuid, image_path)], True)
@@ -161,11 +161,11 @@ class IscsiDeployMethodsTestCase(db_base.DbTestCase):
 
     @mock.patch.object(ironic_utils, 'unlink_without_raise', autospec=True)
     @mock.patch.object(utils, 'rmtree_without_raise', autospec=True)
-    @mock.patch.object(iscsi_deploy, 'InstanceImageCache', autospec=True)
+    @mock.patch.object(deploy_utils, 'InstanceImageCache', autospec=True)
     def test_destroy_images(self, mock_cache, mock_rmtree, mock_unlink):
         self.config(images_path='/path', group='pxe')
 
-        iscsi_deploy.destroy_images('uuid')
+        deploy_utils.destroy_images('uuid')
 
         mock_cache.return_value.clean_up.assert_called_once_with()
         mock_unlink.assert_called_once_with('/path/uuid/disk')
@@ -173,7 +173,7 @@ class IscsiDeployMethodsTestCase(db_base.DbTestCase):
 
     @mock.patch.object(driver_utils, 'collect_ramdisk_logs', autospec=True)
     @mock.patch.object(iscsi_deploy, '_save_disk_layout', autospec=True)
-    @mock.patch.object(iscsi_deploy, 'InstanceImageCache', autospec=True)
+    @mock.patch.object(deploy_utils, 'InstanceImageCache', autospec=True)
     @mock.patch.object(manager_utils, 'node_power_action', autospec=True)
     @mock.patch.object(deploy_utils, 'deploy_partition_image', autospec=True)
     def test_continue_deploy_fail(
@@ -206,7 +206,7 @@ class IscsiDeployMethodsTestCase(db_base.DbTestCase):
 
     @mock.patch.object(driver_utils, 'collect_ramdisk_logs', autospec=True)
     @mock.patch.object(iscsi_deploy, '_save_disk_layout', autospec=True)
-    @mock.patch.object(iscsi_deploy, 'InstanceImageCache', autospec=True)
+    @mock.patch.object(deploy_utils, 'InstanceImageCache', autospec=True)
     @mock.patch.object(manager_utils, 'node_power_action', autospec=True)
     @mock.patch.object(deploy_utils, 'deploy_partition_image', autospec=True)
     def test_continue_deploy_unexpected_fail(
@@ -237,7 +237,7 @@ class IscsiDeployMethodsTestCase(db_base.DbTestCase):
 
     @mock.patch.object(driver_utils, 'collect_ramdisk_logs', autospec=True)
     @mock.patch.object(iscsi_deploy, '_save_disk_layout', autospec=True)
-    @mock.patch.object(iscsi_deploy, 'InstanceImageCache', autospec=True)
+    @mock.patch.object(deploy_utils, 'InstanceImageCache', autospec=True)
     @mock.patch.object(manager_utils, 'node_power_action', autospec=True)
     @mock.patch.object(deploy_utils, 'deploy_partition_image', autospec=True)
     def test_continue_deploy_fail_no_root_uuid_or_disk_id(
@@ -267,7 +267,7 @@ class IscsiDeployMethodsTestCase(db_base.DbTestCase):
 
     @mock.patch.object(driver_utils, 'collect_ramdisk_logs', autospec=True)
     @mock.patch.object(iscsi_deploy, '_save_disk_layout', autospec=True)
-    @mock.patch.object(iscsi_deploy, 'InstanceImageCache', autospec=True)
+    @mock.patch.object(deploy_utils, 'InstanceImageCache', autospec=True)
     @mock.patch.object(manager_utils, 'node_power_action', autospec=True)
     @mock.patch.object(deploy_utils, 'deploy_partition_image', autospec=True)
     def test_continue_deploy_fail_empty_root_uuid(
@@ -298,7 +298,7 @@ class IscsiDeployMethodsTestCase(db_base.DbTestCase):
     @mock.patch.object(iscsi_deploy, '_save_disk_layout', autospec=True)
     @mock.patch.object(iscsi_deploy, 'LOG', autospec=True)
     @mock.patch.object(iscsi_deploy, 'get_deploy_info', autospec=True)
-    @mock.patch.object(iscsi_deploy, 'InstanceImageCache', autospec=True)
+    @mock.patch.object(deploy_utils, 'InstanceImageCache', autospec=True)
     @mock.patch.object(manager_utils, 'node_power_action', autospec=True)
     @mock.patch.object(deploy_utils, 'deploy_partition_image', autospec=True)
     def test_continue_deploy(self, deploy_mock, power_mock, mock_image_cache,
@@ -350,7 +350,7 @@ class IscsiDeployMethodsTestCase(db_base.DbTestCase):
 
     @mock.patch.object(iscsi_deploy, 'LOG', autospec=True)
     @mock.patch.object(iscsi_deploy, 'get_deploy_info', autospec=True)
-    @mock.patch.object(iscsi_deploy, 'InstanceImageCache', autospec=True)
+    @mock.patch.object(deploy_utils, 'InstanceImageCache', autospec=True)
     @mock.patch.object(manager_utils, 'node_power_action', autospec=True)
     @mock.patch.object(deploy_utils, 'deploy_disk_image', autospec=True)
     def test_continue_deploy_whole_disk_image(
@@ -703,7 +703,7 @@ class ISCSIDeployTestCase(db_base.DbTestCase):
 
     @mock.patch.object(manager_utils, 'node_power_action', autospec=True)
     @mock.patch.object(iscsi_deploy, 'check_image_size', autospec=True)
-    @mock.patch.object(iscsi_deploy, 'cache_instance_image', autospec=True)
+    @mock.patch.object(deploy_utils, 'cache_instance_image', autospec=True)
     def test_deploy(self, mock_cache_instance_image,
                     mock_check_image_size, mock_node_power_action):
         with task_manager.acquire(self.context,
@@ -728,7 +728,7 @@ class ISCSIDeployTestCase(db_base.DbTestCase):
                        spec_set=True, autospec=True)
     @mock.patch.object(manager_utils, 'node_power_action', autospec=True)
     @mock.patch.object(iscsi_deploy, 'check_image_size', autospec=True)
-    @mock.patch.object(iscsi_deploy, 'cache_instance_image', autospec=True)
+    @mock.patch.object(deploy_utils, 'cache_instance_image', autospec=True)
     def test_deploy_storage_check_write_image_false(self,
                                                     mock_cache_instance_image,
                                                     mock_check_image_size,
@@ -790,7 +790,7 @@ class ISCSIDeployTestCase(db_base.DbTestCase):
     @mock.patch('ironic.common.dhcp_factory.DHCPFactory.clean_dhcp')
     @mock.patch.object(pxe.PXEBoot, 'clean_up_instance', autospec=True)
     @mock.patch.object(pxe.PXEBoot, 'clean_up_ramdisk', autospec=True)
-    @mock.patch.object(iscsi_deploy, 'destroy_images', autospec=True)
+    @mock.patch.object(deploy_utils, 'destroy_images', autospec=True)
     def test_clean_up(self, destroy_images_mock, clean_up_ramdisk_mock,
                       clean_up_instance_mock, clean_dhcp_mock,
                       set_dhcp_provider_mock):
@@ -982,9 +982,9 @@ class CleanUpFullFlowTestCase(db_base.DbTestCase):
         os.makedirs(self.node_tftp_dir)
         self.kernel_path = os.path.join(self.node_tftp_dir,
                                         'kernel')
-        self.node_image_dir = iscsi_deploy._get_image_dir_path(self.node.uuid)
+        self.node_image_dir = deploy_utils._get_image_dir_path(self.node.uuid)
         os.makedirs(self.node_image_dir)
-        self.image_path = iscsi_deploy._get_image_file_path(self.node.uuid)
+        self.image_path = deploy_utils._get_image_file_path(self.node.uuid)
         self.config_path = pxe_utils.get_pxe_config_file_path(self.node.uuid)
         self.mac_path = pxe_utils._get_pxe_mac_path(self.port.address)
 
diff --git a/releasenotes/notes/agent-http-provisioning-d116b3ff36669d16.yaml b/releasenotes/notes/agent-http-provisioning-d116b3ff36669d16.yaml
new file mode 100644
index 0000000000..14c6b675b3
--- /dev/null
+++ b/releasenotes/notes/agent-http-provisioning-d116b3ff36669d16.yaml
@@ -0,0 +1,15 @@
+---
+features:
+  - Adds the ability to provision with ``direct`` deploy interface and custom
+    HTTP service running at ironic conductor node. A new configuration option
+    ``[agent]image_download_source`` is introduced. When set to ``swift``,
+    the ``direct`` deploy interface uses tempurl generated via the Object
+    service as the source of instance image during provisioning, this is the
+    default configuration. When set to ``http``, the ``direct`` deploy
+    interface downloads instance image from the Image service, and caches the
+    image in the ironic conductor node. The cached instance images are
+    referenced by symbolic links located at subdirectory
+    ``[deploy]http_image_subdir`` under path ``[deploy]http_root``. The custom
+    HTTP server running at ironic conductor node is supposed to be configured
+    properly to make IPA has unauthenticated access to image URL described
+    above.
diff --git a/zuul.d/ironic-jobs.yaml b/zuul.d/ironic-jobs.yaml
index 691bcb17bf..6411a3fb87 100644
--- a/zuul.d/ironic-jobs.yaml
+++ b/zuul.d/ironic-jobs.yaml
@@ -247,3 +247,22 @@
         s-container: True
         s-object: True
         s-proxy: True
+
+- job:
+    name: ironic-tempest-dsvm-ipa-wholedisk-bios-agent_ipmitool-tinyipa-indirect
+    description: ironic-tempest-dsvm-ipa-wholedisk-bios-agent_ipmitool-tinyipa-indirect
+    parent: ironic-tempest-dsvm-ipa-wholedisk-bios-agent_ipmitool-tinyipa
+    timeout: 5400
+    vars:
+      devstack_localrc:
+        IRONIC_AGENT_IMAGE_DOWNLOAD_SOURCE: http
+
+- job:
+    name: ironic-tempest-dsvm-ipa-partition-bios-agent_ipmitool-tinyipa-indirect
+    description: ironic-tempest-dsvm-ipa-partition-bios-agent_ipmitool-tinyipa-indirect
+    parent: ironic-tempest-dsvm-ipa-wholedisk-bios-agent_ipmitool-tinyipa
+    timeout: 5400
+    vars:
+      devstack_localrc:
+        IRONIC_AGENT_IMAGE_DOWNLOAD_SOURCE: http
+        IRONIC_TEMPEST_WHOLE_DISK_IMAGE: False
diff --git a/zuul.d/project.yaml b/zuul.d/project.yaml
index 0abaf60c4a..2b14980edb 100644
--- a/zuul.d/project.yaml
+++ b/zuul.d/project.yaml
@@ -20,6 +20,8 @@
         - ironic-tempest-dsvm-ipa-partition-uefi-pxe_ipmitool-tinyipa
         - ironic-tempest-dsvm-ipa-wholedisk-agent_ipmitool-tinyipa-multinode
         - ironic-tempest-dsvm-ipa-wholedisk-bios-agent_ipmitool-tinyipa
+        - ironic-tempest-dsvm-ipa-wholedisk-bios-agent_ipmitool-tinyipa-indirect
+        - ironic-tempest-dsvm-ipa-partition-bios-agent_ipmitool-tinyipa-indirect
         # Non-voting jobs
         - ironic-tempest-dsvm-ipa-wholedisk-bios-pxe_snmp-tinyipa:
             voting: false
@@ -43,6 +45,8 @@
         - ironic-tempest-dsvm-ipa-partition-uefi-pxe_ipmitool-tinyipa
         - ironic-tempest-dsvm-ipa-wholedisk-agent_ipmitool-tinyipa-multinode
         - ironic-tempest-dsvm-ipa-wholedisk-bios-agent_ipmitool-tinyipa
+        - ironic-tempest-dsvm-ipa-wholedisk-bios-agent_ipmitool-tinyipa-indirect
+        - ironic-tempest-dsvm-ipa-partition-bios-agent_ipmitool-tinyipa-indirect
         - openstack-tox-lower-constraints
         - openstack-tox-cover
     experimental: