Repository: antrea-io/antrea
Pull request #3939 (2022-06-24 23:35:55)

Refactor Pod creation for e2e tests
Functions createPodOnNode and CreatePodOnNodeInNamespace have long lists
of parameters, and some of them are almost never used. Their naming is
also confusing since they both let you specify an arbitrary Namespace,
and the only difference is that CreatePodOnNodeInNamespace takes a
container name as parameter, while createPodOnNode auto-generates it
based on the container image name.

We try a different approach, using a "PodBuilder" object. The advantage
is the ability to add new methods to the builder, with explicit names,
instead of adding new function parameters or using the mutate function.

Signed-off-by: Antonin Bas <abas@vmware.com>

codecov-commenter: # [Codecov](https://codecov.io/gh/antrea-io/antrea/pull/3939?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) Report
> Merging [#3939](https://codecov.io/gh/antrea-io/antrea/pull/3939?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) (64a6e6f) into [main](https://codecov.io/gh/antrea-io/antrea/commit/a07959ee419c626e868e1d13511fc86607523660?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) (a07959e) will **increase** coverage by `0.19%`.
> The diff coverage is `n/a`.

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

```diff
@@            Coverage Diff             @@
##             main    #3939      +/-   ##
==========================================
+ Coverage   49.03%   49.23%   +0.19%     
==========================================
  Files         261      414     +153     
  Lines       39245    60392   +21147     
==========================================
+ Hits        19245    29734   +10489     
- Misses      18128    28219   +10091     
- Partials     1872     2439     +567     
```

| Flag | Coverage Δ | |
|---|---|---|
| integration-tests | `35.22% <ø> (?)` | |
| kind-e2e-tests | `32.62% <ø> (-16.41%)` | :arrow_down: |
| unit-tests | `44.53% <ø> (?)` | |

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/3939?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) | Coverage Δ | |
|---|---|---|
| [pkg/agent/nodeportlocal/k8s/annotations.go](https://codecov.io/gh/antrea-io/antrea/pull/3939/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%> (-84.45%)` | :arrow_down: |
| [pkg/monitor/agent.go](https://codecov.io/gh/antrea-io/antrea/pull/3939/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL21vbml0b3IvYWdlbnQuZ28=) | `0.00% <0.00%> (-70.59%)` | :arrow_down: |
| [pkg/agent/nodeportlocal/util/parse\_port.go](https://codecov.io/gh/antrea-io/antrea/pull/3939/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2FnZW50L25vZGVwb3J0bG9jYWwvdXRpbC9wYXJzZV9wb3J0Lmdv) | `0.00% <0.00%> (-70.00%)` | :arrow_down: |
| [pkg/agent/util/ipset/ipset.go](https://codecov.io/gh/antrea-io/antrea/pull/3939/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2FnZW50L3V0aWwvaXBzZXQvaXBzZXQuZ28=) | `0.00% <0.00%> (-67.57%)` | :arrow_down: |
| [pkg/agent/nodeportlocal/k8s/npl\_controller.go](https://codecov.io/gh/antrea-io/antrea/pull/3939/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2FnZW50L25vZGVwb3J0bG9jYWwvazhzL25wbF9jb250cm9sbGVyLmdv) | `0.00% <0.00%> (-61.13%)` | :arrow_down: |
| [pkg/apis/controlplane/v1beta2/helper.go](https://codecov.io/gh/antrea-io/antrea/pull/3939/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2FwaXMvY29udHJvbHBsYW5lL3YxYmV0YTIvaGVscGVyLmdv) | `40.00% <0.00%> (-60.00%)` | :arrow_down: |
| [...agent/controller/traceflow/traceflow\_controller.go](https://codecov.io/gh/antrea-io/antrea/pull/3939/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2FnZW50L2NvbnRyb2xsZXIvdHJhY2VmbG93L3RyYWNlZmxvd19jb250cm9sbGVyLmdv) | `14.03% <0.00%> (-58.93%)` | :arrow_down: |
| [pkg/agent/nodeportlocal/portcache/port\_table.go](https://codecov.io/gh/antrea-io/antrea/pull/3939/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2FnZW50L25vZGVwb3J0bG9jYWwvcG9ydGNhY2hlL3BvcnRfdGFibGUuZ28=) | `8.75% <0.00%> (-58.75%)` | :arrow_down: |
| [pkg/agent/controller/networkpolicy/packetin.go](https://codecov.io/gh/antrea-io/antrea/pull/3939/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2FnZW50L2NvbnRyb2xsZXIvbmV0d29ya3BvbGljeS9wYWNrZXRpbi5nbw==) | `0.00% <0.00%> (-57.75%)` | :arrow_down: |
| [pkg/agent/nodeportlocal/npl\_agent\_init.go](https://codecov.io/gh/antrea-io/antrea/pull/3939/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2FnZW50L25vZGVwb3J0bG9jYWwvbnBsX2FnZW50X2luaXQuZ28=) | `0.00% <0.00%> (-57.15%)` | :arrow_down: |
| ... and [317 more](https://codecov.io/gh/antrea-io/antrea/pull/3939/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) | |


Patch:
diff --git a/.github/workflows/kind.yml b/.github/workflows/kind.yml
index 3ba9f45a4..9de62ee63 100644
--- a/.github/workflows/kind.yml
+++ b/.github/workflows/kind.yml
@@ -149,7 +149,7 @@ jobs:
       run: |
         mkdir log
         mkdir test-e2e-encap-no-proxy-coverage
-        ANTREA_LOG_DIR=$PWD/log ANTREA_COV_DIR=$PWD/test-e2e-encap-no-proxy-coverage ./ci/kind/test-e2e-kind.sh --encap-mode encap --no-proxy --coverage --skip mode-irrelevant
+        ANTREA_LOG_DIR=$PWD/log ANTREA_COV_DIR=$PWD/test-e2e-encap-no-proxy-coverage ./ci/kind/test-e2e-kind.sh --encap-mode encap --feature-gates AntreaProxy=false --coverage --skip mode-irrelevant
     - name: Tar coverage files
       run: tar -czf test-e2e-encap-no-proxy-coverage.tar.gz test-e2e-encap-no-proxy-coverage
     - name: Upload coverage for test-e2e-encap-no-proxy-coverage
@@ -177,10 +177,10 @@ jobs:
         path: log.tar.gz
         retention-days: 30
 
-  test-e2e-encap-proxy-all:
-    name: E2e tests on a Kind cluster on Linux with AntreaProxy all Service support
-    needs: [ build-antrea-coverage-image ]
-    runs-on: [ ubuntu-latest ]
+  test-e2e-encap-all-features-enabled:
+    name: E2e tests on a Kind cluster on Linux with all features enabled
+    needs: [build-antrea-coverage-image]
+    runs-on: [ubuntu-latest]
     steps:
       - name: Free disk space
         # https://github.com/actions/virtual-environments/issues/709
@@ -204,15 +204,16 @@ jobs:
       - name: Run e2e tests
         run: |
           mkdir log
-          mkdir test-e2e-encap-proxy-all-coverage
-          ANTREA_LOG_DIR=$PWD/log ANTREA_COV_DIR=$PWD/test-e2e-encap-proxy-all-coverage ./ci/kind/test-e2e-kind.sh --encap-mode encap --proxy-all --coverage --skip mode-irrelevant
+          mkdir test-e2e-encap-all-features-enabled-coverage
+          # Currently multicast tests require specific testbeds, exclude it for now.
+          ANTREA_LOG_DIR=$PWD/log ANTREA_COV_DIR=$PWD/test-e2e-encap-all-features-enabled-coverage ./ci/kind/test-e2e-kind.sh --encap-mode encap --coverage --feature-gates AllAlpha=true,AllBeta=true,Multicast=false --proxy-all
       - name: Tar coverage files
-        run: tar -czf test-e2e-encap-proxy-all-coverage.tar.gz test-e2e-encap-proxy-all-coverage
-      - name: Upload coverage for test-e2e-encap-proxy-all-coverage
+        run: tar -czf test-e2e-encap-all-features-enabled-coverage.tar.gz test-e2e-encap-all-features-enabled-coverage
+      - name: Upload coverage for test-e2e-encap-all-features-enabled-coverage
         uses: actions/upload-artifact@v3
         with:
-          name: test-e2e-encap-proxy-all-coverage
-          path: test-e2e-encap-proxy-all-coverage.tar.gz
+          name: test-e2e-encap-all-features-enabled-coverage
+          path: test-e2e-encap-all-features-enabled-coverage.tar.gz
           retention-days: 30
       - name: Codecov
         uses: codecov/codecov-action@v3
@@ -220,8 +221,8 @@ jobs:
           token: ${{ secrets.CODECOV_TOKEN }}
           file: '*.cov.out*'
           flags: kind-e2e-tests
-          name: codecov-test-e2e-encap-proxy-all
-          directory: test-e2e-encap-proxy-all-coverage
+          name: codecov-test-e2e-encap-all-features-enabled
+          directory: test-e2e-encap-all-features-enabled-coverage
       - name: Tar log files
         if: ${{ failure() }}
         run: tar -czf log.tar.gz log
@@ -229,7 +230,7 @@ jobs:
         uses: actions/upload-artifact@v3
         if: ${{ failure() }}
         with:
-          name: e2e-kind-encap-proxy-all.tar.gz
+          name: e2e-kind-encap-all-features-enabled.tar.gz
           path: log.tar.gz
           retention-days: 30
 
@@ -345,65 +346,6 @@ jobs:
         path: log.tar.gz
         retention-days: 30
 
-  # TODO: remove when https://github.com/antrea-io/antrea/issues/897 is fixed.
-  # In the mean time, we keep this test around to ensure that at least one Kind
-  # test uses a Geneve overlay.
-  test-e2e-encap-no-np:
-    name: E2e tests on a Kind cluster on Linux with Antrea-native policies disabled
-    needs: [build-antrea-coverage-image]
-    runs-on: [ubuntu-latest]
-    steps:
-    - name: Free disk space
-      # https://github.com/actions/virtual-environments/issues/709
-      run: |
-        sudo apt-get clean
-        df -h
-    - uses: actions/checkout@v3
-    - uses: actions/setup-go@v3
-      with:
-        go-version: 1.17
-    - name: Download Antrea images from previous jobs
-      uses: actions/download-artifact@v3
-    - name: Load Antrea image
-      run: |
-        docker load -i antrea-ubuntu-cov/antrea-ubuntu.tar
-    - name: Install Kind
-      run: |
-        curl -Lo ./kind https://github.com/kubernetes-sigs/kind/releases/download/${KIND_VERSION}/kind-$(uname)-amd64
-        chmod +x ./kind
-        sudo mv kind /usr/local/bin
-    - name: Run e2e tests
-      run: |
-        mkdir log
-        mkdir test-e2e-encap-no-np-coverage
-        ANTREA_LOG_DIR=$PWD/log ANTREA_COV_DIR=$PWD/test-e2e-encap-no-np-coverage ./ci/kind/test-e2e-kind.sh --encap-mode encap --no-np --coverage --skip mode-irrelevant
-    - name: Tar coverage files
-      run: tar -czf test-e2e-encap-no-np-coverage.tar.gz test-e2e-encap-no-np-coverage
-    - name: Upload coverage for test-e2e-encap-no-np-coverage
-      uses: actions/upload-artifact@v3
-      with:
-        name: test-e2e-encap-no-np-coverage
-        path: test-e2e-encap-no-np-coverage.tar.gz
-        retention-days: 30
-    - name: Codecov
-      uses: codecov/codecov-action@v3
-      with:
-        token: ${{ secrets.CODECOV_TOKEN }}
-        file: '*.cov.out*'
-        flags: kind-e2e-tests
-        name: codecov-test-e2e-no-np-encap
-        directory: test-e2e-encap-no-np-coverage
-    - name: Tar log files
-      if: ${{ failure() }}
-      run: tar -czf log.tar.gz log
-    - name: Upload test log
-      uses: actions/upload-artifact@v3
-      if: ${{ failure() }}
-      with:
-        name: e2e-kind-encap-no-np.tar.gz
-        path: log.tar.gz
-        retention-days: 30
-
   test-e2e-flow-visibility:
     name: E2e tests on a Kind cluster on Linux for Flow Visibility
     needs: [build-antrea-coverage-image, build-flow-aggregator-coverage-image]
@@ -536,7 +478,7 @@ jobs:
   # yet.
   artifact-cleanup:
     name: Delete uploaded images
-    needs: [build-antrea-coverage-image, build-flow-aggregator-coverage-image, test-e2e-encap, test-e2e-encap-no-proxy, test-e2e-encap-proxy-all, test-e2e-noencap, test-e2e-hybrid, test-e2e-encap-no-np, test-netpol-tmp, validate-prometheus-metrics-doc, test-e2e-flow-visibility]
+    needs: [build-antrea-coverage-image, build-flow-aggregator-coverage-image, test-e2e-encap, test-e2e-encap-no-proxy, test-e2e-encap-all-features-enabled, test-e2e-noencap, test-e2e-hybrid, test-netpol-tmp, validate-prometheus-metrics-doc, test-e2e-flow-visibility]
     if: ${{ always() &&  (needs.build-antrea-coverage-image.result == 'success' || needs.build-flow-aggregator-coverage-image.result == 'success') }}
     runs-on: [ubuntu-latest]
     steps:
diff --git a/build/charts/antrea/conf/antrea-agent.conf b/build/charts/antrea/conf/antrea-agent.conf
index e41025688..6b3d5f198 100644
--- a/build/charts/antrea/conf/antrea-agent.conf
+++ b/build/charts/antrea/conf/antrea-agent.conf
@@ -1,5 +1,11 @@
 # FeatureGates is a map of feature names to bools that enable or disable experimental features.
 featureGates:
+# AllAlpha is a global toggle for alpha features. Per-feature key values override the default set by AllAlpha.
+{{- include "featureGate" (dict "featureGates" .Values.featureGates "name" "AllAlpha" "default" false) }}
+
+# AllBeta is a global toggle for beta features. Per-feature key values override the default set by AllBeta.
+{{- include "featureGate" (dict "featureGates" .Values.featureGates "name" "AllBeta" "default" false) }}
+
 # Enable AntreaProxy which provides ServiceLB for in-cluster Services in antrea-agent.
 # It should be enabled on Windows, otherwise NetworkPolicy will not take effect on
 # Service traffic.
diff --git a/build/charts/antrea/conf/antrea-controller.conf b/build/charts/antrea/conf/antrea-controller.conf
index f3d89e136..440952a87 100644
--- a/build/charts/antrea/conf/antrea-controller.conf
+++ b/build/charts/antrea/conf/antrea-controller.conf
@@ -1,5 +1,11 @@
 # FeatureGates is a map of feature names to bools that enable or disable experimental features.
 featureGates:
+# AllAlpha is a global toggle for alpha features. Per-feature key values override the default set by AllAlpha.
+{{- include "featureGate" (dict "featureGates" .Values.featureGates "name" "AllAlpha" "default" false) }}
+
+# AllBeta is a global toggle for beta features. Per-feature key values override the default set by AllBeta.
+{{- include "featureGate" (dict "featureGates" .Values.featureGates "name" "AllBeta" "default" false) }}
+
 # Enable traceflow which provides packet tracing feature to diagnose network issue.
 {{- include "featureGate" (dict "featureGates" .Values.featureGates "name" "Traceflow" "default" true) }}
 
diff --git a/build/yamls/antrea-aks.yml b/build/yamls/antrea-aks.yml
index 613d25759..bb41cc558 100644
--- a/build/yamls/antrea-aks.yml
+++ b/build/yamls/antrea-aks.yml
@@ -2537,6 +2537,12 @@ data:
   antrea-agent.conf: |
     # FeatureGates is a map of feature names to bools that enable or disable experimental features.
     featureGates:
+    # AllAlpha is a global toggle for alpha features. Per-feature key values override the default set by AllAlpha.
+    #  AllAlpha: false
+
+    # AllBeta is a global toggle for beta features. Per-feature key values override the default set by AllBeta.
+    #  AllBeta: false
+
     # Enable AntreaProxy which provides ServiceLB for in-cluster Services in antrea-agent.
     # It should be enabled on Windows, otherwise NetworkPolicy will not take effect on
     # Service traffic.
@@ -2845,6 +2851,12 @@ data:
   antrea-controller.conf: |
     # FeatureGates is a map of feature names to bools that enable or disable experimental features.
     featureGates:
+    # AllAlpha is a global toggle for alpha features. Per-feature key values override the default set by AllAlpha.
+    #  AllAlpha: false
+
+    # AllBeta is a global toggle for beta features. Per-feature key values override the default set by AllBeta.
+    #  AllBeta: false
+
     # Enable traceflow which provides packet tracing feature to diagnose network issue.
     #  Traceflow: true
 
@@ -3676,7 +3688,7 @@ spec:
         kubectl.kubernetes.io/default-container: antrea-agent
         # Automatically restart Pods with a RollingUpdate if the ConfigMap changes
         # See https://helm.sh/docs/howto/charts_tips_and_tricks/#automatically-roll-deployments
-        checksum/config: a4e7182f54a10234001d8baeb7d950a5053ffb97b2f6951db516244ac3620cd8
+        checksum/config: 890f1364c9b89811375830c94fab2fa9f1957518351cc52c623e22b6964e5e75
       labels:
         app: antrea
         component: antrea-agent
@@ -3916,7 +3928,7 @@ spec:
       annotations:
         # Automatically restart Pod if the ConfigMap changes
         # See https://helm.sh/docs/howto/charts_tips_and_tricks/#automatically-roll-deployments
-        checksum/config: a4e7182f54a10234001d8baeb7d950a5053ffb97b2f6951db516244ac3620cd8
+        checksum/config: 890f1364c9b89811375830c94fab2fa9f1957518351cc52c623e22b6964e5e75
       labels:
         app: antrea
         component: antrea-controller
diff --git a/build/yamls/antrea-eks.yml b/build/yamls/antrea-eks.yml
index bf90bf557..45eb18a7d 100644
--- a/build/yamls/antrea-eks.yml
+++ b/build/yamls/antrea-eks.yml
@@ -2537,6 +2537,12 @@ data:
   antrea-agent.conf: |
     # FeatureGates is a map of feature names to bools that enable or disable experimental features.
     featureGates:
+    # AllAlpha is a global toggle for alpha features. Per-feature key values override the default set by AllAlpha.
+    #  AllAlpha: false
+
+    # AllBeta is a global toggle for beta features. Per-feature key values override the default set by AllBeta.
+    #  AllBeta: false
+
     # Enable AntreaProxy which provides ServiceLB for in-cluster Services in antrea-agent.
     # It should be enabled on Windows, otherwise NetworkPolicy will not take effect on
     # Service traffic.
@@ -2845,6 +2851,12 @@ data:
   antrea-controller.conf: |
     # FeatureGates is a map of feature names to bools that enable or disable experimental features.
     featureGates:
+    # AllAlpha is a global toggle for alpha features. Per-feature key values override the default set by AllAlpha.
+    #  AllAlpha: false
+
+    # AllBeta is a global toggle for beta features. Per-feature key values override the default set by AllBeta.
+    #  AllBeta: false
+
     # Enable traceflow which provides packet tracing feature to diagnose network issue.
     #  Traceflow: true
 
@@ -3676,7 +3688,7 @@ spec:
         kubectl.kubernetes.io/default-container: antrea-agent
         # Automatically restart Pods with a RollingUpdate if the ConfigMap changes
         # See https://helm.sh/docs/howto/charts_tips_and_tricks/#automatically-roll-deployments
-        checksum/config: a4e7182f54a10234001d8baeb7d950a5053ffb97b2f6951db516244ac3620cd8
+        checksum/config: 890f1364c9b89811375830c94fab2fa9f1957518351cc52c623e22b6964e5e75
       labels:
         app: antrea
         component: antrea-agent
@@ -3918,7 +3930,7 @@ spec:
       annotations:
         # Automatically restart Pod if the ConfigMap changes
         # See https://helm.sh/docs/howto/charts_tips_and_tricks/#automatically-roll-deployments
-        checksum/config: a4e7182f54a10234001d8baeb7d950a5053ffb97b2f6951db516244ac3620cd8
+        checksum/config: 890f1364c9b89811375830c94fab2fa9f1957518351cc52c623e22b6964e5e75
       labels:
         app: antrea
         component: antrea-controller
diff --git a/build/yamls/antrea-gke.yml b/build/yamls/antrea-gke.yml
index 82bf6c0bf..75f23b9a4 100644
--- a/build/yamls/antrea-gke.yml
+++ b/build/yamls/antrea-gke.yml
@@ -2537,6 +2537,12 @@ data:
   antrea-agent.conf: |
     # FeatureGates is a map of feature names to bools that enable or disable experimental features.
     featureGates:
+    # AllAlpha is a global toggle for alpha features. Per-feature key values override the default set by AllAlpha.
+    #  AllAlpha: false
+
+    # AllBeta is a global toggle for beta features. Per-feature key values override the default set by AllBeta.
+    #  AllBeta: false
+
     # Enable AntreaProxy which provides ServiceLB for in-cluster Services in antrea-agent.
     # It should be enabled on Windows, otherwise NetworkPolicy will not take effect on
     # Service traffic.
@@ -2845,6 +2851,12 @@ data:
   antrea-controller.conf: |
     # FeatureGates is a map of feature names to bools that enable or disable experimental features.
     featureGates:
+    # AllAlpha is a global toggle for alpha features. Per-feature key values override the default set by AllAlpha.
+    #  AllAlpha: false
+
+    # AllBeta is a global toggle for beta features. Per-feature key values override the default set by AllBeta.
+    #  AllBeta: false
+
     # Enable traceflow which provides packet tracing feature to diagnose network issue.
     #  Traceflow: true
 
@@ -3676,7 +3688,7 @@ spec:
         kubectl.kubernetes.io/default-container: antrea-agent
         # Automatically restart Pods with a RollingUpdate if the ConfigMap changes
         # See https://helm.sh/docs/howto/charts_tips_and_tricks/#automatically-roll-deployments
-        checksum/config: afbe6d81852e79d710d0350ab5173f7d0e05b79c75744cffb94ce2294afc328c
+        checksum/config: cc8af4219d403a137ab87500ae0ab15b681fc635e41057b5623df6154443fddf
       labels:
         app: antrea
         component: antrea-agent
@@ -3916,7 +3928,7 @@ spec:
       annotations:
         # Automatically restart Pod if the ConfigMap changes
         # See https://helm.sh/docs/howto/charts_tips_and_tricks/#automatically-roll-deployments
-        checksum/config: afbe6d81852e79d710d0350ab5173f7d0e05b79c75744cffb94ce2294afc328c
+        checksum/config: cc8af4219d403a137ab87500ae0ab15b681fc635e41057b5623df6154443fddf
       labels:
         app: antrea
         component: antrea-controller
diff --git a/build/yamls/antrea-ipsec.yml b/build/yamls/antrea-ipsec.yml
index a62bb3cc0..785e8f135 100644
--- a/build/yamls/antrea-ipsec.yml
+++ b/build/yamls/antrea-ipsec.yml
@@ -2550,6 +2550,12 @@ data:
   antrea-agent.conf: |
     # FeatureGates is a map of feature names to bools that enable or disable experimental features.
     featureGates:
+    # AllAlpha is a global toggle for alpha features. Per-feature key values override the default set by AllAlpha.
+    #  AllAlpha: false
+
+    # AllBeta is a global toggle for beta features. Per-feature key values override the default set by AllBeta.
+    #  AllBeta: false
+
     # Enable AntreaProxy which provides ServiceLB for in-cluster Services in antrea-agent.
     # It should be enabled on Windows, otherwise NetworkPolicy will not take effect on
     # Service traffic.
@@ -2858,6 +2864,12 @@ data:
   antrea-controller.conf: |
     # FeatureGates is a map of feature names to bools that enable or disable experimental features.
     featureGates:
+    # AllAlpha is a global toggle for alpha features. Per-feature key values override the default set by AllAlpha.
+    #  AllAlpha: false
+
+    # AllBeta is a global toggle for beta features. Per-feature key values override the default set by AllBeta.
+    #  AllBeta: false
+
     # Enable traceflow which provides packet tracing feature to diagnose network issue.
     #  Traceflow: true
 
@@ -3689,7 +3701,7 @@ spec:
         kubectl.kubernetes.io/default-container: antrea-agent
         # Automatically restart Pods with a RollingUpdate if the ConfigMap changes
         # See https://helm.sh/docs/howto/charts_tips_and_tricks/#automatically-roll-deployments
-        checksum/config: 626cd437985469e9b0e55e2faacbab203aee641d8a99ff9831c2bb3319f02e95
+        checksum/config: df5271a5c42a550d3f8e73fbe8e5fad8d178884fb74d81c7322128187546db86
         checksum/ipsec-secret: d0eb9c52d0cd4311b6d252a951126bf9bea27ec05590bed8a394f0f792dcb2a4
       labels:
         app: antrea
@@ -3975,7 +3987,7 @@ spec:
       annotations:
         # Automatically restart Pod if the ConfigMap changes
         # See https://helm.sh/docs/howto/charts_tips_and_tricks/#automatically-roll-deployments
-        checksum/config: 626cd437985469e9b0e55e2faacbab203aee641d8a99ff9831c2bb3319f02e95
+        checksum/config: df5271a5c42a550d3f8e73fbe8e5fad8d178884fb74d81c7322128187546db86
       labels:
         app: antrea
         component: antrea-controller
diff --git a/build/yamls/antrea.yml b/build/yamls/antrea.yml
index 1bb9be1c0..4fcbc4001 100644
--- a/build/yamls/antrea.yml
+++ b/build/yamls/antrea.yml
@@ -2537,6 +2537,12 @@ data:
   antrea-agent.conf: |
     # FeatureGates is a map of feature names to bools that enable or disable experimental features.
     featureGates:
+    # AllAlpha is a global toggle for alpha features. Per-feature key values override the default set by AllAlpha.
+    #  AllAlpha: false
+
+    # AllBeta is a global toggle for beta features. Per-feature key values override the default set by AllBeta.
+    #  AllBeta: false
+
     # Enable AntreaProxy which provides ServiceLB for in-cluster Services in antrea-agent.
     # It should be enabled on Windows, otherwise NetworkPolicy will not take effect on
     # Service traffic.
@@ -2845,6 +2851,12 @@ data:
   antrea-controller.conf: |
     # FeatureGates is a map of feature names to bools that enable or disable experimental features.
     featureGates:
+    # AllAlpha is a global toggle for alpha features. Per-feature key values override the default set by AllAlpha.
+    #  AllAlpha: false
+
+    # AllBeta is a global toggle for beta features. Per-feature key values override the default set by AllBeta.
+    #  AllBeta: false
+
     # Enable traceflow which provides packet tracing feature to diagnose network issue.
     #  Traceflow: true
 
@@ -3676,7 +3688,7 @@ spec:
         kubectl.kubernetes.io/default-container: antrea-agent
         # Automatically restart Pods with a RollingUpdate if the ConfigMap changes
         # See https://helm.sh/docs/howto/charts_tips_and_tricks/#automatically-roll-deployments
-        checksum/config: 750097fa4ed60fda68fedf2f31b06a1a62dbbe5dc9a38e9c1f9ddc861a389401
+        checksum/config: 033b7f8c7b77a918ad1a90c3db034bbfc1df67de264a77f8aee6a035836b6812
       labels:
         app: antrea
         component: antrea-agent
@@ -3916,7 +3928,7 @@ spec:
       annotations:
         # Automatically restart Pod if the ConfigMap changes
         # See https://helm.sh/docs/howto/charts_tips_and_tricks/#automatically-roll-deployments
-        checksum/config: 750097fa4ed60fda68fedf2f31b06a1a62dbbe5dc9a38e9c1f9ddc861a389401
+        checksum/config: 033b7f8c7b77a918ad1a90c3db034bbfc1df67de264a77f8aee6a035836b6812
       labels:
         app: antrea
         component: antrea-controller
diff --git a/build/yamls/flow-aggregator.yml b/build/yamls/flow-aggregator.yml
index 9dbbe4ba7..56ecc83bd 100644
--- a/build/yamls/flow-aggregator.yml
+++ b/build/yamls/flow-aggregator.yml
@@ -96,6 +96,14 @@ rules:
   - secrets
   verbs:
   - create
+- apiGroups:
+  - ""
+  resourceNames:
+  - flow-aggregator-configmap-7572tg842t
+  resources:
+  - configmaps
+  verbs:
+  - update
 ---
 apiVersion: rbac.authorization.k8s.io/v1
 kind: RoleBinding
@@ -301,6 +309,10 @@ spec:
           valueFrom:
             fieldRef:
               fieldPath: metadata.name
+        - name: POD_NAMESPACE
+          valueFrom:
+            fieldRef:
+              fieldPath: metadata.namespace
         - name: CH_USERNAME
           valueFrom:
             secretKeyRef:
@@ -311,16 +323,17 @@ spec:
             secretKeyRef:
               key: password
               name: clickhouse-secret
+        - name: FA_CONFIG_MAP_NAME
+          value: flow-aggregator-configmap-7572tg842t
         image: projects.registry.vmware.com/antrea/flow-aggregator:latest
         imagePullPolicy: IfNotPresent
         name: flow-aggregator
         ports:
         - containerPort: 4739
         volumeMounts:
-        - mountPath: /etc/flow-aggregator/flow-aggregator.conf
+        - mountPath: /etc/flow-aggregator
           name: flow-aggregator-config
           readOnly: true
-          subPath: flow-aggregator.conf
         - mountPath: /var/log/antrea/flow-aggregator
           name: host-var-log-antrea-flow-aggregator
       nodeSelector:
diff --git a/build/yamls/flow-aggregator/base/flow-aggregator.yml b/build/yamls/flow-aggregator/base/flow-aggregator.yml
index 009a2a121..533eb19ee 100644
--- a/build/yamls/flow-aggregator/base/flow-aggregator.yml
+++ b/build/yamls/flow-aggregator/base/flow-aggregator.yml
@@ -42,6 +42,10 @@ rules:
   - apiGroups: [""]
     resources: ["secrets"]
     verbs: ["create"]
+  - apiGroups: [ "" ]
+    resources: [ "configmaps" ]
+    resourceNames: [ "flow-aggregator-configmap" ]
+    verbs: [ "update" ]
 ---
 kind: ClusterRoleBinding
 apiVersion: rbac.authorization.k8s.io/v1
@@ -145,6 +149,10 @@ spec:
             valueFrom:
               fieldRef:
                 fieldPath: metadata.name
+          - name: POD_NAMESPACE
+            valueFrom:
+              fieldRef:
+                fieldPath: metadata.namespace
           - name: CH_USERNAME
             valueFrom:
               secretKeyRef:
@@ -155,13 +163,14 @@ spec:
               secretKeyRef:
                 name: clickhouse-secret
                 key: password
+          - name: FA_CONFIG_MAP_NAME
+            value: "$(FA_CONFIG_MAP_NAME)"
         ports:
           - containerPort: 4739
         volumeMounts:
-        - mountPath: /etc/flow-aggregator/flow-aggregator.conf
+        - mountPath: /etc/flow-aggregator
           name: flow-aggregator-config
           readOnly: true
-          subPath: flow-aggregator.conf
         - mountPath: /var/log/antrea/flow-aggregator
           name: host-var-log-antrea-flow-aggregator
       nodeSelector:
diff --git a/build/yamls/flow-aggregator/base/kustomization.yml b/build/yamls/flow-aggregator/base/kustomization.yml
index 03d8c9dea..961a8fdca 100644
--- a/build/yamls/flow-aggregator/base/kustomization.yml
+++ b/build/yamls/flow-aggregator/base/kustomization.yml
@@ -9,3 +9,9 @@ commonLabels:
 namespace: flow-aggregator
 apiVersion: kustomize.config.k8s.io/v1beta1
 kind: Kustomization
+vars:
+  - name: FA_CONFIG_MAP_NAME
+    objref:
+      kind: ConfigMap
+      name: flow-aggregator-configmap
+      apiVersion: v1
diff --git a/cmd/flow-aggregator/flow-aggregator.go b/cmd/flow-aggregator/flow-aggregator.go
index 976f2e49e..0ba89dc9a 100644
--- a/cmd/flow-aggregator/flow-aggregator.go
+++ b/cmd/flow-aggregator/flow-aggregator.go
@@ -16,22 +16,16 @@ package main
 
 import (
 	"fmt"
-	"hash/fnv"
-	"os"
 	"sync"
 	"time"
 
-	"github.com/google/uuid"
-	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/informers"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 	"k8s.io/klog/v2"
 
-	"antrea.io/antrea/pkg/clusteridentity"
 	aggregator "antrea.io/antrea/pkg/flowaggregator"
 	"antrea.io/antrea/pkg/flowaggregator/apiserver"
-	"antrea.io/antrea/pkg/flowaggregator/clickhouseclient"
 	"antrea.io/antrea/pkg/log"
 	"antrea.io/antrea/pkg/signals"
 	"antrea.io/antrea/pkg/util/cipher"
@@ -39,44 +33,7 @@ import (
 
 const informerDefaultResync = 12 * time.Hour
 
-// genObservationDomainID generates an IPFIX Observation Domain ID when one is not provided by the
-// user through the flow aggregator configuration. It will first try to generate one
-// deterministically based on the cluster UUID (if available, with a timeout of 10s). Otherwise, it
-// will generate a random one. The cluster UUID should be available if Antrea is deployed to the
-// cluster ahead of the flow aggregator, which is the expectation since when deploying flow
-// aggregator as a Pod, networking needs to be configured by the CNI plugin.
-func genObservationDomainID(k8sClient kubernetes.Interface) uint32 {
-	const retryInterval = time.Second
-	const timeout = 10 * time.Second
-	const defaultAntreaNamespace = "kube-system"
-
-	clusterIdentityProvider := clusteridentity.NewClusterIdentityProvider(
-		defaultAntreaNamespace,
-		clusteridentity.DefaultClusterIdentityConfigMapName,
-		k8sClient,
-	)
-	var clusterUUID uuid.UUID
-	if err := wait.PollImmediate(retryInterval, timeout, func() (bool, error) {
-		clusterIdentity, _, err := clusterIdentityProvider.Get()
-		if err != nil {
-			return false, nil
-		}
-		clusterUUID = clusterIdentity.UUID
-		return true, nil
-	}); err != nil {
-		klog.Warningf(
-			"Unable to retrieve cluster UUID after %v (does ConfigMap '%s/%s' exist?); will generate a random observation domain ID",
-			timeout, defaultAntreaNamespace, clusteridentity.DefaultClusterIdentityConfigMapName,
-		)
-		clusterUUID = uuid.New()
-	}
-	h := fnv.New32()
-	h.Write(clusterUUID[:])
-	observationDomainID := h.Sum32()
-	return observationDomainID
-}
-
-func run(o *Options) error {
+func run(configFile string) error {
 	klog.Infof("Flow aggregator starting...")
 	// Set up signal capture: the first SIGTERM / SIGINT signal is handled gracefully and will
 	// cause the stopCh channel to be closed; if another signal is received before the program
@@ -93,72 +50,28 @@ func run(o *Options) error {
 	informerFactory := informers.NewSharedInformerFactory(k8sClient, informerDefaultResync)
 	podInformer := informerFactory.Core().V1().Pods()
 
-	var observationDomainID uint32
-	if o.config.FlowCollector.ObservationDomainID != nil {
-		observationDomainID = *o.config.FlowCollector.ObservationDomainID
-	} else {
-		observationDomainID = genObservationDomainID(k8sClient)
-	}
-	klog.Infof("Flow aggregator Observation Domain ID: %d", observationDomainID)
-
-	var sendJSONRecord bool
-	if o.format == "JSON" {
-		sendJSONRecord = true
-	} else {
-		sendJSONRecord = false
-	}
-
-	flowAggregator := aggregator.NewFlowAggregator(
-		o.externalFlowCollectorAddr,
-		o.externalFlowCollectorProto,
-		o.activeFlowRecordTimeout,
-		o.inactiveFlowRecordTimeout,
-		o.aggregatorTransportProtocol,
-		o.flowAggregatorAddress,
-		o.includePodLabels,
+	flowAggregator, err := aggregator.NewFlowAggregator(
 		k8sClient,
-		observationDomainID,
 		podInformer,
-		sendJSONRecord,
+		configFile,
 	)
-	err = flowAggregator.InitCollectingProcess()
-	if err != nil {
-		return fmt.Errorf("error when creating collecting process: %v", err)
-	}
-	err = flowAggregator.InitAggregationProcess()
-	if err != nil {
-		return fmt.Errorf("error when creating aggregation process: %v", err)
-	}
 
-	if o.config.ClickHouse.Enable {
-		chInput := clickhouseclient.ClickHouseInput{
-			Username:       os.Getenv("CH_USERNAME"),
-			Password:       os.Getenv("CH_PASSWORD"),
-			Database:       o.config.ClickHouse.Database,
-			DatabaseURL:    o.config.ClickHouse.DatabaseURL,
-			Debug:          o.config.ClickHouse.Debug,
-			Compress:       o.config.ClickHouse.Compress,
-			CommitInterval: o.clickHouseCommitInterval,
-		}
-		err = flowAggregator.InitDBExportProcess(chInput)
-		if err != nil {
-			return fmt.Errorf("error when creating db export process: %v", err)
-		}
+	if err != nil {
+		return err
 	}
-
 	var wg sync.WaitGroup
 	wg.Add(1)
 	go flowAggregator.Run(stopCh, &wg)
 
-	cipherSuites, err := cipher.GenerateCipherSuitesList(o.config.APIServer.TLSCipherSuites)
+	cipherSuites, err := cipher.GenerateCipherSuitesList(flowAggregator.APIServer.TLSCipherSuites)
 	if err != nil {
 		return fmt.Errorf("error generating Cipher Suite list: %v", err)
 	}
 	apiServer, err := apiserver.New(
 		flowAggregator,
-		o.config.APIServer.APIPort,
+		flowAggregator.APIServer.APIPort,
 		cipherSuites,
-		cipher.TLSVersionMap[o.config.APIServer.TLSMinVersion])
+		cipher.TLSVersionMap[flowAggregator.APIServer.TLSMinVersion])
 	if err != nil {
 		return fmt.Errorf("error when creating flow aggregator API server: %v", err)
 	}
@@ -167,7 +80,7 @@ func run(o *Options) error {
 	informerFactory.Start(stopCh)
 
 	<-stopCh
-	klog.Infof("Stopping flow aggregator")
+	klog.InfoS("Stopping flow aggregator")
 	wg.Wait()
 	return nil
 }
diff --git a/cmd/flow-aggregator/main.go b/cmd/flow-aggregator/main.go
index 4868c0820..9ee62175e 100644
--- a/cmd/flow-aggregator/main.go
+++ b/cmd/flow-aggregator/main.go
@@ -35,7 +35,6 @@ func main() {
 }
 
 func newFlowAggregatorCommand() *cobra.Command {
-	opts := newOptions()
 
 	cmd := &cobra.Command{
 		Use:  "flow-aggregator",
@@ -43,21 +42,18 @@ func newFlowAggregatorCommand() *cobra.Command {
 		Run: func(cmd *cobra.Command, args []string) {
 			log.InitLogs(cmd.Flags())
 			defer log.FlushLogs()
-			if err := opts.complete(args); err != nil {
-				klog.Fatalf("Failed to complete args: %v", err)
+			configFile, err := cmd.Flags().GetString("config")
+			if err != nil {
+				klog.Fatalf("Error when finding the path of config: %v", err)
 			}
-			if err := opts.validate(args); err != nil {
-				klog.Fatalf("Failed to validate args: %v", err)
-			}
-			if err := run(opts); err != nil {
+			if err := run(configFile); err != nil {
 				klog.Fatalf("Error running flow aggregator: %v", err)
 			}
 		},
 		Version: version.GetFullVersionWithRuntimeInfo(),
 	}
-
 	flags := cmd.Flags()
-	opts.addFlags(flags)
+	flags.String("config", "", "The path to the configuration file")
 	log.AddFlags(flags)
 	return cmd
 }
diff --git a/cmd/flow-aggregator/options.go b/cmd/flow-aggregator/options.go
deleted file mode 100644
index d25bd8d25..000000000
--- a/cmd/flow-aggregator/options.go
+++ /dev/null
@@ -1,216 +0,0 @@
-// Copyright 2020 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 main
-
-import (
-	"errors"
-	"fmt"
-	"io/ioutil"
-	"net"
-	"time"
-
-	"github.com/spf13/pflag"
-	"gopkg.in/yaml.v2"
-
-	"antrea.io/antrea/pkg/apis"
-	flowaggregatorconfig "antrea.io/antrea/pkg/config/flowaggregator"
-	"antrea.io/antrea/pkg/flowaggregator"
-	"antrea.io/antrea/pkg/util/flowexport"
-)
-
-const (
-	defaultExternalFlowCollectorTransport = "tcp"
-	defaultExternalFlowCollectorPort      = "4739"
-	defaultActiveFlowRecordTimeout        = "60s"
-	defaultInactiveFlowRecordTimeout      = "90s"
-	defaultAggregatorTransportProtocol    = "TLS"
-	defaultFlowAggregatorAddress          = "flow-aggregator.flow-aggregator.svc"
-	defaultRecordFormat                   = "IPFIX"
-	defaultClickHouseDatabase             = "default"
-	defaultClickHouseCommitInterval       = "8s"
-	minClickHouseCommitInterval           = 1 * time.Second
-	defaultClickHouseDatabaseUrl          = "tcp://clickhouse-clickhouse.flow-visibility.svc:9000"
-)
-
-type Options struct {
-	// The path of configuration file.
-	configFile string
-	// The configuration object
-	config *flowaggregatorconfig.FlowAggregatorConfig
-	// Expiration timeout for active flow records in the flow aggregator
-	activeFlowRecordTimeout time.Duration
-	// Expiration timeout for inactive flow records in the flow aggregator
-	inactiveFlowRecordTimeout time.Duration
-	// Transport protocol over which the aggregator collects IPFIX records from all Agents
-	aggregatorTransportProtocol flowaggregator.AggregatorTransportProtocol
-	// DNS name or IP address of flow aggregator for generating TLS certificate
-	flowAggregatorAddress string
-	// includePodLabels indicates whether source and destination Pod labels are included or not
-	includePodLabels bool
-	// IPFIX flow collector address
-	externalFlowCollectorAddr string
-	// IPFIX flow collector transport protocol
-	externalFlowCollectorProto string
-	// Format for record sent to the configured flow collector
-	format string
-	// clickHouseCommitInterval flow records batch commit interval to clickhouse in the flow aggregator
-	clickHouseCommitInterval time.Duration
-}
-
-func newOptions() *Options {
-	return &Options{
-		config: &flowaggregatorconfig.FlowAggregatorConfig{},
-	}
-}
-
-// addFlags adds flags to fs and binds them to options.
-func (o *Options) addFlags(fs *pflag.FlagSet) {
-	fs.StringVar(&o.configFile, "config", o.configFile, "The path to the configuration file")
-}
-
-// complete completes all the required options.
-func (o *Options) complete(args []string) error {
-	if len(o.configFile) > 0 {
-		c, err := o.loadConfigFromFile(o.configFile)
-		if err != nil {
-			return err
-		}
-		o.config = c
-		o.setDefaults()
-	}
-	return nil
-}
-
-// validate validates all the required options.
-func (o *Options) validate(args []string) error {
-	var err error
-	if len(args) != 0 {
-		return errors.New("no positional arguments are supported")
-	}
-	if o.config.FlowCollector.Enable && o.config.FlowCollector.Address == "" {
-		return fmt.Errorf("external flow collector enabled without providing address")
-	}
-	if !o.isFlowCollectorSet() && !o.isClickHouseSet() {
-		return fmt.Errorf("external flow collector or ClickHouse should be configured")
-	}
-
-	// Validate common parameters
-	o.activeFlowRecordTimeout, err = time.ParseDuration(o.config.ActiveFlowRecordTimeout)
-	if err != nil {
-		return err
-	}
-	o.inactiveFlowRecordTimeout, err = time.ParseDuration(o.config.ActiveFlowRecordTimeout)
-	if err != nil {
-		return err
-	}
-	transportProtocol, err := flowexport.ParseTransportProtocol(o.config.AggregatorTransportProtocol)
-	if err != nil {
-		return err
-	}
-	o.aggregatorTransportProtocol = transportProtocol
-	o.flowAggregatorAddress = o.config.FlowAggregatorAddress
-	o.includePodLabels = o.config.RecordContents.PodLabels
-
-	// Validate flow collector specific parameters
-	if o.isFlowCollectorSet() {
-		host, port, proto, err := flowexport.ParseFlowCollectorAddr(
-			o.config.FlowCollector.Address, defaultExternalFlowCollectorPort,
-			defaultExternalFlowCollectorTransport)
-		if err != nil {
-			return err
-		}
-		o.externalFlowCollectorAddr = net.JoinHostPort(host, port)
-		o.externalFlowCollectorProto = proto
-
-		if o.config.FlowCollector.RecordFormat != "IPFIX" &&
-			o.config.FlowCollector.RecordFormat != "JSON" {
-			return fmt.Errorf("record format %s is not supported", o.config.FlowCollector.RecordFormat)
-		} else {
-			o.format = o.config.FlowCollector.RecordFormat
-		}
-	}
-
-	// Validate clickhouse specific parameters
-	if o.isClickHouseSet() {
-		o.clickHouseCommitInterval, err = time.ParseDuration(o.config.ClickHouse.CommitInterval)
-		if err != nil {
-			return err
-		}
-		if o.clickHouseCommitInterval < minClickHouseCommitInterval {
-			return fmt.Errorf("commitInterval %s is too small: shortest supported interval is %s",
-				o.config.ClickHouse.CommitInterval, minClickHouseCommitInterval)
-		}
-	}
-	return nil
-}
-
-func (o *Options) loadConfigFromFile(file string) (*flowaggregatorconfig.FlowAggregatorConfig, error) {
-	data, err := ioutil.ReadFile(file)
-	if err != nil {
-		return nil, err
-	}
-	c := flowaggregatorconfig.FlowAggregatorConfig{}
-	err = yaml.UnmarshalStrict(data, &c)
-	if err != nil {
-		return nil, err
-	}
-	return &c, nil
-}
-
-func (o *Options) isFlowCollectorSet() bool {
-	return o.config.FlowCollector.Enable && len(o.config.FlowCollector.Address) > 0
-}
-
-func (o *Options) isClickHouseSet() bool {
-	return o.config.ClickHouse.Enable
-}
-
-func (o *Options) setDefaults() {
-	if o.config.ActiveFlowRecordTimeout == "" {
-		o.config.ActiveFlowRecordTimeout = defaultActiveFlowRecordTimeout
-	}
-	if o.config.InactiveFlowRecordTimeout == "" {
-		o.config.InactiveFlowRecordTimeout = defaultInactiveFlowRecordTimeout
-	}
-	if o.config.AggregatorTransportProtocol == "" {
-		o.config.AggregatorTransportProtocol = defaultAggregatorTransportProtocol
-	}
-	if o.config.FlowAggregatorAddress == "" {
-		o.config.FlowAggregatorAddress = defaultFlowAggregatorAddress
-	}
-	if o.config.APIServer.APIPort == 0 {
-		o.config.APIServer.APIPort = apis.FlowAggregatorAPIPort
-	}
-	if o.isFlowCollectorSet() {
-		if o.config.FlowCollector.RecordFormat == "" {
-			o.config.FlowCollector.RecordFormat = defaultRecordFormat
-		}
-	}
-	if o.isClickHouseSet() {
-		if o.config.ClickHouse.Database == "" {
-			o.config.ClickHouse.Database = defaultClickHouseDatabase
-		}
-		if o.config.ClickHouse.DatabaseURL == "" {
-			o.config.ClickHouse.DatabaseURL = defaultClickHouseDatabaseUrl
-		}
-		if o.config.ClickHouse.Compress == nil {
-			o.config.ClickHouse.Compress = new(bool)
-			*o.config.ClickHouse.Compress = true
-		}
-		if o.config.ClickHouse.CommitInterval == "" {
-			o.config.ClickHouse.CommitInterval = defaultClickHouseCommitInterval
-		}
-	}
-}
diff --git a/go.mod b/go.mod
index d2d4998e7..7645c9913 100644
--- a/go.mod
+++ b/go.mod
@@ -18,6 +18,7 @@ require (
 	github.com/containernetworking/cni v0.8.1
 	github.com/containernetworking/plugins v0.8.7
 	github.com/coreos/go-iptables v0.6.0
+	github.com/fsnotify/fsnotify v1.5.4
 	github.com/gammazero/deque v0.1.0
 	github.com/go-logr/logr v1.2.0
 	github.com/gogo/protobuf v1.3.2
@@ -55,7 +56,7 @@ require (
 	golang.org/x/mod v0.6.0-dev.0.20220106191415-9b9b3d81d5e3
 	golang.org/x/net v0.0.0-20220127200216-cd36cc0744dd
 	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c
-	golang.org/x/sys v0.0.0-20220209214540-3681064d5158
+	golang.org/x/sys v0.0.0-20220412211240-33da011f77ad
 	golang.org/x/time v0.0.0-20220210224613-90d013bbcef8
 	golang.zx2c4.com/wireguard/wgctrl v0.0.0-20210506160403-92e472f520a5
 	google.golang.org/grpc v1.40.0
@@ -109,7 +110,6 @@ require (
 	github.com/fatih/color v1.10.0 // indirect
 	github.com/felixge/httpsnoop v1.0.1 // indirect
 	github.com/form3tech-oss/jwt-go v3.2.3+incompatible // indirect
-	github.com/fsnotify/fsnotify v1.5.1 // indirect
 	github.com/go-logr/zapr v1.2.0 // indirect
 	github.com/go-openapi/jsonpointer v0.19.5 // indirect
 	github.com/go-openapi/jsonreference v0.19.5 // indirect
diff --git a/go.sum b/go.sum
index 288bb45ec..b5ecc364d 100644
--- a/go.sum
+++ b/go.sum
@@ -284,8 +284,9 @@ github.com/fortytw2/leaktest v1.3.0/go.mod h1:jDsjWgpAGjm2CA7WthBh/CdZYEPF31XHqu
 github.com/frankban/quicktest v1.10.2/go.mod h1:K+q6oSqb0W0Ininfk863uOk1lMy69l/P6txr3mVT54s=
 github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
 github.com/fsnotify/fsnotify v1.4.9/go.mod h1:znqG4EE+3YCdAaPaxE2ZRY/06pZUdp0tY4IgpuI1SZQ=
-github.com/fsnotify/fsnotify v1.5.1 h1:mZcQUHVQUQWoPXXtuf9yuEXKudkV2sx1E06UadKWpgI=
 github.com/fsnotify/fsnotify v1.5.1/go.mod h1:T3375wBYaZdLLcVNkcVbzGHY7f1l/uK5T5Ai1i3InKU=
+github.com/fsnotify/fsnotify v1.5.4 h1:jRbGcIw6P2Meqdwuo0H1p6JVLbL5DHKAKlYndzMwVZI=
+github.com/fsnotify/fsnotify v1.5.4/go.mod h1:OVB6XrOHzAwXMpEM7uPOzcehqUV2UqJxmVXmkdnm1bU=
 github.com/fvbommel/sortorder v1.0.1/go.mod h1:uk88iVf1ovNn1iLfgUVU2F9o5eO30ui720w+kxuqRs0=
 github.com/gammazero/deque v0.1.0 h1:f9LnNmq66VDeuAlSAapemq/U7hJ2jpIWa4c09q8Dlik=
 github.com/gammazero/deque v0.1.0/go.mod h1:KQw7vFau1hHuM8xmI9RbgKFbAsQFWmBpqQ2KenFLk6M=
@@ -1214,8 +1215,9 @@ golang.org/x/sys v0.0.0-20210831042530-f4d43177bf5e/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20211019181941-9d821ace8654/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220114195835-da31bd327af9/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.0.0-20220209214540-3681064d5158 h1:rm+CHSpPEEW2IsXUib1ThaHIjuBVZjxNgSKmBLFfD4c=
 golang.org/x/sys v0.0.0-20220209214540-3681064d5158/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220412211240-33da011f77ad h1:ntjMns5wyP/fN65tdBD4g8J5w8n015+iIIs9rtjXkY0=
+golang.org/x/sys v0.0.0-20220412211240-33da011f77ad/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210220032956-6a3ed077a48d/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
diff --git a/hack/generate-manifest.sh b/hack/generate-manifest.sh
index 96362b0a0..d753b2980 100755
--- a/hack/generate-manifest.sh
+++ b/hack/generate-manifest.sh
@@ -20,18 +20,14 @@ function echoerr {
     >&2 echo "$@"
 }
 
-_usage="Usage: $0 [--mode (dev|release)] [--encap-mode] [--ipsec] [--no-proxy] [--no-np] [--keep] [--tun (geneve|vxlan|gre|stt)] [--verbose-log] [--help|-h]
+_usage="Usage: $0 [--mode (dev|release)] [--encap-mode] [--ipsec] [--tun (geneve|vxlan|gre|stt)] [--verbose-log] [--help|-h]
 Generate a YAML manifest for Antrea using Helm and print it to stdout.
         --mode (dev|release)          Choose the configuration variant that you need (default is 'dev')
         --encap-mode                  Traffic encapsulation mode. (default is 'encap')
         --cloud                       Generate a manifest appropriate for running Antrea in Public Cloud
         --ipsec                       Generate a manifest with IPsec encryption of tunnel traffic enabled
-        --all-features                Generate a manifest with all alpha features enabled
-        --no-proxy                    Generate a manifest with Antrea proxy disabled
+        --feature-gates               A comma-separated list of key=value pairs that describe feature gates, e.g. AntreaProxy=true,Egress=false.
         --proxy-all                   Generate a manifest with Antrea proxy with all Service support enabled
-        --endpointslice               Generate a manifest with EndpointSlice support enabled
-        --flow-exporter               Generate a manifest with FlowExporter support enabled
-        --no-np                       Generate a manifest with Antrea-native policies disabled
         --tun (geneve|vxlan|gre|stt)  Choose encap tunnel type from geneve, gre, stt and vxlan (default is geneve)
         --verbose-log                 Generate a manifest with increased log-level (level 4) for Antrea agent and controller.
                                       This option will work only in 'dev' mode.
@@ -67,15 +63,10 @@ function print_help {
     echoerr "Try '$0 --help' for more information."
 }
 
+FEATURE_GATES=""
 MODE="dev"
 IPSEC=false
-ALLFEATURES=false
-PROXY=true
 PROXY_ALL=false
-ENDPOINTSLICE=false
-FLOW_EXPORTER=false
-NP=true
-KEEP=false
 ENCAP_MODE=""
 CLOUD=""
 TUN_TYPE="geneve"
@@ -118,42 +109,21 @@ case $key in
     IPSEC=true
     shift
     ;;
-    --all-features)
-    ALLFEATURES=true
-    shift
-    ;;
-    --no-proxy)
-    PROXY=false
-    shift
+    --feature-gates)
+    FEATURE_GATES="$2"
+    shift 2
     ;;
     --proxy-all)
     PROXY=true
     PROXY_ALL=true
     shift
     ;;
-    --endpointslice)
-    PROXY=true
-    ENDPOINTSLICE=true
-    shift
-    ;;
-    --flow-exporter)
-    FLOW_EXPORTER=true
-    shift
-    ;;
-    --no-np)
-    NP=false
-    shift
-    ;;
     --k8s-1.15)
     echoerr "The --k8s-1.15 flag is no longer supported"
     exit 1
     K8S_115=true
     shift
     ;;
-    --keep)
-    KEEP=true
-    shift
-    ;;
     --tun)
     TUN_TYPE="$2"
     shift 2
@@ -225,18 +195,6 @@ case $key in
 esac
 done
 
-if [ "$PROXY" == false ] && [ "$ENDPOINTSLICE" == true ]; then
-    echoerr "--endpointslice requires AntreaProxy, so it cannot be used with --no-proxy"
-    print_help
-    exit 1
-fi
-
-if [ "$PROXY" == false ] && [ "$PROXY_ALL" == true ]; then
-    echoerr "--proxy-all requires AntreaProxy, so it cannot be used with --no-proxy"
-    print_help
-    exit 1
-fi
-
 if [ "$MODE" != "dev" ] && [ "$MODE" != "release" ]; then
     echoerr "--mode must be one of 'dev' or 'release'"
     print_help
@@ -278,11 +236,6 @@ if $COVERAGE && $VERBOSE_LOG; then
     VERBOSE_LOG=false
 fi
 
-if [[ "$ENCAP_MODE" != "" ]] && [[ "$ENCAP_MODE" != "encap" ]] && ! $PROXY; then
-    echoerr "Cannot use '--no-proxy' when '--encap-mode' is not 'encap'"
-    exit 1
-fi
-
 if [[ "$ENCAP_MODE" != "" ]] && [[ "$ENCAP_MODE" != "encap" ]] && $IPSEC; then
     echoerr "Encap mode '$ENCAP_MODE' does not make sense with IPsec"
     exit 1
@@ -314,30 +267,15 @@ if $MULTICAST; then
     HELM_VALUES+=("trafficEncapMode=noEncap" "featureGates.Multicast=true" "multicast.multicastInterfaces={$MULTICAST_INTERFACES}")
 fi
 
-if $ALLFEATURES; then
-    HELM_VALUES+=("featureGates.FlowExporter=true" "featureGates.EndpointSlice=true" "antreaProxy.proxyAll=true")
-fi
-
-if ! $PROXY; then
-    HELM_VALUES+=("featureGates.AntreaProxy=false")
-fi
+IFS=',' read -r -a feature_gates <<< "$FEATURE_GATES"
+for feature_gate in "${feature_gates[@]}"; do
+    HELM_VALUES+=("featureGates.${feature_gate}")
+done
 
 if $PROXY_ALL; then
     HELM_VALUES+=("antreaProxy.proxyAll=true")
 fi
 
-if $ENDPOINTSLICE; then
-    HELM_VALUES+=("featureGates.EndpointSlice=true")
-fi
-
-if $FLOW_EXPORTER; then
-    HELM_VALUES+=("featureGates.FlowExporter=true")
-fi
-
-if ! $NP; then
-    HELM_VALUES+=("featureGates.AntreaPolicy=false")
-fi
-
 if [[ $ENCAP_MODE != "" ]]; then
     HELM_VALUES+=("trafficEncapMode=$ENCAP_MODE")
 fi
diff --git a/pkg/agent/nodeportlocal/k8s/annotations.go b/pkg/agent/nodeportlocal/k8s/annotations.go
index c5a69a73d..8601d73b7 100644
--- a/pkg/agent/nodeportlocal/k8s/annotations.go
+++ b/pkg/agent/nodeportlocal/k8s/annotations.go
@@ -26,29 +26,18 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	clientset "k8s.io/client-go/kubernetes"
 	"k8s.io/klog/v2"
-)
 
-const (
-	NPLAnnotationKey          = "nodeportlocal.antrea.io"
-	NPLEnabledAnnotationKey   = "nodeportlocal.antrea.io/enabled"
-	NPLEnabledAnnotationIndex = "nplEnabledAnnotation"
+	npltypes "antrea.io/antrea/pkg/agent/nodeportlocal/types"
 )
 
-// NPLAnnotation is the structure used for setting NodePortLocal annotation on the Pods.
-type NPLAnnotation struct {
-	PodPort   int      `json:"podPort"`
-	NodeIP    string   `json:"nodeIP"`
-	NodePort  int      `json:"nodePort"`
-	Protocol  string   `json:"protocol"`
-	Protocols []string `json:"protocols"` // deprecated, array with a single member which is equal to the Protocol field
-}
+const NPLEnabledAnnotationIndex = "nplEnabledAnnotation"
 
 func toJSON(serialize interface{}) string {
 	jsonMarshalled, _ := json.Marshal(serialize)
 	return string(jsonMarshalled)
 }
 
-func (c *NPLController) updatePodNPLAnnotation(pod *corev1.Pod, annotations []NPLAnnotation) error {
+func (c *NPLController) updatePodNPLAnnotation(pod *corev1.Pod, annotations []npltypes.NPLAnnotation) error {
 	if err := patchPod(annotations, pod, c.kubeClient); err != nil {
 		klog.Warningf("Unable to patch NodePortLocal annotation for Pod %s/%s: %v", pod.Namespace, pod.Name, err)
 	}
@@ -56,13 +45,13 @@ func (c *NPLController) updatePodNPLAnnotation(pod *corev1.Pod, annotations []NP
 	return nil
 }
 
-func patchPod(value []NPLAnnotation, pod *corev1.Pod, kubeClient clientset.Interface) error {
+func patchPod(value []npltypes.NPLAnnotation, pod *corev1.Pod, kubeClient clientset.Interface) error {
 	payloadValue := make(map[string]*string)
 	if len(value) > 0 {
 		valueStr := string(toJSON(value))
-		payloadValue[NPLAnnotationKey] = &valueStr
+		payloadValue[npltypes.NPLAnnotationKey] = &valueStr
 	} else {
-		payloadValue[NPLAnnotationKey] = nil
+		payloadValue[npltypes.NPLAnnotationKey] = nil
 	}
 
 	newPayload := map[string]interface{}{
@@ -82,11 +71,11 @@ func patchPod(value []NPLAnnotation, pod *corev1.Pod, kubeClient clientset.Inter
 
 // compareNPLAnnotationLists returns true if and only if the two lists contain the same set of
 // annotations, irrespective of the order.
-func compareNPLAnnotationLists(annotations1, annotations2 []NPLAnnotation) bool {
+func compareNPLAnnotationLists(annotations1, annotations2 []npltypes.NPLAnnotation) bool {
 	if len(annotations1) != len(annotations2) {
 		return false
 	}
-	nplAnnotationLess := func(a1, a2 *NPLAnnotation) bool {
+	nplAnnotationLess := func(a1, a2 *npltypes.NPLAnnotation) bool {
 		return a1.NodePort < a2.NodePort
 
 	}
diff --git a/pkg/agent/nodeportlocal/k8s/npl_controller.go b/pkg/agent/nodeportlocal/k8s/npl_controller.go
index 362dfa2bf..0bdb187f6 100644
--- a/pkg/agent/nodeportlocal/k8s/npl_controller.go
+++ b/pkg/agent/nodeportlocal/k8s/npl_controller.go
@@ -23,6 +23,7 @@ import (
 
 	"antrea.io/antrea/pkg/agent/nodeportlocal/portcache"
 	"antrea.io/antrea/pkg/agent/nodeportlocal/rules"
+	"antrea.io/antrea/pkg/agent/nodeportlocal/types"
 	"antrea.io/antrea/pkg/agent/nodeportlocal/util"
 	utilsets "antrea.io/antrea/pkg/util/sets"
 
@@ -100,7 +101,7 @@ func NewNPLController(kubeClient clientset.Interface,
 				if !ok {
 					return []string{}, nil
 				}
-				if val, ok := svc.GetAnnotations()[NPLEnabledAnnotationKey]; ok {
+				if val, ok := svc.GetAnnotations()[types.NPLEnabledAnnotationKey]; ok {
 					return []string{val}, nil
 				}
 				return []string{}, nil
@@ -214,8 +215,8 @@ func (c *NPLController) enqueueSvcUpdate(oldObj, newObj interface{}) {
 	newSvc := newObj.(*corev1.Service)
 	oldSvc := oldObj.(*corev1.Service)
 
-	oldSvcAnnotation := oldSvc.Annotations[NPLEnabledAnnotationKey]
-	newSvcAnnotation := newSvc.Annotations[NPLEnabledAnnotationKey]
+	oldSvcAnnotation := oldSvc.Annotations[types.NPLEnabledAnnotationKey]
+	newSvcAnnotation := newSvc.Annotations[types.NPLEnabledAnnotationKey]
 	// Return if both Services do not have the NPL annotation.
 	if oldSvcAnnotation != "true" && newSvcAnnotation != "true" {
 		return
@@ -266,7 +267,7 @@ func (c *NPLController) enqueueSvc(obj interface{}) {
 	}
 
 	// Process Pods corresponding to Service with valid NPL annotation.
-	if svc.Annotations[NPLEnabledAnnotationKey] == "true" {
+	if svc.Annotations[types.NPLEnabledAnnotationKey] == "true" {
 		validateNPLService(svc)
 		for _, podKey := range c.getPodsFromService(svc) {
 			c.queue.Add(podKey)
@@ -299,7 +300,7 @@ func (c *NPLController) getTargetPortsForServicesOfPod(obj interface{}) (sets.St
 	pod := obj.(*corev1.Pod)
 	services, err := c.svcInformer.GetIndexer().ByIndex(NPLEnabledAnnotationIndex, "true")
 	if err != nil {
-		klog.Errorf("Got error while listing Services with annotation %s: %v", NPLEnabledAnnotationKey, err)
+		klog.Errorf("Got error while listing Services with annotation %s: %v", types.NPLEnabledAnnotationKey, err)
 		return targetPortsInt, targetPortsStr
 	}
 
@@ -437,9 +438,9 @@ func (c *NPLController) handleAddUpdatePod(key string, obj interface{}) error {
 	var nodePort int
 	podPorts := make(map[string]struct{})
 	podContainers := pod.Spec.Containers
-	nplAnnotations := []NPLAnnotation{}
+	nplAnnotations := []types.NPLAnnotation{}
 
-	podAnnotation, nplExists := pod.GetAnnotations()[NPLAnnotationKey]
+	podAnnotation, nplExists := pod.GetAnnotations()[types.NPLAnnotationKey]
 	if nplExists {
 		if err := json.Unmarshal([]byte(podAnnotation), &nplAnnotations); err != nil {
 			klog.Warningf("Unable to unmarshal NodePortLocal annotation for Pod %s", key)
@@ -447,8 +448,8 @@ func (c *NPLController) handleAddUpdatePod(key string, obj interface{}) error {
 		}
 	}
 
-	nplAnnotationsRequiredMap := map[string]NPLAnnotation{}
-	nplAnnotationsRequired := []NPLAnnotation{}
+	nplAnnotationsRequiredMap := map[string]types.NPLAnnotation{}
+	nplAnnotationsRequired := []types.NPLAnnotation{}
 
 	hostPorts := make(map[string]int)
 	for _, container := range podContainers {
@@ -476,7 +477,7 @@ func (c *NPLController) handleAddUpdatePod(key string, obj interface{}) error {
 		if err := c.deleteAllPortRulesIfAny(podIP); err != nil {
 			return err
 		}
-		if _, exists := pod.Annotations[NPLAnnotationKey]; exists {
+		if _, exists := pod.Annotations[types.NPLAnnotationKey]; exists {
 			return c.cleanupNPLAnnotationForPod(pod)
 		}
 		return nil
@@ -511,7 +512,7 @@ func (c *NPLController) handleAddUpdatePod(key string, obj interface{}) error {
 		}
 
 		if _, ok := nplAnnotationsRequiredMap[portcache.NodePortProtoFormat(nodePort, protocol)]; !ok {
-			nplAnnotationsRequiredMap[portcache.NodePortProtoFormat(nodePort, protocol)] = NPLAnnotation{
+			nplAnnotationsRequiredMap[portcache.NodePortProtoFormat(nodePort, protocol)] = types.NPLAnnotation{
 				PodPort:   port,
 				NodeIP:    pod.Status.HostIP,
 				NodePort:  nodePort,
@@ -577,11 +578,11 @@ func (c *NPLController) waitForRulesInitialization() {
 		//   rules buffer.
 		pod := podList[i]
 		annotations := pod.GetAnnotations()
-		nplAnnotation, ok := annotations[NPLAnnotationKey]
+		nplAnnotation, ok := annotations[types.NPLAnnotationKey]
 		if !ok {
 			continue
 		}
-		nplData := []NPLAnnotation{}
+		nplData := []types.NPLAnnotation{}
 		if err := json.Unmarshal([]byte(nplAnnotation), &nplData); err != nil {
 			klog.InfoS("Found invalid NodePortLocal annotation for Pod that cannot be parsed, cleaning it up", "pod", klog.KObj(pod))
 			// if there's an error in this NodePortLocal annotation, clean it up
@@ -625,7 +626,7 @@ func (c *NPLController) addRulesForNPLPorts(allNPLPorts []rules.PodNodePort, syn
 
 // cleanupNPLAnnotationForPod removes the NodePortLocal annotation from the Pod's annotations map entirely.
 func (c *NPLController) cleanupNPLAnnotationForPod(pod *corev1.Pod) error {
-	_, ok := pod.Annotations[NPLAnnotationKey]
+	_, ok := pod.Annotations[types.NPLAnnotationKey]
 	if !ok {
 		return nil
 	}
diff --git a/pkg/agent/nodeportlocal/types/types.go b/pkg/agent/nodeportlocal/types/types.go
new file mode 100644
index 000000000..b876242b3
--- /dev/null
+++ b/pkg/agent/nodeportlocal/types/types.go
@@ -0,0 +1,29 @@
+// 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
+
+const (
+	NPLAnnotationKey        = "nodeportlocal.antrea.io"
+	NPLEnabledAnnotationKey = "nodeportlocal.antrea.io/enabled"
+)
+
+// NPLAnnotation is the structure used for setting NodePortLocal annotation on the Pods.
+type NPLAnnotation struct {
+	PodPort   int      `json:"podPort"`
+	NodeIP    string   `json:"nodeIP"`
+	NodePort  int      `json:"nodePort"`
+	Protocol  string   `json:"protocol"`
+	Protocols []string `json:"protocols"` // deprecated, array with a single member which is equal to the Protocol field
+}
diff --git a/pkg/antctl/antctl.go b/pkg/antctl/antctl.go
index cdd076bd6..88038d12e 100644
--- a/pkg/antctl/antctl.go
+++ b/pkg/antctl/antctl.go
@@ -28,6 +28,7 @@ import (
 	"antrea.io/antrea/pkg/antctl/raw/featuregates"
 	"antrea.io/antrea/pkg/antctl/raw/multicluster"
 	"antrea.io/antrea/pkg/antctl/raw/proxy"
+	"antrea.io/antrea/pkg/antctl/raw/set"
 	"antrea.io/antrea/pkg/antctl/raw/supportbundle"
 	"antrea.io/antrea/pkg/antctl/raw/traceflow"
 	"antrea.io/antrea/pkg/antctl/transform/addressgroup"
@@ -620,6 +621,12 @@ $ antctl get podmulticaststats pod -n namespace`,
 			supportController: false,
 			commandGroup:      mc,
 		},
+		{
+			cobraCommand:          set.SetCmd,
+			supportAgent:          false,
+			supportController:     false,
+			supportFlowAggregator: true,
+		},
 	},
 	codec: scheme.Codecs,
 }
diff --git a/pkg/antctl/command_definition.go b/pkg/antctl/command_definition.go
index 3e93aa846..192c5a64b 100644
--- a/pkg/antctl/command_definition.go
+++ b/pkg/antctl/command_definition.go
@@ -188,10 +188,11 @@ type flagInfo struct {
 // write complex client-side tasks. Only the global flags of the antctl framework will
 // be passed to the cobra.Command.
 type rawCommand struct {
-	cobraCommand      *cobra.Command
-	supportAgent      bool
-	supportController bool
-	commandGroup      commandGroup
+	cobraCommand          *cobra.Command
+	supportAgent          bool
+	supportController     bool
+	supportFlowAggregator bool
+	commandGroup          commandGroup
 }
 
 // commandDefinition defines options to create a cobra.Command for an antctl client.
diff --git a/pkg/antctl/command_list.go b/pkg/antctl/command_list.go
index e4f911916..cd5b3e218 100644
--- a/pkg/antctl/command_list.go
+++ b/pkg/antctl/command_list.go
@@ -62,6 +62,7 @@ func (cl *commandList) applyToRootCommand(root *cobra.Command, client AntctlClie
 	for _, cmd := range cl.rawCommands {
 		if (runtime.Mode == runtime.ModeAgent && cmd.supportAgent) ||
 			(runtime.Mode == runtime.ModeController && cmd.supportController) ||
+			(runtime.Mode == runtime.ModeFlowAggregator && cmd.supportFlowAggregator) ||
 			(!runtime.InPod && cmd.commandGroup == mc) {
 			if groupCommand, ok := groupCommands[cmd.commandGroup]; ok {
 				groupCommand.AddCommand(cmd.cobraCommand)
diff --git a/pkg/antctl/raw/set/command.go b/pkg/antctl/raw/set/command.go
new file mode 100644
index 000000000..00efd5a80
--- /dev/null
+++ b/pkg/antctl/raw/set/command.go
@@ -0,0 +1,30 @@
+// 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 set
+
+import (
+	"github.com/spf13/cobra"
+
+	flow_aggregator "antrea.io/antrea/pkg/antctl/raw/set/flow-aggregator"
+)
+
+var SetCmd = &cobra.Command{
+	Use:   "set",
+	Short: "Set configuration parameters",
+}
+
+func init() {
+	SetCmd.AddCommand(flow_aggregator.NewFlowAggregatorSetCommand())
+}
diff --git a/pkg/antctl/raw/set/flow-aggregator/command.go b/pkg/antctl/raw/set/flow-aggregator/command.go
new file mode 100644
index 000000000..5d3bf0a23
--- /dev/null
+++ b/pkg/antctl/raw/set/flow-aggregator/command.go
@@ -0,0 +1,179 @@
+// 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 flow_aggregator
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/spf13/cobra"
+	"gopkg.in/yaml.v2"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+
+	"antrea.io/antrea/pkg/antctl/raw"
+	flowaggregatorconfig "antrea.io/antrea/pkg/config/flowaggregator"
+)
+
+// Command is the support bundle command implementation.
+var Command *cobra.Command
+
+type FlowAggregatorConfigMutator func(c *flowaggregatorconfig.FlowAggregatorConfig, value string) error
+
+var mutators map[string]FlowAggregatorConfigMutator
+
+var example = strings.Trim(`
+  Enable ClickHouse
+  $ antctl set flow-aggregator clickHouse.enable=true
+  Update ClickHouse database
+  $ antctl set flow-aggregator clickHouse.database=name
+  Update ClickHouse databaseURL
+  $ antctl set flow-aggregator clickHouse.databaseURL=http://xxxxx
+  Update ClickHouse debug
+  $ antctl set flow-aggregator clickHouse.debug=true
+  Update ClickHouse compress
+  $ antctl set flow-aggregator clickHouse.compress=true
+  Update ClickHouse commitInterval
+  $ antctl set flow-aggregator clickHouse.commitInterval=10s
+  Update IPFIX Flow Collector address
+  $ antctl set flow-aggregator flowCollector.address=<IP>:<port>[:<proto>]
+  Enable IPFIX Flow Collector
+  $ antctl set flow-aggregator flowCollector.enable=true
+`, "\n")
+
+func NewFlowAggregatorSetCommand() *cobra.Command {
+	Command = &cobra.Command{
+		Use:     "flow-aggregator",
+		Short:   "Update configuration parameters used in Flow Aggregator",
+		Example: example,
+		RunE:    updateRunE,
+	}
+	mutators = map[string]FlowAggregatorConfigMutator{
+		"clickHouse.enable": func(c *flowaggregatorconfig.FlowAggregatorConfig, value string) error {
+			return setBoolOrFail(&c.ClickHouse.Enable, value)
+		},
+		"clickHouse.database": func(c *flowaggregatorconfig.FlowAggregatorConfig, value string) error {
+			return setStringOrFail(&c.ClickHouse.Database, value)
+		},
+		"clickHouse.databaseURL": func(c *flowaggregatorconfig.FlowAggregatorConfig, value string) error {
+			return setStringOrFail(&c.ClickHouse.DatabaseURL, value)
+		},
+		"clickHouse.debug": func(c *flowaggregatorconfig.FlowAggregatorConfig, value string) error {
+			return setBoolOrFail(&c.ClickHouse.Debug, value)
+		},
+		"clickHouse.compress": func(c *flowaggregatorconfig.FlowAggregatorConfig, value string) error {
+			if c.ClickHouse.Compress == nil {
+				c.ClickHouse.Compress = new(bool)
+			}
+			return setBoolOrFail(c.ClickHouse.Compress, value)
+		},
+		"clickHouse.commitInterval": func(c *flowaggregatorconfig.FlowAggregatorConfig, value string) error {
+			return setCommitIntervalOrFail(&c.ClickHouse.CommitInterval, value)
+		},
+		"flowCollector.enable": func(c *flowaggregatorconfig.FlowAggregatorConfig, value string) error {
+			return setBoolOrFail(&c.FlowCollector.Enable, value)
+		},
+		"flowCollector.address": func(c *flowaggregatorconfig.FlowAggregatorConfig, value string) error {
+			return setStringOrFail(&c.FlowCollector.Address, value)
+		},
+	}
+	return Command
+}
+
+func updateRunE(cmd *cobra.Command, args []string) error {
+	kubeconfig, err := raw.ResolveKubeconfig(cmd)
+	if err != nil {
+		return err
+	}
+	k8sClient, _, err := raw.SetupClients(kubeconfig)
+	if err != nil {
+		return fmt.Errorf("failed to create clientset: %w", err)
+	}
+	configMapName := os.Getenv("FA_CONFIG_MAP_NAME")
+	configMap, err := GetFAConfigMap(k8sClient, configMapName)
+	if err != nil {
+		return err
+	}
+	var flowAggregatorConf flowaggregatorconfig.FlowAggregatorConfig
+	if err := yaml.Unmarshal([]byte(configMap.Data["flow-aggregator.conf"]), &flowAggregatorConf); err != nil {
+		return err
+	}
+	for _, query := range args {
+		pair := strings.Split(query, "=")
+		if len(pair) != 2 {
+			return fmt.Errorf("query should contain exactly one '='")
+		}
+		mutatorFn, ok := mutators[pair[0]]
+		if !ok {
+			return fmt.Errorf("unknown configuration parameter, please check antctl set flow-aggregator -h")
+		}
+		if err := mutatorFn(&flowAggregatorConf, pair[1]); err != nil {
+			return err
+		}
+	}
+	// marshal back the changed parameters to configmap
+	b, err := yaml.Marshal(&flowAggregatorConf)
+	if err != nil {
+		return err
+	}
+	configMap.Data["flow-aggregator.conf"] = string(b)
+	if _, err := k8sClient.CoreV1().ConfigMaps(os.Getenv("POD_NAMESPACE")).Update(context.TODO(), configMap, metav1.UpdateOptions{}); err != nil {
+		return err
+	}
+	return nil
+}
+
+func setBoolOrFail(b *bool, value string) error {
+	boolValue, err := strconv.ParseBool(value)
+	if err != nil {
+		return err
+	}
+	*b = boolValue
+	return nil
+}
+
+func setStringOrFail(b *string, value string) error {
+	*b = value
+	return nil
+}
+
+func setCommitIntervalOrFail(b *string, value string) error {
+	commitInterval, err := time.ParseDuration(value)
+	if err != nil {
+		return err
+	}
+	if commitInterval < flowaggregatorconfig.MinClickHouseCommitInterval {
+		return fmt.Errorf("commitInterval %s is too small: shortest supported interval is %s", commitInterval, flowaggregatorconfig.MinClickHouseCommitInterval)
+	}
+	*b = value
+	return nil
+}
+
+// GetFAConfigMap is used to get and return the flow-aggregator configmap
+func GetFAConfigMap(k8sClient kubernetes.Interface, configMapName string) (*corev1.ConfigMap, error) {
+	if len(configMapName) == 0 {
+		return nil, fmt.Errorf("failed to locate %s ConfigMap volume", "flow-aggregator-config")
+	}
+	configMap, err := k8sClient.CoreV1().ConfigMaps("flow-aggregator").Get(context.TODO(), configMapName, metav1.GetOptions{})
+	if err != nil {
+		return nil, fmt.Errorf("failed to get ConfigMap %s: %v", configMapName, err)
+	}
+	return configMap, nil
+}
diff --git a/pkg/config/flowaggregator/config.go b/pkg/config/flowaggregator/config.go
index 139d97ce1..b3054f32b 100644
--- a/pkg/config/flowaggregator/config.go
+++ b/pkg/config/flowaggregator/config.go
@@ -14,8 +14,12 @@
 
 package config
 
-import (
-	"antrea.io/antrea/pkg/flowaggregator"
+type AggregatorTransportProtocol string
+
+const (
+	AggregatorTransportProtocolTCP AggregatorTransportProtocol = "TCP"
+	AggregatorTransportProtocolTLS AggregatorTransportProtocol = "TLS"
+	AggregatorTransportProtocolUDP AggregatorTransportProtocol = "UDP"
 )
 
 type FlowAggregatorConfig struct {
@@ -36,7 +40,7 @@ type FlowAggregatorConfig struct {
 	InactiveFlowRecordTimeout string `yaml:"inactiveFlowRecordTimeout,omitempty"`
 	// Transport protocol over which the aggregator collects IPFIX records from all Agents.
 	// Defaults to "tls"
-	AggregatorTransportProtocol flowaggregator.AggregatorTransportProtocol `yaml:"aggregatorTransportProtocol,omitempty"`
+	AggregatorTransportProtocol AggregatorTransportProtocol `yaml:"aggregatorTransportProtocol,omitempty"`
 	// Provide DNS name or IP address of flow aggregator for generating TLS certificate.
 	// Defaults to "flow-aggregator.flow-aggregator.svc"
 	FlowAggregatorAddress string `yaml:"flowAggregatorAddress,omitempty"`
diff --git a/pkg/config/flowaggregator/default.go b/pkg/config/flowaggregator/default.go
new file mode 100644
index 000000000..6d4586bb4
--- /dev/null
+++ b/pkg/config/flowaggregator/default.go
@@ -0,0 +1,69 @@
+// 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 config
+
+import (
+	"time"
+
+	"antrea.io/antrea/pkg/apis"
+)
+
+const (
+	DefaultExternalFlowCollectorTransport = "tcp"
+	DefaultExternalFlowCollectorPort      = "4739"
+	DefaultActiveFlowRecordTimeout        = "60s"
+	DefaultInactiveFlowRecordTimeout      = "90s"
+	DefaultAggregatorTransportProtocol    = "TLS"
+	DefaultFlowAggregatorAddress          = "flow-aggregator.flow-aggregator.svc"
+	DefaultRecordFormat                   = "IPFIX"
+	DefaultClickHouseDatabase             = "default"
+	DefaultClickHouseCommitInterval       = "8s"
+	MinClickHouseCommitInterval           = 1 * time.Second
+	DefaultClickHouseDatabaseUrl          = "tcp://clickhouse-clickhouse.flow-visibility.svc:9000"
+)
+
+func SetConfigDefaults(flowAggregatorConf *FlowAggregatorConfig) {
+	if flowAggregatorConf.ActiveFlowRecordTimeout == "" {
+		flowAggregatorConf.ActiveFlowRecordTimeout = DefaultActiveFlowRecordTimeout
+	}
+	if flowAggregatorConf.InactiveFlowRecordTimeout == "" {
+		flowAggregatorConf.InactiveFlowRecordTimeout = DefaultInactiveFlowRecordTimeout
+	}
+	if flowAggregatorConf.AggregatorTransportProtocol == "" {
+		flowAggregatorConf.AggregatorTransportProtocol = DefaultAggregatorTransportProtocol
+	}
+	if flowAggregatorConf.FlowAggregatorAddress == "" {
+		flowAggregatorConf.FlowAggregatorAddress = DefaultFlowAggregatorAddress
+	}
+	if flowAggregatorConf.APIServer.APIPort == 0 {
+		flowAggregatorConf.APIServer.APIPort = apis.FlowAggregatorAPIPort
+	}
+	if flowAggregatorConf.FlowCollector.RecordFormat == "" {
+		flowAggregatorConf.FlowCollector.RecordFormat = DefaultRecordFormat
+	}
+	if flowAggregatorConf.ClickHouse.Database == "" {
+		flowAggregatorConf.ClickHouse.Database = DefaultClickHouseDatabase
+	}
+	if flowAggregatorConf.ClickHouse.DatabaseURL == "" {
+		flowAggregatorConf.ClickHouse.DatabaseURL = DefaultClickHouseDatabaseUrl
+	}
+	if flowAggregatorConf.ClickHouse.Compress == nil {
+		flowAggregatorConf.ClickHouse.Compress = new(bool)
+		*flowAggregatorConf.ClickHouse.Compress = true
+	}
+	if flowAggregatorConf.ClickHouse.CommitInterval == "" {
+		flowAggregatorConf.ClickHouse.CommitInterval = DefaultClickHouseCommitInterval
+	}
+}
diff --git a/pkg/flowaggregator/clickhouseclient/clickhouseclient.go b/pkg/flowaggregator/clickhouseclient/clickhouseclient.go
index e9f5e3a2d..c1f75f8bd 100644
--- a/pkg/flowaggregator/clickhouseclient/clickhouseclient.go
+++ b/pkg/flowaggregator/clickhouseclient/clickhouseclient.go
@@ -24,6 +24,7 @@ import (
 	"github.com/ClickHouse/clickhouse-go"
 	"github.com/gammazero/deque"
 	ipfixentities "github.com/vmware/go-ipfix/pkg/entities"
+	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/klog/v2"
 )
 
@@ -94,8 +95,10 @@ type ClickHouseExportProcess struct {
 	queueSize int
 	// commitInterval is the interval between batch commits
 	commitInterval time.Duration
-	// stopChan is the channel to receive stop message
-	stopChan chan bool
+	// stopCh is the channel to receive stop message
+	stopCh chan bool
+	// commitTicker is a ticker, containing a channel used to trigger batchCommitAll() for every commitInterval period
+	commitTicker *time.Ticker
 }
 
 type ClickHouseInput struct {
@@ -121,9 +124,13 @@ func (ci *ClickHouseInput) getDataSourceName() (string, error) {
 	}
 	if ci.Debug {
 		sb.WriteString("&debug=true")
+	} else {
+		sb.WriteString("&debug=false")
 	}
 	if *ci.Compress {
 		sb.WriteString("&compress=true")
+	} else {
+		sb.WriteString("&compress=false")
 	}
 
 	return sb.String(), nil
@@ -180,37 +187,11 @@ type ClickHouseFlowRow struct {
 }
 
 func NewClickHouseClient(input ClickHouseInput) (*ClickHouseExportProcess, error) {
-	dsn, err := input.getDataSourceName()
-	if err != nil {
-		klog.ErrorS(err, "Error parsing ClickHouse DSN")
-		return nil, err
-	}
-	connect, err := sql.Open("clickhouse", dsn)
-	if err != nil {
-		klog.ErrorS(err, "Error when opening DB connection", "dbURL", input.DatabaseURL)
-		return nil, err
-	}
 
-	// Test ping DB
-	if err = connect.Ping(); err != nil {
-		if exception, ok := err.(*clickhouse.Exception); ok {
-			klog.ErrorS(err, "Error pinging DB", "errCode", exception.Code)
-		} else {
-			klog.ErrorS(err, "Error pinging DB")
-		}
-		return nil, err
-	}
-
-	// Test open Transaction
-	tx, err := connect.Begin()
-	if err == nil {
-		_, err = tx.Prepare(insertQuery)
-	}
+	dsn, connect, err := PrepareConnection(input)
 	if err != nil {
-		klog.ErrorS(err, "Error when preparing insert statement")
 		return nil, err
 	}
-	_ = tx.Commit()
 
 	chClient := &ClickHouseExportProcess{
 		db:             connect,
@@ -219,9 +200,9 @@ func NewClickHouseClient(input ClickHouseInput) (*ClickHouseExportProcess, error
 		mutex:          sync.RWMutex{},
 		queueSize:      maxQueueSize,
 		commitInterval: input.CommitInterval,
-		stopChan:       make(chan bool),
+		stopCh:         make(chan bool),
+		commitTicker:   time.NewTicker(input.CommitInterval),
 	}
-
 	return chClient, nil
 }
 
@@ -239,11 +220,11 @@ func (ch *ClickHouseExportProcess) CacheSet(set ipfixentities.Set) {
 
 func (ch *ClickHouseExportProcess) Start() {
 	go ch.flowRecordPeriodicCommit()
-	<-ch.stopChan
+	<-ch.stopCh
 }
 
 func (ch *ClickHouseExportProcess) Stop() {
-	close(ch.stopChan)
+	close(ch.stopCh)
 }
 
 func (ch *ClickHouseExportProcess) getClickHouseFlowRow(record ipfixentities.Record) *ClickHouseFlowRow {
@@ -399,16 +380,23 @@ func (ch *ClickHouseExportProcess) getClickHouseFlowRow(record ipfixentities.Rec
 }
 
 func (ch *ClickHouseExportProcess) flowRecordPeriodicCommit() {
-	commitTicker := time.NewTicker(ch.commitInterval)
 	logTicker := time.NewTicker(time.Minute)
 	committedRec := 0
 	for {
 		select {
-		case <-ch.stopChan:
-			commitTicker.Stop()
+		case <-ch.stopCh:
+			klog.InfoS("Stopping ClickHouse exporting process")
+			committed, err := ch.batchCommitAll()
+			if err != nil {
+				klog.ErrorS(err, "Error when doing last batchCommitAll")
+			} else {
+				committedRec += committed
+				klog.V(4).InfoS("Total number of records committed to DB", "count", committedRec)
+			}
+			ch.commitTicker.Stop()
 			logTicker.Stop()
 			return
-		case <-commitTicker.C:
+		case <-ch.commitTicker.C:
 			committed, err := ch.batchCommitAll()
 			if err == nil {
 				committedRec += committed
@@ -518,3 +506,86 @@ func (ch *ClickHouseExportProcess) batchCommitAll() (int, error) {
 
 	return currSize, nil
 }
+
+func PrepareConnection(input ClickHouseInput) (string, *sql.DB, error) {
+	dsn, err := input.getDataSourceName()
+	if err != nil {
+		return "", nil, fmt.Errorf("error when parsing ClickHouse DSN: %v", err)
+	}
+	connect, err := ConnectClickHouse(dsn)
+	if err != nil {
+		return "", nil, err
+	}
+	// Test open Transaction
+	tx, err := connect.Begin()
+	if err == nil {
+		_, err = tx.Prepare(insertQuery)
+	}
+	if err != nil {
+		return "", nil, fmt.Errorf("error when preparing insert statement, %v", err)
+	}
+	_ = tx.Commit()
+	return dsn, connect, err
+}
+
+func (ch *ClickHouseExportProcess) GetDsnMap() map[string]string {
+	parseURL := strings.Split(ch.dsn, "?")
+	m := make(map[string]string)
+	m["databaseURL"] = parseURL[0]
+	for _, v := range strings.Split(parseURL[1], "&") {
+		pair := strings.Split(v, "=")
+		m[pair[0]] = pair[1]
+	}
+	return m
+}
+
+func (ch *ClickHouseExportProcess) UpdateCH(input *ClickHouseExportProcess, dsn string, connect *sql.DB) {
+	input.mutex.Lock()
+	defer input.mutex.Unlock()
+	input.dsn = dsn
+	input.db = connect
+}
+
+func (ch *ClickHouseExportProcess) GetCommitInterval() time.Duration {
+	return ch.commitInterval
+}
+
+func (ch *ClickHouseExportProcess) SetCommitInterval(commitInterval time.Duration) {
+	ch.commitInterval = commitInterval
+	ch.commitTicker.Reset(ch.commitInterval)
+}
+
+func (ch *ClickHouseExportProcess) GetDsn() string {
+	return ch.dsn
+}
+
+func ConnectClickHouse(url string) (*sql.DB, error) {
+	var connect *sql.DB
+	var connErr error
+	connRetryInterval := 1 * time.Second
+	connTimeout := 10 * time.Second
+
+	// Connect to ClickHouse in a loop
+	if err := wait.PollImmediate(connRetryInterval, connTimeout, func() (bool, error) {
+		// Open the database and ping it
+		var err error
+		connect, err = sql.Open("clickhouse", url)
+		if err != nil {
+			connErr = fmt.Errorf("error when opening DB connection: %v", err)
+			return false, nil
+		}
+		if err := connect.Ping(); err != nil {
+			if exception, ok := err.(*clickhouse.Exception); ok {
+				connErr = fmt.Errorf("failed to ping ClickHouse: %v", exception.Message)
+			} else {
+				connErr = fmt.Errorf("failed to ping ClickHouse: %v", err)
+			}
+			return false, nil
+		} else {
+			return true, nil
+		}
+	}); err != nil {
+		return nil, fmt.Errorf("failed to connect to ClickHouse after %s: %v", connTimeout, connErr)
+	}
+	return connect, nil
+}
diff --git a/pkg/flowaggregator/flowaggregator.go b/pkg/flowaggregator/flowaggregator.go
index 17d58e2fa..c43cb231b 100644
--- a/pkg/flowaggregator/flowaggregator.go
+++ b/pkg/flowaggregator/flowaggregator.go
@@ -18,21 +18,31 @@ import (
 	"bytes"
 	"encoding/json"
 	"fmt"
+	"hash/fnv"
+	"io/ioutil"
+	"os"
+	"path/filepath"
 	"sync"
 	"time"
 
+	"github.com/fsnotify/fsnotify"
+	"github.com/google/uuid"
 	"github.com/vmware/go-ipfix/pkg/collector"
 	ipfixentities "github.com/vmware/go-ipfix/pkg/entities"
 	"github.com/vmware/go-ipfix/pkg/exporter"
 	ipfixintermediate "github.com/vmware/go-ipfix/pkg/intermediate"
 	ipfixregistry "github.com/vmware/go-ipfix/pkg/registry"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/util/wait"
 	coreinformers "k8s.io/client-go/informers/core/v1"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/klog/v2"
 
+	"antrea.io/antrea/pkg/clusteridentity"
+	flowaggregatorconfig "antrea.io/antrea/pkg/config/flowaggregator"
 	"antrea.io/antrea/pkg/flowaggregator/clickhouseclient"
+	"antrea.io/antrea/pkg/flowaggregator/options"
 	"antrea.io/antrea/pkg/flowaggregator/querier"
 	"antrea.io/antrea/pkg/ipfix"
 )
@@ -183,18 +193,25 @@ const (
 	podInfoIndex = "podInfo"
 )
 
-type AggregatorTransportProtocol string
+type updateParam int
 
 const (
-	AggregatorTransportProtocolTCP AggregatorTransportProtocol = "TCP"
-	AggregatorTransportProtocolTLS AggregatorTransportProtocol = "TLS"
-	AggregatorTransportProtocolUDP AggregatorTransportProtocol = "UDP"
+	updateExternalFlowCollectorAddr updateParam = iota
+	updateClickHouseParam
+	enableClickHouse
+	disableClickHouse
+	disableFlowCollector
 )
 
+type updateMsg struct {
+	param updateParam
+	value interface{}
+}
+
 type flowAggregator struct {
 	externalFlowCollectorAddr   string
 	externalFlowCollectorProto  string
-	aggregatorTransportProtocol AggregatorTransportProtocol
+	aggregatorTransportProtocol flowaggregatorconfig.AggregatorTransportProtocol
 	collectingProcess           ipfix.IPFIXCollectingProcess
 	aggregationProcess          ipfix.IPFIXAggregationProcess
 	dbExportProcess             *clickhouseclient.ClickHouseExportProcess
@@ -213,40 +230,141 @@ type flowAggregator struct {
 	sendJSONRecord              bool
 	numRecordsExported          int64
 	numRecordsReceived          int64
+	updateCh                    chan updateMsg
+	configFile                  string
+	configWatcher               *fsnotify.Watcher
+	configData                  []byte
+	APIServer                   flowaggregatorconfig.APIServerConfig
 }
 
 func NewFlowAggregator(
-	externalFlowCollectorAddr string,
-	externalFlowCollectorProto string,
-	activeFlowRecTimeout time.Duration,
-	inactiveFlowRecTimeout time.Duration,
-	aggregatorTransportProtocol AggregatorTransportProtocol,
-	flowAggregatorAddress string,
-	includePodLabels bool,
 	k8sClient kubernetes.Interface,
-	observationDomainID uint32,
 	podInformer coreinformers.PodInformer,
-	sendJSONRecord bool,
-) *flowAggregator {
+	configFile string,
+) (*flowAggregator, error) {
+	if len(configFile) == 0 {
+		return nil, fmt.Errorf("configFile is empty string")
+	}
 	registry := ipfix.NewIPFIXRegistry()
 	registry.LoadRegistry()
+
+	var err error
+	configWatcher, err := fsnotify.NewWatcher()
+	if err != nil {
+		return nil, fmt.Errorf("error when creating file watcher for configuration file: %v", err)
+	}
+	// When watching the configuration file directly, we have to add the file back to our watcher whenever the configuration
+	// file is modified (The watcher cannot track the config file  when the config file is replaced).
+	// Watching the directory can prevent us from above situation.
+	if err = configWatcher.Add(filepath.Dir(configFile)); err != nil {
+		return nil, fmt.Errorf("error when starting file watch on configuration dir: %v", err)
+	}
+
+	data, err := ioutil.ReadFile(configFile)
+	if err != nil {
+		return nil, fmt.Errorf("cannot read FlowAggregator configuration file: %v", err)
+	}
+	opt, err := options.LoadConfig(data)
+	if err != nil {
+		return nil, err
+	}
+
+	var observationDomainID uint32
+	if opt.Config.FlowCollector.ObservationDomainID != nil {
+		observationDomainID = *opt.Config.FlowCollector.ObservationDomainID
+	} else {
+		observationDomainID = genObservationDomainID(k8sClient)
+	}
+	klog.InfoS("Flow aggregator Observation Domain ID", "Domain ID", observationDomainID)
+
+	var sendJSONRecord bool
+	if opt.Config.FlowCollector.RecordFormat == "JSON" {
+		sendJSONRecord = true
+	} else {
+		sendJSONRecord = false
+	}
+
 	fa := &flowAggregator{
-		externalFlowCollectorAddr:   externalFlowCollectorAddr,
-		externalFlowCollectorProto:  externalFlowCollectorProto,
-		aggregatorTransportProtocol: aggregatorTransportProtocol,
-		activeFlowRecordTimeout:     activeFlowRecTimeout,
-		inactiveFlowRecordTimeout:   inactiveFlowRecTimeout,
+		externalFlowCollectorAddr:   opt.ExternalFlowCollectorAddr,
+		externalFlowCollectorProto:  opt.ExternalFlowCollectorProto,
+		aggregatorTransportProtocol: opt.AggregatorTransportProtocol,
+		activeFlowRecordTimeout:     opt.ActiveFlowRecordTimeout,
+		inactiveFlowRecordTimeout:   opt.InactiveFlowRecordTimeout,
 		registry:                    registry,
 		set:                         ipfixentities.NewSet(false),
-		flowAggregatorAddress:       flowAggregatorAddress,
-		includePodLabels:            includePodLabels,
+		flowAggregatorAddress:       opt.Config.FlowAggregatorAddress,
+		includePodLabels:            opt.Config.RecordContents.PodLabels,
 		k8sClient:                   k8sClient,
 		observationDomainID:         observationDomainID,
 		podInformer:                 podInformer,
 		sendJSONRecord:              sendJSONRecord,
+		updateCh:                    make(chan updateMsg),
+		configFile:                  configFile,
+		configWatcher:               configWatcher,
+		configData:                  data,
+		APIServer:                   opt.Config.APIServer,
+	}
+	err = fa.InitCollectingProcess()
+	if err != nil {
+		return nil, fmt.Errorf("error when creating collecting process: %v", err)
+	}
+	err = fa.InitAggregationProcess()
+	if err != nil {
+		return nil, fmt.Errorf("error when creating aggregation process: %v", err)
+	}
+	if opt.Config.ClickHouse.Enable {
+		chInput := clickhouseclient.ClickHouseInput{
+			Username:       os.Getenv("CH_USERNAME"),
+			Password:       os.Getenv("CH_PASSWORD"),
+			Database:       opt.Config.ClickHouse.Database,
+			DatabaseURL:    opt.Config.ClickHouse.DatabaseURL,
+			Debug:          opt.Config.ClickHouse.Debug,
+			Compress:       opt.Config.ClickHouse.Compress,
+			CommitInterval: opt.ClickHouseCommitInterval,
+		}
+		err = fa.InitDBExportProcess(chInput)
+		if err != nil {
+			return nil, fmt.Errorf("error when creating db export process: %v", err)
+		}
 	}
 	podInformer.Informer().AddIndexers(cache.Indexers{podInfoIndex: podInfoIndexFunc})
-	return fa
+	return fa, nil
+}
+
+// genObservationDomainID generates an IPFIX Observation Domain ID when one is not provided by the
+// user through the flow aggregator configuration. It will first try to generate one
+// deterministically based on the cluster UUID (if available, with a timeout of 10s). Otherwise, it
+// will generate a random one. The cluster UUID should be available if Antrea is deployed to the
+// cluster ahead of the flow aggregator, which is the expectation since when deploying flow
+// aggregator as a Pod, networking needs to be configured by the CNI plugin.
+func genObservationDomainID(k8sClient kubernetes.Interface) uint32 {
+	const retryInterval = time.Second
+	const timeout = 10 * time.Second
+	const defaultAntreaNamespace = "kube-system"
+
+	clusterIdentityProvider := clusteridentity.NewClusterIdentityProvider(
+		defaultAntreaNamespace,
+		clusteridentity.DefaultClusterIdentityConfigMapName,
+		k8sClient,
+	)
+	var clusterUUID uuid.UUID
+	if err := wait.PollImmediate(retryInterval, timeout, func() (bool, error) {
+		clusterIdentity, _, err := clusterIdentityProvider.Get()
+		if err != nil {
+			return false, nil
+		}
+		clusterUUID = clusterIdentity.UUID
+		return true, nil
+	}); err != nil {
+		klog.InfoS(
+			"Unable to retrieve cluster UUID; will generate a random observation domain ID", "timeout", timeout, "ConfigMapNameSpace", defaultAntreaNamespace, "ConfigMapName", clusteridentity.DefaultClusterIdentityConfigMapName,
+		)
+		clusterUUID = uuid.New()
+	}
+	h := fnv.New32()
+	h.Write(clusterUUID[:])
+	observationDomainID := h.Sum32()
+	return observationDomainID
 }
 
 func podInfoIndexFunc(obj interface{}) ([]string, error) {
@@ -266,7 +384,7 @@ func podInfoIndexFunc(obj interface{}) ([]string, error) {
 
 func (fa *flowAggregator) InitCollectingProcess() error {
 	var cpInput collector.CollectorInput
-	if fa.aggregatorTransportProtocol == AggregatorTransportProtocolTLS {
+	if fa.aggregatorTransportProtocol == flowaggregatorconfig.AggregatorTransportProtocolTLS {
 		parentCert, privateKey, caCert, err := generateCACertKey()
 		if err != nil {
 			return fmt.Errorf("error when generating CA certificate: %v", err)
@@ -294,7 +412,7 @@ func (fa *flowAggregator) InitCollectingProcess() error {
 			ServerKey:     serverKey,
 			ServerCert:    serverCert,
 		}
-	} else if fa.aggregatorTransportProtocol == AggregatorTransportProtocolTCP {
+	} else if fa.aggregatorTransportProtocol == flowaggregatorconfig.AggregatorTransportProtocolTCP {
 		cpInput = collector.CollectorInput{
 			Address:       collectorAddress,
 			Protocol:      tcpTransport,
@@ -411,12 +529,12 @@ func (fa *flowAggregator) Run(stopCh <-chan struct{}, wg *sync.WaitGroup) {
 		go fa.dbExportProcess.Start()
 		defer fa.dbExportProcess.Stop()
 	}
-	go fa.flowRecordExpiryCheck(stopCh)
-
+	go fa.flowExportLoop(stopCh)
+	go fa.watchConfiguration(stopCh)
 	<-stopCh
 }
 
-func (fa *flowAggregator) flowRecordExpiryCheck(stopCh <-chan struct{}) {
+func (fa *flowAggregator) flowExportLoop(stopCh <-chan struct{}) {
 	expireTimer := time.NewTimer(fa.activeFlowRecordTimeout)
 	logTicker := time.NewTicker(time.Minute)
 	for {
@@ -431,7 +549,7 @@ func (fa *flowAggregator) flowRecordExpiryCheck(stopCh <-chan struct{}) {
 			if fa.externalFlowCollectorAddr != "" && fa.exportingProcess == nil {
 				err := fa.initExportingProcess()
 				if err != nil {
-					klog.Errorf("Error when initializing exporting process: %v, will retry in %s", err, fa.activeFlowRecordTimeout)
+					klog.ErrorS(err, "Error when initializing exporting process", "wait time for retry", fa.activeFlowRecordTimeout)
 					// Initializing exporting process fails, will retry in next cycle.
 					expireTimer.Reset(fa.activeFlowRecordTimeout)
 					continue
@@ -440,7 +558,7 @@ func (fa *flowAggregator) flowRecordExpiryCheck(stopCh <-chan struct{}) {
 			// Pop the flow record item from expire priority queue in the Aggregation
 			// Process and send the flow records.
 			if err := fa.aggregationProcess.ForAllExpiredFlowRecordsDo(fa.sendFlowKeyRecord); err != nil {
-				klog.Errorf("Error when sending expired flow records: %v", err)
+				klog.ErrorS(err, "Error when sending expired flow records")
 				// If there is an error when sending flow records because of intermittent connectivity, we reset the connection
 				// to IPFIX collector and retry in the next export cycle to reinitialize the connection and send flow records.
 				if fa.exportingProcess != nil {
@@ -458,6 +576,70 @@ func (fa *flowAggregator) flowRecordExpiryCheck(stopCh <-chan struct{}) {
 			klog.V(4).InfoS("Total number of records exported", "count", fa.numRecordsExported)
 			klog.V(4).InfoS("Total number of flows stored in Flow Aggregator", "count", fa.aggregationProcess.GetNumFlows())
 			klog.V(4).InfoS("Number of exporters connected with Flow Aggregator", "count", fa.collectingProcess.GetNumConnToCollector())
+		case msg := <-fa.updateCh:
+			switch msg.param {
+			case updateExternalFlowCollectorAddr:
+				//modify addr and proto if changes.
+				newAddr := msg.value.(querier.ExternalFlowCollectorAddr)
+				if newAddr.Address == fa.externalFlowCollectorAddr && newAddr.Protocol == fa.externalFlowCollectorProto {
+					continue
+				}
+				klog.InfoS("Updating flow-collector address")
+				fa.externalFlowCollectorAddr = newAddr.Address
+				fa.externalFlowCollectorProto = newAddr.Protocol
+				klog.InfoS("Config ExternalFlowCollectorAddr is changed", "address", fa.externalFlowCollectorAddr, "protocol", fa.externalFlowCollectorProto)
+				if fa.exportingProcess != nil {
+					fa.exportingProcess.CloseConnToCollector()
+					fa.exportingProcess = nil
+				}
+			case enableClickHouse:
+				klog.InfoS("Enabling ClickHouse")
+				chInput := msg.value.(clickhouseclient.ClickHouseInput)
+				err := fa.InitDBExportProcess(chInput)
+				if err != nil {
+					klog.ErrorS(err, "Error when creating db export process")
+					continue
+				}
+				klog.InfoS("Clickhouse param is", "database", chInput.Database, "databaseURL", chInput.DatabaseURL, "debug", chInput.Debug, "compress", *chInput.Compress, "commitInterval", fa.dbExportProcess.GetCommitInterval().String())
+				go fa.dbExportProcess.Start()
+				defer fa.dbExportProcess.Stop()
+			case updateClickHouseParam:
+				chInput := msg.value.(clickhouseclient.ClickHouseInput)
+				dsn, connect, err := clickhouseclient.PrepareConnection(chInput)
+				if err != nil {
+					klog.ErrorS(err, "Error when checking new connection")
+					continue
+				}
+				if dsn == fa.dbExportProcess.GetDsn() && chInput.CommitInterval.String() == fa.dbExportProcess.GetCommitInterval().String() {
+					continue
+				}
+				klog.InfoS("Updating ClickHouse")
+				if dsn != fa.dbExportProcess.GetDsn() {
+					fa.dbExportProcess.UpdateCH(fa.dbExportProcess, dsn, connect)
+				}
+				if chInput.CommitInterval.String() != fa.dbExportProcess.GetCommitInterval().String() {
+					fa.dbExportProcess.SetCommitInterval(chInput.CommitInterval)
+				}
+				klog.InfoS("New clickhouse param is", "database", chInput.Database, "databaseURL", chInput.DatabaseURL, "debug", chInput.Debug, "compress", *chInput.Compress, "commitInterval", fa.dbExportProcess.GetCommitInterval().String())
+			case disableFlowCollector:
+				if fa.exportingProcess != nil || fa.externalFlowCollectorAddr != "" {
+					klog.InfoS("Disabling Flow-Collector")
+					fa.externalFlowCollectorAddr = ""
+					fa.externalFlowCollectorProto = ""
+					if fa.exportingProcess != nil {
+						fa.exportingProcess.CloseConnToCollector()
+						fa.exportingProcess = nil
+					}
+					klog.Info("Flow-collector disabled ")
+				}
+			case disableClickHouse:
+				if fa.dbExportProcess != nil {
+					klog.InfoS("Disabling Clickhouse")
+					fa.dbExportProcess.Stop()
+					fa.dbExportProcess = nil
+					klog.InfoS("Clickhouse disabled")
+				}
+			}
 		}
 	}
 }
@@ -489,7 +671,7 @@ func (fa *flowAggregator) sendFlowKeyRecord(key ipfixintermediate.FlowKey, recor
 		if err != nil {
 			return err
 		}
-		klog.V(4).Infof("Data set sent successfully: %d Bytes sent", sentBytes)
+		klog.V(4).InfoS("Data set sent successfully", "bytes sent", sentBytes)
 	}
 	if fa.dbExportProcess != nil {
 		fa.dbExportProcess.CacheSet(fa.set)
@@ -724,3 +906,99 @@ func (fa *flowAggregator) createInfoElementForTemplateSet(ieName string, enterpr
 	}
 	return ie, nil
 }
+
+func (fa *flowAggregator) watchConfiguration(stopCh <-chan struct{}) {
+	klog.InfoS("Watching for FlowAggregator configuration file")
+	for {
+		select {
+		case <-stopCh:
+			close(fa.updateCh)
+			return
+		case event, ok := <-fa.configWatcher.Events:
+			klog.InfoS("Event happened", "event", event.String())
+			if !ok {
+				// If configWatcher event channel is closed, we kill the flow-aggregator Pod to restore
+				// the channel.
+				klog.Fatal("ConfigWatcher event channel closed")
+			}
+			if err := fa.handleWatcherEvent(); err != nil {
+				// If the watcher cannot add mounted configuration file or the configuration file is not readable,
+				// we kill the flow-aggregator Pod (serious error)
+				klog.Fatalf("Cannot watch or read configMap: %v", err)
+			}
+		case err := <-fa.configWatcher.Errors:
+			if err != nil {
+				// If the error happens to watcher, we kill the flow-aggregator Pod.
+				// watcher might be shut-down or broken in this situation.
+				klog.Fatalf("configWatcher err: %v", err)
+			}
+		}
+	}
+}
+
+func (fa *flowAggregator) handleWatcherEvent() error {
+	data, err := ioutil.ReadFile(fa.configFile)
+	if err != nil {
+		return fmt.Errorf("cannot read FlowAggregator configuration file: %v", err)
+	}
+	opt, err := options.LoadConfig(data)
+	if err != nil {
+		klog.ErrorS(err, "Error when loading configuration from config file")
+		return nil
+	}
+	if bytes.Equal(data, fa.configData) {
+		klog.InfoS("Flow-aggregator configuration didn't changed")
+		return nil
+	}
+	fa.configData = data
+	fa.updateFlowAggregator(opt)
+	return nil
+}
+
+func (fa *flowAggregator) updateFlowAggregator(opt *options.Options) {
+	klog.InfoS("Updating Flow Aggregator")
+	if opt.Config.FlowCollector.Enable {
+		query := querier.ExternalFlowCollectorAddr{
+			Address:  opt.ExternalFlowCollectorAddr,
+			Protocol: opt.ExternalFlowCollectorProto,
+		}
+		fa.updateCh <- updateMsg{
+			param: updateExternalFlowCollectorAddr,
+			value: query,
+		}
+	} else {
+		if fa.exportingProcess != nil || fa.externalFlowCollectorAddr != "" {
+			fa.updateCh <- updateMsg{
+				param: disableFlowCollector,
+			}
+		}
+	}
+	if opt.Config.ClickHouse.Enable {
+		chInput := clickhouseclient.ClickHouseInput{
+			Username:       os.Getenv("CH_USERNAME"),
+			Password:       os.Getenv("CH_PASSWORD"),
+			Database:       opt.Config.ClickHouse.Database,
+			DatabaseURL:    opt.Config.ClickHouse.DatabaseURL,
+			Debug:          opt.Config.ClickHouse.Debug,
+			Compress:       opt.Config.ClickHouse.Compress,
+			CommitInterval: opt.ClickHouseCommitInterval,
+		}
+		if fa.dbExportProcess == nil {
+			fa.updateCh <- updateMsg{
+				param: enableClickHouse,
+				value: chInput,
+			}
+		} else {
+			fa.updateCh <- updateMsg{
+				param: updateClickHouseParam,
+				value: chInput,
+			}
+		}
+	} else {
+		if fa.dbExportProcess != nil {
+			fa.updateCh <- updateMsg{
+				param: disableClickHouse,
+			}
+		}
+	}
+}
diff --git a/pkg/flowaggregator/options/options.go b/pkg/flowaggregator/options/options.go
new file mode 100644
index 000000000..10f294c46
--- /dev/null
+++ b/pkg/flowaggregator/options/options.go
@@ -0,0 +1,99 @@
+// 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 options
+
+import (
+	"fmt"
+	"net"
+	"time"
+
+	"gopkg.in/yaml.v2"
+
+	flowaggregatorconfig "antrea.io/antrea/pkg/config/flowaggregator"
+	"antrea.io/antrea/pkg/util/flowexport"
+)
+
+type Options struct {
+	// The configuration object
+	Config *flowaggregatorconfig.FlowAggregatorConfig
+	// Expiration timeout for active flow records in the flow aggregator
+	ActiveFlowRecordTimeout time.Duration
+	// Expiration timeout for inactive flow records in the flow aggregator
+	InactiveFlowRecordTimeout time.Duration
+	// Transport protocol over which the aggregator collects IPFIX records from all Agents
+	AggregatorTransportProtocol flowaggregatorconfig.AggregatorTransportProtocol
+	// IPFIX flow collector address
+	ExternalFlowCollectorAddr string
+	// IPFIX flow collector transport protocol
+	ExternalFlowCollectorProto string
+	// clickHouseCommitInterval flow records batch commit interval to clickhouse in the flow aggregator
+	ClickHouseCommitInterval time.Duration
+}
+
+func LoadConfig(configBytes []byte) (*Options, error) {
+	var opt Options
+	if err := yaml.UnmarshalStrict(configBytes, &opt.Config); err != nil {
+		return nil, fmt.Errorf("failed to unmarshal FlowAggregator config from ConfigMap: %v", err)
+	}
+	flowaggregatorconfig.SetConfigDefaults(opt.Config)
+	// validate all the required options.
+	if opt.Config.FlowCollector.Enable && opt.Config.FlowCollector.Address == "" {
+		return nil, fmt.Errorf("external flow collector enabled without providing address")
+	}
+	if !opt.Config.FlowCollector.Enable && !opt.Config.ClickHouse.Enable {
+		return nil, fmt.Errorf("external flow collector or ClickHouse should be configured")
+	}
+	// Validate common parameters
+	var err error
+	opt.ActiveFlowRecordTimeout, err = time.ParseDuration(opt.Config.ActiveFlowRecordTimeout)
+	if err != nil {
+		return nil, err
+	}
+	opt.InactiveFlowRecordTimeout, err = time.ParseDuration(opt.Config.InactiveFlowRecordTimeout)
+	if err != nil {
+		return nil, err
+	}
+	opt.AggregatorTransportProtocol, err = flowexport.ParseTransportProtocol(opt.Config.AggregatorTransportProtocol)
+	if err != nil {
+		return nil, err
+	}
+	// Validate flow collector specific parameters
+	if opt.Config.FlowCollector.Enable && len(opt.Config.FlowCollector.Address) > 0 {
+		host, port, proto, err := flowexport.ParseFlowCollectorAddr(
+			opt.Config.FlowCollector.Address, flowaggregatorconfig.DefaultExternalFlowCollectorPort,
+			flowaggregatorconfig.DefaultExternalFlowCollectorTransport)
+		if err != nil {
+			return nil, err
+		}
+		opt.ExternalFlowCollectorAddr = net.JoinHostPort(host, port)
+		opt.ExternalFlowCollectorProto = proto
+
+		if opt.Config.FlowCollector.RecordFormat != "IPFIX" && opt.Config.FlowCollector.RecordFormat != "JSON" {
+			return nil, fmt.Errorf("record format %s is not supported", opt.Config.FlowCollector.RecordFormat)
+		}
+	}
+	// Validate clickhouse specific parameters
+	if opt.Config.ClickHouse.Enable {
+		opt.ClickHouseCommitInterval, err = time.ParseDuration(opt.Config.ClickHouse.CommitInterval)
+		if err != nil {
+			return nil, err
+		}
+		if opt.ClickHouseCommitInterval < flowaggregatorconfig.MinClickHouseCommitInterval {
+			return nil, fmt.Errorf("commitInterval %s is too small: shortest supported interval is %s",
+				opt.Config.ClickHouse.CommitInterval, flowaggregatorconfig.MinClickHouseCommitInterval)
+		}
+	}
+	return &opt, nil
+}
diff --git a/pkg/flowaggregator/querier/querier.go b/pkg/flowaggregator/querier/querier.go
index 456b1dc36..de694375a 100644
--- a/pkg/flowaggregator/querier/querier.go
+++ b/pkg/flowaggregator/querier/querier.go
@@ -29,3 +29,8 @@ type FlowAggregatorQuerier interface {
 	GetFlowRecords(flowKey *ipfixintermediate.FlowKey) []map[string]interface{}
 	GetRecordMetrics() Metrics
 }
+
+type ExternalFlowCollectorAddr struct {
+	Address  string
+	Protocol string
+}
diff --git a/pkg/util/flowexport/flowexport.go b/pkg/util/flowexport/flowexport.go
index 3e204ad94..caa1f96c9 100644
--- a/pkg/util/flowexport/flowexport.go
+++ b/pkg/util/flowexport/flowexport.go
@@ -20,7 +20,7 @@ import (
 	"strings"
 	"time"
 
-	"antrea.io/antrea/pkg/flowaggregator"
+	flowaggregatorconfig "antrea.io/antrea/pkg/config/flowaggregator"
 )
 
 // ParseFlowCollectorAddr parses the flow collector address input for flow exporter and aggregator
@@ -76,10 +76,10 @@ func ParseFlowIntervalString(intervalString string) (time.Duration, error) {
 }
 
 // ParseTransportProtocol parses the transport protocol input for the flow aggregator
-func ParseTransportProtocol(trasnportProtocolInput flowaggregator.AggregatorTransportProtocol) (flowaggregator.AggregatorTransportProtocol, error) {
-	upperProtocolInput := flowaggregator.AggregatorTransportProtocol(strings.ToUpper(string(trasnportProtocolInput)))
-	if (upperProtocolInput != flowaggregator.AggregatorTransportProtocolTLS) && (upperProtocolInput != flowaggregator.AggregatorTransportProtocolUDP) && (upperProtocolInput != flowaggregator.AggregatorTransportProtocolTCP) {
-		return "", fmt.Errorf("collecting process over %s proto is not supported", trasnportProtocolInput)
+func ParseTransportProtocol(transportProtocolInput flowaggregatorconfig.AggregatorTransportProtocol) (flowaggregatorconfig.AggregatorTransportProtocol, error) {
+	upperProtocolInput := flowaggregatorconfig.AggregatorTransportProtocol(strings.ToUpper(string(transportProtocolInput)))
+	if (upperProtocolInput != flowaggregatorconfig.AggregatorTransportProtocolTLS) && (upperProtocolInput != flowaggregatorconfig.AggregatorTransportProtocolUDP) && (upperProtocolInput != flowaggregatorconfig.AggregatorTransportProtocolTCP) {
+		return "", fmt.Errorf("collecting process over %s proto is not supported", transportProtocolInput)
 	}
 	return upperProtocolInput, nil
 }
diff --git a/plugins/octant/go.mod b/plugins/octant/go.mod
index eacd5a2d9..3cfcfe0f9 100644
--- a/plugins/octant/go.mod
+++ b/plugins/octant/go.mod
@@ -33,7 +33,7 @@ require (
 	github.com/dop251/goja v0.0.0-20200629185240-bfd59704b500 // indirect
 	github.com/dop251/goja_nodejs v0.0.0-20200706082813-b2775b86b9e0 // indirect
 	github.com/fatih/color v1.10.0 // indirect
-	github.com/fsnotify/fsnotify v1.5.1 // indirect
+	github.com/fsnotify/fsnotify v1.5.4 // indirect
 	github.com/ghodss/yaml v1.0.0 // indirect
 	github.com/go-logr/logr v1.2.0 // indirect
 	github.com/go-sourcemap/sourcemap v2.1.3+incompatible // indirect
@@ -106,7 +106,7 @@ require (
 	golang.org/x/net v0.0.0-20220127200216-cd36cc0744dd // indirect
 	golang.org/x/oauth2 v0.0.0-20211104180415-d3ed0bb246c8 // indirect
 	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c // indirect
-	golang.org/x/sys v0.0.0-20220209214540-3681064d5158 // indirect
+	golang.org/x/sys v0.0.0-20220412211240-33da011f77ad // indirect
 	golang.org/x/term v0.0.0-20210927222741-03fcf44c2211 // indirect
 	golang.org/x/text v0.3.7 // indirect
 	golang.org/x/time v0.0.0-20220210224613-90d013bbcef8 // indirect
diff --git a/plugins/octant/go.sum b/plugins/octant/go.sum
index 7bb425daa..8e261505b 100644
--- a/plugins/octant/go.sum
+++ b/plugins/octant/go.sum
@@ -388,8 +388,9 @@ github.com/frankban/quicktest v1.10.2/go.mod h1:K+q6oSqb0W0Ininfk863uOk1lMy69l/P
 github.com/frankban/quicktest v1.11.3/go.mod h1:wRf/ReqHper53s+kmmSZizM8NamnL3IM0I9ntUbOk+k=
 github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
 github.com/fsnotify/fsnotify v1.4.9/go.mod h1:znqG4EE+3YCdAaPaxE2ZRY/06pZUdp0tY4IgpuI1SZQ=
-github.com/fsnotify/fsnotify v1.5.1 h1:mZcQUHVQUQWoPXXtuf9yuEXKudkV2sx1E06UadKWpgI=
 github.com/fsnotify/fsnotify v1.5.1/go.mod h1:T3375wBYaZdLLcVNkcVbzGHY7f1l/uK5T5Ai1i3InKU=
+github.com/fsnotify/fsnotify v1.5.4 h1:jRbGcIw6P2Meqdwuo0H1p6JVLbL5DHKAKlYndzMwVZI=
+github.com/fsnotify/fsnotify v1.5.4/go.mod h1:OVB6XrOHzAwXMpEM7uPOzcehqUV2UqJxmVXmkdnm1bU=
 github.com/fullsailor/pkcs7 v0.0.0-20190404230743-d7302db945fa/go.mod h1:KnogPXtdwXqoenmZCw6S+25EAm2MkxbG0deNDu4cbSA=
 github.com/fvbommel/sortorder v1.0.1/go.mod h1:uk88iVf1ovNn1iLfgUVU2F9o5eO30ui720w+kxuqRs0=
 github.com/gammazero/deque v0.1.0/go.mod h1:KQw7vFau1hHuM8xmI9RbgKFbAsQFWmBpqQ2KenFLk6M=
@@ -1432,8 +1433,9 @@ golang.org/x/sys v0.0.0-20210831042530-f4d43177bf5e/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20211019181941-9d821ace8654/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220114195835-da31bd327af9/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.0.0-20220209214540-3681064d5158 h1:rm+CHSpPEEW2IsXUib1ThaHIjuBVZjxNgSKmBLFfD4c=
 golang.org/x/sys v0.0.0-20220209214540-3681064d5158/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220412211240-33da011f77ad h1:ntjMns5wyP/fN65tdBD4g8J5w8n015+iIIs9rtjXkY0=
+golang.org/x/sys v0.0.0-20220412211240-33da011f77ad/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210220032956-6a3ed077a48d/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=


Test patch:
diff --git a/ci/jenkins/test-vmc.sh b/ci/jenkins/test-vmc.sh
index 768f14f5d..58d91a6be 100755
--- a/ci/jenkins/test-vmc.sh
+++ b/ci/jenkins/test-vmc.sh
@@ -37,8 +37,7 @@ SECRET_EXIST=false
 TEST_FAILURE=false
 CLUSTER_READY=false
 DOCKER_REGISTRY=""
-# TODO: change to "control-plane" when testbeds are updated to K8s v1.20
-CONTROL_PLANE_NODE_ROLE="master"
+CONTROL_PLANE_NODE_ROLE="master|control-plane"
 
 _usage="Usage: $0 [--cluster-name <VMCClusterNameToUse>] [--kubeconfig <KubeconfigSavePath>] [--workdir <HomePath>]
                   [--log-mode <SonobuoyResultLogLevel>] [--testcase <e2e|conformance|all-features-conformance|whole-conformance|networkpolicy>]
@@ -574,10 +573,10 @@ function run_conformance {
 
     if [[ "$TESTCASE" == "all-features-conformance" ]]; then
       if [[ "$COVERAGE" == true ]]; then
-        $GIT_CHECKOUT_DIR/hack/generate-manifest.sh --mode dev --all-features --coverage > $GIT_CHECKOUT_DIR/build/yamls/antrea-all-coverage.yml
+        $GIT_CHECKOUT_DIR/hack/generate-manifest.sh --mode dev --feature-gates AllAlpha=true,AllBeta=true --proxy-all --coverage > $GIT_CHECKOUT_DIR/build/yamls/antrea-all-coverage.yml
         antrea_yml="antrea-all-coverage.yml"
       else
-        $GIT_CHECKOUT_DIR/hack/generate-manifest.sh --mode dev --all-features --verbose-log > $GIT_CHECKOUT_DIR/build/yamls/antrea-all.yml
+        $GIT_CHECKOUT_DIR/hack/generate-manifest.sh --mode dev --feature-gates AllAlpha=true,AllBeta=true --proxy-all --verbose-log > $GIT_CHECKOUT_DIR/build/yamls/antrea-all.yml
         antrea_yml="antrea-all.yml"
       fi
     fi
diff --git a/ci/jenkins/test.sh b/ci/jenkins/test.sh
index 057eb4e1a..4efb8f988 100755
--- a/ci/jenkins/test.sh
+++ b/ci/jenkins/test.sh
@@ -40,7 +40,6 @@ WINDOWS_NETWORKPOLICY_SKIP="SCTP"
 CONFORMANCE_SKIP="\[Slow\]|\[Serial\]|\[Disruptive\]|\[Flaky\]|\[Feature:.+\]|\[sig-cli\]|\[sig-storage\]|\[sig-auth\]|\[sig-api-machinery\]|\[sig-apps\]|\[sig-node\]"
 NETWORKPOLICY_SKIP="should allow egress access to server in CIDR block|should enforce except clause while egress access to server in CIDR block"
 
-# TODO: change to "control-plane" when testbeds are updated to K8s v1.20
 CONTROL_PLANE_NODE_ROLE="master|control-plane"
 
 CLEAN_STALE_IMAGES="docker system prune --force --all --filter until=48h"
diff --git a/ci/kind/test-e2e-kind.sh b/ci/kind/test-e2e-kind.sh
index f4ad6e217..a849969a8 100755
--- a/ci/kind/test-e2e-kind.sh
+++ b/ci/kind/test-e2e-kind.sh
@@ -22,13 +22,12 @@ function echoerr {
     >&2 echo "$@"
 }
 
-_usage="Usage: $0 [--encap-mode <mode>] [--ip-family <v4|v6>] [--no-proxy] [--np] [--coverage] [--help|-h]
+_usage="Usage: $0 [--encap-mode <mode>] [--ip-family <v4|v6>] [--coverage] [--help|-h]
         --encap-mode                  Traffic encapsulation mode. (default is 'encap').
         --ip-family                   Configures the ipFamily for the KinD cluster.
-        --no-proxy                    Disables Antrea proxy.
+        --feature-gates               A comma-separated list of key=value pairs that describe feature gates, e.g. AntreaProxy=true,Egress=false.
+        --run                         Run only tests matching the regexp.
         --proxy-all                   Enables Antrea proxy with all Service support.
-        --endpointslice               Enables Antrea proxy and EndpointSlice support.
-        --no-np                       Disables Antrea-native policies.
         --flow-visibility             Only run flow visibility related e2e tests.
         --skip                        A comma-separated list of keywords, with which tests should be skipped.
         --coverage                    Enables measure Antrea code coverage when run e2e tests on kind.
@@ -61,24 +60,27 @@ function quit {
 
 mode=""
 ipfamily="v4"
-proxy=true
+feature_gates=""
 proxy_all=false
-endpointslice=false
-np=true
 flow_visibility=false
 coverage=false
 skiplist=""
 setup_only=false
 cleanup_only=false
 test_only=false
+run=""
 while [[ $# -gt 0 ]]
 do
 key="$1"
 
 case $key in
-    --no-proxy)
-    proxy=false
-    shift
+    --run)
+    run="$2"
+    shift 2
+    ;;
+    --feature-gates)
+    feature_gates="$2"
+    shift 2
     ;;
     --proxy-all)
     proxy_all=true
@@ -88,14 +90,6 @@ case $key in
     ipfamily="$2"
     shift 2
     ;;
-    --endpointslice)
-    endpointslice=true
-    shift
-    ;;
-    --no-np)
-    np=false
-    shift
-    ;;
     --flow-visibility)
     flow_visibility=true
     shift
@@ -142,25 +136,15 @@ fi
 
 trap "quit" INT EXIT
 
-manifest_args=""
-if ! $proxy; then
-    manifest_args="$manifest_args --no-proxy"
+manifest_args="$manifest_args --verbose-log"
+if [ -n "$feature_gates" ]; then
+  manifest_args="$manifest_args --feature-gates $feature_gates"
 fi
 if $proxy_all; then
-    if ! $proxy; then
-      echoerr "--proxy-all requires AntreaProxy, so it cannot be used with --no-proxy"
-      exit 1
-    fi
     manifest_args="$manifest_args --proxy-all"
 fi
-if $endpointslice; then
-    manifest_args="$manifest_args --endpointslice"
-fi
-if ! $np; then
-    manifest_args="$manifest_args --no-np"
-fi
 if $flow_visibility; then
-    manifest_args="$manifest_args --flow-exporter --extra-helm-values-file $FLOW_VISIBILITY_HELM_VALUES"
+    manifest_args="$manifest_args --feature-gates FlowExporter=true --extra-helm-values-file $FLOW_VISIBILITY_HELM_VALUES"
 fi
 
 COMMON_IMAGES_LIST=("k8s.gcr.io/e2e-test-images/agnhost:2.29" \
@@ -254,7 +238,11 @@ function run_test {
   fi
   sleep 1
 
-  go test -v -timeout=$timeout antrea.io/antrea/test/e2e $flow_visibility_args -provider=kind --logs-export-dir=$ANTREA_LOG_DIR --skip=$skiplist $coverage_args
+  RUN_OPT=""
+  if [ -n "$run" ]; then
+    RUN_OPT="-run $run"
+  fi
+  go test -v -timeout=$timeout $RUN_OPT antrea.io/antrea/test/e2e $flow_visibility_args -provider=kind --logs-export-dir=$ANTREA_LOG_DIR --skip=$skiplist $coverage_args
 }
 
 if [[ "$mode" == "" ]] || [[ "$mode" == "encap" ]]; then
diff --git a/multicluster/test/e2e/framework.go b/multicluster/test/e2e/framework.go
index bdad74359..4cbfac8d9 100644
--- a/multicluster/test/e2e/framework.go
+++ b/multicluster/test/e2e/framework.go
@@ -155,7 +155,12 @@ func (data *MCTestData) deleteTestNamespaces(timeout time.Duration) error {
 func (data *MCTestData) createPod(clusterName, name, nodeName, namespace, ctrName, image string, command []string,
 	args []string, env []corev1.EnvVar, ports []corev1.ContainerPort, hostNetwork bool, mutateFunc func(pod *corev1.Pod)) error {
 	if d, ok := data.clusterTestDataMap[clusterName]; ok {
-		return d.CreatePodOnNodeInNamespace(name, namespace, nodeName, ctrName, image, command, args, env, ports, hostNetwork, mutateFunc)
+		return antreae2e.NewPodBuilder(name, namespace, image).
+			OnNode(nodeName).WithContainerName(ctrName).
+			WithCommand(command).WithArgs(args).
+			WithEnv(env).WithPorts(ports).WithHostNetwork(hostNetwork).
+			WithMutateFunc(mutateFunc).
+			Create(&d)
 	}
 	return fmt.Errorf("clusterName %s not found", clusterName)
 }
diff --git a/pkg/agent/nodeportlocal/k8s/annotations_test.go b/pkg/agent/nodeportlocal/k8s/annotations_test.go
index d05ecc9b5..83608f5e8 100644
--- a/pkg/agent/nodeportlocal/k8s/annotations_test.go
+++ b/pkg/agent/nodeportlocal/k8s/annotations_test.go
@@ -21,58 +21,60 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+
+	"antrea.io/antrea/pkg/agent/nodeportlocal/types"
 )
 
 func BenchmarkCompareNPLAnnotationLists(b *testing.B) {
 	const nodeIP = "127.0.0.1"
-	a1 := NPLAnnotation{
+	a1 := types.NPLAnnotation{
 		PodPort:  80,
 		NodeIP:   nodeIP,
 		NodePort: 61000,
 	}
-	a2 := NPLAnnotation{
+	a2 := types.NPLAnnotation{
 		PodPort:  8080,
 		NodeIP:   nodeIP,
 		NodePort: 61001,
 	}
-	a3 := NPLAnnotation{
+	a3 := types.NPLAnnotation{
 		PodPort:  81,
 		NodeIP:   nodeIP,
 		NodePort: 61002,
 	}
-	a4 := NPLAnnotation{
+	a4 := types.NPLAnnotation{
 		PodPort:  8081,
 		NodeIP:   nodeIP,
 		NodePort: 61003,
 	}
 	benchmarkCases := []struct {
 		name         string
-		annotations1 []NPLAnnotation
-		annotations2 []NPLAnnotation
+		annotations1 []types.NPLAnnotation
+		annotations2 []types.NPLAnnotation
 		equal        bool
 	}{
 		{
 			"EqualSameOrder",
-			[]NPLAnnotation{a2, a1, a3},
-			[]NPLAnnotation{a2, a1, a3},
+			[]types.NPLAnnotation{a2, a1, a3},
+			[]types.NPLAnnotation{a2, a1, a3},
 			true,
 		},
 		{
 			"EqualDifferentOrder",
-			[]NPLAnnotation{a1, a2, a3},
-			[]NPLAnnotation{a3, a2, a1},
+			[]types.NPLAnnotation{a1, a2, a3},
+			[]types.NPLAnnotation{a3, a2, a1},
 			true,
 		},
 		{
 			"NotEqualSameLength",
-			[]NPLAnnotation{a1, a2, a3},
-			[]NPLAnnotation{a3, a2, a4},
+			[]types.NPLAnnotation{a1, a2, a3},
+			[]types.NPLAnnotation{a3, a2, a4},
 			false,
 		},
 		{
 			"NotEqualDifferentLength",
-			[]NPLAnnotation{a1, a2, a3},
-			[]NPLAnnotation{a1, a2, a3, a4},
+			[]types.NPLAnnotation{a1, a2, a3},
+			[]types.NPLAnnotation{a1, a2, a3, a4},
 			false,
 		},
 	}
diff --git a/pkg/agent/nodeportlocal/npl_agent_test.go b/pkg/agent/nodeportlocal/npl_agent_test.go
index 3cc5957e4..95a09547d 100644
--- a/pkg/agent/nodeportlocal/npl_agent_test.go
+++ b/pkg/agent/nodeportlocal/npl_agent_test.go
@@ -40,12 +40,13 @@ import (
 	k8sfake "k8s.io/client-go/kubernetes/fake"
 	"k8s.io/client-go/tools/cache"
 
-	nplk8s "antrea.io/antrea/pkg/agent/nodeportlocal/k8s"
+	"antrea.io/antrea/pkg/agent/nodeportlocal/k8s"
 	"antrea.io/antrea/pkg/agent/nodeportlocal/portcache"
 	portcachetesting "antrea.io/antrea/pkg/agent/nodeportlocal/portcache/testing"
 	"antrea.io/antrea/pkg/agent/nodeportlocal/rules"
 	rulestesting "antrea.io/antrea/pkg/agent/nodeportlocal/rules/testing"
 	npltesting "antrea.io/antrea/pkg/agent/nodeportlocal/testing"
+	"antrea.io/antrea/pkg/agent/nodeportlocal/types"
 )
 
 const (
@@ -138,7 +139,7 @@ func getTestSvc(targetPorts ...int32) *corev1.Service {
 		ObjectMeta: metav1.ObjectMeta{
 			Name:        defaultSvcName,
 			Namespace:   defaultNS,
-			Annotations: map[string]string{nplk8s.NPLEnabledAnnotationKey: "true"},
+			Annotations: map[string]string{types.NPLEnabledAnnotationKey: "true"},
 		},
 		Spec: corev1.ServiceSpec{
 			Type:     corev1.ServiceTypeClusterIP,
@@ -153,7 +154,7 @@ func getTestSvcWithPortName(portName string) *corev1.Service {
 		ObjectMeta: metav1.ObjectMeta{
 			Name:        defaultSvcName,
 			Namespace:   defaultNS,
-			Annotations: map[string]string{nplk8s.NPLEnabledAnnotationKey: "true"},
+			Annotations: map[string]string{types.NPLEnabledAnnotationKey: "true"},
 		},
 		Spec: corev1.ServiceSpec{
 			Type:     corev1.ServiceTypeClusterIP,
@@ -179,7 +180,7 @@ type testData struct {
 	wg        sync.WaitGroup
 }
 
-func (t *testData) runWrapper(c *nplk8s.NPLController) {
+func (t *testData) runWrapper(c *k8s.NPLController) {
 	t.wg.Add(1)
 	go func() {
 		defer t.wg.Done()
@@ -253,7 +254,7 @@ func setUp(t *testing.T, tc *testConfig, objects ...runtime.Object) *testData {
 	)
 	svcInformer := informerFactory.Core().V1().Services().Informer()
 
-	c := nplk8s.NewNPLController(data.k8sClient, localPodInformer, svcInformer, data.portTable, defaultNodeName)
+	c := k8s.NewNPLController(data.k8sClient, localPodInformer, svcInformer, data.portTable, defaultNodeName)
 
 	data.runWrapper(c)
 	informerFactory.Start(data.stopCh)
@@ -302,7 +303,7 @@ func (t *testData) tearDown() {
 	os.Unsetenv("NODE_NAME")
 }
 
-func (t *testData) pollForPodAnnotation(podName string, found bool) ([]nplk8s.NPLAnnotation, error) {
+func (t *testData) pollForPodAnnotation(podName string, found bool) ([]types.NPLAnnotation, error) {
 	var data string
 	var exists bool
 	// do not use PollImmediate: 1 second is reserved for the controller to do his job and
@@ -311,7 +312,7 @@ func (t *testData) pollForPodAnnotation(podName string, found bool) ([]nplk8s.NP
 		updatedPod, err := t.k8sClient.CoreV1().Pods(defaultNS).Get(context.TODO(), podName, metav1.GetOptions{})
 		require.NoError(t, err, "Failed to get Pod")
 		annotation := updatedPod.GetAnnotations()
-		data, exists = annotation[nplk8s.NPLAnnotationKey]
+		data, exists = annotation[types.NPLAnnotationKey]
 		if found {
 			return exists, nil
 		}
@@ -319,12 +320,12 @@ func (t *testData) pollForPodAnnotation(podName string, found bool) ([]nplk8s.NP
 	})
 
 	if err != nil {
-		return []nplk8s.NPLAnnotation{}, err
+		return []types.NPLAnnotation{}, err
 	}
 	if data == "" {
-		return []nplk8s.NPLAnnotation{}, nil
+		return []types.NPLAnnotation{}, nil
 	}
-	var nplValue []nplk8s.NPLAnnotation
+	var nplValue []types.NPLAnnotation
 	err = json.Unmarshal([]byte(data), &nplValue)
 	return nplValue, err
 }
@@ -394,7 +395,7 @@ func TestSvcUpdateAnnotation(t *testing.T) {
 	defer testData.tearDown()
 
 	// Disable NPL.
-	testSvc.Annotations = map[string]string{nplk8s.NPLEnabledAnnotationKey: "false"}
+	testSvc.Annotations = map[string]string{types.NPLEnabledAnnotationKey: "false"}
 	testData.updateServiceOrFail(testSvc)
 
 	// Check that annotation and the rule is removed.
@@ -403,7 +404,7 @@ func TestSvcUpdateAnnotation(t *testing.T) {
 	assert.False(t, testData.portTable.RuleExists(defaultPodIP, defaultPort, protocolTCP))
 
 	// Enable NPL back.
-	testSvc.Annotations = map[string]string{nplk8s.NPLEnabledAnnotationKey: "true"}
+	testSvc.Annotations = map[string]string{types.NPLEnabledAnnotationKey: "true"}
 	testData.updateServiceOrFail(testSvc)
 
 	_, err = testData.pollForPodAnnotation(testPod.Name, true)
@@ -743,7 +744,7 @@ func TestInitInvalidPod(t *testing.T) {
 	testPod := getTestPod()
 	// assign an invalid annotation
 	annotations := map[string]string{
-		nplk8s.NPLAnnotationKey: "[{\"podPort\":53,\"nodeIP\":\"10.10.10.10\", \"nodePort\": 30000}]",
+		types.NPLAnnotationKey: "[{\"podPort\":53,\"nodeIP\":\"10.10.10.10\", \"nodePort\": 30000}]",
 	}
 	testPod.SetAnnotations(annotations)
 	testData := setUp(t, newTestConfig(), testSvc, testPod)
diff --git a/pkg/agent/nodeportlocal/testing/annotations.go b/pkg/agent/nodeportlocal/testing/annotations.go
index 415a231af..57f9a999d 100644
--- a/pkg/agent/nodeportlocal/testing/annotations.go
+++ b/pkg/agent/nodeportlocal/testing/annotations.go
@@ -22,14 +22,14 @@ import (
 
 	"github.com/stretchr/testify/assert"
 
-	nplk8s "antrea.io/antrea/pkg/agent/nodeportlocal/k8s"
+	"antrea.io/antrea/pkg/agent/nodeportlocal/types"
 )
 
 type ExpectedNPLAnnotations struct {
 	nodeIP       *string
 	nplStartPort int
 	nplEndPort   int
-	annotations  []nplk8s.NPLAnnotation
+	annotations  []types.NPLAnnotation
 }
 
 func NewExpectedNPLAnnotations(nodeIP *string, nplStartPort, nplEndPort int) *ExpectedNPLAnnotations {
@@ -40,7 +40,7 @@ func NewExpectedNPLAnnotations(nodeIP *string, nplStartPort, nplEndPort int) *Ex
 	}
 }
 
-func (a *ExpectedNPLAnnotations) find(podPort int, protocol string) *nplk8s.NPLAnnotation {
+func (a *ExpectedNPLAnnotations) find(podPort int, protocol string) *types.NPLAnnotation {
 	for _, annotation := range a.annotations {
 		if annotation.PodPort == podPort && annotation.Protocol == protocol {
 			return &annotation
@@ -51,7 +51,7 @@ func (a *ExpectedNPLAnnotations) find(podPort int, protocol string) *nplk8s.NPLA
 
 func (a *ExpectedNPLAnnotations) Add(nodePort *int, podPort int, protocol string) *ExpectedNPLAnnotations {
 	protocols := []string{protocol}
-	annotation := nplk8s.NPLAnnotation{PodPort: podPort, Protocol: protocol, Protocols: protocols}
+	annotation := types.NPLAnnotation{PodPort: podPort, Protocol: protocol, Protocols: protocols}
 	if nodePort != nil {
 		annotation.NodePort = *nodePort
 	}
@@ -62,7 +62,7 @@ func (a *ExpectedNPLAnnotations) Add(nodePort *int, podPort int, protocol string
 	return a
 }
 
-func (a *ExpectedNPLAnnotations) Check(t *testing.T, nplValue []nplk8s.NPLAnnotation) {
+func (a *ExpectedNPLAnnotations) Check(t *testing.T, nplValue []types.NPLAnnotation) {
 	assert.Equal(t, len(a.annotations), len(nplValue), "Invalid number of NPL annotations")
 	for _, nplAnnotation := range nplValue {
 		expectedAnnotation := a.find(nplAnnotation.PodPort, nplAnnotation.Protocol)
diff --git a/pkg/flowaggregator/flowaggregator_test.go b/pkg/flowaggregator/flowaggregator_test.go
index c5622faed..07379a607 100644
--- a/pkg/flowaggregator/flowaggregator_test.go
+++ b/pkg/flowaggregator/flowaggregator_test.go
@@ -16,18 +16,28 @@ package flowaggregator
 
 import (
 	"bytes"
+	"fmt"
+	"os"
+	"path/filepath"
+	"sync"
 	"testing"
 	"time"
 
+	"github.com/fsnotify/fsnotify"
 	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 	ipfixentities "github.com/vmware/go-ipfix/pkg/entities"
 	ipfixentitiestesting "github.com/vmware/go-ipfix/pkg/entities/testing"
 	ipfixintermediate "github.com/vmware/go-ipfix/pkg/intermediate"
 	ipfixregistry "github.com/vmware/go-ipfix/pkg/registry"
+	"gopkg.in/yaml.v2"
 	"k8s.io/client-go/informers"
 	"k8s.io/client-go/kubernetes/fake"
 
+	flowaggregatorconfig "antrea.io/antrea/pkg/config/flowaggregator"
+	"antrea.io/antrea/pkg/flowaggregator/clickhouseclient"
+	"antrea.io/antrea/pkg/flowaggregator/options"
 	ipfixtest "antrea.io/antrea/pkg/ipfix/testing"
 )
 
@@ -278,3 +288,85 @@ func createElement(name string, enterpriseID uint32) ipfixentities.InfoElementWi
 	ieWithValue, _ := ipfixentities.DecodeAndCreateInfoElementWithValue(element, nil)
 	return ieWithValue
 }
+
+func TestFlowAggregator_watchConfiguration(t *testing.T) {
+	dbEP := &clickhouseclient.ClickHouseExportProcess{}
+	var wg sync.WaitGroup
+	testcases := []struct {
+		FlowCollectorEnable       bool
+		FlowCollectorAddress      string
+		ClickHouseEnable          bool
+		externalFlowCollectorAddr string
+		message                   []updateParam
+		dbExportProcess           *clickhouseclient.ClickHouseExportProcess
+	}{
+		{true, "10.10.10.10:155", false, "", []updateParam{updateExternalFlowCollectorAddr}, nil},
+		{false, "", true, "addr", []updateParam{disableFlowCollector}, nil},
+		{false, "", true, "", []updateParam{enableClickHouse}, nil},
+		{false, "", true, "", []updateParam{updateClickHouseParam}, dbEP},
+		{true, "10.10.10.10:155", false, "", []updateParam{updateExternalFlowCollectorAddr, disableClickHouse}, dbEP},
+	}
+
+	for i, tc := range testcases {
+		t.Run(fmt.Sprintf("subtest: %d", i), func(t *testing.T) {
+			stopCh := make(chan struct{})
+			opt := options.Options{
+				Config: &flowaggregatorconfig.FlowAggregatorConfig{
+					FlowCollector: flowaggregatorconfig.FlowCollectorConfig{
+						Enable:  tc.FlowCollectorEnable,
+						Address: tc.FlowCollectorAddress,
+					},
+					ClickHouse: flowaggregatorconfig.ClickHouseConfig{
+						Enable: tc.ClickHouseEnable,
+					},
+				},
+			}
+			// create watcher
+			fileName := fmt.Sprintf("test_%d.config", i)
+			configWatcher, err := fsnotify.NewWatcher()
+			require.NoError(t, err)
+			flowAggregator := &flowAggregator{
+				// use a larger buffer to prevent the buffered channel from blocking
+				updateCh:                  make(chan updateMsg, 100),
+				externalFlowCollectorAddr: tc.externalFlowCollectorAddr,
+				dbExportProcess:           tc.dbExportProcess,
+				configFile:                fileName,
+				configWatcher:             configWatcher,
+			}
+			dir := filepath.Dir(fileName)
+			f, err := os.Create(flowAggregator.configFile)
+			require.NoError(t, err)
+			err = flowAggregator.configWatcher.Add(dir)
+			require.NoError(t, err)
+			err = f.Close()
+			require.NoError(t, err)
+			err = os.Remove(flowAggregator.configFile)
+			require.NoError(t, err)
+			f, err = os.Create(flowAggregator.configFile)
+			require.NoError(t, err)
+			b, err := yaml.Marshal(opt.Config)
+			require.NoError(t, err)
+			_, err = f.Write(b)
+			require.NoError(t, err)
+			err = f.Close()
+			require.NoError(t, err)
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				flowAggregator.watchConfiguration(stopCh)
+			}()
+
+			for _, message := range tc.message {
+				select {
+				case msg := <-flowAggregator.updateCh:
+					assert.Equal(t, message, msg.param)
+				case <-time.After(1 * time.Minute):
+					assert.NoError(t, fmt.Errorf("timeout"))
+				}
+			}
+			close(stopCh)
+			wg.Wait()
+			os.Remove(flowAggregator.configFile)
+		})
+	}
+}
diff --git a/pkg/util/flowexport/flowexport_test.go b/pkg/util/flowexport/flowexport_test.go
index 825e8cfe0..095c88768 100644
--- a/pkg/util/flowexport/flowexport_test.go
+++ b/pkg/util/flowexport/flowexport_test.go
@@ -21,7 +21,7 @@ import (
 
 	"github.com/stretchr/testify/assert"
 
-	"antrea.io/antrea/pkg/flowaggregator"
+	flowaggregatorconfig "antrea.io/antrea/pkg/config/flowaggregator"
 )
 
 const (
@@ -133,24 +133,24 @@ func TestParseFlowIntervalString(t *testing.T) {
 func TestParseTransportProtocol(t *testing.T) {
 	testcases := []struct {
 		// input
-		transportProtocolInput flowaggregator.AggregatorTransportProtocol
+		transportProtocolInput flowaggregatorconfig.AggregatorTransportProtocol
 		// expectations
-		expectedTransportProtocol flowaggregator.AggregatorTransportProtocol
+		expectedTransportProtocol flowaggregatorconfig.AggregatorTransportProtocol
 		expectedError             error
 	}{
 		{
 			transportProtocolInput:    "tcp",
-			expectedTransportProtocol: flowaggregator.AggregatorTransportProtocolTCP,
+			expectedTransportProtocol: flowaggregatorconfig.AggregatorTransportProtocolTCP,
 			expectedError:             nil,
 		},
 		{
 			transportProtocolInput:    "UDP",
-			expectedTransportProtocol: flowaggregator.AggregatorTransportProtocolUDP,
+			expectedTransportProtocol: flowaggregatorconfig.AggregatorTransportProtocolUDP,
 			expectedError:             nil,
 		},
 		{
 			transportProtocolInput:    "Tcp",
-			expectedTransportProtocol: flowaggregator.AggregatorTransportProtocolTCP,
+			expectedTransportProtocol: flowaggregatorconfig.AggregatorTransportProtocolTCP,
 			expectedError:             nil,
 		},
 		{
diff --git a/test/e2e/antreaipam_test.go b/test/e2e/antreaipam_test.go
index 0a1780a9a..aef5b9e1f 100644
--- a/test/e2e/antreaipam_test.go
+++ b/test/e2e/antreaipam_test.go
@@ -28,7 +28,6 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	appsv1 "k8s.io/api/apps/v1"
-	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/wait"
 	utilnet "k8s.io/utils/net"
@@ -328,16 +327,12 @@ func testAntreaIPAMStatefulSet(t *testing.T, data *TestData, dedicatedIPPoolKey
 	}
 	checkStatefulSetIPPoolAllocation(t, data, stsName, testAntreaIPAMNamespace, ipPoolName, ipOffsets, reservedIPOffsets)
 
-	podMutateFunc := func(pod *corev1.Pod) {
-		if pod.Annotations == nil {
-			pod.Annotations = map[string]string{}
-		}
-		if dedicatedIPPoolKey != nil {
-			pod.Annotations[annotation.AntreaIPAMAnnotationKey] = ipPoolName
-		}
-	}
 	podName := randName("test-standalone-pod-")
-	err = data.createPodOnNode(podName, testAntreaIPAMNamespace, controlPlaneNodeName(), agnhostImage, []string{"sleep", "3600"}, nil, nil, nil, false, podMutateFunc)
+	podAnnotations := map[string]string{}
+	if dedicatedIPPoolKey != nil {
+		podAnnotations[annotation.AntreaIPAMAnnotationKey] = ipPoolName
+	}
+	err = NewPodBuilder(podName, testAntreaIPAMNamespace, agnhostImage).OnNode(controlPlaneNodeName()).WithCommand([]string{"sleep", "3600"}).WithAnnotations(podAnnotations).Create(data)
 	if err != nil {
 		t.Fatalf("Error when creating Pod '%s': %v", podName, err)
 	}
diff --git a/test/e2e/antreapolicy_test.go b/test/e2e/antreapolicy_test.go
index 09af4823b..98f45b124 100644
--- a/test/e2e/antreapolicy_test.go
+++ b/test/e2e/antreapolicy_test.go
@@ -75,6 +75,7 @@ const (
 func TestAntreaPolicyStats(t *testing.T) {
 	skipIfHasWindowsNodes(t)
 	skipIfAntreaPolicyDisabled(t)
+	skipIfNetworkPolicyStatsDisabled(t)
 
 	data, err := setupTest(t)
 	if err != nil {
@@ -83,11 +84,9 @@ func TestAntreaPolicyStats(t *testing.T) {
 	defer teardownTest(t, data)
 
 	t.Run("testANPNetworkPolicyStatsWithDropAction", func(t *testing.T) {
-		skipIfNetworkPolicyStatsDisabled(t)
 		testANPNetworkPolicyStatsWithDropAction(t, data)
 	})
 	t.Run("testAntreaClusterNetworkPolicyStats", func(t *testing.T) {
-		skipIfNetworkPolicyStatsDisabled(t)
 		testAntreaClusterNetworkPolicyStats(t, data)
 	})
 }
diff --git a/test/e2e/bandwidth_test.go b/test/e2e/bandwidth_test.go
index a48aa867c..ecc97f9d9 100644
--- a/test/e2e/bandwidth_test.go
+++ b/test/e2e/bandwidth_test.go
@@ -68,14 +68,13 @@ func TestBenchmarkBandwidth(t *testing.T) {
 
 // testBenchmarkBandwidthIntraNode runs the bandwidth benchmark between Pods on same node.
 func testBenchmarkBandwidthIntraNode(t *testing.T, data *TestData) {
-
-	if err := data.createPodOnNode("perftest-a", data.testNamespace, controlPlaneNodeName(), perftoolImage, nil, nil, nil, nil, false, nil); err != nil {
+	if err := NewPodBuilder("perftest-a", data.testNamespace, perftoolImage).OnNode(controlPlaneNodeName()).Create(data); err != nil {
 		t.Fatalf("Error when creating the perftest client Pod: %v", err)
 	}
 	if err := data.podWaitForRunning(defaultTimeout, "perftest-a", data.testNamespace); err != nil {
 		t.Fatalf("Error when waiting for the perftest client Pod: %v", err)
 	}
-	if err := data.createPodOnNode("perftest-b", data.testNamespace, controlPlaneNodeName(), perftoolImage, nil, nil, nil, []v1.ContainerPort{{Protocol: v1.ProtocolTCP, ContainerPort: iperfPort}}, false, nil); err != nil {
+	if err := NewPodBuilder("perftest-b", data.testNamespace, perftoolImage).OnNode(controlPlaneNodeName()).WithPorts([]v1.ContainerPort{{Protocol: v1.ProtocolTCP, ContainerPort: iperfPort}}).Create(data); err != nil {
 		t.Fatalf("Error when creating the perftest server Pod: %v", err)
 	}
 	podBIPs, err := data.podWaitForIPs(defaultTimeout, "perftest-b", data.testNamespace)
@@ -96,13 +95,13 @@ func benchmarkBandwidthService(t *testing.T, endpointNode, clientNode string, da
 	if err != nil {
 		t.Fatalf("Error when creating perftest service: %v", err)
 	}
-	if err := data.createPodOnNode("perftest-a", data.testNamespace, clientNode, perftoolImage, nil, nil, nil, nil, false, nil); err != nil {
+	if err := NewPodBuilder("perftest-a", data.testNamespace, perftoolImage).OnNode(clientNode).Create(data); err != nil {
 		t.Fatalf("Error when creating the perftest client Pod: %v", err)
 	}
 	if err := data.podWaitForRunning(defaultTimeout, "perftest-a", data.testNamespace); err != nil {
 		t.Fatalf("Error when waiting for the perftest client Pod: %v", err)
 	}
-	if err := data.createPodOnNode("perftest-b", data.testNamespace, endpointNode, perftoolImage, nil, nil, nil, []v1.ContainerPort{{Protocol: v1.ProtocolTCP, ContainerPort: iperfPort}}, false, nil); err != nil {
+	if err := NewPodBuilder("perftest-b", data.testNamespace, perftoolImage).OnNode(clientNode).WithPorts([]v1.ContainerPort{{Protocol: v1.ProtocolTCP, ContainerPort: iperfPort}}).Create(data); err != nil {
 		t.Fatalf("Error when creating the perftest server Pod: %v", err)
 	}
 	if err := data.podWaitForRunning(defaultTimeout, "perftest-b", data.testNamespace); err != nil {
@@ -159,22 +158,22 @@ func testPodTrafficShaping(t *testing.T, data *TestData) {
 		t.Run(tt.name, func(t *testing.T) {
 			clientPodName := fmt.Sprintf("client-a-%d", i)
 			serverPodName := fmt.Sprintf("server-a-%d", i)
-			if err := data.createPodOnNode(clientPodName, data.testNamespace, nodeName, perftoolImage, nil, nil, nil, nil, false, func(pod *v1.Pod) {
-				pod.Annotations = map[string]string{
+			if err := NewPodBuilder(clientPodName, data.testNamespace, perftoolImage).OnNode(nodeName).WithAnnotations(
+				map[string]string{
 					"kubernetes.io/egress-bandwidth": fmt.Sprintf("%dM", tt.clientEgressBandwidth),
-				}
-			}); err != nil {
+				},
+			).Create(data); err != nil {
 				t.Fatalf("Error when creating the perftest client Pod: %v", err)
 			}
 			defer deletePodWrapper(t, data, data.testNamespace, clientPodName)
 			if err := data.podWaitForRunning(defaultTimeout, clientPodName, data.testNamespace); err != nil {
 				t.Fatalf("Error when waiting for the perftest client Pod: %v", err)
 			}
-			if err := data.createPodOnNode(serverPodName, data.testNamespace, nodeName, perftoolImage, nil, nil, nil, []v1.ContainerPort{{Protocol: v1.ProtocolTCP, ContainerPort: iperfPort}}, false, func(pod *v1.Pod) {
-				pod.Annotations = map[string]string{
+			if err := NewPodBuilder(serverPodName, data.testNamespace, perftoolImage).OnNode(nodeName).WithPorts([]v1.ContainerPort{{Protocol: v1.ProtocolTCP, ContainerPort: iperfPort}}).WithAnnotations(
+				map[string]string{
 					"kubernetes.io/ingress-bandwidth": fmt.Sprintf("%dM", tt.serverIngressBandwidth),
-				}
-			}); err != nil {
+				},
+			).Create(data); err != nil {
 				t.Fatalf("Error when creating the perftest server Pod: %v", err)
 			}
 			defer deletePodWrapper(t, data, data.testNamespace, serverPodName)
diff --git a/test/e2e/egress_test.go b/test/e2e/egress_test.go
index 70f9ed86f..fa48e45d4 100644
--- a/test/e2e/egress_test.go
+++ b/test/e2e/egress_test.go
@@ -125,10 +125,7 @@ ip netns exec %[1]s ip link set dev %[1]s-a up && \
 ip netns exec %[1]s ip route replace default via %[3]s && \
 ip netns exec %[1]s /agnhost netexec
 `, tt.fakeServer, tt.serverIP, tt.localIP0, tt.localIP1, tt.ipMaskLen)
-			if err := data.createPodOnNode(tt.fakeServer, data.testNamespace, egressNode, agnhostImage, []string{"sh", "-c", cmd}, nil, nil, nil, true, func(pod *v1.Pod) {
-				privileged := true
-				pod.Spec.Containers[0].SecurityContext = &v1.SecurityContext{Privileged: &privileged}
-			}); err != nil {
+			if err := NewPodBuilder(tt.fakeServer, data.testNamespace, agnhostImage).OnNode(egressNode).WithCommand([]string{"sh", "-c", cmd}).InHostNetwork().Privileged().Create(data); err != nil {
 				t.Fatalf("Failed to create server Pod: %v", err)
 			}
 			defer deletePodWrapper(t, data, data.testNamespace, tt.fakeServer)
@@ -239,9 +236,7 @@ ip netns exec %[1]s /agnhost netexec
 				clientIPStr = fmt.Sprintf("[%s]", clientIPStr)
 			}
 			cmd = fmt.Sprintf("wget -T 3 -O - %s:8080/clientip | grep %s:", serverIPStr, clientIPStr)
-			if err := data.createPodOnNode(initialIPChecker, data.testNamespace, egressNode, busyboxImage, []string{"sh", "-c", cmd}, nil, nil, nil, false, func(pod *v1.Pod) {
-				pod.Spec.RestartPolicy = v1.RestartPolicyNever
-			}); err != nil {
+			if err := NewPodBuilder(initialIPChecker, data.testNamespace, agnhostImage).OnNode(egressNode).WithCommand([]string{"sh", "-c", cmd}).Create(data); err != nil {
 				t.Fatalf("Failed to create Pod initial-ip-checker: %v", err)
 			}
 			defer data.DeletePod(data.testNamespace, initialIPChecker)
diff --git a/test/e2e/fixtures.go b/test/e2e/fixtures.go
index 8c9a910e1..4c5b3e62b 100644
--- a/test/e2e/fixtures.go
+++ b/test/e2e/fixtures.go
@@ -228,7 +228,7 @@ func setupTestForFlowAggregator(tb testing.TB) (*TestData, bool, bool, error) {
 		return testData, v4Enabled, v6Enabled, err
 	}
 	// Create pod using ipfix collector image
-	if err = testData.createPodOnNode("ipfix-collector", testData.testNamespace, "", ipfixCollectorImage, nil, nil, nil, nil, true, nil); err != nil {
+	if err := NewPodBuilder("ipfix-collector", testData.testNamespace, ipfixCollectorImage).InHostNetwork().Create(testData); err != nil {
 		tb.Errorf("Error when creating the ipfix collector Pod: %v", err)
 	}
 	ipfixCollectorIP, err := testData.podWaitForIPs(defaultTimeout, "ipfix-collector", testData.testNamespace)
diff --git a/test/e2e/flowaggregator_test.go b/test/e2e/flowaggregator_test.go
index c8e18b49f..8809a3525 100644
--- a/test/e2e/flowaggregator_test.go
+++ b/test/e2e/flowaggregator_test.go
@@ -1269,7 +1269,11 @@ func deployDenyNetworkPolicies(t *testing.T, data *TestData, pod1, pod2 string)
 }
 
 func createPerftestPods(data *TestData) (podAIPs *PodIPs, podBIPs *PodIPs, podCIPs *PodIPs, podDIPs *PodIPs, podEIPs *PodIPs, err error) {
-	if err := data.createPodOnNode("perftest-a", data.testNamespace, controlPlaneNodeName(), perftoolImage, nil, nil, nil, nil, false, nil); err != nil {
+	create := func(name string, nodeName string, ports []corev1.ContainerPort) error {
+		return NewPodBuilder(name, data.testNamespace, perftoolImage).OnNode(nodeName).WithPorts(ports).Create(data)
+	}
+
+	if err := create("perftest-a", controlPlaneNodeName(), nil); err != nil {
 		return nil, nil, nil, nil, nil, fmt.Errorf("Error when creating the perftest client Pod: %v", err)
 	}
 	podAIPs, err = data.podWaitForIPs(defaultTimeout, "perftest-a", data.testNamespace)
@@ -1277,7 +1281,7 @@ func createPerftestPods(data *TestData) (podAIPs *PodIPs, podBIPs *PodIPs, podCI
 		return nil, nil, nil, nil, nil, fmt.Errorf("Error when waiting for the perftest client Pod: %v", err)
 	}
 
-	if err := data.createPodOnNode("perftest-b", data.testNamespace, controlPlaneNodeName(), perftoolImage, nil, nil, nil, []corev1.ContainerPort{{Protocol: corev1.ProtocolTCP, ContainerPort: iperfPort}}, false, nil); err != nil {
+	if err := create("perftest-b", controlPlaneNodeName(), []corev1.ContainerPort{{Protocol: corev1.ProtocolTCP, ContainerPort: iperfPort}}); err != nil {
 		return nil, nil, nil, nil, nil, fmt.Errorf("Error when creating the perftest server Pod: %v", err)
 	}
 	podBIPs, err = data.podWaitForIPs(defaultTimeout, "perftest-b", data.testNamespace)
@@ -1285,7 +1289,7 @@ func createPerftestPods(data *TestData) (podAIPs *PodIPs, podBIPs *PodIPs, podCI
 		return nil, nil, nil, nil, nil, fmt.Errorf("Error when getting the perftest server Pod's IPs: %v", err)
 	}
 
-	if err := data.createPodOnNode("perftest-c", data.testNamespace, workerNodeName(1), perftoolImage, nil, nil, nil, []corev1.ContainerPort{{Protocol: corev1.ProtocolTCP, ContainerPort: iperfPort}}, false, nil); err != nil {
+	if err := create("perftest-c", workerNodeName(1), []corev1.ContainerPort{{Protocol: corev1.ProtocolTCP, ContainerPort: iperfPort}}); err != nil {
 		return nil, nil, nil, nil, nil, fmt.Errorf("Error when creating the perftest server Pod: %v", err)
 	}
 	podCIPs, err = data.podWaitForIPs(defaultTimeout, "perftest-c", data.testNamespace)
@@ -1293,7 +1297,7 @@ func createPerftestPods(data *TestData) (podAIPs *PodIPs, podBIPs *PodIPs, podCI
 		return nil, nil, nil, nil, nil, fmt.Errorf("Error when getting the perftest server Pod's IPs: %v", err)
 	}
 
-	if err := data.createPodOnNode("perftest-d", data.testNamespace, controlPlaneNodeName(), perftoolImage, nil, nil, nil, []corev1.ContainerPort{{Protocol: corev1.ProtocolTCP, ContainerPort: iperfPort}}, false, nil); err != nil {
+	if err := create("perftest-d", controlPlaneNodeName(), []corev1.ContainerPort{{Protocol: corev1.ProtocolTCP, ContainerPort: iperfPort}}); err != nil {
 		return nil, nil, nil, nil, nil, fmt.Errorf("Error when creating the perftest server Pod: %v", err)
 	}
 	podDIPs, err = data.podWaitForIPs(defaultTimeout, "perftest-d", data.testNamespace)
@@ -1301,7 +1305,7 @@ func createPerftestPods(data *TestData) (podAIPs *PodIPs, podBIPs *PodIPs, podCI
 		return nil, nil, nil, nil, nil, fmt.Errorf("Error when getting the perftest server Pod's IPs: %v", err)
 	}
 
-	if err := data.createPodOnNode("perftest-e", data.testNamespace, workerNodeName(1), perftoolImage, nil, nil, nil, []corev1.ContainerPort{{Protocol: corev1.ProtocolTCP, ContainerPort: iperfPort}}, false, nil); err != nil {
+	if err := create("perftest-e", workerNodeName(1), []corev1.ContainerPort{{Protocol: corev1.ProtocolTCP, ContainerPort: iperfPort}}); err != nil {
 		return nil, nil, nil, nil, nil, fmt.Errorf("Error when creating the perftest server Pod: %v", err)
 	}
 	podEIPs, err = data.podWaitForIPs(defaultTimeout, "perftest-e", data.testNamespace)
diff --git a/test/e2e/framework.go b/test/e2e/framework.go
index 0c626a767..ea5c97218 100644
--- a/test/e2e/framework.go
+++ b/test/e2e/framework.go
@@ -1100,58 +1100,150 @@ func getImageName(uri string) string {
 	return paths[len(paths)-1]
 }
 
-// createPodOnNode creates a pod in the test namespace with a container whose type is decided by imageName.
-// Pod will be scheduled on the specified Node (if nodeName is not empty).
-// mutateFunc can be used to customize the Pod if the other parameters don't meet the requirements.
-func (data *TestData) createPodOnNode(name string, ns string, nodeName string, image string, command []string, args []string, env []corev1.EnvVar, ports []corev1.ContainerPort, hostNetwork bool, mutateFunc func(*corev1.Pod)) error {
-	// image could be a fully qualified URI which can't be used as container name and label value,
-	// extract the image name from it.
-	imageName := getImageName(image)
-	return data.CreatePodOnNodeInNamespace(name, ns, nodeName, imageName, image, command, args, env, ports, hostNetwork, mutateFunc)
+type PodBuilder struct {
+	Name               string
+	Namespace          string
+	Image              string
+	ContainerName      string
+	Command            []string
+	Args               []string
+	Env                []corev1.EnvVar
+	Ports              []corev1.ContainerPort
+	HostNetwork        bool
+	IsPrivileged       bool
+	ServiceAccountName string
+	Annotations        map[string]string
+	Labels             map[string]string
+	NodeName           string
+	MutateFunc         func(*corev1.Pod)
 }
 
-// CreatePodOnNodeInNamespace creates a pod in the provided namespace with a container whose type is decided by imageName.
-// Pod will be scheduled on the specified Node (if nodeName is not empty).
-// mutateFunc can be used to customize the Pod if the other parameters don't meet the requirements.
-func (data *TestData) CreatePodOnNodeInNamespace(name, ns string, nodeName, ctrName string, image string, command []string, args []string, env []corev1.EnvVar, ports []corev1.ContainerPort, hostNetwork bool, mutateFunc func(*corev1.Pod)) error {
+func NewPodBuilder(name, ns, image string) *PodBuilder {
+	return &PodBuilder{
+		Name:      name,
+		Namespace: ns,
+		Image:     image,
+	}
+}
+
+func (b *PodBuilder) WithContainerName(ctrName string) *PodBuilder {
+	b.ContainerName = ctrName
+	return b
+}
+
+func (b *PodBuilder) WithCommand(command []string) *PodBuilder {
+	b.Command = command
+	return b
+}
+
+func (b *PodBuilder) WithArgs(args []string) *PodBuilder {
+	b.Args = args
+	return b
+}
+
+func (b *PodBuilder) WithEnv(env []corev1.EnvVar) *PodBuilder {
+	b.Env = env
+	return b
+}
+
+func (b *PodBuilder) WithPorts(ports []corev1.ContainerPort) *PodBuilder {
+	b.Ports = ports
+	return b
+}
+
+func (b *PodBuilder) WithHostNetwork(v bool) *PodBuilder {
+	b.HostNetwork = v
+	return b
+}
+
+func (b *PodBuilder) InHostNetwork() *PodBuilder {
+	return b.WithHostNetwork(true)
+}
+
+func (b *PodBuilder) Privileged() *PodBuilder {
+	b.IsPrivileged = true
+	return b
+}
+
+func (b *PodBuilder) WithServiceAccountName(name string) *PodBuilder {
+	b.ServiceAccountName = name
+	return b
+}
+
+func (b *PodBuilder) WithAnnotations(annotations map[string]string) *PodBuilder {
+	b.Annotations = annotations
+	return b
+}
+
+func (b *PodBuilder) WithLabels(labels map[string]string) *PodBuilder {
+	b.Labels = labels
+	return b
+}
+
+func (b *PodBuilder) OnNode(nodeName string) *PodBuilder {
+	b.NodeName = nodeName
+	return b
+}
+
+func (b *PodBuilder) WithMutateFunc(f func(*corev1.Pod)) *PodBuilder {
+	b.MutateFunc = f
+	return b
+}
+
+func (b *PodBuilder) Create(data *TestData) error {
+	containerName := b.ContainerName
+	if containerName == "" {
+		containerName = getImageName(b.Image)
+	}
 	podSpec := corev1.PodSpec{
 		Containers: []corev1.Container{
 			{
-				Name:            ctrName,
-				Image:           image,
+				Name:            containerName,
+				Image:           b.Image,
 				ImagePullPolicy: corev1.PullIfNotPresent,
-				Command:         command,
-				Args:            args,
-				Env:             env,
-				Ports:           ports,
+				Command:         b.Command,
+				Args:            b.Args,
+				Env:             b.Env,
+				Ports:           b.Ports,
+				SecurityContext: &corev1.SecurityContext{
+					Privileged: &b.IsPrivileged,
+				},
 			},
 		},
-		RestartPolicy: corev1.RestartPolicyNever,
-		HostNetwork:   hostNetwork,
+		RestartPolicy:      corev1.RestartPolicyNever,
+		HostNetwork:        b.HostNetwork,
+		ServiceAccountName: b.ServiceAccountName,
 	}
-	if nodeName != "" {
+	if b.NodeName != "" {
 		podSpec.NodeSelector = map[string]string{
-			"kubernetes.io/hostname": nodeName,
+			"kubernetes.io/hostname": b.NodeName,
 		}
 	}
-	if nodeName == controlPlaneNodeName() {
+	if b.NodeName == controlPlaneNodeName() {
 		// tolerate NoSchedule taint if we want Pod to run on control-plane Node
 		podSpec.Tolerations = controlPlaneNoScheduleTolerations()
 	}
 	pod := &corev1.Pod{
 		ObjectMeta: metav1.ObjectMeta{
-			Name: name,
+			Name:        b.Name,
+			Annotations: map[string]string{},
 			Labels: map[string]string{
-				"antrea-e2e": name,
-				"app":        ctrName,
+				"antrea-e2e": b.Name,
+				"app":        containerName,
 			},
 		},
 		Spec: podSpec,
 	}
-	if mutateFunc != nil {
-		mutateFunc(pod)
+	for k, v := range b.Annotations {
+		pod.Annotations[k] = v
 	}
-	if _, err := data.clientset.CoreV1().Pods(ns).Create(context.TODO(), pod, metav1.CreateOptions{}); err != nil {
+	for k, v := range b.Labels {
+		pod.Labels[k] = v
+	}
+	if b.MutateFunc != nil {
+		b.MutateFunc(pod)
+	}
+	if _, err := data.clientset.CoreV1().Pods(b.Namespace).Create(context.TODO(), pod, metav1.CreateOptions{}); err != nil {
 		return err
 	}
 	return nil
@@ -1160,41 +1252,35 @@ func (data *TestData) CreatePodOnNodeInNamespace(name, ns string, nodeName, ctrN
 // createBusyboxPodOnNode creates a Pod in the test namespace with a single busybox container. The
 // Pod will be scheduled on the specified Node (if nodeName is not empty).
 func (data *TestData) createBusyboxPodOnNode(name string, ns string, nodeName string, hostNetwork bool) error {
-	sleepDuration := 3600 // seconds
-	return data.createPodOnNode(name, ns, nodeName, busyboxImage, []string{"sleep", strconv.Itoa(sleepDuration)}, nil, nil, nil, hostNetwork, nil)
+	return NewPodBuilder(name, ns, busyboxImage).OnNode(nodeName).WithCommand([]string{"sleep", "3600"}).WithHostNetwork(hostNetwork).Create(data)
 }
 
 // createMcJoinPodOnNode creates a Pod in the test namespace with a single mcjoin container. The
 // Pod will be scheduled on the specified Node (if nodeName is not empty).
 func (data *TestData) createMcJoinPodOnNode(name string, ns string, nodeName string, hostNetwork bool) error {
-	sleepDuration := 3600 // seconds
-	return data.createPodOnNode(name, ns, nodeName, mcjoinImage, []string{"sleep", strconv.Itoa(sleepDuration)}, nil, nil, nil, hostNetwork, nil)
+	return NewPodBuilder(name, ns, mcjoinImage).OnNode(nodeName).WithCommand([]string{"sleep", "3600"}).WithHostNetwork(hostNetwork).Create(data)
 }
 
 // createNetshootPodOnNode creates a Pod in the test namespace with a single netshoot container. The
 // Pod will be scheduled on the specified Node (if nodeName is not empty).
 func (data *TestData) createNetshootPodOnNode(name string, ns string, nodeName string, hostNetwork bool) error {
-	sleepDuration := 3600 // seconds
-	return data.createPodOnNode(name, ns, nodeName, netshootImage, []string{"sleep", strconv.Itoa(sleepDuration)}, nil, nil, nil, hostNetwork, nil)
+	return NewPodBuilder(name, ns, netshootImage).OnNode(nodeName).WithCommand([]string{"sleep", "3600"}).WithHostNetwork(hostNetwork).Create(data)
 }
 
 // createNginxPodOnNode creates a Pod in the test namespace with a single nginx container. The
 // Pod will be scheduled on the specified Node (if nodeName is not empty).
 func (data *TestData) createNginxPodOnNode(name string, ns string, nodeName string, hostNetwork bool) error {
-	var mutateImage func(*corev1.Pod)
-	mutateImage = nil
+	image := nginxImage
 	if clusterInfo.nodesOS[nodeName] == "windows" {
-		mutateImage = func(pod *corev1.Pod) {
-			pod.Spec.Containers[0].Image = iisImage
-		}
+		image = iisImage
 	}
-	return data.createPodOnNode(name, ns, nodeName, nginxImage, []string{}, nil, nil, []corev1.ContainerPort{
+	return NewPodBuilder(name, ns, image).OnNode(nodeName).WithPorts([]corev1.ContainerPort{
 		{
 			Name:          "http",
 			ContainerPort: 80,
 			Protocol:      corev1.ProtocolTCP,
 		},
-	}, hostNetwork, mutateImage)
+	}).WithHostNetwork(hostNetwork).Create(data)
 }
 
 // createServerPod creates a Pod that can listen to specified port and have named port set.
@@ -1207,7 +1293,7 @@ func (data *TestData) createServerPod(name string, ns string, portName string, p
 		// If hostPort is to be set, it must match the container port number.
 		port.HostPort = int32(portNum)
 	}
-	return data.createPodOnNode(name, ns, "", agnhostImage, nil, []string{cmd}, []corev1.EnvVar{env}, []corev1.ContainerPort{port}, hostNetwork, nil)
+	return NewPodBuilder(name, ns, agnhostImage).WithArgs([]string{cmd}).WithEnv([]corev1.EnvVar{env}).WithPorts([]corev1.ContainerPort{port}).WithHostNetwork(hostNetwork).Create(data)
 }
 
 // createCustomPod creates a Pod in given Namespace with custom labels.
@@ -1216,12 +1302,7 @@ func (data *TestData) createServerPodWithLabels(name, ns string, portNum int32,
 	env := corev1.EnvVar{Name: fmt.Sprintf("SERVE_PORT_%d", portNum), Value: "foo"}
 	port := corev1.ContainerPort{ContainerPort: portNum}
 	containerName := fmt.Sprintf("c%v", portNum)
-	mutateLabels := func(pod *corev1.Pod) {
-		for k, v := range labels {
-			pod.Labels[k] = v
-		}
-	}
-	return data.CreatePodOnNodeInNamespace(name, ns, "", containerName, agnhostImage, cmd, nil, []corev1.EnvVar{env}, []corev1.ContainerPort{port}, false, mutateLabels)
+	return NewPodBuilder(name, ns, agnhostImage).WithContainerName(containerName).WithCommand(cmd).WithEnv([]corev1.EnvVar{env}).WithPorts([]corev1.ContainerPort{port}).WithLabels(labels).Create(data)
 }
 
 // DeletePod deletes a Pod in the test namespace.
@@ -1981,32 +2062,30 @@ func (data *TestData) isProxyAll() (bool, error) {
 	return agentConf.AntreaProxy.ProxyAll, nil
 }
 
-func getFeatures(confName string) (featuregate.FeatureGate, error) {
+func GetAgentFeatures() (featuregate.FeatureGate, error) {
 	featureGate := features.DefaultMutableFeatureGate.DeepCopy()
-	var cfg interface{}
-	if err := yaml.Unmarshal([]byte(AntreaConfigMap.Data[confName]), &cfg); err != nil {
+	var cfg agentconfig.AgentConfig
+	if err := yaml.Unmarshal([]byte(AntreaConfigMap.Data[antreaAgentConfName]), &cfg); err != nil {
 		return nil, err
 	}
-	rawFeatureGateMap, ok := cfg.(map[interface{}]interface{})["featureGates"]
-	if !ok || rawFeatureGateMap == nil {
-		return featureGate, nil
-	}
-	featureGateMap := make(map[string]bool)
-	for k, v := range rawFeatureGateMap.(map[interface{}]interface{}) {
-		featureGateMap[k.(string)] = v.(bool)
-	}
-	if err := featureGate.SetFromMap(featureGateMap); err != nil {
+	err := featureGate.SetFromMap(cfg.FeatureGates)
+	if err != nil {
 		return nil, err
 	}
 	return featureGate, nil
 }
 
-func GetAgentFeatures() (featuregate.FeatureGate, error) {
-	return getFeatures(antreaAgentConfName)
-}
-
 func GetControllerFeatures() (featuregate.FeatureGate, error) {
-	return getFeatures(antreaControllerConfName)
+	featureGate := features.DefaultMutableFeatureGate.DeepCopy()
+	var cfg controllerconfig.ControllerConfig
+	if err := yaml.Unmarshal([]byte(AntreaConfigMap.Data[antreaControllerConfName]), &cfg); err != nil {
+		return nil, err
+	}
+	err := featureGate.SetFromMap(cfg.FeatureGates)
+	if err != nil {
+		return nil, err
+	}
+	return featureGate, nil
 }
 
 func (data *TestData) GetAntreaWindowsConfigMap(antreaNamespace string) (*corev1.ConfigMap, error) {
@@ -2428,29 +2507,18 @@ func (data *TestData) copyNodeFiles(nodeName string, fileName string, covDir str
 // createAgnhostPodOnNode creates a Pod in the test namespace with a single agnhost container. The
 // Pod will be scheduled on the specified Node (if nodeName is not empty).
 func (data *TestData) createAgnhostPodOnNode(name string, ns string, nodeName string, hostNetwork bool) error {
-	sleepDuration := 3600 // seconds
-	return data.createPodOnNode(name, ns, nodeName, agnhostImage, []string{"sleep", strconv.Itoa(sleepDuration)}, nil, nil, nil, hostNetwork, nil)
+	return NewPodBuilder(name, ns, agnhostImage).OnNode(nodeName).WithCommand([]string{"sleep", "3600"}).WithHostNetwork(hostNetwork).Create(data)
 }
 
 // createAgnhostPodWithSAOnNode creates a Pod in the test namespace with a single
 // agnhost container and a specific ServiceAccount. The Pod will be scheduled on
 // the specified Node (if nodeName is not empty).
 func (data *TestData) createAgnhostPodWithSAOnNode(name string, ns string, nodeName string, hostNetwork bool, serviceAccountName string) error {
-	sleepDuration := 3600 // seconds
-	mutateFunc := func(pod *corev1.Pod) {
-		pod.Spec.ServiceAccountName = serviceAccountName
-	}
-	return data.createPodOnNode(name, ns, nodeName, agnhostImage, []string{"sleep", strconv.Itoa(sleepDuration)}, nil, nil, nil, hostNetwork, mutateFunc)
+	return NewPodBuilder(name, ns, agnhostImage).OnNode(nodeName).WithCommand([]string{"sleep", "3600"}).WithHostNetwork(hostNetwork).WithServiceAccountName(serviceAccountName).Create(data)
 }
 
 func (data *TestData) createAgnhostPodOnNodeWithAnnotations(name string, ns string, nodeName string, annotations map[string]string) error {
-	sleepDuration := 3600 // seconds
-	mutateFunc := func(pod *corev1.Pod) {
-		for k, v := range annotations {
-			pod.Annotations[k] = v
-		}
-	}
-	return data.createPodOnNode(name, ns, nodeName, agnhostImage, []string{"sleep", strconv.Itoa(sleepDuration)}, nil, nil, nil, false, mutateFunc)
+	return NewPodBuilder(name, ns, agnhostImage).OnNode(nodeName).WithCommand([]string{"sleep", "3600"}).WithAnnotations(annotations).Create(data)
 }
 
 func (data *TestData) createDaemonSet(name string, ns string, ctrName string, image string, cmd []string, args []string) (*appsv1.DaemonSet, func() error, error) {
diff --git a/test/e2e/infra/vagrant/push_antrea.sh b/test/e2e/infra/vagrant/push_antrea.sh
index 80fe5b5b2..275df1270 100755
--- a/test/e2e/infra/vagrant/push_antrea.sh
+++ b/test/e2e/infra/vagrant/push_antrea.sh
@@ -170,7 +170,7 @@ FLOW_VIS_YML="/tmp/flow-visibility.yml"
 # manifest to enable FlowExporter.
 if [[ $FLOW_COLLECTOR != "" ]]; then
     echo "Generating manifest with FlowExporter enabled"
-    $THIS_DIR/../../../../hack/generate-manifest.sh --mode dev --flow-exporter > "${ANTREA_YML}"
+    $THIS_DIR/../../../../hack/generate-manifest.sh --mode dev --feature-gates FlowExporter=true > "${ANTREA_YML}"
 fi
 
 # Push Antrea image and related manifest.
diff --git a/test/e2e/ipsec_test.go b/test/e2e/ipsec_test.go
index 054331fe4..3e58bfc6b 100644
--- a/test/e2e/ipsec_test.go
+++ b/test/e2e/ipsec_test.go
@@ -25,6 +25,7 @@ import (
 	"antrea.io/antrea/pkg/agent/util"
 	agentconfig "antrea.io/antrea/pkg/config/agent"
 	controllerconfig "antrea.io/antrea/pkg/config/controller"
+	"antrea.io/antrea/pkg/features"
 )
 
 // TestIPSec is the top-level test which contains all subtests for
@@ -60,15 +61,8 @@ func TestIPSec(t *testing.T) {
 	})
 
 	t.Run("testIPSecCertificateAuth", func(t *testing.T) {
-		// restart the Controller first as Agent needs to get the CSR signed.
-		cc := func(config *controllerconfig.ControllerConfig) {
-			config.FeatureGates["IPsecCertAuth"] = true
-		}
-		if err := data.mutateAntreaConfigMap(cc, nil, true, false); err != nil {
-			t.Fatalf("Failed to enable IPsecCertAuth feature: %v", err)
-		}
+		skipIfFeatureDisabled(t, features.IPsecCertAuth, true, true)
 		ac := func(config *agentconfig.AgentConfig) {
-			config.FeatureGates["IPsecCertAuth"] = true
 			config.IPsec.AuthenticationMode = "cert"
 		}
 		if err := data.mutateAntreaConfigMap(nil, ac, false, true); err != nil {
diff --git a/test/e2e/nodeportlocal_test.go b/test/e2e/nodeportlocal_test.go
index 725036cd6..075a73bec 100644
--- a/test/e2e/nodeportlocal_test.go
+++ b/test/e2e/nodeportlocal_test.go
@@ -32,8 +32,8 @@ import (
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/wait"
 
-	"antrea.io/antrea/pkg/agent/nodeportlocal/k8s"
 	npltesting "antrea.io/antrea/pkg/agent/nodeportlocal/testing"
+	"antrea.io/antrea/pkg/agent/nodeportlocal/types"
 	agentconfig "antrea.io/antrea/pkg/config/agent"
 	"antrea.io/antrea/pkg/features"
 )
@@ -78,13 +78,14 @@ func configureNPLForAgent(t *testing.T, data *TestData, startPort, endPort int)
 func TestNodePortLocal(t *testing.T) {
 	skipIfNotIPv4Cluster(t)
 	skipIfHasWindowsNodes(t)
+	skipIfNodePortLocalDisabled(t)
 
 	data, err := setupTest(t)
 	if err != nil {
 		t.Fatalf("Error when setting up test: %v", err)
 	}
 	defer teardownTest(t, data)
-	skipIfNodePortLocalDisabled(t)
+
 	configureNPLForAgent(t, data, defaultStartPort, defaultEndPort)
 	t.Run("testNPLAddPod", func(t *testing.T) { testNPLAddPod(t, data) })
 	t.Run("testNPLMultiplePodsAgentRestart", func(t *testing.T) { testNPLMultiplePodsAgentRestart(t, data) })
@@ -127,7 +128,7 @@ func getNPLAnnotation(t *testing.T, data *TestData, r *require.Assertions, testP
 
 			ann := pod.GetAnnotations()
 			t.Logf("Got annotations %v for Pod with IP %v", ann, testPodIP.ipv4.String())
-			nplAnn, found = ann[k8s.NPLAnnotationKey]
+			nplAnn, found = ann[types.NPLAnnotationKey]
 			return found, nil
 		})
 		if err == nil {
@@ -140,14 +141,14 @@ func getNPLAnnotation(t *testing.T, data *TestData, r *require.Assertions, testP
 	return nplAnn, testPodIP.ipv4.String()
 }
 
-func getNPLAnnotations(t *testing.T, data *TestData, r *require.Assertions, testPodName string) ([]k8s.NPLAnnotation, string) {
+func getNPLAnnotations(t *testing.T, data *TestData, r *require.Assertions, testPodName string) ([]types.NPLAnnotation, string) {
 	nplAnnotationString, testPodIP := getNPLAnnotation(t, data, r, testPodName)
-	var nplAnnotations []k8s.NPLAnnotation
+	var nplAnnotations []types.NPLAnnotation
 	json.Unmarshal([]byte(nplAnnotationString), &nplAnnotations)
 	return nplAnnotations, testPodIP
 }
 
-func checkNPLRules(t *testing.T, data *TestData, r *require.Assertions, nplAnnotations []k8s.NPLAnnotation, antreaPod, podIP string, nodeName string, present bool) {
+func checkNPLRules(t *testing.T, data *TestData, r *require.Assertions, nplAnnotations []types.NPLAnnotation, antreaPod, podIP string, nodeName string, present bool) {
 	if clusterInfo.nodesOS[nodeName] == "windows" {
 		checkNPLRulesForWindowsPod(t, data, r, nplAnnotations, antreaPod, podIP, nodeName, present)
 	} else {
@@ -155,7 +156,7 @@ func checkNPLRules(t *testing.T, data *TestData, r *require.Assertions, nplAnnot
 	}
 }
 
-func checkNPLRulesForPod(t *testing.T, data *TestData, r *require.Assertions, nplAnnotations []k8s.NPLAnnotation, antreaPod, podIP string, present bool) {
+func checkNPLRulesForPod(t *testing.T, data *TestData, r *require.Assertions, nplAnnotations []types.NPLAnnotation, antreaPod, podIP string, present bool) {
 	var rules []nplRuleData
 	for _, ann := range nplAnnotations {
 		for _, protocol := range ann.Protocols {
@@ -173,7 +174,7 @@ func checkNPLRulesForPod(t *testing.T, data *TestData, r *require.Assertions, np
 	checkForNPLListeningSockets(t, data, r, antreaPod, rules, present)
 }
 
-func checkNPLRulesForWindowsPod(t *testing.T, data *TestData, r *require.Assertions, nplAnnotations []k8s.NPLAnnotation, antreaPod, podIP string, nodeName string, present bool) {
+func checkNPLRulesForWindowsPod(t *testing.T, data *TestData, r *require.Assertions, nplAnnotations []types.NPLAnnotation, antreaPod, podIP string, nodeName string, present bool) {
 	var rules []nplRuleData
 	for _, ann := range nplAnnotations {
 		for _, protocol := range ann.Protocols {
@@ -338,7 +339,7 @@ func deleteNPLRuleFromNetNat(t *testing.T, data *TestData, r *require.Assertions
 	r.NoError(err, "Error when deleting Netnat rule")
 }
 
-func checkTrafficForNPL(data *TestData, r *require.Assertions, nplAnnotations []k8s.NPLAnnotation, clientName string) {
+func checkTrafficForNPL(data *TestData, r *require.Assertions, nplAnnotations []types.NPLAnnotation, clientName string) {
 	for i := range nplAnnotations {
 		for j := range nplAnnotations[i].Protocols {
 			err := data.runNetcatCommandFromTestPodWithProtocol(clientName, data.testNamespace, "agnhost", nplAnnotations[i].NodeIP, int32(nplAnnotations[i].NodePort), nplAnnotations[i].Protocols[j])
@@ -364,7 +365,7 @@ func NPLTestMultiplePods(t *testing.T, data *TestData) {
 	r := require.New(t)
 
 	annotation := make(map[string]string)
-	annotation[k8s.NPLEnabledAnnotationKey] = "true"
+	annotation[types.NPLEnabledAnnotationKey] = "true"
 	ipFamily := corev1.IPv4Protocol
 	testData.createNginxClusterIPServiceWithAnnotations(false, &ipFamily, annotation)
 	node := nodeName(0)
@@ -416,7 +417,7 @@ func NPLTestPodAddMultiPort(t *testing.T, data *TestData) {
 	testPodName := randName("test-pod-")
 
 	annotation := make(map[string]string)
-	annotation[k8s.NPLEnabledAnnotationKey] = "true"
+	annotation[types.NPLEnabledAnnotationKey] = "true"
 	selector := make(map[string]string)
 	selector["app"] = "agnhost"
 	ipFamily := corev1.IPv4Protocol
@@ -425,17 +426,16 @@ func NPLTestPodAddMultiPort(t *testing.T, data *TestData) {
 	expectedAnnotations := newExpectedNPLAnnotations(defaultStartPort, defaultEndPort).
 		Add(nil, 80, "tcp").Add(nil, 8080, "tcp")
 
-	podcmd := "porter"
-
+	podCmd := "porter"
 	// Creating a Pod using agnhost image to support multiple ports, instead of nginx.
-	err := testData.createPodOnNode(testPodName, data.testNamespace, node, agnhostImage, nil, []string{podcmd}, []corev1.EnvVar{
+	err := NewPodBuilder(testPodName, data.testNamespace, agnhostImage).OnNode(node).WithArgs([]string{podCmd}).WithEnv([]corev1.EnvVar{
 		{
 			Name: fmt.Sprintf("SERVE_PORT_%d", 80), Value: "foo",
 		},
 		{
 			Name: fmt.Sprintf("SERVE_PORT_%d", 8080), Value: "bar",
 		},
-	}, []corev1.ContainerPort{
+	}).WithPorts([]corev1.ContainerPort{
 		{
 			Name:          "http1",
 			ContainerPort: 80,
@@ -446,8 +446,7 @@ func NPLTestPodAddMultiPort(t *testing.T, data *TestData) {
 			ContainerPort: 8080,
 			Protocol:      corev1.ProtocolTCP,
 		},
-	}, false, nil)
-
+	}).Create(data)
 	r.NoError(err, "Error creating test Pod: %v", err)
 
 	nplAnnotations, testPodIP := getNPLAnnotations(t, testData, r, testPodName)
@@ -484,7 +483,7 @@ func NPLTestPodAddMultiProtocol(t *testing.T, data *TestData) {
 	testPodName := randName("test-pod-")
 
 	annotation := make(map[string]string)
-	annotation[k8s.NPLEnabledAnnotationKey] = "true"
+	annotation[types.NPLEnabledAnnotationKey] = "true"
 	selector := make(map[string]string)
 	selector["app"] = "agnhost"
 	ipFamily := corev1.IPv4Protocol
@@ -500,13 +499,7 @@ func NPLTestPodAddMultiProtocol(t *testing.T, data *TestData) {
 	}
 	port := corev1.ContainerPort{ContainerPort: 8080}
 	containerName := fmt.Sprintf("c%v", 8080)
-	mutateLabels := func(pod *corev1.Pod) {
-		for k, v := range selector {
-			pod.Labels[k] = v
-		}
-	}
-	err := testData.CreatePodOnNodeInNamespace(testPodName, data.testNamespace, node, containerName, agnhostImage, cmd, args, []corev1.EnvVar{}, []corev1.ContainerPort{port}, false, mutateLabels)
-
+	err := NewPodBuilder(testPodName, data.testNamespace, agnhostImage).OnNode(node).WithContainerName(containerName).WithCommand(cmd).WithArgs(args).WithPorts([]corev1.ContainerPort{port}).WithLabels(selector).Create(testData)
 	r.NoError(err, "Error creating test Pod: %v", err)
 
 	nplAnnotations, testPodIP := getNPLAnnotations(t, testData, r, testPodName)
@@ -539,7 +532,7 @@ func NPLTestLocalAccess(t *testing.T, data *TestData) {
 	r := require.New(t)
 
 	annotation := make(map[string]string)
-	annotation[k8s.NPLEnabledAnnotationKey] = "true"
+	annotation[types.NPLEnabledAnnotationKey] = "true"
 	ipFamily := corev1.IPv4Protocol
 	testData.createNginxClusterIPServiceWithAnnotations(false, &ipFamily, annotation)
 	expectedAnnotations := newExpectedNPLAnnotations(defaultStartPort, defaultEndPort).Add(nil, defaultTargetPort, "tcp")
@@ -583,7 +576,7 @@ func testNPLMultiplePodsAgentRestart(t *testing.T, data *TestData) {
 	r := require.New(t)
 
 	annotation := make(map[string]string)
-	annotation[k8s.NPLEnabledAnnotationKey] = "true"
+	annotation[types.NPLEnabledAnnotationKey] = "true"
 	ipFamily := corev1.IPv4Protocol
 	data.createNginxClusterIPServiceWithAnnotations(false, &ipFamily, annotation)
 
@@ -659,7 +652,7 @@ func testNPLChangePortRangeAgentRestart(t *testing.T, data *TestData) {
 	r := require.New(t)
 
 	annotation := make(map[string]string)
-	annotation[k8s.NPLEnabledAnnotationKey] = "true"
+	annotation[types.NPLEnabledAnnotationKey] = "true"
 	ipFamily := corev1.IPv4Protocol
 	data.createNginxClusterIPServiceWithAnnotations(false, &ipFamily, annotation)
 
diff --git a/test/e2e/proxy_test.go b/test/e2e/proxy_test.go
index 4289b1558..db9ac8880 100644
--- a/test/e2e/proxy_test.go
+++ b/test/e2e/proxy_test.go
@@ -448,10 +448,7 @@ ip netns exec %[1]s ip link set dev %[1]s-a up && \
 ip netns exec %[1]s ip route replace default via %[3]s && \
 sleep 3600
 `, testNetns, "1.1.1.1", "1.1.1.254", 24)
-	if err := data.createPodOnNode(testPod, data.testNamespace, controlPlaneNodeName(), agnhostImage, []string{"sh", "-c", cmd}, nil, nil, nil, true, func(pod *corev1.Pod) {
-		privileged := true
-		pod.Spec.Containers[0].SecurityContext = &corev1.SecurityContext{Privileged: &privileged}
-	}); err != nil {
+	if err := NewPodBuilder(testPod, data.testNamespace, agnhostImage).OnNode(controlPlaneNodeName()).WithCommand([]string{"sh", "-c", cmd}).InHostNetwork().Privileged().Create(data); err != nil {
 		t.Fatalf("Failed to create client Pod: %v", err)
 	}
 	defer deletePodWrapper(t, data, data.testNamespace, testPod)
@@ -474,7 +471,7 @@ func createAgnhostPod(t *testing.T, data *TestData, podName string, node string,
 		},
 	}
 
-	require.NoError(t, data.createPodOnNode(podName, data.testNamespace, node, agnhostImage, []string{}, args, nil, ports, hostNetwork, nil))
+	require.NoError(t, NewPodBuilder(podName, data.testNamespace, agnhostImage).OnNode(node).WithArgs(args).WithPorts(ports).WithHostNetwork(hostNetwork).Create(data))
 	_, err := data.podWaitForIPs(defaultTimeout, podName, data.testNamespace)
 	require.NoError(t, err)
 	require.NoError(t, data.podWaitForRunning(defaultTimeout, podName, data.testNamespace))
diff --git a/test/e2e/secondary_network_ipam_test.go b/test/e2e/secondary_network_ipam_test.go
index 6bb634a36..86bc47f75 100644
--- a/test/e2e/secondary_network_ipam_test.go
+++ b/test/e2e/secondary_network_ipam_test.go
@@ -22,8 +22,7 @@ import (
 
 	"antrea.io/antrea/pkg/agent/config"
 	crdv1alpha2 "antrea.io/antrea/pkg/apis/crd/v1alpha2"
-	agentconfig "antrea.io/antrea/pkg/config/agent"
-	controllerconfig "antrea.io/antrea/pkg/config/controller"
+	"antrea.io/antrea/pkg/features"
 )
 
 var (
@@ -242,6 +241,7 @@ func TestSecondaryNetworkIPAM(t *testing.T) {
 	skipIfProxyDisabled(t)
 	skipIfNotIPv4Cluster(t)
 	skipIfAntreaIPAMTest(t)
+	skipIfFeatureDisabled(t, features.AntreaIPAM, true, true)
 
 	data, err := setupTest(t)
 	if err != nil {
@@ -250,16 +250,6 @@ func TestSecondaryNetworkIPAM(t *testing.T) {
 	defer teardownTest(t, data)
 	skipIfEncapModeIsNot(t, data, config.TrafficEncapModeEncap)
 
-	cc := func(config *controllerconfig.ControllerConfig) {
-		config.FeatureGates["AntreaIPAM"] = true
-	}
-	ac := func(config *agentconfig.AgentConfig) {
-		config.FeatureGates["AntreaIPAM"] = true
-	}
-	if err = data.mutateAntreaConfigMap(cc, ac, true, true); err != nil {
-		t.Fatalf("Failed to enable AntreaIPAM feature: %v", err)
-	}
-
 	_, err = data.crdClient.CrdV1alpha2().IPPools().Create(context.TODO(), testIPPoolv4, metav1.CreateOptions{})
 	defer deleteIPPoolWrapper(t, data, testIPPoolv4.Name)
 	if err != nil {
diff --git a/test/e2e/service_externalip_test.go b/test/e2e/service_externalip_test.go
index 9d1bc4f03..82e7e1920 100644
--- a/test/e2e/service_externalip_test.go
+++ b/test/e2e/service_externalip_test.go
@@ -37,8 +37,7 @@ import (
 
 	antreaagenttypes "antrea.io/antrea/pkg/agent/types"
 	"antrea.io/antrea/pkg/apis/crd/v1alpha2"
-	agentconfig "antrea.io/antrea/pkg/config/agent"
-	controllerconfig "antrea.io/antrea/pkg/config/controller"
+	"antrea.io/antrea/pkg/features"
 	"antrea.io/antrea/pkg/querier"
 )
 
@@ -46,6 +45,7 @@ func TestServiceExternalIP(t *testing.T) {
 	skipIfHasWindowsNodes(t)
 	skipIfNumNodesLessThan(t, 2)
 	skipIfAntreaIPAMTest(t)
+	skipIfFeatureDisabled(t, features.ServiceExternalIP, true, true)
 
 	data, err := setupTest(t)
 	if err != nil {
@@ -53,17 +53,6 @@ func TestServiceExternalIP(t *testing.T) {
 	}
 	defer teardownTest(t, data)
 
-	cc := func(config *controllerconfig.ControllerConfig) {
-		config.FeatureGates["ServiceExternalIP"] = true
-	}
-	ac := func(config *agentconfig.AgentConfig) {
-		config.FeatureGates["ServiceExternalIP"] = true
-	}
-
-	if err := data.mutateAntreaConfigMap(cc, ac, true, true); err != nil {
-		t.Fatalf("Failed to enable ServiceExternalIP feature: %v", err)
-	}
-
 	t.Run("testServiceWithExternalIPCRUD", func(t *testing.T) { testServiceWithExternalIPCRUD(t, data) })
 	t.Run("testServiceUpdateExternalIP", func(t *testing.T) { testServiceUpdateExternalIP(t, data) })
 	t.Run("testServiceExternalTrafficPolicyLocal", func(t *testing.T) { testServiceExternalTrafficPolicyLocal(t, data) })
@@ -656,9 +645,7 @@ sleep 3600`, tt.clientName, tt.clientIP, tt.localIP, tt.clientIPMaskLen)
 
 					baseUrl := net.JoinHostPort(externalIP, strconv.FormatInt(int64(port), 10))
 
-					require.NoError(t, data.createPodOnNode(tt.clientName, data.testNamespace, host, agnhostImage, []string{"sh", "-c", cmd}, nil, nil, nil, true, func(pod *v1.Pod) {
-						privileged := true
-						pod.Spec.Containers[0].SecurityContext = &v1.SecurityContext{Privileged: &privileged}
+					require.NoError(t, NewPodBuilder(tt.clientName, data.testNamespace, agnhostImage).OnNode(host).WithCommand([]string{"sh", "-c", cmd}).InHostNetwork().Privileged().WithMutateFunc(func(pod *v1.Pod) {
 						delete(pod.Labels, "app")
 						// curl will exit immediately if the destination IP is unreachable and will NOT retry despite having retry flags set.
 						// Use an exec readiness probe to ensure the route is configured to the interface.
@@ -673,7 +660,7 @@ sleep 3600`, tt.clientName, tt.clientIP, tt.localIP, tt.clientIPMaskLen)
 							InitialDelaySeconds: 1,
 							PeriodSeconds:       1,
 						}
-					}))
+					}).Create(data))
 
 					_, err = data.PodWaitFor(defaultTimeout, tt.clientName, data.testNamespace, func(p *v1.Pod) (bool, error) {
 						for _, condition := range p.Status.Conditions {
diff --git a/test/e2e/service_test.go b/test/e2e/service_test.go
index 267a6d732..4ea59e5b9 100644
--- a/test/e2e/service_test.go
+++ b/test/e2e/service_test.go
@@ -178,13 +178,13 @@ func (data *TestData) testNodePort(t *testing.T, isWindows bool, clientNamespace
 func (data *TestData) createAgnhostServiceAndBackendPods(t *testing.T, name, namespace string, node string, svcType corev1.ServiceType) (*corev1.Service, func()) {
 	ipProtocol := corev1.IPv4Protocol
 	args := []string{"netexec", "--http-port=80", "--udp-port=80"}
-	require.NoError(t, data.createPodOnNode(name, namespace, node, agnhostImage, []string{}, args, nil, []corev1.ContainerPort{
+	require.NoError(t, NewPodBuilder(name, namespace, agnhostImage).OnNode(node).WithArgs(args).WithPorts([]corev1.ContainerPort{
 		{
 			Name:          "http",
 			ContainerPort: 80,
 			Protocol:      corev1.ProtocolTCP,
 		},
-	}, false, nil))
+	}).Create(data))
 	podIPs, err := data.podWaitForIPs(defaultTimeout, name, namespace)
 	require.NoError(t, err)
 	t.Logf("Created service Pod IPs %v", podIPs.ipStrings)
diff --git a/test/e2e/traceflow_test.go b/test/e2e/traceflow_test.go
index 9c81a9770..c2e1fae58 100644
--- a/test/e2e/traceflow_test.go
+++ b/test/e2e/traceflow_test.go
@@ -19,7 +19,6 @@ import (
 	"fmt"
 	"net"
 	"reflect"
-	"strconv"
 	"strings"
 	"testing"
 	"time"
@@ -1140,10 +1139,7 @@ func testTraceflowInterNode(t *testing.T, data *TestData) {
 	// Create Service backend Pod. The "hairpin" testcases require the Service to have a single backend Pod,
 	// and no more, in order to be deterministic.
 	agnhostPodName := "agnhost"
-	mutateFunc := func(pod *corev1.Pod) {
-		pod.Labels["app"] = "agnhost-server"
-	}
-	require.NoError(t, data.createPodOnNode(agnhostPodName, data.testNamespace, node2, agnhostImage, []string{"sleep", strconv.Itoa(3600)}, nil, nil, nil, false, mutateFunc))
+	require.NoError(t, NewPodBuilder(agnhostPodName, data.testNamespace, agnhostImage).OnNode(node2).WithCommand([]string{"sleep", "3600"}).WithLabels(map[string]string{"app": "agnhost-server"}).Create(data))
 	agnhostIP, err := data.podWaitForIPs(defaultTimeout, agnhostPodName, data.testNamespace)
 	require.NoError(t, err)
 
diff --git a/test/e2e/trafficcontrol_test.go b/test/e2e/trafficcontrol_test.go
index fffd0b601..834464b71 100644
--- a/test/e2e/trafficcontrol_test.go
+++ b/test/e2e/trafficcontrol_test.go
@@ -28,7 +28,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	"antrea.io/antrea/pkg/apis/crd/v1alpha2"
-	agentconfig "antrea.io/antrea/pkg/config/agent"
+	"antrea.io/antrea/pkg/features"
 )
 
 type trafficControlTestConfig struct {
@@ -54,6 +54,7 @@ var (
 
 func TestTrafficControl(t *testing.T) {
 	skipIfHasWindowsNodes(t)
+	skipIfFeatureDisabled(t, features.TrafficControl, true, false)
 
 	data, err := setupTest(t)
 	if err != nil {
@@ -61,14 +62,6 @@ func TestTrafficControl(t *testing.T) {
 	}
 	defer teardownTest(t, data)
 
-	ac := func(config *agentconfig.AgentConfig) {
-		config.FeatureGates["TrafficControl"] = true
-	}
-
-	if err = data.mutateAntreaConfigMap(nil, ac, true, true); err != nil {
-		t.Fatalf("Failed to enable TrafficControl feature: %v", err)
-	}
-
 	tcTestConfig.nodeName = controlPlaneNodeName()
 
 	createTrafficControlTestPod(t, data, tcTestConfig.podName)
@@ -88,13 +81,8 @@ func createTrafficControlTestPod(t *testing.T, data *TestData, podName string) {
 			Protocol:      corev1.ProtocolTCP,
 		},
 	}
-	mutateLabels := func(pod *corev1.Pod) {
-		for k, v := range labels {
-			pod.Labels[k] = v
-		}
-	}
 
-	require.NoError(t, data.createPodOnNode(podName, data.testNamespace, tcTestConfig.nodeName, agnhostImage, []string{}, args, nil, ports, false, mutateLabels))
+	require.NoError(t, NewPodBuilder(podName, data.testNamespace, agnhostImage).OnNode(tcTestConfig.nodeName).WithArgs(args).WithPorts(ports).WithLabels(labels).Create(data))
 	ips, err := data.podWaitForIPs(defaultTimeout, podName, data.testNamespace)
 	if err != nil {
 		t.Fatalf("Error when waiting for IP for Pod '%s': %v", podName, err)
@@ -110,10 +98,7 @@ func createTrafficControlTestPod(t *testing.T, data *TestData, podName string) {
 }
 
 func createTrafficControlPacketsCollectorPod(t *testing.T, data *TestData, podName string) {
-	require.NoError(t, data.createPodOnNode(podName, data.testNamespace, tcTestConfig.nodeName, agnhostImage, []string{"sleep"}, []string{"3600"}, nil, nil, false, func(pod *corev1.Pod) {
-		privileged := true
-		pod.Spec.Containers[0].SecurityContext = &corev1.SecurityContext{Privileged: &privileged}
-	}))
+	require.NoError(t, NewPodBuilder(podName, data.testNamespace, agnhostImage).OnNode(tcTestConfig.nodeName).WithCommand([]string{"sleep", "3600"}).Privileged().Create(data))
 	ips, err := data.podWaitForIPs(defaultTimeout, podName, data.testNamespace)
 	if err != nil {
 		t.Fatalf("Error when waiting for IP for Pod '%s': %v", podName, err)
@@ -314,10 +299,7 @@ func testRedirectToLocal(t *testing.T, data *TestData) {
 ip link add dev %[1]s type veth peer name %[2]s && \
 ip link set dev %[1]s up && \
 ip link set dev %[2]s up`, targetPortName, returnPortName)
-	if err := data.createPodOnNode(tempPodName, data.testNamespace, tcTestConfig.nodeName, agnhostImage, []string{"sleep", "3600"}, nil, nil, nil, true, func(pod *corev1.Pod) {
-		privileged := true
-		pod.Spec.Containers[0].SecurityContext = &corev1.SecurityContext{Privileged: &privileged}
-	}); err != nil {
+	if err := NewPodBuilder(tempPodName, data.testNamespace, agnhostImage).OnNode(tcTestConfig.nodeName).WithCommand([]string{"sleep", "3600"}).InHostNetwork().Privileged().Create(data); err != nil {
 		t.Fatalf("Failed to create Pod %s: %v", tempPodName, err)
 	}
 	require.NoError(t, data.podWaitForRunning(defaultTimeout, tempPodName, data.testNamespace))
diff --git a/test/e2e/wireguard_test.go b/test/e2e/wireguard_test.go
index b6667ab7d..6be3dc1cd 100644
--- a/test/e2e/wireguard_test.go
+++ b/test/e2e/wireguard_test.go
@@ -17,7 +17,6 @@ package e2e
 import (
 	"fmt"
 	"net"
-	"strconv"
 	"strings"
 	"testing"
 
@@ -117,7 +116,7 @@ func testServiceConnectivity(t *testing.T, data *TestData) {
 	defer cleanup()
 
 	// Create the a hostNetwork Pod on a Node different from the service's backend Pod, so the service traffic will be transferred across the tunnel.
-	require.NoError(t, data.createPodOnNode(clientPodName, data.testNamespace, clientPodNode, busyboxImage, []string{"sleep", strconv.Itoa(3600)}, nil, nil, nil, true, nil))
+	require.NoError(t, NewPodBuilder(clientPodName, data.testNamespace, busyboxImage).OnNode(clientPodNode).WithCommand([]string{"sleep", "3600"}).InHostNetwork().Create(data))
 	defer data.deletePodAndWait(defaultTimeout, clientPodName, data.testNamespace)
 	require.NoError(t, data.podWaitForRunning(defaultTimeout, clientPodName, data.testNamespace))