path: root/0001-add-distributed-traffic-feature-support.patch

Author: wangkuntian <wangkuntian@uniontech.com>
Date:   Fri Oct 13 16:25:17 2023 +0800

    feat: add distributed traffic feature
---
 agent/l3/dvr_edge_ha_router.py                                  |   4 +-
 agent/l3/extensions/rg_port_forwarding.py                       | 398 ++++++++++++++++++++++++++++
 agent/l3/ha.py                                                  |   9 -
 agent/l3/ha_router.py                                           |  74 +++---
 agent/l3/keepalived_state_change.py                             |  12 +-
 agent/l3/router_info.py                                         |  19 +-
 agent/linux/dhcp.py                                             | 137 +++++++++-
 agent/linux/interface.py                                        |  43 +--
 agent/linux/keepalived.py                                       |   1 +
 api/rpc/agentnotifiers/dhcp_rpc_agent_api.py                    |   9 +
 api/rpc/callbacks/resources.py                                  |   3 +
 conf/agent/l3/keepalived.py                                     |   2 +
 conf/common.py                                                  |  16 +-
 conf/policies/__init__.py                                       |   2 +
 conf/policies/rg_port_forwarding.py                             |  76 ++++++
 db/l3_attrs_db.py                                               |   7 +-
 db/l3_db.py                                                     |  27 +-
 db/l3_hamode_db.py                                              |  97 +++++++
 db/migration/alembic_migrations/versions/EXPAND_HEAD            |   2 +-
 .../train/expand/1c19a98b5eef_add_router_configurations.py      |  36 +++
 .../expand/cab12b72ed90_add_router_gateway_port_forwarding.py   |  55 ++++
 db/models/l3_attrs.py                                           |   2 +
 db/models/rg_port_forwarding.py                                 |  59 +++++
 extensions/rg_port_forwarding.py                                | 119 +++++++++
 objects/rg_port_forwarding.py                                   |  87 ++++++
 objects/router.py                                               |  15 +-
 scheduler/l3_agent_scheduler.py                                 | 107 ++++++--
 services/l3_router/l3_router_plugin.py                          |  12 +
 services/rg_portforwarding/__init__.py                          |   0
 services/rg_portforwarding/common/__init__.py                   |   0
 services/rg_portforwarding/common/exceptions.py                 |  77 ++++++
 services/rg_portforwarding/pf_plugin.py                         | 369 ++++++++++++++++++++++++++
 32 files changed, 1749 insertions(+), 127 deletions(-)

diff --git a/agent/l3/dvr_edge_ha_router.py b/agent/l3/dvr_edge_ha_router.py
index 71f740bef9..b92f70b70f 100644
--- a/agent/l3/dvr_edge_ha_router.py
+++ b/agent/l3/dvr_edge_ha_router.py
@@ -114,9 +114,7 @@ class DvrEdgeHaRouter(dvr_edge_router.DvrEdgeRouter,
 
     def _external_gateway_added(self, ex_gw_port, interface_name,
                                 ns_name, preserve_ips):
-        link_up = self.external_gateway_link_up()
-        self._plug_external_gateway(ex_gw_port, interface_name, ns_name,
-                                    link_up=link_up)
+        self._plug_external_gateway(ex_gw_port, interface_name, ns_name)
 
     def _is_this_snat_host(self):
         return self.agent_conf.agent_mode == constants.L3_AGENT_MODE_DVR_SNAT
diff --git a/agent/l3/extensions/rg_port_forwarding.py b/agent/l3/extensions/rg_port_forwarding.py
new file mode 100644
index 0000000000..a159e4df34
--- /dev/null
+++ b/agent/l3/extensions/rg_port_forwarding.py
@@ -0,0 +1,398 @@
+# Copyright (c) 2023 UnionTech
+# All rights reserved
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+import collections
+from typing import Optional, List
+from oslo_concurrency import lockutils
+from oslo_log import log as logging
+
+from neutron_lib import constants
+from neutron_lib.rpc import Connection
+from neutron_lib.context import Context
+from neutron_lib.agent import l3_extension
+
+from neutron.agent.linux.ip_lib import IPDevice
+from neutron.agent.l3.router_info import RouterInfo
+from neutron.agent.linux.iptables_manager import IptablesManager
+
+from neutron.api.rpc.handlers import resources_rpc
+from neutron.api.rpc.callbacks import resources, events
+from neutron.api.rpc.callbacks.consumer import registry
+
+from neutron.common import coordination
+
+from neutron.objects.ports import Port
+from neutron.objects.router import Router
+from neutron.objects.rg_port_forwarding import RGPortForwarding
+
+LOG = logging.getLogger(__name__)
+
+PORT_FORWARDING_PREFIX = 'rg_portforwarding-'
+DEFAULT_PORT_FORWARDING_CHAIN = 'rg-pf'
+PORT_FORWARDING_CHAIN_PREFIX = 'pf-'
+
+
+def _get_port_forwarding_chain_name(pf_id):
+    chain_name = PORT_FORWARDING_CHAIN_PREFIX + pf_id
+    return chain_name[:constants.MAX_IPTABLES_CHAIN_LEN_WRAP]
+
+
+class RGPortForwardingMapping(object):
+    def __init__(self):
+        self.managed_port_forwardings = {}
+        self.router_pf_mapping = collections.defaultdict(set)
+
+    @lockutils.synchronized('rg-port-forwarding-cache')
+    def check_port_forwarding_changes(self, new_pf: RGPortForwarding) -> bool:
+        old_pf = self.managed_port_forwardings.get(new_pf.id)
+        return old_pf != new_pf
+
+    @lockutils.synchronized('rg-port-forwarding-cache')
+    def set_port_forwardings(self, port_forwardings: List[RGPortForwarding]):
+        for port_forwarding in port_forwardings:
+            self._set_router_port_forwarding(port_forwarding,
+                                             port_forwarding.router_id)
+
+    def _set_router_port_forwarding(self,
+                                    port_forwarding: RGPortForwarding,
+                                    router_id: str):
+        self.router_pf_mapping[router_id].add(port_forwarding.id)
+        self.managed_port_forwardings[port_forwarding.id] = port_forwarding
+
+    @lockutils.synchronized('rg-port-forwarding-cache')
+    def update_port_forwardings(self, port_forwardings):
+        for port_forwarding in port_forwardings:
+            self.managed_port_forwardings[port_forwarding.id] = port_forwarding
+
+    @lockutils.synchronized('rg-port-forwarding-cache')
+    def del_port_forwardings(self, port_forwardings):
+        for port_forwarding in port_forwardings:
+            if not self.managed_port_forwardings.get(port_forwarding.id):
+                continue
+            self.managed_port_forwardings.pop(port_forwarding.id)
+            self.router_pf_mapping[port_forwarding.router_id].discard(
+                port_forwarding.id)
+            if not self.router_pf_mapping[port_forwarding.router_id]:
+                self.router_pf_mapping.pop(port_forwarding.router_id)
+
+    @lockutils.synchronized('rg-port-forwarding-cache')
+    def clean_port_forwardings_by_router_id(self, router_id: str):
+        pf_ids = self.router_pf_mapping.pop(router_id, [])
+        for pf_id in pf_ids:
+            self.managed_port_forwardings.pop(pf_id, None)
+
+
+class RGPortForwardingAgentExtension(l3_extension.L3AgentExtension):
+    SUPPORTED_RESOURCE_TYPES = [resources.RGPORTFORWARDING]
+
+    def consume_api(self, agent_api):
+        self.agent_api = agent_api
+
+    def initialize(self, connection, driver_type):
+        self.mapping = RGPortForwardingMapping()
+        self.resource_rpc = resources_rpc.ResourcesPullRpcApi()
+        self._register_rpc_consumers()
+
+    def _register_rpc_consumers(self):
+        registry.register(self._handle_notification,
+                          resources.RGPORTFORWARDING)
+        self._connection = Connection()
+        endpoints = [resources_rpc.ResourcesPushRpcCallback()]
+        topic = resources_rpc.resource_type_versioned_topic(
+            resources.RGPORTFORWARDING)
+        self._connection.create_consumer(topic, endpoints, fanout=True)
+        self._connection.consume_in_threads()
+
+    def _handle_notification(self, context: Context,
+                             resource_type: str,
+                             forwardings, event_type):
+        for forwarding in forwardings:
+            self._process_port_forwarding_event(
+                context, forwarding, event_type)
+
+    def _get_gw_port_and_ip(self,
+                            ri: RouterInfo) -> (Optional[Port], Optional[str]):
+        ex_gw_port = ri.get_ex_gw_port()
+        ex_gw_port_ip = self._get_gw_port_ip(ex_gw_port)
+        if not ex_gw_port_ip:
+            LOG.error(f"Router {ri.router_id} external port "
+                      f"{ex_gw_port['id']} does not have any IP addresses")
+            return None, None
+        return ex_gw_port, ex_gw_port_ip
+
+    def _process_port_forwarding_event(self, context: Context,
+                                       port_forwarding: RGPortForwarding,
+                                       event_type: str):
+        router_id = port_forwarding.router_id
+        ri = self._get_router_info(router_id)
+        if not self._check_if_need_process(ri, force=True):
+            return
+
+        ex_gw_port, ex_gw_port_ip = self._get_gw_port_and_ip(ri)
+        if not ex_gw_port or not ex_gw_port_ip:
+            return
+
+        (interface_name, namespace,
+         iptables_manager) = self._get_resource_by_router(ri, ex_gw_port)
+
+        if event_type == events.CREATED:
+            self._process_create([port_forwarding], ri, interface_name,
+                                 ex_gw_port_ip, namespace, iptables_manager)
+        elif event_type == events.UPDATED:
+            self._process_update([port_forwarding], interface_name,
+                                 ex_gw_port_ip, namespace, iptables_manager)
+        elif event_type == events.DELETED:
+            self._process_delete([port_forwarding], interface_name,
+                                 ex_gw_port_ip, namespace, iptables_manager)
+
+    def ha_state_change(self, context: Context, data: Router) -> None:
+        pass
+
+    def update_network(self, context: Context, data: dict) -> None:
+        pass
+
+    def add_router(self, context: Context, data: Router) -> None:
+        LOG.info(f"call add_router for {data['id']}")
+        self.process_port_forwarding(context, data)
+
+    def update_router(self, context: Context, data: Router) -> None:
+        LOG.info(f"call update_router for {data['id']}")
+        self.process_port_forwarding(context, data)
+
+    def delete_router(self, context: Context, data: Router) -> None:
+        self.mapping.clean_port_forwardings_by_router_id(data['id'])
+
+    def _get_router_info(self, router_id) -> Optional[RouterInfo]:
+        router_info = self.agent_api.get_router_info(router_id)
+        if router_info:
+            return router_info
+        LOG.debug("Router %s is not managed by this agent. "
+                  "It was possibly deleted concurrently.", router_id)
+
+    @staticmethod
+    def _check_if_need_process(ri: RouterInfo, force: bool = False) -> bool:
+        if not ri or not ri.get_ex_gw_port():
+            return False
+
+        if force:
+            return True
+
+        is_distributed = ri.router.get('distributed')
+        agent_mode = ri.agent_conf.agent_mode
+        if (is_distributed and
+                agent_mode in [constants.L3_AGENT_MODE_DVR_NO_EXTERNAL,
+                               constants.L3_AGENT_MODE_DVR]):
+            # just support centralized cases
+            return False
+
+        if is_distributed and not ri.snat_namespace.exists():
+            return False
+
+        return True
+
+    def process_port_forwarding(self, context: Context, data: Router):
+        ri = self._get_router_info(data['id'])
+        if not self._check_if_need_process(ri):
+            return
+        self.check_local_port_forwardings(context, ri)
+
+    @staticmethod
+    def _get_gw_port_ip(gw_port: dict) -> Optional[str]:
+        fixed_ips = gw_port.get('fixed_ips', [])
+        if not fixed_ips:
+            return
+        return fixed_ips[0].get('ip_address', None)
+
+    @staticmethod
+    def _get_resource_by_router(ri: RouterInfo, ex_gw_port: dict) -> (
+            str, str, IptablesManager):
+        is_distributed = ri.router.get('distributed')
+        if is_distributed:
+            interface_name = ri.get_snat_external_device_interface_name(
+                ex_gw_port)
+            namespace = ri.snat_namespace.name
+            iptables_manager = ri.snat_iptables_manager
+        else:
+            interface_name = ri.get_external_device_interface_name(ex_gw_port)
+            namespace = ri.ns_name
+            iptables_manager = ri.iptables_manager
+
+        return interface_name, namespace, iptables_manager
+
+    def check_local_port_forwardings(self, context: Context, ri: RouterInfo):
+        pfs = self.resource_rpc.bulk_pull(
+            context, resources.RGPORTFORWARDING,
+            filter_kwargs={'router_id': ri.router_id})
+        if not pfs:
+            return
+        ex_gw_port, ex_gw_port_ip = self._get_gw_port_and_ip(ri)
+        if not ex_gw_port_ip or not ex_gw_port_ip:
+            return
+        (interface_name, namespace,
+         iptables_manager) = self._get_resource_by_router(ri, ex_gw_port)
+        local_pfs = set(self.mapping.managed_port_forwardings.keys())
+        new_pfs = []
+        updated_pfs = []
+        current_pfs = set()
+        for pf in pfs:
+            if pf.id in self.mapping.managed_port_forwardings:
+                if self.mapping.check_port_forwarding_changes(pf):
+                    updated_pfs.append(pf)
+            else:
+                new_pfs.append(pf)
+            current_pfs.add(pf.id)
+
+        remove_pf_ids_set = local_pfs - current_pfs
+        remove_pfs = [self.mapping.managed_port_forwardings[pf_id]
+                      for pf_id in remove_pf_ids_set]
+
+        self._process_create(new_pfs, ri, interface_name, ex_gw_port_ip,
+                             namespace, iptables_manager)
+
+        self._process_update(updated_pfs, interface_name, ex_gw_port_ip,
+                             namespace, iptables_manager)
+
+        self._process_delete(remove_pfs, interface_name, ex_gw_port_ip,
+                             namespace, iptables_manager)
+
+    @staticmethod
+    def _install_default_rules(iptables_manager: IptablesManager):
+        default_rule = '-j %s-%s' % (iptables_manager.wrap_name,
+                                     DEFAULT_PORT_FORWARDING_CHAIN)
+        LOG.info(f'Add default chain {DEFAULT_PORT_FORWARDING_CHAIN}')
+        LOG.info(f'Add default rule {default_rule}')
+        iptables_manager.ipv4['nat'].add_chain(DEFAULT_PORT_FORWARDING_CHAIN)
+        iptables_manager.ipv4['nat'].add_rule('PREROUTING', default_rule)
+        iptables_manager.apply()
+
+    @staticmethod
+    def _get_rg_rules(port_forward: RGPortForwarding, wrap_name: str):
+        chain_rule_list = []
+        pf_chain_name = _get_port_forwarding_chain_name(port_forward.id)
+        chain_rule_list.append(
+            (DEFAULT_PORT_FORWARDING_CHAIN, f'-j {wrap_name}-{pf_chain_name}'))
+        gw_ip_address = port_forward.gw_ip_address
+        protocol = port_forward.protocol
+        internal_ip_address = str(port_forward.internal_ip_address)
+        internal_port = port_forward.internal_port
+        external_port = port_forward.external_port
+        chain_rule = (
+            pf_chain_name,
+            f'-d {gw_ip_address}/32 -p {protocol} -m {protocol} '
+            f'--dport {external_port} '
+            f'-j DNAT --to-destination {internal_ip_address}:{internal_port}'
+        )
+        chain_rule_list.append(chain_rule)
+        return chain_rule_list
+
+    def _rule_apply(self,
+                    iptables_manager: IptablesManager,
+                    port_forwarding: RGPortForwarding,
+                    rule_tag: str):
+        iptables_manager.ipv4['nat'].clear_rules_by_tag(rule_tag)
+        if (DEFAULT_PORT_FORWARDING_CHAIN not in
+                iptables_manager.ipv4['nat'].chains):
+            self._install_default_rules(iptables_manager)
+
+        for chain, rule in self._get_rg_rules(port_forwarding,
+                                              iptables_manager.wrap_name):
+            LOG.info(f'Add router gateway port forwarding '
+                     f'rule {rule} in {chain}')
+            if chain not in iptables_manager.ipv4['nat'].chains:
+                iptables_manager.ipv4['nat'].add_chain(chain)
+            iptables_manager.ipv4['nat'].add_rule(chain, rule, tag=rule_tag)
+
+    def _store_local(self, pf_objs: List[RGPortForwarding], event_type: str):
+        if event_type == events.CREATED:
+            self.mapping.set_port_forwardings(pf_objs)
+        elif event_type == events.UPDATED:
+            self.mapping.update_port_forwardings(pf_objs)
+        elif event_type == events.DELETED:
+            self.mapping.del_port_forwardings(pf_objs)
+
+    def _process_create(self,
+                        port_forwardings: List[RGPortForwarding],
+                        ri: RouterInfo,
+                        interface_name: str,
+                        interface_ip: str,
+                        namespace: str,
+                        iptables_manager: IptablesManager):
+        if not port_forwardings:
+            return
+
+        ha_port = ri.router.get(constants.HA_INTERFACE_KEY, None)
+        if ha_port and ha_port['status'] == constants.PORT_STATUS_ACTIVE:
+            ri.enable_keepalived()
+
+        for port_forwarding in port_forwardings:
+            if port_forwarding.id in self.mapping.managed_port_forwardings:
+                LOG.debug("Skip port forwarding %s for create, as it had been "
+                          "managed by agent", port_forwarding.id)
+                continue
+            rule_tag = PORT_FORWARDING_PREFIX + port_forwarding.id
+            port_forwarding.gw_ip_address = interface_ip
+            self._rule_apply(iptables_manager, port_forwarding, rule_tag)
+        iptables_manager.apply()
+        self._store_local(port_forwardings, events.CREATED)
+
+    def _process_update(self,
+                        port_forwardings: List[RGPortForwarding],
+                        interface_name: str,
+                        interface_ip: str,
+                        namespace: str,
+                        iptables_manager: IptablesManager):
+        if not port_forwardings:
+            return
+        device = IPDevice(interface_name, namespace=namespace)
+        for port_forwarding in port_forwardings:
+            # check if port forwarding change from OVO and router rpc
+            if not self.mapping.check_port_forwarding_changes(port_forwarding):
+                LOG.debug("Skip port forwarding %s for update, as there is no "
+                          "difference between the memory managed by agent",
+                          port_forwarding.id)
+                continue
+            current_chain = _get_port_forwarding_chain_name(port_forwarding.id)
+            iptables_manager.ipv4['nat'].remove_chain(current_chain)
+            ori_pf = self.mapping.managed_port_forwardings[port_forwarding.id]
+            device.delete_socket_conntrack_state(interface_ip,
+                                                 ori_pf.external_port,
+                                                 protocol=ori_pf.protocol)
+            rule_tag = PORT_FORWARDING_PREFIX + port_forwarding.id
+            port_forwarding.gw_ip_address = interface_ip
+            self._rule_apply(iptables_manager, port_forwarding, rule_tag)
+        iptables_manager.apply()
+        self._store_local(port_forwardings, events.UPDATED)
+
+    @coordination.synchronized('router-lock-ns-{namespace}')
+    def _process_delete(self,
+                        port_forwardings: List[RGPortForwarding],
+                        interface_name: str,
+                        interface_ip: str,
+                        namespace: str,
+                        iptables_manager: IptablesManager):
+        if not port_forwardings:
+            return
+        device = IPDevice(interface_name, namespace=namespace)
+        for port_forwarding in port_forwardings:
+            current_chain = _get_port_forwarding_chain_name(port_forwarding.id)
+            iptables_manager.ipv4['nat'].remove_chain(current_chain)
+            device.delete_socket_conntrack_state(
+                interface_ip,
+                port_forwarding.external_port,
+                protocol=port_forwarding.protocol)
+
+        iptables_manager.apply()
+
+        self._store_local(port_forwardings, events.DELETED)
diff --git a/agent/l3/ha.py b/agent/l3/ha.py
index 17891dc983..182fa68175 100644
--- a/agent/l3/ha.py
+++ b/agent/l3/ha.py
@@ -163,15 +163,6 @@ class AgentMixin(object):
                  'agent %(host)s',
                  state_change_data)
 
-        # Set external gateway port link up or down according to state
-        if state == 'master':
-            ri.set_external_gw_port_link_status(link_up=True, set_gw=True)
-        elif state == 'backup':
-            ri.set_external_gw_port_link_status(link_up=False)
-        else:
-            LOG.warning('Router %s has status %s, '
-                        'no action to router gateway device.',
-                        router_id, state)
         # TODO(dalvarez): Fix bug 1677279 by moving the IPv6 parameters
         # configuration to keepalived-state-change in order to remove the
         # dependency that currently exists on l3-agent running for the IPv6
diff --git a/agent/l3/ha_router.py b/agent/l3/ha_router.py
index 0a21902771..ef10ff76e4 100644
--- a/agent/l3/ha_router.py
+++ b/agent/l3/ha_router.py
@@ -17,6 +17,7 @@ import shutil
 import signal
 
 import netaddr
+from typing import Optional, List
 from neutron_lib.api.definitions import portbindings
 from neutron_lib import constants as n_consts
 from neutron_lib.utils import runtime
@@ -137,6 +138,22 @@ class HaRouter(router.RouterInfo):
         else:
             return False
 
+    @property
+    def configurations(self) -> Optional[dict]:
+        return self.router.get('configurations', {})
+
+    @property
+    def master(self) -> Optional[str]:
+        if self.configurations:
+            return self.configurations.get('master_agent', None)
+        return None
+
+    @property
+    def slaves(self) -> Optional[List[str]]:
+        if self.configurations:
+            return self.configurations.get('slave_agents', [])
+        return []
+
     def initialize(self, process_monitor):
         ha_port = self.router.get(n_consts.HA_INTERFACE_KEY)
         if not ha_port:
@@ -162,19 +179,32 @@ class HaRouter(router.RouterInfo):
             throttle_restart_value=(
                 self.agent_conf.ha_vrrp_advert_int * THROTTLER_MULTIPLIER))
 
+        # The following call is required to ensure that if the state path does
+        # not exist it gets created.
+        self.keepalived_manager.get_full_config_file_path('test')
+
         config = self.keepalived_manager.config
 
         interface_name = self.get_ha_device_name()
         subnets = self.ha_port.get('subnets', [])
         ha_port_cidrs = [subnet['cidr'] for subnet in subnets]
+        nopreempt = True
+        state = 'BACKUP'
+        priority = self.ha_priority
+        if self.slaves and self.master:
+            nopreempt = False
+            if self.master == self.agent_conf.host:
+                state = 'MASTER'
+                priority = keepalived.HA_DEFAULT_MASTER_PRIORITY
+
         instance = keepalived.KeepalivedInstance(
-            'BACKUP',
+            state,
             interface_name,
             self.ha_vr_id,
             ha_port_cidrs,
-            nopreempt=True,
+            nopreempt=nopreempt,
             advert_int=self.agent_conf.ha_vrrp_advert_int,
-            priority=self.ha_priority,
+            priority=priority,
             vrrp_health_check_interval=(
                 self.agent_conf.ha_vrrp_health_check_interval),
             ha_conf_dir=self.keepalived_manager.get_conf_dir())
@@ -396,13 +426,16 @@ class HaRouter(router.RouterInfo):
         ha_device = self.get_ha_device_name()
         ha_cidr = self._get_primary_vip()
         config_dir = self.keepalived_manager.get_conf_dir()
-        state_change_log = (
-            "%s/neutron-keepalived-state-change.log") % config_dir
+        state_change_log = f"{config_dir}/neutron-keepalived-state-change.log"
 
         def callback(pid_file):
+            LOG.info(f'Router: {self.router_id} master is {self.master}, '
+                     f'salves are {self.slaves}.')
             cmd = [
                 STATE_CHANGE_PROC_NAME,
                 '--router_id=%s' % self.router_id,
+                '--master_agent=%s' % self.master,
+                '--slave_agents=%s' % ','.join(self.slaves),
                 '--namespace=%s' % self.ha_namespace,
                 '--conf_dir=%s' % config_dir,
                 '--log-file=%s' % state_change_log,
@@ -453,9 +486,7 @@ class HaRouter(router.RouterInfo):
         return port1_filtered == port2_filtered
 
     def external_gateway_added(self, ex_gw_port, interface_name):
-        link_up = self.external_gateway_link_up()
-        self._plug_external_gateway(ex_gw_port, interface_name,
-                                    self.ns_name, link_up=link_up)
+        self._plug_external_gateway(ex_gw_port, interface_name, self.ns_name)
         self._add_gateway_vip(ex_gw_port, interface_name)
         self._disable_ipv6_addressing_on_interface(interface_name)
 
@@ -519,30 +550,3 @@ class HaRouter(router.RouterInfo):
         if (self.keepalived_manager.get_process().active and
                 self.ha_state == 'master'):
             super(HaRouter, self).enable_radvd(internal_ports)
-
-    def external_gateway_link_up(self):
-        # Check HA router ha_state for its gateway port link state.
-        # 'backup' instance will not link up the gateway port.
-        return self.ha_state == 'master'
-
-    def set_external_gw_port_link_status(self, link_up, set_gw=False):
-        link_state = "up" if link_up else "down"
-        LOG.info('Set router %s gateway device link state to %s.',
-                 self.router_id, link_state)
-
-        ex_gw_port = self.get_ex_gw_port()
-        ex_gw_port_id = (ex_gw_port and ex_gw_port['id'] or
-                         self.ex_gw_port and self.ex_gw_port['id'])
-        if ex_gw_port_id:
-            interface_name = self.get_external_device_name(ex_gw_port_id)
-            ns_name = self.get_gw_ns_name()
-            if (not self.driver.set_link_status(
-                    interface_name, namespace=ns_name, link_up=link_up) and
-                    link_up):
-                LOG.error('Gateway interface for router %s was not set up; '
-                          'router will not work properly', self.router_id)
-            if link_up and set_gw:
-                preserve_ips = self.get_router_preserve_ips()
-                self._external_gateway_settings(ex_gw_port, interface_name,
-                                                ns_name, preserve_ips)
-                self.routes_updated([], self.routes)
diff --git a/agent/l3/keepalived_state_change.py b/agent/l3/keepalived_state_change.py
index 7fd9e4269e..8c10a8b00f 100644
--- a/agent/l3/keepalived_state_change.py
+++ b/agent/l3/keepalived_state_change.py
@@ -47,9 +47,12 @@ class KeepalivedUnixDomainConnection(agent_utils.UnixDomainHTTPConnection):
 
 
 class MonitorDaemon(daemon.Daemon):
-    def __init__(self, pidfile, router_id, user, group, namespace, conf_dir,
-                 interface, cidr):
+    def __init__(self, pidfile, host, router_id, master, slaves, user, group,
+                 namespace, conf_dir, interface, cidr):
+        self.host = host
         self.router_id = router_id
+        self.master = master
+        self.slaves = slaves
         self.namespace = namespace
         self.conf_dir = conf_dir
         self.interface = interface
@@ -62,6 +65,8 @@ class MonitorDaemon(daemon.Daemon):
                                             user=user, group=group)
 
     def run(self):
+        LOG.debug(f'Router: {self.router_id} master is {self.master}, '
+                  f'salves are {self.slaves}.')
         self._thread_ip_monitor = threading.Thread(
             target=ip_lib.ip_monitor,
             args=(self.namespace, self.queue, self.event_stop,
@@ -169,7 +174,10 @@ def main():
     keepalived.register_l3_agent_keepalived_opts()
     configure(cfg.CONF)
     MonitorDaemon(cfg.CONF.pid_file,
+                  cfg.CONF.host,
                   cfg.CONF.router_id,
+                  cfg.CONF.master_agent,
+                  cfg.CONF.slave_agents,
                   cfg.CONF.user,
                   cfg.CONF.group,
                   cfg.CONF.namespace,
diff --git a/agent/l3/router_info.py b/agent/l3/router_info.py
index ea2b488cd2..eabdcc6e54 100644
--- a/agent/l3/router_info.py
+++ b/agent/l3/router_info.py
@@ -697,16 +697,14 @@ class RouterInfo(BaseRouterInfo):
         return [common_utils.ip_to_cidr(ip['floating_ip_address'])
                 for ip in floating_ips]
 
-    def _plug_external_gateway(self, ex_gw_port, interface_name, ns_name,
-                               link_up=True):
+    def _plug_external_gateway(self, ex_gw_port, interface_name, ns_name):
         self.driver.plug(ex_gw_port['network_id'],
                          ex_gw_port['id'],
                          interface_name,
                          ex_gw_port['mac_address'],
                          namespace=ns_name,
                          prefix=EXTERNAL_DEV_PREFIX,
-                         mtu=ex_gw_port.get('mtu'),
-                         link_up=link_up)
+                         mtu=ex_gw_port.get('mtu'))
 
     def _get_external_gw_ips(self, ex_gw_port):
         gateway_ips = []
@@ -766,11 +764,7 @@ class RouterInfo(BaseRouterInfo):
         LOG.debug("External gateway added: port(%s), interface(%s), ns(%s)",
                   ex_gw_port, interface_name, ns_name)
         self._plug_external_gateway(ex_gw_port, interface_name, ns_name)
-        self._external_gateway_settings(ex_gw_port, interface_name,
-                                        ns_name, preserve_ips)
 
-    def _external_gateway_settings(self, ex_gw_port, interface_name,
-                                   ns_name, preserve_ips):
         # Build up the interface and gateway IP addresses that
         # will be added to the interface.
         ip_cidrs = common_utils.fixed_ip_cidrs(ex_gw_port['fixed_ips'])
@@ -815,19 +809,18 @@ class RouterInfo(BaseRouterInfo):
         return any(netaddr.IPAddress(gw_ip).version == 6
                    for gw_ip in gateway_ips)
 
-    def get_router_preserve_ips(self):
+    def external_gateway_added(self, ex_gw_port, interface_name):
         preserve_ips = self._list_floating_ip_cidrs() + list(
             self.centralized_port_forwarding_fip_set)
         preserve_ips.extend(self.agent.pd.get_preserve_ips(self.router_id))
-        return preserve_ips
 
-    def external_gateway_added(self, ex_gw_port, interface_name):
-        preserve_ips = self.get_router_preserve_ips()
         self._external_gateway_added(
             ex_gw_port, interface_name, self.ns_name, preserve_ips)
 
     def external_gateway_updated(self, ex_gw_port, interface_name):
-        preserve_ips = self.get_router_preserve_ips()
+        preserve_ips = self._list_floating_ip_cidrs() + list(
+            self.centralized_port_forwarding_fip_set)
+        preserve_ips.extend(self.agent.pd.get_preserve_ips(self.router_id))
         self._external_gateway_added(
             ex_gw_port, interface_name, self.ns_name, preserve_ips)
 
diff --git a/agent/linux/dhcp.py b/agent/linux/dhcp.py
index 249e1a8199..5b82fe328f 100644
--- a/agent/linux/dhcp.py
+++ b/agent/linux/dhcp.py
@@ -19,6 +19,7 @@ import os
 import re
 import shutil
 import time
+from typing import List
 
 import netaddr
 from neutron_lib.api.definitions import extra_dhcp_opt as edo_ext
@@ -40,6 +41,7 @@ from neutron.agent.linux import iptables_manager
 from neutron.cmd import runtime_checks as checks
 from neutron.common import ipv6_utils
 from neutron.common import utils as common_utils
+from neutron.conf.common import NETWORK_HOST_OPTS
 from neutron.ipam import utils as ipam_utils
 
 LOG = logging.getLogger(__name__)
@@ -81,6 +83,51 @@ def port_requires_dhcp_configuration(port):
         constants.DEVICE_OWNER_DHCP]
 
 
+class Octopus(object):
+    def __init__(self):
+        self.tentacles = collections.defaultdict(Tentacle)
+
+    def __str__(self):
+        lines = ['']
+        for subnet_id, tentacle in self.tentacles.items():
+            line = (f"Subnet {subnet_id} has multi routers "
+                    f"{len(tentacle.gateway_ports) > 1} "
+                    f"{tentacle}")
+            lines.append(line)
+        return '\n'.join(lines)
+
+
+class Tentacle(object):
+    def __init__(self, gateway_ip: str):
+        self.gateway_ip = gateway_ip
+        self.tags = []
+        self.gateway_ports = collections.defaultdict(Sucker)
+        self.suckers = collections.defaultdict(Sucker)
+
+    def __str__(self):
+        lines = [""]
+        for port_id, sucker in self.suckers.items():
+            line = f"  Port {port_id} {sucker}"
+            lines.append(line)
+        return '\n'.join(lines)
+
+
+class Sucker(object):
+    def __init__(self, host: str, device_owner: str, ip_address: str):
+        self.host = host
+        self.device_owner = device_owner
+        self.ip_address = ip_address
+        self.tag = None
+
+    def subnet_tag(self, subnet_id: str):
+        return f'{self.host}-subnet-{subnet_id}'
+
+    def __str__(self):
+        return (f"ip: {self.ip_address} \t"
+                f"binding_host: {self.host} \t"
+                f"device_owner: {self.device_owner}")
+
+
 class DictModel(dict):
     """Convert dict into an object that provides attribute access to values."""
 
@@ -149,6 +196,10 @@ class DhcpBase(object):
         self.process_monitor = process_monitor
         self.device_manager = DeviceManager(self.conf, plugin)
         self.version = version
+        self.octopus = Octopus()
+        self._init_octopus()
+        self.compute_to_network = dict()
+        self._init_compute_to_network()
 
     @abc.abstractmethod
     def enable(self):
@@ -193,6 +244,31 @@ class DhcpBase(object):
         """True if the metadata-proxy should be enabled for the network."""
         raise NotImplementedError()
 
+    def _init_compute_to_network(self):
+        for network_node in self.conf.network_nodes:
+            self.conf.register_opts(NETWORK_HOST_OPTS, group=network_node)
+            network_group = self.conf.get(network_node, None)
+            if network_group:
+                compute_nodes = network_group.get('compute_nodes', [])
+                for compute_node in compute_nodes:
+                    self.compute_to_network[compute_node] = network_node
+
+    def _init_octopus(self):
+        for subnet in self.network.subnets:
+            self.octopus.tentacles[subnet.id] = Tentacle(subnet.gateway_ip)
+
+        for port in self.network.ports:
+            for ip in port.fixed_ips:
+                host = port.get('binding:host_id')
+                device_owner = port.get('device_owner')
+                ip_address = ip.get('ip_address')
+                tentacle = self.octopus.tentacles[ip.subnet_id]
+                sucker = Sucker(host, device_owner, ip_address)
+                tentacle.suckers[port.id] = sucker
+                if (device_owner in (constants.DEVICE_OWNER_HA_REPLICATED_INT,
+                                     constants.DEVICE_OWNER_ROUTER_INTF)):
+                    tentacle.gateway_ports[host] = sucker
+
 
 @six.add_metaclass(abc.ABCMeta)
 class DhcpLocalProcess(DhcpBase):
@@ -841,8 +917,18 @@ class Dnsmasq(DhcpLocalProcess):
                               (port.mac_address, tag, name, ip_address,
                                'set:', self._PORT_TAG_PREFIX % port.id))
             else:
-                buf.write('%s,%s%s,%s\n' %
-                          (port.mac_address, tag, name, ip_address))
+                subnet_tag = f'subnet-{alloc.subnet_id}'
+                if self.conf.enable_set_route_for_single_port:
+                    tentacle = self.octopus.tentacles[alloc.subnet_id]
+                    sucker = tentacle.suckers[port.id]
+                    tentacle.tags.append(subnet_tag)
+                    if sucker.device_owner.startswith(
+                            constants.DEVICE_OWNER_COMPUTE_PREFIX):
+                        subnet_tag = sucker.subnet_tag(alloc.subnet_id)
+                        sucker.tag = subnet_tag
+
+                buf.write(f'{port.mac_address},{tag}{name},{ip_address},'
+                          f'set:{subnet_tag}\n')
 
         file_utils.replace_file(filename, buf.getvalue())
         LOG.debug('Done building host file %s', filename)
@@ -1059,7 +1145,8 @@ class Dnsmasq(DhcpLocalProcess):
         """Write a dnsmasq compatible options file."""
         options, subnet_index_map = self._generate_opts_per_subnet()
         options += self._generate_opts_per_port(subnet_index_map)
-
+        if self.conf.enable_set_route_for_single_port:
+            options += self._generate_opts_for_compute_port(options)
         name = self.get_conf_file_name('opts')
         file_utils.replace_file(name, '\n'.join(options))
         return name
@@ -1220,6 +1307,50 @@ class Dnsmasq(DhcpLocalProcess):
                                                                   vx_ips))))
         return options
 
+    def _generate_opts_for_compute_port(self, options: List[str]) -> List[str]:
+        new_options = []
+        LOG.debug(self.octopus)
+        if not self.compute_to_network:
+            LOG.warning('CONF.enable_set_route_for_single_port is True, '
+                        'but not configured.')
+            return new_options
+        for subnet_id, tentacle in self.octopus.tentacles.items():
+            if len(tentacle.tags) == 0:
+                continue
+            if len(tentacle.gateway_ports) <= 1:
+                LOG.info(f'Subnet {subnet_id} is not bound '
+                         f'to different routers, '
+                         f'so skip generate options for compute ports.')
+                continue
+            for port_id, sucker in tentacle.suckers.items():
+                if not sucker.tag:
+                    continue
+                if not sucker.device_owner.startswith(
+                        constants.DEVICE_OWNER_COMPUTE_PREFIX):
+                    continue
+                network_node = self.compute_to_network.get(sucker.host, None)
+                if not network_node:
+                    LOG.warning(f'Compute host {sucker.host} not configured.')
+                    continue
+                port = tentacle.gateway_ports.get(network_node, None)
+                if not port:
+                    LOG.warning(f'Subnet {subnet_id} does not have gateway '
+                                f'port on network host {network_node}.')
+                    continue
+                if tentacle.gateway_ip == port.ip_address:
+                    continue
+                for option in options.copy():
+                    if subnet_id in option:
+                        option = option.replace(f'subnet-{subnet_id}',
+                                                sucker.tag)
+                        if ('option:classless-static-route' in option or
+                                ',249,' in option or 'option:router' in option):
+                            gateway_ip = option.split(',')[-1]
+                            option = option.replace(gateway_ip,
+                                                    port.ip_address)
+                        new_options.append(option)
+        return new_options
+
     def _make_subnet_interface_ip_map(self):
         subnet_lookup = dict(
             (netaddr.IPNetwork(subnet.cidr), subnet.id)
diff --git a/agent/linux/interface.py b/agent/linux/interface.py
index 3ac476d7ba..2e6455707c 100644
--- a/agent/linux/interface.py
+++ b/agent/linux/interface.py
@@ -259,17 +259,16 @@ class LinuxInterfaceDriver(object):
 
     @abc.abstractmethod
     def plug_new(self, network_id, port_id, device_name, mac_address,
-                 bridge=None, namespace=None, prefix=None, mtu=None,
-                 link_up=True):
+                 bridge=None, namespace=None, prefix=None, mtu=None):
         """Plug in the interface only for new devices that don't exist yet."""
 
     def plug(self, network_id, port_id, device_name, mac_address,
-             bridge=None, namespace=None, prefix=None, mtu=None, link_up=True):
+             bridge=None, namespace=None, prefix=None, mtu=None):
         if not ip_lib.device_exists(device_name,
                                     namespace=namespace):
             self._safe_plug_new(
                 network_id, port_id, device_name, mac_address, bridge,
-                namespace, prefix, mtu, link_up)
+                namespace, prefix, mtu)
         else:
             LOG.info("Device %s already exists", device_name)
             if mtu:
@@ -279,11 +278,11 @@ class LinuxInterfaceDriver(object):
                 LOG.warning("No MTU configured for port %s", port_id)
 
     def _safe_plug_new(self, network_id, port_id, device_name, mac_address,
-            bridge=None, namespace=None, prefix=None, mtu=None, link_up=True):
+            bridge=None, namespace=None, prefix=None, mtu=None):
         try:
             self.plug_new(
                 network_id, port_id, device_name, mac_address, bridge,
-                namespace, prefix, mtu, link_up)
+                namespace, prefix, mtu)
         except TypeError:
             LOG.warning("Interface driver's plug_new() method should now "
                         "accept additional optional parameter 'link_up'. "
@@ -321,27 +320,10 @@ class LinuxInterfaceDriver(object):
             LOG.warning("Interface driver cannot update MTU for ports")
             self._mtu_update_warn_logged = True
 
-    def set_link_status(self, device_name, namespace=None, link_up=True):
-        ns_dev = ip_lib.IPWrapper(namespace=namespace).device(device_name)
-        try:
-            utils.wait_until_true(ns_dev.exists, timeout=3)
-        except utils.WaitTimeout:
-            LOG.debug('Device %s may have been deleted concurrently',
-                      device_name)
-            return False
-
-        if link_up:
-            ns_dev.link.set_up()
-        else:
-            ns_dev.link.set_down()
-
-        return True
-
 
 class NullDriver(LinuxInterfaceDriver):
     def plug_new(self, network_id, port_id, device_name, mac_address,
-                 bridge=None, namespace=None, prefix=None, mtu=None,
-                 link_up=True):
+                 bridge=None, namespace=None, prefix=None, mtu=None):
         pass
 
     def unplug(self, device_name, bridge=None, namespace=None, prefix=None):
@@ -377,8 +359,7 @@ class OVSInterfaceDriver(LinuxInterfaceDriver):
         ovs.replace_port(device_name, *attrs)
 
     def plug_new(self, network_id, port_id, device_name, mac_address,
-                 bridge=None, namespace=None, prefix=None, mtu=None,
-                 link_up=True):
+                 bridge=None, namespace=None, prefix=None, mtu=None):
         """Plug in the interface."""
         if not bridge:
             bridge = self.conf.ovs_integration_bridge
@@ -442,8 +423,8 @@ class OVSInterfaceDriver(LinuxInterfaceDriver):
         else:
             LOG.warning("No MTU configured for port %s", port_id)
 
-        if link_up:
-            ns_dev.link.set_up()
+        ns_dev.link.set_up()
+
         if self.conf.ovs_use_veth:
             # ovs-dpdk does not do checksum calculations for veth interface
             # (bug 1832021)
@@ -488,8 +469,7 @@ class BridgeInterfaceDriver(LinuxInterfaceDriver):
     DEV_NAME_PREFIX = 'ns-'
 
     def plug_new(self, network_id, port_id, device_name, mac_address,
-                 bridge=None, namespace=None, prefix=None, mtu=None,
-                 link_up=True):
+                 bridge=None, namespace=None, prefix=None, mtu=None):
         """Plugin the interface."""
         ip = ip_lib.IPWrapper()
 
@@ -508,8 +488,7 @@ class BridgeInterfaceDriver(LinuxInterfaceDriver):
             LOG.warning("No MTU configured for port %s", port_id)
 
         root_veth.link.set_up()
-        if link_up:
-            ns_veth.link.set_up()
+        ns_veth.link.set_up()
 
     def unplug(self, device_name, bridge=None, namespace=None, prefix=None):
         """Unplug the interface."""
diff --git a/agent/linux/keepalived.py b/agent/linux/keepalived.py
index f47a27f1d1..405a781f0b 100644
--- a/agent/linux/keepalived.py
+++ b/agent/linux/keepalived.py
@@ -32,6 +32,7 @@ from neutron.common import utils
 VALID_STATES = ['MASTER', 'BACKUP']
 VALID_AUTH_TYPES = ['AH', 'PASS']
 HA_DEFAULT_PRIORITY = 50
+HA_DEFAULT_MASTER_PRIORITY = 100
 PRIMARY_VIP_RANGE_SIZE = 24
 KEEPALIVED_SERVICE_NAME = 'keepalived'
 KEEPALIVED_EMAIL_FROM = 'neutron@openstack.local'
diff --git a/api/rpc/agentnotifiers/dhcp_rpc_agent_api.py b/api/rpc/agentnotifiers/dhcp_rpc_agent_api.py
index 29b61b2c9b..fbad5f1d9a 100644
--- a/api/rpc/agentnotifiers/dhcp_rpc_agent_api.py
+++ b/api/rpc/agentnotifiers/dhcp_rpc_agent_api.py
@@ -312,6 +312,15 @@ class DhcpAgentNotifyAPI(object):
             return False
         if set(orig.keys()) != set(new.keys()):
             return False
+
+        if cfg.CONF.enable_set_route_for_single_port:
+            device_owner = new.get('device_owner', None)
+            orig_device_owner = orig.get('device_owner', None)
+            if (not orig_device_owner and device_owner and
+                    device_owner.startswith(
+                        constants.DEVICE_OWNER_COMPUTE_PREFIX)):
+                return True
+
         for k in orig.keys():
             if k in ('status', 'updated_at', 'revision_number'):
                 continue
diff --git a/api/rpc/callbacks/resources.py b/api/rpc/callbacks/resources.py
index 734f05eb6f..de56211c53 100644
--- a/api/rpc/callbacks/resources.py
+++ b/api/rpc/callbacks/resources.py
@@ -15,6 +15,7 @@ from neutron.objects import conntrack_helper
 from neutron.objects.logapi import logging_resource as log_object
 from neutron.objects import network
 from neutron.objects import port_forwarding
+from neutron.objects import rg_port_forwarding
 from neutron.objects import ports
 from neutron.objects.qos import policy
 from neutron.objects import securitygroup
@@ -33,6 +34,7 @@ SUBNET = subnet.Subnet.obj_name()
 SECURITYGROUP = securitygroup.SecurityGroup.obj_name()
 SECURITYGROUPRULE = securitygroup.SecurityGroupRule.obj_name()
 PORTFORWARDING = port_forwarding.PortForwarding.obj_name()
+RGPORTFORWARDING = rg_port_forwarding.RGPortForwarding.obj_name()
 CONNTRACKHELPER = conntrack_helper.ConntrackHelper.obj_name()
 
 
@@ -47,6 +49,7 @@ _VALID_CLS = (
     securitygroup.SecurityGroupRule,
     log_object.Log,
     port_forwarding.PortForwarding,
+    rg_port_forwarding.RGPortForwarding,
     conntrack_helper.ConntrackHelper,
 )
 
diff --git a/conf/agent/l3/keepalived.py b/conf/agent/l3/keepalived.py
index bd46c723fc..5ff3492280 100644
--- a/conf/agent/l3/keepalived.py
+++ b/conf/agent/l3/keepalived.py
@@ -20,6 +20,8 @@ from neutron._i18n import _
 
 CLI_OPTS = [
     cfg.StrOpt('router_id', help=_('ID of the router')),
+    cfg.StrOpt('master_agent', help=_('The master agent of router')),
+    cfg.ListOpt('slave_agents', help=_('The slave agents of router')),
     cfg.StrOpt('namespace', help=_('Namespace of the router')),
     cfg.StrOpt('conf_dir', help=_('Path to the router directory')),
     cfg.StrOpt('monitor_interface', help=_('Interface to monitor')),
diff --git a/conf/common.py b/conf/common.py
index f885429613..45e0b48723 100644
--- a/conf/common.py
+++ b/conf/common.py
@@ -145,7 +145,21 @@ core_opts = [
                       "Setting to any positive integer means that on failure "
                       "the connection is retried that many times. "
                       "For example, setting to 3 means total attempts to "
-                      "connect will be 4."))
+                      "connect will be 4.")),
+    cfg.BoolOpt('enable_set_route_for_single_port', default=False,
+                help=_("To set route path for every single port "
+                       "when the same subnet has multi ports on router.")),
+    cfg.ListOpt('network_nodes',
+                default=[],
+                help=_("The list of network hosts to "
+                       "make a network map "
+                       "with compute node and network node.")),
+]
+
+NETWORK_HOST_OPTS = [
+    cfg.ListOpt('compute_nodes',
+                default=[],
+                help=_("The list of compute hosts."))
 ]
 
 core_cli_opts = [
diff --git a/conf/policies/__init__.py b/conf/policies/__init__.py
index aa4dda63d0..15cdaea45a 100644
--- a/conf/policies/__init__.py
+++ b/conf/policies/__init__.py
@@ -34,6 +34,7 @@ from neutron.conf.policies import port
 from neutron.conf.policies import qos
 from neutron.conf.policies import rbac
 from neutron.conf.policies import router
+from neutron.conf.policies import rg_port_forwarding
 from neutron.conf.policies import security_group
 from neutron.conf.policies import segment
 from neutron.conf.policies import service_type
@@ -63,6 +64,7 @@ def list_rules():
         qos.list_rules(),
         rbac.list_rules(),
         router.list_rules(),
+        rg_port_forwarding.list_rules(),
         security_group.list_rules(),
         segment.list_rules(),
         service_type.list_rules(),
diff --git a/conf/policies/rg_port_forwarding.py b/conf/policies/rg_port_forwarding.py
new file mode 100644
index 0000000000..19e2cd5e2f
--- /dev/null
+++ b/conf/policies/rg_port_forwarding.py
@@ -0,0 +1,76 @@
+# Copyright (c) 2023 UnionTech
+# All rights reserved
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+
+from oslo_policy import policy
+from neutron.conf.policies import base
+
+COLLECTION_PATH = '/routers/{router_id}/port_forwardings'
+RESOURCE_PATH = '/routers/{router_id}/port_forwardings/{port_forwarding_id}'
+
+rules = [
+    policy.DocumentedRuleDefault(
+        'create_router_gateway_port_forwarding',
+        base.RULE_ADMIN_OR_PARENT_OWNER,
+        'Create a router gateway port forwarding',
+        [
+            {
+                'method': 'POST',
+                'path': COLLECTION_PATH,
+            },
+        ]
+    ),
+    policy.DocumentedRuleDefault(
+        'get_router_gateway_forwarding',
+        base.RULE_ADMIN_OR_PARENT_OWNER,
+        'Get a router gateway port forwarding',
+        [
+            {
+                'method': 'GET',
+                'path': COLLECTION_PATH,
+            },
+            {
+                'method': 'GET',
+                'path': RESOURCE_PATH,
+            },
+        ]
+    ),
+    policy.DocumentedRuleDefault(
+        'update_router_gateway_port_forwarding',
+        base.RULE_ADMIN_OR_PARENT_OWNER,
+        'Update a floating IP port forwarding',
+        [
+            {
+                'method': 'PUT',
+                'path': RESOURCE_PATH,
+            },
+        ]
+    ),
+    policy.DocumentedRuleDefault(
+        'delete_router_gateway_port_forwarding',
+        base.RULE_ADMIN_OR_PARENT_OWNER,
+        'Delete a floating IP port forwarding',
+        [
+            {
+                'method': 'DELETE',
+                'path': RESOURCE_PATH,
+            },
+        ]
+    ),
+]
+
+
+def list_rules():
+    return rules
diff --git a/db/l3_attrs_db.py b/db/l3_attrs_db.py
index e6d4e298b1..f292b7aa32 100644
--- a/db/l3_attrs_db.py
+++ b/db/l3_attrs_db.py
@@ -19,6 +19,7 @@ from oslo_config import cfg
 
 from neutron._i18n import _
 from neutron.db.models import l3_attrs
+from neutron.objects.base import NeutronDbObject
 
 
 def get_attr_info():
@@ -29,7 +30,11 @@ def get_attr_info():
             'availability_zone_hints': {
                 'default': '[]',
                 'transform_to_db': az_validator.convert_az_list_to_string,
-                'transform_from_db': az_validator.convert_az_string_to_list}
+                'transform_from_db': az_validator.convert_az_string_to_list},
+            'configurations': {
+                'default': '{}',
+                'transform_to_db': NeutronDbObject.filter_to_json_str,
+                'transform_from_db': NeutronDbObject.load_json_from_str}
             }
 
 
diff --git a/db/l3_db.py b/db/l3_db.py
index 565b422532..b625dc1959 100644
--- a/db/l3_db.py
+++ b/db/l3_db.py
@@ -47,6 +47,7 @@ from neutron.common import ipv6_utils
 from neutron.common import utils
 from neutron.db import _utils as db_utils
 from neutron.db.models import l3 as l3_models
+from neutron.plugins.ml2 import models as ml2_models
 from neutron.db import models_v2
 from neutron.db import standardattrdescription_db as st_attr
 from neutron.extensions import l3
@@ -1086,21 +1087,37 @@ class L3_NAT_dbonly_mixin(l3.RouterPluginBase,
         # with subnet's gateway-ip, return that router.
         # Otherwise return the first router.
         RouterPort = l3_models.RouterPort
+        RouterPortBinding = orm.aliased(ml2_models.PortBinding,
+                                        name="router_port_binding")
+        ComputePortBinding = orm.aliased(ml2_models.PortBinding,
+                                         name="compute_port_binding")
         gw_port = orm.aliased(models_v2.Port, name="gw_port")
         # TODO(lujinluo): Need IPAllocation and Port object
         routerport_qry = context.session.query(
-            RouterPort.router_id, models_v2.IPAllocation.ip_address).join(
-            RouterPort.port, models_v2.Port.fixed_ips).filter(
+            RouterPort.router_id, models_v2.IPAllocation.ip_address,
+            RouterPortBinding.host, ComputePortBinding.host,
+        ).join(
+            RouterPort.port, models_v2.Port.fixed_ips
+        ).filter(
             models_v2.Port.network_id == internal_port['network_id'],
             RouterPort.port_type.in_(constants.ROUTER_INTERFACE_OWNERS),
-            models_v2.IPAllocation.subnet_id == internal_subnet['id']
-        ).join(gw_port, gw_port.device_id == RouterPort.router_id).filter(
+            models_v2.IPAllocation.subnet_id == internal_subnet['id'],
+            ComputePortBinding.port_id == internal_port['id'],
+        ).join(
+            gw_port, gw_port.device_id == RouterPort.router_id
+        ).filter(
             gw_port.network_id == external_network_id,
             gw_port.device_owner == DEVICE_OWNER_ROUTER_GW
+        ).join(
+            RouterPortBinding, RouterPortBinding.port_id == models_v2.Port.id
         ).distinct()
 
         first_router_id = None
-        for router_id, interface_ip in routerport_qry:
+        for (router_id, interface_ip,
+             network_host, compute_host) in routerport_qry:
+            network_node = self.compute_to_network.get(compute_host, None)
+            if network_node and network_node == network_host:
+                return router_id
             if interface_ip == internal_subnet['gateway_ip']:
                 return router_id
             if not first_router_id:
diff --git a/db/l3_hamode_db.py b/db/l3_hamode_db.py
index bff388e166..4c414016dd 100644
--- a/db/l3_hamode_db.py
+++ b/db/l3_hamode_db.py
@@ -32,6 +32,7 @@ from neutron_lib import constants
 from neutron_lib.db import api as db_api
 from neutron_lib import exceptions as n_exc
 from neutron_lib.exceptions import l3 as l3_exc
+from neutron_lib.exceptions import agent as agent_exc
 from neutron_lib.exceptions import l3_ext_ha_mode as l3ha_exc
 from neutron_lib.objects import exceptions as obj_base
 from neutron_lib.plugins import utils as p_utils
@@ -54,6 +55,7 @@ from neutron.db import l3_dvr_db
 from neutron.objects import base
 from neutron.objects import l3_hamode
 from neutron.objects import router as l3_obj
+from neutron.objects import agent as agent_obj
 
 
 VR_ID_RANGE = set(range(1, 255))
@@ -378,12 +380,85 @@ class L3_HA_NAT_db_mixin(l3_dvr_db.L3_NAT_with_dvr_db_mixin,
         if not self.get_ha_network(context, router['tenant_id']):
             self._create_ha_network(context, router['tenant_id'])
 
+    @staticmethod
+    def _check_router_configurations_creation(context,
+                                              configurations: dict,
+                                              is_ha: bool = True):
+        agents = agent_obj.Agent.get_objects(context,
+                                             binary='neutron-l3-agent')
+        agents = [agent['host'] for agent in agents]
+        master = configurations.get('master_agent', None)
+        slaves = configurations.get('slave_agents', [])
+        preferred_agent = configurations.get('preferred_agent', None)
+        if is_ha:
+            if master and slaves:
+                for agent in [master] + slaves:
+                    if agent not in agents:
+                        raise agent_exc.AgentNotFound(id=agent)
+                if master in slaves:
+                    raise l3_exc.RouterAgentConflict()
+            else:
+                if master or slaves:
+                    raise l3_exc.RouterAgentNotGiven()
+        else:
+            if preferred_agent and preferred_agent not in agents:
+                raise agent_exc.AgentNotFound(id=preferred_agent)
+
+    @staticmethod
+    def _check_router_configurations_update(context,
+                                            configurations: dict,
+                                            old_configurations: dict,
+                                            is_ha: bool = True) -> bool:
+        if configurations == old_configurations:
+            return False
+
+        agents = agent_obj.Agent.get_objects(context,
+                                             binary='neutron-l3-agent')
+        agents = [agent['host'] for agent in agents]
+        master = configurations.get('master_agent', None)
+        slaves = configurations.get('slave_agents', [])
+        preferred_agent = configurations.get('preferred_agent', None)
+        old_master = old_configurations.get('master_agent', None)
+        old_slaves = old_configurations.get('slave_agents', [])
+        old_preferred_agent = old_configurations.get('preferred_agent', None)
+        if is_ha:
+            if master:
+                if master != old_master:
+                    if master not in agents:
+                        raise agent_exc.AgentNotFound(id=master)
+                    old_configurations['master_agent'] = master
+            if slaves:
+                if slaves != old_slaves:
+                    for slave in slaves:
+                        if slave not in agents:
+                            raise agent_exc.AgentNotFound(id=slave)
+                    old_configurations['slave_agents'] = slaves
+            if (old_configurations['master_agent'] in
+                    old_configurations['slave_agents']):
+                raise l3_exc.RouterAgentConflict()
+        else:
+            if preferred_agent:
+                if preferred_agent != old_preferred_agent:
+                    if preferred_agent not in agents:
+                        raise agent_exc.AgentNotFound(id=preferred_agent)
+                    old_configurations['preferred_agent'] = preferred_agent
+            else:
+                old_configurations['preferred_agent'] = None
+
+        return True
+
     @registry.receives(resources.ROUTER, [events.PRECOMMIT_CREATE],
                        priority_group.PRIORITY_ROUTER_EXTENDED_ATTRIBUTE)
     def _precommit_router_create(self, resource, event, trigger, context,
                                  router, router_db, **kwargs):
         """Event handler to set ha flag and status on creation."""
         is_ha = self._is_ha(router)
+        configurations = router.get('configurations', {})
+        if configurations:
+            self._check_router_configurations_creation(context, configurations,
+                                                       is_ha)
+        self.set_extra_attr_value(context, router_db, 'configurations',
+                                  configurations)
         router['ha'] = is_ha
         self.set_extra_attr_value(context, router_db, 'ha', is_ha)
         if not is_ha:
@@ -465,6 +540,28 @@ class L3_HA_NAT_db_mixin(l3_dvr_db.L3_NAT_with_dvr_db_mixin,
         self.set_extra_attr_value(
             payload.context, payload.desired_state, 'ha', requested_ha_state)
 
+    @registry.receives(resources.ROUTER, [events.PRECOMMIT_UPDATE],
+                       priority_group.PRIORITY_ROUTER_EXTENDED_ATTRIBUTE)
+    def _validate_configurations(self, resource, event, trigger, payload=None):
+        old_configurations = payload.states[0].get('configurations', {})
+        configurations = payload.request_body.get('configurations', {})
+
+        if not configurations:
+            return
+
+        if payload.desired_state.admin_state_up:
+            msg = _('Cannot change configurations of active routers. Please '
+                    'set router admin_state_up to False prior to upgrade')
+            raise n_exc.BadRequest(resource='router', msg=msg)
+
+        need_update = self._check_router_configurations_update(
+            payload.context, configurations, old_configurations,
+            payload.states[0]['ha'])
+
+        if need_update:
+            self.set_extra_attr_value(payload.context, payload.desired_state,
+                                      'configurations', old_configurations)
+
     @registry.receives(resources.ROUTER, [events.AFTER_UPDATE],
                        priority_group.PRIORITY_ROUTER_EXTENDED_ATTRIBUTE)
     def _reconfigure_ha_resources(self, resource, event, trigger, context,
diff --git a/db/migration/alembic_migrations/versions/EXPAND_HEAD b/db/migration/alembic_migrations/versions/EXPAND_HEAD
index ffa2bbaaf6..0c8e4a2178 100644
--- a/db/migration/alembic_migrations/versions/EXPAND_HEAD
+++ b/db/migration/alembic_migrations/versions/EXPAND_HEAD
@@ -1 +1 @@
-c613d0b82681
+1c19a98b5eef
diff --git a/db/migration/alembic_migrations/versions/train/expand/1c19a98b5eef_add_router_configurations.py b/db/migration/alembic_migrations/versions/train/expand/1c19a98b5eef_add_router_configurations.py
new file mode 100644
index 0000000000..f12600ef4f
--- /dev/null
+++ b/db/migration/alembic_migrations/versions/train/expand/1c19a98b5eef_add_router_configurations.py
@@ -0,0 +1,36 @@
+# Copyright 2023 OpenStack Foundation
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+#
+
+from alembic import op
+import sqlalchemy as sa
+
+"""add router configurations
+
+Revision ID: 1c19a98b5eef
+Revises: cab12b72ed90
+Create Date: 2023-08-01 10:05:56.412167
+
+"""
+
+# revision identifiers, used by Alembic.
+revision = '1c19a98b5eef'
+down_revision = 'cab12b72ed90'
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column('router_extra_attributes',
+                  sa.Column('configurations', sa.String(length=4095)))
+    # ### end Alembic commands ###
diff --git a/db/migration/alembic_migrations/versions/train/expand/cab12b72ed90_add_router_gateway_port_forwarding.py b/db/migration/alembic_migrations/versions/train/expand/cab12b72ed90_add_router_gateway_port_forwarding.py
new file mode 100644
index 0000000000..ad511a7ed8
--- /dev/null
+++ b/db/migration/alembic_migrations/versions/train/expand/cab12b72ed90_add_router_gateway_port_forwarding.py
@@ -0,0 +1,55 @@
+# Copyright 2023 OpenStack Foundation
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+#
+
+from alembic import op
+import sqlalchemy as sa
+
+"""add router gateway port forwarding
+
+Revision ID: cab12b72ed90
+Revises: c613d0b82681
+Create Date: 2023-07-04 10:27:54.485453
+
+"""
+
+# revision identifiers, used by Alembic.
+revision = 'cab12b72ed90'
+down_revision = 'c613d0b82681'
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table(
+        'rgportforwardings',
+        sa.Column('id', sa.String(length=36), nullable=False),
+        sa.Column('router_id', sa.String(length=36), nullable=False),
+        sa.Column('external_port', sa.Integer(), nullable=False),
+        sa.Column('internal_neutron_port_id', sa.String(length=36),
+                  nullable=False),
+        sa.Column('protocol', sa.String(length=40), nullable=False),
+        sa.Column('socket', sa.String(length=36), nullable=False),
+        sa.ForeignKeyConstraint(['internal_neutron_port_id'], ['ports.id'],
+                                ondelete='CASCADE'),
+        sa.ForeignKeyConstraint(['router_id'], ['routers.id'],
+                                ondelete='CASCADE'),
+        sa.PrimaryKeyConstraint('id'),
+        sa.UniqueConstraint(
+            'internal_neutron_port_id', 'socket', 'protocol',
+            name='uniq_port_forwardings0internal_neutron_port_id0socket0protocol'),
+        sa.UniqueConstraint(
+            'router_id', 'external_port', 'protocol',
+            name='uniq_rg_port_forwardings0router_id0external_port0protocol')
+    )
+    # ### end Alembic commands ###
diff --git a/db/models/l3_attrs.py b/db/models/l3_attrs.py
index 6c30ac2c16..904f4ef08d 100644
--- a/db/models/l3_attrs.py
+++ b/db/models/l3_attrs.py
@@ -41,6 +41,8 @@ class RouterExtraAttributes(model_base.BASEV2):
     # Availability Zone support
     availability_zone_hints = sa.Column(sa.String(255))
 
+    configurations = sa.Column(sa.String(4095))
+
     router = orm.relationship(
         'Router', load_on_pending=True,
         backref=orm.backref("extra_attributes", lazy='joined',
diff --git a/db/models/rg_port_forwarding.py b/db/models/rg_port_forwarding.py
new file mode 100644
index 0000000000..e7963169a8
--- /dev/null
+++ b/db/models/rg_port_forwarding.py
@@ -0,0 +1,59 @@
+# Copyright (c) 2023 UnionTech
+# All rights reserved
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+import sqlalchemy as sa
+from sqlalchemy import orm
+from neutron_lib.db import model_base
+from neutron_lib.db import constants as db_const
+
+from neutron.db.models import l3
+from neutron.db import models_v2
+
+
+class RGPortForwarding(model_base.BASEV2, model_base.HasId):
+    __table_args__ = (
+        sa.UniqueConstraint('router_id', 'external_port', 'protocol',
+                            name='uniq_rg_port_forwardings0router_id0'
+                                 'external_port0protocol'),
+        sa.UniqueConstraint('internal_neutron_port_id', 'socket', 'protocol',
+                            name='uniq_port_forwardings0'
+                                 'internal_neutron_port_id0socket0'
+                                 'protocol')
+    )
+
+    router_id = sa.Column(sa.String(db_const.UUID_FIELD_SIZE),
+                          sa.ForeignKey('routers.id',
+                                        ondelete="CASCADE"),
+                          nullable=False)
+    external_port = sa.Column(sa.Integer, nullable=False)
+    internal_neutron_port_id = sa.Column(
+        sa.String(db_const.UUID_FIELD_SIZE),
+        sa.ForeignKey('ports.id', ondelete="CASCADE"),
+        nullable=False)
+    protocol = sa.Column(sa.String(40), nullable=False)
+    socket = sa.Column(sa.String(36), nullable=False)
+    port = orm.relationship(
+        models_v2.Port, load_on_pending=True,
+        backref=orm.backref("rg_port_forwardings",
+                            lazy='subquery', uselist=True,
+                            cascade='delete')
+    )
+    router = orm.relationship(
+        l3.Router, load_on_pending=True,
+        backref=orm.backref("rg_port_forwardings",
+                            lazy='subquery', uselist=True,
+                            cascade='delete')
+    )
+    revises_on_change = ('router', 'port',)
diff --git a/extensions/rg_port_forwarding.py b/extensions/rg_port_forwarding.py
new file mode 100644
index 0000000000..c9888cd1ec
--- /dev/null
+++ b/extensions/rg_port_forwarding.py
@@ -0,0 +1,119 @@
+# Copyright (c) 2023 UnionTech
+# All rights reserved
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+import six
+import abc
+import itertools
+from typing import List
+
+from neutron_lib.context import Context
+from neutron_lib.plugins import constants
+from neutron_lib.plugins import directory
+from neutron_lib.services import base as service_base
+from neutron_lib.api.extensions import APIExtensionDescriptor
+from neutron_lib.api.definitions import rg_port_forwarding as apidef
+
+from neutron.api.v2 import base
+from neutron.api.v2 import resource_helper
+from neutron.api.extensions import ResourceExtension
+
+
+class Rg_port_forwarding(APIExtensionDescriptor):
+    api_definition = apidef
+
+    @classmethod
+    def get_plugin_interface(cls):
+        return RGPortForwardingPluginBase
+
+    @classmethod
+    def get_resources(cls):
+        special_mappings = {'routers': 'router'}
+        plural_mappings = resource_helper.build_plural_mappings(
+            special_mappings,
+            itertools.chain(
+                apidef.RESOURCE_ATTRIBUTE_MAP,
+                apidef.SUB_RESOURCE_ATTRIBUTE_MAP
+            )
+        )
+
+        resources = resource_helper.build_resource_info(
+            plural_mappings,
+            apidef.RESOURCE_ATTRIBUTE_MAP,
+            constants.ROUTER_GATEWAY_PORTFORWARDING,
+            translate_name=True,
+            allow_bulk=True)
+
+        plugin = directory.get_plugin(constants.ROUTER_GATEWAY_PORTFORWARDING)
+
+        parent = apidef.SUB_RESOURCE_ATTRIBUTE_MAP[apidef.COLLECTION_NAME].get(
+            'parent')
+        params = apidef.SUB_RESOURCE_ATTRIBUTE_MAP[apidef.COLLECTION_NAME].get(
+            'parameters')
+        controller = base.create_resource(
+            apidef.COLLECTION_NAME, apidef.RESOURCE_NAME, plugin, params,
+            allow_bulk=True, parent=parent, allow_pagination=True,
+            allow_sorting=True)
+
+        resource = ResourceExtension(
+            apidef.COLLECTION_NAME, controller, parent, attr_map=params)
+        resources.append(resource)
+
+        return resources
+
+
+@six.add_metaclass(abc.ABCMeta)
+class RGPortForwardingPluginBase(service_base.ServicePluginBase):
+    path_prefix = apidef.API_PREFIX
+
+    @classmethod
+    def get_plugin_type(cls):
+        return constants.ROUTER_GATEWAY_PORTFORWARDING
+
+    def get_plugin_description(self):
+        return "Router Gateway Port Forwarding Service Plugin"
+
+    @abc.abstractmethod
+    def create_router_gateway_port_forwarding(self, context: Context,
+                                              router_id: str,
+                                              gateway_port_forwarding: dict):
+        pass
+
+    @abc.abstractmethod
+    def update_router_gateway_port_forwarding(self, context: Context, id: str,
+                                              router_id: str,
+                                              gateway_port_forwarding: dict):
+        pass
+
+    @abc.abstractmethod
+    def get_router_gateway_port_forwarding(self, context: Context, id: str,
+                                           router_id: str,
+                                           fields: List[str] = None):
+        pass
+
+    @abc.abstractmethod
+    def get_router_gateway_port_forwardings(self, context: Context,
+                                            router_id: str,
+                                            filters: List[str] = None,
+                                            fields: List[str] = None,
+                                            sorts: List[str] = None,
+                                            limit: int = None,
+                                            marker: str = None,
+                                            page_reverse: bool = False):
+        pass
+
+    @abc.abstractmethod
+    def delete_router_gateway_port_forwarding(self, context: Context, id: str,
+                                              router_id: str):
+        pass
diff --git a/objects/rg_port_forwarding.py b/objects/rg_port_forwarding.py
new file mode 100644
index 0000000000..28cb4e1d4d
--- /dev/null
+++ b/objects/rg_port_forwarding.py
@@ -0,0 +1,87 @@
+# Copyright (c) 2023 UnionTech
+# All rights reserved
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+import netaddr
+from neutron_lib import constants as lib_const
+from oslo_versionedobjects import fields as obj_fields
+
+from neutron.objects import base, common_types
+from neutron.db.models import rg_port_forwarding as models
+
+FIELDS_NOT_SUPPORT_FILTER = ['internal_ip_address', 'internal_port']
+
+
+@base.NeutronObjectRegistry.register
+class RGPortForwarding(base.NeutronDbObject):
+    VERSION = '1.0'
+
+    db_model = models.RGPortForwarding
+
+    primary_keys = ['id']
+    foreign_keys = {
+        'Router': {'router_id': 'id'},
+        'Port': {'internal_port_id': 'id'}
+    }
+    fields_need_translation = {
+        'socket': 'socket',
+        'internal_port_id': 'internal_neutron_port_id'
+    }
+
+    fields = {
+        'id': common_types.UUIDField(),
+        'router_id': common_types.UUIDField(nullable=False),
+        'external_port': common_types.PortRangeField(nullable=False),
+        'protocol': common_types.IpProtocolEnumField(nullable=False),
+        'internal_port_id': common_types.UUIDField(nullable=False),
+        'internal_ip_address': obj_fields.IPV4AddressField(),
+        'internal_port': common_types.PortRangeField(nullable=False),
+        'gw_ip_address': obj_fields.IPV4AddressField(),
+    }
+
+    synthetic_fields = ['gw_ip_address']
+    fields_no_update = {'id', 'router_id'}
+
+    def __eq__(self, other):
+        for attr in self.fields:
+            if getattr(self, attr) != getattr(other, attr):
+                return False
+        return True
+
+    def obj_load_attr(self, attrname):
+        super(RGPortForwarding, self).obj_load_attr(attrname)
+
+    def from_db_object(self, db_obj):
+        super(RGPortForwarding, self).from_db_object(db_obj)
+
+    @classmethod
+    def modify_fields_from_db(cls, db_obj):
+        result = super(RGPortForwarding, cls).modify_fields_from_db(db_obj)
+        if 'socket' in result:
+            groups = result['socket'].split(":")
+            result['internal_ip_address'] = netaddr.IPAddress(
+                groups[0], version=lib_const.IP_VERSION_4)
+            result['internal_port'] = int(groups[1])
+            del result['socket']
+        return result
+
+    @classmethod
+    def modify_fields_to_db(cls, fields):
+        result = super(RGPortForwarding, cls).modify_fields_to_db(fields)
+        if 'internal_ip_address' in result and 'internal_port' in result:
+            result['socket'] = (f"{result['internal_ip_address']}:"
+                                f"{result['internal_port']}")
+            del result['internal_ip_address']
+            del result['internal_port']
+        return result
diff --git a/objects/router.py b/objects/router.py
index 1373f89515..9590a109f6 100644
--- a/objects/router.py
+++ b/objects/router.py
@@ -18,6 +18,7 @@ from neutron_lib.api.definitions import availability_zone as az_def
 from neutron_lib.api.validators import availability_zone as az_validator
 from neutron_lib import constants as n_const
 from neutron_lib.utils import net as net_utils
+from neutron_lib.objects import utils as obj_utils
 from oslo_versionedobjects import fields as obj_fields
 import six
 from sqlalchemy import func
@@ -70,7 +71,8 @@ class RouterRoute(base.NeutronDbObject):
 @base.NeutronObjectRegistry.register
 class RouterExtraAttributes(base.NeutronDbObject):
     # Version 1.0: Initial version
-    VERSION = '1.0'
+    # Version 1.1: Add configurations
+    VERSION = '1.1'
 
     db_model = l3_attrs.RouterExtraAttributes
 
@@ -80,7 +82,8 @@ class RouterExtraAttributes(base.NeutronDbObject):
         'service_router': obj_fields.BooleanField(default=False),
         'ha': obj_fields.BooleanField(default=False),
         'ha_vr_id': obj_fields.IntegerField(nullable=True),
-        'availability_zone_hints': obj_fields.ListOfStringsField(nullable=True)
+        'availability_zone_hints': obj_fields.ListOfStringsField(nullable=True),
+        'configurations': common_types.DictOfMiscValuesField(nullable=True),
     }
 
     primary_keys = ['router_id']
@@ -95,6 +98,9 @@ class RouterExtraAttributes(base.NeutronDbObject):
             result[az_def.AZ_HINTS] = (
                 az_validator.convert_az_string_to_list(
                     result[az_def.AZ_HINTS]))
+        if 'configurations' in result:
+            result['configurations'] = cls.load_json_from_str(
+                result['configurations'], default={})
         return result
 
     @classmethod
@@ -104,6 +110,11 @@ class RouterExtraAttributes(base.NeutronDbObject):
             result[az_def.AZ_HINTS] = (
                 az_validator.convert_az_list_to_string(
                     result[az_def.AZ_HINTS]))
+        if ('configurations' in result and
+                not isinstance(result['configurations'],
+                               obj_utils.StringMatchingFilterObj)):
+            result['configurations'] = (
+                cls.filter_to_json_str(result['configurations']))
         return result
 
     @classmethod
diff --git a/scheduler/l3_agent_scheduler.py b/scheduler/l3_agent_scheduler.py
index 5810cf85b8..7a428aef03 100644
--- a/scheduler/l3_agent_scheduler.py
+++ b/scheduler/l3_agent_scheduler.py
@@ -14,13 +14,15 @@
 #    under the License.
 
 import abc
-import collections
+import random
 import functools
 import itertools
-import random
+import collections
+from typing import List, Optional
 
 from neutron_lib.api.definitions import availability_zone as az_def
 from neutron_lib import constants as lib_const
+from neutron_lib.context import Context
 from neutron_lib.db import api as lib_db_api
 from neutron_lib.exceptions import l3 as l3_exc
 from oslo_config import cfg
@@ -31,8 +33,9 @@ import six
 from neutron.common import utils
 from neutron.conf.db import l3_hamode_db
 from neutron.db.models import l3agent as rb_model
+from neutron.objects.agent import Agent
 from neutron.objects import l3agent as rb_obj
-
+from neutron.services.l3_router.l3_router_plugin import L3RouterPlugin
 
 LOG = logging.getLogger(__name__)
 cfg.CONF.register_opts(l3_hamode_db.L3_HA_OPTS)
@@ -228,26 +231,25 @@ class L3Scheduler(object):
         if not candidates:
             return
         elif sync_router.get('ha', False):
-            chosen_agents = self._bind_ha_router(plugin, context,
-                                                 router_id,
-                                                 sync_router.get('tenant_id'),
+            chosen_agents = self._bind_ha_router(plugin, context, sync_router,
                                                  candidates)
             if not chosen_agents:
                 return
             chosen_agent = chosen_agents[-1]
         else:
             chosen_agent = self._choose_router_agent(
-                plugin, context, candidates)
+                context, plugin, candidates, sync_router)
             self.bind_router(plugin, context, router_id, chosen_agent.id)
         return chosen_agent
 
     @abc.abstractmethod
-    def _choose_router_agent(self, plugin, context, candidates):
+    def _choose_router_agent(self, context, plugin, candidates, sync_router):
         """Choose an agent from candidates based on a specific policy."""
         pass
 
     @abc.abstractmethod
-    def _choose_router_agents_for_ha(self, plugin, context, candidates):
+    def _choose_router_agents_for_ha(self, context, plugin, candidates,
+                                     sync_router):
         """Choose agents from candidates based on a specific policy."""
         pass
 
@@ -315,19 +317,19 @@ class L3Scheduler(object):
         hosting_list = [tuple(host) for host in hosting]
         return list(set(candidates) - set(hosting_list))
 
-    def _bind_ha_router(self, plugin, context, router_id,
-                        tenant_id, candidates):
+    def _bind_ha_router(self, plugin, context, sync_router, candidates):
         """Bind a HA router to agents based on a specific policy."""
-
+        router_id = sync_router.get('id')
+        tenant_id = sync_router.get('tenant_id')
         candidates = self._filter_scheduled_agents(plugin, context, router_id,
                                                    candidates)
 
         chosen_agents = self._choose_router_agents_for_ha(
-            plugin, context, candidates)
+            context, plugin, candidates, sync_router)
 
         for agent in chosen_agents:
-            self.create_ha_port_and_bind(plugin, context, router_id,
-                                         tenant_id, agent)
+            self.create_ha_port_and_bind(plugin, context, router_id, tenant_id,
+                                         agent)
 
         return chosen_agents
 
@@ -335,10 +337,11 @@ class L3Scheduler(object):
 class ChanceScheduler(L3Scheduler):
     """Randomly allocate an L3 agent for a router."""
 
-    def _choose_router_agent(self, plugin, context, candidates):
+    def _choose_router_agent(self, context, plugin, candidates, sync_router):
         return random.choice(candidates)
 
-    def _choose_router_agents_for_ha(self, plugin, context, candidates):
+    def _choose_router_agents_for_ha(self, context, plugin, candidates,
+                                     sync_router):
         num_agents = self._get_num_of_agents_for_ha(len(candidates))
         return random.sample(candidates, num_agents)
 
@@ -346,13 +349,14 @@ class ChanceScheduler(L3Scheduler):
 class LeastRoutersScheduler(L3Scheduler):
     """Allocate to an L3 agent with the least number of routers bound."""
 
-    def _choose_router_agent(self, plugin, context, candidates):
+    def _choose_router_agent(self, context, plugin, candidates, sync_router):
         candidate_ids = [candidate['id'] for candidate in candidates]
         chosen_agent = plugin.get_l3_agent_with_min_routers(
             context, candidate_ids)
         return chosen_agent
 
-    def _choose_router_agents_for_ha(self, plugin, context, candidates):
+    def _choose_router_agents_for_ha(self, context, plugin, candidates,
+                                     sync_router):
         num_agents = self._get_num_of_agents_for_ha(len(candidates))
         ordered_agents = plugin.get_l3_agents_ordered_by_num_routers(
             context, [candidate['id'] for candidate in candidates])
@@ -397,7 +401,8 @@ class AZLeastRoutersScheduler(LeastRoutersScheduler):
 
         return candidates
 
-    def _choose_router_agents_for_ha(self, plugin, context, candidates):
+    def _choose_router_agents_for_ha(self, context, plugin, candidates,
+                                     sync_router):
         ordered_agents = plugin.get_l3_agents_ordered_by_num_routers(
             context, [candidate['id'] for candidate in candidates])
         num_agents = self._get_num_of_agents_for_ha(len(ordered_agents))
@@ -416,3 +421,65 @@ class AZLeastRoutersScheduler(LeastRoutersScheduler):
             if len(selected_agents) >= num_agents:
                 break
         return selected_agents
+
+
+class PreferredL3AgentRoutersScheduler(LeastRoutersScheduler):
+
+    @staticmethod
+    def get_preferred_agent(sync_router: dict) -> Optional[str]:
+        configurations = sync_router.get('configurations', {})
+        if configurations:
+            return configurations.get('preferred_agent', None)
+        return None
+
+    @staticmethod
+    def get_agents(sync_router: dict) -> Optional[List[str]]:
+        configurations = sync_router.get('configurations', {})
+        if configurations:
+            master = configurations.get('master_agent', None)
+            slaves = configurations.get('slave_agents', [])
+            if master and slaves:
+                return slaves + [master]
+        return []
+
+    def _choose_router_agent(self, context: Context,
+                             plugin: L3RouterPlugin,
+                             candidates: List[Agent],
+                             sync_router: dict) -> Agent:
+        agent = self.get_preferred_agent(sync_router)
+        if agent:
+            new_candidates = [candidate for candidate in candidates
+                              if candidate['host'] == agent]
+            if not new_candidates:
+                LOG.warning(f"Router {sync_router['id']} failed to "
+                            f"schedule l3 agent on {agent}.")
+            else:
+                agent = new_candidates[0]
+                LOG.debug(f"Router {sync_router['id']} l3 agent is {agent}.")
+                return agent
+        agent = super()._choose_router_agent(context, plugin, candidates,
+                                             sync_router)
+        return agent
+
+    def _choose_router_agents_for_ha(self, context: Context,
+                                     plugin: L3RouterPlugin,
+                                     candidates: List[Agent],
+                                     sync_router: dict) -> List[Agent]:
+
+        agents = self.get_agents(sync_router)
+        if agents:
+            if self.max_ha_agents < len(agents):
+                agents = agents[len(agents) - self.max_ha_agents:]
+            new_candidates = [candidate for candidate in candidates if
+                              candidate['host'] in agents]
+            if len(new_candidates) != len(agents):
+                LOG.warning(f"Router {sync_router['id']} failed to "
+                            f"schedule l3 agents on {agents}.")
+            else:
+                LOG.debug(f"Router {sync_router['id']} l3 agents are "
+                          f"{new_candidates}.")
+                return new_candidates
+
+        return super(
+            PreferredL3AgentRoutersScheduler, self
+        )._choose_router_agents_for_ha(context, plugin, candidates, sync_router)
diff --git a/services/l3_router/l3_router_plugin.py b/services/l3_router/l3_router_plugin.py
index 2e8a762764..9825138261 100644
--- a/services/l3_router/l3_router_plugin.py
+++ b/services/l3_router/l3_router_plugin.py
@@ -38,6 +38,7 @@ from oslo_utils import importutils
 
 from neutron.api.rpc.agentnotifiers import l3_rpc_agent_api
 from neutron.api.rpc.handlers import l3_rpc
+from neutron.conf.common import NETWORK_HOST_OPTS
 from neutron.db import dns_db
 from neutron.db import extraroute_db
 from neutron.db import l3_dvr_ha_scheduler_db
@@ -135,6 +136,17 @@ class L3RouterPlugin(service_base.ServicePluginBase,
 
         self.add_worker(rpc_worker)
         self.l3_driver_controller = driver_controller.DriverController(self)
+        self.compute_to_network = dict()
+        self._init_compute_to_network()
+
+    def _init_compute_to_network(self):
+        for network_node in cfg.CONF.network_nodes:
+            cfg.CONF.register_opts(NETWORK_HOST_OPTS, group=network_node)
+            network_group = cfg.CONF.get(network_node, None)
+            if network_group:
+                compute_nodes = network_group.get('compute_nodes', [])
+                for compute_node in compute_nodes:
+                    self.compute_to_network[compute_node] = network_node
 
     @property
     def supported_extension_aliases(self):
diff --git a/services/rg_portforwarding/__init__.py b/services/rg_portforwarding/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/services/rg_portforwarding/common/__init__.py b/services/rg_portforwarding/common/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/services/rg_portforwarding/common/exceptions.py b/services/rg_portforwarding/common/exceptions.py
new file mode 100644
index 0000000000..73cea68e32
--- /dev/null
+++ b/services/rg_portforwarding/common/exceptions.py
@@ -0,0 +1,77 @@
+# Copyright (c) 2023 UnionTech
+# All rights reserved
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+from neutron._i18n import _
+from neutron_lib import exceptions
+
+
+class PortForwardingNotSupportFilterField(exceptions.BadRequest):
+    message = _("Port Forwarding filter %(filter)s is not supported.")
+
+
+class RouterDoesNotHaveGateway(exceptions.BadRequest):
+    message = _("Router %(router_id)s does not have any gateways.")
+
+
+class RouterGatewayPortNotFound(exceptions.NotFound):
+    message = _("Router %(router_id)s 's gateway port %(gw_port_id)s "
+                "could not be found.")
+
+
+class RouterGatewayPortDoesNotHaveAnyIPAddresses(exceptions.NotFound):
+    message = _("Router %(router_id)s 's gateway port %(gw_port_id)s "
+                "does not have any IP addresses.")
+
+
+class RouterGatewayPortForwardingNotFound(exceptions.NotFound):
+    message = _("Router Gateway Port Forwarding %(id)s could not be found.")
+
+
+class PortHasBindingFloatingIP(exceptions.InUse):
+    message = _("Cannot create port forwarding to floating IP "
+                "%(floating_ip_address)s (%(fip_id)s) with port %(port_id)s "
+                "using fixed IP %(fixed_ip)s, as that port already "
+                "has a binding floating IP.")
+
+
+class InconsistentPortAndIP(exceptions.BadRequest):
+    message = _("Port %(port_id)s does not have ip address %(ip_address)s.")
+
+
+class RouterGatewayPortForwardingAlreadyExists(exceptions.BadRequest):
+    message = _("A duplicate router gateway port forwarding entry "
+                "with same attributes already exists, "
+                "conflicting values are %(conflict)s.")
+
+
+class PortNetworkNotBindOnRouter(exceptions.BadRequest):
+    message = _("Port %(port_id)s 's network %(network_id)s "
+                "not bind on router %(router_id)s.")
+
+
+class RouterGatewayPortForwardingUpdateFailed(exceptions.BadRequest):
+    message = _("Another router port forwarding entry with the same "
+                "attributes already exists, conflicting "
+                "values are %(conflict)s.")
+
+
+class DeletedRouterWithRGForwarding(exceptions.InUse):
+    message = _("Cant not delete router, "
+                "router %(router_id)s has port forwardings to remove.")
+
+
+class DeletedRouterGatewayWithRGForwarding(exceptions.InUse):
+    message = _("Cant not delete or update router gateway, "
+                "router %(router_id)s has port forwardings to remove.")
diff --git a/services/rg_portforwarding/pf_plugin.py b/services/rg_portforwarding/pf_plugin.py
new file mode 100644
index 0000000000..ed8e68e53c
--- /dev/null
+++ b/services/rg_portforwarding/pf_plugin.py
@@ -0,0 +1,369 @@
+# Copyright (c) 2023 UnionTech
+# All rights reserved
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+
+from oslo_log import log as logging
+from typing import List, Dict, Optional
+
+from neutron_lib import constants
+from neutron_lib.context import Context
+from neutron_lib.plugins import directory
+from neutron_lib.db import resource_extend
+from neutron_lib.plugins.constants import L3
+from neutron_lib.db.api import CONTEXT_WRITER
+from neutron_lib.exceptions import PortNotFound
+from neutron_lib.exceptions.l3 import RouterNotFound
+from neutron_lib.callbacks import registry, resources
+from neutron_lib.callbacks import events as lib_events
+from neutron_lib.callbacks.events import DBEventPayload
+from neutron_lib.api.definitions import rg_port_forwarding as apidef
+from neutron_lib.objects.exceptions import NeutronDbObjectDuplicateEntry
+
+from neutron.db import db_base_plugin_common
+from neutron.db.l3_dvr_db import is_distributed_router
+from neutron.db.l3_hamode_db import is_ha_router
+
+from neutron.objects.base import Pager
+from neutron.objects.ports import Port
+from neutron.objects.router import Router, FloatingIP
+from neutron.objects.rg_port_forwarding import RGPortForwarding
+from neutron.objects.rg_port_forwarding import FIELDS_NOT_SUPPORT_FILTER
+
+from neutron.extensions.rg_port_forwarding import RGPortForwardingPluginBase
+from neutron.services.l3_router.l3_router_plugin import L3RouterPlugin
+from neutron.services.rg_portforwarding.common import exceptions
+
+from neutron.api.rpc.callbacks import events
+from neutron.api.rpc.handlers import resources_rpc
+
+LOG = logging.getLogger(__name__)
+
+
+@resource_extend.has_resource_extenders
+@registry.has_registry_receivers
+class RGPortForwardingPlugin(RGPortForwardingPluginBase):
+    required_service_plugins = ['router']
+
+    supported_extension_aliases = [apidef.ALIAS]
+
+    __native_pagination_support = True
+    __native_sorting_support = True
+    __filter_validation_support = True
+
+    def __init__(self):
+        super(RGPortForwardingPlugin, self).__init__()
+        self.push_api = resources_rpc.ResourcesPushRpcApi()
+        self.l3_plugin = directory.get_plugin(L3)
+        self.core_plugin = directory.get_plugin()
+
+    @staticmethod
+    def _get_router(context: Context, router_id: str) -> Optional[Router]:
+        router = Router.get_object(context, id=router_id)
+        if not router:
+            raise RouterNotFound(router_id=router_id)
+        return router
+
+    @staticmethod
+    def _get_router_gateway(context: Context, router: Router) -> str:
+        gw_port_id = router.get('gw_port_id', None)
+        if not gw_port_id:
+            raise exceptions.RouterDoesNotHaveGateway(router_id=router.id)
+        gw_port = Port.get_object(
+            context, id=gw_port_id)
+        if not gw_port:
+            raise exceptions.RouterGatewayPortNotFound(router_id=router.id,
+                                                       gw_port_id=gw_port_id)
+        gw_port_ips = gw_port.get("fixed_ips", [])
+        if len(gw_port_ips) <= 0:
+            raise exceptions.RouterGatewayPortDoesNotHaveAnyIPAddresses(
+                router_id=router.id, gw_port_id=gw_port_id)
+        gw_ip_address = gw_port_ips[0].get('ip_address')
+        return gw_ip_address
+
+    @staticmethod
+    def _get_port(context: Context, port_id: str) -> Optional[Port]:
+        port = Port.get_object(context, id=port_id)
+        if not port:
+            raise PortNotFound(port_id=port_id)
+        return port
+
+    @staticmethod
+    def _get_ports(context: Context, router_id: str, port: Port,
+                   device_owner: str) -> Optional[List[Port]]:
+        ports = Port.get_ports_by_router_and_network(
+            context, router_id, device_owner, port.network_id)
+        if not ports:
+            raise exceptions.PortNetworkNotBindOnRouter(
+                port_id=port.id,
+                network_id=port.network_id,
+                router_id=router_id)
+        return ports
+
+    @staticmethod
+    def _validate_filter_for_port_forwarding(filters: Dict[str, str]) -> None:
+        if not filters:
+            return
+        for filter_member_key in filters.keys():
+            if filter_member_key in FIELDS_NOT_SUPPORT_FILTER:
+                raise exceptions.PortForwardingNotSupportFilterField(
+                    filter=filter_member_key)
+
+    @staticmethod
+    def _check_port_has_binding_floating_ip(context: Context, port_id: str,
+                                            ip_address: str) -> None:
+        floatingip_objs = FloatingIP.get_objects(
+            context.elevated(),
+            fixed_port_id=port_id)
+        if floatingip_objs:
+            floating_ip_address = floatingip_objs[0].floating_ip_address
+            raise exceptions.PortHasBindingFloatingIP(
+                floating_ip_address=floating_ip_address,
+                fip_id=floatingip_objs[0].id,
+                port_id=port_id,
+                fixed_ip=ip_address)
+
+    @staticmethod
+    def _get_device_owner(router: Router) -> str:
+        if is_distributed_router(router):
+            return constants.DEVICE_OWNER_DVR_INTERFACE
+        elif is_ha_router(router):
+            return constants.DEVICE_OWNER_HA_REPLICATED_INT
+        return constants.DEVICE_OWNER_ROUTER_INTF
+
+    def _check_router_port(self, context: Context, router: Router,
+                           port: Port):
+        device_owner = self._get_device_owner(router)
+        self._get_ports(context, router.id, port, device_owner)
+
+    def _check_port(self, context: Context, port_id: str, ip: str) -> Port:
+        port = self._get_port(context, port_id)
+        self._check_port_has_binding_floating_ip(context, port_id, ip)
+        fixed_ips = port.get('fixed_ips', [])
+        result = list(map(lambda x: str(x.get('ip_address')) == ip, fixed_ips))
+        if not any(result):
+            raise exceptions.InconsistentPortAndIP(port_id=port, ip_address=ip)
+        return port
+
+    def _check_router(self, context: Context, router_id: str) -> (Router, str):
+        router = self._get_router(context, router_id)
+        gw_ip_address = self._get_router_gateway(context, router)
+        return router, gw_ip_address
+
+    def _check_port_forwarding_create(self, context: Context, router_id: str,
+                                      pf_dict: Dict) -> None:
+        router, gw_ip_address = self._check_router(context, router_id)
+        pf_dict['router_id'] = router_id
+        pf_dict[apidef.GW_IP_ADDRESS] = gw_ip_address
+        internal_port_id = pf_dict[apidef.INTERNAL_PORT_ID]
+        internal_ip_address = pf_dict[apidef.INTERNAL_IP_ADDRESS]
+        internal_port = self._check_port(context, internal_port_id,
+                                         internal_ip_address)
+        self._check_router_port(context, router, internal_port)
+
+    @staticmethod
+    def _check_port_forwarding(context: Context, pf_obj: RGPortForwarding):
+        pf_objs = RGPortForwarding.get_objects(
+            context,
+            router_id=pf_obj.router_id,
+            protocol=pf_obj.protocol)
+
+        for obj in pf_objs:
+            if obj.id == pf_obj.get('id', None):
+                continue
+            # Ensure there are no conflicts on the outside
+            if obj.external_port == pf_obj.external_port:
+                raise exceptions.RouterGatewayPortForwardingAlreadyExists(
+                    conflict={
+                        'router_id': pf_obj.router_id,
+                        'protocol': pf_obj.protocol,
+                        'external_port': obj.external_port,
+                    }
+                )
+            # Ensure there are no conflicts in the inside
+            # socket: internal_ip_address + internal_port
+            if (obj.internal_port_id == pf_obj.internal_port_id and
+                    obj.internal_ip_address == pf_obj.internal_ip_address and
+                    obj.internal_port == pf_obj.internal_port):
+                raise exceptions.RouterGatewayPortForwardingAlreadyExists(
+                    conflict={
+                        'router_id': pf_obj.router_id,
+                        'protocol': pf_obj.protocol,
+                        'internal_port_id': obj.internal_port_id,
+                        'internal_ip_address': str(obj.internal_ip_address),
+                        'internal_port': obj.internal_port
+                    }
+                )
+
+    @staticmethod
+    def _find_existing_rg_port_forwarding(context: Context,
+                                          router_id: str,
+                                          port_forwarding: Dict,
+                                          specify_params: List = None):
+        # Because the session had been flushed by NeutronDbObjectDuplicateEntry
+        # so if we want to use the context to get another db queries, we need
+        # to rollback first.
+        context.session.rollback()
+        if not specify_params:
+            specify_params = [
+                {
+                    'router_id': router_id,
+                    'external_port': port_forwarding['external_port'],
+                    'protocol': port_forwarding['protocol']
+                },
+                {
+                    'internal_port_id': port_forwarding['internal_port_id'],
+                    'internal_ip_address': port_forwarding[
+                        'internal_ip_address'],
+                    'internal_port': port_forwarding['internal_port'],
+                    'protocol': port_forwarding['protocol']
+                }]
+        for param in specify_params:
+            objs = RGPortForwarding.get_objects(context, **param)
+            if objs:
+                return objs[0], param
+
+    @db_base_plugin_common.make_result_with_fields
+    @db_base_plugin_common.convert_result_to_dict
+    def get_router_gateway_port_forwardings(self, context: Context,
+                                            router_id: str,
+                                            filters: List[str] = None,
+                                            fields: List[str] = None,
+                                            sorts: List[str] = None,
+                                            limit: int = None,
+                                            marker: str = None,
+                                            page_reverse: bool = False):
+
+        router, gw_ip_address = self._check_router(context, router_id)
+        filters = filters or {}
+        self._validate_filter_for_port_forwarding(filters)
+        pager = Pager(sorts, limit, page_reverse, marker)
+        port_forwardings = RGPortForwarding.get_objects(
+            context, _pager=pager, router_id=router_id, **filters)
+        for pf in port_forwardings:
+            setattr(pf, 'gw_ip_address', gw_ip_address)
+        return port_forwardings
+
+    @db_base_plugin_common.convert_result_to_dict
+    def create_router_gateway_port_forwarding(self, context: Context,
+                                              router_id: str,
+                                              gateway_port_forwarding: dict):
+        port_forwarding = gateway_port_forwarding.get(apidef.RESOURCE_NAME)
+        self._check_port_forwarding_create(context, router_id, port_forwarding)
+        with CONTEXT_WRITER.using(context):
+            pf_obj = RGPortForwarding(context, **port_forwarding)
+            self._check_port_forwarding(context, pf_obj)
+            try:
+                pf_obj.create()
+            except NeutronDbObjectDuplicateEntry:
+                _, conflict = self._find_existing_rg_port_forwarding(
+                    context, router_id, port_forwarding)
+                raise exceptions.RouterGatewayPortForwardingAlreadyExists(
+                    conflict=conflict
+                )
+        self.push_api.push(context, [pf_obj], events.CREATED)
+        return pf_obj
+
+    @db_base_plugin_common.convert_result_to_dict
+    def update_router_gateway_port_forwarding(self, context: Context, id: str,
+                                              router_id: str,
+                                              gateway_port_forwarding: dict):
+
+        router = self._get_router(context, router_id)
+        gw_ip_address = self._get_router_gateway(context, router)
+        pf_obj = RGPortForwarding.get_object(context, id=id)
+        if not pf_obj:
+            raise exceptions.RouterGatewayPortForwardingNotFound(id=id)
+
+        port_forwarding = gateway_port_forwarding.get(apidef.RESOURCE_NAME, {})
+        port_forwarding[apidef.GW_IP_ADDRESS] = gw_ip_address
+        new_port_id = port_forwarding.get(apidef.INTERNAL_PORT_ID)
+        new_internal_ip = port_forwarding.get(apidef.INTERNAL_IP_ADDRESS, None)
+
+        if new_port_id and new_port_id != pf_obj.internal_port_id:
+            self._check_port_has_binding_floating_ip(context,
+                                                     new_port_id,
+                                                     new_internal_ip)
+
+        if any([new_internal_ip, new_port_id]):
+            port_forwarding.update({
+                apidef.INTERNAL_IP_ADDRESS: new_internal_ip
+                if new_internal_ip else
+                str(pf_obj.internal_ip_address),
+                apidef.INTERNAL_PORT_ID: new_port_id
+                if new_port_id else pf_obj.internal_port
+            })
+
+        with CONTEXT_WRITER.using(context):
+            pf_obj.update_fields(port_forwarding, reset_changes=True)
+            self._check_port_forwarding(context, pf_obj)
+            try:
+                pf_obj.update()
+            except NeutronDbObjectDuplicateEntry:
+                _, conflict = self._find_existing_rg_port_forwarding(
+                    context, router_id, port_forwarding)
+                raise exceptions.RouterGatewayPortForwardingAlreadyExists(
+                    conflict=conflict
+                )
+        self.push_api.push(context, [pf_obj], events.UPDATED)
+        return pf_obj
+
+    @db_base_plugin_common.make_result_with_fields
+    @db_base_plugin_common.convert_result_to_dict
+    def get_router_gateway_port_forwarding(self, context: Context, id: str,
+                                           router_id: str,
+                                           fields: List[str] = None):
+        _, gw_ip_address = self._check_router(context, router_id)
+        pf_obj = RGPortForwarding.get_object(context, id=id)
+        if not pf_obj:
+            raise exceptions.RouterGatewayPortForwardingNotFound(id=id)
+        setattr(pf_obj, apidef.GW_IP_ADDRESS, gw_ip_address)
+        return pf_obj
+
+    def delete_router_gateway_port_forwarding(self, context: Context, id: str,
+                                              router_id: str):
+        pf_obj = RGPortForwarding.get_object(context, id=id)
+        if not pf_obj:
+            raise exceptions.RouterGatewayPortForwardingNotFound(id=id)
+        with CONTEXT_WRITER.using(context):
+            pf_obj.delete()
+        self.push_api.push(context, [pf_obj], events.DELETED)
+
+    @registry.receives(resources.ROUTER, [lib_events.BEFORE_DELETE])
+    def _receive_router_before_delete(self, resource: str, event: str,
+                                      trigger: L3RouterPlugin,
+                                      payload: DBEventPayload):
+        router_id = payload.resource_id
+        context = payload.context
+        port_forwardings = RGPortForwarding.get_objects(context,
+                                                        router_id=router_id)
+        if port_forwardings:
+            ex = exceptions.DeletedRouterWithRGForwarding(router_id=router_id)
+            LOG.info(ex.msg)
+            raise ex
+
+    @registry.receives(resources.ROUTER_GATEWAY, [lib_events.BEFORE_DELETE,
+                                                  lib_events.BEFORE_UPDATE])
+    def _receive_router_gateway_before_delete(self, resource: str, event: str,
+                                              trigger: L3RouterPlugin,
+                                              payload: DBEventPayload):
+        router_id = payload.resource_id
+        context = payload.context
+        port_forwardings = RGPortForwarding.get_objects(context,
+                                                        router_id=router_id)
+        if port_forwardings:
+            ex = exceptions.DeletedRouterGatewayWithRGForwarding(
+                router_id=router_id)
+            LOG.info(ex.msg)
+            raise ex