Repository: antrea-io/antrea
Pull request #3608 (2022-04-08 21:50:21)

Support ClickHouse deployment with Persistent Volume
This commit supports deploying ClickHouse with Persistent Volume. It provides examples using Local PV, NFS PV or other customized StorageClass.

Signed-off-by: Yanjun Zhou <zhouya@vmware.com>

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

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

```diff
@@             Coverage Diff             @@
##             main    #3608       +/-   ##
===========================================
- Coverage   63.12%   41.67%   -21.46%     
===========================================
  Files         278      324       +46     
  Lines       27829    40590    +12761     
===========================================
- Hits        17568    16914      -654     
- Misses       8388    22101    +13713     
+ Partials     1873     1575      -298     
```

| Flag | Coverage Δ | |
|---|---|---|
| integration-tests | `38.72% <ø> (?)` | |
| kind-e2e-tests | `?` | |
| unit-tests | `43.44% <ø> (-0.08%)` | :arrow_down: |

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

| [Impacted Files](https://codecov.io/gh/antrea-io/antrea/pull/3608?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) | Coverage Δ | |
|---|---|---|
| [pkg/ovs/openflow/default.go](https://codecov.io/gh/antrea-io/antrea/pull/3608/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL292cy9vcGVuZmxvdy9kZWZhdWx0Lmdv) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [pkg/util/runtime/runtime.go](https://codecov.io/gh/antrea-io/antrea/pull/3608/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL3V0aWwvcnVudGltZS9ydW50aW1lLmdv) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [pkg/agent/openflow/groups.go](https://codecov.io/gh/antrea-io/antrea/pull/3608/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2FnZW50L29wZW5mbG93L2dyb3Vwcy5nbw==) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [pkg/agent/nodeportlocal/k8s/annotations.go](https://codecov.io/gh/antrea-io/antrea/pull/3608/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2FnZW50L25vZGVwb3J0bG9jYWwvazhzL2Fubm90YXRpb25zLmdv) | `0.00% <0.00%> (-93.55%)` | :arrow_down: |
| [pkg/ovs/openflow/ofctrl\_pipeline.go](https://codecov.io/gh/antrea-io/antrea/pull/3608/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL292cy9vcGVuZmxvdy9vZmN0cmxfcGlwZWxpbmUuZ28=) | `0.00% <0.00%> (-93.34%)` | :arrow_down: |
| [pkg/ovs/openflow/logs.go](https://codecov.io/gh/antrea-io/antrea/pull/3608/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL292cy9vcGVuZmxvdy9sb2dzLmdv) | `9.52% <0.00%> (-90.48%)` | :arrow_down: |
| [pkg/apis/controlplane/register.go](https://codecov.io/gh/antrea-io/antrea/pull/3608/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2FwaXMvY29udHJvbHBsYW5lL3JlZ2lzdGVyLmdv) | `0.00% <0.00%> (-90.48%)` | :arrow_down: |
| [pkg/agent/controller/networkpolicy/reject.go](https://codecov.io/gh/antrea-io/antrea/pull/3608/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2FnZW50L2NvbnRyb2xsZXIvbmV0d29ya3BvbGljeS9yZWplY3QuZ28=) | `0.00% <0.00%> (-85.08%)` | :arrow_down: |
| [pkg/flowaggregator/certificate.go](https://codecov.io/gh/antrea-io/antrea/pull/3608/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2Zsb3dhZ2dyZWdhdG9yL2NlcnRpZmljYXRlLmdv) | `0.00% <0.00%> (-77.48%)` | :arrow_down: |
| [pkg/agent/client.go](https://codecov.io/gh/antrea-io/antrea/pull/3608/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2FnZW50L2NsaWVudC5nbw==) | `0.00% <0.00%> (-77.42%)` | :arrow_down: |
| ... and [308 more](https://codecov.io/gh/antrea-io/antrea/pull/3608/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) | |

yanjunz97: Hi @antoninbas Could you take another look at this PR?
antoninbas: @yanjunz97 I approved but looks like you have a merge conflict
yanjunz97: > @yanjunz97 I approved but looks like you have a merge conflict

Thanks! I have rebased and resolved the conflicts.
yanjunz97: /test-e2e
/test-conformance
/test-networkpolicy

Patch:
diff --git a/.github/workflows/docker_update_ethtool.yml b/.github/workflows/docker_update_ethtool.yml
index 8ec7637d3..0d7bbdd71 100644
--- a/.github/workflows/docker_update_ethtool.yml
+++ b/.github/workflows/docker_update_ethtool.yml
@@ -26,16 +26,16 @@ jobs:
       with:
         ref: ${{ github.event.inputs.antrea-ref }}
     - name: Set up QEMU
-      uses: docker/setup-qemu-action@v1
+      uses: docker/setup-qemu-action@v2
     - name: Set up Docker Buildx
-      uses: docker/setup-buildx-action@v1
+      uses: docker/setup-buildx-action@v2
     - name: Login to DockerHub
-      uses: docker/login-action@v1
+      uses: docker/login-action@v2
       with:
         username: ${{ secrets.DOCKER_USERNAME }}
         password: ${{ secrets.DOCKER_PASSWORD }}
     - name: Build and push Docker image
-      uses: docker/build-push-action@v2
+      uses: docker/build-push-action@v3
       with:
         context: build/images/ethtool
         platforms: linux/amd64,linux/arm64,linux/arm/v7
diff --git a/.github/workflows/docker_update_flow_visibility.yml b/.github/workflows/docker_update_flow_visibility.yml
index 0784213c8..95e818325 100644
--- a/.github/workflows/docker_update_flow_visibility.yml
+++ b/.github/workflows/docker_update_flow_visibility.yml
@@ -22,7 +22,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - name: Login to DockerHub
-        uses: docker/login-action@v1
+        uses: docker/login-action@v2
         with:
           username: ${{ secrets.DOCKER_USERNAME }}
           password: ${{ secrets.DOCKER_PASSWORD }}
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index 6ca34583e..f4fa7ce75 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -34,7 +34,6 @@ contributing to project Antrea:
 2. Check out the [Architecture document](docs/design/architecture.md) for the Antrea
    architecture and design.
 3. Set up necessary [accounts](#accounts-setup).
-4. Set up your [development environment](docs/contributors/manual-installation.md)
 
 Now that you're setup, skip ahead to learn how to [contribute](#contribute).
 
diff --git a/build/charts/antrea/README.md b/build/charts/antrea/README.md
index 9005e7c0b..2b14782d0 100644
--- a/build/charts/antrea/README.md
+++ b/build/charts/antrea/README.md
@@ -61,7 +61,7 @@ Kubernetes: `>= 1.16.0-0`
 | controller.podLabels | object | `{}` | Labels to be added to antrea-controller Pod. |
 | controller.priorityClassName | string | `"system-cluster-critical"` | Prority class to use for the antrea-controller Pod. |
 | controller.selfSignedCert | bool | `true` | Indicates whether to use auto-generated self-signed TLS certificates. If false, a Secret named "antrea-controller-tls" must be provided with the following keys: ca.crt, tls.crt, tls.key. |
-| controller.tolerations | list | `[{"key":"CriticalAddonsOnly","operator":"Exists"},{"effect":"NoSchedule","key":"node-role.kubernetes.io/master"}]` | Tolerations for the antrea-controller Pod. |
+| controller.tolerations | list | `[{"key":"CriticalAddonsOnly","operator":"Exists"},{"effect":"NoSchedule","key":"node-role.kubernetes.io/master"},{"effect":"NoSchedule","key":"node-role.kubernetes.io/control-plane"}]` | Tolerations for the antrea-controller Pod. |
 | defaultMTU | int | `0` | Default MTU to use for the host gateway interface and the network interface of each Pod. By default, antrea-agent will discover the MTU of the Node's primary interface and adjust it to accommodate for tunnel encapsulation overhead if applicable. |
 | egress.exceptCIDRs | list | `[]` | CIDR ranges to which outbound Pod traffic will not be SNAT'd by Egresses. |
 | enableBridgingMode | bool | `false` | Enable bridging mode of Pod network on Nodes, in which the Node's transport interface is connected to the OVS bridge. |
diff --git a/build/charts/antrea/templates/agent/secret.yaml b/build/charts/antrea/templates/agent/secret.yaml
new file mode 100644
index 000000000..90053e027
--- /dev/null
+++ b/build/charts/antrea/templates/agent/secret.yaml
@@ -0,0 +1,8 @@
+apiVersion: v1
+kind: Secret
+metadata:
+  name: antrea-agent-service-account-token
+  namespace: {{ .Release.Namespace }}
+  annotations:
+    kubernetes.io/service-account.name: antrea-agent
+type: kubernetes.io/service-account-token
diff --git a/build/charts/antrea/templates/antctl/secret.yaml b/build/charts/antrea/templates/antctl/secret.yaml
new file mode 100644
index 000000000..c964bc87f
--- /dev/null
+++ b/build/charts/antrea/templates/antctl/secret.yaml
@@ -0,0 +1,8 @@
+apiVersion: v1
+kind: Secret
+metadata:
+  name: antctl-service-account-token
+  namespace: {{ .Release.Namespace }}
+  annotations:
+    kubernetes.io/service-account.name: antctl
+type: kubernetes.io/service-account-token
diff --git a/build/charts/antrea/templates/crds/clusternetworkpolicy.yaml b/build/charts/antrea/templates/crds/clusternetworkpolicy.yaml
index 307048661..a2a90da43 100644
--- a/build/charts/antrea/templates/crds/clusternetworkpolicy.yaml
+++ b/build/charts/antrea/templates/crds/clusternetworkpolicy.yaml
@@ -494,6 +494,8 @@ spec:
                               type: object
                               properties:
                                 match:
+                                  enum:
+                                    - Self
                                   type: string
                             ipBlock:
                               type: object
diff --git a/build/charts/antrea/values.yaml b/build/charts/antrea/values.yaml
index 8a587f202..45a1297ba 100644
--- a/build/charts/antrea/values.yaml
+++ b/build/charts/antrea/values.yaml
@@ -203,6 +203,9 @@ controller:
     # Allow it to schedule onto master nodes.
     - key: node-role.kubernetes.io/master
       effect: NoSchedule
+    # Control-plane taint for Kubernetes >= 1.24.
+    - key: node-role.kubernetes.io/control-plane
+      effect: NoSchedule
   # -- Node selector for the antrea-controller Pod.
   nodeSelector:
     kubernetes.io/os: linux
diff --git a/build/yamls/antrea-aks.yml b/build/yamls/antrea-aks.yml
index c5fb6bfbe..6f04524d2 100644
--- a/build/yamls/antrea-aks.yml
+++ b/build/yamls/antrea-aks.yml
@@ -26,6 +26,26 @@ metadata:
   labels:
     app: antrea
 ---
+# Source: antrea/templates/agent/secret.yaml
+apiVersion: v1
+kind: Secret
+metadata:
+  name: antrea-agent-service-account-token
+  namespace: kube-system
+  annotations:
+    kubernetes.io/service-account.name: antrea-agent
+type: kubernetes.io/service-account-token
+---
+# Source: antrea/templates/antctl/secret.yaml
+apiVersion: v1
+kind: Secret
+metadata:
+  name: antctl-service-account-token
+  namespace: kube-system
+  annotations:
+    kubernetes.io/service-account.name: antctl
+type: kubernetes.io/service-account-token
+---
 # Source: antrea/templates/configmap.yaml
 apiVersion: v1
 kind: ConfigMap
@@ -1245,6 +1265,8 @@ spec:
                               type: object
                               properties:
                                 match:
+                                  enum:
+                                    - Self
                                   type: string
                             ipBlock:
                               type: object
@@ -3723,6 +3745,8 @@ spec:
           operator: Exists
         - effect: NoSchedule
           key: node-role.kubernetes.io/master
+        - effect: NoSchedule
+          key: node-role.kubernetes.io/control-plane
       serviceAccountName: antrea-controller
       containers:
         - name: antrea-controller
diff --git a/build/yamls/antrea-eks.yml b/build/yamls/antrea-eks.yml
index 80487df05..9b6b2b40f 100644
--- a/build/yamls/antrea-eks.yml
+++ b/build/yamls/antrea-eks.yml
@@ -26,6 +26,26 @@ metadata:
   labels:
     app: antrea
 ---
+# Source: antrea/templates/agent/secret.yaml
+apiVersion: v1
+kind: Secret
+metadata:
+  name: antrea-agent-service-account-token
+  namespace: kube-system
+  annotations:
+    kubernetes.io/service-account.name: antrea-agent
+type: kubernetes.io/service-account-token
+---
+# Source: antrea/templates/antctl/secret.yaml
+apiVersion: v1
+kind: Secret
+metadata:
+  name: antctl-service-account-token
+  namespace: kube-system
+  annotations:
+    kubernetes.io/service-account.name: antctl
+type: kubernetes.io/service-account-token
+---
 # Source: antrea/templates/configmap.yaml
 apiVersion: v1
 kind: ConfigMap
@@ -1245,6 +1265,8 @@ spec:
                               type: object
                               properties:
                                 match:
+                                  enum:
+                                    - Self
                                   type: string
                             ipBlock:
                               type: object
@@ -3725,6 +3747,8 @@ spec:
           operator: Exists
         - effect: NoSchedule
           key: node-role.kubernetes.io/master
+        - effect: NoSchedule
+          key: node-role.kubernetes.io/control-plane
       serviceAccountName: antrea-controller
       containers:
         - name: antrea-controller
diff --git a/build/yamls/antrea-gke.yml b/build/yamls/antrea-gke.yml
index a0ed8f188..9d6de7b5a 100644
--- a/build/yamls/antrea-gke.yml
+++ b/build/yamls/antrea-gke.yml
@@ -26,6 +26,26 @@ metadata:
   labels:
     app: antrea
 ---
+# Source: antrea/templates/agent/secret.yaml
+apiVersion: v1
+kind: Secret
+metadata:
+  name: antrea-agent-service-account-token
+  namespace: kube-system
+  annotations:
+    kubernetes.io/service-account.name: antrea-agent
+type: kubernetes.io/service-account-token
+---
+# Source: antrea/templates/antctl/secret.yaml
+apiVersion: v1
+kind: Secret
+metadata:
+  name: antctl-service-account-token
+  namespace: kube-system
+  annotations:
+    kubernetes.io/service-account.name: antctl
+type: kubernetes.io/service-account-token
+---
 # Source: antrea/templates/configmap.yaml
 apiVersion: v1
 kind: ConfigMap
@@ -1245,6 +1265,8 @@ spec:
                               type: object
                               properties:
                                 match:
+                                  enum:
+                                    - Self
                                   type: string
                             ipBlock:
                               type: object
@@ -3723,6 +3745,8 @@ spec:
           operator: Exists
         - effect: NoSchedule
           key: node-role.kubernetes.io/master
+        - effect: NoSchedule
+          key: node-role.kubernetes.io/control-plane
       serviceAccountName: antrea-controller
       containers:
         - name: antrea-controller
diff --git a/build/yamls/antrea-ipsec.yml b/build/yamls/antrea-ipsec.yml
index d029f1a8e..681db390b 100644
--- a/build/yamls/antrea-ipsec.yml
+++ b/build/yamls/antrea-ipsec.yml
@@ -39,6 +39,26 @@ stringData:
   # Preshared Key used by IKE for authentication with peers.
   psk: "changeme"
 ---
+# Source: antrea/templates/agent/secret.yaml
+apiVersion: v1
+kind: Secret
+metadata:
+  name: antrea-agent-service-account-token
+  namespace: kube-system
+  annotations:
+    kubernetes.io/service-account.name: antrea-agent
+type: kubernetes.io/service-account-token
+---
+# Source: antrea/templates/antctl/secret.yaml
+apiVersion: v1
+kind: Secret
+metadata:
+  name: antctl-service-account-token
+  namespace: kube-system
+  annotations:
+    kubernetes.io/service-account.name: antctl
+type: kubernetes.io/service-account-token
+---
 # Source: antrea/templates/configmap.yaml
 apiVersion: v1
 kind: ConfigMap
@@ -1258,6 +1278,8 @@ spec:
                               type: object
                               properties:
                                 match:
+                                  enum:
+                                    - Self
                                   type: string
                             ipBlock:
                               type: object
@@ -3772,6 +3794,8 @@ spec:
           operator: Exists
         - effect: NoSchedule
           key: node-role.kubernetes.io/master
+        - effect: NoSchedule
+          key: node-role.kubernetes.io/control-plane
       serviceAccountName: antrea-controller
       containers:
         - name: antrea-controller
diff --git a/build/yamls/antrea-prometheus.yml b/build/yamls/antrea-prometheus.yml
index 3ee508fa0..84a5bbbb7 100644
--- a/build/yamls/antrea-prometheus.yml
+++ b/build/yamls/antrea-prometheus.yml
@@ -12,6 +12,16 @@ metadata:
   name: prometheus
   namespace: monitoring
 ---
+# This is useful for testing and manual access to the /metrics endpoints
+apiVersion: v1
+kind: Secret
+metadata:
+  name: prometheus-service-account-token
+  namespace: monitoring
+  annotations:
+    kubernetes.io/service-account.name: prometheus
+type: kubernetes.io/service-account-token
+---
 # Authorize Prometheus to view Kubernetes cluster components for service discovery purposes
 # Authorize Prometheus to retrieve metrics
 apiVersion: rbac.authorization.k8s.io/v1
diff --git a/build/yamls/antrea.yml b/build/yamls/antrea.yml
index 1ceb4e174..d9465dfee 100644
--- a/build/yamls/antrea.yml
+++ b/build/yamls/antrea.yml
@@ -26,6 +26,26 @@ metadata:
   labels:
     app: antrea
 ---
+# Source: antrea/templates/agent/secret.yaml
+apiVersion: v1
+kind: Secret
+metadata:
+  name: antrea-agent-service-account-token
+  namespace: kube-system
+  annotations:
+    kubernetes.io/service-account.name: antrea-agent
+type: kubernetes.io/service-account-token
+---
+# Source: antrea/templates/antctl/secret.yaml
+apiVersion: v1
+kind: Secret
+metadata:
+  name: antctl-service-account-token
+  namespace: kube-system
+  annotations:
+    kubernetes.io/service-account.name: antctl
+type: kubernetes.io/service-account-token
+---
 # Source: antrea/templates/configmap.yaml
 apiVersion: v1
 kind: ConfigMap
@@ -1245,6 +1265,8 @@ spec:
                               type: object
                               properties:
                                 match:
+                                  enum:
+                                    - Self
                                   type: string
                             ipBlock:
                               type: object
@@ -3723,6 +3745,8 @@ spec:
           operator: Exists
         - effect: NoSchedule
           key: node-role.kubernetes.io/master
+        - effect: NoSchedule
+          key: node-role.kubernetes.io/control-plane
       serviceAccountName: antrea-controller
       containers:
         - name: antrea-controller
diff --git a/build/yamls/flow-visibility.yml b/build/yamls/flow-visibility.yml
index 7217afb84..3f8ba4598 100644
--- a/build/yamls/flow-visibility.yml
+++ b/build/yamls/flow-visibility.yml
@@ -86,26 +86,27 @@ data:
     UInt64,\n        reverseThroughputFromDestinationNode UInt64,\n        trusted
     UInt8 DEFAULT 0\n    ) engine=MergeTree\n    ORDER BY (timeInserted, flowEndSeconds)\n
     \   TTL timeInserted + INTERVAL 1 HOUR\n    SETTINGS merge_with_ttl_timeout =
-    3600;\n\n    CREATE MATERIALIZED VIEW flows_pod_view\n    ENGINE = SummingMergeTree\n
-    \   ORDER BY (\n        timeInserted,\n        flowEndSeconds,\n        flowEndSecondsFromSourceNode,\n
-    \       flowEndSecondsFromDestinationNode,\n        sourcePodName,\n        destinationPodName,\n
-    \       destinationIP,\n        destinationServicePortName,\n        flowType,\n
-    \       sourcePodNamespace,\n        destinationPodNamespace)\n    TTL timeInserted
-    + INTERVAL 1 HOUR\n    SETTINGS merge_with_ttl_timeout = 3600\n    POPULATE\n
-    \   AS SELECT\n        timeInserted,\n        flowEndSeconds,\n        flowEndSecondsFromSourceNode,\n
-    \       flowEndSecondsFromDestinationNode,\n        sourcePodName,\n        destinationPodName,\n
-    \       destinationIP,\n        destinationServicePortName,\n        flowType,\n
-    \       sourcePodNamespace,\n        destinationPodNamespace,\n        sum(octetDeltaCount)
-    AS octetDeltaCount,\n        sum(reverseOctetDeltaCount) AS reverseOctetDeltaCount,\n
-    \       sum(throughput) AS throughput,\n        sum(reverseThroughput) AS reverseThroughput,\n
-    \       sum(throughputFromSourceNode) AS throughputFromSourceNode,\n        sum(throughputFromDestinationNode)
-    AS throughputFromDestinationNode\n    FROM flows\n    GROUP BY\n        timeInserted,\n
+    3600;\n\n    CREATE MATERIALIZED VIEW IF NOT EXISTS flows_pod_view\n    ENGINE
+    = SummingMergeTree\n    ORDER BY (\n        timeInserted,\n        flowEndSeconds,\n
+    \       flowEndSecondsFromSourceNode,\n        flowEndSecondsFromDestinationNode,\n
+    \       sourcePodName,\n        destinationPodName,\n        destinationIP,\n
+    \       destinationServicePortName,\n        flowType,\n        sourcePodNamespace,\n
+    \       destinationPodNamespace)\n    TTL timeInserted + INTERVAL 1 HOUR\n    SETTINGS
+    merge_with_ttl_timeout = 3600\n    POPULATE\n    AS SELECT\n        timeInserted,\n
     \       flowEndSeconds,\n        flowEndSecondsFromSourceNode,\n        flowEndSecondsFromDestinationNode,\n
     \       sourcePodName,\n        destinationPodName,\n        destinationIP,\n
     \       destinationServicePortName,\n        flowType,\n        sourcePodNamespace,\n
-    \       destinationPodNamespace;\n\n    CREATE MATERIALIZED VIEW flows_node_view\n
-    \   ENGINE = SummingMergeTree\n    ORDER BY (\n        timeInserted,\n        flowEndSeconds,\n
+    \       destinationPodNamespace,\n        sum(octetDeltaCount) AS octetDeltaCount,\n
+    \       sum(reverseOctetDeltaCount) AS reverseOctetDeltaCount,\n        sum(throughput)
+    AS throughput,\n        sum(reverseThroughput) AS reverseThroughput,\n        sum(throughputFromSourceNode)
+    AS throughputFromSourceNode,\n        sum(throughputFromDestinationNode) AS throughputFromDestinationNode\n
+    \   FROM flows\n    GROUP BY\n        timeInserted,\n        flowEndSeconds,\n
     \       flowEndSecondsFromSourceNode,\n        flowEndSecondsFromDestinationNode,\n
+    \       sourcePodName,\n        destinationPodName,\n        destinationIP,\n
+    \       destinationServicePortName,\n        flowType,\n        sourcePodNamespace,\n
+    \       destinationPodNamespace;\n\n    CREATE MATERIALIZED VIEW IF NOT EXISTS
+    flows_node_view\n    ENGINE = SummingMergeTree\n    ORDER BY (\n        timeInserted,\n
+    \       flowEndSeconds,\n        flowEndSecondsFromSourceNode,\n        flowEndSecondsFromDestinationNode,\n
     \       sourceNodeName,\n        destinationNodeName,\n        sourcePodNamespace,\n
     \       destinationPodNamespace)\n    TTL timeInserted + INTERVAL 1 HOUR\n    SETTINGS
     merge_with_ttl_timeout = 3600\n    POPULATE\n    AS SELECT\n        timeInserted,\n
@@ -120,9 +121,9 @@ data:
     AS reverseThroughputFromDestinationNode\n    FROM flows\n    GROUP BY\n        timeInserted,\n
     \       flowEndSeconds,\n        flowEndSecondsFromSourceNode,\n        flowEndSecondsFromDestinationNode,\n
     \       sourceNodeName,\n        destinationNodeName,\n        sourcePodNamespace,\n
-    \       destinationPodNamespace;\n\n    CREATE MATERIALIZED VIEW flows_policy_view\n
-    \   ENGINE = SummingMergeTree\n    ORDER BY (\n        timeInserted,\n        flowEndSeconds,\n
-    \       flowEndSecondsFromSourceNode,\n        flowEndSecondsFromDestinationNode,\n
+    \       destinationPodNamespace;\n\n    CREATE MATERIALIZED VIEW IF NOT EXISTS
+    flows_policy_view\n    ENGINE = SummingMergeTree\n    ORDER BY (\n        timeInserted,\n
+    \       flowEndSeconds,\n        flowEndSecondsFromSourceNode,\n        flowEndSecondsFromDestinationNode,\n
     \       egressNetworkPolicyName,\n        egressNetworkPolicyRuleAction,\n        ingressNetworkPolicyName,\n
     \       ingressNetworkPolicyRuleAction,\n        sourcePodNamespace,\n        destinationPodNamespace)\n
     \   TTL timeInserted + INTERVAL 1 HOUR\n    SETTINGS merge_with_ttl_timeout =
@@ -145,7 +146,7 @@ data:
     \   ORDER BY (timeCreated);\n    \nEOSQL\n"
 kind: ConfigMap
 metadata:
-  name: clickhouse-mounted-configmap-dkbmg82ctg
+  name: clickhouse-mounted-configmap-58fkkt9b56
   namespace: flow-visibility
 ---
 apiVersion: v1
@@ -4934,12 +4935,14 @@ spec:
             value: default.flows
           - name: MV_NAMES
             value: default.flows_pod_view default.flows_node_view default.flows_policy_view
+          - name: STORAGE_SIZE
+            value: 8Gi
           image: projects.registry.vmware.com/antrea/flow-visibility-clickhouse-monitor:latest
           imagePullPolicy: IfNotPresent
           name: clickhouse-monitor
         volumes:
         - configMap:
-            name: clickhouse-mounted-configmap-dkbmg82ctg
+            name: clickhouse-mounted-configmap-58fkkt9b56
           name: clickhouse-configmap-volume
         - emptyDir:
             medium: Memory
diff --git a/build/yamls/flow-visibility/base/clickhouse.yml b/build/yamls/flow-visibility/base/clickhouse.yml
index 18fe72ee7..cfe352936 100644
--- a/build/yamls/flow-visibility/base/clickhouse.yml
+++ b/build/yamls/flow-visibility/base/clickhouse.yml
@@ -45,32 +45,7 @@ spec:
               volumeMounts:
                 - name: clickhouse-configmap-volume
                   mountPath: /docker-entrypoint-initdb.d
-                - name: clickhouse-storage-volume
-                  mountPath: /var/lib/clickhouse
-            - name: clickhouse-monitor
-              image: flow-visibility-clickhouse-monitor
-              env:
-                - name: CLICKHOUSE_USERNAME
-                  valueFrom:
-                    secretKeyRef: 
-                      name: clickhouse-secret
-                      key: username
-                - name: CLICKHOUSE_PASSWORD
-                  valueFrom:
-                    secretKeyRef:
-                      name: clickhouse-secret
-                      key: password
-                - name: DB_URL
-                  value: "tcp://localhost:9000"
-                - name: TABLE_NAME
-                  value: "default.flows"
-                - name: MV_NAMES
-                  value: "default.flows_pod_view default.flows_node_view default.flows_policy_view"
           volumes:
             - name: clickhouse-configmap-volume
               configMap:
                 name: $(CLICKHOUSE_CONFIG_MAP_NAME)
-            - name: clickhouse-storage-volume
-              emptyDir:
-                medium: Memory
-                sizeLimit: 8Gi
diff --git a/build/yamls/flow-visibility/base/provisioning/datasources/create_table.sh b/build/yamls/flow-visibility/base/provisioning/datasources/create_table.sh
index 9f1355794..423fa45d0 100644
--- a/build/yamls/flow-visibility/base/provisioning/datasources/create_table.sh
+++ b/build/yamls/flow-visibility/base/provisioning/datasources/create_table.sh
@@ -72,7 +72,7 @@ clickhouse client -n -h 127.0.0.1 <<-EOSQL
     TTL timeInserted + INTERVAL 1 HOUR
     SETTINGS merge_with_ttl_timeout = 3600;
 
-    CREATE MATERIALIZED VIEW flows_pod_view
+    CREATE MATERIALIZED VIEW IF NOT EXISTS flows_pod_view
     ENGINE = SummingMergeTree
     ORDER BY (
         timeInserted,
@@ -121,7 +121,7 @@ clickhouse client -n -h 127.0.0.1 <<-EOSQL
         sourcePodNamespace,
         destinationPodNamespace;
 
-    CREATE MATERIALIZED VIEW flows_node_view
+    CREATE MATERIALIZED VIEW IF NOT EXISTS flows_node_view
     ENGINE = SummingMergeTree
     ORDER BY (
         timeInserted,
@@ -163,7 +163,7 @@ clickhouse client -n -h 127.0.0.1 <<-EOSQL
         sourcePodNamespace,
         destinationPodNamespace;
 
-    CREATE MATERIALIZED VIEW flows_policy_view
+    CREATE MATERIALIZED VIEW IF NOT EXISTS flows_policy_view
     ENGINE = SummingMergeTree
     ORDER BY (
         timeInserted,
diff --git a/build/yamls/flow-visibility/patches/chmonitor/chMonitor.yml b/build/yamls/flow-visibility/patches/chmonitor/chMonitor.yml
new file mode 100644
index 000000000..68abcc0bb
--- /dev/null
+++ b/build/yamls/flow-visibility/patches/chmonitor/chMonitor.yml
@@ -0,0 +1,24 @@
+- op: add
+  path: /spec/templates/podTemplates/0/spec/containers/-
+  value: 
+    name: clickhouse-monitor
+    image: flow-visibility-clickhouse-monitor
+    env:
+      - name: CLICKHOUSE_USERNAME
+        valueFrom:
+          secretKeyRef: 
+            name: clickhouse-secret
+            key: username
+      - name: CLICKHOUSE_PASSWORD
+        valueFrom:
+          secretKeyRef:
+            name: clickhouse-secret
+            key: password
+      - name: DB_URL
+        value: "tcp://localhost:9000"
+      - name: TABLE_NAME
+        value: "default.flows"
+      - name: MV_NAMES
+        value: "default.flows_pod_view default.flows_node_view default.flows_policy_view"
+      - name: STORAGE_SIZE
+        value: STORAGE_SIZE_VALUE
diff --git a/build/yamls/flow-visibility/patches/pv/createLocalPv.yml b/build/yamls/flow-visibility/patches/pv/createLocalPv.yml
new file mode 100644
index 000000000..bda1ac4e3
--- /dev/null
+++ b/build/yamls/flow-visibility/patches/pv/createLocalPv.yml
@@ -0,0 +1,28 @@
+apiVersion: storage.k8s.io/v1
+kind: StorageClass
+metadata:
+  name: clickhouse-storage
+provisioner: kubernetes.io/no-provisioner
+volumeBindingMode: WaitForFirstConsumer
+reclaimPolicy: Retain
+allowVolumeExpansion: True
+---
+apiVersion: v1
+kind: PersistentVolume
+metadata:
+  name: clickhouse-pv
+spec:
+  storageClassName: clickhouse-storage
+  capacity:
+    storage: STORAGE_SIZE
+  accessModes:
+    - ReadWriteOnce
+  volumeMode: Filesystem
+  local:
+    path: LOCAL_PATH
+  nodeAffinity:
+    required:
+      nodeSelectorTerms:
+      - matchExpressions:
+        - key: antrea.io/clickhouse-data-node
+          operator: Exists
diff --git a/build/yamls/flow-visibility/patches/pv/createNfsPv.yml b/build/yamls/flow-visibility/patches/pv/createNfsPv.yml
new file mode 100644
index 000000000..2cef3949a
--- /dev/null
+++ b/build/yamls/flow-visibility/patches/pv/createNfsPv.yml
@@ -0,0 +1,23 @@
+apiVersion: storage.k8s.io/v1
+kind: StorageClass
+metadata:
+  name: clickhouse-storage
+provisioner: kubernetes.io/no-provisioner
+volumeBindingMode: WaitForFirstConsumer
+reclaimPolicy: Retain
+allowVolumeExpansion: True
+---
+apiVersion: v1
+kind: PersistentVolume
+metadata:
+  name: clickhouse-pv
+spec:
+  storageClassName: clickhouse-storage
+  capacity:
+    storage: STORAGE_SIZE
+  accessModes:
+    - ReadWriteOnce
+  volumeMode: Filesystem
+  nfs:
+    path: NFS_SERVER_PATH
+    server: NFS_SERVER_ADDRESS
diff --git a/build/yamls/flow-visibility/patches/pv/mountPv.yml b/build/yamls/flow-visibility/patches/pv/mountPv.yml
new file mode 100644
index 000000000..c3d4d898b
--- /dev/null
+++ b/build/yamls/flow-visibility/patches/pv/mountPv.yml
@@ -0,0 +1,14 @@
+- op: add
+  path: /spec/defaults/templates/dataVolumeClaimTemplate
+  value: clickhouse-storage-template
+- op: add
+  path: /spec/templates/volumeClaimTemplates
+  value: 
+    - name: clickhouse-storage-template
+      spec:
+        storageClassName: STORAGECLASS_NAME
+        accessModes:
+          - ReadWriteOnce
+        resources:
+          requests:
+            storage: STORAGE_SIZE
diff --git a/build/yamls/flow-visibility/patches/ram/mountRam.yml b/build/yamls/flow-visibility/patches/ram/mountRam.yml
new file mode 100644
index 000000000..541136de9
--- /dev/null
+++ b/build/yamls/flow-visibility/patches/ram/mountRam.yml
@@ -0,0 +1,12 @@
+- op: add
+  path: /spec/templates/podTemplates/0/spec/volumes/-
+  value: 
+    name: clickhouse-storage-volume
+    emptyDir:
+      medium: Memory
+      sizeLimit: STORAGE_SIZE
+- op: add
+  path: /spec/templates/podTemplates/0/spec/containers/0/volumeMounts/-
+  value: 
+    name: clickhouse-storage-volume
+    mountPath: /var/lib/clickhouse
diff --git a/cmd/antrea-agent/agent.go b/cmd/antrea-agent/agent.go
index 94a91d8fa..8cd98f60d 100644
--- a/cmd/antrea-agent/agent.go
+++ b/cmd/antrea-agent/agent.go
@@ -579,11 +579,13 @@ func run(o *Options) error {
 		}
 		mcastController := multicast.NewMulticastController(
 			ofClient,
+			v4GroupIDAllocator,
 			nodeConfig,
 			ifaceStore,
 			multicastSocket,
 			sets.NewString(append(o.config.MulticastInterfaces, nodeConfig.NodeTransportInterfaceName)...),
-			ovsBridgeClient)
+			ovsBridgeClient,
+			podUpdateChannel)
 		if err := mcastController.Initialize(); err != nil {
 			return err
 		}
diff --git a/docs/configuration.md b/docs/configuration.md
index 8a2d70c57..c443eb4d8 100644
--- a/docs/configuration.md
+++ b/docs/configuration.md
@@ -15,9 +15,9 @@ Use `antrea-agent -h` to see complete options.
 
 The `antrea-agent` configuration file specifies the agent configuration
 parameters. For all the agent configuration parameters of a Linux Node, refer to
-this [base configuration file](/build/yamls/base/conf/antrea-agent.conf).
+this [base configuration file](../build/charts/antrea/conf/antrea-agent.conf).
 For all the configuration parameters of a Windows Node, refer to this [base
-configuration file](/build/yamls/windows/base/conf/antrea-agent.conf)
+configuration file](../build/yamls/windows/base/conf/antrea-agent.conf)
 
 ## antrea-controller
 
@@ -34,7 +34,7 @@ Use `antrea-controller -h` to see complete options.
 
 The `antrea-controller` configuration file specifies the controller
 configuration parameters. For all the controller configuration parameters,
-refer to this [base configuration file](/build/yamls/base/conf/antrea-controller.conf).
+refer to this [base configuration file](../build/charts/antrea/conf/antrea-controller.conf).
 
 ## CNI configuration
 
diff --git a/docs/contributors/manual-installation.md b/docs/contributors/manual-installation.md
deleted file mode 100644
index 13aa595c2..000000000
--- a/docs/contributors/manual-installation.md
+++ /dev/null
@@ -1,171 +0,0 @@
-# Manual Installation
-
-## Overview
-
-There are four components which need to be deployed in order to run Antrea:
-
-* The OpenVSwitch daemons `ovs-vswitchd` and `ovsdb-server`
-
-* The controller `antrea-controller`
-
-* The agent `antrea-agent`
-
-* The CNI plugin `antrea-cni`
-
-## Instructions
-
-Prior to bringing up the individual components, follow the common steps:
-
-* Ensure Go v1.15 is [installed](https://golang.org/doc/install)
-
-* Git clone your forked Antrea repository and `cd` into the `antrea` directory
-
-    ```bash
-    git clone https://github.com/$user/antrea
-    cd antrea
-    ```
-
-* Build the binaries for all components under `bin` directory
-
-    ```bash
-    make bin
-    ```
-
-### OpenVSwitch
-
-Open vSwitch >= 2.8.0 userspace daemon `ovs-vswitchd` and `ovsdb-server` should run on all worker nodes. See
-[Installing Open vSwitch](https://docs.openvswitch.org/en/latest/intro/install/#installation-from-packages) for details.
-
-### antrea-controller
-
-`antrea-controller` is required to implement Kubernetes Network Policies. At any time, there should be only a single
-active replica of `antrea-controller`.
-
-1. Grant the `antrea-controller` ServiceAccount necessary permissions to Kubernetes APIs. You can apply
-[controller-rbac.yaml](/build/yamls/base/controller-rbac.yml) to do it.
-
-    ```bash
-    kubectl apply -f build/yamls/base/controller-rbac.yml
-    ```
-
-2. Create the kubeconfig file that contains the K8s APIServer endpoint and the token of ServiceAccount created in the
-above step. See [Configure Access to Multiple Clusters](
-https://kubernetes.io/docs/tasks/access-application-cluster/configure-access-multiple-clusters/) for more information.
-
-    ```bash
-    APISERVER=$(kubectl config view --minify -o jsonpath='{.clusters[0].cluster.server}')
-    TOKEN=$(kubectl get secrets -n kube-system -o jsonpath="{.items[?(@.metadata.annotations['kubernetes\.io/service-account\.name']=='antrea-controller')].data.token}"|base64 --decode)
-    kubectl config --kubeconfig=antrea-controller.kubeconfig set-cluster kubernetes --server=$APISERVER --insecure-skip-tls-verify
-    kubectl config --kubeconfig=antrea-controller.kubeconfig set-credentials antrea-controller --token=$TOKEN
-    kubectl config --kubeconfig=antrea-controller.kubeconfig set-context antrea-controller@kubernetes --cluster=kubernetes --user=antrea-controller
-    kubectl config --kubeconfig=antrea-controller.kubeconfig use-context antrea-controller@kubernetes
-    ```
-
-3. Create the `antrea-controller` config file, see [Configuration](../configuration.md) for details.
-
-    ```bash
-    cat >antrea-controller.conf <<EOF
-    clientConnection:
-      kubeconfig: antrea-controller.kubeconfig
-    EOF
-    ```
-
-4. Start `antrea-controller`.
-
-    ```bash
-    bin/antrea-controller --config antrea-controller.conf
-    ```
-
-### antrea-agent
-
-`antrea-agent` must run all worker nodes.
-
-1. Grant the `antrea-agent` ServiceAccount necessary permissions to Kubernetes APIs. You can apply [agent-rbac.yaml](
-/build/yamls/base/agent-rbac.yml) to do it.
-
-    ```bash
-    kubectl apply -f build/yamls/base/agent-rbac.yml
-    ```
-
-2. Create the kubeconfig file that contains the K8s APIServer endpoint and the token of ServiceAccount created in the
-above step. See [Configure Access to Multiple Clusters](
-https://kubernetes.io/docs/tasks/access-application-cluster/configure-access-multiple-clusters/) for more information.
-
-    ```bash
-    APISERVER=$(kubectl config view --minify -o jsonpath='{.clusters[0].cluster.server}')
-    TOKEN=$(kubectl get secrets -n kube-system -o jsonpath="{.items[?(@.metadata.annotations['kubernetes\.io/service-account\.name']=='antrea-agent')].data.token}"|base64 --decode)
-    kubectl config --kubeconfig=antrea-agent.kubeconfig set-cluster kubernetes --server=$APISERVER --insecure-skip-tls-verify
-    kubectl config --kubeconfig=antrea-agent.kubeconfig set-credentials antrea-agent --token=$TOKEN
-    kubectl config --kubeconfig=antrea-agent.kubeconfig set-context antrea-agent@kubernetes --cluster=kubernetes --user=antrea-agent
-    kubectl config --kubeconfig=antrea-agent.kubeconfig use-context antrea-agent@kubernetes
-    ```
-
-3. Create the kubeconfig file that contains the `antrea-controller` APIServer endpoint and the token of ServiceAccount
-created in the above step.
-
-    ```bash
-    # Change it to the correct endpoint if you are running antrea-controller somewhere else.
-    ANTREA_APISERVER=https://localhost
-    TOKEN=$(kubectl get secrets -n kube-system -o jsonpath="{.items[?(@.metadata.annotations['kubernetes\.io/service-account\.name']=='antrea-agent')].data.token}"|base64 --decode)
-    kubectl config --kubeconfig=antrea-agent.antrea.kubeconfig set-cluster antrea --server=$ANTREA_APISERVER --insecure-skip-tls-verify
-    kubectl config --kubeconfig=antrea-agent.antrea.kubeconfig set-credentials antrea-agent --token=$TOKEN
-    kubectl config --kubeconfig=antrea-agent.antrea.kubeconfig set-context antrea-agent@antrea --cluster=antrea --user=antrea-agent
-    kubectl config --kubeconfig=antrea-agent.antrea.kubeconfig use-context antrea-agent@antrea
-    ```
-
-4. Create the `antrea-agent` config file, see [Configuration](../configuration.md) for details.
-
-    ```bash
-    cat >antrea-agent.conf <<EOF
-    clientConnection:
-      kubeconfig: antrea-agent.kubeconfig
-    antreaClientConnection:
-      kubeconfig: antrea-agent.antrea.kubeconfig
-    hostProcPathPrefix: "/"
-    EOF
-    ```
-
-5. Start `antrea-agent`.
-
-    ```bash
-    bin/antrea-agent --config antrea-agent.conf
-    ```
-
-### antrea-cni
-
-`antrea-cni` should be installed on all worker nodes.
-
-1. Create the cni config file on all worker nodes.
-
-    ```bash
-    mkdir -p /etc/cni/net.d
-
-    cat >/etc/cni/net.d/10-antrea.conflist <<EOF
-    {
-      "cniVersion":"0.3.0",
-      "name": "antrea",
-      "plugins": [
-        {
-          "type": "antrea",
-          "ipam": {
-            "type": "host-local"
-          }
-        },
-        {
-          "type": "portmap",
-          "capabilities": {"portMappings": true}
-        },
-        {
-          "type": "bandwidth",
-          "capabilities": {"bandwidth": true}
-        }
-      ]
-    }
-    EOF
-    ```
-
-2. Install `antrea-cni` to `/opt/cni/bin/antrea`.
-
-    ```bash
-    cp bin/antrea-cni /opt/cni/bin/antrea
-    ```
diff --git a/docs/getting-started.md b/docs/getting-started.md
index fc892c6e3..758832507 100644
--- a/docs/getting-started.md
+++ b/docs/getting-started.md
@@ -146,10 +146,6 @@ or simply reboot your Nodes.
 To build the image locally, you can follow the instructions in the [Contributor
 Guide](../CONTRIBUTING.md#building-and-testing-your-change).
 
-Antrea components can also be run manually as processes for development
-purposes. See [Manual Installation](contributors/manual-installation.md) for
-information.
-
 ### Deploying Antrea in Kind
 
 To deploy Antrea in a [Kind](https://github.com/kubernetes-sigs/kind) cluster,
diff --git a/docs/network-flow-visibility.md b/docs/network-flow-visibility.md
index f8ce96954..d4ff6ef16 100644
--- a/docs/network-flow-visibility.md
+++ b/docs/network-flow-visibility.md
@@ -35,7 +35,10 @@
     - [About Grafana and ClickHouse](#about-grafana-and-clickhouse)
     - [Deployment Steps](#deployment-steps-2)
       - [Credentials Configuration](#credentials-configuration)
-      - [ClickHouse Configuration](#clickhouse-configuration)
+    - [ClickHouse Configuration](#clickhouse-configuration)
+      - [Service Customization](#service-customization)
+      - [Performance Configuration](#performance-configuration)
+      - [Persistent Volumes](#persistent-volumes)
     - [Pre-built Dashboards](#pre-built-dashboards)
       - [Flow Records Dashboard](#flow-records-dashboard)
       - [Pod-to-Pod Flows Dashboard](#pod-to-pod-flows-dashboard)
@@ -778,7 +781,9 @@ type: Opaque
 We recommend changing all the credentials above if you are going to run the Flow
 Collector in production.
 
-##### ClickHouse Configuration
+#### ClickHouse Configuration
+
+##### Service Customization
 
 The ClickHouse database can be accessed through the Service `clickhouse-clickhouse`.
 The Pod exposes HTTP port at 8123 and TCP port at 9000 by default. The ports are
@@ -828,6 +833,8 @@ metadata:
   namespace: flow-visibility
 ```
 
+##### Performance Configuration
+
 The ClickHouse throughput depends on two factors - the storage size of the ClickHouse
 and the time interval between the batch commits to the ClickHouse. Larger storage
 size and longer commit interval provide higher throughput.
@@ -847,11 +854,121 @@ storage size, please modify the `sizeLimit` in the following section.
   name: clickhouse-storage-volume
 ```
 
+To deploy ClickHouse with Persistent Volumes and limited storage size, please refer
+to [Persistent Volumes](#persistent-volumes).
+
 The time interval between the batch commits to the ClickHouse is specified in the
 [Flow Aggregator Configuration](#configuration-1) as `commitInterval`. The
 ClickHouse throughput grows sightly when the commit interval grows from 1s to 8s.
 A commit interval larger than 8s provides little improvement on the throughput.
 
+##### Persistent Volumes
+
+By default, ClickHouse is deployed in memory. From Antrea v1.7, we support deploying
+ClickHouse with Persistent Volumes.
+
+[PersistentVolume](https://kubernetes.io/docs/concepts/storage/persistent-volumes/)
+(PV) is a piece of storage in the K8s cluster, which requires to be manually
+provisioned by an administrator or dynamically provisioned using Storage Classes.
+A PersistentVolumeClaim (PVC) is a request for storage which consumes PV. As
+ClickHouse is deployed as a StatefulSet, the volume can be claimed using
+`volumeClaimTemplate`.
+
+To generate the manifest automatically with default settings, you can clone the
+repository and run one of the following commands:
+
+```yaml
+# To generate a manifest with Local PV for the ClickHouse
+./hack/generate-manifest-flow-visibility.sh --volume pv --local <local_path> > flow-visibility.yml
+ 
+# To generate a manifest with NFS PV for the ClickHouse
+./hack/generate-manifest-flow-visibility.sh --volume pv --nfs <nfs_server_address>:/<nfs_server_path> > flow-visibility.yml
+ 
+# To generate a manifest with a customized StorageClass for the ClickHouse
+./hack/generate-manifest-flow-visibility.sh --volume pv --storageclass <storageclass_name> > flow-visibility.yml
+```
+
+If you prefer not to clone the repository and prefer to create a customized
+manifest manually, please follow the steps below to deploy the ClickHouse with
+Persistent Volumes:
+
+1. Provision the PersistentVolume. K8s supports a great number of
+[PersistentVolume types](https://kubernetes.io/docs/concepts/storage/persistent-volumes/#types-of-persistent-volumes).
+You can provision your own PersistentVolume per your requirements. Here are
+two simple examples for your reference.
+
+    - Local PV allows you to store the ClickHouse data at a pre-defined path on
+    a specific Node. Refer to [createLocalPv.yml][local_pv_yaml] to create the
+    PV. Please replace `LOCAL_PATH` with the path to store the ClickHouse data
+    and label the Node used to store the ClickHouse data with
+    `antrea.io/clickhouse-data-node=`.
+
+    - NFS PV allows you to store the ClickHouse data on an existing NFS server.
+    Refer to [createNfsPv.yml][nfs_pv_yaml] to create the PV. Please replace
+    `NFS_SERVER_ADDRESS` with the host name of the NFS server and `NFS_SERVER_PATH`
+    with the exported path on the NFS server.
+
+    In both examples, you can set `.spec.capacity.storage` in PersistentVolume
+    to your storage size. This value is for informative purpose as K8s does not
+    enforce the capacity of PVs. If you want to limit the storage usage, you need
+    to ask for your storage system to enforce that. For example, you can create
+    a Local PV on a partition with the limited size. We recommend using a dedicated
+    saving space for the ClickHouse if you are going to run the Flow Collector in
+    production.
+
+    As these examples do not use any dynamic provisioner, the reclaim policy
+    for the PVs is `Retain` by default. After stopping the Grafana Flow Collector,
+    if you no long need the data for future use, you may need to manually clean
+    up the data on the local disk or NFS server.
+
+1. Request the PV for ClickHouse. Please add a `volumeClaimTemplate` section
+under `.spec.templates` for the resource `ClickHouseInstallation` in
+`flow-visibility.yml` as shown in the example below. `storageClassName` should
+be set to your own `StorageClass` name, and `.resources.requests.storage`
+should be set to your storage size.
+
+    ```yaml
+    volumeClaimTemplates:
+    - name: clickhouse-storage-template
+      spec:
+        accessModes:
+        - ReadWriteOnce
+        resources:
+          requests:
+            storage: 8Gi
+        storageClassName: clickhouse-storage
+    ```
+
+    Then add this template as `dataVolumeClaimTemplate` to the section below.
+
+    ```yaml
+    defaults:
+      templates:
+        dataVolumeClaimTemplate: clickhouse-storage-template
+        podTemplate: pod-template
+        serviceTemplate: service-template
+    ```
+
+1. Remove the in-memory related deployment options, by removing the appropriate
+`volume` and `volumeMount` for the `ClickHouseInstallation` resource in
+`flow-visibility.yml`.
+
+    The `volumeMounts` entry to be removed is the following one:
+
+    ```yaml
+    - mountPath: /var/lib/clickhouse
+      name: clickhouse-storage-volume
+    ```
+
+    The `volumes` entry to be removed is the following one:
+
+    ```yaml
+    - emptyDir:
+        medium: Memory
+        sizeLimit: 8Gi
+      name: clickhouse-storage-volume
+    ```
+
 #### Pre-built Dashboards
 
 The following dashboards are pre-built and are recommended for Antrea flow
@@ -1145,3 +1262,5 @@ With filters applied:
 Visualization Network Policy Dashboard">
 
 [flow_visibility_kustomization_yaml]: ../build/yamls/flow-visibility/base/kustomization.yml
+[local_pv_yaml]: ../build/yamls/flow-visibility/patches/pv/createLocalPv.yml
+[nfs_pv_yaml]: ../build/yamls/flow-visibility/patches/pv/createNfsPv.yml
diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md
index ea449e886..65d717601 100644
--- a/docs/troubleshooting.md
+++ b/docs/troubleshooting.md
@@ -126,7 +126,7 @@ address and pass an authentication token when accessing it, like this:
 # Get the antrea Service address
 ANTREA_SVC=$(kubectl get service antrea -n kube-system -o jsonpath='{.spec.clusterIP}')
 # Get the token value of antctl account, you can use any ServiceAccount that has permissions to antrea API.
-TOKEN=$(kubectl get secrets -n kube-system -o jsonpath="{.items[?(@.metadata.annotations['kubernetes\.io/service-account\.name']=='antctl')].data.token}"|base64 --decode)
+TOKEN=$(kubectl get secret/antctl-service-account-token -n kube-system -o jsonpath="{.data.token}"|base64 --decode)
 # Access antrea API with TOKEN
 curl --insecure --header "Authorization: Bearer $TOKEN" https://$ANTREA_SVC/apis
 ```
@@ -176,13 +176,14 @@ using the authentication token of the `antctl` ServiceAccount:
 
 ```bash
 # Get the token value of antctl account.
-TOKEN=$(kubectl get secrets -n kube-system -o jsonpath="{.items[?(@.metadata.annotations['kubernetes\.io/service-account\.name']=='antctl')].data.token}"|base64 --decode)
+TOKEN=$(kubectl get secret/antctl-service-account-token -n kube-system -o jsonpath="{.data.token}"|base64 --decode)
 # Access antrea API with TOKEN
 curl --insecure --header "Authorization: Bearer $TOKEN" https://<Node IP address>:10350/podinterfaces
 ```
 
 However, in this case you will be limited to the endpoints that `antctl` is
-allowed to access, as defined [here](/build/yamls/base/antctl.yml).
+allowed to access, as defined
+[here](../build/charts/antrea/templates/antctl/clusterrole.yaml).
 
 ## Accessing the flow-aggregator API
 
diff --git a/hack/.notableofcontents b/hack/.notableofcontents
index 6f9811354..720e2f957 100644
--- a/hack/.notableofcontents
+++ b/hack/.notableofcontents
@@ -9,7 +9,6 @@ docs/contributors/code-generation.md
 docs/contributors/docker-desktop-alternatives.md
 docs/contributors/eks-terraform.md
 docs/contributors/github-labels.md
-docs/contributors/manual-installation.md
 docs/configuration.md
 docs/cookbooks/multus/README.md
 docs/cookbooks/multus/build/cni-dhcp-daemon/README.md
diff --git a/hack/generate-manifest-flow-visibility.sh b/hack/generate-manifest-flow-visibility.sh
index ddd312a89..5dd22fa60 100755
--- a/hack/generate-manifest-flow-visibility.sh
+++ b/hack/generate-manifest-flow-visibility.sh
@@ -21,17 +21,28 @@ function echoerr {
 }
 
 _usage="Usage: $0 [--mode (dev|release|e2e)] [--keep] [--help|-h]
-Generate a YAML manifest for the Clickhouse-Grafana Flow-visibility Solution, using Kustomize, and
+Generate a YAML manifest for the ClickHouse-Grafana Flow-visibility Solution, using Kustomize, and
 print it to stdout.
-        --mode (dev|release|e2e)  Choose the configuration variant that you need (default is 'dev')
-                                  e2e mode generates YAML manifest for e2e test, which includes
-                                  clickhouse operator and server with default credentials,
-                                  but not Grafana-related functionality
-        --keep                    Debug flag which will preserve the generated kustomization.yml
-        --help, -h                Print this message and exit
-
+        --mode (dev|release|e2e)    Choose the configuration variant that you need (default is 'dev')
+                                    e2e mode generates YAML manifest for e2e test, which includes
+                                    ClickHouse operator and server with default credentials,
+                                    but not Grafana-related functionality and ClickHouse monitor.
+        --keep                      Debug flag which will preserve the generated kustomization.yml
+        --volume (ram|pv)           Choose the volume provider that you need (default is 'ram').
+        --storageclass -sc <name>   Provide the StorageClass used to dynamically provision the 
+                                    PersistentVolume for ClickHouse storage.
+        --local <path>              Create the PersistentVolume for ClickHouse with a provided
+                                    local path.
+        --nfs <hostname:path>       Create the PersistentVolume for ClickHouse with a provided
+                                    NFS server hostname or IP address and the path exported in the
+                                    form of hostname:path.
+        --size <size>               Deploy the ClickHouse with a specific storage size. Can be a 
+                                    plain integer or as a fixed-point number using one of these quantity
+                                    suffixes: E, P, T, G, M, K. Or the power-of-two equivalents:
+                                    Ei, Pi, Ti, Gi, Mi, Ki.  The default is 8Gi.
+        --help, -h                  Print this message and exit
 This tool uses kustomize (https://github.com/kubernetes-sigs/kustomize) to generate manifests for
-Clickhouse-Grafana Flow-visibility Solution. You can set the KUSTOMIZE environment variable to the
+ClickHouse-Grafana Flow-visibility Solution. You can set the KUSTOMIZE environment variable to the
 path of the kustomize binary you want us to use. Otherwise we will look for kustomize in your PATH
 and your GOPATH. If we cannot find kustomize there, we will try to install it."
 
@@ -45,11 +56,15 @@ function print_help {
 
 MODE="dev"
 KEEP=false
+VOLUME="ram"
+STORAGECLASS=""
+LOCALPATH=""
+NFSPATH=""
+SIZE="8Gi"
 
 while [[ $# -gt 0 ]]
 do
 key="$1"
-
 case $key in
     --mode)
     MODE="$2"
@@ -59,6 +74,26 @@ case $key in
     KEEP=true
     shift
     ;;
+    --volume)
+    VOLUME="$2"
+    shift 2
+    ;;
+    -sc|--storageclass)
+    STORAGECLASS="$2"
+    shift 2
+    ;;
+    --local)
+    LOCALPATH="$2"
+    shift 2
+    ;;
+    --nfs)
+    NFSPATH="$2"
+    shift 2
+    ;;
+    --size)
+    SIZE="$2"
+    shift 2
+    ;;
     -h|--help)
     print_usage
     exit 0
@@ -71,7 +106,7 @@ esac
 done
 
 if [ "$MODE" != "dev" ] && [ "$MODE" != "release" ] && [ "$MODE" != "e2e" ]; then
-    echoerr "--mode must be one of 'dev' or 'release'"
+    echoerr "--mode must be one of 'dev', 'release' or 'e2e'"
     print_help
     exit 1
 fi
@@ -88,6 +123,33 @@ if [ "$MODE" == "release" ] && [ -z "$IMG_TAG" ]; then
     exit 1
 fi
 
+if [ "$VOLUME" != "ram" ] && [ "$VOLUME" != "pv" ]; then
+    echoerr "--volume must be one of 'ram' or 'pv'"
+    print_help
+    exit 1
+fi
+
+if [ "$VOLUME" == "pv" ] && [ "$LOCALPATH" == "" ] && [ "$NFSPATH" == "" ] && [ "$STORAGECLASS" == "" ]; then
+    echoerr "When deploying with 'pv', one of '--local', '--nfs', '--storageclass' should be set"
+    print_help
+    exit 1
+fi
+
+if ([ "$LOCALPATH" != "" ] && [ "$NFSPATH" != "" ]) || ([ "$LOCALPATH" != "" ] && [ "$STORAGECLASS" != "" ]) || ([ "$STORAGECLASS" != "" ] && [ "$NFSPATH" != "" ]); then
+    echoerr "Cannot set '--local', '--nfs' or '--storageclass' at the same time"
+    print_help
+    exit 1
+fi
+
+if [ "$NFSPATH" != "" ]; then
+    pathPair=(${NFSPATH//:/ })
+    if [ ${#pathPair[@]} != 2 ]; then
+        echoerr "--nfs must be in the form of hostname:path"
+        print_help
+        exit 1
+    fi
+fi
+
 THIS_DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" >/dev/null 2>&1 && pwd )"
 
 source $THIS_DIR/verify-kustomize.sh
@@ -120,12 +182,14 @@ if [ "$MODE" == "e2e" ]; then
     cp $KUSTOMIZATION_DIR/base/kustomize-config.yml base/kustomize-config.yml
     cp $KUSTOMIZATION_DIR/base/provisioning/datasources/create_table.sh base/provisioning/datasources/create_table.sh
     cp $KUSTOMIZATION_DIR/../clickhouse-operator-install-bundle.yml clickhouse-operator-install-bundle.yml
-
     $KUSTOMIZE edit add base base
-    $KUSTOMIZE edit set image flow-visibility-clickhouse-monitor=projects.registry.vmware.com/antrea/flow-visibility-clickhouse-monitor:latest
     $KUSTOMIZE edit add patch --path imagePullPolicyClickhouse.yml --group clickhouse.altinity.com --version v1 --kind ClickHouseInstallation --name clickhouse
 else
+    # patch the clickhouse monitor with desired storage size
+    cp $KUSTOMIZATION_DIR/patches/chmonitor/*.yml .
     $KUSTOMIZE edit add base $BASE
+    sed -i.bak -E "s/STORAGE_SIZE_VALUE/$SIZE/" chMonitor.yml
+    $KUSTOMIZE edit add patch --path chMonitor.yml --group clickhouse.altinity.com --version v1 --kind ClickHouseInstallation --name clickhouse
 fi
 
 if [ "$MODE" == "dev" ]; then
@@ -136,6 +200,43 @@ fi
 if [ "$MODE" == "release" ]; then
     $KUSTOMIZE edit set image flow-visibility-clickhouse-monitor=$IMG_NAME:$IMG_TAG
 fi
+BASE=../$MODE
+cd ..
+
+if [ "$VOLUME" == "ram" ]; then
+    mkdir ram && cd ram
+    cp $KUSTOMIZATION_DIR/patches/ram/*.yml .
+    touch kustomization.yml
+    $KUSTOMIZE edit add base $BASE
+    sed -i.bak -E "s/STORAGE_SIZE/$SIZE/" mountRam.yml
+    $KUSTOMIZE edit add patch --path mountRam.yml --group clickhouse.altinity.com --version v1 --kind ClickHouseInstallation --name clickhouse
+fi
+
+if [ "$VOLUME" == "pv" ]; then
+    mkdir pv && cd pv
+    cp $KUSTOMIZATION_DIR/patches/pv/*.yml .
+    touch kustomization.yml
+    $KUSTOMIZE edit add base $BASE
+
+    if [[ $STORAGECLASS != "" ]]; then
+        sed -i.bak -E "s/STORAGECLASS_NAME/$STORAGECLASS/" mountPv.yml
+    else
+        sed -i.bak -E "s/STORAGECLASS_NAME/clickhouse-storage/" mountPv.yml
+    fi
+    if [[ $LOCALPATH != "" ]]; then
+        sed -i.bak -E "s~LOCAL_PATH~$LOCALPATH~" createLocalPv.yml
+        sed -i.bak -E "s/STORAGE_SIZE/$SIZE/" createLocalPv.yml
+        $KUSTOMIZE edit add base createLocalPv.yml
+    fi
+    if [[ $NFSPATH != "" ]]; then
+        sed -i.bak -E "s~NFS_SERVER_ADDRESS~${pathPair[0]}~" createNfsPv.yml
+        sed -i.bak -E "s~NFS_SERVER_PATH~${pathPair[1]}~" createNfsPv.yml
+        sed -i.bak -E "s/STORAGE_SIZE/$SIZE/" createNfsPv.yml
+        $KUSTOMIZE edit add base createNfsPv.yml
+    fi
+    sed -i.bak -E "s/STORAGE_SIZE/$SIZE/" mountPv.yml
+    $KUSTOMIZE edit add patch --path mountPv.yml --group clickhouse.altinity.com --version v1 --kind ClickHouseInstallation --name clickhouse
+fi
 
 $KUSTOMIZE build
 
diff --git a/hack/windows/Helper.psm1 b/hack/windows/Helper.psm1
index 7f6ac3371..be7768113 100644
--- a/hack/windows/Helper.psm1
+++ b/hack/windows/Helper.psm1
@@ -127,7 +127,7 @@ function Install-AntreaAgent {
     $APIServer=$(kubectl --kubeconfig=$KubeConfig get service kubernetes -o jsonpath='{.spec.clusterIP}')
     $APIServerPort=$(kubectl --kubeconfig=$KubeConfig get service kubernetes -o jsonpath='{.spec.ports[0].port}')
     $APIServer="https://$APIServer" + ":" + $APIServerPort
-    $TOKEN=$(kubectl --kubeconfig=$KubeConfig get secrets -n kube-system -o jsonpath="{.items[?(@.metadata.annotations['kubernetes\.io/service-account\.name']=='antrea-agent')].data.token}")
+    $TOKEN=$(kubectl --kubeconfig=$KubeConfig get secret/antrea-agent-service-account-token -n kube-system -o jsonpath="{.data.token}")
     $TOKEN=$([System.Text.Encoding]::UTF8.GetString([System.Convert]::FromBase64String($TOKEN)))
     kubectl config --kubeconfig=$AntreaEtc\antrea-agent.kubeconfig set-cluster kubernetes --server=$APIServer --insecure-skip-tls-verify
     kubectl config --kubeconfig=$AntreaEtc\antrea-agent.kubeconfig set-credentials antrea-agent --token=$TOKEN
diff --git a/pkg/agent/cniserver/pod_configuration.go b/pkg/agent/cniserver/pod_configuration.go
index 0c2027df6..f5d68c530 100644
--- a/pkg/agent/cniserver/pod_configuration.go
+++ b/pkg/agent/cniserver/pod_configuration.go
@@ -32,6 +32,7 @@ import (
 	"antrea.io/antrea/pkg/agent/openflow"
 	"antrea.io/antrea/pkg/agent/route"
 	"antrea.io/antrea/pkg/agent/secondarynetwork/cnipodcache"
+	agenttypes "antrea.io/antrea/pkg/agent/types"
 	"antrea.io/antrea/pkg/agent/util"
 	"antrea.io/antrea/pkg/ovs/ovsconfig"
 	"antrea.io/antrea/pkg/util/channel"
@@ -312,6 +313,7 @@ func (pc *podConfigurator) removeInterfaces(containerID string) error {
 	if err := pc.routeClient.DeleteLocalAntreaFlexibleIPAMPodRule(containerConfig.IPs); err != nil {
 		return err
 	}
+
 	return nil
 }
 
@@ -495,7 +497,13 @@ func (pc *podConfigurator) connectInterfaceToOVSCommon(ovsPortName string, conta
 	// Add containerConfig into local cache
 	pc.ifaceStore.AddInterface(containerConfig)
 	// Notify the Pod update event to required components.
-	pc.podUpdateNotifier.Notify(k8s.NamespacedName(containerConfig.PodNamespace, containerConfig.PodName))
+	event := agenttypes.PodUpdate{
+		PodName:      containerConfig.PodName,
+		PodNamespace: containerConfig.PodNamespace,
+		IsAdd:        true,
+		ContainerID:  containerConfig.ContainerID,
+	}
+	pc.podUpdateNotifier.Notify(event)
 	return nil
 }
 
@@ -518,6 +526,13 @@ func (pc *podConfigurator) disconnectInterfaceFromOVS(containerConfig *interface
 	}
 	// Remove container configuration from cache.
 	pc.ifaceStore.DeleteInterface(containerConfig)
+	event := agenttypes.PodUpdate{
+		PodName:      containerConfig.PodName,
+		PodNamespace: containerConfig.PodNamespace,
+		IsAdd:        false,
+		ContainerID:  containerConfig.ContainerID,
+	}
+	pc.podUpdateNotifier.Notify(event)
 	klog.Infof("Removed interfaces for container %s", containerID)
 	return nil
 }
diff --git a/pkg/agent/cniserver/pod_configuration_windows.go b/pkg/agent/cniserver/pod_configuration_windows.go
index a2dc28340..16f824f84 100644
--- a/pkg/agent/cniserver/pod_configuration_windows.go
+++ b/pkg/agent/cniserver/pod_configuration_windows.go
@@ -24,6 +24,7 @@ import (
 	"k8s.io/klog/v2"
 
 	"antrea.io/antrea/pkg/agent/interfacestore"
+	"antrea.io/antrea/pkg/agent/types"
 	"antrea.io/antrea/pkg/agent/util"
 	"antrea.io/antrea/pkg/util/k8s"
 )
@@ -49,7 +50,13 @@ func (pc *podConfigurator) connectInterfaceToOVSAsync(ifConfig *interfacestore.I
 		// Update interface config with the ofPort.
 		ifConfig.OVSPortConfig.OFPort = ofPort
 		// Notify the Pod update event to required components.
-		pc.podUpdateNotifier.Notify(k8s.NamespacedName(ifConfig.PodNamespace, ifConfig.PodName))
+		event := types.PodUpdate{
+			PodName:      ifConfig.PodName,
+			PodNamespace: ifConfig.PodNamespace,
+			IsAdd:        true,
+			ContainerID:  ifConfig.ContainerID,
+		}
+		pc.podUpdateNotifier.Notify(event)
 		return nil
 	})
 }
diff --git a/pkg/agent/controller/egress/egress_controller.go b/pkg/agent/controller/egress/egress_controller.go
index 167d0756a..a683a8d84 100644
--- a/pkg/agent/controller/egress/egress_controller.go
+++ b/pkg/agent/controller/egress/egress_controller.go
@@ -41,6 +41,7 @@ import (
 	"antrea.io/antrea/pkg/agent/memberlist"
 	"antrea.io/antrea/pkg/agent/openflow"
 	"antrea.io/antrea/pkg/agent/route"
+	"antrea.io/antrea/pkg/agent/types"
 	cpv1b2 "antrea.io/antrea/pkg/apis/controlplane/v1beta2"
 	crdv1a2 "antrea.io/antrea/pkg/apis/crd/v1alpha2"
 	clientsetversioned "antrea.io/antrea/pkg/client/clientset/versioned"
@@ -217,9 +218,11 @@ func NewEgressController(
 
 // processPodUpdate will be called when CNIServer publishes a Pod update event.
 // It triggers reconciling the effective Egress of the Pod.
-func (c *EgressController) processPodUpdate(pod string) {
+func (c *EgressController) processPodUpdate(e interface{}) {
 	c.egressBindingsMutex.Lock()
 	defer c.egressBindingsMutex.Unlock()
+	podEvent := e.(types.PodUpdate)
+	pod := k8s.NamespacedName(podEvent.PodNamespace, podEvent.PodName)
 	binding, exists := c.egressBindings[pod]
 	if !exists {
 		return
diff --git a/pkg/agent/controller/networkpolicy/cache.go b/pkg/agent/controller/networkpolicy/cache.go
index 32011eaf8..e0fc264f2 100644
--- a/pkg/agent/controller/networkpolicy/cache.go
+++ b/pkg/agent/controller/networkpolicy/cache.go
@@ -28,6 +28,7 @@ import (
 	"k8s.io/klog/v2"
 
 	"antrea.io/antrea/pkg/agent/metrics"
+	agenttypes "antrea.io/antrea/pkg/agent/types"
 	v1beta "antrea.io/antrea/pkg/apis/controlplane/v1beta2"
 	crdv1alpha1 "antrea.io/antrea/pkg/apis/crd/v1alpha1"
 	"antrea.io/antrea/pkg/querier"
@@ -361,12 +362,12 @@ func newRuleCache(dirtyRuleHandler func(string), podUpdateSubscriber channel.Sub
 // done if antrea-controller has computed the Pods' policies and propagated
 // them to this Node by their labels and NodeName, instead of waiting for their
 // IPs are reported to kube-apiserver and processed by antrea-controller.
-func (c *ruleCache) processPodUpdate(pod string) {
-	namespace, name := k8s.SplitNamespacedName(pod)
+func (c *ruleCache) processPodUpdate(e interface{}) {
+	podEvent := e.(agenttypes.PodUpdate)
 	member := &v1beta.GroupMember{
 		Pod: &v1beta.PodReference{
-			Name:      name,
-			Namespace: namespace,
+			Name:      podEvent.PodName,
+			Namespace: podEvent.PodNamespace,
 		},
 	}
 	c.appliedToSetLock.RLock()
diff --git a/pkg/agent/multicast/mcast_controller.go b/pkg/agent/multicast/mcast_controller.go
index c11a92337..481dcebb2 100644
--- a/pkg/agent/multicast/mcast_controller.go
+++ b/pkg/agent/multicast/mcast_controller.go
@@ -28,7 +28,11 @@ import (
 	"antrea.io/antrea/pkg/agent/config"
 	"antrea.io/antrea/pkg/agent/interfacestore"
 	"antrea.io/antrea/pkg/agent/openflow"
+	"antrea.io/antrea/pkg/agent/types"
+	"antrea.io/antrea/pkg/agent/util"
+	binding "antrea.io/antrea/pkg/ovs/openflow"
 	"antrea.io/antrea/pkg/ovs/ovsconfig"
+	"antrea.io/antrea/pkg/util/channel"
 )
 
 type eventType uint8
@@ -59,10 +63,11 @@ type mcastGroupEvent struct {
 
 type GroupMemberStatus struct {
 	group net.IP
-	// localMembers is a set for the local Pod's interface name which has joined in the multicast group.
-	localMembers   sets.String
+	// localMembers is a map for the local Pod member and its last update time, key is the Pod's interface name,
+	// and value is its last update time.
+	localMembers   map[string]time.Time
 	lastIGMPReport time.Time
-	mutex          sync.RWMutex
+	ofGroupID      binding.GroupIDType
 }
 
 // eventHandler process the multicast Group membership report or leave messages.
@@ -82,7 +87,8 @@ func (c *Controller) addGroupMemberStatus(e *mcastGroupEvent) {
 	status := &GroupMemberStatus{
 		group:          e.group,
 		lastIGMPReport: e.time,
-		localMembers:   sets.NewString(e.iface.InterfaceName),
+		localMembers:   map[string]time.Time{e.iface.InterfaceName: e.time},
+		ofGroupID:      c.v4GroupAllocator.Allocate(),
 	}
 	c.groupCache.Add(status)
 	c.queue.Add(e.group.String())
@@ -97,29 +103,37 @@ func (c *Controller) addGroupMemberStatus(e *mcastGroupEvent) {
 // is sent out to check if there are still local members in the group.
 func (c *Controller) updateGroupMemberStatus(obj interface{}, e *mcastGroupEvent) {
 	status := obj.(*GroupMemberStatus)
-	status.mutex.Lock()
-	defer status.mutex.Unlock()
 	newStatus := &GroupMemberStatus{
 		group:          status.group,
-		localMembers:   status.localMembers.Union(nil),
+		localMembers:   make(map[string]time.Time),
 		lastIGMPReport: status.lastIGMPReport,
+		ofGroupID:      status.ofGroupID,
 	}
-	exist := status.localMembers.Has(e.iface.InterfaceName)
+	for m, t := range status.localMembers {
+		newStatus.localMembers[m] = t
+	}
+	_, exist := status.localMembers[e.iface.InterfaceName]
 	switch e.eType {
 	case groupJoin:
 		newStatus.lastIGMPReport = e.time
+		newStatus.localMembers[e.iface.InterfaceName] = e.time
+		c.groupCache.Update(newStatus)
 		if !exist {
-			newStatus.localMembers.Insert(e.iface.InterfaceName)
 			klog.InfoS("Added member to multicast group", "group", e.group.String(), "member", e.iface.InterfaceName)
+			c.queue.Add(newStatus.group.String())
 		}
-		c.groupCache.Update(newStatus)
-		c.queue.Add(newStatus.group.String())
 	case groupLeave:
 		if exist {
-			newStatus.localMembers.Delete(e.iface.InterfaceName)
+			delete(newStatus.localMembers, e.iface.InterfaceName)
 			c.groupCache.Update(newStatus)
 			klog.InfoS("Deleted member from multicast group", "group", e.group.String(), "member", e.iface.InterfaceName)
-			if len(newStatus.localMembers) == 0 {
+			_, found := c.ifaceStore.GetInterfaceByName(e.iface.InterfaceName)
+			// Notify worker immediately about the member leave event if the member doesn't exist on the Node, or there are
+			// other local members in the multicast group.
+			if !found || len(newStatus.localMembers) > 0 {
+				c.queue.Add(newStatus.group.String())
+			} else {
+				// Check if all local members have left the multicast group.
 				klog.InfoS("Check last member in multicast group", "group", e.group.String(), "member", e.iface.InterfaceName)
 				c.checkLastMember(e.group)
 			}
@@ -139,28 +153,70 @@ func (c *Controller) checkLastMember(group net.IP) {
 	c.queue.AddAfter(group.String(), igmpMaxResponseTime)
 }
 
-// clearStaleGroups checks the stale groups which have not been updated for mcastGroupTimeout, and then notifies worker
+// clearStaleGroups checks the stale group members which have not been updated for mcastGroupTimeout, and then notifies worker
 // to remove them from groupCache.
 func (c *Controller) clearStaleGroups() {
 	now := time.Now()
 	for _, obj := range c.groupCache.List() {
 		status := obj.(*GroupMemberStatus)
-		status.mutex.RLock()
 		diff := now.Sub(status.lastIGMPReport)
-		status.mutex.RUnlock()
 		if diff > mcastGroupTimeout {
+			// Notify worker to remove the group from groupCache if all its members are not updated before mcastGroupTimeout.
 			c.queue.Add(status.group.String())
+		} else {
+			// Create a "leave" event for a local member if it is not updated before mcastGroupTimeout.
+			for member, lastUpdate := range status.localMembers {
+				if now.Sub(lastUpdate) > mcastGroupTimeout {
+					ifConfig := &interfacestore.InterfaceConfig{
+						InterfaceName: member,
+					}
+					event := &mcastGroupEvent{
+						group: status.group,
+						eType: groupLeave,
+						time:  now,
+						iface: ifConfig,
+					}
+					c.groupEventCh <- event
+				}
+			}
 		}
 	}
 }
 
+// removeLocalInterface searches the GroupMemberStatus which the deleted interface has joined, and then triggers a member
+// leave event so that Antrea can remove the corresponding interface from local multicast receivers on OVS. This function
+// should be called if the removed Pod receiver fails to send IGMP leave message before deletion.
+func (c *Controller) removeLocalInterface(e interface{}) {
+	podEvent := e.(types.PodUpdate)
+	// Ignore Pod creation event.
+	if podEvent.IsAdd {
+		return
+	}
+	interfaceName := util.GenerateContainerInterfaceName(podEvent.PodName, podEvent.PodNamespace, podEvent.ContainerID)
+	ifConfig := &interfacestore.InterfaceConfig{
+		InterfaceName: interfaceName,
+	}
+	groupStatuses := c.getGroupMemberStatusesByPod(interfaceName)
+	for _, g := range groupStatuses {
+		event := &mcastGroupEvent{
+			group: g.group,
+			eType: groupLeave,
+			time:  time.Now(),
+			iface: ifConfig,
+		}
+		c.groupEventCh <- event
+	}
+}
+
 type Controller struct {
-	ofClient     openflow.Client
-	nodeConfig   *config.NodeConfig
-	igmpSnooper  *IGMPSnooper
-	groupEventCh chan *mcastGroupEvent
-	groupCache   cache.Indexer
-	queue        workqueue.RateLimitingInterface
+	ofClient         openflow.Client
+	v4GroupAllocator openflow.GroupAllocator
+	ifaceStore       interfacestore.InterfaceStore
+	nodeConfig       *config.NodeConfig
+	igmpSnooper      *IGMPSnooper
+	groupEventCh     chan *mcastGroupEvent
+	groupCache       cache.Indexer
+	queue            workqueue.RateLimitingInterface
 	// installedGroups saves the groups which are configured on both OVS and the host.
 	installedGroups      sets.String
 	installedGroupsMutex sync.RWMutex
@@ -168,33 +224,39 @@ type Controller struct {
 	ovsBridgeClient      ovsconfig.OVSBridgeClient
 }
 
-func NewMulticastController(ofClient openflow.Client, nodeConfig *config.NodeConfig, ifaceStore interfacestore.InterfaceStore, multicastSocket RouteInterface, multicastInterfaces sets.String, ovsBridgeClient ovsconfig.OVSBridgeClient) *Controller {
+func NewMulticastController(ofClient openflow.Client,
+	v4GroupAllocator openflow.GroupAllocator,
+	nodeConfig *config.NodeConfig,
+	ifaceStore interfacestore.InterfaceStore,
+	multicastSocket RouteInterface,
+	multicastInterfaces sets.String,
+	ovsBridgeClient ovsconfig.OVSBridgeClient,
+	podUpdateSubscriber channel.Subscriber) *Controller {
 	eventCh := make(chan *mcastGroupEvent, workerCount)
 	groupSnooper := newSnooper(ofClient, ifaceStore, eventCh)
 	groupCache := cache.NewIndexer(getGroupEventKey, cache.Indexers{
 		podInterfaceIndex: podInterfaceIndexFunc,
 	})
 	multicastRouteClient := newRouteClient(nodeConfig, groupCache, multicastSocket, multicastInterfaces)
-	return &Controller{
-		ofClient:        ofClient,
-		nodeConfig:      nodeConfig,
-		igmpSnooper:     groupSnooper,
-		groupEventCh:    eventCh,
-		groupCache:      groupCache,
-		installedGroups: sets.NewString(),
-		queue:           workqueue.NewNamedRateLimitingQueue(workqueue.NewItemExponentialFailureRateLimiter(minRetryDelay, maxRetryDelay), "multicastgroup"),
-		mRouteClient:    multicastRouteClient,
-		ovsBridgeClient: ovsBridgeClient,
+	c := &Controller{
+		ofClient:         ofClient,
+		ifaceStore:       ifaceStore,
+		v4GroupAllocator: v4GroupAllocator,
+		nodeConfig:       nodeConfig,
+		igmpSnooper:      groupSnooper,
+		groupEventCh:     eventCh,
+		groupCache:       groupCache,
+		installedGroups:  sets.NewString(),
+		queue:            workqueue.NewNamedRateLimitingQueue(workqueue.NewItemExponentialFailureRateLimiter(minRetryDelay, maxRetryDelay), "multicastgroup"),
+		mRouteClient:     multicastRouteClient,
+		ovsBridgeClient:  ovsBridgeClient,
 	}
+	podUpdateSubscriber.Subscribe(c.removeLocalInterface)
+	return c
 }
 
 func (c *Controller) Initialize() error {
-	err := c.configureOVSMulticast()
-	if err != nil {
-		klog.ErrorS(err, "Failed to configure multicast for the OVS bridge")
-		return err
-	}
-	err = c.mRouteClient.Initialize()
+	err := c.mRouteClient.Initialize()
 	if err != nil {
 		return err
 	}
@@ -221,7 +283,6 @@ func (c *Controller) Run(stopCh <-chan struct{}) {
 
 	// Periodically check the group member status, and remove the groups in which no members exist
 	go wait.NonSlidingUntil(c.clearStaleGroups, queryInterval, stopCh)
-
 	go c.eventHandler(stopCh)
 
 	for i := 0; i < int(workerCount); i++ {
@@ -298,15 +359,28 @@ func (c *Controller) syncGroup(groupKey string) error {
 		return nil
 	}
 	status := obj.(*GroupMemberStatus)
+	memberPorts := make([]uint32, 0, len(status.localMembers))
+	for memberInterfaceName := range status.localMembers {
+		obj, found := c.ifaceStore.GetInterfaceByName(memberInterfaceName)
+		if !found {
+			klog.InfoS("Failed to find interface from cache", "interface", memberInterfaceName)
+			continue
+		}
+		memberPorts = append(memberPorts, uint32(obj.OFPort))
+	}
 	if c.groupHasInstalled(groupKey) {
-		status.mutex.Lock()
-		defer status.mutex.Unlock()
 		if c.groupIsStale(status) {
 			// Remove the multicast flow entry if no local Pod is in the group.
-			if err := c.ofClient.UninstallMulticastFlow(status.group); err != nil {
+			if err := c.ofClient.UninstallMulticastFlows(status.group); err != nil {
 				klog.ErrorS(err, "Failed to uninstall multicast flows", "group", groupKey)
 				return err
 			}
+			// Remove the multicast flow entry if no local Pod is in the group.
+			if err := c.ofClient.UninstallGroup(status.ofGroupID); err != nil {
+				klog.ErrorS(err, "Failed to uninstall multicast group", "group", groupKey)
+				return err
+			}
+			c.v4GroupAllocator.Release(status.ofGroupID)
 			err := c.mRouteClient.deleteInboundMrouteEntryByGroup(status.group)
 			if err != nil {
 				klog.ErrorS(err, "Cannot delete multicast group", "group", groupKey)
@@ -320,10 +394,22 @@ func (c *Controller) syncGroup(groupKey string) error {
 			c.delInstalledGroup(groupKey)
 			c.groupCache.Delete(status)
 			klog.InfoS("Removed multicast group from cache after all members left", "group", groupKey)
+			return nil
 		}
+		// Reinstall OpenFlow group because the local Pod receivers have changed.
+		if err := c.ofClient.InstallMulticastGroup(status.ofGroupID, memberPorts); err != nil {
+			return err
+		}
+		klog.V(2).InfoS("Updated OpenFlow group for local receivers", "group", groupKey, "ofGroup", status.ofGroupID, "localReceivers", memberPorts)
 		return nil
 	}
-	if err := c.ofClient.InstallMulticastFlow(status.group); err != nil {
+	// Install OpenFlow group for a new multicast group which has local Pod receivers joined.
+	if err := c.ofClient.InstallMulticastGroup(status.ofGroupID, memberPorts); err != nil {
+		return err
+	}
+	klog.V(2).InfoS("Installed OpenFlow group for local receivers", "group", groupKey, "ofGroup", status.ofGroupID, "localReceivers", memberPorts)
+	// Install OpenFlow flow to forward packets to local Pod receivers which are included in the group.
+	if err := c.ofClient.InstallMulticastFlows(status.group, status.ofGroupID); err != nil {
 		klog.ErrorS(err, "Failed to install multicast flows", "group", status.group)
 		return err
 	}
@@ -388,21 +474,6 @@ func podInterfaceIndexFunc(obj interface{}) ([]string, error) {
 	return podInterfaces, nil
 }
 
-func (c *Controller) configureOVSMulticast() error {
-	// Configure bridge to enable multicast snooping.
-	err := c.ovsBridgeClient.SetBridgeMcastSnooping(true)
-	if err != nil {
-		return err
-	}
-	// Disable flooding of unregistered multicast packets to all ports.
-	otherConfig := map[string]interface{}{"mcast-snooping-disable-flood-unregistered": "true"}
-	err = c.ovsBridgeClient.AddBridgeOtherConfig(otherConfig)
-	if err != nil {
-		return err
-	}
-	return nil
-}
-
 func getGroupEventKey(obj interface{}) (string, error) {
 	groupState := obj.(*GroupMemberStatus)
 	return groupState.group.String(), nil
diff --git a/pkg/agent/multicast/mcast_discovery.go b/pkg/agent/multicast/mcast_discovery.go
index 5f418513c..7d4b7fe91 100644
--- a/pkg/agent/multicast/mcast_discovery.go
+++ b/pkg/agent/multicast/mcast_discovery.go
@@ -131,7 +131,7 @@ func (s *IGMPSnooper) processPacketIn(pktIn *ofctrl.PacketIn) error {
 		fallthrough
 	case protocol.IGMPv2Report:
 		mgroup := igmp.(*protocol.IGMPv1or2).GroupAddress
-		klog.InfoS("Received IGMPv1or2 Report message", "group", mgroup.String(), "interface", iface.InterfaceName, "pod", podName)
+		klog.V(2).InfoS("Received IGMPv1or2 Report message", "group", mgroup.String(), "interface", iface.InterfaceName, "pod", podName)
 		event := &mcastGroupEvent{
 			group: mgroup,
 			eType: groupJoin,
@@ -143,7 +143,7 @@ func (s *IGMPSnooper) processPacketIn(pktIn *ofctrl.PacketIn) error {
 		msg := igmp.(*protocol.IGMPv3MembershipReport)
 		for _, gr := range msg.GroupRecords {
 			mgroup := gr.MulticastAddress
-			klog.InfoS("Received IGMPv3 Report message", "group", mgroup.String(), "interface", iface.InterfaceName, "pod", podName, "recordType", gr.Type, "sourceCount", gr.NumberOfSources)
+			klog.V(2).InfoS("Received IGMPv3 Report message", "group", mgroup.String(), "interface", iface.InterfaceName, "pod", podName, "recordType", gr.Type, "sourceCount", gr.NumberOfSources)
 			evtType := groupJoin
 			if (gr.Type == protocol.IGMPIsIn || gr.Type == protocol.IGMPToIn) && gr.NumberOfSources == 0 {
 				evtType = groupLeave
@@ -159,7 +159,7 @@ func (s *IGMPSnooper) processPacketIn(pktIn *ofctrl.PacketIn) error {
 
 	case protocol.IGMPv2LeaveGroup:
 		mgroup := igmp.(*protocol.IGMPv1or2).GroupAddress
-		klog.InfoS("Received IGMPv2 Leave message", "group", mgroup.String(), "interface", iface.InterfaceName, "pod", podName)
+		klog.V(2).InfoS("Received IGMPv2 Leave message", "group", mgroup.String(), "interface", iface.InterfaceName, "pod", podName)
 		event := &mcastGroupEvent{
 			group: mgroup,
 			eType: groupLeave,
diff --git a/pkg/agent/openflow/client.go b/pkg/agent/openflow/client.go
index 076306829..28980801b 100644
--- a/pkg/agent/openflow/client.go
+++ b/pkg/agent/openflow/client.go
@@ -83,9 +83,9 @@ type Client interface {
 	// InstallServiceGroup installs a group for Service LB. Each endpoint
 	// is a bucket of the group. For now, each bucket has the same weight.
 	InstallServiceGroup(groupID binding.GroupIDType, withSessionAffinity bool, endpoints []proxy.Endpoint) error
-	// UninstallServiceGroup removes the group and its buckets that are
-	// installed by InstallServiceGroup.
-	UninstallServiceGroup(groupID binding.GroupIDType) error
+	// UninstallGroup removes the group and its buckets that are
+	// installed by InstallServiceGroup or InstallMulticastGroup.
+	UninstallGroup(groupID binding.GroupIDType) error
 
 	// InstallEndpointFlows installs flows for accessing Endpoints.
 	// If an Endpoint is on the current Node, then flows for hairpin and endpoint
@@ -269,11 +269,11 @@ type Client interface {
 	// InstallMulticastInitialFlows installs OpenFlow to packetIn the IGMP messages and output the Multicast traffic to
 	// antrea-gw0 so that local Pods could access external Multicast servers.
 	InstallMulticastInitialFlows(pktInReason uint8) error
-	// InstallMulticastFlow installs the flow to forward Multicast traffic normally, and output it to antrea-gw0
+	// InstallMulticastFlows installs the flow to forward Multicast traffic normally, and output it to antrea-gw0
 	// to ensure it can be forwarded to the external addresses.
-	InstallMulticastFlow(multicastIP net.IP) error
-	// UninstallMulticastFlow removes the flow matching the given multicastIP.
-	UninstallMulticastFlow(multicastIP net.IP) error
+	InstallMulticastFlows(multicastIP net.IP, groupID binding.GroupIDType) error
+	// UninstallMulticastFlows removes the flow matching the given multicastIP.
+	UninstallMulticastFlows(multicastIP net.IP) error
 	// SendIGMPQueryPacketOut sends the IGMPQuery packet as a packet-out to OVS from the gateway port.
 	SendIGMPQueryPacketOut(
 		dstMAC net.HardwareAddr,
@@ -292,6 +292,8 @@ type Client interface {
 
 	// UninstallTrafficControlReturnPortFlow removes the flow to classify the packets from a return port.
 	UninstallTrafficControlReturnPortFlow(returnOFPort uint32) error
+
+	InstallMulticastGroup(ofGroupID binding.GroupIDType, localReceivers []uint32) error
 }
 
 // GetFlowTableStatus returns an array of flow table status.
@@ -545,7 +547,7 @@ func (c *client) InstallServiceGroup(groupID binding.GroupIDType, withSessionAff
 	return nil
 }
 
-func (c *client) UninstallServiceGroup(groupID binding.GroupIDType) error {
+func (c *client) UninstallGroup(groupID binding.GroupIDType) error {
 	c.replayMutex.RLock()
 	defer c.replayMutex.RUnlock()
 	if !c.bridge.DeleteGroup(groupID) {
@@ -744,7 +746,7 @@ func (c *client) generatePipelines() {
 
 	if c.enableMulticast {
 		// TODO: add support for IPv6 protocol
-		c.featureMulticast = newFeatureMulticast(c.cookieAllocator, []binding.Protocol{binding.ProtocolIP})
+		c.featureMulticast = newFeatureMulticast(c.cookieAllocator, []binding.Protocol{binding.ProtocolIP}, c.bridge)
 		c.activatedFeatures = append(c.activatedFeatures, c.featureMulticast)
 	}
 	c.featureTraceflow = newFeatureTraceflow()
@@ -834,6 +836,9 @@ func (c *client) ReplayFlows() {
 	}
 
 	c.featureService.replayGroups()
+	if c.enableMulticast {
+		c.featureMulticast.replayGroups()
+	}
 
 	for _, activeFeature := range c.activatedFeatures {
 		if err := c.ofEntryOperations.AddAll(activeFeature.replayFlows()); err != nil {
@@ -1109,15 +1114,15 @@ func (c *client) InstallMulticastInitialFlows(pktInReason uint8) error {
 	return c.addFlows(c.featureMulticast.cachedFlows, cacheKey, flows)
 }
 
-func (c *client) InstallMulticastFlow(multicastIP net.IP) error {
-	flows := c.featureMulticast.localMulticastForwardFlow(multicastIP)
+func (c *client) InstallMulticastFlows(multicastIP net.IP, groupID binding.GroupIDType) error {
+	flows := c.featureMulticast.localMulticastForwardFlows(multicastIP, groupID)
 	cacheKey := fmt.Sprintf("multicast_%s", multicastIP.String())
 	c.replayMutex.RLock()
 	defer c.replayMutex.RUnlock()
 	return c.addFlows(c.featureMulticast.cachedFlows, cacheKey, flows)
 }
 
-func (c *client) UninstallMulticastFlow(multicastIP net.IP) error {
+func (c *client) UninstallMulticastFlows(multicastIP net.IP) error {
 	c.replayMutex.RLock()
 	defer c.replayMutex.RUnlock()
 	cacheKey := fmt.Sprintf("multicast_%s", multicastIP.String())
@@ -1173,3 +1178,14 @@ func (c *client) UninstallTrafficControlReturnPortFlow(returnOFPort uint32) erro
 	defer c.replayMutex.RUnlock()
 	return c.deleteFlows(c.featurePodConnectivity.tcCachedFlows, cacheKey)
 }
+
+func (c *client) InstallMulticastGroup(groupID binding.GroupIDType, localReceivers []uint32) error {
+	c.replayMutex.RLock()
+	defer c.replayMutex.RUnlock()
+
+	targetPorts := append([]uint32{config.HostGatewayOFPort}, localReceivers...)
+	if err := c.featureMulticast.multicastReceiversGroup(groupID, targetPorts...); err != nil {
+		return err
+	}
+	return nil
+}
diff --git a/pkg/agent/openflow/framework.go b/pkg/agent/openflow/framework.go
index 16c9392aa..8007bc976 100644
--- a/pkg/agent/openflow/framework.go
+++ b/pkg/agent/openflow/framework.go
@@ -245,7 +245,8 @@ func (f *featureEgress) getRequiredTables() []*Table {
 
 func (f *featureMulticast) getRequiredTables() []*Table {
 	return []*Table{
-		MulticastTable,
+		MulticastRoutingTable,
+		MulticastOutputTable,
 	}
 }
 
diff --git a/pkg/agent/openflow/multicast.go b/pkg/agent/openflow/multicast.go
index 3d416aa71..8a2adedfc 100644
--- a/pkg/agent/openflow/multicast.go
+++ b/pkg/agent/openflow/multicast.go
@@ -15,7 +15,11 @@
 package openflow
 
 import (
+	"fmt"
 	"net"
+	"sync"
+
+	"k8s.io/klog/v2"
 
 	"antrea.io/antrea/pkg/agent/openflow/cookie"
 	binding "antrea.io/antrea/pkg/ovs/openflow"
@@ -26,8 +30,10 @@ var _, mcastCIDR, _ = net.ParseCIDR("224.0.0.0/4")
 type featureMulticast struct {
 	cookieAllocator cookie.Allocator
 	ipProtocols     []binding.Protocol
+	bridge          binding.Bridge
 
 	cachedFlows *flowCategoryCache
+	groupCache  sync.Map
 
 	category cookie.Category
 }
@@ -36,12 +42,14 @@ func (f *featureMulticast) getFeatureName() string {
 	return "Multicast"
 }
 
-func newFeatureMulticast(cookieAllocator cookie.Allocator, ipProtocols []binding.Protocol) *featureMulticast {
+func newFeatureMulticast(cookieAllocator cookie.Allocator, ipProtocols []binding.Protocol, bridge binding.Bridge) *featureMulticast {
 	return &featureMulticast{
 		cookieAllocator: cookieAllocator,
 		ipProtocols:     ipProtocols,
 		cachedFlows:     newFlowCategoryCache(),
+		bridge:          bridge,
 		category:        cookie.Multicast,
+		groupCache:      sync.Map{},
 	}
 }
 
@@ -56,10 +64,48 @@ func multicastPipelineClassifyFlow(cookieID uint64, pipeline binding.Pipeline) b
 }
 
 func (f *featureMulticast) initFlows() []binding.Flow {
-	return []binding.Flow{}
+	cookieID := f.cookieAllocator.Request(f.category).Raw()
+	return []binding.Flow{
+		f.multicastOutputFlow(cookieID),
+	}
 }
 
 func (f *featureMulticast) replayFlows() []binding.Flow {
 	// Get cached flows.
 	return getCachedFlows(f.cachedFlows)
 }
+
+func (f *featureMulticast) multicastReceiversGroup(groupID binding.GroupIDType, ports ...uint32) error {
+	group := f.bridge.CreateGroupTypeAll(groupID).ResetBuckets()
+	for i := range ports {
+		group = group.Bucket().
+			LoadToRegField(OFPortFoundRegMark.GetField(), OFPortFoundRegMark.GetValue()).
+			LoadToRegField(TargetOFPortField, ports[i]).
+			ResubmitToTable(MulticastRoutingTable.GetNext()).
+			Done()
+	}
+	if err := group.Add(); err != nil {
+		return fmt.Errorf("error when installing Multicast receiver Group: %w", err)
+	}
+	f.groupCache.Store(groupID, group)
+	return nil
+}
+
+func (f *featureMulticast) multicastOutputFlow(cookieID uint64) binding.Flow {
+	return MulticastOutputTable.ofTable.BuildFlow(priorityNormal).
+		Cookie(cookieID).
+		MatchRegMark(OFPortFoundRegMark).
+		Action().OutputToRegField(TargetOFPortField).
+		Done()
+}
+
+func (f *featureMulticast) replayGroups() {
+	f.groupCache.Range(func(id, value interface{}) bool {
+		group := value.(binding.Group)
+		group.Reset()
+		if err := group.Add(); err != nil {
+			klog.ErrorS(err, "Error when replaying cached group", "group", id)
+		}
+		return true
+	})
+}
diff --git a/pkg/agent/openflow/pipeline.go b/pkg/agent/openflow/pipeline.go
index 0c3539cf9..d609d2f67 100644
--- a/pkg/agent/openflow/pipeline.go
+++ b/pkg/agent/openflow/pipeline.go
@@ -180,7 +180,10 @@ var (
 	// Tables of pipelineMulticast are declared below. Do don't declare any tables of other pipelines here!
 
 	// Tables in stageRouting:
-	MulticastTable = newTable("Multicast", stageRouting, pipelineMulticast)
+	MulticastRoutingTable = newTable("MulticastRouting", stageRouting, pipelineMulticast)
+
+	// Tables in stageOutput
+	MulticastOutputTable = newTable("MulticastOutput", stageOutput, pipelineMulticast)
 
 	// Flow priority level
 	priorityHigh            = uint16(210)
@@ -2637,8 +2640,8 @@ func pipelineClassifyFlow(cookieID uint64, protocol binding.Protocol, pipeline b
 		Done()
 }
 
-// igmpPktInFlows generates the flow to load CustomReasonIGMPRegMark to mark the IGMP packet in MulticastTable and sends
-// it to antrea-agent on MulticastTable.
+// igmpPktInFlows generates the flow to load CustomReasonIGMPRegMark to mark the IGMP packet in MulticastRoutingTable and sends
+// it to antrea-agent.
 func (f *featureMulticast) igmpPktInFlows(reason uint8) []binding.Flow {
 	flows := []binding.Flow{
 		// Set a custom reason for the IGMP packets, and then send it to antrea-agent and forward it normally in the
@@ -2646,7 +2649,7 @@ func (f *featureMulticast) igmpPktInFlows(reason uint8) []binding.Flow {
 		// group and its members in the meanwhile.
 		// Do not set dst IP address because IGMPv1 report message uses target multicast group as IP destination in
 		// the packet.
-		MulticastTable.ofTable.BuildFlow(priorityHigh).
+		MulticastRoutingTable.ofTable.BuildFlow(priorityHigh).
 			Cookie(f.cookieAllocator.Request(f.category).Raw()).
 			MatchProtocol(binding.ProtocolIGMP).
 			MatchRegMark(FromLocalRegMark).
@@ -2658,33 +2661,34 @@ func (f *featureMulticast) igmpPktInFlows(reason uint8) []binding.Flow {
 	return flows
 }
 
-// localMulticastForwardFlow generates the flow to forward multicast packets with OVS action "normal", and outputs
+// localMulticastForwardFlows generates the flow to forward multicast packets with OVS action "normal", and outputs
 // it to Antrea gateway in the meanwhile, so that the packet can be forwarded to local Pods which have joined the Multicast
 // group and to the external receivers. For external multicast packets accessing to the given multicast IP also hits the
 // flow, and the packet is not sent back to Antrea gateway because OVS datapath will drop it when it finds the output
 // port is the same as the input port.
-func (f *featureMulticast) localMulticastForwardFlow(multicastIP net.IP) []binding.Flow {
+func (f *featureMulticast) localMulticastForwardFlows(multicastIP net.IP, groupID binding.GroupIDType) []binding.Flow {
 	return []binding.Flow{
-		MulticastTable.ofTable.BuildFlow(priorityNormal).
+		MulticastRoutingTable.ofTable.BuildFlow(priorityNormal).
 			Cookie(f.cookieAllocator.Request(f.category).Raw()).
 			MatchProtocol(binding.ProtocolIP).
 			MatchDstIP(multicastIP).
-			Action().Output(config.HostGatewayOFPort).
-			Action().Normal().
+			Action().Group(groupID).
 			Done(),
 	}
 }
 
 // externalMulticastReceiverFlow generates the flow to output multicast packets to Antrea gateway, so that local Pods can
 // send multicast packets to access the external receivers. For the case that one or more local Pods have joined the target
-// multicast group, it is handled by the flows created by function "localMulticastForwardFlow" after local Pods report the
+// multicast group, it is handled by the flows created by function "localMulticastForwardFlows" after local Pods report the
 // IGMP membership.
 func (f *featureMulticast) externalMulticastReceiverFlow() binding.Flow {
-	return MulticastTable.ofTable.BuildFlow(priorityLow).
+	return MulticastRoutingTable.ofTable.BuildFlow(priorityLow).
 		Cookie(f.cookieAllocator.Request(f.category).Raw()).
 		MatchProtocol(binding.ProtocolIP).
 		MatchDstIPNet(*mcastCIDR).
-		Action().Output(config.HostGatewayOFPort).
+		Action().LoadRegMark(OFPortFoundRegMark).
+		Action().LoadToRegField(TargetOFPortField, config.HostGatewayOFPort).
+		Action().NextTable().
 		Done()
 }
 
diff --git a/pkg/agent/proxy/proxier.go b/pkg/agent/proxy/proxier.go
index d3e6a4d31..96f10c3c4 100644
--- a/pkg/agent/proxy/proxier.go
+++ b/pkg/agent/proxy/proxier.go
@@ -173,7 +173,7 @@ func (p *proxier) removeStaleServices() {
 		// Remove Service group whose Endpoints are local.
 		if svcInfo.NodeLocalExternal() {
 			if groupIDLocal, exist := p.groupCounter.Get(svcPortName, true); exist {
-				if err := p.ofClient.UninstallServiceGroup(groupIDLocal); err != nil {
+				if err := p.ofClient.UninstallGroup(groupIDLocal); err != nil {
 					klog.ErrorS(err, "Failed to remove Group of local Endpoints for Service", "Service", svcPortName)
 					continue
 				}
@@ -182,7 +182,7 @@ func (p *proxier) removeStaleServices() {
 		}
 		// Remove Service group which has all Endpoints.
 		if groupID, exist := p.groupCounter.Get(svcPortName, false); exist {
-			if err := p.ofClient.UninstallServiceGroup(groupID); err != nil {
+			if err := p.ofClient.UninstallGroup(groupID); err != nil {
 				klog.ErrorS(err, "Failed to remove Group of all Endpoints for Service", "Service", svcPortName)
 				continue
 			}
@@ -555,7 +555,7 @@ func (p *proxier) installServices() {
 					continue
 				}
 				if groupID, exist := p.groupCounter.Get(svcPortName, !nodeLocalVal); exist {
-					if err := p.ofClient.UninstallServiceGroup(groupID); err != nil {
+					if err := p.ofClient.UninstallGroup(groupID); err != nil {
 						klog.ErrorS(err, "Failed to uninstall Group of all Endpoints for Service", "Service", svcPortName)
 						continue
 					}
diff --git a/pkg/agent/types/event.go b/pkg/agent/types/event.go
new file mode 100644
index 000000000..eea70f66e
--- /dev/null
+++ b/pkg/agent/types/event.go
@@ -0,0 +1,22 @@
+// Copyright 2022 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 types
+
+type PodUpdate struct {
+	PodNamespace string
+	PodName      string
+	IsAdd        bool
+	ContainerID  string
+}
diff --git a/pkg/ovs/openflow/interfaces.go b/pkg/ovs/openflow/interfaces.go
index a086f9a40..74e362e69 100644
--- a/pkg/ovs/openflow/interfaces.go
+++ b/pkg/ovs/openflow/interfaces.go
@@ -94,6 +94,7 @@ type Bridge interface {
 	CreateTable(table Table, next uint8, missAction MissActionType) Table
 	// AddTable adds table on the Bridge. Return true if the operation succeeds, otherwise return false.
 	DeleteTable(id uint8) bool
+	CreateGroupTypeAll(id GroupIDType) Group
 	CreateGroup(id GroupIDType) Group
 	DeleteGroup(id GroupIDType) bool
 	CreateMeter(id MeterIDType, flags ofctrl.MeterFlag) Meter
diff --git a/pkg/ovs/openflow/ofctrl_bridge.go b/pkg/ovs/openflow/ofctrl_bridge.go
index c1b5e11df..e5b2b886c 100644
--- a/pkg/ovs/openflow/ofctrl_bridge.go
+++ b/pkg/ovs/openflow/ofctrl_bridge.go
@@ -189,8 +189,16 @@ type OFBridge struct {
 	multipartReplyChs map[uint32]chan *openflow13.MultipartReply
 }
 
+func (b *OFBridge) CreateGroupTypeAll(id GroupIDType) Group {
+	return b.createGroupWithType(id, ofctrl.GroupAll)
+}
+
 func (b *OFBridge) CreateGroup(id GroupIDType) Group {
-	ofctrlGroup, err := b.ofSwitch.NewGroup(uint32(id), ofctrl.GroupSelect)
+	return b.createGroupWithType(id, ofctrl.GroupSelect)
+}
+
+func (b *OFBridge) createGroupWithType(id GroupIDType, groupType ofctrl.GroupType) Group {
+	ofctrlGroup, err := b.ofSwitch.NewGroup(uint32(id), groupType)
 	if err != nil { // group already exists
 		ofctrlGroup = b.ofSwitch.GetGroup(uint32(id))
 	}
diff --git a/pkg/util/channel/channel.go b/pkg/util/channel/channel.go
index b10e9b4d4..019eee536 100644
--- a/pkg/util/channel/channel.go
+++ b/pkg/util/channel/channel.go
@@ -25,7 +25,7 @@ const (
 	notifyTimeout = time.Second
 )
 
-type eventHandler func(string)
+type eventHandler func(interface{})
 
 type Subscriber interface {
 	// Subscribe registers an eventHandler which will be called when an event is sent to the channel.
@@ -37,7 +37,7 @@ type Subscriber interface {
 
 type Notifier interface {
 	// Notify sends an event to the channel.
-	Notify(string) bool
+	Notify(interface{}) bool
 }
 
 // SubscribableChannel is different from the Go channel which dispatches every event to only single consumer regardless
@@ -47,7 +47,7 @@ type SubscribableChannel struct {
 	// The name of the channel, used for logging purpose to differentiate multiple channels.
 	name string
 	// eventCh is the channel used for buffering the pending events.
-	eventCh chan string
+	eventCh chan interface{}
 	// handlers is a slice of callbacks registered by consumers.
 	handlers []eventHandler
 }
@@ -55,7 +55,7 @@ type SubscribableChannel struct {
 func NewSubscribableChannel(name string, bufferSize int) *SubscribableChannel {
 	n := &SubscribableChannel{
 		name:    name,
-		eventCh: make(chan string, bufferSize),
+		eventCh: make(chan interface{}, bufferSize),
 	}
 	return n
 }
@@ -64,7 +64,7 @@ func (n *SubscribableChannel) Subscribe(h eventHandler) {
 	n.handlers = append(n.handlers, h)
 }
 
-func (n *SubscribableChannel) Notify(e string) bool {
+func (n *SubscribableChannel) Notify(e interface{}) bool {
 	timer := time.NewTimer(notifyTimeout)
 	defer timer.Stop()
 	select {
diff --git a/plugins/flow-visibility/clickhouse-monitor/go.mod b/plugins/flow-visibility/clickhouse-monitor/go.mod
index 0204c59c9..f3dd5d76a 100644
--- a/plugins/flow-visibility/clickhouse-monitor/go.mod
+++ b/plugins/flow-visibility/clickhouse-monitor/go.mod
@@ -11,5 +11,7 @@ require (
 require (
 	github.com/cloudflare/golz4 v0.0.0-20150217214814-ef862a3cdc58 // indirect
 	github.com/go-logr/logr v1.2.0 // indirect
+	github.com/gogo/protobuf v1.3.2 // indirect
+	gopkg.in/inf.v0 v0.9.1 // indirect
 	k8s.io/utils v0.0.0-20211116205334-6203023598ed // indirect
 )
diff --git a/plugins/flow-visibility/clickhouse-monitor/go.sum b/plugins/flow-visibility/clickhouse-monitor/go.sum
index 60371aec4..d187dd50a 100644
--- a/plugins/flow-visibility/clickhouse-monitor/go.sum
+++ b/plugins/flow-visibility/clickhouse-monitor/go.sum
@@ -35,6 +35,7 @@ github.com/go-openapi/jsonpointer v0.19.5/go.mod h1:Pl9vOtqEWErmShwVjC8pYs9cog34
 github.com/go-openapi/jsonreference v0.19.3/go.mod h1:rjx6GuL8TTa9VaixXglHmQmIL98+wF9xc8zWvFonSJ8=
 github.com/go-openapi/swag v0.19.5/go.mod h1:POnQmlKehdgb5mhVOsnJFsivZCEZ/vjK9gh66Z9tfKk=
 github.com/go-sql-driver/mysql v1.4.0/go.mod h1:zAC/RDZ24gD3HViQzih4MyKcchzm+sOG5ZlKdlhCg5w=
+github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q=
 github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q=
 github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q=
 github.com/golang/mock v1.1.1/go.mod h1:oTYuIxOrZwtPieC+H1uAHpcLFnEyAGVDL/k47Jfbm0A=
@@ -55,6 +56,7 @@ github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMyw
 github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
+github.com/google/gofuzz v1.1.0 h1:Hsa8mG0dQ46ij8Sl2AYJDUv1oA9/d6Vk+3LG99Oe02g=
 github.com/google/gofuzz v1.1.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
 github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/googleapis/gnostic v0.5.1/go.mod h1:6U4PtQXGIEt/Z3h5MAT7FNofLnw9vXk2cUuW7uA/OeU=
@@ -100,6 +102,7 @@ github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZb
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
 github.com/spf13/afero v1.2.2/go.mod h1:9ZxEEn6pIJ8Rxe320qSDBk6AsU0r9pR7Q4OcevTdifk=
+github.com/spf13/pflag v1.0.5 h1:iy+VFUOCP1a+8yFto/drg2CJ5u0yRoB7fZw3DKv/JXA=
 github.com/spf13/pflag v1.0.5/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
 github.com/stoewer/go-strcase v1.2.0/go.mod h1:IBiWB2sKIp3wVVQ3Y035++gc+knqhUQag1KpM8ahLw8=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
@@ -202,6 +205,7 @@ gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8
 gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20200227125254-8fa46927fb4f/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/fsnotify.v1 v1.4.7/go.mod h1:Tz8NjZHkW78fSQdbUxIjBTcgA1z1m8ZHf0WmKUhAMys=
+gopkg.in/inf.v0 v0.9.1 h1:73M5CoZyi3ZLMOyDlQh031Cx6N9NDJ2Vvfl76EDAgDc=
 gopkg.in/inf.v0 v0.9.1/go.mod h1:cWUDdTG/fYaXco+Dcufb5Vnc6Gp2YChqWtbxRZE0mXw=
 gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7/go.mod h1:dt/ZhP58zS4L8KSrWDmTeBkI65Dw0HsyUHuEVlX15mw=
 gopkg.in/yaml.v2 v2.2.1/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
diff --git a/plugins/flow-visibility/clickhouse-monitor/main.go b/plugins/flow-visibility/clickhouse-monitor/main.go
index cc1f6c61e..4e779c3b9 100644
--- a/plugins/flow-visibility/clickhouse-monitor/main.go
+++ b/plugins/flow-visibility/clickhouse-monitor/main.go
@@ -22,6 +22,7 @@ import (
 	"time"
 
 	"github.com/ClickHouse/clickhouse-go"
+	"k8s.io/apimachinery/pkg/api/resource"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/klog/v2"
 )
@@ -48,6 +49,8 @@ const (
 )
 
 var (
+	// Storage size allocated for the ClickHouse in number of bytes
+	allocatedSpace uint64
 	// The name of the table to store the flow records
 	tableName = os.Getenv("TABLE_NAME")
 	// The names of the materialized views
@@ -58,16 +61,26 @@ var (
 
 func main() {
 	// Check environment variables
-	if len(tableName) == 0 || len(mvNames) == 0 {
-		klog.ErrorS(nil, "Unable to load environment variables, TABLE_NAME and MV_NAMES must be defined")
+	allocatedSpaceStr := os.Getenv("STORAGE_SIZE")
+
+	if len(tableName) == 0 || len(mvNames) == 0 || len(allocatedSpaceStr) == 0 {
+		klog.ErrorS(nil, "Unable to load environment variables, TABLE_NAME, MV_NAMES and STORAGE_SIZE must be defined")
+		return
+	}
+	var err error
+	quantity, err := resource.ParseQuantity(allocatedSpaceStr)
+	if err != nil {
+		klog.ErrorS(err, "Error when parsing STORAGE_SIZE")
 		return
 	}
+	allocatedSpace = uint64(quantity.Value())
 
 	connect, err := connectLoop()
 	if err != nil {
 		klog.ErrorS(err, "Error when connecting to ClickHouse")
 		os.Exit(1)
 	}
+	checkStorageCondition(connect)
 	wait.Forever(func() {
 		// The monitor stops working for several rounds after a deletion
 		// as the release of memory space by the ClickHouse MergeTree engine requires time
@@ -118,28 +131,69 @@ func connectLoop() (*sql.DB, error) {
 	return connect, nil
 }
 
-// Checks the memory usage in the ClickHouse, and deletes records when it exceeds the threshold.
-func monitorMemory(connect *sql.DB) {
+// Check if ClickHouse shares storage space with other software
+func checkStorageCondition(connect *sql.DB) {
 	var (
 		freeSpace  uint64
+		usedSpace  uint64
 		totalSpace uint64
 	)
-	// Get memory usage from ClickHouse system table
+	getDiskUsage(connect, &freeSpace, &totalSpace)
+	getClickHouseUsage(connect, &usedSpace)
+	availablePercentage := float64(freeSpace+usedSpace) / float64(totalSpace)
+	klog.InfoS("Low available percentage implies ClickHouse does not save data on a dedicated disk", "availablePercentage", availablePercentage)
+}
+
+func getDiskUsage(connect *sql.DB, freeSpace *uint64, totalSpace *uint64) {
+	// Get free space from ClickHouse system table
+	if err := wait.PollImmediate(queryRetryInterval, queryTimeout, func() (bool, error) {
+		if err := connect.QueryRow("SELECT free_space, total_space FROM system.disks").Scan(freeSpace, totalSpace); err != nil {
+			klog.ErrorS(err, "Failed to get the disk usage")
+			return false, nil
+		} else {
+			return true, nil
+		}
+	}); err != nil {
+		klog.ErrorS(err, "Failed to get the disk usage", "timeout", queryTimeout)
+		return
+	}
+}
+
+func getClickHouseUsage(connect *sql.DB, usedSpace *uint64) {
+	// Get space usage from ClickHouse system table
 	if err := wait.PollImmediate(queryRetryInterval, queryTimeout, func() (bool, error) {
-		if err := connect.QueryRow("SELECT free_space, total_space FROM system.disks").Scan(&freeSpace, &totalSpace); err != nil {
-			klog.ErrorS(err, "Failed to get memory usage for ClickHouse")
+		if err := connect.QueryRow("SELECT SUM(bytes) FROM system.parts").Scan(usedSpace); err != nil {
+			klog.ErrorS(err, "Failed to get the used space size by the ClickHouse")
 			return false, nil
 		} else {
 			return true, nil
 		}
 	}); err != nil {
-		klog.ErrorS(err, "Failed to get memory usage for ClickHouse", "timeout", queryTimeout)
+		klog.ErrorS(err, "Failed to get the used space size by the ClickHouse", "timeout", queryTimeout)
 		return
 	}
+}
+
+// Checks the memory usage in the ClickHouse, and deletes records when it exceeds the threshold.
+func monitorMemory(connect *sql.DB) {
+	var (
+		freeSpace  uint64
+		usedSpace  uint64
+		totalSpace uint64
+	)
+	getDiskUsage(connect, &freeSpace, &totalSpace)
+	getClickHouseUsage(connect, &usedSpace)
+
+	// Total space for ClickHouse is the smaller one of the user allocated space size and the actual space size on the disk
+	if (freeSpace + usedSpace) < allocatedSpace {
+		totalSpace = freeSpace + usedSpace
+	} else {
+		totalSpace = allocatedSpace
+	}
 
 	// Calculate the memory usage
-	usagePercentage := float64(totalSpace-freeSpace) / float64(totalSpace)
-	klog.InfoS("Memory usage", "total", totalSpace, "used", totalSpace-freeSpace, "percentage", usagePercentage)
+	usagePercentage := float64(usedSpace) / float64(totalSpace)
+	klog.InfoS("Memory usage", "total", totalSpace, "used", usedSpace, "percentage", usagePercentage)
 	// Delete records when memory usage is larger than threshold
 	if usagePercentage > threshold {
 		timeBoundary, err := getTimeBoundary(connect)


Test patch:
diff --git a/build/yamls/flow-visibility/patches/e2e/imagePullPolicyClickhouse.yml b/build/yamls/flow-visibility/patches/e2e/imagePullPolicyClickhouse.yml
index 484649ec3..b49e45178 100644
--- a/build/yamls/flow-visibility/patches/e2e/imagePullPolicyClickhouse.yml
+++ b/build/yamls/flow-visibility/patches/e2e/imagePullPolicyClickhouse.yml
@@ -1,6 +1,3 @@
 - op: add
   path: /spec/templates/podTemplates/0/spec/containers/0/imagePullPolicy
   value: IfNotPresent
-- op: add
-  path: /spec/templates/podTemplates/0/spec/containers/1/imagePullPolicy
-  value: IfNotPresent
diff --git a/pkg/agent/controller/egress/egress_controller_test.go b/pkg/agent/controller/egress/egress_controller_test.go
index 9c448dd1c..0d241bf7e 100644
--- a/pkg/agent/controller/egress/egress_controller_test.go
+++ b/pkg/agent/controller/egress/egress_controller_test.go
@@ -38,6 +38,7 @@ import (
 	ipassignertest "antrea.io/antrea/pkg/agent/ipassigner/testing"
 	openflowtest "antrea.io/antrea/pkg/agent/openflow/testing"
 	routetest "antrea.io/antrea/pkg/agent/route/testing"
+	"antrea.io/antrea/pkg/agent/types"
 	"antrea.io/antrea/pkg/agent/util"
 	cpv1b2 "antrea.io/antrea/pkg/apis/controlplane/v1beta2"
 	crdv1a2 "antrea.io/antrea/pkg/apis/crd/v1alpha2"
@@ -582,7 +583,11 @@ func TestPodUpdateShouldSyncEgress(t *testing.T) {
 	c.mockIPAssigner.EXPECT().UnassignIP(fakeLocalEgressIP1)
 	// Mock CNIServer
 	addPodInterface(c.ifaceStore, "ns1", "pendingPod", 10)
-	c.podUpdateChannel.Notify("ns1/pendingPod")
+	ev := types.PodUpdate{
+		PodName:      "pendingPod",
+		PodNamespace: "ns1",
+	}
+	c.podUpdateChannel.Notify(ev)
 	require.NoError(t, wait.PollImmediate(10*time.Millisecond, time.Second, func() (done bool, err error) {
 		return c.queue.Len() == 1, nil
 	}))
diff --git a/pkg/agent/controller/networkpolicy/cache_test.go b/pkg/agent/controller/networkpolicy/cache_test.go
index 35878d26c..00c01d3ed 100644
--- a/pkg/agent/controller/networkpolicy/cache_test.go
+++ b/pkg/agent/controller/networkpolicy/cache_test.go
@@ -25,8 +25,10 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/sets"
 
+	"antrea.io/antrea/pkg/agent/types"
 	"antrea.io/antrea/pkg/apis/controlplane/v1beta2"
 	"antrea.io/antrea/pkg/util/channel"
+	"antrea.io/antrea/pkg/util/k8s"
 )
 
 var (
@@ -1184,7 +1186,12 @@ func TestRuleCacheProcessPodUpdates(t *testing.T) {
 			stopCh := make(chan struct{})
 			defer close(stopCh)
 			go podUpdateNotifier.Run(stopCh)
-			podUpdateNotifier.Notify(tt.podUpdate)
+			ns, name := k8s.SplitNamespacedName(tt.podUpdate)
+			e := types.PodUpdate{
+				PodNamespace: ns,
+				PodName:      name,
+			}
+			podUpdateNotifier.Notify(e)
 			func() {
 				// Drain the channel with 10 ms timeout so we can know it's done.
 				for {
diff --git a/pkg/agent/multicast/mcast_controller_test.go b/pkg/agent/multicast/mcast_controller_test.go
index 9918b5ffe..0536dfda6 100644
--- a/pkg/agent/multicast/mcast_controller_test.go
+++ b/pkg/agent/multicast/mcast_controller_test.go
@@ -36,8 +36,10 @@ import (
 	"antrea.io/antrea/pkg/agent/interfacestore"
 	ifaceStoretest "antrea.io/antrea/pkg/agent/interfacestore/testing"
 	multicasttest "antrea.io/antrea/pkg/agent/multicast/testing"
+	"antrea.io/antrea/pkg/agent/openflow"
 	openflowtest "antrea.io/antrea/pkg/agent/openflow/testing"
 	ovsconfigtest "antrea.io/antrea/pkg/ovs/ovsconfig/testing"
+	"antrea.io/antrea/pkg/util/channel"
 )
 
 var (
@@ -45,16 +47,21 @@ var (
 	mockMulticastSocket *multicasttest.MockRouteInterface
 	mockIfaceStore      *ifaceStoretest.MockInterfaceStore
 	ovsClient           *ovsconfigtest.MockOVSBridgeClient
-	mgroup              = net.ParseIP("224.96.1.3")
 	if1                 = &interfacestore.InterfaceConfig{
 		Type:          interfacestore.ContainerInterface,
 		InterfaceName: "if1",
 		IPs:           []net.IP{net.ParseIP("192.168.1.1")},
+		OVSPortConfig: &interfacestore.OVSPortConfig{
+			OFPort: 1,
+		},
 	}
 	if2 = &interfacestore.InterfaceConfig{
 		Type:          interfacestore.ContainerInterface,
 		InterfaceName: "if2",
 		IPs:           []net.IP{net.ParseIP("192.168.1.2")},
+		OVSPortConfig: &interfacestore.OVSPortConfig{
+			OFPort: 2,
+		},
 	}
 	nodeIf1IP           = net.ParseIP("192.168.20.22")
 	externalInterfaceIP = net.ParseIP("192.168.50.23")
@@ -63,6 +70,7 @@ var (
 )
 
 func TestAddGroupMemberStatus(t *testing.T) {
+	mgroup := net.ParseIP("224.96.1.3")
 	event := &mcastGroupEvent{
 		group: mgroup,
 		eType: groupJoin,
@@ -82,7 +90,9 @@ func TestAddGroupMemberStatus(t *testing.T) {
 	key, ok := obj.(string)
 	assert.True(t, ok)
 	assert.Equal(t, mgroup.String(), key)
-	mockOFClient.EXPECT().InstallMulticastFlow(mgroup).Times(1)
+	mockIfaceStore.EXPECT().GetInterfaceByName(if1.InterfaceName).Return(if1, true)
+	mockOFClient.EXPECT().InstallMulticastGroup(gomock.Any(), gomock.Any())
+	mockOFClient.EXPECT().InstallMulticastFlows(mgroup, gomock.Any()).Times(1)
 	mockMulticastSocket.EXPECT().MulticastInterfaceJoinMgroup(mgroup.To4(), nodeIf1IP.To4(), if1.InterfaceName).Times(1)
 	err = mctrl.syncGroup(key)
 	assert.Nil(t, err)
@@ -94,6 +104,7 @@ func TestUpdateGroupMemberStatus(t *testing.T) {
 	err := mctrl.initialize(t)
 	assert.Nil(t, err)
 	igmpMaxResponseTime = time.Second * 1
+	mgroup := net.ParseIP("224.96.1.4")
 	event := &mcastGroupEvent{
 		group: mgroup,
 		eType: groupJoin,
@@ -101,7 +112,6 @@ func TestUpdateGroupMemberStatus(t *testing.T) {
 		iface: if1,
 	}
 	mctrl.addGroupMemberStatus(event)
-	obj, _, _ := mctrl.groupCache.GetByKey(event.group.String())
 	mockOFClient.EXPECT().SendIGMPQueryPacketOut(igmpQueryDstMac, mcastAllHosts, uint32(openflow13.P_NORMAL), gomock.Any()).Times(len(queryVersions))
 	for _, e := range []*mcastGroupEvent{
 		{group: mgroup, eType: groupJoin, time: event.time.Add(time.Second * 20), iface: if1},
@@ -110,6 +120,10 @@ func TestUpdateGroupMemberStatus(t *testing.T) {
 		{group: mgroup, eType: groupLeave, time: event.time.Add(time.Second * 61), iface: if1},
 		{group: mgroup, eType: groupLeave, time: event.time.Add(time.Second * 62), iface: if2},
 	} {
+		obj, _, _ := mctrl.groupCache.GetByKey(event.group.String())
+		if e.eType == groupLeave {
+			mockIfaceStore.EXPECT().GetInterfaceByName(e.iface.InterfaceName).Return(e.iface, true)
+		}
 		mctrl.updateGroupMemberStatus(obj, e)
 		groupCache := mctrl.groupCache
 		compareGroupStatus(t, groupCache, e)
@@ -127,15 +141,22 @@ func TestCheckLastMember(t *testing.T) {
 	workerCount = 1
 	igmpMaxResponseTime = time.Second * 1
 	lastProbe := time.Now()
+	mgroup := net.ParseIP("224.96.1.2")
 	testCheckLastMember := func(ev *mcastGroupEvent, expExist bool) {
 		status := &GroupMemberStatus{
-			localMembers:   sets.NewString(),
+			localMembers:   map[string]time.Time{},
 			lastIGMPReport: lastProbe,
 		}
 		if ev != nil {
 			status.group = ev.group
+			if ev.eType == groupLeave {
+				mockOFClient.EXPECT().UninstallGroup(gomock.Any())
+				mockOFClient.EXPECT().UninstallMulticastFlows(status.group)
+			}
 		} else {
 			status.group = mgroup
+			mockOFClient.EXPECT().UninstallGroup(gomock.Any())
+			mockOFClient.EXPECT().UninstallMulticastFlows(status.group)
 		}
 		_ = mctrl.groupCache.Add(status)
 		mctrl.addInstalledGroup(status.group.String())
@@ -167,7 +188,8 @@ func TestCheckLastMember(t *testing.T) {
 		}
 		mctrl.queue.Forget(obj)
 	}
-	mockOFClient.EXPECT().UninstallMulticastFlow(gomock.Any()).Times(2)
+	mockIfaceStore.EXPECT().GetInterfaceByName(if1.InterfaceName).Return(if1, true).Times(1)
+	mockOFClient.EXPECT().InstallMulticastGroup(gomock.Any(), gomock.Any()).Times(1)
 	for _, tc := range []struct {
 		ev     *mcastGroupEvent
 		exists bool
@@ -194,30 +216,31 @@ func TestClearStaleGroups(t *testing.T) {
 		mctrl.worker()
 		wg.Done()
 	}()
-
 	now := time.Now()
+	validUpdateTime := now.Add(-queryInterval)
 	validGroups := []*GroupMemberStatus{
 		{
 			group:          net.ParseIP("224.96.1.2"),
-			localMembers:   sets.NewString("p1", "p2"),
-			lastIGMPReport: now.Add(-queryInterval),
+			localMembers:   map[string]time.Time{"p1": now, "p2": validUpdateTime},
+			lastIGMPReport: validUpdateTime,
 		},
 		{
 			group:          net.ParseIP("224.96.1.3"),
-			localMembers:   sets.NewString(),
-			lastIGMPReport: now.Add(-queryInterval),
+			localMembers:   map[string]time.Time{"p2": validUpdateTime},
+			lastIGMPReport: validUpdateTime,
 		},
 	}
+	staleUpdateTime := now.Add(-mcastGroupTimeout - time.Second)
 	staleGroups := []*GroupMemberStatus{
 		{
 			group:          net.ParseIP("224.96.1.4"),
-			localMembers:   sets.NewString("p1", "p3"),
-			lastIGMPReport: now.Add(-mcastGroupTimeout - time.Second),
+			localMembers:   map[string]time.Time{"p1": staleUpdateTime, "p3": staleUpdateTime},
+			lastIGMPReport: staleUpdateTime,
 		},
 		{
 			group:          net.ParseIP("224.96.1.5"),
-			localMembers:   sets.NewString(),
-			lastIGMPReport: now.Add(-mcastGroupTimeout - time.Second),
+			localMembers:   map[string]time.Time{},
+			lastIGMPReport: staleUpdateTime,
 		},
 	}
 	for _, g := range validGroups {
@@ -225,12 +248,19 @@ func TestClearStaleGroups(t *testing.T) {
 		assert.Nil(t, err)
 		mctrl.addInstalledGroup(g.group.String())
 	}
+	fakePort := int32(1)
 	for _, g := range staleGroups {
 		err := mctrl.groupCache.Add(g)
 		assert.Nil(t, err)
 		mctrl.addInstalledGroup(g.group.String())
+		for m := range g.localMembers {
+			mockIface := &interfacestore.InterfaceConfig{InterfaceName: m, OVSPortConfig: &interfacestore.OVSPortConfig{OFPort: fakePort}}
+			mockIfaceStore.EXPECT().GetInterfaceByName(m).Return(mockIface, true)
+			fakePort++
+		}
 	}
-	mockOFClient.EXPECT().UninstallMulticastFlow(gomock.Any()).Times(len(staleGroups))
+	mockOFClient.EXPECT().UninstallGroup(gomock.Any()).Times(len(staleGroups))
+	mockOFClient.EXPECT().UninstallMulticastFlows(gomock.Any()).Times(len(staleGroups))
 	mockMulticastSocket.EXPECT().MulticastInterfaceLeaveMgroup(gomock.Any(), gomock.Any(), gomock.Any()).Times(len(staleGroups))
 	mctrl.clearStaleGroups()
 	mctrl.queue.ShutDown()
@@ -251,7 +281,9 @@ func TestProcessPacketIn(t *testing.T) {
 	snooper := mockController.igmpSnooper
 	stopCh := make(chan struct{})
 	defer close(stopCh)
-	go mockController.eventHandler(stopCh)
+	go func() {
+		mockController.eventHandler(stopCh)
+	}()
 
 	getIPs := func(ipStrs []string) []net.IP {
 		ips := make([]net.IP, len(ipStrs))
@@ -285,6 +317,7 @@ func TestProcessPacketIn(t *testing.T) {
 			version:      3,
 		},
 	} {
+		mockIfaceStore.EXPECT().GetInterfaceByName(tc.iface.InterfaceName).Return(tc.iface, true).AnyTimes()
 		packets := createIGMPReportPacketIn(getIPs(tc.joinedGroups.List()), getIPs(tc.leftGroups.List()), tc.version, uint32(tc.iface.OFPort))
 		mockOFClient.EXPECT().SendIGMPQueryPacketOut(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes()
 		for _, pkt := range packets {
@@ -311,11 +344,11 @@ func compareGroupStatus(t *testing.T, cache cache.Indexer, event *mcastGroupEven
 	assert.Equal(t, event.group, status.group)
 	if event.eType == groupJoin {
 		assert.True(t, status.lastIGMPReport.Equal(event.time) || status.lastIGMPReport.After(event.time))
-		exists := status.localMembers.Has(event.iface.InterfaceName)
+		_, exists := status.localMembers[event.iface.InterfaceName]
 		assert.Truef(t, exists, "member is not added into cache")
 	} else {
 		assert.True(t, status.lastIGMPReport.Before(event.time))
-		exists := status.localMembers.Has(event.iface.InterfaceName)
+		_, exists := status.localMembers[event.iface.InterfaceName]
 		assert.Falsef(t, exists, "member is not removed from cache")
 	}
 }
@@ -329,13 +362,13 @@ func newMockMulticastController(t *testing.T) *Controller {
 	addr := &net.IPNet{IP: nodeIf1IP, Mask: net.IPv4Mask(255, 255, 255, 0)}
 	nodeConfig := &config.NodeConfig{GatewayConfig: &config.GatewayConfig{Name: "antrea-gw0"}, NodeIPv4Addr: addr}
 	mockOFClient.EXPECT().RegisterPacketInHandler(gomock.Any(), gomock.Any(), gomock.Any()).Times(1)
-	mctrl := NewMulticastController(mockOFClient, nodeConfig, mockIfaceStore, mockMulticastSocket, sets.NewString(), ovsClient)
+	groupAllocator := openflow.NewGroupAllocator(false)
+	podUpdateSubscriber := channel.NewSubscribableChannel("PodUpdate", 100)
+	mctrl := NewMulticastController(mockOFClient, groupAllocator, nodeConfig, mockIfaceStore, mockMulticastSocket, sets.NewString(), ovsClient, podUpdateSubscriber)
 	return mctrl
 }
 
 func (c *Controller) initialize(t *testing.T) error {
-	ovsClient.EXPECT().SetBridgeMcastSnooping(true).Times(1)
-	ovsClient.EXPECT().AddBridgeOtherConfig(map[string]interface{}{"mcast-snooping-disable-flood-unregistered": "true"}).Times(1)
 	mockOFClient.EXPECT().InstallMulticastInitialFlows(uint8(0)).Times(1)
 	mockMulticastSocket.EXPECT().AllocateVIFs(gomock.Any(), uint16(0)).Times(1).Return([]uint16{0}, nil)
 	mockMulticastSocket.EXPECT().AllocateVIFs(gomock.Any(), uint16(1)).Times(1).Return([]uint16{1, 2}, nil)
diff --git a/pkg/agent/multicast/mcast_route_test.go b/pkg/agent/multicast/mcast_route_test.go
index 081171e7c..b9f4c14d6 100644
--- a/pkg/agent/multicast/mcast_route_test.go
+++ b/pkg/agent/multicast/mcast_route_test.go
@@ -21,6 +21,7 @@ import (
 	"fmt"
 	"net"
 	"testing"
+	"time"
 
 	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/assert"
@@ -76,12 +77,12 @@ func TestProcessIGMPNocacheMsg(t *testing.T) {
 	mRoute.internalInterfaceVIF = uint16(0)
 	status1 := &GroupMemberStatus{
 		group:        net.ParseIP("224.3.3.8"),
-		localMembers: sets.NewString("aa"),
+		localMembers: map[string]time.Time{"aa": time.Now()},
 	}
 	mRoute.groupCache.Add(status1)
 	status2 := &GroupMemberStatus{
 		group:        net.ParseIP("224.3.3.9"),
-		localMembers: sets.NewString(),
+		localMembers: map[string]time.Time{},
 	}
 	mRoute.groupCache.Add(status2)
 	for _, m := range []struct {
diff --git a/pkg/agent/openflow/testing/mock_openflow.go b/pkg/agent/openflow/testing/mock_openflow.go
index 34415abdb..8e47d69fc 100644
--- a/pkg/agent/openflow/testing/mock_openflow.go
+++ b/pkg/agent/openflow/testing/mock_openflow.go
@@ -282,18 +282,32 @@ func (mr *MockClientMockRecorder) InstallEndpointFlows(arg0, arg1 interface{}) *
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "InstallEndpointFlows", reflect.TypeOf((*MockClient)(nil).InstallEndpointFlows), arg0, arg1)
 }
 
-// InstallMulticastFlow mocks base method
-func (m *MockClient) InstallMulticastFlow(arg0 net.IP) error {
+// InstallMulticastFlows mocks base method
+func (m *MockClient) InstallMulticastFlows(arg0 net.IP, arg1 openflow.GroupIDType) error {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "InstallMulticastFlow", arg0)
+	ret := m.ctrl.Call(m, "InstallMulticastFlows", arg0, arg1)
 	ret0, _ := ret[0].(error)
 	return ret0
 }
 
-// InstallMulticastFlow indicates an expected call of InstallMulticastFlow
-func (mr *MockClientMockRecorder) InstallMulticastFlow(arg0 interface{}) *gomock.Call {
+// InstallMulticastFlows indicates an expected call of InstallMulticastFlows
+func (mr *MockClientMockRecorder) InstallMulticastFlows(arg0, arg1 interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "InstallMulticastFlow", reflect.TypeOf((*MockClient)(nil).InstallMulticastFlow), arg0)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "InstallMulticastFlows", reflect.TypeOf((*MockClient)(nil).InstallMulticastFlows), arg0, arg1)
+}
+
+// InstallMulticastGroup mocks base method
+func (m *MockClient) InstallMulticastGroup(arg0 openflow.GroupIDType, arg1 []uint32) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "InstallMulticastGroup", arg0, arg1)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// InstallMulticastGroup indicates an expected call of InstallMulticastGroup
+func (mr *MockClientMockRecorder) InstallMulticastGroup(arg0, arg1 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "InstallMulticastGroup", reflect.TypeOf((*MockClient)(nil).InstallMulticastGroup), arg0, arg1)
 }
 
 // InstallMulticastInitialFlows mocks base method
@@ -640,18 +654,32 @@ func (mr *MockClientMockRecorder) UninstallEndpointFlows(arg0, arg1 interface{})
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UninstallEndpointFlows", reflect.TypeOf((*MockClient)(nil).UninstallEndpointFlows), arg0, arg1)
 }
 
-// UninstallMulticastFlow mocks base method
-func (m *MockClient) UninstallMulticastFlow(arg0 net.IP) error {
+// UninstallGroup mocks base method
+func (m *MockClient) UninstallGroup(arg0 openflow.GroupIDType) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "UninstallGroup", arg0)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// UninstallGroup indicates an expected call of UninstallGroup
+func (mr *MockClientMockRecorder) UninstallGroup(arg0 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UninstallGroup", reflect.TypeOf((*MockClient)(nil).UninstallGroup), arg0)
+}
+
+// UninstallMulticastFlows mocks base method
+func (m *MockClient) UninstallMulticastFlows(arg0 net.IP) error {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "UninstallMulticastFlow", arg0)
+	ret := m.ctrl.Call(m, "UninstallMulticastFlows", arg0)
 	ret0, _ := ret[0].(error)
 	return ret0
 }
 
-// UninstallMulticastFlow indicates an expected call of UninstallMulticastFlow
-func (mr *MockClientMockRecorder) UninstallMulticastFlow(arg0 interface{}) *gomock.Call {
+// UninstallMulticastFlows indicates an expected call of UninstallMulticastFlows
+func (mr *MockClientMockRecorder) UninstallMulticastFlows(arg0 interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UninstallMulticastFlow", reflect.TypeOf((*MockClient)(nil).UninstallMulticastFlow), arg0)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UninstallMulticastFlows", reflect.TypeOf((*MockClient)(nil).UninstallMulticastFlows), arg0)
 }
 
 // UninstallNodeFlows mocks base method
@@ -739,20 +767,6 @@ func (mr *MockClientMockRecorder) UninstallServiceFlows(arg0, arg1, arg2 interfa
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UninstallServiceFlows", reflect.TypeOf((*MockClient)(nil).UninstallServiceFlows), arg0, arg1, arg2)
 }
 
-// UninstallServiceGroup mocks base method
-func (m *MockClient) UninstallServiceGroup(arg0 openflow.GroupIDType) error {
-	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "UninstallServiceGroup", arg0)
-	ret0, _ := ret[0].(error)
-	return ret0
-}
-
-// UninstallServiceGroup indicates an expected call of UninstallServiceGroup
-func (mr *MockClientMockRecorder) UninstallServiceGroup(arg0 interface{}) *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UninstallServiceGroup", reflect.TypeOf((*MockClient)(nil).UninstallServiceGroup), arg0)
-}
-
 // UninstallTraceflowFlows mocks base method
 func (m *MockClient) UninstallTraceflowFlows(arg0 byte) error {
 	m.ctrl.T.Helper()
diff --git a/pkg/agent/proxy/proxier_test.go b/pkg/agent/proxy/proxier_test.go
index fc809925f..6c6ec07c2 100644
--- a/pkg/agent/proxy/proxier_test.go
+++ b/pkg/agent/proxy/proxier_test.go
@@ -347,7 +347,7 @@ func testLoadBalancer(t *testing.T, nodePortAddresses []net.IP, svcIP, ep1IP, ep
 			mockOFClient.EXPECT().InstallServiceFlows(groupID, loadBalancerIP, uint16(svcPort), bindingProtocol, uint16(0), nodeLocalVal, corev1.ServiceTypeLoadBalancer).Times(1)
 		}
 		groupID = fp.groupCounter.AllocateIfNotExist(svcPortName, !nodeLocalVal)
-		mockOFClient.EXPECT().UninstallServiceGroup(groupID).Times(1)
+		mockOFClient.EXPECT().UninstallGroup(groupID).Times(1)
 	}
 	mockRouteClient.EXPECT().AddClusterIPRoute(svcIP).Times(1)
 	if proxyLoadBalancerIPs {
@@ -466,7 +466,7 @@ func testNodePort(t *testing.T, nodePortAddresses []net.IP, svcIP, ep1IP, ep2IP
 		mockOFClient.EXPECT().InstallServiceFlows(groupID, gomock.Any(), uint16(svcNodePort), bindingProtocol, uint16(0), nodeLocalVal, corev1.ServiceTypeNodePort).Times(1)
 
 		groupID = fp.groupCounter.AllocateIfNotExist(svcPortName, !nodeLocalVal)
-		mockOFClient.EXPECT().UninstallServiceGroup(groupID).Times(1)
+		mockOFClient.EXPECT().UninstallGroup(groupID).Times(1)
 	}
 	mockRouteClient.EXPECT().AddClusterIPRoute(svcIP).Times(1)
 	mockRouteClient.EXPECT().AddNodePort(gomock.Any(), uint16(svcNodePort), bindingProtocol).Times(1)
@@ -748,7 +748,7 @@ func testClusterIPRemoval(t *testing.T, svcIP net.IP, epIP net.IP, isIPv6 bool)
 	mockRouteClient.EXPECT().AddClusterIPRoute(svcIP).Times(1)
 	mockOFClient.EXPECT().UninstallServiceFlows(svcIP, uint16(svcPort), bindingProtocol).Times(1)
 	mockOFClient.EXPECT().UninstallEndpointFlows(bindingProtocol, gomock.Any()).Times(1)
-	mockOFClient.EXPECT().UninstallServiceGroup(gomock.Any()).Times(1)
+	mockOFClient.EXPECT().UninstallGroup(gomock.Any()).Times(1)
 	fp.syncProxyRules()
 
 	fp.serviceChanges.OnServiceUpdate(service, nil)
@@ -1231,8 +1231,8 @@ func TestServicesWithSameEndpoints(t *testing.T) {
 	mockOFClient.EXPECT().InstallServiceFlows(groupID2, svcIP2, uint16(svcPort), bindingProtocol, uint16(0), false, corev1.ServiceTypeClusterIP).Times(1)
 	mockOFClient.EXPECT().UninstallServiceFlows(svcIP1, uint16(svcPort), bindingProtocol).Times(1)
 	mockOFClient.EXPECT().UninstallServiceFlows(svcIP2, uint16(svcPort), bindingProtocol).Times(1)
-	mockOFClient.EXPECT().UninstallServiceGroup(groupID1).Times(1)
-	mockOFClient.EXPECT().UninstallServiceGroup(groupID2).Times(1)
+	mockOFClient.EXPECT().UninstallGroup(groupID1).Times(1)
+	mockOFClient.EXPECT().UninstallGroup(groupID2).Times(1)
 	// Since these two Services reference to the same Endpoint, there should only be one operation.
 	mockOFClient.EXPECT().UninstallEndpointFlows(bindingProtocol, gomock.Any()).Times(1)
 
diff --git a/pkg/ovs/openflow/testing/mock_openflow.go b/pkg/ovs/openflow/testing/mock_openflow.go
index 283a606c5..45a1e3f47 100644
--- a/pkg/ovs/openflow/testing/mock_openflow.go
+++ b/pkg/ovs/openflow/testing/mock_openflow.go
@@ -134,6 +134,20 @@ func (mr *MockBridgeMockRecorder) CreateGroup(arg0 interface{}) *gomock.Call {
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CreateGroup", reflect.TypeOf((*MockBridge)(nil).CreateGroup), arg0)
 }
 
+// CreateGroupTypeAll mocks base method
+func (m *MockBridge) CreateGroupTypeAll(arg0 openflow.GroupIDType) openflow.Group {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "CreateGroupTypeAll", arg0)
+	ret0, _ := ret[0].(openflow.Group)
+	return ret0
+}
+
+// CreateGroupTypeAll indicates an expected call of CreateGroupTypeAll
+func (mr *MockBridgeMockRecorder) CreateGroupTypeAll(arg0 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CreateGroupTypeAll", reflect.TypeOf((*MockBridge)(nil).CreateGroupTypeAll), arg0)
+}
+
 // CreateMeter mocks base method
 func (m *MockBridge) CreateMeter(arg0 openflow.MeterIDType, arg1 ofctrl.MeterFlag) openflow.Meter {
 	m.ctrl.T.Helper()
diff --git a/pkg/util/channel/channel_test.go b/pkg/util/channel/channel_test.go
index 2eee503f7..e848bd019 100644
--- a/pkg/util/channel/channel_test.go
+++ b/pkg/util/channel/channel_test.go
@@ -36,10 +36,10 @@ func newEventReceiver() *eventReceiver {
 	}
 }
 
-func (r *eventReceiver) receive(e string) {
+func (r *eventReceiver) receive(e interface{}) {
 	r.mutex.Lock()
 	defer r.mutex.Unlock()
-	r.receivedEvents.Insert(e)
+	r.receivedEvents.Insert(e.(string))
 }
 
 func (r *eventReceiver) received() sets.String {
diff --git a/test/e2e/framework.go b/test/e2e/framework.go
index e5dc0f213..9a8121103 100644
--- a/test/e2e/framework.go
+++ b/test/e2e/framework.go
@@ -379,12 +379,19 @@ func labelNodeRoleControlPlane() string {
 	return labelNodeRoleControlPlane
 }
 
-func controlPlaneNoScheduleToleration() corev1.Toleration {
+func controlPlaneNoScheduleTolerations() []corev1.Toleration {
 	// the Node taint still uses "master" in K8s v1.20
-	return corev1.Toleration{
-		Key:      "node-role.kubernetes.io/master",
-		Operator: corev1.TolerationOpExists,
-		Effect:   corev1.TaintEffectNoSchedule,
+	return []corev1.Toleration{
+		{
+			Key:      "node-role.kubernetes.io/master",
+			Operator: corev1.TolerationOpExists,
+			Effect:   corev1.TaintEffectNoSchedule,
+		},
+		{
+			Key:      "node-role.kubernetes.io/control-plane",
+			Operator: corev1.TolerationOpExists,
+			Effect:   corev1.TaintEffectNoSchedule,
+		},
 	}
 }
 
@@ -417,6 +424,15 @@ func (data *TestData) RunCommandOnNodeExt(nodeName, cmd string, envs map[string]
 }
 
 func (data *TestData) collectClusterInfo() error {
+	// retrieve K8s server version
+	// this needs to be done first, as there may be dependencies on the
+	// version later in this function (e.g., for labelNodeRoleControlPlane()).
+	serverVersion, err := testData.clientset.Discovery().ServerVersion()
+	if err != nil {
+		return err
+	}
+	clusterInfo.k8sServerVersion = serverVersion.String()
+
 	// retrieve Node information
 	nodes, err := testData.clientset.CoreV1().Nodes().List(context.TODO(), metav1.ListOptions{})
 	if err != nil {
@@ -560,14 +576,6 @@ func (data *TestData) collectClusterInfo() error {
 	clusterInfo.svcV4NetworkCIDR = svcCIDRs[0]
 	clusterInfo.svcV6NetworkCIDR = svcCIDRs[1]
 
-	// retrieve K8s server version
-
-	serverVersion, err := testData.clientset.Discovery().ServerVersion()
-	if err != nil {
-		return err
-	}
-	clusterInfo.k8sServerVersion = serverVersion.String()
-
 	// Retrieve kubernetes Service host and Port
 	svc, err := testData.clientset.CoreV1().Services("default").Get(context.TODO(), "kubernetes", metav1.GetOptions{})
 	if err != nil {
@@ -1118,8 +1126,7 @@ func (data *TestData) CreatePodOnNodeInNamespace(name, ns string, nodeName, ctrN
 	}
 	if nodeName == controlPlaneNodeName() {
 		// tolerate NoSchedule taint if we want Pod to run on control-plane Node
-		noScheduleToleration := controlPlaneNoScheduleToleration()
-		podSpec.Tolerations = []corev1.Toleration{noScheduleToleration}
+		podSpec.Tolerations = controlPlaneNoScheduleTolerations()
 	}
 	pod := &corev1.Pod{
 		ObjectMeta: metav1.ObjectMeta{
@@ -2389,9 +2396,7 @@ func (data *TestData) createAgnhostPodOnNodeWithAnnotations(name string, ns stri
 
 func (data *TestData) createDaemonSet(name string, ns string, ctrName string, image string, cmd []string, args []string) (*appsv1.DaemonSet, func() error, error) {
 	podSpec := corev1.PodSpec{
-		Tolerations: []corev1.Toleration{
-			controlPlaneNoScheduleToleration(),
-		},
+		Tolerations: controlPlaneNoScheduleTolerations(),
 		Containers: []corev1.Container{
 			{
 				Name:            ctrName,
@@ -2463,9 +2468,7 @@ func (data *TestData) waitForDaemonSetPods(timeout time.Duration, dsName string,
 
 func (data *TestData) createStatefulSet(name string, ns string, size int32, ctrName string, image string, cmd []string, args []string, mutateFunc func(*appsv1.StatefulSet)) (*appsv1.StatefulSet, func() error, error) {
 	podSpec := corev1.PodSpec{
-		Tolerations: []corev1.Toleration{
-			controlPlaneNoScheduleToleration(),
-		},
+		Tolerations: controlPlaneNoScheduleTolerations(),
 		Containers: []corev1.Container{
 			{
 				Name:            ctrName,
diff --git a/test/e2e/multicast_test.go b/test/e2e/multicast_test.go
index 31a49751b..0d7755550 100644
--- a/test/e2e/multicast_test.go
+++ b/test/e2e/multicast_test.go
@@ -83,6 +83,7 @@ func TestMulticast(t *testing.T) {
 		for _, mc := range testcases {
 			mc := mc
 			t.Run(mc.name, func(t *testing.T) {
+				t.Parallel()
 				runTestMulticastBetweenPods(t, data, mc, nodeMulticastInterfaces)
 			})
 		}
@@ -122,6 +123,7 @@ func TestMulticast(t *testing.T) {
 		for _, mc := range testcases {
 			mc := mc
 			t.Run(mc.name, func(t *testing.T) {
+				t.Parallel()
 				runTestMulticastBetweenPods(t, data, mc, nodeMulticastInterfaces)
 			})
 		}
@@ -189,7 +191,7 @@ func runTestMulticastForwardToMultipleInterfaces(t *testing.T, data *TestData, s
 	}
 }
 
-func runTestMulticastBetweenPods(t *testing.T, data *TestData, mc multicastTestcase, nodeMulticastInterfaces [][]string) {
+func runTestMulticastBetweenPods(t *testing.T, data *TestData, mc multicastTestcase, nodeMulticastInterfaces map[int][]string) {
 	mcjoinWaitTimeout := defaultTimeout / time.Second
 	gatewayInterface, err := data.GetGatewayInterfaceName(antreaNamespace)
 	failOnError(err, t)
@@ -209,10 +211,10 @@ func runTestMulticastBetweenPods(t *testing.T, data *TestData, mc multicastTestc
 			defer wg.Done()
 			// The following command joins a multicast group and sets the timeout to 100 seconds(-W 100) before exit.
 			// The command will return after receiving 1 packet(-c 1).
-			receiveMulticastCommand := []string{"/bin/sh", "-c", fmt.Sprintf("mcjoin -c 1 -o -p %d -W %d %s", mc.port, mcjoinWaitTimeout, mc.group.String())}
+			receiveMulticastCommand := []string{"/bin/sh", "-c", fmt.Sprintf("mcjoin -c 10 -o -p %d -W %d %s", mc.port, mcjoinWaitTimeout, mc.group.String())}
 			res, _, err := data.RunCommandFromPod(testNamespace, r, mcjoinContainerName, receiveMulticastCommand)
 			failOnError(err, t)
-			assert.Contains(t, res, "Total: 1 packets")
+			assert.Contains(t, res, "Total: 10 packets")
 		}()
 	}
 	// Wait 2 seconds(-w 2) before sending multicast traffic.
@@ -222,23 +224,32 @@ func runTestMulticastBetweenPods(t *testing.T, data *TestData, mc multicastTestc
 		data.RunCommandFromPod(testNamespace, senderName, mcjoinContainerName, sendMulticastCommand)
 	}()
 
-	if err := wait.Poll(5*time.Second, defaultTimeout, func() (bool, error) {
-		// Sender pods should add an outbound multicast route except running as HostNetwork.
-		_, mrouteResult, _, err := data.RunCommandOnNode(nodeName(mc.senderConfig.nodeIdx), fmt.Sprintf("ip mroute show to %s iif %s | grep '%s'", mc.group.String(), gatewayInterface, strings.Join(nodeMulticastInterfaces[mc.senderConfig.nodeIdx], " ")))
-		if err != nil {
-			return false, err
-		}
-		if !mc.senderConfig.isHostNetwork {
-			if len(mrouteResult) == 0 {
-				return false, nil
+	readyReceivers := sets.NewInt()
+	senderReady := false
+	if err := wait.Poll(3*time.Second, defaultTimeout, func() (bool, error) {
+		if !senderReady {
+			// Sender pods should add an outbound multicast route except running as HostNetwork.
+			_, mrouteResult, _, err := data.RunCommandOnNode(nodeName(mc.senderConfig.nodeIdx), fmt.Sprintf("ip mroute show to %s iif %s | grep '%s'", mc.group.String(), gatewayInterface, strings.Join(nodeMulticastInterfaces[mc.senderConfig.nodeIdx], " ")))
+			if err != nil {
+				return false, err
 			}
-		} else {
-			if len(mrouteResult) != 0 {
-				return false, nil
+			if !mc.senderConfig.isHostNetwork {
+				if len(mrouteResult) == 0 {
+					return false, nil
+				}
+			} else {
+				if len(mrouteResult) != 0 {
+					return false, nil
+				}
 			}
+			senderReady = true
 		}
+
 		// Check inbound multicast route and whether multicast interfaces has joined the multicast group.
 		for _, receiver := range mc.receiverConfigs {
+			if readyReceivers.Has(receiver.nodeIdx) {
+				continue
+			}
 			for _, receiverMulticastInterface := range nodeMulticastInterfaces[receiver.nodeIdx] {
 				_, mRouteResult, _, err := data.RunCommandOnNode(nodeName(receiver.nodeIdx), fmt.Sprintf("ip mroute show to %s iif %s ", mc.group.String(), receiverMulticastInterface))
 				if err != nil {
@@ -272,6 +283,7 @@ func runTestMulticastBetweenPods(t *testing.T, data *TestData, mc multicastTestc
 					}
 				}
 			}
+			readyReceivers = readyReceivers.Insert(receiver.nodeIdx)
 		}
 		return true, nil
 	}); err != nil {
@@ -282,7 +294,7 @@ func runTestMulticastBetweenPods(t *testing.T, data *TestData, mc multicastTestc
 
 // computeMulticastInterfaces computes multicastInterfaces for each node.
 // It returns [][]string with its index as node index and value as multicastInterfaces for this node.
-func computeMulticastInterfaces(t *testing.T, data *TestData) ([][]string, error) {
+func computeMulticastInterfaces(t *testing.T, data *TestData) (map[int][]string, error) {
 	multicastInterfaces, err := data.GetMulticastInterfaces(antreaNamespace)
 	if err != nil {
 		return nil, err
@@ -291,7 +303,7 @@ func computeMulticastInterfaces(t *testing.T, data *TestData) ([][]string, error
 	if err != nil {
 		t.Fatalf("Error getting transport interfaces: %v", err)
 	}
-	nodeMulticastInterfaces := make([][]string, 0, len(clusterInfo.nodes))
+	nodeMulticastInterfaces := make(map[int][]string)
 	for nodeIdx := range clusterInfo.nodes {
 		_, localInterfacesStr, _, err := data.RunCommandOnNode(nodeName(nodeIdx), "ls /sys/class/net")
 		if err != nil {
@@ -303,7 +315,7 @@ func computeMulticastInterfaces(t *testing.T, data *TestData) ([][]string, error
 		externalMulticastInterfaces := localInterfacesSet.Intersection(multicastInterfaceSet)
 		currNodeMulticastInterfaces := externalMulticastInterfaces.Insert(transportInterface).List()
 		t.Logf("Multicast interfaces for node index %d is %+v", nodeIdx, currNodeMulticastInterfaces)
-		nodeMulticastInterfaces = append(nodeMulticastInterfaces, currNodeMulticastInterfaces)
+		nodeMulticastInterfaces[nodeIdx] = currNodeMulticastInterfaces
 	}
 	return nodeMulticastInterfaces, nil
 }
diff --git a/test/e2e/networkpolicy_test.go b/test/e2e/networkpolicy_test.go
index 313ad31b1..c4d3939f4 100644
--- a/test/e2e/networkpolicy_test.go
+++ b/test/e2e/networkpolicy_test.go
@@ -816,8 +816,7 @@ func testIngressPolicyWithEndPort(t *testing.T, data *TestData) {
 		}
 		if nodeName == controlPlaneNodeName() {
 			// tolerate NoSchedule taint if we want Pod to run on control-plane Node
-			noScheduleToleration := controlPlaneNoScheduleToleration()
-			podSpec.Tolerations = []corev1.Toleration{noScheduleToleration}
+			podSpec.Tolerations = controlPlaneNoScheduleTolerations()
 		}
 		pod := &corev1.Pod{
 			ObjectMeta: metav1.ObjectMeta{
diff --git a/test/e2e/performance_test.go b/test/e2e/performance_test.go
index 02b32f666..c8b2c61de 100644
--- a/test/e2e/performance_test.go
+++ b/test/e2e/performance_test.go
@@ -50,8 +50,6 @@ var (
 	customizePolicyRules = flag.Int("perf.http.policy_rules", 0, "Number of CIDRs in the network policy")
 	httpConcurrency      = flag.Int("perf.http.concurrency", 1, "Number of multiple requests to make at a time")
 	realizeTimeout       = flag.Duration("perf.realize.timeout", 5*time.Minute, "Timeout of the realization of network policies")
-	// tolerate NoSchedule taint to let the Pod run on control-plane Node
-	noScheduleToleration = controlPlaneNoScheduleToleration()
 	labelSelector        = &metav1.LabelSelector{
 		MatchLabels: map[string]string{"app": perfTestAppLabel},
 	}
@@ -118,7 +116,7 @@ func createPerfTestPodDefinition(name, containerName, image string) *corev1.Pod
 		"kubernetes.io/hostname": controlPlaneNodeName(),
 	}
 
-	podSpec.Tolerations = []corev1.Toleration{noScheduleToleration}
+	podSpec.Tolerations = controlPlaneNoScheduleTolerations()
 	pod := &corev1.Pod{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:   name,
diff --git a/test/e2e/prometheus_test.go b/test/e2e/prometheus_test.go
index 08571fd34..b3e1732b1 100644
--- a/test/e2e/prometheus_test.go
+++ b/test/e2e/prometheus_test.go
@@ -109,20 +109,14 @@ func skipIfPrometheusDisabled(t *testing.T) {
 // getMonitoringAuthToken retrieves monitoring authorization token, required for access to Antrea apiserver/metrics
 // resource
 func getMonitoringAuthToken(t *testing.T, data *TestData) string {
-	secrets, err := data.clientset.CoreV1().Secrets(monitoringNamespace).List(context.TODO(), metav1.ListOptions{})
+	const secretName = "prometheus-service-account-token"
+	secret, err := data.clientset.CoreV1().Secrets(monitoringNamespace).Get(context.TODO(), secretName, metav1.GetOptions{})
 	if err != nil {
-		t.Fatalf("Error fetching monitoring secrets: %v", err)
+		t.Fatalf("Error fetching monitoring secret '%s': %v", secretName, err)
 	}
-
-	var token string
-	for _, secret := range secrets.Items {
-		if secret.Annotations["kubernetes.io/service-account.name"] == "prometheus" {
-			token = string(secret.Data[v1.ServiceAccountTokenKey])
-		}
-	}
-
+	token := string(secret.Data[v1.ServiceAccountTokenKey])
 	if len(token) == 0 {
-		t.Fatal("Prometheus ServiceAccount secret not found")
+		t.Fatalf("Monitoring secret '%s' does not include token", secretName)
 	}
 
 	return token
diff --git a/test/integration/agent/openflow_test.go b/test/integration/agent/openflow_test.go
index 528aba271..e7c9d557b 100644
--- a/test/integration/agent/openflow_test.go
+++ b/test/integration/agent/openflow_test.go
@@ -704,7 +704,7 @@ func uninstallServiceFlowsFunc(t *testing.T, gid uint32, svc svcConfig, endpoint
 	groupID := ofconfig.GroupIDType(gid)
 	err := c.UninstallServiceFlows(svc.ip, svc.port, svc.protocol)
 	assert.Nil(t, err)
-	err = c.UninstallServiceGroup(groupID)
+	err = c.UninstallGroup(groupID)
 	assert.Nil(t, err)
 	for _, ep := range endpointList {
 		err := c.UninstallEndpointFlows(svc.protocol, ep)
diff --git a/test/integration/ovs/ofctrl_test.go b/test/integration/ovs/ofctrl_test.go
index b646cceeb..73769a10e 100644
--- a/test/integration/ovs/ofctrl_test.go
+++ b/test/integration/ovs/ofctrl_test.go
@@ -269,6 +269,8 @@ func TestOFctrlGroup(t *testing.T) {
 
 	ovsCtlClient := ovsctl.NewClient(brName)
 
+	id := 1
+
 	for name, buckets := range map[string][]struct {
 		weight        uint16      // Must have non-zero value.
 		reg2reg       [][4]uint32 // regNum, data, startIndex, endIndex
@@ -278,12 +280,25 @@ func TestOFctrlGroup(t *testing.T) {
 			{weight: 100, reg2reg: [][4]uint32{{0, 1, 0, 31}, {1, 2, 15, 31}}, resubmitTable: 31},
 			{weight: 110, resubmitTable: 42},
 		},
+		"TypeAll": {
+			{weight: 100, reg2reg: [][4]uint32{{0, 1, 0, 31}, {1, 2, 15, 31}}, resubmitTable: 31},
+			{weight: 110, resubmitTable: 42},
+		},
 	} {
 		t.Run(name, func(t *testing.T) {
-			group := br.CreateGroup(1)
+			var group binding.Group
+			gid := binding.GroupIDType(id)
+			if name == "TypeAll" {
+				group = br.CreateGroupTypeAll(gid)
+			} else {
+				group = br.CreateGroup(gid)
+			}
 			for _, bucket := range buckets {
 				require.NotZero(t, bucket.weight, "Weight value of a bucket must be specified")
-				bucketBuilder := group.Bucket().Weight(bucket.weight)
+				bucketBuilder := group.Bucket()
+				if name == "Normal" {
+					bucketBuilder = bucketBuilder.Weight(bucket.weight)
+				}
 				if bucket.resubmitTable != 0 {
 					bucketBuilder = bucketBuilder.ResubmitToTable(bucket.resubmitTable)
 				}
@@ -303,8 +318,10 @@ func TestOFctrlGroup(t *testing.T) {
 			}), "Failed to install group")
 			dumpedGroup := groups[0]
 			for i, bucket := range buckets {
-				// Must have weight
-				assert.True(t, strings.Contains(dumpedGroup[i+1], fmt.Sprintf("weight:%d", bucket.weight)))
+				if name == "Normal" {
+					// Must have weight
+					assert.True(t, strings.Contains(dumpedGroup[i+1], fmt.Sprintf("weight:%d", bucket.weight)))
+				}
 				for _, loading := range bucket.reg2reg {
 					rngStr := "[]"
 					if !(loading[2] == 0 && loading[3] == 31) {
@@ -326,6 +343,7 @@ func TestOFctrlGroup(t *testing.T) {
 				return len(groups) == 0, nil
 			}), "Failed to delete group")
 		})
+		id++
 	}
 }