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

.github/workflows/tests.yml

@@ -6,7 +6,7 @@ jobs:
   unit-benchmark-test:
     strategy:
       matrix:
-        go: [ 1.17, 1.18, 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:
@@ -34,3 +30,7 @@ linters:
     - staticcheck
 issues:
   exclude-use-default: true
+  exclude-files:
+    - ".*\\.mock\\.go$"
+  exclude-dirs:
+    - kitex_gen

client/circuit_breaker.go

@@ -15,100 +15,13 @@
 package client
 
 import (
-	"context"
-	"strings"
-
+	cwclient "github.com/cloudwego-contrib/cwgo-pkg/config/zookeeper/client"
 	"github.com/cloudwego/kitex/client"
-	"github.com/cloudwego/kitex/pkg/circuitbreak"
-	"github.com/cloudwego/kitex/pkg/klog"
-	"github.com/cloudwego/kitex/pkg/rpcinfo"
 	"github.com/kitex-contrib/config-zookeeper/utils"
 	"github.com/kitex-contrib/config-zookeeper/zookeeper"
 )
 
 // WithCircuitBreaker sets the circuit breaker policy from zookeeper configuration center.
 func WithCircuitBreaker(dest, src string, zookeeperClient zookeeper.Client, opts utils.Options) []client.Option {
-	param, err := zookeeperClient.ClientConfigParam(&zookeeper.ConfigParamConfig{
-		Category:          circuitBreakerConfigName,
-		ServerServiceName: dest,
-		ClientServiceName: src,
-	})
-	if err != nil {
-		panic(err)
-	}
-
-	for _, f := range opts.ZookeeperCustomFunctions {
-		f(&param)
-	}
-
-	uid := zookeeper.GetUniqueID()
-	path := param.Prefix + "/" + param.Path
-
-	cbSuite := initCircuitBreaker(path, uid, dest, zookeeperClient)
-
-	return []client.Option{
-		client.WithCircuitBreaker(cbSuite),
-		client.WithCloseCallbacks(func() error {
-			// cancel the configuration listener when client is closed.
-			zookeeperClient.DeregisterConfig(path, uid)
-			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(path string, uniqueID int64, dest string, zookeeperClient zookeeper.Client) *circuitbreak.CBSuite {
-	cb := circuitbreak.NewCBSuite(genServiceCBKeyWithRPCInfo)
-	lcb := utils.ThreadSafeSet{}
-
-	onChangeCallback := func(restoreDefault bool, data string, parser zookeeper.ConfigParser) {
-		set := utils.Set{}
-		configs := map[string]circuitbreak.CBConfig{}
-
-		if !restoreDefault {
-			err := parser.Decode(data, &configs)
-			if err != nil {
-				klog.Warnf("[zookeeper] %s client zookeeper circuit breaker: unmarshal data %s failed: %s, skip...", path, 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())
-		}
-	}
-
-	zookeeperClient.RegisterConfigCallback(context.Background(), path, uniqueID, onChangeCallback)
-
-	return cb
+	return cwclient.WithCircuitBreaker(dest, src, zookeeperClient, opts)
 }

client/degradation.go

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

client/retry.go

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

client/rpc_timeout.go

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

client/suite.go

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