Repository: antrea-io/antrea
Pull request #3365 (2022-02-27 06:27:36)

Update Antrea IPAM documentation for enableBridgingMode parameter
Signed-off-by: Jianjun Shen <shenj@vmware.com>

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

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

```diff
@@             Coverage Diff             @@
##             main    #3365       +/-   ##
===========================================
- Coverage   62.05%   42.13%   -19.92%     
===========================================
  Files         268      200       -68     
  Lines       26725    24150     -2575     
===========================================
- Hits        16583    10176     -6407     
- Misses       8314    12918     +4604     
+ Partials     1828     1056      -772     
```

| Flag | Coverage Δ | |
|---|---|---|
| kind-e2e-tests | `?` | |
| unit-tests | `42.13% <ø> (-0.01%)` | :arrow_down: |

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

| [Impacted Files](https://codecov.io/gh/antrea-io/antrea/pull/3365?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) | Coverage Δ | |
|---|---|---|
| [pkg/ovs/openflow/default.go](https://codecov.io/gh/antrea-io/antrea/pull/3365/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL292cy9vcGVuZmxvdy9kZWZhdWx0Lmdv) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [pkg/util/runtime/runtime.go](https://codecov.io/gh/antrea-io/antrea/pull/3365/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL3V0aWwvcnVudGltZS9ydW50aW1lLmdv) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [pkg/agent/openflow/groups.go](https://codecov.io/gh/antrea-io/antrea/pull/3365/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2FnZW50L29wZW5mbG93L2dyb3Vwcy5nbw==) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [pkg/agent/nodeportlocal/k8s/annotations.go](https://codecov.io/gh/antrea-io/antrea/pull/3365/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%> (-100.00%)` | :arrow_down: |
| [pkg/ovs/openflow/logs.go](https://codecov.io/gh/antrea-io/antrea/pull/3365/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL292cy9vcGVuZmxvdy9sb2dzLmdv) | `9.52% <0.00%> (-90.48%)` | :arrow_down: |
| [pkg/apis/controlplane/register.go](https://codecov.io/gh/antrea-io/antrea/pull/3365/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2FwaXMvY29udHJvbHBsYW5lL3JlZ2lzdGVyLmdv) | `0.00% <0.00%> (-90.00%)` | :arrow_down: |
| [pkg/agent/controller/networkpolicy/reject.go](https://codecov.io/gh/antrea-io/antrea/pull/3365/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2FnZW50L2NvbnRyb2xsZXIvbmV0d29ya3BvbGljeS9yZWplY3QuZ28=) | `0.00% <0.00%> (-84.68%)` | :arrow_down: |
| [pkg/agent/client.go](https://codecov.io/gh/antrea-io/antrea/pull/3365/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2FnZW50L2NsaWVudC5nbw==) | `0.00% <0.00%> (-77.42%)` | :arrow_down: |
| [pkg/ovs/ovsconfig/ovs\_client\_linux.go](https://codecov.io/gh/antrea-io/antrea/pull/3365/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL292cy9vdnNjb25maWcvb3ZzX2NsaWVudF9saW51eC5nbw==) | `0.00% <0.00%> (-76.93%)` | :arrow_down: |
| [pkg/flowaggregator/certificate.go](https://codecov.io/gh/antrea-io/antrea/pull/3365/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2Zsb3dhZ2dyZWdhdG9yL2NlcnRpZmljYXRlLmdv) | `0.00% <0.00%> (-76.58%)` | :arrow_down: |
| ... and [184 more](https://codecov.io/gh/antrea-io/antrea/pull/3365/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/benchmark.yml b/.github/workflows/benchmark.yml
index 44b549040..cd75247dd 100644
--- a/.github/workflows/benchmark.yml
+++ b/.github/workflows/benchmark.yml
@@ -13,7 +13,7 @@ jobs:
       - uses: actions/checkout@v2
         with:
           fetch-depth: 0
-      - uses: antrea-io/antrea/ci/gh-actions/has-changes@main
+      - uses: antrea-io/has-changes@v1
         id: check_diff
         with:
           args: pkg/*
diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml
index dd6fdc760..8fe705dc2 100644
--- a/.github/workflows/build.yml
+++ b/.github/workflows/build.yml
@@ -18,7 +18,7 @@ jobs:
     - uses: actions/checkout@v2
       with:
         fetch-depth: 0
-    - uses: antrea-io/antrea/ci/gh-actions/has-changes@main
+    - uses: antrea-io/has-changes@v1
       id: check_diff
       with:
         args: docs/* ci/jenkins/* *.md hack/.notableofcontents
@@ -34,7 +34,7 @@ jobs:
     - name: Build Antrea amd64 Docker image without pushing to registry
       if: ${{ github.repository != 'antrea-io/antrea' || github.event_name != 'push' || github.ref != 'refs/heads/main' }}
       run: |
-        ./hack/build-antrea-ubuntu-all.sh --pull
+        ./hack/build-antrea-linux-all.sh --pull
     - name: Build and push Antrea amd64 Docker image to registry
       if: ${{ github.repository == 'antrea-io/antrea' && github.event_name == 'push' && github.ref == 'refs/heads/main' }}
       env:
@@ -42,7 +42,7 @@ jobs:
         DOCKER_PASSWORD: ${{ secrets.DOCKER_PASSWORD }}
       run: |
         echo "$DOCKER_PASSWORD" | docker login -u "$DOCKER_USERNAME" --password-stdin
-        ./hack/build-antrea-ubuntu-all.sh --pull --push-base-images
+        ./hack/build-antrea-linux-all.sh --pull --push-base-images
         docker tag antrea/antrea-ubuntu:latest antrea/antrea-ubuntu-amd64:latest
         docker push antrea/antrea-ubuntu-amd64:latest
     - name: Trigger Antrea arm builds and multi-arch manifest update
@@ -55,6 +55,26 @@ jobs:
         token: ${{ secrets.ANTREA_BUILD_INFRA_WORKFLOW_DISPATCH_PAT }}
         inputs: ${{ format('{{ "antrea-repository":"antrea-io/antrea", "antrea-ref":"{0}", "docker-tag":"{1}" }}', github.ref, 'latest') }}
 
+  build-ubi:
+    needs: check-changes
+    if: ${{ needs.check-changes.outputs.has_changes == 'yes' || github.event_name == 'push' }}
+    runs-on: [ubuntu-latest]
+    steps:
+    - uses: actions/checkout@v2
+    - name: Build Antrea UBI8 Docker image without pushing to registry
+      if: ${{ github.repository != 'antrea-io/antrea' || github.event_name != 'push' || github.ref != 'refs/heads/main' }}
+      run: |
+        ./hack/build-antrea-linux-all.sh --pull --distro ubi
+    - name: Build and push Antrea UBI8 Docker image to registry
+      if: ${{ github.repository == 'antrea-io/antrea' && github.event_name == 'push' && github.ref == 'refs/heads/main' }}
+      env:
+        DOCKER_USERNAME: ${{ secrets.DOCKER_USERNAME }}
+        DOCKER_PASSWORD: ${{ secrets.DOCKER_PASSWORD }}
+      run: |
+        echo "$DOCKER_PASSWORD" | docker login -u "$DOCKER_USERNAME" --password-stdin
+        ./hack/build-antrea-linux-all.sh --pull --push-base-images --distro ubi
+        docker push antrea/antrea-ubi:latest
+
   build-scale:
     needs: check-changes
     if: ${{ needs.check-changes.outputs.has_changes == 'yes' || github.event_name == 'push' }}
diff --git a/.github/workflows/build_tag.yml b/.github/workflows/build_tag.yml
index 0128ed2de..5bce4c495 100644
--- a/.github/workflows/build_tag.yml
+++ b/.github/workflows/build_tag.yml
@@ -24,13 +24,13 @@ jobs:
     needs: get-version
     steps:
     - uses: actions/checkout@v2
-    - name: Build and push Antrea amd64 Docker image to registry
+    - name: Build and push Antrea Ubuntu amd64 Docker image to registry
       env:
         DOCKER_USERNAME: ${{ secrets.DOCKER_USERNAME }}
         DOCKER_PASSWORD: ${{ secrets.DOCKER_PASSWORD }}
         VERSION: ${{ needs.get-version.outputs.version }}
       run: |
-        ./hack/build-antrea-ubuntu-all.sh --pull
+        ./hack/build-antrea-linux-all.sh --pull
         echo "$DOCKER_PASSWORD" | docker login -u "$DOCKER_USERNAME" --password-stdin
         docker tag antrea/antrea-ubuntu:"${VERSION}" antrea/antrea-ubuntu-amd64:"${VERSION}"
         docker push antrea/antrea-ubuntu-amd64:"${VERSION}"
@@ -43,6 +43,21 @@ jobs:
         token: ${{ secrets.ANTREA_BUILD_INFRA_WORKFLOW_DISPATCH_PAT }}
         inputs: ${{ format('{{ "antrea-repository":"antrea-io/antrea", "antrea-ref":"{0}", "docker-tag":"{1}" }}', github.ref, needs.get-version.outputs.version) }}
 
+  build-ubi:
+    runs-on: [ubuntu-latest]
+    needs: get-version
+    steps:
+      - uses: actions/checkout@v2
+      - name: Build and push Antrea UBI8 amd64 Docker image to registry
+        env:
+          DOCKER_USERNAME: ${{ secrets.DOCKER_USERNAME }}
+          DOCKER_PASSWORD: ${{ secrets.DOCKER_PASSWORD }}
+          VERSION: ${{ needs.get-version.outputs.version }}
+        run: |
+          ./hack/build-antrea-linux-all.sh --pull --distro ubi
+          echo "$DOCKER_PASSWORD" | docker login -u "$DOCKER_USERNAME" --password-stdin
+          docker push antrea/antrea-ubi:"${VERSION}"
+
   build-windows:
     runs-on: [windows-2019]
     needs: get-version
diff --git a/.github/workflows/golicense.yml b/.github/workflows/golicense.yml
index 8dd23bc66..3ef6cef7b 100644
--- a/.github/workflows/golicense.yml
+++ b/.github/workflows/golicense.yml
@@ -20,7 +20,7 @@ jobs:
     - uses: actions/checkout@v2
       with:
         fetch-depth: 0
-    - uses: antrea-io/antrea/ci/gh-actions/has-changes@main
+    - uses: antrea-io/has-changes@v1
       id: check_diff
       with:
         args: docs/* ci/jenkins/* *.md hack/.notableofcontents
diff --git a/.github/workflows/kind.yml b/.github/workflows/kind.yml
index 4111da0bd..53f5ecfaf 100755
--- a/.github/workflows/kind.yml
+++ b/.github/workflows/kind.yml
@@ -20,7 +20,7 @@ jobs:
     - uses: actions/checkout@v2
       with:
         fetch-depth: 0
-    - uses: antrea-io/antrea/ci/gh-actions/has-changes@main
+    - uses: antrea-io/has-changes@v1
       id: check_diff
       with:
         args: docs/* ci/jenkins/* *.md hack/.notableofcontents
@@ -36,7 +36,7 @@ jobs:
     - uses: actions/checkout@v2
     - name: Build Antrea Docker image with code coverage support
       run: |
-        ./hack/build-antrea-ubuntu-all.sh --pull --coverage
+        ./hack/build-antrea-linux-all.sh --pull --coverage
     - name: Save Antrea image to tarball
       run:  docker save -o antrea-ubuntu.tar antrea/antrea-ubuntu-coverage:latest
     - name: Upload Antrea image for subsequent jobs
diff --git a/.github/workflows/kind_upgrade.yml b/.github/workflows/kind_upgrade.yml
index 08646cd54..72f027f30 100644
--- a/.github/workflows/kind_upgrade.yml
+++ b/.github/workflows/kind_upgrade.yml
@@ -20,7 +20,7 @@ jobs:
     - uses: actions/checkout@v2
       with:
         fetch-depth: 0
-    - uses: antrea-io/antrea/ci/gh-actions/has-changes@main
+    - uses: antrea-io/has-changes@v1
       id: check_diff
       with:
         args: docs/* ci/jenkins/* *.md hack/.notableofcontents
@@ -36,7 +36,7 @@ jobs:
     - uses: actions/checkout@v2
     - name: Build Antrea Docker image
       run: |
-        ./hack/build-antrea-ubuntu-all.sh --pull
+        ./hack/build-antrea-linux-all.sh --pull
     - name: Save Antrea image to tarball
       run:  docker save -o antrea-ubuntu.tar projects.registry.vmware.com/antrea/antrea-ubuntu:latest
     - name: Upload Antrea image for subsequent jobs
diff --git a/Makefile b/Makefile
index adcf330ae..762a65532 100644
--- a/Makefile
+++ b/Makefile
@@ -66,11 +66,6 @@ antrea-cni:
 	@mkdir -p $(BINDIR)
 	GOOS=linux CGO_ENABLED=0 $(GO) build -o $(BINDIR) $(GOFLAGS) -ldflags '$(LDFLAGS)' antrea.io/antrea/cmd/antrea-cni
 
-.PHONY: antctl-ubuntu
-antctl-ubuntu:
-	@mkdir -p $(BINDIR)
-	GOOS=linux $(GO) build -o $(BINDIR) $(GOFLAGS) -ldflags '$(LDFLAGS)' antrea.io/antrea/cmd/antctl
-
 .PHONY: antctl-instr-binary
 antctl-instr-binary:
 	@mkdir -p $(BINDIR)
@@ -307,6 +302,19 @@ endif
 	docker tag antrea/antrea-ubuntu:$(DOCKER_IMG_VERSION) projects.registry.vmware.com/antrea/antrea-ubuntu
 	docker tag antrea/antrea-ubuntu:$(DOCKER_IMG_VERSION) projects.registry.vmware.com/antrea/antrea-ubuntu:$(DOCKER_IMG_VERSION)
 
+# Build bins in a golang container, and build the antrea-ubi Docker image.
+.PHONY: build-ubi
+build-ubi:
+	@echo "===> Building Antrea bins and antrea/antrea-ubi Docker image <==="
+ifneq ($(NO_PULL),"")
+	docker build -t antrea/antrea-ubi:$(DOCKER_IMG_VERSION) -f build/images/Dockerfile.build.ubi $(DOCKER_BUILD_ARGS) .
+else
+	docker build --pull -t antrea/antrea-ubi:$(DOCKER_IMG_VERSION) -f build/images/Dockerfile.build.ubi $(DOCKER_BUILD_ARGS) .
+endif
+	docker tag antrea/antrea-ubi:$(DOCKER_IMG_VERSION) antrea/antrea-ubi
+	docker tag antrea/antrea-ubi:$(DOCKER_IMG_VERSION) projects.registry.vmware.com/antrea/antrea-ubi
+	docker tag antrea/antrea-ubi:$(DOCKER_IMG_VERSION) projects.registry.vmware.com/antrea/antrea-ubi:$(DOCKER_IMG_VERSION)
+
 .PHONY: build-windows
 build-windows:
 	@echo "===> Building Antrea bins and antrea/antrea-windows Docker image <==="
diff --git a/build/images/Dockerfile.build.coverage b/build/images/Dockerfile.build.coverage
index 6e880dca6..dd9869bfb 100644
--- a/build/images/Dockerfile.build.coverage
+++ b/build/images/Dockerfile.build.coverage
@@ -10,7 +10,8 @@ RUN go mod download
 
 COPY . /antrea
 
-RUN make antrea-agent antrea-controller antrea-cni antctl-ubuntu antrea-controller-instr-binary antrea-agent-instr-binary antctl-instr-binary
+RUN make antrea-agent antrea-controller antrea-cni antctl-linux antrea-controller-instr-binary antrea-agent-instr-binary antctl-instr-binary
+RUN mv bin/antctl-linux bin/antctl
 
 FROM antrea/base-ubuntu:${OVS_VERSION}
 
diff --git a/build/images/Dockerfile.build.ubi b/build/images/Dockerfile.build.ubi
new file mode 100644
index 000000000..29cbb8132
--- /dev/null
+++ b/build/images/Dockerfile.build.ubi
@@ -0,0 +1,24 @@
+ARG GO_VERSION
+ARG OVS_VERSION
+FROM golang:${GO_VERSION} as antrea-build
+
+WORKDIR /antrea
+
+COPY go.mod /antrea/go.mod
+
+RUN go mod download
+
+COPY . /antrea
+
+RUN make antrea-agent antrea-controller antrea-cni antctl-linux
+RUN mv bin/antctl-linux bin/antctl
+
+FROM antrea/base-ubi:${OVS_VERSION}
+
+LABEL maintainer="Antrea <projectantrea-dev@googlegroups.com>"
+LABEL description="The Docker image to deploy the Antrea CNI. "
+
+USER root
+
+COPY build/images/scripts/* /usr/local/bin/
+COPY --from=antrea-build /antrea/bin/* /usr/local/bin/
diff --git a/build/images/Dockerfile.build.ubuntu b/build/images/Dockerfile.build.ubuntu
index 5a03d1eea..cb25d626e 100644
--- a/build/images/Dockerfile.build.ubuntu
+++ b/build/images/Dockerfile.build.ubuntu
@@ -10,7 +10,8 @@ RUN go mod download
 
 COPY . /antrea
 
-RUN make antrea-agent antrea-controller antrea-cni antctl-ubuntu
+RUN make antrea-agent antrea-controller antrea-cni antctl-linux
+RUN mv bin/antctl-linux bin/antctl
 
 FROM antrea/base-ubuntu:${OVS_VERSION}
 
diff --git a/build/images/base/Dockerfile b/build/images/base/Dockerfile
index 34404822f..a6d1e375d 100644
--- a/build/images/base/Dockerfile
+++ b/build/images/base/Dockerfile
@@ -26,7 +26,7 @@ RUN set -eux; \
 FROM antrea/openvswitch:${OVS_VERSION}
 
 LABEL maintainer="Antrea <projectantrea-dev@googlegroups.com>"
-LABEL description="Takes care of building the Antrea binaries as part of building the image."
+LABEL description="An Ubuntu based Docker base image for Antrea."
 
 USER root
 
diff --git a/build/images/base/Dockerfile.ubi b/build/images/base/Dockerfile.ubi
new file mode 100644
index 000000000..3d5f86385
--- /dev/null
+++ b/build/images/base/Dockerfile.ubi
@@ -0,0 +1,35 @@
+ARG OVS_VERSION
+FROM ubuntu:20.04 as cni-binaries
+
+ARG CNI_BINARIES_VERSION
+ARG WHEREABOUTS_VERSION=v0.4.2
+
+RUN apt-get update && \
+    apt-get install -y --no-install-recommends wget ca-certificates
+
+# Leading dot is required for the tar command below
+ENV CNI_PLUGINS="./host-local ./loopback ./portmap ./bandwidth"
+
+# Download containernetworking plugin binaries for the correct architecture
+RUN set -eux; \
+    dpkgArch="$(dpkg --print-architecture)"; \
+    case "${dpkgArch##*-}" in \
+         amd64) pluginsArch='amd64' ;; \
+	 armhf) pluginsArch='arm' ;; \
+	 arm64) pluginsArch='arm64' ;; \
+         *) pluginsArch=''; echo >&2; echo >&2 "unsupported architecture '$dpkgArch'"; echo >&2 ; exit 1 ;; \
+    esac; \
+    mkdir -p /opt/cni/bin; \
+    wget -q -O - https://github.com/containernetworking/plugins/releases/download/$CNI_BINARIES_VERSION/cni-plugins-linux-${pluginsArch}-$CNI_BINARIES_VERSION.tgz | tar xz -C /opt/cni/bin $CNI_PLUGINS; \
+    wget -q -O - https://downloads.antrea.io/whereabouts/$WHEREABOUTS_VERSION/whereabouts-linux-${pluginsArch}.tgz | tar xz -C /opt/cni/bin/ whereabouts-linux-${pluginsArch}/whereabouts --strip-components=1 --no-same-owner
+
+FROM antrea/openvswitch-ubi:${OVS_VERSION}
+
+LABEL maintainer="Antrea <projectantrea-dev@googlegroups.com>"
+LABEL description="An UBI8 based Docker base image for Antrea."
+
+USER root
+
+RUN yum install ipset jq -y && yum clean all 
+
+COPY --from=cni-binaries /opt/cni/bin /opt/cni/bin
diff --git a/build/images/base/build.sh b/build/images/base/build.sh
index 54057b193..1348242b1 100755
--- a/build/images/base/build.sh
+++ b/build/images/base/build.sh
@@ -23,11 +23,12 @@ function echoerr {
     >&2 echo "$@"
 }
 
-_usage="Usage: $0 [--pull] [--push] [--platform <PLATFORM>]
+_usage="Usage: $0 [--pull] [--push] [--platform <PLATFORM>] [--distro [ubuntu|ubi]]
 Build the antrea/base-ubuntu:<OVS_VERSION> image.
         --pull                  Always attempt to pull a newer version of the base images
         --push                  Push the built image to the registry
-        --platform <PLATFORM>   Target platform for the image if server is multi-platform capable"
+        --platform <PLATFORM>   Target platform for the image if server is multi-platform capable
+        --distro <distro>       Target Linux distribution"
 
 function print_usage {
     echoerr "$_usage"
@@ -36,6 +37,7 @@ function print_usage {
 PULL=false
 PUSH=false
 PLATFORM=""
+DISTRO="ubuntu"
 
 while [[ $# -gt 0 ]]
 do
@@ -54,6 +56,10 @@ case $key in
     PLATFORM="$2"
     shift 2
     ;;
+    --distro)
+    DISTRO="$2"
+    shift 2
+    ;;
     -h|--help)
     print_usage
     exit 0
@@ -75,6 +81,11 @@ if [ "$PLATFORM" != "" ]; then
     PLATFORM_ARG="--platform $PLATFORM"
 fi
 
+if [ "$DISTRO" != "ubuntu" ] && [ "$DISTRO" != "ubi" ]; then
+    echoerr "Invalid distribution $DISTRO"
+    exit 1
+fi
+
 THIS_DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" >/dev/null 2>&1 && pwd )"
 
 pushd $THIS_DIR > /dev/null
@@ -89,11 +100,20 @@ if $PULL; then
         docker pull ${DOCKER_REGISTRY}/antrea/ubuntu:20.04
         docker tag ${DOCKER_REGISTRY}/antrea/ubuntu:20.04 ubuntu:20.04
     fi
-    IMAGES_LIST=(
-        "antrea/openvswitch:$OVS_VERSION"
-        "antrea/cni-binaries:$CNI_BINARIES_VERSION"
-        "antrea/base-ubuntu:$OVS_VERSION"
-    )
+
+    if [ "$DISTRO" == "ubuntu" ]; then
+        IMAGES_LIST=(
+            "antrea/openvswitch:$OVS_VERSION"
+            "antrea/cni-binaries:$CNI_BINARIES_VERSION"
+            "antrea/base-ubuntu:$OVS_VERSION"
+        )
+    elif [ "$DISTRO" == "ubi" ]; then
+        IMAGES_LIST=(
+            "antrea/openvswitch-ubi:$OVS_VERSION"
+            "antrea/cni-binaries:$CNI_BINARIES_VERSION"
+            "antrea/base-ubi:$OVS_VERSION"
+        )
+    fi
     for image in "${IMAGES_LIST[@]}"; do
         if [[ ${DOCKER_REGISTRY} == "" ]]; then
             docker pull $PLATFORM_ARG "${image}" || true
@@ -113,16 +133,26 @@ docker build $PLATFORM_ARG --target cni-binaries \
        --build-arg CNI_BINARIES_VERSION=$CNI_BINARIES_VERSION \
        --build-arg OVS_VERSION=$OVS_VERSION .
 
-docker build $PLATFORM_ARG \
-       --cache-from antrea/cni-binaries:$CNI_BINARIES_VERSION \
-       --cache-from antrea/base-ubuntu:$OVS_VERSION \
-       -t antrea/base-ubuntu:$OVS_VERSION \
-       --build-arg CNI_BINARIES_VERSION=$CNI_BINARIES_VERSION \
-       --build-arg OVS_VERSION=$OVS_VERSION .
+if [ "$DISTRO" == "ubuntu" ]; then
+    docker build $PLATFORM_ARG \
+           --cache-from antrea/cni-binaries:$CNI_BINARIES_VERSION \
+           --cache-from antrea/base-ubuntu:$OVS_VERSION \
+           -t antrea/base-ubuntu:$OVS_VERSION \
+           --build-arg CNI_BINARIES_VERSION=$CNI_BINARIES_VERSION \
+           --build-arg OVS_VERSION=$OVS_VERSION .
+elif [ "$DISTRO" == "ubi" ]; then
+    docker build $PLATFORM_ARG \
+           --cache-from antrea/cni-binaries:$CNI_BINARIES_VERSION \
+           --cache-from antrea/base-ubuntu:$OVS_VERSION \
+           -t antrea/base-ubi:$OVS_VERSION \
+           -f Dockerfile.ubi \
+           --build-arg CNI_BINARIES_VERSION=$CNI_BINARIES_VERSION \
+           --build-arg OVS_VERSION=$OVS_VERSION .
+fi
 
 if $PUSH; then
     docker push antrea/cni-binaries:$CNI_BINARIES_VERSION
-    docker push antrea/base-ubuntu:$OVS_VERSION
+    docker push antrea/base-$DISTRO:$OVS_VERSION
 fi
 
 popd > /dev/null
diff --git a/build/images/flow-aggregator/Dockerfile b/build/images/flow-aggregator/Dockerfile
index 37fa899da..7140a9695 100644
--- a/build/images/flow-aggregator/Dockerfile
+++ b/build/images/flow-aggregator/Dockerfile
@@ -5,7 +5,8 @@ WORKDIR /antrea
 
 COPY . /antrea
 
-RUN make flow-aggregator antctl-ubuntu
+RUN make flow-aggregator antctl-linux
+RUN mv bin/antctl-linux bin/antctl
 
 # Chose this base image so that a shell is available for users to exec into the container, run antctl and run tools like pprof easily
 FROM ubuntu:20.04
diff --git a/build/images/flow-aggregator/Dockerfile.coverage b/build/images/flow-aggregator/Dockerfile.coverage
index 90f97fb7b..ce53c8901 100644
--- a/build/images/flow-aggregator/Dockerfile.coverage
+++ b/build/images/flow-aggregator/Dockerfile.coverage
@@ -5,7 +5,8 @@ WORKDIR /antrea
 
 COPY . /antrea
 
-RUN make flow-aggregator antctl-ubuntu flow-aggregator-instr-binary antctl-instr-binary
+RUN make flow-aggregator antctl-linux flow-aggregator-instr-binary antctl-instr-binary
+RUN mv bin/antctl-linux bin/antctl
 
 FROM ubuntu:20.04
 
diff --git a/build/images/ovs/CentOS.repo b/build/images/ovs/CentOS.repo
new file mode 100644
index 000000000..9eb710e79
--- /dev/null
+++ b/build/images/ovs/CentOS.repo
@@ -0,0 +1,23 @@
+[AppStream]
+name=CentOS-8-stream - AppStream
+mirrorlist=http://mirrorlist.centos.org/?release=8-stream&arch=$basearch&repo=AppStream&infra=$infra
+#baseurl=http://mirror.centos.org/$contentdir/8-stream/AppStream/$basearch/os/
+gpgcheck=1
+enabled=1
+gpgkey=file:///etc/pki/rpm-gpg/RPM-GPG-KEY-centosofficial
+
+[BaseOS]
+name=CentOS-8-stream - Base
+mirrorlist=http://mirrorlist.centos.org/?release=8-stream&arch=$basearch&repo=BaseOS&infra=$infra
+#baseurl=http://mirror.centos.org/$contentdir/8-stream/BaseOS/$basearch/os/
+gpgcheck=1
+enabled=1
+gpgkey=file:///etc/pki/rpm-gpg/RPM-GPG-KEY-centosofficial
+
+[extras]
+name=CentOS-8-stream - Extras
+mirrorlist=http://mirrorlist.centos.org/?release=8-stream&arch=$basearch&repo=extras&infra=$infra
+#baseurl=http://mirror.centos.org/$contentdir/8-stream/extras/$basearch/os/
+gpgcheck=1
+enabled=1
+gpgkey=file:///etc/pki/rpm-gpg/RPM-GPG-KEY-centosofficial
diff --git a/build/images/ovs/Dockerfile.ubi b/build/images/ovs/Dockerfile.ubi
new file mode 100644
index 000000000..12eed251b
--- /dev/null
+++ b/build/images/ovs/Dockerfile.ubi
@@ -0,0 +1,46 @@
+# OVS build scripts are only applicable for RHEL 7.x:
+# https://docs.openvswitch.org/en/latest/intro/install/fedora
+FROM centos:centos7 as ovs-rpms
+
+# Some patches may not apply cleanly if a non-default version is provided.
+# See build/images/deps/ovs-version for the default version.
+ARG OVS_VERSION
+
+# Install RPM tools and generic build dependencies.
+RUN yum update -y && yum install wget git yum-utils python38 rpm-build epel-release -y
+
+COPY apply-patches.sh /
+
+# Download OVS source code
+RUN wget -q -O - https://www.openvswitch.org/releases/openvswitch-$OVS_VERSION.tar.gz  | tar xz -C /tmp
+RUN cd /tmp/openvswitch* && \
+    /apply-patches.sh && \
+    sed -e "s/@VERSION@/$OVS_VERSION/" rhel/openvswitch-fedora.spec.in > /tmp/ovs.spec && \
+    yum-builddep -y /tmp/ovs.spec && ./boot.sh && \
+    ./configure --prefix=/usr --localstatedir=/var --sysconfdir=/etc && \
+    make rpm-fedora && mkdir -p /tmp/ovs-rpms && \
+    mv /tmp/openvswitch-$OVS_VERSION/rpm/rpmbuild/RPMS/*/*.rpm  /tmp/ovs-rpms && \
+    rm -rf /tmp/openvswitch*
+
+
+FROM registry.access.redhat.com/ubi8
+
+LABEL maintainer="Antrea <projectantrea-dev@googlegroups.com>"
+LABEL description="A Docker image based on UBI8 which includes Open vSwitch built from source."
+
+# Change Repository from UBI8’s to CentOS because UBI8's repository does not contain
+# enough packages required by OVS installation.
+# Using the official RHEL repository would be the best choice but it's not publicly accessible.
+# TODO: update the strongSwan logging config.
+COPY CentOS.repo /tmp/CentOS.repo
+COPY charon-logging.conf /tmp
+COPY --from=ovs-rpms /tmp/ovs-rpms/* /tmp/ovs-rpms/
+RUN rm -f /etc/yum.repos.d/* && mv /tmp/CentOS.repo /etc/yum.repos.d/CentOS.repo && \
+    curl https://www.centos.org/keys/RPM-GPG-KEY-CentOS-Official -o /etc/pki/rpm-gpg/RPM-GPG-KEY-centosofficial && \
+    subscription-manager config --rhsm.manage_repos=0 && \
+    yum clean all -y && yum reinstall yum -y && \
+    yum install /tmp/ovs-rpms/* -y && yum install epel-release -y && \
+    yum install iptables logrotate strongswan -y && \
+    mv /etc/logrotate.d/openvswitch /etc/logrotate.d/openvswitch-switch && \
+    sed -i "/rotate /a\    #size 100M" /etc/logrotate.d/openvswitch-switch && \
+    rm -rf /tmp/* && yum clean all
diff --git a/build/images/ovs/build.sh b/build/images/ovs/build.sh
index 7421515c2..3e09006db 100755
--- a/build/images/ovs/build.sh
+++ b/build/images/ovs/build.sh
@@ -23,11 +23,12 @@ function echoerr {
     >&2 echo "$@"
 }
 
-_usage="Usage: $0 [--pull] [--push] [--platform <PLATFORM>]
-Build the antrea/ovs:<VERSION> image.
+_usage="Usage: $0 [--pull] [--push] [--platform <PLATFORM>] [--distro [ubuntu|ubi]]
+Build the antrea/base-ubuntu:<OVS_VERSION> image.
         --pull                  Always attempt to pull a newer version of the base images
         --push                  Push the built image to the registry
-        --platform <PLATFORM>   Target platform for the image if server is multi-platform capable"
+        --platform <PLATFORM>   Target platform for the image if server is multi-platform capable
+        --distro <distro>       Target Linux distribution"
 
 function print_usage {
     echoerr "$_usage"
@@ -36,6 +37,7 @@ function print_usage {
 PULL=false
 PUSH=false
 PLATFORM=""
+DISTRO="ubuntu"
 
 while [[ $# -gt 0 ]]
 do
@@ -54,6 +56,10 @@ case $key in
     PLATFORM="$2"
     shift 2
     ;;
+    --distro)
+    DISTRO="$2"
+    shift 2
+    ;;
     -h|--help)
     print_usage
     exit 0
@@ -75,6 +81,11 @@ if [ "$PLATFORM" != "" ]; then
     PLATFORM_ARG="--platform $PLATFORM"
 fi
 
+if [ "$DISTRO" != "ubuntu" ] && [ "$DISTRO" != "ubi" ]; then
+    echoerr "Invalid distribution $DISTRO"
+    exit 1
+fi
+
 THIS_DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" >/dev/null 2>&1 && pwd )"
 
 pushd $THIS_DIR > /dev/null
@@ -95,10 +106,17 @@ if $PULL; then
         docker pull ${DOCKER_REGISTRY}/antrea/ubuntu:20.04
         docker tag ${DOCKER_REGISTRY}/antrea/ubuntu:20.04 ubuntu:20.04
     fi
-    IMAGES_LIST=(
-        "antrea/openvswitch-debs:$OVS_VERSION"
-        "antrea/openvswitch:$OVS_VERSION"
-    )
+    if [ "$DISTRO" == "ubuntu" ]; then
+        IMAGES_LIST=(
+            "antrea/openvswitch-debs:$OVS_VERSION"
+            "antrea/openvswitch:$OVS_VERSION"
+        )
+    elif [ "$DISTRO" == "ubi" ]; then
+        IMAGES_LIST=(
+            "antrea/openvswitch-rpms:$OVS_VERSION"
+            "antrea/openvswitch-ubi:$OVS_VERSION"
+        )
+    fi
     for image in "${IMAGES_LIST[@]}"; do
         if [[ ${DOCKER_REGISTRY} == "" ]]; then
             docker pull $PLATFORM_ARG "${image}" || true
@@ -112,20 +130,40 @@ if $PULL; then
     done
 fi
 
-docker build $PLATFORM_ARG --target ovs-debs \
-       --cache-from antrea/openvswitch-debs:$OVS_VERSION \
-       -t antrea/openvswitch-debs:$OVS_VERSION \
-       --build-arg OVS_VERSION=$OVS_VERSION .
-
-docker build $PLATFORM_ARG \
-       --cache-from antrea/openvswitch-debs:$OVS_VERSION \
-       --cache-from antrea/openvswitch:$OVS_VERSION \
-       -t antrea/openvswitch:$OVS_VERSION \
-       --build-arg OVS_VERSION=$OVS_VERSION .
+if [ "$DISTRO" == "ubuntu" ]; then
+    docker build $PLATFORM_ARG --target ovs-debs \
+           --cache-from antrea/openvswitch-debs:$OVS_VERSION \
+           -t antrea/openvswitch-debs:$OVS_VERSION \
+           --build-arg OVS_VERSION=$OVS_VERSION .
+
+    docker build $PLATFORM_ARG \
+           --cache-from antrea/openvswitch-debs:$OVS_VERSION \
+           --cache-from antrea/openvswitch:$OVS_VERSION \
+           -t antrea/openvswitch:$OVS_VERSION \
+           --build-arg OVS_VERSION=$OVS_VERSION .
+elif [ "$DISTRO" == "ubi" ]; then
+    docker build $PLATFORM_ARG --target ovs-rpms \
+           --cache-from antrea/openvswitch-rpms:$OVS_VERSION \
+           -t antrea/openvswitch-rpms:$OVS_VERSION \
+           --build-arg OVS_VERSION=$OVS_VERSION \
+           -f Dockerfile.ubi .
+
+    docker build \
+           --cache-from antrea/openvswitch-rpms:$OVS_VERSION \
+           --cache-from antrea/openvswitch-ubi:$OVS_VERSION \
+           -t antrea/openvswitch-ubi:$OVS_VERSION \
+           --build-arg OVS_VERSION=$OVS_VERSION \
+           -f Dockerfile.ubi .
+fi
 
 if $PUSH; then
-    docker push antrea/openvswitch-debs:$OVS_VERSION
-    docker push antrea/openvswitch:$OVS_VERSION
+    if [ "$DISTRO" == "ubuntu" ]; then
+        docker push antrea/openvswitch-debs:$OVS_VERSION
+        docker push antrea/openvswitch:$OVS_VERSION
+    elif [ "$DISTRO" == "ubi" ]; then
+        docker push antrea/openvswitch-rpms:$OVS_VERSION
+        docker push antrea/openvswitch-ubi:$OVS_VERSION
+    fi
 fi
 
 popd > /dev/null
diff --git a/build/yamls/antrea-aks.yml b/build/yamls/antrea-aks.yml
index b6d3a14b5..10695ecc8 100644
--- a/build/yamls/antrea-aks.yml
+++ b/build/yamls/antrea-aks.yml
@@ -3567,6 +3567,7 @@ webhooks:
     apiVersions:
     - v1alpha2
     operations:
+    - CREATE
     - UPDATE
     - DELETE
     resources:
diff --git a/build/yamls/antrea-eks.yml b/build/yamls/antrea-eks.yml
index 67f8eae63..600d1ac7d 100644
--- a/build/yamls/antrea-eks.yml
+++ b/build/yamls/antrea-eks.yml
@@ -3569,6 +3569,7 @@ webhooks:
     apiVersions:
     - v1alpha2
     operations:
+    - CREATE
     - UPDATE
     - DELETE
     resources:
diff --git a/build/yamls/antrea-gke.yml b/build/yamls/antrea-gke.yml
index d1a948295..d5a3fc74e 100644
--- a/build/yamls/antrea-gke.yml
+++ b/build/yamls/antrea-gke.yml
@@ -3567,6 +3567,7 @@ webhooks:
     apiVersions:
     - v1alpha2
     operations:
+    - CREATE
     - UPDATE
     - DELETE
     resources:
diff --git a/build/yamls/antrea-ipsec.yml b/build/yamls/antrea-ipsec.yml
index e5aed0897..69122ca26 100644
--- a/build/yamls/antrea-ipsec.yml
+++ b/build/yamls/antrea-ipsec.yml
@@ -3616,6 +3616,7 @@ webhooks:
     apiVersions:
     - v1alpha2
     operations:
+    - CREATE
     - UPDATE
     - DELETE
     resources:
diff --git a/build/yamls/antrea-kind.yml b/build/yamls/antrea-kind.yml
index c1e317884..1b7ba036a 100644
--- a/build/yamls/antrea-kind.yml
+++ b/build/yamls/antrea-kind.yml
@@ -3568,6 +3568,7 @@ webhooks:
     apiVersions:
     - v1alpha2
     operations:
+    - CREATE
     - UPDATE
     - DELETE
     resources:
diff --git a/build/yamls/antrea-windows.yml b/build/yamls/antrea-windows.yml
index 09e87dad5..100e0c778 100644
--- a/build/yamls/antrea-windows.yml
+++ b/build/yamls/antrea-windows.yml
@@ -23,6 +23,11 @@ data:
     # Service traffic.
     #  AntreaProxy: true
 
+    # Enable EndpointSlice support in AntreaProxy. Don't enable this feature unless that EndpointSlice
+    # API version v1beta1 is supported and set as enabled in Kubernetes. If AntreaProxy is not enabled,
+    # this flag will not take effect.
+    #  EndpointSlice: false
+
     # Enable flowexporter which exports polled conntrack connections as IPFIX flow records from each agent to a configured collector.
     #  FlowExporter: false
 
@@ -144,7 +149,7 @@ kind: ConfigMap
 metadata:
   labels:
     app: antrea
-  name: antrea-windows-config-25k48fh55f
+  name: antrea-windows-config-mf82kffb45
   namespace: kube-system
 ---
 apiVersion: apps/v1
@@ -232,7 +237,7 @@ spec:
         operator: Exists
       volumes:
       - configMap:
-          name: antrea-windows-config-25k48fh55f
+          name: antrea-windows-config-mf82kffb45
         name: antrea-windows-config
       - configMap:
           defaultMode: 420
diff --git a/build/yamls/antrea.yml b/build/yamls/antrea.yml
index 08dff9cda..773da3719 100644
--- a/build/yamls/antrea.yml
+++ b/build/yamls/antrea.yml
@@ -3572,6 +3572,7 @@ webhooks:
     apiVersions:
     - v1alpha2
     operations:
+    - CREATE
     - UPDATE
     - DELETE
     resources:
diff --git a/build/yamls/base/controller.yml b/build/yamls/base/controller.yml
index f52931537..7ce4a28da 100644
--- a/build/yamls/base/controller.yml
+++ b/build/yamls/base/controller.yml
@@ -175,7 +175,7 @@ webhooks:
         namespace: "kube-system"
         path: "/validate/ippool"
     rules:
-      - operations: ["UPDATE", "DELETE"]
+      - operations: ["CREATE", "UPDATE", "DELETE"]
         apiGroups: ["crd.antrea.io"]
         apiVersions: ["v1alpha2"]
         resources: ["ippools"]
diff --git a/build/yamls/windows/base/conf/antrea-agent.conf b/build/yamls/windows/base/conf/antrea-agent.conf
index 23bcaed14..7fbb69524 100644
--- a/build/yamls/windows/base/conf/antrea-agent.conf
+++ b/build/yamls/windows/base/conf/antrea-agent.conf
@@ -5,6 +5,11 @@ featureGates:
 # Service traffic.
 #  AntreaProxy: true
 
+# Enable EndpointSlice support in AntreaProxy. Don't enable this feature unless that EndpointSlice
+# API version v1beta1 is supported and set as enabled in Kubernetes. If AntreaProxy is not enabled,
+# this flag will not take effect.
+#  EndpointSlice: false
+
 # Enable flowexporter which exports polled conntrack connections as IPFIX flow records from each agent to a configured collector.
 #  FlowExporter: false
 
diff --git a/ci/gh-actions/has-changes/README.md b/ci/gh-actions/has-changes/README.md
index 49aed7af0..c30f20399 100644
--- a/ci/gh-actions/has-changes/README.md
+++ b/ci/gh-actions/has-changes/README.md
@@ -1,5 +1,10 @@
 # "Has Changes" Docker Action
 
+**This Action has been moved to its own
+  [repository](https://github.com/antrea-io/has-changes). This directory will be
+  removed in the near future. We preserve it for now so that existing PRs can be
+  merged without requiring a rebase.**
+
 This action sets a boolean output (`has_changes`) if the diff (`push` or
 `pull_request` event) includes changes outside of a provided list of paths.
 
diff --git a/cmd/antrea-agent/agent.go b/cmd/antrea-agent/agent.go
index 2fb4e22a9..dcb00e867 100644
--- a/cmd/antrea-agent/agent.go
+++ b/cmd/antrea-agent/agent.go
@@ -53,7 +53,6 @@ import (
 	"antrea.io/antrea/pkg/agent/secondarynetwork/cnipodcache"
 	"antrea.io/antrea/pkg/agent/secondarynetwork/podwatch"
 	"antrea.io/antrea/pkg/agent/stats"
-	"antrea.io/antrea/pkg/agent/types"
 	crdinformers "antrea.io/antrea/pkg/client/informers/externalversions"
 	"antrea.io/antrea/pkg/controller/externalippool"
 	"antrea.io/antrea/pkg/features"
@@ -62,6 +61,7 @@ import (
 	ofconfig "antrea.io/antrea/pkg/ovs/openflow"
 	"antrea.io/antrea/pkg/ovs/ovsconfig"
 	"antrea.io/antrea/pkg/signals"
+	"antrea.io/antrea/pkg/util/channel"
 	"antrea.io/antrea/pkg/util/cipher"
 	"antrea.io/antrea/pkg/util/k8s"
 	"antrea.io/antrea/pkg/version"
@@ -257,10 +257,10 @@ func run(o *Options) error {
 		}
 	}
 
-	// entityUpdates is a channel for receiving entity updates from CNIServer and
-	// notifying NetworkPolicyController to reconcile rules related to the
-	// updated entities.
-	entityUpdates := make(chan types.EntityReference, 100)
+	// podUpdateChannel is a channel for receiving Pod updates from CNIServer and
+	// notifying NetworkPolicyController and EgressController to reconcile rules
+	// related to the updated Pods.
+	podUpdateChannel := channel.NewSubscribableChannel("PodUpdate", 100)
 	// We set flow poll interval as the time interval for rule deletion in the async
 	// rule cache, which is implemented as part of the idAllocator. This is to preserve
 	// the rule info for populating NetworkPolicy fields in the Flow Exporter even
@@ -278,7 +278,7 @@ func run(o *Options) error {
 		ofClient,
 		ifaceStore,
 		nodeConfig.Name,
-		entityUpdates,
+		podUpdateChannel,
 		groupCounters,
 		groupIDUpdates,
 		antreaPolicyEnabled,
@@ -328,7 +328,7 @@ func run(o *Options) error {
 	if features.DefaultFeatureGate.Enabled(features.Egress) {
 		egressController, err = egress.NewEgressController(
 			ofClient, antreaClientProvider, crdClient, ifaceStore, routeClient, nodeConfig.Name, nodeConfig.NodeTransportInterfaceName,
-			memberlistCluster, egressInformer, nodeInformer,
+			memberlistCluster, egressInformer, podUpdateChannel,
 		)
 		if err != nil {
 			return fmt.Errorf("error creating new Egress controller: %v", err)
@@ -365,12 +365,12 @@ func run(o *Options) error {
 	var cniPodInfoStore cnipodcache.CNIPodInfoStore
 	if features.DefaultFeatureGate.Enabled(features.SecondaryNetwork) {
 		cniPodInfoStore = cnipodcache.NewCNIPodInfoStore()
-		err = cniServer.Initialize(ovsBridgeClient, ofClient, ifaceStore, entityUpdates, cniPodInfoStore)
+		err = cniServer.Initialize(ovsBridgeClient, ofClient, ifaceStore, podUpdateChannel, cniPodInfoStore)
 		if err != nil {
 			return fmt.Errorf("error initializing CNI server with cniPodInfoStore cache: %v", err)
 		}
 	} else {
-		err = cniServer.Initialize(ovsBridgeClient, ofClient, ifaceStore, entityUpdates, nil)
+		err = cniServer.Initialize(ovsBridgeClient, ofClient, ifaceStore, podUpdateChannel, nil)
 		if err != nil {
 			return fmt.Errorf("error initializing CNI server: %v", err)
 		}
@@ -460,6 +460,8 @@ func run(o *Options) error {
 
 	log.StartLogFileNumberMonitor(stopCh)
 
+	go podUpdateChannel.Run(stopCh)
+
 	go routeClient.Run(stopCh)
 
 	go cniServer.Run(stopCh)
diff --git a/docs/antrea-ipam.md b/docs/antrea-ipam.md
index 0287a8ae4..3d4dbf26c 100644
--- a/docs/antrea-ipam.md
+++ b/docs/antrea-ipam.md
@@ -47,43 +47,48 @@ cluster. Valid range is 16 to 30. Default is 24.
 - `nodeCIDRMaskSizeIPv6`: Mask size for IPv6 Node CIDR in IPv6 or dual-stack
 cluster. Valid range is 64 to 126. Default is 64.
 
-## Antrea Flexible-IPAM
-
-Antrea supports flexible control over Pod IP addressing since version 1.4. This can be
-achieved by configuring `IPPool` CRD with a desired set of IP ranges. The `IPPool` can be
-annotated to Namespace, Pod and PodTemplate of StatefulSet/Deployment. Antrea will manage
-IP address assignment for corresponding Pods according to `IPPool` spec.
-
-Note that the IP pool annotation cannot be updated or deleted without recreating the
-resource. An `IPPool` can be extended, but cannot be shrunk if already assigned to a
-resource. The IP ranges of IPPools must not overlap, otherwise it would lead to undefined
-behavior.
-
-Regular `Subnet per Node` IPAM will continue to be used for resources without the IP pool
-annotation, or when `AntreaIPAM` feature is disabled.
+## Antrea Flexible IPAM
+
+Antrea supports flexible control over Pod IP addressing since version 1.4. Pod
+IP addresses can be allocated from an `IPPool`. When a Pod's IP is allocated
+from an IPPool, the traffic from the Pod to Pods on another Node or to external
+network will be sent to the underlay network through the Node's transport
+network interface, and will be forwarded/routed by the underlay network. We also
+call this forwarding mode `bridging mode`.
+
+`IPPool` CRD defines a desired set of IP ranges. An `IPPool` can be annotated to
+Namespace, Pod and PodTemplate of StatefulSet/Deployment. Then Antrea will
+manage IP address assignment for corresponding Pods according to `IPPool` spec.
+Note that the IP pool annotation cannot be updated or deleted without recreating
+the resource. An `IPPool` can be extended, but cannot be shrunk if already
+assigned to a resource. The IP ranges of IPPools must not overlap, otherwise it
+would lead to undefined behavior.
+
+Regular `Subnet per Node` IPAM will continue to be used for resources without the
+IPPool annotation, or when the `AntreaIPAM` feature is disabled.
 
 ### Usage
 
-#### Enable AntreaIPAM feature gate
+#### Enable AntreaIPAM feature gate and bridging mode
 
-Users need to enable `AntreaIPAM` from the featureGates map defined in antrea.yml for
-both Controller and Agent:
+To enable flexible IPAM, you need to enable the `AntreaIPAM` feature gate for
+for both `antrea-controller` and `antrea-agent`, and set the `enableBridgingMode`
+configuration parameter of `antrea-agent` to `true`. The needed changes in the
+Antrea deployment YAML are:
 
 ```yaml
   antrea-controller.conf: |
     ...
     featureGates:
-    # Enable flexible IPAM mode for Antrea. This mode allows to assign IP Ranges to Namespaces,
-    # Deployments and StatefulSets via IP Pool annotation.
       AntreaIPAM: true
     ...
   antrea-agent.conf: |
     ...
     featureGates:
-    # Enable flexible IPAM mode for Antrea. This mode allows to assign IP Ranges to Namespaces,
-    # Deployments and StatefulSets via IP Pool annotation.
       AntreaIPAM: true
     ...
+    enableBridgingMode: true
+    ...
 ```
 
 #### Create IPPool CR
diff --git a/go.mod b/go.mod
index dbbe609a6..73ccd838a 100644
--- a/go.mod
+++ b/go.mod
@@ -46,7 +46,7 @@ require (
 	github.com/stretchr/testify v1.7.0
 	github.com/ti-mo/conntrack v0.4.0
 	github.com/vishvananda/netlink v1.1.1-0.20210510164352-d17758a128bf
-	github.com/vmware/go-ipfix v0.5.11
+	github.com/vmware/go-ipfix v0.5.12
 	go.uber.org/multierr v1.6.0
 	golang.org/x/crypto v0.0.0-20210503195802-e9a32991a82e
 	golang.org/x/exp v0.0.0-20200224162631-6cc2880d07d6
diff --git a/go.sum b/go.sum
index 930c3d8c0..10751fffa 100644
--- a/go.sum
+++ b/go.sum
@@ -747,8 +747,8 @@ github.com/vishvananda/netns v0.0.0-20180720170159-13995c7128cc/go.mod h1:ZjcWmF
 github.com/vishvananda/netns v0.0.0-20191106174202-0a2b9b5464df/go.mod h1:JP3t17pCcGlemwknint6hfoeCVQrEMVwxRLRjXpq+BU=
 github.com/vishvananda/netns v0.0.0-20200728191858-db3c7e526aae h1:4hwBBUfQCFe3Cym0ZtKyq7L16eZUtYKs+BaHDN6mAns=
 github.com/vishvananda/netns v0.0.0-20200728191858-db3c7e526aae/go.mod h1:DD4vA1DwXk04H54A1oHXtwZmA0grkVMdPxx/VGLCah0=
-github.com/vmware/go-ipfix v0.5.11 h1:1731EiUCTkhrK0YTxVbpT3YVyfyIj3ACua+QjL+9eq0=
-github.com/vmware/go-ipfix v0.5.11/go.mod h1:yzbG1rv+yJ8GeMrRm+MDhOV3akygNZUHLhC1pDoD2AY=
+github.com/vmware/go-ipfix v0.5.12 h1:mqQknlvnvDY25apPNy9c27ri3FMDFIhzvO68Kk5Qp58=
+github.com/vmware/go-ipfix v0.5.12/go.mod h1:yzbG1rv+yJ8GeMrRm+MDhOV3akygNZUHLhC1pDoD2AY=
 github.com/xdg/scram v0.0.0-20180814205039-7eeb5667e42c/go.mod h1:lB8K/P019DLNhemzwFU4jHLhdvlE6uDZjXFejJXr49I=
 github.com/xdg/stringprep v1.0.0/go.mod h1:Jhud4/sHMO4oL310DaZAKk9ZaJ08SJfe+sJh0HrGL1Y=
 github.com/xiang90/probing v0.0.0-20190116061207-43a291ad63a2 h1:eY9dn8+vbi4tKz5Qo6v2eYzo7kUS51QINcR5jNpbZS8=
diff --git a/hack/build-antrea-ubuntu-all.sh b/hack/build-antrea-linux-all.sh
similarity index 74%
rename from hack/build-antrea-ubuntu-all.sh
rename to hack/build-antrea-linux-all.sh
index 184e03057..bd931040d 100755
--- a/hack/build-antrea-ubuntu-all.sh
+++ b/hack/build-antrea-linux-all.sh
@@ -20,14 +20,15 @@ function echoerr {
     >&2 echo "$@"
 }
 
-_usage="Usage: $0 [--pull] [--push-base-images] [--coverage] [--platform <PLATFORM>]
+_usage="Usage: $0 [--pull] [--push-base-images] [--coverage] [--platform <PLATFORM>] [--distro [ubuntu|ubi]]
 Build the antrea/antrea-ubuntu image, as well as all the base images in the build chain. This is
 typically used in CI to build the image with the latest version of all dependencies, taking into
 account changes to all Dockerfiles.
         --pull                  Always attempt to pull a newer version of the base images.
         --push-base-images      Push built images to the registry. Only base images will be pushed.
         --coverage              Build the image with support for code coverage.
-        --platform <PLATFORM>   Target platform for the images if server is multi-platform capable."
+        --platform <PLATFORM>   Target platform for the images if server is multi-platform capable.
+        --distro <distro>       Target Linux distribution."
 
 function print_usage {
     echoerr "$_usage"
@@ -37,6 +38,7 @@ PULL=false
 PUSH=false
 COVERAGE=false
 PLATFORM=""
+DISTRO="ubuntu"
 
 while [[ $# -gt 0 ]]
 do
@@ -59,6 +61,10 @@ case $key in
     PLATFORM="$2"
     shift 2
     ;;
+    --distro)
+    DISTRO="$2"
+    shift 2
+    ;;
     -h|--help)
     print_usage
     exit 0
@@ -83,6 +89,17 @@ if [ "$PLATFORM" != "" ]; then
     ARGS="$ARGS --platform $PLATFORM"
     PLATFORM_ARG="--platform $PLATFORM"
 fi
+if [ "$DISTRO" != "ubuntu" ] && [ "$DISTRO" != "ubi" ]; then
+    echoerr "Invalid distribution $DISTRO"
+    exit 1
+fi
+if [ "$DISTRO" == "ubi" ]; then
+    if $COVERAGE ; then
+        echoerr "No coverage build for UBI8"
+        exit 1
+    fi
+    ARGS="$ARGS --distro ubi"
+fi
 
 OVS_VERSION=$(head -n 1 build/images/deps/ovs-version)
 CNI_BINARIES_VERSION=$(head -n 1 build/images/deps/cni-binaries-version)
@@ -103,12 +120,21 @@ if $PULL; then
         docker pull ${DOCKER_REGISTRY}/antrea/golang:$GO_VERSION
         docker tag ${DOCKER_REGISTRY}/antrea/golang:$GO_VERSION golang:$GO_VERSION
     fi
-    IMAGES_LIST=(
-        "antrea/openvswitch-debs:$OVS_VERSION"
-        "antrea/openvswitch:$OVS_VERSION"
-        "antrea/cni-binaries:$CNI_BINARIES_VERSION"
-        "antrea/base-ubuntu:$OVS_VERSION"
-    )
+    if [ "$DISTRO" == "ubuntu" ]; then
+        IMAGES_LIST=(
+            "antrea/openvswitch-debs:$OVS_VERSION"
+            "antrea/openvswitch:$OVS_VERSION"
+            "antrea/cni-binaries:$CNI_BINARIES_VERSION"
+            "antrea/base-ubuntu:$OVS_VERSION"
+        )
+    elif [ "$DISTRO" == "ubi" ]; then
+        IMAGES_LIST=(
+            "antrea/openvswitch-rpms:$OVS_VERSION"
+            "antrea/openvswitch-ubi:$OVS_VERSION"
+            "antrea/cni-binaries:$CNI_BINARIES_VERSION"
+            "antrea/base-ubi:$OVS_VERSION"
+        )
+    fi
     for image in "${IMAGES_LIST[@]}"; do
         if [[ ${DOCKER_REGISTRY} == "" ]]; then
             docker pull $PLATFORM_ARG "${image}" || true
@@ -131,10 +157,14 @@ cd build/images/base
 cd -
 
 export NO_PULL=1
-if $COVERAGE; then
-    make build-ubuntu-coverage
-else
-    make
+if [ "$DISTRO" == "ubuntu" ]; then
+    if $COVERAGE; then
+        make build-ubuntu-coverage
+    else
+        make
+    fi
+elif [ "$DISTRO" == "ubi" ]; then
+    make build-ubi
 fi
 
 popd > /dev/null
diff --git a/multicluster/controllers/multicluster/common/helper.go b/multicluster/controllers/multicluster/common/helper.go
index a94453815..839421b53 100644
--- a/multicluster/controllers/multicluster/common/helper.go
+++ b/multicluster/controllers/multicluster/common/helper.go
@@ -16,7 +16,8 @@ package common
 import corev1 "k8s.io/api/core/v1"
 
 const (
-	AntreaMCServiceAnnotation = "multicluster.antrea.io/imported-service"
+	AntreaMCServiceAnnotation   = "multicluster.antrea.io/imported-service"
+	AntreaMCClusterIDAnnotation = "multicluster.antrea.io/local-cluster-id"
 
 	AntreaMCSPrefix   = "antrea-mc-"
 	ServiceKind       = "Service"
diff --git a/multicluster/controllers/multicluster/commonarea/remote_common_area.go b/multicluster/controllers/multicluster/commonarea/remote_common_area.go
index 43e64b31f..6ccd30ee7 100644
--- a/multicluster/controllers/multicluster/commonarea/remote_common_area.go
+++ b/multicluster/controllers/multicluster/commonarea/remote_common_area.go
@@ -378,6 +378,7 @@ func (r *remoteCommonArea) StartWatching() error {
 		r.ClusterManager.GetClient(),
 		r.ClusterManager.GetScheme(),
 		r.localClusterClient,
+		string(r.remoteCommonAreaManager.GetLocalClusterID()),
 		r,
 	)
 
diff --git a/multicluster/controllers/multicluster/commonarea/resourceimport_controller.go b/multicluster/controllers/multicluster/commonarea/resourceimport_controller.go
index 12e197c37..b5952acf8 100644
--- a/multicluster/controllers/multicluster/commonarea/resourceimport_controller.go
+++ b/multicluster/controllers/multicluster/commonarea/resourceimport_controller.go
@@ -59,15 +59,17 @@ type ResourceImportReconciler struct {
 	client.Client
 	Scheme              *runtime.Scheme
 	localClusterClient  client.Client
+	localClusterID      string
 	remoteCommonArea    RemoteCommonArea
 	installedResImports cache.Indexer
 }
 
-func NewResourceImportReconciler(client client.Client, scheme *runtime.Scheme, localClusterClient client.Client, remoteCommonArea RemoteCommonArea) *ResourceImportReconciler {
+func NewResourceImportReconciler(client client.Client, scheme *runtime.Scheme, localClusterClient client.Client, localClusterID string, remoteCommonArea RemoteCommonArea) *ResourceImportReconciler {
 	return &ResourceImportReconciler{
 		Client:             client,
 		Scheme:             scheme,
 		localClusterClient: localClusterClient,
+		localClusterID:     localClusterID,
 		remoteCommonArea:   remoteCommonArea,
 		installedResImports: cache.NewIndexer(resImportIndexerKeyFunc, cache.Indexers{
 			resImportIndexer: resImportIndexerFunc,
@@ -173,7 +175,7 @@ func (r *ResourceImportReconciler) handleResImpUpdateForService(ctx context.Cont
 	if err != nil && !svcImpNotFound {
 		return ctrl.Result{}, err
 	}
-	svcImpObj := getMCServiceImport(resImp)
+	svcImpObj := getMCServiceImport(resImp, r.localClusterID)
 	// Set multi-cluster Service's ClusterIP as ServiceImport's ClusterSetIP
 	if svc.Spec.ClusterIP != "" {
 		svcImpObj.Spec.IPs = []string{svc.Spec.ClusterIP}
@@ -205,6 +207,7 @@ func (r *ResourceImportReconciler) handleResImpUpdateForService(ctx context.Cont
 
 	if !apiequality.Semantic.DeepEqual(svcImp.Spec, svcImpObj.Spec) {
 		svcImp.Spec = svcImpObj.Spec
+		addAnnotation(svcImp, r.localClusterID)
 		err = r.localClusterClient.Update(ctx, svcImp, &client.UpdateOptions{})
 		if err != nil {
 			klog.ErrorS(err, "failed to update ServiceImport", "serviceimport", svcImpName.String())
@@ -364,11 +367,12 @@ func getMCService(resImp *multiclusterv1alpha1.ResourceImport) *corev1.Service {
 	return mcs
 }
 
-func getMCServiceImport(resImp *multiclusterv1alpha1.ResourceImport) *k8smcsv1alpha1.ServiceImport {
+func getMCServiceImport(resImp *multiclusterv1alpha1.ResourceImport, clusterID string) *k8smcsv1alpha1.ServiceImport {
 	svcImp := &k8smcsv1alpha1.ServiceImport{
 		ObjectMeta: metav1.ObjectMeta{
-			Name:      resImp.Spec.Name,
-			Namespace: resImp.Spec.Namespace,
+			Name:        resImp.Spec.Name,
+			Namespace:   resImp.Spec.Namespace,
+			Annotations: map[string]string{common.AntreaMCClusterIDAnnotation: clusterID},
 		},
 		Spec: resImp.Spec.ServiceImport.Spec,
 	}
@@ -397,6 +401,15 @@ func removeLocalSubsets(local []corev1.EndpointSubset, allSubsets []corev1.Endpo
 	return newSubsets
 }
 
+func addAnnotation(svcImport *k8smcsv1alpha1.ServiceImport, localClusterID string) {
+	if svcImport.Annotations == nil {
+		svcImport.Annotations = make(map[string]string)
+	}
+	if _, ok := svcImport.Annotations[common.AntreaMCClusterIDAnnotation]; !ok {
+		svcImport.Annotations[common.AntreaMCClusterIDAnnotation] = localClusterID
+	}
+}
+
 // SetupWithManager sets up the controller with the ClusterManager
 // which will set up controllers for resources that need to be monitored
 // in the remoteCommonArea.
diff --git a/multicluster/controllers/multicluster/resourceexport_controller.go b/multicluster/controllers/multicluster/resourceexport_controller.go
index af5cb422e..8f3254b8c 100644
--- a/multicluster/controllers/multicluster/resourceexport_controller.go
+++ b/multicluster/controllers/multicluster/resourceexport_controller.go
@@ -71,13 +71,8 @@ func NewResourceExportReconciler(
 
 // Reconcile is part of the main kubernetes reconciliation loop which aims to
 // move the current state of the cluster closer to the desired state.
-// TODO(user): Modify the Reconcile function to compare the state specified by
-// the ResourceExport object against the actual cluster state, and then
-// perform operations to make the cluster state reflect the state specified by
-// the user.
-//
-// For more details, check Reconcile and its Result here:
-// - https://pkg.go.dev/sigs.k8s.io/controller-runtime@v0.8.3/pkg/reconcile
+// Reconcile will process all kinds of ResourceExport. Service and Endpoint kinds of ResourceExport
+// will be handled in this file, and all other kinds will have their own handler files, eg: newkind_handler.go
 func (r *ResourceExportReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
 	klog.V(2).InfoS("reconciling ResourceExport", "resourceexport", req.NamespacedName)
 	var resExport mcsv1alpha1.ResourceExport
@@ -85,6 +80,20 @@ func (r *ResourceExportReconciler) Reconcile(ctx context.Context, req ctrl.Reque
 		return ctrl.Result{}, client.IgnoreNotFound(err)
 	}
 
+	switch resExport.Spec.Kind {
+	case common.ServiceKind:
+		klog.V(2).InfoS("Reconciling Service type of ResourceExport", "resourceexport", req.NamespacedName)
+	case common.EndpointsKind:
+		klog.V(2).InfoS("Reconciling Endpoint type of ResourceExport", "resourceexport", req.NamespacedName)
+	// Developer can add more supported kinds here in the future.
+	// eg: add a new case and a new method 'handleNewKind' in a new file like 'newkind_handler.go'
+	//  case common.NewKind:
+	//    return r.handleNewKind(ctx, req, resExport)
+	default:
+		klog.InfoS("It's not expected kind, skip reconciling ResourceExport", "resourceexport", req.NamespacedName)
+		return ctrl.Result{}, nil
+	}
+
 	// We are using Finalizers to implement asynchronous pre-delete hooks.
 	// When ResourceExport is deleted, it will have non-zero DeletionTimestamp
 	// but controller can still get the deleted ResourceExport object, so we can
@@ -96,11 +105,7 @@ func (r *ResourceExportReconciler) Reconcile(ctx context.Context, req ctrl.Reque
 			if err != nil {
 				return ctrl.Result{}, err
 			}
-			resExport.SetFinalizers(common.RemoveStringFromSlice(resExport.Finalizers, common.ResourceExportFinalizer))
-			if err := r.Client.Update(ctx, &resExport, &client.UpdateOptions{}); err != nil {
-				return ctrl.Result{}, err
-			}
-			return ctrl.Result{}, nil
+			return r.deleteResourceExport(&resExport)
 		}
 		return ctrl.Result{}, nil
 	}
@@ -424,6 +429,15 @@ func (r *ResourceExportReconciler) updateResourceExportStatus(resExport *mcsv1al
 	}
 }
 
+// deleteResourceExport removes ResourceExport finalizer string and updates it, so Kubernetes can complete deletion.
+func (r *ResourceExportReconciler) deleteResourceExport(resExport *mcsv1alpha1.ResourceExport) (ctrl.Result, error) {
+	resExport.SetFinalizers(common.RemoveStringFromSlice(resExport.Finalizers, common.ResourceExportFinalizer))
+	if err := r.Client.Update(ctx, resExport, &client.UpdateOptions{}); err != nil {
+		return ctrl.Result{}, err
+	}
+	return ctrl.Result{}, nil
+}
+
 // SetupWithManager sets up the controller with the Manager.
 func (r *ResourceExportReconciler) SetupWithManager(mgr ctrl.Manager) error {
 	// ignore status update event via GenerationChangedPredicate
diff --git a/multicluster/controllers/multicluster/serviceexport_controller.go b/multicluster/controllers/multicluster/serviceexport_controller.go
index 14887e565..00268c09e 100644
--- a/multicluster/controllers/multicluster/serviceexport_controller.go
+++ b/multicluster/controllers/multicluster/serviceexport_controller.go
@@ -330,6 +330,11 @@ func (r *ServiceExportReconciler) Reconcile(ctx context.Context, req ctrl.Reques
 			return ctrl.Result{}, err
 		}
 	}
+
+	if err = r.updateSvcExportAnnotation(&svcExport); err != nil {
+		// Ignore the error since it's not critical and we can update in next event.
+		klog.ErrorS(err, "Failed to update ServiceExport annotation", "serviceexport", klog.KObj(&svcExport))
+	}
 	return ctrl.Result{}, nil
 }
 
@@ -378,12 +383,21 @@ func (r *ServiceExportReconciler) handleEndpointDeleteEvent(ctx context.Context,
 	return nil
 }
 
+func (r *ServiceExportReconciler) updateSvcExportAnnotation(svcExport *k8smcsv1alpha1.ServiceExport) error {
+	addAnnotation(svcExport, r.localClusterID)
+	if err := r.Client.Update(ctx, svcExport, &client.UpdateOptions{}); err != nil {
+		return err
+	}
+	return nil
+}
+
 func (r *ServiceExportReconciler) updateSvcExportStatus(ctx context.Context, req ctrl.Request, cause reason) error {
 	svcExport := &k8smcsv1alpha1.ServiceExport{}
 	err := r.Client.Get(ctx, req.NamespacedName, svcExport)
 	if err != nil {
 		return client.IgnoreNotFound(err)
 	}
+	addAnnotation(svcExport, r.localClusterID)
 	now := metav1.Now()
 	var res, message *string
 	switch cause {
@@ -580,6 +594,15 @@ func (r *ServiceExportReconciler) updateOrCreateResourceExport(resName string,
 	return nil
 }
 
+func addAnnotation(svcExport *k8smcsv1alpha1.ServiceExport, localClusterID string) {
+	if svcExport.Annotations == nil {
+		svcExport.Annotations = make(map[string]string)
+	}
+	if _, ok := svcExport.Annotations[common.AntreaMCClusterIDAnnotation]; !ok {
+		svcExport.Annotations[common.AntreaMCClusterIDAnnotation] = localClusterID
+	}
+}
+
 func getEndPointsAddress(ep *corev1.Endpoints) []string {
 	var epAddrs []string
 	for _, s := range ep.Subsets {
diff --git a/pkg/agent/cniserver/pod_configuration.go b/pkg/agent/cniserver/pod_configuration.go
index 930dae05f..25930b022 100644
--- a/pkg/agent/cniserver/pod_configuration.go
+++ b/pkg/agent/cniserver/pod_configuration.go
@@ -30,10 +30,9 @@ import (
 	"antrea.io/antrea/pkg/agent/openflow"
 	"antrea.io/antrea/pkg/agent/route"
 	"antrea.io/antrea/pkg/agent/secondarynetwork/cnipodcache"
-	"antrea.io/antrea/pkg/agent/types"
 	"antrea.io/antrea/pkg/agent/util"
-	"antrea.io/antrea/pkg/apis/controlplane/v1beta2"
 	"antrea.io/antrea/pkg/ovs/ovsconfig"
+	"antrea.io/antrea/pkg/util/channel"
 	"antrea.io/antrea/pkg/util/k8s"
 )
 
@@ -63,9 +62,9 @@ type podConfigurator struct {
 	ifaceStore      interfacestore.InterfaceStore
 	gatewayMAC      net.HardwareAddr
 	ifConfigurator  *ifConfigurator
-	// entityUpdates is a channel for notifying updates of local endpoints / entities (most notably Pod)
-	// to other components which may benefit from this information, i.e NetworkPolicyController.
-	entityUpdates chan<- types.EntityReference
+	// podUpdateNotifier is used for notifying updates of local Pods to other components which may benefit from this
+	// information, i.e. NetworkPolicyController, EgressController.
+	podUpdateNotifier channel.Notifier
 	// consumed by secondary network creation.
 	podInfoStore cnipodcache.CNIPodInfoStore
 }
@@ -78,7 +77,7 @@ func newPodConfigurator(
 	gatewayMAC net.HardwareAddr,
 	ovsDatapathType ovsconfig.OVSDatapathType,
 	isOvsHardwareOffloadEnabled bool,
-	entityUpdates chan<- types.EntityReference,
+	podUpdateNotifier channel.Notifier,
 	podInfoStore cnipodcache.CNIPodInfoStore,
 ) (*podConfigurator, error) {
 	ifConfigurator, err := newInterfaceConfigurator(ovsDatapathType, isOvsHardwareOffloadEnabled)
@@ -86,14 +85,14 @@ func newPodConfigurator(
 		return nil, err
 	}
 	return &podConfigurator{
-		ovsBridgeClient: ovsBridgeClient,
-		ofClient:        ofClient,
-		routeClient:     routeClient,
-		ifaceStore:      ifaceStore,
-		gatewayMAC:      gatewayMAC,
-		ifConfigurator:  ifConfigurator,
-		entityUpdates:   entityUpdates,
-		podInfoStore:    podInfoStore,
+		ovsBridgeClient:   ovsBridgeClient,
+		ofClient:          ofClient,
+		routeClient:       routeClient,
+		ifaceStore:        ifaceStore,
+		gatewayMAC:        gatewayMAC,
+		ifConfigurator:    ifConfigurator,
+		podUpdateNotifier: podUpdateNotifier,
+		podInfoStore:      podInfoStore,
 	}, nil
 }
 
@@ -486,9 +485,7 @@ func (pc *podConfigurator) connectInterfaceToOVSCommon(ovsPortName string, conta
 	// Add containerConfig into local cache
 	pc.ifaceStore.AddInterface(containerConfig)
 	// Notify the Pod update event to required components.
-	pc.entityUpdates <- types.EntityReference{
-		Pod: &v1beta2.PodReference{Name: containerConfig.PodName, Namespace: containerConfig.PodNamespace},
-	}
+	pc.podUpdateNotifier.Notify(k8s.NamespacedName(containerConfig.PodNamespace, containerConfig.PodName))
 	return nil
 }
 
diff --git a/pkg/agent/cniserver/pod_configuration_windows.go b/pkg/agent/cniserver/pod_configuration_windows.go
index d42578e1e..6f61e241d 100644
--- a/pkg/agent/cniserver/pod_configuration_windows.go
+++ b/pkg/agent/cniserver/pod_configuration_windows.go
@@ -24,9 +24,7 @@ import (
 	"k8s.io/klog/v2"
 
 	"antrea.io/antrea/pkg/agent/interfacestore"
-	"antrea.io/antrea/pkg/agent/types"
 	"antrea.io/antrea/pkg/agent/util"
-	"antrea.io/antrea/pkg/apis/controlplane/v1beta2"
 	"antrea.io/antrea/pkg/util/k8s"
 )
 
@@ -51,9 +49,7 @@ func (pc *podConfigurator) connectInterfaceToOVSAsync(ifConfig *interfacestore.I
 		// Update interface config with the ofPort.
 		ifConfig.OVSPortConfig.OFPort = ofPort
 		// Notify the Pod update event to required components.
-		pc.entityUpdates <- types.EntityReference{
-			Pod: &v1beta2.PodReference{Name: ifConfig.PodName, Namespace: ifConfig.PodNamespace},
-		}
+		pc.podUpdateNotifier.Notify(k8s.NamespacedName(ifConfig.PodNamespace, ifConfig.PodName))
 		return nil
 	})
 }
diff --git a/pkg/agent/cniserver/server.go b/pkg/agent/cniserver/server.go
index 6059ebd0e..82c9c8d0d 100644
--- a/pkg/agent/cniserver/server.go
+++ b/pkg/agent/cniserver/server.go
@@ -40,11 +40,11 @@ import (
 	"antrea.io/antrea/pkg/agent/openflow"
 	"antrea.io/antrea/pkg/agent/route"
 	"antrea.io/antrea/pkg/agent/secondarynetwork/cnipodcache"
-	"antrea.io/antrea/pkg/agent/types"
 	"antrea.io/antrea/pkg/agent/util"
 	cnipb "antrea.io/antrea/pkg/apis/cni/v1beta1"
 	"antrea.io/antrea/pkg/cni"
 	"antrea.io/antrea/pkg/ovs/ovsconfig"
+	"antrea.io/antrea/pkg/util/channel"
 )
 
 const (
@@ -589,7 +589,7 @@ func (s *CNIServer) Initialize(
 	ovsBridgeClient ovsconfig.OVSBridgeClient,
 	ofClient openflow.Client,
 	ifaceStore interfacestore.InterfaceStore,
-	entityUpdates chan<- types.EntityReference,
+	podUpdateNotifier channel.Notifier,
 	podInfoStore cnipodcache.CNIPodInfoStore,
 ) error {
 	var err error
@@ -602,7 +602,7 @@ func (s *CNIServer) Initialize(
 
 	s.podConfigurator, err = newPodConfigurator(
 		ovsBridgeClient, ofClient, s.routeClient, ifaceStore, s.nodeConfig.GatewayConfig.MAC,
-		ovsBridgeClient.GetOVSDatapathType(), ovsBridgeClient.IsHardwareOffloadEnabled(), entityUpdates,
+		ovsBridgeClient.GetOVSDatapathType(), ovsBridgeClient.IsHardwareOffloadEnabled(), podUpdateNotifier,
 		podInfoStore,
 	)
 	if err != nil {
diff --git a/pkg/agent/controller/egress/egress_controller.go b/pkg/agent/controller/egress/egress_controller.go
index 510c952ad..167d0756a 100644
--- a/pkg/agent/controller/egress/egress_controller.go
+++ b/pkg/agent/controller/egress/egress_controller.go
@@ -30,7 +30,6 @@ import (
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/apimachinery/pkg/watch"
-	coreinformers "k8s.io/client-go/informers/core/v1"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/util/retry"
 	"k8s.io/client-go/util/workqueue"
@@ -48,6 +47,7 @@ import (
 	crdinformers "antrea.io/antrea/pkg/client/informers/externalversions/crd/v1alpha2"
 	crdlisters "antrea.io/antrea/pkg/client/listers/crd/v1alpha2"
 	"antrea.io/antrea/pkg/controller/metrics"
+	"antrea.io/antrea/pkg/util/channel"
 	"antrea.io/antrea/pkg/util/k8s"
 )
 
@@ -154,7 +154,7 @@ func NewEgressController(
 	nodeTransportInterface string,
 	cluster *memberlist.Cluster,
 	egressInformer crdinformers.EgressInformer,
-	nodeInformer coreinformers.NodeInformer,
+	podUpdateSubscriber channel.Subscriber,
 ) (*EgressController, error) {
 	c := &EgressController{
 		ofClient:             ofClient,
@@ -207,11 +207,26 @@ func NewEgressController(
 		},
 		resyncPeriod,
 	)
+	// Subscribe Pod update events from CNIServer to enforce Egress earlier, instead of waiting for their IPs are
+	// reported to kube-apiserver and processed by antrea-controller.
+	podUpdateSubscriber.Subscribe(c.processPodUpdate)
 	c.localIPDetector.AddEventHandler(c.onLocalIPUpdate)
 	c.cluster.AddClusterEventHandler(c.enqueueEgressesByExternalIPPool)
 	return c, nil
 }
 
+// processPodUpdate will be called when CNIServer publishes a Pod update event.
+// It triggers reconciling the effective Egress of the Pod.
+func (c *EgressController) processPodUpdate(pod string) {
+	c.egressBindingsMutex.Lock()
+	defer c.egressBindingsMutex.Unlock()
+	binding, exists := c.egressBindings[pod]
+	if !exists {
+		return
+	}
+	c.queue.Add(binding.effectiveEgress)
+}
+
 // addEgress processes Egress ADD events.
 func (c *EgressController) addEgress(obj interface{}) {
 	egress := obj.(*crdv1a2.Egress)
diff --git a/pkg/agent/controller/networkpolicy/cache.go b/pkg/agent/controller/networkpolicy/cache.go
index 8d3ad10b9..1adc88664 100644
--- a/pkg/agent/controller/networkpolicy/cache.go
+++ b/pkg/agent/controller/networkpolicy/cache.go
@@ -28,10 +28,10 @@ import (
 	"k8s.io/klog/v2"
 
 	"antrea.io/antrea/pkg/agent/metrics"
-	antreatypes "antrea.io/antrea/pkg/agent/types"
 	v1beta "antrea.io/antrea/pkg/apis/controlplane/v1beta2"
 	crdv1alpha1 "antrea.io/antrea/pkg/apis/crd/v1alpha1"
 	"antrea.io/antrea/pkg/querier"
+	"antrea.io/antrea/pkg/util/channel"
 	"antrea.io/antrea/pkg/util/k8s"
 )
 
@@ -156,9 +156,6 @@ type ruleCache struct {
 	// dirtyRuleHandler is a callback that is run upon finding a rule out-of-sync.
 	dirtyRuleHandler func(string)
 
-	// entityUpdates is a channel for receiving entity (e.g. Pod) updates from CNIServer.
-	entityUpdates <-chan antreatypes.EntityReference
-
 	// groupIDUpdates is a channel for receiving groupID for Service is assigned
 	// or released events from groupCounters.
 	groupIDUpdates <-chan string
@@ -337,7 +334,7 @@ func toServicesIndexFunc(obj interface{}) ([]string, error) {
 }
 
 // newRuleCache returns a new *ruleCache.
-func newRuleCache(dirtyRuleHandler func(string), podUpdate <-chan antreatypes.EntityReference, serviceGroupIDUpdate <-chan string) *ruleCache {
+func newRuleCache(dirtyRuleHandler func(string), podUpdateSubscriber channel.Subscriber, serviceGroupIDUpdate <-chan string) *ruleCache {
 	rules := cache.NewIndexer(
 		ruleKeyFunc,
 		cache.Indexers{addressGroupIndex: addressGroupIndexFunc, appliedToGroupIndex: appliedToGroupIndexFunc, policyIndex: policyIndexFunc, toServicesIndex: toServicesIndexFunc},
@@ -348,38 +345,34 @@ func newRuleCache(dirtyRuleHandler func(string), podUpdate <-chan antreatypes.En
 		policyMap:           make(map[string]*v1beta.NetworkPolicy),
 		rules:               rules,
 		dirtyRuleHandler:    dirtyRuleHandler,
-		entityUpdates:       podUpdate,
 		groupIDUpdates:      serviceGroupIDUpdate,
 	}
-	go cache.processEntityUpdates()
+	// Subscribe Pod update events from CNIServer.
+	podUpdateSubscriber.Subscribe(cache.processPodUpdate)
 	go cache.processGroupIDUpdates()
 	return cache
 }
 
-// processEntityUpdates is an infinite loop that takes entity (e.g. Pod)
-// update events from the channel, finds out AppliedToGroups that contains
-// this Pod and trigger reconciling of related rules.
+// processPodUpdate will be called when CNIServer publishes a Pod update event.
+// It finds out AppliedToGroups that contains this Pod and trigger reconciling
+// of related rules.
 // It can enforce NetworkPolicies to newly added Pods right after CNI ADD is
 // done if antrea-controller has computed the Pods' policies and propagated
 // them to this Node by their labels and NodeName, instead of waiting for their
 // IPs are reported to kube-apiserver and processed by antrea-controller.
-func (c *ruleCache) processEntityUpdates() {
-	for {
-		select {
-		case entity := <-c.entityUpdates:
-			func() {
-				member := &v1beta.GroupMember{
-					Pod:            entity.Pod,
-					ExternalEntity: entity.ExternalEntity,
-				}
-				c.appliedToSetLock.RLock()
-				defer c.appliedToSetLock.RUnlock()
-				for group, memberSet := range c.appliedToSetByGroup {
-					if memberSet.Has(member) {
-						c.onAppliedToGroupUpdate(group)
-					}
-				}
-			}()
+func (c *ruleCache) processPodUpdate(pod string) {
+	namespace, name := k8s.SplitNamespacedName(pod)
+	member := &v1beta.GroupMember{
+		Pod: &v1beta.PodReference{
+			Name:      name,
+			Namespace: namespace,
+		},
+	}
+	c.appliedToSetLock.RLock()
+	defer c.appliedToSetLock.RUnlock()
+	for group, memberSet := range c.appliedToSetByGroup {
+		if memberSet.Has(member) {
+			c.onAppliedToGroupUpdate(group)
 		}
 	}
 }
diff --git a/pkg/agent/controller/networkpolicy/networkpolicy_controller.go b/pkg/agent/controller/networkpolicy/networkpolicy_controller.go
index 33ba8111d..3396f1073 100644
--- a/pkg/agent/controller/networkpolicy/networkpolicy_controller.go
+++ b/pkg/agent/controller/networkpolicy/networkpolicy_controller.go
@@ -37,6 +37,7 @@ import (
 	"antrea.io/antrea/pkg/agent/types"
 	"antrea.io/antrea/pkg/apis/controlplane/v1beta2"
 	"antrea.io/antrea/pkg/querier"
+	"antrea.io/antrea/pkg/util/channel"
 )
 
 const (
@@ -112,7 +113,7 @@ func NewNetworkPolicyController(antreaClientGetter agent.AntreaClientProvider,
 	ofClient openflow.Client,
 	ifaceStore interfacestore.InterfaceStore,
 	nodeName string,
-	entityUpdates <-chan types.EntityReference,
+	podUpdateSubscriber channel.Subscriber,
 	groupCounters []proxytypes.GroupCounter,
 	groupIDUpdates <-chan string,
 	antreaPolicyEnabled bool,
@@ -143,7 +144,7 @@ func NewNetworkPolicyController(antreaClientGetter agent.AntreaClientProvider,
 		}
 	}
 	c.reconciler = newReconciler(ofClient, ifaceStore, idAllocator, c.fqdnController, groupCounters, v4Enabled, v6Enabled)
-	c.ruleCache = newRuleCache(c.enqueueRule, entityUpdates, groupIDUpdates)
+	c.ruleCache = newRuleCache(c.enqueueRule, podUpdateSubscriber, groupIDUpdates)
 	if statusManagerEnabled {
 		c.statusManager = newStatusController(antreaClientGetter, nodeName, c.ruleCache)
 	}
diff --git a/pkg/agent/types/networkpolicy.go b/pkg/agent/types/networkpolicy.go
index 38d6e81d3..4bac731c2 100644
--- a/pkg/agent/types/networkpolicy.go
+++ b/pkg/agent/types/networkpolicy.go
@@ -149,11 +149,3 @@ type BitRange struct {
 	Value uint16
 	Mask  *uint16
 }
-
-// EntityReference represents a reference to either a Pod or an ExternalEntity.
-type EntityReference struct {
-	// Pod maintains the reference to the Pod.
-	Pod *v1beta2.PodReference
-	// ExternalEntity maintains the reference to the ExternalEntity.
-	ExternalEntity *v1beta2.ExternalEntityReference
-}
diff --git a/pkg/apis/crd/v1alpha1/types.go b/pkg/apis/crd/v1alpha1/types.go
index 38bc5d857..12cc33f2e 100644
--- a/pkg/apis/crd/v1alpha1/types.go
+++ b/pkg/apis/crd/v1alpha1/types.go
@@ -372,7 +372,7 @@ type Rule struct {
 	// or Ports. If this field and To are both empty or missing, this rule matches all
 	// destinations.
 	// +optional
-	ToServices []ServiceReference `json:"toServices,omitempty"`
+	ToServices []NamespacedName `json:"toServices,omitempty"`
 	// Name describes the intention of this rule.
 	// Name should be unique within the policy.
 	// +optional
@@ -477,14 +477,6 @@ type NetworkPolicyPort struct {
 	EndPort *int32 `json:"endPort,omitempty"`
 }
 
-// ServiceReference represents a reference to a v1.Service.
-type ServiceReference struct {
-	// Name of the Service
-	Name string `json:"name"`
-	// Namespace of the Service
-	Namespace string `json:"namespace,omitempty"`
-}
-
 // RuleAction describes the action to be applied on traffic matching a rule.
 type RuleAction string
 
diff --git a/pkg/apis/crd/v1alpha1/zz_generated.deepcopy.go b/pkg/apis/crd/v1alpha1/zz_generated.deepcopy.go
index 20959c31e..345650c66 100644
--- a/pkg/apis/crd/v1alpha1/zz_generated.deepcopy.go
+++ b/pkg/apis/crd/v1alpha1/zz_generated.deepcopy.go
@@ -523,7 +523,7 @@ func (in *Rule) DeepCopyInto(out *Rule) {
 	}
 	if in.ToServices != nil {
 		in, out := &in.ToServices, &out.ToServices
-		*out = make([]ServiceReference, len(*in))
+		*out = make([]NamespacedName, len(*in))
 		copy(*out, *in)
 	}
 	if in.AppliedTo != nil {
@@ -546,22 +546,6 @@ func (in *Rule) DeepCopy() *Rule {
 	return out
 }
 
-// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
-func (in *ServiceReference) DeepCopyInto(out *ServiceReference) {
-	*out = *in
-	return
-}
-
-// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ServiceReference.
-func (in *ServiceReference) DeepCopy() *ServiceReference {
-	if in == nil {
-		return nil
-	}
-	out := new(ServiceReference)
-	in.DeepCopyInto(out)
-	return out
-}
-
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *Source) DeepCopyInto(out *Source) {
 	*out = *in
diff --git a/pkg/apis/crd/v1alpha2/types.go b/pkg/apis/crd/v1alpha2/types.go
index 906c3ea4b..0b04bc13a 100644
--- a/pkg/apis/crd/v1alpha2/types.go
+++ b/pkg/apis/crd/v1alpha2/types.go
@@ -77,14 +77,6 @@ type ExternalEntityList struct {
 	Items []ExternalEntity `json:"items,omitempty"`
 }
 
-// ServiceReference represent reference to a v1.Service.
-type ServiceReference struct {
-	// Name of the Service
-	Name string `json:"name,omitempty"`
-	// Namespace of the Service
-	Namespace string `json:"namespace,omitempty"`
-}
-
 // ClusterGroupReference represent reference to a ClusterGroup.
 type ClusterGroupReference string
 
@@ -131,7 +123,7 @@ type GroupSpec struct {
 	// Select backend Pods of the referred Service.
 	// Cannot be set with any other selector or ipBlock.
 	// +optional
-	ServiceReference *ServiceReference `json:"serviceReference,omitempty"`
+	ServiceReference *v1alpha1.NamespacedName `json:"serviceReference,omitempty"`
 	// Select ExternalEntities from all Namespaces as workloads
 	// in AppliedTo/To/From fields. If set with NamespaceSelector,
 	// ExternalEntities are matched from Namespaces matched by the
diff --git a/pkg/apis/crd/v1alpha2/zz_generated.deepcopy.go b/pkg/apis/crd/v1alpha2/zz_generated.deepcopy.go
index 739d500e7..4334cb9f3 100644
--- a/pkg/apis/crd/v1alpha2/zz_generated.deepcopy.go
+++ b/pkg/apis/crd/v1alpha2/zz_generated.deepcopy.go
@@ -471,7 +471,7 @@ func (in *GroupSpec) DeepCopyInto(out *GroupSpec) {
 	}
 	if in.ServiceReference != nil {
 		in, out := &in.ServiceReference, &out.ServiceReference
-		*out = new(ServiceReference)
+		*out = new(v1alpha1.NamespacedName)
 		**out = **in
 	}
 	if in.ExternalEntitySelector != nil {
@@ -716,22 +716,6 @@ func (in *PodOwner) DeepCopy() *PodOwner {
 	return out
 }
 
-// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
-func (in *ServiceReference) DeepCopyInto(out *ServiceReference) {
-	*out = *in
-	return
-}
-
-// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ServiceReference.
-func (in *ServiceReference) DeepCopy() *ServiceReference {
-	if in == nil {
-		return nil
-	}
-	out := new(ServiceReference)
-	in.DeepCopyInto(out)
-	return out
-}
-
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *StatefulSetOwner) DeepCopyInto(out *StatefulSetOwner) {
 	*out = *in
diff --git a/pkg/apis/crd/v1alpha3/types.go b/pkg/apis/crd/v1alpha3/types.go
index 0bba85530..4913d42b6 100644
--- a/pkg/apis/crd/v1alpha3/types.go
+++ b/pkg/apis/crd/v1alpha3/types.go
@@ -57,7 +57,7 @@ type GroupSpec struct {
 	// Select backend Pods of the referred Service.
 	// Cannot be set with any other selector or ipBlock.
 	// +optional
-	ServiceReference *ServiceReference `json:"serviceReference,omitempty"`
+	ServiceReference *v1alpha1.NamespacedName `json:"serviceReference,omitempty"`
 	// Select ExternalEntities from all Namespaces as workloads
 	// in AppliedTo/To/From fields. If set with NamespaceSelector,
 	// ExternalEntities are matched from Namespaces matched by the
@@ -87,14 +87,6 @@ type GroupStatus struct {
 	Conditions []GroupCondition `json:"conditions,omitempty"`
 }
 
-// ServiceReference represent reference to a v1.Service.
-type ServiceReference struct {
-	// Name of the Service
-	Name string `json:"name,omitempty"`
-	// Namespace of the Service
-	Namespace string `json:"namespace,omitempty"`
-}
-
 // ClusterGroupReference represent reference to a ClusterGroup.
 type ClusterGroupReference string
 
diff --git a/pkg/apis/crd/v1alpha3/zz_generated.deepcopy.go b/pkg/apis/crd/v1alpha3/zz_generated.deepcopy.go
index 2e14a876f..67fe1bd3b 100644
--- a/pkg/apis/crd/v1alpha3/zz_generated.deepcopy.go
+++ b/pkg/apis/crd/v1alpha3/zz_generated.deepcopy.go
@@ -1,7 +1,7 @@
 //go:build !ignore_autogenerated
 // +build !ignore_autogenerated
 
-// Copyright 2021 Antrea Authors
+// 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.
@@ -123,7 +123,7 @@ func (in *GroupSpec) DeepCopyInto(out *GroupSpec) {
 	}
 	if in.ServiceReference != nil {
 		in, out := &in.ServiceReference, &out.ServiceReference
-		*out = new(ServiceReference)
+		*out = new(v1alpha1.NamespacedName)
 		**out = **in
 	}
 	if in.ExternalEntitySelector != nil {
@@ -171,19 +171,3 @@ func (in *GroupStatus) DeepCopy() *GroupStatus {
 	in.DeepCopyInto(out)
 	return out
 }
-
-// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
-func (in *ServiceReference) DeepCopyInto(out *ServiceReference) {
-	*out = *in
-	return
-}
-
-// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ServiceReference.
-func (in *ServiceReference) DeepCopy() *ServiceReference {
-	if in == nil {
-		return nil
-	}
-	out := new(ServiceReference)
-	in.DeepCopyInto(out)
-	return out
-}
diff --git a/pkg/controller/egress/controller.go b/pkg/controller/egress/controller.go
index d7df808c0..0cd162ffe 100644
--- a/pkg/controller/egress/controller.go
+++ b/pkg/controller/egress/controller.go
@@ -358,12 +358,7 @@ func (c *EgressController) syncEgress(key string) error {
 	for _, pod := range pods {
 		// Ignore Pod if it's not scheduled or not running. And Egress does not support HostNetwork Pods, so also ignore
 		// Pod if it's HostNetwork Pod.
-		// TODO: If a Pod is scheduled but not running, it can be included in the EgressGroup so that the agent can
-		// install its SNAT rule right after the Pod's CNI request is processed, which just requires a notification from
-		// CNIServer to the agent's EgressController. However the current notification mechanism (the entityUpdate
-		// channel) allows only single consumer. Once it allows multiple consumers, we can change the condition to
-		// include scheduled Pods that have no IPs.
-		if pod.Spec.NodeName == "" || len(pod.Status.PodIPs) == 0 || pod.Spec.HostNetwork {
+		if pod.Spec.NodeName == "" || pod.Spec.HostNetwork {
 			continue
 		}
 		podNum++
diff --git a/pkg/controller/ipam/validate.go b/pkg/controller/ipam/validate.go
index f584a63b4..abdb04296 100644
--- a/pkg/controller/ipam/validate.go
+++ b/pkg/controller/ipam/validate.go
@@ -15,8 +15,10 @@
 package ipam
 
 import (
+	"bytes"
 	"encoding/json"
 	"fmt"
+	"net"
 	"strings"
 
 	admv1 "k8s.io/api/admission/v1"
@@ -27,7 +29,6 @@ import (
 )
 
 func ValidateIPPool(review *admv1.AdmissionReview) *admv1.AdmissionResponse {
-	var result *metav1.Status
 	var msg string
 	allowed := true
 
@@ -48,16 +49,46 @@ func ValidateIPPool(review *admv1.AdmissionReview) *admv1.AdmissionResponse {
 
 	switch review.Request.Operation {
 	case admv1.Create:
-		// This shouldn't happen with the webhook configuration we include in the Antrea YAML manifests.
 		klog.V(2).Info("Validating CREATE request for IPPool")
-		// Always allow CREATE request.
+
+		// Validate individual ranges
+		for _, r := range newObj.Spec.IPRanges {
+			allowed, msg = validateIPRange(r, newObj.Spec.IPVersion)
+			if !allowed {
+				return validationResult(allowed, msg)
+			}
+		}
+
+		// Validate that the added ipRanges do not overlap
+		for i, r1 := range newObj.Spec.IPRanges[0 : len(newObj.Spec.IPRanges)-1] {
+			for _, r2 := range newObj.Spec.IPRanges[i+1 : len(newObj.Spec.IPRanges)] {
+				if rangesOverlap(r1, r2) {
+					msg = fmt.Sprintf("IPRanges %s overlap", humanReadableIPRanges([]crdv1alpha2.SubnetIPRange{r1, r2}))
+					return validationResult(false, msg)
+				}
+			}
+		}
 	case admv1.Update:
 		klog.V(2).Info("Validating UPDATE request for IPPool")
 		deletedIPRanges := getIPRangeDifference(oldObj.Spec.IPRanges, newObj.Spec.IPRanges)
 		if len(deletedIPRanges) > 0 {
-			allowed = false
-
 			msg = fmt.Sprintf("existing IPRanges %s cannot be updated or deleted", humanReadableIPRanges(deletedIPRanges))
+			return validationResult(false, msg)
+		}
+
+		addedIPRanges := getIPRangeDifference(newObj.Spec.IPRanges, oldObj.Spec.IPRanges)
+		for _, r1 := range addedIPRanges {
+			allowed, msg = validateIPRange(r1, newObj.Spec.IPVersion)
+			if !allowed {
+				return validationResult(allowed, msg)
+			}
+			for _, r2 := range newObj.Spec.IPRanges {
+				if r1 != r2 && rangesOverlap(r1, r2) {
+					msg = fmt.Sprintf("IPRanges %s overlap",
+						humanReadableIPRanges([]crdv1alpha2.SubnetIPRange{r1, r2}))
+					return validationResult(false, msg)
+				}
+			}
 		}
 	case admv1.Delete:
 		klog.V(2).Info("Validating DELETE request for IPPool")
@@ -67,6 +98,12 @@ func ValidateIPPool(review *admv1.AdmissionReview) *admv1.AdmissionResponse {
 		}
 	}
 
+	return validationResult(allowed, msg)
+}
+
+func validationResult(allowed bool, msg string) *admv1.AdmissionResponse {
+	var result *metav1.Status
+
 	if msg != "" {
 		result = &metav1.Status{
 			Message: msg,
@@ -107,6 +144,97 @@ func humanReadableIPRanges(ipRanges []crdv1alpha2.SubnetIPRange) string {
 	return fmt.Sprintf("[%s]", strings.Join(strs, ","))
 }
 
+func rangesOverlap(r1, r2 crdv1alpha2.SubnetIPRange) bool {
+	if r1.CIDR == "" {
+		if r2.CIDR == "" {
+			r1start := net.ParseIP(r1.Start)
+			r1end := net.ParseIP(r1.End)
+			r2start := net.ParseIP(r2.Start)
+			r2end := net.ParseIP(r2.End)
+
+			if ipInRange(r1start, r1end, r2start) || ipInRange(r2start, r2end, r1start) {
+				return true
+			}
+		} else {
+			_, cidr2, _ := net.ParseCIDR(r2.CIDR)
+			r1start := net.ParseIP(r1.Start)
+			r1end := net.ParseIP(r1.End)
+			if cidr2.Contains(r1start) || cidr2.Contains(r1end) {
+				return true
+			}
+			return ipInRange(r1start, r1end, cidr2.IP)
+		}
+	} else {
+		_, cidr1, _ := net.ParseCIDR(r1.CIDR)
+		if r2.CIDR == "" {
+			r2start := net.ParseIP(r2.Start)
+			r2end := net.ParseIP(r2.End)
+			if cidr1.Contains(r2start) || cidr1.Contains(r2end) {
+				return true
+			}
+			return ipInRange(r2start, r2end, cidr1.IP)
+		} else {
+			_, cidr2, _ := net.ParseCIDR(r2.CIDR)
+			return cidr1.Contains(cidr2.IP) || cidr2.Contains(cidr1.IP)
+		}
+	}
+	return false
+}
+
+func ipInRange(rangeStart, rangeEnd, ip net.IP) bool {
+	// Validate that ip is within start and end of range
+	ip16 := ip.To16()
+	return bytes.Compare(ip16, rangeStart.To16()) >= 0 && bytes.Compare(ip16, rangeEnd.To16()) <= 0
+}
+
+func ipVersion(ip net.IP) int {
+	if ip.To4() != nil {
+		return 4
+	}
+	return 6
+}
+
+func validateIPRange(r crdv1alpha2.SubnetIPRange, poolIPVersion int) (bool, string) {
+	// Validate the integrity the IP range:
+	//  Verify that all the IP ranges have the same IP family as the IP pool
+	//  Verify that the gateway IP is reachable from the IP range
+	gateway := net.ParseIP(r.Gateway)
+	var mask net.IPMask
+	if ipVersion(gateway) == 4 {
+		mask = net.CIDRMask(int(r.PrefixLength), 32)
+	} else {
+		mask = net.CIDRMask(int(r.PrefixLength), 128)
+	}
+	netCIDR := net.IPNet{IP: gateway, Mask: mask}
+
+	if r.CIDR != "" {
+		_, cidr, _ := net.ParseCIDR(r.CIDR)
+
+		if !netCIDR.Contains(cidr.IP) {
+			return false, fmt.Sprintf(
+				"Range is invalid. CIDR %s is not contained within subnet %s/%d",
+				r.CIDR, netCIDR.IP.String(), r.PrefixLength)
+		}
+		if ipVersion(cidr.IP) != poolIPVersion {
+			return false, fmt.Sprintf(
+				"Range is invalid. IP version of range %s differs from Pool IP version", r.CIDR)
+		}
+	} else {
+		rStart := net.ParseIP(r.Start)
+		rEnd := net.ParseIP(r.End)
+		if !netCIDR.Contains(rStart) || !netCIDR.Contains(rEnd) {
+			return false, fmt.Sprintf(
+				"Range is invalid. range %s-%s is not contained within subnet %s/%d",
+				r.Start, r.End, netCIDR.IP.String(), r.PrefixLength)
+		}
+		if ipVersion(rStart) != poolIPVersion || ipVersion(rEnd) != poolIPVersion {
+			return false, fmt.Sprintf(
+				"Range is invalid. IP version of range %s-%s differs from Pool IP version", r.Start, r.End)
+		}
+	}
+	return true, ""
+}
+
 func newAdmissionResponseForErr(err error) *admv1.AdmissionResponse {
 	return &admv1.AdmissionResponse{
 		Result: &metav1.Status{
diff --git a/pkg/controller/networkpolicy/crd_utils.go b/pkg/controller/networkpolicy/crd_utils.go
index 91ad94a60..6abf8c41a 100644
--- a/pkg/controller/networkpolicy/crd_utils.go
+++ b/pkg/controller/networkpolicy/crd_utils.go
@@ -141,7 +141,7 @@ func (n *NetworkPolicyController) toNamespacedPeerForCRD(peers []v1alpha1.Networ
 // ServiceReference in ToServices field. For ANP, we will use the
 // defaultNamespace(policy Namespace) as the Namespace of ServiceReference that
 // doesn't set Namespace.
-func (n *NetworkPolicyController) svcRefToPeerForCRD(svcRefs []v1alpha1.ServiceReference, defaultNamespace string) *controlplane.NetworkPolicyPeer {
+func (n *NetworkPolicyController) svcRefToPeerForCRD(svcRefs []v1alpha1.NamespacedName, defaultNamespace string) *controlplane.NetworkPolicyPeer {
 	var controlplaneSvcRefs []controlplane.ServiceReference
 	for _, svcRef := range svcRefs {
 		curNS := defaultNamespace
diff --git a/pkg/flowaggregator/flowaggregator.go b/pkg/flowaggregator/flowaggregator.go
index 95f4b4d9a..95c5f83f3 100644
--- a/pkg/flowaggregator/flowaggregator.go
+++ b/pkg/flowaggregator/flowaggregator.go
@@ -475,7 +475,7 @@ func (fa *flowAggregator) sendFlowKeyRecord(key ipfixintermediate.FlowKey, recor
 	if err != nil {
 		return err
 	}
-	if err = fa.aggregationProcess.ResetStatElementsInRecord(record.Record); err != nil {
+	if err = fa.aggregationProcess.ResetStatAndThroughputElementsInRecord(record.Record); err != nil {
 		return err
 	}
 	klog.V(4).Infof("Data set sent successfully: %d Bytes sent", sentBytes)
diff --git a/pkg/ipfix/ipfix_intermediate.go b/pkg/ipfix/ipfix_intermediate.go
index 0987a32e1..83a54657c 100644
--- a/pkg/ipfix/ipfix_intermediate.go
+++ b/pkg/ipfix/ipfix_intermediate.go
@@ -31,7 +31,7 @@ type IPFIXAggregationProcess interface {
 	ForAllExpiredFlowRecordsDo(callback ipfixintermediate.FlowKeyRecordMapCallBack) error
 	GetExpiryFromExpirePriorityQueue() time.Duration
 	GetRecords(flowKey *ipfixintermediate.FlowKey) []map[string]interface{}
-	ResetStatElementsInRecord(record ipfixentities.Record) error
+	ResetStatAndThroughputElementsInRecord(record ipfixentities.Record) error
 	SetCorrelatedFieldsFilled(record *ipfixintermediate.AggregationFlowRecord)
 	AreCorrelatedFieldsFilled(record ipfixintermediate.AggregationFlowRecord) bool
 	IsAggregatedRecordIPv4(record ipfixintermediate.AggregationFlowRecord) bool
@@ -76,8 +76,8 @@ func (ap *ipfixAggregationProcess) GetRecords(flowKey *ipfixintermediate.FlowKey
 	return ap.AggregationProcess.GetRecords(flowKey)
 }
 
-func (ap *ipfixAggregationProcess) ResetStatElementsInRecord(record ipfixentities.Record) error {
-	return ap.AggregationProcess.ResetStatElementsInRecord(record)
+func (ap *ipfixAggregationProcess) ResetStatAndThroughputElementsInRecord(record ipfixentities.Record) error {
+	return ap.AggregationProcess.ResetStatAndThroughputElementsInRecord(record)
 }
 
 func (ap *ipfixAggregationProcess) SetCorrelatedFieldsFilled(record *ipfixintermediate.AggregationFlowRecord) {
diff --git a/pkg/util/channel/channel.go b/pkg/util/channel/channel.go
new file mode 100644
index 000000000..b10e9b4d4
--- /dev/null
+++ b/pkg/util/channel/channel.go
@@ -0,0 +1,95 @@
+// 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 channel
+
+import (
+	"time"
+
+	"k8s.io/klog/v2"
+)
+
+const (
+	// notifyTimeout is the timeout for failing to publish an event to the channel.
+	notifyTimeout = time.Second
+)
+
+type eventHandler func(string)
+
+type Subscriber interface {
+	// Subscribe registers an eventHandler which will be called when an event is sent to the channel.
+	// It's not thread-safe and it's supposed to be called serially before first event is published.
+	// The eventHandler is supposed to execute quickly and not perform blocking operation. Blocking operation should be
+	// deferred to a routine that is triggered by the eventHandler.
+	Subscribe(h eventHandler)
+}
+
+type Notifier interface {
+	// Notify sends an event to the channel.
+	Notify(string) bool
+}
+
+// SubscribableChannel is different from the Go channel which dispatches every event to only single consumer regardless
+// of the number of consumers. Instead, it dispatches every event to all consumers by calling the eventHandlers they
+// have registered.
+type SubscribableChannel struct {
+	// The name of the channel, used for logging purpose to differentiate multiple channels.
+	name string
+	// eventCh is the channel used for buffering the pending events.
+	eventCh chan string
+	// handlers is a slice of callbacks registered by consumers.
+	handlers []eventHandler
+}
+
+func NewSubscribableChannel(name string, bufferSize int) *SubscribableChannel {
+	n := &SubscribableChannel{
+		name:    name,
+		eventCh: make(chan string, bufferSize),
+	}
+	return n
+}
+
+func (n *SubscribableChannel) Subscribe(h eventHandler) {
+	n.handlers = append(n.handlers, h)
+}
+
+func (n *SubscribableChannel) Notify(e string) bool {
+	timer := time.NewTimer(notifyTimeout)
+	defer timer.Stop()
+	select {
+	case n.eventCh <- e:
+		return true
+	case <-timer.C:
+		// This shouldn't happen as we expect handlers to execute quickly and eventCh can buffer some messages.
+		// If the error is ever seen, either the buffer is too small, or some handlers have improper workload blocking
+		// the event consumption.
+		klog.ErrorS(nil, "Failed to send event to channel, will discard it", "name", n.name, "event", e)
+		return false
+	}
+}
+
+func (n *SubscribableChannel) Run(stopCh <-chan struct{}) {
+	klog.InfoS("Starting SubscribableChannel", "name", n.name)
+	for {
+		select {
+		case <-stopCh:
+			klog.InfoS("Stopping SubscribableChannel", "name", n.name)
+			return
+		case obj := <-n.eventCh:
+			for _, h := range n.handlers {
+				h(obj)
+			}
+		}
+	}
+}
diff --git a/plugins/octant/go.sum b/plugins/octant/go.sum
index e674d8959..20fabbcb1 100644
--- a/plugins/octant/go.sum
+++ b/plugins/octant/go.sum
@@ -1032,7 +1032,7 @@ github.com/vishvananda/netns v0.0.0-20191106174202-0a2b9b5464df/go.mod h1:JP3t17
 github.com/vishvananda/netns v0.0.0-20200728191858-db3c7e526aae/go.mod h1:DD4vA1DwXk04H54A1oHXtwZmA0grkVMdPxx/VGLCah0=
 github.com/vmware-tanzu/octant v0.24.0 h1:PMLU2QG6czSdCl6/xn7lHkHGi+Dv43rxL+AZL6sPJ24=
 github.com/vmware-tanzu/octant v0.24.0/go.mod h1:yK0Nu7wzzvV25/T8nf4NQFMUUA7sLcp0gk96CjOTUiQ=
-github.com/vmware/go-ipfix v0.5.11/go.mod h1:yzbG1rv+yJ8GeMrRm+MDhOV3akygNZUHLhC1pDoD2AY=
+github.com/vmware/go-ipfix v0.5.12/go.mod h1:yzbG1rv+yJ8GeMrRm+MDhOV3akygNZUHLhC1pDoD2AY=
 github.com/willf/bitset v1.1.11-0.20200630133818-d5bec3311243/go.mod h1:RjeCKbqT1RxIR/KWY6phxZiaY1IyutSBfGjNPySAYV4=
 github.com/willf/bitset v1.1.11/go.mod h1:83CECat5yLh5zVOf4P1ErAgKA5UDvKtgyUABdr3+MjI=
 github.com/xdg/scram v0.0.0-20180814205039-7eeb5667e42c/go.mod h1:lB8K/P019DLNhemzwFU4jHLhdvlE6uDZjXFejJXr49I=
diff --git a/third_party/proxy/config/config.go b/third_party/proxy/config/config.go
index 69733c5b0..728c845bc 100644
--- a/third_party/proxy/config/config.go
+++ b/third_party/proxy/config/config.go
@@ -330,7 +330,7 @@ func (c *EndpointSliceConfig) handleAddEndpointSlice(obj interface{}) {
 		return
 	}
 	for _, h := range c.eventHandlers {
-		klog.V(4).Infof("Calling handler.OnEndpointSliceAdd %+v", endpointSlice)
+		klog.V(4).InfoS("Calling handler.OnEndpointSliceAdd", "endpointSlice", endpointSlice)
 		h.OnEndpointSliceAdd(endpointSlice)
 	}
 }
@@ -347,7 +347,7 @@ func (c *EndpointSliceConfig) handleUpdateEndpointSlice(oldObj, newObj interface
 		return
 	}
 	for _, h := range c.eventHandlers {
-		klog.V(4).Infof("Calling handler.OnEndpointSliceUpdate")
+		klog.V(4).InfoS("Calling handler.OnEndpointSliceUpdate", "from", oldEndpointSlice, "to", newEndpointSlice)
 		h.OnEndpointSliceUpdate(oldEndpointSlice, newEndpointSlice)
 	}
 }
@@ -366,7 +366,7 @@ func (c *EndpointSliceConfig) handleDeleteEndpointSlice(obj interface{}) {
 		}
 	}
 	for _, h := range c.eventHandlers {
-		klog.V(4).Infof("Calling handler.OnEndpointsDelete")
+		klog.V(4).InfoS("Calling handler.OnEndpointSliceDelete", "endpointSlice", endpointSlice)
 		h.OnEndpointSliceDelete(endpointSlice)
 	}
 }


Test patch:
diff --git a/ci/jenkins/test-mc.sh b/ci/jenkins/test-mc.sh
index d794146c2..f83c0272e 100755
--- a/ci/jenkins/test-mc.sh
+++ b/ci/jenkins/test-mc.sh
@@ -198,7 +198,7 @@ function deliver_antrea_multicluster {
     ${CLEAN_STALE_IMAGES}
 
     cp -f build/yamls/*.yml $WORKDIR
-    DOCKER_REGISTRY="${DOCKER_REGISTRY}" ./hack/build-antrea-ubuntu-all.sh --pull
+    DOCKER_REGISTRY="${DOCKER_REGISTRY}" ./hack/build-antrea-linux-all.sh --pull
     echo "====== Delivering Antrea to all the Nodes ======"
     docker save -o ${WORKDIR}/antrea-ubuntu.tar $DOCKER_REGISTRY/antrea/antrea-ubuntu:latest
 
diff --git a/ci/jenkins/test-vmc.sh b/ci/jenkins/test-vmc.sh
index 657ed8fb8..768f14f5d 100755
--- a/ci/jenkins/test-vmc.sh
+++ b/ci/jenkins/test-vmc.sh
@@ -359,18 +359,18 @@ function deliver_antrea {
     # Pull images from Dockerhub first then try Harbor.
     for i in `seq 3`; do
         if [[ "$COVERAGE" == true ]]; then
-            VERSION="$CLUSTER" ./hack/build-antrea-ubuntu-all.sh --pull --coverage && break
+            VERSION="$CLUSTER" ./hack/build-antrea-linux-all.sh --pull --coverage && break
         else
-            VERSION="$CLUSTER" ./hack/build-antrea-ubuntu-all.sh --pull && break
+            VERSION="$CLUSTER" ./hack/build-antrea-linux-all.sh --pull && break
         fi
     done
     if [ $? -ne 0 ]; then
         echoerr "Failed to build antrea images with Dockerhub"
         for i in `seq 3`; do
             if [[ "$COVERAGE" == true ]]; then
-                VERSION="$CLUSTER" DOCKER_REGISTRY="${DOCKER_REGISTRY}" ./hack/build-antrea-ubuntu-all.sh --pull --coverage && break
+                VERSION="$CLUSTER" DOCKER_REGISTRY="${DOCKER_REGISTRY}" ./hack/build-antrea-linux-all.sh --pull --coverage && break
             else
-                VERSION="$CLUSTER" DOCKER_REGISTRY="${DOCKER_REGISTRY}" ./hack/build-antrea-ubuntu-all.sh --pull && break
+                VERSION="$CLUSTER" DOCKER_REGISTRY="${DOCKER_REGISTRY}" ./hack/build-antrea-linux-all.sh --pull && break
             fi
         done
         if [ $? -ne 0 ]; then
diff --git a/ci/jenkins/test.sh b/ci/jenkins/test.sh
index 254f4d1f6..282ea2677 100755
--- a/ci/jenkins/test.sh
+++ b/ci/jenkins/test.sh
@@ -276,7 +276,7 @@ function deliver_antrea_windows {
     ${CLEAN_STALE_IMAGES}
     chmod -R g-w build/images/ovs
     chmod -R g-w build/images/base
-    DOCKER_REGISTRY="${DOCKER_REGISTRY}" ./hack/build-antrea-ubuntu-all.sh --pull
+    DOCKER_REGISTRY="${DOCKER_REGISTRY}" ./hack/build-antrea-linux-all.sh --pull
     if [[ "$TESTCASE" == "windows-networkpolicy-process" ]]; then
         make windows-bin
         make antctl-windows
@@ -431,7 +431,7 @@ function deliver_antrea {
     fi
     chmod -R g-w build/images/ovs
     chmod -R g-w build/images/base
-    DOCKER_REGISTRY="${DOCKER_REGISTRY}" ./hack/build-antrea-ubuntu-all.sh --pull
+    DOCKER_REGISTRY="${DOCKER_REGISTRY}" ./hack/build-antrea-linux-all.sh --pull
     make flow-aggregator-image
 
     # Enable verbose log for troubleshooting.
diff --git a/ci/kind/test-e2e-kind.sh b/ci/kind/test-e2e-kind.sh
index 12016de09..2ea92db2b 100755
--- a/ci/kind/test-e2e-kind.sh
+++ b/ci/kind/test-e2e-kind.sh
@@ -129,7 +129,7 @@ COMMON_IMAGES_LIST=("k8s.gcr.io/e2e-test-images/agnhost:2.29" \
                     "projects.registry.vmware.com/library/busybox"  \
                     "projects.registry.vmware.com/antrea/nginx:1.21.6-alpine" \
                     "projects.registry.vmware.com/antrea/perftool" \
-                    "projects.registry.vmware.com/antrea/ipfix-collector:v0.5.11" \
+                    "projects.registry.vmware.com/antrea/ipfix-collector:v0.5.12" \
                     "projects.registry.vmware.com/antrea/wireguard-go:0.0.20210424")
 for image in "${COMMON_IMAGES_LIST[@]}"; do
     for i in `seq 3`; do
diff --git a/multicluster/controllers/multicluster/commonarea/resourceimport_controller_test.go b/multicluster/controllers/multicluster/commonarea/resourceimport_controller_test.go
index ce389376a..3f895a2e2 100644
--- a/multicluster/controllers/multicluster/commonarea/resourceimport_controller_test.go
+++ b/multicluster/controllers/multicluster/commonarea/resourceimport_controller_test.go
@@ -141,31 +141,32 @@ func TestResourceImportReconciler_handleCreateEvent(t *testing.T) {
 		},
 	}
 
-	r := NewResourceImportReconciler(fakeClient, scheme, fakeClient, remoteCluster)
+	r := NewResourceImportReconciler(fakeClient, scheme, fakeClient, localClusterID, remoteCluster)
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			if _, err := r.Reconcile(ctx, tt.req); err != nil {
 				if !assert.Contains(t, err.Error(), "ClusterSetIP is empty") {
 					t.Errorf("ResourceImport Reconciler should handle create event successfully but got error = %v", err)
 				}
-			} else {
-				switch tt.objType {
-				case "Service":
-					svc := &corev1.Service{}
-					if err := fakeClient.Get(ctx, types.NamespacedName{Namespace: "default", Name: "antrea-mc-nginx"}, svc); err != nil {
-						t.Errorf("ResourceImport Reconciler should import a Service successfully but got error = %v", err)
-					}
-					svcImp := &k8smcsapi.ServiceImport{}
-					if err := fakeClient.Get(ctx, types.NamespacedName{Namespace: "default", Name: "nginx"}, svcImp); err != nil {
-						t.Errorf("ResourceImport Reconciler should create a ServiceImport successfully but got error = %v", err)
-					}
-				case "Endpoints":
-					ep := &corev1.Endpoints{}
-					if err := fakeClient.Get(ctx, types.NamespacedName{Namespace: "default", Name: "antrea-mc-nginx"}, ep); err != nil {
-						t.Errorf("ResourceImport Reconciler should import an Endpoint successfully but got error = %v", err)
-					}
+			}
+			switch tt.objType {
+			case "Service":
+				svc := &corev1.Service{}
+				if err := fakeClient.Get(ctx, types.NamespacedName{Namespace: "default", Name: "antrea-mc-nginx"}, svc); err != nil {
+					t.Errorf("ResourceImport Reconciler should import a Service successfully but got error = %v", err)
+				}
+				svcImp := &k8smcsapi.ServiceImport{}
+				if err := fakeClient.Get(ctx, types.NamespacedName{Namespace: "default", Name: "nginx"}, svcImp); err != nil {
+					t.Errorf("ResourceImport Reconciler should create a ServiceImport successfully but got error = %v", err)
+				}
+				checkAnnotation(t, svcImp)
+			case "Endpoints":
+				ep := &corev1.Endpoints{}
+				if err := fakeClient.Get(ctx, types.NamespacedName{Namespace: "default", Name: "antrea-mc-nginx"}, ep); err != nil {
+					t.Errorf("ResourceImport Reconciler should import an Endpoint successfully but got error = %v", err)
 				}
 			}
+
 		})
 	}
 }
@@ -214,7 +215,7 @@ func TestResourceImportReconciler_handleDeleteEvent(t *testing.T) {
 		},
 	}
 
-	r := NewResourceImportReconciler(fakeClient, scheme, fakeClient, remoteCluster)
+	r := NewResourceImportReconciler(fakeClient, scheme, fakeClient, localClusterID, remoteCluster)
 	r.installedResImports.Add(*svcResImport)
 	r.installedResImports.Add(*epResImport)
 
@@ -442,7 +443,7 @@ func TestResourceImportReconciler_handleUpdateEvent(t *testing.T) {
 		},
 	}
 
-	r := NewResourceImportReconciler(fakeClient, scheme, fakeClient, remoteCluster)
+	r := NewResourceImportReconciler(fakeClient, scheme, fakeClient, localClusterID, remoteCluster)
 	r.installedResImports.Add(*svcResImport)
 	r.installedResImports.Add(*epResImport)
 
@@ -471,10 +472,12 @@ func TestResourceImportReconciler_handleUpdateEvent(t *testing.T) {
 							Namespace: tt.resNamespaceName.Namespace,
 							Name:      strings.TrimPrefix(tt.resNamespaceName.Name, common.AntreaMCSPrefix)}, svcImp); err != nil {
 							t.Errorf("ResourceImport Reconciler should update a ServiceImport successfully but got error = %v", err)
+							checkAnnotation(t, svcImp)
 						} else {
 							if !reflect.DeepEqual(svcImp.Spec.Ports, newPorts) {
 								t.Errorf("expected ServiceImport ports are %v but got %v", newPorts, svc.Spec.Ports)
 							}
+							checkAnnotation(t, svcImp)
 						}
 					}
 				case "Endpoints":
@@ -491,3 +494,10 @@ func TestResourceImportReconciler_handleUpdateEvent(t *testing.T) {
 		})
 	}
 }
+
+func checkAnnotation(t *testing.T, svcImport *k8smcsapi.ServiceImport) {
+	id, ok := svcImport.Annotations[common.AntreaMCClusterIDAnnotation]
+	if id != localClusterID || !ok {
+		t.Errorf("latest ServiceImport annotation should be %v but got %v", localClusterID, id)
+	}
+}
diff --git a/multicluster/controllers/multicluster/serviceexport_controller_test.go b/multicluster/controllers/multicluster/serviceexport_controller_test.go
index 72b3b27a5..38623cbc1 100644
--- a/multicluster/controllers/multicluster/serviceexport_controller_test.go
+++ b/multicluster/controllers/multicluster/serviceexport_controller_test.go
@@ -124,10 +124,18 @@ func TestServiceExportReconciler_ExportNotFoundService(t *testing.T) {
 			if *reason != "not_found_service" {
 				t.Errorf("latest ServiceExport status should be 'not_found_service' but got %v", reason)
 			}
+			checkAnnotation(t, newSvcExport)
 		}
 	}
 }
 
+func checkAnnotation(t *testing.T, svcExport *k8smcsv1alpha1.ServiceExport) {
+	id, ok := svcExport.Annotations[common.AntreaMCClusterIDAnnotation]
+	if id != localClusterID || !ok {
+		t.Errorf("latest ServiceExport annotation should be %v but got %v", localClusterID, id)
+	}
+}
+
 func TestServiceExportReconciler_ExportMCSService(t *testing.T) {
 	localClusterID = "cluster-a"
 	leaderNamespace = "default"
@@ -160,6 +168,7 @@ func TestServiceExportReconciler_ExportMCSService(t *testing.T) {
 			if *reason != "imported_service" {
 				t.Errorf("latest ServiceExport status should be 'imported_service' but got %v", reason)
 			}
+			checkAnnotation(t, newSvcExport)
 		}
 	}
 }
@@ -195,6 +204,9 @@ func TestServiceExportReconciler_handleServiceExportCreateEvent(t *testing.T) {
 		if err != nil {
 			t.Errorf("ServiceExport Reconciler should get new Endpoints kind of ResourceExport successfully but got error = %v", err)
 		}
+		newSvcExport := &k8smcsv1alpha1.ServiceExport{}
+		fakeClient.Get(ctx, types.NamespacedName{Namespace: "default", Name: "nginx"}, newSvcExport)
+		checkAnnotation(t, newSvcExport)
 	}
 }
 
@@ -302,6 +314,9 @@ func TestServiceExportReconciler_handleServiceUpdateEvent(t *testing.T) {
 				t.Errorf("expected Endpoints subsets are %v but got %v", expectedSubsets, subsets)
 			}
 		}
+		newSvcExport := &k8smcsv1alpha1.ServiceExport{}
+		fakeClient.Get(ctx, types.NamespacedName{Namespace: "default", Name: "nginx"}, newSvcExport)
+		checkAnnotation(t, newSvcExport)
 	}
 }
 
diff --git a/multicluster/test/e2e/service_test.go b/multicluster/test/e2e/service_test.go
index 6deca699d..be7077231 100644
--- a/multicluster/test/e2e/service_test.go
+++ b/multicluster/test/e2e/service_test.go
@@ -118,7 +118,7 @@ func (data *TestData) testServiceExport(t *testing.T) {
 	anpBuilder = anpBuilder.SetName(multiClusterTestNamespace, "block-west-exported-service").
 		SetPriority(1.0).
 		SetAppliedToGroup([]e2euttils.ANPAppliedToSpec{{PodSelector: map[string]string{"app": "client"}}}).
-		AddToServicesRule([]crdv1alpha1.ServiceReference{{
+		AddToServicesRule([]crdv1alpha1.NamespacedName{{
 			Name:      fmt.Sprintf("antrea-mc-%s", westClusterTestService),
 			Namespace: multiClusterTestNamespace},
 		}, "", nil, crdv1alpha1.RuleActionDrop)
diff --git a/pkg/agent/cniserver/server_test.go b/pkg/agent/cniserver/server_test.go
index bf4cd8a97..afbbd2119 100644
--- a/pkg/agent/cniserver/server_test.go
+++ b/pkg/agent/cniserver/server_test.go
@@ -42,12 +42,12 @@ import (
 	"antrea.io/antrea/pkg/agent/interfacestore"
 	openflowtest "antrea.io/antrea/pkg/agent/openflow/testing"
 	routetest "antrea.io/antrea/pkg/agent/route/testing"
-	antreatypes "antrea.io/antrea/pkg/agent/types"
 	"antrea.io/antrea/pkg/agent/util"
 	cnipb "antrea.io/antrea/pkg/apis/cni/v1beta1"
 	"antrea.io/antrea/pkg/cni"
 	"antrea.io/antrea/pkg/ovs/ovsconfig"
 	ovsconfigtest "antrea.io/antrea/pkg/ovs/ovsconfig/testing"
+	"antrea.io/antrea/pkg/util/channel"
 )
 
 const (
@@ -400,7 +400,7 @@ func TestValidatePrevResult(t *testing.T) {
 		cniConfig.Netns = "invalid_netns"
 		sriovVFDeviceID := ""
 		prevResult.Interfaces = []*current.Interface{hostIface, containerIface}
-		cniServer.podConfigurator, _ = newPodConfigurator(nil, nil, nil, nil, nil, "", false, make(chan antreatypes.EntityReference, 100), nil)
+		cniServer.podConfigurator, _ = newPodConfigurator(nil, nil, nil, nil, nil, "", false, channel.NewSubscribableChannel("PodUpdate", 100), nil)
 		response := cniServer.validatePrevResult(cniConfig.CniCmdArgs, k8sPodArgs, prevResult, sriovVFDeviceID)
 		checkErrorResponse(t, response, cnipb.ErrorCode_CHECK_INTERFACE_FAILURE, "")
 	})
@@ -411,7 +411,7 @@ func TestValidatePrevResult(t *testing.T) {
 		cniConfig.Netns = "invalid_netns"
 		sriovVFDeviceID := "0000:03:00.6"
 		prevResult.Interfaces = []*current.Interface{hostIface, containerIface}
-		cniServer.podConfigurator, _ = newPodConfigurator(nil, nil, nil, nil, nil, "", true, make(chan antreatypes.EntityReference, 100), nil)
+		cniServer.podConfigurator, _ = newPodConfigurator(nil, nil, nil, nil, nil, "", true, channel.NewSubscribableChannel("PodUpdate", 100), nil)
 		response := cniServer.validatePrevResult(cniConfig.CniCmdArgs, k8sPodArgs, prevResult, sriovVFDeviceID)
 		checkErrorResponse(t, response, cnipb.ErrorCode_CHECK_INTERFACE_FAILURE, "")
 	})
@@ -531,7 +531,7 @@ func TestRemoveInterface(t *testing.T) {
 	ifaceStore := interfacestore.NewInterfaceStore()
 	routeMock := routetest.NewMockInterface(controller)
 	gwMAC, _ := net.ParseMAC("00:00:11:11:11:11")
-	podConfigurator, err := newPodConfigurator(mockOVSBridgeClient, mockOFClient, routeMock, ifaceStore, gwMAC, "system", false, make(chan antreatypes.EntityReference, 100), nil)
+	podConfigurator, err := newPodConfigurator(mockOVSBridgeClient, mockOFClient, routeMock, ifaceStore, gwMAC, "system", false, channel.NewSubscribableChannel("PodUpdate", 100), nil)
 	require.Nil(t, err, "No error expected in podConfigurator constructor")
 
 	containerMAC, _ := net.ParseMAC("aa:bb:cc:dd:ee:ff")
diff --git a/pkg/agent/controller/egress/egress_controller_test.go b/pkg/agent/controller/egress/egress_controller_test.go
index 5d43a4227..9c448dd1c 100644
--- a/pkg/agent/controller/egress/egress_controller_test.go
+++ b/pkg/agent/controller/egress/egress_controller_test.go
@@ -44,6 +44,7 @@ import (
 	"antrea.io/antrea/pkg/client/clientset/versioned"
 	fakeversioned "antrea.io/antrea/pkg/client/clientset/versioned/fake"
 	crdinformers "antrea.io/antrea/pkg/client/informers/externalversions"
+	"antrea.io/antrea/pkg/util/channel"
 	"antrea.io/antrea/pkg/util/k8s"
 )
 
@@ -92,6 +93,7 @@ type fakeController struct {
 	crdClient          *fakeversioned.Clientset
 	crdInformerFactory crdinformers.SharedInformerFactory
 	mockIPAssigner     *ipassignertest.MockIPAssigner
+	podUpdateChannel   *channel.SubscribableChannel
 }
 
 func newFakeController(t *testing.T, initObjects []runtime.Object) *fakeController {
@@ -114,6 +116,8 @@ func newFakeController(t *testing.T, initObjects []runtime.Object) *fakeControll
 	addPodInterface(ifaceStore, "ns3", "pod3", 3)
 	addPodInterface(ifaceStore, "ns4", "pod4", 4)
 
+	podUpdateChannel := channel.NewSubscribableChannel("PodUpdate", 100)
+
 	egressController := &EgressController{
 		ofClient:             mockOFClient,
 		routeClient:          mockRouteClient,
@@ -133,6 +137,7 @@ func newFakeController(t *testing.T, initObjects []runtime.Object) *fakeControll
 		egressIPStates:       map[string]*egressIPState{},
 		ipAssigner:           mockIPAssigner,
 	}
+	podUpdateChannel.Subscribe(egressController.processPodUpdate)
 	return &fakeController{
 		EgressController:   egressController,
 		mockController:     controller,
@@ -141,6 +146,7 @@ func newFakeController(t *testing.T, initObjects []runtime.Object) *fakeControll
 		crdClient:          crdClient,
 		crdInformerFactory: crdInformerFactory,
 		mockIPAssigner:     mockIPAssigner,
+		podUpdateChannel:   podUpdateChannel,
 	}
 }
 
@@ -541,6 +547,51 @@ func TestSyncEgress(t *testing.T) {
 	}
 }
 
+func TestPodUpdateShouldSyncEgress(t *testing.T) {
+	egress := &crdv1a2.Egress{
+		ObjectMeta: metav1.ObjectMeta{Name: "egressA", UID: "uidA"},
+		Spec:       crdv1a2.EgressSpec{EgressIP: fakeLocalEgressIP1},
+	}
+	egressGroup := &cpv1b2.EgressGroup{
+		ObjectMeta: metav1.ObjectMeta{Name: "egressA", UID: "uidA"},
+		GroupMembers: []cpv1b2.GroupMember{
+			{Pod: &cpv1b2.PodReference{Name: "pod1", Namespace: "ns1"}},
+			{Pod: &cpv1b2.PodReference{Name: "pendingPod", Namespace: "ns1"}},
+		},
+	}
+	c := newFakeController(t, []runtime.Object{egress})
+	defer c.mockController.Finish()
+	stopCh := make(chan struct{})
+	defer close(stopCh)
+	go c.podUpdateChannel.Run(stopCh)
+	c.crdInformerFactory.Start(stopCh)
+	c.crdInformerFactory.WaitForCacheSync(stopCh)
+
+	c.mockOFClient.EXPECT().InstallSNATMarkFlows(net.ParseIP(fakeLocalEgressIP1), uint32(1))
+	c.mockOFClient.EXPECT().InstallPodSNATFlows(uint32(1), net.ParseIP(fakeLocalEgressIP1), uint32(1))
+	c.mockRouteClient.EXPECT().AddSNATRule(net.ParseIP(fakeLocalEgressIP1), uint32(1))
+	c.mockIPAssigner.EXPECT().UnassignIP(fakeLocalEgressIP1)
+	c.addEgressGroup(egressGroup)
+	require.Equal(t, 1, c.queue.Len())
+	item, _ := c.queue.Get()
+	require.Equal(t, egress.Name, item)
+	require.NoError(t, c.syncEgress(item.(string)))
+	c.queue.Done(item)
+
+	c.mockOFClient.EXPECT().InstallPodSNATFlows(uint32(10), net.ParseIP(fakeLocalEgressIP1), uint32(1))
+	c.mockIPAssigner.EXPECT().UnassignIP(fakeLocalEgressIP1)
+	// Mock CNIServer
+	addPodInterface(c.ifaceStore, "ns1", "pendingPod", 10)
+	c.podUpdateChannel.Notify("ns1/pendingPod")
+	require.NoError(t, wait.PollImmediate(10*time.Millisecond, time.Second, func() (done bool, err error) {
+		return c.queue.Len() == 1, nil
+	}))
+	item, _ = c.queue.Get()
+	require.Equal(t, egress.Name, item)
+	require.NoError(t, c.syncEgress(item.(string)))
+	c.queue.Done(item)
+}
+
 func TestSyncOverlappingEgress(t *testing.T) {
 	egress1 := &crdv1a2.Egress{
 		ObjectMeta: metav1.ObjectMeta{Name: "egressA", UID: "uidA"},
diff --git a/pkg/agent/controller/networkpolicy/cache_test.go b/pkg/agent/controller/networkpolicy/cache_test.go
index 7f47d9fc8..c57b894c3 100644
--- a/pkg/agent/controller/networkpolicy/cache_test.go
+++ b/pkg/agent/controller/networkpolicy/cache_test.go
@@ -25,8 +25,8 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/sets"
 
-	"antrea.io/antrea/pkg/agent/types"
 	"antrea.io/antrea/pkg/apis/controlplane/v1beta2"
+	"antrea.io/antrea/pkg/util/channel"
 )
 
 var (
@@ -260,12 +260,12 @@ func TestRuleCacheAddAddressGroup(t *testing.T) {
 	}
 }
 
-func newFakeRuleCache() (*ruleCache, *dirtyRuleRecorder, chan types.EntityReference) {
+func newFakeRuleCache() (*ruleCache, *dirtyRuleRecorder, *channel.SubscribableChannel) {
 	recorder := newDirtyRuleRecorder()
-	ch := make(chan types.EntityReference, 100)
+	podUpdateChannel := channel.NewSubscribableChannel("PodUpdate", 100)
 	ch2 := make(chan string, 100)
-	c := newRuleCache(recorder.Record, ch, ch2)
-	return c, recorder, ch
+	c := newRuleCache(recorder.Record, podUpdateChannel, ch2)
+	return c, recorder, podUpdateChannel
 }
 
 func TestRuleCacheReplaceAppliedToGroups(t *testing.T) {
@@ -1134,21 +1134,21 @@ func TestRuleCacheProcessPodUpdates(t *testing.T) {
 		name               string
 		rules              []*rule
 		podSetByGroup      map[string]v1beta2.GroupMemberSet
-		podUpdate          types.EntityReference
+		podUpdate          string
 		expectedDirtyRules sets.String
 	}{
 		{
 			"non-matching-group",
 			nil,
 			nil,
-			types.EntityReference{Pod: &v1beta2.PodReference{Name: "foo", Namespace: "bar"}},
+			"bar/foo",
 			sets.NewString(),
 		},
 		{
 			"matching-one-group-affecting-one-rule",
 			[]*rule{rule1, rule2},
 			map[string]v1beta2.GroupMemberSet{"group2": v1beta2.NewGroupMemberSet(newAppliedToGroupMember("pod1", "ns1"))},
-			types.EntityReference{Pod: &v1beta2.PodReference{Name: "pod1", Namespace: "ns1"}},
+			"ns1/pod1",
 			sets.NewString("rule2"),
 		},
 		{
@@ -1158,19 +1158,21 @@ func TestRuleCacheProcessPodUpdates(t *testing.T) {
 				"group1": v1beta2.NewGroupMemberSet(newAppliedToGroupMember("pod1", "ns1")),
 				"group2": v1beta2.NewGroupMemberSet(newAppliedToGroupMember("pod1", "ns1")),
 			},
-			types.EntityReference{Pod: &v1beta2.PodReference{Name: "pod1", Namespace: "ns1"}},
+			"ns1/pod1",
 			sets.NewString("rule1", "rule2"),
 		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			c, recorder, ch := newFakeRuleCache()
+			c, recorder, podUpdateNotifier := newFakeRuleCache()
 			c.appliedToSetByGroup = tt.podSetByGroup
 			for _, rule := range tt.rules {
 				c.rules.Add(rule)
 			}
-			ch <- tt.podUpdate
-
+			stopCh := make(chan struct{})
+			defer close(stopCh)
+			go podUpdateNotifier.Run(stopCh)
+			podUpdateNotifier.Notify(tt.podUpdate)
 			func() {
 				// Drain the channel with 10 ms timeout so we can know it's done.
 				for {
diff --git a/pkg/agent/controller/networkpolicy/networkpolicy_controller_test.go b/pkg/agent/controller/networkpolicy/networkpolicy_controller_test.go
index ac17fd167..b6b9e2034 100644
--- a/pkg/agent/controller/networkpolicy/networkpolicy_controller_test.go
+++ b/pkg/agent/controller/networkpolicy/networkpolicy_controller_test.go
@@ -39,6 +39,7 @@ import (
 	"antrea.io/antrea/pkg/client/clientset/versioned"
 	"antrea.io/antrea/pkg/client/clientset/versioned/fake"
 	"antrea.io/antrea/pkg/querier"
+	"antrea.io/antrea/pkg/util/channel"
 )
 
 const testNamespace = "ns1"
@@ -53,11 +54,11 @@ func (g *antreaClientGetter) GetAntreaClient() (versioned.Interface, error) {
 
 func newTestController() (*Controller, *fake.Clientset, *mockReconciler) {
 	clientset := &fake.Clientset{}
-	ch := make(chan agenttypes.EntityReference, 100)
+	podUpdateChannel := channel.NewSubscribableChannel("PodUpdate", 100)
 	ch2 := make(chan string, 100)
 	groupIDAllocator := openflow.NewGroupAllocator(false)
 	groupCounters := []proxytypes.GroupCounter{proxytypes.NewGroupCounter(groupIDAllocator, ch2)}
-	controller, _ := NewNetworkPolicyController(&antreaClientGetter{clientset}, nil, nil, "node1", ch, groupCounters, ch2,
+	controller, _ := NewNetworkPolicyController(&antreaClientGetter{clientset}, nil, nil, "node1", podUpdateChannel, groupCounters, ch2,
 		true, true, true, true, testAsyncDeleteInterval, "8.8.8.8:53", true, false)
 	reconciler := newMockReconciler()
 	controller.reconciler = reconciler
diff --git a/pkg/agent/controller/networkpolicy/status_controller_test.go b/pkg/agent/controller/networkpolicy/status_controller_test.go
index 37c9173fa..d719e8baa 100644
--- a/pkg/agent/controller/networkpolicy/status_controller_test.go
+++ b/pkg/agent/controller/networkpolicy/status_controller_test.go
@@ -24,8 +24,8 @@ import (
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/wait"
 
-	"antrea.io/antrea/pkg/agent/types"
 	"antrea.io/antrea/pkg/apis/controlplane/v1beta2"
+	"antrea.io/antrea/pkg/util/channel"
 )
 
 const (
@@ -51,7 +51,7 @@ func (c *fakeNetworkPolicyControl) getNetworkPolicyStatus() *v1beta2.NetworkPoli
 }
 
 func newTestStatusController() (*StatusController, *ruleCache, *fakeNetworkPolicyControl) {
-	ruleCache := newRuleCache(func(s string) {}, make(<-chan types.EntityReference), make(chan string, 100))
+	ruleCache := newRuleCache(func(s string) {}, channel.NewSubscribableChannel("PodUpdate", 100), make(chan string, 100))
 	statusControl := &fakeNetworkPolicyControl{}
 	statusController := newStatusController(nil, testNode1, ruleCache)
 	statusController.statusControlInterface = statusControl
diff --git a/pkg/controller/egress/controller_test.go b/pkg/controller/egress/controller_test.go
index 025e3af37..d5880ced7 100644
--- a/pkg/controller/egress/controller_test.go
+++ b/pkg/controller/egress/controller_test.go
@@ -55,7 +55,7 @@ var (
 	podBar1                 = newPod("default", "podBar1", map[string]string{"app": "bar"}, node1, "1.1.1.2", false)
 	podFoo1InOtherNamespace = newPod("other", "podFoo1", map[string]string{"app": "foo"}, node1, "1.1.1.3", false)
 	podUnscheduled          = newPod("default", "podUnscheduled", map[string]string{"app": "foo"}, "", "", false)
-	podNonIP                = newPod("default", "podNonIP", map[string]string{"app": "foo"}, "node1", "", false)
+	podNonIP                = newPod("default", "podNonIP", map[string]string{"app": "foo"}, node1, "", false)
 	podWithHostNetwork      = newPod("default", "podHostNetwork", map[string]string{"app": "bar"}, node1, "172.16.100.1", true)
 	// Fake Namespaces
 	nsDefault = newNamespace("default", map[string]string{"company": "default"})
@@ -188,6 +188,7 @@ func TestAddEgress(t *testing.T) {
 					ObjectMeta: metav1.ObjectMeta{Name: "egressA", UID: "uidA"},
 					GroupMembers: []controlplane.GroupMember{
 						{Pod: &controlplane.PodReference{Name: podFoo1.Name, Namespace: podFoo1.Namespace}},
+						{Pod: &controlplane.PodReference{Name: podNonIP.Name, Namespace: podNonIP.Namespace}},
 					},
 				},
 				node2: {
@@ -219,6 +220,7 @@ func TestAddEgress(t *testing.T) {
 					GroupMembers: []controlplane.GroupMember{
 						{Pod: &controlplane.PodReference{Name: podFoo1.Name, Namespace: podFoo1.Namespace}},
 						{Pod: &controlplane.PodReference{Name: podBar1.Name, Namespace: podBar1.Namespace}},
+						{Pod: &controlplane.PodReference{Name: podNonIP.Name, Namespace: podNonIP.Namespace}},
 					},
 				},
 				node2: {
@@ -250,6 +252,7 @@ func TestAddEgress(t *testing.T) {
 					GroupMembers: []controlplane.GroupMember{
 						{Pod: &controlplane.PodReference{Name: podFoo1.Name, Namespace: podFoo1.Namespace}},
 						{Pod: &controlplane.PodReference{Name: podFoo1InOtherNamespace.Name, Namespace: podFoo1InOtherNamespace.Namespace}},
+						{Pod: &controlplane.PodReference{Name: podNonIP.Name, Namespace: podNonIP.Namespace}},
 					},
 				},
 				node2: {
@@ -282,6 +285,7 @@ func TestAddEgress(t *testing.T) {
 					GroupMembers: []controlplane.GroupMember{
 						{Pod: &controlplane.PodReference{Name: podFoo1.Name, Namespace: podFoo1.Namespace}},
 						{Pod: &controlplane.PodReference{Name: podFoo1InOtherNamespace.Name, Namespace: podFoo1InOtherNamespace.Namespace}},
+						{Pod: &controlplane.PodReference{Name: podNonIP.Name, Namespace: podNonIP.Namespace}},
 					},
 				},
 				node2: {
diff --git a/pkg/controller/ipam/validate_test.go b/pkg/controller/ipam/validate_test.go
index de87ad5e4..8b86d4779 100644
--- a/pkg/controller/ipam/validate_test.go
+++ b/pkg/controller/ipam/validate_test.go
@@ -92,6 +92,102 @@ func TestEgressControllerValidateExternalIPPool(t *testing.T) {
 			},
 			expectedResponse: &admv1.AdmissionResponse{Allowed: true},
 		},
+		{
+			name: "CREATE operation with CIDR partially overlap with IP range should not be allowed",
+			request: &admv1.AdmissionRequest{
+				Name:      "foo",
+				Operation: "CREATE",
+				Object: runtime.RawExtension{Raw: marshal(copyAndMutateIPPool(testIPPool, func(pool *crdv1alpha2.IPPool) {
+					pool.Spec.IPRanges = append(pool.Spec.IPRanges, crdv1alpha2.SubnetIPRange{
+						IPRange: crdv1alpha2.IPRange{
+							CIDR: "192.168.3.0/26",
+						},
+						SubnetInfo: crdv1alpha2.SubnetInfo{
+							Gateway:      "192.168.3.1",
+							PrefixLength: 24,
+						},
+					})
+				}))},
+			},
+			expectedResponse: &admv1.AdmissionResponse{
+				Allowed: false,
+				Result: &metav1.Status{
+					Message: "IPRanges [192.168.3.10-192.168.3.20,192.168.3.0/26] overlap",
+				},
+			},
+		},
+		{
+			name: "CREATE operation with CIDR contained within with IP range should not be allowed",
+			request: &admv1.AdmissionRequest{
+				Name:      "foo",
+				Operation: "CREATE",
+				Object: runtime.RawExtension{Raw: marshal(copyAndMutateIPPool(testIPPool, func(pool *crdv1alpha2.IPPool) {
+					pool.Spec.IPRanges = append(pool.Spec.IPRanges, crdv1alpha2.SubnetIPRange{
+						IPRange: crdv1alpha2.IPRange{
+							CIDR: "192.168.3.12/30",
+						},
+						SubnetInfo: crdv1alpha2.SubnetInfo{
+							Gateway:      "192.168.3.13",
+							PrefixLength: 24,
+						},
+					})
+				}))},
+			},
+			expectedResponse: &admv1.AdmissionResponse{
+				Allowed: false,
+				Result: &metav1.Status{
+					Message: "IPRanges [192.168.3.10-192.168.3.20,192.168.3.12/30] overlap",
+				},
+			},
+		},
+		{
+			name: "CREATE operation with mixed IP version should not be allowed",
+			request: &admv1.AdmissionRequest{
+				Name:      "foo",
+				Operation: "CREATE",
+				Object: runtime.RawExtension{Raw: marshal(copyAndMutateIPPool(testIPPool, func(pool *crdv1alpha2.IPPool) {
+					pool.Spec.IPRanges = append(pool.Spec.IPRanges, crdv1alpha2.SubnetIPRange{
+						IPRange: crdv1alpha2.IPRange{
+							CIDR: "10:2400::0/96",
+						},
+						SubnetInfo: crdv1alpha2.SubnetInfo{
+							Gateway:      "10:2400::01",
+							PrefixLength: 64,
+						},
+					})
+				}))},
+			},
+			expectedResponse: &admv1.AdmissionResponse{
+				Allowed: false,
+				Result: &metav1.Status{
+					Message: "Range is invalid. IP version of range 10:2400::0/96 differs from Pool IP version",
+				},
+			},
+		},
+		{
+			name: "CREATE operation with bad gateway not be allowed",
+			request: &admv1.AdmissionRequest{
+				Name:      "foo",
+				Operation: "CREATE",
+				Object: runtime.RawExtension{Raw: marshal(copyAndMutateIPPool(testIPPool, func(pool *crdv1alpha2.IPPool) {
+					pool.Spec.IPRanges = append(pool.Spec.IPRanges, crdv1alpha2.SubnetIPRange{
+						IPRange: crdv1alpha2.IPRange{
+							CIDR: "192.168.10.0/26",
+						},
+						SubnetInfo: crdv1alpha2.SubnetInfo{
+							Gateway:      "192.168.1.1",
+							PrefixLength: 24,
+						},
+					})
+				}))},
+			},
+			expectedResponse: &admv1.AdmissionResponse{
+				Allowed: false,
+				Result: &metav1.Status{
+					Message: "Range is invalid. CIDR 192.168.10.0/26 is not contained within subnet 192.168.1.1/24",
+				},
+			},
+		},
 		{
 			name: "Deleting IPRange should not be allowed",
 			request: &admv1.AdmissionRequest{
@@ -146,6 +242,32 @@ func TestEgressControllerValidateExternalIPPool(t *testing.T) {
 			},
 			expectedResponse: &admv1.AdmissionResponse{Allowed: true},
 		},
+		{
+			name: "Adding overlapping IPRange should not be allowed",
+			request: &admv1.AdmissionRequest{
+				Name:      "foo",
+				Operation: "UPDATE",
+				OldObject: runtime.RawExtension{Raw: marshal(testIPPool)},
+				Object: runtime.RawExtension{Raw: marshal(copyAndMutateIPPool(testIPPool, func(pool *crdv1alpha2.IPPool) {
+					pool.Spec.IPRanges = append(pool.Spec.IPRanges, crdv1alpha2.SubnetIPRange{
+						IPRange: crdv1alpha2.IPRange{
+							Start: "192.168.3.10",
+							End:   "192.168.3.30",
+						},
+						SubnetInfo: crdv1alpha2.SubnetInfo{
+							Gateway:      "192.168.3.1",
+							PrefixLength: 24,
+						},
+					})
+				}))},
+			},
+			expectedResponse: &admv1.AdmissionResponse{
+				Allowed: false,
+				Result: &metav1.Status{
+					Message: "IPRanges [192.168.3.10-192.168.3.30,192.168.3.10-192.168.3.20] overlap",
+				},
+			},
+		},
 		{
 			name: "Deleting IPPool in use should not be allowed",
 			request: &admv1.AdmissionRequest{
diff --git a/pkg/controller/networkpolicy/antreanetworkpolicy_test.go b/pkg/controller/networkpolicy/antreanetworkpolicy_test.go
index 562af54cd..efb4f8e1a 100644
--- a/pkg/controller/networkpolicy/antreanetworkpolicy_test.go
+++ b/pkg/controller/networkpolicy/antreanetworkpolicy_test.go
@@ -405,7 +405,7 @@ func TestProcessAntreaNetworkPolicy(t *testing.T) {
 					Priority: p10,
 					Egress: []crdv1alpha1.Rule{
 						{
-							ToServices: []crdv1alpha1.ServiceReference{
+							ToServices: []crdv1alpha1.NamespacedName{
 								{
 									Namespace: "ns5",
 									Name:      "svc1",
diff --git a/pkg/controller/networkpolicy/clustergroup_test.go b/pkg/controller/networkpolicy/clustergroup_test.go
index 68ab0a0bd..adfb7532f 100644
--- a/pkg/controller/networkpolicy/clustergroup_test.go
+++ b/pkg/controller/networkpolicy/clustergroup_test.go
@@ -115,7 +115,7 @@ func TestProcessClusterGroup(t *testing.T) {
 			inputGroup: &crdv1alpha3.ClusterGroup{
 				ObjectMeta: metav1.ObjectMeta{Name: "cgE", UID: "uidE"},
 				Spec: crdv1alpha3.GroupSpec{
-					ServiceReference: &crdv1alpha3.ServiceReference{
+					ServiceReference: &crdv1alpha1.NamespacedName{
 						Name:      "test-svc",
 						Namespace: "test-ns",
 					},
@@ -334,7 +334,7 @@ func TestUpdateClusterGroup(t *testing.T) {
 			updatedGroup: &crdv1alpha3.ClusterGroup{
 				ObjectMeta: metav1.ObjectMeta{Name: "cgA", UID: "uidA"},
 				Spec: crdv1alpha3.GroupSpec{
-					ServiceReference: &crdv1alpha3.ServiceReference{
+					ServiceReference: &crdv1alpha1.NamespacedName{
 						Name:      "test-svc",
 						Namespace: "test-ns",
 					},
diff --git a/pkg/controller/networkpolicy/clusternetworkpolicy_test.go b/pkg/controller/networkpolicy/clusternetworkpolicy_test.go
index 573ac95ea..1dbf8009d 100644
--- a/pkg/controller/networkpolicy/clusternetworkpolicy_test.go
+++ b/pkg/controller/networkpolicy/clusternetworkpolicy_test.go
@@ -920,7 +920,7 @@ func TestProcessClusterNetworkPolicy(t *testing.T) {
 					Priority: p10,
 					Egress: []crdv1alpha1.Rule{
 						{
-							ToServices: []crdv1alpha1.ServiceReference{
+							ToServices: []crdv1alpha1.NamespacedName{
 								{
 									Namespace: "nsA",
 									Name:      "svcA",
@@ -977,7 +977,7 @@ func TestProcessClusterNetworkPolicy(t *testing.T) {
 					Priority: p10,
 					Egress: []crdv1alpha1.Rule{
 						{
-							ToServices: []crdv1alpha1.ServiceReference{
+							ToServices: []crdv1alpha1.NamespacedName{
 								{
 									Namespace: "nsA",
 									Name:      "svcA",
diff --git a/pkg/controller/networkpolicy/networkpolicy_controller_test.go b/pkg/controller/networkpolicy/networkpolicy_controller_test.go
index ba5b3e627..d2a431f1a 100644
--- a/pkg/controller/networkpolicy/networkpolicy_controller_test.go
+++ b/pkg/controller/networkpolicy/networkpolicy_controller_test.go
@@ -37,6 +37,7 @@ import (
 	"k8s.io/client-go/util/workqueue"
 
 	"antrea.io/antrea/pkg/apis/controlplane"
+	"antrea.io/antrea/pkg/apis/crd/v1alpha1"
 	"antrea.io/antrea/pkg/apis/crd/v1alpha2"
 	"antrea.io/antrea/pkg/apis/crd/v1alpha3"
 	"antrea.io/antrea/pkg/apiserver/storage"
@@ -1920,7 +1921,7 @@ func TestAddAndUpdateService(t *testing.T) {
 			Name: "cg-1",
 		},
 		Spec: v1alpha3.GroupSpec{
-			ServiceReference: &v1alpha3.ServiceReference{
+			ServiceReference: &v1alpha1.NamespacedName{
 				Name:      "test-svc-1",
 				Namespace: "test-ns",
 			},
@@ -1931,7 +1932,7 @@ func TestAddAndUpdateService(t *testing.T) {
 			Name: "cg-2",
 		},
 		Spec: v1alpha3.GroupSpec{
-			ServiceReference: &v1alpha3.ServiceReference{
+			ServiceReference: &v1alpha1.NamespacedName{
 				Name:      "test-svc-2",
 				Namespace: "test-ns",
 			},
@@ -2026,7 +2027,7 @@ func TestDeleteService(t *testing.T) {
 			Name: "test-cg",
 		},
 		Spec: v1alpha3.GroupSpec{
-			ServiceReference: &v1alpha3.ServiceReference{
+			ServiceReference: &v1alpha1.NamespacedName{
 				Name:      "test-svc",
 				Namespace: "test-ns",
 			},
diff --git a/pkg/controller/networkpolicy/validate_test.go b/pkg/controller/networkpolicy/validate_test.go
index 8297815b0..e19e8150d 100644
--- a/pkg/controller/networkpolicy/validate_test.go
+++ b/pkg/controller/networkpolicy/validate_test.go
@@ -695,7 +695,7 @@ func TestValidateAntreaPolicy(t *testing.T) {
 									},
 								},
 							},
-							ToServices: []crdv1alpha1.ServiceReference{
+							ToServices: []crdv1alpha1.NamespacedName{
 								{
 									Name:      "foo",
 									Namespace: "bar",
@@ -729,7 +729,7 @@ func TestValidateAntreaPolicy(t *testing.T) {
 									Port: &int80,
 								},
 							},
-							ToServices: []crdv1alpha1.ServiceReference{
+							ToServices: []crdv1alpha1.NamespacedName{
 								{
 									Name:      "foo",
 									Namespace: "bar",
@@ -758,7 +758,7 @@ func TestValidateAntreaPolicy(t *testing.T) {
 					Egress: []crdv1alpha1.Rule{
 						{
 							Action: &allowAction,
-							ToServices: []crdv1alpha1.ServiceReference{
+							ToServices: []crdv1alpha1.NamespacedName{
 								{
 									Name:      "foo",
 									Namespace: "bar",
diff --git a/pkg/flowaggregator/flowaggregator_test.go b/pkg/flowaggregator/flowaggregator_test.go
index 877bb959e..c5622faed 100644
--- a/pkg/flowaggregator/flowaggregator_test.go
+++ b/pkg/flowaggregator/flowaggregator_test.go
@@ -147,7 +147,7 @@ func TestFlowAggregator_sendFlowKeyRecord(t *testing.T) {
 		mockRecord.EXPECT().GetOrderedElementList().Return(elementList)
 		mockDataSet.EXPECT().AddRecord(elementList, templateID).Return(nil)
 		mockIPFIXExpProc.EXPECT().SendSet(mockDataSet).Return(0, nil)
-		mockAggregationProcess.EXPECT().ResetStatElementsInRecord(mockRecord).Return(nil)
+		mockAggregationProcess.EXPECT().ResetStatAndThroughputElementsInRecord(mockRecord).Return(nil)
 		mockAggregationProcess.EXPECT().AreCorrelatedFieldsFilled(*tc.flowRecord).Return(false)
 		emptyStr := make([]byte, 0)
 		sourcePodNameElem, _ := ipfixentities.DecodeAndCreateInfoElementWithValue(ipfixentities.NewInfoElement("sourcePodName", 0, 0, ipfixregistry.AntreaEnterpriseID, 0), emptyStr)
diff --git a/pkg/ipfix/testing/mock_ipfix.go b/pkg/ipfix/testing/mock_ipfix.go
index 25a0fc0f8..b7d2d01fb 100644
--- a/pkg/ipfix/testing/mock_ipfix.go
+++ b/pkg/ipfix/testing/mock_ipfix.go
@@ -1,4 +1,4 @@
-// Copyright 2021 Antrea Authors
+// 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.
@@ -351,18 +351,18 @@ func (mr *MockIPFIXAggregationProcessMockRecorder) IsAggregatedRecordIPv4(arg0 i
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "IsAggregatedRecordIPv4", reflect.TypeOf((*MockIPFIXAggregationProcess)(nil).IsAggregatedRecordIPv4), arg0)
 }
 
-// ResetStatElementsInRecord mocks base method
-func (m *MockIPFIXAggregationProcess) ResetStatElementsInRecord(arg0 entities.Record) error {
+// ResetStatAndThroughputElementsInRecord mocks base method
+func (m *MockIPFIXAggregationProcess) ResetStatAndThroughputElementsInRecord(arg0 entities.Record) error {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "ResetStatElementsInRecord", arg0)
+	ret := m.ctrl.Call(m, "ResetStatAndThroughputElementsInRecord", arg0)
 	ret0, _ := ret[0].(error)
 	return ret0
 }
 
-// ResetStatElementsInRecord indicates an expected call of ResetStatElementsInRecord
-func (mr *MockIPFIXAggregationProcessMockRecorder) ResetStatElementsInRecord(arg0 interface{}) *gomock.Call {
+// ResetStatAndThroughputElementsInRecord indicates an expected call of ResetStatAndThroughputElementsInRecord
+func (mr *MockIPFIXAggregationProcessMockRecorder) ResetStatAndThroughputElementsInRecord(arg0 interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ResetStatElementsInRecord", reflect.TypeOf((*MockIPFIXAggregationProcess)(nil).ResetStatElementsInRecord), arg0)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ResetStatAndThroughputElementsInRecord", reflect.TypeOf((*MockIPFIXAggregationProcess)(nil).ResetStatAndThroughputElementsInRecord), arg0)
 }
 
 // SetCorrelatedFieldsFilled mocks base method
diff --git a/pkg/util/channel/channel_test.go b/pkg/util/channel/channel_test.go
new file mode 100644
index 000000000..2eee503f7
--- /dev/null
+++ b/pkg/util/channel/channel_test.go
@@ -0,0 +1,109 @@
+// 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 channel
+
+import (
+	"fmt"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/apimachinery/pkg/util/wait"
+)
+
+type eventReceiver struct {
+	receivedEvents sets.String
+	mutex          sync.RWMutex
+}
+
+func newEventReceiver() *eventReceiver {
+	return &eventReceiver{
+		receivedEvents: sets.NewString(),
+	}
+}
+
+func (r *eventReceiver) receive(e string) {
+	r.mutex.Lock()
+	defer r.mutex.Unlock()
+	r.receivedEvents.Insert(e)
+}
+
+func (r *eventReceiver) received() sets.String {
+	r.mutex.RLock()
+	defer r.mutex.RUnlock()
+	// Return a copy to prevent race condition
+	return r.receivedEvents.Union(nil)
+}
+
+func TestSubscribe(t *testing.T) {
+	c := NewSubscribableChannel("foo", 100)
+	stopCh := make(chan struct{})
+	defer close(stopCh)
+	go c.Run(stopCh)
+
+	var eventReceivers []*eventReceiver
+	for i := 0; i < 100; i++ {
+		receiver := newEventReceiver()
+		c.Subscribe(receiver.receive)
+		eventReceivers = append(eventReceivers, receiver)
+	}
+
+	desiredEvents := sets.NewString()
+	for i := 0; i < 1000; i++ {
+		e := fmt.Sprintf("event-%d", i)
+		c.Notify(e)
+		desiredEvents.Insert(e)
+	}
+
+	var errReceiver int
+	var errReceivedEvents sets.String
+	assert.NoError(t, wait.PollImmediate(10*time.Millisecond, 100*time.Millisecond, func() (done bool, err error) {
+		for i, r := range eventReceivers {
+			receivedEvents := r.received()
+			if !receivedEvents.Equal(desiredEvents) {
+				errReceiver = i
+				errReceivedEvents = receivedEvents
+				return false, nil
+			}
+		}
+		return true, nil
+	}), "Receiver %d failed to receive all events, expected %d events, got %d events", errReceiver, len(desiredEvents), len(errReceivedEvents))
+}
+
+func TestNotify(t *testing.T) {
+	bufferSize := 100
+	c := NewSubscribableChannel("foo", bufferSize)
+	stopCh := make(chan struct{})
+	defer close(stopCh)
+	// Do not run the channel so first N events should be published successfully and later events should fail.
+	for i := 0; i < bufferSize; i++ {
+		e := fmt.Sprintf("event-%d", i)
+		assert.True(t, c.Notify(e), "Failed to publish event when it doesn't exceed the buffer's capacity")
+	}
+
+	notifyRes := make(chan bool)
+	defer close(notifyRes)
+	go func() {
+		notifyRes <- c.Notify("foo")
+	}()
+	select {
+	case res := <-notifyRes:
+		assert.False(t, res)
+	case <-time.After(notifyTimeout + time.Second):
+		t.Errorf("Notify() didn't return in time")
+	}
+}
diff --git a/test/e2e/antreapolicy_test.go b/test/e2e/antreapolicy_test.go
index 3b2b80317..52dec43a4 100644
--- a/test/e2e/antreapolicy_test.go
+++ b/test/e2e/antreapolicy_test.go
@@ -2467,12 +2467,12 @@ func testToServices(t *testing.T) {
 		services = append(services, ipv6Svc)
 	}
 
-	var svcRefs []crdv1alpha1.ServiceReference
+	var svcRefs []crdv1alpha1.NamespacedName
 	var builtSvcs []*v1.Service
 	for _, service := range services {
 		builtSvc, _ := k8sUtils.CreateOrUpdateService(service)
 		failOnError(waitForResourceReady(service, timeout), t)
-		svcRefs = append(svcRefs, crdv1alpha1.ServiceReference{
+		svcRefs = append(svcRefs, crdv1alpha1.NamespacedName{
 			Name:      service.Name,
 			Namespace: service.Namespace,
 		})
diff --git a/test/e2e/clustergroup_test.go b/test/e2e/clustergroup_test.go
index cbb129bdf..723fb3c87 100644
--- a/test/e2e/clustergroup_test.go
+++ b/test/e2e/clustergroup_test.go
@@ -96,7 +96,7 @@ func testInvalidCGServiceRefWithPodSelector(t *testing.T) {
 	invalidErr := fmt.Errorf("clustergroup created with serviceReference and podSelector")
 	cgName := "svcref-pod-selector"
 	pSel := &metav1.LabelSelector{MatchLabels: map[string]string{"pod": "x"}}
-	svcRef := &crdv1alpha3.ServiceReference{
+	svcRef := &crdv1alpha1.NamespacedName{
 		Namespace: "y",
 		Name:      "test-svc",
 	}
@@ -119,7 +119,7 @@ func testInvalidCGServiceRefWithNSSelector(t *testing.T) {
 	invalidErr := fmt.Errorf("clustergroup created with serviceReference and namespaceSelector")
 	cgName := "svcref-ns-selector"
 	nSel := &metav1.LabelSelector{MatchLabels: map[string]string{"ns": "y"}}
-	svcRef := &crdv1alpha3.ServiceReference{
+	svcRef := &crdv1alpha1.NamespacedName{
 		Namespace: "y",
 		Name:      "test-svc",
 	}
@@ -143,7 +143,7 @@ func testInvalidCGServiceRefWithIPBlock(t *testing.T) {
 	cgName := "ipb-svcref"
 	cidr := "10.0.0.10/32"
 	ipb := []crdv1alpha1.IPBlock{{CIDR: cidr}}
-	svcRef := &crdv1alpha3.ServiceReference{
+	svcRef := &crdv1alpha1.NamespacedName{
 		Namespace: "y",
 		Name:      "test-svc",
 	}
@@ -206,7 +206,7 @@ func testInvalidCGChildGroupWithPodSelector(t *testing.T) {
 func testInvalidCGChildGroupWithServiceReference(t *testing.T) {
 	invalidErr := fmt.Errorf("clustergroup created with childGroups and ServiceReference")
 	cgName := "child-group-svcref"
-	svcRef := &crdv1alpha3.ServiceReference{
+	svcRef := &crdv1alpha1.NamespacedName{
 		Namespace: "y",
 		Name:      "test-svc",
 	}
diff --git a/test/e2e/egress_test.go b/test/e2e/egress_test.go
index 403f0c4d6..163bcb434 100644
--- a/test/e2e/egress_test.go
+++ b/test/e2e/egress_test.go
@@ -161,12 +161,13 @@ ip netns exec %[1]s /agnhost netexec
 				t.Fatalf("Error when waiting for Pod '%s' to be in the Running state", remotePod)
 			}
 
+			serverIPStr := tt.serverIP
+			if utilnet.IsIPv6String(tt.localIP0) {
+				serverIPStr = fmt.Sprintf("[%s]", tt.serverIP)
+			}
+
 			// getClientIP gets the translated client IP by accessing the API that replies the request's client IP.
 			getClientIP := func(pod string) (string, string, error) {
-				serverIPStr := tt.serverIP
-				if utilnet.IsIPv6String(tt.localIP0) {
-					serverIPStr = fmt.Sprintf("[%s]", tt.serverIP)
-				}
 				cmd := []string{"wget", "-T", "3", "-O", "-", fmt.Sprintf("%s:8080/clientip", serverIPStr)}
 				return data.runCommandFromPod(testNamespace, pod, busyboxContainerName, cmd)
 			}
@@ -217,7 +218,7 @@ ip netns exec %[1]s /agnhost netexec
 			assertClientIP(localPod, tt.localIP0, tt.localIP1)
 			assertConnError(remotePod)
 
-			t.Logf("Creating an Egress applying to both Pods")
+			t.Logf("Creating an Egress applying to all e2e Pods")
 			matchExpressions := []metav1.LabelSelectorRequirement{
 				{
 					Key:      "antrea-e2e",
@@ -239,6 +240,27 @@ ip netns exec %[1]s /agnhost netexec
 			})
 			assert.NoError(t, err, "Egress failed to reach expected status")
 
+			t.Log("Checking the client IP of a Pod whose Egress has been created in advance")
+			initialIPChecker := "initial-ip-checker"
+			clientIPStr := egress.Spec.EgressIP
+			if utilnet.IsIPv6String(clientIPStr) {
+				clientIPStr = fmt.Sprintf("[%s]", clientIPStr)
+			}
+			cmd = fmt.Sprintf("wget -T 3 -O - %s:8080/clientip | grep %s:", serverIPStr, clientIPStr)
+			if err := data.createPodOnNode(initialIPChecker, testNamespace, egressNode, busyboxImage, []string{"sh", "-c", cmd}, nil, nil, nil, false, func(pod *v1.Pod) {
+				pod.Spec.RestartPolicy = v1.RestartPolicyNever
+			}); err != nil {
+				t.Fatalf("Failed to create Pod initial-ip-checker: %v", err)
+			}
+			defer data.deletePod(testNamespace, initialIPChecker)
+			_, err = data.podWaitFor(timeout, initialIPChecker, testNamespace, func(pod *v1.Pod) (bool, error) {
+				if pod.Status.Phase == v1.PodFailed {
+					return false, fmt.Errorf("Pod terminated with failure")
+				}
+				return pod.Status.Phase == v1.PodSucceeded, nil
+			})
+			assert.NoError(t, err, "Failed to get expected client IP %s for Pod initial-ip-checker", initialIPChecker)
+
 			t.Log("Updating the Egress's AppliedTo to remotePod only")
 			egress.Spec.AppliedTo = v1alpha2.AppliedTo{
 				PodSelector: &metav1.LabelSelector{
diff --git a/test/e2e/framework.go b/test/e2e/framework.go
index 80762eac6..32bad5fbf 100644
--- a/test/e2e/framework.go
+++ b/test/e2e/framework.go
@@ -120,7 +120,7 @@ const (
 	netshootImage       = "projects.registry.vmware.com/antrea/netshoot:v0.1"
 	nginxImage          = "projects.registry.vmware.com/antrea/nginx:1.21.6-alpine"
 	perftoolImage       = "projects.registry.vmware.com/antrea/perftool"
-	ipfixCollectorImage = "projects.registry.vmware.com/antrea/ipfix-collector:v0.5.11"
+	ipfixCollectorImage = "projects.registry.vmware.com/antrea/ipfix-collector:v0.5.12"
 	ipfixCollectorPort  = "4739"
 
 	nginxLBService = "nginx-loadbalancer"
diff --git a/test/e2e/utils/anpspecbuilder.go b/test/e2e/utils/anpspecbuilder.go
index f209ac9b9..af6279237 100644
--- a/test/e2e/utils/anpspecbuilder.go
+++ b/test/e2e/utils/anpspecbuilder.go
@@ -190,7 +190,7 @@ func (b *AntreaNetworkPolicySpecBuilder) AddEgress(protoc v1.Protocol,
 	return b
 }
 
-func (b *AntreaNetworkPolicySpecBuilder) AddToServicesRule(svcRefs []crdv1alpha1.ServiceReference,
+func (b *AntreaNetworkPolicySpecBuilder) AddToServicesRule(svcRefs []crdv1alpha1.NamespacedName,
 	name string, ruleAppliedToSpecs []ANPAppliedToSpec, action crdv1alpha1.RuleAction) *AntreaNetworkPolicySpecBuilder {
 	var appliedTos []crdv1alpha1.NetworkPolicyPeer
 	for _, at := range ruleAppliedToSpecs {
diff --git a/test/e2e/utils/cgspecbuilder.go b/test/e2e/utils/cgspecbuilder.go
index b4b651a97..68c9b88d0 100644
--- a/test/e2e/utils/cgspecbuilder.go
+++ b/test/e2e/utils/cgspecbuilder.go
@@ -80,7 +80,7 @@ func (b *ClusterGroupV1Alpha2SpecBuilder) SetIPBlocks(ipBlocks []crdv1alpha1.IPB
 }
 
 func (b *ClusterGroupV1Alpha2SpecBuilder) SetServiceReference(svcNS, svcName string) *ClusterGroupV1Alpha2SpecBuilder {
-	svcRef := &crdv1alpha2.ServiceReference{
+	svcRef := &crdv1alpha1.NamespacedName{
 		Namespace: svcNS,
 		Name:      svcName,
 	}
@@ -151,7 +151,7 @@ func (b *ClusterGroupV1Alpha3SpecBuilder) SetIPBlocks(ipBlocks []crdv1alpha1.IPB
 }
 
 func (b *ClusterGroupV1Alpha3SpecBuilder) SetServiceReference(svcNS, svcName string) *ClusterGroupV1Alpha3SpecBuilder {
-	svcRef := &crdv1alpha3.ServiceReference{
+	svcRef := &crdv1alpha1.NamespacedName{
 		Namespace: svcNS,
 		Name:      svcName,
 	}
diff --git a/test/e2e/utils/cnpspecbuilder.go b/test/e2e/utils/cnpspecbuilder.go
index 643d49d3c..1f92f7ba9 100644
--- a/test/e2e/utils/cnpspecbuilder.go
+++ b/test/e2e/utils/cnpspecbuilder.go
@@ -259,7 +259,7 @@ func (b *ClusterNetworkPolicySpecBuilder) AddFQDNRule(fqdn string,
 	return b
 }
 
-func (b *ClusterNetworkPolicySpecBuilder) AddToServicesRule(svcRefs []crdv1alpha1.ServiceReference,
+func (b *ClusterNetworkPolicySpecBuilder) AddToServicesRule(svcRefs []crdv1alpha1.NamespacedName,
 	name string, ruleAppliedToSpecs []ACNPAppliedToSpec, action crdv1alpha1.RuleAction) *ClusterNetworkPolicySpecBuilder {
 	var appliedTos []crdv1alpha1.NetworkPolicyPeer
 	for _, at := range ruleAppliedToSpecs {
diff --git a/test/integration/agent/cniserver_test.go b/test/integration/agent/cniserver_test.go
index 7117f7cb4..28ef31969 100644
--- a/test/integration/agent/cniserver_test.go
+++ b/test/integration/agent/cniserver_test.go
@@ -53,11 +53,11 @@ import (
 	"antrea.io/antrea/pkg/agent/metrics"
 	openflowtest "antrea.io/antrea/pkg/agent/openflow/testing"
 	routetest "antrea.io/antrea/pkg/agent/route/testing"
-	antreatypes "antrea.io/antrea/pkg/agent/types"
 	"antrea.io/antrea/pkg/agent/util"
 	cnimsg "antrea.io/antrea/pkg/apis/cni/v1beta1"
 	"antrea.io/antrea/pkg/ovs/ovsconfig"
 	ovsconfigtest "antrea.io/antrea/pkg/ovs/ovsconfig/testing"
+	"antrea.io/antrea/pkg/util/channel"
 )
 
 const (
@@ -575,7 +575,7 @@ func newTester() *cmdAddDelTester {
 		false,
 		routeMock,
 		tester.networkReadyCh)
-	tester.server.Initialize(ovsServiceMock, ofServiceMock, ifaceStore, make(chan antreatypes.EntityReference, 100), nil)
+	tester.server.Initialize(ovsServiceMock, ofServiceMock, ifaceStore, channel.NewSubscribableChannel("PodUpdate", 100), nil)
 	ctx := context.Background()
 	tester.ctx = ctx
 	return tester
@@ -799,7 +799,7 @@ func TestCNIServerChaining(t *testing.T) {
 			ifaceStore := interfacestore.NewInterfaceStore()
 			ovsServiceMock.EXPECT().IsHardwareOffloadEnabled().Return(false).AnyTimes()
 			ovsServiceMock.EXPECT().GetOVSDatapathType().Return(ovsconfig.OVSDatapathSystem).AnyTimes()
-			err = server.Initialize(ovsServiceMock, ofServiceMock, ifaceStore, make(chan antreatypes.EntityReference, 100), nil)
+			err = server.Initialize(ovsServiceMock, ofServiceMock, ifaceStore, channel.NewSubscribableChannel("PodUpdate", 100), nil)
 			testRequire.Nil(err)
 		}