provider: add new provider SemaphoreProvider

pkg/credentials/provider/go.mod

@@ -1,3 +1,5 @@
 module github.com/AliyunContainerService/ack-ram-tool/pkg/credentials/provider
 
 go 1.16
+
+require golang.org/x/sync v0.7.0

pkg/credentials/provider/go.sum

@@ -0,0 +1,2 @@
+golang.org/x/sync v0.7.0 h1:YsImfSBoP9QPYL0xyKJPq0gcaJdG3rInoqxTWbfQu9M=
+golang.org/x/sync v0.7.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=

pkg/credentials/provider/semaphore_provider.go

@@ -0,0 +1,42 @@
+package provider
+
+import (
+	"context"
+	"fmt"
+	"golang.org/x/sync/semaphore"
+)
+
+type SemaphoreProvider struct {
+	weighted *semaphore.Weighted
+
+	cp CredentialsProvider
+}
+
+type SemaphoreProviderOptions struct {
+	MaxWeight int64
+}
+
+func NewSemaphoreProvider(cp CredentialsProvider, opts SemaphoreProviderOptions) *SemaphoreProvider {
+	opts.applyDefaults()
+
+	w := semaphore.NewWeighted(opts.MaxWeight)
+	return &SemaphoreProvider{
+		weighted: w,
+		cp:       cp,
+	}
+}
+
+func (p *SemaphoreProvider) Credentials(ctx context.Context) (*Credentials, error) {
+	if err := p.weighted.Acquire(ctx, 1); err != nil {
+		return nil, fmt.Errorf("acquire semaphore: %w", err)
+	}
+	defer p.weighted.Release(1)
+
+	return p.cp.Credentials(ctx)
+}
+
+func (o *SemaphoreProviderOptions) applyDefaults() {
+	if o.MaxWeight <= 0 {
+		o.MaxWeight = 1
+	}
+}

pkg/credentials/provider/semaphore_provider_test.go

@@ -0,0 +1,48 @@
+package provider
+
+import (
+	"context"
+	"runtime"
+	"strings"
+	"testing"
+	"time"
+)
+
+func TestSemaphoreProvider_Credentials(t *testing.T) {
+	cp := NewFunctionProvider(func(ctx context.Context) (*Credentials, error) {
+		time.Sleep(time.Millisecond * 200)
+		return &Credentials{}, nil
+	})
+
+	p := NewSemaphoreProvider(cp, SemaphoreProviderOptions{})
+
+	go p.Credentials(context.TODO())
+	runtime.Gosched()
+
+	ctx, cancel := context.WithTimeout(context.TODO(), time.Millisecond*100)
+	defer cancel()
+
+	// no free lock
+	_, err := p.Credentials(ctx)
+	if err == nil {
+		t.Error("err should not be nil")
+	} else {
+		if !strings.Contains(err.Error(), "acquire semaphore: context deadline exceeded") {
+			t.Log(err)
+			t.Error("err should include 'context deadline exceeded'")
+		}
+	}
+
+	time.Sleep(time.Millisecond * 300)
+	// has free lock
+	ctx2, cancel2 := context.WithTimeout(context.TODO(), time.Millisecond*100)
+	defer cancel2()
+	cred, err := p.Credentials(ctx2)
+	if err != nil {
+		t.Log(err)
+		t.Error("err should be nil")
+	}
+	if cred == nil {
+		t.Error("cred should not be nil")
+	}
+}