Repository: antrea-io/antrea
Pull request #2357 (2021-07-07 14:46:55)

[CI] Power on VM after reverting to a snapshot
Signed-off-by: Zhecheng Li <lzhecheng@vmware.com>

lzhecheng: /test-windows-conformance

Patch:
diff --git a/build/images/base/Dockerfile b/build/images/base/Dockerfile
index d9f680da0..e2b25ecfc 100644
--- a/build/images/base/Dockerfile
+++ b/build/images/base/Dockerfile
@@ -20,7 +20,7 @@ RUN set -eux; \
     esac; \
     mkdir -p /opt/cni/bin; \
     wget -q -O - https://github.com/containernetworking/plugins/releases/download/$CNI_BINARIES_VERSION/cni-plugins-linux-${pluginsArch}-$CNI_BINARIES_VERSION.tgz | tar xz -C /opt/cni/bin $CNI_PLUGINS; \
-    wget -q -O - https://downloads.antrea.io/whereabouts/v0.4/whereabouts-linux-${pluginsArch}.tgz | tar xz -C /opt/cni/bin/ whereabouts-linux-${pluginsArch}/whereabouts --strip-components=1
+    wget -q -O - https://downloads.antrea.io/whereabouts/v0.4/whereabouts-linux-${pluginsArch}.tgz | tar xz -C /opt/cni/bin/ whereabouts-linux-${pluginsArch}/whereabouts --strip-components=1 --no-same-owner
 
 FROM antrea/openvswitch:${OVS_VERSION}
 
diff --git a/build/yamls/antrea-aks.yml b/build/yamls/antrea-aks.yml
index 5c8c59ef4..be50c48bb 100644
--- a/build/yamls/antrea-aks.yml
+++ b/build/yamls/antrea-aks.yml
@@ -1345,6 +1345,10 @@ spec:
     - jsonPath: .metadata.creationTimestamp
       name: Age
       type: date
+    - description: The Owner Node of egress IP
+      jsonPath: .status.egressNode
+      name: Node
+      type: string
     name: v1alpha2
     schema:
       openAPIV3Schema:
@@ -1415,11 +1419,18 @@ spec:
             required:
             - appliedTo
             type: object
+          status:
+            properties:
+              egressNode:
+                type: string
+            type: object
         required:
         - spec
         type: object
     served: true
     storage: true
+    subresources:
+      status: {}
 ---
 apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
@@ -3282,6 +3293,20 @@ rules:
   - get
   - watch
   - list
+- apiGroups:
+  - crd.antrea.io
+  resources:
+  - egresses/status
+  verbs:
+  - update
+- apiGroups:
+  - crd.antrea.io
+  resources:
+  - externalippools
+  verbs:
+  - get
+  - watch
+  - list
 ---
 apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRole
@@ -3497,6 +3522,7 @@ rules:
   - watch
   - list
   - update
+  - patch
 - apiGroups:
   - crd.antrea.io
   resources:
@@ -3663,7 +3689,7 @@ data:
     #  FlowExporter: false
 
     # Enable collecting and exposing NetworkPolicy statistics.
-    #  NetworkPolicyStats: false
+    #  NetworkPolicyStats: true
 
     # Enable controlling SNAT IPs of Pod egress traffic.
     #  Egress: false
@@ -3812,7 +3838,7 @@ data:
     #  AntreaPolicy: true
 
     # Enable collecting and exposing NetworkPolicy statistics.
-    #  NetworkPolicyStats: false
+    #  NetworkPolicyStats: true
 
     # Enable controlling SNAT IPs of Pod egress traffic.
     #  Egress: false
@@ -3861,7 +3887,7 @@ metadata:
   annotations: {}
   labels:
     app: antrea
-  name: antrea-config-t9hc8tf75d
+  name: antrea-config-gb8bgg4d2m
   namespace: kube-system
 ---
 apiVersion: v1
@@ -3932,7 +3958,7 @@ spec:
             fieldRef:
               fieldPath: spec.serviceAccountName
         - name: ANTREA_CONFIG_MAP_NAME
-          value: antrea-config-t9hc8tf75d
+          value: antrea-config-gb8bgg4d2m
         image: projects.registry.vmware.com/antrea/antrea-ubuntu:latest
         imagePullPolicy: IfNotPresent
         livenessProbe:
@@ -3983,7 +4009,7 @@ spec:
         key: node-role.kubernetes.io/master
       volumes:
       - configMap:
-          name: antrea-config-t9hc8tf75d
+          name: antrea-config-gb8bgg4d2m
         name: antrea-config
       - name: antrea-controller-tls
         secret:
@@ -4279,7 +4305,7 @@ spec:
         operator: Exists
       volumes:
       - configMap:
-          name: antrea-config-t9hc8tf75d
+          name: antrea-config-gb8bgg4d2m
         name: antrea-config
       - hostPath:
           path: /etc/cni/net.d
diff --git a/build/yamls/antrea-eks.yml b/build/yamls/antrea-eks.yml
index 6b0a25e65..488b73a57 100644
--- a/build/yamls/antrea-eks.yml
+++ b/build/yamls/antrea-eks.yml
@@ -1345,6 +1345,10 @@ spec:
     - jsonPath: .metadata.creationTimestamp
       name: Age
       type: date
+    - description: The Owner Node of egress IP
+      jsonPath: .status.egressNode
+      name: Node
+      type: string
     name: v1alpha2
     schema:
       openAPIV3Schema:
@@ -1415,11 +1419,18 @@ spec:
             required:
             - appliedTo
             type: object
+          status:
+            properties:
+              egressNode:
+                type: string
+            type: object
         required:
         - spec
         type: object
     served: true
     storage: true
+    subresources:
+      status: {}
 ---
 apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
@@ -3282,6 +3293,20 @@ rules:
   - get
   - watch
   - list
+- apiGroups:
+  - crd.antrea.io
+  resources:
+  - egresses/status
+  verbs:
+  - update
+- apiGroups:
+  - crd.antrea.io
+  resources:
+  - externalippools
+  verbs:
+  - get
+  - watch
+  - list
 ---
 apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRole
@@ -3497,6 +3522,7 @@ rules:
   - watch
   - list
   - update
+  - patch
 - apiGroups:
   - crd.antrea.io
   resources:
@@ -3663,7 +3689,7 @@ data:
     #  FlowExporter: false
 
     # Enable collecting and exposing NetworkPolicy statistics.
-    #  NetworkPolicyStats: false
+    #  NetworkPolicyStats: true
 
     # Enable controlling SNAT IPs of Pod egress traffic.
     #  Egress: false
@@ -3812,7 +3838,7 @@ data:
     #  AntreaPolicy: true
 
     # Enable collecting and exposing NetworkPolicy statistics.
-    #  NetworkPolicyStats: false
+    #  NetworkPolicyStats: true
 
     # Enable controlling SNAT IPs of Pod egress traffic.
     #  Egress: false
@@ -3861,7 +3887,7 @@ metadata:
   annotations: {}
   labels:
     app: antrea
-  name: antrea-config-t9hc8tf75d
+  name: antrea-config-gb8bgg4d2m
   namespace: kube-system
 ---
 apiVersion: v1
@@ -3932,7 +3958,7 @@ spec:
             fieldRef:
               fieldPath: spec.serviceAccountName
         - name: ANTREA_CONFIG_MAP_NAME
-          value: antrea-config-t9hc8tf75d
+          value: antrea-config-gb8bgg4d2m
         image: projects.registry.vmware.com/antrea/antrea-ubuntu:latest
         imagePullPolicy: IfNotPresent
         livenessProbe:
@@ -3983,7 +4009,7 @@ spec:
         key: node-role.kubernetes.io/master
       volumes:
       - configMap:
-          name: antrea-config-t9hc8tf75d
+          name: antrea-config-gb8bgg4d2m
         name: antrea-config
       - name: antrea-controller-tls
         secret:
@@ -4281,7 +4307,7 @@ spec:
         operator: Exists
       volumes:
       - configMap:
-          name: antrea-config-t9hc8tf75d
+          name: antrea-config-gb8bgg4d2m
         name: antrea-config
       - hostPath:
           path: /etc/cni/net.d
diff --git a/build/yamls/antrea-gke.yml b/build/yamls/antrea-gke.yml
index 21e82a208..2bc1f5b11 100644
--- a/build/yamls/antrea-gke.yml
+++ b/build/yamls/antrea-gke.yml
@@ -1345,6 +1345,10 @@ spec:
     - jsonPath: .metadata.creationTimestamp
       name: Age
       type: date
+    - description: The Owner Node of egress IP
+      jsonPath: .status.egressNode
+      name: Node
+      type: string
     name: v1alpha2
     schema:
       openAPIV3Schema:
@@ -1415,11 +1419,18 @@ spec:
             required:
             - appliedTo
             type: object
+          status:
+            properties:
+              egressNode:
+                type: string
+            type: object
         required:
         - spec
         type: object
     served: true
     storage: true
+    subresources:
+      status: {}
 ---
 apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
@@ -3282,6 +3293,20 @@ rules:
   - get
   - watch
   - list
+- apiGroups:
+  - crd.antrea.io
+  resources:
+  - egresses/status
+  verbs:
+  - update
+- apiGroups:
+  - crd.antrea.io
+  resources:
+  - externalippools
+  verbs:
+  - get
+  - watch
+  - list
 ---
 apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRole
@@ -3497,6 +3522,7 @@ rules:
   - watch
   - list
   - update
+  - patch
 - apiGroups:
   - crd.antrea.io
   resources:
@@ -3663,7 +3689,7 @@ data:
     #  FlowExporter: false
 
     # Enable collecting and exposing NetworkPolicy statistics.
-    #  NetworkPolicyStats: false
+    #  NetworkPolicyStats: true
 
     # Enable controlling SNAT IPs of Pod egress traffic.
     #  Egress: false
@@ -3812,7 +3838,7 @@ data:
     #  AntreaPolicy: true
 
     # Enable collecting and exposing NetworkPolicy statistics.
-    #  NetworkPolicyStats: false
+    #  NetworkPolicyStats: true
 
     # Enable controlling SNAT IPs of Pod egress traffic.
     #  Egress: false
@@ -3861,7 +3887,7 @@ metadata:
   annotations: {}
   labels:
     app: antrea
-  name: antrea-config-9g829tktd6
+  name: antrea-config-4gm249hc95
   namespace: kube-system
 ---
 apiVersion: v1
@@ -3932,7 +3958,7 @@ spec:
             fieldRef:
               fieldPath: spec.serviceAccountName
         - name: ANTREA_CONFIG_MAP_NAME
-          value: antrea-config-9g829tktd6
+          value: antrea-config-4gm249hc95
         image: projects.registry.vmware.com/antrea/antrea-ubuntu:latest
         imagePullPolicy: IfNotPresent
         livenessProbe:
@@ -3983,7 +4009,7 @@ spec:
         key: node-role.kubernetes.io/master
       volumes:
       - configMap:
-          name: antrea-config-9g829tktd6
+          name: antrea-config-4gm249hc95
         name: antrea-config
       - name: antrea-controller-tls
         secret:
@@ -4282,7 +4308,7 @@ spec:
           path: /home/kubernetes/bin
         name: host-cni-bin
       - configMap:
-          name: antrea-config-9g829tktd6
+          name: antrea-config-4gm249hc95
         name: antrea-config
       - hostPath:
           path: /etc/cni/net.d
diff --git a/build/yamls/antrea-ipsec.yml b/build/yamls/antrea-ipsec.yml
index c8958d062..bc78304cf 100644
--- a/build/yamls/antrea-ipsec.yml
+++ b/build/yamls/antrea-ipsec.yml
@@ -1345,6 +1345,10 @@ spec:
     - jsonPath: .metadata.creationTimestamp
       name: Age
       type: date
+    - description: The Owner Node of egress IP
+      jsonPath: .status.egressNode
+      name: Node
+      type: string
     name: v1alpha2
     schema:
       openAPIV3Schema:
@@ -1415,11 +1419,18 @@ spec:
             required:
             - appliedTo
             type: object
+          status:
+            properties:
+              egressNode:
+                type: string
+            type: object
         required:
         - spec
         type: object
     served: true
     storage: true
+    subresources:
+      status: {}
 ---
 apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
@@ -3282,6 +3293,20 @@ rules:
   - get
   - watch
   - list
+- apiGroups:
+  - crd.antrea.io
+  resources:
+  - egresses/status
+  verbs:
+  - update
+- apiGroups:
+  - crd.antrea.io
+  resources:
+  - externalippools
+  verbs:
+  - get
+  - watch
+  - list
 ---
 apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRole
@@ -3497,6 +3522,7 @@ rules:
   - watch
   - list
   - update
+  - patch
 - apiGroups:
   - crd.antrea.io
   resources:
@@ -3663,7 +3689,7 @@ data:
     #  FlowExporter: false
 
     # Enable collecting and exposing NetworkPolicy statistics.
-    #  NetworkPolicyStats: false
+    #  NetworkPolicyStats: true
 
     # Enable controlling SNAT IPs of Pod egress traffic.
     #  Egress: false
@@ -3817,7 +3843,7 @@ data:
     #  AntreaPolicy: true
 
     # Enable collecting and exposing NetworkPolicy statistics.
-    #  NetworkPolicyStats: false
+    #  NetworkPolicyStats: true
 
     # Enable controlling SNAT IPs of Pod egress traffic.
     #  Egress: false
@@ -3866,7 +3892,7 @@ metadata:
   annotations: {}
   labels:
     app: antrea
-  name: antrea-config-h5kbhh859d
+  name: antrea-config-4949869kb7
   namespace: kube-system
 ---
 apiVersion: v1
@@ -3946,7 +3972,7 @@ spec:
             fieldRef:
               fieldPath: spec.serviceAccountName
         - name: ANTREA_CONFIG_MAP_NAME
-          value: antrea-config-h5kbhh859d
+          value: antrea-config-4949869kb7
         image: projects.registry.vmware.com/antrea/antrea-ubuntu:latest
         imagePullPolicy: IfNotPresent
         livenessProbe:
@@ -3997,7 +4023,7 @@ spec:
         key: node-role.kubernetes.io/master
       volumes:
       - configMap:
-          name: antrea-config-h5kbhh859d
+          name: antrea-config-4949869kb7
         name: antrea-config
       - name: antrea-controller-tls
         secret:
@@ -4328,7 +4354,7 @@ spec:
         operator: Exists
       volumes:
       - configMap:
-          name: antrea-config-h5kbhh859d
+          name: antrea-config-4949869kb7
         name: antrea-config
       - hostPath:
           path: /etc/cni/net.d
diff --git a/build/yamls/antrea.yml b/build/yamls/antrea.yml
index b75e88e6f..6e3c9f03e 100644
--- a/build/yamls/antrea.yml
+++ b/build/yamls/antrea.yml
@@ -1345,6 +1345,10 @@ spec:
     - jsonPath: .metadata.creationTimestamp
       name: Age
       type: date
+    - description: The Owner Node of egress IP
+      jsonPath: .status.egressNode
+      name: Node
+      type: string
     name: v1alpha2
     schema:
       openAPIV3Schema:
@@ -1415,11 +1419,18 @@ spec:
             required:
             - appliedTo
             type: object
+          status:
+            properties:
+              egressNode:
+                type: string
+            type: object
         required:
         - spec
         type: object
     served: true
     storage: true
+    subresources:
+      status: {}
 ---
 apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
@@ -3282,6 +3293,20 @@ rules:
   - get
   - watch
   - list
+- apiGroups:
+  - crd.antrea.io
+  resources:
+  - egresses/status
+  verbs:
+  - update
+- apiGroups:
+  - crd.antrea.io
+  resources:
+  - externalippools
+  verbs:
+  - get
+  - watch
+  - list
 ---
 apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRole
@@ -3497,6 +3522,7 @@ rules:
   - watch
   - list
   - update
+  - patch
 - apiGroups:
   - crd.antrea.io
   resources:
@@ -3663,7 +3689,7 @@ data:
     #  FlowExporter: false
 
     # Enable collecting and exposing NetworkPolicy statistics.
-    #  NetworkPolicyStats: false
+    #  NetworkPolicyStats: true
 
     # Enable controlling SNAT IPs of Pod egress traffic.
     #  Egress: false
@@ -3817,7 +3843,7 @@ data:
     #  AntreaPolicy: true
 
     # Enable collecting and exposing NetworkPolicy statistics.
-    #  NetworkPolicyStats: false
+    #  NetworkPolicyStats: true
 
     # Enable controlling SNAT IPs of Pod egress traffic.
     #  Egress: false
@@ -3866,7 +3892,7 @@ metadata:
   annotations: {}
   labels:
     app: antrea
-  name: antrea-config-cbfh568k9m
+  name: antrea-config-dcfb6k2hkm
   namespace: kube-system
 ---
 apiVersion: v1
@@ -3937,7 +3963,7 @@ spec:
             fieldRef:
               fieldPath: spec.serviceAccountName
         - name: ANTREA_CONFIG_MAP_NAME
-          value: antrea-config-cbfh568k9m
+          value: antrea-config-dcfb6k2hkm
         image: projects.registry.vmware.com/antrea/antrea-ubuntu:latest
         imagePullPolicy: IfNotPresent
         livenessProbe:
@@ -3988,7 +4014,7 @@ spec:
         key: node-role.kubernetes.io/master
       volumes:
       - configMap:
-          name: antrea-config-cbfh568k9m
+          name: antrea-config-dcfb6k2hkm
         name: antrea-config
       - name: antrea-controller-tls
         secret:
@@ -4284,7 +4310,7 @@ spec:
         operator: Exists
       volumes:
       - configMap:
-          name: antrea-config-cbfh568k9m
+          name: antrea-config-dcfb6k2hkm
         name: antrea-config
       - hostPath:
           path: /etc/cni/net.d
diff --git a/build/yamls/base/agent-rbac.yml b/build/yamls/base/agent-rbac.yml
index 61db6ab0f..df9758149 100644
--- a/build/yamls/base/agent-rbac.yml
+++ b/build/yamls/base/agent-rbac.yml
@@ -149,6 +149,20 @@ rules:
       - get
       - watch
       - list
+  - apiGroups:
+      - crd.antrea.io
+    resources:
+      - egresses/status
+    verbs:
+      - update
+  - apiGroups:
+      - crd.antrea.io
+    resources:
+      - externalippools
+    verbs:
+      - get
+      - watch
+      - list
 ---
 kind: ClusterRoleBinding
 apiVersion: rbac.authorization.k8s.io/v1
diff --git a/build/yamls/base/conf/antrea-agent.conf b/build/yamls/base/conf/antrea-agent.conf
index 01591e983..ae2213ab2 100644
--- a/build/yamls/base/conf/antrea-agent.conf
+++ b/build/yamls/base/conf/antrea-agent.conf
@@ -26,7 +26,7 @@ featureGates:
 #  FlowExporter: false
 
 # Enable collecting and exposing NetworkPolicy statistics.
-#  NetworkPolicyStats: false
+#  NetworkPolicyStats: true
 
 # Enable controlling SNAT IPs of Pod egress traffic.
 #  Egress: false
diff --git a/build/yamls/base/conf/antrea-controller.conf b/build/yamls/base/conf/antrea-controller.conf
index f67763102..18c2910db 100644
--- a/build/yamls/base/conf/antrea-controller.conf
+++ b/build/yamls/base/conf/antrea-controller.conf
@@ -9,7 +9,7 @@ featureGates:
 #  AntreaPolicy: true
 
 # Enable collecting and exposing NetworkPolicy statistics.
-#  NetworkPolicyStats: false
+#  NetworkPolicyStats: true
 
 # Enable controlling SNAT IPs of Pod egress traffic.
 #  Egress: false
diff --git a/build/yamls/base/controller-rbac.yml b/build/yamls/base/controller-rbac.yml
index 4855d2e3f..af7b90e76 100644
--- a/build/yamls/base/controller-rbac.yml
+++ b/build/yamls/base/controller-rbac.yml
@@ -207,6 +207,7 @@ rules:
       - watch
       - list
       - update
+      - patch
   - apiGroups:
       - crd.antrea.io
     resources:
diff --git a/build/yamls/base/crds.yml b/build/yamls/base/crds.yml
index 38a1867b2..7eb3ab70b 100644
--- a/build/yamls/base/crds.yml
+++ b/build/yamls/base/crds.yml
@@ -81,6 +81,11 @@ spec:
                 - format: ipv6
               externalIPPool:
                 type: string
+          status:
+            type: object
+            properties:
+              egressNode:
+                type: string
     additionalPrinterColumns:
     - description: Specifies the SNAT IP address for the selected workloads.
       jsonPath: .spec.egressIP
@@ -89,6 +94,12 @@ spec:
     - jsonPath: .metadata.creationTimestamp
       name: Age
       type: date
+    - description: The Owner Node of egress IP
+      jsonPath: .status.egressNode
+      name: Node
+      type: string
+    subresources:
+      status: {}
   scope: Cluster
   names:
     plural: egresses
diff --git a/cmd/antrea-agent/agent.go b/cmd/antrea-agent/agent.go
index 904185587..8789033da 100644
--- a/cmd/antrea-agent/agent.go
+++ b/cmd/antrea-agent/agent.go
@@ -77,6 +77,8 @@ func run(o *Options) error {
 	crdInformerFactory := crdinformers.NewSharedInformerFactory(crdClient, informerDefaultResync)
 	traceflowInformer := crdInformerFactory.Crd().V1alpha1().Traceflows()
 	egressInformer := crdInformerFactory.Crd().V1alpha2().Egresses()
+	nodeInformer := informerFactory.Core().V1().Nodes()
+	externalIPPoolInformer := crdInformerFactory.Crd().V1alpha2().ExternalIPPools()
 
 	// Create Antrea Clientset for the given config.
 	antreaClientProvider := agent.NewAntreaClientProvider(o.config.AntreaClientConnection, k8sClient)
@@ -223,7 +225,13 @@ func run(o *Options) error {
 
 	var egressController *egress.EgressController
 	if features.DefaultFeatureGate.Enabled(features.Egress) {
-		egressController = egress.NewEgressController(ofClient, egressInformer, antreaClientProvider, ifaceStore, routeClient, nodeConfig.Name)
+		egressController, err = egress.NewEgressController(
+			ofClient, antreaClientProvider, crdClient, ifaceStore, routeClient, nodeConfig.Name, nodeConfig.NodeIPAddr.IP,
+			o.config.ClusterMembershipPort, egressInformer, nodeInformer, externalIPPoolInformer,
+		)
+		if err != nil {
+			return fmt.Errorf("error creating new Egress controller: %v", err)
+		}
 	}
 
 	isChaining := false
diff --git a/cmd/antrea-agent/config.go b/cmd/antrea-agent/config.go
index 9e3f13725..e9b36915d 100644
--- a/cmd/antrea-agent/config.go
+++ b/cmd/antrea-agent/config.go
@@ -98,6 +98,11 @@ type AgentConfig struct {
 	// APIPort is the port for the antrea-agent APIServer to serve on.
 	// Defaults to 10350.
 	APIPort int `yaml:"apiPort,omitempty"`
+
+	// ClusterMembershipPort is the server port used by the antrea-agent to run a gossip-based cluster membership protocol. Currently it's used only when the Egress feature is enabled.
+	// Defaults to 10351.
+	ClusterMembershipPort int `yaml:"clusterPort,omitempty"`
+
 	// Enable metrics exposure via Prometheus. Initializes Prometheus metrics listener
 	// Defaults to true.
 	EnablePrometheusMetrics bool `yaml:"enablePrometheusMetrics,omitempty"`
diff --git a/cmd/antrea-agent/options.go b/cmd/antrea-agent/options.go
index 7e4aa315d..04037aa03 100644
--- a/cmd/antrea-agent/options.go
+++ b/cmd/antrea-agent/options.go
@@ -193,6 +193,10 @@ func (o *Options) setDefaults() {
 		o.config.APIPort = apis.AntreaAgentAPIPort
 	}
 
+	if o.config.ClusterMembershipPort == 0 {
+		o.config.ClusterMembershipPort = apis.AntreaAgentClusterMembershipPort
+	}
+
 	if features.DefaultFeatureGate.Enabled(features.FlowExporter) {
 		if o.config.FlowCollectorAddr == "" {
 			o.config.FlowCollectorAddr = defaultFlowCollectorAddress
diff --git a/docs/feature-gates.md b/docs/feature-gates.md
index 48a277f43..4faa0baca 100644
--- a/docs/feature-gates.md
+++ b/docs/feature-gates.md
@@ -40,7 +40,7 @@ example, to enable `AntreaProxy` on Linux, edit the Agent configuration in the
 | `AntreaPolicy`          | Agent + Controller | `true`  | Beta  | v0.8          | v1.0         | N/A        | No                 | Agent side config required from v0.9.0+. |
 | `Traceflow`             | Agent + Controller | `true`  | Beta  | v0.8          | v0.11        | N/A        | Yes                |       |
 | `FlowExporter`          | Agent              | `false` | Alpha | v0.9          | N/A          | N/A        | Yes                |       |
-| `NetworkPolicyStats`    | Agent + Controller | `false` | Alpha | v0.10         | N/A          | N/A        | No                 |       |
+| `NetworkPolicyStats`    | Agent + Controller | `false` | Alpha | v0.10         | v1.2         | N/A        | No                 |       |
 | `NodePortLocal`         | Agent              | `false` | Alpha | v0.13         | N/A          | N/A        | Yes                |       |
 | `Egress`                | Agent + Controller | `false` | Alpha | v1.0          | N/A          | N/A        | Yes                |       |
 
diff --git a/go.mod b/go.mod
index 6c6b1d438..cab28d6dd 100644
--- a/go.mod
+++ b/go.mod
@@ -23,9 +23,11 @@ require (
 	github.com/elazarl/goproxy v0.0.0-20190911111923-ecfe977594f1 // indirect
 	github.com/go-openapi/spec v0.19.5
 	github.com/gogo/protobuf v1.3.2
-	github.com/golang/mock v1.5.0
+	github.com/golang/groupcache v0.0.0-20200121045136-8c9f03a8e57e
+	github.com/golang/mock v1.6.0
 	github.com/golang/protobuf v1.5.0
 	github.com/google/uuid v1.1.2
+	github.com/hashicorp/memberlist v0.2.4
 	github.com/k8snetworkplumbingwg/sriov-cni v2.1.0+incompatible
 	github.com/kevinburke/ssh_config v0.0.0-20190725054713-01f96b0aa0cd
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822
@@ -45,10 +47,10 @@ require (
 	github.com/vmware/go-ipfix v0.5.3
 	golang.org/x/crypto v0.0.0-20210220033148-5ea612d1eb83
 	golang.org/x/exp v0.0.0-20200224162631-6cc2880d07d6
-	golang.org/x/mod v0.4.0
-	golang.org/x/net v0.0.0-20210224082022-3d97a244fca7
-	golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9
-	golang.org/x/sys v0.0.0-20210225134936-a50acf3fe073
+	golang.org/x/mod v0.4.2
+	golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4
+	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c
+	golang.org/x/sys v0.0.0-20210510120138-977fb7262007
 	golang.org/x/time v0.0.0-20210220033141-f8bda1e9f3ba
 	google.golang.org/grpc v1.27.1
 	gopkg.in/natefinch/lumberjack.v2 v2.0.0
diff --git a/go.sum b/go.sum
index b9472712d..a77bcb7b0 100644
--- a/go.sum
+++ b/go.sum
@@ -69,6 +69,7 @@ github.com/antoninbas/go-powershell v0.1.0 h1:LKwuZJt3loyr++Y2Qc+FZoUt8TwbrTWB3H
 github.com/antoninbas/go-powershell v0.1.0/go.mod h1:01pgKhz1CJxGnCWqXVDgvmp/QmHgWgEdxdYP+1azopE=
 github.com/armon/circbuf v0.0.0-20150827004946-bbbad097214e/go.mod h1:3U/XgcO3hCbHZ8TKRvWD2dDTCfh9M9ya+I9JpbB7O8o=
 github.com/armon/consul-api v0.0.0-20180202201655-eb2c6b5be1b6/go.mod h1:grANhF5doyWs3UAsr3K4I6qtAmlQcZDesFNEHPZAzj8=
+github.com/armon/go-metrics v0.0.0-20180917152333-f0300d1749da h1:8GUt8eRujhVEGZFFEjBj46YV4rDjvGrNxb0KMWYkL2I=
 github.com/armon/go-metrics v0.0.0-20180917152333-f0300d1749da/go.mod h1:Q73ZrmVTwzkszR9V5SSuryQ31EELlFMUz1kKyl939pY=
 github.com/armon/go-radix v0.0.0-20180808171621-7fddfc383310/go.mod h1:ufUuZ+zHj4x4TnLV4JWEpy2hxWSpsRywHrMgIH9cCH8=
 github.com/asaskevich/govalidator v0.0.0-20180720115003-f9ffefc3facf/go.mod h1:lB+ZfQJz7igIIfQNfa7Ml4HSf2uFQQRzpGGRXenZAgY=
@@ -280,8 +281,8 @@ github.com/golang/mock v1.3.1/go.mod h1:sBzyDLLjw3U8JLTeZvSv8jJB+tU5PVekmnlKIyFU
 github.com/golang/mock v1.4.0/go.mod h1:UOMv5ysSaYNkG+OFQykRIcU/QvvxJf3p21QfJ2Bt3cw=
 github.com/golang/mock v1.4.1/go.mod h1:UOMv5ysSaYNkG+OFQykRIcU/QvvxJf3p21QfJ2Bt3cw=
 github.com/golang/mock v1.4.3/go.mod h1:UOMv5ysSaYNkG+OFQykRIcU/QvvxJf3p21QfJ2Bt3cw=
-github.com/golang/mock v1.5.0 h1:jlYHihg//f7RRwuPfptm04yp4s7O6Kw8EZiVYIGcH0g=
-github.com/golang/mock v1.5.0/go.mod h1:CWnOUgYIOo4TcNZ0wHX3YZCqsaM1I1Jvs6v3mP3KVu8=
+github.com/golang/mock v1.6.0 h1:ErTB+efbowRARo13NNdxyJji2egdxLGQhRaY+DUumQc=
+github.com/golang/mock v1.6.0/go.mod h1:p6yTPP+5HYm5mzsMV8JkE6ZKdX+/wYM6Hr+LicevLPs=
 github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
 github.com/golang/protobuf v1.3.1/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
 github.com/golang/protobuf v1.3.2/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
@@ -346,12 +347,17 @@ github.com/grpc-ecosystem/grpc-gateway v1.9.5 h1:UImYN5qQ8tuGpGE16ZmjvcTtTw24zw1
 github.com/grpc-ecosystem/grpc-gateway v1.9.5/go.mod h1:vNeuVxBJEsws4ogUvrchl83t/GYV9WGTSLVdBhOQFDY=
 github.com/hashicorp/consul/api v1.1.0/go.mod h1:VmuI/Lkw1nC05EYQWNKwWGbkg+FbDBtguAZLlVdkD9Q=
 github.com/hashicorp/consul/sdk v0.1.1/go.mod h1:VKf9jXwCTEY1QZP2MOLRhb5i/I/ssyNV1vwHyQBF0x8=
+github.com/hashicorp/errwrap v1.0.0 h1:hLrqtEDnRye3+sgx6z4qVLNuviH3MR5aQ0ykNJa/UYA=
 github.com/hashicorp/errwrap v1.0.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4=
 github.com/hashicorp/go-cleanhttp v0.5.1/go.mod h1:JpRdi6/HCYpAwUzNwuwqhbovhLtngrth3wmdIIUrZ80=
+github.com/hashicorp/go-immutable-radix v1.0.0 h1:AKDB1HM5PWEA7i4nhcpwOrO2byshxBjXVn/J/3+z5/0=
 github.com/hashicorp/go-immutable-radix v1.0.0/go.mod h1:0y9vanUI8NX6FsYoO3zeMjhV/C5i9g4Q3DwcSNZ4P60=
+github.com/hashicorp/go-msgpack v0.5.3 h1:zKjpN5BK/P5lMYrLmBHdBULWbJ0XpYR+7NGzqkZzoD4=
 github.com/hashicorp/go-msgpack v0.5.3/go.mod h1:ahLV/dePpqEmjfWmKiqvPkv/twdG7iPBM1vqhUKIvfM=
+github.com/hashicorp/go-multierror v1.0.0 h1:iVjPR7a6H0tWELX5NxNe7bYopibicUzc7uPribsnS6o=
 github.com/hashicorp/go-multierror v1.0.0/go.mod h1:dHtQlpGsu+cZNNAkkCN/P3hoUDHhCYQXV3UM06sGGrk=
 github.com/hashicorp/go-rootcerts v1.0.0/go.mod h1:K6zTfqpRlCUIjkwsN4Z+hiSfzSTQa6eBIzfwKfwNnHU=
+github.com/hashicorp/go-sockaddr v1.0.0 h1:GeH6tui99pF4NJgfnhp+L6+FfobzVW3Ah46sLo0ICXs=
 github.com/hashicorp/go-sockaddr v1.0.0/go.mod h1:7Xibr9yA9JjQq1JpNB2Vw7kxv8xerXegt+ozgdvDeDU=
 github.com/hashicorp/go-syslog v1.0.0/go.mod h1:qPfqrKkXGihmCqbJM2mZgkZGvKG1dFdvsLplgctolz4=
 github.com/hashicorp/go-uuid v1.0.0/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro=
@@ -366,6 +372,8 @@ github.com/hashicorp/hcl v1.0.0/go.mod h1:E5yfLk+7swimpb2L/Alb/PJmXilQ/rhwaUYs4T
 github.com/hashicorp/logutils v1.0.0/go.mod h1:QIAnNjmIWmVIIkWDTG1z5v++HQmx9WQRO+LraFDTW64=
 github.com/hashicorp/mdns v1.0.0/go.mod h1:tL+uN++7HEJ6SQLQ2/p+z2pH24WQKWjBPkE0mNTz8vQ=
 github.com/hashicorp/memberlist v0.1.3/go.mod h1:ajVTdAv/9Im8oMAAj5G31PhhMCZJV2pPBoIllUwCN7I=
+github.com/hashicorp/memberlist v0.2.4 h1:OOhYzSvFnkFQXm1ysE8RjXTHsqSRDyP4emusC9K7DYg=
+github.com/hashicorp/memberlist v0.2.4/go.mod h1:MS2lj3INKhZjWNqd3N0m3J+Jxf3DAOnAH9VT3Sh9MUE=
 github.com/hashicorp/serf v0.8.2/go.mod h1:6hOLApaqBFA1NXqRQAsxw9QxuDEvNxSQRwA/JwenrHc=
 github.com/hpcloud/tail v1.0.0 h1:nfCOvKYfkgYP8hkirhJocXT2+zOD8yUNjXaWfTlyFKI=
 github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU=
@@ -444,6 +452,8 @@ github.com/mdlayher/netlink v1.0.0/go.mod h1:KxeJAFOFLG6AjpyDkQ/iIhxygIUKD+vcwqc
 github.com/mdlayher/netlink v1.1.0 h1:mpdLgm+brq10nI9zM1BpX1kpDbh3NLl3RSnVq6ZSkfg=
 github.com/mdlayher/netlink v1.1.0/go.mod h1:H4WCitaheIsdF9yOYu8CFmCgQthAPIWZmcKp9uZHgmY=
 github.com/miekg/dns v1.0.14/go.mod h1:W1PPwlIAgtquWBMBEV9nkV9Cazfe8ScdGz/Lj7v3Nrg=
+github.com/miekg/dns v1.1.26 h1:gPxPSwALAeHJSjarOs00QjVdV9QoBvc1D2ujQUr5BzU=
+github.com/miekg/dns v1.1.26/go.mod h1:bPDLeHnStXmXAq1m/Ch/hvfNHr14JKNPMBo3VZKjuso=
 github.com/mitchellh/cli v1.0.0/go.mod h1:hNIlj7HEI86fIcpObd7a0FcrxTWetlwJDGcceTlRvqc=
 github.com/mitchellh/go-homedir v1.0.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0=
 github.com/mitchellh/go-homedir v1.1.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0=
@@ -487,6 +497,7 @@ github.com/opencontainers/go-digest v0.0.0-20180430190053-c9281466c8b2/go.mod h1
 github.com/opencontainers/go-digest v1.0.0/go.mod h1:0JzlMkj0TRzQZfJkVvzbP0HBR3IKzErnv2BNG4W4MAM=
 github.com/opencontainers/runc v0.0.0-20190115041553-12f6a991201f/go.mod h1:qT5XzbpPznkRYVz/mWwUaVBUv2rmF59PVA73FjuZG0U=
 github.com/opencontainers/runtime-spec v1.0.2/go.mod h1:jwyrGlmzljRJv/Fgzds9SsS/C5hL+LL3ko9hs6T5lQ0=
+github.com/pascaldekloe/goe v0.0.0-20180627143212-57f6aae5913c h1:Lgl0gzECD8GnQ5QCWA8o6BtfL6mDH5rQgM4/fX3avOs=
 github.com/pascaldekloe/goe v0.0.0-20180627143212-57f6aae5913c/go.mod h1:lzWF7FIEvWOWxwDKqyGYQf6ZUaNfKdP144TG7ZOy1lc=
 github.com/pborman/uuid v1.2.0/go.mod h1:X/NO0urCmaxf9VXbdlT7C2Yzkj2IKimNn4k+gtPdI/k=
 github.com/pelletier/go-toml v1.2.0/go.mod h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/94hg7ilaic=
@@ -550,6 +561,7 @@ github.com/safchain/ethtool v0.0.0-20190326074333-42ed695e3de8 h1:2c1EFnZHIPCW8q
 github.com/safchain/ethtool v0.0.0-20190326074333-42ed695e3de8/go.mod h1:Z0q5wiBQGYcxhMZ6gUqHn6pYNLypFAvaL3UvgZLR0U4=
 github.com/satori/go.uuid v1.2.0 h1:0uYX9dsZ2yD7q2RtLRtPSdGDWzjeM3TbMJP9utgA0ww=
 github.com/satori/go.uuid v1.2.0/go.mod h1:dA0hQrYB0VpLJoorglMZABFdXlWrHn1NEOzdhQKdks0=
+github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529 h1:nn5Wsu0esKSJiIVhscUtVbo7ada43DJhG55ua/hjS5I=
 github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529/go.mod h1:DxrIzT+xaE7yg65j358z/aeFdxmN0P9QXhEzd20vsDc=
 github.com/sergi/go-diff v1.0.0/go.mod h1:0CfEIISq7TuYL3j771MWULgwwjU+GofnZX9QAmXWZgo=
 github.com/sergi/go-diff v1.1.0/go.mod h1:STckp+ISIX8hZLjrqAeVduY0gWCT9IjLuqbuNXdaHfM=
@@ -625,6 +637,7 @@ github.com/xlab/treeprint v0.0.0-20181112141820-a009c3971eca/go.mod h1:ce1O1j6Ut
 github.com/xordataexchange/crypt v0.0.3-0.20170626215501-b2862e3d0a77/go.mod h1:aYKd//L2LvnjZzWKhF00oedf4jCCReLcmhLdhm1A27Q=
 github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
+github.com/yuin/goldmark v1.3.5/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k=
 go.etcd.io/bbolt v1.3.2/go.mod h1:IbVyRI1SCnLcuJnV2u8VeU0CEYM7e686BmAb1XKL+uU=
 go.etcd.io/bbolt v1.3.3/go.mod h1:IbVyRI1SCnLcuJnV2u8VeU0CEYM7e686BmAb1XKL+uU=
 go.etcd.io/bbolt v1.3.5 h1:XAzx9gjCb0Rxj7EoqcClPD1d5ZBxZJk0jbuoPHenBt0=
@@ -657,6 +670,7 @@ golang.org/x/crypto v0.0.0-20190605123033-f99c8df09eb5/go.mod h1:yigFU9vqHzYiE8U
 golang.org/x/crypto v0.0.0-20190611184440-5c40567a22f8/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20190617133340-57b3e21c3d56/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20190820162420-60c769a6c586/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
+golang.org/x/crypto v0.0.0-20190923035154-9ee001bba392/go.mod h1:/lpIB1dKB+9EgE3H3cr1v9wB50oz8l4C4h62xy7jSTY=
 golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20200820211705-5c72a883971a/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
@@ -695,8 +709,8 @@ golang.org/x/mod v0.1.1-0.20191107180719-034126e5016b/go.mod h1:QqPTAvyqsEbceGzB
 golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.3.1-0.20200828183125-ce943fd02449/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
-golang.org/x/mod v0.4.0 h1:8pl+sMODzuvGJkmj2W4kZihvVb5mKm8pB/X44PIQHv8=
-golang.org/x/mod v0.4.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
+golang.org/x/mod v0.4.2 h1:Gz96sIWK3OalVv/I/qNygP42zyoKp3xptRVCWRFEBvo=
+golang.org/x/mod v0.4.2/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
@@ -720,6 +734,7 @@ golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLL
 golang.org/x/net v0.0.0-20190724013045-ca1201d0de80/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20190813141303-74dc4d7220e7/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20190827160401-ba9fcec4b297/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/net v0.0.0-20190923162816-aa69164e4478/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20191004110552-13f9640d40b9/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20191007182048-72f939374954/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20191209160850-c0dbc17a3553/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
@@ -734,8 +749,9 @@ golang.org/x/net v0.0.0-20200822124328-c89045814202/go.mod h1:/O7V0waA8r7cgGh81R
 golang.org/x/net v0.0.0-20200904194848-62affa334b73/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA=
 golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
 golang.org/x/net v0.0.0-20201110031124-69a78807bb2b/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
-golang.org/x/net v0.0.0-20210224082022-3d97a244fca7 h1:OgUuv8lsRpBibGNbSizVwKWlysjaNzmC9gYMhPVfqFM=
 golang.org/x/net v0.0.0-20210224082022-3d97a244fca7/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
+golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4 h1:4nGaVu0QrbjT/AK2PRLuQfQuh6DJve+pELhqTdAj3x0=
+golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
 golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
@@ -748,8 +764,9 @@ golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20190227155943-e225da77a7e6/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9 h1:SQFwaSi55rU7vdNs9Yr0Z324VNlrF+0wMqRXT4St8ck=
 golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20210220032951-036812b2e83c h1:5KslGYwFpkhGh+Q16bwMP3cOontH8FOep7tGV86Y7SQ=
+golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20180823144017-11551d06cbcc/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
@@ -773,6 +790,8 @@ golang.org/x/sys v0.0.0-20190624142023-c5567b49c5d0/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20190726091711-fc99dfbffb4e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190826190057-c7b8b68b1456/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190916202348-b4ddaad3f8a3/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20190922100055-0a153f010e69/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20190924154521-2837fb4f24fe/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20191001151750-bb3f8db39f24/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20191002063906-3421d5a6bb1c/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20191005200804-aed5e4c7ecf9/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
@@ -798,8 +817,10 @@ golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20201201145000-ef89a241ccb3/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210119212857-b64e53b001e4/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20210225134936-a50acf3fe073 h1:8qxJSnu+7dRq6upnbntrmriWByIakBuct5OM/MdQC1M=
 golang.org/x/sys v0.0.0-20210225134936-a50acf3fe073/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210510120138-977fb7262007 h1:gG67DSER+11cZvqIMb8S8bt0vZtiN6xWYARwirrOSfE=
+golang.org/x/sys v0.0.0-20210510120138-977fb7262007/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210220032956-6a3ed077a48d h1:SZxvLBoTP5yHO3Frd4z4vrF+DBX9vMVanchswa69toE=
@@ -837,6 +858,7 @@ golang.org/x/tools v0.0.0-20190621195816-6e04913cbbac/go.mod h1:/rFqwRUd4F7ZHNgw
 golang.org/x/tools v0.0.0-20190624222133-a101b041ded4/go.mod h1:/rFqwRUd4F7ZHNgwSSTFct+R/Kf4OFW1sUzUTQQTgfc=
 golang.org/x/tools v0.0.0-20190628153133-6cdbf07be9d0/go.mod h1:/rFqwRUd4F7ZHNgwSSTFct+R/Kf4OFW1sUzUTQQTgfc=
 golang.org/x/tools v0.0.0-20190816200558-6889da9d5479/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
+golang.org/x/tools v0.0.0-20190907020128-2ca718005c18/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20190911174233-4f2ddba30aff/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20191012152004-8de300cfc20a/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20191112195655-aa38f8e97acc/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
@@ -859,6 +881,7 @@ golang.org/x/tools v0.0.0-20200505023115-26f46d2f7ef8/go.mod h1:EkVYQZoAsY45+roY
 golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
 golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
 golang.org/x/tools v0.1.0/go.mod h1:xkSsbof2nBLbhDlRMhhhyNLN/zl3eTqcnHD5viDpcZ0=
+golang.org/x/tools v0.1.1/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
diff --git a/hack/update-codegen-dockerized.sh b/hack/update-codegen-dockerized.sh
index d091a5e3f..ac735d4f7 100755
--- a/hack/update-codegen-dockerized.sh
+++ b/hack/update-codegen-dockerized.sh
@@ -167,6 +167,7 @@ MOCKGEN_TARGETS=(
   "pkg/agent/proxy Proxier testing"
   "pkg/agent/querier AgentQuerier testing"
   "pkg/agent/route Interface testing"
+  "pkg/agent/controller/egress/ipassigner IPAssigner testing"
   "pkg/antctl AntctlClient ."
   "pkg/controller/networkpolicy EndpointQuerier testing"
   "pkg/controller/querier ControllerQuerier testing"
diff --git a/pkg/agent/controller/egress/egress_controller.go b/pkg/agent/controller/egress/egress_controller.go
index f75761517..ab7c7c9bd 100644
--- a/pkg/agent/controller/egress/egress_controller.go
+++ b/pkg/agent/controller/egress/egress_controller.go
@@ -29,18 +29,23 @@ import (
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/apimachinery/pkg/watch"
+	coreinformers "k8s.io/client-go/informers/core/v1"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/util/workqueue"
 	"k8s.io/klog/v2"
 
 	"antrea.io/antrea/pkg/agent"
+	"antrea.io/antrea/pkg/agent/controller/egress/ipassigner"
 	"antrea.io/antrea/pkg/agent/interfacestore"
+	"antrea.io/antrea/pkg/agent/memberlist"
 	"antrea.io/antrea/pkg/agent/openflow"
 	"antrea.io/antrea/pkg/agent/route"
 	cpv1b2 "antrea.io/antrea/pkg/apis/controlplane/v1beta2"
 	crdv1a2 "antrea.io/antrea/pkg/apis/crd/v1alpha2"
+	clientsetversioned "antrea.io/antrea/pkg/client/clientset/versioned"
 	crdinformers "antrea.io/antrea/pkg/client/informers/externalversions/crd/v1alpha2"
 	crdlisters "antrea.io/antrea/pkg/client/listers/crd/v1alpha2"
+	"antrea.io/antrea/pkg/controller/metrics"
 	"antrea.io/antrea/pkg/util/k8s"
 )
 
@@ -58,7 +63,10 @@ const (
 	// maxEgressMark is the maximum mark of Egress IPs can be configured on a Node.
 	maxEgressMark = 255
 
-	egressIPIndex = "egressIP"
+	egressIPIndex       = "egressIP"
+	externalIPPoolIndex = "externalIPPool"
+
+	DefaultEgressRunDir = "/var/run/antrea/egress"
 )
 
 var emptyWatch = watch.NewEmptyWatch()
@@ -101,6 +109,7 @@ type egressBinding struct {
 type EgressController struct {
 	ofClient             openflow.Client
 	routeClient          route.Interface
+	crdClient            clientsetversioned.Interface
 	antreaClientProvider agent.AntreaClientProvider
 
 	egressInformer     cache.SharedIndexInformer
@@ -127,21 +136,30 @@ type EgressController struct {
 
 	egressIPStates      map[string]*egressIPState
 	egressIPStatesMutex sync.Mutex
+
+	cluster    *memberlist.Cluster
+	ipAssigner ipassigner.IPAssigner
 }
 
 func NewEgressController(
 	ofClient openflow.Client,
-	egressInformer crdinformers.EgressInformer,
 	antreaClientGetter agent.AntreaClientProvider,
+	crdClient clientsetversioned.Interface,
 	ifaceStore interfacestore.InterfaceStore,
 	routeClient route.Interface,
 	nodeName string,
-) *EgressController {
+	nodeIP net.IP,
+	clusterPort int,
+	egressInformer crdinformers.EgressInformer,
+	nodeInformer coreinformers.NodeInformer,
+	externalIPPoolInformer crdinformers.ExternalIPPoolInformer,
+) (*EgressController, error) {
 	localIPDetector := NewLocalIPDetector()
 	c := &EgressController{
 		ofClient:             ofClient,
 		routeClient:          routeClient,
 		antreaClientProvider: antreaClientGetter,
+		crdClient:            crdClient,
 		queue:                workqueue.NewNamedRateLimitingQueue(workqueue.NewItemExponentialFailureRateLimiter(minRetryDelay, maxRetryDelay), "egressgroup"),
 		egressInformer:       egressInformer.Informer(),
 		egressLister:         egressInformer.Lister(),
@@ -155,6 +173,17 @@ func NewEgressController(
 		localIPDetector:      localIPDetector,
 		idAllocator:          newIDAllocator(minEgressMark, maxEgressMark),
 	}
+	ipAssigner, err := ipassigner.NewIPAssigner(nodeIP, DefaultEgressRunDir)
+	if err != nil {
+		return nil, fmt.Errorf("initializing egressIP assigner failed: %v", err)
+	}
+	c.ipAssigner = ipAssigner
+
+	cluster, err := memberlist.NewCluster(clusterPort, nodeIP, nodeName, nodeInformer, externalIPPoolInformer)
+	if err != nil {
+		return nil, fmt.Errorf("initializing memberlist cluster failed: %v", err)
+	}
+	c.cluster = cluster
 
 	c.egressInformer.AddIndexers(cache.Indexers{egressIPIndex: func(obj interface{}) ([]string, error) {
 		egress, ok := obj.(*crdv1a2.Egress)
@@ -163,6 +192,17 @@ func NewEgressController(
 		}
 		return []string{egress.Spec.EgressIP}, nil
 	}})
+	// externalIPPoolIndex will be used to get all Egresses associated with a given ExternalIPPool.
+	c.egressInformer.AddIndexers(cache.Indexers{externalIPPoolIndex: func(obj interface{}) (strings []string, e error) {
+		egress, ok := obj.(*crdv1a2.Egress)
+		if !ok {
+			return nil, fmt.Errorf("obj is not Egress: %+v", obj)
+		}
+		if egress.Spec.ExternalIPPool == "" {
+			return nil, nil
+		}
+		return []string{egress.Spec.ExternalIPPool}, nil
+	}})
 	c.egressInformer.AddEventHandlerWithResyncPeriod(
 		cache.ResourceEventHandlerFuncs{
 			AddFunc: c.enqueueEgress,
@@ -174,7 +214,8 @@ func NewEgressController(
 		resyncPeriod,
 	)
 	localIPDetector.AddEventHandler(c.onLocalIPUpdate)
-	return c
+	c.cluster.AddClusterEventHandler(c.enqueueEgressesByExternalIPPool)
+	return c, nil
 }
 
 func (c *EgressController) enqueueEgress(obj interface{}) {
@@ -209,6 +250,18 @@ func (c *EgressController) onLocalIPUpdate(ip string, added bool) {
 	}
 }
 
+// enqueueEgressesByExternalIPPool enqueues all Egresses that refer to the provided ExternalIPPool,
+// the ExternalIPPool is affected by a Node update/create/delete event or
+// Node leaves/join cluster event or ExternalIPPool changed.
+func (c *EgressController) enqueueEgressesByExternalIPPool(eipName string) {
+	objects, _ := c.egressInformer.GetIndexer().ByIndex(externalIPPoolIndex, eipName)
+	for _, object := range objects {
+		egress := object.(*crdv1a2.Egress)
+		c.queue.Add(egress.Name)
+	}
+	klog.InfoS("Detected ExternalIPPool event", "ExternalIPPool", eipName, "enqueueEgressNum", len(objects))
+}
+
 // Run will create defaultWorkers workers (go routines) which will process the Egress events from the
 // workqueue.
 func (c *EgressController) Run(stopCh <-chan struct{}) {
@@ -223,6 +276,21 @@ func (c *EgressController) Run(stopCh <-chan struct{}) {
 		return
 	}
 
+	go c.cluster.Run(stopCh)
+
+	// The Egress has been deleted but assigned IP has not been deleted, so agent should delete those IPs when it starts.
+	for egressName := range c.ipAssigner.AssignedIPs() {
+		if _, err := c.egressLister.Get(egressName); err != nil {
+			if errors.IsNotFound(err) {
+				if err := c.ipAssigner.UnassignEgressIP(egressName); err != nil {
+					klog.ErrorS(err, "Unassign EgressIP failed")
+				}
+			} else {
+				klog.ErrorS(err, "Get Egress error", "egressName", egressName)
+			}
+		}
+	}
+
 	go wait.NonSlidingUntil(c.watchEgressGroup, 5*time.Second, stopCh)
 
 	for i := 0; i < defaultWorkers; i++ {
@@ -443,6 +511,21 @@ func (c *EgressController) unbindPodEgress(pod, egress string) (string, bool) {
 	return "", false
 }
 
+func (c *EgressController) updateEgressStatus(egress *crdv1a2.Egress, nodeName string) error {
+	if egress.Status.EgressNode == nodeName {
+		return nil
+	}
+	klog.V(2).InfoS("Updating Egress status", "Egress", egress.Name, "oldNode", egress.Status.EgressNode, "newNode", nodeName)
+	toUpdate := egress.DeepCopy()
+	toUpdate.Status.EgressNode = nodeName
+	if _, err := c.crdClient.CrdV1alpha2().Egresses().UpdateStatus(context.TODO(), toUpdate, metav1.UpdateOptions{}); err != nil {
+		return fmt.Errorf("updating Egress %s status error: %v", egress.Name, err)
+	}
+	klog.V(2).InfoS("Updated Egress status", "Egress", egress.Name)
+	metrics.AntreaEgressStatusUpdates.Inc()
+	return nil
+}
+
 func (c *EgressController) syncEgress(egressName string) error {
 	startTime := time.Now()
 	defer func() {
@@ -461,11 +544,38 @@ func (c *EgressController) syncEgress(egressName string) error {
 			if err := c.uninstallEgress(egressName, eState); err != nil {
 				return err
 			}
+			// Unassign the Egress IP (assigned by agent) from the local Node.
+			if err := c.ipAssigner.UnassignEgressIP(egressName); err != nil {
+				return err
+			}
 			return nil
 		}
 		return err
 	}
 
+	localNodeSelected, err := c.cluster.ShouldSelectEgress(egress)
+	if err != nil {
+		return err
+	}
+	if localNodeSelected {
+		// Assign Egress IP to the local Node.
+		if !c.localIPDetector.IsLocalIP(egress.Spec.EgressIP) {
+			err := c.ipAssigner.AssignEgressIP(egress.Spec.EgressIP, egressName)
+			if err != nil {
+				return err
+			}
+			if err := c.updateEgressStatus(egress, c.nodeName); err != nil {
+				return err
+			}
+			klog.InfoS("Assigned Egress IP", "Egress", egressName, "ip", egress.Spec.EgressIP, "nodeName", c.nodeName)
+		}
+	} else {
+		// Unassign the Egress IP (assigned by agent) from the local Node.
+		if err := c.ipAssigner.UnassignEgressIP(egressName); err != nil {
+			return err
+		}
+	}
+
 	eState, exist := c.getEgressState(egressName)
 	// If the EgressIP changes, uninstalls this Egress first.
 	if exist && eState.egressIP != egress.Spec.EgressIP {
diff --git a/pkg/agent/controller/egress/ipassigner/ip_assigner.go b/pkg/agent/controller/egress/ipassigner/ip_assigner.go
new file mode 100644
index 000000000..8dd2775c5
--- /dev/null
+++ b/pkg/agent/controller/egress/ipassigner/ip_assigner.go
@@ -0,0 +1,22 @@
+// 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 ipassigner
+
+// IPAssigner provides methods to assign or unassign egressIP.
+type IPAssigner interface {
+	AssignEgressIP(egressIP, egressName string) error
+	UnassignEgressIP(egressName string) error
+	AssignedIPs() (ips map[string]string)
+}
diff --git a/pkg/agent/controller/egress/ipassigner/ip_assigner_linux.go b/pkg/agent/controller/egress/ipassigner/ip_assigner_linux.go
new file mode 100644
index 000000000..833c53f12
--- /dev/null
+++ b/pkg/agent/controller/egress/ipassigner/ip_assigner_linux.go
@@ -0,0 +1,186 @@
+// 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 ipassigner
+
+import (
+	"bytes"
+	"errors"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"net"
+	"os"
+	"path/filepath"
+	"sync"
+
+	"github.com/vishvananda/netlink"
+	"k8s.io/klog/v2"
+
+	"antrea.io/antrea/pkg/agent/util"
+	"antrea.io/antrea/pkg/agent/util/arping"
+)
+
+var ipv6NotSupportErr = errors.New("IPv6 not supported")
+
+type ipAssigner struct {
+	egressInterface *net.Interface
+	egressLink      netlink.Link
+	egressRunDir    string
+	mutex           sync.RWMutex
+	assignedEgress  map[string]string
+}
+
+// NewIPAssigner returns an *ipAssigner.
+func NewIPAssigner(nodeIPAddr net.IP, dir string) (*ipAssigner, error) {
+	_, egressInterface, err := util.GetIPNetDeviceFromIP(nodeIPAddr)
+	if err != nil {
+		return nil, fmt.Errorf("get IPNetDevice from ip %v error: %+v", nodeIPAddr, err)
+	}
+	ifaceName := egressInterface.Name
+	egressLink, err := netlink.LinkByName(ifaceName)
+	if err != nil {
+		return nil, fmt.Errorf("get netlink by name(%s) error: %+v", ifaceName, err)
+	}
+	a := &ipAssigner{
+		egressInterface: egressInterface,
+		egressLink:      egressLink,
+		egressRunDir:    dir,
+		assignedEgress:  make(map[string]string),
+	}
+	if err := a.initAssignedIPFiles(); err != nil {
+		return nil, err
+	}
+	return a, nil
+}
+
+func (a *ipAssigner) initAssignedIPFiles() (err error) {
+	files, err := ioutil.ReadDir(a.egressRunDir)
+	if err != nil {
+		if err := os.MkdirAll(a.egressRunDir, os.ModeDir); err != nil {
+			return fmt.Errorf("error when creating Egress run dir: %v", err)
+		}
+		return nil
+	}
+	a.mutex.Lock()
+	defer a.mutex.Unlock()
+	for _, file := range files {
+		egressName := file.Name()
+		fileName := ipSavedFile(a.egressRunDir, egressName)
+		if ip, err := ioutil.ReadFile(fileName); err == nil {
+			a.assignedEgress[egressName] = string(ip)
+		}
+	}
+	return
+}
+
+// AssignEgressIP assign an egressIP and save persistent files.
+func (a *ipAssigner) AssignEgressIP(egressIP, egressName string) error {
+	egressSpecIP := net.ParseIP(egressIP)
+	if egressSpecIP == nil {
+		return fmt.Errorf("invalid IP %s", egressIP)
+	}
+	addr := netlink.Addr{IPNet: &net.IPNet{IP: egressSpecIP, Mask: net.CIDRMask(32, 32)}}
+	ifaceName := a.egressInterface.Name
+	if err := netlink.AddrAdd(a.egressLink, &addr); err != nil {
+		return fmt.Errorf("failed to add ip %v to interface %s: %v", addr, ifaceName, err)
+	}
+	if err := a.saveIPAssignFile(egressIP, egressName); err != nil {
+		return fmt.Errorf("failed to save Egress IP assign, Egress %s, IP: %s: %v", egressName, egressIP, err)
+	}
+	isIPv4 := egressSpecIP.To4()
+	if isIPv4 != nil {
+		if err := arping.GratuitousARPOverIface(isIPv4, a.egressInterface); err != nil {
+			return fmt.Errorf("failed to send gratuitous ARP: %v", err)
+		}
+		klog.V(2).InfoS("Sent gratuitous ARP", "ip", isIPv4)
+	} else if isIPv6 := egressSpecIP.To16(); isIPv6 != nil {
+		return fmt.Errorf("failed to send Advertisement: %v", ipv6NotSupportErr)
+	}
+	klog.V(2).InfoS("Added ip", "ip", addr, "interface", ifaceName)
+	return nil
+}
+
+// UnassignEgressIP unassign an egressIP and delete the persistent files.
+func (a *ipAssigner) UnassignEgressIP(egressName string) error {
+	egressIP, has := a.isAssignedIP(egressName)
+	if !has {
+		return nil
+	}
+	egressSpecIP := net.ParseIP(egressIP)
+	addr := netlink.Addr{IPNet: &net.IPNet{IP: egressSpecIP, Mask: net.CIDRMask(32, 32)}}
+	ifaceName := a.egressLink.Attrs().Name
+	if err := netlink.AddrDel(a.egressLink, &addr); err != nil {
+		return fmt.Errorf("failed to delete ip %v from interface %s: %v", addr, ifaceName, err)
+	}
+	if err := a.removeAssignedIPFile(egressName); err != nil {
+		return fmt.Errorf("failed to remove Egress IP assign file, egressName: %s, egressIP: %s: %v", egressName, egressIP, err)
+	}
+	klog.V(2).InfoS("Deleted ip", "ip", addr, "interface", ifaceName)
+	return nil
+}
+
+// isAssignedIP check that if an IP address has been assigned with a specific egressName.
+func (a *ipAssigner) isAssignedIP(egressName string) (egressIP string, has bool) {
+	a.mutex.RLock()
+	defer a.mutex.RUnlock()
+	if ip, ok := a.assignedEgress[egressName]; ok {
+		return ip, true
+	}
+	return
+}
+
+func ipSavedFile(dir, name string) string {
+	return filepath.Join(dir, name)
+}
+
+func (a *ipAssigner) saveIPAssignFile(egressIP, egressName string) error {
+	var buffer bytes.Buffer
+	buffer.WriteString(egressIP)
+	f, err := os.Create(ipSavedFile(a.egressRunDir, egressName))
+	if err != nil {
+		return err
+	}
+	defer f.Close()
+	_, err = io.Copy(f, &buffer)
+	if err == nil {
+		a.mutex.Lock()
+		defer a.mutex.Unlock()
+		a.assignedEgress[egressName] = egressIP
+		return nil
+	}
+	return err
+}
+
+func (a *ipAssigner) removeAssignedIPFile(egressName string) error {
+	fileName := ipSavedFile(a.egressRunDir, egressName)
+	if err := os.Remove(fileName); err != nil && !os.IsNotExist(err) {
+		return err
+	}
+	a.mutex.Lock()
+	defer a.mutex.Unlock()
+	delete(a.assignedEgress, egressName)
+	return nil
+}
+
+// AssignedIPs returns a map of the allocated IPs ([egressName]IP).
+func (a *ipAssigner) AssignedIPs() map[string]string {
+	ips := make(map[string]string)
+	a.mutex.RLock()
+	defer a.mutex.RUnlock()
+	for k, v := range a.assignedEgress {
+		ips[k] = v
+	}
+	return ips
+}
diff --git a/pkg/agent/controller/egress/ipassigner/ip_assigner_windows.go b/pkg/agent/controller/egress/ipassigner/ip_assigner_windows.go
new file mode 100644
index 000000000..0b0530652
--- /dev/null
+++ b/pkg/agent/controller/egress/ipassigner/ip_assigner_windows.go
@@ -0,0 +1,38 @@
+// 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 ipassigner
+
+import (
+	"net"
+)
+
+type ipAssigner struct {
+}
+
+func NewIPAssigner(nodeIPAddr net.IP, dir string) (*ipAssigner, error) {
+	return nil, nil
+}
+
+func (a *ipAssigner) AssignEgressIP(egressIP, egressName string) error {
+	return nil
+}
+
+func (a *ipAssigner) UnassignEgressIP(egressName string) error {
+	return nil
+}
+
+func (a *ipAssigner) AssignedIPs() (ips map[string]string) {
+	return
+}
diff --git a/pkg/agent/flowexporter/connections/conntrack.go b/pkg/agent/flowexporter/connections/conntrack.go
index ef8cbf982..4dc509757 100644
--- a/pkg/agent/flowexporter/connections/conntrack.go
+++ b/pkg/agent/flowexporter/connections/conntrack.go
@@ -46,11 +46,9 @@ func filterAntreaConns(conns []*flowexporter.Connection, nodeConfig *config.Node
 
 		// Consider Pod-to-Pod, Pod-To-Service and Pod-To-External flows.
 		if srcIP.Equal(nodeConfig.GatewayConfig.IPv4) || dstIP.Equal(nodeConfig.GatewayConfig.IPv4) {
-			klog.V(4).Infof("Detected flow for which one of the endpoint is host gateway %s :%+v", nodeConfig.GatewayConfig.IPv4.String(), conn)
 			continue
 		}
 		if srcIP.Equal(nodeConfig.GatewayConfig.IPv6) || dstIP.Equal(nodeConfig.GatewayConfig.IPv6) {
-			klog.V(4).Infof("Detected flow for which one of the endpoint is host gateway %s :%+v", nodeConfig.GatewayConfig.IPv6.String(), conn)
 			continue
 		}
 
diff --git a/pkg/agent/flowexporter/connections/conntrack_connections.go b/pkg/agent/flowexporter/connections/conntrack_connections.go
index 1bb703b3e..9a6b2a61d 100644
--- a/pkg/agent/flowexporter/connections/conntrack_connections.go
+++ b/pkg/agent/flowexporter/connections/conntrack_connections.go
@@ -226,7 +226,9 @@ func (cs *ConntrackConnectionStore) AddOrUpdateConn(conn *flowexporter.Connectio
 			}
 		}
 		cs.addNetworkPolicyMetadata(conn)
-
+		if conn.StartTime.IsZero() {
+			conn.StartTime = time.Now()
+		}
 		metrics.TotalAntreaConnectionsInConnTrackTable.Inc()
 		klog.V(4).Infof("New Antrea flow added: %v", conn)
 		// Add new antrea connection to connection store
diff --git a/pkg/agent/flowexporter/exporter/exporter.go b/pkg/agent/flowexporter/exporter/exporter.go
index 3d4069370..dd364c1f3 100644
--- a/pkg/agent/flowexporter/exporter/exporter.go
+++ b/pkg/agent/flowexporter/exporter/exporter.go
@@ -313,6 +313,7 @@ func (exp *flowExporter) sendFlowRecords() error {
 			} else {
 				exp.flowRecords.ValidateAndUpdateStats(key, record)
 			}
+			klog.V(4).InfoS("Record sent successfully", "flowKey", key, "record", record)
 		}
 		return nil
 	}
@@ -330,6 +331,7 @@ func (exp *flowExporter) sendFlowRecords() error {
 				return err
 			}
 			exp.numDataSetsSent = exp.numDataSetsSent + 1
+			klog.V(4).InfoS("Record for deny connection sent successfully", "flowKey", connKey, "connection", conn)
 			exp.denyConnStore.ResetConnStatsWithoutLock(connKey)
 		}
 		if time.Since(conn.LastExportTime) >= exp.idleFlowTimeout {
@@ -340,6 +342,7 @@ func (exp *flowExporter) sendFlowRecords() error {
 				return err
 			}
 			exp.numDataSetsSent = exp.numDataSetsSent + 1
+			klog.V(4).InfoS("Record for deny connection sent successfully", "flowKey", connKey, "connection", conn)
 			exp.denyConnStore.DeleteConnWithoutLock(connKey)
 		}
 		return nil
diff --git a/pkg/agent/memberlist/cluster.go b/pkg/agent/memberlist/cluster.go
new file mode 100644
index 000000000..27b38ba4a
--- /dev/null
+++ b/pkg/agent/memberlist/cluster.go
@@ -0,0 +1,487 @@
+// 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 memberlist
+
+import (
+	"fmt"
+	"io/ioutil"
+	"net"
+	"reflect"
+	"sync"
+	"time"
+
+	"github.com/golang/groupcache/consistenthash"
+	"github.com/hashicorp/memberlist"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/apimachinery/pkg/util/wait"
+	coreinformers "k8s.io/client-go/informers/core/v1"
+	corelisters "k8s.io/client-go/listers/core/v1"
+	"k8s.io/client-go/tools/cache"
+	"k8s.io/client-go/util/workqueue"
+	"k8s.io/klog/v2"
+
+	"antrea.io/antrea/pkg/apis/crd/v1alpha2"
+	crdinformers "antrea.io/antrea/pkg/client/informers/externalversions/crd/v1alpha2"
+	crdlister "antrea.io/antrea/pkg/client/listers/crd/v1alpha2"
+	"antrea.io/antrea/pkg/util/k8s"
+)
+
+const (
+	controllerName = "MemberListCluster"
+	// Set resyncPeriod to 0 to disable resyncing.
+	resyncPeriod time.Duration = 0
+	// Set default virtual node replicas num of consistent hash
+	// in order to improve the quality of the hash distribution, refs https://github.com/golang/groupcache/issues/29
+	defaultVirtualNodeReplicas = 50
+	// How long to wait before retrying the processing of an ExternalIPPool change.
+	minRetryDelay = 5 * time.Second
+	maxRetryDelay = 300 * time.Second
+	// Default number of workers processing an ExternalIPPool change.
+	defaultWorkers = 4
+
+	nodeEventTypeJoin   nodeEventType = "Join"
+	nodeEventTypeLeave  nodeEventType = "Leave"
+	nodeEventTypeUpdate nodeEventType = "Update"
+)
+
+type nodeEventType string
+
+// Default Hash Fn is crc32.ChecksumIEEE.
+var defaultHashFn func(data []byte) uint32
+
+var (
+	errDecodingObject          = fmt.Errorf("received unexpected object")
+	errDecodingObjectTombstone = fmt.Errorf("deletedFinalStateUnknown contains unexpected object")
+)
+
+var mapNodeEventType = map[memberlist.NodeEventType]nodeEventType{
+	memberlist.NodeJoin:   nodeEventTypeJoin,
+	memberlist.NodeLeave:  nodeEventTypeLeave,
+	memberlist.NodeUpdate: nodeEventTypeUpdate,
+}
+
+type clusterNodeEventHandler func(objName string)
+
+// Cluster implements ClusterInterface.
+type Cluster struct {
+	bindPort int
+	// IP addr of local Node.
+	localNodeIP net.IP
+	// Name of local Node. Node name must be unique in the cluster.
+	nodeName string
+
+	mList *memberlist.Memberlist
+	// consistentHash hold the consistentHashMap, when a Node join cluster, use method Add() to add a key to the hash.
+	// when a Node leave the cluster, the consistentHashMap should be update.
+	consistentHashMap     map[string]*consistenthash.Map
+	consistentHashRWMutex sync.RWMutex
+	// nodeEventsCh, the Node join/leave events will be notified via it.
+	nodeEventsCh chan memberlist.NodeEvent
+
+	// clusterNodeEventHandlers contains eventHandler which will run when consistentHashMap is updated,
+	// which caused by an ExternalIPPool or Node event, such as cluster Node status update(leave of join cluster),
+	// ExternalIPPool events(create/update/delete).
+	// For example, when a new Node joins the cluster, each Node should compute whether it should still hold all
+	// its existing Egresses, and when a Node leaves the cluster,
+	// each Node should check whether it is now responsible for some of the Egresses from that Node.
+	clusterNodeEventHandlers []clusterNodeEventHandler
+
+	nodeInformer     coreinformers.NodeInformer
+	nodeLister       corelisters.NodeLister
+	nodeListerSynced cache.InformerSynced
+
+	externalIPPoolInformer          cache.SharedIndexInformer
+	externalIPPoolLister            crdlister.ExternalIPPoolLister
+	externalIPPoolInformerHasSynced cache.InformerSynced
+
+	// queue maintains the ExternalIPPool names that need to be synced.
+	queue workqueue.RateLimitingInterface
+}
+
+// NewCluster returns a new *Cluster.
+func NewCluster(
+	clusterBindPort int,
+	localNodeIP net.IP,
+	nodeName string,
+	nodeInformer coreinformers.NodeInformer,
+	externalIPPoolInformer crdinformers.ExternalIPPoolInformer,
+) (*Cluster, error) {
+	// The Node join/leave events will be notified via it.
+	nodeEventCh := make(chan memberlist.NodeEvent, 1024)
+	c := &Cluster{
+		bindPort:                        clusterBindPort,
+		localNodeIP:                     localNodeIP,
+		nodeName:                        nodeName,
+		consistentHashMap:               make(map[string]*consistenthash.Map),
+		nodeEventsCh:                    nodeEventCh,
+		nodeInformer:                    nodeInformer,
+		nodeLister:                      nodeInformer.Lister(),
+		nodeListerSynced:                nodeInformer.Informer().HasSynced,
+		externalIPPoolInformer:          externalIPPoolInformer.Informer(),
+		externalIPPoolLister:            externalIPPoolInformer.Lister(),
+		externalIPPoolInformerHasSynced: externalIPPoolInformer.Informer().HasSynced,
+		queue:                           workqueue.NewNamedRateLimitingQueue(workqueue.NewItemExponentialFailureRateLimiter(minRetryDelay, maxRetryDelay), "externalIPPool"),
+	}
+
+	conf := memberlist.DefaultLocalConfig()
+	conf.Name = c.nodeName
+	conf.BindPort = c.bindPort
+	conf.AdvertisePort = c.bindPort
+	conf.Events = &memberlist.ChannelEventDelegate{Ch: nodeEventCh}
+	conf.LogOutput = ioutil.Discard
+	klog.V(1).InfoS("New memberlist cluster", "config", conf)
+
+	mList, err := memberlist.Create(conf)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create memberlist cluster: %v", err)
+	}
+	c.mList = mList
+
+	nodeInformer.Informer().AddEventHandlerWithResyncPeriod(
+		cache.ResourceEventHandlerFuncs{
+			AddFunc:    c.handleCreateNode,
+			UpdateFunc: c.handleUpdateNode,
+			DeleteFunc: c.handleDeleteNode,
+		},
+		resyncPeriod,
+	)
+	externalIPPoolInformer.Informer().AddEventHandlerWithResyncPeriod(
+		cache.ResourceEventHandlerFuncs{
+			AddFunc: c.enqueueExternalIPPool,
+			UpdateFunc: func(oldObj, newObj interface{}) {
+				c.enqueueExternalIPPool(newObj)
+			},
+			DeleteFunc: c.enqueueExternalIPPool,
+		},
+		resyncPeriod,
+	)
+	return c, nil
+}
+
+func (c *Cluster) handleCreateNode(obj interface{}) {
+	node := obj.(*corev1.Node)
+	if member, err := c.newClusterMember(node); err == nil {
+		_, err := c.mList.Join([]string{member})
+		if err != nil {
+			klog.ErrorS(err, "Processing Node CREATE event error, join cluster failed", "member", member)
+		}
+	} else {
+		klog.ErrorS(err, "Processing Node CREATE event error", "nodeName", node.Name)
+	}
+
+	affectedEIPs := c.filterEIPsFromNodeLabels(node)
+	c.enqueueExternalIPPools(affectedEIPs)
+	klog.V(2).InfoS("Processed Node CREATE event", "nodeName", node.Name, "affectedExternalIPPoolNum", affectedEIPs.Len())
+}
+
+func (c *Cluster) handleDeleteNode(obj interface{}) {
+	node, ok := obj.(*corev1.Node)
+	if !ok {
+		tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
+		if !ok {
+			klog.ErrorS(errDecodingObject, "Processing Node DELETE event error", "obj", obj)
+			return
+		}
+		node, ok = tombstone.Obj.(*corev1.Node)
+		if !ok {
+			klog.ErrorS(errDecodingObjectTombstone, "Processing Node DELETE event error", "obj", tombstone.Obj)
+			return
+		}
+	}
+	affectedEIPs := c.filterEIPsFromNodeLabels(node)
+	c.enqueueExternalIPPools(affectedEIPs)
+	klog.V(2).InfoS("Processed Node DELETE event", "nodeName", node.Name, "affectedExternalIPPoolNum", affectedEIPs.Len())
+}
+
+func (c *Cluster) handleUpdateNode(oldObj, newObj interface{}) {
+	node := newObj.(*corev1.Node)
+	oldNode := oldObj.(*corev1.Node)
+	if reflect.DeepEqual(node.GetLabels(), oldNode.GetLabels()) {
+		klog.V(2).InfoS("Processing Node UPDATE event error, labels not changed", "nodeName", node.Name)
+		return
+	}
+	oldMatches, newMatches := c.filterEIPsFromNodeLabels(oldNode), c.filterEIPsFromNodeLabels(node)
+	if oldMatches.Equal(newMatches) {
+		klog.V(2).InfoS("Processing Node UPDATE event error, Node cluster status not changed", "nodeName", node.Name)
+		return
+	}
+	affectedEIPs := oldMatches.Union(newMatches)
+	c.enqueueExternalIPPools(affectedEIPs)
+	klog.V(2).InfoS("Processed Node UPDATE event", "nodeName", node.Name, "affectedExternalIPPoolNum", affectedEIPs.Len())
+}
+
+func (c *Cluster) enqueueExternalIPPools(eips sets.String) {
+	for eip := range eips {
+		c.queue.Add(eip)
+	}
+}
+
+func (c *Cluster) enqueueExternalIPPool(obj interface{}) {
+	eip, ok := obj.(*v1alpha2.ExternalIPPool)
+	if !ok {
+		deletedState, ok := obj.(cache.DeletedFinalStateUnknown)
+		if !ok {
+			klog.ErrorS(errDecodingObject, "Processing ExternalIPPool DELETE event error", "obj", obj)
+			return
+		}
+		eip, ok = deletedState.Obj.(*v1alpha2.ExternalIPPool)
+		if !ok {
+			klog.ErrorS(errDecodingObjectTombstone, "Processing ExternalIPPool DELETE event error", "obj", deletedState.Obj)
+			return
+		}
+	}
+	c.queue.Add(eip.Name)
+}
+
+// newClusterMember gets the Node's IP and returns a cluster member "<IP>:<clusterMemberlistPort>"
+// representing that Node in the memberlist cluster.
+func (c *Cluster) newClusterMember(node *corev1.Node) (string, error) {
+	nodeAddr, err := k8s.GetNodeAddr(node)
+	if err != nil {
+		return "", fmt.Errorf("obtain IP address from K8s Node failed: %v", err)
+	}
+	member := fmt.Sprintf("%s:%d", nodeAddr, c.bindPort)
+	return member, nil
+}
+
+func (c *Cluster) allClusterMembers() (clusterNodes []string, err error) {
+	nodes, err := c.nodeLister.List(labels.Everything())
+	if err != nil {
+		return nil, fmt.Errorf("listing Nodes error: %v", err)
+	}
+
+	for _, node := range nodes {
+		member, err := c.newClusterMember(node)
+		if err != nil {
+			klog.ErrorS(err, "Get Node failed")
+			continue
+		}
+		clusterNodes = append(clusterNodes, member)
+	}
+	return
+}
+
+func (c *Cluster) filterEIPsFromNodeLabels(node *corev1.Node) sets.String {
+	pools := sets.NewString()
+	eips, err := c.externalIPPoolLister.List(labels.Everything())
+	if err != nil {
+		klog.ErrorS(err, "Filter ExternalIPPools from nodeLabels failed")
+		return pools
+	}
+	for _, eip := range eips {
+		nodeSelector, _ := metav1.LabelSelectorAsSelector(&eip.Spec.NodeSelector)
+		if nodeSelector.Matches(labels.Set(node.GetLabels())) {
+			pools.Insert(eip.Name)
+		}
+	}
+	return pools
+}
+
+// Run will join all the other K8s Nodes in a memberlist cluster
+// and will create defaultWorkers workers (go routines) which will process the ExternalIPPool or Node events
+// from the work queue.
+func (c *Cluster) Run(stopCh <-chan struct{}) {
+	defer c.queue.ShutDown()
+	// In order to exit the cluster more gracefully, call Leave prior to shutting down.
+	defer close(c.nodeEventsCh)
+	defer c.mList.Shutdown()
+	defer c.mList.Leave(time.Second)
+
+	klog.InfoS("Starting", "controllerName", controllerName)
+	defer klog.InfoS("Shutting down", "controllerName", controllerName)
+
+	if !cache.WaitForNamedCacheSync(controllerName, stopCh, c.externalIPPoolInformerHasSynced, c.nodeListerSynced) {
+		return
+	}
+
+	members, err := c.allClusterMembers()
+	if err != nil {
+		klog.ErrorS(err, "List cluster members failed")
+	} else if members != nil {
+		_, err := c.mList.Join(members)
+		if err != nil {
+			klog.ErrorS(err, "Join cluster failed")
+		}
+	}
+
+	for i := 0; i < defaultWorkers; i++ {
+		go wait.Until(c.worker, time.Second, stopCh)
+	}
+
+	for {
+		select {
+		case <-stopCh:
+			return
+		case nodeEvent := <-c.nodeEventsCh:
+			c.handleClusterNodeEvents(&nodeEvent)
+		}
+	}
+}
+
+func (c *Cluster) worker() {
+	for c.processNextWorkItem() {
+	}
+}
+
+func (c *Cluster) processNextWorkItem() bool {
+	obj, quit := c.queue.Get()
+	if quit {
+		return false
+	}
+	defer c.queue.Done(obj)
+
+	// We expect strings (ExternalIPPool name) to come off the work queue.
+	if key, ok := obj.(string); !ok {
+		// As the item in the work queue is actually invalid, we call Forget here else we'd
+		// go into a loop of attempting to process a work item that is invalid.
+		// This should not happen.
+		c.queue.Forget(obj)
+		klog.Errorf("Expected string in work queue but got %#v", obj)
+		return true
+	} else if err := c.syncConsistentHash(key); err == nil {
+		// If no error occurs we Forget this item so it does not get queued again until
+		// another change happens.
+		c.queue.Forget(key)
+	} else {
+		// Put the item back on the work queue to handle any transient errors.
+		c.queue.AddRateLimited(key)
+		klog.ErrorS(err, "Syncing consistentHash by ExternalIPPool failed, requeue", "ExternalIPPool", key)
+	}
+	return true
+}
+
+func (c *Cluster) syncConsistentHash(eipName string) error {
+	startTime := time.Now()
+	defer func() {
+		klog.V(4).InfoS("Finished syncing consistentHash", "ExternalIPPool", eipName, "durationTime", time.Since(startTime))
+	}()
+
+	eip, err := c.externalIPPoolLister.Get(eipName)
+	if err != nil {
+		if errors.IsNotFound(err) {
+			c.consistentHashRWMutex.Lock()
+			defer c.consistentHashRWMutex.Unlock()
+			delete(c.consistentHashMap, eipName)
+			return nil
+		}
+		return err
+	}
+
+	// updateConsistentHash refreshes the consistentHashMap.
+	updateConsistentHash := func(eip *v1alpha2.ExternalIPPool) error {
+		nodeSel, err := metav1.LabelSelectorAsSelector(&eip.Spec.NodeSelector)
+		if err != nil {
+			return err
+		}
+		nodes, err := c.nodeLister.List(nodeSel)
+		if err != nil {
+			return err
+		}
+		aliveNodes := c.aliveNodes()
+		// Node alive and Node labels match ExternalIPPool nodeSelector.
+		var aliveAndMatchedNodes []string
+		for _, node := range nodes {
+			nodeName := node.Name
+			if aliveNodes.Has(nodeName) {
+				aliveAndMatchedNodes = append(aliveAndMatchedNodes, nodeName)
+			}
+		}
+		consistentHashMap := newNodeConsistentHashMap()
+		consistentHashMap.Add(aliveAndMatchedNodes...)
+		c.consistentHashRWMutex.Lock()
+		defer c.consistentHashRWMutex.Unlock()
+		c.consistentHashMap[eip.Name] = consistentHashMap
+		c.notify(eip.Name)
+		return nil
+	}
+
+	if err := updateConsistentHash(eip); err != nil {
+		return err
+	}
+	return nil
+}
+
+func newNodeConsistentHashMap() *consistenthash.Map {
+	return consistenthash.New(defaultVirtualNodeReplicas, defaultHashFn)
+}
+
+func (c *Cluster) handleClusterNodeEvents(nodeEvent *memberlist.NodeEvent) {
+	node, event := nodeEvent.Node, nodeEvent.Event
+	switch event {
+	case memberlist.NodeJoin, memberlist.NodeLeave:
+		// When a Node joins cluster, all matched ExternalIPPools consistentHash should be updated;
+		// when a Node leaves cluster, the Node may have failed or have been deleted,
+		// if the Node has been deleted, affected ExternalIPPool should be enqueued, and deleteNode handler has been executed,
+		// if the Node has failed, ExternalIPPools consistentHash maybe changed, and affected ExternalIPPool should be enqueued.
+		coreNode, err := c.nodeLister.Get(node.Name)
+		if err != nil {
+			if errors.IsNotFound(err) {
+				// Node has been deleted, and deleteNode handler has been executed.
+				klog.ErrorS(err, "Processing Node event, not found", "eventType", event)
+				return
+			}
+			klog.ErrorS(err, "Processing Node event, get Node failed", "eventType", event)
+			return
+		}
+		affectedEIPs := c.filterEIPsFromNodeLabels(coreNode)
+		c.enqueueExternalIPPools(affectedEIPs)
+		klog.InfoS("Processed Node event", "eventType", mapNodeEventType[event], "nodeName", node.Name, "affectedExternalIPPoolNum", len(affectedEIPs))
+	default:
+		klog.InfoS("Processed Node event", "eventType", mapNodeEventType[event], "nodeName", node.Name)
+	}
+}
+
+// aliveNodes returns the list of nodeNames in the cluster.
+func (c *Cluster) aliveNodes() sets.String {
+	nodes := sets.NewString()
+	for _, node := range c.mList.Members() {
+		nodes.Insert(node.Name)
+	}
+	return nodes
+}
+
+// ShouldSelectEgress returns true if the local Node selected as the owner Node of the Egress,
+// the local Node in the cluster holds the same consistent hash ring for each ExternalIPPool,
+// consistentHash.Get gets the closest item (Node name) in the hash to the provided key(egressIP),
+// if the name of the local Node is equal to the name of the selected Node, returns true.
+func (c *Cluster) ShouldSelectEgress(egress *v1alpha2.Egress) (bool, error) {
+	eipName := egress.Spec.ExternalIPPool
+	if eipName == "" || egress.Spec.EgressIP == "" {
+		return false, nil
+	}
+	c.consistentHashRWMutex.RLock()
+	defer c.consistentHashRWMutex.RUnlock()
+	consistentHash, ok := c.consistentHashMap[eipName]
+	if !ok {
+		return false, fmt.Errorf("local Node consistentHashMap has not synced, ExternalIPPool %s", eipName)
+	}
+	return consistentHash.Get(egress.Spec.EgressIP) == c.nodeName, nil
+}
+
+func (c *Cluster) notify(objName string) {
+	for _, handler := range c.clusterNodeEventHandlers {
+		handler(objName)
+	}
+}
+
+// AddClusterEventHandler adds a clusterNodeEventHandler, which will run when consistentHashMap is updated,
+// due to an ExternalIPPool or Node event.
+func (c *Cluster) AddClusterEventHandler(handler clusterNodeEventHandler) {
+	c.clusterNodeEventHandlers = append(c.clusterNodeEventHandlers, handler)
+}
diff --git a/pkg/apis/crd/v1alpha2/types.go b/pkg/apis/crd/v1alpha2/types.go
index 6393886a4..671920b69 100644
--- a/pkg/apis/crd/v1alpha2/types.go
+++ b/pkg/apis/crd/v1alpha2/types.go
@@ -190,7 +190,6 @@ type AppliedTo struct {
 
 // +genclient
 // +genclient:nonNamespaced
-// +genclient:noStatus
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
 
 // Egress defines which egress (SNAT) IP the traffic from the selected Pods to
@@ -202,6 +201,15 @@ type Egress struct {
 
 	// Specification of the desired behavior of Egress.
 	Spec EgressSpec `json:"spec"`
+
+	// EgressStatus represents the current status of an Egress.
+	Status EgressStatus `json:"status"`
+}
+
+// EgressStatus represents the current status of an Egress.
+type EgressStatus struct {
+	// The name of the Node that holds the Egress IP.
+	EgressNode string `json:"egressNode"`
 }
 
 // EgressSpec defines the desired state for Egress.
diff --git a/pkg/apis/crd/v1alpha2/zz_generated.deepcopy.go b/pkg/apis/crd/v1alpha2/zz_generated.deepcopy.go
index cdf7e2358..bb2587fb0 100644
--- a/pkg/apis/crd/v1alpha2/zz_generated.deepcopy.go
+++ b/pkg/apis/crd/v1alpha2/zz_generated.deepcopy.go
@@ -122,6 +122,7 @@ func (in *Egress) DeepCopyInto(out *Egress) {
 	out.TypeMeta = in.TypeMeta
 	in.ObjectMeta.DeepCopyInto(&out.ObjectMeta)
 	in.Spec.DeepCopyInto(&out.Spec)
+	out.Status = in.Status
 	return
 }
 
@@ -193,6 +194,22 @@ func (in *EgressSpec) DeepCopy() *EgressSpec {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *EgressStatus) DeepCopyInto(out *EgressStatus) {
+	*out = *in
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new EgressStatus.
+func (in *EgressStatus) DeepCopy() *EgressStatus {
+	if in == nil {
+		return nil
+	}
+	out := new(EgressStatus)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *Endpoint) DeepCopyInto(out *Endpoint) {
 	*out = *in
diff --git a/pkg/apis/ports.go b/pkg/apis/ports.go
index f8e54df92..c22a0bd5a 100644
--- a/pkg/apis/ports.go
+++ b/pkg/apis/ports.go
@@ -19,4 +19,7 @@ const (
 	AntreaControllerAPIPort = 10349
 	// AntreaAgentAPIPort is the default port for the antrea-agent APIServer.
 	AntreaAgentAPIPort = 10350
+	// AntreaAgentClusterMembershipPort is the default port for the antrea-agent cluster.
+	// A gossip-based cluster will be created in the background when the egress feature is turned on.
+	AntreaAgentClusterMembershipPort = 10351
 )
diff --git a/pkg/apis/system/v1beta1/zz_generated.deepcopy.go b/pkg/apis/system/v1beta1/zz_generated.deepcopy.go
index 103dd4a15..45b077887 100644
--- a/pkg/apis/system/v1beta1/zz_generated.deepcopy.go
+++ b/pkg/apis/system/v1beta1/zz_generated.deepcopy.go
@@ -1,6 +1,6 @@
 // +build !ignore_autogenerated
 
-// Copyright 2020 Antrea Authors
+// 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.
diff --git a/pkg/client/clientset/versioned/typed/crd/v1alpha2/egress.go b/pkg/client/clientset/versioned/typed/crd/v1alpha2/egress.go
index 784d5c1ae..52326a2f0 100644
--- a/pkg/client/clientset/versioned/typed/crd/v1alpha2/egress.go
+++ b/pkg/client/clientset/versioned/typed/crd/v1alpha2/egress.go
@@ -38,6 +38,7 @@ type EgressesGetter interface {
 type EgressInterface interface {
 	Create(ctx context.Context, egress *v1alpha2.Egress, opts v1.CreateOptions) (*v1alpha2.Egress, error)
 	Update(ctx context.Context, egress *v1alpha2.Egress, opts v1.UpdateOptions) (*v1alpha2.Egress, error)
+	UpdateStatus(ctx context.Context, egress *v1alpha2.Egress, opts v1.UpdateOptions) (*v1alpha2.Egress, error)
 	Delete(ctx context.Context, name string, opts v1.DeleteOptions) error
 	DeleteCollection(ctx context.Context, opts v1.DeleteOptions, listOpts v1.ListOptions) error
 	Get(ctx context.Context, name string, opts v1.GetOptions) (*v1alpha2.Egress, error)
@@ -126,6 +127,21 @@ func (c *egresses) Update(ctx context.Context, egress *v1alpha2.Egress, opts v1.
 	return
 }
 
+// UpdateStatus was generated because the type contains a Status member.
+// Add a +genclient:noStatus comment above the type to avoid generating UpdateStatus().
+func (c *egresses) UpdateStatus(ctx context.Context, egress *v1alpha2.Egress, opts v1.UpdateOptions) (result *v1alpha2.Egress, err error) {
+	result = &v1alpha2.Egress{}
+	err = c.client.Put().
+		Resource("egresses").
+		Name(egress.Name).
+		SubResource("status").
+		VersionedParams(&opts, scheme.ParameterCodec).
+		Body(egress).
+		Do(ctx).
+		Into(result)
+	return
+}
+
 // Delete takes name of the egress and deletes it. Returns an error if one occurs.
 func (c *egresses) Delete(ctx context.Context, name string, opts v1.DeleteOptions) error {
 	return c.client.Delete().
diff --git a/pkg/client/clientset/versioned/typed/crd/v1alpha2/fake/fake_egress.go b/pkg/client/clientset/versioned/typed/crd/v1alpha2/fake/fake_egress.go
index 6e142d760..9c463ebf4 100644
--- a/pkg/client/clientset/versioned/typed/crd/v1alpha2/fake/fake_egress.go
+++ b/pkg/client/clientset/versioned/typed/crd/v1alpha2/fake/fake_egress.go
@@ -94,6 +94,17 @@ func (c *FakeEgresses) Update(ctx context.Context, egress *v1alpha2.Egress, opts
 	return obj.(*v1alpha2.Egress), err
 }
 
+// UpdateStatus was generated because the type contains a Status member.
+// Add a +genclient:noStatus comment above the type to avoid generating UpdateStatus().
+func (c *FakeEgresses) UpdateStatus(ctx context.Context, egress *v1alpha2.Egress, opts v1.UpdateOptions) (*v1alpha2.Egress, error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewRootUpdateSubresourceAction(egressesResource, "status", egress), &v1alpha2.Egress{})
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1alpha2.Egress), err
+}
+
 // Delete takes name of the egress and deletes it. Returns an error if one occurs.
 func (c *FakeEgresses) Delete(ctx context.Context, name string, opts v1.DeleteOptions) error {
 	_, err := c.Fake.
diff --git a/pkg/controller/metrics/prometheus.go b/pkg/controller/metrics/prometheus.go
index 1d8735c0c..d346ab67e 100644
--- a/pkg/controller/metrics/prometheus.go
+++ b/pkg/controller/metrics/prometheus.go
@@ -96,6 +96,13 @@ var (
 		Help:           "The total number of actual status updates performed for Antrea NetworkPolicy Custom Resources",
 		StabilityLevel: metrics.ALPHA,
 	})
+	AntreaEgressStatusUpdates = metrics.NewCounter(&metrics.CounterOpts{
+		Namespace:      metricNamespaceAntrea,
+		Subsystem:      metricSubsystemController,
+		Name:           "eg_status_updates",
+		Help:           "The total number of actual status updates performed for Antrea Egress Custom Resources",
+		StabilityLevel: metrics.ALPHA,
+	})
 	AntreaClusterNetworkPolicyStatusUpdates = metrics.NewCounter(&metrics.CounterOpts{
 		Namespace:      metricNamespaceAntrea,
 		Subsystem:      metricSubsystemController,
diff --git a/pkg/features/antrea_features.go b/pkg/features/antrea_features.go
index 5f448c350..5b6178e67 100644
--- a/pkg/features/antrea_features.go
+++ b/pkg/features/antrea_features.go
@@ -56,6 +56,7 @@ const (
 	FlowExporter featuregate.Feature = "FlowExporter"
 
 	// alpha: v0.10
+	// beta: v1.2
 	// Enable collecting and exposing NetworkPolicy statistics.
 	NetworkPolicyStats featuregate.Feature = "NetworkPolicyStats"
 
@@ -86,7 +87,7 @@ var (
 		EndpointSlice:      {Default: false, PreRelease: featuregate.Alpha},
 		Traceflow:          {Default: true, PreRelease: featuregate.Beta},
 		FlowExporter:       {Default: false, PreRelease: featuregate.Alpha},
-		NetworkPolicyStats: {Default: false, PreRelease: featuregate.Alpha},
+		NetworkPolicyStats: {Default: true, PreRelease: featuregate.Beta},
 		NodePortLocal:      {Default: false, PreRelease: featuregate.Alpha},
 	}
 
diff --git a/plugins/octant/go.sum b/plugins/octant/go.sum
index 4c8ca2023..1329a118d 100644
--- a/plugins/octant/go.sum
+++ b/plugins/octant/go.sum
@@ -300,8 +300,8 @@ github.com/golang/mock v1.4.0/go.mod h1:UOMv5ysSaYNkG+OFQykRIcU/QvvxJf3p21QfJ2Bt
 github.com/golang/mock v1.4.1/go.mod h1:UOMv5ysSaYNkG+OFQykRIcU/QvvxJf3p21QfJ2Bt3cw=
 github.com/golang/mock v1.4.3/go.mod h1:UOMv5ysSaYNkG+OFQykRIcU/QvvxJf3p21QfJ2Bt3cw=
 github.com/golang/mock v1.4.4/go.mod h1:l3mdAwkq5BuhzHwde/uurv3sEJeZMXNpwsxVWU71h+4=
-github.com/golang/mock v1.5.0 h1:jlYHihg//f7RRwuPfptm04yp4s7O6Kw8EZiVYIGcH0g=
-github.com/golang/mock v1.5.0/go.mod h1:CWnOUgYIOo4TcNZ0wHX3YZCqsaM1I1Jvs6v3mP3KVu8=
+github.com/golang/mock v1.6.0 h1:ErTB+efbowRARo13NNdxyJji2egdxLGQhRaY+DUumQc=
+github.com/golang/mock v1.6.0/go.mod h1:p6yTPP+5HYm5mzsMV8JkE6ZKdX+/wYM6Hr+LicevLPs=
 github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
 github.com/golang/protobuf v1.3.1/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
 github.com/golang/protobuf v1.3.2/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
@@ -404,6 +404,7 @@ github.com/hashicorp/hcl v1.0.0/go.mod h1:E5yfLk+7swimpb2L/Alb/PJmXilQ/rhwaUYs4T
 github.com/hashicorp/logutils v1.0.0/go.mod h1:QIAnNjmIWmVIIkWDTG1z5v++HQmx9WQRO+LraFDTW64=
 github.com/hashicorp/mdns v1.0.0/go.mod h1:tL+uN++7HEJ6SQLQ2/p+z2pH24WQKWjBPkE0mNTz8vQ=
 github.com/hashicorp/memberlist v0.1.3/go.mod h1:ajVTdAv/9Im8oMAAj5G31PhhMCZJV2pPBoIllUwCN7I=
+github.com/hashicorp/memberlist v0.2.4/go.mod h1:MS2lj3INKhZjWNqd3N0m3J+Jxf3DAOnAH9VT3Sh9MUE=
 github.com/hashicorp/serf v0.8.2/go.mod h1:6hOLApaqBFA1NXqRQAsxw9QxuDEvNxSQRwA/JwenrHc=
 github.com/hashicorp/yamux v0.0.0-20180604194846-3520598351bb/go.mod h1:+NfK9FKeTrX5uv1uIXGdwYDTeHna2qgaIlx54MXqjAM=
 github.com/hashicorp/yamux v0.0.0-20181012175058-2f1d1f20f75d h1:kJCB4vdITiW1eC1vq2e6IsrXKrZit1bv/TDYFGMp4BQ=
@@ -480,6 +481,7 @@ github.com/mdlayher/netlink v0.0.0-20190409211403-11939a169225/go.mod h1:eQB3mZE
 github.com/mdlayher/netlink v1.0.0/go.mod h1:KxeJAFOFLG6AjpyDkQ/iIhxygIUKD+vcwqcnu43w/+M=
 github.com/mdlayher/netlink v1.1.0/go.mod h1:H4WCitaheIsdF9yOYu8CFmCgQthAPIWZmcKp9uZHgmY=
 github.com/miekg/dns v1.0.14/go.mod h1:W1PPwlIAgtquWBMBEV9nkV9Cazfe8ScdGz/Lj7v3Nrg=
+github.com/miekg/dns v1.1.26/go.mod h1:bPDLeHnStXmXAq1m/Ch/hvfNHr14JKNPMBo3VZKjuso=
 github.com/mitchellh/cli v1.0.0/go.mod h1:hNIlj7HEI86fIcpObd7a0FcrxTWetlwJDGcceTlRvqc=
 github.com/mitchellh/go-homedir v1.0.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0=
 github.com/mitchellh/go-homedir v1.1.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0=
@@ -660,6 +662,7 @@ github.com/yuin/goldmark v1.1.25/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9de
 github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.1.32/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
+github.com/yuin/goldmark v1.3.5/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k=
 go.etcd.io/bbolt v1.3.2/go.mod h1:IbVyRI1SCnLcuJnV2u8VeU0CEYM7e686BmAb1XKL+uU=
 go.etcd.io/bbolt v1.3.3/go.mod h1:IbVyRI1SCnLcuJnV2u8VeU0CEYM7e686BmAb1XKL+uU=
 go.etcd.io/bbolt v1.3.5/go.mod h1:G5EMThwa9y8QZGBClrRx5EY+Yw9kAhnjy3bSjsnlVTQ=
@@ -699,6 +702,7 @@ golang.org/x/crypto v0.0.0-20190605123033-f99c8df09eb5/go.mod h1:yigFU9vqHzYiE8U
 golang.org/x/crypto v0.0.0-20190611184440-5c40567a22f8/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20190617133340-57b3e21c3d56/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20190820162420-60c769a6c586/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
+golang.org/x/crypto v0.0.0-20190923035154-9ee001bba392/go.mod h1:/lpIB1dKB+9EgE3H3cr1v9wB50oz8l4C4h62xy7jSTY=
 golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20191206172530-e9b2fee46413/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
@@ -738,8 +742,8 @@ golang.org/x/mod v0.1.1-0.20191107180719-034126e5016b/go.mod h1:QqPTAvyqsEbceGzB
 golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.3.1-0.20200828183125-ce943fd02449/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
-golang.org/x/mod v0.4.0 h1:8pl+sMODzuvGJkmj2W4kZihvVb5mKm8pB/X44PIQHv8=
-golang.org/x/mod v0.4.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
+golang.org/x/mod v0.4.2 h1:Gz96sIWK3OalVv/I/qNygP42zyoKp3xptRVCWRFEBvo=
+golang.org/x/mod v0.4.2/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
@@ -764,6 +768,7 @@ golang.org/x/net v0.0.0-20190628185345-da137c7871d7/go.mod h1:z5CRVTTTmAJ677TzLL
 golang.org/x/net v0.0.0-20190724013045-ca1201d0de80/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20190813141303-74dc4d7220e7/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20190827160401-ba9fcec4b297/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/net v0.0.0-20190923162816-aa69164e4478/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20191004110552-13f9640d40b9/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20191007182048-72f939374954/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20191209160850-c0dbc17a3553/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
@@ -784,8 +789,9 @@ golang.org/x/net v0.0.0-20200904194848-62affa334b73/go.mod h1:/O7V0waA8r7cgGh81R
 golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
 golang.org/x/net v0.0.0-20201024042810-be3efd7ff127/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
 golang.org/x/net v0.0.0-20201110031124-69a78807bb2b/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
-golang.org/x/net v0.0.0-20210224082022-3d97a244fca7 h1:OgUuv8lsRpBibGNbSizVwKWlysjaNzmC9gYMhPVfqFM=
 golang.org/x/net v0.0.0-20210224082022-3d97a244fca7/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
+golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4 h1:4nGaVu0QrbjT/AK2PRLuQfQuh6DJve+pELhqTdAj3x0=
+golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
 golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
@@ -801,8 +807,9 @@ golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20200317015054-43a5402ce75a/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20200625203802-6e8e738ad208/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9 h1:SQFwaSi55rU7vdNs9Yr0Z324VNlrF+0wMqRXT4St8ck=
 golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20210220032951-036812b2e83c h1:5KslGYwFpkhGh+Q16bwMP3cOontH8FOep7tGV86Y7SQ=
+golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20180823144017-11551d06cbcc/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
@@ -828,6 +835,8 @@ golang.org/x/sys v0.0.0-20190624142023-c5567b49c5d0/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20190726091711-fc99dfbffb4e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190826190057-c7b8b68b1456/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190916202348-b4ddaad3f8a3/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20190922100055-0a153f010e69/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20190924154521-2837fb4f24fe/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20191001151750-bb3f8db39f24/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20191002063906-3421d5a6bb1c/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20191005200804-aed5e4c7ecf9/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
@@ -858,8 +867,10 @@ golang.org/x/sys v0.0.0-20201024232916-9f70ab9862d5/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20201112073958-5cba982894dd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210119212857-b64e53b001e4/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20210225134936-a50acf3fe073 h1:8qxJSnu+7dRq6upnbntrmriWByIakBuct5OM/MdQC1M=
 golang.org/x/sys v0.0.0-20210225134936-a50acf3fe073/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210510120138-977fb7262007 h1:gG67DSER+11cZvqIMb8S8bt0vZtiN6xWYARwirrOSfE=
+golang.org/x/sys v0.0.0-20210510120138-977fb7262007/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1 h1:v+OssWQX+hTHEmOBgwxdZxK4zHq3yOs8F9J7mk0PY8E=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
@@ -898,6 +909,7 @@ golang.org/x/tools v0.0.0-20190621195816-6e04913cbbac/go.mod h1:/rFqwRUd4F7ZHNgw
 golang.org/x/tools v0.0.0-20190624222133-a101b041ded4/go.mod h1:/rFqwRUd4F7ZHNgwSSTFct+R/Kf4OFW1sUzUTQQTgfc=
 golang.org/x/tools v0.0.0-20190628153133-6cdbf07be9d0/go.mod h1:/rFqwRUd4F7ZHNgwSSTFct+R/Kf4OFW1sUzUTQQTgfc=
 golang.org/x/tools v0.0.0-20190816200558-6889da9d5479/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
+golang.org/x/tools v0.0.0-20190907020128-2ca718005c18/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20190911174233-4f2ddba30aff/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20191012152004-8de300cfc20a/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20191029041327-9cc4af7d6b2c/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
@@ -932,8 +944,9 @@ golang.org/x/tools v0.0.0-20200729194436-6467de6f59a7/go.mod h1:njjCfa9FT2d7l9Bc
 golang.org/x/tools v0.0.0-20200804011535-6c149bb5ef0d/go.mod h1:njjCfa9FT2d7l9Bc6FUM5FLjQPp3cFF28FI3qnDFljA=
 golang.org/x/tools v0.0.0-20200825202427-b303f430e36d/go.mod h1:njjCfa9FT2d7l9Bc6FUM5FLjQPp3cFF28FI3qnDFljA=
 golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
-golang.org/x/tools v0.1.0 h1:po9/4sTYwZU9lPhi1tOrb4hCv3qrhiQ77LZfGa2OjwY=
 golang.org/x/tools v0.1.0/go.mod h1:xkSsbof2nBLbhDlRMhhhyNLN/zl3eTqcnHD5viDpcZ0=
+golang.org/x/tools v0.1.1 h1:wGiQel/hW0NnEkJUk8lbzkX2gFJU6PFxf1v5OlCfuOs=
+golang.org/x/tools v0.1.1/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=


Test patch:
diff --git a/ci/jenkins/test.sh b/ci/jenkins/test.sh
index 25d4eba56..1805190ef 100755
--- a/ci/jenkins/test.sh
+++ b/ci/jenkins/test.sh
@@ -258,7 +258,10 @@ function deliver_antrea_windows {
     rm -f antrea-windows.tar.gz
     sed -i 's/if (!(Test-Path $AntreaAgentConfigPath))/if ($true)/' hack/windows/Helper.psm1
     kubectl get nodes -o wide --no-headers=true | awk -v role="$CONTROL_PLANE_NODE_ROLE" '$3 != role && $1 ~ /win/ {print $6}' | while read IP; do
+        WORKER_NAME=$(govc vm.info -vm.ip ${IP} -json | jq -r '.VirtualMachines[0].Name')
+        echo "==== Reverting Windows VM ${WORKER_NAME} ====="
         govc snapshot.revert -vm.ip ${IP} win-initial
+        govc vm.power -on ${WORKER_NAME} || true
         # Some tests need us.gcr.io/k8s-artifacts-prod/e2e-test-images/agnhost:2.13 image but it is not for windows/amd64 10.0.17763
         # Use e2eteam/agnhost:2.13 instead
         harbor_images=("sigwindowstools-kube-proxy:v1.18.0" "agnhost:2.13" "agnhost:2.13" "e2eteam-jessie-dnsutils:1.0" "e2eteam-pause:3.2")
@@ -513,8 +516,10 @@ function run_install_windows_ovs {
     export_govc_env_var
     OVS_VM_NAME="antrea-microsoft-ovs"
 
-    IP=$(govc vm.ip $OVS_VM_NAME)
     govc snapshot.revert -vm $OVS_VM_NAME initial
+    govc vm.power -on $OVS_VM_NAME || true
+    echo "===== Testing VM has been reverted and powered on ====="
+    IP=$(govc vm.ip $OVS_VM_NAME)
     scp -o StrictHostKeyChecking=no -i ${WORKDIR}/.ssh/id_rsa -T hack/windows/Install-OVS.ps1 Administrator@${IP}:
     ssh -o StrictHostKeyChecking=no -i ${WORKDIR}/.ssh/id_rsa -n Administrator@${IP} '/bin/bash -lc "cp Install-OVS.ps1 C:/k && powershell.exe -File C:/k/Install-OVS.ps1"'
 
diff --git a/pkg/agent/controller/egress/egress_controller_test.go b/pkg/agent/controller/egress/egress_controller_test.go
index 832cd74e1..4da787dfe 100644
--- a/pkg/agent/controller/egress/egress_controller_test.go
+++ b/pkg/agent/controller/egress/egress_controller_test.go
@@ -30,6 +30,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/util/workqueue"
 
+	ipassignertest "antrea.io/antrea/pkg/agent/controller/egress/ipassigner/testing"
 	"antrea.io/antrea/pkg/agent/interfacestore"
 	openflowtest "antrea.io/antrea/pkg/agent/openflow/testing"
 	routetest "antrea.io/antrea/pkg/agent/route/testing"
@@ -87,6 +88,7 @@ type fakeController struct {
 	mockRouteClient    *routetest.MockInterface
 	crdClient          *fakeversioned.Clientset
 	crdInformerFactory crdinformers.SharedInformerFactory
+	mockIPAssigner     *ipassignertest.MockIPAssigner
 }
 
 func newFakeController(t *testing.T, initObjects []runtime.Object) *fakeController {
@@ -94,6 +96,7 @@ func newFakeController(t *testing.T, initObjects []runtime.Object) *fakeControll
 
 	mockOFClient := openflowtest.NewMockClient(controller)
 	mockRouteClient := routetest.NewMockInterface(controller)
+	mockIPAssigner := ipassignertest.NewMockIPAssigner(controller)
 
 	clientset := &fake.Clientset{}
 	crdClient := fakeversioned.NewSimpleClientset(initObjects...)
@@ -124,6 +127,7 @@ func newFakeController(t *testing.T, initObjects []runtime.Object) *fakeControll
 		egressBindings:       map[string]*egressBinding{},
 		egressStates:         map[string]*egressState{},
 		egressIPStates:       map[string]*egressIPState{},
+		ipAssigner:           mockIPAssigner,
 	}
 	return &fakeController{
 		EgressController:   egressController,
@@ -132,8 +136,10 @@ func newFakeController(t *testing.T, initObjects []runtime.Object) *fakeControll
 		mockRouteClient:    mockRouteClient,
 		crdClient:          crdClient,
 		crdInformerFactory: crdInformerFactory,
+		mockIPAssigner:     mockIPAssigner,
 	}
 }
+
 func TestSyncEgress(t *testing.T) {
 	tests := []struct {
 		name                string
@@ -142,7 +148,7 @@ func TestSyncEgress(t *testing.T) {
 		existingEgressGroup *cpv1b2.EgressGroup
 		newEgressGroup      *cpv1b2.EgressGroup
 		newLocalIPs         sets.String
-		expectedCalls       func(mockOFClient *openflowtest.MockClient, mockRouteClient *routetest.MockInterface)
+		expectedCalls       func(mockOFClient *openflowtest.MockClient, mockRouteClient *routetest.MockInterface, mockIPAssigner *ipassignertest.MockIPAssigner)
 	}{
 		{
 			name: "Local IP becomes non local",
@@ -169,7 +175,7 @@ func TestSyncEgress(t *testing.T) {
 				},
 			},
 			newLocalIPs: sets.NewString(),
-			expectedCalls: func(mockOFClient *openflowtest.MockClient, mockRouteClient *routetest.MockInterface) {
+			expectedCalls: func(mockOFClient *openflowtest.MockClient, mockRouteClient *routetest.MockInterface, mockIPAssigner *ipassignertest.MockIPAssigner) {
 				mockOFClient.EXPECT().InstallSNATMarkFlows(net.ParseIP(fakeLocalEgressIP1), uint32(1))
 				mockOFClient.EXPECT().InstallPodSNATFlows(uint32(1), net.ParseIP(fakeLocalEgressIP1), uint32(1))
 				mockOFClient.EXPECT().InstallPodSNATFlows(uint32(2), net.ParseIP(fakeLocalEgressIP1), uint32(1))
@@ -182,6 +188,9 @@ func TestSyncEgress(t *testing.T) {
 
 				mockOFClient.EXPECT().InstallPodSNATFlows(uint32(1), net.ParseIP(fakeLocalEgressIP1), uint32(0))
 				mockOFClient.EXPECT().InstallPodSNATFlows(uint32(3), net.ParseIP(fakeLocalEgressIP1), uint32(0))
+				mockIPAssigner.EXPECT().UnassignEgressIP("egressA")
+				mockIPAssigner.EXPECT().UnassignEgressIP("egressA")
+				mockIPAssigner.EXPECT().UnassignEgressIP("egressA")
 			},
 		},
 		{
@@ -209,7 +218,7 @@ func TestSyncEgress(t *testing.T) {
 				},
 			},
 			newLocalIPs: sets.NewString(fakeRemoteEgressIP1),
-			expectedCalls: func(mockOFClient *openflowtest.MockClient, mockRouteClient *routetest.MockInterface) {
+			expectedCalls: func(mockOFClient *openflowtest.MockClient, mockRouteClient *routetest.MockInterface, mockIPAssigner *ipassignertest.MockIPAssigner) {
 				mockOFClient.EXPECT().InstallPodSNATFlows(uint32(1), net.ParseIP(fakeRemoteEgressIP1), uint32(0))
 				mockOFClient.EXPECT().InstallPodSNATFlows(uint32(2), net.ParseIP(fakeRemoteEgressIP1), uint32(0))
 
@@ -220,6 +229,9 @@ func TestSyncEgress(t *testing.T) {
 				mockOFClient.EXPECT().InstallPodSNATFlows(uint32(1), net.ParseIP(fakeRemoteEgressIP1), uint32(1))
 				mockOFClient.EXPECT().InstallPodSNATFlows(uint32(3), net.ParseIP(fakeRemoteEgressIP1), uint32(1))
 				mockRouteClient.EXPECT().AddSNATRule(net.ParseIP(fakeRemoteEgressIP1), uint32(1))
+				mockIPAssigner.EXPECT().UnassignEgressIP("egressA")
+				mockIPAssigner.EXPECT().UnassignEgressIP("egressA")
+				mockIPAssigner.EXPECT().UnassignEgressIP("egressA")
 			},
 		},
 		{
@@ -246,7 +258,7 @@ func TestSyncEgress(t *testing.T) {
 					{Pod: &cpv1b2.PodReference{Name: "pod3", Namespace: "ns3"}},
 				},
 			},
-			expectedCalls: func(mockOFClient *openflowtest.MockClient, mockRouteClient *routetest.MockInterface) {
+			expectedCalls: func(mockOFClient *openflowtest.MockClient, mockRouteClient *routetest.MockInterface, mockIPAssigner *ipassignertest.MockIPAssigner) {
 				mockOFClient.EXPECT().InstallSNATMarkFlows(net.ParseIP(fakeLocalEgressIP1), uint32(1))
 				mockOFClient.EXPECT().InstallPodSNATFlows(uint32(1), net.ParseIP(fakeLocalEgressIP1), uint32(1))
 				mockOFClient.EXPECT().InstallPodSNATFlows(uint32(2), net.ParseIP(fakeLocalEgressIP1), uint32(1))
@@ -261,6 +273,9 @@ func TestSyncEgress(t *testing.T) {
 				mockOFClient.EXPECT().InstallPodSNATFlows(uint32(1), net.ParseIP(fakeLocalEgressIP2), uint32(1))
 				mockOFClient.EXPECT().InstallPodSNATFlows(uint32(3), net.ParseIP(fakeLocalEgressIP2), uint32(1))
 				mockRouteClient.EXPECT().AddSNATRule(net.ParseIP(fakeLocalEgressIP2), uint32(1))
+				mockIPAssigner.EXPECT().UnassignEgressIP("egressA")
+				mockIPAssigner.EXPECT().UnassignEgressIP("egressA")
+				mockIPAssigner.EXPECT().UnassignEgressIP("egressA")
 			},
 		},
 		{
@@ -287,7 +302,7 @@ func TestSyncEgress(t *testing.T) {
 					{Pod: &cpv1b2.PodReference{Name: "pod3", Namespace: "ns3"}},
 				},
 			},
-			expectedCalls: func(mockOFClient *openflowtest.MockClient, mockRouteClient *routetest.MockInterface) {
+			expectedCalls: func(mockOFClient *openflowtest.MockClient, mockRouteClient *routetest.MockInterface, mockIPAssigner *ipassignertest.MockIPAssigner) {
 				mockOFClient.EXPECT().InstallSNATMarkFlows(net.ParseIP(fakeLocalEgressIP1), uint32(1))
 				mockOFClient.EXPECT().InstallPodSNATFlows(uint32(1), net.ParseIP(fakeLocalEgressIP1), uint32(1))
 				mockOFClient.EXPECT().InstallPodSNATFlows(uint32(2), net.ParseIP(fakeLocalEgressIP1), uint32(1))
@@ -300,6 +315,9 @@ func TestSyncEgress(t *testing.T) {
 
 				mockOFClient.EXPECT().InstallPodSNATFlows(uint32(1), net.ParseIP(fakeRemoteEgressIP1), uint32(0))
 				mockOFClient.EXPECT().InstallPodSNATFlows(uint32(3), net.ParseIP(fakeRemoteEgressIP1), uint32(0))
+				mockIPAssigner.EXPECT().UnassignEgressIP("egressA")
+				mockIPAssigner.EXPECT().UnassignEgressIP("egressA")
+				mockIPAssigner.EXPECT().UnassignEgressIP("egressA")
 			},
 		},
 		{
@@ -326,7 +344,7 @@ func TestSyncEgress(t *testing.T) {
 					{Pod: &cpv1b2.PodReference{Name: "pod3", Namespace: "ns3"}},
 				},
 			},
-			expectedCalls: func(mockOFClient *openflowtest.MockClient, mockRouteClient *routetest.MockInterface) {
+			expectedCalls: func(mockOFClient *openflowtest.MockClient, mockRouteClient *routetest.MockInterface, mockIPAssigner *ipassignertest.MockIPAssigner) {
 				mockOFClient.EXPECT().InstallPodSNATFlows(uint32(1), net.ParseIP(fakeRemoteEgressIP1), uint32(0))
 				mockOFClient.EXPECT().InstallPodSNATFlows(uint32(2), net.ParseIP(fakeRemoteEgressIP1), uint32(0))
 
@@ -337,6 +355,9 @@ func TestSyncEgress(t *testing.T) {
 				mockOFClient.EXPECT().InstallPodSNATFlows(uint32(1), net.ParseIP(fakeLocalEgressIP1), uint32(1))
 				mockOFClient.EXPECT().InstallPodSNATFlows(uint32(3), net.ParseIP(fakeLocalEgressIP1), uint32(1))
 				mockRouteClient.EXPECT().AddSNATRule(net.ParseIP(fakeLocalEgressIP1), uint32(1))
+				mockIPAssigner.EXPECT().UnassignEgressIP("egressA")
+				mockIPAssigner.EXPECT().UnassignEgressIP("egressA")
+				mockIPAssigner.EXPECT().UnassignEgressIP("egressA")
 			},
 		},
 		{
@@ -363,7 +384,7 @@ func TestSyncEgress(t *testing.T) {
 					{Pod: &cpv1b2.PodReference{Name: "pod3", Namespace: "ns3"}},
 				},
 			},
-			expectedCalls: func(mockOFClient *openflowtest.MockClient, mockRouteClient *routetest.MockInterface) {
+			expectedCalls: func(mockOFClient *openflowtest.MockClient, mockRouteClient *routetest.MockInterface, mockIPAssigner *ipassignertest.MockIPAssigner) {
 				mockOFClient.EXPECT().InstallSNATMarkFlows(net.ParseIP(fakeLocalEgressIP1), uint32(1))
 				mockOFClient.EXPECT().InstallPodSNATFlows(uint32(1), net.ParseIP(fakeLocalEgressIP1), uint32(1))
 				mockOFClient.EXPECT().InstallPodSNATFlows(uint32(2), net.ParseIP(fakeLocalEgressIP1), uint32(1))
@@ -372,6 +393,9 @@ func TestSyncEgress(t *testing.T) {
 				mockOFClient.EXPECT().InstallSNATMarkFlows(net.ParseIP(fakeLocalEgressIP2), uint32(2))
 				mockOFClient.EXPECT().InstallPodSNATFlows(uint32(3), net.ParseIP(fakeLocalEgressIP2), uint32(2))
 				mockRouteClient.EXPECT().AddSNATRule(net.ParseIP(fakeLocalEgressIP2), uint32(2))
+				mockIPAssigner.EXPECT().UnassignEgressIP("egressA")
+				mockIPAssigner.EXPECT().UnassignEgressIP("egressB")
+				mockIPAssigner.EXPECT().UnassignEgressIP("egressB")
 			},
 		},
 		{
@@ -398,13 +422,15 @@ func TestSyncEgress(t *testing.T) {
 					{Pod: &cpv1b2.PodReference{Name: "pod3", Namespace: "ns3"}},
 				},
 			},
-			expectedCalls: func(mockOFClient *openflowtest.MockClient, mockRouteClient *routetest.MockInterface) {
+			expectedCalls: func(mockOFClient *openflowtest.MockClient, mockRouteClient *routetest.MockInterface, mockIPAssigner *ipassignertest.MockIPAssigner) {
 				mockOFClient.EXPECT().InstallSNATMarkFlows(net.ParseIP(fakeLocalEgressIP1), uint32(1))
 				mockOFClient.EXPECT().InstallPodSNATFlows(uint32(1), net.ParseIP(fakeLocalEgressIP1), uint32(1))
 				mockOFClient.EXPECT().InstallPodSNATFlows(uint32(2), net.ParseIP(fakeLocalEgressIP1), uint32(1))
 				mockRouteClient.EXPECT().AddSNATRule(net.ParseIP(fakeLocalEgressIP1), uint32(1))
-
 				mockOFClient.EXPECT().InstallPodSNATFlows(uint32(3), net.ParseIP(fakeLocalEgressIP1), uint32(1))
+				mockIPAssigner.EXPECT().UnassignEgressIP("egressA")
+				mockIPAssigner.EXPECT().UnassignEgressIP("egressB")
+				mockIPAssigner.EXPECT().UnassignEgressIP("egressB")
 			},
 		},
 	}
@@ -419,7 +445,7 @@ func TestSyncEgress(t *testing.T) {
 			c.crdInformerFactory.WaitForCacheSync(stopCh)
 			c.addEgressGroup(tt.existingEgressGroup)
 
-			tt.expectedCalls(c.mockOFClient, c.mockRouteClient)
+			tt.expectedCalls(c.mockOFClient, c.mockRouteClient, c.mockIPAssigner)
 			err := c.syncEgress(tt.existingEgress.Name)
 			assert.NoError(t, err)
 
@@ -504,16 +530,19 @@ func TestSyncOverlappingEgress(t *testing.T) {
 	c.mockOFClient.EXPECT().InstallPodSNATFlows(uint32(1), net.ParseIP(fakeLocalEgressIP1), uint32(1))
 	c.mockOFClient.EXPECT().InstallPodSNATFlows(uint32(2), net.ParseIP(fakeLocalEgressIP1), uint32(1))
 	c.mockRouteClient.EXPECT().AddSNATRule(net.ParseIP(fakeLocalEgressIP1), uint32(1))
+	c.mockIPAssigner.EXPECT().UnassignEgressIP(egress1.Name)
 	err := c.syncEgress(egress1.Name)
 	assert.NoError(t, err)
 
 	// egress2's IP is not local and pod1 has enforced egress1, so only one Pod SNAT flow is expected.
 	c.mockOFClient.EXPECT().InstallPodSNATFlows(uint32(3), net.ParseIP(fakeRemoteEgressIP1), uint32(0))
+	c.mockIPAssigner.EXPECT().UnassignEgressIP(egress2.Name)
 	err = c.syncEgress(egress2.Name)
 	assert.NoError(t, err)
 
 	// egress3 shares the same IP as egress1 and pod2 has enforced egress1, so only one Pod SNAT flow is expected.
 	c.mockOFClient.EXPECT().InstallPodSNATFlows(uint32(4), net.ParseIP(fakeLocalEgressIP1), uint32(1))
+	c.mockIPAssigner.EXPECT().UnassignEgressIP(egress3.Name)
 	err = c.syncEgress(egress3.Name)
 	assert.NoError(t, err)
 
@@ -527,6 +556,7 @@ func TestSyncOverlappingEgress(t *testing.T) {
 		_, err := c.egressLister.Get(egress1.Name)
 		return err != nil, nil
 	}))
+	c.mockIPAssigner.EXPECT().UnassignEgressIP(egress1.Name)
 	err = c.syncEgress(egress1.Name)
 	assert.NoError(t, err)
 	require.Equal(t, 2, c.queue.Len())
@@ -541,11 +571,13 @@ func TestSyncOverlappingEgress(t *testing.T) {
 
 	// pod1 is expected to enforce egress2.
 	c.mockOFClient.EXPECT().InstallPodSNATFlows(uint32(1), net.ParseIP(fakeRemoteEgressIP1), uint32(0))
+	c.mockIPAssigner.EXPECT().UnassignEgressIP(egress2.Name)
 	err = c.syncEgress(egress2.Name)
 	assert.NoError(t, err)
 
 	// pod2 is expected to enforce egress3.
 	c.mockOFClient.EXPECT().InstallPodSNATFlows(uint32(2), net.ParseIP(fakeLocalEgressIP1), uint32(1))
+	c.mockIPAssigner.EXPECT().UnassignEgressIP(egress3.Name)
 	err = c.syncEgress(egress3.Name)
 	assert.NoError(t, err)
 
@@ -553,6 +585,7 @@ func TestSyncOverlappingEgress(t *testing.T) {
 	c.mockOFClient.EXPECT().UninstallPodSNATFlows(uint32(1))
 	c.mockOFClient.EXPECT().UninstallPodSNATFlows(uint32(3))
 	c.crdClient.CrdV1alpha2().Egresses().Delete(context.TODO(), egress2.Name, metav1.DeleteOptions{})
+	c.mockIPAssigner.EXPECT().UnassignEgressIP(egress2.Name)
 	assert.NoError(t, wait.Poll(time.Millisecond*100, time.Second, func() (bool, error) {
 		_, err := c.egressLister.Get(egress2.Name)
 		return err != nil, nil
@@ -567,6 +600,7 @@ func TestSyncOverlappingEgress(t *testing.T) {
 	c.mockOFClient.EXPECT().UninstallPodSNATFlows(uint32(2))
 	c.mockOFClient.EXPECT().UninstallPodSNATFlows(uint32(4))
 	c.crdClient.CrdV1alpha2().Egresses().Delete(context.TODO(), egress3.Name, metav1.DeleteOptions{})
+	c.mockIPAssigner.EXPECT().UnassignEgressIP(egress3.Name)
 	assert.NoError(t, wait.Poll(time.Millisecond*100, time.Second, func() (bool, error) {
 		_, err := c.egressLister.Get(egress3.Name)
 		return err != nil, nil
diff --git a/pkg/agent/controller/egress/ipassigner/testing/mock_ipassigner.go b/pkg/agent/controller/egress/ipassigner/testing/mock_ipassigner.go
new file mode 100644
index 000000000..4ab48d29f
--- /dev/null
+++ b/pkg/agent/controller/egress/ipassigner/testing/mock_ipassigner.go
@@ -0,0 +1,90 @@
+// 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.
+//
+
+// Code generated by MockGen. DO NOT EDIT.
+// Source: antrea.io/antrea/pkg/agent/controller/egress/ipassigner (interfaces: IPAssigner)
+
+// Package testing is a generated GoMock package.
+package testing
+
+import (
+	gomock "github.com/golang/mock/gomock"
+	reflect "reflect"
+)
+
+// MockIPAssigner is a mock of IPAssigner interface
+type MockIPAssigner struct {
+	ctrl     *gomock.Controller
+	recorder *MockIPAssignerMockRecorder
+}
+
+// MockIPAssignerMockRecorder is the mock recorder for MockIPAssigner
+type MockIPAssignerMockRecorder struct {
+	mock *MockIPAssigner
+}
+
+// NewMockIPAssigner creates a new mock instance
+func NewMockIPAssigner(ctrl *gomock.Controller) *MockIPAssigner {
+	mock := &MockIPAssigner{ctrl: ctrl}
+	mock.recorder = &MockIPAssignerMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use
+func (m *MockIPAssigner) EXPECT() *MockIPAssignerMockRecorder {
+	return m.recorder
+}
+
+// AssignEgressIP mocks base method
+func (m *MockIPAssigner) AssignEgressIP(arg0, arg1 string) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "AssignEgressIP", arg0, arg1)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// AssignEgressIP indicates an expected call of AssignEgressIP
+func (mr *MockIPAssignerMockRecorder) AssignEgressIP(arg0, arg1 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AssignEgressIP", reflect.TypeOf((*MockIPAssigner)(nil).AssignEgressIP), arg0, arg1)
+}
+
+// AssignedIPs mocks base method
+func (m *MockIPAssigner) AssignedIPs() map[string]string {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "AssignedIPs")
+	ret0, _ := ret[0].(map[string]string)
+	return ret0
+}
+
+// AssignedIPs indicates an expected call of AssignedIPs
+func (mr *MockIPAssignerMockRecorder) AssignedIPs() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AssignedIPs", reflect.TypeOf((*MockIPAssigner)(nil).AssignedIPs))
+}
+
+// UnassignEgressIP mocks base method
+func (m *MockIPAssigner) UnassignEgressIP(arg0 string) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "UnassignEgressIP", arg0)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// UnassignEgressIP indicates an expected call of UnassignEgressIP
+func (mr *MockIPAssignerMockRecorder) UnassignEgressIP(arg0 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UnassignEgressIP", reflect.TypeOf((*MockIPAssigner)(nil).UnassignEgressIP), arg0)
+}
diff --git a/pkg/agent/flowexporter/connections/conntrack_connections_test.go b/pkg/agent/flowexporter/connections/conntrack_connections_test.go
index c64e925c9..d33998d82 100644
--- a/pkg/agent/flowexporter/connections/conntrack_connections_test.go
+++ b/pkg/agent/flowexporter/connections/conntrack_connections_test.go
@@ -137,6 +137,8 @@ func TestConntrackConnectionStore_AddOrUpdateConn(t *testing.T) {
 	// 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{
+		StartTime: refTime.Add(-(time.Second * 50)),
+		StopTime:  refTime,
 		FlowKey:   tuple4,
 		Mark:      openflow.ServiceCTMark,
 		IsPresent: true,
diff --git a/pkg/agent/memberlist/cluster_test.go b/pkg/agent/memberlist/cluster_test.go
new file mode 100644
index 000000000..f08d5688e
--- /dev/null
+++ b/pkg/agent/memberlist/cluster_test.go
@@ -0,0 +1,581 @@
+// 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 memberlist
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"reflect"
+	"testing"
+	"time"
+
+	"github.com/golang/groupcache/consistenthash"
+	"github.com/hashicorp/memberlist"
+	"github.com/stretchr/testify/assert"
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/informers"
+	"k8s.io/client-go/kubernetes/fake"
+	"k8s.io/client-go/tools/cache"
+
+	"antrea.io/antrea/pkg/agent/config"
+	"antrea.io/antrea/pkg/apis"
+	crdv1a2 "antrea.io/antrea/pkg/apis/crd/v1alpha2"
+	fakeversioned "antrea.io/antrea/pkg/client/clientset/versioned/fake"
+	crdinformers "antrea.io/antrea/pkg/client/informers/externalversions"
+)
+
+type fakeCluster struct {
+	cluster   *Cluster
+	clientSet *fake.Clientset
+	crdClient *fakeversioned.Clientset
+}
+
+func newFakeCluster(nodeConfig *config.NodeConfig, stopCh <-chan struct{}, i int) (*fakeCluster, error) {
+	port := apis.AntreaAgentClusterMembershipPort + i
+
+	clientset := fake.NewSimpleClientset()
+	informerFactory := informers.NewSharedInformerFactory(clientset, 0)
+
+	nodeInformer := informerFactory.Core().V1().Nodes()
+
+	crdClient := fakeversioned.NewSimpleClientset([]runtime.Object{}...)
+	crdInformerFactory := crdinformers.NewSharedInformerFactory(crdClient, 0)
+	ipPoolInformer := crdInformerFactory.Crd().V1alpha2().ExternalIPPools()
+
+	cluster, err := NewCluster(port, nodeConfig.NodeIPAddr.IP, nodeConfig.Name, nodeInformer, ipPoolInformer)
+	if err != nil {
+		return nil, err
+	}
+
+	// Make sure informers are running.
+	informerFactory.Start(stopCh)
+	crdInformerFactory.Start(stopCh)
+
+	cache.WaitForCacheSync(stopCh, nodeInformer.Informer().HasSynced)
+	cache.WaitForCacheSync(stopCh, ipPoolInformer.Informer().HasSynced)
+	return &fakeCluster{
+		cluster:   cluster,
+		clientSet: clientset,
+		crdClient: crdClient,
+	}, nil
+}
+
+func createNode(cs *fake.Clientset, node *v1.Node) error {
+	_, err := cs.CoreV1().Nodes().Create(context.TODO(), node, metav1.CreateOptions{})
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func createExternalIPPool(crdClient *fakeversioned.Clientset, eip *crdv1a2.ExternalIPPool) error {
+	_, err := crdClient.CrdV1alpha2().ExternalIPPools().Create(context.TODO(), eip, metav1.CreateOptions{})
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func TestCluster_Run(t *testing.T) {
+	localNodeName := "localNodeName"
+	testCases := []struct {
+		name                     string
+		egress                   *crdv1a2.Egress
+		externalIPPool           *crdv1a2.ExternalIPPool
+		localNode                *v1.Node
+		expectEgressSelectResult bool
+	}{
+		{
+			name: "Local Node matches ExternalIPPool nodeSelectors",
+			egress: &crdv1a2.Egress{
+				Spec: crdv1a2.EgressSpec{ExternalIPPool: "", EgressIP: "1.1.1.1"},
+			},
+			externalIPPool: &crdv1a2.ExternalIPPool{
+				TypeMeta:   metav1.TypeMeta{Kind: "CustomResourceDefinition"},
+				ObjectMeta: metav1.ObjectMeta{Name: "fakeExternalIPPool"},
+				Spec:       crdv1a2.ExternalIPPoolSpec{NodeSelector: metav1.LabelSelector{MatchLabels: map[string]string{"env": "pro"}}},
+			},
+			localNode: &v1.Node{
+				ObjectMeta: metav1.ObjectMeta{Name: localNodeName, Labels: map[string]string{"env": "pro"}},
+				Status:     v1.NodeStatus{Addresses: []v1.NodeAddress{{Type: v1.NodeInternalIP, Address: "127.0.0.1"}}},
+			},
+			expectEgressSelectResult: true,
+		},
+		{
+			name: "Local Node not match ExternalIPPool nodeSelectors",
+			egress: &crdv1a2.Egress{
+				Spec: crdv1a2.EgressSpec{ExternalIPPool: "", EgressIP: "1.1.1.1"},
+			},
+			externalIPPool: &crdv1a2.ExternalIPPool{
+				TypeMeta:   metav1.TypeMeta{Kind: "CustomResourceDefinition"},
+				ObjectMeta: metav1.ObjectMeta{Name: "fakeExternalIPPool1"},
+				Spec:       crdv1a2.ExternalIPPoolSpec{NodeSelector: metav1.LabelSelector{MatchLabels: map[string]string{"env": "pro"}}},
+			},
+			localNode: &v1.Node{
+				ObjectMeta: metav1.ObjectMeta{Name: localNodeName},
+				Status:     v1.NodeStatus{Addresses: []v1.NodeAddress{{Type: v1.NodeInternalIP, Address: "127.0.0.1"}}},
+			},
+			expectEgressSelectResult: false,
+		},
+	}
+	for i, tCase := range testCases {
+		t.Run(tCase.name, func(t *testing.T) {
+			nodeConfig := &config.NodeConfig{
+				Name:       localNodeName,
+				NodeIPAddr: &net.IPNet{IP: net.IPv4(127, 0, 0, 1), Mask: net.IPv4Mask(255, 255, 255, 255)},
+			}
+			stopCh := make(chan struct{})
+			defer close(stopCh)
+			fakeCluster, err := newFakeCluster(nodeConfig, stopCh, i)
+			if err != nil {
+				t.Fatalf("New fake memberlist server error: %v", err)
+			}
+
+			eip := tCase.externalIPPool
+			assert.NoError(t, createExternalIPPool(fakeCluster.crdClient, eip))
+			assert.NoError(t, createNode(fakeCluster.clientSet, tCase.localNode))
+
+			go fakeCluster.cluster.Run(stopCh)
+
+			assert.NoError(t, wait.Poll(time.Millisecond*100, time.Second, func() (done bool, err error) {
+				newEIP, _ := fakeCluster.cluster.externalIPPoolLister.Get(eip.Name)
+				return reflect.DeepEqual(newEIP, eip), nil
+			}))
+
+			tCase.egress.Spec.ExternalIPPool = eip.Name
+			res, err := fakeCluster.cluster.ShouldSelectEgress(tCase.egress)
+			// Cluster should hold the same consistent hash ring for each ExternalIPPool.
+			assert.NoError(t, err)
+			allMembers, err := fakeCluster.cluster.allClusterMembers()
+			assert.NoError(t, err)
+			assert.Equal(t, 1, len(allMembers), "expected Node member num is 1")
+			assert.Equal(t, 1, fakeCluster.cluster.mList.NumMembers(), "expected alive Node num is 1")
+			assert.Equal(t, tCase.expectEgressSelectResult, res, "select Node for Egress result not match")
+		})
+	}
+}
+
+func TestCluster_RunClusterEvents(t *testing.T) {
+	stopCh := make(chan struct{})
+	defer close(stopCh)
+
+	nodeName := "localNodeName"
+	nodeConfig := &config.NodeConfig{
+		Name:       nodeName,
+		NodeIPAddr: &net.IPNet{IP: net.IPv4(127, 0, 0, 1)},
+	}
+	localNode := &v1.Node{
+		ObjectMeta: metav1.ObjectMeta{Name: nodeName},
+		Status:     v1.NodeStatus{Addresses: []v1.NodeAddress{{Type: v1.NodeInternalIP, Address: "127.0.0.1"}}}}
+	fakeEIP1 := &crdv1a2.ExternalIPPool{
+		TypeMeta:   metav1.TypeMeta{Kind: "CustomResourceDefinition"},
+		ObjectMeta: metav1.ObjectMeta{Name: "fakeExternalIPPool1"},
+		Spec:       crdv1a2.ExternalIPPoolSpec{NodeSelector: metav1.LabelSelector{MatchLabels: map[string]string{"env": "pro"}}},
+	}
+	fakeEgress1 := &crdv1a2.Egress{
+		ObjectMeta: metav1.ObjectMeta{Name: "fakeEgress1", UID: "fakeUID1"},
+		Spec:       crdv1a2.EgressSpec{ExternalIPPool: fakeEIP1.Name, EgressIP: "1.1.1.2"},
+	}
+
+	fakeCluster, err := newFakeCluster(nodeConfig, stopCh, 10)
+	if err != nil {
+		t.Fatalf("New fake memberlist server error: %v", err)
+	}
+	// Test Cluster AddClusterEventHandler.
+	fakeCluster.cluster.AddClusterEventHandler(func(objName string) {
+		t.Logf("Detected cluster Node event, running fake handler, obj: %s", objName)
+	})
+
+	// Create local Node and ExternalIPPool.
+	assert.NoError(t, createNode(fakeCluster.clientSet, localNode))
+	assert.NoError(t, createExternalIPPool(fakeCluster.crdClient, fakeEIP1))
+
+	go fakeCluster.cluster.Run(stopCh)
+
+	assert.NoError(t, wait.Poll(time.Millisecond*100, time.Second, func() (done bool, err error) {
+		newEIP, _ := fakeCluster.cluster.externalIPPoolLister.Get(fakeEIP1.Name)
+		return reflect.DeepEqual(newEIP, fakeEIP1), nil
+	}))
+
+	// Test updating Node labels.
+	testCasesUpdateNode := []struct {
+		name                     string
+		expectEgressSelectResult bool
+		newNodeLabels            map[string]string
+		egress                   *crdv1a2.Egress
+	}{
+		{
+			name:                     "Update Node with the same labels then local Node should not be selected",
+			expectEgressSelectResult: false,
+			newNodeLabels:            localNode.Labels,
+			egress:                   fakeEgress1,
+		},
+		{
+			name:                     "Update Node with matched labels then local Node should be selected",
+			expectEgressSelectResult: true,
+			newNodeLabels:            map[string]string{"env": "pro"},
+			egress:                   fakeEgress1,
+		},
+		{
+			name:                     "Update Node with different but matched labels then local Node should be selected",
+			expectEgressSelectResult: true,
+			newNodeLabels:            map[string]string{"env": "pro", "env1": "test"},
+			egress:                   fakeEgress1,
+		},
+		{
+			name:                     "Update Node with not matched labels then local Node should not be selected",
+			expectEgressSelectResult: false,
+			newNodeLabels:            map[string]string{"env": "test"},
+			egress:                   fakeEgress1,
+		},
+	}
+	updateNode := func(node *v1.Node) {
+		_, err = fakeCluster.clientSet.CoreV1().Nodes().Update(context.TODO(), node, metav1.UpdateOptions{})
+		if err != nil {
+			t.Fatalf("Update Node error: %v", err)
+		}
+		assert.NoError(t, wait.Poll(time.Millisecond*100, time.Second, func() (done bool, err error) {
+			newNode, _ := fakeCluster.cluster.nodeLister.Get(node.Name)
+			return reflect.DeepEqual(node, newNode), nil
+		}))
+	}
+	for _, tCase := range testCasesUpdateNode {
+		t.Run(tCase.name, func(t *testing.T) {
+			localNode.Labels = tCase.newNodeLabels
+			updateNode(localNode)
+			res, err := fakeCluster.cluster.ShouldSelectEgress(tCase.egress)
+			assert.NoError(t, err)
+			assert.Equal(t, tCase.expectEgressSelectResult, res, "select Node for Egress result not match")
+		})
+	}
+
+	// Test updating ExternalIPPool.
+	localNode.Labels = map[string]string{"env": "test"}
+	updateNode(localNode)
+	testCasesUpdateEIP := []struct {
+		name                     string
+		expectEgressSelectResult bool
+		newEIPnodeSelectors      metav1.LabelSelector
+	}{
+		{
+			name:                     "Update ExternalIPPool with the same nodeSelector then local Node should not be selected",
+			expectEgressSelectResult: false,
+			newEIPnodeSelectors:      fakeEIP1.Spec.NodeSelector,
+		},
+		{
+			name:                     "Update ExternalIPPool with the matched nodeSelector then local Node should be selected",
+			expectEgressSelectResult: true,
+			newEIPnodeSelectors:      metav1.LabelSelector{MatchLabels: map[string]string{"env": "test"}},
+		},
+		{
+			name:                     "Update ExternalIPPool with nil nodeSelector then local Node should be selected",
+			expectEgressSelectResult: true,
+			newEIPnodeSelectors:      metav1.LabelSelector{},
+		},
+		{
+			name:                     "Update ExternalIPPool refresh back then local Node should not be selected",
+			expectEgressSelectResult: false,
+			newEIPnodeSelectors:      fakeEIP1.Spec.NodeSelector,
+		},
+	}
+	for _, tCase := range testCasesUpdateEIP {
+		t.Run(tCase.name, func(t *testing.T) {
+			fakeEIP1.Spec.NodeSelector = tCase.newEIPnodeSelectors
+			_, err := fakeCluster.crdClient.CrdV1alpha2().ExternalIPPools().Update(context.TODO(), fakeEIP1, metav1.UpdateOptions{})
+			if err != nil {
+				t.Fatalf("Update ExternalIPPool error: %v", err)
+			}
+			assert.NoError(t, wait.Poll(time.Millisecond*100, time.Second, func() (done bool, err error) {
+				newEIP, _ := fakeCluster.cluster.externalIPPoolLister.Get(fakeEIP1.Name)
+				return reflect.DeepEqual(fakeEIP1, newEIP), nil
+			}))
+			res, err := fakeCluster.cluster.ShouldSelectEgress(fakeEgress1)
+			assert.NoError(t, err)
+			assert.Equal(t, tCase.expectEgressSelectResult, res, "select Node for Egress result not match")
+		})
+	}
+
+	// Test creating new ExternalIPPool.
+	fakeEIP2 := &crdv1a2.ExternalIPPool{
+		TypeMeta: metav1.TypeMeta{
+			Kind: "CustomResourceDefinition",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "fakeExternalIPPool2",
+		},
+		Spec: crdv1a2.ExternalIPPoolSpec{NodeSelector: metav1.LabelSelector{MatchLabels: map[string]string{"env": "test"}}},
+	}
+	fakeEgressIP2 := "1.1.1.2"
+	fakeEgress2 := &crdv1a2.Egress{
+		ObjectMeta: metav1.ObjectMeta{Name: "fakeEgress2", UID: "fakeUID2"},
+		Spec:       crdv1a2.EgressSpec{ExternalIPPool: fakeEIP2.Name, EgressIP: fakeEgressIP2},
+	}
+	assert.NoError(t, createExternalIPPool(fakeCluster.crdClient, fakeEIP2))
+	assert.NoError(t, wait.Poll(time.Millisecond*100, time.Second, func() (done bool, err error) {
+		newEIP, _ := fakeCluster.cluster.externalIPPoolLister.Get(fakeEIP2.Name)
+		return reflect.DeepEqual(newEIP, fakeEIP2), nil
+	}))
+	assertEgressSelectResult := func(egress *crdv1a2.Egress, expectedRes bool, hasSyncedErr bool) {
+		res, err := fakeCluster.cluster.ShouldSelectEgress(egress)
+		if !hasSyncedErr {
+			assert.NoError(t, err)
+		}
+		assert.Equal(t, expectedRes, res, "select Node for Egress result not match")
+	}
+	assertEgressSelectResult(fakeEgress2, true, false)
+	assertEgressSelectResult(fakeEgress1, false, false)
+
+	// Test deleting ExternalIPPool.
+	deleteExternalIPPool := func(eipName string) {
+		err := fakeCluster.crdClient.CrdV1alpha2().ExternalIPPools().Delete(context.TODO(), eipName, metav1.DeleteOptions{})
+		if err != nil {
+			t.Fatalf("Delete ExternalIPPool error: %v", err)
+		}
+		assert.NoError(t, wait.Poll(time.Millisecond*100, time.Second, func() (done bool, err error) {
+			newEIP, _ := fakeCluster.cluster.externalIPPoolLister.Get(eipName)
+			return nil == newEIP, nil
+		}))
+	}
+	deleteExternalIPPool(fakeEIP2.Name)
+	assertEgressSelectResult(fakeEgress2, false, true)
+	assertEgressSelectResult(fakeEgress1, false, false)
+
+	// Test Node update event.
+	fakeCluster.cluster.nodeEventsCh <- memberlist.NodeEvent{Node: &memberlist.Node{Name: "fakeNodeNameUpdate"}, Event: memberlist.NodeUpdate}
+	assertEgressSelectResult(fakeEgress2, false, true)
+	assertEgressSelectResult(fakeEgress1, false, false)
+
+	// Test Node leave event.
+	fakeCluster.cluster.nodeEventsCh <- memberlist.NodeEvent{Node: &memberlist.Node{Name: localNode.Name}, Event: memberlist.NodeLeave}
+	assertEgressSelectResult(fakeEgress2, false, true)
+	assertEgressSelectResult(fakeEgress1, false, false)
+
+	// Test Node leave event, Node not found.
+	fakeCluster.cluster.nodeEventsCh <- memberlist.NodeEvent{Node: &memberlist.Node{Name: "fakeNodeNameLeave"}, Event: memberlist.NodeLeave}
+	assertEgressSelectResult(fakeEgress2, false, true)
+	assertEgressSelectResult(fakeEgress1, false, false)
+
+	// Test Node join event.
+	fakeCluster.cluster.nodeEventsCh <- memberlist.NodeEvent{Node: &memberlist.Node{Name: localNode.Name}, Event: memberlist.NodeJoin}
+	assertEgressSelectResult(fakeEgress2, false, true)
+	assertEgressSelectResult(fakeEgress1, false, false)
+
+	// Test creating Node with invalid IP.
+	fakeNode := &v1.Node{
+		ObjectMeta: metav1.ObjectMeta{Name: "fakeNode0"},
+		Status:     v1.NodeStatus{Addresses: []v1.NodeAddress{{Type: v1.NodeInternalIP, Address: "x"}}},
+	}
+	assert.NoError(t, createNode(fakeCluster.clientSet, fakeNode))
+	assertEgressSelectResult(fakeEgress2, false, true)
+	assertEgressSelectResult(fakeEgress1, false, false)
+
+	// Test deleting Node.
+	deleteNode := func(node *v1.Node) {
+		err := fakeCluster.clientSet.CoreV1().Nodes().Delete(context.TODO(), node.Name, metav1.DeleteOptions{})
+		if err != nil {
+			t.Fatalf("Delete Node error: %v", err)
+		}
+		assert.NoError(t, wait.Poll(time.Millisecond*100, time.Second, func() (done bool, err error) {
+			newNode, _ := fakeCluster.cluster.nodeLister.Get(node.Name)
+			return nil == newNode, nil
+		}))
+	}
+	deleteNode(localNode)
+	assertEgressSelectResult(fakeEgress2, false, true)
+	assertEgressSelectResult(fakeEgress1, false, false)
+
+	// Test creating Node with valid IP.
+	fakeNode1 := &v1.Node{
+		ObjectMeta: metav1.ObjectMeta{Name: "fakeNode1"},
+		Status:     v1.NodeStatus{Addresses: []v1.NodeAddress{{Type: v1.NodeInternalIP, Address: "1.1.1.1"}}},
+	}
+	assert.NoError(t, createNode(fakeCluster.clientSet, fakeNode1))
+	assertEgressSelectResult(fakeEgress2, false, true)
+	assertEgressSelectResult(fakeEgress1, false, false)
+}
+
+func genLocalNodeCluster(localNodeNme, eipName string, nodes []string) *Cluster {
+	cluster := &Cluster{
+		nodeName:          localNodeNme,
+		consistentHashMap: make(map[string]*consistenthash.Map),
+	}
+	cluster.consistentHashMap[eipName] = newNodeConsistentHashMap()
+	cluster.consistentHashMap[eipName].Add(nodes...)
+	return cluster
+}
+
+func genNodes(n int) []string {
+	nodes := make([]string, n)
+	for i := 0; i < n; i++ {
+		nodes[i] = fmt.Sprintf("node-%d", i)
+	}
+	return nodes
+}
+
+// TestCluster_ConsistentHashDistribute test the distributions of Egresses in Nodes
+func TestCluster_ConsistentHashDistribute(t *testing.T) {
+	egressNum := 10
+	testCases := []struct {
+		name                  string
+		nodes                 []string
+		expectedDistributions map[string][]int
+	}{
+		{
+			name:                  fmt.Sprintf("Assign owner Node for %d Egress", egressNum),
+			nodes:                 []string{"node0", "node1", "node2"},
+			expectedDistributions: map[string][]int{"node0": {1, 4, 9}, "node1": {0, 2, 5, 8}, "node2": {3, 6, 7}},
+		},
+		{
+			// Failover, when Node failed, Egress should move to available Node.
+			name:                  "A Node fail then Egress should move",
+			nodes:                 []string{"node1", "node2"},
+			expectedDistributions: map[string][]int{"node1": {0, 2, 4, 5, 8, 9}, "node2": {1, 3, 6, 7}},
+		},
+		{
+			// Egress should move when Node added in cluster.
+			name:                  "Add new Node then Egress should move",
+			nodes:                 []string{"node0", "node1", "node2", "node3"},
+			expectedDistributions: map[string][]int{"node0": {1, 4, 9}, "node1": {0, 5, 8}, "node2": {3, 6}, "node3": {2, 7}},
+		},
+	}
+	for _, testC := range testCases {
+		t.Run(testC.name, func(t *testing.T) {
+			actualDistribute := map[string][]int{}
+			for _, node := range testC.nodes {
+				fakeEIPName := "fakeExternalIPPool"
+				fakeCluster := genLocalNodeCluster(node, fakeEIPName, testC.nodes)
+				selectedNodes := []int{}
+				for i := 0; i < egressNum; i++ {
+					fakeEgress := &crdv1a2.Egress{
+						ObjectMeta: metav1.ObjectMeta{Name: "fakeEgress"},
+						Spec:       crdv1a2.EgressSpec{ExternalIPPool: fakeEIPName, EgressIP: fmt.Sprintf("10.1.1.%d", i)},
+					}
+					selected, err := fakeCluster.ShouldSelectEgress(fakeEgress)
+					assert.NoError(t, err)
+					if selected {
+						selectedNodes = append(selectedNodes, i)
+					}
+				}
+				actualDistribute[node] = selectedNodes
+				t.Logf("Distributions of Egresses in Node %s: %#v", node, selectedNodes)
+			}
+			assert.Equal(t, testC.expectedDistributions, actualDistribute, "Egress distributions not match")
+		})
+	}
+}
+
+func TestCluster_ShouldSelectEgress(t *testing.T) {
+	testCases := []struct {
+		name         string
+		nodeNum      int
+		egressIP     string
+		expectedNode string
+	}{
+		{
+			name:         "Select Node from 0 Nodes",
+			nodeNum:      0,
+			egressIP:     "1.1.1.1",
+			expectedNode: "",
+		},
+		{
+			name:         "Select Node from 1 Nodes",
+			nodeNum:      1,
+			egressIP:     "1.1.1.1",
+			expectedNode: "node-0",
+		},
+		{
+			name:         "Select Node from 3 Nodes",
+			nodeNum:      3,
+			egressIP:     "1.1.1.1",
+			expectedNode: "node-1",
+		},
+		{
+			name:         "Select Node from 10 Nodes",
+			nodeNum:      10,
+			egressIP:     "1.1.1.1",
+			expectedNode: "node-1",
+		},
+		{
+			name:         "Select Node from 100 Nodes",
+			nodeNum:      100,
+			egressIP:     "1.1.1.1",
+			expectedNode: "node-79",
+		},
+	}
+	for _, tCase := range testCases {
+		t.Run(tCase.name, func(t *testing.T) {
+			fakeEIPName := "fakeExternalIPPool"
+			fakeEgress := &crdv1a2.Egress{
+				ObjectMeta: metav1.ObjectMeta{Name: "fakeEgress"},
+				Spec:       crdv1a2.EgressSpec{ExternalIPPool: fakeEIPName, EgressIP: tCase.egressIP},
+			}
+			consistentHashMap := newNodeConsistentHashMap()
+			consistentHashMap.Add(genNodes(tCase.nodeNum)...)
+
+			fakeCluster := &Cluster{
+				consistentHashMap: map[string]*consistenthash.Map{fakeEIPName: consistentHashMap},
+			}
+
+			for i := 0; i < tCase.nodeNum; i++ {
+				node := fmt.Sprintf("node-%d", i)
+				fakeCluster.nodeName = node
+				selected, err := fakeCluster.ShouldSelectEgress(fakeEgress)
+				assert.NoError(t, err)
+				assert.Equal(t, node == tCase.expectedNode, selected, "Selected Node for Egress not match")
+			}
+		})
+	}
+}
+
+// BenchmarkCluster_ShouldSelect
+// BenchmarkCluster_ShouldSelect/Select_Node_from_10000_alive_nodes-nodeSelectedForEgress
+// BenchmarkCluster_ShouldSelect/Select_Node_from_10000_alive_nodes-nodeSelectedForEgress-16         	 9190818	       128 ns/op
+// BenchmarkCluster_ShouldSelect/Select_Node_from_1000_alive_nodes-nodeSelectedForEgress
+// BenchmarkCluster_ShouldSelect/Select_Node_from_1000_alive_nodes-nodeSelectedForEgress-16          	 9474440	       125 ns/op
+// PASS
+func BenchmarkCluster_ShouldSelect(b *testing.B) {
+	benchmarkCases := []struct {
+		name  string
+		nodes []string
+	}{
+		{
+			name:  "Select Node from 10000 alive Nodes",
+			nodes: genNodes(10000),
+		},
+		{
+			name:  "Select Node from 1000 alive Nodes",
+			nodes: genNodes(1000),
+		},
+	}
+
+	for _, bc := range benchmarkCases {
+		fakeEIPName := "fakeExternalIPPool"
+		fakeEgress := &crdv1a2.Egress{
+			ObjectMeta: metav1.ObjectMeta{Name: "fakeEgress"},
+			Spec:       crdv1a2.EgressSpec{ExternalIPPool: fakeEIPName, EgressIP: "1.1.1.1"},
+		}
+		fakeCluster := genLocalNodeCluster("fakeLocalNodeName", fakeEIPName, bc.nodes)
+		b.Run(fmt.Sprintf("%s-nodeSelectedForEgress", bc.name), func(b *testing.B) {
+			b.ResetTimer()
+			for i := 0; i < b.N; i++ {
+				fakeCluster.ShouldSelectEgress(fakeEgress)
+			}
+		})
+	}
+}
diff --git a/pkg/apiserver/handlers/featuregates/handler_test.go b/pkg/apiserver/handlers/featuregates/handler_test.go
index 3b480dfc5..bd8ee76e9 100644
--- a/pkg/apiserver/handlers/featuregates/handler_test.go
+++ b/pkg/apiserver/handlers/featuregates/handler_test.go
@@ -51,7 +51,7 @@ func Test_getGatesResponse(t *testing.T) {
 				{Component: "agent", Name: "EndpointSlice", Status: "Disabled", Version: "ALPHA"},
 				{Component: "agent", Name: "Traceflow", Status: "Enabled", Version: "BETA"},
 				{Component: "agent", Name: "FlowExporter", Status: "Disabled", Version: "ALPHA"},
-				{Component: "agent", Name: "NetworkPolicyStats", Status: "Disabled", Version: "ALPHA"},
+				{Component: "agent", Name: "NetworkPolicyStats", Status: "Enabled", Version: "BETA"},
 				{Component: "agent", Name: "NodePortLocal", Status: "Disabled", Version: "ALPHA"},
 			},
 		},
@@ -136,14 +136,14 @@ func TestHandleFunc(t *testing.T) {
 				{Component: "controller", Name: "AntreaPolicy", Status: "Enabled", Version: "BETA"},
 				{Component: "controller", Name: "Egress", Status: "Disabled", Version: "ALPHA"},
 				{Component: "controller", Name: "Traceflow", Status: "Enabled", Version: "BETA"},
-				{Component: "controller", Name: "NetworkPolicyStats", Status: "Disabled", Version: "ALPHA"},
+				{Component: "controller", Name: "NetworkPolicyStats", Status: "Enabled", Version: "BETA"},
 				{Component: "agent", Name: "AntreaPolicy", Status: "Enabled", Version: "BETA"},
 				{Component: "agent", Name: "AntreaProxy", Status: "Enabled", Version: "BETA"},
 				{Component: "agent", Name: "Egress", Status: "Disabled", Version: "ALPHA"},
 				{Component: "agent", Name: "EndpointSlice", Status: "Disabled", Version: "ALPHA"},
 				{Component: "agent", Name: "Traceflow", Status: "Enabled", Version: "BETA"},
 				{Component: "agent", Name: "FlowExporter", Status: "Disabled", Version: "ALPHA"},
-				{Component: "agent", Name: "NetworkPolicyStats", Status: "Disabled", Version: "ALPHA"},
+				{Component: "agent", Name: "NetworkPolicyStats", Status: "Enabled", Version: "BETA"},
 				{Component: "agent", Name: "NodePortLocal", Status: "Disabled", Version: "ALPHA"},
 			},
 		},
@@ -187,7 +187,7 @@ func Test_getControllerGatesResponse(t *testing.T) {
 				{Component: "controller", Name: "AntreaPolicy", Status: "Enabled", Version: "BETA"},
 				{Component: "controller", Name: "Egress", Status: "Disabled", Version: "ALPHA"},
 				{Component: "controller", Name: "Traceflow", Status: "Enabled", Version: "BETA"},
-				{Component: "controller", Name: "NetworkPolicyStats", Status: "Disabled", Version: "ALPHA"},
+				{Component: "controller", Name: "NetworkPolicyStats", Status: "Enabled", Version: "BETA"},
 			},
 		},
 	}
diff --git a/test/e2e/antreapolicy_test.go b/test/e2e/antreapolicy_test.go
index e51260da4..dd8baac80 100644
--- a/test/e2e/antreapolicy_test.go
+++ b/test/e2e/antreapolicy_test.go
@@ -155,11 +155,7 @@ func initialize(t *testing.T, data *TestData) {
 }
 
 func skipIfAntreaPolicyDisabled(tb testing.TB, data *TestData) {
-	if featureGate, err := data.GetControllerFeatures(antreaNamespace); err != nil {
-		tb.Fatalf("Cannot determine if ACNP enabled: %v", err)
-	} else if !featureGate.Enabled(features.AntreaPolicy) {
-		tb.Skipf("Skipping test as it required ACNP to be enabled")
-	}
+	skipIfFeatureDisabled(tb, data, features.AntreaPolicy, true, true)
 }
 
 func applyDefaultDenyToAllNamespaces(k8s *KubernetesUtils, namespaces []string) error {
@@ -2843,16 +2839,7 @@ func TestANPNetworkPolicyStatsWithDropAction(t *testing.T) {
 	}
 	defer teardownTest(t, data)
 	skipIfAntreaPolicyDisabled(t, data)
-
-	cc := []configChange{
-		{"NetworkPolicyStats", "true", true},
-	}
-	ac := []configChange{
-		{"NetworkPolicyStats", "true", true},
-	}
-	if err := testData.mutateAntreaConfigMap(cc, ac, true, true); err != nil {
-		t.Fatalf("Failed to enable NetworkPolicyStats feature: %v", err)
-	}
+	skipIfNetworkPolicyStatsDisabled(t, data)
 
 	serverName, serverIPs, cleanupFunc := createAndWaitForPod(t, data, data.createNginxPodOnNode, "test-server-", "")
 	defer cleanupFunc()
@@ -2996,16 +2983,8 @@ func TestAntreaClusterNetworkPolicyStats(t *testing.T) {
 	}
 	defer teardownTest(t, data)
 	skipIfAntreaPolicyDisabled(t, data)
+	skipIfNetworkPolicyStatsDisabled(t, data)
 
-	cc := []configChange{
-		{"NetworkPolicyStats", "true", true},
-	}
-	ac := []configChange{
-		{"NetworkPolicyStats", "true", true},
-	}
-	if err := testData.mutateAntreaConfigMap(cc, ac, true, true); err != nil {
-		t.Fatalf("Failed to enable NetworkPolicyStats feature: %v", err)
-	}
 	serverName, serverIPs, cleanupFunc := createAndWaitForPod(t, data, data.createNginxPodOnNode, "test-server-", "")
 	defer cleanupFunc()
 
diff --git a/test/e2e/fixtures.go b/test/e2e/fixtures.go
index 156ab1f3b..dbf0cec1d 100644
--- a/test/e2e/fixtures.go
+++ b/test/e2e/fixtures.go
@@ -23,6 +23,8 @@ import (
 	"testing"
 	"time"
 
+	"k8s.io/component-base/featuregate"
+
 	"antrea.io/antrea/pkg/agent/config"
 )
 
@@ -105,6 +107,23 @@ func skipIfNoWindowsNodes(tb testing.TB) {
 	}
 }
 
+func skipIfFeatureDisabled(tb testing.TB, data *TestData, feature featuregate.Feature, checkAgent bool, checkController bool) {
+	if checkAgent {
+		if featureGate, err := data.GetAgentFeatures(antreaNamespace); err != nil {
+			tb.Fatalf("Cannot determine if %s is enabled in the Agent: %v", feature, err)
+		} else if !featureGate.Enabled(feature) {
+			tb.Skipf("Skipping test because %s is not enabled in the Agent", feature)
+		}
+	}
+	if checkController {
+		if featureGate, err := data.GetControllerFeatures(antreaNamespace); err != nil {
+			tb.Fatalf("Cannot determine if %s is enabled in the Controller: %v", feature, err)
+		} else if !featureGate.Enabled(feature) {
+			tb.Skipf("Skipping test because %s is not enabled in the Controller", feature)
+		}
+	}
+}
+
 func ensureAntreaRunning(tb testing.TB, data *TestData) error {
 	tb.Logf("Applying Antrea YAML")
 	if err := data.deployAntrea(); err != nil {
diff --git a/test/e2e/flowaggregator_test.go b/test/e2e/flowaggregator_test.go
index 2922bc938..1705b2813 100644
--- a/test/e2e/flowaggregator_test.go
+++ b/test/e2e/flowaggregator_test.go
@@ -119,6 +119,9 @@ const (
 	testIngressRuleName            = "test-ingress-rule-name"
 	testEgressRuleName             = "test-egress-rule-name"
 	iperfTimeSec                   = 12
+)
+
+var (
 	// Single iperf run results in two connections with separate ports (control connection and actual data connection).
 	// 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.
@@ -144,6 +147,21 @@ func TestFlowAggregator(t *testing.T) {
 	defer teardownTest(t, data)
 	defer teardownFlowAggregator(t, data)
 
+	if testOptions.providerName == "kind" {
+		// Currently, in Kind clusters, OVS userspace datapath does not support
+		// packet statistics in the conntrack entries. Because of that Flow Exporter
+		// at Antrea agent cannot consider flows to be active and keep sending active
+		// records. Currently, Flow Exporter sends two records for a iperf flow
+		// in kind cluster with a duration of 12s: 1. A new iperf connection gets
+		// idled out after exporter idle timeout, which is after 1s in the test.
+		// In this case, flow aggregator sends the record after 4.5s 2. When the
+		// connection dies and TCP state becomes TIME_WAIT, which is
+		// at 12s in the test. Here, Flow Aggregator sends the record at 15.5s.
+		// We will remove this workaround once OVS userspace datapath supports packet
+		// statistics in conntrack entries.
+		expectedNumDataRecords = 2
+	}
+
 	k8sUtils, err = NewKubernetesUtils(data)
 	if err != nil {
 		t.Fatalf("Error when creating Kubernetes utils client: %v", err)
@@ -487,16 +505,15 @@ func checkRecordsForFlows(t *testing.T, data *TestData, srcIP string, dstIP stri
 	timeStartSec := timeStart.Unix()
 	var cmdStr string
 	if !isIPv6 {
-		cmdStr = fmt.Sprintf("iperf3 -c %s -t %d|grep sender|awk '{print $7,$8}'", dstIP, iperfTimeSec)
+		cmdStr = fmt.Sprintf("iperf3 -c %s -t %d", dstIP, iperfTimeSec)
 	} else {
-		cmdStr = fmt.Sprintf("iperf3 -6 -c %s -t %d|grep sender|awk '{print $7,$8}'", dstIP, iperfTimeSec)
+		cmdStr = fmt.Sprintf("iperf3 -6 -c %s -t %d", 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, " ")
+	bwSlice, srcPort := getBandwidthAndSourcePort(stdout)
 	// 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])
@@ -509,13 +526,15 @@ func checkRecordsForFlows(t *testing.T, data *TestData, srcIP string, dstIP stri
 		t.Fatalf("Unit of the traffic bandwidth reported by iperf should either be Mbits or Gbits, failing the test.")
 	}
 
-	collectorOutput := getCollectorOutput(t, srcIP, dstIP, timeStart, true)
+	collectorOutput, recordSlices := getCollectorOutput(t, srcIP, dstIP, srcPort, timeStart, true)
 	// 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) {
+		// Check the source port along with source and destination IPs as there
+		// are flow records for control flows during the iperf  with same IPs
+		// and destination port.
+		if strings.Contains(record, srcIP) && strings.Contains(record, dstIP) && strings.Contains(record, srcPort) {
 			dataRecordsCount = dataRecordsCount + 1
 			// Check if record has both Pod name of source and destination Pod.
 			if isIntraNode {
@@ -581,7 +600,7 @@ func checkRecordsForFlows(t *testing.T, data *TestData, srcIP string, dstIP stri
 	}
 	// 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))
+	assert.GreaterOrEqualf(t, dataRecordsCount, expectedNumDataRecords, "IPFIX collector should receive expected number of flow records. Considered records: %s \n Collector output: %s", recordSlices, collectorOutput)
 }
 
 func checkRecordsForToExternalFlows(t *testing.T, data *TestData, srcNodeName string, srcPodName string, srcIP string, dstIP string, dstPort int32, isIPv6 bool) {
@@ -595,8 +614,7 @@ func checkRecordsForToExternalFlows(t *testing.T, data *TestData, srcNodeName st
 	stdout, stderr, err := data.runCommandFromPod(testNamespace, srcPodName, busyboxContainerName, strings.Fields(cmd))
 	require.NoErrorf(t, err, "Error when running wget command, stdout: %s, stderr: %s", stdout, stderr)
 
-	collectorOutput := getCollectorOutput(t, srcIP, dstIP, timeStart, false)
-	recordSlices := getRecordsFromOutput(collectorOutput)
+	_, recordSlices := getCollectorOutput(t, srcIP, dstIP, "", timeStart, false)
 	for _, record := range recordSlices {
 		if strings.Contains(record, srcIP) && strings.Contains(record, dstIP) {
 			checkPodAndNodeData(t, record, srcPodName, srcNodeName, "", "")
@@ -625,9 +643,8 @@ func checkRecordsForDenyFlows(t *testing.T, data *TestData, testFlow1, testFlow2
 	_, _, err = data.runCommandFromPod(testNamespace, testFlow2.srcPodName, "", []string{"timeout", "2", "bash", "-c", cmdStr2})
 	assert.Error(t, err)
 
-	collectorOutput := getCollectorOutput(t, testFlow1.srcIP, testFlow2.srcIP, timeStart, false)
+	_, recordSlices := getCollectorOutput(t, testFlow1.srcIP, testFlow2.srcIP, "", timeStart, false)
 	// 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) {
@@ -730,22 +747,27 @@ func getUnit64FieldFromRecord(t *testing.T, record string, field string) uint64
 	return 0
 }
 
-func getCollectorOutput(t *testing.T, srcIP string, dstIP string, timeStart time.Time, checkAllRecords bool) string {
+// getCollectorOutput polls the output of go-ipfix collector and checks if we have
+// received all the expected records for a given flow with source IP, destination IP
+// and source port. We send source port to ignore the control flows during the
+// iperf test.
+func getCollectorOutput(t *testing.T, srcIP, dstIP, srcPort string, timeStart time.Time, checkAllRecords bool) (string, []string) {
 	var collectorOutput string
+	var recordSlices []string
 	err := wait.PollImmediate(500*time.Millisecond, aggregatorInactiveFlowRecordTimeout, func() (bool, error) {
 		var rc int
 		var err error
 		// `pod-running-timeout` option is added to cover scenarios where ipfix flow-collector has crashed after being deployed
-		rc, collectorOutput, _, err = provider.RunCommandOnNode(controlPlaneNodeName(), fmt.Sprintf("kubectl logs --since=%v --pod-running-timeout=%v ipfix-collector -n antrea-test", time.Since(timeStart).String(), aggregatorInactiveFlowRecordTimeout.String()))
+		rc, collectorOutput, _, err = provider.RunCommandOnNode(controlPlaneNodeName(), fmt.Sprintf("kubectl logs --pod-running-timeout=%v ipfix-collector -n antrea-test", aggregatorInactiveFlowRecordTimeout.String()))
 		if err != nil || rc != 0 {
 			return false, err
 		}
 		// Checking that all the data records which correspond to the iperf flow are received
+		recordSlices = getRecordsFromOutput(collectorOutput)
 		if checkAllRecords {
-			recordSlices := getRecordsFromOutput(collectorOutput)
 			for _, record := range recordSlices {
 				exportTime := int64(getUnit64FieldFromRecord(t, record, "flowEndSeconds"))
-				if strings.Contains(record, srcIP) && strings.Contains(record, dstIP) {
+				if strings.Contains(record, srcIP) && strings.Contains(record, dstIP) && strings.Contains(record, srcPort) {
 					if exportTime >= timeStart.Unix()+iperfTimeSec {
 						return true, nil
 					}
@@ -753,11 +775,11 @@ func getCollectorOutput(t *testing.T, srcIP string, dstIP string, timeStart time
 			}
 			return false, nil
 		} else {
-			return strings.Contains(collectorOutput, srcIP) && strings.Contains(collectorOutput, dstIP), nil
+			return strings.Contains(collectorOutput, srcIP) && strings.Contains(collectorOutput, dstIP) && strings.Contains(collectorOutput, srcPort), nil
 		}
 	})
-	require.NoErrorf(t, err, "IPFIX collector did not receive the expected records and timed out with error")
-	return collectorOutput
+	require.NoErrorf(t, err, "IPFIX collector did not receive the expected records in collector output: %v time start: %s iperf source port: %s", collectorOutput, timeStart.String(), srcPort)
+	return collectorOutput, recordSlices
 }
 
 func getRecordsFromOutput(output string) []string {
@@ -765,10 +787,6 @@ func getRecordsFromOutput(output string) []string {
 	output = re.ReplaceAllString(output, "")
 	output = strings.TrimSpace(output)
 	recordSlices := strings.Split(output, "IPFIX-HDR:")
-	// Delete the first element from recordSlices
-	recordSlices[0] = recordSlices[len(recordSlices)-1]
-	recordSlices[len(recordSlices)-1] = ""
-	recordSlices = recordSlices[:len(recordSlices)-1]
 	return recordSlices
 }
 
@@ -1001,3 +1019,23 @@ func deletePerftestServices(t *testing.T, data *TestData) {
 		}
 	}
 }
+
+// getBandwidthAndSourcePort parses iperf commands output and returns bandwidth
+// and source port. Bandwidth is returned as a slice containing two strings (bandwidth
+// value and bandwidth unit).
+func getBandwidthAndSourcePort(iperfStdout string) ([]string, string) {
+	var bandwidth []string
+	var srcPort string
+	outputLines := strings.Split(iperfStdout, "\n")
+	for _, line := range outputLines {
+		if strings.Contains(line, "sender") {
+			fields := strings.Fields(line)
+			bandwidth = fields[6:8]
+		}
+		if strings.Contains(line, "connected") {
+			fields := strings.Fields(line)
+			srcPort = fields[5]
+		}
+	}
+	return bandwidth, srcPort
+}
diff --git a/test/e2e/framework.go b/test/e2e/framework.go
index 29b419554..73fd70776 100644
--- a/test/e2e/framework.go
+++ b/test/e2e/framework.go
@@ -107,7 +107,7 @@ const (
 	nginxLBService = "nginx-loadbalancer"
 
 	exporterActiveFlowExportTimeout     = 2 * time.Second
-	exporterInactiveFlowExportTimeout   = 1 * time.Second
+	exporterIdleFlowExportTimeout       = 1 * time.Second
 	aggregatorActiveFlowRecordTimeout   = 3500 * time.Millisecond
 	aggregatorInactiveFlowRecordTimeout = 6 * time.Second
 )
@@ -571,7 +571,7 @@ func (data *TestData) deployAntreaFlowExporter(ipfixCollector string) error {
 		{"FlowExporter", "true", true},
 		{"flowPollInterval", "\"1s\"", false},
 		{"activeFlowExportTimeout", fmt.Sprintf("\"%v\"", exporterActiveFlowExportTimeout), false},
-		{"inactiveFlowExportTimeout", fmt.Sprintf("\"%v\"", exporterInactiveFlowExportTimeout), false},
+		{"idleFlowExportTimeout", fmt.Sprintf("\"%v\"", exporterIdleFlowExportTimeout), false},
 	}
 	if ipfixCollector != "" {
 		ac = append(ac, configChange{"flowCollectorAddr", fmt.Sprintf("\"%s\"", ipfixCollector), false})
diff --git a/test/e2e/legacyantreapolicy_test.go b/test/e2e/legacyantreapolicy_test.go
index 6e30c925a..903843f2f 100644
--- a/test/e2e/legacyantreapolicy_test.go
+++ b/test/e2e/legacyantreapolicy_test.go
@@ -2310,16 +2310,7 @@ func TestLegacyANPNetworkPolicyStatsWithDropAction(t *testing.T) {
 	}
 	defer teardownTest(t, data)
 	skipIfAntreaPolicyDisabled(t, data)
-
-	cc := []configChange{
-		{"NetworkPolicyStats", "true", true},
-	}
-	ac := []configChange{
-		{"NetworkPolicyStats", "true", true},
-	}
-	if err := testData.mutateAntreaConfigMap(cc, ac, true, true); err != nil {
-		t.Fatalf("Failed to enable NetworkPolicyStats feature: %v", err)
-	}
+	skipIfNetworkPolicyStatsDisabled(t, data)
 
 	serverName, serverIPs, cleanupFunc := createAndWaitForPod(t, data, data.createNginxPodOnNode, "test-server-", "")
 	defer cleanupFunc()
@@ -2463,16 +2454,8 @@ func TestLegacyAntreaClusterNetworkPolicyStats(t *testing.T) {
 	}
 	defer teardownTest(t, data)
 	skipIfAntreaPolicyDisabled(t, data)
+	skipIfNetworkPolicyStatsDisabled(t, data)
 
-	cc := []configChange{
-		{"NetworkPolicyStats", "true", true},
-	}
-	ac := []configChange{
-		{"NetworkPolicyStats", "true", true},
-	}
-	if err := testData.mutateAntreaConfigMap(cc, ac, true, true); err != nil {
-		t.Fatalf("Failed to enable NetworkPolicyStats feature: %v", err)
-	}
 	serverName, serverIPs, cleanupFunc := createAndWaitForPod(t, data, data.createNginxPodOnNode, "test-server-", "")
 	defer cleanupFunc()
 
diff --git a/test/e2e/networkpolicy_test.go b/test/e2e/networkpolicy_test.go
index aa3d3e03b..24e8b3415 100644
--- a/test/e2e/networkpolicy_test.go
+++ b/test/e2e/networkpolicy_test.go
@@ -32,8 +32,13 @@ import (
 	"antrea.io/antrea/pkg/agent/apiserver/handlers/agentinfo"
 	"antrea.io/antrea/pkg/apis/crd/v1beta1"
 	"antrea.io/antrea/pkg/apis/stats/v1alpha1"
+	"antrea.io/antrea/pkg/features"
 )
 
+func skipIfNetworkPolicyStatsDisabled(tb testing.TB, data *TestData) {
+	skipIfFeatureDisabled(tb, data, features.NetworkPolicyStats, true, true)
+}
+
 func TestNetworkPolicyStats(t *testing.T) {
 	skipIfNotIPv4Cluster(t)
 	skipIfHasWindowsNodes(t)
@@ -43,16 +48,7 @@ func TestNetworkPolicyStats(t *testing.T) {
 		t.Fatalf("Error when setting up test: %v", err)
 	}
 	defer teardownTest(t, data)
-
-	cc := []configChange{
-		{"NetworkPolicyStats", "true", true},
-	}
-	ac := []configChange{
-		{"NetworkPolicyStats", "true", true},
-	}
-	if err := data.mutateAntreaConfigMap(cc, ac, true, true); err != nil {
-		t.Fatalf("Failed to enable NetworkPolicyStats feature: %v", err)
-	}
+	skipIfNetworkPolicyStatsDisabled(t, data)
 
 	serverName, serverIPs, cleanupFunc := createAndWaitForPod(t, data, data.createNginxPodOnNode, "test-server-", "")
 	defer cleanupFunc()
diff --git a/test/e2e/proxy_test.go b/test/e2e/proxy_test.go
index 863dab462..f547b912b 100644
--- a/test/e2e/proxy_test.go
+++ b/test/e2e/proxy_test.go
@@ -35,11 +35,7 @@ type expectTableFlows struct {
 }
 
 func skipIfProxyDisabled(t *testing.T, data *TestData) {
-	if featureGate, err := data.GetAgentFeatures(antreaNamespace); err != nil {
-		t.Fatalf("Error when detecting proxy: %v", err)
-	} else if !featureGate.Enabled(features.AntreaProxy) {
-		t.Skip("Skipping test because AntreaProxy is not enabled")
-	}
+	skipIfFeatureDisabled(t, data, features.AntreaProxy, true /* checkAgent */, false /* checkController */)
 }
 
 func TestProxyServiceSessionAffinity(t *testing.T) {
diff --git a/test/e2e/traceflow_test.go b/test/e2e/traceflow_test.go
index 6b1623301..5077c3633 100644
--- a/test/e2e/traceflow_test.go
+++ b/test/e2e/traceflow_test.go
@@ -50,16 +50,7 @@ type testcase struct {
 }
 
 func skipIfTraceflowDisabled(t *testing.T, data *TestData) {
-	if featureGate, err := data.GetAgentFeatures(antreaNamespace); err != nil {
-		t.Fatalf("Error when detecting traceflow: %v", err)
-	} else if !featureGate.Enabled(features.AntreaProxy) {
-		t.Skip("Skipping test because Traceflow is not enabled in the Agent")
-	}
-	if featureGate, err := data.GetControllerFeatures(antreaNamespace); err != nil {
-		t.Fatalf("Error when detecting traceflow: %v", err)
-	} else if !featureGate.Enabled(features.AntreaProxy) {
-		t.Skip("Skipping test because Traceflow is not enabled in the Controller")
-	}
+	skipIfFeatureDisabled(t, data, features.Traceflow, true, true)
 }
 
 var (
diff --git a/test/integration/agent/egress_ip_assign_linux_test.go b/test/integration/agent/egress_ip_assign_linux_test.go
new file mode 100644
index 000000000..3ae0f5ddc
--- /dev/null
+++ b/test/integration/agent/egress_ip_assign_linux_test.go
@@ -0,0 +1,73 @@
+// 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 agent
+
+import (
+	"io/ioutil"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	"antrea.io/antrea/pkg/agent/controller/egress"
+	"antrea.io/antrea/pkg/agent/controller/egress/ipassigner"
+	"antrea.io/antrea/pkg/apis/crd/v1alpha2"
+)
+
+const defaultEgressRunDir = egress.DefaultEgressRunDir
+
+func TestEgressIPAssigner(t *testing.T) {
+	ipAssigner, err := ipassigner.NewIPAssigner(nodeIP.IP, defaultEgressRunDir)
+	if err != nil {
+		t.Fatalf("Initializing egressIP assigner failed: %v", err)
+	}
+
+	egressName := "test-egress"
+	egressIP := "10.10.10.10"
+	eg := &v1alpha2.Egress{
+		ObjectMeta: metav1.ObjectMeta{Name: egressName},
+		Spec:       v1alpha2.EgressSpec{EgressIP: egressIP},
+	}
+	assignErr := ipAssigner.AssignEgressIP("", eg.Name)
+	assert.Error(t, assignErr, "invalid IP ")
+	assignErr = ipAssigner.AssignEgressIP("x", eg.Name)
+	assert.Error(t, assignErr, "invalid IP x")
+	// Assign EgressIP.
+	if err := ipAssigner.AssignEgressIP(eg.Spec.EgressIP, eg.Name); err != nil {
+		t.Fatalf("Assigning egress IP error: %v", err)
+	}
+
+	ips := ipAssigner.AssignedIPs()
+	assert.Equal(t, ips, map[string]string{eg.Name: eg.Spec.EgressIP}, "List assigned IPs not match")
+
+	readEgressIPFile := func(egressName string) string {
+		fileName := strings.Join([]string{defaultEgressRunDir, egressName}, "/")
+		ipStr, err := ioutil.ReadFile(fileName)
+		if err != nil {
+			return ""
+		}
+		return string(ipStr)
+	}
+	assert.Equal(t, eg.Spec.EgressIP, readEgressIPFile(eg.Name), "Reading IP assigned file not match")
+
+	// Unassign EgressIP.
+	if err := ipAssigner.UnassignEgressIP(eg.Name); err != nil {
+		t.Fatalf("Unassigning egress IP error: %v", err)
+	}
+
+	ips = ipAssigner.AssignedIPs()
+	assert.Equal(t, map[string]string{}, ips, "List assigned IPs not match")
+}