Merge pull request #14776 from sp98/osd-migration

migrate OSDs to enable encryption as day-2 operation

Documentation/CRDs/Cluster/ceph-cluster-crd.md

@@ -357,6 +357,7 @@ The following storage selection settings are specific to Ceph and do not apply t
 * `encryptedDevice`**: Encrypt OSD volumes using dmcrypt ("true" or "false"). By default this option is disabled. See [encryption](http://docs.ceph.com/docs/master/ceph-volume/lvm/encryption/) for more information on encryption in Ceph. (Resizing is not supported for host-based clusters.)
 * `crushRoot`: The value of the `root` CRUSH map label. The default is `default`. Generally, you should not need to change this. However, if any of your topology labels may have the value `default`, you need to change `crushRoot` to avoid conflicts, since CRUSH map values need to be unique.
 * `enableCrushUpdates`: Enables rook to update the pool crush rule using Pool Spec. Can cause data remapping if crush rule changes, Defaults to false.
+* `migration`: Existing PVC based OSDs can be migrated to enable or disable encryption. Refer to the [osd management](../../Storage-Configuration/Advanced/ceph-osd-mgmt.md/#osd-encryption-as-day-2-operation) topic for details.
 
 Allowed configurations are:
 

Documentation/CRDs/specification.md

@@ -8049,6 +8049,65 @@ bool
 </tr>
 </tbody>
 </table>
+<h3 id="ceph.rook.io/v1.Migration">Migration
+</h3>
+<p>
+(<em>Appears on:</em><a href="#ceph.rook.io/v1.StorageScopeSpec">StorageScopeSpec</a>)
+</p>
+<div>
+<p>Migration handles the OSD migration</p>
+</div>
+<table>
+<thead>
+<tr>
+<th>Field</th>
+<th>Description</th>
+</tr>
+</thead>
+<tbody>
+<tr>
+<td>
+<code>confirmation</code><br/>
+<em>
+string
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>A user confirmation to migrate the OSDs. It destroys each OSD one at a time, cleans up the backing disk
+and prepares OSD with same ID on that disk</p>
+</td>
+</tr>
+</tbody>
+</table>
+<h3 id="ceph.rook.io/v1.MigrationStatus">MigrationStatus
+</h3>
+<p>
+(<em>Appears on:</em><a href="#ceph.rook.io/v1.OSDStatus">OSDStatus</a>)
+</p>
+<div>
+<p>MigrationStatus status represents the current status of any OSD migration.</p>
+</div>
+<table>
+<thead>
+<tr>
+<th>Field</th>
+<th>Description</th>
+</tr>
+</thead>
+<tbody>
+<tr>
+<td>
+<code>pending</code><br/>
+<em>
+int
+</em>
+</td>
+<td>
+</td>
+</tr>
+</tbody>
+</table>
 <h3 id="ceph.rook.io/v1.MirrorHealthCheckSpec">MirrorHealthCheckSpec
 </h3>
 <p>
@@ -9471,6 +9530,18 @@ map[string]int
 <p>StoreType is a mapping between the OSD backend stores and number of OSDs using these stores</p>
 </td>
 </tr>
+<tr>
+<td>
+<code>migrationStatus</code><br/>
+<em>
+<a href="#ceph.rook.io/v1.MigrationStatus">
+MigrationStatus
+</a>
+</em>
+</td>
+<td>
+</td>
+</tr>
 </tbody>
 </table>
 <h3 id="ceph.rook.io/v1.OSDStore">OSDStore
@@ -12935,6 +13006,20 @@ Selection
 </tr>
 <tr>
 <td>
+<code>migration</code><br/>
+<em>
+<a href="#ceph.rook.io/v1.Migration">
+Migration
+</a>
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>Migration handles the OSD migration</p>
+</td>
+</tr>
+<tr>
+<td>
 <code>store</code><br/>
 <em>
 <a href="#ceph.rook.io/v1.OSDStore">

Documentation/Storage-Configuration/Advanced/ceph-osd-mgmt.md

@@ -190,3 +190,29 @@ If you don't see a new OSD automatically created, restart the operator (by delet
 
 !!! note
     The OSD might have a different ID than the previous OSD that was replaced.
+
+
+## OSD Migration
+
+Ceph does not support changing certain settings on existing OSDs. To support changing these settings on an OSD, the OSD must be destroyed and re-created with the new settings. Rook will automate this by migrating only one OSD at a time. The operator waits for the data to rebalance (PGs to become `active+clean`) before migrating the next OSD. This ensures that there is no data loss. Refer to the [OSD migration](https://github.com/rook/rook/blob/master/design/ceph/osd-migration.md) design doc for more information. 
+
+The following scenarios are supported for OSD migration:
+
+- Enable or disable OSD encryption for existing PVC-based OSDs by changing the `encrypted` setting under the `storageClassDeviceSets`
+
+For example:
+
+```yaml
+storage:
+    migration:
+        confirmation: "yes-really-migrate-osds" 
+    storageClassDeviceSets:
+        - name: set1
+          count: 3
+          encrypted: true  # change to true or false based on whether encryption needs to enable or disabled.
+```
+
+Details about the migration status can be found under the cephCluster `status.storage.osd.migrationStatus.pending` field which shows the total number of OSDs that are pending migration.
+
+!!! note
+    Performance of the cluster might be impacted during data rebalancing while OSDs are being migrated.

PendingReleaseNotes.md

@@ -8,3 +8,4 @@
 
 - Enable mirroring for CephBlockPoolRadosNamespaces (see [#14701](https://github.com/rook/rook/pull/14701)).
 - Enable periodic monitoring for CephBlockPoolRadosNamespaces mirroring (see [#14896](https://github.com/rook/rook/pull/14896)).
+- Allow migration of PVC based OSDs to enable or disable encryption (see [#14776](https://github.com/rook/rook/pull/14776)).

cmd/rook/ceph/osd.go

@@ -261,10 +261,10 @@ func prepareOSD(cmd *cobra.Command, args []string) error {
 	}
 
 	// destroy the OSD using the OSD ID
-	var replaceOSD *oposd.OSDReplaceInfo
+	var replaceOSD *oposd.OSDInfo
 	if replaceOSDID != -1 {
 		logger.Infof("destroying osd.%d and cleaning its backing device", replaceOSDID)
-		replaceOSD, err = osddaemon.DestroyOSD(context, &clusterInfo, replaceOSDID, cfg.pvcBacked, cfg.storeConfig.EncryptedDevice)
+		replaceOSD, err = osddaemon.DestroyOSD(context, &clusterInfo, replaceOSDID, cfg.pvcBacked)
 		if err != nil {
 			rook.TerminateFatal(errors.Wrapf(err, "failed to destroy OSD %d.", replaceOSDID))
 		}

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

@@ -3445,6 +3445,16 @@ spec:
                       minimum: 0
                       nullable: true
                       type: number
+                    migration:
+                      description: Migration handles the OSD migration
+                      properties:
+                        confirmation:
+                          description: |-
+                            A user confirmation to migrate the OSDs. It destroys each OSD one at a time, cleans up the backing disk
+                            and prepares OSD with same ID on that disk
+                          pattern: ^$|^yes-really-migrate-osds$
+                          type: string
+                      type: object
                     nearFullRatio:
                       description: NearFullRatio is the ratio at which the cluster is considered nearly full and will raise a ceph health warning. Default is 0.85.
                       maximum: 1
@@ -5538,6 +5548,12 @@ spec:
                     osd:
                       description: OSDStatus represents OSD status of the ceph Cluster
                       properties:
+                        migrationStatus:
+                          description: MigrationStatus status represents the current status of any OSD migration.
+                          properties:
+                            pending:
+                              type: integer
+                          type: object
                         storeType:
                           additionalProperties:
                             type: integer

deploy/examples/crds.yaml

@@ -3443,6 +3443,16 @@ spec:
                       minimum: 0
                       nullable: true
                       type: number
+                    migration:
+                      description: Migration handles the OSD migration
+                      properties:
+                        confirmation:
+                          description: |-
+                            A user confirmation to migrate the OSDs. It destroys each OSD one at a time, cleans up the backing disk
+                            and prepares OSD with same ID on that disk
+                          pattern: ^$|^yes-really-migrate-osds$
+                          type: string
+                      type: object
                     nearFullRatio:
                       description: NearFullRatio is the ratio at which the cluster is considered nearly full and will raise a ceph health warning. Default is 0.85.
                       maximum: 1
@@ -5536,6 +5546,12 @@ spec:
                     osd:
                       description: OSDStatus represents OSD status of the ceph Cluster
                       properties:
+                        migrationStatus:
+                          description: MigrationStatus status represents the current status of any OSD migration.
+                          properties:
+                            pending:
+                              type: integer
+                          type: object
                         storeType:
                           additionalProperties:
                             type: integer

design/ceph/osd-migration.md

@@ -25,6 +25,7 @@ considered in the future:
   and application data on slower media.
 - Setups with multiple OSDs per drive, though with recent Ceph releases the
   motivation for deploying this way is mostly obviated.
+- OSDs where Persistent Volumes are using partitioned disks due to a [ceph issue](https://tracker.ceph.com/issues/68977). 
 
 ## Proposal
 - Since migration requires destroying of the OSD and cleaning data from the disk,

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

@@ -500,7 +500,13 @@ type DeviceClasses struct {
 // 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"`
+	StoreType       map[string]int  `json:"storeType,omitempty"`
+	MigrationStatus MigrationStatus `json:"migrationStatus,omitempty"`
+}
+
+// MigrationStatus status represents the current status of any OSD migration.
+type MigrationStatus struct {
+	Pending int `json:"pending,omitempty"`
 }
 
 // ClusterVersion represents the version of a Ceph Cluster
@@ -3051,6 +3057,9 @@ type StorageScopeSpec struct {
 	// +nullable
 	// +optional
 	StorageClassDeviceSets []StorageClassDeviceSet `json:"storageClassDeviceSets,omitempty"`
+	// Migration handles the OSD migration
+	// +optional
+	Migration Migration `json:"migration,omitempty"`
 	// +optional
 	Store OSDStore `json:"store,omitempty"`
 	// +optional
@@ -3089,6 +3098,15 @@ type StorageScopeSpec struct {
 	AllowOsdCrushWeightUpdate bool `json:"allowOsdCrushWeightUpdate,omitempty"`
 }
 
+// Migration handles the OSD migration
+type Migration struct {
+	// A user confirmation to migrate the OSDs. It destroys each OSD one at a time, cleans up the backing disk
+	// and prepares OSD with same ID on that disk
+	// +optional
+	// +kubebuilder:validation:Pattern=`^$|^yes-really-migrate-osds$`
+	Confirmation string `json:"confirmation,omitempty"`
+}
+
 // OSDStore is the backend storage type used for creating the OSDs
 type OSDStore struct {
 	// Type of backend storage to be used while creating OSDs. If empty, then bluestore will be used

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

@@ -3081,6 +3081,38 @@ func (in *MgrSpec) DeepCopy() *MgrSpec {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *Migration) DeepCopyInto(out *Migration) {
+	*out = *in
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new Migration.
+func (in *Migration) DeepCopy() *Migration {
+	if in == nil {
+		return nil
+	}
+	out := new(Migration)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *MigrationStatus) DeepCopyInto(out *MigrationStatus) {
+	*out = *in
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new MigrationStatus.
+func (in *MigrationStatus) DeepCopy() *MigrationStatus {
+	if in == nil {
+		return nil
+	}
+	out := new(MigrationStatus)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *MirrorHealthCheckSpec) DeepCopyInto(out *MirrorHealthCheckSpec) {
 	*out = *in
@@ -3615,6 +3647,7 @@ func (in *OSDStatus) DeepCopyInto(out *OSDStatus) {
 			(*out)[key] = val
 		}
 	}
+	out.MigrationStatus = in.MigrationStatus
 	return
 }
 
@@ -4799,6 +4832,7 @@ func (in *StorageScopeSpec) DeepCopyInto(out *StorageScopeSpec) {
 			(*in)[i].DeepCopyInto(&(*out)[i])
 		}
 	}
+	out.Migration = in.Migration
 	out.Store = in.Store
 	if in.FullRatio != nil {
 		in, out := &in.FullRatio, &out.FullRatio

pkg/daemon/ceph/osd/agent.go

@@ -38,13 +38,13 @@ type OsdAgent struct {
 	storeConfig    config.StoreConfig
 	kv             *k8sutil.ConfigMapKVStore
 	pvcBacked      bool
-	replaceOSD     *oposd.OSDReplaceInfo
+	replaceOSD     *oposd.OSDInfo
 }
 
 // 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,
-	replaceOSD *oposd.OSDReplaceInfo, pvcBacked bool) *OsdAgent {
+	replaceOSD *oposd.OSDInfo, pvcBacked bool) *OsdAgent {
 
 	return &OsdAgent{
 		devices:        devices,
@@ -71,7 +71,7 @@ func getDeviceLVPath(context *clusterd.Context, deviceName string) string {
 
 // GetReplaceOSDId returns the OSD ID based on the device name
 func (a *OsdAgent) GetReplaceOSDId(device string) int {
-	if device == a.replaceOSD.Path {
+	if device == a.replaceOSD.BlockPath {
 		return a.replaceOSD.ID
 	}
 

pkg/daemon/ceph/osd/remove.go

@@ -248,15 +248,12 @@ func archiveCrash(clusterdContext *clusterd.Context, clusterInfo *client.Cluster
 }
 
 // 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) (*oposd.OSDReplaceInfo, error) {
-	var block string
+func DestroyOSD(context *clusterd.Context, clusterInfo *client.ClusterInfo, id int, isPVC bool) (*oposd.OSDInfo, error) {
 	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()
@@ -265,7 +262,7 @@ func DestroyOSD(context *clusterd.Context, clusterInfo *client.ClusterInfo, id i
 	}
 	logger.Infof("successfully destroyed osd.%d", osdInfo.ID)
 
-	if isPVC && isEncrypted {
+	if isPVC && osdInfo.Encrypted {
 		// remove the dm device
 		pvcName := os.Getenv(oposd.PVCNameEnvVarName)
 		target := oposd.EncryptionDMName(pvcName, oposd.DmcryptBlockType)
@@ -280,17 +277,17 @@ func DestroyOSD(context *clusterd.Context, clusterInfo *client.ClusterInfo, id i
 		if err != nil {
 			return nil, errors.Wrapf(err, "failed to get device info for %q", blockPath)
 		}
-		block = diskInfo.RealPath
+		osdInfo.BlockPath = 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")
+	logger.Infof("zap OSD.%d path %q", osdInfo.ID, osdInfo.BlockPath)
+	output, err := context.Executor.ExecuteCommandWithCombinedOutput("stdbuf", "-oL", "ceph-volume", "lvm", "zap", osdInfo.BlockPath, "--destroy")
 	if err != nil {
-		return nil, errors.Wrapf(err, "failed to zap osd.%d path %q. %s.", osdInfo.ID, block, output)
+		return nil, errors.Wrapf(err, "failed to zap osd.%d path %q. %s.", osdInfo.ID, osdInfo.BlockPath, output)
 	}
 
 	logger.Infof("%s\n", output)
-	logger.Infof("successfully zapped osd.%d path %q", osdInfo.ID, block)
+	logger.Infof("successfully zapped osd.%d path %q", osdInfo.ID, osdInfo.BlockPath)
 
-	return &oposd.OSDReplaceInfo{ID: osdInfo.ID, Path: block}, nil
+	return osdInfo, nil
 }

pkg/daemon/ceph/osd/volume.go

@@ -1180,7 +1180,7 @@ func GetCephVolumeRawOSDs(context *clusterd.Context, clusterInfo *client.Cluster
 		// pod
 		// For the cleanup pod we don't want to close the encrypted block since it will sanitize it
 		// first and then close it
-		if os.Getenv(oposd.CephVolumeEncryptedKeyEnvVarName) != "" {
+		if osd.Encrypted && os.Getenv(oposd.CephVolumeEncryptedKeyEnvVarName) != "" {
 			// If label and subsystem are not set on the encrypted block let's set it
 			// They will be set if the OSD deployment has been removed manually and the prepare job
 			// runs again.

pkg/daemon/ceph/osd/volume_test.go

@@ -1574,7 +1574,7 @@ func TestInitializeBlockPVC(t *testing.T) {
 	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: &oposd.OSDReplaceInfo{ID: 3, Path: "/dev/sda"}}
+	a = &OsdAgent{clusterInfo: clusterInfo, nodeName: "node1", storeConfig: config.StoreConfig{StoreType: "bluestore"}, replaceOSD: &oposd.OSDInfo{ID: 3, BlockPath: "/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"}},
@@ -1596,7 +1596,7 @@ func TestInitializeBlockPVC(t *testing.T) {
 	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: &oposd.OSDReplaceInfo{ID: 3, Path: "/dev/sda"}}
+	a = &OsdAgent{clusterInfo: clusterInfo, nodeName: "node1", storeConfig: config.StoreConfig{StoreType: "bluestore"}, replaceOSD: &oposd.OSDInfo{ID: 3, BlockPath: "/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"}},

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

@@ -187,9 +187,9 @@ func (c *Cluster) startProvisioningOverPVCs(config *provisionConfig, errs *provi
 		}
 
 		// 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)
+		if c.migrateOSD != nil {
+			if strings.Contains(c.migrateOSD.BlockPath, dataSource.ClaimName) {
+				logger.Infof("updating OSD prepare pod to replace OSD.%d", c.migrateOSD.ID)
 				skipPreparePod = false
 			}
 		}

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

@@ -64,6 +64,12 @@ func (c *Cluster) getOSDLabels(osd OSDInfo, failureDomainValue string, portable
 		labels[deviceType] = osd.DeviceType
 	}
 
+	encryptedOSD := "false"
+	if osd.Encrypted {
+		encryptedOSD = "true"
+	}
+	labels[encrypted] = encryptedOSD
+
 	for k, v := range getOSDTopologyLocationLabels(osd.Location) {
 		labels[k] = v
 	}

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

@@ -0,0 +1,217 @@
+/*
+Copyright 2024 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 (
+	"fmt"
+	"strconv"
+
+	"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"
+	appsv1 "k8s.io/api/apps/v1"
+	v1 "k8s.io/api/core/v1"
+	kerrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+const (
+	// OSDMigrationConfirmation is the confirmation provided by the user in the cephCluster spec.
+	OSDMigrationConfirmation = "yes-really-migrate-osds"
+	// OSDUpdateStoreConfirmation is the confirmation provided by the user to updated OSD backend store
+	OSDUpdateStoreConfirmation = "yes-really-update-store"
+	// OSDMigrationConfigName is the configMap that stores the ID of the last migrated OSD
+	OSDMigrationConfigName = "osd-migration-config"
+	// OSDIdKey is the key used to store the OSD ID inside the `osd-migration-config` configMap
+	OSDIdKey = "osdID"
+)
+
+// migrationConfig represents the OSDs that need migration
+type migrationConfig struct {
+	// osds that require migration (map key is the OSD id)
+	osds map[int]*OSDInfo
+}
+
+func (c *Cluster) newMigrationConfig() (*migrationConfig, error) {
+	mc := migrationConfig{
+		osds: map[int]*OSDInfo{},
+	}
+
+	osdDeployments, err := c.getOSDDeployments()
+	if err != nil {
+		return nil, errors.Wrapf(err, "failed to get existing OSD deployments in namespace %q", c.clusterInfo.Namespace)
+	}
+
+	// get OSDs that require migration due to change in encryption settings
+	err = mc.migrateForEncryption(c, osdDeployments)
+	if err != nil {
+		return nil, errors.Wrapf(err, "failed to get OSDs that require migration due to change in encryption setting")
+	}
+
+	// get OSDs that require migration due the change in OSD store type settings
+	err = mc.migrateForOSDStore(c, osdDeployments)
+	if err != nil {
+		return nil, errors.Wrapf(err, "failed to get OSDs that require migration due to change in OSD Store type setting")
+	}
+
+	return &mc, nil
+}
+
+// migrateForEncryption gets all the OSDs that require migration due to change in the cephCluster encryption setting
+func (m *migrationConfig) migrateForEncryption(c *Cluster, osdDeployments *appsv1.DeploymentList) error {
+	deviceSetMap := map[string]cephv1.StorageClassDeviceSet{}
+	for i := range c.spec.Storage.StorageClassDeviceSets {
+		deviceSetMap[c.spec.Storage.StorageClassDeviceSets[i].Name] = c.spec.Storage.StorageClassDeviceSets[i]
+	}
+
+	for i := range osdDeployments.Items {
+		osdDeviceSetName := osdDeployments.Items[i].Labels[CephDeviceSetLabelKey]
+		requestedEncryptionSetting := deviceSetMap[osdDeviceSetName].Encrypted
+		actualEncryptedSetting := false
+		if osdDeployments.Items[i].Labels["encrypted"] == "true" {
+			actualEncryptedSetting = true
+		}
+
+		if requestedEncryptionSetting != actualEncryptedSetting {
+			osdInfo, err := c.getOSDInfo(&osdDeployments.Items[i])
+			if err != nil {
+				return errors.Wrapf(err, "failed to details about the OSD %q", osdDeployments.Items[i].Name)
+			}
+			logger.Infof("migration is required for OSD.%d due to change in encryption settings from %t to %t in storageClassDeviceSet %q", osdInfo.ID, actualEncryptedSetting, requestedEncryptionSetting, osdDeviceSetName)
+			if _, exists := m.osds[osdInfo.ID]; !exists {
+				m.osds[osdInfo.ID] = &osdInfo
+			}
+		}
+	}
+	return nil
+}
+
+// migrateForOSDStore gets all the OSDs that require migration due to change in the cephCluster OSD storeType setting
+func (m *migrationConfig) migrateForOSDStore(c *Cluster, osdDeployments *appsv1.DeploymentList) error {
+	for i := range osdDeployments.Items {
+		if osdStore, ok := osdDeployments.Items[i].Labels[osdStore]; ok {
+			if osdStore != string(c.spec.Storage.Store.Type) {
+				osdInfo, err := c.getOSDInfo(&osdDeployments.Items[i])
+				if err != nil {
+					return errors.Wrapf(err, "failed to details about the OSD %q", osdDeployments.Items[i].Name)
+				}
+				logger.Infof("migration is required for OSD.%d to update storeType from %q to %q", osdInfo.ID, osdStore, c.spec.Storage.Store.Type)
+				if _, exists := m.osds[osdInfo.ID]; !exists {
+					m.osds[osdInfo.ID] = &osdInfo
+				}
+			}
+		}
+	}
+	return nil
+}
+
+// getOSDToMigrate returns the next OSD to migrate from the list of OSDs that are pending migration.
+func (m *migrationConfig) getOSDToMigrate() *OSDInfo {
+	osdInfo := &OSDInfo{}
+	osdID := -1
+	for k, v := range m.osds {
+		osdID, osdInfo = k, v
+		break
+	}
+	delete(m.osds, osdID)
+	return osdInfo
+}
+
+func (m *migrationConfig) getOSDIds() []int {
+	osdIds := make([]int, len(m.osds))
+	i := 0
+	for k := range m.osds {
+		osdIds[i] = k
+		i++
+	}
+	return osdIds
+}
+
+// saveMigrationConfig saves the ID of the migrated OSD to a configMap
+func saveMigrationConfig(context *clusterd.Context, clusterInfo *cephclient.ClusterInfo, osdID int) error {
+	newConfigMap := &v1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      OSDMigrationConfigName,
+			Namespace: clusterInfo.Namespace,
+		},
+		Data: map[string]string{
+			OSDIdKey: strconv.Itoa(osdID),
+		},
+	}
+
+	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
+}
+
+// isLastOSDMigrationComplete checks if the deployment for the migrated OSD got created successfully.
+func isLastOSDMigrationComplete(c *Cluster) (bool, error) {
+	migratedOSDId, err := getLastMigratedOSDId(c.context, c.clusterInfo)
+	if err != nil {
+		return false, errors.Wrapf(err, "failed to get last migrated OSD ID")
+	}
+
+	if migratedOSDId == -1 {
+		return true, nil
+	}
+
+	deploymentName := fmt.Sprintf("rook-ceph-osd-%d", migratedOSDId)
+	_, err = c.context.Clientset.AppsV1().Deployments(c.clusterInfo.Namespace).Get(c.clusterInfo.Context, deploymentName, metav1.GetOptions{})
+	if err != nil {
+		if kerrors.IsNotFound(err) {
+			logger.Infof("deployment for the last migrated OSD with ID - %d is not found.", migratedOSDId)
+			return false, nil
+		}
+	}
+
+	logger.Infof("migration of OSD.%d was successful", migratedOSDId)
+	return true, nil
+}
+
+// getLastMigratedOSDId fetches the ID of the last migrated OSD from the "osd-migration-config" configmap
+func getLastMigratedOSDId(context *clusterd.Context, clusterInfo *cephclient.ClusterInfo) (int, error) {
+	cm, err := context.Clientset.CoreV1().ConfigMaps(clusterInfo.Namespace).Get(clusterInfo.Context, OSDMigrationConfigName, metav1.GetOptions{})
+	if err != nil {
+		if kerrors.IsNotFound(err) {
+			return -1, nil
+		}
+	}
+
+	osdID, ok := cm.Data[OSDIdKey]
+	if !ok || osdID == "" {
+		logger.Debugf("empty config map %q", OSDMigrationConfigName)
+		return -1, nil
+	}
+
+	osdIDInt, err := strconv.Atoi(osdID)
+	if err != nil {
+		return -1, errors.Wrapf(err, "failed to convert OSD id %q to integer.", osdID)
+	}
+
+	return osdIDInt, nil
+}

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

@@ -0,0 +1,225 @@
+/*
+Copyright 2024 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"
+
+	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/stretchr/testify/assert"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes/fake"
+)
+
+func TestMigrateForEncryption(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)
+
+	c := New(ctx, clusterInfo, cephv1.ClusterSpec{}, "rook/rook:master")
+
+	t.Run("no OSD migration is required due to encryption", func(t *testing.T) {
+		c.spec.Storage.StorageClassDeviceSets = []cephv1.StorageClassDeviceSet{
+			{
+				Name:      "set1",
+				Encrypted: true,
+			},
+
+			{
+				Name:      "set2",
+				Encrypted: true,
+			},
+		}
+
+		d1 := getDummyDeploymentOnNode(clientset, c, "node2", 1)
+		d1.Labels["encrypted"] = "true"
+		d1.Labels["ceph.rook.io/DeviceSet"] = "set1"
+		createDeploymentOrPanic(clientset, d1)
+
+		d2 := getDummyDeploymentOnNode(clientset, c, "node2", 2)
+		d2.Labels["encrypted"] = "true"
+		d2.Labels["ceph.rook.io/DeviceSet"] = "set1"
+		createDeploymentOrPanic(clientset, d2)
+
+		deployments, err := c.getOSDDeployments()
+		assert.NoError(t, err)
+
+		mc := migrationConfig{
+			osds: map[int]*OSDInfo{},
+		}
+
+		err = mc.migrateForEncryption(c, deployments)
+		assert.NoError(t, err)
+		assert.Equal(t, 0, len(mc.osds))
+	})
+	t.Run("osd.1 on set1 needs migration", func(t *testing.T) {
+		c.clusterInfo.Namespace = namespace2
+		c.spec.Storage.StorageClassDeviceSets = []cephv1.StorageClassDeviceSet{
+			{
+				Name:      "set1",
+				Encrypted: true,
+			},
+		}
+
+		d1 := getDummyDeploymentOnNode(clientset, c, "node2", 1)
+		d1.Labels["encrypted"] = "false"
+		d1.Labels["ceph.rook.io/DeviceSet"] = "set1"
+		createDeploymentOrPanic(clientset, d1)
+
+		deployments, err := c.getOSDDeployments()
+		assert.NoError(t, err)
+
+		mc := migrationConfig{
+			osds: map[int]*OSDInfo{},
+		}
+
+		err = mc.migrateForEncryption(c, deployments)
+		assert.NoError(t, err)
+		assert.Equal(t, 1, len(mc.osds))
+		assert.Equal(t, 1, mc.osds[1].ID)
+	})
+
+}
+
+func TestMigrationForOSDStore(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)
+
+	c := New(ctx, clusterInfo, cephv1.ClusterSpec{}, "rook/rook:master")
+
+	t.Run("no OSD migration is required due to OSD store change", func(t *testing.T) {
+		c.spec.Storage.Store.Type = "store1"
+
+		d1 := getDummyDeploymentOnNode(clientset, c, "node2", 1)
+		d1.Labels[osdStore] = "store1"
+		createDeploymentOrPanic(clientset, d1)
+
+		d2 := getDummyDeploymentOnNode(clientset, c, "node2", 2)
+		d2.Labels[osdStore] = "store1"
+		createDeploymentOrPanic(clientset, d2)
+
+		deployments, err := c.getOSDDeployments()
+		assert.NoError(t, err)
+
+		mc := migrationConfig{
+			osds: map[int]*OSDInfo{},
+		}
+
+		err = mc.migrateForEncryption(c, deployments)
+		assert.NoError(t, err)
+		assert.Equal(t, 0, len(mc.osds))
+	})
+	t.Run("osd.1 needs migration due to change is OSD store type", func(t *testing.T) {
+		c.clusterInfo.Namespace = namespace2
+		c.spec.Storage.Store.Type = "newStore"
+
+		d1 := getDummyDeploymentOnNode(clientset, c, "node2", 1)
+		d1.Labels[osdStore] = "store1" // store type is set to `store1` but spec has `newStore`
+		createDeploymentOrPanic(clientset, d1)
+
+		d2 := getDummyDeploymentOnNode(clientset, c, "node2", 2)
+		d2.Labels[osdStore] = "newStore" // store type label matches with the spec
+		createDeploymentOrPanic(clientset, d2)
+
+		deployments, err := c.getOSDDeployments()
+		assert.NoError(t, err)
+
+		mc := migrationConfig{
+			osds: map[int]*OSDInfo{},
+		}
+
+		err = mc.migrateForOSDStore(c, deployments)
+		assert.NoError(t, err)
+		assert.Equal(t, 1, len(mc.osds))
+		assert.Equal(t, 1, mc.osds[1].ID)
+	})
+}
+
+func createMigrationConfigmap(osdID, ns string, clientset *fake.Clientset) error {
+	ctx := context.TODO()
+	data := make(map[string]string, 1)
+	data[OSDIdKey] = osdID
+	cm := &corev1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      OSDMigrationConfigName,
+			Namespace: ns,
+		},
+		Data: data,
+	}
+	_, err := clientset.CoreV1().ConfigMaps(ns).Create(ctx, cm, metav1.CreateOptions{})
+	return err
+}
+func TestIsLastOSDMigrationComplete(t *testing.T) {
+	namespace := "rook-ceph"
+	clientset := fake.NewSimpleClientset()
+	ctx := &clusterd.Context{
+		Clientset: clientset,
+	}
+	clusterInfo := &cephclient.ClusterInfo{
+		Namespace: namespace,
+		Context:   context.TODO(),
+	}
+	clusterInfo.SetName("mycluster")
+	clusterInfo.OwnerInfo = cephclient.NewMinimumOwnerInfo(t)
+
+	c := New(ctx, clusterInfo, cephv1.ClusterSpec{}, "rook/rook:master")
+	t.Run("no OSD migration config found", func(t *testing.T) {
+		result, err := isLastOSDMigrationComplete(c)
+		assert.NoError(t, err)
+		assert.Equal(t, true, result)
+	})
+
+	t.Run("osd.1 was migrated but its not up yet", func(t *testing.T) {
+		err := createMigrationConfigmap("1", namespace, clientset)
+		assert.NoError(t, err)
+		result, err := isLastOSDMigrationComplete(c)
+		assert.NoError(t, err)
+		assert.Equal(t, false, result)
+	})
+
+	t.Run("migrated osd.1 is up", func(t *testing.T) {
+		d1 := getDummyDeploymentOnNode(clientset, c, "node2", 1)
+		createDeploymentOrPanic(clientset, d1)
+		result, err := isLastOSDMigrationComplete(c)
+		assert.NoError(t, err)
+		assert.Equal(t, true, result)
+	})
+}

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

@@ -74,6 +74,7 @@ const (
 	deviceClass                    = "device-class"
 	osdStore                       = "osd-store"
 	deviceType                     = "device-type"
+	encrypted                      = "encrypted"
 )
 
 // Cluster keeps track of the OSDs
@@ -85,7 +86,7 @@ type Cluster struct {
 	ValidStorage   cephv1.StorageScopeSpec // valid subset of `Storage`, computed at runtime
 	kv             *k8sutil.ConfigMapKVStore
 	deviceSets     []deviceSet
-	replaceOSD     *OSDReplaceInfo
+	migrateOSD     *OSDInfo
 	deprecatedOSDs map[string][]int
 }
 
@@ -218,26 +219,27 @@ func (c *Cluster) Start() error {
 	}
 	logger.Infof("wait timeout for healthy OSDs during upgrade or restart is %q", c.clusterInfo.OsdUpgradeTimeout)
 
-	// replace OSDs for a new backing store
-	osdsToBeReplaced, err := c.replaceOSDForNewStore()
-	if err != nil {
-		return errors.Wrapf(err, "failed to replace OSD for new backing store %q in namespace %q", c.spec.Storage.Store.Type, namespace)
-	}
-
 	osdsToSkipReconcile, err := controller.GetDaemonsToSkipReconcile(c.clusterInfo.Context, c.context, c.clusterInfo.Namespace, OsdIdLabelKey, AppName)
 	if err != nil {
 		logger.Warningf("failed to get osds to skip reconcile. %v", err)
 	}
 
+	migrationConfig, err := c.startOSDMigration()
+	if err != nil {
+		return errors.Wrapf(err, "failed to start OSD migration")
+	}
+
 	// 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)
 	}
 
-	// OSDs that are to be replaced should not be upgraded. So remove them from the `updateQueue`
-	if len(osdsToBeReplaced) > 0 {
-		updateQueue.Remove(osdsToBeReplaced.getOSDIds())
+	if migrationConfig != nil {
+		if len(migrationConfig.osds) != 0 {
+			// prevent upgrade of OSDs that require migration
+			updateQueue.Remove(migrationConfig.getOSDIds())
+		}
 	}
 
 	logger.Debugf("%d of %d OSD Deployments need update", updateQueue.Len(), deployments.Len())
@@ -295,32 +297,69 @@ func (c *Cluster) Start() error {
 		return errors.Wrapf(err, "failed to update ceph storage status")
 	}
 
-	if c.spec.Storage.Store.UpdateStore == OSDStoreUpdateConfirmation {
-		delOpts := &k8sutil.DeleteOptions{WaitOptions: k8sutil.WaitOptions{Wait: true}}
-		err := k8sutil.DeleteConfigMap(c.clusterInfo.Context, c.context.Clientset, OSDReplaceConfigName, namespace, delOpts)
-		if err != nil {
-			if kerrors.IsNotFound(err) {
-				logger.Debugf("config map %q not found. Ignoring since object must be deleted.", OSDReplaceConfigName)
-			} else {
-				return errors.Wrapf(err, "failed to delete the %q configmap", OSDReplaceConfigName)
-			}
-		}
-
-		if len(osdsToBeReplaced) > 0 {
-			// wait for the pgs to be healthy before attempting to migrate the next OSD
-			_, err := c.waitForHealthyPGs()
-			if err != nil {
-				return errors.Wrapf(err, "failed to wait for pgs to be healhty")
-			}
-			// return with error to reconcile the operator since there are OSDs that are pending migration
-			return errors.New("reconcile operator to replace OSDs that are pending migration")
-		}
-	}
-
 	logger.Infof("finished running OSDs in namespace %q", namespace)
 	return nil
 }
 
+func (c *Cluster) startOSDMigration() (*migrationConfig, error) {
+	if !c.isMigrationRequested() {
+		logger.Debug("no OSD migration is requested")
+		return nil, nil
+	}
+
+	logger.Info("osd migration is requested")
+
+	// start migration only if PGs are active+clean
+	pgsHealhty, err := c.waitForHealthyPGs()
+	if err != nil {
+		return nil, errors.Wrapf(err, "failed to wait for pgs to be healthy")
+	}
+
+	if !pgsHealhty {
+		return nil, errors.Wrapf(err, "failed to start migration due to unhealthy PGs")
+	}
+
+	// skip migration if previously migrated OSD is not up yet.
+	migrationComplete, err := isLastOSDMigrationComplete(c)
+	if err != nil {
+		return nil, errors.Wrapf(err, "failed to check if the last migration was successful or not")
+	}
+
+	if !migrationComplete {
+		return nil, errors.Wrapf(err, "migration of the last OSD is not complete")
+	}
+
+	migrationConfig, err := c.newMigrationConfig()
+	if err != nil {
+		return nil, errors.Wrapf(err, "failed to get new OSD migration config")
+	}
+
+	// delete deployment of the osd that needs migration
+	if migrationConfig != nil && len(migrationConfig.osds) > 0 {
+		osdToMigrate := migrationConfig.getOSDToMigrate()
+		logger.Infof("deleting OSD.%d deployment for migration ", osdToMigrate.ID)
+		err = c.deleteOSDDeployment(osdToMigrate.ID)
+		if err != nil {
+			return nil, errors.Wrapf(err, "failed to delete deployment for osd.%d that needs migration %q", osdToMigrate.ID, c.clusterInfo.Namespace)
+		}
+		err = saveMigrationConfig(c.context, c.clusterInfo, osdToMigrate.ID)
+		if err != nil {
+			return nil, errors.Wrapf(err, "failed to save migrated OSD ID %din the config map", osdToMigrate.ID)
+		}
+		c.migrateOSD = osdToMigrate
+	}
+
+	return migrationConfig, nil
+}
+
+func (c *Cluster) isMigrationRequested() bool {
+	// check for OSDUpdateStoreConfirmation as well for backwards compatibility
+	if c.spec.Storage.Migration.Confirmation == OSDMigrationConfirmation || c.spec.Storage.Store.UpdateStore == OSDUpdateStoreConfirmation {
+		return true
+	}
+	return false
+}
+
 func (c *Cluster) postReconcileUpdateOSDProperties(desiredOSDs map[int]*OSDInfo) error {
 	osdUsage, err := cephclient.GetOSDUsage(c.context, c.clusterInfo)
 	if err != nil {
@@ -671,6 +710,10 @@ func (c *Cluster) getOSDInfo(d *appsv1.Deployment) (OSDInfo, error) {
 	}
 
 	osd.Store = d.Labels[osdStore]
+	osd.Encrypted = false
+	if d.Labels[encrypted] == "true" {
+		osd.Encrypted = true
+	}
 
 	if isPVC {
 		osd.PVCName = d.Labels[OSDOverPVCLabelKey]
@@ -894,7 +937,7 @@ func (c *Cluster) deleteOSDDeployment(osdID int) error {
 		}
 	}
 
-	return c.replaceOSD.saveAsConfig(c.context, c.clusterInfo)
+	return nil
 }
 
 func (c *Cluster) waitForHealthyPGs() (bool, error) {
@@ -941,6 +984,15 @@ func (c *Cluster) updateCephStorageStatus() error {
 	// Add the status about deprecated OSDs
 	cephClusterStorage.DeprecatedOSDs = c.deprecatedOSDs
 
+	// Update pending migration status
+	if c.isMigrationRequested() {
+		migrationConfig, err := c.newMigrationConfig()
+		if err != nil {
+			return errors.Wrapf(err, "failed to get osd migration config to update cluster status")
+		}
+		cephClusterStorage.OSD.MigrationStatus.Pending = len(migrationConfig.osds)
+	}
+
 	err = c.context.Client.Get(c.clusterInfo.Context, c.clusterInfo.NamespacedName(), &cephCluster)
 	if err != nil {
 		if kerrors.IsNotFound(err) {

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

@@ -18,7 +18,6 @@ package osd
 
 import (
 	"context"
-	"encoding/json"
 	"testing"
 
 	"github.com/pkg/errors"
@@ -796,174 +795,6 @@ func TestGetOSDInfoWithCustomRoot(t *testing.T) {
 	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)
-		osdsToBeReplaced, err := c.replaceOSDForNewStore()
-		assert.NoError(t, err)
-		assert.Equal(t, 0, len(osdsToBeReplaced))
-		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)
-		osdsToBeReplaced, err := c.replaceOSDForNewStore()
-		assert.NoError(t, err)
-		assert.Equal(t, 1, len(osdsToBeReplaced))
-		assert.NotNil(t, c.replaceOSD)
-		assert.Equal(t, 2, c.replaceOSD.ID)
-		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{})
-		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, "pvc1", 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")
-		osdsToBeReplaced, err := c.replaceOSDForNewStore()
-		assert.NoError(t, err)
-		assert.Equal(t, 0, len(osdsToBeReplaced))
-		assert.Nil(t, c.replaceOSD)
-	})
-}
-
 func TestUpdateCephStorageStatus(t *testing.T) {
 	ctx := context.TODO()
 	clusterInfo := cephclient.AdminTestClusterInfo("fake")

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

@@ -308,16 +308,16 @@ func (c *Cluster) provisionOSDContainer(osdProps osdProperties, copyBinariesMoun
 
 	// Add OSD ID as environment variables.
 	// When this env is set, prepare pod job will destroy this OSD.
-	if c.replaceOSD != nil {
+	if c.migrateOSD != 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)))
+			if strings.Contains(c.migrateOSD.PVCName, osdProps.pvc.ClaimName) {
+				envVars = append(envVars, replaceOSDIDEnvVar(fmt.Sprint(c.migrateOSD.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)))
+			if c.migrateOSD.NodeName == osdProps.crushHostname {
+				envVars = append(envVars, replaceOSDIDEnvVar(fmt.Sprint(c.migrateOSD.ID)))
 			}
 		}
 	}

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

@@ -1,191 +0,0 @@
-/*
-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"
-
-	"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"`
-}
-
-type OSDReplaceInfoList []OSDReplaceInfo
-
-func (o *OSDReplaceInfoList) getOSDIds() []int {
-	osdIDs := []int{}
-	for _, osd := range *o {
-		osdIDs = append(osdIDs, osd.ID)
-	}
-	return osdIDs
-}
-
-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
-}
-
-func (c *Cluster) replaceOSDForNewStore() (OSDReplaceInfoList, error) {
-	if c.spec.Storage.Store.UpdateStore != OSDStoreUpdateConfirmation {
-		logger.Info("no replacement of osds is requested")
-		return nil, nil
-	}
-
-	osdsToBeReplaced, err := c.getOSDWithNonMatchingStore()
-	if err != nil {
-		return nil, errors.Wrapf(err, "failed to get information about the OSDs where backing store does not match the spec in namespace %q", c.clusterInfo.Namespace)
-	}
-
-	if len(osdsToBeReplaced) == 0 {
-		logger.Debug("all OSDs are using the desired backing store. No replacement is required.")
-		return osdsToBeReplaced, nil
-	}
-
-	// replace an OSD only if Pgs are healthy
-	pgHealthMsg, pgClean, err := cephclient.IsClusterClean(c.context, c.clusterInfo, c.spec.DisruptionManagement.PGHealthyRegex)
-	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 osdsToBeReplaced, nil
-	}
-
-	// Check for an existing OSDs in the configmap
-	osdToBeReplaced, err := GetOSDReplaceConfigMap(c.context, c.clusterInfo)
-	if err != nil {
-		return nil, errors.Wrap(err, "failed to get any existing OSD in replace configmap")
-	}
-	if osdToBeReplaced != nil {
-		c.replaceOSD = osdToBeReplaced
-	} else {
-		c.replaceOSD = &osdsToBeReplaced[0]
-	}
-
-	logger.Infof("replacing OSD.%d to the new backing store %q", c.replaceOSD.ID, c.spec.Storage.Store.Type)
-	err = c.deleteOSDDeployment(c.replaceOSD.ID)
-	if err != nil {
-		return nil, errors.Wrapf(err, "failed to delete OSD deployment that needs migration to new backend store in namespace %q", c.clusterInfo.Namespace)
-	}
-
-	return osdsToBeReplaced, nil
-}
-
-// getOSDWithNonMatchingStore returns OSDs with osd-store label different from expected store in cephCluster spec
-func (c *Cluster) getOSDWithNonMatchingStore() (OSDReplaceInfoList, error) {
-	osdReplaceList := []OSDReplaceInfo{}
-	// get existing OSD deployments
-	osdDeployments, err := c.getOSDDeployments()
-	if err != nil {
-		return osdReplaceList, errors.Wrapf(err, "failed to get existing OSD deployments in namespace %q", c.clusterInfo.Namespace)
-	}
-	for i := range osdDeployments.Items {
-		if osdStore, ok := osdDeployments.Items[i].Labels[osdStore]; ok {
-			if osdStore != string(c.spec.Storage.Store.Type) {
-				osdInfo, err := c.getOSDInfo(&osdDeployments.Items[i])
-				if err != nil {
-					return nil, errors.Wrapf(err, "failed to details about the OSD %q", osdDeployments.Items[i].Name)
-				}
-				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})
-			}
-		}
-	}
-
-	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

@@ -1,162 +0,0 @@
-/*
-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"
-
-	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/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, "pvc0", osdList[0].Path)
-	})
-}