diff --git a/playbooks/devstack-tempest-ipv6.yaml b/playbooks/devstack-tempest-ipv6.yaml
index 568077e20c..89eec6de39 100644
--- a/playbooks/devstack-tempest-ipv6.yaml
+++ b/playbooks/devstack-tempest-ipv6.yaml
@@ -17,6 +17,16 @@
     # fail early if anything missing the IPv6 settings or deployments.
     - devstack-ipv6-only-deployments-verification
   tasks:
+    - name: Run tempest cleanup init-saved-state
+      include_role:
+        name: tempest-cleanup
+      vars:
+        init_saved_state: true
+      when: (run_tempest_dry_cleanup is defined and run_tempest_dry_cleanup | bool) or
+            (run_tempest_cleanup is defined and run_tempest_cleanup | bool) or
+            (run_tempest_fail_if_leaked_resources is defined and run_tempest_fail_if_leaked_resources | bool) or
+            (run_tempest_cleanup_prefix is defined and run_tempest_cleanup_prefix | bool)
+
     - name: Run Tempest version <= 26.0.0
       include_role:
         name: run-tempest-26
@@ -30,3 +40,15 @@
       when:
         - zuul.branch is defined
         - zuul.branch not in ["stable/ocata", "stable/pike", "stable/queens", "stable/rocky", "stable/stein"]
+
+    - name: Run tempest cleanup dry-run
+      include_role:
+        name: tempest-cleanup
+      vars:
+        dry_run: true
+      when: run_tempest_dry_cleanup is defined and run_tempest_dry_cleanup | bool
+
+    - name: Run tempest cleanup
+      include_role:
+        name: tempest-cleanup
+      when: run_tempest_cleanup is defined and run_tempest_cleanup | bool
diff --git a/playbooks/devstack-tempest.yaml b/playbooks/devstack-tempest.yaml
index 6d70bc349e..5fb1afcb89 100644
--- a/playbooks/devstack-tempest.yaml
+++ b/playbooks/devstack-tempest.yaml
@@ -32,7 +32,8 @@
         init_saved_state: true
       when: (run_tempest_dry_cleanup is defined and run_tempest_dry_cleanup | bool) or
             (run_tempest_cleanup is defined and run_tempest_cleanup | bool) or
-            (run_tempest_fail_if_leaked_resources is defined and run_tempest_fail_if_leaked_resources | bool)
+            (run_tempest_fail_if_leaked_resources is defined and run_tempest_fail_if_leaked_resources | bool) or
+            (run_tempest_cleanup_prefix is defined and run_tempest_cleanup_prefix | bool)
 
     - name: Run Tempest version <= 26.0.0
       include_role:
diff --git a/releasenotes/notes/Allow-tempest-cleanup-delete-resources-based-on-prefix-96d9562f1f30e979.yaml b/releasenotes/notes/Allow-tempest-cleanup-delete-resources-based-on-prefix-96d9562f1f30e979.yaml
new file mode 100644
index 0000000000..872f664fd1
--- /dev/null
+++ b/releasenotes/notes/Allow-tempest-cleanup-delete-resources-based-on-prefix-96d9562f1f30e979.yaml
@@ -0,0 +1,10 @@
+---
+features:
+  - |
+    We add a new argument, ``--prefix``, to ``tempest cleanup`` tool that will
+    allow users delete only resources that match the prefix. When this option
+    is used, ``saved_state.json`` file is not needed (no need to run with
+    ``--init-saved-state`` first). If there is one, it will be ignored and the
+    cleanup will be done based on the given prefix only.
+    Note, that some resources are not named thus they will not be deleted when
+    filtering based on the prefix.
diff --git a/roles/tempest-cleanup/README.rst b/roles/tempest-cleanup/README.rst
index d1fad900a4..d43319c33d 100644
--- a/roles/tempest-cleanup/README.rst
+++ b/roles/tempest-cleanup/README.rst
@@ -40,6 +40,12 @@ saved_state.json file.
    some must have been leaked. This can be also used to verify that tempest
    cleanup was successful.
 
+.. zuul:rolevar:: run_tempest_cleanup_prefix
+   :default: false
+
+   When true, tempest cleanup will be called with '--prefix tempest' to delete
+   only resources with names that match the prefix. This option can be used
+   together with dry_run.
 
 Role usage
 ----------
diff --git a/roles/tempest-cleanup/defaults/main.yaml b/roles/tempest-cleanup/defaults/main.yaml
index ce78bdbe38..8060b295cd 100644
--- a/roles/tempest-cleanup/defaults/main.yaml
+++ b/roles/tempest-cleanup/defaults/main.yaml
@@ -2,3 +2,4 @@ devstack_base_dir: /opt/stack
 init_saved_state: false
 dry_run: false
 run_tempest_fail_if_leaked_resources: false
+run_tempest_cleanup_prefix: false
diff --git a/roles/tempest-cleanup/tasks/dry_run.yaml b/roles/tempest-cleanup/tasks/dry_run.yaml
index 46749ab5af..07e1b634cd 100644
--- a/roles/tempest-cleanup/tasks/dry_run.yaml
+++ b/roles/tempest-cleanup/tasks/dry_run.yaml
@@ -5,3 +5,12 @@
   command: tox -evenv-tempest -- tempest cleanup --dry-run --debug
   args:
     chdir: "{{ devstack_base_dir }}/tempest"
+  when: not run_tempest_cleanup_prefix
+
+- name: Run tempest cleanup dry-run with tempest prefix
+  become: yes
+  become_user: tempest
+  command: tox -evenv-tempest -- tempest cleanup --dry-run --debug --prefix tempest
+  args:
+    chdir: "{{ devstack_base_dir }}/tempest"
+  when: run_tempest_cleanup_prefix
\ No newline at end of file
diff --git a/roles/tempest-cleanup/tasks/main.yaml b/roles/tempest-cleanup/tasks/main.yaml
index c1d63f03eb..7ef4928042 100644
--- a/roles/tempest-cleanup/tasks/main.yaml
+++ b/roles/tempest-cleanup/tasks/main.yaml
@@ -27,6 +27,15 @@
       command: tox -evenv-tempest -- tempest cleanup --debug
       args:
         chdir: "{{ devstack_base_dir }}/tempest"
+      when: not run_tempest_cleanup_prefix
+
+    - name: Run tempest cleanup with tempest prefix
+      become: yes
+      become_user: tempest
+      command: tox -evenv-tempest -- tempest cleanup --debug --prefix tempest
+      args:
+        chdir: "{{ devstack_base_dir }}/tempest"
+      when: run_tempest_cleanup_prefix
 
 - when:
     - run_tempest_fail_if_leaked_resources
diff --git a/tempest/cmd/cleanup.py b/tempest/cmd/cleanup.py
index a8a344a691..2a406dedcf 100644
--- a/tempest/cmd/cleanup.py
+++ b/tempest/cmd/cleanup.py
@@ -26,6 +26,10 @@ specified in ``tempest.conf`` is never deleted.
 Example Run
 -----------
 
+.. warning::
+
+    We advice not to run tempest cleanup on production environments.
+
 .. warning::
 
     If step 1 is skipped in the example below, the cleanup procedure
@@ -45,7 +49,10 @@ Runtime Arguments
 * ``--init-saved-state``: Initializes the saved state of the OpenStack
   deployment and will output a ``saved_state.json`` file containing resources
   from your deployment that will be preserved from the cleanup command. This
-  should be done prior to running Tempest tests.
+  should be done prior to running Tempest tests. Note, that if other users of
+  your cloud could have created resources after running ``--init-saved-state``,
+  it would not protect those resources as they wouldn't be present in the
+  saved_state.json file.
 
 * ``--delete-tempest-conf-objects``: If option is present, then the command
   will delete the admin project in addition to the resources associated with
@@ -58,7 +65,27 @@ Runtime Arguments
   global objects that will be removed (domains, flavors, images, roles,
   projects, and users). Once the cleanup command is executed (e.g. run without
   parameters), running it again with ``--dry-run`` should yield an empty
-  report.
+  report. We STRONGLY ENCOURAGE to run ``tempest cleanup`` with ``--dry-run``
+  first and then verify that the resources listed in the ``dry_run.json`` file
+  are meant to be deleted.
+
+* ``--prefix``: Only resources that match the prefix will be deleted. When this
+  option is used, ``saved_state.json`` file is not needed (no need to run with
+  ``--init-saved-state`` first).
+
+  All tempest resources are created with the prefix value from the config
+  option ``resource_name_prefix`` in tempest.conf. To cleanup only the
+  resources created by tempest, you should use the prefix set in your
+  tempest.conf (the default value of ``resource_name_prefix`` is ``tempest``.
+
+  Note, that some resources are not named thus they will not be deleted when
+  filtering based on the prefix. This option will be ignored when
+  ``--init-saved-state`` is used so that it can capture the true init state -
+  all resources present at that moment. If there is any ``saved_state.json``
+  file present (e.g. if you ran the tempest cleanup with ``--init-saved-state``
+  before) and you run the tempest cleanup with ``--prefix``, the
+  ``saved_state.json`` file will be ignored and cleanup will be done based on
+  the passed prefix only.
 
 * ``--help``: Print the help text for the command and parameters.
 
@@ -157,6 +184,7 @@ class TempestCleanup(command.Command):
         is_dry_run = self.options.dry_run
         is_preserve = not self.options.delete_tempest_conf_objects
         is_save_state = False
+        cleanup_prefix = self.options.prefix
 
         if is_dry_run:
             self.dry_run_data["_projects_to_clean"] = {}
@@ -168,7 +196,8 @@ class TempestCleanup(command.Command):
                   'is_dry_run': is_dry_run,
                   'saved_state_json': self.json_data,
                   'is_preserve': False,
-                  'is_save_state': is_save_state}
+                  'is_save_state': is_save_state,
+                  'prefix': cleanup_prefix}
         project_service = cleanup_service.ProjectService(admin_mgr, **kwargs)
         projects = project_service.list()
         LOG.info("Processing %s projects", len(projects))
@@ -182,6 +211,7 @@ class TempestCleanup(command.Command):
                   'saved_state_json': self.json_data,
                   'is_preserve': is_preserve,
                   'is_save_state': is_save_state,
+                  'prefix': cleanup_prefix,
                   'got_exceptions': self.GOT_EXCEPTIONS}
         LOG.info("Processing global services")
         for service in self.global_services:
@@ -206,6 +236,7 @@ class TempestCleanup(command.Command):
         project_id = project['id']
         project_name = project['name']
         project_data = None
+        cleanup_prefix = self.options.prefix
         if is_dry_run:
             project_data = dry_run_data["_projects_to_clean"][project_id] = {}
             project_data['name'] = project_name
@@ -216,6 +247,7 @@ class TempestCleanup(command.Command):
                   'is_preserve': is_preserve,
                   'is_save_state': False,
                   'project_id': project_id,
+                  'prefix': cleanup_prefix,
                   'got_exceptions': self.GOT_EXCEPTIONS}
         for service in self.project_associated_services:
             svc = service(self.admin_mgr, **kwargs)
@@ -243,10 +275,26 @@ class TempestCleanup(command.Command):
                             help="Generate JSON file:" + DRY_RUN_JSON +
                             ", that reports the objects that would have "
                             "been deleted had a full cleanup been run.")
+        parser.add_argument('--prefix', dest='prefix', default=None,
+                            help="Only resources that match the prefix will "
+                            "be deleted (resources in saved_state.json are "
+                            "not taken into account). All tempest resources "
+                            "are created with the prefix value set by "
+                            "resource_name_prefix in tempest.conf, default "
+                            "prefix is tempest. Note that some resources are "
+                            "not named thus they will not be deleted when "
+                            "filtering based on the prefix. This opt will be "
+                            "ignored when --init-saved-state is used so that "
+                            "it can capture the true init state - all "
+                            "resources present at that moment.")
         return parser
 
     def get_description(self):
-        return 'Cleanup after tempest run'
+        return ('tempest cleanup tool, read the full documentation before '
+                'using this tool. We advice not to run it on production '
+                'environments. On environments where also other users may '
+                'create resources, we strongly advice using --dry-run '
+                'argument first and verify the content of dry_run.json file.')
 
     def _init_state(self):
         LOG.info("Initializing saved state.")
@@ -257,6 +305,10 @@ class TempestCleanup(command.Command):
                   'saved_state_json': data,
                   'is_preserve': False,
                   'is_save_state': True,
+                  # must be None as we want to capture true init state
+                  # (all resources present) thus no filtering based
+                  # on the prefix
+                  'prefix': None,
                   'got_exceptions': self.GOT_EXCEPTIONS}
         for service in self.global_services:
             svc = service(admin_mgr, **kwargs)
diff --git a/tempest/cmd/cleanup_service.py b/tempest/cmd/cleanup_service.py
index f2370f30ad..8651ab064d 100644
--- a/tempest/cmd/cleanup_service.py
+++ b/tempest/cmd/cleanup_service.py
@@ -115,6 +115,16 @@ class BaseService(object):
         return [item for item in item_list
                 if item['tenant_id'] == self.tenant_id]
 
+    def _filter_by_prefix(self, item_list):
+        items = [item for item in item_list
+                 if item['name'].startswith(self.prefix)]
+        return items
+
+    def _filter_out_ids_from_saved(self, item_list, attr):
+        items = [item for item in item_list if item['id']
+                 not in self.saved_state_json[attr].keys()]
+        return items
+
     def list(self):
         pass
 
@@ -156,10 +166,11 @@ class SnapshotService(BaseService):
     def list(self):
         client = self.client
         snaps = client.list_snapshots()['snapshots']
-        if not self.is_save_state:
+        if self.prefix:
+            snaps = self._filter_by_prefix(snaps)
+        elif not self.is_save_state:
             # recreate list removing saved snapshots
-            snaps = [snap for snap in snaps if snap['id']
-                     not in self.saved_state_json['snapshots'].keys()]
+            snaps = self._filter_out_ids_from_saved(snaps, 'snapshots')
         LOG.debug("List count, %s Snapshots", len(snaps))
         return snaps
 
@@ -194,10 +205,11 @@ class ServerService(BaseService):
         client = self.client
         servers_body = client.list_servers()
         servers = servers_body['servers']
-        if not self.is_save_state:
+        if self.prefix:
+            servers = self._filter_by_prefix(servers)
+        elif not self.is_save_state:
             # recreate list removing saved servers
-            servers = [server for server in servers if server['id']
-                       not in self.saved_state_json['servers'].keys()]
+            servers = self._filter_out_ids_from_saved(servers, 'servers')
         LOG.debug("List count, %s Servers", len(servers))
         return servers
 
@@ -227,10 +239,11 @@ class ServerGroupService(ServerService):
     def list(self):
         client = self.server_groups_client
         sgs = client.list_server_groups()['server_groups']
-        if not self.is_save_state:
+        if self.prefix:
+            sgs = self._filter_by_prefix(sgs)
+        elif not self.is_save_state:
             # recreate list removing saved server_groups
-            sgs = [sg for sg in sgs if sg['id']
-                   not in self.saved_state_json['server_groups'].keys()]
+            sgs = self._filter_out_ids_from_saved(sgs, 'server_groups')
         LOG.debug("List count, %s Server Groups", len(sgs))
         return sgs
 
@@ -263,7 +276,9 @@ class KeyPairService(BaseService):
     def list(self):
         client = self.client
         keypairs = client.list_keypairs()['keypairs']
-        if not self.is_save_state:
+        if self.prefix:
+            keypairs = self._filter_by_prefix(keypairs)
+        elif not self.is_save_state:
             # recreate list removing saved keypairs
             keypairs = [keypair for keypair in keypairs
                         if keypair['keypair']['name']
@@ -302,10 +317,11 @@ class VolumeService(BaseService):
     def list(self):
         client = self.client
         vols = client.list_volumes()['volumes']
-        if not self.is_save_state:
+        if self.prefix:
+            vols = self._filter_by_prefix(vols)
+        elif not self.is_save_state:
             # recreate list removing saved volumes
-            vols = [vol for vol in vols if vol['id']
-                    not in self.saved_state_json['volumes'].keys()]
+            vols = self._filter_out_ids_from_saved(vols, 'volumes')
         LOG.debug("List count, %s Volumes", len(vols))
         return vols
 
@@ -336,6 +352,10 @@ class VolumeQuotaService(BaseService):
         self.client = manager.volume_quotas_client_latest
 
     def delete(self):
+        if self.prefix:
+            # this means we're cleaning resources based on a certain prefix,
+            # this resource doesn't have a name, therefore do nothing
+            return
         client = self.client
         try:
             LOG.debug("Deleting Volume Quotas for project with id %s",
@@ -346,6 +366,10 @@ class VolumeQuotaService(BaseService):
                           self.project_id)
 
     def dry_run(self):
+        if self.prefix:
+            # this means we're cleaning resources based on a certain prefix,
+            # this resource doesn't have a name, therefore do nothing
+            return
         quotas = self.client.show_quota_set(
             self.project_id, params={'usage': True})['quota_set']
         self.data['volume_quotas'] = quotas
@@ -358,6 +382,10 @@ class NovaQuotaService(BaseService):
         self.limits_client = manager.limits_client
 
     def delete(self):
+        if self.prefix:
+            # this means we're cleaning resources based on a certain prefix,
+            # this resource doesn't have a name, therefore do nothing
+            return
         client = self.client
         try:
             LOG.debug("Deleting Nova Quotas for project with id %s",
@@ -368,6 +396,10 @@ class NovaQuotaService(BaseService):
                           self.project_id)
 
     def dry_run(self):
+        if self.prefix:
+            # this means we're cleaning resources based on a certain prefix,
+            # this resource doesn't have a name, therefore do nothing
+            return
         client = self.limits_client
         quotas = client.show_limits()['limits']
         self.data['compute_quotas'] = quotas['absolute']
@@ -379,6 +411,10 @@ class NetworkQuotaService(BaseService):
         self.client = manager.network_quotas_client
 
     def delete(self):
+        if self.prefix:
+            # this means we're cleaning resources based on a certain prefix,
+            # this resource doesn't have a name, therefore do nothing
+            return
         client = self.client
         try:
             LOG.debug("Deleting Network Quotas for project with id %s",
@@ -389,6 +425,10 @@ class NetworkQuotaService(BaseService):
                           self.project_id)
 
     def dry_run(self):
+        if self.prefix:
+            # this means we're cleaning resources based on a certain prefix,
+            # this resource doesn't have a name, therefore do nothing
+            return
         resp = [quota for quota in self.client.list_quotas()['quotas']
                 if quota['project_id'] == self.project_id]
         self.data['network_quotas'] = resp
@@ -422,11 +462,13 @@ class NetworkService(BaseNetworkService):
         client = self.networks_client
         networks = client.list_networks(**self.tenant_filter)
         networks = networks['networks']
-
-        if not self.is_save_state:
-            # recreate list removing saved networks
-            networks = [network for network in networks if network['id']
-                        not in self.saved_state_json['networks'].keys()]
+        if self.prefix:
+            networks = self._filter_by_prefix(networks)
+        else:
+            if not self.is_save_state:
+                # recreate list removing saved networks
+                networks = self._filter_out_ids_from_saved(
+                    networks, 'networks')
         # filter out networks declared in tempest.conf
         if self.is_preserve:
             networks = [network for network in networks
@@ -458,14 +500,17 @@ class NetworkService(BaseNetworkService):
 class NetworkFloatingIpService(BaseNetworkService):
 
     def list(self):
+        if self.prefix:
+            # this means we're cleaning resources based on a certain prefix,
+            # this resource doesn't have a name, therefore return empty list
+            return []
         client = self.floating_ips_client
         flips = client.list_floatingips(**self.tenant_filter)
         flips = flips['floatingips']
 
         if not self.is_save_state:
             # recreate list removing saved flips
-            flips = [flip for flip in flips if flip['id']
-                     not in self.saved_state_json['floatingips'].keys()]
+            flips = self._filter_out_ids_from_saved(flips, 'floatingips')
         LOG.debug("List count, %s Network Floating IPs", len(flips))
         return flips
 
@@ -498,15 +543,15 @@ class NetworkRouterService(BaseNetworkService):
         client = self.routers_client
         routers = client.list_routers(**self.tenant_filter)
         routers = routers['routers']
-
-        if not self.is_save_state:
-            # recreate list removing saved routers
-            routers = [router for router in routers if router['id']
-                       not in self.saved_state_json['routers'].keys()]
+        if self.prefix:
+            routers = self._filter_by_prefix(routers)
+        else:
+            if not self.is_save_state:
+                # recreate list removing saved routers
+                routers = self._filter_out_ids_from_saved(routers, 'routers')
         if self.is_preserve:
             routers = [router for router in routers
                        if router['id'] != CONF_PUB_ROUTER]
-
         LOG.debug("List count, %s Routers", len(routers))
         return routers
 
@@ -547,15 +592,19 @@ class NetworkRouterService(BaseNetworkService):
 class NetworkMeteringLabelRuleService(NetworkService):
 
     def list(self):
+        if self.prefix:
+            # this means we're cleaning resources based on a certain prefix,
+            # this resource doesn't have a name, therefore return empty list
+            return []
         client = self.metering_label_rules_client
         rules = client.list_metering_label_rules()
         rules = rules['metering_label_rules']
         rules = self._filter_by_tenant_id(rules)
 
         if not self.is_save_state:
-            saved_rules = self.saved_state_json['metering_label_rules'].keys()
+            rules = self._filter_out_ids_from_saved(
+                rules, 'metering_label_rules')
             # recreate list removing saved rules
-            rules = [rule for rule in rules if rule['id'] not in saved_rules]
         LOG.debug("List count, %s Metering Label Rules", len(rules))
         return rules
 
@@ -589,11 +638,12 @@ class NetworkMeteringLabelService(BaseNetworkService):
         labels = client.list_metering_labels()
         labels = labels['metering_labels']
         labels = self._filter_by_tenant_id(labels)
-
-        if not self.is_save_state:
+        if self.prefix:
+            labels = self._filter_by_prefix(labels)
+        elif not self.is_save_state:
             # recreate list removing saved labels
-            labels = [label for label in labels if label['id']
-                      not in self.saved_state_json['metering_labels'].keys()]
+            labels = self._filter_out_ids_from_saved(
+                labels, 'metering_labels')
         LOG.debug("List count, %s Metering Labels", len(labels))
         return labels
 
@@ -627,14 +677,14 @@ class NetworkPortService(BaseNetworkService):
                  client.list_ports(**self.tenant_filter)['ports']
                  if port["device_owner"] == "" or
                  port["device_owner"].startswith("compute:")]
-
-        if not self.is_save_state:
-            # recreate list removing saved ports
-            ports = [port for port in ports if port['id']
-                     not in self.saved_state_json['ports'].keys()]
+        if self.prefix:
+            ports = self._filter_by_prefix(ports)
+        else:
+            if not self.is_save_state:
+                # recreate list removing saved ports
+                ports = self._filter_out_ids_from_saved(ports, 'ports')
         if self.is_preserve:
             ports = self._filter_by_conf_networks(ports)
-
         LOG.debug("List count, %s Ports", len(ports))
         return ports
 
@@ -667,16 +717,18 @@ class NetworkSecGroupService(BaseNetworkService):
         secgroups = [secgroup for secgroup in
                      client.list_security_groups(**filter)['security_groups']
                      if secgroup['name'] != 'default']
-
-        if not self.is_save_state:
-            # recreate list removing saved security_groups
-            secgroups = [secgroup for secgroup in secgroups if secgroup['id']
-                         not in self.saved_state_json['security_groups'].keys()
-                         ]
+        if self.prefix:
+            secgroups = self._filter_by_prefix(secgroups)
+        else:
+            if not self.is_save_state:
+                # recreate list removing saved security_groups
+                secgroups = self._filter_out_ids_from_saved(
+                    secgroups, 'security_groups')
         if self.is_preserve:
-            secgroups = [secgroup for secgroup in secgroups
-                         if secgroup['security_group_rules'][0]['project_id']
-                         not in CONF_PROJECTS]
+            secgroups = [
+                secgroup for secgroup in secgroups
+                if secgroup['security_group_rules'][0]['project_id']
+                not in CONF_PROJECTS]
         LOG.debug("List count, %s security_groups", len(secgroups))
         return secgroups
 
@@ -708,10 +760,12 @@ class NetworkSubnetService(BaseNetworkService):
         client = self.subnets_client
         subnets = client.list_subnets(**self.tenant_filter)
         subnets = subnets['subnets']
-        if not self.is_save_state:
-            # recreate list removing saved subnets
-            subnets = [subnet for subnet in subnets if subnet['id']
-                       not in self.saved_state_json['subnets'].keys()]
+        if self.prefix:
+            subnets = self._filter_by_prefix(subnets)
+        else:
+            if not self.is_save_state:
+                # recreate list removing saved subnets
+                subnets = self._filter_out_ids_from_saved(subnets, 'subnets')
         if self.is_preserve:
             subnets = self._filter_by_conf_networks(subnets)
         LOG.debug("List count, %s Subnets", len(subnets))
@@ -743,10 +797,12 @@ class NetworkSubnetPoolsService(BaseNetworkService):
     def list(self):
         client = self.subnetpools_client
         pools = client.list_subnetpools(**self.tenant_filter)['subnetpools']
-        if not self.is_save_state:
-            # recreate list removing saved subnet pools
-            pools = [pool for pool in pools if pool['id']
-                     not in self.saved_state_json['subnetpools'].keys()]
+        if self.prefix:
+            pools = self._filter_by_prefix(pools)
+        else:
+            if not self.is_save_state:
+                # recreate list removing saved subnet pools
+                pools = self._filter_out_ids_from_saved(pools, 'subnetpools')
         if self.is_preserve:
             pools = [pool for pool in pools if pool['project_id']
                      not in CONF_PROJECTS]
@@ -782,11 +838,15 @@ class RegionService(BaseService):
         self.client = manager.regions_client
 
     def list(self):
+        if self.prefix:
+            # this means we're cleaning resources based on a certain prefix,
+            # this resource doesn't have a name, therefore return empty list
+            return []
         client = self.client
         regions = client.list_regions()
         if not self.is_save_state:
-            regions = [region for region in regions['regions'] if region['id']
-                       not in self.saved_state_json['regions'].keys()]
+            regions = self._filter_out_ids_from_saved(
+                regions['regions'], 'regions')
             LOG.debug("List count, %s Regions", len(regions))
             return regions
         else:
@@ -824,11 +884,12 @@ class FlavorService(BaseService):
     def list(self):
         client = self.client
         flavors = client.list_flavors({"is_public": None})['flavors']
-        if not self.is_save_state:
-            # recreate list removing saved flavors
-            flavors = [flavor for flavor in flavors if flavor['id']
-                       not in self.saved_state_json['flavors'].keys()]
-
+        if self.prefix:
+            flavors = self._filter_by_prefix(flavors)
+        else:
+            if not self.is_save_state:
+                # recreate list removing saved flavors
+                flavors = self._filter_out_ids_from_saved(flavors, 'flavors')
         if self.is_preserve:
             flavors = [flavor for flavor in flavors
                        if flavor['id'] not in CONF_FLAVORS]
@@ -871,10 +932,11 @@ class ImageService(BaseService):
             marker = urllib.parse_qs(parsed.query)['marker'][0]
             response = client.list_images(params={"marker": marker})
             images.extend(response['images'])
-
-        if not self.is_save_state:
-            images = [image for image in images if image['id']
-                      not in self.saved_state_json['images'].keys()]
+        if self.prefix:
+            images = self._filter_by_prefix(images)
+        else:
+            if not self.is_save_state:
+                images = self._filter_out_ids_from_saved(images, 'images')
         if self.is_preserve:
             images = [image for image in images
                       if image['id'] not in CONF_IMAGES]
@@ -910,19 +972,17 @@ class UserService(BaseService):
 
     def list(self):
         users = self.client.list_users()['users']
-
-        if not self.is_save_state:
-            users = [user for user in users if user['id']
-                     not in self.saved_state_json['users'].keys()]
-
+        if self.prefix:
+            users = self._filter_by_prefix(users)
+        else:
+            if not self.is_save_state:
+                users = self._filter_out_ids_from_saved(users, 'users')
         if self.is_preserve:
             users = [user for user in users if user['name']
                      not in CONF_USERS]
-
         elif not self.is_save_state:  # Never delete admin user
             users = [user for user in users if user['name'] !=
                      CONF.auth.admin_username]
-
         LOG.debug("List count, %s Users after reconcile", len(users))
         return users
 
@@ -955,13 +1015,14 @@ class RoleService(BaseService):
     def list(self):
         try:
             roles = self.client.list_roles()['roles']
-            # reconcile roles with saved state and never list admin role
-            if not self.is_save_state:
-                roles = [role for role in roles if
-                         (role['id'] not in
-                          self.saved_state_json['roles'].keys() and
-                          role['name'] != CONF.identity.admin_role)]
-                LOG.debug("List count, %s Roles after reconcile", len(roles))
+            if self.prefix:
+                roles = self._filter_by_prefix(roles)
+            elif not self.is_save_state:
+                # reconcile roles with saved state and never list admin role
+                roles = self._filter_out_ids_from_saved(roles, 'roles')
+                roles = [role for role in roles
+                         if role['name'] != CONF.identity.admin_role]
+            LOG.debug("List count, %s Roles after reconcile", len(roles))
             return roles
         except Exception:
             LOG.exception("Cannot retrieve Roles.")
@@ -995,18 +1056,17 @@ class ProjectService(BaseService):
 
     def list(self):
         projects = self.client.list_projects()['projects']
-        if not self.is_save_state:
-            project_ids = self.saved_state_json['projects']
-            projects = [project
-                        for project in projects
-                        if (project['id'] not in project_ids and
-                            project['name'] != CONF.auth.admin_project_name)]
-
+        if self.prefix:
+            projects = self._filter_by_prefix(projects)
+        else:
+            if not self.is_save_state:
+                projects = self._filter_out_ids_from_saved(
+                    projects, 'projects')
+                projects = [project for project in projects
+                            if project['name'] != CONF.auth.admin_project_name]
         if self.is_preserve:
-            projects = [project
-                        for project in projects
+            projects = [project for project in projects
                         if project['name'] not in CONF_PROJECTS]
-
         LOG.debug("List count, %s Projects after reconcile", len(projects))
         return projects
 
@@ -1039,10 +1099,10 @@ class DomainService(BaseService):
     def list(self):
         client = self.client
         domains = client.list_domains()['domains']
-        if not self.is_save_state:
-            domains = [domain for domain in domains if domain['id']
-                       not in self.saved_state_json['domains'].keys()]
-
+        if self.prefix:
+            domains = self._filter_by_prefix(domains)
+        elif not self.is_save_state:
+            domains = self._filter_out_ids_from_saved(domains, 'domains')
         LOG.debug("List count, %s Domains after reconcile", len(domains))
         return domains
 
diff --git a/tempest/tests/cmd/test_cleanup_services.py b/tempest/tests/cmd/test_cleanup_services.py
index 2301be6ed8..6b3b4b7a59 100644
--- a/tempest/tests/cmd/test_cleanup_services.py
+++ b/tempest/tests/cmd/test_cleanup_services.py
@@ -44,6 +44,7 @@ class TestBaseService(base.TestCase):
                   'saved_state_json': {'saved': 'data'},
                   'is_preserve': False,
                   'is_save_state': True,
+                  'prefix': 'tempest',
                   'tenant_id': 'project_id',
                   'got_exceptions': []}
         base = cleanup_service.BaseService(kwargs)
@@ -54,6 +55,7 @@ class TestBaseService(base.TestCase):
         self.assertTrue(base.is_save_state)
         self.assertEqual(base.tenant_filter['project_id'], kwargs['tenant_id'])
         self.assertEqual(base.got_exceptions, kwargs['got_exceptions'])
+        self.assertEqual(base.prefix, kwargs['prefix'])
 
     def test_not_implemented_ex(self):
         kwargs = {'data': {'data': 'test'},
@@ -61,6 +63,7 @@ class TestBaseService(base.TestCase):
                   'saved_state_json': {'saved': 'data'},
                   'is_preserve': False,
                   'is_save_state': False,
+                  'prefix': 'tempest',
                   'tenant_id': 'project_id',
                   'got_exceptions': []}
         base = self.TestException(kwargs)
@@ -188,7 +191,8 @@ class BaseCmdServiceTests(MockFunctionsBase):
     service_name = 'default'
 
     def _create_cmd_service(self, service_type, is_save_state=False,
-                            is_preserve=False, is_dry_run=False):
+                            is_preserve=False, is_dry_run=False,
+                            prefix=''):
         creds = fake_credentials.FakeKeystoneV3Credentials()
         os = clients.Manager(creds)
         return getattr(cleanup_service, service_type)(
@@ -196,6 +200,7 @@ class BaseCmdServiceTests(MockFunctionsBase):
             is_save_state=is_save_state,
             is_preserve=is_preserve,
             is_dry_run=is_dry_run,
+            prefix=prefix,
             project_id='b8e3ece07bb049138d224436756e3b57',
             data={},
             saved_state_json=self.saved_state
diff --git a/zuul.d/tempest-specific.yaml b/zuul.d/tempest-specific.yaml
index 10490b4aea..296682efa7 100644
--- a/zuul.d/tempest-specific.yaml
+++ b/zuul.d/tempest-specific.yaml
@@ -58,6 +58,8 @@
       Base integration test with Neutron networking, IPv6 and py3.
     vars:
       tox_envlist: full
+      run_tempest_cleanup: true
+      run_tempest_cleanup_prefix: true
       devstack_localrc:
         USE_PYTHON3: true
         FORCE_CONFIG_DRIVE: true