最近在利用confd实现prometheus配置管理，不过confd项目很久没人维护了，所以打算看看源码，使用过程有了问题也可以自己修改。
# 启动流程
代码位置: confd/confd.go
这是main函数所在文件
主要执行了以下步骤
func main() {
// 1. 解析命令行参数，然后执行initConfig()函数初始化
flag.Parse()
if err := initConfig(); err != nil {
log.Fatal(err.Error())
}
// 2\. 根据配置文件创建一个存储客户端（如etcd zookeeper consul等）
storeClient, err := backends.New(config.BackendsConfig)
······
config.TemplateConfig.StoreClient = storeClient
······
// 3 根据配置实现不同的processor接口并运行，confd主要功能都在这里
switch {
case config.Watch:
processor = template.WatchProcessor(config.TemplateConfig, stopChan, doneChan, errChan)
default:
processor = template.IntervalProcessor(config.TemplateConfig, stopChan, doneChan, errChan, config.Interval)
}
go processor.Process()
# 初始化配置文件
confd/config.go
初始化配置这部分没什么好讲的，先略过
# backends
confd/backends/client.go
## backends.New
// 这里有一个StoreClient接口，有两个方法，一个是GetValues获取键值对，WatchPrefix 观察key是否有event触发
type StoreClient interface {
GetValues(keys []string) (map[string]string, error)
WatchPrefix(prefix string, keys []string, waitIndex uint64, stopChan chan bool) (uint64, error)
}
// 根据后端类型，实现这个接口并返回
func New(config Config) (StoreClient, error) {
······
case "etcd":
// Create the etcd client upfront and use it for the life of the process.
// The etcdClient is an http.Client and designed to be reused.
return etcd.NewEtcdClient(backendNodes, config.ClientCert, config.ClientKey, config.ClientCaKeys, config.ClientInsecure, config.BasicAuth, config.Username, config.Password)
case "etcdv3":
return etcdv3.NewEtcdClient(backendNodes, config.ClientCert, config.ClientKey, config.ClientCaKeys, config.BasicAuth, config.Username, config.Password)
case "zookeeper":
return zookeeper.NewZookeeperClient(backendNodes)
······
}
confd/backends
这里的各个文件夹存放了具体实现接口的方法
## zookeeper
### GetValues
由于我使用的是zookeeper，就先分析zookeeper client相关代码了
作用：遍历传入的key数组，获取对应的value或子节点value
confd/backends/zookeeper
这是StoreClient接口GetValues的实现
func (c *Client) GetValues(keys []string) (map[string]string, error) {
vars := make(map[string]string)
// 遍历传入的key数组
for _, v := range keys {
v = strings.Replace(v, "/*", "", -1)
_, _, err := c.client.Exists(v)
if err != nil {
return vars, fmt.Errorf("%w. %s",err,v)
}
// 遍历key下的所有子节点获取value存入vars并返回
err = nodeWalk(v, c, vars)
if err != nil {
return vars, fmt.Errorf("%w. %s",err,v)
}
}
return vars, nil
}
confd/backends/zookeeper/client.go 26
// 遍历子节点具体实现
func nodeWalk(prefix string, c *Client, vars map[string]string) error {
var s string
// 获取所有子节点
l, stat, err := c.client.Children(prefix)
if err != nil {
return fmt.Errorf("%w. %s",err,prefix)
}
// 无子节点则获取value返回
if stat.NumChildren == 0 {
b, _, err := c.client.Get(prefix)
if err != nil {
return fmt.Errorf("%w. %s",err,prefix)
}
vars[prefix] = string(b)
} else {
// 遍历子节点
for _, key := range l {
if prefix == "/" {
s = "/" + key
} else {
s = prefix + "/" + key
}
_, stat, err := c.client.Exists(s)
if err != nil {
return fmt.Errorf("%w. %s",err,s)
}
// 没有子节点则存入vars，有的话继续遍历
if stat.NumChildren == 0 {
b, _, err := c.client.Get(s)
if err != nil {
return err
}
vars[s] = string(b)
} else {
nodeWalk(s, c, vars)
}
}
}
return nil
}
### WatchPrefix
confd/backends/zookeeper/client.go:119
作用： 根据传入的prefix + keys处理，调用watch方法进行watch
func (c *Client) WatchPrefix(prefix string, keys []string, waitIndex uint64, stopChan chan bool) (uint64, error) {
entries, err := c.GetValues([]string{prefix})
watchMap := make(map[string]string)
// watch key的父级节点，直到为/再进行下一次循环
for k, _ := range entries {
for _, v := range keys {
if strings.HasPrefix(k, v) {
// 这里发现windows下会出现dir一直为\目录，造成死循环，改了一下
for dir := filepath.ToSlash(filepath.Dir(k)); dir != "/"; dir = filepath.ToSlash(filepath.Dir(dir)) {
if _, ok := watchMap[dir]; !ok {
watchMap[dir] = ""
// 主要方法watch
go c.watch(dir, respChan, cancelRoutine)
······
// watch key
for k, _ := range entries {
for _, v := range keys {
if strings.HasPrefix(k, v) {
log.Debug("Watching: " + k)
// 主要方法watch
go c.watch(k, respChan, cancelRoutine)
break
}
}
}
}
confd/backends/zookeeper/client.go:89
作用：对key进行watch，这里的watch是一次性事件，触发一次后就会结束。
zk包中声明了以下五种事件类型, 分别对应的api是 ：
EventNodeCreated：节点创建事件，需要watch一个不存在的节点，当节点被创建时触发，此watch通过conn.ExistsW(path string)设置
EventNodeDeleted：节点删除事件，需要watch一个已存在的节点，当节点被移除时触发，此watch通过conn.ExistsW(path string)设置
EventNodeDataChanged：节点数据变化事件，此watch通过conn.GetW(path string) 以及 conn.ExistsW(path string) 设置，
EventNodeChildrenChanged：子节点改变事件（数量改变），此watch通过conn.ChildrenW(path string)设置， 当path 下面增删子节点时触发（修改path下的子节点的内容时，不会触发通知）。
EventNoWatching：watch移除事件，服务端出于某些原因不再为客户端watch节点时触发。
func (c *Client) watch(key string, respChan chan watchResponse, cancelRoutine chan bool) {
// watch key事件
_, _, keyEventCh, err := c.client.GetW(key)
······
// watch key的子节点事件
_, _, childEventCh, err := c.client.ChildrenW(key)
for {
select {
case e := <-keyEventCh:
if e.Type == zk.EventNodeDataChanged {
respChan <- watchResponse{1, e.Err}
}
case e := <-childEventCh:
if e.Type == zk.EventNodeChildrenChanged {
respChan <- watchResponse{1, e.Err}
}
case <-cancelRoutine:
// There is no way to stop GetW/ChildrenW so just quit
return
}
}
}
到这backends相关的内容就结束了，这里的大概流程就是
backends.New实现StoreClient接口，赋值到config中给后面的主要干活的processor使用
# processor
TODO
# 参考文章：
https://aimuke.github.io/zookeeper/2019/06/18/zookeeper-watch/
