diff --git a/heat/engine/resources/openstack/heat/delay.py b/heat/engine/resources/openstack/heat/delay.py
new file mode 100644
index 0000000000..643a2b2c0c
--- /dev/null
+++ b/heat/engine/resources/openstack/heat/delay.py
@@ -0,0 +1,172 @@
+#
+#    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 random
+
+from heat.common import exception
+from heat.common.i18n import _
+from heat.engine import constraints
+from heat.engine import properties
+from heat.engine import resource
+from heat.engine import support
+
+from oslo_utils import timeutils
+
+
+class Delay(resource.Resource):
+    """A resource that pauses for a configurable delay.
+
+    By manipulating the dependency relationships between resources in the
+    template, a delay can be inserted at an arbitrary point during e.g. stack
+    creation or deletion. They delay will occur after any resource that it
+    depends on during CREATE or SUSPEND, and before any resource that it
+    depends on during DELETE or RESUME. Similarly, it will occur before any
+    resource that depends on it during CREATE or SUSPEND, and after any
+    resource thet depends on it during DELETE or RESUME.
+
+    If a non-zero maximum jitter is specified, a random amount of jitter -
+    chosen with uniform probability in the range from 0 to the product of the
+    maximum jitter value and the jitter multiplier (1s by default) - is added
+    to the minimum delay time. This can be used, for example, in the scaled
+    unit of a large scaling group to prevent 'thundering herd' issues.
+    """
+
+    support_status = support.SupportStatus(version='11.0.0')
+
+    _ALLOWED_ACTIONS = (
+        resource.Resource.CREATE,
+        resource.Resource.DELETE,
+        resource.Resource.SUSPEND,
+        resource.Resource.RESUME,
+    )
+
+    PROPERTIES = (
+        MIN_WAIT_SECS, MAX_JITTER, JITTER_MULTIPLIER_SECS, ACTIONS,
+    ) = (
+        'min_wait', 'max_jitter', 'jitter_multiplier', 'actions',
+    )
+
+    properties_schema = {
+        MIN_WAIT_SECS: properties.Schema(
+            properties.Schema.NUMBER,
+            _('Minimum time in seconds to wait during the specified actions.'),
+            update_allowed=True,
+            default=0,
+            constraints=[
+                constraints.Range(min=0)
+            ]
+        ),
+        MAX_JITTER: properties.Schema(
+            properties.Schema.NUMBER,
+            _('Maximum jitter to add to the minimum wait time.'),
+            update_allowed=True,
+            default=0,
+            constraints=[
+                constraints.Range(min=0),
+            ]
+        ),
+        JITTER_MULTIPLIER_SECS: properties.Schema(
+            properties.Schema.NUMBER,
+            _('Number of seconds to multiply the maximum jitter value by.'),
+            update_allowed=True,
+            default=1.0,
+            constraints=[
+                constraints.Range(min=0),
+            ]
+        ),
+        ACTIONS: properties.Schema(
+            properties.Schema.LIST,
+            _('Actions during which the delay will occur.'),
+            update_allowed=True,
+            default=[resource.Resource.CREATE],
+            constraints=[constraints.AllowedValues(_ALLOWED_ACTIONS)]
+        ),
+    }
+
+    attributes_schema = {}
+
+    def _delay_parameters(self):
+        """Return a tuple of the min delay and max jitter, in seconds."""
+        min_wait_secs = self.properties[self.MIN_WAIT_SECS]
+        max_jitter_secs = (self.properties[self.MAX_JITTER] *
+                           self.properties[self.JITTER_MULTIPLIER_SECS])
+        return min_wait_secs, max_jitter_secs
+
+    def validate(self):
+        result = super(Delay, self).validate()
+        if not self.stack.strict_validate:
+            return result
+
+        min_wait_secs, max_jitter_secs = self._delay_parameters()
+        max_wait = min_wait_secs + max_jitter_secs
+        if max_wait > self.stack.timeout_secs():
+            raise exception.StackValidationFailed(_('%(res_type)s maximum '
+                                                    'delay %(max_wait)ss '
+                                                    'exceeds stack timeout.') %
+                                                  {'res_type': self.type,
+                                                   'max_wait': max_wait})
+        return result
+
+    def _wait_secs(self, action):
+        """Return a (randomised) wait time for the specified action."""
+        if action not in self.properties[self.ACTIONS]:
+            return 0
+
+        min_wait_secs, max_jitter_secs = self._delay_parameters()
+        return min_wait_secs + (max_jitter_secs * random.random())
+
+    def _handle_action(self):
+        """Return a tuple of the start time in UTC and the time to wait."""
+        return timeutils.utcnow(), self._wait_secs(self.action)
+
+    @staticmethod
+    def _check_complete(started_at, wait_secs):
+        if not wait_secs:
+            return True
+        elapsed_secs = (timeutils.utcnow() - started_at).total_seconds()
+        if elapsed_secs >= wait_secs:
+            return True
+        remaining = wait_secs - elapsed_secs
+        if remaining >= 4:
+            raise resource.PollDelay(int(remaining // 2))
+        return False
+
+    def handle_create(self):
+        return self._handle_action()
+
+    def handle_delete(self):
+        return self._handle_action()
+
+    def handle_suspend(self):
+        return self._handle_action()
+
+    def handle_resume(self):
+        return self._handle_action()
+
+    def check_create_complete(self, cookie):
+        return self._check_complete(*cookie)
+
+    def check_delete_complete(self, cookie):
+        return self._check_complete(*cookie)
+
+    def check_suspend_complete(self, cookie):
+        return self._check_complete(*cookie)
+
+    def check_resume_complete(self, cookie):
+        return self._check_status_complete(*cookie)
+
+
+def resource_mapping():
+    return {
+        'OS::Heat::Delay': Delay,
+    }
diff --git a/heat/tests/openstack/heat/test_delay.py b/heat/tests/openstack/heat/test_delay.py
new file mode 100644
index 0000000000..9a56e0355a
--- /dev/null
+++ b/heat/tests/openstack/heat/test_delay.py
@@ -0,0 +1,154 @@
+#
+#    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.
+
+from heat.common import exception
+from heat.common import template_format
+from heat.engine import resource
+from heat.engine.resources.openstack.heat import delay
+from heat.engine import status
+from heat.tests import common
+from heat.tests import utils
+
+from oslo_utils import fixture as utils_fixture
+from oslo_utils import timeutils
+
+
+class TestDelay(common.HeatTestCase):
+
+    simple_template = template_format.parse('''
+heat_template_version: '2016-10-14'
+resources:
+  constant:
+    type: OS::Heat::Delay
+    properties:
+      min_wait: 3
+  variable:
+    type: OS::Heat::Delay
+    properties:
+      min_wait: 1.6
+      max_jitter: 4.2
+      actions:
+        - CREATE
+        - DELETE
+  variable_prod:
+    type: OS::Heat::Delay
+    properties:
+      min_wait: 2
+      max_jitter: 666
+      jitter_multiplier: 0.1
+      actions:
+        - DELETE
+''')
+
+    def test_delay_params(self):
+        stk = utils.parse_stack(self.simple_template)
+
+        self.assertEqual((3, 0), stk['constant']._delay_parameters())
+
+        self.assertEqual((1.6, 4.2), stk['variable']._delay_parameters())
+
+        min_wait, max_jitter = stk['variable_prod']._delay_parameters()
+        self.assertEqual(2, min_wait)
+        self.assertAlmostEqual(66.6, max_jitter)
+
+    def test_wait_secs_create(self):
+        stk = utils.parse_stack(self.simple_template)
+        action = status.ResourceStatus.CREATE
+
+        self.assertEqual(3, stk['constant']._wait_secs(action))
+
+        variable = stk['variable']._wait_secs(action)
+        self.assertGreaterEqual(variable, 1.6)
+        self.assertLessEqual(variable, 5.8)
+        self.assertNotEqual(variable, stk['variable']._wait_secs(action))
+
+        self.assertEqual(0, stk['variable_prod']._wait_secs(action))
+
+    def test_wait_secs_delete(self):
+        stk = utils.parse_stack(self.simple_template)
+        action = status.ResourceStatus.DELETE
+
+        self.assertEqual(0, stk['constant']._wait_secs(action))
+
+        variable = stk['variable']._wait_secs(action)
+        self.assertGreaterEqual(variable, 1.6)
+        self.assertLessEqual(variable, 5.8)
+        self.assertNotEqual(variable, stk['variable']._wait_secs(action))
+
+        variable_prod = stk['variable_prod']._wait_secs(action)
+        self.assertGreaterEqual(variable_prod, 2.0)
+        self.assertLessEqual(variable_prod, 68.6)
+        self.assertNotEqual(variable_prod,
+                            stk['variable_prod']._wait_secs(action))
+
+    def test_wait_secs_update(self):
+        stk = utils.parse_stack(self.simple_template)
+        action = status.ResourceStatus.UPDATE
+
+        self.assertEqual(0, stk['constant']._wait_secs(action))
+        self.assertEqual(0, stk['variable']._wait_secs(action))
+        self.assertEqual(0, stk['variable_prod']._wait_secs(action))
+
+    def test_validate_success(self):
+        stk = utils.parse_stack(self.simple_template)
+        for res in stk.resources.values():
+            self.assertIsNone(res.validate())
+
+    def test_validate_failure(self):
+        stk = utils.parse_stack(self.simple_template)
+        stk.timeout_mins = 1
+        self.assertRaises(exception.StackValidationFailed,
+                          stk['variable_prod'].validate)
+
+
+class DelayCompletionTest(common.HeatTestCase):
+    def setUp(self):
+        super(DelayCompletionTest, self).setUp()
+        self.time_fixture = utils_fixture.TimeFixture()
+        self.useFixture(self.time_fixture)
+
+    def test_complete_no_wait(self):
+        now = timeutils.utcnow()
+        self.time_fixture.advance_time_seconds(-1)
+        self.assertEqual(True, delay.Delay._check_complete(now, 0))
+
+    def test_complete(self):
+        now = timeutils.utcnow()
+        self.time_fixture.advance_time_seconds(5.1)
+        self.assertEqual(True, delay.Delay._check_complete(now, 5.1))
+
+    def test_already_complete(self):
+        now = timeutils.utcnow()
+        self.time_fixture.advance_time_seconds(5.1)
+        self.assertEqual(True, delay.Delay._check_complete(now, 5))
+
+    def test_incomplete_short_delay(self):
+        now = timeutils.utcnow()
+        self.time_fixture.advance_time_seconds(2)
+        self.assertEqual(False, delay.Delay._check_complete(now, 5))
+
+    def test_incomplete_moderate_delay(self):
+        now = timeutils.utcnow()
+        self.time_fixture.advance_time_seconds(2)
+        poll_del = self.assertRaises(resource.PollDelay,
+                                     delay.Delay._check_complete,
+                                     now, 6)
+        self.assertEqual(2, poll_del.period)
+
+    def test_incomplete_long_delay(self):
+        now = timeutils.utcnow()
+        self.time_fixture.advance_time_seconds(0.1)
+        poll_del = self.assertRaises(resource.PollDelay,
+                                     delay.Delay._check_complete,
+                                     now, 62)
+        self.assertEqual(30, poll_del.period)
diff --git a/releasenotes/notes/delay-resource-e20ba61f31799f6e.yaml b/releasenotes/notes/delay-resource-e20ba61f31799f6e.yaml
new file mode 100644
index 0000000000..a5e8f0400d
--- /dev/null
+++ b/releasenotes/notes/delay-resource-e20ba61f31799f6e.yaml
@@ -0,0 +1,6 @@
+---
+features:
+  - |
+    A new OS::Heat::Delay resource type allows users to work around thundering
+    herd issues in large templates by adding a random delay (with configurable
+    jitter) into the workflow.