Allow multiple IPv6 ports on router from same network ml2/ovs+dvr

On a recent change [1], some additional checks was added to avoid
ports overlapping cidrs on a router. On this change was also added
a check to do not attach more than one port IPv6 from same network,
but this check need to allow multiple ports when a deployment is
done using ml2/ovs+xvlan+dvr and the router has an external gateway
configured, because two ports are added:

- one with device_owner as network:router_interface_distributed
- another one with device_owner as network:router_centralized_snat

Also an small improvement is done for just run this check over the
existing ports on the router for the same network_id.

[1] https://review.opendev.org/c/openstack/neutron/+/859143

Closes-Bug: #2002800
Change-Id: I765a7b41e7e84f42a3180dfd15e3a41a8e085284
(cherry picked from commit e209715)

Author: fernandoroyosanchez

neutron/db/l3_db.py

@@ -843,8 +843,11 @@ def _validate_one_router_ipv6_port_per_network(self, router, port):
         if self._port_has_ipv6_address(port):
             for existing_port in (rp.port for rp in router.attached_ports):
                 if (existing_port["id"] != port["id"] and
-                    existing_port["network_id"] == port["network_id"] and
-                        self._port_has_ipv6_address(existing_port)):
+                        existing_port["network_id"] == port["network_id"] and
+                        self._port_has_ipv6_address(existing_port) and
+                        port["device_owner"] not in [
+                            constants.DEVICE_OWNER_ROUTER_SNAT,
+                            constants.DEVICE_OWNER_DVR_INTERFACE]):
                     msg = _("Router already contains IPv6 port %(p)s "
                         "belonging to network id %(nid)s. Only one IPv6 port "
                         "from the same network subnet can be connected to a "
@@ -971,7 +974,7 @@ def add_router_interface(self, context, router_id, interface_info=None):
                                                  port=port,
                                                  interface_info=interface_info)
                 self._add_router_port(
-                    context, port['id'], router, device_owner)
+                    context, port, router, device_owner)
 
         gw_ips = []
         gw_network_id = None
@@ -999,10 +1002,10 @@ def add_router_interface(self, context, router_id, interface_info=None):
             subnets[-1]['id'], [subnet['id'] for subnet in subnets])
 
     @db_api.retry_if_session_inactive()
-    def _add_router_port(self, context, port_id, router, device_owner):
+    def _add_router_port(self, context, port, router, device_owner):
         l3_obj.RouterPort(
             context,
-            port_id=port_id,
+            port_id=port['id'],
             router_id=router.id,
             port_type=device_owner
         ).create()
@@ -1019,15 +1022,19 @@ def _add_router_port(self, context, port_id, router, device_owner):
         if len(router_ports) > 1:
             subnets_id = []
             for rp in router_ports:
-                port = port_obj.Port.get_object(context.elevated(),
-                                                id=rp.port_id)
-                if port:
+                router_port = port_obj.Port.get_object(context.elevated(),
+                                                       id=rp.port_id)
+                if router_port:
+                    # NOTE(froyo): Just run the validation in case the new port
+                    # added is on the same network than an existing one.
                     # Only allow one router port with IPv6 subnets per network
-                    # id
-                    self._validate_one_router_ipv6_port_per_network(
-                        router, port)
-                    subnets_id.extend([fixed_ip['subnet_id']
-                                       for fixed_ip in port['fixed_ips']])
+                    # id.
+                    if router_port['network_id'] == port['network_id']:
+                        self._validate_one_router_ipv6_port_per_network(
+                            router, router_port)
+                    subnets_id.extend(
+                        [fixed_ip["subnet_id"]
+                         for fixed_ip in router_port["fixed_ips"]])
                 else:
                     # due to race conditions maybe the port under analysis is
                     # deleted, so instead returning a RouterInterfaceNotFound
@@ -1055,8 +1062,8 @@ def _add_router_port(self, context, port_id, router, device_owner):
         # make sure the records in routerports table and ports
         # table are consistent.
         self._core_plugin.update_port(
-            context, port_id, {'port': {'device_id': router.id,
-                                        'device_owner': device_owner}})
+            context, port['id'], {'port': {'device_id': router.id,
+                                           'device_owner': device_owner}})
 
     def _check_router_interface_not_in_use(self, router_id, subnet):
         context = n_ctx.get_admin_context()

neutron/tests/unit/db/test_l3_db.py

@@ -577,6 +577,53 @@ def test__validate_one_router_ipv6_port_per_network_mix_ipv4_ipv6(self):
         self.db._validate_one_router_ipv6_port_per_network(
             router, new_port)
 
+    def test__validate_one_router_ipv6_port_per_network_distributed_port(self):
+        port = models_v2.Port(
+                id=uuidutils.generate_uuid(),
+                network_id='foo_network',
+                device_owner=n_const.DEVICE_OWNER_DVR_INTERFACE,
+                fixed_ips=[models_v2.IPAllocation(
+                    ip_address=str(netaddr.IPNetwork(
+                        '2001:db8::/32').ip + 1),
+                    subnet_id='foo_subnet')])
+        rports = [l3_models.RouterPort(router_id='foo_router', port=port)]
+        router = l3_models.Router(
+            id='foo_router', attached_ports=rports, route_list=[],
+            gw_port_id=None)
+        new_port = models_v2.Port(
+                id=uuidutils.generate_uuid(),
+                network_id='foo_network',
+                device_owner=n_const.DEVICE_OWNER_ROUTER_SNAT,
+                fixed_ips=[models_v2.IPAllocation(
+                    ip_address=str(netaddr.IPNetwork(
+                        '2001:db8::/32').ip + 2),
+                    subnet_id='foo_subnet')])
+        self.db._validate_one_router_ipv6_port_per_network(router, new_port)
+
+    def test__validate_one_router_ipv6_port_per_network_centralized_snat_port(
+            self):
+        port = models_v2.Port(
+                id=uuidutils.generate_uuid(),
+                network_id='foo_network',
+                device_owner=n_const.DEVICE_OWNER_ROUTER_SNAT,
+                fixed_ips=[models_v2.IPAllocation(
+                    ip_address=str(netaddr.IPNetwork(
+                        '2001:db8::/32').ip + 1),
+                    subnet_id='foo_subnet')])
+        rports = [l3_models.RouterPort(router_id='foo_router', port=port)]
+        router = l3_models.Router(
+            id='foo_router', attached_ports=rports, route_list=[],
+            gw_port_id=None)
+        new_port = models_v2.Port(
+                id=uuidutils.generate_uuid(),
+                network_id='foo_network',
+                device_owner=n_const.DEVICE_OWNER_DVR_INTERFACE,
+                fixed_ips=[models_v2.IPAllocation(
+                    ip_address=str(netaddr.IPNetwork(
+                        '2001:db8::/32').ip + 2),
+                    subnet_id='foo_subnet')])
+        self.db._validate_one_router_ipv6_port_per_network(router, new_port)
+
     def test__validate_one_router_ipv6_port_per_network_failed(self):
         port = models_v2.Port(
                 id=uuidutils.generate_uuid(),