Update maximum number of buckets in OVS group add/insert_bucket message

go.mod

@@ -3,7 +3,7 @@ module antrea.io/antrea
 go 1.21.0
 
 require (
-	antrea.io/libOpenflow v0.13.0
+	antrea.io/libOpenflow v0.14.0
 	antrea.io/ofnet v0.12.0
 	github.com/ClickHouse/clickhouse-go/v2 v2.6.1
 	github.com/DATA-DOG/go-sqlmock v1.5.2

go.sum

@@ -1,5 +1,5 @@
-antrea.io/libOpenflow v0.13.0 h1:CelrRAMUk2wZyLI7KrYy2+cbFsJrWVIQp3xlP/pnNu4=
-antrea.io/libOpenflow v0.13.0/go.mod h1:U8YR0ithWrjwLdUUhyeCsYnlKg/mEFjG5CbPNt1V+j0=
+antrea.io/libOpenflow v0.14.0 h1:6MS1E52nGQyz/AJ8j3OrotgAc/1ubef5vc5i8ytIxFE=
+antrea.io/libOpenflow v0.14.0/go.mod h1:U8YR0ithWrjwLdUUhyeCsYnlKg/mEFjG5CbPNt1V+j0=
 antrea.io/ofnet v0.12.0 h1:pgygAsEZJUPK/kGmeuIesDh5hoGLpYeavSLdG/Dp8Ao=
 antrea.io/ofnet v0.12.0/go.mod h1:MB3qaInEimj+T8qtpBcIQK+EqeNiY1S/WbUdGk+TzNg=
 cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=

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,18 @@
 package openflow
 
 import (
+	"fmt"
+	"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"
+	openflowtest "antrea.io/antrea/pkg/ovs/openflow/testing"
 )
 
 func multicastInitFlows(isEncap bool) []string {
@@ -80,3 +87,63 @@ 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) {
+	fm := &featureMulticast{
+		bridge: binding.NewOFBridge(bridgeName, ""),
+	}
+
+	testCases := []struct {
+		name         string
+		ports        []uint32
+		remoteIPs    []net.IP
+		expectedCall func(*openflowtest.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 *openflowtest.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 *openflowtest.MockTable) {
+				table.EXPECT().GetID().Return(uint8(1)).Times(binding.MaxBucketsPerMessage)
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			ctrl := gomock.NewController(t)
+			fakeOfTable := openflowtest.NewMockTable(ctrl)
+			MulticastOutputTable.ofTable = fakeOfTable
+			defer func() {
+				MulticastOutputTable.ofTable = nil
+			}()
+
+			tc.expectedCall(fakeOfTable)
+			group := fm.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)
+			errorMsg := fmt.Sprintf("The GroupMod size with %d buckets exceeds the OpenFlow message's maximum allowable size, please consider setting binding.MaxBucketsPerMessage to a smaller value.", binding.MaxBucketsPerMessage)
+			require.LessOrEqual(t, getGroupModLen(groupMod), uint32(openflow15.MSG_MAX_LEN), errorMsg)
+		})
+	}
+}

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

@@ -15,13 +15,20 @@
 package openflow
 
 import (
+	"fmt"
 	"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"
+	openflowtest "antrea.io/antrea/pkg/ovs/openflow/testing"
+	"antrea.io/antrea/third_party/proxy"
 )
 
 func pipelineDefaultFlows(egressTrafficShapingEnabled, externalNodeEnabled, isEncap, isIPv4 bool) []string {
@@ -228,3 +235,69 @@ 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) {
+	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) {
+			ctrl := gomock.NewController(t)
+			fakeOfTable := openflowtest.NewMockTable(ctrl)
+			ServiceLBTable.ofTable = fakeOfTable
+			defer func() {
+				ServiceLBTable.ofTable = nil
+			}()
+
+			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)
+			errorMsg := fmt.Sprintf("The GroupMod size with %d buckets exceeds the OpenFlow message's maximum allowable size, please consider setting binding.MaxBucketsPerMessage to a smaller value.", binding.MaxBucketsPerMessage)
+			require.LessOrEqual(t, getGroupModLen(groupMod), uint32(openflow15.MSG_MAX_LEN), errorMsg)
+		})
+	}
+}
+
+// For openflow15.GroupMod, it provides a built-in method for calculating the message length. However,considering that
+// the GroupMod size we test might exceed the maximum uint16 value, we use uint32 as the return value type.
+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 {