disk: implement ModifyVolume

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/disk/constants.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"
@@ -128,9 +135,6 @@ const (
 	VOLUME_EXPAND_AUTO_SNAPSHOT_OP_KEY             = "volumeExpandAutoSnapshot"
 	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/controllerserver.go

@@ -31,7 +31,10 @@ import (
 	"github.com/aliyun/alibaba-cloud-sdk-go/services/ecs"
 	"github.com/container-storage-interface/spec/lib/go/csi"
 	"github.com/golang/protobuf/ptypes/timestamp"
+	"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"
 	"github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/features"
 	"github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/utils"
 	"google.golang.org/grpc/codes"
@@ -43,11 +46,13 @@ import (
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/client-go/tools/record"
 	"k8s.io/klog/v2"
+	"k8s.io/utils/clock"
 )
 
 // controller server try to create/delete volumes/snapshots
 type controllerServer struct {
 	recorder record.EventRecorder
+	modify   ModifyServer
 	common.GenericControllerServer
 }
 
@@ -81,10 +86,37 @@ var veasp = struct {
 
 var delVolumeSnap sync.Map
 
+func newTaskStatusWaiter() waitstatus.StatusWaiter[ecs.Task] {
+	client := desc.Task{Client: GlobalConfigVar.EcsClient}
+	if GlobalConfigVar.DiskMultiTenantEnable {
+		waiter := waitstatus.NewSimple(client, clock.RealClock{})
+		waiter.ClientFactory = func(ctx context.Context) (desc.Client[ecs.Task], error) {
+			tenantUserUID := GetTenantUserUID(ctx)
+			client, err := getEcsClientByID("", tenantUserUID)
+			if err != nil {
+				return nil, err
+			}
+			return desc.Task{Client: client}, nil
+		}
+		return waiter
+	} else {
+		waiter := waitstatus.NewBatched(client, clock.RealClock{})
+		waiter.PollHook = func() desc.Client[ecs.Task] {
+			return desc.Task{Client: updateEcsClient(GlobalConfigVar.EcsClient)}
+		}
+		go waiter.Run(context.Background())
+		return waiter
+	}
+}
+
 // NewControllerServer is to create controller server
 func NewControllerServer() csi.ControllerServer {
 	c := &controllerServer{
 		recorder: utils.NewEventRecorder(),
+		modify: ModifyServer{
+			ecsClient:  GlobalConfigVar.EcsClient,
+			taskWaiter: newTaskStatusWaiter(),
+		},
 	}
 	return c
 }
@@ -97,6 +129,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
 }
@@ -137,6 +170,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)
+	}
+
 	// 兼容 serverless 拓扑感知场景；
 	// req参数里面包含了云盘ID，则直接使用云盘ID进行返回；
 	csiVolume, err := staticVolumeCreate(req, snapshotID)
@@ -1003,12 +1044,17 @@ func (cs *controllerServer) deleteUntagAutoSnapshot(snapshotID, diskID string) {
 	}
 }
 
-func (cs *controllerServer) ControllerGetVolume(ctx context.Context, req *csi.ControllerGetVolumeRequest,
-) (*csi.ControllerGetVolumeResponse, error) {
-	return nil, status.Error(codes.Unimplemented, "")
-}
-
-func (cs *controllerServer) ControllerModifyVolume(ctx context.Context, req *csi.ControllerModifyVolumeRequest,
-) (*csi.ControllerModifyVolumeResponse, error) {
-	return nil, status.Error(codes.Unimplemented, "")
+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/desc/task.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"
+}