confd的源码参考：https://github.com/kelseyhightower/confd
本文分析的confd的版本是v0.16.0，代码参考：https://github.com/kelseyhightower/confd/tree/v0.16.0。
1. Main
confd的入口函数 Main 函数，先解析参数，如果是打印版本信息的参数，则执行打印版本的命令。
func main() {
flag.Parse()
if config.PrintVersion {
fmt.Printf("confd %s (Git SHA: %s, Go Version: %s)\n", Version, GitSHA, runtime.Version())
os.Exit(0)
}
...
}
其中版本信息记录在https://github.com/kelseyhightower/confd/blob/v0.16.0/version.go#L3
const Version = "0.16.0"
1.1. initConfig
初始化配置文件。
if err := initConfig(); err != nil {
log.Fatal(err.Error())
}
initConfig函数对基本的配置内容做初始化，当没有指定后端存储的时候，设置默认存储。
// initConfig initializes the confd configuration by first setting defaults,
// then overriding settings from the confd config file, then overriding
// settings from environment variables, and finally overriding
// settings from flags set on the command line.
// It returns an error if any.
func initConfig() error {
_, err := os.Stat(config.ConfigFile)
if os.IsNotExist(err) {
log.Debug("Skipping confd config file.")
} else {
log.Debug("Loading " + config.ConfigFile)
configBytes, err := ioutil.ReadFile(config.ConfigFile)
if err != nil {
return err
}
_, err = toml.Decode(string(configBytes), &config)
if err != nil {
return err
}
}
// Update config from environment variables.
processEnv()
if config.SecretKeyring != "" {
kr, err := os.Open(config.SecretKeyring)
if err != nil {
log.Fatal(err.Error())
}
defer kr.Close()
config.PGPPrivateKey, err = ioutil.ReadAll(kr)
if err != nil {
log.Fatal(err.Error())
}
}
if config.LogLevel != "" {
log.SetLevel(config.LogLevel)
}
if config.SRVDomain != "" && config.SRVRecord == "" {
config.SRVRecord = fmt.Sprintf("_%s._tcp.%s.", config.Backend, config.SRVDomain)
}
// Update BackendNodes from SRV records.
if config.Backend != "env" && config.SRVRecord != "" {
log.Info("SRV record set to " + config.SRVRecord)
srvNodes, err := getBackendNodesFromSRV(config.SRVRecord)
if err != nil {
return errors.New("Cannot get nodes from SRV records " + err.Error())
}
switch config.Backend {
case "etcd":
vsm := make([]string, len(srvNodes))
for i, v := range srvNodes {
vsm[i] = config.Scheme + "://" + v
}
srvNodes = vsm
}
config.BackendNodes = srvNodes
}
if len(config.BackendNodes) == 0 {
switch config.Backend {
case "consul":
config.BackendNodes = []string{"127.0.0.1:8500"}
case "etcd":
peerstr := os.Getenv("ETCDCTL_PEERS")
if len(peerstr) > 0 {
config.BackendNodes = strings.Split(peerstr, ",")
} else {
config.BackendNodes = []string{"http://127.0.0.1:4001"}
}
case "etcdv3":
config.BackendNodes = []string{"127.0.0.1:2379"}
case "redis":
config.BackendNodes = []string{"127.0.0.1:6379"}
case "vault":
config.BackendNodes = []string{"http://127.0.0.1:8200"}
case "zookeeper":
config.BackendNodes = []string{"127.0.0.1:2181"}
}
}
// Initialize the storage client
log.Info("Backend set to " + config.Backend)
if config.Watch {
unsupportedBackends := map[string]bool{
"dynamodb": true,
"ssm": true,
}
if unsupportedBackends[config.Backend] {
log.Info(fmt.Sprintf("Watch is not supported for backend %s. Exiting...", config.Backend))
os.Exit(1)
}
}
if config.Backend == "dynamodb" && config.Table == "" {
return errors.New("No DynamoDB table configured")
}
config.ConfigDir = filepath.Join(config.ConfDir, "conf.d")
config.TemplateDir = filepath.Join(config.ConfDir, "templates")
return nil
}
1.2. storeClient
log.Info("Starting confd")
storeClient, err := backends.New(config.BackendsConfig)
if err != nil {
log.Fatal(err.Error())
}
根据配置文件中的存储后端类型构造一个存储后端的client，其中主要调用的函数为backends.New(config.BackendsConfig)。
当没有设置存储后端时，默认为etcd。
if config.Backend == "" {
config.Backend = "etcd"
}
backendNodes := config.BackendNodes
当存储后端为file类型的处理。
if config.Backend == "file" {
log.Info("Backend source(s) set to " + strings.Join(config.YAMLFile, ", "))
} else {
log.Info("Backend source(s) set to " + strings.Join(backendNodes, ", "))
}
最后再根据不同类型的存储后端，调用不同的存储后端构建函数，本文只分析redis类型的存储后端。
switch config.Backend {
case "consul":
return consul.New(config.BackendNodes, config.Scheme,
config.ClientCert, config.ClientKey,
config.ClientCaKeys,
config.BasicAuth,
config.Username,
config.Password,
)
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.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)
case "rancher":
return rancher.NewRancherClient(backendNodes)
case "redis":
return redis.NewRedisClient(backendNodes, config.ClientKey, config.Separator)
case "env":
return env.NewEnvClient()
case "file":
return file.NewFileClient(config.YAMLFile, config.Filter)
case "vault":
vaultConfig := map[string]string{
"app-id": config.AppID,
"user-id": config.UserID,
"role-id": config.RoleID,
"secret-id": config.SecretID,
"username": config.Username,
"password": config.Password,
"token": config.AuthToken,
"cert": config.ClientCert,
"key": config.ClientKey,
"caCert": config.ClientCaKeys,
"path": config.Path,
}
return vault.New(backendNodes[0], config.AuthType, vaultConfig)
case "dynamodb":
table := config.Table
log.Info("DynamoDB table set to " + table)
return dynamodb.NewDynamoDBClient(table)
case "ssm":
return ssm.New()
}
return nil, errors.New("Invalid backend")
其中redis类型的存储后端调用了NewRedisClient方法来构造redis的client。
case "redis":
return redis.NewRedisClient(backendNodes, config.ClientKey, config.Separator)
其中涉及三个参数：
backendNodes：redis的节点地址。ClientKey：redis的密码。Separator：查找redis键的分隔符，该参数只用在redis类型。
NewRedisClient函数方法如下：
// NewRedisClient returns an *redis.Client with a connection to named machines.
// It returns an error if a connection to the cluster cannot be made.
func NewRedisClient(machines []string, password string, separator string) (*Client, error) {
if separator == "" {
separator = "/"
}
log.Debug(fmt.Sprintf("Redis Separator: %#v", separator))
var err error
clientWrapper := &Client{machines: machines, password: password, separator: separator, client: nil, pscChan: make(chan watchResponse), psc: redis.PubSubConn{Conn: nil} }
clientWrapper.client, _, err = tryConnect(machines, password, true)
return clientWrapper, err
}
1.3. processor
stopChan := make(chan bool)
doneChan := make(chan bool)
errChan := make(chan error, 10)
var processor template.Processor
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()
当开启watch参数的时候，则构造WatchProcessor，否则构造IntervalProcessor，最后起一个goroutine。
go processor.Process()
这块的逻辑在本文第二部分析。
1.4. signalChan
signalChan := make(chan os.Signal, 1)
signal.Notify(signalChan, syscall.SIGINT, syscall.SIGTERM)
for {
select {
case err := <-errChan:
log.Error(err.Error())
case s := <-signalChan:
log.Info(fmt.Sprintf("Captured %v. Exiting...", s))
close(doneChan)
case <-doneChan:
os.Exit(0)
}
}
2. Process
type Processor interface {
Process()
}
Processor是一个接口类型，主要的实现体有：
intervalProcessor：默认的实现体，即没有添加watch参数。watchProcessor：添加watch参数的实现体。
2.1. intervalProcessor
type intervalProcessor struct {
config Config
stopChan chan bool
doneChan chan bool
errChan chan error
interval int
}
intervalProcessor根据config内容和几个channel构造一个intervalProcessor。
func IntervalProcessor(config Config, stopChan, doneChan chan bool, errChan chan error, interval int) Processor {
return &intervalProcessor{config, stopChan, doneChan, errChan, interval}
}
2.1.1. intervalProcessor.Process
func (p *intervalProcessor) Process() {
defer close(p.doneChan)
for {
ts, err := getTemplateResources(p.config)
if err != nil {
log.Fatal(err.Error())
break
}
process(ts)
select {
case <-p.stopChan:
break
case <-time.After(time.Duration(p.interval) * time.Second):
continue
}
}
}
通过解析config内容获取TemplateResources，其中核心函数为process(ts)，然后执行t.process()，该函数中会调用t.sync()。t.process()的具体逻辑后文分析。
func process(ts []*TemplateResource) error {
var lastErr error
for _, t := range ts {
if err := t.process(); err != nil {
log.Error(err.Error())
lastErr = err
}
}
return lastErr
}
2.2. watchProcessor
type watchProcessor struct {
config Config
stopChan chan bool
doneChan chan bool
errChan chan error
wg sync.WaitGroup
}
watchProcessor根据config内容和几个channel构造一个watchProcessor。
func WatchProcessor(config Config, stopChan, doneChan chan bool, errChan chan error) Processor {
var wg sync.WaitGroup
return &watchProcessor{config, stopChan, doneChan, errChan, wg}
}
2.2.1. watchProcessor.Process
func (p *watchProcessor) Process() {
defer close(p.doneChan)
ts, err := getTemplateResources(p.config)
if err != nil {
log.Fatal(err.Error())
return
}
for _, t := range ts {
t := t
p.wg.Add(1)
go p.monitorPrefix(t)
}
p.wg.Wait()
}
watchProcessor.Process方法实现了Processor接口中定义的方法，通过解析config内容获取TemplateResources，再遍历TemplateResources执行monitorPrefix，有多少个TemplateResources就运行多少个monitorPrefix的goroutine。
2.2.2. monitorPrefix
func (p *watchProcessor) monitorPrefix(t *TemplateResource) {
defer p.wg.Done()
keys := util.AppendPrefix(t.Prefix, t.Keys)
for {
index, err := t.storeClient.WatchPrefix(t.Prefix, keys, t.lastIndex, p.stopChan)
if err != nil {
p.errChan <- err
// Prevent backend errors from consuming all resources.
time.Sleep(time.Second * 2)
continue
}
t.lastIndex = index
if err := t.process(); err != nil {
p.errChan <- err
}
}
}
先对配置文件中的prefix和keys参数进行拼接。
keys := util.AppendPrefix(t.Prefix, t.Keys)
AppendPrefix函数如下：
func AppendPrefix(prefix string, keys []string) []string {
s := make([]string, len(keys))
for i, k := range keys {
s[i] = path.Join(prefix, k)
}
return s
}
接着再执行storeClient的WatchPrefix方法，因为storeClient是一个接口，对应不同类型的存储后端，WatchPrefix的实现逻辑也不同，本文分析的存储类型为redis。
index, err := t.storeClient.WatchPrefix(t.Prefix, keys, t.lastIndex, p.stopChan)
if err != nil {
p.errChan <- err
// Prevent backend errors from consuming all resources.
time.Sleep(time.Second * 2)
continue
}
storeClient.WatchPrefix主要是获取lastIndex的值，这个值在t.process()中使用。
t.lastIndex = index
if err := t.process(); err != nil {
p.errChan <- err
}
2.3. TemplateResource.process
无论是否加watch参数，即intervalProcessor和watchProcessor最终都会调用到TemplateResource.process这个函数，而这个函数中的核心函数为t.sync()。
// process is a convenience function that wraps calls to the three main tasks
// required to keep local configuration files in sync. First we gather vars
// from the store, then we stage a candidate configuration file, and finally sync
// things up.
// It returns an error if any.
func (t *TemplateResource) process() error {
if err := t.setFileMode(); err != nil {
return err
}
if err := t.setVars(); err != nil {
return err
}
if err := t.createStageFile(); err != nil {
return err
}
if err := t.sync(); err != nil {
return err
}
return nil
}
2.3.1. setFileMode
setFileMode设置文件的权限，如果没有在配置文件指定mode参数则默认为0644，否则根据配置文件中指定的mode来设置文件权限。
// setFileMode sets the FileMode.
func (t *TemplateResource) setFileMode() error {
if t.Mode == "" {
if !util.IsFileExist(t.Dest) {
t.FileMode = 0644
} else {
fi, err := os.Stat(t.Dest)
if err != nil {
return err
}
t.FileMode = fi.Mode()
}
} else {
mode, err := strconv.ParseUint(t.Mode, 0, 32)
if err != nil {
return err
}
t.FileMode = os.FileMode(mode)
}
return nil
}
2.3.2. setVars
setVars将后端存储中最新的值拿出来暂存到内存中供后续进程使用。其中根据不同的后端，storeClient.GetValues的逻辑可能不同，但通过接口的方式可以让不同的存储后端实现不同的获取值的方法。
// setVars sets the Vars for template resource.
func (t *TemplateResource) setVars() error {
var err error
log.Debug("Retrieving keys from store")
log.Debug("Key prefix set to " + t.Prefix)
result, err := t.storeClient.GetValues(util.AppendPrefix(t.Prefix, t.Keys))
if err != nil {
return err
}
log.Debug("Got the following map from store: %v", result)
t.store.Purge()
for k, v := range result {
t.store.Set(path.Join("/", strings.TrimPrefix(k, t.Prefix)), v)
}
return nil
}
2.3.3. createStageFile
createStageFile通过src的template文件和最新内存中的变量数据生成StageFile，该文件在sync中和目标文件进行比较，看是否有修改。即StageFile实际上是根据后端存储生成的最新的配置文件，如果这份配置文件跟当前的配置文件不同，表明后端存储的数据被更新了需要重新生成一份新的配置文件。
// createStageFile stages the src configuration file by processing the src
// template and setting the desired owner, group, and mode. It also sets the
// StageFile for the template resource.
// It returns an error if any.
func (t *TemplateResource) createStageFile() error {
log.Debug("Using source template " + t.Src)
if !util.IsFileExist(t.Src) {
return errors.New("Missing template: " + t.Src)
}
log.Debug("Compiling source template " + t.Src)
tmpl, err := template.New(filepath.Base(t.Src)).Funcs(t.funcMap).ParseFiles(t.Src)
if err != nil {
return fmt.Errorf("Unable to process template %s, %s", t.Src, err)
}
// create TempFile in Dest directory to avoid cross-filesystem issues
temp, err := ioutil.TempFile(filepath.Dir(t.Dest), "."+filepath.Base(t.Dest))
if err != nil {
return err
}
if err = tmpl.Execute(temp, nil); err != nil {
temp.Close()
os.Remove(temp.Name())
return err
}
defer temp.Close()
// Set the owner, group, and mode on the stage file now to make it easier to
// compare against the destination configuration file later.
os.Chmod(temp.Name(), t.FileMode)
os.Chown(temp.Name(), t.Uid, t.Gid)
t.StageFile = temp
return nil
}
2.3.4. sync
if err := t.sync(); err != nil {
return err
}
t.sync()是执行confd核心功能的函数，将配置文件通过模板的方式自动生成，并执行检查命令和reload命令。该部分逻辑在本文第三部分分析。
3. sync
sync通过比较源文件和目标文件的差别，如果不同则重新生成新的配置，当设置了check_cmd和reload_cmd的时候，会执行check_cmd指定的检查命令，如果都没有问题则执行reload_cmd中指定的reload命令。
3.1. IsConfigChanged
IsConfigChanged比较源文件和目标文件是否相等，其中比较内容包括：Uid、Gid、Mode、Md5。只要其中任意值不同则认为两个文件不同。
// IsConfigChanged reports whether src and dest config files are equal.
// Two config files are equal when they have the same file contents and
// Unix permissions. The owner, group, and mode must match.
// It return false in other cases.
func IsConfigChanged(src, dest string) (bool, error) {
if !IsFileExist(dest) {
return true, nil
}
d, err := FileStat(dest)
if err != nil {
return true, err
}
s, err := FileStat(src)
if err != nil {
return true, err
}
if d.Uid != s.Uid {
log.Info(fmt.Sprintf("%s has UID %d should be %d", dest, d.Uid, s.Uid))
}
if d.Gid != s.Gid {
log.Info(fmt.Sprintf("%s has GID %d should be %d", dest, d.Gid, s.Gid))
}
if d.Mode != s.Mode {
log.Info(fmt.Sprintf("%s has mode %s should be %s", dest, os.FileMode(d.Mode), os.FileMode(s.Mode)))
}
if d.Md5 != s.Md5 {
log.Info(fmt.Sprintf("%s has md5sum %s should be %s", dest, d.Md5, s.Md5))
}
if d.Uid != s.Uid || d.Gid != s.Gid || d.Mode != s.Mode || d.Md5 != s.Md5 {
return true, nil
}
return false, nil
}
如果文件发生改变则执行check_cmd命令（有配置的情况下），重新生成配置文件，并执行reload_cmd命令（有配置的情况下）。
if ok {
log.Info("Target config " + t.Dest + " out of sync")
if !t.syncOnly && t.CheckCmd != "" {
if err := t.check(); err != nil {
return errors.New("Config check failed: " + err.Error())
}
}
log.Debug("Overwriting target config " + t.Dest)
err := os.Rename(staged, t.Dest)
if err != nil {
if strings.Contains(err.Error(), "device or resource busy") {
log.Debug("Rename failed - target is likely a mount. Trying to write instead")
// try to open the file and write to it
var contents []byte
var rerr error
contents, rerr = ioutil.ReadFile(staged)
if rerr != nil {
return rerr
}
err := ioutil.WriteFile(t.Dest, contents, t.FileMode)
// make sure owner and group match the temp file, in case the file was created with WriteFile
os.Chown(t.Dest, t.Uid, t.Gid)
if err != nil {
return err
}
} else {
return err
}
}
if !t.syncOnly && t.ReloadCmd != "" {
if err := t.reload(); err != nil {
return err
}
}
log.Info("Target config " + t.Dest + " has been updated")
} else {
log.Debug("Target config " + t.Dest + " in sync")
}
3.2. check
check检查暂存的配置文件即stageFile，该文件是由最新的后端存储中的数据生成的。
if !t.syncOnly && t.CheckCmd != "" {
if err := t.check(); err != nil {
return errors.New("Config check failed: " + err.Error())
}
}
t.check()只是执行配置文件中checkcmd参数指定的命令而已，根据是否执行成功来返回报错。当check命令产生错误的是，则直接return报错，不再执行重新生成配置文件和`reload的操作了。
// check executes the check command to validate the staged config file. The
// command is modified so that any references to src template are substituted
// with a string representing the full path of the staged file. This allows the
// check to be run on the staged file before overwriting the destination config
// file.
// It returns nil if the check command returns 0 and there are no other errors.
func (t *TemplateResource) check() error {
var cmdBuffer bytes.Buffer
data := make(map[string]string)
data["src"] = t.StageFile.Name()
tmpl, err := template.New("checkcmd").Parse(t.CheckCmd)
if err != nil {
return err
}
if err := tmpl.Execute(&cmdBuffer, data); err != nil {
return err
}
return runCommand(cmdBuffer.String())
}
check会通过模板解析的方式解析出checkcmd中的{{.src}}部分，并用stageFile来替代。即check的命令是拉取最新后端存储的数据形成临时配置文件（stageFile），并通过指定的checkcmd来检查最新的临时配置文件是否合法，如果合法则替换会新的配置文件，否则返回错误。
3.3. Overwriting
将staged文件命名为Dest文件的名字，读取staged文件中的内容并将它写入到Dest文件中，该过程实际上就是重新生成一份新的配置文件。staged文件的生成逻辑在函数createStageFile中。
log.Debug("Overwriting target config " + t.Dest)
err := os.Rename(staged, t.Dest)
if err != nil {
if strings.Contains(err.Error(), "device or resource busy") {
log.Debug("Rename failed - target is likely a mount. Trying to write instead")
// try to open the file and write to it
var contents []byte
var rerr error
contents, rerr = ioutil.ReadFile(staged)
if rerr != nil {
return rerr
}
err := ioutil.WriteFile(t.Dest, contents, t.FileMode)
// make sure owner and group match the temp file, in case the file was created with WriteFile
os.Chown(t.Dest, t.Uid, t.Gid)
if err != nil {
return err
}
} else {
return err
}
}
3.4. reload
如果没有指定syncOnly参数并且指定了ReloadCmd则执行reload操作。
if !t.syncOnly && t.ReloadCmd != "" {
if err := t.reload(); err != nil {
return err
}
}
其中t.reload()实现如下：
// reload executes the reload command.
// It returns nil if the reload command returns 0.
func (t *TemplateResource) reload() error {
return runCommand(t.ReloadCmd)
}
t.reload()和t.check()都调用了runCommand函数：
// runCommand is a shared function used by check and reload
// to run the given command and log its output.
// It returns nil if the given cmd returns 0.
// The command can be run on unix and windows.
func runCommand(cmd string) error {
log.Debug("Running " + cmd)
var c *exec.Cmd
if runtime.GOOS == "windows" {
c = exec.Command("cmd", "/C", cmd)
} else {
c = exec.Command("/bin/sh", "-c", cmd)
}
output, err := c.CombinedOutput()
if err != nil {
log.Error(fmt.Sprintf("%q", string(output)))
return err
}
log.Debug(fmt.Sprintf("%q", string(output)))
return nil
}
4. redisClient.WatchPrefix
redisClient.WatchPrefix是当用户设置了watch参数的时候，并且存储后端为redis，则会调用到redis的watch机制。其中redisClient.WatchPrefix是redis存储类型的时候实现了StoreClient接口的WatchPrefix方法。
// The StoreClient interface is implemented by objects that can retrieve
// key/value pairs from a backend store.
type StoreClient interface {
GetValues(keys []string) (map[string]string, error)
WatchPrefix(prefix string, keys []string, waitIndex uint64, stopChan chan bool) (uint64, error)
}
StoreClient是对后端存储类型的抽象，常用的后端存储类型有Etcd和Redis等，不同的后端存储类型GetValues和WatchPrefix的具体实现不同，本文主要分析Redis类型的watch机制。
4.1. WatchPrefix
WatchPrefix的调用函数在monitorPrefix的部分，具体参考：
func (p *watchProcessor) monitorPrefix(t *TemplateResource) {
defer p.wg.Done()
keys := util.AppendPrefix(t.Prefix, t.Keys)
for {
index, err := t.storeClient.WatchPrefix(t.Prefix, keys, t.lastIndex, p.stopChan)
if err != nil {
p.errChan <- err
// Prevent backend errors from consuming all resources.
time.Sleep(time.Second * 2)
continue
}
t.lastIndex = index
if err := t.process(); err != nil {
p.errChan <- err
}
}
}
redis的watch主要通过pub-sub的机制，即WatchPrefix会根据传入的prefix起一个sub的监听机制，而在写入redis的数据的同时需要执行redis的publish操作，channel为符合prefix的值，value为给定命令之一，实际上是给定命令之一，具体是什么命令并没有关系，则会触发watch机制，从而自动更新配置，给定的命令列表如下：
"del", "append", "rename_from", "rename_to", "expire", "set", "incrby", "incrbyfloat", "hset", "hincrby", "hincrbyfloat", "hdel"
sub监听的key的格式如下：
__keyspace@0__:{prefix}/*
如果只是写入redis数据而没有自动执行publish的操作，并不会触发redis的watch机制来自动更新配置。但是如果使用etcd，则etcd的watch机制，只需要用户写入或更新数据就可以自动触发更新配置。
WatchPrefix源码如下：
func (c *Client) WatchPrefix(prefix string, keys []string, waitIndex uint64, stopChan chan bool) (uint64, error) {
if waitIndex == 0 {
return 1, nil
}
if len(c.pscChan) > 0 {
var respChan watchResponse
for len(c.pscChan) > 0 {
respChan = <-c.pscChan
}
return respChan.waitIndex, respChan.err
}
go func() {
if c.psc.Conn == nil {
rClient, db, err := tryConnect(c.machines, c.password, false);
if err != nil {
c.psc = redis.PubSubConn{Conn: nil}
c.pscChan <- watchResponse{0, err}
return
}
c.psc = redis.PubSubConn{Conn: rClient}
go func() {
defer func() {
c.psc.Close()
c.psc = redis.PubSubConn{Conn: nil}
}()
for {
switch n := c.psc.Receive().(type) {
case redis.PMessage:
log.Debug(fmt.Sprintf("Redis Message: %s %s\n", n.Channel, n.Data))
data := string(n.Data)
commands := [12]string{"del", "append", "rename_from", "rename_to", "expire", "set", "incrby", "incrbyfloat", "hset", "hincrby", "hincrbyfloat", "hdel"}
for _, command := range commands {
if command == data {
c.pscChan <- watchResponse{1, nil}
break
}
}
case redis.Subscription:
log.Debug(fmt.Sprintf("Redis Subscription: %s %s %d\n", n.Kind, n.Channel, n.Count))
if n.Count == 0 {
c.pscChan <- watchResponse{0, nil}
return
}
case error:
log.Debug(fmt.Sprintf("Redis error: %v\n", n))
c.pscChan <- watchResponse{0, n}
return
}
}
}()
c.psc.PSubscribe("__keyspace@" + strconv.Itoa(db) + "__:" + c.transform(prefix) + "*")
}
}()
select {
case <-stopChan:
c.psc.PUnsubscribe()
return waitIndex, nil
case r := <- c.pscChan:
return r.waitIndex, r.err
}
}
5. 总结
- confd的作用是通过将配置存放到存储后端，来自动触发更新配置的功能，其中常用的后端有
Etcd和Redis等。 - 不同的存储后端，watch机制不同，例如Etcd只需要更新key便可以触发自动更新配置的操作，而redis除了更新key还需要执行
publish的操作。 - 可以通过配置
check_cmd来校验配置文件是否正确，如果配置文件非法则不会执行自动更新配置和reload的操作，但是当存储后端存入的非法数据，会导致每次校验都是失败的，即使后面新增的配置部分是合法的，所以需要有机制来控制存入存储后端的数据始终是合法的。
参考：