Repository: rook/rook
Pull request #11859 (2023-03-08 22:52:04)

csi: Update port to 3300 if msgr2 is required
<!-- Please take a look at our [Contributing](https://rook.io/docs/rook/latest/Contributing/development-flow/)
documentation before submitting a Pull Request!
Thank you for contributing to Rook! -->

**Description of your changes:**
When msgr2 is required, ensure the mon endpoints passed to the csi configmap are on port 3300. The mons will still be listening on both 6789 and 3300 if they were created before msgr2 was required. Existing volumes can continue using port 6789 while new volumes will use port 3300.

**Checklist:**

- [ ] **Commit Message Formatting**: Commit titles and messages follow guidelines in the [developer guide](https://rook.io/docs/rook/latest/Contributing/development-flow/#commit-structure)).
- [ ] **Skip Tests for Docs**: If this is only a documentation change, add the label `skip-ci` on the PR.
- [ ] Reviewed the developer guide on [Submitting a Pull Request](https://rook.io/docs/rook/latest/Contributing/development-flow/#submitting-a-pull-request)
- [ ] [Pending release notes](https://github.com/rook/rook/blob/master/PendingReleaseNotes.md) updated with breaking and/or notable changes for the next minor release.
- [ ] Documentation has been updated, if necessary.
- [ ] Unit tests have been added, if necessary.
- [ ] Integration tests have been added, if necessary.



Patch:
diff --git a/pkg/operator/ceph/cluster/cluster_external.go b/pkg/operator/ceph/cluster/cluster_external.go
index 7e747138b..e5b5c2a93 100644
--- a/pkg/operator/ceph/cluster/cluster_external.go
+++ b/pkg/operator/ceph/cluster/cluster_external.go
@@ -113,7 +113,8 @@ func (c *ClusterController) configureExternalCephCluster(cluster *cluster) error
 	}
 
 	// Save CSI configmap
-	err = csi.SaveClusterConfig(c.context.Clientset, c.namespacedName.Namespace, cluster.ClusterInfo, &csi.CsiClusterConfigEntry{Namespace: cluster.ClusterInfo.Namespace, Monitors: csi.MonEndpoints(cluster.ClusterInfo.Monitors)})
+	monEndpoints := csi.MonEndpoints(cluster.ClusterInfo.Monitors, cluster.Spec.RequireMsgr2())
+	err = csi.SaveClusterConfig(c.context.Clientset, c.namespacedName.Namespace, cluster.ClusterInfo, &csi.CsiClusterConfigEntry{Namespace: cluster.ClusterInfo.Namespace, Monitors: monEndpoints})
 	if err != nil {
 		return errors.Wrap(err, "failed to update csi cluster config")
 	}
diff --git a/pkg/operator/ceph/cluster/mon/mon.go b/pkg/operator/ceph/cluster/mon/mon.go
index 826627ef1..ddc8ca31c 100644
--- a/pkg/operator/ceph/cluster/mon/mon.go
+++ b/pkg/operator/ceph/cluster/mon/mon.go
@@ -1078,7 +1078,8 @@ func (c *Cluster) saveMonConfig() error {
 		return errors.Wrap(err, "failed to write connection config for new mons")
 	}
 
-	if err := csi.SaveClusterConfig(c.context.Clientset, c.Namespace, c.ClusterInfo, &csi.CsiClusterConfigEntry{Namespace: c.ClusterInfo.Namespace, Monitors: csi.MonEndpoints(c.ClusterInfo.Monitors)}); err != nil {
+	monEndpoints := csi.MonEndpoints(c.ClusterInfo.Monitors, c.spec.RequireMsgr2())
+	if err := csi.SaveClusterConfig(c.context.Clientset, c.Namespace, c.ClusterInfo, &csi.CsiClusterConfigEntry{Namespace: c.ClusterInfo.Namespace, Monitors: monEndpoints}); err != nil {
 		return errors.Wrap(err, "failed to update csi cluster config")
 	}
 
diff --git a/pkg/operator/ceph/csi/cluster_config.go b/pkg/operator/ceph/csi/cluster_config.go
index d00350c54..043b094e6 100644
--- a/pkg/operator/ceph/csi/cluster_config.go
+++ b/pkg/operator/ceph/csi/cluster_config.go
@@ -19,11 +19,14 @@ package csi
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"os"
+	"strings"
 	"sync"
 
 	"github.com/coreos/pkg/capnslog"
 	"github.com/pkg/errors"
+	"github.com/rook/rook/pkg/daemon/ceph/client"
 	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
 	"github.com/rook/rook/pkg/operator/k8sutil"
 	v1 "k8s.io/api/core/v1"
@@ -99,10 +102,23 @@ func formatCsiClusterConfig(cc csiClusterConfig) (string, error) {
 	return string(ccJson), nil
 }
 
-func MonEndpoints(mons map[string]*cephclient.MonInfo) []string {
+func MonEndpoints(mons map[string]*cephclient.MonInfo, requireMsgr2 bool) []string {
 	endpoints := make([]string, 0)
 	for _, m := range mons {
-		endpoints = append(endpoints, m.Endpoint)
+		endpoint := m.Endpoint
+		if requireMsgr2 {
+			logger.Debugf("evaluating mon %q for msgr1 on endpoint %q", m.Name, m.Endpoint)
+			if strings.HasSuffix(m.Endpoint, fmt.Sprintf(":%d", client.Msgr1port)) {
+				parts := strings.Split(m.Endpoint, ":")
+				if len(parts) != 2 {
+					logger.Errorf("endpoint %q does not contain two parts to extract the port", m.Endpoint)
+					continue
+				}
+				endpoint = fmt.Sprintf("%s:%d", parts[0], client.Msgr2port)
+				logger.Debugf("mon %q will use the msgrv2 port: %q", m.Name, endpoint)
+			}
+		}
+		endpoints = append(endpoints, endpoint)
 	}
 	return endpoints
 }
diff --git a/pkg/operator/ceph/csi/spec.go b/pkg/operator/ceph/csi/spec.go
index 8e6afae2c..cbe26cecd 100644
--- a/pkg/operator/ceph/csi/spec.go
+++ b/pkg/operator/ceph/csi/spec.go
@@ -874,7 +874,7 @@ func (r *ReconcileCSI) configureHolder(driver driverDetails, c ClusterDetail, tp
 	}
 
 	clusterConfigEntry := &CsiClusterConfigEntry{
-		Monitors: MonEndpoints(c.clusterInfo.Monitors),
+		Monitors: MonEndpoints(c.clusterInfo.Monitors, c.cluster.Spec.RequireMsgr2()),
 		RBD:      &CsiRBDSpec{},
 		CephFS:   &CsiCephFSSpec{},
 		NFS:      &CsiNFSSpec{},
diff --git a/pkg/operator/ceph/file/subvolumegroup/controller.go b/pkg/operator/ceph/file/subvolumegroup/controller.go
index a520629de..7a0875dae 100644
--- a/pkg/operator/ceph/file/subvolumegroup/controller.go
+++ b/pkg/operator/ceph/file/subvolumegroup/controller.go
@@ -267,7 +267,7 @@ func (r *ReconcileCephFilesystemSubVolumeGroup) updateClusterConfig(cephFilesyst
 	// config map, so no special care is needed in this controller
 	csiClusterConfigEntry := csi.CsiClusterConfigEntry{
 		Namespace: r.clusterInfo.Namespace,
-		Monitors:  csi.MonEndpoints(r.clusterInfo.Monitors),
+		Monitors:  csi.MonEndpoints(r.clusterInfo.Monitors, cephCluster.Spec.RequireMsgr2()),
 		CephFS: &csi.CsiCephFSSpec{
 			SubvolumeGroup: cephFilesystemSubVolumeGroup.Name,
 		},
diff --git a/pkg/operator/ceph/pool/radosnamespace/controller.go b/pkg/operator/ceph/pool/radosnamespace/controller.go
index 386581045..1ce89ef31 100644
--- a/pkg/operator/ceph/pool/radosnamespace/controller.go
+++ b/pkg/operator/ceph/pool/radosnamespace/controller.go
@@ -259,7 +259,7 @@ func (r *ReconcileCephBlockPoolRadosNamespace) updateClusterConfig(cephBlockPool
 	// config map, so no special care is needed in this controller
 	csiClusterConfigEntry := csi.CsiClusterConfigEntry{
 		Namespace: r.clusterInfo.Namespace,
-		Monitors:  csi.MonEndpoints(r.clusterInfo.Monitors),
+		Monitors:  csi.MonEndpoints(r.clusterInfo.Monitors, cephCluster.Spec.RequireMsgr2()),
 		RBD: &csi.CsiRBDSpec{
 			RadosNamespace: cephBlockPoolRadosNamespace.Name,
 		},


Test patch:
diff --git a/pkg/operator/ceph/csi/cluster_config_test.go b/pkg/operator/ceph/csi/cluster_config_test.go
index 9899767c1..dc64ed60f 100644
--- a/pkg/operator/ceph/csi/cluster_config_test.go
+++ b/pkg/operator/ceph/csi/cluster_config_test.go
@@ -17,8 +17,10 @@ limitations under the License.
 package csi
 
 import (
+	"strings"
 	"testing"
 
+	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -336,3 +338,51 @@ func contains(src, dest []string) bool {
 
 	return false
 }
+
+func TestMonEndpoints(t *testing.T) {
+	monInfo := map[string]*cephclient.MonInfo{
+		"a": {Name: "a", Endpoint: "1.2.3.4:6789"},
+		"b": {Name: "b", Endpoint: "1.2.3.5:6789"},
+		"c": {Name: "c", Endpoint: "1.2.3.6:6789"},
+	}
+
+	t.Run("msgrv1 when not require msgr2", func(t *testing.T) {
+		endpoints := MonEndpoints(monInfo, false)
+		assert.Equal(t, 3, len(endpoints))
+		verifyEndpointPort(t, endpoints, "6789")
+	})
+
+	t.Run("convert to msgr2", func(t *testing.T) {
+		endpoints := MonEndpoints(monInfo, true)
+		assert.Equal(t, 3, len(endpoints))
+		verifyEndpointPort(t, endpoints, "3300")
+	})
+
+	t.Run("remains msgr2", func(t *testing.T) {
+		monInfo := map[string]*cephclient.MonInfo{
+			"a": {Name: "a", Endpoint: "1.2.3.4:3300"},
+			"b": {Name: "b", Endpoint: "1.2.3.5:3300"},
+			"c": {Name: "c", Endpoint: "1.2.3.6:3300"},
+		}
+		endpoints := MonEndpoints(monInfo, false)
+		assert.Equal(t, 3, len(endpoints))
+		verifyEndpointPort(t, endpoints, "3300")
+	})
+
+	t.Run("bad endpoint", func(t *testing.T) {
+		monInfo := map[string]*cephclient.MonInfo{
+			"a": {Name: "a", Endpoint: "1.2.3.4:5:6789"},
+			"b": {Name: "b", Endpoint: "1.2.3.5:6789"},
+			"c": {Name: "c", Endpoint: "1.2.3.6:6789"},
+		}
+		endpoints := MonEndpoints(monInfo, true)
+		assert.Equal(t, 2, len(endpoints))
+		verifyEndpointPort(t, endpoints, "3300")
+	})
+}
+
+func verifyEndpointPort(t *testing.T, endpoints []string, expectedPort string) {
+	for _, endpoint := range endpoints {
+		assert.True(t, strings.HasSuffix(endpoint, expectedPort))
+	}
+}
diff --git a/tests/integration/ceph_upgrade_test.go b/tests/integration/ceph_upgrade_test.go
index 227659ab6..da1dc841a 100644
--- a/tests/integration/ceph_upgrade_test.go
+++ b/tests/integration/ceph_upgrade_test.go
@@ -437,6 +437,9 @@ func (s *UpgradeSuite) upgradeToMaster() {
 	s.installer.Manifests = installer.NewCephManifests(s.settings)
 
 	if s.settings.UseHelm {
+		logger.Info("Requiring msgr2 during helm upgrade to test the port conversion from 6789 to 3300")
+		s.settings.RequireMsgr2 = true
+
 		// Upgrade the operator chart
 		err := s.installer.UpgradeRookOperatorViaHelm()
 		require.NoError(s.T(), err, "failed to upgrade the operator chart")