From dc01f5b33011675a919bd80cc4f021066d7f2fc4 Mon Sep 17 00:00:00 2001 From: Rodolfo Alonso Hernandez Date: Fri, 29 Jan 2021 16:42:40 +0000 Subject: [PATCH] Process DHCP events in order if related When processing port events (create, update, delete), the port provisioning (port creation) has priority over the other events [1]. As reported in the related bug, if a port deletion with an IP address and another port creation with the same IP address arrive to the DHCP agent, those events can be processed in the same queue. Because of the creation event priority, even when this event arrived after the deletion event, it will be processed first. That will clash with the DHCP agent cache, that contains a port (not deleted yet) with the same IP address. That will trigger an unwanted resync. This patch implements a specific logic to store the events in "ResourceProcessingQueue" (that uses "PriorityQueue" [2]). When a port event arrives, the event comparison method checks the (subnet, fixed_ips) tuple set of both elements. If there is a coincidence, that means those ports are the same or are using the same IP addreses (the race condition explained in the bug). In this case, the priority is defined only by the timestamp; that means the events are processed in order of arrival. Because the Neutron server do not allow to have two ports in the same subnet with the same IP address, the order of the events is guaranteed. In the case explained in the bug, the deletion event will be processed first. [1]https://review.opendev.org/c/openstack/neutron/+/626830 [2]https://docs.python.org/3/library/queue.html#queue.PriorityQueue Closes-Bug: #1913723 Change-Id: I89438feae3c0244f6da5e6a2a035d45b956ac247 --- neutron/agent/dhcp/agent.py | 101 +++++++++++------- .../rpc/agentnotifiers/dhcp_rpc_agent_api.py | 5 +- neutron/tests/unit/agent/dhcp/test_agent.py | 61 +++++++++++ .../agentnotifiers/test_dhcp_rpc_agent_api.py | 4 +- 4 files changed, 129 insertions(+), 42 deletions(-) diff --git a/neutron/agent/dhcp/agent.py b/neutron/agent/dhcp/agent.py index 3e52fb47426..8c11566369d 100644 --- a/neutron/agent/dhcp/agent.py +++ b/neutron/agent/dhcp/agent.py @@ -74,6 +74,35 @@ def _wait_if_syncing(f): return wrapped +class DHCPResourceUpdate(queue.ResourceUpdate): + + def __init__(self, _id, priority, action=None, resource=None, + timestamp=None, tries=5, obj_type=None): + super().__init__(_id, priority, action=action, resource=resource, + timestamp=timestamp, tries=tries) + self.obj_type = obj_type + + def __lt__(self, other): + if other.obj_type == self.obj_type == 'port': + # NOTE(ralonsoh): both resources should have "fixed_ips" + # information. That key was added to the deleted ports in this + # patch but this code runs in the Neutron API (server). Both the + # server and the DHCP agent should be updated. + # This check could be removed in Y release. + if ('fixed_ips' not in self.resource or + 'fixed_ips' not in other.resource): + return super().__lt__(other) + + self_ips = set(str(fixed_ip['ip_address']) for + fixed_ip in self.resource['fixed_ips']) + other_ips = set(str(fixed_ip['ip_address']) for + fixed_ip in other.resource['fixed_ips']) + if self_ips & other_ips: + return self.timestamp < other.timestamp + + return super().__lt__(other) + + class DhcpAgent(manager.Manager): """DHCP agent service manager. @@ -429,11 +458,10 @@ class DhcpAgent(manager.Manager): def network_create_end(self, context, payload): """Handle the network.create.end notification event.""" - update = queue.ResourceUpdate(payload['network']['id'], - payload.get('priority', - DEFAULT_PRIORITY), - action='_network_create', - resource=payload) + update = DHCPResourceUpdate(payload['network']['id'], + payload.get('priority', DEFAULT_PRIORITY), + action='_network_create', + resource=payload, obj_type='network') self._queue.add(update) @_wait_if_syncing @@ -444,11 +472,10 @@ class DhcpAgent(manager.Manager): def network_update_end(self, context, payload): """Handle the network.update.end notification event.""" - update = queue.ResourceUpdate(payload['network']['id'], - payload.get('priority', - DEFAULT_PRIORITY), - action='_network_update', - resource=payload) + update = DHCPResourceUpdate(payload['network']['id'], + payload.get('priority', DEFAULT_PRIORITY), + action='_network_update', + resource=payload, obj_type='network') self._queue.add(update) @_wait_if_syncing @@ -462,11 +489,10 @@ class DhcpAgent(manager.Manager): def network_delete_end(self, context, payload): """Handle the network.delete.end notification event.""" - update = queue.ResourceUpdate(payload['network_id'], - payload.get('priority', - DEFAULT_PRIORITY), - action='_network_delete', - resource=payload) + update = DHCPResourceUpdate(payload['network_id'], + payload.get('priority', DEFAULT_PRIORITY), + action='_network_delete', + resource=payload, obj_type='network') self._queue.add(update) @_wait_if_syncing @@ -477,11 +503,10 @@ class DhcpAgent(manager.Manager): def subnet_update_end(self, context, payload): """Handle the subnet.update.end notification event.""" - update = queue.ResourceUpdate(payload['subnet']['network_id'], - payload.get('priority', - DEFAULT_PRIORITY), - action='_subnet_update', - resource=payload) + update = DHCPResourceUpdate(payload['subnet']['network_id'], + payload.get('priority', DEFAULT_PRIORITY), + action='_subnet_update', + resource=payload, obj_type='subnet') self._queue.add(update) @_wait_if_syncing @@ -516,11 +541,10 @@ class DhcpAgent(manager.Manager): network_id = self._get_network_lock_id(payload) if not network_id: return - update = queue.ResourceUpdate(network_id, - payload.get('priority', - DEFAULT_PRIORITY), - action='_subnet_delete', - resource=payload) + update = DHCPResourceUpdate(network_id, + payload.get('priority', DEFAULT_PRIORITY), + action='_subnet_delete', + resource=payload, obj_type='subnet') self._queue.add(update) @_wait_if_syncing @@ -564,11 +588,10 @@ class DhcpAgent(manager.Manager): if self.cache.is_port_message_stale(updated_port): LOG.debug("Discarding stale port update: %s", updated_port) return - update = queue.ResourceUpdate(updated_port.network_id, - payload.get('priority', - DEFAULT_PRIORITY), - action='_port_update', - resource=updated_port) + update = DHCPResourceUpdate(updated_port.network_id, + payload.get('priority', DEFAULT_PRIORITY), + action='_port_update', + resource=updated_port, obj_type='port') self._queue.add(update) @_wait_if_syncing @@ -624,11 +647,10 @@ class DhcpAgent(manager.Manager): def port_create_end(self, context, payload): """Handle the port.create.end notification event.""" created_port = dhcp.DictModel(payload['port']) - update = queue.ResourceUpdate(created_port.network_id, - payload.get('priority', - DEFAULT_PRIORITY), - action='_port_create', - resource=created_port) + update = DHCPResourceUpdate(created_port.network_id, + payload.get('priority', DEFAULT_PRIORITY), + action='_port_create', + resource=created_port, obj_type='port') self._queue.add(update) @_wait_if_syncing @@ -667,11 +689,10 @@ class DhcpAgent(manager.Manager): network_id = self._get_network_lock_id(payload) if not network_id: return - update = queue.ResourceUpdate(network_id, - payload.get('priority', - DEFAULT_PRIORITY), - action='_port_delete', - resource=payload) + update = DHCPResourceUpdate(network_id, + payload.get('priority', DEFAULT_PRIORITY), + action='_port_delete', + resource=payload, obj_type='port') self._queue.add(update) @_wait_if_syncing diff --git a/neutron/api/rpc/agentnotifiers/dhcp_rpc_agent_api.py b/neutron/api/rpc/agentnotifiers/dhcp_rpc_agent_api.py index fb04cd4ee61..ced6a023bf8 100644 --- a/neutron/api/rpc/agentnotifiers/dhcp_rpc_agent_api.py +++ b/neutron/api/rpc/agentnotifiers/dhcp_rpc_agent_api.py @@ -268,7 +268,8 @@ class DhcpAgentNotifyAPI(object): def _after_router_interface_deleted(self, resource, event, trigger, **kwargs): self._notify_agents(kwargs['context'], 'port_delete_end', - {'port_id': kwargs['port']['id']}, + {'port_id': kwargs['port']['id'], + 'fixed_ips': kwargs['port']['fixed_ips']}, kwargs['port']['network_id']) def _native_event_send_dhcp_notification(self, resource, event, trigger, @@ -343,6 +344,8 @@ class DhcpAgentNotifyAPI(object): payload = {obj_type + '_id': obj_value['id']} if obj_type != 'network': payload['network_id'] = network_id + if obj_type == 'port': + payload['fixed_ips'] = obj_value['fixed_ips'] self._notify_agents(context, method_name, payload, network_id) else: self._notify_agents(context, method_name, data, network_id) diff --git a/neutron/tests/unit/agent/dhcp/test_agent.py b/neutron/tests/unit/agent/dhcp/test_agent.py index 465694f857d..b88f0619914 100644 --- a/neutron/tests/unit/agent/dhcp/test_agent.py +++ b/neutron/tests/unit/agent/dhcp/test_agent.py @@ -15,6 +15,7 @@ import collections import copy +import datetime import sys import uuid @@ -2305,3 +2306,63 @@ class TestDeviceManager(base.BaseTestCase): self.assertEqual(2, device.route.get_gateway.call_count) self.assertFalse(device.route.delete_gateway.called) device.route.add_gateway.assert_has_calls(expected) + + +class TestDHCPResourceUpdate(base.BaseTestCase): + + date1 = datetime.datetime(year=2021, month=2, day=1, hour=9, minute=1, + second=2) + date2 = datetime.datetime(year=2021, month=2, day=1, hour=9, minute=1, + second=1) # older than date1 + + def test__lt__no_port_event(self): + # Lower numerical priority always gets precedence. DHCPResourceUpdate + # (and ResourceUpdate) objects with more precedence will return as + # "lower" in a "__lt__" method comparison. + update1 = dhcp_agent.DHCPResourceUpdate('id1', 5, obj_type='network') + update2 = dhcp_agent.DHCPResourceUpdate('id2', 6, obj_type='network') + self.assertLess(update1, update2) + + def test__lt__no_port_event_timestamp(self): + update1 = dhcp_agent.DHCPResourceUpdate( + 'id1', 5, timestamp=self.date1, obj_type='network') + update2 = dhcp_agent.DHCPResourceUpdate( + 'id2', 6, timestamp=self.date2, obj_type='network') + self.assertLess(update1, update2) + + def test__lt__port_no_fixed_ips(self): + update1 = dhcp_agent.DHCPResourceUpdate( + 'id1', 5, timestamp=self.date1, resource={}, obj_type='port') + update2 = dhcp_agent.DHCPResourceUpdate( + 'id2', 6, timestamp=self.date2, resource={}, obj_type='port') + self.assertLess(update1, update2) + + def test__lt__port_fixed_ips_not_matching(self): + resource1 = {'fixed_ips': [ + {'subnet_id': 'subnet1', 'ip_address': '10.0.0.1'}]} + resource2 = {'fixed_ips': [ + {'subnet_id': 'subnet1', 'ip_address': '10.0.0.2'}, + {'subnet_id': 'subnet2', 'ip_address': '10.0.1.1'}]} + update1 = dhcp_agent.DHCPResourceUpdate( + 'id1', 5, timestamp=self.date1, resource=resource1, + obj_type='port') + update2 = dhcp_agent.DHCPResourceUpdate( + 'id2', 6, timestamp=self.date2, resource=resource2, + obj_type='port') + self.assertLess(update1, update2) + + def test__lt__port_fixed_ips_matching(self): + resource1 = {'fixed_ips': [ + {'subnet_id': 'subnet1', 'ip_address': '10.0.0.1'}]} + resource2 = {'fixed_ips': [ + {'subnet_id': 'subnet1', 'ip_address': '10.0.0.1'}, + {'subnet_id': 'subnet2', 'ip_address': '10.0.0.2'}]} + update1 = dhcp_agent.DHCPResourceUpdate( + 'id1', 5, timestamp=self.date1, resource=resource1, + obj_type='port') + update2 = dhcp_agent.DHCPResourceUpdate( + 'id2', 6, timestamp=self.date2, resource=resource2, + obj_type='port') + # In this case, both "port" events have matching IPs. "__lt__" method + # uses the timestamp: date2 < date1 + self.assertLess(update2, update1) diff --git a/neutron/tests/unit/api/rpc/agentnotifiers/test_dhcp_rpc_agent_api.py b/neutron/tests/unit/api/rpc/agentnotifiers/test_dhcp_rpc_agent_api.py index d47d6dbb7e3..de3bd75beac 100644 --- a/neutron/tests/unit/api/rpc/agentnotifiers/test_dhcp_rpc_agent_api.py +++ b/neutron/tests/unit/api/rpc/agentnotifiers/test_dhcp_rpc_agent_api.py @@ -247,7 +247,9 @@ class TestDhcpAgentNotifyAPI(base.BaseTestCase): self._test__notify_agents_with_function( lambda: self.notifier._after_router_interface_deleted( mock.ANY, mock.ANY, mock.ANY, context=mock.Mock(), - port={'id': 'foo_port_id', 'network_id': 'foo_network_id'}), + port={'id': 'foo_port_id', 'network_id': 'foo_network_id', + 'fixed_ips': {'subnet_id': 'subnet1', + 'ip_address': '10.0.0.1'}}), expected_scheduling=0, expected_casts=1) def test__fanout_message(self):