Repository: antrea-io/antrea
Pull request #3914 (2022-06-19 21:21:49)

StretchedNetworkPolicy Agent Implementation
This PR is based on https://github.com/antrea-io/antrea/pull/3913. 

1. Add implementation of Stretched NetworkPolicy in Antrea agent.
1.1 Add OVS flows with tun_id matching.
1.2 Agent also realizes a security rule using unknown LabelIdentity for each Stretched NetworkPolicy rule.
2. Add implementation of LabelIdentity on datapath
2.1 Pod will load its LabelIdentity in tun_id in ClassifierFlow.
2.2 Pod Classifier flow will have a different cacheKey for easy updates.
2.3 StretchedNetworkPolicy will sync Pod ClassifierFlow according to Pod/NS/LabelID events.
3. Add UT and E2E tests.


codecov-commenter: # [Codecov](https://codecov.io/gh/antrea-io/antrea/pull/3914?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) Report
> Merging [#3914](https://codecov.io/gh/antrea-io/antrea/pull/3914?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) (0bf305c) into [main](https://codecov.io/gh/antrea-io/antrea/commit/52bb02e0acdbf46c4756df092e9f008a3e49370d?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) (52bb02e) will **decrease** coverage by `30.89%`.
> The diff coverage is `27.83%`.

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

```diff
@@             Coverage Diff             @@
##             main    #3914       +/-   ##
===========================================
- Coverage   63.91%   33.01%   -30.90%     
===========================================
  Files         293      253       -40     
  Lines       43227    38710     -4517     
===========================================
- Hits        27628    12780    -14848     
- Misses      13363    24568    +11205     
+ Partials     2236     1362      -874     
```

| Flag | Coverage Δ | |
|---|---|---|
| kind-e2e-tests | `33.01% <27.83%> (-17.06%)` | :arrow_down: |
| unit-tests | `?` | |

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/3914?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) | Coverage Δ | |
|---|---|---|
| [pkg/agent/config/node\_config.go](https://codecov.io/gh/antrea-io/antrea/pull/3914/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2FnZW50L2NvbmZpZy9ub2RlX2NvbmZpZy5nbw==) | `72.22% <ø> (-27.78%)` | :arrow_down: |
| [pkg/agent/openflow/multicluster.go](https://codecov.io/gh/antrea-io/antrea/pull/3914/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2FnZW50L29wZW5mbG93L211bHRpY2x1c3Rlci5nbw==) | `13.88% <0.00%> (-29.11%)` | :arrow_down: |
| [pkg/agent/openflow/network\_policy.go](https://codecov.io/gh/antrea-io/antrea/pull/3914/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2FnZW50L29wZW5mbG93L25ldHdvcmtfcG9saWN5Lmdv) | `46.33% <0.00%> (-33.19%)` | :arrow_down: |
| [pkg/agent/types/networkpolicy.go](https://codecov.io/gh/antrea-io/antrea/pull/3914/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2FnZW50L3R5cGVzL25ldHdvcmtwb2xpY3kuZ28=) | `59.45% <ø> (-21.63%)` | :arrow_down: |
| [pkg/apis/controlplane/types.go](https://codecov.io/gh/antrea-io/antrea/pull/3914/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2FwaXMvY29udHJvbHBsYW5lL3R5cGVzLmdv) | `0.00% <ø> (ø)` | |
| [...kg/controller/networkpolicy/antreanetworkpolicy.go](https://codecov.io/gh/antrea-io/antrea/pull/3914/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2NvbnRyb2xsZXIvbmV0d29ya3BvbGljeS9hbnRyZWFuZXR3b3JrcG9saWN5Lmdv) | `67.01% <0.00%> (-20.17%)` | :arrow_down: |
| [...g/controller/networkpolicy/clusternetworkpolicy.go](https://codecov.io/gh/antrea-io/antrea/pull/3914/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2NvbnRyb2xsZXIvbmV0d29ya3BvbGljeS9jbHVzdGVybmV0d29ya3BvbGljeS5nbw==) | `9.00% <0.00%> (-60.22%)` | :arrow_down: |
| [...ntroller/networkpolicy/networkpolicy\_controller.go](https://codecov.io/gh/antrea-io/antrea/pull/3914/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2NvbnRyb2xsZXIvbmV0d29ya3BvbGljeS9uZXR3b3JrcG9saWN5X2NvbnRyb2xsZXIuZ28=) | `58.59% <7.69%> (-23.73%)` | :arrow_down: |
| [pkg/controller/labelidentity/label\_group\_index.go](https://codecov.io/gh/antrea-io/antrea/pull/3914/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2NvbnRyb2xsZXIvbGFiZWxpZGVudGl0eS9sYWJlbF9ncm91cF9pbmRleC5nbw==) | `12.41% <12.41%> (ø)` | |
| [.../multicluster/streched\_networkpolicy\_controller.go](https://codecov.io/gh/antrea-io/antrea/pull/3914/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGtnL2FnZW50L211bHRpY2x1c3Rlci9zdHJlY2hlZF9uZXR3b3JrcG9saWN5X2NvbnRyb2xsZXIuZ28=) | `36.01% <36.01%> (ø)` | |
| ... and [219 more](https://codecov.io/gh/antrea-io/antrea/pull/3914/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) | |

lgtm-com[bot]: This pull request **introduces 1 alert** when merging 0bf305c869a30f07615c3867ce5e1df202951309 into 52bb02e0acdbf46c4756df092e9f008a3e49370d - [view on LGTM.com](https://lgtm.com/projects/g/antrea-io/antrea/rev/pr-50a5bf2ac26de034ef2969d52d9333c7b9c37b70)

**new alerts:**

* 1 for Useless assignment to local variable
luolanzone: Hi @GraysonWu  Could you update the summary and commit message so reviewer can learn about this PR with more  details? thanks.
GraysonWu: /test-multicluster-e2e
GraysonWu: /test-multicluster-e2e
GraysonWu: > Hi @GraysonWu Could you update the summary and commit message so reviewer can learn about this PR with more details? thanks.

Updated.
GraysonWu: /test-multicluster-e2e
GraysonWu: /test-multicluster-e2e
GraysonWu: /test-multicluster-e2e
GraysonWu: /test-multicluster-e2e
GraysonWu: /test-multicluster-e2e
GraysonWu: /test-multicluster-e2e
GraysonWu: /test-multicluster-e2e
GraysonWu: /test-multicluster-e2e
GraysonWu: /test-multicluster-e2e
GraysonWu: /test-multicluster-e2e
GraysonWu: /test-multicluster-e2e
GraysonWu: /test-multicluster-e2e
GraysonWu: /test-multicluster-e2e
GraysonWu: /test-multicluster-e2e
GraysonWu: /test-multicluster-e2e
GraysonWu: /test-multicluster-e2e
GraysonWu: /test-multicluster-e2e
GraysonWu: /test-multicluster-e2e
GraysonWu: /test-multicluster-e2e
GraysonWu: /test-multicluster-e2e
GraysonWu: /test-multicluster-e2e
GraysonWu: /test-multicluster-e2e
GraysonWu: /test-multicluster-e2e
GraysonWu: /test-multicluster-e2e
GraysonWu: /test-multicluster-e2e
GraysonWu: /test-multicluster-e2e
GraysonWu: /test-multicluster-e2e
GraysonWu: /test-multicluster-e2e
GraysonWu: /test-multicluster-e2e
GraysonWu: /test-multicluster-e2e
GraysonWu: /test-multicluster-e2e
luolanzone: Hi @wenyingd @hongliangl ,Could you help to take a look at StretchedNetworkPolicy agent part. @GraysonWu  and @Dyanngg  gave a presentation about the details of StretchedNetworkPolicy on community meeting. You can refer to the [pdf](https://downloads.antrea.io/community/multicluster-stretched-netpols-2.pdf) and [recording](https://www.youtube.com/watch?v=X61eEI79Bzc&list=PLuzde2hYeDBdw0BuQCYbYqxzoJYY1hfwv&index=78) to understand the details. 
GraysonWu: /test-multicluster-e2e
GraysonWu: /test-multicluster-e2e
GraysonWu: /test-multicluster-e2e
GraysonWu: /test-multicluster-e2e
GraysonWu: /test-multicluster-e2e
GraysonWu: /test-multicluster-e2e
GraysonWu: /test-multicluster-e2e
GraysonWu: /test-multicluster-e2e
GraysonWu: /test-multicluster-e2e
GraysonWu: /test-multicluster-e2e
GraysonWu: /test-multicluster-e2e
GraysonWu: /test-all
luolanzone: @GraysonWu  could you re-base and resolve conflicts, thanks.
GraysonWu: /test-multicluster-e2e
luolanzone: @jianjuns @tnqn @wenyingd @hongliangl any new comment?
GraysonWu: /test-multicluster-e2e
/test-all
GraysonWu: /test-e2e

Patch:
diff --git a/build/charts/antrea/README.md b/build/charts/antrea/README.md
index 40d703d39..4d655ac6b 100644
--- a/build/charts/antrea/README.md
+++ b/build/charts/antrea/README.md
@@ -85,7 +85,7 @@ Kubernetes: `>= 1.16.0-0`
 | multicast.igmpQueryInterval | string | `"125s"` | The interval at which the antrea-agent sends IGMP queries to Pods. Valid time units are "ns", "us" (or "µs"), "ms", "s", "m", "h". |
 | multicast.multicastInterfaces | list | `[]` | Names of the interfaces on Nodes that are used to forward multicast traffic. |
 | multicluster.enable | bool | `false` | Enable Antrea Multi-cluster Gateway to support cross-cluster traffic. This feature is supported only with encap mode. |
-| multicluster.enableStretchedNetworkPolicy | bool | `false` | Enable Multicluster which allow Antrea-native policies to select peers from other clusters in a ClusterSet. This feature is supported only with encap mode when the tunnel type is Geneve. |
+| multicluster.enableStretchedNetworkPolicy | bool | `false` | Enable StretchedNetworkPolicy which allow Antrea-native policies to select peers from other clusters in a ClusterSet. This feature is supported only with encap mode when the tunnel type is Geneve. |
 | multicluster.namespace | string | `""` | The Namespace where Antrea Multi-cluster Controller is running. The default is antrea-agent's Namespace. |
 | noSNAT | bool | `false` | Whether or not to SNAT (using the Node IP) the egress traffic from a Pod to the external network. |
 | nodeIPAM.clusterCIDRs | list | `[]` | CIDR ranges to use when allocating Pod IP addresses. |
diff --git a/build/charts/antrea/conf/antrea-agent.conf b/build/charts/antrea/conf/antrea-agent.conf
index a9044639b..0a4abd40b 100644
--- a/build/charts/antrea/conf/antrea-agent.conf
+++ b/build/charts/antrea/conf/antrea-agent.conf
@@ -333,4 +333,7 @@ multicluster:
 # The Namespace where Antrea Multi-cluster Controller is running.
 # The default is antrea-agent's Namespace.
   namespace: {{ .namespace | quote }}
+# Enable StretchedNetworkPolicy which could be enforced on cross-cluster traffic.
+# This feature is supported only with encap mode when the tunnel type is Geneve.
+  enableStretchedNetworkPolicy: {{ .enableStretchedNetworkPolicy }}
 {{- end }}
diff --git a/build/charts/antrea/conf/antrea-controller.conf b/build/charts/antrea/conf/antrea-controller.conf
index aaf6095dd..520b0a0c3 100644
--- a/build/charts/antrea/conf/antrea-controller.conf
+++ b/build/charts/antrea/conf/antrea-controller.conf
@@ -111,7 +111,7 @@ ipsecCSRSigner:
 
 multicluster:
 {{- with .Values.multicluster }}
-  # Enable Multicluster which allow Antrea-native policies to select peers
+  # Enable StretchedNetworkPolicy which allow Antrea-native policies to select peers
   # from other clusters in a ClusterSet.
   enableStretchedNetworkPolicy: {{ .enableStretchedNetworkPolicy }}
 {{- end }}
diff --git a/build/charts/antrea/crds/traceflow.yaml b/build/charts/antrea/crds/traceflow.yaml
index 45d00fbb2..113e98bd4 100644
--- a/build/charts/antrea/crds/traceflow.yaml
+++ b/build/charts/antrea/crds/traceflow.yaml
@@ -189,6 +189,10 @@ spec:
                               type: string
                             tunnelDstIP:
                               type: string
+                            egressIP:
+                              type: string
+                            egress:
+                              type: string
                 capturedPacket:
                   properties:
                     srcIP:
diff --git a/build/charts/antrea/templates/agent/clusterrole.yaml b/build/charts/antrea/templates/agent/clusterrole.yaml
index ab81673b4..384d3d4e8 100644
--- a/build/charts/antrea/templates/agent/clusterrole.yaml
+++ b/build/charts/antrea/templates/agent/clusterrole.yaml
@@ -213,6 +213,7 @@ rules:
     - multicluster.crd.antrea.io
     resources:
     - clusterinfoimports
+    - labelidentities
     verbs:
     - get
     - list
diff --git a/build/charts/antrea/values.yaml b/build/charts/antrea/values.yaml
index 572ba01ca..215dc8904 100644
--- a/build/charts/antrea/values.yaml
+++ b/build/charts/antrea/values.yaml
@@ -307,7 +307,7 @@ multicluster:
   # -- The Namespace where Antrea Multi-cluster Controller is running.
   # The default is antrea-agent's Namespace.
   namespace: ""
-  # -- Enable Multicluster which allow Antrea-native policies to select peers
+  # -- Enable StretchedNetworkPolicy which allow Antrea-native policies to select peers
   # from other clusters in a ClusterSet.
   # This feature is supported only with encap mode when the tunnel type is Geneve.
   enableStretchedNetworkPolicy: false
diff --git a/build/images/base/Dockerfile b/build/images/base/Dockerfile
index ce380849f..f83ec20f7 100644
--- a/build/images/base/Dockerfile
+++ b/build/images/base/Dockerfile
@@ -39,6 +39,8 @@ RUN set -eux; \
 
 FROM antrea/openvswitch:${BUILD_TAG}
 
+ARG SURICATA_VERSION
+
 LABEL maintainer="Antrea <projectantrea-dev@googlegroups.com>"
 LABEL description="An Ubuntu based Docker base image for Antrea."
 
@@ -50,8 +52,9 @@ USER root
 # chmod in the RUN command below instead.
 ADD https://raw.githubusercontent.com/kubernetes-sigs/iptables-wrappers/9e6ce59c864623ea71a6f7d59c35fcb13a919b87/iptables-wrapper-installer.sh /iptables-wrapper-installer.sh
 
-RUN apt-get update && apt-get install -y --no-install-recommends ipset jq inotify-tools && \
-    rm -rf /var/lib/apt/lists/* && \
+RUN apt-get update && apt-get install -y --no-install-recommends ipset jq inotify-tools gpg-agent software-properties-common && \
+    add-apt-repository ppa:oisf/suricata-${SURICATA_VERSION} && apt-get update && apt-get install -y suricata && \
+    apt-get remove -y gpg-agent software-properties-common && apt-get autoremove -y && rm -rf /var/cache/apt/* /var/lib/apt/lists/* && \
     chmod +x /iptables-wrapper-installer.sh && \
     /iptables-wrapper-installer.sh
 
diff --git a/build/images/base/Dockerfile.ubi b/build/images/base/Dockerfile.ubi
index ee03fa1a4..c65e931e8 100644
--- a/build/images/base/Dockerfile.ubi
+++ b/build/images/base/Dockerfile.ubi
@@ -39,11 +39,16 @@ RUN set -eux; \
 
 FROM antrea/openvswitch-ubi:${BUILD_TAG}
 
+ARG SURICATA_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 
+# Skip installing weak dependencies (geolite2-city and geolite2-country) for Suricata as they are not required for Antrea's use case.
+RUN yum install ipset jq yum-plugin-copr -y && \
+    yum copr enable @oisf/suricata-${SURICATA_VERSION} -y && yum install suricata --setopt=install_weak_deps=False -y && \
+    yum remove yum-plugin-copr -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 7f16ce987..36d282345 100755
--- a/build/images/base/build.sh
+++ b/build/images/base/build.sh
@@ -91,6 +91,7 @@ THIS_DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" >/dev/null 2>&1 && pwd )"
 pushd $THIS_DIR > /dev/null
 
 CNI_BINARIES_VERSION=$(head -n 1 ../deps/cni-binaries-version)
+SURICATA_VERSION=$(head -n 1 ../deps/suricata-version)
 
 BUILD_TAG=$(../build-tag.sh)
 
@@ -140,6 +141,7 @@ if [ "$DISTRO" == "ubuntu" ]; then
            --cache-from antrea/base-ubuntu:$BUILD_TAG \
            -t antrea/base-ubuntu:$BUILD_TAG \
            --build-arg CNI_BINARIES_VERSION=$CNI_BINARIES_VERSION \
+           --build-arg SURICATA_VERSION=$SURICATA_VERSION \
            --build-arg BUILD_TAG=$BUILD_TAG .
 elif [ "$DISTRO" == "ubi" ]; then
     docker build $PLATFORM_ARG \
@@ -148,6 +150,7 @@ elif [ "$DISTRO" == "ubi" ]; then
            -t antrea/base-ubi:$BUILD_TAG \
            -f Dockerfile.ubi \
            --build-arg CNI_BINARIES_VERSION=$CNI_BINARIES_VERSION \
+           --build-arg SURICATA_VERSION=$SURICATA_VERSION \
            --build-arg BUILD_TAG=$BUILD_TAG .
 fi
 
diff --git a/build/images/deps/suricata-version b/build/images/deps/suricata-version
new file mode 100644
index 000000000..e0ea36fee
--- /dev/null
+++ b/build/images/deps/suricata-version
@@ -0,0 +1,1 @@
+6.0
diff --git a/build/yamls/antrea-aks.yml b/build/yamls/antrea-aks.yml
index 43e077337..7f4665a27 100644
--- a/build/yamls/antrea-aks.yml
+++ b/build/yamls/antrea-aks.yml
@@ -2508,6 +2508,10 @@ spec:
                               type: string
                             tunnelDstIP:
                               type: string
+                            egressIP:
+                              type: string
+                            egress:
+                              type: string
                 capturedPacket:
                   properties:
                     srcIP:
@@ -3220,6 +3224,9 @@ data:
     # The Namespace where Antrea Multi-cluster Controller is running.
     # The default is antrea-agent's Namespace.
       namespace: ""
+    # Enable StretchedNetworkPolicy which could be enforced on cross-cluster traffic.
+    # This feature is supported only with encap mode when the tunnel type is Geneve.
+      enableStretchedNetworkPolicy: false
   antrea-cni.conflist: |
     {
         "cniVersion":"0.3.0",
@@ -3349,7 +3356,7 @@ data:
       selfSignedCA: true
 
     multicluster:
-      # Enable Multicluster which allow Antrea-native policies to select peers
+      # Enable StretchedNetworkPolicy which allow Antrea-native policies to select peers
       # from other clusters in a ClusterSet.
       enableStretchedNetworkPolicy: false
 ---
@@ -3699,6 +3706,7 @@ rules:
     - multicluster.crd.antrea.io
     resources:
     - clusterinfoimports
+    - labelidentities
     verbs:
     - get
     - list
@@ -4273,7 +4281,7 @@ spec:
         kubectl.kubernetes.io/default-container: antrea-agent
         # Automatically restart Pods with a RollingUpdate if the ConfigMap changes
         # See https://helm.sh/docs/howto/charts_tips_and_tricks/#automatically-roll-deployments
-        checksum/config: 0bfe61fa131f03545550f3a41480c66a3122c1a87390077d700ca01df6371f9a
+        checksum/config: f06bc0519d38f9edaaa4516e816c564429500bf878c777abc41033b346e8edff
       labels:
         app: antrea
         component: antrea-agent
@@ -4514,7 +4522,7 @@ spec:
       annotations:
         # Automatically restart Pod if the ConfigMap changes
         # See https://helm.sh/docs/howto/charts_tips_and_tricks/#automatically-roll-deployments
-        checksum/config: 0bfe61fa131f03545550f3a41480c66a3122c1a87390077d700ca01df6371f9a
+        checksum/config: f06bc0519d38f9edaaa4516e816c564429500bf878c777abc41033b346e8edff
       labels:
         app: antrea
         component: antrea-controller
diff --git a/build/yamls/antrea-crds.yml b/build/yamls/antrea-crds.yml
index fa7b247d8..82e0fcb8c 100644
--- a/build/yamls/antrea-crds.yml
+++ b/build/yamls/antrea-crds.yml
@@ -2483,6 +2483,10 @@ spec:
                               type: string
                             tunnelDstIP:
                               type: string
+                            egressIP:
+                              type: string
+                            egress:
+                              type: string
                 capturedPacket:
                   properties:
                     srcIP:
diff --git a/build/yamls/antrea-eks.yml b/build/yamls/antrea-eks.yml
index aa2b89901..82a84eeca 100644
--- a/build/yamls/antrea-eks.yml
+++ b/build/yamls/antrea-eks.yml
@@ -2508,6 +2508,10 @@ spec:
                               type: string
                             tunnelDstIP:
                               type: string
+                            egressIP:
+                              type: string
+                            egress:
+                              type: string
                 capturedPacket:
                   properties:
                     srcIP:
@@ -3220,6 +3224,9 @@ data:
     # The Namespace where Antrea Multi-cluster Controller is running.
     # The default is antrea-agent's Namespace.
       namespace: ""
+    # Enable StretchedNetworkPolicy which could be enforced on cross-cluster traffic.
+    # This feature is supported only with encap mode when the tunnel type is Geneve.
+      enableStretchedNetworkPolicy: false
   antrea-cni.conflist: |
     {
         "cniVersion":"0.3.0",
@@ -3349,7 +3356,7 @@ data:
       selfSignedCA: true
 
     multicluster:
-      # Enable Multicluster which allow Antrea-native policies to select peers
+      # Enable StretchedNetworkPolicy which allow Antrea-native policies to select peers
       # from other clusters in a ClusterSet.
       enableStretchedNetworkPolicy: false
 ---
@@ -3699,6 +3706,7 @@ rules:
     - multicluster.crd.antrea.io
     resources:
     - clusterinfoimports
+    - labelidentities
     verbs:
     - get
     - list
@@ -4273,7 +4281,7 @@ spec:
         kubectl.kubernetes.io/default-container: antrea-agent
         # Automatically restart Pods with a RollingUpdate if the ConfigMap changes
         # See https://helm.sh/docs/howto/charts_tips_and_tricks/#automatically-roll-deployments
-        checksum/config: 0bfe61fa131f03545550f3a41480c66a3122c1a87390077d700ca01df6371f9a
+        checksum/config: f06bc0519d38f9edaaa4516e816c564429500bf878c777abc41033b346e8edff
       labels:
         app: antrea
         component: antrea-agent
@@ -4516,7 +4524,7 @@ spec:
       annotations:
         # Automatically restart Pod if the ConfigMap changes
         # See https://helm.sh/docs/howto/charts_tips_and_tricks/#automatically-roll-deployments
-        checksum/config: 0bfe61fa131f03545550f3a41480c66a3122c1a87390077d700ca01df6371f9a
+        checksum/config: f06bc0519d38f9edaaa4516e816c564429500bf878c777abc41033b346e8edff
       labels:
         app: antrea
         component: antrea-controller
diff --git a/build/yamls/antrea-gke.yml b/build/yamls/antrea-gke.yml
index 46d0517e2..2055bf2d0 100644
--- a/build/yamls/antrea-gke.yml
+++ b/build/yamls/antrea-gke.yml
@@ -2508,6 +2508,10 @@ spec:
                               type: string
                             tunnelDstIP:
                               type: string
+                            egressIP:
+                              type: string
+                            egress:
+                              type: string
                 capturedPacket:
                   properties:
                     srcIP:
@@ -3220,6 +3224,9 @@ data:
     # The Namespace where Antrea Multi-cluster Controller is running.
     # The default is antrea-agent's Namespace.
       namespace: ""
+    # Enable StretchedNetworkPolicy which could be enforced on cross-cluster traffic.
+    # This feature is supported only with encap mode when the tunnel type is Geneve.
+      enableStretchedNetworkPolicy: false
   antrea-cni.conflist: |
     {
         "cniVersion":"0.3.0",
@@ -3349,7 +3356,7 @@ data:
       selfSignedCA: true
 
     multicluster:
-      # Enable Multicluster which allow Antrea-native policies to select peers
+      # Enable StretchedNetworkPolicy which allow Antrea-native policies to select peers
       # from other clusters in a ClusterSet.
       enableStretchedNetworkPolicy: false
 ---
@@ -3699,6 +3706,7 @@ rules:
     - multicluster.crd.antrea.io
     resources:
     - clusterinfoimports
+    - labelidentities
     verbs:
     - get
     - list
@@ -4273,7 +4281,7 @@ spec:
         kubectl.kubernetes.io/default-container: antrea-agent
         # Automatically restart Pods with a RollingUpdate if the ConfigMap changes
         # See https://helm.sh/docs/howto/charts_tips_and_tricks/#automatically-roll-deployments
-        checksum/config: db1a9feabdabaa45a5a006e8d89bd1b3b4a4e3c67573cb98d5f3630e15d4d757
+        checksum/config: 4c94da583c601f67c4369808e59e884c0ea07d35b3f295e7a9fe3e14b80221a5
       labels:
         app: antrea
         component: antrea-agent
@@ -4513,7 +4521,7 @@ spec:
       annotations:
         # Automatically restart Pod if the ConfigMap changes
         # See https://helm.sh/docs/howto/charts_tips_and_tricks/#automatically-roll-deployments
-        checksum/config: db1a9feabdabaa45a5a006e8d89bd1b3b4a4e3c67573cb98d5f3630e15d4d757
+        checksum/config: 4c94da583c601f67c4369808e59e884c0ea07d35b3f295e7a9fe3e14b80221a5
       labels:
         app: antrea
         component: antrea-controller
diff --git a/build/yamls/antrea-ipsec.yml b/build/yamls/antrea-ipsec.yml
index 870714525..e1ef225c7 100644
--- a/build/yamls/antrea-ipsec.yml
+++ b/build/yamls/antrea-ipsec.yml
@@ -2508,6 +2508,10 @@ spec:
                               type: string
                             tunnelDstIP:
                               type: string
+                            egressIP:
+                              type: string
+                            egress:
+                              type: string
                 capturedPacket:
                   properties:
                     srcIP:
@@ -3233,6 +3237,9 @@ data:
     # The Namespace where Antrea Multi-cluster Controller is running.
     # The default is antrea-agent's Namespace.
       namespace: ""
+    # Enable StretchedNetworkPolicy which could be enforced on cross-cluster traffic.
+    # This feature is supported only with encap mode when the tunnel type is Geneve.
+      enableStretchedNetworkPolicy: false
   antrea-cni.conflist: |
     {
         "cniVersion":"0.3.0",
@@ -3362,7 +3369,7 @@ data:
       selfSignedCA: true
 
     multicluster:
-      # Enable Multicluster which allow Antrea-native policies to select peers
+      # Enable StretchedNetworkPolicy which allow Antrea-native policies to select peers
       # from other clusters in a ClusterSet.
       enableStretchedNetworkPolicy: false
 ---
@@ -3712,6 +3719,7 @@ rules:
     - multicluster.crd.antrea.io
     resources:
     - clusterinfoimports
+    - labelidentities
     verbs:
     - get
     - list
@@ -4286,7 +4294,7 @@ spec:
         kubectl.kubernetes.io/default-container: antrea-agent
         # Automatically restart Pods with a RollingUpdate if the ConfigMap changes
         # See https://helm.sh/docs/howto/charts_tips_and_tricks/#automatically-roll-deployments
-        checksum/config: 1cc89e2ac8e3f6c3c1297fb1d3d8ba1f8eb1f69a7ff915fc23322d9e45237d3f
+        checksum/config: 8a0ea8f57943caf7de83015414c2bb78c4283a7280118ce3a2160e121e8829b9
         checksum/ipsec-secret: d0eb9c52d0cd4311b6d252a951126bf9bea27ec05590bed8a394f0f792dcb2a4
       labels:
         app: antrea
@@ -4572,7 +4580,7 @@ spec:
       annotations:
         # Automatically restart Pod if the ConfigMap changes
         # See https://helm.sh/docs/howto/charts_tips_and_tricks/#automatically-roll-deployments
-        checksum/config: 1cc89e2ac8e3f6c3c1297fb1d3d8ba1f8eb1f69a7ff915fc23322d9e45237d3f
+        checksum/config: 8a0ea8f57943caf7de83015414c2bb78c4283a7280118ce3a2160e121e8829b9
       labels:
         app: antrea
         component: antrea-controller
diff --git a/build/yamls/antrea.yml b/build/yamls/antrea.yml
index ca5bb9745..98c0784b9 100644
--- a/build/yamls/antrea.yml
+++ b/build/yamls/antrea.yml
@@ -2508,6 +2508,10 @@ spec:
                               type: string
                             tunnelDstIP:
                               type: string
+                            egressIP:
+                              type: string
+                            egress:
+                              type: string
                 capturedPacket:
                   properties:
                     srcIP:
@@ -3220,6 +3224,9 @@ data:
     # The Namespace where Antrea Multi-cluster Controller is running.
     # The default is antrea-agent's Namespace.
       namespace: ""
+    # Enable StretchedNetworkPolicy which could be enforced on cross-cluster traffic.
+    # This feature is supported only with encap mode when the tunnel type is Geneve.
+      enableStretchedNetworkPolicy: false
   antrea-cni.conflist: |
     {
         "cniVersion":"0.3.0",
@@ -3349,7 +3356,7 @@ data:
       selfSignedCA: true
 
     multicluster:
-      # Enable Multicluster which allow Antrea-native policies to select peers
+      # Enable StretchedNetworkPolicy which allow Antrea-native policies to select peers
       # from other clusters in a ClusterSet.
       enableStretchedNetworkPolicy: false
 ---
@@ -3699,6 +3706,7 @@ rules:
     - multicluster.crd.antrea.io
     resources:
     - clusterinfoimports
+    - labelidentities
     verbs:
     - get
     - list
@@ -4273,7 +4281,7 @@ spec:
         kubectl.kubernetes.io/default-container: antrea-agent
         # Automatically restart Pods with a RollingUpdate if the ConfigMap changes
         # See https://helm.sh/docs/howto/charts_tips_and_tricks/#automatically-roll-deployments
-        checksum/config: bb8e267e96249bf4d28379cb852eaada9d0e8d20467d58c8e8ab54e33a29fd93
+        checksum/config: 7ef1495672b7b4b2ed3d7729f27af861e3f619bd3343488d45b9c8c3195106cd
       labels:
         app: antrea
         component: antrea-agent
@@ -4513,7 +4521,7 @@ spec:
       annotations:
         # Automatically restart Pod if the ConfigMap changes
         # See https://helm.sh/docs/howto/charts_tips_and_tricks/#automatically-roll-deployments
-        checksum/config: bb8e267e96249bf4d28379cb852eaada9d0e8d20467d58c8e8ab54e33a29fd93
+        checksum/config: 7ef1495672b7b4b2ed3d7729f27af861e3f619bd3343488d45b9c8c3195106cd
       labels:
         app: antrea
         component: antrea-controller
diff --git a/cmd/antrea-agent/agent.go b/cmd/antrea-agent/agent.go
index b55a0612a..fe0de0c5c 100644
--- a/cmd/antrea-agent/agent.go
+++ b/cmd/antrea-agent/agent.go
@@ -134,6 +134,8 @@ func run(o *Options) error {
 	egressEnabled := features.DefaultFeatureGate.Enabled(features.Egress)
 	enableAntreaIPAM := features.DefaultFeatureGate.Enabled(features.AntreaIPAM)
 	enableBridgingMode := enableAntreaIPAM && o.config.EnableBridgingMode
+	enableNodePortLocal := features.DefaultFeatureGate.Enabled(features.NodePortLocal) && o.config.NodePortLocal.Enable
+	enableMulticluster := features.DefaultFeatureGate.Enabled(features.Multicluster) && o.config.Multicluster.Enable
 	// Bridging mode will connect the uplink interface to the OVS bridge.
 	connectUplinkToBridge := enableBridgingMode
 
@@ -281,10 +283,42 @@ func run(o *Options) error {
 		)
 	}
 
+	// podUpdateChannel is a channel for receiving Pod updates from CNIServer and
+	// notifying NetworkPolicyController, StretchedNetworkPolicyController and
+	// EgressController to reconcile rules related to the updated Pods.
+	var podUpdateChannel *channel.SubscribableChannel
+	// externalEntityUpdateChannel is a channel for receiving ExternalEntity updates from ExternalNodeController and
+	// notifying NetworkPolicyController to reconcile rules related to the updated ExternalEntities.
+	var externalEntityUpdateChannel *channel.SubscribableChannel
+	if o.nodeType == config.K8sNode {
+		podUpdateChannel = channel.NewSubscribableChannel("PodUpdate", 100)
+	} else {
+		externalEntityUpdateChannel = channel.NewSubscribableChannel("ExternalEntityUpdate", 100)
+	}
+
+	// Initialize localPodInformer for NPLAgent, AntreaIPAMController,
+	// StretchedNetworkPolicyController, and secondary network controller.
+	var localPodInformer cache.SharedIndexInformer
+	if enableNodePortLocal || enableBridgingMode || enableMulticluster ||
+		features.DefaultFeatureGate.Enabled(features.SecondaryNetwork) ||
+		features.DefaultFeatureGate.Enabled(features.TrafficControl) {
+		listOptions := func(options *metav1.ListOptions) {
+			options.FieldSelector = fields.OneTermEqualSelector("spec.nodeName", nodeConfig.Name).String()
+		}
+		localPodInformer = coreinformers.NewFilteredPodInformer(
+			k8sClient,
+			metav1.NamespaceAll,
+			resyncPeriodDisabled,
+			cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}, // NamespaceIndex is used in NPLController.
+			listOptions,
+		)
+	}
+
 	var mcRouteController *mcroute.MCRouteController
+	var mcStrechedNetworkPolicyController *mcroute.StretchedNetworkPolicyController
 	var mcInformerFactory mcinformers.SharedInformerFactory
 
-	if features.DefaultFeatureGate.Enabled(features.Multicluster) && o.config.Multicluster.Enable {
+	if enableMulticluster {
 		mcNamespace := env.GetPodNamespace()
 		if o.config.Multicluster.Namespace != "" {
 			mcNamespace = o.config.Multicluster.Namespace
@@ -292,6 +326,7 @@ func run(o *Options) error {
 		mcInformerFactory = mcinformers.NewSharedInformerFactory(mcClient, informerDefaultResync)
 		gwInformer := mcInformerFactory.Multicluster().V1alpha1().Gateways()
 		ciImportInformer := mcInformerFactory.Multicluster().V1alpha1().ClusterInfoImports()
+		labelIDInformer := mcInformerFactory.Multicluster().V1alpha1().LabelIdentities()
 		mcRouteController = mcroute.NewMCRouteController(
 			mcClient,
 			gwInformer,
@@ -301,7 +336,18 @@ func run(o *Options) error {
 			ifaceStore,
 			nodeConfig,
 			mcNamespace,
+			o.config.Multicluster.EnableStretchedNetworkPolicy,
 		)
+		if o.config.Multicluster.EnableStretchedNetworkPolicy {
+			mcStrechedNetworkPolicyController = mcroute.NewMCAgentStretchedNetworkPolicyController(
+				ofClient,
+				ifaceStore,
+				localPodInformer,
+				informerFactory.Core().V1().Namespaces(),
+				labelIDInformer,
+				podUpdateChannel,
+			)
+		}
 	}
 	var groupCounters []proxytypes.GroupCounter
 	groupIDUpdates := make(chan string, 100)
@@ -333,19 +379,6 @@ func run(o *Options) error {
 		}
 	}
 
-	// podUpdateChannel is a channel for receiving Pod updates from CNIServer and
-	// notifying NetworkPolicyController and EgressController to reconcile rules
-	// related to the updated Pods.
-	var podUpdateChannel *channel.SubscribableChannel
-	// externalEntityUpdateChannel is a channel for receiving ExternalEntity updates from ExternalNodeController and
-	// notifying NetworkPolicyController to reconcile rules related to the updated ExternalEntities.
-	var externalEntityUpdateChannel *channel.SubscribableChannel
-	if o.nodeType == config.K8sNode {
-		podUpdateChannel = channel.NewSubscribableChannel("PodUpdate", 100)
-	} else {
-		externalEntityUpdateChannel = channel.NewSubscribableChannel("ExternalEntityUpdate", 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
@@ -502,6 +535,7 @@ func run(o *Options) error {
 			traceflowInformer,
 			ofClient,
 			networkPolicyController,
+			egressController,
 			ovsBridgeClient,
 			ifaceStore,
 			networkConfig,
@@ -558,25 +592,6 @@ func run(o *Options) error {
 		networkPolicyController.SetDenyConnStore(flowExporter.GetDenyConnStore())
 	}
 
-	enableNodePortLocal := features.DefaultFeatureGate.Enabled(features.NodePortLocal) && o.config.NodePortLocal.Enable
-
-	// Initialize localPodInformer for NPLAgent, AntreaIPAMController, and secondary network controller.
-	var localPodInformer cache.SharedIndexInformer
-	if enableNodePortLocal || enableBridgingMode ||
-		features.DefaultFeatureGate.Enabled(features.SecondaryNetwork) ||
-		features.DefaultFeatureGate.Enabled(features.TrafficControl) {
-		listOptions := func(options *metav1.ListOptions) {
-			options.FieldSelector = fields.OneTermEqualSelector("spec.nodeName", nodeConfig.Name).String()
-		}
-		localPodInformer = coreinformers.NewFilteredPodInformer(
-			k8sClient,
-			metav1.NamespaceAll,
-			resyncPeriodDisabled,
-			cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}, // NamespaceIndex is used in NPLController.
-			listOptions,
-		)
-	}
-
 	log.StartLogFileNumberMonitor(stopCh)
 
 	if o.nodeType == config.K8sNode {
@@ -730,9 +745,12 @@ func run(o *Options) error {
 		go mcastController.Run(stopCh)
 	}
 
-	if features.DefaultFeatureGate.Enabled(features.Multicluster) && o.config.Multicluster.Enable {
+	if enableMulticluster {
 		mcInformerFactory.Start(stopCh)
 		go mcRouteController.Run(stopCh)
+		if o.config.Multicluster.EnableStretchedNetworkPolicy {
+			go mcStrechedNetworkPolicyController.Run(stopCh)
+		}
 	}
 
 	// statsCollector collects stats and reports to the antrea-controller periodically. For now it's only used for
diff --git a/cmd/antrea-agent/options.go b/cmd/antrea-agent/options.go
index 5c9cce17e..737997e37 100644
--- a/cmd/antrea-agent/options.go
+++ b/cmd/antrea-agent/options.go
@@ -306,6 +306,23 @@ func (o *Options) validateAntreaIPAMConfig() error {
 	return nil
 }
 
+func (o *Options) validateMulticlusterConfig(encapMode config.TrafficEncapModeType) error {
+	if !o.config.Multicluster.Enable {
+		o.config.Multicluster.EnableStretchedNetworkPolicy = false
+		return nil
+	}
+
+	if !features.DefaultFeatureGate.Enabled(features.Multicluster) {
+		return fmt.Errorf("Multicluster can only be enabled when Multicluster feature gate is enabled")
+	}
+
+	if encapMode != config.TrafficEncapModeEncap {
+		// Only Encap mode is supported for Multi-cluster Gateway.
+		return fmt.Errorf("Multicluster is only applicable to the %s mode", config.TrafficEncapModeEncap)
+	}
+	return nil
+}
+
 func (o *Options) setK8sNodeDefaultOptions() {
 	if o.config.CNISocket == "" {
 		o.config.CNISocket = cni.AntreaCNISocketAddr
@@ -453,11 +470,10 @@ func (o *Options) validateK8sNodeOptions() error {
 			}
 		}
 	}
-	if (features.DefaultFeatureGate.Enabled(features.Multicluster) || o.config.Multicluster.Enable) &&
-		encapMode != config.TrafficEncapModeEncap {
-		// Only Encap mode is supported for Multi-cluster feature.
-		return fmt.Errorf("Multicluster is only applicable to the %s mode", config.TrafficEncapModeEncap)
+	if err := o.validateMulticlusterConfig(encapMode); err != nil {
+		return err
 	}
+
 	if features.DefaultFeatureGate.Enabled(features.NodePortLocal) {
 		startPort, endPort, err := parsePortRange(o.config.NodePortLocal.PortRange)
 		if err != nil {
diff --git a/docs/external-node.md b/docs/external-node.md
index 22c9ed648..3bc59733d 100644
--- a/docs/external-node.md
+++ b/docs/external-node.md
@@ -13,6 +13,8 @@
   - [Installation on Linux VM](#installation-on-linux-vm)
     - [Prerequisites on Linux VM](#prerequisites-on-linux-vm)
     - [Installation steps on Linux VM](#installation-steps-on-linux-vm)
+      - [Service Installation](#service-installation)
+      - [Container Installation](#container-installation)
   - [Installation on Windows VM](#installation-on-windows-vm)
     - [Prerequisites on Windows VM](#prerequisites-on-windows-vm)
     - [Installation steps on Windows VM](#installation-steps-on-windows-vm)
@@ -262,7 +264,11 @@ please refer to the [getting-started guide](getting-started.md#open-vswitch).
 
 #### Installation steps on Linux VM
 
-1. Build `antrea-agent` binary in the root of the antrea code tree and copy the
+`Antrea Agent` can be installed as a native service or can be installed in a container.
+
+##### Service Installation
+
+1. Build `antrea-agent` binary in the root of the Antrea code tree and copy the
    `antrea-agent` binary from the `bin` directory to the Linux VM.
 
    ```bash
@@ -277,7 +283,7 @@ please refer to the [getting-started guide](getting-started.md#open-vswitch).
 3. Bootstrap `antrea-agent` using one of these 2 methods:
 
    1. Bootstrap `antrea-agent` using the [installation script](../hack/externalnode/install-vm.sh)
-      as shown below (Ubuntu 18.04 and 20.04 only).
+      as shown below (Ubuntu 18.04 and 20.04, and Red Hat Enterprise Linux 8.4).
 
       ```bash
       ./install-vm.sh --ns vm-ns --bin ./antrea-agent --config ./antrea-agent.conf \
@@ -334,6 +340,66 @@ please refer to the [getting-started guide](getting-started.md#open-vswitch).
       sudo systemctl start antrea-agent
       ```
 
+##### Container Installation
+
+1. `Docker` is used as the container runtime for Linux VMs. The Docker image can be built from source code
+   or can be downloaded from the Antrea repository.
+
+   1. From Source
+
+      Build `antrea-ubuntu` Docker image in the root of the Antrea code tree.
+
+      ```bash
+      make
+      ```
+
+      Note: The image repository name should be `antrea/antrea-ubuntu` and tag should be `latest`.
+
+      Copy the `antrea/antrea-ubuntu:latest` image to the target VM. Please follow
+      the below steps.
+
+      ```bash
+      # Save it in a tar file
+      docker save -o <tar file path in source host machine> antrea/antrea-ubuntu:latest
+
+      # Copy this tar file to the target VM.
+      # Then load that image on the target VM.
+      docker load -i <path to image tar file>
+      ```
+
+   2. Docker Repository
+
+      The released version of `antrea-ubuntu` Docker image can be downloaded from Antrea `Dockerhub`
+      repository. Pick a version from the [list of releases](https://github.com/antrea-io/antrea/releases). For any given
+      release `<TAG>` (e.g. `v1.9.0`), download `antrea-ubuntu` Docker image as follows:
+
+      ```bash
+      docker pull antrea/antrea-ubuntu:<TAG>
+      ```
+
+      The [installation script](../hack/externalnode/install-vm.sh) automatically downloads the specific released
+      version of `antrea-ubuntu` Docker image on VM by specifying the installation argument `--antrea-version`.
+      Also, the script automatically loads that image into Docker. For any given release `<TAG>` (e.g. `v1.9.0`), specify
+      it in the --antrea-version argument as follows.
+
+      ```bash
+         --antrea-version <TAG>
+      ```
+
+2. Copy configuration files to the VM, including [antrea-agent.conf](../build/yamls/externalnode/conf/antrea-agent.conf),
+   which specifies agent configuration parameters;
+   `antrea-agent.antrea.kubeconfig` and `antrea-agent.kubeconfig`, which were
+   generated in steps 4 and 5 of [Prerequisites on Kubernetes cluster](#prerequisites-on-kubernetes-cluster).
+
+3. Bootstrap `antrea-agent` using the [installation script](../hack/externalnode/install-vm.sh)
+   as shown below (Ubuntu 18.04, 20.04, and Rhel 8.4).
+
+    ```bash
+      ./install-vm.sh --ns vm-ns --config ./antrea-agent.conf \
+      --kubeconfig ./antrea-agent.kubeconfig \
+      --antrea-kubeconfig ./antrea-agent.antrea.kubeconfig --containerize --antrea-version v1.9.0
+    ```
+
 ### Installation on Windows VM
 
 #### Prerequisites on Windows VM
diff --git a/hack/externalnode/install-vm.sh b/hack/externalnode/install-vm.sh
index 662ae0f15..14aadaa51 100644
--- a/hack/externalnode/install-vm.sh
+++ b/hack/externalnode/install-vm.sh
@@ -20,7 +20,7 @@ function echoerr {
     >&2 echo "$@"
 }
 
-_usage="Usage: $0 [--ns <Namespace>] [--bin <AntreaAgentSavePath>] [--config <AgentConfigSavePath>] [--kubeconfig <KubeconfigSavePath>] [--antrea-kubeconfig <AntreaKubeconfigSavePath>] [--nodename <ExternalNodeName>] [--ovs-bridge <OVSBridgeName>] [--validate-ovs] [--help|-h]
+_usage="Usage: $0 [--ns <Namespace>] [--bin <AntreaAgentSavePath>] [--config <AgentConfigSavePath>] [--kubeconfig <KubeconfigSavePath>] [--antrea-kubeconfig <AntreaKubeconfigSavePath>] [--nodename <ExternalNodeName>] [--ovs-bridge <OVSBridgeName>] [--containerize] [--validate-ovs] [--antrea-version] [--help|-h]
         --ns                          Namespace to be used by the antrea-agent
         --bin                         Path of the antrea-agent binary
         --config                      Path of the antrea-agent configuration file
@@ -29,6 +29,8 @@ _usage="Usage: $0 [--ns <Namespace>] [--bin <AntreaAgentSavePath>] [--config <Ag
         --nodename                    ExternalNode name to be used by the antrea-agent
         --ovs-bridge                  Specify the OVS bridge name
         --validate-ovs                Validate OVS configuration and performs cleanup when any error is detected.
+        --containerize                Run Antrea Agent in a container.
+        --antrea-version              Specify the version of Antrea.
         --help, -h                    Print this message and exit
 
 Please run the script as sudo user"
@@ -47,26 +49,36 @@ AGENT_LOG_DIR="/var/log/antrea"
 AGENT_CONF_PATH="/etc/antrea"
 OVS_BRIDGE="br-int"
 OVS_VSWITCHD="ovs-vswitchd.service"
+OVS_BIN="start_ovs"
+
+# Antrea agent files
+ANTREA_AGENT_BIN="antrea-agent"
+ANTREA_AGENT_KUBECONFIG="antrea-agent.kubeconfig"
+ANTREA_AGENT_ANTREA_KUBECONFIG="antrea-agent.antrea.kubeconfig"
+
+# Docker variables
+ANTREA_VERSION="latest"
+ANTREA_DOCKER_REGISTRY="antrea/antrea-ubuntu"
+ANTREA_DOCKER_IMAGE="${ANTREA_DOCKER_REGISTRY}:${ANTREA_VERSION}"
 
-# Optional arguments
-VALIDATE_OVS_CONFIG=false
 NODE_NAME="$(hostname)"
+OS_NAME=""
 
-# List of supported OS versions, verified by antrea.
-declare -a SUPPORTED_OS=("Ubuntu 18.04" "Ubuntu 20.04" "Red Hat Enterprise Linux 8.4")
+declare -a SUPPORTED_OS=("ubuntu 18.04" "ubuntu 20.04" "rhel 8.4")
 
+# List of supported OS versions, verified with Antrea.
 check_supported_platform() {
-  echo "Checking supported OS platform"
-  os_name=`grep -Po '^NAME=\K.*' /etc/os-release | sed -e 's/^"//' -e 's/"$//'`
-  os_version=`grep -Po '^VERSION_ID=\K.*' /etc/os-release | sed -e 's/^"//' -e 's/"$//'`
-  dist_version="${os_name} ${os_version}"
-  for ver in "${SUPPORTED_OS[@]}"; do
-      if [ "$ver" == "$dist_version" ]; then
-          return
-      fi
-  done
-  echoerr "Error ${SUPPORTED_OS[*]} are supported"
-  exit 1
+    echo "Checking supported OS platform"
+    OS_NAME=`grep -Po '^ID=\K.*' /etc/os-release | sed -e 's/^"//' -e 's/"$//'`
+    os_version=`grep -Po '^VERSION_ID=\K.*' /etc/os-release | sed -e 's/^"//' -e 's/"$//'`
+    dist_version="${OS_NAME} ${os_version}"
+    for ver in "${SUPPORTED_OS[@]}"; do
+        if [ "$ver" == "$dist_version" ]; then
+            return
+        fi
+    done
+    echoerr "${SUPPORTED_OS[*]} are supported"
+    exit 1
 }
 
 copy_antrea_agent_files() {
@@ -77,35 +89,32 @@ copy_antrea_agent_files() {
     mkdir -p $AGENT_CONF_PATH
     echo "Copying $CONFIG_PATH to $AGENT_CONF_PATH"
     cp "$CONFIG_PATH" $AGENT_CONF_PATH
-
     if [[ ! -f "$KUBECONFIG" ]]; then
         echoerr "Error $KUBECONFIG file not found"
         exit 1
     fi
-
     echo "Copying $KUBECONFIG to $AGENT_CONF_PATH"
-    cp "$KUBECONFIG" "${AGENT_CONF_PATH}/antrea-agent.kubeconfig"
-    chmod 600 "${AGENT_CONF_PATH}/antrea-agent.kubeconfig"
-
+    cp "$KUBECONFIG" "${AGENT_CONF_PATH}/${ANTREA_AGENT_KUBECONFIG}"
+    chmod 600 "${AGENT_CONF_PATH}/${ANTREA_AGENT_KUBECONFIG}"
     if [[ ! -f "$ANTREA_KUBECONFIG" ]]; then
         echoerr "Error $ANTREA_KUBECONFIG file not found"
         exit 1
     fi
     echo "Copying $ANTREA_KUBECONFIG to $AGENT_CONF_PATH"
-    cp "$ANTREA_KUBECONFIG" "${AGENT_CONF_PATH}/antrea-agent.antrea.kubeconfig"
-    chmod 600 "${AGENT_CONF_PATH}/antrea-agent.antrea.kubeconfig"
+    cp "$ANTREA_KUBECONFIG" "${AGENT_CONF_PATH}/${ANTREA_AGENT_ANTREA_KUBECONFIG}"
+    chmod 600 "${AGENT_CONF_PATH}/${ANTREA_AGENT_ANTREA_KUBECONFIG}"
 }
 
 update_antrea_agent_conf() {
-  echo "Updating clientConnection and antreaClientConnection"
-  sed -i "s|kubeconfig: |kubeconfig: $AGENT_CONF_PATH/|g" $AGENT_CONF_PATH/antrea-agent.conf
-  if [[ -z "$AGENT_NAMESPACE" ]]; then
-      AGENT_NAMESPACE="default"
-  fi
-  echo "Updating externalNodeNamespace to $AGENT_NAMESPACE"
-  sed -i "s|#externalNodeNamespace: default|externalNodeNamespace: $AGENT_NAMESPACE|g" $AGENT_CONF_PATH/antrea-agent.conf
-  echo "Updating ovsBridge to $OVS_BRIDGE"
-  sed -i "s|#ovsBridge: br-int|ovsBridge: $OVS_BRIDGE|g" $AGENT_CONF_PATH/antrea-agent.conf
+    echo "Updating clientConnection and antreaClientConnection"
+    sed -i "s|kubeconfig: |kubeconfig: $AGENT_CONF_PATH/|g" $AGENT_CONF_PATH/antrea-agent.conf
+    if [[ -z "$AGENT_NAMESPACE" ]]; then
+        AGENT_NAMESPACE="default"
+    fi
+    echo "Updating externalNodeNamespace to $AGENT_NAMESPACE"
+    sed -i "s|#externalNodeNamespace: default|externalNodeNamespace: $AGENT_NAMESPACE|g" $AGENT_CONF_PATH/antrea-agent.conf
+    echo "Updating ovsBridge to $OVS_BRIDGE"
+    sed -i "s|#ovsBridge: br-int|ovsBridge: $OVS_BRIDGE|g" $AGENT_CONF_PATH/antrea-agent.conf
 }
 
 start_antrea_agent_service() {
@@ -141,17 +150,240 @@ EOF
     systemctl status "$ANTREA_AGENT"
 }
 
+# Generic function to execute command with timeout.
+execute_command() {
+    cmd="$1"
+    timeout="$2"
+    echo "Executing $cmd with timeout $timeout at $(date)"
+    timeout --preserve-status --foreground $timeout $cmd
+    ret=$?
+    if [ $ret -ne 0 ]; then
+        echoerr "$cmd failed with status $ret at $(date), exiting"
+        exit 2
+    fi
+}
+
+# Install Docker as container runtime.
+install_container_runtime() {
+    echo "Installing container runtime on $OS_NAME"
+    case "$OS_NAME" in
+        ubuntu)
+        # Uninstall older versions of Docker.
+        docker_uninstall_packages="docker docker-engine docker.io containerd runc"
+        apt-get remove ${docker_uninstall_packages}
+        cmd="apt-get update"
+        execute_command "$cmd" 1m
+        dependent_packages="ca-certificates curl gnupg lsb-release"
+        cmd="apt-get install -y ${dependent_packages}"
+        execute_command "$cmd" 1m
+        docker_gpg_key="/usr/share/keyrings/docker-archive-keyring.gpg"
+        rm -rf ${docker_gpg_key}
+        docker_download_url="https://download.docker.com/linux/ubuntu"
+        timeout --preserve-status --foreground 10s curl -fsSL ${docker_download_url}/gpg | sudo gpg --dearmor -o ${docker_gpg_key}
+        # Setup a stable repository
+        echo \
+        "deb [arch=$(dpkg --print-architecture) signed-by=${docker_gpg_key}] ${docker_download_url} \
+        $(lsb_release -cs) stable" | sudo tee /etc/apt/sources.list.d/docker.list > /dev/null
+        cmd="apt-get update"
+        execute_command "$cmd" 1m
+        docker_install_packages="docker-ce docker-ce-cli containerd.io docker-compose-plugin"
+        cmd="apt-get install -y ${docker_install_packages}"
+        execute_command "$cmd" 2m
+        ;;
+        rhel)
+        # Uninstall older versions of Docker.
+        docker_uninstall_packages="docker docker-client docker-client-latest docker-common docker-latest docker-latest-logrotate docker-logrotate docker-engine podman runc"
+        sudo yum remove ${docker_unistall_packages}
+        cmd="sudo yum install -y yum-utils"
+        execute_command "$cmd" 1m
+        # We are using CentOs Docker images for Rhel as suggested in official website.
+        docker_download_url="https://download.docker.com/linux/centos/docker-ce.repo"
+        cmd="sudo yum-config-manager --add-repo ${docker_download_url}"
+        execute_command "$cmd" 1m
+        docker_install_packages="docker-ce docker-ce-cli containerd.io docker-compose-plugin"
+        cmd="sudo yum install -y ${docker_install_packages}"
+        execute_command "$cmd" 1m
+        cmd="sudo systemctl start docker"
+        execute_command "$cmd" 1m
+        ;;
+    esac
+}
+
+# Install all pre-requisite packages needed for installation.
+install_prerequisite_packages() {
+    set +eo pipefail
+    # Install Docker, docker-compose only when running services in containers.
+    docker version
+    if [[ "$?" -ne 0 ]]; then
+        install_container_runtime
+    fi
+    set -eo pipefail
+}
+
+# Create docker-compose file with OVS and antrea-agent container specific parameters.
+create_docker_compose_file() {
+    cat << EOF > $AGENT_CONF_PATH/docker-compose.yaml
+version: "3"
+
+services:
+  antrea-ovs:
+    image: $ANTREA_DOCKER_IMAGE
+    container_name: antrea-ovs
+    volumes:
+        - /var/log/openvswitch:/var/log/openvswitch
+        - /var/run/openvswitch:/var/run/openvswitch
+        - /lib/modules:/lib/modules
+    command: $OVS_BIN
+    network_mode: host
+    cap_add:
+        - NET_ADMIN
+        - SYS_NICE
+        - SYS_ADMIN
+        - IPC_LOCK
+    deploy:
+        restart_policy:
+          condition: on-failure
+          delay: 3s
+          max_attempts: 500
+          window: 10s
+
+  antrea-agent:
+    image: $ANTREA_DOCKER_IMAGE
+    container_name: antrea-agent
+    volumes:
+        - /etc/antrea/:/etc/antrea
+        - /var/run/antrea:/var/run/antrea
+        - /var/run/openvswitch:/var/run/openvswitch
+        - /var/log/antrea:/var/log/antrea
+    environment:
+        - NODE_NAME=$NODE_NAME
+    command: $ANTREA_AGENT_BIN --config $AGENT_CONF_PATH/antrea-agent.conf --logtostderr=false --log_file=$AGENT_LOG_DIR/antrea-agent.log
+    network_mode: host
+    privileged: true
+    deploy:
+        restart_policy:
+          condition: on-failure
+          delay: 3s
+          max_attempts: 500
+          window: 10s
+EOF
+}
+
+create_antrea_agent_service_file() {
+    cat >$AGENT_CONF_PATH/antrea-agent.sh << 'EOF'
+#!/bin/bash
+
+AGENT_CONF_PATH="/etc/antrea"
+AGENT_LOG_DIR="/var/log/antrea"
+DOCKER_COMPOSE_YAML="$AGENT_CONF_PATH/docker-compose.yaml"
+
+declare -a OVS_KERNEL_MODULES=("openvswitch")
+declare -a CONTAINERS=("antrea-ovs" "antrea-agent")
+
+load_ovs_modules() {
+    for module in "${OVS_KERNEL_MODULES[@]}"; do
+        lsmod | grep -q $module
+        if [[ $? -ne 0 ]]; then
+            echo "Load $module kernel module"
+            modprobe $module
+            if [ "$?" -ne 0 ]; then
+                echoerr "Error loading kernel module $module"
+            fi
+        fi
+    done
+}
+
+docker info
+if [[ $? -ne 0 ]]; then
+    echo "Starting Docker service"
+    systemctl start docker
+    sleep 2
+fi
+
+key=$1
+case $key in
+    start)
+    load_ovs_modules
+    docker compose -f $DOCKER_COMPOSE_YAML up -d
+    ;;
+    stop)
+    docker compose -f $DOCKER_COMPOSE_YAML stop ${CONTAINERS[*]}
+    ;;
+esac
+EOF
+}
+
+# Create antrea-agent service as a systemd service for monitoring/creating Antrea specific containers.
+setup_antrea_agent_service() {
+    DESCRIPTION="Antrea agent service to monitor OVS and antrea-agent containers"
+    SERVICE_NAME="antrea-agent.service"
+    IS_ACTIVE=$(sudo systemctl is-active $SERVICE_NAME)
+    create_antrea_agent_service_file
+    sudo chmod +x $AGENT_CONF_PATH/antrea-agent.sh
+    if [ "$IS_ACTIVE" == "active" ]; then
+        # restart the service
+        echo "Systemd service $SERVICE_NAME is running"
+        echo "Restarting $SERVICE_NAME service"
+        sudo systemctl restart $SERVICE_NAME
+    else
+        # create service file
+        echo "Creating systemd service file for $SERVICE_NAME"
+        sudo cat > /etc/systemd/system/${SERVICE_NAME} << EOF
+[Unit]
+Description=$DESCRIPTION
+After=network.target
+[Service]
+ExecStart=$AGENT_CONF_PATH/antrea-agent.sh start
+ExecStop=$AGENT_CONF_PATH/antrea-agent.sh stop
+RemainAfterExit=yes
+[Install]
+WantedBy=multi-user.target
+EOF
+        # restart daemon, enable and start service.
+        echo "Reloading daemon and enabling service $SERVICE_NAME"
+        sudo systemctl daemon-reload
+        sudo systemctl enable ${SERVICE_NAME//'.service'/} # remove the extension.
+        sudo systemctl start ${SERVICE_NAME//'.service'/}
+        if [ $? -eq 0 ]; then
+            echo "Antrea-agent service started successfully"
+        else
+            echoerr "Error Antrea-agent service could not be started"
+            exit 1
+        fi
+    fi
+}
+
+# Function to deploy and run Antrea specific containers.
+deploy_antrea_containers() {
+    set +eo pipefail
+    if [ -z "$(docker images -q $ANTREA_DOCKER_IMAGE)" ]; then
+        echo "Downloading the Docker image $ANTREA_DOCKER_IMAGE"
+        docker pull $ANTREA_DOCKER_IMAGE
+        if [[ "$?" -ne 0 ]]; then
+            echoerr "Error, Downloading the Docker image $ANTREA_DOCKER_IMAGE"
+            exit 2
+        fi
+    else
+        echo "$ANTREA_DOCKER_IMAGE image is already loaded to Docker"
+    fi
+    # Start Antrea/OVS containers as part of system startup.
+    setup_antrea_agent_service
+
+    set -eo pipefail
+}
+
+# To validate if any errors in OVS bridge config and delete the bridge.
 check_ovs_config_and_cleanup() {
     bridges=$(ovs-vsctl list-br)
     for br in $bridges; do
-        if [ "$br" != "$OVS_BRIDGE" ] ; then
+        if [ "$br" != "$OVS_BRIDGE" ]; then
             continue
         fi
         # Check if any of the interface is in error state.
         ports=$(ovs-vsctl list-ports $OVS_BRIDGE)
         for port in $ports; do
             output=$(ovs-vsctl --no-headings --columns=error list interface "$port")
-            if [ "$output" != '[]' ] ; then
+            if [ "$output" != '[]' ]; then
                 echoerr "Error while listing interface $port, deleting bridge $OVS_BRIDGE"
                 ovs-vsctl del-br "$OVS_BRIDGE"
                 break
@@ -161,6 +393,7 @@ check_ovs_config_and_cleanup() {
     done
 }
 
+
 validate_argument() {
     if [[ $2 == --* || -z $2 ]]; then
         echoerr "Error invalid argument for $1: <$2>"
@@ -173,6 +406,10 @@ while [[ $# -gt 0 ]]
 do
 key="$1"
 case $key in
+    --containerize)
+    CONTAINERIZE=true
+    shift 1
+    ;;
     --ns)
     AGENT_NAMESPACE="$2"
     validate_argument $1 $2
@@ -211,6 +448,12 @@ case $key in
     VALIDATE_OVS_CONFIG=true
     shift 1
     ;;
+    --antrea-version)
+    ANTREA_VERSION="$2"
+    ANTREA_DOCKER_IMAGE="${ANTREA_DOCKER_REGISTRY}:${ANTREA_VERSION}"
+    validate_argument $1 $2
+    shift 2
+    ;;
     -h|--help)
     print_usage
     exit 0
@@ -223,19 +466,27 @@ esac
 done
 
 # Check whether OVS configuration needs to be cleaned up.
-if [ "$VALIDATE_OVS_CONFIG" = true ] ; then
+if [ "$VALIDATE_OVS_CONFIG" = true ] && [ -z "$CONTAINERIZE" ]; then
     check_ovs_config_and_cleanup
     exit 0
 fi
 
-# Check for mandatory arguments.
-if [ -z "$AGENT_BIN_PATH" ] || [ -z "$CONFIG_PATH" ] || [ -z "$KUBECONFIG" ] || [ -z "$ANTREA_KUBECONFIG" ] ; then
-    echoerr "Missing argument(s)"
-    print_usage
-    exit 1
-fi
+validate_mandatory_arguments() {
+    if [ -z "$CONFIG_PATH" ] || [ -z "$KUBECONFIG" ] || [ -z "$ANTREA_KUBECONFIG" ] || ( [ -z "$AGENT_BIN_PATH" ] && [ -z "$CONTAINERIZE" ] ); then
+        echoerr "Missing argument(s)"
+        print_usage
+        exit 1
+    fi
+}
 
+validate_mandatory_arguments
 check_supported_platform
 copy_antrea_agent_files
 update_antrea_agent_conf
-start_antrea_agent_service
+if [ -z "$CONTAINERIZE" ]; then
+    start_antrea_agent_service
+else
+    install_prerequisite_packages
+    create_docker_compose_file
+    deploy_antrea_containers
+fi
diff --git a/multicluster/controllers/multicluster/label_identity_controller.go b/multicluster/controllers/multicluster/label_identity_controller.go
index f370dda08..fca54aabc 100644
--- a/multicluster/controllers/multicluster/label_identity_controller.go
+++ b/multicluster/controllers/multicluster/label_identity_controller.go
@@ -105,7 +105,7 @@ func (r *LabelIdentityReconciler) Reconcile(ctx context.Context, req ctrl.Reques
 		klog.ErrorS(err, "Cannot get corresponding Namespace of the Pod", "pod", req.NamespacedName)
 		return ctrl.Result{}, err
 	}
-	normalizedLabel := getNormalizedLabel(ns.Labels, pod.Labels, ns.Name)
+	normalizedLabel := GetNormalizedLabel(ns.Labels, pod.Labels, ns.Name)
 	r.onPodCreateOrUpdate(req.NamespacedName.String(), normalizedLabel)
 	return ctrl.Result{}, nil
 }
@@ -317,7 +317,7 @@ func (r *LabelIdentityReconciler) getLabelIdentityResourceExport(name, normalize
 	}
 }
 
-func getNormalizedLabel(nsLabels, podLabels map[string]string, ns string) string {
+func GetNormalizedLabel(nsLabels, podLabels map[string]string, ns string) string {
 	if _, ok := nsLabels[v1.LabelMetadataName]; !ok {
 		// NamespaceDefaultLabelName is supported from K8s v1.21. For K8s versions before v1.21,
 		// we append the Namespace name label to the Namespace label set, so that the exported
diff --git a/pkg/agent/cniserver/pod_configuration.go b/pkg/agent/cniserver/pod_configuration.go
index 89b023ab4..c50b24860 100644
--- a/pkg/agent/cniserver/pod_configuration.go
+++ b/pkg/agent/cniserver/pod_configuration.go
@@ -451,6 +451,7 @@ func (pc *podConfigurator) reconcile(pods []corev1.Pod, containerAccess *contain
 				containerConfig.MAC,
 				uint32(containerConfig.OFPort),
 				containerConfig.VLANID,
+				nil,
 			); err != nil {
 				klog.Errorf("Error when re-installing flows for Pod %s", namespacedName)
 			}
@@ -491,7 +492,7 @@ func (pc *podConfigurator) connectInterfaceToOVSCommon(ovsPortName string, conta
 		return fmt.Errorf("failed to get of_port of OVS port %s: %v", ovsPortName, err)
 	}
 	klog.V(2).Infof("Setting up Openflow entries for container %s", containerID)
-	if err := pc.ofClient.InstallPodFlows(ovsPortName, containerConfig.IPs, containerConfig.MAC, uint32(ofPort), containerConfig.VLANID); err != nil {
+	if err := pc.ofClient.InstallPodFlows(ovsPortName, containerConfig.IPs, containerConfig.MAC, uint32(ofPort), containerConfig.VLANID, nil); err != nil {
 		return fmt.Errorf("failed to add Openflow entries for container %s: %v", containerID, err)
 	}
 	containerConfig.OVSPortConfig = &interfacestore.OVSPortConfig{PortUUID: portUUID, OFPort: ofPort}
diff --git a/pkg/agent/cniserver/pod_configuration_windows.go b/pkg/agent/cniserver/pod_configuration_windows.go
index 16f824f84..230a27d71 100644
--- a/pkg/agent/cniserver/pod_configuration_windows.go
+++ b/pkg/agent/cniserver/pod_configuration_windows.go
@@ -44,7 +44,7 @@ func (pc *podConfigurator) connectInterfaceToOVSAsync(ifConfig *interfacestore.I
 		}
 		containerID := ifConfig.ContainerID
 		klog.V(2).Infof("Setting up Openflow entries for container %s", containerID)
-		if err := pc.ofClient.InstallPodFlows(ovsPortName, ifConfig.IPs, ifConfig.MAC, uint32(ofPort), ifConfig.VLANID); err != nil {
+		if err := pc.ofClient.InstallPodFlows(ovsPortName, ifConfig.IPs, ifConfig.MAC, uint32(ofPort), ifConfig.VLANID, nil); err != nil {
 			return fmt.Errorf("failed to add Openflow entries for container %s: %v", containerID, err)
 		}
 		// Update interface config with the ofPort.
diff --git a/pkg/agent/controller/egress/egress_controller.go b/pkg/agent/controller/egress/egress_controller.go
index 2f62bc6b0..6bf53a3e3 100644
--- a/pkg/agent/controller/egress/egress_controller.go
+++ b/pkg/agent/controller/egress/egress_controller.go
@@ -905,3 +905,27 @@ func (c *EgressController) deleteEgressGroup(group *cpv1b2.EgressGroup) {
 	delete(c.egressGroups, group.Name)
 	c.queue.Add(group.Name)
 }
+
+// GetEgressIPByMark returns the Egress IP associated with the snatMark.
+func (c *EgressController) GetEgressIPByMark(mark uint32) (string, error) {
+	c.egressIPStatesMutex.Lock()
+	defer c.egressIPStatesMutex.Unlock()
+	for _, e := range c.egressIPStates {
+		if e.mark == mark {
+			return e.egressIP.String(), nil
+		}
+	}
+	return "", fmt.Errorf("no EgressIP associated with mark %v", mark)
+}
+
+// GetEgress returns effective Egress applied on a Pod.
+func (c *EgressController) GetEgress(ns, podName string) (string, error) {
+	c.egressBindingsMutex.RLock()
+	defer c.egressBindingsMutex.RUnlock()
+	pod := k8s.NamespacedName(ns, podName)
+	binding, exists := c.egressBindings[pod]
+	if !exists {
+		return "", fmt.Errorf("no Egress applied to Pod %v", pod)
+	}
+	return binding.effectiveEgress, nil
+}
diff --git a/pkg/agent/controller/networkpolicy/cache.go b/pkg/agent/controller/networkpolicy/cache.go
index 3b55abc66..b76464256 100644
--- a/pkg/agent/controller/networkpolicy/cache.go
+++ b/pkg/agent/controller/networkpolicy/cache.go
@@ -30,6 +30,7 @@ import (
 
 	"antrea.io/antrea/pkg/agent/config"
 	"antrea.io/antrea/pkg/agent/metrics"
+	"antrea.io/antrea/pkg/agent/openflow"
 	agenttypes "antrea.io/antrea/pkg/agent/types"
 	v1beta "antrea.io/antrea/pkg/apis/controlplane/v1beta2"
 	crdv1alpha1 "antrea.io/antrea/pkg/apis/crd/v1alpha1"
@@ -691,6 +692,38 @@ func toRule(r *v1beta.NetworkPolicyRule, policy *v1beta.NetworkPolicy, maxPriori
 	return rule
 }
 
+// toStretchedNetworkPolicySecurityRule converts v1beta.NetworkPolicyRule to *rule
+// which is used to drop all traffic initiated from Pods with UnknownLabelIdentity.
+// Pods may have UnknownLabelIdentity when their Labels are completely new Labels
+// among the whole ClusterSet. Before the new allocated LabelIdentity is imported
+// in the Cluster, those Pods will have UnknownLabelIdentity.
+func toStretchedNetworkPolicySecurityRule(r *v1beta.NetworkPolicyRule, policy *v1beta.NetworkPolicy, maxPriority int32) *rule {
+	appliedToGroups := policy.AppliedToGroups
+	if len(r.AppliedToGroups) != 0 {
+		appliedToGroups = r.AppliedToGroups
+	}
+	snpSecDropAction := crdv1alpha1.RuleActionDrop
+	rule := &rule{
+		Direction:       r.Direction,
+		From:            v1beta.NetworkPolicyPeer{LabelIdentities: []uint32{openflow.UnknownLabelIdentity}},
+		To:              r.To,
+		Services:        r.Services,
+		Action:          &snpSecDropAction,
+		Priority:        r.Priority,
+		PolicyPriority:  policy.Priority,
+		TierPriority:    policy.TierPriority,
+		AppliedToGroups: appliedToGroups,
+		Name:            r.Name + "-security",
+		PolicyUID:       policy.UID,
+		SourceRef:       policy.SourceRef,
+		EnableLogging:   r.EnableLogging,
+	}
+	rule.ID = hashRule(rule)
+	rule.PolicyName = policy.Name
+	rule.MaxPriority = maxPriority
+	return rule
+}
+
 // getMaxPriority returns the highest rule priority for v1beta.NetworkPolicy that is created
 // by Antrea-native policies. For K8s NetworkPolicies, it always returns -1.
 func getMaxPriority(policy *v1beta.NetworkPolicy) int32 {
@@ -772,22 +805,31 @@ func (c *ruleCache) updateNetworkPolicyLocked(policy *v1beta.NetworkPolicy) bool
 	anyRuleUpdate := false
 	maxPriority := getMaxPriority(policy)
 	for i := range policy.Rules {
-		r := toRule(&policy.Rules[i], policy, maxPriority)
-		if _, exists := ruleByID[r.ID]; exists {
-			// If rule already exists, remove it from the map so the ones left finally are orphaned.
-			klog.V(2).Infof("Rule %v was not changed", r.ID)
-			delete(ruleByID, r.ID)
-		} else {
-			// If rule doesn't exist, add it to cache, mark it as dirty.
-			c.rules.Add(r)
-			// Count up antrea_agent_ingress_networkpolicy_rule_count or antrea_agent_egress_networkpolicy_rule_count
-			if r.Direction == v1beta.DirectionIn {
-				metrics.IngressNetworkPolicyRuleCount.Inc()
+		rules := []*rule{toRule(&policy.Rules[i], policy, maxPriority)}
+		if len(policy.Rules[i].From.LabelIdentities) > 0 {
+			// If the rule is a StretchedNetworkPolicy rule, we also need to add a security rule.
+			// The security rule is used to drop all traffic initiated from Pods with
+			// UnknownLabelIdentity to make sure those traffic won't sneak around the Policy.
+			rules = append(rules, toStretchedNetworkPolicySecurityRule(&policy.Rules[i], policy, maxPriority))
+		}
+		for _, r := range rules {
+			if _, exists := ruleByID[r.ID]; exists {
+				// If rule already exists, remove it from the map so the ones left are orphaned,
+				// which means those rules need to be handled by dirtyRuleHandler.
+				klog.V(2).InfoS("Rule was not changed", "id", r.ID)
+				delete(ruleByID, r.ID)
 			} else {
-				metrics.EgressNetworkPolicyRuleCount.Inc()
+				// If rule doesn't exist, add it to cache and mark it as dirty.
+				c.rules.Add(r)
+				// Count up antrea_agent_ingress_networkpolicy_rule_count or antrea_agent_egress_networkpolicy_rule_count
+				if r.Direction == v1beta.DirectionIn {
+					metrics.IngressNetworkPolicyRuleCount.Inc()
+				} else {
+					metrics.EgressNetworkPolicyRuleCount.Inc()
+				}
+				c.dirtyRuleHandler(r.ID)
+				anyRuleUpdate = true
 			}
-			c.dirtyRuleHandler(r.ID)
-			anyRuleUpdate = true
 		}
 	}
 
diff --git a/pkg/agent/controller/networkpolicy/reconciler.go b/pkg/agent/controller/networkpolicy/reconciler.go
index 7dc160f32..87151f7c6 100644
--- a/pkg/agent/controller/networkpolicy/reconciler.go
+++ b/pkg/agent/controller/networkpolicy/reconciler.go
@@ -550,6 +550,8 @@ func (r *reconciler) computeOFRulesForAdd(rule *CompletedRule, ofPriority *uint1
 		from1 := groupMembersToOFAddresses(rule.FromAddresses)
 		// Get addresses that in From IPBlock but not in Except IPBlocks.
 		from2 := ipBlocksToOFAddresses(rule.From.IPBlocks, r.ipv4Enabled, r.ipv6Enabled, isRuleAppliedToService)
+		from3 := labelIDToOFAddresses(rule.From.LabelIdentities)
+		from := append(from1, append(from2, from3...)...)
 		membersByServicesMap, servicesMap := groupMembersByServices(rule.Services, rule.TargetMembers)
 		for svcKey, members := range membersByServicesMap {
 			var toAddresses []types.Address
@@ -567,7 +569,7 @@ func (r *reconciler) computeOFRulesForAdd(rule *CompletedRule, ofPriority *uint1
 			}
 			ofRuleByServicesMap[svcKey] = &types.PolicyRule{
 				Direction:     v1beta2.DirectionIn,
-				From:          append(from1, from2...),
+				From:          from,
 				To:            toAddresses,
 				Service:       filterUnresolvablePort(servicesMap[svcKey]),
 				Action:        rule.Action,
@@ -1208,6 +1210,15 @@ func ipBlocksToOFAddresses(ipBlocks []v1beta2.IPBlock, ipv4Enabled, ipv6Enabled,
 	return addresses
 }
 
+func labelIDToOFAddresses(labelIDs []uint32) []types.Address {
+	// Must not return nil as it means not restricted by addresses in Openflow implementation.
+	addresses := make([]types.Address, 0, len(labelIDs))
+	for _, labelID := range labelIDs {
+		addresses = append(addresses, openflow.NewLabelIDAddress(labelID))
+	}
+	return addresses
+}
+
 func isIPNetSupportedByAF(ipnet *net.IPNet, ipv4Enabled, ipv6Enabled bool) bool {
 	if (ipnet.IP.To4() != nil && ipv4Enabled) || (ipnet.IP.To4() == nil && ipv6Enabled) {
 		return true
diff --git a/pkg/agent/controller/networkpolicy/reject.go b/pkg/agent/controller/networkpolicy/reject.go
index 0544c9ec0..c468f11a6 100644
--- a/pkg/agent/controller/networkpolicy/reject.go
+++ b/pkg/agent/controller/networkpolicy/reject.go
@@ -299,14 +299,18 @@ func getRejectOFPorts(rejectType RejectType, sIface, dIface *interfacestore.Inte
 	return inPort, outPort
 }
 
-// getRejectPacketOutMutateFunc returns the mutate func of a packetOut based on the RejectType.
+// getRejectPacketOutMutateFunc returns the mutate-func of a packetOut based on the RejectType.
 func getRejectPacketOutMutateFunc(rejectType RejectType, nodeType config.NodeType) func(binding.PacketOutBuilder) binding.PacketOutBuilder {
 	var mutatePacketOut func(binding.PacketOutBuilder) binding.PacketOutBuilder
+	mutatePacketOut = func(packetOutBuilder binding.PacketOutBuilder) binding.PacketOutBuilder {
+		return packetOutBuilder.AddLoadRegMark(openflow.CustomReasonRejectRegMark)
+	}
 	switch rejectType {
 	case RejectServiceLocal:
 		tableID := openflow.ConntrackTable.GetID()
 		mutatePacketOut = func(packetOutBuilder binding.PacketOutBuilder) binding.PacketOutBuilder {
-			return packetOutBuilder.AddResubmitAction(nil, &tableID)
+			return packetOutBuilder.AddLoadRegMark(openflow.CustomReasonRejectRegMark).
+				AddResubmitAction(nil, &tableID)
 		}
 	case RejectLocalToRemote:
 		tableID := openflow.L3ForwardingTable.GetID()
@@ -315,7 +319,8 @@ func getRejectPacketOutMutateFunc(rejectType RejectType, nodeType config.NodeTyp
 			tableID = openflow.L2ForwardingCalcTable.GetID()
 		}
 		mutatePacketOut = func(packetOutBuilder binding.PacketOutBuilder) binding.PacketOutBuilder {
-			return packetOutBuilder.AddResubmitAction(nil, &tableID)
+			return packetOutBuilder.AddLoadRegMark(openflow.CustomReasonRejectRegMark).
+				AddResubmitAction(nil, &tableID)
 		}
 	}
 	return mutatePacketOut
diff --git a/pkg/agent/controller/traceflow/packetin.go b/pkg/agent/controller/traceflow/packetin.go
index f60bca25d..ea82b8415 100644
--- a/pkg/agent/controller/traceflow/packetin.go
+++ b/pkg/agent/controller/traceflow/packetin.go
@@ -76,7 +76,7 @@ func (c *Controller) parsePacketIn(pktIn *ofctrl.PacketIn) (*crdv1alpha1.Tracefl
 	// Directly read data plane tag from packet.
 	var err error
 	var tag uint8
-	var ctNwDst, ctNwSrc, ipDst, ipSrc string
+	var ctNwDst, ctNwSrc, ipDst, ipSrc, ns, srcPod string
 	etherData := new(protocol.Ethernet)
 	if err := etherData.UnmarshalBinary(pktIn.Data.(*util.Buffer).Bytes()); err != nil {
 		return nil, nil, nil, fmt.Errorf("failed to parse Ethernet packet from packet-in message: %v", err)
@@ -148,6 +148,8 @@ func (c *Controller) parsePacketIn(pktIn *ofctrl.PacketIn) (*crdv1alpha1.Tracefl
 	if err != nil {
 		return nil, nil, nil, fmt.Errorf("failed to get Traceflow %s CRD: %v", tfState.name, err)
 	}
+	ns = tf.Spec.Source.Namespace
+	srcPod = tf.Spec.Source.Pod
 
 	obs := []crdv1alpha1.Observation{}
 	tableID := pktIn.TableId
@@ -256,11 +258,48 @@ func (c *Controller) parsePacketIn(pktIn *ofctrl.PacketIn) (*crdv1alpha1.Tracefl
 		gwPort := c.nodeConfig.GatewayConfig.OFPort
 		tunPort := c.nodeConfig.TunnelOFPort
 		if c.networkConfig.TrafficEncapMode.SupportsEncap() && outputPort == tunPort {
+			var isRemoteEgress uint32
+			if match := getMatchRegField(matchers, openflow.RemoteSNATRegMark.GetField()); match != nil {
+				isRemoteEgress, err = getRegValue(match, openflow.RemoteSNATRegMark.GetField().GetRange().ToNXRange())
+				if err != nil {
+					return nil, nil, nil, err
+				}
+			}
+			if isRemoteEgress == 1 { // an Egress packet, currently on source Node and forwarded to Egress Node.
+				egress, err := c.egressQuerier.GetEgress(ns, srcPod)
+				if err != nil {
+					return nil, nil, nil, err
+				}
+				obEgress := getEgressObservation(false, tunnelDstIP, egress)
+				obs = append(obs, *obEgress)
+			}
 			ob.TunnelDstIP = tunnelDstIP
 			ob.Action = crdv1alpha1.ActionForwarded
 		} else if ipDst == gatewayIP.String() && outputPort == gwPort {
 			ob.Action = crdv1alpha1.ActionDelivered
 		} else if c.networkConfig.TrafficEncapMode.SupportsEncap() && outputPort == gwPort {
+			var pktMark uint32
+			if match := getMatchPktMarkField(matchers); match != nil {
+				pktMark, err = getMarkValue(match)
+				if err != nil {
+					return nil, nil, nil, err
+				}
+			}
+			if pktMark != 0 { // Egress packet on Egress Node
+				egressIP, err := c.egressQuerier.GetEgressIPByMark(pktMark)
+				if err != nil {
+					return nil, nil, nil, err
+				}
+				egress := ""
+				if tunnelDstIP == "" { // Egress Node is Source Node of this Egress packet
+					egress, err = c.egressQuerier.GetEgress(ns, srcPod)
+					if err != nil {
+						return nil, nil, nil, err
+					}
+				}
+				obEgress := getEgressObservation(true, egressIP, egress)
+				obs = append(obs, *obEgress)
+			}
 			ob.Action = crdv1alpha1.ActionForwardedOutOfOverlay
 		} else if outputPort == gwPort { // noEncap
 			ob.Action = crdv1alpha1.ActionForwarded
@@ -277,6 +316,10 @@ func (c *Controller) parsePacketIn(pktIn *ofctrl.PacketIn) (*crdv1alpha1.Tracefl
 	return tf, &nodeResult, capturedPacket, nil
 }
 
+func getMatchPktMarkField(matchers *ofctrl.Matchers) *ofctrl.MatchField {
+	return matchers.GetMatchByName("NXM_NX_PKT_MARK")
+}
+
 func getMatchRegField(matchers *ofctrl.Matchers, field *binding.RegField) *ofctrl.MatchField {
 	return openflow.GetMatchFieldByRegID(matchers, field.GetRegID())
 }
@@ -288,6 +331,14 @@ func getMatchTunnelDstField(matchers *ofctrl.Matchers, isIPv6 bool) *ofctrl.Matc
 	return matchers.GetMatchByName("NXM_NX_TUN_IPV4_DST")
 }
 
+func getMarkValue(match *ofctrl.MatchField) (uint32, error) {
+	mark, ok := match.GetValue().(uint32)
+	if !ok {
+		return 0, errors.New("mark value cannot be got")
+	}
+	return mark, nil
+}
+
 func getRegValue(regMatch *ofctrl.MatchField, rng *openflow15.NXRange) (uint32, error) {
 	regValue, ok := regMatch.GetValue().(*ofctrl.NXRegister)
 	if !ok {
@@ -409,3 +460,16 @@ func parseCapturedPacket(pktIn *ofctrl.PacketIn) *crdv1alpha1.Packet {
 	}
 	return &capturedPacket
 }
+
+func getEgressObservation(isEgressNode bool, egressIP, egressName string) *crdv1alpha1.Observation {
+	ob := new(crdv1alpha1.Observation)
+	ob.Component = crdv1alpha1.ComponentEgress
+	ob.EgressIP = egressIP
+	ob.Egress = egressName
+	if isEgressNode {
+		ob.Action = crdv1alpha1.ActionMarkedForSNAT
+	} else {
+		ob.Action = crdv1alpha1.ActionForwardedToEgressNode
+	}
+	return ob
+}
diff --git a/pkg/agent/controller/traceflow/traceflow_controller.go b/pkg/agent/controller/traceflow/traceflow_controller.go
index fd129281b..7f99d9f99 100644
--- a/pkg/agent/controller/traceflow/traceflow_controller.go
+++ b/pkg/agent/controller/traceflow/traceflow_controller.go
@@ -96,6 +96,7 @@ type Controller struct {
 	ovsBridgeClient        ovsconfig.OVSBridgeClient
 	ofClient               openflow.Client
 	networkPolicyQuerier   querier.AgentNetworkPolicyInfoQuerier
+	egressQuerier          querier.EgressQuerier
 	interfaceStore         interfacestore.InterfaceStore
 	networkConfig          *config.NetworkConfig
 	nodeConfig             *config.NodeConfig
@@ -116,6 +117,7 @@ func NewTraceflowController(
 	traceflowInformer crdinformers.TraceflowInformer,
 	client openflow.Client,
 	npQuerier querier.AgentNetworkPolicyInfoQuerier,
+	egressQuerier querier.EgressQuerier,
 	ovsBridgeClient ovsconfig.OVSBridgeClient,
 	interfaceStore interfacestore.InterfaceStore,
 	networkConfig *config.NetworkConfig,
@@ -130,6 +132,7 @@ func NewTraceflowController(
 		ovsBridgeClient:       ovsBridgeClient,
 		ofClient:              client,
 		networkPolicyQuerier:  npQuerier,
+		egressQuerier:         egressQuerier,
 		interfaceStore:        interfaceStore,
 		networkConfig:         networkConfig,
 		nodeConfig:            nodeConfig,
diff --git a/pkg/agent/multicluster/mc_route_controller.go b/pkg/agent/multicluster/mc_route_controller.go
index 6c5966e9b..852f3d4a2 100644
--- a/pkg/agent/multicluster/mc_route_controller.go
+++ b/pkg/agent/multicluster/mc_route_controller.go
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package noderoute
+package multicluster
 
 import (
 	"fmt"
@@ -75,7 +75,8 @@ type MCRouteController struct {
 	// events.
 	installedActiveGW *mcv1alpha1.Gateway
 	// The Namespace where Antrea Multi-cluster Controller is running.
-	namespace string
+	namespace                    string
+	enableStretchedNetworkPolicy bool
 }
 
 func NewMCRouteController(
@@ -87,22 +88,24 @@ func NewMCRouteController(
 	interfaceStore interfacestore.InterfaceStore,
 	nodeConfig *config.NodeConfig,
 	namespace string,
+	enableStretchedNetworkPolicy bool,
 ) *MCRouteController {
 	controller := &MCRouteController{
-		mcClient:             mcClient,
-		ovsBridgeClient:      ovsBridgeClient,
-		ofClient:             client,
-		interfaceStore:       interfaceStore,
-		nodeConfig:           nodeConfig,
-		gwInformer:           gwInformer,
-		gwLister:             gwInformer.Lister(),
-		gwListerSynced:       gwInformer.Informer().HasSynced,
-		ciImportInformer:     ciImportInformer,
-		ciImportLister:       ciImportInformer.Lister(),
-		ciImportListerSynced: ciImportInformer.Informer().HasSynced,
-		queue:                workqueue.NewNamedRateLimitingQueue(workqueue.NewItemExponentialFailureRateLimiter(minRetryDelay, maxRetryDelay), "gatewayroute"),
-		installedCIImports:   make(map[string]*mcv1alpha1.ClusterInfoImport),
-		namespace:            namespace,
+		mcClient:                     mcClient,
+		ovsBridgeClient:              ovsBridgeClient,
+		ofClient:                     client,
+		interfaceStore:               interfaceStore,
+		nodeConfig:                   nodeConfig,
+		gwInformer:                   gwInformer,
+		gwLister:                     gwInformer.Lister(),
+		gwListerSynced:               gwInformer.Informer().HasSynced,
+		ciImportInformer:             ciImportInformer,
+		ciImportLister:               ciImportInformer.Lister(),
+		ciImportListerSynced:         ciImportInformer.Informer().HasSynced,
+		queue:                        workqueue.NewNamedRateLimitingQueue(workqueue.NewItemExponentialFailureRateLimiter(minRetryDelay, maxRetryDelay), "gatewayroute"),
+		installedCIImports:           make(map[string]*mcv1alpha1.ClusterInfoImport),
+		namespace:                    namespace,
+		enableStretchedNetworkPolicy: enableStretchedNetworkPolicy,
 	}
 	controller.gwInformer.Informer().AddEventHandlerWithResyncPeriod(
 		cache.ResourceEventHandlerFuncs{
@@ -372,7 +375,8 @@ func (c *MCRouteController) addMCFlowsForSingleCIImp(activeGW *mcv1alpha1.Gatewa
 			ciImport.Name,
 			peerConfigs,
 			tunnelPeerIPToRemoteGW,
-			localGatewayIP); err != nil {
+			localGatewayIP,
+			c.enableStretchedNetworkPolicy); err != nil {
 			return fmt.Errorf("failed to install flows to remote Gateway in ClusterInfoImport %s: %v", ciImport.Name, err)
 		}
 	} else {
@@ -381,7 +385,8 @@ func (c *MCRouteController) addMCFlowsForSingleCIImp(activeGW *mcv1alpha1.Gatewa
 		if err := c.ofClient.InstallMulticlusterNodeFlows(
 			ciImport.Name,
 			peerConfigs,
-			tunnelPeerIPToLocalGW); err != nil {
+			tunnelPeerIPToLocalGW,
+			c.enableStretchedNetworkPolicy); err != nil {
 			return fmt.Errorf("failed to install flows to Gateway %s: %v", activeGW.Name, err)
 		}
 	}
diff --git a/pkg/agent/multicluster/stretched_networkpolicy_controller.go b/pkg/agent/multicluster/stretched_networkpolicy_controller.go
new file mode 100644
index 000000000..e9a67f1d5
--- /dev/null
+++ b/pkg/agent/multicluster/stretched_networkpolicy_controller.go
@@ -0,0 +1,345 @@
+// 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 multicluster
+
+import (
+	"fmt"
+	"reflect"
+	"sync"
+	"time"
+
+	v1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/wait"
+	coreinformers "k8s.io/client-go/informers/core/v1"
+	corelisters "k8s.io/client-go/listers/core/v1"
+	"k8s.io/client-go/tools/cache"
+	"k8s.io/client-go/util/workqueue"
+	"k8s.io/klog/v2"
+
+	"antrea.io/antrea/multicluster/apis/multicluster/v1alpha1"
+	"antrea.io/antrea/multicluster/controllers/multicluster"
+	mcinformers "antrea.io/antrea/multicluster/pkg/client/informers/externalversions/multicluster/v1alpha1"
+	mclisters "antrea.io/antrea/multicluster/pkg/client/listers/multicluster/v1alpha1"
+	"antrea.io/antrea/pkg/agent/interfacestore"
+	"antrea.io/antrea/pkg/agent/openflow"
+	antreatypes "antrea.io/antrea/pkg/agent/types"
+	"antrea.io/antrea/pkg/util/channel"
+)
+
+const (
+	stretchedNetworkPolicyWorker         = 4
+	stretchedNetworkPolicyControllerName = "AntreaAgentStretchedNetworkPolicyController"
+
+	labelIndex = "Label"
+)
+
+type podSet map[types.NamespacedName]struct{}
+
+// StretchedNetworkPolicyController is used to update classifier flows of Pods.
+// It will make sure the latest LabelIdentity of the Pod, if available, will be
+// loaded into tun_id in the classifier flow of the Pod.
+// If the LabelIdentity of the Pod is not available when updating, the
+// UnknownLabelIdentity will be loaded. When the actual LabelIdentity is created,
+// the classifier flow will be updated accordingly.
+type StretchedNetworkPolicyController struct {
+	ofClient                  openflow.Client
+	interfaceStore            interfacestore.InterfaceStore
+	podInformer               cache.SharedIndexInformer
+	podLister                 corelisters.PodLister
+	podListerSynced           cache.InformerSynced
+	namespaceInformer         coreinformers.NamespaceInformer
+	namespaceLister           corelisters.NamespaceLister
+	namespaceListerSynced     cache.InformerSynced
+	labelIdentityInformer     mcinformers.LabelIdentityInformer
+	labelIdentityLister       mclisters.LabelIdentityLister
+	LabelIdentityListerSynced cache.InformerSynced
+	queue                     workqueue.RateLimitingInterface
+	lock                      sync.RWMutex
+
+	labelToPods map[string]podSet
+	podToLabel  map[types.NamespacedName]string
+}
+
+func NewMCAgentStretchedNetworkPolicyController(
+	client openflow.Client,
+	interfaceStore interfacestore.InterfaceStore,
+	podInformer cache.SharedIndexInformer,
+	namespaceInformer coreinformers.NamespaceInformer,
+	labelIdentityInformer mcinformers.LabelIdentityInformer,
+	podUpdateSubscriber channel.Subscriber,
+) *StretchedNetworkPolicyController {
+	controller := &StretchedNetworkPolicyController{
+		ofClient:                  client,
+		interfaceStore:            interfaceStore,
+		podInformer:               podInformer,
+		podLister:                 corelisters.NewPodLister(podInformer.GetIndexer()),
+		podListerSynced:           podInformer.HasSynced,
+		namespaceInformer:         namespaceInformer,
+		namespaceLister:           namespaceInformer.Lister(),
+		namespaceListerSynced:     namespaceInformer.Informer().HasSynced,
+		labelIdentityInformer:     labelIdentityInformer,
+		labelIdentityLister:       labelIdentityInformer.Lister(),
+		LabelIdentityListerSynced: labelIdentityInformer.Informer().HasSynced,
+		queue:                     workqueue.NewNamedRateLimitingQueue(workqueue.DefaultItemBasedRateLimiter(), "stretchedNetworkPolicy"),
+		labelToPods:               map[string]podSet{},
+		podToLabel:                map[types.NamespacedName]string{},
+	}
+
+	controller.podInformer.AddEventHandlerWithResyncPeriod(
+		cache.ResourceEventHandlerFuncs{
+			// Pod add event will be handled by processPodCNIAddEvent.
+			// We choose to use events from podUpdateSubscriber instead of the informer because
+			// the controller can only update the Pod classifier flow when the Pod container
+			// config is available. Events from the Informer may be received way before the Pod
+			// container config is available, which will cause the work item be continually
+			// re-queued with an exponential increased delay time. When the Pod container
+			// config is ready, the work item could wait for a long time to be processed.
+			UpdateFunc: controller.processPodUpdate,
+			DeleteFunc: controller.processPodDelete,
+		},
+		resyncPeriod,
+	)
+	controller.namespaceInformer.Informer().AddEventHandlerWithResyncPeriod(
+		cache.ResourceEventHandlerFuncs{
+			UpdateFunc: controller.processNamespaceUpdate,
+		},
+		resyncPeriod,
+	)
+	controller.labelIdentityInformer.Informer().AddEventHandlerWithResyncPeriod(
+		cache.ResourceEventHandlerFuncs{
+			AddFunc: controller.processLabelIdentityEvent,
+			UpdateFunc: func(old, cur interface{}) {
+				controller.processLabelIdentityEvent(cur)
+			},
+			DeleteFunc: controller.processLabelIdentityEvent,
+		},
+		resyncPeriod,
+	)
+	podUpdateSubscriber.Subscribe(controller.processPodCNIAddEvent)
+
+	controller.labelIdentityInformer.Informer().AddIndexers(cache.Indexers{
+		labelIndex: func(obj interface{}) ([]string, error) {
+			labelID, ok := obj.(*v1alpha1.LabelIdentity)
+			if !ok {
+				return []string{}, nil
+			}
+			return []string{labelID.Spec.Label}, nil
+		}})
+	return controller
+}
+
+func (s *StretchedNetworkPolicyController) Run(stopCh <-chan struct{}) {
+	defer s.queue.ShutDown()
+
+	klog.InfoS("Starting controller", "controller", stretchedNetworkPolicyControllerName)
+	defer klog.InfoS("Shutting down controller", "controller", stretchedNetworkPolicyControllerName)
+	cacheSyncs := []cache.InformerSynced{s.podListerSynced, s.namespaceListerSynced, s.LabelIdentityListerSynced}
+	if !cache.WaitForNamedCacheSync(stretchedNetworkPolicyControllerName, stopCh, cacheSyncs...) {
+		return
+	}
+	s.enqueueAllPods()
+	for i := 0; i < stretchedNetworkPolicyWorker; i++ {
+		go wait.Until(s.worker, time.Second, stopCh)
+	}
+	<-stopCh
+}
+
+func (s *StretchedNetworkPolicyController) enqueueAllPods() {
+	pods, _ := s.podLister.List(labels.Everything())
+	for _, pod := range pods {
+		if pod.Spec.HostNetwork {
+			continue
+		}
+		s.queue.Add(getPodReference(pod))
+	}
+}
+
+// worker is a long-running function that will continually call the processNextWorkItem
+// function in order to read and process a message on the workqueue.
+func (s *StretchedNetworkPolicyController) worker() {
+	for s.processNextWorkItem() {
+	}
+}
+
+func (s *StretchedNetworkPolicyController) processNextWorkItem() bool {
+	obj, quit := s.queue.Get()
+	if quit {
+		return false
+	}
+	defer s.queue.Done(obj)
+
+	if podRef, ok := obj.(types.NamespacedName); !ok {
+		s.queue.Forget(obj)
+		klog.Errorf("Expected type 'NamespacedName' in work queue but got object", "object", obj)
+	} else if err := s.syncPodClassifierFlow(podRef); err == nil {
+		s.queue.Forget(podRef)
+	} else {
+		// Put the item back on the workqueue to handle any transient errors.
+		s.queue.AddRateLimited(podRef)
+		klog.ErrorS(err, "Error syncing Pod classifier flow, requeuing", "name", podRef.Name, "namespace", podRef.Namespace)
+	}
+	return true
+}
+
+// syncPodClassifierFlow gets containerConfigs and labelIdentity according to a
+// Pod reference and updates this Pod's classifierFlow.
+func (s *StretchedNetworkPolicyController) syncPodClassifierFlow(podRef types.NamespacedName) error {
+	pod, err := s.podLister.Pods(podRef.Namespace).Get(podRef.Name)
+	if err != nil || pod.Spec.HostNetwork {
+		return nil
+	}
+	containerConfigs := s.interfaceStore.GetContainerInterfacesByPod(podRef.Name, podRef.Namespace)
+	if len(containerConfigs) == 0 {
+		klog.InfoS("Pod container config not found, will retry after it's ready", "name", podRef.Name, "namespace", podRef.Namespace)
+		return nil
+	}
+	podNS, err := s.namespaceLister.Get(podRef.Namespace)
+	if err != nil {
+		return fmt.Errorf("can't get Namespace %s: %v", podRef.Namespace, err)
+	}
+	normalizedLabel := multicluster.GetNormalizedLabel(podNS.Labels, pod.Labels, podNS.Name)
+	labelID := s.getLabelIdentity(podRef, normalizedLabel)
+	return s.ofClient.InstallPodFlows(
+		containerConfigs[0].InterfaceName,
+		containerConfigs[0].IPs,
+		containerConfigs[0].MAC,
+		uint32(containerConfigs[0].OFPort),
+		containerConfigs[0].VLANID,
+		&labelID,
+	)
+}
+
+// getLabelIdentity updates labelToPods and podToLabel and returns the
+// LabelIdentity based on the normalizedLabel.
+func (s *StretchedNetworkPolicyController) getLabelIdentity(podRef types.NamespacedName, normalizedLabel string) uint32 {
+	s.lock.Lock()
+	oldNormalizedLabel, ok := s.podToLabel[podRef]
+	if ok && oldNormalizedLabel != normalizedLabel {
+		s.deleteLabelToPod(oldNormalizedLabel, podRef)
+	}
+	if !ok || oldNormalizedLabel != normalizedLabel {
+		s.addLabelToPod(normalizedLabel, podRef)
+		s.podToLabel[podRef] = normalizedLabel
+	}
+	s.lock.Unlock()
+
+	labelID := openflow.UnknownLabelIdentity
+	if objs, err := s.labelIdentityInformer.Informer().GetIndexer().ByIndex(labelIndex, normalizedLabel); err == nil && len(objs) == 1 {
+		labelIdentity := objs[0].(*v1alpha1.LabelIdentity)
+		labelID = labelIdentity.Spec.ID
+	}
+	return labelID
+}
+
+func (s *StretchedNetworkPolicyController) processPodCNIAddEvent(e interface{}) {
+	podEvent := e.(antreatypes.PodUpdate)
+	if !podEvent.IsAdd {
+		return
+	}
+	podRef := types.NamespacedName{
+		Namespace: podEvent.PodNamespace,
+		Name:      podEvent.PodName,
+	}
+	s.queue.Add(podRef)
+}
+
+// processPodUpdate handles Pod update events. It only enqueues the Pod if the
+// Labels of this Pod have been updated.
+func (s *StretchedNetworkPolicyController) processPodUpdate(old, cur interface{}) {
+	oldPod, _ := old.(*v1.Pod)
+	curPod, _ := cur.(*v1.Pod)
+	if curPod.Spec.HostNetwork {
+		klog.V(5).InfoS("Skipped processing hostNetwork Pod update event", "name", curPod.Name, "namespace", curPod.Namespace)
+		return
+	}
+	if reflect.DeepEqual(oldPod.Labels, curPod.Labels) {
+		klog.V(5).InfoS("Pod UpdateFunc received a Pod update event, "+
+			"but labels are the same. Skip it", "name", curPod.Name, "namespace", curPod.Namespace)
+		return
+	}
+	s.queue.Add(getPodReference(curPod))
+}
+
+// processPodDelete handles Pod delete events. It deletes the Pod from the
+// labelToPods and podToLabel. After Pod is deleted, its classifier flow will also
+// be deleted by podConfigurator. So no need to enqueue this Pod to update its
+// classifier flow.
+func (s *StretchedNetworkPolicyController) processPodDelete(old interface{}) {
+	oldPod, _ := old.(*v1.Pod)
+	oldPodRef := getPodReference(oldPod)
+	s.lock.Lock()
+	defer s.lock.Unlock()
+	if podLabel, ok := s.podToLabel[oldPodRef]; ok {
+		s.deleteLabelToPod(podLabel, oldPodRef)
+		delete(s.podToLabel, oldPodRef)
+	}
+}
+
+// processNamespaceUpdate handles Namespace update events. It only enqueues all
+// Pods in this Namespace if the Labels of this Namespace have been updated.
+func (s *StretchedNetworkPolicyController) processNamespaceUpdate(old, cur interface{}) {
+	oldNS, _ := old.(*v1.Namespace)
+	curNS, _ := cur.(*v1.Namespace)
+	if reflect.DeepEqual(oldNS.Labels, curNS.Labels) {
+		klog.V(5).InfoS("Namespace UpdateFunc received a Namespace update event, but labels are the same. Skip it", "namespace", curNS.Name)
+		return
+	}
+	allPodsInNS, _ := s.podLister.Pods(curNS.Name).List(labels.Everything())
+	for _, pod := range allPodsInNS {
+		if pod.Spec.HostNetwork {
+			continue
+		}
+		s.queue.Add(getPodReference(pod))
+	}
+}
+
+// processLabelIdentityEvent handles labelIdentity add/update/delete event.
+// It will enqueue all Pods affected by this labelIdentity
+func (s *StretchedNetworkPolicyController) processLabelIdentityEvent(cur interface{}) {
+	labelIdentity, _ := cur.(*v1alpha1.LabelIdentity)
+	s.lock.RLock()
+	defer s.lock.RUnlock()
+	if podSet, ok := s.labelToPods[labelIdentity.Spec.Label]; ok {
+		for podRef := range podSet {
+			s.queue.Add(podRef)
+		}
+	}
+}
+
+func (s *StretchedNetworkPolicyController) addLabelToPod(normalizedLabel string, podRef types.NamespacedName) {
+	if _, ok := s.labelToPods[normalizedLabel]; ok {
+		s.labelToPods[normalizedLabel][podRef] = struct{}{}
+	} else {
+		s.labelToPods[normalizedLabel] = podSet{podRef: struct{}{}}
+	}
+}
+
+func (s *StretchedNetworkPolicyController) deleteLabelToPod(normalizedLabel string, podRef types.NamespacedName) {
+	if _, ok := s.labelToPods[normalizedLabel]; ok {
+		delete(s.labelToPods[normalizedLabel], podRef)
+		if len(s.labelToPods[normalizedLabel]) == 0 {
+			delete(s.labelToPods, normalizedLabel)
+		}
+	}
+}
+
+func getPodReference(pod *v1.Pod) types.NamespacedName {
+	return types.NamespacedName{
+		Name:      pod.Name,
+		Namespace: pod.Namespace,
+	}
+}
diff --git a/pkg/agent/openflow/client.go b/pkg/agent/openflow/client.go
index 381e84760..36c57e467 100644
--- a/pkg/agent/openflow/client.go
+++ b/pkg/agent/openflow/client.go
@@ -75,7 +75,7 @@ type Client interface {
 	// flows will be installed). Calls to InstallPodFlows are idempotent. Concurrent calls
 	// to InstallPodFlows and / or UninstallPodFlows are supported as long as they are all
 	// for different interfaceNames.
-	InstallPodFlows(interfaceName string, podInterfaceIPs []net.IP, podInterfaceMAC net.HardwareAddr, ofPort uint32, vlanID uint16) error
+	InstallPodFlows(interfaceName string, podInterfaceIPs []net.IP, podInterfaceMAC net.HardwareAddr, ofPort uint32, vlanID uint16, labelID *uint32) error
 
 	// UninstallPodFlows removes the connection to the local Pod specified with the
 	// interfaceName. UninstallPodFlows will do nothing if no connection to the Pod was established.
@@ -326,14 +326,16 @@ type Client interface {
 	InstallMulticlusterNodeFlows(
 		clusterID string,
 		peerConfigs map[*net.IPNet]net.IP,
-		tunnelPeerIP net.IP) error
+		tunnelPeerIP net.IP,
+		enableStretchedNetworkPolicy bool) error
 
 	// InstallMulticlusterGatewayFlows installs flows to handle cross-cluster packets between Gateways.
 	InstallMulticlusterGatewayFlows(
 		clusterID string,
 		peerConfigs map[*net.IPNet]net.IP,
 		tunnelPeerIP net.IP,
-		localGatewayIP net.IP) error
+		localGatewayIP net.IP,
+		enableStretchedNetworkPolicy bool) error
 
 	// InstallMulticlusterClassifierFlows installs flows to classify cross-cluster packets.
 	InstallMulticlusterClassifierFlows(tunnelOFPort uint32, isGateway bool) error
@@ -524,7 +526,7 @@ func (c *client) UninstallNodeFlows(hostname string) error {
 	return c.deleteFlows(c.featurePodConnectivity.nodeCachedFlows, hostname)
 }
 
-func (c *client) InstallPodFlows(interfaceName string, podInterfaceIPs []net.IP, podInterfaceMAC net.HardwareAddr, ofPort uint32, vlanID uint16) error {
+func (c *client) InstallPodFlows(interfaceName string, podInterfaceIPs []net.IP, podInterfaceMAC net.HardwareAddr, ofPort uint32, vlanID uint16, labelID *uint32) error {
 	c.replayMutex.RLock()
 	defer c.replayMutex.RUnlock()
 
@@ -534,7 +536,7 @@ func (c *client) InstallPodFlows(interfaceName string, podInterfaceIPs []net.IP,
 
 	localGatewayMAC := c.nodeConfig.GatewayConfig.MAC
 	flows := []binding.Flow{
-		c.featurePodConnectivity.podClassifierFlow(ofPort, isAntreaFlexibleIPAM),
+		c.featurePodConnectivity.podClassifierFlow(ofPort, isAntreaFlexibleIPAM, labelID),
 		c.featurePodConnectivity.l2ForwardCalcFlow(podInterfaceMAC, ofPort),
 	}
 
@@ -561,7 +563,7 @@ func (c *client) InstallPodFlows(interfaceName string, podInterfaceIPs []net.IP,
 			flows = append(flows, c.featurePodConnectivity.podVLANFlow(ofPort, vlanID))
 		}
 	}
-	err := c.addFlows(c.featurePodConnectivity.podCachedFlows, interfaceName, flows)
+	err := c.modifyFlows(c.featurePodConnectivity.podCachedFlows, interfaceName, flows)
 	if err != nil {
 		return err
 	}
@@ -1349,14 +1351,15 @@ func (c *client) UninstallMulticastGroup(groupID binding.GroupIDType) error {
 // Node and a local Gateway.
 func (c *client) InstallMulticlusterNodeFlows(clusterID string,
 	peerConfigs map[*net.IPNet]net.IP,
-	tunnelPeerIP net.IP) error {
+	tunnelPeerIP net.IP,
+	enableStretchedNetworkPolicy bool) error {
 	c.replayMutex.RLock()
 	defer c.replayMutex.RUnlock()
 	cacheKey := fmt.Sprintf("cluster_%s", clusterID)
 	var flows []binding.Flow
 	localGatewayMAC := c.nodeConfig.GatewayConfig.MAC
 	for peerCIDR, remoteGatewayIP := range peerConfigs {
-		flows = append(flows, c.featureMulticluster.l3FwdFlowToRemoteViaTun(localGatewayMAC, *peerCIDR, tunnelPeerIP, remoteGatewayIP)...)
+		flows = append(flows, c.featureMulticluster.l3FwdFlowToRemoteViaTun(localGatewayMAC, *peerCIDR, tunnelPeerIP, remoteGatewayIP, enableStretchedNetworkPolicy)...)
 	}
 	return c.modifyFlows(c.featureMulticluster.cachedFlows, cacheKey, flows)
 }
@@ -1366,6 +1369,7 @@ func (c *client) InstallMulticlusterGatewayFlows(clusterID string,
 	peerConfigs map[*net.IPNet]net.IP,
 	tunnelPeerIP net.IP,
 	localGatewayIP net.IP,
+	enableStretchedNetworkPolicy bool,
 ) error {
 	c.replayMutex.RLock()
 	defer c.replayMutex.RUnlock()
@@ -1373,7 +1377,7 @@ func (c *client) InstallMulticlusterGatewayFlows(clusterID string,
 	var flows []binding.Flow
 	localGatewayMAC := c.nodeConfig.GatewayConfig.MAC
 	for peerCIDR, remoteGatewayIP := range peerConfigs {
-		flows = append(flows, c.featureMulticluster.l3FwdFlowToRemoteViaTun(localGatewayMAC, *peerCIDR, tunnelPeerIP, remoteGatewayIP)...)
+		flows = append(flows, c.featureMulticluster.l3FwdFlowToRemoteViaTun(localGatewayMAC, *peerCIDR, tunnelPeerIP, remoteGatewayIP, enableStretchedNetworkPolicy)...)
 		// Add SNAT flows to change cross-cluster packets' source IP to local Gateway IP.
 		flows = append(flows, c.featureMulticluster.snatConntrackFlows(*peerCIDR, localGatewayIP)...)
 	}
diff --git a/pkg/agent/openflow/fields.go b/pkg/agent/openflow/fields.go
index 4351d9251..2f6a1e517 100644
--- a/pkg/agent/openflow/fields.go
+++ b/pkg/agent/openflow/fields.go
@@ -80,6 +80,8 @@ var (
 	CustomReasonDenyRegMark    = binding.NewRegMark(CustomReasonField, CustomReasonDeny)
 	CustomReasonDNSRegMark     = binding.NewRegMark(CustomReasonField, CustomReasonDNS)
 	CustomReasonIGMPRegMark    = binding.NewRegMark(CustomReasonField, CustomReasonIGMP)
+	// reg0[18]: Mark to indicate remote SNAT for Egress.
+	RemoteSNATRegMark = binding.NewOneBitRegMark(0, 18)
 
 	// reg1(NXM_NX_REG1)
 	// Field to cache the ofPort of the OVS interface where to output packet.
@@ -165,7 +167,7 @@ var (
 
 // Marks using CT.
 var (
-	//TODO: There is a bug in libOpenflow when CT_MARK range is from 0 to 0, and a wrong mask will be got. As a result,
+	// TODO: There is a bug in libOpenflow when CT_MARK range is from 0 to 0, and a wrong mask will be got. As a result,
 	// don't just use bit 0 of CT_MARK.
 
 	// CTMark (NXM_NX_CT_MARK)
diff --git a/pkg/agent/openflow/multicluster.go b/pkg/agent/openflow/multicluster.go
index 71a50ee56..7761836ef 100644
--- a/pkg/agent/openflow/multicluster.go
+++ b/pkg/agent/openflow/multicluster.go
@@ -25,6 +25,11 @@ import (
 // for cross-cluster traffic to distinguish from in-cluster traffic.
 var GlobalVirtualMACForMulticluster, _ = net.ParseMAC("aa:bb:cc:dd:ee:f0")
 
+// UnknownLabelIdentity represents an unknown label identity.
+// 24 bits in VNI are used for label identity. The max value is reserved for
+// UnknownLabelIdentity.
+const UnknownLabelIdentity = uint32(0xffffff)
+
 type featureMulticluster struct {
 	cookieAllocator cookie.Allocator
 	cachedFlows     *flowCategoryCache
@@ -65,7 +70,8 @@ func (f *featureMulticluster) l3FwdFlowToRemoteViaTun(
 	localGatewayMAC net.HardwareAddr,
 	peerServiceCIDR net.IPNet,
 	tunnelPeer net.IP,
-	remoteGatewayIP net.IP) []binding.Flow {
+	remoteGatewayIP net.IP,
+	enableStretchedNetworkPolicy bool) []binding.Flow {
 	ipProtocol := getIPProtocol(peerServiceCIDR.IP)
 	cookieID := f.cookieAllocator.Request(f.category).Raw()
 	var flows []binding.Flow
@@ -97,6 +103,23 @@ func (f *featureMulticluster) l3FwdFlowToRemoteViaTun(
 			Action().GotoTable(L3DecTTLTable.GetID()).
 			Done(),
 	)
+	if enableStretchedNetworkPolicy {
+		flows = append(flows,
+			// This generates the flow to forward cross-cluster reject traffic based
+			// on Gateway IP and reg.
+			L3ForwardingTable.ofTable.BuildFlow(priorityNormal-1).
+				Cookie(cookieID).
+				MatchProtocol(ipProtocol).
+				MatchRegMark(CustomReasonRejectRegMark).
+				MatchDstIP(remoteGatewayIP).
+				Action().SetSrcMAC(localGatewayMAC).
+				Action().SetDstMAC(GlobalVirtualMACForMulticluster).
+				Action().SetTunnelDst(tunnelPeer). // Flow based tunnel. Set tunnel destination.
+				Action().LoadRegMark(ToTunnelRegMark).
+				Action().GotoTable(L3DecTTLTable.GetID()).
+				Done(),
+		)
+	}
 	return flows
 }
 
diff --git a/pkg/agent/openflow/network_policy.go b/pkg/agent/openflow/network_policy.go
index 784bddbcd..c6fcc582b 100644
--- a/pkg/agent/openflow/network_policy.go
+++ b/pkg/agent/openflow/network_policy.go
@@ -69,6 +69,7 @@ var (
 	MatchICMPv6Code     = types.NewMatchKey(binding.ProtocolICMPv6, types.ICMPAddr, "icmpv6_code")
 	MatchServiceGroupID = types.NewMatchKey(binding.ProtocolIP, types.ServiceGroupIDAddr, "reg7[0..31]")
 	MatchIGMPProtocol   = types.NewMatchKey(binding.ProtocolIGMP, types.IGMPAddr, "igmp")
+	MatchLabelID        = types.NewMatchKey(binding.ProtocolIP, types.LabelIDAddr, "tun_id")
 	Unsupported         = types.NewMatchKey(binding.ProtocolIP, types.UnSupported, "unknown")
 
 	// metricFlowIdentifier is used to identify metric flows in metric table.
@@ -273,6 +274,25 @@ func NewCTIPNetAddress(addr net.IPNet) *CTIPNetAddress {
 	return &ia
 }
 
+type LabelIDAddress uint32
+
+func (a *LabelIDAddress) GetMatchKey(addrType types.AddressType) *types.MatchKey {
+	return MatchLabelID
+}
+
+func (a *LabelIDAddress) GetMatchValue() string {
+	return fmt.Sprintf("%d", uint32(*a))
+}
+
+func (a *LabelIDAddress) GetValue() interface{} {
+	return uint32(*a)
+}
+
+func NewLabelIDAddress(labelID uint32) *LabelIDAddress {
+	a := LabelIDAddress(labelID)
+	return &a
+}
+
 // ConjunctionNotFound is an error response when the specified policyRuleConjunction is not found from the local cache.
 type ConjunctionNotFound uint32
 
diff --git a/pkg/agent/openflow/pipeline.go b/pkg/agent/openflow/pipeline.go
index f5c338638..1715fbdc4 100644
--- a/pkg/agent/openflow/pipeline.go
+++ b/pkg/agent/openflow/pipeline.go
@@ -619,11 +619,21 @@ func (f *featurePodConnectivity) gatewayClassifierFlow() binding.Flow {
 }
 
 // podClassifierFlow generates the flow to mark the packets from a local Pod port.
-func (f *featurePodConnectivity) podClassifierFlow(podOFPort uint32, isAntreaFlexibleIPAM bool) binding.Flow {
+// If multi-cluster is enabled, also load podLabelID into LabelIDField.
+func (f *featurePodConnectivity) podClassifierFlow(podOFPort uint32, isAntreaFlexibleIPAM bool, podLabelID *uint32) binding.Flow {
 	regMarksToLoad := []*binding.RegMark{FromLocalRegMark}
 	if isAntreaFlexibleIPAM {
 		regMarksToLoad = append(regMarksToLoad, AntreaFlexibleIPAMRegMark, RewriteMACRegMark)
 	}
+	if podLabelID != nil {
+		return ClassifierTable.ofTable.BuildFlow(priorityLow).
+			Cookie(f.cookieAllocator.Request(f.category).Raw()).
+			MatchInPort(podOFPort).
+			Action().LoadRegMark(regMarksToLoad...).
+			Action().SetTunnelID(uint64(*podLabelID)).
+			Action().GotoStage(stageValidation).
+			Done()
+	}
 	return ClassifierTable.ofTable.BuildFlow(priorityLow).
 		Cookie(f.cookieAllocator.Request(f.category).Raw()).
 		MatchInPort(podOFPort).
@@ -1971,6 +1981,8 @@ func (f *featureNetworkPolicy) addFlowMatch(fb binding.FlowBuilder, matchKey *ty
 		fb = fb.MatchRegFieldWithValue(ServiceGroupIDField, matchValue.(uint32))
 	case MatchIGMPProtocol:
 		fb = fb.MatchProtocol(matchKey.GetOFProtocol())
+	case MatchLabelID:
+		fb = fb.MatchTunnelID(uint64(matchValue.(uint32)))
 	}
 	return fb
 }
@@ -2186,7 +2198,7 @@ func (f *featureEgress) snatRuleFlow(ofPort uint32, snatIP net.IP, snatMark uint
 		Action().SetSrcMAC(localGatewayMAC).
 		Action().SetDstMAC(GlobalVirtualMAC).
 		Action().SetTunnelDst(snatIP). // Set tunnel destination to the SNAT IP.
-		Action().LoadRegMark(ToTunnelRegMark).
+		Action().LoadRegMark(ToTunnelRegMark, RemoteSNATRegMark).
 		Action().GotoStage(stageSwitching).
 		Done()
 }
diff --git a/pkg/agent/types/networkpolicy.go b/pkg/agent/types/networkpolicy.go
index c020a73b0..da5a5bbd0 100644
--- a/pkg/agent/types/networkpolicy.go
+++ b/pkg/agent/types/networkpolicy.go
@@ -56,6 +56,7 @@ const (
 	ICMPAddr
 	ServiceGroupIDAddr
 	IGMPAddr
+	LabelIDAddr
 	UnSupported
 )
 
diff --git a/pkg/apis/crd/v1alpha1/types.go b/pkg/apis/crd/v1alpha1/types.go
index 2e22d8290..5448e2030 100644
--- a/pkg/apis/crd/v1alpha1/types.go
+++ b/pkg/apis/crd/v1alpha1/types.go
@@ -38,6 +38,7 @@ const (
 	ComponentRouting       TraceflowComponent = "Routing"
 	ComponentNetworkPolicy TraceflowComponent = "NetworkPolicy"
 	ComponentForwarding    TraceflowComponent = "Forwarding"
+	ComponentEgress        TraceflowComponent = "Egress"
 )
 
 type TraceflowAction string
@@ -51,6 +52,8 @@ const (
 	// ActionForwardedOutOfOverlay indicates that the packet has been forwarded out of the network
 	// managed by Antrea. This indicates that the Traceflow request can be considered complete.
 	ActionForwardedOutOfOverlay TraceflowAction = "ForwardedOutOfOverlay"
+	ActionMarkedForSNAT         TraceflowAction = "MarkedForSNAT"
+	ActionForwardedToEgressNode TraceflowAction = "ForwardedToEgressNode"
 )
 
 // List the supported protocols and their codes in traceflow.
@@ -264,6 +267,8 @@ type Observation struct {
 	DstMAC string `json:"dstMAC,omitempty" yaml:"dstMAC,omitempty"`
 	// NetworkPolicy is the combination of Namespace and NetworkPolicyName.
 	NetworkPolicy string `json:"networkPolicy,omitempty" yaml:"networkPolicy,omitempty"`
+	// Egress is the name of the Egress.
+	Egress string `json:"egress,omitempty" yaml:"egress,omitempty"`
 	// TTL is the observation TTL.
 	TTL int32 `json:"ttl,omitempty" yaml:"ttl,omitempty"`
 	// TranslatedSrcIP is the translated source IP.
@@ -272,6 +277,7 @@ type Observation struct {
 	TranslatedDstIP string `json:"translatedDstIP,omitempty" yaml:"translatedDstIP,omitempty"`
 	// TunnelDstIP is the tunnel destination IP.
 	TunnelDstIP string `json:"tunnelDstIP,omitempty" yaml:"tunnelDstIP,omitempty"`
+	EgressIP    string `json:"egressIP,omitempty" yaml:"egressIP,omitempty"`
 }
 
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
diff --git a/pkg/config/agent/config.go b/pkg/config/agent/config.go
index 5a8e47418..06c572624 100644
--- a/pkg/config/agent/config.go
+++ b/pkg/config/agent/config.go
@@ -283,6 +283,8 @@ type MulticlusterConfig struct {
 	// The Namespace where the Antrea Multi-cluster controller is running.
 	// The default is antrea-agent's Namespace.
 	Namespace string `yaml:"namespace,omitempty"`
+	// Enable StretchedNetworkPolicy which could be enforced on cross-cluster traffic.
+	EnableStretchedNetworkPolicy bool `yaml:"enableStretchedNetworkPolicy,omitempty"`
 }
 
 type ExternalNodeConfig struct {
diff --git a/pkg/graphviz/traceflow.go b/pkg/graphviz/traceflow.go
index 9136d165e..175116c35 100644
--- a/pkg/graphviz/traceflow.go
+++ b/pkg/graphviz/traceflow.go
@@ -201,6 +201,12 @@ func getTraceflowMessage(o *crdv1alpha1.Observation, spec *crdv1alpha1.Traceflow
 	if o.Action != crdv1alpha1.ActionDropped && len(o.TunnelDstIP) > 0 {
 		str += "\nTunnel Destination IP : " + o.TunnelDstIP
 	}
+	if o.Action != crdv1alpha1.ActionDropped && len(o.EgressIP) > 0 {
+		str += "\nEgress IP : " + o.EgressIP
+	}
+	if o.Action != crdv1alpha1.ActionDropped && len(o.Egress) > 0 {
+		str += "\nEgress : " + o.Egress
+	}
 	return str
 }
 
diff --git a/pkg/ovs/openflow/interfaces.go b/pkg/ovs/openflow/interfaces.go
index 30e64c3ec..9e7db1349 100644
--- a/pkg/ovs/openflow/interfaces.go
+++ b/pkg/ovs/openflow/interfaces.go
@@ -226,6 +226,7 @@ type Action interface {
 	SetSrcIP(addr net.IP) FlowBuilder
 	SetDstIP(addr net.IP) FlowBuilder
 	SetTunnelDst(addr net.IP) FlowBuilder
+	SetTunnelID(tunnelID uint64) FlowBuilder
 	PopVLAN() FlowBuilder
 	PushVLAN(etherType uint16) FlowBuilder
 	SetVLAN(vlanID uint16) FlowBuilder
@@ -281,6 +282,7 @@ type FlowBuilder interface {
 	MatchICMPv6Type(icmp6Type byte) FlowBuilder
 	MatchICMPv6Code(icmp6Code byte) FlowBuilder
 	MatchTunnelDst(dstIP net.IP) FlowBuilder
+	MatchTunnelID(tunnelID uint64) FlowBuilder
 	MatchTunMetadata(index int, data uint32) FlowBuilder
 	MatchVLAN(nonVLAN bool, vlanId uint16, vlanMask *uint16) FlowBuilder
 	// MatchCTSrcIP matches the source IPv4 address of the connection tracker original direction tuple.
diff --git a/pkg/ovs/openflow/ofctrl_action.go b/pkg/ovs/openflow/ofctrl_action.go
index 84311dbb4..7250e2687 100644
--- a/pkg/ovs/openflow/ofctrl_action.go
+++ b/pkg/ovs/openflow/ofctrl_action.go
@@ -252,6 +252,13 @@ func (a *ofFlowAction) SetTunnelDst(addr net.IP) FlowBuilder {
 	return a.builder
 }
 
+// SetTunnelID is an action to modify packet tunnel ID to the specified ID.
+func (a *ofFlowAction) SetTunnelID(tunnelID uint64) FlowBuilder {
+	setTunIDAct := &ofctrl.SetTunnelIDAction{TunnelID: tunnelID}
+	a.builder.ApplyAction(setTunIDAct)
+	return a.builder
+}
+
 // PopVLAN is an action to pop VLAN ID.
 func (a *ofFlowAction) PopVLAN() FlowBuilder {
 	popVLANAct := &ofctrl.PopVLANAction{}
diff --git a/pkg/ovs/openflow/ofctrl_builder.go b/pkg/ovs/openflow/ofctrl_builder.go
index 8d90f6476..88179adb3 100644
--- a/pkg/ovs/openflow/ofctrl_builder.go
+++ b/pkg/ovs/openflow/ofctrl_builder.go
@@ -296,6 +296,13 @@ func (b *ofFlowBuilder) MatchTunnelDst(dstIP net.IP) FlowBuilder {
 	return b
 }
 
+// MatchTunnelID adds match condition for matching tun_id.
+func (b *ofFlowBuilder) MatchTunnelID(tunnelID uint64) FlowBuilder {
+	b.matchers = append(b.matchers, fmt.Sprintf("tun_id=%d", tunnelID))
+	b.ofFlow.Match.TunnelId = tunnelID
+	return b
+}
+
 func ctLabelRange(high, low uint64, rng *Range, match *ofctrl.FlowMatch) {
 	// [127..64] [63..0]
 	//   high     low
diff --git a/pkg/querier/querier.go b/pkg/querier/querier.go
index f3722a8bc..a848c4486 100644
--- a/pkg/querier/querier.go
+++ b/pkg/querier/querier.go
@@ -60,6 +60,11 @@ type ControllerNetworkPolicyInfoQuerier interface {
 	GetConnectedAgentNum() int
 }
 
+type EgressQuerier interface {
+	GetEgressIPByMark(mark uint32) (string, error)
+	GetEgress(podNamespace, podName string) (string, error)
+}
+
 // GetSelfPod gets current pod.
 func GetSelfPod() v1.ObjectReference {
 	podName := env.GetPodName()


Test patch:
diff --git a/ci/jenkins/test-mc.sh b/ci/jenkins/test-mc.sh
index a8bd6ffd8..ae1c6d7d4 100755
--- a/ci/jenkins/test-mc.sh
+++ b/ci/jenkins/test-mc.sh
@@ -248,6 +248,24 @@ function run_codecov { (set -e
     rm -f trustedkeys.gpg codecov
 )}
 
+function modify_config {
+  if [[ ${ENABLE_MC_GATEWAY} == "true" ]]; then
+  cat > build/yamls/chart-values/antrea.yml << EOF
+multicluster:
+  enable: true
+  enableStretchedNetworkPolicy: true
+featureGates: {
+  Multicluster: true
+}
+EOF
+  make manifest
+  cd multicluster
+  sed -i 's/enableStretchedNetworkPolicy: false/enableStretchedNetworkPolicy: true/g' config/default/configmap/controller_manager_config.yaml
+  make manifests
+  cd ..
+  fi
+}
+
 function deliver_antrea_multicluster {
     echo "====== Building Antrea for the Following Commit ======"
     export GO111MODULE=on
@@ -357,16 +375,6 @@ function run_multicluster_e2e {
     export GOCACHE=${WORKDIR}/.cache/go-build
     export PATH=$GOROOT/bin:$PATH
 
-    if [[ ${ENABLE_MC_GATEWAY} == "true" ]]; then
-    cat > build/yamls/chart-values/antrea.yml <<EOF
-multicluster:
-  enable: true
-featureGates: {
-  Multicluster: true
-}
-EOF
-    make manifest
-    fi
     wait_for_antrea_multicluster_pods_ready "${LEADER_CLUSTER_CONFIG}"
     wait_for_antrea_multicluster_pods_ready "${EAST_CLUSTER_CONFIG}"
     wait_for_antrea_multicluster_pods_ready "${WEST_CLUSTER_CONFIG}"
@@ -471,6 +479,7 @@ set -e
 
 if [[ ${TESTCASE} =~ "e2e" ]]; then
     deliver_antrea_multicluster
+    modify_config
     deliver_multicluster_controller
     run_multicluster_e2e
     if $COVERAGE;then
diff --git a/multicluster/controllers/multicluster/label_identity_controller_test.go b/multicluster/controllers/multicluster/label_identity_controller_test.go
index 89eb212a0..3f6907d96 100644
--- a/multicluster/controllers/multicluster/label_identity_controller_test.go
+++ b/multicluster/controllers/multicluster/label_identity_controller_test.go
@@ -271,7 +271,7 @@ func TestGetNormalizedLabel(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			normalizedLabel := getNormalizedLabel(tt.nsLabels, tt.podLabels, tt.namespace)
+			normalizedLabel := GetNormalizedLabel(tt.nsLabels, tt.podLabels, tt.namespace)
 			assert.Equal(t, tt.expNormalizedLabel, normalizedLabel)
 		})
 	}
diff --git a/multicluster/test/e2e/framework.go b/multicluster/test/e2e/framework.go
index 07757febb..fdf10d283 100644
--- a/multicluster/test/e2e/framework.go
+++ b/multicluster/test/e2e/framework.go
@@ -39,6 +39,8 @@ const (
 	multiClusterTestNamespace string = "antrea-multicluster-test"
 	eastClusterTestService    string = "east-nginx"
 	westClusterTestService    string = "west-nginx"
+	mcEastClusterTestService  string = "antrea-mc-east-nginx"
+	mcWestClusterTestService  string = "antrea-mc-west-nginx"
 	eastCluster               string = "east-cluster"
 	westCluster               string = "west-cluster"
 	leaderCluster             string = "leader-cluster"
@@ -168,6 +170,26 @@ func (data *MCTestData) createPod(clusterName, name, nodeName, namespace, ctrNam
 	return fmt.Errorf("clusterName %s not found", clusterName)
 }
 
+func (data *MCTestData) updatePod(clusterName string, namespace, name string, mutateFunc func(*corev1.Pod)) error {
+	if d, ok := data.clusterTestDataMap[clusterName]; ok {
+		if err := d.UpdatePod(namespace, name, mutateFunc); err != nil {
+			return err
+		}
+		return nil
+	}
+	return fmt.Errorf("clusterName %s not found", clusterName)
+}
+
+func (data *MCTestData) updateNamespace(clusterName string, namespace string, mutateFunc func(*corev1.Namespace)) error {
+	if d, ok := data.clusterTestDataMap[clusterName]; ok {
+		if err := d.UpdateNamespace(namespace, mutateFunc); err != nil {
+			return err
+		}
+		return nil
+	}
+	return fmt.Errorf("clusterName %s not found", clusterName)
+}
+
 func (data *MCTestData) createService(clusterName, serviceName, namespace string, port int32, targetPort int32,
 	protocol corev1.Protocol, selector map[string]string, affinity bool, nodeLocalExternal bool, serviceType corev1.ServiceType,
 	ipFamily *corev1.IPFamily, annotation map[string]string) (*corev1.Service, error) {
@@ -196,6 +218,21 @@ func (data *MCTestData) deleteANP(clusterName, namespace, name string) error {
 	return fmt.Errorf("clusterName %s not found", clusterName)
 }
 
+func (data *MCTestData) createOrUpdateACNP(clusterName string, acnp *crdv1alpha1.ClusterNetworkPolicy) (*crdv1alpha1.ClusterNetworkPolicy, error) {
+	if d, ok := data.clusterTestDataMap[clusterName]; ok {
+		return d.CreateOrUpdateACNP(acnp)
+	}
+	return nil, fmt.Errorf("clusterName %s not found", clusterName)
+}
+
+// deleteACNP is a convenience function for deleting ACNP by name.
+func (data *MCTestData) deleteACNP(clusterName, name string) error {
+	if d, ok := data.clusterTestDataMap[clusterName]; ok {
+		return d.DeleteACNP(name)
+	}
+	return fmt.Errorf("clusterName %s not found", clusterName)
+}
+
 // podWaitFor polls the K8s apiserver until the specified Pod is found (in the test Namespace) and
 // the condition predicate is met (or until the provided timeout expires).
 func (data *MCTestData) podWaitFor(timeout time.Duration, clusterName, name, namespace string, condition antreae2e.PodCondition) (*corev1.Pod, error) {
diff --git a/multicluster/test/e2e/main_test.go b/multicluster/test/e2e/main_test.go
index 8f9a6bd3c..5c40ec8d6 100644
--- a/multicluster/test/e2e/main_test.go
+++ b/multicluster/test/e2e/main_test.go
@@ -106,12 +106,16 @@ func TestConnectivity(t *testing.T) {
 		time.Sleep(5 * time.Second)
 	}
 
-	t.Run("TestMCServiceExport", func(t *testing.T) {
+	t.Run("TestMCService", func(t *testing.T) {
 		defer tearDownForServiceExportsTest(t, data)
 		initializeForServiceExportsTest(t, data)
 		t.Run("Case=MCServiceConnectivity", func(t *testing.T) { testMCServiceConnectivity(t, data) })
 		t.Run("Case=ScaleDownMCServiceEndpoints", func(t *testing.T) { testScaleDownMCServiceEndpoints(t, data) })
 		t.Run("Case=ANPToServices", func(t *testing.T) { testANPToServices(t, data) })
+		t.Run("Case=StretchedNetworkPolicy", func(t *testing.T) { testStretchedNetworkPolicy(t, data) })
+		t.Run("Case=StretchedNetworkPolicyUpdatePod", func(t *testing.T) { testStretchedNetworkPolicyUpdatePod(t, data) })
+		t.Run("Case=StretchedNetworkPolicyUpdateNS", func(t *testing.T) { testStretchedNetworkPolicyUpdateNS(t, data) })
+		t.Run("Case=StretchedNetworkPolicyUpdatePolicy", func(t *testing.T) { testStretchedNetworkPolicyUpdatePolicy(t, data) })
 	})
 
 	t.Run("TestAntreaPolicy", func(t *testing.T) {
diff --git a/multicluster/test/e2e/service_test.go b/multicluster/test/e2e/service_test.go
index 40572d043..ce7c731ac 100644
--- a/multicluster/test/e2e/service_test.go
+++ b/multicluster/test/e2e/service_test.go
@@ -22,6 +22,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/assert"
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 
@@ -106,10 +107,15 @@ func testMCServiceConnectivity(t *testing.T, data *MCTestData) {
 func testScaleDownMCServiceEndpoints(t *testing.T, data *MCTestData) {
 	deletePodWrapper(t, data, eastCluster, multiClusterTestNamespace, testServerPod)
 	time.Sleep(2 * time.Second)
-	mcServiceName := fmt.Sprintf("antrea-mc-%s", eastClusterTestService)
-	_, err := data.getService(westCluster, multiClusterTestNamespace, mcServiceName)
+	_, err := data.getService(westCluster, multiClusterTestNamespace, mcEastClusterTestService)
 	if !apierrors.IsNotFound(err) {
-		t.Fatalf("Expected to get not found error when getting the imported Service %s, but got: %v", mcServiceName, err)
+		t.Fatalf("Expected to get not found error when getting the imported Service %s, but got: %v", mcEastClusterTestService, err)
+	}
+
+	// Revert changes, because we shouldn't make changes to the test env that is also
+	// used by other test cases.
+	if err = createPodWrapper(t, data, eastCluster, multiClusterTestNamespace, testServerPod, "", nginxImage, "nginx", nil, nil, nil, nil, false, nil); err != nil {
+		t.Fatalf("Error when creating nginx Pod in cluster %s: %v", eastCluster, err)
 	}
 }
 
@@ -117,6 +123,20 @@ func testANPToServices(t *testing.T, data *MCTestData) {
 	data.testANPToServices(t)
 }
 
+func testStretchedNetworkPolicy(t *testing.T, data *MCTestData) {
+	data.testStretchedNetworkPolicy(t)
+}
+
+func testStretchedNetworkPolicyUpdatePod(t *testing.T, data *MCTestData) {
+	data.testStretchedNetworkPolicyUpdatePod(t)
+}
+func testStretchedNetworkPolicyUpdateNS(t *testing.T, data *MCTestData) {
+	data.testStretchedNetworkPolicyUpdateNS(t)
+}
+func testStretchedNetworkPolicyUpdatePolicy(t *testing.T, data *MCTestData) {
+	data.testStretchedNetworkPolicyUpdatePolicy(t)
+}
+
 func (data *MCTestData) testMCServiceConnectivity(t *testing.T) {
 	data.probeMCServiceFromCluster(t, eastCluster, westClusterTestService)
 	data.probeMCServiceFromCluster(t, westCluster, eastClusterTestService)
@@ -142,9 +162,9 @@ func (data *MCTestData) probeMCServiceFromCluster(t *testing.T, clusterName stri
 }
 
 func (data *MCTestData) testANPToServices(t *testing.T) {
-	svc, err := data.getService(eastCluster, multiClusterTestNamespace, fmt.Sprintf("antrea-mc-%s", westClusterTestService))
+	svc, err := data.getService(eastCluster, multiClusterTestNamespace, mcWestClusterTestService)
 	if err != nil {
-		t.Fatalf("Error when getting the imported Service %s: %v", fmt.Sprintf("antrea-mc-%s", westClusterTestService), err)
+		t.Fatalf("Error when getting the imported Service %s: %v", mcWestClusterTestService, err)
 	}
 	eastIP := svc.Spec.ClusterIP
 	eastGwClientName := getClusterGatewayClientPodName(eastCluster)
@@ -156,7 +176,7 @@ func (data *MCTestData) testANPToServices(t *testing.T) {
 		SetPriority(1.0).
 		SetAppliedToGroup([]e2euttils.ANPAppliedToSpec{{PodSelector: map[string]string{"app": "client"}}}).
 		AddToServicesRule([]crdv1alpha1.NamespacedName{{
-			Name:      fmt.Sprintf("antrea-mc-%s", westClusterTestService),
+			Name:      mcWestClusterTestService,
 			Namespace: multiClusterTestNamespace},
 		}, "", nil, crdv1alpha1.RuleActionDrop)
 	if _, err := data.createOrUpdateANP(eastCluster, anpBuilder.Get()); err != nil {
@@ -164,21 +184,176 @@ func (data *MCTestData) testANPToServices(t *testing.T) {
 	}
 	defer data.deleteANP(eastCluster, multiClusterTestNamespace, anpBuilder.Name)
 
-	connectivity := data.probeFromPodInCluster(eastCluster, multiClusterTestNamespace, eastGwClientName, "client", eastIP, fmt.Sprintf("antrea-mc-%s", westClusterTestService), 80, corev1.ProtocolTCP)
-	if connectivity == antreae2e.Error {
-		t.Errorf("Failure -- could not complete probeFromPodInCluster: %v", err)
-	} else if connectivity != antreae2e.Dropped {
-		t.Errorf("Failure -- wrong result from probing exported Service from gateway clientPod after applying toServices AntreaNetworkPolicy. Expected: %v, Actual: %v", antreae2e.Dropped, connectivity)
+	connectivity := data.probeFromPodInCluster(eastCluster, multiClusterTestNamespace, eastGwClientName, "client", eastIP, mcWestClusterTestService, 80, corev1.ProtocolTCP)
+	assert.Equal(t, antreae2e.Dropped, connectivity, "Failure -- wrong result from probing exported Service from gateway clientPod after applying toServices AntreaNetworkPolicy")
+
+	connectivity = data.probeFromPodInCluster(eastCluster, multiClusterTestNamespace, eastRegularClientName, "client", eastIP, mcWestClusterTestService, 80, corev1.ProtocolTCP)
+	assert.Equal(t, antreae2e.Dropped, connectivity, "Failure -- wrong result from probing exported Service from regular clientPod after applying toServices AntreaNetworkPolicy")
+}
+
+func (data *MCTestData) testStretchedNetworkPolicy(t *testing.T) {
+	westExpSvc, err := data.getService(eastCluster, multiClusterTestNamespace, mcWestClusterTestService)
+	if err != nil {
+		t.Fatalf("Error when getting the imported Service %s: %v", mcWestClusterTestService, err)
+	}
+	westExpSvcIP := westExpSvc.Spec.ClusterIP
+	eastGwClientName := getClusterGatewayClientPodName(eastCluster)
+	eastRegularClientName := getClusterRegularClientPodName(eastCluster)
+
+	// Verify that Stretched NetworkPolicy works fine with podSelect or podSelect+nsSelector.
+	acnpBuilder1 := &e2euttils.ClusterNetworkPolicySpecBuilder{}
+	acnpBuilder1 = acnpBuilder1.SetName("drop-client-pod-sel").
+		SetPriority(1.0).
+		SetAppliedToGroup([]e2euttils.ACNPAppliedToSpec{{PodSelector: map[string]string{"app": "nginx"}}}).
+		AddStretchedIngressRule(map[string]string{"antrea-e2e": eastGwClientName}, nil, "", nil, crdv1alpha1.RuleActionReject).
+		AddStretchedIngressRule(map[string]string{"antrea-e2e": eastRegularClientName}, map[string]string{"kubernetes.io/metadata.name": multiClusterTestNamespace}, "", nil, crdv1alpha1.RuleActionReject)
+	if _, err := data.createOrUpdateACNP(westCluster, acnpBuilder1.Get()); err != nil {
+		t.Fatalf("Error creating ACNP %s: %v", acnpBuilder1.Name, err)
+	}
+
+	connectivity := data.probeFromPodInCluster(eastCluster, multiClusterTestNamespace, eastGwClientName, "client", westExpSvcIP, mcWestClusterTestService, 80, corev1.ProtocolTCP)
+	assert.Equal(t, antreae2e.Rejected, connectivity, getStretchedNetworkPolicyErrorMessage(eastGwClientName))
+
+	connectivity = data.probeFromPodInCluster(eastCluster, multiClusterTestNamespace, eastRegularClientName, "client", westExpSvcIP, mcWestClusterTestService, 80, corev1.ProtocolTCP)
+	assert.Equal(t, antreae2e.Rejected, connectivity, getStretchedNetworkPolicyErrorMessage(eastRegularClientName))
+	data.deleteACNP(westCluster, acnpBuilder1.Name)
+
+	// Verify that Stretched NetworkPolicy works fine with nsSelect.
+	acnpBuilder2 := &e2euttils.ClusterNetworkPolicySpecBuilder{}
+	acnpBuilder2 = acnpBuilder2.SetName("drop-client-ns-sel").
+		SetPriority(1.0).
+		SetAppliedToGroup([]e2euttils.ACNPAppliedToSpec{{PodSelector: map[string]string{"app": "nginx"}}}).
+		AddStretchedIngressRule(nil, map[string]string{"kubernetes.io/metadata.name": multiClusterTestNamespace}, "", nil, crdv1alpha1.RuleActionDrop)
+
+	if _, err := data.createOrUpdateACNP(westCluster, acnpBuilder2.Get()); err != nil {
+		t.Fatalf("Error creating ACNP %s: %v", acnpBuilder2.Name, err)
+	}
+	defer data.deleteACNP(westCluster, acnpBuilder2.Name)
+
+	connectivity = data.probeFromPodInCluster(eastCluster, multiClusterTestNamespace, eastGwClientName, "client", westExpSvcIP, mcWestClusterTestService, 80, corev1.ProtocolTCP)
+	assert.Equal(t, antreae2e.Dropped, connectivity, getStretchedNetworkPolicyErrorMessage(eastGwClientName))
+
+	connectivity = data.probeFromPodInCluster(eastCluster, multiClusterTestNamespace, eastRegularClientName, "client", westExpSvcIP, mcWestClusterTestService, 80, corev1.ProtocolTCP)
+	assert.Equal(t, antreae2e.Dropped, connectivity, getStretchedNetworkPolicyErrorMessage(eastRegularClientName))
+}
+
+func (data *MCTestData) testStretchedNetworkPolicyUpdatePod(t *testing.T) {
+	westExpSvc, err := data.getService(eastCluster, multiClusterTestNamespace, mcWestClusterTestService)
+	if err != nil {
+		t.Fatalf("Error when getting the imported Service %s: %v", mcWestClusterTestService, err)
+	}
+	westExpSvcIP := westExpSvc.Spec.ClusterIP
+	eastRegularClientName := getClusterRegularClientPodName(eastCluster)
+
+	// Create a Stretched NetworkPolicy that doesn't select any Pods.
+	acnpBuilder := &e2euttils.ClusterNetworkPolicySpecBuilder{}
+	acnpBuilder = acnpBuilder.SetName("drop-client-pod-update").
+		SetPriority(1.0).
+		SetAppliedToGroup([]e2euttils.ACNPAppliedToSpec{{PodSelector: map[string]string{"app": "nginx"}}}).
+		AddStretchedIngressRule(map[string]string{"antrea-e2e": eastRegularClientName, "foo": "bar"}, nil, "", nil, crdv1alpha1.RuleActionDrop)
+	if _, err := data.createOrUpdateACNP(westCluster, acnpBuilder.Get()); err != nil {
+		t.Fatalf("Error creating ACNP %s: %v", acnpBuilder.Name, err)
+	}
+	defer data.deleteACNP(westCluster, acnpBuilder.Name)
+
+	connectivity := data.probeFromPodInCluster(eastCluster, multiClusterTestNamespace, eastRegularClientName, "client", westExpSvcIP, mcWestClusterTestService, 80, corev1.ProtocolTCP)
+	assert.Equal(t, antreae2e.Connected, connectivity, getStretchedNetworkPolicyErrorMessage(eastRegularClientName))
+
+	// Update Pod Label to match the Stretched NetworkPolicy selector.
+	if err = data.updatePod(eastCluster, multiClusterTestNamespace, eastRegularClientName, func(pod *corev1.Pod) { pod.Labels["foo"] = "bar" }); err != nil {
+		t.Errorf("Failure -- fail to update eastRegularClientPod: %v", err)
+	}
+	connectivity = data.probeFromPodInCluster(eastCluster, multiClusterTestNamespace, eastRegularClientName, "client", westExpSvcIP, mcWestClusterTestService, 80, corev1.ProtocolTCP)
+	assert.Equal(t, antreae2e.Dropped, connectivity, getStretchedNetworkPolicyErrorMessage(eastRegularClientName))
+
+	// Revert Pod Label update to test this Pod won't be selected again.
+	if err = data.updatePod(eastCluster, multiClusterTestNamespace, eastRegularClientName, func(pod *corev1.Pod) { delete(pod.Labels, "foo") }); err != nil {
+		t.Errorf("Failure -- fail to update eastRegularClientPod: %v", err)
+	}
+	connectivity = data.probeFromPodInCluster(eastCluster, multiClusterTestNamespace, eastRegularClientName, "client", westExpSvcIP, mcWestClusterTestService, 80, corev1.ProtocolTCP)
+	assert.Equal(t, antreae2e.Connected, connectivity, getStretchedNetworkPolicyErrorMessage(eastRegularClientName))
+}
+
+func (data *MCTestData) testStretchedNetworkPolicyUpdateNS(t *testing.T) {
+	westExpSvc, err := data.getService(eastCluster, multiClusterTestNamespace, mcWestClusterTestService)
+	if err != nil {
+		t.Fatalf("Error when getting the imported Service %s: %v", mcWestClusterTestService, err)
+	}
+	westExpSvcIP := westExpSvc.Spec.ClusterIP
+	eastGwClientName := getClusterGatewayClientPodName(eastCluster)
+	eastRegularClientName := getClusterRegularClientPodName(eastCluster)
+
+	// Verify that Stretched NetworkPolicy works fine with nsSelector.
+	acnpBuilder := &e2euttils.ClusterNetworkPolicySpecBuilder{}
+	acnpBuilder = acnpBuilder.SetName("drop-client-ns-update").
+		SetPriority(1.0).
+		SetAppliedToGroup([]e2euttils.ACNPAppliedToSpec{{PodSelector: map[string]string{"app": "nginx"}}}).
+		AddStretchedIngressRule(nil, map[string]string{"kubernetes.io/metadata.name": multiClusterTestNamespace, "foo": "bar"}, "", nil, crdv1alpha1.RuleActionDrop)
+
+	if _, err := data.createOrUpdateACNP(westCluster, acnpBuilder.Get()); err != nil {
+		t.Fatalf("Error creating ACNP %s: %v", acnpBuilder.Name, err)
+	}
+	defer data.deleteACNP(westCluster, acnpBuilder.Name)
+
+	connectivity := data.probeFromPodInCluster(eastCluster, multiClusterTestNamespace, eastGwClientName, "client", westExpSvcIP, mcWestClusterTestService, 80, corev1.ProtocolTCP)
+	assert.Equal(t, antreae2e.Connected, connectivity, getStretchedNetworkPolicyErrorMessage(eastGwClientName))
+
+	connectivity = data.probeFromPodInCluster(eastCluster, multiClusterTestNamespace, eastRegularClientName, "client", westExpSvcIP, mcWestClusterTestService, 80, corev1.ProtocolTCP)
+	assert.Equal(t, antreae2e.Connected, connectivity, getStretchedNetworkPolicyErrorMessage(eastRegularClientName))
+
+	// Update NS label to match the Stretched NetworkPolicy selector.
+	if err = data.updateNamespace(eastCluster, multiClusterTestNamespace, func(ns *corev1.Namespace) { ns.Labels["foo"] = "bar" }); err != nil {
+		t.Errorf("Failure -- fail to update multiClusterTestNamespace: %v", err)
 	}
 
-	connectivity = data.probeFromPodInCluster(eastCluster, multiClusterTestNamespace, eastRegularClientName, "client", eastIP, fmt.Sprintf("antrea-mc-%s", westClusterTestService), 80, corev1.ProtocolTCP)
-	if connectivity == antreae2e.Error {
-		t.Errorf("Failure -- could not complete probeFromPodInCluster: %v", err)
-	} else if connectivity != antreae2e.Dropped {
-		t.Errorf("Failure -- wrong result from probing exported Service from regular clientPod after applying toServices AntreaNetworkPolicy. Expected: %v, Actual: %v", antreae2e.Dropped, connectivity)
+	connectivity = data.probeFromPodInCluster(eastCluster, multiClusterTestNamespace, eastGwClientName, "client", westExpSvcIP, mcWestClusterTestService, 80, corev1.ProtocolTCP)
+	assert.Equal(t, antreae2e.Dropped, connectivity, getStretchedNetworkPolicyErrorMessage(eastGwClientName))
+
+	connectivity = data.probeFromPodInCluster(eastCluster, multiClusterTestNamespace, eastRegularClientName, "client", westExpSvcIP, mcWestClusterTestService, 80, corev1.ProtocolTCP)
+	assert.Equal(t, antreae2e.Dropped, connectivity, getStretchedNetworkPolicyErrorMessage(eastRegularClientName))
+
+	// Revert Namespace Label update.
+	if err = data.updateNamespace(eastCluster, multiClusterTestNamespace, func(ns *corev1.Namespace) { delete(ns.Labels, "foo") }); err != nil {
+		t.Errorf("Failure -- fail to update multiClusterTestNamespace: %v", err)
 	}
 }
 
+func (data *MCTestData) testStretchedNetworkPolicyUpdatePolicy(t *testing.T) {
+	westExpSvc, err := data.getService(eastCluster, multiClusterTestNamespace, mcWestClusterTestService)
+	if err != nil {
+		t.Fatalf("Error when getting the imported Service %s: %v", mcWestClusterTestService, err)
+	}
+	westExpSvcIP := westExpSvc.Spec.ClusterIP
+	eastRegularClientName := getClusterRegularClientPodName(eastCluster)
+
+	// Create a Stretched NetworkPolicy that doesn't select any Pods.
+	acnpBuilder := &e2euttils.ClusterNetworkPolicySpecBuilder{}
+	acnpBuilder = acnpBuilder.SetName("drop-client-pod-update").
+		SetPriority(1.0).
+		SetAppliedToGroup([]e2euttils.ACNPAppliedToSpec{{PodSelector: map[string]string{"app": "nginx"}}}).
+		AddStretchedIngressRule(map[string]string{"foo": "bar"}, nil, "", nil, crdv1alpha1.RuleActionDrop)
+
+	if _, err := data.createOrUpdateACNP(westCluster, acnpBuilder.Get()); err != nil {
+		t.Fatalf("Error creating ACNP %s: %v", acnpBuilder.Name, err)
+	}
+	defer data.deleteACNP(westCluster, acnpBuilder.Name)
+
+	connectivity := data.probeFromPodInCluster(eastCluster, multiClusterTestNamespace, eastRegularClientName, "client", westExpSvcIP, mcWestClusterTestService, 80, corev1.ProtocolTCP)
+	assert.Equal(t, antreae2e.Connected, connectivity, getStretchedNetworkPolicyErrorMessage(eastRegularClientName))
+
+	// Update the policy to select the eastRegularClient.
+	acnpBuilder.AddStretchedIngressRule(map[string]string{"antrea-e2e": eastRegularClientName}, nil, "", nil, crdv1alpha1.RuleActionDrop)
+	if _, err := data.createOrUpdateACNP(westCluster, acnpBuilder.Get()); err != nil {
+		t.Fatalf("Error updateing ACNP %s: %v", acnpBuilder.Name, err)
+	}
+	connectivity = data.probeFromPodInCluster(eastCluster, multiClusterTestNamespace, eastRegularClientName, "client", westExpSvcIP, mcWestClusterTestService, 80, corev1.ProtocolTCP)
+	assert.Equal(t, antreae2e.Dropped, connectivity, getStretchedNetworkPolicyErrorMessage(eastRegularClientName))
+}
+
+func getStretchedNetworkPolicyErrorMessage(client string) string {
+	return fmt.Sprintf("Failure -- wrong result from probing exported Service from %s clientPod after applying Stretched NetworkPolicy", client)
+}
+
 func (data *MCTestData) createClientPodInCluster(t *testing.T, cluster string, nodeName string, podName string) {
 	if err := data.createPod(cluster, podName, nodeName, multiClusterTestNamespace, "client", agnhostImage,
 		[]string{"sleep", strconv.Itoa(3600)}, nil, nil, nil, false, nil); err != nil {
diff --git a/pkg/agent/controller/egress/egress_controller_test.go b/pkg/agent/controller/egress/egress_controller_test.go
index 0d241bf7e..163bd1892 100644
--- a/pkg/agent/controller/egress/egress_controller_test.go
+++ b/pkg/agent/controller/egress/egress_controller_test.go
@@ -834,3 +834,119 @@ func TestUpdateEgressStatus(t *testing.T) {
 		})
 	}
 }
+
+func TestGetEgress(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"}},
+		},
+	}
+
+	c := newFakeController(t, []runtime.Object{egress})
+	defer c.mockController.Finish()
+
+	stopCh := make(chan struct{})
+	defer close(stopCh)
+	c.crdInformerFactory.Start(stopCh)
+	c.crdInformerFactory.WaitForCacheSync(stopCh)
+	c.addEgressGroup(egressGroup)
+	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)
+	err := c.syncEgress(egress.Name)
+	require.NoError(t, err)
+
+	type args struct {
+		ns      string
+		podName string
+	}
+	tests := []struct {
+		name               string
+		args               args
+		expectedEgressName string
+		expectedErr        string
+	}{
+		{
+			name: "local egress applied on a pod",
+			args: args{
+				ns:      "ns1",
+				podName: "pod1",
+			},
+			expectedEgressName: "egressA",
+		},
+		{
+			name: "no local egress applied on a pod",
+			args: args{
+				ns:      "ns2",
+				podName: "pod2",
+			},
+			expectedErr: "no Egress applied to Pod ns2/pod2",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			gotEgressName, err := c.GetEgress(tt.args.ns, tt.args.podName)
+			if tt.expectedErr == "" {
+				require.NoError(t, err)
+			} else {
+				require.EqualError(t, err, tt.expectedErr)
+			}
+			assert.Equal(t, tt.expectedEgressName, gotEgressName)
+		})
+	}
+}
+
+func TestGetEgressIPByMark(t *testing.T) {
+	egress := &crdv1a2.Egress{
+		ObjectMeta: metav1.ObjectMeta{Name: "egressA", UID: "uidA"},
+		Spec:       crdv1a2.EgressSpec{EgressIP: fakeLocalEgressIP1},
+	}
+
+	c := newFakeController(t, []runtime.Object{egress})
+	defer c.mockController.Finish()
+
+	stopCh := make(chan struct{})
+	defer close(stopCh)
+	c.crdInformerFactory.Start(stopCh)
+	c.crdInformerFactory.WaitForCacheSync(stopCh)
+	c.mockOFClient.EXPECT().InstallSNATMarkFlows(net.ParseIP(fakeLocalEgressIP1), uint32(1))
+	c.mockRouteClient.EXPECT().AddSNATRule(net.ParseIP(fakeLocalEgressIP1), uint32(1))
+	c.mockIPAssigner.EXPECT().UnassignIP(fakeLocalEgressIP1)
+	err := c.syncEgress(egress.Name)
+	require.NoError(t, err)
+
+	tests := []struct {
+		name             string
+		mark             uint32
+		expectedEgressIP string
+		expectedErr      string
+	}{
+		{
+			name:             "snatMark associated with local egressIP",
+			mark:             uint32(1),
+			expectedEgressIP: fakeLocalEgressIP1,
+		},
+		{
+			name:        "snatMark not associated with any local egressIP",
+			mark:        uint32(2),
+			expectedErr: "no EgressIP associated with mark 2",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			gotEgressIP, err := c.GetEgressIPByMark(tt.mark)
+			if tt.expectedErr == "" {
+				require.NoError(t, err)
+			} else {
+				require.EqualError(t, err, tt.expectedErr)
+			}
+			assert.Equal(t, tt.expectedEgressIP, gotEgressIP)
+		})
+	}
+}
diff --git a/pkg/agent/controller/networkpolicy/cache_test.go b/pkg/agent/controller/networkpolicy/cache_test.go
index f0d31e898..5ffa0070e 100644
--- a/pkg/agent/controller/networkpolicy/cache_test.go
+++ b/pkg/agent/controller/networkpolicy/cache_test.go
@@ -627,6 +627,14 @@ func TestRuleCacheAddNetworkPolicy(t *testing.T) {
 		Services:        nil,
 		Priority:        0,
 	}
+	networkPolicyRule4 := &v1beta2.NetworkPolicyRule{
+		Direction:       v1beta2.DirectionIn,
+		AppliedToGroups: []string{"appliedToGroup1"},
+		From:            v1beta2.NetworkPolicyPeer{LabelIdentities: []uint32{1}},
+		To:              v1beta2.NetworkPolicyPeer{},
+		Services:        nil,
+		Priority:        0,
+	}
 	networkPolicy1 := &v1beta2.NetworkPolicy{
 		ObjectMeta:      metav1.ObjectMeta{UID: "policy1", Namespace: "ns1", Name: "name1"},
 		Rules:           nil,
@@ -659,9 +667,21 @@ func TestRuleCacheAddNetworkPolicy(t *testing.T) {
 			UID:       "policy3",
 		},
 	}
+	networkPolicy4 := &v1beta2.NetworkPolicy{
+		ObjectMeta: metav1.ObjectMeta{UID: "policy4", Namespace: "ns4", Name: "name4"},
+		Rules:      []v1beta2.NetworkPolicyRule{*networkPolicyRule4},
+		SourceRef: &v1beta2.NetworkPolicyReference{
+			Type:      v1beta2.AntreaNetworkPolicy,
+			Namespace: "ns4",
+			Name:      "name4",
+			UID:       "policy4",
+		},
+	}
 	rule1 := toRule(networkPolicyRule1, networkPolicy2, k8sNPMaxPriority)
 	rule2 := toRule(networkPolicyRule2, networkPolicy2, k8sNPMaxPriority)
 	rule3 := toRule(networkPolicyRule3, networkPolicy3, 0)
+	rule4 := toRule(networkPolicyRule4, networkPolicy4, 0)
+	rule4Sec := toStretchedNetworkPolicySecurityRule(networkPolicyRule4, networkPolicy4, 0)
 	tests := []struct {
 		name               string
 		args               *v1beta2.NetworkPolicy
@@ -686,6 +706,12 @@ func TestRuleCacheAddNetworkPolicy(t *testing.T) {
 			[]*rule{rule3},
 			sets.NewString(rule3.ID),
 		},
+		{
+			"rule-with-label-identity",
+			networkPolicy4,
+			[]*rule{rule4, rule4Sec},
+			sets.NewString(rule4.ID, rule4Sec.ID),
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
diff --git a/pkg/agent/controller/networkpolicy/reconciler_test.go b/pkg/agent/controller/networkpolicy/reconciler_test.go
index 9cfb32266..d5fc492f2 100644
--- a/pkg/agent/controller/networkpolicy/reconciler_test.go
+++ b/pkg/agent/controller/networkpolicy/reconciler_test.go
@@ -444,6 +444,30 @@ func TestReconcilerReconcile(t *testing.T) {
 			},
 			false,
 		},
+		{
+			"ingress-rule-with-label-identity",
+			&CompletedRule{
+				rule: &rule{
+					ID:        "ingress-rule",
+					Direction: v1beta2.DirectionIn,
+					From:      v1beta2.NetworkPolicyPeer{LabelIdentities: []uint32{1}},
+					SourceRef: &np1,
+				},
+				FromAddresses: nil,
+				ToAddresses:   nil,
+				TargetMembers: appliedToGroup1,
+			},
+			[]*types.PolicyRule{
+				{
+					Direction: v1beta2.DirectionIn,
+					From:      labelIDToOFAddresses([]uint32{1}),
+					To:        ofPortsToOFAddresses(sets.NewInt32(1)),
+					Service:   nil,
+					PolicyRef: &np1,
+				},
+			},
+			false,
+		},
 		{
 			"egress-rule",
 			&CompletedRule{
diff --git a/pkg/agent/controller/traceflow/packetin_test.go b/pkg/agent/controller/traceflow/packetin_test.go
index c2a5eb928..cc2ee83bc 100644
--- a/pkg/agent/controller/traceflow/packetin_test.go
+++ b/pkg/agent/controller/traceflow/packetin_test.go
@@ -15,19 +15,43 @@
 package traceflow
 
 import (
+	"encoding/binary"
 	"net"
 	"reflect"
 	"testing"
 
+	"antrea.io/libOpenflow/openflow15"
 	"antrea.io/libOpenflow/protocol"
 	"antrea.io/libOpenflow/util"
 	"antrea.io/ofnet/ofctrl"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
 
+	"antrea.io/antrea/pkg/agent/config"
 	"antrea.io/antrea/pkg/agent/openflow"
 	crdv1alpha1 "antrea.io/antrea/pkg/apis/crd/v1alpha1"
 )
 
+var (
+	egressName = "dummyEgress"
+	egressIP   = "192.168.100.100"
+)
+
+type fakeEgressQuerier struct {
+	egressName string
+	egressIP   string
+}
+
+func (f *fakeEgressQuerier) GetEgress(podNamespace, podName string) (string, error) {
+	return f.egressName, nil
+}
+
+func (f *fakeEgressQuerier) GetEgressIPByMark(mark uint32) (string, error) {
+	return f.egressIP, nil
+}
+
 func prepareMockTables() {
 	openflow.InitMockTables(
 		map[*openflow.Table]uint8{
@@ -181,3 +205,255 @@ func TestParseCapturedPacket(t *testing.T) {
 		})
 	}
 }
+
+func TestParsePacketIn(t *testing.T) {
+	xreg0 := make([]byte, 8)
+	binary.BigEndian.PutUint32(xreg0[0:4], 262144) // RemoteSNATRegMark in 32bit reg0
+	binary.BigEndian.PutUint32(xreg0[4:8], 2)      // outputPort in 32bit reg1
+	matchOutPort := &openflow15.MatchField{
+		Class: openflow15.OXM_CLASS_PACKET_REGS,
+		Field: openflow15.NXM_NX_REG0,
+		Value: &openflow15.ByteArrayField{
+			Data: xreg0,
+		},
+	}
+	matchPktMark := &openflow15.MatchField{
+		Class: openflow15.OXM_CLASS_NXM_1,
+		Field: openflow15.NXM_NX_PKT_MARK,
+		Value: &openflow15.Uint32Message{
+			Data: 1,
+		},
+	}
+	matchTunDst := openflow15.NewTunnelIpv4DstField(net.ParseIP(egressIP), nil)
+
+	ipPacket := &protocol.IPv4{
+		Version:  0x4,
+		IHL:      5,
+		Protocol: uint8(8),
+		DSCP:     1,
+		Length:   20,
+		NWSrc:    net.IP(pod1IPv4),
+		NWDst:    net.IP(dstIPv4),
+	}
+	ethernetPkt := protocol.NewEthernet()
+	ethernetPkt.HWSrc = pod1MAC
+	ethernetPkt.Ethertype = protocol.IPv4_MSG
+	ethernetPkt.Data = ipPacket
+	pktBytes, _ := ethernetPkt.MarshalBinary()
+
+	egressQuerier := &fakeEgressQuerier{
+		egressName: egressName,
+		egressIP:   egressIP,
+	}
+
+	tests := []struct {
+		name               string
+		networkConfig      *config.NetworkConfig
+		nodeConfig         *config.NodeConfig
+		tfState            *traceflowState
+		pktIn              *ofctrl.PacketIn
+		expectedTf         *crdv1alpha1.Traceflow
+		expectedNodeResult *crdv1alpha1.NodeResult
+	}{
+		{
+			name: "packet at source node for local Egress",
+			networkConfig: &config.NetworkConfig{
+				TrafficEncapMode: 0,
+			},
+			nodeConfig: &config.NodeConfig{
+				TunnelOFPort: 1,
+				GatewayConfig: &config.GatewayConfig{
+					OFPort: 2,
+				},
+			},
+			tfState: &traceflowState{
+				name:     "dummy-traceflow-pod-to-ipv4",
+				tag:      1,
+				isSender: true,
+			},
+			pktIn: &ofctrl.PacketIn{
+				TableId: openflow.L2ForwardingOutTable.GetID(),
+				Match: openflow15.Match{
+					Fields: []openflow15.MatchField{*matchOutPort, *matchPktMark},
+				},
+				Data: util.NewBuffer(pktBytes),
+			},
+			expectedTf: &crdv1alpha1.Traceflow{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "dummy-traceflow-pod-to-ipv4",
+				},
+				Spec: crdv1alpha1.TraceflowSpec{
+					Source: crdv1alpha1.Source{
+						Namespace: pod1.Namespace,
+						Pod:       pod1.Name,
+					},
+					Destination: crdv1alpha1.Destination{
+						IP: dstIPv4,
+					},
+				},
+				Status: crdv1alpha1.TraceflowStatus{
+					Phase:        crdv1alpha1.Running,
+					DataplaneTag: 1,
+				},
+			},
+			expectedNodeResult: &crdv1alpha1.NodeResult{
+				Observations: []crdv1alpha1.Observation{
+					{
+						Component: crdv1alpha1.ComponentSpoofGuard,
+						Action:    crdv1alpha1.ActionForwarded,
+					},
+					{
+						Component: crdv1alpha1.ComponentEgress,
+						Action:    crdv1alpha1.ActionMarkedForSNAT,
+						Egress:    egressName,
+						EgressIP:  egressIP,
+					},
+					{
+						Component:     crdv1alpha1.ComponentForwarding,
+						ComponentInfo: openflow.L2ForwardingOutTable.GetName(),
+						Action:        crdv1alpha1.ActionForwardedOutOfOverlay,
+					},
+				},
+			},
+		},
+		{
+			name: "packet at source node for remote Egress",
+			networkConfig: &config.NetworkConfig{
+				TrafficEncapMode: 0,
+			},
+			nodeConfig: &config.NodeConfig{
+				TunnelOFPort: 2,
+				GatewayConfig: &config.GatewayConfig{
+					OFPort: 1,
+				},
+			},
+			tfState: &traceflowState{
+				name:     "dummy-traceflow-pod-to-ipv4",
+				tag:      1,
+				isSender: true,
+			},
+			pktIn: &ofctrl.PacketIn{
+				TableId: openflow.L2ForwardingOutTable.GetID(),
+				Match: openflow15.Match{
+					Fields: []openflow15.MatchField{*matchTunDst, *matchOutPort},
+				},
+				Data: util.NewBuffer(pktBytes),
+			},
+			expectedTf: &crdv1alpha1.Traceflow{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "dummy-traceflow-pod-to-ipv4",
+				},
+				Spec: crdv1alpha1.TraceflowSpec{
+					Source: crdv1alpha1.Source{
+						Namespace: pod1.Namespace,
+						Pod:       pod1.Name,
+					},
+					Destination: crdv1alpha1.Destination{
+						IP: dstIPv4,
+					},
+				},
+				Status: crdv1alpha1.TraceflowStatus{
+					Phase:        crdv1alpha1.Running,
+					DataplaneTag: 1,
+				},
+			},
+			expectedNodeResult: &crdv1alpha1.NodeResult{
+				Observations: []crdv1alpha1.Observation{
+					{
+						Component: crdv1alpha1.ComponentSpoofGuard,
+						Action:    crdv1alpha1.ActionForwarded,
+					},
+					{
+						Component: crdv1alpha1.ComponentEgress,
+						Action:    crdv1alpha1.ActionForwardedToEgressNode,
+						Egress:    egressName,
+						EgressIP:  egressIP,
+					},
+					{
+						Component:     crdv1alpha1.ComponentForwarding,
+						ComponentInfo: openflow.L2ForwardingOutTable.GetName(),
+						Action:        crdv1alpha1.ActionForwarded,
+						TunnelDstIP:   egressIP,
+					},
+				},
+			},
+		},
+		{
+			name: "packet at remote node for remote Egress",
+			networkConfig: &config.NetworkConfig{
+				TrafficEncapMode: 0,
+			},
+			nodeConfig: &config.NodeConfig{
+				TunnelOFPort: 1,
+				GatewayConfig: &config.GatewayConfig{
+					OFPort: 2,
+				},
+			},
+			tfState: &traceflowState{
+				name: "dummy-traceflow-pod-to-ipv4",
+				tag:  1,
+			},
+			pktIn: &ofctrl.PacketIn{
+				TableId: openflow.L2ForwardingOutTable.GetID(),
+				Match: openflow15.Match{
+					Fields: []openflow15.MatchField{*matchOutPort, *matchTunDst, *matchPktMark},
+				},
+				Data: util.NewBuffer(pktBytes),
+			},
+			expectedTf: &crdv1alpha1.Traceflow{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "dummy-traceflow-pod-to-ipv4",
+				},
+				Spec: crdv1alpha1.TraceflowSpec{
+					Source: crdv1alpha1.Source{
+						Namespace: pod1.Namespace,
+						Pod:       pod1.Name,
+					},
+					Destination: crdv1alpha1.Destination{
+						IP: dstIPv4,
+					},
+				},
+				Status: crdv1alpha1.TraceflowStatus{
+					Phase:        crdv1alpha1.Running,
+					DataplaneTag: 1,
+				},
+			},
+			expectedNodeResult: &crdv1alpha1.NodeResult{
+				Observations: []crdv1alpha1.Observation{
+					{
+						Component: crdv1alpha1.ComponentForwarding,
+						Action:    crdv1alpha1.ActionReceived,
+					},
+					{
+						Component: crdv1alpha1.ComponentEgress,
+						Action:    crdv1alpha1.ActionMarkedForSNAT,
+						EgressIP:  egressIP,
+					},
+					{
+						Component:     crdv1alpha1.ComponentForwarding,
+						ComponentInfo: openflow.L2ForwardingOutTable.GetName(),
+						Action:        crdv1alpha1.ActionForwardedOutOfOverlay,
+					},
+				},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+
+			tfc := newFakeTraceflowController(t, []runtime.Object{tt.expectedTf}, tt.networkConfig, tt.nodeConfig, nil, egressQuerier)
+			defer tfc.mockController.Finish()
+			stopCh := make(chan struct{})
+			defer close(stopCh)
+			tfc.crdInformerFactory.Start(stopCh)
+			tfc.crdInformerFactory.WaitForCacheSync(stopCh)
+			tfc.runningTraceflows[tt.expectedTf.Status.DataplaneTag] = tt.tfState
+
+			tf, nodeResult, _, err := tfc.parsePacketIn(tt.pktIn)
+			require.NoError(t, err)
+			assert.Equal(t, tt.expectedNodeResult.Observations, nodeResult.Observations)
+			assert.Equal(t, tt.expectedTf, tf)
+		})
+	}
+}
diff --git a/pkg/agent/controller/traceflow/traceflow_controller_test.go b/pkg/agent/controller/traceflow/traceflow_controller_test.go
index b4bb842ad..b9fecbc01 100644
--- a/pkg/agent/controller/traceflow/traceflow_controller_test.go
+++ b/pkg/agent/controller/traceflow/traceflow_controller_test.go
@@ -45,6 +45,7 @@ import (
 var (
 	pod1IPv4   = "192.168.10.10"
 	pod2IPv4   = "192.168.11.10"
+	dstIPv4    = "192.168.99.99"
 	pod1MAC, _ = net.ParseMAC("aa:bb:cc:dd:ee:0f")
 	pod2MAC, _ = net.ParseMAC("aa:bb:cc:dd:ee:00")
 	ofPortPod1 = uint32(1)
@@ -80,7 +81,7 @@ type fakeTraceflowController struct {
 	ovsClient          *ovsconfigtest.MockOVSBridgeClient
 }
 
-func newFakeTraceflowController(t *testing.T, initObjects []runtime.Object, networkConfig *config.NetworkConfig, nodeConfig *config.NodeConfig, npQuerier querier.AgentNetworkPolicyInfoQuerier) *fakeTraceflowController {
+func newFakeTraceflowController(t *testing.T, initObjects []runtime.Object, networkConfig *config.NetworkConfig, nodeConfig *config.NodeConfig, npQuerier querier.AgentNetworkPolicyInfoQuerier, egressQuerier querier.EgressQuerier) *fakeTraceflowController {
 	controller := gomock.NewController(t)
 	kubeClient := fake.NewSimpleClientset(&pod1, &pod2)
 	mockOFClient := openflowtest.NewMockClient(controller)
@@ -103,6 +104,7 @@ func newFakeTraceflowController(t *testing.T, initObjects []runtime.Object, netw
 		traceflowListerSynced: traceflowInformer.Informer().HasSynced,
 		ofClient:              mockOFClient,
 		networkPolicyQuerier:  npQuerier,
+		egressQuerier:         egressQuerier,
 		ovsBridgeClient:       ovsClient,
 		interfaceStore:        ifaceStore,
 		networkConfig:         networkConfig,
@@ -301,7 +303,7 @@ func TestPreparePacket(t *testing.T) {
 
 	for _, tt := range tcs {
 		t.Run(tt.name, func(t *testing.T) {
-			tfc := newFakeTraceflowController(t, []runtime.Object{tt.tf}, nil, nil, nil)
+			tfc := newFakeTraceflowController(t, []runtime.Object{tt.tf}, nil, nil, nil, nil)
 			defer tfc.mockController.Finish()
 
 			podInterfaces := tfc.interfaceStore.GetContainerInterfacesByPod(pod1.Name, pod1.Namespace)
@@ -343,7 +345,7 @@ func TestErrTraceflowCRD(t *testing.T) {
 	expectedTf.Status.Phase = crdv1alpha1.Failed
 	expectedTf.Status.Reason = reason
 
-	tfc := newFakeTraceflowController(t, []runtime.Object{tf}, nil, nil, nil)
+	tfc := newFakeTraceflowController(t, []runtime.Object{tf}, nil, nil, nil, nil)
 	defer tfc.mockController.Finish()
 
 	gotTf, err := tfc.errorTraceflowCRD(tf, reason)
@@ -358,7 +360,6 @@ func TestStartTraceflow(t *testing.T) {
 		ofPort       uint32
 		receiverOnly bool
 		packet       *binding.Packet
-		expectedErr  string
 	}{
 		{
 			name: "pod-to-pod traceflow",
@@ -400,7 +401,7 @@ func TestStartTraceflow(t *testing.T) {
 						Pod:       pod1.Name,
 					},
 					Destination: crdv1alpha1.Destination{
-						IP: "192.168.13.3",
+						IP: dstIPv4,
 					},
 				},
 				Status: crdv1alpha1.TraceflowStatus{
@@ -412,7 +413,7 @@ func TestStartTraceflow(t *testing.T) {
 			packet: &binding.Packet{
 				SourceIP:      net.ParseIP(pod1IPv4),
 				SourceMAC:     pod1MAC,
-				DestinationIP: net.ParseIP("192.168.13.3"),
+				DestinationIP: net.ParseIP(dstIPv4),
 				IPProto:       1,
 				TTL:           64,
 				ICMPType:      8,
@@ -422,7 +423,7 @@ func TestStartTraceflow(t *testing.T) {
 
 	for _, tt := range tcs {
 		t.Run(tt.name, func(t *testing.T) {
-			tfc := newFakeTraceflowController(t, []runtime.Object{tt.tf}, nil, nil, nil)
+			tfc := newFakeTraceflowController(t, []runtime.Object{tt.tf}, nil, nil, nil, nil)
 			defer tfc.mockController.Finish()
 
 			stopCh := make(chan struct{})
@@ -434,11 +435,7 @@ func TestStartTraceflow(t *testing.T) {
 			tfc.mockOFClient.EXPECT().SendTraceflowPacket(tt.tf.Status.DataplaneTag, tt.packet, tt.ofPort, int32(-1))
 
 			err := tfc.startTraceflow(tt.tf)
-			if tt.expectedErr == "" {
-				require.NoError(t, err)
-			} else {
-				require.EqualError(t, err, tt.expectedErr)
-			}
+			require.NoError(t, err)
 		})
 	}
 }
@@ -505,7 +502,7 @@ func TestSyncTraceflow(t *testing.T) {
 
 	for _, tt := range tcs {
 		t.Run(tt.name, func(t *testing.T) {
-			tfc := newFakeTraceflowController(t, []runtime.Object{tt.tf}, nil, nil, nil)
+			tfc := newFakeTraceflowController(t, []runtime.Object{tt.tf}, nil, nil, nil, nil)
 			defer tfc.mockController.Finish()
 			tfc.runningTraceflows[tt.tf.Status.DataplaneTag] = tt.tfState
 			stopCh := make(chan struct{})
@@ -560,7 +557,7 @@ func TestProcessTraceflowItem(t *testing.T) {
 		},
 		expected: true,
 	}
-	tfc := newFakeTraceflowController(t, []runtime.Object{tc.tf}, nil, nil, nil)
+	tfc := newFakeTraceflowController(t, []runtime.Object{tc.tf}, nil, nil, nil, nil)
 	defer tfc.mockController.Finish()
 
 	stopCh := make(chan struct{})
diff --git a/pkg/agent/multicluster/mc_route_controller_test.go b/pkg/agent/multicluster/mc_route_controller_test.go
index 0f582fbf6..c745e5c35 100644
--- a/pkg/agent/multicluster/mc_route_controller_test.go
+++ b/pkg/agent/multicluster/mc_route_controller_test.go
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package noderoute
+package multicluster
 
 import (
 	"context"
@@ -60,6 +60,7 @@ func newMCRouteController(t *testing.T, nodeConfig *config.NodeConfig) (*fakeRou
 		interfaceStore,
 		nodeConfig,
 		"default",
+		true,
 	)
 	return &fakeRouteController{
 		MCRouteController: c,
@@ -153,14 +154,14 @@ func TestMCRouteControllerAsGateway(t *testing.T) {
 			Create(context.TODO(), &clusterInfoImport1, metav1.CreateOptions{})
 		peerNodeIP1 := getPeerGatewayIP(clusterInfoImport1.Spec)
 		c.ofClient.EXPECT().InstallMulticlusterGatewayFlows(clusterInfoImport1.Name,
-			gomock.Any(), peerNodeIP1, gw1GatewayIP).Times(1)
+			gomock.Any(), peerNodeIP1, gw1GatewayIP, true).Times(1)
 		c.processNextWorkItem()
 
 		c.mcClient.MulticlusterV1alpha1().ClusterInfoImports(clusterInfoImport2.GetNamespace()).
 			Create(context.TODO(), &clusterInfoImport2, metav1.CreateOptions{})
 		peerNodeIP2 := getPeerGatewayIP(clusterInfoImport2.Spec)
 		c.ofClient.EXPECT().InstallMulticlusterGatewayFlows(clusterInfoImport2.Name,
-			gomock.Any(), peerNodeIP2, gw1GatewayIP).Times(1)
+			gomock.Any(), peerNodeIP2, gw1GatewayIP, true).Times(1)
 		c.processNextWorkItem()
 
 		// Update a ClusterInfoImport
@@ -168,7 +169,7 @@ func TestMCRouteControllerAsGateway(t *testing.T) {
 		c.mcClient.MulticlusterV1alpha1().ClusterInfoImports(clusterInfoImport1.GetNamespace()).
 			Update(context.TODO(), &clusterInfoImport1, metav1.UpdateOptions{})
 		c.ofClient.EXPECT().InstallMulticlusterGatewayFlows(clusterInfoImport1.Name,
-			gomock.Any(), peerNodeIP1, gw1GatewayIP).Times(1)
+			gomock.Any(), peerNodeIP1, gw1GatewayIP, true).Times(1)
 		c.processNextWorkItem()
 
 		// Delete a ClusterInfoImport
@@ -184,7 +185,7 @@ func TestMCRouteControllerAsGateway(t *testing.T) {
 		c.mcClient.MulticlusterV1alpha1().Gateways(updatedGateway1a.GetNamespace()).Update(context.TODO(),
 			updatedGateway1a, metav1.UpdateOptions{})
 		c.ofClient.EXPECT().InstallMulticlusterGatewayFlows(clusterInfoImport1.Name,
-			gomock.Any(), peerNodeIP1, updatedGateway1aIP).Times(1)
+			gomock.Any(), peerNodeIP1, updatedGateway1aIP, true).Times(1)
 		c.processNextWorkItem()
 
 		// Update Gateway1's InternalIP
@@ -199,7 +200,7 @@ func TestMCRouteControllerAsGateway(t *testing.T) {
 			&gateway2, metav1.CreateOptions{})
 		c.ofClient.EXPECT().UninstallMulticlusterFlows(clusterInfoImport1.Name).Times(1)
 		c.ofClient.EXPECT().InstallMulticlusterClassifierFlows(uint32(1), false).Times(1)
-		c.ofClient.EXPECT().InstallMulticlusterNodeFlows(clusterInfoImport1.Name, gomock.Any(), gw2InternalIP).Times(1)
+		c.ofClient.EXPECT().InstallMulticlusterNodeFlows(clusterInfoImport1.Name, gomock.Any(), gw2InternalIP, true).Times(1)
 		c.processNextWorkItem()
 
 		// Delete Gateway2, then Gateway1 become active Gateway
@@ -208,7 +209,7 @@ func TestMCRouteControllerAsGateway(t *testing.T) {
 		c.ofClient.EXPECT().UninstallMulticlusterFlows(clusterInfoImport1.Name).Times(1)
 		c.ofClient.EXPECT().InstallMulticlusterClassifierFlows(uint32(1), true).Times(1)
 		c.ofClient.EXPECT().InstallMulticlusterGatewayFlows(clusterInfoImport1.Name,
-			gomock.Any(), peerNodeIP1, updatedGateway1aIP).Times(1)
+			gomock.Any(), peerNodeIP1, updatedGateway1aIP, true).Times(1)
 		c.processNextWorkItem()
 
 		// Delete last Gateway
@@ -250,13 +251,13 @@ func TestMCRouteControllerAsRegularNode(t *testing.T) {
 		c.mcClient.MulticlusterV1alpha1().ClusterInfoImports(clusterInfoImport1.GetNamespace()).
 			Create(context.TODO(), &clusterInfoImport1, metav1.CreateOptions{})
 		c.ofClient.EXPECT().InstallMulticlusterNodeFlows(clusterInfoImport1.Name,
-			gomock.Any(), peerNodeIP1).Times(1)
+			gomock.Any(), peerNodeIP1, true).Times(1)
 		c.processNextWorkItem()
 
 		c.mcClient.MulticlusterV1alpha1().ClusterInfoImports(clusterInfoImport2.GetNamespace()).
 			Create(context.TODO(), &clusterInfoImport2, metav1.CreateOptions{})
 		c.ofClient.EXPECT().InstallMulticlusterNodeFlows(clusterInfoImport2.Name,
-			gomock.Any(), peerNodeIP1).Times(1)
+			gomock.Any(), peerNodeIP1, true).Times(1)
 		c.processNextWorkItem()
 
 		// Update a ClusterInfoImport
@@ -264,7 +265,7 @@ func TestMCRouteControllerAsRegularNode(t *testing.T) {
 		c.mcClient.MulticlusterV1alpha1().ClusterInfoImports(clusterInfoImport1.GetNamespace()).
 			Update(context.TODO(), &clusterInfoImport1, metav1.UpdateOptions{})
 		c.ofClient.EXPECT().InstallMulticlusterNodeFlows(clusterInfoImport1.Name,
-			gomock.Any(), peerNodeIP1).Times(1)
+			gomock.Any(), peerNodeIP1, true).Times(1)
 		c.processNextWorkItem()
 
 		// Delete a ClusterInfoImport
@@ -287,14 +288,14 @@ func TestMCRouteControllerAsRegularNode(t *testing.T) {
 		c.mcClient.MulticlusterV1alpha1().Gateways(updatedGateway1b.GetNamespace()).Update(context.TODO(),
 			updatedGateway1b, metav1.UpdateOptions{})
 		c.ofClient.EXPECT().InstallMulticlusterNodeFlows(clusterInfoImport1.Name,
-			gomock.Any(), updatedGateway1bIP).Times(1)
+			gomock.Any(), updatedGateway1bIP, true).Times(1)
 		c.processNextWorkItem()
 
 		// Create Gateway2 as the active Gateway
 		c.mcClient.MulticlusterV1alpha1().Gateways(gateway2.GetNamespace()).Create(context.TODO(),
 			&gateway2, metav1.CreateOptions{})
 		c.ofClient.EXPECT().InstallMulticlusterClassifierFlows(uint32(1), false).Times(1)
-		c.ofClient.EXPECT().InstallMulticlusterNodeFlows(clusterInfoImport1.Name, gomock.Any(), peerNodeIP2).Times(1)
+		c.ofClient.EXPECT().InstallMulticlusterNodeFlows(clusterInfoImport1.Name, gomock.Any(), peerNodeIP2, true).Times(1)
 		c.ofClient.EXPECT().UninstallMulticlusterFlows(clusterInfoImport1.Name).Times(1)
 		c.processNextWorkItem()
 
@@ -304,7 +305,7 @@ func TestMCRouteControllerAsRegularNode(t *testing.T) {
 		c.ofClient.EXPECT().UninstallMulticlusterFlows(clusterInfoImport1.Name).Times(1)
 		c.ofClient.EXPECT().InstallMulticlusterClassifierFlows(uint32(1), false).Times(1)
 		c.ofClient.EXPECT().InstallMulticlusterNodeFlows(clusterInfoImport1.Name,
-			gomock.Any(), updatedGateway1bIP).Times(1)
+			gomock.Any(), updatedGateway1bIP, true).Times(1)
 		c.processNextWorkItem()
 
 		// Delete last Gateway
diff --git a/pkg/agent/multicluster/stretched_networkpolicy_controller_test.go b/pkg/agent/multicluster/stretched_networkpolicy_controller_test.go
new file mode 100644
index 000000000..8a259ae59
--- /dev/null
+++ b/pkg/agent/multicluster/stretched_networkpolicy_controller_test.go
@@ -0,0 +1,587 @@
+// 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 multicluster
+
+import (
+	"context"
+	"net"
+	"testing"
+	"time"
+
+	"github.com/golang/mock/gomock"
+	"github.com/stretchr/testify/assert"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/fields"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/informers"
+	coreinformers "k8s.io/client-go/informers/core/v1"
+	"k8s.io/client-go/kubernetes/fake"
+	"k8s.io/client-go/tools/cache"
+
+	"antrea.io/antrea/multicluster/apis/multicluster/v1alpha1"
+	mcfake "antrea.io/antrea/multicluster/pkg/client/clientset/versioned/fake"
+	mcinformers "antrea.io/antrea/multicluster/pkg/client/informers/externalversions"
+	"antrea.io/antrea/pkg/agent/interfacestore"
+	interfacestoretest "antrea.io/antrea/pkg/agent/interfacestore/testing"
+	"antrea.io/antrea/pkg/agent/openflow"
+	oftest "antrea.io/antrea/pkg/agent/openflow/testing"
+	antreatypes "antrea.io/antrea/pkg/agent/types"
+	ovsconfigtest "antrea.io/antrea/pkg/ovs/ovsconfig/testing"
+	"antrea.io/antrea/pkg/util/channel"
+)
+
+const (
+	interval = time.Millisecond
+	timeout  = time.Second
+)
+
+type fakeStretchedNetworkPolicyController struct {
+	*StretchedNetworkPolicyController
+	clientset         *fake.Clientset
+	mcClient          *mcfake.Clientset
+	informerFactory   informers.SharedInformerFactory
+	mcInformerFactory mcinformers.SharedInformerFactory
+	ofClient          *oftest.MockClient
+	ovsClient         *ovsconfigtest.MockOVSBridgeClient
+	interfaceStore    *interfacestoretest.MockInterfaceStore
+	podUpdateChannel  *channel.SubscribableChannel
+}
+
+func newStretchedNetworkPolicyController(t *testing.T, clientset *fake.Clientset, mcClient *mcfake.Clientset) (*fakeStretchedNetworkPolicyController, func()) {
+	informerFactory := informers.NewSharedInformerFactory(clientset, 12*time.Hour)
+	listOptions := func(options *metav1.ListOptions) {
+		options.FieldSelector = fields.OneTermEqualSelector("spec.nodeName", "test-node").String()
+	}
+	podInformer := coreinformers.NewFilteredPodInformer(
+		clientset,
+		metav1.NamespaceAll,
+		0,
+		cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}, // NamespaceIndex is used in NPLController.
+		listOptions,
+	)
+	nsInformer := informerFactory.Core().V1().Namespaces()
+	mcInformerFactory := mcinformers.NewSharedInformerFactory(mcClient, 60*time.Second)
+	labelIDInformer := mcInformerFactory.Multicluster().V1alpha1().LabelIdentities()
+
+	podUpdateChannel := channel.NewSubscribableChannel("PodUpdate", 100)
+	ctrl := gomock.NewController(t)
+	ofClient := oftest.NewMockClient(ctrl)
+	ovsClient := ovsconfigtest.NewMockOVSBridgeClient(ctrl)
+	interfaceStore := interfacestoretest.NewMockInterfaceStore(ctrl)
+	c := NewMCAgentStretchedNetworkPolicyController(
+		ofClient,
+		interfaceStore,
+		podInformer,
+		nsInformer,
+		labelIDInformer,
+		podUpdateChannel,
+	)
+	return &fakeStretchedNetworkPolicyController{
+		StretchedNetworkPolicyController: c,
+		clientset:                        clientset,
+		mcClient:                         mcClient,
+		informerFactory:                  informerFactory,
+		mcInformerFactory:                mcInformerFactory,
+		ofClient:                         ofClient,
+		ovsClient:                        ovsClient,
+		interfaceStore:                   interfaceStore,
+		podUpdateChannel:                 podUpdateChannel,
+	}, ctrl.Finish
+}
+
+var (
+	interfaceConfig = interfacestore.InterfaceConfig{
+		InterfaceName: "foo",
+		OVSPortConfig: &interfacestore.OVSPortConfig{
+			OFPort: 2,
+		},
+		IPs: []net.IP{[]byte("1.1.1.1")},
+	}
+	unknownLabelIdentity = openflow.UnknownLabelIdentity
+)
+
+func TestEnqueueAllPods(t *testing.T) {
+	ns := &corev1.Namespace{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "ns",
+			Labels: map[string]string{
+				"env": "test",
+			},
+		},
+	}
+	pod := &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "pod",
+			Namespace: "ns",
+			Labels: map[string]string{
+				"foo": "bar1",
+			},
+		},
+		Spec: corev1.PodSpec{
+			NodeName: "test-node",
+		},
+	}
+	labelIdentity := &v1alpha1.LabelIdentity{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "labelIdentity1",
+		},
+		Spec: v1alpha1.LabelIdentitySpec{
+			Label: "ns:env=test,kubernetes.io/metadata.name=ns&pod:foo=bar1",
+			ID:    1,
+		},
+	}
+
+	clientset := fake.NewSimpleClientset(ns, pod)
+	mcClient := mcfake.NewSimpleClientset(labelIdentity)
+	c, closeFn := newStretchedNetworkPolicyController(t, clientset, mcClient)
+	defer closeFn()
+	defer c.queue.ShutDown()
+
+	stopCh := make(chan struct{})
+	defer close(stopCh)
+	c.informerFactory.Start(stopCh)
+	c.informerFactory.WaitForCacheSync(stopCh)
+	c.mcInformerFactory.Start(stopCh)
+	c.mcInformerFactory.WaitForCacheSync(stopCh)
+	go c.podInformer.Run(stopCh)
+	if err := waitForPodRealized(c, pod); err != nil {
+		t.Errorf("error when waiting for Pod '%s/%s' to be realized: %v", pod.Namespace, pod.Name, err)
+	}
+	if err := waitForLabelIdentityRealized(c, labelIdentity); err != nil {
+		t.Errorf("error when waiting for LabelIdentity '%s' to be realized: %v", labelIdentity.Name, err)
+	}
+	c.enqueueAllPods()
+
+	finishCh := make(chan struct{})
+	go func() {
+		defer close(finishCh)
+		c.interfaceStore.EXPECT().GetContainerInterfacesByPod(pod.Name, pod.Namespace).Return([]*interfacestore.InterfaceConfig{&interfaceConfig}).Times(1)
+		c.ofClient.EXPECT().InstallPodFlows(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Eq(&labelIdentity.Spec.ID)).Times(1)
+		c.processNextWorkItem()
+		assert.Equal(t, map[types.NamespacedName]string{{Name: pod.Name, Namespace: pod.Namespace}: labelIdentity.Spec.Label}, c.podToLabel)
+		assert.Equal(t, map[string]podSet{labelIdentity.Spec.Label: {types.NamespacedName{Name: pod.Name, Namespace: pod.Namespace}: struct{}{}}}, c.labelToPods)
+	}()
+	select {
+	case <-time.After(5 * time.Second):
+		t.Errorf("Test didn't finish in time")
+	case <-finishCh:
+	}
+}
+
+func TestStretchedNetworkPolicyControllerPodEvent(t *testing.T) {
+	ns := &corev1.Namespace{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "ns",
+			Labels: map[string]string{
+				"env": "test",
+			},
+		},
+	}
+	pod := &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "pod",
+			Namespace: "ns",
+			Labels: map[string]string{
+				"foo": "bar1",
+			},
+		},
+		Spec: corev1.PodSpec{
+			NodeName: "test-node",
+		},
+	}
+	labelIdentity1 := &v1alpha1.LabelIdentity{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "labelIdentity1",
+		},
+		Spec: v1alpha1.LabelIdentitySpec{
+			Label: "ns:env=test,kubernetes.io/metadata.name=ns&pod:foo=bar1",
+			ID:    1,
+		},
+	}
+	labelIdentity2 := &v1alpha1.LabelIdentity{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "labelIdentity2",
+		},
+		Spec: v1alpha1.LabelIdentitySpec{
+			Label: "ns:env=test,kubernetes.io/metadata.name=ns&pod:foo=bar2",
+			ID:    2,
+		},
+	}
+
+	clientset := fake.NewSimpleClientset()
+	mcClient := mcfake.NewSimpleClientset()
+	c, closeFn := newStretchedNetworkPolicyController(t, clientset, mcClient)
+	defer closeFn()
+	defer c.queue.ShutDown()
+
+	stopCh := make(chan struct{})
+	defer close(stopCh)
+	c.informerFactory.Start(stopCh)
+	c.informerFactory.WaitForCacheSync(stopCh)
+	c.mcInformerFactory.Start(stopCh)
+	c.mcInformerFactory.WaitForCacheSync(stopCh)
+	go c.podInformer.Run(stopCh)
+	go c.podUpdateChannel.Run(stopCh)
+
+	finishCh := make(chan struct{})
+	go func() {
+		defer close(finishCh)
+		c.clientset.CoreV1().Namespaces().Create(context.TODO(), ns, metav1.CreateOptions{})
+		if err := waitForNSRealized(c, ns); err != nil {
+			t.Errorf("error when waiting for Namespace '%s' to be realized: %v", ns.Name, err)
+		}
+
+		// Create a Pod whose LabelIdentity doesn't exist.
+		c.clientset.CoreV1().Pods(pod.Namespace).Create(context.TODO(), pod, metav1.CreateOptions{})
+		if err := waitForPodRealized(c, pod); err != nil {
+			t.Errorf("error when waiting for Pod '%s/%s' to be realized: %v", pod.Namespace, pod.Name, err)
+		}
+		c.interfaceStore.EXPECT().GetContainerInterfacesByPod(pod.Name, pod.Namespace).Return([]*interfacestore.InterfaceConfig{&interfaceConfig}).Times(1)
+		c.ofClient.EXPECT().InstallPodFlows(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Eq(&unknownLabelIdentity)).Times(1)
+		c.podUpdateChannel.Notify(toPodAddEvent(pod))
+		c.processNextWorkItem()
+		assert.Equal(t, map[types.NamespacedName]string{{Name: pod.Name, Namespace: pod.Namespace}: labelIdentity1.Spec.Label}, c.podToLabel)
+		assert.Equal(t, map[string]podSet{labelIdentity1.Spec.Label: {types.NamespacedName{Name: pod.Name, Namespace: pod.Namespace}: struct{}{}}}, c.labelToPods)
+
+		// Delete a Pod.
+		c.clientset.CoreV1().Pods(pod.Namespace).Delete(context.TODO(), pod.Name, metav1.DeleteOptions{})
+
+		// Create a Pod whose LabelIdentity already exist.
+		c.mcClient.MulticlusterV1alpha1().LabelIdentities().Create(context.TODO(), labelIdentity1, metav1.CreateOptions{})
+		if err := waitForLabelIdentityRealized(c, labelIdentity1); err != nil {
+			t.Errorf("error when waiting for LabelIdentity '%s' to be realized: %v", labelIdentity1.Name, err)
+		}
+		c.interfaceStore.EXPECT().GetContainerInterfacesByPod(pod.Name, pod.Namespace).Return([]*interfacestore.InterfaceConfig{&interfaceConfig}).Times(1)
+		c.clientset.CoreV1().Pods(pod.Namespace).Create(context.TODO(), pod, metav1.CreateOptions{})
+		if err := waitForPodRealized(c, pod); err != nil {
+			t.Errorf("error when waiting for Pod '%s/%s' to be realized: %v", pod.Namespace, pod.Name, err)
+		}
+		c.ofClient.EXPECT().InstallPodFlows(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Eq(&labelIdentity1.Spec.ID)).Times(1)
+		c.podUpdateChannel.Notify(toPodAddEvent(pod))
+		c.processNextWorkItem()
+		assert.Equal(t, map[types.NamespacedName]string{{Name: pod.Name, Namespace: pod.Namespace}: labelIdentity1.Spec.Label}, c.podToLabel)
+		assert.Equal(t, map[string]podSet{labelIdentity1.Spec.Label: {types.NamespacedName{Name: pod.Name, Namespace: pod.Namespace}: struct{}{}}}, c.labelToPods)
+
+		// Update Pod label.
+		c.mcClient.MulticlusterV1alpha1().LabelIdentities().Create(context.TODO(), labelIdentity2, metav1.CreateOptions{})
+		if err := waitForLabelIdentityRealized(c, labelIdentity2); err != nil {
+			t.Errorf("error when waiting for LabelIdentity '%s' to be realized: %v", labelIdentity2.Name, err)
+		}
+		c.interfaceStore.EXPECT().GetContainerInterfacesByPod(pod.Name, pod.Namespace).Return([]*interfacestore.InterfaceConfig{&interfaceConfig}).Times(1)
+		pod.Labels["foo"] = "bar2"
+		c.clientset.CoreV1().Pods(pod.Namespace).Update(context.TODO(), pod, metav1.UpdateOptions{})
+		c.ofClient.EXPECT().InstallPodFlows(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Eq(&labelIdentity2.Spec.ID)).Times(1)
+		c.processNextWorkItem()
+		assert.Equal(t, map[types.NamespacedName]string{{Name: pod.Name, Namespace: pod.Namespace}: labelIdentity2.Spec.Label}, c.podToLabel)
+		assert.Equal(t, map[string]podSet{labelIdentity2.Spec.Label: {types.NamespacedName{Name: pod.Name, Namespace: pod.Namespace}: struct{}{}}}, c.labelToPods)
+	}()
+	select {
+	case <-time.After(5 * time.Second):
+		t.Errorf("Test didn't finish in time")
+	case <-finishCh:
+	}
+}
+
+func TestStretchedNetworkPolicyControllerNSEvent(t *testing.T) {
+	ns := &corev1.Namespace{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "ns",
+			Labels: map[string]string{
+				"env": "test",
+			},
+		},
+	}
+	pod1 := &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "pod1",
+			Namespace: "ns",
+			Labels: map[string]string{
+				"foo": "bar1",
+			},
+		},
+		Spec: corev1.PodSpec{
+			NodeName: "test-node",
+		},
+	}
+	pod2 := &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "pod2",
+			Namespace: "ns",
+			Labels: map[string]string{
+				"foo": "bar2",
+			},
+		},
+		Spec: corev1.PodSpec{
+			NodeName: "test-node",
+		},
+	}
+	labelIdentity1 := &v1alpha1.LabelIdentity{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "labelIdentity1",
+		},
+		Spec: v1alpha1.LabelIdentitySpec{
+			Label: "ns:env=test,kubernetes.io/metadata.name=ns&pod:foo=bar1",
+			ID:    1,
+		},
+	}
+	labelIdentity2 := &v1alpha1.LabelIdentity{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "labelIdentity2",
+		},
+		Spec: v1alpha1.LabelIdentitySpec{
+			Label: "ns:env=test,kubernetes.io/metadata.name=ns&pod:foo=bar2",
+			ID:    2,
+		},
+	}
+	labelIdentity3 := &v1alpha1.LabelIdentity{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "labelIdentity3",
+		},
+		Spec: v1alpha1.LabelIdentitySpec{
+			Label: "ns:env=prod,kubernetes.io/metadata.name=ns&pod:foo=bar1",
+			ID:    3,
+		},
+	}
+	labelIdentity4 := &v1alpha1.LabelIdentity{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "labelIdentity4",
+		},
+		Spec: v1alpha1.LabelIdentitySpec{
+			Label: "ns:env=prod,kubernetes.io/metadata.name=ns&pod:foo=bar2",
+			ID:    4,
+		},
+	}
+
+	clientset := fake.NewSimpleClientset()
+	mcClient := mcfake.NewSimpleClientset()
+	c, closeFn := newStretchedNetworkPolicyController(t, clientset, mcClient)
+	defer closeFn()
+	defer c.queue.ShutDown()
+
+	stopCh := make(chan struct{})
+	defer close(stopCh)
+	c.informerFactory.Start(stopCh)
+	c.informerFactory.WaitForCacheSync(stopCh)
+	c.mcInformerFactory.Start(stopCh)
+	c.mcInformerFactory.WaitForCacheSync(stopCh)
+	go c.podInformer.Run(stopCh)
+	go c.podUpdateChannel.Run(stopCh)
+
+	finishCh := make(chan struct{})
+	go func() {
+		defer close(finishCh)
+
+		c.mcClient.MulticlusterV1alpha1().LabelIdentities().Create(context.TODO(), labelIdentity1, metav1.CreateOptions{})
+		if err := waitForLabelIdentityRealized(c, labelIdentity1); err != nil {
+			t.Errorf("error when waiting for LabelIdentity '%s' to be realized: %v", labelIdentity1.Name, err)
+		}
+		c.mcClient.MulticlusterV1alpha1().LabelIdentities().Create(context.TODO(), labelIdentity2, metav1.CreateOptions{})
+		if err := waitForLabelIdentityRealized(c, labelIdentity2); err != nil {
+			t.Errorf("error when waiting for LabelIdentity '%s' to be realized: %v", labelIdentity2.Name, err)
+		}
+		c.mcClient.MulticlusterV1alpha1().LabelIdentities().Create(context.TODO(), labelIdentity3, metav1.CreateOptions{})
+		if err := waitForLabelIdentityRealized(c, labelIdentity3); err != nil {
+			t.Errorf("error when waiting for LabelIdentity '%s' to be realized: %v", labelIdentity3.Name, err)
+		}
+		c.mcClient.MulticlusterV1alpha1().LabelIdentities().Create(context.TODO(), labelIdentity4, metav1.CreateOptions{})
+		if err := waitForLabelIdentityRealized(c, labelIdentity4); err != nil {
+			t.Errorf("error when waiting for LabelIdentity '%s' to be realized: %v", labelIdentity4.Name, err)
+		}
+
+		c.clientset.CoreV1().Namespaces().Create(context.TODO(), ns, metav1.CreateOptions{})
+		if err := waitForNSRealized(c, ns); err != nil {
+			t.Errorf("error when waiting for Namespace '%s' to be realized: %v", ns.Name, err)
+		}
+
+		c.clientset.CoreV1().Pods(pod1.Namespace).Create(context.TODO(), pod1, metav1.CreateOptions{})
+		if err := waitForPodRealized(c, pod1); err != nil {
+			t.Errorf("error when waiting for Pod '%s/%s' to be realized: %v", pod1.Namespace, pod1.Name, err)
+		}
+		c.clientset.CoreV1().Pods(pod2.Namespace).Create(context.TODO(), pod2, metav1.CreateOptions{})
+		if err := waitForPodRealized(c, pod2); err != nil {
+			t.Errorf("error when waiting for Pod '%s/%s' to be realized: %v", pod2.Namespace, pod2.Name, err)
+		}
+		c.interfaceStore.EXPECT().GetContainerInterfacesByPod(pod1.Name, pod1.Namespace).Return([]*interfacestore.InterfaceConfig{&interfaceConfig}).Times(1)
+		c.interfaceStore.EXPECT().GetContainerInterfacesByPod(pod2.Name, pod2.Namespace).Return([]*interfacestore.InterfaceConfig{&interfaceConfig}).Times(1)
+		c.ofClient.EXPECT().InstallPodFlows(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Eq(&labelIdentity1.Spec.ID)).Times(1)
+		c.ofClient.EXPECT().InstallPodFlows(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Eq(&labelIdentity2.Spec.ID)).Times(1)
+		c.podUpdateChannel.Notify(toPodAddEvent(pod1))
+		c.processNextWorkItem()
+		c.podUpdateChannel.Notify(toPodAddEvent(pod2))
+		c.processNextWorkItem()
+
+		// Update Namespace label.
+		ns.Labels["env"] = "prod"
+		c.clientset.CoreV1().Namespaces().Update(context.TODO(), ns, metav1.UpdateOptions{})
+		c.interfaceStore.EXPECT().GetContainerInterfacesByPod(pod1.Name, pod1.Namespace).Return([]*interfacestore.InterfaceConfig{&interfaceConfig}).Times(1)
+		c.interfaceStore.EXPECT().GetContainerInterfacesByPod(pod2.Name, pod2.Namespace).Return([]*interfacestore.InterfaceConfig{&interfaceConfig}).Times(1)
+		c.ofClient.EXPECT().InstallPodFlows(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Eq(&labelIdentity3.Spec.ID)).Times(1)
+		c.ofClient.EXPECT().InstallPodFlows(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Eq(&labelIdentity4.Spec.ID)).Times(1)
+		c.processNextWorkItem()
+		c.processNextWorkItem()
+		assert.Equal(t, map[types.NamespacedName]string{
+			{Name: pod1.Name, Namespace: pod1.Namespace}: labelIdentity3.Spec.Label,
+			{Name: pod2.Name, Namespace: pod2.Namespace}: labelIdentity4.Spec.Label,
+		}, c.podToLabel)
+		assert.Equal(t, map[string]podSet{
+			labelIdentity3.Spec.Label: {types.NamespacedName{Name: pod1.Name, Namespace: pod1.Namespace}: struct{}{}},
+			labelIdentity4.Spec.Label: {types.NamespacedName{Name: pod2.Name, Namespace: pod2.Namespace}: struct{}{}},
+		}, c.labelToPods)
+	}()
+	select {
+	case <-time.After(5 * time.Second):
+		t.Errorf("Test didn't finish in time")
+	case <-finishCh:
+	}
+}
+
+func TestStretchedNetworkPolicyControllerLabelIdentityEvent(t *testing.T) {
+	ns := &corev1.Namespace{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "ns",
+			Labels: map[string]string{
+				"env": "test",
+			},
+		},
+	}
+	pod := &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "pod",
+			Namespace: "ns",
+			Labels: map[string]string{
+				"foo": "bar1",
+			},
+		},
+		Spec: corev1.PodSpec{
+			NodeName: "test-node",
+		},
+	}
+	labelIdentity := &v1alpha1.LabelIdentity{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "labelIdentity",
+		},
+		Spec: v1alpha1.LabelIdentitySpec{
+			Label: "ns:env=test,kubernetes.io/metadata.name=ns&pod:foo=bar1",
+			ID:    1,
+		},
+	}
+	clientset := fake.NewSimpleClientset()
+	mcClient := mcfake.NewSimpleClientset()
+	c, closeFn := newStretchedNetworkPolicyController(t, clientset, mcClient)
+	defer closeFn()
+	defer c.queue.ShutDown()
+
+	stopCh := make(chan struct{})
+	defer close(stopCh)
+	c.informerFactory.Start(stopCh)
+	c.informerFactory.WaitForCacheSync(stopCh)
+	c.mcInformerFactory.Start(stopCh)
+	c.mcInformerFactory.WaitForCacheSync(stopCh)
+	go c.podInformer.Run(stopCh)
+	go c.podUpdateChannel.Run(stopCh)
+
+	finishCh := make(chan struct{})
+	go func() {
+		defer close(finishCh)
+		c.clientset.CoreV1().Namespaces().Create(context.TODO(), ns, metav1.CreateOptions{})
+		if err := waitForNSRealized(c, ns); err != nil {
+			t.Errorf("error when waiting for Namespace '%s' to be realized: %v", ns.Name, err)
+		}
+
+		c.clientset.CoreV1().Pods(pod.Namespace).Create(context.TODO(), pod, metav1.CreateOptions{})
+		if err := waitForPodRealized(c, pod); err != nil {
+			t.Errorf("error when waiting for Pod '%s/%s' to be realized: %v", pod.Namespace, pod.Name, err)
+		}
+		c.interfaceStore.EXPECT().GetContainerInterfacesByPod(pod.Name, pod.Namespace).Return([]*interfacestore.InterfaceConfig{&interfaceConfig}).Times(1)
+		c.ofClient.EXPECT().InstallPodFlows(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Eq(&unknownLabelIdentity)).Times(1)
+		c.podUpdateChannel.Notify(toPodAddEvent(pod))
+		c.processNextWorkItem()
+		assert.Equal(t, map[types.NamespacedName]string{{Name: pod.Name, Namespace: pod.Namespace}: labelIdentity.Spec.Label}, c.podToLabel)
+		assert.Equal(t, map[string]podSet{labelIdentity.Spec.Label: {types.NamespacedName{Name: pod.Name, Namespace: pod.Namespace}: struct{}{}}}, c.labelToPods)
+
+		// Create LabelIdentity
+		c.mcClient.MulticlusterV1alpha1().LabelIdentities().Create(context.TODO(), labelIdentity, metav1.CreateOptions{})
+		if err := waitForLabelIdentityRealized(c, labelIdentity); err != nil {
+			t.Errorf("error when waiting for LabelIdentity '%s' to be realized: %v", labelIdentity.Name, err)
+		}
+		c.interfaceStore.EXPECT().GetContainerInterfacesByPod(pod.Name, pod.Namespace).Return([]*interfacestore.InterfaceConfig{&interfaceConfig}).Times(1)
+		c.ofClient.EXPECT().InstallPodFlows(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Eq(&labelIdentity.Spec.ID)).Times(1)
+		c.processNextWorkItem()
+		assert.Equal(t, map[types.NamespacedName]string{{Name: pod.Name, Namespace: pod.Namespace}: labelIdentity.Spec.Label}, c.podToLabel)
+		assert.Equal(t, map[string]podSet{labelIdentity.Spec.Label: {types.NamespacedName{Name: pod.Name, Namespace: pod.Namespace}: struct{}{}}}, c.labelToPods)
+
+		// Update LabelIdentity
+		labelIdentity.Spec.ID = 2
+		c.mcClient.MulticlusterV1alpha1().LabelIdentities().Update(context.TODO(), labelIdentity, metav1.UpdateOptions{})
+		c.interfaceStore.EXPECT().GetContainerInterfacesByPod(pod.Name, pod.Namespace).Return([]*interfacestore.InterfaceConfig{&interfaceConfig}).Times(1)
+		c.ofClient.EXPECT().InstallPodFlows(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Eq(&labelIdentity.Spec.ID)).Times(1)
+		c.processNextWorkItem()
+		assert.Equal(t, map[types.NamespacedName]string{{Name: pod.Name, Namespace: pod.Namespace}: labelIdentity.Spec.Label}, c.podToLabel)
+		assert.Equal(t, map[string]podSet{labelIdentity.Spec.Label: {types.NamespacedName{Name: pod.Name, Namespace: pod.Namespace}: struct{}{}}}, c.labelToPods)
+
+		// Delete LabelIdentity
+		c.mcClient.MulticlusterV1alpha1().LabelIdentities().Delete(context.TODO(), labelIdentity.Name, metav1.DeleteOptions{})
+		c.interfaceStore.EXPECT().GetContainerInterfacesByPod(pod.Name, pod.Namespace).Return([]*interfacestore.InterfaceConfig{&interfaceConfig}).Times(1)
+		c.ofClient.EXPECT().InstallPodFlows(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Eq(&unknownLabelIdentity)).Times(1)
+		c.processNextWorkItem()
+		assert.Equal(t, map[types.NamespacedName]string{{Name: pod.Name, Namespace: pod.Namespace}: labelIdentity.Spec.Label}, c.podToLabel)
+		assert.Equal(t, map[string]podSet{labelIdentity.Spec.Label: {types.NamespacedName{Name: pod.Name, Namespace: pod.Namespace}: struct{}{}}}, c.labelToPods)
+	}()
+	select {
+	case <-time.After(5 * time.Second):
+		t.Errorf("Test didn't finish in time")
+	case <-finishCh:
+	}
+}
+
+func toPodAddEvent(pod *corev1.Pod) antreatypes.PodUpdate {
+	return antreatypes.PodUpdate{
+		PodNamespace: pod.Namespace,
+		PodName:      pod.Name,
+		IsAdd:        true,
+	}
+}
+
+func waitForPodRealized(c *fakeStretchedNetworkPolicyController, pod *corev1.Pod) error {
+	return wait.Poll(interval, timeout, func() (bool, error) {
+		_, err := c.podLister.Pods(pod.Namespace).Get(pod.Name)
+		if err != nil {
+			return false, nil
+		}
+		return true, err
+	})
+}
+
+func waitForNSRealized(c *fakeStretchedNetworkPolicyController, ns *corev1.Namespace) error {
+	return wait.Poll(interval, timeout, func() (bool, error) {
+		_, err := c.namespaceLister.Get(ns.Name)
+		if err != nil {
+			return false, nil
+		}
+		return true, err
+	})
+}
+
+func waitForLabelIdentityRealized(c *fakeStretchedNetworkPolicyController, labelIdentity *v1alpha1.LabelIdentity) error {
+	return wait.Poll(interval, timeout, func() (bool, error) {
+		_, err := c.labelIdentityLister.Get(labelIdentity.Name)
+		if err != nil {
+			return false, nil
+		}
+		return true, err
+	})
+}
diff --git a/pkg/agent/openflow/client_test.go b/pkg/agent/openflow/client_test.go
index 899b74039..fc75cdf8b 100644
--- a/pkg/agent/openflow/client_test.go
+++ b/pkg/agent/openflow/client_test.go
@@ -142,7 +142,21 @@ func installPodFlows(ofClient Client, cacheKey string) (int, error) {
 	podMAC, _ := net.ParseMAC("AA:BB:CC:DD:EE:EE")
 	podIP := net.ParseIP("10.0.0.2")
 	ofPort := uint32(10)
-	err := ofClient.InstallPodFlows(containerID, []net.IP{podIP}, podMAC, ofPort, 0)
+	err := ofClient.InstallPodFlows(containerID, []net.IP{podIP}, podMAC, ofPort, 0, nil)
+	client := ofClient.(*client)
+	fCacheI, ok := client.featurePodConnectivity.podCachedFlows.Load(containerID)
+	if ok {
+		return len(fCacheI.(flowCache)), err
+	}
+	return 0, err
+}
+
+func installPodFlowsWithLabelID(ofClient Client, cacheKey string, labelID *uint32) (int, error) {
+	containerID := cacheKey
+	podMAC, _ := net.ParseMAC("AA:BB:CC:DD:EE:EE")
+	podIP := net.ParseIP("10.0.0.2")
+	ofPort := uint32(10)
+	err := ofClient.InstallPodFlows(containerID, []net.IP{podIP}, podMAC, ofPort, 0, labelID)
 	client := ofClient.(*client)
 	fCacheI, ok := client.featurePodConnectivity.podCachedFlows.Load(containerID)
 	if ok {
@@ -153,13 +167,16 @@ func installPodFlows(ofClient Client, cacheKey string) (int, error) {
 
 // TestIdempotentFlowInstallation checks that InstallNodeFlows and InstallPodFlows are idempotent.
 func TestIdempotentFlowInstallation(t *testing.T) {
+	labelID := uint32(1)
 	testCases := []struct {
 		name      string
 		cacheKey  string
+		labelID   *uint32
 		numFlows  int
-		installFn func(ofClient Client, cacheKey string) (int, error)
+		installFn func(ofClient Client, cacheKey string, labelID *uint32) (int, error)
 	}{
-		{"PodFlows", "aaaa-bbbb-cccc-dddd", 5, installPodFlows},
+		{"PodFlows", "aaaa-bbbb-cccc-dddd", nil, 5, installPodFlowsWithLabelID},
+		{"SNPPodFlows", "eeee-ffff-gggg-hhhh", &labelID, 5, installPodFlowsWithLabelID},
 	}
 
 	// Check the flows are installed only once even though InstallNodeFlows/InstallPodFlows is called multiple times.
@@ -173,13 +190,14 @@ func TestIdempotentFlowInstallation(t *testing.T) {
 
 			m.EXPECT().AddAll(gomock.Any()).Return(nil).Times(1)
 			// Installing the flows should succeed, and all the flows should be added into the cache.
-			numCached1, err := tc.installFn(fc, tc.cacheKey)
+			numCached1, err := tc.installFn(fc, tc.cacheKey, tc.labelID)
 			require.Nil(t, err, "Error when installing Node flows")
 			assert.Equal(t, tc.numFlows, numCached1)
 
 			// Installing the same flows again must not return an error and should not
 			// add additional flows to the cache.
-			numCached2, err := tc.installFn(fc, tc.cacheKey)
+			m.EXPECT().BundleOps(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).Times(1)
+			numCached2, err := tc.installFn(fc, tc.cacheKey, tc.labelID)
 			require.Nil(t, err, "Error when installing Node flows again")
 
 			assert.Equal(t, numCached1, numCached2)
@@ -195,16 +213,16 @@ func TestIdempotentFlowInstallation(t *testing.T) {
 			fc := newFakeClient(m, true, false, config.K8sNode, config.TrafficEncapModeEncap)
 			defer resetPipelines()
 
-			errorCall := m.EXPECT().AddAll(gomock.Any()).Return(errors.New("Bundle error")).Times(1)
+			errorCall := m.EXPECT().AddAll(gomock.Any()).Return(errors.New("Bundle error"))
 			m.EXPECT().AddAll(gomock.Any()).Return(nil).After(errorCall)
 
 			// Installing the flows failed at the first time, and no flow cache is created.
-			numCached1, err := tc.installFn(fc, tc.cacheKey)
+			numCached1, err := tc.installFn(fc, tc.cacheKey, tc.labelID)
 			require.NotNil(t, err, "Installing flows in bundle is expected to fail")
 			assert.Equal(t, 0, numCached1)
 
 			// Installing the same flows successfully at the second time, and add flows to the cache.
-			numCached2, err := tc.installFn(fc, tc.cacheKey)
+			numCached2, err := tc.installFn(fc, tc.cacheKey, tc.labelID)
 			require.Nil(t, err, "Error when installing Node flows again")
 
 			assert.Equal(t, tc.numFlows, numCached2)
@@ -220,8 +238,8 @@ func TestFlowInstallationFailed(t *testing.T) {
 		numAddCalls int
 		installFn   func(ofClient Client, cacheKey string) (int, error)
 	}{
-		{"NodeFlows", "host", 2, installNodeFlows},
-		{"PodFlows", "aaaa-bbbb-cccc-dddd", 5, installPodFlows},
+		{"NodeFlows", "host", 1, installNodeFlows},
+		{"PodFlows", "aaaa-bbbb-cccc-dddd", 1, installPodFlows},
 	}
 
 	for _, tc := range testCases {
@@ -233,7 +251,7 @@ func TestFlowInstallationFailed(t *testing.T) {
 			defer resetPipelines()
 
 			// We generate an error for AddAll call.
-			m.EXPECT().AddAll(gomock.Any()).Return(errors.New("Bundle error"))
+			m.EXPECT().AddAll(gomock.Any()).Return(errors.New("Bundle error")).Times(tc.numAddCalls)
 
 			var err error
 			var numCached int
@@ -753,7 +771,7 @@ func Test_client_InstallPodFlows(t *testing.T) {
 			interfaceName := "pod1"
 			cacheKey := fmt.Sprintf("multicast_pod_metric_%s", interfaceName)
 
-			assert.NoError(t, fc.InstallPodFlows(interfaceName, tc.podInterfaceIPs, podMAC, podOfPort, tc.vlanID))
+			assert.NoError(t, fc.InstallPodFlows(interfaceName, tc.podInterfaceIPs, podMAC, podOfPort, tc.vlanID, nil))
 			fCacheI, ok := fc.featurePodConnectivity.podCachedFlows.Load(interfaceName)
 			require.True(t, ok)
 			flows := getFlowStrings(fCacheI)
@@ -789,7 +807,7 @@ func Test_client_GetPodFlowKeys(t *testing.T) {
 	podInterfaceIPs := []net.IP{net.ParseIP("10.10.0.11")}
 	podMAC, _ := net.ParseMAC("00:00:10:10:00:11")
 
-	assert.NoError(t, fc.InstallPodFlows(interfaceName, podInterfaceIPs, podMAC, uint32(11), 0))
+	assert.NoError(t, fc.InstallPodFlows(interfaceName, podInterfaceIPs, podMAC, uint32(11), 0, nil))
 	flowKeys := fc.GetPodFlowKeys(interfaceName)
 	expectedFlowKeys := []string{
 		"table=1,arp,in_port=11,arp_sha=00:00:10:10:00:11,arp_spa=10.10.0.11",
@@ -1218,7 +1236,7 @@ func Test_client_InstallPodSNATFlows(t *testing.T) {
 		{
 			name: "SNAT on Remote",
 			expectedFlows: []string{
-				"cookie=0x1040000000000, table=EgressMark, priority=200,ip,in_port=100 actions=set_field:0a:00:00:00:00:01->eth_src,set_field:aa:bb:cc:dd:ee:ff->eth_dst,set_field:192.168.77.101->tun_dst,set_field:0x10/0xf0->reg0,goto_table:L2ForwardingCalc",
+				"cookie=0x1040000000000, table=EgressMark, priority=200,ip,in_port=100 actions=set_field:0a:00:00:00:00:01->eth_src,set_field:aa:bb:cc:dd:ee:ff->eth_dst,set_field:192.168.77.101->tun_dst,set_field:0x10/0xf0->reg0,set_field:0x40000/0x40000->reg0,goto_table:L2ForwardingCalc",
 			},
 		},
 	}
@@ -2009,6 +2027,7 @@ func Test_client_InstallMulticlusterNodeFlows(t *testing.T) {
 			expectedFlows: []string{
 				"cookie=0x1060000000000, table=L3Forwarding, priority=200,ip,nw_dst=10.97.0.0/16 actions=set_field:0a:00:00:00:00:01->eth_src,set_field:aa:bb:cc:dd:ee:f0->eth_dst,set_field:192.168.78.101->tun_dst,set_field:0x10/0xf0->reg0,goto_table:L3DecTTL",
 				"cookie=0x1060000000000, table=L3Forwarding, priority=200,ct_state=+rpl+trk,ip,nw_dst=192.168.78.101 actions=set_field:0a:00:00:00:00:01->eth_src,set_field:aa:bb:cc:dd:ee:f0->eth_dst,set_field:192.168.78.101->tun_dst,set_field:0x10/0xf0->reg0,goto_table:L3DecTTL",
+				"cookie=0x1060000000000, table=L3Forwarding, priority=199,ip,reg0=0x4000/0x3e000,nw_dst=192.168.78.101 actions=set_field:0a:00:00:00:00:01->eth_src,set_field:aa:bb:cc:dd:ee:f0->eth_dst,set_field:192.168.78.101->tun_dst,set_field:0x10/0xf0->reg0,goto_table:L3DecTTL",
 			},
 		},
 		//TODO: IPv6
@@ -2025,7 +2044,7 @@ func Test_client_InstallMulticlusterNodeFlows(t *testing.T) {
 			m.EXPECT().AddAll(gomock.Any()).Return(nil).Times(1)
 			m.EXPECT().DeleteAll(gomock.Any()).Return(nil).Times(1)
 
-			assert.NoError(t, fc.InstallMulticlusterNodeFlows(clusterID, tc.peerConfigs, tc.tunnelPeerIP))
+			assert.NoError(t, fc.InstallMulticlusterNodeFlows(clusterID, tc.peerConfigs, tc.tunnelPeerIP, true))
 			cacheKey := fmt.Sprintf("cluster_%s", clusterID)
 			fCacheI, ok := fc.featureMulticluster.cachedFlows.Load(cacheKey)
 			require.True(t, ok)
@@ -2060,6 +2079,7 @@ func Test_client_InstallMulticlusterGatewayFlows(t *testing.T) {
 				"cookie=0x1060000000000, table=UnSNAT, priority=200,ip,nw_dst=192.168.77.100 actions=ct(table=ConntrackZone,zone=65521,exec(nat))",
 				"cookie=0x1060000000000, table=L3Forwarding, priority=200,ip,nw_dst=10.97.0.0/16 actions=set_field:0a:00:00:00:00:01->eth_src,set_field:aa:bb:cc:dd:ee:f0->eth_dst,set_field:192.168.78.101->tun_dst,set_field:0x10/0xf0->reg0,goto_table:L3DecTTL",
 				"cookie=0x1060000000000, table=L3Forwarding, priority=200,ct_state=+rpl+trk,ip,nw_dst=192.168.78.101 actions=set_field:0a:00:00:00:00:01->eth_src,set_field:aa:bb:cc:dd:ee:f0->eth_dst,set_field:192.168.78.101->tun_dst,set_field:0x10/0xf0->reg0,goto_table:L3DecTTL",
+				"cookie=0x1060000000000, table=L3Forwarding, priority=199,ip,reg0=0x4000/0x3e000,nw_dst=192.168.78.101 actions=set_field:0a:00:00:00:00:01->eth_src,set_field:aa:bb:cc:dd:ee:f0->eth_dst,set_field:192.168.78.101->tun_dst,set_field:0x10/0xf0->reg0,goto_table:L3DecTTL",
 				"cookie=0x1060000000000, table=SNATMark, priority=210,ct_state=+new+trk,ip,nw_dst=10.97.0.0/16 actions=ct(commit,table=SNAT,zone=65520,exec(set_field:0x20/0x20->ct_mark))",
 				"cookie=0x1060000000000, table=SNAT, priority=200,ct_state=+new+trk,ip,nw_dst=10.97.0.0/16 actions=ct(commit,table=L2ForwardingCalc,zone=65521,exec(nat(src=192.168.77.100)))",
 			},
@@ -2080,7 +2100,7 @@ func Test_client_InstallMulticlusterGatewayFlows(t *testing.T) {
 
 			cacheKey := fmt.Sprintf("cluster_%s", clusterID)
 
-			assert.NoError(t, fc.InstallMulticlusterGatewayFlows(clusterID, tc.peerConfigs, tc.tunnelPeerIP, tc.localGatewayIP))
+			assert.NoError(t, fc.InstallMulticlusterGatewayFlows(clusterID, tc.peerConfigs, tc.tunnelPeerIP, tc.localGatewayIP, true))
 			fCacheI, ok := fc.featureMulticluster.cachedFlows.Load(cacheKey)
 			require.True(t, ok)
 			assert.ElementsMatch(t, tc.expectedFlows, getFlowStrings(fCacheI))
diff --git a/pkg/agent/openflow/network_policy_test.go b/pkg/agent/openflow/network_policy_test.go
index d222e4522..fbd1f66c2 100644
--- a/pkg/agent/openflow/network_policy_test.go
+++ b/pkg/agent/openflow/network_policy_test.go
@@ -69,6 +69,7 @@ var (
 	protocolICMP = v1beta2.ProtocolICMP
 	priority100  = uint16(100)
 	priority200  = uint16(200)
+	priority201  = uint16(201)
 	icmpType8    = int32(8)
 	icmpCode0    = int32(0)
 
@@ -453,6 +454,22 @@ func TestBatchInstallPolicyRuleFlows(t *testing.T) {
 						UID:       "id3",
 					},
 				},
+				{
+					Direction: v1beta2.DirectionIn,
+					From:      parseLabelIdentityAddresses([]uint32{1, 2}),
+					Action:    &actionDrop,
+					Priority:  &priority201,
+					To:        []types.Address{NewOFPortAddress(1)},
+					Service:   []v1beta2.Service{},
+					FlowID:    uint32(13),
+					TableID:   AntreaPolicyIngressRuleTable.GetID(),
+					PolicyRef: &v1beta2.NetworkPolicyReference{
+						Type:      v1beta2.AntreaNetworkPolicy,
+						Namespace: "ns1",
+						Name:      "np4",
+						UID:       "id4",
+					},
+				},
 			},
 			expectedFlowsFn: func(c *client) []binding.Flow {
 				cookiePolicy := c.cookieAllocator.Request(cookie.NetworkPolicy).Raw()
@@ -471,6 +488,11 @@ func TestBatchInstallPolicyRuleFlows(t *testing.T) {
 						Action().LoadToRegField(CNPConjIDField, 12).
 						Action().LoadRegMark(CnpDenyRegMark).
 						Action().GotoTable(IngressMetricTable.GetID()).Done(),
+					AntreaPolicyIngressRuleTable.ofTable.BuildFlow(priority201).Cookie(cookiePolicy).
+						MatchConjID(13).
+						Action().LoadToRegField(CNPConjIDField, 13).
+						Action().LoadRegMark(CnpDenyRegMark).
+						Action().GotoTable(IngressMetricTable.GetID()).Done(),
 					AntreaPolicyIngressRuleTable.ofTable.BuildFlow(priority100).Cookie(cookiePolicy).
 						MatchProtocol(binding.ProtocolIP).MatchSrcIP(net.ParseIP("192.168.1.40")).
 						Action().Conjunction(10, 1, 2).
@@ -484,6 +506,12 @@ func TestBatchInstallPolicyRuleFlows(t *testing.T) {
 					AntreaPolicyIngressRuleTable.ofTable.BuildFlow(priority100).Cookie(cookiePolicy).
 						MatchProtocol(binding.ProtocolIP).MatchSrcIP(net.ParseIP("192.168.1.51")).
 						Action().Conjunction(11, 1, 3).Done(),
+					AntreaPolicyIngressRuleTable.ofTable.BuildFlow(priority201).Cookie(cookiePolicy).
+						MatchTunnelID(1).
+						Action().Conjunction(13, 1, 3).Done(),
+					AntreaPolicyIngressRuleTable.ofTable.BuildFlow(priority201).Cookie(cookiePolicy).
+						MatchTunnelID(2).
+						Action().Conjunction(13, 1, 3).Done(),
 					AntreaPolicyIngressRuleTable.ofTable.BuildFlow(priority100).Cookie(cookiePolicy).
 						MatchRegFieldWithValue(TargetOFPortField, uint32(1)).
 						Action().Conjunction(10, 2, 2).
@@ -497,6 +525,9 @@ func TestBatchInstallPolicyRuleFlows(t *testing.T) {
 					AntreaPolicyIngressRuleTable.ofTable.BuildFlow(priority100).Cookie(cookiePolicy).
 						MatchRegFieldWithValue(TargetOFPortField, uint32(3)).
 						Action().Conjunction(11, 2, 3).Done(),
+					AntreaPolicyIngressRuleTable.ofTable.BuildFlow(priority201).Cookie(cookiePolicy).
+						MatchRegFieldWithValue(TargetOFPortField, uint32(1)).
+						Action().Conjunction(13, 2, 3).Done(),
 					AntreaPolicyIngressRuleTable.ofTable.BuildFlow(priority100).Cookie(cookiePolicy).
 						MatchProtocol(binding.ProtocolTCP).MatchDstPort(8080, nil).
 						Action().Conjunction(11, 3, 3).Done(),
@@ -518,6 +549,9 @@ func TestBatchInstallPolicyRuleFlows(t *testing.T) {
 					IngressMetricTable.ofTable.BuildFlow(priorityNormal).Cookie(cookiePolicy).
 						MatchRegMark(CnpDenyRegMark).MatchRegFieldWithValue(CNPConjIDField, 12).
 						Action().Drop().Done(),
+					IngressMetricTable.ofTable.BuildFlow(priorityNormal).Cookie(cookiePolicy).
+						MatchRegMark(CnpDenyRegMark).MatchRegFieldWithValue(CNPConjIDField, 13).
+						Action().Drop().Done(),
 				}
 			},
 		},
@@ -977,6 +1011,14 @@ func parseAddresses(addrs []string) []types.Address {
 	return addresses
 }
 
+func parseLabelIdentityAddresses(labelIdentities []uint32) []types.Address {
+	var addresses = make([]types.Address, 0)
+	for _, labelIdentity := range labelIdentities {
+		addresses = append(addresses, NewLabelIDAddress(labelIdentity))
+	}
+	return addresses
+}
+
 func preparePipelines() {
 	pipelineID := pipelineIP
 	requiredTablesMap := make(map[*Table]struct{})
diff --git a/pkg/agent/openflow/testing/mock_openflow.go b/pkg/agent/openflow/testing/mock_openflow.go
index 5ea6d08d8..596c9c6da 100644
--- a/pkg/agent/openflow/testing/mock_openflow.go
+++ b/pkg/agent/openflow/testing/mock_openflow.go
@@ -340,31 +340,31 @@ func (mr *MockClientMockRecorder) InstallMulticlusterClassifierFlows(arg0, arg1
 }
 
 // InstallMulticlusterGatewayFlows mocks base method
-func (m *MockClient) InstallMulticlusterGatewayFlows(arg0 string, arg1 map[*net.IPNet]net.IP, arg2, arg3 net.IP) error {
+func (m *MockClient) InstallMulticlusterGatewayFlows(arg0 string, arg1 map[*net.IPNet]net.IP, arg2, arg3 net.IP, arg4 bool) error {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "InstallMulticlusterGatewayFlows", arg0, arg1, arg2, arg3)
+	ret := m.ctrl.Call(m, "InstallMulticlusterGatewayFlows", arg0, arg1, arg2, arg3, arg4)
 	ret0, _ := ret[0].(error)
 	return ret0
 }
 
 // InstallMulticlusterGatewayFlows indicates an expected call of InstallMulticlusterGatewayFlows
-func (mr *MockClientMockRecorder) InstallMulticlusterGatewayFlows(arg0, arg1, arg2, arg3 interface{}) *gomock.Call {
+func (mr *MockClientMockRecorder) InstallMulticlusterGatewayFlows(arg0, arg1, arg2, arg3, arg4 interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "InstallMulticlusterGatewayFlows", reflect.TypeOf((*MockClient)(nil).InstallMulticlusterGatewayFlows), arg0, arg1, arg2, arg3)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "InstallMulticlusterGatewayFlows", reflect.TypeOf((*MockClient)(nil).InstallMulticlusterGatewayFlows), arg0, arg1, arg2, arg3, arg4)
 }
 
 // InstallMulticlusterNodeFlows mocks base method
-func (m *MockClient) InstallMulticlusterNodeFlows(arg0 string, arg1 map[*net.IPNet]net.IP, arg2 net.IP) error {
+func (m *MockClient) InstallMulticlusterNodeFlows(arg0 string, arg1 map[*net.IPNet]net.IP, arg2 net.IP, arg3 bool) error {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "InstallMulticlusterNodeFlows", arg0, arg1, arg2)
+	ret := m.ctrl.Call(m, "InstallMulticlusterNodeFlows", arg0, arg1, arg2, arg3)
 	ret0, _ := ret[0].(error)
 	return ret0
 }
 
 // InstallMulticlusterNodeFlows indicates an expected call of InstallMulticlusterNodeFlows
-func (mr *MockClientMockRecorder) InstallMulticlusterNodeFlows(arg0, arg1, arg2 interface{}) *gomock.Call {
+func (mr *MockClientMockRecorder) InstallMulticlusterNodeFlows(arg0, arg1, arg2, arg3 interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "InstallMulticlusterNodeFlows", reflect.TypeOf((*MockClient)(nil).InstallMulticlusterNodeFlows), arg0, arg1, arg2)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "InstallMulticlusterNodeFlows", reflect.TypeOf((*MockClient)(nil).InstallMulticlusterNodeFlows), arg0, arg1, arg2, arg3)
 }
 
 // InstallNodeFlows mocks base method
@@ -382,17 +382,17 @@ func (mr *MockClientMockRecorder) InstallNodeFlows(arg0, arg1, arg2, arg3, arg4
 }
 
 // InstallPodFlows mocks base method
-func (m *MockClient) InstallPodFlows(arg0 string, arg1 []net.IP, arg2 net.HardwareAddr, arg3 uint32, arg4 uint16) error {
+func (m *MockClient) InstallPodFlows(arg0 string, arg1 []net.IP, arg2 net.HardwareAddr, arg3 uint32, arg4 uint16, arg5 *uint32) error {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "InstallPodFlows", arg0, arg1, arg2, arg3, arg4)
+	ret := m.ctrl.Call(m, "InstallPodFlows", arg0, arg1, arg2, arg3, arg4, arg5)
 	ret0, _ := ret[0].(error)
 	return ret0
 }
 
 // InstallPodFlows indicates an expected call of InstallPodFlows
-func (mr *MockClientMockRecorder) InstallPodFlows(arg0, arg1, arg2, arg3, arg4 interface{}) *gomock.Call {
+func (mr *MockClientMockRecorder) InstallPodFlows(arg0, arg1, arg2, arg3, arg4, arg5 interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "InstallPodFlows", reflect.TypeOf((*MockClient)(nil).InstallPodFlows), arg0, arg1, arg2, arg3, arg4)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "InstallPodFlows", reflect.TypeOf((*MockClient)(nil).InstallPodFlows), arg0, arg1, arg2, arg3, arg4, arg5)
 }
 
 // InstallPodSNATFlows mocks base method
diff --git a/pkg/graphviz/traceflow_test.go b/pkg/graphviz/traceflow_test.go
index 98558968e..349f69cad 100644
--- a/pkg/graphviz/traceflow_test.go
+++ b/pkg/graphviz/traceflow_test.go
@@ -181,6 +181,52 @@ func TestGenGraph(t *testing.T) {
 		},
 	}
 
+	tfEgressFromLocalNode := crdv1alpha1.Traceflow{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "test-traceflow-egress-local-node",
+		},
+		Spec: crdv1alpha1.TraceflowSpec{
+			Packet: crdv1alpha1.Packet{
+				IPHeader: crdv1alpha1.IPHeader{
+					Protocol: 1,
+				},
+				TransportHeader: crdv1alpha1.TransportHeader{},
+			},
+			Source: crdv1alpha1.Source{
+				Namespace: "default",
+				Pod:       "pod-1",
+			},
+			Destination: crdv1alpha1.Destination{
+				IP: "192.168.100.100",
+			},
+		},
+		Status: crdv1alpha1.TraceflowStatus{
+			Phase: crdv1alpha1.Succeeded,
+			Results: []crdv1alpha1.NodeResult{
+				{
+					Node: "k8s-node-1",
+					Observations: []crdv1alpha1.Observation{
+						{
+							Action:    crdv1alpha1.ActionForwarded,
+							Component: crdv1alpha1.ComponentSpoofGuard,
+						},
+						{
+							Component: crdv1alpha1.ComponentEgress,
+							Action:    crdv1alpha1.ActionMarkedForSNAT,
+							Egress:    "egressA",
+							EgressIP:  "192.168.225.5",
+						},
+						{
+							Action:        crdv1alpha1.ActionForwardedOutOfOverlay,
+							Component:     crdv1alpha1.ComponentForwarding,
+							ComponentInfo: "Output",
+						},
+					},
+				},
+			},
+		},
+	}
+
 	expectedOutputWithSourcePod := `digraph G {
 	center=true;
 	label="test-live-traceflow";
@@ -295,6 +341,36 @@ Delivered", shape=box, style="rounded,filled,solid" ];
 
 }
 `
+
+	expectedOutputEgressFromLocalNode := `digraph G {
+	center=true;
+	label="test-traceflow-egress-local-node";
+	labelloc=t;
+	"default/pod-1"->cluster_source_1[ color="#C0C0C0", dir=forward, minlen=1, penwidth=2.0 ];
+	cluster_source_1->cluster_source_2[ color="#C0C0C0", dir=forward, minlen=1, penwidth=2.0 ];
+	cluster_source_2->cluster_source_3[ color="#C0C0C0", dir=forward, minlen=1, penwidth=2.0 ];
+	subgraph cluster_source {
+	bgcolor="#F8F8FF";
+	label="k8s-node-1";
+	labeljust=l;
+	style="filled,bold";
+	"default/pod-1" [ color="#808080", fillcolor="#C8C8C8", style="filled,bold" ];
+	cluster_source_1 [ color="#696969", fillcolor="#DCDCDC", label="SpoofGuard
+Forwarded", shape=box, style="rounded,filled,solid" ];
+	cluster_source_2 [ color="#696969", fillcolor="#DCDCDC", label="Egress
+MarkedForSNAT
+Egress IP : 192.168.225.5
+Egress : egressA", shape=box, style="rounded,filled,solid" ];
+	cluster_source_3 [ color="#696969", fillcolor="#DCDCDC", label="Forwarding
+Output
+ForwardedOutOfOverlay", shape=box, style="rounded,filled,solid" ];
+
+}
+;
+
+}
+`
+
 	tests := []struct {
 		name           string
 		traceflow      crdv1alpha1.Traceflow
@@ -315,6 +391,11 @@ Delivered", shape=box, style="rounded,filled,solid" ];
 			traceflow:      tfInOneNode,
 			expectedOutput: expectedOutputInOneNode,
 		},
+		{
+			name:           "traceflow egress from local Node",
+			traceflow:      tfEgressFromLocalNode,
+			expectedOutput: expectedOutputEgressFromLocalNode,
+		},
 	}
 
 	for _, tt := range tests {
diff --git a/pkg/ovs/openflow/testing/mock_openflow.go b/pkg/ovs/openflow/testing/mock_openflow.go
index 4062f5977..ce28d5953 100644
--- a/pkg/ovs/openflow/testing/mock_openflow.go
+++ b/pkg/ovs/openflow/testing/mock_openflow.go
@@ -1216,6 +1216,20 @@ func (mr *MockActionMockRecorder) SetTunnelDst(arg0 interface{}) *gomock.Call {
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SetTunnelDst", reflect.TypeOf((*MockAction)(nil).SetTunnelDst), arg0)
 }
 
+// SetTunnelID mocks base method
+func (m *MockAction) SetTunnelID(arg0 uint64) openflow.FlowBuilder {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "SetTunnelID", arg0)
+	ret0, _ := ret[0].(openflow.FlowBuilder)
+	return ret0
+}
+
+// SetTunnelID indicates an expected call of SetTunnelID
+func (mr *MockActionMockRecorder) SetTunnelID(arg0 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SetTunnelID", reflect.TypeOf((*MockAction)(nil).SetTunnelID), arg0)
+}
+
 // SetVLAN mocks base method
 func (m *MockAction) SetVLAN(arg0 uint16) openflow.FlowBuilder {
 	m.ctrl.T.Helper()
@@ -2072,6 +2086,20 @@ func (mr *MockFlowBuilderMockRecorder) MatchTunnelDst(arg0 interface{}) *gomock.
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "MatchTunnelDst", reflect.TypeOf((*MockFlowBuilder)(nil).MatchTunnelDst), arg0)
 }
 
+// MatchTunnelID mocks base method
+func (m *MockFlowBuilder) MatchTunnelID(arg0 uint64) openflow.FlowBuilder {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "MatchTunnelID", arg0)
+	ret0, _ := ret[0].(openflow.FlowBuilder)
+	return ret0
+}
+
+// MatchTunnelID indicates an expected call of MatchTunnelID
+func (mr *MockFlowBuilderMockRecorder) MatchTunnelID(arg0 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "MatchTunnelID", reflect.TypeOf((*MockFlowBuilder)(nil).MatchTunnelID), arg0)
+}
+
 // MatchVLAN mocks base method
 func (m *MockFlowBuilder) MatchVLAN(arg0 bool, arg1 uint16, arg2 *uint16) openflow.FlowBuilder {
 	m.ctrl.T.Helper()
diff --git a/test/e2e/framework.go b/test/e2e/framework.go
index 5a8308879..0c9c9010f 100644
--- a/test/e2e/framework.go
+++ b/test/e2e/framework.go
@@ -1275,6 +1275,20 @@ func (b *PodBuilder) Create(data *TestData) error {
 	return nil
 }
 
+func (data *TestData) UpdatePod(namespace, name string, mutateFunc func(*corev1.Pod)) error {
+	pod, err := data.clientset.CoreV1().Pods(namespace).Get(context.TODO(), name, metav1.GetOptions{})
+	if err != nil {
+		return fmt.Errorf("error when getting '%s/%s' Pod: %v", namespace, name, err)
+	}
+	if mutateFunc != nil {
+		mutateFunc(pod)
+	}
+	if _, err := data.clientset.CoreV1().Pods(namespace).Update(context.TODO(), pod, metav1.UpdateOptions{}); err != nil {
+		return fmt.Errorf("error when updating '%s/%s' Pod: %v", namespace, name, err)
+	}
+	return nil
+}
+
 // createBusyboxPodOnNode creates a Pod in the test namespace with a single busybox container. The
 // Pod will be scheduled on the specified Node (if nodeName is not empty).
 func (data *TestData) createBusyboxPodOnNode(name string, ns string, nodeName string, hostNetwork bool) error {
diff --git a/test/e2e/traceflow_test.go b/test/e2e/traceflow_test.go
index a3479ceaa..2d5b0c9bb 100644
--- a/test/e2e/traceflow_test.go
+++ b/test/e2e/traceflow_test.go
@@ -26,12 +26,15 @@ import (
 	"github.com/stretchr/testify/require"
 	corev1 "k8s.io/api/core/v1"
 	networkingv1 "k8s.io/api/networking/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/util/retry"
 
 	"antrea.io/antrea/pkg/agent/config"
 	"antrea.io/antrea/pkg/apis/controlplane/v1beta2"
 	"antrea.io/antrea/pkg/apis/crd/v1alpha1"
+	"antrea.io/antrea/pkg/apis/crd/v1alpha2"
 	"antrea.io/antrea/pkg/features"
 	"antrea.io/antrea/pkg/util/k8s"
 )
@@ -77,6 +80,13 @@ func TestTraceflow(t *testing.T) {
 		skipIfEncapModeIsNot(t, data, config.TrafficEncapModeEncap)
 		testTraceflowExternalIP(t, data)
 	})
+	t.Run("testTraceflowEgress", func(t *testing.T) {
+		skipIfHasWindowsNodes(t)
+		skipIfEncapModeIsNot(t, data, config.TrafficEncapModeEncap)
+		skipIfAntreaIPAMTest(t)
+		skipIfEgressDisabled(t)
+		testTraceflowEgress(t, data)
+	})
 }
 
 func skipIfTraceflowDisabled(t *testing.T) {
@@ -2057,6 +2067,167 @@ func testTraceflowExternalIP(t *testing.T, data *TestData) {
 	runTestTraceflow(t, data, testcase)
 }
 
+func testTraceflowEgress(t *testing.T, data *TestData) {
+	egressNode := nodeName(0)
+	egressIP := nodeIP(0)
+	externalDstIP := "1.1.1.1"
+
+	localPodNames, _, localCleanupFn := createTestAgnhostPods(t, data, 1, data.testNamespace, egressNode)
+	defer localCleanupFn()
+
+	matchExpressions := []metav1.LabelSelectorRequirement{
+		{
+			Key:      "antrea-e2e",
+			Operator: metav1.LabelSelectorOpIn,
+			Values:   []string{localPodNames[0]},
+		},
+	}
+
+	egress := data.createEgress(t, "egress-", matchExpressions, nil, "", egressIP)
+	defer data.crdClient.CrdV1alpha2().Egresses().Delete(context.TODO(), egress.Name, metav1.DeleteOptions{})
+
+	testcaseLocalEgress := testcase{
+		name:      "egressFromLocalNode",
+		ipVersion: 4,
+		tf: &v1alpha1.Traceflow{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: randName(fmt.Sprintf("%s-%s-to-%s-%s-", data.testNamespace, localPodNames[0], data.testNamespace, strings.ReplaceAll(externalDstIP, ":", "--"))),
+			},
+			Spec: v1alpha1.TraceflowSpec{
+				Source: v1alpha1.Source{
+					Namespace: data.testNamespace,
+					Pod:       localPodNames[0],
+				},
+				Destination: v1alpha1.Destination{
+					IP: externalDstIP,
+				},
+				Packet: v1alpha1.Packet{
+					IPHeader: v1alpha1.IPHeader{
+						Protocol: protocolICMP,
+					},
+				},
+			},
+		},
+		expectedPhase: v1alpha1.Succeeded,
+		expectedResults: []v1alpha1.NodeResult{
+			{
+				Node: egressNode,
+				Observations: []v1alpha1.Observation{
+					{
+						Component: v1alpha1.ComponentSpoofGuard,
+						Action:    v1alpha1.ActionForwarded,
+					},
+					{
+						Component: v1alpha1.ComponentEgress,
+						Action:    v1alpha1.ActionMarkedForSNAT,
+						Egress:    egress.Name,
+						EgressIP:  egressIP,
+					},
+					{
+						Component:     v1alpha1.ComponentForwarding,
+						ComponentInfo: "Output",
+						Action:        v1alpha1.ActionForwardedOutOfOverlay,
+					},
+				},
+			},
+		},
+	}
+
+	t.Run(testcaseLocalEgress.name, func(t *testing.T) {
+		runTestTraceflow(t, data, testcaseLocalEgress)
+	})
+
+	skipIfNumNodesLessThan(t, 2)
+	remoteNode := nodeName(1)
+	remotePodNames, _, remoteCleanupFn := createTestAgnhostPods(t, data, 1, data.testNamespace, remoteNode)
+	defer remoteCleanupFn()
+
+	toUpdate := egress.DeepCopy()
+	err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
+		toUpdate.Spec.AppliedTo = v1alpha2.AppliedTo{
+			PodSelector: &metav1.LabelSelector{
+				MatchLabels: map[string]string{"antrea-e2e": remotePodNames[0]},
+			},
+		}
+		_, err := data.crdClient.CrdV1alpha2().Egresses().Update(context.TODO(), toUpdate, metav1.UpdateOptions{})
+		if err != nil && errors.IsConflict(err) {
+			toUpdate, _ = data.crdClient.CrdV1alpha2().Egresses().Get(context.TODO(), egress.Name, metav1.GetOptions{})
+		}
+		return err
+	})
+	require.NoError(t, err, "Failed to update Egress")
+
+	testcaseRemoteEgress := testcase{
+		name:      "egressFromRemoteNode",
+		ipVersion: 4,
+		tf: &v1alpha1.Traceflow{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: randName(fmt.Sprintf("%s-%s-to-%s-%s-", data.testNamespace, remotePodNames[0], data.testNamespace, strings.ReplaceAll(externalDstIP, ":", "--"))),
+			},
+			Spec: v1alpha1.TraceflowSpec{
+				Source: v1alpha1.Source{
+					Namespace: data.testNamespace,
+					Pod:       remotePodNames[0],
+				},
+				Destination: v1alpha1.Destination{
+					IP: externalDstIP,
+				},
+				Packet: v1alpha1.Packet{
+					IPHeader: v1alpha1.IPHeader{
+						Protocol: protocolICMP,
+					},
+				},
+			},
+		},
+		expectedPhase: v1alpha1.Succeeded,
+		expectedResults: []v1alpha1.NodeResult{
+			{
+				Node: remoteNode,
+				Observations: []v1alpha1.Observation{
+					{
+						Component: v1alpha1.ComponentSpoofGuard,
+						Action:    v1alpha1.ActionForwarded,
+					},
+					{
+						Component: v1alpha1.ComponentEgress,
+						Action:    v1alpha1.ActionForwardedToEgressNode,
+						Egress:    egress.Name,
+						EgressIP:  egressIP,
+					},
+					{
+						Component:     v1alpha1.ComponentForwarding,
+						ComponentInfo: "Output",
+						Action:        v1alpha1.ActionForwarded,
+					},
+				},
+			},
+			{
+				Node: egressNode,
+				Observations: []v1alpha1.Observation{
+					{
+						Component: v1alpha1.ComponentForwarding,
+						Action:    v1alpha1.ActionReceived,
+					},
+					{
+						Component: v1alpha1.ComponentEgress,
+						Action:    v1alpha1.ActionMarkedForSNAT,
+						EgressIP:  egressIP,
+					},
+					{
+						Component:     v1alpha1.ComponentForwarding,
+						ComponentInfo: "Output",
+						Action:        v1alpha1.ActionForwardedOutOfOverlay,
+					},
+				},
+			},
+		},
+	}
+
+	t.Run(testcaseRemoteEgress.name, func(t *testing.T) {
+		runTestTraceflow(t, data, testcaseRemoteEgress)
+	})
+}
+
 func (data *TestData) waitForTraceflow(t *testing.T, name string, phase v1alpha1.TraceflowPhase) (*v1alpha1.Traceflow, error) {
 	var tf *v1alpha1.Traceflow
 	var err error
@@ -2091,6 +2262,8 @@ func compareObservations(expected v1alpha1.NodeResult, actual v1alpha1.NodeResul
 			exObs[i].ComponentInfo != acObs[i].ComponentInfo ||
 			exObs[i].Pod != acObs[i].Pod ||
 			exObs[i].TranslatedDstIP != acObs[i].TranslatedDstIP ||
+			exObs[i].EgressIP != acObs[i].EgressIP ||
+			exObs[i].Egress != acObs[i].Egress ||
 			exObs[i].Action != acObs[i].Action {
 			return fmt.Errorf("Observations should be %v, but got %v", exObs, acObs)
 		}
diff --git a/test/e2e/utils/cnp_spec_builder.go b/test/e2e/utils/cnp_spec_builder.go
index ea73c34b2..9907e56d9 100644
--- a/test/e2e/utils/cnp_spec_builder.go
+++ b/test/e2e/utils/cnp_spec_builder.go
@@ -261,6 +261,29 @@ func (b *ClusterNetworkPolicySpecBuilder) AddToServicesRule(svcRefs []crdv1alpha
 	return b
 }
 
+func (b *ClusterNetworkPolicySpecBuilder) AddStretchedIngressRule(pSel, nsSel map[string]string,
+	name string, ruleAppliedToSpecs []ACNPAppliedToSpec, action crdv1alpha1.RuleAction) *ClusterNetworkPolicySpecBuilder {
+
+	var appliedTos []crdv1alpha1.AppliedTo
+	for _, at := range ruleAppliedToSpecs {
+		appliedTos = append(appliedTos, b.GetAppliedToPeer(at.PodSelector, at.NSSelector, at.PodSelectorMatchExp, at.NSSelectorMatchExp, at.Group, at.Service))
+	}
+	newRule := crdv1alpha1.Rule{
+		From:      []crdv1alpha1.NetworkPolicyPeer{{Scope: "ClusterSet"}},
+		Action:    &action,
+		Name:      name,
+		AppliedTo: appliedTos,
+	}
+	if len(pSel) > 0 {
+		newRule.From[0].PodSelector = &metav1.LabelSelector{MatchLabels: pSel}
+	}
+	if len(nsSel) > 0 {
+		newRule.From[0].NamespaceSelector = &metav1.LabelSelector{MatchLabels: nsSel}
+	}
+	b.Spec.Ingress = append(b.Spec.Ingress, newRule)
+	return b
+}
+
 // AddEgressDNS mutates the nth policy rule to allow DNS, convenience method
 func (b *ClusterNetworkPolicySpecBuilder) WithEgressDNS() *ClusterNetworkPolicySpecBuilder {
 	protocolUDP, _ := AntreaPolicyProtocolToK8sProtocol(ProtocolUDP)
diff --git a/test/integration/agent/cniserver_test.go b/test/integration/agent/cniserver_test.go
index 8761b4830..88d1c6340 100644
--- a/test/integration/agent/cniserver_test.go
+++ b/test/integration/agent/cniserver_test.go
@@ -608,7 +608,7 @@ func cmdAddDelCheckTest(testNS ns.NetNS, tc testCase, dataDir string) {
 	ovsPortUUID := uuid.New().String()
 	ovsServiceMock.EXPECT().CreatePort(ovsPortname, ovsPortname, mock.Any()).Return(ovsPortUUID, nil).AnyTimes()
 	ovsServiceMock.EXPECT().GetOFPort(ovsPortname, false).Return(int32(10), nil).AnyTimes()
-	ofServiceMock.EXPECT().InstallPodFlows(ovsPortname, mock.Any(), mock.Any(), mock.Any(), uint16(0)).Return(nil)
+	ofServiceMock.EXPECT().InstallPodFlows(ovsPortname, mock.Any(), mock.Any(), mock.Any(), uint16(0), nil).Return(nil)
 
 	close(tester.networkReadyCh)
 	// Test ips allocation
@@ -827,7 +827,7 @@ func TestCNIServerChaining(t *testing.T) {
 			routeMock.EXPECT().MigrateRoutesToGw(hostVeth.Name),
 			ovsServiceMock.EXPECT().CreatePort(ovsPortname, ovsPortname, mock.Any()).Return(ovsPortUUID, nil),
 			ovsServiceMock.EXPECT().GetOFPort(ovsPortname, false).Return(testContainerOFPort, nil),
-			ofServiceMock.EXPECT().InstallPodFlows(ovsPortname, []net.IP{podIP}, containerIntf.HardwareAddr, mock.Any(), uint16(0)),
+			ofServiceMock.EXPECT().InstallPodFlows(ovsPortname, []net.IP{podIP}, containerIntf.HardwareAddr, mock.Any(), uint16(0), nil),
 		)
 		mock.InOrder(orderedCalls...)
 		cniResp, err := server.CmdAdd(ctx, cniReq)
diff --git a/test/integration/agent/openflow_test.go b/test/integration/agent/openflow_test.go
index 0803c8483..52b828ee3 100644
--- a/test/integration/agent/openflow_test.go
+++ b/test/integration/agent/openflow_test.go
@@ -91,14 +91,15 @@ type testPeerConfig struct {
 }
 
 type testConfig struct {
-	bridge                string
-	nodeConfig            *agentconfig.NodeConfig
-	localPods             []*testLocalPodConfig
-	peers                 []*testPeerConfig
-	globalMAC             net.HardwareAddr
-	enableIPv6            bool
-	enableIPv4            bool
-	connectUplinkToBridge bool
+	bridge                       string
+	nodeConfig                   *agentconfig.NodeConfig
+	localPods                    []*testLocalPodConfig
+	peers                        []*testPeerConfig
+	globalMAC                    net.HardwareAddr
+	enableIPv6                   bool
+	enableIPv4                   bool
+	connectUplinkToBridge        bool
+	enableStretchedNetworkPolicy bool
 }
 
 var (
@@ -129,7 +130,7 @@ func TestConnectivityFlows(t *testing.T) {
 		ofClient.ResetOFTable()
 	}()
 
-	config := prepareConfiguration(true, false)
+	config := prepareConfiguration(true, false, false)
 
 	t.Run("testInitialize", func(t *testing.T) {
 		testInitialize(t, config)
@@ -158,6 +159,14 @@ func TestConnectivityFlows(t *testing.T) {
 	t.Run("testExternalFlows", func(t *testing.T) {
 		testExternalFlows(t, config)
 	})
+
+	stretchedNetworkPolicyConfig := prepareConfiguration(true, false, true)
+	t.Run("testInstallPodFlows", func(t *testing.T) {
+		testInstallPodFlows(t, stretchedNetworkPolicyConfig)
+	})
+	t.Run("testUninstallPodFlows", func(t *testing.T) {
+		testUninstallPodFlows(t, stretchedNetworkPolicyConfig)
+	})
 }
 
 func TestAntreaFlexibleIPAMConnectivityFlows(t *testing.T) {
@@ -177,7 +186,7 @@ func TestAntreaFlexibleIPAMConnectivityFlows(t *testing.T) {
 		ofClient.ResetOFTable()
 	}()
 
-	config := prepareConfiguration(true, false)
+	config := prepareConfiguration(true, false, false)
 	config.connectUplinkToBridge = true
 	config.localPods[0].ips = []net.IP{net.ParseIP("192.168.255.3")}
 	vlanID := uint16(100)
@@ -241,7 +250,7 @@ func TestReplayFlowsConnectivityFlows(t *testing.T) {
 		ofClient.ResetOFTable()
 	}()
 
-	config := prepareConfiguration(true, false)
+	config := prepareConfiguration(true, false, false)
 	t.Run("testInitialize", func(t *testing.T) {
 		testInitialize(t, config)
 	})
@@ -274,7 +283,7 @@ func TestReplayFlowsNetworkPolicyFlows(t *testing.T) {
 	err := ofTestUtils.PrepareOVSBridge(br)
 	require.Nil(t, err, fmt.Sprintf("Failed to prepare OVS bridge: %v", err))
 
-	config := prepareConfiguration(true, false)
+	config := prepareConfiguration(true, false, false)
 	_, err = c.Initialize(roundInfo, config.nodeConfig, &agentconfig.NetworkConfig{TrafficEncapMode: agentconfig.TrafficEncapModeEncap, IPv4Enabled: true}, &agentconfig.EgressConfig{}, &agentconfig.ServiceConfig{})
 	require.Nil(t, err, "Failed to initialize OFClient")
 
@@ -419,11 +428,17 @@ func testUninstallNodeFlows(t *testing.T, config *testConfig) {
 func testInstallPodFlows(t *testing.T, config *testConfig) {
 	gatewayConfig := config.nodeConfig.GatewayConfig
 	for _, pod := range config.localPods {
-		err := c.InstallPodFlows(pod.name, pod.ips, pod.mac, pod.ofPort, pod.vlanID)
+		var err error
+		if config.enableStretchedNetworkPolicy {
+			labelIdentity := ofClient.UnknownLabelIdentity
+			err = c.InstallPodFlows(pod.name, pod.ips, pod.mac, pod.ofPort, pod.vlanID, &labelIdentity)
+		} else {
+			err = c.InstallPodFlows(pod.name, pod.ips, pod.mac, pod.ofPort, pod.vlanID, nil)
+		}
 		if err != nil {
 			t.Fatalf("Failed to install Openflow entries for pod: %v", err)
 		}
-		for _, tableFlow := range preparePodFlows(pod.ips, pod.mac, pod.ofPort, gatewayConfig.MAC, config.globalMAC, config.nodeConfig, config.connectUplinkToBridge, pod.vlanID) {
+		for _, tableFlow := range preparePodFlows(pod.ips, pod.mac, pod.ofPort, gatewayConfig.MAC, config.globalMAC, config.nodeConfig, config.connectUplinkToBridge, pod.vlanID, config.enableStretchedNetworkPolicy) {
 			ofTestUtils.CheckFlowExists(t, ovsCtlClient, tableFlow.tableName, 0, true, tableFlow.flows)
 		}
 	}
@@ -436,7 +451,7 @@ func testUninstallPodFlows(t *testing.T, config *testConfig) {
 		if err != nil {
 			t.Fatalf("Failed to uninstall Openflow entries for pod: %v", err)
 		}
-		for _, tableFlow := range preparePodFlows(pod.ips, pod.mac, pod.ofPort, gatewayConfig.MAC, config.globalMAC, config.nodeConfig, config.connectUplinkToBridge, pod.vlanID) {
+		for _, tableFlow := range preparePodFlows(pod.ips, pod.mac, pod.ofPort, gatewayConfig.MAC, config.globalMAC, config.nodeConfig, config.connectUplinkToBridge, pod.vlanID, config.enableStretchedNetworkPolicy) {
 			ofTestUtils.CheckFlowExists(t, ovsCtlClient, tableFlow.tableName, 0, false, tableFlow.flows)
 		}
 	}
@@ -451,7 +466,7 @@ func TestNetworkPolicyFlows(t *testing.T) {
 	err := ofTestUtils.PrepareOVSBridge(br)
 	require.Nil(t, err, fmt.Sprintf("Failed to prepare OVS bridge %s", br))
 
-	config := prepareConfiguration(true, true)
+	config := prepareConfiguration(true, true, false)
 	_, err = c.Initialize(roundInfo, config.nodeConfig, &agentconfig.NetworkConfig{TrafficEncapMode: agentconfig.TrafficEncapModeEncap, IPv4Enabled: true, IPv6Enabled: true}, &agentconfig.EgressConfig{}, &agentconfig.ServiceConfig{})
 	require.Nil(t, err, "Failed to initialize OFClient")
 
@@ -573,7 +588,7 @@ func TestIPv6ConnectivityFlows(t *testing.T) {
 		ofClient.CleanOFTableCache()
 		ofClient.ResetOFTable()
 	}()
-	config := prepareConfiguration(false, true)
+	config := prepareConfiguration(false, true, false)
 	t.Run("testInitialize", func(t *testing.T) {
 		testInitialize(t, config)
 	})
@@ -613,7 +628,7 @@ func TestProxyServiceFlowsAntreaPolicyDisabled(t *testing.T) {
 	err := ofTestUtils.PrepareOVSBridge(br)
 	require.Nil(t, err, fmt.Sprintf("Failed to prepare OVS bridge %s", br))
 
-	config := prepareConfiguration(true, false)
+	config := prepareConfiguration(true, false, false)
 	_, err = c.Initialize(roundInfo, config.nodeConfig, &agentconfig.NetworkConfig{TrafficEncapMode: agentconfig.TrafficEncapModeEncap, IPv4Enabled: true}, &agentconfig.EgressConfig{}, &agentconfig.ServiceConfig{})
 	require.Nil(t, err, "Failed to initialize OFClient")
 
@@ -703,7 +718,7 @@ func TestProxyServiceFlowsAntreaPoilcyEnabled(t *testing.T) {
 	err := ofTestUtils.PrepareOVSBridge(br)
 	require.Nil(t, err, fmt.Sprintf("Failed to prepare OVS bridge %s", br))
 
-	config := prepareConfiguration(true, false)
+	config := prepareConfiguration(true, false, false)
 	_, err = c.Initialize(roundInfo, config.nodeConfig, &agentconfig.NetworkConfig{TrafficEncapMode: agentconfig.TrafficEncapModeEncap, IPv4Enabled: true}, &agentconfig.EgressConfig{}, &agentconfig.ServiceConfig{})
 	require.Nil(t, err, "Failed to initialize OFClient")
 
@@ -1041,7 +1056,7 @@ func testInstallGatewayFlows(t *testing.T, config *testConfig) {
 	}
 }
 
-func prepareConfiguration(enableIPv4, enableIPv6 bool) *testConfig {
+func prepareConfiguration(enableIPv4, enableIPv6, enableStretchedNetworkPolicy bool) *testConfig {
 	podMAC, _ := net.ParseMAC("aa:aa:aa:aa:aa:13")
 	gwMAC, _ := net.ParseMAC("aa:aa:aa:aa:aa:11")
 	uplinkMAC, _ := net.ParseMAC("aa:aa:aa:aa:aa:12")
@@ -1089,32 +1104,37 @@ func prepareConfiguration(enableIPv4, enableIPv6 bool) *testConfig {
 
 	vMAC, _ := net.ParseMAC("aa:bb:cc:dd:ee:ff")
 	return &testConfig{
-		bridge:     br,
-		nodeConfig: nodeConfig,
-		localPods:  []*testLocalPodConfig{podCfg},
-		peers:      []*testPeerConfig{peerNode},
-		globalMAC:  vMAC,
-		enableIPv4: enableIPv4,
-		enableIPv6: enableIPv6,
+		bridge:                       br,
+		nodeConfig:                   nodeConfig,
+		localPods:                    []*testLocalPodConfig{podCfg},
+		peers:                        []*testPeerConfig{peerNode},
+		globalMAC:                    vMAC,
+		enableIPv4:                   enableIPv4,
+		enableIPv6:                   enableIPv6,
+		enableStretchedNetworkPolicy: enableStretchedNetworkPolicy,
 	}
 }
 
-func preparePodFlows(podIPs []net.IP, podMAC net.HardwareAddr, podOFPort uint32, gwMAC, vMAC net.HardwareAddr, nodeConfig *agentconfig.NodeConfig, connectUplinkToBridge bool, vlanID uint16) []expectTableFlows {
+func preparePodFlows(podIPs []net.IP, podMAC net.HardwareAddr, podOFPort uint32, gwMAC, vMAC net.HardwareAddr, nodeConfig *agentconfig.NodeConfig, connectUplinkToBridge bool, vlanID uint16, enableStretchedNetworkPolicy bool) []expectTableFlows {
 	podIPv4 := util.GetIPv4Addr(podIPs)
 	isAntreaFlexibleIPAM := connectUplinkToBridge && podIPv4 != nil && !nodeConfig.PodIPv4CIDR.Contains(podIPv4)
 	actionNotAntreaFlexibleIPAMString := ""
+	actionNotMulticlusterString := ""
 	matchRewriteMACMarkString := ",reg0=0x200/0x200"
 	if isAntreaFlexibleIPAM {
 		actionNotAntreaFlexibleIPAMString = ",set_field:0x100000/0x100000->reg4,set_field:0x200/0x200->reg0"
 		matchRewriteMACMarkString = ""
 	}
+	if enableStretchedNetworkPolicy {
+		actionNotMulticlusterString = ",set_field:0xffffff->tun_id"
+	}
 	flows := []expectTableFlows{
 		{
 			"Classifier",
 			[]*ofTestUtils.ExpectFlow{
 				{
 					MatchStr: fmt.Sprintf("priority=190,in_port=%d", podOFPort),
-					ActStr:   fmt.Sprintf("set_field:0x3/0xf->reg0%s,goto_table:SpoofGuard", actionNotAntreaFlexibleIPAMString),
+					ActStr:   fmt.Sprintf("set_field:0x3/0xf->reg0%s%s,goto_table:SpoofGuard", actionNotAntreaFlexibleIPAMString, actionNotMulticlusterString),
 				},
 			},
 		},
@@ -1701,7 +1721,7 @@ func prepareEgressMarkFlows(snatIP net.IP, mark, podOFPort, podOFPortRemote uint
 				},
 				{
 					MatchStr: fmt.Sprintf("priority=200,%s,in_port=%d", ipProtoStr, podOFPortRemote),
-					ActStr:   fmt.Sprintf("set_field:%s->eth_src,set_field:%s->eth_dst,set_field:%s->%s,set_field:0x10/0xf0->reg0,goto_table:L2ForwardingCalc", localGwMAC.String(), vMAC.String(), snatIP, tunDstFieldName),
+					ActStr:   fmt.Sprintf("set_field:%s->eth_src,set_field:%s->eth_dst,set_field:%s->%s,set_field:0x10/0xf0->reg0,set_field:0x40000/0x40000->reg0,goto_table:L2ForwardingCalc", localGwMAC.String(), vMAC.String(), snatIP, tunDstFieldName),
 				},
 			},
 		},
@@ -1767,7 +1787,7 @@ func TestEgressMarkFlows(t *testing.T) {
 	err := ofTestUtils.PrepareOVSBridge(br)
 	require.Nil(t, err, fmt.Sprintf("Failed to prepare OVS bridge %s", br))
 
-	config := prepareConfiguration(true, true)
+	config := prepareConfiguration(true, true, false)
 	_, err = c.Initialize(roundInfo, config.nodeConfig, &agentconfig.NetworkConfig{TrafficEncapMode: agentconfig.TrafficEncapModeEncap}, &agentconfig.EgressConfig{}, &agentconfig.ServiceConfig{})
 	require.Nil(t, err, "Failed to initialize OFClient")
 
@@ -1824,7 +1844,7 @@ func TestTrafficControlFlows(t *testing.T) {
 	err := ofTestUtils.PrepareOVSBridge(br)
 	require.Nil(t, err, fmt.Sprintf("Failed to prepare OVS bridge %s", br))
 
-	config := prepareConfiguration(true, false)
+	config := prepareConfiguration(true, false, false)
 	_, err = c.Initialize(roundInfo, config.nodeConfig, &agentconfig.NetworkConfig{TrafficEncapMode: agentconfig.TrafficEncapModeEncap, IPv4Enabled: config.enableIPv4}, &agentconfig.EgressConfig{}, &agentconfig.ServiceConfig{})
 	require.Nil(t, err, "Failed to initialize OFClient")