Merge pull request #6982 from travisn/osd-pvc-allow-removal

ceph: Allow removal of arbitrary osds on pvcs and simplify pvc names

Documentation/ceph-osd-mgmt.md

@@ -67,19 +67,22 @@ If you are using `useAllDevices: true`, no change to the CR is necessary.
 
 ### PVC-based cluster
 
-If you want to reduce the storage in your cluster, a healthy PVC-based OSD can be removed as described in this section.
+To reduce the storage in your cluster or remove a failed OSD on a PVC:
 
-1. Shrink the number of OSDs in the `storageClassDeviceSet` in the CephCluster CR.
+1. Shrink the number of OSDs in the `storageClassDeviceSet` in the CephCluster CR. If you have multiple device sets,
+   you may need to change the index of `0` in this example path.
    - `kubectl -n rook-ceph patch CephCluster rook-ceph --type=json -p '[{"op": "replace", "path": "/spec/storage/storageClassDeviceSets/0/count", "value":<desired number>}]'`
-   - Reduce the `count` of the OSDs to the desired number. Rook will not take any action to automatically remove the extra OSD(s), but will effectively stop managing the orphaned OSD.
-2. Identify the orphaned PVC that belongs to the orphaned OSD.
-   - The orphaned PVC will have the highest index among the PVCs for the device set.
+   - Reduce the `count` of the OSDs to the desired number. Rook will not take any action to automatically remove the extra OSD(s).
+2. Identify the PVC that belongs to the OSD that is failed or otherwise being removed.
    - `kubectl -n rook-ceph get pvc -l ceph.rook.io/DeviceSet=<deviceSet>`
-   - For example if the device set is named `set1` and the `count` was reduced from `3` to `2`, the orphaned PVC would have the index `2` and might be named `set1-2-data-vbwcf`
-3. Identify the orphaned OSD.
+3. Identify the OSD you desire to remove.
    - The OSD assigned to the PVC can be found in the labels on the PVC
    - `kubectl -n rook-ceph get pod -l ceph.rook.io/pvc=<orphaned-pvc> -o yaml | grep ceph-osd-id`
    - For example, this might return: `ceph-osd-id: "0"`
+   - Remember the OSD ID for purging the OSD below
+
+If you later increase the count in the device set, note that the operator will create PVCs with the highest index
+that is not currently in use by existing OSD PVCs.
 
 ### Confirm the OSD is down
 
@@ -96,7 +99,7 @@ In the osd-purge.yaml, change the `<OSD-IDs>` to the ID(s) of the OSDs you want
 2. When the job is completed, review the logs to ensure success: `kubectl -n rook-ceph logs -l app=rook-ceph-purge-osd`
 3. When finished, you can delete the job: `kubectl delete -f osd-purge.yaml`
 
-If you want to remove OSDs by hand, please read the following sections. However, we recommend you to use the above-mentioned job to avoid operation errors.
+If you want to remove OSDs by hand, continue with the following sections. However, we recommend you to use the above-mentioned job to avoid operation errors.
 
 ### Purge the OSD manually
 

cluster/examples/kubernetes/ceph/osd-purge.yaml

@@ -63,7 +63,11 @@ spec:
           volumeMounts:
             - mountPath: /etc/ceph
               name: ceph-conf-emptydir
+            - mountPath: /var/lib/rook
+              name: rook-config
       volumes:
         - emptyDir: {}
           name: ceph-conf-emptydir
+        - emptyDir: {}
+          name: rook-config
       restartPolicy: Never

cmd/rook/ceph/osd.go

@@ -284,6 +284,7 @@ func commonOSDInit(cmd *cobra.Command) {
 func getLocation(clientset kubernetes.Interface) (string, error) {
 	// get the value the operator instructed to use as the host name in the CRUSH map
 	hostNameLabel := os.Getenv("ROOK_CRUSHMAP_HOSTNAME")
+
 	rootLabel := os.Getenv(oposd.CrushRootVarName)
 
 	loc, err := oposd.GetLocationWithNode(clientset, os.Getenv(k8sutil.NodeNameEnvVar), rootLabel, hostNameLabel)

pkg/operator/ceph/cluster/controller.go

@@ -547,7 +547,7 @@ func (c *ClusterController) deleteOSDEncryptionKeyFromKMS(currentCluster *cephv1
 	}
 
 	// Fetch PVCs
-	osdPVCs, err := osd.GetExistingOSDPVCs(c.context, currentCluster.Namespace)
+	osdPVCs, _, err := osd.GetExistingPVCs(c.context, currentCluster.Namespace)
 	if err != nil {
 		return errors.Wrap(err, "failed to list osd pvc")
 	}

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

@@ -19,6 +19,7 @@ package osd
 import (
 	"context"
 	"fmt"
+	"strconv"
 	"strings"
 
 	"github.com/pkg/errors"
@@ -27,6 +28,7 @@ import (
 	"github.com/rook/rook/pkg/clusterd"
 	"github.com/rook/rook/pkg/operator/ceph/controller"
 	"github.com/rook/rook/pkg/operator/k8sutil"
+	"github.com/rook/rook/pkg/util"
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
@@ -34,93 +36,127 @@ import (
 func (c *Cluster) prepareStorageClassDeviceSets(config *provisionConfig) []rookv1.VolumeSource {
 	volumeSources := []rookv1.VolumeSource{}
 
-	existingPVCs, err := GetExistingOSDPVCs(c.context, c.clusterInfo.Namespace)
+	existingPVCs, uniqueOSDsPerDeviceSet, err := GetExistingPVCs(c.context, c.clusterInfo.Namespace)
 	if err != nil {
 		config.addError("failed to detect existing OSD PVCs. %v", err)
 		return volumeSources
 	}
 
-	// Iterate over storageClassDeviceSet
-	for _, storageClassDeviceSet := range c.spec.Storage.StorageClassDeviceSets {
-		if err := controller.CheckPodMemory(cephv1.ResourcesKeyPrepareOSD, storageClassDeviceSet.Resources, cephOsdPodMinimumMemory); err != nil {
-			config.addError("cannot use storageClassDeviceSet %q for creating osds %v", storageClassDeviceSet.Name, err)
+	// Iterate over deviceSet
+	for _, deviceSet := range c.spec.Storage.StorageClassDeviceSets {
+		if err := controller.CheckPodMemory(cephv1.ResourcesKeyPrepareOSD, deviceSet.Resources, cephOsdPodMinimumMemory); err != nil {
+			config.addError("cannot use device set %q for creating osds %v", deviceSet.Name, err)
+			continue
+		}
+		// Check if the volume claim template is specified
+		if len(deviceSet.VolumeClaimTemplates) == 0 {
+			logger.Warningf("no volumeClaimTemplate specified for device set %q", deviceSet.Name)
 			continue
 		}
-		for i := 0; i < storageClassDeviceSet.Count; i++ {
-			// Check if the volume claim template has PVCs
-			if len(storageClassDeviceSet.VolumeClaimTemplates) == 0 {
-				logger.Warningf("no PVC available for storageClassDeviceSet %q", storageClassDeviceSet.Name)
-				continue
-			}
 
-			// Create the PVC source for each of the data, metadata, and other types of templates if defined.
-			pvcSources := map[string]v1.PersistentVolumeClaimVolumeSource{}
-			var dataSize string
-			var crushDeviceClass string
-			for _, pvcTemplate := range storageClassDeviceSet.VolumeClaimTemplates {
-				if pvcTemplate.Name == "" {
-					// For backward compatibility a blank name must be treated as a data volume
-					pvcTemplate.Name = bluestorePVCData
-				}
-
-				pvc, err := c.createStorageClassDeviceSetPVC(existingPVCs, storageClassDeviceSet.Name, pvcTemplate, i)
+		// Iterate through existing PVCs to ensure they are up-to-date, no metadata pvcs are missing, etc
+		highestExistingID := -1
+		countInDeviceSet := 0
+		if existingIDs, ok := uniqueOSDsPerDeviceSet[deviceSet.Name]; ok {
+			logger.Infof("verifying pvcs exist for %d osds in device set %q", existingIDs.Count(), deviceSet.Name)
+			for existingID := range existingIDs.Iter() {
+				pvcID, err := strconv.Atoi(existingID)
 				if err != nil {
-					config.addError("failed to create osd for storageClassDeviceSet %q for count %d. %v", storageClassDeviceSet.Name, i, err)
+					config.addError("invalid PVC index %q found for device set %q", existingID, deviceSet.Name)
 					continue
 				}
-
-				// The PVC type must be from a predefined set such as "data" and "metadata". These names must be enforced if the wal/db are specified
-				// with a separate device, but if there is a single volume template we can assume it is always the data template.
-				pvcType := pvcTemplate.Name
-				if len(storageClassDeviceSet.VolumeClaimTemplates) == 1 {
-					pvcType = bluestorePVCData
-				}
-
-				if pvcType == bluestorePVCData {
-					pvcSize := pvc.Spec.Resources.Requests[v1.ResourceStorage]
-					dataSize = pvcSize.String()
-					crushDeviceClass = pvcTemplate.Annotations["crushDeviceClass"]
-				}
-				pvcSources[pvcType] = v1.PersistentVolumeClaimVolumeSource{
-					ClaimName: pvc.GetName(),
-					ReadOnly:  false,
+				// keep track of the max PVC index found so we know what index to start with for new OSDs
+				if pvcID > highestExistingID {
+					highestExistingID = pvcID
 				}
+				volumeSource := c.createDeviceSetPVCsForIndex(config, deviceSet, existingPVCs, pvcID)
+				volumeSources = append(volumeSources, volumeSource)
 			}
-
-			volumeSources = append(volumeSources, rookv1.VolumeSource{
-				Name:                storageClassDeviceSet.Name,
-				Resources:           storageClassDeviceSet.Resources,
-				Placement:           storageClassDeviceSet.Placement,
-				PreparePlacement:    storageClassDeviceSet.PreparePlacement,
-				Config:              storageClassDeviceSet.Config,
-				Size:                dataSize,
-				PVCSources:          pvcSources,
-				Portable:            storageClassDeviceSet.Portable,
-				TuneSlowDeviceClass: storageClassDeviceSet.TuneSlowDeviceClass,
-				TuneFastDeviceClass: storageClassDeviceSet.TuneFastDeviceClass,
-				SchedulerName:       storageClassDeviceSet.SchedulerName,
-				CrushDeviceClass:    crushDeviceClass,
-				Encrypted:           storageClassDeviceSet.Encrypted,
-			})
+			countInDeviceSet = existingIDs.Count()
+		}
+		// Create new PVCs if we are not yet at the expected count
+		// No new PVCs will be created if we have too many
+		pvcsToCreate := deviceSet.Count - countInDeviceSet
+		if pvcsToCreate > 0 {
+			logger.Infof("creating %d new PVCs for device set %q", pvcsToCreate, deviceSet.Name)
+		}
+		for i := 0; i < pvcsToCreate; i++ {
+			pvcID := highestExistingID + i + 1
+			volumeSource := c.createDeviceSetPVCsForIndex(config, deviceSet, existingPVCs, pvcID)
+			volumeSources = append(volumeSources, volumeSource)
+			countInDeviceSet++
 		}
 	}
 
 	return volumeSources
 }
 
-func (c *Cluster) createStorageClassDeviceSetPVC(existingPVCs map[string]*v1.PersistentVolumeClaim, storageClassDeviceSetName string, pvcTemplate v1.PersistentVolumeClaim, setIndex int) (*v1.PersistentVolumeClaim, error) {
+func (c *Cluster) createDeviceSetPVCsForIndex(config *provisionConfig, deviceSet rookv1.StorageClassDeviceSet, existingPVCs map[string]*v1.PersistentVolumeClaim, setIndex int) rookv1.VolumeSource {
+	// Create the PVC source for each of the data, metadata, and other types of templates if defined.
+	pvcSources := map[string]v1.PersistentVolumeClaimVolumeSource{}
+
+	var dataSize string
+	var crushDeviceClass string
+	for _, pvcTemplate := range deviceSet.VolumeClaimTemplates {
+		if pvcTemplate.Name == "" {
+			// For backward compatibility a blank name must be treated as a data volume
+			pvcTemplate.Name = bluestorePVCData
+		}
+
+		pvc, err := c.createDeviceSetPVC(existingPVCs, deviceSet.Name, pvcTemplate, setIndex)
+		if err != nil {
+			config.addError("failed to create osd for device set %q for count %d. %v", deviceSet.Name, setIndex, err)
+			continue
+		}
+
+		// The PVC type must be from a predefined set such as "data", "metadata", and "wal". These names must be enforced if the wal/db are specified
+		// with a separate device, but if there is a single volume template we can assume it is always the data template.
+		pvcType := pvcTemplate.Name
+		if len(deviceSet.VolumeClaimTemplates) == 1 {
+			pvcType = bluestorePVCData
+		}
+
+		if pvcType == bluestorePVCData {
+			pvcSize := pvc.Spec.Resources.Requests[v1.ResourceStorage]
+			dataSize = pvcSize.String()
+			crushDeviceClass = pvcTemplate.Annotations["crushDeviceClass"]
+		}
+		pvcSources[pvcType] = v1.PersistentVolumeClaimVolumeSource{
+			ClaimName: pvc.GetName(),
+			ReadOnly:  false,
+		}
+	}
+
+	return rookv1.VolumeSource{
+		Name:                deviceSet.Name,
+		Resources:           deviceSet.Resources,
+		Placement:           deviceSet.Placement,
+		PreparePlacement:    deviceSet.PreparePlacement,
+		Config:              deviceSet.Config,
+		Size:                dataSize,
+		PVCSources:          pvcSources,
+		Portable:            deviceSet.Portable,
+		TuneSlowDeviceClass: deviceSet.TuneSlowDeviceClass,
+		TuneFastDeviceClass: deviceSet.TuneFastDeviceClass,
+		SchedulerName:       deviceSet.SchedulerName,
+		CrushDeviceClass:    crushDeviceClass,
+		Encrypted:           deviceSet.Encrypted,
+	}
+}
+
+func (c *Cluster) createDeviceSetPVC(existingPVCs map[string]*v1.PersistentVolumeClaim, deviceSetName string, pvcTemplate v1.PersistentVolumeClaim, setIndex int) (*v1.PersistentVolumeClaim, error) {
 	ctx := context.TODO()
 	// old labels and PVC ID for backward compatibility
-	pvcStorageClassDeviceSetPVCId := legacyDeviceSetPVCID(storageClassDeviceSetName, setIndex)
+	pvcID := legacyDeviceSetPVCID(deviceSetName, setIndex)
 
 	// check for the existence of the pvc
-	existingPVC, ok := existingPVCs[pvcStorageClassDeviceSetPVCId]
+	existingPVC, ok := existingPVCs[pvcID]
 	if !ok {
 		// The old name of the PVC didn't exist, now try the new PVC name and label
-		pvcStorageClassDeviceSetPVCId = deviceSetPVCID(storageClassDeviceSetName, pvcTemplate.GetName(), setIndex)
-		existingPVC = existingPVCs[pvcStorageClassDeviceSetPVCId]
+		pvcID = deviceSetPVCID(deviceSetName, pvcTemplate.GetName(), setIndex)
+		existingPVC = existingPVCs[pvcID]
 	}
-	pvc := makeStorageClassDeviceSetPVC(storageClassDeviceSetName, pvcStorageClassDeviceSetPVCId, setIndex, pvcTemplate)
+	pvc := makeDeviceSetPVC(deviceSetName, pvcID, setIndex, pvcTemplate)
 	k8sutil.SetOwnerRef(&pvc.ObjectMeta, &c.clusterInfo.OwnerRef)
 
 	if existingPVC != nil {
@@ -134,9 +170,10 @@ func (c *Cluster) createStorageClassDeviceSetPVC(existingPVCs map[string]*v1.Per
 	// No PVC found, creating a new one
 	deployedPVC, err := c.context.Clientset.CoreV1().PersistentVolumeClaims(c.clusterInfo.Namespace).Create(ctx, pvc, metav1.CreateOptions{})
 	if err != nil {
-		return nil, errors.Wrapf(err, "failed to create pvc %q for storageClassDeviceSet %q", pvc.GetGenerateName(), storageClassDeviceSetName)
+		return nil, errors.Wrapf(err, "failed to create pvc %q for device set %q", pvc.Name, deviceSetName)
 	}
 	logger.Infof("successfully provisioned PVC %q", deployedPVC.Name)
+
 	return deployedPVC, nil
 }
 
@@ -162,20 +199,20 @@ func (c *Cluster) updatePVCIfChanged(desiredPVC *v1.PersistentVolumeClaim, curre
 	}
 }
 
-func makeStorageClassDeviceSetPVC(storageClassDeviceSetName, pvcStorageClassDeviceSetPVCId string, setIndex int, pvcTemplate v1.PersistentVolumeClaim) *v1.PersistentVolumeClaim {
-	pvcLabels := makeStorageClassDeviceSetPVCLabel(storageClassDeviceSetName, pvcStorageClassDeviceSetPVCId, setIndex)
-
-	// pvc naming format <storageClassDeviceSetName>-<SetNumber>-<PVCIndex>
-	pvcGenerateName := pvcStorageClassDeviceSetPVCId + "-"
+func makeDeviceSetPVC(deviceSetName, pvcID string, setIndex int, pvcTemplate v1.PersistentVolumeClaim) *v1.PersistentVolumeClaim {
+	pvcLabels := makeStorageClassDeviceSetPVCLabel(deviceSetName, pvcID, setIndex)
 
 	// Add user provided labels to pvcTemplates
 	for k, v := range pvcTemplate.GetLabels() {
 		pvcLabels[k] = v
 	}
 
+	// pvc naming format rook-ceph-osd-<deviceSetName>-<SetNumber>-<PVCIndex>-<generatedSuffix>
 	return &v1.PersistentVolumeClaim{
 		ObjectMeta: metav1.ObjectMeta{
-			GenerateName: pvcGenerateName,
+			// Use a generated name to avoid the possibility of two OSDs being created with the same ID.
+			// If one is removed and a new one is created later with the same ID, the OSD would fail to start.
+			GenerateName: pvcID,
 			Labels:       pvcLabels,
 			Annotations:  pvcTemplate.Annotations,
 		},
@@ -183,29 +220,40 @@ func makeStorageClassDeviceSetPVC(storageClassDeviceSetName, pvcStorageClassDevi
 	}
 }
 
-// GetExistingOSDPVCs fetches the list of OSD PVCs
-func GetExistingOSDPVCs(clusterdContext *clusterd.Context, namespace string) (map[string]*v1.PersistentVolumeClaim, error) {
+// GetExistingPVCs fetches the list of OSD PVCs
+func GetExistingPVCs(clusterdContext *clusterd.Context, namespace string) (map[string]*v1.PersistentVolumeClaim, map[string]*util.Set, error) {
 	ctx := context.TODO()
 	selector := metav1.ListOptions{LabelSelector: CephDeviceSetPVCIDLabelKey}
 	pvcs, err := clusterdContext.Clientset.CoreV1().PersistentVolumeClaims(namespace).List(ctx, selector)
 	if err != nil {
-		return nil, errors.Wrap(err, "failed to detect pvcs")
+		return nil, nil, errors.Wrap(err, "failed to detect pvcs")
 	}
 	result := map[string]*v1.PersistentVolumeClaim{}
+	uniqueOSDsPerDeviceSet := map[string]*util.Set{}
 	for i, pvc := range pvcs.Items {
+		// Populate the PVCs based on their unique name across all the device sets
 		pvcID := pvc.Labels[CephDeviceSetPVCIDLabelKey]
 		result[pvcID] = &pvcs.Items[i]
+
+		// Create a map of the PVC IDs available in each device set based on PVC index
+		deviceSet := pvc.Labels[CephDeviceSetLabelKey]
+		pvcIndex := pvc.Labels[CephSetIndexLabelKey]
+		if _, ok := uniqueOSDsPerDeviceSet[deviceSet]; !ok {
+			uniqueOSDsPerDeviceSet[deviceSet] = util.NewSet()
+		}
+		uniqueOSDsPerDeviceSet[deviceSet].Add(pvcIndex)
 	}
 
-	return result, nil
+	return result, uniqueOSDsPerDeviceSet, nil
 }
 
-func legacyDeviceSetPVCID(storageClassDeviceSetName string, setIndex int) string {
-	return fmt.Sprintf("%s-%d", storageClassDeviceSetName, setIndex)
+func legacyDeviceSetPVCID(deviceSetName string, setIndex int) string {
+	return fmt.Sprintf("%s-%d", deviceSetName, setIndex)
 }
 
 // This is the new function that generates the labels
 // It includes the pvcTemplateName in it
-func deviceSetPVCID(storageClassDeviceSetName, pvcTemplateName string, setIndex int) string {
-	return fmt.Sprintf("%s-%s-%d", storageClassDeviceSetName, strings.Replace(pvcTemplateName, " ", "-", -1), setIndex)
+func deviceSetPVCID(deviceSetName, pvcTemplateName string, setIndex int) string {
+	cleanName := strings.Replace(pvcTemplateName, " ", "-", -1)
+	return fmt.Sprintf("%s-%s-%d", deviceSetName, cleanName, setIndex)
 }

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

@@ -26,9 +26,13 @@ import (
 	"github.com/rook/rook/pkg/daemon/ceph/client"
 	testexec "github.com/rook/rook/pkg/operator/test"
 	"github.com/stretchr/testify/assert"
+	corev1 "k8s.io/api/core/v1"
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/kubernetes"
+	k8stesting "k8s.io/client-go/testing"
 )
 
 func TestPrepareDeviceSets(t *testing.T) {
@@ -42,10 +46,7 @@ func testPrepareDeviceSets(t *testing.T, setTemplateName bool) {
 	context := &clusterd.Context{
 		Clientset: clientset,
 	}
-	storageClass := "mysource"
-	claim := v1.PersistentVolumeClaim{Spec: v1.PersistentVolumeClaimSpec{
-		StorageClassName: &storageClass,
-	}}
+	claim := testVolumeClaim("")
 	if setTemplateName {
 		claim.Name = "randomname"
 	}
@@ -83,10 +84,157 @@ func testPrepareDeviceSets(t *testing.T, setTemplateName bool) {
 	if !setTemplateName {
 		expectedName = "data"
 	}
-	assert.Equal(t, fmt.Sprintf("mydata-%s-0-", expectedName), pvcs.Items[0].GenerateName)
+	assert.Equal(t, fmt.Sprintf("mydata-%s-0", expectedName), pvcs.Items[0].GenerateName)
 	assert.Equal(t, cluster.clusterInfo.Namespace, pvcs.Items[0].Namespace)
 }
 
+func TestPrepareDeviceSetWithHolesInPVCs(t *testing.T) {
+	ctx := context.TODO()
+	clientset := testexec.New(t, 1)
+	context := &clusterd.Context{
+		Clientset: clientset,
+	}
+
+	deviceSet := rookv1.StorageClassDeviceSet{
+		Name:                 "mydata",
+		Count:                1,
+		Portable:             true,
+		VolumeClaimTemplates: []v1.PersistentVolumeClaim{testVolumeClaim("data"), testVolumeClaim("metadata"), testVolumeClaim("wal")},
+		SchedulerName:        "custom-scheduler",
+	}
+	spec := cephv1.ClusterSpec{
+		Storage: rookv1.StorageScopeSpec{StorageClassDeviceSets: []rookv1.StorageClassDeviceSet{deviceSet}},
+	}
+	ns := "testns"
+	cluster := &Cluster{
+		context:     context,
+		clusterInfo: client.AdminClusterInfo(ns),
+		spec:        spec,
+	}
+
+	pvcSuffix := 0
+	var pvcReactor k8stesting.ReactionFunc = func(action k8stesting.Action) (handled bool, ret runtime.Object, err error) {
+		// PVCs are created with generateName used, and we need to capture the create calls and
+		// generate a name for them in order for PVCs to all have unique names.
+		createAction, ok := action.(k8stesting.CreateAction)
+		if !ok {
+			t.Fatal("err! action is not a create action")
+			return false, nil, nil
+		}
+		obj := createAction.GetObject()
+		pvc, ok := obj.(*corev1.PersistentVolumeClaim)
+		if !ok {
+			t.Fatal("err! action not a PVC")
+			return false, nil, nil
+		}
+		if pvc.Name == "" {
+			pvc.Name = fmt.Sprintf("%s-%d", pvc.GenerateName, pvcSuffix)
+			logger.Info("generated name for PVC:", pvc.Name)
+			pvcSuffix++
+		} else {
+			logger.Info("PVC already has a name:", pvc.Name)
+		}
+		// setting pvc.Name above modifies the action in-place before future reactors occur
+		// we want the default reactor to create the resource, so return false as if we did nothing
+		return false, nil, nil
+	}
+	clientset.PrependReactor("create", "persistentvolumeclaims", pvcReactor)
+
+	// Create 3 PVCs for two OSDs in the device set
+	config := &provisionConfig{}
+	volumeSources := cluster.prepareStorageClassDeviceSets(config)
+	assert.Equal(t, 1, len(volumeSources))
+	assert.Equal(t, 0, len(config.errorMessages))
+	assert.Equal(t, "mydata", volumeSources[0].Name)
+	assert.True(t, volumeSources[0].Portable)
+	_, dataOK := volumeSources[0].PVCSources["data"]
+	assert.True(t, dataOK)
+
+	// Verify the PVCs all exist
+	pvcs, err := clientset.CoreV1().PersistentVolumeClaims(cluster.clusterInfo.Namespace).List(ctx, metav1.ListOptions{})
+	assert.NoError(t, err)
+	assert.Equal(t, 3, len(pvcs.Items))
+	assertPVCExists(t, clientset, ns, "mydata-data-0-0")
+	assertPVCExists(t, clientset, ns, "mydata-metadata-0-1")
+	assertPVCExists(t, clientset, ns, "mydata-wal-0-2")
+
+	// Create 3 more PVCs (6 total) for two OSDs in the device set
+	cluster.spec.Storage.StorageClassDeviceSets[0].Count = 2
+	volumeSources = cluster.prepareStorageClassDeviceSets(config)
+	assert.Equal(t, 2, len(volumeSources))
+	assert.Equal(t, 0, len(config.errorMessages))
+
+	// Verify the PVCs all exist
+	pvcs, err = clientset.CoreV1().PersistentVolumeClaims(cluster.clusterInfo.Namespace).List(ctx, metav1.ListOptions{})
+	assert.NoError(t, err)
+	assert.Equal(t, 6, len(pvcs.Items))
+	assertPVCExists(t, clientset, ns, "mydata-data-0-0")
+	assertPVCExists(t, clientset, ns, "mydata-metadata-0-1")
+	assertPVCExists(t, clientset, ns, "mydata-wal-0-2")
+	assertPVCExists(t, clientset, ns, "mydata-data-1-3")
+	assertPVCExists(t, clientset, ns, "mydata-metadata-1-4")
+	assertPVCExists(t, clientset, ns, "mydata-wal-1-5")
+
+	// Verify the same number of PVCs exist after calling the reconcile again on the PVCs
+	volumeSources = cluster.prepareStorageClassDeviceSets(config)
+	assert.Equal(t, 2, len(volumeSources))
+	pvcs, err = clientset.CoreV1().PersistentVolumeClaims(cluster.clusterInfo.Namespace).List(ctx, metav1.ListOptions{})
+	assert.NoError(t, err)
+	assert.Equal(t, 6, len(pvcs.Items))
+
+	// Delete a single PVC and verify it will be re-created
+	err = clientset.CoreV1().PersistentVolumeClaims(cluster.clusterInfo.Namespace).Delete(ctx, "mydata-wal-0-2", metav1.DeleteOptions{})
+	assert.NoError(t, err)
+	volumeSources = cluster.prepareStorageClassDeviceSets(config)
+	assert.Equal(t, 2, len(volumeSources))
+	pvcs, err = clientset.CoreV1().PersistentVolumeClaims(cluster.clusterInfo.Namespace).List(ctx, metav1.ListOptions{})
+	assert.NoError(t, err)
+	assert.Equal(t, 6, len(pvcs.Items))
+
+	// Delete the PVCs for an OSD and verify it will not be re-created if the count is reduced
+	cluster.spec.Storage.StorageClassDeviceSets[0].Count = 1
+	err = clientset.CoreV1().PersistentVolumeClaims(cluster.clusterInfo.Namespace).Delete(ctx, "mydata-data-0-0", metav1.DeleteOptions{})
+	assert.NoError(t, err)
+	err = clientset.CoreV1().PersistentVolumeClaims(cluster.clusterInfo.Namespace).Delete(ctx, "mydata-metadata-0-1", metav1.DeleteOptions{})
+	assert.NoError(t, err)
+	err = clientset.CoreV1().PersistentVolumeClaims(cluster.clusterInfo.Namespace).Delete(ctx, "mydata-wal-0-6", metav1.DeleteOptions{})
+	assert.NoError(t, err)
+	volumeSources = cluster.prepareStorageClassDeviceSets(config)
+	assert.Equal(t, 1, len(volumeSources))
+	pvcs, err = clientset.CoreV1().PersistentVolumeClaims(cluster.clusterInfo.Namespace).List(ctx, metav1.ListOptions{})
+	assert.NoError(t, err)
+	assert.Equal(t, 3, len(pvcs.Items))
+
+	// Scale back up to a count of two and confirm that a new index is used for the PVCs
+	cluster.spec.Storage.StorageClassDeviceSets[0].Count = 2
+	volumeSources = cluster.prepareStorageClassDeviceSets(config)
+	assert.Equal(t, 2, len(volumeSources))
+	pvcs, err = clientset.CoreV1().PersistentVolumeClaims(cluster.clusterInfo.Namespace).List(ctx, metav1.ListOptions{})
+	assert.NoError(t, err)
+	assert.Equal(t, 6, len(pvcs.Items))
+	assertPVCExists(t, clientset, ns, "mydata-data-1-3")
+	assertPVCExists(t, clientset, ns, "mydata-metadata-1-4")
+	assertPVCExists(t, clientset, ns, "mydata-wal-1-5")
+	assertPVCExists(t, clientset, ns, "mydata-data-2-7")
+	assertPVCExists(t, clientset, ns, "mydata-metadata-2-8")
+	assertPVCExists(t, clientset, ns, "mydata-wal-2-9")
+}
+
+func assertPVCExists(t *testing.T, clientset kubernetes.Interface, namespace, name string) {
+	pvc, err := clientset.CoreV1().PersistentVolumeClaims(namespace).Get(context.TODO(), name, metav1.GetOptions{})
+	assert.NoError(t, err)
+	assert.NotNil(t, pvc)
+}
+
+func testVolumeClaim(name string) v1.PersistentVolumeClaim {
+	storageClass := "mysource"
+	claim := v1.PersistentVolumeClaim{Spec: v1.PersistentVolumeClaimSpec{
+		StorageClassName: &storageClass,
+	}}
+	claim.Name = name
+	return claim
+}
+
 func TestUpdatePVCSize(t *testing.T) {
 	clientset := testexec.New(t, 1)
 	context := &clusterd.Context{