Fix race condition in the sink when deleting records11.0.1

Updated the sink to behave closer to how we handle this type
of operations in designate.api.v2.

- Added object validation to all requests.
- Better test coverage.
- Use recordset update / delete instead of just record delete.

Closes-Bug: #1947765
Change-Id: I867600eb48a3e30a4d17471ab794ca717706823d
(cherry picked from commit 4807c23228fbb23af3087ee072bf73d7fc43aff5)

4 files changed, 336 insertions, 51 deletions

diff --git a/designate/notification_handler/base.py b/designate/notification_handler/base.py
index 07862b9b..a95c0145 100644
--- a/designate/notification_handler/base.py
+++ b/designate/notification_handler/base.py
@@ -78,6 +78,7 @@ class NotificationHandler(ExtensionPlugin):
             }
             recordset = RecordSet(**values)
             recordset.records = RecordList(objects=records)
+            recordset.validate()
             recordset = self.central_api.create_recordset(
                 context, zone_id, recordset
             )
@@ -90,12 +91,51 @@ class NotificationHandler(ExtensionPlugin):
             })
             for record in records:
                 recordset.records.append(record)
+            recordset.validate()
             recordset = self.central_api.update_recordset(
                 context, recordset
             )
         LOG.debug('Creating record in %s / %s', zone_id, recordset['id'])
         return recordset
 
+    def _update_or_delete_recordset(self, context, zone_id, recordset_id,
+                                    record_to_delete):
+        LOG.debug(
+            'Deleting record in %s / %s',
+            zone_id, record_to_delete['id']
+        )
+        try:
+            recordset = self.central_api.get_recordset(
+                context, zone_id, recordset_id
+            )
+
+            # Record not longer in recordset. Lets abort.
+            if record_to_delete not in recordset.records:
+                LOG.debug(
+                    'Record %s not found in recordset %s',
+                    record_to_delete['id'], recordset_id
+                )
+                return
+
+            # Remove the record from the recordset.
+            recordset.records.remove(record_to_delete)
+
+            if not recordset.records:
+                # Recordset is now empty. Remove it.
+                self.central_api.delete_recordset(
+                    context, zone_id, recordset_id
+                )
+                return
+
+            # Recordset still has records, validate it and update it.
+            recordset.validate()
+            self.central_api.update_recordset(context, recordset)
+        except exceptions.RecordSetNotFound:
+            LOG.info(
+                'Recordset %s for record %s was already removed',
+                recordset_id, record_to_delete['id']
+            )
+
 
 class BaseAddressHandler(NotificationHandler):
     default_formatv4 = ('%(hostname)s.%(domain)s',)
@@ -210,11 +250,7 @@ class BaseAddressHandler(NotificationHandler):
         })
 
         records = self.central_api.find_records(context, criterion)
-
         for record in records:
-            LOG.debug('Deleting record %s', record['id'])
-
-            self.central_api.delete_record(context,
-                                           zone_id,
-                                           record['recordset_id'],
-                                           record['id'])
+            self._update_or_delete_recordset(
+                context, zone_id, record['recordset_id'], record
+            )
diff --git a/designate/tests/resources/sample_notifications/nova/compute.instance.create.end-2.json b/designate/tests/resources/sample_notifications/nova/compute.instance.create.end-2.json
new file mode 100644
index 00000000..e8142c1b
--- /dev/null
+++ b/designate/tests/resources/sample_notifications/nova/compute.instance.create.end-2.json
@@ -0,0 +1,180 @@
+{
+    "_context_roles": ["Member", "admin"],
+    "_context_request_id": "req-14a1f71e-78a4-484f-94c5-5d6233533772",
+    "_context_quota_class": null,
+    "event_type": "compute.instance.create.end",
+    "_context_user_name": "user-name",
+    "_context_project_name": "tenant-name",
+    "_context_service_catalog": [{
+        "endpoints_links": [],
+        "endpoints": [{
+            "adminURL": "http://compute/v2/33a88272e06a49c1a0f653abc374b56b",
+            "region": "dub01",
+            "publicURL": "http://compute/v2/33a88272e06a49c1a0f653abc374b56b",
+            "internalURL": "http://compute/v2/33a88272e06a49c1a0f653abc374b56b",
+            "id": "9d4044f4601145eebb60fe0446d640ab"
+        }],
+        "type": "compute",
+        "name": "nova"
+    }, {
+        "endpoints_links": [],
+        "endpoints": [{
+            "adminURL": "http://s3",
+            "region": "dub01",
+            "publicURL": "http://s3",
+            "internalURL": "http://s3",
+            "id": "97e22c600c7141ac916a9be13d8ffbb8"
+        }],
+        "type": "s3",
+        "name": "s3"
+    }, {
+        "endpoints_links": [],
+        "endpoints": [{
+            "adminURL": "http://image:9292/v1",
+            "region": "dub01",
+            "publicURL": "http://image/v1",
+            "internalURL": "http://image:9292/v1",
+            "id": "c5b95f59ccb841e293dfdcafcaaf0a4a"
+        }],
+        "type": "image",
+        "name": "glance"
+    }, {
+        "endpoints_links": [],
+        "endpoints": [{
+            "adminURL": "http://volume/v1/33a88272e06a49c1a0f653abc374b56b",
+            "region": "dub01",
+            "publicURL": "http://volume/v1/33a88272e06a49c1a0f653abc374b56b",
+            "internalURL": "http://volume/v1/33a88272e06a49c1a0f653abc374b56b",
+            "id": "8e9b87c7e3a94697b8656ab845c74017"
+        }],
+        "type": "volume",
+        "name": "cinder"
+    }, {
+        "endpoints_links": [],
+        "endpoints": [{
+            "adminURL": "http://ec2/services/Admin",
+            "region": "dub01",
+            "publicURL": "http://ec2/services/Cloud",
+            "internalURL": "http://ec2/services/Cloud",
+            "id": "e465f3ed91534fe3a3a457f9ff90d839"
+        }],
+        "type": "ec2",
+        "name": "ec2"
+    }, {
+        "endpoints_links": [],
+        "endpoints": [{
+            "adminURL": "http://dns/v1.0",
+            "region": "dub01",
+            "publicURL": "http://dns/v1.0",
+            "internalURL": "http://dns/v1.0",
+            "id": "6dc68974176140c5b1eacd4329e94ae2"
+        }],
+        "type": "dns",
+        "name": "designate"
+    }, {
+        "endpoints_links": [],
+        "endpoints": [{
+            "adminURL": "http://identity:35357/v2.0",
+            "region": "dub01",
+            "publicURL": "http://identity/v2.0",
+            "internalURL": "http://identity/v2.0",
+            "id": "4034b8fb88df41e2abd7b3056fc908fd"
+        }],
+        "type": "identity",
+        "name": "keystone"
+    }],
+    "timestamp": "2012-11-03 17:54:48.797009",
+    "_context_is_admin": true,
+    "message_id": "1b378216-3f76-46db-9989-933172c1d4b2",
+    "_context_auth_token": "d7f4118a789f47b8ab708a00f239268f",
+    "_context_instance_lock_checked": false,
+    "_context_project_id": "33a88272e06a49c1a0f653abc374b56b",
+    "_context_timestamp": "2012-11-03T17:54:27.320555",
+    "_context_read_deleted": "no",
+    "_context_user_id": "953f8394fa044302b7d42f47228e427d",
+    "_context_remote_address": "127.0.0.1",
+    "publisher_id": "compute.stack01",
+    "payload": {
+        "state_description": "",
+        "availability_zone": null,
+        "ramdisk_id": "",
+        "instance_type_id": 2,
+        "deleted_at": "",
+        "fixed_ips": [{
+            "floating_ips": [],
+            "label": "private",
+            "version": 4,
+            "meta": {},
+            "address": "172.16.0.15",
+            "type": "fixed"
+        }],
+        "memory_mb": 512,
+        "user_id": "953f8394fa044302b7d42f47228e427d",
+        "reservation_id": "r-1ekblkfw",
+        "state": "active",
+        "launched_at": "2012-11-03 17:54:48.514631",
+        "metadata": [],
+        "ephemeral_gb": 0,
+        "access_ip_v6": null,
+        "disk_gb": 0,
+        "access_ip_v4": null,
+        "kernel_id": "",
+        "image_name": "ubuntu-precise",
+        "host": "stack01",
+        "display_name": "TestInstance",
+        "image_ref_url": "http://192.0.2.98:9292/images/e52f1321-fb9e-40fb-8057-555a850462e8",
+        "root_gb": 0,
+        "tenant_id": "33a88272e06a49c1a0f653abc374b56b",
+        "created_at": "2012-11-03 17:54:27",
+        "instance_id": "c71977ac-d2e3-479f-8549-3c56a2bfa24a",
+        "instance_type": "m1.tiny",
+        "vcpus": 1,
+        "image_meta": {
+            "base_image_ref": "e52f1321-fb9e-40fb-8057-555a850462e8"
+        },
+        "architecture": null,
+        "os_type": null
+    },
+    "payload_v6": {
+        "state_description": "",
+        "availability_zone": null,
+        "ramdisk_id": "",
+        "instance_type_id": 2,
+        "deleted_at": "",
+        "fixed_ips": [{
+            "floating_ips": [],
+            "label": "private",
+            "version": 6,
+            "meta": {},
+            "address": "172.16.0.15",
+            "type": "fixed"
+        }],
+        "memory_mb": 512,
+        "user_id": "953f8394fa044302b7d42f47228e427d",
+        "reservation_id": "r-1ekblkfw",
+        "state": "active",
+        "launched_at": "2012-11-03 17:54:48.514631",
+        "metadata": [],
+        "ephemeral_gb": 0,
+        "access_ip_v6": null,
+        "disk_gb": 0,
+        "access_ip_v4": null,
+        "kernel_id": "",
+        "image_name": "ubuntu-precise",
+        "host": "stack01",
+        "display_name": "TestInstance",
+        "image_ref_url": "http://192.0.2.98:9292/images/e52f1321-fb9e-40fb-8057-555a850462e8",
+        "root_gb": 0,
+        "tenant_id": "33a88272e06a49c1a0f653abc374b56b",
+        "created_at": "2012-11-03 17:54:27",
+        "instance_id": "c71977ac-d2e3-479f-8549-3c56a2bfa24a",
+        "instance_type": "m1.tiny",
+        "vcpus": 1,
+        "image_meta": {
+            "base_image_ref": "e52f1321-fb9e-40fb-8057-555a850462e8"
+        },
+        "architecture": null,
+        "os_type": null
+    },
+    "priority": "INFO"
+}
diff --git a/designate/tests/test_notification_handler/test_neutron.py b/designate/tests/test_notification_handler/test_neutron.py
index 652844ff..9373a869 100644
--- a/designate/tests/test_notification_handler/test_neutron.py
+++ b/designate/tests/test_notification_handler/test_neutron.py
@@ -54,11 +54,11 @@ class NeutronFloatingHandlerTest(TestCase, NotificationHandlerMixin):
         self.plugin.process_notification(
             self.admin_context.to_dict(), event_type, fixture['payload'])
 
-        # Ensure we now have exactly 1 record, plus SOA & NS
-        records = self.central_service.find_records(self.admin_context,
-                                                    criterion)
+        # Ensure we now have exactly 2 records, plus SOA & NS
+        recordsets = self.central_service.find_recordsets(self.admin_context,
+                                                         criterion)
 
-        self.assertEqual(4, len(records))
+        self.assertEqual(4, len(recordsets))
 
     def test_floatingip_disassociate(self):
         start_event_type = 'floatingip.update.end'
@@ -76,7 +76,7 @@ class NeutronFloatingHandlerTest(TestCase, NotificationHandlerMixin):
 
         criterion = {'zone_id': self.zone_id}
 
-        # Ensure we start with at least 1 record, plus NS and SOA
+        # Ensure we start with exactly 2 records, plus NS and SOA
         records = self.central_service.find_records(self.admin_context,
                                                     criterion)
 
@@ -85,17 +85,11 @@ class NeutronFloatingHandlerTest(TestCase, NotificationHandlerMixin):
         self.plugin.process_notification(
             self.admin_context.to_dict(), event_type, fixture['payload'])
 
-        # Simulate the record having been deleted on the backend
-        zone_serial = self.central_service.get_zone(
-            self.admin_context, self.zone_id).serial
-        self.central_service.update_status(
-            self.admin_context, self.zone_id, "SUCCESS", zone_serial)
+        # Ensure we now have exactly 0 recordsets, plus NS and SOA
+        recordsets = self.central_service.find_recordsets(self.admin_context,
+                                                          criterion)
 
-        # Ensure we now have exactly 0 records, plus NS and SOA
-        records = self.central_service.find_records(self.admin_context,
-                                                    criterion)
-
-        self.assertEqual(2, len(records))
+        self.assertEqual(2, len(recordsets))
 
     def test_floatingip_delete(self):
         start_event_type = 'floatingip.update.end'
@@ -113,7 +107,7 @@ class NeutronFloatingHandlerTest(TestCase, NotificationHandlerMixin):
 
         criterion = {'zone_id': self.zone_id}
 
-        # Ensure we start with at least 1 record, plus NS and SOA
+        # Ensure we start with exactly 2 records, plus NS and SOA
         records = self.central_service.find_records(self.admin_context,
                                                     criterion)
         self.assertEqual(4, len(records))
@@ -121,14 +115,8 @@ class NeutronFloatingHandlerTest(TestCase, NotificationHandlerMixin):
         self.plugin.process_notification(
             self.admin_context.to_dict(), event_type, fixture['payload'])
 
-        # Simulate the record having been deleted on the backend
-        zone_serial = self.central_service.get_zone(
-            self.admin_context, self.zone_id).serial
-        self.central_service.update_status(
-            self.admin_context, self.zone_id, "SUCCESS", zone_serial)
+        # Ensure we now have exactly 0 recordsets, plus NS and SOA
+        recordsets = self.central_service.find_recordsets(self.admin_context,
+                                                          criterion)
 
-        # Ensure we now have exactly 0 records, plus NS and SOA
-        records = self.central_service.find_records(self.admin_context,
-                                                    criterion)
-
-        self.assertEqual(2, len(records))
+        self.assertEqual(2, len(recordsets))
diff --git a/designate/tests/test_notification_handler/test_nova.py b/designate/tests/test_notification_handler/test_nova.py
index 141d7378..4d973987 100644
--- a/designate/tests/test_notification_handler/test_nova.py
+++ b/designate/tests/test_notification_handler/test_nova.py
@@ -47,22 +47,25 @@ class NovaFixedHandlerTest(TestCase, NotificationHandlerMixin):
 
         self.assertIn(event_type, self.plugin.get_event_types())
 
-        criterion = {'zone_id': self.zone_id}
+        criterion = {
+            'zone_id': self.zone_id,
+            'managed': True,
+            'managed_resource_type': 'instance',
+        }
 
-        # Ensure we start with 2 records
         records = self.central_service.find_records(self.admin_context,
                                                     criterion)
-        # Should only be SOA and NS records
-        self.assertEqual(2, len(records))
+        # Ensure we start with zero managed records
+        self.assertFalse(records)
 
         self.plugin.process_notification(
             self.admin_context.to_dict(), event_type, fixture['payload'])
 
-        # Ensure we now have exactly 1 more record
+        # Ensure we now have exactly 2 records.
         records = self.central_service.find_records(self.admin_context,
                                                     criterion)
 
-        self.assertEqual(4, len(records))
+        self.assertEqual(2, len(records))
 
     def test_instance_create_end_utf8(self):
         self.config(formatv4=['%(display_name)s.%(zone)s'],
@@ -77,13 +80,14 @@ class NovaFixedHandlerTest(TestCase, NotificationHandlerMixin):
 
         self.assertIn(event_type, self.plugin.get_event_types())
 
-        criterion = {'zone_id': self.zone_id}
+        criterion = {
+            'zone_id': self.zone_id,
+        }
 
-        # Ensure we start with 2 records
         recordsets = self.central_service.find_recordsets(
             self.admin_context, criterion)
 
-        # Should only be SOA and NS recordsets
+        # Ensure that we only have SOA and NS recordsets.
         self.assertEqual(2, len(recordsets))
 
         self.plugin.process_notification(
@@ -118,28 +122,105 @@ class NovaFixedHandlerTest(TestCase, NotificationHandlerMixin):
 
         self.assertIn(event_type, self.plugin.get_event_types())
 
-        criterion = {'zone_id': self.zone_id}
+        criterion = {
+            'zone_id': self.zone_id,
+            'managed': True,
+            'managed_resource_type': 'instance',
+        }
+
+        # Ensure we start with exactly 2 records.
+        records = self.central_service.find_records(self.admin_context,
+                                                    criterion)
+
+        self.assertEqual(2, len(records))
+
+        self.plugin.process_notification(
+            self.admin_context.to_dict(), event_type, fixture['payload'])
 
-        # Ensure we start with at least 1 record, plus NS and SOA
+        # Ensure we now have exactly zero active records
         records = self.central_service.find_records(self.admin_context,
                                                     criterion)
 
+        self.assertEqual(2, len(records), records)
+
+        # The two deleted records should now be in action state DELETE.
+        for record in records:
+            self.assertEqual('DELETE', record.action)
+            self.assertEqual('172.16.0.14', record.data)
+
+    def test_instance_delete_one_with_multiple_records_with_same_name(self):
+        # Prepare for the test
+        for start_event_type in ['compute.instance.create.end',
+                                 'compute.instance.create.end-2']:
+            start_fixture = self.get_notification_fixture(
+                'nova', start_event_type
+            )
+            self.plugin.process_notification(
+                self.admin_context.to_dict(),
+                start_fixture['event_type'],
+                start_fixture['payload']
+            )
+
+        # Now - Onto the real test
+        event_type = 'compute.instance.delete.start'
+        fixture = self.get_notification_fixture('nova', event_type)
+
+        self.assertIn(event_type, self.plugin.get_event_types())
+
+        criterion = {
+            'zone_id': self.zone_id,
+            'managed': True,
+            'managed_resource_type': 'instance',
+        }
+
+        records = self.central_service.find_records(self.admin_context,
+                                                    criterion)
+
+        # Ensure we start with exactly 4 records.
         self.assertEqual(4, len(records))
 
         self.plugin.process_notification(
             self.admin_context.to_dict(), event_type, fixture['payload'])
 
-        # Simulate the record having been deleted on the backend
-        zone_serial = self.central_service.get_zone(
-            self.admin_context, self.zone_id).serial
-        self.central_service.update_status(
-            self.admin_context, self.zone_id, "SUCCESS", zone_serial)
+        # Ensure we now have exactly 2 records
+        records = self.central_service.find_records(self.admin_context,
+                                                    criterion)
+
+        self.assertEqual(2, len(records), records)
+
+        # The two remaining records should be in waiting UPDATE.
+        for record in records:
+            self.assertEqual('UPDATE', record.action)
+            self.assertEqual('172.16.0.15', record.data)
+
+    def test_instance_delete_with_no_record(self):
+        event_type = 'compute.instance.delete.start'
+        fixture = self.get_notification_fixture('nova', event_type)
+
+        self.assertIn(event_type, self.plugin.get_event_types())
+
+        criterion = {
+            'zone_id': self.zone_id,
+            'managed': True,
+            'managed_resource_type': 'instance',
+        }
 
-        # Ensure we now have exactly 0 records, plus NS and SOA
         records = self.central_service.find_records(self.admin_context,
                                                     criterion)
 
-        self.assertEqual(2, len(records))
+        # Ensure we start with zero records
+        self.assertFalse(records)
+
+        # Make sure we don't fail here, even though there is nothing to
+        # do, since the record we are trying to delete does not actually exist.
+        self.plugin.process_notification(
+            self.admin_context.to_dict(), event_type, fixture['payload'])
+
+        # Ensure we still have zero records
+        records = self.central_service.find_records(self.admin_context,
+                                                    criterion)
+
+        self.assertFalse(records)
 
     def test_label_in_format_v4_v6(self):
         event_type = 'compute.instance.create.end'