Extends the Endpoints support from 500 to 800, extra ones will be dropped in AntreaProxy

docs/feature-gates.md

@@ -51,7 +51,11 @@ example, to enable `AntreaProxy` on Linux, edit the Agent configuration in the
 `AntreaProxy` implements Service load-balancing for ClusterIP Services as part
 of the OVS pipeline, as opposed to relying on kube-proxy. This only applies to
 traffic originating from Pods, and destined to ClusterIP Services. In
-particular, it does not apply to NodePort Services.
+particular, it does not apply to NodePort Services. Please note that due to
+some restrictions on the implementation of Services in Antrea, the maximum
+number of Endpoints that Antrea can support at the moment is 800. If the
+number of Endpoints for a given Service exceeds 800, extra Endpoints will
+be dropped.
 
 Note that this feature must be enabled for Windows. The Antrea Windows YAML
 manifest provided as part of releases enables this feature by default. If you

pkg/agent/openflow/client.go

@@ -434,6 +434,7 @@ func (c *client) GetPodFlowKeys(interfaceName string) []string {
 func (c *client) InstallServiceGroup(groupID binding.GroupIDType, withSessionAffinity bool, endpoints []proxy.Endpoint) error {
 	c.replayMutex.RLock()
 	defer c.replayMutex.RUnlock()
+
 	group := c.serviceEndpointGroup(groupID, withSessionAffinity, endpoints...)
 	if err := group.Add(); err != nil {
 		return fmt.Errorf("error when installing Service Endpoints Group: %w", err)
@@ -497,7 +498,7 @@ func (c *client) InstallServiceFlows(groupID binding.GroupIDType, svcIP net.IP,
 	c.replayMutex.RLock()
 	defer c.replayMutex.RUnlock()
 	var flows []binding.Flow
-	flows = append(flows, c.serviceLBFlow(groupID, svcIP, svcPort, protocol))
+	flows = append(flows, c.serviceLBFlow(groupID, svcIP, svcPort, protocol, affinityTimeout != 0))
 	if affinityTimeout != 0 {
 		flows = append(flows, c.serviceLearnFlow(groupID, svcIP, svcPort, protocol, affinityTimeout))
 	}

pkg/agent/openflow/pipeline.go

@@ -1903,12 +1903,21 @@ func (c *client) serviceLearnFlow(groupID binding.GroupIDType, svcIP net.IP, svc
 
 // serviceLBFlow generates the flow which uses the specific group to do Endpoint
 // selection.
-func (c *client) serviceLBFlow(groupID binding.GroupIDType, svcIP net.IP, svcPort uint16, protocol binding.Protocol) binding.Flow {
+func (c *client) serviceLBFlow(groupID binding.GroupIDType, svcIP net.IP, svcPort uint16, protocol binding.Protocol, withSessionAffinity bool) binding.Flow {
+	var lbResultMark uint32
+	if withSessionAffinity {
+		lbResultMark = marksRegServiceNeedLearn
+	} else {
+		lbResultMark = marksRegServiceSelected
+	}
+
 	return c.pipeline[serviceLBTable].BuildFlow(priorityNormal).
 		MatchProtocol(protocol).
 		MatchDstPort(svcPort, nil).
 		MatchDstIP(svcIP).
 		MatchRegRange(int(serviceLearnReg), marksRegServiceNeedLB, serviceLearnRegRange).
+		Action().LoadRegRange(int(serviceLearnReg), lbResultMark, serviceLearnRegRange).
+		Action().LoadRegRange(int(marksReg), macRewriteMark, macRewriteMarkRange).
 		Action().Group(groupID).
 		Cookie(c.cookieAllocator.Request(cookie.Service).Raw()).
 		Done()
@@ -1974,13 +1983,10 @@ func (c *client) hairpinSNATFlow(endpointIP net.IP) binding.Flow {
 func (c *client) serviceEndpointGroup(groupID binding.GroupIDType, withSessionAffinity bool, endpoints ...proxy.Endpoint) binding.Group {
 	group := c.bridge.CreateGroup(groupID).ResetBuckets()
 	var resubmitTableID binding.TableIDType
-	var lbResultMark uint32
 	if withSessionAffinity {
 		resubmitTableID = serviceLBTable
-		lbResultMark = marksRegServiceNeedLearn
 	} else {
 		resubmitTableID = endpointDNATTable
-		lbResultMark = marksRegServiceSelected
 	}
 
 	for _, endpoint := range endpoints {
@@ -1993,17 +1999,13 @@ func (c *client) serviceEndpointGroup(groupID binding.GroupIDType, withSessionAf
 			group = group.Bucket().Weight(100).
 				LoadReg(int(endpointIPReg), ipVal).
 				LoadRegRange(int(endpointPortReg), uint32(portVal), endpointPortRegRange).
-				LoadRegRange(int(serviceLearnReg), lbResultMark, serviceLearnRegRange).
-				LoadRegRange(int(marksReg), macRewriteMark, macRewriteMarkRange).
 				ResubmitToTable(resubmitTableID).
 				Done()
 		} else if ipProtocol == binding.ProtocolIPv6 {
 			ipVal := []byte(endpointIP)
 			group = group.Bucket().Weight(100).
 				LoadXXReg(int(endpointIPv6XXReg), ipVal).
 				LoadRegRange(int(endpointPortReg), uint32(portVal), endpointPortRegRange).
-				LoadRegRange(int(serviceLearnReg), lbResultMark, serviceLearnRegRange).
-				LoadRegRange(int(marksReg), macRewriteMark, macRewriteMarkRange).
 				ResubmitToTable(resubmitTableID).
 				Done()
 		}

pkg/agent/proxy/endpoints.go

@@ -108,7 +108,7 @@ func (t *endpointsChangesTracker) OnEndpointUpdate(previous, current *corev1.End
 	return len(t.changes) > 0
 }
 
-// EndpointSliceUpdate updates the given service's endpoints change map based on the <previous, current> endpoints pair.
+// OnEndpointSliceUpdate updates the given service's endpoints change map based on the <previous, current> endpoints pair.
 // It returns true if items changed, otherwise it returns false. Will add/update/delete items of endpointsChange Map.
 // If removeSlice is true, slice will be removed, otherwise it will be added or updated.
 func (t *endpointsChangesTracker) OnEndpointSliceUpdate(endpointSlice *discovery.EndpointSlice, removeSlice bool) bool {
@@ -220,3 +220,16 @@ func (t *endpointsChangesTracker) Update(em types.EndpointsMap) {
 		}
 	}
 }
+
+// byEndpoint helps sort Endpoint
+type byEndpoint []k8sproxy.Endpoint
+
+func (p byEndpoint) Len() int {
+	return len(p)
+}
+func (p byEndpoint) Swap(i, j int) {
+	p[i], p[j] = p[j], p[i]
+}
+func (p byEndpoint) Less(i, j int) bool {
+	return p[i].String() < p[j].String()
+}

pkg/agent/proxy/proxier.go

@@ -17,6 +17,7 @@ package proxy
 import (
 	"fmt"
 	"net"
+	"sort"
 	"strings"
 	"sync"
 	"time"
@@ -25,6 +26,7 @@ import (
 	"k8s.io/api/discovery/v1beta1"
 	"k8s.io/apimachinery/pkg/runtime"
 	k8sapitypes "k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/client-go/informers"
 	"k8s.io/client-go/tools/record"
 	"k8s.io/klog"
@@ -42,6 +44,10 @@ import (
 const (
 	resyncPeriod  = time.Minute
 	componentName = "antrea-agent-proxy"
+	// Due to the maximum message size in Openflow 1.3 and the implementation of Services in Antrea, the maximum number
+	// of Endpoints that Antrea can support at the moment is 800. If the number of Endpoints for a given Service exceeds
+	// 800, extra Endpoints will be dropped.
+	maxEndpoints = 800
 )
 
 // Proxier wraps proxy.Provider and adds extra methods. It is introduced for
@@ -90,6 +96,8 @@ type proxier struct {
 	serviceStringMap map[string]k8sproxy.ServicePortName
 	// serviceStringMapMutex protects serviceStringMap object.
 	serviceStringMapMutex sync.Mutex
+	// oversizeServiceSet records the Services that have more than 800 Endpoints.
+	oversizeServiceSet sets.String
 
 	runner              *k8sproxy.BoundedFrequencyRunner
 	stopChan            <-chan struct{}
@@ -117,6 +125,9 @@ func (p *proxier) removeStaleServices() {
 		}
 		svcInfo := svcPort.(*types.ServiceInfo)
 		klog.V(2).Infof("Removing stale Service: %s %s", svcPortName.Name, svcInfo.String())
+		if p.oversizeServiceSet.Has(svcPortName.String()) {
+			p.oversizeServiceSet.Delete(svcPortName.String())
+		}
 		if err := p.ofClient.UninstallServiceFlows(svcInfo.ClusterIP(), uint16(svcInfo.Port()), svcInfo.OFProtocol); err != nil {
 			klog.Errorf("Failed to remove flows of Service %v: %v", svcPortName, err)
 			continue
@@ -258,12 +269,40 @@ func (p *proxier) installServices() {
 		}
 
 		var endpointUpdateList []k8sproxy.Endpoint
-		for _, endpoint := range endpoints { // Check if there is any installed Endpoint which is not expected anymore.
-			if _, ok := endpointsInstalled[endpoint.String()]; !ok { // There is an expected Endpoint which is not installed.
-				needUpdateEndpoints = true
+		if len(endpoints) > maxEndpoints {
+			if !p.oversizeServiceSet.Has(svcPortName.String()) {
+				klog.Warningf("Since Endpoints of Service %s exceeds %d, extra Endpoints will be dropped", svcPortName.String(), maxEndpoints)
+				p.oversizeServiceSet.Insert(svcPortName.String())
+			}
+			// If the length of endpoints > maxEndpoints, endpoints should be cut. However, endpoints is a map. Therefore,
+			// iterate the map and append every Endpoint to a slice endpointList. Since the iteration order of map in
+			// Golang is random, if cut directly without any sorting, some Endpoints may not be installed. So cutting
+			// slice endpointList after sorting can avoid this situation in some degree.
+			var endpointList []k8sproxy.Endpoint
+			for _, endpoint := range endpoints {
+				endpointList = append(endpointList, endpoint)
+			}
+			sort.Sort(byEndpoint(endpointList))
+			endpointList = endpointList[:maxEndpoints]
+
+			for _, endpoint := range endpointList { // Check if there is any installed Endpoint which is not expected anymore.
+				if _, ok := endpointsInstalled[endpoint.String()]; !ok { // There is an expected Endpoint which is not installed.
+					needUpdateEndpoints = true
+				}
+				endpointUpdateList = append(endpointUpdateList, endpoint)
+			}
+		} else {
+			if p.oversizeServiceSet.Has(svcPortName.String()) {
+				p.oversizeServiceSet.Delete(svcPortName.String())
+			}
+			for _, endpoint := range endpoints { // Check if there is any installed Endpoint which is not expected anymore.
+				if _, ok := endpointsInstalled[endpoint.String()]; !ok { // There is an expected Endpoint which is not installed.
+					needUpdateEndpoints = true
+				}
+				endpointUpdateList = append(endpointUpdateList, endpoint)
 			}
-			endpointUpdateList = append(endpointUpdateList, endpoint)
 		}
+
 		if len(endpoints) < len(endpointsInstalled) { // There are Endpoints which expired.
 			klog.V(2).Infof("Some Endpoints of Service %s removed, updating Endpoints", svcInfo.String())
 			needUpdateEndpoints = true
@@ -605,6 +644,7 @@ func NewProxier(
 		endpointsMap:             types.EndpointsMap{},
 		endpointReferenceCounter: map[string]int{},
 		serviceStringMap:         map[string]k8sproxy.ServicePortName{},
+		oversizeServiceSet:       sets.NewString(),
 		groupCounter:             types.NewGroupCounter(),
 		ofClient:                 ofClient,
 		isIPv6:                   isIPv6,

test/e2e/proxy_test.go

@@ -281,9 +281,9 @@ func testProxyServiceLifeCycle(ipFamily *corev1.IPFamily, ingressIPs []string, d
 
 	var groupKeyword string
 	if *ipFamily == corev1.IPv6Protocol {
-		groupKeyword = fmt.Sprintf("set_field:0x%s->xxreg3,load:0x%x->NXM_NX_REG4[0..15],load:0x2->NXM_NX_REG4[16..18]", strings.TrimLeft(hex.EncodeToString(nginxIPs.ipv6.To16()), "0"), 80)
+		groupKeyword = fmt.Sprintf("set_field:0x%s->xxreg3,load:0x%x->NXM_NX_REG4[0..15]", strings.TrimLeft(hex.EncodeToString(nginxIPs.ipv6.To16()), "0"), 80)
 	} else {
-		groupKeyword = fmt.Sprintf("load:0x%s->NXM_NX_REG3[],load:0x%x->NXM_NX_REG4[0..15],load:0x2->NXM_NX_REG4[16..18]", strings.TrimLeft(hex.EncodeToString(nginxIPs.ipv4.To4()), "0"), 80)
+		groupKeyword = fmt.Sprintf("load:0x%s->NXM_NX_REG3[],load:0x%x->NXM_NX_REG4[0..15]", strings.TrimLeft(hex.EncodeToString(nginxIPs.ipv4.To4()), "0"), 80)
 	}
 	groupOutput, _, err := data.runCommandFromPod(metav1.NamespaceSystem, agentName, "antrea-agent", []string{"ovs-ofctl", "dump-groups", defaultBridgeName})
 	require.NoError(t, err)

test/integration/agent/openflow_test.go

@@ -600,11 +600,16 @@ func expectedProxyServiceGroupAndFlows(gid uint32, svc svcConfig, endpointList [
 		nw_proto = 132
 		learnProtoField = "OXM_OF_SCTP_DST[]"
 	}
+
+	serviceLearnReg := 2
+	if stickyAge != 0 {
+		serviceLearnReg = 3
+	}
 	cookieAllocator := cookie.NewAllocator(roundInfo.RoundNum)
 	svcFlows := expectTableFlows{tableID: 41, flows: []*ofTestUtils.ExpectFlow{
 		{
 			MatchStr: fmt.Sprintf("priority=200,%s,reg4=0x10000/0x70000,nw_dst=%s,tp_dst=%d", string(svc.protocol), svc.ip.String(), svc.port),
-			ActStr:   fmt.Sprintf("group:%d", gid),
+			ActStr:   fmt.Sprintf("load:0x%x->NXM_NX_REG4[16..18],load:0x1->NXM_NX_REG0[19],group:%d", serviceLearnReg, gid),
 		},
 		{
 			MatchStr: fmt.Sprintf("priority=190,%s,reg4=0x30000/0x70000,nw_dst=%s,tp_dst=%d", string(svc.protocol), svc.ip.String(), svc.port),
@@ -617,7 +622,7 @@ func expectedProxyServiceGroupAndFlows(gid uint32, svc svcConfig, endpointList [
 	for _, ep := range endpointList {
 		epIP := ipToHexString(net.ParseIP(ep.IP()))
 		epPort, _ := ep.Port()
-		bucket := fmt.Sprintf("weight:100,actions=load:%s->NXM_NX_REG3[],load:0x%x->NXM_NX_REG4[0..15],load:0x2->NXM_NX_REG4[16..18],load:0x1->NXM_NX_REG0[19],resubmit(,42)", epIP, epPort)
+		bucket := fmt.Sprintf("weight:100,actions=load:%s->NXM_NX_REG3[],load:0x%x->NXM_NX_REG4[0..15],resubmit(,42)", epIP, epPort)
 		groupBuckets = append(groupBuckets, bucket)
 
 		unionVal := (0b010 << 16) + uint32(epPort)