Switch from deprecated corev1.Endpoints to discoveryv1.EndpointSlice.

api/v1beta1/rabbitmqcluster_status_test.go

@@ -6,6 +6,7 @@ import (
 	"github.com/rabbitmq/cluster-operator/v2/internal/status"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
+	discoveryv1 "k8s.io/api/discovery/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -40,19 +41,17 @@ var _ = Describe("RabbitmqClusterStatus", func() {
 			Conditions:         nil,
 		}
 
-		endPoints := &corev1.Endpoints{
-			Subsets: []corev1.EndpointSubset{
+		endPointSlice := &discoveryv1.EndpointSlice{
+			Endpoints: []discoveryv1.Endpoint{
 				{
-					Addresses: []corev1.EndpointAddress{
-						{
-							IP: "127.0.0.1",
-						},
+					Addresses: []string{
+						"127.0.0.1",
 					},
 				},
 			},
 		}
 
-		rabbitmqClusterStatus.SetConditions([]runtime.Object{sts, endPoints})
+		rabbitmqClusterStatus.SetConditions([]runtime.Object{sts, endPointSlice})
 
 		Expect(rabbitmqClusterStatus.Conditions).To(HaveLen(4))
 		Expect(rabbitmqClusterStatus.Conditions[0].Type).To(Equal(status.AllReplicasReady))

api/v1beta1/rabbitmqcluster_types_test.go

@@ -14,6 +14,7 @@ import (
 	"github.com/rabbitmq/cluster-operator/v2/internal/status"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
+	discoveryv1 "k8s.io/api/discovery/v1"
 	k8sresource "k8s.io/apimachinery/pkg/api/resource"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/utils/ptr"
@@ -484,19 +485,17 @@ var _ = Describe("RabbitmqCluster", func() {
 				Conditions:         nil,
 			}
 
-			endPoints := &corev1.Endpoints{
-				Subsets: []corev1.EndpointSubset{
+			endPointSlice := &discoveryv1.EndpointSlice{
+				Endpoints: []discoveryv1.Endpoint{
 					{
-						Addresses: []corev1.EndpointAddress{
-							{
-								IP: "127.0.0.1",
-							},
+						Addresses: []string{
+							"127.0.0.1",
 						},
 					},
 				},
 			}
 
-			rabbitmqClusterStatus.SetConditions([]runtime.Object{statefulset, endPoints})
+			rabbitmqClusterStatus.SetConditions([]runtime.Object{statefulset, endPointSlice})
 
 			Expect(rabbitmqClusterStatus.Conditions).To(HaveLen(4))
 			Expect(rabbitmqClusterStatus.Conditions[0].Type).To(Equal(status.AllReplicasReady))

config/rbac/role.yaml

@@ -68,6 +68,13 @@ rules:
   - list
   - update
   - watch
+- apiGroups:
+  - discovery.k8s.io
+  resources:
+  - endpointslices
+  verbs:
+  - get
+  - list
 - apiGroups:
   - rabbitmq.com
   resources:

controllers/rabbitmqcluster_controller.go

@@ -30,6 +30,7 @@ import (
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/record"
 
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/types"
 
 	clientretry "k8s.io/client-go/util/retry"
@@ -43,6 +44,7 @@ import (
 	rabbitmqv1beta1 "github.com/rabbitmq/cluster-operator/v2/api/v1beta1"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
+	discoveryv1 "k8s.io/api/discovery/v1"
 	rbacv1 "k8s.io/api/rbac/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
@@ -77,7 +79,6 @@ type RabbitmqClusterReconciler struct {
 // +kubebuilder:rbac:groups="",resources=pods/exec,verbs=create
 // +kubebuilder:rbac:groups="",resources=pods,verbs=update;get;list;watch
 // +kubebuilder:rbac:groups="",resources=services,verbs=get;list;watch;create;update
-// +kubebuilder:rbac:groups="",resources=endpoints,verbs=get;watch;list
 // +kubebuilder:rbac:groups=apps,resources=statefulsets,verbs=get;list;watch;create;update;delete
 // +kubebuilder:rbac:groups="",resources=configmaps,verbs=get;list;watch;create;update
 // +kubebuilder:rbac:groups="",resources=secrets,verbs=get;list;watch;create;update
@@ -88,6 +89,8 @@ type RabbitmqClusterReconciler struct {
 // +kubebuilder:rbac:groups="",resources=serviceaccounts,verbs=get;list;watch;create;update
 // +kubebuilder:rbac:groups="",resources=persistentvolumeclaims,verbs=get;list;watch;create;update
 // +kubebuilder:rbac:groups="rbac.authorization.k8s.io",resources=roles,verbs=get;list;watch;create;update
+// +kubebuilder:rbac:groups="discovery.k8s.io",resources=endpointslices,verbs=get;list
+// +kubebuilder:rbac:groups="",resources=endpoints,verbs=get;watch;list
 // +kubebuilder:rbac:groups="rbac.authorization.k8s.io",resources=rolebindings,verbs=get;list;watch;create;update
 
 func (r *RabbitmqClusterReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
@@ -342,7 +345,8 @@ func (r *RabbitmqClusterReconciler) updateStatusConditions(ctx context.Context,
 
 func (r *RabbitmqClusterReconciler) getChildResources(ctx context.Context, rmq *rabbitmqv1beta1.RabbitmqCluster) ([]runtime.Object, error) {
 	sts := &appsv1.StatefulSet{}
-	endPoints := &corev1.Endpoints{}
+	endpointSliceList := &discoveryv1.EndpointSliceList{}
+	endpointSlice := &discoveryv1.EndpointSlice{}
 
 	if err := r.Get(ctx,
 		types.NamespacedName{Name: rmq.ChildResourceName("server"), Namespace: rmq.Namespace},
@@ -352,15 +356,25 @@ func (r *RabbitmqClusterReconciler) getChildResources(ctx context.Context, rmq *
 		sts = nil
 	}
 
-	if err := r.Get(ctx,
-		types.NamespacedName{Name: rmq.ChildResourceName(resource.ServiceSuffix), Namespace: rmq.Namespace},
-		endPoints); err != nil && !k8serrors.IsNotFound(err) {
+	selector, err := labels.Parse(fmt.Sprintf("%s=%s", discoveryv1.LabelServiceName, rmq.Name))
+	if err != nil {
 		return nil, err
-	} else if k8serrors.IsNotFound(err) {
-		endPoints = nil
 	}
 
-	return []runtime.Object{sts, endPoints}, nil
+	listOptions := client.ListOptions{
+		LabelSelector: selector,
+		Namespace:     rmq.Namespace,
+	}
+
+	if err := r.List(ctx, endpointSliceList, &listOptions); err != nil {
+		return nil, err
+	} else if len(endpointSliceList.Items) == 0 {
+		endpointSlice = nil
+	} else {
+		endpointSlice = &endpointSliceList.Items[0]
+	}
+
+	return []runtime.Object{sts, endpointSlice}, nil
 }
 
 func (r *RabbitmqClusterReconciler) setReconcileSuccess(ctx context.Context, rabbitmqCluster *rabbitmqv1beta1.RabbitmqCluster, condition corev1.ConditionStatus, reason, msg string) {

internal/status/all_replicas_ready_test.go

@@ -17,6 +17,7 @@ import (
 	rabbitmqstatus "github.com/rabbitmq/cluster-operator/v2/internal/status"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
+	discoveryv1 "k8s.io/api/discovery/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 )
@@ -44,7 +45,7 @@ var _ = Describe("AllReplicasReady", func() {
 			})
 
 			It("returns the expected condition", func() {
-				condition := rabbitmqstatus.AllReplicasReadyCondition([]runtime.Object{&corev1.Endpoints{}, sts}, oldCondition)
+				condition := rabbitmqstatus.AllReplicasReadyCondition([]runtime.Object{&discoveryv1.EndpointSlice{}, sts}, oldCondition)
 
 				By("having status true and reason message", func() {
 					Expect(condition.Status).To(Equal(corev1.ConditionTrue))
@@ -112,7 +113,7 @@ var _ = Describe("AllReplicasReady", func() {
 				})
 
 				It("updates the transition time", func() {
-					condition := rabbitmqstatus.AllReplicasReadyCondition([]runtime.Object{&corev1.Endpoints{}, sts}, oldCondition)
+					condition := rabbitmqstatus.AllReplicasReadyCondition([]runtime.Object{&discoveryv1.EndpointSlice{}, sts}, oldCondition)
 					Expect(condition.LastTransitionTime).ToNot(Equal(emptyTime))
 				})
 			})

internal/status/cluster_available.go

@@ -13,13 +13,14 @@ import (
 	"time"
 
 	corev1 "k8s.io/api/core/v1"
+	discoveryv1 "k8s.io/api/discovery/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 )
 
 type ClusterAvailableConditionManager struct {
-	condition RabbitmqClusterCondition
-	endpoints *corev1.Endpoints
+	condition     RabbitmqClusterCondition
+	endpointslice *discoveryv1.EndpointSlice
 }
 
 func ClusterAvailableCondition(resources []runtime.Object,
@@ -32,16 +33,16 @@ func ClusterAvailableCondition(resources []runtime.Object,
 
 	for _, res := range resources {
 		switch resource := res.(type) {
-		case *corev1.Endpoints:
+		case *discoveryv1.EndpointSlice:
 			if resource == nil {
 				condition.Status = corev1.ConditionUnknown
 				condition.Reason = "CouldNotRetrieveEndpoints"
 				condition.Message = "Could not verify available service endpoints"
 				goto assignLastTransitionTime
 			}
 
-			for _, subset := range resource.Subsets {
-				if len(subset.Addresses) > 0 {
+			for _, endpoint := range resource.Endpoints {
+				if len(endpoint.Addresses) > 0 {
 					condition.Status = corev1.ConditionTrue
 					condition.Reason = "AtLeastOneEndpointAvailable"
 					goto assignLastTransitionTime