Add support for Nacos v2 SDK

v2/client/circuit_breaker.go

@@ -0,0 +1,111 @@
+// 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 (
+	"github.com/kitex-contrib/config-nacos/v2/nacos"
+	"github.com/kitex-contrib/config-nacos/v2/utils"
+	"github.com/nacos-group/nacos-sdk-go/v2/vo"
+	"strings"
+
+	"github.com/cloudwego/kitex/client"
+	"github.com/cloudwego/kitex/pkg/circuitbreak"
+	"github.com/cloudwego/kitex/pkg/klog"
+	"github.com/cloudwego/kitex/pkg/rpcinfo"
+)
+
+// WithCircuitBreaker sets the circuit breaker policy from nacos configuration center.
+func WithCircuitBreaker(dest, src string, nacosClient nacos.Client, opts utils.Options) []client.Option {
+	param, err := nacosClient.ClientConfigParam(&nacos.ConfigParamConfig{
+		Category:          circuitBreakerConfigName,
+		ServerServiceName: dest,
+		ClientServiceName: src,
+	})
+	if err != nil {
+		panic(err)
+	}
+
+	for _, f := range opts.NacosCustomFunctions {
+		f(&param)
+	}
+
+	uniqueID := nacos.GetUniqueID()
+
+	cbSuite := initCircuitBreaker(param, dest, src, nacosClient, uniqueID)
+
+	return []client.Option{
+		client.WithCircuitBreaker(cbSuite),
+		client.WithCloseCallbacks(func() error {
+			err := nacosClient.DeregisterConfig(param, uniqueID)
+			if err != nil {
+				return err
+			}
+			// cancel the configuration listener when client is closed.
+			return cbSuite.Close()
+		}),
+	}
+}
+
+// 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(param vo.ConfigParam, dest, src string,
+	nacosClient nacos.Client, uniqueID int64,
+) *circuitbreak.CBSuite {
+	cb := circuitbreak.NewCBSuite(genServiceCBKeyWithRPCInfo)
+	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 circuit breaker: unmarshal data %s failed: %s, skip...", dest, 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())
+		}
+	}
+
+	nacosClient.RegisterConfigCallback(param, onChangeCallback, uniqueID)
+
+	return cb
+}

v2/client/retry.go

@@ -0,0 +1,83 @@
+package client
+
+import (
+	"github.com/cloudwego/kitex/client"
+	"github.com/cloudwego/kitex/pkg/klog"
+	"github.com/cloudwego/kitex/pkg/retry"
+	"github.com/kitex-contrib/config-nacos/v2/nacos"
+	"github.com/kitex-contrib/config-nacos/v2/utils"
+	"github.com/nacos-group/nacos-sdk-go/v2/vo"
+)
+
+// WithRetryPolicy sets the retry policy from nacos configuration center.
+func WithRetryPolicy(dest, src string, nacosClient nacos.Client, opts utils.Options) []client.Option {
+	param, err := nacosClient.ClientConfigParam(&nacos.ConfigParamConfig{
+		Category:          retryConfigName,
+		ServerServiceName: dest,
+		ClientServiceName: src,
+	})
+	if err != nil {
+		panic(err)
+	}
+
+	for _, f := range opts.NacosCustomFunctions {
+		f(&param)
+	}
+
+	uniqueID := nacos.GetUniqueID()
+
+	rc := initRetryContainer(param, dest, nacosClient, uniqueID)
+	return []client.Option{
+		client.WithRetryContainer(rc),
+		client.WithCloseCallbacks(func() error {
+			// cancel the configuration listener when client is closed.
+			err := nacosClient.DeregisterConfig(param, uniqueID)
+			if err != nil {
+				return err
+			}
+			return rc.Close()
+		}),
+	}
+}
+
+func initRetryContainer(param vo.ConfigParam, dest string,
+	nacosClient nacos.Client, uniqueID int64,
+) *retry.Container {
+	retryContainer := retry.NewRetryContainerWithPercentageLimit()
+
+	ts := utils.ThreadSafeSet{}
+
+	onChangeCallback := func(data string, parser nacos.ConfigParser) {
+		// 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)
+				continue
+			}
+			if policy.BackupPolicy == nil && policy.FailurePolicy == nil {
+				klog.Warnf("[nacos] %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)
+		}
+	}
+
+	nacosClient.RegisterConfigCallback(param, onChangeCallback, uniqueID)
+
+	return retryContainer
+}

v2/client/rpc_timeout.go

@@ -0,0 +1,71 @@
+// 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 (
+	"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-nacos/v2/nacos"
+	"github.com/kitex-contrib/config-nacos/v2/utils"
+	"github.com/nacos-group/nacos-sdk-go/v2/vo"
+)
+
+// WithRPCTimeout sets the RPC timeout policy from nacos configuration center.
+func WithRPCTimeout(dest, src string, nacosClient nacos.Client, opts utils.Options) []client.Option {
+	param, err := nacosClient.ClientConfigParam(&nacos.ConfigParamConfig{
+		Category:          rpcTimeoutConfigName,
+		ServerServiceName: dest,
+		ClientServiceName: src,
+	})
+	if err != nil {
+		panic(err)
+	}
+
+	for _, f := range opts.NacosCustomFunctions {
+		f(&param)
+	}
+
+	uniqueID := nacos.GetUniqueID()
+
+	return []client.Option{
+		client.WithTimeoutProvider(initRPCTimeoutContainer(param, dest, nacosClient, uniqueID)),
+		client.WithCloseCallbacks(func() error {
+			// cancel the configuration listener when client is closed.
+			return nacosClient.DeregisterConfig(param, uniqueID)
+		}),
+	}
+}
+
+func initRPCTimeoutContainer(param vo.ConfigParam, dest string,
+	nacosClient nacos.Client, uniqueID int64,
+) rpcinfo.TimeoutProvider {
+	rpcTimeoutContainer := rpctimeout.NewContainer()
+
+	onChangeCallback := func(data string, parser nacos.ConfigParser) {
+		configs := map[string]*rpctimeout.RPCTimeout{}
+		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
+		}
+		rpcTimeoutContainer.NotifyPolicyChange(configs)
+	}
+
+	nacosClient.RegisterConfigCallback(param, onChangeCallback, uniqueID)
+
+	return rpcTimeoutContainer
+}

v2/client/suite.go

@@ -0,0 +1,57 @@
+// 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 (
+	"github.com/cloudwego/kitex/client"
+	"github.com/kitex-contrib/config-nacos/v2/nacos"
+	"github.com/kitex-contrib/config-nacos/v2/utils"
+)
+
+const (
+	retryConfigName          = "retry"
+	rpcTimeoutConfigName     = "rpc_timeout"
+	circuitBreakerConfigName = "circuit_break"
+)
+
+// NacosClientSuite nacos client config suite, configure retry timeout limit and circuitbreak dynamically from nacos.
+type NacosClientSuite struct {
+	nacosClient nacos.Client
+	service     string
+	client      string
+	opts        utils.Options
+}
+
+// NewSuite service is the destination service name and client is the local identity.
+func NewSuite(service, client string, cli nacos.Client, opts ...utils.Option) *NacosClientSuite {
+	su := &NacosClientSuite{
+		service:     service,
+		client:      client,
+		nacosClient: cli,
+	}
+	for _, f := range opts {
+		f.Apply(&su.opts)
+	}
+	return su
+}
+
+// Options return a list client.Option
+func (s *NacosClientSuite) Options() []client.Option {
+	opts := make([]client.Option, 0, 7)
+	opts = append(opts, WithRetryPolicy(s.service, s.client, s.nacosClient, s.opts)...)
+	opts = append(opts, WithRPCTimeout(s.service, s.client, s.nacosClient, s.opts)...)
+	opts = append(opts, WithCircuitBreaker(s.service, s.client, s.nacosClient, s.opts)...)
+	return opts
+}

v2/example/client/main.go

@@ -0,0 +1,72 @@
+// 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 main
+
+import (
+	"context"
+	"fmt"
+	"github.com/cloudwego/kitex-examples/kitex_gen/api"
+	"github.com/cloudwego/kitex-examples/kitex_gen/api/echo"
+	"github.com/cloudwego/kitex/client"
+	"github.com/cloudwego/kitex/pkg/klog"
+	nacosclient "github.com/kitex-contrib/config-nacos/v2/client"
+	"github.com/kitex-contrib/config-nacos/v2/nacos"
+	"github.com/kitex-contrib/config-nacos/v2/utils"
+	"github.com/nacos-group/nacos-sdk-go/v2/vo"
+	"log"
+	"time"
+)
+
+type configLog struct{}
+
+func (cl *configLog) Apply(opt *utils.Options) {
+	fn := func(cp *vo.ConfigParam) {
+		klog.Infof("nacos config %v", cp)
+	}
+	opt.NacosCustomFunctions = append(opt.NacosCustomFunctions, fn)
+}
+
+func main() {
+	nacosClient, err := nacos.NewClient(nacos.Options{})
+	if err != nil {
+		panic(err)
+	}
+	fmt.Println(nacosClient)
+
+	cl := &configLog{}
+
+	serviceName := "ServiceName" // your server-side service name
+	clientName := "ClientName"   // your client-side service name
+	client, err := echo.NewClient(
+		serviceName,
+		client.WithHostPorts("0.0.0.0:8888"),
+		client.WithSuite(nacosclient.NewSuite(serviceName, clientName, nacosClient, cl)),
+	)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	for {
+		req := &api.Request{Message: "my request"}
+		resp, err := client.Echo(context.Background(), req)
+		if err != nil {
+			klog.Errorf("take request error: %v", err)
+		} else {
+			klog.Infof("receive response %v", resp)
+		}
+		time.Sleep(time.Second * 10)
+	}
+}