migrate from v1.Endpoints to discoveryv1.EndpointSlices (kedacore#1298)

* Migration to EndpointSlices

Signed-off-by: Jirka Kremser <jiri.kremser@gmail.com>

* finish migration from v1.Endpoints to discoveryv1.EndpointSlices

Signed-off-by: Jan Wozniak <wozniak.jan@gmail.com>

* handle multiple EndpointSlices per service

Signed-off-by: Jan Wozniak <wozniak.jan@gmail.com>

* remove deletecollection rbac for endpoint slices

Signed-off-by: Jan Wozniak <wozniak.jan@gmail.com>

---------

Signed-off-by: Jirka Kremser <jiri.kremser@gmail.com>
Signed-off-by: Jan Wozniak <wozniak.jan@gmail.com>
Co-authored-by: Jan Wozniak <wozniak.jan@gmail.com>

Author: 2 people

CHANGELOG.md

@@ -21,6 +21,7 @@ This changelog keeps track of work items that have been completed and are ready
 
 ### Breaking Changes
 
+- **General**: `v1.Endpoints` -> `EndpointSlices` ([#1297](https://github.com/kedacore/http-add-on/issues/1297)) Potentially breaking for old versions of k8s.
 - **General**: TODO ([#TODO](https://github.com/kedacore/http-add-on/issues/TODO))
 
 ### New

config/interceptor/role.yaml

@@ -7,15 +7,15 @@ rules:
 - apiGroups:
   - ""
   resources:
-  - endpoints
+  - services
   verbs:
   - get
   - list
   - watch
 - apiGroups:
-  - ""
+  - discovery.k8s.io
   resources:
-  - services
+  - endpointslices
   verbs:
   - get
   - list

config/scaler/role.yaml

@@ -5,9 +5,9 @@ metadata:
   name: scaler
 rules:
 - apiGroups:
-  - ""
+  - discovery.k8s.io
   resources:
-  - endpoints
+  - endpointslices
   verbs:
   - get
   - list

interceptor/forward_wait_func.go

@@ -5,7 +5,7 @@ import (
 	"fmt"
 
 	"github.com/go-logr/logr"
-	v1 "k8s.io/api/core/v1"
+	discov1 "k8s.io/api/discovery/v1"
 
 	"github.com/kedacore/http-add-on/pkg/k8s"
 )
@@ -14,10 +14,10 @@ import (
 // before proceeding to serve the request.
 type forwardWaitFunc func(context.Context, string, string) (bool, error)
 
-func workloadActiveEndpoints(endpoints v1.Endpoints) int {
+func workloadActiveEndpoints(endpoints discov1.EndpointSlice) int {
 	total := 0
-	for _, subset := range endpoints.Subsets {
-		total += len(subset.Addresses)
+	for _, e := range endpoints.Endpoints {
+		total += len(e.Addresses)
 	}
 	return total
 }
@@ -55,7 +55,7 @@ func newWorkloadReplicasForwardWaitFunc(
 		for {
 			select {
 			case event := <-eventCh:
-				endpoints, ok := event.Object.(*v1.Endpoints)
+				endpoints, ok := event.Object.(*discov1.EndpointSlice)
 				if !ok {
 					lggr.Info(
 						"Didn't get a endpoints back in event",

interceptor/forward_wait_func_test.go

@@ -8,7 +8,7 @@ import (
 	"github.com/go-logr/logr"
 	"github.com/stretchr/testify/require"
 	"golang.org/x/sync/errgroup"
-	v1 "k8s.io/api/core/v1"
+	discov1 "k8s.io/api/discovery/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/watch"
 
@@ -27,7 +27,7 @@ func TestForwardWaitFuncOneReplica(t *testing.T) {
 	endpoints := *newEndpoint(ns, endpointsName)
 	cache := k8s.NewFakeEndpointsCache()
 	cache.Set(endpoints)
-	r.NoError(cache.SetSubsets(ns, endpointsName, 1))
+	r.NoError(cache.SetEndpoints(ns, endpointsName, 1))
 
 	ctx, done := context.WithTimeout(ctx, waitFuncWait)
 	defer done()
@@ -97,10 +97,10 @@ func TestWaitFuncWaitsUntilReplicas(t *testing.T) {
 		watcher := cache.GetWatcher(ns, endpointsName)
 		r.NotNil(watcher, "watcher was not found")
 		modifiedEndpoints := endpoints.DeepCopy()
-		modifiedEndpoints.Subsets = []v1.EndpointSubset{
+		modifiedEndpoints.Endpoints = []discov1.Endpoint{
 			{
-				Addresses: []v1.EndpointAddress{
-					{IP: "1.2.3.4"},
+				Addresses: []string{
+					"1.2.3.4",
 				},
 			},
 		}
@@ -119,11 +119,14 @@ func TestWaitFuncWaitsUntilReplicas(t *testing.T) {
 func newEndpoint(
 	namespace,
 	name string,
-) *v1.Endpoints {
-	endpoints := &v1.Endpoints{
+) *discov1.EndpointSlice {
+	endpoints := &discov1.EndpointSlice{
 		ObjectMeta: metav1.ObjectMeta{
-			Name:      name,
-			Namespace: namespace,
+			GenerateName: name,
+			Namespace:    namespace,
+			Labels: map[string]string{
+				discov1.LabelServiceName: name,
+			},
 		},
 	}
 

interceptor/main.go

@@ -46,7 +46,7 @@ var (
 )
 
 // +kubebuilder:rbac:groups=http.keda.sh,resources=httpscaledobjects,verbs=get;list;watch
-// +kubebuilder:rbac:groups="",resources=endpoints,verbs=get;list;watch
+// +kubebuilder:rbac:groups=discovery.k8s.io,resources=endpointslices,verbs=get;list;watch
 // +kubebuilder:rbac:groups="",resources=services,verbs=get;list;watch
 
 func main() {

interceptor/proxy_handlers_integration_test.go

@@ -17,7 +17,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"golang.org/x/sync/errgroup"
-	v1 "k8s.io/api/core/v1"
+	discov1 "k8s.io/api/discovery/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/wait"
 
@@ -55,13 +55,16 @@ func TestIntegrationHappyPath(t *testing.T) {
 	)
 	h.routingTable.Memory[hostForTest(t)] = target
 
-	h.endpCache.Set(v1.Endpoints{
+	h.endpCache.Set(discov1.EndpointSlice{
 		ObjectMeta: metav1.ObjectMeta{
-			Name:      serviceName,
-			Namespace: target.GetNamespace(),
+			GenerateName: serviceName,
+			Namespace:    target.GetNamespace(),
+			Labels: map[string]string{
+				discov1.LabelServiceName: serviceName,
+			},
 		},
 	})
-	r.NoError(h.endpCache.SetSubsets(target.GetNamespace(), serviceName, 1))
+	r.NoError(h.endpCache.SetEndpoints(target.GetNamespace(), serviceName, 1))
 
 	// happy path
 	res, err := doRequest(
@@ -122,10 +125,13 @@ func TestIntegrationNoReplicas(t *testing.T) {
 	h.routingTable.Memory[hostForTest(t)] = target
 
 	// 0 replicas
-	h.endpCache.Set(v1.Endpoints{
+	h.endpCache.Set(discov1.EndpointSlice{
 		ObjectMeta: metav1.ObjectMeta{
-			Name:      serviceName,
-			Namespace: target.GetNamespace(),
+			GenerateName: serviceName,
+			Namespace:    target.GetNamespace(),
+			Labels: map[string]string{
+				discov1.LabelServiceName: serviceName,
+			},
 		},
 	})
 
@@ -173,10 +179,13 @@ func TestIntegrationWaitReplicas(t *testing.T) {
 	// set up a endpoint with zero replicas and create
 	// a watcher we can use later to fake-send a endpoint
 	// event
-	h.endpCache.Set(v1.Endpoints{
+	h.endpCache.Set(discov1.EndpointSlice{
 		ObjectMeta: metav1.ObjectMeta{
-			Name:      serviceName,
-			Namespace: target.GetNamespace(),
+			GenerateName: serviceName,
+			Namespace:    target.GetNamespace(),
+			Labels: map[string]string{
+				discov1.LabelServiceName: serviceName,
+			},
 		},
 	})
 	endpoints, _ := h.endpCache.Get(target.GetNamespace(), serviceName)
@@ -207,10 +216,10 @@ func TestIntegrationWaitReplicas(t *testing.T) {
 		t.Logf("Woke up, setting replicas to 10")
 
 		modifiedEndpoints := endpoints.DeepCopy()
-		modifiedEndpoints.Subsets = []v1.EndpointSubset{
+		modifiedEndpoints.Endpoints = []discov1.Endpoint{
 			{
-				Addresses: []v1.EndpointAddress{
-					{IP: "1.2.3.4"},
+				Addresses: []string{
+					"1.2.3.4",
 				},
 			},
 		}

operator/controllers/http/ping_test.go

@@ -6,6 +6,7 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/require"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 
 	"github.com/kedacore/http-add-on/pkg/k8s"
@@ -30,9 +31,13 @@ func TestPingInterceptors(t *testing.T) {
 	r.NoError(err)
 	defer srv.Close()
 	ctx := context.Background()
-	endpoints, err := k8s.FakeEndpointsForURL(url, ns, svcName, 2)
+	endpoints, err := k8s.FakeEndpointsForURL(url, ns, svcName, 1)
 	r.NoError(err)
-	cl := fake.NewClientBuilder().WithObjects(endpoints).Build()
+	eps := []client.Object{}
+	for i := range endpoints.Items {
+		eps = append(eps, &endpoints.Items[i])
+	}
+	cl := fake.NewClientBuilder().WithObjects(eps...).Build()
 	r.NoError(pingInterceptors(
 		ctx,
 		cl,
@@ -42,5 +47,11 @@ func TestPingInterceptors(t *testing.T) {
 		url.Port(),
 	))
 	reqs := hdl.IncomingRequests()
-	r.Equal(len(endpoints.Subsets[0].Addresses), len(reqs))
+	var endpointsAddrs []string
+	for _, es := range endpoints.Items {
+		for _, e := range es.Endpoints {
+			endpointsAddrs = append(endpointsAddrs, e.Addresses...)
+		}
+	}
+	r.Equal(len(endpointsAddrs), len(reqs))
 }

pkg/k8s/endpoints.go

@@ -3,79 +3,88 @@ package k8s
 import (
 	"context"
 	"fmt"
+	"maps"
 	"net/url"
+	"slices"
 
-	v1 "k8s.io/api/core/v1"
+	discov1 "k8s.io/api/discovery/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
+// Endpoints represents a set of ready and not ready addresses
+type Endpoints struct {
+	ReadyAddresses    []string
+	NotReadyAddresses []string
+}
+
 // GetEndpointsFunc is a type that represents a function that can
 // fetch endpoints
-type GetEndpointsFunc func(
-	ctx context.Context,
-	namespace,
-	serviceName string,
-) (*v1.Endpoints, error)
+type GetEndpointsFunc func(ctx context.Context, namespace, serviceName string) (Endpoints, error)
 
-func EndpointsForService(
-	ctx context.Context,
-	ns,
-	serviceName,
-	servicePort string,
-	endpointsFn GetEndpointsFunc,
-) ([]*url.URL, error) {
+// EndpointsForService fetches the ready endpoints for a given service and returns them as a slice of URLs with the specified port
+func EndpointsForService(ctx context.Context, ns, serviceName, servicePort string, endpointsFn GetEndpointsFunc) ([]url.URL, error) {
 	endpoints, err := endpointsFn(ctx, ns, serviceName)
 	if err != nil {
-		return nil, fmt.Errorf("pkg.k8s.EndpointsForService: %w", err)
+		return []url.URL{}, fmt.Errorf("pkg.k8s.EndpointsForService: %w", err)
 	}
-	ret := []*url.URL{}
-	for _, subset := range endpoints.Subsets {
-		for _, addr := range subset.Addresses {
-			u, err := url.Parse(
-				fmt.Sprintf("http://%s:%s", addr.IP, servicePort),
-			)
-			if err != nil {
-				return nil, err
-			}
-			ret = append(ret, u)
+	ret := make([]url.URL, 0, len(endpoints.ReadyAddresses))
+	for _, addr := range endpoints.ReadyAddresses {
+		u := url.URL{
+			Scheme: "http",
+			Host:   fmt.Sprintf("%s:%s", addr, servicePort),
 		}
+		ret = append(ret, u)
 	}
-
 	return ret, nil
 }
 
-// EndpointsFuncForControllerClient returns a new GetEndpointsFunc
-// that uses the controller-runtime client.Client to fetch endpoints
-func EndpointsFuncForControllerClient(
-	cl client.Client,
-) GetEndpointsFunc {
-	return func(
-		ctx context.Context,
-		namespace,
-		serviceName string,
-	) (*v1.Endpoints, error) {
-		endpts := &v1.Endpoints{}
-		if err := cl.Get(ctx, client.ObjectKey{
-			Namespace: namespace,
-			Name:      serviceName,
-		}, endpts); err != nil {
-			return nil, err
+// EndpointsFuncForControllerClient returns a new GetEndpointsFunc that uses the controller-runtime client.Client to fetch endpoints
+func EndpointsFuncForControllerClient(cl client.Client) GetEndpointsFunc {
+	return func(ctx context.Context, namespace, serviceName string) (Endpoints, error) {
+		ess := &discov1.EndpointSliceList{}
+
+		if err := cl.List(ctx, ess, client.InNamespace(namespace), client.MatchingLabels{discov1.LabelServiceName: serviceName}); err != nil {
+			return Endpoints{}, err
+		}
+		return extractAddresses(ess.Items), nil
+	}
+}
+
+// EndpointsFuncForK8sClientset returns a new GetEndpointsFunc that uses the kubernetes.Clientset to fetch endpoints
+// TODO: this should be eventually removed because it causes high load on the API server, there is EndpointsFuncForControllerClient instead
+func EndpointsFuncForK8sClientset(cl *kubernetes.Clientset) GetEndpointsFunc {
+	return func(ctx context.Context, namespace, serviceName string) (Endpoints, error) {
+		endpointSlCl := cl.DiscoveryV1().EndpointSlices(namespace)
+		ess, err := endpointSlCl.List(ctx, metav1.ListOptions{LabelSelector: fmt.Sprintf("%s=%s", discov1.LabelServiceName, serviceName)})
+		if err != nil {
+			return Endpoints{}, err
 		}
-		return endpts, nil
+		return extractAddresses(ess.Items), nil
 	}
 }
 
-func EndpointsFuncForK8sClientset(
-	cl *kubernetes.Clientset,
-) GetEndpointsFunc {
-	return func(
-		ctx context.Context,
-		namespace,
-		serviceName string,
-	) (*v1.Endpoints, error) {
-		endpointsCl := cl.CoreV1().Endpoints(namespace)
-		return endpointsCl.Get(ctx, serviceName, metav1.GetOptions{})
+// extractAddresses extracts ready and not ready addresses from the given list of EndpointSlice
+func extractAddresses(eps []discov1.EndpointSlice) Endpoints {
+	// addresses from EndpointSlices need to be deduplicated
+	ready, notReady := make(map[string]bool), make(map[string]bool)
+	for _, ep := range eps {
+		for _, addr := range ep.Endpoints {
+			// see also https://github.com/kubernetes/api/blob/v0.33.0/discovery/v1/types.go#L137-L144
+			if addr.Conditions.Ready == nil || *addr.Conditions.Ready {
+				for _, a := range addr.Addresses {
+					ready[a] = true
+				}
+			} else {
+				for _, a := range addr.Addresses {
+					notReady[a] = true
+				}
+			}
+		}
+	}
+	return Endpoints{
+		ReadyAddresses:    slices.Collect(maps.Keys(ready)),
+		NotReadyAddresses: slices.Collect(maps.Keys(notReady)),
 	}
 }