mon: allow mons on msgr1 even if v2 settings required

Mons that are configured on msgr1 will be using the v1 port
in the csi driver since only one port is allowed in the configmap.
To allow mons to be started on v2 while keeping quorum with the
v1 mons, the v1 mons must continue running on the v1 port.
So even if msgr2 is required, allow existing v1 mons to continue
running on that port.

Signed-off-by: Travis Nielsen <tnielsen@redhat.com>

pkg/daemon/ceph/client/config.go

@@ -257,16 +257,13 @@ func PopulateMonHostMembers(clusterInfo *ClusterInfo) ([]string, []string) {
 		}
 		monMembers = append(monMembers, monitor.Name)
 		monIP := cephutil.GetIPFromEndpoint(monitor.Endpoint)
-		if clusterInfo.RequireMsgr2 {
+		// Detect the current port if the mon already exists
+		// so the same msgr1 port can be preserved if needed (6789 or 6790)
+		currentMonPort := cephutil.GetPortFromEndpoint(monitor.Endpoint)
+
+		if currentMonPort == Msgr2port {
 			monHosts = append(monHosts, fmt.Sprintf("[v2:%s:%d]", monIP, Msgr2port))
 		} else {
-			// Detect the current port if the mon already exists
-			// so the same msgr1 port can be preserved if needed (6789 or 6790)
-			currentMonPort := cephutil.GetPortFromEndpoint(monitor.Endpoint)
-			// Ensure we're setting a msgr1 port, rather than duplicating msgr2
-			if currentMonPort == Msgr2port {
-				currentMonPort = Msgr1port
-			}
 			msgr2Endpoint := net.JoinHostPort(monIP, strconv.Itoa(int(Msgr2port)))
 			msgr1Endpoint := net.JoinHostPort(monIP, strconv.Itoa(int(currentMonPort)))
 			monHosts = append(monHosts, "[v2:"+msgr2Endpoint+",v1:"+msgr1Endpoint+"]")

pkg/daemon/ceph/client/info.go

@@ -43,7 +43,6 @@ type ClusterInfo struct {
 	CephVersion   cephver.CephVersion
 	Namespace     string
 	OwnerInfo     *k8sutil.OwnerInfo
-	RequireMsgr2  bool
 	// Hide the name of the cluster since in 99% of uses we want to use the cluster namespace.
 	// If the CR name is needed, access it through the NamespacedName() method.
 	name              string

pkg/operator/ceph/cluster/cluster.go

@@ -197,7 +197,6 @@ func (c *ClusterController) initializeCluster(cluster *cluster) error {
 		} else {
 			clusterInfo.OwnerInfo = cluster.ownerInfo
 			clusterInfo.SetName(c.namespacedName.Name)
-			clusterInfo.RequireMsgr2 = cluster.Spec.RequireMsgr2()
 			cluster.ClusterInfo = clusterInfo
 		}
 		// If the local cluster has already been configured, immediately start monitoring the cluster.

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

@@ -498,7 +498,6 @@ func (c *Cluster) initClusterInfo(cephVersion cephver.CephVersion, clusterName s
 	c.ClusterInfo.OwnerInfo = c.ownerInfo
 	c.ClusterInfo.Context = context
 	c.ClusterInfo.SetName(clusterName)
-	c.ClusterInfo.RequireMsgr2 = c.spec.RequireMsgr2()
 
 	// save cluster monitor config
 	if err = c.saveMonConfig(); err != nil {

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

@@ -40,9 +40,9 @@ func (c *Cluster) createService(mon *monConfig) (string, error) {
 		return "", errors.Wrapf(err, "failed to set owner reference to mon service %q", svcDef.Name)
 	}
 
-	// If the cluster doesn't require msgr2 we allow msgr1 as well
-	if !c.ClusterInfo.RequireMsgr2 {
-		addServicePort(svcDef, "tcp-msgr1", DefaultMsgr1Port)
+	// If the mon port was not msgr2, add the msgr1 port
+	if mon.Port != DefaultMsgr2Port {
+		addServicePort(svcDef, "tcp-msgr1", mon.Port)
 	}
 	addServicePort(svcDef, "tcp-msgr2", DefaultMsgr2Port)
 

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

@@ -320,11 +320,11 @@ func (c *Cluster) makeMonDaemonContainer(monConfig *monConfig) corev1.Container
 		WorkingDir:    config.VarLogCephDir,
 	}
 
-	if !c.spec.RequireMsgr2() {
+	if monConfig.Port != DefaultMsgr2Port {
 		// Add messenger 1 port
 		container.Ports = append(container.Ports, corev1.ContainerPort{
 			Name:          "tcp-msgr1",
-			ContainerPort: DefaultMsgr1Port,
+			ContainerPort: monConfig.Port,
 			Protocol:      corev1.ProtocolTCP,
 		})
 	}

pkg/operator/ceph/config/store_test.go

@@ -40,13 +40,17 @@ func TestStore(t *testing.T) {
 	ownerInfo := cephclient.NewMinimumOwnerInfoWithOwnerRef()
 
 	s := GetStore(ctx, ns, ownerInfo)
-
+	mon1EndpointsEnabled := false
 	assertConfigStore := func(ci *cephclient.ClusterInfo) {
 		sec, e := clientset.CoreV1().Secrets(ns).Get(ctxt, StoreName, metav1.GetOptions{})
 		assert.NoError(t, e)
-		mh := strings.Split(sec.StringData["mon_host"], ",")                    // list of mon ip:port pairs in cluster
-		assert.Equal(t, len(ci.Monitors)*2, len(mh), ci.Monitors["a"].Endpoint) // we need to pass x2 since we split on "," above and that returns msgr1 and msgr2 addresses
-		mim := strings.Split(sec.StringData["mon_initial_members"], ",")        // list of mon ids in cluster
+		mh := strings.Split(sec.StringData["mon_host"], ",") // list of mon ip:port pairs in cluster
+		expectedEndpoints := len(ci.Monitors)
+		if mon1EndpointsEnabled {
+			expectedEndpoints *= 2
+		}
+		assert.Equal(t, expectedEndpoints, len(mh), ci.Monitors["a"].Endpoint) // we need to pass x2 since we split on "," above and that returns msgr1 and msgr2 addresses
+		mim := strings.Split(sec.StringData["mon_initial_members"], ",")       // list of mon ids in cluster
 		assert.Equal(t, len(ci.Monitors), len(mim))
 		// make sure every mon has its id/ip:port in mon_initial_members/mon_host
 		for _, id := range mim {
@@ -73,6 +77,22 @@ func TestStore(t *testing.T) {
 	err = s.CreateOrUpdate(i3)
 	assert.NoError(t, err)
 	assertConfigStore(i3)
+
+	// Now run the same test for v1 endpoints
+	mon1EndpointsEnabled = true
+	for _, mon := range i1.Monitors {
+		mon.Endpoint = "1.2.3.4:6789"
+	}
+	err = s.CreateOrUpdate(i1)
+	assert.NoError(t, err)
+	assertConfigStore(i1)
+
+	for _, mon := range i3.Monitors {
+		mon.Endpoint = "1.2.3.4:6789"
+	}
+	err = s.CreateOrUpdate(i3)
+	assert.NoError(t, err)
+	assertConfigStore(i3)
 }
 
 func TestEnvVarsAndFlags(t *testing.T) {