Support dynamic ofport on OVS gateway/tunnel port (#3976)

Originally the OpenFlow entries are using static port numbers for
gateway/tunnel/uplink which are used in the ofport_request fields in OVS
port creation. This is possible to introduce traffic issues if the
actual numbers assigned by OVS are different from the requests.

With this change, the predefined port numbers are still used in the OVS
port creation request, and save the actual port numbers in NodeConfig
after the OVS ports are created. And for the OpenFlow entries, the
actual port numbers are used.

Signed-off-by: wenyingd <wenyingd@vmware.com>

Author: wenyingd

cmd/antrea-agent/agent.go

@@ -348,7 +348,9 @@ func run(o *Options) error {
 		asyncRuleDeleteInterval,
 		o.dnsServerOverride,
 		v4Enabled,
-		v6Enabled)
+		v6Enabled,
+		nodeConfig.GatewayConfig.OFPort,
+		nodeConfig.TunnelOFPort)
 	if err != nil {
 		return fmt.Errorf("error creating new NetworkPolicy controller: %v", err)
 	}

pkg/agent/agent.go

@@ -608,11 +608,17 @@ func (i *Initializer) setupGatewayInterface() error {
 		}
 		gwPortUUID, err := i.ovsBridgeClient.CreateInternalPort(i.hostGateway, config.HostGatewayOFPort, externalIDs)
 		if err != nil {
-			klog.Errorf("Failed to create gateway port %s on OVS bridge: %v", i.hostGateway, err)
+			klog.ErrorS(err, "Failed to create gateway port on OVS bridge", "port", i.hostGateway)
 			return err
 		}
+		gwPort, err := i.ovsBridgeClient.GetOFPort(i.hostGateway, false)
+		if err != nil {
+			klog.ErrorS(err, "Failed to get gateway ofport", "port", i.hostGateway)
+			return err
+		}
+		klog.InfoS("Allocated OpenFlow port for gateway interface", "port", i.hostGateway, "ofPort", gwPort)
 		gatewayIface = interfacestore.NewGatewayInterface(i.hostGateway)
-		gatewayIface.OVSPortConfig = &interfacestore.OVSPortConfig{PortUUID: gwPortUUID, OFPort: config.HostGatewayOFPort}
+		gatewayIface.OVSPortConfig = &interfacestore.OVSPortConfig{PortUUID: gwPortUUID, OFPort: gwPort}
 		i.ifaceStore.AddInterface(gatewayIface)
 	} else {
 		klog.V(2).Infof("Gateway port %s already exists on OVS bridge", i.hostGateway)
@@ -657,7 +663,7 @@ func (i *Initializer) configureGatewayInterface(gatewayIface *interfacestore.Int
 		return err
 	}
 
-	i.nodeConfig.GatewayConfig = &config.GatewayConfig{Name: i.hostGateway, MAC: gwMAC}
+	i.nodeConfig.GatewayConfig = &config.GatewayConfig{Name: i.hostGateway, MAC: gwMAC, OFPort: uint32(gatewayIface.OFPort)}
 	gatewayIface.MAC = gwMAC
 	gatewayIface.IPs = []net.IP{}
 	if i.networkConfig.TrafficEncapMode.IsNetworkPolicyOnly() {
@@ -728,6 +734,7 @@ func (i *Initializer) setupDefaultTunnelInterface() error {
 				}
 				tunnelIface.TunnelInterfaceConfig.Csum = true
 			}
+			i.nodeConfig.TunnelOFPort = uint32(tunnelIface.OFPort)
 			return nil
 		}
 
@@ -755,12 +762,19 @@ func (i *Initializer) setupDefaultTunnelInterface() error {
 		}
 		tunnelPortUUID, err := i.ovsBridgeClient.CreateTunnelPortExt(tunnelPortName, i.networkConfig.TunnelType, config.DefaultTunOFPort, shouldEnableCsum, localIPStr, "", "", "", nil, externalIDs)
 		if err != nil {
-			klog.Errorf("Failed to create tunnel port %s type %s on OVS bridge: %v", tunnelPortName, i.networkConfig.TunnelType, err)
+			klog.ErrorS(err, "Failed to create tunnel port on OVS bridge", "port", tunnelPortName, "type", i.networkConfig.TunnelType)
 			return err
 		}
+		tunPort, err := i.ovsBridgeClient.GetOFPort(tunnelPortName, false)
+		if err != nil {
+			klog.ErrorS(err, "Failed to get tunnel ofport on OVS bridge", "port", tunnelPortName, "type", i.networkConfig.TunnelType)
+			return err
+		}
+		klog.InfoS("Allocated OpenFlow port for tunnel interface", "port", tunnelPortName, "ofPort", tunPort)
 		tunnelIface = interfacestore.NewTunnelInterface(tunnelPortName, i.networkConfig.TunnelType, localIP, shouldEnableCsum)
-		tunnelIface.OVSPortConfig = &interfacestore.OVSPortConfig{PortUUID: tunnelPortUUID, OFPort: config.DefaultTunOFPort}
+		tunnelIface.OVSPortConfig = &interfacestore.OVSPortConfig{PortUUID: tunnelPortUUID, OFPort: tunPort}
 		i.ifaceStore.AddInterface(tunnelIface)
+		i.nodeConfig.TunnelOFPort = uint32(tunPort)
 	}
 	return nil
 }
@@ -1126,3 +1140,24 @@ func (i *Initializer) patchNodeAnnotations(nodeName, key string, value interface
 func (i *Initializer) getNodeInterfaceFromIP(nodeIPs *utilip.DualStackIPs) (v4IPNet *net.IPNet, v6IPNet *net.IPNet, iface *net.Interface, err error) {
 	return getIPNetDeviceFromIP(nodeIPs, sets.NewString(i.hostGateway))
 }
+
+// getFreeOFPort returns an OpenFlow port number which is not used by any existing OVS port. Note that, the returned port
+// is not saved in OVSDB yet before the real port is created, so it might introduce an issue for the same return value
+// if it is called multiple times before OVS port creation.
+func (i *Initializer) getFreeOFPort(startPort int) (int32, error) {
+	existingOFPorts := sets.NewInt32()
+	ports, err := i.ovsBridgeClient.GetPortList()
+	if err != nil {
+		return 0, err
+	}
+	for _, p := range ports {
+		existingOFPorts.Insert(p.OFPort)
+	}
+	port := int32(startPort)
+	for ; ; port++ {
+		if !existingOFPorts.Has(port) {
+			break
+		}
+	}
+	return port, nil
+}

pkg/agent/agent_linux.go

@@ -92,7 +92,8 @@ func (i *Initializer) prepareOVSBridge() error {
 			klog.V(2).Infof("Found ports from OVS bridge: %+v", ports)
 			var uplinkPort *ovsconfig.OVSPortData
 			for index := range ports {
-				if ports[index].OFPort == config.UplinkOFPort {
+				antreaIfaceType, _ := ports[index].ExternalIDs[interfacestore.AntreaInterfaceTypeKey]
+				if antreaIfaceType == interfacestore.AntreaUplink {
 					uplinkPort = &ports[index]
 					break
 				}
@@ -105,10 +106,11 @@ func (i *Initializer) prepareOVSBridge() error {
 				return fmt.Errorf("cannot find uplink port %s: err=%w", uplinkPort.Name, err2)
 			}
 			klog.Infof("Found uplink device %s", adapter.Name)
+			uplinkNetConfig.OFPort = uint32(uplinkPort.OFPort)
 			uplinkNetConfig.Name = adapter.Name
 			uplinkNetConfig.Index = adapter.Index
 			uplinkInterface := interfacestore.NewUplinkInterface(uplinkPort.Name)
-			uplinkInterface.OVSPortConfig = &interfacestore.OVSPortConfig{uplinkPort.UUID, config.UplinkOFPort} //nolint: govet
+			uplinkInterface.OVSPortConfig = &interfacestore.OVSPortConfig{uplinkPort.UUID, uplinkPort.OFPort} //nolint: govet
 			i.ifaceStore.AddInterface(uplinkInterface)
 		}
 	} else {
@@ -122,6 +124,17 @@ func (i *Initializer) prepareOVSBridge() error {
 		}
 	}
 
+	i.nodeConfig.HostInterfaceOFPort = config.BridgeOFPort
+	if uplinkNetConfig.OFPort == 0 {
+		freePort, err := i.getFreeOFPort(config.UplinkOFPort)
+		if err != nil {
+			klog.ErrorS(err, "Failed to find a free port on OVS")
+			return err
+		}
+		uplinkNetConfig.OFPort = uint32(freePort)
+		klog.InfoS("Set OpenFlow port in UplinkNetConfig", "ofport", freePort)
+	}
+
 	return nil
 }
 
@@ -197,18 +210,23 @@ func (i *Initializer) ConnectUplinkToOVSBridge() error {
 
 	// If uplink is already exists, return.
 	uplink := uplinkNetConfig.Name
-	if _, err := i.ovsBridgeClient.GetOFPort(uplink, false); err == nil {
-		klog.Infof("Uplink %s already exists, skip the configuration", uplink)
+	if uplinkOFPort, err := i.ovsBridgeClient.GetOFPort(uplink, false); err == nil {
+		klog.InfoS("Uplink already exists, skip the configuration", "uplink", uplink, "port", uplinkOFPort)
 		return nil
 	}
 	// Create uplink port.
-	uplinkPortUUID, err := i.ovsBridgeClient.CreateUplinkPort(uplink, config.UplinkOFPort, map[string]interface{}{interfacestore.AntreaInterfaceTypeKey: interfacestore.AntreaUplink})
+	uplinkPortUUID, err := i.ovsBridgeClient.CreateUplinkPort(uplink, int32(i.nodeConfig.UplinkNetConfig.OFPort), map[string]interface{}{interfacestore.AntreaInterfaceTypeKey: interfacestore.AntreaUplink})
 	if err != nil {
 		return fmt.Errorf("failed to add uplink port %s: err=%w", uplink, err)
 	}
 	// Add newly created uplinkInterface to interface cache. This will be overwritten by initInterfaceStore.
 	uplinkInterface := interfacestore.NewUplinkInterface(uplink)
-	uplinkInterface.OVSPortConfig = &interfacestore.OVSPortConfig{uplinkPortUUID, config.UplinkOFPort} //nolint: govet
+	uplinkOFPort, err := i.ovsBridgeClient.GetOFPort(uplink, false)
+	if err != nil {
+		return fmt.Errorf("failed to get uplink ofport %s: err=%w", uplink, err)
+	}
+	klog.InfoS("Allocated OpenFlow port for uplink interface", "port", uplink, "ofPort", uplinkOFPort)
+	uplinkInterface.OVSPortConfig = &interfacestore.OVSPortConfig{uplinkPortUUID, uplinkOFPort} //nolint: govet
 	i.ifaceStore.AddInterface(uplinkInterface)
 
 	// Move network configuration of uplink interface to OVS bridge local interface.

pkg/agent/agent_windows.go

@@ -158,19 +158,33 @@ func (i *Initializer) prepareOVSBridge() error {
 	// If uplink is already exists, return.
 	uplinkNetConfig := i.nodeConfig.UplinkNetConfig
 	uplink := uplinkNetConfig.Name
-	if _, err = i.ovsBridgeClient.GetOFPort(uplink, false); err == nil {
-		klog.Infof("Uplink %s already exists, skip the configuration", uplink)
-		return err
+	if ofport, err := i.ovsBridgeClient.GetOFPort(uplink, false); err == nil {
+		klog.InfoS("Uplink already exists, skip the configuration", "uplink", uplink, "port", ofport)
+		i.nodeConfig.UplinkNetConfig.OFPort = uint32(ofport)
+		i.nodeConfig.HostInterfaceOFPort = config.BridgeOFPort
+		return nil
 	}
 	// Create uplink port.
+	freePort, err := i.getFreeOFPort(config.UplinkOFPort)
+	if err != nil {
+		klog.ErrorS(err, "Failed to find a free port on OVS")
+		return err
+	}
 	var uplinkPortUUID string
-	uplinkPortUUID, err = i.ovsBridgeClient.CreateUplinkPort(uplink, config.UplinkOFPort, nil)
+	uplinkPortUUID, err = i.ovsBridgeClient.CreateUplinkPort(uplink, freePort, nil)
 	if err != nil {
 		klog.Errorf("Failed to add uplink port %s: %v", uplink, err)
 		return err
 	}
+	uplinkOFPort, err := i.ovsBridgeClient.GetOFPort(uplink, false)
+	if err != nil {
+		return fmt.Errorf("failed to get uplink ofport %s: err=%w", uplink, err)
+	}
+	klog.InfoS("Allocated OpenFlow port for uplink interface", "port", uplink, "ofPort", uplinkOFPort)
+	i.nodeConfig.UplinkNetConfig.OFPort = uint32(uplinkOFPort)
+	i.nodeConfig.HostInterfaceOFPort = config.BridgeOFPort
 	uplinkInterface := interfacestore.NewUplinkInterface(uplink)
-	uplinkInterface.OVSPortConfig = &interfacestore.OVSPortConfig{uplinkPortUUID, config.UplinkOFPort} //nolint: govet
+	uplinkInterface.OVSPortConfig = &interfacestore.OVSPortConfig{uplinkPortUUID, uplinkOFPort} //nolint: govet
 	i.ifaceStore.AddInterface(uplinkInterface)
 	ovsCtlClient := ovsctl.NewClient(i.ovsBridge)
 
@@ -184,7 +198,7 @@ func (i *Initializer) prepareOVSBridge() error {
 	}
 	// Set the uplink with "no-flood" config, so that the IP of local Pods and "antrea-gw0" will not be leaked to the
 	// underlay network by the "normal" flow entry.
-	if err = ovsCtlClient.SetPortNoFlood(config.UplinkOFPort); err != nil {
+	if err = ovsCtlClient.SetPortNoFlood(int(uplinkOFPort)); err != nil {
 		klog.Errorf("Failed to set the uplink port with no-flood config: %v", err)
 		return err
 	}

pkg/agent/config/node_config.go

@@ -73,6 +73,9 @@ type GatewayConfig struct {
 	MAC  net.HardwareAddr
 	// LinkIndex is the link index of host gateway.
 	LinkIndex int
+
+	// OFPort is the OpenFlow port number of host gateway allocated by OVS.
+	OFPort uint32
 }
 
 func (g *GatewayConfig) String() string {
@@ -87,6 +90,8 @@ type AdapterNetConfig struct {
 	Gateway    string
 	DNSServers string
 	Routes     []interface{}
+	// OFPort is the OpenFlow port number of the uplink interface allocated by OVS.
+	OFPort uint32
 }
 
 type WireGuardConfig struct {
@@ -136,6 +141,12 @@ type NodeConfig struct {
 	// Auto discovery will use MTU value of the Node's primary interface.
 	// For Encap and Hybrid mode, Node MTU will be adjusted to account for encap header.
 	NodeMTU int
+	// TunnelOFPort is the OpenFlow port number of tunnel interface allocated by OVS. With noEncap mode, the value is 0.
+	TunnelOFPort uint32
+	// HostInterfaceOFPort is the OpenFlow port number of the host interface allocated by OVS. The host interface is the
+	// one which the IP/MAC of the uplink is moved to. If the host interface is the OVS bridge interface (br-int), the
+	// value is config.BridgeOFPort.
+	HostInterfaceOFPort uint32
 	// The config of the gateway interface on the OVS bridge.
 	GatewayConfig *GatewayConfig
 	// The config of the OVS bridge uplink interface. Only for Windows Node.

pkg/agent/controller/networkpolicy/fqdn.go

@@ -32,7 +32,6 @@ import (
 	"k8s.io/client-go/util/workqueue"
 	"k8s.io/klog/v2"
 
-	"antrea.io/antrea/pkg/agent/config"
 	"antrea.io/antrea/pkg/agent/openflow"
 	"antrea.io/antrea/pkg/agent/types"
 	binding "antrea.io/antrea/pkg/ovs/openflow"
@@ -150,9 +149,10 @@ type fqdnController struct {
 	selectorItemToRuleIDs map[fqdnSelectorItem]sets.String
 	ipv4Enabled           bool
 	ipv6Enabled           bool
+	gwPort                uint32
 }
 
-func newFQDNController(client openflow.Client, allocator *idAllocator, dnsServerOverride string, dirtyRuleHandler func(string), v4Enabled, v6Enabled bool) (*fqdnController, error) {
+func newFQDNController(client openflow.Client, allocator *idAllocator, dnsServerOverride string, dirtyRuleHandler func(string), v4Enabled, v6Enabled bool, gwPort uint32) (*fqdnController, error) {
 	controller := &fqdnController{
 		ofClient:               client,
 		dirtyRuleHandler:       dirtyRuleHandler,
@@ -166,6 +166,7 @@ func newFQDNController(client openflow.Client, allocator *idAllocator, dnsServer
 		selectorItemToRuleIDs:  map[fqdnSelectorItem]sets.String{},
 		ipv4Enabled:            v4Enabled,
 		ipv6Enabled:            v6Enabled,
+		gwPort:                 gwPort,
 	}
 	if controller.ofClient != nil {
 		if err := controller.ofClient.NewDNSpacketInConjunction(dnsInterceptRuleID); err != nil {
@@ -809,6 +810,16 @@ func (f *fqdnController) sendDNSPacketout(pktIn *ofctrl.PacketIn) error {
 		}
 	}
 	if prot == protocol.Type_UDP {
+		inPort := f.gwPort
+		if inPort == 0 {
+			// Use the original in_port number in the packetIn message to avoid an invalid input port number. Note that,
+			// this should not happen in container case as antrea-gw0 always exists. This check is for security.
+			matches := pktIn.GetMatches()
+			inPortField := matches.GetMatchByName("OXM_OF_IN_PORT")
+			if inPortField != nil {
+				inPort = inPortField.GetValue().(uint32)
+			}
+		}
 		udpSrcPort, udpDstPort, err := binding.GetUDPHeaderData(pktIn.Data.Data)
 		if err != nil {
 			klog.ErrorS(err, "Failed to get UDP header data")
@@ -822,7 +833,7 @@ func (f *fqdnController) sendDNSPacketout(pktIn *ofctrl.PacketIn) error {
 			pktIn.Data.HWDst.String(),
 			srcIP,
 			dstIP,
-			uint32(config.HostGatewayOFPort),
+			inPort,
 			0,
 			isIPv6,
 			udpSrcPort,

pkg/agent/controller/networkpolicy/fqdn_test.go

@@ -24,6 +24,7 @@ import (
 	"github.com/stretchr/testify/require"
 	"k8s.io/apimachinery/pkg/util/sets"
 
+	"antrea.io/antrea/pkg/agent/config"
 	openflowtest "antrea.io/antrea/pkg/agent/openflow/testing"
 )
 
@@ -42,6 +43,7 @@ func newMockFQDNController(t *testing.T, controller *gomock.Controller, dnsServe
 		dirtyRuleHandler,
 		true,
 		false,
+		config.HostGatewayOFPort,
 	)
 	require.NoError(t, err)
 	return f, mockOFClient

pkg/agent/controller/networkpolicy/networkpolicy_controller.go

@@ -111,6 +111,8 @@ type Controller struct {
 	ifaceStore            interfacestore.InterfaceStore
 	// denyConnStore is for storing deny connections for flow exporter.
 	denyConnStore *connections.DenyConnectionStore
+	gwPort        uint32
+	tunPort       uint32
 }
 
 // NewNetworkPolicyController returns a new *Controller.
@@ -129,7 +131,8 @@ func NewNetworkPolicyController(antreaClientGetter agent.AntreaClientProvider,
 	asyncRuleDeleteInterval time.Duration,
 	dnsServerOverride string,
 	v4Enabled bool,
-	v6Enabled bool) (*Controller, error) {
+	v6Enabled bool,
+	gwPort, tunPort uint32) (*Controller, error) {
 	idAllocator := newIDAllocator(asyncRuleDeleteInterval, dnsInterceptRuleID)
 	c := &Controller{
 		antreaClientProvider: antreaClientGetter,
@@ -140,11 +143,13 @@ func NewNetworkPolicyController(antreaClientGetter agent.AntreaClientProvider,
 		statusManagerEnabled: statusManagerEnabled,
 		multicastEnabled:     multicastEnabled,
 		loggingEnabled:       loggingEnabled,
+		gwPort:               gwPort,
+		tunPort:              tunPort,
 	}
 
 	if antreaPolicyEnabled {
 		var err error
-		if c.fqdnController, err = newFQDNController(ofClient, idAllocator, dnsServerOverride, c.enqueueRule, v4Enabled, v6Enabled); err != nil {
+		if c.fqdnController, err = newFQDNController(ofClient, idAllocator, dnsServerOverride, c.enqueueRule, v4Enabled, v6Enabled, gwPort); err != nil {
 			return nil, err
 		}
 

pkg/agent/controller/networkpolicy/networkpolicy_controller_test.go

@@ -32,6 +32,7 @@ import (
 	k8stesting "k8s.io/client-go/testing"
 	"k8s.io/component-base/metrics/legacyregistry"
 
+	"antrea.io/antrea/pkg/agent/config"
 	"antrea.io/antrea/pkg/agent/metrics"
 	"antrea.io/antrea/pkg/agent/openflow"
 	proxytypes "antrea.io/antrea/pkg/agent/proxy/types"
@@ -60,7 +61,7 @@ func newTestController() (*Controller, *fake.Clientset, *mockReconciler) {
 	ch2 := make(chan string, 100)
 	groupIDAllocator := openflow.NewGroupAllocator(false)
 	groupCounters := []proxytypes.GroupCounter{proxytypes.NewGroupCounter(groupIDAllocator, ch2)}
-	controller, _ := NewNetworkPolicyController(&antreaClientGetter{clientset}, nil, nil, "node1", podUpdateChannel, groupCounters, ch2, true, true, true, false, true, testAsyncDeleteInterval, "8.8.8.8:53", true, false)
+	controller, _ := NewNetworkPolicyController(&antreaClientGetter{clientset}, nil, nil, "node1", podUpdateChannel, groupCounters, ch2, true, true, true, false, true, testAsyncDeleteInterval, "8.8.8.8:53", true, false, config.HostGatewayOFPort, config.DefaultTunOFPort)
 	reconciler := newMockReconciler()
 	controller.reconciler = reconciler
 	controller.antreaPolicyLogger = nil