Add volume expansion support

Author: bertinatto

examples/csi-storageclass.yaml

@@ -5,3 +5,4 @@ metadata:
 provisioner: hostpath.csi.k8s.io
 reclaimPolicy: Delete
 volumeBindingMode: Immediate
+allowVolumeExpansion: true

pkg/hostpath/controllerserver.go

@@ -63,6 +63,7 @@ func NewControllerServer(ephemeral bool) *controllerServer {
 				csi.ControllerServiceCapability_RPC_CREATE_DELETE_SNAPSHOT,
 				csi.ControllerServiceCapability_RPC_LIST_SNAPSHOTS,
 				csi.ControllerServiceCapability_RPC_CLONE_VOLUME,
+				csi.ControllerServiceCapability_RPC_EXPAND_VOLUME,
 			}),
 	}
 }
@@ -457,6 +458,42 @@ func (cs *controllerServer) ListSnapshots(ctx context.Context, req *csi.ListSnap
 	}, nil
 }
 
+func (cs *controllerServer) ControllerExpandVolume(ctx context.Context, req *csi.ControllerExpandVolumeRequest) (*csi.ControllerExpandVolumeResponse, error) {
+
+	volID := req.GetVolumeId()
+	if len(volID) == 0 {
+		return nil, status.Error(codes.InvalidArgument, "Volume ID missing in request")
+	}
+
+	capRange := req.GetCapacityRange()
+	if capRange == nil {
+		return nil, status.Error(codes.InvalidArgument, "Capacity range not provided")
+	}
+
+	capacity := int64(capRange.GetRequiredBytes())
+	if capacity >= maxStorageCapacity {
+		return nil, status.Errorf(codes.OutOfRange, "Requested capacity %d exceeds maximum allowed %d", capacity, maxStorageCapacity)
+	}
+
+	exVol, err := getVolumeByID(volID)
+	if err != nil {
+		// Assume not found error
+		return nil, status.Errorf(codes.NotFound, "Could not get volume %s: %v", volID, err)
+	}
+
+	if exVol.VolSize < capacity {
+		exVol.VolSize = capacity
+		if err := updateHostpathVolume(volID, exVol); err != nil {
+			return nil, status.Errorf(codes.Internal, "Could not update volume %s: %v", volID, err)
+		}
+	}
+
+	return &csi.ControllerExpandVolumeResponse{
+		CapacityBytes:         exVol.VolSize,
+		NodeExpansionRequired: true,
+	}, nil
+}
+
 func convertSnapshot(snap hostPathSnapshot) *csi.ListSnapshotsResponse {
 	entries := []*csi.ListSnapshotsResponse_Entry{
 		{

pkg/hostpath/hostpath.go

@@ -209,6 +209,18 @@ func createHostpathVolume(volID, name string, cap int64, volAccessType accessTyp
 	return &hostpathVol, nil
 }
 
+// updateVolume updates the existing hostpath volume.
+func updateHostpathVolume(volID string, volume hostPathVolume) error {
+	glog.V(4).Infof("updating hostpath volume: %s", volID)
+
+	if _, err := getVolumeByID(volID); err != nil {
+		return err
+	}
+
+	hostPathVolumes[volID] = volume
+	return nil
+}
+
 // deleteVolume deletes the directory for the hostpath volume.
 func deleteHostpathVolume(volID string) error {
 	glog.V(4).Infof("deleting hostpath volume: %s", volID)

pkg/hostpath/nodeserver.go

@@ -277,10 +277,57 @@ func (ns *nodeServer) NodeGetCapabilities(ctx context.Context, req *csi.NodeGetC
 					},
 				},
 			},
+			{
+				Type: &csi.NodeServiceCapability_Rpc{
+					Rpc: &csi.NodeServiceCapability_RPC{
+						Type: csi.NodeServiceCapability_RPC_EXPAND_VOLUME,
+					},
+				},
+			},
 		},
 	}, nil
 }
 
 func (ns *nodeServer) NodeGetVolumeStats(ctx context.Context, in *csi.NodeGetVolumeStatsRequest) (*csi.NodeGetVolumeStatsResponse, error) {
 	return nil, status.Error(codes.Unimplemented, "")
 }
+
+// NodeExpandVolume is only implemented so the driver can be used for e2e testing.
+func (ns *nodeServer) NodeExpandVolume(ctx context.Context, req *csi.NodeExpandVolumeRequest) (*csi.NodeExpandVolumeResponse, error) {
+
+	volID := req.GetVolumeId()
+	if len(volID) == 0 {
+		return nil, status.Error(codes.InvalidArgument, "Volume ID not provided")
+	}
+
+	vol, err := getVolumeByID(volID)
+	if err != nil {
+		// Assume not found error
+		return nil, status.Errorf(codes.NotFound, "Could not get volume %s: %v", volID, err)
+	}
+
+	volPath := req.GetVolumePath()
+	if len(volPath) == 0 {
+		return nil, status.Error(codes.InvalidArgument, "Volume path not provided")
+	}
+
+	info, err := os.Lstat(volPath)
+	if err != nil {
+		return nil, status.Errorf(codes.InvalidArgument, "Could not get file information from %s: %v", volPath, err)
+	}
+
+	switch m := info.Mode(); {
+	case m.IsDir():
+		if vol.VolAccessType != mountAccess {
+			return nil, status.Errorf(codes.InvalidArgument, "Volume %s is not a directory", volID)
+		}
+	case m&os.ModeCharDevice != 0:
+		if vol.VolAccessType != blockAccess {
+			return nil, status.Errorf(codes.InvalidArgument, "Volume %s is not a block device", volID)
+		}
+	default:
+		return nil, status.Errorf(codes.InvalidArgument, "Volume %s is invalid", volID)
+	}
+
+	return &csi.NodeExpandVolumeResponse{}, nil
+}