Repository: antrea-io/antrea
Pull request #1802 (2021-01-30 00:03:10)

Fortify the bandwidth e2e test of flow aggregator
Currently, bandwith test is disabled in flow aggregator e2e test because it's flaky especially for intra-node flows. We found the root cause of it is the first and final records exported by flow exporter don't have desired `octectDeltaCount` because iperf traffic may not fill the entire export interval of the first and final records.

So in this commit, we check bandwidth using `octetDeltaCountFromSourceNode` for the records in the middle and calculate the average bandwidth using the `octetTotalCountFromSourceNode`, then compares with the iperf bandwidth. The difference should be lower than 15%.

Also I changed the iperf time to 12s and flow aggregator active timeout to 3.5s to make the tests more stable.

Sample output of updated bandwidth test:
```
=== RUN   TestFlowAggregator/IPv4/IntraNodeFlows
    flowaggregator_test.go:425: Network Policies are realized.
    flowaggregator_test.go:313: Check the bandwidth using octetDeltaCountFromSourceNode in data record.
    flowaggregator_test.go:331: Iperf throughput: 27545.60 Mbits/s, IPFIX record throughput calculated through octetDeltaCountFromSourceNode: 23907.06 Mbits/s
    flowaggregator_test.go:321: Check the average bandwidth using octetTotalCountFromSourceNode in data record.
    flowaggregator_test.go:331: Iperf throughput: 27545.60 Mbits/s, IPFIX record throughput calculated through octetTotalCountFromSourceNode: 25696.05 Mbits/s
=== RUN   TestFlowAggregator/IPv4/InterNodeFlows
    flowaggregator_test.go:425: Network Policies are realized.
    flowaggregator_test.go:313: Check the bandwidth using octetDeltaCountFromSourceNode in data record.
    flowaggregator_test.go:331: Iperf throughput: 917.00 Mbits/s, IPFIX record throughput calculated through octetDeltaCountFromSourceNode: 875.63 Mbits/s
    flowaggregator_test.go:321: Check the average bandwidth using octetTotalCountFromSourceNode in data record.
    flowaggregator_test.go:331: Iperf throughput: 917.00 Mbits/s, IPFIX record throughput calculated through octetTotalCountFromSourceNode: 873.56 Mbits/s
=== RUN   TestFlowAggregator/IPv4/LocalServiceAccess
    flowaggregator_test.go:313: Check the bandwidth using octetDeltaCountFromSourceNode in data record.
    flowaggregator_test.go:331: Iperf throughput: 29798.40 Mbits/s, IPFIX record throughput calculated through octetDeltaCountFromSourceNode: 28904.05 Mbits/s
    flowaggregator_test.go:321: Check the average bandwidth using octetTotalCountFromSourceNode in data record.
    flowaggregator_test.go:331: Iperf throughput: 29798.40 Mbits/s, IPFIX record throughput calculated through octetTotalCountFromSourceNode: 27773.72 Mbits/s
=== RUN   TestFlowAggregator/IPv4/RemoteServiceAccess
    flowaggregator_test.go:313: Check the bandwidth using octetDeltaCountFromSourceNode in data record.
    flowaggregator_test.go:331: Iperf throughput: 903.00 Mbits/s, IPFIX record throughput calculated through octetDeltaCountFromSourceNode: 880.78 Mbits/s
    flowaggregator_test.go:321: Check the average bandwidth using octetTotalCountFromSourceNode in data record.
    flowaggregator_test.go:331: Iperf throughput: 903.00 Mbits/s, IPFIX record throughput calculated through octetTotalCountFromSourceNode: 859.51 Mbits/s
```

dreamtalen: > As network policy issue in userspace datapath is being fixed. Please reflect that in title/PR description.

Thanks, added.
srikartati: Could you rebase and change the title of PR as there are changes in this PR?
dreamtalen: > Could you rebase and change the title of PR as there are changes in this PR?

sure, I'm debugging on a unstable bandwidth testing problem after rebased the main, will push the change after then.
dreamtalen: /test-all
codecov-commenter: # [Codecov](https://codecov.io/gh/antrea-io/antrea/pull/1802?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) Report
> Merging [#1802](https://codecov.io/gh/antrea-io/antrea/pull/1802?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) (58a8a19) into [main](https://codecov.io/gh/antrea-io/antrea/commit/c21592c0e718539f6a190843eb9fff9af8d3d175?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) (c21592c) will **decrease** coverage by `0.29%`.
> The diff coverage is `n/a`.

[![Impacted file tree graph](https://codecov.io/gh/antrea-io/antrea/pull/1802/graphs/tree.svg?width=650&height=150&src=pr&token=TjWaLL2GoC&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None)](https://codecov.io/gh/antrea-io/antrea/pull/1802?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None)

```diff
@@            Coverage Diff             @@
##             main    #1802      +/-   ##
==========================================
- Coverage   61.63%   61.33%   -0.30%     
==========================================
  Files         274      274              
  Lines       20673    20673              
==========================================
- Hits        12742    12680      -62     
- Misses       6593     6660      +67     
+ Partials     1338     1333       -5     
```

| Flag | Coverage Δ | |
|---|---|---|
| e2e-tests | `∅ <ø> (?)` | |
| kind-e2e-tests | `52.09% <ø> (-0.51%)` | :arrow_down: |
| unit-tests | `41.10% <ø> (ø)` | |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#carryforward-flags-in-the-pull-request-comment) to find out more.

| [Impacted Files](https://codecov.io/gh/antrea-io/antrea/pull/1802?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) | Coverage Δ | |
|---|---|---|
| [pkg/agent/controller/traceflow/packetin.go](https://codecov.io/gh/antrea-io/antrea/pull/1802/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2FnZW50L2NvbnRyb2xsZXIvdHJhY2VmbG93L3BhY2tldGluLmdv) | `56.94% <0.00%> (-9.73%)` | :arrow_down: |
| [...agent/controller/traceflow/traceflow\_controller.go](https://codecov.io/gh/antrea-io/antrea/pull/1802/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2FnZW50L2NvbnRyb2xsZXIvdHJhY2VmbG93L3RyYWNlZmxvd19jb250cm9sbGVyLmdv) | `69.61% <0.00%> (-3.54%)` | :arrow_down: |
| [pkg/controller/traceflow/controller.go](https://codecov.io/gh/antrea-io/antrea/pull/1802/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2NvbnRyb2xsZXIvdHJhY2VmbG93L2NvbnRyb2xsZXIuZ28=) | `69.27% <0.00%> (-2.41%)` | :arrow_down: |
| [pkg/agent/openflow/pipeline.go](https://codecov.io/gh/antrea-io/antrea/pull/1802/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2FnZW50L29wZW5mbG93L3BpcGVsaW5lLmdv) | `69.42% <0.00%> (-1.73%)` | :arrow_down: |
| [pkg/controller/networkpolicy/status\_controller.go](https://codecov.io/gh/antrea-io/antrea/pull/1802/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2NvbnRyb2xsZXIvbmV0d29ya3BvbGljeS9zdGF0dXNfY29udHJvbGxlci5nbw==) | `85.16% <0.00%> (-1.30%)` | :arrow_down: |
| [pkg/agent/proxy/proxier.go](https://codecov.io/gh/antrea-io/antrea/pull/1802/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2FnZW50L3Byb3h5L3Byb3hpZXIuZ28=) | `62.97% <0.00%> (-0.64%)` | :arrow_down: |
| [pkg/agent/route/route\_linux.go](https://codecov.io/gh/antrea-io/antrea/pull/1802/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2FnZW50L3JvdXRlL3JvdXRlX2xpbnV4Lmdv) | `43.08% <0.00%> (-0.55%)` | :arrow_down: |
| [pkg/agent/openflow/client.go](https://codecov.io/gh/antrea-io/antrea/pull/1802/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2FnZW50L29wZW5mbG93L2NsaWVudC5nbw==) | `58.99% <0.00%> (-0.27%)` | :arrow_down: |

dreamtalen: /test-all
dreamtalen: /test-networkpolicy
/test-windows-conformance
/test-e2e
dreamtalen: /test-all
dreamtalen: /test-windows-e2e
/test-e2e
dreamtalen: /test-all
dreamtalen: /test-networkpolicy
/test-e2e
dreamtalen: /test-all
dreamtalen: /test-all
dreamtalen: /test-networkpolicy
dreamtalen: /test-e2e
/test-networkpolicy
dreamtalen: /test-networkpolicy
dreamtalen: /test-networkpolicy
dreamtalen: /test-all

Patch:
diff --git a/build/yamls/antrea-windows.yml b/build/yamls/antrea-windows.yml
index f3a512676..abdfedf2d 100644
--- a/build/yamls/antrea-windows.yml
+++ b/build/yamls/antrea-windows.yml
@@ -86,6 +86,16 @@ data:
 
     # Enable TLS communication from flow exporter to flow aggregator.
     #enableTLSToFlowAggregator: true
+
+    # Determines how traffic is encapsulated. It has the following options:
+    # encap(default):    Inter-node Pod traffic is always encapsulated and Pod to external network
+    #                    traffic is SNAT'd.
+    # noEncap:           Inter-node Pod traffic is not encapsulated; Pod to external network traffic is
+    #                    SNAT'd if noSNAT is not set to true. Underlying network must be capable of
+    #                    supporting Pod traffic across IP subnets.
+    # hybrid:            noEncap if source and destination Nodes are on the same subnet, otherwise encap.
+    #
+    #trafficEncapMode: encap
   antrea-cni.conflist: |
     {
         "cniVersion":"0.3.0",
@@ -104,7 +114,7 @@ kind: ConfigMap
 metadata:
   labels:
     app: antrea
-  name: antrea-windows-config-6cmd972m6b
+  name: antrea-windows-config-g5hmmkbh55
   namespace: kube-system
 ---
 apiVersion: apps/v1
@@ -192,7 +202,7 @@ spec:
         operator: Exists
       volumes:
       - configMap:
-          name: antrea-windows-config-6cmd972m6b
+          name: antrea-windows-config-g5hmmkbh55
         name: antrea-windows-config
       - configMap:
           defaultMode: 420
diff --git a/build/yamls/windows/base/conf/antrea-agent.conf b/build/yamls/windows/base/conf/antrea-agent.conf
index a0a7af490..ead6b3cf1 100644
--- a/build/yamls/windows/base/conf/antrea-agent.conf
+++ b/build/yamls/windows/base/conf/antrea-agent.conf
@@ -68,3 +68,13 @@ featureGates:
 
 # Enable TLS communication from flow exporter to flow aggregator.
 #enableTLSToFlowAggregator: true
+
+# Determines how traffic is encapsulated. It has the following options:
+# encap(default):    Inter-node Pod traffic is always encapsulated and Pod to external network
+#                    traffic is SNAT'd.
+# noEncap:           Inter-node Pod traffic is not encapsulated; Pod to external network traffic is
+#                    SNAT'd if noSNAT is not set to true. Underlying network must be capable of
+#                    supporting Pod traffic across IP subnets.
+# hybrid:            noEncap if source and destination Nodes are on the same subnet, otherwise encap.
+#
+#trafficEncapMode: encap
diff --git a/cmd/antrea-agent/options_windows.go b/cmd/antrea-agent/options_windows.go
index 483714c75..1e72a5092 100644
--- a/cmd/antrea-agent/options_windows.go
+++ b/cmd/antrea-agent/options_windows.go
@@ -42,7 +42,7 @@ func (o *Options) checkUnsupportedFeatures() error {
 		unsupported = append(unsupported, "OVSDatapathType: "+o.config.OVSDatapathType)
 	}
 	_, encapMode := config.GetTrafficEncapModeFromStr(o.config.TrafficEncapMode)
-	if encapMode != config.TrafficEncapModeEncap {
+	if encapMode == config.TrafficEncapModeNetworkPolicyOnly {
 		unsupported = append(unsupported, "TrafficEncapMode: "+encapMode.String())
 	}
 	if o.config.TunnelType == ovsconfig.GRETunnel {
diff --git a/pkg/agent/controller/noderoute/node_route_controller.go b/pkg/agent/controller/noderoute/node_route_controller.go
index 3b90b8c5d..7d2dd79dd 100644
--- a/pkg/agent/controller/noderoute/node_route_controller.go
+++ b/pkg/agent/controller/noderoute/node_route_controller.go
@@ -35,6 +35,7 @@ import (
 	"antrea.io/antrea/pkg/agent/interfacestore"
 	"antrea.io/antrea/pkg/agent/openflow"
 	"antrea.io/antrea/pkg/agent/route"
+	"antrea.io/antrea/pkg/agent/types"
 	"antrea.io/antrea/pkg/agent/util"
 	"antrea.io/antrea/pkg/ovs/ovsconfig"
 	utilip "antrea.io/antrea/pkg/util/ip"
@@ -136,6 +137,7 @@ type nodeRouteInfo struct {
 	podCIDRs  []*net.IPNet
 	nodeIP    net.IP
 	gatewayIP []net.IP
+	nodeMAC   net.HardwareAddr
 }
 
 // enqueueNode adds an object to the controller work queue
@@ -407,8 +409,16 @@ func (c *Controller) deleteNodeRoute(nodeName string) error {
 }
 
 func (c *Controller) addNodeRoute(nodeName string, node *corev1.Node) error {
-	if _, installed, _ := c.installedNodes.GetByKey(nodeName); installed {
-		// Route is already added for this Node.
+	// It is only for Windows Noencap mode to get Node MAC.
+	peerNodeMAC, err := getNodeMAC(node)
+	if err != nil {
+		return fmt.Errorf("error when retrieving MAC of Node %s: %v", nodeName, err)
+	}
+
+	nrInfo, installed, _ := c.installedNodes.GetByKey(nodeName)
+
+	if installed && nrInfo.(*nodeRouteInfo).nodeMAC.String() == peerNodeMAC.String() {
+		// Route is already added for this Node and Node MAC isn't changed.
 		return nil
 	}
 
@@ -432,7 +442,7 @@ func (c *Controller) addNodeRoute(nodeName string, node *corev1.Node) error {
 			return nil
 		}
 
-		nodesHaveSamePodCIDR, _ := c.installedNodes.ByIndex(nodeRouteInfoPodCIDRIndexName, podCIDR)
+		nodesHaveSamePodCIDR, _ := c.installedNodes.IndexKeys(nodeRouteInfoPodCIDRIndexName, podCIDR)
 		// PodCIDRs can be released from deleted Nodes and allocated to new Nodes. For server side, it won't happen that a
 		// PodCIDR is allocated to more than one Node at any point. However, for client side, if a resync happens to occur
 		// when there are Node creation and deletion events, the informer will generate the events in a way that all
@@ -442,9 +452,12 @@ func (c *Controller) addNodeRoute(nodeName string, node *corev1.Node) error {
 		// stale routes, flows, and relevant cache of this podCIDR are removed appropriately, we wait for the Node deletion
 		// event to be processed before proceeding, or the route installation and uninstallation operations may override or
 		// conflict with each other.
-		if len(nodesHaveSamePodCIDR) > 0 {
+		// For Windows Noencap case, it is possible that nodesHaveSamePodCIDR is the Node itself because the Node
+		// MAC annotation was not set yet when the Node was initially installed. Then it is processed for the second
+		// time when its MAC annotation is updated.
+		if len(nodesHaveSamePodCIDR) > 0 && (len(nodesHaveSamePodCIDR) != 1 || nodesHaveSamePodCIDR[0] != nodeName) {
 			// Return an error so that the Node will be put back to the workqueue and will be retried later.
-			return fmt.Errorf("skipping addNodeRoute for Node %s because podCIDR %s is duplicate with Node %s, will retry later", nodeName, podCIDR, nodesHaveSamePodCIDR[0].(*nodeRouteInfo).nodeName)
+			return fmt.Errorf("skipping addNodeRoute for Node %s because podCIDR %s is duplicate with Node %s, will retry later", nodeName, podCIDR, nodesHaveSamePodCIDR[0])
 		}
 
 		peerPodCIDRAddr, peerPodCIDR, err := net.ParseCIDR(podCIDR)
@@ -477,7 +490,8 @@ func (c *Controller) addNodeRoute(nodeName string, node *corev1.Node) error {
 		nodeName,
 		peerConfig,
 		peerNodeIP,
-		uint32(ipsecTunOFPort))
+		uint32(ipsecTunOFPort),
+		peerNodeMAC)
 	if err != nil {
 		return fmt.Errorf("failed to install flows to Node %s: %v", nodeName, err)
 	}
@@ -494,6 +508,7 @@ func (c *Controller) addNodeRoute(nodeName string, node *corev1.Node) error {
 		podCIDRs:  podCIDRs,
 		nodeIP:    peerNodeIP,
 		gatewayIP: peerGatewayIPs,
+		nodeMAC:   peerNodeMAC,
 	})
 	return err
 }
@@ -629,3 +644,16 @@ func (c *Controller) IPInPodSubnets(ip net.IP) bool {
 	nodeInCluster, _ := c.installedNodes.ByIndex(nodeRouteInfoPodCIDRIndexName, ipCIDRStr)
 	return len(nodeInCluster) > 0 || ipCIDRStr == curNodeCIDRStr
 }
+
+// getNodeMAC gets Node's br-int MAC from its annotation. It is only for Windows Noencap mode.
+func getNodeMAC(node *corev1.Node) (net.HardwareAddr, error) {
+	macStr := node.Annotations[types.NodeMACAddressAnnotationKey]
+	if macStr == "" {
+		return nil, nil
+	}
+	mac, err := net.ParseMAC(macStr)
+	if err != nil {
+		return nil, fmt.Errorf("failed to parse MAC `%s`: %v", macStr, err)
+	}
+	return mac, nil
+}
diff --git a/pkg/agent/openflow/client.go b/pkg/agent/openflow/client.go
index d1163fadb..77fd7a001 100644
--- a/pkg/agent/openflow/client.go
+++ b/pkg/agent/openflow/client.go
@@ -68,7 +68,8 @@ type Client interface {
 		hostname string,
 		peerConfigs map[*net.IPNet]net.IP,
 		tunnelPeerIP net.IP,
-		ipsecTunOFPort uint32) error
+		ipsecTunOFPort uint32,
+		peerNodeMAC net.HardwareAddr) error
 
 	// UninstallNodeFlows removes the connection to the remote Node specified with the
 	// hostname. UninstallNodeFlows will do nothing if no connection to the host was established.
@@ -312,6 +313,45 @@ func (c *client) addFlows(cache *flowCategoryCache, flowCacheKey string, flows [
 	return nil
 }
 
+// modifyFlows sets the flows of flowCategoryCache be exactly same as the provided slice for the given flowCacheKey.
+func (c *client) modifyFlows(cache *flowCategoryCache, flowCacheKey string, flows []binding.Flow) error {
+	oldFlowCacheI, ok := cache.Load(flowCacheKey)
+	fCache := flowCache{}
+	var err error
+	if !ok {
+		for _, flow := range flows {
+			fCache[flow.MatchString()] = flow
+		}
+
+		err = c.ofEntryOperations.AddAll(flows)
+	} else {
+		var adds, mods, dels []binding.Flow
+		oldFlowCache := oldFlowCacheI.(flowCache)
+		for _, flow := range flows {
+			matchString := flow.MatchString()
+			if _, ok := oldFlowCache[matchString]; ok {
+				mods = append(mods, flow)
+			} else {
+				adds = append(adds, flow)
+			}
+			fCache[matchString] = flow
+		}
+		for k, v := range oldFlowCache {
+			if _, ok := fCache[k]; !ok {
+				dels = append(dels, v)
+			}
+		}
+		err = c.ofEntryOperations.BundleOps(adds, mods, dels)
+	}
+	if err != nil {
+		return err
+	}
+
+	// Modify the flows in the flow cache.
+	cache.Store(flowCacheKey, fCache)
+	return nil
+}
+
 // deleteFlows deletes all the flows in the flow cache indexed by the provided flowCacheKey.
 func (c *client) deleteFlows(cache *flowCategoryCache, flowCacheKey string) error {
 	fCacheI, ok := cache.Load(flowCacheKey)
@@ -332,10 +372,12 @@ func (c *client) deleteFlows(cache *flowCategoryCache, flowCacheKey string) erro
 	return nil
 }
 
+// InstallNodeFlows installs flows for peer Nodes. Parameter remoteGatewayMAC is only for Windows.
 func (c *client) InstallNodeFlows(hostname string,
 	peerConfigs map[*net.IPNet]net.IP,
 	tunnelPeerIP net.IP,
-	ipsecTunOFPort uint32) error {
+	ipsecTunOFPort uint32,
+	remoteGatewayMAC net.HardwareAddr) error {
 	c.replayMutex.RLock()
 	defer c.replayMutex.RUnlock()
 
@@ -353,7 +395,7 @@ func (c *client) InstallNodeFlows(hostname string,
 			// IPv6 one is decided by the address family of Node Internal Address.
 			flows = append(flows, c.l3FwdFlowToRemote(localGatewayMAC, *peerPodCIDR, tunnelPeerIP, cookie.Node))
 		} else {
-			flows = append(flows, c.l3FwdFlowToRemoteViaGW(localGatewayMAC, *peerPodCIDR, cookie.Node))
+			flows = append(flows, c.l3FwdFlowToRemoteViaRouting(localGatewayMAC, remoteGatewayMAC, cookie.Node, tunnelPeerIP, peerPodCIDR)...)
 		}
 	}
 
@@ -364,7 +406,9 @@ func (c *client) InstallNodeFlows(hostname string,
 		flows = append(flows, c.tunnelClassifierFlow(ipsecTunOFPort, cookie.Node))
 	}
 
-	return c.addFlows(c.nodeFlowCache, hostname, flows)
+	// For Windows Noencap Mode, the OVS flows for Node need be be exactly same as the provided 'flows' slice because
+	// the Node flows may be processed more than once if the MAC annotation is updated.
+	return c.modifyFlows(c.nodeFlowCache, hostname, flows)
 }
 
 func (c *client) UninstallNodeFlows(hostname string) error {
@@ -576,10 +620,7 @@ func (c *client) InstallGatewayFlows() error {
 	// Add flow to ensure the liveness check packet could be forwarded correctly.
 	flows = append(flows, c.localProbeFlow(gatewayIPs, cookie.Default)...)
 	flows = append(flows, c.ctRewriteDstMACFlows(gatewayConfig.MAC, cookie.Default)...)
-	// In NoEncap , no traffic from tunnel port
-	if c.encapMode.SupportsEncap() {
-		flows = append(flows, c.l3FwdFlowToGateway(gatewayIPs, gatewayConfig.MAC, cookie.Default)...)
-	}
+	flows = append(flows, c.l3FwdFlowToGateway(gatewayIPs, gatewayConfig.MAC, cookie.Default)...)
 
 	if err := c.ofEntryOperations.AddAll(flows); err != nil {
 		return err
diff --git a/pkg/agent/openflow/pipeline.go b/pkg/agent/openflow/pipeline.go
index f9e855b38..92da130bf 100644
--- a/pkg/agent/openflow/pipeline.go
+++ b/pkg/agent/openflow/pipeline.go
@@ -359,6 +359,7 @@ type OFEntryOperations interface {
 	Delete(flow binding.Flow) error
 	AddAll(flows []binding.Flow) error
 	ModifyAll(flows []binding.Flow) error
+	BundleOps(adds []binding.Flow, mods []binding.Flow, dels []binding.Flow) error
 	DeleteAll(flows []binding.Flow) error
 	AddOFEntries(ofEntries []binding.OFEntry) error
 	DeleteOFEntries(ofEntries []binding.OFEntry) error
@@ -425,32 +426,34 @@ func (c *client) GetTunnelVirtualMAC() net.HardwareAddr {
 	return globalVirtualMAC
 }
 
-func (c *client) changeAll(flows []binding.Flow, action ofAction) error {
-	if len(flows) == 0 {
+func (c *client) changeAll(flowsMap map[ofAction][]binding.Flow) error {
+	if len(flowsMap) == 0 {
 		return nil
 	}
-	var adds, mods, dels []binding.Flow
-	if action == add {
-		adds = flows
-	} else if action == mod {
-		mods = flows
-	} else if action == del {
-		dels = flows
-	} else {
-		return fmt.Errorf("OF Action not exists: %s", action)
-	}
 
 	startTime := time.Now()
 	defer func() {
 		d := time.Since(startTime)
-		metrics.OVSFlowOpsLatency.WithLabelValues(action.String()).Observe(float64(d.Milliseconds()))
+		for k, v := range flowsMap {
+			if len(v) != 0 {
+				metrics.OVSFlowOpsLatency.WithLabelValues(k.String()).Observe(float64(d.Milliseconds()))
+			}
+		}
 	}()
 
-	if err := c.bridge.AddFlowsInBundle(adds, mods, dels); err != nil {
-		metrics.OVSFlowOpsErrorCount.WithLabelValues(action.String()).Inc()
+	if err := c.bridge.AddFlowsInBundle(flowsMap[add], flowsMap[mod], flowsMap[del]); err != nil {
+		for k, v := range flowsMap {
+			if len(v) != 0 {
+				metrics.OVSFlowOpsErrorCount.WithLabelValues(k.String()).Inc()
+			}
+		}
 		return err
 	}
-	metrics.OVSFlowOpsCount.WithLabelValues(action.String()).Inc()
+	for k, v := range flowsMap {
+		if len(v) != 0 {
+			metrics.OVSFlowOpsCount.WithLabelValues(k.String()).Inc()
+		}
+	}
 	return nil
 }
 
@@ -467,15 +470,19 @@ func (c *client) Delete(flow binding.Flow) error {
 }
 
 func (c *client) AddAll(flows []binding.Flow) error {
-	return c.changeAll(flows, add)
+	return c.changeAll(map[ofAction][]binding.Flow{add: flows})
 }
 
 func (c *client) ModifyAll(flows []binding.Flow) error {
-	return c.changeAll(flows, mod)
+	return c.changeAll(map[ofAction][]binding.Flow{mod: flows})
 }
 
 func (c *client) DeleteAll(flows []binding.Flow) error {
-	return c.changeAll(flows, del)
+	return c.changeAll(map[ofAction][]binding.Flow{del: flows})
+}
+
+func (c *client) BundleOps(adds []binding.Flow, mods []binding.Flow, dels []binding.Flow) error {
+	return c.changeAll(map[ofAction][]binding.Flow{add: adds, mod: mods, del: dels})
 }
 
 func (c *client) changeOFEntries(ofEntries []binding.OFEntry, action ofAction) error {
@@ -492,7 +499,6 @@ func (c *client) changeOFEntries(ofEntries []binding.OFEntry, action ofAction) e
 	} else {
 		return fmt.Errorf("OF Entries Action not exists: %s", action)
 	}
-
 	startTime := time.Now()
 	defer func() {
 		d := time.Since(startTime)
@@ -1171,9 +1177,8 @@ func (c *client) l3FwdFlowRouteToGW(gwMAC net.HardwareAddr, category cookie.Cate
 	return flows
 }
 
-// l3FwdFlowToGateway generates the L3 forward flows for traffic from tunnel to
-// the local gateway. It rewrites the destination MAC (should be
-// globalVirtualMAC) of the packets to the gateway interface MAC.
+// l3FwdFlowToGateway generates the L3 forward flows to rewrite the destination MAC of the packets to the gateway interface
+// MAC if the destination IP is the gateway IP.
 func (c *client) l3FwdFlowToGateway(localGatewayIPs []net.IP, localGatewayMAC net.HardwareAddr, category cookie.Category) []binding.Flow {
 	l3FwdTable := c.pipeline[l3ForwardingTable]
 	var flows []binding.Flow
diff --git a/pkg/agent/openflow/pipeline_other.go b/pkg/agent/openflow/pipeline_other.go
index 3968a22e9..5c11630da 100644
--- a/pkg/agent/openflow/pipeline_other.go
+++ b/pkg/agent/openflow/pipeline_other.go
@@ -35,3 +35,8 @@ func (c *client) externalFlows(nodeIP net.IP, localSubnet net.IPNet, localGatewa
 func (c *client) snatMarkFlows(snatIP net.IP, mark uint32) []binding.Flow {
 	return []binding.Flow{c.snatIPFromTunnelFlow(snatIP, mark)}
 }
+
+func (c *client) l3FwdFlowToRemoteViaRouting(localGatewayMAC net.HardwareAddr, remoteGatewayMAC net.HardwareAddr,
+	category cookie.Category, peerIP net.IP, peerPodCIDR *net.IPNet) []binding.Flow {
+	return []binding.Flow{c.l3FwdFlowToRemoteViaGW(localGatewayMAC, *peerPodCIDR, category)}
+}
diff --git a/pkg/agent/openflow/pipeline_windows.go b/pkg/agent/openflow/pipeline_windows.go
index 187d84c5a..17abde490 100644
--- a/pkg/agent/openflow/pipeline_windows.go
+++ b/pkg/agent/openflow/pipeline_windows.go
@@ -281,5 +281,33 @@ func (c *client) hostBridgeUplinkFlows(localSubnet net.IPNet, category cookie.Ca
 			Cookie(c.cookieAllocator.Request(category).Raw()).
 			Done(),
 	}
+	if c.encapMode.SupportsNoEncap() {
+		// If NoEncap is enabled, the reply packets from remote Pod can be forwarded to local Pod directly.
+		// by explicitly resubmitting them to endpointDNATTable and marking "macRewriteMark" at same time.
+		flows = append(flows, c.pipeline[conntrackStateTable].BuildFlow(priorityHigh).MatchProtocol(binding.ProtocolIP).
+			MatchRegRange(int(marksReg), markTrafficFromUplink, binding.Range{0, 15}).
+			MatchDstIPNet(localSubnet).
+			Action().LoadRegRange(int(marksReg), macRewriteMark, macRewriteMarkRange).
+			Action().GotoTable(endpointDNATTable).
+			Cookie(c.cookieAllocator.Request(category).Raw()).
+			Done())
+	}
 	return flows
 }
+
+func (c *client) l3FwdFlowToRemoteViaRouting(localGatewayMAC net.HardwareAddr, remoteGatewayMAC net.HardwareAddr,
+	category cookie.Category, peerIP net.IP, peerPodCIDR *net.IPNet) []binding.Flow {
+	if !c.encapMode.NeedsRoutingToPeer(peerIP, c.nodeConfig.NodeIPAddr) && remoteGatewayMAC != nil {
+		// It enhances Windows Noencap mode performance by bypassing host network.
+		flows := []binding.Flow{c.pipeline[l2ForwardingCalcTable].BuildFlow(priorityNormal).
+			MatchDstMAC(remoteGatewayMAC).
+			Action().LoadRegRange(int(PortCacheReg), config.UplinkOFPort, ofPortRegRange).
+			Action().LoadRegRange(int(marksReg), macRewriteMark, ofPortMarkRange).
+			Action().GotoTable(conntrackCommitTable).
+			Cookie(c.cookieAllocator.Request(category).Raw()).
+			Done()}
+		flows = append(flows, c.l3FwdFlowToRemoteViaGW(remoteGatewayMAC, *peerPodCIDR, category))
+		return flows
+	}
+	return []binding.Flow{c.l3FwdFlowToRemoteViaGW(localGatewayMAC, *peerPodCIDR, category)}
+}
diff --git a/pkg/agent/route/route_windows.go b/pkg/agent/route/route_windows.go
index b0d4a3032..59f1a69cc 100644
--- a/pkg/agent/route/route_windows.go
+++ b/pkg/agent/route/route_windows.go
@@ -18,6 +18,7 @@ package route
 
 import (
 	"errors"
+	"fmt"
 	"net"
 	"sync"
 
@@ -36,11 +37,13 @@ const (
 )
 
 type Client struct {
-	nr          netroute.Interface
-	nodeConfig  *config.NodeConfig
-	serviceCIDR *net.IPNet
-	hostRoutes  *sync.Map
-	fwClient    *winfirewall.Client
+	nr             netroute.Interface
+	nodeConfig     *config.NodeConfig
+	networkConfig  *config.NetworkConfig
+	serviceCIDR    *net.IPNet
+	hostRoutes     *sync.Map
+	fwClient       *winfirewall.Client
+	bridgeInfIndex int
 }
 
 // NewClient returns a route client.
@@ -48,10 +51,11 @@ type Client struct {
 func NewClient(serviceCIDR *net.IPNet, networkConfig *config.NetworkConfig, noSNAT bool) (*Client, error) {
 	nr := netroute.New()
 	return &Client{
-		nr:          nr,
-		serviceCIDR: serviceCIDR,
-		hostRoutes:  &sync.Map{},
-		fwClient:    winfirewall.NewClient(),
+		nr:            nr,
+		networkConfig: networkConfig,
+		serviceCIDR:   serviceCIDR,
+		hostRoutes:    &sync.Map{},
+		fwClient:      winfirewall.NewClient(),
 	}, nil
 }
 
@@ -59,6 +63,11 @@ func NewClient(serviceCIDR *net.IPNet, networkConfig *config.NetworkConfig, noSN
 // Service LoadBalancing is provided by OpenFlow.
 func (c *Client) Initialize(nodeConfig *config.NodeConfig, done func()) error {
 	c.nodeConfig = nodeConfig
+	bridgeInf, err := net.InterfaceByName(nodeConfig.OVSBridge)
+	if err != nil {
+		return fmt.Errorf("failed to find the interface %s: %v", nodeConfig.OVSBridge, err)
+	}
+	c.bridgeInfIndex = bridgeInf.Index
 	if err := c.initFwRules(); err != nil {
 		return err
 	}
@@ -100,26 +109,43 @@ func (c *Client) Reconcile(podCIDRs []string) error {
 // It overrides the routes if they already exist, without error.
 func (c *Client) AddRoutes(podCIDR *net.IPNet, nodeName string, peerNodeIP, peerGwIP net.IP) error {
 	obj, found := c.hostRoutes.Load(podCIDR.String())
+	route := &netroute.Route{
+		DestinationSubnet: podCIDR,
+	}
+	if c.networkConfig.TrafficEncapMode.NeedsEncapToPeer(peerNodeIP, c.nodeConfig.NodeIPAddr) {
+		route.LinkIndex = c.nodeConfig.GatewayConfig.LinkIndex
+		route.GatewayAddress = peerGwIP
+	} else if !c.networkConfig.TrafficEncapMode.NeedsRoutingToPeer(peerNodeIP, c.nodeConfig.NodeIPAddr) {
+		// NoEncap traffic to Node on the same subnet.
+		// Set the peerNodeIP as next hop.
+		route.LinkIndex = c.bridgeInfIndex
+		route.GatewayAddress = peerNodeIP
+	}
+	// NoEncap traffic to Node on the different subnet needs underlying routing support.
+	// Use host default route inside the Node.
+
 	if found {
-		rt := obj.(*netroute.Route)
-		if rt.GatewayAddress.Equal(peerGwIP) {
+		existingRoute := obj.(*netroute.Route)
+		if existingRoute.GatewayAddress.Equal(route.GatewayAddress) {
 			klog.V(4).Infof("Route with destination %s already exists on %s (%s)", podCIDR.String(), nodeName, peerNodeIP)
 			return nil
 		}
 		// Remove the existing route entry if the gateway address is not as expected.
-		if err := c.nr.RemoveNetRoute(rt.LinkIndex, rt.DestinationSubnet, rt.GatewayAddress); err != nil {
+		if err := c.nr.RemoveNetRoute(existingRoute.LinkIndex, existingRoute.DestinationSubnet, existingRoute.GatewayAddress); err != nil {
 			klog.Errorf("Failed to delete existing route entry with destination %s gateway %s on %s (%s)", podCIDR.String(), peerGwIP.String(), nodeName, peerNodeIP)
 			return err
 		}
 	}
-	if err := c.nr.NewNetRoute(c.nodeConfig.GatewayConfig.LinkIndex, podCIDR, peerGwIP); err != nil {
+
+	if route.GatewayAddress == nil {
+		return nil
+	}
+
+	if err := c.nr.NewNetRoute(route.LinkIndex, route.DestinationSubnet, route.GatewayAddress); err != nil {
 		return err
 	}
-	c.hostRoutes.Store(podCIDR.String(), &netroute.Route{
-		LinkIndex:         c.nodeConfig.GatewayConfig.LinkIndex,
-		DestinationSubnet: podCIDR,
-		GatewayAddress:    peerGwIP,
-	})
+
+	c.hostRoutes.Store(podCIDR.String(), route)
 	klog.V(2).Infof("Added route with destination %s via %s on host gateway on %s (%s)", podCIDR.String(), peerGwIP.String(), nodeName, peerNodeIP)
 	return nil
 }


Test patch:
diff --git a/cmd/antrea-agent/options_windows_test.go b/cmd/antrea-agent/options_windows_test.go
index 7c5a57185..f27632b28 100644
--- a/cmd/antrea-agent/options_windows_test.go
+++ b/cmd/antrea-agent/options_windows_test.go
@@ -57,7 +57,7 @@ func TestCheckUnsupportedFeatures(t *testing.T) {
 		{
 			"noEncap mode",
 			AgentConfig{TrafficEncapMode: config.TrafficEncapModeNoEncap.String()},
-			false,
+			true,
 		},
 		{
 			"GRE tunnel",
diff --git a/pkg/agent/controller/noderoute/node_route_controller_test.go b/pkg/agent/controller/noderoute/node_route_controller_test.go
index 66873b14a..162055cbe 100644
--- a/pkg/agent/controller/noderoute/node_route_controller_test.go
+++ b/pkg/agent/controller/noderoute/node_route_controller_test.go
@@ -133,7 +133,7 @@ func TestControllerWithDuplicatePodCIDR(t *testing.T) {
 		c.clientset.CoreV1().Nodes().Create(context.TODO(), node1, metav1.CreateOptions{})
 		// The 2nd argument is Any() because the argument is unpredictable when it uses pointer as the key of map.
 		// The argument type is map[*net.IPNet]net.IP.
-		c.ofClient.EXPECT().InstallNodeFlows("node1", gomock.Any(), nodeIP1, uint32(0)).Times(1)
+		c.ofClient.EXPECT().InstallNodeFlows("node1", gomock.Any(), nodeIP1, uint32(0), nil).Times(1)
 		c.routeClient.EXPECT().AddRoutes(podCIDR, "node1", nodeIP1, podCIDRGateway).Times(1)
 		c.processNextWorkItem()
 
@@ -150,7 +150,7 @@ func TestControllerWithDuplicatePodCIDR(t *testing.T) {
 		// After node1 is deleted, routes and flows should be installed for node2 successfully.
 		// The 2nd argument is Any() because the argument is unpredictable when it uses pointer as the key of map.
 		// The argument type is map[*net.IPNet]net.IP.
-		c.ofClient.EXPECT().InstallNodeFlows("node2", gomock.Any(), nodeIP2, uint32(0)).Times(1)
+		c.ofClient.EXPECT().InstallNodeFlows("node2", gomock.Any(), nodeIP2, uint32(0), nil).Times(1)
 		c.routeClient.EXPECT().AddRoutes(podCIDR, "node2", nodeIP2, podCIDRGateway).Times(1)
 		c.processNextWorkItem()
 	}()
@@ -214,12 +214,12 @@ func TestIPInPodSubnets(t *testing.T) {
 	c.clientset.CoreV1().Nodes().Create(context.TODO(), node1, metav1.CreateOptions{})
 	// The 2nd argument is Any() because the argument is unpredictable when it uses pointer as the key of map.
 	// The argument type is map[*net.IPNet]net.IP.
-	c.ofClient.EXPECT().InstallNodeFlows("node1", gomock.Any(), nodeIP1, uint32(0)).Times(1)
+	c.ofClient.EXPECT().InstallNodeFlows("node1", gomock.Any(), nodeIP1, uint32(0), nil).Times(1)
 	c.routeClient.EXPECT().AddRoutes(podCIDR, "node1", nodeIP1, podCIDRGateway).Times(1)
 	c.processNextWorkItem()
 
 	c.clientset.CoreV1().Nodes().Create(context.TODO(), node2, metav1.CreateOptions{})
-	c.ofClient.EXPECT().InstallNodeFlows("node2", gomock.Any(), nodeIP2, uint32(0)).Times(1)
+	c.ofClient.EXPECT().InstallNodeFlows("node2", gomock.Any(), nodeIP2, uint32(0), nil).Times(1)
 	c.routeClient.EXPECT().AddRoutes(podCIDR2, "node2", nodeIP2, podCIDR2Gateway).Times(1)
 	c.processNextWorkItem()
 
diff --git a/pkg/agent/openflow/client_test.go b/pkg/agent/openflow/client_test.go
index 7c12985cd..7df57c190 100644
--- a/pkg/agent/openflow/client_test.go
+++ b/pkg/agent/openflow/client_test.go
@@ -57,7 +57,7 @@ func installNodeFlows(ofClient Client, cacheKey string) (int, error) {
 	peerConfig := map[*net.IPNet]net.IP{
 		ipNet: gwIP,
 	}
-	err := ofClient.InstallNodeFlows(hostName, peerConfig, peerNodeIP, 0)
+	err := ofClient.InstallNodeFlows(hostName, peerConfig, peerNodeIP, 0, nil)
 	client := ofClient.(*client)
 	fCacheI, ok := client.nodeFlowCache.Load(hostName)
 	if ok {
@@ -90,7 +90,6 @@ func TestIdempotentFlowInstallation(t *testing.T) {
 		numFlows  int
 		installFn func(ofClient Client, cacheKey string) (int, error)
 	}{
-		{"NodeFlows", "host", 2, installNodeFlows},
 		{"PodFlows", "aaaa-bbbb-cccc-dddd", 5, installPodFlows},
 	}
 
diff --git a/pkg/agent/openflow/testing/mock_openflow.go b/pkg/agent/openflow/testing/mock_openflow.go
index 0f68c6550..917fca885 100644
--- a/pkg/agent/openflow/testing/mock_openflow.go
+++ b/pkg/agent/openflow/testing/mock_openflow.go
@@ -349,17 +349,17 @@ func (mr *MockClientMockRecorder) InstallLoadBalancerServiceFromOutsideFlows(arg
 }
 
 // InstallNodeFlows mocks base method
-func (m *MockClient) InstallNodeFlows(arg0 string, arg1 map[*net.IPNet]net.IP, arg2 net.IP, arg3 uint32) error {
+func (m *MockClient) InstallNodeFlows(arg0 string, arg1 map[*net.IPNet]net.IP, arg2 net.IP, arg3 uint32, arg4 net.HardwareAddr) error {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "InstallNodeFlows", arg0, arg1, arg2, arg3)
+	ret := m.ctrl.Call(m, "InstallNodeFlows", arg0, arg1, arg2, arg3, arg4)
 	ret0, _ := ret[0].(error)
 	return ret0
 }
 
 // InstallNodeFlows indicates an expected call of InstallNodeFlows
-func (mr *MockClientMockRecorder) InstallNodeFlows(arg0, arg1, arg2, arg3 interface{}) *gomock.Call {
+func (mr *MockClientMockRecorder) InstallNodeFlows(arg0, arg1, arg2, arg3, arg4 interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "InstallNodeFlows", reflect.TypeOf((*MockClient)(nil).InstallNodeFlows), arg0, arg1, arg2, arg3)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "InstallNodeFlows", reflect.TypeOf((*MockClient)(nil).InstallNodeFlows), arg0, arg1, arg2, arg3, arg4)
 }
 
 // InstallPodFlows mocks base method
@@ -828,6 +828,20 @@ func (mr *MockOFEntryOperationsMockRecorder) AddOFEntries(arg0 interface{}) *gom
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AddOFEntries", reflect.TypeOf((*MockOFEntryOperations)(nil).AddOFEntries), arg0)
 }
 
+// BundleOps mocks base method
+func (m *MockOFEntryOperations) BundleOps(arg0, arg1, arg2 []openflow.Flow) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "BundleOps", arg0, arg1, arg2)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// BundleOps indicates an expected call of BundleOps
+func (mr *MockOFEntryOperationsMockRecorder) BundleOps(arg0, arg1, arg2 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "BundleOps", reflect.TypeOf((*MockOFEntryOperations)(nil).BundleOps), arg0, arg1, arg2)
+}
+
 // Delete mocks base method
 func (m *MockOFEntryOperations) Delete(arg0 openflow.Flow) error {
 	m.ctrl.T.Helper()
diff --git a/pkg/agent/route/route_windows_test.go b/pkg/agent/route/route_windows_test.go
index bef208b77..7ce0d43c4 100644
--- a/pkg/agent/route/route_windows_test.go
+++ b/pkg/agent/route/route_windows_test.go
@@ -56,6 +56,7 @@ func TestRouteOperation(t *testing.T) {
 	client, err := NewClient(serviceCIDR, &config.NetworkConfig{}, false)
 	require.Nil(t, err)
 	nodeConfig := &config.NodeConfig{
+		OVSBridge: "Loopback Pseudo-Interface 1",
 		GatewayConfig: &config.GatewayConfig{
 			Name:      hostGateway,
 			LinkIndex: gwLink,
diff --git a/test/e2e/flowaggregator_test.go b/test/e2e/flowaggregator_test.go
index 5c29a7d3e..4177a6ecd 100644
--- a/test/e2e/flowaggregator_test.go
+++ b/test/e2e/flowaggregator_test.go
@@ -106,11 +106,14 @@ const (
 	egressRejectANPName           = "test-flow-aggregator-anp-engress-reject"
 	egressDropANPName             = "test-flow-aggregator-anp-engress-drop"
 	egressDenyNPName              = "test-flow-aggregator-np-egress-deny"
-	collectorCheckTimeout         = 10 * time.Second
+	collectorCheckTimeout         = 12 * time.Second
+	iperfTimeSec                  = 12
 	// Single iperf run results in two connections with separate ports (control connection and actual data connection).
-	// As 5s is export interval and iperf traffic runs for 10s, we expect about 4 records exporting to the flow aggregator.
-	// Since flow aggregator will aggregate records based on 5-tuple connection key, we expect 2 records.
-	expectedNumDataRecords = 2
+	// As 2s is the export active timeout of flow exporter and iperf traffic runs for 12s, we expect totally 12 records
+	// exporting to the flow aggregator at time 2s, 4s, 6s, 8s, 10s, and 12s after iperf traffic begins.
+	// Since flow aggregator will aggregate records based on 5-tuple connection key and active timeout is 3.5 seconds,
+	// we expect 3 records at time 5.5s, 9s, and 12.5s after iperf traffic begins.
+	expectedNumDataRecords = 3
 )
 
 type testFlow struct {
@@ -153,6 +156,8 @@ func testHelper(t *testing.T, data *TestData, podAIPs, podBIPs, podCIPs, podDIPs
 	// Wait for the Service to be realized.
 	time.Sleep(3 * time.Second)
 
+	// OVS userspace implementation of conntrack doesn't maintain packet or byte counter statistics, so we ignore the bandwidth test in Kind cluster.
+	checkBandwidth := testOptions.providerName != "kind"
 	// IntraNodeFlows tests the case, where Pods are deployed on same Node and their flow information is exported as IPFIX flow records.
 	t.Run("IntraNodeFlows", func(t *testing.T) {
 		np1, np2 := deployNetworkPolicies(t, data, "perftest-a", "perftest-b")
@@ -168,11 +173,10 @@ func testHelper(t *testing.T, data *TestData, podAIPs, podBIPs, podCIPs, podDIPs
 				}
 			}
 		}()
-		// TODO: Skipping bandwidth check for Intra-Node flows as it is flaky.
 		if !isIPv6 {
-			checkRecordsForFlows(t, data, podAIPs.ipv4.String(), podBIPs.ipv4.String(), isIPv6, true, false, true, false)
+			checkRecordsForFlows(t, data, podAIPs.ipv4.String(), podBIPs.ipv4.String(), isIPv6, true, false, true, checkBandwidth)
 		} else {
-			checkRecordsForFlows(t, data, podAIPs.ipv6.String(), podBIPs.ipv6.String(), isIPv6, true, false, true, false)
+			checkRecordsForFlows(t, data, podAIPs.ipv6.String(), podBIPs.ipv6.String(), isIPv6, true, false, true, checkBandwidth)
 		}
 	})
 
@@ -298,9 +302,9 @@ func testHelper(t *testing.T, data *TestData, podAIPs, podBIPs, podCIPs, podDIPs
 			}
 		}()
 		if !isIPv6 {
-			checkRecordsForFlows(t, data, podAIPs.ipv4.String(), podCIPs.ipv4.String(), isIPv6, false, false, true, true)
+			checkRecordsForFlows(t, data, podAIPs.ipv4.String(), podCIPs.ipv4.String(), isIPv6, false, false, true, checkBandwidth)
 		} else {
-			checkRecordsForFlows(t, data, podAIPs.ipv6.String(), podCIPs.ipv6.String(), isIPv6, false, false, true, true)
+			checkRecordsForFlows(t, data, podAIPs.ipv6.String(), podCIPs.ipv6.String(), isIPv6, false, false, true, checkBandwidth)
 		}
 	})
 
@@ -412,11 +416,10 @@ func testHelper(t *testing.T, data *TestData, podAIPs, podBIPs, podCIPs, podDIPs
 	// LocalServiceAccess tests the case, where Pod and Service are deployed on the same Node and their flow information is exported as IPFIX flow records.
 	t.Run("LocalServiceAccess", func(t *testing.T) {
 		skipIfProxyDisabled(t, data)
-		// TODO: Skipping bandwidth check for LocalServiceAccess flows as it is flaky.
 		if !isIPv6 {
-			checkRecordsForFlows(t, data, podAIPs.ipv4.String(), svcB.Spec.ClusterIP, isIPv6, true, true, false, false)
+			checkRecordsForFlows(t, data, podAIPs.ipv4.String(), svcB.Spec.ClusterIP, isIPv6, true, true, false, checkBandwidth)
 		} else {
-			checkRecordsForFlows(t, data, podAIPs.ipv6.String(), svcB.Spec.ClusterIP, isIPv6, true, true, false, false)
+			checkRecordsForFlows(t, data, podAIPs.ipv6.String(), svcB.Spec.ClusterIP, isIPv6, true, true, false, checkBandwidth)
 		}
 	})
 
@@ -424,26 +427,39 @@ func testHelper(t *testing.T, data *TestData, podAIPs, podBIPs, podCIPs, podDIPs
 	t.Run("RemoteServiceAccess", func(t *testing.T) {
 		skipIfProxyDisabled(t, data)
 		if !isIPv6 {
-			checkRecordsForFlows(t, data, podAIPs.ipv4.String(), svcC.Spec.ClusterIP, isIPv6, false, true, false, true)
+			checkRecordsForFlows(t, data, podAIPs.ipv4.String(), svcC.Spec.ClusterIP, isIPv6, false, true, false, checkBandwidth)
 		} else {
-			checkRecordsForFlows(t, data, podAIPs.ipv6.String(), svcC.Spec.ClusterIP, isIPv6, false, true, false, true)
+			checkRecordsForFlows(t, data, podAIPs.ipv6.String(), svcC.Spec.ClusterIP, isIPv6, false, true, false, checkBandwidth)
 		}
 	})
 }
 
 func checkRecordsForFlows(t *testing.T, data *TestData, srcIP string, dstIP string, isIPv6 bool, isIntraNode bool, checkService bool, checkNetworkPolicy bool, checkBandwidth bool) {
 	timeStart := time.Now()
+	timeStartSec := timeStart.Unix()
 	var cmdStr string
 	if !isIPv6 {
-		cmdStr = fmt.Sprintf("iperf3 -c %s|grep sender|awk '{print $7,$8}'", dstIP)
+		cmdStr = fmt.Sprintf("iperf3 -c %s -t %d|grep sender|awk '{print $7,$8}'", dstIP, iperfTimeSec)
 	} else {
-		cmdStr = fmt.Sprintf("iperf3 -6 -c %s|grep sender|awk '{print $7,$8}'", dstIP)
+		cmdStr = fmt.Sprintf("iperf3 -6 -c %s -t %d|grep sender|awk '{print $7,$8}'", dstIP, iperfTimeSec)
 	}
 	stdout, _, err := data.runCommandFromPod(testNamespace, "perftest-a", "perftool", []string{"bash", "-c", cmdStr})
 	if err != nil {
 		t.Errorf("Error when running iperf3 client: %v", err)
 	}
 	bandwidth := strings.TrimSpace(stdout)
+	bwSlice := strings.Split(bandwidth, " ")
+	// bandwidth from iperf output
+	bandwidthInFloat, err := strconv.ParseFloat(bwSlice[0], 64)
+	require.NoErrorf(t, err, "Error when converting iperf bandwidth %s to float64 type", bwSlice[0])
+	var bandwidthInMbps float64
+	if strings.Contains(bwSlice[1], "Mbits") {
+		bandwidthInMbps = bandwidthInFloat
+	} else if strings.Contains(bwSlice[1], "Gbits") {
+		bandwidthInMbps = bandwidthInFloat * float64(1024)
+	} else {
+		t.Fatalf("Unit of the traffic bandwidth reported by iperf should either be Mbits or Gbits, failing the test.")
+	}
 
 	// Polling to make sure all the data records corresponding to the iperf flow
 	// are received.
@@ -452,7 +468,16 @@ func checkRecordsForFlows(t *testing.T, data *TestData, srcIP string, dstIP stri
 		if err != nil || rc != 0 {
 			return false, err
 		}
-		return strings.Count(collectorOutput, srcIP) >= expectedNumDataRecords && strings.Count(collectorOutput, dstIP) >= expectedNumDataRecords, nil
+		recordSlices := getRecordsFromOutput(collectorOutput)
+		for _, record := range recordSlices {
+			exportTime := int64(getUnit64FieldFromRecord(t, record, "flowEndSeconds"))
+			if strings.Contains(record, srcIP) && strings.Contains(record, dstIP) {
+				if exportTime >= timeStartSec+iperfTimeSec {
+					return true, nil
+				}
+			}
+		}
+		return false, nil
 	})
 	require.NoErrorf(t, err, "IPFIX collector did not receive the expected records and timed out with error: %v", err)
 
@@ -460,58 +485,71 @@ func checkRecordsForFlows(t *testing.T, data *TestData, srcIP string, dstIP stri
 	if err != nil || rc != 0 {
 		t.Errorf("Error when getting logs %v, rc: %v", err, rc)
 	}
-
 	// Iterate over recordSlices and build some results to test with expected results
 	recordSlices := getRecordsFromOutput(collectorOutput)
 	dataRecordsCount := 0
+	var octetTotalCount uint64
 	for _, record := range recordSlices {
 		if strings.Contains(record, srcIP) && strings.Contains(record, dstIP) {
 			dataRecordsCount = dataRecordsCount + 1
-			// In Kind clusters, there are two flow records for the iperf flow.
-			// One of them has no bytes and we ignore that flow record.
-			if !strings.Contains(record, "octetDeltaCount: 0") {
-				// Check if record has both Pod name of source and destination pod.
+			// Check if record has both Pod name of source and destination Pod.
+			if isIntraNode {
+				checkPodAndNodeData(t, record, "perftest-a", controlPlaneNodeName(), "perftest-b", controlPlaneNodeName())
+				checkFlowType(t, record, ipfixregistry.FlowTypeIntraNode)
+			} else {
+				checkPodAndNodeData(t, record, "perftest-a", controlPlaneNodeName(), "perftest-c", workerNodeName(1))
+				checkFlowType(t, record, ipfixregistry.FlowTypeInterNode)
+			}
+
+			if checkService {
 				if isIntraNode {
-					checkPodAndNodeData(t, record, "perftest-a", controlPlaneNodeName(), "perftest-b", controlPlaneNodeName())
-					checkFlowType(t, record, ipfixregistry.FlowTypeIntraNode)
+					if !strings.Contains(record, "antrea-test/perftest-b") {
+						t.Errorf("Record with ServiceIP does not have Service name")
+					}
 				} else {
-					checkPodAndNodeData(t, record, "perftest-a", controlPlaneNodeName(), "perftest-c", workerNodeName(1))
-					checkFlowType(t, record, ipfixregistry.FlowTypeInterNode)
-				}
-
-				if checkService {
-					if isIntraNode {
-						if !strings.Contains(record, "antrea-test/perftest-b") {
-							t.Errorf("Record with ServiceIP does not have Service name")
-						}
-					} else {
-						if !strings.Contains(record, "antrea-test/perftest-c") {
-							t.Errorf("Record with ServiceIP does not have Service name")
-						}
+					if !strings.Contains(record, "antrea-test/perftest-c") {
+						t.Errorf("Record with ServiceIP does not have Service name")
 					}
 				}
-				if checkNetworkPolicy {
-					// Check if records have both ingress and egress network policies.
-					if !strings.Contains(record, ingressAllowNetworkPolicyName) {
-						t.Errorf("Record does not have NetworkPolicy name with ingress rule")
-					}
-					if !strings.Contains(record, fmt.Sprintf("%s: %s", "ingressNetworkPolicyNamespace", testNamespace)) {
-						t.Errorf("Record does not have correct ingressNetworkPolicyNamespace")
-					}
-					if !strings.Contains(record, egressAllowNetworkPolicyName) {
-						t.Errorf("Record does not have NetworkPolicy name with egress rule")
-					}
-					if !strings.Contains(record, fmt.Sprintf("%s: %s", "egressNetworkPolicyNamespace", testNamespace)) {
-						t.Errorf("Record does not have correct egressNetworkPolicyNamespace")
-					}
+			}
+			if checkNetworkPolicy {
+				// Check if records have both ingress and egress network policies.
+				if !strings.Contains(record, ingressAllowNetworkPolicyName) {
+					t.Errorf("Record does not have NetworkPolicy name with ingress rule")
 				}
-				// Check the bandwidth using octetDeltaCount in data record.
-				if checkBandwidth {
-					checkBandwidthFromRecord(t, record, bandwidth)
+				if !strings.Contains(record, fmt.Sprintf("%s: %s", "ingressNetworkPolicyNamespace", testNamespace)) {
+					t.Errorf("Record does not have correct ingressNetworkPolicyNamespace")
+				}
+				if !strings.Contains(record, egressAllowNetworkPolicyName) {
+					t.Errorf("Record does not have NetworkPolicy name with egress rule")
+				}
+				if !strings.Contains(record, fmt.Sprintf("%s: %s", "egressNetworkPolicyNamespace", testNamespace)) {
+					t.Errorf("Record does not have correct egressNetworkPolicyNamespace")
+				}
+			}
+			// Skip the bandwidth check for the iperf control flow records which have 0 delta count.
+			if checkBandwidth && !strings.Contains(record, "octetDeltaCount: 0") {
+				exportTime := int64(getUnit64FieldFromRecord(t, record, "flowEndSeconds"))
+				curOctetTotalCount := getUnit64FieldFromRecord(t, record, "octetTotalCountFromSourceNode")
+				if curOctetTotalCount > octetTotalCount {
+					octetTotalCount = curOctetTotalCount
+				}
+				curOctetDeltaCount := getUnit64FieldFromRecord(t, record, "octetDeltaCountFromSourceNode")
+				// Check the bandwidth using octetDeltaCountFromSourceNode, if this record
+				// is not either the first record or the last in the stream of records.
+				if curOctetDeltaCount != curOctetTotalCount && exportTime < timeStartSec+iperfTimeSec {
+					t.Logf("Check the bandwidth using octetDeltaCountFromSourceNode in data record.")
+					// This middle record should aggregate two records from Flow Exporter
+					checkBandwidthByInterval(t, bandwidthInMbps, curOctetDeltaCount, float64(2*exporterActiveFlowExportTimeout/time.Second), "octetDeltaCountFromSourceNode")
 				}
 			}
 		}
 	}
+	// Average bandwidth check is done after iterating through all records using the largest octetTotalCountFromSourceNode.
+	if checkBandwidth && octetTotalCount > 0 {
+		t.Logf("Check the average bandwidth using octetTotalCountFromSourceNode %v in data record.", octetTotalCount)
+		checkBandwidthByInterval(t, bandwidthInMbps, octetTotalCount, float64(iperfTimeSec), "octetTotalCountFromSourceNode")
+	}
 	// Checking only data records as data records cannot be decoded without template
 	// record.
 	assert.GreaterOrEqualf(t, dataRecordsCount, expectedNumDataRecords, "IPFIX collector should receive expected number of flow records. Considered records: ", len(recordSlices))
@@ -597,6 +635,12 @@ func checkRecordsForDenyFlows(t *testing.T, data *TestData, testFlow1, testFlow2
 	}
 }
 
+func checkBandwidthByInterval(t *testing.T, bandwidthInMbps float64, octetCount uint64, interval float64, field string) {
+	recBandwidth := float64(octetCount) * 8 / 1000000 / interval
+	t.Logf("Iperf throughput: %.2f Mbits/s, IPFIX record throughput calculated through %s: %.2f Mbits/s", bandwidthInMbps, field, recBandwidth)
+	assert.InDeltaf(t, recBandwidth, bandwidthInMbps, bandwidthInMbps*0.15, "Difference between Iperf bandwidth and IPFIX record bandwidth calculated through %s should be lower than 15%%", field)
+}
+
 func checkPodAndNodeData(t *testing.T, record, srcPod, srcNode, dstPod, dstNode string) {
 	if !strings.Contains(record, srcPod) {
 		t.Errorf("Record with srcIP does not have Pod name")
@@ -618,35 +662,17 @@ func checkPodAndNodeData(t *testing.T, record, srcPod, srcNode, dstPod, dstNode
 	}
 }
 
-func checkBandwidthFromRecord(t *testing.T, record, bandwidth string) {
-	// Split the record in lines to compute bandwidth
+func getUnit64FieldFromRecord(t *testing.T, record string, field string) uint64 {
 	splitLines := strings.Split(record, "\n")
 	for _, line := range splitLines {
-		if strings.Contains(line, "octetDeltaCount:") {
+		if strings.Contains(line, field) {
 			lineSlice := strings.Split(line, ":")
-			deltaBytes, err := strconv.ParseFloat(strings.TrimSpace(lineSlice[1]), 64)
-			if err != nil {
-				t.Errorf("Error in converting octetDeltaCount to float type")
-			}
-			// Flow Aggregator uses 5s as export interval; we use
-			// 2s as export interval for Flow Exporter.
-			recBandwidth := (deltaBytes * 8.0) / float64(5*time.Second.Nanoseconds())
-			// bandwidth from iperf output
-			bwSlice := strings.Split(bandwidth, " ")
-			iperfBandwidth, err := strconv.ParseFloat(bwSlice[0], 64)
-			if err != nil {
-				t.Errorf("Error in converting iperf bandwidth to float64 type")
-			}
-			if strings.Contains(bwSlice[1], "Mbits") {
-				iperfBandwidth = iperfBandwidth / float64(1000)
-			}
-			t.Logf("Iperf bandwidth: %v", iperfBandwidth)
-			t.Logf("IPFIX record bandwidth: %v", recBandwidth)
-			// TODO: Make bandwidth test more robust.
-			assert.InDeltaf(t, recBandwidth, iperfBandwidth, 10, "Difference between Iperf bandwidth and IPFIX record bandwidth should be lower than 10")
-			break
+			value, err := strconv.ParseUint(strings.TrimSpace(lineSlice[1]), 10, 64)
+			require.NoError(t, err, "Error when converting %s to uint64 type", field)
+			return value
 		}
 	}
+	return 0
 }
 
 // TODO: Add a test that checks the functionality of Pod-To-External flow.
diff --git a/test/e2e/framework.go b/test/e2e/framework.go
index 807d1d301..c7fecbdef 100644
--- a/test/e2e/framework.go
+++ b/test/e2e/framework.go
@@ -105,6 +105,11 @@ const (
 	ipfixCollectorPort  = "4739"
 
 	nginxLBService = "nginx-loadbalancer"
+
+	exporterActiveFlowExportTimeout     = 2 * time.Second
+	exporterInactiveFlowExportTimeout   = 1 * time.Second
+	aggregatorActiveFlowRecordTimeout   = 3500 * time.Millisecond
+	aggregatorInactiveFlowRecordTimeout = 6 * time.Second
 )
 
 type ClusterNode struct {
@@ -563,8 +568,8 @@ func (data *TestData) deployAntreaFlowExporter(ipfixCollector string) error {
 	ac := []configChange{
 		{"FlowExporter", "true", true},
 		{"flowPollInterval", "\"1s\"", false},
-		{"activeFlowExportTimeout", "\"2s\"", false},
-		{"idleFlowExportTimeout", "\"1s\"", false},
+		{"activeFlowExportTimeout", fmt.Sprintf("\"%v\"", exporterActiveFlowExportTimeout), false},
+		{"inactiveFlowExportTimeout", fmt.Sprintf("\"%v\"", exporterInactiveFlowExportTimeout), false},
 	}
 	if ipfixCollector != "" {
 		ac = append(ac, configChange{"flowCollectorAddr", fmt.Sprintf("\"%s\"", ipfixCollector), false})
@@ -604,11 +609,8 @@ func (data *TestData) mutateFlowAggregatorConfigMap(ipfixCollector string, faClu
 	}
 	flowAggregatorConf, _ := configMap.Data[flowAggregatorConfName]
 	flowAggregatorConf = strings.Replace(flowAggregatorConf, "#externalFlowCollectorAddr: \"\"", fmt.Sprintf("externalFlowCollectorAddr: \"%s\"", ipfixCollector), 1)
-	// We expect at least two flow records at the external flow collector, so picked
-	// 4s and 6s for active and inactive flow timeouts, respectively, considering
-	// the fact that test flows run for 10s.
-	flowAggregatorConf = strings.Replace(flowAggregatorConf, "#activeFlowRecordTimeout: 60s", "activeFlowRecordTimeout: 4s", 1)
-	flowAggregatorConf = strings.Replace(flowAggregatorConf, "#inactiveFlowRecordTimeout: 90s", "inactiveFlowRecordTimeout: 6s", 1)
+	flowAggregatorConf = strings.Replace(flowAggregatorConf, "#activeFlowRecordTimeout: 60s", fmt.Sprintf("activeFlowRecordTimeout: %v", aggregatorActiveFlowRecordTimeout), 1)
+	flowAggregatorConf = strings.Replace(flowAggregatorConf, "#inactiveFlowRecordTimeout: 90s", fmt.Sprintf("inactiveFlowRecordTimeout: %v", aggregatorInactiveFlowRecordTimeout), 1)
 	if testOptions.providerName == "kind" {
 		// In Kind cluster, there are issues with DNS name resolution on worker nodes.
 		// We will use flow aggregator service cluster IP to generate the server certificate for tls communication
diff --git a/test/integration/agent/openflow_test.go b/test/integration/agent/openflow_test.go
index 897ccac79..74ff7e16a 100644
--- a/test/integration/agent/openflow_test.go
+++ b/test/integration/agent/openflow_test.go
@@ -290,7 +290,7 @@ func testInstallNodeFlows(t *testing.T, config *testConfig) {
 		peerConfig := map[*net.IPNet]net.IP{
 			&node.subnet: node.gateway,
 		}
-		err := c.InstallNodeFlows(node.name, peerConfig, node.nodeAddress, 0)
+		err := c.InstallNodeFlows(node.name, peerConfig, node.nodeAddress, 0, nil)
 		if err != nil {
 			t.Fatalf("Failed to install Openflow entries for node connectivity: %v", err)
 		}