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