routeTable

1.简介

1.1.定义

• routeTable是calico路由状态调谐工具，根据期望路由状态和内核实际路由状态计算路由差异，基于netlink/conntrack调整路由相关规则。

• // RouteTable manages calico routes for a specific table.
  type RouteTable struct {
      ...
  	
  	// 强制同步标记
  	reSync                bool
      // 接口更新类型(add/update/del)
  	ifaceNameToUpdateType map[string]updateType
      // 接口正则匹配(cali*)
  	ifacePrefixRegexp     *regexp.Regexp
      // 无设备路由处理标记
  	includeNoInterface    bool
  
      // 接口应有路由(veth-->cidr-->target)
  	ifaceNameToTargets             map[string]map[ip.CIDR]Target
      // 接口L2路由(veth-->arp/neighbor/mac...)
  	ifaceNameToL2Targets           map[string][]L2Target
  	...
      
      // 暂存未应用的接口变更
  	pendingIfaceNameToDeltaTargets map[string]map[ip.CIDR]*Target
  	pendingIfaceNameToL2Targets    map[string][]L2Target
  
      // route删除,待清理的tcp/udp conntrack state
  	pendingConntrackCleanups map[ip.Addr]chan struct{}
  
  	// vxlan route标记
  	vxlan bool
  
      // route流量的出网卡IP
  	deviceRouteSourceAddress net.IP
  
      // route标记(calico创建的路由)
  	deviceRouteProtocol  netlink.RouteProtocol
      // 清理非calico route标记
  	removeExternalRoutes bool
  
      // netlink连接工具
  	nl *handlemgr.HandleManager
  
  	// route table编号
  	tableIndex int
  	...
      
      // contrack连接工具
  	conntrack         conntrackIface
  	...
      
      // 健康状态回调
  	livenessCallback func()
  
  	// 路由优雅删除时间
  	routeCleanupGracePeriod time.Duration
  }
  

• 注意

  route变更会暂存到pendingIfaceNameToDeltaTargets和pendingIfaceNameToL2Targets，利用netlink修改路由

1.2.初始化

• routeTable.New()会基于NewWithShims()初始化routeTable工具，构造netlink/conntrack连接Linux内核调整路由流量。

• func New(interfaceRegexes []string, ipVersion uint8, vxlan bool, netlinkTimeout time.Duration...) *RouteTable {
  	return NewWithShims(
  		interfaceRegexes,
  		ipVersion,
  		netlinkshim.NewRealNetlink,
  		vxlan,
  		netlinkTimeout,
  		addStaticARPEntry,
  		conntrack.New(),
  		timeshim.RealTime(),
  		deviceRouteSourceAddress,
  		deviceRouteProtocol,
  		removeExternalRoutes,
  		tableIndex,
  		opReporter,
  		featureDetector,
  		opts...,
  	)
  }
  
  // NewWithShims is a test constructor, which allows netlink, arp and time to be replaced by shims.
  func NewWithShims(interfaceRegexes []string, ipVersion uint8, newNetlinkHandle func() (...)...) *RouteTable {
  	...
  
      // 决定calico管理的网卡
  	for _, interfaceRegex := range interfaceRegexes {
          // 无设备路由
  		if interfaceRegex == InterfaceNone {
  			includeNoOIF = true
          // 网卡设备匹配(cali.*)
  		} else {
  			filteredRegexes = append(filteredRegexes, interfaceRegex)
  		}
  	}
  
      // route table索引
  	if tableIndex == 0 {
          // 0xfe
  		tableIndex = unix.RT_TABLE_MAIN
  	}
  	...
      
  	// Create a regexp matching the interfaces this route table manages.
  	if len(filteredRegexes) == 0 && len(interfaceRegexes) > 0 {
          // 无有效接口匹配
  		logCxt.Info("No interface matches required for routetable")
  	} else {
  		// 接口匹配转为正则
  		ifaceNamePattern := strings.Join(filteredRegexes, "|")
  		ifacePrefixRegexp = regexp.MustCompile(ifaceNamePattern)
  	}
  
  	family := netlink.FAMILY_V4
  	if ipVersion == 6 {
  		family = netlink.FAMILY_V6
  	}
  
  	rt := &RouteTable{
  		logCxt:                         logCxt,
  		ipVersion:                      ipVersion,
  		netlinkFamily:                  family,
  		ifacePrefixRegexp:              ifacePrefixRegexp,
  		includeNoInterface:             includeNoOIF,
  		ifaceNameToTargets:             map[string]map[ip.CIDR]Target{},
  		ifaceNameToL2Targets:           map[string][]L2Target{},
  		ifaceNameToFirstSeen:           map[string]time.Time{},
  		pendingIfaceNameToDeltaTargets: map[string]map[ip.CIDR]*Target{},
  		pendingIfaceNameToL2Targets:    map[string][]L2Target{},
  		reSync:                         true,
  		ifaceNameToUpdateType:          map[string]updateType{},
  		pendingConntrackCleanups:       map[ip.Addr]chan struct{}{},
  		addStaticARPEntry:              addStaticARPEntry,
  		conntrack:                      conntrack,
  		time:                           timeShim,
  		vxlan:                          vxlan,
  		deviceRouteSourceAddress:       deviceRouteSourceAddress,
  		deviceRouteProtocol:            deviceRouteProtocol,
  		removeExternalRoutes:           removeExternalRoutes,
  		tableIndex:                     tableIndex,
  		opReporter:                     opReporter,
  		livenessCallback:               func() {},
  		nl: handlemgr.NewHandleManager(
  			family,
  			featureDetector,
  			handlemgr.WithNewHandleOverride(newNetlinkHandle),
  			handlemgr.WithSocketTimeout(netlinkTimeout),
  		),
  	}
  
  	for _, o := range opts {
  		o(rt)
  	}
  
  	return rt
  }
  

• 注意

  routeTable基于interface regex匹配接口管理路由，无有效接口匹配会转为无设备路由模式

1.3.onIface

• dataplane.ifaceUpdates消息由onIfaceMonitorMessage消费，ifaceStateUpdate分发到rt.OnIfaceStateChanged处理。

• func (d *InternalDataplane) loopUpdatingDataplane() {
  	...
  
  	for {
  		select {
  		...
          // iface monitor监听的网卡变化
  		case ifaceUpdate := <-d.ifaceUpdates:
  			d.onIfaceMonitorMessage(ifaceUpdate)
  		...
  		}
  		...
  	}
  }
  
  // onIfaceMonitorMessage is called when we get a message from the interface monitor.
  func (d *InternalDataplane) onIfaceMonitorMessage(ifaceUpdate any) {
  	...
  
  	// As for datastore messages, the interface monitor can send many messages in one go,
  	d.processIfaceUpdate(ifaceUpdate)
  	drainChan(d.ifaceUpdates, d.processIfaceUpdate)
  	...
  }
  
  func (d *InternalDataplane) processIfaceUpdate(ifaceUpdate any) {
  	switch ifaceUpdateMsg := ifaceUpdate.(type) {
      // iface state变化才会推给routeTable
  	case *ifaceStateUpdate:
  		d.processIfaceStateUpdate(ifaceUpdateMsg)
  	...
  	}
  }
  
  func (d *InternalDataplane) processIfaceStateUpdate(ifaceUpdate *ifaceStateUpdate) {
  	...
  
      // 这里会分发给其它manager
  	for _, mgr := range d.allManagers {
  		mgr.OnUpdate(ifaceUpdate)
  	}
  
      // 这里分发给routeTable
  	for _, mgr := range d.managersWithRouteTables {
  		for _, routeTable := range mgr.GetRouteTableSyncers() {
  			routeTable.OnIfaceStateChanged(ifaceUpdate.Name, ifaceUpdate.State)
  		}
  	}
  }
  
  func (r *RouteTable) OnIfaceStateChanged(ifaceName string, state ifacemonitor.State) {
      // 过滤不关心的iface
  	if r.ifacePrefixRegexp == nil || !r.ifacePrefixRegexp.MatchString(ifaceName) {
  		return
  	}
      
      // iface状态为UP
  	if state == ifacemonitor.StateUp {
          // 标记full resync
  		r.ifaceNameToUpdateType[ifaceName] = updateTypeFullResync
          // 标记首次出现时间
  		r.onIfaceSeen(ifaceName)
  	}
  }
  

• 

• 注意

  ifaceMonitor监听的iface状态变化会推送到routeTable登记，后续的apply阶段正式下发路由

2.应用

2.1.apply

• dataplane会基于ifaceMonitor监听的ifaceUpdate事件执行d.Apply()，进一步执行routeTable.Apply()增量或全量同步路由。

• func (d *InternalDataplane) loopUpdatingDataplane() {
  	...
  
  	for {
  		select {
  		...
  		case ifaceUpdate := <-d.ifaceUpdates:
  			d.onIfaceMonitorMessage(ifaceUpdate)
  		...
  		}
  
  		if d.datastoreInSync && d.ifaceMonitorInSync && d.dataplaneNeedsSync {
  			// Dataplane is out-of-sync, check if we're throttled.
  			if d.applyThrottle.Admit() {
  				...
  
  				// Actually apply the changes to the dataplane.
  				d.apply()
  				...
  			}
              ...
  		}
  	}
  }
  
  func (d *InternalDataplane) apply() {
  	...
  
  	// Update routing table in parallel with the other updates.  We'll wait for it to finish before we return.
  	for _, r := range d.routeTableSyncers() {
  		routesWG.Add(1)
  		go func(r routetable.RouteTableSyncer) {
              // route table执行
  			r.Apply()
  			...
  			routesWG.Done()
  		}(r)
  	}
      ...
  
  	// Update routing rules in parallel with the other updates.  We'll wait for it to finish before we return.
  	for _, r := range d.routeRules() {
  		rulesWG.Add(1)
  		go func(r routeRules) {
              // route rule执行
  			r.Apply()
  			...
  			rulesWG.Done()
  		}(r)
  	}
  	...
  
  	// Wait for the route updates to finish.
  	routesWG.Wait()
  
  	// Wait for the rule updates to finish.
  	rulesWG.Wait()
  	...
  }
  
  func (r *RouteTable) Apply() error {
      // 首次启动/queueSync强制刷新
  	if r.reSync {
  
          // iface过滤
  		if r.ifacePrefixRegexp != nil {
  			// 获取netlink
  			nl, err := r.nl.Handle()
  			...
              
              // 扫描所有iface
  			links, err := nl.LinkList()
  			...
  			
  			seen := set.New[string]()
              for _, link := range links {
                  // report health
  				r.livenessCallback()
                  // 获取接口属性
  				attrs := link.Attrs()
  				if attrs == nil {
  					continue
  				}
  				ifaceName := attrs.Name
                  // 接口名匹配
  				if r.ifacePrefixRegexp.MatchString(ifaceName) {
                      // iface标记full resync
  					r.markIfaceForUpdate(ifaceName, true)
  					r.onIfaceSeen(ifaceName)
  					seen.Add(ifaceName)
  				}
  			}
              
  			// Clean up first-seen timestamps for old interfaces.
  			for name, firstSeen := range r.ifaceNameToFirstSeen {
                  // iface未过期
  				if seen.Contains(name) {
  					// Interface still present.
  					continue
  				}
                  
                  // iface宽限期未结束
  				if r.time.Since(firstSeen) < r.routeCleanupGracePeriod {
  					// Interface first seen recently.
  					continue
  				}
                  
                  // 清理过期iface的首次出现时间
  				delete(r.ifaceNameToFirstSeen, name)
  			}
  		}
  
  		// If we are managing no-OIF routes then add that to our dirty set.
  		if r.includeNoInterface {
              // 标记ifaceNone full sync
  			r.markIfaceForUpdate(InterfaceNone, true)
  		}
  
          // resync重置
  		r.reSync = false
  		...
  	}
  
  	graceIfaces := 0
  	for retry := 0; retry < 2; retry++ {
  	ifaceLoop:
          // 待更新的iface
  		for ifaceName, ia := range r.ifaceNameToUpdateType {
  			...
              
  			// vxlan模式
  			if r.vxlan {
  				// Sync L2 routes first.
  				err = r.syncL2RoutesForLink(ifaceName)
  			}
              
  			if err == nil {
  				// No errors syncing L2, sync L3 routes.
  				err = r.syncRoutesForLink(ifaceName, fullResync, firstTry)
  			}
  
  			// Handle errors from syncing either L2 or L3 routes.
  			switch err {
              // sync成功/iface未找到/iface down
  			case nil,IfaceNotPresent,IfaceDown:
                  // dirty set剔除iface
  				delete(r.ifaceNameToUpdateType, ifaceName)
  				continue ifaceLoop
              // 暂时无法同步
  			case IfaceGrace:
                  // 轮空
  				graceIfaces++
  				continue ifaceLoop
  			}
  
              // 最后一次重试
  			if lastTry {
  				// iface标记full resync
  				r.markIfaceForUpdate(ifaceName, true)
  			}
  		}
  	}
      ...
      
      // tcp/udp conntrack清理
  	r.cleanUpPendingConntrackDeletions()
  	...
  
  	return nil
  }
  

• 

• 补充

  resync基于内核和内存对齐iface，由syncRoutesForLink对齐路由，cleanUpPendingConntrackDeletions清理过期conntrack

2.2.vxlanRoute

• r.syncL2RoutesForLink()用于配置Pod流量处于vxlan模式对应的ARP/FDB，确保vxlan出接口封包相关的对象vxlan及目标节点。

• func (r *RouteTable) syncL2RoutesForLink(ifaceName string) error {
  	// vxlanMgr.OnUpdates-->vtepsByNode-->vxlanMgr.CompleteDeferredWork-->routeTable.SetL2Routes
      // -->pendingIfaceNameToL2Targets
  	if updatedTargets, ok := r.pendingIfaceNameToL2Targets[ifaceName]; ok {
          // DEL-->清理iface正式数据
  		if updatedTargets == nil {
  			delete(r.ifaceNameToL2Targets, ifaceName)
          // ADD/UPDATE-->更新iface正式数据
  		} else {
  			r.ifaceNameToL2Targets[ifaceName] = updatedTargets
  		}
          
          // 摘除iface pending
  		delete(r.pendingIfaceNameToL2Targets, ifaceName)
  	}
      
      // 获取iface关联目标
  	expectedTargets := r.ifaceNameToL2Targets[ifaceName]
  
  	// 获取iface属性
  	linkAttrs, err := r.getLinkAttributes(ifaceName)
  	...
  
  	// iface关联的VTEP	MAC
  	expected := map[string]bool{}
  	for _, target := range expectedTargets {
  		expected[target.VTEPMAC.String()] = true
  	}
  
      // iface相应的邻居(ip neighbor show)
  	existingNeigh, err := netlink.NeighList(linkAttrs.Index, netlink.FAMILY_V4)
  	...
  
  	// For each existing neighbor, if don't expect an entry for its MAC to be programmed on this link, then del.
  	for _, existing := range existingNeigh {
  		if existing.HardwareAddr == nil {
  			continue
  		}
          
          // neighbor过期
  		if _, ok := expected[existing.HardwareAddr.String()]; !ok {
  			// Remove the FDB entry for this neighbor.
  			n := netlink.Neigh{
  				LinkIndex:    existing.LinkIndex,
  				State:        netlink.NUD_PERMANENT,
  				Family:       syscall.AF_BRIDGE,
  				Flags:        netlink.NTF_SELF,
  				IP:           existing.IP,
  				HardwareAddr: existing.HardwareAddr,
  			}
  			
              // Delete the FDB entry
              netlink.NeighDel(&n)
              ...
  
  			// Delete the ARP entry.
  			netlink.NeighDel(&existing)
              ...
  		}
  	}
  
  	// If the value has changed since last programming, this  will update it.
  	for _, target := range expectedTargets {
  		r.ensureL2Dataplane(linkAttrs, target)
          ...
  	}
      ...
  
  	return nil
  }
  
  func (r *RouteTable) ensureL2Dataplane(linkAttrs *netlink.LinkAttrs, target L2Target) error {
  	// Add a static ARP entry.
  	a := &netlink.Neigh{
  		LinkIndex:    linkAttrs.Index,
  		State:        netlink.NUD_PERMANENT,
  		Type:         syscall.RTN_UNICAST,
  		IP:           target.GW.AsNetIP(),
  		HardwareAddr: target.VTEPMAC,
  	}
      
  	netlink.NeighSet(a)
      ...
  
  	// Add a FDB entry for this neighbor.
  	n := &netlink.Neigh{
  		LinkIndex:    linkAttrs.Index,
  		State:        netlink.NUD_PERMANENT,
  		Family:       syscall.AF_BRIDGE,
  		Flags:        netlink.NTF_SELF,
  		IP:           target.IP.AsNetIP(),
  		HardwareAddr: target.VTEPMAC,
  	}
      
  	netlink.NeighSet(n)
      ...
      
  	return nil
  }
  

• 

• 注意

  vxlan模式会额外补充ARP/FDB相关路由，确保路由的下一跳正确封包

2.3.syncRoute

• r.syncRoutesForLink()将pending delta合并至期望状态，基于内核路由对账清理多余route，注册缺失route/arp及清理conntrack。

• func (r *RouteTable) syncRoutesForLink(ifaceName string, fullSync bool, firstTry bool) error {
  	...
  
  	defer func() {
          // CIDR路由清理后conntrack回收
  		cidrsToTarget := r.ifaceNameToTargets[ifaceName]
  		deletedConnCIDRs.Iter(func(cidr ip.CIDR) error {
  			if _, ok := cidrsToTarget[cidr]; !ok {
  				// Route is deleted and CIDR should not be routable anymore - remove conntrack entries.
  				r.startConntrackDeletion(cidr.Addr())
  			}
  			return nil
  		})
  	}()
      ...
  
  	if fullSync {
  		// pending route转至ifaceNameToTargets
  		_, _ = r.applyRouteDeltas(ifaceName, deletedConnCIDRs)
  
  		// 内核和内存route对齐-->未期望的旧route
  		routesToDelete, resyncErr = r.fullResyncRoutesForLink(logCxt, ifaceName, deletedConnCIDRs)
          ...
  
  		// Ensure we have static ARP entries for all of our existing routes.
  		for _, target := range r.ifaceNameToTargets[ifaceName] {
              // except route注册ARP(vxlan模式ARP注册的是VTEP MAC/GW)
  			if r.ipVersion == 4 && target.DestMAC != nil {
  				...
  				r.addStaticARPEntry(target.CIDR, target.DestMAC, ifaceName)
  				...
  			}
  		}
  	}
  
  	// Update the cached values from the deltas and get the set of targets to create and delete.
  	targetsToCreate, targetsToDelete := r.applyRouteDeltas(ifaceName, deletedConnCIDRs)
  
  	// 获取iface属性
  	linkAttrs, err := r.getLinkAttributes(ifaceName)
  	...
  	nl, err := r.nl.Handle()
  	...
  
  	// merge待清理route
  	// data that we use to tidy up routes and conntrack entries).
  	for _, target := range targetsToDelete {
  		routesToDelete = append(routesToDelete, r.createL3Route(linkAttrs, target))
  	}
  
  	// Delete the combined set of routes.
  	for _, route := range routesToDelete {
  		r.livenessCallback()
  		nl.RouteDel(&route)
          ...
  	}
  
  	// Now add target routes.
  	for _, target := range targetsToCreate {
  		r.livenessCallback()
          // 初始化route对象
  		route := r.createL3Route(linkAttrs, target)
  
  		// wait for any previous conntrack entry to be cleaned up.
  		r.waitForPendingConntrackDeletion(target.CIDR.Addr())
          // 注册路由
  		nl.RouteAdd(&route)
          ...
          
          // 注册路由相关ARP
  		if r.ipVersion == 4 && target.DestMAC != nil {
  			// TODO(smc) clean up/sync old ARP entries
  			r.addStaticARPEntry(target.CIDR, target.DestMAC, ifaceName)
  			...
  		}
  	}
  	...
  
  	// Return any un-handled re-sync error.
  	return resyncErr
  }
  
  func (r *RouteTable) applyRouteDeltas(ifaceName string, deletedConnCIDRs set.Set[ip.CIDR]) (c, d []Target) {
  	// iface路由缓存
  	cidrsToTarget := r.ifaceNameToTargets[ifaceName]
  	if cidrsToTarget == nil {
  		cidrsToTarget = map[ip.CIDR]Target{}
  	}
  
  	// ifaceMonitor监听iface/ifaceAddr设置
  	deltaTargets := r.pendingIfaceNameToDeltaTargets[ifaceName]
      // 扫描待更新路由
  	for cidr, target := range deltaTargets {
          // 清理旧route
  		if current, ok := cidrsToTarget[cidr]; ok {
  			// cidr标记清理
  			targetsToDelete = append(targetsToDelete, current)
              // 标记conntrack回收
  			deletedConnCIDRs.Add(cidr)
              // 由内存缓存剔除cidr
  			delete(cidrsToTarget, cidr)
  		}
          
          // 重新补充路由
  		if target != nil {
  			// Delta adds a new entry.
  			targetsToCreate = append(targetsToCreate, *target)
  			cidrsToTarget[cidr] = *target
  		}
  	}
  
  	// 处理完成清理pending route
  	delete(r.pendingIfaceNameToDeltaTargets, ifaceName)
  
  	// 更新内存route缓存
  	if len(cidrsToTarget) == 0 {
  		delete(r.ifaceNameToTargets, ifaceName)
  	} else {
  		r.ifaceNameToTargets[ifaceName] = cidrsToTarget
  	}
  
  	return
  }
  

• 

• 注意

  vxlan模式route出网口是vxlan.tun，下一跳是vtep gw，host直连模式出网卡是物理网络设备，下一条是对端PodIP