Ceph messengers 2 support

This commit introduces the necessary changes to support the new
messenger feature coming with Ceph Nautilus (currently in development).

What changes? Now the monitor listens on two port:

* old 6789 for messenger v1, which will help us support older client
(e,g: krbd)
* new 3300 for messengers v2, which brings new improvement in the
messaging layer. This new transport layer brings numerous advantages
such as encryption improvement, speed improvement, pluggable nature to
support different network stack than TCP and many more.

We still have one Service IP, however it has 2 ports, see:

```
NAME                      TYPE        CLUSTER-IP       EXTERNAL-IP   PORT(S)             AGE
rook-ceph-mon-a           ClusterIP   10.106.217.160   <none>        3300/TCP,6789/TCP   4h
rook-ceph-mon-b           ClusterIP   10.99.36.175     <none>        3300/TCP,6789/TCP   4h
rook-ceph-mon-c           ClusterIP   10.108.220.74    <none>        3300/TCP,6789/TCP   4h
````

The `ceph.conf` has changed and we don't force the port when using an IP
address (public addr etc). Ceph, depending on its version will naturally
start the monitors on their right port, 6789.

A new --ceph-version-name CLI argument has been added to the Rook binary
so that when the pod starts it passes the ceph version name and the
configuration of the ceph.conf, as well as daemon startup flags, happen
properly.

Given that the Rook Operator remembers the port of all the monitors it
deployed (through Pod definition), this change is not an issue and will
maintain backward compatibility.

Note that to test this you must build rook with dev container image,
which contains the dev Nautilus version. So you should do something
like:

`make -j4 BASEIMAGE='ceph/daemon-base:latest-master' IMAGES='ceph' build`

Resolves: #2525
Signed-off-by: Sébastien Han <seb@redhat.com>

cmd/rook/ceph/ceph.go

@@ -90,6 +90,7 @@ func addCephFlags(command *cobra.Command) {
 	command.Flags().StringVar(&cfg.monEndpoints, "mon-endpoints", "", "ceph mon endpoints")
 	command.Flags().StringVar(&cfg.dataDir, "config-dir", "/var/lib/rook", "directory for storing configuration")
 	command.Flags().StringVar(&cfg.cephConfigOverride, "ceph-config-override", "", "optional path to a ceph config file that will be appended to the config files that rook generates")
+	command.Flags().StringVar(&clusterInfo.CephVersionName, "ceph-version-name", "", "name of the ceph version")
 
 	// deprecated ipv4 format address
 	// TODO: remove these legacy flags in the future

pkg/daemon/ceph/config/config.go

@@ -19,14 +19,18 @@ package config
 
 import (
 	"fmt"
+	"net"
 	"os"
 	"path"
+	"strconv"
 	"strings"
 
 	"github.com/coreos/pkg/capnslog"
 	"github.com/go-ini/ini"
+	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	"github.com/rook/rook/pkg/clusterd"
 	"github.com/rook/rook/pkg/daemon/ceph/client"
+	cephutil "github.com/rook/rook/pkg/daemon/ceph/util"
 )
 
 var logger = capnslog.NewPackageLogger("github.com/rook/rook", "cephconfig")
@@ -38,6 +42,8 @@ const (
 	DefaultConfigFile = "ceph.conf"
 	// DefaultKeyringFile is the default name of the file where Ceph stores its keyring info
 	DefaultKeyringFile = "keyring"
+	// Msgr2port is the listening port of the messenger v2 protocol
+	Msgr2port = 3300
 )
 
 // GlobalConfig represents the [global] sections of Ceph's config file.
@@ -124,11 +130,7 @@ func GenerateAdminConnectionConfigWithSettings(context *clusterd.Context, cluste
 }
 
 // GenerateConfigFile generates and writes a config file to disk.
-func GenerateConfigFile(context *clusterd.Context,
-	cluster *ClusterInfo,
-	pathRoot, user, keyringPath string,
-	globalConfig *CephConfig,
-	clientSettings map[string]string) (string, error) {
+func GenerateConfigFile(context *clusterd.Context, cluster *ClusterInfo, pathRoot, user, keyringPath string, globalConfig *CephConfig, clientSettings map[string]string) (string, error) {
 
 	// create the config directory
 	if err := os.MkdirAll(pathRoot, 0744); err != nil {
@@ -189,7 +191,22 @@ func CreateDefaultCephConfig(context *clusterd.Context, cluster *ClusterInfo, ru
 	i := 0
 	for _, monitor := range cluster.Monitors {
 		monMembers[i] = monitor.Name
-		monHosts[i] = monitor.Endpoint
+		monIP := cephutil.GetIPFromEndpoint(monitor.Endpoint)
+
+		// This tries to detect the current port if the mon already exists
+		// This basically handles the transtion between monitors running on 6790 to msgr2
+		// So whatever the previous monitor port was we keep it
+		currentMonPort := cephutil.GetPortFromEndpoint(monitor.Endpoint)
+
+		monPorts := [2]string{strconv.Itoa(int(Msgr2port)), strconv.Itoa(int(currentMonPort))}
+		msgr2Endpoint := net.JoinHostPort(monIP, monPorts[0])
+		msgr1Endpoint := net.JoinHostPort(monIP, monPorts[1])
+
+		if cephv1.VersionAtLeast(cluster.CephVersionName, cephv1.Nautilus) {
+			monHosts[i] = "[v2:" + msgr2Endpoint + ",v1:" + msgr1Endpoint + "]"
+		} else {
+			monHosts[i] = msgr1Endpoint
+		}
 		i++
 	}
 

pkg/daemon/ceph/config/config_test.go

@@ -25,6 +25,7 @@ import (
 
 	"github.com/coreos/pkg/capnslog"
 	"github.com/go-ini/ini"
+	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	"github.com/rook/rook/pkg/clusterd"
 	"github.com/stretchr/testify/assert"
 )
@@ -39,10 +40,9 @@ func TestCreateDefaultCephConfig(t *testing.T) {
 			"node0": {Name: "mon0", Endpoint: "10.0.0.1:6789"},
 			"node1": {Name: "mon1", Endpoint: "10.0.0.2:6789"},
 		},
+		CephVersionName: "mimic",
 	}
 
-	monMembers := "mon0 mon1"
-
 	// start with INFO level logging
 	context := &clusterd.Context{
 		LogLevel: capnslog.INFO,
@@ -55,13 +55,13 @@ func TestCreateDefaultCephConfig(t *testing.T) {
 	}
 
 	cephConfig := CreateDefaultCephConfig(context, clusterInfo, "/var/lib/rook1")
-	verifyConfig(t, cephConfig, monMembers, 0)
+	verifyConfig(t, cephConfig, clusterInfo, 0)
 
 	// now use DEBUG level logging
 	context.LogLevel = capnslog.DEBUG
 
 	cephConfig = CreateDefaultCephConfig(context, clusterInfo, "/var/lib/rook1")
-	verifyConfig(t, cephConfig, monMembers, 10)
+	verifyConfig(t, cephConfig, clusterInfo, 10)
 
 	// verify the network info config
 	assert.Equal(t, "10.1.1.1", cephConfig.PublicAddr)
@@ -100,6 +100,7 @@ debug bluestore = 1234`
 		Monitors: map[string]*MonInfo{
 			"node0": {Name: "mon0", Endpoint: "10.0.0.1:6789"},
 		},
+		CephVersionName: "mimic",
 	}
 
 	// generate the config file to disk now
@@ -116,18 +117,39 @@ debug bluestore = 1234`
 	verifyConfigValue(t, actualConf, "global", "debug bluestore", "1234")
 }
 
-func verifyConfig(t *testing.T, cephConfig *CephConfig, expectedMonMembers string, loggingLevel int) {
-
-	for _, expectedMon := range strings.Split(expectedMonMembers, " ") {
+func verifyConfig(t *testing.T, cephConfig *CephConfig, cluster *ClusterInfo, loggingLevel int) {
+	monMembers := make([]string, len(cluster.Monitors))
+	i := 0
+	for _, expectedMon := range cluster.Monitors {
 		contained := false
+		monMembers[i] = expectedMon.Name
 		for _, actualMon := range strings.Split(cephConfig.MonMembers, " ") {
+			if expectedMon.Name == actualMon {
+				contained = true
+				break
+			}
+		}
+
+		assert.True(t, contained)
+	}
+
+	// Testing mon_host
+	expectedMons := "10.0.0.1:6789,10.0.0.2:6789"
+
+	if cephv1.VersionAtLeast(cluster.CephVersionName, cephv1.Nautilus) {
+		expectedMons = "[v2:10.0.0.1:3300,v1:10.0.0.1:6789],[v2:10.0.0.2:3300,v1:10.0.0.2:6789]"
+	}
+
+	for _, expectedMon := range strings.Split(expectedMons, ",") {
+		contained := false
+		for _, actualMon := range strings.Split(cephConfig.MonHost, ",") {
 			if expectedMon == actualMon {
 				contained = true
 				break
 			}
 		}
 
-		assert.True(t, contained, "expectedMons: %+v, actualMons: %+v", expectedMonMembers, cephConfig.MonMembers)
+		assert.True(t, contained, "expectedMons: %+v, actualMons: %+v", expectedMons, cephConfig.MonHost)
 	}
 
 	assert.Equal(t, loggingLevel, cephConfig.DebugLogDefaultLevel)

pkg/daemon/ceph/config/info.go

@@ -27,11 +27,12 @@ import (
 // ClusterInfo is a collection of information about a particular Ceph cluster. Rook uses information
 // about the cluster to configure daemons to connect to the desired cluster.
 type ClusterInfo struct {
-	FSID          string
-	MonitorSecret string
-	AdminSecret   string
-	Name          string
-	Monitors      map[string]*MonInfo
+	FSID            string
+	MonitorSecret   string
+	AdminSecret     string
+	Name            string
+	Monitors        map[string]*MonInfo
+	CephVersionName string
 }
 
 // MonInfo is a collection of information about a Ceph mon.

pkg/daemon/ceph/util/util.go

@@ -19,9 +19,13 @@ package util
 import (
 	"fmt"
 	"io/ioutil"
+	"net"
 	"os"
 	"path/filepath"
+	"strconv"
 	"strings"
+
+	"github.com/coreos/pkg/capnslog"
 )
 
 const (
@@ -30,6 +34,8 @@ const (
 	RBDDevicePathPrefix  = "/dev/rbd"
 )
 
+var logger = capnslog.NewPackageLogger("github.com/rook/rook", "op-ceph-util")
+
 // FindRBDMappedFile search for the mapped RBD volume and returns its device path
 func FindRBDMappedFile(imageName, poolName, sysBusDir string) (string, error) {
 
@@ -57,3 +63,27 @@ func FindRBDMappedFile(imageName, poolName, sysBusDir string) (string, error) {
 	}
 	return "", nil
 }
+
+// GetIPFromEndpoint return the IP from an endpoint string (192.168.0.1:6789)
+func GetIPFromEndpoint(endpoint string) string {
+	host, _, err := net.SplitHostPort(endpoint)
+	if err != nil {
+		logger.Errorf("failed to split ip and port for endpoint %s", endpoint)
+	}
+	return host
+}
+
+// GetPortFromEndpoint return the port from an endpoint string (192.168.0.1:6789)
+func GetPortFromEndpoint(endpoint string) int32 {
+	var port int
+	_, portString, err := net.SplitHostPort(endpoint)
+	if err != nil {
+		logger.Errorf("failed to split host and port for endpoint %s, assuming default Ceph port %s", endpoint, portString)
+	} else {
+		port, err = strconv.Atoi(portString)
+		if err != nil {
+			logger.Errorf("failed to convert %s to integer", portString)
+		}
+	}
+	return int32(port)
+}

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

@@ -27,7 +27,7 @@ import (
 	opspec "github.com/rook/rook/pkg/operator/ceph/spec"
 	"github.com/rook/rook/pkg/operator/k8sutil"
 	apps "k8s.io/api/apps/v1"
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 

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

@@ -32,7 +32,7 @@ import (
 	"github.com/rook/rook/pkg/operator/k8sutil"
 	"github.com/rook/rook/pkg/util/exec"
 	"github.com/rook/rook/pkg/util/sys"
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"

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

@@ -16,7 +16,7 @@ limitations under the License.
 
 package mon
 
-import "k8s.io/api/core/v1"
+import v1 "k8s.io/api/core/v1"
 
 // ClusterNameEnvVar is the cluster name environment var
 func ClusterNameEnvVar(name string) v1.EnvVar {

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

@@ -33,7 +33,7 @@ import (
 	exectest "github.com/rook/rook/pkg/util/exec/test"
 	"github.com/stretchr/testify/assert"
 	apps "k8s.io/api/apps/v1"
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/kubernetes/pkg/kubelet/apis"
 )

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

@@ -33,15 +33,20 @@ import (
 	"github.com/rook/rook/pkg/clusterd"
 	"github.com/rook/rook/pkg/daemon/ceph/client"
 	cephconfig "github.com/rook/rook/pkg/daemon/ceph/config"
+	cephutil "github.com/rook/rook/pkg/daemon/ceph/util"
 	"github.com/rook/rook/pkg/operator/ceph/config"
 	"github.com/rook/rook/pkg/operator/ceph/config/keyring"
 	"github.com/rook/rook/pkg/operator/k8sutil"
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
-var logger = capnslog.NewPackageLogger("github.com/rook/rook", "op-mon")
+var (
+	logger = capnslog.NewPackageLogger("github.com/rook/rook", "op-mon")
+	// DefaultPort is the default port Ceph mons use to communicate amongst themselves.
+	DefaultPort int32 = 6789
+)
 
 const (
 	// EndpointConfigMapName is the name of the configmap with mon endpoints
@@ -62,12 +67,12 @@ const (
 	adminSecretName   = "admin-secret"
 	clusterSecretName = "cluster-name"
 
-	// DefaultPort is the default port mons use to communicate with each other
-	DefaultPort = 6789
 	// DefaultMonCount Default mon count for a cluster
 	DefaultMonCount = 3
 	// MaxMonCount Maximum allowed mon count for a cluster
 	MaxMonCount = 9
+	// Msgr2port is the listening port of the messenger v2 protocol
+	Msgr2port int32 = 3300
 )
 
 // Cluster represents the Rook and environment configuration settings needed to set up Ceph mons.
@@ -252,6 +257,8 @@ func (c *Cluster) initClusterInfo() error {
 	var err error
 	// get the cluster info from secret
 	c.clusterInfo, c.maxMonID, c.mapping, err = CreateOrLoadClusterInfo(c.context, c.Namespace, &c.ownerRef)
+	c.clusterInfo.CephVersionName = c.cephVersion.Name
+
 	if err != nil {
 		return fmt.Errorf("failed to get cluster info. %+v", err)
 	}
@@ -282,7 +289,7 @@ func (c *Cluster) initMonConfig(size int) (int, []*monConfig) {
 		mons = append(mons, &monConfig{
 			ResourceName: resourceName(monitor.Name),
 			DaemonName:   monitor.Name,
-			Port:         getPortFromEndpoint(monitor.Endpoint),
+			Port:         cephutil.GetPortFromEndpoint(monitor.Endpoint),
 			DataPathMap: config.NewStatefulDaemonDataPathMap(
 				c.dataDirHostPath, dataDirRelativeHostPath(monitor.Name), config.MonType, monitor.Name),
 		})
@@ -300,10 +307,11 @@ func (c *Cluster) initMonConfig(size int) (int, []*monConfig) {
 
 func (c *Cluster) newMonConfig(monID int) *monConfig {
 	daemonName := k8sutil.IndexToName(monID)
+
 	return &monConfig{
 		ResourceName: resourceName(daemonName),
 		DaemonName:   daemonName,
-		Port:         int32(DefaultPort),
+		Port:         DefaultPort,
 		DataPathMap: config.NewStatefulDaemonDataPathMap(
 			c.dataDirHostPath, dataDirRelativeHostPath(daemonName), config.MonType, daemonName),
 	}

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

@@ -24,7 +24,7 @@ import (
 	"github.com/rook/rook/pkg/clusterd"
 	"github.com/rook/rook/pkg/operator/test"
 	"github.com/stretchr/testify/assert"
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 

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

@@ -18,9 +18,11 @@ package mon
 
 import (
 	"fmt"
+	"strconv"
 
+	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	"github.com/rook/rook/pkg/operator/k8sutil"
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/intstr"
@@ -50,6 +52,11 @@ func (c *Cluster) createService(mon *monConfig) (string, error) {
 		s.Spec.ClusterIP = v1.ClusterIPNone
 	}
 
+	// If deploying Nautilus or newer we need a new port for the monitor service
+	if cephv1.VersionAtLeast(c.cephVersion.Name, cephv1.Nautilus) {
+		addServicePort(s, "msgr2", Msgr2port)
+	}
+
 	s, err := c.context.Clientset.CoreV1().Services(c.Namespace).Create(s)
 	if err != nil {
 		if !errors.IsAlreadyExists(err) {
@@ -66,6 +73,13 @@ func (c *Cluster) createService(mon *monConfig) (string, error) {
 		return "", nil
 	}
 
-	logger.Infof("mon %s endpoint is %s:%d", mon.DaemonName, s.Spec.ClusterIP, mon.Port)
+	// mon endpoint are not actually like, they remain with the mgrs1 format
+	// however it's interesting to show that monitors can be addressed via 2 different ports
+	// in the end the service has msgr1 and msgr2 ports configured so it's not entirely wrong
+	if cephv1.VersionAtLeast(c.cephVersion.Name, cephv1.Nautilus) {
+		logger.Infof("mon %s endpoint are [v2:%s:%s,v1:%s:%d]", mon.DaemonName, s.Spec.ClusterIP, strconv.Itoa(int(Msgr2port)), s.Spec.ClusterIP, mon.Port)
+	} else {
+		logger.Infof("mon %s endpoint is %s:%d", mon.DaemonName, s.Spec.ClusterIP, mon.Port)
+	}
 	return s.Spec.ClusterIP, nil
 }

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

@@ -17,16 +17,15 @@ limitations under the License.
 package mon
 
 import (
-	"fmt"
-	"net"
 	"os"
 
+	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	"github.com/rook/rook/pkg/operator/ceph/config"
 
 	opspec "github.com/rook/rook/pkg/operator/ceph/spec"
 	"github.com/rook/rook/pkg/operator/k8sutil"
 	apps "k8s.io/api/apps/v1"
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/kubernetes/pkg/kubelet/apis"
 )
@@ -138,6 +137,9 @@ func (c *Cluster) makeMonFSInitContainer(monConfig *monConfig) v1.Container {
 		},
 		Args: append(
 			opspec.DaemonFlags(c.clusterInfo, config.MonType, monConfig.DaemonName),
+			// needed so we can generate an initial monmap
+			// otherwise the mkfs will say: "0  no local addrs match monmap"
+			config.NewFlag("public-addr", monConfig.PublicIP),
 			"--mkfs",
 		),
 		Image:           c.cephVersion.Image,
@@ -150,7 +152,7 @@ func (c *Cluster) makeMonFSInitContainer(monConfig *monConfig) v1.Container {
 }
 
 func (c *Cluster) makeMonDaemonContainer(monConfig *monConfig) v1.Container {
-	return v1.Container{
+	container := v1.Container{
 		Name: "mon",
 		Command: []string{
 			cephMonCommand,
@@ -158,8 +160,8 @@ func (c *Cluster) makeMonDaemonContainer(monConfig *monConfig) v1.Container {
 		Args: append(
 			opspec.DaemonFlags(c.clusterInfo, config.MonType, monConfig.DaemonName),
 			"--foreground",
-			config.NewFlag("public-addr", joinHostPort(monConfig.PublicIP, monConfig.Port)),
-			config.NewFlag("public-bind-addr", joinHostPort("$(ROOK_PRIVATE_IP)", monConfig.Port)),
+			config.NewFlag("public-addr", monConfig.PublicIP),
+			config.NewFlag("public-bind-addr", "$(ROOK_PRIVATE_IP)"),
 		),
 		Image:           c.cephVersion.Image,
 		VolumeMounts:    opspec.DaemonVolumeMounts(monConfig.DataPathMap, keyringStoreName),
@@ -177,8 +179,11 @@ func (c *Cluster) makeMonDaemonContainer(monConfig *monConfig) v1.Container {
 		),
 		Resources: c.resources,
 	}
-}
 
-func joinHostPort(host string, port int32) string {
-	return net.JoinHostPort(host, fmt.Sprintf("%d", port))
+	// If deploying Nautilus and newer we need a new port of the monitor container
+	if cephv1.VersionAtLeast(c.cephVersion.Name, cephv1.Nautilus) {
+		addContainerPort(container, "msgr2", 3300)
+	}
+
+	return container
 }

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

@@ -105,9 +105,11 @@ func testPodSpec(t *testing.T, monID string) {
 
 	// mon fs init container
 	monFsInitContDef := test_opceph.ContainerTestDefinition{
-		Image:            &cephImage,
-		Command:          []string{"ceph-mon"},
-		Args:             append(commonFlags, []string{"--mkfs"}),
+		Image:   &cephImage,
+		Command: []string{"ceph-mon"},
+		Args: append(commonFlags,
+			[]string{"--public-addr=2.4.6.1"},
+			[]string{"--mkfs"}),
 		VolumeMountNames: cephVolumeMountNames,
 		EnvCount:         &envVars,
 		Ports:            []v1.ContainerPort{},
@@ -126,8 +128,8 @@ func testPodSpec(t *testing.T, monID string) {
 			"ceph-mon"},
 		Args: append(commonFlags,
 			[]string{"--foreground"},
-			[]string{"--public-addr=2.4.6.1:6789"},
-			[]string{"--public-bind-addr=$(ROOK_PRIVATE_IP):6789"}),
+			[]string{"--public-addr=2.4.6.1"},
+			[]string{"--public-bind-addr=$(ROOK_PRIVATE_IP)"}),
 		VolumeMountNames: cephVolumeMountNames,
 		EnvCount:         &monDaemonEnvs,
 		Ports: []v1.ContainerPort{

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

@@ -18,12 +18,13 @@ package mon
 
 import (
 	"fmt"
-	"net"
 	"strconv"
 	"strings"
 
 	"github.com/rook/rook/pkg/daemon/ceph/client"
 	"github.com/rook/rook/pkg/operator/k8sutil"
+	v1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/util/intstr"
 )
 
 const (
@@ -58,14 +59,27 @@ func fullNameToIndex(name string) (int, error) {
 	return id, nil
 }
 
-// getPortFromEndpoint return the port from an endpoint string (192.168.0.1:6790)
-func getPortFromEndpoint(endpoint string) int32 {
-	port := DefaultPort
-	_, portString, err := net.SplitHostPort(endpoint)
-	if err != nil {
-		logger.Errorf("failed to split host and port for endpoint %s, assuming default Ceph port %d", endpoint, port)
-	} else {
-		port, _ = strconv.Atoi(portString)
+// addServicePort adds a port to a service
+func addServicePort(service *v1.Service, name string, port int32) {
+	if port == 0 {
+		return
 	}
-	return int32(port)
+	service.Spec.Ports = append(service.Spec.Ports, v1.ServicePort{
+		Name:       name,
+		Port:       port,
+		TargetPort: intstr.FromInt(int(port)),
+		Protocol:   v1.ProtocolTCP,
+	})
+}
+
+// addContainerPort adds a port to a container
+func addContainerPort(container v1.Container, name string, port int32) {
+	if port == 0 {
+		return
+	}
+	container.Ports = append(container.Ports, v1.ContainerPort{
+		Name:          name,
+		ContainerPort: port,
+		Protocol:      v1.ProtocolTCP,
+	})
 }

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

@@ -32,7 +32,7 @@ import (
 	"github.com/rook/rook/pkg/operator/k8sutil"
 	apps "k8s.io/api/apps/v1"
 	batch "k8s.io/api/batch/v1"
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/kubernetes/pkg/kubelet/apis"
 )

pkg/operator/ceph/cluster/osd/spec_test.go

@@ -28,7 +28,7 @@ import (
 	exectest "github.com/rook/rook/pkg/util/exec/test"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes/fake"
@@ -36,7 +36,7 @@ import (
 )
 
 func TestPodContainer(t *testing.T) {
-	cluster := &Cluster{Namespace: "myosd", rookVersion: "23"}
+	cluster := &Cluster{Namespace: "myosd", rookVersion: "23", cephVersion: cephv1.CephVersionSpec{Name: "mimic"}}
 	c, err := cluster.provisionPodTemplateSpec([]rookalpha.Device{}, rookalpha.Selection{}, v1.ResourceRequirements{}, config.StoreConfig{}, "", "node", "", v1.RestartPolicyAlways)
 	assert.NotNil(t, c)
 	assert.Nil(t, err)

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

@@ -22,7 +22,7 @@ import (
 	opspec "github.com/rook/rook/pkg/operator/ceph/spec"
 	"github.com/rook/rook/pkg/operator/k8sutil"
 	apps "k8s.io/api/apps/v1"
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 

pkg/operator/ceph/config/store.go

@@ -21,13 +21,17 @@ package config
 import (
 	"bytes"
 	"fmt"
+	"net"
+	"strconv"
 	"strings"
 
 	"github.com/go-ini/ini"
+	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	"github.com/rook/rook/pkg/clusterd"
 	cephconfig "github.com/rook/rook/pkg/daemon/ceph/config"
+	cephutil "github.com/rook/rook/pkg/daemon/ceph/util"
 	"github.com/rook/rook/pkg/operator/k8sutil"
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
@@ -40,6 +44,8 @@ const (
 	confFileName         = "ceph.conf"
 	monHostKey           = "mon_host"
 	monInitialMembersKey = "mon_initial_members"
+	// Msgr2port is the listening port of the messenger v2 protocol
+	Msgr2port = 3300
 )
 
 // Store manages storage of the Ceph config file shared by all daemons (if applicable) as well as an
@@ -152,7 +158,23 @@ func (s *Store) createOrUpdateMonHostSecrets(clusterInfo *cephconfig.ClusterInfo
 	members := make([]string, len(clusterInfo.Monitors))
 	i := 0
 	for _, m := range clusterInfo.Monitors {
-		hosts[i] = m.Endpoint
+		monIP := cephutil.GetIPFromEndpoint(m.Endpoint)
+
+		// This tries to detect the current port if the mon already exists
+		// This basically handles the transtion between monitors running on 6790 to msgr2
+		// So whatever the previous monitor port was we keep it
+		currentMonPort := cephutil.GetPortFromEndpoint(m.Endpoint)
+
+		monPorts := [2]string{strconv.Itoa(int(Msgr2port)), strconv.Itoa(int(currentMonPort))}
+		msgr2Endpoint := net.JoinHostPort(monIP, monPorts[0])
+		msgr1Endpoint := net.JoinHostPort(monIP, monPorts[1])
+
+		if cephv1.VersionAtLeast(clusterInfo.CephVersionName, cephv1.Nautilus) {
+			hosts[i] = "[v2:" + msgr2Endpoint + ",v1:" + msgr1Endpoint + "]"
+		} else {
+			hosts[i] = msgr1Endpoint
+		}
+
 		members[i] = m.Name
 		i++
 	}

pkg/operator/ceph/file/spec.go

@@ -24,7 +24,7 @@ import (
 	opspec "github.com/rook/rook/pkg/operator/ceph/spec"
 	"github.com/rook/rook/pkg/operator/k8sutil"
 	apps "k8s.io/api/apps/v1"
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 

pkg/operator/ceph/file/spec_test.go

@@ -29,7 +29,7 @@ import (
 	testop "github.com/rook/rook/pkg/operator/test"
 	"github.com/stretchr/testify/assert"
 	apps "k8s.io/api/apps/v1"
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
@@ -52,7 +52,7 @@ func testDeploymentObject(hostNetwork bool) *apps.Deployment {
 	c := newCluster(
 		&clusterd.Context{Clientset: testop.New(1)},
 		"rook/rook:myversion",
-		cephv1.CephVersionSpec{Image: "ceph/ceph:testversion"},
+		cephv1.CephVersionSpec{Image: "ceph/ceph:testversion", Name: "mimic"},
 		hostNetwork,
 		fs,
 		&client.CephFilesystemDetails{ID: 15},

pkg/operator/ceph/nfs/spec.go

@@ -26,7 +26,7 @@ import (
 	opspec "github.com/rook/rook/pkg/operator/ceph/spec"
 	"github.com/rook/rook/pkg/operator/k8sutil"
 	apps "k8s.io/api/apps/v1"
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/intstr"
@@ -177,7 +177,8 @@ func (c *CephNFSController) daemonContainer(n cephv1.CephNFS, name string, binar
 		),
 		Env: append(
 			k8sutil.ClusterDaemonEnvVars(),
-			v1.EnvVar{Name: "ROOK_CEPH_NFS_NAME", Value: name}),
+			v1.EnvVar{Name: "ROOK_CEPH_NFS_NAME", Value: name},
+		),
 		Resources: n.Spec.Server.Resources,
 	}
 }

pkg/operator/ceph/object/spec_test.go

@@ -41,7 +41,7 @@ func TestPodSpecs(t *testing.T) {
 		},
 	}
 
-	c := &config{store: store, rookVersion: "rook/rook:myversion", cephVersion: cephv1.CephVersionSpec{Image: "ceph/ceph:v13.2.1"}, hostNetwork: true}
+	c := &config{store: store, rookVersion: "rook/rook:myversion", cephVersion: cephv1.CephVersionSpec{Image: "ceph/ceph:v13.2.1", Name: "mimic"}, hostNetwork: true}
 	s := c.makeRGWPodSpec()
 	assert.NotNil(t, s)
 	//assert.Equal(t, instanceName(store), s.Name)

pkg/operator/ceph/spec/spec.go

@@ -25,7 +25,7 @@ import (
 	"github.com/rook/rook/pkg/operator/ceph/config"
 	"github.com/rook/rook/pkg/operator/ceph/config/keyring"
 	"github.com/rook/rook/pkg/operator/k8sutil"
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 )
 
 const (

pkg/operator/k8sutil/pod.go

@@ -28,7 +28,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
 
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 )
 
@@ -304,3 +304,8 @@ func ClusterDaemonEnvVars() []v1.EnvVar {
 		{Name: "NODE_NAME", ValueFrom: &v1.EnvVarSource{FieldRef: &v1.ObjectFieldSelector{FieldPath: "spec.nodeName"}}},
 	}
 }
+
+// CephVersionName is the name of the ceph version
+func CephVersionName(version string) v1.EnvVar {
+	return v1.EnvVar{Name: "ROOK_CEPH_VERSION_NAME", Value: version}
+}

pkg/operator/test/spec.go

@@ -46,8 +46,8 @@ func ArgumentsMatchExpected(actualArgs []string, expectedArgs [][]string) error
 	fullArgString := strings.Join(actualArgs, " ")
 	logger.Infof("testing that actual args: %s\nmatch expected args:%v", fullArgString, actualArgs)
 	for _, arg := range expectedArgs {
-		// We join each individual argument together the same was as the big string
 		validArgMatcher := strings.Join(arg, " ")
+		// We join each individual argument together the same was as the big string
 		if validArgMatcher == "" {
 			return fmt.Errorf("Expected argument %v evaluated to empty string; ArgumentsMatchExpected() doesn't know what to do", arg)
 		}