Repository: kubernetes/kubernetes (language: go)

Issue:
kube-proxy: EndpointSliceCache memory is leaked
### What happened?

The kube-proxy which is using iptables mode occupies about 8G memory in a cluster of our production environment.
[root ~]$ kubectl top pod kube-proxy-22dcr -n kube-system
NAME               CPU(cores)   MEMORY(bytes)   
kube-proxy-22dcr   2127m        7935Mi          
[root ~]$
[root ~]$ kubectl top pod kube-proxy-s5nk6 -n kube-system
NAME               CPU(cores)   MEMORY(bytes)   
kube-proxy-s5nk6   521m         7931Mi          
[ root ~]$ 
[root ~]$ kubectl top pod kube-proxy-zhqnr -n kube-system
NAME               CPU(cores)   MEMORY(bytes)   
kube-proxy-zhqnr   389m         7933Mi          
[root ~]$ kubectl top pod kube-proxy-xsvn7 -n kube-system
NAME               CPU(cores)   MEMORY(bytes)   
kube-proxy-xsvn7   464m         7390Mi          
[ root ~]$ 
<img width="840" alt="Clipboard_Screenshot_1732443822" src="https://github.com/user-attachments/assets/8714624f-bd39-49f1-a467-a8e4ef947274">
<img width="1043" alt="Clipboard_Screenshot_1732443860" src="https://github.com/user-attachments/assets/36998270-1af6-4583-b037-c0c55170a5e7">

On the node, this is only about two thousand three hundred iptables rules.
[centos ~]# iptables-save | wc -l
2358
[centos ~]# 

And the number of pod and service in this cluster is less than 1000.
[root ~]$ kubectl get svc -A | wc -l
414
[root ~]$ kubectl get endpoints -A | wc -l
417
[ root ~]$ kubectl get pod -A | wc -l
831
[root ~]$ kubectl get endpointslices -A | wc -l
552
[root ~]$ 



Here are some results of the reproduction：
1， 
<img width="1223" alt="Clipboard_Screenshot_1732323625" src="https://github.com/user-attachments/assets/c47becc3-013d-4d13-8302-ca43d344d0a7">

2,
  HELP kubeproxy_sync_proxy_rules_endpoint_changes_total [ALPHA] Cumulative proxy rules Endpoint changes
  TYPE kubeproxy_sync_proxy_rules_endpoint_changes_total counter
kubeproxy_sync_proxy_rules_endpoint_changes_total 681443
  HELP kubeproxy_sync_proxy_rules_service_changes_total [ALPHA] Cumulative proxy rules Service changes
  TYPE kubeproxy_sync_proxy_rules_service_changes_total counter
kubeproxy_sync_proxy_rules_service_changes_total 1.363076e+06


### What did you expect to happen?

when the number of pod and service is less than 1000, the memory should not occupy so much memory (about 8G).

### How can we reproduce it (as minimally and precisely as possible)?

1, Create service and delete it repeatedly.
2, When creating service, the namespace name or service name should be different from all previously created.
Memory will slowly increase.

### Anything else we need to know?

_No response_

### Kubernetes version

v1.30

### Cloud provider

tke

### OS version
ubuntu2204



Discussion / hints:
This issue is currently awaiting triage.

If a SIG or subproject determines this is a relevant issue, they will accept it by applying the `triage/accepted` label and provide further guidance.

The `triage/accepted` label can be added by org members by writing `/triage accepted` in a comment.


<details>

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>
> There are no sig labels on this issue. Please add an appropriate label by using one of the following commands:
> 
> * `/sig <group-name>`
> * `/wg <group-name>`
> * `/committee <group-name>`
> 
> Please see the [group list](https://git.k8s.io/community/sig-list.md) for a listing of the SIGs, working groups, and committees available.

/sig network
/priority important-soon

Fix patch:
diff --git a/pkg/proxy/endpointslicecache.go b/pkg/proxy/endpointslicecache.go
index c967f9f517222..5671404fcd3db 100644
--- a/pkg/proxy/endpointslicecache.go
+++ b/pkg/proxy/endpointslicecache.go
@@ -149,6 +149,9 @@ func (cache *EndpointSliceCache) checkoutChanges() map[types.NamespacedName]*end
 			}
 
 			delete(esTracker.pending, name)
+			if len(esTracker.applied) == 0 && len(esTracker.pending) == 0 {
+				delete(cache.trackerByServiceMap, serviceNN)
+			}
 		}
 
 		change.current = cache.getEndpointsMap(serviceNN, esTracker.applied)


Test patch:
diff --git a/pkg/proxy/endpointslicecache_test.go b/pkg/proxy/endpointslicecache_test.go
index c51203c13579e..f1afd63a53cd2 100644
--- a/pkg/proxy/endpointslicecache_test.go
+++ b/pkg/proxy/endpointslicecache_test.go
@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"reflect"
 	"testing"
+	"time"
 
 	v1 "k8s.io/api/core/v1"
 	discovery "k8s.io/api/discovery/v1"
@@ -559,3 +560,107 @@ func (cmc *cacheMutationCheck) Check(t *testing.T) {
 		}
 	}
 }
+
+func TestEndpointSliceCacheClearedCorrectly(t *testing.T) {
+	initEndpointsPorts := func(eps *discovery.EndpointSlice) {
+		eps.Endpoints = []discovery.Endpoint{{
+			Addresses: []string{"1.1.1.1"},
+			NodeName:  ptr.To(testHostname),
+		}}
+		eps.Ports = []discovery.EndpointPort{{
+			Name:     ptr.To("n1"),
+			Port:     ptr.To[int32](11),
+			Protocol: ptr.To(v1.ProtocolUDP),
+		}}
+	}
+
+	testCases := []struct {
+		name               string
+		currEndpointSlices []*discovery.EndpointSlice
+	}{
+		{
+			name: "one endpoint slice",
+			currEndpointSlices: []*discovery.EndpointSlice{
+				makeTestEndpointSlice("ns1", "ep1", 1, initEndpointsPorts),
+			},
+		},
+		{
+			name: "two endpoint slices, same namespace",
+			currEndpointSlices: []*discovery.EndpointSlice{
+				makeTestEndpointSlice("ns1", "ep1", 1, initEndpointsPorts),
+				makeTestEndpointSlice("ns1", "ep2", 2, initEndpointsPorts),
+			},
+		},
+		{
+			name: "two endpoint slices, same service",
+			currEndpointSlices: []*discovery.EndpointSlice{
+				makeTestEndpointSlice("ns1", "ep1", 1, initEndpointsPorts),
+				makeTestEndpointSlice("ns1", "ep1", 2, initEndpointsPorts),
+			},
+		},
+		{
+			name: "two endpoint slices, different namespace",
+			currEndpointSlices: []*discovery.EndpointSlice{
+				makeTestEndpointSlice("ns1", "ep1", 1, initEndpointsPorts),
+				makeTestEndpointSlice("ns2", "ep1", 2, initEndpointsPorts),
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			fp := newFakeProxier(v1.IPv4Protocol, time.Time{})
+			fp.hostname = testHostname
+
+			for _, epSlice := range tc.currEndpointSlices {
+				fp.addEndpointSlice(epSlice)
+			}
+			fp.endpointsMap.Update(fp.endpointsChanges)
+
+			for _, epSlice := range tc.currEndpointSlices {
+				fp.deleteEndpointSlice(epSlice)
+			}
+			fp.endpointsMap.Update(fp.endpointsChanges)
+
+			if len(fp.endpointsChanges.endpointSliceCache.trackerByServiceMap) != 0 {
+				t.Errorf("expected: endpointSliceCache does not have any entries, got: %v", fp.endpointsChanges.endpointSliceCache.trackerByServiceMap)
+			}
+		})
+	}
+}
+
+func TestSameServiceEndpointSliceCacheClearedCorrectly(t *testing.T) {
+	initEndpointsPorts := func(eps *discovery.EndpointSlice) {
+		eps.Endpoints = []discovery.Endpoint{{
+			Addresses: []string{"1.1.1.1"},
+			NodeName:  ptr.To(testHostname),
+		}}
+		eps.Ports = []discovery.EndpointPort{{
+			Name:     ptr.To("n1"),
+			Port:     ptr.To[int32](11),
+			Protocol: ptr.To(v1.ProtocolUDP),
+		}}
+	}
+
+	currEndpointSlices := []*discovery.EndpointSlice{
+		makeTestEndpointSlice("ns1", "svc1", 1, initEndpointsPorts),
+		makeTestEndpointSlice("ns1", "svc1", 2, initEndpointsPorts),
+	}
+
+	fp := newFakeProxier(v1.IPv4Protocol, time.Time{})
+	fp.hostname = testHostname
+
+	for _, epSlice := range currEndpointSlices {
+		fp.addEndpointSlice(epSlice)
+	}
+	fp.endpointsMap.Update(fp.endpointsChanges)
+
+	// only delete the first endpoint slice
+	fp.deleteEndpointSlice(currEndpointSlices[0])
+
+	fp.endpointsMap.Update(fp.endpointsChanges)
+
+	if len(fp.endpointsChanges.endpointSliceCache.trackerByServiceMap) != 1 {
+		t.Errorf("expected: endpointSliceCache to have one entries, got: %v", fp.endpointsChanges.endpointSliceCache.trackerByServiceMap)
+	}
+}
diff --git a/pkg/proxy/servicechangetracker_test.go b/pkg/proxy/servicechangetracker_test.go
index e139d6648247f..8f58424754b46 100644
--- a/pkg/proxy/servicechangetracker_test.go
+++ b/pkg/proxy/servicechangetracker_test.go
@@ -900,3 +900,23 @@ func TestBuildServiceMapServiceUpdate(t *testing.T) {
 		t.Errorf("expected healthcheck ports length 0, got %v", healthCheckNodePorts)
 	}
 }
+
+func TestServiceCacheLeaks(t *testing.T) {
+	fp := newFakeProxier(v1.IPv4Protocol, time.Time{})
+
+	service := makeTestService("ns1", "svc1", func(svc *v1.Service) {
+		svc.Spec.Type = v1.ServiceTypeClusterIP
+		svc.Spec.ClusterIP = "172.16.55.4"
+		svc.Spec.Ports = addTestPort(svc.Spec.Ports, "p1", "UDP", 1234, 4321, 0)
+		svc.Spec.Ports = addTestPort(svc.Spec.Ports, "p2", "TCP", 1235, 5321, 0)
+	})
+	fp.addService(service)
+	if len(fp.serviceChanges.items) != 1 {
+		t.Errorf("Found %d items on the cache, 1 expected", len(fp.serviceChanges.items))
+	}
+
+	fp.deleteService(service)
+	if len(fp.serviceChanges.items) > 0 {
+		t.Errorf("Found %d items on the cache, 0 expected", len(fp.serviceChanges.items))
+	}
+}