Repository: cilium/cilium
Pull request #22705 (2022-12-13 14:01:41)

build(deps): bump azure/setup-helm from 3.4 to 3.5
Bumps [azure/setup-helm](https://github.com/azure/setup-helm) from 3.4 to 3.5.
<details>
<summary>Release notes</summary>
<p><em>Sourced from <a href="https://github.com/azure/setup-helm/releases">azure/setup-helm's releases</a>.</em></p>
<blockquote>
<h2>v3.5 release</h2>
<p>Bump <code>@​actions/core</code> version to remove output warning.</p>
</blockquote>
</details>
<details>
<summary>Commits</summary>
<ul>
<li><a href="https://github.com/Azure/setup-helm/commit/5119fcb9089d432beecbf79bb2c7915207344b78"><code>5119fcb</code></a> v3 new release (<a href="https://github-redirect.dependabot.com/azure/setup-helm/issues/104">#104</a>)</li>
<li>See full diff in <a href="https://github.com/azure/setup-helm/compare/f382f75448129b3be48f8121b9857be18d815a82...5119fcb9089d432beecbf79bb2c7915207344b78">compare view</a></li>
</ul>
</details>
<br />


[![Dependabot compatibility score](https://dependabot-badges.githubapp.com/badges/compatibility_score?dependency-name=azure/setup-helm&package-manager=github_actions&previous-version=3.4&new-version=3.5)](https://docs.github.com/en/github/managing-security-vulnerabilities/about-dependabot-security-updates#about-compatibility-scores)

You can trigger a rebase of this PR by commenting `@dependabot rebase`.

[//]: # (dependabot-automerge-start)
[//]: # (dependabot-automerge-end)

---

<details>
<summary>Dependabot commands and options</summary>
<br />

You can trigger Dependabot actions by commenting on this PR:
- `@dependabot rebase` will rebase this PR
- `@dependabot recreate` will recreate this PR, overwriting any edits that have been made to it
- `@dependabot merge` will merge this PR after your CI passes on it
- `@dependabot squash and merge` will squash and merge this PR after your CI passes on it
- `@dependabot cancel merge` will cancel a previously requested merge and block automerging
- `@dependabot reopen` will reopen this PR if it is closed
- `@dependabot close` will close this PR and stop Dependabot recreating it. You can achieve the same result by closing it manually
- `@dependabot ignore this major version` will close this PR and stop Dependabot creating any more for this major version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this minor version` will close this PR and stop Dependabot creating any more for this minor version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this dependency` will close this PR and stop Dependabot creating any more for this dependency (unless you reopen the PR or upgrade to it yourself)


</details>


Patch:
diff --git a/.github/workflows/conformance-aws-cni-v1.10.yaml b/.github/workflows/conformance-aws-cni-v1.10.yaml
index c6d351749b..c70f8176e3 100644
--- a/.github/workflows/conformance-aws-cni-v1.10.yaml
+++ b/.github/workflows/conformance-aws-cni-v1.10.yaml
@@ -181,7 +181,7 @@ jobs:
           rm eksctl_$(uname -s)_amd64.tar.gz
 
       - name: Install helm
-        uses: azure/setup-helm@f382f75448129b3be48f8121b9857be18d815a82
+        uses: azure/setup-helm@5119fcb9089d432beecbf79bb2c7915207344b78
         with:
           # Due to the below issue, v3.8.2 is pinned currently to avoid
           # exec plugin: invalid apiVersion "client.authentication.k8s.io/v1alpha1"
diff --git a/.github/workflows/conformance-eks-v1.10.yaml b/.github/workflows/conformance-eks-v1.10.yaml
index 04b0cceb8c..f3131ccd0e 100644
--- a/.github/workflows/conformance-eks-v1.10.yaml
+++ b/.github/workflows/conformance-eks-v1.10.yaml
@@ -195,7 +195,7 @@ jobs:
           rm eksctl_$(uname -s)_amd64.tar.gz
 
       - name: Install helm
-        uses: azure/setup-helm@f382f75448129b3be48f8121b9857be18d815a82
+        uses: azure/setup-helm@5119fcb9089d432beecbf79bb2c7915207344b78
         with:
           # Due to the below issue, v3.8.2 is pinned currently to avoid
           # exec plugin: invalid apiVersion "client.authentication.k8s.io/v1alpha1"
diff --git a/.github/workflows/conformance-gke-v1.12.yaml b/.github/workflows/conformance-gke-v1.12.yaml
index 7dc3530413..9c5a0286b4 100644
--- a/.github/workflows/conformance-gke-v1.12.yaml
+++ b/.github/workflows/conformance-gke-v1.12.yaml
@@ -265,8 +265,6 @@ jobs:
       - name: Clean up Cilium
         run: |
           pkill -f "cilium.*hubble.*port-forward|kubectl.*port-forward.*hubble-relay"
-          # Temporary workaround for https://github.com/cilium/cilium-cli/issues/255
-          kubectl delete ns cilium-test
           cilium uninstall --chart-directory=install/kubernetes/cilium --wait
 
       - name: Install Cilium with tunnel datapath
@@ -291,8 +289,6 @@ jobs:
       - name: Clean up Cilium
         run: |
           pkill -f "cilium.*hubble.*port-forward|kubectl.*port-forward.*hubble-relay"
-          # Temporary workaround for https://github.com/cilium/cilium-cli/issues/255
-          kubectl delete ns cilium-test
           cilium uninstall --chart-directory=install/kubernetes/cilium --wait
 
       - name: Create custom IPsec secret
@@ -321,8 +317,6 @@ jobs:
       - name: Clean up Cilium
         run: |
           pkill -f "cilium.*hubble.*port-forward|kubectl.*port-forward.*hubble-relay"
-          # Temporary workaround for https://github.com/cilium/cilium-cli/issues/255
-          kubectl delete ns cilium-test
           cilium uninstall --chart-directory=install/kubernetes/cilium --wait
 
       - name: Install Cilium with encryption and tunnel datapath
diff --git a/daemon/cmd/daemon.go b/daemon/cmd/daemon.go
index 55d6a05b93..b1b70768b3 100644
--- a/daemon/cmd/daemon.go
+++ b/daemon/cmd/daemon.go
@@ -7,6 +7,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"math/big"
 	"net"
 	"net/netip"
 	"os"
@@ -132,6 +133,10 @@ type Daemon struct {
 	// apply to locally running endpoints.
 	dnsNameManager *fqdn.NameManager
 
+	// dnsProxyContext contains fields relevant to the DNS proxy. See each
+	// field's godoc for more details.
+	dnsProxyContext dnsProxyContext
+
 	// Used to synchronize generation of daemon's BPF programs and endpoint BPF
 	// programs.
 	compilationMutex *lock.RWMutex
@@ -207,6 +212,28 @@ type Daemon struct {
 	bigTCPConfig bigtcp.Configuration
 }
 
+func (d *Daemon) initDNSProxyContext(size int) {
+	d.dnsProxyContext = dnsProxyContext{
+		responseMutexes: make([]*lock.Mutex, size),
+		modulus:         big.NewInt(int64(size)),
+	}
+	for i := range d.dnsProxyContext.responseMutexes {
+		d.dnsProxyContext.responseMutexes[i] = new(lock.Mutex)
+	}
+}
+
+// dnsProxyContext is a meta struct containing fields relevant to the DNS proxy
+// of the daemon, for organizational purposes.
+type dnsProxyContext struct {
+	// responseMutexes is used to serialized the critical path of
+	// notifyOnDNSMsg() to ensure that identity allocation and ipcache
+	// insertion happen atomically between parallel DNS request handling.
+	responseMutexes []*lock.Mutex
+	// modulus is used when computing the hash of the DNS response IPs in order
+	// to map them to the mutexes inside responseMutexes.
+	modulus *big.Int
+}
+
 // GetPolicyRepository returns the policy repository of the daemon
 func (d *Daemon) GetPolicyRepository() *policy.Repository {
 	return d.policy
diff --git a/daemon/cmd/daemon_main.go b/daemon/cmd/daemon_main.go
index 20f5b8938c..5478aa215f 100644
--- a/daemon/cmd/daemon_main.go
+++ b/daemon/cmd/daemon_main.go
@@ -867,6 +867,14 @@ func initializeFlags() {
 	flags.Duration(option.DNSProxyConcurrencyProcessingGracePeriod, 0, "Grace time to wait when DNS proxy concurrent limit has been reached during DNS message processing")
 	option.BindEnv(Vp, option.DNSProxyConcurrencyProcessingGracePeriod)
 
+	flags.Int(option.DNSProxyLockCount, 128, "Array size containing mutexes which protect against parallel handling of DNS response IPs")
+	flags.MarkHidden(option.DNSProxyLockCount)
+	option.BindEnv(Vp, option.DNSProxyLockCount)
+
+	flags.Duration(option.DNSProxyLockTimeout, 500*time.Millisecond, fmt.Sprintf("Timeout when acquiring the locks controlled by --%s", option.DNSProxyLockCount))
+	flags.MarkHidden(option.DNSProxyLockTimeout)
+	option.BindEnv(Vp, option.DNSProxyLockTimeout)
+
 	flags.Int(option.PolicyQueueSize, defaults.PolicyQueueSize, "Size of queues for policy-related events")
 	option.BindEnv(Vp, option.PolicyQueueSize)
 
diff --git a/daemon/cmd/fqdn.go b/daemon/cmd/fqdn.go
index f4bbbfc8ba..2b31ace67b 100644
--- a/daemon/cmd/fqdn.go
+++ b/daemon/cmd/fqdn.go
@@ -7,10 +7,12 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"math/big"
 	"net"
 	"net/netip"
 	"os"
 	"regexp"
+	"sort"
 	"strconv"
 	"strings"
 	"sync"
@@ -158,6 +160,8 @@ func (d *Daemon) updateSelectorCacheFQDNs(ctx context.Context, selectors map[pol
 // default DNS cache. The proxy binds to all interfaces, and uses the
 // configured DNS proxy port (this may be 0 and so OS-assigned).
 func (d *Daemon) bootstrapFQDN(possibleEndpoints map[uint16]*endpoint.Endpoint, preCachePath string) (err error) {
+	d.initDNSProxyContext(option.Config.DNSProxyLockCount)
+
 	cfg := fqdn.Config{
 		MinTTL:          option.Config.ToFQDNsMinTTL,
 		Cache:           fqdn.NewDNSCache(option.Config.ToFQDNsMinTTL),
@@ -533,6 +537,55 @@ func (d *Daemon) notifyOnDNSMsg(lookupTime time.Time, ep *endpoint.Endpoint, epI
 
 	if msg.Response && msg.Rcode == dns.RcodeSuccess && len(responseIPs) > 0 {
 		stat.PolicyGenerationTime.Start()
+		// Create a critical section especially for when multiple DNS requests
+		// are in-flight for the same name (i.e. cilium.io).
+		//
+		// In the absence of such a critical section, consider the following
+		// race condition:
+		//
+		//              G1                                    G2
+		//
+		// T0 --> NotifyOnDNSMsg()               NotifyOnDNSMsg()            <-- T0
+		//
+		// T1 --> UpdateGenerateDNS()            UpdateGenerateDNS()         <-- T1
+		//
+		// T2 ----> mutex.Lock()                 +---------------------------+
+		//                                       |No identities need updating|
+		// T3 ----> mutex.Unlock()               +---------------------------+
+		//
+		// T4 --> UpsertGeneratedIdentities()    UpsertGeneratedIdentities() <-- T4
+		//
+		// T5 ---->  Upsert()                    DNS released back to pod    <-- T5
+		//                                                    |
+		// T6 --> DNS released back to pod                    |
+		//              |                                     |
+		//              |                                     |
+		//              v                                     v
+		//       Traffic flows fine                   Leads to policy drop
+		//
+		// Note how G2 releases the DNS msg back to the pod at T5 because
+		// UpdateGenerateDNS() was a no-op. It's a no-op because G1 had executed
+		// UpdateGenerateDNS() first at T1 and performed the necessary identity
+		// allocation for the response IPs. Due to G1 performing all the work
+		// first, G2 executes T4 also as a no-op and releases the msg back to the
+		// pod at T5 before G1 would at T6.
+		mutexAcquireStart := time.Now()
+		mutexes := d.dnsProxyContext.getMutexesForResponseIPs(responseIPs)
+		for _, m := range mutexes {
+			d.dnsProxyContext.responseMutexes[m].Lock()
+			defer d.dnsProxyContext.responseMutexes[m].Unlock()
+		}
+		if d := time.Since(mutexAcquireStart); d >= option.Config.DNSProxyLockTimeout {
+			log.WithFields(logrus.Fields{
+				logfields.DNSName:  qname,
+				logfields.Duration: d,
+				logfields.Expected: option.Config.DNSProxyLockTimeout,
+			}).Warnf("Lock acquisition time took longer than expected. "+
+				"Potentially too many parallel DNS requests being processed, "+
+				"consider adjusting --%s and/or --%s",
+				option.DNSProxyLockCount, option.DNSProxyLockTimeout)
+		}
+
 		// This must happen before the NameManager update below, to ensure that
 		// this data is included in the serialized Endpoint object.
 		// We also need to add to the cache before we purge any matching zombies
@@ -595,6 +648,39 @@ func (d *Daemon) notifyOnDNSMsg(lookupTime time.Time, ep *endpoint.Endpoint, epI
 	return nil
 }
 
+// getMutexesForResponseIPs returns a slice of indices for accessing the
+// mutexes in dnsProxyContext. There's a many-to-one mapping from IP to mutex,
+// meaning multiple IPs may map to a single mutex. The many-to-one property is
+// obtained by hashing each IP inside responseIPs. In order to prevent the
+// caller from acquiring the mutexes in an undesirable order, this function
+// ensures that the slice returned in ascending order. This is the order in
+// which the mutexes should be taken and released. The slice is de-duplicated
+// to avoid acquiring and releasing the same mutex in order.
+func (dpc *dnsProxyContext) getMutexesForResponseIPs(responseIPs []net.IP) []int {
+	// cache stores all unique indices for mutexes. Prevents the same mutex
+	// index from being added to indices which prevents the caller from
+	// attempting to acquire the same mutex multiple times.
+	cache := make(map[int]struct{}, len(responseIPs))
+	indices := make([]int, 0, len(responseIPs))
+	for _, ip := range responseIPs {
+		h := ipToInt(ip)
+		m := h.Mod(h, dpc.modulus)
+		i := int(m.Int64())
+		if _, exists := cache[i]; !exists {
+			cache[i] = struct{}{}
+			indices = append(indices, i)
+		}
+	}
+	sort.Ints(indices)
+	return indices
+}
+
+func ipToInt(addr net.IP) *big.Int {
+	i := big.NewInt(0)
+	i.SetBytes(addr)
+	return i
+}
+
 type getFqdnCache struct {
 	daemon *Daemon
 }
diff --git a/pkg/endpoint/bpf.go b/pkg/endpoint/bpf.go
index 193ba72972..7ed6e3609e 100644
--- a/pkg/endpoint/bpf.go
+++ b/pkg/endpoint/bpf.go
@@ -253,8 +253,14 @@ func (e *Endpoint) addNewRedirectsFromDesiredPolicy(ingress bool, desiredRedirec
 				var err error
 				redirectPort, err, finalizeFunc, revertFunc = e.proxy.CreateOrUpdateRedirect(e.aliveCtx, l4, proxyID, e, proxyWaitGroup)
 				if err != nil {
-					revertStack.Revert() // Ignore errors while reverting. This is best-effort.
-					return err, nil, nil
+					// Skip redirects that can not be created or updated.  This
+					// can happen when a listener is missing, for example when
+					// restarting and k8s delivers the CNP before the related
+					// CEC.
+					// Policy is regenerated when listeners are added or removed
+					// to fix this condition when the listener is available.
+					e.getLogger().WithField(logfields.Listener, l4.GetListener()).WithError(err).Debug("Redirect rule with missing listener skipped, will be applied once the listener is available")
+					continue
 				}
 				finalizeList.Append(finalizeFunc)
 				revertStack.Push(revertFunc)
diff --git a/pkg/envoy/ciliumenvoyconfig.go b/pkg/envoy/ciliumenvoyconfig.go
index 943e62675e..9105aa3152 100644
--- a/pkg/envoy/ciliumenvoyconfig.go
+++ b/pkg/envoy/ciliumenvoyconfig.go
@@ -6,7 +6,6 @@ package envoy
 import (
 	"context"
 	"fmt"
-	"strings"
 	"time"
 
 	envoy_config_cluster "github.com/cilium/proxy/go/envoy/config/cluster/v3"
@@ -15,6 +14,7 @@ import (
 	envoy_config_listener "github.com/cilium/proxy/go/envoy/config/listener/v3"
 	envoy_config_route "github.com/cilium/proxy/go/envoy/config/route/v3"
 	envoy_config_http "github.com/cilium/proxy/go/envoy/extensions/filters/network/http_connection_manager/v3"
+	envoy_config_tcp "github.com/cilium/proxy/go/envoy/extensions/filters/network/tcp_proxy/v3"
 	envoy_config_tls "github.com/cilium/proxy/go/envoy/extensions/transport_sockets/tls/v3"
 	"google.golang.org/protobuf/types/known/anypb"
 	"google.golang.org/protobuf/types/known/wrapperspb"
@@ -24,9 +24,10 @@ import (
 	cilium_v2 "github.com/cilium/cilium/pkg/k8s/apis/cilium.io/v2"
 	lb "github.com/cilium/cilium/pkg/loadbalancer"
 	"github.com/cilium/cilium/pkg/option"
+	"github.com/cilium/cilium/pkg/policy/api"
 
 	// Imports for Envoy extensions not used directly from Cilium Agent, but that we want to
-	// be registered for use in Cilium Enco Config CRDs:
+	// be registered for use in Cilium Envoy Config CRDs:
 	_ "github.com/cilium/proxy/go/envoy/extensions/clusters/dynamic_forward_proxy/v3"
 	_ "github.com/cilium/proxy/go/envoy/extensions/filters/http/dynamic_forward_proxy/v3"
 	_ "github.com/cilium/proxy/go/envoy/extensions/filters/http/ext_authz/v3"
@@ -63,9 +64,42 @@ type PortAllocator interface {
 	ReleaseProxyPort(name string) error
 }
 
-// ParseResources parses all supported Envoy resource types from CiliumEnvoyConfig CRD to Resources type
-// cecNamespace and cecName parameters, if not empty, will be prepended to the Envoy resource names.
-func ParseResources(cecNamespace string, cecName string, anySlice []cilium_v2.XDSResource, validate bool, portAllocator PortAllocator) (Resources, error) {
+// ListenersAddedOrDeleted returns 'true' if a listener is added or removed when updating from 'old'
+// to 'new'
+func (old *Resources) ListenersAddedOrDeleted(new *Resources) bool {
+	// Typically the number of listeners in a CEC is small (e.g, one), so it should be OK to
+	// scan the slices like here
+	for _, nl := range new.Listeners {
+		found := false
+		for _, ol := range old.Listeners {
+			if ol.Name == nl.Name {
+				found = true
+				break
+			}
+		}
+		if !found {
+			return true // a listener was added
+		}
+	}
+	for _, ol := range old.Listeners {
+		found := false
+		for _, nl := range new.Listeners {
+			if nl.Name == ol.Name {
+				found = true
+				break
+			}
+		}
+		if !found {
+			return true // a listener was removed
+		}
+	}
+	return false
+}
+
+// ParseResources parses all supported Envoy resource types from CiliumEnvoyConfig CRD to Resources
+// type cecNamespace and cecName parameters, if not empty, will be prepended to the Envoy resource
+// names.
+func ParseResources(cecNamespace string, cecName string, anySlice []cilium_v2.XDSResource, validate bool, portAllocator PortAllocator, isL7LB bool) (Resources, error) {
 	resources := Resources{}
 	for _, r := range anySlice {
 		// Skip empty TypeURLs, which are left behind when Unmarshaling resource JSON fails
@@ -108,7 +142,7 @@ func ParseResources(cecNamespace string, cecName string, anySlice []cilium_v2.XD
 				}
 			}
 			if !found {
-				listener.ListenerFilters = append(listener.ListenerFilters, getListenerFilter(false /* not ingress */, false, true))
+				listener.ListenerFilters = append(listener.ListenerFilters, getListenerFilter(false /* not ingress */, !isL7LB, isL7LB))
 			}
 			// Inject listener socket option for Cilium datapath
 			listener.SocketOptions = append(listener.SocketOptions, getListenerSocketMarkOption(false /* not ingress */))
@@ -122,68 +156,85 @@ func ParseResources(cecNamespace string, cecName string, anySlice []cilium_v2.XD
 						foundCiliumNetworkFilter = true
 					}
 					tc := filter.GetTypedConfig()
-					if tc == nil || tc.GetTypeUrl() != HttpConnectionManagerTypeURL {
-						continue
-					}
-					any, err := tc.UnmarshalNew()
-					if err != nil {
+					if tc == nil {
 						continue
 					}
-					hcmConfig, ok := any.(*envoy_config_http.HttpConnectionManager)
-					if !ok {
-						continue
-					}
-					updated := false
-					if rds := hcmConfig.GetRds(); rds != nil {
-						if rds.ConfigSource == nil {
-							rds.ConfigSource = ciliumXDS
-							updated = true
+					switch tc.GetTypeUrl() {
+					case HttpConnectionManagerTypeURL:
+						any, err := tc.UnmarshalNew()
+						if err != nil {
+							continue
+						}
+						hcmConfig, ok := any.(*envoy_config_http.HttpConnectionManager)
+						if !ok {
+							continue
+						}
+						updated := false
+						if rds := hcmConfig.GetRds(); rds != nil {
+							// Since we are prepending CEC namespace and name to Routes name,
+							// we must do the same here to point to the correct Route resource.
+							if rds.RouteConfigName != "" {
+								rds.RouteConfigName = api.ResourceQualifiedName(cecNamespace, cecName, rds.RouteConfigName)
+								updated = true
+							}
+							if rds.ConfigSource == nil {
+								rds.ConfigSource = ciliumXDS
+								updated = true
+							}
+						}
+						if listener.GetAddress() == nil {
+							foundCiliumL7Filter := false
+						loop:
+							for j, httpFilter := range hcmConfig.HttpFilters {
+								switch httpFilter.Name {
+								case "cilium.l7policy":
+									foundCiliumL7Filter = true
+								case "envoy.filters.http.router":
+									if !foundCiliumL7Filter {
+										// Inject Cilium HTTP filter just before the HTTP Router filter
+										hcmConfig.HttpFilters = append(hcmConfig.HttpFilters[:j+1], hcmConfig.HttpFilters[j:]...)
+										hcmConfig.HttpFilters[j] = getCiliumHttpFilter()
+										updated = true
+									}
+									break loop
+								}
+							}
+						}
+						if updated {
+							filter.ConfigType = &envoy_config_listener.Filter_TypedConfig{
+								TypedConfig: toAny(hcmConfig),
+							}
+						}
+					case TCPProxyTypeURL:
+						any, err := tc.UnmarshalNew()
+						if err != nil {
+							continue
 						}
-						// Since we are prepending CEC namespace and name to Routes name,
-						// we must do the same here to point to the correct Route resource.
-						if rds.RouteConfigName != "" {
-							rds.RouteConfigName = resourceQualifiedName(cecNamespace, cecName, rds.RouteConfigName)
-							updated = true
+						_, ok := any.(*envoy_config_tcp.TcpProxy)
+						if !ok {
+							continue
 						}
+					default:
+						continue
 					}
 					// Only inject Cilium policy enforcement filters for
 					// listeners for which Cilium agent allocates address
 					// for (see below)
 					if listener.GetAddress() == nil {
 						if !foundCiliumNetworkFilter {
-							// Inject Cilium network filter just before the HTTP Connection Manager filter
+							// Inject Cilium network filter just before the HTTP Connection Manager or TCPProxy filter
 							fc.Filters = append(fc.Filters[:i+1], fc.Filters[i:]...)
 							fc.Filters[i] = &envoy_config_listener.Filter{
 								Name: "cilium.network",
 							}
 						}
-						foundCiliumL7Filter := false
-						for j, httpFilter := range hcmConfig.HttpFilters {
-							switch httpFilter.Name {
-							case "cilium.l7policy":
-								foundCiliumL7Filter = true
-							case "envoy.filters.http.router":
-								if !foundCiliumL7Filter {
-									// Inject Cilium HTTP filter just before the HTTP Router filter
-									hcmConfig.HttpFilters = append(hcmConfig.HttpFilters[:j+1], hcmConfig.HttpFilters[j:]...)
-									hcmConfig.HttpFilters[j] = getCiliumHttpFilter()
-									updated = true
-								}
-								break
-							}
-						}
-					}
-					if updated {
-						filter.ConfigType = &envoy_config_listener.Filter_TypedConfig{
-							TypedConfig: toAny(hcmConfig),
-						}
 					}
 					break // Done with this filter chain
 				}
 			}
 
 			name := listener.Name
-			listener.Name = resourceQualifiedName(cecNamespace, cecName, listener.Name)
+			listener.Name = api.ResourceQualifiedName(cecNamespace, cecName, listener.Name)
 			resources.Listeners = append(resources.Listeners, listener)
 
 			log.Debugf("ParseResources: Parsed listener %q: %v", name, listener)
@@ -209,7 +260,7 @@ func ParseResources(cecNamespace string, cecName string, anySlice []cilium_v2.XD
 			}
 
 			name := route.Name
-			route.Name = resourceQualifiedName(cecNamespace, cecName, route.Name)
+			route.Name = api.ResourceQualifiedName(cecNamespace, cecName, route.Name)
 			resources.Routes = append(resources.Routes, route)
 
 			log.Debugf("ParseResources: Parsed route %q: %v", name, route)
@@ -358,6 +409,13 @@ func (s *XDSServer) UpsertEnvoyResources(ctx context.Context, resources Resource
 		log.Debugf("UpsertEnvoyResources: Upserting %s...", msg)
 	}
 	var wg *completion.WaitGroup
+	// Listener config may fail if it refers to a cluster that has not been added yet, so we
+	// must wait for Envoy to ACK cluster config before adding Listeners to be sure Listener
+	// config does not fail for this reason.
+	// Enable wait before new Listeners are added if clusters are also added.
+	if len(resources.Listeners) > 0 && len(resources.Clusters) > 0 {
+		wg = completion.NewWaitGroup(ctx)
+	}
 	var revertFuncs xds.AckingResourceMutatorRevertFuncList
 	// Do not wait for the addition of routes, clusters, endpoints, routes,
 	// or secrets as there are no guarantees that these additions will be
@@ -366,6 +424,7 @@ func (s *XDSServer) UpsertEnvoyResources(ctx context.Context, resources Resource
 	// in which case we could wait forever for the ACKs. This could also
 	// happen if there is no listener referring to these named
 	// resources to begin with.
+	// If both listeners and clusters are added then wait for clusters.
 	for _, r := range resources.Secrets {
 		log.Debugf("Envoy upsertSecret %s %v", r.Name, r)
 		revertFuncs = append(revertFuncs, s.upsertSecret(r.Name, r, nil, nil))
@@ -376,12 +435,27 @@ func (s *XDSServer) UpsertEnvoyResources(ctx context.Context, resources Resource
 	}
 	for _, r := range resources.Clusters {
 		log.Debugf("Envoy upsertCluster %s %v", r.Name, r)
-		revertFuncs = append(revertFuncs, s.upsertCluster(r.Name, r, nil, nil))
+		revertFuncs = append(revertFuncs, s.upsertCluster(r.Name, r, wg, nil))
 	}
 	for _, r := range resources.Routes {
 		log.Debugf("Envoy upsertRoute %s %v", r.Name, r)
 		revertFuncs = append(revertFuncs, s.upsertRoute(r.Name, r, nil, nil))
 	}
+	// Wait before new Listeners are added if clusters were also added above.
+	if wg != nil {
+		start := time.Now()
+		log.Debug("UpsertEnvoyResources: Waiting for cluster updates to complete...")
+		err := wg.Wait()
+		log.Debugf("UpsertEnvoyResources: Wait time for cluster updates %v (err: %s)", time.Since(start), err)
+
+		// revert all changes in case of failure
+		if err != nil {
+			revertFuncs.Revert(nil)
+			log.Debug("UpsertEnvoyResources: Finished reverting failed xDS transactions")
+			return err
+		}
+		wg = nil
+	}
 	// Wait only if new Listeners are added, as they will always be acked.
 	// (unreferenced routes or endpoints (and maybe clusters) are not ACKed or NACKed).
 	if len(resources.Listeners) > 0 {
@@ -573,12 +647,23 @@ func (s *XDSServer) UpdateEnvoyResources(ctx context.Context, old, new Resources
 	}
 	// Add new Clusters
 	for _, r := range new.Clusters {
-		revertFuncs = append(revertFuncs, s.upsertCluster(r.Name, r, nil, nil))
+		revertFuncs = append(revertFuncs, s.upsertCluster(r.Name, r, wg, nil))
 	}
 	// Add new Routes
 	for _, r := range new.Routes {
 		revertFuncs = append(revertFuncs, s.upsertRoute(r.Name, r, nil, nil))
 	}
+	if wg != nil && len(new.Clusters) > 0 {
+		start := time.Now()
+		log.Debug("UpdateEnvoyResources: Waiting for cluster updates to complete...")
+		err := wg.Wait()
+		if err != nil {
+			log.Debug("UpdateEnvoyResources: cluster update failed: ", err)
+		}
+		log.Debug("UpdateEnvoyResources: Wait time for cluster updates: ", time.Since(start))
+		// new wait group for adds
+		wg = completion.NewWaitGroup(ctx)
+	}
 	// Add new Listeners
 	for _, r := range new.Listeners {
 		listenerName := r.Name
@@ -768,25 +853,3 @@ func fillInTransportSocketXDS(ts *envoy_config_core.TransportSocket) {
 		}
 	}
 }
-
-// resourceQualifiedName returns the qualified name of an Envoy resource,
-// prepending CEC namespace and CEC name to the resource name and using
-// '/' as a separator.
-//
-// In case of an empty CEC namespace or an empty CEC name, leading separators
-// are stripped away.
-func resourceQualifiedName(namespace, name, resource string) string {
-	var sb strings.Builder
-
-	if namespace != "" {
-		sb.WriteString(namespace)
-		sb.WriteRune('/')
-	}
-	if name != "" {
-		sb.WriteString(name)
-		sb.WriteRune('/')
-	}
-	sb.WriteString(resource)
-
-	return sb.String()
-}
diff --git a/pkg/envoy/resources.go b/pkg/envoy/resources.go
index 2b1c7b20b5..7c4eb9e689 100644
--- a/pkg/envoy/resources.go
+++ b/pkg/envoy/resources.go
@@ -29,9 +29,12 @@ const (
 	// ClusterTypeURL is the type URL of Cluster resources.
 	ClusterTypeURL = "type.googleapis.com/envoy.config.cluster.v3.Cluster"
 
-	// HttpConnectionManagerTypeURL is the type URL of HttpConnectionManager resources.
+	// HttpConnectionManagerTypeURL is the type URL of HttpConnectionManager filter.
 	HttpConnectionManagerTypeURL = "type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager"
 
+	// TCPProxyTypeURL is the type URL of TCPProxy filter.
+	TCPProxyTypeURL = "type.googleapis.com/envoy.extensions.filters.network.tcp_proxy.v3.TcpProxy"
+
 	// EndpointTypeURL is the type URL of Endpoint resources.
 	EndpointTypeURL = "type.googleapis.com/envoy.config.endpoint.v3.ClusterLoadAssignment"
 
diff --git a/pkg/envoy/server.go b/pkg/envoy/server.go
index f8bffa9894..421d67028d 100644
--- a/pkg/envoy/server.go
+++ b/pkg/envoy/server.go
@@ -1428,11 +1428,11 @@ func getWildcardNetworkPolicyRule(selectors policy.L7DataMap) *cilium.PortNetwor
 			remoteMap[uint64(id)] = struct{}{}
 		}
 
-		if !l7.IsEmpty() {
-			// If it is not empty then we issue the warning.
+		if l7.IsRedirect() {
+			// Issue a warning if this port-0 rule is a redirect.
 			// Deny rules don't support L7 therefore for the deny case
-			// l7.IsEmpty() will always return true.
-			log.Warningf("L3-only rule for selector %v surprisingly has L7 rules (%v)!", sel, *l7)
+			// l7.IsRedirect() will always return false.
+			log.Warningf("L3-only rule for selector %v surprisingly requires proxy redirection (%v)!", sel, *l7)
 		}
 	}
 
diff --git a/pkg/k8s/apis/cilium.io/client/crds/v2/ciliumclusterwidenetworkpolicies.yaml b/pkg/k8s/apis/cilium.io/client/crds/v2/ciliumclusterwidenetworkpolicies.yaml
index a65dce6fe1..454af63b5f 100644
--- a/pkg/k8s/apis/cilium.io/client/crds/v2/ciliumclusterwidenetworkpolicies.yaml
+++ b/pkg/k8s/apis/cilium.io/client/crds/v2/ciliumclusterwidenetworkpolicies.yaml
@@ -339,6 +339,45 @@ spec:
                         description: PortRule is a list of ports/protocol combinations
                           with optional Layer 7 rules which must be met.
                         properties:
+                          listener:
+                            description: listener specifies the name of a custom Envoy
+                              listener to which this traffic should be redirected
+                              to.
+                            properties:
+                              envoyConfig:
+                                description: EnvoyConfig is a reference to the CEC
+                                  or CCNP resource in which the listener is defined.
+                                properties:
+                                  kind:
+                                    description: Kind is the resource type being referred
+                                      to. Defaults to CiliumEnvoyConfig or CiliumClusterwideEnvoyConfig
+                                      for CiliumNetworkPolicy and CiliumClusterwideNetworkPolicy,
+                                      respectively. The only case this is currently
+                                      explicitly needed is when referring to a CiliumClusterwideEnvoyConfig
+                                      from CiliumNetworkPolicy, as using a namespaced
+                                      listener from a cluster scoped policy is not
+                                      allowed.
+                                    enum:
+                                    - CiliumEnvoyConfig
+                                    - CiliumClusterwideEnvoyConfig
+                                    type: string
+                                  name:
+                                    description: Name is the resource name of the
+                                      CiliumEnvoyConfig or CiliumClusterwideEnvoyConfig
+                                      where the listener is defined in.
+                                    minLength: 1
+                                    type: string
+                                required:
+                                - name
+                                type: object
+                              name:
+                                description: Name is the name of the listener.
+                                minLength: 1
+                                type: string
+                            required:
+                            - envoyConfig
+                            - name
+                            type: object
                           originatingTLS:
                             description: OriginatingTLS is the TLS context for the
                               connections originated by the L7 proxy.  For egress
@@ -1623,6 +1662,45 @@ spec:
                         description: PortRule is a list of ports/protocol combinations
                           with optional Layer 7 rules which must be met.
                         properties:
+                          listener:
+                            description: listener specifies the name of a custom Envoy
+                              listener to which this traffic should be redirected
+                              to.
+                            properties:
+                              envoyConfig:
+                                description: EnvoyConfig is a reference to the CEC
+                                  or CCNP resource in which the listener is defined.
+                                properties:
+                                  kind:
+                                    description: Kind is the resource type being referred
+                                      to. Defaults to CiliumEnvoyConfig or CiliumClusterwideEnvoyConfig
+                                      for CiliumNetworkPolicy and CiliumClusterwideNetworkPolicy,
+                                      respectively. The only case this is currently
+                                      explicitly needed is when referring to a CiliumClusterwideEnvoyConfig
+                                      from CiliumNetworkPolicy, as using a namespaced
+                                      listener from a cluster scoped policy is not
+                                      allowed.
+                                    enum:
+                                    - CiliumEnvoyConfig
+                                    - CiliumClusterwideEnvoyConfig
+                                    type: string
+                                  name:
+                                    description: Name is the resource name of the
+                                      CiliumEnvoyConfig or CiliumClusterwideEnvoyConfig
+                                      where the listener is defined in.
+                                    minLength: 1
+                                    type: string
+                                required:
+                                - name
+                                type: object
+                              name:
+                                description: Name is the name of the listener.
+                                minLength: 1
+                                type: string
+                            required:
+                            - envoyConfig
+                            - name
+                            type: object
                           originatingTLS:
                             description: OriginatingTLS is the TLS context for the
                               connections originated by the L7 proxy.  For egress
@@ -2711,6 +2789,46 @@ spec:
                           description: PortRule is a list of ports/protocol combinations
                             with optional Layer 7 rules which must be met.
                           properties:
+                            listener:
+                              description: listener specifies the name of a custom
+                                Envoy listener to which this traffic should be redirected
+                                to.
+                              properties:
+                                envoyConfig:
+                                  description: EnvoyConfig is a reference to the CEC
+                                    or CCNP resource in which the listener is defined.
+                                  properties:
+                                    kind:
+                                      description: Kind is the resource type being
+                                        referred to. Defaults to CiliumEnvoyConfig
+                                        or CiliumClusterwideEnvoyConfig for CiliumNetworkPolicy
+                                        and CiliumClusterwideNetworkPolicy, respectively.
+                                        The only case this is currently explicitly
+                                        needed is when referring to a CiliumClusterwideEnvoyConfig
+                                        from CiliumNetworkPolicy, as using a namespaced
+                                        listener from a cluster scoped policy is not
+                                        allowed.
+                                      enum:
+                                      - CiliumEnvoyConfig
+                                      - CiliumClusterwideEnvoyConfig
+                                      type: string
+                                    name:
+                                      description: Name is the resource name of the
+                                        CiliumEnvoyConfig or CiliumClusterwideEnvoyConfig
+                                        where the listener is defined in.
+                                      minLength: 1
+                                      type: string
+                                  required:
+                                  - name
+                                  type: object
+                                name:
+                                  description: Name is the name of the listener.
+                                  minLength: 1
+                                  type: string
+                              required:
+                              - envoyConfig
+                              - name
+                              type: object
                             originatingTLS:
                               description: OriginatingTLS is the TLS context for the
                                 connections originated by the L7 proxy.  For egress
@@ -4014,6 +4132,46 @@ spec:
                           description: PortRule is a list of ports/protocol combinations
                             with optional Layer 7 rules which must be met.
                           properties:
+                            listener:
+                              description: listener specifies the name of a custom
+                                Envoy listener to which this traffic should be redirected
+                                to.
+                              properties:
+                                envoyConfig:
+                                  description: EnvoyConfig is a reference to the CEC
+                                    or CCNP resource in which the listener is defined.
+                                  properties:
+                                    kind:
+                                      description: Kind is the resource type being
+                                        referred to. Defaults to CiliumEnvoyConfig
+                                        or CiliumClusterwideEnvoyConfig for CiliumNetworkPolicy
+                                        and CiliumClusterwideNetworkPolicy, respectively.
+                                        The only case this is currently explicitly
+                                        needed is when referring to a CiliumClusterwideEnvoyConfig
+                                        from CiliumNetworkPolicy, as using a namespaced
+                                        listener from a cluster scoped policy is not
+                                        allowed.
+                                      enum:
+                                      - CiliumEnvoyConfig
+                                      - CiliumClusterwideEnvoyConfig
+                                      type: string
+                                    name:
+                                      description: Name is the resource name of the
+                                        CiliumEnvoyConfig or CiliumClusterwideEnvoyConfig
+                                        where the listener is defined in.
+                                      minLength: 1
+                                      type: string
+                                  required:
+                                  - name
+                                  type: object
+                                name:
+                                  description: Name is the name of the listener.
+                                  minLength: 1
+                                  type: string
+                              required:
+                              - envoyConfig
+                              - name
+                              type: object
                             originatingTLS:
                               description: OriginatingTLS is the TLS context for the
                                 connections originated by the L7 proxy.  For egress
diff --git a/pkg/k8s/apis/cilium.io/client/crds/v2/ciliumnetworkpolicies.yaml b/pkg/k8s/apis/cilium.io/client/crds/v2/ciliumnetworkpolicies.yaml
index 97984299c6..ee42b7b15c 100644
--- a/pkg/k8s/apis/cilium.io/client/crds/v2/ciliumnetworkpolicies.yaml
+++ b/pkg/k8s/apis/cilium.io/client/crds/v2/ciliumnetworkpolicies.yaml
@@ -343,6 +343,45 @@ spec:
                         description: PortRule is a list of ports/protocol combinations
                           with optional Layer 7 rules which must be met.
                         properties:
+                          listener:
+                            description: listener specifies the name of a custom Envoy
+                              listener to which this traffic should be redirected
+                              to.
+                            properties:
+                              envoyConfig:
+                                description: EnvoyConfig is a reference to the CEC
+                                  or CCNP resource in which the listener is defined.
+                                properties:
+                                  kind:
+                                    description: Kind is the resource type being referred
+                                      to. Defaults to CiliumEnvoyConfig or CiliumClusterwideEnvoyConfig
+                                      for CiliumNetworkPolicy and CiliumClusterwideNetworkPolicy,
+                                      respectively. The only case this is currently
+                                      explicitly needed is when referring to a CiliumClusterwideEnvoyConfig
+                                      from CiliumNetworkPolicy, as using a namespaced
+                                      listener from a cluster scoped policy is not
+                                      allowed.
+                                    enum:
+                                    - CiliumEnvoyConfig
+                                    - CiliumClusterwideEnvoyConfig
+                                    type: string
+                                  name:
+                                    description: Name is the resource name of the
+                                      CiliumEnvoyConfig or CiliumClusterwideEnvoyConfig
+                                      where the listener is defined in.
+                                    minLength: 1
+                                    type: string
+                                required:
+                                - name
+                                type: object
+                              name:
+                                description: Name is the name of the listener.
+                                minLength: 1
+                                type: string
+                            required:
+                            - envoyConfig
+                            - name
+                            type: object
                           originatingTLS:
                             description: OriginatingTLS is the TLS context for the
                               connections originated by the L7 proxy.  For egress
@@ -1627,6 +1666,45 @@ spec:
                         description: PortRule is a list of ports/protocol combinations
                           with optional Layer 7 rules which must be met.
                         properties:
+                          listener:
+                            description: listener specifies the name of a custom Envoy
+                              listener to which this traffic should be redirected
+                              to.
+                            properties:
+                              envoyConfig:
+                                description: EnvoyConfig is a reference to the CEC
+                                  or CCNP resource in which the listener is defined.
+                                properties:
+                                  kind:
+                                    description: Kind is the resource type being referred
+                                      to. Defaults to CiliumEnvoyConfig or CiliumClusterwideEnvoyConfig
+                                      for CiliumNetworkPolicy and CiliumClusterwideNetworkPolicy,
+                                      respectively. The only case this is currently
+                                      explicitly needed is when referring to a CiliumClusterwideEnvoyConfig
+                                      from CiliumNetworkPolicy, as using a namespaced
+                                      listener from a cluster scoped policy is not
+                                      allowed.
+                                    enum:
+                                    - CiliumEnvoyConfig
+                                    - CiliumClusterwideEnvoyConfig
+                                    type: string
+                                  name:
+                                    description: Name is the resource name of the
+                                      CiliumEnvoyConfig or CiliumClusterwideEnvoyConfig
+                                      where the listener is defined in.
+                                    minLength: 1
+                                    type: string
+                                required:
+                                - name
+                                type: object
+                              name:
+                                description: Name is the name of the listener.
+                                minLength: 1
+                                type: string
+                            required:
+                            - envoyConfig
+                            - name
+                            type: object
                           originatingTLS:
                             description: OriginatingTLS is the TLS context for the
                               connections originated by the L7 proxy.  For egress
@@ -2715,6 +2793,46 @@ spec:
                           description: PortRule is a list of ports/protocol combinations
                             with optional Layer 7 rules which must be met.
                           properties:
+                            listener:
+                              description: listener specifies the name of a custom
+                                Envoy listener to which this traffic should be redirected
+                                to.
+                              properties:
+                                envoyConfig:
+                                  description: EnvoyConfig is a reference to the CEC
+                                    or CCNP resource in which the listener is defined.
+                                  properties:
+                                    kind:
+                                      description: Kind is the resource type being
+                                        referred to. Defaults to CiliumEnvoyConfig
+                                        or CiliumClusterwideEnvoyConfig for CiliumNetworkPolicy
+                                        and CiliumClusterwideNetworkPolicy, respectively.
+                                        The only case this is currently explicitly
+                                        needed is when referring to a CiliumClusterwideEnvoyConfig
+                                        from CiliumNetworkPolicy, as using a namespaced
+                                        listener from a cluster scoped policy is not
+                                        allowed.
+                                      enum:
+                                      - CiliumEnvoyConfig
+                                      - CiliumClusterwideEnvoyConfig
+                                      type: string
+                                    name:
+                                      description: Name is the resource name of the
+                                        CiliumEnvoyConfig or CiliumClusterwideEnvoyConfig
+                                        where the listener is defined in.
+                                      minLength: 1
+                                      type: string
+                                  required:
+                                  - name
+                                  type: object
+                                name:
+                                  description: Name is the name of the listener.
+                                  minLength: 1
+                                  type: string
+                              required:
+                              - envoyConfig
+                              - name
+                              type: object
                             originatingTLS:
                               description: OriginatingTLS is the TLS context for the
                                 connections originated by the L7 proxy.  For egress
@@ -4018,6 +4136,46 @@ spec:
                           description: PortRule is a list of ports/protocol combinations
                             with optional Layer 7 rules which must be met.
                           properties:
+                            listener:
+                              description: listener specifies the name of a custom
+                                Envoy listener to which this traffic should be redirected
+                                to.
+                              properties:
+                                envoyConfig:
+                                  description: EnvoyConfig is a reference to the CEC
+                                    or CCNP resource in which the listener is defined.
+                                  properties:
+                                    kind:
+                                      description: Kind is the resource type being
+                                        referred to. Defaults to CiliumEnvoyConfig
+                                        or CiliumClusterwideEnvoyConfig for CiliumNetworkPolicy
+                                        and CiliumClusterwideNetworkPolicy, respectively.
+                                        The only case this is currently explicitly
+                                        needed is when referring to a CiliumClusterwideEnvoyConfig
+                                        from CiliumNetworkPolicy, as using a namespaced
+                                        listener from a cluster scoped policy is not
+                                        allowed.
+                                      enum:
+                                      - CiliumEnvoyConfig
+                                      - CiliumClusterwideEnvoyConfig
+                                      type: string
+                                    name:
+                                      description: Name is the resource name of the
+                                        CiliumEnvoyConfig or CiliumClusterwideEnvoyConfig
+                                        where the listener is defined in.
+                                      minLength: 1
+                                      type: string
+                                  required:
+                                  - name
+                                  type: object
+                                name:
+                                  description: Name is the name of the listener.
+                                  minLength: 1
+                                  type: string
+                              required:
+                              - envoyConfig
+                              - name
+                              type: object
                             originatingTLS:
                               description: OriginatingTLS is the TLS context for the
                                 connections originated by the L7 proxy.  For egress
diff --git a/pkg/k8s/watchers/cilium_clusterwide_envoy_config.go b/pkg/k8s/watchers/cilium_clusterwide_envoy_config.go
index 059e5fb003..4a5f793487 100644
--- a/pkg/k8s/watchers/cilium_clusterwide_envoy_config.go
+++ b/pkg/k8s/watchers/cilium_clusterwide_envoy_config.go
@@ -93,6 +93,7 @@ func (k *K8sWatcher) addCiliumClusterwideEnvoyConfig(ccec *cilium_v2.CiliumClust
 		ccec.Spec.Resources,
 		true,
 		k.envoyConfigManager,
+		len(ccec.Spec.Services) > 0,
 	)
 	if err != nil {
 		scopedLog.WithError(err).Warn("Failed to add CiliumClusterwideEnvoyConfig: malformed Envoy config")
@@ -112,6 +113,14 @@ func (k *K8sWatcher) addCiliumClusterwideEnvoyConfig(ccec *cilium_v2.CiliumClust
 		return err
 	}
 
+	if len(resources.Listeners) > 0 {
+		// TODO: Policy does not need to be recomputed for this, but if we do not 'force'
+		// the bpf maps are not updated with the new proxy ports either. Move from the
+		// simple boolean to an enum that can more selectively skip regeneration steps (like
+		// we do for the datapath recompilations already?)
+		k.policyManager.TriggerPolicyUpdates(true, "Envoy Listeners added")
+	}
+
 	scopedLog.Debug("Added CiliumClusterwideEnvoyConfig")
 	return err
 }
@@ -129,6 +138,7 @@ func (k *K8sWatcher) updateCiliumClusterwideEnvoyConfig(oldCCEC *cilium_v2.Ciliu
 		oldCCEC.Spec.Resources,
 		false,
 		k.envoyConfigManager,
+		len(oldCCEC.Spec.Services) > 0,
 	)
 	if err != nil {
 		scopedLog.WithError(err).Warn("Failed to update CiliumClusterwideEnvoyConfig: malformed old Envoy config")
@@ -140,6 +150,7 @@ func (k *K8sWatcher) updateCiliumClusterwideEnvoyConfig(oldCCEC *cilium_v2.Ciliu
 		newCCEC.Spec.Resources,
 		true,
 		k.envoyConfigManager,
+		len(newCCEC.Spec.Services) > 0,
 	)
 	if err != nil {
 		scopedLog.WithError(err).Warn("Failed to update CiliumClusterwideEnvoyConfig: malformed new Envoy config")
@@ -163,6 +174,10 @@ func (k *K8sWatcher) updateCiliumClusterwideEnvoyConfig(oldCCEC *cilium_v2.Ciliu
 		return err
 	}
 
+	if oldResources.ListenersAddedOrDeleted(&newResources) {
+		k.policyManager.TriggerPolicyUpdates(true, "Envoy Listeners added or deleted")
+	}
+
 	scopedLog.Debug("Updated CiliumClusterwideEnvoyConfig")
 	return nil
 }
@@ -180,6 +195,7 @@ func (k *K8sWatcher) deleteCiliumClusterwideEnvoyConfig(ccec *cilium_v2.CiliumCl
 		ccec.Spec.Resources,
 		false,
 		k.envoyConfigManager,
+		len(ccec.Spec.Services) > 0,
 	)
 	if err != nil {
 		scopedLog.WithError(err).Warn("Failed to delete CiliumClusterwideEnvoyConfig: parsing rersource names failed")
@@ -199,6 +215,10 @@ func (k *K8sWatcher) deleteCiliumClusterwideEnvoyConfig(ccec *cilium_v2.CiliumCl
 		return err
 	}
 
+	if len(resources.Listeners) > 0 {
+		k.policyManager.TriggerPolicyUpdates(true, "Envoy Listeners deleted")
+	}
+
 	scopedLog.Debug("Deleted CiliumClusterwideEnvoyConfig")
 	return nil
 }
diff --git a/pkg/k8s/watchers/cilium_envoy_config.go b/pkg/k8s/watchers/cilium_envoy_config.go
index f7df3ec20f..00c8c03499 100644
--- a/pkg/k8s/watchers/cilium_envoy_config.go
+++ b/pkg/k8s/watchers/cilium_envoy_config.go
@@ -97,6 +97,7 @@ func (k *K8sWatcher) addCiliumEnvoyConfig(cec *cilium_v2.CiliumEnvoyConfig) erro
 		cec.Spec.Resources,
 		true,
 		k.envoyConfigManager,
+		len(cec.Spec.Services) > 0,
 	)
 	if err != nil {
 		scopedLog.WithError(err).Warn("Failed to add CiliumEnvoyConfig: malformed Envoy config")
@@ -116,6 +117,14 @@ func (k *K8sWatcher) addCiliumEnvoyConfig(cec *cilium_v2.CiliumEnvoyConfig) erro
 		return err
 	}
 
+	if len(resources.Listeners) > 0 {
+		// TODO: Policy does not need to be recomputed for this, but if we do not 'force'
+		// the bpf maps are not updated with the new proxy ports either. Move from the
+		// simple boolean to an enum that can more selectively skip regeneration steps (like
+		// we do for the datapath recompilations already?)
+		k.policyManager.TriggerPolicyUpdates(true, "Envoy Listeners added")
+	}
+
 	scopedLog.Debug("Added CiliumEnvoyConfig")
 	return err
 }
@@ -191,6 +200,7 @@ func (k *K8sWatcher) updateCiliumEnvoyConfig(oldCEC *cilium_v2.CiliumEnvoyConfig
 		oldCEC.Spec.Resources,
 		false,
 		k.envoyConfigManager,
+		len(oldCEC.Spec.Services) > 0,
 	)
 	if err != nil {
 		scopedLog.WithError(err).Warn("Failed to update CiliumEnvoyConfig: malformed old Envoy config")
@@ -202,6 +212,7 @@ func (k *K8sWatcher) updateCiliumEnvoyConfig(oldCEC *cilium_v2.CiliumEnvoyConfig
 		newCEC.Spec.Resources,
 		true,
 		k.envoyConfigManager,
+		len(newCEC.Spec.Services) > 0,
 	)
 	if err != nil {
 		scopedLog.WithError(err).Warn("Failed to update CiliumEnvoyConfig: malformed new Envoy config")
@@ -226,6 +237,10 @@ func (k *K8sWatcher) updateCiliumEnvoyConfig(oldCEC *cilium_v2.CiliumEnvoyConfig
 		return err
 	}
 
+	if oldResources.ListenersAddedOrDeleted(&newResources) {
+		k.policyManager.TriggerPolicyUpdates(true, "Envoy Listeners added or deleted")
+	}
+
 	scopedLog.Debug("Updated CiliumEnvoyConfig")
 	return nil
 }
@@ -302,6 +317,7 @@ func (k *K8sWatcher) deleteCiliumEnvoyConfig(cec *cilium_v2.CiliumEnvoyConfig) e
 		cec.Spec.Resources,
 		false,
 		k.envoyConfigManager,
+		len(cec.Spec.Services) > 0,
 	)
 	if err != nil {
 		scopedLog.WithError(err).Warn("Failed to delete CiliumEnvoyConfig: parsing rersource names failed")
@@ -321,6 +337,10 @@ func (k *K8sWatcher) deleteCiliumEnvoyConfig(cec *cilium_v2.CiliumEnvoyConfig) e
 		return err
 	}
 
+	if len(resources.Listeners) > 0 {
+		k.policyManager.TriggerPolicyUpdates(true, "Envoy Listeners deleted")
+	}
+
 	scopedLog.Debug("Deleted CiliumEnvoyConfig")
 	return nil
 }
diff --git a/pkg/logging/logfields/logfields.go b/pkg/logging/logfields/logfields.go
index 46742d6cb2..3a68373504 100644
--- a/pkg/logging/logfields/logfields.go
+++ b/pkg/logging/logfields/logfields.go
@@ -316,6 +316,9 @@ const (
 	// CiliumEnvoyConfigName is the name of a CiliumEnvoyConfig
 	CiliumEnvoyConfigName = "ciliumEnvoyConfigName"
 
+	// Listener is the name of an Envoy Listener defined in CEC or CCEC
+	Listener = "listener"
+
 	// BPFMapKey is a key from a BPF map
 	BPFMapKey = "bpfMapKey"
 
@@ -634,4 +637,7 @@ const (
 
 	// CGroupId is the numerical cgroup id
 	CGroupID = "cgroupID"
+
+	// Expected is an expected value
+	Expected = "expected"
 )
diff --git a/pkg/option/config.go b/pkg/option/config.go
index ccc683b648..27c34cc6cd 100644
--- a/pkg/option/config.go
+++ b/pkg/option/config.go
@@ -481,6 +481,14 @@ const (
 	// been reached.
 	DNSProxyConcurrencyProcessingGracePeriod = "dnsproxy-concurrency-processing-grace-period"
 
+	// DNSProxyLockCount is the array size containing mutexes which protect
+	// against parallel handling of DNS response IPs.
+	DNSProxyLockCount = "dnsproxy-lock-count"
+
+	// DNSProxyLockTimeout is timeout when acquiring the locks controlled by
+	// DNSProxyLockCount.
+	DNSProxyLockTimeout = "dnsproxy-lock-timeout"
+
 	// MTUName is the name of the MTU option
 	MTUName = "mtu"
 
@@ -1712,6 +1720,14 @@ type DaemonConfig struct {
 	// been reached.
 	DNSProxyConcurrencyProcessingGracePeriod time.Duration
 
+	// DNSProxyLockCount is the array size containing mutexes which protect
+	// against parallel handling of DNS response IPs.
+	DNSProxyLockCount int
+
+	// DNSProxyLockTimeout is timeout when acquiring the locks controlled by
+	// DNSProxyLockCount.
+	DNSProxyLockTimeout time.Duration
+
 	// EnableXTSocketFallback allows disabling of kernel's ip_early_demux
 	// sysctl option if `xt_socket` kernel module is not available.
 	EnableXTSocketFallback bool
@@ -3089,6 +3105,8 @@ func (c *DaemonConfig) Populate(vp *viper.Viper) {
 	c.FQDNProxyResponseMaxDelay = vp.GetDuration(FQDNProxyResponseMaxDelay)
 	c.DNSProxyConcurrencyLimit = vp.GetInt(DNSProxyConcurrencyLimit)
 	c.DNSProxyConcurrencyProcessingGracePeriod = vp.GetDuration(DNSProxyConcurrencyProcessingGracePeriod)
+	c.DNSProxyLockCount = vp.GetInt(DNSProxyLockCount)
+	c.DNSProxyLockTimeout = vp.GetDuration(DNSProxyLockTimeout)
 
 	// Convert IP strings into net.IPNet types
 	subnets, invalid := ip.ParseCIDRs(vp.GetStringSlice(IPv4PodSubnets))
diff --git a/pkg/policy/api/l4.go b/pkg/policy/api/l4.go
index 51cb415825..023d8e4e82 100644
--- a/pkg/policy/api/l4.go
+++ b/pkg/policy/api/l4.go
@@ -112,6 +112,41 @@ type TLSContext struct {
 	PrivateKey string `json:"privateKey,omitempty"`
 }
 
+// EnvoyConfig defines a reference to a CiliumEnvoyConfig or CiliumClusterwideEnvoyConfig
+type EnvoyConfig struct {
+	// Kind is the resource type being referred to. Defaults to CiliumEnvoyConfig or
+	// CiliumClusterwideEnvoyConfig for CiliumNetworkPolicy and CiliumClusterwideNetworkPolicy,
+	// respectively. The only case this is currently explicitly needed is when referring to a
+	// CiliumClusterwideEnvoyConfig from CiliumNetworkPolicy, as using a namespaced listener
+	// from a cluster scoped policy is not allowed.
+	//
+	// +kubebuilder:validation:Enum=CiliumEnvoyConfig;CiliumClusterwideEnvoyConfig
+	// +kubebuilder:validation:Optional
+	Kind string `json:"kind"`
+
+	// Name is the resource name of the CiliumEnvoyConfig or CiliumClusterwideEnvoyConfig where
+	// the listener is defined in.
+	//
+	// +kubebuilder:validation:MinLength=1
+	// +kubebuilder:validation:Required
+	Name string `json:"name"`
+}
+
+// Listener defines a reference to an Envoy listener specified in a CEC or CCEC resource.
+type Listener struct {
+	// EnvoyConfig is a reference to the CEC or CCNP resource in which
+	// the listener is defined.
+	//
+	// +kubebuilder:validation:Required
+	EnvoyConfig *EnvoyConfig `json:"envoyConfig"`
+
+	// Name is the name of the listener.
+	//
+	// +kubebuilder:validation:MinLength=1
+	// +kubebuilder:validation:Required
+	Name string `json:"name"`
+}
+
 // PortRule is a list of ports/protocol combinations with optional Layer 7
 // rules which must be met.
 type PortRule struct {
@@ -147,6 +182,12 @@ type PortRule struct {
 	// +kubebuilder:validation:Optional
 	ServerNames []string `json:"serverNames,omitempty"`
 
+	// listener specifies the name of a custom Envoy listener to which this traffic should be
+	// redirected to.
+	//
+	// +kubebuilder:validation:Optional
+	Listener *Listener `json:"listener,omitempty"`
+
 	// Rules is a list of additional port level rules which must be met in
 	// order for the PortRule to allow the traffic. If omitted or empty,
 	// no layer 7 rules are enforced.
diff --git a/pkg/policy/api/rule_validation.go b/pkg/policy/api/rule_validation.go
index a1ea6b608b..25fa9e417f 100644
--- a/pkg/policy/api/rule_validation.go
+++ b/pkg/policy/api/rule_validation.go
@@ -381,6 +381,23 @@ func (pr *PortRule) sanitize(ingress bool) error {
 		}
 	}
 
+	listener := pr.Listener
+	if listener != nil {
+		// For now we have only tested custom listener support on the egress path.  TODO
+		// (jrajahalme): Lift this limitation in follow-up work once proper testing has been
+		// done on the ingress path.
+		if ingress {
+			return fmt.Errorf("Listener is not allowed on ingress (%s)", listener.Name)
+		}
+		// There is no quarantee that Listener will support Cilium policy enforcement.  Even
+		// now proxylib-based enforcement (e.g, Kafka) may work, but has not been tested.
+		// TODO (jrajahalme): Lift this limitation in follow-up work for proxylib based
+		// parsers if needed and when tested.
+		if !pr.Rules.IsEmpty() {
+			return fmt.Errorf("Listener is not allowed with L7 rules (%s)", listener.Name)
+		}
+	}
+
 	// Sanitize L7 rules
 	if !pr.Rules.IsEmpty() {
 		if haveZeroPort {
diff --git a/pkg/policy/api/utils.go b/pkg/policy/api/utils.go
index 5ce5523f15..780985c046 100644
--- a/pkg/policy/api/utils.go
+++ b/pkg/policy/api/utils.go
@@ -119,3 +119,25 @@ func ParseL4Proto(proto string) (L4Proto, error) {
 	p := L4Proto(strings.ToUpper(proto))
 	return p, p.Validate()
 }
+
+// ResourceQualifiedName returns the qualified name of an Envoy resource,
+// prepending CEC namespace and CEC name to the resource name and using
+// '/' as a separator.
+//
+// In case of an empty CEC namespace or an empty CEC name, leading separators
+// are stripped away.
+func ResourceQualifiedName(namespace, name, resource string) string {
+	var sb strings.Builder
+
+	if namespace != "" {
+		sb.WriteString(namespace)
+		sb.WriteRune('/')
+	}
+	if name != "" {
+		sb.WriteString(name)
+		sb.WriteRune('/')
+	}
+	sb.WriteString(resource)
+
+	return sb.String()
+}
diff --git a/pkg/policy/api/zz_generated.deepcopy.go b/pkg/policy/api/zz_generated.deepcopy.go
index 49e2d6bce8..836bc859e6 100644
--- a/pkg/policy/api/zz_generated.deepcopy.go
+++ b/pkg/policy/api/zz_generated.deepcopy.go
@@ -319,6 +319,22 @@ func (in EntitySlice) DeepCopy() EntitySlice {
 	return *out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *EnvoyConfig) DeepCopyInto(out *EnvoyConfig) {
+	*out = *in
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new EnvoyConfig.
+func (in *EnvoyConfig) DeepCopy() *EnvoyConfig {
+	if in == nil {
+		return nil
+	}
+	out := new(EnvoyConfig)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *FQDNSelector) DeepCopyInto(out *FQDNSelector) {
 	*out = *in
@@ -630,6 +646,27 @@ func (in *L7Rules) DeepCopy() *L7Rules {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *Listener) DeepCopyInto(out *Listener) {
+	*out = *in
+	if in.EnvoyConfig != nil {
+		in, out := &in.EnvoyConfig, &out.EnvoyConfig
+		*out = new(EnvoyConfig)
+		**out = **in
+	}
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new Listener.
+func (in *Listener) DeepCopy() *Listener {
+	if in == nil {
+		return nil
+	}
+	out := new(Listener)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *PortDenyRule) DeepCopyInto(out *PortDenyRule) {
 	*out = *in
@@ -712,6 +749,11 @@ func (in *PortRule) DeepCopyInto(out *PortRule) {
 		*out = make([]string, len(*in))
 		copy(*out, *in)
 	}
+	if in.Listener != nil {
+		in, out := &in.Listener, &out.Listener
+		*out = new(Listener)
+		(*in).DeepCopyInto(*out)
+	}
 	if in.Rules != nil {
 		in, out := &in.Rules, &out.Rules
 		*out = new(L7Rules)
diff --git a/pkg/policy/api/zz_generated.deepequal.go b/pkg/policy/api/zz_generated.deepequal.go
index f140bce79e..4a405f6488 100644
--- a/pkg/policy/api/zz_generated.deepequal.go
+++ b/pkg/policy/api/zz_generated.deepequal.go
@@ -460,6 +460,23 @@ func (in *EntitySlice) DeepEqual(other *EntitySlice) bool {
 	return true
 }
 
+// DeepEqual is an autogenerated deepequal function, deeply comparing the
+// receiver with other. in must be non-nil.
+func (in *EnvoyConfig) DeepEqual(other *EnvoyConfig) bool {
+	if other == nil {
+		return false
+	}
+
+	if in.Kind != other.Kind {
+		return false
+	}
+	if in.Name != other.Name {
+		return false
+	}
+
+	return true
+}
+
 // DeepEqual is an autogenerated deepequal function, deeply comparing the
 // receiver with other. in must be non-nil.
 func (in *FQDNSelector) DeepEqual(other *FQDNSelector) bool {
@@ -853,6 +870,28 @@ func (in *L7Rules) DeepEqual(other *L7Rules) bool {
 	return true
 }
 
+// DeepEqual is an autogenerated deepequal function, deeply comparing the
+// receiver with other. in must be non-nil.
+func (in *Listener) DeepEqual(other *Listener) bool {
+	if other == nil {
+		return false
+	}
+
+	if (in.EnvoyConfig == nil) != (other.EnvoyConfig == nil) {
+		return false
+	} else if in.EnvoyConfig != nil {
+		if !in.EnvoyConfig.DeepEqual(other.EnvoyConfig) {
+			return false
+		}
+	}
+
+	if in.Name != other.Name {
+		return false
+	}
+
+	return true
+}
+
 // DeepEqual is an autogenerated deepequal function, deeply comparing the
 // receiver with other. in must be non-nil.
 func (in *PortDenyRule) DeepEqual(other *PortDenyRule) bool {
@@ -974,6 +1013,14 @@ func (in *PortRule) DeepEqual(other *PortRule) bool {
 		}
 	}
 
+	if (in.Listener == nil) != (other.Listener == nil) {
+		return false
+	} else if in.Listener != nil {
+		if !in.Listener.DeepEqual(other.Listener) {
+			return false
+		}
+	}
+
 	if (in.Rules == nil) != (other.Rules == nil) {
 		return false
 	} else if in.Rules != nil {
diff --git a/pkg/policy/l4.go b/pkg/policy/l4.go
index e105ac9846..d444a0c64a 100644
--- a/pkg/policy/l4.go
+++ b/pkg/policy/l4.go
@@ -183,6 +183,9 @@ type PerSelectorPolicy struct {
 	// TLS handshake.
 	ServerNames StringSet `json:"serverNames,omitempty"`
 
+	// isRedirect is 'true' when traffic must be redirected
+	isRedirect bool `json:"-"`
+
 	// Pre-computed HTTP rules, computed after rule merging is complete
 	EnvoyHTTPRules *cilium.HttpNetworkPolicyRules `json:"-"`
 
@@ -202,22 +205,14 @@ func (a *PerSelectorPolicy) Equal(b *PerSelectorPolicy) bool {
 		a.TerminatingTLS.Equal(b.TerminatingTLS) &&
 		a.OriginatingTLS.Equal(b.OriginatingTLS) &&
 		a.ServerNames.Equal(b.ServerNames) &&
+		a.isRedirect == b.isRedirect &&
 		a.IsDeny == b.IsDeny &&
 		a.L7Rules.DeepEqual(&b.L7Rules)
 }
 
 // IsRedirect returns true if the L7Rules are a redirect.
 func (a *PerSelectorPolicy) IsRedirect() bool {
-	// policy is for an proxy redirect if it:
-	// - is not empty, i.e., has L7 rules or TLS config, and
-	// - it is not a deny policy, as deny policies do not support L7 rules
-	return !a.IsEmpty() && !a.IsDeny
-}
-
-// IsEmpty returns whether the `L7Rules` is nil or has no L7 rules and no TLS config.
-func (a *PerSelectorPolicy) IsEmpty() bool {
-	return a == nil ||
-		(a.L7Rules.IsEmpty() && a.TerminatingTLS == nil && a.OriginatingTLS == nil && len(a.ServerNames) > 0)
+	return a != nil && a.isRedirect
 }
 
 // HasL7Rules returns whether the `L7Rules` contains any L7 rules.
@@ -286,9 +281,13 @@ func (l7 L7ParserType) String() string {
 const (
 	// ParserTypeNone represents the case where no parser type is provided.
 	ParserTypeNone L7ParserType = ""
-	// ParserTypeTLS is used when TLS origination, termination, or SNI filtering is used
-	// without any L7 parsing. If TLS policies are used with HTTP, ParserTypeHTTP is used.
+	// ParserTypeTLS is used for TLS origination, termination, or SNI filtering without any L7
+	// parsing. If TLS policies are used with HTTP rules, ParserTypeHTTP is used instead.
 	ParserTypeTLS L7ParserType = "tls"
+	// ParserTypeCRD is used with a custom CiliumEnvoyConfig redirection. Incompatible with any
+	// parser type with L7 enforcement (HTTP, Kafka, proxylib), as the custom Listener generally
+	// does not support them.
+	ParserTypeCRD L7ParserType = "crd"
 	// ParserTypeHTTP specifies a HTTP parser type
 	ParserTypeHTTP L7ParserType = "http"
 	// ParserTypeKafka specifies a Kafka parser type
@@ -297,15 +296,31 @@ const (
 	ParserTypeDNS L7ParserType = "dns"
 )
 
+// redirectTypes is a bitmask of redirection types of multiple filters
+type redirectTypes uint16
+
+const (
+	// redirectTypeDNS bit is set when policy contains a redirection to DNS proxy
+	redirectTypeDNS redirectTypes = 1 << iota
+	// redirectTypeEnvoy bit is set when policy contains a redirection to Envoy
+	redirectTypeEnvoy
+	// redirectTypeProxylib bits are set when policy contains a redirection to Proxylib (via
+	// Envoy)
+	redirectTypeProxylib redirectTypes = 1<<iota | redirectTypeEnvoy
+
+	// redirectTypeNone represents the case where there is no proxy redirect
+	redirectTypeNone redirectTypes = redirectTypes(0)
+)
+
 func (from L7ParserType) canPromoteTo(to L7ParserType) bool {
 	switch from {
 	case ParserTypeNone:
 		// ParserTypeNone can be promoted to any other type
 		return true
 	case ParserTypeTLS:
-		// ParserTypeTLS can be promoted to any other type, except for DNS,
+		// ParserTypeTLS can be promoted to any other type, except for DNS or CRD,
 		// but ParserTypeTLS can not be demoted to ParserTypeNone
-		if to != ParserTypeNone && to != ParserTypeDNS {
+		if to != ParserTypeNone && to != ParserTypeDNS && to != ParserTypeCRD {
 			return true
 		}
 	}
@@ -353,6 +368,9 @@ type L4Filter struct {
 	// L7Parser specifies the L7 protocol parser (optional). If specified as
 	// an empty string, then means that no L7 proxy redirect is performed.
 	L7Parser L7ParserType `json:"-"`
+	// Listener is an optional fully qualified name of a Envoy Listner defined in a CiliumEnvoyConfig CRD that should be
+	// used for this traffic instead of the default listener
+	Listener string `json:"listener,omitempty"`
 	// Ingress is true if filter applies at ingress; false if it applies at egress.
 	Ingress bool `json:"-"`
 	// The rule labels of this Filter
@@ -395,6 +413,11 @@ func (l4 *L4Filter) GetPort() uint16 {
 	return uint16(l4.Port)
 }
 
+// GetListener returns the optional listener name.
+func (l4 *L4Filter) GetListener() string {
+	return l4.Listener
+}
+
 // ToMapState converts filter into a MapState with two possible values:
 //
 //	Entry with ProxyPort = 0: No proxy redirection is needed for this key
@@ -574,12 +597,13 @@ func (l4 *L4Filter) cacheFQDNSelector(sel api.FQDNSelector, selectorCache *Selec
 
 // add L7 rules for all endpoints in the L7DataMap
 
-func (l7 L7DataMap) addRulesForEndpoints(rules *api.L7Rules, terminatingTLS, originatingTLS *TLSContext, deny bool, sni []string) {
+func (l7 L7DataMap) addRulesForEndpoints(rules *api.L7Rules, terminatingTLS, originatingTLS *TLSContext, deny bool, sni []string, forceRedirect bool) {
 	l7policy := &PerSelectorPolicy{
 		TerminatingTLS: terminatingTLS,
 		OriginatingTLS: originatingTLS,
 		IsDeny:         deny,
 		ServerNames:    NewStringSet(sni),
+		isRedirect:     !deny && (forceRedirect || terminatingTLS != nil || originatingTLS != nil || len(sni) > 0 || !rules.IsEmpty()),
 	}
 	if rules != nil {
 		l7policy.L7Rules = *rules
@@ -678,6 +702,7 @@ func createL4Filter(policyCtx PolicyContext, peerEndpoints api.EndpointSelectorS
 		if err != nil {
 			return nil, err
 		}
+
 		// Set parser type to TLS, if TLS. This will be overridden by L7 below, if rules
 		// exists.
 		if terminatingTLS != nil || originatingTLS != nil || len(pr.ServerNames) > 0 {
@@ -702,8 +727,34 @@ func createL4Filter(policyCtx PolicyContext, peerEndpoints api.EndpointSelectorS
 			}
 		}
 
+		// Override the parser type to CRD is applicable.
+		forceRedirect := false
+		if pr.Listener != nil {
+			l4.L7Parser = ParserTypeCRD
+			ns := policyCtx.GetNamespace()
+			resource := pr.Listener.EnvoyConfig
+			switch resource.Kind {
+			case "CiliumEnvoyConfig":
+				if ns == "" {
+					// Cluster-scoped CCNP tries to use namespaced
+					// CiliumEnvoyConfig
+					//
+					// TODO: Catch this in rule validation once we have a
+					// validation context in there so that we can differentiate
+					// between CNP and CCNP at validation time.
+					return nil, fmt.Errorf("Listener %q in CCNP can not use Kind CiliumEnvoyConfig", pr.Listener.Name)
+				}
+			case "CiliumClusterwideEnvoyConfig":
+				// CNP refers to a cluster-scoped listener
+				ns = ""
+			default:
+			}
+			l4.Listener = api.ResourceQualifiedName(ns, resource.Name, pr.Listener.Name)
+			forceRedirect = true
+		}
+
 		if l4.L7Parser != ParserTypeNone {
-			l4.L7RulesPerSelector.addRulesForEndpoints(pr.Rules, terminatingTLS, originatingTLS, policyCtx.IsDeny(), pr.ServerNames)
+			l4.L7RulesPerSelector.addRulesForEndpoints(pr.Rules, terminatingTLS, originatingTLS, policyCtx.IsDeny(), pr.ServerNames, forceRedirect)
 		}
 	}
 
@@ -761,15 +812,11 @@ func createL4IngressFilter(policyCtx PolicyContext, fromEndpoints api.EndpointSe
 		return nil, err
 	}
 
-	pr := rule.GetPortRule()
-	if pr == nil {
-		return filter, nil
-	}
-	// If the filter would apply L7 rules for the Host, when we should accept everything from host,
-	// then wildcard Host at L7.
-	if !pr.Rules.IsEmpty() && len(hostWildcardL7) > 0 {
-		for cs := range filter.L7RulesPerSelector {
-			if cs.Selects(identity.ReservedIdentityHost) {
+	// If the filter would apply proxy redirection for the Host, when we should accept
+	// everything from host, then wildcard Host at L7.
+	if len(hostWildcardL7) > 0 {
+		for cs, l7 := range filter.L7RulesPerSelector {
+			if l7.IsRedirect() && cs.Selects(identity.ReservedIdentityHost) {
 				for _, name := range hostWildcardL7 {
 					selector := api.ReservedEndpointSelectors[name]
 					filter.cacheIdentitySelector(selector, policyCtx.GetSelectorCache(), policyCtx.IsDeny())
@@ -791,21 +838,26 @@ func createL4EgressFilter(policyCtx PolicyContext, toEndpoints api.EndpointSelec
 	return createL4Filter(policyCtx, toEndpoints, rule, port, protocol, ruleLabels, false, fqdns)
 }
 
+// redirectType returns the redirectType for this filter
+func (l4 *L4Filter) redirectType() redirectTypes {
+	switch l4.L7Parser {
+	case ParserTypeNone:
+		return redirectTypeNone
+	case ParserTypeDNS:
+		return redirectTypeDNS
+	case ParserTypeHTTP, ParserTypeTLS, ParserTypeCRD:
+		return redirectTypeEnvoy
+	default:
+		// all other (non-empty) values are used for proxylib redirects
+		return redirectTypeProxylib
+	}
+}
+
 // IsRedirect returns true if the L4 filter contains a port redirection
 func (l4 *L4Filter) IsRedirect() bool {
 	return l4.L7Parser != ParserTypeNone
 }
 
-// IsEnvoyRedirect returns true if the L4 filter contains a port redirected to Envoy
-func (l4 *L4Filter) IsEnvoyRedirect() bool {
-	return l4.IsRedirect() && l4.L7Parser != ParserTypeDNS
-}
-
-// IsProxylibRedirect returns true if the L4 filter contains a port redirected to Proxylib (via Envoy)
-func (l4 *L4Filter) IsProxylibRedirect() bool {
-	return l4.IsEnvoyRedirect() && l4.L7Parser != ParserTypeHTTP && l4.L7Parser != ParserTypeTLS
-}
-
 // Marshal returns the `L4Filter` in a JSON string.
 func (l4 *L4Filter) Marshal() string {
 	b, err := json.Marshal(l4)
@@ -898,43 +950,14 @@ func (l4 L4PolicyMap) Detach(selectorCache *SelectorCache) {
 
 // Attach makes all the L4Filters to point back to the L4Policy that contains them.
 // This is done before the L4PolicyMap is exposed to concurrent access.
-func (l4 L4PolicyMap) Attach(ctx PolicyContext, l4Policy *L4Policy) {
+// Returns the bitmask of all redirect types for this policymap.
+func (l4 L4PolicyMap) attach(ctx PolicyContext, l4Policy *L4Policy) redirectTypes {
+	var redirectTypes redirectTypes
 	for _, f := range l4 {
 		f.attach(ctx, l4Policy)
+		redirectTypes |= f.redirectType()
 	}
-}
-
-// HasRedirect returns true if at least one L4 filter contains a port
-// redirection
-func (l4 L4PolicyMap) HasRedirect() bool {
-	for _, f := range l4 {
-		if f.IsRedirect() {
-			return true
-		}
-	}
-	return false
-}
-
-// HasEnvoyRedirect returns true if at least one L4 filter contains a port
-// redirection that is forwarded to Envoy
-func (l4 L4PolicyMap) HasEnvoyRedirect() bool {
-	for _, f := range l4 {
-		if f.IsEnvoyRedirect() {
-			return true
-		}
-	}
-	return false
-}
-
-// HasProxylibRedirect returns true if at least one L4 filter contains a port
-// redirection that is forwarded to Proxylib (via Envoy)
-func (l4 L4PolicyMap) HasProxylibRedirect() bool {
-	for _, f := range l4 {
-		if f.IsProxylibRedirect() {
-			return true
-		}
-	}
-	return false
+	return redirectTypes
 }
 
 // containsAllL3L4 checks if the L4PolicyMap contains all L4 ports in `ports`.
@@ -1017,6 +1040,11 @@ type L4Policy struct {
 	// Revision is the repository revision used to generate this policy.
 	Revision uint64
 
+	// redirectTypes is a bitmap containing the types of redirect contained by this policy.  It
+	// is computed after the policy maps to avoid scanning them repeatedly when using the
+	// L4Policy
+	redirectTypes redirectTypes
+
 	// Endpoint policies using this L4Policy
 	// These are circular references, cleaned up in Detach()
 	// This mutex is taken while Endpoint mutex is held, so Endpoint lock
@@ -1118,8 +1146,9 @@ func (l4 *L4Policy) Detach(selectorCache *SelectorCache) {
 // Attach makes all the L4Filters to point back to the L4Policy that contains them.
 // This is done before the L4Policy is exposed to concurrent access.
 func (l4 *L4Policy) Attach(ctx PolicyContext) {
-	l4.Ingress.Attach(ctx, l4)
-	l4.Egress.Attach(ctx, l4)
+	ingressRedirects := l4.Ingress.attach(ctx, l4)
+	egressRedirects := l4.Egress.attach(ctx, l4)
+	l4.redirectTypes = ingressRedirects | egressRedirects
 }
 
 // IngressCoversContext checks if the receiver's ingress L4Policy contains
@@ -1140,17 +1169,17 @@ func (l4 *L4PolicyMap) EgressCoversContext(ctx *SearchContext) api.Decision {
 
 // HasRedirect returns true if the L4 policy contains at least one port redirection
 func (l4 *L4Policy) HasRedirect() bool {
-	return l4 != nil && (l4.Ingress.HasRedirect() || l4.Egress.HasRedirect())
+	return l4 != nil && l4.redirectTypes != redirectTypeNone
 }
 
 // HasEnvoyRedirect returns true if the L4 policy contains at least one port redirection to Envoy
 func (l4 *L4Policy) HasEnvoyRedirect() bool {
-	return l4 != nil && (l4.Ingress.HasEnvoyRedirect() || l4.Egress.HasEnvoyRedirect())
+	return l4 != nil && l4.redirectTypes&redirectTypeEnvoy == redirectTypeEnvoy
 }
 
 // HasProxylibRedirect returns true if the L4 policy contains at least one port redirection to Proxylib
 func (l4 *L4Policy) HasProxylibRedirect() bool {
-	return l4 != nil && (l4.Ingress.HasProxylibRedirect() || l4.Egress.HasProxylibRedirect())
+	return l4 != nil && l4.redirectTypes&redirectTypeProxylib == redirectTypeProxylib
 }
 
 func (l4 *L4Policy) GetModel() *models.L4Policy {
@@ -1187,4 +1216,5 @@ type ProxyPolicy interface {
 	GetL7Parser() L7ParserType
 	GetIngress() bool
 	GetPort() uint16
+	GetListener() string
 }
diff --git a/pkg/policy/repository.go b/pkg/policy/repository.go
index 73e1266595..b0bddc74db 100644
--- a/pkg/policy/repository.go
+++ b/pkg/policy/repository.go
@@ -34,6 +34,9 @@ type CertificateManager interface {
 // PolicyContext is an interface policy resolution functions use to access the Repository.
 // This way testing code can run without mocking a full Repository.
 type PolicyContext interface {
+	// return the namespace in which the policy rule is being resolved
+	GetNamespace() string
+
 	// return the SelectorCache
 	GetSelectorCache() *SelectorCache
 
@@ -68,6 +71,11 @@ type policyContext struct {
 	isDeny bool
 }
 
+// GetNamespace() returns the namespace for the policy rule being resolved
+func (p *policyContext) GetNamespace() string {
+	return p.ns
+}
+
 // GetSelectorCache() returns the selector cache used by the Repository
 func (p *policyContext) GetSelectorCache() *SelectorCache {
 	return p.repo.GetSelectorCache()
diff --git a/pkg/policy/rule.go b/pkg/policy/rule.go
index 87751d9061..eb0b687aef 100644
--- a/pkg/policy/rule.go
+++ b/pkg/policy/rule.go
@@ -106,6 +106,15 @@ func mergePortProto(ctx *SearchContext, existingFilter, filterToMerge *L4Filter,
 		return err
 	}
 
+	if existingFilter.Listener == "" || filterToMerge.Listener == "" {
+		if filterToMerge.Listener != "" {
+			existingFilter.Listener = filterToMerge.Listener
+		}
+	} else if filterToMerge.Listener != existingFilter.Listener {
+		ctx.PolicyTrace("   Merge conflict: mismatching CiliumEnvoyConfig listeners %v/%v\n", filterToMerge.Listener, existingFilter.Listener)
+		return fmt.Errorf("cannot merge conflicting CiliumEnvoyConfig Listeners (%v/%v)", filterToMerge.Listener, existingFilter.Listener)
+	}
+
 	for cs, newL7Rules := range filterToMerge.L7RulesPerSelector {
 		// 'cs' will be merged or moved (see below), either way it needs
 		// to be removed from the map it is in now.
@@ -148,6 +157,9 @@ func mergePortProto(ctx *SearchContext, existingFilter, filterToMerge *L4Filter,
 				newL7Rules = &PerSelectorPolicy{}
 			}
 
+			// Merge isRedirect flag
+			l7Rules.isRedirect = l7Rules.isRedirect || newL7Rules.isRedirect
+
 			if l7Rules.TerminatingTLS == nil || newL7Rules.TerminatingTLS == nil {
 				if newL7Rules.TerminatingTLS != nil {
 					l7Rules.TerminatingTLS = newL7Rules.TerminatingTLS
diff --git a/pkg/policy/visibility.go b/pkg/policy/visibility.go
index 734e2c94bc..da2600b042 100644
--- a/pkg/policy/visibility.go
+++ b/pkg/policy/visibility.go
@@ -211,3 +211,8 @@ func (v *VisibilityMetadata) GetIngress() bool {
 func (v *VisibilityMetadata) GetPort() uint16 {
 	return v.Port
 }
+
+// GetListener returns the optional listener name.
+func (l4 *VisibilityMetadata) GetListener() string {
+	return ""
+}
diff --git a/pkg/proxy/proxy.go b/pkg/proxy/proxy.go
index 0d9736820f..178cc08d90 100644
--- a/pkg/proxy/proxy.go
+++ b/pkg/proxy/proxy.go
@@ -283,7 +283,7 @@ func (p *Proxy) AckProxyPort(ctx context.Context, name string) error {
 // to keep the use count up-to-date.
 // Must be called with proxyPortsMutex held!
 func (p *Proxy) ackProxyPort(ctx context.Context, name string, pp *ProxyPort) error {
-	scopedLog := log.WithField("proxy port name", name)
+	scopedLog := log.WithField(fieldProxyRedirectID, name)
 
 	// Datapath rules are added only after we know the proxy configuration
 	// with the actual port number has succeeded. Deletion of the rules
@@ -303,6 +303,7 @@ func (p *Proxy) ackProxyPort(ctx context.Context, name string, pp *ProxyPort) er
 		pp.rulesPort = pp.proxyPort
 	}
 	pp.nRedirects++
+	scopedLog.Debugf("AckProxyPort: acked proxy port %d (%v)", pp.proxyPort, *pp)
 	return nil
 }
 
@@ -320,6 +321,8 @@ func (p *Proxy) releaseProxyPort(name string) error {
 			return fmt.Errorf("Can't release proxy port: proxy %s on %d has a static listener", name, pp.proxyPort)
 		}
 
+		log.WithField(fieldProxyRedirectID, name).Debugf("Delayed release of proxy port %d", pp.proxyPort)
+
 		// Allow the port to be reallocated for other use if needed.
 		allocatedPorts[pp.proxyPort] = false
 		pp.proxyPort = 0
@@ -329,24 +332,30 @@ func (p *Proxy) releaseProxyPort(name string) error {
 		// Leave the datapath rules behind on the hope that they get reused later.
 		// This becomes possible when we are able to keep the proxy listeners
 		// configured also when there are no redirects.
-		log.WithField(fieldProxyRedirectID, name).Debugf("Delayed release of proxy port %d", pp.proxyPort)
 	}
 
 	return nil
 }
 
-// findProxyPortByType returns a ProxyPort matching the given type and direction, if found.
-// This is mainly useful for getting a statically defined proxy port. Dynamic types
-// may have multiple instances and any one of them can be returned.
-// Must be called with proxyPortsMutex held!
-func findProxyPortByType(l7Type ProxyType, ingress bool) (string, *ProxyPort) {
+// findProxyPortByType returns a ProxyPort matching the given type, listener name, and direction, if
+// found.  Must be called with proxyPortsMutex held!
+func findProxyPortByType(l7Type ProxyType, listener string, ingress bool) (string, *ProxyPort) {
 	portType := l7Type
 	switch l7Type {
-	case ProxyTypeDNS, ProxyTypeHTTP, ProxyTypeCRD:
+	case ProxyTypeCRD:
+		// CRD proxy ports are dynamically created, look up by name
+		if pp, ok := proxyPorts[listener]; ok && pp.proxyType == ProxyTypeCRD {
+			return listener, pp
+		}
+		log.Debugf("findProxyPortByType: can not find crd listener %s from %v", listener, proxyPorts)
+		return "", nil
+	case ProxyTypeDNS, ProxyTypeHTTP:
+		// Look up by the given type
 	default:
 		// "Unknown" parsers are assumed to be Proxylib (TCP) parsers, which
 		// is registered with an empty string.
-		// This works also for explicit TCP and TLS parser types.
+		// This works also for explicit TCP and TLS parser types, which are backed by the
+		// TCP Proxy filter chain.
 		portType = ProxyTypeAny
 	}
 	// proxyPorts is small enough to not bother indexing it.
@@ -358,12 +367,12 @@ func findProxyPortByType(l7Type ProxyType, ingress bool) (string, *ProxyPort) {
 	return "", nil
 }
 
-func proxyTypeNotFoundError(proxyType ProxyType, ingress bool) error {
+func proxyTypeNotFoundError(proxyType ProxyType, listener string, ingress bool) error {
 	dir := "egress"
 	if ingress {
 		dir = "ingress"
 	}
-	return fmt.Errorf("unrecognized %s proxy type: %s", dir, proxyType)
+	return fmt.Errorf("unrecognized %s proxy type for %s: %s", dir, listener, proxyType)
 }
 
 func proxyNotFoundError(name string) error {
@@ -415,6 +424,8 @@ func (p *Proxy) AllocateProxyPort(name string, ingress bool) (uint16, error) {
 	proxyPorts[name] = pp
 	pp.reservePort() // marks port as reserved, 'pp' as configured
 
+	log.WithField(fieldProxyRedirectID, name).Debugf("AllocateProxyPort: allocated proxy port %d (%v)", pp.proxyPort, *pp)
+
 	return pp.proxyPort, nil
 }
 
@@ -530,9 +541,9 @@ func (p *Proxy) CreateOrUpdateRedirect(ctx context.Context, l4 policy.ProxyPolic
 
 	proxyPortsMutex.Lock()
 	defer proxyPortsMutex.Unlock()
-	ppName, pp := findProxyPortByType(ProxyType(l4.GetL7Parser()), l4.GetIngress())
+	ppName, pp := findProxyPortByType(ProxyType(l4.GetL7Parser()), l4.GetListener(), l4.GetIngress())
 	if pp == nil {
-		err = proxyTypeNotFoundError(ProxyType(l4.GetL7Parser()), l4.GetIngress())
+		err = proxyTypeNotFoundError(ProxyType(l4.GetL7Parser()), l4.GetListener(), l4.GetIngress())
 		revertFunc()
 		return 0, err, nil, nil
 	}
@@ -564,7 +575,14 @@ func (p *Proxy) CreateOrUpdateRedirect(ctx context.Context, l4 policy.ProxyPolic
 		case policy.ParserTypeDNS:
 			redir.implementation, err = createDNSRedirect(redir, dnsConfiguration{}, p.defaultEndpointInfoRegistry)
 		default:
-			redir.implementation, err = createEnvoyRedirect(redir, p.stateDir, p.XDSServer, p.datapathUpdater.SupportsOriginalSourceAddr(), wg)
+			if pp.proxyType == ProxyTypeCRD {
+				// CRD Listeners already exist, create a no-op implementation
+				redir.implementation = &CRDRedirect{}
+				err = nil
+			} else {
+				// create an Envoy Listener for Cilium policy enforcement
+				redir.implementation, err = createEnvoyRedirect(redir, p.stateDir, p.XDSServer, p.datapathUpdater.SupportsOriginalSourceAddr(), wg)
+			}
 		}
 
 		if err == nil {
@@ -610,7 +628,7 @@ func (p *Proxy) CreateOrUpdateRedirect(ctx context.Context, l4 policy.ProxyPolic
 	}
 
 	// an error occurred, and we have no more retries
-	scopedLog.WithError(err).Error("Unable to create ", l4.GetL7Parser(), " proxy")
+	scopedLog.WithError(err).Errorf("Unable to create %s proxy %s", l4.GetL7Parser(), l4.GetListener())
 	revertFunc() // Ignore errors while reverting. This is best-effort.
 	return 0, err, nil, nil
 }
diff --git a/pkg/proxy/redirect.go b/pkg/proxy/redirect.go
index 06cdc4c79b..721114c8bc 100644
--- a/pkg/proxy/redirect.go
+++ b/pkg/proxy/redirect.go
@@ -28,6 +28,17 @@ type RedirectImplementation interface {
 	Close(wg *completion.WaitGroup) (revert.FinalizeFunc, revert.RevertFunc)
 }
 
+// Redirect type for custom Listeners, which are managed externally.
+type CRDRedirect struct{}
+
+func (r *CRDRedirect) UpdateRules(wg *completion.WaitGroup) (revert.RevertFunc, error) {
+	return func() error { return nil }, nil
+}
+
+func (r *CRDRedirect) Close(wg *completion.WaitGroup) (revert.FinalizeFunc, revert.RevertFunc) {
+	return nil, func() error { return nil }
+}
+
 type Redirect struct {
 	// The following fields are only written to during initialization, it
 	// is safe to read these fields without locking the mutex


Test patch:
diff --git a/daemon/cmd/fqdn_test.go b/daemon/cmd/fqdn_test.go
index d53c6ad9e4..74b00dd2f0 100644
--- a/daemon/cmd/fqdn_test.go
+++ b/daemon/cmd/fqdn_test.go
@@ -13,6 +13,7 @@ import (
 	"sync"
 	"time"
 
+	miekgdns "github.com/miekg/dns"
 	. "gopkg.in/check.v1"
 	k8sCache "k8s.io/client-go/tools/cache"
 
@@ -23,14 +24,19 @@ import (
 	"github.com/cilium/cilium/pkg/endpoint"
 	"github.com/cilium/cilium/pkg/fqdn"
 	"github.com/cilium/cilium/pkg/fqdn/dns"
+	"github.com/cilium/cilium/pkg/fqdn/dnsproxy"
 	"github.com/cilium/cilium/pkg/fqdn/re"
 	"github.com/cilium/cilium/pkg/identity"
 	"github.com/cilium/cilium/pkg/identity/cache"
+	"github.com/cilium/cilium/pkg/ipcache"
 	"github.com/cilium/cilium/pkg/k8s/client/clientset/versioned"
 	"github.com/cilium/cilium/pkg/kvstore"
 	"github.com/cilium/cilium/pkg/labels"
+	"github.com/cilium/cilium/pkg/lock"
 	"github.com/cilium/cilium/pkg/policy"
 	"github.com/cilium/cilium/pkg/policy/api"
+	"github.com/cilium/cilium/pkg/proxy/accesslog"
+	"github.com/cilium/cilium/pkg/proxy/logger"
 	testidentity "github.com/cilium/cilium/pkg/testutils/identity"
 )
 
@@ -112,6 +118,7 @@ func (f *FakeRefcountingIdentityAllocator) WatchRemoteIdentities(kvstore.Backend
 
 func (ds *DaemonFQDNSuite) SetUpTest(c *C) {
 	d := &Daemon{}
+	d.initDNSProxyContext(128)
 	d.identityAllocator = NewFakeIdentityAllocator(nil)
 	d.policy = policy.NewPolicyRepository(d.identityAllocator, nil, nil)
 	d.dnsNameManager = fqdn.NewNameManager(fqdn.Config{
@@ -121,7 +128,20 @@ func (ds *DaemonFQDNSuite) SetUpTest(c *C) {
 	})
 	d.endpointManager = WithCustomEndpointManager(&dummyEpSyncher{})
 	d.policy.GetSelectorCache().SetLocalIdentityNotifier(d.dnsNameManager)
+	d.ipcache = ipcache.NewIPCache(&ipcache.Configuration{
+		Context:           context.TODO(),
+		IdentityAllocator: d.identityAllocator,
+		PolicyHandler:     d.policy.GetSelectorCache(),
+		DatapathHandler:   d.endpointManager,
+	})
 	ds.d = d
+
+	logger.SetEndpointInfoRegistry(&dummyInfoRegistry{})
+}
+
+type dummyInfoRegistry struct{}
+
+func (*dummyInfoRegistry) FillEndpointInfo(info *accesslog.EndpointInfo, ip net.IP, id identity.NumericIdentity) {
 }
 
 // makeIPs generates count sequential IPv4 IPs
@@ -156,6 +176,91 @@ func (ds *DaemonSuite) BenchmarkFqdnCache(c *C) {
 	extractDNSLookups(endpoints, "0.0.0.0/0", "*", "")
 }
 
+// Benchmark_notifyOnDNSMsg stresses the main callback function for the DNS
+// proxy path, which is called on every DNS request and response.
+func (ds *DaemonFQDNSuite) Benchmark_notifyOnDNSMsg(c *C) {
+	var (
+		nameManager             = ds.d.dnsNameManager
+		ciliumIOSel             = api.FQDNSelector{MatchName: "cilium.io"}
+		ciliumIOSelMatchPattern = api.FQDNSelector{MatchPattern: "*cilium.io."}
+		ebpfIOSel               = api.FQDNSelector{MatchName: "ebpf.io"}
+		ciliumDNSRecord         = map[string]*fqdn.DNSIPRecords{
+			dns.FQDN("cilium.io"): {TTL: 60, IPs: []net.IP{net.ParseIP("192.0.2.3")}},
+		}
+		ebpfDNSRecord = map[string]*fqdn.DNSIPRecords{
+			dns.FQDN("ebpf.io"): {TTL: 60, IPs: []net.IP{net.ParseIP("192.0.2.4")}},
+		}
+
+		wg sync.WaitGroup
+	)
+
+	// Register rules (simulates applied policies).
+	selectorsToAdd := api.FQDNSelectorSlice{ciliumIOSel, ciliumIOSelMatchPattern, ebpfIOSel}
+	nameManager.Lock()
+	for _, sel := range selectorsToAdd {
+		nameManager.RegisterForIdentityUpdatesLocked(sel)
+	}
+	nameManager.Unlock()
+
+	// Initialize the endpoints.
+	endpoints := make([]*endpoint.Endpoint, c.N)
+	for i := range endpoints {
+		endpoints[i] = &endpoint.Endpoint{
+			ID:   uint16(c.N % 65000),
+			IPv4: netip.MustParseAddr(fmt.Sprintf("10.96.%d.%d", (c.N>>16)%8, c.N%256)),
+			SecurityIdentity: &identity.Identity{
+				ID: identity.NumericIdentity(c.N % int(identity.MaximumAllocationIdentity)),
+			},
+			DNSZombies: &fqdn.DNSZombieMappings{
+				Mutex: lock.Mutex{},
+			},
+		}
+		ep := endpoints[i]
+		ep.UpdateLogger(nil)
+		ep.DNSHistory = fqdn.NewDNSCache(0)
+	}
+
+	c.ResetTimer()
+	// Simulate parallel DNS responses from the upstream DNS for cilium.io and
+	// ebpf.io, done by every endpoint.
+	for i := 0; i < c.N; i++ {
+		wg.Add(1)
+		go func(ep *endpoint.Endpoint) {
+			defer wg.Done()
+			// Using a hardcoded string representing endpoint IP:port as this
+			// parameter is only used in logging. Not using the endpoint's IP
+			// so we don't spend any time in the benchmark on converting from
+			// net.IP to string.
+			c.Assert(ds.d.notifyOnDNSMsg(time.Now(), ep, "10.96.64.8:12345", 0, "10.96.64.1:53", &miekgdns.Msg{
+				MsgHdr: miekgdns.MsgHdr{
+					Response: true,
+				},
+				Question: []miekgdns.Question{{
+					Name: dns.FQDN("cilium.io"),
+				}},
+				Answer: []miekgdns.RR{&miekgdns.A{
+					Hdr: miekgdns.RR_Header{Name: dns.FQDN("cilium.io")},
+					A:   ciliumDNSRecord[dns.FQDN("cilium.io")].IPs[0],
+				}}}, "udp", true, &dnsproxy.ProxyRequestContext{}), IsNil)
+
+			c.Assert(ds.d.notifyOnDNSMsg(time.Now(), ep, "10.96.64.4:54321", 0, "10.96.64.1:53", &miekgdns.Msg{
+				MsgHdr: miekgdns.MsgHdr{
+					Response: true,
+				},
+				Compress: false,
+				Question: []miekgdns.Question{{
+					Name: dns.FQDN("ebpf.io"),
+				}},
+				Answer: []miekgdns.RR{&miekgdns.A{
+					Hdr: miekgdns.RR_Header{Name: dns.FQDN("ebpf.io")},
+					A:   ebpfDNSRecord[dns.FQDN("ebpf.io")].IPs[0],
+				}}}, "udp", true, &dnsproxy.ProxyRequestContext{}), IsNil)
+		}(endpoints[i%len(endpoints)])
+	}
+
+	wg.Wait()
+}
+
 func (ds *DaemonFQDNSuite) TestFQDNIdentityReferenceCounting(c *C) {
 	var (
 		idAllocator             = ds.d.identityAllocator.(*FakeRefcountingIdentityAllocator)
@@ -216,3 +321,23 @@ func (ds *DaemonFQDNSuite) TestFQDNIdentityReferenceCounting(c *C) {
 	c.Assert(idAllocator.IdentityReferenceCounter(), checker.DeepEquals, counter.IntCounter{},
 		Commentf("The Daemon code leaked references to one or more identities"))
 }
+
+func (ds *DaemonFQDNSuite) Test_getMutexesForResponseIPs(c *C) {
+	r := make([]net.IP, 0)
+	for i := 0; i < 64; i++ {
+		r = append(r, net.ParseIP(fmt.Sprintf("1.1.1.%d", i)))
+	}
+	m := ds.d.dnsProxyContext.getMutexesForResponseIPs(r)
+	c.Assert(m, HasLen, 64)
+	c.Assert(m[0], Equals, 0)
+	c.Assert(m[len(m)-1], Equals, len(m)-1)
+
+	r = make([]net.IP, 0)
+	for i := 0; i < 64; i++ {
+		r = append(r, net.ParseIP(fmt.Sprintf("1.%d.1.1", i)))
+	}
+	m = ds.d.dnsProxyContext.getMutexesForResponseIPs(r)
+	fmt.Println(m)
+	c.Assert(m, HasLen, 1)
+	c.Assert(m[0], Equals, 1)
+}
diff --git a/pkg/envoy/ciliumenvoyconfig_test.go b/pkg/envoy/ciliumenvoyconfig_test.go
index dc30480c9a..07f9c557f5 100644
--- a/pkg/envoy/ciliumenvoyconfig_test.go
+++ b/pkg/envoy/ciliumenvoyconfig_test.go
@@ -4,16 +4,21 @@
 package envoy
 
 import (
+	"bytes"
 	"context"
 	"encoding/json"
 	"fmt"
 
+	cilium "github.com/cilium/proxy/go/cilium/api"
 	envoy_config_cluster "github.com/cilium/proxy/go/envoy/config/cluster/v3"
 	envoy_config_core "github.com/cilium/proxy/go/envoy/config/core/v3"
+	envoy_config_listener "github.com/cilium/proxy/go/envoy/config/listener/v3"
 	envoy_config_http "github.com/cilium/proxy/go/envoy/extensions/filters/network/http_connection_manager/v3"
+	envoy_config_tcp "github.com/cilium/proxy/go/envoy/extensions/filters/network/tcp_proxy/v3"
 	envoy_config_tls "github.com/cilium/proxy/go/envoy/extensions/transport_sockets/tls/v3"
 	"sigs.k8s.io/yaml"
 
+	"github.com/cilium/cilium/pkg/bpf"
 	cilium_v2 "github.com/cilium/cilium/pkg/k8s/apis/cilium.io/v2"
 
 	. "gopkg.in/check.v1"
@@ -162,7 +167,7 @@ func (s *JSONSuite) TestCiliumEnvoyConfig(c *C) {
 	c.Assert(cec.Spec.Resources, HasLen, 1)
 	c.Assert(cec.Spec.Resources[0].TypeUrl, Equals, "type.googleapis.com/envoy.config.listener.v3.Listener")
 
-	resources, err := ParseResources("namespace", "name", cec.Spec.Resources, true, portAllocator)
+	resources, err := ParseResources("namespace", "name", cec.Spec.Resources, true, portAllocator, false)
 	c.Assert(err, IsNil)
 	c.Assert(resources.Listeners, HasLen, 1)
 	c.Assert(resources.Listeners[0].Name, Equals, "namespace/name/envoy-prometheus-metrics-listener")
@@ -253,7 +258,7 @@ func (s *JSONSuite) TestCiliumEnvoyConfigValidation(c *C) {
 	c.Assert(cec.Spec.Resources, HasLen, 1)
 	c.Assert(cec.Spec.Resources[0].TypeUrl, Equals, "type.googleapis.com/envoy.config.listener.v3.Listener")
 
-	resources, err := ParseResources("namespace", "name", cec.Spec.Resources, false, portAllocator)
+	resources, err := ParseResources("namespace", "name", cec.Spec.Resources, false, portAllocator, false)
 	c.Assert(err, IsNil)
 	c.Assert(resources.Listeners, HasLen, 1)
 	c.Assert(resources.Listeners[0].Name, Equals, "namespace/name/envoy-prometheus-metrics-listener")
@@ -286,7 +291,7 @@ func (s *JSONSuite) TestCiliumEnvoyConfigValidation(c *C) {
 	//
 	// Same with validation fails
 	//
-	resources, err = ParseResources("namespace", "name", cec.Spec.Resources, true, portAllocator)
+	resources, err = ParseResources("namespace", "name", cec.Spec.Resources, true, portAllocator, false)
 	c.Assert(err, Not(IsNil))
 }
 
@@ -326,7 +331,7 @@ func (s *JSONSuite) TestCiliumEnvoyConfigNoAddress(c *C) {
 	c.Assert(cec.Spec.Resources, HasLen, 1)
 	c.Assert(cec.Spec.Resources[0].TypeUrl, Equals, "type.googleapis.com/envoy.config.listener.v3.Listener")
 
-	resources, err := ParseResources("namespace", "name", cec.Spec.Resources, true, portAllocator)
+	resources, err := ParseResources("namespace", "name", cec.Spec.Resources, true, portAllocator, false)
 	c.Assert(err, IsNil)
 	c.Assert(resources.Listeners, HasLen, 1)
 	c.Assert(resources.Listeners[0].Name, Equals, "namespace/name/envoy-prometheus-metrics-listener")
@@ -440,7 +445,7 @@ func (s *JSONSuite) TestCiliumEnvoyConfigMulti(c *C) {
 	c.Assert(cec.Spec.Resources, HasLen, 5)
 	c.Assert(cec.Spec.Resources[0].TypeUrl, Equals, "type.googleapis.com/envoy.config.listener.v3.Listener")
 
-	resources, err := ParseResources("namespace", "name", cec.Spec.Resources, true, portAllocator)
+	resources, err := ParseResources("namespace", "name", cec.Spec.Resources, true, portAllocator, false)
 	c.Assert(err, IsNil)
 	c.Assert(resources.Listeners, HasLen, 1)
 	c.Assert(resources.Listeners[0].Name, Equals, "namespace/name/multi-resource-listener")
@@ -551,6 +556,253 @@ func (s *JSONSuite) TestCiliumEnvoyConfigMulti(c *C) {
 	c.Assert(addr.GetSocketAddress().GetPortValue(), Equals, uint32(5678))
 }
 
+var ciliumEnvoyConfigTCPProxy = `apiVersion: cilium.io/v2
+kind: CiliumEnvoyConfig
+metadata:
+  name: envoy-test-listener
+spec:
+  resources:
+  - "@type": type.googleapis.com/envoy.config.listener.v3.Listener
+    name: tcp_proxy_test-2
+    filter_chains:
+    - filters:
+      - name: envoy.filters.network.tcp_proxy
+        typed_config:
+          "@type": type.googleapis.com/envoy.extensions.filters.network.tcp_proxy.v3.TcpProxy
+          stat_prefix: tcp_stats
+          cluster: "cluster_0"
+          tunneling_config:
+            hostname: host.com:443
+            use_post: true
+  - "@type": type.googleapis.com/envoy.config.cluster.v3.Cluster
+    name: "cluster_0"
+    connect_timeout: 5s
+    # This ensures HTTP/2 POST is used for establishing the tunnel.
+    typed_extension_protocol_options:
+      envoy.extensions.upstreams.http.v3.HttpProtocolOptions:
+        "@type": type.googleapis.com/envoy.extensions.upstreams.http.v3.HttpProtocolOptions
+        explicit_http_config:
+          http2_protocol_options: {}
+    load_assignment:
+      cluster_name: cluster_0
+      endpoints:
+      - lb_endpoints:
+        - endpoint:
+            address:
+              socket_address:
+                address: 127.0.0.1
+                port_value: 10001
+`
+
+func (s *JSONSuite) TestCiliumEnvoyConfigTCPProxy(c *C) {
+	portAllocator := NewMockPortAllocator()
+	jsonBytes, err := yaml.YAMLToJSON([]byte(ciliumEnvoyConfigTCPProxy))
+	c.Assert(err, IsNil)
+
+	var buf bytes.Buffer
+	json.Indent(&buf, jsonBytes, "", "\t")
+	fmt.Printf("JSON spec:\n%s\n", buf.String())
+
+	cec := &cilium_v2.CiliumEnvoyConfig{}
+	err = json.Unmarshal(jsonBytes, cec)
+	c.Assert(err, IsNil)
+	c.Assert(cec.Spec.Resources, Not(IsNil))
+	c.Assert(cec.Spec.Resources, HasLen, 2)
+	c.Assert(cec.Spec.Resources[0].TypeUrl, Equals, "type.googleapis.com/envoy.config.listener.v3.Listener")
+
+	resources, err := ParseResources("namespace", "name", cec.Spec.Resources, true, portAllocator, false)
+	c.Assert(err, IsNil)
+	c.Assert(resources.Listeners, HasLen, 1)
+	c.Assert(resources.Listeners[0].Address, Not(IsNil))
+	c.Assert(resources.Listeners[0].Address.GetSocketAddress(), Not(IsNil))
+	c.Assert(resources.Listeners[0].Address.GetSocketAddress().GetPortValue(), Not(Equals), 0)
+	//
+	// Check injected listener filter config
+	//
+	c.Assert(resources.Listeners[0].ListenerFilters, HasLen, 1)
+	c.Assert(resources.Listeners[0].ListenerFilters[0].Name, Equals, "cilium.bpf_metadata")
+	lfMsg, err := resources.Listeners[0].ListenerFilters[0].GetTypedConfig().UnmarshalNew()
+	c.Assert(err, IsNil)
+	c.Assert(lfMsg, Not(IsNil))
+	lf, ok := lfMsg.(*cilium.BpfMetadata)
+	c.Assert(ok, Equals, true)
+	c.Assert(lf, Not(IsNil))
+	c.Assert(lf.IsIngress, Equals, false)
+	c.Assert(lf.MayUseOriginalSourceAddress, Equals, true)
+	c.Assert(lf.BpfRoot, Equals, bpf.GetMapRoot())
+	c.Assert(lf.EgressMarkSourceEndpointId, Equals, false)
+
+	c.Assert(resources.Listeners[0].FilterChains, HasLen, 1)
+	chain := resources.Listeners[0].FilterChains[0]
+	c.Assert(chain.Filters, HasLen, 2)
+	c.Assert(chain.Filters[0].Name, Equals, "cilium.network")
+	c.Assert(chain.Filters[1].Name, Equals, "envoy.filters.network.tcp_proxy")
+	message, err := chain.Filters[1].GetTypedConfig().UnmarshalNew()
+	c.Assert(err, IsNil)
+	c.Assert(message, Not(IsNil))
+	tcp, ok := message.(*envoy_config_tcp.TcpProxy)
+	c.Assert(ok, Equals, true)
+	c.Assert(tcp, Not(IsNil))
+	//
+	// Check TCP config
+	//
+	tc := tcp.GetTunnelingConfig()
+	c.Assert(tc, Not(IsNil))
+	c.Assert(tc.Hostname, Equals, "host.com:443")
+	c.Assert(tc.UsePost, Equals, true)
+	//
+	// Check cluster resource
+	//
+	c.Assert(cec.Spec.Resources[1].TypeUrl, Equals, "type.googleapis.com/envoy.config.cluster.v3.Cluster")
+	c.Assert(resources.Clusters, HasLen, 1)
+	c.Assert(resources.Clusters[0].Name, Equals, "cluster_0")
+	c.Assert(resources.Clusters[0].ConnectTimeout.Seconds, Equals, int64(5))
+	c.Assert(resources.Clusters[0].ConnectTimeout.Nanos, Equals, int32(0))
+	c.Assert(resources.Clusters[0].LoadAssignment.ClusterName, Equals, "cluster_0")
+	c.Assert(resources.Clusters[0].LoadAssignment.Endpoints, HasLen, 1)
+	c.Assert(resources.Clusters[0].LoadAssignment.Endpoints[0].LbEndpoints, HasLen, 1)
+	addr := resources.Clusters[0].LoadAssignment.Endpoints[0].LbEndpoints[0].GetEndpoint().Address
+	c.Assert(addr, Not(IsNil))
+	c.Assert(addr.GetSocketAddress(), Not(IsNil))
+	c.Assert(addr.GetSocketAddress().GetAddress(), Equals, "127.0.0.1")
+	c.Assert(addr.GetSocketAddress().GetPortValue(), Equals, uint32(10001))
+}
+
+var ciliumEnvoyConfigTCPProxyTermination = `apiVersion: cilium.io/v2
+kind: CiliumEnvoyConfig
+metadata:
+  name: tcp-proxy-ingress-listener
+spec:
+  services:
+  - name: tcp-proxy-ingress
+    namespace: cilium-test
+  resources:
+  - "@type": type.googleapis.com/envoy.config.listener.v3.Listener
+    name: envoy-ingress-listener
+    filter_chains:
+    - filters:
+      - name: envoy.filters.network.http_connection_manager
+        typed_config:
+          "@type": type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager
+          stat_prefix: tcp-proxy-ingress-listener
+          route_config:
+            name: local_route
+            virtual_hosts:
+            - name: local_service
+              domains:
+              - "*"
+              routes:
+              - match:
+                  prefix: "/"
+                  headers:
+                  - name: ":method"
+                    string_match:
+                      exact: "POST"
+                route:
+                  cluster: service_google
+                  upgrade_configs:
+                  - upgrade_type: CONNECT
+                    connect_config:
+                      allow_post: true
+          http_filters:
+          - name: envoy.filters.http.router
+          http2_protocol_options:
+            allow_connect: true
+  - "@type": type.googleapis.com/envoy.config.cluster.v3.Cluster
+    name: service_google
+    connect_timeout: 5s
+    type: LOGICAL_DNS
+    # Comment out the following line to test on v6 networks
+    dns_lookup_family: V4_ONLY
+    lb_policy: ROUND_ROBIN
+    load_assignment:
+      cluster_name: service_google
+      endpoints:
+      - lb_endpoints:
+        - endpoint:
+            address:
+              socket_address:
+                address: www.google.com
+                port_value: 443
+`
+
+func (s *JSONSuite) TestCiliumEnvoyConfigTCPProxyTermination(c *C) {
+	portAllocator := NewMockPortAllocator()
+	jsonBytes, err := yaml.YAMLToJSON([]byte(ciliumEnvoyConfigTCPProxyTermination))
+	c.Assert(err, IsNil)
+
+	var buf bytes.Buffer
+	json.Indent(&buf, jsonBytes, "", "\t")
+	fmt.Printf("JSON spec:\n%s\n", buf.String())
+
+	cec := &cilium_v2.CiliumEnvoyConfig{}
+	err = json.Unmarshal(jsonBytes, cec)
+	c.Assert(err, IsNil)
+	c.Assert(cec.Spec.Resources, Not(IsNil))
+	c.Assert(cec.Spec.Resources, HasLen, 2)
+	c.Assert(cec.Spec.Resources[0].TypeUrl, Equals, "type.googleapis.com/envoy.config.listener.v3.Listener")
+
+	resources, err := ParseResources("namespace", "name", cec.Spec.Resources, true, portAllocator, true)
+	c.Assert(err, IsNil)
+	c.Assert(resources.Listeners, HasLen, 1)
+	c.Assert(resources.Listeners[0].Address, Not(IsNil))
+	c.Assert(resources.Listeners[0].Address.GetSocketAddress(), Not(IsNil))
+	c.Assert(resources.Listeners[0].Address.GetSocketAddress().GetPortValue(), Not(Equals), 0)
+	//
+	// Check injected listener filter config
+	//
+	c.Assert(resources.Listeners[0].ListenerFilters, HasLen, 1)
+	c.Assert(resources.Listeners[0].ListenerFilters[0].Name, Equals, "cilium.bpf_metadata")
+	lfMsg, err := resources.Listeners[0].ListenerFilters[0].GetTypedConfig().UnmarshalNew()
+	c.Assert(err, IsNil)
+	c.Assert(lfMsg, Not(IsNil))
+	lf, ok := lfMsg.(*cilium.BpfMetadata)
+	c.Assert(ok, Equals, true)
+	c.Assert(lf, Not(IsNil))
+	c.Assert(lf.IsIngress, Equals, false)
+	c.Assert(lf.MayUseOriginalSourceAddress, Equals, false)
+	c.Assert(lf.BpfRoot, Equals, bpf.GetMapRoot())
+	c.Assert(lf.EgressMarkSourceEndpointId, Equals, true)
+
+	c.Assert(resources.Listeners[0].FilterChains, HasLen, 1)
+	chain := resources.Listeners[0].FilterChains[0]
+	c.Assert(chain.Filters, HasLen, 2)
+	c.Assert(chain.Filters[0].Name, Equals, "cilium.network")
+	c.Assert(chain.Filters[1].Name, Equals, "envoy.filters.network.http_connection_manager")
+	message, err := chain.Filters[1].GetTypedConfig().UnmarshalNew()
+	c.Assert(err, IsNil)
+	c.Assert(message, Not(IsNil))
+	hcm, ok := message.(*envoy_config_http.HttpConnectionManager)
+	c.Assert(ok, Equals, true)
+	c.Assert(hcm, Not(IsNil))
+	//
+	// Check HTTP config
+	//
+	c.Assert(hcm.HttpFilters, HasLen, 2)
+	c.Assert(hcm.HttpFilters[0].Name, Equals, "cilium.l7policy")
+	c.Assert(hcm.HttpFilters[1].Name, Equals, "envoy.filters.http.router")
+	//
+	// Check cluster resource
+	//
+	c.Assert(cec.Spec.Resources[1].TypeUrl, Equals, "type.googleapis.com/envoy.config.cluster.v3.Cluster")
+	c.Assert(resources.Clusters, HasLen, 1)
+	c.Assert(resources.Clusters[0].Name, Equals, "service_google")
+	c.Assert(resources.Clusters[0].ConnectTimeout.Seconds, Equals, int64(5))
+	c.Assert(resources.Clusters[0].ConnectTimeout.Nanos, Equals, int32(0))
+	c.Assert(resources.Clusters[0].GetType(), Equals, envoy_config_cluster.Cluster_LOGICAL_DNS)
+	c.Assert(resources.Clusters[0].GetDnsLookupFamily(), Equals, envoy_config_cluster.Cluster_V4_ONLY)
+	c.Assert(resources.Clusters[0].LbPolicy, Equals, envoy_config_cluster.Cluster_ROUND_ROBIN)
+
+	c.Assert(resources.Clusters[0].LoadAssignment.ClusterName, Equals, "service_google")
+	c.Assert(resources.Clusters[0].LoadAssignment.Endpoints, HasLen, 1)
+	c.Assert(resources.Clusters[0].LoadAssignment.Endpoints[0].LbEndpoints, HasLen, 1)
+	addr := resources.Clusters[0].LoadAssignment.Endpoints[0].LbEndpoints[0].GetEndpoint().Address
+	c.Assert(addr, Not(IsNil))
+	c.Assert(addr.GetSocketAddress(), Not(IsNil))
+	c.Assert(addr.GetSocketAddress().GetAddress(), Equals, "www.google.com")
+	c.Assert(addr.GetSocketAddress().GetPortValue(), Equals, uint32(443))
+}
+
 func checkCiliumXDS(c *C, cs *envoy_config_core.ConfigSource) {
 	c.Assert(cs, Not(IsNil))
 	c.Assert(cs.ResourceApiVersion, Equals, envoy_config_core.ApiVersion_V3)
@@ -565,27 +817,60 @@ func checkCiliumXDS(c *C, cs *envoy_config_core.ConfigSource) {
 	c.Assert(eg.ClusterName, Equals, "xds-grpc-cilium")
 }
 
-func (s *JSONSuite) TestResourceQualifiedName(c *C) {
-	var fullName, namespace, name, resource string
-
-	resource = "test-resource"
-	fullName = resourceQualifiedName(namespace, name, resource)
-	c.Assert(fullName, Equals, "test-resource")
-
-	name = "test-name"
-	resource = "test-resource"
-	fullName = resourceQualifiedName(namespace, name, resource)
-	c.Assert(fullName, Equals, "test-name/test-resource")
-
-	namespace = "test-namespace"
-	name = ""
-	resource = "test-resource"
-	fullName = resourceQualifiedName(namespace, name, resource)
-	c.Assert(fullName, Equals, "test-namespace/test-resource")
-
-	namespace = "test-namespace"
-	name = "test-name"
-	resource = "test-resource"
-	fullName = resourceQualifiedName(namespace, name, resource)
-	c.Assert(fullName, Equals, "test-namespace/test-name/test-resource")
+func (s *JSONSuite) TestListenersAddedOrDeleted(c *C) {
+	var old Resources
+	var new Resources
+
+	// Both empty
+	res := old.ListenersAddedOrDeleted(&new)
+	c.Assert(res, Equals, false)
+
+	// new adds a listener
+	new.Listeners = append(old.Listeners, &envoy_config_listener.Listener{Name: "foo"})
+	res = old.ListenersAddedOrDeleted(&new)
+	c.Assert(res, Equals, true)
+	res = new.ListenersAddedOrDeleted(&old)
+	c.Assert(res, Equals, true)
+
+	// Now both have 'foo'
+	old.Listeners = append(old.Listeners, &envoy_config_listener.Listener{Name: "foo"})
+	res = old.ListenersAddedOrDeleted(&new)
+	c.Assert(res, Equals, false)
+	res = new.ListenersAddedOrDeleted(&old)
+	c.Assert(res, Equals, false)
+
+	// New has no listeners
+	new.Listeners = nil
+	res = old.ListenersAddedOrDeleted(&new)
+	c.Assert(res, Equals, true)
+	res = new.ListenersAddedOrDeleted(&old)
+	c.Assert(res, Equals, true)
+
+	// New has a different listener
+	new.Listeners = append(new.Listeners, &envoy_config_listener.Listener{Name: "bar"})
+	res = old.ListenersAddedOrDeleted(&new)
+	c.Assert(res, Equals, true)
+	res = new.ListenersAddedOrDeleted(&old)
+	c.Assert(res, Equals, true)
+
+	// New adds the listener in old, but still has the other listener
+	new.Listeners = append(new.Listeners, &envoy_config_listener.Listener{Name: "foo"})
+	res = old.ListenersAddedOrDeleted(&new)
+	c.Assert(res, Equals, true)
+	res = new.ListenersAddedOrDeleted(&old)
+	c.Assert(res, Equals, true)
+
+	// Same listeners but in different order
+	old.Listeners = append(old.Listeners, &envoy_config_listener.Listener{Name: "bar"})
+	res = old.ListenersAddedOrDeleted(&new)
+	c.Assert(res, Equals, false)
+	res = new.ListenersAddedOrDeleted(&old)
+	c.Assert(res, Equals, false)
+
+	// Old has no listeners
+	old.Listeners = nil
+	res = old.ListenersAddedOrDeleted(&new)
+	c.Assert(res, Equals, true)
+	res = new.ListenersAddedOrDeleted(&old)
+	c.Assert(res, Equals, true)
 }
diff --git a/pkg/k8s/watchers/cilium_envoy_config_test.go b/pkg/k8s/watchers/cilium_envoy_config_test.go
index 7cb0a852bc..fac17a5df9 100644
--- a/pkg/k8s/watchers/cilium_envoy_config_test.go
+++ b/pkg/k8s/watchers/cilium_envoy_config_test.go
@@ -65,7 +65,7 @@ func (s *K8sWatcherSuite) TestParseEnvoySpec(c *C) {
 	c.Assert(cec.Spec.Resources, HasLen, 1)
 	c.Assert(cec.Spec.Resources[0].TypeUrl, Equals, "type.googleapis.com/envoy.config.listener.v3.Listener")
 
-	resources, err := envoy.ParseResources("namespace", "name", cec.Spec.Resources, true, nil)
+	resources, err := envoy.ParseResources("namespace", "name", cec.Spec.Resources, true, nil, len(cec.Spec.Services) > 0)
 	c.Assert(err, IsNil)
 	c.Assert(resources.Listeners, HasLen, 1)
 	c.Assert(resources.Listeners[0].Address.GetSocketAddress().GetPortValue(), Equals, uint32(10000))
diff --git a/pkg/policy/api/rule_validation_test.go b/pkg/policy/api/rule_validation_test.go
index a2af518730..ca992ae7eb 100644
--- a/pkg/policy/api/rule_validation_test.go
+++ b/pkg/policy/api/rule_validation_test.go
@@ -338,6 +338,140 @@ func (s *PolicyAPITestSuite) TestL7RulesWithNonTCPProtocols(c *C) {
 	}
 	err = validPortRule.Sanitize()
 	c.Assert(err, IsNil)
+
+	// Rule is valid because Listener is allowed on egress, default Kind
+	validPortRule = Rule{
+		EndpointSelector: WildcardEndpointSelector,
+		Egress: []EgressRule{
+			{
+				EgressCommonRule: EgressCommonRule{
+					ToEndpoints: []EndpointSelector{WildcardEndpointSelector},
+				},
+				ToPorts: []PortRule{{
+					Ports: []PortProtocol{
+						{Port: "443", Protocol: ProtoTCP},
+					},
+					Listener: &Listener{
+						EnvoyConfig: &EnvoyConfig{
+							Name: "test-config",
+						},
+						Name: "myCustomListener",
+					},
+				}},
+			},
+		},
+	}
+	err = validPortRule.Sanitize()
+	c.Assert(err, IsNil)
+
+	// Rule is valid because Listener is allowed on egress, Kind CiliumClusterwideEnvoyConfig
+	validPortRule = Rule{
+		EndpointSelector: WildcardEndpointSelector,
+		Egress: []EgressRule{
+			{
+				EgressCommonRule: EgressCommonRule{
+					ToEndpoints: []EndpointSelector{WildcardEndpointSelector},
+				},
+				ToPorts: []PortRule{{
+					Ports: []PortProtocol{
+						{Port: "443", Protocol: ProtoTCP},
+					},
+					Listener: &Listener{
+						EnvoyConfig: &EnvoyConfig{
+							Kind: "CiliumClusterwideEnvoyConfig",
+							Name: "shared-config",
+						},
+						Name: "myCustomListener",
+					},
+				}},
+			},
+		},
+	}
+	err = validPortRule.Sanitize()
+	c.Assert(err, IsNil)
+
+	// Rule is valid because Listener is allowed on egress, Kind CiliumEnvoyConfig
+	validPortRule = Rule{
+		EndpointSelector: WildcardEndpointSelector,
+		Egress: []EgressRule{
+			{
+				EgressCommonRule: EgressCommonRule{
+					ToEndpoints: []EndpointSelector{WildcardEndpointSelector},
+				},
+				ToPorts: []PortRule{{
+					Ports: []PortProtocol{
+						{Port: "443", Protocol: ProtoTCP},
+					},
+					Listener: &Listener{
+						EnvoyConfig: &EnvoyConfig{
+							Kind: "CiliumEnvoyConfig",
+							Name: "shared-config",
+						},
+						Name: "myCustomListener",
+					},
+				}},
+			},
+		},
+	}
+	err = validPortRule.Sanitize()
+	c.Assert(err, IsNil)
+
+	// Rule is invalid because Listener is not allowed on ingress (yet)
+	invalidPortRule = Rule{
+		EndpointSelector: WildcardEndpointSelector,
+		Ingress: []IngressRule{
+			{
+				IngressCommonRule: IngressCommonRule{
+					FromEndpoints: []EndpointSelector{WildcardEndpointSelector},
+				},
+				ToPorts: []PortRule{{
+					Ports: []PortProtocol{
+						{Port: "443", Protocol: ProtoTCP},
+					},
+					Listener: &Listener{
+						EnvoyConfig: &EnvoyConfig{
+							Name: "test-config",
+						},
+						Name: "myCustomListener",
+					},
+				}},
+			},
+		},
+	}
+	err = invalidPortRule.Sanitize()
+	c.Assert(err, Not(IsNil))
+	c.Assert(err.Error(), Equals, "Listener is not allowed on ingress (myCustomListener)")
+
+	// Rule is invalid because Listener is not allowed with L7 rules
+	invalidPortRule = Rule{
+		EndpointSelector: WildcardEndpointSelector,
+		Egress: []EgressRule{
+			{
+				EgressCommonRule: EgressCommonRule{
+					ToEndpoints: []EndpointSelector{WildcardEndpointSelector},
+				},
+				ToPorts: []PortRule{{
+					Ports: []PortProtocol{
+						{Port: "443", Protocol: ProtoTCP},
+					},
+					Listener: &Listener{
+						EnvoyConfig: &EnvoyConfig{
+							Name: "test-config",
+						},
+						Name: "myCustomListener",
+					},
+					Rules: &L7Rules{
+						HTTP: []PortRuleHTTP{
+							{Method: "GET", Path: "/"},
+						},
+					},
+				}},
+			},
+		},
+	}
+	err = invalidPortRule.Sanitize()
+	c.Assert(err, Not(IsNil))
+	c.Assert(err.Error(), Equals, "Listener is not allowed with L7 rules (myCustomListener)")
 }
 
 // This test ensures that L7 rules reject unspecified ports.
diff --git a/pkg/policy/api/utils_test.go b/pkg/policy/api/utils_test.go
index a9998fb722..4e740c764a 100644
--- a/pkg/policy/api/utils_test.go
+++ b/pkg/policy/api/utils_test.go
@@ -111,3 +111,28 @@ func (s *PolicyAPITestSuite) TestParseL4Proto(c *C) {
 	_, err = ParseL4Proto("foo2")
 	c.Assert(err, Not(IsNil))
 }
+
+func (s *PolicyAPITestSuite) TestResourceQualifiedName(c *C) {
+	var fullName, namespace, name, resource string
+
+	resource = "test-resource"
+	fullName = ResourceQualifiedName(namespace, name, resource)
+	c.Assert(fullName, Equals, "test-resource")
+
+	name = "test-name"
+	resource = "test-resource"
+	fullName = ResourceQualifiedName(namespace, name, resource)
+	c.Assert(fullName, Equals, "test-name/test-resource")
+
+	namespace = "test-namespace"
+	name = ""
+	resource = "test-resource"
+	fullName = ResourceQualifiedName(namespace, name, resource)
+	c.Assert(fullName, Equals, "test-namespace/test-resource")
+
+	namespace = "test-namespace"
+	name = "test-name"
+	resource = "test-resource"
+	fullName = ResourceQualifiedName(namespace, name, resource)
+	c.Assert(fullName, Equals, "test-namespace/test-name/test-resource")
+}
diff --git a/pkg/policy/l4_filter_deny_test.go b/pkg/policy/l4_filter_deny_test.go
index bfcf1e1ff7..016ecbbb08 100644
--- a/pkg/policy/l4_filter_deny_test.go
+++ b/pkg/policy/l4_filter_deny_test.go
@@ -561,6 +561,7 @@ func (ds *PolicyTestSuite) TestL3L4AllowRuleWithByL3DenyAll(c *C) {
 				L7Rules: api.L7Rules{
 					HTTP: []api.PortRuleHTTP{{Path: "/", Method: "GET"}},
 				},
+				isRedirect: true,
 			},
 		},
 		Ingress:          true,
@@ -638,6 +639,7 @@ func (ds *PolicyTestSuite) TestL3L4AllowRuleWithByL3DenyAll(c *C) {
 				L7Rules: api.L7Rules{
 					HTTP: []api.PortRuleHTTP{{Path: "/", Method: "GET"}},
 				},
+				isRedirect: true,
 			},
 		},
 		Ingress:          true,
diff --git a/pkg/policy/l4_filter_test.go b/pkg/policy/l4_filter_test.go
index fe8d49f4dc..4692593909 100644
--- a/pkg/policy/l4_filter_test.go
+++ b/pkg/policy/l4_filter_test.go
@@ -52,6 +52,11 @@ var (
 
 type testPolicyContextType struct {
 	isDeny bool
+	ns     string
+}
+
+func (p *testPolicyContextType) GetNamespace() string {
+	return p.ns
 }
 
 func (p *testPolicyContextType) GetSelectorCache() *SelectorCache {
@@ -286,6 +291,7 @@ func (ds *PolicyTestSuite) TestMergeAllowAllL3AndShadowedL7(c *C) {
 				L7Rules: api.L7Rules{
 					HTTP: []api.PortRuleHTTP{{Path: "/", Method: "GET"}, {}},
 				},
+				isRedirect: true,
 			},
 		},
 		Ingress:          true,
@@ -403,6 +409,7 @@ func (ds *PolicyTestSuite) TestMergeIdenticalAllowAllL3AndRestrictedL7HTTP(c *C)
 				L7Rules: api.L7Rules{
 					HTTP: []api.PortRuleHTTP{{Path: "/", Method: "GET"}},
 				},
+				isRedirect: true,
 			},
 		},
 		Ingress:          true,
@@ -488,6 +495,7 @@ func (ds *PolicyTestSuite) TestMergeIdenticalAllowAllL3AndRestrictedL7Kafka(c *C
 				L7Rules: api.L7Rules{
 					Kafka: []kafka.PortRule{{Topic: "foo"}},
 				},
+				isRedirect: true,
 			},
 		},
 		Ingress:          true,
@@ -790,6 +798,7 @@ func (ds *PolicyTestSuite) TestMergeTLSTCPPolicy(c *C) {
 				EnvoyHTTPRules:  nil,
 				CanShortCircuit: false,
 				L7Rules:         api.L7Rules{},
+				isRedirect:      true,
 			},
 		},
 		Ingress:          false,
@@ -881,6 +890,7 @@ func (ds *PolicyTestSuite) TestMergeTLSHTTPPolicy(c *C) {
 				L7Rules: api.L7Rules{
 					HTTP: []api.PortRuleHTTP{{}},
 				},
+				isRedirect: true,
 			},
 		},
 		Ingress:          false,
@@ -989,6 +999,7 @@ func (ds *PolicyTestSuite) TestMergeTLSSNIPolicy(c *C) {
 				L7Rules: api.L7Rules{
 					HTTP: []api.PortRuleHTTP{{}},
 				},
+				isRedirect: true,
 			},
 		},
 		Ingress:          false,
@@ -1003,6 +1014,298 @@ func (ds *PolicyTestSuite) TestMergeTLSSNIPolicy(c *C) {
 	log.Infof("res: %v", res)
 }
 
+func (ds *PolicyTestSuite) TestMergeListenerPolicy(c *C) {
+	policyContext := &testPolicyContextType{}
+
+	//
+	// no namespace in policyContext (Clusterwide policy): Can not refer to EnvoyConfig
+	//
+	egressRule := &rule{
+		Rule: api.Rule{
+			EndpointSelector: fooSelector,
+			Egress: []api.EgressRule{
+				{
+					EgressCommonRule: api.EgressCommonRule{
+						ToEndpoints: []api.EndpointSelector{endpointSelectorA},
+					},
+					ToPorts: []api.PortRule{{
+						Ports: []api.PortProtocol{
+							{Port: "443", Protocol: api.ProtoTCP},
+						},
+					}},
+				},
+				{
+					EgressCommonRule: api.EgressCommonRule{
+						ToEndpoints: []api.EndpointSelector{endpointSelectorC},
+					},
+					ToPorts: []api.PortRule{{
+						Ports: []api.PortProtocol{
+							{Port: "443", Protocol: api.ProtoTCP},
+						},
+						Listener: &api.Listener{
+							EnvoyConfig: &api.EnvoyConfig{
+								Kind: "CiliumEnvoyConfig",
+								Name: "test-cec",
+							},
+							Name: "test",
+						},
+					}},
+				},
+			},
+		}}
+
+	buffer := new(bytes.Buffer)
+	ctxFromFoo := SearchContext{From: labels.ParseSelectLabelArray("foo"), Trace: TRACE_VERBOSE}
+	ctxFromFoo.Logging = stdlog.New(buffer, "", 0)
+	c.Log(buffer)
+
+	err := egressRule.Sanitize()
+	c.Assert(err, IsNil)
+
+	state := traceState{}
+	res, err := egressRule.resolveEgressPolicy(policyContext, &ctxFromFoo, &state, L4PolicyMap{}, nil, nil)
+	c.Log(buffer)
+	c.Assert(err, ErrorMatches, "Listener \"test\" in CCNP can not use Kind CiliumEnvoyConfig")
+	c.Assert(res, IsNil)
+
+	//
+	// no namespace in policyContext (Clusterwide policy): Must to ClusterwideEnvoyConfig
+	//
+	egressRule = &rule{
+		Rule: api.Rule{
+			EndpointSelector: fooSelector,
+			Egress: []api.EgressRule{
+				{
+					EgressCommonRule: api.EgressCommonRule{
+						ToEndpoints: []api.EndpointSelector{endpointSelectorA},
+					},
+					ToPorts: []api.PortRule{{
+						Ports: []api.PortProtocol{
+							{Port: "443", Protocol: api.ProtoTCP},
+						},
+					}},
+				},
+				{
+					EgressCommonRule: api.EgressCommonRule{
+						ToEndpoints: []api.EndpointSelector{endpointSelectorC},
+					},
+					ToPorts: []api.PortRule{{
+						Ports: []api.PortProtocol{
+							{Port: "443", Protocol: api.ProtoTCP},
+						},
+						Listener: &api.Listener{
+							EnvoyConfig: &api.EnvoyConfig{
+								Kind: "CiliumClusterwideEnvoyConfig",
+								Name: "shared-cec",
+							},
+							Name: "test",
+						},
+					}},
+				},
+			},
+		}}
+
+	buffer = new(bytes.Buffer)
+	ctxFromFoo = SearchContext{From: labels.ParseSelectLabelArray("foo"), Trace: TRACE_VERBOSE}
+	ctxFromFoo.Logging = stdlog.New(buffer, "", 0)
+	c.Log(buffer)
+
+	err = egressRule.Sanitize()
+	c.Assert(err, IsNil)
+
+	state = traceState{}
+	res, err = egressRule.resolveEgressPolicy(policyContext, &ctxFromFoo, &state, L4PolicyMap{}, nil, nil)
+	c.Log(buffer)
+	c.Assert(err, IsNil)
+	c.Assert(res, Not(IsNil))
+
+	// Since cachedSelectorA's map entry is 'nil', it will not be redirected to the proxy.
+	expected := L4PolicyMap{"443/TCP": &L4Filter{
+		Port:     443,
+		Protocol: api.ProtoTCP,
+		U8Proto:  6,
+		wildcard: nil,
+		L7Parser: ParserTypeCRD,
+		L7RulesPerSelector: L7DataMap{
+			cachedSelectorA: nil, // no proxy redirect
+			cachedSelectorC: &PerSelectorPolicy{
+				EnvoyHTTPRules:  nil,
+				CanShortCircuit: false,
+				isRedirect:      true,
+			},
+		},
+		Listener:         "shared-cec/test",
+		Ingress:          false,
+		DerivedFromRules: labels.LabelArrayList{nil},
+	}}
+
+	c.Assert(res, checker.Equals, expected)
+
+	l4Filter := res["443/TCP"]
+	c.Assert(l4Filter, Not(IsNil))
+	c.Assert(l4Filter.L7Parser, Equals, ParserTypeCRD)
+	log.Infof("res: %v", res)
+
+	//
+	// namespace in policyContext (Namespaced policy): Can refer to EnvoyConfig
+	//
+	egressRule = &rule{
+		Rule: api.Rule{
+			EndpointSelector: fooSelector,
+			Egress: []api.EgressRule{
+				{
+					EgressCommonRule: api.EgressCommonRule{
+						ToEndpoints: []api.EndpointSelector{endpointSelectorA},
+					},
+					ToPorts: []api.PortRule{{
+						Ports: []api.PortProtocol{
+							{Port: "443", Protocol: api.ProtoTCP},
+						},
+					}},
+				},
+				{
+					EgressCommonRule: api.EgressCommonRule{
+						ToEndpoints: []api.EndpointSelector{endpointSelectorC},
+					},
+					ToPorts: []api.PortRule{{
+						Ports: []api.PortProtocol{
+							{Port: "443", Protocol: api.ProtoTCP},
+						},
+						Listener: &api.Listener{
+							EnvoyConfig: &api.EnvoyConfig{
+								Kind: "CiliumEnvoyConfig",
+								Name: "test-cec",
+							},
+							Name: "test",
+						},
+					}},
+				},
+			},
+		}}
+
+	buffer = new(bytes.Buffer)
+	ctxFromFoo = SearchContext{From: labels.ParseSelectLabelArray("foo"), Trace: TRACE_VERBOSE}
+	ctxFromFoo.Logging = stdlog.New(buffer, "", 0)
+	c.Log(buffer)
+
+	err = egressRule.Sanitize()
+	c.Assert(err, IsNil)
+
+	state = traceState{}
+	policyContext.ns = "default"
+	res, err = egressRule.resolveEgressPolicy(policyContext, &ctxFromFoo, &state, L4PolicyMap{}, nil, nil)
+	c.Log(buffer)
+	c.Assert(err, IsNil)
+	c.Assert(res, Not(IsNil))
+
+	// Since cachedSelectorA's map entry is 'nil', it will not be redirected to the proxy.
+	expected = L4PolicyMap{"443/TCP": &L4Filter{
+		Port:     443,
+		Protocol: api.ProtoTCP,
+		U8Proto:  6,
+		wildcard: nil,
+		L7Parser: ParserTypeCRD,
+		L7RulesPerSelector: L7DataMap{
+			cachedSelectorA: nil, // no proxy redirect
+			cachedSelectorC: &PerSelectorPolicy{
+				EnvoyHTTPRules:  nil,
+				CanShortCircuit: false,
+				isRedirect:      true,
+			},
+		},
+		Listener:         "default/test-cec/test",
+		Ingress:          false,
+		DerivedFromRules: labels.LabelArrayList{nil},
+	}}
+
+	c.Assert(res, checker.Equals, expected)
+
+	l4Filter = res["443/TCP"]
+	c.Assert(l4Filter, Not(IsNil))
+	c.Assert(l4Filter.L7Parser, Equals, ParserTypeCRD)
+	log.Infof("res: %v", res)
+
+	//
+	// namespace in policyContext (Namespaced policy): Can refer to Cluster-socoped
+	// CiliumClusterwideEnvoyConfig
+	//
+	egressRule = &rule{
+		Rule: api.Rule{
+			EndpointSelector: fooSelector,
+			Egress: []api.EgressRule{
+				{
+					EgressCommonRule: api.EgressCommonRule{
+						ToEndpoints: []api.EndpointSelector{endpointSelectorA},
+					},
+					ToPorts: []api.PortRule{{
+						Ports: []api.PortProtocol{
+							{Port: "443", Protocol: api.ProtoTCP},
+						},
+					}},
+				},
+				{
+					EgressCommonRule: api.EgressCommonRule{
+						ToEndpoints: []api.EndpointSelector{endpointSelectorC},
+					},
+					ToPorts: []api.PortRule{{
+						Ports: []api.PortProtocol{
+							{Port: "443", Protocol: api.ProtoTCP},
+						},
+						Listener: &api.Listener{
+							EnvoyConfig: &api.EnvoyConfig{
+								Kind: "CiliumClusterwideEnvoyConfig",
+								Name: "shared-cec",
+							},
+							Name: "test",
+						},
+					}},
+				},
+			},
+		}}
+
+	buffer = new(bytes.Buffer)
+	ctxFromFoo = SearchContext{From: labels.ParseSelectLabelArray("foo"), Trace: TRACE_VERBOSE}
+	ctxFromFoo.Logging = stdlog.New(buffer, "", 0)
+	c.Log(buffer)
+
+	err = egressRule.Sanitize()
+	c.Assert(err, IsNil)
+
+	state = traceState{}
+	policyContext.ns = "default"
+	res, err = egressRule.resolveEgressPolicy(policyContext, &ctxFromFoo, &state, L4PolicyMap{}, nil, nil)
+	c.Log(buffer)
+	c.Assert(err, IsNil)
+	c.Assert(res, Not(IsNil))
+
+	// Since cachedSelectorA's map entry is 'nil', it will not be redirected to the proxy.
+	expected = L4PolicyMap{"443/TCP": &L4Filter{
+		Port:     443,
+		Protocol: api.ProtoTCP,
+		U8Proto:  6,
+		wildcard: nil,
+		L7Parser: ParserTypeCRD,
+		L7RulesPerSelector: L7DataMap{
+			cachedSelectorA: nil, // no proxy redirect
+			cachedSelectorC: &PerSelectorPolicy{
+				EnvoyHTTPRules:  nil,
+				CanShortCircuit: false,
+				isRedirect:      true,
+			},
+		},
+		Listener:         "shared-cec/test",
+		Ingress:          false,
+		DerivedFromRules: labels.LabelArrayList{nil},
+	}}
+
+	c.Assert(res, checker.Equals, expected)
+
+	l4Filter = res["443/TCP"]
+	c.Assert(l4Filter, Not(IsNil))
+	c.Assert(l4Filter.L7Parser, Equals, ParserTypeCRD)
+	log.Infof("res: %v", res)
+}
+
 // Case 6: allow all at L3/L7 in one rule, and select an endpoint and allow all on L7
 // in another rule. Should resolve to just allowing all on L3/L7 (first rule
 // shadows the second).
@@ -1193,6 +1496,7 @@ func (ds *PolicyTestSuite) TestL3RuleWithL7RulePartiallyShadowedByL3AllowAll(c *
 				L7Rules: api.L7Rules{
 					HTTP: []api.PortRuleHTTP{{Path: "/", Method: "GET"}},
 				},
+				isRedirect: true,
 			},
 		},
 		Ingress:          true,
@@ -1268,6 +1572,7 @@ func (ds *PolicyTestSuite) TestL3RuleWithL7RulePartiallyShadowedByL3AllowAll(c *
 				L7Rules: api.L7Rules{
 					HTTP: []api.PortRuleHTTP{{Path: "/", Method: "GET"}},
 				},
+				isRedirect: true,
 			},
 		},
 		Ingress:          true,
@@ -1355,11 +1660,13 @@ func (ds *PolicyTestSuite) TestL3RuleWithL7RuleShadowedByL3AllowAll(c *C) {
 				L7Rules: api.L7Rules{
 					HTTP: []api.PortRuleHTTP{{Path: "/", Method: "GET"}},
 				},
+				isRedirect: true,
 			},
 			cachedSelectorA: &PerSelectorPolicy{
 				L7Rules: api.L7Rules{
 					HTTP: []api.PortRuleHTTP{{Path: "/", Method: "GET"}},
 				},
+				isRedirect: true,
 			},
 		},
 		Ingress:          true,
@@ -1439,11 +1746,13 @@ func (ds *PolicyTestSuite) TestL3RuleWithL7RuleShadowedByL3AllowAll(c *C) {
 				L7Rules: api.L7Rules{
 					HTTP: []api.PortRuleHTTP{{Path: "/", Method: "GET"}},
 				},
+				isRedirect: true,
 			},
 			cachedSelectorA: &PerSelectorPolicy{
 				L7Rules: api.L7Rules{
 					HTTP: []api.PortRuleHTTP{{Path: "/", Method: "GET"}},
 				},
+				isRedirect: true,
 			},
 		},
 		Ingress:          true,
@@ -1642,11 +1951,13 @@ func (ds *PolicyTestSuite) TestMergingWithDifferentEndpointsSelectedAllowSameL7(
 				L7Rules: api.L7Rules{
 					HTTP: []api.PortRuleHTTP{{Path: "/", Method: "GET"}},
 				},
+				isRedirect: true,
 			},
 			cachedSelectorA: &PerSelectorPolicy{
 				L7Rules: api.L7Rules{
 					HTTP: []api.PortRuleHTTP{{Path: "/", Method: "GET"}},
 				},
+				isRedirect: true,
 			},
 		},
 		Ingress:          true,
@@ -1798,6 +2109,7 @@ func (ds *PolicyTestSuite) TestAllowingLocalhostShadowsL7(c *C) {
 				L7Rules: api.L7Rules{
 					HTTP: []api.PortRuleHTTP{{Path: "/", Method: "GET"}},
 				},
+				isRedirect: true,
 			},
 			cachedSelectorHost: nil, // no proxy redirect
 		},
diff --git a/pkg/policy/l4_test.go b/pkg/policy/l4_test.go
index a88b62e459..541f7137e3 100644
--- a/pkg/policy/l4_test.go
+++ b/pkg/policy/l4_test.go
@@ -6,6 +6,7 @@ package policy
 import (
 	"bytes"
 	"encoding/json"
+	"fmt"
 	"sort"
 
 	"github.com/kr/pretty"
@@ -17,6 +18,14 @@ import (
 	"github.com/cilium/cilium/pkg/policy/api"
 )
 
+func (s *PolicyTestSuite) TestRedirectType(c *C) {
+	c.Assert(redirectTypeNone, Equals, redirectTypes(0))
+	c.Assert(redirectTypeDNS, Equals, redirectTypes(0x1))
+	c.Assert(redirectTypeEnvoy, Equals, redirectTypes(0x2))
+	c.Assert(redirectTypeProxylib, Equals, redirectTypes(0x4)|redirectTypeEnvoy)
+	c.Assert(redirectTypeProxylib&redirectTypeEnvoy, Equals, redirectTypeEnvoy)
+}
+
 func (s *PolicyTestSuite) TestParserTypeMerge(c *C) {
 	for _, t := range []struct {
 		a, b, c L7ParserType
@@ -46,6 +55,23 @@ func (s *PolicyTestSuite) TestParserTypeMerge(c *C) {
 		{ParserTypeNone, ParserTypeTLS, ParserTypeTLS, true},
 		{ParserTypeTLS, ParserTypeNone, ParserTypeTLS, true},
 
+		{ParserTypeNone, ParserTypeCRD, ParserTypeCRD, true},
+		{ParserTypeCRD, ParserTypeNone, ParserTypeCRD, true},
+
+		// None of the actual parser types can be promoted to CRD
+
+		{ParserTypeHTTP, ParserTypeCRD, ParserTypeNone, false},
+		{ParserTypeCRD, ParserTypeHTTP, ParserTypeNone, false},
+
+		{ParserTypeTLS, ParserTypeCRD, ParserTypeNone, false},
+		{ParserTypeCRD, ParserTypeTLS, ParserTypeNone, false},
+
+		{ParserTypeKafka, ParserTypeCRD, ParserTypeNone, false},
+		{ParserTypeCRD, ParserTypeKafka, ParserTypeNone, false},
+
+		{L7ParserType("foo"), ParserTypeCRD, ParserTypeNone, false},
+		{ParserTypeCRD, L7ParserType("foo"), ParserTypeNone, false},
+
 		// TLS can also be promoted to any other type except for DNS (but not demoted to
 		// None)
 
@@ -60,6 +86,9 @@ func (s *PolicyTestSuite) TestParserTypeMerge(c *C) {
 
 		// DNS does not merge with anything else
 
+		{ParserTypeCRD, ParserTypeDNS, ParserTypeNone, false},
+		{ParserTypeDNS, ParserTypeCRD, ParserTypeNone, false},
+
 		{ParserTypeTLS, ParserTypeDNS, ParserTypeNone, false},
 		{ParserTypeDNS, ParserTypeTLS, ParserTypeNone, false},
 
@@ -89,6 +118,9 @@ func (s *PolicyTestSuite) TestParserTypeMerge(c *C) {
 		} else {
 			c.Assert(err, Not(Equals), nil)
 		}
+		if res != t.c {
+			fmt.Printf("Merge %s with %s, expecting %s\n", t.a, t.b, t.c)
+		}
 		c.Assert(res, Equals, t.c)
 	}
 }
@@ -116,14 +148,12 @@ func (s *PolicyTestSuite) TestCreateL4Filter(c *C) {
 		filter, err := createL4IngressFilter(testPolicyContext, eps, nil, portrule, tuple, tuple.Protocol, nil)
 		c.Assert(err, IsNil)
 		c.Assert(len(filter.L7RulesPerSelector), Equals, 1)
-		c.Assert(filter.IsEnvoyRedirect(), Equals, true)
-		c.Assert(filter.IsProxylibRedirect(), Equals, false)
+		c.Assert(filter.redirectType(), Equals, redirectTypeEnvoy)
 
 		filter, err = createL4EgressFilter(testPolicyContext, eps, portrule, tuple, tuple.Protocol, nil, nil)
 		c.Assert(err, IsNil)
 		c.Assert(len(filter.L7RulesPerSelector), Equals, 1)
-		c.Assert(filter.IsEnvoyRedirect(), Equals, true)
-		c.Assert(filter.IsProxylibRedirect(), Equals, false)
+		c.Assert(filter.redirectType(), Equals, redirectTypeEnvoy)
 	}
 }
 
@@ -235,6 +265,7 @@ func (s *PolicyTestSuite) TestJSONMarshal(c *C) {
 		},
 	}
 
+	policy.Attach(testPolicyContext)
 	model = policy.GetModel()
 	c.Assert(model, NotNil)
 
diff --git a/pkg/policy/repository_test.go b/pkg/policy/repository_test.go
index 920d208248..5665f045fc 100644
--- a/pkg/policy/repository_test.go
+++ b/pkg/policy/repository_test.go
@@ -757,6 +757,7 @@ func (ds *PolicyTestSuite) TestWildcardL3RulesIngress(c *C) {
 					L7Rules: api.L7Rules{
 						Kafka: []kafka.PortRule{kafkaRule.Ingress[0].ToPorts[0].Rules.Kafka[0]},
 					},
+					isRedirect: true,
 				},
 			},
 			DerivedFromRules: labels.LabelArrayList{labelsKafka},
@@ -772,6 +773,7 @@ func (ds *PolicyTestSuite) TestWildcardL3RulesIngress(c *C) {
 					L7Rules: api.L7Rules{
 						HTTP: []api.PortRuleHTTP{httpRule.Ingress[0].ToPorts[0].Rules.HTTP[0]},
 					},
+					isRedirect: true,
 				},
 			},
 			DerivedFromRules: labels.LabelArrayList{labelsHTTP},
@@ -788,6 +790,7 @@ func (ds *PolicyTestSuite) TestWildcardL3RulesIngress(c *C) {
 						L7Proto: "tester",
 						L7:      []api.PortRuleL7{l7Rule.Ingress[0].ToPorts[0].Rules.L7[0]},
 					},
+					isRedirect: true,
 				},
 			},
 			DerivedFromRules: labels.LabelArrayList{labelsL7},
@@ -922,6 +925,7 @@ func (ds *PolicyTestSuite) TestWildcardL4RulesIngress(c *C) {
 					L7Rules: api.L7Rules{
 						HTTP: []api.PortRuleHTTP{httpRule.Ingress[0].ToPorts[0].Rules.HTTP[0]},
 					},
+					isRedirect: true,
 				},
 			},
 			DerivedFromRules: labels.LabelArrayList{labelsL4HTTP, labelsL7HTTP},
@@ -938,6 +942,7 @@ func (ds *PolicyTestSuite) TestWildcardL4RulesIngress(c *C) {
 					L7Rules: api.L7Rules{
 						Kafka: []kafka.PortRule{kafkaRule.Ingress[0].ToPorts[0].Rules.Kafka[0]},
 					},
+					isRedirect: true,
 				},
 			},
 			DerivedFromRules: labels.LabelArrayList{labelsL4Kafka, labelsL7Kafka},
@@ -1250,6 +1255,7 @@ func (ds *PolicyTestSuite) TestWildcardL3RulesEgress(c *C) {
 					L7Rules: api.L7Rules{
 						DNS: []api.PortRuleDNS{dnsRule.Egress[0].ToPorts[0].Rules.DNS[0]},
 					},
+					isRedirect: true,
 				},
 			},
 			DerivedFromRules: labels.LabelArrayList{labelsDNS},
@@ -1265,6 +1271,7 @@ func (ds *PolicyTestSuite) TestWildcardL3RulesEgress(c *C) {
 					L7Rules: api.L7Rules{
 						HTTP: []api.PortRuleHTTP{httpRule.Egress[0].ToPorts[0].Rules.HTTP[0]},
 					},
+					isRedirect: true,
 				},
 			},
 			DerivedFromRules: labels.LabelArrayList{labelsHTTP},
@@ -1436,6 +1443,7 @@ func (ds *PolicyTestSuite) TestWildcardL4RulesEgress(c *C) {
 					L7Rules: api.L7Rules{
 						HTTP: []api.PortRuleHTTP{httpRule.Egress[0].ToPorts[0].Rules.HTTP[0]},
 					},
+					isRedirect: true,
 				},
 			},
 			DerivedFromRules: labels.LabelArrayList{labelsL3HTTP, labelsL7HTTP},
@@ -1452,6 +1460,7 @@ func (ds *PolicyTestSuite) TestWildcardL4RulesEgress(c *C) {
 					L7Rules: api.L7Rules{
 						DNS: []api.PortRuleDNS{dnsRule.Egress[0].ToPorts[0].Rules.DNS[0]},
 					},
+					isRedirect: true,
 				},
 			},
 			DerivedFromRules: labels.LabelArrayList{labelsL3DNS, labelsL7DNS},
@@ -1557,6 +1566,7 @@ func (ds *PolicyTestSuite) TestWildcardCIDRRulesEgress(c *C) {
 							Path:    "/",
 						}},
 					},
+					isRedirect: true,
 				},
 			},
 			DerivedFromRules: labels.LabelArrayList{labelsHTTP},
@@ -1693,6 +1703,7 @@ func (ds *PolicyTestSuite) TestWildcardL3RulesIngressFromEntities(c *C) {
 					L7Rules: api.L7Rules{
 						Kafka: []kafka.PortRule{kafkaRule.Ingress[0].ToPorts[0].Rules.Kafka[0]},
 					},
+					isRedirect: true,
 				},
 			},
 			DerivedFromRules: labels.LabelArrayList{labelsKafka},
@@ -1708,6 +1719,7 @@ func (ds *PolicyTestSuite) TestWildcardL3RulesIngressFromEntities(c *C) {
 					L7Rules: api.L7Rules{
 						HTTP: []api.PortRuleHTTP{httpRule.Ingress[0].ToPorts[0].Rules.HTTP[0]},
 					},
+					isRedirect: true,
 				},
 			},
 			DerivedFromRules: labels.LabelArrayList{labelsHTTP},
@@ -1831,6 +1843,7 @@ func (ds *PolicyTestSuite) TestWildcardL3RulesEgressToEntities(c *C) {
 					L7Rules: api.L7Rules{
 						DNS: []api.PortRuleDNS{dnsRule.Egress[0].ToPorts[0].Rules.DNS[0]},
 					},
+					isRedirect: true,
 				},
 			},
 			DerivedFromRules: labels.LabelArrayList{labelsDNS},
@@ -1846,6 +1859,7 @@ func (ds *PolicyTestSuite) TestWildcardL3RulesEgressToEntities(c *C) {
 					L7Rules: api.L7Rules{
 						HTTP: []api.PortRuleHTTP{httpRule.Egress[0].ToPorts[0].Rules.HTTP[0]},
 					},
+					isRedirect: true,
 				},
 			},
 			DerivedFromRules: labels.LabelArrayList{labelsHTTP},
@@ -1968,6 +1982,7 @@ func (ds *PolicyTestSuite) TestMinikubeGettingStarted(c *C) {
 				L7Rules: api.L7Rules{
 					HTTP: []api.PortRuleHTTP{{Method: "GET", Path: "/"}, {}},
 				},
+				isRedirect: true,
 			},
 		},
 		Ingress:          true,
diff --git a/pkg/policy/resolve_test.go b/pkg/policy/resolve_test.go
index 1a082367be..679f1cce54 100644
--- a/pkg/policy/resolve_test.go
+++ b/pkg/policy/resolve_test.go
@@ -311,12 +311,14 @@ func (ds *PolicyTestSuite) TestL7WithIngressWildcard(c *C) {
 									HTTP: []api.PortRuleHTTP{{Method: "GET", Path: "/good"}},
 								},
 								CanShortCircuit: true,
+								isRedirect:      true,
 							},
 						},
 						DerivedFromRules: labels.LabelArrayList{nil},
 					},
 				},
-				Egress: L4PolicyMap{},
+				Egress:        L4PolicyMap{},
+				redirectTypes: redirectTypeEnvoy,
 			},
 			IngressPolicyEnabled: true,
 			EgressPolicyEnabled:  false,
@@ -405,13 +407,15 @@ func (ds *PolicyTestSuite) TestL7WithLocalHostWildcardd(c *C) {
 									HTTP: []api.PortRuleHTTP{{Method: "GET", Path: "/good"}},
 								},
 								CanShortCircuit: true,
+								isRedirect:      true,
 							},
 							cachedSelectorHost: nil,
 						},
 						DerivedFromRules: labels.LabelArrayList{nil},
 					},
 				},
-				Egress: L4PolicyMap{},
+				Egress:        L4PolicyMap{},
+				redirectTypes: redirectTypeEnvoy,
 			},
 			IngressPolicyEnabled: true,
 			EgressPolicyEnabled:  false,
diff --git a/pkg/policy/rule_test.go b/pkg/policy/rule_test.go
index 8e3866a14a..d781e104d2 100644
--- a/pkg/policy/rule_test.go
+++ b/pkg/policy/rule_test.go
@@ -60,7 +60,8 @@ func (ds *PolicyTestSuite) TestL4Policy(c *C) {
 	}
 	l7map := L7DataMap{
 		wildcardCachedSelector: &PerSelectorPolicy{
-			L7Rules: l7rules,
+			L7Rules:    l7rules,
+			isRedirect: true,
 		},
 	}
 
@@ -193,6 +194,7 @@ func (ds *PolicyTestSuite) TestL4Policy(c *C) {
 				L7Rules: api.L7Rules{
 					HTTP: []api.PortRuleHTTP{{Path: "/", Method: "GET"}, {}},
 				},
+				isRedirect: true,
 			},
 		},
 		Ingress:          true,
@@ -445,11 +447,13 @@ func (ds *PolicyTestSuite) TestMergeL7PolicyIngress(c *C) {
 				L7Rules: api.L7Rules{
 					HTTP: []api.PortRuleHTTP{{Path: "/", Method: "GET"}, {}},
 				},
+				isRedirect: true,
 			},
 			cachedFooSelector: &PerSelectorPolicy{
 				L7Rules: api.L7Rules{
 					HTTP: []api.PortRuleHTTP{{Path: "/", Method: "GET"}},
 				},
+				isRedirect: true,
 			},
 		},
 		Ingress:          true,
@@ -513,10 +517,12 @@ func (ds *PolicyTestSuite) TestMergeL7PolicyIngress(c *C) {
 	}
 	l7map := L7DataMap{
 		wildcardCachedSelector: &PerSelectorPolicy{
-			L7Rules: l7rules,
+			L7Rules:    l7rules,
+			isRedirect: true,
 		},
 		cachedFooSelector: &PerSelectorPolicy{
-			L7Rules: l7rules,
+			L7Rules:    l7rules,
+			isRedirect: true,
 		},
 	}
 
@@ -606,10 +612,12 @@ func (ds *PolicyTestSuite) TestMergeL7PolicyIngress(c *C) {
 	// The L3-dependent L7 rules are not merged together.
 	l7map = L7DataMap{
 		cachedFooSelector: &PerSelectorPolicy{
-			L7Rules: fooRules,
+			L7Rules:    fooRules,
+			isRedirect: true,
 		},
 		wildcardCachedSelector: &PerSelectorPolicy{
-			L7Rules: barRules,
+			L7Rules:    barRules,
+			isRedirect: true,
 		},
 	}
 	expected = L4PolicyMap{"80/TCP": &L4Filter{
@@ -690,11 +698,13 @@ func (ds *PolicyTestSuite) TestMergeL7PolicyEgress(c *C) {
 				L7Rules: api.L7Rules{
 					HTTP: []api.PortRuleHTTP{{Path: "/public", Method: "GET"}, {}},
 				},
+				isRedirect: true,
 			},
 			cachedFooSelector: &PerSelectorPolicy{
 				L7Rules: api.L7Rules{
 					HTTP: []api.PortRuleHTTP{{Path: "/private", Method: "GET"}},
 				},
+				isRedirect: true,
 			},
 		},
 		Ingress:          false,
@@ -770,11 +780,13 @@ func (ds *PolicyTestSuite) TestMergeL7PolicyEgress(c *C) {
 				L7Rules: api.L7Rules{
 					Kafka: []kafka.PortRule{{Topic: "foo"}, {}},
 				},
+				isRedirect: true,
 			},
 			cachedFooSelector: &PerSelectorPolicy{
 				L7Rules: api.L7Rules{
 					Kafka: []kafka.PortRule{{Topic: "foo"}},
 				},
+				isRedirect: true,
 			},
 		},
 		Ingress:          false,
@@ -851,10 +863,12 @@ func (ds *PolicyTestSuite) TestMergeL7PolicyEgress(c *C) {
 	// The l3-dependent l7 rules are not merged together.
 	l7map := L7DataMap{
 		cachedFooSelector: &PerSelectorPolicy{
-			L7Rules: fooRules,
+			L7Rules:    fooRules,
+			isRedirect: true,
 		},
 		wildcardCachedSelector: &PerSelectorPolicy{
-			L7Rules: barRules,
+			L7Rules:    barRules,
+			isRedirect: true,
 		},
 	}
 	expected = L4PolicyMap{"80/TCP": &L4Filter{
@@ -2127,6 +2141,7 @@ func (ds *PolicyTestSuite) TestL4WildcardMerge(c *C) {
 				L7Rules: api.L7Rules{
 					HTTP: []api.PortRuleHTTP{{Path: "/", Method: "GET"}},
 				},
+				isRedirect: true,
 			},
 		},
 		Ingress:          true,
@@ -2162,6 +2177,7 @@ func (ds *PolicyTestSuite) TestL4WildcardMerge(c *C) {
 					L7Proto: "testparser",
 					L7:      []api.PortRuleL7{{"Key": "Value"}, {}},
 				},
+				isRedirect: true,
 			},
 		},
 		Ingress:          true,
@@ -2447,6 +2463,7 @@ func (ds *PolicyTestSuite) TestL3L4L7Merge(c *C) {
 				L7Rules: api.L7Rules{
 					HTTP: []api.PortRuleHTTP{{Path: "/", Method: "GET"}},
 				},
+				isRedirect: true,
 			},
 		},
 		Ingress:          true,
@@ -2510,6 +2527,7 @@ func (ds *PolicyTestSuite) TestL3L4L7Merge(c *C) {
 				L7Rules: api.L7Rules{
 					HTTP: []api.PortRuleHTTP{{Path: "/", Method: "GET"}},
 				},
+				isRedirect: true,
 			},
 		},
 		Ingress:          true,
diff --git a/pkg/proxy/proxy_test.go b/pkg/proxy/proxy_test.go
index fc7d53b17e..185bd89465 100644
--- a/pkg/proxy/proxy_test.go
+++ b/pkg/proxy/proxy_test.go
@@ -35,7 +35,7 @@ func (s *ProxySuite) TestPortAllocator(c *C) {
 	p := StartProxySupport(10000, 20000, stateDir, nil, nil, mockDatapathUpdater, nil,
 		testipcache.NewMockIPCache())
 
-	port, err := p.AllocateProxyPort("listener1", true)
+	port, err := p.AllocateProxyPort("listener1", false)
 	c.Assert(err, IsNil)
 	c.Assert(port, Not(Equals), 0)
 
@@ -44,15 +44,15 @@ func (s *ProxySuite) TestPortAllocator(c *C) {
 	c.Assert(port1, Equals, port)
 
 	// Another allocation for the same name gets the same port
-	port1a, err := p.AllocateProxyPort("listener1", true)
+	port1a, err := p.AllocateProxyPort("listener1", false)
 	c.Assert(err, IsNil)
 	c.Assert(port1a, Equals, port1)
 
-	name, pp := findProxyPortByType(ProxyTypeCRD, true)
+	name, pp := findProxyPortByType(ProxyTypeCRD, "listener1", false)
 	c.Assert(name, Equals, "listener1")
 	c.Assert(pp.proxyType, Equals, ProxyTypeCRD)
 	c.Assert(pp.proxyPort, Equals, port)
-	c.Assert(pp.ingress, Equals, true)
+	c.Assert(pp.ingress, Equals, false)
 	c.Assert(pp.configured, Equals, true)
 	c.Assert(pp.isStatic, Equals, false)
 	c.Assert(pp.nRedirects, Equals, 0)
@@ -77,7 +77,7 @@ func (s *ProxySuite) TestPortAllocator(c *C) {
 	c.Assert(err, IsNil)
 	c.Assert(port2, Not(Equals), port)
 	c.Assert(pp.proxyType, Equals, ProxyTypeCRD)
-	c.Assert(pp.ingress, Equals, true)
+	c.Assert(pp.ingress, Equals, false)
 	c.Assert(pp.proxyPort, Equals, port2)
 	c.Assert(pp.configured, Equals, true)
 	c.Assert(pp.isStatic, Equals, false)
@@ -129,7 +129,7 @@ func (s *ProxySuite) TestPortAllocator(c *C) {
 	c.Assert(port3, Not(Equals), port2)
 	c.Assert(port3, Not(Equals), port1)
 	c.Assert(pp.proxyType, Equals, ProxyTypeCRD)
-	c.Assert(pp.ingress, Equals, true)
+	c.Assert(pp.ingress, Equals, false)
 	c.Assert(pp.proxyPort, Equals, port3)
 	c.Assert(pp.configured, Equals, true)
 	c.Assert(pp.isStatic, Equals, false)
@@ -159,7 +159,7 @@ func (s *ProxySuite) TestPortAllocator(c *C) {
 	c.Assert(err, IsNil)
 	c.Assert(port4, Equals, port3)
 	c.Assert(pp.proxyType, Equals, ProxyTypeCRD)
-	c.Assert(pp.ingress, Equals, true)
+	c.Assert(pp.ingress, Equals, false)
 	c.Assert(pp.proxyPort, Equals, port4)
 	c.Assert(pp.configured, Equals, true)
 	c.Assert(pp.isStatic, Equals, false)