dphandler

1.主循环

1.1.loop

• d.loopUpdatingDataplane()是dataplane的核心调度循环，负责接收各类事件、维护期望状态及利用限流机制驱动apply建规则同步到底层。

• func (d *InternalDataplane) loopUpdatingDataplane() {
  	...
  
  	for {
  		select {
          // 1.calGraph计算的事件
  		case msg := <-d.toDataplane:
              // 更新缓存,标记同步
  			d.onDatastoreMessage(msg)
          // 2.监听到的iface事件
  		case ifaceUpdate := <-d.ifaceUpdates:
              // 更新缓存,标记同步
  			d.onIfaceMonitorMessage(ifaceUpdate)
          // 3.ipset刷新
  		case <-ipSetsRefreshC:
  			d.forceIPSetsRefresh = true
  			d.dataplaneNeedsSync = true
          // 4.route刷新
  		case <-routeRefreshC:
  			d.forceRouteRefresh = true
  			d.dataplaneNeedsSync = true
          // 5.XDP刷新
  		case <-xdpRefreshC:
  			d.forceXDPRefresh = true
  			d.dataplaneNeedsSync = true
          // 6.resched周期同步
  		case <-d.reschedC:
  			d.dataplaneNeedsSync = true
  			// nil out the channel to record that the timer is now inactive.
  			d.reschedC = nil
          // 7.限流token颁发(100ms生成一个)
  		case <-throttleC:
  			d.applyThrottle.Refill()
  		...
          // 8.失败任务重试(间隔10s) 
  		case <-retryTicker.C:
  		...
  		}
  
          // 9.datastore同步完成 & iface状态同步完成 & 存在待同步变更
  		if d.datastoreInSync && d.ifaceMonitorInSync && d.dataplaneNeedsSync {
  			// token申请
  			if d.applyThrottle.Admit() {
                  // 未触发限流,重置标志位
  				if beingThrottled && d.applyThrottle.WouldAdmit() {
  					beingThrottled = false
  				}
  
  				// 规则应用
  				d.apply()
  
  				...
  
                  // 第一个触发apply,执行回调触发GC
  				if !d.doneFirstApply {
  					d.doneFirstApply = true
  					if d.config.PostInSyncCallback != nil {
  						d.config.PostInSyncCallback()
  					}
  				}
  				...
  			} else {
                  // 标记限流
  				if !beingThrottled {
  					beingThrottled = true
  				}
  			}
  		}
  	}
  }
  

• 

• 注意

  d.apply()会基于ep/route/xdp... manager更新底层规则，是比较核心的模块

1.2.handler

• d.onDatastoreMessage()/d.onIfaceMonitorMessage()根据datastore/iface监听的变化同步修改manager/route内存状态。

• // called when get message from calGraph it opportunistically processes a match of msg from its channel.
  func (d *InternalDataplane) onDatastoreMessage(msg interface{}) {
  	...
  
  	// 1.处理当前消息
  	d.processMsgFromCalcGraph(msg)
      
      // 2.继续消费100条chan msg(减少apply次数)
  	drainChan(d.toDataplane, d.processMsgFromCalcGraph)
  	...
  }
  
  func (d *InternalDataplane) processMsgFromCalcGraph(msg interface{}) {
  	...
      
      // 1.标记出现变更
  	d.dataplaneNeedsSync = true
  	...
      
      // 2.分发更新不同mgr的内存模型
  	for _, mgr := range d.allManagers {
  		mgr.OnUpdate(msg)
  	}
      
      // 3.标记同步完成
  	switch msg.(type) {
  	case *proto.InSync:
  		d.datastoreInSync = true
  	}
  }
  
  // called when we get message from interface monitor it opportunistically processes match msg from its channel.
  func (d *InternalDataplane) onIfaceMonitorMessage(ifaceUpdate any) {
  	...
  
  	// 1.处理当前消息
  	d.processIfaceUpdate(ifaceUpdate)
      
      // 2.继续消费100条chan msg(减少apply次数)
  	drainChan(d.ifaceUpdates, d.processIfaceUpdate)
  
      // 3.标记出现变更
  	d.dataplaneNeedsSync = true
  	...
  }
  
  func (d *InternalDataplane) processIfaceUpdate(ifaceUpdate any) {
  	switch ifaceUpdateMsg := ifaceUpdate.(type) {
      // 1.iface状态变化
  	case *ifaceStateUpdate:
  		d.processIfaceStateUpdate(ifaceUpdateMsg)
      // 2.iface地址变化
  	case *ifaceAddrsUpdate:
  		d.processIfaceAddrsUpdate(ifaceUpdateMsg)
      // 3.iface监控进度通知
  	case *ifaceInSync:
  		d.processIfaceInSync()
  	}
  }
  
  func (d *InternalDataplane) processIfaceStateUpdate(ifaceUpdate *ifaceStateUpdate) {
  	d.dataplaneNeedsSync = true
  	...
      
      // 1.IPVS支持状态变化,发送终止信号
  	if ifaceUpdate.Name == KubeIPVSInterface {
  		d.checkIPVSConfigOnStateUpdate(ifaceUpdate.State)
  		return
  	}
  
      // 2.分发更新不同mgr的内存模型
  	for _, mgr := range d.allManagers {
  		mgr.OnUpdate(ifaceUpdate)
  	}
  
      // 3.带路由表的mgr
  	for _, mgr := range d.managersWithRouteTables {
          // mgr相关的路由同步器
  		for _, routeTable := range mgr.GetRouteTableSyncers() {
              // 通知同步器网络状态变化
  			routeTable.OnIfaceStateChanged(ifaceUpdate.Name, ifaceUpdate.State)
  		}
  	}
  }
  

• 

• 注意 

  这里的mgr.OnUpdate()会更新内存模型，还没有正式发到底层规则

2.更新

2.1.apply

• d.apply()会计算内存的期望状态，处理XDP/Route/ipset/iptables的一致性与重试，基于严格顺序将变更批量应用到内核实现网络策略管理。

• func (d *InternalDataplane) apply() {
  	// 1.重置同步标志
  	d.dataplaneNeedsSync = false
  
      // 2.Batch实现解析(endpointMgr)
  	for _, mgr := range d.allManagers {
          // aiface-->hostEndpoint缓存更新
  		if handler, ok := mgr.(UpdateBatchResolver); ok {
  			handler.ResolveUpdateBatch()
  			...
  		}
  	}
  
  	// 3.生成内存rule规则
  	for _, mgr := range d.allManagers {
  		mgr.CompleteDeferredWork()
  		...
  	}
  
      // 4.XDP相关
  	if d.xdpState != nil {
          // 标记强制刷新
  		if d.forceXDPRefresh {
  			// Refresh timer popped.
  			d.xdpState.QueueResync()
  			d.forceXDPRefresh = false
  		}
  		...
  
          // endpoint差异计算
  		d.xdpState.ProcessPendingDiffState(d.endpointsSourceV4)
          // 更新BPFMap及重载BPF程序
  		d.applyXDPActions()
          ...
          
          // member更新
          err := d.xdpState.ProcessMemberUpdates()
          
          // 清除diff buffer
          d.xdpState.DropPendingDiffState()
          if err != nil {
              d.applyXDPActions()
              ...
          }
          
          // 标记state收敛
          d.xdpState.UpdateState()
          
          // xdp异常降级为iptables
  		if applyXDPError != nil {
  			d.shutdownXDPCompletely()
              ...
  		}
  	}
  	...
  
      // 5.route相关
  	if d.forceRouteRefresh {
  		// route相关manager
  		for _, r := range d.routeTableSyncers() {
  			// 标记强制刷新route
  			r.QueueResync()
  		}
          
          // route rule相关manager
  		for _, r := range d.routeRules() {
  			// 标记强制刷新route
  			r.QueueResync()
  		}
  		d.forceRouteRefresh = false
  	}
  
      // 6.ipset相关
  	if d.forceIPSetsRefresh {
  		// 标记强制刷新
  		for _, r := range d.ipSets {
  			// Queue a resync on the next Apply().
  			r.QueueResync()
  		}
  		d.forceIPSetsRefresh = false
  	}
  
  	...
      
      // 7.ipset list/restore重建规则
  	for _, ipSets := range d.ipSets {
  		ipSetsWG.Add(1)
  		go func(ipSets common.IPSetsDataplane) {
  			ipSets.ApplyUpdates()
  			...
  			ipSetsWG.Done()
  		}(ipSets)
  	}
  
  	...
      // 8.route add/del注册路由
  	for _, r := range d.routeTableSyncers() {
  		routesWG.Add(1)
  		go func(r routetable.RouteTableSyncer) {
  			r.Apply()
  			...
  			routesWG.Done()
  		}(r)
  	}
  
  	...
      // 9.route rule规则更新
  	for _, r := range d.routeRules() {
  		rulesWG.Add(1)
  		go func(r routeRules) {
  			r.Apply()
  			...
  			rulesWG.Done()
  		}(r)
  	}
  	...
      
  	// 10.iptables规则更新
  	for _, t := range d.allIptablesTables {
  		iptablesWG.Add(1)
  		go func(t *iptables.Table) {
  			t.Apply()
  			...
  			iptablesWG.Done()
  		}(t)
  	}
  	...
  
  	// 11.ipset过期规则清理
  	for _, ipSets := range d.ipSets {
  		ipSetsWG.Add(1)
  		go func(s common.IPSetsDataplane) {
  			s.ApplyDeletions()
  			...
  			ipSetsWG.Done()
  		}(ipSets)
  	}
  	...
  
  	// 12.resched timer重新初始化
  	if d.reschedC != nil {
  		if !d.reschedTimer.Stop() {
  			// Timer had already popped, drain its channel.
  			<-d.reschedC
  		}
  		// Nil out our copy of the channel to record that the timer is inactive.
  		d.reschedC = nil
  	}
      
  	if reschedDelay != 0 {
  		// We need to reschedule.
  		if d.reschedTimer == nil {
  			// First time, create the timer.
  			d.reschedTimer = time.NewTimer(reschedDelay)
  		} else {
  			// Have an existing timer, reset it.
  			d.reschedTimer.Reset(reschedDelay)
  		}
  		d.reschedC = d.reschedTimer.C
  	}
  }
  

• 

• 注意

  apply会先计算内存预期状态，之后按照顺序将规则apply到底层，具体依赖不同manager各自的接口实现，后面会分析

2.2.connector

• dataplane connector是syncer/typha-->dataplane的桥梁，syncer同步的事件经calGraph计算基于connector推到dataplane。

• func newConnector(configParams *config.Config,
  	configUpdChan chan<- map[string]string,
  	datastore bapi.Client,
  	datastorev3 client.Interface,
  	dataplane dp.DataplaneDriver,
  	failureReportChan chan<- string,
  ) *DataplaneConnector {
  	felixConn := &DataplaneConnector{
  		config:                           configParams,
  		configUpdChan:                    configUpdChan,
  		datastore:                        datastore,
  		datastorev3:                      datastorev3,
  		ToDataplane:                      make(chan interface{}),
  		StatusUpdatesFromDataplane:       make(chan interface{}),
  		InSync:                           make(chan bool, 1),
  		failureReportChan:                failureReportChan,
  		dataplane:                        dataplane,
  		wireguardStatUpdateFromDataplane: make(chan *proto.WireguardStatusUpdate, 1),
  	}
  	return felixConn
  }
  

• 注意

  Syncer->channel->Validator->calGraph->channel->dataplane