Merge pull request #12752 from rook/mergify/bp/release-1.12/pr-12507

Replace OSD to use new backend store (backport #12507)

Documentation/CRDs/specification.md

@@ -3612,6 +3612,18 @@ string
 <td>
 </td>
 </tr>
+<tr>
+<td>
+<code>osd</code><br/>
+<em>
+<a href="#ceph.rook.io/v1.OSDStatus">
+OSDStatus
+</a>
+</em>
+</td>
+<td>
+</td>
+</tr>
 </tbody>
 </table>
 <h3 id="ceph.rook.io/v1.CephVersionSpec">CephVersionSpec
@@ -8262,6 +8274,35 @@ string
 </tr>
 </tbody>
 </table>
+<h3 id="ceph.rook.io/v1.OSDStatus">OSDStatus
+</h3>
+<p>
+(<em>Appears on:</em><a href="#ceph.rook.io/v1.CephStorage">CephStorage</a>)
+</p>
+<div>
+<p>OSDStatus represents OSD status of the ceph Cluster</p>
+</div>
+<table>
+<thead>
+<tr>
+<th>Field</th>
+<th>Description</th>
+</tr>
+</thead>
+<tbody>
+<tr>
+<td>
+<code>storeType</code><br/>
+<em>
+map[string]int
+</em>
+</td>
+<td>
+<p>StoreType is a mapping between the OSD backend stores and number of OSDs using these stores</p>
+</td>
+</tr>
+</tbody>
+</table>
 <h3 id="ceph.rook.io/v1.OSDStore">OSDStore
 </h3>
 <p>
@@ -8290,6 +8331,19 @@ string
 <p>Type of backend storage to be used while creating OSDs. If empty, then bluestore will be used</p>
 </td>
 </tr>
+<tr>
+<td>
+<code>updateStore</code><br/>
+<em>
+string
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>UpdateStore updates the backend store for existing OSDs. It destroys each OSD one at a time, cleans up the backing disk
+and prepares same OSD on that disk</p>
+</td>
+</tr>
 </tbody>
 </table>
 <h3 id="ceph.rook.io/v1.ObjectEndpoints">ObjectEndpoints

cmd/rook/ceph/osd.go

@@ -19,6 +19,7 @@ package ceph
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"os"
 	"path"
 	"strconv"
@@ -29,8 +30,12 @@ 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"
 	oposd "github.com/rook/rook/pkg/operator/ceph/cluster/osd"
 	osdcfg "github.com/rook/rook/pkg/operator/ceph/cluster/osd/config"
 	opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
@@ -66,6 +71,7 @@ var (
 	ownerRefID              string
 	clusterName             string
 	osdID                   int
+	replaceOSDID            int
 	osdStoreType            string
 	osdStringID             string
 	osdUUID                 string
@@ -88,6 +94,7 @@ func addOSDFlags(command *cobra.Command) {
 	addOSDConfigFlags(provisionCmd)
 
 	// flags specific to provisioning
+	provisionCmd.Flags().IntVar(&replaceOSDID, "replace-osd", -1, "osd to be destroyed")
 	provisionCmd.Flags().StringVar(&cfg.devices, "data-devices", "", "comma separated list of devices to use for storage")
 	provisionCmd.Flags().StringVar(&osdDataDeviceFilter, "data-device-filter", "", "a regex filter for the device names to use, or \"all\"")
 	provisionCmd.Flags().StringVar(&osdDataDevicePathFilter, "data-device-path-filter", "", "a regex filter for the device path names to use")
@@ -196,6 +203,7 @@ func writeOSDConfig(cmd *cobra.Command, args []string) error {
 
 // Provision a device or directory for an OSD
 func prepareOSD(cmd *cobra.Command, args []string) error {
+
 	if err := verifyConfigFlags(provisionCmd); err != nil {
 		return err
 	}
@@ -251,8 +259,19 @@ func prepareOSD(cmd *cobra.Command, args []string) error {
 	clusterInfo.OwnerInfo = ownerInfo
 	clusterInfo.Context = cmd.Context()
 	kv := k8sutil.NewConfigMapKVStore(clusterInfo.Namespace, context.Clientset, ownerInfo)
+
+	// destroy the OSD using the OSD ID
+	var replaceOSD *osd.OSDReplaceInfo
+	if replaceOSDID != -1 {
+		osdInfo, err := destroyOSD(context, &clusterInfo, replaceOSDID)
+		if err != nil {
+			rook.TerminateFatal(errors.Wrapf(err, "failed to destroy OSD %d.", osdInfo.ID))
+		}
+		replaceOSD = &oposd.OSDReplaceInfo{ID: osdInfo.ID, Path: osdInfo.BlockPath}
+	}
+
 	agent := osddaemon.NewAgent(context, dataDevices, cfg.metadataDevice, forceFormat,
-		cfg.storeConfig, &clusterInfo, cfg.nodeName, kv, cfg.pvcBacked)
+		cfg.storeConfig, &clusterInfo, cfg.nodeName, kv, replaceOSD, cfg.pvcBacked)
 
 	if cfg.metadataDevice != "" {
 		metaDevice = cfg.metadataDevice
@@ -398,3 +417,36 @@ 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
+}

deploy/charts/rook-ceph/templates/resources.yaml

@@ -4619,6 +4619,10 @@ spec:
                             - bluestore
                             - bluestore-rdr
                           type: string
+                        updateStore:
+                          description: UpdateStore updates the backend store for existing OSDs. It destroys each OSD one at a time, cleans up the backing disk and prepares same OSD on that disk
+                          pattern: ^$|^yes-really-update-store$
+                          type: string
                       type: object
                     useAllDevices:
                       description: Whether to consume all the storage devices found on a machine
@@ -4987,6 +4991,15 @@ spec:
                             type: string
                         type: object
                       type: array
+                    osd:
+                      description: OSDStatus represents OSD status of the ceph Cluster
+                      properties:
+                        storeType:
+                          additionalProperties:
+                            type: integer
+                          description: StoreType is a mapping between the OSD backend stores and number of OSDs using these stores
+                          type: object
+                      type: object
                   type: object
                 version:
                   description: ClusterVersion represents the version of a Ceph Cluster

deploy/examples/crds.yaml

@@ -4617,6 +4617,10 @@ spec:
                             - bluestore
                             - bluestore-rdr
                           type: string
+                        updateStore:
+                          description: UpdateStore updates the backend store for existing OSDs. It destroys each OSD one at a time, cleans up the backing disk and prepares same OSD on that disk
+                          pattern: ^$|^yes-really-update-store$
+                          type: string
                       type: object
                     useAllDevices:
                       description: Whether to consume all the storage devices found on a machine
@@ -4985,6 +4989,15 @@ spec:
                             type: string
                         type: object
                       type: array
+                    osd:
+                      description: OSDStatus represents OSD status of the ceph Cluster
+                      properties:
+                        storeType:
+                          additionalProperties:
+                            type: integer
+                          description: StoreType is a mapping between the OSD backend stores and number of OSDs using these stores
+                          type: object
+                      type: object
                   type: object
                 version:
                   description: ClusterVersion represents the version of a Ceph Cluster

pkg/apis/ceph.rook.io/v1/types.go

@@ -420,6 +420,7 @@ type Capacity struct {
 // CephStorage represents flavors of Ceph Cluster Storage
 type CephStorage struct {
 	DeviceClasses []DeviceClasses `json:"deviceClasses,omitempty"`
+	OSD           OSDStatus       `json:"osd,omitempty"`
 }
 
 // DeviceClasses represents device classes of a Ceph Cluster
@@ -427,6 +428,12 @@ type DeviceClasses struct {
 	Name string `json:"name,omitempty"`
 }
 
+// OSDStatus represents OSD status of the ceph Cluster
+type OSDStatus struct {
+	// StoreType is a mapping between the OSD backend stores and number of OSDs using these stores
+	StoreType map[string]int `json:"storeType,omitempty"`
+}
+
 // ClusterVersion represents the version of a Ceph Cluster
 type ClusterVersion struct {
 	Image   string `json:"image,omitempty"`
@@ -2602,6 +2609,11 @@ type OSDStore struct {
 	// +optional
 	// +kubebuilder:validation:Enum=bluestore;bluestore-rdr;
 	Type string `json:"type,omitempty"`
+	// UpdateStore updates the backend store for existing OSDs. It destroys each OSD one at a time, cleans up the backing disk
+	// and prepares same OSD on that disk
+	// +optional
+	// +kubebuilder:validation:Pattern=`^$|^yes-really-update-store$`
+	UpdateStore string `json:"updateStore,omitempty"`
 }
 
 // Node is a storage nodes

pkg/apis/ceph.rook.io/v1/zz_generated.deepcopy.go

@@ -1649,6 +1649,7 @@ func (in *CephStorage) DeepCopyInto(out *CephStorage) {
 		*out = make([]DeviceClasses, len(*in))
 		copy(*out, *in)
 	}
+	in.OSD.DeepCopyInto(&out.OSD)
 	return
 }
 
@@ -3197,6 +3198,29 @@ func (in *NotificationKeyFilterRule) DeepCopy() *NotificationKeyFilterRule {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *OSDStatus) DeepCopyInto(out *OSDStatus) {
+	*out = *in
+	if in.StoreType != nil {
+		in, out := &in.StoreType, &out.StoreType
+		*out = make(map[string]int, len(*in))
+		for key, val := range *in {
+			(*out)[key] = val
+		}
+	}
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new OSDStatus.
+func (in *OSDStatus) DeepCopy() *OSDStatus {
+	if in == nil {
+		return nil
+	}
+	out := new(OSDStatus)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *OSDStore) DeepCopyInto(out *OSDStore) {
 	*out = *in

pkg/daemon/ceph/cleanup/disk.go

@@ -63,7 +63,7 @@ func (s *DiskSanitizer) StartSanitizeDisks() {
 		logger.Errorf("failed to list lvm osd(s). %v", err)
 	} else {
 		// Start the sanitizing sequence
-		s.sanitizeLVMDisk(osdLVMList)
+		s.SanitizeLVMDisk(osdLVMList)
 	}
 
 	// Raw based OSDs
@@ -72,11 +72,11 @@ func (s *DiskSanitizer) StartSanitizeDisks() {
 		logger.Errorf("failed to list raw osd(s). %v", err)
 	} else {
 		// Start the sanitizing sequence
-		s.sanitizeRawDisk(osdRawList)
+		s.SanitizeRawDisk(osdRawList)
 	}
 }
 
-func (s *DiskSanitizer) sanitizeRawDisk(osdRawList []oposd.OSDInfo) {
+func (s *DiskSanitizer) SanitizeRawDisk(osdRawList []oposd.OSDInfo) {
 	// Initialize work group to wait for completion of all the go routine
 	var wg sync.WaitGroup
 
@@ -93,7 +93,7 @@ func (s *DiskSanitizer) sanitizeRawDisk(osdRawList []oposd.OSDInfo) {
 	wg.Wait()
 }
 
-func (s *DiskSanitizer) sanitizeLVMDisk(osdLVMList []oposd.OSDInfo) {
+func (s *DiskSanitizer) SanitizeLVMDisk(osdLVMList []oposd.OSDInfo) {
 	// Initialize work group to wait for completion of all the go routine
 	var wg sync.WaitGroup
 	pvs := []string{}
@@ -112,7 +112,7 @@ func (s *DiskSanitizer) sanitizeLVMDisk(osdLVMList []oposd.OSDInfo) {
 	wg.Wait()
 
 	var wg2 sync.WaitGroup
-	// // purge remaining LVM2 metadata from PV
+	// purge remaining LVM2 metadata from PV
 	for _, pv := range pvs {
 		wg2.Add(1)
 		go s.executeSanitizeCommand(oposd.OSDInfo{BlockPath: pv}, &wg2)

pkg/daemon/ceph/osd/agent.go

@@ -19,6 +19,7 @@ package osd
 import (
 	"github.com/rook/rook/pkg/clusterd"
 	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
+	oposd "github.com/rook/rook/pkg/operator/ceph/cluster/osd"
 	"github.com/rook/rook/pkg/operator/ceph/cluster/osd/config"
 	"github.com/rook/rook/pkg/operator/k8sutil"
 )
@@ -37,11 +38,13 @@ type OsdAgent struct {
 	storeConfig    config.StoreConfig
 	kv             *k8sutil.ConfigMapKVStore
 	pvcBacked      bool
+	replaceOSD     *oposd.OSDReplaceInfo
 }
 
 // NewAgent is the instantiation of the OSD agent
 func NewAgent(context *clusterd.Context, devices []DesiredDevice, metadataDevice string, forceFormat bool,
-	storeConfig config.StoreConfig, clusterInfo *cephclient.ClusterInfo, nodeName string, kv *k8sutil.ConfigMapKVStore, pvcBacked bool) *OsdAgent {
+	storeConfig config.StoreConfig, clusterInfo *cephclient.ClusterInfo, nodeName string, kv *k8sutil.ConfigMapKVStore,
+	replaceOSD *oposd.OSDReplaceInfo, pvcBacked bool) *OsdAgent {
 
 	return &OsdAgent{
 		devices:        devices,
@@ -52,6 +55,7 @@ func NewAgent(context *clusterd.Context, devices []DesiredDevice, metadataDevice
 		nodeName:       nodeName,
 		kv:             kv,
 		pvcBacked:      pvcBacked,
+		replaceOSD:     replaceOSD,
 	}
 }
 
@@ -64,3 +68,12 @@ func getDeviceLVPath(context *clusterd.Context, deviceName string) string {
 	logger.Debugf("logical volume path for device %q is %q", deviceName, output)
 	return output
 }
+
+// GetReplaceOSDId returns the OSD ID based on the device name
+func (a *OsdAgent) GetReplaceOSDId(device string) int {
+	if device == a.replaceOSD.Path {
+		return a.replaceOSD.ID
+	}
+
+	return -1
+}

pkg/daemon/ceph/osd/daemon.go

@@ -601,3 +601,32 @@ 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) {
+	// 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)")
+	}
+
+	for _, osdInfo := range osdLVMList {
+		if osdInfo.ID == osdID {
+			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)")
+	}
+
+	for _, osdInfo := range osdRawList {
+		if osdInfo.ID == osdID {
+			return osdInfo, nil
+		}
+	}
+
+	return oposd.OSDInfo{}, fmt.Errorf("failed to get details for OSD %d using ceph-volume list", osdID)
+}

pkg/daemon/ceph/osd/device.go

@@ -63,6 +63,7 @@ type DeviceOsdIDEntry struct {
 	Config                DesiredDevice // Device specific config options
 	PersistentDevicePaths []string
 	DeviceInfo            *sys.LocalDisk // low-level info about the device
+	RestoreOSD            bool           // Restore OSD by reparing it with with OSD ID
 }
 
 func (m *DeviceOsdMapping) String() string {

pkg/daemon/ceph/osd/volume.go

@@ -282,6 +282,16 @@ func (a *OsdAgent) initializeBlockPVC(context *clusterd.Context, devices *Device
 				deviceArg,
 			}...)
 
+			if a.replaceOSD != nil {
+				replaceOSDID := a.GetReplaceOSDId(device.DeviceInfo.RealPath)
+				if replaceOSDID != -1 {
+					immediateExecuteArgs = append(immediateExecuteArgs, []string{
+						"--osd-id",
+						fmt.Sprintf("%d", replaceOSDID),
+					}...)
+				}
+			}
+
 			crushDeviceClass := os.Getenv(oposd.CrushDeviceClassVarName)
 			if crushDeviceClass != "" {
 				immediateExecuteArgs = append(immediateExecuteArgs, []string{crushDeviceClassFlag, crushDeviceClass}...)
@@ -526,6 +536,16 @@ func (a *OsdAgent) initializeDevicesRawMode(context *clusterd.Context, devices *
 				deviceArg,
 			}...)
 
+			if a.replaceOSD != nil {
+				restoreOSDID := a.GetReplaceOSDId(deviceArg)
+				if restoreOSDID != -1 {
+					immediateExecuteArgs = append(immediateExecuteArgs, []string{
+						"--osd-id",
+						fmt.Sprintf("%d", restoreOSDID),
+					}...)
+				}
+			}
+
 			// assign the device class specific to the device
 			immediateExecuteArgs = a.appendDeviceClassArg(device, immediateExecuteArgs)
 
@@ -860,6 +880,7 @@ func GetCephVolumeLVMOSDs(context *clusterd.Context, clusterInfo *client.Cluster
 			lvPath = lv
 		}
 
+		// TODO: Don't read osd store type from env variable
 		osdStore := os.Getenv(oposd.OSDStoreTypeVarName)
 		if osdStore == "" {
 			osdStore = string(cephv1.StoreTypeBlueStore)
@@ -1053,10 +1074,7 @@ func GetCephVolumeRawOSDs(context *clusterd.Context, clusterInfo *client.Cluster
 			blockPath = block
 		}
 
-		osdStore := os.Getenv(oposd.OSDStoreTypeVarName)
-		if osdStore == "" {
-			osdStore = string(cephv1.StoreTypeBlueStore)
-		}
+		osdStore := osdInfo.Type
 
 		osd := oposd.OSDInfo{
 			ID:      osdID,

pkg/daemon/ceph/osd/volume_test.go

@@ -26,6 +26,7 @@ import (
 	"github.com/pkg/errors"
 	"github.com/rook/rook/pkg/clusterd"
 	cephclient "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/ceph/cluster/osd/config"
 	cephver "github.com/rook/rook/pkg/operator/ceph/version"
@@ -1189,6 +1190,48 @@ func TestInitializeBlockPVC(t *testing.T) {
 	assert.Equal(t, "", blockPath)
 	assert.Equal(t, "", metadataBlockPath)
 	assert.Equal(t, "", walBlockPath)
+
+	// Test for condition when osd is prepared with existing osd ID
+	a = &OsdAgent{clusterInfo: clusterInfo, nodeName: "node1", storeConfig: config.StoreConfig{StoreType: "bluestore"}, replaceOSD: &osd.OSDReplaceInfo{ID: 3, Path: "/dev/sda"}}
+	devices = &DeviceOsdMapping{
+		Entries: map[string]*DeviceOsdIDEntry{
+			"data": {Data: -1, Metadata: nil, Config: DesiredDevice{Name: "/mnt/set1-data-0-rpf2k"}, DeviceInfo: &sys.LocalDisk{RealPath: "/dev/sda"}},
+		},
+	}
+	executor.MockExecuteCommandWithCombinedOutput = func(command string, args ...string) (string, error) {
+		logger.Infof("%s %v", command, args)
+		if args[1] == "ceph-volume" && args[2] == "raw" && args[3] == "prepare" && args[4] == "--bluestore" && args[7] == "--osd-id" && args[8] == "3" {
+			return initializeBlockPVCTestResult, nil
+		}
+
+		return "", errors.Errorf("unknown command %s %s", command, args)
+	}
+	blockPath, metadataBlockPath, walBlockPath, err = a.initializeBlockPVC(context, devices, false)
+	assert.Nil(t, err)
+	assert.Equal(t, "/mnt/set1-data-0-rpf2k", blockPath)
+	assert.Equal(t, "", metadataBlockPath)
+	assert.Equal(t, "", walBlockPath)
+
+	// Test for condition that --osd-id is not passed for the devices that don't match the OSD to be replaced.
+	a = &OsdAgent{clusterInfo: clusterInfo, nodeName: "node1", storeConfig: config.StoreConfig{StoreType: "bluestore"}, replaceOSD: &osd.OSDReplaceInfo{ID: 3, Path: "/dev/sda"}}
+	devices = &DeviceOsdMapping{
+		Entries: map[string]*DeviceOsdIDEntry{
+			"data": {Data: -1, Metadata: nil, Config: DesiredDevice{Name: "/mnt/set1-data-0-rpf2k"}, DeviceInfo: &sys.LocalDisk{RealPath: "/dev/sdb"}},
+		},
+	}
+	executor.MockExecuteCommandWithCombinedOutput = func(command string, args ...string) (string, error) {
+		logger.Infof("%s %v", command, args)
+		if args[1] == "ceph-volume" && args[2] == "raw" && args[3] == "prepare" && args[4] == "--bluestore" && args[7] != "--osd-id" && args[8] != "3" {
+			return initializeBlockPVCTestResult, nil
+		}
+
+		return "", errors.Errorf("unknown command %s %s", command, args)
+	}
+	blockPath, metadataBlockPath, walBlockPath, err = a.initializeBlockPVC(context, devices, false)
+	assert.Nil(t, err)
+	assert.Equal(t, "/mnt/set1-data-0-rpf2k", blockPath)
+	assert.Equal(t, "", metadataBlockPath)
+	assert.Equal(t, "", walBlockPath)
 }
 
 func TestInitializeBlockPVCWithMetadata(t *testing.T) {

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

@@ -18,6 +18,7 @@ package osd
 
 import (
 	"fmt"
+	"strings"
 
 	"github.com/pkg/errors"
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
@@ -180,7 +181,20 @@ func (c *Cluster) startProvisioningOverPVCs(config *provisionConfig, errs *provi
 
 		// Skip OSD prepare if deployment already exists for the PVC
 		// Also skip the encryption work part to avoid overriding the existing encryption key
+		skipPreparePod := false
 		if existingDeployments.Has(dataSource.ClaimName) {
+			skipPreparePod = true
+		}
+
+		// Allow updating OSD prepare pod if the OSD needs migration
+		if c.replaceOSD != nil {
+			if strings.Contains(c.replaceOSD.Path, dataSource.ClaimName) {
+				logger.Infof("updating OSD prepare pod to replace OSD.%d", c.replaceOSD.ID)
+				skipPreparePod = false
+			}
+		}
+
+		if skipPreparePod {
 			logger.Infof("skipping OSD prepare job creation for PVC %q because OSD daemon using the PVC already exists", osdProps.crushHostname)
 			continue
 		}

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

@@ -49,6 +49,7 @@ const (
 	CrushDeviceClassVarName             = "ROOK_OSD_CRUSH_DEVICE_CLASS"
 	CrushInitialWeightVarName           = "ROOK_OSD_CRUSH_INITIAL_WEIGHT"
 	OSDStoreTypeVarName                 = "ROOK_OSD_STORE_TYPE"
+	ReplaceOSDIDVarName                 = "ROOK_REPLACE_OSD"
 	CrushRootVarName                    = "ROOK_CRUSHMAP_ROOT"
 	tcmallocMaxTotalThreadCacheBytesEnv = "TCMALLOC_MAX_TOTAL_THREAD_CACHE_BYTES"
 )
@@ -175,6 +176,10 @@ func osdStoreTypeEnvVar(storeType string) v1.EnvVar {
 	return v1.EnvVar{Name: OSDStoreTypeVarName, Value: storeType}
 }
 
+func replaceOSDIDEnvVar(id string) v1.EnvVar {
+	return v1.EnvVar{Name: ReplaceOSDIDVarName, Value: id}
+}
+
 func crushInitialWeightEnvVar(crushInitialWeight string) v1.EnvVar {
 	return v1.EnvVar{Name: CrushInitialWeightVarName, Value: crushInitialWeight}
 }

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

@@ -82,6 +82,7 @@ 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)
@@ -102,23 +103,6 @@ func (m *OSDHealthMonitor) checkOSDHealth() {
 	if err != nil {
 		logger.Debugf("failed to check OSD Dump. %v", err)
 	}
-	err = m.checkDeviceClasses()
-	if err != nil {
-		logger.Debugf("failed to check device classes. %v", err)
-	}
-}
-
-func (m *OSDHealthMonitor) checkDeviceClasses() error {
-	devices, err := client.GetDeviceClasses(m.context, m.clusterInfo)
-	if err != nil {
-		return errors.Wrap(err, "failed to get osd device classes")
-	}
-
-	if len(devices) > 0 {
-		m.updateCephStatus(devices)
-	}
-
-	return nil
 }
 
 func (m *OSDHealthMonitor) checkOSDDump() error {
@@ -191,15 +175,30 @@ func (m *OSDHealthMonitor) removeOSDDeploymentIfSafeToDestroy(outOSDid int) erro
 	return nil
 }
 
-// updateCephStorage updates the CR with deviceclass details
-func (m *OSDHealthMonitor) updateCephStatus(devices []string) {
+// updateCephStorageStatus updates the Storage details in the ceph cluster CR
+func (m *OSDHealthMonitor) updateCephStorageStatus() {
 	cephCluster := cephv1.CephCluster{}
 	cephClusterStorage := cephv1.CephStorage{}
 
-	for _, device := range devices {
-		cephClusterStorage.DeviceClasses = append(cephClusterStorage.DeviceClasses, cephv1.DeviceClasses{Name: device})
+	deviceClasses, err := client.GetDeviceClasses(m.context, m.clusterInfo)
+	if err != nil {
+		logger.Errorf("failed to get osd device classes. %v", err)
+		return
 	}
-	err := m.context.Client.Get(m.clusterInfo.Context, m.clusterInfo.NamespacedName(), &cephCluster)
+
+	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.")
@@ -216,3 +215,25 @@ func (m *OSDHealthMonitor) updateCephStatus(devices []string) {
 		}
 	}
 }
+
+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

@@ -141,46 +141,103 @@ func TestNewOSDHealthMonitor(t *testing.T) {
 	}
 }
 
-func TestDeviceClasses(t *testing.T) {
+func TestUpdateCephStorageStatus(t *testing.T) {
+	ctx := context.TODO()
 	clusterInfo := client.AdminTestClusterInfo("fake")
-	clusterInfo.SetName("rook-ceph")
-
-	var execCount = 0
 	executor := &exectest.MockExecutor{
 		MockExecuteCommandWithOutput: func(command string, args ...string) (string, error) {
-			return "{\"key\":\"mysecurekey\", \"osdid\":3.0}", nil
+			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
 		},
 	}
-	executor.MockExecuteCommandWithOutput = func(command string, args ...string) (string, error) {
-		logger.Infof("ExecuteCommandWithOutputFile: %s %v", command, args)
-		execCount++
-		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{}
+	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.
-	cl := fake.NewClientBuilder().WithScheme(s).WithRuntimeObjects(object...).Build()
+	c := fake.NewClientBuilder().WithScheme(s).WithRuntimeObjects(object...).Build()
 
 	context := &clusterd.Context{
-		Executor: executor,
-		Client:   cl,
+		Executor:  executor,
+		Client:    c,
+		Clientset: testexec.New(t, 2),
 	}
 
 	// Initializing an OSD monitoring
-	osdMon := NewOSDHealthMonitor(context, clusterInfo, true, cephv1.CephClusterHealthCheckSpec{})
+	osdHealthMonitor := NewOSDHealthMonitor(context, clusterInfo, true, cephv1.CephClusterHealthCheckSpec{})
 
-	// Run OSD monitoring routine
-	err := osdMon.checkDeviceClasses()
-	assert.Nil(t, err)
-	// checkDeviceClasses has 1 mocked cmd for fetching the device classes
-	assert.Equal(t, 1, execCount)
+	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/labels.go

@@ -59,7 +59,7 @@ func (c *Cluster) getOSDLabels(osd OSDInfo, failureDomainValue string, portable
 	labels[FailureDomainKey] = failureDomainValue
 	labels[portableKey] = strconv.FormatBool(portable)
 	labels[deviceClass] = osd.DeviceClass
-	labels[osdStore] = c.spec.Storage.GetOSDStore()
+	labels[osdStore] = osd.Store
 
 	for k, v := range getOSDTopologyLocationLabels(osd.Location) {
 		labels[k] = v

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

@@ -34,6 +34,7 @@ import (
 	"github.com/rook/rook/pkg/operator/ceph/controller"
 	cephver "github.com/rook/rook/pkg/operator/ceph/version"
 	"github.com/rook/rook/pkg/operator/k8sutil"
+	"github.com/rook/rook/pkg/util"
 
 	"github.com/coreos/pkg/capnslog"
 	"github.com/pkg/errors"
@@ -46,7 +47,9 @@ import (
 )
 
 var (
-	logger = capnslog.NewPackageLogger("github.com/rook/rook", "op-osd")
+	logger                   = capnslog.NewPackageLogger("github.com/rook/rook", "op-osd")
+	waitForHealthyPGInterval = 10 * time.Second
+	waitForHealthyPGTimeout  = 15 * time.Minute
 )
 
 const (
@@ -79,6 +82,7 @@ type Cluster struct {
 	ValidStorage cephv1.StorageScopeSpec // valid subset of `Storage`, computed at runtime
 	kv           *k8sutil.ConfigMapKVStore
 	deviceSets   []deviceSet
+	replaceOSD   *OSDReplaceInfo
 }
 
 // New creates an instance of the OSD manager
@@ -112,6 +116,7 @@ type OSDInfo struct {
 	TopologyAffinity string `json:"topologyAffinity"`
 	Encrypted        bool   `json:"encrypted"`
 	ExportService    bool   `json:"exportService"`
+	NodeName         string `json:"nodeName"`
 }
 
 // OrchestrationStatus represents the status of an OSD orchestration
@@ -193,16 +198,24 @@ func (c *Cluster) Start() error {
 	}
 	logger.Infof("wait timeout for healthy OSDs during upgrade or restart is %q", c.clusterInfo.OsdUpgradeTimeout)
 
+	// identify OSDs that need migration to a new backend store
+	err := c.replaceOSDForNewStore()
+	if err != nil {
+		return errors.Wrapf(err, "failed to replace an OSD that needs migration to new backend store in namespace %q", namespace)
+	}
+
+	osdsToSkipReconcile, err := c.getOSDsToSkipReconcile()
+	if err != nil {
+		logger.Warningf("failed to get osds to skip reconcile. %v", err)
+	}
+
 	// prepare for updating existing OSDs
 	updateQueue, deployments, err := c.getOSDUpdateInfo(errs)
 	if err != nil {
 		return errors.Wrapf(err, "failed to get information about currently-running OSD Deployments in namespace %q", namespace)
 	}
-	osdsToSkipReconcile, err := c.getOSDsToSkipReconcile()
-	if err != nil {
-		logger.Warningf("failed to get osds to skip reconcile. %v", err)
-	}
-	logger.Debugf("%d of %d OSD Deployments need updated", updateQueue.Len(), deployments.Len())
+
+	logger.Debugf("%d of %d OSD Deployments need update", updateQueue.Len(), deployments.Len())
 	updateConfig := c.newUpdateConfig(config, updateQueue, deployments, osdsToSkipReconcile)
 
 	// prepare for creating new OSDs
@@ -247,6 +260,22 @@ func (c *Cluster) Start() error {
 		return errors.Wrapf(err, "failed to reconcile key rotation cron jobs")
 	}
 
+	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)
+		if err != nil {
+			return errors.Wrapf(err, "failed to delete the %q configmap", OSDReplaceConfigName)
+		}
+
+		// Wait for PGs to be healthy before continuing the reconcile
+		_, err = c.waitForHealthyPGs()
+		if err != nil {
+			return errors.Wrapf(err, "failed to wait for pgs to be healhty")
+		}
+
+		return errors.New("reconcile operator to replace OSDs that are pending migration")
+	}
+
 	logger.Infof("finished running OSDs in namespace %q", namespace)
 	return nil
 }
@@ -496,6 +525,9 @@ func (c *Cluster) getOSDInfo(d *appsv1.Deployment) (OSDInfo, error) {
 	isPVC := false
 
 	for _, envVar := range d.Spec.Template.Spec.Containers[0].Env {
+		if envVar.Name == "ROOK_NODE_NAME" {
+			osd.NodeName = envVar.Value
+		}
 		if envVar.Name == "ROOK_OSD_UUID" {
 			osd.UUID = envVar.Value
 		}
@@ -577,6 +609,8 @@ func (c *Cluster) getOSDInfo(d *appsv1.Deployment) (OSDInfo, error) {
 		return OSDInfo{}, errors.Errorf("failed to get required osdInfo. %+v", osd)
 	}
 
+	osd.Store = d.Labels[osdStore]
+
 	return osd, nil
 }
 
@@ -779,3 +813,61 @@ func (c *Cluster) applyUpgradeOSDFunctionality() {
 		}
 	}
 }
+
+// replaceOSDForNewStore deletes an existing OSD deployment that does not not match the expected OSD backend store provided in the storage spec
+func (c *Cluster) replaceOSDForNewStore() error {
+	if c.spec.Storage.Store.UpdateStore != OSDStoreUpdateConfirmation {
+		logger.Debugf("no OSD migration to a new backend store is requested")
+		return nil
+	}
+
+	osdToReplace, err := c.getOSDReplaceInfo()
+	if err != nil {
+		return errors.Wrapf(err, "failed to get OSD replace info")
+	}
+
+	if osdToReplace == nil {
+		logger.Info("no osd to replace")
+		return nil
+	}
+
+	logger.Infof("starting migration of the OSD.%d", osdToReplace.ID)
+
+	// Delete the OSD deployment
+	deploymentName := fmt.Sprintf("rook-ceph-osd-%d", osdToReplace.ID)
+	logger.Infof("removing the OSD deployment %q", deploymentName)
+	if err := k8sutil.DeleteDeployment(c.clusterInfo.Context, c.context.Clientset, c.clusterInfo.Namespace, deploymentName); err != nil {
+		if err != nil {
+			if kerrors.IsNotFound(err) {
+				logger.Debugf("osd deployment %q not found. Ignoring since object must be deleted.", deploymentName)
+			} else {
+				return errors.Wrapf(err, "failed to delete OSD deployment %q.", deploymentName)
+			}
+		}
+	}
+
+	c.replaceOSD = osdToReplace
+
+	return osdToReplace.saveAsConfig(c.context, c.clusterInfo)
+}
+
+func (c *Cluster) waitForHealthyPGs() (bool, error) {
+	waitFunc := func() (done bool, err error) {
+		pgHealthMsg, pgClean, err := cephclient.IsClusterClean(c.context, c.clusterInfo)
+		if err != nil {
+			return false, errors.Wrap(err, "failed to check pg are healthy")
+		}
+		if pgClean {
+			return true, nil
+		}
+		logger.Infof("waiting for PGs to be healthy after replacing an OSD, status: %q", pgHealthMsg)
+		return false, nil
+	}
+
+	err := util.RetryWithTimeout(waitFunc, waitForHealthyPGInterval, waitForHealthyPGTimeout, "pgs to be healthy")
+	if err != nil {
+		return false, err
+	}
+
+	return true, nil
+}

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

@@ -18,6 +18,8 @@ package osd
 
 import (
 	"context"
+	"encoding/json"
+	"fmt"
 	"testing"
 
 	"github.com/pkg/errors"
@@ -45,6 +47,11 @@ import (
 	k8stesting "k8s.io/client-go/testing"
 )
 
+const (
+	healthyCephStatus   = `{"fsid":"877a47e0-7f6c-435e-891a-76983ab8c509","health":{"checks":{},"status":"HEALTH_OK"},"election_epoch":12,"quorum":[0,1,2],"quorum_names":["a","b","c"],"monmap":{"epoch":3,"fsid":"877a47e0-7f6c-435e-891a-76983ab8c509","modified":"2020-11-02 09:58:23.015313","created":"2020-11-02 09:57:37.719235","min_mon_release":14,"min_mon_release_name":"nautilus","features":{"persistent":["kraken","luminous","mimic","osdmap-prune","nautilus"],"optional":[]},"mons":[{"rank":0,"name":"a","public_addrs":{"addrvec":[{"type":"v2","addr":"172.30.74.42:3300","nonce":0},{"type":"v1","addr":"172.30.74.42:6789","nonce":0}]},"addr":"172.30.74.42:6789/0","public_addr":"172.30.74.42:6789/0"},{"rank":1,"name":"b","public_addrs":{"addrvec":[{"type":"v2","addr":"172.30.101.61:3300","nonce":0},{"type":"v1","addr":"172.30.101.61:6789","nonce":0}]},"addr":"172.30.101.61:6789/0","public_addr":"172.30.101.61:6789/0"},{"rank":2,"name":"c","public_addrs":{"addrvec":[{"type":"v2","addr":"172.30.250.55:3300","nonce":0},{"type":"v1","addr":"172.30.250.55:6789","nonce":0}]},"addr":"172.30.250.55:6789/0","public_addr":"172.30.250.55:6789/0"}]},"osdmap":{"osdmap":{"epoch":19,"num_osds":3,"num_up_osds":3,"num_in_osds":3,"num_remapped_pgs":0}},"pgmap":{"pgs_by_state":[{"state_name":"active+clean","count":96}],"num_pgs":96,"num_pools":3,"num_objects":79,"data_bytes":81553681,"bytes_used":3255447552,"bytes_avail":1646011994112,"bytes_total":1649267441664,"read_bytes_sec":853,"write_bytes_sec":5118,"read_op_per_sec":1,"write_op_per_sec":0},"fsmap":{"epoch":9,"id":1,"up":1,"in":1,"max":1,"by_rank":[{"filesystem_id":1,"rank":0,"name":"ocs-storagecluster-cephfilesystem-b","status":"up:active","gid":14161},{"filesystem_id":1,"rank":0,"name":"ocs-storagecluster-cephfilesystem-a","status":"up:standby-replay","gid":24146}],"up:standby":0},"mgrmap":{"epoch":10,"active_gid":14122,"active_name":"a","active_addrs":{"addrvec":[{"type":"v2","addr":"10.131.0.28:6800","nonce":1},{"type":"v1","addr":"10.131.0.28:6801","nonce":1}]}}}`
+	unHealthyCephStatus = `{"fsid":"613975f3-3025-4802-9de1-a2280b950e75","health":{"checks":{"OSD_DOWN":{"severity":"HEALTH_WARN","summary":{"message":"1 osds down"}},"OSD_HOST_DOWN":{"severity":"HEALTH_WARN","summary":{"message":"1 host (1 osds) down"}},"PG_AVAILABILITY":{"severity":"HEALTH_WARN","summary":{"message":"Reduced data availability: 101 pgs stale"}},"POOL_APP_NOT_ENABLED":{"severity":"HEALTH_WARN","summary":{"message":"application not enabled on 1 pool(s)"}}},"status":"HEALTH_WARN","overall_status":"HEALTH_WARN"},"election_epoch":12,"quorum":[0,1,2],"quorum_names":["rook-ceph-mon0","rook-ceph-mon2","rook-ceph-mon1"],"monmap":{"epoch":3,"fsid":"613975f3-3025-4802-9de1-a2280b950e75","modified":"2017-08-11 20:13:02.075679","created":"2017-08-11 20:12:35.314510","features":{"persistent":["kraken","luminous"],"optional":[]},"mons":[{"rank":0,"name":"rook-ceph-mon0","addr":"10.3.0.45:6789/0","public_addr":"10.3.0.45:6789/0"},{"rank":1,"name":"rook-ceph-mon2","addr":"10.3.0.249:6789/0","public_addr":"10.3.0.249:6789/0"},{"rank":2,"name":"rook-ceph-mon1","addr":"10.3.0.252:6789/0","public_addr":"10.3.0.252:6789/0"}]},"osdmap":{"osdmap":{"epoch":17,"num_osds":2,"num_up_osds":1,"num_in_osds":2,"full":false,"nearfull":true,"num_remapped_pgs":0}},"pgmap":{"pgs_by_state":[{"state_name":"stale+active+clean","count":101},{"state_name":"active+clean","count":99}],"num_pgs":200,"num_pools":2,"num_objects":243,"data_bytes":976793635,"bytes_used":13611479040,"bytes_avail":19825307648,"bytes_total":33436786688},"fsmap":{"epoch":1,"by_rank":[]},"mgrmap":{"epoch":3,"active_gid":14111,"active_name":"rook-ceph-mgr0","active_addr":"10.2.73.6:6800/9","available":true,"standbys":[],"modules":["restful","status"],"available_modules":["dashboard","prometheus","restful","status","zabbix"]},"servicemap":{"epoch":1,"modified":"0.000000","services":{}}}`
+)
+
 func TestOSDProperties(t *testing.T) {
 	osdProps := []osdProperties{
 		{pvc: corev1.PersistentVolumeClaimVolumeSource{ClaimName: "claim"},
@@ -640,3 +647,169 @@ func TestGetOSDInfoWithCustomRoot(t *testing.T) {
 	_, err := c.getOSDInfo(d3)
 	assert.Error(t, err)
 }
+
+func TestReplaceOSDForNewStore(t *testing.T) {
+	clusterInfo := &cephclient.ClusterInfo{Namespace: "ns", Context: context.TODO()}
+	clusterInfo.SetName("test")
+	clusterInfo.OwnerInfo = cephclient.NewMinimumOwnerInfo(t)
+	executor := &exectest.MockExecutor{
+		MockExecuteCommandWithOutput: func(command string, args ...string) (string, error) {
+			logger.Infof("Command: %s %v", command, args)
+			if args[0] == "status" {
+				return healthyCephStatus, nil
+			}
+			return "", errors.Errorf("unexpected ceph command '%v'", args)
+		},
+	}
+	clientset := fake.NewSimpleClientset()
+	context := &clusterd.Context{
+		Clientset: clientset,
+		Executor:  executor,
+	}
+
+	t.Run("no osd migration is requested in the cephcluster spec", func(t *testing.T) {
+		spec := cephv1.ClusterSpec{
+			Storage: cephv1.StorageScopeSpec{
+				Store: cephv1.OSDStore{},
+			},
+		}
+		c := New(context, clusterInfo, spec, "myversion")
+		err := c.replaceOSDForNewStore()
+		assert.NoError(t, err)
+		assert.Nil(t, c.replaceOSD)
+	})
+
+	t.Run("migration is requested but no osd pods are running", func(t *testing.T) {
+		spec := cephv1.ClusterSpec{
+			Storage: cephv1.StorageScopeSpec{
+				Store: cephv1.OSDStore{
+					Type:        "bluestore-rdr",
+					UpdateStore: "yes-really-update-store",
+				},
+			},
+		}
+		c := New(context, clusterInfo, spec, "myversion")
+		err := c.replaceOSDForNewStore()
+		assert.NoError(t, err)
+		assert.Nil(t, c.replaceOSD)
+	})
+
+	t.Run("migration is requested but all OSDs are running on expected backed store", func(t *testing.T) {
+		spec := cephv1.ClusterSpec{
+			Storage: cephv1.StorageScopeSpec{
+				Store: cephv1.OSDStore{
+					Type:        "bluestore-rdr",
+					UpdateStore: "yes-really-update-store",
+				},
+			},
+		}
+		c := New(context, clusterInfo, spec, "myversion")
+		d := getDummyDeploymentOnNode(clientset, c, "node2", 0)
+		d.Labels[osdStore] = "bluestore-rdr"
+		createDeploymentOrPanic(clientset, d)
+		err := c.replaceOSDForNewStore()
+		assert.NoError(t, err)
+		assert.Nil(t, c.replaceOSD)
+	})
+
+	t.Run("migration is requested and one OSD on node is running legacy backend store", func(t *testing.T) {
+		spec := cephv1.ClusterSpec{
+			Storage: cephv1.StorageScopeSpec{
+				Store: cephv1.OSDStore{
+					Type:        "bluestore-rdr",
+					UpdateStore: "yes-really-update-store",
+				},
+			},
+		}
+		c := New(context, clusterInfo, spec, "myversion")
+		// create osd deployment with `bluestore` backend store
+		d := getDummyDeploymentOnNode(clientset, c, "node2", 1)
+		createDeploymentOrPanic(clientset, d)
+		err := c.replaceOSDForNewStore()
+		assert.NoError(t, err)
+		assert.NotNil(t, c.replaceOSD)
+		assert.Equal(t, 1, c.replaceOSD.ID)
+		assert.Equal(t, "node2", c.replaceOSD.Node)
+
+		// assert that OSD.1 deployment got deleted
+		_, err = clientset.AppsV1().Deployments(clusterInfo.Namespace).Get(clusterInfo.Context, deploymentName(1), metav1.GetOptions{})
+		assert.Equal(t, true, k8serrors.IsNotFound(err))
+
+		// validate the osd replace config map
+		actualCM, err := clientset.CoreV1().ConfigMaps(clusterInfo.Namespace).Get(clusterInfo.Context, OSDReplaceConfigName, metav1.GetOptions{})
+		assert.NoError(t, err)
+		assert.NotNil(t, actualCM)
+		expectedOSDInfo := OSDReplaceInfo{}
+		err = json.Unmarshal([]byte(actualCM.Data[OSDReplaceConfigKey]), &expectedOSDInfo)
+		assert.NoError(t, err)
+		assert.Equal(t, 1, expectedOSDInfo.ID)
+		assert.Equal(t, "node2", expectedOSDInfo.Node)
+
+		// delete configmap
+		err = k8sutil.DeleteConfigMap(clusterInfo.Context, clientset, OSDReplaceConfigName, clusterInfo.Namespace, &k8sutil.DeleteOptions{})
+		assert.NoError(t, err)
+	})
+
+	t.Run("migration is requested and one osd on pvc is running on legacy backend store", func(t *testing.T) {
+		spec := cephv1.ClusterSpec{
+			Storage: cephv1.StorageScopeSpec{
+				Store: cephv1.OSDStore{
+					Type:        "bluestore-rdr",
+					UpdateStore: "yes-really-update-store",
+				},
+			},
+		}
+		c := New(context, clusterInfo, spec, "myversion")
+		d := getDummyDeploymentOnPVC(clientset, c, "pvc1", 2)
+		createDeploymentOrPanic(clientset, d)
+		err := c.replaceOSDForNewStore()
+		assert.NoError(t, err)
+		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 that OSD.2 deployment got deleted
+		_, err = clientset.AppsV1().Deployments(clusterInfo.Namespace).Get(clusterInfo.Context, deploymentName(2), metav1.GetOptions{})
+		assert.Equal(t, true, k8serrors.IsNotFound(err))
+
+		// validate the osd replace config map
+		actualCM, err := clientset.CoreV1().ConfigMaps(clusterInfo.Namespace).Get(clusterInfo.Context, OSDReplaceConfigName, metav1.GetOptions{})
+		assert.NoError(t, err)
+		assert.NotNil(t, actualCM)
+		expectedOSDInfo := OSDReplaceInfo{}
+		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)
+
+		// delete configmap
+		err = k8sutil.DeleteConfigMap(clusterInfo.Context, clientset, OSDReplaceConfigName, clusterInfo.Namespace, &k8sutil.DeleteOptions{})
+		assert.NoError(t, err)
+	})
+
+	t.Run("migration is requested but pgs are not clean", func(t *testing.T) {
+		spec := cephv1.ClusterSpec{
+			Storage: cephv1.StorageScopeSpec{
+				Store: cephv1.OSDStore{
+					Type:        "bluestore-rdr",
+					UpdateStore: "yes-really-update-store",
+				},
+			},
+		}
+		executor := &exectest.MockExecutor{
+			MockExecuteCommandWithOutput: func(command string, args ...string) (string, error) {
+				logger.Infof("Command: %s %v", command, args)
+				if args[0] == "status" {
+					return unHealthyCephStatus, nil
+				}
+				return "", errors.Errorf("unexpected ceph command '%v'", args)
+			},
+		}
+		context.Executor = executor
+		c := New(context, clusterInfo, spec, "myversion")
+		err := c.replaceOSDForNewStore()
+		assert.NoError(t, err)
+		assert.Nil(t, c.replaceOSD)
+	})
+}

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

@@ -20,6 +20,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"path"
+	"strings"
 
 	"github.com/pkg/errors"
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
@@ -303,6 +304,22 @@ func (c *Cluster) provisionOSDContainer(osdProps osdProperties, copyBinariesMoun
 		volumeMounts = append(volumeMounts, v1.VolumeMount{Name: "rootfs", MountPath: "/rootfs", ReadOnly: true})
 	}
 
+	// Add OSD ID as environment variables.
+	// When this env is set, prepare pod job will destroy this OSD.
+	if c.replaceOSD != nil {
+		// Compare pvc claim name in case of OSDs on PVC
+		if osdProps.onPVC() {
+			if strings.Contains(c.replaceOSD.Path, osdProps.pvc.ClaimName) {
+				envVars = append(envVars, replaceOSDIDEnvVar(fmt.Sprint(c.replaceOSD.ID)))
+			}
+		} else {
+			// Compare the node name in case of OSDs on disk
+			if c.replaceOSD.Node == osdProps.crushHostname {
+				envVars = append(envVars, replaceOSDIDEnvVar(fmt.Sprint(c.replaceOSD.ID)))
+			}
+		}
+	}
+
 	// run privileged always since we always mount /dev
 	privileged := true
 	runAsUser := int64(0)

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

@@ -0,0 +1,167 @@
+/*
+Copyright 2023 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 osd for the Ceph OSDs.
+package osd
+
+import (
+	"encoding/json"
+	"fmt"
+
+	"github.com/pkg/errors"
+	"github.com/rook/rook/pkg/clusterd"
+	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
+	"github.com/rook/rook/pkg/operator/k8sutil"
+	v1 "k8s.io/api/core/v1"
+	kerrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+const (
+	OSDReplaceConfigName       = "osd-replace-config"
+	OSDReplaceConfigKey        = "config"
+	OSDStoreUpdateConfirmation = "yes-really-update-store"
+)
+
+// OSDReplaceInfo represents an OSD that needs to replaced
+type OSDReplaceInfo struct {
+	ID   int    `json:"id"`
+	Path string `json:"path"`
+	Node string `json:"node"`
+}
+
+func (o *OSDReplaceInfo) saveAsConfig(context *clusterd.Context, clusterInfo *cephclient.ClusterInfo) error {
+	configStr, err := o.string()
+	if err != nil {
+		return errors.Wrapf(err, "failed to convert osd replace config to string")
+	}
+
+	newConfigMap := &v1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      OSDReplaceConfigName,
+			Namespace: clusterInfo.Namespace,
+		},
+		Data: map[string]string{
+			OSDReplaceConfigKey: configStr,
+		},
+	}
+
+	err = clusterInfo.OwnerInfo.SetControllerReference(newConfigMap)
+	if err != nil {
+		return errors.Wrapf(err, "failed to set owner reference on %q configMap", newConfigMap.Name)
+	}
+
+	_, err = k8sutil.CreateOrUpdateConfigMap(clusterInfo.Context, context.Clientset, newConfigMap)
+	if err != nil {
+		return errors.Wrapf(err, "failed to create or update %q configMap", newConfigMap.Name)
+	}
+
+	return nil
+}
+
+func (o *OSDReplaceInfo) string() (string, error) {
+	configInBytes, err := json.Marshal(o)
+	if err != nil {
+		return "", errors.Wrap(err, "failed to marshal osd replace config")
+	}
+
+	return string(configInBytes), nil
+}
+
+// getOSDReplaceInfo returns an existing OSD that needs to be replaced for a new backend store
+func (c *Cluster) getOSDReplaceInfo() (*OSDReplaceInfo, error) {
+	osdReplaceInfo, err := GetOSDReplaceConfigMap(c.context, c.clusterInfo)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to get any existing OSD in replace configmap")
+	}
+
+	if osdReplaceInfo != nil {
+		return osdReplaceInfo, nil
+	}
+
+	pgHealthMsg, pgClean, err := cephclient.IsClusterClean(c.context, c.clusterInfo)
+	if err != nil {
+		return nil, errors.Wrapf(err, "failed to check if the pgs are clean before replacing OSDs")
+	}
+	if !pgClean {
+		logger.Warningf("skipping OSD replacement because pgs are not healthy. PG status: %q", pgHealthMsg)
+		return nil, nil
+	}
+
+	logger.Infof("placement group status: %q", pgHealthMsg)
+
+	osdsToReplace, err := c.getOSDWithNonMatchingStore()
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to list out OSDs with non matching backend store")
+	}
+
+	if len(osdsToReplace) == 0 {
+		logger.Infof("all osds have already been migrated to backend store %q", c.spec.Storage.Store.Type)
+		return nil, nil
+	}
+
+	logger.Infof("%d osd(s) require migration to new backend store %q.", len(osdsToReplace), c.spec.Storage.Store.Type)
+
+	return &osdsToReplace[0], nil
+}
+
+// getOSDWithNonMatchingStore returns OSDs with osd-store label different from expected store in cephCluster spec
+func (c *Cluster) getOSDWithNonMatchingStore() ([]OSDReplaceInfo, error) {
+	listOpts := metav1.ListOptions{LabelSelector: fmt.Sprintf("%s=%s", k8sutil.AppAttr, AppName)}
+	deployments, err := c.context.Clientset.AppsV1().Deployments(c.clusterInfo.Namespace).List(c.clusterInfo.Context, listOpts)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to query OSDs to skip reconcile")
+	}
+
+	osdReplaceList := []OSDReplaceInfo{}
+	for i := range deployments.Items {
+		if osdStore, ok := deployments.Items[i].Labels[osdStore]; ok {
+			if osdStore != string(c.spec.Storage.Store.Type) {
+				osdInfo, err := c.getOSDInfo(&deployments.Items[i])
+				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})
+			}
+		}
+	}
+
+	return osdReplaceList, nil
+}
+
+// GetOSDReplaceConfigMap returns the OSD replace config map
+func GetOSDReplaceConfigMap(context *clusterd.Context, clusterInfo *cephclient.ClusterInfo) (*OSDReplaceInfo, error) {
+	cm, err := context.Clientset.CoreV1().ConfigMaps(clusterInfo.Namespace).Get(clusterInfo.Context, OSDReplaceConfigName, metav1.GetOptions{})
+	if err != nil {
+		if kerrors.IsNotFound(err) {
+			return nil, nil
+		}
+	}
+
+	configStr, ok := cm.Data[OSDReplaceConfigKey]
+	if !ok || configStr == "" {
+		logger.Debugf("empty config map %q", OSDReplaceConfigName)
+		return nil, nil
+	}
+
+	config := &OSDReplaceInfo{}
+	err = json.Unmarshal([]byte(configStr), config)
+	if err != nil {
+		return nil, errors.Wrapf(err, "failed to JSON unmarshal osd replace status from the (%q)", configStr)
+	}
+
+	return config, nil
+}

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

@@ -0,0 +1,287 @@
+/*
+Copyright 2023 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 osd for the Ceph OSDs.
+package osd
+
+import (
+	"context"
+	"testing"
+
+	"github.com/pkg/errors"
+	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
+	"github.com/rook/rook/pkg/clusterd"
+	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
+	"github.com/rook/rook/pkg/operator/k8sutil"
+	exectest "github.com/rook/rook/pkg/util/exec/test"
+	"github.com/stretchr/testify/assert"
+	appsv1 "k8s.io/api/apps/v1"
+	"k8s.io/client-go/kubernetes/fake"
+)
+
+func TestGetOSDWithNonMatchingStoreOnNodes(t *testing.T) {
+	namespace := "rook-ceph"
+	namespace2 := "rook-ceph2"
+	clientset := fake.NewSimpleClientset()
+	ctx := &clusterd.Context{
+		Clientset: clientset,
+	}
+	clusterInfo := &cephclient.ClusterInfo{
+		Namespace: namespace,
+		Context:   context.TODO(),
+	}
+	clusterInfo.SetName("mycluster")
+	clusterInfo.OwnerInfo = cephclient.NewMinimumOwnerInfo(t)
+	spec := cephv1.ClusterSpec{
+		Storage: cephv1.StorageScopeSpec{
+			Store: cephv1.OSDStore{
+				Type: "bluestore-rdr",
+			},
+		},
+	}
+	c := New(ctx, clusterInfo, spec, "rook/rook:master")
+
+	var d *appsv1.Deployment
+
+	t.Run("all osd deployments are running on bluestore-rdr osd store", func(t *testing.T) {
+		d = getDummyDeploymentOnNode(clientset, c, "node2", 0)
+		d.Labels[osdStore] = "bluestore-rdr"
+		createDeploymentOrPanic(clientset, d)
+
+		d = getDummyDeploymentOnNode(clientset, c, "node3", 1)
+		d.Labels[osdStore] = "bluestore-rdr"
+		createDeploymentOrPanic(clientset, d)
+
+		d = getDummyDeploymentOnNode(clientset, c, "node4", 2)
+		d.Labels[osdStore] = "bluestore-rdr"
+		createDeploymentOrPanic(clientset, d)
+
+		osdList, err := c.getOSDWithNonMatchingStore()
+		assert.NoError(t, err)
+		assert.Equal(t, 0, len(osdList))
+	})
+
+	t.Run("all osd deployments are not running on bluestore-rdr store", func(t *testing.T) {
+		c.clusterInfo.Namespace = namespace2
+
+		// osd.0 is still using bluestore
+		d = getDummyDeploymentOnNode(clientset, c, "node2", 0)
+		createDeploymentOrPanic(clientset, d)
+
+		// osd.1 and osd.2 are using `bluestore-rdr`
+		d = getDummyDeploymentOnNode(clientset, c, "node3", 1)
+		d.Labels[osdStore] = "bluestore-rdr"
+		createDeploymentOrPanic(clientset, d)
+
+		d = getDummyDeploymentOnNode(clientset, c, "node4", 2)
+		d.Labels[osdStore] = "bluestore-rdr"
+		createDeploymentOrPanic(clientset, d)
+
+		osdList, err := c.getOSDWithNonMatchingStore()
+		assert.NoError(t, err)
+		assert.Equal(t, 1, len(osdList))
+		assert.Equal(t, 0, osdList[0].ID)
+		assert.Equal(t, "node2", osdList[0].Node)
+		assert.Equal(t, "/dev/vda", osdList[0].Path)
+	})
+}
+
+func TestGetOSDWithNonMatchingStoreOnPVCs(t *testing.T) {
+	namespace := "rook-ceph"
+	namespace2 := "rook-ceph2"
+	clientset := fake.NewSimpleClientset()
+	ctx := &clusterd.Context{
+		Clientset: clientset,
+	}
+	clusterInfo := &cephclient.ClusterInfo{
+		Namespace: namespace,
+		Context:   context.TODO(),
+	}
+	clusterInfo.SetName("mycluster")
+	clusterInfo.OwnerInfo = cephclient.NewMinimumOwnerInfo(t)
+	spec := cephv1.ClusterSpec{
+		Storage: cephv1.StorageScopeSpec{
+			Store: cephv1.OSDStore{
+				Type: "bluestore-rdr",
+			},
+		},
+	}
+	c := New(ctx, clusterInfo, spec, "rook/rook:master")
+
+	var d *appsv1.Deployment
+
+	t.Run("all osd deployments are running on bluestore-rdr osd store", func(t *testing.T) {
+		d = getDummyDeploymentOnPVC(clientset, c, "pvc0", 0)
+		d.Labels[osdStore] = "bluestore-rdr"
+		createDeploymentOrPanic(clientset, d)
+
+		d = getDummyDeploymentOnPVC(clientset, c, "pvc1", 1)
+		d.Labels[osdStore] = "bluestore-rdr"
+		createDeploymentOrPanic(clientset, d)
+
+		d = getDummyDeploymentOnPVC(clientset, c, "pvc2", 2)
+		d.Labels[osdStore] = "bluestore-rdr"
+		createDeploymentOrPanic(clientset, d)
+
+		osdList, err := c.getOSDWithNonMatchingStore()
+		assert.NoError(t, err)
+		assert.Equal(t, 0, len(osdList))
+	})
+
+	t.Run("all osd deployments are not running on bluestore-rdr store", func(t *testing.T) {
+		c.clusterInfo.Namespace = namespace2
+
+		// osd.0 is still using `bluestore`
+		d = getDummyDeploymentOnPVC(clientset, c, "pvc0", 0)
+		createDeploymentOrPanic(clientset, d)
+
+		d = getDummyDeploymentOnPVC(clientset, c, "pvc1", 1)
+		d.Labels[osdStore] = "bluestore-rdr"
+		createDeploymentOrPanic(clientset, d)
+
+		d = getDummyDeploymentOnPVC(clientset, c, "pvc2", 2)
+		d.Labels[osdStore] = "bluestore-rdr"
+		createDeploymentOrPanic(clientset, d)
+
+		osdList, err := c.getOSDWithNonMatchingStore()
+		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)
+	})
+}
+
+func TestGetOSDReplaceInfo(t *testing.T) {
+	namespace := "rook-ceph"
+	clientset := fake.NewSimpleClientset()
+	executor := &exectest.MockExecutor{
+		MockExecuteCommandWithOutput: func(command string, args ...string) (string, error) {
+			logger.Infof("Command: %s %v", command, args)
+			if args[0] == "status" {
+				return healthyCephStatus, nil
+			}
+			return "", errors.Errorf("unexpected ceph command '%v'", args)
+		},
+	}
+	ctx := &clusterd.Context{
+		Clientset: clientset,
+		Executor:  executor,
+	}
+	clusterInfo := &cephclient.ClusterInfo{
+		Namespace: namespace,
+		Context:   context.TODO(),
+	}
+	clusterInfo.SetName("mycluster")
+	clusterInfo.OwnerInfo = cephclient.NewMinimumOwnerInfo(t)
+
+	spec := cephv1.ClusterSpec{}
+	c := New(ctx, clusterInfo, spec, "rook/rook:master")
+
+	t.Run("no OSD replace info available", func(t *testing.T) {
+		actualOSDInfo, err := c.getOSDReplaceInfo()
+		assert.NoError(t, err)
+		assert.Nil(t, actualOSDInfo)
+	})
+
+	t.Run("ensure no OSD to replace if pgs are not healthy", func(t *testing.T) {
+		executor := &exectest.MockExecutor{
+			MockExecuteCommandWithOutput: func(command string, args ...string) (string, error) {
+				logger.Infof("Command: %s %v", command, args)
+				if args[0] == "status" {
+					return unHealthyCephStatus, nil
+				}
+				return "", errors.Errorf("unexpected ceph command '%v'", args)
+			},
+		}
+
+		ctx.Executor = executor
+		c := New(ctx, clusterInfo, spec, "rook/rook:master")
+		actualOSDInfo, err := c.getOSDReplaceInfo()
+		assert.NoError(t, err)
+		assert.Nil(t, actualOSDInfo)
+	})
+
+	t.Run("read OSD 0 replace info from config map", func(t *testing.T) {
+		actualOSDInfo := &OSDReplaceInfo{ID: 0, Path: "/dev/sda", Node: "node1"}
+		err := actualOSDInfo.saveAsConfig(c.context, c.clusterInfo)
+		assert.NoError(t, err)
+		expectedOSDInfo, err := c.getOSDReplaceInfo()
+		assert.NoError(t, err)
+		assert.NotNil(t, expectedOSDInfo)
+		assert.Equal(t, 0, expectedOSDInfo.ID)
+		err = k8sutil.DeleteConfigMap(clusterInfo.Context, ctx.Clientset, OSDReplaceConfigName, namespace, &k8sutil.DeleteOptions{})
+		assert.NoError(t, err)
+	})
+
+	t.Run("ensure no OSD replace info if all OSDs using expected OSD store", func(t *testing.T) {
+		spec := cephv1.ClusterSpec{
+			Storage: cephv1.StorageScopeSpec{
+				Store: cephv1.OSDStore{
+					Type:        "bluestore-rdr",
+					UpdateStore: "yes-really-update-store",
+				},
+			},
+		}
+		executor := &exectest.MockExecutor{
+			MockExecuteCommandWithOutput: func(command string, args ...string) (string, error) {
+				logger.Infof("Command: %s %v", command, args)
+				if args[0] == "status" {
+					return healthyCephStatus, nil
+				}
+				return "", errors.Errorf("unexpected ceph command '%v'", args)
+			},
+		}
+
+		ctx.Executor = executor
+		c := New(ctx, clusterInfo, spec, "myversion")
+		d := getDummyDeploymentOnNode(clientset, c, "node2", 1)
+		d.Labels[osdStore] = "bluestore-rdr"
+		createDeploymentOrPanic(clientset, d)
+		expectedOSDInfo, err := c.getOSDReplaceInfo()
+		assert.NoError(t, err)
+		assert.Nil(t, expectedOSDInfo)
+	})
+
+	t.Run("ensure OSD replace info if OSD store is not up to date", func(t *testing.T) {
+		spec := cephv1.ClusterSpec{
+			Storage: cephv1.StorageScopeSpec{
+				Store: cephv1.OSDStore{
+					Type:        "bluestore-rdr",
+					UpdateStore: "yes-really-update-store",
+				},
+			},
+		}
+		executor := &exectest.MockExecutor{
+			MockExecuteCommandWithOutput: func(command string, args ...string) (string, error) {
+				logger.Infof("Command: %s %v", command, args)
+				if args[0] == "status" {
+					return healthyCephStatus, nil
+				}
+				return "", errors.Errorf("unexpected ceph command '%v'", args)
+			},
+		}
+
+		ctx.Executor = executor
+		c := New(ctx, clusterInfo, spec, "myversion")
+		d := getDummyDeploymentOnNode(clientset, c, "node3", 2)
+		createDeploymentOrPanic(clientset, d)
+		expectedOSDInfo, err := c.getOSDReplaceInfo()
+		assert.NoError(t, err)
+		assert.NotNil(t, expectedOSDInfo)
+		assert.Equal(t, 2, expectedOSDInfo.ID)
+		assert.Equal(t, "node3", expectedOSDInfo.Node)
+	})
+}

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

@@ -718,6 +718,7 @@ func getDummyDeploymentOnPVC(clientset *fake.Clientset, c *Cluster, pvcName stri
 		UUID:      "some-uuid",
 		BlockPath: "/some/path",
 		CVMode:    "raw",
+		Store:     "bluestore",
 	}
 	c.deviceSets = append(c.deviceSets, deviceSet{
 		Name: pvcName,
@@ -741,6 +742,7 @@ func getDummyDeploymentOnNode(clientset *fake.Clientset, c *Cluster, nodeName st
 		UUID:      "some-uuid",
 		BlockPath: "/dev/vda",
 		CVMode:    "raw",
+		Store:     "bluestore",
 	}
 	c.ValidStorage.Nodes = append(c.ValidStorage.Nodes, cephv1.Node{Name: nodeName})
 	config := c.newProvisionConfig()

pkg/operator/k8sutil/configmap.go

@@ -23,7 +23,10 @@ import (
 	"sync"
 	"time"
 
+	"github.com/pkg/errors"
+	v1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	kerrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
 )
@@ -43,6 +46,31 @@ func DeleteConfigMap(ctx context.Context, clientset kubernetes.Interface, cmName
 	return DeleteResource(delete, verify, resource, opts, defaultWaitOptions)
 }
 
+func CreateOrUpdateConfigMap(ctx context.Context, clientset kubernetes.Interface, cm *v1.ConfigMap) (*v1.ConfigMap, error) {
+	name := cm.GetName()
+	namespace := cm.GetNamespace()
+	existingCm, err := clientset.CoreV1().ConfigMaps(namespace).Get(ctx, name, metav1.GetOptions{})
+	if err != nil {
+		if kerrors.IsNotFound(err) {
+			cm, err := clientset.CoreV1().ConfigMaps(namespace).Create(ctx, cm, metav1.CreateOptions{})
+			if err != nil {
+				return nil, errors.Wrapf(err, "failed to create %q configmap", name)
+			}
+			return cm, nil
+		}
+
+		return nil, errors.Wrapf(err, "failed to retrieve %q configmap.", name)
+	}
+
+	existingCm.Data = cm.Data
+	existingCm.OwnerReferences = cm.OwnerReferences
+	if existingCm, err := clientset.CoreV1().ConfigMaps(namespace).Update(ctx, existingCm, metav1.UpdateOptions{}); err != nil {
+		return nil, errors.Wrapf(err, "failed to update existing %q configmap", existingCm.Name)
+	}
+
+	return existingCm, nil
+}
+
 // GetOperatorSetting gets the operator setting from ConfigMap or Env Var
 // returns defaultValue if setting is not found
 func GetOperatorSetting(context context.Context, clientset kubernetes.Interface, configMapName, settingName, defaultValue string) (string, error) {

pkg/operator/k8sutil/configmap_test.go

@@ -137,3 +137,35 @@ func TestLogChangedSettings(t *testing.T) {
 	ok = logChangedSettings(key, value)
 	assert.True(t, true, ok)
 }
+
+func TestCreateOrUpdateConfigMap(t *testing.T) {
+	k8s := fake.NewSimpleClientset()
+	ctx := context.TODO()
+
+	cm := &v1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "test-configmap",
+			Namespace: "test-namespace",
+		},
+		Data: map[string]string{
+			"test": "data",
+		},
+	}
+
+	_, err := CreateOrUpdateConfigMap(ctx, k8s, cm)
+	assert.NoError(t, err)
+
+	actualCM, err := k8s.CoreV1().ConfigMaps("test-namespace").Get(ctx, "test-configmap", metav1.GetOptions{})
+	assert.NoError(t, err)
+	assert.NotNil(t, actualCM)
+	assert.Equal(t, "data", actualCM.Data["test"])
+
+	// update config map
+	cm.Data["test"] = "updatedData"
+	_, err = CreateOrUpdateConfigMap(ctx, k8s, cm)
+	assert.NoError(t, err)
+	actualCM, err = k8s.CoreV1().ConfigMaps("test-namespace").Get(ctx, "test-configmap", metav1.GetOptions{})
+	assert.NoError(t, err)
+	assert.NotNil(t, actualCM)
+	assert.Equal(t, "updatedData", actualCM.Data["test"])
+}