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

[jiuxia211]

背景

Support integration with well-known config centers

目标

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

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

配置存储参考规范

zookeeper中，配置信息由节点的形式存储，每个节点有一个唯一的路径(path)和对应的数据(data)

字段	取值	说明
path	格式为$Prefix/$ClientName/$ServerName/$ConfigCategory，也可以由用户自定义	路径（Path）是用来唯一标识每个Znode（ZooKeeper节点）的字符串。以"/"作为分隔符。默认Prefix为“/KitexConfig”，其中 Category 为枚举值："rpc_timeout", "retry", "circuit_break", "limit"，用于指定配置类型，超时、重试、熔断、限流

在 ZooKeeper 中，创建节点时，节点的父节点必须存在。

客户端配置

超时：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 方法）

熔断：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）

服务端

限流：Category=limit

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

配置信息格式为 JSON，其 Schema 为 limiter.LimiterConfig

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

动态配置

import (
    "fmt"
    "time"

    "github.com/cloudwego/kitex/pkg/klog"
    "github.com/go-zookeeper/zk"
)

func main() {
    // 连接zookeeper
    hosts := []string{"127.0.0.1:2181"}
    conn, _, err := zk.Connect(hosts, time.Second*5)
    if err != nil {
        fmt.Println("Failed to Connected to zookeeper")
        return
    }
    defer conn.Close()
    // 监听节点路径
    path := "/path/to/node"
    // 开启goroutine监听
    go func() {
        for {
            _, _, watchChan, err := conn.ExistsW(path)
            if err != nil {
                klog.Debugf("Watch node %s failed %v", path, err)
                return
            }
            watchEvent := <-watchChan
            switch watchEvent.Type {
            case zk.EventNodeDataChanged:
                // 监听到配置更新
                // 获取配置信息
                data, _, err := conn.Get(path)
                if err != nil {
                    klog.Warnf("Get data %s from zookeeper failed: %v", path, err)
                }
                noteValue := string(data)
                fmt.Printf("Node path: %s\n", watchEvent.Path)
                fmt.Printf("Node data updated. New data: %s\n", noteValue)
            case zk.EventNodeDeleted:
                // 监听到配置信息被删除
                fmt.Printf("Node path: %s\n", watchEvent.Path)
                fmt.Printf("Node data deleted\n")
            default:
                // 监听到其他event
                fmt.Println("path: ", watchEvent.Path)
                fmt.Println("type: ", watchEvent.Type.String())
                fmt.Println("state: ", watchEvent.State.String())
            }
        }
    }()
}

参考

• Zookeeper
  • Go的zookeeper客户端库：https://github.com/go-zookeeper/zk

[felix021]

zookeeper 没有 「dataid」这个概念，只有 path，一般应该是用 / 作为分隔符，所以一个典型的path应该是 $Prefix/$ClientName/$ServerName/$ConfigCategory（默认prefix可以是 /KitexConfig）；由于用户会有更复杂的需求，因此该路径应当是可以被用户自定义的。

[ViolaPioggia]

开启goroutine监听后，_, _, watchChan, err := conn.ExistsW(path)也许不太适合放在 for 循环内部

[jiuxia211]

zookeeper 没有 「dataid」这个概念，只有 path，一般应该是用 / 作为分隔符，所以一个典型的path应该是 $Prefix/$ClientName/$ServerName/$ConfigCategory（默认prefix可以是 /KitexConfig）；由于用户会有更复杂的需求，因此该路径应当是可以被用户自定义的。

我理解了你的意思，已经修改

[jiuxia211]

开启goroutine监听后，_, _, watchChan, err := conn.ExistsW(path)也许不太适合放在 for 循环内部

__, _, watchChan, err := conn.ExistsW(path)的watchChan一旦被触发，就会被删除，因此需要在每次收到 Watch 通知后重新设置 Watch。

[ViolaPioggia]

__, _, watchChan, err := conn.ExistsW(path)的watchChan一旦被触发，就会被删除，因此需要在每次收到 Watch 通知后重新设置 Watch。

这样的话就没问题了，后续可以关注一下取消对 path 的监听如何不影响另外的监听同一个 path 的 watchChan

[felix021]

OK，方案整体没什么太大问题，我们后面关注具体的代码

[li-jin-gou]

Thanks for the contribution, I'm closing this issue since it's done.