Remove duplicate rows in MySQL query output

Previously we would expect duplicates rows from MySQL and then filter
them with _unique_floatingip_iterator. This means that we were
converting rows to ORM-mapped objects unnecessarily. This conversion is
very CPU intensive.

Instead, we can remove the duplicates as a part of the query which means
that only unique rows are returned and the number of conversions from
rows to ORM-mapped objects is reduced significantly. It also means that
the _unique_floatingip_iterator function is no longer needed.

Change-Id: I05136302f7b8abc0a985b91c993008595234ad6b
Signed-off-by: Adam Oswick <[email protected]>
(cherry picked from commit 8946684)
(cherry picked from commit f96691b)

Author: adamoswick

neutron/objects/router.py

@@ -10,8 +10,6 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
-import itertools
-
 import netaddr
 
 from neutron_lib.api.definitions import availability_zone as az_def
@@ -409,20 +407,11 @@ def get_scoped_floating_ips(cls, context, router_ids):
 
         # Filter out on router_ids
         query = query.filter(l3.FloatingIP.router_id.in_(router_ids))
-        return cls._unique_floatingip_iterator(context, query)
 
-    @classmethod
-    def _unique_floatingip_iterator(cls, context, query):
-        """Iterates over only one row per floating ip. Ignores others."""
-        # Group rows by fip id. They must be sorted by same.
-        q = query.order_by(l3.FloatingIP.id)
-        keyfunc = lambda row: row[0]['id']
-        group_iterator = itertools.groupby(q, keyfunc)
-
-        # Just hit the first row of each group
-        for key, value in group_iterator:
-            # pylint: disable=stop-iteration-return
-            row = list(next(value))
+        # Remove duplicate rows based on FIP IDs
+        query = query.group_by(l3.FloatingIP.id)
+
+        for row in query:
             yield (cls._load_object(context, row[0]), row[1])
 
     @classmethod

neutron/tests/unit/db/test_l3_db.py

@@ -252,27 +252,6 @@ def test__make_floatingip_dict_with_scope(self, make_fip_dict):
             'fixed_ip_address_scope': mock.sentinel.address_scope_id,
             'id': mock.sentinel.fip_ip}, result)
 
-    def test__unique_floatingip_iterator(self):
-        context = mock.MagicMock()
-        query = mock.MagicMock()
-        query.order_by().__iter__.return_value = [
-            ({'id': 'id1'}, 'scope1'),
-            ({'id': 'id1'}, 'scope1'),
-            ({'id': 'id2'}, 'scope2'),
-            ({'id': 'id2'}, 'scope2'),
-            ({'id': 'id2'}, 'scope2'),
-            ({'id': 'id3'}, 'scope3')]
-        query.reset_mock()
-        with mock.patch.object(
-                l3_obj.FloatingIP, '_load_object',
-                side_effect=({'id': 'id1'}, {'id': 'id2'}, {'id': 'id3'})):
-            result = list(
-                l3_obj.FloatingIP._unique_floatingip_iterator(context, query))
-            query.order_by.assert_called_once_with(l3_models.FloatingIP.id)
-            self.assertEqual([({'id': 'id1'}, 'scope1'),
-                              ({'id': 'id2'}, 'scope2'),
-                              ({'id': 'id3'}, 'scope3')], result)
-
     @mock.patch.object(directory, 'get_plugin')
     def test_prevent_l3_port_deletion_port_not_found(self, gp):
         # port not found doesn't prevent

neutron/tests/unit/objects/test_router.py

@@ -12,8 +12,11 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+from itertools import chain
 from unittest import mock
 
+import netaddr
+
 from neutron_lib.db import api as db_api
 from oslo_utils import uuidutils
 
@@ -293,6 +296,51 @@ def test_v1_2_to_v1_1_drops_qos_network_policy_id(self):
         self.assertNotIn('qos_network_policy_id',
                          obj_v1_1['versioned_object.data'])
 
+    def test_get_scoped_floating_ips(self):
+        def compare_results(router_ids, original_fips):
+            self.assertCountEqual(
+                original_fips,
+                [
+                    fip[0].id
+                    for fip in router.FloatingIP.get_scoped_floating_ips(
+                        self.context, router_ids)
+                ]
+            )
+
+        # Setup three routers, networks and external networks
+        routers = {}
+        for i in range(3):
+            router_id = self._create_test_router_id(name=f'router-{i}')
+            routers[router_id] = []
+            net_id = self._create_test_network_id()
+            fip_net_id = self._create_external_network_id()
+
+            # Create three subnets and three FIPs using the
+            # aforementioned networks and routers
+            for j in range(3):
+                self._create_test_subnet_id(net_id)
+                fip = router.FloatingIP(
+                    self.context,
+                    floating_ip_address=netaddr.IPAddress(f'10.{i}.{j}.3'),
+                    floating_network_id=fip_net_id,
+                    floating_port_id=self._create_test_port_id(
+                        network_id=fip_net_id),
+                    fixed_port_id=self._create_test_port_id(
+                        network_id=net_id),
+                    router_id=router_id,
+                )
+                fip.create()
+                routers[router_id].append(fip.id)
+
+        # For each router we created, fetch the fips and ensure the
+        # results match what we originally created
+        for router_id, original_fips in routers.items():
+            compare_results([router_id], original_fips)
+
+        # Now try to fetch all the fips for all the routers at once
+        original_fips = list(chain.from_iterable(routers.values()))
+        compare_results(routers.keys(), original_fips)
+
 
 class DvrFipGatewayPortAgentBindingTestCase(
         obj_test_base.BaseObjectIfaceTestCase):