Migration to EndpointSlices

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

Author: jkremser

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/kustomization.yaml

@@ -16,3 +16,31 @@ labels:
   includeTemplates: true
   pairs:
     app.kubernetes.io/instance: interceptor
+images:
+- name: ghcr.io/kedacore/http-add-on-interceptor
+  newName: ghcr.io/kedacore/http-add-on-interceptor
+  newTag: main
+patches:
+- path: e2e-test/otel/deployment.yaml
+  target:
+    group: apps
+    kind: Deployment
+    name: interceptor
+    version: v1
+- path: e2e-test/otel/scaledobject.yaml
+  target:
+    group: keda.sh
+    kind: ScaledObject
+    name: interceptor
+    version: v1alpha1
+- path: e2e-test/tls/deployment.yaml
+  target:
+    group: apps
+    kind: Deployment
+    name: interceptor
+    version: v1
+- path: e2e-test/tls/proxy.service.yaml
+  target:
+    kind: Service
+    name: interceptor-proxy
+    version: v1

config/interceptor/role.yaml

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

config/scaler/kustomization.yaml

@@ -11,3 +11,14 @@ labels:
   includeTemplates: true
   pairs:
     app.kubernetes.io/instance: external-scaler
+images:
+- name: ghcr.io/kedacore/http-add-on-scaler
+  newName: ghcr.io/kedacore/http-add-on-scaler
+  newTag: main
+patches:
+- path: e2e-test/otel/deployment.yaml
+  target:
+    group: apps
+    kind: Deployment
+    name: scaler
+    version: v1

config/scaler/role.yaml

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

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

@@ -45,7 +45,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;deletecollection
 // +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

@@ -42,5 +42,9 @@ func TestPingInterceptors(t *testing.T) {
 		url.Port(),
 	))
 	reqs := hdl.IncomingRequests()
-	r.Equal(len(endpoints.Subsets[0].Addresses), len(reqs))
+	var endpointsAddrs []string
+	for _, e := range endpoints.Endpoints {
+		endpointsAddrs = append(endpointsAddrs, e.Addresses...)
+	}
+	r.Equal(len(endpointsAddrs), len(reqs))
 }