disk: implement ModifyVolume

Author: huww98

docs/ram-policies/disk/controller.json

@@ -20,8 +20,10 @@
                 "ecs:DescribeSnapshots",
                 "ecs:DescribeTags",
                 "ecs:DescribeTaskAttribute",
+                "ecs:DescribeTasks",
                 "ecs:DetachDisk",
                 "ecs:ListTagResources",
+                "ecs:ModifyDiskAttribute",
                 "ecs:ModifyDiskSpec",
                 "ecs:RemoveTags",
                 "ecs:ResizeDisk",

pkg/cloud/ecsinterface.go

@@ -11,4 +11,10 @@ type ECSInterface interface {
 	DescribeDisks(request *ecs.DescribeDisksRequest) (response *ecs.DescribeDisksResponse, err error)
 	ResizeDisk(request *ecs.ResizeDiskRequest) (response *ecs.ResizeDiskResponse, err error)
 	DescribeSnapshots(request *ecs.DescribeSnapshotsRequest) (response *ecs.DescribeSnapshotsResponse, err error)
+
+	TagResources(request *ecs.TagResourcesRequest) (response *ecs.TagResourcesResponse, err error)
+	UntagResources(request *ecs.UntagResourcesRequest) (response *ecs.UntagResourcesResponse, err error)
+	ModifyDiskSpec(request *ecs.ModifyDiskSpecRequest) (response *ecs.ModifyDiskSpecResponse, err error)
+	ModifyDiskAttribute(request *ecs.ModifyDiskAttributeRequest) (response *ecs.ModifyDiskAttributeResponse, err error)
+	DescribeTasks(request *ecs.DescribeTasksRequest) (response *ecs.DescribeTasksResponse, err error)
 }

pkg/cloud/ecsmock.go

@@ -1,9 +1,6 @@
 package disk
 
 const (
-
-	// ESSD_PERFORMANCE_LEVEL is storage class
-	ESSD_PERFORMANCE_LEVEL = "performanceLevel"
 	// DISKTAGKEY1 tag
 	DISKTAGKEY1 = "k8s.aliyun.com"
 	// DISKTAGVALUE1 value
@@ -18,6 +15,16 @@ const (
 	SNAPSHOTTAGKEY1 = "force.delete.snapshot.k8s.aliyun.com"
 )
 
+// keys used in CreateVolumeRequest.Parameters and MutableParameters
+const (
+	DISK_TYPE              = "type"
+	ESSD_PERFORMANCE_LEVEL = "performanceLevel"
+	PROVISIONED_IOPS_KEY   = "provisionedIops"
+	BURSTING_ENABLED_KEY   = "burstingEnabled"
+	DISK_TAG_PREFIX        = "diskTags/"
+	REMOVE_DISK_TAG_PREFIX = "-diskTags/"
+)
+
 // keys used in CreateSnapshotRequest.Parameters
 const (
 	SNAPSHOTTYPE               = "snapshotType"
@@ -129,9 +136,6 @@ const (
 	VolumeDeleteAutoSnapshotKey                    = "csi.alibabacloud.com/volume-delete-autosnapshot-retentiondays"
 	VOLUME_DELETE_AUTO_SNAPSHOT_OP_RETENT_DAYS_KEY = "volumeDeleteSnapshotRetentionDays"
 
-	PROVISIONED_IOPS_KEY = "provisionedIops"
-	BURSTING_ENABLED_KEY = "burstingEnabled"
-
 	CSI_DEFAULT_FS_TYPE = "csi.storage.k8s.io/fstype"
 	FS_TYPE             = "fsType"
 	EXT4_FSTYPE         = "ext4"

pkg/disk/constants.go

@@ -30,6 +30,7 @@ import (
 	"github.com/aliyun/alibaba-cloud-sdk-go/sdk/requests"
 	"github.com/aliyun/alibaba-cloud-sdk-go/services/ecs"
 	"github.com/container-storage-interface/spec/lib/go/csi"
+	"github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/cloud/wrap"
 	"github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/common"
 	"github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/disk/desc"
 	"github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/disk/waitstatus"
@@ -52,6 +53,7 @@ type controllerServer struct {
 	recorder       record.EventRecorder
 	ad             DiskAttachDetach
 	snapshotWaiter waitstatus.StatusWaiter[ecs.Snapshot]
+	modify         ModifyServer
 	common.GenericControllerServer
 }
 
@@ -85,6 +87,16 @@ var veasp = struct {
 
 var delVolumeSnap sync.Map
 
+func newTaskStatusWaiter() waitstatus.StatusWaiter[ecs.Task] {
+	client := desc.Task{Client: GlobalConfigVar.EcsClient}
+	waiter := waitstatus.NewBatched(client, clock.RealClock{}, 3*time.Second, 10*time.Second)
+	waiter.PollHook = func() desc.Client[ecs.Task] {
+		return desc.Task{Client: updateEcsClient(GlobalConfigVar.EcsClient)}
+	}
+	go waiter.Run(context.Background())
+	return waiter
+}
+
 func newSnapshotStatusWaiter() waitstatus.StatusWaiter[ecs.Snapshot] {
 	client := desc.Snapshots{
 		Client: GlobalConfigVar.EcsClient,
@@ -112,6 +124,10 @@ func NewControllerServer(csiCfg utils.Config) csi.ControllerServer {
 			detachThrottler: defaultThrottler(),
 		},
 		snapshotWaiter: newSnapshotStatusWaiter(),
+		modify: ModifyServer{
+			ecsClient:  GlobalConfigVar.EcsClient,
+			taskWaiter: newTaskStatusWaiter(),
+		},
 	}
 	detachConcurrency := 1
 	attachConcurrency := 1
@@ -135,6 +151,7 @@ func (cs *controllerServer) ControllerGetCapabilities(ctx context.Context, req *
 			csi.ControllerServiceCapability_RPC_CREATE_DELETE_SNAPSHOT,
 			csi.ControllerServiceCapability_RPC_LIST_SNAPSHOTS,
 			csi.ControllerServiceCapability_RPC_EXPAND_VOLUME,
+			csi.ControllerServiceCapability_RPC_MODIFY_VOLUME,
 		),
 	}, nil
 }
@@ -169,6 +186,14 @@ func (cs *controllerServer) CreateVolume(ctx context.Context, req *csi.CreateVol
 		return nil, status.Errorf(codes.InvalidArgument, "Invalid parameters from input: %v, with error: %v", req.Name, err)
 	}
 
+	if len(req.MutableParameters) > 0 {
+		mutable, err := parseMutableParameters(req.MutableParameters)
+		if err != nil {
+			return nil, status.Errorf(codes.InvalidArgument, "Invalid mutable parameters: %v", err)
+		}
+		importMutableParameters(diskVol, &mutable)
+	}
+
 	sharedDisk := len(diskVol.Type) == 1 && (diskVol.Type[0] == DiskSharedEfficiency || diskVol.Type[0] == DiskSharedSSD)
 
 	var supportedTypes sets.Set[Category]
@@ -805,3 +830,18 @@ func (cs *controllerServer) deleteVolumeExpandAutoSnapshot(ctx context.Context,
 	cs.recorder.Event(pvc, v1.EventTypeNormal, snapshotDeletedSuccessfully, str)
 	return nil
 }
+
+func (cs *controllerServer) ControllerModifyVolume(ctx context.Context, req *csi.ControllerModifyVolumeRequest) (*csi.ControllerModifyVolumeResponse, error) {
+	params, err := parseMutableParameters(req.MutableParameters)
+	if err != nil {
+		return nil, status.Error(codes.InvalidArgument, err.Error())
+	}
+	err = cs.modify.Modify(ctx, req.VolumeId, params)
+	if err != nil {
+		if errors.Is(err, wrap.ErrorCode("InvalidDiskId.NotFound")) {
+			return nil, status.Error(codes.NotFound, err.Error())
+		}
+		return nil, err
+	}
+	return &csi.ControllerModifyVolumeResponse{}, nil
+}

pkg/disk/controllerserver.go

@@ -0,0 +1,43 @@
+package desc
+
+import (
+	"strings"
+
+	"github.com/aliyun/alibaba-cloud-sdk-go/sdk/requests"
+	"github.com/aliyun/alibaba-cloud-sdk-go/services/ecs"
+)
+
+type Task struct {
+	*ecs.Client
+}
+
+func (c Task) Describe(ids []string) (Response[ecs.Task], error) {
+	req := ecs.CreateDescribeTasksRequest()
+	req.TaskIds = strings.Join(ids, ",")
+	req.PageSize = requests.NewInteger(batchSize)
+
+	ret := Response[ecs.Task]{}
+	resp, err := c.Client.DescribeTasks(req)
+	if err != nil {
+		return ret, err
+	}
+	ret.RequestID = resp.RequestId
+	ret.Resources = resp.TaskSet.Task
+	return ret, nil
+}
+
+func (c Task) GetID(resource *ecs.Task) string {
+	return resource.TaskId
+}
+
+func (c Task) Type() string {
+	return "task"
+}
+
+func (c Task) BatchSize() int {
+	return batchSize
+}
+
+func TaskSattled(t *ecs.Task) bool {
+	return t.TaskStatus == "Finished" || t.TaskStatus == "Failed"
+}