Merge pull request #49 from whywishfree/feature/for_opensource

Fixing the issue of concurrent requests

Dockerfile

@@ -16,4 +16,4 @@ RUN apk add --no-cache ca-certificates && \
 COPY --from=builder /go/src/github.com/AliyunContainerService/ack-secret-manager/build/bin/ack-secret-manager /bin/ack-secret-manager
 #ADD ./build/bin/ack-secret-manager /bin/ack-secret-manager
 
-CMD ["./ack-secret-manager"]
+CMD ["./ack-secret-manager"]

charts/ack-secret-manager/Chart.yaml

@@ -1,14 +1,14 @@
 apiVersion: v1
 name: ack-secret-manager
-version: 0.5.0
-appVersion: 0.5.0
+version: 0.5.2
+appVersion: 0.5.2
 description: Provides external secret definitions for importing secret managed in Alibaba Cloud KMS secret-manager
 keywords:
   - secret-manager
   - secrets
   - namespace:kube-system
   - releaseName:ack-secret-manager
-  - arch:amd64
+  - arch:amd64,arm64
 home: https://github.com/AliyunContainerService/ack-secret-manager
 sources:
   - https://github.com/AliyunContainerService/ack-secret-manager

charts/ack-secret-manager/README.md

@@ -89,6 +89,7 @@
 | command.region                                      | 从指定region拉取secret凭据                                   |                        |
 | command.disablePolling                              | 关闭从KMS后端自动同步拉取最新的凭据内容，默认false           | false                  |
 | command.pollingInterval                             | 从KMS后端同步存量secret实例的间隔时间                        | 120s                   |
+| command.maxConcurrentSecretPulls                    | secret 同步的最大并发数量                                    | 5                      |
 | image.repository                                    | 指定的ack-secret-manager 镜像仓库名称                        | acs/ack-secret-manager |
 | image.tag                                           | 指定的ack-secret-manager 镜像tag                             | v0.5.0                 |
 | image.pullPolicy                                    | 镜像拉取策略，默认为Always                                   | Always                 |
@@ -486,4 +487,7 @@ ack-secret-manager 涉及了两种 CRD，SecretStore 用于存放访问凭据（
 | 版本号  | 变更时间       | 变更内容                                                     |
 | ------- | -------------- | ------------------------------------------------------------ |
 | `0.4.0` | 2022年12月22日 | 支持基于JMES解析提取JSON格式的密文字段                       |
-| `0.5.0` | 2023年10月10日 | 1.支持专属版 KMS 凭据同步<br />2.多阿里云访问凭据管理<br />3.凭据自解析与键规则替换<br />4.共享版 KMS 跨账号凭据同步 |
+| `0.5.0` | 2023年10月10日 | 1.支持专属版 KMS 凭据同步<br />2.多阿里云访问凭据管理<br />3.凭据自解析与键规则替换<br />4.共享版 KMS 跨账号凭据同步 |
+| `0.5.1` | 2023年10月18日 | 部分功能与性能优化                                           |
+| `0.5.2` | 2024年8月1日   | 大规模资源同步并发优化                                       |
+

charts/ack-secret-manager/README_EN.md

@@ -87,6 +87,7 @@ Make sure that the current account has sufficient permissions to access the Alib
 | command.region                                      | Pull secret credentials from the specified region            |                        |
 | command.disablePolling                              | Turn off automatic synchronization of pulling the latest credential content from the KMS backend, default false | false                  |
 | command.pollingInterval                             | The interval for synchronizing existing secret instances from the KMS backend | 120s                   |
+| command.maxConcurrentSecretPulls                    | Maximum concurrent synchronization of secrets                | 5                      |
 | image.repository                                    | Specified ack-secret-manager mirror warehouse name           | acs/ack-secret-manager |
 | image.tag                                           | Specified ack-secret-manager image tag                       | v0.5.0                 |
 | image.pullPolicy                                    | Image pull strategy, default is Always                       | Always                 |
@@ -482,4 +483,6 @@ ack-secret-manager involves two CRDs. SecretStore is used to store access creden
 | Version | Date       | Changes                                                      |
 | ------- | ---------- | ------------------------------------------------------------ |
 | `0.4.0` | 2022/12/22 | Support sync specific key-value pairs extract from a JSON-formatted secret based on JMES path |
-| `0.5.0` | 2023/10/10 | 1.dedicated KMS credential synchronization<br />2.multiple Alibaba Cloud access credentials management,<br />3.self-resolving credentials and key rule replacement<br />4.shared KMS cross-account credential synchronization. |
+| `0.5.0` | 2023/10/10 | 1.dedicated KMS credential synchronization<br />2.multiple Alibaba Cloud access credentials management,<br />3.self-resolving credentials and key rule replacement<br />4.shared KMS cross-account credential synchronization. |
+| `0.5.1` | 2023/10/18 | Function and performance optimization                        |
+| `0.5.2` | 2024/08/01 | Large-scale resource synchronization concurrency optimization |

charts/ack-secret-manager/values.yaml

@@ -11,6 +11,7 @@ command:
   region:
   disablePolling: false
   pollingInterval: 120s
+  maxConcurrentSecretPulls: 5
 # watchamespaces:
 # excludeamespaces:
 
@@ -65,7 +66,7 @@ replicaCount: 2
 
 image:
   repository: registry-cn-hangzhou.ack.aliyuncs.com/acs/ack-secret-manager
-  tag: v0.5.0
+  tag: v0.5.2
   pullPolicy: Always
 
 cleanupImage:

cmd/manager/main.go

@@ -19,6 +19,7 @@ import (
 	"context"
 	"flag"
 	"fmt"
+	"golang.org/x/sync/semaphore"
 	"os"
 	"runtime"
 	"strings"
@@ -67,6 +68,7 @@ func main() {
 	var excludeNamespaces string
 	var region string
 	var tokenRotationPeriod time.Duration
+	var maxConcurrentSecretPulls int
 
 	flag.StringVar(&selectedBackend, "backend", "alicloud-kms", "Selected backend. Only alicloud-kms supported")
 	flag.DurationVar(&rotationInterval, "polling-interval", 120*time.Second, "How often the controller will sync existing secret from kms")
@@ -77,6 +79,8 @@ func main() {
 	flag.StringVar(&region, "region", "", "Region id, change it according to where you want to pull the secret from")
 	flag.StringVar(&watchNamespaces, "watch-namespaces", "", "Comma separated list of namespaces that ack-secret-manager watch. By default all namespaces are watched.")
 	flag.StringVar(&excludeNamespaces, "exclude-namespaces", "", "Comma separated list of namespaces that that ack-secret-manager will not watch. By default all namespaces are watched.")
+	flag.IntVar(&maxConcurrentSecretPulls, "max-concurrent-secret-pulls", 5, "used to control how many secrets are pulled at the same time.\n\n")
+
 	flag.Parse()
 
 	ctrl.SetLogger(zap.New())
@@ -100,9 +104,13 @@ func main() {
 	if region == "" || region != instanceRegion {
 		region = instanceRegion
 	}
+	opts := &backend.ProviderOptions{
+		Region:        region,
+		MaxConcurrent: maxConcurrentSecretPulls,
+	}
 	for providerName, f := range backend.SupportProvider {
 		log.Info("new provider ", providerName)
-		f(region)
+		f(opts)
 	}
 
 	err = backend.NewProviderClientByENV(ctx, region)
@@ -142,6 +150,7 @@ func main() {
 			watchNs[ns] = false
 		}
 	}
+	w := semaphore.NewWeighted(int64(opts.MaxConcurrent))
 	esReconciler := externalsecret.ExternalSecretReconciler{
 		Client:               mgr.GetClient(),
 		APIReader:            mgr.GetAPIReader(),
@@ -150,6 +159,7 @@ func main() {
 		ReconciliationPeriod: reconcilePeriod,
 		WatchNamespaces:      watchNs,
 		RotationInterval:     rotationInterval,
+		ConcurrentController: w,
 	}
 	err = (&esReconciler).SetupWithManager(mgr, reconcileCount)
 	if err != nil {

go.mod

@@ -3,7 +3,7 @@ module github.com/AliyunContainerService/ack-secret-manager
 go 1.18
 
 require (
-	github.com/AliyunContainerService/ack-ram-tool/pkg/credentials/provider v0.10.0
+	github.com/AliyunContainerService/ack-ram-tool/pkg/credentials/provider v0.13.0
 	github.com/alibabacloud-go/darabonba-openapi v0.1.4
 	github.com/alibabacloud-go/kms-20160120/v2 v2.0.0
 	github.com/alibabacloud-go/tea v1.2.1
@@ -75,6 +75,7 @@ require (
 	golang.org/x/crypto v0.10.0 // indirect
 	golang.org/x/mod v0.8.0 // indirect
 	golang.org/x/oauth2 v0.0.0-20200107190931-bf48bf16ab8d // indirect
+	golang.org/x/sync v0.7.0 // indirect
 	golang.org/x/sys v0.12.0 // indirect
 	golang.org/x/term v0.9.0 // indirect
 	golang.org/x/text v0.10.0 // indirect

go.sum

@@ -22,8 +22,10 @@ cloud.google.com/go/storage v1.0.0/go.mod h1:IhtSnM/ZTZV8YYJWCY8RULGVqBDmpoyjwiy
 cloud.google.com/go/storage v1.5.0/go.mod h1:tpKbwo567HUNpVclU5sGELwQWBDZ8gh0ZeosJ0Rtdos=
 cloud.google.com/go/storage v1.6.0/go.mod h1:N7U0C8pVQ/+NIKOBQyamJIeKQKkZ+mxpohlUTyfDhBk=
 dmitri.shuralyov.com/gpu/mtl v0.0.0-20190408044501-666a987793e9/go.mod h1:H6x//7gZCb22OMCxBHrMx7a5I7Hp++hsVxbQ4BYO7hU=
-github.com/AliyunContainerService/ack-ram-tool/pkg/credentials/provider v0.10.0 h1:I/2/vwxNbykn6fQViKe/EDo578RdPX3+4R1Rs5yLNjU=
-github.com/AliyunContainerService/ack-ram-tool/pkg/credentials/provider v0.10.0/go.mod h1:ULtI7L9xkNeJ07YNqSeT5EhjQAl1CpTgPcUn4KoNcuc=
+github.com/AliyunContainerService/ack-ram-tool/pkg/credentials/provider v0.12.0 h1:nUTgdlM3aLzOHQhdxrtUdO2T4vqBoW3kn6C9QIeClKk=
+github.com/AliyunContainerService/ack-ram-tool/pkg/credentials/provider v0.12.0/go.mod h1:tlqp9mUGbsP+0z3Q+c0Q5MgSdq/OMwQhm5bffR3Q3ss=
+github.com/AliyunContainerService/ack-ram-tool/pkg/credentials/provider v0.13.0 h1:J6JXctkQI6QCdPOKTyERNf2KdgNkBVQfwJUn2qy12TQ=
+github.com/AliyunContainerService/ack-ram-tool/pkg/credentials/provider v0.13.0/go.mod h1:tlqp9mUGbsP+0z3Q+c0Q5MgSdq/OMwQhm5bffR3Q3ss=
 github.com/Azure/go-ansiterm v0.0.0-20170929234023-d6e3b3328b78/go.mod h1:LmzpDX56iTiv29bbRTIsUNlaFfuhWRQBWjQdVyAevI8=
 github.com/Azure/go-autorest v14.2.0+incompatible/go.mod h1:r+4oMnoxhatjLLJ6zxSWATqVooLgysK6ZNox3g/xq24=
 github.com/Azure/go-autorest/autorest v0.11.1/go.mod h1:JFgpikqFJ/MleTTxwepExTKnFUKKszPS8UavbQYUMuw=
@@ -671,6 +673,8 @@ golang.org/x/sync v0.0.0-20200317015054-43a5402ce75a/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.7.0 h1:YsImfSBoP9QPYL0xyKJPq0gcaJdG3rInoqxTWbfQu9M=
+golang.org/x/sync v0.7.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
 golang.org/x/sys v0.0.0-20170830134202-bb24a47a89ea/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180823144017-11551d06cbcc/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=

pkg/backend/kms/auth.go

@@ -32,6 +32,7 @@ type authConfig struct {
 
 func (a *authConfig) getKMSAuthCred(region string, p *Provider) (credentials.Credential, error) {
 	providers := make([]provider.CredentialsProvider, 0)
+	var semaphoreProvider *provider.SemaphoreProvider
 	if a.oidcArn != "" && a.roleArn != "" {
 		oidcProvider := provider.NewOIDCProvider(provider.OIDCProviderOptions{
 			STSEndpoint:     provider.GetSTSEndpoint(region, true),
@@ -60,20 +61,24 @@ func (a *authConfig) getKMSAuthCred(region string, p *Provider) (credentials.Cre
 	}))
 	chainProvider := provider.NewChainProvider(providers...)
 	var remoteRoleProvider *provider.RoleArnProvider
+	var cred *provider.CredentialForV2SDK
 	if a.remoteRoleArn != "" && a.remoteRoleSessionName != "" {
 		remoteRoleProvider = provider.NewRoleArnProvider(chainProvider, a.remoteRoleArn, provider.RoleArnProviderOptions{
 			STSEndpoint:   provider.GetSTSEndpoint(region, true),
 			SessionName:   a.remoteRoleSessionName,
 			RefreshPeriod: a.refreshPeriod,
 		})
-	}
-	var cred *provider.CredentialForV2SDK
-	if remoteRoleProvider != nil {
-		p.RegisterRamProvider(a.clientName, remoteRoleProvider)
-		cred = provider.NewCredentialForV2SDK(remoteRoleProvider, provider.CredentialForV2SDKOptions{})
+		semaphoreProvider = provider.NewSemaphoreProvider(remoteRoleProvider, provider.SemaphoreProviderOptions{
+			MaxWeight: int64(p.maxConcurrentCount),
+		})
 	} else {
-		p.RegisterRamProvider(a.clientName, chainProvider)
-		cred = provider.NewCredentialForV2SDK(chainProvider, provider.CredentialForV2SDKOptions{})
+		semaphoreProvider = provider.NewSemaphoreProvider(chainProvider, provider.SemaphoreProviderOptions{
+			MaxWeight: int64(p.maxConcurrentCount),
+		})
 	}
+	p.RegisterRamProvider(a.clientName, semaphoreProvider)
+	cred = provider.NewCredentialForV2SDK(semaphoreProvider, provider.CredentialForV2SDKOptions{
+		CredentialRetrievalTimeout: 10 * time.Minute,
+	})
 	return cred, nil
 }

pkg/backend/kms/client_manager.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"sync"
+	"time"
 
 	"k8s.io/klog"
 
@@ -35,7 +36,10 @@ func (m *Manager) RegisterRamProvider(clientName string, stopper provider.Stoppe
 	defer m.ramLock.Unlock()
 	providerIns, ok := m.ramProvider[clientName]
 	if ok {
-		providerIns.Stop(context.Background())
+		timeoutCtx, cancel := context.WithTimeout(context.Background(), 30*time.Minute)
+		// cancel is earlier than m.ramLock.Unlock
+		defer cancel()
+		providerIns.Stop(timeoutCtx)
 	}
 	m.ramProvider[clientName] = stopper
 	klog.Infof("register provider %v success", clientName)
@@ -48,7 +52,10 @@ func (m *Manager) StopProvider(clientName string) {
 	if !ok || providerIns == nil {
 		return
 	}
-	providerIns.Stop(context.Background())
+	timeoutCtx, cancel := context.WithTimeout(context.Background(), 30*time.Minute)
+	// cancel is earlier than m.ramLock.Unlock
+	defer cancel()
+	providerIns.Stop(timeoutCtx)
 	delete(m.ramProvider, clientName)
 	klog.Infof("stop provider %v success", clientName)
 }

pkg/backend/kms/client_provider.go

@@ -36,15 +36,17 @@ func init() {
 // Provider provides the ability to generate kms clients and manage kms clients
 type Provider struct {
 	*Manager
-	region string
-	name   string
+	region             string
+	name               string
+	maxConcurrentCount int
 }
 
-func NewProvider(region string) {
+func NewProvider(opts *backend.ProviderOptions) {
 	provider := &Provider{
-		Manager: NewManager(region),
-		region:  region,
-		name:    ProviderName,
+		Manager:            NewManager(opts.Region),
+		region:             opts.Region,
+		name:               ProviderName,
+		maxConcurrentCount: 5,
 	}
 	backend.RegisterProvider(ProviderName, provider)
 }

pkg/backend/provider.go

@@ -12,7 +12,12 @@ import (
 
 const EnvClient = "env.client"
 
-type CreateProvider func(region string)
+type CreateProvider func(opt *ProviderOptions)
+
+type ProviderOptions struct {
+	Region        string
+	MaxConcurrent int
+}
 
 var (
 	SupportProvider map[string]CreateProvider

pkg/controller/externalsecret/secret_controller.go

@@ -19,6 +19,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"golang.org/x/sync/semaphore"
 	"reflect"
 	"sync"
 	"time"
@@ -57,6 +58,7 @@ type ExternalSecretReconciler struct {
 	rotationTicker       *time.Ticker
 	secrets              sync.Map  // secrets map is the cache for secrets.
 	closing              chan bool // close channel.
+	ConcurrentController *semaphore.Weighted
 }
 
 var (
@@ -255,9 +257,16 @@ func (r *ExternalSecretReconciler) rotate() {
 		es := v.(*api.ExternalSecret)
 		r.Log.Info("rotate checking secret", "index", k)
 		// Re-generate secret if update in kms secret-manager.
+		timeoutContext, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
+		if err := r.ConcurrentController.Acquire(timeoutContext, 1); err != nil {
+			cancel()
+			return true
+		}
 		wg.Add(1)
 		go func() {
 			defer wg.Done()
+			defer r.ConcurrentController.Release(1)
+			defer cancel()
 			updated, _ := r.syncIfNeedUpdate(es)
 			if updated {
 				secretMap.Store(fmt.Sprintf("%s/%s", es.Namespace, es.Name), es)