Merge pull request #3739 from leseb/external-cluster-upgrades

ceph: external be a bit more flexible on upgrade

pkg/operator/ceph/cluster/cluster.go

@@ -36,10 +36,9 @@ import (
 	"github.com/rook/rook/pkg/operator/ceph/cluster/mon"
 	"github.com/rook/rook/pkg/operator/ceph/cluster/osd"
 	"github.com/rook/rook/pkg/operator/ceph/cluster/rbd"
+	"github.com/rook/rook/pkg/operator/ceph/config"
+	cephspec "github.com/rook/rook/pkg/operator/ceph/spec"
 	cephver "github.com/rook/rook/pkg/operator/ceph/version"
-	"github.com/rook/rook/pkg/operator/k8sutil"
-	v1 "k8s.io/api/core/v1"
-	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
@@ -124,17 +123,19 @@ func (c *cluster) detectCephVersion(rookImage, cephImage string, timeout time.Du
 }
 
 func (c *cluster) validateCephVersion(version *cephver.CephVersion) error {
-	if !version.IsAtLeast(cephver.Minimum) {
-		return fmt.Errorf("the version does not meet the minimum version: %s", cephver.Minimum.String())
-	}
+	// The external has its own supported versions
+	if !c.Spec.External.Enable {
+		if !version.IsAtLeast(cephver.Minimum) {
+			return fmt.Errorf("the version does not meet the minimum version: %s", cephver.Minimum.String())
+		}
 
-	if !version.Supported() {
-		logger.Warningf("unsupported ceph version detected: %s.", version)
-		if !c.Spec.CephVersion.AllowUnsupported {
-			return fmt.Errorf("allowUnsupported must be set to true to run with this version: %v", version)
+		if !version.Supported() {
+			logger.Warningf("unsupported ceph version detected: %s.", version)
+			if !c.Spec.CephVersion.AllowUnsupported {
+				return fmt.Errorf("allowUnsupported must be set to true to run with this version: %v", version)
+			}
 		}
 	}
-
 	// The following tries to determine if the operator can proceed with an upgrade because we come from an OnAdd() call
 	// If the cluster was unhealthy and someone injected a new image version, an upgrade was triggered but failed because the cluster is not healthy
 	// Then after this, if the operator gets restarted we are not able to fail if the cluster is not healthy, the following tries to determine the
@@ -156,6 +157,13 @@ func (c *cluster) validateCephVersion(version *cephver.CephVersion) error {
 		return nil
 	}
 
+	if c.Spec.External.Enable {
+		c.Info.CephVersion, err = cephspec.ValidateCephVersionsBetweenLocalAndExternalClusters(c.context, c.Namespace, *version)
+		if err != nil {
+			return fmt.Errorf("failed to validate ceph version between external and local. %+v", err)
+		}
+	}
+
 	// Get cluster running versions
 	versions, err := client.GetAllCephDaemonVersions(c.context, c.Namespace)
 	if err != nil {
@@ -216,66 +224,71 @@ func (c *cluster) createInstance(rookImage string, cephVersion cephver.CephVersi
 func (c *cluster) doOrchestration(rookImage string, cephVersion cephver.CephVersion, spec *cephv1.ClusterSpec) error {
 	// Create a configmap for overriding ceph config settings
 	// These settings should only be modified by a user after they are initialized
-	placeholderConfig := map[string]string{
-		k8sutil.ConfigOverrideVal: "",
-	}
-	cm := &v1.ConfigMap{
-		ObjectMeta: metav1.ObjectMeta{
-			Name: k8sutil.ConfigOverrideName,
-		},
-		Data: placeholderConfig,
-	}
-	k8sutil.SetOwnerRef(&cm.ObjectMeta, &c.ownerRef)
-	_, err := c.context.Clientset.CoreV1().ConfigMaps(c.Namespace).Create(cm)
-	if err != nil && !errors.IsAlreadyExists(err) {
-		return fmt.Errorf("failed to create override configmap %s. %+v", c.Namespace, err)
-	}
-
-	// This gets triggered on CR update so let's not run that (mon/mgr/osd daemons)
-	// Start the mon pods
-	clusterInfo, err := c.mons.Start(c.Info, rookImage, cephVersion, *c.Spec, c.isUpgrade)
+	err := populateConfigOverrideConfigMap(c.context, c.Namespace, c.ownerRef)
 	if err != nil {
-		return fmt.Errorf("failed to start the mons. %+v", err)
-	}
-	c.Info = clusterInfo // mons return the cluster's info
-
-	// The cluster Identity must be established at this point
-	if !c.Info.IsInitialized() {
-		return fmt.Errorf("the cluster identity was not established: %+v", c.Info)
+		return fmt.Errorf("failed to populate config override config map. %+v", err)
 	}
 
-	mgrs := mgr.New(c.Info, c.context, c.Namespace, rookImage,
-		spec.CephVersion, cephv1.GetMgrPlacement(spec.Placement), cephv1.GetMgrAnnotations(c.Spec.Annotations),
-		spec.Network, spec.Dashboard, spec.Monitoring, cephv1.GetMgrResources(spec.Resources), c.ownerRef, c.Spec.DataDirHostPath, c.isUpgrade)
-	err = mgrs.Start()
-	if err != nil {
-		return fmt.Errorf("failed to start the ceph mgr. %+v", err)
-	}
+	if c.Spec.External.Enable {
+		// Apply CRD ConfigOverrideName to the external cluster
+		err = config.SetDefaultAndUserConfigs(c.context, c.Namespace, c.Info, c.Spec.ConfigOverrides)
+		if err != nil {
+			// Mons are up, so something else is wrong
+			return fmt.Errorf("failed to set Rook and/or user-defined Ceph config options on the external cluster monitors. %+v", err)
+		}
 
-	// Start the OSDs
-	osds := osd.New(c.Info, c.context, c.Namespace, rookImage, spec.CephVersion, spec.Storage, spec.DataDirHostPath,
-		cephv1.GetOSDPlacement(spec.Placement), cephv1.GetOSDAnnotations(spec.Annotations), spec.Network,
-		cephv1.GetOSDResources(spec.Resources), c.ownerRef, c.isUpgrade)
-	err = osds.Start()
-	if err != nil {
-		return fmt.Errorf("failed to start the osds. %+v", err)
-	}
+		// The cluster Identity must be established at this point
+		if !c.Info.IsInitialized() {
+			return fmt.Errorf("the cluster identity was not established: %+v", c.Info)
+		}
+	} else {
+		// This gets triggered on CR update so let's not run that (mon/mgr/osd daemons)
+		// Start the mon pods
+		clusterInfo, err := c.mons.Start(c.Info, rookImage, cephVersion, *c.Spec, c.isUpgrade)
+		if err != nil {
+			return fmt.Errorf("failed to start the mons. %+v", err)
+		}
+		c.Info = clusterInfo // mons return the cluster's info
 
-	// Start the rbd mirroring daemon(s)
-	rbdmirror := rbd.New(c.Info, c.context, c.Namespace, rookImage, spec.CephVersion, cephv1.GetRBDMirrorPlacement(spec.Placement),
-		cephv1.GetRBDMirrorAnnotations(spec.Annotations), spec.Network, spec.RBDMirroring,
-		cephv1.GetRBDMirrorResources(spec.Resources), c.ownerRef, c.Spec.DataDirHostPath, c.isUpgrade)
-	err = rbdmirror.Start()
-	if err != nil {
-		return fmt.Errorf("failed to start the rbd mirrors. %+v", err)
-	}
+		// The cluster Identity must be established at this point
+		if !c.Info.IsInitialized() {
+			return fmt.Errorf("the cluster identity was not established: %+v", c.Info)
+		}
 
-	logger.Infof("Done creating rook instance in namespace %s", c.Namespace)
-	c.initCompleted = true
+		mgrs := mgr.New(c.Info, c.context, c.Namespace, rookImage,
+			spec.CephVersion, cephv1.GetMgrPlacement(spec.Placement), cephv1.GetMgrAnnotations(c.Spec.Annotations),
+			spec.Network, spec.Dashboard, spec.Monitoring, cephv1.GetMgrResources(spec.Resources), c.ownerRef, c.Spec.DataDirHostPath, c.isUpgrade)
+		err = mgrs.Start()
+		if err != nil {
+			return fmt.Errorf("failed to start the ceph mgr. %+v", err)
+		}
+
+		// Start the OSDs
+		osds := osd.New(c.Info, c.context, c.Namespace, rookImage, spec.CephVersion, spec.Storage, spec.DataDirHostPath,
+			cephv1.GetOSDPlacement(spec.Placement), cephv1.GetOSDAnnotations(spec.Annotations), spec.Network,
+			cephv1.GetOSDResources(spec.Resources), c.ownerRef, c.isUpgrade)
+		err = osds.Start()
+		if err != nil {
+			return fmt.Errorf("failed to start the osds. %+v", err)
+		}
+
+		// Start the rbd mirroring daemon(s)
+		rbdmirror := rbd.New(c.Info, c.context, c.Namespace, rookImage, spec.CephVersion, cephv1.GetRBDMirrorPlacement(spec.Placement),
+			cephv1.GetRBDMirrorAnnotations(spec.Annotations), spec.Network, spec.RBDMirroring,
+			cephv1.GetRBDMirrorResources(spec.Resources), c.ownerRef, c.Spec.DataDirHostPath, c.isUpgrade)
+		err = rbdmirror.Start()
+		if err != nil {
+			return fmt.Errorf("failed to start the rbd mirrors. %+v", err)
+		}
+
+		logger.Infof("Done creating rook instance in namespace %s", c.Namespace)
+		c.initCompleted = true
+	}
 
 	// Notify the child controllers that the cluster spec might have changed
+	logger.Debug("notifying CR childs of the potential upgrade")
 	for _, child := range c.childControllers {
-		child.ParentClusterChanged(*c.Spec, clusterInfo, c.isUpgrade)
+		child.ParentClusterChanged(*c.Spec, c.Info, c.isUpgrade)
 	}
 
 	return nil

pkg/operator/ceph/cluster/controller.go

@@ -30,6 +30,7 @@ import (
 	"github.com/rook/rook/pkg/clusterd"
 	"github.com/rook/rook/pkg/daemon/ceph/agent/flexvolume/attachment"
 	"github.com/rook/rook/pkg/daemon/ceph/client"
+	cephconfig "github.com/rook/rook/pkg/daemon/ceph/config"
 	discoverDaemon "github.com/rook/rook/pkg/daemon/discover"
 	"github.com/rook/rook/pkg/operator/ceph/cluster/mon"
 	"github.com/rook/rook/pkg/operator/ceph/cluster/osd"
@@ -61,6 +62,7 @@ const (
 	updateClusterInterval    = 30 * time.Second
 	updateClusterTimeout     = 1 * time.Hour
 	detectCephVersionTimeout = 15 * time.Minute
+	externalConnectionRetry  = 60 * time.Second
 )
 
 const (
@@ -254,19 +256,19 @@ func (c *ClusterController) configureExternalCephCluster(namespace, name string,
 	c.updateClusterStatus(namespace, name, cephv1.ClusterStateConnecting, "")
 
 	// loop until we find the secret necessary to connect to the external cluster
-	for {
-		var err error
-		cluster.Info, _, _, err = mon.LoadClusterInfo(c.context, namespace)
-		if err != nil {
-			logger.Warningf("waiting for the connection info of the external cluster. %+v", err)
-			time.Sleep(10 * time.Second)
-			continue
-		} else {
-			break
-		}
+	// then populate clusterInfo
+	cluster.Info = populateExternalClusterInfo(c.context, namespace)
+
+	// Validate versions (local and external)
+	_, _, err = c.detectAndValidateCephVersion(cluster, cluster.Spec.CephVersion.Image)
+	if err != nil {
+		return fmt.Errorf("failed to detect and validate ceph version. %+v", err)
 	}
 
-	logger.Infof("found the cluster info to connect to the external cluster. mons=%+v", cluster.Info.Monitors)
+	err = populateConfigOverrideConfigMap(cluster.context, namespace, cluster.ownerRef)
+	if err != nil {
+		return fmt.Errorf("failed to populate config override config map. %+v", err)
+	}
 
 	// Write the rook-ceph-config configmap (used by various daemons to apply config overrides)
 	// If we don't do this, daemons will never start, waiting forever for this configmap to be present
@@ -275,36 +277,11 @@ func (c *ClusterController) configureExternalCephCluster(namespace, name string,
 		return err
 	}
 
-	// Let's write connection info (ceph config file and keyring) to the operator for health checks
-	err = mon.WriteConnectionConfig(cluster.context, cluster.Info)
+	// Apply CRD ConfigOverrideName to the external cluster
+	err = config.SetDefaultAndUserConfigs(cluster.context, namespace, cluster.Info, cluster.Spec.ConfigOverrides)
 	if err != nil {
-		return fmt.Errorf("failed to write connection info %+v", err)
-	}
-
-	// Get Ceph monitors version on the external cluster
-	cephMonVersion, err := client.GetCephMonVersion(c.context, namespace)
-	if err != nil {
-		return fmt.Errorf("failed to get ceph mon version. %+v", err)
-	}
-
-	// We only support Nautilus or newer
-	if !cephMonVersion.IsAtLeastNautilus() {
-		return fmt.Errorf("unsupported ceph version %s, need at least nautilus", cephMonVersion.String())
-	}
-
-	logger.Infof("detecting the image version provided for the external cluster...")
-	specCephVersionImage, err := cluster.detectCephVersion(c.rookImage, cluster.Spec.CephVersion.Image, detectCephVersionTimeout)
-	if err != nil {
-		return fmt.Errorf("unknown ceph major version. %+v", err)
-	}
-	specCephVersion := *specCephVersionImage
-
-	// Populate clusterInfo with the external cluster version
-	cluster.Info.CephVersion = *cephMonVersion
-
-	// Make sure the external cluster version and the ceph version in the provided image are identical
-	if !cephver.IsIdentical(specCephVersion, cluster.Info.CephVersion) {
-		return fmt.Errorf("wrong ceph version %s, external cluster version is %s, they must match", specCephVersion.String(), cephMonVersion.String())
+		// Mons are up, so something else is wrong
+		return fmt.Errorf("failed to set Rook and/or user-defined Ceph config options on the external cluster monitors. %+v", err)
 	}
 
 	// The cluster Identity must be established at this point
@@ -981,3 +958,43 @@ func validateExternalClusterSpec(cluster *cluster) error {
 
 	return nil
 }
+
+// Add validation in the code to fail if the external cluster has no OSDs keep waiting
+func populateExternalClusterInfo(context *clusterd.Context, namespace string) *cephconfig.ClusterInfo {
+	var clusterInfo *cephconfig.ClusterInfo
+	for {
+		var err error
+		clusterInfo, _, _, err = mon.LoadClusterInfo(context, namespace)
+		if err != nil {
+			logger.Warningf("waiting for the connection info of the external cluster. retrying in %s.", externalConnectionRetry.String())
+			time.Sleep(externalConnectionRetry)
+			continue
+		} else {
+			logger.Infof("found the cluster info to connect to the external cluster. mons=%+v", clusterInfo.Monitors)
+			break
+		}
+	}
+
+	return clusterInfo
+}
+
+func populateConfigOverrideConfigMap(context *clusterd.Context, namespace string, ownerRef metav1.OwnerReference) error {
+	placeholderConfig := map[string]string{
+		k8sutil.ConfigOverrideVal: "",
+	}
+
+	cm := &v1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: k8sutil.ConfigOverrideName,
+		},
+		Data: placeholderConfig,
+	}
+
+	k8sutil.SetOwnerRef(&cm.ObjectMeta, &ownerRef)
+	_, err := context.Clientset.CoreV1().ConfigMaps(namespace).Create(cm)
+	if err != nil && !errors.IsAlreadyExists(err) {
+		return fmt.Errorf("failed to create override configmap %s. %+v", namespace, err)
+	}
+
+	return nil
+}

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

@@ -25,6 +25,7 @@ import (
 	"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"
+	cephspec "github.com/rook/rook/pkg/operator/ceph/spec"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
@@ -296,6 +297,10 @@ func removeMonitorFromQuorum(context *clusterd.Context, clusterName, name string
 }
 
 func (c *Cluster) handleExternalMonStatus(status client.MonStatusResponse) error {
+	_, err := cephspec.ValidateCephVersionsBetweenLocalAndExternalClusters(c.context, c.Namespace, c.ClusterInfo.CephVersion)
+	if err != nil {
+		return fmt.Errorf("failed to validate external ceph version. %+v", err)
+	}
 
 	changed, err := c.addOrRemoveExternalMonitor(status)
 	if err != nil {
@@ -321,6 +326,15 @@ func (c *Cluster) addOrRemoveExternalMonitor(status client.MonStatusResponse) (b
 		monsNotFound[mon.Name] = struct{}{}
 	}
 
+	monCount := len(status.MonMap.Mons)
+	if monCount%2 == 0 {
+		logger.Warningf("external cluster mon count is even (%d), should be uneven, continuing.", monCount)
+	}
+
+	if monCount == 1 {
+		logger.Warning("external cluster mon count is 1, consider adding new monitors.")
+	}
+
 	// Iterate over the mons first and compare it with ClusterInfo
 	for _, mon := range status.MonMap.Mons {
 		inQuorum := monInQuorum(mon, status.Quorum)

pkg/operator/ceph/config/config.go

@@ -24,13 +24,9 @@ import (
 	"strings"
 
 	"github.com/coreos/pkg/capnslog"
-	"github.com/go-ini/ini"
 	rookceph "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"
-	"github.com/rook/rook/pkg/operator/k8sutil"
-	"k8s.io/apimachinery/pkg/api/errors"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
 var logger = capnslog.NewPackageLogger("github.com/rook/rook", "op-config")
@@ -132,77 +128,3 @@ func SetDefaultAndUserConfigs(
 
 	return nil
 }
-
-// LegacyConfigMapOverrideDefined checks to see if the legacy ConfigMap override exists with
-// overrides defined. If it exists and has config overrides defined, this will return true. If it
-// does not exist or exists without config overrides set, it will return false.
-func LegacyConfigMapOverrideDefined(context *clusterd.Context, namespace string) (bool, error) {
-	cm, err := context.Clientset.CoreV1().ConfigMaps(namespace).
-		Get(k8sutil.ConfigOverrideName, metav1.GetOptions{})
-	if err != nil {
-		if errors.IsNotFound(err) {
-			logger.Debugf("override configmap does not exist")
-			return false, nil
-		}
-		// if we can't determine existence or not, assume it exists
-		return true, fmt.Errorf("failed to see if legacy override ConfigMap exists. %+v", err)
-	}
-
-	o, ok := cm.Data[k8sutil.ConfigOverrideVal]
-	if !ok || (ok && o == "") {
-		// if no override set or override is empty string
-		return false, nil
-	}
-
-	return true, nil
-}
-
-func configFileTextToOverrides(txt string) (rookceph.ConfigOverridesSpec, error) {
-	overrides := []rookceph.ConfigOverride{}
-
-	f, err := configFileTxtToINI(txt)
-	if err != nil {
-		return nil, err
-	}
-	sections := f.Sections()
-	for _, s := range sections {
-		who := s.Name()
-		keys := s.Keys()
-		for _, k := range keys {
-			option := k.Name()
-			value := k.String()
-			overrides = append(overrides, configOverride(who, option, value))
-		}
-	}
-
-	return overrides, nil
-}
-
-func configFileTxtToINI(txt string) (*ini.File, error) {
-	f := ini.Empty()
-	ovrTxt := []byte(txt)
-	if err := f.Append(ovrTxt); err != nil {
-		return nil, fmt.Errorf("failed to interpret text as INI file '''%s'''. %+v", ovrTxt, err)
-	}
-	return f, nil
-}
-
-func getOverrideConfigFromConfigMap(context *clusterd.Context, namespace string) string {
-	cm, err := context.Clientset.CoreV1().ConfigMaps(namespace).
-		Get(k8sutil.ConfigOverrideName, metav1.GetOptions{})
-	if err != nil {
-		if errors.IsNotFound(err) {
-			logger.Debugf("override configmap does not exist")
-			return "" // return empty string if cannot get override text
-		}
-		logger.Warningf("Error getting override configmap. %+v", err)
-		return "" // return empty string if cannot get override text
-	}
-	// the override config map exists
-	o, ok := cm.Data[k8sutil.ConfigOverrideVal]
-	if !ok {
-		logger.Warningf("The override config map does not have override text defined. %+v", o)
-		return "" // return empty string if cannot get override text
-	}
-	return o
-}

pkg/operator/ceph/file/controller.go

@@ -27,6 +27,7 @@ import (
 	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"
+	cephspec "github.com/rook/rook/pkg/operator/ceph/spec"
 	apiextensionsv1beta1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/tools/cache"
@@ -105,6 +106,16 @@ func (c *FilesystemController) onAdd(obj interface{}) {
 	c.acquireOrchestrationLock()
 	defer c.releaseOrchestrationLock()
 
+	if c.clusterSpec.External.Enable {
+		_, err := cephspec.ValidateCephVersionsBetweenLocalAndExternalClusters(c.context, c.namespace, c.clusterInfo.CephVersion)
+		if err != nil {
+			// This handles the case where the operator is running, the external cluster has been upgraded and a CR creation is called
+			// If that's a major version upgrade we fail, if it's a minor version, we continue, it's not ideal but not critical
+			logger.Errorf("refusing to run new crd. %+v", err)
+			return
+		}
+	}
+
 	err = createFilesystem(c.clusterInfo, c.context, *filesystem, c.rookVersion, c.clusterSpec, c.filesystemOwners(filesystem), c.clusterSpec.DataDirHostPath, c.isUpgrade)
 	if err != nil {
 		logger.Errorf("failed to create filesystem %s: %+v", filesystem.Name, err)

pkg/operator/ceph/object/controller.go

@@ -27,6 +27,7 @@ import (
 	"github.com/rook/rook/pkg/clusterd"
 	daemonconfig "github.com/rook/rook/pkg/daemon/ceph/config"
 	cephconfig "github.com/rook/rook/pkg/operator/ceph/config"
+	cephspec "github.com/rook/rook/pkg/operator/ceph/spec"
 	apiextensionsv1beta1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/tools/cache"
@@ -104,6 +105,15 @@ func (c *ObjectStoreController) onAdd(obj interface{}) {
 	c.acquireOrchestrationLock()
 	defer c.releaseOrchestrationLock()
 
+	if c.clusterSpec.External.Enable {
+		_, err := cephspec.ValidateCephVersionsBetweenLocalAndExternalClusters(c.context, c.namespace, c.clusterInfo.CephVersion)
+		if err != nil {
+			// This handles the case where the operator is running, the external cluster has been upgraded and a CR creation is called
+			// If that's a major version upgrade we fail, if it's a minor version, we continue, it's not ideal but not critical
+			logger.Errorf("refusing to run new crd. %+v", err)
+			return
+		}
+	}
 	c.createOrUpdateStore(objectstore)
 }
 

pkg/operator/ceph/spec/version.go

@@ -0,0 +1,37 @@
+/*
+Copyright 2018 The Rook Authors. All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+	http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package spec
+
+import (
+	"fmt"
+
+	"github.com/rook/rook/pkg/clusterd"
+	"github.com/rook/rook/pkg/daemon/ceph/client"
+	cephver "github.com/rook/rook/pkg/operator/ceph/version"
+)
+
+// ValidateCephVersionsBetweenLocalAndExternalClusters makes sure an external cluster can be connected
+// by checking the external ceph versions available and comparing it with the local image provided
+func ValidateCephVersionsBetweenLocalAndExternalClusters(context *clusterd.Context, namespace string, localVersion cephver.CephVersion) (cephver.CephVersion, error) {
+	// health check should tell us if the external cluster has been upgraded and display a message
+	externalVersion, err := client.GetCephMonVersion(context, namespace)
+	if err != nil {
+		return cephver.CephVersion{}, fmt.Errorf("failed to get ceph mon version. %+v", err)
+	}
+
+	return *externalVersion, cephver.ValidateCephVersionsBetweenLocalAndExternalClusters(localVersion, *externalVersion)
+}

pkg/operator/ceph/version/version.go

@@ -20,6 +20,8 @@ import (
 	"fmt"
 	"regexp"
 	"strconv"
+
+	"github.com/coreos/pkg/capnslog"
 )
 
 // CephVersion represents the Ceph version format
@@ -51,6 +53,8 @@ var (
 
 	// for parsing the output of `ceph --version`
 	versionPattern = regexp.MustCompile(`ceph version (\d+)\.(\d+)\.(\d+)`)
+
+	logger = capnslog.NewPackageLogger("github.com/rook/rook", "cephver")
 )
 
 func (v *CephVersion) String() string {
@@ -214,3 +218,40 @@ func IsInferior(a, b CephVersion) bool {
 
 	return false
 }
+
+// ValidateCephVersionsBetweenLocalAndExternalClusters makes sure an external cluster can be connected
+// by checking the external ceph versions available and comparing it with the local image provided
+func ValidateCephVersionsBetweenLocalAndExternalClusters(localVersion, externalVersion CephVersion) error {
+	logger.Debugf("local version is %s, external version is %s", localVersion.String(), externalVersion.String())
+
+	// We only support Nautilus or newer
+	if !externalVersion.IsAtLeastNautilus() {
+		return fmt.Errorf("unsupported ceph version %s, need at least nautilus, delete your cluster CR and create a new one with a correct ceph version", externalVersion.String())
+	}
+
+	// Identical version, regardless if other CRs are running, it's ok!
+	if IsIdentical(localVersion, externalVersion) {
+		return nil
+	}
+
+	// Local version must never be higher than the external one
+	if IsSuperior(localVersion, externalVersion) {
+		return fmt.Errorf("local cluster ceph version is higher %s than the external cluster %s, this must never happen", externalVersion.String(), localVersion.String())
+	}
+
+	// External cluster was updated to a minor version higher, consider updating too!
+	if localVersion.Major == externalVersion.Major {
+		if IsSuperior(externalVersion, localVersion) {
+			logger.Warningf("external cluster ceph version is a minor version higher %s than the local cluster %s, consider upgrading", externalVersion.String(), localVersion.String())
+			return nil
+		}
+	}
+
+	// The external cluster was upgraded, consider upgrading too!
+	if localVersion.Major < externalVersion.Major {
+		logger.Errorf("external cluster ceph version is a major version higher %s than the local cluster %s, consider upgrading", externalVersion.String(), localVersion.String())
+		return nil
+	}
+
+	return nil
+}

pkg/operator/ceph/version/version_test.go

@@ -175,3 +175,36 @@ func TestIsInferior(t *testing.T) {
 	assert.True(t, IsInferior(CephVersion{15, 1, 3}, CephVersion{15, 2, 2}))
 	assert.True(t, IsInferior(CephVersion{15, 2, 1}, CephVersion{15, 2, 2}))
 }
+
+func TestValidateCephVersionsBetweenLocalAndExternalClusters(t *testing.T) {
+	// TEST 1: versions are identical
+	localCephVersion := CephVersion{Major: 14, Minor: 2, Extra: 1}
+	externalCephVersion := CephVersion{Major: 14, Minor: 2, Extra: 1}
+	err := ValidateCephVersionsBetweenLocalAndExternalClusters(localCephVersion, externalCephVersion)
+	assert.NoError(t, err)
+
+	// TEST 2: local cluster version major is lower than external cluster version
+	localCephVersion = CephVersion{Major: 14, Minor: 2, Extra: 1}
+	externalCephVersion = CephVersion{Major: 15, Minor: 2, Extra: 1}
+	err = ValidateCephVersionsBetweenLocalAndExternalClusters(localCephVersion, externalCephVersion)
+	assert.NoError(t, err)
+
+	// TEST 3: local cluster version major is higher than external cluster version
+	localCephVersion = CephVersion{Major: 15, Minor: 2, Extra: 1}
+	externalCephVersion = CephVersion{Major: 14, Minor: 2, Extra: 1}
+	err = ValidateCephVersionsBetweenLocalAndExternalClusters(localCephVersion, externalCephVersion)
+	assert.Error(t, err)
+
+	// TEST 4: local version is > but from a minor release
+	// local version must never be higher
+	localCephVersion = CephVersion{Major: 14, Minor: 2, Extra: 2}
+	externalCephVersion = CephVersion{Major: 14, Minor: 2, Extra: 1}
+	err = ValidateCephVersionsBetweenLocalAndExternalClusters(localCephVersion, externalCephVersion)
+	assert.Error(t, err)
+
+	// TEST 5: external version is > but from a minor release
+	localCephVersion = CephVersion{Major: 14, Minor: 2, Extra: 1}
+	externalCephVersion = CephVersion{Major: 14, Minor: 2, Extra: 2}
+	err = ValidateCephVersionsBetweenLocalAndExternalClusters(localCephVersion, externalCephVersion)
+	assert.NoError(t, err)
+}