Merge "Add 'deploy steps' parameter for provisioning API"

23 files changed, 630 insertions, 128 deletions

diff --git a/api-ref/source/baremetal-api-v1-node-management.inc b/api-ref/source/baremetal-api-v1-node-management.inc
index 407d3c3f7..cae2c0689 100644
--- a/api-ref/source/baremetal-api-v1-node-management.inc
+++ b/api-ref/source/baremetal-api-v1-node-management.inc
@@ -359,6 +359,10 @@ detailed documentation of the Ironic State Machine is available
 .. versionadded:: 1.59
    A ``configdrive`` now accepts ``vendor_data``.
 
+.. versionadded:: 1.69
+   ``deploy_steps`` can be provided when settings the node's provision target
+   state to ``active`` or ``rebuild``.
+
 Normal response code: 202
 
 Error codes:
@@ -376,12 +380,17 @@ Request
     - target: req_provision_state
     - configdrive: configdrive
     - clean_steps: clean_steps
+    - deploy_steps: deploy_steps
     - rescue_password: rescue_password
 
 **Example request to deploy a Node, using a configdrive served via local webserver:**
 
 .. literalinclude:: samples/node-set-active-state.json
 
+**Example request to deploy a Node with custom deploy step:**
+
+.. literalinclude:: samples/node-set-active-state-deploy-steps.json
+
 **Example request to clean a Node, with custom clean step:**
 
 .. literalinclude:: samples/node-set-clean-state.json
diff --git a/api-ref/source/parameters.yaml b/api-ref/source/parameters.yaml
index 882223d6f..62d3f2ec4 100644
--- a/api-ref/source/parameters.yaml
+++ b/api-ref/source/parameters.yaml
@@ -727,6 +727,15 @@ deploy_step:
   in: body
   required: false
   type: string
+deploy_steps:
+  description: |
+    A list of deploy steps that will be performed on the node. A deploy step is
+    a dictionary with required keys 'interface', 'step', 'priority' and optional
+    key 'args'. If specified, the value for 'args' is a keyword variable
+    argument dictionary that is passed to the deploy step method.
+  in: body
+  required: False
+  type: array
 deploy_template_name:
   description: |
     The unique name of the deploy template.
diff --git a/api-ref/source/samples/node-set-active-state-deploy-steps.json b/api-ref/source/samples/node-set-active-state-deploy-steps.json
new file mode 100644
index 000000000..e5778d23a
--- /dev/null
+++ b/api-ref/source/samples/node-set-active-state-deploy-steps.json
@@ -0,0 +1,14 @@
+{
+    "target": "active",
+    "deploy_steps": [
+        {
+            "interface": "deploy",
+            "step": "upgrade_firmware",
+            "args": {
+                "force": "True"
+            },
+            "priority": 95
+        }
+    ]
+}
+
diff --git a/doc/source/admin/node-deployment.rst b/doc/source/admin/node-deployment.rst
index 0e6e6966c..d9e736a8c 100644
--- a/doc/source/admin/node-deployment.rst
+++ b/doc/source/admin/node-deployment.rst
@@ -89,6 +89,31 @@ More deploy steps can be provided by the ramdisk, see
 :ironic-python-agent-doc:`IPA hardware managers documentation
 <admin/hardware_managers.html>` for a listing.
 
+Requesting steps
+----------------
+
+Starting with Bare Metal API version 1.69 user can optionally supply deploy
+steps for node deployment when invoking deployment or rebuilding. Overlapping
+steps will take precedence over `Agent steps`_ and `Deploy Templates`_
+steps.
+
+Using "baremetal" client deploy steps can be passed via ``--deploy-steps``
+argument. The argument ``--deploy-steps`` is one of:
+
+- a JSON string
+- path to a JSON file whose contents are passed to the API
+- '-', to read from stdin. This allows piping in the deploy steps.
+
+An example by passing a JSON string:
+
+.. code-block:: console
+
+    baremetal node deploy <node> \
+        --deloy-steps '[{"interface": "bios", "step": "apply_configuration", "args": {"settings": [{"name": "LogicalProc", "value": "Enabled"}]}, "priority": 150}]'
+
+Format of JSON for deploy steps argument is described in `Deploy step format`_
+section.
+
 Writing a Deploy Step
 ---------------------
 
diff --git a/doc/source/contributor/webapi-version-history.rst b/doc/source/contributor/webapi-version-history.rst
index 15de0644e..960fdbdbd 100644
--- a/doc/source/contributor/webapi-version-history.rst
+++ b/doc/source/contributor/webapi-version-history.rst
@@ -2,6 +2,13 @@
 REST API Version History
 ========================
 
+1.69 (Wallaby, master)
+----------------------
+
+Add support for ``deploy-steps`` parameter to provisioning endpoint
+``/v1/nodes/{node_ident}/states/provision``. Available and optional when target
+is 'active' or 'rebuild'.
+
 1.68 (Victoria, 16.0)
 -----------------------
 
diff --git a/ironic/api/controllers/v1/deploy_template.py b/ironic/api/controllers/v1/deploy_template.py
index 9e774f948..e54b0d63b 100644
--- a/ironic/api/controllers/v1/deploy_template.py
+++ b/ironic/api/controllers/v1/deploy_template.py
@@ -30,7 +30,6 @@ from ironic.api import method
 from ironic.common import args
 from ironic.common import exception
 from ironic.common.i18n import _
-from ironic.conductor import steps as conductor_steps
 import ironic.conf
 from ironic import objects
 
@@ -40,30 +39,14 @@ METRICS = metrics_utils.get_metrics_logger(__name__)
 
 DEFAULT_RETURN_FIELDS = ['uuid', 'name']
 
-INTERFACE_NAMES = list(conductor_steps.DEPLOYING_INTERFACE_PRIORITY)
-
-STEP_SCHEMA = {
-    'type': 'object',
-    'properties': {
-        'args': {'type': 'object'},
-        'interface': {'type': 'string', 'enum': INTERFACE_NAMES},
-        'priority': {'anyOf': [
-            {'type': 'integer', 'minimum': 0},
-            {'type': 'string', 'minLength': 1, 'pattern': '^[0-9]+$'}
-        ]},
-        'step': {'type': 'string', 'minLength': 1},
-    },
-    'required': ['interface', 'step', 'args', 'priority'],
-    'additionalProperties': False,
-}
-
 TEMPLATE_SCHEMA = {
     'type': 'object',
     'properties': {
         'description': {'type': ['string', 'null'], 'maxLength': 255},
         'extra': {'type': ['object', 'null']},
         'name': api_utils.TRAITS_SCHEMA,
-        'steps': {'type': 'array', 'items': STEP_SCHEMA, 'minItems': 1},
+        'steps': {'type': 'array', 'items': api_utils.DEPLOY_STEP_SCHEMA,
+                  'minItems': 1},
         'uuid': {'type': ['string', 'null']},
     },
     'required': ['steps', 'name'],
@@ -307,7 +290,7 @@ class DeployTemplatesController(rest.RestController):
         # validate the result with the patch schema
         for step in template.get('steps', []):
             api_utils.patched_validate_with_schema(
-                step, STEP_SCHEMA)
+                step, api_utils.DEPLOY_STEP_SCHEMA)
         api_utils.patched_validate_with_schema(
             template, TEMPLATE_SCHEMA, TEMPLATE_VALIDATOR)
 
diff --git a/ironic/api/controllers/v1/node.py b/ironic/api/controllers/v1/node.py
index e6b444c30..de87d566d 100644
--- a/ironic/api/controllers/v1/node.py
+++ b/ironic/api/controllers/v1/node.py
@@ -84,6 +84,13 @@ _CLEAN_STEPS_SCHEMA = {
     }
 }
 
+_DEPLOY_STEPS_SCHEMA = {
+    "$schema": "http://json-schema.org/schema#",
+    "title": "Deploy steps schema",
+    "type": "array",
+    "items": api_utils.DEPLOY_STEP_SCHEMA
+}
+
 METRICS = metrics_utils.get_metrics_logger(__name__)
 
 # Vendor information for node's driver:
@@ -784,18 +791,22 @@ class NodeStatesController(rest.RestController):
         api.response.location = link.build_url('nodes', url_args)
 
     def _do_provision_action(self, rpc_node, target, configdrive=None,
-                             clean_steps=None, rescue_password=None):
+                             clean_steps=None, deploy_steps=None,
+                             rescue_password=None):
         topic = api.request.rpcapi.get_topic_for(rpc_node)
         # Note that there is a race condition. The node state(s) could change
         # by the time the RPC call is made and the TaskManager manager gets a
         # lock.
         if target in (ir_states.ACTIVE, ir_states.REBUILD):
             rebuild = (target == ir_states.REBUILD)
+            if deploy_steps:
+                _check_deploy_steps(deploy_steps)
             api.request.rpcapi.do_node_deploy(context=api.request.context,
                                               node_id=rpc_node.uuid,
                                               rebuild=rebuild,
                                               configdrive=configdrive,
-                                              topic=topic)
+                                              topic=topic,
+                                              deploy_steps=deploy_steps)
         elif (target == ir_states.VERBS['unrescue']):
             api.request.rpcapi.do_node_unrescue(
                 api.request.context, rpc_node.uuid, topic)
@@ -836,9 +847,11 @@ class NodeStatesController(rest.RestController):
     @args.validate(node_ident=args.uuid_or_name, target=args.string,
                    configdrive=args.types(type(None), dict, str),
                    clean_steps=args.types(type(None), list),
+                   deploy_steps=args.types(type(None), list),
                    rescue_password=args.string)
     def provision(self, node_ident, target, configdrive=None,
-                  clean_steps=None, rescue_password=None):
+                  clean_steps=None, deploy_steps=None,
+                  rescue_password=None):
         """Asynchronous trigger the provisioning of the node.
 
         This will set the target provision state of the node, and a
@@ -871,6 +884,27 @@ class NodeStatesController(rest.RestController):
                 'args': {'force': True} }
 
             This is required (and only valid) when target is "clean".
+        :param deploy_steps: A list of deploy steps that will be performed on
+            the node. A deploy step is a dictionary with required keys
+            'interface', 'step', 'priority' and 'args'. If specified, the value
+            for 'args' is a keyword variable argument dictionary that is passed
+            to the deploy step method.::
+
+              { 'interface': <driver_interface>,
+                'step': <name_of_deploy_step>,
+                'args': {<arg1>: <value1>, ..., <argn>: <valuen>}
+                'priority': <integer>}
+
+            For example (this isn't a real example, this deploy step doesn't
+            exist)::
+
+              { 'interface': 'deploy',
+                'step': 'upgrade_firmware',
+                'args': {'force': True},
+                'priority': 90 }
+
+            This is used only when target is "active" or "rebuild" and is
+            optional.
         :param rescue_password: A string representing the password to be set
             inside the rescue environment. This is required (and only valid),
             when target is "rescue".
@@ -878,7 +912,7 @@ class NodeStatesController(rest.RestController):
         :raises: ClientSideError (HTTP 409) if the node is already being
                  provisioned.
         :raises: InvalidParameterValue (HTTP 400), if validation of
-                 clean_steps or power driver interface fails.
+                 clean_steps, deploy_steps or power driver interface fails.
         :raises: InvalidStateRequested (HTTP 400) if the requested transition
                  is not possible from the current state.
         :raises: NodeInMaintenance (HTTP 400), if operation cannot be
@@ -923,6 +957,8 @@ class NodeStatesController(rest.RestController):
             raise exception.ClientSideError(
                 msg, status_code=http_client.BAD_REQUEST)
 
+        api_utils.check_allow_deploy_steps(target, deploy_steps)
+
         if (rescue_password is not None
             and target != ir_states.VERBS['rescue']):
             msg = (_('"rescue_password" is only valid when setting target '
@@ -936,7 +972,7 @@ class NodeStatesController(rest.RestController):
                 raise exception.NotAcceptable()
 
         self._do_provision_action(rpc_node, target, configdrive, clean_steps,
-                                  rescue_password)
+                                  deploy_steps, rescue_password)
 
         # Set the HTTP Location Header
         url_args = '/'.join([node_ident, 'states'])
@@ -944,20 +980,43 @@ class NodeStatesController(rest.RestController):
 
 
 def _check_clean_steps(clean_steps):
-    """Ensure all necessary keys are present and correct in clean steps.
-
-    Check that the user-specified clean steps are in the expected format and
-    include the required information.
+    """Ensure all necessary keys are present and correct in steps for clean
 
-    :param clean_steps: a list of clean steps. For more details, see the
+    :param clean_steps: a list of steps. For more details, see the
         clean_steps parameter of :func:`NodeStatesController.provision`.
-    :raises: InvalidParameterValue if validation of clean steps fails.
+    :raises: InvalidParameterValue if validation of steps fails.
+    """
+    _check_steps(clean_steps, 'clean', _CLEAN_STEPS_SCHEMA)
+
+
+def _check_deploy_steps(deploy_steps):
+    """Ensure all necessary keys are present and correct in steps for deploy
+
+    :param deploy_steps: a list of steps. For more details, see the
+        deploy_steps parameter of :func:`NodeStatesController.provision`.
+    :raises: InvalidParameterValue if validation of steps fails.
+    """
+    _check_steps(deploy_steps, 'deploy', _DEPLOY_STEPS_SCHEMA)
+
+
+def _check_steps(steps, step_type, schema):
+    """Ensure all necessary keys are present and correct in steps.
+
+    Check that the user-specified steps are in the expected format and include
+    the required information.
+
+    :param steps: a list of steps. For more details, see the
+        clean_steps and deploy_steps parameter of
+        :func:`NodeStatesController.provision`.
+    :param step_type: 'clean' or 'deploy' step type
+    :param schema: JSON schema to use for validation.
+    :raises: InvalidParameterValue if validation of steps fails.
     """
     try:
-        jsonschema.validate(clean_steps, _CLEAN_STEPS_SCHEMA)
+        jsonschema.validate(steps, schema)
     except jsonschema.ValidationError as exc:
-        raise exception.InvalidParameterValue(_('Invalid clean_steps: %s') %
-                                              exc)
+        raise exception.InvalidParameterValue(_('Invalid %s_steps: %s') %
+                                              (step_type, exc))
 
 
 def _get_chassis_uuid(node):
diff --git a/ironic/api/controllers/v1/utils.py b/ironic/api/controllers/v1/utils.py
index 4af137645..1156fa606 100644
--- a/ironic/api/controllers/v1/utils.py
+++ b/ironic/api/controllers/v1/utils.py
@@ -37,6 +37,7 @@ from ironic.common.i18n import _
 from ironic.common import policy
 from ironic.common import states
 from ironic.common import utils
+from ironic.conductor import steps as conductor_steps
 from ironic import objects
 from ironic.objects import fields as ofields
 
@@ -121,6 +122,24 @@ LOCAL_LINK_CONN_SCHEMA = {'anyOf': [
     {'type': 'object', 'additionalProperties': False},
 ]}
 
+DEPLOY_STEP_SCHEMA = {
+    'type': 'object',
+    'properties': {
+        'args': {'type': 'object'},
+        'interface': {
+            'type': 'string',
+            'enum': list(conductor_steps.DEPLOYING_INTERFACE_PRIORITY)
+        },
+        'priority': {'anyOf': [
+            {'type': 'integer', 'minimum': 0},
+            {'type': 'string', 'minLength': 1, 'pattern': '^[0-9]+$'}
+        ]},
+        'step': {'type': 'string', 'minLength': 1},
+    },
+    'required': ['interface', 'step', 'args', 'priority'],
+    'additionalProperties': False,
+}
+
 
 def local_link_normalize(name, value):
     if not value:
@@ -1689,3 +1708,29 @@ def allow_local_link_connection_network_type():
 def allow_verify_ca_in_heartbeat():
     """Check if heartbeat accepts agent_verify_ca."""
     return api.request.version.minor >= versions.MINOR_68_HEARTBEAT_VERIFY_CA
+
+
+def allow_deploy_steps():
+    """Check if deploy_steps are available."""
+    return api.request.version.minor >= versions.MINOR_69_DEPLOY_STEPS
+
+
+def check_allow_deploy_steps(target, deploy_steps):
+    """Check if deploy steps are allowed"""
+
+    if not deploy_steps:
+        return
+
+    if not allow_deploy_steps():
+        raise exception.NotAcceptable(_(
+            "Request not acceptable. The minimal required API version "
+            "should be %(base)s.%(opr)s") %
+            {'base': versions.BASE_VERSION,
+             'opr': versions.MINOR_69_DEPLOY_STEPS})
+
+    allowed_states = (states.ACTIVE, states.REBUILD)
+    if target not in allowed_states:
+        msg = (_('"deploy_steps" is only valid when setting target '
+                 'provision state to %s or %s') % allowed_states)
+        raise exception.ClientSideError(
+            msg, status_code=http_client.BAD_REQUEST)
diff --git a/ironic/api/controllers/v1/versions.py b/ironic/api/controllers/v1/versions.py
index b9c2d6690..d00d68f60 100644
--- a/ironic/api/controllers/v1/versions.py
+++ b/ironic/api/controllers/v1/versions.py
@@ -106,6 +106,7 @@ BASE_VERSION = 1
 # v1.66: Add support for node network_data field.
 # v1.67: Add support for port_uuid/portgroup_uuid in node vif_attach
 # v1.68: Add agent_verify_ca to heartbeat.
+# v1.69: Add deploy_steps to provisioning
 
 MINOR_0_JUNO = 0
 MINOR_1_INITIAL_VERSION = 1
@@ -176,6 +177,7 @@ MINOR_65_NODE_LESSEE = 65
 MINOR_66_NODE_NETWORK_DATA = 66
 MINOR_67_NODE_VIF_ATTACH_PORT = 67
 MINOR_68_HEARTBEAT_VERIFY_CA = 68
+MINOR_69_DEPLOY_STEPS = 69
 
 # When adding another version, update:
 # - MINOR_MAX_VERSION
@@ -183,7 +185,7 @@ MINOR_68_HEARTBEAT_VERIFY_CA = 68
 #   explanation of what changed in the new version
 # - common/release_mappings.py, RELEASE_MAPPING['master']['api']
 
-MINOR_MAX_VERSION = MINOR_68_HEARTBEAT_VERIFY_CA
+MINOR_MAX_VERSION = MINOR_69_DEPLOY_STEPS
 
 # String representations of the minor and maximum versions
 _MIN_VERSION_STRING = '{}.{}'.format(BASE_VERSION, MINOR_1_INITIAL_VERSION)
diff --git a/ironic/common/release_mappings.py b/ironic/common/release_mappings.py
index abc9a2e38..8ed93a00d 100644
--- a/ironic/common/release_mappings.py
+++ b/ironic/common/release_mappings.py
@@ -302,8 +302,8 @@ RELEASE_MAPPING = {
         }
     },
     'master': {
-        'api': '1.68',
-        'rpc': '1.51',
+        'api': '1.69',
+        'rpc': '1.52',
         'objects': {
             'Allocation': ['1.1'],
             'Node': ['1.35'],
diff --git a/ironic/conductor/deployments.py b/ironic/conductor/deployments.py
index d6ab54aea..a37ecd29a 100644
--- a/ironic/conductor/deployments.py
+++ b/ironic/conductor/deployments.py
@@ -58,7 +58,8 @@ def validate_node(task, event='deploy'):
 
 @METRICS.timer('start_deploy')
 @task_manager.require_exclusive_lock
-def start_deploy(task, manager, configdrive=None, event='deploy'):
+def start_deploy(task, manager, configdrive=None, event='deploy',
+                 deploy_steps=None):
     """Start deployment or rebuilding on a node.
 
     This function does not check the node suitability for deployment, it's left
@@ -68,6 +69,7 @@ def start_deploy(task, manager, configdrive=None, event='deploy'):
     :param manager: a ConductorManager to run tasks on.
     :param configdrive: a configdrive, if requested.
     :param event: event to process: deploy or rebuild.
+    :param deploy_steps: Optional deploy steps.
     """
     node = task.node
 
@@ -98,7 +100,8 @@ def start_deploy(task, manager, configdrive=None, event='deploy'):
         task.driver.power.validate(task)
         task.driver.deploy.validate(task)
         utils.validate_instance_info_traits(task.node)
-        conductor_steps.validate_deploy_templates(task, skip_missing=True)
+        conductor_steps.validate_user_deploy_steps_and_templates(
+            task, deploy_steps, skip_missing=True)
     except exception.InvalidParameterValue as e:
         raise exception.InstanceDeployFailure(
             _("Failed to validate deploy or power info for node "
@@ -110,7 +113,7 @@ def start_deploy(task, manager, configdrive=None, event='deploy'):
             event,
             callback=manager._spawn_worker,
             call_args=(do_node_deploy, task,
-                       manager.conductor.id, configdrive),
+                       manager.conductor.id, configdrive, deploy_steps),
             err_handler=utils.provisioning_error_handler)
     except exception.InvalidState:
         raise exception.InvalidStateRequested(
@@ -120,7 +123,8 @@ def start_deploy(task, manager, configdrive=None, event='deploy'):
 
 @METRICS.timer('do_node_deploy')
 @task_manager.require_exclusive_lock
-def do_node_deploy(task, conductor_id=None, configdrive=None):
+def do_node_deploy(task, conductor_id=None, configdrive=None,
+                   deploy_steps=None):
     """Prepare the environment and deploy a node."""
     node = task.node
     utils.wipe_deploy_internal_info(task)
@@ -181,7 +185,16 @@ def do_node_deploy(task, conductor_id=None, configdrive=None):
                 traceback=True, clean_up=False)
 
     try:
-        # This gets the deploy steps (if any) and puts them in the node's
+        # If any deploy steps provided by user, save them to node. They will be
+        # validated & processed later together with driver and deploy template
+        # steps.
+        if deploy_steps:
+            info = node.driver_internal_info
+            info['user_deploy_steps'] = deploy_steps
+            node.driver_internal_info = info
+            node.save()
+        # This gets the deploy steps (if any) from driver, deploy template and
+        # deploy_steps argument and updates them in the node's
         # driver_internal_info['deploy_steps']. In-band steps are skipped since
         # we know that an agent is not running yet.
         conductor_steps.set_node_deployment_steps(task, skip_missing=True)
@@ -350,7 +363,7 @@ def continue_node_deploy(task):
     # Agent is now running, we're ready to validate the remaining steps
     if not node.driver_internal_info.get('steps_validated'):
         try:
-            conductor_steps.validate_deploy_templates(task)
+            conductor_steps.validate_user_deploy_steps_and_templates(task)
             conductor_steps.set_node_deployment_steps(
                 task, reset_current=False)
         except exception.IronicException as exc:
diff --git a/ironic/conductor/manager.py b/ironic/conductor/manager.py
index d22cd8525..8397be581 100644
--- a/ironic/conductor/manager.py
+++ b/ironic/conductor/manager.py
@@ -91,7 +91,7 @@ class ConductorManager(base_manager.BaseConductorManager):
     # NOTE(rloo): This must be in sync with rpcapi.ConductorAPI's.
     # NOTE(pas-ha): This also must be in sync with
     #               ironic.common.release_mappings.RELEASE_MAPPING['master']
-    RPC_API_VERSION = '1.51'
+    RPC_API_VERSION = '1.52'
 
     target = messaging.Target(version=RPC_API_VERSION)
 
@@ -809,7 +809,7 @@ class ConductorManager(base_manager.BaseConductorManager):
                                    exception.InvalidStateRequested,
                                    exception.NodeProtected)
     def do_node_deploy(self, context, node_id, rebuild=False,
-                       configdrive=None):
+                       configdrive=None, deploy_steps=None):
         """RPC method to initiate deployment to a node.
 
         Initiate the deployment of a node. Validations are done
@@ -823,6 +823,7 @@ class ConductorManager(base_manager.BaseConductorManager):
                         all disk. The ephemeral partition, if it exists, can
                         optionally be preserved.
         :param configdrive: Optional. A gzipped and base64 encoded configdrive.
+        :param deploy_steps: Optional. Deploy steps.
         :raises: InstanceDeployFailure
         :raises: NodeInMaintenance if the node is in maintenance mode.
         :raises: NoFreeConductorWorker when there is no free worker to start
@@ -841,7 +842,8 @@ class ConductorManager(base_manager.BaseConductorManager):
         with task_manager.acquire(context, node_id, shared=False,
                                   purpose='node deployment') as task:
             deployments.validate_node(task, event=event)
-            deployments.start_deploy(task, self, configdrive, event=event)
+            deployments.start_deploy(task, self, configdrive, event=event,
+                                     deploy_steps=deploy_steps)
 
     @METRICS.timer('ConductorManager.continue_node_deploy')
     def continue_node_deploy(self, context, node_id):
@@ -1890,8 +1892,9 @@ class ConductorManager(base_manager.BaseConductorManager):
                         # NOTE(dtantsur): without the agent running we cannot
                         # have the complete list of steps, so skip ones that we
                         # don't know.
-                        conductor_steps.validate_deploy_templates(
-                            task, skip_missing=True)
+                        (conductor_steps
+                         .validate_user_deploy_steps_and_templates(
+                             task, skip_missing=True))
                     result = True
                 except (exception.InvalidParameterValue,
                         exception.UnsupportedDriverExtension) as e:
diff --git a/ironic/conductor/rpcapi.py b/ironic/conductor/rpcapi.py
index 2c88e4872..e41cb7791 100644
--- a/ironic/conductor/rpcapi.py
+++ b/ironic/conductor/rpcapi.py
@@ -104,13 +104,14 @@ class ConductorAPI(object):
     |    1.50 - Added set_indicator_state, get_indicator_state and
     |           get_supported_indicators.
     |    1.51 - Added agent_verify_ca to heartbeat.
+    |    1.52 - Added deploy steps argument to provisioning
 
     """
 
     # NOTE(rloo): This must be in sync with manager.ConductorManager's.
     # NOTE(pas-ha): This also must be in sync with
     #               ironic.common.release_mappings.RELEASE_MAPPING['master']
-    RPC_API_VERSION = '1.51'
+    RPC_API_VERSION = '1.52'
 
     def __init__(self, topic=None):
         super(ConductorAPI, self).__init__()
@@ -399,7 +400,7 @@ class ConductorAPI(object):
                           driver_name=driver_name)
 
     def do_node_deploy(self, context, node_id, rebuild, configdrive,
-                       topic=None):
+                       topic=None, deploy_steps=None):
         """Signal to conductor service to perform a deployment.
 
         :param context: request context.
@@ -407,6 +408,7 @@ class ConductorAPI(object):
         :param rebuild: True if this is a rebuild request.
         :param configdrive: A gzipped and base64 encoded configdrive.
         :param topic: RPC topic. Defaults to self.topic.
+        :param deploy_steps: Deploy steps
         :raises: InstanceDeployFailure
         :raises: InvalidParameterValue if validation fails
         :raises: MissingParameterValue if a required parameter is missing
@@ -417,9 +419,15 @@ class ConductorAPI(object):
         undeployed state before this method is called.
 
         """
-        cctxt = self.client.prepare(topic=topic or self.topic, version='1.22')
+        version = '1.22'
+        new_kws = {}
+        if deploy_steps:
+            version = '1.52'
+            new_kws['deploy_steps'] = deploy_steps
+
+        cctxt = self.client.prepare(topic=topic or self.topic, version=version)
         return cctxt.call(context, 'do_node_deploy', node_id=node_id,
-                          rebuild=rebuild, configdrive=configdrive)
+                          rebuild=rebuild, configdrive=configdrive, **new_kws)
 
     def do_node_tear_down(self, context, node_id, topic=None):
         """Signal to conductor service to tear down a deployment.
diff --git a/ironic/conductor/steps.py b/ironic/conductor/steps.py
index a6663db01..b38186610 100644
--- a/ironic/conductor/steps.py
+++ b/ironic/conductor/steps.py
@@ -284,12 +284,23 @@ def _get_all_deployment_steps(task, skip_missing=False):
         deploy steps.
     :returns: A list of deploy step dictionaries
     """
+    # Get deploy steps provided by user via argument if any. These steps
+    # override template and driver steps when overlap.
+    user_steps = _get_validated_user_deploy_steps(
+        task, skip_missing=skip_missing)
+
     # Gather deploy steps from deploy templates and validate.
     # NOTE(mgoddard): although we've probably just validated the templates in
     # do_node_deploy, they may have changed in the DB since we last checked, so
     # validate again.
-    user_steps = _get_validated_steps_from_templates(task,
-                                                     skip_missing=skip_missing)
+    template_steps = _get_validated_steps_from_templates(
+        task, skip_missing=skip_missing)
+
+    # Take only template steps that are not already provided by user
+    user_step_keys = {(s['interface'], s['step']) for s in user_steps}
+    new_template_steps = [s for s in template_steps
+                          if (s['interface'], s['step']) not in user_step_keys]
+    user_steps.extend(new_template_steps)
 
     # Gather enabled deploy steps from drivers.
     driver_steps = _get_deployment_steps(task, enabled=True, sort=False)
@@ -548,7 +559,8 @@ def _validate_user_steps(task, user_steps, driver_steps, step_type,
         result.append(user_step)
 
     if step_type == 'deploy':
-        # Deploy steps should be unique across all combined templates.
+        # Deploy steps should be unique across all combined templates or passed
+        # deploy_steps argument.
         dup_errors = _validate_deploy_steps_unique(result)
         errors.extend(dup_errors)
 
@@ -617,14 +629,49 @@ def _validate_user_deploy_steps(task, user_steps, error_prefix=None,
                                 skip_missing=skip_missing)
 
 
-def validate_deploy_templates(task, skip_missing=False):
-    """Validate the deploy templates for a node.
+def _get_validated_user_deploy_steps(task, deploy_steps=None,
+                                     skip_missing=False):
+    """Validate the deploy steps for a node.
+
+    :param task: A TaskManager object
+    :param deploy_steps: Deploy steps to validate. Optional. If not provided
+        then will check node's driver internal info.
+    :param skip_missing: whether skip missing steps that are not yet available
+        at the time of validation.
+    :raises: InvalidParameterValue if deploy steps are unsupported by the
+        node's driver interfaces.
+    :raises: InstanceDeployFailure if there was a problem getting the deploy
+        steps from the driver.
+    """
+    if not deploy_steps:
+        deploy_steps = task.node.driver_internal_info.get('user_deploy_steps')
+
+    if deploy_steps:
+        error_prefix = (_('Validation of deploy steps from "deploy steps" '
+                          'argument failed.'))
+        return _validate_user_deploy_steps(task, deploy_steps,
+                                           error_prefix=error_prefix,
+                                           skip_missing=skip_missing)
+    else:
+        return []
+
+
+def validate_user_deploy_steps_and_templates(task, deploy_steps=None,
+                                             skip_missing=False):
+    """Validate the user deploy steps and the deploy templates for a node.
 
     :param task: A TaskManager object
+    :param deploy_steps: Deploy steps to validate. Optional. If not provided
+        then will check node's driver internal info.
+    :param skip_missing: whether skip missing steps that are not yet available
+        at the time of validation.
     :raises: InvalidParameterValue if the instance has traits that map to
-        deploy steps that are unsupported by the node's driver interfaces.
+        deploy steps that are unsupported by the node's driver interfaces or
+        user deploy steps are unsupported by the node's driver interfaces
     :raises: InstanceDeployFailure if there was a problem getting the deploy
         steps from the driver.
     """
     # Gather deploy steps from matching deploy templates and validate them.
     _get_validated_steps_from_templates(task, skip_missing=skip_missing)
+    # Validate steps from passed argument or stored on the node.
+    _get_validated_user_deploy_steps(task, deploy_steps, skip_missing)
diff --git a/ironic/conductor/utils.py b/ironic/conductor/utils.py
index 4823b84a6..25070109d 100644
--- a/ironic/conductor/utils.py
+++ b/ironic/conductor/utils.py
@@ -523,6 +523,7 @@ def wipe_deploy_internal_info(task):
     # Clear any leftover metadata about deployment.
     info = task.node.driver_internal_info
     info['deploy_steps'] = None
+    info.pop('user_deploy_steps', None)
     info.pop('agent_cached_deploy_steps', None)
     info.pop('deploy_step_index', None)
     info.pop('deployment_reboot', None)
diff --git a/ironic/tests/unit/api/controllers/v1/test_node.py b/ironic/tests/unit/api/controllers/v1/test_node.py
index d450d78ee..eb9873239 100644
--- a/ironic/tests/unit/api/controllers/v1/test_node.py
+++ b/ironic/tests/unit/api/controllers/v1/test_node.py
@@ -4978,7 +4978,8 @@ class TestPut(test_api_base.BaseApiTest):
                                               node_id=self.node.uuid,
                                               rebuild=False,
                                               configdrive=None,
-                                              topic='test-topic')
+                                              topic='test-topic',
+                                              deploy_steps=None)
         # Check location header
         self.assertIsNotNone(ret.location)
         expected_location = '/v1/nodes/%s/states' % self.node.uuid
@@ -5002,7 +5003,8 @@ class TestPut(test_api_base.BaseApiTest):
                                               node_id=self.node.uuid,
                                               rebuild=False,
                                               configdrive=None,
-                                              topic='test-topic')
+                                              topic='test-topic',
+                                              deploy_steps=None)
 
     def test_provision_with_deploy_configdrive(self):
         ret = self.put_json('/nodes/%s/states/provision' % self.node.uuid,
@@ -5013,7 +5015,8 @@ class TestPut(test_api_base.BaseApiTest):
                                               node_id=self.node.uuid,
                                               rebuild=False,
                                               configdrive='foo',
-                                              topic='test-topic')
+                                              topic='test-topic',
+                                              deploy_steps=None)
         # Check location header
         self.assertIsNotNone(ret.location)
         expected_location = '/v1/nodes/%s/states' % self.node.uuid
@@ -5031,7 +5034,8 @@ class TestPut(test_api_base.BaseApiTest):
                                               node_id=self.node.uuid,
                                               rebuild=False,
                                               configdrive={'user_data': 'foo'},
-                                              topic='test-topic')
+                                              topic='test-topic',
+                                              deploy_steps=None)
 
     def test_provision_with_deploy_configdrive_as_dict_all_fields(self):
         fake_cd = {'user_data': {'serialize': 'me'},
@@ -5048,7 +5052,8 @@ class TestPut(test_api_base.BaseApiTest):
                                               node_id=self.node.uuid,
                                               rebuild=False,
                                               configdrive=fake_cd,
-                                              topic='test-topic')
+                                              topic='test-topic',
+                                              deploy_steps=None)
 
     def test_provision_with_deploy_configdrive_as_dict_unsupported(self):
         ret = self.put_json('/nodes/%s/states/provision' % self.node.uuid,
@@ -5057,6 +5062,39 @@ class TestPut(test_api_base.BaseApiTest):
                             expect_errors=True)
         self.assertEqual(http_client.BAD_REQUEST, ret.status_code)
 
+    @mock.patch.object(api_utils, 'check_allow_deploy_steps', autospec=True)
+    def test_provision_with_deploy_deploy_steps(self, mock_check):
+        deploy_steps = [{'interface': 'bios',
+                         'step': 'factory_reset',
+                         'priority': 95,
+                         'args': {}}]
+        ret = self.put_json('/nodes/%s/states/provision' % self.node.uuid,
+                            {'target': states.ACTIVE,
+                             'deploy_steps': deploy_steps})
+        self.assertEqual(http_client.ACCEPTED, ret.status_code)
+        self.assertEqual(b'', ret.body)
+        self.mock_dnd.assert_called_once_with(context=mock.ANY,
+                                              node_id=self.node.uuid,
+                                              rebuild=False,
+                                              configdrive=None,
+                                              topic='test-topic',
+                                              deploy_steps=deploy_steps)
+        # Check location header
+        self.assertIsNotNone(ret.location)
+        expected_location = '/v1/nodes/%s/states' % self.node.uuid
+        self.assertEqual(urlparse.urlparse(ret.location).path,
+                         expected_location)
+
+    def test_provision_with_deploy_deploy_steps_fail(self):
+        # Mandatory 'priority' missing in the step
+        deploy_steps = [{'interface': 'bios',
+                         'step': 'factory_reset'}]
+        ret = self.put_json('/nodes/%s/states/provision' % self.node.uuid,
+                            {'target': states.ACTIVE,
+                             'deploy_steps': deploy_steps},
+                            expect_errors=True)
+        self.assertEqual(http_client.NOT_ACCEPTABLE, ret.status_code)
+
     def test_provision_with_rebuild(self):
         node = self.node
         node.provision_state = states.ACTIVE
@@ -5070,7 +5108,8 @@ class TestPut(test_api_base.BaseApiTest):
                                               node_id=self.node.uuid,
                                               rebuild=True,
                                               configdrive=None,
-                                              topic='test-topic')
+                                              topic='test-topic',
+                                              deploy_steps=None)
         # Check location header
         self.assertIsNotNone(ret.location)
         expected_location = '/v1/nodes/%s/states' % self.node.uuid
@@ -5101,7 +5140,8 @@ class TestPut(test_api_base.BaseApiTest):
                                               node_id=self.node.uuid,
                                               rebuild=True,
                                               configdrive='foo',
-                                              topic='test-topic')
+                                              topic='test-topic',
+                                              deploy_steps=None)
         # Check location header
         self.assertIsNotNone(ret.location)
         expected_location = '/v1/nodes/%s/states' % self.node.uuid
@@ -5114,6 +5154,33 @@ class TestPut(test_api_base.BaseApiTest):
                             expect_errors=True)
         self.assertEqual(http_client.BAD_REQUEST, ret.status_code)
 
+    @mock.patch.object(api_utils, 'check_allow_deploy_steps', autospec=True)
+    def test_provision_with_rebuild_deploy_steps(self, mock_check):
+        node = self.node
+        node.provision_state = states.ACTIVE
+        node.target_provision_state = states.NOSTATE
+        node.save()
+        deploy_steps = [{'interface': 'bios',
+                         'step': 'factory_reset',
+                         'priority': 95,
+                         'args': {}}]
+        ret = self.put_json('/nodes/%s/states/provision' % self.node.uuid,
+                            {'target': states.REBUILD,
+                             'deploy_steps': deploy_steps})
+        self.assertEqual(http_client.ACCEPTED, ret.status_code)
+        self.assertEqual(b'', ret.body)
+        self.mock_dnd.assert_called_once_with(context=mock.ANY,
+                                              node_id=self.node.uuid,
+                                              rebuild=True,
+                                              configdrive=None,
+                                              topic='test-topic',
+                                              deploy_steps=deploy_steps)
+        # Check location header
+        self.assertIsNotNone(ret.location)
+        expected_location = '/v1/nodes/%s/states' % self.node.uuid
+        self.assertEqual(urlparse.urlparse(ret.location).path,
+                         expected_location)
+
     def test_provision_with_tear_down(self):
         node = self.node
         node.provision_state = states.ACTIVE
@@ -5191,7 +5258,8 @@ class TestPut(test_api_base.BaseApiTest):
                                               node_id=self.node.uuid,
                                               rebuild=False,
                                               configdrive=None,
-                                              topic='test-topic')
+                                              topic='test-topic',
+                                              deploy_steps=None)
         # Check location header
         self.assertIsNotNone(ret.location)
         expected_location = '/v1/nodes/%s/states' % node.uuid
diff --git a/ironic/tests/unit/api/controllers/v1/test_utils.py b/ironic/tests/unit/api/controllers/v1/test_utils.py
index 761c0df17..67b0b55d5 100644
--- a/ironic/tests/unit/api/controllers/v1/test_utils.py
+++ b/ironic/tests/unit/api/controllers/v1/test_utils.py
@@ -760,6 +760,32 @@ class TestCheckAllowFields(base.TestCase):
         mock_request.version.minor = 61
         self.assertFalse(utils.allow_agent_token())
 
+    def test_allow_deploy_steps(self, mock_request):
+        mock_request.version.minor = 69
+        self.assertTrue(utils.allow_deploy_steps())
+        mock_request.version.minor = 68
+        self.assertFalse(utils.allow_deploy_steps())
+
+    def test_check_allow_deploy_steps(self, mock_request):
+        mock_request.version.minor = 69
+        utils.check_allow_deploy_steps(states.ACTIVE, {'a': 1})
+        utils.check_allow_deploy_steps(states.REBUILD, {'a': 1})
+
+    def test_check_allow_deploy_steps_empty(self, mock_request):
+        utils.check_allow_deploy_steps(states.ACTIVE, None)
+
+    def test_check_allow_deploy_steps_version_older(self, mock_request):
+        mock_request.version.minor = 68
+        self.assertRaises(exception.NotAcceptable,
+                          utils.check_allow_deploy_steps,
+                          states.ACTIVE, {'a': 1})
+
+    def test_check_allow_deploy_steps_target_unsupported(self, mock_request):
+        mock_request.version.minor = 69
+        self.assertRaises(exception.ClientSideError,
+                          utils.check_allow_deploy_steps,
+                          states.MANAGEABLE, {'a': 1})
+
 
 @mock.patch.object(api, 'request', spec_set=['context', 'version'])
 class TestNodeIdent(base.TestCase):
diff --git a/ironic/tests/unit/api/utils.py b/ironic/tests/unit/api/utils.py
index 58970f2b3..d3963fc15 100644
--- a/ironic/tests/unit/api/utils.py
+++ b/ironic/tests/unit/api/utils.py
@@ -191,7 +191,7 @@ def deploy_template_post_data(**kw):
     # These values are not part of the API object
     template.pop('version')
     # Remove internal attributes from each step.
-    step_internal = dt_controller.STEP_SCHEMA['properties']
+    step_internal = api_utils.DEPLOY_STEP_SCHEMA['properties']
     template['steps'] = [remove_other_fields(step, step_internal)
                          for step in template['steps']]
     # Remove internal attributes from the template.
diff --git a/ironic/tests/unit/conductor/test_deployments.py b/ironic/tests/unit/conductor/test_deployments.py
index 56ec17072..d029b7994 100644
--- a/ironic/tests/unit/conductor/test_deployments.py
+++ b/ironic/tests/unit/conductor/test_deployments.py
@@ -388,33 +388,37 @@ class DoNodeDeployTestCase(mgr_utils.ServiceSetUpMixin, db_base.DbTestCase):
                 autospec=True)
     @mock.patch('ironic.drivers.modules.fake.FakeDeploy.validate',
                 autospec=True)
-    @mock.patch.object(conductor_steps, 'validate_deploy_templates',
+    @mock.patch.object(conductor_steps,
+                       'validate_user_deploy_steps_and_templates',
                        autospec=True)
     @mock.patch.object(conductor_utils, 'validate_instance_info_traits',
                        autospec=True)
     @mock.patch.object(images, 'is_whole_disk_image', autospec=True)
     def test_start_deploy(self, mock_iwdi, mock_validate_traits,
-                          mock_validate_templates, mock_deploy_validate,
+                          mock_validate_deploy_user_steps_and_templates,
+                          mock_deploy_validate,
                           mock_power_validate, mock_process_event):
         self._start_service()
         mock_iwdi.return_value = False
+        deploy_steps = [{"interface": "bios", "step": "factory_reset",
+                         "priority": 95}]
         node = obj_utils.create_test_node(self.context, driver='fake-hardware',
                                           provision_state=states.AVAILABLE,
                                           target_provision_state=states.ACTIVE)
         task = task_manager.TaskManager(self.context, node.uuid)
 
         deployments.start_deploy(task, self.service, configdrive=None,
-                                 event='deploy')
+                                 event='deploy', deploy_steps=deploy_steps)
         node.refresh()
         self.assertTrue(mock_iwdi.called)
         mock_power_validate.assert_called_once_with(task.driver.power, task)
         mock_deploy_validate.assert_called_once_with(task.driver.deploy, task)
         mock_validate_traits.assert_called_once_with(task.node)
-        mock_validate_templates.assert_called_once_with(
-            task, skip_missing=True)
+        mock_validate_deploy_user_steps_and_templates.assert_called_once_with(
+            task, deploy_steps, skip_missing=True)
         mock_process_event.assert_called_with(
             mock.ANY, 'deploy', call_args=(
-                deployments.do_node_deploy, task, 1, None),
+                deployments.do_node_deploy, task, 1, None, deploy_steps),
             callback=mock.ANY, err_handler=mock.ANY)
 
 
@@ -849,9 +853,12 @@ class DoNextDeployStepTestCase(mgr_utils.ServiceSetUpMixin,
             mock.ANY, mock.ANY, self.deploy_steps[0])
 
     @mock.patch.object(deployments, 'do_next_deploy_step', autospec=True)
-    def _continue_node_deploy(self, mock_next_step, skip=True):
+    @mock.patch.object(conductor_steps, '_get_steps', autospec=True)
+    def _continue_node_deploy(self, mock__get_steps, mock_next_step,
+                              skip=True):
+        mock__get_steps.return_value = self.deploy_steps
         driver_info = {'deploy_steps': self.deploy_steps,
-                       'deploy_step_index': 0,
+                       'deploy_step_index': 1,
                        'deployment_polling': 'value'}
         if not skip:
             driver_info['skip_current_deploy_step'] = skip
@@ -862,7 +869,7 @@ class DoNextDeployStepTestCase(mgr_utils.ServiceSetUpMixin,
             deploy_step=self.deploy_steps[0])
         with task_manager.acquire(self.context, node.uuid) as task:
             deployments.continue_node_deploy(task)
-            expected_step_index = None if skip else 0
+            expected_step_index = None if skip else 1
             self.assertNotIn(
                 'skip_current_deploy_step', task.node.driver_internal_info)
             self.assertNotIn(
@@ -899,7 +906,8 @@ class DoNextDeployStepTestCase(mgr_utils.ServiceSetUpMixin,
             mock_next_step.assert_called_once_with(task, 1)
 
     @mock.patch.object(deployments, 'do_next_deploy_step', autospec=True)
-    @mock.patch.object(conductor_steps, 'validate_deploy_templates',
+    @mock.patch.object(conductor_steps,
+                       'validate_user_deploy_steps_and_templates',
                        autospec=True)
     def test_continue_node_steps_validation(self, mock_validate,
                                             mock_next_step):
diff --git a/ironic/tests/unit/conductor/test_manager.py b/ironic/tests/unit/conductor/test_manager.py
index 076963d5c..d8e1d387d 100644
--- a/ironic/tests/unit/conductor/test_manager.py
+++ b/ironic/tests/unit/conductor/test_manager.py
@@ -1455,7 +1455,8 @@ class ServiceDoNodeDeployTestCase(mgr_utils.ServiceSetUpMixin,
                                                  mock_iwdi):
         self._test_do_node_deploy_validate_fail(mock_validate, mock_iwdi)
 
-    @mock.patch.object(conductor_steps, 'validate_deploy_templates',
+    @mock.patch.object(conductor_steps,
+                       'validate_user_deploy_steps_and_templates',
                        autospec=True)
     def test_do_node_deploy_validate_template_fail(self, mock_validate,
                                                    mock_iwdi):
@@ -1485,7 +1486,7 @@ class ServiceDoNodeDeployTestCase(mgr_utils.ServiceSetUpMixin,
             # Verify reservation has been cleared.
             self.assertIsNone(node.reservation)
             mock_spawn.assert_called_once_with(mock.ANY, mock.ANY,
-                                               mock.ANY, None)
+                                               mock.ANY, None, None)
             mock_iwdi.assert_called_once_with(self.context, node.instance_info)
             self.assertFalse(node.driver_internal_info['is_whole_disk_image'])
 
@@ -3216,7 +3217,8 @@ class MiscTestCase(mgr_utils.ServiceSetUpMixin, mgr_utils.CommonMixIn,
         node = obj_utils.create_test_node(self.context, driver='fake-hardware',
                                           network_interface='noop')
         with mock.patch(
-                'ironic.conductor.steps.validate_deploy_templates',
+                'ironic.conductor.steps'
+                '.validate_user_deploy_steps_and_templates',
                 autospec=True) as mock_validate:
             reason = 'fake reason'
             mock_validate.side_effect = exception.InvalidParameterValue(reason)
diff --git a/ironic/tests/unit/conductor/test_rpcapi.py b/ironic/tests/unit/conductor/test_rpcapi.py
index 1cb0b7d9b..3f0ae2040 100644
--- a/ironic/tests/unit/conductor/test_rpcapi.py
+++ b/ironic/tests/unit/conductor/test_rpcapi.py
@@ -327,6 +327,15 @@ class RPCAPITestCase(db_base.DbTestCase):
                           rebuild=False,
                           configdrive=None)
 
+    def test_do_node_deploy_with_deploy_steps(self):
+        self._test_rpcapi('do_node_deploy',
+                          'call',
+                          version='1.52',
+                          node_id=self.fake_node['uuid'],
+                          rebuild=False,
+                          configdrive=None,
+                          deploy_steps={'key': 'value'})
+
     def test_do_node_tear_down(self):
         self._test_rpcapi('do_node_tear_down',
                           'call',
diff --git a/ironic/tests/unit/conductor/test_steps.py b/ironic/tests/unit/conductor/test_steps.py
index 845d639d4..1053ff236 100644
--- a/ironic/tests/unit/conductor/test_steps.py
+++ b/ironic/tests/unit/conductor/test_steps.py
@@ -182,63 +182,111 @@ class NodeDeployStepsTestCase(db_base.DbTestCase):
                 task, [template1, template2])
             self.assertEqual(expected, steps)
 
+    @mock.patch.object(conductor_steps, '_get_validated_user_deploy_steps',
+                       autospec=True)
     @mock.patch.object(conductor_steps, '_get_validated_steps_from_templates',
                        autospec=True)
     @mock.patch.object(conductor_steps, '_get_deployment_steps', autospec=True)
-    def _test__get_all_deployment_steps(self, user_steps, driver_steps,
-                                        expected_steps, mock_steps,
-                                        mock_validated):
-        mock_validated.return_value = user_steps
+    def _test__get_all_deployment_steps(self, user_steps, template_steps,
+                                        driver_steps, expected_steps,
+                                        mock_steps, mock_validated_template,
+                                        mock_validated_user):
+        returned_user_steps = user_steps.copy()
+        mock_validated_user.return_value = returned_user_steps
+        mock_validated_template.return_value = template_steps
         mock_steps.return_value = driver_steps
-
         with task_manager.acquire(
                 self.context, self.node.uuid, shared=False) as task:
+
             steps = conductor_steps._get_all_deployment_steps(task)
             self.assertEqual(expected_steps, steps)
-            mock_validated.assert_called_once_with(task, skip_missing=False)
+            mock_validated_template.assert_called_once_with(task,
+                                                            skip_missing=False)
             mock_steps.assert_called_once_with(task, enabled=True, sort=False)
+            mock_validated_user.assert_called_once_with(
+                task, skip_missing=False)
 
     def test__get_all_deployment_steps_no_steps(self):
         # Nothing in -> nothing out.
         user_steps = []
+        template_steps = []
         driver_steps = []
         expected_steps = []
-        self._test__get_all_deployment_steps(user_steps, driver_steps,
-                                             expected_steps)
+        self._test__get_all_deployment_steps(user_steps, template_steps,
+                                             driver_steps, expected_steps)
 
-    def test__get_all_deployment_steps_no_user_steps(self):
+    def test__get_all_deployment_steps_no_template_and_user_steps(self):
         # Only driver steps in -> only driver steps out.
         user_steps = []
+        template_steps = []
         driver_steps = self.deploy_steps
         expected_steps = self.deploy_steps
-        self._test__get_all_deployment_steps(user_steps, driver_steps,
-                                             expected_steps)
+        self._test__get_all_deployment_steps(user_steps, template_steps,
+                                             driver_steps, expected_steps)
+
+    def test__get_all_deployment_steps_no_user_and_driver_steps(self):
+        # Only template steps in -> only template steps out.
+        user_steps = []
+        template_steps = self.deploy_steps
+        driver_steps = []
+        expected_steps = self.deploy_steps
+        self._test__get_all_deployment_steps(user_steps, template_steps,
+                                             driver_steps, expected_steps)
 
-    def test__get_all_deployment_steps_no_driver_steps(self):
-        # Only user steps in -> only user steps out.
+    def test__get_all_deployment_steps_no_template_and_driver_steps(self):
+        # Only template steps in -> only template steps out.
         user_steps = self.deploy_steps
+        template_steps = []
         driver_steps = []
         expected_steps = self.deploy_steps
-        self._test__get_all_deployment_steps(user_steps, driver_steps,
-                                             expected_steps)
+        self._test__get_all_deployment_steps(user_steps, template_steps,
+                                             driver_steps, expected_steps)
+
+    def test__get_all_deployment_steps_template_and_driver_steps(self):
+        # Driver and template steps in -> driver and template steps out.
+        user_steps = []
+        template_steps = self.deploy_steps[:2]
+        driver_steps = self.deploy_steps[2:]
+        expected_steps = self.deploy_steps
+        self._test__get_all_deployment_steps(user_steps, template_steps,
+                                             driver_steps, expected_steps)
 
     def test__get_all_deployment_steps_user_and_driver_steps(self):
         # Driver and user steps in -> driver and user steps out.
         user_steps = self.deploy_steps[:2]
+        template_steps = []
         driver_steps = self.deploy_steps[2:]
         expected_steps = self.deploy_steps
-        self._test__get_all_deployment_steps(user_steps, driver_steps,
-                                             expected_steps)
+        self._test__get_all_deployment_steps(user_steps, template_steps,
+                                             driver_steps, expected_steps)
+
+    def test__get_all_deployment_steps_user_and_template_steps(self):
+        # Template and user steps in -> template and user steps out.
+        user_steps = self.deploy_steps[:2]
+        template_steps = self.deploy_steps[2:]
+        driver_steps = []
+        expected_steps = self.deploy_steps
+        self._test__get_all_deployment_steps(user_steps, template_steps,
+                                             driver_steps, expected_steps)
+
+    def test__get_all_deployment_steps_all_steps(self):
+        # All steps in -> all steps out.
+        user_steps = self.deploy_steps[:1]
+        template_steps = self.deploy_steps[1:3]
+        driver_steps = self.deploy_steps[3:]
+        expected_steps = self.deploy_steps
+        self._test__get_all_deployment_steps(user_steps, template_steps,
+                                             driver_steps, expected_steps)
 
     @mock.patch.object(conductor_steps, '_get_validated_steps_from_templates',
                        autospec=True)
     @mock.patch.object(conductor_steps, '_get_deployment_steps', autospec=True)
     def test__get_all_deployment_steps_skip_missing(self, mock_steps,
                                                     mock_validated):
-        user_steps = self.deploy_steps[:2]
+        template_steps = self.deploy_steps[:2]
         driver_steps = self.deploy_steps[2:]
         expected_steps = self.deploy_steps
-        mock_validated.return_value = user_steps
+        mock_validated.return_value = template_steps
         mock_steps.return_value = driver_steps
 
         with task_manager.acquire(
@@ -251,40 +299,76 @@ class NodeDeployStepsTestCase(db_base.DbTestCase):
 
     def test__get_all_deployment_steps_disable_core_steps(self):
         # User steps can disable core driver steps.
-        user_steps = [self.deploy_core.copy()]
-        user_steps[0].update({'priority': 0})
+        template_steps = [self.deploy_core.copy()]
+        template_steps[0].update({'priority': 0})
         driver_steps = [self.deploy_core]
         expected_steps = []
-        self._test__get_all_deployment_steps(user_steps, driver_steps,
+        self._test__get_all_deployment_steps([], template_steps, driver_steps,
                                              expected_steps)
 
     def test__get_all_deployment_steps_override_driver_steps(self):
         # User steps override non-core driver steps.
-        user_steps = [step.copy() for step in self.deploy_steps[:2]]
-        user_steps[0].update({'priority': 200})
-        user_steps[1].update({'priority': 100})
+        template_steps = [step.copy() for step in self.deploy_steps[:2]]
+        template_steps[0].update({'priority': 200})
+        template_steps[1].update({'priority': 100})
         driver_steps = self.deploy_steps
-        expected_steps = user_steps + self.deploy_steps[2:]
-        self._test__get_all_deployment_steps(user_steps, driver_steps,
+        expected_steps = template_steps + self.deploy_steps[2:]
+        self._test__get_all_deployment_steps([], template_steps, driver_steps,
                                              expected_steps)
 
-    def test__get_all_deployment_steps_duplicate_user_steps(self):
-        # Duplicate user steps override non-core driver steps.
+    def test__get_all_deployment_steps_override_template_steps(self):
+        # User steps override template steps.
+        user_steps = [step.copy() for step in self.deploy_steps[:1]]
+        user_steps[0].update({'priority': 300})
+        template_steps = [step.copy() for step in self.deploy_steps[:2]]
+        template_steps[0].update({'priority': 200})
+        template_steps[1].update({'priority': 100})
+        driver_steps = self.deploy_steps
+        expected_steps = (user_steps[:1]
+                          + template_steps[1:2]
+                          + self.deploy_steps[2:])
+        self._test__get_all_deployment_steps(user_steps, template_steps,
+                                             driver_steps, expected_steps)
+
+    def test__get_all_deployment_steps_duplicate_template_steps(self):
+        # Duplicate template steps override non-core driver steps.
 
         # NOTE(mgoddard): This case is currently prevented by the API and
         # conductor - the interface/step must be unique across all enabled
         # steps. This test ensures that we can support this case, in case we
         # choose to allow it in future.
-        user_steps = [self.deploy_start.copy(), self.deploy_start.copy()]
-        user_steps[0].update({'priority': 200})
-        user_steps[1].update({'priority': 100})
+        template_steps = [self.deploy_start.copy(), self.deploy_start.copy()]
+        template_steps[0].update({'priority': 200})
+        template_steps[1].update({'priority': 100})
         driver_steps = self.deploy_steps
         # Each user invocation of the deploy_start step should be included, but
         # not the default deploy_start from the driver.
-        expected_steps = user_steps + self.deploy_steps[1:]
-        self._test__get_all_deployment_steps(user_steps, driver_steps,
+        expected_steps = template_steps + self.deploy_steps[1:]
+        self._test__get_all_deployment_steps([], template_steps, driver_steps,
                                              expected_steps)
 
+    def test__get_all_deployment_steps_duplicate_template_and_user_steps(self):
+        # Duplicate user steps override non-core driver steps.
+
+        # NOTE(ajya):
+        # See also test__get_all_deployment_steps_duplicate_template_steps.
+        # As user steps provided via API arguments take over template steps,
+        # currently it will override all duplicated steps as it cannot know
+        # which to keep. If duplicates are getting supported, then
+        # _get_all_deployment_steps needs to be updated. Until then this case
+        # tests currently desired outcome.
+        user_steps = [self.deploy_start.copy()]
+        user_steps[0].update({'priority': 300})
+        template_steps = [self.deploy_start.copy(), self.deploy_start.copy()]
+        template_steps[0].update({'priority': 200})
+        template_steps[1].update({'priority': 100})
+        driver_steps = self.deploy_steps
+        # Each user invocation of the deploy_start step should be included, but
+        # not the default deploy_start from the driver.
+        expected_steps = user_steps + self.deploy_steps[1:]
+        self._test__get_all_deployment_steps(user_steps, template_steps,
+                                             driver_steps, expected_steps)
+
     @mock.patch.object(conductor_steps, '_get_validated_steps_from_templates',
                        autospec=True)
     @mock.patch.object(conductor_steps, '_get_deployment_steps', autospec=True)
@@ -775,32 +859,104 @@ class GetValidatedStepsFromTemplatesTestCase(db_base.DbTestCase):
 
 @mock.patch.object(conductor_steps, '_get_validated_steps_from_templates',
                    autospec=True)
-class ValidateDeployTemplatesTestCase(db_base.DbTestCase):
+@mock.patch.object(conductor_steps, '_get_validated_user_deploy_steps',
+                   autospec=True)
+class ValidateUserDeployStepsAndTemplatesTestCase(db_base.DbTestCase):
 
     def setUp(self):
-        super(ValidateDeployTemplatesTestCase, self).setUp()
+        super(ValidateUserDeployStepsAndTemplatesTestCase, self).setUp()
         self.node = obj_utils.create_test_node(self.context,
                                                driver='fake-hardware')
 
-    def test_ok(self, mock_validated):
+    def test_ok(self, mock_validated_steps, mock_validated_template):
         with task_manager.acquire(
                 self.context, self.node.uuid, shared=False) as task:
-            result = conductor_steps.validate_deploy_templates(task)
+            result = conductor_steps.validate_user_deploy_steps_and_templates(
+                task, {'key': 'value'})
             self.assertIsNone(result)
-            mock_validated.assert_called_once_with(task, skip_missing=False)
+            mock_validated_template.assert_called_once_with(
+                task, skip_missing=False)
+            mock_validated_steps.assert_called_once_with(
+                task, {'key': 'value'}, skip_missing=False)
 
-    def test_skip_missing(self, mock_validated):
+    def test_skip_missing(self, mock_validated_steps, mock_validated_template):
         with task_manager.acquire(
                 self.context, self.node.uuid, shared=False) as task:
-            result = conductor_steps.validate_deploy_templates(
-                task, skip_missing=True)
+            result = conductor_steps.validate_user_deploy_steps_and_templates(
+                task, {'key': 'value'}, skip_missing=True)
             self.assertIsNone(result)
-            mock_validated.assert_called_once_with(task, skip_missing=True)
+            mock_validated_template.assert_called_once_with(
+                task, skip_missing=True)
+            mock_validated_steps.assert_called_once_with(
+                task, {'key': 'value'}, skip_missing=True)
 
-    def test_error(self, mock_validated):
+    def test_error_on_template(
+            self, mock_validated_steps, mock_validated_template):
         with task_manager.acquire(
                 self.context, self.node.uuid, shared=False) as task:
-            mock_validated.side_effect = exception.InvalidParameterValue('foo')
-            self.assertRaises(exception.InvalidParameterValue,
-                              conductor_steps.validate_deploy_templates, task)
-            mock_validated.assert_called_once_with(task, skip_missing=False)
+            mock_validated_template.side_effect =\
+                exception.InvalidParameterValue('foo')
+            self.assertRaises(
+                exception.InvalidParameterValue,
+                conductor_steps.validate_user_deploy_steps_and_templates,
+                task,
+                {'key': 'value'})
+            mock_validated_template.assert_called_once_with(
+                task, skip_missing=False)
+
+    def test_error_on_usersteps(
+            self, mock_validated_steps, mock_validated_template):
+        with task_manager.acquire(
+                self.context, self.node.uuid, shared=False) as task:
+            mock_validated_steps.side_effect =\
+                exception.InvalidParameterValue('foo')
+            self.assertRaises(
+                exception.InvalidParameterValue,
+                conductor_steps.validate_user_deploy_steps_and_templates,
+                task,
+                {'key': 'value'})
+            mock_validated_template.assert_called_once_with(
+                task, skip_missing=False)
+            mock_validated_steps.assert_called_once_with(
+                task, {'key': 'value'}, skip_missing=False)
+
+
+@mock.patch.object(conductor_steps, '_validate_user_deploy_steps',
+                   autospec=True)
+class ValidateUserDeployStepsTestCase(db_base.DbTestCase):
+
+    def setUp(self):
+        super(ValidateUserDeployStepsTestCase, self).setUp()
+        self.node = obj_utils.create_test_node(self.context,
+                                               driver='fake-hardware')
+
+    def test__get_validate_user_deploy_steps(self, mock_validated):
+        deploy_steps = [{"interface": "bios", "step": "factory_reset",
+                         "priority": 95}]
+        with task_manager.acquire(
+                self.context, self.node.uuid, shared=False) as task:
+            result = conductor_steps._get_validated_user_deploy_steps(
+                task, deploy_steps)
+            self.assertIsNotNone(result)
+            mock_validated.assert_called_once_with(task, deploy_steps,
+                                                   mock.ANY,
+                                                   skip_missing=False)
+
+    def test__get_validate_user_deploy_steps_on_node(self, mock_validated):
+        deploy_steps = [{"interface": "bios", "step": "factory_reset",
+                         "priority": 95}]
+        with task_manager.acquire(
+                self.context, self.node.uuid, shared=False) as task:
+            task.node.driver_internal_info['user_deploy_steps'] = deploy_steps
+            result = conductor_steps._get_validated_user_deploy_steps(task)
+            self.assertIsNotNone(result)
+            mock_validated.assert_called_once_with(task, deploy_steps,
+                                                   mock.ANY,
+                                                   skip_missing=False)
+
+    def test__get_validate_user_deploy_steps_no_steps(self, mock_validated):
+        with task_manager.acquire(
+                self.context, self.node.uuid, shared=False) as task:
+            result = conductor_steps._get_validated_user_deploy_steps(task)
+            self.assertEqual([], result)
+            mock_validated.assert_not_called()
diff --git a/releasenotes/notes/add-deploy-steps-arg-9d8c58559c14288c.yaml b/releasenotes/notes/add-deploy-steps-arg-9d8c58559c14288c.yaml
new file mode 100644
index 000000000..4b85dc4ec
--- /dev/null
+++ b/releasenotes/notes/add-deploy-steps-arg-9d8c58559c14288c.yaml
@@ -0,0 +1,8 @@
+---
+features:
+  - |
+    Adds support for ``deploy_steps`` parameter to provisioning endpoint
+    ``/v1/nodes/{node_ident}/states/provision``. Available and optional when
+    target is 'active' or 'rebuild'. When overlapping, these steps override
+    deploy template and driver steps. ``deploy_steps`` is a list of
+    dictionaries with required keys 'interface', 'step', 'priority' and 'args'.