Repository: kubernetes/kubernetes (language: go)

Issue:
Excessive conntrack cleanup causes high memory (12GB) and CPU usage when any Pod with a UDP port changes
### What happened?

We are encountering a severe performance issue in kube-proxy (v1.32) when any Pod with a UDP port is updated (e.g., CoreDNS). In the new kube-proxy implementation, changes to Services or Pods that expose UDP ports trigger a full conntrack cleanup. This cleanup process iterates over the entire conntrack table, leading to extremely high resource consumption—sometimes up to 12 GB of memory and 1.5 CPU cores per kube-proxy instance.

In a simple test, we observed 2,780 instances of the log message "Adding conntrack filter for cleanup", which caused an OOM when kube-proxy was limited to 256 MB of memory. Without that limit, kube-proxy memory usage spiked to 12 GB. On nodes with large conntrack tables, kube-proxy effectively becomes stuck, consuming all available memory each time there is a UDP endpoint change.

This issue appears to be systemic; every change in a Pod with a UDP port triggers all kube-proxy instances to perform the extensive cleanup. Currently, there is no option to disable or throttle this behavior, which disrupts cluster stability and can lead to service degradation or outages. We request that the cleanup logic be revised to target only the relevant conntrack entries or that a mechanism be provided to disable or limit this aggressive cleanup behavior.


https://github.com/kubernetes/kubernetes/pull/127318
https://github.com/kubernetes/kubernetes/issues/126130

### What did you expect to happen?

We expected kube-proxy to handle conntrack cleanup in a more efficient and targeted way. Even if it needs to scan a significant portion of the conntrack table, it should do so without causing a spike to 12 GB of memory usage. Ideally, it would either:

- Limit its cleanup to entries relevant to the specific changed UDP endpoint.
- Provide a way to configure or disable this aggressive cleanup process so it does not risk out-of-memory (OOM) events or excessively high CPU usage.

### How can we reproduce it (as minimally and precisely as possible)?

- Deploy multiple Pods that generate a high volume of DNS requests, for example:
- A simple Golang application making repeated DNS lookups without any caching mechanism.
- Observe kube-proxy resource usage (memory and CPU) on that node.
- Delete or update the coredns Pod (which also uses UDP DNS).
- Watch the logs and resource usage of kube-proxy closely, noting the surge in memory (potentially up to 12 GB) and CPU usage as it performs the conntrack cleanup.

### Anything else we need to know?

![Image](https://github.com/user-attachments/assets/f75f0bc0-e394-45fa-b323-2f6fc0570386)

<img width="1708" alt="Image" src="https://github.com/user-attachments/assets/a4ff03e2-156c-4b60-bf87-7caf53859e51" />

<img width="1679" alt="Image" src="https://github.com/user-attachments/assets/81b4bea7-5cf6-498e-9cd4-ec807f267912" />

### Kubernetes version

<details>

```console
$ kubectl version
Client Version: v1.31.2
Kustomize Version: v5.4.2
Server Version: v1.32.0-eks-5ca49cb
```

</details>


### Cloud provider

<details>
AWS
</details>


### OS version

<details>

```console
# On Linux: Amazon Linux 2
5.10.230-223.885.amzn2.aarch64

```

</details>


### Install tools

<details>
EKS
</details>


### Container runtime (CRI) and version (if applicable)

<details>
containerd://1.7.23
</details>


### Related plugins (CNI, CSI, ...) and versions (if applicable)

<details>
kube-proxy:v1.32.0-minimal-eksbuild.2
</details>



Discussion / hints:
/sig network
It might be a critical bug.
/cc @aroradaman @aojea @danwinship 

Since it was reported on EKS, for visibility:
/cc @dims 
@sergeylanzman if you haven't already please open a support ticket with AWS support
I tried to reproduce this locally with profiling enabled and found 
https://github.com/vishvananda/netlink/blob/b1ce50cfa9bea7652382b8c5ef6083f8d3f5f853/conntrack_linux.go#L175 to be the root cause of the memory spike. 

In **vishvananda/netlink** all the flow entries for which delete call fails are collected in an array, and bloating of this array is causing the memory spikes.



![Image](https://github.com/user-attachments/assets/3273afc8-025a-4cb9-9b82-dde012b24b60)

/triage accepted
/assign

[EDIT]

I introduced that change in **vishvananda/netlink** with the intention of collecting and aggregating all error messages.
https://github.com/vishvananda/netlink/pull/1014
I cross checked by pointing `vishvananda/netlink` to my debug branch https://github.com/vishvananda/netlink/compare/main...aroradaman:netlink:debug and observed the following logs in kube-proxy.
```
####################################################### Netlink ConntrackDeleteFilters
error occurred
length 7538373
####################################################### Netlink ConntrackDeleteFilters
```
Hi!

We encountered the same issue after upgrading kube-proxy to version 1.32.0-eksbuild.2 through EKS Add-ons. Following the upgrade, we observed a memory spike from 40Mi to 27Gi on m5.2xlarge nodes (8 vCPUs, 32 GiB RAM), causing pod evictions and impacting the cluster’s availability.

To mitigate the issue, we downgraded kube-proxy to v1.31 and also raised a support case with AWS.

Below is a capture from the kube-capacity tool before and after the upgrade:

```
# 1.31
POD                CPU REQUESTS   CPU LIMITS   CPU UTIL   MEMORY REQUESTS   MEMORY LIMITS   MEMORY UTIL
kube-proxy-jndmk   100m (1%)      0m (0%)      18m (0%)   0Mi (0%)          0Mi (0%)        40Mi (0%)

# 1.32
POD                CPU REQUESTS   CPU LIMITS   CPU UTIL      MEMORY REQUESTS   MEMORY LIMITS   MEMORY UTIL
kube-proxy-22hjt   100m (1%)      0m (0%)      1593m (20%)   0Mi (0%)          0Mi (0%)        26814Mi (95%)
```

Thanks!
/priority critical-urgent

> I cross checked by pointing `vishvananda/netlink` to my debug branch [vishvananda/netlink@main...aroradaman:netlink:debug](https://github.com/vishvananda/netlink/compare/main...aroradaman:netlink:debug) and observed the following logs in kube-proxy.
> 
> ```
> ####################################################### Netlink ConntrackDeleteFilters
> error occurred
> length 7538373
> ####################################################### Netlink ConntrackDeleteFilters
> ```

then what are we logging there?
@aroradaman  can you paste the error we are receiving ?

the for loop is entries and filters , so that 7538373 has to be equal to the number of entries x numer of filters
> > I cross checked by pointing `vishvananda/netlink` to my debug branch [vishvananda/netlink@main...aroradaman:netlink:debug](https://github.com/vishvananda/netlink/compare/main...aroradaman:netlink:debug) and observed the following logs in kube-proxy.
> > ```
> > ####################################################### Netlink ConntrackDeleteFilters
> > error occurred
> > length 7538373
> > ####################################################### Netlink ConntrackDeleteFilters
> > ```
> 
> then what are we logging there? [@aroradaman](https://github.com/aroradaman) can you paste the error we are receiving ?
> 
> the for loop is entries and filters , so that 7538373 has to be equal to the number of entries x numer of filters

ENOENT / "no such file or directory" was thrown, My assumption is that the conntrack entry got expired before deletion.
> > > I cross checked by pointing `vishvananda/netlink` to my debug branch [vishvananda/netlink@main...aroradaman:netlink:debug](https://github.com/vishvananda/netlink/compare/main...aroradaman:netlink:debug) and observed the following logs in kube-proxy.
> > > ```
> > > ####################################################### Netlink ConntrackDeleteFilters
> > > error occurred
> > > length 7538373
> > > ####################################################### Netlink ConntrackDeleteFilters
> > > ```
> > 
> > 
> > then what are we logging there? [@aroradaman](https://github.com/aroradaman) can you paste the error we are receiving ?
> > the for loop is entries and filters , so that 7538373 has to be equal to the number of entries x numer of filters
> 
> ENOENT / "no such file or directory" was thrown, My assumption is that the conntrack entry got expired before deletion.

what I;m surprised is that there are 7538373 entries on that cluster? is EKS doing a total conntrack flush? 
how can we get into this situation? do we have some bad logic in the reconciler?
> what I;m surprised is that there are 7538373 entries on that cluster? is EKS doing a total conntrack flush?
> how can we get into this situation? do we have some bad logic in the reconciler?

This is not on EKS, I tried to reproduce this locally bombarding the cluster with DNS lookup requests.
10 pods each running 10000 go-routines and doing DNS lookup in loop.
> This is not on EKS, I tried to reproduce this locally bombarding the cluster with DNS lookup requests.
> 10 pods each running 10000 go-routines and doing DNS lookup in loop.

yeah, but for the other reports claiming 12gb of memory ... how many entries are there?
just for tracking / release notes, is this a regression in kube 1.32?
> just for tracking / release notes, is this a regression in kube 1.32?

it is
I don't believe this is fixed. As far as I can tell the change only resolved the excessive memory used by the error buffer. 

We're still seeing kube-proxy cleaning up conntrack entries for things that it shouldn't, like flannel vxlan connections between nodes. This is visible if you increase the kube-proxy log level. While doing so it consumes excessive CPU time and memory, to the point where its liveness checks fail and the pod is continuously restarted.

From the above-linked issue:

> turned on the verbose logging and just see this constantly:
>```
>I0226 19:48:14.281563 1 filter.go:99] "Deleting conntrack entry" flow=<
>udp 17 src=10.1.40.120 dst=10.1.40.118 sport=57973 dport=8472 packets=12 bytes=1482 src=10.1.40.118 dst=10.1.40.120 sport=8472 dport=57973 packets=0 bytes=0 mark=0x0 start=1970-01-01 00:00:00 +0000 UTC stop=1970-01-01 00:00:00 +0000 UTC timeout=3(sec)
>I0226 19:48:14.281622 1 filter.go:99] "Deleting conntrack entry" flow=<
>udp 17 src=10.1.40.120 dst=10.1.40.118 sport=57973 dport=8472 packets=12 bytes=1482 src=10.1.40.118 dst=10.1.40.120 sport=8472 dport=57973 packets=0 bytes=0 mark=0x0 start=1970-01-01 00:00:00 +0000 UTC stop=1970-01-01 00:00:00 +0000 UTC timeout=3(sec)
>I0226 19:48:14.281683 1 filter.go:99] "Deleting conntrack entry" flow=<
>udp 17 src=10.1.40.120 dst=10.1.40.118 sport=57973 dport=8472 packets=12 bytes=1482 src=10.1.40.118 dst=10.1.40.120 sport=8472 dport=57973 packets=0 bytes=0 mark=0x0 start=1970-01-01 00:00:00 +0000 UTC stop=1970-01-01 00:00:00 +0000 UTC timeout=3(sec)
>I0226 19:48:14.281751 1 filter.go:99] "Deleting conntrack entry" flow=<
>udp 17 src=10.1.40.120 dst=10.1.40.118 sport=57973 dport=8472 packets=12 bytes=1482 src=10.1.40.118 dst=10.1.40.120 sport=8472 dport=57973 packets=0 bytes=0 mark=0x0 start=1970-01-01 00:00:00 +0000 UTC stop=1970-01-01 00:00:00 +0000 UTC timeout=3(sec)
>I0226 19:48:14.281811 1 filter.go:99] "Deleting conntrack entry" flow=<
>udp 17 src=10.1.40.120 dst=10.1.40.118 sport=57973 dport=8472 packets=12 bytes=1482 src=10.1.40.118 dst=10.1.40.120 sport=8472 dport=57973 packets=0 bytes=0 mark=0x0 start=1970-01-01 00:00:00 +0000 UTC stop=1970-01-01 00:00:00 +0000 UTC timeout=3(sec)
> I0226 19:48:14.281869 1 filter.go:99] "Deleting conntrack entry" flow=<
>udp 17 src=10.1.40.120 dst=10.1.40.118 sport=57973 dport=8472 packets=12 bytes=1482 src=10.1.40.118 dst=10.1.40.120 sport=8472 dport=57973 packets=0 bytes=0 mark=0x0 start=1970-01-01 00:00:00 +0000 UTC stop=1970-01-01 00:00:00 +0000 UTC timeout=3(sec)
>```

These are conntrack entries for flannel vxlan traffic between nodes, there is no reason kube-proxy should be deleting them.
cc @aroradaman 
Currently we clear all the UDP entries with --orig-dst=`<service_ip>` and --reply-src=`<!serving_endpoint_ip>`. 
https://github.com/kubernetes/kubernetes/blob/931ad2a9fdedaf1e47126f5b3e5880eb3708bfb2/pkg/proxy/conntrack/cleanup.go#L100-L107

service-ip here can be one of ClusterIP, LoadBalancerIP or ExternalIP. I think it is safe to assume that all flow entries with destination as a ClusterIP will be owned by kube-proxy, but not for LoadBalancerIPs and ExternalIPs.

From the [logs shared in rancher issue](https://github.com/rancher/rke2/issues/7818#issuecomment-2686128988) its is visible that one NodeIPs are being used as ExternalIPs and this lead reconciler to pickup all the flow entries with destination as NodeIPs.


I had this concern before implementing the reconciler  https://github.com/kubernetes/kubernetes/issues/126130#issuecomment-2344658176.
> By using CONNMARK / ct mark we can set a mark on the flow entry itself, which will help distinguish flows that are created or DNATed by the proxy. From a reconciliation perspective, this marking can be used as a filter to prevent accidental cleanup of non-kube-proxy owned flows.


I can think of three ways to figure out if a conntrack flow is owned by kube-proxy:
1. Using service port and target port for filtering
   --orig-dst=`<service_ip>` && --orig-port-dst=`<service_port>` && --reply-port-src=`<endpoint/target_port>`
   This is not a full-proof solution but should cover almost all the cases.
2. Using pod-cidr
    We can filter out kube-proxy owned flows first  using --orig-dst=`<service_ip>` and --reply-src=`<belongs to pod-cidr>`. 
    Though this can mis the cases of pods running in host network namespace.
3. Using conntrack labels (as Dan suggested [here](https://github.com/kubernetes/kubernetes/issues/126130#issuecomment-2348883750) )
 
[1] seems a straight forward way [2] and [3] may required API changes.

cc. @danwinship @aojea 

> We're still seeing kube-proxy cleaning up conntrack entries for things that it shouldn't, like flannel vxlan connections between nodes. This is visible if you increase the kube-proxy log level. While doing so it consumes excessive CPU time and memory, to the point where its liveness checks fail and the pod is continuously restarted.

/reopen

@brandond do you have an idea what part of  filter is causing the problem? I'd like to understand what IP is causing the filter to match

EDIT

~the kube-proxy logs should show the services at that point~

the infor is in the links , checking ...
@aojea: Reopened this issue.

<details>

In response to [this](https://github.com/kubernetes/kubernetes/issues/129982#issuecomment-2688944027):

>> We're still seeing kube-proxy cleaning up conntrack entries for things that it shouldn't, like flannel vxlan connections between nodes. This is visible if you increase the kube-proxy log level. While doing so it consumes excessive CPU time and memory, to the point where its liveness checks fail and the pod is continuously restarted.
>
>/reopen
>
>@brandond can you give an idea what part of  filter is causing the problem? I'd like to understand what IP is causing the filter to match


Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes-sigs/prow](https://github.com/kubernetes-sigs/prow/issues/new?title=Prow%20issue:) repository.
</details>
/assign @aojea 

Fix patch:
diff --git a/pkg/proxy/conntrack/cleanup.go b/pkg/proxy/conntrack/cleanup.go
index 5c8dded75efbf..8627e0b261d6c 100644
--- a/pkg/proxy/conntrack/cleanup.go
+++ b/pkg/proxy/conntrack/cleanup.go
@@ -21,6 +21,8 @@ package conntrack
 
 import (
 	"errors"
+	"net"
+	"strconv"
 	"time"
 
 	"github.com/vishvananda/netlink"
@@ -34,8 +36,14 @@ import (
 	netutils "k8s.io/utils/net"
 )
 
+// Kubernetes UDP services can be affected by stale conntrack entries.
+// These entries may point to endpoints that no longer exist,
+// leading to packet loss and connectivity problems.
+
 // CleanStaleEntries scans conntrack table and removes any entries
 // for a service that do not correspond to a serving endpoint.
+// List existing conntrack entries and calculate the desired conntrack state
+// based on the current Services and Endpoints.
 func CleanStaleEntries(ct Interface, ipFamily v1.IPFamily,
 	svcPortMap proxy.ServicePortMap, endpointsMap proxy.EndpointsMap) {
 
@@ -52,7 +60,7 @@ func CleanStaleEntries(ct Interface, ipFamily v1.IPFamily,
 		}
 	}
 
-	// serviceIPEndpointIPs maps service IPs (ClusterIP, LoadBalancerIPs and ExternalIPs)
+	// serviceIPEndpointIPs maps service IPs (ClusterIP, LoadBalancerIPs and ExternalIPs) and Service Port
 	// to the set of serving endpoint IPs.
 	serviceIPEndpointIPs := make(map[string]sets.Set[string])
 	// serviceNodePortEndpointIPs maps service NodePort to the set of serving endpoint IPs.
@@ -79,14 +87,28 @@ func CleanStaleEntries(ct Interface, ipFamily v1.IPFamily,
 			}
 		}
 
-		serviceIPEndpointIPs[svc.ClusterIP().String()] = endpointIPs
+		// a Service without endpoints does not require to clean the conntrack entries associated.
+		if endpointIPs.Len() == 0 {
+			continue
+		}
+
+		// we need to filter entries that are directed to a Service IP:Port frontend
+		// that does not have a backend as part of the endpoints IPs
+		portStr := strconv.Itoa(svc.Port())
+		// clusterIP:Port
+		serviceIPEndpointIPs[net.JoinHostPort(svc.ClusterIP().String(), portStr)] = endpointIPs
+		// loadbalancerIP:Port
 		for _, loadBalancerIP := range svc.LoadBalancerVIPs() {
-			serviceIPEndpointIPs[loadBalancerIP.String()] = endpointIPs
+			serviceIPEndpointIPs[net.JoinHostPort(loadBalancerIP.String(), portStr)] = endpointIPs
 		}
+		// externalIP:Port
 		for _, externalIP := range svc.ExternalIPs() {
-			serviceIPEndpointIPs[externalIP.String()] = endpointIPs
+			serviceIPEndpointIPs[net.JoinHostPort(externalIP.String(), portStr)] = endpointIPs
 		}
+		// we need to filter entries that are directed to a *:NodePort
+		// that does not have a backend as part of the endpoints IPs
 		if svc.NodePort() != 0 {
+			// *:NodePort
 			serviceNodePortEndpointIPs[svc.NodePort()] = endpointIPs
 		}
 	}
@@ -98,26 +120,25 @@ func CleanStaleEntries(ct Interface, ipFamily v1.IPFamily,
 			continue
 		}
 
-		origDst := entry.Forward.DstIP.String()
-		origPortDst := int(entry.Forward.DstPort)
-		replySrc := entry.Reverse.SrcIP.String()
+		origDst := entry.Forward.DstIP.String()   // match Service IP
+		origPortDst := int(entry.Forward.DstPort) // match Service Port
+		origPortDstStr := strconv.Itoa(origPortDst)
+		replySrc := entry.Reverse.SrcIP.String() // match Serving Endpoint IP
 
 		// if the original destination (--orig-dst) of the entry is service IP (ClusterIP,
-		// LoadBalancerIPs or ExternalIPs) and the reply source (--reply-src) is not IP of
-		// any serving endpoint, we clear the entry.
-		if _, ok := serviceIPEndpointIPs[origDst]; ok {
-			if !serviceIPEndpointIPs[origDst].Has(replySrc) {
-				filters = append(filters, filterForNAT(origDst, replySrc, v1.ProtocolUDP))
-			}
+		// LoadBalancerIPs or ExternalIPs) and (--orig-port-dst) of the flow is service Port
+		// and the reply source (--reply-src) is not IP of any serving endpoint, we clear the entry.
+		endpoints, ok := serviceIPEndpointIPs[net.JoinHostPort(origDst, origPortDstStr)]
+		if ok && !endpoints.Has(replySrc) {
+			filters = append(filters, filterForIPPortNAT(origDst, replySrc, entry.Forward.DstPort, v1.ProtocolUDP))
 		}
 
 		// if the original port destination (--orig-port-dst) of the flow is service
 		// NodePort and the reply source (--reply-src) is not IP of any serving endpoint,
 		// we clear the entry.
-		if _, ok := serviceNodePortEndpointIPs[origPortDst]; ok {
-			if !serviceNodePortEndpointIPs[origPortDst].Has(replySrc) {
-				filters = append(filters, filterForPortNAT(replySrc, origPortDst, v1.ProtocolUDP))
-			}
+		endpoints, ok = serviceNodePortEndpointIPs[origPortDst]
+		if ok && !endpoints.Has(replySrc) {
+			filters = append(filters, filterForPortNAT(replySrc, origPortDst, v1.ProtocolUDP))
 		}
 	}
 
@@ -145,14 +166,16 @@ var protocolMap = map[v1.Protocol]uint8{
 	v1.ProtocolSCTP: unix.IPPROTO_SCTP,
 }
 
-// filterForNAT returns *conntrackFilter to delete the conntrack entries for connections
-// specified by the destination IP (original direction) and source IP (reply direction).
-func filterForNAT(origin, dest string, protocol v1.Protocol) *conntrackFilter {
+// filterForIPPortNAT returns *conntrackFilter to delete the conntrack entries for connections
+// specified by the destination IP (original direction) and destination port (original direction)
+// and source IP (reply direction).
+func filterForIPPortNAT(origin, dest string, dstPort uint16, protocol v1.Protocol) *conntrackFilter {
 	klog.V(6).InfoS("Adding conntrack filter for cleanup", "org-dst", origin, "reply-src", dest, "protocol", protocol)
 	return &conntrackFilter{
 		protocol: protocolMap[protocol],
 		original: &connectionTuple{
-			dstIP: netutils.ParseIPSloppy(origin),
+			dstIP:   netutils.ParseIPSloppy(origin),
+			dstPort: dstPort,
 		},
 		reply: &connectionTuple{
 			srcIP: netutils.ParseIPSloppy(dest),
diff --git a/pkg/proxy/conntrack/fake.go b/pkg/proxy/conntrack/fake.go
index bedbb99b6e9ac..8902da7a6f2bb 100644
--- a/pkg/proxy/conntrack/fake.go
+++ b/pkg/proxy/conntrack/fake.go
@@ -25,7 +25,8 @@ import (
 
 // FakeInterface implements Interface by just recording entries that have been cleared.
 type FakeInterface struct {
-	entries []*netlink.ConntrackFlow
+	entries         []*netlink.ConntrackFlow
+	netlinkRequests int // try to get the estimated number of netlink request
 }
 
 var _ Interface = &FakeInterface{}
@@ -37,18 +38,30 @@ func NewFake() *FakeInterface {
 
 // ListEntries is part of Interface
 func (fake *FakeInterface) ListEntries(_ uint8) ([]*netlink.ConntrackFlow, error) {
-	return fake.entries, nil
+	entries := make([]*netlink.ConntrackFlow, len(fake.entries))
+	copy(entries, fake.entries)
+	// 1 netlink request to dump the table
+	// https://github.com/vishvananda/netlink/blob/0af32151e72b990c271ef6268e8aadb7e015f2bd/conntrack_linux.go#L93-L94
+	fake.netlinkRequests++
+	return entries, nil
 }
 
 // ClearEntries is part of Interface
 func (fake *FakeInterface) ClearEntries(_ uint8, filters ...netlink.CustomConntrackFilter) (int, error) {
 	var flows []*netlink.ConntrackFlow
 	before := len(fake.entries)
+	// 1 netlink request to dump the table
+	// https://github.com/vishvananda/netlink/blob/0af32151e72b990c271ef6268e8aadb7e015f2bd/conntrack_linux.go#L163
+	fake.netlinkRequests++
+
 	for _, flow := range fake.entries {
 		var matched bool
 		for _, filter := range filters {
 			matched = filter.MatchConntrackFlow(flow)
 			if matched {
+				// 1 netlink request to delete the flow
+				// https://github.com/vishvananda/netlink/blob/0af32151e72b990c271ef6268e8aadb7e015f2bd/conntrack_linux.go#L182
+				fake.netlinkRequests++
 				break
 			}
 		}


Test patch:
diff --git a/pkg/proxy/conntrack/cleanup_test.go b/pkg/proxy/conntrack/cleanup_test.go
index 68c8a8f1e2aea..ff72e73e78f84 100644
--- a/pkg/proxy/conntrack/cleanup_test.go
+++ b/pkg/proxy/conntrack/cleanup_test.go
@@ -21,9 +21,11 @@ package conntrack
 
 import (
 	"fmt"
+	"math/rand"
 	"sort"
 	"testing"
 
+	"github.com/google/go-cmp/cmp"
 	"github.com/stretchr/testify/require"
 	"github.com/vishvananda/netlink"
 	"golang.org/x/sys/unix"
@@ -50,8 +52,11 @@ const (
 	testNonServingEndpointIP = "10.240.1.5"
 	testDeletedEndpointIP    = "10.240.2.6"
 
-	testPort     = 8000
-	testNodePort = 32000
+	testServicePort     = 8000
+	testServiceNodePort = 32000
+	// testNonServicePort is used to mock conntrack flow entries which are not owned by
+	// kube-proxy and reconciler should not consider these for cleanup
+	testNonServicePort = 3000
 )
 
 func TestCleanStaleEntries(t *testing.T) {
@@ -74,19 +79,19 @@ func TestCleanStaleEntries(t *testing.T) {
 			Ports: []v1.ServicePort{
 				{
 					Name:     "test-tcp",
-					Port:     testPort,
+					Port:     testServicePort,
 					Protocol: v1.ProtocolTCP,
 				},
 				{
 					Name:     "test-udp",
-					Port:     testPort,
-					NodePort: testNodePort,
+					Port:     testServicePort,
+					NodePort: testServiceNodePort,
 					Protocol: v1.ProtocolUDP,
 				},
 				{
 					Name:     "test-sctp",
-					Port:     testPort,
-					NodePort: testNodePort,
+					Port:     testServicePort,
+					NodePort: testServiceNodePort,
 					Protocol: v1.ProtocolSCTP,
 				},
 			},
@@ -126,17 +131,17 @@ func TestCleanStaleEntries(t *testing.T) {
 		Ports: []discovery.EndpointPort{
 			{
 				Name:     ptr.To("test-tcp"),
-				Port:     ptr.To(int32(testPort)),
+				Port:     ptr.To(int32(testServicePort)),
 				Protocol: ptr.To(v1.ProtocolTCP),
 			},
 			{
 				Name:     ptr.To("test-udp"),
-				Port:     ptr.To(int32(testPort)),
+				Port:     ptr.To(int32(testServicePort)),
 				Protocol: ptr.To(v1.ProtocolUDP),
 			},
 			{
 				Name:     ptr.To("test-sctp"),
-				Port:     ptr.To(int32(testPort)),
+				Port:     ptr.To(int32(testServicePort)),
 				Protocol: ptr.To(v1.ProtocolSCTP),
 			},
 		},
@@ -212,35 +217,47 @@ func TestCleanStaleEntries(t *testing.T) {
 		}
 	}
 
-	// mock existing entries before cleanup
-	// we create 36 fake flow entries ( 3 Endpoints * 3 Protocols * ( 3 (ServiceIPs) + 1 (NodePort))
-	var mockEntries []*netlink.ConntrackFlow
-	// expectedEntries are the entries on which we will assert the cleanup logic
-	var expectedEntries []*netlink.ConntrackFlow
+	// The following mock conntrack flow entries `entriesBeforeCleanup` and `entriesAfterCleanup`
+	// represent conntrack flow entries before and after reconciler cleanup loop. Before cleanup,
+	// reconciler lists the conntrack flows, receiving `entriesBeforeCleanup` and after cleanup,
+	// we list the conntrack flows and assert them to match with `entriesAfterCleanup`.
+	// {entriesBeforeCleanup} - {entriesAfterCleanup} = entries cleared by conntrack reconciler
+	var entriesBeforeCleanup []*netlink.ConntrackFlow
+	// entriesBeforeCleanup - entriesAfterCleanup = entries cleared by conntrack reconciler
+	var entriesAfterCleanup []*netlink.ConntrackFlow
+
+	// we create 63 fake flow entries ( 3 Endpoints * 3 Protocols * ( 3 (ServiceIP:ServicePort) + 3 (ServiceIP:NonServicePort) + 1 (NodePort))
 	for _, dnatDest := range []string{testServingEndpointIP, testNonServingEndpointIP, testDeletedEndpointIP} {
 		for _, proto := range []uint8{unix.IPPROTO_TCP, unix.IPPROTO_UDP, unix.IPPROTO_SCTP} {
+
 			for _, origDest := range []string{testClusterIP, testLoadBalancerIP, testExternalIP} {
-				entry := &netlink.ConntrackFlow{
-					FamilyType: unix.AF_INET,
-					Forward: netlink.IPTuple{
-						DstIP:    netutils.ParseIPSloppy(origDest),
-						Protocol: proto,
-					},
-					Reverse: netlink.IPTuple{
-						Protocol: proto,
-						SrcIP:    netutils.ParseIPSloppy(dnatDest),
-					},
-				}
-				mockEntries = append(mockEntries, entry)
-				// we do not expect deleted or non-serving UDP endpoints flows to be present after cleanup
-				if !(proto == unix.IPPROTO_UDP && (dnatDest == testNonServingEndpointIP || dnatDest == testDeletedEndpointIP)) {
-					expectedEntries = append(expectedEntries, entry)
+				for _, port := range []uint16{testServicePort, testNonServicePort} {
+					entry := &netlink.ConntrackFlow{
+						FamilyType: unix.AF_INET,
+						Forward: netlink.IPTuple{
+							DstIP:    netutils.ParseIPSloppy(origDest),
+							DstPort:  port,
+							Protocol: proto,
+						},
+						Reverse: netlink.IPTuple{
+							Protocol: proto,
+							SrcIP:    netutils.ParseIPSloppy(dnatDest),
+						},
+					}
+					entriesBeforeCleanup = append(entriesBeforeCleanup, entry)
+					if proto == unix.IPPROTO_UDP && port == testServicePort && dnatDest != testServingEndpointIP {
+						// we do not expect UDP entries with destination port `testServicePort` and DNATed destination
+						// address not an address of serving endpoint to be present after cleanup.
+					} else {
+						entriesAfterCleanup = append(entriesAfterCleanup, entry)
+					}
 				}
 			}
+
 			entry := &netlink.ConntrackFlow{
 				FamilyType: unix.AF_INET,
 				Forward: netlink.IPTuple{
-					DstPort:  testNodePort,
+					DstPort:  testServiceNodePort,
 					Protocol: proto,
 				},
 				Reverse: netlink.IPTuple{
@@ -248,50 +265,264 @@ func TestCleanStaleEntries(t *testing.T) {
 					SrcIP:    netutils.ParseIPSloppy(dnatDest),
 				},
 			}
-			mockEntries = append(mockEntries, entry)
-			// we do not expect deleted or non-serving UDP endpoints entries to be present after cleanup
-			if !(proto == unix.IPPROTO_UDP && (dnatDest == testNonServingEndpointIP || dnatDest == testDeletedEndpointIP)) {
-				expectedEntries = append(expectedEntries, entry)
+			entriesBeforeCleanup = append(entriesBeforeCleanup, entry)
+			if proto == unix.IPPROTO_UDP && dnatDest != testServingEndpointIP {
+				// we do not expect UDP entries with DNATed destination address not
+				// an address of serving endpoint to be present after cleanup.
+			} else {
+				entriesAfterCleanup = append(entriesAfterCleanup, entry)
 			}
 		}
 	}
 
-	// add some non-DNATed mock entries which should be cleared up by reconciler
+	// add 6 non-DNATed mock entries which should be cleared up by reconciler
 	// These will exist if the proxy don't have DROP/REJECT rule for service with
 	// no endpoints, --orig-dst and --reply-src will be same for these entries.
 	for _, ip := range []string{testClusterIP, testLoadBalancerIP, testExternalIP} {
-		entry := &netlink.ConntrackFlow{
-			FamilyType: unix.AF_INET,
-			Forward: netlink.IPTuple{
-				DstIP:    netutils.ParseIPSloppy(ip),
-				Protocol: unix.IPPROTO_UDP,
-			},
-			Reverse: netlink.IPTuple{
-				Protocol: unix.IPPROTO_UDP,
-				SrcIP:    netutils.ParseIPSloppy(ip),
-			},
+		for _, port := range []uint16{testServicePort, testNonServicePort} {
+			entry := &netlink.ConntrackFlow{
+				FamilyType: unix.AF_INET,
+				Forward: netlink.IPTuple{
+					DstIP:    netutils.ParseIPSloppy(ip),
+					DstPort:  port,
+					Protocol: unix.IPPROTO_UDP,
+				},
+				Reverse: netlink.IPTuple{
+					Protocol: unix.IPPROTO_UDP,
+					SrcIP:    netutils.ParseIPSloppy(ip),
+				},
+			}
+
+			entriesBeforeCleanup = append(entriesBeforeCleanup, entry)
+			// we do not expect entries with destination port `testServicePort` to be
+			// present after cleanup.
+			if port != testServicePort {
+				entriesAfterCleanup = append(entriesAfterCleanup, entry)
+			}
+
 		}
-		mockEntries = append(mockEntries, entry)
 	}
 
+	t.Logf("entries before cleanup %d after cleanup %d", len(entriesBeforeCleanup), len(entriesAfterCleanup))
 	fake := NewFake()
-	fake.entries = mockEntries
+	fake.entries = entriesBeforeCleanup
 	CleanStaleEntries(fake, testIPFamily, svcPortMap, endpointsMap)
 
 	actualEntries, _ := fake.ListEntries(ipFamilyMap[testIPFamily])
-	require.Equal(t, len(expectedEntries), len(actualEntries))
+	require.Equal(t, len(entriesAfterCleanup), len(actualEntries))
 
 	// sort the actual flows before comparison
 	sort.Slice(actualEntries, func(i, j int) bool {
 		return actualEntries[i].String() < actualEntries[j].String()
 	})
 	// sort the expected flows before comparison
-	sort.Slice(expectedEntries, func(i, j int) bool {
-		return expectedEntries[i].String() < expectedEntries[j].String()
+	sort.Slice(entriesAfterCleanup, func(i, j int) bool {
+		return entriesAfterCleanup[i].String() < entriesAfterCleanup[j].String()
 	})
 
-	for i := 0; i < len(expectedEntries); i++ {
-		require.Equal(t, expectedEntries[i], actualEntries[i])
+	if diff := cmp.Diff(entriesAfterCleanup, actualEntries); len(diff) > 0 {
+		t.Errorf("unexpected entries after cleanup: %s", diff)
+	}
+}
+
+func TestPerformanceCleanStaleEntries(t *testing.T) {
+	sct := proxy.NewServiceChangeTracker(v1.IPv4Protocol, nil, nil)
+	svc := &v1.Service{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      testServiceName,
+			Namespace: testServiceNamespace,
+		},
+		Spec: v1.ServiceSpec{
+			ClusterIP:   testClusterIP,
+			ExternalIPs: []string{testExternalIP},
+			Ports: []v1.ServicePort{
+				{
+					Name:     "test-udp",
+					Port:     testServicePort,
+					Protocol: v1.ProtocolUDP,
+				},
+			},
+		},
+		Status: v1.ServiceStatus{
+			LoadBalancer: v1.LoadBalancerStatus{
+				Ingress: []v1.LoadBalancerIngress{{
+					IP: testLoadBalancerIP,
+				}},
+			},
+		},
+	}
+
+	sct.Update(nil, svc)
+	svcPortMap := make(proxy.ServicePortMap)
+	_ = svcPortMap.Update(sct)
+
+	ect := proxy.NewEndpointsChangeTracker(v1.IPv4Protocol, "test-worker", nil, nil)
+	eps := &discovery.EndpointSlice{
+		TypeMeta:    metav1.TypeMeta{},
+		AddressType: discovery.AddressTypeIPv4,
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      fmt.Sprintf("%s-0", testServiceName),
+			Namespace: testServiceNamespace,
+			Labels:    map[string]string{discovery.LabelServiceName: testServiceName},
+		},
+		Endpoints: []discovery.Endpoint{
+			{
+				Addresses:  []string{testServingEndpointIP},
+				Conditions: discovery.EndpointConditions{Serving: ptr.To(true)},
+			},
+			{
+				Addresses:  []string{testNonServingEndpointIP},
+				Conditions: discovery.EndpointConditions{Serving: ptr.To(false)},
+			},
+		},
+		Ports: []discovery.EndpointPort{
+			{
+				Name:     ptr.To("test-udp"),
+				Port:     ptr.To(int32(testServicePort)),
+				Protocol: ptr.To(v1.ProtocolUDP),
+			},
+		},
+	}
+
+	ect.EndpointSliceUpdate(eps, false)
+	endpointsMap := make(proxy.EndpointsMap)
+	_ = endpointsMap.Update(ect)
+
+	genConntrackEntry := func(dstPort uint16, revSrc string) *netlink.ConntrackFlow {
+		return &netlink.ConntrackFlow{
+			FamilyType: unix.AF_INET,
+			Forward: netlink.IPTuple{
+				DstIP:    netutils.ParseIPSloppy(testExternalIP),
+				DstPort:  dstPort,
+				Protocol: unix.IPPROTO_UDP,
+			},
+			Reverse: netlink.IPTuple{
+				Protocol: unix.IPPROTO_UDP,
+				SrcIP:    netutils.ParseIPSloppy(revSrc),
+			},
+		}
+	}
+
+	fake := NewFake()
+	// 1 valid entry
+	fake.entries = append(fake.entries, genConntrackEntry(uint16(testServicePort), testServingEndpointIP))
+	expectedEntries := 1
+	// 1 stale entry
+	fake.entries = append(fake.entries, genConntrackEntry(uint16(testServicePort), testDeletedEndpointIP))
+	expectedDeleted := 1
+	// 1 M to the Service IP with random ports
+	for i := 0; i < 1000*1000; i++ {
+		port := uint16(rand.Intn(65535))
+		if port == testServicePort {
+			expectedDeleted++
+		} else {
+			expectedEntries++
+		}
+		fake.entries = append(fake.entries, genConntrackEntry(port, testDeletedEndpointIP))
+	}
+
+	CleanStaleEntries(fake, testIPFamily, svcPortMap, endpointsMap)
+	actualEntries, _ := fake.ListEntries(ipFamilyMap[testIPFamily])
+	if len(actualEntries) != expectedEntries {
+		t.Errorf("unexpected number of entries, got %d expected %d", len(actualEntries), expectedEntries)
+	}
+	// expected conntrack entries
+	// 1 for CleanStaleEntries + 1 for ListEntries + 1 for ConntrackDeleteFilters to dump the conntrack table
+	// n for the expected deleted stale entries
+	t.Logf("expected deleted %d", expectedDeleted)
+	if fake.netlinkRequests != 3+expectedDeleted {
+		t.Errorf("expected %d netlink requests, got %d", 3+expectedDeleted, fake.netlinkRequests)
+	}
+}
+
+func TestServiceWithoutEndpoints(t *testing.T) {
+	sct := proxy.NewServiceChangeTracker(v1.IPv4Protocol, nil, nil)
+	svc := &v1.Service{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      testServiceName,
+			Namespace: testServiceNamespace,
+		},
+		Spec: v1.ServiceSpec{
+			ClusterIP:   testClusterIP,
+			ExternalIPs: []string{testExternalIP},
+			Ports: []v1.ServicePort{
+				{
+					Name:     "test-udp",
+					Port:     testServicePort,
+					Protocol: v1.ProtocolUDP,
+				},
+			},
+		},
+		Status: v1.ServiceStatus{
+			LoadBalancer: v1.LoadBalancerStatus{
+				Ingress: []v1.LoadBalancerIngress{{
+					IP: testLoadBalancerIP,
+				}},
+			},
+		},
+	}
+
+	sct.Update(nil, svc)
+	svcPortMap := make(proxy.ServicePortMap)
+	_ = svcPortMap.Update(sct)
+
+	ect := proxy.NewEndpointsChangeTracker(v1.IPv4Protocol, "test-worker", nil, nil)
+	eps := &discovery.EndpointSlice{
+		TypeMeta:    metav1.TypeMeta{},
+		AddressType: discovery.AddressTypeIPv4,
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      fmt.Sprintf("%s-0", testServiceName),
+			Namespace: testServiceNamespace,
+			Labels:    map[string]string{discovery.LabelServiceName: "non-existing-service"},
+		},
+		Endpoints: []discovery.Endpoint{
+			{
+				Addresses:  []string{testServingEndpointIP},
+				Conditions: discovery.EndpointConditions{Serving: ptr.To(true)},
+			},
+			{
+				Addresses:  []string{testNonServingEndpointIP},
+				Conditions: discovery.EndpointConditions{Serving: ptr.To(false)},
+			},
+		},
+		Ports: []discovery.EndpointPort{
+			{
+				Name:     ptr.To("test-udp"),
+				Port:     ptr.To(int32(testServicePort)),
+				Protocol: ptr.To(v1.ProtocolUDP),
+			},
+		},
+	}
+
+	ect.EndpointSliceUpdate(eps, false)
+	endpointsMap := make(proxy.EndpointsMap)
+	_ = endpointsMap.Update(ect)
+
+	genConntrackEntry := func(dstPort uint16, revSrc string) *netlink.ConntrackFlow {
+		return &netlink.ConntrackFlow{
+			FamilyType: unix.AF_INET,
+			Forward: netlink.IPTuple{
+				DstIP:    netutils.ParseIPSloppy(testExternalIP),
+				DstPort:  dstPort,
+				Protocol: unix.IPPROTO_UDP,
+			},
+			Reverse: netlink.IPTuple{
+				Protocol: unix.IPPROTO_UDP,
+				SrcIP:    netutils.ParseIPSloppy(revSrc),
+			},
+		}
+	}
+
+	fake := NewFake()
+	// 1 valid entry
+	fake.entries = append(fake.entries, genConntrackEntry(uint16(testServicePort), testServingEndpointIP))
+	// 1 stale entry
+	fake.entries = append(fake.entries, genConntrackEntry(uint16(testServicePort), testDeletedEndpointIP))
+
+	CleanStaleEntries(fake, testIPFamily, svcPortMap, endpointsMap)
+	actualEntries, _ := fake.ListEntries(ipFamilyMap[testIPFamily])
+	if len(actualEntries) != 2 {
+		t.Errorf("unexpected number of entries, got %d expected %d", len(actualEntries), 2)
 	}
 }
 
@@ -300,6 +531,7 @@ func TestFilterForNAT(t *testing.T) {
 		name           string
 		orig           string
 		dest           string
+		dstPort        uint16
 		protocol       v1.Protocol
 		expectedFilter *conntrackFilter
 	}{
@@ -329,7 +561,7 @@ func TestFilterForNAT(t *testing.T) {
 
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
-			require.Equal(t, tc.expectedFilter, filterForNAT(tc.orig, tc.dest, tc.protocol))
+			require.Equal(t, tc.expectedFilter, filterForIPPortNAT(tc.orig, tc.dest, tc.dstPort, tc.protocol))
 		})
 	}
 }