Merge pull request #15522 from patrostkowski/feature/mon-active-ep-14986

mon: manage endpoints containing set of mon ip addresses

Documentation/Storage-Configuration/Advanced/ceph-mon-health.md

@@ -125,6 +125,32 @@ CephCluster CR:
 - `spec.network.Provider` : When updated from being empty to "host", Rook fails over all monitors, configuring them to enable or disable host networking.
 - `spec.network.multiClusterService`: When enabled or disabled, Rook fails over all monitors, configuring them to start (or stop) using service IPs compatible with the multi-cluster service.
 
+## Tracking Mon Endpoints
+
+An EndpointSlice resource provides dynamic DNS resolution, allowing clients to resolve mon endpoints via DNS without requiring manual updates. Dynamic DNS resolution helps address challenges such as virtual machine live migration by ensuring seamless and automatic updates to mon endpoint addresses. The Ceph client can connect to `rook-ceph-active-mons.<namespace>.svc.cluster.local` to dynamically resolve mon endpoints and receive automatic updates when mon IPs change.
+
+To enable the dynamic DNS resolution, create a headless service, that endpointslice resource references. Below is the required configuration for the headless service:
+
+```bash
+kubectl apply -f deploy/examples/headless-mon-service.yaml
+```
+
+To confirm the dynamic resolution of active mon endpoints, we can perform a check from within the cluster:
+
+```console
+$ nslookup rook-ceph-active-mons.rook-ceph.svc.cluster.local
+Server:    127.0.0.53
+Address:   127.0.0.53#53
+
+Non-authoritative answer:
+Name:   rook-ceph-active-mons.rook-ceph.svc.cluster.local
+Address: 10.233.49.126
+Name:   rook-ceph-active-mons.rook-ceph.svc.cluster.local
+Address: 10.233.37.99
+Name:   rook-ceph-active-mons.rook-ceph.svc.cluster.local
+Address: 10.233.1.212
+```
+
 ## External Monitors
 
 !!! attention
@@ -149,7 +175,7 @@ Here is a step-by-step guide on how to add external monitors to a Rook cluster:
         allowMultiplePerNode: false
         # ID of external Mon
         externalMonIDs:
-        - ext-mon-1 
+        - ext-mon-1
     ```
 
     This will tell Rook to create two internal monitors in the cluster and to keep the external monitor with the ID `ext-mon-1` if it is found in the quorum.

PendingReleaseNotes.md

@@ -15,3 +15,4 @@ Object:
 ## Features
 
 - Support external mons for local Rook cluster (see [#14733](https://github.com/rook/rook/issues/14733)).
+- Manage EndpointSlice resources containing monitor IPs to support DNS-based resolution for Ceph clients (see [#14986](https://github.com/rook/rook/issues/14986)).

deploy/charts/rook-ceph/templates/clusterrole.yaml

@@ -99,6 +99,7 @@ rules:
   - watch
 - apiGroups:
   - ""
+  - "discovery.k8s.io"
   resources:
   # Rook creates events for its custom resources
   - events
@@ -108,6 +109,7 @@ rules:
   # Rook creates endpoints for mgr and object store access
   - endpoints
   - services
+  - endpointslices
   verbs:
   - get
   - list

deploy/examples/common.yaml

@@ -294,6 +294,7 @@ rules:
       - watch
   - apiGroups:
       - ""
+      - "discovery.k8s.io"
     resources:
       # Rook creates events for its custom resources
       - events
@@ -303,6 +304,7 @@ rules:
       # Rook creates endpoints for mgr and object store access
       - endpoints
       - services
+      - endpointslices
     verbs:
       - get
       - list

deploy/examples/headless-mon-service.yaml

@@ -0,0 +1,9 @@
+# Create a headless service which will effectively referenced by the EndpointSlice resources
+# of the same name which is created automatically by Rook
+apiVersion: v1
+kind: Service
+metadata:
+  name: rook-ceph-active-mons
+  namespace: rook-ceph # namespace:cluster
+spec:
+  clusterIP: None # headless service

pkg/operator/ceph/cluster/mon/health_test.go

@@ -231,7 +231,7 @@ func TestTrackMonsOutOfQuorum(t *testing.T) {
 	assert.NoError(t, err)
 
 	// initialize the configmap
-	err = c.persistExpectedMonDaemons()
+	err = c.persistExpectedMonDaemonsInConfigMap()
 	assert.NoError(t, err)
 
 	// Track mon.a as out of quorum

pkg/operator/ceph/cluster/mon/mon.go

@@ -23,6 +23,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"net"
 	"reflect"
 	"strconv"
 	"strings"
@@ -43,17 +44,24 @@ import (
 	cephver "github.com/rook/rook/pkg/operator/ceph/version"
 	"github.com/rook/rook/pkg/operator/k8sutil"
 	apps "k8s.io/api/apps/v1"
+	corev1 "k8s.io/api/core/v1"
 	v1 "k8s.io/api/core/v1"
+	discoveryv1 "k8s.io/api/discovery/v1"
 	kerrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/utils/ptr"
 
 	cephcsi "github.com/ceph/ceph-csi/api/deploy/kubernetes"
 )
 
 const (
+	// endpointSliceNameIPv4 is the name of the endpointslice with IPv4 mon addresses
+	endpointSliceNameIPv4 = "rook-ceph-active-mons-ipv4"
+	// endpointSliceNameIPv6 is the name of the endpointslice with IPv6 mon addresses
+	endpointSliceNameIPv6 = "rook-ceph-active-mons-ipv6"
 	// EndpointConfigMapName is the name of the configmap with mon endpoints
 	EndpointConfigMapName = "rook-ceph-mon-endpoints"
 	// EndpointDataKey is the name of the key inside the mon configmap to get the endpoints
@@ -78,6 +86,11 @@ const (
 	// Nautilus. In Nautilus and a few Ceph releases after, Ceph can use both v1 and v2 protocols.
 	DefaultMsgr2Port int32 = 3300
 
+	// DefaultMsgr1PortName is the name used for the Ceph msgr1 TCP port
+	DefaultMsgr1PortName string = "tcp-msgr1"
+	// DefaultMsgr2PortName is the name used for the Ceph msgr2 TCP port
+	DefaultMsgr2PortName string = "tcp-msgr2"
+
 	// minimum amount of memory in MB to run the pod
 	cephMonPodMinimumMemory uint64 = 1024
 
@@ -1146,7 +1159,7 @@ func (c *Cluster) waitForMonsToJoin(mons []*monConfig, requireAllInQuorum bool)
 
 func (c *Cluster) saveMonConfig() error {
 	if err := c.persistExpectedMonDaemons(); err != nil {
-		return errors.Wrap(err, "failed to persist expected mons")
+		return errors.Wrap(err, "failed to persist expected mon daemons")
 	}
 
 	// Every time the mon config is updated, must also update the global config so that all daemons
@@ -1188,6 +1201,130 @@ func (c *Cluster) saveMonConfig() error {
 }
 
 func (c *Cluster) persistExpectedMonDaemons() error {
+	if err := c.persistExpectedMonDaemonsInConfigMap(); err != nil {
+		return errors.Wrap(err, "failed to persist expected mons in ConfigMap")
+	}
+
+	if err := c.persistExpectedMonDaemonsAsEndpointSlice(); err != nil {
+		return errors.Wrap(err, "failed to persist expected mons in EndpointSlice")
+	}
+	return nil
+}
+
+func (c *Cluster) persistExpectedMonDaemonsAsEndpointSlice() error {
+	monitors := c.ClusterInfo.AllMonitors()
+	if len(monitors) == 0 {
+		// Theoretically, we could now go ahead with the normal code path to
+		// delete (both IPv4 and IPv6) EndpointSlices, but 0 mons is certainly
+		// an error state, so it's better to do nothing destructive right now.
+		logger.Debug("no mon addresses found, skipping endpointslice resource reconciliation")
+		return nil
+	}
+
+	ipv4Addresses := []string{}
+	ipv6Addresses := []string{}
+
+	for _, mon := range monitors {
+		host, _, err := net.SplitHostPort(mon.Endpoint)
+		if err != nil {
+			return errors.Wrapf(err, "failed to parse mon addr %q", mon.Endpoint)
+		}
+		ip := net.ParseIP(host)
+		if ip == nil {
+			logger.Warningf("invalid IP parsed from mon endpoint: %s", mon.Endpoint)
+			continue
+		}
+		if ip.To4() != nil {
+			ipv4Addresses = append(ipv4Addresses, host)
+		} else {
+			ipv6Addresses = append(ipv6Addresses, host)
+		}
+	}
+
+	if err := c.createEndpointSliceForAddresses(ipv4Addresses, discoveryv1.AddressTypeIPv4); err != nil {
+		return err
+	}
+
+	if err := c.createEndpointSliceForAddresses(ipv6Addresses, discoveryv1.AddressTypeIPv6); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (c *Cluster) createEndpointSliceForAddresses(addresses []string, addressType discoveryv1.AddressType) error {
+	client := c.context.Clientset.DiscoveryV1().EndpointSlices(c.Namespace)
+
+	sliceName := endpointSliceNameIPv4
+	if addressType == discoveryv1.AddressTypeIPv6 {
+		sliceName = endpointSliceNameIPv6
+	}
+
+	if len(addresses) == 0 {
+		logger.Debugf("no %s addresses found, deleting existing %q endpointslice if exists", addressType, sliceName)
+		if err := client.Delete(c.ClusterInfo.Context, sliceName, metav1.DeleteOptions{}); err != nil {
+			if kerrors.IsNotFound(err) {
+				logger.Debugf("endpointslice %q not found, nothing to delete", sliceName)
+			} else {
+				logger.Errorf("failed to delete endpointslice %q: %v", sliceName, err)
+			}
+		}
+		return nil
+	}
+
+	endpointSlicePorts := []discoveryv1.EndpointPort{}
+	endpointSlicePorts = append(endpointSlicePorts, discoveryv1.EndpointPort{
+		Name:     ptr.To(DefaultMsgr2PortName),
+		Port:     ptr.To(DefaultMsgr2Port),
+		Protocol: ptr.To(corev1.ProtocolTCP),
+	})
+	if !c.spec.RequireMsgr2() {
+		endpointSlicePorts = append(endpointSlicePorts, discoveryv1.EndpointPort{
+			Name:     ptr.To(DefaultMsgr1PortName),
+			Port:     ptr.To(DefaultMsgr1Port),
+			Protocol: ptr.To(corev1.ProtocolTCP),
+		})
+	}
+
+	endpointSlice := &discoveryv1.EndpointSlice{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      sliceName,
+			Namespace: c.Namespace,
+			Labels: map[string]string{
+				"kubernetes.io/service-name": "rook-ceph-active-mons",
+			},
+		},
+		Endpoints: []discoveryv1.Endpoint{
+			{
+				Addresses: addresses,
+			},
+		},
+		Ports:       endpointSlicePorts,
+		AddressType: addressType,
+	}
+
+	cephv1.GetClusterMetadataAnnotations(c.spec.Annotations).ApplyToObjectMeta(&endpointSlice.ObjectMeta)
+
+	if err := c.ownerInfo.SetControllerReference(endpointSlice); err != nil {
+		return errors.Wrapf(err, "failed to set controller reference on endpointslice %q", sliceName)
+	}
+
+	if _, err := client.Create(c.ClusterInfo.Context, endpointSlice, metav1.CreateOptions{}); err != nil {
+		if !kerrors.IsAlreadyExists(err) {
+			return errors.Wrapf(err, "failed to create %s endpointslice", addressType)
+		}
+
+		logger.Debugf("updating existing %s endpointslice %s", addressType, sliceName)
+		if _, err = client.Update(c.ClusterInfo.Context, endpointSlice, metav1.UpdateOptions{}); err != nil {
+			return errors.Wrapf(err, "failed to update %s endpointslice", addressType)
+		}
+	}
+
+	logger.Infof("created/updated %s endpointslice with addresses: %+v", addressType, addresses)
+	return nil
+}
+
+func (c *Cluster) persistExpectedMonDaemonsInConfigMap() error {
 	configMap := &v1.ConfigMap{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:       EndpointConfigMapName,
@@ -1492,7 +1629,8 @@ func (c *Cluster) startMon(m *monConfig, schedule *controller.MonScheduleInfo) e
 		return errors.Wrapf(err, "failed to commit maxMonId after starting mon %q", m.DaemonName)
 	}
 
-	// Persist the expected list of mons to the configmap in case the operator is interrupted before the mon failover is completed
+	// Persist the expected list of mons to the ConfigMap and EndpointSlice resources
+	// in case the operator is interrupted before the mon failover is completed.
 	// The config on disk won't be updated until the mon failover is completed
 	if err := c.persistExpectedMonDaemons(); err != nil {
 		return errors.Wrap(err, "failed to persist expected mon daemons")

pkg/operator/ceph/cluster/mon/mon_test.go

@@ -43,8 +43,11 @@ import (
 	"github.com/stretchr/testify/assert"
 	apps "k8s.io/api/apps/v1"
 	v1 "k8s.io/api/core/v1"
+	discoveryv1 "k8s.io/api/discovery/v1"
+	kerrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/utils/ptr"
 )
 
 // generate a standard mon config from a mon id w/ default port and IP 2.4.6.{1,2,3,...}
@@ -278,8 +281,15 @@ func TestPersistMons(t *testing.T) {
 	c := New(context.TODO(), &clusterd.Context{Clientset: clientset}, "ns", cephv1.ClusterSpec{Annotations: cephv1.AnnotationsSpec{cephv1.KeyClusterMetadata: cephv1.Annotations{"key": "value"}}}, ownerInfo)
 	setCommonMonProperties(c, 1, cephv1.MonSpec{Count: 3, AllowMultiplePerNode: true}, "myversion")
 
+	expectedPorts := []discoveryv1.EndpointPort{
+		{Name: ptr.To(DefaultMsgr2PortName), Protocol: ptr.To(v1.ProtocolTCP), Port: ptr.To(DefaultMsgr2Port)},
+		{Name: ptr.To(DefaultMsgr1PortName), Protocol: ptr.To(v1.ProtocolTCP), Port: ptr.To(DefaultMsgr1Port)},
+	}
+
 	// Persist mon a
-	err := c.persistExpectedMonDaemons()
+	err := c.persistExpectedMonDaemonsInConfigMap()
+	assert.NoError(t, err)
+	err = c.persistExpectedMonDaemonsAsEndpointSlice()
 	assert.NoError(t, err)
 
 	cm, err := c.context.Clientset.CoreV1().ConfigMaps(c.Namespace).Get(context.TODO(), EndpointConfigMapName, metav1.GetOptions{})
@@ -287,15 +297,142 @@ func TestPersistMons(t *testing.T) {
 	assert.Equal(t, "a=1.2.3.1:3300", cm.Data[EndpointDataKey])
 	assert.Equal(t, map[string]string{"key": "value"}, cm.Annotations)
 
+	ep, err := c.context.Clientset.DiscoveryV1().EndpointSlices(c.Namespace).Get(context.TODO(), endpointSliceNameIPv4, metav1.GetOptions{})
+	assert.NoError(t, err)
+	assert.Equal(t, "1.2.3.1", ep.Endpoints[0].Addresses[0])
+	assert.ElementsMatch(t, expectedPorts, ep.Ports)
+	assert.Equal(t, map[string]string{"key": "value"}, cm.Annotations)
+
 	// Persist mon b, and remove mon a for simply testing the configmap is updated
 	c.ClusterInfo.InternalMonitors["b"] = &cephclient.MonInfo{Name: "b", Endpoint: "4.5.6.7:3300"}
 	delete(c.ClusterInfo.InternalMonitors, "a")
-	err = c.persistExpectedMonDaemons()
+	err = c.persistExpectedMonDaemonsInConfigMap()
+	assert.NoError(t, err)
+	err = c.persistExpectedMonDaemonsAsEndpointSlice()
 	assert.NoError(t, err)
 
 	cm, err = c.context.Clientset.CoreV1().ConfigMaps(c.Namespace).Get(context.TODO(), EndpointConfigMapName, metav1.GetOptions{})
 	assert.NoError(t, err)
 	assert.Equal(t, "b=4.5.6.7:3300", cm.Data[EndpointDataKey])
+
+	ep, err = c.context.Clientset.DiscoveryV1().EndpointSlices(c.Namespace).Get(context.TODO(), endpointSliceNameIPv4, metav1.GetOptions{})
+	assert.NoError(t, err)
+	assert.Equal(t, "4.5.6.7", ep.Endpoints[0].Addresses[0])
+	assert.ElementsMatch(t, expectedPorts, ep.Ports)
+	assert.Equal(t, map[string]string{"key": "value"}, cm.Annotations)
+}
+
+func TestCreateEndpointSlices(t *testing.T) {
+	clientset := test.New(t, 1)
+	ownerInfo := cephclient.NewMinimumOwnerInfoWithOwnerRef()
+
+	// RequireMsgr2=false
+	c := New(context.TODO(), &clusterd.Context{Clientset: clientset}, "ns", cephv1.ClusterSpec{}, ownerInfo)
+	expectedPorts := []discoveryv1.EndpointPort{
+		{Name: ptr.To(DefaultMsgr2PortName), Protocol: ptr.To(v1.ProtocolTCP), Port: ptr.To(DefaultMsgr2Port)},
+		{Name: ptr.To(DefaultMsgr1PortName), Protocol: ptr.To(v1.ProtocolTCP), Port: ptr.To(DefaultMsgr1Port)},
+	}
+	testCreateEndpointSlicesForCluster(t, c, expectedPorts)
+
+	// RequireMsgr2=true
+	c = New(
+		context.TODO(),
+		&clusterd.Context{Clientset: clientset},
+		"ns",
+		cephv1.ClusterSpec{
+			Network: cephv1.NetworkSpec{
+				Connections: &cephv1.ConnectionsSpec{
+					RequireMsgr2: true,
+				},
+			},
+		}, ownerInfo)
+	expectedPorts = []discoveryv1.EndpointPort{
+		{Name: ptr.To(DefaultMsgr2PortName), Protocol: ptr.To(v1.ProtocolTCP), Port: ptr.To(DefaultMsgr2Port)},
+	}
+	testCreateEndpointSlicesForCluster(t, c, expectedPorts)
+}
+
+func testCreateEndpointSlicesForCluster(t *testing.T, c *Cluster, expectedPorts []discoveryv1.EndpointPort) {
+	ipv4Addresses := []string{"10.0.0.1", "10.0.0.2", "10.0.0.3"}
+	ipv4Mons := []cephclient.MonInfo{
+		{Name: "a", Endpoint: fmt.Sprintf("%s:6789", ipv4Addresses[0])},
+		{Name: "b", Endpoint: fmt.Sprintf("%s:6789", ipv4Addresses[1])},
+		{Name: "c", Endpoint: fmt.Sprintf("%s:6789", ipv4Addresses[2])},
+	}
+	ipv6Addresses := []string{"2001:db8::1", "2001:db8::2", "2001:db8::3"}
+	ipv6Mons := []cephclient.MonInfo{
+		{Name: "d", Endpoint: fmt.Sprintf("[%s]:6789", ipv6Addresses[0])},
+		{Name: "e", Endpoint: fmt.Sprintf("[%s]:6789", ipv6Addresses[1])},
+		{Name: "f", Endpoint: fmt.Sprintf("[%s]:6789", ipv6Addresses[2])},
+	}
+
+	// IPv4 test
+	c.ClusterInfo.InternalMonitors = map[string]*cephclient.MonInfo{}
+	for _, mon := range ipv4Mons {
+		c.ClusterInfo.InternalMonitors[mon.Name] = &mon
+	}
+
+	err := c.persistExpectedMonDaemonsAsEndpointSlice()
+	assert.NoError(t, err)
+
+	epSliceIPv4, err := c.context.Clientset.DiscoveryV1().EndpointSlices(c.Namespace).Get(context.TODO(), endpointSliceNameIPv4, metav1.GetOptions{})
+	assert.NoError(t, err)
+	assert.Equal(t, discoveryv1.AddressTypeIPv4, epSliceIPv4.AddressType)
+	assert.Len(t, epSliceIPv4.Endpoints, 1)
+	assert.ElementsMatch(t, ipv4Addresses, epSliceIPv4.Endpoints[0].Addresses)
+	assert.ElementsMatch(t, expectedPorts, epSliceIPv4.Ports)
+
+	_, err = c.context.Clientset.DiscoveryV1().EndpointSlices(c.Namespace).Get(context.TODO(), endpointSliceNameIPv6, metav1.GetOptions{})
+	assert.True(t, kerrors.IsNotFound(err))
+
+	// IPv6 test
+	c.ClusterInfo.InternalMonitors = map[string]*cephclient.MonInfo{}
+	for _, mon := range ipv6Mons {
+		c.ClusterInfo.InternalMonitors[mon.Name] = &mon
+	}
+
+	err = c.persistExpectedMonDaemonsAsEndpointSlice()
+	assert.NoError(t, err)
+
+	_, err = c.context.Clientset.DiscoveryV1().EndpointSlices(c.Namespace).Get(context.TODO(), endpointSliceNameIPv4, metav1.GetOptions{})
+	assert.True(t, kerrors.IsNotFound(err))
+
+	epSliceIPv6, err := c.context.Clientset.DiscoveryV1().EndpointSlices(c.Namespace).Get(context.TODO(), endpointSliceNameIPv6, metav1.GetOptions{})
+	assert.NoError(t, err)
+	assert.Equal(t, discoveryv1.AddressTypeIPv6, epSliceIPv6.AddressType)
+	assert.Len(t, epSliceIPv6.Endpoints, 1)
+	assert.ElementsMatch(t, ipv6Addresses, epSliceIPv6.Endpoints[0].Addresses)
+	assert.ElementsMatch(t, expectedPorts, epSliceIPv6.Ports)
+
+	// Mixed IPv4 and IPv6 test.
+	// Note that this normally doesn't happen, because rook uses only IPv4 or
+	// only IPv6 mons. But in the future, migration between IPv4 and IPv6 might
+	// be supported, and then, during migration, the old mons would use the
+	// old IP family and the new mons would use the new IP family.
+	c.ClusterInfo.InternalMonitors = map[string]*cephclient.MonInfo{}
+	for _, mon := range ipv4Mons {
+		c.ClusterInfo.InternalMonitors[mon.Name] = &mon
+	}
+	for _, mon := range ipv6Mons {
+		c.ClusterInfo.InternalMonitors[mon.Name] = &mon
+	}
+
+	err = c.persistExpectedMonDaemonsAsEndpointSlice()
+	assert.NoError(t, err)
+
+	epSliceIPv4, err = c.context.Clientset.DiscoveryV1().EndpointSlices(c.Namespace).Get(context.TODO(), endpointSliceNameIPv4, metav1.GetOptions{})
+	assert.NoError(t, err)
+	assert.Equal(t, discoveryv1.AddressTypeIPv4, epSliceIPv4.AddressType)
+	assert.Len(t, epSliceIPv4.Endpoints, 1)
+	assert.ElementsMatch(t, ipv4Addresses, epSliceIPv4.Endpoints[0].Addresses)
+	assert.ElementsMatch(t, expectedPorts, epSliceIPv4.Ports)
+
+	epSliceIPv6, err = c.context.Clientset.DiscoveryV1().EndpointSlices(c.Namespace).Get(context.TODO(), endpointSliceNameIPv6, metav1.GetOptions{})
+	assert.NoError(t, err)
+	assert.Equal(t, discoveryv1.AddressTypeIPv6, epSliceIPv6.AddressType)
+	assert.Len(t, epSliceIPv6.Endpoints, 1)
+	assert.ElementsMatch(t, ipv6Addresses, epSliceIPv6.Endpoints[0].Addresses)
+	assert.ElementsMatch(t, expectedPorts, epSliceIPv6.Ports)
 }
 
 func TestSaveMonEndpoints(t *testing.T) {

pkg/operator/ceph/cluster/mon/service.go

@@ -44,9 +44,9 @@ func (c *Cluster) createService(mon *monConfig) (*v1.Service, error) {
 
 	// If the mon port was not msgr2, add the msgr1 port
 	if mon.Port != DefaultMsgr2Port {
-		addServicePort(svcDef, "tcp-msgr1", mon.Port)
+		addServicePort(svcDef, DefaultMsgr1PortName, mon.Port)
 	}
-	addServicePort(svcDef, "tcp-msgr2", DefaultMsgr2Port)
+	addServicePort(svcDef, DefaultMsgr2PortName, DefaultMsgr2Port)
 
 	// Set the ClusterIP if the service does not exist and we expect a certain cluster IP
 	// For example, in disaster recovery the service might have been deleted accidentally, but we have the

pkg/operator/ceph/cluster/mon/spec.go

@@ -313,7 +313,7 @@ func (c *Cluster) makeMonDaemonContainer(monConfig *monConfig) corev1.Container
 		SecurityContext: controller.PodSecurityContext(),
 		Ports: []corev1.ContainerPort{
 			{
-				Name:          "tcp-msgr2",
+				Name:          DefaultMsgr2PortName,
 				ContainerPort: DefaultMsgr2Port,
 				Protocol:      corev1.ProtocolTCP,
 			},
@@ -350,7 +350,7 @@ func (c *Cluster) makeMonDaemonContainer(monConfig *monConfig) corev1.Container
 	} else {
 		// Add messenger 1 port
 		container.Ports = append(container.Ports, corev1.ContainerPort{
-			Name:          "tcp-msgr1",
+			Name:          DefaultMsgr1PortName,
 			ContainerPort: monConfig.Port,
 			Protocol:      corev1.ProtocolTCP,
 		})