osd: replace existing OSDs to use new store

This follow up the #12507
- fixes replacing of encrypted OSDs.
- Updates OSD status at the end of reconcile

Signed-off-by: sp98 <sapillai@redhat.com>
(cherry picked from commit 11b8d10a5a)

cmd/rook/ceph/osd.go

@@ -19,7 +19,6 @@ package ceph
 import (
 	"context"
 	"encoding/json"
-	"fmt"
 	"os"
 	"path"
 	"strconv"
@@ -30,9 +29,6 @@ import (
 	"github.com/pkg/errors"
 	"github.com/rook/rook/cmd/rook/rook"
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
-	"github.com/rook/rook/pkg/clusterd"
-	cleanup "github.com/rook/rook/pkg/daemon/ceph/cleanup"
-	"github.com/rook/rook/pkg/daemon/ceph/client"
 	osddaemon "github.com/rook/rook/pkg/daemon/ceph/osd"
 	"github.com/rook/rook/pkg/operator/ceph/cluster/mon"
 	"github.com/rook/rook/pkg/operator/ceph/cluster/osd"
@@ -263,11 +259,11 @@ func prepareOSD(cmd *cobra.Command, args []string) error {
 	// destroy the OSD using the OSD ID
 	var replaceOSD *osd.OSDReplaceInfo
 	if replaceOSDID != -1 {
-		osdInfo, err := destroyOSD(context, &clusterInfo, replaceOSDID)
+		logger.Infof("destroying osd.%d and cleaning its backing device", replaceOSDID)
+		replaceOSD, err = osddaemon.DestroyOSD(context, &clusterInfo, replaceOSDID, cfg.pvcBacked, cfg.storeConfig.EncryptedDevice)
 		if err != nil {
-			rook.TerminateFatal(errors.Wrapf(err, "failed to destroy OSD %d.", osdInfo.ID))
+			rook.TerminateFatal(errors.Wrapf(err, "failed to destroy OSD %d.", replaceOSDID))
 		}
-		replaceOSD = &oposd.OSDReplaceInfo{ID: osdInfo.ID, Path: osdInfo.BlockPath}
 	}
 
 	agent := osddaemon.NewAgent(context, dataDevices, cfg.metadataDevice, forceFormat,
@@ -417,36 +413,3 @@ func readCephSecret(path string) error {
 	}
 	return nil
 }
-
-func destroyOSD(context *clusterd.Context, clusterInfo *client.ClusterInfo, osdID int) (*oposd.OSDInfo, error) {
-	osdInfo, err := osddaemon.GetOSDInfoById(context, clusterInfo, osdID)
-	if err != nil {
-		return nil, errors.Wrapf(err, "failed to get OSD info for OSD.%d", osdID)
-	}
-
-	// destroy the osd
-	logger.Infof("destroying OSD %d on path %q in %q mode", osdInfo.ID, osdInfo.BlockPath, osdInfo.CVMode)
-	destroyOSDArgs := []string{"osd", "destroy", fmt.Sprintf("osd.%d", osdInfo.ID), "--yes-i-really-mean-it"}
-	_, err = client.NewCephCommand(context, clusterInfo, destroyOSDArgs).Run()
-	if err != nil {
-		return nil, errors.Wrapf(err, "failed to destroy osd.%d.", osdInfo.ID)
-	}
-
-	// Sanitize OSD disk
-	s := cleanup.NewDiskSanitizer(context, clusterInfo,
-		&cephv1.SanitizeDisksSpec{
-			Method:     cephv1.SanitizeMethodProperty(cephv1.SanitizeMethodComplete),
-			DataSource: cephv1.SanitizeDataSourceProperty(cephv1.SanitizeDataSourceZero),
-			Iteration:  1,
-		},
-	)
-
-	// TODO: handle disk sanitization errors
-	if osdInfo.CVMode == "raw" {
-		s.SanitizeRawDisk([]oposd.OSDInfo{osdInfo})
-	} else if osdInfo.CVMode == "lvm" {
-		s.SanitizeLVMDisk([]oposd.OSDInfo{osdInfo})
-	}
-
-	return &osdInfo, nil
-}

pkg/daemon/ceph/client/deviceclass.go

@@ -34,7 +34,7 @@ func GetDeviceClasses(context *clusterd.Context, clusterInfo *ClusterInfo) ([]st
 
 	var deviceclass []string
 	if err := json.Unmarshal(buf, &deviceclass); err != nil {
-		return nil, errors.Wrap(err, "failed to unmarshal osd crush class list response")
+		return nil, errors.Wrapf(err, "failed to unmarshal osd crush class list response %q", string(buf))
 	}
 
 	return deviceclass, nil

pkg/daemon/ceph/osd/daemon.go

@@ -32,6 +32,7 @@ import (
 	"github.com/pkg/errors"
 	"github.com/rook/rook/pkg/clusterd"
 	"github.com/rook/rook/pkg/daemon/ceph/client"
+	"github.com/rook/rook/pkg/operator/ceph/cluster/osd"
 	oposd "github.com/rook/rook/pkg/operator/ceph/cluster/osd"
 	cephver "github.com/rook/rook/pkg/operator/ceph/version"
 	"github.com/rook/rook/pkg/util/sys"
@@ -371,6 +372,7 @@ func getAvailableDevices(context *clusterd.Context, agent *OsdAgent) (*DeviceOsd
 				// This handles the case where the OSD deployment has been removed and the prepare
 				// job kicks in again to re-deploy the OSD.
 				continue
+
 			} else if device.Filesystem == "mpath_member" && agent.pvcBacked {
 				logger.Infof("allowing multipath disk %q with filesystem %q", device.Name, device.Filesystem)
 			} else {
@@ -602,31 +604,31 @@ func getVolumeGroupName(lvPath string) string {
 	return vgSlice[2]
 }
 
-// GetOSDInfoByID returns the OSDInfo using the ceph volume list
-func GetOSDInfoById(context *clusterd.Context, clusterInfo *client.ClusterInfo, osdID int) (oposd.OSDInfo, error) {
+// GetOSDInfoById returns the osdInfo using the ceph volume list
+func GetOSDInfoById(context *clusterd.Context, clusterInfo *client.ClusterInfo, osdID int) (*osd.OSDInfo, error) {
 	// LVM mode OSDs
 	osdLVMList, err := GetCephVolumeLVMOSDs(context, clusterInfo, clusterInfo.FSID, "", false, false)
 	if err != nil {
-		return oposd.OSDInfo{}, errors.Wrap(err, "failed to list lvm osd(s)")
+		return nil, errors.Wrap(err, "failed to list lvm osd(s)")
 	}
 
 	for _, osdInfo := range osdLVMList {
 		if osdInfo.ID == osdID {
-			return osdInfo, nil
+			return &osdInfo, nil
 		}
 	}
 
 	// Raw mode OSDs
 	osdRawList, err := GetCephVolumeRawOSDs(context, clusterInfo, clusterInfo.FSID, "", "", "", false, true)
 	if err != nil {
-		return oposd.OSDInfo{}, errors.Wrap(err, "failed to list raw osd(s)")
+		return nil, errors.Wrap(err, "failed to list raw osd(s)")
 	}
 
 	for _, osdInfo := range osdRawList {
 		if osdInfo.ID == osdID {
-			return osdInfo, nil
+			return &osdInfo, nil
 		}
 	}
 
-	return oposd.OSDInfo{}, fmt.Errorf("failed to get details for OSD %d using ceph-volume list", osdID)
+	return nil, fmt.Errorf("failed to get details for OSD %d using ceph-volume list", osdID)
 }

pkg/daemon/ceph/osd/remove.go

@@ -18,6 +18,7 @@ package osd
 
 import (
 	"fmt"
+	"os"
 	"strconv"
 	"time"
 
@@ -28,6 +29,7 @@ import (
 	"github.com/rook/rook/pkg/clusterd"
 	"github.com/rook/rook/pkg/daemon/ceph/client"
 	"github.com/rook/rook/pkg/operator/ceph/cluster/osd"
+	oposd "github.com/rook/rook/pkg/operator/ceph/cluster/osd"
 	"github.com/rook/rook/pkg/operator/k8sutil"
 )
 
@@ -245,3 +247,51 @@ func archiveCrash(clusterdContext *clusterd.Context, clusterInfo *client.Cluster
 		logger.Errorf("failed to archive the crash %q. %v", crashID, err)
 	}
 }
+
+// DestroyOSD fetches the OSD to be replaced based on the ID and then destroys that OSD and zaps the backing device
+func DestroyOSD(context *clusterd.Context, clusterInfo *client.ClusterInfo, id int, isPVC, isEncrypted bool) (*osd.OSDReplaceInfo, error) {
+	var block string
+	osdInfo, err := GetOSDInfoById(context, clusterInfo, id)
+	if err != nil {
+		return nil, errors.Wrapf(err, "failed to get OSD info for OSD.%d", id)
+	}
+
+	block = osdInfo.BlockPath
+
+	logger.Infof("destroying osd.%d", osdInfo.ID)
+	destroyOSDArgs := []string{"osd", "destroy", fmt.Sprintf("osd.%d", osdInfo.ID), "--yes-i-really-mean-it"}
+	_, err = client.NewCephCommand(context, clusterInfo, destroyOSDArgs).Run()
+	if err != nil {
+		return nil, errors.Wrapf(err, "failed to destroy osd.%d.", osdInfo.ID)
+	}
+	logger.Infof("successfully destroyed osd.%d", osdInfo.ID)
+
+	if isPVC && isEncrypted {
+		// remove the dm device
+		pvcName := os.Getenv(oposd.PVCNameEnvVarName)
+		target := oposd.EncryptionDMName(pvcName, oposd.DmcryptBlockType)
+		err = removeEncryptedDevice(context, target)
+		if err != nil {
+			return nil, errors.Wrapf(err, "failed to remove dm device %q", target)
+		}
+		// ceph-volume uses `/dev/mapper/*` for encrypted disks. This is not a block device. So we need to fetch the corresponding
+		// block device for cleanup using `ceph-volume lvm zap`
+		blockPath := fmt.Sprintf("/mnt/%s", pvcName)
+		diskInfo, err := clusterd.PopulateDeviceInfo(blockPath, context.Executor)
+		if err != nil {
+			return nil, errors.Wrapf(err, "failed to get device info for %q", blockPath)
+		}
+		block = diskInfo.RealPath
+	}
+
+	logger.Infof("zap OSD.%d path %q", osdInfo.ID, block)
+	output, err := context.Executor.ExecuteCommandWithCombinedOutput("stdbuf", "-oL", "ceph-volume", "lvm", "zap", block, "--destroy")
+	if err != nil {
+		return nil, errors.Wrapf(err, "failed to zap osd.%d path %q. %s.", osdInfo.ID, block, output)
+	}
+
+	logger.Infof("%s\n", output)
+	logger.Infof("successfully zapped osd.%d path %q", osdInfo.ID, block)
+
+	return &osd.OSDReplaceInfo{ID: osdInfo.ID, Path: block}, nil
+}

pkg/daemon/ceph/osd/volume.go

@@ -1129,9 +1129,11 @@ func GetCephVolumeRawOSDs(context *clusterd.Context, clusterInfo *client.Cluster
 			}
 
 			// Close encrypted device
-			err = CloseEncryptedDevice(context, block)
-			if err != nil {
-				return nil, errors.Wrapf(err, "failed to close encrypted device %q for osd %d", block, osdID)
+			if block != "" {
+				err = CloseEncryptedDevice(context, block)
+				if err != nil {
+					return nil, errors.Wrapf(err, "failed to close encrypted device %q for osd %d", block, osdID)
+				}
 			}
 
 			// If there is a metadata block

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

@@ -18,7 +18,6 @@ package osd
 
 import (
 	"fmt"
-	"reflect"
 	"time"
 
 	"github.com/pkg/errors"
@@ -26,7 +25,6 @@ import (
 	"github.com/rook/rook/pkg/clusterd"
 	"github.com/rook/rook/pkg/daemon/ceph/client"
 	opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
-	"github.com/rook/rook/pkg/operator/ceph/reporting"
 	"github.com/rook/rook/pkg/operator/k8sutil"
 	kerrors "k8s.io/apimachinery/pkg/api/errors"
 )
@@ -82,7 +80,6 @@ func (m *OSDHealthMonitor) Start(monitoringRoutines map[string]*opcontroller.Clu
 		case <-time.After(*m.interval):
 			logger.Debug("checking osd processes status.")
 			m.checkOSDHealth()
-			m.updateCephStorageStatus()
 
 		case <-monitoringRoutines[daemon].InternalCtx.Done():
 			logger.Infof("stopping monitoring of OSDs in namespace %q", m.clusterInfo.Namespace)
@@ -174,66 +171,3 @@ func (m *OSDHealthMonitor) removeOSDDeploymentIfSafeToDestroy(outOSDid int) erro
 	}
 	return nil
 }
-
-// updateCephStorageStatus updates the Storage details in the ceph cluster CR
-func (m *OSDHealthMonitor) updateCephStorageStatus() {
-	cephCluster := cephv1.CephCluster{}
-	cephClusterStorage := cephv1.CephStorage{}
-
-	deviceClasses, err := client.GetDeviceClasses(m.context, m.clusterInfo)
-	if err != nil {
-		logger.Errorf("failed to get osd device classes. %v", err)
-		return
-	}
-
-	for _, deviceClass := range deviceClasses {
-		cephClusterStorage.DeviceClasses = append(cephClusterStorage.DeviceClasses, cephv1.DeviceClasses{Name: deviceClass})
-	}
-
-	osdStore, err := m.getOSDStoreStatus()
-	if err != nil {
-		logger.Errorf("failed to get osd store status. %v", err)
-		return
-	}
-
-	cephClusterStorage.OSD = *osdStore
-
-	err = m.context.Client.Get(m.clusterInfo.Context, m.clusterInfo.NamespacedName(), &cephCluster)
-	if err != nil {
-		if kerrors.IsNotFound(err) {
-			logger.Debug("CephCluster resource not found. Ignoring since object must be deleted.")
-			return
-		}
-		logger.Errorf("failed to retrieve ceph cluster %q to update ceph Storage. %v", m.clusterInfo.NamespacedName().Name, err)
-		return
-	}
-	if !reflect.DeepEqual(cephCluster.Status.CephStorage, cephClusterStorage) {
-		cephCluster.Status.CephStorage = &cephClusterStorage
-		if err := reporting.UpdateStatus(m.context.Client, &cephCluster); err != nil {
-			logger.Errorf("failed to update cluster %q Storage. %v", m.clusterInfo.NamespacedName().Name, err)
-			return
-		}
-	}
-}
-
-func (m *OSDHealthMonitor) getOSDStoreStatus() (*cephv1.OSDStatus, error) {
-	label := fmt.Sprintf("%s=%s", k8sutil.AppAttr, AppName)
-	osdDeployments, err := k8sutil.GetDeployments(m.clusterInfo.Context, m.context.Clientset, m.clusterInfo.Namespace, label)
-	if err != nil {
-		if kerrors.IsNotFound(err) {
-			return nil, nil
-		}
-		return nil, errors.Wrap(err, "failed to get osd deployments")
-	}
-
-	storeType := map[string]int{}
-	for i := range osdDeployments.Items {
-		if osdStore, ok := osdDeployments.Items[i].Labels[osdStore]; ok {
-			storeType[osdStore]++
-		}
-	}
-
-	return &cephv1.OSDStatus{
-		StoreType: storeType,
-	}, nil
-}

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

@@ -23,7 +23,6 @@ import (
 	"time"
 
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
-	"github.com/rook/rook/pkg/client/clientset/versioned/scheme"
 	"github.com/rook/rook/pkg/clusterd"
 	"github.com/rook/rook/pkg/daemon/ceph/client"
 	opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
@@ -33,8 +32,6 @@ import (
 	"github.com/stretchr/testify/assert"
 	apps "k8s.io/api/apps/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/runtime"
-	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 )
 
 func TestOSDHealthCheck(t *testing.T) {
@@ -140,104 +137,3 @@ func TestNewOSDHealthMonitor(t *testing.T) {
 		})
 	}
 }
-
-func TestUpdateCephStorageStatus(t *testing.T) {
-	ctx := context.TODO()
-	clusterInfo := client.AdminTestClusterInfo("fake")
-	executor := &exectest.MockExecutor{
-		MockExecuteCommandWithOutput: func(command string, args ...string) (string, error) {
-			logger.Infof("ExecuteCommandWithOutputFile: %s %v", command, args)
-			if args[1] == "crush" && args[2] == "class" && args[3] == "ls" {
-				// Mock executor for OSD crush class list command, returning ssd as available device class
-				return `["ssd"]`, nil
-			}
-			return "", nil
-		},
-	}
-
-	cephCluster := &cephv1.CephCluster{
-		ObjectMeta: metav1.ObjectMeta{
-			Name:      "testing",
-			Namespace: "fake",
-		},
-	}
-	// Objects to track in the fake client.
-	object := []runtime.Object{
-		cephCluster,
-	}
-	s := scheme.Scheme
-	// Create a fake client to mock API calls.
-	c := fake.NewClientBuilder().WithScheme(s).WithRuntimeObjects(object...).Build()
-
-	context := &clusterd.Context{
-		Executor:  executor,
-		Client:    c,
-		Clientset: testexec.New(t, 2),
-	}
-
-	// Initializing an OSD monitoring
-	osdHealthMonitor := NewOSDHealthMonitor(context, clusterInfo, true, cephv1.CephClusterHealthCheckSpec{})
-
-	t.Run("verify ssd device class added to storage status", func(t *testing.T) {
-		osdHealthMonitor.updateCephStorageStatus()
-		err := context.Client.Get(clusterInfo.Context, clusterInfo.NamespacedName(), cephCluster)
-		assert.NoError(t, err)
-		assert.Equal(t, 1, len(cephCluster.Status.CephStorage.DeviceClasses))
-		assert.Equal(t, "ssd", cephCluster.Status.CephStorage.DeviceClasses[0].Name)
-	})
-
-	t.Run("verify bluestore OSD count in storage status", func(t *testing.T) {
-		labels := map[string]string{
-			k8sutil.AppAttr:     AppName,
-			k8sutil.ClusterAttr: clusterInfo.Namespace,
-			OsdIdLabelKey:       "0",
-			osdStore:            "bluestore",
-		}
-
-		deployment := &apps.Deployment{
-			ObjectMeta: metav1.ObjectMeta{
-				Name:      "osd0",
-				Namespace: clusterInfo.Namespace,
-				Labels:    labels,
-			},
-		}
-		if _, err := context.Clientset.AppsV1().Deployments(clusterInfo.Namespace).Create(ctx, deployment, metav1.CreateOptions{}); err != nil {
-			logger.Errorf("Error creating fake deployment: %v", err)
-		}
-		osdHealthMonitor.updateCephStorageStatus()
-		err := context.Client.Get(clusterInfo.Context, clusterInfo.NamespacedName(), cephCluster)
-		assert.NoError(t, err)
-		assert.Equal(t, 1, len(cephCluster.Status.CephStorage.DeviceClasses))
-		assert.Equal(t, "ssd", cephCluster.Status.CephStorage.DeviceClasses[0].Name)
-		assert.Equal(t, 1, cephCluster.Status.CephStorage.OSD.StoreType["bluestore"])
-		assert.Equal(t, 0, cephCluster.Status.CephStorage.OSD.StoreType["bluestore-rdr"])
-
-	})
-
-	t.Run("verify bluestoreRDR OSD count in storage status", func(t *testing.T) {
-		labels := map[string]string{
-			k8sutil.AppAttr:     AppName,
-			k8sutil.ClusterAttr: clusterInfo.Namespace,
-			OsdIdLabelKey:       "1",
-			osdStore:            "bluestore-rdr",
-		}
-
-		deployment := &apps.Deployment{
-			ObjectMeta: metav1.ObjectMeta{
-				Name:      "osd1",
-				Namespace: clusterInfo.Namespace,
-				Labels:    labels,
-			},
-		}
-		if _, err := context.Clientset.AppsV1().Deployments(clusterInfo.Namespace).Create(ctx, deployment, metav1.CreateOptions{}); err != nil {
-			logger.Errorf("Error creating fake deployment: %v", err)
-		}
-		osdHealthMonitor.updateCephStorageStatus()
-		err := context.Client.Get(clusterInfo.Context, clusterInfo.NamespacedName(), cephCluster)
-		assert.NoError(t, err)
-		assert.Equal(t, 1, len(cephCluster.Status.CephStorage.DeviceClasses))
-		assert.Equal(t, "ssd", cephCluster.Status.CephStorage.DeviceClasses[0].Name)
-		assert.Equal(t, 1, cephCluster.Status.CephStorage.OSD.StoreType["bluestore-rdr"])
-		assert.Equal(t, 1, cephCluster.Status.CephStorage.OSD.StoreType["bluestore"])
-	})
-}

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

@@ -28,6 +28,7 @@ import (
 	"github.com/coreos/pkg/capnslog"
 	"github.com/pkg/errors"
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
+	"github.com/rook/rook/pkg/client/clientset/versioned/scheme"
 	"github.com/rook/rook/pkg/clusterd"
 	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
 	cephclientfake "github.com/rook/rook/pkg/daemon/ceph/client/fake"
@@ -45,6 +46,7 @@ import (
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/kubernetes/fake"
 	k8stesting "k8s.io/client-go/testing"
+	clientfake "sigs.k8s.io/controller-runtime/pkg/client/fake"
 )
 
 var (
@@ -192,12 +194,28 @@ func testOSDIntegration(t *testing.T) {
 	clusterInfo.Context = ctx
 	executor := osdIntegrationTestExecutor(t, clientset, namespace)
 
+	cephCluster := &cephv1.CephCluster{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      clusterName,
+			Namespace: namespace,
+		},
+		Spec: cephv1.ClusterSpec{},
+	}
+	// Objects to track in the fake client.
+	object := []runtime.Object{
+		cephCluster,
+	}
+	s := scheme.Scheme
+	// Create a fake client to mock API calls.
+	client := clientfake.NewClientBuilder().WithScheme(s).WithRuntimeObjects(object...).Build()
+
 	rootCtx := &clusterd.Context{
 		Clientset: clientset,
 		ConfigDir: "/var/lib/rook",
 		Executor:  executor,
+		Client:    client,
 	}
-	spec := cephv1.ClusterSpec{
+	cephCluster.Spec = cephv1.ClusterSpec{
 		CephVersion: cephv1.CephVersionSpec{
 			Image: "quay.io/ceph/ceph:v16.2.0",
 		},
@@ -219,13 +237,13 @@ func testOSDIntegration(t *testing.T) {
 	}
 	osdsPerNode := 2 // vda and vdb
 
-	c := New(rootCtx, clusterInfo, spec, "myversion")
+	c := New(rootCtx, clusterInfo, cephCluster.Spec, "myversion")
 
 	var startErr error
 	var done bool
 	runReconcile := func(ctx context.Context) {
 		// reset environment
-		c = New(rootCtx, clusterInfo, spec, "myversion")
+		c = New(rootCtx, clusterInfo, cephCluster.Spec, "myversion")
 		clusterInfo.Context = ctx
 		statusMapWatcher.Reset()
 
@@ -280,10 +298,10 @@ func testOSDIntegration(t *testing.T) {
 	})
 
 	t.Run("increase number of OSDs", func(t *testing.T) {
-		spec.Storage.Selection.DeviceFilter = "/dev/vd[abc]" // 3 more (1 more per node)
-		spec.Storage.StorageClassDeviceSets[0].Count = 8     // 2 more portable
-		spec.Storage.StorageClassDeviceSets[1].Count = 6     // 3 more (1 more per node)
-		osdsPerNode = 3                                      // vda, vdb, vdc
+		cephCluster.Spec.Storage.Selection.DeviceFilter = "/dev/vd[abc]" // 3 more (1 more per node)
+		cephCluster.Spec.Storage.StorageClassDeviceSets[0].Count = 8     // 2 more portable
+		cephCluster.Spec.Storage.StorageClassDeviceSets[1].Count = 6     // 3 more (1 more per node)
+		osdsPerNode = 3                                                  // vda, vdb, vdc
 
 		go runReconcile(contextCancel)
 
@@ -301,9 +319,9 @@ func testOSDIntegration(t *testing.T) {
 	})
 
 	t.Run("mixed create and update, cancel reconcile, and continue reconcile", func(t *testing.T) {
-		spec.Storage.Selection.DeviceFilter = "/dev/vd[abcd]" // 3 more (1 more per node)
-		spec.Storage.StorageClassDeviceSets[0].Count = 10     // 2 more portable
-		osdsPerNode = 4                                       // vd[a-d]
+		cephCluster.Spec.Storage.Selection.DeviceFilter = "/dev/vd[abcd]" // 3 more (1 more per node)
+		cephCluster.Spec.Storage.StorageClassDeviceSets[0].Count = 10     // 2 more portable
+		osdsPerNode = 4                                                   // vd[a-d]
 
 		go runReconcile(contextCancel)
 		cms := waitForNumConfigMaps(clientset, namespace, 5) // 3 nodes + 2 new PVCs
@@ -352,8 +370,8 @@ func testOSDIntegration(t *testing.T) {
 	})
 
 	t.Run("failures reported in status configmaps", func(t *testing.T) {
-		spec.Storage.Selection.DeviceFilter = "/dev/vd[abcde]" // 3 more (1 more per node)
-		osdsPerNode = 5                                        // vd[a-e]
+		cephCluster.Spec.Storage.Selection.DeviceFilter = "/dev/vd[abcde]" // 3 more (1 more per node)
+		osdsPerNode = 5                                                    // vd[a-e]
 
 		go runReconcile(contextCancel)
 		cms := waitForNumConfigMaps(clientset, namespace, 3) // 3 nodes
@@ -416,8 +434,8 @@ func testOSDIntegration(t *testing.T) {
 	})
 
 	t.Run("failures during deployment creation", func(t *testing.T) {
-		spec.Storage.Selection.DeviceFilter = "/dev/vd[abcdef]" // 3 more (1 more per node)
-		osdsPerNode = 6                                         // vd[a-f]
+		cephCluster.Spec.Storage.Selection.DeviceFilter = "/dev/vd[abcdef]" // 3 more (1 more per node)
+		osdsPerNode = 6                                                     // vd[a-f]
 
 		failCreatingDeployments = []string{"osd-31", "osd-33"}
 		go runReconcile(contextCancel)
@@ -454,7 +472,7 @@ func testOSDIntegration(t *testing.T) {
 			Portable:             true,
 			VolumeClaimTemplates: []corev1.PersistentVolumeClaim{},
 		}
-		spec.Storage.StorageClassDeviceSets = append(spec.Storage.StorageClassDeviceSets, newSCDS)
+		cephCluster.Spec.Storage.StorageClassDeviceSets = append(cephCluster.Spec.Storage.StorageClassDeviceSets, newSCDS)
 
 		go runReconcile(contextCancel)
 		cms := waitForNumConfigMaps(clientset, namespace, 3) // 3 nodes
@@ -469,7 +487,7 @@ func testOSDIntegration(t *testing.T) {
 		assert.Len(t, deploymentsCreated, 0)
 		assert.Len(t, deploymentsUpdated, 34)
 
-		spec.Storage.StorageClassDeviceSets[2].VolumeClaimTemplates = []corev1.PersistentVolumeClaim{
+		cephCluster.Spec.Storage.StorageClassDeviceSets[2].VolumeClaimTemplates = []corev1.PersistentVolumeClaim{
 			newDummyPVC("data", namespace, "100Gi", "ec2"),
 			newDummyPVC("metadata", namespace, "10Gi", "uncle-rogers-secret-stuff"),
 		}
@@ -562,6 +580,10 @@ func osdIntegrationTestExecutor(t *testing.T, clientset *fake.Clientset, namespa
 					if args[2] == "get-device-class" {
 						return cephclientfake.OSDDeviceClassOutput(args[3]), nil
 					}
+					if args[2] == "class" && args[3] == "ls" {
+						// Mock executor for OSD crush class list command, returning ssd as available device class
+						return `["ssd"]`, nil
+					}
 				}
 			}
 			if args[0] == "versions" {

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

@@ -21,6 +21,7 @@ import (
 	"bufio"
 	"context"
 	"fmt"
+	"reflect"
 	"sort"
 	"strconv"
 	"strings"
@@ -28,10 +29,12 @@ import (
 
 	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"
 	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
 	osdconfig "github.com/rook/rook/pkg/operator/ceph/cluster/osd/config"
 	"github.com/rook/rook/pkg/operator/ceph/cluster/osd/topology"
 	"github.com/rook/rook/pkg/operator/ceph/controller"
+	"github.com/rook/rook/pkg/operator/ceph/reporting"
 	cephver "github.com/rook/rook/pkg/operator/ceph/version"
 	"github.com/rook/rook/pkg/operator/k8sutil"
 	"github.com/rook/rook/pkg/util"
@@ -117,6 +120,7 @@ type OSDInfo struct {
 	Encrypted        bool   `json:"encrypted"`
 	ExportService    bool   `json:"exportService"`
 	NodeName         string `json:"nodeName"`
+	PVCName          string `json:"pvcName"`
 }
 
 // OrchestrationStatus represents the status of an OSD orchestration
@@ -260,6 +264,11 @@ func (c *Cluster) Start() error {
 		return errors.Wrapf(err, "failed to reconcile key rotation cron jobs")
 	}
 
+	err = c.updateCephStorageStatus()
+	if err != nil {
+		return errors.Wrapf(err, "failed to update ceph storage status")
+	}
+
 	if c.replaceOSD != nil {
 		delOpts := &k8sutil.DeleteOptions{MustDelete: true, WaitOptions: k8sutil.WaitOptions{Wait: true}}
 		err := k8sutil.DeleteConfigMap(c.clusterInfo.Context, c.context.Clientset, OSDReplaceConfigName, namespace, delOpts)
@@ -611,6 +620,10 @@ func (c *Cluster) getOSDInfo(d *appsv1.Deployment) (OSDInfo, error) {
 
 	osd.Store = d.Labels[osdStore]
 
+	if isPVC {
+		osd.PVCName = d.Labels[OSDOverPVCLabelKey]
+	}
+
 	return osd, nil
 }
 
@@ -871,3 +884,63 @@ func (c *Cluster) waitForHealthyPGs() (bool, error) {
 
 	return true, nil
 }
+
+func (c *Cluster) updateCephStorageStatus() error {
+	cephCluster := cephv1.CephCluster{}
+	cephClusterStorage := cephv1.CephStorage{}
+
+	deviceClasses, err := client.GetDeviceClasses(c.context, c.clusterInfo)
+	if err != nil {
+		return errors.Wrap(err, "failed to get osd device classes")
+	}
+
+	for _, deviceClass := range deviceClasses {
+		cephClusterStorage.DeviceClasses = append(cephClusterStorage.DeviceClasses, cephv1.DeviceClasses{Name: deviceClass})
+	}
+
+	osdStore, err := c.getOSDStoreStatus()
+	if err != nil {
+		return errors.Wrapf(err, "failed to get osd store status")
+	}
+
+	cephClusterStorage.OSD = *osdStore
+
+	err = c.context.Client.Get(c.clusterInfo.Context, c.clusterInfo.NamespacedName(), &cephCluster)
+	if err != nil {
+		if kerrors.IsNotFound(err) {
+			logger.Debug("CephCluster resource not found. Ignoring since object must be deleted.")
+			return nil
+		}
+		return errors.Wrapf(err, "failed to retrieve ceph cluster %q to update ceph Storage", c.clusterInfo.NamespacedName().Name)
+	}
+	if !reflect.DeepEqual(cephCluster.Status.CephStorage, cephClusterStorage) {
+		cephCluster.Status.CephStorage = &cephClusterStorage
+		if err := reporting.UpdateStatus(c.context.Client, &cephCluster); err != nil {
+			return errors.Wrapf(err, "failed to update cluster %q Storage.", c.clusterInfo.NamespacedName().Name)
+		}
+	}
+
+	return nil
+}
+
+func (c *Cluster) getOSDStoreStatus() (*cephv1.OSDStatus, error) {
+	label := fmt.Sprintf("%s=%s", k8sutil.AppAttr, AppName)
+	osdDeployments, err := k8sutil.GetDeployments(c.clusterInfo.Context, c.context.Clientset, c.clusterInfo.Namespace, label)
+	if err != nil {
+		if kerrors.IsNotFound(err) {
+			return nil, nil
+		}
+		return nil, errors.Wrap(err, "failed to get osd deployments")
+	}
+
+	storeType := map[string]int{}
+	for i := range osdDeployments.Items {
+		if osdStore, ok := osdDeployments.Items[i].Labels[osdStore]; ok {
+			storeType[osdStore]++
+		}
+	}
+
+	return &cephv1.OSDStatus{
+		StoreType: storeType,
+	}, nil
+}

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

@@ -25,7 +25,9 @@ import (
 	"github.com/pkg/errors"
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	fakeclient "github.com/rook/rook/pkg/client/clientset/versioned/fake"
+	"github.com/rook/rook/pkg/client/clientset/versioned/scheme"
 	"github.com/rook/rook/pkg/clusterd"
+	"github.com/rook/rook/pkg/daemon/ceph/client"
 	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
 	cephclientfake "github.com/rook/rook/pkg/daemon/ceph/client/fake"
 	discoverDaemon "github.com/rook/rook/pkg/daemon/discover"
@@ -34,6 +36,7 @@ import (
 	cephver "github.com/rook/rook/pkg/operator/ceph/version"
 	"github.com/rook/rook/pkg/operator/k8sutil"
 	"github.com/rook/rook/pkg/operator/test"
+	testexec "github.com/rook/rook/pkg/operator/test"
 	exectest "github.com/rook/rook/pkg/util/exec/test"
 	"github.com/stretchr/testify/assert"
 	apps "k8s.io/api/apps/v1"
@@ -45,6 +48,7 @@ import (
 	"k8s.io/apimachinery/pkg/watch"
 	"k8s.io/client-go/kubernetes/fake"
 	k8stesting "k8s.io/client-go/testing"
+	clientfake "sigs.k8s.io/controller-runtime/pkg/client/fake"
 )
 
 const (
@@ -70,13 +74,40 @@ func TestOSDProperties(t *testing.T) {
 }
 
 func TestStart(t *testing.T) {
+	namespace := "ns"
 	clientset := fake.NewSimpleClientset()
+	executor := &exectest.MockExecutor{
+		MockExecuteCommandWithOutput: func(command string, args ...string) (string, error) {
+			logger.Infof("ExecuteCommandWithOutputFile: %s %v", command, args)
+			if args[1] == "crush" && args[2] == "class" && args[3] == "ls" {
+				// Mock executor for OSD crush class list command, returning ssd as available device class
+				return `["ssd"]`, nil
+			}
+			return "", nil
+		},
+	}
+
+	cephCluster := &cephv1.CephCluster{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "testing",
+			Namespace: namespace,
+		},
+		Spec: cephv1.ClusterSpec{},
+	}
+	// Objects to track in the fake client.
+	object := []runtime.Object{
+		cephCluster,
+	}
+	s := scheme.Scheme
+	// Create a fake client to mock API calls.
+	client := clientfake.NewClientBuilder().WithScheme(s).WithRuntimeObjects(object...).Build()
 	clusterInfo := &cephclient.ClusterInfo{
-		Namespace:   "ns",
+		Namespace:   namespace,
 		CephVersion: cephver.Quincy,
 		Context:     context.TODO(),
 	}
-	context := &clusterd.Context{Clientset: clientset, ConfigDir: "/var/lib/rook", Executor: &exectest.MockExecutor{}}
+	clusterInfo.SetName("rook-ceph-test")
+	context := &clusterd.Context{Clientset: clientset, Client: client, ConfigDir: "/var/lib/rook", Executor: executor}
 	spec := cephv1.ClusterSpec{}
 	c := New(context, clusterInfo, spec, "myversion")
 
@@ -163,6 +194,11 @@ func TestAddRemoveNode(t *testing.T) {
 	generateKey := "expected key"
 	executor := &exectest.MockExecutor{
 		MockExecuteCommandWithOutput: func(command string, args ...string) (string, error) {
+			logger.Infof("Command: %s %v", command, args)
+			if args[1] == "crush" && args[2] == "class" && args[3] == "ls" {
+				// Mock executor for OSD crush class list command, returning ssd as available device class
+				return `["ssd"]`, nil
+			}
 			return "{\"key\": \"" + generateKey + "\"}", nil
 		},
 	}
@@ -173,20 +209,37 @@ func TestAddRemoveNode(t *testing.T) {
 		Executor:      executor,
 		RookClientset: fakeclient.NewSimpleClientset(),
 	}
-	spec := cephv1.ClusterSpec{
-		DataDirHostPath: context.ConfigDir,
-		Storage: cephv1.StorageScopeSpec{
-			Nodes: []cephv1.Node{
-				{
-					Name: nodeName,
-					Selection: cephv1.Selection{
-						Devices: []cephv1.Device{{Name: "sdx"}},
+
+	cephCluster := &cephv1.CephCluster{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "testing",
+			Namespace: namespace,
+		},
+		Spec: cephv1.ClusterSpec{
+			DataDirHostPath: context.ConfigDir,
+			Storage: cephv1.StorageScopeSpec{
+				Nodes: []cephv1.Node{
+					{
+						Name: nodeName,
+						Selection: cephv1.Selection{
+							Devices: []cephv1.Device{{Name: "sdx"}},
+						},
 					},
 				},
 			},
 		},
 	}
-	c := New(context, clusterInfo, spec, "myversion")
+
+	// Objects to track in the fake client.
+	object := []runtime.Object{
+		cephCluster,
+	}
+	s := scheme.Scheme
+	// Create a fake client to mock API calls.
+	client := clientfake.NewClientBuilder().WithScheme(s).WithRuntimeObjects(object...).Build()
+	context.Client = client
+
+	c := New(context, clusterInfo, cephCluster.Spec, "myversion")
 
 	// kick off the start of the orchestration in a goroutine
 	var startErr error
@@ -246,6 +299,9 @@ func TestAddRemoveNode(t *testing.T) {
 					if args[2] == "get-device-class" {
 						return cephclientfake.OSDDeviceClassOutput(args[3]), nil
 					}
+					if args[2] == "class" {
+						return `["ssd"]`, nil
+					}
 				}
 				if args[1] == "out" {
 					return "", nil
@@ -276,8 +332,8 @@ func TestAddRemoveNode(t *testing.T) {
 	}
 
 	// modify the storage spec to remove the node from the cluster
-	spec.Storage.Nodes = []cephv1.Node{}
-	c = New(context, clusterInfo, spec, "myversion")
+	cephCluster.Spec.Storage.Nodes = []cephv1.Node{}
+	c = New(context, clusterInfo, cephCluster.Spec, "myversion")
 
 	// reset the orchestration status watcher
 	statusMapWatcher = watch.NewFake()
@@ -767,7 +823,7 @@ func TestReplaceOSDForNewStore(t *testing.T) {
 		fmt.Printf("%+v", c.replaceOSD)
 		assert.NotNil(t, c.replaceOSD)
 		assert.Equal(t, 2, c.replaceOSD.ID)
-		assert.Equal(t, "/some/path", c.replaceOSD.Path)
+		assert.Equal(t, "pvc1", c.replaceOSD.Path)
 
 		// assert that OSD.2 deployment got deleted
 		_, err = clientset.AppsV1().Deployments(clusterInfo.Namespace).Get(clusterInfo.Context, deploymentName(2), metav1.GetOptions{})
@@ -781,7 +837,7 @@ func TestReplaceOSDForNewStore(t *testing.T) {
 		err = json.Unmarshal([]byte(actualCM.Data[OSDReplaceConfigKey]), &expectedOSDInfo)
 		assert.NoError(t, err)
 		assert.Equal(t, 2, expectedOSDInfo.ID)
-		assert.Equal(t, "/some/path", c.replaceOSD.Path)
+		assert.Equal(t, "pvc1", c.replaceOSD.Path)
 
 		// delete configmap
 		err = k8sutil.DeleteConfigMap(clusterInfo.Context, clientset, OSDReplaceConfigName, clusterInfo.Namespace, &k8sutil.DeleteOptions{})
@@ -813,3 +869,108 @@ func TestReplaceOSDForNewStore(t *testing.T) {
 		assert.Nil(t, c.replaceOSD)
 	})
 }
+
+func TestUpdateCephStorageStatus(t *testing.T) {
+	ctx := context.TODO()
+	clusterInfo := client.AdminTestClusterInfo("fake")
+	executor := &exectest.MockExecutor{
+		MockExecuteCommandWithOutput: func(command string, args ...string) (string, error) {
+			logger.Infof("ExecuteCommandWithOutputFile: %s %v", command, args)
+			if args[1] == "crush" && args[2] == "class" && args[3] == "ls" {
+				// Mock executor for OSD crush class list command, returning ssd as available device class
+				return `["ssd"]`, nil
+			}
+			return "", nil
+		},
+	}
+
+	cephCluster := &cephv1.CephCluster{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "testing",
+			Namespace: "fake",
+		},
+		Spec: cephv1.ClusterSpec{},
+	}
+	// Objects to track in the fake client.
+	object := []runtime.Object{
+		cephCluster,
+	}
+	s := scheme.Scheme
+	// Create a fake client to mock API calls.
+	client := clientfake.NewClientBuilder().WithScheme(s).WithRuntimeObjects(object...).Build()
+
+	context := &clusterd.Context{
+		Executor:  executor,
+		Client:    client,
+		Clientset: testexec.New(t, 2),
+	}
+
+	// Initializing an OSD monitoring
+	c := New(context, clusterInfo, cephCluster.Spec, "myversion")
+
+	t.Run("verify ssd device class added to storage status", func(t *testing.T) {
+		err := c.updateCephStorageStatus()
+		assert.NoError(t, err)
+		err = context.Client.Get(clusterInfo.Context, clusterInfo.NamespacedName(), cephCluster)
+		assert.NoError(t, err)
+		assert.Equal(t, 1, len(cephCluster.Status.CephStorage.DeviceClasses))
+		assert.Equal(t, "ssd", cephCluster.Status.CephStorage.DeviceClasses[0].Name)
+	})
+
+	t.Run("verify bluestore OSD count in storage status", func(t *testing.T) {
+		labels := map[string]string{
+			k8sutil.AppAttr:     AppName,
+			k8sutil.ClusterAttr: clusterInfo.Namespace,
+			OsdIdLabelKey:       "0",
+			osdStore:            "bluestore",
+		}
+
+		deployment := &apps.Deployment{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "osd0",
+				Namespace: clusterInfo.Namespace,
+				Labels:    labels,
+			},
+		}
+		if _, err := context.Clientset.AppsV1().Deployments(clusterInfo.Namespace).Create(ctx, deployment, metav1.CreateOptions{}); err != nil {
+			logger.Errorf("Error creating fake deployment: %v", err)
+		}
+		err := c.updateCephStorageStatus()
+		assert.NoError(t, err)
+		err = context.Client.Get(clusterInfo.Context, clusterInfo.NamespacedName(), cephCluster)
+		assert.NoError(t, err)
+		assert.Equal(t, 1, len(cephCluster.Status.CephStorage.DeviceClasses))
+		assert.Equal(t, "ssd", cephCluster.Status.CephStorage.DeviceClasses[0].Name)
+		assert.Equal(t, 1, cephCluster.Status.CephStorage.OSD.StoreType["bluestore"])
+		assert.Equal(t, 0, cephCluster.Status.CephStorage.OSD.StoreType["bluestore-rdr"])
+
+	})
+
+	t.Run("verify bluestoreRDR OSD count in storage status", func(t *testing.T) {
+		labels := map[string]string{
+			k8sutil.AppAttr:     AppName,
+			k8sutil.ClusterAttr: clusterInfo.Namespace,
+			OsdIdLabelKey:       "1",
+			osdStore:            "bluestore-rdr",
+		}
+
+		deployment := &apps.Deployment{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "osd1",
+				Namespace: clusterInfo.Namespace,
+				Labels:    labels,
+			},
+		}
+		if _, err := context.Clientset.AppsV1().Deployments(clusterInfo.Namespace).Create(ctx, deployment, metav1.CreateOptions{}); err != nil {
+			logger.Errorf("Error creating fake deployment: %v", err)
+		}
+		err := c.updateCephStorageStatus()
+		assert.NoError(t, err)
+		err = context.Client.Get(clusterInfo.Context, clusterInfo.NamespacedName(), cephCluster)
+		assert.NoError(t, err)
+		assert.Equal(t, 1, len(cephCluster.Status.CephStorage.DeviceClasses))
+		assert.Equal(t, "ssd", cephCluster.Status.CephStorage.DeviceClasses[0].Name)
+		assert.Equal(t, 1, cephCluster.Status.CephStorage.OSD.StoreType["bluestore-rdr"])
+		assert.Equal(t, 1, cephCluster.Status.CephStorage.OSD.StoreType["bluestore"])
+	})
+}

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

@@ -134,7 +134,13 @@ func (c *Cluster) getOSDWithNonMatchingStore() ([]OSDReplaceInfo, error) {
 				if err != nil {
 					return nil, errors.Wrapf(err, "failed to details about the OSD %q", deployments.Items[i].Name)
 				}
-				osdReplaceList = append(osdReplaceList, OSDReplaceInfo{ID: osdInfo.ID, Path: osdInfo.BlockPath, Node: osdInfo.NodeName})
+				var path string
+				if osdInfo.PVCName != "" {
+					path = osdInfo.PVCName
+				} else {
+					path = osdInfo.BlockPath
+				}
+				osdReplaceList = append(osdReplaceList, OSDReplaceInfo{ID: osdInfo.ID, Path: path, Node: osdInfo.NodeName})
 			}
 		}
 	}

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

@@ -160,7 +160,7 @@ func TestGetOSDWithNonMatchingStoreOnPVCs(t *testing.T) {
 		assert.NoError(t, err)
 		assert.Equal(t, 1, len(osdList))
 		assert.Equal(t, 0, osdList[0].ID)
-		assert.Equal(t, "/some/path", osdList[0].Path)
+		assert.Equal(t, "pvc0", osdList[0].Path)
 	})
 }