Proposal: 对接 etcd 配置中心 | etcd As Config Center

[ViolaPioggia]

目标

基于本文描述的配置存储参考规范：

• 按 Kitex 的 [Suite 扩展规范]，实现 EtcdClientSuite、EtcdServerSuite；
• 基于上面实现的方案，写一个简单的例子（Kitex Client & Server），用于展示实际用法；

配置存储参考规范

根据 [etcd官方提供的Go语言客户端库clientv3]，配置以键值对的形式存储，每个配置项都有一个对应的键（Key）和值（Value）。配置数据存储在etcd的分布式键值存储空间中。

和例如 nacos 等配置中心不同的是，etcd 没有类似于 namespace 的命名空间，只有前缀（prefix）的方式来作区分

字段	取值	说明
prefix	默认 KitexConfig，也可以由用户自己自定义	用于指定键的前缀。它在etcd中用于进行键的模式匹配和筛选，以便对特定前缀的键进行操作
dataid	格式为 ${ClientServiceName}.${ServerServiceName}.${Category}；也可以由用户自定义	其中 Category 为枚举值："rpc_timeout", "retry", "circuit_break", "limit"，用于指定配置类型，超时、重试、熔断、限流

## 客户端配置

超时：Category=rpc_timeout

配置信息格式为 JSON，其 Schema 为

map[string]*rpctimeout.RPCTimeout // 分method指定超时配置

说明：

• rpctimeout.RPCTimeout: https://github.com/cloudwego/kitex/blob/develop/pkg/rpctimeout/item_rpc_timeout.go#L42
• map 的 key 是 methodName，用于分 method 指定超时配置
• 当 methodName = * 时，表示这是默认配置，如果找不到 method 配置，会默认使用这个配置

例如：Echo 方法链接超时 50ms、请求超时 1s，其他方法连接超时 100ms、请求超时 2s

{
  "*": {
    "conn_timeout_ms": 100,
    "rpc_timeout_ms": 2000
  },
  "Echo": {
    "conn_timeout_ms": 50,
    "rpc_timeout_ms": 1000
  }
}

注：Kitex 只区分连接超时和读写超时，因此只需要指定两个字段即可。

重试：Category=retry

配置信息格式为 JSON，其 Schema 为

map[string]*retry.Policy

说明：

• retry.Policy: https://github.com/cloudwego/kitex/blob/develop/pkg/retry/policy.go#L63

例如：Echo 方法启用备用请求（200ms后未返回则发出），其他请求都默认失败重试3次

{
    "*": {
        "enable": true,
        "type": 0,                 // 失败重试(type=0)
        "failure_policy": {
            "stop_policy": {
                "max_retry_times": 3,
                "max_duration_ms": 2000,
                "cb_policy": {
                    "error_rate": 0.1
                }
            },
            "backoff_policy": {
                "backoff_type": "fixed",
                "cfg_items": {
                    "fix_ms": 50
                }
            }
        }
    },
    "Echo": {
        "enable": true,
        "type": 1,                // 备用请求(type=1)
        "backup_policy": {
            "retry_delay_ms": 200,
            "stop_policy": {
                "max_retry_times": 2,
                "max_duration_ms": 1000,
                "cb_policy": {
                    "error_rate": 0.3
                }
            }
        }
    }
}

注：retry.Container 内置支持用 * 通配符指定默认配置（详见 [getRetryer](https://github.com/cloudwego/kitex/blob/v0.5.1/pkg/retry/retryer.go#L240) 方法）

熔断：Category=circuit_break

配置信息格式为 JSON，其 Schema 为

map[string]*circuitbreak.CBConfigItem

注：

• circuitbreak.CBConfigItem: https://github.com/cloudwego/kitex/blob/develop/pkg/circuitbreak/item_circuit_breaker.go#L30

例如：Echo 方法使用下面的配置（0.3、100），其他方法使用全局默认配置（0.5、200）

{
  "Echo": {
    "enable": true,
    "err_rate": 0.3,
    "min_sample": 100
  }
}

注：kitex 的熔断实现目前不支持修改全局默认配置（详见 [initServiceCB](https://github.com/cloudwego/kitex/blob/v0.5.1/pkg/circuitbreak/cbsuite.go#L195)）

服务端

限流：Category=limit

Group 里 CallerName 的值留空，表示是服务端配置，与客户端无关

配置信息格式为 JSON，其 Schema 为 [limiter.LimiterConfig](https://github.com/cloudwego/kitex/blob/develop/pkg/limiter/item_limiter.go#L33)

type LimiterConfig struct {
	ConnectionLimit int64 `json:"connection_limit"`
	QPSLimit        int64 `json:"qps_limit"`
}

例如：最大100并发 && 每 100ms 内最大 2000QPS

{
  "connection_limit": 100,
  "qps_limit": 200
}

注：

1. 限流配置的粒度是 Server 全局，不分 client、method
2. 「未配置」或「取值为 0」表示不开启
3. connection_limit 和 qps_limit 可以独立配置，例如 connection_limit = 100, qps_limit = 0

动态配置

在这个 demo 中我们启动了一个循环，通过client.Watch()方法来监听指定键的变化。当配置发生变化时，我们会收到相应的事件通知，我们可以根据事件的类型进行相应的处理。

import (
	"context"
	"fmt"
	"github.com/coreos/etcd/clientv3"
	"github.com/coreos/etcd/mvcc/mvccpb"
)

func main() {
	// 创建etcd客户端连接
	client, err := clientv3.New(clientv3.Config{
		Endpoints: []string{"http://localhost:2379"}, // etcd服务器的地址和端口
	})
	if err != nil {
		fmt.Println("Failed to connect to etcd:", err)
		return
	}
	defer client.Close()

	// 指定要监听的键
	key := "/config/database/url"

	// 启动监听
	watchChan := client.Watch(context.Background(), key)
	for watchResp := range watchChan {
		for _, event := range watchResp.Events {
			// 检查事件类型
			if event.Type == mvccpb.PUT {
				// 配置被更新
				value := string(event.Kv.Value)
				fmt.Println("Config updated. New value:", value)
			} else if event.Type == mvccpb.DELETE {
				// 配置被删除
				fmt.Println("Config deleted.")
			}
		}
	}
}

参考

• etcd
  • 官网：https://etcd.io/
  • Open API：https://pkg.go.dev/go.etcd.io/etcd/client/v3

[ozline]

Using context.Background() may not be a good choice, how do you close watchChan?

[ViolaPioggia]

Using context.Background() may not be a good choice, how do you close watchChan?

这里我是想实现一个长期的配置监听，context.Background()在正式的代码中我的想法是传入一个context，这样就能实现程序停止运行的时候监听也停止运行

[ozline]

Using context.Background() may not be a good choice, how do you close watchChan?

这里我是想实现一个长期的配置监听，context.Background()在正式的代码中我的想法是传入一个context，这样就能实现程序停止运行的时候监听也停止运行

应该要有一个退出时的一个处理，可以考虑增加一些细节

[ViolaPioggia]

Using context.Background() may not be a good choice, how do you close watchChan?

这里我是想实现一个长期的配置监听，context.Background()在正式的代码中我的想法是传入一个context，这样就能实现程序停止运行的时候监听也停止运行

应该要有一个退出时的一个处理，可以考虑增加一些细节

那就监听停止信号，用 cancel() 实现一个优雅退出

[whalecold]

配置项的 Key 值可以加点描述，还有默认的 prefix 是不是使用 kitexConfig 更好

[ViolaPioggia]

已将默认值设置为 KitexConfig，Key 加上了下列描述：
这里的键（Key）就是前缀（prefix），推荐将配置信息以 yaml 或者 json 等其他形式序列化之后存放到一个前缀目录下

[felix021]

已将默认值设置为 KitexConfig，Key 加上了下列描述： 这里的键（Key）就是前缀（prefix），推荐将配置信息以 yaml 或者 json 等其他形式序列化之后存放到一个前缀目录下

这个还是有点模糊。举个具体的例子，我有2个服务 A、B，A要请求B的Echo方法，那么对应的超时、重试配置在 etcd 里存储的路径是什么呢？

[ViolaPioggia]

这个还是有点模糊。举个具体的例子，我有2个服务 A、B，A要请求B的Echo方法，那么对应的超时、重试配置在 etcd 里存储的路径是什么呢？

好的，我觉得可以参考 nacos，存储在"rpc_timeout", "retry", "circuit_break", "limit"下，用于指定配置类型，超时、重试、熔断、限流

[felix021]

这个还是有点模糊。举个具体的例子，我有2个服务 A、B，A要请求B的Echo方法，那么对应的超时、重试配置在 etcd 里存储的路径是什么呢？

好的，我觉得可以参考 nacos，存储在"rpc_timeout", "retry", "circuit_break", "limit"下，用于指定配置类型，超时、重试、熔断、限流

和 nacos 还是不一样，建议你在本机实际搭一个单机版 etcd，用 etcdctl 命令写一个超时配置进去，你才能理解我上面的问题：「完整的路径是什么」，前缀，服务(client, server)名称，类型，还要考虑允许用户自定义（参考 nacos 的 template）