Refine Endpoint selection for multi-cluster Service

When the Endpoint of Multi-cluster Service is a local Service ClusterIP,
refine the action to let it go to the corresponding exported Service's
group to do final Endpoint selection. This can avoid the case that the
traffic goes out of antrea-gw0 and goes back to OVS again when a
local Pod is trying to access a MC Service but a local Service's
Endpoint is selected.

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

build/charts/antrea/README.md

@@ -87,6 +87,7 @@ Kubernetes: `>= 1.16.0-0`
 | multicluster.enableGateway | bool | `false` | Enable Antrea Multi-cluster Gateway to support cross-cluster traffic. This feature is supported only with encap mode. |
 | multicluster.enableStretchedNetworkPolicy | bool | `false` | Enable StretchedNetworkPolicy which allows Antrea-native policies to select peers from other clusters in a ClusterSet. Multi-cluster Gateway must be enabled to enable StretchedNetworkPolicy. |
 | multicluster.namespace | string | `""` | The Namespace where Antrea Multi-cluster Controller is running. The default is antrea-agent's Namespace. |
+| multicluster.serviceCIDR | string | `""` | ClusterIP CIDR range for Services. It should be set to the same value as the one specified by --service-cluster-ip-range for kube-apiserver. |
 | noSNAT | bool | `false` | Whether or not to SNAT (using the Node IP) the egress traffic from a Pod to the external network. |
 | nodeIPAM.clusterCIDRs | list | `[]` | CIDR ranges to use when allocating Pod IP addresses. |
 | nodeIPAM.enable | bool | `false` | Enable Node IPAM in Antrea |

build/charts/antrea/conf/antrea-agent.conf

@@ -340,6 +340,9 @@ multicluster:
 # from other clusters in a ClusterSet.
 # Multi-cluster Gateway must be enabled to enable StretchedNetworkPolicy.
   enableStretchedNetworkPolicy: {{ .enableStretchedNetworkPolicy }}
+# ClusterIP CIDR range for Services. It should be set to the same value as the one specified
+# by --service-cluster-ip-range for kube-apiserver.
+  serviceCIDR: ""
 {{- end }}
 
 {{- if .Values.featureGates.SecondaryNetwork }}

build/charts/antrea/values.yaml

@@ -331,6 +331,9 @@ multicluster:
   # from other clusters in a ClusterSet.
   # Multi-cluster Gateway must be enabled to enable StretchedNetworkPolicy.
   enableStretchedNetworkPolicy: false
+  # -- ClusterIP CIDR range for Services. It should be set to the same value as the one specified
+  # by --service-cluster-ip-range for kube-apiserver.
+  serviceCIDR: ""
 
 testing:
   ## -- enable code coverage measurement (used when testing Antrea only).

build/yamls/antrea-aks.yml

@@ -3235,6 +3235,9 @@ data:
     # from other clusters in a ClusterSet.
     # Multi-cluster Gateway must be enabled to enable StretchedNetworkPolicy.
       enableStretchedNetworkPolicy: false
+    # ClusterIP CIDR range for Services. It should be set to the same value as the one specified
+    # by --service-cluster-ip-range for kube-apiserver.
+      serviceCIDR: ""
   antrea-cni.conflist: |
     {
         "cniVersion":"0.3.0",
@@ -4293,7 +4296,7 @@ spec:
         kubectl.kubernetes.io/default-container: antrea-agent
         # Automatically restart Pods with a RollingUpdate if the ConfigMap changes
         # See https://helm.sh/docs/howto/charts_tips_and_tricks/#automatically-roll-deployments
-        checksum/config: 5863dc7db71990f4adebcf03cd8311ce14d9a233129c909a872f2f14e0e022fa
+        checksum/config: b657fff0248b1eb6d0de88e87b7461f05cc14169157544e2c594107df99a6b7a
       labels:
         app: antrea
         component: antrea-agent
@@ -4534,7 +4537,7 @@ spec:
       annotations:
         # Automatically restart Pod if the ConfigMap changes
         # See https://helm.sh/docs/howto/charts_tips_and_tricks/#automatically-roll-deployments
-        checksum/config: 5863dc7db71990f4adebcf03cd8311ce14d9a233129c909a872f2f14e0e022fa
+        checksum/config: b657fff0248b1eb6d0de88e87b7461f05cc14169157544e2c594107df99a6b7a
       labels:
         app: antrea
         component: antrea-controller

build/yamls/antrea-eks.yml

@@ -3235,6 +3235,9 @@ data:
     # from other clusters in a ClusterSet.
     # Multi-cluster Gateway must be enabled to enable StretchedNetworkPolicy.
       enableStretchedNetworkPolicy: false
+    # ClusterIP CIDR range for Services. It should be set to the same value as the one specified
+    # by --service-cluster-ip-range for kube-apiserver.
+      serviceCIDR: ""
   antrea-cni.conflist: |
     {
         "cniVersion":"0.3.0",
@@ -4293,7 +4296,7 @@ spec:
         kubectl.kubernetes.io/default-container: antrea-agent
         # Automatically restart Pods with a RollingUpdate if the ConfigMap changes
         # See https://helm.sh/docs/howto/charts_tips_and_tricks/#automatically-roll-deployments
-        checksum/config: 5863dc7db71990f4adebcf03cd8311ce14d9a233129c909a872f2f14e0e022fa
+        checksum/config: b657fff0248b1eb6d0de88e87b7461f05cc14169157544e2c594107df99a6b7a
       labels:
         app: antrea
         component: antrea-agent
@@ -4536,7 +4539,7 @@ spec:
       annotations:
         # Automatically restart Pod if the ConfigMap changes
         # See https://helm.sh/docs/howto/charts_tips_and_tricks/#automatically-roll-deployments
-        checksum/config: 5863dc7db71990f4adebcf03cd8311ce14d9a233129c909a872f2f14e0e022fa
+        checksum/config: b657fff0248b1eb6d0de88e87b7461f05cc14169157544e2c594107df99a6b7a
       labels:
         app: antrea
         component: antrea-controller

build/yamls/antrea-gke.yml

@@ -3235,6 +3235,9 @@ data:
     # from other clusters in a ClusterSet.
     # Multi-cluster Gateway must be enabled to enable StretchedNetworkPolicy.
       enableStretchedNetworkPolicy: false
+    # ClusterIP CIDR range for Services. It should be set to the same value as the one specified
+    # by --service-cluster-ip-range for kube-apiserver.
+      serviceCIDR: ""
   antrea-cni.conflist: |
     {
         "cniVersion":"0.3.0",
@@ -4293,7 +4296,7 @@ spec:
         kubectl.kubernetes.io/default-container: antrea-agent
         # Automatically restart Pods with a RollingUpdate if the ConfigMap changes
         # See https://helm.sh/docs/howto/charts_tips_and_tricks/#automatically-roll-deployments
-        checksum/config: 91c75022e7c9a8203230275e9cccae335c2590ebd03432e97c62db5318bda8f1
+        checksum/config: 0df53ddddfc534c014af9646f5495e9eada38dac56608c2b7534bbc3111cdf38
       labels:
         app: antrea
         component: antrea-agent
@@ -4533,7 +4536,7 @@ spec:
       annotations:
         # Automatically restart Pod if the ConfigMap changes
         # See https://helm.sh/docs/howto/charts_tips_and_tricks/#automatically-roll-deployments
-        checksum/config: 91c75022e7c9a8203230275e9cccae335c2590ebd03432e97c62db5318bda8f1
+        checksum/config: 0df53ddddfc534c014af9646f5495e9eada38dac56608c2b7534bbc3111cdf38
       labels:
         app: antrea
         component: antrea-controller

build/yamls/antrea-ipsec.yml

@@ -3248,6 +3248,9 @@ data:
     # from other clusters in a ClusterSet.
     # Multi-cluster Gateway must be enabled to enable StretchedNetworkPolicy.
       enableStretchedNetworkPolicy: false
+    # ClusterIP CIDR range for Services. It should be set to the same value as the one specified
+    # by --service-cluster-ip-range for kube-apiserver.
+      serviceCIDR: ""
   antrea-cni.conflist: |
     {
         "cniVersion":"0.3.0",
@@ -4306,7 +4309,7 @@ spec:
         kubectl.kubernetes.io/default-container: antrea-agent
         # Automatically restart Pods with a RollingUpdate if the ConfigMap changes
         # See https://helm.sh/docs/howto/charts_tips_and_tricks/#automatically-roll-deployments
-        checksum/config: f5f411ec50782205ed224633b7ff5c9c2712e2ac10666bc78ff1d2929f850999
+        checksum/config: 93def1c832776b9a2a676f2557358a57074f142e0d2578a539bdfef207a1b4a8
         checksum/ipsec-secret: d0eb9c52d0cd4311b6d252a951126bf9bea27ec05590bed8a394f0f792dcb2a4
       labels:
         app: antrea
@@ -4592,7 +4595,7 @@ spec:
       annotations:
         # Automatically restart Pod if the ConfigMap changes
         # See https://helm.sh/docs/howto/charts_tips_and_tricks/#automatically-roll-deployments
-        checksum/config: f5f411ec50782205ed224633b7ff5c9c2712e2ac10666bc78ff1d2929f850999
+        checksum/config: 93def1c832776b9a2a676f2557358a57074f142e0d2578a539bdfef207a1b4a8
       labels:
         app: antrea
         component: antrea-controller

build/yamls/antrea.yml

@@ -3235,6 +3235,9 @@ data:
     # from other clusters in a ClusterSet.
     # Multi-cluster Gateway must be enabled to enable StretchedNetworkPolicy.
       enableStretchedNetworkPolicy: false
+    # ClusterIP CIDR range for Services. It should be set to the same value as the one specified
+    # by --service-cluster-ip-range for kube-apiserver.
+      serviceCIDR: ""
   antrea-cni.conflist: |
     {
         "cniVersion":"0.3.0",
@@ -4293,7 +4296,7 @@ spec:
         kubectl.kubernetes.io/default-container: antrea-agent
         # Automatically restart Pods with a RollingUpdate if the ConfigMap changes
         # See https://helm.sh/docs/howto/charts_tips_and_tricks/#automatically-roll-deployments
-        checksum/config: 78d8f9c7ce769c65649147b1c66f47998739b3b7b73f6981ccb488f320d04452
+        checksum/config: 0db09333704fa2a3550e5e7bbc6692d84beb2d8159e6b51fb0f0145ed2fe2a51
       labels:
         app: antrea
         component: antrea-agent
@@ -4533,7 +4536,7 @@ spec:
       annotations:
         # Automatically restart Pod if the ConfigMap changes
         # See https://helm.sh/docs/howto/charts_tips_and_tricks/#automatically-roll-deployments
-        checksum/config: 78d8f9c7ce769c65649147b1c66f47998739b3b7b73f6981ccb488f320d04452
+        checksum/config: 0db09333704fa2a3550e5e7bbc6692d84beb2d8159e6b51fb0f0145ed2fe2a51
       labels:
         app: antrea
         component: antrea-controller

cmd/antrea-agent/agent.go

@@ -323,6 +323,8 @@ func run(o *Options) error {
 	var mcRouteController *mcroute.MCRouteController
 	var mcStrechedNetworkPolicyController *mcroute.StretchedNetworkPolicyController
 	var mcInformerFactory mcinformers.SharedInformerFactory
+	var clusterServiceCIDR *net.IPNet
+
 	if enableMulticlusterGW {
 		mcNamespace := env.GetPodNamespace()
 		if o.config.Multicluster.Namespace != "" {
@@ -342,6 +344,7 @@ func run(o *Options) error {
 			mcNamespace,
 			o.config.Multicluster.EnableStretchedNetworkPolicy,
 		)
+		_, clusterServiceCIDR, _ = net.ParseCIDR(o.config.Multicluster.ServiceCIDR)
 	}
 	if enableMulticlusterNP {
 		labelIDInformer := mcInformerFactory.Multicluster().V1alpha1().LabelIdentities()
@@ -372,13 +375,13 @@ func run(o *Options) error {
 
 		switch {
 		case v4Enabled && v6Enabled:
-			proxier = proxy.NewDualStackProxier(nodeConfig.Name, informerFactory, ofClient, routeClient, nodePortAddressesIPv4, nodePortAddressesIPv6, proxyAll, skipServices, proxyLoadBalancerIPs, v4GroupCounter, v6GroupCounter)
+			proxier = proxy.NewDualStackProxier(nodeConfig.Name, informerFactory, ofClient, routeClient, nodePortAddressesIPv4, nodePortAddressesIPv6, proxyAll, skipServices, proxyLoadBalancerIPs, v4GroupCounter, v6GroupCounter, clusterServiceCIDR)
 			groupCounters = append(groupCounters, v4GroupCounter, v6GroupCounter)
 		case v4Enabled:
-			proxier = proxy.NewProxier(nodeConfig.Name, informerFactory, ofClient, false, routeClient, nodePortAddressesIPv4, proxyAll, skipServices, proxyLoadBalancerIPs, v4GroupCounter)
+			proxier = proxy.NewProxier(nodeConfig.Name, informerFactory, ofClient, false, routeClient, nodePortAddressesIPv4, proxyAll, skipServices, proxyLoadBalancerIPs, v4GroupCounter, clusterServiceCIDR)
 			groupCounters = append(groupCounters, v4GroupCounter)
 		case v6Enabled:
-			proxier = proxy.NewProxier(nodeConfig.Name, informerFactory, ofClient, true, routeClient, nodePortAddressesIPv6, proxyAll, skipServices, proxyLoadBalancerIPs, v6GroupCounter)
+			proxier = proxy.NewProxier(nodeConfig.Name, informerFactory, ofClient, true, routeClient, nodePortAddressesIPv6, proxyAll, skipServices, proxyLoadBalancerIPs, v6GroupCounter, clusterServiceCIDR)
 			groupCounters = append(groupCounters, v6GroupCounter)
 		default:
 			return fmt.Errorf("at least one of IPv4 or IPv6 should be enabled")

cmd/antrea-agent/options.go

@@ -324,6 +324,13 @@ func (o *Options) validateMulticlusterConfig(encapMode config.TrafficEncapModeTy
 		// Only Encap mode is supported for Multi-cluster Gateway.
 		return fmt.Errorf("Multicluster is only applicable to the %s mode", config.TrafficEncapModeEncap)
 	}
+
+	if o.config.Multicluster.ServiceCIDR != "" {
+		_, _, err := net.ParseCIDR(o.config.Multicluster.ServiceCIDR)
+		if err != nil {
+			return fmt.Errorf("serviceCIDR %s is invalid in Multicluster config", o.config.Multicluster.ServiceCIDR)
+		}
+	}
 	return nil
 }
 

cmd/antrea-agent/options_linux_test.go

@@ -71,6 +71,26 @@ func TestMulticlusterOptions(t *testing.T) {
 			featureGate: true,
 			expectedErr: "",
 		},
+		{
+			name: "EnableGateway and EnableStretchedNetworkPolicy with right serviceCIDR",
+			mcConfig: agentconfig.MulticlusterConfig{
+				EnableGateway:                true,
+				EnableStretchedNetworkPolicy: true,
+				ServiceCIDR:                  "10.96.0.1/24",
+			},
+			featureGate: true,
+			expectedErr: "",
+		},
+		{
+			name: "EnableGateway and EnableStretchedNetworkPolicy with wrong serviceCIDR",
+			mcConfig: agentconfig.MulticlusterConfig{
+				EnableGateway:                true,
+				EnableStretchedNetworkPolicy: true,
+				ServiceCIDR:                  "192.168.1.",
+			},
+			featureGate: true,
+			expectedErr: "serviceCIDR 192.168.1. is invalid in Multicluster config",
+		},
 		{
 			name: "EnableGateway false and EnableStretchedNetworkPolicy",
 			mcConfig: agentconfig.MulticlusterConfig{

pkg/agent/openflow/client.go

@@ -84,7 +84,7 @@ type Client interface {
 
 	// InstallServiceGroup installs a group for Service LB. Each endpoint
 	// is a bucket of the group. For now, each bucket has the same weight.
-	InstallServiceGroup(groupID binding.GroupIDType, withSessionAffinity bool, endpoints []proxy.Endpoint) error
+	InstallServiceGroup(groupID, exportedSvcGroupID binding.GroupIDType, withSessionAffinity bool, endpoints []proxy.Endpoint) error
 	// UninstallServiceGroup removes the group and its buckets that are
 	// installed by InstallServiceGroup.
 	UninstallServiceGroup(groupID binding.GroupIDType) error
@@ -618,11 +618,11 @@ func (c *client) GetPodFlowKeys(interfaceName string) []string {
 	return c.getFlowKeysFromCache(c.featurePodConnectivity.podCachedFlows, interfaceName)
 }
 
-func (c *client) InstallServiceGroup(groupID binding.GroupIDType, withSessionAffinity bool, endpoints []proxy.Endpoint) error {
+func (c *client) InstallServiceGroup(groupID, exportedSvcGroupID binding.GroupIDType, withSessionAffinity bool, endpoints []proxy.Endpoint) error {
 	c.replayMutex.RLock()
 	defer c.replayMutex.RUnlock()
 
-	group := c.featureService.serviceEndpointGroup(groupID, withSessionAffinity, endpoints...)
+	group := c.featureService.serviceEndpointGroup(groupID, exportedSvcGroupID, withSessionAffinity, endpoints...)
 	_, installed := c.featureService.groupCache.Load(groupID)
 	if !installed {
 		if err := c.ofEntryOperations.AddOFEntries([]binding.OFEntry{group}); err != nil {

pkg/agent/openflow/client_test.go

@@ -840,29 +840,45 @@ func Test_client_GetPodFlowKeys(t *testing.T) {
 
 func Test_client_InstallServiceGroup(t *testing.T) {
 	groupID := binding.GroupIDType(100)
+	zeroGroupID := binding.GroupIDType(0)
 
 	testCases := []struct {
 		name                string
 		withSessionAffinity bool
 		endpoints           []proxy.Endpoint
 		expectedGroup       string
+		exportedGroupID     binding.GroupIDType
 	}{
 		{
 			name: "IPv4 Endpoints",
 			endpoints: []proxy.Endpoint{
 				proxy.NewBaseEndpointInfo("10.10.0.100", "", "", 80, false, true, false, false, nil),
 				proxy.NewBaseEndpointInfo("10.10.0.101", "", "", 80, false, true, false, false, nil),
 			},
+			exportedGroupID: zeroGroupID,
 			expectedGroup: "group_id=100,type=select," +
 				"bucket=bucket_id:0,weight:100,actions=set_field:0xa0a0064->reg3,set_field:0x50/0xffff->reg4,resubmit:EndpointDNAT," +
 				"bucket=bucket_id:1,weight:100,actions=set_field:0xa0a0065->reg3,set_field:0x50/0xffff->reg4,resubmit:EndpointDNAT",
 		},
+		{
+			name: "IPv4 Endpoints with multi-cluster enabled",
+			endpoints: []proxy.Endpoint{
+				proxy.NewBaseEndpointInfo("10.10.0.100", "", "", 80, false, true, false, false, nil),
+				proxy.NewBaseEndpointInfo("10.10.0.101", "", "", 80, false, true, false, false, nil),
+			},
+			exportedGroupID: binding.GroupIDType(2),
+			expectedGroup: "group_id=100,type=select," +
+				"bucket=bucket_id:0,weight:100,actions=set_field:0xa0a0064->reg3,set_field:0x50/0xffff->reg4,resubmit:EndpointDNAT," +
+				"bucket=bucket_id:1,weight:100,actions=set_field:0xa0a0065->reg3,set_field:0x50/0xffff->reg4,resubmit:EndpointDNAT," +
+				"bucket=bucket_id:2,weight:100,actions=group:2",
+		},
 		{
 			name: "IPv6 Endpoints",
 			endpoints: []proxy.Endpoint{
 				proxy.NewBaseEndpointInfo("fec0:10:10::100", "", "", 80, false, true, false, false, nil),
 				proxy.NewBaseEndpointInfo("fec0:10:10::101", "", "", 80, false, true, false, false, nil),
 			},
+			exportedGroupID: zeroGroupID,
 			expectedGroup: "group_id=100,type=select," +
 				"bucket=bucket_id:0,weight:100,actions=set_field:0xfec00010001000000000000000000100->xxreg3,set_field:0x50/0xffff->reg4,resubmit:EndpointDNAT," +
 				"bucket=bucket_id:1,weight:100,actions=set_field:0xfec00010001000000000000000000101->xxreg3,set_field:0x50/0xffff->reg4,resubmit:EndpointDNAT",
@@ -874,6 +890,7 @@ func Test_client_InstallServiceGroup(t *testing.T) {
 				proxy.NewBaseEndpointInfo("10.10.0.100", "", "", 80, false, true, false, false, nil),
 				proxy.NewBaseEndpointInfo("10.10.0.101", "", "", 80, false, true, false, false, nil),
 			},
+			exportedGroupID: zeroGroupID,
 			expectedGroup: "group_id=100,type=select," +
 				"bucket=bucket_id:0,weight:100,actions=set_field:0xa0a0064->reg3,set_field:0x50/0xffff->reg4,resubmit:ServiceLB," +
 				"bucket=bucket_id:1,weight:100,actions=set_field:0xa0a0065->reg3,set_field:0x50/0xffff->reg4,resubmit:ServiceLB",
@@ -885,6 +902,7 @@ func Test_client_InstallServiceGroup(t *testing.T) {
 				proxy.NewBaseEndpointInfo("fec0:10:10::100", "", "", 80, false, true, false, false, nil),
 				proxy.NewBaseEndpointInfo("fec0:10:10::101", "", "", 80, false, true, false, false, nil),
 			},
+			exportedGroupID: zeroGroupID,
 			expectedGroup: "group_id=100,type=select," +
 				"bucket=bucket_id:0,weight:100,actions=set_field:0xfec00010001000000000000000000100->xxreg3,set_field:0x50/0xffff->reg4,resubmit:ServiceLB," +
 				"bucket=bucket_id:1,weight:100,actions=set_field:0xfec00010001000000000000000000101->xxreg3,set_field:0x50/0xffff->reg4,resubmit:ServiceLB",
@@ -896,14 +914,12 @@ func Test_client_InstallServiceGroup(t *testing.T) {
 			ctrl := gomock.NewController(t)
 			defer ctrl.Finish()
 			m := oftest.NewMockOFEntryOperations(ctrl)
-
 			fc := newFakeClient(m, true, true, config.K8sNode, config.TrafficEncapModeEncap)
 			defer resetPipelines()
 
 			m.EXPECT().AddOFEntries(gomock.Any()).Return(nil).Times(1)
 			m.EXPECT().DeleteOFEntries(gomock.Any()).Return(nil).Times(1)
-
-			assert.NoError(t, fc.InstallServiceGroup(groupID, tc.withSessionAffinity, tc.endpoints))
+			assert.NoError(t, fc.InstallServiceGroup(groupID, tc.exportedGroupID, tc.withSessionAffinity, tc.endpoints))
 			gCacheI, ok := fc.featureService.groupCache.Load(groupID)
 			require.True(t, ok)
 			group := getGroupFromCache(gCacheI.(binding.Group))

pkg/agent/openflow/pipeline.go

@@ -2428,7 +2428,9 @@ func (f *featureService) endpointDNATFlow(endpointIP net.IP, endpointPort uint16
 // serviceEndpointGroup creates/modifies the group/buckets of Endpoints. If the withSessionAffinity is true, then buckets
 // will resubmit packets back to ServiceLBTable to trigger the learn flow, the learn flow will then send packets to
 // EndpointDNATTable. Otherwise, buckets will resubmit packets to EndpointDNATTable directly.
-func (f *featureService) serviceEndpointGroup(groupID binding.GroupIDType, withSessionAffinity bool, endpoints ...proxy.Endpoint) binding.Group {
+// When exportedSvcGroupID is not zero, it means the corresponding Endpoint is a local exported Service's ClusterIP,
+// the action will go to the group of the exported Service.
+func (f *featureService) serviceEndpointGroup(groupID, exportedSvcGroupID binding.GroupIDType, withSessionAffinity bool, endpoints ...proxy.Endpoint) binding.Group {
 	group := f.bridge.CreateGroup(groupID).ResetBuckets()
 	var resubmitTableID uint8
 	if withSessionAffinity {
@@ -2458,6 +2460,11 @@ func (f *featureService) serviceEndpointGroup(groupID binding.GroupIDType, withS
 				Done()
 		}
 	}
+	if exportedSvcGroupID != 0 {
+		group = group.Bucket().Weight(100).
+			ActionGroup(uint32(exportedSvcGroupID)).
+			Done()
+	}
 	return group
 }