Merge branch 'etcd'

etcd/client/client.go

@@ -1,8 +1,7 @@
-package client
+package etcdclient
 
 import (
 	clientv3 "go.etcd.io/etcd/client/v3"
-	"go.uber.org/zap"
 	"text/template"
 	"time"
 )
@@ -13,7 +12,6 @@ type ReaderOptions struct {
 	Prefix       string
 	PathFormat   string
 	Timeout      time.Duration
-	LoggerConfig *zap.Config
 	ConfigParser ConfigParser
 	MyConfig     Config
 }
@@ -35,8 +33,8 @@ func NewReader(opts ReaderOptions) (*EtcdReader, error) {
 		opts.PathFormat = EtcdClientDefaultPath
 	}
 	etcdClient, err := clientv3.New(clientv3.Config{
-		Endpoints: opts.Node,
-		LogConfig: opts.LoggerConfig,
+		Endpoints:   opts.Node,
+		DialTimeout: opts.Timeout,
 	})
 	if err != nil {
 		return nil, err

etcd/client/decoder.go

@@ -1,20 +1,20 @@
-package client
+package etcdclient
 
 import (
 	"encoding/json"
 	"fmt"
-	"time"
+	"strings"
 )
 
 type ConfigParser interface {
-	Decode(data []byte, config EtcdConfig) error
+	Decode(data []byte, config *EtcdConfig) error
 }
 
 type defaultParser struct {
 }
 
-func (p *defaultParser) Decode(data []byte, config EtcdConfig) error {
-	return json.Unmarshal(data, &config)
+func (p *defaultParser) Decode(data []byte, config *EtcdConfig) error {
+	return json.Unmarshal(data, config)
 }
 
 type Config interface {
@@ -23,25 +23,41 @@ type Config interface {
 
 type EtcdConfig struct {
 	ClientBasicInfo *EndpointBasicInfo `mapstructure:"ClientBasicInfo"`
-	HostPorts       *string            `mapstructure:"HostPorts"`
+	HostPorts       []string           `mapstructure:"HostPorts"`
 	DestService     *string            `mapstructure:"DestService"`
 	Protocol        *string            `mapstructure:"Protocol"`
 	Connection      *Connection        `mapstructure:"Connection"`
 	MyConfig        Config             `mapstructure:"MyConfig"`
 }
 
 func (c *EtcdConfig) String() string {
-	baseInfo := "nil"
+	var builder strings.Builder
+
+	if c.ClientBasicInfo != nil {
+		builder.WriteString(fmt.Sprintf("ClientBasicInfo: %v\n", *c.ClientBasicInfo))
+	}
+
+	if c.HostPorts != nil {
+		builder.WriteString(fmt.Sprintf("HostPorts: %v\n", c.HostPorts))
+	}
+
+	if c.DestService != nil {
+		builder.WriteString(fmt.Sprintf("DestService: %v\n", *c.DestService))
+	}
+
+	if c.Protocol != nil {
+		builder.WriteString(fmt.Sprintf("Protocol: %v\n", *c.Protocol))
+	}
+
+	if c.Connection != nil {
+		builder.WriteString(fmt.Sprintf("Connection: %v\n", *c.Connection))
+	}
+
 	if c.MyConfig != nil {
-		baseInfo = c.MyConfig.String()
+		builder.WriteString(c.MyConfig.String())
 	}
-	return fmt.Sprintf("ClientBasicInfo: %v\n"+
-		" HostPorts: %s\n"+
-		" DestService: %s\n"+
-		" Protocol: %s\n"+
-		" Connection: %v\n"+
-		" MyConfig: %s\n",
-		*c.ClientBasicInfo, *c.HostPorts, *c.DestService, *c.Protocol, *c.Connection, baseInfo)
+
+	return builder.String()
 }
 
 type EndpointBasicInfo struct {
@@ -51,13 +67,13 @@ type EndpointBasicInfo struct {
 }
 
 type IdleConfig struct {
-	MinIdlePerAddress int           `mapstructure:"MinIdlePerAddress"`
-	MaxIdlePerAddress int           `mapstructure:"MaxIdlePerAddress"`
-	MaxIdleGlobal     int           `mapstructure:"MaxIdleGlobal"`
-	MaxIdleTimeout    time.Duration `mapstructure:"MaxIdleTimeout"`
+	MinIdlePerAddress int    `mapstructure:"MinIdlePerAddress"`
+	MaxIdlePerAddress int    `mapstructure:"MaxIdlePerAddress"`
+	MaxIdleGlobal     int    `mapstructure:"MaxIdleGlobal"`
+	MaxIdleTimeout    string `mapstructure:"MaxIdleTimeout"`
 }
 type MuxConnection struct {
-	ConnNum int `json:"connNum"`
+	ConnNum int `mapstructure:"ConnNum"`
 }
 
 type Connection struct {

etcd/client/loader.go

@@ -1,4 +1,4 @@
-package client
+package etcdclient
 
 import (
 	kitexclient "github.com/cloudwego/kitex/client"
@@ -22,6 +22,11 @@ type EtcdLoader struct {
 }
 
 func (l *EtcdLoader) Load() error {
+	path := Path{ClientServiceName: l.ClientServiceName, ServerServiceName: l.ServerServiceName}
+	err := l.reader.ReadToConfig(&path)
+	if err != nil {
+		return err
+	}
 	config, err := l.reader.GetConfig()
 	if err != nil {
 		return err

etcd/client/reader.go

@@ -1,4 +1,4 @@
-package client
+package etcdclient
 
 import (
 	"bytes"
@@ -13,7 +13,7 @@ const (
 	EtcdDefaultNode         = "http://127.0.0.1:2379"
 	EtcdDefaultConfigPrefix = "/KitexConfig"
 	EtcdDefaultTimeout      = 5 * time.Second
-	EtcdClientDefaultPath   = "{{.ClientServiceName}}/{{.ServerServiceName}}"
+	EtcdClientDefaultPath   = "/{{.ClientServiceName}}/{{.ServerServiceName}}"
 )
 
 type Reader interface {
@@ -55,7 +55,7 @@ func (r *EtcdReader) ReadToConfig(p *Path) error {
 		klog.Debugf("[etcd] key: %s config get value failed", key)
 		return err
 	}
-	err = r.parser.Decode(data.Kvs[0].Value, *r.config)
+	err = r.parser.Decode(data.Kvs[0].Value, r.config)
 	if err != nil {
 		return err
 	}

etcd/client/suite.go

@@ -1,4 +1,4 @@
-package client
+package etcdclient
 
 import (
 	"github.com/cloudwego/kitex/client"

etcd/client/translator.go

@@ -1,13 +1,12 @@
-package client
+package etcdclient
 
 import (
 	"fmt"
+	"github.com/Printemps417/optionloader/utils"
 	kitexclient "github.com/cloudwego/kitex/client"
 	"github.com/cloudwego/kitex/pkg/connpool"
 	"github.com/cloudwego/kitex/pkg/rpcinfo"
 	"github.com/cloudwego/kitex/transport"
-	"strconv"
-	"strings"
 )
 
 // Protocol indicates the transport protocol.
@@ -38,6 +37,9 @@ var protocolMap = map[string]Protocol{
 
 func basicInfoTranslator(config *EtcdConfig) ([]kitexclient.Option, error) {
 	c := config.ClientBasicInfo
+	if c == nil {
+		return nil, nil
+	}
 	var res []kitexclient.Option
 	rpcInfo := rpcinfo.EndpointBasicInfo{
 		ServiceName: c.ServiceName,
@@ -49,6 +51,9 @@ func basicInfoTranslator(config *EtcdConfig) ([]kitexclient.Option, error) {
 }
 func protocolTranslator(config *EtcdConfig) ([]kitexclient.Option, error) {
 	c := config.Protocol
+	if c == nil {
+		return nil, nil
+	}
 	var res []kitexclient.Option
 	protocol, ok := protocolMap[*c]
 	if !ok {
@@ -60,40 +65,43 @@ func protocolTranslator(config *EtcdConfig) ([]kitexclient.Option, error) {
 }
 func destServiceTranslator(config *EtcdConfig) ([]kitexclient.Option, error) {
 	c := config.DestService
+	if c == nil {
+		return nil, nil
+	}
 	var res []kitexclient.Option
 	res = append(res, kitexclient.WithDestService(*c))
 	return res, nil
 }
 
 func hostPortsTranslator(config *EtcdConfig) ([]kitexclient.Option, error) {
 	c := config.HostPorts
-	var res []kitexclient.Option
-
-	ports := strings.Split(*c, ",")
-	for _, port := range ports {
-		hostPort := strings.Split(port, ":")
-		portNum, err := strconv.Atoi(hostPort[len(hostPort)-1])
-		if err != nil || portNum < 1 || portNum > 65535 {
-			return nil, fmt.Errorf("invalid port number: %s", port)
-		}
+	if c == nil {
+		return nil, nil
 	}
-
-	res = append(res, kitexclient.WithHostPorts(*c))
+	var res []kitexclient.Option
+	res = append(res, kitexclient.WithHostPorts(c...))
 	return res, nil
 }
 func connectionTranslator(config *EtcdConfig) ([]kitexclient.Option, error) {
 	c := config.Connection
+	if c == nil {
+		return nil, nil
+	}
 	var res []kitexclient.Option
 
 	switch c.Method {
 	case "ShortConnection":
 		res = append(res, kitexclient.WithShortConnection())
 	case "LongConnection":
+		MaxIdleTimeout, err := utils.ParseDuration(c.LongConnection.MaxIdleTimeout)
+		if err != nil {
+			return nil, err
+		}
 		idleConfig := connpool.IdleConfig{
 			MinIdlePerAddress: c.LongConnection.MinIdlePerAddress,
 			MaxIdlePerAddress: c.LongConnection.MaxIdlePerAddress,
 			MaxIdleGlobal:     c.LongConnection.MaxIdleGlobal,
-			MaxIdleTimeout:    c.LongConnection.MaxIdleTimeout,
+			MaxIdleTimeout:    MaxIdleTimeout,
 		}
 		res = append(res, kitexclient.WithLongConnection(idleConfig))
 	case "MuxConnection":

etcd/server/decoder.go

@@ -1,19 +1,19 @@
-package server
+package etcdserver
 
 import (
 	"encoding/json"
 	"fmt"
 )
 
 type ConfigParser interface {
-	Decode(data []byte, config EtcdConfig) error
+	Decode(data []byte, config *EtcdConfig) error
 }
 
 type defaultParser struct {
 }
 
-func (p *defaultParser) Decode(data []byte, config EtcdConfig) error {
-	return json.Unmarshal(data, &config)
+func (p *defaultParser) Decode(data []byte, config *EtcdConfig) error {
+	return json.Unmarshal(data, config)
 }
 
 type Config interface {

etcd/server/loader.go

@@ -1,4 +1,4 @@
-package server
+package etcdserver
 
 import (
 	"github.com/cloudwego/kitex/pkg/klog"
@@ -21,6 +21,11 @@ type EtcdLoader struct {
 }
 
 func (l *EtcdLoader) Load() error {
+	path := Path{ServerServiceName: l.ServerServiceName}
+	err := l.reader.ReadToConfig(&path)
+	if err != nil {
+		return err
+	}
 	config, err := l.reader.GetConfig()
 	if err != nil {
 		return err

etcd/server/reader.go

@@ -1,4 +1,4 @@
-package server
+package etcdserver
 
 import (
 	"bytes"
@@ -13,7 +13,7 @@ const (
 	EtcdDefaultNode         = "http://127.0.0.1:2379"
 	EtcdDefaultConfigPrefix = "/KitexConfig"
 	EtcdDefaultTimeout      = 5 * time.Second
-	EtcdServerDefaultPath   = "{{.ServerServiceName}}"
+	EtcdServerDefaultPath   = "/{{.ServerServiceName}}"
 )
 
 type Reader interface {
@@ -33,7 +33,6 @@ type EtcdReader struct {
 }
 
 type Path struct {
-	ClientServiceName string
 	ServerServiceName string
 }
 
@@ -55,7 +54,7 @@ func (r *EtcdReader) ReadToConfig(p *Path) error {
 		klog.Debugf("[etcd] key: %s config get value failed", key)
 		return err
 	}
-	err = r.parser.Decode(data.Kvs[0].Value, *r.config)
+	err = r.parser.Decode(data.Kvs[0].Value, r.config)
 	if err != nil {
 		return err
 	}

etcd/server/server.go

@@ -1,8 +1,7 @@
-package server
+package etcdserver
 
 import (
 	clientv3 "go.etcd.io/etcd/client/v3"
-	"go.uber.org/zap"
 	"text/template"
 	"time"
 )
@@ -13,7 +12,6 @@ type ReaderOptions struct {
 	Prefix       string
 	PathFormat   string
 	Timeout      time.Duration
-	LoggerConfig *zap.Config
 	ConfigParser ConfigParser
 	MyConfig     Config
 }
@@ -35,8 +33,8 @@ func NewReader(opts ReaderOptions) (*EtcdReader, error) {
 		opts.PathFormat = EtcdServerDefaultPath
 	}
 	etcdClient, err := clientv3.New(clientv3.Config{
-		Endpoints: opts.Node,
-		LogConfig: opts.LoggerConfig,
+		Endpoints:   opts.Node,
+		DialTimeout: opts.Timeout,
 	})
 	if err != nil {
 		return nil, err

etcd/server/suite.go

@@ -1,4 +1,4 @@
-package server
+package etcdserver
 
 import (
 	"github.com/cloudwego/kitex/server"