Repository: rook/rook
Pull request #11879 (2023-03-10 21:52:01)

csi: Update port to 3300 if msgr2 is required (backport #11859)
This is an automatic backport of pull request #11859 done by [Mergify](https://mergify.com).


---


<details>
<summary>Mergify commands and options</summary>

<br />

More conditions and actions can be found in the [documentation](https://docs.mergify.com/).

You can also trigger Mergify actions by commenting on this pull request:

- `@Mergifyio refresh` will re-evaluate the rules
- `@Mergifyio rebase` will rebase this PR on its base branch
- `@Mergifyio update` will merge the base branch into this PR
- `@Mergifyio backport <destination>` will backport this PR on `<destination>` branch

Additionally, on Mergify [dashboard](https://dashboard.mergify.com/) you can:

- look at your merge queues
- generate the Mergify configuration with the config editor.

Finally, you can contact us on https://mergify.com
</details>


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")