Multi-cluster support with networkPolicyOnly mode

In order to support multi-cluster traffic when the member cluster is
deployed with networkPolicyOnly mode, antrea-agent will be responsible to do
following things:

1. Create tunnel interface `antrea-tun0` for cross-cluster traffic
2. Watch all Pods on the Gateway and set up one rule per Pod in L3Fowarding
table as long as the Pod is running in a general Node instead of the Gateway.
3. Update container interface's MTU to minus tunnel overload.

Signed-off-by: Lan Luo <luola@vmware.com>

cmd/antrea-agent/agent.go

@@ -111,6 +111,7 @@ func run(o *Options) error {
 	serviceInformer := informerFactory.Core().V1().Services()
 	endpointsInformer := informerFactory.Core().V1().Endpoints()
 	namespaceInformer := informerFactory.Core().V1().Namespaces()
+	podInformer := informerFactory.Core().V1().Pods()
 
 	// Create Antrea Clientset for the given config.
 	antreaClientProvider := agent.NewAntreaClientProvider(o.config.AntreaClientConnection, k8sClient)
@@ -134,6 +135,7 @@ func run(o *Options) error {
 	enableBridgingMode := enableAntreaIPAM && o.config.EnableBridgingMode
 	// Bridging mode will connect the uplink interface to the OVS bridge.
 	connectUplinkToBridge := enableBridgingMode
+	multiclusterEnabled := features.DefaultFeatureGate.Enabled(features.Multicluster) && o.config.Multicluster.Enable
 
 	ovsDatapathType := ovsconfig.OVSDatapathType(o.config.OVSDatapathType)
 	ovsBridgeClient := ovsconfig.NewOVSBridge(o.config.OVSBridge, ovsDatapathType, ovsdbConnection)
@@ -148,7 +150,7 @@ func run(o *Options) error {
 		connectUplinkToBridge,
 		multicastEnabled,
 		features.DefaultFeatureGate.Enabled(features.TrafficControl),
-		features.DefaultFeatureGate.Enabled(features.Multicluster),
+		multiclusterEnabled,
 	)
 
 	var serviceCIDRNet *net.IPNet
@@ -248,7 +250,8 @@ func run(o *Options) error {
 		o.config.ExternalNode.ExternalNodeNamespace,
 		features.DefaultFeatureGate.Enabled(features.AntreaProxy),
 		o.config.AntreaProxy.ProxyAll,
-		connectUplinkToBridge)
+		connectUplinkToBridge,
+		multiclusterEnabled)
 	err = agentInitializer.Initialize()
 	if err != nil {
 		return fmt.Errorf("error initializing agent: %v", err)
@@ -281,8 +284,10 @@ func run(o *Options) error {
 
 	var mcRouteController *mcroute.MCRouteController
 	var mcInformerFactory mcinformers.SharedInformerFactory
+	var mcPolicyOnlyRouteController *mcroute.MCWithPolicyOnlyNodeRouteController
 
-	if features.DefaultFeatureGate.Enabled(features.Multicluster) && o.config.Multicluster.Enable {
+	isNetworkPolicyOnly := networkConfig.TrafficEncapMode.IsNetworkPolicyOnly()
+	if multiclusterEnabled {
 		mcNamespace := env.GetPodNamespace()
 		if o.config.Multicluster.Namespace != "" {
 			mcNamespace = o.config.Multicluster.Namespace
@@ -300,6 +305,19 @@ func run(o *Options) error {
 			nodeConfig,
 			mcNamespace,
 		)
+
+		if isNetworkPolicyOnly {
+			mcPolicyOnlyRouteController = mcroute.NewMCWithPolicyOnlyNodeRouteController(
+				mcClient,
+				podInformer.Informer(),
+				gwInformer.Informer(),
+				ofClient,
+				ovsBridgeClient,
+				ifaceStore,
+				nodeConfig,
+				mcNamespace,
+			)
+		}
 	}
 	var groupCounters []proxytypes.GroupCounter
 	groupIDUpdates := make(chan string, 100)
@@ -441,10 +459,7 @@ func run(o *Options) error {
 	var externalNodeController *externalnode.ExternalNodeController
 	var localExternalNodeInformer cache.SharedIndexInformer
 	if o.nodeType == config.K8sNode {
-		isChaining := false
-		if networkConfig.TrafficEncapMode.IsNetworkPolicyOnly() {
-			isChaining = true
-		}
+		isChaining := isNetworkPolicyOnly
 		cniServer = cniserver.New(
 			o.config.CNISocket,
 			o.config.HostProcPathPrefix,
@@ -455,6 +470,7 @@ func run(o *Options) error {
 			enableBridgingMode,
 			enableAntreaIPAM,
 			o.config.DisableTXChecksumOffload,
+			multiclusterEnabled,
 			networkReadyCh)
 
 		if features.DefaultFeatureGate.Enabled(features.SecondaryNetwork) {
@@ -724,9 +740,12 @@ func run(o *Options) error {
 		go mcastController.Run(stopCh)
 	}
 
-	if features.DefaultFeatureGate.Enabled(features.Multicluster) && o.config.Multicluster.Enable {
+	if multiclusterEnabled {
 		mcInformerFactory.Start(stopCh)
 		go mcRouteController.Run(stopCh)
+		if isNetworkPolicyOnly {
+			go mcPolicyOnlyRouteController.Run(stopCh)
+		}
 	}
 
 	// statsCollector collects stats and reports to the antrea-controller periodically. For now it's only used for

cmd/antrea-agent/options.go

@@ -454,9 +454,9 @@ func (o *Options) validateK8sNodeOptions() error {
 		}
 	}
 	if (features.DefaultFeatureGate.Enabled(features.Multicluster) || o.config.Multicluster.Enable) &&
-		encapMode != config.TrafficEncapModeEncap {
-		// Only Encap mode is supported for Multi-cluster feature.
-		return fmt.Errorf("Multicluster is only applicable to the %s mode", config.TrafficEncapModeEncap)
+		!(encapMode == config.TrafficEncapModeEncap || encapMode == config.TrafficEncapModeNetworkPolicyOnly) {
+		// Only Encap or networkPolicyOnly is supported for Multi-cluster feature.
+		return fmt.Errorf("Multicluster is only applicable to the %s mode or %s mode", config.TrafficEncapModeEncap, config.TrafficEncapModeNetworkPolicyOnly)
 	}
 	if features.DefaultFeatureGate.Enabled(features.NodePortLocal) {
 		startPort, endPort, err := parsePortRange(o.config.NodePortLocal.PortRange)

multicluster/controllers/multicluster/resourceexport_controller.go

@@ -356,11 +356,11 @@ func (r *ResourceExportReconciler) refreshEndpointsResourceImport(
 		if len(svcResExport.Status.Conditions) > 0 {
 			if svcResExport.Status.Conditions[0].Status != corev1.ConditionTrue {
 				return newResImport, false, fmt.Errorf("corresponding Service type of ResourceExport " + svcResExportName.String() +
-					"has not been converged successfully, retry later")
+					" has not been converged successfully, retry later")
 			}
 		} else {
 			return newResImport, false, fmt.Errorf("corresponding Service type of ResourceExport " + svcResExportName.String() +
-				"has not been converged yet, retry later")
+				" has not been converged yet, retry later")
 		}
 	}
 

pkg/agent/agent.go

@@ -105,6 +105,7 @@ type Initializer struct {
 	// networkReadyCh should be closed once the Node's network is ready.
 	// The CNI server will wait for it before handling any CNI Add requests.
 	proxyAll              bool
+	enableMulticluster    bool
 	networkReadyCh        chan<- struct{}
 	stopCh                <-chan struct{}
 	nodeType              config.NodeType
@@ -132,6 +133,7 @@ func NewInitializer(
 	enableProxy bool,
 	proxyAll bool,
 	connectUplinkToBridge bool,
+	enableMulticluster bool,
 ) *Initializer {
 	return &Initializer{
 		ovsBridgeClient:       ovsBridgeClient,
@@ -154,6 +156,7 @@ func NewInitializer(
 		enableProxy:           enableProxy,
 		proxyAll:              proxyAll,
 		connectUplinkToBridge: connectUplinkToBridge,
+		enableMulticluster:    enableMulticluster,
 	}
 }
 
@@ -755,10 +758,12 @@ func (i *Initializer) setupDefaultTunnelInterface() error {
 	// It's not necessary for new Linux kernel versions with the following patch:
 	// https://github.com/torvalds/linux/commit/89e5c58fc1e2857ccdaae506fb8bc5fed57ee063.
 	shouldEnableCsum := i.networkConfig.TunnelCsum && (i.networkConfig.TunnelType == ovsconfig.GeneveTunnel || i.networkConfig.TunnelType == ovsconfig.VXLANTunnel)
+	tunnelInterfaceSupported := i.networkConfig.TrafficEncapMode.SupportsEncap() ||
+		(i.networkConfig.TrafficEncapMode.IsNetworkPolicyOnly() && i.enableMulticluster)
 
 	// Check the default tunnel port.
 	if portExists {
-		if i.networkConfig.TrafficEncapMode.SupportsEncap() &&
+		if tunnelInterfaceSupported &&
 			tunnelIface.TunnelInterfaceConfig.Type == i.networkConfig.TunnelType &&
 			tunnelIface.TunnelInterfaceConfig.DestinationPort == i.networkConfig.TunnelPort &&
 			tunnelIface.TunnelInterfaceConfig.LocalIP.Equal(localIP) {
@@ -775,7 +780,7 @@ func (i *Initializer) setupDefaultTunnelInterface() error {
 		}
 
 		if err := i.ovsBridgeClient.DeletePort(tunnelIface.PortUUID); err != nil {
-			if i.networkConfig.TrafficEncapMode.SupportsEncap() {
+			if tunnelInterfaceSupported {
 				return fmt.Errorf("failed to remove tunnel port %s with wrong tunnel type: %s", tunnelPortName, err)
 			}
 			klog.Errorf("Failed to remove tunnel port %s in NoEncapMode: %v", tunnelPortName, err)
@@ -786,7 +791,7 @@ func (i *Initializer) setupDefaultTunnelInterface() error {
 	}
 
 	// Create the default tunnel port and interface.
-	if i.networkConfig.TrafficEncapMode.SupportsEncap() {
+	if tunnelInterfaceSupported {
 		if tunnelPortName != defaultTunInterfaceName {
 			// Reset the tunnel interface name to the desired name before
 			// recreating the tunnel port and interface.

pkg/agent/cniserver/interface_configuration_linux.go

@@ -371,6 +371,42 @@ func (ic *ifConfigurator) configureContainerLink(
 	}
 }
 
+func (ic *ifConfigurator) configureContainerMTU(
+	podName string,
+	podNameSpace string,
+	containerID string,
+	containerNetNS string,
+	containerIFDev string,
+	hostInterfaceName string,
+	mtu int,
+) error {
+	updateMTU := func(interfaceName string) error {
+		link, err := netlink.LinkByName(interfaceName)
+		if err != nil {
+			return fmt.Errorf("failed to find interface %s: %v", interfaceName, err)
+		}
+		err = netlink.LinkSetMTU(link, mtu)
+		if err != nil {
+			return fmt.Errorf("failed to set MTU for interface %s: %v", interfaceName, err)
+		}
+		return nil
+	}
+
+	if err := ns.WithNetNSPath(containerNetNS, func(hostNS ns.NetNS) error {
+		if err := updateMTU(containerIFDev); err != nil {
+			return fmt.Errorf("error when updating container interface MTU: %v", err)
+		}
+		return nil
+	}); err != nil {
+		return err
+	}
+
+	if err := updateMTU(hostInterfaceName); err != nil {
+		return fmt.Errorf("error when updating host interface MTU: %v", err)
+	}
+	return nil
+}
+
 func (ic *ifConfigurator) removeContainerLink(containerID, hostInterfaceName string) error {
 	klog.V(2).Infof("Deleting veth devices for container %s", containerID)
 	// Don't return an error if the device is already removed as CniDel can be called multiple times.

pkg/agent/cniserver/interface_configuration_windows.go

@@ -173,6 +173,24 @@ func (ic *ifConfigurator) configureContainerLink(
 	return nil
 }
 
+func (ic *ifConfigurator) configureContainerMTU(
+	podName string,
+	podNameSpace string,
+	containerID string,
+	containerNetNS string,
+	containerIFDev string,
+	hostInterfaceName string,
+	mtu int,
+) error {
+	infraContainerID := getInfraContainer(containerID, containerNetNS)
+	epName := util.GenerateContainerInterfaceName(podName, podNameSpace, infraContainerID)
+	ifaceName := util.VirtualAdapterName(epName)
+	if err := util.SetInterfaceMTU(ifaceName, mtu); err != nil {
+		return fmt.Errorf("failed to configure MTU on container interface '%s': %v", ifaceName, err)
+	}
+	return nil
+}
+
 // createContainerLink creates HNSEndpoint using the IP configuration in the IPAM result.
 func (ic *ifConfigurator) createContainerLink(endpointName string, result *current.Result, containerID, podName, podNamespace string) (hostLink *hcsshim.HNSEndpoint, err error) {
 	containerIP, err := findContainerIPConfig(result.IPs)

pkg/agent/cniserver/pod_configuration.go

@@ -196,6 +196,21 @@ func ParseOVSPortInterfaceConfig(portData *ovsconfig.OVSPortData, portConfig *in
 	return interfaceConfig
 }
 
+func (pc *podConfigurator) configureInterfacesMTU(
+	podName string,
+	podNameSpace string,
+	containerID string,
+	containerNetNS string,
+	containerIFDev string,
+	result *current.Result,
+	mtu int) error {
+	hostIface := result.Interfaces[0]
+	if err := pc.ifConfigurator.configureContainerMTU(podName, podNameSpace, containerID, containerNetNS, containerIFDev, hostIface.Name, mtu); err != nil {
+		return err
+	}
+	return nil
+}
+
 func (pc *podConfigurator) configureInterfaces(
 	podName string,
 	podNameSpace string,

pkg/agent/cniserver/server.go

@@ -117,6 +117,7 @@ type CNIServer struct {
 	enableSecondaryNetworkIPAM bool
 	disableTXChecksumOffload   bool
 	secondaryNetworkEnabled    bool
+	multiClusterEnabled        bool
 	// networkReadyCh notifies that the network is ready so new Pods can be created. Therefore, CmdAdd waits for it.
 	networkReadyCh <-chan struct{}
 }
@@ -624,7 +625,7 @@ func New(
 	nodeConfig *config.NodeConfig,
 	kubeClient clientset.Interface,
 	routeClient route.Interface,
-	isChaining, enableBridgingMode, enableSecondaryNetworkIPAM, disableTXChecksumOffload bool,
+	isChaining, enableBridgingMode, enableSecondaryNetworkIPAM, disableTXChecksumOffload, multiclusterEnabled bool,
 	networkReadyCh <-chan struct{},
 ) *CNIServer {
 	return &CNIServer{
@@ -640,6 +641,7 @@ func New(
 		enableBridgingMode:         enableBridgingMode,
 		disableTXChecksumOffload:   disableTXChecksumOffload,
 		enableSecondaryNetworkIPAM: enableSecondaryNetworkIPAM,
+		multiClusterEnabled:        multiclusterEnabled,
 		networkReadyCh:             networkReadyCh,
 	}
 }
@@ -716,6 +718,22 @@ func (s *CNIServer) interceptAdd(cniConfig *CNIConfig) (*cnipb.CniCmdResponse, e
 		return &cnipb.CniCmdResponse{CniResult: []byte("")}, fmt.Errorf("failed to connect container %s to ovs: %w", cniConfig.ContainerId, err)
 	}
 
+	if s.multiClusterEnabled {
+		// Antrea multi-cluster supports Geneve tunnel mode only.
+		mtu := cniConfig.MTU - config.GeneveOverhead
+		if err := s.podConfigurator.configureInterfacesMTU(
+			podName,
+			podNamespace,
+			cniConfig.ContainerId,
+			s.hostNetNsPath(cniConfig.Netns),
+			cniConfig.Ifname,
+			prevResult,
+			mtu,
+		); err != nil {
+			return &cnipb.CniCmdResponse{CniResult: []byte("")}, fmt.Errorf("failed to configure container %s's MTU: %w", cniConfig.ContainerId, err)
+		}
+	}
+
 	// we return prevResult, which should be exactly what we received from
 	// the runtime, potentially converted to the current CNI version used by
 	// Antrea.