Feat(circuit_breaker): support circuit_breaker policy by nacos config

client/circuit_breaker.go

@@ -0,0 +1,105 @@
+// Copyright 2023 CloudWeGo Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package client
+
+import (
+	"strings"
+
+	"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/nacos-group/nacos-sdk-go/vo"
+
+	"github.com/kitex-contrib/config-nacos/nacos"
+	"github.com/kitex-contrib/config-nacos/utils"
+)
+
+// WithCircuitBreaker sets the circuit breaker policy from nacos configuration center.
+func WithCircuitBreaker(dest, src string, nacosClient nacos.Client,
+	cfs ...nacos.CustomFunction,
+) []client.Option {
+	param := nacos.NacosConfigParam(&nacos.ConfigParamConfig{
+		Category:          circuitBreakerConfigName,
+		ServerServiceName: dest,
+		ClientServiceName: src,
+	}, cfs...)
+
+	cbSuite := initCircuitBreaker(param, dest, src, nacosClient)
+
+	return []client.Option{
+		// the client identity is necessary when generate the key for service circuit breaker.
+		client.WithClientBasicInfo(&rpcinfo.EndpointBasicInfo{
+			ServiceName: src,
+		}),
+		client.WithCircuitBreaker(cbSuite),
+		client.WithCloseCallbacks(func() error {
+			err := cbSuite.Close()
+			if err != nil {
+				return err
+			}
+			// cancel the configuration listener when client is closed.
+			return nacosClient.DeregisterConfig(param)
+		}),
+	}
+}
+
+// Be consistent with the RPCInfo2Key function in kitex/pkg/circuitbreak/cbsuite.go.
+// NOTE The fromService should keep consistent with the ServiceName field in rpcinfo.EndpointBasicInfo
+// which can be set using client.WithClientBasicInfo function, should
+// set it correctly.
+func genServiceCBKey(fromService, toService, method string) string {
+	sum := len(fromService) + len(toService) + len(method) + 2
+	var buf strings.Builder
+	buf.Grow(sum)
+	buf.WriteString(fromService)
+	buf.WriteByte('/')
+	buf.WriteString(toService)
+	buf.WriteByte('/')
+	buf.WriteString(method)
+	return buf.String()
+}
+
+func initCircuitBreaker(param vo.ConfigParam, dest, src string,
+	nacosClient nacos.Client,
+) *circuitbreak.CBSuite {
+	cb := circuitbreak.NewCBSuite(nil)
+	lcb := utils.ThreadSafeSet{}
+
+	onChangeCallback := func(data string, parser nacos.ConfigParser) {
+		set := utils.Set{}
+		configs := map[string]circuitbreak.CBConfig{}
+		err := parser.Decode(param.Type, data, &configs)
+		if err != nil {
+			klog.Warnf("[nacos] %s client nacos rpc timeout: unmarshal data %s failed: %s, skip...", dest, data, err)
+			return
+		}
+
+		for method, config := range configs {
+			set[method] = true
+			key := genServiceCBKey(src, dest, method)
+			cb.UpdateServiceCBConfig(key, config)
+		}
+
+		for _, method := range lcb.DiffAndEmplace(set) {
+			key := genServiceCBKey(src, dest, method)
+			cb.UpdateServiceCBConfig(key, circuitbreak.GetDefaultCBConfig())
+		}
+	}
+
+	nacosClient.RegisterConfigCallback(param, onChangeCallback)
+
+	return cb
+}

client/client_suite.go

@@ -16,13 +16,13 @@ package client
 
 import (
 	"github.com/cloudwego/kitex/client"
-
 	"github.com/kitex-contrib/config-nacos/nacos"
 )
 
 const (
-	retryConfigName      = "retry_config"
-	rpcTimeoutConfigName = "rpc_timeout"
+	retryConfigName          = "retry_config"
+	rpcTimeoutConfigName     = "rpc_timeout"
+	circuitBreakerConfigName = "cb_config"
 )
 
 // NacosClientSuite nacos client config suite, configure retry timeout limit and circuitbreak dynamically from nacos.
@@ -47,8 +47,9 @@ func NewSuite(service, client string, cli nacos.Client,
 
 // Options return a list client.Option
 func (s *NacosClientSuite) Options() []client.Option {
-	opts := make([]client.Option, 0, 8)
+	opts := make([]client.Option, 0, 7)
 	opts = append(opts, WithRetryPolicy(s.service, s.client, s.nacosClient, s.fns...)...)
 	opts = append(opts, WithRPCTimeout(s.service, s.client, s.nacosClient, s.fns...)...)
+	opts = append(opts, WithCircuitBreaker(s.service, s.client, s.nacosClient, s.fns...)...)
 	return opts
 }

client/retry.go

@@ -16,6 +16,7 @@ package client
 
 import (
 	"github.com/kitex-contrib/config-nacos/nacos"
+	"github.com/kitex-contrib/config-nacos/utils"
 	"github.com/nacos-group/nacos-sdk-go/vo"
 
 	"github.com/cloudwego/kitex/client"
@@ -42,23 +43,25 @@ func WithRetryPolicy(dest, src string, nacosClient nacos.Client,
 	}
 }
 
-// the key is method name, wildcard "*" can match anything.
-type retryConfigs map[string]*retry.Policy
-
 func initRetryContainer(param vo.ConfigParam, dest string,
 	nacosClient nacos.Client,
 ) *retry.Container {
 	retryContainer := retry.NewRetryContainer()
 
+	ts := utils.ThreadSafeSet{}
+
 	onChangeCallback := func(data string, parser nacos.ConfigParser) {
-		rcs := retryConfigs{}
-		err := parser.Decode(param.Type, data, rcs)
+		// the key is method name, wildcard "*" can match anything.
+		rcs := map[string]*retry.Policy{}
+		err := parser.Decode(param.Type, data, &rcs)
 		if err != nil {
 			klog.Warnf("[nacos] %s client nacos retry: unmarshal data %s failed: %s, skip...", dest, data, err)
 			return
 		}
 
+		set := utils.Set{}
 		for method, policy := range rcs {
+			set[method] = true
 			if policy.BackupPolicy != nil && policy.FailurePolicy != nil {
 				klog.Warnf("[nacos] %s client policy for method %s BackupPolicy and FailurePolicy must not be set at same time",
 					dest, method)
@@ -71,13 +74,13 @@ func initRetryContainer(param vo.ConfigParam, dest string,
 			}
 			retryContainer.NotifyPolicyChange(method, *policy)
 		}
+
+		for _, method := range ts.DiffAndEmplace(set) {
+			retryContainer.DeletePolicy(method)
+		}
 	}
 
-	nacosClient.RegisterConfigCallback(dest,
-		retryConfigName,
-		param,
-		onChangeCallback,
-	)
+	nacosClient.RegisterConfigCallback(param, onChangeCallback)
 
 	return retryContainer
 }

client/rpc_timeout.go

@@ -57,11 +57,7 @@ func initRPCTimeoutContainer(param vo.ConfigParam, dest string,
 		rpcTimeoutContainer.NotifyPolicyChange(configs)
 	}
 
-	nacosClient.RegisterConfigCallback(dest,
-		rpcTimeoutConfigName,
-		param,
-		onChangeCallback,
-	)
+	nacosClient.RegisterConfigCallback(param, onChangeCallback)
 
 	return rpcTimeoutContainer
 }

go.mod

@@ -3,7 +3,7 @@ module github.com/kitex-contrib/config-nacos
 go 1.19
 
 require (
-	github.com/cloudwego/kitex v0.6.2-0.20230725114715-4865773820d5
+	github.com/cloudwego/kitex v0.6.2-0.20230804065152-067aad9c9eef
 	github.com/cloudwego/kitex-examples v0.2.0
 	github.com/kitex-contrib/registry-nacos v0.1.0
 	github.com/nacos-group/nacos-sdk-go v1.1.4
@@ -15,7 +15,7 @@ require (
 	github.com/aliyun/alibaba-cloud-sdk-go v1.61.18 // indirect
 	github.com/apache/thrift v0.13.0 // indirect
 	github.com/buger/jsonparser v1.1.1 // indirect
-	github.com/bytedance/gopkg v0.0.0-20230531144706-a12972768317 // indirect
+	github.com/bytedance/gopkg v0.0.0-20230728082804-614d0af6619b // indirect
 	github.com/bytedance/sonic v1.9.1 // indirect
 	github.com/chenzhuoyu/base64x v0.0.0-20221115062448-fe3a3abad311 // indirect
 	github.com/chenzhuoyu/iasm v0.9.0 // indirect
@@ -24,8 +24,9 @@ require (
 	github.com/cloudwego/dynamicgo v0.1.2 // indirect
 	github.com/cloudwego/fastpb v0.0.4 // indirect
 	github.com/cloudwego/frugal v0.1.7 // indirect
+	github.com/cloudwego/localsession v0.0.2 // indirect
 	github.com/cloudwego/netpoll v0.4.1 // indirect
-	github.com/cloudwego/thriftgo v0.2.11 // indirect
+	github.com/cloudwego/thriftgo v0.2.13-0.20230804030519-2ce040e1b8ba // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
 	github.com/fatih/structtag v1.2.0 // indirect
 	github.com/go-errors/errors v1.0.1 // indirect
@@ -37,6 +38,7 @@ require (
 	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/klauspost/cpuid/v2 v2.2.4 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
+	github.com/modern-go/gls v0.0.0-20220109145502-612d0167dce5 // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/oleiade/lane v1.0.1 // indirect
 	github.com/pkg/errors v0.9.1 // indirect

go.sum

@@ -427,8 +427,9 @@ github.com/bytedance/gopkg v0.0.0-20220413063733-65bf48ffb3a7/go.mod h1:2ZlV9BaU
 github.com/bytedance/gopkg v0.0.0-20220509134931-d1878f638986/go.mod h1:2ZlV9BaUH4+NXIBF0aMdKKAnHTzqH+iMU4KUjAbL23Q=
 github.com/bytedance/gopkg v0.0.0-20220531084716-665b4f21126f/go.mod h1:2ZlV9BaUH4+NXIBF0aMdKKAnHTzqH+iMU4KUjAbL23Q=
 github.com/bytedance/gopkg v0.0.0-20220817015305-b879a72dc90f/go.mod h1:2ZlV9BaUH4+NXIBF0aMdKKAnHTzqH+iMU4KUjAbL23Q=
-github.com/bytedance/gopkg v0.0.0-20230531144706-a12972768317 h1:SReMVmTCeJ5Nf0hU8nyWu7gAaFVD8mu5yvSH/+uLT1E=
 github.com/bytedance/gopkg v0.0.0-20230531144706-a12972768317/go.mod h1:FtQG3YbQG9L/91pbKSw787yBQPutC+457AvDW77fgUQ=
+github.com/bytedance/gopkg v0.0.0-20230728082804-614d0af6619b h1:R6PWoQtxEMpWJPHnpci+9LgFxCS7iJCfOGBvCgZeTKI=
+github.com/bytedance/gopkg v0.0.0-20230728082804-614d0af6619b/go.mod h1:FtQG3YbQG9L/91pbKSw787yBQPutC+457AvDW77fgUQ=
 github.com/bytedance/mockey v1.2.0 h1:847+X2fBSM4s/AIN4loO5d16PCgEj53j7Q8YVB+8P6c=
 github.com/bytedance/mockey v1.2.0/go.mod h1:+Jm/fzWZAuhEDrPXVjDf/jLM2BlLXJkwk94zf2JZ3X4=
 github.com/bytedance/sonic v1.5.0/go.mod h1:ED5hyg4y6t3/9Ku1R6dU/4KyJ48DZ4jPhfY1O2AihPM=
@@ -483,10 +484,12 @@ github.com/cloudwego/kitex v0.4.4/go.mod h1:3FcH5h9Qw+dhRljSzuGSpWuThttA8DvK0BsL
 github.com/cloudwego/kitex v0.5.0/go.mod h1:yhw7XikNVG4RstmlQAidBuxMlZYpIiCLsDU8eHPGEMo=
 github.com/cloudwego/kitex v0.5.1/go.mod h1:B3oH+MTQ7/wBL3BrCAMlyeyjAqOpi4pRzCvQcXN7RgM=
 github.com/cloudwego/kitex v0.6.1/go.mod h1:zI1GBrjT0qloTikcCfQTgxg3Ws+yQMyaChEEOcGNUvA=
-github.com/cloudwego/kitex v0.6.2-0.20230725114715-4865773820d5 h1:71ynYIunvwxFUAYhpq2REuDSkp6F1LG7GOO9ULXjsF8=
-github.com/cloudwego/kitex v0.6.2-0.20230725114715-4865773820d5/go.mod h1:5DimETWVNf0xmMtlEpyD00HuOfiSDw5spU/e5MvN6CQ=
+github.com/cloudwego/kitex v0.6.2-0.20230804065152-067aad9c9eef h1:TijrLYx5RqH/Q9AXl2pd24+3/ID5jWNvQNJfbphF8Ic=
+github.com/cloudwego/kitex v0.6.2-0.20230804065152-067aad9c9eef/go.mod h1:juSMg+Ae9uvch5bET2IdjJpOurx+IPjWQSKLrZyqjCY=
 github.com/cloudwego/kitex-examples v0.2.0 h1:77PXV+5q5OCKE38q2DVvHO7ePemtNapwnX+n6zaVSjo=
 github.com/cloudwego/kitex-examples v0.2.0/go.mod h1:jmqlyPgALXoYXiZwWJSpzb8wEycu9PIIyRRs2cN9OJs=
+github.com/cloudwego/localsession v0.0.2 h1:N9/IDtCPj1fCL9bCTP+DbXx3f40YjVYWcwkJG0YhQkY=
+github.com/cloudwego/localsession v0.0.2/go.mod h1:kiJxmvAcy4PLgKtEnPS5AXed3xCiXcs7Z+KBHP72Wv8=
 github.com/cloudwego/netpoll v0.0.2/go.mod h1:rZOiNI0FYjuvNybXKKhAPUja03loJi/cdv2F55AE6E8=
 github.com/cloudwego/netpoll v0.0.3/go.mod h1:rZOiNI0FYjuvNybXKKhAPUja03loJi/cdv2F55AE6E8=
 github.com/cloudwego/netpoll v0.1.0/go.mod h1:rZOiNI0FYjuvNybXKKhAPUja03loJi/cdv2F55AE6E8=
@@ -503,8 +506,9 @@ github.com/cloudwego/thriftgo v0.1.2/go.mod h1:LzeafuLSiHA9JTiWC8TIMIq64iadeObgR
 github.com/cloudwego/thriftgo v0.2.4/go.mod h1:8i9AF5uDdWHGqzUhXDlubCjx4MEfKvWXGQlMWyR0tM4=
 github.com/cloudwego/thriftgo v0.2.7/go.mod h1:8i9AF5uDdWHGqzUhXDlubCjx4MEfKvWXGQlMWyR0tM4=
 github.com/cloudwego/thriftgo v0.2.8/go.mod h1:dAyXHEmKXo0LfMCrblVEY3mUZsdeuA5+i0vF5f09j7E=
-github.com/cloudwego/thriftgo v0.2.11 h1:uwFyTMBwmBJKpwxRdBvn46aHEVJJSgxkHo93RN0r3fw=
 github.com/cloudwego/thriftgo v0.2.11/go.mod h1:dAyXHEmKXo0LfMCrblVEY3mUZsdeuA5+i0vF5f09j7E=
+github.com/cloudwego/thriftgo v0.2.13-0.20230804030519-2ce040e1b8ba h1:K+9Hjh6XZvvMN/yCnyErVIMtXoKMS249scXMXVa8qD8=
+github.com/cloudwego/thriftgo v0.2.13-0.20230804030519-2ce040e1b8ba/go.mod h1:AvH0iEjvKHu3cdxG7JvhSAaffkS4h2f4/ZxpJbm48W4=
 github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGXZJjfX53e64911xZQV5JYwmTeXPW+k8Sc=
 github.com/cncf/udpa/go v0.0.0-20200629203442-efcf912fb354/go.mod h1:WmhPx2Nbnhtbo57+VJT5O0JRkEi1Wbu0z5j0R8u5Hbk=
 github.com/cncf/udpa/go v0.0.0-20201120205902-5459f2c99403/go.mod h1:WmhPx2Nbnhtbo57+VJT5O0JRkEi1Wbu0z5j0R8u5Hbk=
@@ -726,6 +730,8 @@ github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5
 github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
 github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg=
 github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
+github.com/modern-go/gls v0.0.0-20220109145502-612d0167dce5 h1:uiS4zKYKJVj5F3ID+5iylfKPsEQmBEOucSD9Vgmn0i0=
+github.com/modern-go/gls v0.0.0-20220109145502-612d0167dce5/go.mod h1:I8AX+yW//L8Hshx6+a1m3bYkwXkpsVjA2795vP4f4oQ=
 github.com/modern-go/reflect2 v0.0.0-20180701023420-4b7aa43c6742/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=
 github.com/modern-go/reflect2 v1.0.1/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=
 github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9Gz0M=

nacos/nacos.go

@@ -25,7 +25,7 @@ import (
 // Client the wrapper of nacos client.
 type Client interface {
 	SetParser(ConfigParser)
-	RegisterConfigCallback(string, string, vo.ConfigParam, func(string, ConfigParser))
+	RegisterConfigCallback(vo.ConfigParam, func(string, ConfigParser))
 	DeregisterConfig(vo.ConfigParam) error
 }
 
@@ -75,13 +75,12 @@ func (c *client) DeregisterConfig(cfg vo.ConfigParam) error {
 }
 
 // RegisterConfigCallback register the callback function to nacos client.
-func (c *client) RegisterConfigCallback(dest, category string,
-	param vo.ConfigParam,
+func (c *client) RegisterConfigCallback(param vo.ConfigParam,
 	callback func(string, ConfigParser),
 ) {
 	param.OnChange = func(namespace, group, dataId, data string) {
-		klog.Debugf("[nacos] %s client %s config updated, namespace %s group %s dataId %s data %s",
-			dest, category, namespace, group, dataId, data)
+		klog.Debugf("[nacos] config %s updated, namespace %s group %s dataId %s data %s",
+			param.DataId, namespace, group, dataId, data)
 		callback(data, c.parser)
 	}
 	data, err := c.ncli.GetConfig(param)

utils/set.go

@@ -0,0 +1,46 @@
+// Copyright 2023 CloudWeGo Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package utils
+
+import "sync"
+
+// ThreadSafeSet wrapper of Set.
+type ThreadSafeSet struct {
+	sync.RWMutex
+	s Set
+}
+
+// DiffAndEmplace returns the keys that are not in other and emplace the old set.
+func (ts *ThreadSafeSet) DiffAndEmplace(other Set) []string {
+	ts.Lock()
+	defer ts.Unlock()
+	out := ts.s.Diff(other)
+	ts.s = other
+	return out
+}
+
+// Set map template.
+type Set map[string]bool
+
+// Diff returns the keys that are not in other
+func (s Set) Diff(other Set) []string {
+	out := make([]string, 0, len(s))
+	for key := range s {
+		if _, ok := other[key]; !ok {
+			out = append(out, key)
+		}
+	}
+	return out
+}