Merge pull request #2559 from leseb/mon-msgr2

Support for Ceph messenger 2

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