Merge pull request #92838 from tnqn/endpointslicetrack-leak

Fix memory leak in endpointSliceTracker

pkg/controller/endpointslice/BUILD

@@ -66,6 +66,7 @@ go_test(
         "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/runtime:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/runtime/schema:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/types:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/util/intstr:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/util/rand:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/util/wait:go_default_library",

pkg/controller/endpointslice/endpointslice_controller.go

@@ -307,6 +307,7 @@ func (c *Controller) syncService(key string) error {
 		if apierrors.IsNotFound(err) {
 			c.triggerTimeTracker.DeleteService(namespace, name)
 			c.reconciler.deleteService(namespace, name)
+			c.endpointSliceTracker.DeleteService(namespace, name)
 			// The service has been deleted, return nil so that it won't be retried.
 			return nil
 		}

pkg/controller/endpointslice/endpointslice_tracker.go

@@ -51,8 +51,11 @@ func (est *endpointSliceTracker) Has(endpointSlice *discovery.EndpointSlice) boo
 	est.lock.Lock()
 	defer est.lock.Unlock()
 
-	rrv := est.relatedResourceVersions(endpointSlice)
+	rrv, ok := est.relatedResourceVersions(endpointSlice)
-	_, ok := rrv[endpointSlice.Name]
+	if !ok {
+		return false
+	}
+	_, ok = rrv[endpointSlice.Name]
 	return ok
 }
 
@@ -63,7 +66,10 @@ func (est *endpointSliceTracker) Stale(endpointSlice *discovery.EndpointSlice) b
 	est.lock.Lock()
 	defer est.lock.Unlock()
 
-	rrv := est.relatedResourceVersions(endpointSlice)
+	rrv, ok := est.relatedResourceVersions(endpointSlice)
+	if !ok {
+		return true
+	}
 	return rrv[endpointSlice.Name] != endpointSlice.ResourceVersion
 }
 
@@ -73,33 +79,42 @@ func (est *endpointSliceTracker) Update(endpointSlice *discovery.EndpointSlice)
 	est.lock.Lock()
 	defer est.lock.Unlock()
 
-	rrv := est.relatedResourceVersions(endpointSlice)
+	rrv, ok := est.relatedResourceVersions(endpointSlice)
+	if !ok {
+		rrv = endpointSliceResourceVersions{}
+		est.resourceVersionsByService[getServiceNN(endpointSlice)] = rrv
+	}
 	rrv[endpointSlice.Name] = endpointSlice.ResourceVersion
 }
 
+// DeleteService removes the set of resource versions tracked for the Service.
+func (est *endpointSliceTracker) DeleteService(namespace, name string) {
+	est.lock.Lock()
+	defer est.lock.Unlock()
+
+	serviceNN := types.NamespacedName{Name: name, Namespace: namespace}
+	delete(est.resourceVersionsByService, serviceNN)
+}
+
 // Delete removes the resource version in this endpointSliceTracker for the
 // provided EndpointSlice.
 func (est *endpointSliceTracker) Delete(endpointSlice *discovery.EndpointSlice) {
 	est.lock.Lock()
 	defer est.lock.Unlock()
 
-	rrv := est.relatedResourceVersions(endpointSlice)
+	rrv, ok := est.relatedResourceVersions(endpointSlice)
+	if ok {
 		delete(rrv, endpointSlice.Name)
 	}
+}
 
 // relatedResourceVersions returns the set of resource versions tracked for the
-// Service corresponding to the provided EndpointSlice. If no resource versions
+// Service corresponding to the provided EndpointSlice, and a bool to indicate
-// are currently tracked for this service, an empty set is initialized.
+// if it exists.
-func (est *endpointSliceTracker) relatedResourceVersions(endpointSlice *discovery.EndpointSlice) endpointSliceResourceVersions {
+func (est *endpointSliceTracker) relatedResourceVersions(endpointSlice *discovery.EndpointSlice) (endpointSliceResourceVersions, bool) {
 	serviceNN := getServiceNN(endpointSlice)
 	vers, ok := est.resourceVersionsByService[serviceNN]
-
+	return vers, ok
-	if !ok {
-		vers = endpointSliceResourceVersions{}
-		est.resourceVersionsByService[serviceNN] = vers
-	}
-
-	return vers
 }
 
 // getServiceNN returns a namespaced name for the Service corresponding to the

pkg/controller/endpointslice/endpointslice_tracker_test.go

@@ -19,8 +19,11 @@ package endpointslice
 import (
 	"testing"
 
+	"github.com/stretchr/testify/assert"
+
 	discovery "k8s.io/api/discovery/v1beta1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
 )
 
 func TestEndpointSliceTrackerUpdate(t *testing.T) {
@@ -47,30 +50,51 @@ func TestEndpointSliceTrackerUpdate(t *testing.T) {
 		checksParam                     *discovery.EndpointSlice
 		expectHas                       bool
 		expectStale                     bool
+		expectResourceVersionsByService map[types.NamespacedName]endpointSliceResourceVersions
 	}{
 		"same slice": {
 			updateParam: epSlice1,
 			checksParam: epSlice1,
 			expectHas:   true,
 			expectStale: false,
+			expectResourceVersionsByService: map[types.NamespacedName]endpointSliceResourceVersions{
+				{Namespace: epSlice1.Namespace, Name: "svc1"}: {
+					epSlice1.Name: epSlice1.ResourceVersion,
+				},
+			},
 		},
 		"different namespace": {
 			updateParam: epSlice1,
 			checksParam: epSlice1DifferentNS,
 			expectHas:   false,
 			expectStale: true,
+			expectResourceVersionsByService: map[types.NamespacedName]endpointSliceResourceVersions{
+				{Namespace: epSlice1.Namespace, Name: "svc1"}: {
+					epSlice1.Name: epSlice1.ResourceVersion,
+				},
+			},
 		},
 		"different service": {
 			updateParam: epSlice1,
 			checksParam: epSlice1DifferentService,
 			expectHas:   false,
 			expectStale: true,
+			expectResourceVersionsByService: map[types.NamespacedName]endpointSliceResourceVersions{
+				{Namespace: epSlice1.Namespace, Name: "svc1"}: {
+					epSlice1.Name: epSlice1.ResourceVersion,
+				},
+			},
 		},
 		"different resource version": {
 			updateParam: epSlice1,
 			checksParam: epSlice1DifferentRV,
 			expectHas:   true,
 			expectStale: true,
+			expectResourceVersionsByService: map[types.NamespacedName]endpointSliceResourceVersions{
+				{Namespace: epSlice1.Namespace, Name: "svc1"}: {
+					epSlice1.Name: epSlice1.ResourceVersion,
+				},
+			},
 		},
 	}
 
@@ -84,6 +108,7 @@ func TestEndpointSliceTrackerUpdate(t *testing.T) {
 			if esTracker.Stale(tc.checksParam) != tc.expectStale {
 				t.Errorf("tc.tracker.Stale(%+v) == %t, expected %t", tc.checksParam, esTracker.Stale(tc.checksParam), tc.expectStale)
 			}
+			assert.Equal(t, tc.expectResourceVersionsByService, esTracker.resourceVersionsByService)
 		})
 	}
 }
@@ -172,3 +197,69 @@ func TestEndpointSliceTrackerDelete(t *testing.T) {
 		})
 	}
 }
+
+func TestEndpointSliceTrackerDeleteService(t *testing.T) {
+	svcName1, svcNS1 := "svc1", "ns1"
+	svcName2, svcNS2 := "svc2", "ns2"
+	epSlice1 := &discovery.EndpointSlice{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:            "example-1",
+			Namespace:       svcNS1,
+			ResourceVersion: "rv1",
+			Labels:          map[string]string{discovery.LabelServiceName: svcName1},
+		},
+	}
+
+	testCases := map[string]struct {
+		updateParam                     *discovery.EndpointSlice
+		deleteServiceParam              *types.NamespacedName
+		expectHas                       bool
+		expectStale                     bool
+		expectResourceVersionsByService map[types.NamespacedName]endpointSliceResourceVersions
+	}{
+		"same service": {
+			updateParam:                     epSlice1,
+			deleteServiceParam:              &types.NamespacedName{Namespace: svcNS1, Name: svcName1},
+			expectHas:                       false,
+			expectStale:                     true,
+			expectResourceVersionsByService: map[types.NamespacedName]endpointSliceResourceVersions{},
+		},
+		"different namespace": {
+			updateParam:        epSlice1,
+			deleteServiceParam: &types.NamespacedName{Namespace: svcNS2, Name: svcName1},
+			expectHas:          true,
+			expectStale:        false,
+			expectResourceVersionsByService: map[types.NamespacedName]endpointSliceResourceVersions{
+				{Namespace: epSlice1.Namespace, Name: "svc1"}: {
+					epSlice1.Name: epSlice1.ResourceVersion,
+				},
+			},
+		},
+		"different service": {
+			updateParam:        epSlice1,
+			deleteServiceParam: &types.NamespacedName{Namespace: svcNS1, Name: svcName2},
+			expectHas:          true,
+			expectStale:        false,
+			expectResourceVersionsByService: map[types.NamespacedName]endpointSliceResourceVersions{
+				{Namespace: epSlice1.Namespace, Name: "svc1"}: {
+					epSlice1.Name: epSlice1.ResourceVersion,
+				},
+			},
+		},
+	}
+
+	for name, tc := range testCases {
+		t.Run(name, func(t *testing.T) {
+			esTracker := newEndpointSliceTracker()
+			esTracker.Update(tc.updateParam)
+			esTracker.DeleteService(tc.deleteServiceParam.Namespace, tc.deleteServiceParam.Name)
+			if esTracker.Has(tc.updateParam) != tc.expectHas {
+				t.Errorf("tc.tracker.Has(%+v) == %t, expected %t", tc.updateParam, esTracker.Has(tc.updateParam), tc.expectHas)
+			}
+			if esTracker.Stale(tc.updateParam) != tc.expectStale {
+				t.Errorf("tc.tracker.Stale(%+v) == %t, expected %t", tc.updateParam, esTracker.Stale(tc.updateParam), tc.expectStale)
+			}
+			assert.Equal(t, tc.expectResourceVersionsByService, esTracker.resourceVersionsByService)
+		})
+	}
+}

pkg/controller/endpointslice/reconciler_test.go

@@ -983,7 +983,7 @@ func expectActions(t *testing.T, actions []k8stesting.Action, num int, verb, res
 }
 
 func expectTrackedResourceVersion(t *testing.T, tracker *endpointSliceTracker, slice *discovery.EndpointSlice, expectedRV string) {
-	rrv := tracker.relatedResourceVersions(slice)
+	rrv, _ := tracker.relatedResourceVersions(slice)
 	rv, tracked := rrv[slice.Name]
 	if !tracked {
 		t.Fatalf("Expected EndpointSlice %s to be tracked", slice.Name)