Handle Delete operation if pool not found

If the backend rbd or cephfs pool is already deleted
we need to return success to the  DeleteVolume RPC
call to make it idempotent.

Signed-off-by: Madhu Rajanna <madhupr007@gmail.com>

pkg/cephfs/cephfs_util.go

@@ -84,10 +84,10 @@ func getMetadataPool(ctx context.Context, monitors string, cr *util.Credentials,
 		}
 	}
 
-	return "", fmt.Errorf("fsName (%s) not found in Ceph cluster", fsName)
+	return "", util.ErrPoolNotFound{Pool: fsName, Err: fmt.Errorf("fsName (%s) not found in Ceph cluster", fsName)}
 }
 
-// CephFilesystemDetails is a representation of the main json structure returned by 'ceph fs dump'
+// CephFilesystemDump is a representation of the main json structure returned by 'ceph fs dump'
 type CephFilesystemDump struct {
 	Filesystems []CephFilesystemDetails `json:"filesystems"`
 }
@@ -114,5 +114,5 @@ func getFsName(ctx context.Context, monitors string, cr *util.Credentials, fscID
 		}
 	}
 
-	return "", fmt.Errorf("fscID (%d) not found in Ceph cluster", fscID)
+	return "", util.ErrPoolNotFound{Pool: string(fscID), Err: fmt.Errorf("fscID (%d) not found in Ceph cluster", fscID)}
 }

pkg/cephfs/controllerserver.go

@@ -231,6 +231,12 @@ func (cs *ControllerServer) DeleteVolume(ctx context.Context, req *csi.DeleteVol
 	// Find the volume using the provided VolumeID
 	volOptions, vID, err := newVolumeOptionsFromVolID(ctx, string(volID), nil, secrets)
 	if err != nil {
+		// if error is ErrPoolNotFound, the pool is already deleted we dont
+		// need to worry about deleting subvolume or omap data, return success
+		if _, ok := err.(util.ErrPoolNotFound); ok {
+			klog.Warningf(util.Log(ctx, "failed to get backend volume for %s: %v"), string(volID), err)
+			return &csi.DeleteVolumeResponse{}, nil
+		}
 		// if error is ErrKeyNotFound, then a previous attempt at deletion was complete
 		// or partially complete (subvolume and imageOMap are garbage collected already), hence
 		// return success as deletion is complete

pkg/rbd/controllerserver.go

@@ -258,6 +258,12 @@ func (cs *ControllerServer) checkSnapshot(ctx context.Context, req *csi.CreateVo
 		if _, ok := err.(ErrSnapNotFound); !ok {
 			return status.Error(codes.Internal, err.Error())
 		}
+
+		if _, ok := err.(util.ErrPoolNotFound); ok {
+			klog.Errorf(util.Log(ctx, "failed to get backend snapshot for %s: %v"), snapshotID, err)
+			return status.Error(codes.InvalidArgument, err.Error())
+		}
+
 		return status.Error(codes.InvalidArgument, "missing requested Snapshot ID")
 	}
 
@@ -344,6 +350,11 @@ func (cs *ControllerServer) DeleteVolume(ctx context.Context, req *csi.DeleteVol
 
 	rbdVol := &rbdVolume{}
 	if err = genVolFromVolID(ctx, rbdVol, volumeID, cr, req.GetSecrets()); err != nil {
+		if _, ok := err.(util.ErrPoolNotFound); ok {
+			klog.Warningf(util.Log(ctx, "failed to get backend volume for %s: %v"), volumeID, err)
+			return &csi.DeleteVolumeResponse{}, nil
+		}
+
 		// If error is ErrInvalidVolID it could be a version 1.0.0 or lower volume, attempt
 		// to process it as such
 		if _, ok := err.(ErrInvalidVolID); ok {
@@ -360,6 +371,7 @@ func (cs *ControllerServer) DeleteVolume(ctx context.Context, req *csi.DeleteVol
 		// or partially complete (image and imageOMap are garbage collected already), hence return
 		// success as deletion is complete
 		if _, ok := err.(util.ErrKeyNotFound); ok {
+			klog.Warningf(util.Log(ctx, "Failed to volume options for %s: %v"), volumeID, err)
 			return &csi.DeleteVolumeResponse{}, nil
 		}
 
@@ -458,6 +470,11 @@ func (cs *ControllerServer) CreateSnapshot(ctx context.Context, req *csi.CreateS
 		if _, ok := err.(ErrImageNotFound); ok {
 			return nil, status.Errorf(codes.NotFound, "source Volume ID %s not found", req.GetSourceVolumeId())
 		}
+
+		if _, ok := err.(util.ErrPoolNotFound); ok {
+			klog.Errorf(util.Log(ctx, "failed to get backend volume for %s: %v"), req.GetSourceVolumeId(), err)
+			return nil, status.Errorf(codes.Internal, err.Error())
+		}
 		return nil, status.Errorf(codes.Internal, err.Error())
 	}
 
@@ -625,6 +642,13 @@ func (cs *ControllerServer) DeleteSnapshot(ctx context.Context, req *csi.DeleteS
 
 	rbdSnap := &rbdSnapshot{}
 	if err = genSnapFromSnapID(ctx, rbdSnap, snapshotID, cr); err != nil {
+		// if error is ErrPoolNotFound, the pool is already deleted we dont
+		// need to worry about deleting snapshot or omap data, return success
+		if _, ok := err.(util.ErrPoolNotFound); ok {
+			klog.Warningf(util.Log(ctx, "failed to get backend snapshot for %s: %v"), snapshotID, err)
+			return &csi.DeleteSnapshotResponse{}, nil
+		}
+
 		// if error is ErrKeyNotFound, then a previous attempt at deletion was complete
 		// or partially complete (snap and snapOMap are garbage collected already), hence return
 		// success as deletion is complete
@@ -715,6 +739,12 @@ func (cs *ControllerServer) ControllerExpandVolume(ctx context.Context, req *csi
 		if _, ok := err.(ErrImageNotFound); ok {
 			return nil, status.Errorf(codes.NotFound, "volume ID %s not found", volID)
 		}
+
+		if _, ok := err.(util.ErrPoolNotFound); ok {
+			klog.Errorf(util.Log(ctx, "failed to get backend volume for %s: %v"), volID, err)
+			return nil, status.Errorf(codes.InvalidArgument, err.Error())
+		}
+
 		return nil, status.Errorf(codes.Internal, err.Error())
 	}
 

pkg/util/cephcmds.go

@@ -114,7 +114,7 @@ func GetPoolName(ctx context.Context, monitors string, cr *Credentials, poolID i
 		}
 	}
 
-	return "", fmt.Errorf("pool ID (%d) not found in Ceph cluster", poolID)
+	return "", ErrPoolNotFound{string(poolID), fmt.Errorf("pool ID (%d) not found in Ceph cluster", poolID)}
 }
 
 // SetOMapKeyValue sets the given key and value into the provided Ceph omap name

pkg/util/errors.go

@@ -56,3 +56,13 @@ type ErrSnapNameConflict struct {
 func (e ErrSnapNameConflict) Error() string {
 	return e.err.Error()
 }
+
+// ErrPoolNotFound is returned when pool is not found
+type ErrPoolNotFound struct {
+	Pool string
+	Err  error
+}
+
+func (e ErrPoolNotFound) Error() string {
+	return e.Err.Error()
+}