nas: use golang/x rate limiter

go.mod

@@ -27,7 +27,6 @@ require (
 	github.com/sirupsen/logrus v1.9.3
 	github.com/spf13/pflag v1.0.5
 	github.com/stretchr/testify v1.10.0
-	go.uber.org/ratelimit v0.1.0
 	golang.org/x/sys v0.37.0
 	golang.org/x/time v0.7.0
 	google.golang.org/grpc v1.75.0

go.sum

@@ -375,8 +375,6 @@ go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
 go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
 go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0=
 go.uber.org/multierr v1.11.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y=
-go.uber.org/ratelimit v0.1.0 h1:U2AruXqeTb4Eh9sYQSTrMhH8Cb7M0Ian2ibBOnBcnAw=
-go.uber.org/ratelimit v0.1.0/go.mod h1:2X8KaoNd1J0lZV+PxJk/5+DGbO/tpwLR1m++a7FnB/Y=
 go.uber.org/zap v1.27.0 h1:aJMhYGrd5QSmlpLMr2MftRKl7t8J8PTZPA732ud/XR8=
 go.uber.org/zap v1.27.0/go.mod h1:GB2qFLM7cTU87MWRP2mPIjqfIDnGu+VIO4V/SdhGo2E=
 golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=

pkg/nas/cloud/nas_client_factory.go

@@ -5,16 +5,15 @@ import (
 	"strconv"
 
 	"github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/nas/interfaces"
+	"golang.org/x/time/rate"
 	"k8s.io/klog/v2"
-
-	"go.uber.org/ratelimit"
 )
 
 const defaultQps = 2
 
 type NasClientFactory struct {
 	// ratelimiter only takes effect on v2 client
-	limiter ratelimit.Limiter
+	limiter *rate.Limiter
 }
 
 func NewNasClientFactory() *NasClientFactory {
@@ -30,7 +29,7 @@ func NewNasClientFactory() *NasClientFactory {
 		}
 	}
 	return &NasClientFactory{
-		limiter: ratelimit.New(qps),
+		limiter: rate.NewLimiter(rate.Limit(qps), 10),
 	}
 }
 

pkg/nas/cloud/nas_client_factory_test.go

@@ -1,28 +1,29 @@
 package cloud
 
 import (
-	"github.com/stretchr/testify/assert"
-	"go.uber.org/ratelimit"
 	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"golang.org/x/time/rate"
 )
 
 func TestNasClientFactory(t *testing.T) {
 	t.Parallel()
 	actual := NewNasClientFactory()
-	expected := &NasClientFactory{ratelimit.New(defaultQps)}
+	expected := &NasClientFactory{rate.NewLimiter(defaultQps, 10)}
 	assert.Equal(t, expected, actual)
 }
 
 func TestNasClientFactoryValidEnv(t *testing.T) {
 	t.Setenv("NAS_LIMIT_PERSECOND", "3")
-	expected := &NasClientFactory{ratelimit.New(3)}
+	expected := &NasClientFactory{rate.NewLimiter(3, 10)}
 	actual := NewNasClientFactory()
 	assert.Equal(t, expected, actual)
 }
 
 func TestNasClientFactoryInvalidEnv(t *testing.T) {
 	t.Setenv("NAS_LIMIT_PERSECOND", "3i")
-	expected := &NasClientFactory{ratelimit.New(defaultQps)}
+	expected := &NasClientFactory{rate.NewLimiter(defaultQps, 10)}
 	actual := NewNasClientFactory()
 	assert.Equal(t, expected, actual)
 }

pkg/nas/cloud/nas_client_v2.go

@@ -1,6 +1,7 @@
 package cloud
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"os"
@@ -12,7 +13,7 @@ import (
 	"github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/credentials"
 	"github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/nas/interfaces"
 	utilshttp "github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/utils/http"
-	"go.uber.org/ratelimit"
+	"golang.org/x/time/rate"
 	"k8s.io/klog/v2"
 )
 
@@ -60,12 +61,14 @@ func NewNasClientV2(region string) (*sdk.Client, error) {
 
 type NasClientV2 struct {
 	region  string
-	limiter ratelimit.Limiter
+	limiter *rate.Limiter
 	client  interfaces.NasV2Interface
 }
 
 func (c *NasClientV2) CreateDir(req *sdk.CreateDirRequest) error {
-	c.limiter.Take()
+	if err := c.limiter.Wait(context.TODO()); err != nil {
+		return fmt.Errorf("error while waiting for rate limiter: %w", err)
+	}
 	resp, err := c.client.CreateDir(req)
 	logger := klog.Background().WithValues("request", req, "response", resp)
 	if err == nil {
@@ -77,7 +80,9 @@ func (c *NasClientV2) CreateDir(req *sdk.CreateDirRequest) error {
 }
 
 func (c *NasClientV2) SetDirQuota(req *sdk.SetDirQuotaRequest) error {
-	c.limiter.Take()
+	if err := c.limiter.Wait(context.TODO()); err != nil {
+		return fmt.Errorf("error while waiting for rate limiter: %w", err)
+	}
 	resp, err := c.client.SetDirQuota(req)
 	if err == nil && resp.Body != nil && !tea.BoolValue(resp.Body.Success) {
 		err = errors.New("response indicates a failure")
@@ -92,7 +97,9 @@ func (c *NasClientV2) SetDirQuota(req *sdk.SetDirQuotaRequest) error {
 }
 
 func (c *NasClientV2) CancelDirQuota(req *sdk.CancelDirQuotaRequest) error {
-	c.limiter.Take()
+	if err := c.limiter.Wait(context.TODO()); err != nil {
+		return fmt.Errorf("error while waiting for rate limiter: %w", err)
+	}
 	resp, err := c.client.CancelDirQuota(req)
 	if err == nil {
 		if !tea.BoolValue(resp.Body.Success) {
@@ -115,13 +122,17 @@ func (c *NasClientV2) CancelDirQuota(req *sdk.CancelDirQuotaRequest) error {
 }
 
 func (c *NasClientV2) GetRecycleBinAttribute(filesystemId string) (*sdk.GetRecycleBinAttributeResponse, error) {
-	c.limiter.Take()
+	if err := c.limiter.Wait(context.TODO()); err != nil {
+		return nil, fmt.Errorf("error while waiting for rate limiter: %w", err)
+	}
 	req := &sdk.GetRecycleBinAttributeRequest{FileSystemId: &filesystemId}
 	return c.client.GetRecycleBinAttribute(req)
 }
 
 func (c *NasClientV2) CreateAccesspoint(req *sdk.CreateAccessPointRequest) (*sdk.CreateAccessPointResponse, error) {
-	c.limiter.Take()
+	if err := c.limiter.Wait(context.TODO()); err != nil {
+		return nil, fmt.Errorf("error while waiting for rate limiter: %w", err)
+	}
 	resp, err := c.client.CreateAccessPoint(req)
 	logger := klog.Background().WithValues("request", req, "response", resp)
 	if err == nil {
@@ -133,7 +144,9 @@ func (c *NasClientV2) CreateAccesspoint(req *sdk.CreateAccessPointRequest) (*sdk
 }
 
 func (c *NasClientV2) DeleteAccesspoint(filesystemId, accessPointId string) error {
-	c.limiter.Take()
+	if err := c.limiter.Wait(context.TODO()); err != nil {
+		return fmt.Errorf("error while waiting for rate limiter: %w", err)
+	}
 	req := &sdk.DeleteAccessPointRequest{
 		AccessPointId: &accessPointId,
 		FileSystemId:  &filesystemId,
@@ -149,15 +162,19 @@ func (c *NasClientV2) DeleteAccesspoint(filesystemId, accessPointId string) erro
 }
 
 func (c *NasClientV2) DescribeAccesspoint(filesystemId, accessPointId string) (*sdk.DescribeAccessPointResponse, error) {
-	c.limiter.Take()
+	if err := c.limiter.Wait(context.TODO()); err != nil {
+		return nil, fmt.Errorf("error while waiting for rate limiter: %w", err)
+	}
 	return c.client.DescribeAccessPoint(&sdk.DescribeAccessPointRequest{
 		AccessPointId: &accessPointId,
 		FileSystemId:  &filesystemId,
 	})
 }
 
 func (c *NasClientV2) DescribeFileSystems(filesystemID string) (*sdk.DescribeFileSystemsResponse, error) {
-	c.limiter.Take()
+	if err := c.limiter.Wait(context.TODO()); err != nil {
+		return nil, fmt.Errorf("error while waiting for rate limiter: %w", err)
+	}
 	return c.client.DescribeFileSystems(&sdk.DescribeFileSystemsRequest{
 		FileSystemId: &filesystemID,
 	})

pkg/nas/cloud/nas_client_v2_test.go

@@ -8,7 +8,7 @@ import (
 	"github.com/golang/mock/gomock"
 	"github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/nas/interfaces"
 	"github.com/stretchr/testify/assert"
-	"go.uber.org/ratelimit"
+	"golang.org/x/time/rate"
 )
 
 const nasV2Region = "cn-hangzhou"
@@ -39,7 +39,7 @@ func newNasClientV2ForTest(t *testing.T, mockExpects func(*interfaces.MockNasV2I
 	mockExpects(mockNas)
 	return &NasClientV2{
 		region:  nasV2Region,
-		limiter: ratelimit.New(2),
+		limiter: rate.NewLimiter(2, 10),
 		client:  mockNas,
 	}
 }