From e7320ad63bc4e4e0ae6c3e6c2d89b6f07e3fa289 Mon Sep 17 00:00:00 2001 From: Travis Nielsen Date: Tue, 7 Feb 2023 14:12:46 -0700 Subject: [PATCH] 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 --- pkg/daemon/ceph/client/config.go | 13 +++++------ pkg/daemon/ceph/client/info.go | 1 - pkg/operator/ceph/cluster/cluster.go | 1 - pkg/operator/ceph/cluster/mon/mon.go | 1 - pkg/operator/ceph/cluster/mon/service.go | 6 ++--- pkg/operator/ceph/cluster/mon/spec.go | 4 ++-- pkg/operator/ceph/config/store_test.go | 28 ++++++++++++++++++++---- 7 files changed, 34 insertions(+), 20 deletions(-) diff --git a/pkg/daemon/ceph/client/config.go b/pkg/daemon/ceph/client/config.go index 4f531e51c..bdcd0e705 100644 --- a/pkg/daemon/ceph/client/config.go +++ b/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+"]") diff --git a/pkg/daemon/ceph/client/info.go b/pkg/daemon/ceph/client/info.go index d192e10a7..b6930cf31 100644 --- a/pkg/daemon/ceph/client/info.go +++ b/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 diff --git a/pkg/operator/ceph/cluster/cluster.go b/pkg/operator/ceph/cluster/cluster.go index bf3564ef0..73b195268 100755 --- a/pkg/operator/ceph/cluster/cluster.go +++ b/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. diff --git a/pkg/operator/ceph/cluster/mon/mon.go b/pkg/operator/ceph/cluster/mon/mon.go index 4c6d1860e..12ba3b2b5 100644 --- a/pkg/operator/ceph/cluster/mon/mon.go +++ b/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 { diff --git a/pkg/operator/ceph/cluster/mon/service.go b/pkg/operator/ceph/cluster/mon/service.go index b01d4a314..ed2536842 100644 --- a/pkg/operator/ceph/cluster/mon/service.go +++ b/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) diff --git a/pkg/operator/ceph/cluster/mon/spec.go b/pkg/operator/ceph/cluster/mon/spec.go index 9de80bc7c..b1b07e3e3 100644 --- a/pkg/operator/ceph/cluster/mon/spec.go +++ b/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, }) } diff --git a/pkg/operator/ceph/config/store_test.go b/pkg/operator/ceph/config/store_test.go index 735f55999..902a72e4c 100644 --- a/pkg/operator/ceph/config/store_test.go +++ b/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) {