Add accepted egress direct flow

Do not flood the packets to bridge, since we have the
bridge port list, we can add a simple direct flow to
the right port only.

Closes-Bug: #1732067
Related-Bug: #1841622
Change-Id: I14fefe289a19b718b247bf0740ca9bc47f8903f4

neutron/agent/linux/openvswitch_firewall/firewall.py

@@ -23,11 +23,14 @@ from neutron_lib.callbacks import events as callbacks_events
 from neutron_lib.callbacks import registry as callbacks_registry
 from neutron_lib.callbacks import resources as callbacks_resources
 from neutron_lib import constants as lib_const
+from neutron_lib.plugins import utils as p_utils
+from neutron_lib.utils import helpers
 from oslo_config import cfg
 from oslo_log import log as logging
 from oslo_utils import netutils
 
 from neutron._i18n import _
+from neutron.agent.common import ovs_lib
 from neutron.agent import firewall
 from neutron.agent.linux.openvswitch_firewall import constants as ovsfw_consts
 from neutron.agent.linux.openvswitch_firewall import exceptions
@@ -82,6 +85,26 @@ def get_segmentation_id_from_other_config(bridge, port_name):
         pass
 
 
+def get_network_type_from_other_config(bridge, port_name):
+    """Return network_type stored in OVSDB other_config metadata.
+
+    :param bridge: OVSBridge instance where port is.
+    :param port_name: Name of the port.
+    """
+    other_config = bridge.db_get_val('Port', port_name, 'other_config')
+    return other_config.get('network_type')
+
+
+def get_physical_network_from_other_config(bridge, port_name):
+    """Return physical_network stored in OVSDB other_config metadata.
+
+    :param bridge: OVSBridge instance where port is.
+    :param port_name: Name of the port.
+    """
+    other_config = bridge.db_get_val('Port', port_name, 'other_config')
+    return other_config.get('physical_network')
+
+
 def get_tag_from_other_config(bridge, port_name):
     """Return tag stored in OVSDB other_config metadata.
 
@@ -135,7 +158,8 @@ class SecurityGroup(object):
 
 
 class OFPort(object):
-    def __init__(self, port_dict, ovs_port, vlan_tag, segment_id=None):
+    def __init__(self, port_dict, ovs_port, vlan_tag, segment_id=None,
+                 network_type=None, physical_network=None):
         self.id = port_dict['device']
         self.vlan_tag = vlan_tag
         self.segment_id = segment_id
@@ -148,6 +172,8 @@ class OFPort(object):
         self.neutron_port_dict = port_dict.copy()
         self.allowed_pairs_v4 = self._get_allowed_pairs(port_dict, version=4)
         self.allowed_pairs_v6 = self._get_allowed_pairs(port_dict, version=6)
+        self.network_type = network_type
+        self.physical_network = physical_network
 
     @staticmethod
     def _get_allowed_pairs(port_dict, version):
@@ -538,6 +564,14 @@ class OVSFirewallDriver(firewall.FirewallDriver):
         return get_segmentation_id_from_other_config(
             self.int_br.br, port_name)
 
+    def _get_port_network_type(self, port_name):
+        return get_network_type_from_other_config(
+            self.int_br.br, port_name)
+
+    def _get_port_physical_network(self, port_name):
+        return get_physical_network_from_other_config(
+            self.int_br.br, port_name)
+
     def get_ofport(self, port):
         port_id = port['device']
         return self.sg_port_map.ports.get(port_id)
@@ -555,8 +589,13 @@ class OVSFirewallDriver(firewall.FirewallDriver):
             port_vlan_id = self._get_port_vlan_tag(ovs_port.port_name)
             segment_id = self._get_port_segmentation_id(
                 ovs_port.port_name)
+            network_type = self._get_port_network_type(
+                ovs_port.port_name)
+            physical_network = self._get_port_physical_network(
+                ovs_port.port_name)
             of_port = OFPort(port, ovs_port, port_vlan_id,
-                             segment_id)
+                             segment_id,
+                             network_type, physical_network)
             self.sg_port_map.create_port(of_port, port)
         else:
             if of_port.ofport != ovs_port.ofport:
@@ -835,7 +874,7 @@ class OVSFirewallDriver(firewall.FirewallDriver):
                       {'port_id': port_id,
                        'err': not_found_e})
             return
-        self.sg_port_map.unfiltered[port_id] = ovs_port.ofport
+        self.sg_port_map.unfiltered[port_id] = (ovs_port, vlan_tag)
         self._add_flow(
             table=ovs_consts.TRANSIENT_TABLE,
             priority=100,
@@ -857,20 +896,33 @@ class OVSFirewallDriver(firewall.FirewallDriver):
                 ovs_consts.ACCEPTED_EGRESS_TRAFFIC_NORMAL_TABLE)
         )
 
+        tunnel_direct_info = {
+            "network_type": self._get_port_network_type(ovs_port.port_name),
+            "physical_network": self._get_port_physical_network(
+                ovs_port.port_name)
+        }
+        self.install_accepted_egress_direct_flow(
+            ovs_port.vif_mac, vlan_tag, ovs_port.ofport,
+            tunnel_direct_info=tunnel_direct_info)
+
     def _remove_egress_no_port_security(self, port_id):
         try:
-            ofport = self.sg_port_map.unfiltered[port_id]
+            ovs_port, vlan_tag = self.sg_port_map.unfiltered[port_id]
         except KeyError:
             raise exceptions.OVSFWPortNotHandled(port_id=port_id)
 
         self._delete_flows(
             table=ovs_consts.TRANSIENT_TABLE,
-            in_port=ofport
+            in_port=ovs_port.ofport
         )
         self._delete_flows(
             table=ovs_consts.ACCEPT_OR_INGRESS_TABLE,
-            reg_port=ofport
+            reg_port=ovs_port.ofport
         )
+
+        self.delete_accepted_egress_direct_flow(
+            ovs_port.vif_mac, vlan_tag)
+
         del self.sg_port_map.unfiltered[port_id]
 
     def _initialize_egress(self, port):
@@ -1030,6 +1082,72 @@ class OVSFirewallDriver(firewall.FirewallDriver):
                 ovs_consts.ACCEPTED_EGRESS_TRAFFIC_NORMAL_TABLE)
         )
 
+        tunnel_direct_info = {"network_type": port.network_type,
+                              "physical_network": port.physical_network}
+        self.install_accepted_egress_direct_flow(
+            port.mac, port.vlan_tag, port.ofport,
+            tunnel_direct_info=tunnel_direct_info)
+
+    def install_accepted_egress_direct_flow(self, mac, vlan_tag, dst_port,
+                                            tunnel_direct_info=None):
+        if not cfg.CONF.AGENT.explicitly_egress_direct:
+            return
+
+        # Prevent flood for accepted egress traffic
+        self._add_flow(
+            table=ovs_consts.ACCEPTED_EGRESS_TRAFFIC_NORMAL_TABLE,
+            priority=12,
+            dl_dst=mac,
+            reg_net=vlan_tag,
+            actions='output:{:d}'.format(dst_port)
+        )
+
+        # The former flow may not match, that means the destination port is
+        # not in this host. So, we direct the packet to mapped bridge(s).
+        if tunnel_direct_info:
+            patch_ofport = ovs_lib.INVALID_OFPORT
+            if tunnel_direct_info["network_type"] in (
+                    lib_const.TYPE_VXLAN, lib_const.TYPE_GRE,
+                    lib_const.TYPE_GENEVE):
+                # Some ports like router internal gateway will not install
+                # the l2pop related flows, so we will transmit the ARP request
+                # packet to tunnel bridge use NORMAL action as usual.
+                port_name = cfg.CONF.OVS.int_peer_patch_port
+                patch_ofport = self.int_br.br.get_port_ofport(port_name)
+            elif tunnel_direct_info["network_type"] == lib_const.TYPE_VLAN:
+                physical_network = tunnel_direct_info["physical_network"]
+                if not physical_network:
+                    return
+                bridge_mappings = helpers.parse_mappings(
+                    cfg.CONF.OVS.bridge_mappings)
+                bridge = bridge_mappings.get(physical_network)
+                port_name = p_utils.get_interface_name(
+                    bridge, prefix=ovs_consts.PEER_INTEGRATION_PREFIX)
+                patch_ofport = self.int_br.br.get_port_ofport(port_name)
+
+            if patch_ofport is not ovs_lib.INVALID_OFPORT:
+                self._add_flow(
+                    table=ovs_consts.ACCEPTED_EGRESS_TRAFFIC_NORMAL_TABLE,
+                    priority=10,
+                    dl_src=mac,
+                    dl_dst="00:00:00:00:00:00/01:00:00:00:00:00",
+                    reg_net=vlan_tag,
+                    actions='mod_vlan_vid:{:d},output:{:d}'.format(
+                        vlan_tag,
+                        patch_ofport)
+                )
+
+    def delete_accepted_egress_direct_flow(self, mac, vlan_tag):
+        self._delete_flows(
+            table=ovs_consts.ACCEPTED_EGRESS_TRAFFIC_NORMAL_TABLE,
+            dl_dst=mac,
+            reg_net=vlan_tag)
+
+        self._delete_flows(
+            table=ovs_consts.ACCEPTED_EGRESS_TRAFFIC_NORMAL_TABLE,
+            dl_src=mac,
+            reg_net=vlan_tag)
+
     def _initialize_tracked_egress(self, port):
         # Drop invalid packets
         self._add_flow(
@@ -1291,6 +1409,9 @@ class OVSFirewallDriver(firewall.FirewallDriver):
             self.delete_vlan_direct_flow(mac_addr, port.segment_id)
             self._delete_flows(table=ovs_consts.ACCEPT_OR_INGRESS_TABLE,
                                dl_dst=mac_addr, reg_net=port.vlan_tag)
+
+        self.delete_accepted_egress_direct_flow(
+            port.mac, port.vlan_tag)
         self._strict_delete_flow(priority=100,
                                  table=ovs_consts.TRANSIENT_TABLE,
                                  in_port=port.ofport)
@@ -1298,6 +1419,9 @@ class OVSFirewallDriver(firewall.FirewallDriver):
 
     def delete_flows_for_ip_addresses(
             self, ip_addresses, direction, ethertype, vlan_tag):
+        if not cfg.CONF.AGENT.explicitly_egress_direct:
+            return
+
         for ip_addr in ip_addresses:
             # Generate deletion template with bogus conj_id.
             flows = rules.create_flows_for_ip_address(

neutron/agent/securitygroups_rpc.py

@@ -123,6 +123,10 @@ class SecurityGroupAgentRpc(object):
                             *args, **kwargs)
         return decorated_function
 
+    @skip_if_noopfirewall_or_firewall_disabled
+    def init_ovs_dvr_firewall(self, dvr_agent):
+        dvr_agent.set_firewall(self.firewall)
+
     @skip_if_noopfirewall_or_firewall_disabled
     def prepare_devices_filter(self, device_ids):
         if not device_ids:

neutron/conf/plugins/ml2/drivers/ovs_conf.py

@@ -170,6 +170,12 @@ agent_opts = [
                        "outgoing IP packet carrying GRE/VXLAN tunnel.")),
     cfg.BoolOpt('baremetal_smartnic', default=False,
                 help=_("Enable the agent to process Smart NIC ports.")),
+    cfg.BoolOpt('explicitly_egress_direct', default=False,
+                help=_("When set to True, the accepted egress unicast "
+                       "traffic will not use action NORMAL. The accepted "
+                       "egress packets will be taken care of in the final "
+                       "egress tables direct output flows for unicast "
+                       "traffic.")),
 ]
 
 

neutron/plugins/ml2/drivers/openvswitch/agent/common/constants.py

@@ -56,6 +56,7 @@ MAC_SPOOF_TABLE = 25
 
 # Table to decide whether further filtering is needed
 TRANSIENT_TABLE = 60
+TRANSIENT_EGRESS_TABLE = 61
 
 # Tables used for ovs firewall
 BASE_EGRESS_TABLE = 71
@@ -86,6 +87,7 @@ INT_BR_ALL_TABLES = (
     ARP_SPOOF_TABLE,
     MAC_SPOOF_TABLE,
     TRANSIENT_TABLE,
+    TRANSIENT_EGRESS_TABLE,
     BASE_EGRESS_TABLE,
     RULES_EGRESS_TABLE,
     ACCEPT_OR_INGRESS_TABLE,

neutron/plugins/ml2/drivers/openvswitch/agent/openflow/native/br_int.py

@@ -48,6 +48,10 @@ class OVSIntegrationBridge(ovs_bridge.OVSAgentBridge):
         self.install_drop(table_id=constants.LOCAL_SWITCHING,
                           priority=constants.OPENFLOW_MAX_PRIORITY,
                           vlan_vid=constants.DEAD_VLAN_TAG)
+        # When openflow firewall is not enabled, we use this table to
+        # deal with all egress flow.
+        self.install_normal(table_id=constants.TRANSIENT_EGRESS_TABLE,
+                            priority=3)
 
     def setup_canary_table(self):
         self.install_drop(constants.CANARY_TABLE)
@@ -157,7 +161,7 @@ class OVSIntegrationBridge(ovs_bridge.OVSAgentBridge):
             ofpp.OFPInstructionGotoTable(table_id=constants.TRANSIENT_TABLE),
         ]
         self.install_instructions(table_id=table_id,
-                                  priority=4,
+                                  priority=20,
                                   match=match,
                                   instructions=instructions)
         actions = [
@@ -165,7 +169,7 @@ class OVSIntegrationBridge(ovs_bridge.OVSAgentBridge):
             ofpp.OFPActionOutput(dst_port, 0),
         ]
         self.install_apply_actions(table_id=constants.TRANSIENT_TABLE,
-                                   priority=4,
+                                   priority=20,
                                    match=match,
                                    actions=actions)
 
@@ -176,7 +180,7 @@ class OVSIntegrationBridge(ovs_bridge.OVSAgentBridge):
                                            vlan_tag=vlan_tag, dst_mac=dst_mac)
         for table in (table_id, constants.TRANSIENT_TABLE):
             self.uninstall_flows(
-                strict=True, priority=4, table_id=table, match=match)
+                strict=True, priority=20, table_id=table, match=match)
 
     def add_dvr_mac_vlan(self, mac, port):
         self.install_goto(table_id=constants.LOCAL_SWITCHING,

neutron/plugins/ml2/drivers/openvswitch/agent/ovs_dvr_neutron_agent.py

@@ -17,11 +17,13 @@ import sys
 
 import netaddr
 from neutron_lib import constants as n_const
+from oslo_config import cfg
 from oslo_log import log as logging
 import oslo_messaging
 from oslo_utils import excutils
 from osprofiler import profiler
 
+from neutron.agent.linux.openvswitch_firewall import firewall as ovs_firewall
 from neutron.common import utils as n_utils
 from neutron.plugins.ml2.drivers.openvswitch.agent.common import constants
 
@@ -135,6 +137,11 @@ class OVSDVRNeutronAgent(object):
         self.dvr_mac_address = None
         if self.enable_distributed_routing:
             self.get_dvr_mac_address()
+        self.conf = cfg.CONF
+        self.firewall = None
+
+    def set_firewall(self, firewall=None):
+        self.firewall = firewall
 
     def setup_dvr_flows(self):
         self.setup_dvr_flows_on_integ_br()
@@ -390,6 +397,15 @@ class OVSDVRNeutronAgent(object):
 
         subnet_info = ldm.get_subnet_info()
         ip_version = subnet_info['ip_version']
+
+        if self.firewall and isinstance(self.firewall,
+                                        ovs_firewall.OVSFirewallDriver):
+            tunnel_direct_info = {"network_type": lvm.network_type,
+                                  "physical_network": lvm.physical_network}
+            self.firewall.install_accepted_egress_direct_flow(
+                subnet_info['gateway_mac'], lvm.vlan, port.ofport,
+                tunnel_direct_info=tunnel_direct_info)
+
         local_compute_ports = (
             self.plugin_rpc.get_ports_on_host_by_subnet(
                 self.context, self.host, subnet_uuid))
@@ -648,6 +664,11 @@ class OVSDVRNeutronAgent(object):
                     vlan_tag=lvm.vlan, gateway_mac=subnet_info['gateway_mac'])
             ovsport.remove_subnet(sub_uuid)
 
+            if self.firewall and isinstance(self.firewall,
+                                            ovs_firewall.OVSFirewallDriver):
+                self.firewall.delete_accepted_egress_direct_flow(
+                    subnet_info['gateway_mac'], lvm.vlan)
+
         if lvm.network_type == n_const.TYPE_VLAN:
             br = self.phys_brs[physical_network]
         if lvm.network_type in constants.TUNNEL_NETWORK_TYPES:

neutron/plugins/ml2/drivers/openvswitch/agent/ovs_neutron_agent.py

@@ -21,6 +21,7 @@ import signal
 import sys
 import time
 
+import eventlet
 import netaddr
 from neutron_lib.agent import constants as agent_consts
 from neutron_lib.agent import topics
@@ -53,6 +54,7 @@ from neutron.agent.common import ip_lib
 from neutron.agent.common import ovs_lib
 from neutron.agent.common import polling
 from neutron.agent.common import utils
+from neutron.agent import firewall as agent_firewall
 from neutron.agent.l2 import l2_agent_extensions_manager as ext_manager
 from neutron.agent.linux import xenapi_root_helper
 from neutron.agent import rpc as agent_rpc
@@ -297,6 +299,8 @@ class OVSNeutronAgent(l2population_rpc.L2populationRpcCallBackTunnelMixin,
         self.sg_plugin_rpc.register_legacy_sg_notification_callbacks(
             self.sg_agent)
 
+        self.sg_agent.init_ovs_dvr_firewall(self.dvr_agent)
+
         # we default to False to provide backward compat with out of tree
         # firewall drivers that expect the logic that existed on the Neutron
         # server which only enabled hybrid plugging based on the use of the
@@ -636,26 +640,49 @@ class OVSNeutronAgent(l2population_rpc.L2populationRpcCallBackTunnelMixin,
                 port_set.remove(port_id)
                 break
 
+    def _get_port_local_vlan(self, port_id):
+        for network_id, port_set in self.network_ports.items():
+            if port_id in port_set:
+                lvm = self.vlan_manager.get(network_id)
+                return lvm.vlan
+
     def process_deleted_ports(self, port_info):
         # don't try to process removed ports as deleted ports since
         # they are already gone
         if 'removed' in port_info:
             self.deleted_ports -= port_info['removed']
         deleted_ports = list(self.deleted_ports)
-        while self.deleted_ports:
-            port_id = self.deleted_ports.pop()
-            port = self.int_br.get_vif_port_by_id(port_id)
-            self._clean_network_ports(port_id)
-            self.ext_manager.delete_port(self.context,
-                                         {"vif_port": port,
-                                          "port_id": port_id})
-            # move to dead VLAN so deleted ports no
-            # longer have access to the network
-            if port:
-                # don't log errors since there is a chance someone will be
-                # removing the port from the bridge at the same time
-                self.port_dead(port, log_errors=False)
-            self.port_unbound(port_id)
+
+        with self.int_br.deferred(full_ordered=True,
+                                  use_bundle=True) as int_br:
+            while self.deleted_ports:
+                port_id = self.deleted_ports.pop()
+                port = self.int_br.get_vif_port_by_id(port_id)
+
+                if (isinstance(self.sg_agent.firewall,
+                               agent_firewall.NoopFirewallDriver) or
+                        not agent_sg_rpc.is_firewall_enabled()):
+                    try:
+                        self.delete_accepted_egress_direct_flow(
+                            int_br,
+                            port.ofport,
+                            port.mac, self._get_port_local_vlan(port_id))
+                    except Exception as err:
+                        LOG.debug("Failed to remove accepted egress flows "
+                                  "for port %s, error: %s", port_id, err)
+
+                self._clean_network_ports(port_id)
+                self.ext_manager.delete_port(self.context,
+                                             {"vif_port": port,
+                                              "port_id": port_id})
+                # move to dead VLAN so deleted ports no
+                # longer have access to the network
+                if port:
+                    # don't log errors since there is a chance someone will be
+                    # removing the port from the bridge at the same time
+                    self.port_dead(port, log_errors=False)
+                self.port_unbound(port_id)
+
         # Flush firewall rules after ports are put on dead VLAN to be
         # more secure
         self.sg_agent.remove_devices_filter(deleted_ports)
@@ -2010,6 +2037,9 @@ class OVSNeutronAgent(l2population_rpc.L2populationRpcCallBackTunnelMixin,
         added_ports = (port_info.get('added', set()) - skipped_devices -
                        binding_no_activated_devices)
         self._add_port_tag_info(need_binding_devices)
+
+        self.process_install_ports_egress_flows(need_binding_devices)
+
         self.sg_agent.setup_port_filters(added_ports,
                                          port_info.get('updated', set()))
         LOG.info("process_network_ports - iteration:%(iter_num)d - "
@@ -2035,6 +2065,83 @@ class OVSNeutronAgent(l2population_rpc.L2populationRpcCallBackTunnelMixin,
                       'elapsed': time.time() - start})
         return failed_devices
 
+    def process_install_ports_egress_flows(self, ports):
+        if not self.conf.AGENT.explicitly_egress_direct:
+            return
+
+        if (isinstance(self.sg_agent.firewall,
+                       agent_firewall.NoopFirewallDriver) or
+                not agent_sg_rpc.is_firewall_enabled()):
+            with self.int_br.deferred(full_ordered=True,
+                                      use_bundle=True) as int_br:
+                for port in ports:
+                    try:
+                        self.install_accepted_egress_direct_flow(port, int_br)
+                        # give other coroutines a chance to run
+                        eventlet.sleep(0)
+                    except Exception as err:
+                        LOG.debug("Failed to install accepted egress flows "
+                                  "for port %s, error: %s",
+                                  port['port_id'], err)
+
+    def install_accepted_egress_direct_flow(self, port_detail, br_int):
+        lvm = self.vlan_manager.get(port_detail['network_id'])
+        port = port_detail['vif_port']
+
+        br_int.add_flow(
+            table=constants.TRANSIENT_TABLE,
+            priority=9,
+            in_port=port.ofport,
+            dl_src=port_detail['mac_address'],
+            actions='resubmit(,{:d})'.format(
+                constants.TRANSIENT_EGRESS_TABLE))
+
+        br_int.add_flow(
+            table=constants.TRANSIENT_EGRESS_TABLE,
+            priority=12,
+            dl_dst=port_detail['mac_address'],
+            actions='output:{:d}'.format(port.ofport))
+
+        patch_ofport = None
+        if lvm.network_type in (
+                n_const.TYPE_VXLAN, n_const.TYPE_GRE,
+                n_const.TYPE_GENEVE):
+            port_name = self.conf.OVS.int_peer_patch_port
+            patch_ofport = self.int_br.get_port_ofport(port_name)
+        elif lvm.network_type == n_const.TYPE_VLAN:
+            bridge = self.bridge_mappings.get(lvm.physical_network)
+            port_name = plugin_utils.get_interface_name(
+                bridge, prefix=constants.PEER_INTEGRATION_PREFIX)
+            patch_ofport = self.int_br.get_port_ofport(port_name)
+        if patch_ofport is not None:
+            br_int.add_flow(
+                table=constants.TRANSIENT_EGRESS_TABLE,
+                priority=10,
+                dl_src=port_detail['mac_address'],
+                dl_dst="00:00:00:00:00:00/01:00:00:00:00:00",
+                in_port=port.ofport,
+                actions='mod_vlan_vid:{:d},'
+                        'output:{:d}'.format(
+                            lvm.vlan,
+                            patch_ofport))
+
+    def delete_accepted_egress_direct_flow(self, br_int, ofport, mac, vlan):
+        if not self.conf.AGENT.explicitly_egress_direct:
+            return
+
+        br_int.delete_flows(
+            table=constants.TRANSIENT_TABLE,
+            in_port=ofport,
+            dl_src=mac)
+        self.delete_flows(
+            table=constants.TRANSIENT_EGRESS_TABLE,
+            dl_dst=mac)
+
+        self.delete_flows(
+            table=constants.TRANSIENT_EGRESS_TABLE,
+            dl_src=mac,
+            in_port=ofport)
+
     def process_ancillary_network_ports(self, port_info):
         failed_devices = {'added': set(), 'removed': set()}
         if 'added' in port_info and port_info['added']:

neutron/tests/unit/agent/linux/openvswitch_firewall/test_firewall.py

@@ -17,6 +17,8 @@ from neutron_lib.callbacks import events as callbacks_events
 from neutron_lib.callbacks import registry as callbacks_registry
 from neutron_lib.callbacks import resources as callbacks_resources
 from neutron_lib import constants
+from neutron_lib.utils import helpers
+from oslo_config import cfg
 import testtools
 
 from neutron.agent.common import ovs_lib
@@ -35,11 +37,13 @@ TESTING_VLAN_TAG = 1
 TESTING_SEGMENT = 1000
 
 
-def create_ofport(port_dict):
+def create_ofport(port_dict, network_type=None, physical_network=None):
     ovs_port = mock.Mock(vif_mac='00:00:00:00:00:00', ofport=1,
                          port_name="port-name")
     return ovsfw.OFPort(port_dict, ovs_port, vlan_tag=TESTING_VLAN_TAG,
-                        segment_id=TESTING_SEGMENT)
+                        segment_id=TESTING_SEGMENT,
+                        network_type=network_type,
+                        physical_network=physical_network)
 
 
 class TestCreateRegNumbers(base.BaseTestCase):
@@ -381,6 +385,7 @@ class TestOVSFirewallDriver(base.BaseTestCase):
         self.fake_ovs_port = FakeOVSPort('port', 1, '00:00:00:00:00:00')
         self.mock_bridge.br.get_vif_port_by_id.return_value = \
             self.fake_ovs_port
+        cfg.CONF.set_override('explicitly_egress_direct', True, 'AGENT')
 
     def _prepare_security_group(self):
         security_group_rules = [
@@ -572,10 +577,14 @@ class TestOVSFirewallDriver(base.BaseTestCase):
         port_dict = {
             'device': 'port-id',
             'security_groups': [1]}
-        of_port = create_ofport(port_dict)
+        of_port = create_ofport(port_dict,
+                                network_type=constants.TYPE_VXLAN)
         self.firewall.sg_port_map.ports[of_port.id] = of_port
         port = self.firewall.get_or_create_ofport(port_dict)
 
+        fake_patch_port = 999
+        self.mock_bridge.br.get_port_ofport.return_value = fake_patch_port
+
         self.firewall.initialize_port_flows(port)
 
         call_args1 = {
@@ -624,11 +633,29 @@ class TestOVSFirewallDriver(base.BaseTestCase):
         self.mock_bridge.br.add_flow.assert_has_calls(
             [egress_flow_call, ingress_flow_call1, ingress_flow_call2])
 
+    def test_initialize_port_flows_vlan_dvr_conntrack_direct_vlan(self):
+        port_dict = {
+            'device': 'port-id',
+            'security_groups': [1]}
+        of_port = create_ofport(port_dict,
+                                network_type=constants.TYPE_VLAN,
+                                physical_network='vlan1')
+        self.firewall.sg_port_map.ports[of_port.id] = of_port
+        port = self.firewall.get_or_create_ofport(port_dict)
+
+        fake_patch_port = 999
+        self.mock_bridge.br.get_port_ofport.return_value = fake_patch_port
+
+        with mock.patch.object(helpers, "parse_mappings",
+                               return_value={"vlan1": "br-vlan1"}):
+            self.firewall.initialize_port_flows(port)
+
     def test_delete_all_port_flows(self):
         port_dict = {
             'device': 'port-id',
             'security_groups': [1]}
-        of_port = create_ofport(port_dict)
+        of_port = create_ofport(port_dict,
+                                network_type=constants.TYPE_VXLAN)
         self.firewall.sg_port_map.ports[of_port.id] = of_port
         port = self.firewall.get_or_create_ofport(port_dict)
 
@@ -662,8 +689,18 @@ class TestOVSFirewallDriver(base.BaseTestCase):
         call_args5 = {"reg5": port.ofport}
         flow5 = mock.call(**call_args5)
 
+        call_args6 = {"table": ovs_consts.ACCEPTED_EGRESS_TRAFFIC_NORMAL_TABLE,
+                      "dl_dst": port.mac,
+                      "reg6": port.vlan_tag}
+        flow6 = mock.call(**call_args6)
+
+        call_args7 = {"table": ovs_consts.ACCEPTED_EGRESS_TRAFFIC_NORMAL_TABLE,
+                      "dl_src": port.mac,
+                      "reg6": port.vlan_tag}
+        flow7 = mock.call(**call_args7)
+
         self.mock_bridge.br.delete_flows.assert_has_calls(
-            [flow1, flow2, flow3, flow4, flow5])
+            [flow1, flow2, flow3, flow6, flow7, flow4, flow5])
 
     def test_prepare_port_filter_initialized_port(self):
         port_dict = {'device': 'port-id',
@@ -830,7 +867,8 @@ class TestOVSFirewallDriver(base.BaseTestCase):
 
     def test__remove_egress_no_port_security_deletes_flow(self):
         self.mock_bridge.br.db_get_val.return_value = {'tag': TESTING_VLAN_TAG}
-        self.firewall.sg_port_map.unfiltered['port_id'] = 1
+        self.firewall.sg_port_map.unfiltered['port_id'] = (
+            self.fake_ovs_port, 100)
         self.firewall._remove_egress_no_port_security('port_id')
         expected_call = mock.call(
             table=ovs_consts.TRANSIENT_TABLE,
@@ -848,9 +886,10 @@ class TestOVSFirewallDriver(base.BaseTestCase):
         with mock.patch.object(self.firewall.int_br.br, 'get_vifs_by_ids',
                                return_value={'port_id': vif_port}):
             self.firewall.process_trusted_ports(['port_id'])
-            self.assertEqual(1, len(self.firewall.sg_port_map.unfiltered))
-            self.assertEqual(vif_port.ofport,
-                             self.firewall.sg_port_map.unfiltered['port_id'])
+            self.assertEqual(1,
+                             len(self.firewall.sg_port_map.unfiltered.keys()))
+            ofport, _ = self.firewall.sg_port_map.unfiltered['port_id']
+            self.assertEqual(vif_port.ofport, ofport.ofport)
 
     def test_process_trusted_ports_port_not_found(self):
         """Check that exception is not propagated outside."""
@@ -861,7 +900,8 @@ class TestOVSFirewallDriver(base.BaseTestCase):
             self.assertEqual(0, len(self.firewall.sg_port_map.unfiltered))
 
     def test_remove_trusted_ports_clears_cached_port_id(self):
-        self.firewall.sg_port_map.unfiltered['port_id'] = 1
+        self.firewall.sg_port_map.unfiltered['port_id'] = (
+            self.fake_ovs_port, 100)
         self.firewall.remove_trusted_ports(['port_id'])
         self.assertNotIn('port_id', self.firewall.sg_port_map.unfiltered)
 

neutron/tests/unit/plugins/ml2/drivers/openvswitch/agent/openflow/native/test_br_int.py

@@ -77,6 +77,18 @@ class OVSIntegrationBridgeTest(ovs_bridge_test_base.OVSBridgeTestBase):
                 priority=65535,
                 table_id=0),
                            active_bundle=None),
+            call._send_msg(ofpp.OFPFlowMod(dp,
+                cookie=self.stamp,
+                instructions=[
+                    ofpp.OFPInstructionActions(
+                        ofp.OFPIT_APPLY_ACTIONS, [
+                            ofpp.OFPActionOutput(ofp.OFPP_NORMAL, 0)
+                        ]),
+                ],
+                match=ofpp.OFPMatch(),
+                priority=3,
+                table_id=61),
+                           active_bundle=None),
         ]
         self.assertEqual(expected, self.mock.mock_calls)
 
@@ -183,7 +195,7 @@ class OVSIntegrationBridgeTest(ovs_bridge_test_base.OVSBridgeTestBase):
                 match=ofpp.OFPMatch(
                     eth_dst=dst_mac,
                     vlan_vid=vlan_tag | ofp.OFPVID_PRESENT),
-                priority=4,
+                priority=20,
                 table_id=1),
                            active_bundle=None),
             call._send_msg(ofpp.OFPFlowMod(dp,
@@ -197,7 +209,7 @@ class OVSIntegrationBridgeTest(ovs_bridge_test_base.OVSBridgeTestBase):
                 match=ofpp.OFPMatch(
                     eth_dst=dst_mac,
                     vlan_vid=vlan_tag | ofp.OFPVID_PRESENT),
-                priority=4,
+                priority=20,
                 table_id=60),
                            active_bundle=None),
         ]
@@ -214,14 +226,14 @@ class OVSIntegrationBridgeTest(ovs_bridge_test_base.OVSBridgeTestBase):
         expected = [
             call.uninstall_flows(
                 strict=True,
-                priority=4,
+                priority=20,
                 table_id=1,
                 match=ofpp.OFPMatch(
                     eth_dst=dst_mac,
                     vlan_vid=vlan_tag | ofp.OFPVID_PRESENT)),
             call.uninstall_flows(
                 strict=True,
-                priority=4,
+                priority=20,
                 table_id=60,
                 match=ofpp.OFPMatch(
                     eth_dst=dst_mac,
@@ -253,7 +265,7 @@ class OVSIntegrationBridgeTest(ovs_bridge_test_base.OVSBridgeTestBase):
                 match=ofpp.OFPMatch(
                     eth_dst=dst_mac,
                     vlan_vid=vlan_tag | ofp.OFPVID_PRESENT),
-                priority=4,
+                priority=20,
                 table_id=2),
                            active_bundle=None),
             call._send_msg(ofpp.OFPFlowMod(dp,
@@ -267,7 +279,7 @@ class OVSIntegrationBridgeTest(ovs_bridge_test_base.OVSBridgeTestBase):
                 match=ofpp.OFPMatch(
                     eth_dst=dst_mac,
                     vlan_vid=vlan_tag | ofp.OFPVID_PRESENT),
-                priority=4,
+                priority=20,
                 table_id=60),
                            active_bundle=None),
         ]
@@ -284,14 +296,14 @@ class OVSIntegrationBridgeTest(ovs_bridge_test_base.OVSBridgeTestBase):
         expected = [
             call.uninstall_flows(
                 strict=True,
-                priority=4,
+                priority=20,
                 table_id=2,
                 match=ofpp.OFPMatch(
                     eth_dst=dst_mac,
                     vlan_vid=vlan_tag | ofp.OFPVID_PRESENT)),
             call.uninstall_flows(
                 strict=True,
-                priority=4,
+                priority=20,
                 table_id=60,
                 match=ofpp.OFPMatch(
                     eth_dst=dst_mac,

neutron/tests/unit/plugins/ml2/drivers/openvswitch/agent/test_ovs_neutron_agent.py

@@ -12,6 +12,7 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+import contextlib
 import sys
 import time
 
@@ -1274,6 +1275,12 @@ class TestOvsNeutronAgent(object):
         self.agent.port_delete(context=None, port_id=TEST_PORT_ID1)
         self.agent.sg_agent = mock.Mock()
         self.agent.int_br = mock.Mock()
+
+        @contextlib.contextmanager
+        def bridge_deferred(*args, **kwargs):
+            yield
+
+        self.agent.int_br.deferred = mock.Mock(side_effect=bridge_deferred)
         self.agent.process_deleted_ports(port_info={})
         self.assertEqual(set(), self.agent.network_ports[TEST_NETWORK_ID1])
 

neutron/tests/unit/plugins/ml2/drivers/openvswitch/agent/test_ovs_tunnel.py

@@ -88,6 +88,7 @@ class TunnelTest(object):
                              'neutron.agent.firewall.NoopFirewallDriver',
                              group='SECURITYGROUP')
         cfg.CONF.set_override('report_interval', 0, 'AGENT')
+        cfg.CONF.set_override('explicitly_egress_direct', True, 'AGENT')
 
         self.INT_BRIDGE = 'integration_bridge'
         self.TUN_BRIDGE = 'tunnel_bridge'
@@ -583,8 +584,14 @@ class TunnelTest(object):
 
         self.mock_int_bridge_expected += [
             mock.call.check_canary_table(),
+            mock.call.deferred(full_ordered=True, use_bundle=True),
+            mock.call.deferred().__enter__(),
+            mock.call.deferred().__exit__(None, None, None),
             mock.call.cleanup_flows(),
-            mock.call.check_canary_table()
+            mock.call.check_canary_table(),
+            mock.call.deferred(full_ordered=True, use_bundle=True),
+            mock.call.deferred().__enter__(),
+            mock.call.deferred().__exit__(None, None, None),
         ]
         self.mock_map_tun_bridge_expected += [
             mock.call.cleanup_flows(),

releasenotes/notes/accepted_egress_direct-cc23873e213c6919.yaml

@@ -0,0 +1,20 @@
+---
+fixes:
+  - |
+    Bug https://bugs.launchpad.net/neutron/+bug/1732067 described a flooding
+    issue on the neutron-ovs-agent integration bridge. And bug
+    https://bugs.launchpad.net/neutron/+bug/1841622 proposed a solution for
+    it. The accepted egress packets will be taken care in the final egress
+    tables (61 when openflow firewall is not enabled, table 94 otherwise) with
+    direct output flows for unicast traffic with a minimum influence on the
+    existing cloud networking.
+    A new config option ``explicitly_egress_direct``, with default value False,
+    was added for the aim of distinguishing clouds which are running the
+    network node mixed with compute services, upstream neutron CI should be
+    an example. In such situation, this ``explicitly_egress_direct`` should be
+    set to False, because there are numerous cases from HA routers which can
+    not be covered, particularly when you have centralized floating IPs running
+    in such mixed hosts.
+    Otherwise, set ``explicitly_egress_direct`` to True to avoid the flooding.
+    One more note is if your network nodes are for networing services only, we
+    recommand you disable all the security_group to get a higher performance.