confd

1.router

1.1.initial

• NewRouteGenerator()会初始化路由生成器及svc/endpoint informer，基于监听到的svc/endpoint事件驱动svc路由发布及撤销。

• // NewRouteGenerator initializes a kube-api client and the informers
  func NewRouteGenerator(c *client) (rg *routeGenerator, err error) {
  	// 1.加载nodeName
  	nodename := template.NodeName
  	if n := os.Getenv("CALICO_K8S_NODE_REF"); n != "" {
  		nodename = n
  	}
  
  	// 2.初始化route生成器
  	rg = &routeGenerator{
  		client:                     c,
  		nodeName:                   nodename,
  		svcRouteMap:                make(map[string]map[string]bool),
  		routeAdvertisementRefCount: make(map[string]int),
  		resyncKnownRoutesTrigger:   make(chan struct{}, 1),
  	}
  
  	// 3.初始化k8s client
  	cfgFile := os.Getenv("KUBECONFIG")
  	cfg, err := clientcmd.BuildConfigFromFlags("", cfgFile)
  	...
  	client, err := kubernetes.NewForConfig(cfg)
  	...
  
  	// 4.svc监听处理
  	svcWatcher := cache.NewListWatchFromClient(client.CoreV1().RESTClient(), "services", "",fields.Everything())
  	svcHandler := cache.ResourceEventHandlerFuncs{rg.onSvcAdd, rg.onSvcUpdate, rg.onSvcDelete}
  	rg.svcIndexer, rg.svcInformer =cache.NewIndexerInformer(svcWatcher,&Service{},0,svcHandler,cache.Indexers{})
  
  	// 5.初始化endpoint监听处理
  	epWatcher := cache.NewListWatchFromClient(client.CoreV1().RESTClient(), "endpoints", "",fields.Everything())
  	epHandler := cache.ResourceEventHandlerFuncs{rg.onEPAdd, rg.onEPUpdate, rg.onEPDelete}
  	rg.epIndexer, rg.epInformer = cache.NewIndexerInformer(epWatcher,&Endpoints{},0,epHandler,cache.Indexers{})
  
  	return
  }
  

• 注意

  svc路由及撤销绑定node，发布的路由明确告知外部svc流量的下一跳是nodeIP

1.2.start

• rg.start()会启动svc/endpoint informer监听，根据节点拥有的后端池动态发布或撤销service路由，实现流量配置自动收敛。

• // Start starts the RouteGenerator so that it will monitor Kubernetes services.
  func (rg *routeGenerator) Start() {
  	...
      
      // 1.informer监听
  	go rg.svcInformer.Run(ch)
  	go rg.epInformer.Run(ch)
  
  	// Wait for informers to sync, then notify the main client.
  	go func() {
  		for !rg.svcInformer.HasSynced() || !rg.epInformer.HasSynced() {
  			time.Sleep(100 * time.Millisecond)
  		}
  
  		// 2.标记同步完成
  		rg.client.OnSyncChange(SourceRouteGenerator, true)
  
          // 3.路由发布(client.onUpdates触发)
  		for range rg.resyncKnownRoutesTrigger {
  			rg.resyncKnownRoutes()
  		}
  	}()
  }
  
  func (rg *routeGenerator) resyncKnownRoutes() {
  	// 1.获取所有svc
  	svcIfaces := rg.svcIndexer.List()
      
      // 2.遍历svc广播路由
  	for _, svcIface := range svcIfaces {
  		svc, ok := svcIface.(*v1.Service)
  		if !ok {
  			continue
  		}
          
  		// Update the routes advertised for this service
  		rg.setRouteForSvc(svc, nil)
  	}
  }
  
  // handles the main logic to check if a specified service or endpoint should have its route advertised.
  func (rg *routeGenerator) setRouteForSvc(svc *v1.Service, ep *v1.Endpoints) {
  	// 1.svc/ep均为空不处理
  	if svc == nil && ep == nil {
  		return
  	}
  	...
      
      // 2.svc为空,基于ep获取
  	if svc == nil {
  		// ep received but svc nil
  		if svc, key = rg.getServiceForEndpoints(ep); svc == nil {
  			return
  		}
      // 3.ep为空,基于svc获取
  	} else if ep == nil {
  		// svc received but ep nil
  		if ep, key = rg.getEndpointsForService(svc); ep == nil {
  			return
  		}
  	}
  	...
  
      // 4.路由发布检查
      // a) node label未注入node.kubernetes.io/exclude-from-external-load-balancers=true
      // b) svc是clusterIP/nodePort/lb类型
      // c) clusterIP是有效地址
      // d) loadbalancer+cluster策略+singleIP  √
      // e) local策略+本地有后端   √
  	advertise := rg.advertiseThisService(svc, ep)
      
      // 5.路由发布
  	if advertise {
          // 5.1.生成svc路由
  		routes := rg.getAllRoutesForService(svc)
          // 5.2.路由发布到节点
  		rg.setRoutesForKey(key, routes)
      // 6.路由撤销
  	} else {
          // 6.1.svc已发布路由
  		routes := rg.getAdvertisedRoutes(key)
          // 6.2.路由由节点撤销
  		rg.withdrawRoutesForKey(key, routes)
  	}
  }
  

• 

• 注意

  client.onUpdates执行配置回调更新，会触发一次路由发布或撤销

1.3.advertise

• rg.setRoutesForKey()基于待发布routes和已发布routes计算差异，将过期路由撤销及发布新路由，发布本质上是保存到缓存供后续配置生成。

• // returns all the routes that should be advertised for the given service.
  func (rg *routeGenerator) getAllRoutesForService(svc *v1.Service) []string {
  	...
      
      // 1.允许发布clusterIP
  	if rg.client.AdvertiseClusterIPs() {
  		// Only advertise cluster IPs if we've been told to.
  		routes = append(routes, svc.Spec.ClusterIP)
  	}
  	svcID := fmt.Sprintf("%s/%s", svc.Namespace, svc.Name)
  
      // 2.允许发布的externalP
  	if svc.Spec.ExternalIPs != nil {
  		for _, externalIP := range svc.Spec.ExternalIPs {
  			// Only advertise allowed external IPs
  			if !rg.isAllowedExternalIP(externalIP) {
  				continue
  			}
  			routes = append(routes, externalIP)
  		}
  	}
  
      // 3.允许发布的loadbalancIP
  	if svc.Status.LoadBalancer.Ingress != nil {
  		for _, lbIngress := range svc.Status.LoadBalancer.Ingress {
  			if len(lbIngress.IP) > 0 {
  				// Only advertise allowed LB IPs
  				if !rg.isAllowedLoadBalancerIP(lbIngress.IP) {
  					continue
  				}
  				routes = append(routes, lbIngress.IP)
  			}
  		}
  	}
  
      // 4.补充后缀/32 /128
  	return addFullIPLength(routes)
  }
  
  // associates only the given routes with the given key, and advertises the given routes.
  func (rg *routeGenerator) setRoutesForKey(key string, routes []string) {
  	// 1.获取svc已发布路由
      advertisedRoutes := rg.svcRouteMap[key]
  	...
  
  	// 2.本轮路由没有的已发布路由,撤销
  	for route := range advertisedRoutes {
  		if !contains(routes, route) {
  			rg.withdrawRoute(key, route)
  		}
  	}
  
  	// 3.已发布路由没有的本轮路由,发布
  	for _, route := range routes {
  		// Advertise route if not already advertised for this key.
  		if _, ok := advertisedRoutes[route]; !ok {
  			rg.advertiseRoute(key, route)
  		}
  	}
  }
  

• 

• 注意

  svc可发布的路由基于初始化阶段允许的clusterCidr/externalCidr/loadbalaceCidr决定

2.consume

2.1.event

• typha/syncer监听到资源变化会解析KV推送到client.syncerC，secretMgr监听到secret变化会发送通知信号到client.recheckC。

• func NewCalicoClient(confdConfig *config.Config) (*client, error) {
  	...
  
  	go func() {
  		for {
  			select {
              // typha/syncer推送
  			case e := <-c.syncerC:
  				switch event := e.(type) {
  				case []api.Update:
  					// 更新cache
                      c.onUpdates(event, false)
  				case api.SyncStatus:
                      // 标记syncer同步完成
  					c.onStatusUpdated(event)
  				default:
  				}
              // secretMgr推送
  			case <-c.recheckC:
  				c.onUpdates(nil, true)
  			}
  		}
  	}()
  
  	return c, nil
  }
  

• 注意

  c.recheckC基于信号触发路由重新发布，c.syncerC才是真正的处理事件

2.2.onupdate

• c.onUpdate()会处理datastore的资源变更(BGPPeer/BGPConfig/Node..)，更新本地缓存及重建BGPPeer，触发路由生成器同步。

• func (c *client) onUpdates(updates []api.Update, needUpdatePeersV1 bool) {
  	...
  
  	// 1.cache rev自增
  	c.incrementCacheRevision()
  	...
      
      // 2.event处理
  	for _, u := range updates {
  
  		v3key, ok := u.Key.(model.ResourceKey)
  		...
  
  		// 2.1.calico-node变化
  		if v3key.Kind == libapiv3.KindNode {
  			// Convert to v1 key/value pairs.
  			kvps, err := c.nodeV1Processor.Process(&u.KVPair)
  			...
              
              // v3 Node转换成v1 KV键值(兼容)
  			for _, kvp := range kvps {
                  // DEL事件
  				if kvp.Value == nil {
  					// 清理节点IP缓存
  					nodeIPv4, nodeIPv6, _, _ := c.nodeToBGPFields(v3key.Name)
  					delete(c.nodeIPs, nodeIPv4)
  					delete(c.nodeIPs, nodeIPv6)
  
  					// 清理cache
  					if c.updateCache(api.UpdateTypeKVDeleted, kvp) {
  						needUpdatePeersV1 = true
  					}
                  // Add/Update事件
  				} else {
  					// Check if the node already has IPs in our node IP cache.
  					oldNodeIPv4, oldNodeIPv6, _, _ := c.nodeToBGPFields(v3key.Name)
  
  					// 更新cache
  					if c.updateCache(u.UpdateType, kvp) {
  						needUpdatePeersV1 = true
  					}
  
  					// Add the node IPs to our node IP cache.
  					nodeIPv4, nodeIPv6, _, _ := c.nodeToBGPFields(v3key.Name)
                      
                      // nodeIP
  					if oldNodeIPv4 != "" && oldNodeIPv4 != nodeIPv4 {
  						// IPv4 address is updated, remove the old IPv4 address.
  						delete(c.nodeIPs, oldNodeIPv4)
  					}
  					if oldNodeIPv6 != "" && oldNodeIPv6 == nodeIPv6 {
  						// IPv6 address is updated, remove the old IPv6 address.
  						delete(c.nodeIPs, oldNodeIPv6)
  					}
  					if nodeIPv4 != "" {
  						// There is an IPv4 address for this node.
  						c.nodeIPs[nodeIPv4] = struct{}{}
  					}
  					if nodeIPv6 != "" {
  						// There is an IPv6 address for this node.
  						c.nodeIPs[nodeIPv6] = struct{}{}
  					}
  				}
  			}
  
  			// DEL事件
  			if u.Value == nil {
  				// 清理node label缓存
  				if c.nodeLabelManager.nodeExists(v3key.Name) {
  					c.nodeLabelManager.deleteNode(v3key.Name)
  					needUpdatePeersV1 = true
  				}
              // ADD/Update事件
  			} else {
  				// 转换为calico-node
  				v3res, ok := u.Value.(*libapiv3.Node)
                  ...
  
                  // 缓存labels
  				if changed := c.nodeLabelManager.setLabels(v3key.Name, v3res.Labels); changed {
  					needUpdatePeersV1 = true
  					
                      // 本节点
  					if v3key.Name == template.NodeName && c.rg != nil {
  						// 标记触发svc路由发布
  						needServiceAdvertisementUpdates = true
  					}
  
  				}
  			}
  		}
  
          // 2.2.BGPPeer资源变化
  		if v3key.Kind == apiv3.KindBGPPeer {
  			// DEL事件
  			if u.Value == nil || u.UpdateType == api.UpdateTypeKVDeleted {
                  // 清理bgppeer缓存
  				delete(c.bgpPeers, v3key.Name)
              // ADD/Update事件
  			} else if v3res, ok := u.Value.(*apiv3.BGPPeer); ok {
  				c.bgpPeers[v3key.Name] = v3res
  			} else {
  				continue
  			}
  
  			// Note need to recompute equivalent v1 peerings.
  			needUpdatePeersV1 = true
  		}
  
          // 2.3.BGPFilter资源变化
  		if v3key.Kind == apiv3.KindBGPFilter {
  			needUpdatePeersV1 = true
  		}
  	}
  
  	// 3.event再处理
  	for _, u := range updates {
          // BGPConfiguration资源
  		if v3key, ok := u.Key.(model.ResourceKey); ok && v3key.Kind == apiv3.KindBGPConfiguration {
  			v3res, _ := u.KVPair.Value.(*apiv3.BGPConfiguration)
              // 更新cache配置
  			c.updateBGPConfigCache(v3key.Name, v3res, &needServiceAdvertisementUpdates, &needUpdatePeersV1, &needUpdatePeersReasons)
  		}
          
          // 更新cache其它配置
  		c.updateCache(u.UpdateType, &u.KVPair)
  	}
  
  	// 4.BGPPeer重新计算
  	if needUpdatePeersV1 {
  		// 标记旧secret过期
  		if c.secretWatcher != nil {
  			c.secretWatcher.MarkStale()
  		}
  		
          // 计算BGPPeer
  		c.updatePeersV1()
  
  		// 更新node-to-node mesh password缓存
  		if c.globalBGPConfig != nil {
  			c.getNodeMeshPasswordKVPair(c.globalBGPConfig, model.GlobalBGPConfigKey{})
  		}
  
  		// 基于BGPPeer secret清理无用secret
  		if c.secretWatcher != nil {
  			c.secretWatcher.SweepStale()
  		}
  	}
  
  	// svc路由发布
  	if needServiceAdvertisementUpdates {
          // 加载路由生成器
  		if c.rg == nil {
  			// If this is the first time we've needed to start the route generator, then do so here.
  			if c.rg, err = NewRouteGenerator(c); err != nil {
  				c.rg = nil
  			} else {
  				c.rg.Start()
  			}
  		}
  		...
          
  		// 加载externalIPs
  		if len(c.cache["/calico/bgp/v1/global/svc_external_ips"]) > 0 {
  			externalIPs = strings.Split(c.cache["/calico/bgp/v1/global/svc_external_ips"], ",")
  		}
  		c.onExternalIPsUpdate(externalIPs)
  		...
          
          // 加载clusterIPs
  		if len(c.cache["/calico/bgp/v1/global/svc_cluster_ips"]) > 0 {
  			clusterIPs = strings.Split(c.cache["/calico/bgp/v1/global/svc_cluster_ips"], ",")
  		}
  		c.onClusterIPsUpdate(clusterIPs)
  		...
          
  		// 加载loadbalanceIPs 
  		if len(c.cache["/calico/bgp/v1/global/svc_loadbalancer_ips"]) > 0 {
  			loadBalancerIPs = strings.Split(c.cache["/calico/bgp/v1/global/svc_loadbalancer_ips"], ",")
  		}
  		c.onLoadBalancerIPsUpdate(loadBalancerIPs)
  
          // 触发路由更新
  		if c.rg != nil {
  			// Trigger the route generator to recheck and advertise or withdraw
  			// node-specific routes.
  			c.rg.TriggerResync()
  		}
  	}
  
  	// 通知watcher更新
  	c.onNewUpdates()
  }
  

• 

• 注意

  client.onUpdates()本质将BGPPeer/Node/BGPConfiguration...原生资源及转换的内部格式更新到cache供后续配置生成使用

2.3.updatepeer

• c.updatePeersV1()会将BGPConfiguration/BGPPeer配置转换为Felix/Bird使用的v1 peer数据模型，对比缓存后增量下发变化配置。

• func (c *client) updatePeersV1() {
  	...
  
  	// 1.注册peersV1回调
  	emit := func(key model.Key, peer *bgpPeer) {
          // 1.1.计算V1路径(/calico/bgp/v1/global/peer_v4/10.0.0.1)
  		k, err := model.KeyToDefaultPath(key)
  		...
  
  		if _, ok := peersV1[k]; ok {
  			return
  		}
  
  		// 1.2.nodePeer
  		if nodeKey, ok := key.(model.NodeBGPPeerKey); ok {
              // 先检查global peer(优先)
  			globalKey := model.GlobalBGPPeerKey{PeerIP: nodeKey.PeerIP, Port: nodeKey.Port}
  			globalPath, _ := model.KeyToDefaultPath(globalKey)
  			if _, ok = peersV1[globalPath]; ok {
  				return
  			}
  		}
  
  		// 1.3.序列化及注册
  		value, err := json.Marshal(peer)
  		...
  		peersV1[k] = string(value)
  	}
  
      // 2.跑两轮生成正向peer(全局/非全局)——local-->peer
  	for _, globalPass := range []bool{true, false} {
  		for _, v3res := range c.bgpPeers {
              // 2.1.Peer作用域检查
  			if globalPass != ((v3res.Spec.NodeSelector == "") && (v3res.Spec.Node == "")) {
  				continue
  			}
  			...
              
              // 2.2.非全局Peer对应nodeName
  			if v3res.Spec.NodeSelector != "" {
  				localNodeNames = c.nodeLabelManager.nodesMatching(v3res.Spec.NodeSelector)
  			} else if v3res.Spec.Node != "" {
  				localNodeNames = []string{v3res.Spec.Node}
  			}
  			...
  
              // 2.3.基于标签限制邻居
  			if v3res.Spec.PeerSelector != "" {
                  // 获取邻居节点构造Peer
  				for _, peerNodeName := range c.nodeLabelManager.nodesMatching(v3res.Spec.PeerSelector) {
  					peers = append(peers, c.nodeAsBGPPeers(peerNodeName, true, true, v3res)...)
  				}
              // 2.4.未限制邻居
  			} else {
  				// 解析ip-port
  				ip, port := parseIPPort(v3res.Spec.PeerIP)
  				if ip == nil {
  					continue
  				}
  
  				// port为空
  				if port == 0 {
  					// 匹配IP/ASNum满足的节点
  					nodeNames := c.nodesWithIPPortAndAS(host, v3res.Spec.ASNumber, port)
  					if len(nodeNames) != 0 {
                          // 优先使用calico-node监听的BGP端口
  						if nodePort, ok := c.nodeListenPorts[nodeNames[0]]; ok {
  							port = nodePort
                          // 回退使用Global BGP端口
  						} else if c.globalListenPort != 0 {
  							port = c.globalListenPort
  						}
  					}
  				}
  
  				// 检查peer是不是calico-node
  				_, isCalicoNode := c.nodeIPs[host]
  				...
                  
  				// 设置asNum
  				if v3res.Spec.NumAllowedLocalASNumbers != nil {
  					numLocalAS = *v3res.Spec.NumAllowedLocalASNumbers
  				}
  				...
                  
                  // 设置TTL策略
  				if v3res.Spec.TTLSecurity != nil {
  					ttlSecurityHopCount = *v3res.Spec.TTLSecurity
  				}
  				...
                  
                  // 解析路由可达地址
  				if v3res.Spec.ReachableBy != "" {
  					reachableByAddr := cnet.ParseIP(v3res.Spec.ReachableBy)
  					if reachableByAddr == nil {
  						continue
  					}
  
  					if reachableByAddr.Version() != ip.Version() {
  						continue
  					}
  					reachableBy = v3res.Spec.ReachableBy
  				}
  
  				peers = append(peers, &bgpPeer{
  					PeerIP:          *ip,
  					ASNum:           v3res.Spec.ASNumber,
  					SourceAddr:      string(v3res.Spec.SourceAddress),
  					Port:            port,
  					KeepNextHop:     v3res.Spec.KeepOriginalNextHop,
  					CalicoNode:      isCalicoNode,
  					TTLSecurity:     ttlSecurityHopCount,
  					Filters:         v3res.Spec.Filters,
  					NumAllowLocalAS: numLocalAS,
  					ReachableBy:     reachableBy,
  				})
  			}
  
  			if len(peers) == 0 {
  				continue
  			}
  
              // 设置peer sourceAddr/password/restartTime
  			c.setPeerConfigFieldsFromV3Resource(peers, v3res)
  
              // 注册到peersV1
  			for _, peer := range peers {
  				if globalPass {
  					key := model.GlobalBGPPeerKey{PeerIP: peer.PeerIP, Port: peer.Port}
  					emit(key, peer)
  				} else {
  					for _, localNodeName := range localNodeNames {
  						key := NodeBGPPeerKey{Nodename: localNodeName, PeerIP: peer.PeerIP, Port: peer.Port}
  						emit(key, peer)
  					}
  				}
  			}
  		}
  	}
  
      // 3.生成反向peer(peer-->local)
  	for _, v3res := range c.bgpPeers {
  		...
          
  		// 3.1.获取作为local的peerNode
  		if v3res.Spec.PeerSelector != "" {
  			localNodeNames = c.nodeLabelManager.nodesMatching(v3res.Spec.PeerSelector)
  			// Peering on label selector, so we should reverse the peering over IPv4 and IPv6.
  			includeV4 = true
  			includeV6 = true
          // 3.2.基于IP/Port匹配peerNode
  		} else {
  			ip, port := parseIPPort(v3res.Spec.PeerIP)
  			localNodeNames = c.nodesWithIPPortAndAS(ip, v3res.Spec.ASNumber, port)
  			if strings.Contains(ip, ":") {
  				includeV6 = true
  			} else {
  				includeV4 = true
  			}
  		}
  
  		// Skip peer computation if there are no local nodes.
  		if len(localNodeNames) == 0 {
  			continue
  		}
  		...
          
  		// 3.3.获取作为peer的localNode
  		if v3res.Spec.NodeSelector != "" {
  			peerNodeNames = c.nodeLabelManager.nodesMatching(v3res.Spec.NodeSelector)
  		} else if v3res.Spec.Node != "" {
  			peerNodeNames = []string{v3res.Spec.Node}
  		} else {
  			peerNodeNames = c.nodeLabelManager.nodesMatching("all()")
  		}
  
  		if len(peerNodeNames) == 0 {
  			continue
  		}
  		...
  
          // 3.4.生成peer(peer->local)
  		for _, peerNodeName := range peerNodeNames {
  			peers = append(peers, c.nodeAsBGPPeers(peerNodeName, includeV4, includeV6, v3res)...)
  		}
          
  		if len(peers) == 0 {
  			continue
  		}
  
          // 3.5.设置peer sourceAddr/password/restartTime
  		c.setPeerConfigFieldsFromV3Resource(peers, v3res)
  
          // 注册到peersV1
  		for _, peer := range peers {
  			for _, localNodeName := range localNodeNames {
  				key := model.NodeBGPPeerKey{Nodename: localNodeName, PeerIP: peer.PeerIP, Port: peer.Port}
  				emit(key, peer)
  			}
  		}
  	}
  
  	// 4.peer状态对账
  	for k, value := range c.peeringCache {
  		newValue, ok := peersV1[k]
          // 过期的peerCache
  		if !ok {
  			// This cache entry should be deleted.
  			delete(c.peeringCache, k)
  			c.keyUpdated(k)
          // peerCache更新
  		} else if newValue != value {
  			// This cache entry should be updated.
  			c.peeringCache[k] = newValue
  			c.keyUpdated(k)
  			delete(peersV1, k)
  		} else {
  			// Value in cache is already correct.  Delete from peersV1 so that we
  			// don't generate a spurious keyUpdated for this key.
  			delete(peersV1, k)
  		}
  	}
      
  	// 5.增量的peer
  	for k, newValue := range peersV1 {
  		c.peeringCache[k] = newValue
  		c.keyUpdated(k)
  	}
  }
  

• 

• 注意

  Peer状态更新本质是基于BGPPeer获取本地和远端节点，生成最新Peer对象更新PeerCache

3.process

3.1.sync

• template.Process()会加载配置文件的resource template模板，基于client监听的资源变化和配置模板渲染新的上下文进行替换。

• func Process(config Config) error {
  	// 1.template resource加载
  	ts, err := getTemplateResources(config)
  	...
  
  	// 2.更新client.revPrefix
  	setClientPrefixes(config, ts)
      ...
  
      // 3.渲染及替换配置
  	for _, t := range ts {
  		t.process("")
          ...
  	}
      
  	return lastErr
  }
  
  func getTemplateResources(config Config) ([]*TemplateResource, error) {
  	...
  
      // 1.检查配置目录
  	if !isFileExist(config.ConfDir) {
  		return nil, nil
  	}
      
      // 2.加载所有.toml文件路径
  	paths, err := recursiveFindFiles(config.ConfigDir, "*toml")
  	...
  
  	if len(paths) < 1 {
  		log.Warning("Found no templates")
  	}
  
      // 3.遍历加载.toml配置
  	for _, p := range paths {
  		t, err := NewTemplateResource(p, config)
  		...
          
  		templates = append(templates, t)
  	}
  	return templates, lastError
  }
  
  // process is a convenience function that wraps calls to the three main tasks
  // required to keep local configuration files in sync.
  func (t *TemplateResource) process(key string) error {
      // 1.获取dest文件权限
  	t.setFileMode()
      ...
      
      // 2.渲染变量上下文
  	t.setVars()
      ...
      
      // 3.生成临时文件
  	t.createStageFile()
      ...
      
      // 4.配置文件替换+reload
  	t.sync(key)
      ...
      
  	return nil
  }
  

• 

• syncOnce.process会加载配置目录的.toml模板，基于模板解析获取src conf及client.cache配置内容渲染到临时文件替换目标

3.2.watch

• p.Process()会加载template resource进行初次初始化，之后监听revPrefix事件(route/revPrefix push)，进行后续配置同步。

• func (p *watchProcessor) Process() {
  	...
      
  	// 1.template resources加载
  	ts, err := getTemplateResources(p.config)
  	...
  
  	// 2.更新client.revPrefix
  	setClientPrefixes(p.config, ts)
      ...
  
  	// 3.激活template监控
  	for _, t := range ts {
  		t := t
  		p.wg.Add(1)
  		go p.monitorPrefix(t)
  	}
  	p.wg.Wait()
  }
  
  func (p *watchProcessor) monitorPrefix(t *TemplateResource) {
  	...
      
  	for {
  		// 1.阻塞监听revPrefix变化的key
  		key, err := t.storeClient.WatchPrefix(t.Prefix, t.ExpandedKeys, revision, p.stopChan)
  		...
  
  		for {
              // 2.获取最新已同步版本
  			revision = t.storeClient.GetCurrentRevision()
              
              // 3.重新渲染bird配置
  			if err = t.process(key); err == nil {
  				break
  			}
              ...
              
              // 4.间隔[250ms,5s]
  			time.Sleep(retryInterval)
  			retryInterval *= 2
  			if retryInterval > maxProcessRetryInterval {
  				retryInterval = maxProcessRetryInterval
  			}
  		}
  	}
  }
  
  // called from confd. It blocks waiting for updates to the data which have any of the requested set of prefixes.
  func (c *client) WatchPrefix(prefix string, keys []string, lastRevision uint64, stopChan chan bool) (...) {
  	...
  
      // 1.初始化加载
  	if lastRevision == 0 {
  		// If this is the first iteration, we always exit to ensure we render with the initial
  		// synced settings.
  		return "", nil
  	}
  
  	for {
  		// 2.循环监听更新的key
  		for _, key := range keys {
  			rev, ok := c.revisionsByPrefix[key]
  			...
  
  			if rev > lastRevision {
  				return key, nil
  			}
  		}
  
  		// 3.route/revPrefix更新会被唤醒
  		c.watcherCond.Wait()
  	}
  }
  

• 

• 注意

  route/revPrefix更新会执行c.watcherCond.Broadcast()唤醒WatchPrefix()进行配置更新

3.3.conf

• confd基于datastore BGP配置的变化，诸如BGPPeer、ASNum用模板引擎渲染出BIRD相关配置，基于配置文件替换和BIRD重载实现热加载。

• # /etc/calico/confd/config/bird.cfg
  router id 192.168.1.10;
  
  protocol kernel {
      scan time 2;
      import all;
      export all;
  }
  
  protocol static {
      route 10.244.0.0/26 blackhole;  # block CIDR 聚合
  }
  
  template bgp bgp_template {
      local as 64512;
      graceful restart;
  }
  
  protocol bgp Node_192_168_1_20 from bgp_template {
      neighbor 192.168.1.20 as 64512;
  }
  
  protocol bgp Node_192_168_1_30 from bgp_template {
      neighbor 192.168.1.30 as 64512;
  }
  

• 注意

  neighbor本质上是BGP邻居，也就是相互交换路由的对端