diff --git a/pkg/operator/ceph/controller/spec.go b/pkg/operator/ceph/controller/spec.go index 07916d56b..38d72c209 100644 --- a/pkg/operator/ceph/controller/spec.go +++ b/pkg/operator/ceph/controller/spec.go @@ -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") } diff --git a/pkg/operator/ceph/controller/spec_test.go b/pkg/operator/ceph/controller/spec_test.go index 127ac0326..d7341c73a 100644 --- a/pkg/operator/ceph/controller/spec_test.go +++ b/pkg/operator/ceph/controller/spec_test.go @@ -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) }) } diff --git a/pkg/operator/ceph/object/spec.go b/pkg/operator/ceph/object/spec.go index 16fefa340..3819c42fc 100644 --- a/pkg/operator/ceph/object/spec.go +++ b/pkg/operator/ceph/object/spec.go @@ -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 { diff --git a/pkg/operator/k8sutil/endpoint.go b/pkg/operator/k8sutil/endpoint.go index 2cef36f18..895f475ea 100644 --- a/pkg/operator/k8sutil/endpoint.go +++ b/pkg/operator/k8sutil/endpoint.go @@ -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 }