Merge "Add oslo db retry decorator to the RPC handlers"

neutron/api/rpc/handlers/dhcp_rpc.py

@@ -26,6 +26,7 @@ from neutron.api.v2 import attributes
 from neutron.common import constants
 from neutron.common import exceptions as n_exc
 from neutron.common import utils
+from neutron.db import api as db_api
 from neutron.extensions import portbindings
 from neutron.i18n import _LW
 from neutron import manager
@@ -157,6 +158,7 @@ class DhcpRpcCallback(object):
         network['ports'] = plugin.get_ports(context, filters=filters)
         return network
 
+    @db_api.retry_db_errors
     def release_dhcp_port(self, context, **kwargs):
         """Release the port currently being used by a DHCP agent."""
         host = kwargs.get('host')
@@ -169,6 +171,7 @@ class DhcpRpcCallback(object):
         plugin = manager.NeutronManager.get_plugin()
         plugin.delete_ports_by_device_id(context, device_id, network_id)
 
+    @db_api.retry_db_errors
     def release_port_fixed_ip(self, context, **kwargs):
         """Release the fixed_ip associated the subnet on a port."""
         host = kwargs.get('host')
@@ -203,6 +206,7 @@ class DhcpRpcCallback(object):
         LOG.warning(_LW('Updating lease expiration is now deprecated. Issued  '
                         'from host %s.'), host)
 
+    @db_api.retry_db_errors
     @resource_registry.mark_resources_dirty
     def create_dhcp_port(self, context, **kwargs):
         """Create and return dhcp port information.
@@ -224,6 +228,7 @@ class DhcpRpcCallback(object):
         plugin = manager.NeutronManager.get_plugin()
         return self._port_action(plugin, context, port, 'create_port')
 
+    @db_api.retry_db_errors
     def update_dhcp_port(self, context, **kwargs):
         """Update the dhcp port."""
         host = kwargs.get('host')

neutron/api/rpc/handlers/l3_rpc.py

@@ -23,6 +23,7 @@ from neutron.common import constants
 from neutron.common import exceptions
 from neutron.common import utils
 from neutron import context as neutron_context
+from neutron.db import api as db_api
 from neutron.extensions import l3
 from neutron.extensions import portbindings
 from neutron.i18n import _LE
@@ -59,6 +60,7 @@ class L3RpcCallback(object):
                 plugin_constants.L3_ROUTER_NAT]
         return self._l3plugin
 
+    @db_api.retry_db_errors
     def sync_routers(self, context, **kwargs):
         """Sync routers according to filters to a specific agent.
 
@@ -234,6 +236,7 @@ class L3RpcCallback(object):
         filters = {'fixed_ips': {'subnet_id': [subnet_id]}}
         return self.plugin.get_ports(context, filters=filters)
 
+    @db_api.retry_db_errors
     def get_agent_gateway_port(self, context, **kwargs):
         """Get Agent Gateway port for FIP.
 

neutron/plugins/ml2/plugin.py

@@ -1276,8 +1276,6 @@ class Ml2Plugin(db_base_plugin_v2.NeutronDbPluginV2,
                 raise e.errors[0].error
             raise exc.ServicePortInUse(port_id=port_id, reason=e)
 
-    @oslo_db_api.wrap_db_retry(max_retries=db_api.MAX_RETRIES,
-                               retry_on_deadlock=True)
     def delete_port(self, context, id, l3_port_check=True):
         self._pre_delete_port(context, id, l3_port_check)
         # TODO(armax): get rid of the l3 dependency in the with block

neutron/tests/unit/plugins/ml2/test_plugin.py

@@ -617,6 +617,20 @@ class TestMl2PortsV2(test_plugin.TestPortsV2, Ml2PluginV2TestCase):
             # by the called method
             self.assertIsNone(l3plugin.disassociate_floatingips(ctx, port_id))
 
+    def test_create_port_tolerates_db_deadlock(self):
+        ctx = context.get_admin_context()
+        with self.network() as net:
+            with self.subnet(network=net) as subnet:
+                segments = ml2_db.get_network_segments(ctx.session,
+                                                       net['network']['id'])
+                with mock.patch('neutron.plugins.ml2.plugin.'
+                                'db.get_network_segments') as get_seg_mock:
+                    get_seg_mock.side_effect = [db_exc.DBDeadlock, segments,
+                                                segments, segments]
+                    with self.port(subnet=subnet) as port:
+                        self.assertTrue(port['port']['id'])
+                        self.assertEqual(4, get_seg_mock.call_count)
+
     def test_delete_port_tolerates_db_deadlock(self):
         ctx = context.get_admin_context()
         plugin = manager.NeutronManager.get_plugin()
@@ -627,7 +641,9 @@ class TestMl2PortsV2(test_plugin.TestPortsV2, Ml2PluginV2TestCase):
                             'db.get_locked_port_and_binding') as lock:
                 lock.side_effect = [db_exc.DBDeadlock,
                                     (port_db, binding)]
-                plugin.delete_port(ctx, port['port']['id'])
+                req = self.new_delete_request('ports', port['port']['id'])
+                res = req.get_response(self.api)
+                self.assertEqual(204, res.status_int)
                 self.assertEqual(2, lock.call_count)
                 self.assertRaises(
                     exc.PortNotFound, plugin.get_port, ctx, port['port']['id'])