Add resize check for XFS formatted FS

Lock out parellel requests against same volumeID
Remove pod after resize and validation in E2E

Signed-off-by: Humble Chirammal <hchiramm@redhat.com>

e2e/rbd.go

@@ -116,7 +116,6 @@ var _ = Describe("RBD", func() {
 			By("create a PVC and Bind it to an app with normal user", func() {
 				validateNormalUserPVCAccess(pvcPath, f)
 			})
-			// Skipping ext4 FS testing
 
 			By("create a PVC and Bind it to an app with ext4 as the FS ", func() {
 				deleteResource(rbdExamplePath + "storageclass.yaml")
@@ -250,8 +249,16 @@ var _ = Describe("RBD", func() {
 						e2elog.Logf("failed to resize PVC %v", err)
 						Fail(err.Error())
 					}
-				}
 
+					deleteResource(rbdExamplePath + "storageclass.yaml")
+					createRBDStorageClass(f.ClientSet, f, map[string]string{"csi.storage.k8s.io/fstype": "xfs"})
+					err = resizePVCAndValidateSize(pvcPath, appPath, f)
+					if err != nil {
+						e2elog.Logf("failed to resize PVC %v", err)
+						Fail(err.Error())
+
+					}
+				}
 			})
 
 			By("Test unmount after nodeplugin restart", func() {

e2e/utils.go

@@ -834,12 +834,17 @@ func expandPVCSize(c kubernetes.Interface, pvc *v1.PersistentVolumeClaim, size s
 		}
 		pvcConditions := updatedPVC.Status.Conditions
 		if len(pvcConditions) > 0 {
-			if pvcConditions[0].Type == v1.PersistentVolumeClaimResizing {
-				return true, nil
-			}
 			e2elog.Logf("pvc state %v", pvcConditions[0].Type)
+			if pvcConditions[0].Type == v1.PersistentVolumeClaimResizing || pvcConditions[0].Type == v1.PersistentVolumeClaimFileSystemResizePending {
+				return false, nil
+			}
 		}
-		return false, nil
+
+		if updatedPVC.Status.Capacity[v1.ResourceStorage] != resource.MustParse(size) {
+			e2elog.Logf("current size in status %v,expected size %v", updatedPVC.Status.Capacity[v1.ResourceStorage], resource.MustParse(size))
+			return false, nil
+		}
+		return true, nil
 	})
 }
 
@@ -889,6 +894,10 @@ func resizePVCAndValidateSize(pvcPath, appPath string, f *framework.Framework) e
 		return err
 	}
 	err = checkDirSize(app, f, &opt, expandSize, deployTimeout)
+	if err != nil {
+		return err
+	}
+	err = deletePVCAndApp("", f, resizePvc, app)
 	return err
 }
 

pkg/rbd/controllerserver.go

@@ -535,7 +535,6 @@ func (cs *ControllerServer) doSnapshot(ctx context.Context, rbdSnap *rbdSnapshot
 			err = status.Error(codes.Internal, err.Error())
 		}
 	}()
-
 	err = protectSnapshot(ctx, rbdSnap, cr)
 	if err != nil {
 		klog.Errorf(util.Log(ctx, "failed to protect snapshot: %v"), err)
@@ -645,7 +644,6 @@ func (cs *ControllerServer) DeleteSnapshot(ctx context.Context, req *csi.DeleteS
 
 // ControllerExpandVolume expand RBD Volumes on demand based on resizer request
 func (cs *ControllerServer) ControllerExpandVolume(ctx context.Context, req *csi.ControllerExpandVolumeRequest) (*csi.ControllerExpandVolumeResponse, error) {
-
 	if err := cs.Driver.ValidateControllerServiceRequest(csi.ControllerServiceCapability_RPC_EXPAND_VOLUME); err != nil {
 		klog.Warningf("invalid expand volume req: %v", protosanitizer.StripSecrets(req))
 		return nil, err
@@ -653,14 +651,21 @@ func (cs *ControllerServer) ControllerExpandVolume(ctx context.Context, req *csi
 
 	volID := req.GetVolumeId()
 	if volID == "" {
-		return nil, status.Error(codes.InvalidArgument, "Volume ID cannot be empty")
+		return nil, status.Error(codes.InvalidArgument, "volume ID cannot be empty")
 	}
 
 	capRange := req.GetCapacityRange()
 	if capRange == nil {
-		return nil, status.Error(codes.InvalidArgument, "CapacityRange cannot be empty")
+		return nil, status.Error(codes.InvalidArgument, "capacityRange cannot be empty")
 	}
 
+	// lock out parallel requests against the same volume ID
+	if acquired := cs.VolumeLocks.TryAcquire(volID); !acquired {
+		klog.Infof(util.Log(ctx, util.VolumeOperationAlreadyExistsFmt), volID)
+		return nil, status.Errorf(codes.Aborted, util.VolumeOperationAlreadyExistsFmt, volID)
+	}
+	defer cs.VolumeLocks.Release(volID)
+
 	cr, err := util.NewUserCredentials(req.GetSecrets())
 	if err != nil {
 		return nil, status.Error(codes.Internal, err.Error())
@@ -670,28 +675,31 @@ func (cs *ControllerServer) ControllerExpandVolume(ctx context.Context, req *csi
 	rbdVol := &rbdVolume{}
 	err = genVolFromVolID(ctx, rbdVol, volID, cr)
 	if err != nil {
-		return nil, status.Errorf(codes.NotFound, "Source Volume ID %s cannot found", volID)
+		if _, ok := err.(ErrImageNotFound); ok {
+			return nil, status.Errorf(codes.NotFound, "volume ID %s not found", volID)
+		}
+		return nil, status.Errorf(codes.Internal, err.Error())
 	}
 
 	// always round up the request size in bytes to the nearest MiB/GiB
 	volSize := util.RoundOffBytes(req.GetCapacityRange().GetRequiredBytes())
 
-	klog.V(4).Infof(util.Log(ctx, "rbd volume %s/%s size is %vMiB,resizing to %vMiB"), rbdVol.Pool, rbdVol.RbdImageName, rbdVol.VolSize, util.RoundOffVolSize(volSize))
 	// resize volume if required
+	nodeExpansion := false
 	if rbdVol.VolSize < volSize {
-		rbdVol.VolSize = util.RoundOffVolSize(volSize)
+		volSizeVal := util.RoundOffVolSize(volSize)
-		err = resizeRBDImage(ctx, rbdVol, cr)
+		klog.V(4).Infof(util.Log(ctx, "rbd volume %s/%s size is %v,resizing to %v"), rbdVol.Pool, rbdVol.RbdImageName, rbdVol.VolSize, volSize)
+		rbdVol.VolSize = volSize
+		nodeExpansion = true
+		err = resizeRBDImage(rbdVol, volSizeVal, cr)
 		if err != nil {
-			klog.Errorf("failed to resize rbd image: %s/%s with error: %v", rbdVol.Pool, rbdVol.RbdImageName, err)
+			klog.Errorf(util.Log(ctx, "failed to resize rbd image: %s/%s with error: %v"), rbdVol.Pool, rbdVol.RbdImageName, err)
 			return nil, status.Error(codes.Internal, err.Error())
 		}
-		klog.V(4).Infof(util.Log(ctx, "successfully resized %s to %v"), volID, volSize)
 	}
 
-	nodeExpansion := true
 	return &csi.ControllerExpandVolumeResponse{
-		CapacityBytes:         volSize,
+		CapacityBytes:         rbdVol.VolSize,
 		NodeExpansionRequired: nodeExpansion,
 	}, nil
-
 }

pkg/rbd/nodeserver.go

@@ -538,45 +538,51 @@ func (ns *NodeServer) NodeUnstageVolume(ctx context.Context, req *csi.NodeUnstag
 }
 
 func (ns *NodeServer) NodeExpandVolume(ctx context.Context, req *csi.NodeExpandVolumeRequest) (*csi.NodeExpandVolumeResponse, error) {
-
 	volumeID := req.GetVolumeId()
 	if volumeID == "" {
-		return nil, status.Error(codes.InvalidArgument, "NodeExpandVolume volume ID must be provided")
+		return nil, status.Error(codes.InvalidArgument, "volume ID must be provided")
 	}
 	volumePath := req.GetVolumePath()
 	if volumePath == "" {
-		return nil, status.Error(codes.InvalidArgument, "NodeExpandVolume volume path must be provided")
+		return nil, status.Error(codes.InvalidArgument, "volume path must be provided")
+	}
+
+	if acquired := ns.VolumeLocks.TryAcquire(volumeID); !acquired {
+		klog.Infof(util.Log(ctx, util.VolumeOperationAlreadyExistsFmt), volumeID)
+		return nil, status.Errorf(codes.Aborted, util.VolumeOperationAlreadyExistsFmt, volumeID)
+	}
+	defer ns.VolumeLocks.Release(volumeID)
+
+	_, err := getVolumeName(volumeID)
+	if err != nil {
+		return nil, status.Error(codes.InvalidArgument, err.Error())
 	}
 
 	diskMounter := &mount.SafeFormatAndMount{Interface: ns.mounter, Exec: mount.NewOsExec()}
-	volumePath = volumePath + "/" + req.GetVolumeId()
+	devicePath, err := getDevicePath(ctx, volumePath)
-	devicePath, err := getDevicePath(ctx, diskMounter, volumePath)
 	if err != nil {
 		return nil, status.Error(codes.Internal, err.Error())
 	}
 	// TODO check size and return success or error
+	volumePath += "/" + volumeID
 	resizer := resizefs.NewResizeFs(diskMounter)
 	ok, err := resizer.Resize(devicePath, volumePath)
 	if !ok {
 		return nil, fmt.Errorf("rbd: resize failed on path %s, error: %v", req.GetVolumePath(), err)
 	}
-	return &csi.NodeExpandVolumeResponse{
+	return &csi.NodeExpandVolumeResponse{}, nil
-		CapacityBytes: req.GetCapacityRange().GetRequiredBytes(),
-	}, nil
 }
 
-func getDevicePath(ctx context.Context, mnt *mount.SafeFormatAndMount, stagingTargetPath string) (string, error) {
+func getDevicePath(ctx context.Context, volumePath string) (string, error) {
-	mointPoints, err := mnt.List()
+	imgInfo, err := lookupRBDImageMetadataStash(volumePath)
 	if err != nil {
-		return "", err
+		klog.Errorf(util.Log(ctx, "failed to find image metadata: %v"), err)
 	}
-	for _, m := range mointPoints {
+	device, found := findDeviceMappingImage(ctx, imgInfo.Pool, imgInfo.ImageName, imgInfo.NbdAccess)
-		if strings.EqualFold(stagingTargetPath, m.Path) {
+	if found {
-			klog.V(3).Infof(util.Log(ctx, "found device %v for %v"), m.Device, stagingTargetPath)
+		return device, nil
-			return m.Device, nil
-		}
 	}
-	return "", fmt.Errorf("failed to get device for stagingtarget path %v", stagingTargetPath)
+	return "", fmt.Errorf("failed to get device for stagingtarget path %v", volumePath)
 }
 
 // NodeGetCapabilities returns the supported capabilities of the node server

pkg/rbd/rbd_util.go

@@ -815,14 +815,13 @@ func cleanupRBDImageMetadataStash(path string) error {
 }
 
 // resizeRBDImage resizes the given volume to new size
-func resizeRBDImage(ctx context.Context, rbdVol *rbdVolume, cr *util.Credentials) error {
+func resizeRBDImage(rbdVol *rbdVolume, newSize int64, cr *util.Credentials) error {
 	var output []byte
 
 	mon := rbdVol.Monitors
 	image := rbdVol.RbdImageName
-	volSzMiB := fmt.Sprintf("%dM", rbdVol.VolSize)
+	volSzMiB := fmt.Sprintf("%dM", newSize)
 
-	klog.V(4).Infof(util.Log(ctx, "rbd: resize %s size %s using mon %s, pool %s "), image, volSzMiB, mon, rbdVol.Pool)
 	args := []string{"resize", image, "--size", volSzMiB, "--pool", rbdVol.Pool, "--id", cr.ID, "-m", mon, "--keyfile=" + cr.KeyFile}
 	output, err := execCommand("rbd", args)