【kubernetes/k8s源码分析】calico node confd源码分析

刺骨的言语ヽ痛彻心扉 2021-11-16 13:50 279阅读 0赞

github: [https://github.com/kelseyhightower/confd][https_github.com_kelseyhightower_confd]

fork: [https://github.com/projectcalico/confd][https_github.com_projectcalico_confd]

# WHAT confd #

confd：根据etcd上状态信息，与本地模板，生成并更新BIRD配置

通过lsdir获取当前目录下的所有子目录。第一层子目录为域名，根据域名即可生成acl规则、规则使用、后端名称等数据。  
再重新通过瓶装域名目录，对域名目录执行lsdir，读取目录下的每个主机名，创建后端的server条目（一个域名下的负载均衡后段服务器），同时获取挂在这个目录下的属性键值对

# WHY confd #

当系统变的复杂，配置项越来越多，一方面配置管理变得繁琐，另一方面配置修改后需要重新上线同样十分痛苦。这时候，需要有一套集中化配置管理系统，一方面提供统一的配置管理，另一方面提供配置变更的自动下发，及时生效。

统一配置管理系统，常见的：zookeeper、etcd、consul、git等

# 1. InitConfig 函数 #

首先初始化 confd 配置使用默认设置，从 confd 配置文件覆盖参数，从环境变量覆盖配置参数，最后从命令行参数覆盖配置参数

// 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(ignoreFlags bool) (*Config, error) {
    	if configFile == "" {
    		if _, err := os.Stat(defaultConfigFile); !os.IsNotExist(err) {
    			configFile = defaultConfigFile
    		}
    	}

###     1.1 首先使用默认设置配置参数 ###

// Set defaults.
    config := Config{
    	ConfDir:  "/etc/confd",
    	Interval: 600,
    	Prefix:   "",
    	Typha: TyphaConfig{
    		// Non-zero defaults copied from <felix>/config/config_params.go.
    		K8sNamespace: "kube-system",
    		ReadTimeout:  30 * time.Second,
    		WriteTimeout: 10 * time.Second,
    	},
    }

###     1.2 如果指定配置文件则从配置文件覆盖参数 ###

// Update config from the TOML configuration file.
    if configFile == "" {
    	log.Info("Skipping confd config file.")
    } else {
    	log.Info("Loading " + configFile)
    	configBytes, err := ioutil.ReadFile(configFile)
    	if err != nil {
    		return nil, err
    	}
    	_, err = toml.Decode(string(configBytes), &config)
    	if err != nil {
    		return nil, err
    	}
    }

###     1.3 迭代从命令行参数读取配置参数进行覆盖 ###

// processFlags iterates through each flag set on the command line and
    // overrides corresponding configuration settings.
    func processFlags(config *Config) {
    	log.Info("Processing command line flags")
    	v := ConfigVisitor{config: config}
    	flag.Visit(v.setConfigFromFlag)
    }
    
    type ConfigVisitor struct {
    	config *Config
    }
    
    func (c *ConfigVisitor) setConfigFromFlag(f *flag.Flag) {
    	switch f.Name {
    	case "confdir":
    		c.config.ConfDir = confdir
    	case "interval":
    		c.config.Interval = interval
    	case "noop":
    		c.config.Noop = noop
    	case "prefix":
    		c.config.Prefix = prefix
    	case "sync-only":
    		c.config.SyncOnly = syncOnly
    	case "calicoconfig":
    		c.config.CalicoConfig = calicoconfig
    	case "onetime":
    		c.config.Onetime = onetime
    	case "keep-stage-file":
    		c.config.Onetime = keepStageFile
    	}
    }

###     1.4 从环境变量获取参数进行配置覆盖 ###

环境变量前缀 CONFD\_ FELIX\_ CALICO\_，中间 TYPHA

func readTyphaConfig(typhaConfig *TyphaConfig) {
    	// When Typha is in use, there will already be variables prefixed with FELIX_, so it's
    	// convenient if confd honours those too.  However there may use cases for confd to
    	// have independent settings, so honour CONFD_ also.  Longer-term it would be nice to
    	// coalesce around CALICO_, so support that as well.
    	supportedPrefixes := []string{"CONFD_", "FELIX_", "CALICO_"}
    	kind := reflect.TypeOf(*typhaConfig)
    	for ii := 0; ii < kind.NumField(); ii++ {
    		field := kind.Field(ii)
    		nameUpper := strings.ToUpper(field.Name)
    		for _, prefix := range supportedPrefixes {
    			varName := prefix + "TYPHA" + nameUpper
    			if value := os.Getenv(varName); value != "" && value != "none" {
    				log.Infof("Found %v=%v", varName, value)
    				if field.Type.Name() == "Duration" {
    					seconds, err := strconv.ParseFloat(value, 64)
    					if err != nil {
    						log.Error("Invalid float")
    					}
    					duration := time.Duration(seconds * float64(time.Second))
    					reflect.ValueOf(typhaConfig).Elem().FieldByName(field.Name).Set(reflect.ValueOf(duration))
    				} else {
    					reflect.ValueOf(typhaConfig).Elem().FieldByName(field.Name).Set(reflect.ValueOf(value))
    				}
    				break
    			}
    		}
    	}
    }

# 2. Run 函数 #

func Run(config *config.Config) {
    	log.Info("Starting calico-confd")
    	storeClient, err := calico.NewCalicoClient(config)
    	if err != nil {
    		log.Fatal(err.Error())
    	}

###     2.1 NewCalicoClient ###

路径 kelseyhightower/confd/pkg/backends/calico/client.go

**     2.1.1 loadClientConfig **

如果指定配置文件则从配置文件读取配置参数，未指定则从环境变量获取配置参数

// LoadClientConfig loads the ClientConfig from the specified file (if specified)
    // or from environment variables (if the file is not specified).
    func LoadClientConfig(filename string) (*CalicoAPIConfig, error) {
    
    	// Override / merge with values loaded from the specified file.
    	if filename != "" {
    		b, err := ioutil.ReadFile(filename)
    		if err != nil {
    			return nil, err
    		}
    
    		c, err := LoadClientConfigFromBytes(b)
    		if err != nil {
    			return nil, fmt.Errorf("syntax error in %s: %v", filename, err)
    		}
    		return c, nil
    	}
    	return LoadClientConfigFromEnvironment()
    }

**    2.1.2 使用 etcdv3 或者 kubernetes 作为 backend**

// Query the current BGP configuration to determine if the node to node mesh is enabled or
    // not.  If it is we need to monitor all node configuration.  If it is not enabled then we
    // only need to monitor our own node.  If this setting changes, we terminate confd (so that
    // when restarted it will start watching the correct resources).
    cc, err := clientv3.New(*config)
    if err != nil {
    	log.Errorf("Failed to create main Calico client: %v", err)
    	return nil, err
    }

**  2.1.3 查询名为 default 资源为 bgpconfigurations**

cfg, err := cc.BGPConfigurations().Get(
    	context.Background(),
    	"default",
    	options.GetOptions{},
    )
    if _, ok := err.(lerr.ErrorResourceDoesNotExist); err != nil && !ok {
    	// Failed to get the BGP configuration (and not because it doesn't exist).
    	// Exit.
    	log.Errorf("Failed to query current BGP settings: %v", err)
    	return nil, err
    }

**    2.1.4 实例化模板配置 Config**

templateConfig := template.Config{
    	ConfDir:       config.ConfDir,
    	ConfigDir:     filepath.Join(config.ConfDir, "conf.d"),
    	KeepStageFile: config.KeepStageFile,
    	Noop:          config.Noop,
    	Prefix:        config.Prefix,
    	SyncOnly:      config.SyncOnly,
    	TemplateDir:   filepath.Join(config.ConfDir, "templates"),
    	StoreClient:   storeClient,
    }

# 3. WatchProcessor 实例化 #

func WatchProcessor(config Config, stopChan, doneChan chan bool, errChan chan error) Processor {
    	return &watchProcessor{config, stopChan, doneChan, errChan, sync.WaitGroup{}}
    }

# 4. sync 函数 #

监控同步配置，如果更新过则重新 reload 命令

// sync compares the staged and dest config files and attempts to sync them
    // if they differ. sync will run a config check command if set before
    // overwriting the target config file. Finally, sync will run a reload command
    // if set to have the application or service pick up the changes.
    // It returns an error if any.
    func (t *TemplateResource) sync() error {
    	staged := t.StageFile.Name()
    	if t.keepStageFile {
    		log.Info("Keeping staged file: " + staged)
    	} else {
    		defer func() {
    			if e := os.Remove(staged); e != nil {
    				if !os.IsNotExist(e) {
    					// Just log the error but don't carry it up the calling stack.
    					log.WithError(e).WithField("filename", staged).Error("error removing file")
    				} else {
    					log.Debugf("Ignore not exists err. %s", e.Error())
    				}
    			}
    		}()
    	}

[https_github.com_kelseyhightower_confd]: https://github.com/kelseyhightower/confd
[https_github.com_projectcalico_confd]: https://github.com/projectcalico/confd