diff --git a/doc/source/contributor/webapi-version-history.rst b/doc/source/contributor/webapi-version-history.rst
index 3c6729178c..1e95f61893 100644
--- a/doc/source/contributor/webapi-version-history.rst
+++ b/doc/source/contributor/webapi-version-history.rst
@@ -2,6 +2,23 @@
 REST API Version History
 ========================
 
+1.52 (Stein, master)
+--------------------
+
+Added allocation API, allowing reserving a node for deployment based on
+resource class and traits. The new endpoints are:
+
+* ``POST /v1/allocations`` to request an allocation.
+* ``GET /v1/allocations`` to list all allocations.
+* ``GET /v1/allocations/<ID or name>`` to retrieve the allocation details.
+* ``GET /v1/nodes/<ID or name>/allocation`` to retrieve an allocation
+  associated with the node.
+* ``DELETE /v1/allocations/<ID or name`` to remove the allocation.
+* ``DELETE /v1/nodes/<ID or name/allocation`` to remove an allocation
+  associated with the node.
+
+Also added a new field ``allocation_uuid`` to the node resource.
+
 1.51 (Stein, master)
 --------------------
 
diff --git a/ironic/api/controllers/v1/__init__.py b/ironic/api/controllers/v1/__init__.py
index 074f6d84ba..52a7cfe6c6 100644
--- a/ironic/api/controllers/v1/__init__.py
+++ b/ironic/api/controllers/v1/__init__.py
@@ -25,6 +25,7 @@ from wsme import types as wtypes
 
 from ironic.api.controllers import base
 from ironic.api.controllers import link
+from ironic.api.controllers.v1 import allocation
 from ironic.api.controllers.v1 import chassis
 from ironic.api.controllers.v1 import conductor
 from ironic.api.controllers.v1 import driver
@@ -104,6 +105,9 @@ class V1(base.APIBase):
     conductors = [link.Link]
     """Links to the conductors resource"""
 
+    allocations = [link.Link]
+    """Links to the allocations resource"""
+
     version = version.Version
     """Version discovery information."""
 
@@ -191,6 +195,15 @@ class V1(base.APIBase):
                                                  'conductors', '',
                                                  bookmark=True)
                              ]
+        if utils.allow_allocations():
+            v1.allocations = [link.Link.make_link('self',
+                                                  pecan.request.public_url,
+                                                  'allocations', ''),
+                              link.Link.make_link('bookmark',
+                                                  pecan.request.public_url,
+                                                  'allocations', '',
+                                                  bookmark=True)
+                              ]
         v1.version = version.default_version()
         return v1
 
@@ -207,6 +220,7 @@ class Controller(rest.RestController):
     lookup = ramdisk.LookupController()
     heartbeat = ramdisk.HeartbeatController()
     conductors = conductor.ConductorsController()
+    allocations = allocation.AllocationsController()
 
     @expose.expose(V1)
     def get(self):
diff --git a/ironic/api/controllers/v1/allocation.py b/ironic/api/controllers/v1/allocation.py
new file mode 100644
index 0000000000..af346dd208
--- /dev/null
+++ b/ironic/api/controllers/v1/allocation.py
@@ -0,0 +1,465 @@
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+import datetime
+
+from ironic_lib import metrics_utils
+from oslo_utils import uuidutils
+import pecan
+from six.moves import http_client
+import wsme
+from wsme import types as wtypes
+
+from ironic.api.controllers import base
+from ironic.api.controllers import link
+from ironic.api.controllers.v1 import collection
+from ironic.api.controllers.v1 import notification_utils as notify
+from ironic.api.controllers.v1 import types
+from ironic.api.controllers.v1 import utils as api_utils
+from ironic.api import expose
+from ironic.common import exception
+from ironic.common.i18n import _
+from ironic.common import policy
+from ironic.common import states as ir_states
+from ironic import objects
+
+METRICS = metrics_utils.get_metrics_logger(__name__)
+
+
+class Allocation(base.APIBase):
+    """API representation of an allocation.
+
+    This class enforces type checking and value constraints, and converts
+    between the internal object model and the API representation of a
+    allocation.
+    """
+
+    uuid = types.uuid
+    """Unique UUID for this allocation"""
+
+    extra = {wtypes.text: types.jsontype}
+    """This allocation's meta data"""
+
+    node_uuid = wsme.wsattr(types.uuid, readonly=True)
+    """The UUID of the node this allocation belongs to"""
+
+    name = wsme.wsattr(wtypes.text)
+    """The logical name for this allocation"""
+
+    links = wsme.wsattr([link.Link], readonly=True)
+    """A list containing a self link and associated allocation links"""
+
+    state = wsme.wsattr(wtypes.text, readonly=True)
+    """The current state of the allocation"""
+
+    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)
+    """Requested resource class for this allocation"""
+
+    # NOTE(dtantsur): candidate_nodes is a list of UUIDs on the database level,
+    # but the API level also accept names, converting them on fly.
+    candidate_nodes = wsme.wsattr([wtypes.text])
+    """Candidate nodes for this allocation"""
+
+    traits = wsme.wsattr([wtypes.text])
+    """Requested traits for the allocation"""
+
+    def __init__(self, **kwargs):
+        self.fields = []
+        fields = list(objects.Allocation.fields)
+        # NOTE: node_uuid is not part of objects.Allocation.fields
+        #       because it's an API-only attribute
+        fields.append('node_uuid')
+        for field in fields:
+            # Skip fields we do not expose.
+            if not hasattr(self, field):
+                continue
+            self.fields.append(field)
+            setattr(self, field, kwargs.get(field, wtypes.Unset))
+
+    @staticmethod
+    def _convert_with_links(allocation, url):
+        """Add links to the allocation."""
+        allocation.links = [
+            link.Link.make_link('self', url, 'allocations', allocation.uuid),
+            link.Link.make_link('bookmark', url, 'allocations',
+                                allocation.uuid, bookmark=True)
+        ]
+        return allocation
+
+    @classmethod
+    def convert_with_links(cls, rpc_allocation, fields=None, sanitize=True):
+        """Add links to the allocation."""
+        allocation = Allocation(**rpc_allocation.as_dict())
+
+        if rpc_allocation.node_id:
+            try:
+                allocation.node_uuid = objects.Node.get_by_id(
+                    pecan.request.context,
+                    rpc_allocation.node_id).uuid
+            except exception.NodeNotFound:
+                allocation.node_uuid = None
+        else:
+            allocation.node_uuid = None
+
+        if fields is not None:
+            api_utils.check_for_invalid_fields(fields, allocation.fields)
+
+        allocation = cls._convert_with_links(allocation,
+                                             pecan.request.host_url)
+
+        if not sanitize:
+            return allocation
+
+        allocation.sanitize(fields)
+
+        return allocation
+
+    def sanitize(self, fields=None):
+        """Removes sensitive and unrequested data.
+
+        Will only keep the fields specified in the ``fields`` parameter.
+
+        :param fields:
+            list of fields to preserve, or ``None`` to preserve them all
+        :type fields: list of str
+        """
+
+        if fields is not None:
+            self.unset_fields_except(fields)
+
+    @classmethod
+    def sample(cls):
+        """Return a sample of the allocation."""
+        sample = cls(uuid='a594544a-2daf-420c-8775-17a8c3e0852f',
+                     node_uuid='7ae81bb3-dec3-4289-8d6c-da80bd8001ae',
+                     name='node1-allocation-01',
+                     state=ir_states.ALLOCATING,
+                     last_error=None,
+                     resource_class='baremetal',
+                     traits=['CUSTOM_GPU'],
+                     candidate_nodes=[],
+                     extra={'foo': 'bar'},
+                     created_at=datetime.datetime(2000, 1, 1, 12, 0, 0),
+                     updated_at=datetime.datetime(2000, 1, 1, 12, 0, 0))
+        return cls._convert_with_links(sample, 'http://localhost:6385')
+
+
+class AllocationCollection(collection.Collection):
+    """API representation of a collection of allocations."""
+
+    allocations = [Allocation]
+    """A list containing allocation objects"""
+
+    def __init__(self, **kwargs):
+        self._type = 'allocations'
+
+    @staticmethod
+    def convert_with_links(rpc_allocations, limit, url=None, fields=None,
+                           **kwargs):
+        collection = AllocationCollection()
+        collection.allocations = [
+            Allocation.convert_with_links(p, fields=fields, sanitize=False)
+            for p in rpc_allocations
+        ]
+        collection.next = collection.get_next(limit, url=url, **kwargs)
+
+        for item in collection.allocations:
+            item.sanitize(fields=fields)
+
+        return collection
+
+    @classmethod
+    def sample(cls):
+        """Return a sample of the allocation."""
+        sample = cls()
+        sample.allocations = [Allocation.sample()]
+        return sample
+
+
+class AllocationsController(pecan.rest.RestController):
+    """REST controller for allocations."""
+
+    invalid_sort_key_list = ['extra', 'candidate_nodes', 'traits']
+
+    def _get_allocations_collection(self, node_ident=None, resource_class=None,
+                                    state=None, marker=None, limit=None,
+                                    sort_key='id', sort_dir='asc',
+                                    resource_url=None, fields=None):
+        """Return allocations collection.
+
+        :param node_ident: UUID or name of a node.
+        :param marker: Pagination marker for large data sets.
+        :param limit: Maximum number of resources to return in a single result.
+        :param sort_key: Column to sort results by. Default: id.
+        :param sort_dir: Direction to sort. "asc" or "desc". Default: asc.
+        :param resource_url: Optional, URL to the allocation resource.
+        :param fields: Optional, a list with a specified set of fields
+                       of the resource to be returned.
+        """
+        limit = api_utils.validate_limit(limit)
+        sort_dir = api_utils.validate_sort_dir(sort_dir)
+
+        if sort_key in self.invalid_sort_key_list:
+            raise exception.InvalidParameterValue(
+                _("The sort_key value %(key)s is an invalid field for "
+                  "sorting") % {'key': sort_key})
+
+        marker_obj = None
+        if marker:
+            marker_obj = objects.Allocation.get_by_uuid(pecan.request.context,
+                                                        marker)
+
+        if node_ident:
+            try:
+                node_uuid = api_utils.get_rpc_node(node_ident).uuid
+            except exception.NodeNotFound as exc:
+                exc.code = http_client.BAD_REQUEST
+                raise
+        else:
+            node_uuid = None
+
+        possible_filters = {
+            'node_uuid': node_uuid,
+            'resource_class': resource_class,
+            'state': state
+        }
+
+        filters = {}
+        for key, value in possible_filters.items():
+            if value is not None:
+                filters[key] = value
+
+        allocations = objects.Allocation.list(pecan.request.context,
+                                              limit=limit,
+                                              marker=marker_obj,
+                                              sort_key=sort_key,
+                                              sort_dir=sort_dir,
+                                              filters=filters)
+        return AllocationCollection.convert_with_links(allocations, limit,
+                                                       url=resource_url,
+                                                       fields=fields,
+                                                       sort_key=sort_key,
+                                                       sort_dir=sort_dir)
+
+    @METRICS.timer('AllocationsController.get_all')
+    @expose.expose(AllocationCollection, types.uuid_or_name, wtypes.text,
+                   wtypes.text, types.uuid, int, wtypes.text, wtypes.text,
+                   types.listtype)
+    def get_all(self, node=None, resource_class=None, state=None, marker=None,
+                limit=None, sort_key='id', sort_dir='asc', fields=None):
+        """Retrieve a list of allocations.
+
+        :param node: UUID or name of a node, to get only allocations for that
+                     node.
+        :param resource_class: Filter by requested resource class.
+        :param state: Filter by allocation state.
+        :param marker: pagination marker for large data sets.
+        :param limit: maximum number of resources to return in a single result.
+                      This value cannot be larger than the value of max_limit
+                      in the [api] section of the ironic configuration, or only
+                      max_limit resources will be returned.
+        :param sort_key: column to sort results by. Default: id.
+        :param sort_dir: direction to sort. "asc" or "desc". Default: asc.
+        :param fields: Optional, a list with a specified set of fields
+                       of the resource to be returned.
+        """
+        if not api_utils.allow_allocations():
+            raise exception.NotFound()
+
+        cdict = pecan.request.context.to_policy_values()
+        policy.authorize('baremetal:allocation:get', cdict, cdict)
+
+        return self._get_allocations_collection(node, resource_class, state,
+                                                marker, limit,
+                                                sort_key, sort_dir,
+                                                fields=fields)
+
+    @METRICS.timer('AllocationsController.get_one')
+    @expose.expose(Allocation, types.uuid_or_name, types.listtype)
+    def get_one(self, allocation_ident, fields=None):
+        """Retrieve information about the given allocation.
+
+        :param allocation_ident: UUID or logical name of an allocation.
+        :param fields: Optional, a list with a specified set of fields
+                       of the resource to be returned.
+        """
+        if not api_utils.allow_allocations():
+            raise exception.NotFound()
+
+        cdict = pecan.request.context.to_policy_values()
+        policy.authorize('baremetal:allocation:get', cdict, cdict)
+
+        rpc_allocation = api_utils.get_rpc_allocation_with_suffix(
+            allocation_ident)
+        return Allocation.convert_with_links(rpc_allocation, fields=fields)
+
+    @METRICS.timer('AllocationsController.post')
+    @expose.expose(Allocation, body=Allocation,
+                   status_code=http_client.CREATED)
+    def post(self, allocation):
+        """Create a new allocation.
+
+        :param allocation: an allocation within the request body.
+        """
+        if not api_utils.allow_allocations():
+            raise exception.NotFound()
+
+        context = pecan.request.context
+        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 "
+                    "'%(name)s'") % {'name': allocation.name}
+            raise exception.Invalid(msg)
+
+        if allocation.traits:
+            for trait in allocation.traits:
+                api_utils.validate_trait(trait)
+
+        if allocation.candidate_nodes:
+            # Convert nodes from names to UUIDs and check their validity
+            converted = []
+            for node in allocation.candidate_nodes:
+                try:
+                    node = api_utils.get_rpc_node(node)
+                except exception.NodeNotFound as exc:
+                    exc.code = http_client.BAD_REQUEST
+                    raise
+                else:
+                    converted.append(node.uuid)
+            allocation.candidate_nodes = converted
+
+        all_dict = allocation.as_dict()
+
+        # NOTE(yuriyz): UUID is mandatory for notifications payload
+        if not all_dict.get('uuid'):
+            all_dict['uuid'] = uuidutils.generate_uuid()
+
+        new_allocation = objects.Allocation(context, **all_dict)
+        topic = pecan.request.rpcapi.get_random_topic()
+
+        notify.emit_start_notification(context, new_allocation, 'create')
+        with notify.handle_error_notification(context, new_allocation,
+                                              'create'):
+            new_allocation = pecan.request.rpcapi.create_allocation(
+                context, new_allocation, topic)
+        notify.emit_end_notification(context, new_allocation, 'create')
+
+        # Set the HTTP Location Header
+        pecan.response.location = link.build_url('allocations',
+                                                 new_allocation.uuid)
+        return Allocation.convert_with_links(new_allocation)
+
+    @METRICS.timer('AllocationsController.delete')
+    @expose.expose(None, types.uuid_or_name,
+                   status_code=http_client.NO_CONTENT)
+    def delete(self, allocation_ident):
+        """Delete an allocation.
+
+        :param allocation_ident: UUID or logical name of an allocation.
+        """
+        if not api_utils.allow_allocations():
+            raise exception.NotFound()
+
+        context = pecan.request.context
+        cdict = context.to_policy_values()
+        policy.authorize('baremetal:allocation:delete', cdict, cdict)
+
+        rpc_allocation = api_utils.get_rpc_allocation_with_suffix(
+            allocation_ident)
+        if rpc_allocation.node_id:
+            node_uuid = objects.Node.get_by_id(pecan.request.context,
+                                               rpc_allocation.node_id).uuid
+        else:
+            node_uuid = None
+
+        notify.emit_start_notification(context, rpc_allocation, 'delete',
+                                       node_uuid=node_uuid)
+        with notify.handle_error_notification(context, rpc_allocation,
+                                              'delete', node_uuid=node_uuid):
+            topic = pecan.request.rpcapi.get_random_topic()
+            pecan.request.rpcapi.destroy_allocation(context, rpc_allocation,
+                                                    topic)
+        notify.emit_end_notification(context, rpc_allocation, 'delete',
+                                     node_uuid=node_uuid)
+
+
+class NodeAllocationController(pecan.rest.RestController):
+    """REST controller for allocations."""
+
+    invalid_sort_key_list = ['extra', 'candidate_nodes', 'traits']
+
+    def __init__(self, node_ident):
+        super(NodeAllocationController, self).__init__()
+        self.parent_node_ident = node_ident
+        self.inner = AllocationsController()
+
+    @METRICS.timer('NodeAllocationController.get_all')
+    @expose.expose(Allocation, types.listtype)
+    def get_all(self, fields=None):
+        if not api_utils.allow_allocations():
+            raise exception.NotFound()
+
+        cdict = pecan.request.context.to_policy_values()
+        policy.authorize('baremetal:allocation:get', cdict, cdict)
+
+        result = self.inner._get_allocations_collection(self.parent_node_ident,
+                                                        fields=fields)
+        try:
+            return result.allocations[0]
+        except IndexError:
+            raise exception.AllocationNotFound(
+                _("Allocation for node %s was not found") %
+                self.parent_node_ident)
+
+    @METRICS.timer('NodeAllocationController.delete')
+    @expose.expose(None, status_code=http_client.NO_CONTENT)
+    def delete(self):
+        if not api_utils.allow_allocations():
+            raise exception.NotFound()
+
+        context = pecan.request.context
+        cdict = context.to_policy_values()
+        policy.authorize('baremetal:allocation:delete', cdict, cdict)
+
+        rpc_node = api_utils.get_rpc_node_with_suffix(self.parent_node_ident)
+        allocations = objects.Allocation.list(
+            pecan.request.context,
+            filters={'node_uuid': rpc_node.uuid})
+
+        try:
+            rpc_allocation = allocations[0]
+        except IndexError:
+            raise exception.AllocationNotFound(
+                _("Allocation for node %s was not found") %
+                self.parent_node_ident)
+
+        notify.emit_start_notification(context, rpc_allocation, 'delete',
+                                       node_uuid=rpc_node.uuid)
+        with notify.handle_error_notification(context, rpc_allocation,
+                                              'delete',
+                                              node_uuid=rpc_node.uuid):
+            topic = pecan.request.rpcapi.get_random_topic()
+            pecan.request.rpcapi.destroy_allocation(context, rpc_allocation,
+                                                    topic)
+        notify.emit_end_notification(context, rpc_allocation, 'delete',
+                                     node_uuid=rpc_node.uuid)
diff --git a/ironic/api/controllers/v1/node.py b/ironic/api/controllers/v1/node.py
index 03beebb41a..1eb03fd7e0 100644
--- a/ironic/api/controllers/v1/node.py
+++ b/ironic/api/controllers/v1/node.py
@@ -28,6 +28,7 @@ from wsme import types as wtypes
 
 from ironic.api.controllers import base
 from ironic.api.controllers import link
+from ironic.api.controllers.v1 import allocation
 from ironic.api.controllers.v1 import bios
 from ironic.api.controllers.v1 import collection
 from ironic.api.controllers.v1 import notification_utils as notify
@@ -1083,6 +1084,9 @@ class Node(base.APIBase):
     description = wsme.wsattr(wtypes.text)
     """Field for node description"""
 
+    allocation_uuid = wsme.wsattr(types.uuid, readonly=True)
+    """The UUID of the allocation this node belongs"""
+
     # NOTE(deva): "conductor_affinity" shouldn't be presented on the
     #             API because it's an internal value. Don't add it here.
 
@@ -1174,8 +1178,21 @@ class Node(base.APIBase):
                           '%(node)s.', {'node': rpc_node.uuid})
                 node.conductor = None
 
+        if (api_utils.allow_allocations()
+                and (fields is None or 'allocation_uuid' in fields)):
+            node.allocation_uuid = None
+            if rpc_node.allocation_id:
+                try:
+                    allocation = objects.Allocation.get_by_id(
+                        pecan.request.context,
+                        rpc_node.allocation_id)
+                    node.allocation_uuid = allocation.uuid
+                except exception.AllocationNotFound:
+                    pass
+
         if fields is not None:
-            api_utils.check_for_invalid_fields(fields, node.as_dict())
+            api_utils.check_for_invalid_fields(
+                fields, set(node.as_dict()) | {'allocation_uuid'})
 
         show_states_links = (
             api_utils.allow_links_node_states_and_driver_properties())
@@ -1285,7 +1302,8 @@ class Node(base.APIBase):
                      storage_interface=None, traits=[], rescue_interface=None,
                      bios_interface=None, conductor_group="",
                      automated_clean=None, protected=False,
-                     protected_reason=None, owner=None)
+                     protected_reason=None, owner=None,
+                     allocation_uuid='982ddb5b-bce5-4d23-8fb8-7f710f648cd5')
         # NOTE(matty_dubs): The chassis_uuid getter() is based on the
         # _chassis_uuid variable:
         sample._chassis_uuid = 'edcad704-b2da-41d5-96d9-afd580ecfa12'
@@ -1311,7 +1329,7 @@ class NodePatchType(types.JsonPatchType):
                            '/inspection_started_at', '/clean_step',
                            '/deploy_step',
                            '/raid_config', '/target_raid_config',
-                           '/fault', '/conductor']
+                           '/fault', '/conductor', '/allocation_uuid']
 
 
 class NodeCollection(collection.Collection):
@@ -1563,6 +1581,7 @@ class NodesController(rest.RestController):
         'volume': volume.VolumeController,
         'traits': NodeTraitsController,
         'bios': bios.NodeBiosController,
+        'allocation': allocation.NodeAllocationController,
     }
 
     @pecan.expose()
@@ -1578,7 +1597,9 @@ class NodesController(rest.RestController):
             or (remainder[0] == 'vifs'
                 and not api_utils.allow_vifs_subcontroller())
             or (remainder[0] == 'bios' and
-                not api_utils.allow_bios_interface())):
+                not api_utils.allow_bios_interface())
+            or (remainder[0] == 'allocation'
+                and not api_utils.allow_allocations())):
             pecan.abort(http_client.NOT_FOUND)
         if remainder[0] == 'traits' and not api_utils.allow_traits():
             # NOTE(mgoddard): Returning here will ensure we exhibit the
@@ -2007,6 +2028,10 @@ class NodesController(rest.RestController):
                     "characters") % _NODE_DESCRIPTION_MAX_LENGTH
             raise exception.Invalid(msg)
 
+        if node.allocation_uuid is not wtypes.Unset:
+            msg = _("Allocation UUID cannot be specified, use allocations API")
+            raise exception.Invalid(msg)
+
         # NOTE(deva): get_topic_for checks if node.driver is in the hash ring
         #             and raises NoValidHost if it is not.
         #             We need to ensure that node has a UUID before it can
diff --git a/ironic/api/controllers/v1/notification_utils.py b/ironic/api/controllers/v1/notification_utils.py
index 57268104de..31464e02be 100644
--- a/ironic/api/controllers/v1/notification_utils.py
+++ b/ironic/api/controllers/v1/notification_utils.py
@@ -21,6 +21,7 @@ from wsme import types as wtypes
 
 from ironic.common import exception
 from ironic.common.i18n import _
+from ironic.objects import allocation as allocation_objects
 from ironic.objects import chassis as chassis_objects
 from ironic.objects import fields
 from ironic.objects import node as node_objects
@@ -35,6 +36,8 @@ CONF = cfg.CONF
 
 
 CRUD_NOTIFY_OBJ = {
+    'allocation': (allocation_objects.AllocationCRUDNotification,
+                   allocation_objects.AllocationCRUDPayload),
     'chassis': (chassis_objects.ChassisCRUDNotification,
                 chassis_objects.ChassisCRUDPayload),
     'node': (node_objects.NodeCRUDNotification,
diff --git a/ironic/api/controllers/v1/utils.py b/ironic/api/controllers/v1/utils.py
index 5140774671..770dec62dd 100644
--- a/ironic/api/controllers/v1/utils.py
+++ b/ironic/api/controllers/v1/utils.py
@@ -260,6 +260,45 @@ def get_rpc_portgroup_with_suffix(portgroup_ident):
                             exception.PortgroupNotFound)
 
 
+def get_rpc_allocation(allocation_ident):
+    """Get the RPC allocation from the allocation UUID or logical name.
+
+    :param allocation_ident: the UUID or logical name of an allocation.
+
+    :returns: The RPC allocation.
+    :raises: InvalidUuidOrName if the name or uuid provided is not valid.
+    :raises: AllocationNotFound if the allocation is not found.
+    """
+    # Check to see if the allocation_ident is a valid UUID.  If it is, treat it
+    # as a UUID.
+    if uuidutils.is_uuid_like(allocation_ident):
+        return objects.Allocation.get_by_uuid(pecan.request.context,
+                                              allocation_ident)
+
+    # We can refer to allocations by their name
+    if utils.is_valid_logical_name(allocation_ident):
+        return objects.Allocation.get_by_name(pecan.request.context,
+                                              allocation_ident)
+    raise exception.InvalidUuidOrName(name=allocation_ident)
+
+
+def get_rpc_allocation_with_suffix(allocation_ident):
+    """Get the RPC allocation from the allocation UUID or logical name.
+
+    If HAS_JSON_SUFFIX flag is set in the pecan environment, try also looking
+    for allocation_ident with '.json' suffix. Otherwise identical
+    to get_rpc_allocation.
+
+    :param allocation_ident: the UUID or logical name of an allocation.
+
+    :returns: The RPC allocation.
+    :raises: InvalidUuidOrName if the name or uuid provided is not valid.
+    :raises: AllocationNotFound if the allocation is not found.
+    """
+    return _get_with_suffix(get_rpc_allocation, allocation_ident,
+                            exception.AllocationNotFound)
+
+
 def is_valid_node_name(name):
     """Determine if the provided name is a valid node name.
 
@@ -381,6 +420,7 @@ VERSIONED_FIELDS = {
     'conductor': versions.MINOR_49_CONDUCTORS,
     'owner': versions.MINOR_50_NODE_OWNER,
     'description': versions.MINOR_51_NODE_DESCRIPTION,
+    'allocation_uuid': versions.MINOR_52_ALLOCATION,
 }
 
 for field in V31_FIELDS:
@@ -963,3 +1003,12 @@ def check_allow_filter_by_conductor(conductor):
             "should be %(base)s.%(opr)s") %
             {'base': versions.BASE_VERSION,
              'opr': versions.MINOR_49_CONDUCTORS})
+
+
+def allow_allocations():
+    """Check if accessing allocation endpoints is allowed.
+
+    Version 1.52 of the API exposed allocation endpoints and allocation_uuid
+    field for the node.
+    """
+    return pecan.request.version.minor >= versions.MINOR_52_ALLOCATION
diff --git a/ironic/api/controllers/v1/versions.py b/ironic/api/controllers/v1/versions.py
index e28db74635..d456738de8 100644
--- a/ironic/api/controllers/v1/versions.py
+++ b/ironic/api/controllers/v1/versions.py
@@ -89,6 +89,7 @@ BASE_VERSION = 1
 # v1.49: Exposes current conductor on the node object.
 # v1.50: Add owner to the node object.
 # v1.51: Add description to the node object.
+# v1.52: Add allocation API.
 
 MINOR_0_JUNO = 0
 MINOR_1_INITIAL_VERSION = 1
@@ -142,6 +143,7 @@ MINOR_48_NODE_PROTECTED = 48
 MINOR_49_CONDUCTORS = 49
 MINOR_50_NODE_OWNER = 50
 MINOR_51_NODE_DESCRIPTION = 51
+MINOR_52_ALLOCATION = 52
 
 # When adding another version, update:
 # - MINOR_MAX_VERSION
@@ -149,7 +151,7 @@ MINOR_51_NODE_DESCRIPTION = 51
 #   explanation of what changed in the new version
 # - common/release_mappings.py, RELEASE_MAPPING['master']['api']
 
-MINOR_MAX_VERSION = MINOR_51_NODE_DESCRIPTION
+MINOR_MAX_VERSION = MINOR_52_ALLOCATION
 
 # String representations of the minor and maximum versions
 _MIN_VERSION_STRING = '{}.{}'.format(BASE_VERSION, MINOR_1_INITIAL_VERSION)
diff --git a/ironic/common/policy.py b/ironic/common/policy.py
index a01708b1e9..0c55f4cf13 100644
--- a/ironic/common/policy.py
+++ b/ironic/common/policy.py
@@ -404,6 +404,27 @@ conductor_policies = [
          {'path': '/conductors/{hostname}', 'method': 'GET'}]),
 ]
 
+allocation_policies = [
+    policy.DocumentedRuleDefault(
+        'baremetal:allocation:get',
+        'rule:is_admin or rule:is_observer',
+        'Retrieve Allocation records',
+        [{'path': '/allocations', 'method': 'GET'},
+         {'path': '/allocations/{allocation_id}', 'method': 'GET'},
+         {'path': '/nodes/{node_ident}/allocation', 'method': 'GET'}]),
+    policy.DocumentedRuleDefault(
+        'baremetal:allocation:create',
+        'rule:is_admin',
+        'Create Allocation records',
+        [{'path': '/allocations', 'method': 'POST'}]),
+    policy.DocumentedRuleDefault(
+        'baremetal:allocation:delete',
+        'rule:is_admin',
+        'Delete Allocation records',
+        [{'path': '/allocations/{allocation_id}', 'method': 'DELETE'},
+         {'path': '/nodes/{node_ident}/allocation', 'method': 'DELETE'}]),
+]
+
 
 def list_policies():
     policies = itertools.chain(
@@ -416,7 +437,8 @@ def list_policies():
         vendor_passthru_policies,
         utility_policies,
         volume_policies,
-        conductor_policies
+        conductor_policies,
+        allocation_policies,
     )
     return policies
 
diff --git a/ironic/objects/allocation.py b/ironic/objects/allocation.py
index 655b9ad1ec..a279b886dd 100644
--- a/ironic/objects/allocation.py
+++ b/ironic/objects/allocation.py
@@ -295,6 +295,6 @@ class AllocationCRUDPayload(notification.NotificationPayloadBase):
         'updated_at': object_fields.DateTimeField(nullable=True),
     }
 
-    def __init__(self, allocation, node_uuid):
+    def __init__(self, allocation, node_uuid=None):
         super(AllocationCRUDPayload, self).__init__(node_uuid=node_uuid)
         self.populate_schema(allocation=allocation)
diff --git a/ironic/tests/unit/api/controllers/v1/test_allocation.py b/ironic/tests/unit/api/controllers/v1/test_allocation.py
new file mode 100644
index 0000000000..bcb3ccad66
--- /dev/null
+++ b/ironic/tests/unit/api/controllers/v1/test_allocation.py
@@ -0,0 +1,697 @@
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+"""
+Tests for the API /allocations/ methods.
+"""
+
+import datetime
+
+import fixtures
+import mock
+from oslo_config import cfg
+from oslo_utils import timeutils
+from oslo_utils import uuidutils
+import six
+from six.moves import http_client
+from six.moves.urllib import parse as urlparse
+from wsme import types as wtypes
+
+from ironic.api.controllers import base as api_base
+from ironic.api.controllers import v1 as api_v1
+from ironic.api.controllers.v1 import allocation as api_allocation
+from ironic.api.controllers.v1 import notification_utils
+from ironic.common import exception
+from ironic.conductor import rpcapi
+from ironic import objects
+from ironic.objects import fields as obj_fields
+from ironic.tests import base
+from ironic.tests.unit.api import base as test_api_base
+from ironic.tests.unit.api import utils as apiutils
+from ironic.tests.unit.objects import utils as obj_utils
+
+
+class TestAllocationObject(base.TestCase):
+
+    def test_allocation_init(self):
+        allocation_dict = apiutils.allocation_post_data(node_id=None)
+        del allocation_dict['extra']
+        allocation = api_allocation.Allocation(**allocation_dict)
+        self.assertEqual(wtypes.Unset, allocation.extra)
+
+
+class TestListAllocations(test_api_base.BaseApiTest):
+    headers = {api_base.Version.string: str(api_v1.max_version())}
+
+    def setUp(self):
+        super(TestListAllocations, self).setUp()
+        self.node = obj_utils.create_test_node(self.context, name='node-1')
+
+    def test_empty(self):
+        data = self.get_json('/allocations', headers=self.headers)
+        self.assertEqual([], data['allocations'])
+
+    def test_one(self):
+        allocation = obj_utils.create_test_allocation(self.context,
+                                                      node_id=self.node.id)
+        data = self.get_json('/allocations', headers=self.headers)
+        self.assertEqual(allocation.uuid, data['allocations'][0]["uuid"])
+        self.assertEqual(allocation.name, data['allocations'][0]['name'])
+        self.assertEqual({}, data['allocations'][0]["extra"])
+        self.assertEqual(self.node.uuid, data['allocations'][0]["node_uuid"])
+        # never expose the node_id
+        self.assertNotIn('node_id', data['allocations'][0])
+
+    def test_get_one(self):
+        allocation = obj_utils.create_test_allocation(self.context,
+                                                      node_id=self.node.id)
+        data = self.get_json('/allocations/%s' % allocation.uuid,
+                             headers=self.headers)
+        self.assertEqual(allocation.uuid, data['uuid'])
+        self.assertEqual({}, data["extra"])
+        self.assertEqual(self.node.uuid, data["node_uuid"])
+        # never expose the node_id
+        self.assertNotIn('node_id', data)
+
+    def test_get_one_with_json(self):
+        allocation = obj_utils.create_test_allocation(self.context,
+                                                      node_id=self.node.id)
+        data = self.get_json('/allocations/%s.json' % allocation.uuid,
+                             headers=self.headers)
+        self.assertEqual(allocation.uuid, data['uuid'])
+
+    def test_get_one_with_json_in_name(self):
+        allocation = obj_utils.create_test_allocation(self.context,
+                                                      name='pg.json',
+                                                      node_id=self.node.id)
+        data = self.get_json('/allocations/%s' % allocation.uuid,
+                             headers=self.headers)
+        self.assertEqual(allocation.uuid, data['uuid'])
+
+    def test_get_one_with_suffix(self):
+        allocation = obj_utils.create_test_allocation(self.context,
+                                                      name='pg.1',
+                                                      node_id=self.node.id)
+        data = self.get_json('/allocations/%s' % allocation.uuid,
+                             headers=self.headers)
+        self.assertEqual(allocation.uuid, data['uuid'])
+
+    def test_get_one_custom_fields(self):
+        allocation = obj_utils.create_test_allocation(self.context,
+                                                      node_id=self.node.id)
+        fields = 'resource_class,extra'
+        data = self.get_json(
+            '/allocations/%s?fields=%s' % (allocation.uuid, fields),
+            headers=self.headers)
+        # We always append "links"
+        self.assertItemsEqual(['resource_class', 'extra', 'links'], data)
+
+    def test_get_collection_custom_fields(self):
+        fields = 'uuid,extra'
+        for i in range(3):
+            obj_utils.create_test_allocation(
+                self.context,
+                node_id=self.node.id,
+                uuid=uuidutils.generate_uuid(),
+                name='allocation%s' % i)
+
+        data = self.get_json(
+            '/allocations?fields=%s' % fields,
+            headers=self.headers)
+
+        self.assertEqual(3, len(data['allocations']))
+        for allocation in data['allocations']:
+            # We always append "links"
+            self.assertItemsEqual(['uuid', 'extra', 'links'], allocation)
+
+    def test_get_custom_fields_invalid_fields(self):
+        allocation = obj_utils.create_test_allocation(self.context,
+                                                      node_id=self.node.id)
+        fields = 'uuid,spongebob'
+        response = self.get_json(
+            '/allocations/%s?fields=%s' % (allocation.uuid, fields),
+            headers=self.headers, expect_errors=True)
+        self.assertEqual(http_client.BAD_REQUEST, response.status_int)
+        self.assertEqual('application/json', response.content_type)
+        self.assertIn('spongebob', response.json['error_message'])
+
+    def test_get_one_invalid_api_version(self):
+        allocation = obj_utils.create_test_allocation(self.context,
+                                                      node_id=self.node.id)
+        response = self.get_json(
+            '/allocations/%s' % (allocation.uuid),
+            headers={api_base.Version.string: str(api_v1.min_version())},
+            expect_errors=True)
+        self.assertEqual(http_client.NOT_FOUND, response.status_int)
+
+    def test_many(self):
+        allocations = []
+        for id_ in range(5):
+            allocation = obj_utils.create_test_allocation(
+                self.context, node_id=self.node.id,
+                uuid=uuidutils.generate_uuid(),
+                name='allocation%s' % id_)
+            allocations.append(allocation.uuid)
+        data = self.get_json('/allocations', headers=self.headers)
+        self.assertEqual(len(allocations), len(data['allocations']))
+
+        uuids = [n['uuid'] for n in data['allocations']]
+        six.assertCountEqual(self, allocations, uuids)
+
+    def test_links(self):
+        uuid = uuidutils.generate_uuid()
+        obj_utils.create_test_allocation(self.context,
+                                         uuid=uuid,
+                                         node_id=self.node.id)
+        data = self.get_json('/allocations/%s' % uuid, headers=self.headers)
+        self.assertIn('links', data)
+        self.assertEqual(2, len(data['links']))
+        self.assertIn(uuid, data['links'][0]['href'])
+        for l in data['links']:
+            bookmark = l['rel'] == 'bookmark'
+            self.assertTrue(self.validate_link(l['href'], bookmark=bookmark,
+                                               headers=self.headers))
+
+    def test_collection_links(self):
+        allocations = []
+        for id_ in range(5):
+            allocation = obj_utils.create_test_allocation(
+                self.context,
+                uuid=uuidutils.generate_uuid(),
+                name='allocation%s' % id_)
+            allocations.append(allocation.uuid)
+        data = self.get_json('/allocations/?limit=3', headers=self.headers)
+        self.assertEqual(3, len(data['allocations']))
+
+        next_marker = data['allocations'][-1]['uuid']
+        self.assertIn(next_marker, data['next'])
+
+    def test_collection_links_default_limit(self):
+        cfg.CONF.set_override('max_limit', 3, 'api')
+        allocations = []
+        for id_ in range(5):
+            allocation = obj_utils.create_test_allocation(
+                self.context,
+                uuid=uuidutils.generate_uuid(),
+                name='allocation%s' % id_)
+            allocations.append(allocation.uuid)
+        data = self.get_json('/allocations', headers=self.headers)
+        self.assertEqual(3, len(data['allocations']))
+
+        next_marker = data['allocations'][-1]['uuid']
+        self.assertIn(next_marker, data['next'])
+
+    def test_get_collection_pagination_no_uuid(self):
+        fields = 'node_uuid'
+        limit = 2
+        allocations = []
+        for id_ in range(3):
+            allocation = obj_utils.create_test_allocation(
+                self.context,
+                node_id=self.node.id,
+                uuid=uuidutils.generate_uuid(),
+                name='allocation%s' % id_)
+            allocations.append(allocation)
+
+        data = self.get_json(
+            '/allocations?fields=%s&limit=%s' % (fields, limit),
+            headers=self.headers)
+
+        self.assertEqual(limit, len(data['allocations']))
+        self.assertIn('marker=%s' % allocations[limit - 1].uuid, data['next'])
+
+    def test_allocation_get_all_invalid_api_version(self):
+        obj_utils.create_test_allocation(
+            self.context, node_id=self.node.id, uuid=uuidutils.generate_uuid(),
+            name='allocation_1')
+        response = self.get_json('/allocations',
+                                 headers={api_base.Version.string: '1.14'},
+                                 expect_errors=True)
+        self.assertEqual(http_client.NOT_FOUND, response.status_int)
+
+    def test_sort_key(self):
+        allocations = []
+        for id_ in range(3):
+            allocation = obj_utils.create_test_allocation(
+                self.context,
+                node_id=self.node.id,
+                uuid=uuidutils.generate_uuid(),
+                name='allocation%s' % id_)
+            allocations.append(allocation.uuid)
+        data = self.get_json('/allocations?sort_key=uuid',
+                             headers=self.headers)
+        uuids = [n['uuid'] for n in data['allocations']]
+        self.assertEqual(sorted(allocations), uuids)
+
+    def test_sort_key_invalid(self):
+        invalid_keys_list = ['foo', 'extra', 'internal_info', 'properties']
+        for invalid_key in invalid_keys_list:
+            response = self.get_json('/allocations?sort_key=%s' % invalid_key,
+                                     expect_errors=True, headers=self.headers)
+            self.assertEqual(http_client.BAD_REQUEST, response.status_int)
+            self.assertEqual('application/json', response.content_type)
+            self.assertIn(invalid_key, response.json['error_message'])
+
+    def test_sort_key_allowed(self):
+        allocation_uuids = []
+        for id_ in range(3, 0, -1):
+            allocation = obj_utils.create_test_allocation(
+                self.context,
+                uuid=uuidutils.generate_uuid(),
+                name='allocation%s' % id_)
+            allocation_uuids.append(allocation.uuid)
+        allocation_uuids.reverse()
+        data = self.get_json('/allocations?sort_key=name',
+                             headers=self.headers)
+        data_uuids = [p['uuid'] for p in data['allocations']]
+        self.assertEqual(allocation_uuids, data_uuids)
+
+    def test_get_all_by_state(self):
+        for i in range(5):
+            if i < 3:
+                state = 'allocating'
+            else:
+                state = 'active'
+            obj_utils.create_test_allocation(
+                self.context,
+                state=state,
+                uuid=uuidutils.generate_uuid(),
+                name='allocation%s' % i)
+        data = self.get_json("/allocations?state=allocating",
+                             headers=self.headers)
+        self.assertEqual(3, len(data['allocations']))
+
+    def test_get_all_by_node_name(self):
+        for i in range(5):
+            if i < 3:
+                node_id = self.node.id
+            else:
+                node_id = 100000 + i
+            obj_utils.create_test_allocation(
+                self.context,
+                node_id=node_id,
+                uuid=uuidutils.generate_uuid(),
+                name='allocation%s' % i)
+        data = self.get_json("/allocations?node=%s" % self.node.name,
+                             headers=self.headers)
+        self.assertEqual(3, len(data['allocations']))
+
+    def test_get_all_by_node_uuid(self):
+        obj_utils.create_test_allocation(self.context, node_id=self.node.id)
+        data = self.get_json('/allocations?node=%s' % (self.node.uuid),
+                             headers=self.headers)
+        self.assertEqual(1, len(data['allocations']))
+
+    def test_get_all_by_non_existing_node(self):
+        obj_utils.create_test_allocation(self.context, node_id=self.node.id)
+        response = self.get_json('/allocations?node=banana',
+                                 headers=self.headers, expect_errors=True)
+        self.assertEqual('application/json', response.content_type)
+        self.assertEqual(http_client.BAD_REQUEST, response.status_int)
+
+    def test_get_by_node_resource(self):
+        allocation = obj_utils.create_test_allocation(self.context,
+                                                      node_id=self.node.id)
+        data = self.get_json('/nodes/%s/allocation' % self.node.uuid,
+                             headers=self.headers)
+        self.assertEqual(allocation.uuid, data['uuid'])
+        self.assertEqual({}, data["extra"])
+        self.assertEqual(self.node.uuid, data["node_uuid"])
+
+    def test_get_by_node_resource_with_fields(self):
+        obj_utils.create_test_allocation(self.context, node_id=self.node.id)
+        data = self.get_json('/nodes/%s/allocation?fields=name,extra' %
+                             self.node.uuid,
+                             headers=self.headers)
+        self.assertNotIn('uuid', data)
+        self.assertIn('name', data)
+        self.assertEqual({}, data["extra"])
+
+    def test_get_by_node_resource_and_id(self):
+        allocation = obj_utils.create_test_allocation(self.context,
+                                                      node_id=self.node.id)
+        response = self.get_json('/nodes/%s/allocation/%s' % (self.node.uuid,
+                                                              allocation.uuid),
+                                 headers=self.headers, expect_errors=True)
+        self.assertEqual('application/json', response.content_type)
+        self.assertEqual(http_client.METHOD_NOT_ALLOWED, response.status_int)
+
+    def test_by_node_resource_not_existed(self):
+        node = obj_utils.create_test_node(self.context,
+                                          uuid=uuidutils.generate_uuid())
+        res = self.get_json('/node/%s/allocation' % node.uuid,
+                            expect_errors=True, headers=self.headers)
+        self.assertEqual(http_client.NOT_FOUND, res.status_code)
+
+    def test_by_node_invalid_node(self):
+        res = self.get_json('/node/%s/allocation' % uuidutils.generate_uuid(),
+                            expect_errors=True, headers=self.headers)
+        self.assertEqual(http_client.NOT_FOUND, res.status_code)
+
+
+class TestPatch(test_api_base.BaseApiTest):
+    headers = {api_base.Version.string: str(api_v1.max_version())}
+
+    def setUp(self):
+        super(TestPatch, self).setUp()
+        self.allocation = obj_utils.create_test_allocation(self.context)
+
+    def test_update_not_allowed(self):
+        response = self.patch_json('/allocations/%s' % self.allocation.uuid,
+                                   [{'path': '/extra/foo',
+                                     'value': 'bar',
+                                     'op': 'add'}],
+                                   expect_errors=True,
+                                   headers=self.headers)
+        self.assertEqual('application/json', response.content_type)
+        self.assertEqual(http_client.METHOD_NOT_ALLOWED, response.status_int)
+
+
+def _create_locally(_api, _ctx, allocation, _topic):
+    allocation.create()
+    return allocation
+
+
+@mock.patch.object(rpcapi.ConductorAPI, 'create_allocation', _create_locally)
+class TestPost(test_api_base.BaseApiTest):
+    headers = {api_base.Version.string: str(api_v1.max_version())}
+
+    def setUp(self):
+        super(TestPost, self).setUp()
+        self.mock_get_topic = self.useFixture(
+            fixtures.MockPatchObject(rpcapi.ConductorAPI, 'get_random_topic')
+        ).mock
+        self.mock_get_topic.return_value = 'some-topic'
+
+    @mock.patch.object(notification_utils, '_emit_api_notification')
+    @mock.patch.object(timeutils, 'utcnow', autospec=True)
+    def test_create_allocation(self, mock_utcnow, mock_notify):
+        adict = apiutils.allocation_post_data()
+        test_time = datetime.datetime(2000, 1, 1, 0, 0)
+        mock_utcnow.return_value = test_time
+        response = self.post_json('/allocations', adict,
+                                  headers=self.headers)
+        self.assertEqual(http_client.CREATED, response.status_int)
+        self.assertEqual(adict['uuid'], response.json['uuid'])
+        self.assertEqual('allocating', response.json['state'])
+        self.assertIsNone(response.json['node_uuid'])
+        result = self.get_json('/allocations/%s' % adict['uuid'],
+                               headers=self.headers)
+        self.assertEqual(adict['uuid'], result['uuid'])
+        self.assertFalse(result['updated_at'])
+        self.assertIsNone(result['node_uuid'])
+        return_created_at = timeutils.parse_isotime(
+            result['created_at']).replace(tzinfo=None)
+        self.assertEqual(test_time, return_created_at)
+        # Check location header
+        self.assertIsNotNone(response.location)
+        expected_location = '/v1/allocations/%s' % adict['uuid']
+        self.assertEqual(urlparse.urlparse(response.location).path,
+                         expected_location)
+        mock_notify.assert_has_calls([
+            mock.call(mock.ANY, mock.ANY, 'create',
+                      obj_fields.NotificationLevel.INFO,
+                      obj_fields.NotificationStatus.START),
+            mock.call(mock.ANY, mock.ANY, 'create',
+                      obj_fields.NotificationLevel.INFO,
+                      obj_fields.NotificationStatus.END),
+        ])
+
+    def test_create_allocation_invalid_api_version(self):
+        adict = apiutils.allocation_post_data()
+        response = self.post_json(
+            '/allocations', adict, headers={api_base.Version.string: '1.50'},
+            expect_errors=True)
+        self.assertEqual(http_client.NOT_FOUND, response.status_int)
+
+    def test_create_allocation_doesnt_contain_id(self):
+        with mock.patch.object(self.dbapi, 'create_allocation',
+                               wraps=self.dbapi.create_allocation) as cp_mock:
+            adict = apiutils.allocation_post_data(extra={'foo': 123})
+            self.post_json('/allocations', adict, headers=self.headers)
+            result = self.get_json('/allocations/%s' % adict['uuid'],
+                                   headers=self.headers)
+            self.assertEqual(adict['extra'], result['extra'])
+            cp_mock.assert_called_once_with(mock.ANY)
+            # Check that 'id' is not in first arg of positional args
+            self.assertNotIn('id', cp_mock.call_args[0][0])
+
+    @mock.patch.object(notification_utils.LOG, 'exception', autospec=True)
+    @mock.patch.object(notification_utils.LOG, 'warning', autospec=True)
+    def test_create_allocation_generate_uuid(self, mock_warn, mock_except):
+        adict = apiutils.allocation_post_data()
+        del adict['uuid']
+        response = self.post_json('/allocations', adict, headers=self.headers)
+        result = self.get_json('/allocations/%s' % response.json['uuid'],
+                               headers=self.headers)
+        self.assertTrue(uuidutils.is_uuid_like(result['uuid']))
+        self.assertFalse(mock_warn.called)
+        self.assertFalse(mock_except.called)
+
+    @mock.patch.object(notification_utils, '_emit_api_notification')
+    @mock.patch.object(objects.Allocation, 'create')
+    def test_create_allocation_error(self, mock_create, mock_notify):
+        mock_create.side_effect = Exception()
+        adict = apiutils.allocation_post_data()
+        self.post_json('/allocations', adict, headers=self.headers,
+                       expect_errors=True)
+        mock_notify.assert_has_calls([
+            mock.call(mock.ANY, mock.ANY, 'create',
+                      obj_fields.NotificationLevel.INFO,
+                      obj_fields.NotificationStatus.START),
+            mock.call(mock.ANY, mock.ANY, 'create',
+                      obj_fields.NotificationLevel.ERROR,
+                      obj_fields.NotificationStatus.ERROR),
+        ])
+
+    def test_create_allocation_with_candidate_nodes(self):
+        node1 = obj_utils.create_test_node(self.context,
+                                           name='node-1')
+        node2 = obj_utils.create_test_node(self.context,
+                                           uuid=uuidutils.generate_uuid())
+        adict = apiutils.allocation_post_data(
+            candidate_nodes=[node1.name, node2.uuid])
+        response = self.post_json('/allocations', adict,
+                                  headers=self.headers)
+        self.assertEqual(http_client.CREATED, response.status_int)
+        result = self.get_json('/allocations/%s' % adict['uuid'],
+                               headers=self.headers)
+        self.assertEqual(adict['uuid'], result['uuid'])
+        self.assertEqual([node1.uuid, node2.uuid], result['candidate_nodes'])
+
+    def test_create_allocation_valid_extra(self):
+        adict = apiutils.allocation_post_data(
+            extra={'str': 'foo', 'int': 123, 'float': 0.1, 'bool': True,
+                   'list': [1, 2], 'none': None, 'dict': {'cat': 'meow'}})
+        self.post_json('/allocations', adict, headers=self.headers)
+        result = self.get_json('/allocations/%s' % adict['uuid'],
+                               headers=self.headers)
+        self.assertEqual(adict['extra'], result['extra'])
+
+    def test_create_allocation_with_no_extra(self):
+        adict = apiutils.allocation_post_data()
+        del adict['extra']
+        response = self.post_json('/allocations', adict, headers=self.headers)
+        self.assertEqual('application/json', response.content_type)
+        self.assertEqual(http_client.CREATED, response.status_int)
+
+    def test_create_allocation_no_mandatory_field_resource_class(self):
+        adict = apiutils.allocation_post_data()
+        del adict['resource_class']
+        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_create_allocation_resource_class_too_long(self):
+        adict = apiutils.allocation_post_data()
+        adict['resource_class'] = 'f' * 81
+        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_create_allocation_with_traits(self):
+        adict = apiutils.allocation_post_data()
+        adict['traits'] = ['CUSTOM_GPU', 'CUSTOM_FOO_BAR']
+        response = self.post_json('/allocations', adict, headers=self.headers)
+        self.assertEqual('application/json', response.content_type)
+        self.assertEqual(http_client.CREATED, response.status_int)
+        self.assertEqual(['CUSTOM_GPU', 'CUSTOM_FOO_BAR'],
+                         response.json['traits'])
+        result = self.get_json('/allocations/%s' % adict['uuid'],
+                               headers=self.headers)
+        self.assertEqual(adict['uuid'], result['uuid'])
+        self.assertEqual(['CUSTOM_GPU', 'CUSTOM_FOO_BAR'],
+                         result['traits'])
+
+    def test_create_allocation_invalid_trait(self):
+        adict = apiutils.allocation_post_data()
+        adict['traits'] = ['CUSTOM_GPU', 'FOO_BAR']
+        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_create_allocation_invalid_candidate_node_format(self):
+        adict = apiutils.allocation_post_data(
+            candidate_nodes=['invalid-format'])
+        response = self.post_json('/allocations', adict, expect_errors=True,
+                                  headers=self.headers)
+        self.assertEqual('application/json', response.content_type)
+        self.assertEqual(http_client.BAD_REQUEST, response.status_int)
+        self.assertTrue(response.json['error_message'])
+
+    def test_create_allocation_candidate_node_not_found(self):
+        adict = apiutils.allocation_post_data(
+            candidate_nodes=['1a1a1a1a-2b2b-3c3c-4d4d-5e5e5e5e5e5e'])
+        response = self.post_json('/allocations', adict, expect_errors=True,
+                                  headers=self.headers)
+        self.assertEqual('application/json', response.content_type)
+        self.assertEqual(http_client.BAD_REQUEST, response.status_int)
+        self.assertTrue(response.json['error_message'])
+
+    def test_create_allocation_name_ok(self):
+        name = 'foo'
+        adict = apiutils.allocation_post_data(name=name)
+        self.post_json('/allocations', adict, headers=self.headers)
+        result = self.get_json('/allocations/%s' % adict['uuid'],
+                               headers=self.headers)
+        self.assertEqual(name, result['name'])
+
+    def test_create_allocation_name_invalid(self):
+        name = 'aa:bb_cc'
+        adict = apiutils.allocation_post_data(name=name)
+        response = self.post_json('/allocations', adict, headers=self.headers,
+                                  expect_errors=True)
+        self.assertEqual(http_client.BAD_REQUEST, response.status_int)
+
+    def test_create_by_node_not_allowed(self):
+        node = obj_utils.create_test_node(self.context)
+        adict = apiutils.allocation_post_data()
+        response = self.post_json('/nodes/%s/allocation' % node.uuid,
+                                  adict, headers=self.headers,
+                                  expect_errors=True)
+        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):
+        adict = apiutils.allocation_post_data()
+        adict['node_uuid'] = 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'])
+
+
+@mock.patch.object(rpcapi.ConductorAPI, 'destroy_allocation')
+class TestDelete(test_api_base.BaseApiTest):
+    headers = {api_base.Version.string: str(api_v1.max_version())}
+
+    def setUp(self):
+        super(TestDelete, self).setUp()
+        self.node = obj_utils.create_test_node(self.context)
+        self.allocation = obj_utils.create_test_allocation(
+            self.context, node_id=self.node.id, name='alloc1')
+
+        self.mock_get_topic = self.useFixture(
+            fixtures.MockPatchObject(rpcapi.ConductorAPI, 'get_random_topic')
+        ).mock
+
+    @mock.patch.object(notification_utils, '_emit_api_notification')
+    def test_delete_allocation_by_id(self, mock_notify, mock_destroy):
+        self.delete('/allocations/%s' % self.allocation.uuid,
+                    headers=self.headers)
+        self.assertTrue(mock_destroy.called)
+        mock_notify.assert_has_calls([
+            mock.call(mock.ANY, mock.ANY, 'delete',
+                      obj_fields.NotificationLevel.INFO,
+                      obj_fields.NotificationStatus.START,
+                      node_uuid=self.node.uuid),
+            mock.call(mock.ANY, mock.ANY, 'delete',
+                      obj_fields.NotificationLevel.INFO,
+                      obj_fields.NotificationStatus.END,
+                      node_uuid=self.node.uuid),
+        ])
+
+    @mock.patch.object(notification_utils, '_emit_api_notification')
+    def test_delete_allocation_node_locked(self, mock_notify, mock_destroy):
+        self.node.reserve(self.context, 'fake', self.node.uuid)
+        mock_destroy.side_effect = exception.NodeLocked(node='fake-node',
+                                                        host='fake-host')
+        ret = self.delete('/allocations/%s' % self.allocation.uuid,
+                          expect_errors=True, headers=self.headers)
+        self.assertEqual(http_client.CONFLICT, ret.status_code)
+        self.assertTrue(ret.json['error_message'])
+        self.assertTrue(mock_destroy.called)
+        mock_notify.assert_has_calls([
+            mock.call(mock.ANY, mock.ANY, 'delete',
+                      obj_fields.NotificationLevel.INFO,
+                      obj_fields.NotificationStatus.START,
+                      node_uuid=self.node.uuid),
+            mock.call(mock.ANY, mock.ANY, 'delete',
+                      obj_fields.NotificationLevel.ERROR,
+                      obj_fields.NotificationStatus.ERROR,
+                      node_uuid=self.node.uuid),
+        ])
+
+    def test_delete_allocation_invalid_api_version(self, mock_destroy):
+        response = self.delete('/allocations/%s' % self.allocation.uuid,
+                               expect_errors=True,
+                               headers={api_base.Version.string: '1.14'})
+        self.assertEqual(http_client.NOT_FOUND, response.status_int)
+
+    def test_delete_allocation_by_name(self, mock_destroy):
+        self.delete('/allocations/%s' % self.allocation.name,
+                    headers=self.headers)
+        self.assertTrue(mock_destroy.called)
+
+    def test_delete_allocation_by_name_with_json(self, mock_destroy):
+        self.delete('/allocations/%s.json' % self.allocation.name,
+                    headers=self.headers)
+        self.assertTrue(mock_destroy.called)
+
+    def test_delete_allocation_by_name_not_existed(self, mock_destroy):
+        res = self.delete('/allocations/%s' % 'blah', expect_errors=True,
+                          headers=self.headers)
+        self.assertEqual(http_client.NOT_FOUND, res.status_code)
+
+    @mock.patch.object(notification_utils, '_emit_api_notification')
+    def test_delete_allocation_by_node(self, mock_notify, mock_destroy):
+        self.delete('/nodes/%s/allocation' % self.node.uuid,
+                    headers=self.headers)
+        self.assertTrue(mock_destroy.called)
+        mock_notify.assert_has_calls([
+            mock.call(mock.ANY, mock.ANY, 'delete',
+                      obj_fields.NotificationLevel.INFO,
+                      obj_fields.NotificationStatus.START,
+                      node_uuid=self.node.uuid),
+            mock.call(mock.ANY, mock.ANY, 'delete',
+                      obj_fields.NotificationLevel.INFO,
+                      obj_fields.NotificationStatus.END,
+                      node_uuid=self.node.uuid),
+        ])
+
+    def test_delete_allocation_by_node_not_existed(self, mock_destroy):
+        node = obj_utils.create_test_node(self.context,
+                                          uuid=uuidutils.generate_uuid())
+        res = self.delete('/nodes/%s/allocation' % node.uuid,
+                          expect_errors=True, headers=self.headers)
+        self.assertEqual(http_client.NOT_FOUND, res.status_code)
+
+    def test_delete_allocation_invalid_node(self, mock_destroy):
+        res = self.delete('/nodes/%s/allocation' % uuidutils.generate_uuid(),
+                          expect_errors=True, headers=self.headers)
+        self.assertEqual(http_client.NOT_FOUND, res.status_code)
diff --git a/ironic/tests/unit/api/controllers/v1/test_node.py b/ironic/tests/unit/api/controllers/v1/test_node.py
index 14ee668b24..15e01ddd50 100644
--- a/ironic/tests/unit/api/controllers/v1/test_node.py
+++ b/ironic/tests/unit/api/controllers/v1/test_node.py
@@ -175,6 +175,8 @@ class TestListNodes(test_api_base.BaseApiTest):
         self.assertIn('protected', data)
         self.assertIn('protected_reason', data)
         self.assertIn('owner', data)
+        self.assertNotIn('allocation_id', data)
+        self.assertIn('allocation_uuid', data)
 
     def test_get_one_with_json(self):
         # Test backward compatibility with guess_content_type_from_ext
@@ -557,6 +559,15 @@ class TestListNodes(test_api_base.BaseApiTest):
                                  headers={api_base.Version.string: '1.51'})
         self.assertIn('description', response)
 
+    def test_get_with_allocation(self):
+        allocation = obj_utils.create_test_allocation(self.context)
+        node = obj_utils.create_test_node(self.context,
+                                          allocation_id=allocation.id)
+        fields = 'allocation_uuid'
+        response = self.get_json('/nodes/%s?fields=%s' % (node.uuid, fields),
+                                 headers={api_base.Version.string: '1.52'})
+        self.assertEqual(allocation.uuid, response['allocation_uuid'])
+
     def test_detail(self):
         node = obj_utils.create_test_node(self.context,
                                           chassis_id=self.chassis.id)
@@ -593,6 +604,8 @@ class TestListNodes(test_api_base.BaseApiTest):
         self.assertIn('owner', data['nodes'][0])
         # never expose the chassis_id
         self.assertNotIn('chassis_id', data['nodes'][0])
+        self.assertNotIn('allocation_id', data['nodes'][0])
+        self.assertIn('allocation_uuid', data['nodes'][0])
 
     def test_detail_using_query(self):
         node = obj_utils.create_test_node(self.context,
@@ -2814,6 +2827,19 @@ class TestPatch(test_api_base.BaseApiTest):
         self.assertEqual(http_client.BAD_REQUEST, response.status_code)
         self.assertTrue(response.json['error_message'])
 
+    def test_patch_allocation_uuid_forbidden(self):
+        node = obj_utils.create_test_node(self.context,
+                                          uuid=uuidutils.generate_uuid())
+        response = self.patch_json('/nodes/%s' % node.uuid,
+                                   [{'path': '/allocation_uuid',
+                                     'op': 'replace',
+                                     'value': uuidutils.generate_uuid()}],
+                                   headers={api_base.Version.string: "1.52"},
+                                   expect_errors=True)
+        self.assertEqual('application/json', response.content_type)
+        self.assertEqual(http_client.BAD_REQUEST, response.status_code)
+        self.assertTrue(response.json['error_message'])
+
     def test_update_conductor_group(self):
         node = obj_utils.create_test_node(self.context,
                                           uuid=uuidutils.generate_uuid())
@@ -2996,6 +3022,20 @@ class TestPatch(test_api_base.BaseApiTest):
         self.assertEqual('application/json', response.content_type)
         self.assertEqual(http_client.BAD_REQUEST, response.status_code)
 
+    def test_patch_allocation_forbidden(self):
+        node = obj_utils.create_test_node(self.context,
+                                          uuid=uuidutils.generate_uuid())
+        response = self.patch_json('/nodes/%s' % node.uuid,
+                                   [{'path': '/allocation_uuid',
+                                     'op': 'replace',
+                                     'value': uuidutils.generate_uuid()}],
+                                   headers={api_base.Version.string:
+                                            str(api_v1.max_version())},
+                                   expect_errors=True)
+        self.assertEqual('application/json', response.content_type)
+        self.assertEqual(http_client.BAD_REQUEST, response.status_code)
+        self.assertTrue(response.json['error_message'])
+
 
 def _create_node_locally(node):
     driver_factory.check_and_update_node_interfaces(node)
diff --git a/ironic/tests/unit/api/utils.py b/ironic/tests/unit/api/utils.py
index f6f95aaacb..9f73a4b39e 100644
--- a/ironic/tests/unit/api/utils.py
+++ b/ironic/tests/unit/api/utils.py
@@ -187,3 +187,14 @@ def post_get_test_portgroup(**kw):
     node = db_utils.get_test_node()
     portgroup['node_uuid'] = kw.get('node_uuid', node['uuid'])
     return portgroup
+
+
+_ALLOCATION_POST_FIELDS = {'resource_class', 'uuid', 'traits',
+                           'candidate_nodes', 'name', 'extra'}
+
+
+def allocation_post_data(**kw):
+    """Return an Allocation object without internal attributes."""
+    allocation = db_utils.get_test_allocation(**kw)
+    return {key: value for key, value in allocation.items()
+            if key in _ALLOCATION_POST_FIELDS}
diff --git a/releasenotes/notes/allocation-api-6ac2d262689f5f59.yaml b/releasenotes/notes/allocation-api-6ac2d262689f5f59.yaml
new file mode 100644
index 0000000000..a62079fb05
--- /dev/null
+++ b/releasenotes/notes/allocation-api-6ac2d262689f5f59.yaml
@@ -0,0 +1,10 @@
+---
+features:
+  - |
+    Introduces allocation API. This API allows finding and reserving a node
+    by its resource class, traits and optional list of candidate nodes.
+    Introduces new API endpoints:
+
+    * ``GET/POST /v1/allocations``
+    * ``GET/DELETE /v1/allocations/<ID or name>``
+    * ``GET/DELETE /v1/nodes/<ID or name>/allocation``