【kubernetes/k8s源码分析】calico node felix源码分析之二

怼烎@ 2021-11-17 01:56 421阅读 0赞

>     // Now create the calculation graph, which receives updates from the
>     // datastore and outputs dataplane updates for the dataplane driver.
>     //
>     // The Syncer has its own thread and we use an extra thread for the
>     // Validator, just to pipeline that part of the calculation then the
>     // main calculation graph runs in a single thread for simplicity.
>     // The output of the calculation graph arrives at the dataplane
>     // connection via channel.
>     //
>     // Syncer -chan-> Validator -chan-> Calc graph -chan->   dataplane
>     //        KVPair            KVPair             protobufs

` Syncer `协程负责监听 datastore 中的更新，并将更新的内容通过 channel 发送给 `Validator `协程。Validator 完成校验后，将其发送给 `Calc graph `协程。Calc graph 完成计算后，发送给dataplane协程。最后dataplane完成数据平面處理

>     // Get a Syncer from the datastore, or a connection to our remote sync daemon, Typha,
>     // which will feed the calculation graph with updates, bringing Felix into sync.
>     var syncer Startable
>     var typhaConnection *syncclient.SyncerClient
>     syncerToValidator := calc.NewSyncerCallbacksDecoupler()

# 1. NewSyncerCallbacksDecoupler 汉书实例化 SyncerCallbacksDecoupler #

func NewSyncerCallbacksDecoupler() *SyncerCallbacksDecoupler {
    	return &SyncerCallbacksDecoupler{
    		c: make(chan interface{}),
    	}
    }
    
    type SyncerCallbacksDecoupler struct {
    	c chan interface{}
    }

###     1.1 三個方法比较简单，主要是往 channel 发送数据 ###

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 {
    		switch obj := obj.(type) {
    		case api.SyncStatus:
    			sink.OnStatusUpdated(obj)
    		case []api.Update:
    			sink.OnUpdates(obj)
    		}
    	}
    }

# 2. 创建一個 felix syncer #

// New creates a new Felix v1 Syncer.
    func New(client api.Client, cfg apiconfig.CalicoAPIConfigSpec, callbacks api.SyncerCallbacks) api.Syncer {
    	// Create the set of ResourceTypes required for Felix.  Since the update processors
    	// also cache state, we need to create individual ones per syncer rather than create
    	// a common global set.

###     2.1 ResourceListOptions 结构体 ###

> type ResourceListOptions struct \{  
>    // The name of the resource.  
>    Name string  
>    // The namespace of the resource.  Not required if the resource is not namespaced.  
>    Namespace string  
>    // The resource kind.  
>    Kind string  
>    // Whether the name is prefix rather than the full name.  
>    Prefix bool  
> \}

###     2.2 资源类型有 ###

*  ClusterInformation
 *  FelixConfiguration
 *  GlobalNetworkPolicy
 *  GlobalNetworkSet
 *  IPPool
 *  Node
 *  Profile
 *  WorkloadEndpoint
 *  NetworkPolicy
 *  NetworkSet
 *  HostEndpoint

resourceTypes := []watchersyncer.ResourceType{
    	{
    		ListInterface:   model.ResourceListOptions{Kind: apiv3.KindClusterInformation},
    		UpdateProcessor: updateprocessors.NewClusterInfoUpdateProcessor(),
    	},
    	{
    		ListInterface:   model.ResourceListOptions{Kind: apiv3.KindFelixConfiguration},
    		UpdateProcessor: updateprocessors.NewFelixConfigUpdateProcessor(),
    	},
    	{
    		ListInterface:   model.ResourceListOptions{Kind: apiv3.KindGlobalNetworkPolicy},
    		UpdateProcessor: updateprocessors.NewGlobalNetworkPolicyUpdateProcessor(),
    	},
    	{
    		ListInterface:   model.ResourceListOptions{Kind: apiv3.KindGlobalNetworkSet},
    		UpdateProcessor: updateprocessors.NewGlobalNetworkSetUpdateProcessor(),
    	},
    }

watchersyncer.New 第 5 章节讲解

// If using Calico IPAM, include IPAM resources the felix cares about.
    if !cfg.K8sUsePodCIDR {
    	resourceTypes = append(resourceTypes, watchersyncer.ResourceType{ListInterface: model.BlockListOptions{}})
    }
    
    return watchersyncer.New(
    	client,
    	resourceTypes,
    	callbacks,
    )

# 3. NewAsyncCalcGraph 函數 #

实例化 CalcGraph，ipsets/active policy calculation graph，包含了两个主要成员

NewCalculationGraph 注册了一大堆，请看下文

>     // AllUpdDispatcher is the input node to the calculation graph.
>     AllUpdDispatcher      *dispatcher.Dispatcher
>     activeRulesCalculator *ActiveRulesCalculator

func NewAsyncCalcGraph(
    	conf *config.Config,
    	outputChannels []chan<- interface{},
    	healthAggregator *health.HealthAggregator,
    ) *AsyncCalcGraph {
    	eventBuffer := NewEventSequencer(conf)
    	calcGraph := NewCalculationGraph(eventBuffer, conf)
    	g := &AsyncCalcGraph{
    		CalcGraph:        calcGraph,
    		inputEvents:      make(chan interface{}, 10),
    		outputChannels:   outputChannels,
    		eventBuffer:      eventBuffer,
    		healthAggregator: healthAggregator,
    	}
    	if conf.DebugSimulateCalcGraphHangAfter != 0 {
    		log.WithField("delay", conf.DebugSimulateCalcGraphHangAfter).Warn(
    			"Simulating a calculation graph hang.")
    		g.debugHangC = time.After(conf.DebugSimulateCalcGraphHangAfter)
    	}
    	eventBuffer.Callback = g.onEvent
    	if healthAggregator != nil {
    		healthAggregator.RegisterReporter(healthName, &health.HealthReport{Live: true, Ready: true}, healthInterval*2)
    	}
    	return g
    }

###     3.1 NewCalculationGraph 函數 ###

allUpdDispatcher 接收所有更新从 datastore 到 valid 到 calc graph，将他们分发给注册的接收方进行处理

>     // 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

func NewCalculationGraph(callbacks PipelineCallbacks, conf *config.Config) *CalcGraph {
    	hostname := conf.FelixHostname
    	log.Infof("Creating calculation graph, filtered to hostname %v", hostname)
    
    	allUpdDispatcher := dispatcher.NewDispatcher()

**    3.1.1 localEndpointDispatcher 用来过滤非本地的 endpoints**

有些 receivers 只需要知道本地的 endpoints

> // 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  
> //

localEndpointDispatcher := dispatcher.NewDispatcher()
    (*localEndpointDispatcherReg)(localEndpointDispatcher).RegisterWith(allUpdDispatcher)
    localEndpointFilter := &endpointHostnameFilter{hostname: hostname}
    localEndpointFilter.RegisterWith(localEndpointDispatcher)

**     3.1.1.1 调用 dispatcher 的 Register 方法注册 handler**

<table> 
 <tbody> 
  <tr> 
   <td> <pre><code>allUpdDispatcher</code></pre> <p>&nbsp;</p> </td> 
   <td> <pre><code>WorkloadEndpointKey: localEndpointDispatcher.OnUpdate</code></pre> </td> 
  </tr> 
  <tr> 
   <td> <pre><code>HostEndpointKey: localEndpointDispatcher.OnUpdate</code></pre> </td> 
  </tr> 
  <tr> 
   <td> <pre><code>statusHandlers: localEndpointDispatcher.OnDatamodelStatus
</code></pre> </td> 
  </tr> 
  <tr> 
   <td>&nbsp;</td> 
  </tr> 
 </tbody> 
</table>

func (l *localEndpointDispatcherReg) RegisterWith(disp *dispatcher.Dispatcher) {
    	led := (*dispatcher.Dispatcher)(l)
    	disp.Register(model.WorkloadEndpointKey{}, led.OnUpdate)
    	disp.Register(model.HostEndpointKey{}, led.OnUpdate)
    	disp.RegisterStatusHandler(led.OnDatamodelStatus)
    }

**    3.1.2 注册 active 规则**

根据 policies 和 profiles 计算那些在 host 上是起作用的

>     // 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
>     //             ...
>     //

activeRulesCalc := NewActiveRulesCalculator()
    activeRulesCalc.RegisterWith(localEndpointDispatcher, allUpdDispatcher)

**     3.1.2.1 注册 ActiveRulesCalculator**

<table> 
 <tbody> 
  <tr> 
   <td colspan="1"> <pre><code>localEndpointDispatcher</code></pre> </td> 
   <td> <pre><code>WorkloadEndpointKey： ActiveRulesCalculator.OnUpdate</code></pre> </td> 
  </tr> 
  <tr> 
   <td> <pre><code>HostEndpointKey： ActiveRulesCalculator.OnUpdate</code></pre> </td> 
  </tr> 
  <tr> 
   <td colspan="1">allUpdDispatcher</td> 
   <td> <pre><code>PolicyKey： ActiveRulesCalculator.OnUpdate</code></pre> </td> 
  </tr> 
  <tr> 
   <td> <pre><code>ProfileRulesKey： ActiveRulesCalculator.OnUpdate</code></pre> </td> 
  </tr> 
  <tr> 
   <td> <pre><code>ProfileLabelsKey： ActiveRulesCalculator.OnUpdate</code></pre> </td> 
  </tr> 
  <tr> 
   <td> <pre><code>ProfileTagsKey： ActiveRulesCalculator.OnUpdate</code></pre> </td> 
  </tr> 
  <tr> 
   <td>&nbsp;</td> 
  </tr> 
 </tbody> 
</table>

func (arc *ActiveRulesCalculator) RegisterWith(localEndpointDispatcher, allUpdDispatcher *dispatcher.Dispatcher) {
    	// It needs the filtered endpoints...
    	localEndpointDispatcher.Register(model.WorkloadEndpointKey{}, arc.OnUpdate)
    	localEndpointDispatcher.Register(model.HostEndpointKey{}, arc.OnUpdate)
    	// ...as well as all the policies and profiles.
    	allUpdDispatcher.Register(model.PolicyKey{}, arc.OnUpdate)
    	allUpdDispatcher.Register(model.ProfileRulesKey{}, arc.OnUpdate)
    	allUpdDispatcher.Register(model.ProfileLabelsKey{}, arc.OnUpdate)
    	allUpdDispatcher.Register(model.ProfileTagsKey{}, arc.OnUpdate)
    	allUpdDispatcher.RegisterStatusHandler(arc.OnStatusUpdate)
    }

**   3.1.3 runScanner 用来从 rules 抽出信息 tags/selectors/named ports**

>     // 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, tags, and named ports.  It
>     // generates events when a new selector/tag/named port starts/stops being used.
>     //
>     //             ...
>     //     Active Rules Calculator
>     //              |
>     //              | Locally active policies/profiles
>     //              |
>     //         Rule scanner
>     //          |    \
>     //          |     \ Locally active tags/selectors/named ports
>     //          |      \
>     //          |      ...
>     //          |
>     //          | IP set active/inactive
>     //          |
>     //     <dataplane>
>     //

ruleScanner := NewRuleScanner()
    // Wire up the rule scanner's inputs.
    activeRulesCalc.RuleScanner = ruleScanner
    // Send IP set added/removed events to the dataplane.  We'll hook up the other outputs
    // below.
    ruleScanner.RulesUpdateCallbacks = callbacks

**    3.1.4 计算哪些 endpoints 在每一个 tag/selector/named 端口**

**     **ipsetMemberIndex 计算在每一个 IP set  中的 IP 集合和 named ports，主要是通过  active selectors/tags/named ports 进行匹配

>     // The rule scanner only goes as far as figuring out which tags/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/tags/named
>     // ports extracted by the rule scanner against all the endpoints.
>     //
>     //        ...
>     //     Dispatcher (all updates)
>     //      |
>     //      | All endpoints
>     //      |
>     //      |       ...
>     //      |    Rule scanner
>     //      |     |       \
>     //      |    ...       \ Locally active tags/selectors/named ports
>     //       \              |
>     //        \_____        |
>     //              \       |
>     //            IP set member index
>     //                   |
>     //                   | IP set member added/removed
>     //                   |
>     //               <dataplane>
>     //

<table> 
 <tbody> 
  <tr> 
   <td> <pre><code>allUpdDispatcher</code></pre> </td> 
   <td> <pre><code>ProfileTagsKey: SelectorAndNamedPortIndex.OnUpdate</code></pre> </td> 
  </tr> 
  <tr> 
   <td> <pre><code>ProfileLabelsKey: SelectorAndNamedPortIndex.OnUpdate</code></pre> </td> 
  </tr> 
  <tr> 
   <td> <pre><code>WorkloadEndpointKey: SelectorAndNamedPortIndex.OnUpdate</code></pre> </td> 
  </tr> 
  <tr> 
   <td> <pre><code>HostEndpointKey: SelectorAndNamedPortIndex.OnUpdate</code></pre> </td> 
  </tr> 
  <tr> 
   <td> <pre><code>NetworkSetKey: SelectorAndNamedPortIndex.OnUpdate</code></pre> </td> 
  </tr> 
 </tbody> 
</table>

ipsetMemberIndex := labelindex.NewSelectorAndNamedPortIndex()
    // Wire up the inputs to the IP set member index.
    ipsetMemberIndex.RegisterWith(allUpdDispatcher)
    ruleScanner.OnIPSetActive = func(ipSet *IPSetData) {
    	log.WithField("ipSet", ipSet).Info("IPSet now active")
    	callbacks.OnIPSetAdded(ipSet.UniqueID(), ipSet.DataplaneProtocolType())
    	ipsetMemberIndex.UpdateIPSet(ipSet.UniqueID(), ipSet.Selector, ipSet.NamedPortProtocol, ipSet.NamedPort)
    	gaugeNumActiveSelectors.Inc()
    }
    ruleScanner.OnIPSetInactive = func(ipSet *IPSetData) {
    	log.WithField("ipSet", ipSet).Info("IPSet now inactive")
    	ipsetMemberIndex.DeleteIPSet(ipSet.UniqueID())
    	callbacks.OnIPSetRemoved(ipSet.UniqueID())
    	gaugeNumActiveSelectors.Dec()
    }

**    3.1.5 发送 IP set 给数据平面**

// Send the IP set member index's outputs to the dataplane.
    ipsetMemberIndex.OnMemberAdded = func(ipSetID string, member labelindex.IPSetMember) {
    	if log.GetLevel() >= log.DebugLevel {
    		log.WithFields(log.Fields{
    			"ipSetID": ipSetID,
    			"member":  member,
    		}).Debug("Member added to IP set.")
    	}
    	callbacks.OnIPSetMemberAdded(ipSetID, member)
    }
    ipsetMemberIndex.OnMemberRemoved = func(ipSetID string, member labelindex.IPSetMember) {
    	if log.GetLevel() >= log.DebugLevel {
    		log.WithFields(log.Fields{
    			"ipSetID": ipSetID,
    			"member":  member,
    		}).Debug("Member removed from IP set.")
    	}
    	callbacks.OnIPSetMemberRemoved(ipSetID, member)
    }

**    3.1.6 polResolver 汇总本地 endpoints 的 active 策略，计算完整、有序的策略集，然后应用于每个端点**

>     // 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>
>     //

<table> 
 <tbody> 
  <tr> 
   <td> <pre><code>allUpdDispatcher</code></pre> </td> 
   <td> <pre><code>PolicyKey: PolicyResolver.OnUpdate</code></pre> </td> 
  </tr> 
  <tr> 
   <td colspan="1"> <pre><code>localEndpointDispatcher</code></pre> </td> 
   <td> <pre><code>WorkloadEndpointKey: PolicyResolver.OnUpdate</code></pre> </td> 
  </tr> 
  <tr> 
   <td> <pre><code>HostEndpointKey: PolicyResolver.OnUpdate</code></pre> </td> 
  </tr> 
  <tr> 
   <td> <pre><code>statusHandlers: PolicyResolver.OnDatamodelStatus</code></pre> </td> 
  </tr> 
 </tbody> 
</table>

polResolver := NewPolicyResolver()
    // Hook up the inputs to the policy resolver.
    activeRulesCalc.PolicyMatchListener = polResolver
    polResolver.RegisterWith(allUpdDispatcher, localEndpointDispatcher)
    // And hook its output to the callbacks.
    polResolver.Callbacks = callbacks

**    3.1.7 为 host IP 更新进行注册**

>     // Register for host IP updates.
>     //
>     //        ...
>     //     Dispatcher (all updates)
>     //         |
>     //         | host IPs
>     //         |
>     //       passthru
>     //         |
>     //         |
>     //         |
>     //      <dataplane>
>     //

<table> 
 <tbody> 
  <tr> 
   <td> <pre><code>allUpdDispatcher</code></pre> </td> 
   <td> <pre><code>HostIPKey: DataplanePassthru.OnUpdate</code></pre> </td> 
  </tr> 
  <tr> 
   <td> <pre><code>IPPoolKey: DataplanePassthru.OnUpdate</code></pre> </td> 
  </tr> 
 </tbody> 
</table>

hostIPPassthru := NewDataplanePassthru(callbacks)
    hostIPPassthru.RegisterWith(allUpdDispatcher)

**    3.1.8 vxlanResolver 计算 VXLAN 路由**

if conf.VXLANEnabled {
    	vxlanResolver := NewVXLANResolver(hostname, callbacks)
    	vxlanResolver.RegisterWith(allUpdDispatcher)
    }

**    3.1.9 configBatcher 用于配置更新**

>     // Register for config updates.
>     //
>     //        ...
>     //     Dispatcher (all updates)
>     //         |
>     //         | separate config updates foo=bar, baz=biff
>     //         |
>     //       config batcher
>     //         |
>     //         | combined config {foo=bar, bax=biff}
>     //         |
>     //      <dataplane>
>     //

<table> 
 <tbody> 
  <tr> 
   <td> <pre><code>allUpdDispatcher</code></pre> </td> 
   <td> <pre><code>GlobalConfigKey: ConfigBatcher.OnUpdate</code></pre> </td> 
  </tr> 
  <tr> 
   <td> <pre><code>HostConfigKey: ConfigBatcher.OnUpdate</code></pre> </td> 
  </tr> 
  <tr> 
   <td> <pre><code>ReadyFlagKey: ConfigBatcher.OnUpdate</code></pre> </td> 
  </tr> 
  <tr> 
   <td> <pre><code>StatusHandler: ConfigBatcher.OnDatamodelStatus</code></pre> </td> 
  </tr> 
 </tbody> 
</table>

configBatcher := NewConfigBatcher(hostname, callbacks)
    configBatcher.RegisterWith(allUpdDispatcher)

**    3.1.10 profileDecoder **

>     // 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>
>     //

profileDecoder := NewProfileDecoder(callbacks)
    profileDecoder.RegisterWith(allUpdDispatcher)

###    总结 allUpdDispatcher 中注册的 handler ###

<table> 
 <tbody> 
  <tr> 
   <td colspan="1">localEndpointDispatcher</td> 
   <td>WorkloadEndpointKey</td> 
  </tr> 
  <tr> 
   <td>HostEndpointKey</td> 
  </tr> 
  <tr> 
   <td>statusHandlers</td> 
  </tr> 
  <tr> 
   <td colspan="1">ActiveRulesCalculator</td> 
   <td>PolicyKey</td> 
  </tr> 
  <tr> 
   <td>ProfileRulesKey</td> 
  </tr> 
  <tr> 
   <td>ProfileLabelsKey</td> 
  </tr> 
  <tr> 
   <td>ProfileTagsKey</td> 
  </tr> 
  <tr> 
   <td>statusHandlers</td> 
  </tr> 
  <tr> 
   <td colspan="1">SelectorAndNamedPortIndex</td> 
   <td>ProfileTagsKey</td> 
  </tr> 
  <tr> 
   <td>ProfileLabelsKey</td> 
  </tr> 
  <tr> 
   <td>WorkloadEndpointKey</td> 
  </tr> 
  <tr> 
   <td>HostEndpointKey</td> 
  </tr> 
  <tr> 
   <td>NetworkSetKey</td> 
  </tr> 
  <tr> 
   <td>PolicyResolver</td> 
   <td>PolicyKey</td> 
  </tr> 
  <tr> 
   <td colspan="1">DataplanePassthru</td> 
   <td>HostIPKey</td> 
  </tr> 
  <tr> 
   <td>IPPoolKey</td> 
  </tr> 
  <tr> 
   <td colspan="1">ConfigBatcher</td> 
   <td>GlobalConfigKey</td> 
  </tr> 
  <tr> 
   <td>HostConfigKey</td> 
  </tr> 
  <tr> 
   <td>ReadyFlagKey</td> 
  </tr> 
  <tr> 
   <td>statusHandlers</td> 
  </tr> 
  <tr> 
   <td colspan="1">ProfileDecoder</td> 
   <td>ProfileLabelsKey</td> 
  </tr> 
 </tbody> 
</table>

###    总结 localEndpointDispatcher 中注册的 handler ###

<table> 
 <tbody> 
  <tr> 
   <td colspan="1">ActiveRulesCalculator</td> 
   <td>WorkloadEndpointKey</td> 
  </tr> 
  <tr> 
   <td>HostEndpointKey</td> 
  </tr> 
  <tr> 
   <td colspan="1">PolicyResolver</td> 
   <td>WorkloadEndpointKey</td> 
  </tr> 
  <tr> 
   <td>HostEndpointKey</td> 
  </tr> 
  <tr> 
   <td>localEndpointFilter</td> 
   <td>&nbsp;</td> 
  </tr> 
 </tbody> 
</table>

###     3.2 Start 函数 ###

func (acg *AsyncCalcGraph) Start() {
    	log.Info("Starting AsyncCalcGraph")
    	flushTicker := time.NewTicker(tickInterval)
    	acg.flushTicks = flushTicker.C
    	go acg.loop()
    }

**     3.2.1 AsyncCalcGraph loop**

如果 update 类型为 \[\]api.Update ，发送到 disaptcher

case update := <-acg.inputEvents:
    	switch update := update.(type) {
    	case []api.Update:
    		// Update; send it to the dispatcher.
    		log.Debug("Pulled []KVPair off channel")
    		updStartTime := time.Now()
    		acg.AllUpdDispatcher.OnUpdates(update)
    		summaryUpdateTime.Observe(time.Since(updStartTime).Seconds())
    		// Record stats for the number of messages processed.
    		for _, upd := range update {
    			typeName := reflect.TypeOf(upd.Key).Name()
    			count := countUpdatesProcessed.WithLabelValues(typeName)
    			count.Inc()
    		}

# 4. 实例化 ValidationFilter #

作为 syncer 和 calculation graph 的中间人

func NewValidationFilter(sink api.SyncerCallbacks) *ValidationFilter {
    	return &ValidationFilter{
    		sink: sink,
    	}
    }
    
    type ValidationFilter struct {
    	sink api.SyncerCallbacks
    }

###     4.1 OnUpdates 函数 ###

遍历所有更新 updates，最后调用的是 AsyncCalcGraph 的 OnUpdates 方法，主要是发送到 channel 中

func (v *ValidationFilter) OnUpdates(updates []api.Update) {
    	filteredUpdates := make([]api.Update, len(updates))
    	for i, update := range updates {
    		if update.Value != nil {
    			val := reflect.ValueOf(update.Value)
    			if val.Kind() == reflect.Ptr {
    				elem := val.Elem()
    				if elem.Kind() == reflect.Struct {
    					if err := validator.Validate(elem.Interface()); err != nil {
    						logCxt.WithError(err).Warn("Validation failed; treating as missing")
    						update.Value = nil
    					}
    				}
    			}
    
    			switch v := update.Value.(type) {
    			case *model.WorkloadEndpoint:
    				if v.Name == "" {
    					logCxt.WithError(errors.New("Missing name")).Warn("Validation failed; treating as missing")
    					update.Value = nil
    				}
    			}
    		}
    		filteredUpdates[i] = update
    	}
    	v.sink.OnUpdates(filteredUpdates)
    }

# 5. syncer New 方法 #

路径 projectcalico/libcalico-go/lib/backend/watchersyncer/watchersyncer.go

callbacks 是 syncerToValidator := calc.NewSyncerCallbacksDecoupler()

// 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
    }

###     5.1 Start 函数 ###

wgwc 用来所有 watcher cacher goroutines，wgws 用来 syncer watcher

func (ws *watcherSyncer) Start() {
    	log.Info("Start called")
    
    	// Create a context and a wait group.
    	// The context is passed to the run() method where it is passed on to all the watcher caches.
    	// The cancel function is stored off and when called it signals to the caches that they need to wrap up their work.
    	// watcher caches wait group (wswc) is used to signal the completion of all of the watcher cache goroutines.
    	// watcher syncer wait group (wswc) is used to signal the completion of the watcher syncer itself.
    	ctx, cancel := context.WithCancel(context.Background())
    	ws.cancel = cancel
    	ws.wgwc = &sync.WaitGroup{}
    	ws.wgws = &sync.WaitGroup{}
    
    	go func() {
    		ws.run(ctx)
    		log.Debug("Watcher syncer run completed")
    	}()
    }

###     5.2 run 函数 ###

实现了 syncer 循环接收 watch 事件，传送 syncer updates，首先发送状态 WaitForDatastore （wait-for-ready）

// run implements the main syncer loop that loops forever receiving watch events and translating
    // to syncer updates.
    func (ws *watcherSyncer) run(ctx context.Context) {
    	log.Debug("Sending initial status event and starting watchers")
    	ws.wgws.Add(1)
    	ws.sendStatusUpdate(api.WaitForDatastore)

**   5.2.1 对于所有注册的 watcher cahcers 异步执行 run 方法**

for _, wc := range ws.watcherCaches {
    	ws.wgwc.Add(1)
    	go func(wc *watcherCache) {
    		wc.run(ctx)
    		log.Debug("Watcher cache run completed")
    		ws.wgwc.Done()
    	}(wc)
    }

watchercache 第 6 章讲解

###     5.3 sendStatusUpdate 函数 ###

调用 func (a \*SyncerCallbacksDecoupler) OnStatusUpdated(status api.SyncStatus)，将 status 发送到 channel

// Send a status update and store the status.
    func (ws *watcherSyncer) sendStatusUpdate(status api.SyncStatus) {
    	log.WithField("Status", status).Info("Sending status update")
    	ws.callbacks.OnStatusUpdated(status)
    	ws.status = status
    }

# 6. watcherCache 结构体 #

// The watcherCache provides watcher/syncer support for a single key type in the
    // backend.  These results are sent to the main WatcherSyncer on a buffered "results"
    // channel.  To ensure the order of events is received correctly by the main WatcherSyncer,
    // we send all notification types in this channel.  Note that because of this the results
    // channel is untyped - however the watcherSyncer only expects one of the following
    // types:
    // -  An error
    // -  An api.Update
    // -  A api.SyncStatus (only for the very first InSync notification)
    type watcherCache struct {
    	logger               *logrus.Entry
    	client               api.Client
    	watch                api.WatchInterface
    	resources            map[string]cacheEntry
    	oldResources         map[string]cacheEntry
    	results              chan<- interface{}
    	hasSynced            bool
    	errors               int
    	resourceType         ResourceType
    	currentWatchRevision string
    }

### **    6.1 newWatcherCache 实例化 watcherCache** ###

resourceType 主要是 2.2 章节注册的一大堆

// Create a new watcherCache.
    func newWatcherCache(client api.Client, resourceType ResourceType, results chan<- interface{}) *watcherCache {
    	return &watcherCache{
    		logger:       logrus.WithField("ListRoot", model.ListOptionsToDefaultPathRoot(resourceType.ListInterface)),
    		client:       client,
    		resourceType: resourceType,
    		results:      results,
    		resources:    make(map[string]cacheEntry, 0),
    	}
    }

###     6.2 run 函数 ###

// run creates the watcher and loops indefinitely reading from the watcher.
    func (wc *watcherCache) run(ctx context.Context) {
    	wc.logger.Debug("Watcher cache starting, start initial sync processing")
    	wc.resyncAndCreateWatcher(ctx)

**     6.2.1 resyncAndCreateWatcher 函数 **

循环重新同步过程，直到成功的完成了重新同步以及启动一个 watcher

// resyncAndCreateWatcher loops performing resync processing until it successfully
    // completes a resync and starts a watcher.
    func (wc *watcherCache) resyncAndCreateWatcher(ctx context.Context) {
    	// The passed in context allows a resync to be stopped mid-resync. The resync should be stopped as quickly as
    	// possible, but there should be usable data available in wc.resources so that delete events can be sent.
    	// The strategy is to
    	// - cancel any long running functions calls made from here, i.e. pass ctx to the client.list() calls
    	//    - but if it finishes, then ensure that the listing gets processed.
    	// - cancel any sleeps if the context is cancelled

**     6.2.1.1 调用资源 API List 方法**

OnSyncerStarting 每一个资源都有该方法，调用 List API 方法获得资源

if performFullResync {
    	wc.logger.Debug("Full resync is required")
    
    	// Notify the converter that we are resyncing.
    	if wc.resourceType.UpdateProcessor != nil {
    		wc.logger.Debug("Trigger converter resync notification")
    		wc.resourceType.UpdateProcessor.OnSyncerStarting()
    	}
    
    	// Start the sync by Listing the current resources.
    	l, err := wc.client.List(ctx, wc.resourceType.ListInterface, "")

**     6.2.1.2 Watch 从当前版本开始**

// And now start watching from the revision returned by the List, or from a previous watch event
    // (depending on whether we were performing a full resync).
    w, err := wc.client.Watch(ctx, wc.resourceType.ListInterface, wc.currentWatchRevision)

6.2.2 如果 watch nil 或者 ctx Done 则重新开始

for {
    	if wc.watch == nil {
    		// The watcher will be nil if the context cancelled during a resync.
    		wc.logger.Debug("Watch is nil. Returning")
    		break mainLoop
    	}
    	select {
    	case <-ctx.Done():
    		wc.logger.Debug("Context is done. Returning")
    		wc.cleanExistingWatcher()
    		break mainLoop

**   6.2.3 循环处理事件**

case event, ok := <-wc.watch.ResultChan():
    	if !ok {
    		// If the channel is closed then resync/recreate the watch.
    		wc.logger.Info("Watch channel closed by remote - recreate watcher")
    		wc.resyncAndCreateWatcher(ctx)
    		continue
    	}
    	wc.logger.WithField("RC", wc.watch.ResultChan()).Debug("Reading event from results channel")
    
    	// Handle the specific event type.
    	switch event.Type {
    	case api.WatchAdded, api.WatchModified:
    		kvp := event.New
    		wc.handleWatchListEvent(kvp)
    	case api.WatchDeleted:
    		// Nil out the value to indicate a delete.
    		kvp := event.Old
    		if kvp == nil {
    			// Bug, we're about to panic when we hit the nil pointer, log something useful.
    			wc.logger.WithField("watcher", wc).WithField("event", event).Panic("Deletion event without old value")
    		}
    		kvp.Value = nil
    		wc.handleWatchListEvent(kvp)

# 7. DataplaneConnector #

ToDataplane：是发送到数据平面的

>     type DataplaneConnector struct {
>        config                     *config.Config
>        configUpdChan              chan<- map[string]string
>        ToDataplane                chan interface{}
>        StatusUpdatesFromDataplane chan interface{}
>        InSync                     chan bool
>        failureReportChan          chan<- string
>        dataplane                  dp.DataplaneDriver
>        datastore                  bapi.Client
>        statusReporter             *statusrep.EndpointStatusReporter
>     
>        datastoreInSync bool
>     
>        firstStatusReportSent bool
>     }

###     7.1 Start 函数异步执行了两个方法  ###

*  sendMessagesToDataplaneDriver异步写入到数据平面 （7.2 章节讲解）
 *  readMessagesFromDataplane 异步从数据平面读如数据 （7.3章节讲解）

func (fc *DataplaneConnector) Start() {
    	// Start a background thread to write to the dataplane driver.
    	go fc.sendMessagesToDataplaneDriver()
    
    	// Start background thread to read messages from dataplane driver.
    	go fc.readMessagesFromDataplane()
    }

###     7.2 sendMessagesToDataplaneDriver 函数 ###

异步循环发送消息到数据平面

func (fc *DataplaneConnector) sendMessagesToDataplaneDriver() {
    	defer func() {
    		fc.shutDownProcess("Failed to send messages to dataplane")
    	}()

7.2.1 如果消息类型为 InSync 意味着在进行同步数据中

msg := <-fc.ToDataplane
    switch msg := msg.(type) {
    case *proto.InSync:
    	log.Info("Datastore now in sync.")
    	if !fc.datastoreInSync {
    		log.Info("Datastore in sync for first time, sending message to status reporter.")
    		fc.datastoreInSync = true
    		fc.InSync <- true
    	}

7.2.2 如果消息类型为 ConfigUpdate

使用 var config map\[string\]string 保存

case *proto.ConfigUpdate:
    	if config != nil {
    		log.WithFields(log.Fields{
    			"old": config,
    			"new": msg.Config,
    		}).Info("Config updated, checking whether we need to restart")
    		restartNeeded := false
    		for kNew, vNew := range msg.Config {
    			logCxt := log.WithFields(log.Fields{"key": kNew, "new": vNew})
    			if vOld, prs := config[kNew]; !prs {
    				logCxt = logCxt.WithField("updateType", "add")
    			} else if vNew != vOld {
    				logCxt = logCxt.WithFields(log.Fields{"old": vOld, "updateType": "update"})
    			} else {
    				continue
    			}
    			if handledConfigChanges.Contains(kNew) {
    				logCxt.Info("Config change can be handled without restart")
    				continue
    			}
    			logCxt.Warning("Config change requires restart")
    			restartNeeded = true
    		}
    
    	if fc.configUpdChan != nil {
    		// Send the config over to the usage reporter.
    		fc.configUpdChan <- config
    	}

7.2.3 SendMessage

实现为 dataplane/linux/int\_dataplane.go 中的方法，其实就是向 todataplane channel 发送 msg

if err := fc.dataplane.SendMessage(msg); err != nil {
    	fc.shutDownProcess("Failed to write to dataplane driver")
    }

数据平面 SendMessage 方法的实现

>     func (d *InternalDataplane) SendMessage(msg interface{}) error {
>        d.toDataplane <- msg
>        return nil
>     }

#     7.3 readMessagesFromDataplane #

循环异步的从 dataplane 读取消息进行处理

func (fc *DataplaneConnector) readMessagesFromDataplane() {
    	defer func() {
    		fc.shutDownProcess("Failed to read messages from dataplane")
    	}()
    	log.Info("Reading from dataplane driver pipe...")
    	ctx := context.Background()
    	for {

7.3.1 从数据平面的 fromDataplane channel 读取数据

payload, err := fc.dataplane.RecvMessage()
    if err != nil {
    	log.WithError(err).Error("Failed to read from front-end socket")
    	fc.shutDownProcess("Failed to read from front-end socket")
    }

数据平面 RecvMessage 实现

>     func (d *InternalDataplane) RecvMessage() (interface{}, error) {
>        return <-d.fromDataplane, nil
>     }

7.3.2 如果消息类型为 proto.ProcessStatusUpdate

handleProcessStatusUpdate 函数主要是封装了 KVPair  结构图，调用后端 client 的Apply 方法

func (fc *DataplaneConnector) handleProcessStatusUpdate(ctx context.Context, msg *proto.ProcessStatusUpdate) {
    	log.Debugf("Status update from dataplane driver: %v", *msg)
    	statusReport := model.StatusReport{
    		Timestamp:     msg.IsoTimestamp,
    		UptimeSeconds: msg.Uptime,
    		FirstUpdate:   !fc.firstStatusReportSent,
    	}
    	kv := model.KVPair{
    		Key:   model.ActiveStatusReportKey{Hostname: fc.config.FelixHostname, RegionString: model.RegionString(fc.config.OpenstackRegion)},
    		Value: &statusReport,
    		TTL:   fc.config.ReportingTTLSecs,
    	}
    	applyCtx, cancel := context.WithTimeout(ctx, 2*time.Second)
    	_, err := fc.datastore.Apply(applyCtx, &kv)

7.3.3根据不同类型的消息，发送到 StatusUpdatesFromDataplane channel 中

case *proto.WorkloadEndpointStatusUpdate:
    	if fc.statusReporter != nil {
    		fc.StatusUpdatesFromDataplane <- msg
    	}
    case *proto.WorkloadEndpointStatusRemove:
    	if fc.statusReporter != nil {
    		fc.StatusUpdatesFromDataplane <- msg
    	}
    case *proto.HostEndpointStatusUpdate:
    	if fc.statusReporter != nil {
    		fc.StatusUpdatesFromDataplane <- msg
    	}
    case *proto.HostEndpointStatusRemove:
    	if fc.statusReporter != nil {
    		fc.StatusUpdatesFromDataplane <- msg
    	}
    default:
    	log.WithField("msg", msg).Warning("Unknown message from dataplane")