[RayService] Migrate from Endpoints API to EndpointSlice API for RayService (#4245)

* Migrate from Endpoints API to EndpointSlice API for RayService

Signed-off-by: seanlaii <qazwsx0939059006@gmail.com>

* trigger test

* add back endpoints rule for backward compatibility

* add comment

* fix comment

* de-duplicate endpoint based on pod uid

* address comment

* change TODO message

* trigger test

* remove endpoint RBAC

* move comment

* change logging level

---------

Signed-off-by: seanlaii <qazwsx0939059006@gmail.com>

Author: seanlaii

.buildkite/test-sample-yamls.yml

@@ -28,7 +28,7 @@
     - kubectl config set clusters.kind-kind.server https://docker:6443
     # Deploy KubeRay operator
     - pushd ray-operator
-    - IMG=quay.io/kuberay/operator:v1.5.1 make deploy
+    - helm install kuberay-operator kuberay/kuberay-operator --version 1.5.1
     - kubectl wait --timeout=90s --for=condition=Available=true deployment kuberay-operator
     # Run sample YAML tests
     - echo "--- START:Running Sample YAMLs (latest release) tests"

helm-chart/kuberay-operator/templates/_helpers.tpl

@@ -119,14 +119,6 @@ Create a template to ensure consistency for Role and ClusterRole.
 */}}
 {{- define "role.consistentRules" -}}
 rules:
-- apiGroups:
-  - ""
-  resources:
-  - endpoints
-  verbs:
-  - get
-  - list
-  - watch
 - apiGroups:
   - ""
   resources:
@@ -218,6 +210,14 @@ rules:
   - get
   - list
   - update
+- apiGroups:
+  - discovery.k8s.io
+  resources:
+  - endpointslices
+  verbs:
+  - get
+  - list
+  - watch
 - apiGroups:
   - extensions
   - networking.k8s.io

ray-operator/config/rbac/role.yaml

@@ -4,14 +4,6 @@ kind: ClusterRole
 metadata:
   name: kuberay-operator
 rules:
-- apiGroups:
-  - ""
-  resources:
-  - endpoints
-  verbs:
-  - get
-  - list
-  - watch
 - apiGroups:
   - ""
   resources:
@@ -103,6 +95,14 @@ rules:
   - get
   - list
   - update
+- apiGroups:
+  - discovery.k8s.io
+  resources:
+  - endpointslices
+  verbs:
+  - get
+  - list
+  - watch
 - apiGroups:
   - extensions
   - networking.k8s.io

ray-operator/controllers/ray/rayservice_controller.go

@@ -14,10 +14,12 @@ import (
 	"github.com/go-logr/logr"
 	cmap "github.com/orcaman/concurrent-map/v2"
 	corev1 "k8s.io/api/core/v1"
+	discoveryv1 "k8s.io/api/discovery/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/json"
 	"k8s.io/apimachinery/pkg/util/yaml"
 	"k8s.io/client-go/tools/record"
@@ -87,7 +89,6 @@ func NewRayServiceReconciler(_ context.Context, mgr manager.Manager, provider ut
 // +kubebuilder:rbac:groups=core,resources=pods,verbs=get;list;watch;create;update;patch;delete
 // +kubebuilder:rbac:groups=core,resources=pods/status,verbs=get;list;watch;create;update;patch;delete
 // +kubebuilder:rbac:groups=core,resources=pods/proxy,verbs=get;update;patch
-// +kubebuilder:rbac:groups=core,resources=endpoints,verbs=get;list;watch
 // +kubebuilder:rbac:groups=core,resources=services,verbs=get;list;watch;create;update;patch;delete
 // +kubebuilder:rbac:groups=core,resources=services/status,verbs=get;update;patch
 // +kubebuilder:rbac:groups=core,resources=services/proxy,verbs=get;update;patch
@@ -97,6 +98,7 @@ func NewRayServiceReconciler(_ context.Context, mgr manager.Manager, provider ut
 // +kubebuilder:rbac:groups="gateway.networking.k8s.io",resources=httproutes,verbs=get;list;watch;create;update;
 // +kubebuilder:rbac:groups="rbac.authorization.k8s.io",resources=roles,verbs=get;list;watch;create;delete;update
 // +kubebuilder:rbac:groups="rbac.authorization.k8s.io",resources=rolebindings,verbs=get;list;watch;create;delete
+// +kubebuilder:rbac:groups=discovery.k8s.io,resources=endpointslices,verbs=get;list;watch
 
 // [WARNING]: There MUST be a newline after kubebuilder markers.
 // Reconcile is part of the main kubernetes reconciliation loop which aims to
@@ -443,33 +445,26 @@ func (r *RayServiceReconciler) calculateStatus(
 		}
 	}
 
-	serveEndPoints := &corev1.Endpoints{}
-	serveServiceName := common.RayServiceServeServiceNamespacedName(rayServiceInstance)
+	// Calculate the number of ready serve endpoints for the active cluster.
+	serveServiceNamespacedName := common.RayServiceServeServiceNamespacedName(rayServiceInstance)
 	if utils.IsIncrementalUpgradeEnabled(&rayServiceInstance.Spec) && activeCluster != nil {
 		// The Serve service name is based on the unique RayCluster name, since we use the
 		// per-cluster Serve services for traffic routing during an incremental upgrade.
-		serveServiceName.Name = utils.GenerateServeServiceName(activeCluster.Name)
+		serveServiceNamespacedName.Name = utils.GenerateServeServiceName(activeCluster.Name)
 	}
-	if err := r.Get(ctx, serveServiceName, serveEndPoints); err != nil && !errors.IsNotFound(err) {
+	numServeEndpoints, err := r.calculateNumServeEndpointsFromSlices(ctx, serveServiceNamespacedName)
+	if err != nil {
 		return err
 	}
 
-	numServeEndpoints := 0
-	// Ray Pod addresses are categorized into subsets based on the IPs they share.
-	// subset.Addresses contains a list of Ray Pod addresses with ready serve port.
-	for _, subset := range serveEndPoints.Subsets {
-		numServeEndpoints += len(subset.Addresses)
-	}
-
 	// During NewClusterWithIncrementalUpgrade, the pending RayCluster is also serving.
 	if utils.IsIncrementalUpgradeEnabled(&rayServiceInstance.Spec) && pendingCluster != nil {
-		pendingServeServiceName := common.RayClusterServeServiceNamespacedName(pendingCluster)
-		if err := r.Get(ctx, pendingServeServiceName, serveEndPoints); err != nil && !errors.IsNotFound(err) {
+		pendingServeServiceNamespacedName := common.RayClusterServeServiceNamespacedName(pendingCluster)
+		pendingEndpoints, err := r.calculateNumServeEndpointsFromSlices(ctx, pendingServeServiceNamespacedName)
+		if err != nil {
 			return err
 		}
-		for _, subset := range serveEndPoints.Subsets {
-			numServeEndpoints += len(subset.Addresses)
-		}
+		numServeEndpoints += pendingEndpoints
 	}
 
 	if numServeEndpoints > math.MaxInt32 {
@@ -1751,6 +1746,57 @@ func generateHashWithoutReplicasAndWorkersToDelete(rayClusterSpec rayv1.RayClust
 	return utils.GenerateJsonHash(updatedRayClusterSpec)
 }
 
+// calculateNumServeEndpointsFromSlices calculates the number of unique ready Pods
+// from EndpointSlices associated with a given service.
+//
+// In dual-stack environments (IPv4 + IPv6), the same Pod may appear in multiple
+// EndpointSlices with different IP addresses. This function deduplicates by TargetRef.UID
+// to ensure each Pod is counted only once, regardless of how many IP families it serves.
+//
+// An endpoint is considered ready if it has the `Ready` condition set to true.
+// This replaces the legacy Endpoints API approach.
+func (r *RayServiceReconciler) calculateNumServeEndpointsFromSlices(ctx context.Context, serviceNamespacedName client.ObjectKey) (int, error) {
+	logger := ctrl.LoggerFrom(ctx)
+
+	// List all EndpointSlices for the service.
+	// EndpointSlices are automatically created by Kubernetes and labeled with
+	// kubernetes.io/service-name to associate them with their parent Service.
+	endpointSliceList := &discoveryv1.EndpointSliceList{}
+	listOpts := []client.ListOption{
+		client.InNamespace(serviceNamespacedName.Namespace),
+		client.MatchingLabels{discoveryv1.LabelServiceName: serviceNamespacedName.Name},
+	}
+
+	if err := r.List(ctx, endpointSliceList, listOpts...); err != nil {
+		logger.Error(err, "Failed to list EndpointSlices", "serviceName", serviceNamespacedName.Name, "serviceNamespace", serviceNamespacedName.Namespace)
+		return 0, err
+	}
+
+	uniqueNumReadyPods := make(map[types.UID]struct{})
+
+	for _, endpointSlice := range endpointSliceList.Items {
+		for _, endpoint := range endpointSlice.Endpoints {
+			// Only count endpoints that are ready to serve traffic
+			if ptr.Deref(endpoint.Conditions.Ready, false) {
+				if endpoint.TargetRef != nil && endpoint.TargetRef.UID != "" {
+					uniqueNumReadyPods[endpoint.TargetRef.UID] = struct{}{}
+				}
+			}
+		}
+	}
+
+	numPods := len(uniqueNumReadyPods)
+
+	logger.V(1).Info("Counted serve-ready pods via EndpointSlices",
+		"serviceName", serviceNamespacedName.Name,
+		"serviceNamespace", serviceNamespacedName.Namespace,
+		"numSlices", len(endpointSliceList.Items),
+		"numReadyPods", numPods,
+	)
+
+	return numPods, nil
+}
+
 // isHeadPodRunningAndReady checks if the head pod of the RayCluster is running and ready.
 func (r *RayServiceReconciler) isHeadPodRunningAndReady(ctx context.Context, instance *rayv1.RayCluster) (bool, error) {
 	headPod, err := common.GetRayClusterHeadPod(ctx, r, instance)

ray-operator/controllers/ray/rayservice_controller_test.go

@@ -26,6 +26,7 @@ import (
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 	corev1 "k8s.io/api/core/v1"
+	discoveryv1 "k8s.io/api/discovery/v1"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/util/retry"
@@ -119,18 +120,29 @@ func rayServiceTemplate(name string, namespace string, serveAppName string) *ray
 	}
 }
 
-func endpointsTemplate(name string, namespace string) *corev1.Endpoints {
-	return &corev1.Endpoints{
+func endpointSliceTemplate(name string, namespace string) *discoveryv1.EndpointSlice {
+	return &discoveryv1.EndpointSlice{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      name,
 			Namespace: namespace,
+			Labels: map[string]string{
+				discoveryv1.LabelServiceName: name,
+			},
 		},
-		Subsets: []corev1.EndpointSubset{
+		AddressType: discoveryv1.AddressTypeIPv4,
+		Endpoints: []discoveryv1.Endpoint{
 			{
-				Addresses: []corev1.EndpointAddress{
-					{
-						IP: "10.9.8.7",
-					},
+				Addresses: []string{"10.9.8.7"},
+				Conditions: discoveryv1.EndpointConditions{
+					Ready: ptr.To(true),
+				},
+				// TargetRef should always be set in production to identify the backing Pod.
+				// This allows proper deduplication when a Pod appears in multiple EndpointSlices.
+				TargetRef: &corev1.ObjectReference{
+					Kind:      "Pod",
+					Namespace: namespace,
+					Name:      "test-pod-1",
+					UID:       "test-uid-12345",
 				},
 			},
 		},
@@ -275,7 +287,7 @@ var _ = Context("RayService env tests", func() {
 		ctx := context.Background()
 		var rayService *rayv1.RayService
 		var rayCluster *rayv1.RayCluster
-		var endpoints *corev1.Endpoints
+		var endpointSlice *discoveryv1.EndpointSlice
 		serveAppName := "app1"
 		namespace := "default"
 
@@ -363,9 +375,9 @@ var _ = Context("RayService env tests", func() {
 			// TODO: Verify the serve service by checking labels and annotations.
 
 			By("The RayServiceReady condition should be true when the number of endpoints is greater than 0")
-			endpoints = endpointsTemplate(utils.GenerateServeServiceName(rayService.Name), namespace)
-			err = k8sClient.Create(ctx, endpoints)
-			Expect(err).NotTo(HaveOccurred(), "failed to create Endpoints resource")
+			endpointSlice = endpointSliceTemplate(utils.GenerateServeServiceName(rayService.Name), namespace)
+			err = k8sClient.Create(ctx, endpointSlice)
+			Expect(err).NotTo(HaveOccurred(), "failed to create EndpointSlice resource")
 			Eventually(func() int32 {
 				if err := k8sClient.Get(ctx, client.ObjectKey{Name: rayService.Name, Namespace: namespace}, rayService); err != nil {
 					return 0
@@ -379,9 +391,9 @@ var _ = Context("RayService env tests", func() {
 			By(fmt.Sprintf("Delete the RayService custom resource %v", rayService.Name))
 			err := k8sClient.Delete(ctx, rayService)
 			Expect(err).NotTo(HaveOccurred(), "failed to delete the test RayService resource")
-			By(fmt.Sprintf("Delete the Endpoints %v", endpoints.Name))
-			err = k8sClient.Delete(ctx, endpoints)
-			Expect(err).NotTo(HaveOccurred(), "failed to delete the test Endpoints resource")
+			By(fmt.Sprintf("Delete the EndpointSlice %v", endpointSlice.Name))
+			err = k8sClient.Delete(ctx, endpointSlice)
+			Expect(err).NotTo(HaveOccurred(), "failed to delete the test EndpointSlice resource")
 		})
 
 		When("Testing in-place update: updating the serveConfigV2", Ordered, func() {

ray-operator/test/e2erayservice/rayservice_redeploy_test.go

@@ -59,11 +59,10 @@ func TestRedeployRayServe(t *testing.T) {
 	LogWithTimestamp(test.T(), "Checking that the K8s serve service eventually has 1 endpoint and the endpoint is not the old head Pod")
 	g.Eventually(func(g Gomega) {
 		svcName := utils.GenerateServeServiceName(rayService.Name)
-		endpoints, err := test.Client().Core().CoreV1().Endpoints(namespace.Name).Get(test.Ctx(), svcName, metav1.GetOptions{})
+		readyEndpoints, err := GetReadyEndpointsFromSlices(test.Ctx(), test.Client(), namespace.Name, svcName)
 		g.Expect(err).NotTo(HaveOccurred())
-		g.Expect(endpoints.Subsets).To(HaveLen(1))
-		g.Expect(endpoints.Subsets[0].Addresses).To(HaveLen(1))
-		g.Expect(endpoints.Subsets[0].Addresses[0].TargetRef.UID).NotTo(Equal(oldHeadPod.UID))
+		g.Expect(readyEndpoints).To(HaveLen(1))
+		g.Expect(readyEndpoints[0].TargetRefUID).NotTo(Equal(oldHeadPod.UID))
 	}, TestTimeoutMedium).Should(Succeed())
 
 	LogWithTimestamp(test.T(), "Waiting for RayService %s/%s to be ready", rayService.Namespace, rayService.Name)

ray-operator/test/e2erayservice/rayservice_upgrade_test.go

@@ -54,11 +54,11 @@ func TestOldHeadPodFailDuringUpgrade(t *testing.T) {
 
 	svcName := utils.GenerateServeServiceName(rayService.Name)
 	LogWithTimestamp(test.T(), "Checking that the K8s serve service %s has exactly one endpoint because the cluster only has a head Pod", svcName)
-	endpoints, err := test.Client().Core().CoreV1().Endpoints(namespace.Name).Get(test.Ctx(), svcName, metav1.GetOptions{})
+	readyEndpoints, err := GetReadyEndpointsFromSlices(test.Ctx(), test.Client(), namespace.Name, svcName)
 	g.Expect(err).NotTo(HaveOccurred())
-	g.Expect(endpoints.Subsets).To(HaveLen(1))
-	g.Expect(endpoints.Subsets[0].Addresses).To(HaveLen(1))
-	headPodName := endpoints.Subsets[0].Addresses[0].TargetRef.Name
+	g.Expect(readyEndpoints).To(HaveLen(1))
+	headPodName := readyEndpoints[0].TargetRefName
+	headPodUID := readyEndpoints[0].TargetRefUID
 
 	LogWithTimestamp(test.T(), "Upgrading the RayService to trigger a zero downtime upgrade")
 	rayService, err = GetRayService(test, namespace.Name, rayService.Name)
@@ -118,11 +118,11 @@ func TestOldHeadPodFailDuringUpgrade(t *testing.T) {
 
 	LogWithTimestamp(test.T(), "Checking that the K8s serve service eventually has 1 endpoint and the endpoint is not the old head Pod")
 	g.Eventually(func(g Gomega) {
-		endpoints, err = test.Client().Core().CoreV1().Endpoints(namespace.Name).Get(test.Ctx(), svcName, metav1.GetOptions{})
+		readyEndpoints, err := GetReadyEndpointsFromSlices(test.Ctx(), test.Client(), namespace.Name, svcName)
 		g.Expect(err).NotTo(HaveOccurred())
-		g.Expect(endpoints.Subsets).To(HaveLen(1))
-		g.Expect(endpoints.Subsets[0].Addresses).To(HaveLen(1))
-		g.Expect(endpoints.Subsets[0].Addresses[0].TargetRef.Name).NotTo(Equal(headPodName))
+		g.Expect(readyEndpoints).To(HaveLen(1))
+		g.Expect(readyEndpoints[0].TargetRefName).NotTo(Equal(headPodName))
+		g.Expect(readyEndpoints[0].TargetRefUID).NotTo(Equal(headPodUID))
 	}, TestTimeoutMedium).Should(Succeed())
 
 	LogWithTimestamp(test.T(), "Waiting for RayService %s/%s UpgradeInProgress condition to be false", rayService.Namespace, rayService.Name)

ray-operator/test/e2eupgrade/rayservice_operator_upgrade_test.go

@@ -64,11 +64,10 @@ func TestZeroDowntimeUpgradeAfterOperatorUpgrade(t *testing.T) {
 
 	// Validate RayService serve service correctly configured
 	svcName := utils.GenerateServeServiceName(rayService.Name)
-	test.T().Logf("Checking that the K8s serve service %s has exactly one endpoint and two addresses", svcName)
-	endpoints, err := test.Client().Core().CoreV1().Endpoints(namespace.Name).Get(test.Ctx(), svcName, metav1.GetOptions{})
+	test.T().Logf("Checking that the K8s serve service %s has two ready endpoints", svcName)
+	readyEndpoints, err := GetReadyEndpointsFromSlices(test.Ctx(), test.Client(), namespace.Name, svcName)
 	g.Expect(err).NotTo(HaveOccurred())
-	g.Expect(endpoints.Subsets).To(HaveLen(1))
-	g.Expect(endpoints.Subsets[0].Addresses).To(HaveLen(2))
+	g.Expect(readyEndpoints).To(HaveLen(2))
 
 	// Upgrade KubeRay operator to latest version and replace CRDs
 	test.T().Logf("Upgrading the KubeRay operator to %s", upgradeVersion)

ray-operator/test/support/utils.go

@@ -1,11 +1,16 @@
 package support
 
 import (
+	"context"
 	"io/fs"
 	"os"
 	"path"
 
 	"github.com/stretchr/testify/require"
+	discoveryv1 "k8s.io/api/discovery/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/utils/ptr"
 )
 
 func Ptr[T any](v T) *T {
@@ -23,3 +28,37 @@ func WriteToOutputDir(t Test, fileName string, fileType OutputType, data []byte)
 	err := os.WriteFile(path.Join(t.OutputDir(), fileName+"."+string(fileType)), data, fs.ModePerm)
 	require.NoError(t.T(), err)
 }
+
+// EndpointInfo contains information about a ready endpoint from an EndpointSlice.
+type EndpointInfo struct {
+	TargetRefName string
+	TargetRefUID  types.UID
+}
+
+// GetReadyEndpointsFromSlices retrieves all ready endpoints from EndpointSlices
+// associated with the given service name in the specified namespace.
+// It returns a slice of EndpointInfo containing target reference names and UIDs of ready endpoints.
+func GetReadyEndpointsFromSlices(ctx context.Context, client Client, namespace, serviceName string) ([]EndpointInfo, error) {
+	endpointSliceList, err := client.Core().DiscoveryV1().EndpointSlices(namespace).List(ctx, metav1.ListOptions{
+		LabelSelector: discoveryv1.LabelServiceName + "=" + serviceName,
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	var readyEndpoints []EndpointInfo
+	for _, slice := range endpointSliceList.Items {
+		for _, endpoint := range slice.Endpoints {
+			if ptr.Deref(endpoint.Conditions.Ready, false) {
+				if endpoint.TargetRef != nil && endpoint.TargetRef.UID != "" {
+					readyEndpoints = append(readyEndpoints, EndpointInfo{
+						TargetRefName: endpoint.TargetRef.Name,
+						TargetRefUID:  endpoint.TargetRef.UID,
+					})
+				}
+			}
+		}
+	}
+
+	return readyEndpoints, nil
+}