csi: add check for pending clone

Incase any pvc is in pending state and the backend
subvolume is create, it will be counted as stale.
this commit adds the error check to make sure user
doesn't delete the subvolume if it the pvc is in
pending state.

Signed-off-by: yati1998 <[email protected]>

pkg/filesystem/subvolume.go

@@ -175,8 +175,13 @@ func listCephFSSubvolumes(ctx context.Context, clientsets *k8sutil.Clientsets, o
 			}
 			// append the subvolume which doesn't have any snapshot attached to it.
 			for _, sv := range subvol {
-				state := getSubvolumeState(ctx, clientsets, operatorNamespace, clusterNamespace, fs.Name, sv.Name, svg.Name)
-
+				state, err := getSubvolumeState(ctx, clientsets, operatorNamespace, clusterNamespace, fs.Name, sv.Name, svg.Name)
+				// subvolume info returns error in case of pending clone or if it is not ready
+				// it is suggested to delete the pvc before deleting the subvolume.
+				if err != nil {
+					logging.Info("Subvolume info not found for: %q", err)
+					logging.Info("Please delete the pending pv if any before deleting the subvolume %s", sv)
+				}
 				// Assume the volume is stale unless proven otherwise
 				stalevol := true
 				// lookup for subvolume in list of the PV references
@@ -195,12 +200,14 @@ func listCephFSSubvolumes(ctx context.Context, clientsets *k8sutil.Clientsets, o
 					// check the state of the stale subvolume
 					// if it is snapshot-retained then skip listing it.
 					if state == "snapshot-retained" {
-						status = state
 						continue
 					}
-					// check if the stale subvolume has snapshots.
-					if checkSnapshot(ctx, clientsets, operatorNamespace, clusterNamespace, fs.Name, sv.Name, svg.Name) {
-						status = staleWithSnapshot
+					// check for snapshot only if the subvolume info doesn't return any error.
+					if err == nil {
+						// check if the stale subvolume has snapshots.
+						if checkSnapshot(ctx, clientsets, operatorNamespace, clusterNamespace, fs.Name, sv.Name, svg.Name) {
+							status = staleWithSnapshot
+						}
 					}
 
 				}
@@ -212,14 +219,14 @@ func listCephFSSubvolumes(ctx context.Context, clientsets *k8sutil.Clientsets, o
 }
 
 // getSubvolumeState returns the state of the subvolume
-func getSubvolumeState(ctx context.Context, clientsets *k8sutil.Clientsets, operatorNamespace, clusterNamespace, fsName, SubVol, SubvolumeGroup string) string {
+func getSubvolumeState(ctx context.Context, clientsets *k8sutil.Clientsets, operatorNamespace, clusterNamespace, fsName, SubVol, SubvolumeGroup string) (string, error) {
 	cmd := "ceph"
 	args := []string{"fs", "subvolume", "info", fsName, SubVol, SubvolumeGroup, "--format", "json"}
 
 	subVolumeInfo, errvol := runCommand(ctx, clientsets, operatorNamespace, clusterNamespace, cmd, args)
 	if errvol != nil {
-		logging.Error(errvol, "failed to get filesystems")
-		return ""
+		logging.Error(errvol, "failed to get subvolume info")
+		return "", errvol
 	}
 	var info map[string]interface{}
 	err := json.Unmarshal([]byte(subVolumeInfo), &info)
@@ -230,7 +237,7 @@ func getSubvolumeState(ctx context.Context, clientsets *k8sutil.Clientsets, oper
 	if !ok {
 		logging.Fatal(fmt.Errorf("failed to get the state of subvolume: %q", SubVol))
 	}
-	return state
+	return state, nil
 }
 
 // gets list of filesystem