diff --git a/designate/central/rpcapi.py b/designate/central/rpcapi.py index a82862e9e..6d442dba5 100644 --- a/designate/central/rpcapi.py +++ b/designate/central/rpcapi.py @@ -51,14 +51,15 @@ class CentralAPI(object): 5.2 - Add Zone Import methods 5.3 - Add Zone Export method 5.4 - Add asynchronous Zone Export methods + 5.5 - Add deleted zone purging task """ - RPC_API_VERSION = '5.4' + RPC_API_VERSION = '5.5' def __init__(self, topic=None): topic = topic if topic else cfg.CONF.central_topic target = messaging.Target(topic=topic, version=self.RPC_API_VERSION) - self.client = rpc.get_client(target, version_cap='5.4') + self.client = rpc.get_client(target, version_cap='5.5') @classmethod def get_instance(cls): @@ -177,6 +178,14 @@ class CentralAPI(object): LOG.info(_LI("delete_domain: Calling central's delete_domain.")) return self.client.call(context, 'delete_domain', domain_id=domain_id) + def purge_domains(self, context, criterion=None, limit=None): + LOG.info(_LI( + "purge_domains: Calling central's purge_domains." + )) + cctxt = self.client.prepare(version='5.5') + return cctxt.call(context, 'purge_domains', + criterion=criterion, limit=limit) + def count_domains(self, context, criterion=None): LOG.info(_LI("count_domains: Calling central's count_domains.")) return self.client.call(context, 'count_domains', criterion=criterion) @@ -512,7 +521,7 @@ class CentralAPI(object): request_body=request_body) def find_zone_imports(self, context, criterion=None, marker=None, - limit=None, sort_key=None, sort_dir=None): + limit=None, sort_key=None, sort_dir=None): LOG.info(_LI("find_zone_imports: Calling central's " "find_zone_imports.")) return self.client.call(context, 'find_zone_imports', diff --git a/designate/central/service.py b/designate/central/service.py index 6f99b18eb..9b83b9d93 100644 --- a/designate/central/service.py +++ b/designate/central/service.py @@ -260,7 +260,7 @@ def notification(notification_type): class Service(service.RPCService, service.Service): - RPC_API_VERSION = '5.4' + RPC_API_VERSION = '5.5' target = messaging.Target(version=RPC_API_VERSION) @@ -1010,6 +1010,12 @@ class Service(service.RPCService, service.Service): @notification('dns.domain.delete') @synchronized_domain() def delete_domain(self, context, domain_id): + """Delete or abandon a domain + On abandon, delete the domain from the DB immediately. + Otherwise, set action to DELETE and status to PENDING and poke + Pool Manager's "delete_domain" to update the resolvers. PM will then + poke back to set action to NONE and status to DELETED + """ domain = self.storage.get_domain(context, domain_id) target = { @@ -1041,6 +1047,9 @@ class Service(service.RPCService, service.Service): @transaction def _delete_domain_in_storage(self, context, domain): + """Set domain action to DELETE and status to PENDING + to have the domain soft-deleted later on + """ domain.action = 'DELETE' domain.status = 'PENDING' @@ -1049,6 +1058,19 @@ class Service(service.RPCService, service.Service): return domain + def purge_domains(self, context, criterion=None, limit=None): + """Purge deleted zones. + :returns: number of purged domains + """ + policy.check('purge_domains', context, criterion) + if not criterion: + raise exceptions.BadRequest("A criterion is required") + + LOG.debug("Performing purge with limit of %r and criterion of %r" + % (limit, criterion)) + + return self.storage.purge_domains(context, criterion, limit) + def xfr_domain(self, context, domain_id): domain = self.storage.get_domain(context, domain_id) diff --git a/designate/sqlalchemy/base.py b/designate/sqlalchemy/base.py index a891792ca..8605177dc 100644 --- a/designate/sqlalchemy/base.py +++ b/designate/sqlalchemy/base.py @@ -225,6 +225,7 @@ class SQLAlchemy(object): query = self._apply_criterion(table, query, criterion) if apply_tenant_criteria: query = self._apply_tenant_criteria(context, table, query) + query = self._apply_deleted_criteria(context, table, query) # Execute the Query @@ -491,11 +492,20 @@ class SQLAlchemy(object): return _set_object_from_model(obj, resultproxy.fetchone()) - def _delete(self, context, table, obj, exc_notfound): - if hasattr(table.c, 'deleted'): - # Perform a Soft Delete + def _delete(self, context, table, obj, exc_notfound, hard_delete=False): + """Perform item deletion or soft-delete. + """ + + if hasattr(table.c, 'deleted') and not hard_delete: + # Perform item soft-delete. + # Set the "status" column to "DELETED" and populate + # the "deleted_at" column + # TODO(kiall): If the object has any changed fields, they will be # persisted here when we don't want that. + + # "deleted" is populated with the object id (rather than being a + # boolean) to keep (name, deleted) unique obj.deleted = obj.id.replace('-', '') obj.deleted_at = timeutils.utcnow() diff --git a/designate/storage/base.py b/designate/storage/base.py index c8427f0ff..f8a67af6f 100644 --- a/designate/storage/base.py +++ b/designate/storage/base.py @@ -287,6 +287,15 @@ class Storage(DriverPlugin): :param domain_id: Domain ID to delete. """ + @abc.abstractmethod + def purge_domain(self, context, zone): + """ + Purge a Domain + + :param context: RPC Context. + :param domain: Zone to delete. + """ + @abc.abstractmethod def count_domains(self, context, criterion=None): """ @@ -385,7 +394,7 @@ class Storage(DriverPlugin): :param domain_id: Domain ID to create the record in. :param recordset_id: RecordSet ID to create the record in. :param record: Record object with the values to be created. - """ + """ @abc.abstractmethod def get_record(self, context, record_id): @@ -650,7 +659,7 @@ class Storage(DriverPlugin): @abc.abstractmethod def find_zone_imports(self, context, criterion=None, marker=None, - limit=None, sort_key=None, sort_dir=None): + limit=None, sort_key=None, sort_dir=None): """ Find Zone Imports diff --git a/designate/storage/impl_sqlalchemy/__init__.py b/designate/storage/impl_sqlalchemy/__init__.py index 3ae941fcf..6cf198547 100644 --- a/designate/storage/impl_sqlalchemy/__init__.py +++ b/designate/storage/impl_sqlalchemy/__init__.py @@ -24,6 +24,7 @@ from sqlalchemy.sql.expression import or_ from designate import exceptions from designate import objects +from designate.i18n import _LI from designate.sqlalchemy import base as sqlalchemy_base from designate.storage import base as storage_base from designate.storage.impl_sqlalchemy import tables @@ -31,6 +32,8 @@ from designate.storage.impl_sqlalchemy import tables LOG = logging.getLogger(__name__) +MAXIMUM_SUBDOMAIN_DEPTH = 128 + cfg.CONF.register_group(cfg.OptGroup( name='storage:sqlalchemy', title="Configuration for SQLAlchemy Storage" )) @@ -423,11 +426,72 @@ class SQLAlchemyStorage(sqlalchemy_base.SQLAlchemy, storage_base.Storage): return updated_domain def delete_domain(self, context, domain_id): + """ + """ # Fetch the existing domain, we'll need to return it. domain = self._find_domains(context, {'id': domain_id}, one=True) return self._delete(context, tables.domains, domain, exceptions.DomainNotFound) + def purge_domain(self, context, zone): + """Effectively remove a zone database record. + """ + return self._delete(context, tables.domains, zone, + exceptions.DomainNotFound, hard_delete=True) + + def _walk_up_domains(self, current, zones_by_id): + """Walk upwards in a zone hierarchy until we find a parent zone + that does not belong to "zones_by_id" + :returns: parent zone ID or None + """ + max_steps = MAXIMUM_SUBDOMAIN_DEPTH + while current.parent_domain_id in zones_by_id: + current = zones_by_id[current.parent_domain_id] + max_steps -= 1 + if max_steps == 0: + raise exceptions.IllegalParentDomain("Loop detected in the" + " domain hierarchy") + + return current.parent_domain_id + + def purge_domains(self, context, criterion, limit): + """Purge deleted zones. + Reparent orphan childrens, if any. + Transactions/locks are not needed. + :returns: number of purged domains + """ + if 'deleted' in criterion: + context.show_deleted = True + + zones = self.find_domains( + context=context, + criterion=criterion, + limit=limit, + ) + if not zones: + LOG.info(_LI("No zones to be purged")) + return + + LOG.debug(_LI("Purging %d zones"), len(zones)) + + zones_by_id = {z.id: z for z in zones} + + for zone in zones: + + # Reparent child zones, if any. + surviving_parent_id = self._walk_up_domains(zone, zones_by_id) + query = tables.domains.update().\ + where(tables.domains.c.parent_domain_id == zone.id).\ + values(parent_domain_id=surviving_parent_id) + + resultproxy = self.session.execute(query) + LOG.debug(_LI("%d child zones updated"), resultproxy.rowcount) + + self.purge_domain(context, zone) + + LOG.info(_LI("Purged %d zones"), len(zones)) + return len(zones) + def count_domains(self, context, criterion=None): query = select([func.count(tables.domains.c.id)]) query = self._apply_criterion(tables.domains, query, criterion) diff --git a/designate/tests/fixtures.py b/designate/tests/fixtures.py index 186cb61bb..37af5410a 100644 --- a/designate/tests/fixtures.py +++ b/designate/tests/fixtures.py @@ -15,6 +15,7 @@ # License for the specific language governing permissions and limitations # under the License. from __future__ import absolute_import + import shutil import tempfile @@ -130,3 +131,14 @@ class NetworkAPIFixture(fixtures.Fixture): self.api = network_api.get_network_api(cfg.CONF.network_api) self.fake = fake_network_api self.addCleanup(self.fake.reset_floatingips) + + +class ZoneManagerTaskFixture(fixtures.Fixture): + def __init__(self, task_cls): + super(ZoneManagerTaskFixture, self).__init__() + self._task_cls = task_cls + + def setUp(self): + super(ZoneManagerTaskFixture, self).setUp() + self.task = self._task_cls() + self.task.on_partition_change(range(0, 4095), None, None) diff --git a/designate/tests/test_central/test_service.py b/designate/tests/test_central/test_service.py index a7033688b..1215bbe62 100644 --- a/designate/tests/test_central/test_service.py +++ b/designate/tests/test_central/test_service.py @@ -14,8 +14,11 @@ # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # License for the specific language governing permissions and limitations # under the License. + +import datetime import copy import random +from collections import namedtuple import mock import testtools @@ -28,6 +31,7 @@ from oslo_messaging.notify import notifier from designate import exceptions from designate import objects from designate.tests.test_central import CentralTestCase +from designate.storage.impl_sqlalchemy import tables LOG = logging.getLogger(__name__) @@ -974,6 +978,284 @@ class CentralServiceTest(CentralTestCase): with testtools.ExpectedException(exceptions.Forbidden): self.central_service.count_domains(self.get_context()) + def _fetch_all_domains(self): + """Fetch all domains including deleted ones + """ + query = tables.domains.select() + return self.central_service.storage.session.execute(query).fetchall() + + def _log_all_domains(self, zones, msg=None): + """Log out a summary of zones + """ + if msg: + LOG.debug("--- %s ---" % msg) + cols = ('name', 'status', 'action', 'deleted', 'deleted_at', + 'parent_domain_id') + tpl = "%-35s | %-11s | %-11s | %-32s | %-20s | %s" + LOG.debug(tpl % cols) + for z in zones: + LOG.debug(tpl % tuple(z[k] for k in cols)) + + def _assert_count_all_domains(self, n): + """Assert count ALL domains including deleted ones + """ + zones = self._fetch_all_domains() + if len(zones) == n: + return + + msg = "failed: %d zones expected, %d found" % (n, len(zones)) + self._log_all_domains(zones, msg=msg) + raise Exception("Unexpected number of zones") + + def _create_deleted_domain(self, name, mock_deletion_time): + # Create a domain and set it as deleted + domain = self.create_domain(name=name) + self._delete_domain(domain, mock_deletion_time) + return domain + + def _delete_domain(self, domain, mock_deletion_time): + # Set a domain as deleted + zid = domain.id.replace('-', '') + query = tables.domains.update().\ + where(tables.domains.c.id == zid).\ + values( + action='NONE', + deleted=zid, + deleted_at=mock_deletion_time, + status='DELETED', + ) + + pxy = self.central_service.storage.session.execute(query) + self.assertEqual(pxy.rowcount, 1) + return domain + + @mock.patch.object(notifier.Notifier, "info") + def test_purge_domains_nothing_to_purge(self, mock_notifier): + # Create a zone + self.create_domain() + mock_notifier.reset_mock() + self._assert_count_all_domains(1) + + now = datetime.datetime(2015, 7, 31, 0, 0) + self.central_service.purge_domains( + self.admin_context, + limit=100, + criterion={ + 'status': 'DELETED', + 'deleted': '!0', + 'deleted_at': "<=%s" % now + }, + ) + self._assert_count_all_domains(1) + + @mock.patch.object(notifier.Notifier, "info") + def test_purge_domains_one_to_purge(self, mock_notifier): + self.create_domain() + new = datetime.datetime(2015, 7, 30, 0, 0) + now = datetime.datetime(2015, 7, 31, 0, 0) + self._create_deleted_domain('example2.org.', new) + mock_notifier.reset_mock() + self._assert_count_all_domains(2) + + self.central_service.purge_domains( + self.admin_context, + limit=100, + criterion={ + 'deleted': '!0', + 'deleted_at': "<=%s" % now + }, + ) + self._assert_count_all_domains(1) + + @mock.patch.object(notifier.Notifier, "info") + def test_purge_domains_one_to_purge_out_of_three(self, mock_notifier): + self.create_domain() + old = datetime.datetime(2015, 7, 20, 0, 0) + time_threshold = datetime.datetime(2015, 7, 25, 0, 0) + new = datetime.datetime(2015, 7, 30, 0, 0) + self._create_deleted_domain('old.org.', old) + self._create_deleted_domain('new.org.', new) + mock_notifier.reset_mock() + self._assert_count_all_domains(3) + + purge_cnt = self.central_service.purge_domains( + self.admin_context, + limit=100, + criterion={ + 'deleted': '!0', + 'deleted_at': "<=%s" % time_threshold + }, + ) + self._assert_count_all_domains(2) + self.assertEqual(purge_cnt, 1) + + @mock.patch.object(notifier.Notifier, "info") + def test_purge_domains_without_time_threshold(self, mock_notifier): + self.create_domain() + old = datetime.datetime(2015, 7, 20, 0, 0) + new = datetime.datetime(2015, 7, 30, 0, 0) + self._create_deleted_domain('old.org.', old) + self._create_deleted_domain('new.org.', new) + mock_notifier.reset_mock() + self._assert_count_all_domains(3) + + purge_cnt = self.central_service.purge_domains( + self.admin_context, + limit=100, + criterion={ + 'deleted': '!0', + }, + ) + self._assert_count_all_domains(1) + self.assertEqual(purge_cnt, 2) + + @mock.patch.object(notifier.Notifier, "info") + def test_purge_domains_without_deleted_criterion(self, mock_notifier): + self.create_domain() + old = datetime.datetime(2015, 7, 20, 0, 0) + time_threshold = datetime.datetime(2015, 7, 25, 0, 0) + new = datetime.datetime(2015, 7, 30, 0, 0) + self._create_deleted_domain('old.org.', old) + self._create_deleted_domain('new.org.', new) + mock_notifier.reset_mock() + self._assert_count_all_domains(3) + + # Nothing should be purged + purge_cnt = self.central_service.purge_domains( + self.admin_context, + limit=100, + criterion={ + 'deleted_at': "<=%s" % time_threshold + }, + ) + self._assert_count_all_domains(3) + self.assertEqual(purge_cnt, None) + + @mock.patch.object(notifier.Notifier, "info") + def test_purge_domains_by_name(self, mock_notifier): + self.create_domain() + + # The domain is purged (even if it was not deleted) + purge_cnt = self.central_service.purge_domains( + self.admin_context, + limit=100, + criterion={ + 'name': 'example.com.' + }, + ) + self._assert_count_all_domains(0) + self.assertEqual(purge_cnt, 1) + + @mock.patch.object(notifier.Notifier, "info") + def test_purge_domains_without_any_criterion(self, mock_notifier): + with testtools.ExpectedException(exceptions.BadRequest): + self.central_service.purge_domains( + self.admin_context, + limit=100, + criterion={}, + ) + + @mock.patch.object(notifier.Notifier, "info") + def test_purge_domains_with_sharding(self, mock_notifier): + old = datetime.datetime(2015, 7, 20, 0, 0) + time_threshold = datetime.datetime(2015, 7, 25, 0, 0) + domain = self._create_deleted_domain('old.org.', old) + mock_notifier.reset_mock() + + # purge domains in an empty shard + self.central_service.purge_domains( + self.admin_context, + limit=100, + criterion={ + 'deleted': '!0', + 'deleted_at': "<=%s" % time_threshold, + 'shard': 'BETWEEN 99998, 99999', + }, + ) + n_zones = self.central_service.count_domains(self.admin_context) + self.assertEqual(n_zones, 1) + + # purge domains in a shard that contains the domain created above + self.central_service.purge_domains( + self.admin_context, + limit=100, + criterion={ + 'deleted': '!0', + 'deleted_at': "<=%s" % time_threshold, + 'shard': 'BETWEEN 0, %d' % domain.shard, + }, + ) + n_zones = self.central_service.count_domains(self.admin_context) + self.assertEqual(n_zones, 0) + + def test_purge_domains_walk_up_domains(self): + Zone = namedtuple('Zone', 'id parent_domain_id') + zones = [Zone(x + 1, x) for x in range(1234, 1237)] + + zones_by_id = {z.id: z for z in zones} + sid = self.central_service.storage._walk_up_domains( + zones[0], zones_by_id) + self.assertEqual(sid, 1234) + + sid = self.central_service.storage._walk_up_domains( + zones[-1], zones_by_id) + self.assertEqual(sid, 1234) + + def test_purge_domains_walk_up_domains_loop(self): + Zone = namedtuple('Zone', 'id parent_domain_id') + zones = [Zone(2, 1), Zone(3, 2), Zone(1, 3)] + zones_by_id = {z.id: z for z in zones} + with testtools.ExpectedException(exceptions.IllegalParentDomain): + self.central_service.storage._walk_up_domains( + zones[0], zones_by_id) + + @mock.patch.object(notifier.Notifier, "info") + def test_purge_domains_with_orphans(self, mock_notifier): + old = datetime.datetime(2015, 7, 20, 0, 0) + time_threshold = datetime.datetime(2015, 7, 25, 0, 0) + + # Create a tree of alive and deleted [sub]domains + z1 = self.create_domain(name='alive.org.') + z2 = self.create_domain(name='deleted.alive.org.') + z3 = self.create_domain(name='del2.deleted.alive.org.') + z4 = self.create_domain(name='del3.del2.deleted.alive.org.') + z5 = self.create_domain(name='alive2.del3.del2.deleted.alive.org.') + + self._delete_domain(z2, old) + self._delete_domain(z3, old) + self._delete_domain(z4, old) + + self.assertEqual(z2['parent_domain_id'], z1.id) + self.assertEqual(z3['parent_domain_id'], z2.id) + self.assertEqual(z4['parent_domain_id'], z3.id) + self.assertEqual(z5['parent_domain_id'], z4.id) + + self._assert_count_all_domains(5) + mock_notifier.reset_mock() + + zones = self._fetch_all_domains() + self._log_all_domains(zones) + self.central_service.purge_domains( + self.admin_context, + limit=100, + criterion={ + 'deleted': '!0', + 'deleted_at': "<=%s" % time_threshold + }, + ) + self._assert_count_all_domains(2) + zones = self._fetch_all_domains() + self._log_all_domains(zones) + for z in zones: + if z.name == 'alive.org.': + self.assertEqual(z.parent_domain_id, None) + elif z.name == 'alive2.del3.del2.deleted.alive.org.': + # alive2.del2.deleted.alive.org is to be reparented under + # alive.org + self.assertEqual(z.parent_domain_id, z1.id) + else: + raise Exception("Unexpected zone %r" % z) + def test_touch_domain(self): # Create a domain expected_domain = self.create_domain() @@ -1315,7 +1597,7 @@ class CentralServiceTest(CentralTestCase): raise db_exception.DBDeadlock() with mock.patch.object(self.central_service.storage, 'commit', - side_effect=fail_once_then_pass): + side_effect=fail_once_then_pass): # Perform the update recordset = self.central_service.update_recordset( self.admin_context, recordset) diff --git a/designate/tests/test_zone_manager/test_tasks.py b/designate/tests/test_zone_manager/test_tasks.py new file mode 100644 index 000000000..4a0475e3f --- /dev/null +++ b/designate/tests/test_zone_manager/test_tasks.py @@ -0,0 +1,107 @@ +# Copyright 2015 Hewlett-Packard Development Company, L.P. +# +# Author: Endre Karlson +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + +import datetime + +from oslo_log import log as logging +from oslo_utils import timeutils + +from designate.zone_manager import tasks +from designate.tests import TestCase +from designate.storage.impl_sqlalchemy import tables +from designate.tests import fixtures + +LOG = logging.getLogger(__name__) + + +class TaskTest(TestCase): + def setUp(self): + super(TaskTest, self).setUp() + + def _enable_tasks(self, tasks): + self.config( + enabled_tasks=tasks, + group="service:zone_manager") + + +class DeletedDomainPurgeTest(TaskTest): + def setUp(self): + super(DeletedDomainPurgeTest, self).setUp() + + self.config( + interval=3600, + time_threshold=604800, + batch_size=100, + group="zone_manager_task:domain_purge" + ) + + self.purge_task_fixture = self.useFixture( + fixtures.ZoneManagerTaskFixture(tasks.DeletedDomainPurgeTask) + ) + + def _create_deleted_zone(self, name, mock_deletion_time): + # Create a domain and set it as deleted + domain = self.create_domain(name=name) + self._delete_domain(domain, mock_deletion_time) + return domain + + def _fetch_all_domains(self): + """Fetch all domains including deleted ones + """ + query = tables.domains.select() + return self.central_service.storage.session.execute(query).fetchall() + + def _delete_domain(self, domain, mock_deletion_time): + # Set a domain as deleted + zid = domain.id.replace('-', '') + query = tables.domains.update().\ + where(tables.domains.c.id == zid).\ + values( + action='NONE', + deleted=zid, + deleted_at=mock_deletion_time, + status='DELETED', + ) + + pxy = self.central_service.storage.session.execute(query) + self.assertEqual(pxy.rowcount, 1) + return domain + + def _create_deleted_zones(self): + # Create a number of deleted zones in the past days + zones = [] + now = timeutils.utcnow() + for age in range(18): + age *= (24 * 60 * 60) # seconds + delta = datetime.timedelta(seconds=age) + deletion_time = now - delta + name = "example%d.org." % len(zones) + z = self._create_deleted_zone(name, deletion_time) + zones.append(z) + + return zones + + def test_purge_zones(self): + """Create 18 zones, run zone_manager, check if 7 zones are remaining + """ + self.config(quota_domains=1000) + self._create_deleted_zones() + + self.purge_task_fixture.task() + + zones = self._fetch_all_domains() + LOG.info("Number of zones: %d", len(zones)) + self.assertEqual(len(zones), 7) diff --git a/designate/tests/unit/test_central/test_basic.py b/designate/tests/unit/test_central/test_basic.py index 1c4fdf0a0..8e89d6efb 100644 --- a/designate/tests/unit/test_central/test_basic.py +++ b/designate/tests/unit/test_central/test_basic.py @@ -21,9 +21,9 @@ from mock import patch from oslo_config import cfg from oslo_config import fixture as cfg_fixture from oslotest import base -from testtools import ExpectedException as raises # with raises(...): ... import fixtures import mock +import testtools from designate import exceptions from designate.central.service import Service @@ -104,13 +104,13 @@ class MockObjectTest(base.BaseTestCase): o = RoObject(a=1) self.assertEqual(o['a'], 1) self.assertEqual(o.a, 1) - with raises(NotImplementedError): + with testtools.ExpectedException(NotImplementedError): o.a = 2 - with raises(NotImplementedError): + with testtools.ExpectedException(NotImplementedError): o.new = 1 - with raises(NotImplementedError): + with testtools.ExpectedException(NotImplementedError): o['a'] = 2 - with raises(NotImplementedError): + with testtools.ExpectedException(NotImplementedError): o['new'] = 1 def test_rw(self): @@ -123,9 +123,9 @@ class MockObjectTest(base.BaseTestCase): o['a'] = 3 self.assertEqual(o.a, 3) self.assertEqual(o['a'], 3) - with raises(NotImplementedError): + with testtools.ExpectedException(NotImplementedError): o.new = 1 - with raises(NotImplementedError): + with testtools.ExpectedException(NotImplementedError): o['new'] = 1 @@ -325,7 +325,7 @@ class CentralServiceTestCase(CentralBasic): designate.central.service.policy.check = mock.Mock( side_effect=exceptions.Forbidden ) - with raises(exceptions.InvalidTTL): + with testtools.ExpectedException(exceptions.InvalidTTL): self.service._is_valid_ttl(self.context, 3) def test__update_soa_secondary(self): @@ -440,7 +440,7 @@ class CentralServiceTestCase(CentralBasic): # self.assertEqual(parent_domain, '') self.service.check_for_tlds = False - with raises(exceptions.Forbidden): + with testtools.ExpectedException(exceptions.Forbidden): self.service.create_domain(self.context, MockDomain()) # TODO(Federico) add more create_domain tests @@ -463,28 +463,28 @@ class CentralDomainTestCase(CentralBasic): def test__is_valid_domain_name_invalid(self): self.service._is_blacklisted_domain_name = Mock() - with raises(exceptions.InvalidDomainName): + with testtools.ExpectedException(exceptions.InvalidDomainName): self.service._is_valid_domain_name(self.context, 'example^org.') def test__is_valid_domain_name_invalid_2(self): self.service._is_blacklisted_domain_name = Mock() - with raises(exceptions.InvalidDomainName): + with testtools.ExpectedException(exceptions.InvalidDomainName): self.service._is_valid_domain_name(self.context, 'example.tld.') def test__is_valid_domain_name_invalid_same_as_tld(self): self.service._is_blacklisted_domain_name = Mock() - with raises(exceptions.InvalidDomainName): + with testtools.ExpectedException(exceptions.InvalidDomainName): self.service._is_valid_domain_name(self.context, 'com.com.') def test__is_valid_domain_name_invalid_tld(self): self.service._is_blacklisted_domain_name = Mock() - with raises(exceptions.InvalidDomainName): + with testtools.ExpectedException(exceptions.InvalidDomainName): self.service._is_valid_domain_name(self.context, 'tld.') def test__is_valid_domain_name_blacklisted(self): self.service._is_blacklisted_domain_name = Mock( side_effect=exceptions.InvalidDomainName) - with raises(exceptions.InvalidDomainName): + with testtools.ExpectedException(exceptions.InvalidDomainName): self.service._is_valid_domain_name(self.context, 'valid.com.') def test__is_blacklisted_domain_name(self): @@ -510,7 +510,7 @@ class CentralDomainTestCase(CentralBasic): def test__is_valid_recordset_name_no_dot(self): domain = RoObject(name='example.org.') - with raises(ValueError): + with testtools.ExpectedException(ValueError): self.service._is_valid_recordset_name(self.context, domain, 'foo.example.org') @@ -519,20 +519,21 @@ class CentralDomainTestCase(CentralBasic): designate.central.service.cfg.CONF['service:central'].\ max_recordset_name_len = 255 rs_name = 'a' * 255 + '.org.' - with raises(exceptions.InvalidRecordSetName) as e: + with testtools.ExpectedException(exceptions.InvalidRecordSetName) as e: self.service._is_valid_recordset_name(self.context, domain, rs_name) self.assertEqual(e.message, 'Name too long') def test__is_valid_recordset_name_wrong_domain(self): domain = RoObject(name='example.org.') - with raises(exceptions.InvalidRecordSetLocation): + with testtools.ExpectedException(exceptions.InvalidRecordSetLocation): self.service._is_valid_recordset_name(self.context, domain, 'foo.example.com.') def test_is_valid_recordset_placement_cname(self): domain = RoObject(name='example.org.') - with raises(exceptions.InvalidRecordSetLocation) as e: + with testtools.ExpectedException(exceptions.InvalidRecordSetLocation) \ + as e: self.service._is_valid_recordset_placement( self.context, domain, @@ -549,7 +550,8 @@ class CentralDomainTestCase(CentralBasic): self.service.storage.find_recordsets.return_value = [ RoObject(id='2') ] - with raises(exceptions.InvalidRecordSetLocation) as e: + with testtools.ExpectedException(exceptions.InvalidRecordSetLocation) \ + as e: self.service._is_valid_recordset_placement( self.context, domain, @@ -567,7 +569,8 @@ class CentralDomainTestCase(CentralBasic): RoObject(), RoObject() ] - with raises(exceptions.InvalidRecordSetLocation) as e: + with testtools.ExpectedException(exceptions.InvalidRecordSetLocation) \ + as e: self.service._is_valid_recordset_placement( self.context, domain, @@ -616,7 +619,7 @@ class CentralDomainTestCase(CentralBasic): self.service.storage.find_domains.return_value = [ RoObject(name='foo.example.org.') ] - with raises(exceptions.InvalidRecordSetLocation): + with testtools.ExpectedException(exceptions.InvalidRecordSetLocation): self.service._is_valid_recordset_placement_subdomain( self.context, domain, @@ -702,7 +705,7 @@ class CentralDomainTestCase(CentralBasic): ns_records=[] ) - with raises(exceptions.NoServersConfigured): + with testtools.ExpectedException(exceptions.NoServersConfigured): self.service.create_domain( self.context, RoObject(tenant_id='1', name='example.com.', ttl=60, @@ -793,7 +796,7 @@ class CentralDomainTestCase(CentralBasic): tenant_id='2', ) self.service.storage.count_domains.return_value = 2 - with raises(exceptions.DomainHasSubdomain): + with testtools.ExpectedException(exceptions.DomainHasSubdomain): self.service.delete_domain(self.context, '1') pcheck, ctx, target = \ @@ -876,7 +879,7 @@ class CentralDomainTestCase(CentralBasic): tenant_id='2', type='PRIMARY' ) - with raises(exceptions.BadRequest): + with testtools.ExpectedException(exceptions.BadRequest): self.service.xfr_domain(self.context, '1') def test_count_report(self): @@ -923,7 +926,7 @@ class CentralDomainTestCase(CentralBasic): self.service.count_domains = Mock(return_value=1) self.service.count_records = Mock(return_value=2) self.service.count_tenants = Mock(return_value=3) - with raises(exceptions.ReportNotFound): + with testtools.ExpectedException(exceptions.ReportNotFound): self.service.count_report( self.context, criterion='bogus' @@ -951,7 +954,7 @@ class CentralDomainTestCase(CentralBasic): self.service.storage.get_recordset.return_value = RoObject( domain_id='3' ) - with raises(exceptions.RecordSetNotFound): + with testtools.ExpectedException(exceptions.RecordSetNotFound): self.service.get_recordset( self.context, '1', @@ -1010,15 +1013,15 @@ class CentralDomainTestCase(CentralBasic): recordset.obj_get_original_value.return_value = '1' recordset.obj_get_changes.return_value = ['tenant_id', 'foo'] - with raises(exceptions.BadRequest): + with testtools.ExpectedException(exceptions.BadRequest): self.service.update_recordset(self.context, recordset) recordset.obj_get_changes.return_value = ['domain_id', 'foo'] - with raises(exceptions.BadRequest): + with testtools.ExpectedException(exceptions.BadRequest): self.service.update_recordset(self.context, recordset) recordset.obj_get_changes.return_value = ['type', 'foo'] - with raises(exceptions.BadRequest): + with testtools.ExpectedException(exceptions.BadRequest): self.service.update_recordset(self.context, recordset) def test_update_recordset_action_delete(self): @@ -1027,7 +1030,7 @@ class CentralDomainTestCase(CentralBasic): ) recordset = Mock() recordset.obj_get_changes.return_value = ['foo'] - with raises(exceptions.BadRequest): + with testtools.ExpectedException(exceptions.BadRequest): self.service.update_recordset(self.context, recordset) def test_update_recordset_action_fail_on_managed(self): @@ -1042,7 +1045,7 @@ class CentralDomainTestCase(CentralBasic): recordset.managed = True self.context = Mock() self.context.edit_managed_records = False - with raises(exceptions.BadRequest): + with testtools.ExpectedException(exceptions.BadRequest): self.service.update_recordset(self.context, recordset) def test_update_recordset(self): @@ -1169,7 +1172,7 @@ class CentralDomainTestCase(CentralBasic): ) self.context = Mock() self.context.edit_managed_records = False - with raises(exceptions.RecordSetNotFound): + with testtools.ExpectedException(exceptions.RecordSetNotFound): self.service.delete_recordset(self.context, 'd', 'r') def test_delete_recordset_action_delete(self): @@ -1187,7 +1190,7 @@ class CentralDomainTestCase(CentralBasic): ) self.context = Mock() self.context.edit_managed_records = False - with raises(exceptions.BadRequest): + with testtools.ExpectedException(exceptions.BadRequest): self.service.delete_recordset(self.context, 'd', 'r') def test_delete_recordset_managed(self): @@ -1205,7 +1208,7 @@ class CentralDomainTestCase(CentralBasic): ) self.context = Mock() self.context.edit_managed_records = False - with raises(exceptions.BadRequest): + with testtools.ExpectedException(exceptions.BadRequest): self.service.delete_recordset(self.context, 'd', 'r') def test_delete_recordset(self): @@ -1291,7 +1294,7 @@ class CentralDomainTestCase(CentralBasic): tenant_id='2', type='foo', ) - with raises(exceptions.BadRequest): + with testtools.ExpectedException(exceptions.BadRequest): self.service.create_record( self.context, 1, @@ -1360,7 +1363,7 @@ class CentralDomainTestCase(CentralBasic): self.service.storage.get_recordset.return_value = RoObject( domain_id=3 ) - with raises(exceptions.RecordNotFound): + with testtools.ExpectedException(exceptions.RecordNotFound): self.service.get_record(self.context, 1, 2, 3) def test_get_record_not_found_2(self): @@ -1379,7 +1382,7 @@ class CentralDomainTestCase(CentralBasic): domain_id=2, recordset_id=3 ) - with raises(exceptions.RecordNotFound): + with testtools.ExpectedException(exceptions.RecordNotFound): self.service.get_record(self.context, 1, 2, 3) def test_get_record(self): @@ -1425,15 +1428,15 @@ class CentralDomainTestCase(CentralBasic): record.obj_get_original_value.return_value = 1 record.obj_get_changes.return_value = ['tenant_id', 'foo'] - with raises(exceptions.BadRequest): + with testtools.ExpectedException(exceptions.BadRequest): self.service.update_record(self.context, record) record.obj_get_changes.return_value = ['domain_id', 'foo'] - with raises(exceptions.BadRequest): + with testtools.ExpectedException(exceptions.BadRequest): self.service.update_record(self.context, record) record.obj_get_changes.return_value = ['recordset_id', 'foo'] - with raises(exceptions.BadRequest): + with testtools.ExpectedException(exceptions.BadRequest): self.service.update_record(self.context, record) def test_update_record_action_delete(self): @@ -1441,7 +1444,7 @@ class CentralDomainTestCase(CentralBasic): action='DELETE', ) record = Mock() - with raises(exceptions.BadRequest): + with testtools.ExpectedException(exceptions.BadRequest): self.service.update_record(self.context, record) def test_update_record_action_fail_on_managed(self): @@ -1459,7 +1462,7 @@ class CentralDomainTestCase(CentralBasic): record.obj_get_changes.return_value = ['foo'] self.context = Mock() self.context.edit_managed_records = False - with raises(exceptions.BadRequest): + with testtools.ExpectedException(exceptions.BadRequest): self.service.update_record(self.context, record) def test_update_record(self): @@ -1518,7 +1521,7 @@ class CentralDomainTestCase(CentralBasic): self.service.storage.get_domain.return_value = RoObject( action='DELETE', ) - with raises(exceptions.BadRequest): + with testtools.ExpectedException(exceptions.BadRequest): self.service.delete_record(self.context, 1, 2, 3) def test_delete_record_not_found(self): @@ -1533,7 +1536,7 @@ class CentralDomainTestCase(CentralBasic): id=888, ) # domain.id != record.domain_id - with raises(exceptions.RecordNotFound): + with testtools.ExpectedException(exceptions.RecordNotFound): self.service.delete_record(self.context, 1, 2, 3) self.service.storage.get_record.return_value = RoObject( @@ -1542,7 +1545,7 @@ class CentralDomainTestCase(CentralBasic): recordset_id=7777, ) # recordset.id != record.recordset_id - with raises(exceptions.RecordNotFound): + with testtools.ExpectedException(exceptions.RecordNotFound): self.service.delete_record(self.context, 1, 2, 3) def test_delete_record(self): @@ -1607,7 +1610,7 @@ class CentralDomainTestCase(CentralBasic): self.context.edit_managed_records = False with fx_pool_manager: - with raises(exceptions.BadRequest): + with testtools.ExpectedException(exceptions.BadRequest): self.service.delete_record(self.context, 1, 2, 3) def test__delete_record_in_storage(self): diff --git a/designate/tests/unit/test_objects/test_base.py b/designate/tests/unit/test_objects/test_base.py index 765fdea5f..ba80b1958 100644 --- a/designate/tests/unit/test_objects/test_base.py +++ b/designate/tests/unit/test_objects/test_base.py @@ -19,7 +19,6 @@ import copy import unittest from oslo_log import log as logging -from testtools import ExpectedException as raises # with raises(...): ... import mock from oslo_serialization import jsonutils import oslotest.base @@ -188,7 +187,7 @@ class DesignateObjectTest(oslotest.base.BaseTestCase): self.assertEqual(set(['id', 'nested_list']), obj.obj_what_changed()) def test_from_list(self): - with raises(NotImplementedError): + with testtools.ExpectedException(NotImplementedError): TestObject.from_list([]) def test_get_schema(self): @@ -212,7 +211,7 @@ class DesignateObjectTest(oslotest.base.BaseTestCase): schema = obj._obj_validator.schema self.assertEqual(schema, expected) - with raises(AttributeError): # bug + with testtools.ExpectedException(AttributeError): # bug schema = obj.obj_get_schema() @unittest.expectedFailure # bug @@ -354,7 +353,7 @@ class DesignateObjectTest(oslotest.base.BaseTestCase): def test_update_unexpected_attribute(self): obj = TestObject(id='MyID', name='test') - with raises(AttributeError): + with testtools.ExpectedException(AttributeError): obj.update({'id': 'new_id', 'new_key': 3}) def test_is_valid(self): @@ -609,7 +608,7 @@ class DictObjectMixinTest(oslotest.base.BaseTestCase): def test_get_missing(self): obj = TestObjectDict(name=1) self.assertFalse(obj.obj_attr_is_set('foo')) - with raises(AttributeError): + with testtools.ExpectedException(AttributeError): obj.get('foo') def test_get_default(self): diff --git a/designate/tests/unit/test_objects/test_domain.py b/designate/tests/unit/test_objects/test_domain.py index 723201e94..94cc70815 100644 --- a/designate/tests/unit/test_objects/test_domain.py +++ b/designate/tests/unit/test_objects/test_domain.py @@ -17,8 +17,8 @@ import unittest from oslo_log import log as logging -from testtools import ExpectedException as raises # with raises(...): ... import oslotest.base +import testtools from designate import exceptions from designate import objects @@ -41,7 +41,7 @@ class DomainTest(oslotest.base.BaseTestCase): def test_masters_none(self): domain = objects.Domain() - with raises(exceptions.RelationNotLoaded): + with testtools.ExpectedException(exceptions.RelationNotLoaded): self.assertEqual(domain.masters, None) def test_masters(self): @@ -87,5 +87,5 @@ class DomainTest(oslotest.base.BaseTestCase): domain = objects.Domain( type='SECONDARY', ) - with raises(exceptions.InvalidObject): + with testtools.ExpectedException(exceptions.InvalidObject): domain.validate() diff --git a/designate/tests/unit/test_objects/test_recordset.py b/designate/tests/unit/test_objects/test_recordset.py index c1bf7d938..58ba537ac 100644 --- a/designate/tests/unit/test_objects/test_recordset.py +++ b/designate/tests/unit/test_objects/test_recordset.py @@ -18,9 +18,9 @@ import itertools import unittest from oslo_log import log as logging -from testtools import ExpectedException as raises # with raises(...): ... import mock import oslotest.base +import testtools from designate import exceptions from designate import objects @@ -186,9 +186,9 @@ class RecordSetTest(oslotest.base.BaseTestCase): def test_validate_handle_exception(self): rs = create_test_recordset() - with mock.patch('designate.objects.DesignateObject.obj_cls_from_name') \ - as patched: + fn_name = 'designate.objects.DesignateObject.obj_cls_from_name' + with mock.patch(fn_name) as patched: patched.side_effect = KeyError - with raises(exceptions.InvalidObject): + with testtools.ExpectedException(exceptions.InvalidObject): # TODO(Federico): check the attributes of the exception rs.validate() diff --git a/designate/zone_manager/tasks.py b/designate/zone_manager/tasks.py index 61cd08e71..23cc921bc 100644 --- a/designate/zone_manager/tasks.py +++ b/designate/zone_manager/tasks.py @@ -29,6 +29,8 @@ LOG = logging.getLogger(__name__) class PeriodicTask(plugin.ExtensionPlugin): + """Abstract Zone Manager periodic task + """ __plugin_ns__ = 'designate.zone_manager_tasks' __plugin_type__ = 'zone_manager_task' __interval__ = None @@ -40,7 +42,11 @@ class PeriodicTask(plugin.ExtensionPlugin): @classmethod def get_base_opts(cls): options = [ - cfg.IntOpt('interval', default=cls.__interval__), + cfg.IntOpt( + 'interval', + default=cls.__interval__, + help='Run interval in seconds' + ), cfg.IntOpt('per_page', default=100), ] return options @@ -50,12 +56,18 @@ class PeriodicTask(plugin.ExtensionPlugin): return rpcapi.CentralAPI.get_instance() def on_partition_change(self, my_partitions, members, event): + """Refresh partitions attribute + """ self.my_partitions = my_partitions def _my_range(self): + """Returns first and last partitions + """ return self.my_partitions[0], self.my_partitions[-1] def _filter_between(self, col): + """Generate BETWEEN filter based on _my_range + """ return {col: "BETWEEN %s,%s" % self._my_range()} def _iter(self, method, *args, **kwargs): @@ -116,6 +128,62 @@ class PeriodicExistsTask(PeriodicTask): for zone in self._iter_zones(ctxt): zone_data = dict(zone) zone_data.update(data) - self.notifier.info(ctxt, 'dns.domain.exists', zone_data) + LOG.info(_LI("Finished emitting events.")) + + +class DeletedDomainPurgeTask(PeriodicTask): + """Purge deleted domains that are exceeding the grace period time interval. + Deleted domains have values in the deleted_at column. + Purging means removing them from the database entirely. + """ + + __plugin_name__ = 'domain_purge' + __interval__ = 3600 + + def __init__(self): + super(DeletedDomainPurgeTask, self).__init__() + + @classmethod + def get_cfg_opts(cls): + group = cfg.OptGroup(cls.get_canonical_name()) + options = cls.get_base_opts() + [ + cfg.IntOpt( + 'time_threshold', + default=604800, + help="How old deleted domains should be (deleted_at) to be " + "purged, in seconds" + ), + cfg.IntOpt( + 'batch_size', + default=100, + help='How many domains to be purged on each run' + ), + ] + return [(group, options)] + + def __call__(self): + """Call the Central API to perform a purge of deleted zones based on + expiration time and sharding range. + """ + pstart, pend = self._my_range() + msg = _LI("Performing deleted domain purging for %(start)s to %(end)s") + LOG.info(msg % {"start": pstart, "end": pend}) + + delta = datetime.timedelta(seconds=self.options.time_threshold) + time_threshold = timeutils.utcnow() - delta + LOG.debug("Filtering deleted domains before %s", time_threshold) + + criterion = self._filter_between('shard') + criterion['deleted'] = '!0' + criterion['deleted_at'] = "<=%s" % time_threshold + + ctxt = context.DesignateContext.get_admin_context() + ctxt.all_tenants = True + + self.central_api.purge_domains( + ctxt, + criterion=criterion, + limit=self.options.batch_size, + ) diff --git a/etc/designate/designate.conf.sample b/etc/designate/designate.conf.sample index 388bbb9f1..07b128b83 100644 --- a/etc/designate/designate.conf.sample +++ b/etc/designate/designate.conf.sample @@ -219,6 +219,19 @@ debug = False # Whether to allow synchronous zone exports #export_synchronous = True +#------------------------ +# Deleted domains purging +#------------------------ +[zone_manager_task:domain_purge] +# How frequently to purge deleted domains, in seconds +#interval = 3600 # 1h + +# How many records to be deleted on each run +#batch_size = 100 + +# How old deleted records should be (deleted_at) to be purged, in seconds +#time_threshold = 604800 # 7 days + #----------------------- # Pool Manager Service #----------------------- @@ -311,6 +324,7 @@ debug = False #masters = 192.168.27.100:5354 #type = bind9 + ############## ## Network API ############## diff --git a/etc/designate/policy.json b/etc/designate/policy.json index 03daf8a32..9f101d572 100644 --- a/etc/designate/policy.json +++ b/etc/designate/policy.json @@ -51,6 +51,7 @@ "xfr_domain": "rule:admin_or_owner", "abandon_domain": "rule:admin", "count_domains": "rule:admin_or_owner", + "purge_domains": "rule:admin", "touch_domain": "rule:admin_or_owner", "create_recordset": "rule:domain_primary_or_admin", diff --git a/setup.cfg b/setup.cfg index a107d4cc0..37498767c 100644 --- a/setup.cfg +++ b/setup.cfg @@ -112,6 +112,7 @@ designate.manage = designate.zone_manager_tasks = periodic_exists = designate.zone_manager.tasks:PeriodicExistsTask + domain_purge = designate.zone_manager.tasks:DeletedDomainPurgeTask [build_sphinx] all_files = 1