Merge pull request kubernetes-csi#160 from mlmhl/resize

 make mock driver support volume expand feature

Gopkg.toml

@@ -26,8 +26,8 @@
 
 
 [[constraint]]
+  branch = "master"
   name = "github.com/container-storage-interface/spec"
-  version = "v1.0.0"
 
 [[constraint]]
   name = "github.com/golang/mock"

mock/main.go

@@ -33,6 +33,7 @@ func main() {
 	flag.BoolVar(&config.DisableAttach, "disable-attach", false, "Disables RPC_PUBLISH_UNPUBLISH_VOLUME capability.")
 	flag.StringVar(&config.DriverName, "name", service.Name, "CSI driver name.")
 	flag.Int64Var(&config.AttachLimit, "attach-limit", 0, "number of attachable volumes on a node")
+	flag.BoolVar(&config.NodeExpansionRequired, "node-expand-required", false, "Enables NodeServiceCapability_RPC_EXPAND_VOLUME capacity.")
 	flag.Parse()
 
 	endpoint := os.Getenv("CSI_ENDPOINT")

mock/service/controller.go

@@ -386,6 +386,13 @@ func (s *service) ControllerGetCapabilities(
 				},
 			},
 		},
+		{
+			Type: &csi.ControllerServiceCapability_Rpc{
+				Rpc: &csi.ControllerServiceCapability_RPC{
+					Type: csi.ControllerServiceCapability_RPC_EXPAND_VOLUME,
+				},
+			},
+		},
 	}
 
 	if !s.config.DisableAttach {
@@ -469,6 +476,49 @@ func (s *service) ListSnapshots(ctx context.Context,
 	return getAllSnapshots(s, req)
 }
 
+func (s *service) ControllerExpandVolume(
+	ctx context.Context,
+	req *csi.ControllerExpandVolumeRequest) (*csi.ControllerExpandVolumeResponse, error) {
+	if len(req.VolumeId) == 0 {
+		return nil, status.Error(codes.InvalidArgument, "Volume ID cannot be empty")
+	}
+
+	if req.CapacityRange == nil {
+		return nil, status.Error(codes.InvalidArgument, "Request capacity cannot be empty")
+	}
+
+	s.volsRWL.Lock()
+	defer s.volsRWL.Unlock()
+
+	i, v := s.findVolNoLock("id", req.VolumeId)
+	if i < 0 {
+		return nil, status.Error(codes.NotFound, req.VolumeId)
+	}
+
+	requestBytes := req.CapacityRange.RequiredBytes
+
+	if v.CapacityBytes > requestBytes {
+		return nil, status.Error(codes.InvalidArgument, "cannot change volume capacity to a smaller size")
+	}
+
+	resp := &csi.ControllerExpandVolumeResponse{
+		CapacityBytes:         requestBytes,
+		NodeExpansionRequired: s.config.NodeExpansionRequired,
+	}
+
+	// Check to see if the volume already satisfied request size.
+	if v.CapacityBytes == requestBytes {
+		log.WithField("volumeID", v.VolumeId).Infof("Volume capacity is already %s, no need to expand", requestBytes)
+		return resp, nil
+	}
+
+	// Update volume's capacity to the requested size.
+	v.CapacityBytes = requestBytes
+	s.vols[i] = v
+
+	return resp, nil
+}
+
 func getSnapshotById(s *service, req *csi.ListSnapshotsRequest) (*csi.ListSnapshotsResponse, error) {
 	if len(req.GetSnapshotId()) != 0 {
 		i, snapshot := s.snapshots.FindSnapshot("id", req.GetSnapshotId())

mock/service/node.go

@@ -2,6 +2,7 @@ package service
 
 import (
 	"path"
+	"strconv"
 
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
@@ -201,35 +202,94 @@ func (s *service) NodeUnpublishVolume(
 	return &csi.NodeUnpublishVolumeResponse{}, nil
 }
 
+func (s *service) NodeExpandVolume(ctx context.Context, req *csi.NodeExpandVolumeRequest) (*csi.NodeExpandVolumeResponse, error) {
+	if len(req.GetVolumeId()) == 0 {
+		return nil, status.Error(codes.InvalidArgument, "Volume ID cannot be empty")
+	}
+	if len(req.GetVolumePath()) == 0 {
+		return nil, status.Error(codes.InvalidArgument, "Volume Path cannot be empty")
+	}
+	if req.GetCapacityRange() == nil {
+		return nil, status.Error(codes.InvalidArgument, "Request capacity cannot be empty")
+	}
+
+	s.volsRWL.Lock()
+	defer s.volsRWL.Unlock()
+
+	i, v := s.findVolNoLock("id", req.VolumeId)
+	if i < 0 {
+		return nil, status.Error(codes.NotFound, req.VolumeId)
+	}
+
+	// TODO: NodeExpandVolume MUST be called after successful NodeStageVolume as we has STAGE_UNSTAGE_VOLUME node capacity.
+
+	requestCapacity := req.GetCapacityRange().RequiredBytes
+	resp := &csi.NodeExpandVolumeResponse{CapacityBytes: requestCapacity}
+
+	// fsCapacityKey is the key in the volume's attributes that is set to the file system's size.
+	fsCapacityKey := path.Join(s.nodeID, req.GetVolumePath(), "size")
+	oldCapacityStr, exist := v.VolumeContext[fsCapacityKey]
+	if exist {
+		oldCapacity, err := strconv.ParseInt(oldCapacityStr, 10, 64)
+		if err != nil {
+			return nil, status.Error(codes.Internal, err.Error())
+		}
+		if oldCapacity > requestCapacity {
+			return nil, status.Error(codes.InvalidArgument, "cannot change file system size to a smaller size")
+		}
+		if oldCapacity == requestCapacity {
+			// File system capacity is equal to requested size, no need to expand.
+			return resp, nil
+		}
+	}
+
+	// Update volume's fs capacity to requested size.
+	v.VolumeContext[fsCapacityKey] = strconv.FormatInt(requestCapacity, 10)
+	s.vols[i] = v
+
+	return resp, nil
+}
+
 func (s *service) NodeGetCapabilities(
 	ctx context.Context,
 	req *csi.NodeGetCapabilitiesRequest) (
 	*csi.NodeGetCapabilitiesResponse, error) {
 
-	return &csi.NodeGetCapabilitiesResponse{
-		Capabilities: []*csi.NodeServiceCapability{
-			{
-				Type: &csi.NodeServiceCapability_Rpc{
-					Rpc: &csi.NodeServiceCapability_RPC{
-						Type: csi.NodeServiceCapability_RPC_UNKNOWN,
-					},
+	capabilities := []*csi.NodeServiceCapability{
+		{
+			Type: &csi.NodeServiceCapability_Rpc{
+				Rpc: &csi.NodeServiceCapability_RPC{
+					Type: csi.NodeServiceCapability_RPC_UNKNOWN,
 				},
 			},
-			{
-				Type: &csi.NodeServiceCapability_Rpc{
-					Rpc: &csi.NodeServiceCapability_RPC{
-						Type: csi.NodeServiceCapability_RPC_STAGE_UNSTAGE_VOLUME,
-					},
+		},
+		{
+			Type: &csi.NodeServiceCapability_Rpc{
+				Rpc: &csi.NodeServiceCapability_RPC{
+					Type: csi.NodeServiceCapability_RPC_STAGE_UNSTAGE_VOLUME,
 				},
 			},
-			{
-				Type: &csi.NodeServiceCapability_Rpc{
-					Rpc: &csi.NodeServiceCapability_RPC{
-						Type: csi.NodeServiceCapability_RPC_GET_VOLUME_STATS,
-					},
+		},
+		{
+			Type: &csi.NodeServiceCapability_Rpc{
+				Rpc: &csi.NodeServiceCapability_RPC{
+					Type: csi.NodeServiceCapability_RPC_GET_VOLUME_STATS,
 				},
 			},
 		},
+	}
+	if s.config.NodeExpansionRequired {
+		capabilities = append(capabilities, &csi.NodeServiceCapability{
+			Type: &csi.NodeServiceCapability_Rpc{
+				Rpc: &csi.NodeServiceCapability_RPC{
+					Type: csi.NodeServiceCapability_RPC_EXPAND_VOLUME,
+				},
+			},
+		})
+	}
+
+	return &csi.NodeGetCapabilitiesResponse{
+		Capabilities: capabilities,
 	}, nil
 }
 

mock/service/service.go

@@ -27,9 +27,10 @@ var Manifest = map[string]string{
 }
 
 type Config struct {
-	DisableAttach bool
-	DriverName    string
-	AttachLimit   int64
+	DisableAttach         bool
+	DriverName            string
+	AttachLimit           int64
+	NodeExpansionRequired bool
 }
 
 // Service is the CSI Mock service provider.

pkg/sanity/controller.go

@@ -126,6 +126,7 @@ var _ = DescribeSanity("Controller Service", func(sc *SanityContext) {
 				case csi.ControllerServiceCapability_RPC_LIST_SNAPSHOTS:
 				case csi.ControllerServiceCapability_RPC_PUBLISH_READONLY:
 				case csi.ControllerServiceCapability_RPC_CLONE_VOLUME:
+				case csi.ControllerServiceCapability_RPC_EXPAND_VOLUME:
 				default:
 					Fail(fmt.Sprintf("Unknown capability: %v\n", cap.GetRpc().GetType()))
 				}