From eb3c424d57cc9637bf07c9cc43c8ffb8eb7b61a1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?S=C3=A9bastien=20Han?= Date: Thu, 24 Jan 2019 14:58:46 +0100 Subject: [PATCH] Ceph messengers 2 support MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 3300/TCP,6789/TCP 4h rook-ceph-mon-b ClusterIP 10.99.36.175 3300/TCP,6789/TCP 4h rook-ceph-mon-c ClusterIP 10.108.220.74 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 --- cmd/rook/ceph/ceph.go | 1 + pkg/daemon/ceph/config/config.go | 29 +++++++++++---- pkg/daemon/ceph/config/config_test.go | 38 +++++++++++++++----- pkg/daemon/ceph/config/info.go | 11 +++--- pkg/daemon/ceph/util/util.go | 30 ++++++++++++++++ pkg/operator/ceph/cluster/mgr/spec.go | 2 +- pkg/operator/ceph/cluster/mon/config.go | 2 +- pkg/operator/ceph/cluster/mon/env.go | 2 +- pkg/operator/ceph/cluster/mon/health_test.go | 2 +- pkg/operator/ceph/cluster/mon/mon.go | 20 +++++++---- pkg/operator/ceph/cluster/mon/node_test.go | 2 +- pkg/operator/ceph/cluster/mon/service.go | 18 ++++++++-- pkg/operator/ceph/cluster/mon/spec.go | 23 +++++++----- pkg/operator/ceph/cluster/mon/spec_test.go | 12 ++++--- pkg/operator/ceph/cluster/mon/util.go | 34 ++++++++++++------ pkg/operator/ceph/cluster/osd/spec.go | 2 +- pkg/operator/ceph/cluster/osd/spec_test.go | 4 +-- pkg/operator/ceph/cluster/rbd/spec.go | 2 +- pkg/operator/ceph/config/store.go | 26 ++++++++++++-- pkg/operator/ceph/file/spec.go | 2 +- pkg/operator/ceph/file/spec_test.go | 4 +-- pkg/operator/ceph/nfs/spec.go | 5 +-- pkg/operator/ceph/object/spec_test.go | 2 +- pkg/operator/ceph/spec/spec.go | 2 +- pkg/operator/k8sutil/pod.go | 7 +++- pkg/operator/test/spec.go | 2 +- 26 files changed, 213 insertions(+), 71 deletions(-) diff --git a/cmd/rook/ceph/ceph.go b/cmd/rook/ceph/ceph.go index f40d389fa..7424a08e2 100644 --- a/cmd/rook/ceph/ceph.go +++ b/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 diff --git a/pkg/daemon/ceph/config/config.go b/pkg/daemon/ceph/config/config.go index 68f1579f6..104f4ae06 100644 --- a/pkg/daemon/ceph/config/config.go +++ b/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++ } diff --git a/pkg/daemon/ceph/config/config_test.go b/pkg/daemon/ceph/config/config_test.go index 2cd6d0f5e..48ba097a4 100644 --- a/pkg/daemon/ceph/config/config_test.go +++ b/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) diff --git a/pkg/daemon/ceph/config/info.go b/pkg/daemon/ceph/config/info.go index c4bb88ce9..e5e6ddfb1 100644 --- a/pkg/daemon/ceph/config/info.go +++ b/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. diff --git a/pkg/daemon/ceph/util/util.go b/pkg/daemon/ceph/util/util.go index 84abb8a7e..0db9c937b 100644 --- a/pkg/daemon/ceph/util/util.go +++ b/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) +} diff --git a/pkg/operator/ceph/cluster/mgr/spec.go b/pkg/operator/ceph/cluster/mgr/spec.go index b45b66db2..b3d7b1881 100644 --- a/pkg/operator/ceph/cluster/mgr/spec.go +++ b/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" ) diff --git a/pkg/operator/ceph/cluster/mon/config.go b/pkg/operator/ceph/cluster/mon/config.go index b237798af..d6e6ba2e7 100644 --- a/pkg/operator/ceph/cluster/mon/config.go +++ b/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" diff --git a/pkg/operator/ceph/cluster/mon/env.go b/pkg/operator/ceph/cluster/mon/env.go index 79341d52a..41f18b239 100644 --- a/pkg/operator/ceph/cluster/mon/env.go +++ b/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 { diff --git a/pkg/operator/ceph/cluster/mon/health_test.go b/pkg/operator/ceph/cluster/mon/health_test.go index 3b1914dc0..1ac37cc16 100644 --- a/pkg/operator/ceph/cluster/mon/health_test.go +++ b/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" ) diff --git a/pkg/operator/ceph/cluster/mon/mon.go b/pkg/operator/ceph/cluster/mon/mon.go index e1da8dbbd..c09765aea 100644 --- a/pkg/operator/ceph/cluster/mon/mon.go +++ b/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), } diff --git a/pkg/operator/ceph/cluster/mon/node_test.go b/pkg/operator/ceph/cluster/mon/node_test.go index e794bf5b4..948235c48 100644 --- a/pkg/operator/ceph/cluster/mon/node_test.go +++ b/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" ) diff --git a/pkg/operator/ceph/cluster/mon/service.go b/pkg/operator/ceph/cluster/mon/service.go index d69de0344..e6e1b728f 100644 --- a/pkg/operator/ceph/cluster/mon/service.go +++ b/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 } diff --git a/pkg/operator/ceph/cluster/mon/spec.go b/pkg/operator/ceph/cluster/mon/spec.go index 66c524d03..87accd11f 100644 --- a/pkg/operator/ceph/cluster/mon/spec.go +++ b/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 } diff --git a/pkg/operator/ceph/cluster/mon/spec_test.go b/pkg/operator/ceph/cluster/mon/spec_test.go index 173338466..5f23e8d44 100644 --- a/pkg/operator/ceph/cluster/mon/spec_test.go +++ b/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{ diff --git a/pkg/operator/ceph/cluster/mon/util.go b/pkg/operator/ceph/cluster/mon/util.go index 956a10c13..0c0fbfc11 100644 --- a/pkg/operator/ceph/cluster/mon/util.go +++ b/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, + }) } diff --git a/pkg/operator/ceph/cluster/osd/spec.go b/pkg/operator/ceph/cluster/osd/spec.go index 24e811f46..4cdfc05cd 100644 --- a/pkg/operator/ceph/cluster/osd/spec.go +++ b/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" ) diff --git a/pkg/operator/ceph/cluster/osd/spec_test.go b/pkg/operator/ceph/cluster/osd/spec_test.go index 2517e820c..99ed3324f 100644 --- a/pkg/operator/ceph/cluster/osd/spec_test.go +++ b/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) diff --git a/pkg/operator/ceph/cluster/rbd/spec.go b/pkg/operator/ceph/cluster/rbd/spec.go index f246c51b8..8dbbcc6c0 100644 --- a/pkg/operator/ceph/cluster/rbd/spec.go +++ b/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" ) diff --git a/pkg/operator/ceph/config/store.go b/pkg/operator/ceph/config/store.go index c7cb122bf..39cde4add 100644 --- a/pkg/operator/ceph/config/store.go +++ b/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++ } diff --git a/pkg/operator/ceph/file/spec.go b/pkg/operator/ceph/file/spec.go index a63d4032e..611d78576 100644 --- a/pkg/operator/ceph/file/spec.go +++ b/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" ) diff --git a/pkg/operator/ceph/file/spec_test.go b/pkg/operator/ceph/file/spec_test.go index dc0e62591..2682d04f4 100644 --- a/pkg/operator/ceph/file/spec_test.go +++ b/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}, diff --git a/pkg/operator/ceph/nfs/spec.go b/pkg/operator/ceph/nfs/spec.go index 7e03a2e11..637a6ee1c 100644 --- a/pkg/operator/ceph/nfs/spec.go +++ b/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, } } diff --git a/pkg/operator/ceph/object/spec_test.go b/pkg/operator/ceph/object/spec_test.go index e964730ce..98f14883d 100644 --- a/pkg/operator/ceph/object/spec_test.go +++ b/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) diff --git a/pkg/operator/ceph/spec/spec.go b/pkg/operator/ceph/spec/spec.go index 42a4fae25..77d5bd24f 100644 --- a/pkg/operator/ceph/spec/spec.go +++ b/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 ( diff --git a/pkg/operator/k8sutil/pod.go b/pkg/operator/k8sutil/pod.go index 121d0cea2..7eb3c16aa 100644 --- a/pkg/operator/k8sutil/pod.go +++ b/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} +} diff --git a/pkg/operator/test/spec.go b/pkg/operator/test/spec.go index 19a7c2fbd..41c6dd882 100644 --- a/pkg/operator/test/spec.go +++ b/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) }