dpstart

1.简介

1.1.原理

• 前文介绍过sycner，这里监听datastore的更新会分发到cahnnel供Validator校验，之后会转发给calGraph以分发给dataplane应用。

• 

• 注意

  syncer-->validator中间会加一层syncCallbacksDecoupler链接，calGraph-->dataplane中间会加一层dpconnector

1.2.decoupler

• syncCallbacksDecoupler实现上比较简单，作为syncer-->validator的缓冲，将syncer监听到的数据经缓冲channel转给validator。

• type SyncerCallbacksDecoupler struct {
  	c chan interface{}
  }
  
  func NewSyncerCallbacksDecoupler() *SyncerCallbacksDecoupler {
  	return &SyncerCallbacksDecoupler{
  		c: make(chan interface{}),
  	}
  }
  
  func (a *SyncerCallbacksDecoupler) OnStatusUpdated(status api.SyncStatus) {
  	a.c <- status
  }
  
  func (a *SyncerCallbacksDecoupler) OnUpdates(updates []api.Update) {
  	a.c <- updates
  }
  
  func (a *SyncerCallbacksDecoupler) SendTo(sink api.SyncerCallbacks) {
  	for obj := range a.c {
          // 这里会将对象转到validator
  		switch obj := obj.(type) {
  		case api.SyncStatus:
  			sink.OnStatusUpdated(obj)
  		case []api.Update:
  			sink.OnUpdates(obj)
  		}
  	}
  }
  

• 注意

  syncCallbacksDecoupler未进一步处理数据，作为缓冲区将sycner监听事件推到validator模块进行校验

1.3.calGraph

• AsyncCalcGraph基于calGraph计算网络资源差异，通过dispatcher将变更暂存到eventSequencer，后续会flush到dpconnector。

• func NewAsyncCalcGraph(conf *config.Config, outputChannels []chan<- interface{}, ...) *AsyncCalcGraph {
  	eventSequencer := NewEventSequencer(conf)
  	g := &AsyncCalcGraph{
  		inputEvents:      make(chan interface{}, 10),
  		outputChannels:   outputChannels,
  		eventSequencer:   eventSequencer,
  		...
  	}
      
  	g.CalcGraph = NewCalculationGraph(eventSequencer, conf, g.reportHealth)
  	...
      
  	eventSequencer.Callback = g.onEvent
  	...
      
  	return g
  }
  
  // 事件推送到outchan(toDataplane/toPolicySync)
  func (acg *AsyncCalcGraph) onEvent(event interface{}) {
  	...
  channelLoop:
  	for _, c := range acg.outputChannels {
  		for {
  			select {
  			case c <- event:
  				continue channelLoop
  			...
  		}
  	}
  	countOutputEvents.Inc()
  }
  
  func NewCalculationGraph(callbacks PipelineCallbacks, conf *config.Config, liveCallback func()) *CalcGraph {
  	...
  
  	// The source of the processing graph, this dispatcher will be fed all the updates from the
  	// datastore, fanning them out to the registered receivers.
  	//
  	//               Syncer
  	//                 ||
  	//                 || All updates
  	//                 \/
  	//             Dispatcher (all updates)
  	//                / | \
  	//               /  |  \  Updates filtered by type
  	//              /   |   \
  	//     receiver_1  ...  receiver_n
  	//
      // 事件总线
  	allUpdDispatcher := dispatcher.NewDispatcher()
  
  	// Some of the receivers only need to know about local endpoints. Create a second dispatcher
  	// that will filter out non-local endpoints.
  	//
  	//          ...
  	//       Dispatcher (all updates)
  	//          ... \
  	//               \  All Host/Workload Endpoints
  	//                \
  	//              Dispatcher (local updates)
  	//               <filter>
  	//                / | \
  	//               /  |  \  Local Host/Workload Endpoints only
  	//              /   |   \
  	//     receiver_1  ...  receiver_n
  	//
      // 注册local endpoint分发器
  	localEndpointDispatcher := dispatcher.NewDispatcher()
  	(*localEndpointDispatcherReg)(localEndpointDispatcher).RegisterWith(allUpdDispatcher)
      // local endpoint dispatcher包一层local endpoint filter
  	localEndpointFilter := &endpointHostnameFilter{hostname: hostname}
  	localEndpointFilter.RegisterWith(localEndpointDispatcher)
  
  	// The active rules calculator matches local endpoints against policies and profiles to figure
  	// out which policies/profiles are active on this host.  Limiting to policies that apply to
  	// local endpoints significantly cuts down the number of policies that Felix has to
  	// render into the dataplane.
  	//
  	//           ...
  	//        Dispatcher (all updates)
  	//           /   \
  	//          /     \  All Host/Workload Endpoints
  	//         /       \
  	//        /      Dispatcher (local updates)
  	//       /            |
  	//       | Policies   | Local Host/Workload Endpoints only
  	//       | Profiles   |
  	//       |            |
  	//     Active Rules Calculator
  	//              |
  	//              | Locally active policies/profiles
  	//             ...
  	//
      // 计算本节点有效的policy/profile,注册到local endpoint dispatcher和all dispatcher
  	activeRulesCalc := NewActiveRulesCalculator()
  	activeRulesCalc.RegisterWith(localEndpointDispatcher, allUpdDispatcher)
  
  	// The active rules calculator only figures out which rules are active, it doesn't extract
  	// any information from the rules.  The rule scanner takes the output from the active rules
  	// calculator and scans the individual rules for selectors and named ports.  It
  	// generates events when a new selector/named port starts/stops being used.
  	//
  	//             ...
  	//     Active Rules Calculator
  	//              |
  	//              | Locally active policies/profiles
  	//              |
  	//         Rule scanner
  	//          |    \
  	//          |     \ Locally active selectors/named ports
  	//          |      \
  	//          |      ...
  	//          |
  	//          | IP set active/inactive
  	//          |
  	//     <dataplane>
  	//
      // 向active rule calc注册rule scanner,将生效的policy/profiler拆解为selector/named port/ipset规则
  	ruleScanner := NewRuleScanner()
  	// Wire up the rule scanner's inputs.
  	activeRulesCalc.RuleScanner = ruleScanner
  	// rule分发给calGraph.eventSequencer，间接转给dataplane
  	ruleScanner.RulesUpdateCallbacks = callbacks
  
      // service indexer注册到all dispatcher
  	serviceIndex := serviceindex.NewServiceIndex()
  	serviceIndex.RegisterWith(allUpdDispatcher)
  	// ipset member更新,会经过calGraph.eventSequencer间接转给dataplane
  	serviceIndex.OnMemberAdded = func(ipSetID string, member labelindex.IPSetMember) {
  		...
  		callbacks.OnIPSetMemberAdded(ipSetID, member)
  	}
  	serviceIndex.OnMemberRemoved = func(ipSetID string, member labelindex.IPSetMember) {
  		...
  		callbacks.OnIPSetMemberRemoved(ipSetID, member)
  	}
  	...
  
  	// The rule scanner only goes as far as figuring out which selectors/named ports are
  	// active. Next we need to figure out which endpoints (and hence which IP addresses/ports) are
  	// in each tag/selector/named port. The IP set member index calculates the set of IPs and named
  	// ports that should be in each IP set.  To do that, it matches the active selectors/named
  	// ports extracted by the rule scanner against all the endpoints. The service index does the same
  	// for service based rules, building IP set contributions from endpoint slices.
  	//
  	//        ...
  	//     Dispatcher (all updates)
  	//      |
  	//      | All endpoints
  	//      |
  	//      |       ...
  	//      |    Rule scanner
  	//      |     |       \
  	//      |    ...       \ Locally active selectors/named ports
  	//       \              |
  	//        \_____        |
  	//              \       |
  	//            IP set member index / service index
  	//                   |
  	//                   | IP set member added/removed
  	//                   |
  	//               <dataplane>
  	//
      // ipset member indexer初始化
  	ipsetMemberIndex := labelindex.NewSelectorAndNamedPortIndex()
  	...
  	// 注册到all dispatcher
  	ipsetMemberIndex.RegisterWith(allUpdDispatcher)
      
      // rule scanner输出规则处理
  	ruleScanner.OnIPSetActive = func(ipSet *IPSetData) {
          // 基于calGraph.eventSequencer通知dataplane
  		callbacks.OnIPSetAdded(ipSet.UniqueID(), ipSet.DataplaneProtocolType())
          // svc类型,注册到service indexer
  		if ipSet.Service != "" {
  			serviceIndex.UpdateIPSet(ipSet.UniqueID(), ipSet.Service)
          // 非service类型,注册到ipset indexer
  		} else {
  			ipsetMemberIndex.UpdateIPSet(ipSet.UniqueID(), ipSet.Selector, ipSet.NamedPortProtocol, ipSet.NamedPort)
  		}
  		...
  	}
  	ruleScanner.OnIPSetInactive = func(ipSet *IPSetData) {
  		if ipSet.Service != "" {
  			serviceIndex.DeleteIPSet(ipSet.UniqueID())
  		} else {
  			ipsetMemberIndex.DeleteIPSet(ipSet.UniqueID())
  		}
  		callbacks.OnIPSetRemoved(ipSet.UniqueID())
  		...
  	}
  	// 基于calGraph.eventSequencer通知dataplane变更
  	ipsetMemberIndex.OnMemberAdded = func(ipSetID string, member labelindex.IPSetMember) {
  		...
  		callbacks.OnIPSetMemberAdded(ipSetID, member)
  	}
  	ipsetMemberIndex.OnMemberRemoved = func(ipSetID string, member labelindex.IPSetMember) {
  		...
  		callbacks.OnIPSetMemberRemoved(ipSetID, member)
  	}
  
  	// The endpoint policy resolver marries up the active policies with local endpoints and
  	// calculates the complete, ordered set of policies that apply to each endpoint.
  	//
  	//        ...
  	//     Dispatcher (all updates)
  	//      |
  	//      | All policies
  	//      |
  	//      |       ...
  	//       \   Active rules calculator
  	//        \       \
  	//         \       \
  	//          \       | Policy X matches endpoint Y
  	//           \      | Policy Z matches endpoint Y
  	//            \     |
  	//           Policy resolver
  	//                  |
  	//                  | Endpoint Y has policies [Z, X] in that order
  	//                  |
  	//             <dataplane>
  	//
      // active rule calc注册policy resolver
  	polResolver := NewPolicyResolver()
  	// Hook up the inputs to the policy resolver.
  	activeRulesCalc.PolicyMatchListener = polResolver
      // policy resolver注册到all dispatcher和local endpoint dispatcher
  	polResolver.RegisterWith(allUpdDispatcher, localEndpointDispatcher)
  	// And hook its output to the callbacks.
  	polResolver.Callbacks = callbacks
  
  	// Register for host IP updates.
  	//
  	//        ...
  	//     Dispatcher (all updates)
  	//         |
  	//         | host IPs
  	//         |
  	//       passthru
  	//         |
  	//         |
  	//         |
  	//      <dataplane>
  	//
      // hostIP-->dataplane转发模块
  	hostIPPassthru := NewDataplanePassthru(callbacks)
      // 注册到all dispatcher
  	hostIPPassthru.RegisterWith(allUpdDispatcher)
  
      // BPF/Vxlan/WireGuard模式
  	if conf.BPFEnabled || conf.Encapsulation.VXLANEnabled || conf.Encapsulation.VXLANEnabledV6 || conf.WireguardEnabled || conf.WireguardEnabledV6 {
  		// Calculate simple node-ownership routes.
  		//        ...
  		//     Dispatcher (all updates)
  		//         |
  		//         | host IPs, host config, IP pools, IPAM blocks
  		//         |
  		//       L3 resolver
  		//         |
  		//         | routes
  		//         |
  		//      <dataplane>
  		//
          // 向all dispatcher/local endpoint dispatcher注册route resolver
  		l3RR := NewL3RouteResolver(hostname, callbacks, conf.UseNodeResourceUpdates(), conf.RouteSource)
  		l3RR.RegisterWith(allUpdDispatcher, localEndpointDispatcher)
  		...
  	}
  
  	// Calculate VXLAN routes.
  	//        ...
  	//     Dispatcher (all updates)
  	//         |
  	//         | host IPs, host config, IP pools, IPAM blocks
  	//         |
  	//       vxlan resolver
  	//         |
  	//         | VTEPs, routes
  	//         |
  	//      <dataplane>
  	//
      // vxlan模式
  	if conf.Encapsulation.VXLANEnabled || conf.Encapsulation.VXLANEnabledV6 {
          // 节点-->vtep-->route关系计算
  		vxlanResolver := NewVXLANResolver(hostname, callbacks, conf.UseNodeResourceUpdates())
          // 注册到all dispatcher
  		vxlanResolver.RegisterWith(allUpdDispatcher)
  	}
  
  	// Register for config updates.
  	//
  	//        ...
  	//     Dispatcher (all updates)
  	//         |
  	//         | separate config updates foo=bar, baz=biff
  	//         |
  	//       config batcher
  	//         |
  	//         | combined config {foo=bar, bax=biff}
  	//         |
  	//      <dataplane>
  	//
      // all dispatcher注册config batcher(基于key-val合并配置更新)
  	configBatcher := NewConfigBatcher(hostname, callbacks)
  	configBatcher.RegisterWith(allUpdDispatcher)
  
  	// The profile decoder identifies objects with special dataplane significance which have
  	// been encoded as profiles by libcalico-go. At present this includes Kubernetes Service
  	// Accounts and Kubernetes Namespaces.
  	//        ...
  	//     Dispatcher (all updates)
  	//         |
  	//         | Profiles
  	//         |
  	//       profile decoder
  	//         |
  	//         |
  	//         |
  	//      <dataplane>
  	//
      // all dispatcher注册profile decoder(由profile提取dataplane关心对象)
  	profileDecoder := NewProfileDecoder(callbacks)
  	profileDecoder.RegisterWith(allUpdDispatcher)
  
  	// Register for IP Pool updates. EncapsulationResolver will send a message to the
  	// dataplane so that Felix is restarted if IPIP and/or VXLAN encapsulation changes
  	// due to IP pool changes, so that it is recalculated at Felix startup.
  	//
  	//        ...
  	//     Dispatcher (all updates)
  	//         |
  	//         | IP pools
  	//         |
  	//       encapsulation resolver
  	//
      // all dispatcher注册封装解析器(IPPool事件)
  	encapsulationResolver := NewEncapsulationResolver(conf, callbacks)
  	encapsulationResolver.RegisterWith(allUpdDispatcher)
  
  	return &CalcGraph{
  		AllUpdDispatcher:      allUpdDispatcher,
  		activeRulesCalculator: activeRulesCalc,
  	}
  }
  

• 

• 注意

  1.dispatcher会根据syncer/typha监听资源计算本节点生效的ipset/policy/route...推送到eventSequencer暂存

  2.eventSequencer.Flush()会将暂存事件推送到dpconnector.toDataplane

  3.dpconnector.toDataplane事件会转发到dataplane.toDataplane

2.启动

2.1.validator

• validator是事件过滤模块，校验合法的数据会基于回调推到下游(calGraph)，不合法的数据会清空value，相当于交给下游删除事件。

• func NewValidationFilter(sink api.SyncerCallbacks, felixConfig *config.Config) *ValidationFilter {
  	return &ValidationFilter{
  		sink:   sink,
  		config: felixConfig,
  	}
  }
  
  func (v *ValidationFilter) OnStatusUpdated(status api.SyncStatus) {
  	// 直接转给calGraph.inputChan
  	v.sink.OnStatusUpdated(status)
  }
  
  func (v *ValidationFilter) OnUpdates(updates []api.Update) {
  	...
      
      // 资源变更校验
  	for i, update := range updates {
  		...
          
          // validator选择
  		validatorFunc := v1v.Validate
  		if _, isV3 := update.Key.(model.ResourceKey); isV3 {
  			validatorFunc = v3v.Validate
  		}
          
          // value不为空,需校验结构合法性
  		if update.Value != nil {
  			val := reflect.ValueOf(update.Value)
  			if val.Kind() == reflect.Ptr {
  				elem := val.Elem()
  				if elem.Kind() == reflect.Struct {
  					if err := validatorFunc(elem.Interface()); err != nil {
  						update.Value = nil
  					}
  				}
  			}
  
  			switch value := update.Value.(type) {
  			case *model.WorkloadEndpoint:
                  // name不为空 && 启用spoof prefix且config允许
  				err := v.validateWorkloadEndpoint(value)
  				if err != nil {
  					update.Value = nil
  				}
  			}
  		}
          
  		filteredUpdates[i] = update
  	}
      
      // 批事件推给calGraph.inputChan
  	v.sink.OnUpdates(filteredUpdates)
  }
  

• 

• 注意

  validator负责检查struct结构体数据的合法性，非法结构数据value会置空，推到calGraph相当于删除事件

2.2.calGraph

• acg.Start()会执行调度循环，将validator发送的事件分发到dispatcher缓存，基于令牌限流策略间隔10ms触发flush到dataplane。

• func (acg *AsyncCalcGraph) Start() {
      // 10ms的flush令牌
  	acg.flushTicks = time.NewTicker(tickInterval).C
  	...
      
      // 事件处理
  	go acg.loop()
  }
  
  func (acg *AsyncCalcGraph) loop() {
  	...
  	for {
  		select {
          // validator推送事件
  		case update := <-acg.inputEvents:
  			switch update := update.(type) {
              // policy/endpoint/ipset/service更新
  			case []api.Update:
  				// Update; send it to the dispatcher.
  				for i, upd := range update {
                      // dispatcher-->subcriber
  					acg.AllUpdDispatcher.OnUpdates(update[i : i+1])
  					...
  				}
              // 状态同步
  			case api.SyncStatus:
  				// Sync status changed, check if we're now in-sync.
  				acg.syncStatusNow = update
                  // dispatcher->subcriber
  				acg.AllUpdDispatcher.OnStatusUpdated(update)
                  // 第一次状态事件
  				if update == api.InSync && !acg.beenInSync {
                      // 标记insync
  					acg.beenInSync = true
  					acg.needToSendInSync = true
  					acg.dirty = true
                      // 强制允许一次flush
  					if acg.flushLeakyBucket == 0 {
  						// Force a flush.
  						acg.flushLeakyBucket++
  					}
  				}
  			...
  			}
              // 标记有变更
  			acg.dirty = true
          // 间隔10ms补充令牌,控制flush速率
  		case <-acg.flushTicks:
  			// Timer tick: fill up the leaky bucket.
  			if acg.flushLeakyBucket < leakyBucketSize {
  				acg.flushLeakyBucket++
  			}
  		...
  		}
          
          // 执行flush
  		acg.maybeFlush()
  	}
  }
  
  // maybeFlush flushes the event buffer if: we know it's dirty and we're not throttled.
  func (acg *AsyncCalcGraph) maybeFlush() {
      // 必须有变更才允许flush
  	if !acg.dirty {
  		return
  	}
      
      // 可以申请到令牌
  	if acg.flushLeakyBucket > 0 {
  		acg.flushLeakyBucket--
          // 缓冲事件批量推到dataplane
  		acg.eventSequencer.Flush()
  		...
          
          // insync事件只触发一次
  		if acg.needToSendInSync {
  			acg.onEvent(&proto.InSync{})
  			acg.needToSendInSync = false
  		}
  		acg.dirty = false
  	}
  }
  

• 

• 注意

  dispatcher将事件回调暂存到eventSequencer，eventSequencer.Flush()会执行calGraph.onEvent()转发事件到dpconnector

2.3.connector

• dpconnector作为controlplane和dataplane桥梁，报告dataplane状态到statusReporter，将calGraph消息同步到dataplane。

• func (fc *DataplaneConnector) Start() {
  	// write to the dataplane
  	go fc.sendMessagesToDataplaneDriver()
  
  	// read messages from dataplane
  	go fc.readMessagesFromDataplane()
  
  	// handle Wireguard update to Node.
  	go fc.handleWireguardStatUpdateFromDataplane()
  }
  
  func (fc *DataplaneConnector) sendMessagesToDataplaneDriver() {
  	defer func() {
          // 崩溃终止进程
  		fc.shutDownProcess("Failed to send messages to dataplane")
  	}()
  
  	for {
          // calGraph推送的事件
  		msg := <-fc.ToDataplane
  		switch msg := msg.(type) {
          // 同步消息
  		case *proto.InSync:
              // 通知status reporter同步完成
  			if !fc.datastoreInSync {
  				fc.datastoreInSync = true
  				fc.InSync <- true
  			}
          // 配置变更
  		case *proto.ConfigUpdate:
  			fc.handleConfigUpdate(msg)
          // datastore未就绪
  		case *calc.DatastoreNotReady:
              // 终止进程
  			fc.shutDownProcess("datastore became unready")
          // 封装消息
  		case *proto.Encapsulation:
              // 获取封装配置
  			encap := func() config.Encapsulation {
  				// Using a func() here to limit the scope of our defer.
  				fc.configLock.Lock()
  				defer fc.configLock.Unlock()
  				return fc.config.Encapsulation
  			}()
              // 封装能力变更
  			if msg.IpipEnabled != encap.IPIPEnabled || msg.VxlanEnabled != encap.VXLANEnabled ||
  				msg.VxlanEnabledV6 != encap.VXLANEnabledV6 {
                  // 终止进程
  				fc.shutDownProcess(reasonEncapChanged)
  			}
  		}
          
          // 向dataplane.toDataplane推送消息
  		if err := fc.dataplane.SendMessage(msg); err != nil {
              // 推送失败终止进程
  			fc.shutDownProcess("Failed to write to dataplane driver")
  		}
  	}
  }
  
  func (fc *DataplaneConnector) readMessagesFromDataplane() {
  	defer func() {
          // 崩溃终止进程
  		fc.shutDownProcess("Failed to read messages from dataplane")
  	}()
      
  	for {
          // dataplane推送的消息
  		payload, err := fc.dataplane.RecvMessage()
  		if err != nil {
              // 消费失败重置进程
  			fc.shutDownProcess("Failed to read from front-end socket")
  		}
          
  		switch msg := payload.(type) {
          // 进程状态消息 
  		case *proto.ProcessStatusUpdate:
  			fc.handleProcessStatusUpdate(context.TODO(), msg)
          // workload endpoint状态变更消息
  		case *proto.WorkloadEndpointStatusUpdate:
  			if fc.statusReporter != nil {
  				fc.StatusUpdatesFromDataplane <- msg
  			}
          // workload endpoint状态删除消息
  		case *proto.WorkloadEndpointStatusRemove:
  			if fc.statusReporter != nil {
  				fc.StatusUpdatesFromDataplane <- msg
  			}
          // host endpoint状态变更消息
  		case *proto.HostEndpointStatusUpdate:
  			if fc.statusReporter != nil {
  				fc.StatusUpdatesFromDataplane <- msg
  			}
          // host endpoint状态删除消息
  		case *proto.HostEndpointStatusRemove:
  			if fc.statusReporter != nil {
  				fc.StatusUpdatesFromDataplane <- msg
  			}
          // wireGuard tunnel状态消息
  		case *proto.WireguardStatusUpdate:
  			fc.wireguardStatUpdateFromDataplane <- msg
  		...
  		}
  	}
  }
  
  
  func (fc *DataplaneConnector) handleWireguardStatUpdateFromDataplane() {
  	defer func() {
          // 崩溃终止进程
  		fc.shutDownProcess("Failed to read messages from dataplane")
  	}()
  
  	for {
  		select {
          // wireGuard tunnel状态消息
  		case current = <-fc.wireguardStatUpdateFromDataplane:
  		// 周期重试任务
  		case <-retryC:
  		...
  		}
          
          // 先终止定时器
  		if ticker != nil {
  			ticker.Stop()
  		}
  
  		// wireGuard状态对齐到期望状态
  		err := fc.reconcileWireguardStatUpdate(current.PublicKey, current.IpVersion)
          // 成功同步终止重试
  		if err == nil {
  			current = nil
  			retryC = nil
  			ticker = nil
          // 失败设置2~4s重试窗口
  		} else {
  			// retry reconciling between 2-4 seconds.
  			ticker = jitter.NewTicker(2*time.Second, 2*time.Second)
  			retryC = ticker.C
  		}
  	}
  }
  

• 

• 注意

  dpconnector主要负责将calGraph事件推到dataplane及处理wireGuard状态对齐，其它状态消息会上报给status reporter

2.4.handleConf

• dp.handleConfigUpdate()用于处理上游配置变更事件，合并配置过程会检测进程重启条件及适当终止进程，应用可热更新配置至相应模块。

• func (fc *DataplaneConnector) handleConfigUpdate(msg *proto.ConfigUpdate) {
      // msg配置转为source-->raw结构
  	sourceToRaw := map[string]map[string]string{}
  	for _, kvs := range msg.SourceToRawConfig {
  		sourceToRaw[kvs.Source] = kvs.Config
  	}
  	...
  
  	err := func() error {
  		...
          
          // 存量配置
  		oldConfigCopy = fc.config.Copy()
  		...
          
          // 合并最新配置
  		changedFields, err = fc.config.UpdateFromConfigUpdate(msg)
  		newConfigCopy = fc.config.Copy()
  		return err
  	}()
  
      // 配置合并异常终止进程
  	if err != nil {
  		// This shouldn't happen since the config update was _generated_ by the Config object held
  		// by the calculation graph.
  		fc.shutDownProcess(reasonConfigUpdateFailed)
  	}
  	...
      
      // 核心配置变更,标记重启进程
  	changedFields.Iter(func(fieldName string) error {
  		if !handledConfigChanges.Contains(fieldName) {
  			restartNeeded = true
  		}
  		return nil
  	})
  
      // 终止进程
  	if restartNeeded {
  		fc.shutDownProcess(reasonConfigChanged)
  	}
  
      // 应用global health检测超时配置
  	if changedFields.Len() > 0 {
  		fc.ApplyNoRestartConfig(oldConfigCopy, newConfigCopy)
  	}
  	...
  }
  

• 

• 注意

  handleConfigUpdate会将最新配置合并，核心配置变更发送进程重启信号

2.5.handleProcess

• dp.handleProcessStatusUpdate()用于向datastore上报进程状态，基于节点存活检查实现datastore事件分发的全局观测及故障判断。

• func (fc *DataplaneConnector) handleProcessStatusUpdate(ctx context.Context, msg *proto.ProcessStatusUpdate) {
      // 状态报文
  	statusReport := model.StatusReport{
          Timestamp:     msg.IsoTimestamp,
          UptimeSeconds: msg.Uptime,
          FirstUpdate:   !fc.firstStatusReportSent,
      }
  
  	func() {
  		...
          // 获取hostname
  		hostname = fc.config.FelixHostname
          // 获取节点所属region
  		regionString = model.RegionString(fc.config.OpenstackRegion)
          // 获取上报超时TTL
  		reportingTTL = fc.config.ReportingTTLSecs
  	}()
  
      // 初始化active status
  	kv := model.KVPair{
  		Key:   model.ActiveStatusReportKey{Hostname: hostname, RegionString: regionString},
  		Value: &statusReport,
  		TTL:   reportingTTL,
  	}
      
      // 写入datstore
  	applyCtx, cancel := context.WithTimeout(ctx, 2*time.Second)
  	_, err := fc.datastore.Apply(applyCtx, &kv)
  	cancel()
  	...
      // 标记首次上报完成
      fc.firstStatusReportSent = true
      
      // 向datastore上报last status
  	kv = model.KVPair{
  		Key:   model.LastStatusReportKey{Hostname: hostname, RegionString: regionString},
  		Value: &statusReport,
  	}
  	applyCtx, cancel = context.WithTimeout(ctx, 2*time.Second)
  	_, err = fc.datastore.Apply(applyCtx, &kv)
  	cancel()
  	...
  }
  

• 

• 注意

  handleProcessStatusUpdate会将进程状态信息上报给datastore，影响后续的事件规则推送

2.6.handleWire

• dp.reconcileWireguardStatUpdate()基于dataplane上报的wireguard公钥更新calico-node对象，确保网络传输数据加密。

• func (fc *DataplaneConnector) reconcileWireguardStatUpdate(dpPubKey string, ipVersion proto.IPVersion) error {
  	// 重试3次
  	for iter := 0; iter < 3; iter++ {
  		...
  
          // 获取nodeName
  		felixHostname := func() string {
  			...
  			return fc.config.FelixHostname
  		}()
  
          // 由datastore获取calico-node
  		node, err := fc.datastorev3.Nodes().Get(getCtx, felixHostname, options.GetOptions{})
  		...
  
  		// 获取node公钥
  		storedPublicKey := node.Status.WireguardPublicKey
  		if ipVersion == proto.IPVersion_IPV6 {
  			storedPublicKey = node.Status.WireguardPublicKeyV6
  		} else if ipVersion != proto.IPVersion_IPV4 {
  			return fmt.Errorf("Unknown IP version: %d", ipVersion)
  		}
          
          // 公钥变化
  		if storedPublicKey != dpPubKey {
              // 设置calico-node公钥
  			updateCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
  			if ipVersion == proto.IPVersion_IPV4 {
  				node.Status.WireguardPublicKey = dpPubKey
  			} else if ipVersion == proto.IPVersion_IPV6 {
  				node.Status.WireguardPublicKeyV6 = dpPubKey
  			}
              
              // 更新calico-node
  			_, err := fc.datastorev3.Nodes().Update(updateCtx, node, options.SetOptions{})
  			...
              
  			if err != nil {
  				// check if failure is recoverable
  				switch err.(type) {
  				case cerrors.ErrorResourceUpdateConflict:
  					continue
  				}
  				// retry in some time.
  				return err
  			}
  		}
  		break
  	}
  	return nil
  }
  

• 

• 注意

  reconcileWireguardStatUpdate会基于dataplane推送的wireguard公钥对比更新calico-node公钥