diff --git a/api-ref/source/baremetal-api-v1-allocation.inc b/api-ref/source/baremetal-api-v1-allocation.inc
index a233a6dbb5..91b21c99da 100644
--- a/api-ref/source/baremetal-api-v1-allocation.inc
+++ b/api-ref/source/baremetal-api-v1-allocation.inc
@@ -36,9 +36,17 @@ the ``allocating`` state, and the process continues in the background. If it
 succeeds, the ``node_uuid`` field is populated with the Node's UUID, and the
 Node's ``instance_uuid`` field is set to the Allocation's UUID.
 
+If you want to backfill an allocation for an already deployed node, you can
+pass the UUID or name of this node to ``node``. In this case the allocation
+is created immediately, bypassing the normal allocation process. Other
+parameters must be missing or match the provided node.
+
 .. versionadded:: 1.52
     Allocation API was introduced.
 
+.. versionadded:: 1.58
+    Added support for backfilling allocations.
+
 Normal response codes: 201
 
 Error response codes: 400, 401, 403, 409, 503
@@ -54,6 +62,7 @@ Request
    - traits: req_allocation_traits
    - uuid: req_uuid
    - extra: req_extra
+   - node: req_allocation_node
 
 Request Example
 ---------------
diff --git a/api-ref/source/parameters.yaml b/api-ref/source/parameters.yaml
index 36d5a01bdd..fd88e8736c 100644
--- a/api-ref/source/parameters.yaml
+++ b/api-ref/source/parameters.yaml
@@ -443,7 +443,9 @@ allocation_node:
   type: string
 allocation_resource_class:
   description: |
-    The resource class requested for the allocation.
+    The resource class requested for the allocation. Can be ``null`` if
+    the allocation was created via backfilling and the target node did not
+    have the resource class set.
   in: body
   required: true
   type: string
@@ -1237,9 +1239,21 @@ req_allocation_name:
   in: body
   required: false
   type: string
+req_allocation_node:
+  description: |
+    The node UUID or name to create the allocation against, bypassing
+    the normal allocation process.
+
+    .. warning:: This field must not be used to request a normal allocation
+                 with one candidate node, use ``candidate_nodes`` instead.
+  in: body
+  required: false
+  type: string
 req_allocation_resource_class:
   description: |
-    The requested resource class for the allocation.
+    The requested resource class for the allocation. Can only be missing when
+    backfilling an allocation (will be set to the node's ``resource_class``
+    in such case).
   in: body
   required: true
   type: string
diff --git a/doc/source/contributor/webapi-version-history.rst b/doc/source/contributor/webapi-version-history.rst
index 15781b9e04..af1c751d55 100644
--- a/doc/source/contributor/webapi-version-history.rst
+++ b/doc/source/contributor/webapi-version-history.rst
@@ -2,8 +2,14 @@
 REST API Version History
 ========================
 
-1.57 (master)
--------------
+1.58 (Train, master)
+--------------------
+
+Added the ability to backfill allocations for already deployed nodes by
+creating an allocation with ``node`` set.
+
+1.57 (Train, master)
+--------------------
 
 Added the following new endpoint for allocation:
 
diff --git a/ironic/api/controllers/v1/allocation.py b/ironic/api/controllers/v1/allocation.py
index db714de381..7af25dc0c2 100644
--- a/ironic/api/controllers/v1/allocation.py
+++ b/ironic/api/controllers/v1/allocation.py
@@ -53,6 +53,9 @@ class Allocation(base.APIBase):
     node_uuid = wsme.wsattr(types.uuid, readonly=True)
     """The UUID of the node this allocation belongs to"""
 
+    node = wsme.wsattr(wtypes.text)
+    """The node to backfill the allocation for (POST only)"""
+
     name = wsme.wsattr(wtypes.text)
     """The logical name for this allocation"""
 
@@ -65,8 +68,7 @@ class Allocation(base.APIBase):
     last_error = wsme.wsattr(wtypes.text, readonly=True)
     """Last error that happened to this allocation"""
 
-    resource_class = wsme.wsattr(wtypes.StringType(max_length=80),
-                                 mandatory=True)
+    resource_class = wsme.wsattr(wtypes.StringType(max_length=80))
     """Requested resource class for this allocation"""
 
     # NOTE(dtantsur): candidate_nodes is a list of UUIDs on the database level,
@@ -93,6 +95,8 @@ class Allocation(base.APIBase):
     @staticmethod
     def _convert_with_links(allocation, url):
         """Add links to the allocation."""
+        # This field is only used in POST, never return it.
+        allocation.node = wsme.Unset
         allocation.links = [
             link.Link.make_link('self', url, 'allocations', allocation.uuid),
             link.Link.make_link('bookmark', url, 'allocations',
@@ -334,10 +338,6 @@ class AllocationsController(pecan.rest.RestController):
         cdict = context.to_policy_values()
         policy.authorize('baremetal:allocation:create', cdict, cdict)
 
-        if allocation.node_uuid is not wtypes.Unset:
-            msg = _("Cannot set node_uuid when creating an allocation")
-            raise exception.Invalid(msg)
-
         if (allocation.name
                 and not api_utils.is_valid_logical_name(allocation.name)):
             msg = _("Cannot create allocation with invalid name "
@@ -348,6 +348,27 @@ class AllocationsController(pecan.rest.RestController):
             for trait in allocation.traits:
                 api_utils.validate_trait(trait)
 
+        node = None
+        if allocation.node is not wtypes.Unset:
+            if api_utils.allow_allocation_backfill():
+                try:
+                    node = api_utils.get_rpc_node(allocation.node)
+                except exception.NodeNotFound as exc:
+                    exc.code = http_client.BAD_REQUEST
+                    raise
+            else:
+                msg = _("Cannot set node when creating an allocation "
+                        "in this API version")
+                raise exception.Invalid(msg)
+
+        if not allocation.resource_class:
+            if node:
+                allocation.resource_class = node.resource_class
+            else:
+                msg = _("The resource_class field is mandatory when not "
+                        "backfilling")
+                raise exception.Invalid(msg)
+
         if allocation.candidate_nodes:
             # Convert nodes from names to UUIDs and check their validity
             try:
@@ -365,10 +386,19 @@ class AllocationsController(pecan.rest.RestController):
 
         # NOTE(yuriyz): UUID is mandatory for notifications payload
         if not all_dict.get('uuid'):
-            all_dict['uuid'] = uuidutils.generate_uuid()
+            if node and node.instance_uuid:
+                # When backfilling without UUID requested, assume that the
+                # target instance_uuid is the desired UUID
+                all_dict['uuid'] = node.instance_uuid
+            else:
+                all_dict['uuid'] = uuidutils.generate_uuid()
 
         new_allocation = objects.Allocation(context, **all_dict)
-        topic = pecan.request.rpcapi.get_random_topic()
+        if node:
+            new_allocation.node_id = node.id
+            topic = pecan.request.rpcapi.get_topic_for(node)
+        else:
+            topic = pecan.request.rpcapi.get_random_topic()
 
         notify.emit_start_notification(context, new_allocation, 'create')
         with notify.handle_error_notification(context, new_allocation,
diff --git a/ironic/api/controllers/v1/utils.py b/ironic/api/controllers/v1/utils.py
index ebe633aa4b..427cd21254 100644
--- a/ironic/api/controllers/v1/utils.py
+++ b/ironic/api/controllers/v1/utils.py
@@ -1169,3 +1169,11 @@ def allow_allocation_update():
     Version 1.57 of the API added support for updating an allocation.
     """
     return pecan.request.version.minor >= versions.MINOR_57_ALLOCATION_UPDATE
+
+
+def allow_allocation_backfill():
+    """Check if backfilling allocations is allowed.
+
+    Version 1.58 of the API added support for backfilling allocations.
+    """
+    return pecan.request.version.minor >= versions.MINOR_58_ALLOCATION_BACKFILL
diff --git a/ironic/api/controllers/v1/versions.py b/ironic/api/controllers/v1/versions.py
index 1865c90290..cadd2b169b 100644
--- a/ironic/api/controllers/v1/versions.py
+++ b/ironic/api/controllers/v1/versions.py
@@ -95,6 +95,7 @@ BASE_VERSION = 1
 # v1.55: Add deploy templates API.
 # v1.56: Add support for building configdrives.
 # v1.57: Add support for updating an exisiting allocation.
+# v1.58: Add support for backfilling allocations.
 
 MINOR_0_JUNO = 0
 MINOR_1_INITIAL_VERSION = 1
@@ -154,6 +155,7 @@ MINOR_54_EVENTS = 54
 MINOR_55_DEPLOY_TEMPLATES = 55
 MINOR_56_BUILD_CONFIGDRIVE = 56
 MINOR_57_ALLOCATION_UPDATE = 57
+MINOR_58_ALLOCATION_BACKFILL = 58
 
 # When adding another version, update:
 # - MINOR_MAX_VERSION
@@ -161,7 +163,7 @@ MINOR_57_ALLOCATION_UPDATE = 57
 #   explanation of what changed in the new version
 # - common/release_mappings.py, RELEASE_MAPPING['master']['api']
 
-MINOR_MAX_VERSION = MINOR_57_ALLOCATION_UPDATE
+MINOR_MAX_VERSION = MINOR_58_ALLOCATION_BACKFILL
 
 # 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 5fa4dbb8f4..ad57044fe8 100644
--- a/ironic/common/release_mappings.py
+++ b/ironic/common/release_mappings.py
@@ -163,7 +163,7 @@ RELEASE_MAPPING = {
         }
     },
     'master': {
-        'api': '1.57',
+        'api': '1.58',
         'rpc': '1.48',
         'objects': {
             'Allocation': ['1.0'],
diff --git a/ironic/conductor/allocations.py b/ironic/conductor/allocations.py
index c5dd60534b..5c1a7024a2 100644
--- a/ironic/conductor/allocations.py
+++ b/ironic/conductor/allocations.py
@@ -17,6 +17,7 @@ import random
 from ironic_lib import metrics_utils
 from oslo_config import cfg
 from oslo_log import log
+from oslo_utils import excutils
 import retrying
 
 from ironic.common import exception
@@ -85,6 +86,7 @@ def verify_node_for_deallocation(node, allocation):
 def _allocation_failed(allocation, reason):
     """Failure handler for the allocation."""
     try:
+        allocation.node_id = None
         allocation.state = states.ERROR
         allocation.last_error = str(reason)
         allocation.save()
@@ -231,3 +233,87 @@ def _allocate_node(context, allocation, nodes):
         error = _('all nodes were filtered out during reservation')
 
     raise exception.AllocationFailed(uuid=allocation.uuid, error=error)
+
+
+def backfill_allocation(context, allocation, node_id):
+    """Assign the previously allocated node to the node allocation.
+
+    This is not the actual allocation process, but merely backfilling of
+    allocation_uuid for a previously allocated node.
+
+    :param context: an admin context
+    :param allocation: an allocation object associated with the node
+    :param node_id: An ID of the node.
+    :raises: AllocationFailed if the node does not match the allocation
+    :raises: NodeAssociated if the node is already associated with another
+        instance or allocation.
+    :raises: InstanceAssociated if the allocation's UUID is already used
+        on another node as instance_uuid.
+    :raises: NodeNotFound if the node with the provided ID cannot be found.
+    """
+    try:
+        _do_backfill_allocation(context, allocation, node_id)
+    except (exception.AllocationFailed,
+            exception.InstanceAssociated,
+            exception.NodeAssociated,
+            exception.NodeNotFound) as exc:
+        with excutils.save_and_reraise_exception():
+            LOG.error(str(exc))
+            _allocation_failed(allocation, exc)
+    except Exception as exc:
+        with excutils.save_and_reraise_exception():
+            LOG.exception("Unexpected exception during backfilling of "
+                          "allocation %s", allocation.uuid)
+            reason = _("Unexpected exception during allocation: %s") % exc
+            _allocation_failed(allocation, reason)
+
+
+def _do_backfill_allocation(context, allocation, node_id):
+    with task_manager.acquire(context, node_id,
+                              purpose='allocation backfilling') as task:
+        node = task.node
+
+        errors = []
+
+        # NOTE(dtantsur): this feature is not designed to bypass the allocation
+        # mechanism, but to backfill allocations for active nodes, hence this
+        # check.
+        if node.provision_state != states.ACTIVE:
+            errors.append(_('Node must be in the "active" state, but the '
+                            'current state is "%s"') % node.provision_state)
+
+        # NOTE(dtantsur): double-check that the node is still suitable.
+        if (allocation.resource_class
+                and node.resource_class != allocation.resource_class):
+            errors.append(_('Resource class %(curr)s does not match '
+                            'the requested resource class %(rsc)s')
+                          % {'curr': node.resource_class,
+                             'rsc': allocation.resource_class})
+        if (allocation.traits
+                and not _traits_match(set(allocation.traits), node)):
+            errors.append(_('List of traits %(curr)s does not match '
+                            'the requested traits %(traits)s')
+                          % {'curr': node.traits,
+                             'traits': allocation.traits})
+        if (allocation.candidate_nodes
+                and node.uuid not in allocation.candidate_nodes):
+            errors.append(_('Candidate nodes must be empty or contain the '
+                            'target node, but got %s')
+                          % allocation.candidate_nodes)
+
+        if errors:
+            error = _('Cannot backfill an allocation for node %(node)s: '
+                      '%(errors)s') % {'node': node.uuid,
+                                       'errors': '; '.join(errors)}
+            raise exception.AllocationFailed(uuid=allocation.uuid, error=error)
+
+        allocation.node_id = task.node.id
+        allocation.state = states.ACTIVE
+        # NOTE(dtantsur): the node.instance_uuid and allocation_id are
+        # updated inside of the save() call within the same
+        # transaction to avoid races. NodeAssociated can be raised if
+        # another process allocates this node first.
+        allocation.save()
+        LOG.info('Node %(node)s has been successfully reserved for '
+                 'allocation %(uuid)s',
+                 {'node': node.uuid, 'uuid': allocation.uuid})
diff --git a/ironic/conductor/manager.py b/ironic/conductor/manager.py
index 8dd59caece..46358dd416 100644
--- a/ironic/conductor/manager.py
+++ b/ironic/conductor/manager.py
@@ -3498,7 +3498,10 @@ class ConductorManager(base_manager.BaseConductorManager):
                                           trait=trait)
 
     @METRICS.timer('ConductorManager.create_allocation')
-    @messaging.expected_exceptions(exception.InvalidParameterValue)
+    @messaging.expected_exceptions(exception.InvalidParameterValue,
+                                   exception.NodeAssociated,
+                                   exception.InstanceAssociated,
+                                   exception.NodeNotFound)
     def create_allocation(self, context, allocation):
         """Create an allocation in database.
 
@@ -3507,18 +3510,37 @@ class ConductorManager(base_manager.BaseConductorManager):
                            allocation object.
         :returns: created allocation object.
         :raises: InvalidParameterValue if some fields fail validation.
+        :raises: NodeAssociated if allocation backfill is requested for a node
+            that is associated with another instance.
+        :raises: InstanceAssociated if allocation backfill is requested, but
+            the allocation UUID is already used as instance_uuid on another
+            node.
+        :raises: NodeNotFound if allocation backfill is requested for a node
+            that cannot be found.
         """
         LOG.debug("RPC create_allocation called for allocation %s.",
                   allocation.uuid)
         allocation.conductor_affinity = self.conductor.id
+        # Allocation backfilling is handled separately, remove node_id for now.
+        # Cannot use plain getattr here since oslo.versionedobjects raise
+        # NotImplementedError instead of AttributeError (because life is pain).
+        if 'node_id' in allocation and allocation.node_id:
+            node_id = allocation.node_id
+            allocation.node_id = None
+        else:
+            node_id = None
         allocation.create()
 
-        # Spawn an asynchronous worker to process the allocation. Copy it to
-        # avoid data races.
-        self._spawn_worker(allocations.do_allocate,
-                           context, allocation.obj_clone())
+        if node_id:
+            # This is a fast operation and should be done synchronously
+            allocations.backfill_allocation(context, allocation, node_id)
+        else:
+            # Spawn an asynchronous worker to process the allocation. Copy it
+            # to avoid data races.
+            self._spawn_worker(allocations.do_allocate,
+                               context, allocation.obj_clone())
 
-        # Return the unfinished allocation
+        # Return the current status of the allocation
         return allocation
 
     @METRICS.timer('ConductorManager.destroy_allocation')
diff --git a/ironic/db/sqlalchemy/api.py b/ironic/db/sqlalchemy/api.py
index 23c29198c0..6d019a5958 100644
--- a/ironic/db/sqlalchemy/api.py
+++ b/ironic/db/sqlalchemy/api.py
@@ -1758,7 +1758,7 @@ class Connection(api.Connection):
                 ref.update(values)
                 instance_uuid = ref.uuid
 
-                if 'node_id' in values and update_node:
+                if values.get('node_id') and update_node:
                     node = model_query(models.Node, session=session).filter_by(
                         id=ref.node_id).with_lockmode('update').one()
                     node_uuid = node.uuid
diff --git a/ironic/tests/unit/api/controllers/v1/test_allocation.py b/ironic/tests/unit/api/controllers/v1/test_allocation.py
index 20aa72f981..162ded7d20 100644
--- a/ironic/tests/unit/api/controllers/v1/test_allocation.py
+++ b/ironic/tests/unit/api/controllers/v1/test_allocation.py
@@ -561,7 +561,11 @@ class TestPatch(test_api_base.BaseApiTest):
         self.assertTrue(response.json['error_message'])
 
 
-def _create_locally(_api, _ctx, allocation, _topic):
+def _create_locally(_api, _ctx, allocation, topic):
+    if 'node_id' in allocation and allocation.node_id:
+        assert topic == 'node-topic', topic
+    else:
+        assert topic == 'some-topic', topic
     allocation.create()
     return allocation
 
@@ -576,6 +580,10 @@ class TestPost(test_api_base.BaseApiTest):
             fixtures.MockPatchObject(rpcapi.ConductorAPI, 'get_random_topic')
         ).mock
         self.mock_get_topic.return_value = 'some-topic'
+        self.mock_get_topic_for_node = self.useFixture(
+            fixtures.MockPatchObject(rpcapi.ConductorAPI, 'get_topic_for')
+        ).mock
+        self.mock_get_topic_for_node.return_value = 'node-topic'
 
     @mock.patch.object(notification_utils, '_emit_api_notification')
     @mock.patch.object(timeutils, 'utcnow', autospec=True)
@@ -591,6 +599,7 @@ class TestPost(test_api_base.BaseApiTest):
         self.assertIsNone(response.json['node_uuid'])
         self.assertEqual([], response.json['candidate_nodes'])
         self.assertEqual([], response.json['traits'])
+        self.assertNotIn('node', response.json)
         result = self.get_json('/allocations/%s' % adict['uuid'],
                                headers=self.headers)
         self.assertEqual(adict['uuid'], result['uuid'])
@@ -598,6 +607,7 @@ class TestPost(test_api_base.BaseApiTest):
         self.assertIsNone(result['node_uuid'])
         self.assertEqual([], result['candidate_nodes'])
         self.assertEqual([], result['traits'])
+        self.assertNotIn('node', result)
         return_created_at = timeutils.parse_isotime(
             result['created_at']).replace(tzinfo=None)
         self.assertEqual(test_time, return_created_at)
@@ -700,7 +710,7 @@ class TestPost(test_api_base.BaseApiTest):
                                   headers=self.headers)
         self.assertEqual(http_client.BAD_REQUEST, response.status_int)
         self.assertEqual('application/json', response.content_type)
-        self.assertTrue(response.json['error_message'])
+        self.assertIn('resource_class', response.json['error_message'])
 
     def test_create_allocation_resource_class_too_long(self):
         adict = apiutils.allocation_post_data()
@@ -785,15 +795,87 @@ class TestPost(test_api_base.BaseApiTest):
         self.assertEqual('application/json', response.content_type)
         self.assertEqual(http_client.METHOD_NOT_ALLOWED, response.status_int)
 
-    def test_create_with_node_uuid_not_allowed(self):
+    def test_create_node_uuid_not_allowed(self):
+        node = obj_utils.create_test_node(self.context)
         adict = apiutils.allocation_post_data()
-        adict['node_uuid'] = uuidutils.generate_uuid()
+        adict['node_uuid'] = node.uuid
         response = self.post_json('/allocations', adict, expect_errors=True,
                                   headers=self.headers)
         self.assertEqual(http_client.BAD_REQUEST, response.status_int)
         self.assertEqual('application/json', response.content_type)
         self.assertTrue(response.json['error_message'])
 
+    def test_backfill(self):
+        node = obj_utils.create_test_node(self.context)
+        adict = apiutils.allocation_post_data(node=node.uuid)
+        response = self.post_json('/allocations', adict,
+                                  headers=self.headers)
+        self.assertEqual(http_client.CREATED, response.status_int)
+        self.assertNotIn('node', response.json)
+        result = self.get_json('/allocations/%s' % adict['uuid'],
+                               headers=self.headers)
+        self.assertEqual(adict['uuid'], result['uuid'])
+        self.assertEqual(node.uuid, result['node_uuid'])
+        self.assertNotIn('node', result)
+
+    def test_backfill_with_name(self):
+        node = obj_utils.create_test_node(self.context, name='backfill-me')
+        adict = apiutils.allocation_post_data(node=node.name)
+        response = self.post_json('/allocations', adict,
+                                  headers=self.headers)
+        self.assertEqual(http_client.CREATED, response.status_int)
+        self.assertNotIn('node', response.json)
+        result = self.get_json('/allocations/%s' % adict['uuid'],
+                               headers=self.headers)
+        self.assertEqual(adict['uuid'], result['uuid'])
+        self.assertEqual(node.uuid, result['node_uuid'])
+        self.assertNotIn('node', result)
+
+    def test_backfill_without_resource_class(self):
+        node = obj_utils.create_test_node(self.context,
+                                          resource_class='bm-super')
+        adict = {'node': node.uuid}
+        response = self.post_json('/allocations', adict,
+                                  headers=self.headers)
+        self.assertEqual(http_client.CREATED, response.status_int)
+        result = self.get_json('/allocations/%s' % response.json['uuid'],
+                               headers=self.headers)
+        self.assertEqual(node.uuid, result['node_uuid'])
+        self.assertEqual('bm-super', result['resource_class'])
+
+    def test_backfill_copy_instance_uuid(self):
+        uuid = uuidutils.generate_uuid()
+        node = obj_utils.create_test_node(self.context,
+                                          instance_uuid=uuid,
+                                          resource_class='bm-super')
+        adict = {'node': node.uuid}
+        response = self.post_json('/allocations', adict,
+                                  headers=self.headers)
+        self.assertEqual(http_client.CREATED, response.status_int)
+        result = self.get_json('/allocations/%s' % response.json['uuid'],
+                               headers=self.headers)
+        self.assertEqual(uuid, result['uuid'])
+        self.assertEqual(node.uuid, result['node_uuid'])
+        self.assertEqual('bm-super', result['resource_class'])
+
+    def test_backfill_node_not_found(self):
+        adict = apiutils.allocation_post_data(node=uuidutils.generate_uuid())
+        response = self.post_json('/allocations', adict, expect_errors=True,
+                                  headers=self.headers)
+        self.assertEqual(http_client.BAD_REQUEST, response.status_int)
+        self.assertEqual('application/json', response.content_type)
+        self.assertTrue(response.json['error_message'])
+
+    def test_backfill_not_allowed(self):
+        node = obj_utils.create_test_node(self.context)
+        headers = {api_base.Version.string: '1.57'}
+        adict = apiutils.allocation_post_data(node=node.uuid)
+        response = self.post_json('/allocations', adict, expect_errors=True,
+                                  headers=headers)
+        self.assertEqual(http_client.BAD_REQUEST, response.status_int)
+        self.assertEqual('application/json', response.content_type)
+        self.assertTrue(response.json['error_message'])
+
 
 @mock.patch.object(rpcapi.ConductorAPI, 'destroy_allocation')
 class TestDelete(test_api_base.BaseApiTest):
diff --git a/ironic/tests/unit/api/utils.py b/ironic/tests/unit/api/utils.py
index 7addd2e8d0..30444d00ab 100644
--- a/ironic/tests/unit/api/utils.py
+++ b/ironic/tests/unit/api/utils.py
@@ -189,12 +189,16 @@ def post_get_test_portgroup(**kw):
 
 
 _ALLOCATION_POST_FIELDS = {'resource_class', 'uuid', 'traits',
-                           'candidate_nodes', 'name', 'extra'}
+                           'candidate_nodes', 'name', 'extra',
+                           'node'}
 
 
-def allocation_post_data(**kw):
+def allocation_post_data(node=None, **kw):
     """Return an Allocation object without internal attributes."""
     allocation = db_utils.get_test_allocation(**kw)
+    if node:
+        # This is not a database field, so it has to be handled explicitly
+        allocation['node'] = node
     return {key: value for key, value in allocation.items()
             if key in _ALLOCATION_POST_FIELDS}
 
diff --git a/ironic/tests/unit/conductor/test_allocations.py b/ironic/tests/unit/conductor/test_allocations.py
index d490e99e27..dbdb01e866 100644
--- a/ironic/tests/unit/conductor/test_allocations.py
+++ b/ironic/tests/unit/conductor/test_allocations.py
@@ -59,6 +59,30 @@ class AllocationTestCase(mgr_utils.ServiceSetUpMixin, db_base.DbTestCase):
                                            allocations.do_allocate,
                                            self.context, mock.ANY)
 
+    @mock.patch.object(manager.ConductorManager, '_spawn_worker', mock.Mock())
+    @mock.patch.object(allocations, 'backfill_allocation', autospec=True)
+    def test_create_allocation_with_node_id(self, mock_backfill):
+        node = obj_utils.create_test_node(self.context)
+        allocation = obj_utils.get_test_allocation(self.context,
+                                                   node_id=node.id)
+
+        self._start_service()
+        res = self.service.create_allocation(self.context, allocation)
+        mock_backfill.assert_called_once_with(self.context,
+                                              allocation,
+                                              node.id)
+
+        self.assertEqual('allocating', res['state'])
+        self.assertIsNotNone(res['uuid'])
+        self.assertEqual(self.service.conductor.id, res['conductor_affinity'])
+        # create_allocation purges node_id, and since we stub out
+        # backfill_allocation, it does not get populated.
+        self.assertIsNone(res['node_id'])
+        res = objects.Allocation.get_by_uuid(self.context, allocation['uuid'])
+        self.assertEqual('allocating', res['state'])
+        self.assertIsNotNone(res['uuid'])
+        self.assertEqual(self.service.conductor.id, res['conductor_affinity'])
+
     def test_destroy_allocation_without_node(self):
         allocation = obj_utils.create_test_allocation(self.context)
         self.service.destroy_allocation(self.context, allocation)
@@ -422,3 +446,241 @@ class DoAllocateTestCase(db_base.DbTestCase):
 
         # All nodes are filtered out on the database level.
         self.assertFalse(mock_acquire.called)
+
+
+class BackfillAllocationTestCase(db_base.DbTestCase):
+    def test_with_associated_node(self):
+        uuid = uuidutils.generate_uuid()
+        node = obj_utils.create_test_node(self.context,
+                                          instance_uuid=uuid,
+                                          resource_class='x-large',
+                                          provision_state='active')
+        allocation = obj_utils.create_test_allocation(self.context,
+                                                      uuid=uuid,
+                                                      resource_class='x-large')
+
+        allocations.backfill_allocation(self.context, allocation, node.id)
+
+        allocation = objects.Allocation.get_by_uuid(self.context,
+                                                    allocation['uuid'])
+        self.assertIsNone(allocation['last_error'])
+        self.assertEqual('active', allocation['state'])
+
+        node = objects.Node.get_by_uuid(self.context, node['uuid'])
+        self.assertEqual(allocation['uuid'], node['instance_uuid'])
+        self.assertEqual(allocation['id'], node['allocation_id'])
+
+    def test_with_unassociated_node(self):
+        node = obj_utils.create_test_node(self.context,
+                                          instance_uuid=None,
+                                          resource_class='x-large',
+                                          provision_state='active')
+        allocation = obj_utils.create_test_allocation(self.context,
+                                                      resource_class='x-large')
+
+        allocations.backfill_allocation(self.context, allocation, node.id)
+
+        allocation = objects.Allocation.get_by_uuid(self.context,
+                                                    allocation['uuid'])
+        self.assertIsNone(allocation['last_error'])
+        self.assertEqual('active', allocation['state'])
+
+        node = objects.Node.get_by_uuid(self.context, node['uuid'])
+        self.assertEqual(allocation['uuid'], node['instance_uuid'])
+        self.assertEqual(allocation['id'], node['allocation_id'])
+
+    def test_with_candidate_nodes(self):
+        node = obj_utils.create_test_node(self.context,
+                                          instance_uuid=None,
+                                          resource_class='x-large',
+                                          provision_state='active')
+        allocation = obj_utils.create_test_allocation(
+            self.context, candidate_nodes=[node.uuid],
+            resource_class='x-large')
+
+        allocations.backfill_allocation(self.context, allocation, node.id)
+
+        allocation = objects.Allocation.get_by_uuid(self.context,
+                                                    allocation['uuid'])
+        self.assertIsNone(allocation['last_error'])
+        self.assertEqual('active', allocation['state'])
+
+        node = objects.Node.get_by_uuid(self.context, node['uuid'])
+        self.assertEqual(allocation['uuid'], node['instance_uuid'])
+        self.assertEqual(allocation['id'], node['allocation_id'])
+
+    def test_without_resource_class(self):
+        uuid = uuidutils.generate_uuid()
+        node = obj_utils.create_test_node(self.context,
+                                          instance_uuid=uuid,
+                                          resource_class='x-large',
+                                          provision_state='active')
+        allocation = obj_utils.create_test_allocation(self.context,
+                                                      uuid=uuid,
+                                                      resource_class=None)
+
+        allocations.backfill_allocation(self.context, allocation, node.id)
+
+        allocation = objects.Allocation.get_by_uuid(self.context,
+                                                    allocation['uuid'])
+        self.assertIsNone(allocation['last_error'])
+        self.assertEqual('active', allocation['state'])
+
+        node = objects.Node.get_by_uuid(self.context, node['uuid'])
+        self.assertEqual(allocation['uuid'], node['instance_uuid'])
+        self.assertEqual(allocation['id'], node['allocation_id'])
+
+    def test_node_associated_with_another_instance(self):
+        other_uuid = uuidutils.generate_uuid()
+        node = obj_utils.create_test_node(self.context,
+                                          instance_uuid=other_uuid,
+                                          resource_class='x-large',
+                                          provision_state='active')
+        allocation = obj_utils.create_test_allocation(self.context,
+                                                      resource_class='x-large')
+
+        self.assertRaises(exception.NodeAssociated,
+                          allocations.backfill_allocation,
+                          self.context, allocation, node.id)
+
+        allocation = objects.Allocation.get_by_uuid(self.context,
+                                                    allocation['uuid'])
+        self.assertEqual('error', allocation['state'])
+        self.assertIn('associated', allocation['last_error'])
+        self.assertIsNone(allocation['node_id'])
+
+        node = objects.Node.get_by_uuid(self.context, node['uuid'])
+        self.assertEqual(other_uuid, node['instance_uuid'])
+        self.assertIsNone(node['allocation_id'])
+
+    def test_non_existing_node(self):
+        allocation = obj_utils.create_test_allocation(self.context,
+                                                      resource_class='x-large')
+
+        self.assertRaises(exception.NodeNotFound,
+                          allocations.backfill_allocation,
+                          self.context, allocation, 42)
+
+        allocation = objects.Allocation.get_by_uuid(self.context,
+                                                    allocation['uuid'])
+        self.assertEqual('error', allocation['state'])
+        self.assertIn('Node 42 could not be found', allocation['last_error'])
+        self.assertIsNone(allocation['node_id'])
+
+    def test_uuid_associated_with_another_instance(self):
+        uuid = uuidutils.generate_uuid()
+        obj_utils.create_test_node(self.context,
+                                   uuid=uuidutils.generate_uuid(),
+                                   instance_uuid=uuid,
+                                   resource_class='x-large',
+                                   provision_state='active')
+        node = obj_utils.create_test_node(self.context,
+                                          uuid=uuidutils.generate_uuid(),
+                                          resource_class='x-large',
+                                          provision_state='active')
+        allocation = obj_utils.create_test_allocation(self.context,
+                                                      uuid=uuid,
+                                                      resource_class='x-large')
+
+        self.assertRaises(exception.InstanceAssociated,
+                          allocations.backfill_allocation,
+                          self.context, allocation, node.id)
+
+        allocation = objects.Allocation.get_by_uuid(self.context,
+                                                    allocation['uuid'])
+        self.assertEqual('error', allocation['state'])
+        self.assertIn('associated', allocation['last_error'])
+        self.assertIsNone(allocation['node_id'])
+
+        node = objects.Node.get_by_uuid(self.context, node['uuid'])
+        self.assertIsNone(node['instance_uuid'])
+        self.assertIsNone(node['allocation_id'])
+
+    def test_resource_class_mismatch(self):
+        node = obj_utils.create_test_node(self.context,
+                                          resource_class='x-small',
+                                          provision_state='active')
+        allocation = obj_utils.create_test_allocation(self.context,
+                                                      resource_class='x-large')
+
+        self.assertRaises(exception.AllocationFailed,
+                          allocations.backfill_allocation,
+                          self.context, allocation, node.id)
+
+        allocation = objects.Allocation.get_by_uuid(self.context,
+                                                    allocation['uuid'])
+        self.assertEqual('error', allocation['state'])
+        self.assertIn('resource class', allocation['last_error'])
+        self.assertIsNone(allocation['node_id'])
+
+        node = objects.Node.get_by_uuid(self.context, node['uuid'])
+        self.assertIsNone(node['instance_uuid'])
+        self.assertIsNone(node['allocation_id'])
+
+    def test_traits_mismatch(self):
+        node = obj_utils.create_test_node(self.context,
+                                          resource_class='x-large',
+                                          provision_state='active')
+        db_utils.create_test_node_traits(['tr1', 'tr2'], node_id=node.id)
+        allocation = obj_utils.create_test_allocation(self.context,
+                                                      resource_class='x-large',
+                                                      traits=['tr1', 'tr3'])
+
+        self.assertRaises(exception.AllocationFailed,
+                          allocations.backfill_allocation,
+                          self.context, allocation, node.id)
+
+        allocation = objects.Allocation.get_by_uuid(self.context,
+                                                    allocation['uuid'])
+        self.assertEqual('error', allocation['state'])
+        self.assertIn('traits', allocation['last_error'])
+        self.assertIsNone(allocation['node_id'])
+
+        node = objects.Node.get_by_uuid(self.context, node['uuid'])
+        self.assertIsNone(node['instance_uuid'])
+        self.assertIsNone(node['allocation_id'])
+
+    def test_state_not_active(self):
+        node = obj_utils.create_test_node(self.context,
+                                          resource_class='x-large',
+                                          provision_state='available')
+        allocation = obj_utils.create_test_allocation(self.context,
+                                                      resource_class='x-large')
+
+        self.assertRaises(exception.AllocationFailed,
+                          allocations.backfill_allocation,
+                          self.context, allocation, node.id)
+
+        allocation = objects.Allocation.get_by_uuid(self.context,
+                                                    allocation['uuid'])
+        self.assertEqual('error', allocation['state'])
+        self.assertIn('must be in the "active" state',
+                      allocation['last_error'])
+        self.assertIsNone(allocation['node_id'])
+
+        node = objects.Node.get_by_uuid(self.context, node['uuid'])
+        self.assertIsNone(node['instance_uuid'])
+        self.assertIsNone(node['allocation_id'])
+
+    def test_candidate_nodes_mismatch(self):
+        node = obj_utils.create_test_node(self.context,
+                                          resource_class='x-large',
+                                          provision_state='active')
+        allocation = obj_utils.create_test_allocation(
+            self.context,
+            candidate_nodes=[uuidutils.generate_uuid()],
+            resource_class='x-large')
+
+        self.assertRaises(exception.AllocationFailed,
+                          allocations.backfill_allocation,
+                          self.context, allocation, node.id)
+
+        allocation = objects.Allocation.get_by_uuid(self.context,
+                                                    allocation['uuid'])
+        self.assertEqual('error', allocation['state'])
+        self.assertIn('Candidate nodes', allocation['last_error'])
+        self.assertIsNone(allocation['node_id'])
+
+        node = objects.Node.get_by_uuid(self.context, node['uuid'])
+        self.assertIsNone(node['instance_uuid'])
+        self.assertIsNone(node['allocation_id'])
diff --git a/releasenotes/notes/allocation-backfill-c31e84c5fcf24216.yaml b/releasenotes/notes/allocation-backfill-c31e84c5fcf24216.yaml
new file mode 100644
index 0000000000..88b587f7f6
--- /dev/null
+++ b/releasenotes/notes/allocation-backfill-c31e84c5fcf24216.yaml
@@ -0,0 +1,5 @@
+---
+features:
+  - |
+    API version 1.58 allows backfilling allocations for existing deployed nodes
+    by providing ``node`` to ``POST /v1/allocations``.