Ceph: Added version check for ceph osd safe-to-destroy command

It's not 100% safe since before Nautilus osd safe-to-destroy command
didn't have any json output. So added a version check.

Signed-off-by: rohan47 <rohgupta@redhat.com>

pkg/daemon/ceph/client/osd.go

@@ -22,6 +22,7 @@ import (
 
 	"github.com/pkg/errors"
 	"github.com/rook/rook/pkg/clusterd"
+	cephver "github.com/rook/rook/pkg/operator/ceph/version"
 )
 
 type OSDUsage struct {
@@ -259,7 +260,11 @@ func OSDRemove(context *clusterd.Context, clusterName string, osdID int) (string
 	return string(buf), err
 }
 
-func OsdSafeToDestroy(context *clusterd.Context, clusterName string, osdID int) (bool, error) {
+func OsdSafeToDestroy(context *clusterd.Context, clusterName string, osdID int, cephVersion cephver.CephVersion) (bool, error) {
+	if !cephVersion.IsAtLeastNautilus() {
+		logger.Debugf("failed to get safe-to-destroy status: ceph version in lower than Nautilus")
+		return false, nil
+	}
 	args := []string{"osd", "safe-to-destroy", strconv.Itoa(osdID)}
 	cmd := NewCephCommand(context, clusterName, args)
 	buf, err := cmd.Run()

pkg/operator/ceph/cluster/controller.go

@@ -463,7 +463,7 @@ func (c *ClusterController) initializeCluster(cluster *cluster, clusterObj *ceph
 
 	if !cluster.Spec.External.Enable {
 		// Start the osd health checker only if running OSDs in the local ceph cluster
-		c.osdChecker = osd.NewMonitor(c.context, cluster.Namespace, cluster.Spec.RemoveOSDsIfOutAndSafeToRemove)
+		c.osdChecker = osd.NewMonitor(c.context, cluster.Namespace, cluster.Spec.RemoveOSDsIfOutAndSafeToRemove, cluster.Info.CephVersion)
 		go c.osdChecker.Start(cluster.stopCh)
 	}
 

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

@@ -23,6 +23,7 @@ import (
 	"github.com/pkg/errors"
 	"github.com/rook/rook/pkg/clusterd"
 	"github.com/rook/rook/pkg/daemon/ceph/client"
+	cephver "github.com/rook/rook/pkg/operator/ceph/version"
 	"github.com/rook/rook/pkg/operator/k8sutil"
 	kerrors "k8s.io/apimachinery/pkg/api/errors"
 )
@@ -42,11 +43,12 @@ type Monitor struct {
 	context                        *clusterd.Context
 	clusterName                    string
 	removeOSDsIfOUTAndSafeToRemove bool
+	cephVersion                    cephver.CephVersion
 }
 
 // NewMonitor instantiates OSD monitoring
-func NewMonitor(context *clusterd.Context, clusterName string, removeOSDsIfOUTAndSafeToRemove bool) *Monitor {
-	return &Monitor{context, clusterName, removeOSDsIfOUTAndSafeToRemove}
+func NewMonitor(context *clusterd.Context, clusterName string, removeOSDsIfOUTAndSafeToRemove bool, cephVersion cephver.CephVersion) *Monitor {
+	return &Monitor{context, clusterName, removeOSDsIfOUTAndSafeToRemove, cephVersion}
 }
 
 // Start runs monitoring logic for osds status at set intervals
@@ -125,7 +127,7 @@ func (m *Monitor) handleOSDMarkedOut(outOSDid int) error {
 		return errors.Wrapf(err, "failed to get osd deployment of osd id %d", outOSDid)
 	}
 	if len(dp.Items) != 0 {
-		safeToDestroyOSD, err := client.OsdSafeToDestroy(m.context, m.clusterName, outOSDid)
+		safeToDestroyOSD, err := client.OsdSafeToDestroy(m.context, m.clusterName, outOSDid, m.cephVersion)
 		if err != nil {
 			return errors.Wrapf(err, "failed to get osd deployment of osd id %d", outOSDid)
 		}

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

@@ -20,6 +20,7 @@ import (
 	"testing"
 
 	"github.com/rook/rook/pkg/clusterd"
+	cephver "github.com/rook/rook/pkg/operator/ceph/version"
 	"github.com/rook/rook/pkg/operator/k8sutil"
 	testexec "github.com/rook/rook/pkg/operator/test"
 	exectest "github.com/rook/rook/pkg/util/exec/test"
@@ -83,8 +84,12 @@ func TestOSDStatus(t *testing.T) {
 	dp, _ := context.Clientset.AppsV1().Deployments(cluster).List(metav1.ListOptions{LabelSelector: fmt.Sprintf("%v=%d", OsdIdLabelKey, 0)})
 	assert.Equal(t, 1, len(dp.Items))
 
+	cephVersion := cephver.CephVersion{
+		Major: 14,
+	}
+
 	// Initializing an OSD monitoring
-	osdMon := NewMonitor(context, cluster, true)
+	osdMon := NewMonitor(context, cluster, true, cephVersion)
 
 	// Run OSD monitoring routine
 	err := osdMon.osdStatus()
@@ -98,8 +103,12 @@ func TestOSDStatus(t *testing.T) {
 }
 
 func TestMonitorStart(t *testing.T) {
+	cephVersion := cephver.CephVersion{
+		Major: 14,
+	}
+
 	stopCh := make(chan struct{})
-	osdMon := NewMonitor(&clusterd.Context{}, "cluster", true)
+	osdMon := NewMonitor(&clusterd.Context{}, "cluster", true, cephVersion)
 	logger.Infof("starting osd monitor")
 	go osdMon.Start(stopCh)
 	close(stopCh)