feedc0de/my-rook-config
1
0 Fork 0
forked from rook/rook
Code Pull Requests Activity

core: migrate from v1.Endpoints to discoveryv1.EndpointSlice

Browse Source 
This change updates the codebase to replace the deprecated
v1.Endpoints API with discoveryv1.EndpointSlice.
The v1.Endpoints API has been deprecated in Kubernetes v1.33+

Signed-off-by: subhamkrai <srai@redhat.com>
This commit is contained in:
subhamkrai committed 2025-07-23 21:02:02 +05:30
1 parent 1e489fd6ae
commit 1bde614eb5
4 files changed
+64 -49
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

pkg/operator/ceph/controller/spec.go
+24 -12
View File
@@ -37,11 +37,13 @@ import (
"github.com/rook/rook/pkg/operator/k8sutil"
"github.com/rook/rook/pkg/util/display"
v1 "k8s.io/api/core/v1"
discoveryv1 "k8s.io/api/discovery/v1"
kerrors "k8s.io/apimachinery/pkg/api/errors"
"k8s.io/apimachinery/pkg/api/resource"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/util/intstr"
"k8s.io/apimachinery/pkg/util/sets"
"k8s.io/utils/ptr"
)
const (
@@ -858,27 +860,37 @@ func RgwOpsLogSidecarContainer(opsLogFile, ns string, c cephv1.ClusterSpec, env
}
// CreateExternalMetricsEndpoints creates external metric endpoint
func createExternalMetricsEndpoints(namespace string, monitoringSpec cephv1.MonitoringSpec, ownerInfo *k8sutil.OwnerInfo) (*v1.Endpoints, error) {
func createExternalMetricsEndpoints(namespace string, monitoringSpec cephv1.MonitoringSpec, ownerInfo *k8sutil.OwnerInfo) (*discoveryv1.EndpointSlice, error) {
labels := AppLabels("rook-ceph-mgr", namespace)
endpoints := &v1.Endpoints{
// Convert v1.EndpointAddress to string addresses
addresses := make([]string, len(monitoringSpec.ExternalMgrEndpoints))
for i, endpoint := range monitoringSpec.ExternalMgrEndpoints {
addresses[i] = endpoint.IP
}
endpoints := &discoveryv1.EndpointSlice{
ObjectMeta: metav1.ObjectMeta{
Name: ExternalMgrAppName,
Namespace: namespace,
Labels: labels,
},
Subsets: []v1.EndpointSubset{
AddressType: discoveryv1.AddressTypeIPv4,
Endpoints: []discoveryv1.Endpoint{
{
Addresses: monitoringSpec.ExternalMgrEndpoints,
Ports: []v1.EndpointPort{
{
Name: ServiceExternalMetricName,
Port: int32(monitoringSpec.ExternalMgrPrometheusPort),
Protocol: v1.ProtocolTCP,
},
Addresses: addresses,
Conditions: discoveryv1.EndpointConditions{
Ready: ptr.To(true),
},
},
},
Ports: []discoveryv1.EndpointPort{
{
Name: ptr.To(ServiceExternalMetricName),
Port: ptr.To(int32(monitoringSpec.ExternalMgrPrometheusPort)),
Protocol: ptr.To(v1.ProtocolTCP),
},
},
}
err := ownerInfo.SetControllerReference(endpoints)
@@ -919,7 +931,7 @@ func ConfigureExternalMetricsEndpoint(ctx *clusterd.Context, monitoringSpec ceph
}
// Get the endpoint to see if anything needs to be updated
currentEndpoints, err := ctx.Clientset.CoreV1().Endpoints(clusterInfo.Namespace).Get(clusterInfo.Context, endpoint.Name, metav1.GetOptions{})
currentEndpoints, err := ctx.Clientset.DiscoveryV1().EndpointSlices(clusterInfo.Namespace).Get(clusterInfo.Context, endpoint.Name, metav1.GetOptions{})
if err != nil && !kerrors.IsNotFound(err) {
return errors.Wrap(err, "failed to fetch endpoints")
}
@@ -933,7 +945,7 @@ func ConfigureExternalMetricsEndpoint(ctx *clusterd.Context, monitoringSpec ceph
}
logger.Debugf("diff between current endpoint and newly generated one: %v \n", cmp.Diff(currentEndpoints, endpoint, cmp.Comparer(func(x, y resource.Quantity) bool { return x.Cmp(y) == 0 })))
_, err = k8sutil.CreateOrUpdateEndpoint(clusterInfo.Context, ctx.Clientset, clusterInfo.Namespace, endpoint)
_, err = k8sutil.CreateOrUpdateEndpointSlice(clusterInfo.Context, ctx.Clientset, clusterInfo.Namespace, endpoint)
if err != nil {
return errors.Wrap(err, "failed to create or update mgr endpoint")
}
pkg/operator/ceph/controller/spec_test.go
+10 -10
View File
@@ -287,9 +287,9 @@ func TestConfigureExternalMetricsEndpoint(t *testing.T) {
err := ConfigureExternalMetricsEndpoint(ctx, monitoringSpec, clusterInfo, cephclient.NewMinimumOwnerInfo(t))
assert.NoError(t, err)
currentEndpoints, err := ctx.Clientset.CoreV1().Endpoints(namespace).Get(context.TODO(), "rook-ceph-mgr-external", metav1.GetOptions{})
currentEndpoints, err := ctx.Clientset.DiscoveryV1().EndpointSlices(namespace).Get(context.TODO(), "rook-ceph-mgr-external", metav1.GetOptions{})
assert.NoError(t, err)
assert.Equal(t, "192.168.0.1", currentEndpoints.Subsets[0].Addresses[0].IP, currentEndpoints)
assert.Equal(t, "192.168.0.1", currentEndpoints.Endpoints[0].Addresses[0], currentEndpoints)
})
t.Run("spec and current active mgr endpoint different with no existing endpoint object", func(t *testing.T) {
@@ -315,9 +315,9 @@ func TestConfigureExternalMetricsEndpoint(t *testing.T) {
err := ConfigureExternalMetricsEndpoint(ctx, monitoringSpec, clusterInfo, cephclient.NewMinimumOwnerInfo(t))
assert.NoError(t, err)
currentEndpoints, err := ctx.Clientset.CoreV1().Endpoints(namespace).Get(context.TODO(), "rook-ceph-mgr-external", metav1.GetOptions{})
currentEndpoints, err := ctx.Clientset.DiscoveryV1().EndpointSlices(namespace).Get(context.TODO(), "rook-ceph-mgr-external", metav1.GetOptions{})
assert.NoError(t, err)
assert.Equal(t, "172.17.0.12", currentEndpoints.Subsets[0].Addresses[0].IP, currentEndpoints)
assert.Equal(t, "172.17.0.12", currentEndpoints.Endpoints[0].Addresses[0], currentEndpoints)
})
t.Run("spec and current active mgr endpoint different with existing endpoint object", func(t *testing.T) {
@@ -342,15 +342,15 @@ func TestConfigureExternalMetricsEndpoint(t *testing.T) {
ownerInfo := cephclient.NewMinimumOwnerInfo(t)
ep, err := createExternalMetricsEndpoints(clusterInfo.Namespace, monitoringSpec, ownerInfo)
assert.NoError(t, err)
_, err = ctx.Clientset.CoreV1().Endpoints(namespace).Create(context.TODO(), ep, metav1.CreateOptions{})
_, err = ctx.Clientset.DiscoveryV1().EndpointSlices(namespace).Create(context.TODO(), ep, metav1.CreateOptions{})
assert.NoError(t, err)
err = ConfigureExternalMetricsEndpoint(ctx, monitoringSpec, clusterInfo, ownerInfo)
assert.NoError(t, err)
currentEndpoints, err := ctx.Clientset.CoreV1().Endpoints(namespace).Get(context.TODO(), "rook-ceph-mgr-external", metav1.GetOptions{})
currentEndpoints, err := ctx.Clientset.DiscoveryV1().EndpointSlices(namespace).Get(context.TODO(), "rook-ceph-mgr-external", metav1.GetOptions{})
assert.NoError(t, err)
assert.Equal(t, "172.17.0.12", currentEndpoints.Subsets[0].Addresses[0].IP, currentEndpoints)
assert.Equal(t, "172.17.0.12", currentEndpoints.Endpoints[0].Addresses[0], currentEndpoints)
})
t.Run("spec and current active mgr endpoint identical with existing endpoint object", func(t *testing.T) {
@@ -375,15 +375,15 @@ func TestConfigureExternalMetricsEndpoint(t *testing.T) {
ownerInfo := cephclient.NewMinimumOwnerInfo(t)
ep, err := createExternalMetricsEndpoints(clusterInfo.Namespace, monitoringSpec, ownerInfo)
assert.NoError(t, err)
_, err = ctx.Clientset.CoreV1().Endpoints(namespace).Create(context.TODO(), ep, metav1.CreateOptions{})
_, err = ctx.Clientset.DiscoveryV1().EndpointSlices(namespace).Create(context.TODO(), ep, metav1.CreateOptions{})
assert.NoError(t, err)
err = ConfigureExternalMetricsEndpoint(ctx, monitoringSpec, clusterInfo, ownerInfo)
assert.NoError(t, err)
currentEndpoints, err := ctx.Clientset.CoreV1().Endpoints(namespace).Get(context.TODO(), "rook-ceph-mgr-external", metav1.GetOptions{})
currentEndpoints, err := ctx.Clientset.DiscoveryV1().EndpointSlices(namespace).Get(context.TODO(), "rook-ceph-mgr-external", metav1.GetOptions{})
assert.NoError(t, err)
assert.Equal(t, "192.168.0.1", currentEndpoints.Subsets[0].Addresses[0].IP, currentEndpoints)
assert.Equal(t, "192.168.0.1", currentEndpoints.Endpoints[0].Addresses[0], currentEndpoints)
})
}
pkg/operator/ceph/object/spec.go
+20 -17
View File
@@ -37,9 +37,11 @@ import (
"github.com/rook/rook/pkg/operator/k8sutil"
apps "k8s.io/api/apps/v1"
v1 "k8s.io/api/core/v1"
discoveryv1 "k8s.io/api/discovery/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/util/intstr"
"k8s.io/apimachinery/pkg/util/validation"
"k8s.io/utils/ptr"
)
const (
@@ -725,29 +727,31 @@ func (c *clusterConfig) generateService(cephObjectStore *cephv1.CephObjectStore)
return svc
}
func (c *clusterConfig) generateEndpoint(cephObjectStore *cephv1.CephObjectStore) *v1.Endpoints {
func (c *clusterConfig) generateEndpoint(cephObjectStore *cephv1.CephObjectStore) *discoveryv1.EndpointSlice {
labels := getLabels(cephObjectStore.Name, cephObjectStore.Namespace, true)
k8sEndpointAddrs := []v1.EndpointAddress{}
// Convert to string addresses
k8sEndpointAddrs := []string{}
for _, rookEndpoint := range cephObjectStore.Spec.Gateway.ExternalRgwEndpoints {
k8sEndpointAddr := v1.EndpointAddress{
IP: rookEndpoint.IP,
Hostname: rookEndpoint.Hostname,
}
k8sEndpointAddrs = append(k8sEndpointAddrs, k8sEndpointAddr)
k8sEndpointAddrs = append(k8sEndpointAddrs, rookEndpoint.IP)
}
endpoints := &v1.Endpoints{
endpoints := &discoveryv1.EndpointSlice{
ObjectMeta: metav1.ObjectMeta{
Name: instanceName(cephObjectStore.Name),
Namespace: cephObjectStore.Namespace,
Labels: labels,
},
Subsets: []v1.EndpointSubset{
AddressType: discoveryv1.AddressTypeIPv4,
Endpoints: []discoveryv1.Endpoint{
{
Addresses: k8sEndpointAddrs,
Conditions: discoveryv1.EndpointConditions{
Ready: ptr.To(true),
},
},
},
Ports: []discoveryv1.EndpointPort{},
}
addPortToEndpoint(endpoints, "http", cephObjectStore.Spec.Gateway.Port)
@@ -766,7 +770,7 @@ func (c *clusterConfig) reconcileExternalEndpoint(cephObjectStore *cephv1.CephOb
return errors.Wrapf(err, "failed to set owner reference to ceph object store endpoint %q", endpoint.Name)
}
_, err = k8sutil.CreateOrUpdateEndpoint(c.clusterInfo.Context, c.context.Clientset, cephObjectStore.Namespace, endpoint)
_, err = k8sutil.CreateOrUpdateEndpointSlice(c.clusterInfo.Context, c.context.Clientset, cephObjectStore.Namespace, endpoint)
if err != nil {
return errors.Wrapf(err, "failed to create or update object store %q endpoint", cephObjectStore.Name)
}
@@ -868,16 +872,15 @@ func addPort(service *v1.Service, name string, port, destPort int32) {
})
}
func addPortToEndpoint(endpoints *v1.Endpoints, name string, port int32) {
func addPortToEndpoint(endpoints *discoveryv1.EndpointSlice, name string, port int32) {
if port == 0 {
return
}
endpoints.Subsets[0].Ports = append(endpoints.Subsets[0].Ports, v1.EndpointPort{
Name: name,
Port: port,
Protocol: v1.ProtocolTCP,
},
)
endpoints.Ports = append(endpoints.Ports, discoveryv1.EndpointPort{
Name: ptr.To(name),
Port: ptr.To(port),
Protocol: ptr.To(v1.ProtocolTCP),
})
}
func getLabels(name, namespace string, includeNewLabels bool) map[string]string {
pkg/operator/k8sutil/endpoint.go
+10 -10
View File
@@ -20,26 +20,26 @@ import (
"context"
"fmt"
v1 "k8s.io/api/core/v1"
discoveryv1 "k8s.io/api/discovery/v1"
"k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/client-go/kubernetes"
)
// CreateOrUpdateEndpoint creates a service or updates the service declaratively if it already exists.
func CreateOrUpdateEndpoint(ctx context.Context, clientset kubernetes.Interface, namespace string, endpointDefinition *v1.Endpoints) (*v1.Endpoints, error) {
name := endpointDefinition.Name
logger.Debugf("creating endpoint %q. %v", name, endpointDefinition.Subsets)
ep, err := clientset.CoreV1().Endpoints(namespace).Create(ctx, endpointDefinition, metav1.CreateOptions{})
// CreateOrUpdateEndpointSlice creates an endpoint slice or updates the endpoint slice declaratively if it already exists.
func CreateOrUpdateEndpointSlice(ctx context.Context, clientset kubernetes.Interface, namespace string, endpointSliceDefinition *discoveryv1.EndpointSlice) (*discoveryv1.EndpointSlice, error) {
name := endpointSliceDefinition.Name
logger.Debugf("creating endpoint slice %q. %v", name, endpointSliceDefinition.Endpoints)
es, err := clientset.DiscoveryV1().EndpointSlices(namespace).Create(ctx, endpointSliceDefinition, metav1.CreateOptions{})
if err != nil {
if !errors.IsAlreadyExists(err) {
return nil, fmt.Errorf("failed to create endpoint %q. %v", name, err)
return nil, fmt.Errorf("failed to create endpoint slice %q. %v", name, err)
}
ep, err = clientset.CoreV1().Endpoints(namespace).Update(ctx, endpointDefinition, metav1.UpdateOptions{})
es, err = clientset.DiscoveryV1().EndpointSlices(namespace).Update(ctx, endpointSliceDefinition, metav1.UpdateOptions{})
if err != nil {
return nil, fmt.Errorf("failed to update endpoint %q. %v", name, err)
return nil, fmt.Errorf("failed to update endpoint slice %q. %v", name, err)
}
}
return ep, err
return es, err
}