consul as config center for service governance.
go get github.com/kitex-contrib/config-consul
package main
import (
"github.com/kitex-contrib/config-consul/consul"
"context"
"log"
consulserver "github.com/kitex-contrib/config-consul/server"
"github.com/cloudwego/kitex-examples/kitex_gen/api"
"github.com/cloudwego/kitex-examples/kitex_gen/api/echo"
"github.com/cloudwego/kitex/pkg/klog"
"github.com/cloudwego/kitex/pkg/rpcinfo"
"github.com/cloudwego/kitex/server"
)
var _ api.Echo = &EchoImpl{}
// EchoImpl implements the last service interface defined in the IDL.
type EchoImpl struct{}
// Echo implements the Echo interface.
func (s *EchoImpl) Echo(ctx context.Context, req *api.Request) (resp *api.Response, err error) {
klog.Info("echo called")
return &api.Response{Message: req.Message}, nil
}
func main() {
klog.SetLevel(klog.LevelDebug)
serviceName := "ServiceName" // your server-side service name
consulClient, _ := consul.NewClient(consul.Options{})
svr := echo.NewServer(
new(EchoImpl),
server.WithServerBasicInfo(&rpcinfo.EndpointBasicInfo{ServiceName: serviceName}),
server.WithSuite(consulserver.NewSuite(serviceName, consulClient)),
)
if err := svr.Run(); err != nil {
log.Println("server stopped with error:", err)
} else {
log.Println("server stopped")
}
}
package main
import (
"github.com/kitex-contrib/config-consul/consul"
"github.com/kitex-contrib/config-consul/utils"
"context"
"log"
"time"
consulclient "github.com/kitex-contrib/config-consul/client"
"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"
)
type configLog struct{}
func (cl *configLog) Apply(opt *utils.Options) {
fn := func(k *consul.Key) {
klog.Infof("consul config %v", k)
}
opt.ConsulCustomFunctions = append(opt.ConsulCustomFunctions, fn)
}
func main() {
consulClient, err := consul.NewClient(consul.Options{})
if err != nil {
panic(err)
}
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(consulclient.NewSuite(serviceName, clientName, consulClient, 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)
}
}
Provide the mechanism to custom the consul parameter Key
.
type Key struct {
Type ConfigType
Prefix string
Path string
}
Variable Name | Default Value |
---|---|
Addr | 127.0.0.1:8500 |
Prefix | /KitexConfig |
ServerPathFormat | {{.ServerServiceName}}/{{.Category}} |
ClientPathFormat | {{.ClientServiceName}}/{{.ServerServiceName}}/{{.Category}} |
DataCenter | dc1 |
Timeout | 5 * time.Second |
NamespaceId | |
Token | |
Partition | |
LoggerConfig | NULL |
ConfigParser | defaultConfigParser |
The configPath and configPrefix in the following example use default values, the service name is
ServiceName
and the client name isClientName
.
Currently, current limiting only supports the server side, so ClientServiceName is empty.
Variable | Introduction |
---|---|
connection_limit | Maximum concurrent connections |
qps_limit | Maximum request number every 100ms |
Example:
configPath: /KitexConfig/ServiceName/limit
{
"connection_limit": 100,
"qps_limit": 2000
}
Note:
- The granularity of the current limit configuration is server global, regardless of client or method.
- Not configured or value is 0 means not enabled.
- connection_limit and qps_limit can be configured independently, e.g. connection_limit = 100, qps_limit = 0
Variable | Introduction |
---|---|
type | 0: failure_policy 1: backup_policy |
failure_policy.backoff_policy | Can only be set one of fixed none random |
Example:
configPath: /KitexConfig/ClientName/ServiceName/retry
{
"*": {
"enable": true,
"type": 0,
"failure_policy": {
"stop_policy": {
"max_retry_times": 3,
"max_duration_ms": 2000,
"cb_policy": {
"error_rate": 0.3
}
},
"backoff_policy": {
"backoff_type": "fixed",
"cfg_items": {
"fix_ms": 50
}
},
"retry_same_node": false
}
},
"echo": {
"enable": true,
"type": 1,
"backup_policy": {
"retry_delay_ms": 100,
"retry_same_node": false,
"stop_policy": {
"max_retry_times": 2,
"max_duration_ms": 300,
"cb_policy": {
"error_rate": 0.2
}
}
}
}
}
Note: retry.Container has built-in support for specifying the default configuration using the *
wildcard (see the getRetryer method for details).
Example:
configPath: /KitexConfig/ClientName/ServiceName/rpc_timeout
{
"*": {
"conn_timeout_ms": 100,
"rpc_timeout_ms": 3000
},
"echo": {
"conn_timeout_ms": 50,
"rpc_timeout_ms": 1000
}
}
Note: The circuit breaker implementation of kitex does not currently support changing the global default configuration (see initServiceCB for details).
Variable | Introduction |
---|---|
min_sample | Minimum statistical sample number |
Example:
The echo method uses the following configuration (0.3, 100) and other methods use the global default configuration (0.5, 200)
configPath: /KitexConfig/ClientName/ServiceName/circuit_break
{
"echo": {
"enable": true,
"err_rate": 0.3,
"min_sample": 100
}
}
Variable | Introduction |
---|---|
enable | Whether to enable degradation |
percentage | The percentage of dropped requests |
Example:
configPath: /KitexConfig/ClientName/ServiceName/degradation
{
"enable": true,
"percentage": 30
}
Note: Degradation is not enabled by default.
Refer to example for more usage.
the version of Go must >=1.20
maintained by: hiahia12