Repository: antrea-io/antrea
Pull request #3555 (2022-03-30 13:29:17)

Unify the installation of initial flows in AntreaProxy
Functions listed in the following are used to install initial flows. However,
in framework Flexible Pipeline, every feature has a method `initFlows()`
to install initial flows. This PR moves the installation of initial flows in
the following functions to the method `initFlows()` of corresponding
features.

 - InstallGatewayFlows() error -> feature PodConnectivity
 - InstallClusterServiceCIDRFlows(serviceNets []*net.IPNet) error -> feature Service
 - InstallDefaultServiceFlows(nodePortAddressesIPv4, nodePortAddressesIPv6 []net.IP) error -> feature Service
 - InstallDefaultTunnelFlows() error -> feature PodConnectivity
 - InstallBridgeUplinkFlows() error -> feature PodConnectivity
 - InstallExternalFlows(exceptCIDRs []net.IPNet) error  -> feature Egress

 Signed-off-by: Hongliang Liu <lhongliang@vmware.com>

hongliangl: 
/test-all-features-conformance
/test-conformance
/test-e2e
/test-flexible-ipam-e2e
/test-ipv6-conformance
/test-ipv6-e2e
/test-ipv6-networkpolicy
/test-ipv6-only-conformance
/test-ipv6-only-e2e
/test-ipv6-only-networkpolicy
/test-multicluster-e2e
/test-networkpolicy
/test-windows-conformance
/test-windows-e2e
/test-windows-networkpolicy
/test-windows-proxyall-e2e
/test-integration

luolanzone: I suppose this should have no impact to multi-cluster e2e, but copy-span ACNP test failed, rerun it.
/test-multicluster-e2e 
hongliangl: /test-all-features-conformance
/test-conformance
/test-e2e
/test-flexible-ipam-e2e
/test-ipv6-conformance
/test-ipv6-e2e
/test-ipv6-networkpolicy
/test-ipv6-only-conformance
/test-ipv6-only-e2e
/test-ipv6-only-networkpolicy
/test-multicluster-e2e
/test-networkpolicy
/test-windows-conformance
/test-windows-e2e
/test-windows-networkpolicy
/test-windows-proxyall-e2e
/test-integration

Patch:
diff --git a/.github/dependabot.yml b/.github/dependabot.yml
index cc5bb5a6e..c90c7c6b1 100644
--- a/.github/dependabot.yml
+++ b/.github/dependabot.yml
@@ -7,6 +7,8 @@ updates:
     open-pull-requests-limit: 5
     ignore:
     - dependency-name: "k8s.io/*"
+    - dependency-name: "sigs.k8s.io/*"
+    - dependency-name: "github.com/containernetworking/*"
     - dependency-name: "github.com/vmware/go-ipfix"
     - dependency-name: "github.com/TomCodeLV/OVSDB-golang-lib"
     - dependency-name: "github.com/vmware-tanzu/octant"
diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml
index 394e150cd..ad31cfeef 100644
--- a/.github/workflows/build.yml
+++ b/.github/workflows/build.yml
@@ -5,10 +5,12 @@ on:
     branches:
       - main
       - release-*
+      - feature/*
   push:
     branches:
       - main
       - release-*
+      - feature/*
 
 jobs:
   check-changes:
diff --git a/.github/workflows/go.yml b/.github/workflows/go.yml
index 559e48b2f..4da5667d4 100644
--- a/.github/workflows/go.yml
+++ b/.github/workflows/go.yml
@@ -4,10 +4,12 @@ on:
     branches:
     - main
     - release-*
+    - feature/*
   push:
     branches:
     - main
     - release-*
+    - feature/*
 
 env:
   go-cache-name: go
@@ -203,12 +205,21 @@ jobs:
       uses: actions/checkout@v3
     - name: Run verify scripts
       run: make verify
-    - name: Checking for broken Markdown links
+    - name: Checking for broken Markdown links for main branch
+      if: ${{ github.base_ref == 'main' }}
       uses: gaurav-nelson/github-action-markdown-link-check@v1
       with:
         folder-path: './docs'
         file-path: './README.md, ./CHANGELOG.md, ./CONTRIBUTING.md, ./GOVERNANCE.md, ./MAINTAINERS.md, ./ROADMAP.md, ./SECURITY.md'
         config-file: 'hack/.md_links_config.json'
+    - name: Checking for broken Markdown links for release branch
+      if: ${{ startsWith(github.base_ref, 'release-') }}
+      uses: gaurav-nelson/github-action-markdown-link-check@v1
+      with:
+        # Check modified files only as there may be stale links in previous CHANGELOGs and markdown files.
+        check-modified-files-only: yes
+        base-branch: ${{ github.base_ref }}
+        config-file: 'hack/.md_links_config.json'
     - name: Markdownlint
       run: |
         sudo npm install -g markdownlint-cli@0.31.1
diff --git a/.github/workflows/golicense.yml b/.github/workflows/golicense.yml
index 4c7ecdf59..2fc68ce80 100644
--- a/.github/workflows/golicense.yml
+++ b/.github/workflows/golicense.yml
@@ -4,10 +4,12 @@ on:
     branches:
     - main
     - release-*
+    - feature/*
   push:
     branches:
     - main
     - release-*
+    - feature/*
   release:
     types:
       - created
diff --git a/.github/workflows/kind.yml b/.github/workflows/kind.yml
index 0f0946459..f8bced694 100644
--- a/.github/workflows/kind.yml
+++ b/.github/workflows/kind.yml
@@ -4,10 +4,12 @@ on:
     branches:
     - main
     - release-*
+    - feature/*
   push:
     branches:
     - main
     - release-*
+    - feature/*
 
 env:
   KIND_VERSION: v0.11.1
diff --git a/.github/workflows/kind_upgrade.yml b/.github/workflows/kind_upgrade.yml
index 03bb3df34..ee3fe928d 100644
--- a/.github/workflows/kind_upgrade.yml
+++ b/.github/workflows/kind_upgrade.yml
@@ -4,10 +4,12 @@ on:
     branches:
     - main
     - release-*
+    - feature/*
   push:
     branches:
     - main
     - release-*
+    - feature/*
 
 env:
   KIND_VERSION: v0.11.1
diff --git a/CHANGELOG/CHANGELOG-1.6.md b/CHANGELOG/CHANGELOG-1.6.md
index 61d14f1b2..834d97f5a 100644
--- a/CHANGELOG/CHANGELOG-1.6.md
+++ b/CHANGELOG/CHANGELOG-1.6.md
@@ -2,7 +2,8 @@
 
 ## 1.6.0 - 2022-03-29
 
-The Egress feature is graduated from Alpha to Beta and is therefore enabled by default.
+- The Egress feature is graduated from Alpha to Beta and is therefore enabled by default.
+- The support for proxying all Service traffic by Antrea Proxy (enabled by `antreaProxy.proxyAll`) is now Beta.
 
 ### Added
 
@@ -78,7 +79,7 @@ The Egress feature is graduated from Alpha to Beta and is therefore enabled by d
 - Fix CT mark matching without range in flow exporter. ([#3348](https://github.com/antrea-io/antrea/pull/3348), [@hongliangl])
 - [Windows] Enable IP forwarding of the Windows bridge local interface to fix support for Service of type LoadBalancer. ([#3137](https://github.com/antrea-io/antrea/pull/3137), [@hongliangl])
 
-[Antrea Multi-cluster]: https://github.com/antrea-io/antrea/blob/v1.6.0/docs/multicluster/getting-started.md
+[Antrea Multi-cluster]: https://github.com/antrea-io/antrea/blob/v1.6.0/docs/multicluster/user-guide.md
 [Antrea IPAM]: https://github.com/antrea-io/antrea/blob/v1.6.0/docs/antrea-ipam.md
 [AntreaPolicy]: https://github.com/antrea-io/antrea/blob/v1.6.0/docs/antrea-network-policy.md
 
diff --git a/cmd/antrea-agent/agent.go b/cmd/antrea-agent/agent.go
index 0f3bd816c..1741a0774 100644
--- a/cmd/antrea-agent/agent.go
+++ b/cmd/antrea-agent/agent.go
@@ -64,7 +64,6 @@ import (
 	"antrea.io/antrea/pkg/util/channel"
 	"antrea.io/antrea/pkg/util/cipher"
 	"antrea.io/antrea/pkg/util/k8s"
-	"antrea.io/antrea/pkg/util/runtime"
 	"antrea.io/antrea/pkg/version"
 )
 
@@ -84,9 +83,6 @@ var excludeNodePortDevices = []string{"antrea-egress0", "antrea-ingress0", "kube
 func run(o *Options) error {
 	klog.Infof("Starting Antrea agent (version %s)", version.GetFullVersion())
 
-	// Windows platform doesn't support Egress feature yet.
-	egressEnabled := features.DefaultFeatureGate.Enabled(features.Egress) && !runtime.IsWindowsPlatform()
-
 	// Create K8s Clientset, CRD Clientset and SharedInformerFactory for the given config.
 	k8sClient, _, crdClient, _, err := k8s.CreateClients(o.config.ClientConnection, o.config.KubeAPIServerOverride)
 	if err != nil {
@@ -119,6 +115,7 @@ func run(o *Options) error {
 	}
 	defer ovsdbConnection.Close()
 
+	egressEnabled := features.DefaultFeatureGate.Enabled(features.Egress)
 	enableBridgingMode := features.DefaultFeatureGate.Enabled(features.AntreaIPAM) && o.config.EnableBridgingMode
 	// Bridging mode will connect the uplink interface to the OVS bridge.
 	connectUplinkToBridge := enableBridgingMode
@@ -191,6 +188,12 @@ func run(o *Options) error {
 			return fmt.Errorf("getting available NodePort IP addresses failed: %v", err)
 		}
 	}
+	serviceConfig := &config.ServiceConfig{
+		ServiceCIDR:           serviceCIDRNet,
+		ServiceCIDRv6:         serviceCIDRNetv6,
+		NodePortAddressesIPv4: nodePortAddressesIPv4,
+		NodePortAddressesIPv6: nodePortAddressesIPv6,
+	}
 
 	// Initialize agent and node network.
 	agentInitializer := agent.NewInitializer(
@@ -202,17 +205,14 @@ func run(o *Options) error {
 		o.config.OVSBridge,
 		o.config.HostGateway,
 		o.config.DefaultMTU,
-		serviceCIDRNet,
-		serviceCIDRNetv6,
 		networkConfig,
 		wireguardConfig,
 		egressConfig,
+		serviceConfig,
 		networkReadyCh,
 		stopCh,
 		features.DefaultFeatureGate.Enabled(features.AntreaProxy),
 		o.config.AntreaProxy.ProxyAll,
-		nodePortAddressesIPv4,
-		nodePortAddressesIPv6,
 		connectUplinkToBridge)
 	err = agentInitializer.Initialize()
 	if err != nil {
diff --git a/docs/cookbooks/multus/README.md b/docs/cookbooks/multus/README.md
index d159621be..e33d80ad5 100644
--- a/docs/cookbooks/multus/README.md
+++ b/docs/cookbooks/multus/README.md
@@ -25,13 +25,11 @@ corresponding container images will be downloaded from public registries.
 
 For the sake of this guide, we will use macvlan in "bridge" mode, which supports
 the creation of multiple subinterfaces on one parent interface, and connects
-them all using a bridge. For more information on the different macvlan modes,
-you can refer to this [blog post](https://hicu.be/bridge-vs-macvlan). Macvlan in
-"bridge" mode requires the network to be able to handle "promiscuous mode", as
-the same physical interface / virtual adapter ends up being assigned multiple
-MAC addresses. When using a virtual network for the Nodes, some configuration
-changes are usually required, which depend on the virtualization technology. For
-example:
+them all using a bridge. Macvlan in "bridge" mode requires the network to be
+able to handle "promiscuous mode", as the same physical interface / virtual
+adapter ends up being assigned multiple MAC addresses. When using a virtual
+network for the Nodes, some configuration changes are usually required, which
+depend on the virtualization technology. For example:
 
 * when using VirtualBox and [Internal
   Networking](https://www.virtualbox.org/manual/ch06.html#network_internal), set
diff --git a/hack/release/prepare-changelog.sh b/hack/release/prepare-changelog.sh
new file mode 100755
index 000000000..f5ba0bbd6
--- /dev/null
+++ b/hack/release/prepare-changelog.sh
@@ -0,0 +1,137 @@
+#!/usr/bin/env bash
+
+# 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.
+
+set -eo pipefail
+
+function echoerr {
+    >&2 echo "$@"
+}
+
+_usage="Usage: $0 --release <RELEASE> [--from-release <RELEASE>] [--all]
+Draft the changelog for a release based on PR history starting from the last release.
+For a minor release such as 1.5.0, the changelog will base on PRs merged to main branch since the fork of release-1.4.
+For a patch release such as 1.5.2, the changelog will base on PRs merged to release-1.5 branch since the release of 1.5.1.
+
+Options:
+        --release <RELEASE>        The release for which the changelog is generated
+        --from-release <RELEASE>   The last release from which the changelog is generated. It's supposed to be used only
+                                   when running the script for a major release such as 2.0.0
+        --all                      Include PRs that are not labelled with 'action/release-note' in a separate section"
+
+function print_usage {
+  echoerr "$_usage"
+}
+
+if ! command -v gh > /dev/null; then
+  echo "Can't find 'gh' tool in PATH, please install from https://github.com/cli/cli"
+  exit 1
+fi
+
+release=""
+from_release=""
+all="no"
+limit=500
+
+while [[ $# -gt 0 ]]
+do
+key="$1"
+
+case $key in
+  --release)
+  release="$2"
+  shift 2
+  ;;
+  --from-release)
+  from_release="$2"
+  shift 2
+  ;;
+  --all)
+  all="yes"
+  shift
+  ;;
+  -h|--help)
+  print_usage
+  exit 0
+  ;;
+  *)    # unknown option
+  echoerr "Unknown option $1"
+  exit 1
+  ;;
+esac
+done
+
+if [ "$release" == "" ]; then
+  echoerr "--release must be set"
+  exit 1
+fi
+
+major_minor_number="${release%.*}"
+major_number="${major_minor_number%.*}"
+minor_number="${major_minor_number#*.}"
+patch_number="${release##*.}"
+
+if [ "$from_release" == "" ]; then
+  if [ "$patch_number" == "0" ]; then
+    from_release="${major_number}.$(( minor_number-1 )).0"
+  else
+    from_release="${major_number}.${minor_number}.$(( patch_number-1 ))"
+  fi
+fi
+
+# For minor releases, PRs are merged to main branch.
+# For patch releases, PRs are merged to release branches.
+branch="main"
+if [ "$patch_number" != "0" ]; then
+  branch="release-${major_minor_number}"
+fi
+
+# Get the best common ancestor between the current branch and the last release.
+common_ancestor=$(git merge-base "v${from_release}" "${branch}")
+# Get the merge time of the common ancestor and use it as the start time of the current release.
+release_start_time=$(gh pr list --state merged --search "${common_ancestor}" --json mergedAt -q .[0].mergedAt)
+
+# Do not generate the title for patch releases.
+if [ "$patch_number" == "0" ]; then
+  echo "# Changelog ${major_minor_number}"
+  echo ""
+fi
+echo "## ${release} - $(date +%F)"
+echo ""
+echo "### Added"
+echo ""
+
+echo "### Changed"
+echo ""
+# Put the changes under "Changed" first, release manager needs to move them to appropriate sections manually.
+gh pr list -s merged -B ${branch} --search "merged:>${release_start_time} sort:updated-desc label:action/release-note" -L $limit --json number,title,author,labels --template \
+'{{range .}}{{tablerow (printf "- %s. ([#%v](https://github.com/antrea-io/antrea/pull/%v), [@%s])" .title .number .number .author.login)}}{{end}}'
+echo ""
+
+echo "### Fixed"
+echo ""
+
+if [ "$all" == "yes" ]; then
+  # There may be some changes not being labelled properly, release manager needs to move them to appropriate sections manually.
+  echo "### Unlabelled (Remove this section eventually)"
+  echo ""
+  gh pr list -s merged -B ${branch} --search "merged:>${release_start_time} sort:updated-desc -label:action/release-note" -L $limit --json number,title,author,labels --template \
+  '{{range .}}{{tablerow (printf "- %s. ([#%v](https://github.com/antrea-io/antrea/pull/%v), [@%s])" .title .number .number .author.login)}}{{end}}'
+fi
+
+echo ""
+echo ""
+
+gh pr list -s merged -B ${branch} --search "merged:>${release_start_time} sort:author" -L $limit --json author --jq .[].author.login | sort | uniq | xargs -I AUTHOR echo "[@AUTHOR]: https://github.com/AUTHOR"
diff --git a/pkg/agent/agent.go b/pkg/agent/agent.go
index 494b3af14..c37fabd3e 100644
--- a/pkg/agent/agent.go
+++ b/pkg/agent/agent.go
@@ -85,21 +85,18 @@ type Initializer struct {
 	ovsBridge             string
 	hostGateway           string // name of gateway port on the OVS bridge
 	mtu                   int
-	serviceCIDR           *net.IPNet // K8s Service ClusterIP CIDR
-	serviceCIDRv6         *net.IPNet // K8s Service ClusterIP CIDR in IPv6
 	networkConfig         *config.NetworkConfig
 	nodeConfig            *config.NodeConfig
 	wireGuardConfig       *config.WireGuardConfig
 	egressConfig          *config.EgressConfig
+	serviceConfig         *config.ServiceConfig
 	enableProxy           bool
 	connectUplinkToBridge bool
 	// networkReadyCh should be closed once the Node's network is ready.
 	// The CNI server will wait for it before handling any CNI Add requests.
-	proxyAll              bool
-	nodePortAddressesIPv4 []net.IP
-	nodePortAddressesIPv6 []net.IP
-	networkReadyCh        chan<- struct{}
-	stopCh                <-chan struct{}
+	proxyAll       bool
+	networkReadyCh chan<- struct{}
+	stopCh         <-chan struct{}
 }
 
 func NewInitializer(
@@ -111,17 +108,14 @@ func NewInitializer(
 	ovsBridge string,
 	hostGateway string,
 	mtu int,
-	serviceCIDR *net.IPNet,
-	serviceCIDRv6 *net.IPNet,
 	networkConfig *config.NetworkConfig,
 	wireGuardConfig *config.WireGuardConfig,
 	egressConfig *config.EgressConfig,
+	serviceConfig *config.ServiceConfig,
 	networkReadyCh chan<- struct{},
 	stopCh <-chan struct{},
 	enableProxy bool,
 	proxyAll bool,
-	nodePortAddressesIPv4 []net.IP,
-	nodePortAddressesIPv6 []net.IP,
 	connectUplinkToBridge bool,
 ) *Initializer {
 	return &Initializer{
@@ -133,17 +127,14 @@ func NewInitializer(
 		ovsBridge:             ovsBridge,
 		hostGateway:           hostGateway,
 		mtu:                   mtu,
-		serviceCIDR:           serviceCIDR,
-		serviceCIDRv6:         serviceCIDRv6,
 		networkConfig:         networkConfig,
 		wireGuardConfig:       wireGuardConfig,
 		egressConfig:          egressConfig,
+		serviceConfig:         serviceConfig,
 		networkReadyCh:        networkReadyCh,
 		stopCh:                stopCh,
 		enableProxy:           enableProxy,
 		proxyAll:              proxyAll,
-		nodePortAddressesIPv4: nodePortAddressesIPv4,
-		nodePortAddressesIPv6: nodePortAddressesIPv6,
 		connectUplinkToBridge: connectUplinkToBridge,
 	}
 }
@@ -434,60 +425,12 @@ func (i *Initializer) initOpenFlowPipeline() error {
 	roundInfo := getRoundInfo(i.ovsBridgeClient)
 
 	// Set up all basic flows.
-	ofConnCh, err := i.ofClient.Initialize(roundInfo, i.nodeConfig, i.networkConfig)
+	ofConnCh, err := i.ofClient.Initialize(roundInfo, i.nodeConfig, i.networkConfig, i.egressConfig, i.serviceConfig)
 	if err != nil {
 		klog.Errorf("Failed to initialize openflow client: %v", err)
 		return err
 	}
 
-	// On Windows platform, host network flows are needed for host traffic.
-	if err := i.initHostNetworkFlows(); err != nil {
-		klog.Errorf("Failed to install openflow entries for host network: %v", err)
-		return err
-	}
-
-	// Install OpenFlow entries to enable Pod traffic to external IP
-	// addresses.
-	if err := i.ofClient.InstallExternalFlows(i.egressConfig.ExceptCIDRs); err != nil {
-		klog.Errorf("Failed to install openflow entries for external connectivity: %v", err)
-		return err
-	}
-
-	// Set up flow entries for gateway interface, including classifier, skip spoof guard check,
-	// L3 forwarding and L2 forwarding
-	if err := i.ofClient.InstallGatewayFlows(); err != nil {
-		klog.Errorf("Failed to setup openflow entries for gateway: %v", err)
-		return err
-	}
-
-	if i.networkConfig.TrafficEncapMode.SupportsEncap() {
-		// Set up flow entries for the default tunnel port interface.
-		if err := i.ofClient.InstallDefaultTunnelFlows(); err != nil {
-			klog.Errorf("Failed to setup openflow entries for tunnel interface: %v", err)
-			return err
-		}
-	}
-
-	if !i.enableProxy {
-		// Set up flow entries to enable Service connectivity. Upstream kube-proxy is leveraged to
-		// provide load-balancing, and the flows installed by this method ensure that traffic sent
-		// from local Pods to any Service address can be forwarded to the host gateway interface
-		// correctly. Otherwise packets might be dropped by egress rules before they are DNATed to
-		// backend Pods.
-		if err := i.ofClient.InstallClusterServiceCIDRFlows([]*net.IPNet{i.serviceCIDR, i.serviceCIDRv6}); err != nil {
-			klog.Errorf("Failed to setup OpenFlow entries for Service CIDRs: %v", err)
-			return err
-		}
-	} else {
-		// Set up flow entries to enable Service connectivity. The agent proxy handles
-		// ClusterIP Services while the upstream kube-proxy is leveraged to handle
-		// any other kinds of Services.
-		if err := i.ofClient.InstallDefaultServiceFlows(i.nodePortAddressesIPv4, i.nodePortAddressesIPv6); err != nil {
-			klog.Errorf("Failed to setup default OpenFlow entries for ClusterIP Services: %v", err)
-			return err
-		}
-	}
-
 	go func() {
 		// Delete stale flows from previous round. We need to wait long enough to ensure
 		// that all the flow which are still required have received an updated cookie (with
diff --git a/pkg/agent/agent_linux.go b/pkg/agent/agent_linux.go
index af324f839..5bf791fea 100644
--- a/pkg/agent/agent_linux.go
+++ b/pkg/agent/agent_linux.go
@@ -125,15 +125,6 @@ func (i *Initializer) prepareOVSBridge() error {
 	return nil
 }
 
-// initHostNetworkFlows installs Openflow flows between bridge local port and uplink port to support
-// host networking.
-func (i *Initializer) initHostNetworkFlows() error {
-	if err := i.ofClient.InstallBridgeUplinkFlows(); err != nil {
-		return err
-	}
-	return nil
-}
-
 // getTunnelLocalIP returns local_ip of tunnel port.
 // On linux platform, local_ip option is not needed.
 func (i *Initializer) getTunnelPortLocalIP() net.IP {
diff --git a/pkg/agent/agent_windows.go b/pkg/agent/agent_windows.go
index 3ecb469e3..60201a9ce 100644
--- a/pkg/agent/agent_windows.go
+++ b/pkg/agent/agent_windows.go
@@ -191,15 +191,6 @@ func (i *Initializer) prepareOVSBridge() error {
 	return nil
 }
 
-// initHostNetworkFlows installs Openflow flows between bridge local port and uplink port to support
-// host networking.
-func (i *Initializer) initHostNetworkFlows() error {
-	if err := i.ofClient.InstallBridgeUplinkFlows(); err != nil {
-		return err
-	}
-	return nil
-}
-
 // getTunnelLocalIP returns local_ip of tunnel port
 func (i *Initializer) getTunnelPortLocalIP() net.IP {
 	return i.nodeConfig.NodeTransportIPv4Addr.IP
diff --git a/pkg/agent/config/node_config.go b/pkg/agent/config/node_config.go
index 01bd20a96..5b60993e1 100644
--- a/pkg/agent/config/node_config.go
+++ b/pkg/agent/config/node_config.go
@@ -182,3 +182,11 @@ func (nc *NetworkConfig) NeedsTunnelToPeer(peerIP net.IP, localIP *net.IPNet) bo
 func (nc *NetworkConfig) NeedsDirectRoutingToPeer(peerIP net.IP, localIP *net.IPNet) bool {
 	return (nc.TrafficEncapMode == TrafficEncapModeNoEncap || nc.TrafficEncapMode == TrafficEncapModeHybrid) && localIP.Contains(peerIP)
 }
+
+// ServiceConfig includes K8s Service CIDR and available IP addresses for NodePort.
+type ServiceConfig struct {
+	ServiceCIDR           *net.IPNet // K8s Service ClusterIP CIDR
+	ServiceCIDRv6         *net.IPNet // K8s Service ClusterIP CIDR in IPv6
+	NodePortAddressesIPv4 []net.IP
+	NodePortAddressesIPv6 []net.IP
+}
diff --git a/pkg/agent/openflow/client.go b/pkg/agent/openflow/client.go
index ef097e9d7..bb852c62c 100644
--- a/pkg/agent/openflow/client.go
+++ b/pkg/agent/openflow/client.go
@@ -30,7 +30,6 @@ import (
 	"antrea.io/antrea/pkg/agent/util"
 	binding "antrea.io/antrea/pkg/ovs/openflow"
 	utilip "antrea.io/antrea/pkg/util/ip"
-	"antrea.io/antrea/pkg/util/runtime"
 	"antrea.io/antrea/third_party/proxy"
 )
 
@@ -43,22 +42,11 @@ type Client interface {
 	// be called to ensure that the set of OVS flows is correct. All flows programmed in the
 	// switch which match the current round number will be deleted before any new flow is
 	// installed.
-	Initialize(roundInfo types.RoundInfo, config *config.NodeConfig, networkconfig *config.NetworkConfig) (<-chan struct{}, error)
-
-	// InstallGatewayFlows sets up flows related to an OVS gateway port, the gateway must exist.
-	InstallGatewayFlows() error
-
-	// InstallClusterServiceCIDRFlows sets up the appropriate flows so that traffic can reach
-	// the different Services running in the Cluster. This method needs to be invoked once with
-	// the Cluster Service CIDR as a parameter.
-	InstallClusterServiceCIDRFlows(serviceNets []*net.IPNet) error
-
-	// InstallDefaultServiceFlows sets up the appropriate flows so that traffic can reach
-	// the different Services running in the Cluster. This method needs to be invoked once.
-	InstallDefaultServiceFlows(nodePortAddressesIPv4, nodePortAddressesIPv6 []net.IP) error
-
-	// InstallDefaultTunnelFlows sets up the classification flow for the default (flow based) tunnel.
-	InstallDefaultTunnelFlows() error
+	Initialize(roundInfo types.RoundInfo,
+		config *config.NodeConfig,
+		networkConfig *config.NetworkConfig,
+		egressConfig *config.EgressConfig,
+		serviceConfig *config.ServiceConfig) (<-chan struct{}, error)
 
 	// InstallNodeFlows should be invoked when a connection to a remote Node is going to be set
 	// up. The hostname is used to identify the added flows. When IPsec tunnel is enabled,
@@ -140,16 +128,6 @@ type Client interface {
 	// are removed from PolicyRule.From, else from PolicyRule.To.
 	DeletePolicyRuleAddress(ruleID uint32, addrType types.AddressType, addresses []types.Address, priority *uint16) error
 
-	// InstallBridgeUplinkFlows installs Openflow flows between bridge local port and uplink port to support host networking.
-	InstallBridgeUplinkFlows() error
-
-	// InstallExternalFlows sets up flows to enable Pods to communicate to
-	// the external IP addresses. The flows identify the packets from local
-	// Pods to the external IP address, and mark the packets to be SNAT'd
-	// with the configured SNAT IPs. On Windows Node, the flows also perform
-	// SNAT with the Openflow NAT action.
-	InstallExternalFlows(exceptCIDRs []net.IPNet) error
-
 	// InstallSNATMarkFlows installs flows for a local SNAT IP. On Linux, a
 	// single flow is added to mark the packets tunnelled from remote Nodes
 	// that should be SNAT'd with the SNAT IP.
@@ -636,80 +614,6 @@ func (c *client) GetServiceFlowKeys(svcIP net.IP, svcPort uint16, protocol bindi
 	return flowKeys
 }
 
-func (c *client) InstallDefaultServiceFlows(nodePortAddressesIPv4, nodePortAddressesIPv6 []net.IP) error {
-	flows := []binding.Flow{
-		c.featureService.serviceNeedLBFlow(),
-		c.featureService.sessionAffinityReselectFlow(),
-		c.featureService.l2ForwardOutputHairpinServiceFlow(),
-	}
-
-	if c.proxyAll {
-		for _, ipProtocol := range c.ipProtocols {
-			// These flows are used to match the first packet of NodePort. The flows will set a bit of a register to mark
-			// the Service type of the packet as NodePort. The mark will be consumed in table serviceLBTable to match NodePort
-			nodePortAddresses := nodePortAddressesIPv4
-			if ipProtocol == binding.ProtocolIPv6 {
-				nodePortAddresses = nodePortAddressesIPv6
-			}
-			flows = append(flows, c.featureService.nodePortMarkFlows(nodePortAddresses, ipProtocol)...)
-		}
-	}
-	if err := c.ofEntryOperations.AddAll(flows); err != nil {
-		return err
-	}
-	c.featureService.fixedFlows = flows
-	return nil
-}
-
-func (c *client) InstallClusterServiceCIDRFlows(serviceNets []*net.IPNet) error {
-	flows := c.featureService.serviceCIDRDNATFlows(serviceNets)
-	if err := c.ofEntryOperations.AddAll(flows); err != nil {
-		return err
-	}
-	c.featureService.fixedFlows = flows
-	return nil
-}
-
-func (c *client) InstallGatewayFlows() error {
-	gatewayConfig := c.nodeConfig.GatewayConfig
-
-	flows := []binding.Flow{
-		c.featurePodConnectivity.gatewayClassifierFlow(),
-		c.featurePodConnectivity.l2ForwardCalcFlow(gatewayConfig.MAC, config.HostGatewayOFPort),
-	}
-	flows = append(flows, c.featurePodConnectivity.gatewayIPSpoofGuardFlows()...)
-
-	// Add ARP SpoofGuard flow for local gateway interface.
-	if gatewayConfig.IPv4 != nil {
-		flows = append(flows, c.featurePodConnectivity.arpSpoofGuardFlow(gatewayConfig.IPv4, gatewayConfig.MAC, config.HostGatewayOFPort))
-		if c.connectUplinkToBridge {
-			flows = append(flows, c.featurePodConnectivity.arpSpoofGuardFlow(c.nodeConfig.NodeIPv4Addr.IP, gatewayConfig.MAC, config.HostGatewayOFPort))
-		}
-	}
-
-	// Add flow to ensure the liveness check packet could be forwarded correctly.
-	flows = append(flows, c.featurePodConnectivity.localProbeFlow(c.ovsDatapathType)...)
-	flows = append(flows, c.featurePodConnectivity.l3FwdFlowToGateway()...)
-
-	if err := c.ofEntryOperations.AddAll(flows); err != nil {
-		return err
-	}
-	c.featurePodConnectivity.fixedFlows = append(c.featurePodConnectivity.fixedFlows, flows...)
-	return nil
-}
-
-func (c *client) InstallDefaultTunnelFlows() error {
-	flows := []binding.Flow{
-		c.featurePodConnectivity.tunnelClassifierFlow(config.DefaultTunOFPort),
-		c.featurePodConnectivity.l2ForwardCalcFlow(GlobalVirtualMAC, config.DefaultTunOFPort),
-	}
-	if err := c.ofEntryOperations.AddAll(flows); err != nil {
-		return err
-	}
-	c.featurePodConnectivity.fixedFlows = append(c.featurePodConnectivity.fixedFlows, flows...)
-	return nil
-}
-
 func (c *client) initialize() error {
 	if err := c.ofEntryOperations.AddAll(c.defaultFlows()); err != nil {
 		return fmt.Errorf("failed to install default flows: %v", err)
@@ -732,9 +636,15 @@ func (c *client) initialize() error {
 	return nil
 }
 
-func (c *client) Initialize(roundInfo types.RoundInfo, nodeConfig *config.NodeConfig, networkConfig *config.NetworkConfig) (<-chan struct{}, error) {
+func (c *client) Initialize(roundInfo types.RoundInfo,
+	nodeConfig *config.NodeConfig,
+	networkConfig *config.NetworkConfig,
+	egressConfig *config.EgressConfig,
+	serviceConfig *config.ServiceConfig) (<-chan struct{}, error) {
 	c.nodeConfig = nodeConfig
 	c.networkConfig = networkConfig
+	c.egressConfig = egressConfig
+	c.serviceConfig = serviceConfig
 
 	if networkConfig.IPv4Enabled {
 		c.ipProtocols = append(c.ipProtocols, binding.ProtocolIP)
@@ -785,6 +695,7 @@ func (c *client) generatePipelines() {
 		c.ipProtocols,
 		c.nodeConfig,
 		c.networkConfig,
+		c.ovsDatapathType,
 		c.connectUplinkToBridge,
 		c.enableMulticast)
 	c.activatedFeatures = append(c.activatedFeatures, c.featurePodConnectivity)
@@ -803,6 +714,7 @@ func (c *client) generatePipelines() {
 	c.featureService = newFeatureService(c.cookieAllocator,
 		c.ipProtocols,
 		c.nodeConfig,
+		c.serviceConfig,
 		c.bridge,
 		c.enableProxy,
 		c.proxyAll,
@@ -811,7 +723,7 @@ func (c *client) generatePipelines() {
 	c.traceableFeatures = append(c.traceableFeatures, c.featureService)
 
 	if c.enableEgress {
-		c.featureEgress = newFeatureEgress(c.cookieAllocator, c.ipProtocols, c.nodeConfig)
+		c.featureEgress = newFeatureEgress(c.cookieAllocator, c.ipProtocols, c.nodeConfig, c.egressConfig)
 		c.activatedFeatures = append(c.activatedFeatures, c.featureEgress)
 	}
 
@@ -868,17 +780,6 @@ func (c *client) generatePipelines() {
 	}
 }
 
-func (c *client) InstallExternalFlows(exceptCIDRs []net.IPNet) error {
-	if c.enableEgress {
-		flows := c.featureEgress.externalFlows(exceptCIDRs)
-		if err := c.ofEntryOperations.AddAll(flows); err != nil {
-			return fmt.Errorf("failed to install flows for external communication: %v", err)
-		}
-		c.featureEgress.fixedFlows = append(c.featureEgress.fixedFlows, flows...)
-	}
-	return nil
-}
-
 func (c *client) InstallSNATMarkFlows(snatIP net.IP, mark uint32) error {
 	flow := c.featureEgress.snatIPFromTunnelFlow(snatIP, mark)
 	cacheKey := fmt.Sprintf("s%x", mark)
@@ -1190,7 +1091,7 @@ func (c *client) InstallMulticastInitialFlows(pktInReason uint8) error {
 	cacheKey := fmt.Sprintf("multicast")
 	c.replayMutex.RLock()
 	defer c.replayMutex.RUnlock()
-	return c.addFlows(c.featureMulticast.mcastFlowCache, cacheKey, flows)
+	return c.addFlows(c.featureMulticast.cachedFlows, cacheKey, flows)
 }
 
 func (c *client) InstallMulticastFlow(multicastIP net.IP) error {
@@ -1198,14 +1099,14 @@ func (c *client) InstallMulticastFlow(multicastIP net.IP) error {
 	cacheKey := fmt.Sprintf("multicast_%s", multicastIP.String())
 	c.replayMutex.RLock()
 	defer c.replayMutex.RUnlock()
-	return c.addFlows(c.featureMulticast.mcastFlowCache, cacheKey, flows)
+	return c.addFlows(c.featureMulticast.cachedFlows, cacheKey, flows)
 }
 
 func (c *client) UninstallMulticastFlow(multicastIP net.IP) error {
 	c.replayMutex.RLock()
 	defer c.replayMutex.RUnlock()
 	cacheKey := fmt.Sprintf("multicast_%s", multicastIP.String())
-	return c.deleteFlows(c.featureMulticast.mcastFlowCache, cacheKey)
+	return c.deleteFlows(c.featureMulticast.cachedFlows, cacheKey)
 }
 
 func (c *client) SendIGMPQueryPacketOut(
@@ -1227,22 +1128,3 @@ func (c *client) SendIGMPQueryPacketOut(
 	packetOutObj := packetOutBuilder.Done()
 	return c.bridge.SendPacketOut(packetOutObj)
 }
-
-func (c *client) InstallBridgeUplinkFlows() error {
-	if runtime.IsWindowsPlatform() || c.connectUplinkToBridge {
-		podCIDRMap := make(map[binding.Protocol]net.IPNet)
-		if c.nodeConfig.PodIPv4CIDR != nil {
-			podCIDRMap[binding.ProtocolIP] = *c.nodeConfig.PodIPv4CIDR
-		}
-		//TODO: support IPv6
-		flows := c.featurePodConnectivity.hostBridgeUplinkFlows(podCIDRMap)
-		if c.connectUplinkToBridge {
-			flows = append(flows, c.featurePodConnectivity.hostBridgeUplinkVLANFlows()...)
-		}
-		if err := c.ofEntryOperations.AddAll(flows); err != nil {
-			return err
-		}
-		c.featurePodConnectivity.fixedFlows = append(c.featurePodConnectivity.fixedFlows, flows...)
-	}
-	return nil
-}
diff --git a/pkg/agent/openflow/egress.go b/pkg/agent/openflow/egress.go
index 5dacb866b..0d5d980f1 100644
--- a/pkg/agent/openflow/egress.go
+++ b/pkg/agent/openflow/egress.go
@@ -27,10 +27,10 @@ type featureEgress struct {
 	ipProtocols     []binding.Protocol
 
 	cachedFlows *flowCategoryCache
-	fixedFlows  []binding.Flow
 
-	nodeIPs    map[binding.Protocol]net.IP
-	gatewayMAC net.HardwareAddr
+	exceptCIDRs map[binding.Protocol][]net.IPNet
+	nodeIPs     map[binding.Protocol]net.IP
+	gatewayMAC  net.HardwareAddr
 
 	category cookie.Category
 }
@@ -41,7 +41,17 @@ func (f *featureEgress) getFeatureName() string {
 
 func newFeatureEgress(cookieAllocator cookie.Allocator,
 	ipProtocols []binding.Protocol,
-	nodeConfig *config.NodeConfig) *featureEgress {
+	nodeConfig *config.NodeConfig,
+	egressConfig *config.EgressConfig) *featureEgress {
+	exceptCIDRs := make(map[binding.Protocol][]net.IPNet)
+	for _, cidr := range egressConfig.ExceptCIDRs {
+		if cidr.IP.To4() == nil {
+			exceptCIDRs[binding.ProtocolIPv6] = append(exceptCIDRs[binding.ProtocolIPv6], cidr)
+		} else {
+			exceptCIDRs[binding.ProtocolIP] = append(exceptCIDRs[binding.ProtocolIP], cidr)
+		}
+	}
+
 	nodeIPs := make(map[binding.Protocol]net.IP)
 	for _, ipProtocol := range ipProtocols {
 		if ipProtocol == binding.ProtocolIP {
@@ -53,6 +63,7 @@ func newFeatureEgress(cookieAllocator cookie.Allocator,
 	return &featureEgress{
 		cachedFlows:     newFlowCategoryCache(),
 		cookieAllocator: cookieAllocator,
+		exceptCIDRs:     exceptCIDRs,
 		ipProtocols:     ipProtocols,
 		nodeIPs:         nodeIPs,
 		gatewayMAC:      nodeConfig.GatewayConfig.MAC,
@@ -61,17 +72,13 @@ func newFeatureEgress(cookieAllocator cookie.Allocator,
 }
 
 func (f *featureEgress) initFlows() []binding.Flow {
-	return []binding.Flow{}
+	// This installs the flows to enable Pods to communicate to the external IP addresses. The flows identify the packets
+	// from local Pods to the external IP address, and mark the packets to be SNAT'd with the configured SNAT IPs.
+	return f.externalFlows()
 }
 
 func (f *featureEgress) replayFlows() []binding.Flow {
 	var flows []binding.Flow
-
-	// Get fixed flows.
-	for _, flow := range f.fixedFlows {
-		flow.Reset()
-		flows = append(flows, flow)
-	}
 	// Get cached flows.
 	flows = append(flows, getCachedFlows(f.cachedFlows)...)
 
diff --git a/pkg/agent/openflow/multicast.go b/pkg/agent/openflow/multicast.go
index 63bd69736..3d416aa71 100644
--- a/pkg/agent/openflow/multicast.go
+++ b/pkg/agent/openflow/multicast.go
@@ -27,7 +27,7 @@ type featureMulticast struct {
 	cookieAllocator cookie.Allocator
 	ipProtocols     []binding.Protocol
 
-	mcastFlowCache *flowCategoryCache
+	cachedFlows *flowCategoryCache
 
 	category cookie.Category
 }
@@ -40,7 +40,7 @@ func newFeatureMulticast(cookieAllocator cookie.Allocator, ipProtocols []binding
 	return &featureMulticast{
 		cookieAllocator: cookieAllocator,
 		ipProtocols:     ipProtocols,
-		mcastFlowCache:  newFlowCategoryCache(),
+		cachedFlows:     newFlowCategoryCache(),
 		category:        cookie.Multicast,
 	}
 }
@@ -61,5 +61,5 @@ func (f *featureMulticast) initFlows() []binding.Flow {
 
 func (f *featureMulticast) replayFlows() []binding.Flow {
 	// Get cached flows.
-	return getCachedFlows(f.mcastFlowCache)
+	return getCachedFlows(f.cachedFlows)
 }
diff --git a/pkg/agent/openflow/pipeline.go b/pkg/agent/openflow/pipeline.go
index 6c8c3a79a..e1c7e0e4d 100644
--- a/pkg/agent/openflow/pipeline.go
+++ b/pkg/agent/openflow/pipeline.go
@@ -414,6 +414,8 @@ type client struct {
 	replayMutex   sync.RWMutex
 	nodeConfig    *config.NodeConfig
 	networkConfig *config.NetworkConfig
+	egressConfig  *config.EgressConfig
+	serviceConfig *config.ServiceConfig
 	// ovsDatapathType is the type of the datapath used by the bridge.
 	ovsDatapathType ovsconfig.OVSDatapathType
 	// ovsMetersAreSupported indicates whether the OVS datapath supports OpenFlow meters.
@@ -1560,21 +1562,18 @@ func (f *featurePodConnectivity) gatewayIPSpoofGuardFlows() []binding.Flow {
 }
 
 // serviceCIDRDNATFlows generates the flows to match destination IP in Service CIDR and output to the Antrea gateway directly.
-func (f *featureService) serviceCIDRDNATFlows(serviceCIDRs []*net.IPNet) []binding.Flow {
+func (f *featureService) serviceCIDRDNATFlows() []binding.Flow {
 	cookieID := f.cookieAllocator.Request(f.category).Raw()
 	var flows []binding.Flow
-	for _, serviceCIDR := range serviceCIDRs {
-		if serviceCIDR != nil {
-			ipProtocol := getIPProtocol(serviceCIDR.IP)
-			flows = append(flows, DNATTable.ofTable.BuildFlow(priorityNormal).
-				Cookie(cookieID).
-				MatchProtocol(ipProtocol).
-				MatchDstIPNet(*serviceCIDR).
-				Action().LoadToRegField(TargetOFPortField, config.HostGatewayOFPort).
-				Action().LoadRegMark(OFPortFoundRegMark).
-				Action().GotoStage(stageConntrack).
-				Done())
-		}
+	for ipProtocol, serviceCIDR := range f.serviceCIDRs {
+		flows = append(flows, DNATTable.ofTable.BuildFlow(priorityNormal).
+			Cookie(cookieID).
+			MatchProtocol(ipProtocol).
+			MatchDstIPNet(serviceCIDR).
+			Action().LoadToRegField(TargetOFPortField, config.HostGatewayOFPort).
+			Action().LoadRegMark(OFPortFoundRegMark).
+			Action().GotoStage(stageConntrack).
+			Done())
 	}
 	return flows
 }
@@ -2122,10 +2121,10 @@ func (f *featureNetworkPolicy) dnsPacketInFlow(conjunctionID uint32) binding.Flo
 // Note that there is a defect in the latter way that NodePort Service access by external clients will be masqueraded as
 // a local gateway IP to bypass Network Policies. See https://github.com/antrea-io/antrea/issues/280.
 // TODO: Fix it after replacing kube-proxy with AntreaProxy.
-func (f *featurePodConnectivity) localProbeFlow(ovsDatapathType ovsconfig.OVSDatapathType) []binding.Flow {
+func (f *featurePodConnectivity) localProbeFlow() []binding.Flow {
 	cookieID := f.cookieAllocator.Request(f.category).Raw()
 	var flows []binding.Flow
-	if runtime.IsWindowsPlatform() || ovsDatapathType == ovsconfig.OVSDatapathNetdev {
+	if runtime.IsWindowsPlatform() || f.ovsDatapathType == ovsconfig.OVSDatapathNetdev {
 		for ipProtocol, gatewayIP := range f.gatewayIPs {
 			flows = append(flows, IngressSecurityClassifierTable.ofTable.BuildFlow(priorityHigh).
 				Cookie(cookieID).
@@ -2226,29 +2225,31 @@ func (f *featureEgress) snatRuleFlow(ofPort uint32, snatIP net.IP, snatMark uint
 
 // nodePortMarkFlows generates the flows to mark the first packet of Service NodePort connection with ToNodePortAddressRegMark,
 // which indicates the Service type is NodePort.
-func (f *featureService) nodePortMarkFlows(nodePortAddresses []net.IP, ipProtocol binding.Protocol) []binding.Flow {
+func (f *featureService) nodePortMarkFlows() []binding.Flow {
 	cookieID := f.cookieAllocator.Request(f.category).Raw()
 	var flows []binding.Flow
-	// This generates a flow for every NodePort IP. The flows are used to mark the first packet of NodePort connections
-	// from a local Pod.
-	for i := range nodePortAddresses {
+	for ipProtocol, nodePortAddresses := range f.nodePortAddresses {
+		// This generates a flow for every NodePort IP. The flows are used to mark the first packet of NodePort connection
+		// from a local Pod.
+		for i := range nodePortAddresses {
+			flows = append(flows,
+				NodePortMarkTable.ofTable.BuildFlow(priorityNormal).
+					Cookie(cookieID).
+					MatchProtocol(ipProtocol).
+					MatchDstIP(nodePortAddresses[i]).
+					Action().LoadRegMark(ToNodePortAddressRegMark).
+					Done())
+		}
+		// This generates the flow for the virtual IP. The flow is used to mark the first packet of NodePort connection from
+		// the Antrea gateway (the connection is performed DNAT with the virtual IP in host netns).
 		flows = append(flows,
 			NodePortMarkTable.ofTable.BuildFlow(priorityNormal).
 				Cookie(cookieID).
 				MatchProtocol(ipProtocol).
-				MatchDstIP(nodePortAddresses[i]).
+				MatchDstIP(f.virtualIPs[ipProtocol]).
 				Action().LoadRegMark(ToNodePortAddressRegMark).
 				Done())
 	}
-	// This generates the flow for the virtual IP. The flow is used to mark the first packet of NodePort connection from
-	// the Antrea gateway (the connection is performed DNAT with the virtual IP in host netns).
-	flows = append(flows,
-		NodePortMarkTable.ofTable.BuildFlow(priorityNormal).
-			Cookie(cookieID).
-			MatchProtocol(ipProtocol).
-			MatchDstIP(f.virtualIPs[ipProtocol]).
-			Action().LoadRegMark(ToNodePortAddressRegMark).
-			Done())
 
 	return flows
 }
@@ -2481,17 +2482,8 @@ func (f *featurePodConnectivity) decTTLFlows() []binding.Flow {
 
 // externalFlows generates the flows to perform SNAT for the packets of connection to the external network. The flows identify
 // the packets to external network, and send them to EgressMarkTable, where SNAT IPs are looked up for the packets.
-func (f *featureEgress) externalFlows(exceptCIDRs []net.IPNet) []binding.Flow {
+func (f *featureEgress) externalFlows() []binding.Flow {
 	cookieID := f.cookieAllocator.Request(f.category).Raw()
-	exceptCIDRsMap := make(map[binding.Protocol][]net.IPNet)
-	for _, cidr := range exceptCIDRs {
-		if cidr.IP.To4() == nil {
-			exceptCIDRsMap[binding.ProtocolIPv6] = append(exceptCIDRsMap[binding.ProtocolIPv6], cidr)
-		} else {
-			exceptCIDRsMap[binding.ProtocolIP] = append(exceptCIDRsMap[binding.ProtocolIP], cidr)
-		}
-	}
-
 	var flows []binding.Flow
 	for _, ipProtocol := range f.ipProtocols {
 		flows = append(flows,
@@ -2529,7 +2521,7 @@ func (f *featureEgress) externalFlows(exceptCIDRs []net.IPNet) []binding.Flow {
 			f.snatSkipNodeFlow(f.nodeIPs[ipProtocol]),
 		)
 		// This generates the flows to bypass the packets sourced from local Pods and destined for the except CIDRs for Egress.
-		for _, cidr := range exceptCIDRsMap[ipProtocol] {
+		for _, cidr := range f.exceptCIDRs[ipProtocol] {
 			flows = append(flows, EgressMarkTable.ofTable.BuildFlow(priorityHigh).
 				Cookie(cookieID).
 				MatchProtocol(ipProtocol).
diff --git a/pkg/agent/openflow/pod_connectivity.go b/pkg/agent/openflow/pod_connectivity.go
index 8e8354d64..efdb5fa82 100644
--- a/pkg/agent/openflow/pod_connectivity.go
+++ b/pkg/agent/openflow/pod_connectivity.go
@@ -20,6 +20,8 @@ import (
 	"antrea.io/antrea/pkg/agent/config"
 	"antrea.io/antrea/pkg/agent/openflow/cookie"
 	binding "antrea.io/antrea/pkg/ovs/openflow"
+	"antrea.io/antrea/pkg/ovs/ovsconfig"
+	"antrea.io/antrea/pkg/util/runtime"
 )
 
 type featurePodConnectivity struct {
@@ -28,7 +30,6 @@ type featurePodConnectivity struct {
 
 	nodeCachedFlows *flowCategoryCache
 	podCachedFlows  *flowCategoryCache
-	fixedFlows      []binding.Flow
 
 	gatewayIPs    map[binding.Protocol]net.IP
 	ctZones       map[binding.Protocol]int
@@ -36,6 +37,8 @@ type featurePodConnectivity struct {
 	nodeIPs       map[binding.Protocol]net.IP
 	nodeConfig    *config.NodeConfig
 	networkConfig *config.NetworkConfig
+	// ovsDatapathType is the type of the datapath used by the bridge.
+	ovsDatapathType ovsconfig.OVSDatapathType
 
 	connectUplinkToBridge bool
 	ctZoneSrcField        *binding.RegField
@@ -53,6 +56,7 @@ func newFeaturePodConnectivity(
 	ipProtocols []binding.Protocol,
 	nodeConfig *config.NodeConfig,
 	networkConfig *config.NetworkConfig,
+	ovsDatapathType ovsconfig.OVSDatapathType,
 	connectUplinkToBridge bool,
 	enableMulticast bool) *featurePodConnectivity {
 	ctZones := make(map[binding.Protocol]int)
@@ -88,6 +92,7 @@ func newFeaturePodConnectivity(
 		nodeIPs:               nodeIPs,
 		nodeConfig:            nodeConfig,
 		networkConfig:         networkConfig,
+		ovsDatapathType:       ovsDatapathType,
 		connectUplinkToBridge: connectUplinkToBridge,
 		ctZoneSrcField:        getZoneSrcField(connectUplinkToBridge),
 		enableMulticast:       enableMulticast,
@@ -97,14 +102,24 @@ func newFeaturePodConnectivity(
 
 func (f *featurePodConnectivity) initFlows() []binding.Flow {
 	var flows []binding.Flow
+	gatewayMAC := f.nodeConfig.GatewayConfig.MAC
 
 	for _, ipProtocol := range f.ipProtocols {
 		if ipProtocol == binding.ProtocolIPv6 {
 			flows = append(flows, f.ipv6Flows()...)
 		} else if ipProtocol == binding.ProtocolIP {
 			flows = append(flows, f.arpNormalFlow())
+			flows = append(flows, f.arpSpoofGuardFlow(f.gatewayIPs[ipProtocol], gatewayMAC, config.HostGatewayOFPort))
 			if f.connectUplinkToBridge {
-				flows = append(flows, f.arpResponderFlow(f.nodeConfig.GatewayConfig.IPv4, f.nodeConfig.GatewayConfig.MAC))
+				flows = append(flows, f.arpResponderFlow(f.gatewayIPs[ipProtocol], gatewayMAC))
+				flows = append(flows, f.arpSpoofGuardFlow(f.nodeConfig.NodeIPv4Addr.IP, gatewayMAC, config.HostGatewayOFPort))
+				flows = append(flows, f.hostBridgeUplinkVLANFlows()...)
+			}
+			if runtime.IsWindowsPlatform() || f.connectUplinkToBridge {
+				// This installs the flows between bridge local port and uplink port to support host networking.
+				// TODO: support IPv6
+				podCIDRMap := map[binding.Protocol]net.IPNet{binding.ProtocolIP: *f.nodeConfig.PodIPv4CIDR}
+				flows = append(flows, f.hostBridgeUplinkFlows(podCIDRMap)...)
 			}
 		}
 	}
@@ -115,10 +130,21 @@ func (f *featurePodConnectivity) initFlows() []binding.Flow {
 	flows = append(flows, f.decTTLFlows()...)
 	flows = append(flows, f.conntrackFlows()...)
 	flows = append(flows, f.l2ForwardOutputFlow())
+	flows = append(flows, f.gatewayClassifierFlow())
+	flows = append(flows, f.l2ForwardCalcFlow(gatewayMAC, config.HostGatewayOFPort))
+	flows = append(flows, f.gatewayIPSpoofGuardFlows()...)
+	flows = append(flows, f.l3FwdFlowToGateway()...)
+	// Add flow to ensure the liveliness check packet could be forwarded correctly.
+	flows = append(flows, f.localProbeFlow()...)
+
+	if f.networkConfig.TrafficEncapMode.SupportsEncap() {
+		flows = append(flows, f.tunnelClassifierFlow(config.DefaultTunOFPort))
+		flows = append(flows, f.l2ForwardCalcFlow(GlobalVirtualMAC, config.DefaultTunOFPort))
+	}
+
 	if f.networkConfig.TrafficEncapMode.IsNetworkPolicyOnly() {
 		flows = append(flows, f.l3FwdFlowRouteToGW()...)
-		// If IPv6 is enabled, this flow will never get hit.
-		// Replies any ARP request with the same global virtual MAC.
+		// If IPv6 is enabled, this flow will never get hit. Replies any ARP request with the same global virtual MAC.
 		flows = append(flows, f.arpResponderStaticFlow())
 	} else {
 		// If NetworkPolicyOnly mode is enabled, IPAM is implemented by the primary CNI, which may not use the Pod CIDR
@@ -132,11 +158,6 @@ func (f *featurePodConnectivity) initFlows() []binding.Flow {
 func (f *featurePodConnectivity) replayFlows() []binding.Flow {
 	var flows []binding.Flow
 
-	// Get fixed flows.
-	for _, flow := range f.fixedFlows {
-		flow.Reset()
-		flows = append(flows, flow)
-	}
 	// Get cached flows.
 	for _, cachedFlows := range []*flowCategoryCache{f.nodeCachedFlows, f.podCachedFlows} {
 		flows = append(flows, getCachedFlows(cachedFlows)...)
diff --git a/pkg/agent/openflow/service.go b/pkg/agent/openflow/service.go
index 7a102203b..b27dc64c1 100644
--- a/pkg/agent/openflow/service.go
+++ b/pkg/agent/openflow/service.go
@@ -31,14 +31,15 @@ type featureService struct {
 	bridge          binding.Bridge
 
 	cachedFlows *flowCategoryCache
-	fixedFlows  []binding.Flow
 	groupCache  sync.Map
 
-	gatewayIPs  map[binding.Protocol]net.IP
-	virtualIPs  map[binding.Protocol]net.IP
-	dnatCtZones map[binding.Protocol]int
-	snatCtZones map[binding.Protocol]int
-	gatewayMAC  net.HardwareAddr
+	gatewayIPs        map[binding.Protocol]net.IP
+	virtualIPs        map[binding.Protocol]net.IP
+	dnatCtZones       map[binding.Protocol]int
+	snatCtZones       map[binding.Protocol]int
+	gatewayMAC        net.HardwareAddr
+	nodePortAddresses map[binding.Protocol][]net.IP
+	serviceCIDRs      map[binding.Protocol]net.IPNet
 
 	enableProxy           bool
 	proxyAll              bool
@@ -56,6 +57,7 @@ func newFeatureService(
 	cookieAllocator cookie.Allocator,
 	ipProtocols []binding.Protocol,
 	nodeConfig *config.NodeConfig,
+	serviceConfig *config.ServiceConfig,
 	bridge binding.Bridge,
 	enableProxy,
 	proxyAll,
@@ -64,17 +66,27 @@ func newFeatureService(
 	virtualIPs := make(map[binding.Protocol]net.IP)
 	dnatCtZones := make(map[binding.Protocol]int)
 	snatCtZones := make(map[binding.Protocol]int)
+	nodePortAddresses := make(map[binding.Protocol][]net.IP)
+	serviceCIDRs := make(map[binding.Protocol]net.IPNet)
 	for _, ipProtocol := range ipProtocols {
 		if ipProtocol == binding.ProtocolIP {
 			gatewayIPs[ipProtocol] = nodeConfig.GatewayConfig.IPv4
 			virtualIPs[ipProtocol] = config.VirtualServiceIPv4
 			dnatCtZones[ipProtocol] = CtZone
 			snatCtZones[ipProtocol] = SNATCtZone
+			nodePortAddresses[ipProtocol] = serviceConfig.NodePortAddressesIPv4
+			if serviceConfig.ServiceCIDR != nil {
+				serviceCIDRs[ipProtocol] = *serviceConfig.ServiceCIDR
+			}
 		} else if ipProtocol == binding.ProtocolIPv6 {
 			gatewayIPs[ipProtocol] = nodeConfig.GatewayConfig.IPv6
 			virtualIPs[ipProtocol] = config.VirtualServiceIPv6
 			dnatCtZones[ipProtocol] = CtZoneV6
 			snatCtZones[ipProtocol] = SNATCtZoneV6
+			nodePortAddresses[ipProtocol] = serviceConfig.NodePortAddressesIPv6
+			if serviceConfig.ServiceCIDRv6 != nil {
+				serviceCIDRs[ipProtocol] = *serviceConfig.ServiceCIDRv6
+			}
 		}
 	}
 
@@ -88,6 +100,8 @@ func newFeatureService(
 		virtualIPs:            virtualIPs,
 		dnatCtZones:           dnatCtZones,
 		snatCtZones:           snatCtZones,
+		nodePortAddresses:     nodePortAddresses,
+		serviceCIDRs:          serviceCIDRs,
 		gatewayMAC:            nodeConfig.GatewayConfig.MAC,
 		enableProxy:           enableProxy,
 		proxyAll:              proxyAll,
@@ -105,6 +119,20 @@ func (f *featureService) initFlows() []binding.Flow {
 		flows = append(flows, f.l3FwdFlowsToExternalEndpoint()...)
 		flows = append(flows, f.gatewaySNATFlows()...)
 		flows = append(flows, f.snatConntrackFlows()...)
+		flows = append(flows, f.serviceNeedLBFlow())
+		flows = append(flows, f.sessionAffinityReselectFlow())
+		flows = append(flows, f.l2ForwardOutputHairpinServiceFlow())
+		if f.proxyAll {
+			// This installs the flows to match the first packet of NodePort connection. The flows set a bit of a register
+			// to mark the Service type of the packet as NodePort, and the mark is consumed in table serviceLBTable.
+			flows = append(flows, f.nodePortMarkFlows()...)
+		}
+	} else {
+		// This installs the flows to enable Service connectivity. Upstream kube-proxy is leveraged to provide load-balancing,
+		// and the flows installed by this method ensure that traffic sent from local Pods to any Service address can be
+		// forwarded to the host gateway interface correctly, otherwise packets might be dropped by egress rules before
+		// they are DNATed to backend Pods.
+		flows = append(flows, f.serviceCIDRDNATFlows()...)
 	}
 	return flows
 }
@@ -112,11 +140,6 @@ func (f *featureService) initFlows() []binding.Flow {
 func (f *featureService) replayFlows() []binding.Flow {
 	var flows []binding.Flow
 
-	// Get fixed flows.
-	for _, flow := range f.fixedFlows {
-		flow.Reset()
-		flows = append(flows, flow)
-	}
 	// Get cached flows.
 	flows = append(flows, getCachedFlows(f.cachedFlows)...)
 
diff --git a/pkg/agent/proxy/proxier.go b/pkg/agent/proxy/proxier.go
index 2ab8f7eeb..ea286f25e 100644
--- a/pkg/agent/proxy/proxier.go
+++ b/pkg/agent/proxy/proxier.go
@@ -347,7 +347,6 @@ func (p *proxier) uninstallLoadBalancerService(loadBalancerIPStrings []string, s
 func (p *proxier) installServices() {
 	for svcPortName, svcPort := range p.serviceMap {
 		svcInfo := svcPort.(*types.ServiceInfo)
-		groupID := p.groupCounter.AllocateIfNotExist(svcPortName, false)
 		endpointsInstalled, ok := p.endpointsInstalledMap[svcPortName]
 		if !ok {
 			endpointsInstalled = map[string]k8sproxy.Endpoint{}
@@ -366,47 +365,88 @@ func (p *proxier) installServices() {
 			pSvcInfo = installedSvcPort.(*types.ServiceInfo)
 			needRemoval = serviceIdentityChanged(svcInfo, pSvcInfo) || (svcInfo.SessionAffinityType() != pSvcInfo.SessionAffinityType())
 			needUpdateService = needRemoval || (svcInfo.StickyMaxAgeSeconds() != pSvcInfo.StickyMaxAgeSeconds())
-			needUpdateEndpoints = pSvcInfo.SessionAffinityType() != svcInfo.SessionAffinityType() || pSvcInfo.NodeLocalExternal() != svcInfo.NodeLocalExternal()
+			needUpdateEndpoints = pSvcInfo.SessionAffinityType() != svcInfo.SessionAffinityType() ||
+				pSvcInfo.NodeLocalExternal() != svcInfo.NodeLocalExternal() ||
+				pSvcInfo.NodeLocalInternal() != svcInfo.NodeLocalInternal()
 		} else { // Need to install.
 			needUpdateService = true
 		}
 
-		var endpointUpdateList []k8sproxy.Endpoint
+		var internalNodeLocal, externalNodeLocal bool
+		if svcInfo.NodeLocalInternal() {
+			internalNodeLocal = true
+		}
+		if p.proxyAll && svcInfo.NodeLocalExternal() {
+			externalNodeLocal = true
+		}
+
+		var allEndpointUpdateList, localEndpointUpdateList []k8sproxy.Endpoint
 		if len(endpoints) > maxEndpoints {
 			if !p.oversizeServiceSet.Has(svcPortName.String()) {
 				klog.Warningf("Since Endpoints of Service %s exceeds %d, extra Endpoints will be dropped", svcPortName.String(), maxEndpoints)
 				p.oversizeServiceSet.Insert(svcPortName.String())
 			}
-			// If the length of endpoints > maxEndpoints, endpoints should be cut. However, endpoints is a map. Therefore,
-			// iterate the map and append every Endpoint to a slice endpointList. Since the iteration order of map in
-			// Golang is random, if cut directly without any sorting, some Endpoints may not be installed. So cutting
-			// slice endpointList after sorting can avoid this situation in some degree.
-			var endpointList []k8sproxy.Endpoint
+			// If the length of endpoints > maxEndpoints, endpoints should be cut. However, since endpoints is a map, iterate
+			// the map and append every Endpoint to a target slice. When the Endpoint is local, append Endpoint to slice
+			// localEndpointList, otherwise append the Endpoint to slice remoteEndpointList. Since the iteration order
+			// of map in Golang is not guaranteed, if split the Endpoints directly without any sorting, some Endpoints
+			// may not be installed, so split the endpointList after sorting.
+			var remoteEndpointList, localEndpointList []k8sproxy.Endpoint
 			for _, endpoint := range endpoints {
-				endpointList = append(endpointList, endpoint)
+				if endpoint.GetIsLocal() {
+					localEndpointList = append(localEndpointList, endpoint)
+				} else {
+					remoteEndpointList = append(remoteEndpointList, endpoint)
+				}
 			}
-			sort.Sort(byEndpoint(endpointList))
-			endpointList = endpointList[:maxEndpoints]
 
-			for _, endpoint := range endpointList { // Check if there is any installed Endpoint which is not expected anymore.
-				if _, ok := endpointsInstalled[endpoint.String()]; !ok { // There is an expected Endpoint which is not installed.
-					needUpdateEndpoints = true
+			sort.Sort(byEndpoint(remoteEndpointList))
+			sort.Sort(byEndpoint(localEndpointList))
+
+			if len(localEndpointList) > maxEndpoints {
+				// When the number of local Endpoints is greater than maxEndpoints, choose maxEndpoints Endpoints from
+				// localEndpointList to install.
+				allEndpointUpdateList = localEndpointList[:maxEndpoints]
+			} else {
+				// When the number of local Endpoints is smaller than maxEndpoints, choose all Endpoints of localEndpointList
+				// and part of remoteEndpointList to install.
+				localEndpointUpdateList = localEndpointList
+				allEndpointUpdateList = append(localEndpointList, remoteEndpointList[:maxEndpoints-len(localEndpointList)]...)
+			}
+			// Check if there is any installed Endpoint which is not expected anymore. If internalTrafficPolicy and externalTrafficPolicy
+			// are both Local, only local Endpoints should be installed and checked; if internalTrafficPolicy or externalTrafficPolicy
+			// is Cluster, all Endpoints should be installed and checked.
+			for _, endpoint := range allEndpointUpdateList {
+				if internalNodeLocal && externalNodeLocal && endpoint.GetIsLocal() || !internalNodeLocal || !externalNodeLocal {
+					if _, ok := endpointsInstalled[endpoint.String()]; !ok { // There is an expected Endpoint which is not installed.
+						needUpdateEndpoints = true
+						break
+					}
 				}
-				endpointUpdateList = append(endpointUpdateList, endpoint)
 			}
 		} else {
 			if p.oversizeServiceSet.Has(svcPortName.String()) {
 				p.oversizeServiceSet.Delete(svcPortName.String())
 			}
-			for _, endpoint := range endpoints { // Check if there is any installed Endpoint which is not expected anymore.
-				if _, ok := endpointsInstalled[endpoint.String()]; !ok { // There is an expected Endpoint which is not installed.
-					needUpdateEndpoints = true
+			// Check if there is any installed Endpoint which is not expected anymore. If internalTrafficPolicy and externalTrafficPolicy
+			// are both Local, only local Endpoints should be installed and checked; if internalTrafficPolicy or externalTrafficPolicy
+			// is Cluster, all Endpoints should be installed and checked.
+			for _, endpoint := range endpoints {
+				if internalNodeLocal && externalNodeLocal && endpoint.GetIsLocal() || !internalNodeLocal || !externalNodeLocal {
+					if _, ok := endpointsInstalled[endpoint.String()]; !ok { // There is an expected Endpoint which is not installed.
+						needUpdateEndpoints = true
+					}
+				}
+				allEndpointUpdateList = append(allEndpointUpdateList, endpoint)
+				if endpoint.GetIsLocal() {
+					localEndpointUpdateList = append(localEndpointUpdateList, endpoint)
 				}
-				endpointUpdateList = append(endpointUpdateList, endpoint)
 			}
 		}
 
-		if len(endpoints) < len(endpointsInstalled) { // There are Endpoints which expired.
+		// If there are expired Endpoints, Endpoints installed should be updated.
+		if internalNodeLocal && externalNodeLocal && len(localEndpointUpdateList) < len(endpointsInstalled) ||
+			!(internalNodeLocal && externalNodeLocal) && len(allEndpointUpdateList) < len(endpointsInstalled) {
 			klog.V(2).Infof("Some Endpoints of Service %s removed, updating Endpoints", svcInfo.String())
 			needUpdateEndpoints = true
 		}
@@ -437,45 +477,64 @@ func (p *proxier) installServices() {
 		}
 
 		if needUpdateEndpoints {
+			var endpointUpdateList []k8sproxy.Endpoint
+			// If the type of the Service is NodePort or LoadBalancer and both internalTrafficPolicy and externalTrafficPolicy
+			// are Local, or the type of the Service is ClusterIP and internalTrafficPolicy is Local, then only local
+			// Endpoints should be installed, otherwise all Endpoints should be installed.
+			if internalNodeLocal && (externalNodeLocal || svcInfo.NodePort() == 0) {
+				endpointUpdateList = localEndpointUpdateList
+			} else {
+				endpointUpdateList = allEndpointUpdateList
+			}
+			// Install Endpoints.
 			err := p.ofClient.InstallEndpointFlows(svcInfo.OFProtocol, endpointUpdateList)
 			if err != nil {
 				klog.ErrorS(err, "Error when installing Endpoints flows")
 				continue
 			}
-			err = p.ofClient.InstallServiceGroup(groupID, svcInfo.StickyMaxAgeSeconds() != 0, endpointUpdateList)
-			if err != nil {
-				klog.ErrorS(err, "Error when installing Endpoints groups")
-				continue
-			}
-
-			if p.proxyAll {
-				if svcInfo.NodeLocalExternal() {
-					// Install another group when Service externalTrafficPolicy is Local.
-					groupIDLocal := p.groupCounter.AllocateIfNotExist(svcPortName, true)
-					var localEndpointList []k8sproxy.Endpoint
-					for _, ed := range endpointUpdateList {
-						if !ed.GetIsLocal() {
-							continue
-						}
-						localEndpointList = append(localEndpointList, ed)
+			if internalNodeLocal != externalNodeLocal {
+				if svcInfo.NodePort() > 0 {
+					// If the type of the Service is NodePort or LoadBalancer, when internalTrafficPolicy and externalTrafficPolicy
+					// of the Service are different, install two groups. One group has all Endpoints, the other has only
+					// local Endpoints.
+					groupID := p.groupCounter.AllocateIfNotExist(svcPortName, true)
+					if err = p.ofClient.InstallServiceGroup(groupID, svcInfo.StickyMaxAgeSeconds() != 0, localEndpointUpdateList); err != nil {
+						klog.ErrorS(err, "Error when installing Group of local Endpoints for Service", "Service", svcPortName)
+						continue
 					}
-					if err = p.ofClient.InstallServiceGroup(groupIDLocal, svcInfo.StickyMaxAgeSeconds() != 0, localEndpointList); err != nil {
-						klog.ErrorS(err, "Error when installing Group for Service whose externalTrafficPolicy is Local")
+					groupID = p.groupCounter.AllocateIfNotExist(svcPortName, false)
+					if err = p.ofClient.InstallServiceGroup(groupID, svcInfo.StickyMaxAgeSeconds() != 0, allEndpointUpdateList); err != nil {
+						klog.ErrorS(err, "Error when installing Group of all Endpoints for Service", "Service", svcPortName)
 						continue
 					}
 				} else {
-					// Uninstall the group with only local Endpoints when Service externalTrafficPolicy is Cluster
-					// unconditionally. If the group doesn't exist on OVS, then the return value will be nil; if the
-					// group exists on OVS, and after it is uninstalled successfully, then the return value will be also
-					// nil.
-					if groupIDLocal, exist := p.groupCounter.Get(svcPortName, true); exist {
-						if err := p.ofClient.UninstallServiceGroup(groupIDLocal); err != nil {
-							klog.ErrorS(err, "Failed to remove Group of local Endpoints for Service", "Service", svcPortName)
-							continue
-						}
-						p.groupCounter.Recycle(svcPortName, true)
+					// If the type of the Service is ClusterIP, install a group according to internalTrafficPolicy.
+					groupID := p.groupCounter.AllocateIfNotExist(svcPortName, internalNodeLocal)
+					if err = p.ofClient.InstallServiceGroup(groupID, svcInfo.StickyMaxAgeSeconds() != 0, endpointUpdateList); err != nil {
+						klog.ErrorS(err, "Error when installing Group of Endpoints for Service", "Service", svcPortName)
+						continue
 					}
 				}
+			} else {
+				// Regardless of the type of the Service, when internalTrafficPolicy and externalTrafficPolicy of the Service
+				// are the same, only install one group and unconditionally uninstall another group. If both internalTrafficPolicy
+				// and externalTrafficPolicy are Local, install the group that has only local Endpoints and unconditionally
+				// uninstall the group which has all Endpoints; if both internalTrafficPolicy and externalTrafficPolicy are
+				// Cluster, install the group which has all Endpoints and unconditionally uninstall the group which has
+				// only local Endpoints. Note that, if a group doesn't exist on OVS, then the return value will be nil.
+				nodeLocalVal := internalNodeLocal && externalNodeLocal
+				groupID := p.groupCounter.AllocateIfNotExist(svcPortName, nodeLocalVal)
+				if err = p.ofClient.InstallServiceGroup(groupID, svcInfo.StickyMaxAgeSeconds() != 0, endpointUpdateList); err != nil {
+					klog.ErrorS(err, "Error when installing Group of local Endpoints for Service", "Service", svcPortName)
+					continue
+				}
+				if groupID, exist := p.groupCounter.Get(svcPortName, !nodeLocalVal); exist {
+					if err := p.ofClient.UninstallServiceGroup(groupID); err != nil {
+						klog.ErrorS(err, "Failed to uninstall Group of all Endpoints for Service", "Service", svcPortName)
+						continue
+					}
+					p.groupCounter.Recycle(svcPortName, !nodeLocalVal)
+				}
 			}
 
 			for _, e := range endpointUpdateList {
@@ -515,20 +574,15 @@ func (p *proxier) installServices() {
 				}
 			}
 
-			// Install ClusterIP flows of current Service.
-			if err := p.ofClient.InstallServiceFlows(groupID, svcInfo.ClusterIP(), uint16(svcInfo.Port()), svcInfo.OFProtocol, uint16(svcInfo.StickyMaxAgeSeconds()), false, corev1.ServiceTypeClusterIP); err != nil {
+			// Install ClusterIP flows for the Service.
+			groupID := p.groupCounter.AllocateIfNotExist(svcPortName, internalNodeLocal)
+			if err := p.ofClient.InstallServiceFlows(groupID, svcInfo.ClusterIP(), uint16(svcInfo.Port()), svcInfo.OFProtocol, uint16(svcInfo.StickyMaxAgeSeconds()), svcInfo.NodeLocalInternal(), corev1.ServiceTypeClusterIP); err != nil {
 				klog.Errorf("Error when installing Service flows: %v", err)
 				continue
 			}
 
-			// If externalTrafficPolicy of the Service is Local, Service NodePort or LoadBalancer should use the Service
-			// group whose Endpoints are local.
-			nGroupID := groupID
-			if svcInfo.NodeLocalExternal() {
-				nGroupID = p.groupCounter.AllocateIfNotExist(svcPortName, true)
-			}
-
 			if p.proxyAll {
+				nGroupID := p.groupCounter.AllocateIfNotExist(svcPortName, externalNodeLocal)
 				// Install ClusterIP route on Node so that ClusterIP can be accessed on Node. Every time a new ClusterIP
 				// is created, the routing target IP block will be recalculated for expansion to be able to route the new
 				// created ClusterIP. Deleting a ClusterIP will not shrink the target routing IP block. The Service CIDR
@@ -549,6 +603,7 @@ func (p *proxier) installServices() {
 			}
 
 			if p.proxyLoadBalancerIPs {
+				nGroupID := p.groupCounter.AllocateIfNotExist(svcPortName, externalNodeLocal)
 				// Service LoadBalancer flows can be partially updated.
 				var toDelete, toAdd []string
 				if needRemoval {
diff --git a/pkg/features/antrea_features.go b/pkg/features/antrea_features.go
index 01cd6ecf5..6eb6f0adb 100644
--- a/pkg/features/antrea_features.go
+++ b/pkg/features/antrea_features.go
@@ -15,8 +15,10 @@
 package features
 
 import (
-	"k8s.io/apimachinery/pkg/util/runtime"
+	k8sruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/component-base/featuregate"
+
+	"antrea.io/antrea/pkg/util/runtime"
 )
 
 // When editing this file, make sure you edit the documentation as well to keep
@@ -139,7 +141,18 @@ var (
 )
 
 func init() {
-	runtime.Must(DefaultMutableFeatureGate.Add(DefaultAntreaFeatureGates))
+	if runtime.IsWindowsPlatform() {
+		for f := range unsupportedFeaturesOnWindows {
+			// A feature which is enabled by default on Linux might not be supported on
+			// Windows. So, override the default value here.
+			fg := DefaultAntreaFeatureGates[f]
+			if fg.Default {
+				fg.Default = false
+				DefaultAntreaFeatureGates[f] = fg
+			}
+		}
+	}
+	k8sruntime.Must(DefaultMutableFeatureGate.Add(DefaultAntreaFeatureGates))
 }
 
 // SupportedOnWindows checks whether a feature is supported on a Windows Node.


Test patch:
diff --git a/ci/jenkins/test-mc.sh b/ci/jenkins/test-mc.sh
index af4cf0310..dd1ace705 100755
--- a/ci/jenkins/test-mc.sh
+++ b/ci/jenkins/test-mc.sh
@@ -195,8 +195,14 @@ function deliver_antrea_multicluster {
 
     git show --numstat
     make clean
-    ${CLEAN_STALE_IMAGES}
-
+    # Clean up dangling images generated in previous builds.
+    docker image prune -f --filter "until=24h" || true > /dev/null
+
+    # Ensure that files in the Docker context have the correct permissions, or Docker caching cannot
+    # be leveraged successfully
+    chmod -R g-w build/images/ovs
+    chmod -R g-w build/images/base
+    
     cp -f build/yamls/*.yml $WORKDIR
     DOCKER_REGISTRY="${DOCKER_REGISTRY}" ./hack/build-antrea-linux-all.sh --pull
     echo "====== Delivering Antrea to all the Nodes ======"
diff --git a/multicluster/test/integration/stale_controller_test.go b/multicluster/test/integration/stale_controller_test.go
index 390adf485..ae3299e7a 100644
--- a/multicluster/test/integration/stale_controller_test.go
+++ b/multicluster/test/integration/stale_controller_test.go
@@ -41,14 +41,33 @@ var _ = Describe("Stale controller", func() {
 	ctx := context.Background()
 	It("Should clean up MC Service and ServiceImport if no corresponding ResourceImport in leader cluster", func() {
 		By("By claim a Service and ServiceImport without ResourceImport in leader cluster")
+		svcImpNoDelete := &k8smcsv1alpha1.ServiceImport{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: testNSForStale,
+				Name:      "nginxnodelete",
+			},
+			Spec: k8smcsv1alpha1.ServiceImportSpec{
+				Type: k8smcsv1alpha1.ClusterSetIP,
+				Ports: []k8smcsv1alpha1.ServicePort{
+					{
+						Name:     "http",
+						Protocol: corev1.ProtocolTCP,
+						Port:     80,
+					},
+				},
+			},
+		}
+
 		resImport := &mcsv1alpha1.ResourceImport{
 			ObjectMeta: metav1.ObjectMeta{
 				Name:      "resourceimportexist",
 				Namespace: LeaderNamespace,
 			},
 			Spec: mcsv1alpha1.ResourceImportSpec{
-				Name:      "nginxnodelete",
-				Namespace: testNSForStale,
+				Name:          "nginxnodelete",
+				Namespace:     testNSForStale,
+				Kind:          common.ServiceImportKind,
+				ServiceImport: svcImpNoDelete,
 			},
 		}
 		err := k8sClient.Create(ctx, resImport, &client.CreateOptions{})
@@ -106,28 +125,16 @@ var _ = Describe("Stale controller", func() {
 			Spec: svcSpec,
 		}
 
-		svcImpNoDelete := &k8smcsv1alpha1.ServiceImport{
-			ObjectMeta: metav1.ObjectMeta{
-				Namespace: testNSForStale,
-				Name:      "nginxnodelete",
-			},
-			Spec: k8smcsv1alpha1.ServiceImportSpec{
-				Type: k8smcsv1alpha1.ClusterSetIP,
-				Ports: []k8smcsv1alpha1.ServicePort{
-					{
-						Name:     "http",
-						Protocol: corev1.ProtocolTCP,
-						Port:     80,
-					},
-				},
-			},
-		}
-
 		Expect(k8sClient.Create(ctx, svcToDelete)).Should(Succeed())
 		Expect(k8sClient.Create(ctx, svcImpToDelete)).Should(Succeed())
 		Expect(k8sClient.Create(ctx, svcNoDelete)).Should(Succeed())
-		Expect(k8sClient.Create(ctx, mcSvcNoDelete)).Should(Succeed())
-		Expect(k8sClient.Create(ctx, svcImpNoDelete)).Should(Succeed())
+		// ResourceImport controller will watch ResourceImport creation event,
+		// it may create correspoding Service and ServiceImport already, so we
+		// skip it if it's 409 AlreadyExists error.
+		err = k8sClient.Create(ctx, mcSvcNoDelete)
+		Expect(err == nil || apierrors.IsAlreadyExists(err)).Should(BeTrue())
+		err = k8sClient.Create(ctx, svcImpNoDelete)
+		Expect(err == nil || apierrors.IsAlreadyExists(err)).Should(BeTrue())
 
 		Eventually(func() bool {
 			latestSvc := &corev1.Service{}
diff --git a/multicluster/test/integration/suite_test.go b/multicluster/test/integration/suite_test.go
index 8ae58eff6..fab46655e 100644
--- a/multicluster/test/integration/suite_test.go
+++ b/multicluster/test/integration/suite_test.go
@@ -37,10 +37,10 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/log/zap"
 	mcsscheme "sigs.k8s.io/mcs-api/pkg/client/clientset/versioned/scheme"
 
-	// mcsv1alpha1 "antrea.io/antrea/multicluster/apis/multicluster/v1alpha1"
 	mcsv1alpha1 "antrea.io/antrea/multicluster/apis/multicluster/v1alpha1"
 	multiclustercontrollers "antrea.io/antrea/multicluster/controllers/multicluster"
 	antreamcscheme "antrea.io/antrea/multicluster/pkg/client/clientset/versioned/scheme"
+	antreascheme "antrea.io/antrea/pkg/client/clientset/versioned/scheme"
 	"antrea.io/antrea/pkg/signals"
 	//+kubebuilder:scaffold:imports
 )
@@ -97,7 +97,9 @@ var _ = BeforeSuite(func() {
 	By("bootstrapping test environment")
 	useExistingCluster := true
 	testEnv = &envtest.Environment{
-		CRDDirectoryPaths:     []string{filepath.Join("..", "..", "config", "crd", "bases"), filepath.Join("..", "..", "config", "crd", "k8smcs")},
+		CRDDirectoryPaths: []string{filepath.Join("..", "..", "config", "crd", "bases"),
+			filepath.Join("..", "..", "config", "crd", "k8smcs"),
+			filepath.Join("..", "..", "..", "build", "yamls", "base", "crds.yml")},
 		ErrorIfCRDPathMissing: true,
 		UseExistingCluster:    &useExistingCluster,
 	}
@@ -114,6 +116,8 @@ var _ = BeforeSuite(func() {
 	Expect(err).NotTo(HaveOccurred())
 	err = k8sscheme.AddToScheme(scheme)
 	Expect(err).NotTo(HaveOccurred())
+	err = antreascheme.AddToScheme(scheme)
+	Expect(err).NotTo(HaveOccurred())
 	//+kubebuilder:scaffold:scheme
 
 	k8sClient, err = client.New(cfg, client.Options{Scheme: scheme})
diff --git a/pkg/agent/openflow/client_test.go b/pkg/agent/openflow/client_test.go
index 0dba46b77..ec11a49af 100644
--- a/pkg/agent/openflow/client_test.go
+++ b/pkg/agent/openflow/client_test.go
@@ -56,6 +56,8 @@ var (
 		NodeIPv4Addr:    nodeIP,
 	}
 	networkConfig = &config.NetworkConfig{IPv4Enabled: true}
+	egressConfig  = &config.EgressConfig{}
+	serviceConfig = &config.ServiceConfig{}
 )
 
 func installNodeFlows(ofClient Client, cacheKey string) (int, error) {
@@ -111,6 +113,8 @@ func TestIdempotentFlowInstallation(t *testing.T) {
 			client.ofEntryOperations = m
 			client.nodeConfig = nodeConfig
 			client.networkConfig = networkConfig
+			client.egressConfig = egressConfig
+			client.serviceConfig = serviceConfig
 			client.ipProtocols = []binding.Protocol{binding.ProtocolIP}
 			client.generatePipelines()
 
@@ -142,6 +146,8 @@ func TestIdempotentFlowInstallation(t *testing.T) {
 			client.ofEntryOperations = m
 			client.nodeConfig = nodeConfig
 			client.networkConfig = networkConfig
+			client.egressConfig = egressConfig
+			client.serviceConfig = serviceConfig
 			client.ipProtocols = []binding.Protocol{binding.ProtocolIP}
 			client.generatePipelines()
 
@@ -186,6 +192,8 @@ func TestFlowInstallationFailed(t *testing.T) {
 			client.ofEntryOperations = m
 			client.nodeConfig = nodeConfig
 			client.networkConfig = networkConfig
+			client.egressConfig = egressConfig
+			client.serviceConfig = serviceConfig
 			client.ipProtocols = []binding.Protocol{binding.ProtocolIP}
 			client.generatePipelines()
 
@@ -223,6 +231,8 @@ func TestConcurrentFlowInstallation(t *testing.T) {
 			client.ofEntryOperations = m
 			client.nodeConfig = nodeConfig
 			client.networkConfig = networkConfig
+			client.egressConfig = egressConfig
+			client.serviceConfig = serviceConfig
 			client.ipProtocols = []binding.Protocol{binding.ProtocolIP}
 			client.generatePipelines()
 
@@ -415,6 +425,8 @@ func prepareTraceflowFlow(ctrl *gomock.Controller) *client {
 	c.ofEntryOperations = m
 	c.nodeConfig = nodeConfig
 	c.networkConfig = networkConfig
+	c.egressConfig = egressConfig
+	c.serviceConfig = serviceConfig
 	c.ipProtocols = []binding.Protocol{binding.ProtocolIP}
 	c.generatePipelines()
 
diff --git a/pkg/agent/openflow/testing/mock_openflow.go b/pkg/agent/openflow/testing/mock_openflow.go
index a03e5047b..740cad169 100644
--- a/pkg/agent/openflow/testing/mock_openflow.go
+++ b/pkg/agent/openflow/testing/mock_openflow.go
@@ -253,74 +253,18 @@ func (mr *MockClientMockRecorder) InitialTLVMap() *gomock.Call {
 }
 
 // Initialize mocks base method
-func (m *MockClient) Initialize(arg0 types.RoundInfo, arg1 *config.NodeConfig, arg2 *config.NetworkConfig) (<-chan struct{}, error) {
+func (m *MockClient) Initialize(arg0 types.RoundInfo, arg1 *config.NodeConfig, arg2 *config.NetworkConfig, arg3 *config.EgressConfig, arg4 *config.ServiceConfig) (<-chan struct{}, error) {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "Initialize", arg0, arg1, arg2)
+	ret := m.ctrl.Call(m, "Initialize", arg0, arg1, arg2, arg3, arg4)
 	ret0, _ := ret[0].(<-chan struct{})
 	ret1, _ := ret[1].(error)
 	return ret0, ret1
 }
 
 // Initialize indicates an expected call of Initialize
-func (mr *MockClientMockRecorder) Initialize(arg0, arg1, arg2 interface{}) *gomock.Call {
+func (mr *MockClientMockRecorder) Initialize(arg0, arg1, arg2, arg3, arg4 interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Initialize", reflect.TypeOf((*MockClient)(nil).Initialize), arg0, arg1, arg2)
-}
-
-// InstallBridgeUplinkFlows mocks base method
-func (m *MockClient) InstallBridgeUplinkFlows() error {
-	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "InstallBridgeUplinkFlows")
-	ret0, _ := ret[0].(error)
-	return ret0
-}
-
-// InstallBridgeUplinkFlows indicates an expected call of InstallBridgeUplinkFlows
-func (mr *MockClientMockRecorder) InstallBridgeUplinkFlows() *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "InstallBridgeUplinkFlows", reflect.TypeOf((*MockClient)(nil).InstallBridgeUplinkFlows))
-}
-
-// InstallClusterServiceCIDRFlows mocks base method
-func (m *MockClient) InstallClusterServiceCIDRFlows(arg0 []*net.IPNet) error {
-	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "InstallClusterServiceCIDRFlows", arg0)
-	ret0, _ := ret[0].(error)
-	return ret0
-}
-
-// InstallClusterServiceCIDRFlows indicates an expected call of InstallClusterServiceCIDRFlows
-func (mr *MockClientMockRecorder) InstallClusterServiceCIDRFlows(arg0 interface{}) *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "InstallClusterServiceCIDRFlows", reflect.TypeOf((*MockClient)(nil).InstallClusterServiceCIDRFlows), arg0)
-}
-
-// InstallDefaultServiceFlows mocks base method
-func (m *MockClient) InstallDefaultServiceFlows(arg0, arg1 []net.IP) error {
-	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "InstallDefaultServiceFlows", arg0, arg1)
-	ret0, _ := ret[0].(error)
-	return ret0
-}
-
-// InstallDefaultServiceFlows indicates an expected call of InstallDefaultServiceFlows
-func (mr *MockClientMockRecorder) InstallDefaultServiceFlows(arg0, arg1 interface{}) *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "InstallDefaultServiceFlows", reflect.TypeOf((*MockClient)(nil).InstallDefaultServiceFlows), arg0, arg1)
-}
-
-// InstallDefaultTunnelFlows mocks base method
-func (m *MockClient) InstallDefaultTunnelFlows() error {
-	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "InstallDefaultTunnelFlows")
-	ret0, _ := ret[0].(error)
-	return ret0
-}
-
-// InstallDefaultTunnelFlows indicates an expected call of InstallDefaultTunnelFlows
-func (mr *MockClientMockRecorder) InstallDefaultTunnelFlows() *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "InstallDefaultTunnelFlows", reflect.TypeOf((*MockClient)(nil).InstallDefaultTunnelFlows))
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Initialize", reflect.TypeOf((*MockClient)(nil).Initialize), arg0, arg1, arg2, arg3, arg4)
 }
 
 // InstallEndpointFlows mocks base method
@@ -337,34 +281,6 @@ func (mr *MockClientMockRecorder) InstallEndpointFlows(arg0, arg1 interface{}) *
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "InstallEndpointFlows", reflect.TypeOf((*MockClient)(nil).InstallEndpointFlows), arg0, arg1)
 }
 
-// InstallExternalFlows mocks base method
-func (m *MockClient) InstallExternalFlows(arg0 []net.IPNet) error {
-	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "InstallExternalFlows", arg0)
-	ret0, _ := ret[0].(error)
-	return ret0
-}
-
-// InstallExternalFlows indicates an expected call of InstallExternalFlows
-func (mr *MockClientMockRecorder) InstallExternalFlows(arg0 interface{}) *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "InstallExternalFlows", reflect.TypeOf((*MockClient)(nil).InstallExternalFlows), arg0)
-}
-
-// InstallGatewayFlows mocks base method
-func (m *MockClient) InstallGatewayFlows() error {
-	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "InstallGatewayFlows")
-	ret0, _ := ret[0].(error)
-	return ret0
-}
-
-// InstallGatewayFlows indicates an expected call of InstallGatewayFlows
-func (mr *MockClientMockRecorder) InstallGatewayFlows() *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "InstallGatewayFlows", reflect.TypeOf((*MockClient)(nil).InstallGatewayFlows))
-}
-
 // InstallMulticastFlow mocks base method
 func (m *MockClient) InstallMulticastFlow(arg0 net.IP) error {
 	m.ctrl.T.Helper()
diff --git a/pkg/agent/proxy/proxier_test.go b/pkg/agent/proxy/proxier_test.go
index f5685e3cb..24a2b5884 100644
--- a/pkg/agent/proxy/proxier_test.go
+++ b/pkg/agent/proxy/proxier_test.go
@@ -51,6 +51,7 @@ var (
 	loadBalancerIPv6 = net.ParseIP("fec0::169:254:169:1")
 	svcNodePortIPv4  = net.ParseIP("192.168.77.100")
 	svcNodePortIPv6  = net.ParseIP("2001::192:168:77:100")
+	hostname         = "localhost"
 
 	nodePortAddressesIPv4 = []net.IP{svcNodePortIPv4}
 	nodePortAddressesIPv6 = []net.IP{svcNodePortIPv6}
@@ -157,7 +158,7 @@ func NewFakeProxier(routeClient route.Interface, ofClient openflow.Client, nodeP
 	return p
 }
 
-func testClusterIP(t *testing.T, svcIP net.IP, epIP net.IP, isIPv6 bool, extraSvcs []*corev1.Service, extraEps []*corev1.Endpoints) {
+func testClusterIP(t *testing.T, svcIP net.IP, ep1IP, ep2IP net.IP, isIPv6, nodeLocalInternal bool, extraSvcs []*corev1.Service, extraEps []*corev1.Endpoints) {
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()
 	mockOFClient := ofmock.NewMockClient(ctrl)
@@ -171,45 +172,69 @@ func testClusterIP(t *testing.T, svcIP net.IP, epIP net.IP, isIPv6 bool, extraSv
 		Port:           "80",
 		Protocol:       corev1.ProtocolTCP,
 	}
+	internalTrafficPolicy := corev1.ServiceInternalTrafficPolicyCluster
+	if nodeLocalInternal {
+		internalTrafficPolicy = corev1.ServiceInternalTrafficPolicyLocal
+	}
 
-	allServices := append(extraSvcs, makeTestService(svcPortName.Namespace, svcPortName.Name, func(svc *corev1.Service) {
-		svc.Spec.ClusterIP = svcIP.String()
-		svc.Spec.Ports = []corev1.ServicePort{{
-			Name:     svcPortName.Port,
-			Port:     int32(svcPort),
-			Protocol: corev1.ProtocolTCP,
-		}}
-	}))
-	makeServiceMap(fp, allServices...)
-
-	allEps := append(extraEps, makeTestEndpoints(svcPortName.Namespace, svcPortName.Name, func(ept *corev1.Endpoints) {
-		ept.Subsets = []corev1.EndpointSubset{{
-			Addresses: []corev1.EndpointAddress{{
-				IP: epIP.String(),
-			}},
-			Ports: []corev1.EndpointPort{{
+	allServices := append(extraSvcs,
+		makeTestService(svcPortName.Namespace, svcPortName.Name, func(svc *corev1.Service) {
+			svc.Spec.ClusterIP = svcIP.String()
+			svc.Spec.Ports = []corev1.ServicePort{{
 				Name:     svcPortName.Port,
 				Port:     int32(svcPort),
 				Protocol: corev1.ProtocolTCP,
-			}},
-		}}
-	}))
+			}}
+			svc.Spec.InternalTrafficPolicy = &internalTrafficPolicy
+		}))
+	makeServiceMap(fp, allServices...)
+
+	remoteEndpoint := corev1.EndpointSubset{
+		Addresses: []corev1.EndpointAddress{{
+			IP: ep1IP.String(),
+		}},
+		Ports: []corev1.EndpointPort{{
+			Name:     svcPortName.Port,
+			Port:     int32(svcPort),
+			Protocol: corev1.ProtocolTCP,
+		}},
+	}
+	localEndpoint := corev1.EndpointSubset{
+		Addresses: []corev1.EndpointAddress{{
+			IP:       ep2IP.String(),
+			NodeName: &hostname,
+		}},
+		Ports: []corev1.EndpointPort{{
+			Name:     svcPortName.Port,
+			Port:     int32(svcPort),
+			Protocol: corev1.ProtocolTCP,
+		}},
+	}
+	epFunc := func(ept *corev1.Endpoints) { ept.Subsets = []corev1.EndpointSubset{localEndpoint, remoteEndpoint} }
+	allEps := append(extraEps, makeTestEndpoints(svcPortName.Namespace, svcPortName.Name, epFunc))
 	makeEndpointsMap(fp, allEps...)
 
-	groupID := fp.groupCounter.AllocateIfNotExist(svcPortName, false)
-	mockOFClient.EXPECT().InstallServiceGroup(groupID, false, gomock.Any()).Times(1)
+	expectedLocalEps := []k8sproxy.Endpoint{k8sproxy.NewBaseEndpointInfo(ep2IP.String(), svcPort, true, nil)}
+	expectedAllEps := expectedLocalEps
+	if !nodeLocalInternal {
+		expectedAllEps = append(expectedAllEps, k8sproxy.NewBaseEndpointInfo(ep1IP.String(), svcPort, false, nil))
+	}
+
 	bindingProtocol := binding.ProtocolTCP
 	if isIPv6 {
 		bindingProtocol = binding.ProtocolTCPv6
 	}
-	mockOFClient.EXPECT().InstallEndpointFlows(bindingProtocol, gomock.Any()).Times(1)
-	mockOFClient.EXPECT().InstallServiceFlows(groupID, svcIP, uint16(svcPort), bindingProtocol, uint16(0), false, corev1.ServiceTypeClusterIP).Times(1)
+
+	groupID := fp.groupCounter.AllocateIfNotExist(svcPortName, nodeLocalInternal)
+	mockOFClient.EXPECT().InstallEndpointFlows(bindingProtocol, gomock.InAnyOrder(expectedAllEps)).Times(1)
+	mockOFClient.EXPECT().InstallServiceGroup(groupID, false, gomock.InAnyOrder(expectedAllEps)).Times(1)
+	mockOFClient.EXPECT().InstallServiceFlows(groupID, svcIP, uint16(svcPort), bindingProtocol, uint16(0), nodeLocalInternal, corev1.ServiceTypeClusterIP).Times(1)
 	mockRouteClient.EXPECT().AddClusterIPRoute(svcIP).Times(1)
 
 	fp.syncProxyRules()
 }
 
-func testLoadBalancer(t *testing.T, nodePortAddresses []net.IP, svcIP, ep1IP, ep2IP, loadBalancerIP net.IP, isIPv6, nodeLocalExternal, proxyLoadBalancerIPs bool) {
+func testLoadBalancer(t *testing.T, nodePortAddresses []net.IP, svcIP, ep1IP, ep2IP, loadBalancerIP net.IP, isIPv6, nodeLocalInternal, nodeLocalExternal, proxyLoadBalancerIPs bool) {
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()
 	mockOFClient := ofmock.NewMockClient(ctrl)
@@ -232,6 +257,10 @@ func testLoadBalancer(t *testing.T, nodePortAddresses []net.IP, svcIP, ep1IP, ep
 	if nodeLocalExternal {
 		externalTrafficPolicy = corev1.ServiceExternalTrafficPolicyTypeLocal
 	}
+	internalTrafficPolicy := corev1.ServiceInternalTrafficPolicyCluster
+	if nodeLocalInternal {
+		internalTrafficPolicy = corev1.ServiceInternalTrafficPolicyLocal
+	}
 
 	makeServiceMap(fp,
 		makeTestService(svcPortName.Namespace, svcPortName.Name, func(svc *corev1.Service) {
@@ -247,57 +276,78 @@ func testLoadBalancer(t *testing.T, nodePortAddresses []net.IP, svcIP, ep1IP, ep
 				Protocol: corev1.ProtocolTCP,
 			}}
 			svc.Spec.ExternalTrafficPolicy = externalTrafficPolicy
+			svc.Spec.InternalTrafficPolicy = &internalTrafficPolicy
 		}),
 	)
 
-	var eps []*corev1.Endpoints
-	epFunc := func(ept *corev1.Endpoints) {
-		ept.Subsets = []corev1.EndpointSubset{{
-			Addresses: []corev1.EndpointAddress{{
-				IP:       ep1IP.String(),
-				Hostname: "localhost",
-			}},
-			Ports: []corev1.EndpointPort{{
-				Name:     svcPortName.Port,
-				Port:     int32(svcPort),
-				Protocol: corev1.ProtocolTCP,
-			}},
-		}}
+	remoteEndpoint := corev1.EndpointSubset{
+		Addresses: []corev1.EndpointAddress{{
+			IP: ep1IP.String(),
+		}},
+		Ports: []corev1.EndpointPort{{
+			Name:     svcPortName.Port,
+			Port:     int32(svcPort),
+			Protocol: corev1.ProtocolTCP,
+		}},
 	}
-	eps = append(eps, makeTestEndpoints(svcPortName.Namespace, svcPortName.Name, epFunc))
-	if nodeLocalExternal {
-		epFunc = func(ept *corev1.Endpoints) {
-			ept.Subsets = []corev1.EndpointSubset{{
-				Addresses: []corev1.EndpointAddress{{
-					IP:       ep2IP.String(),
-					Hostname: "remote",
-				}},
-				Ports: []corev1.EndpointPort{{
-					Name:     svcPortName.Port,
-					Port:     int32(svcPort),
-					Protocol: corev1.ProtocolTCP,
-				}},
-			}}
-		}
-		eps = append(eps, makeTestEndpoints(svcPortName.Namespace, svcPortName.Name, epFunc))
+	localEndpoint := corev1.EndpointSubset{
+		Addresses: []corev1.EndpointAddress{{
+			IP:       ep2IP.String(),
+			NodeName: &hostname,
+		}},
+		Ports: []corev1.EndpointPort{{
+			Name:     svcPortName.Port,
+			Port:     int32(svcPort),
+			Protocol: corev1.ProtocolTCP,
+		}},
 	}
+
+	epFunc := func(ept *corev1.Endpoints) { ept.Subsets = []corev1.EndpointSubset{localEndpoint, remoteEndpoint} }
+	eps := []*corev1.Endpoints{makeTestEndpoints(svcPortName.Namespace, svcPortName.Name, epFunc)}
 	makeEndpointsMap(fp, eps...)
-	groupID := fp.groupCounter.AllocateIfNotExist(svcPortName, false)
+
+	expectedLocalEps := []k8sproxy.Endpoint{k8sproxy.NewBaseEndpointInfo(ep2IP.String(), svcPort, true, nil)}
+	expectedAllEps := expectedLocalEps
+	if !(nodeLocalInternal && nodeLocalExternal) {
+		expectedAllEps = append(expectedAllEps, k8sproxy.NewBaseEndpointInfo(ep1IP.String(), svcPort, false, nil))
+	}
+
 	bindingProtocol := binding.ProtocolTCP
 	if isIPv6 {
 		bindingProtocol = binding.ProtocolTCPv6
 	}
-	mockOFClient.EXPECT().InstallServiceGroup(groupID, false, gomock.Any()).Times(1)
-	mockOFClient.EXPECT().InstallEndpointFlows(bindingProtocol, gomock.Any()).Times(1)
-	mockOFClient.EXPECT().InstallServiceFlows(groupID, svcIP, uint16(svcPort), bindingProtocol, uint16(0), false, corev1.ServiceTypeClusterIP).Times(1)
-	if nodeLocalExternal {
-		groupID = fp.groupCounter.AllocateIfNotExist(svcPortName, true)
-		mockOFClient.EXPECT().InstallServiceGroup(groupID, false, gomock.Any()).Times(1)
-	}
-	if proxyLoadBalancerIPs {
-		mockOFClient.EXPECT().InstallServiceFlows(groupID, loadBalancerIP, uint16(svcPort), bindingProtocol, uint16(0), nodeLocalExternal, corev1.ServiceTypeLoadBalancer).Times(1)
+
+	mockOFClient.EXPECT().InstallEndpointFlows(bindingProtocol, gomock.InAnyOrder(expectedAllEps)).Times(1)
+	if nodeLocalInternal != nodeLocalExternal {
+		var clusterIPEps, nodePortEps []k8sproxy.Endpoint
+		if nodeLocalInternal {
+			clusterIPEps = expectedLocalEps
+			nodePortEps = expectedAllEps
+		} else {
+			clusterIPEps = expectedAllEps
+			nodePortEps = expectedLocalEps
+		}
+		groupID := fp.groupCounter.AllocateIfNotExist(svcPortName, nodeLocalInternal)
+		mockOFClient.EXPECT().InstallServiceGroup(groupID, false, gomock.InAnyOrder(clusterIPEps)).Times(1)
+		mockOFClient.EXPECT().InstallServiceFlows(groupID, svcIP, uint16(svcPort), bindingProtocol, uint16(0), nodeLocalInternal, corev1.ServiceTypeClusterIP).Times(1)
+		groupID = fp.groupCounter.AllocateIfNotExist(svcPortName, nodeLocalExternal)
+		mockOFClient.EXPECT().InstallServiceGroup(groupID, false, gomock.InAnyOrder(nodePortEps)).Times(1)
+		mockOFClient.EXPECT().InstallServiceFlows(groupID, gomock.Any(), uint16(svcNodePort), bindingProtocol, uint16(0), nodeLocalExternal, corev1.ServiceTypeNodePort).Times(1)
+		if proxyLoadBalancerIPs {
+			mockOFClient.EXPECT().InstallServiceFlows(groupID, loadBalancerIP, uint16(svcPort), bindingProtocol, uint16(0), nodeLocalExternal, corev1.ServiceTypeLoadBalancer).Times(1)
+		}
+	} else {
+		nodeLocalVal := nodeLocalInternal && nodeLocalExternal
+		groupID := fp.groupCounter.AllocateIfNotExist(svcPortName, nodeLocalVal)
+		mockOFClient.EXPECT().InstallServiceGroup(groupID, false, gomock.InAnyOrder(expectedAllEps)).Times(1)
+		mockOFClient.EXPECT().InstallServiceFlows(groupID, svcIP, uint16(svcPort), bindingProtocol, uint16(0), nodeLocalVal, corev1.ServiceTypeClusterIP).Times(1)
+		mockOFClient.EXPECT().InstallServiceFlows(groupID, gomock.Any(), uint16(svcNodePort), bindingProtocol, uint16(0), nodeLocalVal, corev1.ServiceTypeNodePort).Times(1)
+		if proxyLoadBalancerIPs {
+			mockOFClient.EXPECT().InstallServiceFlows(groupID, loadBalancerIP, uint16(svcPort), bindingProtocol, uint16(0), nodeLocalVal, corev1.ServiceTypeLoadBalancer).Times(1)
+		}
+		groupID = fp.groupCounter.AllocateIfNotExist(svcPortName, !nodeLocalVal)
+		mockOFClient.EXPECT().UninstallServiceGroup(groupID).Times(1)
 	}
-	mockOFClient.EXPECT().InstallServiceFlows(groupID, gomock.Any(), uint16(svcNodePort), bindingProtocol, uint16(0), nodeLocalExternal, corev1.ServiceTypeNodePort).Times(1)
 	mockRouteClient.EXPECT().AddClusterIPRoute(svcIP).Times(1)
 	if proxyLoadBalancerIPs {
 		mockRouteClient.EXPECT().AddLoadBalancer([]string{loadBalancerIP.String()}).Times(1)
@@ -307,7 +357,7 @@ func testLoadBalancer(t *testing.T, nodePortAddresses []net.IP, svcIP, ep1IP, ep
 	fp.syncProxyRules()
 }
 
-func testNodePort(t *testing.T, nodePortAddresses []net.IP, svcIP, ep1IP, ep2IP net.IP, isIPv6, nodeLocalExternal bool) {
+func testNodePort(t *testing.T, nodePortAddresses []net.IP, svcIP, ep1IP, ep2IP net.IP, isIPv6, nodeLocalInternal, nodeLocalExternal bool) {
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()
 	mockOFClient := ofmock.NewMockClient(ctrl)
@@ -326,6 +376,10 @@ func testNodePort(t *testing.T, nodePortAddresses []net.IP, svcIP, ep1IP, ep2IP
 	if nodeLocalExternal {
 		externalTrafficPolicy = corev1.ServiceExternalTrafficPolicyTypeLocal
 	}
+	internalTrafficPolicy := corev1.ServiceInternalTrafficPolicyCluster
+	if nodeLocalInternal {
+		internalTrafficPolicy = corev1.ServiceInternalTrafficPolicyLocal
+	}
 
 	makeServiceMap(fp,
 		makeTestService(svcPortName.Namespace, svcPortName.Name, func(svc *corev1.Service) {
@@ -338,104 +392,172 @@ func testNodePort(t *testing.T, nodePortAddresses []net.IP, svcIP, ep1IP, ep2IP
 				Protocol: corev1.ProtocolTCP,
 			}}
 			svc.Spec.ExternalTrafficPolicy = externalTrafficPolicy
+			svc.Spec.InternalTrafficPolicy = &internalTrafficPolicy
 		}),
 	)
 
+	remoteEndpoint := corev1.EndpointSubset{
+		Addresses: []corev1.EndpointAddress{{
+			IP: ep1IP.String(),
+		}},
+		Ports: []corev1.EndpointPort{{
+			Name:     svcPortName.Port,
+			Port:     int32(svcPort),
+			Protocol: corev1.ProtocolTCP,
+		}},
+	}
+	localEndpoint := corev1.EndpointSubset{
+		Addresses: []corev1.EndpointAddress{{
+			IP:       ep2IP.String(),
+			NodeName: &hostname,
+		}},
+		Ports: []corev1.EndpointPort{{
+			Name:     svcPortName.Port,
+			Port:     int32(svcPort),
+			Protocol: corev1.ProtocolTCP,
+		}},
+	}
+
 	var eps []*corev1.Endpoints
 	epFunc := func(ept *corev1.Endpoints) {
-		ept.Subsets = []corev1.EndpointSubset{{
-			Addresses: []corev1.EndpointAddress{{
-				IP:       ep1IP.String(),
-				Hostname: "localhost",
-			}},
-			Ports: []corev1.EndpointPort{{
-				Name:     svcPortName.Port,
-				Port:     int32(svcPort),
-				Protocol: corev1.ProtocolTCP,
-			}},
-		}}
-	}
-	eps = append(eps, makeTestEndpoints(svcPortName.Namespace, svcPortName.Name, epFunc))
-	if nodeLocalExternal {
-		epFunc = func(ept *corev1.Endpoints) {
-			ept.Subsets = []corev1.EndpointSubset{{
-				Addresses: []corev1.EndpointAddress{{
-					IP:       ep2IP.String(),
-					Hostname: "remote",
-				}},
-				Ports: []corev1.EndpointPort{{
-					Name:     svcPortName.Port,
-					Port:     int32(svcPort),
-					Protocol: corev1.ProtocolTCP,
-				}},
-			}}
+		if nodeLocalInternal && nodeLocalExternal {
+			ept.Subsets = []corev1.EndpointSubset{localEndpoint}
+		} else {
+			ept.Subsets = []corev1.EndpointSubset{localEndpoint, remoteEndpoint}
 		}
-		eps = append(eps, makeTestEndpoints(svcPortName.Namespace, svcPortName.Name, epFunc))
 	}
+	eps = append(eps, makeTestEndpoints(svcPortName.Namespace, svcPortName.Name, epFunc))
 	makeEndpointsMap(fp, eps...)
 
-	groupID := fp.groupCounter.AllocateIfNotExist(svcPortName, false)
+	expectedLocalEps := []k8sproxy.Endpoint{k8sproxy.NewBaseEndpointInfo(ep2IP.String(), svcPort, true, nil)}
+	expectedAllEps := expectedLocalEps
+	if !(nodeLocalInternal && nodeLocalExternal) {
+		expectedAllEps = append(expectedAllEps, k8sproxy.NewBaseEndpointInfo(ep1IP.String(), svcPort, false, nil))
+	}
+
 	bindingProtocol := binding.ProtocolTCP
 	if isIPv6 {
 		bindingProtocol = binding.ProtocolTCPv6
 	}
-	mockOFClient.EXPECT().InstallServiceGroup(groupID, false, gomock.Any()).Times(1)
-	mockOFClient.EXPECT().InstallEndpointFlows(bindingProtocol, gomock.Any()).Times(1)
-	mockOFClient.EXPECT().InstallServiceFlows(groupID, svcIP, uint16(svcPort), bindingProtocol, uint16(0), false, corev1.ServiceTypeClusterIP).Times(1)
-	if nodeLocalExternal {
-		groupID = fp.groupCounter.AllocateIfNotExist(svcPortName, true)
-		mockOFClient.EXPECT().InstallServiceGroup(groupID, false, gomock.Any()).Times(1)
-	}
 
-	mockOFClient.EXPECT().InstallServiceFlows(groupID, gomock.Any(), uint16(svcNodePort), bindingProtocol, uint16(0), nodeLocalExternal, corev1.ServiceTypeNodePort).Times(1)
+	mockOFClient.EXPECT().InstallEndpointFlows(bindingProtocol, gomock.InAnyOrder(expectedAllEps)).Times(1)
+	if nodeLocalInternal != nodeLocalExternal {
+		var clusterIPEps, nodePortEps []k8sproxy.Endpoint
+		if nodeLocalInternal {
+			clusterIPEps = expectedLocalEps
+			nodePortEps = expectedAllEps
+		} else {
+			clusterIPEps = expectedAllEps
+			nodePortEps = expectedLocalEps
+		}
+		groupID := fp.groupCounter.AllocateIfNotExist(svcPortName, nodeLocalInternal)
+		mockOFClient.EXPECT().InstallServiceGroup(groupID, false, gomock.InAnyOrder(clusterIPEps)).Times(1)
+		mockOFClient.EXPECT().InstallServiceFlows(groupID, svcIP, uint16(svcPort), bindingProtocol, uint16(0), nodeLocalInternal, corev1.ServiceTypeClusterIP).Times(1)
+
+		groupID = fp.groupCounter.AllocateIfNotExist(svcPortName, nodeLocalExternal)
+		mockOFClient.EXPECT().InstallServiceGroup(groupID, false, gomock.InAnyOrder(nodePortEps)).Times(1)
+		mockOFClient.EXPECT().InstallServiceFlows(groupID, gomock.Any(), uint16(svcNodePort), bindingProtocol, uint16(0), nodeLocalExternal, corev1.ServiceTypeNodePort).Times(1)
+	} else {
+		nodeLocalVal := nodeLocalInternal && nodeLocalExternal
+		groupID := fp.groupCounter.AllocateIfNotExist(svcPortName, nodeLocalVal)
+		mockOFClient.EXPECT().InstallServiceGroup(groupID, false, gomock.InAnyOrder(expectedAllEps)).Times(1)
+		mockOFClient.EXPECT().InstallServiceFlows(groupID, svcIP, uint16(svcPort), bindingProtocol, uint16(0), nodeLocalVal, corev1.ServiceTypeClusterIP).Times(1)
+		mockOFClient.EXPECT().InstallServiceFlows(groupID, gomock.Any(), uint16(svcNodePort), bindingProtocol, uint16(0), nodeLocalVal, corev1.ServiceTypeNodePort).Times(1)
+
+		groupID = fp.groupCounter.AllocateIfNotExist(svcPortName, !nodeLocalVal)
+		mockOFClient.EXPECT().UninstallServiceGroup(groupID).Times(1)
+	}
 	mockRouteClient.EXPECT().AddClusterIPRoute(svcIP).Times(1)
 	mockRouteClient.EXPECT().AddNodePort(gomock.Any(), uint16(svcNodePort), bindingProtocol).Times(1)
 
 	fp.syncProxyRules()
 }
 
-func TestLoadBalancerIPv4(t *testing.T) {
-	testLoadBalancer(t, nodePortAddressesIPv4, svcIPv4, ep1IPv4, nil, loadBalancerIPv4, false, false, true)
-}
-
-func TestLoadBalancerIPv4ExternalLocal(t *testing.T) {
-	testLoadBalancer(t, nodePortAddressesIPv4, svcIPv4, ep1IPv4, ep2IPv4, loadBalancerIPv4, false, true, true)
-}
-
-func TestLoadBalancerIPv6(t *testing.T) {
-	testLoadBalancer(t, nodePortAddressesIPv6, svcIPv6, ep1IPv6, nil, loadBalancerIPv6, true, false, true)
-}
-
-func TestLoadBalancerIPv6ExternalLocal(t *testing.T) {
-	testLoadBalancer(t, nodePortAddressesIPv6, svcIPv6, ep1IPv6, ep2IPv6, loadBalancerIPv6, true, true, true)
-}
-
-func TestLoadBalancerIPv4NoExternalIPs(t *testing.T) {
-	testLoadBalancer(t, nodePortAddressesIPv4, svcIPv4, ep1IPv4, nil, loadBalancerIPv4, false, false, false)
-}
-
-func TestNodePortIPv4(t *testing.T) {
-	testNodePort(t, nodePortAddressesIPv4, svcIPv4, ep1IPv4, nil, false, false)
-}
-
-func TestNodePortIPv4ExternalLocal(t *testing.T) {
-	testNodePort(t, nodePortAddressesIPv4, svcIPv4, ep1IPv4, ep2IPv4, false, true)
-}
-
-func TestNodePortIPv6(t *testing.T) {
-	testNodePort(t, nodePortAddressesIPv6, svcIPv6, ep1IPv6, nil, true, false)
-}
-
-func TestNodePortIPv6ExternalLocal(t *testing.T) {
-	testNodePort(t, nodePortAddressesIPv6, svcIPv6, ep1IPv6, ep2IPv6, true, true)
+func TestCluster(t *testing.T) {
+	t.Run("IPv4", func(t *testing.T) {
+		t.Run("InternalTrafficPolicy Cluster", func(t *testing.T) {
+			testClusterIP(t, svcIPv4, ep1IPv4, ep2IPv4, false, false, []*corev1.Service{}, []*corev1.Endpoints{})
+		})
+		t.Run("InternalTrafficPolicy Local", func(t *testing.T) {
+			testClusterIP(t, svcIPv4, ep1IPv4, ep2IPv4, false, true, []*corev1.Service{}, []*corev1.Endpoints{})
+		})
+	})
+	t.Run("IPv6", func(t *testing.T) {
+		t.Run("InternalTrafficPolicy Cluster", func(t *testing.T) {
+			testClusterIP(t, svcIPv6, ep1IPv6, ep2IPv6, true, false, []*corev1.Service{}, []*corev1.Endpoints{})
+		})
+		t.Run("InternalTrafficPolicy Local", func(t *testing.T) {
+			testClusterIP(t, svcIPv6, ep1IPv6, ep2IPv6, true, true, []*corev1.Service{}, []*corev1.Endpoints{})
+		})
+	})
 }
 
-func TestClusterIPv4(t *testing.T) {
-	testClusterIP(t, svcIPv4, ep1IPv4, false, []*corev1.Service{}, []*corev1.Endpoints{})
+func TestLoadBalancer(t *testing.T) {
+	t.Run("IPv4", func(t *testing.T) {
+		t.Run("InternalTrafficPolicy:Cluster ExternalTrafficPolicy:Cluster", func(t *testing.T) {
+			testLoadBalancer(t, nodePortAddressesIPv4, svcIPv4, ep1IPv4, ep2IPv4, loadBalancerIPv4, false, false, false, true)
+		})
+		t.Run("InternalTrafficPolicy:Cluster ExternalTrafficPolicy:Local", func(t *testing.T) {
+			testLoadBalancer(t, nodePortAddressesIPv4, svcIPv4, ep1IPv4, ep2IPv4, loadBalancerIPv4, false, false, true, true)
+		})
+		t.Run("InternalTrafficPolicy:Local ExternalTrafficPolicy:Cluster", func(t *testing.T) {
+			testLoadBalancer(t, nodePortAddressesIPv4, svcIPv4, ep1IPv4, ep2IPv4, loadBalancerIPv4, false, true, false, true)
+		})
+		t.Run("InternalTrafficPolicy:Local ExternalTrafficPolicy:Local", func(t *testing.T) {
+			testLoadBalancer(t, nodePortAddressesIPv4, svcIPv4, ep1IPv4, ep2IPv4, loadBalancerIPv4, false, true, true, true)
+		})
+		t.Run("No External IPs", func(t *testing.T) {
+			testLoadBalancer(t, nodePortAddressesIPv4, svcIPv4, ep1IPv4, nil, loadBalancerIPv4, false, false, false, false)
+		})
+	})
+	t.Run("IPv6", func(t *testing.T) {
+		t.Run("InternalTrafficPolicy:Cluster ExternalTrafficPolicy:Cluster", func(t *testing.T) {
+			testLoadBalancer(t, nodePortAddressesIPv6, svcIPv6, ep1IPv6, ep2IPv6, loadBalancerIPv6, true, false, false, true)
+		})
+		t.Run("InternalTrafficPolicy:Cluster ExternalTrafficPolicy:Local", func(t *testing.T) {
+			testLoadBalancer(t, nodePortAddressesIPv6, svcIPv6, ep1IPv6, ep2IPv6, loadBalancerIPv6, true, false, true, true)
+		})
+		t.Run("InternalTrafficPolicy:Local ExternalTrafficPolicy:Cluster", func(t *testing.T) {
+			testLoadBalancer(t, nodePortAddressesIPv6, svcIPv6, ep1IPv6, ep2IPv6, loadBalancerIPv6, true, true, false, true)
+		})
+		t.Run("InternalTrafficPolicy:Local ExternalTrafficPolicy:Local", func(t *testing.T) {
+			testLoadBalancer(t, nodePortAddressesIPv6, svcIPv6, ep1IPv6, ep2IPv6, loadBalancerIPv6, true, true, true, true)
+		})
+		t.Run("No External IPs", func(t *testing.T) {
+			testLoadBalancer(t, nodePortAddressesIPv6, svcIPv6, ep1IPv6, nil, loadBalancerIPv6, true, false, false, false)
+		})
+	})
 }
 
-func TestClusterIPv6(t *testing.T) {
-	testClusterIP(t, svcIPv6, ep1IPv6, true, []*corev1.Service{}, []*corev1.Endpoints{})
+func TestNodePort(t *testing.T) {
+	t.Run("IPv4", func(t *testing.T) {
+		t.Run("InternalTrafficPolicy:Cluster ExternalTrafficPolicy:Cluster", func(t *testing.T) {
+			testNodePort(t, nodePortAddressesIPv4, svcIPv4, ep1IPv4, ep2IPv4, false, false, false)
+		})
+		t.Run("InternalTrafficPolicy:Cluster ExternalTrafficPolicy:Local", func(t *testing.T) {
+			testNodePort(t, nodePortAddressesIPv4, svcIPv4, ep1IPv4, ep2IPv4, false, false, true)
+		})
+		t.Run("InternalTrafficPolicy:Local ExternalTrafficPolicy:Cluster", func(t *testing.T) {
+			testNodePort(t, nodePortAddressesIPv4, svcIPv4, ep1IPv4, ep2IPv4, false, true, false)
+		})
+		t.Run("InternalTrafficPolicy:Local ExternalTrafficPolicy:Local", func(t *testing.T) {
+			testNodePort(t, nodePortAddressesIPv4, svcIPv4, ep1IPv4, ep2IPv4, false, true, true)
+		})
+	})
+	t.Run("IPv6", func(t *testing.T) {
+		t.Run("InternalTrafficPolicy:Cluster ExternalTrafficPolicy:Cluster", func(t *testing.T) {
+			testNodePort(t, nodePortAddressesIPv6, svcIPv6, ep1IPv6, ep2IPv6, true, false, false)
+		})
+		t.Run("InternalTrafficPolicy:Cluster ExternalTrafficPolicy:Local", func(t *testing.T) {
+			testNodePort(t, nodePortAddressesIPv6, svcIPv6, ep1IPv6, ep2IPv6, true, false, true)
+		})
+		t.Run("InternalTrafficPolicy:Local ExternalTrafficPolicy:Cluster", func(t *testing.T) {
+			testNodePort(t, nodePortAddressesIPv6, svcIPv6, ep1IPv6, ep2IPv6, true, true, false)
+		})
+		t.Run("InternalTrafficPolicy:Local ExternalTrafficPolicy:Local", func(t *testing.T) {
+			testNodePort(t, nodePortAddressesIPv6, svcIPv6, ep1IPv6, ep2IPv6, true, true, true)
+		})
+	})
 }
 
 func TestClusterSkipServices(t *testing.T) {
@@ -494,7 +616,7 @@ func TestClusterSkipServices(t *testing.T) {
 		}}
 	})
 	eps := []*corev1.Endpoints{ep1, ep2}
-	testClusterIP(t, svcIPv4, ep1IPv4, false, svcs, eps)
+	testClusterIP(t, svcIPv4, ep1IPv4, ep2IPv4, false, false, svcs, eps)
 }
 
 func TestDualStackService(t *testing.T) {
@@ -1114,12 +1236,12 @@ func TestMetrics(t *testing.T) {
 	metrics.Register()
 
 	for _, tc := range []struct {
-		name        string
-		svcIP, epIP string
-		isIPv6      bool
+		name                string
+		svcIP, ep1IP, ep2IP net.IP
+		isIPv6              bool
 	}{
-		{"IPv4", "10.20.30.41", "10.180.0.1", false},
-		{"IPv6", "fc01::1", "fe01::1", true},
+		{"IPv4", svcIPv4, ep1IPv4, ep2IPv4, false},
+		{"IPv6", svcIPv6, ep1IPv6, ep2IPv6, true},
 	} {
 		t.Run(tc.name, func(t *testing.T) {
 			endpointsUpdateTotalMetric := metrics.EndpointsUpdatesTotal.CounterMetric
@@ -1132,7 +1254,7 @@ func TestMetrics(t *testing.T) {
 				endpointsInstallMetric = metrics.EndpointsInstalledTotalV6.GaugeMetric
 				servicesInstallMetric = metrics.ServicesInstalledTotalV6.GaugeMetric
 			}
-			testClusterIP(t, net.ParseIP(tc.svcIP), net.ParseIP(tc.epIP), tc.isIPv6, []*corev1.Service{}, []*corev1.Endpoints{})
+			testClusterIP(t, tc.svcIP, tc.ep1IP, tc.ep2IP, tc.isIPv6, false, []*corev1.Service{}, []*corev1.Endpoints{})
 			v, err := testutil.GetCounterMetricValue(endpointsUpdateTotalMetric)
 			assert.NoError(t, err)
 			assert.Equal(t, 0, int(v))
@@ -1143,10 +1265,10 @@ func TestMetrics(t *testing.T) {
 			assert.Equal(t, 1, int(v))
 			assert.NoError(t, err)
 			v, err = testutil.GetGaugeMetricValue(endpointsInstallMetric)
-			assert.Equal(t, 1, int(v))
+			assert.Equal(t, 2, int(v))
 			assert.NoError(t, err)
 
-			testClusterIPRemoval(t, net.ParseIP(tc.svcIP), net.ParseIP(tc.epIP), tc.isIPv6)
+			testClusterIPRemoval(t, tc.svcIP, tc.ep1IP, tc.isIPv6)
 
 			v, err = testutil.GetCounterMetricValue(endpointsUpdateTotalMetric)
 			assert.NoError(t, err)
diff --git a/pkg/apiserver/handlers/featuregates/handler_test.go b/pkg/apiserver/handlers/featuregates/handler_test.go
index ec2d29210..943797855 100644
--- a/pkg/apiserver/handlers/featuregates/handler_test.go
+++ b/pkg/apiserver/handlers/featuregates/handler_test.go
@@ -29,6 +29,12 @@ import (
 	"k8s.io/client-go/kubernetes/fake"
 
 	"antrea.io/antrea/pkg/features"
+	"antrea.io/antrea/pkg/util/runtime"
+)
+
+var (
+	egressStatus string
+	nplStatus    string
 )
 
 func Test_getGatesResponse(t *testing.T) {
@@ -47,13 +53,13 @@ func Test_getGatesResponse(t *testing.T) {
 			want: []Response{
 				{Component: "agent", Name: "AntreaPolicy", Status: "Disabled", Version: "BETA"},
 				{Component: "agent", Name: "AntreaProxy", Status: "Enabled", Version: "BETA"},
-				{Component: "agent", Name: "Egress", Status: "Enabled", Version: "BETA"},
+				{Component: "agent", Name: "Egress", Status: egressStatus, Version: "BETA"},
 				{Component: "agent", Name: "EndpointSlice", Status: "Disabled", Version: "ALPHA"},
 				{Component: "agent", Name: "AntreaIPAM", Status: "Disabled", Version: "ALPHA"},
 				{Component: "agent", Name: "Traceflow", Status: "Enabled", Version: "BETA"},
 				{Component: "agent", Name: "FlowExporter", Status: "Disabled", Version: "ALPHA"},
 				{Component: "agent", Name: "NetworkPolicyStats", Status: "Enabled", Version: "BETA"},
-				{Component: "agent", Name: "NodePortLocal", Status: "Enabled", Version: "BETA"},
+				{Component: "agent", Name: "NodePortLocal", Status: nplStatus, Version: "BETA"},
 				{Component: "agent", Name: "Multicast", Status: "Disabled", Version: "ALPHA"},
 				{Component: "agent", Name: "ServiceExternalIP", Status: "Disabled", Version: "ALPHA"},
 			},
@@ -128,30 +134,12 @@ func TestHandleFunc(t *testing.T) {
 	)
 
 	tests := []struct {
-		name             string
-		expectedStatus   int
-		expectedResponse []Response
+		name           string
+		expectedStatus int
 	}{
 		{
 			name:           "good path",
 			expectedStatus: http.StatusOK,
-			expectedResponse: []Response{
-				{Component: "controller", Name: "AntreaPolicy", Status: "Enabled", Version: "BETA"},
-				{Component: "controller", Name: "Egress", Status: "Disabled", Version: "ALPHA"},
-				{Component: "controller", Name: "Traceflow", Status: "Enabled", Version: "BETA"},
-				{Component: "controller", Name: "NetworkPolicyStats", Status: "Enabled", Version: "BETA"},
-				{Component: "controller", Name: "NodeIPAM", Status: "Disabled", Version: "ALPHA"},
-				{Component: "controller", Name: "ServiceExternalIP", Status: "Disabled", Version: "ALPHA"},
-				{Component: "agent", Name: "AntreaPolicy", Status: "Enabled", Version: "BETA"},
-				{Component: "agent", Name: "AntreaProxy", Status: "Enabled", Version: "BETA"},
-				{Component: "agent", Name: "Egress", Status: "Disabled", Version: "ALPHA"},
-				{Component: "agent", Name: "EndpointSlice", Status: "Disabled", Version: "ALPHA"},
-				{Component: "agent", Name: "Traceflow", Status: "Enabled", Version: "BETA"},
-				{Component: "agent", Name: "FlowExporter", Status: "Disabled", Version: "ALPHA"},
-				{Component: "agent", Name: "NetworkPolicyStats", Status: "Enabled", Version: "BETA"},
-				{Component: "agent", Name: "NodePortLocal", Status: "Enabled", Version: "BETA"},
-				{Component: "agent", Name: "ServiceExternalIP", Status: "Disabled", Version: "ALPHA"},
-			},
 		},
 	}
 	os.Setenv("POD_NAME", "antrea-controller-wotqiwth")
@@ -191,7 +179,7 @@ func Test_getControllerGatesResponse(t *testing.T) {
 			name: "good path",
 			want: []Response{
 				{Component: "controller", Name: "AntreaPolicy", Status: "Enabled", Version: "BETA"},
-				{Component: "controller", Name: "Egress", Status: "Enabled", Version: "BETA"},
+				{Component: "controller", Name: "Egress", Status: egressStatus, Version: "BETA"},
 				{Component: "controller", Name: "Traceflow", Status: "Enabled", Version: "BETA"},
 				{Component: "controller", Name: "NetworkPolicyStats", Status: "Enabled", Version: "BETA"},
 				{Component: "controller", Name: "NodeIPAM", Status: "Disabled", Version: "ALPHA"},
@@ -214,3 +202,12 @@ func Test_getControllerGatesResponse(t *testing.T) {
 		})
 	}
 }
+
+func init() {
+	egressStatus = "Enabled"
+	nplStatus = "Enabled"
+	if runtime.IsWindowsPlatform() {
+		egressStatus = "Disabled"
+		nplStatus = "Disabled"
+	}
+}
diff --git a/test/integration/agent/openflow_test.go b/test/integration/agent/openflow_test.go
index 834af6536..06d14f5c9 100644
--- a/test/integration/agent/openflow_test.go
+++ b/test/integration/agent/openflow_test.go
@@ -270,7 +270,7 @@ func TestReplayFlowsNetworkPolicyFlows(t *testing.T) {
 	require.Nil(t, err, fmt.Sprintf("Failed to prepare OVS bridge: %v", err))
 
 	config := prepareConfiguration(true, false)
-	_, err = c.Initialize(roundInfo, config.nodeConfig, &config1.NetworkConfig{TrafficEncapMode: config1.TrafficEncapModeEncap})
+	_, err = c.Initialize(roundInfo, config.nodeConfig, &config1.NetworkConfig{TrafficEncapMode: config1.TrafficEncapModeEncap, IPv4Enabled: true}, &config1.EgressConfig{}, &config1.ServiceConfig{})
 	require.Nil(t, err, "Failed to initialize OFClient")
 
 	defer func() {
@@ -320,11 +320,6 @@ func TestReplayFlowsNetworkPolicyFlows(t *testing.T) {
 }
 
 func testExternalFlows(t *testing.T, config *testConfig) {
-	exceptCIDRs := []net.IPNet{}
-	if err := c.InstallExternalFlows(exceptCIDRs); err != nil {
-		t.Errorf("Failed to install OpenFlow entries to allow Pod to communicate to the external addresses: %v", err)
-	}
-
 	gwMACStr := config.nodeConfig.GatewayConfig.MAC.String()
 	if config.enableIPv4 {
 		for _, tableFlow := range expectedExternalFlows("ip", gwMACStr) {
@@ -360,7 +355,7 @@ func testReplayFlows(t *testing.T) {
 }
 
 func testInitialize(t *testing.T, config *testConfig) {
-	if _, err := c.Initialize(roundInfo, config.nodeConfig, &config1.NetworkConfig{TrafficEncapMode: config1.TrafficEncapModeEncap, IPv4Enabled: config.enableIPv4, IPv6Enabled: config.enableIPv6}); err != nil {
+	if _, err := c.Initialize(roundInfo, config.nodeConfig, &config1.NetworkConfig{TrafficEncapMode: config1.TrafficEncapModeEncap, IPv4Enabled: config.enableIPv4, IPv6Enabled: config.enableIPv6}, &config1.EgressConfig{}, &config1.ServiceConfig{}); err != nil {
 		t.Errorf("Failed to initialize openflow client: %v", err)
 	}
 	for _, tableFlow := range prepareDefaultFlows(config) {
@@ -370,20 +365,12 @@ func testInitialize(t *testing.T, config *testConfig) {
 }
 
 func testInstallTunnelFlows(t *testing.T, config *testConfig) {
-	err := c.InstallDefaultTunnelFlows()
-	if err != nil {
-		t.Fatalf("Failed to install Openflow entries for tunnel port: %v", err)
-	}
 	for _, tableFlow := range prepareTunnelFlows(config1.DefaultTunOFPort, config.globalMAC) {
 		ofTestUtils.CheckFlowExists(t, ovsCtlClient, tableFlow.tableName, 0, true, tableFlow.flows)
 	}
 }
 
 func testInstallServiceFlows(t *testing.T, config *testConfig) {
-	err := c.InstallDefaultServiceFlows(nil, nil)
-	if err != nil {
-		t.Fatalf("Failed to install Openflow entries to skip service CIDR from egress table: %v", err)
-	}
 	for _, tableFlow := range prepareServiceHelperFlows() {
 		ofTestUtils.CheckFlowExists(t, ovsCtlClient, tableFlow.tableName, 0, true, tableFlow.flows)
 	}
@@ -460,7 +447,7 @@ func TestNetworkPolicyFlows(t *testing.T) {
 	require.Nil(t, err, fmt.Sprintf("Failed to prepare OVS bridge %s", br))
 
 	config := prepareConfiguration(true, true)
-	_, err = c.Initialize(roundInfo, config.nodeConfig, &config1.NetworkConfig{TrafficEncapMode: config1.TrafficEncapModeEncap, IPv4Enabled: true, IPv6Enabled: true})
+	_, err = c.Initialize(roundInfo, config.nodeConfig, &config1.NetworkConfig{TrafficEncapMode: config1.TrafficEncapModeEncap, IPv4Enabled: true, IPv6Enabled: true}, &config1.EgressConfig{}, &config1.ServiceConfig{})
 	require.Nil(t, err, "Failed to initialize OFClient")
 
 	defer func() {
@@ -622,7 +609,7 @@ func TestProxyServiceFlows(t *testing.T) {
 	require.Nil(t, err, fmt.Sprintf("Failed to prepare OVS bridge %s", br))
 
 	config := prepareConfiguration(true, false)
-	_, err = c.Initialize(roundInfo, config.nodeConfig, &config1.NetworkConfig{TrafficEncapMode: config1.TrafficEncapModeEncap, IPv4Enabled: true})
+	_, err = c.Initialize(roundInfo, config.nodeConfig, &config1.NetworkConfig{TrafficEncapMode: config1.TrafficEncapModeEncap, IPv4Enabled: true}, &config1.EgressConfig{}, &config1.ServiceConfig{})
 	require.Nil(t, err, "Failed to initialize OFClient")
 
 	defer func() {
@@ -961,10 +948,6 @@ func checkOVSFlowMetrics(t *testing.T, client ofClient.Client) {
 
 func testInstallGatewayFlows(t *testing.T, config *testConfig) {
 	gatewayConfig := config.nodeConfig.GatewayConfig
-	err := c.InstallGatewayFlows()
-	if err != nil {
-		t.Fatalf("Failed to install Openflow entries for gateway: %v", err)
-	}
 	var ips []net.IP
 	if config.enableIPv4 {
 		ips = append(ips, gatewayConfig.IPv4)
@@ -980,6 +963,7 @@ func testInstallGatewayFlows(t *testing.T, config *testConfig) {
 func prepareConfiguration(enableIPv4, enableIPv6 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")
 	peerNodeMAC, _ := net.ParseMAC("aa:aa:aa:aa:ab:00")
 
 	nodeIPv4, nodeIPv4Subnet, _ := net.ParseCIDR("10.10.10.1/24")
@@ -990,7 +974,8 @@ func prepareConfiguration(enableIPv4, enableIPv6 bool) *testConfig {
 	_, peerIPv6Subnet, _ := net.ParseCIDR("fd74:ca9b:172:20::/64")
 
 	gatewayConfig := &config1.GatewayConfig{MAC: gwMAC}
-	nodeConfig := &config1.NodeConfig{GatewayConfig: gatewayConfig}
+	uplinkConfig := &config1.AdapterNetConfig{MAC: uplinkMAC}
+	nodeConfig := &config1.NodeConfig{GatewayConfig: gatewayConfig, UplinkNetConfig: uplinkConfig}
 	podCfg := &testLocalPodConfig{
 		name: "container-1",
 		testPortConfig: &testPortConfig{
@@ -1652,7 +1637,7 @@ func TestEgressMarkFlows(t *testing.T) {
 	require.Nil(t, err, fmt.Sprintf("Failed to prepare OVS bridge %s", br))
 
 	config := prepareConfiguration(true, true)
-	_, err = c.Initialize(roundInfo, config.nodeConfig, &config1.NetworkConfig{TrafficEncapMode: config1.TrafficEncapModeEncap})
+	_, err = c.Initialize(roundInfo, config.nodeConfig, &config1.NetworkConfig{TrafficEncapMode: config1.TrafficEncapModeEncap}, &config1.EgressConfig{}, &config1.ServiceConfig{})
 	require.Nil(t, err, "Failed to initialize OFClient")
 
 	defer func() {