Change maximum number of buckets in an OVS group add/insert_bucket me…

…ssage

The current implementation limits the maximum number of buckets in an OVS group
add/insert_bucket message to 800. This constraint is based on the fact that each
bucket has 3 actions, such as `set_field:0xa0a0007->reg0`,
`set_field:0x50/0xffff->reg4`, and `resubmit(,EndpointDNAT)`. However, an update
in antrea-io#5205 introduced a new action, `set_field:0x4000000/0x4000000->reg4`, for
remote Endpoints, making it impossible to accommodate 800 buckets with 4 actions
in an OVS group add/insert_bucket message.

To overcome the limitation, we set the maximum number of buckets to 700, considering
the worst-case scenario where each bucket includes all available actions. For example,
a bucket with all available actions, which is for a remote non-hostNetwork IPv6 Service
Endpoint like this: `set_field:0xa0a0007->xxreg0`, `set_field:0x50/0xffff->reg4`,
`set_field:0x100000/0x100000->reg4`, and `resubmit(,EndpointDNAT)`. The size of such
bucket is 88 bytes, and the header size of an OVS group message is 24 bytes.
According to https://opennetworking.org/wp-content/uploads/2014/10/openflow-switch-v1.5.1.pdf,
the max size of an Openflow 1.5 message is 64000 bytes, as a result, a message can have
a maximum of 727 buckets with the largest size.

Signed-off-by: Hongliang Liu <[email protected]>

pkg/agent/openflow/multicast.go

@@ -97,6 +97,7 @@ func (f *featureMulticast) replayFlows() []*openflow15.FlowMod {
 	return getCachedFlowMessages(f.cachedFlows)
 }
 
+// IMPORTANT: Ensure any changes to this function are tested in TestMulticastReceiversGroupMaxBuckets.
 func (f *featureMulticast) multicastReceiversGroup(groupID binding.GroupIDType, tableID uint8, ports []uint32, remoteIPs []net.IP) binding.Group {
 	group := f.bridge.NewGroupTypeAll(groupID)
 	for i := range ports {

pkg/agent/openflow/multicast_test.go

@@ -15,11 +15,17 @@
 package openflow
 
 import (
+	"net"
 	"testing"
 
+	"antrea.io/libOpenflow/openflow15"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/mock/gomock"
 
 	"antrea.io/antrea/pkg/agent/config"
+	binding "antrea.io/antrea/pkg/ovs/openflow"
+	ovsoftest "antrea.io/antrea/pkg/ovs/openflow/testing"
 )
 
 func multicastInitFlows(isEncap bool) []string {
@@ -80,3 +86,61 @@ func Test_featureMulticast_initFlows(t *testing.T) {
 		})
 	}
 }
+
+// If any test case fails, please consider setting binding.MaxBucketsPerMessage to a smaller value.
+func TestMulticastReceiversGroupMaxBuckets(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	fs := &featureMulticast{
+		bridge: binding.NewOFBridge(bridgeName, ""),
+	}
+	fakeOfTable := ovsoftest.NewMockTable(ctrl)
+	MulticastOutputTable.ofTable = fakeOfTable
+	defer func() {
+		MulticastOutputTable.ofTable = nil
+	}()
+
+	testCases := []struct {
+		name         string
+		ports        []uint32
+		remoteIPs    []net.IP
+		expectedCall func(*ovsoftest.MockTable)
+	}{
+		{
+			name: "Only ports",
+			ports: func() []uint32 {
+				var ports []uint32
+				for i := 0; i < binding.MaxBucketsPerMessage; i++ {
+					ports = append(ports, uint32(i))
+				}
+				return ports
+			}(),
+			expectedCall: func(table *ovsoftest.MockTable) {},
+		},
+		{
+			name: "Only remote IPs",
+			remoteIPs: func() []net.IP {
+				var remoteIPs []net.IP
+				sampleIP := net.ParseIP("192.168.1.1")
+				for i := 0; i < binding.MaxBucketsPerMessage; i++ {
+					remoteIPs = append(remoteIPs, sampleIP)
+				}
+				return remoteIPs
+			}(),
+			expectedCall: func(table *ovsoftest.MockTable) {
+				table.EXPECT().GetID().Return(uint8(1)).Times(binding.MaxBucketsPerMessage)
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			tc.expectedCall(fakeOfTable)
+			group := fs.multicastReceiversGroup(binding.GroupIDType(100), 0, tc.ports, tc.remoteIPs)
+			messages, err := group.GetBundleMessages(binding.AddMessage)
+			require.NoError(t, err)
+			require.Equal(t, 1, len(messages))
+			groupMod := messages[0].GetMessage().(*openflow15.GroupMod)
+			require.LessOrEqual(t, getGroupModLen(groupMod), openflowMessageMaxSize, "The size of the GroupMod message exceeds the maximum allowed size")
+		})
+	}
+}

pkg/agent/openflow/pipeline.go

@@ -2662,6 +2662,7 @@ func (f *featureService) dsrServiceNoDNATFlows() []binding.Flow {
 // 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.
+// IMPORTANT: Ensure any changes to this function are tested in TestServiceEndpointGroupMaxBuckets.
 func (f *featureService) serviceEndpointGroup(groupID binding.GroupIDType, withSessionAffinity bool, endpoints ...proxy.Endpoint) binding.Group {
 	group := f.bridge.NewGroup(groupID)
 

pkg/agent/openflow/pipeline_test.go

@@ -17,13 +17,23 @@ package openflow
 import (
 	"testing"
 
+	"antrea.io/libOpenflow/openflow15"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 	"go.uber.org/mock/gomock"
 
 	"antrea.io/antrea/pkg/agent/config"
+	nodeiptest "antrea.io/antrea/pkg/agent/nodeip/testing"
 	oftest "antrea.io/antrea/pkg/agent/openflow/testing"
+	binding "antrea.io/antrea/pkg/ovs/openflow"
+	ovsoftest "antrea.io/antrea/pkg/ovs/openflow/testing"
+	"antrea.io/antrea/third_party/proxy"
 )
 
+// Referring to page 43 in https://opennetworking.org/wp-content/uploads/2014/10/openflow-switch-v1.5.1.pdf,
+// an OpenFlow message size can reach up to 64KB.
+const openflowMessageMaxSize uint32 = 64 * 1024
+
 func pipelineDefaultFlows(egressTrafficShapingEnabled, externalNodeEnabled, isEncap, isIPv4 bool) []string {
 	if externalNodeEnabled {
 		return []string{
@@ -228,3 +238,65 @@ func Test_client_defaultFlows(t *testing.T) {
 		})
 	}
 }
+
+// If any test case fails, please consider setting binding.MaxBucketsPerMessage to a smaller value.
+func TestServiceEndpointGroupMaxBuckets(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	fakeOfTable := ovsoftest.NewMockTable(ctrl)
+	ServiceLBTable.ofTable = fakeOfTable
+	defer func() {
+		ServiceLBTable.ofTable = nil
+	}()
+	fs := &featureService{
+		bridge:        binding.NewOFBridge(bridgeName, ""),
+		nodeIPChecker: nodeiptest.NewFakeNodeIPChecker(),
+	}
+
+	// Test the Endpoint associated with a bucket containing all available actions.
+	testCases := []struct {
+		name           string
+		sampleEndpoint proxy.Endpoint
+	}{
+		{
+			name:           "IPv6, remote, non-hostNetwork",
+			sampleEndpoint: proxy.NewBaseEndpointInfo("2001::1", "node1", "", 80, false, true, false, false, nil),
+		},
+		{
+			name:           "IPv4, remote, non-hostNetwork",
+			sampleEndpoint: proxy.NewBaseEndpointInfo("192.168.1.1", "node1", "", 80, false, true, false, false, nil),
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			var endpoints []proxy.Endpoint
+			for i := 0; i < binding.MaxBucketsPerMessage; i++ {
+				endpoints = append(endpoints, tc.sampleEndpoint)
+			}
+
+			fakeOfTable.EXPECT().GetID().Return(uint8(1)).Times(1)
+			group := fs.serviceEndpointGroup(binding.GroupIDType(100), true, endpoints...)
+			messages, err := group.GetBundleMessages(binding.AddMessage)
+			require.NoError(t, err)
+			require.Equal(t, 1, len(messages))
+			groupMod := messages[0].GetMessage().(*openflow15.GroupMod)
+			require.LessOrEqual(t, getGroupModLen(groupMod), openflowMessageMaxSize, "The size of the GroupMod message exceeds the maximum allowed size")
+		})
+	}
+}
+
+func getGroupModLen(g *openflow15.GroupMod) uint32 {
+	n := uint32(0)
+
+	n = uint32(g.Header.Len())
+	n += 16
+
+	for _, b := range g.Buckets {
+		n += uint32(b.Len())
+	}
+
+	for _, p := range g.Properties {
+		n += uint32(p.Len())
+	}
+	return n
+}

pkg/ovs/openflow/ofctrl_group.go

@@ -24,7 +24,7 @@ import (
 )
 
 var (
-	MaxBucketsPerMessage = 800
+	MaxBucketsPerMessage = 700
 )
 
 type ofGroup struct {