packages/openstack-neutron
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
openstack-neutron/0001-add-distributed-traffic-feature-support.patch
wangkuntian 784a5b2e38 Add distributed traffic feature
2023-10-27 10:50:02 +08:00

2584 lines
110 KiB
Diff
Raw Permalink Blame History

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
Reference in New Issue View Git Blame Copy Permalink