feat: support VolumeExpansion in ONLINE scenario

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

pkg/azurefile/azurefile.go

@@ -26,12 +26,15 @@ import (
 	azs "github.com/Azure/azure-sdk-for-go/storage"
 	"github.com/container-storage-interface/spec/lib/go/csi"
 	csicommon "github.com/kubernetes-sigs/azurefile-csi-driver/pkg/csi-common"
+	volumehelper "github.com/kubernetes-sigs/azurefile-csi-driver/pkg/util"
 	"github.com/pborman/uuid"
 	"k8s.io/klog"
 	"k8s.io/legacy-cloud-providers/azure"
 
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
+
+	"golang.org/x/net/context"
 )
 
 const (
@@ -111,7 +114,7 @@ func (d *Driver) Run(endpoint string) {
 	})
 
 	d.AddNodeServiceCapabilities([]csi.NodeServiceCapability_RPC_Type{
-		csi.NodeServiceCapability_RPC_UNKNOWN,
+		csi.NodeServiceCapability_RPC_EXPAND_VOLUME,
 	})
 
 	s := csicommon.NewNonBlockingGRPCServer()
@@ -283,3 +286,26 @@ func getSnapshot(id string) (string, error) {
 	}
 	return segments[3], nil
 }
+
+func (d *Driver) expandVolume(ctx context.Context, volumeID string, capacityBytes int64) (int64, error) {
+	if capacityBytes == 0 {
+		return -1, status.Error(codes.InvalidArgument, "volume capacity range missing in request")
+	}
+	requestGiB := int32(volumehelper.RoundUpGiB(capacityBytes))
+
+	shareURL, err := d.getShareUrl(volumeID)
+	if err != nil {
+		return -1, status.Errorf(codes.Internal, "failed to get share url with (%s): %v, returning with success", volumeID, err)
+	}
+
+	if _, err = shareURL.SetQuota(ctx, requestGiB); err != nil {
+		return -1, status.Errorf(codes.Internal, "expand volume error: %v", err)
+	}
+
+	resp, err := shareURL.GetProperties(ctx)
+	if err != nil {
+		return -1, status.Errorf(codes.Internal, "failed to get properties of share(%v): %v", shareURL, err)
+	}
+
+	return volumehelper.GiBToBytes(int64(resp.Quota())), nil
+}

pkg/azurefile/controllerserver.go

@@ -51,8 +51,7 @@ func (d *Driver) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest)
 	}
 
 	capacityBytes := req.GetCapacityRange().GetRequiredBytes()
-	volSizeBytes := int64(capacityBytes)
-	requestGiB := int(volumehelper.RoundUpGiB(volSizeBytes))
+	requestGiB := int(volumehelper.RoundUpGiB(capacityBytes))
 	if requestGiB == 0 {
 		requestGiB = defaultAzureFileQuota
 		klog.Warningf("no quota specified, set as default value(%d GiB)", defaultAzureFileQuota)
@@ -328,30 +327,11 @@ func (d *Driver) ControllerExpandVolume(ctx context.Context, req *csi.Controller
 		return nil, status.Errorf(codes.InvalidArgument, "invalid expand volume request: %v", req)
 	}
 
-	capacityBytes := req.GetCapacityRange().GetRequiredBytes()
-	if capacityBytes == 0 {
-		return nil, status.Error(codes.InvalidArgument, "volume capacity range missing in request")
-	}
-	volSizeBytes := int64(capacityBytes)
-	requestGiB := int32(volumehelper.RoundUpGiB(volSizeBytes))
-
-	volumeID := req.VolumeId
-	shareURL, err := d.getShareUrl(volumeID)
-	if err != nil {
-		return nil, status.Errorf(codes.Internal, "failed to get share url with (%s): %v, returning with success", volumeID, err)
-	}
-
-	if _, err = shareURL.SetQuota(ctx, requestGiB); err != nil {
-		return nil, status.Errorf(codes.Internal, "expand volume error: %v", err)
-	}
-
-	resp, err := shareURL.GetProperties(ctx)
+	currentQuota, err := d.expandVolume(ctx, req.VolumeId, req.GetCapacityRange().GetRequiredBytes())
 	if err != nil {
-		return nil, status.Errorf(codes.Internal, "failed to get properties of share(%v): %v", shareURL, err)
+		return nil, err
 	}
 
-	currentQuota := volumehelper.GiBToBytes(int64(resp.Quota()))
-
 	return &csi.ControllerExpandVolumeResponse{CapacityBytes: currentQuota}, nil
 }
 

pkg/azurefile/nodeserver.go

@@ -251,5 +251,30 @@ func (d *Driver) NodeGetVolumeStats(ctx context.Context, in *csi.NodeGetVolumeSt
 
 // NodeExpandVolume node expand volume
 func (d *Driver) NodeExpandVolume(ctx context.Context, req *csi.NodeExpandVolumeRequest) (*csi.NodeExpandVolumeResponse, error) {
-	return nil, status.Error(codes.Unimplemented, fmt.Sprintf("NodeExpandVolume is not yet implemented"))
+	volumeID := req.GetVolumeId()
+	volumePath := req.GetVolumePath()
+	if len(volumeID) == 0 {
+		return nil, status.Error(codes.InvalidArgument, "Volume ID missing in request")
+	}
+	if len(volumePath) == 0 {
+		return nil, status.Error(codes.InvalidArgument, "Volume Path missing in request")
+	}
+	if err := d.ValidateNodeServiceRequest(csi.NodeServiceCapability_RPC_EXPAND_VOLUME); err != nil {
+		return nil, status.Errorf(codes.InvalidArgument, "invalid expand volume request: %v", req)
+	}
+
+	notMnt, err := d.mounter.IsLikelyNotMountPoint(volumePath)
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, "failed to check volume path(%s): %v", volumePath, err)
+	}
+	if notMnt {
+		return nil, status.Errorf(codes.InvalidArgument, "the specified volume path(%s) is not a mount path", volumePath)
+	}
+
+	currentQuota, err := d.expandVolume(ctx, volumeID, req.GetCapacityRange().GetRequiredBytes())
+	if err != nil {
+		return nil, err
+	}
+
+	return &csi.NodeExpandVolumeResponse{CapacityBytes: currentQuota}, nil
 }

pkg/csi-common/driver.go

@@ -73,6 +73,19 @@ func (d *CSIDriver) ValidateControllerServiceRequest(c csi.ControllerServiceCapa
 	return status.Error(codes.InvalidArgument, c.String())
 }
 
+func (d *CSIDriver) ValidateNodeServiceRequest(c csi.NodeServiceCapability_RPC_Type) error {
+	if c == csi.NodeServiceCapability_RPC_UNKNOWN {
+		return nil
+	}
+
+	for _, cap := range d.NSCap {
+		if c == cap.GetRpc().GetType() {
+			return nil
+		}
+	}
+	return status.Error(codes.InvalidArgument, c.String())
+}
+
 func (d *CSIDriver) AddControllerServiceCapabilities(cl []csi.ControllerServiceCapability_RPC_Type) {
 	var csc []*csi.ControllerServiceCapability