disk: optimize batch create/delete

Integrating batcher and throttle.

Author: huww98

pkg/cloud/throttle/throttle.go

@@ -131,3 +131,14 @@ func (t *Throttler) Throttle(ctx context.Context, f func() error) error {
 		}
 	}
 }
+
+func Throttled[TReq any, TResp any](t *Throttler, f func(TReq) (TResp, error)) func(context.Context, TReq) (TResp, error) {
+	return func(ctx context.Context, req TReq) (TResp, error) {
+		var resp TResp
+		err := t.Throttle(ctx, func() (err error) {
+			resp, err = f(req)
+			return err
+		})
+		return resp, err
+	}
+}

pkg/disk/cloud.go

@@ -77,6 +77,13 @@ type DiskAttachDetach struct {
 	dev *DeviceManager
 }
 
+type DiskCreateDelete struct {
+	ecs             cloud.ECSInterface
+	batcher         batcher.Batcher[ecs.Disk]
+	createThrottler *throttle.Throttler
+	deleteThrottler *throttle.Throttler
+}
+
 func (ad *DiskAttachDetach) possibleDisks(before sets.Set[string]) ([]string, error) {
 	after, err := ad.dev.ListBlocks()
 	if err != nil {
@@ -290,11 +297,7 @@ func (ad *DiskAttachDetach) attachDisk(ctx context.Context, diskID, nodeID strin
 	for key, value := range GlobalConfigVar.RequestBaseInfo {
 		attachRequest.AppendUserAgent(key, value)
 	}
-	var response *ecs.AttachDiskResponse
-	err = ad.attachThrottler.Throttle(ctx, func() error {
-		response, err = ecsClient.AttachDisk(attachRequest)
-		return err
-	})
+	response, err := throttle.Throttled(ad.attachThrottler, ecsClient.AttachDisk)(ctx, attachRequest)
 	if err != nil {
 		var aliErr *alicloudErr.ServerError
 		if errors.As(err, &aliErr) {
@@ -465,11 +468,7 @@ func (ad *DiskAttachDetach) detachDisk(ctx context.Context, ecsClient cloud.ECSI
 	for key, value := range GlobalConfigVar.RequestBaseInfo {
 		detachDiskRequest.AppendUserAgent(key, value)
 	}
-	var response *ecs.DetachDiskResponse
-	err = ad.detachThrottler.Throttle(ctx, func() error {
-		response, err = ecsClient.DetachDisk(detachDiskRequest)
-		return err
-	})
+	response, err := throttle.Throttled(ad.detachThrottler, ecsClient.DetachDisk)(ctx, detachDiskRequest)
 	if err != nil {
 		return status.Errorf(codes.Aborted, "DetachDisk: Fail to detach %s: from Instance: %s with error: %v", disk.DiskId, disk.InstanceId, err)
 	}
@@ -927,7 +926,7 @@ func isValidSnapshotName(name string) bool {
 	return validDiskNameRegexp.MatchString(name)
 }
 
-func createDisk(ecsClient cloud.ECSInterface, diskName, snapshotID string, diskVol *diskVolumeArgs, supportedTypes sets.Set[Category], selectedInstance string, isVirtualNode bool) (string, createAttempt, error) {
+func (c *DiskCreateDelete) createDisk(ctx context.Context, diskName, snapshotID string, diskVol *diskVolumeArgs, supportedTypes sets.Set[Category], selectedInstance string, isVirtualNode bool) (string, createAttempt, error) {
 	// 需要配置external-provisioner启动参数--extra-create-metadata=true，然后ACK的external-provisioner才会将PVC的Annotations传过来
 	createDiskRequest := buildCreateDiskRequest(diskVol)
 	if isValidDiskName(diskName) {
@@ -962,7 +961,7 @@ func createDisk(ecsClient cloud.ECSInterface, diskName, snapshotID string, diskV
 			attempt.Instance = selectedInstance
 		}
 	retry:
-		diskID, final, err := createDiskAttempt(createDiskRequest, attempt, ecsClient)
+		diskID, final, err := c.createDiskAttempt(ctx, createDiskRequest, attempt)
 		if err != nil {
 			if final {
 				return "", attempt, err
@@ -972,7 +971,7 @@ func createDisk(ecsClient cloud.ECSInterface, diskName, snapshotID string, diskV
 					// protect us from infinite loop
 					return "", attempt, status.Error(codes.Internal, "unexpected parameter mismatch")
 				}
-				existingDisk, err := findDiskByName(diskName, ecsClient)
+				existingDisk, err := findDiskByName(diskName, c.ecs)
 				if err != nil {
 					return "", attempt, status.Errorf(codes.Internal, "parameter mismatch detected, but fetch existing disk failed: %v", err)
 				}
@@ -1051,11 +1050,11 @@ func finalizeCreateDiskRequest(template *ecs.CreateDiskRequest, attempt createAt
 
 var ErrParameterMismatch = errors.New("parameter mismatch")
 
-func createDiskAttempt(req *ecs.CreateDiskRequest, attempt createAttempt, ecsClient cloud.ECSInterface) (diskId string, final bool, err error) {
+func (c *DiskCreateDelete) createDiskAttempt(ctx context.Context, req *ecs.CreateDiskRequest, attempt createAttempt) (diskId string, final bool, err error) {
 	req = finalizeCreateDiskRequest(req, attempt)
 	klog.Infof("request: request content: %++v", req)
 
-	volumeRes, err := ecsClient.CreateDisk(req)
+	volumeRes, err := throttle.Throttled(c.createThrottler, c.ecs.CreateDisk)(ctx, req)
 	if err == nil {
 		klog.Infof("request: diskId: %s, reqId: %s", volumeRes.DiskId, volumeRes.RequestId)
 		return volumeRes.DiskId, true, nil
@@ -1157,14 +1156,14 @@ func getDefaultDiskTags(diskVol *diskVolumeArgs) []ecs.CreateDiskTag {
 	return diskTags
 }
 
-func deleteDisk(ctx context.Context, ecsClient cloud.ECSInterface, diskId string) (*ecs.DeleteDiskResponse, error) {
+func (c *DiskCreateDelete) deleteDisk(ctx context.Context, ecsClient cloud.ECSInterface, diskId string) (*ecs.DeleteDiskResponse, error) {
 	deleteDiskRequest := ecs.CreateDeleteDiskRequest()
 	deleteDiskRequest.DiskId = diskId
 
 	var resp *ecs.DeleteDiskResponse
 	err := wait.PollUntilContextTimeout(ctx, time.Second*5, DISK_DELETE_INIT_TIMEOUT, true, func(ctx context.Context) (bool, error) {
 		var err error
-		resp, err = ecsClient.DeleteDisk(deleteDiskRequest)
+		resp, err = throttle.Throttled(c.deleteThrottler, c.ecs.DeleteDisk)(ctx, deleteDiskRequest)
 		if err == nil {
 			klog.Infof("DeleteVolume: Successfully deleted volume: %s, with RequestId: %s", diskId, resp.RequestId)
 			return true, nil

pkg/disk/cloud_test.go

@@ -4,14 +4,19 @@ import (
 	"context"
 	"fmt"
 	"testing"
+	"time"
 
 	alicloudErr "github.com/aliyun/alibaba-cloud-sdk-go/sdk/errors"
 	"github.com/aliyun/alibaba-cloud-sdk-go/sdk/requests"
 	"github.com/aliyun/alibaba-cloud-sdk-go/services/ecs"
 	gomock "github.com/golang/mock/gomock"
 	"github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/cloud"
+	"github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/disk/batcher"
+	"github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/disk/desc"
 	"github.com/stretchr/testify/assert"
 	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/klog/v2/ktesting"
+	"k8s.io/utils/clock"
 )
 
 var (
@@ -31,37 +36,49 @@ func init() {
 }`), resizeDiskResponse)
 }
 
-func TestDeleteDisk(t *testing.T) {
+func testCreateDelete(t *testing.T) (*cloud.MockECSInterface, *DiskCreateDelete) {
 	ctrl := gomock.NewController(t)
 	c := cloud.NewMockECSInterface(ctrl)
 
+	return c, &DiskCreateDelete{
+		ecs:             c,
+		createThrottler: defaultThrottler(),
+		deleteThrottler: defaultThrottler(),
+		batcher:         batcher.NewLowLatency(desc.Disk(c), clock.RealClock{}, 1*time.Second, 8),
+	}
+}
+
+func TestDeleteDisk(t *testing.T) {
+	c, cd := testCreateDelete(t)
+	_, ctx := ktesting.NewTestContext(t)
+
 	c.EXPECT().DeleteDisk(gomock.Any()).Return(deleteDiskResponse, nil)
 
-	_, err := deleteDisk(context.Background(), c, "test-disk")
+	_, err := cd.deleteDisk(ctx, c, "test-disk")
 	assert.Nil(t, err)
 }
 
 func TestDeleteDiskRetryOnInitError(t *testing.T) {
 	t.Parallel()
-	ctrl := gomock.NewController(t)
-	c := cloud.NewMockECSInterface(ctrl)
+	c, cd := testCreateDelete(t)
+	_, ctx := ktesting.NewTestContext(t)
 
 	initErr := alicloudErr.NewServerError(400, `{"Code": "IncorrectDiskStatus.Initializing"}`, "")
 	c.EXPECT().DeleteDisk(gomock.Any()).Return(nil, initErr)
 	c.EXPECT().DeleteDisk(gomock.Any()).Return(deleteDiskResponse, nil)
 
-	_, err := deleteDisk(context.Background(), c, "test-disk")
+	_, err := cd.deleteDisk(ctx, c, "test-disk")
 	assert.Nil(t, err)
 }
 
 func TestDeleteDiskPassthroughError(t *testing.T) {
-	ctrl := gomock.NewController(t)
-	c := cloud.NewMockECSInterface(ctrl)
+	c, cd := testCreateDelete(t)
+	_, ctx := ktesting.NewTestContext(t)
 
 	serverErr := alicloudErr.NewServerError(400, `{"Code": "AnyOtherErrors"}`, "")
 	c.EXPECT().DeleteDisk(gomock.Any()).Return(nil, serverErr)
 
-	_, err := deleteDisk(context.Background(), c, "test-disk")
+	_, err := cd.deleteDisk(ctx, c, "test-disk")
 	assert.Equal(t, serverErr, err)
 }
 
@@ -384,8 +401,8 @@ func TestCreateDisk_Basic(t *testing.T) {
 	}
 	for _, c := range cases {
 		t.Run(c.name, func(t *testing.T) {
-			ctrl := gomock.NewController(t)
-			client := cloud.NewMockECSInterface(ctrl)
+			client, cd := testCreateDelete(t)
+			_, ctx := ktesting.NewTestContext(t)
 
 			if !c.err {
 				client.EXPECT().CreateDisk(gomock.Any()).Return(&ecs.CreateDiskResponse{
@@ -396,7 +413,7 @@ func TestCreateDisk_Basic(t *testing.T) {
 				client.EXPECT().CreateDisk(gomock.Any()).Return(nil, alicloudErr.NewServerError(400, `{"Code": "AnyOtherErrors"}`, ""))
 			}
 
-			diskID, attempt, err := createDisk(client, "disk-name", "", c.args, c.supports, c.instance, c.isVirtualNode)
+			diskID, attempt, err := cd.createDisk(ctx, client, "disk-name", "", c.args, c.supports, c.instance, c.isVirtualNode)
 			if c.err {
 				assert.Error(t, err)
 			} else {
@@ -409,16 +426,16 @@ func TestCreateDisk_Basic(t *testing.T) {
 }
 
 func TestCreateDisk_ServerFailFallback(t *testing.T) {
-	ctrl := gomock.NewController(t)
-	client := cloud.NewMockECSInterface(ctrl)
+	client, cd := testCreateDelete(t)
+	_, ctx := ktesting.NewTestContext(t)
 
 	client.EXPECT().CreateDisk(gomock.Any()).Return(nil, alicloudErr.NewServerError(400, `{"Code": "InvalidDataDiskSize.ValueNotSupported"}`, ""))
 	client.EXPECT().CreateDisk(gomock.Any()).Return(&ecs.CreateDiskResponse{
 		DiskId: "d-123",
 	}, nil)
 
 	args := &diskVolumeArgs{Type: []Category{DiskESSD, DiskESSDAuto}, RequestGB: 20}
-	diskID, attempt, err := createDisk(client, "disk-name", "", args, nil, "", false)
+	diskID, attempt, err := cd.createDisk(ctx, client, "disk-name", "", args, nil, "", false)
 	assert.NoError(t, err)
 	assert.Equal(t, "d-123", diskID)
 	assert.Equal(t, DiskESSDAuto, attempt.Category)
@@ -456,8 +473,8 @@ func TestCreateDisk_ParameterMismatch(t *testing.T) {
 	}
 	for _, c := range cases {
 		t.Run(c.name, func(t *testing.T) {
-			ctrl := gomock.NewController(t)
-			client := cloud.NewMockECSInterface(ctrl)
+			client, cd := testCreateDelete(t)
+			_, ctx := ktesting.NewTestContext(t)
 
 			r1 := client.EXPECT().CreateDisk(gomock.Any()).Return(nil, alicloudErr.NewServerError(400, `{"Code": "IdempotentParameterMismatch"}`, ""))
 			r2 := client.EXPECT().DescribeDisks(gomock.Any()).Return(&ecs.DescribeDisksResponse{
@@ -472,7 +489,7 @@ func TestCreateDisk_ParameterMismatch(t *testing.T) {
 			}
 
 			args := &diskVolumeArgs{Type: []Category{DiskESSD, DiskESSDAuto}, RequestGB: 20}
-			diskID, attempt, err := createDisk(client, "disk-name", "", args, nil, "", false)
+			diskID, attempt, err := cd.createDisk(ctx, client, "disk-name", "", args, nil, "", false)
 			if c.err {
 				assert.Error(t, err)
 			} else {
@@ -492,8 +509,8 @@ func TestCreateDisk_ParameterMismatch(t *testing.T) {
 }
 
 func TestCreateDisk_NoInfiniteLoop(t *testing.T) {
-	ctrl := gomock.NewController(t)
-	client := cloud.NewMockECSInterface(ctrl)
+	client, cd := testCreateDelete(t)
+	_, ctx := ktesting.NewTestContext(t)
 
 	client.EXPECT().CreateDisk(gomock.Any()).Return(nil, alicloudErr.NewServerError(400, `{"Code": "IdempotentParameterMismatch"}`, "")).Times(2)
 	client.EXPECT().DescribeDisks(gomock.Any()).Return(&ecs.DescribeDisksResponse{
@@ -503,7 +520,7 @@ func TestCreateDisk_NoInfiniteLoop(t *testing.T) {
 	}, nil)
 
 	args := &diskVolumeArgs{Type: []Category{DiskESSD}, RequestGB: 20}
-	_, _, err := createDisk(client, "disk-name", "", args, nil, "", false)
+	_, _, err := cd.createDisk(ctx, client, "disk-name", "", args, nil, "", false)
 	assert.Error(t, err)
 }
 

pkg/disk/controllerserver.go

@@ -52,6 +52,7 @@ import (
 type controllerServer struct {
 	recorder       record.EventRecorder
 	ad             DiskAttachDetach
+	cd             DiskCreateDelete
 	meta           metadata.MetadataProvider
 	ecs            cloud.ECSInterface
 	snapshotWaiter waitstatus.StatusWaiter[ecs.Snapshot]
@@ -104,6 +105,12 @@ func NewControllerServer(csiCfg utils.Config, ecs cloud.ECSInterface, m metadata
 			attachThrottler: defaultThrottler(),
 			detachThrottler: defaultThrottler(),
 		},
+		cd: DiskCreateDelete{
+			ecs:             ecs,
+			batcher:         batcher,
+			createThrottler: defaultThrottler(),
+			deleteThrottler: defaultThrottler(),
+		},
 		snapshotWaiter: newSnapshotStatusWaiter(),
 	}
 	detachConcurrency := 1
@@ -177,7 +184,7 @@ func (cs *controllerServer) CreateVolume(ctx context.Context, req *csi.CreateVol
 		isVirtualNode = node.Labels[common.NodeTypeLabelKey] == common.VirtualNodeType
 	}
 
-	diskID, attempt, err := createDisk(cs.ecs, req.GetName(), snapshotID, diskVol, supportedTypes, selectedInstance, isVirtualNode)
+	diskID, attempt, err := cs.cd.createDisk(ctx, req.GetName(), snapshotID, diskVol, supportedTypes, selectedInstance, isVirtualNode)
 	if err != nil {
 		if errors.Is(err, ErrParameterMismatch) {
 			return nil, status.Errorf(codes.AlreadyExists, "volume %s already created but %v", req.Name, err)
@@ -220,7 +227,7 @@ func (cs *controllerServer) DeleteVolume(ctx context.Context, req *csi.DeleteVol
 	var disk *ecs.Disk
 	describeDisk := func() (*csi.DeleteVolumeResponse, error) {
 		var err error
-		disk, err = findDiskByID(req.VolumeId, cs.ecs)
+		disk, err = cs.cd.batcher.Describe(ctx, req.VolumeId)
 		if err != nil {
 			return nil, status.Errorf(codes.Internal, "DeleteVolume: find disk(%s) by id with error: %v", req.VolumeId, err)
 		}
@@ -274,7 +281,7 @@ func (cs *controllerServer) DeleteVolume(ctx context.Context, req *csi.DeleteVol
 		}
 	}
 
-	_, err := deleteDisk(ctx, cs.ecs, req.VolumeId)
+	_, err := cs.cd.deleteDisk(ctx, cs.ecs, req.VolumeId)
 	if err != nil {
 		newErrMsg := utils.FindSuggestionByErrorMessage(err.Error(), utils.DiskDelete)
 		return nil, status.Errorf(codes.Internal, "DeleteVolume: Delete disk with error: %s", newErrMsg)

pkg/disk/desc/describe.go

@@ -5,6 +5,7 @@ import (
 
 	"github.com/aliyun/alibaba-cloud-sdk-go/sdk/requests"
 	"github.com/aliyun/alibaba-cloud-sdk-go/services/ecs"
+	"github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/cloud"
 )
 
 const batchSize = 100
@@ -29,17 +30,21 @@ func encodeIDs(ids []string) string {
 	return string(res)
 }
 
-type Disk struct {
-	*ecs.Client
+type disk struct {
+	cloud.ECSInterface
+}
+
+func Disk(client cloud.ECSInterface) Client[ecs.Disk] {
+	return disk{client}
 }
 
-func (c Disk) Describe(ids []string) (Response[ecs.Disk], error) {
+func (c disk) Describe(ids []string) (Response[ecs.Disk], error) {
 	req := ecs.CreateDescribeDisksRequest()
 	req.DiskIds = encodeIDs(ids)
 	req.PageSize = requests.NewInteger(batchSize)
 
 	ret := Response[ecs.Disk]{}
-	resp, err := c.Client.DescribeDisks(req)
+	resp, err := c.DescribeDisks(req)
 	if err != nil {
 		return ret, err
 	}
@@ -48,15 +53,15 @@ func (c Disk) Describe(ids []string) (Response[ecs.Disk], error) {
 	return ret, nil
 }
 
-func (c Disk) GetID(resource *ecs.Disk) string {
+func (c disk) GetID(resource *ecs.Disk) string {
 	return resource.DiskId
 }
 
-func (c Disk) Type() string {
+func (c disk) Type() string {
 	return "disk"
 }
 
-func (c Disk) BatchSize() int {
+func (c disk) BatchSize() int {
 	return batchSize
 }
 

pkg/disk/disk.go

@@ -243,7 +243,7 @@ func GlobalConfigSet(m metadata.MetadataProvider) utils.Config {
 }
 
 func newBatcher(fromNode bool) (waitstatus.StatusWaiter[ecs.Disk], batcher.Batcher[ecs.Disk]) {
-	client := desc.Disk{Client: GlobalConfigVar.EcsClient}
+	client := desc.Disk(GlobalConfigVar.EcsClient)
 	ctx := context.Background()
 	interval := 1 * time.Second
 	if fromNode {