rbd: Update return error massage

Issue: When delete pv failed, error message shows 'volume deletion failed: rpc error:
code = Internal desc = rbd: ret=-39, Directory not empty', the actual failed reason is 'access denied'

This commit ensures ceph-csi return right error massage.

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

internal/rbd/rbd_util.go

@@ -600,25 +600,29 @@ func isNotMountPoint(mounter mount.Interface, stagingTargetPath string) (bool, e
 
 // isCephMgrSupported determines if the cluster has support for MGR based operation
 // depending on the error.
-func isCephMgrSupported(ctx context.Context, clusterID string, err error) bool {
+func isCephMgrSupported(ctx context.Context, clusterID string, err error) (bool, error) {
 	switch {
 	case err == nil:
-		return true
+		return true, nil
 	case strings.Contains(err.Error(), rbdTaskRemoveCmdInvalidString):
+		msg := fmt.Sprintf("cluster with cluster ID (%s) does not support Ceph manager based rbd commands"+
+			"(minimum ceph version required is v14.2.3)",
+			clusterID)
 		log.WarningLog(
 			ctx,
-			"cluster with cluster ID (%s) does not support Ceph manager based rbd commands"+
-				"(minimum ceph version required is v14.2.3)",
-			clusterID)
+			msg)
 
-		return false
+		return false, errors.New(msg)
 	case strings.Contains(err.Error(), rbdTaskRemoveCmdAccessDeniedMessage):
-		log.WarningLog(ctx, "access denied to Ceph MGR-based rbd commands on cluster ID (%s)", clusterID)
+		msg := fmt.Sprintf("access denied to Ceph MGR-based rbd commands on cluster ID (%s)",
+			clusterID)
+		log.WarningLog(ctx,
+			msg)
 
-		return false
+		return false, errors.New(msg)
 	}
 
-	return true
+	return true, nil
 }
 
 // ensureImageCleanup finds image in trash and if found removes it
@@ -705,20 +709,20 @@ func (ri *rbdImage) trashRemoveImage(ctx context.Context) error {
 
 	_, err = ta.AddTrashRemove(admin.NewImageSpec(ri.Pool, ri.RadosNamespace, ri.ImageID))
 
-	rbdCephMgrSupported := isCephMgrSupported(ctx, ri.ClusterID, err)
+	rbdCephMgrSupported, knownErr := isCephMgrSupported(ctx, ri.ClusterID, err)
 	if rbdCephMgrSupported && err != nil {
 		log.ErrorLog(ctx, "failed to add task to delete rbd image: %s, %v", ri, err)
 
 		return err
 	}
 
-	if !rbdCephMgrSupported {
+	if !rbdCephMgrSupported && knownErr != nil {
 		err = librbd.TrashRemove(ri.ioctx, ri.ImageID, true)
 		if err != nil {
-			log.ErrorLog(ctx, "failed to delete rbd image: %s, %v", ri, err)
-
-			return err
+			log.WarningLog(ctx, "failed to delete rbd image: %s, %v", ri, err)
 		}
+
+		return knownErr
 	} else {
 		log.DebugLog(ctx, "rbd: successfully added task to move image %q with id %q to trash", ri, ri.ImageID)
 	}
@@ -852,7 +856,7 @@ func (ri *rbdImage) flattenRbdImage(
 	}
 
 	_, err = ta.AddFlatten(admin.NewImageSpec(ri.Pool, ri.RadosNamespace, ri.RbdImageName))
-	rbdCephMgrSupported := isCephMgrSupported(ctx, ri.ClusterID, err)
+	rbdCephMgrSupported, knownErr := isCephMgrSupported(ctx, ri.ClusterID, err)
 	if rbdCephMgrSupported {
 		if err != nil {
 			// discard flattening error if the image does not have any parent
@@ -869,7 +873,7 @@ func (ri *rbdImage) flattenRbdImage(
 		}
 		log.DebugLog(ctx, "successfully added task to flatten image %q", ri)
 	}
-	if !rbdCephMgrSupported {
+	if !rbdCephMgrSupported && knownErr != nil {
 		log.ErrorLog(
 			ctx,
 			"task manager does not support flatten,image will be flattened once hardlimit is reached: %v",
@@ -878,10 +882,9 @@ func (ri *rbdImage) flattenRbdImage(
 			err := ri.flatten()
 			if err != nil {
 				log.ErrorLog(ctx, "rbd failed to flatten image %s %s: %v", ri.Pool, ri.RbdImageName, err)
-
-				return err
 			}
 		}
+		return knownErr
 	}
 
 	return nil