OKE (CSI) Expand Volume support.

Author: mrunalpagnis

cmd/oci-csi-controller-driver/csioptions/csioptions.go

@@ -44,6 +44,7 @@ type CSIOptions struct {
 	MetricsPath             string
 	ExtraCreateMetadata     bool
 	ReconcileSync           time.Duration
+	EnableResizer           bool
 }
 
 //NewCSIOptions initializes the flag
@@ -71,6 +72,7 @@ func NewCSIOptions() *CSIOptions {
 		MetricsPath:             *flag.String("metrics-path", "/metrics", "The HTTP path where prometheus metrics will be exposed. Default is `/metrics`."),
 		ExtraCreateMetadata:     *flag.Bool("extra-create-metadata", false, "If set, add pv/pvc metadata to plugin create requests as parameters."),
 		ReconcileSync:           *flag.Duration("reconcile-sync", 1*time.Minute, "Resync interval of the VolumeAttachment reconciler."),
+		EnableResizer:           *flag.Bool("csi-bv-expansion-enabled", false, "Enables go routine csi-resizer."),
 	}
 	return &csioptions
 }

cmd/oci-csi-node-driver/main.go

@@ -16,12 +16,16 @@ package main
 
 import (
 	"flag"
+
+	"github.com/spf13/viper"
+	"go.uber.org/zap/zapcore"
+	"k8s.io/klog"
+
 	"github.com/oracle/oci-cloud-controller-manager/cmd/oci-csi-node-driver/nodedriver"
 	"github.com/oracle/oci-cloud-controller-manager/cmd/oci-csi-node-driver/nodedriveroptions"
 	"github.com/oracle/oci-cloud-controller-manager/cmd/oci-csi-node-driver/nodedriverregistrar"
 	"github.com/oracle/oci-cloud-controller-manager/pkg/csi/driver"
 	"github.com/oracle/oci-cloud-controller-manager/pkg/util/signals"
-	"k8s.io/klog"
 )
 
 func main() {
@@ -44,6 +48,8 @@ func main() {
 	flag.Set("logtostderr", "true")
 	flag.Parse()
 
+	viper.Set("log-level", getLevel(nodecsioptions.LogLevel))
+
 	blockvolumeNodeOptions := nodedriveroptions.NodeOptions{
 		Name:                   "BV",
 		Endpoint:               nodecsioptions.Endpoint,
@@ -72,3 +78,24 @@ func main() {
 	}
 	<-stopCh
 }
+
+func getLevel(loglevel string) int8 {
+	switch loglevel {
+	case "debug":
+		return int8(zapcore.DebugLevel)
+	case "info":
+		return int8(zapcore.InfoLevel)
+	case "warn":
+		return int8(zapcore.WarnLevel)
+	case "error":
+		return int8(zapcore.ErrorLevel)
+	case "dpanic":
+		return int8(zapcore.DPanicLevel)
+	case "panic":
+		return int8(zapcore.PanicLevel)
+	case "fatal":
+		return int8(zapcore.FatalLevel)
+	default:
+		return int8(zapcore.InfoLevel)
+	}
+}

pkg/cloudprovider/providers/oci/instances_test.go

@@ -20,7 +20,10 @@ import (
 	"reflect"
 	"testing"
 
+	"go.uber.org/zap"
+	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/types"
 
 	providercfg "github.com/oracle/oci-cloud-controller-manager/pkg/cloudprovider/providers/oci/config"
@@ -30,9 +33,6 @@ import (
 	"github.com/oracle/oci-go-sdk/v31/filestorage"
 	"github.com/oracle/oci-go-sdk/v31/identity"
 	"github.com/oracle/oci-go-sdk/v31/loadbalancer"
-	"go.uber.org/zap"
-	v1 "k8s.io/api/core/v1"
-	"k8s.io/apimachinery/pkg/labels"
 )
 
 var (
@@ -487,6 +487,10 @@ func (MockBlockStorageClient) CreateVolume(ctx context.Context, details core.Cre
 	return nil, nil
 }
 
+func (c MockBlockStorageClient) UpdateVolume(ctx context.Context, volumeId string, details core.UpdateVolumeDetails) (*core.Volume, error) {
+	return nil, nil
+}
+
 func (MockBlockStorageClient) GetVolume(ctx context.Context, id string) (*core.Volume, error) {
 	return nil, nil
 }

pkg/csi-util/utils.go

@@ -10,7 +10,7 @@ import (
 	"time"
 
 	"go.uber.org/zap"
-
+	"github.com/container-storage-interface/spec/lib/go/csi"
 	kubeAPI "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/sets"
@@ -24,6 +24,18 @@ import (
 )
 
 const (
+	// minimumVolumeSizeInBytes is used to validate that the user is not trying
+	// to create a volume that is smaller than what we support
+	MinimumVolumeSizeInBytes int64 = 50 * client.GiB
+
+	// maximumVolumeSizeInBytes is used to validate that the user is not trying
+	// to create a volume that is larger than what we support
+	MaximumVolumeSizeInBytes int64 = 32 * client.TiB
+
+	// defaultVolumeSizeInBytes is used when the user did not provide a size or
+	// the size they provided did not satisfy our requirements
+	defaultVolumeSizeInBytes int64 = MinimumVolumeSizeInBytes
+
 	waitForPathDelay = 1 * time.Second
 
 	// ociVolumeBackupID is the name of the oci volume backup id annotation.
@@ -254,3 +266,64 @@ func (vl *VolumeLocks) Release(volumeID string) {
 	defer vl.mux.Unlock()
 	vl.locks.Delete(volumeID)
 }
+
+// extractStorage extracts the storage size in bytes from the given capacity
+// range. If the capacity range is not satisfied it returns the default volume
+// size. If the capacity range is below or above supported sizes, it returns an
+// error.
+func ExtractStorage(capRange *csi.CapacityRange) (int64, error) {
+	if capRange == nil {
+		return defaultVolumeSizeInBytes, nil
+	}
+
+	requiredBytes := capRange.GetRequiredBytes()
+	requiredSet := 0 < requiredBytes
+	limitBytes := capRange.GetLimitBytes()
+	limitSet := 0 < limitBytes
+
+	if !requiredSet && !limitSet {
+		return defaultVolumeSizeInBytes, nil
+	}
+
+	if requiredSet && limitSet && limitBytes < requiredBytes {
+		return 0, fmt.Errorf("limit (%v) can not be less than required (%v) size", FormatBytes(limitBytes), FormatBytes(requiredBytes))
+	}
+
+	if requiredSet && !limitSet {
+		return MaxOfInt(requiredBytes, MinimumVolumeSizeInBytes), nil
+	}
+
+	if limitSet {
+		return MaxOfInt(limitBytes, MinimumVolumeSizeInBytes), nil
+	}
+
+	if requiredSet && requiredBytes > MaximumVolumeSizeInBytes {
+		return 0, fmt.Errorf("required (%v) can not exceed maximum supported volume size (%v)", FormatBytes(requiredBytes), FormatBytes(MaximumVolumeSizeInBytes))
+	}
+
+	if !requiredSet && limitSet && limitBytes > MaximumVolumeSizeInBytes {
+		return 0, fmt.Errorf("limit (%v) can not exceed maximum supported volume size (%v)", FormatBytes(limitBytes), FormatBytes(MaximumVolumeSizeInBytes))
+	}
+
+	if requiredSet && limitSet {
+		return MaxOfInt(requiredBytes, limitBytes), nil
+	}
+
+	if requiredSet {
+		return requiredBytes, nil
+	}
+
+	if limitSet {
+		return limitBytes, nil
+	}
+
+	return defaultVolumeSizeInBytes, nil
+}
+
+func RoundUpSize(volumeSizeBytes int64, allocationUnitBytes int64) int64 {
+	return (volumeSizeBytes + allocationUnitBytes - 1) / allocationUnitBytes
+}
+
+func RoundUpMinSize() int64 {
+	return RoundUpSize(MinimumVolumeSizeInBytes, 1*client.GiB)
+}

pkg/csi/driver/bv_node.go

@@ -14,6 +14,7 @@ import (
 	kubeAPI "k8s.io/api/core/v1"
 
 	"github.com/oracle/oci-cloud-controller-manager/pkg/csi-util"
+	"github.com/oracle/oci-cloud-controller-manager/pkg/oci/client"
 	"github.com/oracle/oci-cloud-controller-manager/pkg/util/disk"
 )
 
@@ -416,18 +417,21 @@ func getDevicePathAndAttachmentType(logger *zap.SugaredLogger, path []string) (s
 
 // NodeGetCapabilities returns the supported capabilities of the node server
 func (d BlockVolumeNodeDriver) NodeGetCapabilities(ctx context.Context, req *csi.NodeGetCapabilitiesRequest) (*csi.NodeGetCapabilitiesResponse, error) {
-	nscap := &csi.NodeServiceCapability{
-		Type: &csi.NodeServiceCapability_Rpc{
-			Rpc: &csi.NodeServiceCapability_RPC{
-				Type: csi.NodeServiceCapability_RPC_STAGE_UNSTAGE_VOLUME,
+	var nscaps []*csi.NodeServiceCapability
+	nodeCaps := []csi.NodeServiceCapability_RPC_Type{csi.NodeServiceCapability_RPC_STAGE_UNSTAGE_VOLUME, csi.NodeServiceCapability_RPC_EXPAND_VOLUME}
+	for _, nodeCap := range nodeCaps {
+		c := &csi.NodeServiceCapability{
+			Type: &csi.NodeServiceCapability_Rpc{
+				Rpc: &csi.NodeServiceCapability_RPC{
+					Type: nodeCap,
+				},
 			},
-		},
+		}
+		nscaps = append(nscaps, c)
 	}
 
 	return &csi.NodeGetCapabilitiesResponse{
-		Capabilities: []*csi.NodeServiceCapability{
-			nscap,
-		},
+		Capabilities: nscaps,
 	}, nil
 }
 
@@ -461,5 +465,89 @@ func (d BlockVolumeNodeDriver) NodeGetVolumeStats(ctx context.Context, req *csi.
 
 //NodeExpandVolume returns the expand of the volume
 func (d BlockVolumeNodeDriver) NodeExpandVolume(ctx context.Context, req *csi.NodeExpandVolumeRequest) (*csi.NodeExpandVolumeResponse, error) {
-	return nil, status.Error(codes.Unimplemented, "NodeExpandVolume is not supported yet")
+	volumeID := req.GetVolumeId()
+	if len(volumeID) == 0 {
+		return nil, status.Error(codes.InvalidArgument, "Volume ID not provided")
+	}
+	volumePath := req.GetVolumePath()
+	if len(volumePath) == 0 {
+		return nil, status.Error(codes.InvalidArgument, "volume path must be provided")
+	}
+
+	logger := d.logger.With("volumeId", req.VolumeId, "volumePath", req.VolumePath)
+
+	if acquired := d.volumeLocks.TryAcquire(req.VolumeId); !acquired {
+		logger.Error("Could not acquire lock for NodeUnpublishVolume.")
+		return nil, status.Errorf(codes.Aborted, volumeOperationAlreadyExistsFmt, req.VolumeId)
+	}
+
+	defer d.volumeLocks.Release(req.VolumeId)
+
+	requestedSize, err:= csi_util.ExtractStorage(req.CapacityRange)
+	requestedSizeGB := csi_util.RoundUpSize(requestedSize, 1*client.GiB)
+
+	if err != nil {
+		logger.With(zap.Error(err)).Error("invalid capacity range")
+		return nil, status.Errorf(codes.OutOfRange, "invalid capacity range: %v", err)
+	}
+
+	diskPath, err := disk.GetDiskPathFromMountPath(d.logger, volumePath)
+	if err != nil {
+		// do a clean exit in case of mount point not found
+		if err == disk.ErrMountPointNotFound {
+			logger.With(zap.Error(err)).With("volumePath", volumePath).Warn("unable to fetch mount point")
+			return &csi.NodeExpandVolumeResponse{}, nil
+		}
+		logger.With(zap.Error(err)).With("volumePath", volumePath).Error("unable to get diskPath from mount path")
+		return nil, status.Error(codes.Internal, err.Error())
+	}
+
+	attachmentType, devicePath, err := getDevicePathAndAttachmentType(d.logger, diskPath)
+	if err != nil {
+		logger.With(zap.Error(err)).With("diskPath", diskPath).Error("unable to determine the attachment type")
+		return nil, status.Error(codes.Internal, err.Error())
+	}
+	logger.With("diskPath", diskPath, "attachmentType", attachmentType, "devicePath", devicePath).Infof("Extracted attachment type and device path")
+
+	var mountHandler disk.Interface
+	switch attachmentType {
+	case attachmentTypeISCSI:
+		scsiInfo, _ := csi_util.ExtractISCSIInformationFromMountPath(d.logger, diskPath)
+		if scsiInfo == nil {
+			logger.Warn("unable to get the ISCSI info")
+			return &csi.NodeExpandVolumeResponse{}, nil
+		}
+		mountHandler = disk.NewFromISCSIDisk(d.logger, scsiInfo)
+		d.logger.With("ISCSIInfo", scsiInfo, "mountPath", volumePath).Info("Found ISCSIInfo for NodeExpandVolume.")
+	case attachmentTypeParavirtualized:
+		mountHandler = disk.NewFromPVDisk(d.logger)
+		logger.Info("starting to expand paravirtualized Mounting.")
+	default:
+		logger.Error("unknown attachment type. supported attachment types are iscsi and paravirtualized")
+		return nil, status.Error(codes.InvalidArgument, "unknown attachment type. supported attachment types are iscsi and paravirtualized")
+	}
+
+	if err := mountHandler.Rescan(devicePath); err != nil {
+		return nil, status.Errorf(codes.Internal, "Failed to rescan volume %q (%q):  %v", volumeID, devicePath, err)
+	}
+	logger.With("devicePath", devicePath).Debug("Rescan completed")
+
+	if _, err := mountHandler.Resize(devicePath, volumePath); err != nil {
+		return nil, status.Errorf(codes.Internal, "Failed to resize volume %q (%q):  %v", volumeID, devicePath, err)
+	}
+
+	allocatedSizeBytes, err := mountHandler.GetBlockSizeBytes(devicePath)
+	if err != nil {
+		return nil, status.Error(codes.Internal, fmt.Sprintf("Failed to get size of block volume at path %s: %v", devicePath, err))
+	}
+
+	allocatedSizeGB := csi_util.RoundUpSize(allocatedSizeBytes, 1*client.GiB)
+
+	if allocatedSizeGB < requestedSizeGB {
+		return nil, status.Error(codes.Internal, fmt.Sprintf("Expand Volume Failed, requested size in GB %d but resize allocated only %d", requestedSizeGB, allocatedSizeGB))
+	}
+
+	return &csi.NodeExpandVolumeResponse{
+		CapacityBytes: allocatedSizeBytes,
+	}, nil
 }