address comments

Signed-off-by: wenyingd <[email protected]>

pkg/agent/openflow/client.go

@@ -22,7 +22,6 @@ import (
 	"antrea.io/libOpenflow/openflow15"
 	"antrea.io/libOpenflow/protocol"
 	ofutil "antrea.io/libOpenflow/util"
-	"antrea.io/ofnet/ofctrl"
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/klog/v2"
 
@@ -395,7 +394,7 @@ func (c *client) addFlows(cache *flowCategoryCache, flowCacheKey string, flows [
 // It will skip flows whose cache already exists. All flows will be installed via a bundle.
 func (c *client) addFlowsWithMultipleKeys(cache *flowCategoryCache, keyToFlows map[string][]binding.Flow) error {
 	// allMessages keeps the OpenFlow modification messages we will install via a bundle.
-	var allMessages []ofctrl.OpenFlowModMessage
+	var allMessages []*openflow15.FlowMod
 	// flowCacheMap keeps the flowMessageCache items we will add to the cache on bundle success.
 	flowCacheMap := map[string]flowMessageCache{}
 	for flowCacheKey, flows := range keyToFlows {
@@ -407,9 +406,9 @@ func (c *client) addFlowsWithMultipleKeys(cache *flowCategoryCache, keyToFlows m
 		}
 		fCache := flowMessageCache{}
 		for _, flow := range flows {
-			cf := getCachedFlow(flow, binding.AddMessage)
-			allMessages = append(allMessages, cf.message)
-			fCache[flow.MatchString()] = cf
+			cf := getFlowModMessage(flow, binding.AddMessage)
+			allMessages = append(allMessages, cf)
+			fCache[getFlowKey(cf)] = cf
 		}
 		flowCacheMap[flowCacheKey] = fCache
 	}
@@ -433,30 +432,30 @@ func (c *client) modifyFlows(cache *flowCategoryCache, flowCacheKey string, flow
 	fCache := flowMessageCache{}
 	var err error
 	if !ok {
-		messages := make([]ofctrl.OpenFlowModMessage, 0)
+		messages := make([]*openflow15.FlowMod, 0)
 		for _, flow := range flows {
-			cf := getCachedFlow(flow, binding.AddMessage)
-			messages = append(messages, cf.message)
-			fCache[flow.MatchString()] = cf
+			cf := getFlowModMessage(flow, binding.AddMessage)
+			messages = append(messages, cf)
+			fCache[getFlowKey(cf)] = cf
 		}
 		err = c.ofEntryOperations.AddAll(messages)
 	} else {
-		var adds, mods, dels []ofctrl.OpenFlowModMessage
+		var adds, mods, dels []*openflow15.FlowMod
 		oldFlowCache := oldFlowCacheI.(flowMessageCache)
 		for _, flow := range flows {
 			matchString := flow.MatchString()
-			var cachedFlowInfo *cachedFlow
+			var cachedFlowInfo *openflow15.FlowMod
 			if _, ok := oldFlowCache[matchString]; ok {
-				cachedFlowInfo = getCachedFlow(flow, binding.ModifyMessage)
+				cachedFlowInfo = getFlowModMessage(flow, binding.ModifyMessage)
 			} else {
-				cachedFlowInfo = getCachedFlow(flow, binding.AddMessage)
+				cachedFlowInfo = getFlowModMessage(flow, binding.AddMessage)
 			}
-			adds = append(adds, cachedFlowInfo.message)
+			adds = append(adds, cachedFlowInfo)
 			fCache[matchString] = cachedFlowInfo
 		}
 		for k, v := range oldFlowCache {
 			if _, ok := fCache[k]; !ok {
-				dels = append(dels, v.message)
+				dels = append(dels, v)
 			}
 		}
 		err = c.ofEntryOperations.BundleOps(adds, mods, dels)
@@ -479,7 +478,7 @@ func (c *client) deleteFlows(cache *flowCategoryCache, flowCacheKey string) erro
 // It will skip the keys which are not in the cache. All flows will be uninstalled via a bundle.
 func (c *client) deleteFlowsWithMultipleKeys(cache *flowCategoryCache, keys []string) error {
 	// allFlows keeps the flows we will delete via a bundle.
-	var allFlows []ofctrl.OpenFlowModMessage
+	var allFlows []*openflow15.FlowMod
 	for _, key := range keys {
 		flows, ok := cache.Load(key)
 		// If a flow cache entry of the key does not exist, skip it.
@@ -488,7 +487,7 @@ func (c *client) deleteFlowsWithMultipleKeys(cache *flowCategoryCache, keys []st
 			continue
 		}
 		for _, flow := range flows.(flowMessageCache) {
-			allFlows = append(allFlows, flow.message)
+			allFlows = append(allFlows, flow)
 		}
 	}
 	if len(allFlows) == 0 {
@@ -658,7 +657,7 @@ func (c *client) getFlowKeysFromCache(cache *flowCategoryCache, cacheKey string)
 	c.replayMutex.RLock()
 	defer c.replayMutex.RUnlock()
 	for _, flow := range fCache {
-		flowKeys = append(flowKeys, flow.key)
+		flowKeys = append(flowKeys, getFlowKey(flow))
 	}
 	return flowKeys
 }
@@ -1532,18 +1531,17 @@ func (c *client) UninstallMulticlusterPodFlows(podIP string) error {
 	return c.deleteFlows(c.featureMulticluster.cachedPodFlows, podIP)
 }
 
-func getFlowMessages(flows []binding.Flow) []ofctrl.OpenFlowModMessage {
-	messages := make([]ofctrl.OpenFlowModMessage, 0, len(flows))
+func getFlowMessages(flows []binding.Flow) []*openflow15.FlowMod {
+	messages := make([]*openflow15.FlowMod, 0, len(flows))
 	for i := range flows {
-		cf := getCachedFlow(flows[i], binding.AddMessage)
-		messages = append(messages, cf.message)
+		msg := getFlowModMessage(flows[i], binding.AddMessage)
+		messages = append(messages, msg)
 	}
 	return messages
 }
 
-func getCachedFlow(flow binding.Flow, entryOper binding.OFOperation) *cachedFlow {
+func getFlowModMessage(flow binding.Flow, entryOper binding.OFOperation) *openflow15.FlowMod {
 	bundleMessages, _ := flow.GetBundleMessages(entryOper)
-	flowKey := flow.MatchString()
 	flowMessage := bundleMessages[0]
-	return &cachedFlow{key: flowKey, table: flow.GetTable(), isDropFlow: flow.IsDropFlow(), message: flowMessage}
+	return flowMessage.GetMessage().(*openflow15.FlowMod)
 }

pkg/agent/openflow/client_test.go

@@ -458,7 +458,7 @@ func newFakeClient(mockOFEntryOperations *oftest.MockOFEntryOperations,
 	client.ipProtocols = ipProtocols
 	client.generatePipelines()
 	client.realizePipelines()
-	ovsoftest.TableNameCache = getTableNameCache()
+	binding.TableNameCache = getTableNameCache()
 	client.bridge.(*binding.OFBridge).SetOFSwitch(ofctrl.NewSwitch(&util.MessageStream{}, GlobalVirtualMAC, nil, make(chan int), 32776))
 	client.bridge.(*binding.OFBridge).Initialize()
 	return client
@@ -476,27 +476,26 @@ func getTableNameCache() map[uint8]string {
 }
 
 func getFlowStrings(flows interface{}) []string {
-	getStrings := func(flowMessage ofctrl.OpenFlowModMessage) []string {
+	getStrings := func(message *openflow15.FlowMod) []string {
 		var strs []string
-		message, _ := flowMessage.GetMessage().(*openflow15.FlowMod)
-		f := ovsoftest.FlowModToString(message)
+		f := binding.FlowModToString(message)
 		strs = append(strs, f)
 		return strs
 	}
 	var flowStrings []string
 	switch v := flows.(type) {
 	case flowMessageCache:
 		for _, flow := range v {
-			flowStrings = append(flowStrings, getStrings(flow.message)...)
+			flowStrings = append(flowStrings, getStrings(flow)...)
 		}
 	case []binding.Flow:
 		for _, flow := range v {
 			messages, _ := flow.GetBundleMessages(binding.AddMessage)
 			for _, msg := range messages {
-				flowStrings = append(flowStrings, getStrings(msg)...)
+				flowStrings = append(flowStrings, getStrings(msg.GetMessage().(*openflow15.FlowMod))...)
 			}
 		}
-	case []ofctrl.OpenFlowModMessage:
+	case []*openflow15.FlowMod:
 		for _, flow := range v {
 			flowStrings = append(flowStrings, getStrings(flow)...)
 		}
@@ -505,10 +504,10 @@ func getFlowStrings(flows interface{}) []string {
 }
 
 func getGroupFromCache(groupCache binding.Group) string {
-	ovsoftest.TableNameCache = getTableNameCache()
+	binding.TableNameCache = getTableNameCache()
 
 	messages, _ := groupCache.GetBundleMessages(binding.AddMessage)
-	groupString := ovsoftest.GroupModToString(messages[0].GetMessage().(*openflow15.GroupMod))
+	groupString := binding.GroupModToString(messages[0].GetMessage().(*openflow15.GroupMod))
 	return groupString
 }
 
@@ -838,11 +837,11 @@ func Test_client_GetPodFlowKeys(t *testing.T) {
 	assert.NoError(t, fc.InstallPodFlows(interfaceName, podInterfaceIPs, podMAC, uint32(11), 0, nil))
 	flowKeys := fc.GetPodFlowKeys(interfaceName)
 	expectedFlowKeys := []string{
-		"table=1,arp,in_port=11,arp_sha=00:00:10:10:00:11,arp_spa=10.10.0.11",
-		"table=3,in_port=11",
-		"table=4,ip,in_port=11,dl_src=00:00:10:10:00:11,nw_src=10.10.0.11",
-		"table=17,ip,reg0=0x200/0x200,nw_dst=10.10.0.11",
-		"table=22,dl_dst=00:00:10:10:00:11",
+		"table=1,priority=200,arp,in_port=11,arp_spa=10.10.0.11,arp_sha=00:00:10:10:00:11",
+		"table=3,priority=190,in_port=11",
+		"table=4,priority=200,ip,in_port=11,dl_src=00:00:10:10:00:11,nw_src=10.10.0.11",
+		"table=17,priority=200,ip,reg0=0x200/0x200,nw_dst=10.10.0.11",
+		"table=22,priority=200,dl_dst=00:00:10:10:00:11",
 	}
 	assert.ElementsMatch(t, expectedFlowKeys, flowKeys)
 }
@@ -1225,11 +1224,11 @@ func Test_client_GetServiceFlowKeys(t *testing.T) {
 	assert.NoError(t, fc.InstallEndpointFlows(bindingProtocol, endpoints))
 	flowKeys := fc.GetServiceFlowKeys(svcIP, svcPort, bindingProtocol, endpoints)
 	expectedFlowKeys := []string{
-		"table=11,tcp,tp_dst=0x50,nw_dst=10.96.0.224,reg4=0x10000/0x70000",
-		"table=11,tcp,reg4=0x30000/0x70000,nw_dst=10.96.0.224,tp_dst=0x50",
-		"table=12,tcp,reg4=0x20050/0x7ffff,reg3=0xa0a000b",
-		"table=12,tcp,reg4=0x20050/0x7ffff,reg3=0xa0a000c",
-		"table=20,ip,nw_src=10.10.0.12,nw_dst=10.10.0.12,ct_state=+new+trk",
+		"table=11,priority=200,tcp,reg4=0x10000/0x70000,nw_dst=10.96.0.224,tp_dst=80",
+		"table=11,priority=190,tcp,reg4=0x30000/0x70000,nw_dst=10.96.0.224,tp_dst=80",
+		"table=12,priority=200,tcp,reg3=0xa0a000b,reg4=0x20050/0x7ffff",
+		"table=12,priority=200,tcp,reg3=0xa0a000c,reg4=0x20050/0x7ffff",
+		"table=20,priority=190,ct_state=+new+trk,ip,nw_src=10.10.0.12,nw_dst=10.10.0.12",
 	}
 	assert.ElementsMatch(t, expectedFlowKeys, flowKeys)
 }
@@ -1481,10 +1480,10 @@ func prepareTraceflowFlow(ctrl *gomock.Controller) *client {
 
 	m.EXPECT().AddAll(gomock.Any()).Return(nil).Times(1)
 	_, ipCIDR, _ := net.ParseCIDR("192.168.2.30/32")
-	flowMsg := &ofctrl.FlowBundleMessage{}
-	mFlow := &cachedFlow{isDropFlow: true, message: flowMsg, table: EgressDefaultTable.ofTable}
+	flows, _ := EgressDefaultTable.ofTable.BuildFlow(priority100).Action().Drop().Done().GetBundleMessages(binding.AddMessage)
+	flowMsg := flows[0].GetMessage().(*openflow15.FlowMod)
 	ctx := &conjMatchFlowContext{
-		dropFlow:              mFlow,
+		dropFlow:              flowMsg,
 		dropFlowEnableLogging: false,
 		conjunctiveMatch: &conjunctiveMatch{
 			tableID: 1,
@@ -1496,8 +1495,7 @@ func prepareTraceflowFlow(ctrl *gomock.Controller) *client {
 			},
 		}}
 	fc.featureNetworkPolicy.globalConjMatchFlowCache["mockContext"] = ctx
-	EgressMetricTable.ofTable.SetTable()
-	fc.featureNetworkPolicy.policyCache.Add(&policyRuleConjunction{metricFlows: []*cachedFlow{{isDropFlow: true, message: flowMsg, table: EgressMetricTable.ofTable}}})
+	fc.featureNetworkPolicy.policyCache.Add(&policyRuleConjunction{metricFlows: []*openflow15.FlowMod{flowMsg}})
 	return fc
 }
 
@@ -2266,7 +2264,7 @@ func Test_client_ReplayFlows(t *testing.T) {
 	addFlowInCache := func(cache *flowCategoryCache, cacheKey string, flows []binding.Flow) {
 		fCache := flowMessageCache{}
 		for _, flow := range flows {
-			cf := getCachedFlow(flow, binding.AddMessage)
+			cf := getFlowModMessage(flow, binding.AddMessage)
 			fCache[flow.MatchString()] = cf
 		}
 		cache.Store(cacheKey, fCache)
@@ -2302,14 +2300,14 @@ func Test_client_ReplayFlows(t *testing.T) {
 	priority200 = uint16(200)
 	conj := &policyRuleConjunction{
 		id:          ruleID,
-		actionFlows: []*cachedFlow{getCachedFlow(fc.featureNetworkPolicy.conjunctionActionDenyFlow(ruleID, IngressRuleTable.ofTable, &priority200, DispositionDrop, false), binding.AddMessage)},
-		metricFlows: []*cachedFlow{getCachedFlow(fc.featureNetworkPolicy.denyRuleMetricFlow(ruleID, true, IngressMetricTable.GetID()), binding.AddMessage)},
+		actionFlows: []*openflow15.FlowMod{getFlowModMessage(fc.featureNetworkPolicy.conjunctionActionDenyFlow(ruleID, IngressRuleTable.ofTable, &priority200, DispositionDrop, false), binding.AddMessage)},
+		metricFlows: []*openflow15.FlowMod{getFlowModMessage(fc.featureNetworkPolicy.denyRuleMetricFlow(ruleID, true, IngressMetricTable.GetID()), binding.AddMessage)},
 	}
 	assert.NoError(t, fc.featureNetworkPolicy.policyCache.Add(conj))
 	mp := matchPair{matchKey: MatchDstOFPort, matchValue: int32(podOfPort)}
 	context := &conjMatchFlowContext{
-		flow:     getCachedFlow(fc.featureNetworkPolicy.conjunctiveMatchFlow(IngressRuleTable.GetID(), []matchPair{mp}, &priority200, []*conjunctiveAction{{conjID: ruleID, clauseID: 2, nClause: 2}}), binding.AddMessage),
-		dropFlow: getCachedFlow(fc.featureNetworkPolicy.defaultDropFlow(IngressDefaultTable.ofTable, []matchPair{mp}, true), binding.AddMessage),
+		flow:     getFlowModMessage(fc.featureNetworkPolicy.conjunctiveMatchFlow(IngressRuleTable.GetID(), []matchPair{mp}, &priority200, []*conjunctiveAction{{conjID: ruleID, clauseID: 2, nClause: 2}}), binding.AddMessage),
+		dropFlow: getFlowModMessage(fc.featureNetworkPolicy.defaultDropFlow(IngressDefaultTable.ofTable, []matchPair{mp}, true), binding.AddMessage),
 	}
 	fc.featureNetworkPolicy.globalConjMatchFlowCache["npMatch"] = context
 	replayedFlows = append(replayedFlows,
@@ -2346,11 +2344,43 @@ func Test_client_ReplayFlows(t *testing.T) {
 	expectedFlows = append(expectedFlows, replayedFlows...)
 
 	actualFlows := make([]string, 0)
-	m.EXPECT().AddAll(gomock.Any()).Do(func(flowMessages []ofctrl.OpenFlowModMessage) {
+	m.EXPECT().AddAll(gomock.Any()).Do(func(flowMessages []*openflow15.FlowMod) {
 		flowStrings := getFlowStrings(flowMessages)
 		actualFlows = append(actualFlows, flowStrings...)
 	}).Return(nil).AnyTimes()
 
 	fc.ReplayFlows()
 	assert.ElementsMatch(t, expectedFlows, actualFlows)
 }
+
+func TestCachedFlowIsDrop(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+	m := oftest.NewMockOFEntryOperations(ctrl)
+	fc := newFakeClient(m, true, false, config.K8sNode, config.TrafficEncapModeEncap)
+	defer resetPipelines()
+	fc.bridge = ovsoftest.NewMockBridge(ctrl)
+
+	_, ipCIDR, _ := net.ParseCIDR("192.168.2.30/32")
+	flows, err := EgressDefaultTable.ofTable.
+		BuildFlow(priority100).
+		MatchDstIPNet(*ipCIDR).
+		Action().Drop().
+		Done().
+		GetBundleMessages(binding.AddMessage)
+	assert.NoError(t, err)
+	require.Equal(t, 1, len(flows))
+	msg := flows[0].GetMessage().(*openflow15.FlowMod)
+	assert.True(t, isDropFlow(msg))
+
+	flows, err = EgressDefaultTable.ofTable.
+		BuildFlow(priority100).
+		MatchDstIPNet(*ipCIDR).
+		Action().GotoTable(1).
+		Done().
+		GetBundleMessages(binding.AddMessage)
+	assert.NoError(t, err)
+	require.Equal(t, 1, len(flows))
+	msg = flows[0].GetMessage().(*openflow15.FlowMod)
+	assert.False(t, isDropFlow(msg))
+}

pkg/agent/openflow/egress.go

@@ -17,7 +17,7 @@ package openflow
 import (
 	"net"
 
-	"antrea.io/ofnet/ofctrl"
+	"antrea.io/libOpenflow/openflow15"
 
 	"antrea.io/antrea/pkg/agent/config"
 	"antrea.io/antrea/pkg/agent/openflow/cookie"
@@ -73,12 +73,12 @@ func newFeatureEgress(cookieAllocator cookie.Allocator,
 	}
 }
 
-func (f *featureEgress) initFlows() []ofctrl.OpenFlowModMessage {
+func (f *featureEgress) initFlows() []*openflow15.FlowMod {
 	// This installs the flows to enable Pods to communicate to the external IP addresses. The flows identify the packets
 	// from local Pods to the external IP address, and mark the packets to be SNAT'd with the configured SNAT IPs.
 	return getFlowMessages(f.externalFlows())
 }
 
-func (f *featureEgress) replayFlows() []ofctrl.OpenFlowModMessage {
+func (f *featureEgress) replayFlows() []*openflow15.FlowMod {
 	return getCachedFlowMessages(f.cachedFlows)
 }

pkg/agent/openflow/externalnode_connectivity.go

@@ -17,7 +17,7 @@ package openflow
 import (
 	"net"
 
-	"antrea.io/ofnet/ofctrl"
+	"antrea.io/libOpenflow/openflow15"
 
 	"antrea.io/antrea/pkg/agent/openflow/cookie"
 	binding "antrea.io/antrea/pkg/ovs/openflow"
@@ -101,7 +101,7 @@ func (f *featureExternalNodeConnectivity) vmUplinkFlows(hostOFPort, uplinkOFPort
 	}
 }
 
-func (f *featureExternalNodeConnectivity) initFlows() []ofctrl.OpenFlowModMessage {
+func (f *featureExternalNodeConnectivity) initFlows() []*openflow15.FlowMod {
 	cookieID := f.cookieAllocator.Request(f.category).Raw()
 	flows := []binding.Flow{
 		L2ForwardingOutTable.ofTable.BuildFlow(priorityNormal).
@@ -140,12 +140,12 @@ func (f *featureExternalNodeConnectivity) initFlows() []ofctrl.OpenFlowModMessag
 	return getFlowMessages(flows)
 }
 
-func (f *featureExternalNodeConnectivity) replayFlows() []ofctrl.OpenFlowModMessage {
-	var flows []ofctrl.OpenFlowModMessage
+func (f *featureExternalNodeConnectivity) replayFlows() []*openflow15.FlowMod {
+	var flows []*openflow15.FlowMod
 	rangeFunc := func(key, value interface{}) bool {
-		cachedFlows := value.([]cachedFlow)
+		cachedFlows := value.([]*openflow15.FlowMod)
 		for _, flow := range cachedFlows {
-			flows = append(flows, flow.message)
+			flows = append(flows, flow)
 		}
 		return true
 	}

pkg/agent/openflow/framework.go

@@ -15,7 +15,7 @@
 package openflow
 
 import (
-	"antrea.io/ofnet/ofctrl"
+	"antrea.io/libOpenflow/openflow15"
 
 	"antrea.io/antrea/pkg/agent/config"
 	binding "antrea.io/antrea/pkg/ovs/openflow"
@@ -58,9 +58,9 @@ type feature interface {
 	// getRequiredTables returns a slice of required tables of the feature.
 	getRequiredTables() []*Table
 	// initFlows returns the Openflow messages of initial flows of the feature.
-	initFlows() []ofctrl.OpenFlowModMessage
+	initFlows() []*openflow15.FlowMod
 	// replayFlows returns the Openflow messages of fixed and cached flows that need to be replayed after OVS is reconnected.
-	replayFlows() []ofctrl.OpenFlowModMessage
+	replayFlows() []*openflow15.FlowMod
 }
 
 const (