confd

1.简介

1.1.confd

• confd负责自动完成配置同步，基于datastore监听BGP配置的变化，利用模板引擎渲染BIRD配置文件，通知BIRD重新加载，实现声明式管理。

• 

• 注意

  BGP配置基于calico扩展的CRD声明式管理，confd一般基于informer监听

1.2.入口

• calico-node目录的main函数是confd的唯一入口，基于命令决定confd模块是否开启运行，InitConfig初始化的是confd默认配置。

• func main() {
  	...
  
  	if *runConfd {
  		cfg, err := confdConfig.InitConfig(true)
  		...
          
  		cfg.ConfDir = *confdConfDir
  		cfg.KeepStageFile = *confdKeep
  		cfg.Onetime = *confdRunOnce
  		confd.Run(cfg)
  	}
      ...
  }
  

• 注意

  InitConfig会初始化默认配置，基于传入配置调整后运行confd模块

2.运行

2.1.config

• InitConfig()会生成defaultConfig，基于configFile/cmdLine加载的配置内容更新config对象，作为后续confd运行的配置来源。

• // InitConfig initializes the confd configuration by first setting defaults.
  func InitConfig(ignoreFlags bool) (*Config, error) {
      // 1.未设置配置文件,加载默认(/etc/confd/confd.toml)
  	if configFile == "" {
  		if _, err := os.Stat(defaultConfigFile); !os.IsNotExist(err) {
  			configFile = defaultConfigFile
  		}
  	}
  	
      // 2.初始化config对象
  	config := Config{
  		ConfDir:  "/etc/confd",
  		Interval: 600,
  		Prefix:   "",
          // 读取环境变化typha配置
  		Typha:    syncclientutils.ReadTyphaConfig([]string{"CONFD_", "FELIX_", "CALICO_"}),
  	}  
      
  	// 3.基于配置文件内容更新config
  	if configFile != "" {
  		configBytes, err := os.ReadFile(configFile)
  		...
  		_, err = toml.Decode(string(configBytes), &config)
  		...
  	}
  
      // 4.基于命令行参数更新config
  	if !ignoreFlags {
  		// Update config from commandline flags.
  		processFlags(&config)
  	}
  
  
  	return &config, nil
  }
  

• 

• 注意

  confd配置优先由配置文件加载，其次由命令行加载

2.2.start

• confd.Run()基于配置文件初始化及启动监听模块，利用watch监听配置及节点变化，驱动BGP路由生成与更新，确保BGP网络资源正常处理流量。

• func Run(config *config.Config) {
      // 1.模块初始化及激活
      storeClient, err := calico.NewCalicoClient(config)
  	...
  
  	// 2.仅同步一次
  	if config.Onetime {
          // 处理及渲染配置
  		template.Process(templateConfig)
          ...
  		os.Exit(0)
  	}
  	...
  
      // 3.基于watch监听及同步配置
  	processor := template.WatchProcessor(templateConfig, stopChan, doneChan, errChan)
  	go processor.Process()
  	...
  }
  

• 

• 注意

  NewCalicoClient()负责初始化同步模块及watch配置，process负责消费变更更新BIRD配置

2.3.client

• calico.NewCalicoClient()初始化backend，加载待广播cidr及同步器，驱动BGP路由生成、资源同步及事件驱动更新BIRD配置。

• func NewCalicoClient(confdConfig *config.Config) (*client, error) {
  	...
  
  	// 1.etcd/kube backend
  	cc, err := clientv3.New(*clientCfg)
  	...
      
      // 2.获取名为default bgpconfiguration对象
  	cfg, err := cc.BGPConfigurations().Get(context.Background(), globalConfigName, options.GetOptions{})
  	...
      
  	// 3.初始化client
  	c := &client{ client: cc.(backendClientAccessor).Backend()... }
  	...
  
  	// 4.初始化secret watcher
  	c.secretWatcher, err = NewSecretWatcher(c)
  	...
  
  	// 5.加载ENV/config设置的svccidr
  	if clusterCIDR := os.Getenv(envAdvertiseClusterIPs); len(clusterCIDR) != 0 {
  		clusterCIDRs = []string{clusterCIDR}
  	} else if cfg != nil && cfg.Spec.ServiceClusterIPs != nil {
  		for _, c := range cfg.Spec.ServiceClusterIPs {
  			clusterCIDRs = append(clusterCIDRs, c.CIDR)
  		}
  	}
      // 初始化svc路由规则
  	c.onClusterIPsUpdate(clusterCIDRs)
  	...
      
      // 6.加载config设置的externalcidr
  	if cfg != nil && cfg.Spec.ServiceExternalIPs != nil {
  		for _, c := range cfg.Spec.ServiceExternalIPs {
  			externalCIDRs = append(externalCIDRs, c.CIDR)
  		}
  	}
  	// 初始化external路由规则
  	c.onExternalIPsUpdate(externalCIDRs)
  	...
      
      // 7.加载config设置的lbcidr
  	if cfg != nil && cfg.Spec.ServiceLoadBalancerIPs != nil {
  		for _, c := range cfg.Spec.ServiceLoadBalancerIPs {
  			lbCIDRs = append(lbCIDRs, c.CIDR)
  		}
  	}
  	// 初始化lb路由规则
  	c.onLoadBalancerIPsUpdate(lbCIDRs)
  	...
      
      // 8.初始化bgpnode processor
  	c.nodeV1Processor = updateprocessors.NewBGPNodeUpdateProcessor(clientCfg.Spec.K8sUsePodCIDR)
      
      // 9.typha模式中间层初始化
  	if syncclientutils.MustStartSyncerClientIfTyphaConfigured(&confdConfig.Typha, ...) {
  		log.Debug("Using typha syncclient")
      // 10.本地syncer模式初始化
  	} else {
  		// Use the syncer locally.
  		c.syncer = bgpsyncer.New(c.client, c, template.NodeName, clientCfg.Spec)
  		c.syncer.Start()
  	}
  
      // 10.设置相关cidr,
  	if len(clusterCIDRs) != 0 || len(externalCIDRs) != 0 || len(lbCIDRs) != 0 {
          // 初始化路由生成器及启动(失败标记完成/不可用)
  		if c.rg, err = NewRouteGenerator(c); err != nil {
  			c.OnSyncChange(SourceRouteGenerator, true)
  			c.rg = nil
          // 生成静态或动态路由注入route table
  		} else {
  			c.rg.Start()
  		}
      // 标记完成/不可用
  	} else {
  		c.OnSyncChange(SourceRouteGenerator, true)
  	}
  
  	// 11.事件消费处理
  	go func() {
  		for {
  			select {
              // typha/syncer推送
  			case e := <-c.syncerC:
  				switch event := e.(type) {
  				case []api.Update:
  					c.onUpdates(event, false)
  				case api.SyncStatus:
  					c.onStatusUpdated(event)
  				default:
  				}
              // secretMgr推送
  			case <-c.recheckC:
  				c.onUpdates(nil, true)
  			}
  		}
  	}()
  
  	return c, nil
  }
  

• 

• 注意

  typha模式基于代理减少APIServer/Etcd连接，利用server广播路由；syncer模式则为node-to-node广播，冗余连接较多

3.typha

3.1.initial

• MustStartSyncerClientIfTyphaConfigured()负责自动发现typha服务地址及初始化连接，基于typha监听配置变更及广播给watcher。

• // starts a syncer of the requested type if typha is configured to be running.
  func MustStartSyncerClientIfTyphaConfigured(...) bool {
      // 1.初始化discoverer
  	discoverer := discovery.New(
  		discovery.WithAddrOverride(typhaConfig.Addr),
  		discovery.WithInClusterKubeClient(), /* defer creation of a client until its needed. */
  		discovery.WithKubeService(typhaConfig.K8sNamespace, typhaConfig.K8sServiceName),
  	)
      
      // 2.发现typha地址
  	typhaAddrs, err := discoverer.LoadTyphaAddrs()
  	...
      
      // 3.未启用
  	if !discoverer.TyphaEnabled() {
  		return false
  	}
  
  	// 4.初始化typha connection.
  	typhaConnection := syncclient.New(
  		discoverer,
  		myVersion, myHostname, myInfo,
  		cbs,
  		&syncclient.Options{
  			SyncerType:      syncerType,
  			ReadTimeout:     typhaConfig.ReadTimeout,
  			WriteTimeout:    typhaConfig.WriteTimeout,
  			KeyFile:         typhaConfig.KeyFile,
  			CertFile:        typhaConfig.CertFile,
  			CAFile:          typhaConfig.CAFile,
  			ServerCN:        typhaConfig.CN,
  			ServerURISAN:    typhaConfig.URISAN,
  			FIPSModeEnabled: typhaConfig.FIPSModeEnabled,
  		},
  	)
      
      // 5.执行同步
  	typhaConnection.Start(context.Background())
      ...
      
  	go func() {
  		typhaConnection.Finished.Wait()
  	}()
  
  	return true
  }
  

• 注意

  typha后面会进一步介绍，可以理解为client-->typha-->apiserver的中间代理层，基于统一监听减少无效连接

3.2.loadaddr

• d.LoadTyphaAddrs()用于动态发现typha实例地址，基于local-->remote优先级策略提供给typha syncer建立连接。

• // tries to discover the best address(es) to use to connect to Typha.
  func (d *Discoverer) LoadTyphaAddrs() (ts []Typha, err error) {
  	defer func() {
  		d.allKnownAddrs = ts
  	}()
  	ts, err = d.discoverTyphaAddrs()
  	...
      
      // addr过滤
  	for _, f := range d.filters {
  		ts = f(ts)
  		...
  	}
  	return
  }
  
  func (d *Discoverer) discoverTyphaAddrs() ([]Typha, error) {
  	...
  
      // 1.静态配置的addr优先
  	if d.addrOverride != "" {
  		return []Typha{{Addr: d.addrOverride}}, nil
  	}
  
  	// 2.初始化k8sclient
  	if d.k8sClient == nil && d.inCluster {
  		...
  		d.k8sClient, err = kubernetes.NewForConfig(k8sConf)
  		...
  	}
      ...
  
  	// 3.获取calico-typha service对应的endpoint
  	epClient := d.k8sClient.CoreV1().Endpoints(d.k8sNamespace)
  	eps, err := epClient.Get(context.Background(), d.k8sServiceName, v1.GetOptions{})
  	...
  
      // 4.地址解析
  	for _, subset := range eps.Subsets {
          // 4.1.calico-typha端口解析
  		for _, port := range subset.Ports {
  			if port.Name == d.k8sServicePortName {
  				portForOurVersion = port.Port
  				break
  			}
  		}
  		...
  
  		// 4.2.本地及远程地址解析
  		for _, h := range subset.Addresses {
  			typhaAddr := net.JoinHostPort(h.IP, fmt.Sprint(portForOurVersion))
  			if h.NodeName != nil && *h.NodeName == d.nodeName { // is local
  				local = append(local, Typha{Addr: typhaAddr, IP: h.IP, NodeName: h.NodeName})
  			} else {
  				remote = append(remote, Typha{Addr: typhaAddr, IP: h.IP, NodeName: h.NodeName})
  			}
  			candidates++
  		}
  	}
  	...
  
      // 5.地址随机打乱
  	shuffleInPlace(local)
  	shuffleInPlace(remote)
  	// 本地优先
  	addresses = append(local, remote...)
  
  	return addresses, nil
  }
  

• 

• 注意

  本节点的Pod IP优先，其次才是随机打乱的其它节点Pod IP

3.3.start

• typha.start()会基于重试次数及连接配置初始化connection，设置连接相关参数，基于connection连接typha订阅消息及处理。

• func (s *SyncerClient) Start(cxt context.Context) error {
      // 1.重试次数计算[6,2*len(addrs)]
  	maxTries := s.calculateConnectionAttemptLimit(len(s.discoverer.CachedTyphaAddrs()))
  	remainingTries := maxTries
  	cat := discovery.NewConnAttemptTracker(s.discoverer)
      
      // 2.尝试连接
  	for {
  		remainingTries--
  		if remainingTries < 0 {
  			return fmt.Errorf("failed to connect to Typha after %d tries", maxTries)
  		}
          
          // 取下一个地址
  		addr, err := cat.NextAddr()
  		...
          
          // 尝试连接
  		err = s.connect(cxt, addr)
  		if err == nil {
  			break
  		}
  	}
  	...
      
      // 3.启动事件循环处理
  	go s.loop(cxt, cancelFn)
  	...
  
  	go func() {
  		...
     		// 退出关闭typha连接
  		<-cxt.Done()
  		s.connection.Close()
  		...
  	}()
  	return nil
  }
  
  func (s *SyncerClient) connect(cxt context.Context, typhaAddr discovery.Typha) error {
  	...
      // 1.tls conn初始化
  	if s.options.requiringTLS() {
  		...
  		connFunc = func(addr string) (net.Conn, error) {
  			return tls.DialWithDialer(&net.Dialer{Timeout: 10 * time.Second}, "tcp", addr, tlsConfig)
  		}
      // 2.net conn初始化
  	} else {
  		connFunc = func(addr string) (net.Conn, error) {
  			return net.DialTimeout("tcp", addr, 10*time.Second)
  		}
  	}
  	...
      
      s.connection, err = connFunc(typhaAddr.Addr)
  	...
      s.connR = s.connection
      if s.options.DebugLogReads {
          s.connR = readlogger.New(s.connection)
      }
  	...
  
      // 3.设置bufferSize
  	if s.options.ReadBufferSize != 0 {
          // tls.conn会返回底层net.conn
  		tcpConn := extractTCPConn(s.connection)
  		...
          tcpConn.SetReadBuffer(s.options.ReadBufferSize)
  		...
  	}
      
  	s.connInfo = &typhaAddr
  	...
  
  	return nil
  }
  

• 

• 注意

  typha connection初始化完成后，s.loop()会基于connection循环处理typha推送的消息

3.4.loop

• s.loop()负责完整解析conn握手、通信保活及数据同步，基于callback回调实现状态消息及KV消息推送，提供给上层client处理。

• func (s *SyncerClient) loop(cxt context.Context, cancelFn context.CancelFunc) {
  	...
      
  	// 1.初始化gob编解码器
  	s.encoder = gob.NewEncoder(s.connection)
  	s.decoder = gob.NewDecoder(s.connR)
  	...
      
  	// 2.发送Hello握手
  	s.sendMessageToServer(cxt, logCxt, "send hello to server", syncproto.MsgClientHello{ ... })
  	...
  
  	// 3.接收握手响应
  	msg, err := s.readMessageFromServer(cxt, logCxt)
  	...
  	serverHello, ok := msg.(syncproto.MsgServerHello)
  	...
  
      // 4.匹配支持的同步类型
  	serverSyncerType := serverHello.SyncerType
  	if serverSyncerType == "" {
  		serverSyncerType = syncproto.SyncerTypeFelix
  	}
  	if ourSyncerType != serverSyncerType {
  		return
  	}
  
  	// 5.循环处理订阅消息
  	for cxt.Err() == nil {
  		msg, err := s.readMessageFromServer(cxt, logCxt)
  		...
          
  		switch msg := msg.(type) {
          // 5.1.同步状态更新
  		case syncproto.MsgSyncStatus:
  			s.callbacks.OnStatusUpdated(msg.SyncStatus)
          // 5.2.心跳保活响应
  		case syncproto.MsgPing:
  			s.sendMessageToServer(cxt, logCxt, "write pong to server", syncproto.MsgPong{ msg.Timestamp })
  			...
          // 5.3.KV消息
  		case syncproto.MsgKVs:
  			...
              
              // 反序列化KV封装为更新消息
  			for _, kv := range msg.KVs {
  				update, err := kv.ToUpdate()
  				...
  				updates = append(updates, update)
  			}
              
  			s.callbacks.OnUpdates(updates)
          // 解码器重启,切换server要求的压缩算法
  		case syncproto.MsgDecoderRestart:
  			s.restartDecoder(cxt, logCxt, msg)
  			...
          // 重复握手,断开连接
  		case syncproto.MsgServerHello:
  			return
  		}
  	}
  }
  
  // called from the BGP syncer to indicate that the sync status is updated.
  func (c *client) OnStatusUpdated(status api.SyncStatus) {
  	c.syncerC <- status
  }
  
  // called from the BGP syncer to indicate that new updates are available from the Calico datastore.
  func (c *client) OnUpdates(updates []api.Update) {
  	c.syncerC <- updates
  }
  

• 

• 注意

  s.loop()同步的数据会推送到上面分析的client.syncerC，供上层进一步同步处理

4.syncer

4.1.initial

• syncer.New()负责初始化BGP专用同步器，用于监听calico bgp相关资源(BGPPeer/IPPool...)变更，实现配置的实时同步和处理。

• // creates a new BGP v1 Syncer.
  func New(...) api.Syncer {
  	// Create ResourceTypes required for BGP.
  	resourceTypes := []watchersyncer.ResourceType{
  		{
  			ListInterface:   model.ResourceListOptions{Kind: apiv3.KindIPPool},
  			UpdateProcessor: updateprocessors.NewIPPoolUpdateProcessor(),
  		},
  		{
  			ListInterface: model.ResourceListOptions{Kind: apiv3.KindBGPConfiguration},
  		},
  		{
  			ListInterface: model.ResourceListOptions{Kind: libapiv3.KindNode},
  		},
  		{
  			ListInterface: model.ResourceListOptions{Kind: apiv3.KindBGPPeer},
  		},
  		{
  			ListInterface: model.ResourceListOptions{Kind: apiv3.KindBGPFilter},
  		},
  	}
  
  	// If using Calico IPAM, include IPAM resources.
  	if !cfg.K8sUsePodCIDR {
  		resourceTypes = append(resourceTypes, watchersyncer.ResourceType{
  			ListInterface: model.BlockAffinityListOptions{Host: node},
  		})
  	}
  
  	return watchersyncer.New(client, resourceTypes, callbacks)
  }
  
  // New creates a new multiple Watcher-backed api.Syncer.
  func New(client api.Client, resourceTypes []ResourceType, callbacks api.SyncerCallbacks) api.Syncer {
  	rs := &watcherSyncer{
  		watcherCaches: make([]*watcherCache, len(resourceTypes)),
  		results:       make(chan interface{}, 2000),
  		callbacks:     callbacks,
  	}
  	for i, r := range resourceTypes {
  		rs.watcherCaches[i] = newWatcherCache(client, r, rs.results)
  	}
  	return rs
  }
  

• 注意

  watcherSyncer作为核心实现，各类资源注册为watcherCache进行资源监听及缓存结果到results，基于callback调用推送

4.2.start

• ws.start()后台启动所有watcherCache监听对应资源，相关事件经过ws.results缓存及批量合并后执行callback回调推送到上游client。

• func (ws *watcherSyncer) Start() {
  	...
  	go func() {
  		defer ws.wgws.Done()
  		ws.run(ctx)
  	}()
  }
  
  // implements the main syncer loop that loops forever receiving watch events and translating to syncer updates.
  func (ws *watcherSyncer) run(ctx context.Context) {
      // 1.发送初始状态(WaitForDatastore)
      ws.sendStatusUpdate(api.WaitForDatastore)
      
      // 2.启动所有watcher监听资源
  	for _, wc := range ws.watcherCaches {
  		go func(wc *watcherCache) {
  			defer ws.wgwc.Done()
  			wc.run(ctx)
  		}(wc)
  	}
  	...
      
      // 3.消费监听到的消息
  	for result := range ws.results {
          // 3.1.消息归并
  		updates := ws.processResult(updates, result)
  	consolidatationloop:
          // 3.2.批量合并消息
  		for ii := 0; ii < 1000; ii++ {
  			select {
  			case next := <-ws.results:
                  // 未达到上限或出现err追加到updates,否则提前推到ws.results
  				updates = ws.processResult(updates, next)
  			default:
  				break consolidatationloop
  			}
  		}
  		
          // 3.3.结果推送
  		updates = ws.sendUpdates(updates)
  	}
  }
  

• 

• 注意

  wc.run()监听到的资源事件会发送到ws.results管道，主循环会持续消费管道数据，批量合并后发送到上层client.syncerC

4.3.watcher

• wc.run()会维护本地资源缓存wc.resources，利用List-Watch全量同步和增量监听资源变化更新到缓存及推送到ws.results chan。

• // loops performing resync processing until it successfully completes a resync and starts a watcher.
  func (wc *watcherCache) resyncAndCreateWatcher(ctx context.Context) {
  	// 1.退出exist watcher
  	wc.cleanExistingWatcher()
  	...
      
  	for {
  		select {
          // 2.退出exist watcher
  		case <-ctx.Done():
  			wc.cleanExistingWatcher()
  			return
          // 3.间隔周期
  		case <-wc.resyncThrottleC():
  			wc.logger.Debug("Starting main resync loop")
  		}
  
  		// 4.重置间隔周期
  		wc.resyncBlockedUntil = time.Now().Add(MinResyncInterval)
  
          // 5.全量同步
  		if performFullResync {
  			...
  
  			// 5.1.List最新资源
  			l, err := wc.client.List(ctx, wc.resourceType.ListInterface, wc.currentWatchRevision)
  			...
  
  			// 5.2.更新wc.resources及推到ws.results
  			for _, kvp := range l.KVPairs {
  				wc.handleWatchListEvent(kvp)
  			}
  
  			// 5.3.标记同步完成+剩余的oldResources推送deleted event到ws.results
  			wc.finishResync()
  
  			// Store the current watch revision.  This gets updated on any new add/modified event.
  			wc.currentWatchRevision = l.Revision
  
  			// Mark the resync as complete.
  			performFullResync = false
  		}
  
  		// 6.初始化kube/etcd watcher
  		w, err := wc.client.Watch(ctx, wc.resourceType.ListInterface, wc.currentWatchRevision)
  		...
  
  		// Store the watcher and exit back to the main event loop.
  		wc.watch = w
  		return
  	}
  }
  
  // run creates the watcher and loops indefinitely reading from the watcher.
  func (wc *watcherCache) run(ctx context.Context) {
      // 1.初始化watcher
  	wc.resyncAndCreateWatcher(ctx)
  
  mainLoop:
  	for {
  		if wc.watch == nil {
  			// The watcher will be nil if the context cancelled during a resync.
  			break mainLoop
  		}
  		select {
          // 2.退出exist watcher
  		case <-ctx.Done():
  			wc.cleanExistingWatcher()
  			break mainLoop
         	// 3.消费watcher resultChan事件
  		case event, ok := <-wc.watch.ResultChan():
              // 3.1.watcherChan关闭,重新初始化watcher
  			if !ok {
  				wc.resyncAndCreateWatcher(ctx)
  				continue
  			}
  
  			// Handle the specific event type.
  			switch event.Type {
              // 3.2.分发创建/变更事件
  			case api.WatchAdded, api.WatchModified:
  				kvp := event.New
  				wc.handleWatchListEvent(kvp)
              // 3.3.分发删除事件
  			case api.WatchDeleted:
  				// Nil out the value to indicate a delete.
  				kvp := event.Old
  				...
  				kvp.Value = nil
                  
  				wc.handleWatchListEvent(kvp)
              // 3.4.watch失败,重新初始化watcher
  			case api.WatchError:
  				wc.currentWatchRevision = "0"
  				wc.resyncAndCreateWatcher(ctx)
  			default:
  				// Unknown event type - not much we can do other than log.
  				wc.logger.WithField("EventType", event.Type).Errorf("Unknown event type received")
  			}
  		}
  	}
  
  	// 4.watch退出,推送删除事件
  	for _, value := range wc.resources {
  		wc.results <- []api.Update{{
  			UpdateType: api.UpdateTypeKVDeleted,
  			KVPair: model.KVPair{
  				Key: value.key,
  			},
  		}}
  	}
  }
  

• 

• 注意

  wc.watcher支持kube/etcd两种形式，kube watcher走informer那一套，etcd watcher基于etcd watch chan包装