diff --git a/Documentation/ceph-osd-mgmt.md b/Documentation/ceph-osd-mgmt.md index 0376a74a9..0b47c7677 100644 --- a/Documentation/ceph-osd-mgmt.md +++ b/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":}]'` - - 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=` - - 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= -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 `` 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 diff --git a/cluster/examples/kubernetes/ceph/osd-purge.yaml b/cluster/examples/kubernetes/ceph/osd-purge.yaml index 845d2477d..3923f413c 100644 --- a/cluster/examples/kubernetes/ceph/osd-purge.yaml +++ b/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 diff --git a/cmd/rook/ceph/osd.go b/cmd/rook/ceph/osd.go index 874bdc13a..7a0cb55dc 100644 --- a/cmd/rook/ceph/osd.go +++ b/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) diff --git a/pkg/operator/ceph/cluster/controller.go b/pkg/operator/ceph/cluster/controller.go index c3264a0cc..a687c451e 100644 --- a/pkg/operator/ceph/cluster/controller.go +++ b/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") } diff --git a/pkg/operator/ceph/cluster/osd/deviceSet.go b/pkg/operator/ceph/cluster/osd/deviceSet.go index ef5c89b7b..e91e6f932 100644 --- a/pkg/operator/ceph/cluster/osd/deviceSet.go +++ b/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 -- - 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---- 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) } diff --git a/pkg/operator/ceph/cluster/osd/deviceset_test.go b/pkg/operator/ceph/cluster/osd/deviceset_test.go index f04011dca..a83c33f22 100644 --- a/pkg/operator/ceph/cluster/osd/deviceset_test.go +++ b/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{