[WIP]refactor: softly connect to cloudwego-contrib/cwgo-pkg

.github/workflows/tests.yml

@@ -6,7 +6,7 @@ jobs:
   unit-benchmark-test:
     strategy:
       matrix:
-        go: [ 1.19 ]
+        go: [ 1.21 ]
         os: [ X64, ARM64 ]
     runs-on: ${{ matrix.os }}
     steps:

.golangci.yaml

@@ -2,14 +2,10 @@
 run:
   # include `vendor` `third_party` `testdata` `examples` `Godeps` `builtin`
   skip-dirs-use-default: true
-  skip-dirs:
-    - kitex_gen
-  skip-files:
-    - ".*\\.mock\\.go$"
 # output configuration options
 output:
   # Format: colored-line-number|line-number|json|tab|checkstyle|code-climate|junit-xml|github-actions
-  format: colored-line-number
+  formats: colored-line-number
 # All available settings of specific linters.
 # Refer to https://golangci-lint.run/usage/linters
 linters-settings:
@@ -29,8 +25,10 @@ linters:
   disable:
     - errcheck
     - typecheck
-    - deadcode
-    - varcheck
     - staticcheck
 issues:
   exclude-use-default: true
+  exclude-files:
+    - ".*\\.mock\\.go$"
+  exclude-dirs:
+    - kitex_gen

client/circuit_breaker.go

@@ -15,100 +15,14 @@
 package client
 
 import (
-	"context"
-	"strings"
-
-	"github.com/cloudwego/kitex/pkg/klog"
-
+	cwclient "github.com/cloudwego-contrib/cwgo-pkg/config/etcd/client"
 	"github.com/cloudwego/kitex/client"
-	"github.com/cloudwego/kitex/pkg/circuitbreak"
-	"github.com/cloudwego/kitex/pkg/rpcinfo"
+
 	"github.com/kitex-contrib/config-etcd/etcd"
 	"github.com/kitex-contrib/config-etcd/utils"
 )
 
 // WithCircuitBreaker sets the circuit breaker policy from etcd configuration center.
 func WithCircuitBreaker(dest, src string, etcdClient etcd.Client, uniqueID int64, opts utils.Options) []client.Option {
-	param, err := etcdClient.ClientConfigParam(&etcd.ConfigParamConfig{
-		Category:          circuitBreakerConfigName,
-		ServerServiceName: dest,
-		ClientServiceName: src,
-	})
-	if err != nil {
-		panic(err)
-	}
-
-	for _, f := range opts.EtcdCustomFunctions {
-		f(&param)
-	}
-	key := param.Prefix + "/" + param.Path
-	cbSuite := initCircuitBreaker(key, dest, src, etcdClient, uniqueID)
-
-	return []client.Option{
-		client.WithCircuitBreaker(cbSuite),
-		client.WithCloseCallbacks(func() error {
-			// cancel the configuration listener when client is closed.
-			etcdClient.DeregisterConfig(key, uniqueID)
-			err = cbSuite.Close()
-			if err != nil {
-				return err
-			}
-			return nil
-		}),
-	}
-}
-
-// keep consistent when initialising the circuit breaker suit and updating
-// the circuit breaker policy.
-func genServiceCBKeyWithRPCInfo(ri rpcinfo.RPCInfo) string {
-	if ri == nil {
-		return ""
-	}
-	return genServiceCBKey(ri.To().ServiceName(), ri.To().Method())
-}
-
-func genServiceCBKey(toService, method string) string {
-	sum := len(toService) + len(method) + 2
-	var buf strings.Builder
-	buf.Grow(sum)
-	buf.WriteString(toService)
-	buf.WriteByte('/')
-	buf.WriteString(method)
-	return buf.String()
-}
-
-func initCircuitBreaker(key, dest, src string,
-	etcdClient etcd.Client, uniqueID int64,
-) *circuitbreak.CBSuite {
-	cb := circuitbreak.NewCBSuite(genServiceCBKeyWithRPCInfo)
-	lcb := utils.ThreadSafeSet{}
-
-	onChangeCallback := func(restoreDefault bool, data string, parser etcd.ConfigParser) {
-		set := utils.Set{}
-		configs := map[string]circuitbreak.CBConfig{}
-
-		if !restoreDefault {
-			err := parser.Decode(data, &configs)
-			if err != nil {
-				klog.Warnf("[etcd] %s client etcd circuit breaker: unmarshal data %s failed: %s, skip...", key, data, err)
-				return
-			}
-		}
-
-		for method, config := range configs {
-			set[method] = true
-			key := genServiceCBKey(dest, method)
-			cb.UpdateServiceCBConfig(key, config)
-		}
-
-		for _, method := range lcb.DiffAndEmplace(set) {
-			key := genServiceCBKey(dest, method)
-			// For deleted method configs, set to default policy
-			cb.UpdateServiceCBConfig(key, circuitbreak.GetDefaultCBConfig())
-		}
-	}
-
-	etcdClient.RegisterConfigCallback(context.Background(), key, uniqueID, onChangeCallback)
-
-	return cb
+	return cwclient.WithCircuitBreaker(dest, src, etcdClient, uniqueID, opts)
 }

client/degradation.go

@@ -15,52 +15,13 @@
 package client
 
 import (
-	"context"
+	cwclient "github.com/cloudwego-contrib/cwgo-pkg/config/etcd/client"
 
 	"github.com/cloudwego/kitex/client"
-	"github.com/cloudwego/kitex/pkg/klog"
 	"github.com/kitex-contrib/config-etcd/etcd"
-	"github.com/kitex-contrib/config-etcd/pkg/degradation"
 	"github.com/kitex-contrib/config-etcd/utils"
 )
 
 func WithDegradation(dest, src string, etcdClient etcd.Client, uniqueID int64, opts utils.Options) []client.Option {
-	param, err := etcdClient.ClientConfigParam(&etcd.ConfigParamConfig{
-		Category:          degradationConfigName,
-		ServerServiceName: dest,
-		ClientServiceName: src,
-	})
-	if err != nil {
-		panic(err)
-	}
-	for _, f := range opts.EtcdCustomFunctions {
-		f(&param)
-	}
-	key := param.Prefix + "/" + param.Path
-	container := initDegradationOptions(key, dest, uniqueID, etcdClient)
-	return []client.Option{
-		client.WithACLRules(container.GetAclRule()),
-		client.WithCloseCallbacks(func() error {
-			// cancel the configuration listener when client is closed.
-			etcdClient.DeregisterConfig(key, uniqueID)
-			return nil
-		}),
-	}
-}
-
-func initDegradationOptions(key, dest string, uniqueID int64, etcdClient etcd.Client) *degradation.Container {
-	container := degradation.NewContainer()
-	onChangeCallback := func(restoreDefault bool, data string, parser etcd.ConfigParser) {
-		config := &degradation.Config{}
-		if !restoreDefault {
-			err := parser.Decode(data, config)
-			if err != nil {
-				klog.Warnf("[etcd] %s server etcd degradation config: unmarshal data %s failed: %s, skip...", key, data, err)
-				return
-			}
-		}
-		container.NotifyPolicyChange(config)
-	}
-	etcdClient.RegisterConfigCallback(context.Background(), key, uniqueID, onChangeCallback)
-	return container
+	return cwclient.WithDegradation(dest, src, etcdClient, uniqueID, opts)
 }

client/retry.go

@@ -15,78 +15,14 @@
 package client
 
 import (
-	"context"
+	cwclient "github.com/cloudwego-contrib/cwgo-pkg/config/etcd/client"
 
 	"github.com/cloudwego/kitex/client"
-	"github.com/cloudwego/kitex/pkg/klog"
-	"github.com/cloudwego/kitex/pkg/retry"
 	"github.com/kitex-contrib/config-etcd/etcd"
 	"github.com/kitex-contrib/config-etcd/utils"
 )
 
 // WithRetryPolicy sets the retry policy from etcd configuration center.
 func WithRetryPolicy(dest, src string, etcdClient etcd.Client, uniqueID int64, opts utils.Options) []client.Option {
-	param, err := etcdClient.ClientConfigParam(&etcd.ConfigParamConfig{
-		Category:          retryConfigName,
-		ServerServiceName: dest,
-		ClientServiceName: src,
-	})
-	if err != nil {
-		panic(err)
-	}
-
-	for _, f := range opts.EtcdCustomFunctions {
-		f(&param)
-	}
-	key := param.Prefix + "/" + param.Path
-	rc := initRetryContainer(key, dest, etcdClient, uniqueID)
-	return []client.Option{
-		client.WithRetryContainer(rc),
-		client.WithCloseCallbacks(func() error {
-			// cancel the configuration listener when client is closed.
-			etcdClient.DeregisterConfig(key, uniqueID)
-			return nil
-		}),
-		client.WithCloseCallbacks(rc.Close),
-	}
-}
-
-func initRetryContainer(key, dest string,
-	etcdClient etcd.Client, uniqueID int64,
-) *retry.Container {
-	retryContainer := retry.NewRetryContainerWithPercentageLimit()
-
-	ts := utils.ThreadSafeSet{}
-
-	onChangeCallback := func(restoreDefault bool, data string, parser etcd.ConfigParser) {
-		// the key is method name, wildcard "*" can match anything.
-		rcs := map[string]*retry.Policy{}
-
-		if !restoreDefault {
-			err := parser.Decode(data, &rcs)
-			if err != nil {
-				klog.Warnf("[etcd] %s client etcd retry: unmarshal data %s failed: %s, skip...", key, data, err)
-				return
-			}
-		}
-
-		set := utils.Set{}
-		for method, policy := range rcs {
-			set[method] = true
-			if policy.Enable && policy.BackupPolicy == nil && policy.FailurePolicy == nil {
-				klog.Warnf("[etcd] %s client policy for method %s BackupPolicy and FailurePolicy must not be empty at same time",
-					dest, method)
-				continue
-			}
-			retryContainer.NotifyPolicyChange(method, *policy)
-		}
-
-		for _, method := range ts.DiffAndEmplace(set) {
-			retryContainer.DeletePolicy(method)
-		}
-	}
-
-	etcdClient.RegisterConfigCallback(context.Background(), key, uniqueID, onChangeCallback)
-
-	return retryContainer
+	return cwclient.WithRetryPolicy(dest, src, etcdClient, uniqueID, opts)
 }

client/rpc_timeout.go

@@ -15,59 +15,14 @@
 package client
 
 import (
-	"context"
+	cwclient "github.com/cloudwego-contrib/cwgo-pkg/config/etcd/client"
 
 	"github.com/cloudwego/kitex/client"
-	"github.com/cloudwego/kitex/pkg/klog"
-	"github.com/cloudwego/kitex/pkg/rpcinfo"
-	"github.com/cloudwego/kitex/pkg/rpctimeout"
 	"github.com/kitex-contrib/config-etcd/etcd"
 	"github.com/kitex-contrib/config-etcd/utils"
 )
 
 // WithRPCTimeout sets the RPC timeout policy from etcd configuration center.
 func WithRPCTimeout(dest, src string, etcdClient etcd.Client, uniqueID int64, opts utils.Options) []client.Option {
-	param, err := etcdClient.ClientConfigParam(&etcd.ConfigParamConfig{
-		Category:          rpcTimeoutConfigName,
-		ServerServiceName: dest,
-		ClientServiceName: src,
-	})
-	if err != nil {
-		panic(err)
-	}
-
-	for _, f := range opts.EtcdCustomFunctions {
-		f(&param)
-	}
-	key := param.Prefix + "/" + param.Path
-	return []client.Option{
-		client.WithTimeoutProvider(initRPCTimeoutContainer(key, dest, etcdClient, uniqueID)),
-		client.WithCloseCallbacks(func() error {
-			// cancel the configuration listener when client is closed.
-			etcdClient.DeregisterConfig(key, uniqueID)
-			return nil
-		}),
-	}
-}
-
-func initRPCTimeoutContainer(key, dest string,
-	etcdClient etcd.Client, uniqueID int64,
-) rpcinfo.TimeoutProvider {
-	rpcTimeoutContainer := rpctimeout.NewContainer()
-
-	onChangeCallback := func(restoreDefault bool, data string, parser etcd.ConfigParser) {
-		configs := map[string]*rpctimeout.RPCTimeout{}
-		if !restoreDefault {
-			err := parser.Decode(data, &configs)
-			if err != nil {
-				klog.Warnf("[etcd] %s client etcd rpc timeout: unmarshal data %s failed: %s, skip...", key, data, err)
-				return
-			}
-		}
-		rpcTimeoutContainer.NotifyPolicyChange(configs)
-	}
-
-	etcdClient.RegisterConfigCallback(context.Background(), key, uniqueID, onChangeCallback)
-
-	return rpcTimeoutContainer
+	return cwclient.WithRPCTimeout(dest, src, etcdClient, uniqueID, opts)
 }

client/suite.go

@@ -15,50 +15,17 @@
 package client
 
 import (
-	"github.com/cloudwego/kitex/client"
+	cwclient "github.com/cloudwego-contrib/cwgo-pkg/config/etcd/client"
 	"github.com/kitex-contrib/config-etcd/etcd"
 	"github.com/kitex-contrib/config-etcd/utils"
 )
 
-const (
-	retryConfigName          = "retry"
-	rpcTimeoutConfigName     = "rpc_timeout"
-	circuitBreakerConfigName = "circuit_break"
-	degradationConfigName    = "degradation"
-)
-
 // EtcdClientSuite etcd client config suite, configure retry timeout limit and circuitbreak dynamically from etcd.
-type EtcdClientSuite struct {
-	uid        int64
-	etcdClient etcd.Client
-	service    string
-	client     string
-	opts       utils.Options
-}
+type EtcdClientSuite = cwclient.EtcdClientSuite
 
 // NewSuite service is the destination service name and client is the local identity.
 func NewSuite(service, client string, cli etcd.Client,
 	opts ...utils.Option,
 ) *EtcdClientSuite {
-	uid := etcd.AllocateUniqueID()
-	su := &EtcdClientSuite{
-		uid:        uid,
-		service:    service,
-		client:     client,
-		etcdClient: cli,
-	}
-	for _, opt := range opts {
-		opt.Apply(&su.opts)
-	}
-	return su
-}
-
-// Options return a list client.Option
-func (s *EtcdClientSuite) Options() []client.Option {
-	opts := make([]client.Option, 0, 7)
-	opts = append(opts, WithRetryPolicy(s.service, s.client, s.etcdClient, s.uid, s.opts)...)
-	opts = append(opts, WithRPCTimeout(s.service, s.client, s.etcdClient, s.uid, s.opts)...)
-	opts = append(opts, WithCircuitBreaker(s.service, s.client, s.etcdClient, s.uid, s.opts)...)
-	opts = append(opts, WithDegradation(s.service, s.client, s.etcdClient, s.uid, s.opts)...)
-	return opts
+	return cwclient.NewSuite(service, client, cli, opts...)
 }