Repository: antrea-io/antrea
Pull request #2209 (2021-05-26 18:46:00)

Fix typo in comments

codecov-commenter: # [Codecov](https://codecov.io/gh/antrea-io/antrea/pull/2209?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) Report
> Merging [#2209](https://codecov.io/gh/antrea-io/antrea/pull/2209?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) (33b05cd) into [main](https://codecov.io/gh/antrea-io/antrea/commit/cddfccd8a6059a616c1fa1a17da1319d2913bdc1?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) (cddfccd) will **decrease** coverage by `0.25%`.
> The diff coverage is `n/a`.

[![Impacted file tree graph](https://codecov.io/gh/antrea-io/antrea/pull/2209/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/2209?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None)

```diff
@@            Coverage Diff             @@
##             main    #2209      +/-   ##
==========================================
- Coverage   61.69%   61.44%   -0.26%     
==========================================
  Files         274      274              
  Lines       20673    20673              
==========================================
- Hits        12754    12702      -52     
- Misses       6587     6638      +51     
- Partials     1332     1333       +1     
```

| Flag | Coverage Δ | |
|---|---|---|
| kind-e2e-tests | `52.42% <ø> (-0.25%)` | :arrow_down: |
| unit-tests | `41.08% <ø> (-0.02%)` | :arrow_down: |

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/2209?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) | Coverage Δ | |
|---|---|---|
| [pkg/apiserver/certificate/cacert\_controller.go](https://codecov.io/gh/antrea-io/antrea/pull/2209/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2FwaXNlcnZlci9jZXJ0aWZpY2F0ZS9jYWNlcnRfY29udHJvbGxlci5nbw==) | `58.27% <0.00%> (-7.95%)` | :arrow_down: |
| [pkg/agent/controller/traceflow/packetin.go](https://codecov.io/gh/antrea-io/antrea/pull/2209/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2FnZW50L2NvbnRyb2xsZXIvdHJhY2VmbG93L3BhY2tldGluLmdv) | `61.57% <0.00%> (-5.10%)` | :arrow_down: |
| [...agent/controller/traceflow/traceflow\_controller.go](https://codecov.io/gh/antrea-io/antrea/pull/2209/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/apis/controlplane/v1beta2/sets.go](https://codecov.io/gh/antrea-io/antrea/pull/2209/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2FwaXMvY29udHJvbHBsYW5lL3YxYmV0YTIvc2V0cy5nbw==) | `66.66% <0.00%> (-2.57%)` | :arrow_down: |
| [pkg/controller/traceflow/controller.go](https://codecov.io/gh/antrea-io/antrea/pull/2209/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/apiserver/apiserver.go](https://codecov.io/gh/antrea-io/antrea/pull/2209/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2FwaXNlcnZlci9hcGlzZXJ2ZXIuZ28=) | `89.51% <0.00%> (-1.62%)` | :arrow_down: |
| [pkg/agent/openflow/pipeline.go](https://codecov.io/gh/antrea-io/antrea/pull/2209/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2FnZW50L29wZW5mbG93L3BpcGVsaW5lLmdv) | `70.11% <0.00%> (-1.04%)` | :arrow_down: |
| [pkg/agent/agent.go](https://codecov.io/gh/antrea-io/antrea/pull/2209/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2FnZW50L2FnZW50Lmdv) | `50.57% <0.00%> (ø)` | |

antoninbas: /skip-all
Dyanngg: Seems like there's a DCO warning though  https://github.com/antrea-io/antrea/pull/2209/checks?check_run_id=2678248324
tnqn: @antoninbas should we make DCO check required?
antoninbas: @tnqn just did, thanks for the reminder. I thought I had done it already...
abhiraut: /skip-all
abhiraut: thanks.. pushed this PR mainly to check what modifications are necessary while pushing to antrea.io .. seems like all commits now require to be signed off 

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/elk-flow-collector/logstash/filter.rb b/build/yamls/elk-flow-collector/logstash/filter.rb
index bc84b00e4..dcdaf85b9 100644
--- a/build/yamls/elk-flow-collector/logstash/filter.rb
+++ b/build/yamls/elk-flow-collector/logstash/filter.rb
@@ -31,6 +31,27 @@ def filter(event)
         event.set("[ipfix][flowTypeStr]", "From External")
     end
 
+    ingressRuleAction = event.get("[ipfix][ingressNetworkPolicyRuleAction]")
+    egressRuleAction = event.get("[ipfix][egressNetworkPolicyRuleAction]")
+    if ingressRuleAction == 0
+        event.set("[ipfix][ingressNetworkPolicyRuleActionStr]", "No Action")
+    elsif ingressRuleAction == 1
+        event.set("[ipfix][ingressNetworkPolicyRuleActionStr]", "Allow")
+    elsif ingressRuleAction == 2
+        event.set("[ipfix][ingressNetworkPolicyRuleActionStr]", "Drop")
+    elsif ingressRuleAction == 3
+        event.set("[ipfix][ingressNetworkPolicyRuleActionStr]", "Reject")
+    end
+    if egressRuleAction == 0
+        event.set("[ipfix][egressNetworkPolicyRuleActionStr]", "No Action")
+    elsif egressRuleAction == 1
+        event.set("[ipfix][egressNetworkPolicyRuleActionStr]", "Allow")
+    elsif egressRuleAction == 2
+        event.set("[ipfix][egressNetworkPolicyRuleActionStr]", "Drop")
+    elsif egressRuleAction == 3
+        event.set("[ipfix][egressNetworkPolicyRuleActionStr]", "Reject")
+    end
+
     if event.get("[ipfix][destinationIPv6Address]").nil?
         event.set("[ipfix][destinationIP]", event.get("[ipfix][destinationIPv4Address]"))
     else
diff --git a/build/yamls/elk-flow-collector/logstash/ipfix.yml b/build/yamls/elk-flow-collector/logstash/ipfix.yml
index 7f1caf94d..3de0d40b5 100644
--- a/build/yamls/elk-flow-collector/logstash/ipfix.yml
+++ b/build/yamls/elk-flow-collector/logstash/ipfix.yml
@@ -4124,3 +4124,9 @@
   137:
     - :uint8
     - :flowType
+  139:
+    - :uint8
+    - :ingressNetworkPolicyRuleAction
+  140:
+    - :uint8
+    - :egressNetworkPolicyRuleAction
diff --git a/build/yamls/flow-aggregator.yml b/build/yamls/flow-aggregator.yml
index 13f0f5d97..da71f070e 100644
--- a/build/yamls/flow-aggregator.yml
+++ b/build/yamls/flow-aggregator.yml
@@ -18,7 +18,7 @@ kind: Role
 metadata:
   labels:
     app: flow-aggregator
-  name: flow-aggregator-role
+  name: flow-exporter-role
   namespace: flow-aggregator
 rules:
 - apiGroups:
@@ -29,13 +29,6 @@ rules:
   - configmaps
   verbs:
   - get
-  - update
-- apiGroups:
-  - ""
-  resources:
-  - configmaps
-  verbs:
-  - create
 - apiGroups:
   - ""
   resourceNames:
@@ -44,21 +37,13 @@ rules:
   - secrets
   verbs:
   - get
-  - update
-- apiGroups:
-  - ""
-  resources:
-  - secrets
-  verbs:
-  - create
 ---
 apiVersion: rbac.authorization.k8s.io/v1
-kind: Role
+kind: ClusterRole
 metadata:
   labels:
     app: flow-aggregator
-  name: flow-exporter-role
-  namespace: flow-aggregator
+  name: flow-aggregator-role
 rules:
 - apiGroups:
   - ""
@@ -68,6 +53,21 @@ rules:
   - configmaps
   verbs:
   - get
+  - update
+- apiGroups:
+  - ""
+  resources:
+  - pods
+  verbs:
+  - get
+  - list
+  - watch
+- apiGroups:
+  - ""
+  resources:
+  - configmaps
+  verbs:
+  - create
 - apiGroups:
   - ""
   resourceNames:
@@ -76,49 +76,55 @@ rules:
   - secrets
   verbs:
   - get
+  - update
+- apiGroups:
+  - ""
+  resources:
+  - secrets
+  verbs:
+  - create
 ---
 apiVersion: rbac.authorization.k8s.io/v1
 kind: RoleBinding
 metadata:
   labels:
     app: flow-aggregator
-  name: flow-aggregator-role-binding
+  name: flow-exporter-role-binding
   namespace: flow-aggregator
 roleRef:
   apiGroup: rbac.authorization.k8s.io
   kind: Role
-  name: flow-aggregator-role
+  name: flow-exporter-role
 subjects:
 - kind: ServiceAccount
-  name: flow-aggregator
-  namespace: flow-aggregator
+  name: antrea-agent
+  namespace: kube-system
 ---
 apiVersion: rbac.authorization.k8s.io/v1
-kind: RoleBinding
+kind: ClusterRoleBinding
 metadata:
   labels:
     app: flow-aggregator
-  name: flow-exporter-role-binding
-  namespace: flow-aggregator
+  name: flow-aggregator-cluster-id-reader
 roleRef:
   apiGroup: rbac.authorization.k8s.io
-  kind: Role
-  name: flow-exporter-role
+  kind: ClusterRole
+  name: antrea-cluster-identity-reader
 subjects:
 - kind: ServiceAccount
-  name: antrea-agent
-  namespace: kube-system
+  name: flow-aggregator
+  namespace: flow-aggregator
 ---
 apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRoleBinding
 metadata:
   labels:
     app: flow-aggregator
-  name: flow-aggregator-cluster-id-reader
+  name: flow-aggregator-cluster-role-binding
 roleRef:
   apiGroup: rbac.authorization.k8s.io
   kind: ClusterRole
-  name: antrea-cluster-identity-reader
+  name: flow-aggregator-role
 subjects:
 - kind: ServiceAccount
   name: flow-aggregator
diff --git a/build/yamls/flow-aggregator/base/flow-aggregator.yml b/build/yamls/flow-aggregator/base/flow-aggregator.yml
index 52cdde2fb..55f2a080f 100644
--- a/build/yamls/flow-aggregator/base/flow-aggregator.yml
+++ b/build/yamls/flow-aggregator/base/flow-aggregator.yml
@@ -10,16 +10,18 @@ metadata:
   name: flow-aggregator
   namespace: flow-aggregator
 ---
-kind: Role
+kind: ClusterRole
 apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: flow-aggregator-role
-  namespace: flow-aggregator
 rules:
   - apiGroups: [""]
     resources: ["configmaps"]
     resourceNames: ["flow-aggregator-ca"]
     verbs: ["get", "update"]
+  - apiGroups: [""]
+    resources: ["pods"]
+    verbs: ["get", "list", "watch"]
   - apiGroups: [""]
     resources: ["configmaps"]
     verbs: ["create"]
@@ -31,17 +33,16 @@ rules:
     resources: ["secrets"]
     verbs: ["create"]
 ---
-kind: RoleBinding
+kind: ClusterRoleBinding
 apiVersion: rbac.authorization.k8s.io/v1
 metadata:
-  name: flow-aggregator-role-binding
-  namespace: flow-aggregator
+  name: flow-aggregator-cluster-role-binding
 subjects:
   - kind: ServiceAccount
     name: flow-aggregator
     namespace: flow-aggregator
 roleRef:
-  kind: Role
+  kind: ClusterRole
   name: flow-aggregator-role
   apiGroup: rbac.authorization.k8s.io
 ---
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/agent.go b/cmd/antrea-agent/agent.go
index d372e486b..904185587 100644
--- a/cmd/antrea-agent/agent.go
+++ b/cmd/antrea-agent/agent.go
@@ -101,7 +101,8 @@ func run(o *Options) error {
 	ofClient := openflow.NewClient(o.config.OVSBridge, ovsBridgeMgmtAddr, ovsDatapathType,
 		features.DefaultFeatureGate.Enabled(features.AntreaProxy),
 		features.DefaultFeatureGate.Enabled(features.AntreaPolicy),
-		features.DefaultFeatureGate.Enabled(features.Egress))
+		features.DefaultFeatureGate.Enabled(features.Egress),
+		features.DefaultFeatureGate.Enabled(features.FlowExporter))
 
 	_, serviceCIDRNet, _ := net.ParseCIDR(o.config.ServiceCIDR)
 	var serviceCIDRNetv6 *net.IPNet
@@ -163,6 +164,22 @@ func run(o *Options) error {
 		networkConfig,
 		nodeConfig)
 
+	var proxier proxy.Proxier
+	if features.DefaultFeatureGate.Enabled(features.AntreaProxy) {
+		v4Enabled := config.IsIPv4Enabled(nodeConfig, networkConfig.TrafficEncapMode)
+		v6Enabled := config.IsIPv6Enabled(nodeConfig, networkConfig.TrafficEncapMode)
+		switch {
+		case v4Enabled && v6Enabled:
+			proxier = proxy.NewDualStackProxier(nodeConfig.Name, informerFactory, ofClient)
+		case v4Enabled:
+			proxier = proxy.NewProxier(nodeConfig.Name, informerFactory, ofClient, false)
+		case v6Enabled:
+			proxier = proxy.NewProxier(nodeConfig.Name, informerFactory, ofClient, true)
+		default:
+			return fmt.Errorf("at least one of IPv4 or IPv6 should be enabled")
+		}
+	}
+
 	// entityUpdates is a channel for receiving entity updates from CNIServer and
 	// notifying NetworkPolicyController to reconcile rules related to the
 	// updated entities.
@@ -177,6 +194,11 @@ func run(o *Options) error {
 	// if AntreaPolicy feature is enabled.
 	statusManagerEnabled := antreaPolicyEnabled
 	loggingEnabled := antreaPolicyEnabled
+
+	var denyConnStore *connections.DenyConnectionStore
+	if features.DefaultFeatureGate.Enabled(features.FlowExporter) {
+		denyConnStore = connections.NewDenyConnectionStore(ifaceStore, proxier)
+	}
 	networkPolicyController, err := networkpolicy.NewNetworkPolicyController(
 		antreaClientProvider,
 		ofClient,
@@ -186,6 +208,7 @@ func run(o *Options) error {
 		antreaPolicyEnabled,
 		statusManagerEnabled,
 		loggingEnabled,
+		denyConnStore,
 		asyncRuleDeleteInterval)
 	if err != nil {
 		return fmt.Errorf("error creating new NetworkPolicy controller: %v", err)
@@ -203,22 +226,6 @@ func run(o *Options) error {
 		egressController = egress.NewEgressController(ofClient, egressInformer, antreaClientProvider, ifaceStore, routeClient, nodeConfig.Name)
 	}
 
-	var proxier proxy.Proxier
-	if features.DefaultFeatureGate.Enabled(features.AntreaProxy) {
-		v4Enabled := config.IsIPv4Enabled(nodeConfig, networkConfig.TrafficEncapMode)
-		v6Enabled := config.IsIPv6Enabled(nodeConfig, networkConfig.TrafficEncapMode)
-		switch {
-		case v4Enabled && v6Enabled:
-			proxier = proxy.NewDualStackProxier(nodeConfig.Name, informerFactory, ofClient)
-		case v4Enabled:
-			proxier = proxy.NewProxier(nodeConfig.Name, informerFactory, ofClient, false)
-		case v6Enabled:
-			proxier = proxy.NewProxier(nodeConfig.Name, informerFactory, ofClient, true)
-		default:
-			return fmt.Errorf("at least one of IPv4 or IPv6 should be enabled")
-		}
-	}
-
 	isChaining := false
 	if networkConfig.TrafficEncapMode.IsNetworkPolicyOnly() {
 		isChaining = true
@@ -357,7 +364,7 @@ func run(o *Options) error {
 		isNetworkPolicyOnly := networkConfig.TrafficEncapMode.IsNetworkPolicyOnly()
 
 		flowRecords := flowrecords.NewFlowRecords()
-		connStore := connections.NewConnectionStore(
+		conntrackConnStore := connections.NewConntrackConnectionStore(
 			connections.InitializeConnTrackDumper(nodeConfig, serviceCIDRNet, serviceCIDRNetv6, ovsDatapathType, features.DefaultFeatureGate.Enabled(features.AntreaProxy)),
 			flowRecords,
 			ifaceStore,
@@ -366,11 +373,12 @@ func run(o *Options) error {
 			proxier,
 			networkPolicyController,
 			o.pollInterval)
-		go connStore.Run(stopCh)
+		go conntrackConnStore.Run(stopCh)
 
 		flowExporter, err := exporter.NewFlowExporter(
-			connStore,
+			conntrackConnStore,
 			flowRecords,
+			denyConnStore,
 			o.flowCollectorAddr,
 			o.flowCollectorProto,
 			o.activeFlowTimeout,
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/cmd/flow-aggregator/flow-aggregator.go b/cmd/flow-aggregator/flow-aggregator.go
index e6210a3d7..143818254 100644
--- a/cmd/flow-aggregator/flow-aggregator.go
+++ b/cmd/flow-aggregator/flow-aggregator.go
@@ -21,6 +21,7 @@ import (
 
 	"github.com/google/uuid"
 	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/informers"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 	"k8s.io/klog/v2"
@@ -30,6 +31,8 @@ import (
 	"antrea.io/antrea/pkg/signals"
 )
 
+const informerDefaultResync = 12 * time.Hour
+
 // genObservationDomainID generates an IPFIX Observation Domain ID when one is not provided by the
 // user through the flow aggregator configuration. It will first try to generate one
 // deterministically based on the cluster UUID (if available, with a timeout of 10s). Otherwise, it
@@ -79,6 +82,9 @@ func run(o *Options) error {
 		return fmt.Errorf("error when creating K8s client: %v", err)
 	}
 
+	informerFactory := informers.NewSharedInformerFactory(k8sClient, informerDefaultResync)
+	podInformer := informerFactory.Core().V1().Pods()
+
 	var observationDomainID uint32
 	if o.config.ObservationDomainID != nil {
 		observationDomainID = *o.config.ObservationDomainID
@@ -96,6 +102,7 @@ func run(o *Options) error {
 		o.flowAggregatorAddress,
 		k8sClient,
 		observationDomainID,
+		podInformer,
 	)
 	err = flowAggregator.InitCollectingProcess()
 	if err != nil {
@@ -106,6 +113,9 @@ func run(o *Options) error {
 		return fmt.Errorf("error when creating aggregation process: %v", err)
 	}
 	go flowAggregator.Run(stopCh)
+
+	informerFactory.Start(stopCh)
+
 	<-stopCh
 	klog.Infof("Stopping flow aggregator")
 	return nil
diff --git a/docs/network-flow-visibility.md b/docs/network-flow-visibility.md
index 90da63ccb..037465bcd 100644
--- a/docs/network-flow-visibility.md
+++ b/docs/network-flow-visibility.md
@@ -163,24 +163,26 @@ the flow. All the IEs used by the Antrea Flow Exporter are listed below:
 
 #### IEs from Antrea IE Registry
 
-| IPFIX Information Element    | Enterprise ID | Field ID | Type        |
-|------------------------------|---------------|----------|-------------|
-| sourcePodNamespace           | 56506         | 100      | string      |
-| sourcePodName                | 56506         | 101      | string      |
-| destinationPodNamespace      | 56506         | 102      | string      |
-| destinationPodName           | 56506         | 103      | string      |
-| sourceNodeName               | 56506         | 104      | string      |
-| destinationNodeName          | 56506         | 105      | string      |
-| destinationClusterIPv4       | 56506         | 106      | ipv4Address |
-| destinationClusterIPv6       | 56506         | 107      | ipv6Address |
-| destinationServicePort       | 56506         | 108      | unsigned16  |
-| destinationServicePortName   | 56506         | 109      | string      |
-| ingressNetworkPolicyName     | 56506         | 110      | string      |
-| ingressNetworkPolicyNamespace| 56506         | 111      | string      |
-| egressNetworkPolicyName      | 56506         | 112      | string      |
-| egressNetworkPolicyNamespace | 56506         | 113      | string      |
-| tcpState                     | 56506         | 136      | string      |
-| flowType                     | 56506         | 137      | unsigned8   |
+| IPFIX Information Element     | Enterprise ID | Field ID | Type        |
+|-------------------------------|---------------|----------|-------------|
+| sourcePodNamespace            | 56506         | 100      | string      |
+| sourcePodName                 | 56506         | 101      | string      |
+| destinationPodNamespace       | 56506         | 102      | string      |
+| destinationPodName            | 56506         | 103      | string      |
+| sourceNodeName                | 56506         | 104      | string      |
+| destinationNodeName           | 56506         | 105      | string      |
+| destinationClusterIPv4        | 56506         | 106      | ipv4Address |
+| destinationClusterIPv6        | 56506         | 107      | ipv6Address |
+| destinationServicePort        | 56506         | 108      | unsigned16  |
+| destinationServicePortName    | 56506         | 109      | string      |
+| ingressNetworkPolicyName      | 56506         | 110      | string      |
+| ingressNetworkPolicyNamespace | 56506         | 111      | string      |
+| egressNetworkPolicyName       | 56506         | 112      | string      |
+| egressNetworkPolicyNamespace  | 56506         | 113      | string      |
+| tcpState                      | 56506         | 136      | string      |
+| flowType                      | 56506         | 137      | unsigned8   |
+| ingressNetworkPolicyRuleAction| 56506         | 139      | unsigned8   |
+| egressNetworkPolicyRuleAction | 56506         | 140      | unsigned8   |
 
 ### Supported capabilities
 
@@ -195,11 +197,15 @@ starting with Antrea v0.11. In the future, we will enable the support for Extern
 flows.
 
 Kubernetes information such as Node name, Pod name, Pod Namespace, Service name,
-NetworkPolicy name and NetworkPolicy Namespace, is added to the flow records. For
-flow records that are exported from any given Antrea Agent, the Flow Exporter only
-provides the information of Kubernetes entities that are local to the Antrea Agent.
-In other words, flow records are only complete for intra-Node flows, but incomplete
-for inter-Node flows. It is the responsibility of the [Flow Aggregator](#flow-aggregator)
+NetworkPolicy name and NetworkPolicy Namespace, is added to the flow records.
+Network Policy Rule Action (Allow, Reject, Drop) is also supported for both
+Antrea-native NetworkPolicies and K8s NetworkPolicies. For K8s NetworkPolicies,
+connections dropped due to [isolated Pod behavior](https://kubernetes.io/docs/concepts/services-networking/network-policies/#isolated-and-non-isolated-pods)
+will be assigned the Drop action.
+For flow records that are exported from any given Antrea Agent, the Flow Exporter
+only provides the information of Kubernetes entities that are local to the Antrea
+Agent. In other words, flow records are only complete for intra-Node flows, but
+incomplete for inter-Node flows. It is the responsibility of the [Flow Aggregator](#flow-aggregator)
 to correlate flows from the source and destination Nodes and produce complete flow
 records.
 
@@ -210,7 +216,8 @@ Both Flow Exporter and Flow Aggregator are supported in IPv4 clusters, IPv6 clus
 We support following connection metrics as Prometheus metrics that are exposed
 through [Antrea Agent apiserver endpoint](prometheus-integration.md):
 `antrea_agent_conntrack_total_connection_count`,
-`antrea_agent_conntrack_antrea_connection_count` and
+`antrea_agent_conntrack_antrea_connection_count`,
+`antrea_agent_denied_connection_count` and
 `antrea_agent_conntrack_max_connection_count`
 
 ## Flow Aggregator
diff --git a/docs/prometheus-integration.md b/docs/prometheus-integration.md
index 5e68958a3..a72397029 100644
--- a/docs/prometheus-integration.md
+++ b/docs/prometheus-integration.md
@@ -151,6 +151,9 @@ a configuration parameter for the Agent.
 - **antrea_agent_conntrack_total_connection_count:** Number of connections
 in the conntrack table. This metric gets updated at an interval specified
 by flowPollInterval, a configuration parameter for the Agent.
+- **antrea_agent_denied_connection_count:** Number of denied connections
+detected by Flow Exporter deny connections tracking. This metric gets updated
+when a flow is rejected/dropped by network policy.
 - **antrea_agent_egress_networkpolicy_rule_count:** Number of egress
 NetworkPolicy rules on local Node which are managed by the Antrea Agent.
 - **antrea_agent_ingress_networkpolicy_rule_count:** Number of ingress
diff --git a/hack/update-codegen-dockerized.sh b/hack/update-codegen-dockerized.sh
index 85f6bb3b4..d091a5e3f 100755
--- a/hack/update-codegen-dockerized.sh
+++ b/hack/update-codegen-dockerized.sh
@@ -160,7 +160,7 @@ $GOPATH/bin/deepcopy-gen \
 # Generate mocks for testing with mockgen.
 MOCKGEN_TARGETS=(
   "pkg/agent/cniserver/ipam IPAMDriver testing"
-  "pkg/agent/flowexporter/connections ConnectionStore,ConnTrackDumper,NetFilterConnTrack testing"
+  "pkg/agent/flowexporter/connections ConnTrackDumper,NetFilterConnTrack testing"
   "pkg/agent/interfacestore InterfaceStore testing"
   "pkg/agent/nodeportlocal/rules PodPortRules testing"
   "pkg/agent/openflow Client,OFEntryOperations testing"
diff --git a/pkg/agent/controller/networkpolicy/networkpolicy_controller.go b/pkg/agent/controller/networkpolicy/networkpolicy_controller.go
index 9330fedf7..d29cd7069 100644
--- a/pkg/agent/controller/networkpolicy/networkpolicy_controller.go
+++ b/pkg/agent/controller/networkpolicy/networkpolicy_controller.go
@@ -30,6 +30,7 @@ import (
 	"k8s.io/klog/v2"
 
 	"antrea.io/antrea/pkg/agent"
+	"antrea.io/antrea/pkg/agent/flowexporter/connections"
 	"antrea.io/antrea/pkg/agent/interfacestore"
 	"antrea.io/antrea/pkg/agent/openflow"
 	"antrea.io/antrea/pkg/agent/types"
@@ -91,6 +92,8 @@ type Controller struct {
 	addressGroupWatcher   *watcher
 	fullSyncGroup         sync.WaitGroup
 	ifaceStore            interfacestore.InterfaceStore
+	// denyConnStore is for storing deny connections for flow exporter.
+	denyConnStore *connections.DenyConnectionStore
 }
 
 // NewNetworkPolicyController returns a new *Controller.
@@ -102,6 +105,7 @@ func NewNetworkPolicyController(antreaClientGetter agent.AntreaClientProvider,
 	antreaPolicyEnabled bool,
 	statusManagerEnabled bool,
 	loggingEnabled bool,
+	denyConnStore *connections.DenyConnectionStore,
 	asyncRuleDeleteInterval time.Duration) (*Controller, error) {
 	c := &Controller{
 		antreaClientProvider: antreaClientGetter,
@@ -111,6 +115,7 @@ func NewNetworkPolicyController(antreaClientGetter agent.AntreaClientProvider,
 		antreaPolicyEnabled:  antreaPolicyEnabled,
 		statusManagerEnabled: statusManagerEnabled,
 		loggingEnabled:       loggingEnabled,
+		denyConnStore:        denyConnStore,
 	}
 	c.ruleCache = newRuleCache(c.enqueueRule, entityUpdates)
 	if statusManagerEnabled {
diff --git a/pkg/agent/controller/networkpolicy/packetin.go b/pkg/agent/controller/networkpolicy/packetin.go
index 86b3259a3..a2c77100e 100644
--- a/pkg/agent/controller/networkpolicy/packetin.go
+++ b/pkg/agent/controller/networkpolicy/packetin.go
@@ -21,6 +21,7 @@ import (
 	"log"
 	"os"
 	"path/filepath"
+	"time"
 
 	"github.com/contiv/libOpenflow/openflow13"
 	"github.com/contiv/libOpenflow/protocol"
@@ -29,6 +30,7 @@ import (
 	"k8s.io/klog/v2"
 
 	"antrea.io/antrea/pkg/agent/config"
+	"antrea.io/antrea/pkg/agent/flowexporter"
 	"antrea.io/antrea/pkg/agent/openflow"
 	binding "antrea.io/antrea/pkg/ovs/openflow"
 	"antrea.io/antrea/pkg/util/ip"
@@ -119,7 +121,11 @@ func (c *Controller) HandlePacketIn(pktIn *ofctrl.PacketIn) error {
 			return err
 		}
 	}
-
+	if customReasons&openflow.CustomReasonDeny == openflow.CustomReasonDeny {
+		if err := c.storeDenyConnection(pktIn); err != nil {
+			return err
+		}
+	}
 	return nil
 }
 
@@ -327,3 +333,81 @@ func (c *Controller) rejectRequest(pktIn *ofctrl.PacketIn) error {
 			true)
 	}
 }
+
+func (c *Controller) storeDenyConnection(pktIn *ofctrl.PacketIn) error {
+	packet, err := binding.ParsePacketIn(pktIn)
+	if err != nil {
+		return fmt.Errorf("error in parsing packetin: %v", err)
+	}
+
+	// Get 5-tuple information
+	tuple := flowexporter.Tuple{
+		SourcePort:      packet.SourcePort,
+		DestinationPort: packet.DestinationPort,
+		Protocol:        packet.IPProto,
+	}
+	if packet.IsIPv6 {
+		tuple.SourceAddress = packet.SourceIP.To16()
+		tuple.DestinationAddress = packet.DestinationIP.To16()
+	} else {
+		tuple.SourceAddress = packet.SourceIP.To4()
+		tuple.DestinationAddress = packet.DestinationIP.To4()
+	}
+
+	// Generate deny connection and add to deny connection store
+	denyConn := flowexporter.Connection{}
+	denyConn.FlowKey = tuple
+	denyConn.DestinationServiceAddress = tuple.DestinationAddress
+	denyConn.DestinationServicePort = tuple.DestinationPort
+
+	// No need to obtain connection info again if it already exists in denyConnectionStore.
+	if conn, exist := c.denyConnStore.GetConnByKey(flowexporter.NewConnectionKey(&denyConn)); exist {
+		c.denyConnStore.AddOrUpdateConn(conn, time.Now(), uint64(packet.IPLength))
+		return nil
+	}
+
+	matchers := pktIn.GetMatches()
+	var match *ofctrl.MatchField
+	// Get table ID
+	tableID := binding.TableIDType(pktIn.TableId)
+	// Get disposition Allow, Drop or Reject
+	match = getMatchRegField(matchers, uint32(openflow.DispositionMarkReg))
+	id, err := getInfoInReg(match, openflow.APDispositionMarkRange.ToNXRange())
+	if err != nil {
+		return fmt.Errorf("error when getting disposition from reg: %v", err)
+	}
+	disposition := openflow.DispositionToString[id]
+
+	// For K8s NetworkPolicy implicit drop action, we cannot get name/namespace.
+	if tableID == openflow.IngressDefaultTable {
+		denyConn.IngressNetworkPolicyRuleAction = flowexporter.RuleActionToUint8(disposition)
+	} else if tableID == openflow.EgressDefaultTable {
+		denyConn.EgressNetworkPolicyRuleAction = flowexporter.RuleActionToUint8(disposition)
+	} else { // Get name and namespace for Antrea Network Policy or Antrea Cluster Network Policy
+		// Set match to corresponding ingress/egress reg according to disposition
+		match = getMatch(matchers, tableID, id)
+		ruleID, err := getInfoInReg(match, nil)
+		if err != nil {
+			return fmt.Errorf("error when obtaining rule id from reg: %v", err)
+		}
+		policy := c.GetNetworkPolicyByRuleFlowID(ruleID)
+
+		if policy == nil {
+			// Default drop by K8s NetworkPolicy
+			klog.V(4).Infof("Cannot find NetworkPolicy that has ruleID %v", ruleID)
+		} else {
+			if tableID == openflow.AntreaPolicyIngressRuleTable {
+				denyConn.IngressNetworkPolicyName = policy.Name
+				denyConn.IngressNetworkPolicyNamespace = policy.Namespace
+				denyConn.IngressNetworkPolicyRuleAction = flowexporter.RuleActionToUint8(disposition)
+			} else if tableID == openflow.AntreaPolicyEgressRuleTable {
+				denyConn.EgressNetworkPolicyName = policy.Name
+				denyConn.EgressNetworkPolicyNamespace = policy.Namespace
+				denyConn.EgressNetworkPolicyRuleAction = flowexporter.RuleActionToUint8(disposition)
+			}
+		}
+	}
+
+	c.denyConnStore.AddOrUpdateConn(&denyConn, time.Now(), uint64(packet.IPLength))
+	return nil
+}
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/flowexporter/connections/connections.go b/pkg/agent/flowexporter/connections/connections.go
index 602361d9d..3afc17e04 100644
--- a/pkg/agent/flowexporter/connections/connections.go
+++ b/pkg/agent/flowexporter/connections/connections.go
@@ -15,197 +15,32 @@
 package connections
 
 import (
-	"encoding/binary"
 	"fmt"
 	"sync"
-	"time"
 
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/klog/v2"
 
 	"antrea.io/antrea/pkg/agent/flowexporter"
-	"antrea.io/antrea/pkg/agent/flowexporter/flowrecords"
 	"antrea.io/antrea/pkg/agent/interfacestore"
-	"antrea.io/antrea/pkg/agent/metrics"
-	"antrea.io/antrea/pkg/agent/openflow"
 	"antrea.io/antrea/pkg/agent/proxy"
-	"antrea.io/antrea/pkg/querier"
 )
 
-var serviceProtocolMap = map[uint8]corev1.Protocol{
-	6:   corev1.ProtocolTCP,
-	17:  corev1.ProtocolUDP,
-	132: corev1.ProtocolSCTP,
-}
-
-type ConnectionStore interface {
-	// Run enables the periodical polling of conntrack connections at a given flowPollInterval.
-	Run(stopCh <-chan struct{})
-	// GetConnByKey gets the connection in connection map given the connection key.
-	GetConnByKey(connKey flowexporter.ConnectionKey) (*flowexporter.Connection, bool)
-	// SetExportDone sets DoneExport field of connection to true given the connection key.
-	SetExportDone(connKey flowexporter.ConnectionKey) error
-	// ForAllConnectionsDo execute the callback for each connection in connection map.
-	ForAllConnectionsDo(callback flowexporter.ConnectionMapCallBack) error
-}
-
 type connectionStore struct {
-	connections          map[flowexporter.ConnectionKey]*flowexporter.Connection
-	flowRecords          *flowrecords.FlowRecords
-	connDumper           ConnTrackDumper
-	ifaceStore           interfacestore.InterfaceStore
-	v4Enabled            bool
-	v6Enabled            bool
-	antreaProxier        proxy.Proxier
-	networkPolicyQuerier querier.AgentNetworkPolicyInfoQuerier
-	pollInterval         time.Duration
-	mutex                sync.Mutex
+	connections   map[flowexporter.ConnectionKey]*flowexporter.Connection
+	ifaceStore    interfacestore.InterfaceStore
+	antreaProxier proxy.Proxier
+	mutex         sync.Mutex
 }
 
 func NewConnectionStore(
-	connTrackDumper ConnTrackDumper,
-	flowRecords *flowrecords.FlowRecords,
 	ifaceStore interfacestore.InterfaceStore,
-	v4Enabled bool,
-	v6Enabled bool,
 	proxier proxy.Proxier,
-	npQuerier querier.AgentNetworkPolicyInfoQuerier,
-	pollInterval time.Duration,
-) *connectionStore {
-	return &connectionStore{
-		connections:          make(map[flowexporter.ConnectionKey]*flowexporter.Connection),
-		flowRecords:          flowRecords,
-		connDumper:           connTrackDumper,
-		ifaceStore:           ifaceStore,
-		v4Enabled:            v4Enabled,
-		v6Enabled:            v6Enabled,
-		antreaProxier:        proxier,
-		networkPolicyQuerier: npQuerier,
-		pollInterval:         pollInterval,
-	}
-}
-
-// Run enables the periodical polling of conntrack connections at a given flowPollInterval.
-func (cs *connectionStore) Run(stopCh <-chan struct{}) {
-	klog.Infof("Starting conntrack polling")
-
-	pollTicker := time.NewTicker(cs.pollInterval)
-	defer pollTicker.Stop()
-
-	for {
-		select {
-		case <-stopCh:
-			break
-		case <-pollTicker.C:
-			_, err := cs.Poll()
-			if err != nil {
-				// Not failing here as errors can be transient and could be resolved in future poll cycles.
-				// TODO: Come up with a backoff/retry mechanism by increasing poll interval and adding retry timeout
-				klog.Errorf("Error during conntrack poll cycle: %v", err)
-			}
-			// AddOrUpdateFlowRecord method does not return any error, hence no error handling required.
-			cs.ForAllConnectionsDo(cs.flowRecords.AddOrUpdateFlowRecord)
-			klog.V(2).Infof("Flow records are successfully updated")
-		}
-	}
-}
-
-// addOrUpdateConn updates the connection if it is already present, i.e., update timestamp, counters etc.,
-// or adds a new Connection by 5-tuple of the flow along with local Pod and PodNameSpace.
-func (cs *connectionStore) addOrUpdateConn(conn *flowexporter.Connection) {
-	connKey := flowexporter.NewConnectionKey(conn)
-
-	cs.mutex.Lock()
-	defer cs.mutex.Unlock()
-	existingConn, exists := cs.connections[connKey]
-
-	if exists {
-		existingConn.IsPresent = true
-		// avoid updating stats of the existing connection that is about to close
-		if flowexporter.IsConnectionDying(existingConn) {
-			return
-		}
-		// Update the necessary fields that are used in generating flow records.
-		// Can same 5-tuple flow get deleted and added to conntrack table? If so use ID.
-		existingConn.StopTime = conn.StopTime
-		existingConn.OriginalBytes = conn.OriginalBytes
-		existingConn.OriginalPackets = conn.OriginalPackets
-		existingConn.ReverseBytes = conn.ReverseBytes
-		existingConn.ReversePackets = conn.ReversePackets
-		existingConn.TCPState = conn.TCPState
-		klog.V(4).Infof("Antrea flow updated: %v", existingConn)
-	} else {
-		// sourceIP/destinationIP are mapped only to local pods and not remote pods.
-		sIface, srcFound := cs.ifaceStore.GetInterfaceByIP(conn.TupleOrig.SourceAddress.String())
-		dIface, dstFound := cs.ifaceStore.GetInterfaceByIP(conn.TupleReply.SourceAddress.String())
-		if !srcFound && !dstFound {
-			klog.Warningf("Cannot map any of the IP %s or %s to a local Pod", conn.TupleOrig.SourceAddress.String(), conn.TupleReply.SourceAddress.String())
-		}
-		if srcFound && sIface.Type == interfacestore.ContainerInterface {
-			conn.SourcePodName = sIface.ContainerInterfaceConfig.PodName
-			conn.SourcePodNamespace = sIface.ContainerInterfaceConfig.PodNamespace
-		}
-		if dstFound && dIface.Type == interfacestore.ContainerInterface {
-			conn.DestinationPodName = dIface.ContainerInterfaceConfig.PodName
-			conn.DestinationPodNamespace = dIface.ContainerInterfaceConfig.PodNamespace
-		}
-
-		// Process Pod-to-Service flows when Antrea Proxy is enabled.
-		if cs.antreaProxier != nil {
-			if conn.Mark == openflow.ServiceCTMark {
-				clusterIP := conn.TupleOrig.DestinationAddress.String()
-				svcPort := conn.TupleOrig.DestinationPort
-				protocol, err := lookupServiceProtocol(conn.TupleOrig.Protocol)
-				if err != nil {
-					klog.Warningf("Could not retrieve Service protocol: %v", err)
-				} else {
-					serviceStr := fmt.Sprintf("%s:%d/%s", clusterIP, svcPort, protocol)
-					servicePortName, exists := cs.antreaProxier.GetServiceByIP(serviceStr)
-					if !exists {
-						klog.Warningf("Could not retrieve the Service info from antrea-agent-proxier for the serviceStr: %s", serviceStr)
-					} else {
-						conn.DestinationServicePortName = servicePortName.String()
-					}
-				}
-			}
-		}
-
-		// Retrieve NetworkPolicy Name and Namespace by using the ingress and egress
-		// IDs stored in the connection label.
-		if len(conn.Labels) != 0 {
-			klog.V(4).Infof("connection label: %x; label masks: %x", conn.Labels, conn.LabelsMask)
-			// We always expect labels from conntrack dumper to be added in little-endian format right now
-			// In kernel datapath, the labels uses the "native" endianness for the system, which are little-endian
-			// on most of the modern CPUs based on x86 architecture like Intel, AMD, etc.
-			ingressOfID := binary.LittleEndian.Uint32(conn.Labels[:4])
-			egressOfID := binary.LittleEndian.Uint32(conn.Labels[4:8])
-			if ingressOfID != 0 {
-				policy := cs.networkPolicyQuerier.GetNetworkPolicyByRuleFlowID(ingressOfID)
-				if policy == nil {
-					// This should not happen because the rule flow ID to rule mapping is
-					// preserved for max(5s, flowPollInterval) even after the rule deletion.
-					klog.Warningf("Cannot find NetworkPolicy that has rule with ingressOfID %v", ingressOfID)
-				} else {
-					conn.IngressNetworkPolicyName = policy.Name
-					conn.IngressNetworkPolicyNamespace = policy.Namespace
-				}
-			}
-			if egressOfID != 0 {
-				policy := cs.networkPolicyQuerier.GetNetworkPolicyByRuleFlowID(egressOfID)
-				if policy == nil {
-					// This should not happen because the rule flow ID to rule mapping is
-					// preserved for max(5s, flowPollInterval) even after the rule deletion.
-					klog.Warningf("Cannot find NetworkPolicy that has rule with egressOfID %v", egressOfID)
-				} else {
-					conn.EgressNetworkPolicyName = policy.Name
-					conn.EgressNetworkPolicyNamespace = policy.Namespace
-				}
-			}
-		}
-		metrics.TotalAntreaConnectionsInConnTrackTable.Inc()
-		klog.V(4).Infof("New Antrea flow added: %v", conn)
-		// Add new antrea connection to connection store
-		cs.connections[connKey] = conn
+) connectionStore {
+	return connectionStore{
+		connections:   make(map[flowexporter.ConnectionKey]*flowexporter.Connection),
+		ifaceStore:    ifaceStore,
+		antreaProxier: proxier,
 	}
 }
 
@@ -217,18 +52,6 @@ func (cs *connectionStore) GetConnByKey(connKey flowexporter.ConnectionKey) (*fl
 	return conn, found
 }
 
-// SetExportDone sets DoneExport field of connection to true given the connection key.
-func (cs *connectionStore) SetExportDone(connKey flowexporter.ConnectionKey) error {
-	cs.mutex.Lock()
-	defer cs.mutex.Unlock()
-	if conn, found := cs.connections[connKey]; !found {
-		return fmt.Errorf("connection with key %v does not exist in connection map", connKey)
-	} else {
-		conn.DoneExport = true
-		return nil
-	}
-}
-
 // ForAllConnectionsDo execute the callback for each connection in connection map.
 func (cs *connectionStore) ForAllConnectionsDo(callback flowexporter.ConnectionMapCallBack) error {
 	cs.mutex.Lock()
@@ -243,71 +66,40 @@ func (cs *connectionStore) ForAllConnectionsDo(callback flowexporter.ConnectionM
 	return nil
 }
 
-// Poll calls into conntrackDumper interface to dump conntrack flows. It returns the number of connections for each
-// address family, as a slice. In dual-stack clusters, the slice will contain 2 values (number of IPv4 connections first,
-// then number of IPv6 connections).
-// TODO: As optimization, only poll invalid/closed connections during every poll, and poll the established connections right before the export.
-func (cs *connectionStore) Poll() ([]int, error) {
-	klog.V(2).Infof("Polling conntrack")
-	// Reset IsPresent flag for all connections in connection map before dumping flows in conntrack module.
-	// if the connection does not exist in conntrack table and has been exported, we will delete it from connection map.
-	deleteIfStaleOrResetConn := func(key flowexporter.ConnectionKey, conn *flowexporter.Connection) error {
-		if !conn.IsPresent && conn.DoneExport {
-			if err := cs.deleteConnectionByKeyWithoutLock(key); err != nil {
-				return err
-			}
-		} else {
-			conn.IsPresent = false
-		}
-		return nil
+func (cs *connectionStore) fillPodInfo(conn *flowexporter.Connection) {
+	if cs.ifaceStore == nil {
+		klog.V(4).Info("Interface store is not available to retrieve local Pods information.")
+		return
 	}
-
-	if err := cs.ForAllConnectionsDo(deleteIfStaleOrResetConn); err != nil {
-		return []int{}, err
+	// sourceIP/destinationIP are mapped only to local pods and not remote pods.
+	srcIP := conn.FlowKey.SourceAddress.String()
+	dstIP := conn.FlowKey.DestinationAddress.String()
+
+	sIface, srcFound := cs.ifaceStore.GetInterfaceByIP(srcIP)
+	dIface, dstFound := cs.ifaceStore.GetInterfaceByIP(dstIP)
+	if !srcFound && !dstFound {
+		klog.Warningf("Cannot map any of the IP %s or %s to a local Pod", srcIP, dstIP)
 	}
-
-	var zones []uint16
-	var connsLens []int
-	if cs.v4Enabled {
-		zones = append(zones, openflow.CtZone)
-	}
-	if cs.v6Enabled {
-		zones = append(zones, openflow.CtZoneV6)
-	}
-	var totalConns int
-	for _, zone := range zones {
-		filteredConnsList, totalConnsPerZone, err := cs.connDumper.DumpFlows(zone)
-		if err != nil {
-			return []int{}, err
-		}
-		totalConns += totalConnsPerZone
-		// Update only the Connection store. IPFIX records are generated based on Connection store.
-		for _, conn := range filteredConnsList {
-			cs.addOrUpdateConn(conn)
-		}
-		connsLens = append(connsLens, len(filteredConnsList))
+	if srcFound && sIface.Type == interfacestore.ContainerInterface {
+		conn.SourcePodName = sIface.ContainerInterfaceConfig.PodName
+		conn.SourcePodNamespace = sIface.ContainerInterfaceConfig.PodNamespace
 	}
-	metrics.TotalConnectionsInConnTrackTable.Set(float64(totalConns))
-	maxConns, err := cs.connDumper.GetMaxConnections()
-	if err != nil {
-		return []int{}, err
+	if dstFound && dIface.Type == interfacestore.ContainerInterface {
+		conn.DestinationPodName = dIface.ContainerInterfaceConfig.PodName
+		conn.DestinationPodNamespace = dIface.ContainerInterfaceConfig.PodNamespace
 	}
-	metrics.MaxConnectionsInConnTrackTable.Set(float64(maxConns))
-	klog.V(2).Infof("Conntrack polling successful")
-
-	return connsLens, nil
 }
 
-// deleteConnectionByKeyWithoutLock deletes the connection in connection map given the
-// connection key without grabbing the lock. Caller is expected to grab lock.
-func (cs *connectionStore) deleteConnectionByKeyWithoutLock(connKey flowexporter.ConnectionKey) error {
-	_, exists := cs.connections[connKey]
-	if !exists {
-		return fmt.Errorf("connection with key %v doesn't exist in map", connKey)
+func (cs *connectionStore) fillServiceInfo(conn *flowexporter.Connection, serviceStr string) {
+	// resolve destination Service information
+	if cs.antreaProxier != nil {
+		servicePortName, exists := cs.antreaProxier.GetServiceByIP(serviceStr)
+		if !exists {
+			klog.Warningf("Could not retrieve the Service info from antrea-agent-proxier for the serviceStr: %s", serviceStr)
+		} else {
+			conn.DestinationServicePortName = servicePortName.String()
+		}
 	}
-	delete(cs.connections, connKey)
-	metrics.TotalAntreaConnectionsInConnTrackTable.Dec()
-	return nil
 }
 
 // LookupServiceProtocol returns the corresponding Service protocol string for a given protocol identifier
diff --git a/pkg/agent/flowexporter/connections/conntrack.go b/pkg/agent/flowexporter/connections/conntrack.go
index 32e410c77..ef8cbf982 100644
--- a/pkg/agent/flowexporter/connections/conntrack.go
+++ b/pkg/agent/flowexporter/connections/conntrack.go
@@ -41,8 +41,8 @@ func filterAntreaConns(conns []*flowexporter.Connection, nodeConfig *config.Node
 		if conn.Zone != zoneFilter {
 			continue
 		}
-		srcIP := conn.TupleOrig.SourceAddress
-		dstIP := conn.TupleReply.SourceAddress
+		srcIP := conn.FlowKey.SourceAddress
+		dstIP := conn.FlowKey.DestinationAddress
 
 		// Consider Pod-to-Pod, Pod-To-Service and Pod-To-External flows.
 		if srcIP.Equal(nodeConfig.GatewayConfig.IPv4) || dstIP.Equal(nodeConfig.GatewayConfig.IPv4) {
diff --git a/pkg/agent/flowexporter/connections/conntrack_connections.go b/pkg/agent/flowexporter/connections/conntrack_connections.go
new file mode 100644
index 000000000..7db68deee
--- /dev/null
+++ b/pkg/agent/flowexporter/connections/conntrack_connections.go
@@ -0,0 +1,254 @@
+// Copyright 2021 Antrea Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package connections
+
+import (
+	"encoding/binary"
+	"fmt"
+	"time"
+
+	"github.com/vmware/go-ipfix/pkg/registry"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/klog/v2"
+
+	"antrea.io/antrea/pkg/agent/flowexporter"
+	"antrea.io/antrea/pkg/agent/flowexporter/flowrecords"
+	"antrea.io/antrea/pkg/agent/interfacestore"
+	"antrea.io/antrea/pkg/agent/metrics"
+	"antrea.io/antrea/pkg/agent/openflow"
+	"antrea.io/antrea/pkg/agent/proxy"
+	"antrea.io/antrea/pkg/querier"
+)
+
+var serviceProtocolMap = map[uint8]corev1.Protocol{
+	6:   corev1.ProtocolTCP,
+	17:  corev1.ProtocolUDP,
+	132: corev1.ProtocolSCTP,
+}
+
+type ConntrackConnectionStore struct {
+	flowRecords          *flowrecords.FlowRecords
+	connDumper           ConnTrackDumper
+	v4Enabled            bool
+	v6Enabled            bool
+	networkPolicyQuerier querier.AgentNetworkPolicyInfoQuerier
+	pollInterval         time.Duration
+	connectionStore
+}
+
+func NewConntrackConnectionStore(
+	connTrackDumper ConnTrackDumper,
+	flowRecords *flowrecords.FlowRecords,
+	ifaceStore interfacestore.InterfaceStore,
+	v4Enabled bool,
+	v6Enabled bool,
+	proxier proxy.Proxier,
+	npQuerier querier.AgentNetworkPolicyInfoQuerier,
+	pollInterval time.Duration,
+) *ConntrackConnectionStore {
+	return &ConntrackConnectionStore{
+		flowRecords:          flowRecords,
+		connDumper:           connTrackDumper,
+		v4Enabled:            v4Enabled,
+		v6Enabled:            v6Enabled,
+		networkPolicyQuerier: npQuerier,
+		pollInterval:         pollInterval,
+		connectionStore:      NewConnectionStore(ifaceStore, proxier),
+	}
+}
+
+// Run enables the periodical polling of conntrack connections at a given flowPollInterval.
+func (cs *ConntrackConnectionStore) Run(stopCh <-chan struct{}) {
+	klog.Infof("Starting conntrack polling")
+
+	pollTicker := time.NewTicker(cs.pollInterval)
+	defer pollTicker.Stop()
+
+	for {
+		select {
+		case <-stopCh:
+			break
+		case <-pollTicker.C:
+			_, err := cs.Poll()
+			if err != nil {
+				// Not failing here as errors can be transient and could be resolved in future poll cycles.
+				// TODO: Come up with a backoff/retry mechanism by increasing poll interval and adding retry timeout
+				klog.Errorf("Error during conntrack poll cycle: %v", err)
+			}
+			// AddOrUpdateFlowRecord method does not return any error, hence no error handling required.
+			cs.ForAllConnectionsDo(cs.flowRecords.AddOrUpdateFlowRecord)
+			klog.V(2).Infof("Flow records are successfully updated")
+		}
+	}
+}
+
+// Poll calls into conntrackDumper interface to dump conntrack flows. It returns the number of connections for each
+// address family, as a slice. In dual-stack clusters, the slice will contain 2 values (number of IPv4 connections first,
+// then number of IPv6 connections).
+// TODO: As optimization, only poll invalid/closed connections during every poll, and poll the established connections right before the export.
+func (cs *ConntrackConnectionStore) Poll() ([]int, error) {
+	klog.V(2).Infof("Polling conntrack")
+	// Reset IsPresent flag for all connections in connection map before dumping flows in conntrack module.
+	// if the connection does not exist in conntrack table and has been exported, we will delete it from connection map.
+	deleteIfStaleOrResetConn := func(key flowexporter.ConnectionKey, conn *flowexporter.Connection) error {
+		if !conn.IsPresent && conn.DoneExport {
+			if err := cs.DeleteConnWithoutLock(key); err != nil {
+				return err
+			}
+		} else {
+			conn.IsPresent = false
+		}
+		return nil
+	}
+
+	if err := cs.ForAllConnectionsDo(deleteIfStaleOrResetConn); err != nil {
+		return []int{}, err
+	}
+
+	var zones []uint16
+	var connsLens []int
+	if cs.v4Enabled {
+		zones = append(zones, openflow.CtZone)
+	}
+	if cs.v6Enabled {
+		zones = append(zones, openflow.CtZoneV6)
+	}
+	var totalConns int
+	for _, zone := range zones {
+		filteredConnsList, totalConnsPerZone, err := cs.connDumper.DumpFlows(zone)
+		if err != nil {
+			return []int{}, err
+		}
+		totalConns += totalConnsPerZone
+		// Update only the Connection store. IPFIX records are generated based on Connection store.
+		for _, conn := range filteredConnsList {
+			cs.AddOrUpdateConn(conn)
+		}
+		connsLens = append(connsLens, len(filteredConnsList))
+	}
+	metrics.TotalConnectionsInConnTrackTable.Set(float64(totalConns))
+	maxConns, err := cs.connDumper.GetMaxConnections()
+	if err != nil {
+		return []int{}, err
+	}
+	metrics.MaxConnectionsInConnTrackTable.Set(float64(maxConns))
+	klog.V(2).Infof("Conntrack polling successful")
+
+	return connsLens, nil
+}
+
+func (cs *ConntrackConnectionStore) addNetworkPolicyMetadata(conn *flowexporter.Connection) {
+	// Retrieve NetworkPolicy Name and Namespace by using the ingress and egress
+	// IDs stored in the connection label.
+	if len(conn.Labels) != 0 {
+		klog.V(4).Infof("connection label: %x; label masks: %x", conn.Labels, conn.LabelsMask)
+		ingressOfID := binary.LittleEndian.Uint32(conn.Labels[:4])
+		egressOfID := binary.LittleEndian.Uint32(conn.Labels[4:8])
+		if ingressOfID != 0 {
+			policy := cs.networkPolicyQuerier.GetNetworkPolicyByRuleFlowID(ingressOfID)
+			if policy == nil {
+				// This should not happen because the rule flow ID to rule mapping is
+				// preserved for max(5s, flowPollInterval) even after the rule deletion.
+				klog.Warningf("Cannot find NetworkPolicy that has rule with ingressOfID %v", ingressOfID)
+			} else {
+				conn.IngressNetworkPolicyName = policy.Name
+				conn.IngressNetworkPolicyNamespace = policy.Namespace
+				conn.IngressNetworkPolicyRuleAction = registry.NetworkPolicyRuleActionAllow
+			}
+		}
+		if egressOfID != 0 {
+			policy := cs.networkPolicyQuerier.GetNetworkPolicyByRuleFlowID(egressOfID)
+			if policy == nil {
+				// This should not happen because the rule flow ID to rule mapping is
+				// preserved for max(5s, flowPollInterval) even after the rule deletion.
+				klog.Warningf("Cannot find NetworkPolicy that has rule with egressOfID %v", egressOfID)
+			} else {
+				conn.EgressNetworkPolicyName = policy.Name
+				conn.EgressNetworkPolicyNamespace = policy.Namespace
+				conn.EgressNetworkPolicyRuleAction = registry.NetworkPolicyRuleActionAllow
+			}
+		}
+	}
+}
+
+// AddOrUpdateConn updates the connection if it is already present, i.e., update timestamp, counters etc.,
+// or adds a new connection with the resolved K8s metadata.
+func (cs *ConntrackConnectionStore) AddOrUpdateConn(conn *flowexporter.Connection) {
+	connKey := flowexporter.NewConnectionKey(conn)
+	cs.mutex.Lock()
+	defer cs.mutex.Unlock()
+	existingConn, exists := cs.connections[connKey]
+
+	if exists {
+		existingConn.IsPresent = true
+		// avoid updating stats of the existing connection that is about to close
+		if flowexporter.IsConnectionDying(existingConn) {
+			return
+		}
+		// Update the necessary fields that are used in generating flow records.
+		// Can same 5-tuple flow get deleted and added to conntrack table? If so use ID.
+		existingConn.StopTime = conn.StopTime
+		existingConn.OriginalBytes = conn.OriginalBytes
+		existingConn.OriginalPackets = conn.OriginalPackets
+		existingConn.ReverseBytes = conn.ReverseBytes
+		existingConn.ReversePackets = conn.ReversePackets
+		existingConn.TCPState = conn.TCPState
+		klog.V(4).Infof("Antrea flow updated: %v", existingConn)
+	} else {
+		cs.fillPodInfo(conn)
+		if conn.Mark == openflow.ServiceCTMark {
+			clusterIP := conn.DestinationServiceAddress.String()
+			svcPort := conn.DestinationServicePort
+			protocol, err := lookupServiceProtocol(conn.FlowKey.Protocol)
+			if err != nil {
+				klog.Warningf("Could not retrieve Service protocol: %v", err)
+			} else {
+				serviceStr := fmt.Sprintf("%s:%d/%s", clusterIP, svcPort, protocol)
+				cs.fillServiceInfo(conn, serviceStr)
+			}
+		}
+		cs.addNetworkPolicyMetadata(conn)
+
+		metrics.TotalAntreaConnectionsInConnTrackTable.Inc()
+		klog.V(4).Infof("New Antrea flow added: %v", conn)
+		// Add new antrea connection to connection store
+		cs.connections[connKey] = conn
+	}
+}
+
+// DeleteConnWithoutLock deletes the connection from the connection map given
+// the connection key without grabbing the lock. Caller is expected to grab lock.
+func (cs *ConntrackConnectionStore) DeleteConnWithoutLock(connKey flowexporter.ConnectionKey) error {
+	_, exists := cs.connections[connKey]
+	if !exists {
+		return fmt.Errorf("connection with key %v doesn't exist in map", connKey)
+	}
+	delete(cs.connections, connKey)
+	metrics.TotalAntreaConnectionsInConnTrackTable.Dec()
+	return nil
+}
+
+// SetExportDone sets DoneExport field of conntrack connection to true given the connection key.
+func (cs *ConntrackConnectionStore) SetExportDone(connKey flowexporter.ConnectionKey) error {
+	cs.mutex.Lock()
+	defer cs.mutex.Unlock()
+
+	if conn, found := cs.connections[connKey]; !found {
+		return fmt.Errorf("connection with key %v does not exist in connection map", connKey)
+	} else {
+		conn.DoneExport = true
+		return nil
+	}
+}
diff --git a/pkg/agent/flowexporter/connections/conntrack_linux.go b/pkg/agent/flowexporter/connections/conntrack_linux.go
index f12fa6165..0138bbdbc 100644
--- a/pkg/agent/flowexporter/connections/conntrack_linux.go
+++ b/pkg/agent/flowexporter/connections/conntrack_linux.go
@@ -121,43 +121,36 @@ func (nfct *netFilterConnTrack) DumpFlowsInCtZone(zoneFilter uint16) ([]*flowexp
 }
 
 func NetlinkFlowToAntreaConnection(conn *conntrack.Flow) *flowexporter.Connection {
-	tupleOrig := flowexporter.Tuple{
+	tuple := flowexporter.Tuple{
 		SourceAddress:      conn.TupleOrig.IP.SourceAddress,
-		DestinationAddress: conn.TupleOrig.IP.DestinationAddress,
+		DestinationAddress: conn.TupleReply.IP.SourceAddress,
 		Protocol:           conn.TupleOrig.Proto.Protocol,
 		SourcePort:         conn.TupleOrig.Proto.SourcePort,
-		DestinationPort:    conn.TupleOrig.Proto.DestinationPort,
-	}
-
-	tupleReply := flowexporter.Tuple{
-		SourceAddress:      conn.TupleReply.IP.SourceAddress,
-		DestinationAddress: conn.TupleReply.IP.DestinationAddress,
-		Protocol:           conn.TupleReply.Proto.Protocol,
-		SourcePort:         conn.TupleReply.Proto.SourcePort,
-		DestinationPort:    conn.TupleReply.Proto.DestinationPort,
+		DestinationPort:    conn.TupleReply.Proto.SourcePort,
 	}
 	// Assign all the applicable fields
 	newConn := flowexporter.Connection{
-		ID:                      conn.ID,
-		Timeout:                 conn.Timeout,
-		StartTime:               conn.Timestamp.Start,
-		IsPresent:               true,
-		Zone:                    conn.Zone,
-		Mark:                    conn.Mark,
-		Labels:                  conn.Labels,
-		LabelsMask:              conn.LabelsMask,
-		StatusFlag:              uint32(conn.Status.Value),
-		TupleOrig:               tupleOrig,
-		TupleReply:              tupleReply,
-		OriginalPackets:         conn.CountersOrig.Packets,
-		OriginalBytes:           conn.CountersOrig.Bytes,
-		ReversePackets:          conn.CountersReply.Packets,
-		ReverseBytes:            conn.CountersReply.Bytes,
-		SourcePodNamespace:      "",
-		SourcePodName:           "",
-		DestinationPodNamespace: "",
-		DestinationPodName:      "",
-		TCPState:                "",
+		ID:                        conn.ID,
+		Timeout:                   conn.Timeout,
+		StartTime:                 conn.Timestamp.Start,
+		IsPresent:                 true,
+		Zone:                      conn.Zone,
+		Mark:                      conn.Mark,
+		Labels:                    conn.Labels,
+		LabelsMask:                conn.LabelsMask,
+		StatusFlag:                uint32(conn.Status.Value),
+		FlowKey:                   tuple,
+		DestinationServiceAddress: conn.TupleOrig.IP.DestinationAddress,
+		DestinationServicePort:    conn.TupleOrig.Proto.DestinationPort,
+		OriginalPackets:           conn.CountersOrig.Packets,
+		OriginalBytes:             conn.CountersOrig.Bytes,
+		ReversePackets:            conn.CountersReply.Packets,
+		ReverseBytes:              conn.CountersReply.Bytes,
+		SourcePodNamespace:        "",
+		SourcePodName:             "",
+		DestinationPodNamespace:   "",
+		DestinationPodName:        "",
+		TCPState:                  "",
 	}
 	if conn.ProtoInfo.TCP != nil {
 		newConn.TCPState = stateToString(conn.ProtoInfo.TCP.State)
diff --git a/pkg/agent/flowexporter/connections/conntrack_ovs.go b/pkg/agent/flowexporter/connections/conntrack_ovs.go
index a81830be8..3aa6bacfc 100644
--- a/pkg/agent/flowexporter/connections/conntrack_ovs.go
+++ b/pkg/agent/flowexporter/connections/conntrack_ovs.go
@@ -144,21 +144,18 @@ func flowStringToAntreaConnection(flow string, zoneFilter uint16) (*flowexporter
 			if err != nil {
 				return nil, err
 			}
-			conn.TupleReply.Protocol = proto
-			conn.TupleOrig.Protocol = proto
+			conn.FlowKey.Protocol = proto
 		case strings.Contains(fs, "src"):
 			fields := strings.Split(fs, "=")
 			if !isReply {
-				conn.TupleOrig.SourceAddress = net.ParseIP(fields[len(fields)-1])
+				conn.FlowKey.SourceAddress = net.ParseIP(fields[len(fields)-1])
 			} else {
-				conn.TupleReply.SourceAddress = net.ParseIP(fields[len(fields)-1])
+				conn.FlowKey.DestinationAddress = net.ParseIP(fields[len(fields)-1])
 			}
 		case strings.Contains(fs, "dst"):
 			fields := strings.Split(fs, "=")
 			if !isReply {
-				conn.TupleOrig.DestinationAddress = net.ParseIP(fields[len(fields)-1])
-			} else {
-				conn.TupleReply.DestinationAddress = net.ParseIP(fields[len(fields)-1])
+				conn.DestinationServiceAddress = net.ParseIP(fields[len(fields)-1])
 			}
 		case strings.Contains(fs, "sport"):
 			fields := strings.Split(fs, "=")
@@ -167,9 +164,9 @@ func flowStringToAntreaConnection(flow string, zoneFilter uint16) (*flowexporter
 				return nil, fmt.Errorf("conversion of sport %s to int failed: %v", fields[len(fields)-1], err)
 			}
 			if !isReply {
-				conn.TupleOrig.SourcePort = uint16(val)
+				conn.FlowKey.SourcePort = uint16(val)
 			} else {
-				conn.TupleReply.SourcePort = uint16(val)
+				conn.FlowKey.DestinationPort = uint16(val)
 			}
 		case strings.Contains(fs, "dport"):
 			// dport field could be the last tuple field in ovs-dpctl output format.
@@ -180,9 +177,7 @@ func flowStringToAntreaConnection(flow string, zoneFilter uint16) (*flowexporter
 				return nil, fmt.Errorf("conversion of dport %s to int failed: %v", fields[len(fields)-1], err)
 			}
 			if !isReply {
-				conn.TupleOrig.DestinationPort = uint16(val)
-			} else {
-				conn.TupleReply.DestinationPort = uint16(val)
+				conn.DestinationServicePort = uint16(val)
 			}
 		case strings.Contains(fs, "packets"):
 			fields := strings.Split(fs, "=")
@@ -282,6 +277,7 @@ func flowStringToAntreaConnection(flow string, zoneFilter uint16) (*flowexporter
 	if !inZone {
 		return nil, nil
 	}
+
 	// Add current time as stop time.
 	conn.StopTime = time.Now()
 	conn.IsPresent = true
diff --git a/pkg/agent/flowexporter/connections/deny_connections.go b/pkg/agent/flowexporter/connections/deny_connections.go
new file mode 100644
index 000000000..9c8b4bfef
--- /dev/null
+++ b/pkg/agent/flowexporter/connections/deny_connections.go
@@ -0,0 +1,96 @@
+// Copyright 2021 Antrea Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package connections
+
+import (
+	"fmt"
+	"time"
+
+	"k8s.io/klog/v2"
+
+	"antrea.io/antrea/pkg/agent/flowexporter"
+	"antrea.io/antrea/pkg/agent/interfacestore"
+	"antrea.io/antrea/pkg/agent/metrics"
+	"antrea.io/antrea/pkg/agent/proxy"
+	"antrea.io/antrea/pkg/util/ip"
+)
+
+type DenyConnectionStore struct {
+	connectionStore
+}
+
+func NewDenyConnectionStore(ifaceStore interfacestore.InterfaceStore,
+	proxier proxy.Proxier) *DenyConnectionStore {
+	return &DenyConnectionStore{
+		connectionStore: NewConnectionStore(ifaceStore, proxier),
+	}
+}
+
+// AddOrUpdateConn updates the connection if it is already present, i.e., update timestamp, counters etc.,
+// or adds a new connection with the resolved K8s metadata.
+func (ds *DenyConnectionStore) AddOrUpdateConn(conn *flowexporter.Connection, timeSeen time.Time, bytes uint64) {
+	connKey := flowexporter.NewConnectionKey(conn)
+	ds.mutex.Lock()
+	defer ds.mutex.Unlock()
+	if _, exist := ds.connections[connKey]; exist {
+		conn.DeltaBytes += bytes
+		conn.OriginalBytes += bytes
+		conn.DeltaPackets += 1
+		conn.OriginalPackets += 1
+		conn.StopTime = timeSeen
+		klog.V(2).Infof("Deny connection with flowKey %v has been updated.", connKey)
+		return
+	} else {
+		conn.StartTime = timeSeen
+		conn.StopTime = timeSeen
+		conn.LastExportTime = timeSeen
+		conn.DeltaBytes = bytes
+		conn.OriginalBytes = bytes
+		conn.DeltaPackets = uint64(1)
+		conn.OriginalPackets = uint64(1)
+		ds.fillPodInfo(conn)
+		protocolStr := ip.IPProtocolNumberToString(conn.FlowKey.Protocol, "UnknownProtocol")
+		serviceStr := fmt.Sprintf("%s:%d/%s", conn.DestinationServiceAddress, conn.DestinationServicePort, protocolStr)
+		ds.fillServiceInfo(conn, serviceStr)
+		metrics.TotalDenyConnections.Inc()
+		ds.connections[connKey] = conn
+	}
+}
+
+// ResetConnStatsWithoutLock resets DeltaBytes and DeltaPackets of connection
+// after exporting without grabbing the lock. Caller is expected to grab lock.
+func (ds *DenyConnectionStore) ResetConnStatsWithoutLock(conn *flowexporter.Connection) {
+	connKey := flowexporter.NewConnectionKey(conn)
+	conn, exist := ds.connections[connKey]
+	if !exist {
+		klog.Warningf("Connection with key %s does not exist in deny connection store.", connKey)
+	} else {
+		conn.DeltaBytes = 0
+		conn.DeltaPackets = 0
+		conn.LastExportTime = time.Now()
+	}
+}
+
+// DeleteConnWithoutLock deletes the connection from the connection map given
+// the connection key without grabbing the lock. Caller is expected to grab lock.
+func (ds *DenyConnectionStore) DeleteConnWithoutLock(connKey flowexporter.ConnectionKey) error {
+	_, exists := ds.connections[connKey]
+	if !exists {
+		return fmt.Errorf("connection with key %v doesn't exist in map", connKey)
+	}
+	delete(ds.connections, connKey)
+	metrics.TotalDenyConnections.Dec()
+	return nil
+}
diff --git a/pkg/agent/flowexporter/exporter/exporter.go b/pkg/agent/flowexporter/exporter/exporter.go
index 11dd98126..f0bc2dbc6 100644
--- a/pkg/agent/flowexporter/exporter/exporter.go
+++ b/pkg/agent/flowexporter/exporter/exporter.go
@@ -69,8 +69,10 @@ var (
 		"destinationServicePortName",
 		"ingressNetworkPolicyName",
 		"ingressNetworkPolicyNamespace",
+		"ingressNetworkPolicyRuleAction",
 		"egressNetworkPolicyName",
 		"egressNetworkPolicyNamespace",
+		"egressNetworkPolicyRuleAction",
 		"tcpState",
 		"flowType",
 	}
@@ -79,8 +81,9 @@ var (
 )
 
 type flowExporter struct {
-	connStore           connections.ConnectionStore
+	conntrackConnStore  *connections.ConntrackConnectionStore
 	flowRecords         *flowrecords.FlowRecords
+	denyConnStore       *connections.DenyConnectionStore
 	process             ipfix.IPFIXExportingProcess
 	elementsListv4      []*ipfixentities.InfoElementWithValue
 	elementsListv6      []*ipfixentities.InfoElementWithValue
@@ -130,7 +133,7 @@ func prepareExporterInputArgs(collectorAddr, collectorProto string) (exporter.Ex
 	return expInput, nil
 }
 
-func NewFlowExporter(connStore connections.ConnectionStore, records *flowrecords.FlowRecords,
+func NewFlowExporter(connStore *connections.ConntrackConnectionStore, records *flowrecords.FlowRecords, denyConnStore *connections.DenyConnectionStore,
 	collectorAddr string, collectorProto string, activeFlowTimeout time.Duration, idleFlowTimeout time.Duration,
 	v4Enabled bool, v6Enabled bool, k8sClient kubernetes.Interface,
 	nodeRouteController *noderoute.Controller, isNetworkPolicyOnly bool) (*flowExporter, error) {
@@ -145,8 +148,9 @@ func NewFlowExporter(connStore connections.ConnectionStore, records *flowrecords
 	}
 
 	return &flowExporter{
-		connStore:           connStore,
+		conntrackConnStore:  connStore,
 		flowRecords:         records,
+		denyConnStore:       denyConnStore,
 		registry:            registry,
 		v4Enabled:           v4Enabled,
 		v6Enabled:           v6Enabled,
@@ -311,7 +315,7 @@ func (exp *flowExporter) sendFlowRecords() error {
 				if err := exp.flowRecords.DeleteFlowRecordWithoutLock(key); err != nil {
 					return err
 				}
-				if err := exp.connStore.SetExportDone(key); err != nil {
+				if err := exp.conntrackConnStore.SetExportDone(key); err != nil {
 					return err
 				}
 			} else {
@@ -320,11 +324,38 @@ func (exp *flowExporter) sendFlowRecords() error {
 		}
 		return nil
 	}
-
 	err := exp.flowRecords.ForAllFlowRecordsDo(updateOrSendFlowRecord)
 	if err != nil {
 		return fmt.Errorf("error when iterating flow records: %v", err)
 	}
+
+	exportDenyConn := func(connKey flowexporter.ConnectionKey, conn *flowexporter.Connection) error {
+		if conn.DeltaPackets > 0 && time.Since(conn.LastExportTime) >= exp.activeFlowTimeout {
+			if err := exp.addDenyConnToSet(conn, ipfixregistry.ActiveTimeoutReason); err != nil {
+				return err
+			}
+			if _, err := exp.sendDataSet(); err != nil {
+				return err
+			}
+			exp.numDataSetsSent = exp.numDataSetsSent + 1
+			exp.denyConnStore.ResetConnStatsWithoutLock(conn)
+		}
+		if time.Since(conn.LastExportTime) >= exp.idleFlowTimeout {
+			if err := exp.addDenyConnToSet(conn, ipfixregistry.IdleTimeoutReason); err != nil {
+				return err
+			}
+			if _, err := exp.sendDataSet(); err != nil {
+				return err
+			}
+			exp.numDataSetsSent = exp.numDataSetsSent + 1
+			exp.denyConnStore.DeleteConnWithoutLock(connKey)
+		}
+		return nil
+	}
+	err = exp.denyConnStore.ForAllConnectionsDo(exportDenyConn)
+	if err != nil {
+		return fmt.Errorf("error when iterating deny connections: %v", err)
+	}
 	return nil
 }
 
@@ -407,19 +438,19 @@ func (exp *flowExporter) addRecordToSet(record flowexporter.FlowRecord) error {
 				ie.Value = ipfixregistry.IdleTimeoutReason
 			}
 		case "sourceIPv4Address":
-			ie.Value = record.Conn.TupleOrig.SourceAddress
+			ie.Value = record.Conn.FlowKey.SourceAddress
 		case "destinationIPv4Address":
-			ie.Value = record.Conn.TupleReply.SourceAddress
+			ie.Value = record.Conn.FlowKey.DestinationAddress
 		case "sourceIPv6Address":
-			ie.Value = record.Conn.TupleOrig.SourceAddress
+			ie.Value = record.Conn.FlowKey.SourceAddress
 		case "destinationIPv6Address":
-			ie.Value = record.Conn.TupleReply.SourceAddress
+			ie.Value = record.Conn.FlowKey.DestinationAddress
 		case "sourceTransportPort":
-			ie.Value = record.Conn.TupleOrig.SourcePort
+			ie.Value = record.Conn.FlowKey.SourcePort
 		case "destinationTransportPort":
-			ie.Value = record.Conn.TupleReply.SourcePort
+			ie.Value = record.Conn.FlowKey.DestinationPort
 		case "protocolIdentifier":
-			ie.Value = record.Conn.TupleOrig.Protocol
+			ie.Value = record.Conn.FlowKey.Protocol
 		case "packetTotalCount":
 			ie.Value = record.Conn.OriginalPackets
 		case "octetTotalCount":
@@ -476,7 +507,7 @@ func (exp *flowExporter) addRecordToSet(record flowexporter.FlowRecord) error {
 			}
 		case "destinationClusterIPv4":
 			if record.Conn.DestinationServicePortName != "" {
-				ie.Value = record.Conn.TupleOrig.DestinationAddress
+				ie.Value = record.Conn.DestinationServiceAddress
 			} else {
 				// Sending dummy IP as IPFIX collector expects constant length of data for IP field.
 				// We should probably think of better approach as this involves customization of IPFIX collector to ignore
@@ -485,14 +516,14 @@ func (exp *flowExporter) addRecordToSet(record flowexporter.FlowRecord) error {
 			}
 		case "destinationClusterIPv6":
 			if record.Conn.DestinationServicePortName != "" {
-				ie.Value = record.Conn.TupleOrig.DestinationAddress
+				ie.Value = record.Conn.DestinationServiceAddress
 			} else {
 				// Same as destinationClusterIPv4.
 				ie.Value = net.ParseIP("::")
 			}
 		case "destinationServicePort":
 			if record.Conn.DestinationServicePortName != "" {
-				ie.Value = record.Conn.TupleOrig.DestinationPort
+				ie.Value = record.Conn.DestinationServicePort
 			} else {
 				ie.Value = uint16(0)
 			}
@@ -506,14 +537,18 @@ func (exp *flowExporter) addRecordToSet(record flowexporter.FlowRecord) error {
 			ie.Value = record.Conn.IngressNetworkPolicyName
 		case "ingressNetworkPolicyNamespace":
 			ie.Value = record.Conn.IngressNetworkPolicyNamespace
+		case "ingressNetworkPolicyRuleAction":
+			ie.Value = record.Conn.IngressNetworkPolicyRuleAction
 		case "egressNetworkPolicyName":
 			ie.Value = record.Conn.EgressNetworkPolicyName
 		case "egressNetworkPolicyNamespace":
 			ie.Value = record.Conn.EgressNetworkPolicyNamespace
+		case "egressNetworkPolicyRuleAction":
+			ie.Value = record.Conn.EgressNetworkPolicyRuleAction
 		case "tcpState":
 			ie.Value = record.Conn.TCPState
 		case "flowType":
-			ie.Value = exp.findFlowType(record)
+			ie.Value = exp.findFlowType(record.Conn)
 		}
 	}
 
@@ -528,6 +563,120 @@ func (exp *flowExporter) addRecordToSet(record flowexporter.FlowRecord) error {
 	return nil
 }
 
+func (exp *flowExporter) addDenyConnToSet(conn *flowexporter.Connection, flowEndReason uint8) error {
+	nodeName, _ := env.GetNodeName()
+	exp.ipfixSet.ResetSet()
+
+	eL := exp.elementsListv4
+	templateID := exp.templateIDv4
+	if conn.FlowKey.SourceAddress.To4() == nil {
+		templateID = exp.templateIDv6
+		eL = exp.elementsListv6
+	}
+	if err := exp.ipfixSet.PrepareSet(ipfixentities.Data, templateID); err != nil {
+		return err
+	}
+	// Iterate over all infoElements in the list
+	for _, ie := range eL {
+		switch ieName := ie.Element.Name; ieName {
+		case "flowStartSeconds":
+			ie.Value = uint32(conn.StartTime.Unix())
+		case "flowEndSeconds":
+			ie.Value = uint32(conn.StopTime.Unix())
+		case "flowEndReason":
+			ie.Value = flowEndReason
+		case "sourceIPv4Address":
+			ie.Value = conn.FlowKey.SourceAddress
+		case "destinationIPv4Address":
+			ie.Value = conn.FlowKey.DestinationAddress
+		case "sourceIPv6Address":
+			ie.Value = conn.FlowKey.SourceAddress
+		case "destinationIPv6Address":
+			ie.Value = conn.FlowKey.DestinationAddress
+		case "sourceTransportPort":
+			ie.Value = conn.FlowKey.SourcePort
+		case "destinationTransportPort":
+			ie.Value = conn.FlowKey.DestinationPort
+		case "protocolIdentifier":
+			ie.Value = conn.FlowKey.Protocol
+		case "packetTotalCount":
+			ie.Value = conn.OriginalPackets
+		case "octetTotalCount":
+			ie.Value = conn.OriginalBytes
+		case "packetDeltaCount":
+			ie.Value = conn.DeltaPackets
+		case "octetDeltaCount":
+			ie.Value = conn.DeltaBytes
+		case "reversePacketTotalCount", "reverseOctetTotalCount", "reversePacketDeltaCount", "reverseOctetDeltaCount":
+			ie.Value = uint64(0)
+		case "sourcePodNamespace":
+			ie.Value = conn.SourcePodNamespace
+		case "sourcePodName":
+			ie.Value = conn.SourcePodName
+		case "sourceNodeName":
+			// Add nodeName for only local pods whose pod names are resolved.
+			if conn.SourcePodName != "" {
+				ie.Value = nodeName
+			} else {
+				ie.Value = ""
+			}
+		case "destinationPodNamespace":
+			ie.Value = conn.DestinationPodNamespace
+		case "destinationPodName":
+			ie.Value = conn.DestinationPodName
+		case "destinationNodeName":
+			// Add nodeName for only local pods whose pod names are resolved.
+			if conn.DestinationPodName != "" {
+				ie.Value = nodeName
+			} else {
+				ie.Value = ""
+			}
+		case "destinationClusterIPv4":
+			if conn.DestinationServicePortName != "" {
+				ie.Value = conn.DestinationServiceAddress
+			} else {
+				ie.Value = net.IP{0, 0, 0, 0}
+			}
+		case "destinationClusterIPv6":
+			if conn.DestinationServicePortName != "" {
+				ie.Value = conn.DestinationServiceAddress
+			} else {
+				ie.Value = net.ParseIP("::")
+			}
+		case "destinationServicePort":
+			if conn.DestinationServicePortName != "" {
+				ie.Value = conn.DestinationServicePort
+			} else {
+				ie.Value = uint16(0)
+			}
+		case "destinationServicePortName":
+			ie.Value = conn.DestinationServicePortName
+		case "ingressNetworkPolicyName":
+			ie.Value = conn.IngressNetworkPolicyName
+		case "ingressNetworkPolicyNamespace":
+			ie.Value = conn.IngressNetworkPolicyNamespace
+		case "ingressNetworkPolicyRuleAction":
+			ie.Value = conn.IngressNetworkPolicyRuleAction
+		case "egressNetworkPolicyName":
+			ie.Value = conn.EgressNetworkPolicyName
+		case "egressNetworkPolicyNamespace":
+			ie.Value = conn.EgressNetworkPolicyNamespace
+		case "egressNetworkPolicyRuleAction":
+			ie.Value = conn.EgressNetworkPolicyRuleAction
+		case "tcpState":
+			ie.Value = ""
+		case "flowType":
+			ie.Value = exp.findFlowType(*conn)
+		}
+	}
+
+	err := exp.ipfixSet.AddRecord(eL, templateID)
+	if err != nil {
+		return fmt.Errorf("error in adding record to data set: %v", err)
+	}
+	return nil
+}
+
 func (exp *flowExporter) sendDataSet() (int, error) {
 	sentBytes, err := exp.process.SendSet(exp.ipfixSet)
 	if err != nil {
@@ -537,10 +686,10 @@ func (exp *flowExporter) sendDataSet() (int, error) {
 	return sentBytes, nil
 }
 
-func (exp *flowExporter) findFlowType(record flowexporter.FlowRecord) uint8 {
+func (exp *flowExporter) findFlowType(conn flowexporter.Connection) uint8 {
 	// TODO: support Pod-To-External flows in network policy only mode.
 	if exp.isNetworkPolicyOnly {
-		if record.Conn.SourcePodName == "" || record.Conn.DestinationPodName == "" {
+		if conn.SourcePodName == "" || conn.DestinationPodName == "" {
 			return ipfixregistry.FlowTypeInterNode
 		}
 		return ipfixregistry.FlowTypeIntraNode
@@ -550,9 +699,9 @@ func (exp *flowExporter) findFlowType(record flowexporter.FlowRecord) uint8 {
 		klog.Warningf("Can't find flowType without nodeRouteController")
 		return 0
 	}
-	if exp.nodeRouteController.IPInPodSubnets(record.Conn.TupleOrig.SourceAddress) {
-		if record.Conn.Mark == openflow.ServiceCTMark || exp.nodeRouteController.IPInPodSubnets(record.Conn.TupleOrig.DestinationAddress) {
-			if record.Conn.SourcePodName == "" || record.Conn.DestinationPodName == "" {
+	if exp.nodeRouteController.IPInPodSubnets(conn.FlowKey.SourceAddress) {
+		if conn.Mark == openflow.ServiceCTMark || exp.nodeRouteController.IPInPodSubnets(conn.FlowKey.DestinationAddress) {
+			if conn.SourcePodName == "" || conn.DestinationPodName == "" {
 				return ipfixregistry.FlowTypeInterNode
 			}
 			return ipfixregistry.FlowTypeIntraNode
@@ -561,7 +710,7 @@ func (exp *flowExporter) findFlowType(record flowexporter.FlowRecord) uint8 {
 		}
 	} else {
 		// We do not support External-To-Pod flows for now.
-		klog.Warningf("Source IP: %s doesn't exist in PodCIDRs", record.Conn.TupleOrig.SourceAddress.String())
+		klog.Warningf("Source IP: %s doesn't exist in PodCIDRs", conn.FlowKey.SourceAddress.String())
 		return 0
 	}
 }
diff --git a/pkg/agent/flowexporter/flowrecords/flow_records.go b/pkg/agent/flowexporter/flowrecords/flow_records.go
index aa77dcf7a..9e1c487ac 100644
--- a/pkg/agent/flowexporter/flowrecords/flow_records.go
+++ b/pkg/agent/flowexporter/flowrecords/flow_records.go
@@ -16,7 +16,6 @@ package flowrecords
 
 import (
 	"fmt"
-	"net"
 	"sync"
 	"time"
 
@@ -52,7 +51,7 @@ func (fr *FlowRecords) AddOrUpdateFlowRecord(key flowexporter.ConnectionKey, con
 	record, exists := fr.recordsMap[key]
 	if !exists {
 		isIPv6 := false
-		if net.ParseIP(key[0]).To4() == nil {
+		if conn.FlowKey.SourceAddress.To4() == nil {
 			isIPv6 = true
 		}
 		record = flowexporter.FlowRecord{
diff --git a/pkg/agent/flowexporter/types.go b/pkg/agent/flowexporter/types.go
index 42da673b9..c5e04de8e 100644
--- a/pkg/agent/flowexporter/types.go
+++ b/pkg/agent/flowexporter/types.go
@@ -37,7 +37,8 @@ type Connection struct {
 	ID        uint32
 	Timeout   uint32
 	StartTime time.Time
-	// For invalid and closed connections: StopTime is the time when connection was updated last.
+	// For invalid and closed connections or deny connections: StopTime is the time when connection
+	// was updated last.
 	// For established connections: StopTime is latest time when it was polled.
 	StopTime time.Time
 	// IsPresent flag helps in cleaning up connections when they are not in conntrack table anymore.
@@ -50,20 +51,29 @@ type Connection struct {
 	StatusFlag         uint32
 	Labels, LabelsMask []byte
 	// TODO: Have a separate field for protocol. No need to keep it in Tuple.
-	TupleOrig, TupleReply          Tuple
+	FlowKey                        Tuple
 	OriginalPackets, OriginalBytes uint64
 	ReversePackets, ReverseBytes   uint64
 	// Fields specific to Antrea
-	SourcePodNamespace            string
-	SourcePodName                 string
-	DestinationPodNamespace       string
-	DestinationPodName            string
-	DestinationServicePortName    string
-	IngressNetworkPolicyName      string
-	IngressNetworkPolicyNamespace string
-	EgressNetworkPolicyName       string
-	EgressNetworkPolicyNamespace  string
-	TCPState                      string
+	SourcePodNamespace             string
+	SourcePodName                  string
+	DestinationPodNamespace        string
+	DestinationPodName             string
+	DestinationServicePortName     string
+	DestinationServiceAddress      net.IP
+	DestinationServicePort         uint16
+	IngressNetworkPolicyName       string
+	IngressNetworkPolicyNamespace  string
+	IngressNetworkPolicyRuleAction uint8
+	EgressNetworkPolicyName        string
+	EgressNetworkPolicyNamespace   string
+	EgressNetworkPolicyRuleAction  uint8
+	TCPState                       string
+	// fields specific to deny connections
+	// DeltaBytes and DeltaPackets are octetDeltaCount and packetDeltaCount over each active
+	// flow timeout duration.
+	DeltaBytes, DeltaPackets uint64
+	LastExportTime           time.Time
 }
 
 type FlowRecord struct {
diff --git a/pkg/agent/flowexporter/utils.go b/pkg/agent/flowexporter/utils.go
index d7ea450a4..26171f73e 100644
--- a/pkg/agent/flowexporter/utils.go
+++ b/pkg/agent/flowexporter/utils.go
@@ -16,6 +16,8 @@ package flowexporter
 
 import (
 	"strconv"
+
+	"github.com/vmware/go-ipfix/pkg/registry"
 )
 
 const (
@@ -24,11 +26,11 @@ const (
 
 // NewConnectionKey creates 5-tuple of flow as connection key
 func NewConnectionKey(conn *Connection) ConnectionKey {
-	return ConnectionKey{conn.TupleOrig.SourceAddress.String(),
-		strconv.FormatUint(uint64(conn.TupleOrig.SourcePort), 10),
-		conn.TupleReply.SourceAddress.String(),
-		strconv.FormatUint(uint64(conn.TupleReply.SourcePort), 10),
-		strconv.FormatUint(uint64(conn.TupleOrig.Protocol), 10),
+	return ConnectionKey{conn.FlowKey.SourceAddress.String(),
+		strconv.FormatUint(uint64(conn.FlowKey.SourcePort), 10),
+		conn.FlowKey.DestinationAddress.String(),
+		strconv.FormatUint(uint64(conn.FlowKey.DestinationPort), 10),
+		strconv.FormatUint(uint64(conn.FlowKey.Protocol), 10),
 	}
 }
 
@@ -48,3 +50,17 @@ func IsConnectionDying(conn *Connection) bool {
 	}
 	return false
 }
+
+// RuleActionToUint8 converts network policy rule action to uint8.
+func RuleActionToUint8(action string) uint8 {
+	switch action {
+	case "Allow":
+		return registry.NetworkPolicyRuleActionAllow
+	case "Drop":
+		return registry.NetworkPolicyRuleActionDrop
+	case "Reject":
+		return registry.NetworkPolicyRuleActionReject
+	default:
+		return registry.NetworkPolicyRuleActionNoAction
+	}
+}
diff --git a/pkg/agent/metrics/prometheus.go b/pkg/agent/metrics/prometheus.go
index c32e9ec0d..60fb2de9e 100644
--- a/pkg/agent/metrics/prometheus.go
+++ b/pkg/agent/metrics/prometheus.go
@@ -136,6 +136,16 @@ var (
 		},
 	)
 
+	TotalDenyConnections = metrics.NewGauge(
+		&metrics.GaugeOpts{
+			Namespace:      metricNamespaceAntrea,
+			Subsystem:      metricSubsystemAgent,
+			Name:           "denied_connection_count",
+			Help:           "Number of denied connections detected by Flow Exporter deny connections tracking. This metric gets updated when a flow is rejected/dropped by network policy.",
+			StabilityLevel: metrics.ALPHA,
+		},
+	)
+
 	MaxConnectionsInConnTrackTable = metrics.NewGauge(
 		&metrics.GaugeOpts{
 			Namespace:      metricNamespaceAntrea,
@@ -210,6 +220,9 @@ func InitializeConnectionMetrics() {
 	if err := legacyregistry.Register(TotalAntreaConnectionsInConnTrackTable); err != nil {
 		klog.Errorf("Failed to register antrea_agent_conntrack_antrea_connection_count with error: %v", err)
 	}
+	if err := legacyregistry.Register(TotalDenyConnections); err != nil {
+		klog.Errorf("Failed to register antrea_agent_denied_connection_count with error: %v", err)
+	}
 	if err := legacyregistry.Register(MaxConnectionsInConnTrackTable); err != nil {
 		klog.Errorf("Failed to register antrea_agent_conntrack_max_connection_count with error: %v", err)
 	}
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 46e34190e..92da130bf 100644
--- a/pkg/agent/openflow/pipeline.go
+++ b/pkg/agent/openflow/pipeline.go
@@ -223,7 +223,7 @@ func (rt regType) reg() string {
 
 const (
 	// marksReg stores traffic-source mark and pod-found mark.
-	// traffic-source resides in [0..15], pod-found resides in [16], Antrea Policy disposition in [21-22], Custom Reasons in [24-25]
+	// traffic-source resides in [0..15], pod-found resides in [16], Antrea Policy disposition in [21-22], Custom Reasons in [24-26]
 	marksReg        regType = 0
 	PortCacheReg    regType = 1
 	swapReg         regType = 2
@@ -268,7 +268,7 @@ const (
 	DispositionDrop  = 0b01
 	DispositionRej   = 0b10
 
-	// custom reason is loaded in marksReg [24-25]
+	// custom reason is loaded in marksReg [24-26]
 	// The custom reason mark is used to indicate the reason(s) for sending the packet
 	// to the controller. Reasons can be or-ed to indicate that the packets was sent
 	// for multiple reasons.
@@ -284,6 +284,11 @@ const (
 	// controller send reject packet as packet-out, we want reject response to bypass
 	// the connTrack to avoid unexpected drop.
 	CustomReasonReject = 0b10
+	// CustomReasonDeny is used when sending packet-in message to controller indicating
+	// that the corresponding connection has been dropped or rejected. It can be consumed
+	// by the Flow Exporter to export flow records for connections denied by network
+	// policy rules.
+	CustomReasonDeny = 0b100
 )
 
 var DispositionToString = map[uint32]string{
@@ -313,10 +318,10 @@ var (
 	// disposition of Antrea Policy. It could have more bits to support more disposition
 	// that Antrea policy support in the future.
 	APDispositionMarkRange = binding.Range{21, 22}
-	// CustomReasonMarkRange takes the 24 to 25 bits of register marksReg to indicate
+	// CustomReasonMarkRange takes the 24 to 26 bits of register marksReg to indicate
 	// the reason of sending packet to the controller. It could have more bits to
 	// support more customReason in the future.
-	CustomReasonMarkRange = binding.Range{24, 25}
+	CustomReasonMarkRange = binding.Range{24, 26}
 	// endpointIPRegRange takes a 32-bit range of register endpointIPReg to store
 	// the selected Service Endpoint IP.
 	endpointIPRegRange = binding.Range{0, 31}
@@ -354,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
@@ -376,6 +382,7 @@ func portToUint16(port int) uint16 {
 type client struct {
 	enableProxy        bool
 	enableAntreaPolicy bool
+	enableDenyTracking bool
 	enableEgress       bool
 	roundInfo          types.RoundInfo
 	cookieAllocator    cookie.Allocator
@@ -419,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
 }
 
@@ -461,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 {
@@ -486,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)
@@ -1165,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
@@ -1555,18 +1566,24 @@ func (c *client) conjunctionActionDenyFlow(conjunctionID uint32, tableID binding
 		Action().LoadRegRange(int(CNPDenyConjIDReg), conjunctionID, binding.Range{0, 31}).
 		Action().LoadRegRange(int(marksReg), cnpDenyMark, cnpDenyMarkRange)
 
+	var customReason int
+	if c.enableDenyTracking {
+		customReason += CustomReasonDeny
+		flowBuilder = flowBuilder.
+			Action().LoadRegRange(int(marksReg), disposition, APDispositionMarkRange)
+	}
 	if enableLogging {
-		customReason := CustomReasonLogging
-		if disposition == DispositionRej {
-			customReason += CustomReasonReject
-		}
+		customReason += CustomReasonLogging
 		flowBuilder = flowBuilder.
-			Action().LoadRegRange(int(marksReg), disposition, APDispositionMarkRange).
-			Action().LoadRegRange(int(marksReg), uint32(customReason), CustomReasonMarkRange).
-			Action().SendToController(uint8(PacketInReasonNP))
-	} else if disposition == DispositionRej {
+			Action().LoadRegRange(int(marksReg), disposition, APDispositionMarkRange)
+	}
+	if disposition == DispositionRej {
+		customReason += CustomReasonReject
+	}
+
+	if enableLogging || c.enableDenyTracking || disposition == DispositionRej {
 		flowBuilder = flowBuilder.
-			Action().LoadRegRange(int(marksReg), CustomReasonReject, CustomReasonMarkRange). // Do the rejection.
+			Action().LoadRegRange(int(marksReg), uint32(customReason), CustomReasonMarkRange).
 			Action().SendToController(uint8(PacketInReasonNP))
 	}
 
@@ -1715,6 +1732,15 @@ func (c *client) conjunctiveMatchFlow(tableID binding.TableIDType, matchKey *typ
 // defaultDropFlow generates the flow to drop packets if the match condition is matched.
 func (c *client) defaultDropFlow(tableID binding.TableIDType, matchKey *types.MatchKey, matchValue interface{}) binding.Flow {
 	fb := c.pipeline[tableID].BuildFlow(priorityNormal)
+	if c.enableDenyTracking {
+		return c.addFlowMatch(fb, matchKey, matchValue).
+			Action().Drop().
+			Action().LoadRegRange(int(marksReg), DispositionDrop, APDispositionMarkRange).
+			Action().LoadRegRange(int(marksReg), CustomReasonDeny, CustomReasonMarkRange).
+			Action().SendToController(uint8(PacketInReasonNP)).
+			Cookie(c.cookieAllocator.Request(cookie.Default).Raw()).
+			Done()
+	}
 	return c.addFlowMatch(fb, matchKey, matchValue).
 		Action().Drop().
 		Cookie(c.cookieAllocator.Request(cookie.Default).Raw()).
@@ -2149,7 +2175,7 @@ func (c *client) generatePipeline() {
 }
 
 // NewClient is the constructor of the Client interface.
-func NewClient(bridgeName, mgmtAddr string, ovsDatapathType ovsconfig.OVSDatapathType, enableProxy, enableAntreaPolicy, enableEgress bool) Client {
+func NewClient(bridgeName, mgmtAddr string, ovsDatapathType ovsconfig.OVSDatapathType, enableProxy, enableAntreaPolicy, enableEgress bool, enableDenyTracking bool) Client {
 	bridge := binding.NewOFBridge(bridgeName, mgmtAddr)
 	policyCache := cache.NewIndexer(
 		policyConjKeyFunc,
@@ -2159,6 +2185,7 @@ func NewClient(bridgeName, mgmtAddr string, ovsDatapathType ovsconfig.OVSDatapat
 		bridge:                   bridge,
 		enableProxy:              enableProxy,
 		enableAntreaPolicy:       enableAntreaPolicy,
+		enableDenyTracking:       enableDenyTracking,
 		enableEgress:             enableEgress,
 		nodeFlowCache:            newFlowCategoryCache(),
 		podFlowCache:             newFlowCategoryCache(),
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
 }
diff --git a/pkg/flowaggregator/flowaggregator.go b/pkg/flowaggregator/flowaggregator.go
index e10e9f093..4fa452873 100644
--- a/pkg/flowaggregator/flowaggregator.go
+++ b/pkg/flowaggregator/flowaggregator.go
@@ -24,7 +24,10 @@ import (
 	"github.com/vmware/go-ipfix/pkg/exporter"
 	ipfixintermediate "github.com/vmware/go-ipfix/pkg/intermediate"
 	ipfixregistry "github.com/vmware/go-ipfix/pkg/registry"
+	corev1 "k8s.io/api/core/v1"
+	coreinformers "k8s.io/client-go/informers/core/v1"
 	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/tools/cache"
 	"k8s.io/klog/v2"
 
 	"antrea.io/antrea/pkg/ipfix"
@@ -62,8 +65,10 @@ var (
 		"destinationServicePortName",
 		"ingressNetworkPolicyName",
 		"ingressNetworkPolicyNamespace",
+		"ingressNetworkPolicyRuleAction",
 		"egressNetworkPolicyName",
 		"egressNetworkPolicyNamespace",
+		"egressNetworkPolicyRuleAction",
 		"tcpState",
 		"flowType",
 	}
@@ -130,6 +135,7 @@ var (
 		"destinationServicePortName",
 		"ingressNetworkPolicyName",
 		"ingressNetworkPolicyNamespace",
+		"ingressNetworkPolicyRuleAction",
 		"egressNetworkPolicyName",
 		"egressNetworkPolicyNamespace",
 	}
@@ -140,6 +146,9 @@ const (
 	udpTransport         = "udp"
 	tcpTransport         = "tcp"
 	collectorAddress     = "0.0.0.0:4739"
+
+	// PodInfo index name for Pod cache.
+	podInfoIndex = "podInfo"
 )
 
 type AggregatorTransportProtocol string
@@ -166,6 +175,7 @@ type flowAggregator struct {
 	flowAggregatorAddress       string
 	k8sClient                   kubernetes.Interface
 	observationDomainID         uint32
+	podInformer                 coreinformers.PodInformer
 }
 
 func NewFlowAggregator(
@@ -177,6 +187,7 @@ func NewFlowAggregator(
 	flowAggregatorAddress string,
 	k8sClient kubernetes.Interface,
 	observationDomainID uint32,
+	podInformer coreinformers.PodInformer,
 ) *flowAggregator {
 	registry := ipfix.NewIPFIXRegistry()
 	registry.LoadRegistry()
@@ -191,10 +202,23 @@ func NewFlowAggregator(
 		flowAggregatorAddress:       flowAggregatorAddress,
 		k8sClient:                   k8sClient,
 		observationDomainID:         observationDomainID,
+		podInformer:                 podInformer,
 	}
+	podInformer.Informer().AddIndexers(cache.Indexers{podInfoIndex: podInfoIndexFunc})
 	return fa
 }
 
+func podInfoIndexFunc(obj interface{}) ([]string, error) {
+	pod, ok := obj.(*corev1.Pod)
+	if !ok {
+		return nil, fmt.Errorf("obj is not pod: %+v", obj)
+	}
+	if pod.Status.PodIP == "" {
+		klog.V(4).Infof("Pod %s/%s has not been assigned IP yet. ", pod.Namespace, pod.Name)
+	}
+	return []string{pod.Status.PodIP}, nil
+}
+
 func (fa *flowAggregator) InitCollectingProcess() error {
 	var cpInput collector.CollectorInput
 	if fa.aggregatorTransportProtocol == AggregatorTransportProtocolTLS {
@@ -382,6 +406,10 @@ func (fa *flowAggregator) sendFlowKeyRecord(key ipfixintermediate.FlowKey, recor
 	if err := fa.set.PrepareSet(ipfixentities.Data, templateID); err != nil {
 		return err
 	}
+	if !fa.aggregationProcess.IsMetadataFilled(record) {
+		fa.fillK8sMetadata(key, record.Record)
+		fa.aggregationProcess.SetMetadataFilled(record)
+	}
 	err := fa.set.AddRecord(record.Record.GetOrderedElementList(), templateID)
 	if err != nil {
 		return err
@@ -465,3 +493,54 @@ func (fa *flowAggregator) sendTemplateSet(templateSet ipfixentities.Set, isIPv6
 	bytesSent, err := fa.exportingProcess.SendSet(templateSet)
 	return bytesSent, err
 }
+
+// fillK8sMetadata fills Pod name, Pod namespace and Node name for inter-Node flows
+// that have incomplete info due to deny network policy.
+func (fa *flowAggregator) fillK8sMetadata(key ipfixintermediate.FlowKey, record ipfixentities.Record) {
+	if fa.podInformer == nil {
+		klog.Warning("No pod podInformer is provided for filling k8s metadata.")
+		return
+	}
+	// fill source Pod info when sourcePodName is empty
+	if sourcePodName, exist := record.GetInfoElementWithValue("sourcePodName"); exist {
+		if sourcePodName.Value == "" {
+			pods, err := fa.podInformer.Informer().GetIndexer().ByIndex(podInfoIndex, key.SourceAddress)
+			if err == nil && len(pods) > 0 {
+				pod, ok := pods[0].(*corev1.Pod)
+				if !ok {
+					klog.Warningf("Invalid Pod obj in cache")
+				}
+				sourcePodName.Value = pod.Name
+				if sourcePodNamespace, exist := record.GetInfoElementWithValue("sourcePodNamespace"); exist {
+					sourcePodNamespace.Value = pod.Namespace
+				}
+				if sourceNodeName, exist := record.GetInfoElementWithValue("sourceNodeName"); exist {
+					sourceNodeName.Value = pod.Spec.NodeName
+				}
+			} else {
+				klog.Warning(err)
+			}
+		}
+	}
+	// fill destination Pod info when destinationPodName is empty
+	if destinationPodName, exist := record.GetInfoElementWithValue("destinationPodName"); exist {
+		if destinationPodName.Value == "" {
+			pods, err := fa.podInformer.Informer().GetIndexer().ByIndex(podInfoIndex, key.DestinationAddress)
+			if len(pods) > 0 && err == nil {
+				pod, ok := pods[0].(*corev1.Pod)
+				if !ok {
+					klog.Warningf("Invalid Pod obj in cache")
+				}
+				destinationPodName.Value = pod.Name
+				if destinationPodNamespace, exist := record.GetInfoElementWithValue("destinationPodNamespace"); exist {
+					destinationPodNamespace.Value = pod.Namespace
+				}
+				if destinationNodeName, exist := record.GetInfoElementWithValue("destinationNodeName"); exist {
+					destinationNodeName.Value = pod.Spec.NodeName
+				}
+			} else {
+				klog.Warning(err)
+			}
+		}
+	}
+}
diff --git a/pkg/ipfix/ipfix_intermediate.go b/pkg/ipfix/ipfix_intermediate.go
index 2fb59b2be..db9941567 100644
--- a/pkg/ipfix/ipfix_intermediate.go
+++ b/pkg/ipfix/ipfix_intermediate.go
@@ -31,6 +31,8 @@ type IPFIXAggregationProcess interface {
 	ForAllExpiredFlowRecordsDo(callback ipfixintermediate.FlowKeyRecordMapCallBack) error
 	GetExpiryFromExpirePriorityQueue() time.Duration
 	ResetStatElementsInRecord(record ipfixentities.Record) error
+	SetMetadataFilled(record ipfixintermediate.AggregationFlowRecord)
+	IsMetadataFilled(record ipfixintermediate.AggregationFlowRecord) bool
 }
 
 type ipfixAggregationProcess struct {
@@ -68,3 +70,11 @@ func (ap *ipfixAggregationProcess) GetExpiryFromExpirePriorityQueue() time.Durat
 func (ap *ipfixAggregationProcess) ResetStatElementsInRecord(record ipfixentities.Record) error {
 	return ap.AggregationProcess.ResetStatElementsInRecord(record)
 }
+
+func (ap *ipfixAggregationProcess) SetMetadataFilled(record ipfixintermediate.AggregationFlowRecord) {
+	ap.AggregationProcess.SetMetadataFilled(record, true)
+}
+
+func (ap *ipfixAggregationProcess) IsMetadataFilled(record ipfixintermediate.AggregationFlowRecord) bool {
+	return ap.AggregationProcess.IsMetadataFilled(record)
+}


Test patch:
diff --git a/ci/kind/test-e2e-kind.sh b/ci/kind/test-e2e-kind.sh
index c8ec68d2f..4e44a3c0c 100755
--- a/ci/kind/test-e2e-kind.sh
+++ b/ci/kind/test-e2e-kind.sh
@@ -103,7 +103,7 @@ else
     manifest_args="$manifest_args --no-np"
 fi
 
-COMMON_IMAGES_LIST=("gcr.io/kubernetes-e2e-test-images/agnhost:2.8" "projects.registry.vmware.com/library/busybox" "projects.registry.vmware.com/antrea/nginx" "projects.registry.vmware.com/antrea/perftool" "projects.registry.vmware.com/antrea/ipfix-collector:v0.4.7")
+COMMON_IMAGES_LIST=("gcr.io/kubernetes-e2e-test-images/agnhost:2.8" "projects.registry.vmware.com/library/busybox" "projects.registry.vmware.com/antrea/nginx" "projects.registry.vmware.com/antrea/perftool" "projects.registry.vmware.com/antrea/ipfix-collector:v0.5.2")
 for image in "${COMMON_IMAGES_LIST[@]}"; do
     for i in `seq 3`; do
         docker pull $image && break
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/networkpolicy/networkpolicy_controller_test.go b/pkg/agent/controller/networkpolicy/networkpolicy_controller_test.go
index 4b6e7a827..5adc77176 100644
--- a/pkg/agent/controller/networkpolicy/networkpolicy_controller_test.go
+++ b/pkg/agent/controller/networkpolicy/networkpolicy_controller_test.go
@@ -53,7 +53,7 @@ func newTestController() (*Controller, *fake.Clientset, *mockReconciler) {
 	clientset := &fake.Clientset{}
 	ch := make(chan agenttypes.EntityReference, 100)
 	controller, _ := NewNetworkPolicyController(&antreaClientGetter{clientset}, nil, nil, "node1", ch,
-		true, true, true, testAsyncDeleteInterval)
+		true, true, true, nil, testAsyncDeleteInterval)
 	reconciler := newMockReconciler()
 	controller.reconciler = reconciler
 	return controller, clientset, reconciler
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/flowexporter/connections/connections_test.go b/pkg/agent/flowexporter/connections/connections_test.go
index 3b62dd85b..0c9417f9e 100644
--- a/pkg/agent/flowexporter/connections/connections_test.go
+++ b/pkg/agent/flowexporter/connections/connections_test.go
@@ -15,262 +15,19 @@
 package connections
 
 import (
-	"encoding/binary"
-	"fmt"
 	"net"
-	"strings"
 	"testing"
 	"time"
 
 	"github.com/golang/mock/gomock"
-	"github.com/prometheus/client_golang/prometheus/testutil"
 	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-	v1 "k8s.io/api/core/v1"
-	"k8s.io/apimachinery/pkg/types"
-	"k8s.io/component-base/metrics/legacyregistry"
 
 	"antrea.io/antrea/pkg/agent/flowexporter"
-	connectionstest "antrea.io/antrea/pkg/agent/flowexporter/connections/testing"
-	"antrea.io/antrea/pkg/agent/flowexporter/flowrecords"
-	"antrea.io/antrea/pkg/agent/interfacestore"
 	interfacestoretest "antrea.io/antrea/pkg/agent/interfacestore/testing"
-	"antrea.io/antrea/pkg/agent/metrics"
-	"antrea.io/antrea/pkg/agent/openflow"
-	proxytest "antrea.io/antrea/pkg/agent/proxy/testing"
-	cpv1beta "antrea.io/antrea/pkg/apis/controlplane/v1beta2"
-	queriertest "antrea.io/antrea/pkg/querier/testing"
-	k8sproxy "antrea.io/antrea/third_party/proxy"
-)
-
-var (
-	np1 = cpv1beta.NetworkPolicyReference{
-		Type:      cpv1beta.K8sNetworkPolicy,
-		Namespace: "foo",
-		Name:      "bar",
-		UID:       "uid1",
-	}
-	np2 = cpv1beta.NetworkPolicyReference{
-		Type:      cpv1beta.K8sNetworkPolicy,
-		Namespace: "foo",
-		Name:      "baz",
-		UID:       "uid2",
-	}
 )
 
 const testPollInterval = 0 // Not used in these tests, hence 0.
 
-func makeTuple(srcIP *net.IP, dstIP *net.IP, protoID uint8, srcPort uint16, dstPort uint16) (flowexporter.Tuple, flowexporter.Tuple) {
-	tuple := flowexporter.Tuple{
-		SourceAddress:      *srcIP,
-		DestinationAddress: *dstIP,
-		Protocol:           protoID,
-		SourcePort:         srcPort,
-		DestinationPort:    dstPort,
-	}
-	revTuple := flowexporter.Tuple{
-		SourceAddress:      *dstIP,
-		DestinationAddress: *srcIP,
-		Protocol:           protoID,
-		SourcePort:         dstPort,
-		DestinationPort:    srcPort,
-	}
-	return tuple, revTuple
-}
-
-func TestConnectionStore_addAndUpdateConn(t *testing.T) {
-	ctrl := gomock.NewController(t)
-	defer ctrl.Finish()
-	metrics.InitializeConnectionMetrics()
-	// Create three flows; two are already in connectionStore and another one is new
-	refTime := time.Now()
-	// Flow-1, which is already in connectionStore
-	tuple1, revTuple1 := makeTuple(&net.IP{1, 2, 3, 4}, &net.IP{4, 3, 2, 1}, 6, 65280, 255)
-	testFlow1 := flowexporter.Connection{
-		StartTime:       refTime.Add(-(time.Second * 50)),
-		StopTime:        refTime,
-		OriginalPackets: 0xffff,
-		OriginalBytes:   0xbaaaaa0000000000,
-		ReversePackets:  0xff,
-		ReverseBytes:    0xbaaa,
-		TupleOrig:       tuple1,
-		TupleReply:      revTuple1,
-		IsPresent:       true,
-	}
-	// Flow-2, which is not in connectionStore
-	tuple2, revTuple2 := makeTuple(&net.IP{5, 6, 7, 8}, &net.IP{8, 7, 6, 5}, 6, 60001, 200)
-	testFlow2 := flowexporter.Connection{
-		StartTime:       refTime.Add(-(time.Second * 20)),
-		StopTime:        refTime,
-		OriginalPackets: 0xbb,
-		OriginalBytes:   0xcbbb,
-		ReversePackets:  0xbbbb,
-		ReverseBytes:    0xcbbbb0000000000,
-		TupleOrig:       tuple2,
-		TupleReply:      revTuple2,
-		IsPresent:       true,
-	}
-	// Flow-3 , which is already in connectionStore
-	tuple3, revTuple3 := makeTuple(&net.IP{10, 10, 10, 10}, &net.IP{4, 3, 2, 1}, 6, 60000, 100)
-	testFlow3 := flowexporter.Connection{
-		StartTime:       refTime.Add(-(time.Second * 50)),
-		StopTime:        refTime,
-		OriginalPackets: 0xffff,
-		OriginalBytes:   0xbaaaaa,
-		ReversePackets:  0xff,
-		ReverseBytes:    0xbaaa,
-		TupleOrig:       tuple3,
-		TupleReply:      revTuple3,
-		TCPState:        "TIME_WAIT",
-		IsPresent:       true,
-	}
-	// To test service name mapping.
-	tuple4, revTuple4 := makeTuple(&net.IP{10, 10, 10, 10}, &net.IP{20, 20, 20, 20}, 6, 5000, 80)
-	testFlow4 := flowexporter.Connection{
-		TupleOrig:  tuple4,
-		TupleReply: revTuple4,
-		Mark:       openflow.ServiceCTMark,
-		IsPresent:  true,
-	}
-	// To test NetworkPolicy mapping.
-	tuple5, revTuple5 := makeTuple(&net.IP{30, 30, 30, 30}, &net.IP{20, 20, 20, 20}, 6, 5000, 80)
-	testFlow5 := flowexporter.Connection{
-		TupleOrig:  tuple5,
-		TupleReply: revTuple5,
-		Labels:     []byte{0x0, 0x0, 0x0, 0x1, 0x0, 0x0, 0x0, 0x2},
-		IsPresent:  true,
-	}
-	// Create copy of old conntrack flow for testing purposes.
-	// These two flows are already in connection store.
-	oldTestFlow1 := flowexporter.Connection{
-		StartTime:               testFlow1.StartTime,
-		StopTime:                testFlow1.StopTime.Add(-(time.Second * 30)),
-		OriginalPackets:         0xfff,
-		OriginalBytes:           0xbaaaaa00000000,
-		ReversePackets:          0xf,
-		ReverseBytes:            0xba,
-		TupleOrig:               tuple1,
-		TupleReply:              revTuple1,
-		SourcePodNamespace:      "ns1",
-		SourcePodName:           "pod1",
-		DestinationPodNamespace: "",
-		DestinationPodName:      "",
-		IsPresent:               true,
-		TCPState:                "",
-	}
-	oldTestFlow3 := flowexporter.Connection{
-		StartTime:               testFlow3.StartTime,
-		StopTime:                testFlow3.StopTime.Add(-(time.Second * 30)),
-		OriginalPackets:         0xffff,
-		OriginalBytes:           0xbaaaaa,
-		ReversePackets:          0xff,
-		ReverseBytes:            0xbaaa,
-		TupleOrig:               tuple3,
-		TupleReply:              revTuple3,
-		SourcePodNamespace:      "ns3",
-		SourcePodName:           "pod3",
-		DestinationPodNamespace: "",
-		DestinationPodName:      "",
-		IsPresent:               true,
-		TCPState:                "TIME_WAIT",
-	}
-	podConfigFlow2 := &interfacestore.ContainerInterfaceConfig{
-		ContainerID:  "2",
-		PodName:      "pod2",
-		PodNamespace: "ns2",
-	}
-	interfaceFlow2 := &interfacestore.InterfaceConfig{
-		InterfaceName:            "interface2",
-		IPs:                      []net.IP{{8, 7, 6, 5}},
-		ContainerInterfaceConfig: podConfigFlow2,
-	}
-	servicePortName := k8sproxy.ServicePortName{
-		NamespacedName: types.NamespacedName{
-			Namespace: "serviceNS1",
-			Name:      "service1",
-		},
-		Port:     "255",
-		Protocol: v1.ProtocolTCP,
-	}
-	// Mock interface store with one of the couple of IPs correspond to Pods
-	mockIfaceStore := interfacestoretest.NewMockInterfaceStore(ctrl)
-	mockConnDumper := connectionstest.NewMockConnTrackDumper(ctrl)
-	mockProxier := proxytest.NewMockProxier(ctrl)
-	npQuerier := queriertest.NewMockAgentNetworkPolicyInfoQuerier(ctrl)
-	connStore := NewConnectionStore(mockConnDumper, flowrecords.NewFlowRecords(), mockIfaceStore, true, false, mockProxier, npQuerier, testPollInterval)
-
-	// Add flow1conn and flow3conn to the Connection map
-	testFlow1Tuple := flowexporter.NewConnectionKey(&testFlow1)
-	connStore.connections[testFlow1Tuple] = &oldTestFlow1
-	testFlow3Tuple := flowexporter.NewConnectionKey(&testFlow3)
-	connStore.connections[testFlow3Tuple] = &oldTestFlow3
-	// For testing purposes, increment the metric
-	metrics.TotalAntreaConnectionsInConnTrackTable.Inc()
-	metrics.TotalAntreaConnectionsInConnTrackTable.Inc()
-
-	addOrUpdateConnTests := []struct {
-		flow flowexporter.Connection
-	}{
-		{testFlow1}, // To test update part of function.
-		{testFlow2}, // To test add part of function.
-		{testFlow3}, // To test update part of function for dying connection.
-		{testFlow4}, // To test service name mapping.
-		{testFlow5}, // To test NetworkPolicy mapping.
-	}
-	for i, test := range addOrUpdateConnTests {
-		flowTuple := flowexporter.NewConnectionKey(&test.flow)
-		expConn := test.flow
-		switch i {
-		case 0:
-			// Tests update part of the function.
-			expConn.SourcePodNamespace = "ns1"
-			expConn.SourcePodName = "pod1"
-		case 1:
-			// Tests add part of the function.
-			mockIfaceStore.EXPECT().GetInterfaceByIP(test.flow.TupleOrig.SourceAddress.String()).Return(nil, false)
-			mockIfaceStore.EXPECT().GetInterfaceByIP(test.flow.TupleReply.SourceAddress.String()).Return(interfaceFlow2, true)
-
-			expConn.DestinationPodNamespace = "ns2"
-			expConn.DestinationPodName = "pod2"
-		case 2:
-			// Tests update part of the function for dying connection.
-
-			expConn.SourcePodNamespace = "ns3"
-			expConn.SourcePodName = "pod3"
-			expConn.TCPState = "TIME_WAIT"
-			expConn.StopTime = refTime.Add(-(time.Second * 30))
-		case 3:
-			// Tests service name mapping.
-			mockIfaceStore.EXPECT().GetInterfaceByIP(expConn.TupleOrig.SourceAddress.String()).Return(nil, false)
-			mockIfaceStore.EXPECT().GetInterfaceByIP(expConn.TupleReply.SourceAddress.String()).Return(nil, false)
-
-			protocol, _ := lookupServiceProtocol(expConn.TupleOrig.Protocol)
-			serviceStr := fmt.Sprintf("%s:%d/%s", expConn.TupleOrig.DestinationAddress.String(), expConn.TupleOrig.DestinationPort, protocol)
-			mockProxier.EXPECT().GetServiceByIP(serviceStr).Return(servicePortName, true)
-			expConn.DestinationServicePortName = servicePortName.String()
-		case 4:
-			// Tests NetworkPolicy mapping.
-			mockIfaceStore.EXPECT().GetInterfaceByIP(expConn.TupleOrig.SourceAddress.String()).Return(nil, false)
-			mockIfaceStore.EXPECT().GetInterfaceByIP(expConn.TupleReply.SourceAddress.String()).Return(nil, false)
-
-			ingressOfID := binary.LittleEndian.Uint32(test.flow.Labels[:4])
-			npQuerier.EXPECT().GetNetworkPolicyByRuleFlowID(ingressOfID).Return(&np1)
-			expConn.IngressNetworkPolicyName = np1.Name
-			expConn.IngressNetworkPolicyNamespace = np1.Namespace
-
-			egressOfID := binary.LittleEndian.Uint32(test.flow.Labels[4:8])
-			npQuerier.EXPECT().GetNetworkPolicyByRuleFlowID(egressOfID).Return(&np2)
-			expConn.EgressNetworkPolicyName = np2.Name
-			expConn.EgressNetworkPolicyNamespace = np2.Namespace
-		}
-		connStore.addOrUpdateConn(&test.flow)
-		actualConn, ok := connStore.GetConnByKey(flowTuple)
-		assert.Equal(t, ok, true, "connection should be there in connection store")
-		assert.Equal(t, expConn, *actualConn, "Connections should be equal")
-		checkAntreaConnectionMetrics(t, len(connStore.connections))
-	}
-}
-
 func TestConnectionStore_ForAllConnectionsDo(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()
@@ -279,7 +36,7 @@ func TestConnectionStore_ForAllConnectionsDo(t *testing.T) {
 	testFlowKeys := make([]*flowexporter.ConnectionKey, 2)
 	refTime := time.Now()
 	// Flow-1, which is already in connectionStore
-	tuple1, revTuple1 := makeTuple(&net.IP{1, 2, 3, 4}, &net.IP{4, 3, 2, 1}, 6, 65280, 255)
+	tuple1 := flowexporter.Tuple{SourceAddress: net.IP{1, 2, 3, 4}, DestinationAddress: net.IP{4, 3, 2, 1}, Protocol: 6, SourcePort: 65280, DestinationPort: 255}
 	testFlows[0] = &flowexporter.Connection{
 		StartTime:       refTime.Add(-(time.Second * 50)),
 		StopTime:        refTime,
@@ -287,12 +44,11 @@ func TestConnectionStore_ForAllConnectionsDo(t *testing.T) {
 		OriginalBytes:   0xbaaaaa0000000000,
 		ReversePackets:  0xff,
 		ReverseBytes:    0xbaaa,
-		TupleOrig:       tuple1,
-		TupleReply:      revTuple1,
+		FlowKey:         tuple1,
 		IsPresent:       true,
 	}
 	// Flow-2, which is not in connectionStore
-	tuple2, revTuple2 := makeTuple(&net.IP{5, 6, 7, 8}, &net.IP{8, 7, 6, 5}, 6, 60001, 200)
+	tuple2 := flowexporter.Tuple{SourceAddress: net.IP{5, 6, 7, 8}, DestinationAddress: net.IP{8, 7, 6, 5}, Protocol: 6, SourcePort: 60001, DestinationPort: 200}
 	testFlows[1] = &flowexporter.Connection{
 		StartTime:       refTime.Add(-(time.Second * 20)),
 		StopTime:        refTime,
@@ -300,8 +56,7 @@ func TestConnectionStore_ForAllConnectionsDo(t *testing.T) {
 		OriginalBytes:   0xcbbb,
 		ReversePackets:  0xbbbb,
 		ReverseBytes:    0xcbbbb0000000000,
-		TupleOrig:       tuple2,
-		TupleReply:      revTuple2,
+		FlowKey:         tuple2,
 		IsPresent:       true,
 	}
 	for i, flow := range testFlows {
@@ -310,8 +65,7 @@ func TestConnectionStore_ForAllConnectionsDo(t *testing.T) {
 	}
 	// Create connectionStore
 	mockIfaceStore := interfacestoretest.NewMockInterfaceStore(ctrl)
-	mockConnDumper := connectionstest.NewMockConnTrackDumper(ctrl)
-	connStore := NewConnectionStore(mockConnDumper, flowrecords.NewFlowRecords(), mockIfaceStore, true, false, nil, nil, testPollInterval)
+	connStore := NewConnectionStore(mockIfaceStore, nil)
 	// Add flows to the Connection store
 	for i, flow := range testFlows {
 		connStore.connections[*testFlowKeys[i]] = flow
@@ -331,114 +85,3 @@ func TestConnectionStore_ForAllConnectionsDo(t *testing.T) {
 		assert.Equal(t, conn.OriginalPackets, uint64(0), "OriginalPackets should be reset")
 	}
 }
-
-func TestConnectionStore_DeleteConnectionByKey(t *testing.T) {
-	ctrl := gomock.NewController(t)
-	defer ctrl.Finish()
-	metrics.InitializeConnectionMetrics()
-	// Create two flows; one is already in connectionStore and other one is new
-	testFlows := make([]*flowexporter.Connection, 2)
-	testFlowKeys := make([]*flowexporter.ConnectionKey, 2)
-	refTime := time.Now()
-	// Flow-1, which is already in connectionStore
-	tuple1, revTuple1 := makeTuple(&net.IP{1, 2, 3, 4}, &net.IP{4, 3, 2, 1}, 6, 65280, 255)
-	testFlows[0] = &flowexporter.Connection{
-		StartTime:       refTime.Add(-(time.Second * 50)),
-		StopTime:        refTime,
-		OriginalPackets: 0xffff,
-		OriginalBytes:   0xbaaaaa0000000000,
-		ReversePackets:  0xff,
-		ReverseBytes:    0xbaaa,
-		TupleOrig:       tuple1,
-		TupleReply:      revTuple1,
-		IsPresent:       true,
-	}
-	// Flow-2, which is not in connectionStore
-	tuple2, revTuple2 := makeTuple(&net.IP{5, 6, 7, 8}, &net.IP{8, 7, 6, 5}, 6, 60001, 200)
-	testFlows[1] = &flowexporter.Connection{
-		StartTime:       refTime.Add(-(time.Second * 20)),
-		StopTime:        refTime,
-		OriginalPackets: 0xbb,
-		OriginalBytes:   0xcbbb,
-		ReversePackets:  0xbbbb,
-		ReverseBytes:    0xcbbbb0000000000,
-		TupleOrig:       tuple2,
-		TupleReply:      revTuple2,
-		IsPresent:       true,
-	}
-	for i, flow := range testFlows {
-		connKey := flowexporter.NewConnectionKey(flow)
-		testFlowKeys[i] = &connKey
-	}
-	// For testing purposes, set the metric
-	metrics.TotalAntreaConnectionsInConnTrackTable.Set(float64(len(testFlows)))
-	// Create connectionStore
-	mockIfaceStore := interfacestoretest.NewMockInterfaceStore(ctrl)
-	mockConnDumper := connectionstest.NewMockConnTrackDumper(ctrl)
-	connStore := NewConnectionStore(mockConnDumper, flowrecords.NewFlowRecords(), mockIfaceStore, true, false, nil, nil, testPollInterval)
-	// Add flows to the connection store.
-	for i, flow := range testFlows {
-		connStore.connections[*testFlowKeys[i]] = flow
-	}
-	// Delete the connections in connection store.
-	for i := 0; i < len(testFlows); i++ {
-		err := connStore.deleteConnectionByKeyWithoutLock(*testFlowKeys[i])
-		assert.Nil(t, err, "DeleteConnectionByKey should return nil")
-		_, exists := connStore.GetConnByKey(*testFlowKeys[i])
-		assert.Equal(t, exists, false, "connection should be deleted in connection store")
-		checkAntreaConnectionMetrics(t, len(connStore.connections))
-	}
-}
-
-func TestConnectionStore_MetricSettingInPoll(t *testing.T) {
-	ctrl := gomock.NewController(t)
-	defer ctrl.Finish()
-	metrics.InitializeConnectionMetrics()
-
-	testFlows := make([]*flowexporter.Connection, 0)
-	// Create connectionStore
-	mockIfaceStore := interfacestoretest.NewMockInterfaceStore(ctrl)
-	mockConnDumper := connectionstest.NewMockConnTrackDumper(ctrl)
-	connStore := NewConnectionStore(mockConnDumper, flowrecords.NewFlowRecords(), mockIfaceStore, true, false, nil, nil, testPollInterval)
-	// Hard-coded conntrack occupancy metrics for test
-	TotalConnections := 0
-	MaxConnections := 300000
-	mockConnDumper.EXPECT().DumpFlows(uint16(openflow.CtZone)).Return(testFlows, TotalConnections, nil)
-	mockConnDumper.EXPECT().GetMaxConnections().Return(MaxConnections, nil)
-	connsLens, err := connStore.Poll()
-	require.Nil(t, err, fmt.Sprintf("Failed to add connections to connection store: %v", err))
-	assert.Equal(t, len(connsLens), 1, "length of connsLens is expected to be 1")
-	assert.Equal(t, connsLens[0], len(testFlows), "expected connections should be equal to number of testFlows")
-	checkTotalConnectionsMetric(t, TotalConnections)
-	checkMaxConnectionsMetric(t, MaxConnections)
-}
-
-func checkAntreaConnectionMetrics(t *testing.T, numConns int) {
-	expectedAntreaConnectionCount := `
-	# HELP antrea_agent_conntrack_antrea_connection_count [ALPHA] Number of connections in the Antrea ZoneID of the conntrack table. This metric gets updated at an interval specified by flowPollInterval, a configuration parameter for the Agent.
-	# TYPE antrea_agent_conntrack_antrea_connection_count gauge
-	`
-	expectedAntreaConnectionCount = expectedAntreaConnectionCount + fmt.Sprintf("antrea_agent_conntrack_antrea_connection_count %d\n", numConns)
-	err := testutil.GatherAndCompare(legacyregistry.DefaultGatherer, strings.NewReader(expectedAntreaConnectionCount), "antrea_agent_conntrack_antrea_connection_count")
-	assert.NoError(t, err)
-}
-
-func checkTotalConnectionsMetric(t *testing.T, numConns int) {
-	expectedConnectionCount := `
-	# HELP antrea_agent_conntrack_total_connection_count [ALPHA] Number of connections in the conntrack table. This metric gets updated at an interval specified by flowPollInterval, a configuration parameter for the Agent.
-	# TYPE antrea_agent_conntrack_total_connection_count gauge
-	`
-	expectedConnectionCount = expectedConnectionCount + fmt.Sprintf("antrea_agent_conntrack_total_connection_count %d\n", numConns)
-	err := testutil.GatherAndCompare(legacyregistry.DefaultGatherer, strings.NewReader(expectedConnectionCount), "antrea_agent_conntrack_total_connection_count")
-	assert.NoError(t, err)
-}
-
-func checkMaxConnectionsMetric(t *testing.T, maxConns int) {
-	expectedMaxConnectionsCount := `
-	# HELP antrea_agent_conntrack_max_connection_count [ALPHA] Size of the conntrack table. This metric gets updated at an interval specified by flowPollInterval, a configuration parameter for the Agent.
-	# TYPE antrea_agent_conntrack_max_connection_count gauge
-	`
-	expectedMaxConnectionsCount = expectedMaxConnectionsCount + fmt.Sprintf("antrea_agent_conntrack_max_connection_count %d\n", maxConns)
-	err := testutil.GatherAndCompare(legacyregistry.DefaultGatherer, strings.NewReader(expectedMaxConnectionsCount), "antrea_agent_conntrack_max_connection_count")
-	assert.NoError(t, err)
-}
diff --git a/pkg/agent/flowexporter/connections/conntrack_connections_test.go b/pkg/agent/flowexporter/connections/conntrack_connections_test.go
new file mode 100644
index 000000000..8e2b8bead
--- /dev/null
+++ b/pkg/agent/flowexporter/connections/conntrack_connections_test.go
@@ -0,0 +1,318 @@
+// Copyright 2021 Antrea Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package connections
+
+import (
+	"fmt"
+	"net"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/golang/mock/gomock"
+	"github.com/prometheus/client_golang/prometheus/testutil"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	v1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/component-base/metrics/legacyregistry"
+
+	"antrea.io/antrea/pkg/agent/flowexporter"
+	connectionstest "antrea.io/antrea/pkg/agent/flowexporter/connections/testing"
+	"antrea.io/antrea/pkg/agent/flowexporter/flowrecords"
+	"antrea.io/antrea/pkg/agent/interfacestore"
+	interfacestoretest "antrea.io/antrea/pkg/agent/interfacestore/testing"
+	"antrea.io/antrea/pkg/agent/metrics"
+	"antrea.io/antrea/pkg/agent/openflow"
+	proxytest "antrea.io/antrea/pkg/agent/proxy/testing"
+	queriertest "antrea.io/antrea/pkg/querier/testing"
+	k8sproxy "antrea.io/antrea/third_party/proxy"
+)
+
+func TestConntrackConnectionStore_AddOrUpdateConn(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+	metrics.InitializeConnectionMetrics()
+	// Create three flows; two are already in connectionStore and another one is new
+	refTime := time.Now()
+	// Flow-1, which is already in connectionStore
+	tuple1 := flowexporter.Tuple{SourceAddress: net.IP{1, 2, 3, 4}, DestinationAddress: net.IP{4, 3, 2, 1}, Protocol: 6, SourcePort: 65280, DestinationPort: 255}
+	testFlow1 := flowexporter.Connection{
+		StartTime:       refTime.Add(-(time.Second * 50)),
+		StopTime:        refTime,
+		OriginalPackets: 0xffff,
+		OriginalBytes:   0xbaaaaa0000000000,
+		ReversePackets:  0xff,
+		ReverseBytes:    0xbaaa,
+		FlowKey:         tuple1,
+		IsPresent:       true,
+	}
+	// Flow-2, which is not in connectionStore
+	tuple2 := flowexporter.Tuple{SourceAddress: net.IP{5, 6, 7, 8}, DestinationAddress: net.IP{8, 7, 6, 5}, Protocol: 6, SourcePort: 60001, DestinationPort: 200}
+	testFlow2 := flowexporter.Connection{
+		StartTime:       refTime.Add(-(time.Second * 20)),
+		StopTime:        refTime,
+		OriginalPackets: 0xbb,
+		OriginalBytes:   0xcbbb,
+		ReversePackets:  0xbbbb,
+		ReverseBytes:    0xcbbbb0000000000,
+		FlowKey:         tuple2,
+		IsPresent:       true,
+	}
+	// Flow-3 , which is already in connectionStore
+	tuple3 := flowexporter.Tuple{SourceAddress: net.IP{10, 10, 10, 10}, DestinationAddress: net.IP{4, 3, 2, 1}, Protocol: 6, SourcePort: 60000, DestinationPort: 100}
+	testFlow3 := flowexporter.Connection{
+		StartTime:                 refTime.Add(-(time.Second * 50)),
+		StopTime:                  refTime,
+		OriginalPackets:           0xffff,
+		OriginalBytes:             0xbaaaaa,
+		ReversePackets:            0xff,
+		ReverseBytes:              0xbaaa,
+		FlowKey:                   tuple3,
+		DestinationServiceAddress: tuple3.DestinationAddress,
+		DestinationServicePort:    tuple3.DestinationPort,
+		TCPState:                  "TIME_WAIT",
+		IsPresent:                 true,
+	}
+	// To test service name mapping.
+	tuple4 := flowexporter.Tuple{SourceAddress: net.IP{10, 10, 10, 10}, DestinationAddress: net.IP{20, 20, 20, 20}, Protocol: 6, SourcePort: 5000, DestinationPort: 80}
+	testFlow4 := flowexporter.Connection{
+		FlowKey:   tuple4,
+		Mark:      openflow.ServiceCTMark,
+		IsPresent: true,
+	}
+	// Create copy of old conntrack flow for testing purposes.
+	// These two flows are already in connection store.
+	oldTestFlow1 := flowexporter.Connection{
+		StartTime:               testFlow1.StartTime,
+		StopTime:                testFlow1.StopTime.Add(-(time.Second * 30)),
+		OriginalPackets:         0xfff,
+		OriginalBytes:           0xbaaaaa00000000,
+		ReversePackets:          0xf,
+		ReverseBytes:            0xba,
+		FlowKey:                 tuple1,
+		SourcePodNamespace:      "ns1",
+		SourcePodName:           "pod1",
+		DestinationPodNamespace: "",
+		DestinationPodName:      "",
+		IsPresent:               true,
+		TCPState:                "",
+	}
+	oldTestFlow3 := flowexporter.Connection{
+		StartTime:                 testFlow3.StartTime,
+		StopTime:                  testFlow3.StopTime.Add(-(time.Second * 30)),
+		OriginalPackets:           0xffff,
+		OriginalBytes:             0xbaaaaa,
+		ReversePackets:            0xff,
+		ReverseBytes:              0xbaaa,
+		FlowKey:                   tuple3,
+		DestinationServiceAddress: tuple3.DestinationAddress,
+		DestinationServicePort:    tuple3.DestinationPort,
+		SourcePodNamespace:        "ns3",
+		SourcePodName:             "pod3",
+		DestinationPodNamespace:   "",
+		DestinationPodName:        "",
+		IsPresent:                 true,
+		TCPState:                  "TIME_WAIT",
+	}
+	podConfigFlow2 := &interfacestore.ContainerInterfaceConfig{
+		ContainerID:  "2",
+		PodName:      "pod2",
+		PodNamespace: "ns2",
+	}
+	interfaceFlow2 := &interfacestore.InterfaceConfig{
+		InterfaceName:            "interface2",
+		IPs:                      []net.IP{{8, 7, 6, 5}},
+		ContainerInterfaceConfig: podConfigFlow2,
+	}
+	servicePortName := k8sproxy.ServicePortName{
+		NamespacedName: types.NamespacedName{
+			Namespace: "serviceNS1",
+			Name:      "service1",
+		},
+		Port:     "255",
+		Protocol: v1.ProtocolTCP,
+	}
+	// Mock interface store with one of the couple of IPs correspond to Pods
+	mockIfaceStore := interfacestoretest.NewMockInterfaceStore(ctrl)
+	mockProxier := proxytest.NewMockProxier(ctrl)
+	mockConnDumper := connectionstest.NewMockConnTrackDumper(ctrl)
+	npQuerier := queriertest.NewMockAgentNetworkPolicyInfoQuerier(ctrl)
+	conntrackConnStore := NewConntrackConnectionStore(mockConnDumper, flowrecords.NewFlowRecords(), mockIfaceStore, true, false, mockProxier, npQuerier, testPollInterval)
+
+	// Add flow1conn and flow3conn to the Connection map
+	testFlow1Tuple := flowexporter.NewConnectionKey(&testFlow1)
+	conntrackConnStore.connections[testFlow1Tuple] = &oldTestFlow1
+	testFlow3Tuple := flowexporter.NewConnectionKey(&testFlow3)
+	conntrackConnStore.connections[testFlow3Tuple] = &oldTestFlow3
+	// For testing purposes, increment the metric
+	metrics.TotalAntreaConnectionsInConnTrackTable.Inc()
+	metrics.TotalAntreaConnectionsInConnTrackTable.Inc()
+
+	addOrUpdateConnTests := []struct {
+		flow flowexporter.Connection
+	}{
+		{testFlow1}, // To test update part of function.
+		{testFlow2}, // To test add part of function.
+		{testFlow3}, // To test update part of function for dying connection.
+		{testFlow4}, // To test service name mapping.
+	}
+	for i, test := range addOrUpdateConnTests {
+		flowTuple := flowexporter.NewConnectionKey(&test.flow)
+		expConn := test.flow
+		switch i {
+		case 0:
+			// Tests update part of the function.
+			expConn.SourcePodNamespace = "ns1"
+			expConn.SourcePodName = "pod1"
+		case 1:
+			// Tests add part of the function.
+			mockIfaceStore.EXPECT().GetInterfaceByIP(test.flow.FlowKey.SourceAddress.String()).Return(nil, false)
+			mockIfaceStore.EXPECT().GetInterfaceByIP(test.flow.FlowKey.DestinationAddress.String()).Return(interfaceFlow2, true)
+
+			expConn.DestinationPodNamespace = "ns2"
+			expConn.DestinationPodName = "pod2"
+		case 2:
+			// Tests update part of the function for dying connection.
+
+			expConn.SourcePodNamespace = "ns3"
+			expConn.SourcePodName = "pod3"
+			expConn.TCPState = "TIME_WAIT"
+			expConn.StopTime = refTime.Add(-(time.Second * 30))
+		case 3:
+			// Tests service name mapping.
+			mockIfaceStore.EXPECT().GetInterfaceByIP(expConn.FlowKey.SourceAddress.String()).Return(nil, false)
+			mockIfaceStore.EXPECT().GetInterfaceByIP(expConn.FlowKey.DestinationAddress.String()).Return(nil, false)
+
+			protocol, _ := lookupServiceProtocol(expConn.FlowKey.Protocol)
+			serviceStr := fmt.Sprintf("%s:%d/%s", expConn.DestinationServiceAddress.String(), expConn.DestinationServicePort, protocol)
+			mockProxier.EXPECT().GetServiceByIP(serviceStr).Return(servicePortName, true)
+			expConn.DestinationServicePortName = servicePortName.String()
+		}
+		conntrackConnStore.AddOrUpdateConn(&test.flow)
+		actualConn, ok := conntrackConnStore.GetConnByKey(flowTuple)
+		assert.Equal(t, ok, true, "connection should be there in connection store")
+		assert.Equal(t, expConn, *actualConn, "Connections should be equal")
+		checkAntreaConnectionMetrics(t, len(conntrackConnStore.connections))
+	}
+}
+
+func TestConnectionStore_DeleteConnectionByKey(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+	metrics.InitializeConnectionMetrics()
+	// Create two flows; one is already in connectionStore and other one is new
+	testFlows := make([]*flowexporter.Connection, 2)
+	testFlowKeys := make([]*flowexporter.ConnectionKey, 2)
+	refTime := time.Now()
+	// Flow-1, which is already in connectionStore
+	tuple1 := flowexporter.Tuple{SourceAddress: net.IP{1, 2, 3, 4}, DestinationAddress: net.IP{4, 3, 2, 1}, Protocol: 6, SourcePort: 65280, DestinationPort: 255}
+	testFlows[0] = &flowexporter.Connection{
+		StartTime:       refTime.Add(-(time.Second * 50)),
+		StopTime:        refTime,
+		OriginalPackets: 0xffff,
+		OriginalBytes:   0xbaaaaa0000000000,
+		ReversePackets:  0xff,
+		ReverseBytes:    0xbaaa,
+		FlowKey:         tuple1,
+		IsPresent:       true,
+	}
+	// Flow-2, which is not in connectionStore
+	tuple2 := flowexporter.Tuple{SourceAddress: net.IP{5, 6, 7, 8}, DestinationAddress: net.IP{8, 7, 6, 5}, Protocol: 6, SourcePort: 60001, DestinationPort: 200}
+	testFlows[1] = &flowexporter.Connection{
+		StartTime:       refTime.Add(-(time.Second * 20)),
+		StopTime:        refTime,
+		OriginalPackets: 0xbb,
+		OriginalBytes:   0xcbbb,
+		ReversePackets:  0xbbbb,
+		ReverseBytes:    0xcbbbb0000000000,
+		FlowKey:         tuple2,
+		IsPresent:       true,
+	}
+	for i, flow := range testFlows {
+		connKey := flowexporter.NewConnectionKey(flow)
+		testFlowKeys[i] = &connKey
+	}
+	// For testing purposes, set the metric
+	metrics.TotalAntreaConnectionsInConnTrackTable.Set(float64(len(testFlows)))
+	// Create connectionStore
+	mockIfaceStore := interfacestoretest.NewMockInterfaceStore(ctrl)
+	connStore := NewConntrackConnectionStore(nil, flowrecords.NewFlowRecords(), mockIfaceStore, true, false, nil, nil, testPollInterval)
+	// Add flows to the connection store.
+	for i, flow := range testFlows {
+		connStore.connections[*testFlowKeys[i]] = flow
+	}
+	// Delete the connections in connection store.
+	for i := 0; i < len(testFlows); i++ {
+		err := connStore.DeleteConnWithoutLock(*testFlowKeys[i])
+		assert.Nil(t, err, "DeleteConnectionByKey should return nil")
+		_, exists := connStore.GetConnByKey(*testFlowKeys[i])
+		assert.Equal(t, exists, false, "connection should be deleted in connection store")
+		checkAntreaConnectionMetrics(t, len(connStore.connections))
+	}
+}
+
+func TestConnectionStore_MetricSettingInPoll(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+	metrics.InitializeConnectionMetrics()
+
+	testFlows := make([]*flowexporter.Connection, 0)
+	// Create connectionStore
+	mockIfaceStore := interfacestoretest.NewMockInterfaceStore(ctrl)
+	mockConnDumper := connectionstest.NewMockConnTrackDumper(ctrl)
+	conntrackConnStore := NewConntrackConnectionStore(mockConnDumper, flowrecords.NewFlowRecords(), mockIfaceStore, true, false, nil, nil, testPollInterval)
+	// Hard-coded conntrack occupancy metrics for test
+	TotalConnections := 0
+	MaxConnections := 300000
+	mockConnDumper.EXPECT().DumpFlows(uint16(openflow.CtZone)).Return(testFlows, TotalConnections, nil)
+	mockConnDumper.EXPECT().GetMaxConnections().Return(MaxConnections, nil)
+	connsLens, err := conntrackConnStore.Poll()
+	require.Nil(t, err, fmt.Sprintf("Failed to add connections to connection store: %v", err))
+	assert.Equal(t, len(connsLens), 1, "length of connsLens is expected to be 1")
+	assert.Equal(t, connsLens[0], len(testFlows), "expected connections should be equal to number of testFlows")
+	checkTotalConnectionsMetric(t, TotalConnections)
+	checkMaxConnectionsMetric(t, MaxConnections)
+}
+
+func checkAntreaConnectionMetrics(t *testing.T, numConns int) {
+	expectedAntreaConnectionCount := `
+	# HELP antrea_agent_conntrack_antrea_connection_count [ALPHA] Number of connections in the Antrea ZoneID of the conntrack table. This metric gets updated at an interval specified by flowPollInterval, a configuration parameter for the Agent.
+	# TYPE antrea_agent_conntrack_antrea_connection_count gauge
+	`
+	expectedAntreaConnectionCount = expectedAntreaConnectionCount + fmt.Sprintf("antrea_agent_conntrack_antrea_connection_count %d\n", numConns)
+	err := testutil.GatherAndCompare(legacyregistry.DefaultGatherer, strings.NewReader(expectedAntreaConnectionCount), "antrea_agent_conntrack_antrea_connection_count")
+	assert.NoError(t, err)
+}
+
+func checkTotalConnectionsMetric(t *testing.T, numConns int) {
+	expectedConnectionCount := `
+	# HELP antrea_agent_conntrack_total_connection_count [ALPHA] Number of connections in the conntrack table. This metric gets updated at an interval specified by flowPollInterval, a configuration parameter for the Agent.
+	# TYPE antrea_agent_conntrack_total_connection_count gauge
+	`
+	expectedConnectionCount = expectedConnectionCount + fmt.Sprintf("antrea_agent_conntrack_total_connection_count %d\n", numConns)
+	err := testutil.GatherAndCompare(legacyregistry.DefaultGatherer, strings.NewReader(expectedConnectionCount), "antrea_agent_conntrack_total_connection_count")
+	assert.NoError(t, err)
+}
+
+func checkMaxConnectionsMetric(t *testing.T, maxConns int) {
+	expectedMaxConnectionsCount := `
+	# HELP antrea_agent_conntrack_max_connection_count [ALPHA] Size of the conntrack table. This metric gets updated at an interval specified by flowPollInterval, a configuration parameter for the Agent.
+	# TYPE antrea_agent_conntrack_max_connection_count gauge
+	`
+	expectedMaxConnectionsCount = expectedMaxConnectionsCount + fmt.Sprintf("antrea_agent_conntrack_max_connection_count %d\n", maxConns)
+	err := testutil.GatherAndCompare(legacyregistry.DefaultGatherer, strings.NewReader(expectedMaxConnectionsCount), "antrea_agent_conntrack_max_connection_count")
+	assert.NoError(t, err)
+}
diff --git a/pkg/agent/flowexporter/connections/conntrack_linux_test.go b/pkg/agent/flowexporter/connections/conntrack_linux_test.go
index 8240cf3fb..6bfa6b3a6 100644
--- a/pkg/agent/flowexporter/connections/conntrack_linux_test.go
+++ b/pkg/agent/flowexporter/connections/conntrack_linux_test.go
@@ -54,28 +54,25 @@ func TestConnTrackSystem_DumpFlows(t *testing.T) {
 	defer ctrl.Finish()
 	metrics.InitializeConnectionMetrics()
 	// Create flows for test
-	tuple, revTuple := makeTuple(&net.IP{1, 2, 3, 4}, &net.IP{4, 3, 2, 1}, 6, 65280, 255)
+
+	tuple := flowexporter.Tuple{SourceAddress: net.IP{1, 2, 3, 4}, DestinationAddress: net.IP{4, 3, 2, 1}, Protocol: 6, SourcePort: 65280, DestinationPort: 255}
 	antreaFlow := &flowexporter.Connection{
-		TupleOrig:  tuple,
-		TupleReply: revTuple,
-		Zone:       openflow.CtZone,
+		FlowKey: tuple,
+		Zone:    openflow.CtZone,
 	}
-	tuple, revTuple = makeTuple(&net.IP{1, 2, 3, 4}, &net.IP{100, 50, 25, 5}, 6, 60001, 200)
+	tuple = flowexporter.Tuple{SourceAddress: net.IP{1, 2, 3, 4}, DestinationAddress: net.IP{100, 50, 25, 5}, Protocol: 6, SourcePort: 60001, DestinationPort: 200}
 	antreaServiceFlow := &flowexporter.Connection{
-		TupleOrig:  tuple,
-		TupleReply: revTuple,
-		Zone:       openflow.CtZone,
+		FlowKey: tuple,
+		Zone:    openflow.CtZone,
 	}
-	tuple, revTuple = makeTuple(&net.IP{5, 6, 7, 8}, &net.IP{8, 7, 6, 5}, 6, 60001, 200)
+	tuple = flowexporter.Tuple{SourceAddress: net.IP{5, 6, 7, 8}, DestinationAddress: net.IP{8, 7, 6, 5}, Protocol: 6, SourcePort: 60001, DestinationPort: 200}
 	antreaGWFlow := &flowexporter.Connection{
-		TupleOrig:  tuple,
-		TupleReply: revTuple,
-		Zone:       openflow.CtZone,
+		FlowKey: tuple,
+		Zone:    openflow.CtZone,
 	}
 	nonAntreaFlow := &flowexporter.Connection{
-		TupleOrig:  tuple,
-		TupleReply: revTuple,
-		Zone:       100,
+		FlowKey: tuple,
+		Zone:    100,
 	}
 	testFlows := []*flowexporter.Connection{antreaFlow, antreaServiceFlow, antreaGWFlow, nonAntreaFlow}
 
@@ -154,30 +151,25 @@ func TestConnTrackOvsAppCtl_DumpFlows(t *testing.T) {
 		Zone:       65520,
 		StatusFlag: 302,
 		Mark:       0x21,
-		TupleOrig: flowexporter.Tuple{
+		FlowKey: flowexporter.Tuple{
 			SourceAddress:      net.ParseIP("100.10.0.105"),
-			DestinationAddress: net.ParseIP("10.96.0.1"),
+			DestinationAddress: net.ParseIP("100.10.0.106"),
 			Protocol:           6,
 			SourcePort:         uint16(41284),
-			DestinationPort:    uint16(443),
-		},
-		TupleReply: flowexporter.Tuple{
-			SourceAddress:      net.ParseIP("100.10.0.106"),
-			DestinationAddress: net.ParseIP("100.10.0.105"),
-			Protocol:           6,
-			SourcePort:         6443,
-			DestinationPort:    41284,
+			DestinationPort:    uint16(6443),
 		},
-		OriginalPackets:         343260,
-		OriginalBytes:           19340621,
-		ReversePackets:          381035,
-		ReverseBytes:            181176472,
-		SourcePodNamespace:      "",
-		SourcePodName:           "",
-		DestinationPodNamespace: "",
-		DestinationPodName:      "",
-		TCPState:                "ESTABLISHED",
-		Labels:                  []byte{1, 0, 0, 0, 2, 0, 0, 0},
+		DestinationServiceAddress: net.ParseIP("10.96.0.1"),
+		DestinationServicePort:    uint16(443),
+		OriginalPackets:           343260,
+		OriginalBytes:             19340621,
+		ReversePackets:            381035,
+		ReverseBytes:              181176472,
+		SourcePodNamespace:        "",
+		SourcePodName:             "",
+		DestinationPodNamespace:   "",
+		DestinationPodName:        "",
+		TCPState:                  "ESTABLISHED",
+		Labels:                    []byte{1, 0, 0, 0, 2, 0, 0, 0},
 	}
 	mockOVSCtlClient.EXPECT().RunAppctlCmd("dpctl/dump-conntrack", false, "-m", "-s").Return(ovsctlCmdOutput, nil)
 
@@ -228,25 +220,26 @@ func TestNetLinkFlowToAntreaConnection(t *testing.T) {
 		Timeout: 123, Status: conntrack.Status{Value: conntrack.StatusAssured}, Mark: 0x1234, Zone: 2,
 		Timestamp: conntrack.Timestamp{Start: time.Date(2020, 7, 25, 8, 40, 8, 959000000, time.UTC)},
 	}
-	tuple, _ := makeTuple(&conntrackFlowTuple.IP.SourceAddress, &conntrackFlowTuple.IP.DestinationAddress, conntrackFlowTuple.Proto.Protocol, conntrackFlowTuple.Proto.SourcePort, conntrackFlowTuple.Proto.DestinationPort)
+	tuple := flowexporter.Tuple{SourceAddress: conntrackFlowTuple.IP.SourceAddress, DestinationAddress: conntrackFlowTuple.IP.SourceAddress, Protocol: conntrackFlowTuple.Proto.Protocol, SourcePort: conntrackFlowTuple.Proto.SourcePort, DestinationPort: conntrackFlowTuple.Proto.SourcePort}
 	expectedAntreaFlow := &flowexporter.Connection{
-		Timeout:                 netlinkFlow.Timeout,
-		StartTime:               netlinkFlow.Timestamp.Start,
-		IsPresent:               true,
-		Zone:                    2,
-		StatusFlag:              0x4,
-		Mark:                    0x1234,
-		TupleOrig:               tuple,
-		TupleReply:              tuple,
-		OriginalPackets:         netlinkFlow.CountersOrig.Packets,
-		OriginalBytes:           netlinkFlow.CountersOrig.Bytes,
-		ReversePackets:          netlinkFlow.CountersReply.Packets,
-		ReverseBytes:            netlinkFlow.CountersReply.Bytes,
-		SourcePodNamespace:      "",
-		SourcePodName:           "",
-		DestinationPodNamespace: "",
-		DestinationPodName:      "",
-		TCPState:                "",
+		Timeout:                   netlinkFlow.Timeout,
+		StartTime:                 netlinkFlow.Timestamp.Start,
+		IsPresent:                 true,
+		Zone:                      2,
+		StatusFlag:                0x4,
+		Mark:                      0x1234,
+		FlowKey:                   tuple,
+		DestinationServiceAddress: conntrackFlowTuple.IP.DestinationAddress,
+		DestinationServicePort:    conntrackFlowTuple.Proto.DestinationPort,
+		OriginalPackets:           netlinkFlow.CountersOrig.Packets,
+		OriginalBytes:             netlinkFlow.CountersOrig.Bytes,
+		ReversePackets:            netlinkFlow.CountersReply.Packets,
+		ReverseBytes:              netlinkFlow.CountersReply.Bytes,
+		SourcePodNamespace:        "",
+		SourcePodName:             "",
+		DestinationPodNamespace:   "",
+		DestinationPodName:        "",
+		TCPState:                  "",
 	}
 
 	antreaFlow := NetlinkFlowToAntreaConnection(netlinkFlow)
@@ -265,24 +258,25 @@ func TestNetLinkFlowToAntreaConnection(t *testing.T) {
 		},
 	}
 	expectedAntreaFlow = &flowexporter.Connection{
-		Timeout:                 netlinkFlow.Timeout,
-		StartTime:               netlinkFlow.Timestamp.Start,
-		StopTime:                netlinkFlow.Timestamp.Stop,
-		IsPresent:               true,
-		Zone:                    2,
-		StatusFlag:              0x204,
-		Mark:                    0x1234,
-		TupleOrig:               tuple,
-		TupleReply:              tuple,
-		OriginalPackets:         netlinkFlow.CountersOrig.Packets,
-		OriginalBytes:           netlinkFlow.CountersOrig.Bytes,
-		ReversePackets:          netlinkFlow.CountersReply.Packets,
-		ReverseBytes:            netlinkFlow.CountersReply.Bytes,
-		SourcePodNamespace:      "",
-		SourcePodName:           "",
-		DestinationPodNamespace: "",
-		DestinationPodName:      "",
-		TCPState:                "",
+		Timeout:                   netlinkFlow.Timeout,
+		StartTime:                 netlinkFlow.Timestamp.Start,
+		StopTime:                  netlinkFlow.Timestamp.Stop,
+		IsPresent:                 true,
+		Zone:                      2,
+		StatusFlag:                0x204,
+		Mark:                      0x1234,
+		FlowKey:                   tuple,
+		DestinationServiceAddress: conntrackFlowTuple.IP.DestinationAddress,
+		DestinationServicePort:    conntrackFlowTuple.Proto.DestinationPort,
+		OriginalPackets:           netlinkFlow.CountersOrig.Packets,
+		OriginalBytes:             netlinkFlow.CountersOrig.Bytes,
+		ReversePackets:            netlinkFlow.CountersReply.Packets,
+		ReverseBytes:              netlinkFlow.CountersReply.Bytes,
+		SourcePodNamespace:        "",
+		SourcePodName:             "",
+		DestinationPodNamespace:   "",
+		DestinationPodName:        "",
+		TCPState:                  "",
 	}
 
 	antreaFlow = NetlinkFlowToAntreaConnection(netlinkFlow)
diff --git a/pkg/agent/flowexporter/connections/deny_connections_test.go b/pkg/agent/flowexporter/connections/deny_connections_test.go
new file mode 100644
index 000000000..8ab76f187
--- /dev/null
+++ b/pkg/agent/flowexporter/connections/deny_connections_test.go
@@ -0,0 +1,137 @@
+// Copyright 2021 Antrea Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package connections
+
+import (
+	"fmt"
+	"net"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/golang/mock/gomock"
+	"github.com/prometheus/client_golang/prometheus/testutil"
+	"github.com/stretchr/testify/assert"
+	v1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/component-base/metrics/legacyregistry"
+
+	"antrea.io/antrea/pkg/agent/flowexporter"
+	interfacestoretest "antrea.io/antrea/pkg/agent/interfacestore/testing"
+	"antrea.io/antrea/pkg/agent/metrics"
+	proxytest "antrea.io/antrea/pkg/agent/proxy/testing"
+	k8sproxy "antrea.io/antrea/third_party/proxy"
+)
+
+func TestDenyConnectionStore_AddOrUpdateConn(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+	metrics.InitializeConnectionMetrics()
+	// Create flow for testing adding and updating of same connection.
+	refTime := time.Now()
+	tuple := flowexporter.Tuple{SourceAddress: net.IP{1, 2, 3, 4}, DestinationAddress: net.IP{4, 3, 2, 1}, Protocol: 6, SourcePort: 65280, DestinationPort: 255}
+	servicePortName := k8sproxy.ServicePortName{
+		NamespacedName: types.NamespacedName{
+			Namespace: "serviceNS1",
+			Name:      "service1",
+		},
+		Port:     "255",
+		Protocol: v1.ProtocolTCP,
+	}
+	// flow for testing adding and updating
+	testFlow := flowexporter.Connection{
+		StopTime:                  refTime.Add(-(time.Second * 20)),
+		StartTime:                 refTime.Add(-(time.Second * 20)),
+		LastExportTime:            refTime.Add(-(time.Second * 20)),
+		FlowKey:                   tuple,
+		DestinationServiceAddress: tuple.DestinationAddress,
+		DestinationServicePort:    tuple.DestinationPort,
+		DeltaBytes:                uint64(60),
+		DeltaPackets:              uint64(1),
+		OriginalBytes:             uint64(60),
+		OriginalPackets:           uint64(1),
+	}
+	mockIfaceStore := interfacestoretest.NewMockInterfaceStore(ctrl)
+	mockProxier := proxytest.NewMockProxier(ctrl)
+	protocol, _ := lookupServiceProtocol(tuple.Protocol)
+	serviceStr := fmt.Sprintf("%s:%d/%s", tuple.DestinationAddress.String(), tuple.DestinationPort, protocol)
+	mockProxier.EXPECT().GetServiceByIP(serviceStr).Return(servicePortName, true)
+	mockIfaceStore.EXPECT().GetInterfaceByIP(tuple.SourceAddress.String()).Return(nil, false)
+	mockIfaceStore.EXPECT().GetInterfaceByIP(tuple.DestinationAddress.String()).Return(nil, false)
+
+	denyConnStore := NewDenyConnectionStore(mockIfaceStore, mockProxier)
+
+	denyConnStore.AddOrUpdateConn(&testFlow, refTime.Add(-(time.Second * 20)), uint64(60))
+	expConn := testFlow
+	expConn.DestinationServicePortName = servicePortName.String()
+	actualConn, ok := denyConnStore.GetConnByKey(flowexporter.NewConnectionKey(&testFlow))
+	assert.Equal(t, ok, true, "deny connection should be there in deny connection store")
+	assert.Equal(t, expConn, *actualConn, "deny connections should be equal")
+	checkDenyConnectionMetrics(t, len(denyConnStore.connections))
+
+	denyConnStore.AddOrUpdateConn(&testFlow, refTime.Add(-(time.Second * 10)), uint64(60))
+	expConn.OriginalBytes = uint64(120)
+	expConn.DeltaBytes = uint64(120)
+	expConn.OriginalPackets = uint64(2)
+	expConn.DeltaPackets = uint64(2)
+	expConn.StopTime = refTime.Add(-(time.Second * 10))
+	actualConn, ok = denyConnStore.GetConnByKey(flowexporter.NewConnectionKey(&testFlow))
+	assert.Equal(t, ok, true, "deny connection should be there in deny connection store")
+	assert.Equal(t, expConn, *actualConn, "deny connections should be equal")
+	checkDenyConnectionMetrics(t, len(denyConnStore.connections))
+}
+
+func TestDenyConnectionStore_DeleteConnWithoutLock(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+	metrics.InitializeConnectionMetrics()
+	// Create denyConnectionStore
+	mockIfaceStore := interfacestoretest.NewMockInterfaceStore(ctrl)
+	connStore := NewDenyConnectionStore(mockIfaceStore, nil)
+	refTime := time.Now()
+	tuple1 := flowexporter.Tuple{SourceAddress: net.IP{1, 2, 3, 4}, DestinationAddress: net.IP{4, 3, 2, 1}, Protocol: 6, SourcePort: 65280, DestinationPort: 255}
+	tuple2 := flowexporter.Tuple{SourceAddress: net.IP{1, 2, 3, 4}, DestinationAddress: net.IP{8, 7, 6, 5}, Protocol: 6, SourcePort: 65280, DestinationPort: 255}
+	conn := &flowexporter.Connection{
+		StopTime:   refTime.Add(-(time.Second * 20)),
+		FlowKey:    tuple1,
+		DeltaBytes: uint64(60),
+	}
+	connKey := flowexporter.NewConnectionKey(conn)
+	connStore.connections[connKey] = conn
+	// create invalid connection key to be deleted
+	conn.FlowKey = tuple2
+	invalidConnKey := flowexporter.NewConnectionKey(conn)
+
+	// For testing purposes, set the metric
+	metrics.TotalDenyConnections.Set(1)
+
+	err := connStore.DeleteConnWithoutLock(connKey)
+	assert.Nil(t, err, "DeleteConnWithoutLock should return nil")
+	_, exists := connStore.GetConnByKey(connKey)
+	assert.Equal(t, exists, false, "connection should be deleted in connection store")
+	checkDenyConnectionMetrics(t, len(connStore.connections))
+	assert.NotNil(t, connStore.DeleteConnWithoutLock(invalidConnKey))
+	checkDenyConnectionMetrics(t, len(connStore.connections))
+}
+
+func checkDenyConnectionMetrics(t *testing.T, numConns int) {
+	expectedDenyConnectionCount := `
+	# HELP antrea_agent_denied_connection_count [ALPHA] Number of denied connections detected by Flow Exporter deny connections tracking. This metric gets updated when a flow is rejected/dropped by network policy.
+	# TYPE antrea_agent_denied_connection_count gauge
+	`
+	expectedDenyConnectionCount = expectedDenyConnectionCount + fmt.Sprintf("antrea_agent_denied_connection_count %d\n", numConns)
+	err := testutil.GatherAndCompare(legacyregistry.DefaultGatherer, strings.NewReader(expectedDenyConnectionCount), "antrea_agent_denied_connection_count")
+	assert.NoError(t, err)
+}
diff --git a/pkg/agent/flowexporter/connections/testing/mock_connections.go b/pkg/agent/flowexporter/connections/testing/mock_connections.go
index 3f52ffd32..213e20d1f 100644
--- a/pkg/agent/flowexporter/connections/testing/mock_connections.go
+++ b/pkg/agent/flowexporter/connections/testing/mock_connections.go
@@ -14,7 +14,7 @@
 //
 
 // Code generated by MockGen. DO NOT EDIT.
-// Source: antrea.io/antrea/pkg/agent/flowexporter/connections (interfaces: ConnectionStore,ConnTrackDumper,NetFilterConnTrack)
+// Source: antrea.io/antrea/pkg/agent/flowexporter/connections (interfaces: ConnTrackDumper,NetFilterConnTrack)
 
 // Package testing is a generated GoMock package.
 package testing
@@ -25,84 +25,6 @@ import (
 	reflect "reflect"
 )
 
-// MockConnectionStore is a mock of ConnectionStore interface
-type MockConnectionStore struct {
-	ctrl     *gomock.Controller
-	recorder *MockConnectionStoreMockRecorder
-}
-
-// MockConnectionStoreMockRecorder is the mock recorder for MockConnectionStore
-type MockConnectionStoreMockRecorder struct {
-	mock *MockConnectionStore
-}
-
-// NewMockConnectionStore creates a new mock instance
-func NewMockConnectionStore(ctrl *gomock.Controller) *MockConnectionStore {
-	mock := &MockConnectionStore{ctrl: ctrl}
-	mock.recorder = &MockConnectionStoreMockRecorder{mock}
-	return mock
-}
-
-// EXPECT returns an object that allows the caller to indicate expected use
-func (m *MockConnectionStore) EXPECT() *MockConnectionStoreMockRecorder {
-	return m.recorder
-}
-
-// ForAllConnectionsDo mocks base method
-func (m *MockConnectionStore) ForAllConnectionsDo(arg0 flowexporter.ConnectionMapCallBack) error {
-	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "ForAllConnectionsDo", arg0)
-	ret0, _ := ret[0].(error)
-	return ret0
-}
-
-// ForAllConnectionsDo indicates an expected call of ForAllConnectionsDo
-func (mr *MockConnectionStoreMockRecorder) ForAllConnectionsDo(arg0 interface{}) *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ForAllConnectionsDo", reflect.TypeOf((*MockConnectionStore)(nil).ForAllConnectionsDo), arg0)
-}
-
-// GetConnByKey mocks base method
-func (m *MockConnectionStore) GetConnByKey(arg0 flowexporter.ConnectionKey) (*flowexporter.Connection, bool) {
-	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "GetConnByKey", arg0)
-	ret0, _ := ret[0].(*flowexporter.Connection)
-	ret1, _ := ret[1].(bool)
-	return ret0, ret1
-}
-
-// GetConnByKey indicates an expected call of GetConnByKey
-func (mr *MockConnectionStoreMockRecorder) GetConnByKey(arg0 interface{}) *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetConnByKey", reflect.TypeOf((*MockConnectionStore)(nil).GetConnByKey), arg0)
-}
-
-// Run mocks base method
-func (m *MockConnectionStore) Run(arg0 <-chan struct{}) {
-	m.ctrl.T.Helper()
-	m.ctrl.Call(m, "Run", arg0)
-}
-
-// Run indicates an expected call of Run
-func (mr *MockConnectionStoreMockRecorder) Run(arg0 interface{}) *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Run", reflect.TypeOf((*MockConnectionStore)(nil).Run), arg0)
-}
-
-// SetExportDone mocks base method
-func (m *MockConnectionStore) SetExportDone(arg0 flowexporter.ConnectionKey) error {
-	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "SetExportDone", arg0)
-	ret0, _ := ret[0].(error)
-	return ret0
-}
-
-// SetExportDone indicates an expected call of SetExportDone
-func (mr *MockConnectionStoreMockRecorder) SetExportDone(arg0 interface{}) *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SetExportDone", reflect.TypeOf((*MockConnectionStore)(nil).SetExportDone), arg0)
-}
-
 // MockConnTrackDumper is a mock of ConnTrackDumper interface
 type MockConnTrackDumper struct {
 	ctrl     *gomock.Controller
diff --git a/pkg/agent/flowexporter/exporter/exporter_test.go b/pkg/agent/flowexporter/exporter/exporter_test.go
index 0614906ee..1f877e888 100644
--- a/pkg/agent/flowexporter/exporter/exporter_test.go
+++ b/pkg/agent/flowexporter/exporter/exporter_test.go
@@ -26,6 +26,7 @@ import (
 	ipfixregistry "github.com/vmware/go-ipfix/pkg/registry"
 
 	"antrea.io/antrea/pkg/agent/flowexporter"
+	"antrea.io/antrea/pkg/agent/flowexporter/connections"
 	connectionstest "antrea.io/antrea/pkg/agent/flowexporter/connections/testing"
 	"antrea.io/antrea/pkg/agent/flowexporter/flowrecords"
 	ipfixtest "antrea.io/antrea/pkg/ipfix/testing"
@@ -249,13 +250,13 @@ func getElemList(ianaIE []string, antreaIE []string) []*ipfixentities.InfoElemen
 }
 
 func getConnection(isIPv6 bool, isPresent bool, statusFlag uint32, protoID uint8, tcpState string) *flowexporter.Connection {
-	var tuple, revTuple flowexporter.Tuple
+	var tuple flowexporter.Tuple
 	if !isIPv6 {
-		tuple, revTuple = makeTuple(&net.IP{1, 2, 3, 4}, &net.IP{4, 3, 2, 1}, 6, 65280, 255)
+		tuple = flowexporter.Tuple{SourceAddress: net.IP{1, 2, 3, 4}, DestinationAddress: net.IP{4, 3, 2, 1}, Protocol: 6, SourcePort: 65280, DestinationPort: 255}
 	} else {
-		srcIP := net.IP([]byte{0x20, 0x1, 0x0, 0x0, 0x32, 0x38, 0xdf, 0xe1, 0x0, 0x63, 0x0, 0x0, 0x0, 0x0, 0xfe, 0xfb})
-		dstIP := net.IP([]byte{0x20, 0x1, 0x0, 0x0, 0x32, 0x38, 0xdf, 0xe1, 0x0, 0x63, 0x0, 0x0, 0x0, 0x0, 0xfe, 0xfc})
-		tuple, revTuple = makeTuple(&srcIP, &dstIP, protoID, 65280, 255)
+		srcIP := net.ParseIP("2001:0:3238:dfe1:63::fefb")
+		dstIP := net.ParseIP("2001:0:3238:dfe1:63::fefc")
+		tuple = flowexporter.Tuple{SourceAddress: srcIP, DestinationAddress: dstIP, Protocol: protoID, SourcePort: 65280, DestinationPort: 255}
 	}
 	conn := &flowexporter.Connection{
 		StartTime:                     time.Time{},
@@ -265,8 +266,7 @@ func getConnection(isIPv6 bool, isPresent bool, statusFlag uint32, protoID uint8
 		OriginalBytes:                 0xabcd,
 		ReversePackets:                0xa,
 		ReverseBytes:                  0xab,
-		TupleOrig:                     tuple,
-		TupleReply:                    revTuple,
+		FlowKey:                       tuple,
 		IsPresent:                     isPresent,
 		SourcePodNamespace:            "ns",
 		SourcePodName:                 "pod",
@@ -282,6 +282,26 @@ func getConnection(isIPv6 bool, isPresent bool, statusFlag uint32, protoID uint8
 	return conn
 }
 
+func getDenyConnection(isIPv6 bool, isActive bool) *flowexporter.Connection {
+	var tuple, _ flowexporter.Tuple
+	if !isIPv6 {
+		tuple, _ = makeTuple(&net.IP{1, 2, 3, 4}, &net.IP{4, 3, 2, 1}, 6, 65280, 255)
+	} else {
+		srcIP := net.ParseIP("2001:0:3238:dfe1:63::fefb")
+		dstIP := net.ParseIP("2001:0:3238:dfe1:63::fefc")
+		tuple, _ = makeTuple(&srcIP, &dstIP, 6, 65280, 255)
+	}
+	conn := &flowexporter.Connection{
+		FlowKey:        tuple,
+		LastExportTime: time.Now().Add(-testIdleFlowTimeout),
+	}
+	if isActive {
+		conn.LastExportTime = time.Now().Add(-testActiveFlowTimeout)
+		conn.DeltaPackets = uint64(1)
+	}
+	return conn
+}
+
 func getFlowRecord(conn *flowexporter.Connection, isIPv6 bool, isActive bool) flowexporter.FlowRecord {
 	flowRecord := &flowexporter.FlowRecord{
 		Conn:               *conn,
@@ -340,10 +360,10 @@ func runSendFlowRecordTests(t *testing.T, flowExp *flowExporter, isIPv6 bool) {
 
 	mockIPFIXExpProc := ipfixtest.NewMockIPFIXExportingProcess(ctrl)
 	mockDataSet := ipfixentitiestesting.NewMockSet(ctrl)
-	mockConnStore := connectionstest.NewMockConnectionStore(ctrl)
 	flowExp.process = mockIPFIXExpProc
 	flowExp.ipfixSet = mockDataSet
-	flowExp.connStore = mockConnStore
+	mockConnDumper := connectionstest.NewMockConnTrackDumper(ctrl)
+	flowExp.conntrackConnStore = connections.NewConntrackConnectionStore(mockConnDumper, flowrecords.NewFlowRecords(), nil, !isIPv6, isIPv6, nil, nil, 1)
 
 	tests := []struct {
 		name               string
@@ -354,6 +374,7 @@ func runSendFlowRecordTests(t *testing.T, flowExp *flowExporter, isIPv6 bool) {
 		tcpState           string
 		statusFlag         uint32
 		protoID            uint8
+		isDenyConnActive   bool
 	}{
 		{
 			"active flow record",
@@ -364,6 +385,7 @@ func runSendFlowRecordTests(t *testing.T, flowExp *flowExporter, isIPv6 bool) {
 			"SYN_SENT",
 			0x4,
 			6,
+			true,
 		},
 		{
 			"idle flow record",
@@ -374,6 +396,7 @@ func runSendFlowRecordTests(t *testing.T, flowExp *flowExporter, isIPv6 bool) {
 			"ESTABLISHED",
 			302,
 			6,
+			false,
 		},
 		{
 			"idle flow record that is still inactive",
@@ -384,6 +407,7 @@ func runSendFlowRecordTests(t *testing.T, flowExp *flowExporter, isIPv6 bool) {
 			"",
 			0x204,
 			17,
+			false,
 		},
 		{
 			"idle flow record becomes active",
@@ -394,6 +418,7 @@ func runSendFlowRecordTests(t *testing.T, flowExp *flowExporter, isIPv6 bool) {
 			"SYN_SENT",
 			302,
 			6,
+			true,
 		},
 		{
 			"idle flow record for deleted connection",
@@ -404,17 +429,24 @@ func runSendFlowRecordTests(t *testing.T, flowExp *flowExporter, isIPv6 bool) {
 			"TIME_WAIT",
 			0x204,
 			6,
+			false,
 		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			conn := getConnection(isIPv6, tt.isConnPresent, tt.statusFlag, tt.protoID, tt.tcpState)
 			connKey := flowexporter.NewConnectionKey(conn)
+			flowExp.conntrackConnStore.AddOrUpdateConn(conn)
 			flowExp.flowRecords = flowrecords.NewFlowRecords()
 			err := flowExp.flowRecords.AddOrUpdateFlowRecord(connKey, conn)
 			assert.NoError(t, err)
 			flowExp.numDataSetsSent = 0
 
+			denyConn := getDenyConnection(isIPv6, tt.isDenyConnActive)
+			flowExp.denyConnStore = connections.NewDenyConnectionStore(nil, nil)
+			flowExp.denyConnStore.AddOrUpdateConn(denyConn, denyConn.LastExportTime, denyConn.DeltaBytes)
+			assert.Equal(t, getNumOfConnections(flowExp.denyConnStore), 1)
+
 			// Get the flow record and update it.
 			flowRec, exists := flowExp.flowRecords.GetFlowRecordFromMap(&connKey)
 			if !exists {
@@ -426,32 +458,47 @@ func runSendFlowRecordTests(t *testing.T, flowExp *flowExporter, isIPv6 bool) {
 			flowRec.LastExportTime = time.Now().Add(-tt.lastExportTimeDiff)
 			flowExp.flowRecords.AddFlowRecordToMap(&connKey, flowRec)
 
+			count := 1
 			if tt.isRecordActive {
-				if !isIPv6 {
-					mockDataSet.EXPECT().PrepareSet(ipfixentities.Data, flowExp.templateIDv4).Return(nil)
-					mockDataSet.EXPECT().AddRecord(flowExp.elementsListv4, flowExp.templateIDv4).Return(nil)
-				} else {
-					mockDataSet.EXPECT().PrepareSet(ipfixentities.Data, flowExp.templateIDv6).Return(nil)
-					mockDataSet.EXPECT().AddRecord(flowExp.elementsListv6, flowExp.templateIDv6).Return(nil)
-				}
-				if flowexporter.IsConnectionDying(conn) {
-					mockConnStore.EXPECT().SetExportDone(connKey)
-				}
-				mockIPFIXExpProc.EXPECT().SendSet(mockDataSet).Return(0, nil)
-				mockDataSet.EXPECT().ResetSet()
-				err = flowExp.sendFlowRecords()
-				assert.NoError(t, err)
-				assert.Equalf(t, uint64(1), flowExp.numDataSetsSent, "data set should have been sent.")
-				if flowexporter.IsConnectionDying(conn) {
-					_, recPresent := flowExp.flowRecords.GetFlowRecordFromMap(&connKey)
-					assert.Falsef(t, recPresent, "record should not be in the map")
-				}
+				count += 1
+			}
+			if !isIPv6 {
+				mockDataSet.EXPECT().PrepareSet(ipfixentities.Data, flowExp.templateIDv4).Times(count).Return(nil)
+				mockDataSet.EXPECT().AddRecord(flowExp.elementsListv4, flowExp.templateIDv4).Times(count).Return(nil)
 			} else {
-				err = flowExp.sendFlowRecords()
-				assert.NoError(t, err)
-				assert.Equalf(t, uint64(0), flowExp.numDataSetsSent, "data set should not have been sent.")
+				mockDataSet.EXPECT().PrepareSet(ipfixentities.Data, flowExp.templateIDv6).Times(count).Return(nil)
+				mockDataSet.EXPECT().AddRecord(flowExp.elementsListv6, flowExp.templateIDv6).Times(count).Return(nil)
 			}
+			mockIPFIXExpProc.EXPECT().SendSet(mockDataSet).Times(count).Return(0, nil)
+			mockDataSet.EXPECT().ResetSet().Times(count)
 
+			err = flowExp.sendFlowRecords()
+			assert.NoError(t, err)
+			assert.Equalf(t, uint64(count), flowExp.numDataSetsSent, "%v data sets should have been sent.", count)
+			if tt.isDenyConnActive {
+				connection, exist := flowExp.denyConnStore.GetConnByKey(connKey)
+				assert.True(t, exist)
+				assert.Equal(t, uint64(0), connection.DeltaPackets)
+				assert.Equal(t, uint64(0), connection.DeltaBytes)
+			} else {
+				assert.Equal(t, getNumOfConnections(flowExp.denyConnStore), 0)
+			}
+			if tt.isRecordActive && flowexporter.IsConnectionDying(conn) {
+				_, recPresent := flowExp.flowRecords.GetFlowRecordFromMap(&connKey)
+				assert.Falsef(t, recPresent, "record should not be in the map")
+				connection, _ := flowExp.conntrackConnStore.GetConnByKey(connKey)
+				assert.True(t, connection.DoneExport)
+			}
 		})
 	}
 }
+
+func getNumOfConnections(connStore *connections.DenyConnectionStore) int {
+	count := 0
+	countNumOfConns := func(key flowexporter.ConnectionKey, conn *flowexporter.Connection) error {
+		count++
+		return nil
+	}
+	connStore.ForAllConnectionsDo(countNumOfConns)
+	return count
+}
diff --git a/pkg/agent/openflow/client_test.go b/pkg/agent/openflow/client_test.go
index ec17223bd..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},
 	}
 
@@ -101,7 +100,7 @@ func TestIdempotentFlowInstallation(t *testing.T) {
 			ctrl := gomock.NewController(t)
 			defer ctrl.Finish()
 			m := oftest.NewMockOFEntryOperations(ctrl)
-			ofClient := NewClient(bridgeName, bridgeMgmtAddr, ovsconfig.OVSDatapathSystem, true, false, false)
+			ofClient := NewClient(bridgeName, bridgeMgmtAddr, ovsconfig.OVSDatapathSystem, true, false, false, false)
 			client := ofClient.(*client)
 			client.cookieAllocator = cookie.NewAllocator(0)
 			client.ofEntryOperations = m
@@ -129,7 +128,7 @@ func TestIdempotentFlowInstallation(t *testing.T) {
 			ctrl := gomock.NewController(t)
 			defer ctrl.Finish()
 			m := oftest.NewMockOFEntryOperations(ctrl)
-			ofClient := NewClient(bridgeName, bridgeMgmtAddr, ovsconfig.OVSDatapathSystem, true, false, false)
+			ofClient := NewClient(bridgeName, bridgeMgmtAddr, ovsconfig.OVSDatapathSystem, true, false, false, false)
 			client := ofClient.(*client)
 			client.cookieAllocator = cookie.NewAllocator(0)
 			client.ofEntryOperations = m
@@ -170,7 +169,7 @@ func TestFlowInstallationFailed(t *testing.T) {
 			ctrl := gomock.NewController(t)
 			defer ctrl.Finish()
 			m := oftest.NewMockOFEntryOperations(ctrl)
-			ofClient := NewClient(bridgeName, bridgeMgmtAddr, ovsconfig.OVSDatapathSystem, true, false, false)
+			ofClient := NewClient(bridgeName, bridgeMgmtAddr, ovsconfig.OVSDatapathSystem, true, false, false, false)
 			client := ofClient.(*client)
 			client.cookieAllocator = cookie.NewAllocator(0)
 			client.ofEntryOperations = m
@@ -204,7 +203,7 @@ func TestConcurrentFlowInstallation(t *testing.T) {
 			ctrl := gomock.NewController(t)
 			defer ctrl.Finish()
 			m := oftest.NewMockOFEntryOperations(ctrl)
-			ofClient := NewClient(bridgeName, bridgeMgmtAddr, ovsconfig.OVSDatapathSystem, true, false, false)
+			ofClient := NewClient(bridgeName, bridgeMgmtAddr, ovsconfig.OVSDatapathSystem, true, false, false, false)
 			client := ofClient.(*client)
 			client.cookieAllocator = cookie.NewAllocator(0)
 			client.ofEntryOperations = m
@@ -394,7 +393,7 @@ func Test_client_SendTraceflowPacket(t *testing.T) {
 }
 
 func prepareTraceflowFlow(ctrl *gomock.Controller) *client {
-	ofClient := NewClient(bridgeName, bridgeMgmtAddr, ovsconfig.OVSDatapathSystem, true, true, false)
+	ofClient := NewClient(bridgeName, bridgeMgmtAddr, ovsconfig.OVSDatapathSystem, true, true, false, false)
 	c := ofClient.(*client)
 	c.cookieAllocator = cookie.NewAllocator(0)
 	c.nodeConfig = nodeConfig
@@ -412,7 +411,7 @@ func prepareTraceflowFlow(ctrl *gomock.Controller) *client {
 }
 
 func prepareSendTraceflowPacket(ctrl *gomock.Controller, success bool) *client {
-	ofClient := NewClient(bridgeName, bridgeMgmtAddr, ovsconfig.OVSDatapathSystem, true, true, false)
+	ofClient := NewClient(bridgeName, bridgeMgmtAddr, ovsconfig.OVSDatapathSystem, true, true, false, false)
 	c := ofClient.(*client)
 	c.nodeConfig = nodeConfig
 	m := ovsoftest.NewMockBridge(ctrl)
@@ -500,7 +499,7 @@ func Test_client_setBasePacketOutBuilder(t *testing.T) {
 }
 
 func prepareSetBasePacketOutBuilder(ctrl *gomock.Controller, success bool) *client {
-	ofClient := NewClient(bridgeName, bridgeMgmtAddr, ovsconfig.OVSDatapathSystem, true, true, false)
+	ofClient := NewClient(bridgeName, bridgeMgmtAddr, ovsconfig.OVSDatapathSystem, true, true, false, false)
 	c := ofClient.(*client)
 	m := ovsoftest.NewMockBridge(ctrl)
 	c.bridge = m
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/pkg/flowaggregator/flowaggregator_test.go b/pkg/flowaggregator/flowaggregator_test.go
index a822fe876..47ef22067 100644
--- a/pkg/flowaggregator/flowaggregator_test.go
+++ b/pkg/flowaggregator/flowaggregator_test.go
@@ -114,6 +114,8 @@ func TestFlowAggregator_sendFlowKeyRecord(t *testing.T) {
 		mockDataSet.EXPECT().AddRecord(elementList, templateID).Return(nil)
 		mockIPFIXExpProc.EXPECT().SendSet(mockDataSet).Return(0, nil)
 		mockAggregationProcess.EXPECT().ResetStatElementsInRecord(mockRecord).Return(nil)
+		mockAggregationProcess.EXPECT().IsMetadataFilled(tc.flowRecord).Return(false)
+		mockAggregationProcess.EXPECT().SetMetadataFilled(tc.flowRecord)
 
 		err := fa.sendFlowKeyRecord(tc.flowKey, tc.flowRecord)
 		assert.NoError(t, err, "Error in sending flow key record: %v, key: %v, record: %v", err, tc.flowKey, tc.flowRecord)
diff --git a/pkg/ipfix/testing/mock_ipfix.go b/pkg/ipfix/testing/mock_ipfix.go
index 77d6b5ac7..76e467fb0 100644
--- a/pkg/ipfix/testing/mock_ipfix.go
+++ b/pkg/ipfix/testing/mock_ipfix.go
@@ -253,6 +253,20 @@ func (mr *MockIPFIXAggregationProcessMockRecorder) GetExpiryFromExpirePriorityQu
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetExpiryFromExpirePriorityQueue", reflect.TypeOf((*MockIPFIXAggregationProcess)(nil).GetExpiryFromExpirePriorityQueue))
 }
 
+// IsMetadataFilled mocks base method
+func (m *MockIPFIXAggregationProcess) IsMetadataFilled(arg0 intermediate.AggregationFlowRecord) bool {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "IsMetadataFilled", arg0)
+	ret0, _ := ret[0].(bool)
+	return ret0
+}
+
+// IsMetadataFilled indicates an expected call of IsMetadataFilled
+func (mr *MockIPFIXAggregationProcessMockRecorder) IsMetadataFilled(arg0 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "IsMetadataFilled", reflect.TypeOf((*MockIPFIXAggregationProcess)(nil).IsMetadataFilled), arg0)
+}
+
 // ResetStatElementsInRecord mocks base method
 func (m *MockIPFIXAggregationProcess) ResetStatElementsInRecord(arg0 entities.Record) error {
 	m.ctrl.T.Helper()
@@ -267,6 +281,18 @@ func (mr *MockIPFIXAggregationProcessMockRecorder) ResetStatElementsInRecord(arg
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ResetStatElementsInRecord", reflect.TypeOf((*MockIPFIXAggregationProcess)(nil).ResetStatElementsInRecord), arg0)
 }
 
+// SetMetadataFilled mocks base method
+func (m *MockIPFIXAggregationProcess) SetMetadataFilled(arg0 intermediate.AggregationFlowRecord) {
+	m.ctrl.T.Helper()
+	m.ctrl.Call(m, "SetMetadataFilled", arg0)
+}
+
+// SetMetadataFilled indicates an expected call of SetMetadataFilled
+func (mr *MockIPFIXAggregationProcessMockRecorder) SetMetadataFilled(arg0 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SetMetadataFilled", reflect.TypeOf((*MockIPFIXAggregationProcess)(nil).SetMetadataFilled), arg0)
+}
+
 // Start mocks base method
 func (m *MockIPFIXAggregationProcess) Start() {
 	m.ctrl.T.Helper()
diff --git a/test/e2e/antreapolicy_test.go b/test/e2e/antreapolicy_test.go
index 6bd9deccb..bc1f488ec 100644
--- a/test/e2e/antreapolicy_test.go
+++ b/test/e2e/antreapolicy_test.go
@@ -684,7 +684,7 @@ func testInvalidTierANPRefDelete(t *testing.T) {
 	failOnError(k8sUtils.DeleteTier(tr.Name), t)
 }
 
-// testInvalidACNPPodSelectorNsSelectorMatchExpressions testes creating a ClusterNetworkPolicy with invalid LabelSelector(MatchExpressions)
+// testInvalidACNPPodSelectorNsSelectorMatchExpressions tests creating a ClusterNetworkPolicy with invalid LabelSelector(MatchExpressions)
 func testInvalidACNPPodSelectorNsSelectorMatchExpressions(t *testing.T) {
 	invalidLSErr := fmt.Errorf("create Antrea NetworkPolicy with namespaceSelector but matchExpressions invalid")
 
diff --git a/test/e2e/flowaggregator_test.go b/test/e2e/flowaggregator_test.go
index 5b6f2d8fc..4177a6ecd 100644
--- a/test/e2e/flowaggregator_test.go
+++ b/test/e2e/flowaggregator_test.go
@@ -15,6 +15,7 @@
 package e2e
 
 import (
+	"context"
 	"fmt"
 	"regexp"
 	"strconv"
@@ -30,6 +31,8 @@ import (
 	networkingv1 "k8s.io/api/networking/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/wait"
+
+	secv1alpha1 "antrea.io/antrea/pkg/apis/crd/v1alpha1"
 )
 
 /* Sample output from the collector:
@@ -95,15 +98,31 @@ AntreaProxy enabled (Inter-Node): Flow record from destination Node is ignored,
 */
 
 const (
-	ingressNetworkPolicyName = "test-flow-aggregator-networkpolicy-ingress"
-	egressNetworkPolicyName  = "test-flow-aggregator-networkpolicy-egress"
-	collectorCheckTimeout    = 10 * time.Second
+	ingressAllowNetworkPolicyName = "test-flow-aggregator-networkpolicy-ingress-allow"
+	ingressRejectANPName          = "test-flow-aggregator-anp-ingress-reject"
+	ingressDropANPName            = "test-flow-aggregator-anp-ingress-drop"
+	ingressDenyNPName             = "test-flow-aggregator-np-ingress-deny"
+	egressAllowNetworkPolicyName  = "test-flow-aggregator-networkpolicy-egress-allow"
+	egressRejectANPName           = "test-flow-aggregator-anp-engress-reject"
+	egressDropANPName             = "test-flow-aggregator-anp-engress-drop"
+	egressDenyNPName              = "test-flow-aggregator-np-egress-deny"
+	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 {
+	srcIP      string
+	dstIP      string
+	srcPodName string
+	dstPodName string
+}
+
 func TestFlowAggregator(t *testing.T) {
 	skipIfHasWindowsNodes(t)
 
@@ -114,21 +133,21 @@ func TestFlowAggregator(t *testing.T) {
 	defer teardownTest(t, data)
 	defer teardownFlowAggregator(t, data)
 
-	podAIPs, podBIPs, podCIPs, err := createPerftestPods(data)
+	podAIPs, podBIPs, podCIPs, podDIPs, podEIPs, err := createPerftestPods(data)
 	if err != nil {
 		t.Fatalf("Error when creating perftest Pods: %v", err)
 	}
 
 	if v4Enabled {
-		t.Run("IPv4", func(t *testing.T) { testHelper(t, data, podAIPs, podBIPs, podCIPs, false) })
+		t.Run("IPv4", func(t *testing.T) { testHelper(t, data, podAIPs, podBIPs, podCIPs, podDIPs, podEIPs, false) })
 	}
 
 	if v6Enabled {
-		t.Run("IPv6", func(t *testing.T) { testHelper(t, data, podAIPs, podBIPs, podCIPs, true) })
+		t.Run("IPv6", func(t *testing.T) { testHelper(t, data, podAIPs, podBIPs, podCIPs, podDIPs, podEIPs, true) })
 	}
 }
 
-func testHelper(t *testing.T, data *TestData, podAIPs *PodIPs, podBIPs *PodIPs, podCIPs *PodIPs, isIPv6 bool) {
+func testHelper(t *testing.T, data *TestData, podAIPs, podBIPs, podCIPs, podDIPs, podEIPs *PodIPs, isIPv6 bool) {
 	svcB, svcC, err := createPerftestServices(data, isIPv6)
 	if err != nil {
 		t.Fatalf("Error when creating perftest Services: %v", err)
@@ -137,6 +156,8 @@ func testHelper(t *testing.T, data *TestData, podAIPs *PodIPs, podBIPs *PodIPs,
 	// 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")
@@ -152,11 +173,115 @@ func testHelper(t *testing.T, data *TestData, podAIPs *PodIPs, podBIPs *PodIPs,
 				}
 			}
 		}()
-		// 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, checkBandwidth)
+		}
+	})
+
+	// IntraNodeDenyConnIngressANP tests the case, where Pods are deployed on same Node with an Antrea ingress deny policy rule
+	// applied to destination Pod (one reject rule, one drop rule) and their flow information is exported as IPFIX flow records.
+	// perftest-a -> perftest-b (Ingress reject), perftest-a -> perftest-d (Ingress drop)
+	t.Run("IntraNodeDenyConnIngressANP", func(t *testing.T) {
+		skipIfAntreaPolicyDisabled(t, data)
+		anp1, anp2 := deployDenyAntreaNetworkPolicies(t, data, "perftest-a", "perftest-b", "perftest-d", true)
+		defer func() {
+			if anp1 != nil {
+				if err = data.deleteAntreaNetworkpolicy(anp1); err != nil {
+					t.Errorf("Error when deleting Antrea Network Policy: %v", err)
+				}
+			}
+			if anp2 != nil {
+				if err = data.deleteAntreaNetworkpolicy(anp2); err != nil {
+					t.Errorf("Error when deleting Antrea Network Policy: %v", err)
+				}
+			}
+		}()
+		testFlow1 := testFlow{
+			srcPodName: "perftest-a",
+			dstPodName: "perftest-b",
+		}
+		testFlow2 := testFlow{
+			srcPodName: "perftest-a",
+			dstPodName: "perftest-d",
+		}
+		if !isIPv6 {
+			testFlow1.srcIP, testFlow1.dstIP, testFlow2.srcIP, testFlow2.dstIP = podAIPs.ipv4.String(), podBIPs.ipv4.String(), podAIPs.ipv4.String(), podDIPs.ipv4.String()
+			checkRecordsForDenyFlows(t, data, testFlow1, testFlow2, isIPv6, true, true)
+		} else {
+			testFlow1.srcIP, testFlow1.dstIP, testFlow2.srcIP, testFlow2.dstIP = podAIPs.ipv6.String(), podBIPs.ipv6.String(), podAIPs.ipv6.String(), podDIPs.ipv6.String()
+			checkRecordsForDenyFlows(t, data, testFlow1, testFlow2, isIPv6, true, true)
+		}
+	})
+
+	// IntraNodeDenyConnEgressANP tests the case, where Pods are deployed on same Node with an Antrea egress deny policy rule
+	// applied to source Pods (one reject rule, one drop rule) and their flow information is exported as IPFIX flow records.
+	// perftest-a (Egress reject) -> perftest-b , perftest-a (Egress drop) -> perftest-d
+	t.Run("IntraNodeDenyConnEgressANP", func(t *testing.T) {
+		skipIfAntreaPolicyDisabled(t, data)
+		anp1, anp2 := deployDenyAntreaNetworkPolicies(t, data, "perftest-a", "perftest-b", "perftest-d", false)
+		defer func() {
+			if anp1 != nil {
+				if err = data.deleteAntreaNetworkpolicy(anp1); err != nil {
+					t.Errorf("Error when deleting Antrea Network Policy: %v", err)
+				}
+			}
+			if anp2 != nil {
+				if err = data.deleteAntreaNetworkpolicy(anp2); err != nil {
+					t.Errorf("Error when deleting Antrea Network Policy: %v", err)
+				}
+			}
+		}()
+		testFlow1 := testFlow{
+			srcPodName: "perftest-a",
+			dstPodName: "perftest-b",
+		}
+		testFlow2 := testFlow{
+			srcPodName: "perftest-a",
+			dstPodName: "perftest-d",
+		}
+		if !isIPv6 {
+			testFlow1.srcIP, testFlow1.dstIP, testFlow2.srcIP, testFlow2.dstIP = podAIPs.ipv4.String(), podBIPs.ipv4.String(), podAIPs.ipv4.String(), podDIPs.ipv4.String()
+			checkRecordsForDenyFlows(t, data, testFlow1, testFlow2, isIPv6, true, true)
 		} else {
-			checkRecordsForFlows(t, data, podAIPs.ipv6.String(), podBIPs.ipv6.String(), isIPv6, true, false, true, false)
+			testFlow1.srcIP, testFlow1.dstIP, testFlow2.srcIP, testFlow2.dstIP = podAIPs.ipv6.String(), podBIPs.ipv6.String(), podAIPs.ipv6.String(), podDIPs.ipv6.String()
+			checkRecordsForDenyFlows(t, data, testFlow1, testFlow2, isIPv6, true, true)
+		}
+	})
+
+	// IntraNodeDenyConnNP tests the case, where Pods are deployed on same Node with an ingress and an egress deny policy rule
+	// applied to one destination Pod, one source Pod, respectively and their flow information is exported as IPFIX flow records.
+	// perftest-a -> perftest-b (Ingress deny), perftest-d (Egress deny) -> perftest-a
+	t.Run("IntraNodeDenyConnNP", func(t *testing.T) {
+		skipIfAntreaPolicyDisabled(t, data)
+		np1, np2 := deployDenyNetworkPolicies(t, data, "perftest-b", "perftest-d")
+		defer func() {
+			if np1 != nil {
+				if err = data.deleteNetworkpolicy(np1); err != nil {
+					t.Errorf("Error when deleting Network Policy: %v", err)
+				}
+			}
+			if np2 != nil {
+				if err = data.deleteNetworkpolicy(np2); err != nil {
+					t.Errorf("Error when deleting Network Policy: %v", err)
+				}
+			}
+		}()
+		testFlow1 := testFlow{
+			srcPodName: "perftest-a",
+			dstPodName: "perftest-b",
+		}
+		testFlow2 := testFlow{
+			srcPodName: "perftest-d",
+			dstPodName: "perftest-a",
+		}
+		if !isIPv6 {
+			testFlow1.srcIP, testFlow1.dstIP, testFlow2.srcIP, testFlow2.dstIP = podAIPs.ipv4.String(), podBIPs.ipv4.String(), podDIPs.ipv4.String(), podAIPs.ipv4.String()
+			checkRecordsForDenyFlows(t, data, testFlow1, testFlow2, isIPv6, true, false)
+		} else {
+			testFlow1.srcIP, testFlow1.dstIP, testFlow2.srcIP, testFlow2.dstIP = podAIPs.ipv6.String(), podBIPs.ipv6.String(), podDIPs.ipv6.String(), podAIPs.ipv6.String()
+			checkRecordsForDenyFlows(t, data, testFlow1, testFlow2, isIPv6, true, false)
 		}
 	})
 
@@ -177,20 +302,124 @@ func testHelper(t *testing.T, data *TestData, podAIPs *PodIPs, podBIPs *PodIPs,
 			}
 		}()
 		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)
+		}
+	})
+
+	// InterNodeDenyConnIngressANP tests the case, where Pods are deployed on different Nodes with an Antrea ingress deny policy rule
+	// applied to destination Pod (one reject rule, one drop rule) and their flow information is exported as IPFIX flow records.
+	// perftest-a -> perftest-c (Ingress reject), perftest-a -> perftest-e (Ingress drop)
+	t.Run("InterNodeDenyConnIngressANP", func(t *testing.T) {
+		skipIfAntreaPolicyDisabled(t, data)
+		anp1, anp2 := deployDenyAntreaNetworkPolicies(t, data, "perftest-a", "perftest-c", "perftest-e", true)
+		defer func() {
+			if anp1 != nil {
+				if err = data.deleteAntreaNetworkpolicy(anp1); err != nil {
+					t.Errorf("Error when deleting Antrea Network Policy: %v", err)
+				}
+			}
+			if anp2 != nil {
+				if err = data.deleteAntreaNetworkpolicy(anp2); err != nil {
+					t.Errorf("Error when deleting Antrea Network Policy: %v", err)
+				}
+			}
+		}()
+		testFlow1 := testFlow{
+			srcPodName: "perftest-a",
+			dstPodName: "perftest-c",
+		}
+		testFlow2 := testFlow{
+			srcPodName: "perftest-a",
+			dstPodName: "perftest-e",
+		}
+		if !isIPv6 {
+			testFlow1.srcIP, testFlow1.dstIP, testFlow2.srcIP, testFlow2.dstIP = podAIPs.ipv4.String(), podCIPs.ipv4.String(), podAIPs.ipv4.String(), podEIPs.ipv4.String()
+			checkRecordsForDenyFlows(t, data, testFlow1, testFlow2, isIPv6, false, true)
+		} else {
+			testFlow1.srcIP, testFlow1.dstIP, testFlow2.srcIP, testFlow2.dstIP = podAIPs.ipv6.String(), podCIPs.ipv6.String(), podAIPs.ipv6.String(), podEIPs.ipv6.String()
+			checkRecordsForDenyFlows(t, data, testFlow1, testFlow2, isIPv6, false, true)
+		}
+	})
+
+	// InterNodeDenyConnEgressANP tests the case, where Pods are deployed on different Nodes with an Antrea egress deny policy rule
+	// applied to source Pod (one reject rule, one drop rule) and their flow information is exported as IPFIX flow records.
+	// perftest-a (Egress reject) -> perftest-c, perftest-a (Egress drop)-> perftest-e
+	t.Run("InterNodeDenyConnEgressANP", func(t *testing.T) {
+		skipIfAntreaPolicyDisabled(t, data)
+		anp1, anp2 := deployDenyAntreaNetworkPolicies(t, data, "perftest-a", "perftest-c", "perftest-e", false)
+		defer func() {
+			if anp1 != nil {
+				if err = data.deleteAntreaNetworkpolicy(anp1); err != nil {
+					t.Errorf("Error when deleting Antrea Network Policy: %v", err)
+				}
+			}
+			if anp2 != nil {
+				if err = data.deleteAntreaNetworkpolicy(anp2); err != nil {
+					t.Errorf("Error when deleting Antrea Network Policy: %v", err)
+				}
+			}
+		}()
+		testFlow1 := testFlow{
+			srcPodName: "perftest-a",
+			dstPodName: "perftest-c",
+		}
+		testFlow2 := testFlow{
+			srcPodName: "perftest-a",
+			dstPodName: "perftest-e",
+		}
+		if !isIPv6 {
+			testFlow1.srcIP, testFlow1.dstIP, testFlow2.srcIP, testFlow2.dstIP = podAIPs.ipv4.String(), podCIPs.ipv4.String(), podAIPs.ipv4.String(), podEIPs.ipv4.String()
+			checkRecordsForDenyFlows(t, data, testFlow1, testFlow2, isIPv6, false, true)
+		} else {
+			testFlow1.srcIP, testFlow1.dstIP, testFlow2.srcIP, testFlow2.dstIP = podAIPs.ipv6.String(), podCIPs.ipv6.String(), podAIPs.ipv6.String(), podEIPs.ipv6.String()
+			checkRecordsForDenyFlows(t, data, testFlow1, testFlow2, isIPv6, false, true)
+		}
+	})
+
+	// InterNodeDenyConnNP tests the case, where Pods are deployed on different Nodes with an ingress and an egress deny policy rule
+	// applied to one destination Pod, one source Pod, respectively and their flow information is exported as IPFIX flow records.
+	// perftest-a -> perftest-c (Ingress deny), perftest-b (Egress deny) -> perftest-e
+	t.Run("InterNodeDenyConnNP", func(t *testing.T) {
+		skipIfAntreaPolicyDisabled(t, data)
+		np1, np2 := deployDenyNetworkPolicies(t, data, "perftest-c", "perftest-b")
+		defer func() {
+			if np1 != nil {
+				if err = data.deleteNetworkpolicy(np1); err != nil {
+					t.Errorf("Error when deleting Network Policy: %v", err)
+				}
+			}
+			if np2 != nil {
+				if err = data.deleteNetworkpolicy(np2); err != nil {
+					t.Errorf("Error when deleting Network Policy: %v", err)
+				}
+			}
+		}()
+		testFlow1 := testFlow{
+			srcPodName: "perftest-a",
+			dstPodName: "perftest-c",
+		}
+		testFlow2 := testFlow{
+			srcPodName: "perftest-b",
+			dstPodName: "perftest-e",
+		}
+		if !isIPv6 {
+			testFlow1.srcIP, testFlow1.dstIP, testFlow2.srcIP, testFlow2.dstIP = podAIPs.ipv4.String(), podCIPs.ipv4.String(), podBIPs.ipv4.String(), podEIPs.ipv4.String()
+			checkRecordsForDenyFlows(t, data, testFlow1, testFlow2, isIPv6, false, false)
+		} else {
+			testFlow1.srcIP, testFlow1.dstIP, testFlow2.srcIP, testFlow2.dstIP = podAIPs.ipv6.String(), podCIPs.ipv6.String(), podBIPs.ipv6.String(), podEIPs.ipv6.String()
+			checkRecordsForDenyFlows(t, data, testFlow1, testFlow2, isIPv6, false, false)
 		}
 	})
 
 	// 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)
 		}
 	})
 
@@ -198,26 +427,39 @@ func testHelper(t *testing.T, data *TestData, podAIPs *PodIPs, podBIPs *PodIPs,
 	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.
@@ -226,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)
 
@@ -234,63 +485,162 @@ 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, ingressNetworkPolicyName) {
-						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, egressNetworkPolicyName) {
-						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")
+				}
+				if !strings.Contains(record, fmt.Sprintf("%s: %s", "ingressNetworkPolicyNamespace", testNamespace)) {
+					t.Errorf("Record does not have correct ingressNetworkPolicyNamespace")
 				}
-				// Check the bandwidth using octetDeltaCount in data record.
-				if checkBandwidth {
-					checkBandwidthFromRecord(t, record, bandwidth)
+				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))
 }
 
+func checkRecordsForDenyFlows(t *testing.T, data *TestData, testFlow1, testFlow2 testFlow, isIPv6 bool, isIntraNode bool, isANP bool) {
+	timeStart := time.Now()
+	var cmdStr1, cmdStr2 string
+	if !isIPv6 {
+		cmdStr1 = fmt.Sprintf("iperf3 -c %s -n 1", testFlow1.dstIP)
+		cmdStr2 = fmt.Sprintf("iperf3 -c %s -n 1", testFlow2.dstIP)
+	} else {
+		cmdStr1 = fmt.Sprintf("iperf3 -6 -c %s -n 1", testFlow1.dstIP)
+		cmdStr2 = fmt.Sprintf("iperf3 -6 -c %s -n 1", testFlow2.dstIP)
+	}
+	_, _, err := data.runCommandFromPod(testNamespace, testFlow1.srcPodName, "", []string{"timeout", "2", "bash", "-c", cmdStr1})
+	assert.Error(t, err)
+	_, _, err = data.runCommandFromPod(testNamespace, testFlow2.srcPodName, "", []string{"timeout", "2", "bash", "-c", cmdStr2})
+	assert.Error(t, err)
+
+	err = wait.Poll(250*time.Millisecond, collectorCheckTimeout, func() (bool, error) {
+		rc, collectorOutput, _, err := provider.RunCommandOnNode(controlPlaneNodeName(), fmt.Sprintf("kubectl logs --since=%v ipfix-collector -n antrea-test", time.Since(timeStart).String()))
+		if err != nil || rc != 0 {
+			return false, err
+		}
+		return strings.Contains(collectorOutput, testFlow1.srcIP) && strings.Contains(collectorOutput, testFlow2.srcIP), nil
+	})
+	require.NoErrorf(t, err, "IPFIX collector did not receive the expected records and timed out with error: %v", err)
+
+	rc, collectorOutput, _, err := provider.RunCommandOnNode(controlPlaneNodeName(), fmt.Sprintf("kubectl logs --since=%v ipfix-collector -n antrea-test", time.Since(timeStart).String()))
+	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)
+	for _, record := range recordSlices {
+		var srcPodName, dstPodName string
+		if strings.Contains(record, testFlow1.srcIP) && strings.Contains(record, testFlow1.dstIP) {
+			srcPodName = testFlow1.srcPodName
+			dstPodName = testFlow1.dstPodName
+		} else if strings.Contains(record, testFlow2.srcIP) && strings.Contains(record, testFlow2.dstIP) {
+			srcPodName = testFlow2.srcPodName
+			dstPodName = testFlow2.dstPodName
+		}
+		if (strings.Contains(record, testFlow1.srcIP) && strings.Contains(record, testFlow1.dstIP)) || (strings.Contains(record, testFlow2.srcIP) && strings.Contains(record, testFlow2.dstIP)) {
+			ingressRejectStr := fmt.Sprintf("ingressNetworkPolicyRuleAction: %d", ipfixregistry.NetworkPolicyRuleActionReject)
+			ingressDropStr := fmt.Sprintf("ingressNetworkPolicyRuleAction: %d", ipfixregistry.NetworkPolicyRuleActionDrop)
+			egressRejectStr := fmt.Sprintf("egressNetworkPolicyRuleAction: %d", ipfixregistry.NetworkPolicyRuleActionReject)
+			egressDropStr := fmt.Sprintf("egressNetworkPolicyRuleAction: %d", ipfixregistry.NetworkPolicyRuleActionDrop)
+
+			if isIntraNode {
+				checkPodAndNodeData(t, record, srcPodName, controlPlaneNodeName(), dstPodName, controlPlaneNodeName())
+				checkFlowType(t, record, ipfixregistry.FlowTypeIntraNode)
+			} else {
+				checkPodAndNodeData(t, record, srcPodName, controlPlaneNodeName(), dstPodName, workerNodeName(1))
+				checkFlowType(t, record, ipfixregistry.FlowTypeInterNode)
+			}
+
+			if !isANP { // K8s Network Policies
+				if strings.Contains(record, ingressDropStr) && !strings.Contains(record, ingressDropANPName) {
+					assert.Contains(t, record, testFlow1.dstIP)
+				} else if strings.Contains(record, egressDropStr) && !strings.Contains(record, egressDropANPName) {
+					assert.Contains(t, record, testFlow2.dstIP)
+				}
+			} else { // Antrea Network Policies
+				if strings.Contains(record, ingressRejectStr) {
+					assert.Contains(t, record, ingressRejectANPName, "Record does not have Antrea NetworkPolicy name with ingress reject rule")
+					assert.Contains(t, record, fmt.Sprintf("%s: %s", "ingressNetworkPolicyNamespace", testNamespace), "Record does not have correct ingressNetworkPolicyNamespace")
+				} else if strings.Contains(record, ingressDropStr) {
+					assert.Contains(t, record, ingressDropANPName, "Record does not have Antrea NetworkPolicy name with ingress drop rule")
+					assert.Contains(t, record, fmt.Sprintf("%s: %s", "ingressNetworkPolicyNamespace", testNamespace), "Record does not have correct ingressNetworkPolicyNamespace")
+				} else if strings.Contains(record, egressRejectStr) {
+					assert.Contains(t, record, egressRejectANPName, "Record does not have Antrea NetworkPolicy name with egress reject rule")
+					assert.Contains(t, record, fmt.Sprintf("%s: %s", "egressNetworkPolicyNamespace", testNamespace), "Record does not have correct egressNetworkPolicyNamespace")
+				} else if strings.Contains(record, egressDropStr) {
+					assert.Contains(t, record, egressDropANPName, "Record does not have Antrea NetworkPolicy name with egress drop rule")
+					assert.Contains(t, record, fmt.Sprintf("%s: %s", "egressNetworkPolicyNamespace", testNamespace), "Record does not have correct egressNetworkPolicyNamespace")
+				}
+			}
+
+		}
+	}
+}
+
+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")
@@ -312,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.
@@ -363,7 +695,7 @@ func getRecordsFromOutput(output string) []string {
 func deployNetworkPolicies(t *testing.T, data *TestData, srcPod, dstPod string) (np1 *networkingv1.NetworkPolicy, np2 *networkingv1.NetworkPolicy) {
 	// Add NetworkPolicy between two iperf Pods.
 	var err error
-	np1, err = data.createNetworkPolicy(ingressNetworkPolicyName, &networkingv1.NetworkPolicySpec{
+	np1, err = data.createNetworkPolicy(ingressAllowNetworkPolicyName, &networkingv1.NetworkPolicySpec{
 		PodSelector: metav1.LabelSelector{},
 		PolicyTypes: []networkingv1.PolicyType{networkingv1.PolicyTypeIngress},
 		Ingress: []networkingv1.NetworkPolicyIngressRule{{
@@ -379,7 +711,7 @@ func deployNetworkPolicies(t *testing.T, data *TestData, srcPod, dstPod string)
 	if err != nil {
 		t.Errorf("Error when creating Network Policy: %v", err)
 	}
-	np2, err = data.createNetworkPolicy(egressNetworkPolicyName, &networkingv1.NetworkPolicySpec{
+	np2, err = data.createNetworkPolicy(egressAllowNetworkPolicyName, &networkingv1.NetworkPolicySpec{
 		PodSelector: metav1.LabelSelector{},
 		PolicyTypes: []networkingv1.PolicyType{networkingv1.PolicyTypeEgress},
 		Egress: []networkingv1.NetworkPolicyEgressRule{{
@@ -403,31 +735,108 @@ func deployNetworkPolicies(t *testing.T, data *TestData, srcPod, dstPod string)
 	return np1, np2
 }
 
-func createPerftestPods(data *TestData) (podAIPs *PodIPs, podBIPs *PodIPs, podCIPs *PodIPs, err error) {
+func deployDenyAntreaNetworkPolicies(t *testing.T, data *TestData, srcPod, podReject, podDrop string, isIngress bool) (anp1 *secv1alpha1.NetworkPolicy, anp2 *secv1alpha1.NetworkPolicy) {
+	var err error
+	k8sUtils, _ = NewKubernetesUtils(data)
+	if isIngress {
+		if anp1, err = data.createANPDenyIngress("antrea-e2e", podReject, ingressRejectANPName, true); err != nil {
+			t.Fatalf("Error when creating Antrea NetworkPolicy: %v", err)
+		}
+		if anp2, err = data.createANPDenyIngress("antrea-e2e", podDrop, ingressDropANPName, false); err != nil {
+			t.Fatalf("Error when creating Antrea NetworkPolicy: %v", err)
+		}
+	} else {
+		if anp1, err = data.createANPDenyEgress(srcPod, podReject, egressRejectANPName, true); err != nil {
+			t.Fatalf("Error when creating Antrea NetworkPolicy: %v", err)
+		}
+		if anp2, err = data.createANPDenyEgress(srcPod, podDrop, egressDropANPName, false); err != nil {
+			t.Fatalf("Error when creating Antrea NetworkPolicy: %v", err)
+		}
+	}
+
+	// Wait for Antrea NetworkPolicy to be realized.
+	if err := WaitNetworkPolicyRealize(2, data); err != nil {
+		t.Errorf("Error when waiting for Antrea Network Policy to be realized: %v", err)
+	}
+	t.Log("Antrea Network Policies are realized.")
+	return anp1, anp2
+}
+
+func deployDenyNetworkPolicies(t *testing.T, data *TestData, pod1, pod2 string) (np1 *networkingv1.NetworkPolicy, np2 *networkingv1.NetworkPolicy) {
+	np1, err := data.createNetworkPolicy(ingressDenyNPName, &networkingv1.NetworkPolicySpec{
+		PodSelector: metav1.LabelSelector{
+			MatchLabels: map[string]string{
+				"antrea-e2e": pod1,
+			},
+		},
+		PolicyTypes: []networkingv1.PolicyType{networkingv1.PolicyTypeIngress},
+		Ingress:     []networkingv1.NetworkPolicyIngressRule{},
+	})
+	if err != nil {
+		t.Errorf("Error when creating Network Policy: %v", err)
+	}
+	np2, err = data.createNetworkPolicy(egressDenyNPName, &networkingv1.NetworkPolicySpec{
+		PodSelector: metav1.LabelSelector{
+			MatchLabels: map[string]string{
+				"antrea-e2e": pod2,
+			},
+		},
+		PolicyTypes: []networkingv1.PolicyType{networkingv1.PolicyTypeEgress},
+		Egress:      []networkingv1.NetworkPolicyEgressRule{},
+	})
+	if err != nil {
+		t.Errorf("Error when creating Network Policy: %v", err)
+	}
+	// Wait for NetworkPolicy to be realized.
+	if err := WaitNetworkPolicyRealize(2, data); err != nil {
+		t.Errorf("Error when waiting for Network Policies to be realized: %v", err)
+	}
+	t.Log("Network Policies are realized.")
+	return np1, np2
+}
+
+func createPerftestPods(data *TestData) (podAIPs *PodIPs, podBIPs *PodIPs, podCIPs *PodIPs, podDIPs *PodIPs, podEIPs *PodIPs, err error) {
 	if err := data.createPodOnNode("perftest-a", controlPlaneNodeName(), perftoolImage, nil, nil, nil, nil, false, nil); err != nil {
-		return nil, nil, nil, fmt.Errorf("Error when creating the perftest client Pod: %v", err)
+		return nil, nil, nil, nil, nil, fmt.Errorf("Error when creating the perftest client Pod: %v", err)
 	}
 	podAIPs, err = data.podWaitForIPs(defaultTimeout, "perftest-a", testNamespace)
 	if err != nil {
-		return nil, nil, nil, fmt.Errorf("Error when waiting for the perftest client Pod: %v", err)
+		return nil, nil, nil, nil, nil, fmt.Errorf("Error when waiting for the perftest client Pod: %v", err)
 	}
 
 	if err := data.createPodOnNode("perftest-b", controlPlaneNodeName(), perftoolImage, nil, nil, nil, []v1.ContainerPort{{Protocol: v1.ProtocolTCP, ContainerPort: iperfPort}}, false, nil); err != nil {
-		return nil, nil, nil, fmt.Errorf("Error when creating the perftest server Pod: %v", err)
+		return nil, nil, nil, nil, nil, fmt.Errorf("Error when creating the perftest server Pod: %v", err)
 	}
 	podBIPs, err = data.podWaitForIPs(defaultTimeout, "perftest-b", testNamespace)
 	if err != nil {
-		return nil, nil, nil, fmt.Errorf("Error when getting the perftest server Pod's IPs: %v", err)
+		return nil, nil, nil, nil, nil, fmt.Errorf("Error when getting the perftest server Pod's IPs: %v", err)
 	}
 
 	if err := data.createPodOnNode("perftest-c", workerNodeName(1), perftoolImage, nil, nil, nil, []v1.ContainerPort{{Protocol: v1.ProtocolTCP, ContainerPort: iperfPort}}, false, nil); err != nil {
-		return nil, nil, nil, fmt.Errorf("Error when creating the perftest server Pod: %v", err)
+		return nil, nil, nil, nil, nil, fmt.Errorf("Error when creating the perftest server Pod: %v", err)
 	}
 	podCIPs, err = data.podWaitForIPs(defaultTimeout, "perftest-c", testNamespace)
 	if err != nil {
-		return nil, nil, nil, fmt.Errorf("Error when getting the perftest server Pod's IPs: %v", err)
+		return nil, nil, nil, nil, nil, fmt.Errorf("Error when getting the perftest server Pod's IPs: %v", err)
+	}
+
+	if err := data.createPodOnNode("perftest-d", controlPlaneNodeName(), perftoolImage, nil, nil, nil, []v1.ContainerPort{{Protocol: v1.ProtocolTCP, ContainerPort: iperfPort}}, false, nil); err != nil {
+		return nil, nil, nil, nil, nil, fmt.Errorf("Error when creating the perftest server Pod: %v", err)
+	}
+	podDIPs, err = data.podWaitForIPs(defaultTimeout, "perftest-d", testNamespace)
+	if err != nil {
+		return nil, nil, nil, nil, nil, fmt.Errorf("Error when getting the perftest server Pod's IPs: %v", err)
+	}
+
+	if err := data.createPodOnNode("perftest-e", workerNodeName(1), perftoolImage, nil, nil, nil, []v1.ContainerPort{{Protocol: v1.ProtocolTCP, ContainerPort: iperfPort}}, false, nil); err != nil {
+		return nil, nil, nil, nil, nil, fmt.Errorf("Error when creating the perftest server Pod: %v", err)
+	}
+	podEIPs, err = data.podWaitForIPs(defaultTimeout, "perftest-e", testNamespace)
+	if err != nil {
+		return nil, nil, nil, nil, nil, fmt.Errorf("Error when getting the perftest server Pod's IPs: %v", err)
 	}
-	return podAIPs, podBIPs, podCIPs, nil
+
+	return podAIPs, podBIPs, podCIPs, podDIPs, podEIPs, nil
 }
 
 func createPerftestServices(data *TestData, isIPv6 bool) (svcB *corev1.Service, svcC *corev1.Service, err error) {
@@ -457,3 +866,54 @@ func deletePerftestServices(t *testing.T, data *TestData) {
 		}
 	}
 }
+
+// createANPDenyEgress creates an Antrea NetworkPolicy that denies egress traffic for pods of specific label.
+func (data *TestData) createANPDenyEgress(srcPod, dstPod, name string, isReject bool) (*secv1alpha1.NetworkPolicy, error) {
+	dropACT := secv1alpha1.RuleActionDrop
+	if isReject {
+		dropACT = secv1alpha1.RuleActionReject
+	}
+	anp := secv1alpha1.NetworkPolicy{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: name,
+			Labels: map[string]string{
+				"antrea-e2e": srcPod,
+			},
+		},
+		Spec: secv1alpha1.NetworkPolicySpec{
+			Tier:     defaultTierName,
+			Priority: 250,
+			AppliedTo: []secv1alpha1.NetworkPolicyPeer{
+				{
+					PodSelector: &metav1.LabelSelector{
+						MatchLabels: map[string]string{
+							"antrea-e2e": srcPod,
+						},
+					},
+				},
+			},
+			Ingress: []secv1alpha1.Rule{},
+			Egress: []secv1alpha1.Rule{
+				{
+					Action: &dropACT,
+					Ports:  []secv1alpha1.NetworkPolicyPort{},
+					From:   []secv1alpha1.NetworkPolicyPeer{},
+					To: []secv1alpha1.NetworkPolicyPeer{
+						{
+							PodSelector: &metav1.LabelSelector{
+								MatchLabels: map[string]string{
+									"antrea-e2e": dstPod,
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+	anpCreated, err := k8sUtils.crdClient.CrdV1alpha1().NetworkPolicies(testNamespace).Create(context.TODO(), &anp, metav1.CreateOptions{})
+	if err != nil {
+		return nil, err
+	}
+	return anpCreated, nil
+}
diff --git a/test/e2e/framework.go b/test/e2e/framework.go
index b94d6567c..c7fecbdef 100644
--- a/test/e2e/framework.go
+++ b/test/e2e/framework.go
@@ -101,10 +101,15 @@ const (
 	busyboxImage        = "projects.registry.vmware.com/library/busybox"
 	nginxImage          = "projects.registry.vmware.com/antrea/nginx"
 	perftoolImage       = "projects.registry.vmware.com/antrea/perftool"
-	ipfixCollectorImage = "projects.registry.vmware.com/antrea/ipfix-collector:v0.4.7"
+	ipfixCollectorImage = "projects.registry.vmware.com/antrea/ipfix-collector:v0.5.2"
 	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},
-		{"inactiveFlowExportTimeout", "\"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/flowexporter_test.go b/test/integration/agent/flowexporter_test.go
index c70fd5f9c..df96d842f 100644
--- a/test/integration/agent/flowexporter_test.go
+++ b/test/integration/agent/flowexporter_test.go
@@ -39,24 +39,6 @@ import (
 
 const testPollInterval = 0 // Not used in the test, hence 0.
 
-func makeTuple(srcIP *net.IP, dstIP *net.IP, protoID uint8, srcPort uint16, dstPort uint16) (*flowexporter.Tuple, *flowexporter.Tuple) {
-	tuple := &flowexporter.Tuple{
-		SourceAddress:      *srcIP,
-		DestinationAddress: *dstIP,
-		Protocol:           protoID,
-		SourcePort:         srcPort,
-		DestinationPort:    dstPort,
-	}
-	revTuple := &flowexporter.Tuple{
-		SourceAddress:      *dstIP,
-		DestinationAddress: *srcIP,
-		Protocol:           protoID,
-		SourcePort:         dstPort,
-		DestinationPort:    srcPort,
-	}
-	return tuple, revTuple
-}
-
 func createConnsForTest() ([]*flowexporter.Connection, []*flowexporter.ConnectionKey) {
 	// Reference for flow timestamp
 	refTime := time.Now()
@@ -64,7 +46,7 @@ func createConnsForTest() ([]*flowexporter.Connection, []*flowexporter.Connectio
 	testConns := make([]*flowexporter.Connection, 2)
 	testConnKeys := make([]*flowexporter.ConnectionKey, 2)
 	// Flow-1
-	tuple1, revTuple1 := makeTuple(&net.IP{1, 2, 3, 4}, &net.IP{4, 3, 2, 1}, 6, 65280, 255)
+	tuple1 := flowexporter.Tuple{SourceAddress: net.IP{1, 2, 3, 4}, DestinationAddress: net.IP{4, 3, 2, 1}, Protocol: 6, SourcePort: 65280, DestinationPort: 255}
 	testConn1 := &flowexporter.Connection{
 		StartTime:       refTime.Add(-(time.Second * 50)),
 		StopTime:        refTime,
@@ -72,14 +54,13 @@ func createConnsForTest() ([]*flowexporter.Connection, []*flowexporter.Connectio
 		OriginalBytes:   0xbaaaaa0000000000,
 		ReversePackets:  0xff,
 		ReverseBytes:    0xbaaa,
-		TupleOrig:       *tuple1,
-		TupleReply:      *revTuple1,
+		FlowKey:         tuple1,
 	}
 	testConnKey1 := flowexporter.NewConnectionKey(testConn1)
 	testConns[0] = testConn1
 	testConnKeys[0] = &testConnKey1
 	// Flow-2
-	tuple2, revTuple2 := makeTuple(&net.IP{5, 6, 7, 8}, &net.IP{8, 7, 6, 5}, 6, 60001, 200)
+	tuple2 := flowexporter.Tuple{SourceAddress: net.IP{5, 6, 7, 8}, DestinationAddress: net.IP{8, 7, 6, 5}, Protocol: 6, SourcePort: 60001, DestinationPort: 200}
 	testConn2 := &flowexporter.Connection{
 		StartTime:       refTime.Add(-(time.Second * 20)),
 		StopTime:        refTime,
@@ -87,8 +68,7 @@ func createConnsForTest() ([]*flowexporter.Connection, []*flowexporter.Connectio
 		OriginalBytes:   0xcbbb,
 		ReversePackets:  0xbbbb,
 		ReverseBytes:    0xcbbbb0000000000,
-		TupleOrig:       *tuple2,
-		TupleReply:      *revTuple2,
+		FlowKey:         tuple2,
 	}
 	testConnKey2 := flowexporter.NewConnectionKey(testConn2)
 	testConns[1] = testConn2
@@ -121,28 +101,28 @@ func TestConnectionStoreAndFlowRecords(t *testing.T) {
 	// Prepare connections and interface config for test
 	testConns, testConnKeys := createConnsForTest()
 	testIfConfigs := make([]*interfacestore.InterfaceConfig, 2)
-	testIfConfigs[0] = prepareInterfaceConfigs("1", "pod1", "ns1", "interface1", &testConns[0].TupleOrig.SourceAddress)
-	testIfConfigs[1] = prepareInterfaceConfigs("2", "pod2", "ns2", "interface2", &testConns[1].TupleOrig.DestinationAddress)
+	testIfConfigs[0] = prepareInterfaceConfigs("1", "pod1", "ns1", "interface1", &testConns[0].FlowKey.SourceAddress)
+	testIfConfigs[1] = prepareInterfaceConfigs("2", "pod2", "ns2", "interface2", &testConns[1].FlowKey.DestinationAddress)
 	// Create connectionStore, FlowRecords and associated mocks
 	connDumperMock := connectionstest.NewMockConnTrackDumper(ctrl)
 	ifStoreMock := interfacestoretest.NewMockInterfaceStore(ctrl)
 	npQuerier := queriertest.NewMockAgentNetworkPolicyInfoQuerier(ctrl)
 	// TODO: Enhance the integration test by testing service.
-	connStore := connections.NewConnectionStore(connDumperMock, flowrecords.NewFlowRecords(), ifStoreMock, true, false, nil, npQuerier, testPollInterval)
+	conntrackConnStore := connections.NewConntrackConnectionStore(connDumperMock, flowrecords.NewFlowRecords(), ifStoreMock, true, false, nil, npQuerier, testPollInterval)
 	// Expect calls for connStore.poll and other callees
 	connDumperMock.EXPECT().DumpFlows(uint16(openflow.CtZone)).Return(testConns, 0, nil)
 	connDumperMock.EXPECT().GetMaxConnections().Return(0, nil)
 	for i, testConn := range testConns {
 		if i == 0 {
-			ifStoreMock.EXPECT().GetInterfaceByIP(testConn.TupleOrig.SourceAddress.String()).Return(testIfConfigs[i], true)
-			ifStoreMock.EXPECT().GetInterfaceByIP(testConn.TupleOrig.DestinationAddress.String()).Return(nil, false)
+			ifStoreMock.EXPECT().GetInterfaceByIP(testConn.FlowKey.SourceAddress.String()).Return(testIfConfigs[i], true)
+			ifStoreMock.EXPECT().GetInterfaceByIP(testConn.FlowKey.DestinationAddress.String()).Return(nil, false)
 		} else {
-			ifStoreMock.EXPECT().GetInterfaceByIP(testConn.TupleOrig.SourceAddress.String()).Return(nil, false)
-			ifStoreMock.EXPECT().GetInterfaceByIP(testConn.TupleOrig.DestinationAddress.String()).Return(testIfConfigs[i], true)
+			ifStoreMock.EXPECT().GetInterfaceByIP(testConn.FlowKey.SourceAddress.String()).Return(nil, false)
+			ifStoreMock.EXPECT().GetInterfaceByIP(testConn.FlowKey.DestinationAddress.String()).Return(testIfConfigs[i], true)
 		}
 	}
 	// Execute connStore.Poll
-	connsLens, err := connStore.Poll()
+	connsLens, err := conntrackConnStore.Poll()
 	require.Nil(t, err, fmt.Sprintf("Failed to add connections to connection store: %v", err))
 	assert.Len(t, connsLens, 1, "length of connsLens is expected to be 1")
 	assert.Len(t, testConns, connsLens[0], "expected connections should be equal to number of testConns")
@@ -156,7 +136,7 @@ func TestConnectionStoreAndFlowRecords(t *testing.T) {
 			expConn.DestinationPodName = testIfConfigs[i].PodName
 			expConn.DestinationPodNamespace = testIfConfigs[i].PodNamespace
 		}
-		actualConn, found := connStore.GetConnByKey(*testConnKeys[i])
+		actualConn, found := conntrackConnStore.GetConnByKey(*testConnKeys[i])
 		assert.Equal(t, found, true, "testConn should be present in connection store")
 		assert.Equal(t, expConn, actualConn, "testConn and connection in connection store should be equal")
 	}
diff --git a/test/integration/agent/openflow_test.go b/test/integration/agent/openflow_test.go
index 3b2802a15..74ff7e16a 100644
--- a/test/integration/agent/openflow_test.go
+++ b/test/integration/agent/openflow_test.go
@@ -110,7 +110,7 @@ func TestConnectivityFlows(t *testing.T) {
 		antrearuntime.WindowsOS = runtime.GOOS
 	}
 
-	c = ofClient.NewClient(br, bridgeMgmtAddr, ovsconfig.OVSDatapathNetdev, true, false, true)
+	c = ofClient.NewClient(br, bridgeMgmtAddr, ovsconfig.OVSDatapathNetdev, true, false, true, false)
 	err := ofTestUtils.PrepareOVSBridge(br)
 	require.Nil(t, err, fmt.Sprintf("Failed to prepare OVS bridge: %v", err))
 	defer func() {
@@ -137,7 +137,7 @@ func TestConnectivityFlows(t *testing.T) {
 }
 
 func TestReplayFlowsConnectivityFlows(t *testing.T) {
-	c = ofClient.NewClient(br, bridgeMgmtAddr, ovsconfig.OVSDatapathNetdev, true, false, false)
+	c = ofClient.NewClient(br, bridgeMgmtAddr, ovsconfig.OVSDatapathNetdev, true, false, false, false)
 	err := ofTestUtils.PrepareOVSBridge(br)
 	require.Nil(t, err, fmt.Sprintf("Failed to prepare OVS bridge: %v", err))
 
@@ -164,7 +164,7 @@ func TestReplayFlowsConnectivityFlows(t *testing.T) {
 }
 
 func TestReplayFlowsNetworkPolicyFlows(t *testing.T) {
-	c = ofClient.NewClient(br, bridgeMgmtAddr, ovsconfig.OVSDatapathNetdev, true, false, false)
+	c = ofClient.NewClient(br, bridgeMgmtAddr, ovsconfig.OVSDatapathNetdev, true, false, false, false)
 	err := ofTestUtils.PrepareOVSBridge(br)
 	require.Nil(t, err, fmt.Sprintf("Failed to prepare OVS bridge: %v", err))
 
@@ -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)
 		}
@@ -343,7 +343,7 @@ func TestNetworkPolicyFlows(t *testing.T) {
 	// Initialize ovs metrics (Prometheus) to test them
 	metrics.InitializeOVSMetrics()
 
-	c = ofClient.NewClient(br, bridgeMgmtAddr, ovsconfig.OVSDatapathNetdev, true, false, false)
+	c = ofClient.NewClient(br, bridgeMgmtAddr, ovsconfig.OVSDatapathNetdev, true, false, false, false)
 	err := ofTestUtils.PrepareOVSBridge(br)
 	require.Nil(t, err, fmt.Sprintf("Failed to prepare OVS bridge %s", br))
 
@@ -453,7 +453,7 @@ func TestIPv6ConnectivityFlows(t *testing.T) {
 	// Initialize ovs metrics (Prometheus) to test them
 	metrics.InitializeOVSMetrics()
 
-	c = ofClient.NewClient(br, bridgeMgmtAddr, ovsconfig.OVSDatapathNetdev, true, false, true)
+	c = ofClient.NewClient(br, bridgeMgmtAddr, ovsconfig.OVSDatapathNetdev, true, false, true, false)
 	err := ofTestUtils.PrepareOVSBridge(br)
 	require.Nil(t, err, fmt.Sprintf("Failed to prepare OVS bridge: %v", err))
 
@@ -485,7 +485,7 @@ type svcConfig struct {
 }
 
 func TestProxyServiceFlows(t *testing.T) {
-	c = ofClient.NewClient(br, bridgeMgmtAddr, ovsconfig.OVSDatapathNetdev, true, false, false)
+	c = ofClient.NewClient(br, bridgeMgmtAddr, ovsconfig.OVSDatapathNetdev, true, false, false, false)
 	err := ofTestUtils.PrepareOVSBridge(br)
 	require.Nil(t, err, fmt.Sprintf("Failed to prepare OVS bridge %s", br))
 
@@ -1365,7 +1365,7 @@ func prepareSNATFlows(snatIP net.IP, mark, podOFPort, podOFPortRemote uint32, vM
 }
 
 func TestSNATFlows(t *testing.T) {
-	c = ofClient.NewClient(br, bridgeMgmtAddr, ovsconfig.OVSDatapathNetdev, false, false, true)
+	c = ofClient.NewClient(br, bridgeMgmtAddr, ovsconfig.OVSDatapathNetdev, false, false, true, false)
 	err := ofTestUtils.PrepareOVSBridge(br)
 	require.Nil(t, err, fmt.Sprintf("Failed to prepare OVS bridge %s", br))