diff --git a/HACKING.rst b/HACKING.rst
index cb9821e..c257a0c 100644
--- a/HACKING.rst
+++ b/HACKING.rst
@@ -128,6 +128,12 @@
 Test class level resources should be defined in the `resource_setup` method of
 the test class, except for any credential obtained from the credentials
 provider, which should be set-up in the `setup_credentials` method.
+Cleanup is best scheduled using `addClassResourceCleanup` which ensures that
+the cleanup code is always invoked, and in reverse order with respect to the
+creation order.
+
+In both cases - test level and class level cleanups - a wait loop should be
+scheduled before the actual delete of resources with an asynchronous delete.
 
 The test base class `BaseTestCase` defines Tempest framework for class level
 fixtures. `setUpClass` and `tearDownClass` are defined here and cannot be
diff --git a/doc/source/write_tests.rst b/doc/source/write_tests.rst
index aec55e9..5a2876e 100644
--- a/doc/source/write_tests.rst
+++ b/doc/source/write_tests.rst
@@ -59,10 +59,16 @@
  * setup_clients
  * resource_setup
 
-which is executed in that order. An example of a TestCase which defines all
+which is executed in that order. Cleanup of resources provisioned during
+the resource_setup must be scheduled right after provisioning using
+the addClassResourceCleanp helper. The resource cleanups stacked this way
+are executed in reverse order during tearDownClass, before the cleanup of
+test credentials takes place. An example of a TestCase which defines all
 of these would be::
-
+  
+  from tempest.common import waiters
   from tempest import config
+  from tempest.lib.common.utils import test_utils
   from tempest import test
 
   CONF = config.CONF
@@ -111,6 +117,13 @@
         """
         super(TestExampleCase, cls).resource_setup()
         cls.shared_server = cls.servers_client.create_server(...)
+        cls.addClassResourceCleanup(waiters.wait_for_server_termination,
+                                    cls.servers_client,
+                                    cls.shared_server['id'])
+        cls.addClassResourceCleanup(
+            test_utils.call_and_ignore_notfound_exc(
+                cls.servers_client.delete_server,
+                cls.shared_server['id']))
 
 .. _credentials:
 
diff --git a/tempest/api/compute/admin/test_auto_allocate_network.py b/tempest/api/compute/admin/test_auto_allocate_network.py
index 0c80252..a22f838 100644
--- a/tempest/api/compute/admin/test_auto_allocate_network.py
+++ b/tempest/api/compute/admin/test_auto_allocate_network.py
@@ -143,6 +143,8 @@
             test_utils.call_and_ignore_notfound_exc(
                 cls.networks_client.delete_network, network['id'])
 
+        super(AutoAllocateNetworkTest, cls).resource_cleanup()
+
     @decorators.idempotent_id('5eb7b8fa-9c23-47a2-9d7d-02ed5809dd34')
     def test_server_create_no_allocate(self):
         """Tests that no networking is allocated for the server."""
diff --git a/tempest/test.py b/tempest/test.py
index a4cc2cc..13a91fd 100644
--- a/tempest/test.py
+++ b/tempest/test.py
@@ -109,6 +109,9 @@
     validation_resources = {}
     network_resources = {}
 
+    # Stack of resource cleanups
+    _class_cleanups = []
+
     # NOTE(sdague): log_format is defined inline here instead of using the oslo
     # default because going through the config path recouples config to the
     # stress tests too early, and depending on testr order will fail unit tests
@@ -122,7 +125,14 @@
     TIMEOUT_SCALING_FACTOR = 1
 
     @classmethod
+    def _reset_class(cls):
+        cls.__resource_cleaup_called = False
+        cls._class_cleanups = []
+
+    @classmethod
     def setUpClass(cls):
+        # Reset state
+        cls._reset_class()
         # It should never be overridden by descendants
         if hasattr(super(BaseTestCase, cls), 'setUpClass'):
             super(BaseTestCase, cls).setUpClass()
@@ -171,12 +181,23 @@
             # exception at the end
             try:
                 teardown()
+                if name == 'resources':
+                    if not cls.__resource_cleaup_called:
+                        raise RuntimeError(
+                            "resource_cleanup for %s did not call the "
+                            "super's resource_cleanup" % cls.__name__)
             except Exception as te:
                 sys_exec_info = sys.exc_info()
                 tetype = sys_exec_info[0]
-                # TODO(andreaf): Till we have the ability to cleanup only
-                # resources that were successfully setup in resource_cleanup,
-                # log AttributeError as info instead of exception.
+                # TODO(andreaf): Resource cleanup is often implemented by
+                # storing an array of resources at class level, and cleaning
+                # them up during `resource_cleanup`.
+                # In case of failure during setup, some resource arrays might
+                # not be defined at all, in which case the cleanup code might
+                # trigger an AttributeError. In such cases we log
+                # AttributeError as info instead of exception. Once all
+                # cleanups are migrated to addClassResourceCleanup we can
+                # remove this.
                 if tetype is AttributeError and name == 'resources':
                     LOG.info("tearDownClass of %s failed: %s", name, te)
                 else:
@@ -299,7 +320,65 @@
 
     @classmethod
     def resource_setup(cls):
-        """Class level resource setup for test cases."""
+        """Class level resource setup for test cases.
+
+        `resource_setup` is invoked once all credentials (and related network
+        resources have been provisioned and after client aliases - if any -
+        have been defined.
+
+        The use case for `resource_setup` is test optimization: provisioning
+        of project-specific "expensive" resources that are not dirtied by tests
+        and can thus safely be re-used by multiple tests.
+
+        System wide resources shared by all tests could instead be provisioned
+        only once, before the test run.
+
+        Resources provisioned here must be cleaned up during
+        `resource_cleanup`. This is best achieved by scheduling a cleanup via
+        `addClassResourceCleanup`.
+
+        Some test resources have an asynchronous delete process. It's best
+        practice for them to schedule a wait for delete via
+        `addClassResourceCleanup` to avoid having resources in process of
+        deletion when we reach the credentials cleanup step.
+
+        Example::
+
+            @classmethod
+            def resource_setup(cls):
+                super(MyTest, cls).resource_setup()
+                servers = cls.os_primary.compute.ServersClient()
+                # Schedule delete and wait so that we can first delete the
+                # two servers and then wait for both to delete
+                # Create server 1
+                cls.shared_server = servers.create_server()
+                # Create server 2. If something goes wrong we schedule cleanup
+                # of server 1 anyways.
+                try:
+                    cls.shared_server2 = servers.create_server()
+                    # Wait server 2
+                    cls.addClassResourceCleanup(
+                        waiters.wait_for_server_termination,
+                        servers, cls.shared_server2['id'],
+                        ignore_error=False)
+                finally:
+                    # Wait server 1
+                    cls.addClassResourceCleanup(
+                        waiters.wait_for_server_termination,
+                        servers, cls.shared_server['id'],
+                        ignore_error=False)
+                        # Delete server 1
+                    cls.addClassResourceCleanup(
+                        test_utils.call_and_ignore_notfound_exc,
+                        servers.delete_server,
+                        cls.shared_server['id'])
+                    # Delete server 2 (if it was created)
+                    if hasattr(cls, 'shared_server2'):
+                        cls.addClassResourceCleanup(
+                            test_utils.call_and_ignore_notfound_exc,
+                            servers.delete_server,
+                            cls.shared_server2['id'])
+        """
         if (CONF.validation.ip_version_for_ssh not in (4, 6) and
             CONF.service_available.neutron):
             msg = "Invalid IP version %s in ip_version_for_ssh. Use 4 or 6"
@@ -322,9 +401,45 @@
     def resource_cleanup(cls):
         """Class level resource cleanup for test cases.
 
-        Resource cleanup must be able to handle the case of partially setup
-        resources, in case a failure during `resource_setup` should happen.
+        Resource cleanup processes the stack or cleanups produced by
+        `addClassResourceCleanup` and then cleans up validation resources
+        if any were provisioned.
+
+        All cleanups are processed whatever the outcome. Exceptions are
+        accumulated and re-raised as a `MultipleExceptions` at the end.
+
+        In most cases test cases won't need to override `resource_cleanup`,
+        but if they do they must invoke `resource_cleanup` on super.
+
+        Example::
+
+            class TestWithReallyComplexCleanup(test.BaseTestCase):
+
+                @classmethod
+                def resource_setup(cls):
+                    # provision resource A
+                    cls.addClassResourceCleanup(delete_resource, A)
+                    # provision resource B
+                    cls.addClassResourceCleanup(delete_resource, B)
+
+                @classmethod
+                def resource_cleanup(cls):
+                    # It's possible to override resource_cleanup but in most
+                    # cases it shouldn't be required. Nothing that may fail
+                    # should be executed before the call to super since it
+                    # might cause resource leak in case of error.
+                    super(TestWithReallyComplexCleanup, cls).resource_cleanup()
+                    # At this point test credentials are still available but
+                    # anything from the cleanup stack has been already deleted.
         """
+        cls.__resource_cleaup_called = True
+        cleanup_errors = []
+        while cls._class_cleanups:
+            try:
+                fn, args, kwargs = cls._class_cleanups.pop()
+                fn(*args, **kwargs)
+            except Exception:
+                cleanup_errors.append(sys.exc_info())
         if cls.validation_resources:
             if hasattr(cls, "os_primary"):
                 vr = cls.validation_resources
@@ -335,6 +450,25 @@
             else:
                 LOG.warning("Client manager not found, validation resources "
                             "not deleted")
+        if cleanup_errors:
+            raise testtools.MultipleExceptions(*cleanup_errors)
+
+    @classmethod
+    def addClassResourceCleanup(cls, fn, *arguments, **keywordArguments):
+        """Add a cleanup function to be called during resource_cleanup.
+
+        Functions added with addClassResourceCleanup will be called in reverse
+        order of adding at the beginning of resource_cleanup, before any
+        credential, networking or validation resources cleanup is processed.
+
+        If a function added with addClassResourceCleanup raises an exception,
+        the error will be recorded as a test error, and the next cleanup will
+        then be run.
+
+        Cleanup functions are always called during the test class tearDown
+        fixture, even if an exception occured during setUp or tearDown.
+        """
+        cls._class_cleanups.append((fn, arguments, keywordArguments))
 
     def setUp(self):
         super(BaseTestCase, self).setUp()
diff --git a/tempest/tests/test_test.py b/tempest/tests/test_test.py
new file mode 100644
index 0000000..ed08c3a
--- /dev/null
+++ b/tempest/tests/test_test.py
@@ -0,0 +1,173 @@
+# Copyright 2017 IBM Corp
+# All Rights Reserved.
+#
+#    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 sys
+
+import mock
+from oslo_config import cfg
+import testtools
+
+from tempest import config
+from tempest import test
+from tempest.tests import base
+from tempest.tests import fake_config
+
+
+if sys.version_info >= (2, 7):
+    import unittest
+else:
+    import unittest2 as unittest
+
+
+class LoggingTestResult(testtools.TestResult):
+
+    def __init__(self, log, *args, **kwargs):
+        super(LoggingTestResult, self).__init__(*args, **kwargs)
+        self.log = log
+
+    def addError(self, test, err=None, details=None):
+        self.log.append((test, err, details))
+
+
+class TestTempestBaseTestClass(base.TestCase):
+
+    def setUp(self):
+        super(TestTempestBaseTestClass, self).setUp()
+        self.useFixture(fake_config.ConfigFixture())
+        self.patchobject(config, 'TempestConfigPrivate',
+                         fake_config.FakePrivate)
+
+        class ParentTest(test.BaseTestCase):
+
+            def runTest(self):
+                pass
+
+        self.parent_test = ParentTest
+
+    @mock.patch(
+        'tempest.common.validation_resources.clear_validation_resources',
+        autospec=True)
+    def test_resource_cleanup(self, mock_vr):
+        cfg.CONF.set_default('neutron', False, 'service_available')
+        exp_args = (1, 2,)
+        exp_kwargs = {'a': 1, 'b': 2}
+        exp_vr = {'keypair': 'kp1', 'floating_ip': 'fip2'}
+        mock1 = mock.Mock()
+        mock2 = mock.Mock()
+        exp_functions = [mock1, mock2]
+
+        class TestWithCleanups(self.parent_test):
+
+            # set fake validation resources
+            validation_resources = exp_vr
+
+            # set fake clients
+            os_primary = 'os_primary'
+
+            @classmethod
+            def resource_setup(cls):
+                for fn in exp_functions:
+                    cls.addClassResourceCleanup(fn, *exp_args,
+                                                **exp_kwargs)
+
+        test_cleanups = TestWithCleanups()
+        suite = unittest.TestSuite((test_cleanups,))
+        log = []
+        result = LoggingTestResult(log)
+        suite.run(result)
+        # No exception raised - error log is empty
+        self.assertFalse(log)
+        # All stacked resource cleanups invoked
+        mock1.assert_called_once_with(*exp_args, **exp_kwargs)
+        mock2.assert_called_once_with(*exp_args, **exp_kwargs)
+        self.assertEqual(1, mock_vr.call_count)
+        # Cleanup stack is empty
+        self.assertEqual(0, len(test_cleanups._class_cleanups))
+        # Assert vrs are cleaned up
+        self.assertIn(mock.call(TestWithCleanups.os_primary, use_neutron=False,
+                                **exp_vr), mock_vr.call_args_list)
+
+    @mock.patch(
+        'tempest.common.validation_resources.clear_validation_resources',
+        autospec=True)
+    def test_resource_cleanup_failures(self, mock_vr):
+        cfg.CONF.set_default('neutron', False, 'service_available')
+        exp_args = (1, 2,)
+        exp_kwargs = {'a': 1, 'b': 2}
+        exp_vr = {'keypair': 'kp1', 'floating_ip': 'fip2'}
+        mock1 = mock.Mock()
+        mock1.side_effect = Exception('mock1 resource cleanup failure')
+        mock2 = mock.Mock()
+        exp_functions = [mock1, mock2]
+
+        class TestWithFailingCleanups(self.parent_test):
+
+            # set fake validation resources
+            validation_resources = exp_vr
+
+            # set fake clients
+            os_primary = 'os_primary'
+
+            @classmethod
+            def resource_setup(cls):
+                for fn in exp_functions:
+                    cls.addClassResourceCleanup(fn, *exp_args,
+                                                **exp_kwargs)
+
+        test_cleanups = TestWithFailingCleanups()
+        suite = unittest.TestSuite((test_cleanups,))
+        log = []
+        result = LoggingTestResult(log)
+        suite.run(result)
+        # One multiple exception captured
+        self.assertEqual(1, len(log))
+        # [0]: test, err, details [1] -> exc_info
+        # Type, Exception, traceback [1] -> MultipleException
+        found_exc = log[0][1][1]
+        self.assertTrue(isinstance(found_exc, testtools.MultipleExceptions))
+        self.assertEqual(1, len(found_exc.args))
+        # Each arg is exc_info - match messages and order
+        self.assertIn('mock1 resource', str(found_exc.args[0][1]))
+        # All stacked resource cleanups invoked
+        mock1.assert_called_once_with(*exp_args, **exp_kwargs)
+        mock2.assert_called_once_with(*exp_args, **exp_kwargs)
+        self.assertEqual(1, mock_vr.call_count)
+        # Cleanup stack is empty
+        self.assertEqual(0, len(test_cleanups._class_cleanups))
+        # Assert fake vr are cleaned up
+        self.assertIn(mock.call(TestWithFailingCleanups.os_primary,
+                                use_neutron=False, **exp_vr),
+                      mock_vr.call_args_list)
+
+    def test_super_resource_cleanup_not_invoked(self):
+
+        class BadResourceCleanup(self.parent_test):
+
+            @classmethod
+            def resource_cleanup(cls):
+                pass
+
+        bad_class = BadResourceCleanup()
+        suite = unittest.TestSuite((bad_class,))
+        log = []
+        result = LoggingTestResult(log)
+        suite.run(result)
+        # One multiple exception captured
+        self.assertEqual(1, len(log))
+        # [0]: test, err, details [1] -> exc_info
+        # Type, Exception, traceback [1] -> RuntimeError
+        found_exc = log[0][1][1]
+        self.assertTrue(isinstance(found_exc, RuntimeError))
+        self.assertIn(BadResourceCleanup.__name__, str(found_exc))