Add option to force delete zone-files in delete API.

Designate does not delete the zone-files on the back-end when zone is
deleted. This results in thousands leftover zone files on backend e.g.
bind. Add option in designate zone delete API to force clean/delete
zone-files on the back-end. This option is restricted for admin or
owner roles.

Closes-Bug: 1966517
Change-Id: Ic7b8fee4d4702b0632774d32542b23d7d2a8c253

27 files changed, 190 insertions, 33 deletions

diff --git a/api-ref/source/dns-api-v2-zone.inc b/api-ref/source/dns-api-v2-zone.inc
index 27b5d890..e99a095d 100644
--- a/api-ref/source/dns-api-v2-zone.inc
+++ b/api-ref/source/dns-api-v2-zone.inc
@@ -394,6 +394,7 @@ Request
    - x-auth-token: x-auth-token
    - x-auth-all-projects: x-auth-all-projects
    - x-auth-sudo-project-id: x-auth-sudo-project-id
+   - x-designate-hard-delete: x-designate-hard-delete
    - zone_id: path_zone_id
 
 
diff --git a/api-ref/source/parameters.yaml b/api-ref/source/parameters.yaml
index 421f376e..ef463af8 100644
--- a/api-ref/source/parameters.yaml
+++ b/api-ref/source/parameters.yaml
@@ -30,6 +30,14 @@ x-designate-edit-managed-records:
   required: false
   type: bool
 
+x-designate-hard-delete:
+  description: |
+    If enabled, this will delete the zone resources (i.e. files) on the
+    back-end.
+  in: header
+  required: false
+  type: bool
+
 x-openstack-request-id:
   description: |
     ID of the request
diff --git a/designate/api/middleware.py b/designate/api/middleware.py
index 4b9c7e8e..b888746e 100644
--- a/designate/api/middleware.py
+++ b/designate/api/middleware.py
@@ -89,6 +89,14 @@ class ContextMiddleware(base.Middleware):
             )
 
     @staticmethod
+    def _extract_hard_delete(ctxt, request):
+        ctxt.hard_delete = False
+        if request.headers.get('X-Designate-Hard-Delete'):
+            ctxt.hard_delete = strutils.bool_from_string(
+                request.headers.get('X-Designate-Hard-Delete')
+            )
+
+    @staticmethod
     def _extract_client_addr(ctxt, request):
         if hasattr(request, 'client_addr'):
             ctxt.client_addr = request.client_addr
@@ -103,6 +111,7 @@ class ContextMiddleware(base.Middleware):
             self._extract_sudo(ctxt, request)
             self._extract_all_projects(ctxt, request)
             self._extract_edit_managed_records(ctxt, request)
+            self._extract_hard_delete(ctxt, request)
             self._extract_dns_hide_counts(ctxt, request)
             self._extract_client_addr(ctxt, request)
         finally:
diff --git a/designate/backend/agent_backend/base.py b/designate/backend/agent_backend/base.py
index 78c36874..afe54160 100644
--- a/designate/backend/agent_backend/base.py
+++ b/designate/backend/agent_backend/base.py
@@ -48,5 +48,5 @@ class AgentBackend(DriverPlugin):
         """Zone is a DNSPython Zone object"""
 
     @abc.abstractmethod
-    def delete_zone(self, zone_name):
+    def delete_zone(self, zone_name, zone_params):
         """Delete a DNS zone"""
diff --git a/designate/backend/base.py b/designate/backend/base.py
index 1904474c..1190a699 100644
--- a/designate/backend/base.py
+++ b/designate/backend/base.py
@@ -71,7 +71,7 @@ class Backend(DriverPlugin):
         LOG.debug('Update Zone')
 
     @abc.abstractmethod
-    def delete_zone(self, context, zone):
+    def delete_zone(self, context, zone, zone_params):
         """
         Delete a DNS zone.
 
diff --git a/designate/backend/impl_akamai_v2.py b/designate/backend/impl_akamai_v2.py
index d124b297..2af55071 100644
--- a/designate/backend/impl_akamai_v2.py
+++ b/designate/backend/impl_akamai_v2.py
@@ -189,7 +189,7 @@ class AkamaiBackend(base.Backend):
             zone, self.masters, contract_id, gid, project_id, self.target)
         self.client.create_zone(payload)
 
-    def delete_zone(self, context, zone):
+    def delete_zone(self, context, zone, zone_params=None):
         """Delete a DNS zone"""
         LOG.debug('Delete Zone')
         self.client.delete_zone(zone['name'])
diff --git a/designate/backend/impl_bind9.py b/designate/backend/impl_bind9.py
index 1d484d72..4421875f 100644
--- a/designate/backend/impl_bind9.py
+++ b/designate/backend/impl_bind9.py
@@ -126,7 +126,7 @@ class Bind9Backend(base.Backend):
 
         return True
 
-    def delete_zone(self, context, zone):
+    def delete_zone(self, context, zone, zone_params=None):
         """Delete a new Zone by executin rndc
         Do not raise exceptions if the zone does not exist.
         """
@@ -138,7 +138,8 @@ class Bind9Backend(base.Backend):
             'delzone',
             '%s %s' % (zone['name'].rstrip('.'), view),
         ]
-        if self._clean_zonefile:
+        if (self._clean_zonefile or (zone_params and
+                zone_params.get('hard_delete'))):
             rndc_op.insert(1, '-clean')
 
         try:
diff --git a/designate/backend/impl_designate.py b/designate/backend/impl_designate.py
index 30c08ece..23d68c8c 100644
--- a/designate/backend/impl_designate.py
+++ b/designate/backend/impl_designate.py
@@ -80,7 +80,7 @@ class DesignateBackend(base.Backend):
         self.client.zones.create(
             zone.name, 'SECONDARY', masters=masters)
 
-    def delete_zone(self, context, zone):
+    def delete_zone(self, context, zone, zone_params=None):
         LOG.info('Deleting zone %(d_id)s / %(d_name)s',
                  {'d_id': zone['id'], 'd_name': zone['name']})
 
diff --git a/designate/backend/impl_dynect.py b/designate/backend/impl_dynect.py
index 186cac1c..19f586f5 100755
--- a/designate/backend/impl_dynect.py
+++ b/designate/backend/impl_dynect.py
@@ -356,7 +356,7 @@ class DynECTBackend(base.Backend):
         client.put(url, data={'activate': True})
         client.logout()
 
-    def delete_zone(self, context, zone):
+    def delete_zone(self, context, zone, zone_params=None):
         LOG.info('Deleting zone %(d_id)s / %(d_name)s',
                  {'d_id': zone['id'], 'd_name': zone['name']})
         url = '/Zone/%s' % zone['name'].rstrip('.')
diff --git a/designate/backend/impl_fake.py b/designate/backend/impl_fake.py
index e87a1b6a..2e9f1e07 100644
--- a/designate/backend/impl_fake.py
+++ b/designate/backend/impl_fake.py
@@ -27,5 +27,5 @@ class FakeBackend(base.Backend):
     def create_zone(self, context, zone):
         LOG.info('Create Zone %r', zone)
 
-    def delete_zone(self, context, zone):
+    def delete_zone(self, context, zone, zone_params=None):
         LOG.info('Delete Zone %r', zone)
diff --git a/designate/backend/impl_ns1.py b/designate/backend/impl_ns1.py
index a981f707..a12d971e 100644
--- a/designate/backend/impl_ns1.py
+++ b/designate/backend/impl_ns1.py
@@ -114,7 +114,7 @@ class NS1Backend(base.Backend):
             LOG.info("Can't create zone %s because it already exists",
                 zone.name)
 
-    def delete_zone(self, context, zone):
+    def delete_zone(self, context, zone, zone_params=None):
         """Delete a DNS zone"""
 
         # First verify that the zone exists
diff --git a/designate/backend/impl_nsd4.py b/designate/backend/impl_nsd4.py
index c90e51b5..c8babccb 100644
--- a/designate/backend/impl_nsd4.py
+++ b/designate/backend/impl_nsd4.py
@@ -94,7 +94,7 @@ class NSD4Backend(base.Backend):
             if "already exists" not in str(e):
                 raise
 
-    def delete_zone(self, context, zone):
+    def delete_zone(self, context, zone, zone_params=None):
         LOG.debug('Delete Zone')
         command = 'delzone %s' % zone['name']
 
diff --git a/designate/backend/impl_pdns4.py b/designate/backend/impl_pdns4.py
index 30172233..d320b0ae 100644
--- a/designate/backend/impl_pdns4.py
+++ b/designate/backend/impl_pdns4.py
@@ -125,7 +125,7 @@ class PDNS4Backend(base.Backend):
                     LOG.error('Could not delete errored zone %s', zone)
             raise exceptions.Backend(e)
 
-    def delete_zone(self, context, zone):
+    def delete_zone(self, context, zone, zone_params=None):
         """Delete a DNS zone"""
 
         # First verify that the zone exists -- If it's not present
diff --git a/designate/central/service.py b/designate/central/service.py
index 9ce54515..bf79420c 100644
--- a/designate/central/service.py
+++ b/designate/central/service.py
@@ -1047,7 +1047,11 @@ class Service(service.RPCService):
             zone = self.storage.delete_zone(context, zone.id)
         else:
             zone = self._delete_zone_in_storage(context, zone)
-            self.worker_api.delete_zone(context, zone)
+            delete_zonefile = False
+            if context.hard_delete:
+                delete_zonefile = True
+            self.worker_api.delete_zone(context, zone,
+                                        hard_delete=delete_zonefile)
 
         return zone
 
diff --git a/designate/common/config.py b/designate/common/config.py
index 8a3d7f21..08673367 100644
--- a/designate/common/config.py
+++ b/designate/common/config.py
@@ -27,6 +27,7 @@ def set_defaults():
                        'X-Auth-Sudo-Project-ID',
                        'X-Auth-All-Projects',
                        'X-Designate-Edit-Managed-Records',
+                       'X-Designate-Hard-Delete',
                        'OpenStack-DNS-Hide-Counts'],
         expose_headers=['X-OpenStack-Request-ID',
                         'Host'],
diff --git a/designate/common/policies/context.py b/designate/common/policies/context.py
index 81ab54d5..e5959d0e 100644
--- a/designate/common/policies/context.py
+++ b/designate/common/policies/context.py
@@ -43,6 +43,12 @@ deprecated_use_sudo = policy.DeprecatedRule(
     deprecated_reason=base.DEPRECATED_REASON,
     deprecated_since=versionutils.deprecated.WALLABY
 )
+deprecated_hard_delete = policy.DeprecatedRule(
+    name="hard_delete",
+    check_str=base.RULE_ADMIN,
+    deprecated_reason=base.DEPRECATED_REASON,
+    deprecated_since=versionutils.deprecated.WALLABY
+)
 
 rules = [
     policy.RuleDefault(
@@ -68,7 +74,13 @@ rules = [
         check_str=base.SYSTEM_ADMIN,
         scope_types=['system'],
         description='Accept sudo from user to tenant.',
-        deprecated_rule=deprecated_use_sudo)
+        deprecated_rule=deprecated_use_sudo),
+    policy.RuleDefault(
+        name="hard_delete",
+        check_str=base.SYSTEM_ADMIN,
+        scope_types=['system'],
+        description="Clean backend resources associated with zone",
+        deprecated_rule=deprecated_hard_delete),
 ]
 
 
diff --git a/designate/context.py b/designate/context.py
index 5e033446..2652b7d1 100644
--- a/designate/context.py
+++ b/designate/context.py
@@ -32,16 +32,19 @@ class DesignateContext(context.RequestContext):
     _abandon = None
     original_project_id = None
     _edit_managed_records = False
+    _hard_delete = False
     _client_addr = None
     FROM_DICT_EXTRA_KEYS = [
         'original_project_id', 'service_catalog', 'all_tenants', 'abandon',
         'edit_managed_records', 'tsigkey_id', 'hide_counts', 'client_addr',
+        'hard_delete'
     ]
 
     def __init__(self, service_catalog=None, all_tenants=False, abandon=None,
                  tsigkey_id=None, original_project_id=None,
                  edit_managed_records=False, hide_counts=False,
-                 client_addr=None, user_auth_plugin=None, **kwargs):
+                 client_addr=None, user_auth_plugin=None,
+                 hard_delete=False, **kwargs):
         super(DesignateContext, self).__init__(**kwargs)
 
         self.user_auth_plugin = user_auth_plugin
@@ -53,6 +56,7 @@ class DesignateContext(context.RequestContext):
         self.all_tenants = all_tenants
         self.abandon = abandon
         self.edit_managed_records = edit_managed_records
+        self.hard_delete = hard_delete
         self.hide_counts = hide_counts
         self.client_addr = client_addr
 
@@ -95,6 +99,7 @@ class DesignateContext(context.RequestContext):
             'all_tenants': self.all_tenants,
             'abandon': self.abandon,
             'edit_managed_records': self.edit_managed_records,
+            'hard_delete': self.hard_delete,
             'tsigkey_id': self.tsigkey_id,
             'hide_counts': self.hide_counts,
             'client_addr': self.client_addr,
@@ -103,7 +108,7 @@ class DesignateContext(context.RequestContext):
         return copy.deepcopy(d)
 
     def elevated(self, show_deleted=None, all_tenants=False,
-                 edit_managed_records=False):
+                 edit_managed_records=False, hard_delete=False):
         """Return a version of this context with admin flag set.
         Optionally set all_tenants and edit_managed_records
         """
@@ -124,6 +129,9 @@ class DesignateContext(context.RequestContext):
         if edit_managed_records:
             context.edit_managed_records = True
 
+        if hard_delete:
+            context.hard_delete = True
+
         return context
 
     def sudo(self, project_id):
@@ -183,6 +191,16 @@ class DesignateContext(context.RequestContext):
         self._edit_managed_records = value
 
     @property
+    def hard_delete(self):
+        return self._hard_delete
+
+    @hard_delete.setter
+    def hard_delete(self, value):
+        if value:
+            policy.check('hard_delete', self)
+        self._hard_delete = value
+
+    @property
     def client_addr(self):
         return self._client_addr
 
diff --git a/designate/tests/unit/api/test_middleware.py b/designate/tests/unit/api/test_middleware.py
index 67f28566..3bf7cfda 100644
--- a/designate/tests/unit/api/test_middleware.py
+++ b/designate/tests/unit/api/test_middleware.py
@@ -144,3 +144,22 @@ class KeystoneContextMiddlewareTest(oslotest.base.BaseTestCase):
 
         self.app(self.request)
         self.assertFalse(self.ctxt.edit_managed_records)
+
+    def test_hard_delete_in_headers(self):
+        self.request.headers.update({
+            'X-Tenant-ID': 'TenantID',
+            'X-Roles': 'admin',
+            'X-Designate-Hard-Delete': 'True'
+        })
+
+        self.app(self.request)
+        self.assertTrue(self.ctxt.hard_delete)
+
+    def test_hard_delete_not_set(self):
+        self.request.headers.update({
+            'X-Tenant-ID': 'TenantID',
+            'X-Roles': 'admin',
+        })
+
+        self.app(self.request)
+        self.assertFalse(self.ctxt.hard_delete)
diff --git a/designate/tests/unit/backend/test_bind9.py b/designate/tests/unit/backend/test_bind9.py
index b8c1a3e6..4887f45d 100644
--- a/designate/tests/unit/backend/test_bind9.py
+++ b/designate/tests/unit/backend/test_bind9.py
@@ -163,7 +163,7 @@ class Bind9BackendTestCase(oslotest.base.BaseTestCase):
             objects.PoolTarget.from_dict(self.target)
         )
 
-        backend.delete_zone(self.admin_context, self.zone)
+        backend.delete_zone(self.admin_context, self.zone, {})
 
         mock_execute.assert_called_with(
             ['delzone', 'example.com ']
diff --git a/designate/tests/unit/test_central/test_basic.py b/designate/tests/unit/test_central/test_basic.py
index e9e7a6b9..c371082c 100644
--- a/designate/tests/unit/test_central/test_basic.py
+++ b/designate/tests/unit/test_central/test_basic.py
@@ -264,6 +264,7 @@ class CentralBasic(TestCase):
             'sudo',
             'abandon',
             'all_tenants',
+            'hard_delete'
         ])
 
         self.service = Service()
@@ -922,6 +923,7 @@ class CentralZoneTestCase(CentralBasic):
 
     def test_delete_zone_has_subzone(self):
         self.context.abandon = False
+        self.context.hard_delete = False
         self.service.storage.get_zone.return_value = RoObject(
             name='foo',
             tenant_id='2',
@@ -961,6 +963,7 @@ class CentralZoneTestCase(CentralBasic):
 
     def test_delete_zone(self):
         self.context.abandon = False
+        self.context.hard_delete = False
         self.service.storage.get_zone.return_value = RoObject(
             name='foo',
             tenant_id='2',
@@ -983,6 +986,35 @@ class CentralZoneTestCase(CentralBasic):
         self.assertEqual('foo', out.name)
         pcheck, ctx, target = \
             designate.central.service.policy.check.call_args[0]
+
+        self.assertEqual('delete_zone', pcheck)
+
+    def test_delete_zone_hard_delete(self):
+        self.context.abandon = False
+        self.context.hard_delete = True
+        self.service.storage.get_zone.return_value = RoObject(
+            name='foo',
+            tenant_id='2',
+        )
+        self.service._delete_zone_in_storage = mock.Mock(
+            return_value=RoObject(
+                name='foo'
+            )
+        )
+        self.service.storage.count_zones.return_value = 0
+        out = self.service.delete_zone(self.context,
+                                       CentralZoneTestCase.zone__id)
+        self.assertFalse(self.service.storage.delete_zone.called)
+        self.assertTrue(self.service.worker_api.delete_zone.called)
+        self.assertTrue(designate.central.service.policy.check.called)
+        ctx, deleted_dom = \
+            self.service.worker_api.delete_zone.call_args[0]
+
+        self.assertEqual('foo', deleted_dom.name)
+        self.assertEqual('foo', out.name)
+        pcheck, ctx, target = \
+            designate.central.service.policy.check.call_args[0]
+
         self.assertEqual('delete_zone', pcheck)
 
     def test_delete_zone_in_storage(self):
diff --git a/designate/tests/unit/test_context.py b/designate/tests/unit/test_context.py
index 2a296c1f..62589cb5 100644
--- a/designate/tests/unit/test_context.py
+++ b/designate/tests/unit/test_context.py
@@ -96,6 +96,25 @@ class TestDesignateContext(designate.tests.TestCase):
         with testtools.ExpectedException(exceptions.Forbidden):
             ctxt.edit_managed_records = True
 
+    def test_hard_delete(self):
+        ctxt = context.DesignateContext(
+            user_id='12345', project_id='54321'
+        )
+        admin_ctxt = ctxt.elevated()
+
+        admin_ctxt.hard_delete = True
+
+        self.assertFalse(ctxt.is_admin)
+        self.assertTrue(admin_ctxt.is_admin)
+        self.assertTrue(admin_ctxt.hard_delete)
+
+    def test_hard_delete_failure(self):
+        ctxt = context.DesignateContext(
+            user_id='12345', project_id='54321'
+        )
+        with testtools.ExpectedException(exceptions.Forbidden):
+            ctxt.hard_delete = True
+
     @mock.patch.object(policy, 'check')
     def test_sudo(self, mock_policy_check):
         ctxt = context.DesignateContext(
diff --git a/designate/tests/unit/workers/test_service.py b/designate/tests/unit/workers/test_service.py
index a97361f8..1f4094bb 100644
--- a/designate/tests/unit/workers/test_service.py
+++ b/designate/tests/unit/workers/test_service.py
@@ -179,11 +179,12 @@ class TestService(oslotest.base.BaseTestCase):
 
     def test_delete_zone(self):
         self.service._do_zone_action = mock.Mock()
+        self.zone_params = {}
 
         self.service.delete_zone(self.context, self.zone)
 
         self.service._do_zone_action.assert_called_with(
-            self.context, self.zone
+            self.context, self.zone, self.zone_params
         )
 
     def test_update_zone(self):
@@ -204,15 +205,18 @@ class TestService(oslotest.base.BaseTestCase):
         pool.also_notifies = mock.MagicMock()
         pool.also_notifies.__iter__.return_value = []
         self.service.get_pool.return_value = pool
+        self.zone_params = {}
 
-        self.service._do_zone_action(self.context, self.zone)
+        self.service._do_zone_action(self.context, self.zone,
+                                     self.zone_params)
 
         mock_zone_action.assert_called_with(
             self.service.executor,
             self.context,
             pool,
             self.zone,
-            self.zone.action
+            self.zone.action,
+            self.zone_params
         )
 
         self.service._executor.run.assert_called_with([mock_zone_action()])
@@ -230,15 +234,18 @@ class TestService(oslotest.base.BaseTestCase):
             mock.Mock(host='192.168.1.1', port=53),
         ]
         self.service.get_pool.return_value = pool
+        self.zone_params = {}
 
-        self.service._do_zone_action(self.context, self.zone)
+        self.service._do_zone_action(self.context, self.zone,
+                                     self.zone_params)
 
         mock_zone_action.assert_called_with(
             self.service.executor,
             self.context,
             pool,
             self.zone,
-            self.zone.action
+            self.zone.action,
+            self.zone_params
         )
 
         self.service._executor.run.assert_called_with(
diff --git a/designate/tests/unit/workers/test_zone_tasks.py b/designate/tests/unit/workers/test_zone_tasks.py
index c8e5118b..29795b07 100644
--- a/designate/tests/unit/workers/test_zone_tasks.py
+++ b/designate/tests/unit/workers/test_zone_tasks.py
@@ -166,6 +166,7 @@ class TestZoneActionOnTarget(oslotest.base.BaseTestCase):
 
         self.context = mock.Mock()
         self.executor = mock.Mock()
+        self.zone_params = mock.Mock()
 
     @mock.patch.object(dnsutils, 'notify')
     def test_call_create(self, mock_notify):
@@ -175,6 +176,7 @@ class TestZoneActionOnTarget(oslotest.base.BaseTestCase):
             self.context,
             self.zone,
             self.target,
+            self.zone_params
         )
 
         self.assertTrue(self.actor())
@@ -193,6 +195,7 @@ class TestZoneActionOnTarget(oslotest.base.BaseTestCase):
             self.context,
             self.zone,
             self.target,
+            self.zone_params,
         )
 
         self.assertTrue(self.actor())
@@ -211,6 +214,7 @@ class TestZoneActionOnTarget(oslotest.base.BaseTestCase):
             self.context,
             self.zone,
             self.target,
+            self.zone_params
         )
 
         self.assertTrue(self.actor())
@@ -227,6 +231,7 @@ class TestZoneActionOnTarget(oslotest.base.BaseTestCase):
             self.context,
             self.zone,
             self.target,
+            self.zone_params
         )
 
         self.assertFalse(self.actor())
@@ -289,11 +294,13 @@ class TestZoneActor(oslotest.base.BaseTestCase):
         self.context = mock.Mock()
         self.pool = mock.Mock()
         self.executor = mock.Mock()
+        self.zone_params = mock.Mock()
         self.actor = zone.ZoneActor(
             self.executor,
             self.context,
             self.pool,
             mock.Mock(action='CREATE'),
+            self.zone_params
         )
 
     def test_threshold_from_config(self):
diff --git a/designate/worker/rpcapi.py b/designate/worker/rpcapi.py
index 6f7b4d06..c5c1bb00 100644
--- a/designate/worker/rpcapi.py
+++ b/designate/worker/rpcapi.py
@@ -68,9 +68,9 @@ class WorkerAPI(object):
         return self.client.cast(
             context, 'update_zone', zone=zone)
 
-    def delete_zone(self, context, zone):
+    def delete_zone(self, context, zone, hard_delete):
         return self.client.cast(
-            context, 'delete_zone', zone=zone)
+            context, 'delete_zone', zone=zone, hard_delete=hard_delete)
 
     def recover_shard(self, context, begin, end):
         return self.client.cast(
diff --git a/designate/worker/service.py b/designate/worker/service.py
index cf5a1a5d..c6f4be25 100644
--- a/designate/worker/service.py
+++ b/designate/worker/service.py
@@ -141,11 +141,11 @@ class Service(service.RPCService):
     def stop(self, graceful=True):
         super(Service, self).stop(graceful)
 
-    def _do_zone_action(self, context, zone):
+    def _do_zone_action(self, context, zone, zone_params=None):
         pool = self.get_pool(zone.pool_id)
         all_tasks = [
             zonetasks.ZoneAction(self.executor, context, pool, zone,
-                                 zone.action)
+                                 zone.action, zone_params)
         ]
 
         # Send a NOTIFY to each also-notifies
@@ -177,13 +177,17 @@ class Service(service.RPCService):
         self._do_zone_action(context, zone)
 
     @rpc.expected_exceptions()
-    def delete_zone(self, context, zone):
+    def delete_zone(self, context, zone, hard_delete=False):
         """
         :param context: Security context information.
         :param zone: Zone to be deleted
+        :param hard_delete: Zone resources (files) to be deleted or not
         :return: None
         """
-        self._do_zone_action(context, zone)
+        zone_params = {}
+        if hard_delete:
+            zone_params.update({'hard_delete': True})
+        self._do_zone_action(context, zone, zone_params)
 
     @rpc.expected_exceptions()
     def recover_shard(self, context, begin, end):
diff --git a/designate/worker/tasks/zone.py b/designate/worker/tasks/zone.py
index a724c542..0ee0b3a3 100644
--- a/designate/worker/tasks/zone.py
+++ b/designate/worker/tasks/zone.py
@@ -44,13 +44,14 @@ class ZoneActionOnTarget(base.Task):
     :return: Success/Failure of the target action (bool)
     """
 
-    def __init__(self, executor, context, zone, target):
+    def __init__(self, executor, context, zone, target, zone_params):
         super(ZoneActionOnTarget, self).__init__(executor)
         self.zone = zone
         self.action = zone.action
         self.target = target
         self.context = context
         self.task_name = 'ZoneActionOnTarget-%s' % self.action.title()
+        self.zone_params = zone_params
 
     def __call__(self):
         LOG.debug(
@@ -70,7 +71,8 @@ class ZoneActionOnTarget(base.Task):
                     self.target.backend.create_zone(self.context, self.zone)
                     SendNotify(self.executor, self.zone, self.target)()
                 elif self.action == 'DELETE':
-                    self.target.backend.delete_zone(self.context, self.zone)
+                    self.target.backend.delete_zone(self.context, self.zone,
+                                                    self.zone_params)
                 else:
                     self.target.backend.update_zone(self.context, self.zone)
                     SendNotify(self.executor, self.zone, self.target)()
@@ -195,15 +197,17 @@ class ZoneActor(base.Task):
              of targets (bool)
     """
 
-    def __init__(self, executor, context, pool, zone):
+    def __init__(self, executor, context, pool, zone, zone_params=None):
         super(ZoneActor, self).__init__(executor)
         self.context = context
         self.pool = pool
         self.zone = zone
+        self.zone_params = zone_params
 
     def _execute(self):
         results = self.executor.run([
-            ZoneActionOnTarget(self.executor, self.context, self.zone, target)
+            ZoneActionOnTarget(self.executor, self.context, self.zone, target,
+                               self.zone_params)
             for target in self.pool.targets
         ])
         return results
@@ -250,13 +254,15 @@ class ZoneAction(base.Task):
              number of nameservers (bool)
     """
 
-    def __init__(self, executor, context, pool, zone, action):
+    def __init__(self, executor, context, pool, zone, action,
+                 zone_params=None):
         super(ZoneAction, self).__init__(executor)
         self.context = context
         self.pool = pool
         self.zone = zone
         self.action = action
         self.task_name = 'ZoneAction-%s' % self.action.title()
+        self.zone_params = zone_params
 
     def _wait_for_nameservers(self):
         """
@@ -266,7 +272,7 @@ class ZoneAction(base.Task):
 
     def _zone_action_on_targets(self):
         actor = ZoneActor(
-            self.executor, self.context, self.pool, self.zone
+            self.executor, self.context, self.pool, self.zone, self.zone_params
         )
         return actor()
 
diff --git a/releasenotes/notes/Fix-Delete-back-end-zone-resources-upon-zone-deletion-da0051432c95c8e2.yaml b/releasenotes/notes/Fix-Delete-back-end-zone-resources-upon-zone-deletion-da0051432c95c8e2.yaml
new file mode 100644
index 00000000..6ae1f966
--- /dev/null
+++ b/releasenotes/notes/Fix-Delete-back-end-zone-resources-upon-zone-deletion-da0051432c95c8e2.yaml
@@ -0,0 +1,9 @@
+---
+fixes:
+  - |
+    Currently designate does not provide zone specific option to delete the
+    zone resources (such as files) on the back-end (e.g. bind9) when the zone
+    is deleted. To fix this add a header ``x-designate-hard-delete`` which will
+    be used in the zone delete API to delete zone files on the back-end. This
+    is in addition to the existing per-pool configration option that will
+    override this new header option. This option is restricted for admin.