confd源码分析

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的操作，但是当存储后端存入的非法数据，会导致每次校验都是失败的，即使后面新增的配置部分是合法的，所以需要有机制来控制存入存储后端的数据始终是合法的。

参考：

https://github.com/kelseyhightower/confd/tree/v0.16.0

Feedback

Was this page helpful?

Glad to hear it! Please tell us how we can improve.

Sorry to hear that. Please tell us how we can improve.

最后修改 July 17, 2022: Fix/bug (#1) (d9e9197)