ceph: relax pre-requisite for external cluster

We now differentiate the cases where:

* we only consume the external cluster
* we consume the external cluster as well as creating stateless
resources in Kubernetes (bootstrap mds,rgw, nfs)

This is mostly controlled via the image property spec. If not defined,
not extra CRs won't be able to be created.

Now the external cluster feature supports Ceph cluster as of Luminous 12.2.

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

Documentation/ceph-cluster-crd.md

@@ -89,7 +89,7 @@ Settings can be specified at the global level to apply to the cluster as a whole
 ### Cluster Settings
 
 - `external`:
-  - `enable`: if `true`, the cluster will not be managed by Rook but via an external entity. This mode is intended to connect to an existing cluster. In this case, Rook will only consume the external cluster. However, Rook will be able to deploy various daemons in Kubernetes such as object gateways, mds and nfs. If this setting is enabled **all** the other options will be ignored except `cephVersion.image` and `dataDirHostPath`. See [external cluster configuration](#external-cluster).
+  - `enable`: if `true`, the cluster will not be managed by Rook but via an external entity. This mode is intended to connect to an existing cluster. In this case, Rook will only consume the external cluster. However, Rook will be able to deploy various daemons in Kubernetes such as object gateways, mds and nfs if an image is provided and will refuse otherwise. If this setting is enabled **all** the other options will be ignored except `cephVersion.image` and `dataDirHostPath`. See [external cluster configuration](#external-cluster). If `cephVersion.image` is left blank, Rook will refuse the creation of extra CRs like object, file and nfs.
 - `cephVersion`: The version information for launching the ceph daemons.
   - `image`: The image used for running the ceph daemons. For example, `ceph/ceph:v13.2.6-20190604` or `ceph/ceph:v14.2.4-20190917`.
   For the latest ceph images, see the [Ceph DockerHub](https://hub.docker.com/r/ceph/ceph/tags/).
@@ -672,6 +672,18 @@ spec:
 
 ### External cluster
 
+**The minimum supported Ceph version for the External Cluster is Luminous 12.2.x.**
+
+The features available from the external cluster will vary depending on the version of Ceph. The following table shows the minimum version of Ceph for some of the features:
+
+| FEATURE                                      | CEPH VERSION |
+|----------------------------------------------|--------------|
+| Dynamic provisioning RBD                     | 12.2.X       |
+| Configure extra CRDs (object, file, nfs)[^1] | 13.2.3       |
+| Dynamic provisioning CephFS                  | 14.2.3       |
+
+[^1]: Configure an object store, shared file system, or NFS resources in the local cluster to connect to the external Ceph cluster
+
 #### Pre-requisites
 
 In order to configure an external Ceph cluster with Rook, we need to inject some information in order to connect to that cluster.
@@ -712,12 +724,13 @@ spec:
   external:
     enable: true
   dataDirHostPath: /var/lib/rook
+  # providing an image is optional, do this if you want to create other CRs (rgw, mds, nfs)
   cephVersion:
-    image: ceph/ceph:v14.2.4-20190917 # the image version **must** match the version of the external Ceph cluster
+    image: ceph/ceph:v14.2.4-20190917 # MUST match external cluster version
 ```
 
 Choose the namespace carefully, if you have an existing cluster managed by Rook, you have likely already injected `common.yaml`.
-Additionnally, you now need to inject `common-external.yaml` too.
+Additionally, you now need to inject `common-external.yaml` too.
 
 You can now create it like this:
 

cluster/examples/kubernetes/ceph/cluster-external.yaml

@@ -19,5 +19,6 @@ spec:
   external:
     enable: true
   dataDirHostPath: /var/lib/rook
+  # providing an image is optional, do this if you want to create other CRs (rgw, mds, nfs)
   cephVersion:
     image: ceph/ceph:v14.2.4-20190917 # MUST match external cluster version

pkg/daemon/ceph/client/mon.go

@@ -47,6 +47,7 @@ type MonMapEntry struct {
 	Name        string `json:"name"`
 	Rank        int    `json:"rank"`
 	Address     string `json:"addr"`
+	PublicAddr  string `json:"public_addr"`
 	PublicAddrs struct {
 		Addrvec []AddrvecEntry `json:"addrvec"`
 	} `json:"public_addrs"`

pkg/daemon/ceph/config/config.go

@@ -213,7 +213,7 @@ func CreateDefaultCephConfig(context *clusterd.Context, cluster *ClusterInfo) (*
 		// The operator always fails this test since it does not have the env var 'ROOK_CEPH_VERSION'
 		podName := os.Getenv("POD_NAME")
 		if cluster.CephVersion.IsAtLeastNautilus() {
-			monHosts[i] = "v1:" + msgr1Endpoint
+			monHosts[i] = msgr1Prefix + msgr1Endpoint
 		} else if podName != "" && strings.Contains(podName, "operator") {
 			// This is an operator and its version is always based on Nautilus
 			// so it knows how to parse both msgr1 and msgr2 syntax

pkg/operator/ceph/cluster/cluster.go

@@ -126,7 +126,6 @@ func (c *cluster) detectCephVersion(rookImage, cephImage string, timeout time.Du
 }
 
 func (c *cluster) validateCephVersion(version *cephver.CephVersion) error {
-	// 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())
@@ -139,6 +138,7 @@ func (c *cluster) validateCephVersion(version *cephver.CephVersion) error {
 			}
 		}
 	}
+
 	// 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
@@ -160,13 +160,20 @@ func (c *cluster) validateCephVersion(version *cephver.CephVersion) error {
 		return nil
 	}
 
-	if c.Spec.External.Enable {
+	if c.Spec.External.Enable && c.Spec.CephVersion.Image != "" {
 		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)
 		}
 	}
 
+	// On external cluster setup, if we don't bootstrap any resources in the Kubernetes cluster then
+	// there is no need to validate the Ceph image further
+	if c.Spec.External.Enable && c.Spec.CephVersion.Image == "" {
+		logger.Debug("no spec image specified on external cluster, not validating Ceph version.")
+		return nil
+	}
+
 	// Get cluster running versions
 	versions, err := client.GetAllCephDaemonVersions(c.context, c.Namespace)
 	if err != nil {

pkg/operator/ceph/cluster/controller.go

@@ -267,29 +267,30 @@ func (c *ClusterController) configureExternalCephCluster(namespace, name string,
 	// then populate clusterInfo
 	cluster.Info = populateExternalClusterInfo(c.context, namespace)
 
+	// Write connection info (ceph config file and keyring) for ceph commands
+	if cluster.Spec.CephVersion.Image == "" {
+		err = mon.WriteConnectionConfig(c.context, cluster.Info)
+		if err != nil {
+			logger.Errorf("failed to write config. Attempting to continue. %+v", err)
+		}
+	}
+
 	// 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)
-	}
+	// If no image is specified we don't perform any checks
+	if cluster.Spec.CephVersion.Image != "" {
+		_, _, err = c.detectAndValidateCephVersion(cluster, cluster.Spec.CephVersion.Image)
+		if err != nil {
+			return fmt.Errorf("failed to detect and validate ceph version. %+v", err)
+		}
 
-	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
-	err = config.GetStore(cluster.context, namespace, &cluster.ownerRef).CreateOrUpdate(cluster.Info)
-	if err != nil {
-		return err
-	}
-
-	// Apply CRD ConfigOverrideName to the external cluster
-	err = config.SetDefaultConfigs(cluster.context, namespace, cluster.Info)
-	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)
+		// 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
+		//
+		// Only do this when doing a bit of management...
+		err = config.GetStore(cluster.context, namespace, &cluster.ownerRef).CreateOrUpdate(cluster.Info)
+		if err != nil {
+			return err
+		}
 	}
 
 	// The cluster Identity must be established at this point
@@ -1042,9 +1043,6 @@ func validateExternalClusterSpec(cluster *cluster) error {
 	if cluster.Spec.DataDirHostPath == "" {
 		return fmt.Errorf("dataDirHostPath must be specified")
 	}
-	if cluster.Spec.CephVersion.Image == "" {
-		return fmt.Errorf("spec image must be specified")
-	}
 
 	return nil
 }

pkg/operator/ceph/cluster/controller_test.go

@@ -243,7 +243,7 @@ func TestValidateExternalClusterSpec(t *testing.T) {
 
 	c.Spec.DataDirHostPath = "path"
 	err = validateExternalClusterSpec(c)
-	assert.Error(t, err)
+	assert.NoError(t, err, err)
 
 	c.Spec.CephVersion.Image = "ceph/ceph:v14.2.4-20190917"
 	err = validateExternalClusterSpec(c)

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

@@ -18,6 +18,7 @@ package mon
 
 import (
 	"fmt"
+	"strings"
 	"time"
 
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
@@ -297,9 +298,12 @@ 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)
+	// We don't need to validate Ceph version if no image is present
+	if c.spec.CephVersion.Image == "" {
+		_, 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)
@@ -318,7 +322,6 @@ func (c *Cluster) handleExternalMonStatus(status client.MonStatusResponse) error
 }
 
 func (c *Cluster) addOrRemoveExternalMonitor(status client.MonStatusResponse) (bool, error) {
-	// func addOrRemoveExternalMonitor(status client.MonStatusResponse) (bool, error) {
 	var changed bool
 	// Populate a map with the current monitor from ClusterInfo
 	monsNotFound := map[string]interface{}{}
@@ -343,14 +346,11 @@ func (c *Cluster) addOrRemoveExternalMonitor(status client.MonStatusResponse) (b
 			// If the mon is part of the quorum
 			if inQuorum {
 				// let's add it to ClusterInfo
-				// Always pick the v1 endpoint, that's the easier thing to do since some people might not have activated msgr2
-				// We must run on at least Nautilus so we are 100% certain that will have a field mon.PublicAddrs.Addrvec with 'v1' in it
-				var endpoint string
-				for i := range mon.PublicAddrs.Addrvec {
-					if mon.PublicAddrs.Addrvec[i].Type == "v1" {
-						endpoint = mon.PublicAddrs.Addrvec[i].Addr
-					}
-				}
+				// FYI mon.PublicAddr is "10.97.171.131:6789/0"
+				// so we need to remove '/0'
+				endpointSlash := strings.Split(mon.PublicAddr, "/")
+				endpoint := endpointSlash[0]
+
 				// find IP and Port of that Mon
 				monIP := cephutil.GetIPFromEndpoint(endpoint)
 				monPort := cephutil.GetPortFromEndpoint(endpoint)

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

@@ -229,17 +229,13 @@ func TestAddOrRemoveExternalMonitor(t *testing.T) {
 
 	// populate fake monmap
 	fakeResp := client.MonStatusResponse{Quorum: []int{0}}
-	fakeAddrvecEntry := []client.AddrvecEntry{
-		{
-			Addr: "172.17.0.4:3300",
-		},
-	}
+
 	fakeResp.MonMap.Mons = []client.MonMapEntry{
 		{
 			Name: "a",
 		},
 	}
-	fakeResp.MonMap.Mons[0].PublicAddrs.Addrvec = fakeAddrvecEntry
+	fakeResp.MonMap.Mons[0].PublicAddr = "172.17.0.4:3300"
 
 	// populate fake ClusterInfo
 	c := &Cluster{ClusterInfo: &cephconfig.ClusterInfo{}}
@@ -280,12 +276,7 @@ func TestAddOrRemoveExternalMonitor(t *testing.T) {
 			Name: "b",
 		},
 	}
-	fakeAddrvecEntry2 := []client.AddrvecEntry{
-		{
-			Addr: "172.17.0.5:3300",
-		},
-	}
-	fakeResp.MonMap.Mons[1].PublicAddrs.Addrvec = fakeAddrvecEntry2
+	fakeResp.MonMap.Mons[1].PublicAddr = "172.17.0.5:3300"
 	c.ClusterInfo = test.CreateConfigDir(1)
 	changed, err = c.addOrRemoveExternalMonitor(fakeResp)
 	assert.NoError(t, err)
@@ -299,18 +290,14 @@ func TestAddOrRemoveExternalMonitor(t *testing.T) {
 	// Now let's test the case where the mon is in clusterInfo, part of the monmap but not in quorum!
 	c.ClusterInfo = test.CreateConfigDir(1)
 	fakeResp2 := client.MonStatusResponse{Quorum: []int{1}} // quorum is owned by the mon with the rank 1 and our mon rank is 0
-	fakeAddrvecEntry3 := []client.AddrvecEntry{
-		{
-			Addr: "172.17.0.4:3300",
-		},
-	}
+
 	fakeResp2.MonMap.Mons = []client.MonMapEntry{
 		{
 			Name: "a",
 			Rank: 0,
 		},
 	}
-	fakeResp2.MonMap.Mons[0].PublicAddrs.Addrvec = fakeAddrvecEntry3
+	fakeResp2.MonMap.Mons[0].PublicAddr = "172.17.0.4:3300"
 	changed, err = c.addOrRemoveExternalMonitor(fakeResp2)
 	assert.NoError(t, err)
 	assert.True(t, changed)

pkg/operator/ceph/file/controller.go

@@ -97,6 +97,12 @@ func (c *FilesystemController) StartWatch(namespace string, stopCh chan struct{}
 }
 
 func (c *FilesystemController) onAdd(obj interface{}) {
+
+	if c.clusterSpec.External.Enable && c.clusterSpec.CephVersion.Image == "" {
+		logger.Warningf("Creating filesystems for an external ceph cluster is disabled because no Ceph image is specified")
+		return
+	}
+
 	filesystem, err := getFilesystemObject(obj)
 	if err != nil {
 		logger.Errorf("failed to get filesystem object: %+v", err)
@@ -123,6 +129,11 @@ func (c *FilesystemController) onAdd(obj interface{}) {
 }
 
 func (c *FilesystemController) onUpdate(oldObj, newObj interface{}) {
+	if c.clusterSpec.External.Enable && c.clusterSpec.CephVersion.Image == "" {
+		logger.Warningf("Updating filesystems for an external ceph cluster is disabled because no Ceph image is specified")
+		return
+	}
+
 	oldFS, err := getFilesystemObject(oldObj)
 	if err != nil {
 		logger.Errorf("failed to get old filesystem object: %+v", err)
@@ -182,6 +193,11 @@ func (c *FilesystemController) ParentClusterChanged(cluster cephv1.ClusterSpec,
 }
 
 func (c *FilesystemController) onDelete(obj interface{}) {
+	if c.clusterSpec.External.Enable && c.clusterSpec.CephVersion.Image == "" {
+		logger.Warningf("Deleting filesystems for an external ceph cluster is disabled because no Ceph image is specified")
+		return
+	}
+
 	filesystem, err := getFilesystemObject(obj)
 	if err != nil {
 		logger.Errorf("failed to get filesystem object: %+v", err)

pkg/operator/ceph/nfs/controller.go

@@ -85,6 +85,11 @@ func (c *CephNFSController) StartWatch(namespace string, stopCh chan struct{}) e
 }
 
 func (c *CephNFSController) onAdd(obj interface{}) {
+	if c.clusterSpec.External.Enable && c.clusterSpec.CephVersion.Image == "" {
+		logger.Warningf("Creating nfs for an external ceph cluster is disabled because no Ceph image is specified")
+		return
+	}
+
 	nfs := obj.(*cephv1.CephNFS).DeepCopy()
 	if !c.clusterInfo.CephVersion.IsAtLeastNautilus() {
 		logger.Errorf("Ceph NFS is only supported with Nautilus or newer. CRD %s will be ignored.", nfs.Name)
@@ -101,6 +106,11 @@ func (c *CephNFSController) onAdd(obj interface{}) {
 }
 
 func (c *CephNFSController) onUpdate(oldObj, newObj interface{}) {
+	if c.clusterSpec.External.Enable && c.clusterSpec.CephVersion.Image == "" {
+		logger.Warningf("Updating nfs for an external ceph cluster is disabled because no Ceph image is specified")
+		return
+	}
+
 	oldNFS := oldObj.(*cephv1.CephNFS).DeepCopy()
 	newNFS := newObj.(*cephv1.CephNFS).DeepCopy()
 	if !c.clusterInfo.CephVersion.IsAtLeastNautilus() {
@@ -132,6 +142,11 @@ func (c *CephNFSController) onUpdate(oldObj, newObj interface{}) {
 }
 
 func (c *CephNFSController) onDelete(obj interface{}) {
+	if c.clusterSpec.External.Enable && c.clusterSpec.CephVersion.Image == "" {
+		logger.Warningf("Deleting nfs for an external ceph cluster is disabled because no Ceph image is specified")
+		return
+	}
+
 	nfs := obj.(*cephv1.CephNFS).DeepCopy()
 	if !c.clusterInfo.CephVersion.IsAtLeastNautilus() {
 		logger.Errorf("Ceph NFS is only supported with Nautilus or newer. CRD %s cleanup will be ignored.", nfs.Name)

pkg/operator/ceph/object/controller.go

@@ -96,6 +96,11 @@ func (c *ObjectStoreController) StartWatch(namespace string, stopCh chan struct{
 }
 
 func (c *ObjectStoreController) onAdd(obj interface{}) {
+	if c.clusterSpec.External.Enable && c.clusterSpec.CephVersion.Image == "" {
+		logger.Warningf("Creating object store for an external ceph cluster is disabled because no Ceph image is specified")
+		return
+	}
+
 	objectstore, err := getObjectStoreObject(obj)
 	if err != nil {
 		logger.Errorf("failed to get objectstore object: %+v", err)
@@ -118,6 +123,11 @@ func (c *ObjectStoreController) onAdd(obj interface{}) {
 }
 
 func (c *ObjectStoreController) onUpdate(oldObj, newObj interface{}) {
+	if c.clusterSpec.External.Enable && c.clusterSpec.CephVersion.Image == "" {
+		logger.Warningf("Updating object store for an external ceph cluster is disabled because no Ceph image is specified")
+		return
+	}
+
 	// if the object store spec is modified, update the object store
 	oldStore, err := getObjectStoreObject(oldObj)
 	if err != nil {
@@ -159,6 +169,11 @@ func (c *ObjectStoreController) createOrUpdateStore(objectstore *cephv1.CephObje
 }
 
 func (c *ObjectStoreController) onDelete(obj interface{}) {
+	if c.clusterSpec.External.Enable && c.clusterSpec.CephVersion.Image == "" {
+		logger.Warningf("Deleting object store for an external ceph cluster is disabled because no Ceph image is specified")
+		return
+	}
+
 	objectstore, err := getObjectStoreObject(obj)
 	if err != nil {
 		logger.Errorf("failed to get objectstore object: %+v", err)

pkg/operator/ceph/object/user/controller.go

@@ -89,6 +89,11 @@ func (c *ObjectStoreUserController) StartWatch(stopCh chan struct{}) error {
 }
 
 func (c *ObjectStoreUserController) onAdd(obj interface{}) {
+	if c.clusterSpec.External.Enable && c.clusterSpec.CephVersion.Image == "" {
+		logger.Warningf("Creating object store user for an external ceph cluster is disabled because no Ceph image is specified")
+		return
+	}
+
 	user, err := getObjectStoreUserObject(obj)
 	if err != nil {
 		logger.Errorf("failed to get objectstoreuser object: %+v", err)
@@ -101,10 +106,19 @@ func (c *ObjectStoreUserController) onAdd(obj interface{}) {
 }
 
 func (c *ObjectStoreUserController) onUpdate(oldObj, newObj interface{}) {
+	if c.clusterSpec.External.Enable && c.clusterSpec.CephVersion.Image == "" {
+		logger.Warningf("Updating object store user for an external ceph cluster is disabled because no Ceph image is specified")
+		return
+	}
 	// TODO: Add update code here after features are added which require updates.
 }
 
 func (c *ObjectStoreUserController) onDelete(obj interface{}) {
+	if c.clusterSpec.External.Enable && c.clusterSpec.CephVersion.Image == "" {
+		logger.Warningf("Deleting object store user for an external ceph cluster is disabled because no Ceph image is specified")
+		return
+	}
+
 	user, err := getObjectStoreUserObject(obj)
 	if err != nil {
 		logger.Errorf("failed to get objectstoreuser object: %+v", err)

pkg/operator/ceph/pool/controller.go

@@ -80,6 +80,11 @@ func (c *PoolController) StartWatch(namespace string, stopCh chan struct{}) erro
 }
 
 func (c *PoolController) onAdd(obj interface{}) {
+	if c.clusterSpec.External.Enable && c.clusterSpec.CephVersion.Image == "" {
+		logger.Warningf("Creating pools for an external ceph cluster is disabled because no Ceph image is specified")
+		return
+	}
+
 	pool, err := getPoolObject(obj)
 	if err != nil {
 		logger.Errorf("failed to get pool object: %+v", err)
@@ -93,6 +98,11 @@ func (c *PoolController) onAdd(obj interface{}) {
 }
 
 func (c *PoolController) onUpdate(oldObj, newObj interface{}) {
+	if c.clusterSpec.External.Enable && c.clusterSpec.CephVersion.Image == "" {
+		logger.Warningf("Updating pools for an external ceph cluster is disabled because no Ceph image is specified")
+		return
+	}
+
 	oldPool, err := getPoolObject(oldObj)
 	if err != nil {
 		logger.Errorf("failed to get old pool object: %+v", err)
@@ -138,6 +148,11 @@ func poolChanged(old, new cephv1.PoolSpec) bool {
 }
 
 func (c *PoolController) onDelete(obj interface{}) {
+	if c.clusterSpec.External.Enable && c.clusterSpec.CephVersion.Image == "" {
+		logger.Warningf("Deleting pools for an external ceph cluster is disabled because no Ceph image is specified")
+		return
+	}
+
 	pool, err := getPoolObject(obj)
 	if err != nil {
 		logger.Errorf("failed to get pool object: %+v", err)

pkg/operator/ceph/version/version.go

@@ -38,6 +38,8 @@ const (
 var (
 	// Minimum supported version is 13.2.4 where ceph-volume is supported
 	Minimum = CephVersion{13, 2, 4}
+	// Luminous Ceph version
+	Luminous = CephVersion{12, 0, 0}
 	// Mimic Ceph version
 	Mimic = CephVersion{13, 0, 0}
 	// Nautilus Ceph version
@@ -224,9 +226,9 @@ func IsInferior(a, b CephVersion) bool {
 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())
+	// We only support Luminous or newer
+	if !externalVersion.IsAtLeast(Luminous) {
+		return fmt.Errorf("unsupported external ceph version %s, need at least luminous", externalVersion.String())
 	}
 
 	// Identical version, regardless if other CRs are running, it's ok!