client-go之DeltaFIFO源码分析
1.DeltaFIFO概述
先从名字上来看，DeltaFIFO，首先它是一个FIFO，也就是一个先进先出的队列，而Delta代表变化的资源对象，其包含资源对象数据本身及其变化类型。
Delta的组成：
type Delta struct { Type DeltaType Object interface{} } DeltaFIFO的组成：
type DeltaFIFO struct { ... items map[string]Deltas queue []string ... } type Deltas []Delta 具体来说，DeltaFIFO存储着map[object key]Deltas以及object key的queue，Delta装有对象数据及对象的变化类型。输入输出方面，Reflector负责DeltaFIFO的输入，Controller负责处理DeltaFIFO的输出。
一个对象能算出一个唯一的object key，其对应着一个Deltas，所以一个对象对应着一个Deltas。
而目前Delta有4种Type，分别是: Added、Updated、Deleted、Sync。针对同一个对象，可能有多个不同Type的Delta元素在Deltas中，表示对该对象做了不同的操作，另外，也可能有多个相同Type的Delta元素在Deltas中（除Deleted外，Delted类型会被去重），比如短时间内，多次对某一个对象进行了更新操作，那么就会有多个Updated类型的Delta放入Deltas中。
2.DeltaFIFO的定义与初始化分析
2.1 DeltaFIFO struct
DeltaFIFO struct定义了DeltaFIFO的一些属性，下面挑几个重要的分析一下。
（1）lock：读写锁，操作DeltaFIFO中的items与queue之前都要先加锁；
（2）items：是个map，key根据对象算出，value为Deltas类型；
（3）queue：存储对象key的队列；
（4）keyFunc：计算对象key的函数；
// staging/src/k8s.io/client-go/tools/cache/delta_fifo.go type DeltaFIFO struct { // lock/cond protects access to 'items' and 'queue'. lock sync.RWMutex cond sync.Cond // We depend on the property that items in the set are in // the queue and vice versa, and that all Deltas in this // map have at least one Delta. items map[string]Deltas queue []string // populated is true if the first batch of items inserted by Replace() has been populated // or Delete/Add/Update was called first. populated bool // initialPopulationCount is the number of items inserted by the first call of Replace() initialPopulationCount int // keyFunc is used to make the key used for queued item // insertion and retrieval, and should be deterministic. keyFunc KeyFunc // knownObjects list keys that are "known", for the // purpose of figuring out which items have been deleted // when Replace() or Delete() is called. knownObjects KeyListerGetter // Indication the queue is closed. // Used to indicate a queue is closed so a control loop can exit when a queue is empty. // Currently, not used to gate any of CRED operations. closed bool closedLock sync.Mutex type Deltas
再来看一下Deltas类型，是Delta的切片类型。
type Deltas []Delta type Delta
继续看到Delta类型，其包含两个属性：
（1）Type：代表的是Delta的类型，有Added、Updated、Deleted、Sync四个类型；
（2）Object：存储的资源对象，如pod等资源对象；
type Delta struct { Type DeltaType Object interface{} } // staging/src/k8s.io/client-go/tools/cache/delta_fifo.go type DeltaType string // Change type definition const ( Added DeltaType = "Added" Updated DeltaType = "Updated" Deleted DeltaType = "Deleted" // The other types are obvious. You'll get Sync deltas when: // * A watch expires/errors out and a new list/watch cycle is started. // * You've turned on periodic syncs. // (Anything that trigger's DeltaFIFO's Replace() method.) Sync DeltaType = "Sync" ) 2.2 DeltaFIFO初始化-NewDeltaFIFO
NewDeltaFIFO初始化了一个items和queue都为空的DeltaFIFO并返回。
// staging/src/k8s.io/client-go/tools/cache/delta_fifo.go func NewDeltaFIFO(keyFunc KeyFunc, knownObjects KeyListerGetter) *DeltaFIFO { f := &DeltaFIFO{ items: map[string]Deltas{}, queue: []string{}, keyFunc: keyFunc, knownObjects: knownObjects, } f.cond.L = &f.lock return f } 3.DeltaFIFO核心处理方法分析
在前面分析Reflector时，Reflector的核心处理方法里有调用过几个方法，分别是r.store.Replace、r.store.Add、r.store.Update、r.store.Delete，结合前面文章的k8s informer的初始化与启动分析，或者简要的看一下下面的代码调用，就可以知道Reflector里的r.store其实就是DeltaFIFO，而那几个方法其实就是DeltaFIFO的Replace、Add、Update、Delete方法。
sharedIndexInformer.Run方法中调用NewDeltaFIFO初始化了DeltaFIFO，随后将DeltaFIFO作为参数传入初始化Config；
func (s *sharedIndexInformer) Run(stopCh <-chan struct{}) { ... fifo := NewDeltaFIFO(MetaNamespaceKeyFunc, s.indexer) cfg := &Config{ Queue: fifo, ... } func() { ... s.controller = New(cfg) ... }() ... s.controller.Run(stopCh) 在controller的Run方法中，调用NewReflector初始化Reflector时，将之前的DeltaFIFO传入，赋值给Reflector的store属性，所以Reflector里的r.store其实就是DeltaFIFO，而调用的r.store.Replace、r.store.Add、r.store.Update、r.store.Delete方法其实就是DeltaFIFO的Replace、Add、Update、Delete方法。
func (c *controller) Run(stopCh <-chan struct{}) { ... r := NewReflector( c.config.ListerWatcher, c.config.ObjectType, c.config.Queue, c.config.FullResyncPeriod, ) ... } func NewReflector(lw ListerWatcher, expectedType interface{}, store Store, resyncPeriod time.Duration) *Reflector { return NewNamedReflector(naming.GetNameFromCallsite(internalPackages...), lw, expectedType, store, resyncPeriod) } func NewNamedReflector(name string, lw ListerWatcher, expectedType interface{}, store Store, resyncPeriod time.Duration) *Reflector { r := &Reflector{ ... store: store, ... } ... return r } 所以这里对DeltaFIFO核心处理方法进行分析，主要是分析DeltaFIFO的Replace、Add、Update、Delete方法。
3.1 DeltaFIFO.Add
DeltaFIFO的Add操作，主要逻辑：
（1）加锁；
（2）调用f.queueActionLocked，操作DeltaFIFO中的queue与Deltas，根据对象key构造Added类型的新Delta追加到相应的Deltas中；
（3）释放锁。
func (f *DeltaFIFO) Add(obj interface{}) error { f.lock.Lock() defer f.lock.Unlock() f.populated = true return f.queueActionLocked(Added, obj) } 可以看到基本上DeltaFIFO所有的操作都有加锁操作，所以都是并发安全的。
3.1.1 DeltaFIFO.queueActionLocked
queueActionLocked负责操作DeltaFIFO中的queue与Deltas，根据对象key构造新的Delta追加到对应的Deltas中，主要逻辑：
（1）计算出对象的key；
（2）构造新的Delta，将新的Delta追加到Deltas末尾；
（3）调用dedupDeltas将Delta去重（目前只将Deltas最末尾的两个delete类型的Delta去重）；
（4）判断对象的key是否在queue中，不在则添加入queue中；
（5）根据对象key更新items中的Deltas；
（6）通知所有的消费者解除阻塞；
func (f *DeltaFIFO) queueActionLocked(actionType DeltaType, obj interface{}) error { //（1）计算出对象的key id, err := f.KeyOf(obj) if err != nil { return KeyError{obj, err} } //（2）构造新的Delta，将新的Delta追加到Deltas末尾 newDeltas := append(f.items[id], Delta{actionType, obj}) //（3）调用dedupDeltas将Delta去重（目前只将Deltas最末尾的两个delete类型的Delta去重） newDeltas = dedupDeltas(newDeltas) if len(newDeltas) > 0 { //（4）判断对象的key是否在queue中，不在则添加入queue中 if _, exists := f.items[id]; !exists { f.queue = append(f.queue, id) } //（5）根据对象key更新items中的Deltas f.items[id] = newDeltas //（6）通知所有的消费者解除阻塞 f.cond.Broadcast() } else { // We need to remove this from our map (extra items in the queue are // ignored if they are not in the map). delete(f.items, id) } return nil } 3.2 DeltaFIFO.Update
DeltaFIFO的Update操作，主要逻辑：
（1）加锁；
（2）调用f.queueActionLocked，操作DeltaFIFO中的queue与Deltas，根据对象key构造Updated类型的新Delta追加到相应的Deltas中；
（3）释放锁。
func (f *DeltaFIFO) Update(obj interface{}) error { f.lock.Lock() defer f.lock.Unlock() f.populated = true return f.queueActionLocked(Updated, obj) } 3.3 DeltaFIFO.Delete
DeltaFIFO的Delete操作，主要逻辑：
（1）计算出对象的key；
（2）加锁；
（3）items中不存在对象key，则直接return，跳过处理；
（4）调用f.queueActionLocked，操作DeltaFIFO中的queue与Deltas，根据对象key构造Deleted类型的新Delta追加到相应的Deltas中；
（5）释放锁。
func (f *DeltaFIFO) Delete(obj interface{}) error { id, err := f.KeyOf(obj) if err != nil { return KeyError{obj, err} } f.lock.Lock() defer f.lock.Unlock() f.populated = true // informer的用法中，f.knownObjects不为nil if f.knownObjects == nil { if _, exists := f.items[id]; !exists { // Presumably, this was deleted when a relist happened. // Don't provide a second report of the same deletion. return nil } } else { // We only want to skip the "deletion" action if the object doesn't // exist in knownObjects and it doesn't have corresponding item in items. // Note that even if there is a "deletion" action in items, we can ignore it, // because it will be deduped automatically in "queueActionLocked" _, exists, err := f.knownObjects.GetByKey(id) _, itemsExist := f.items[id] if err == nil && !exists && !itemsExist { // Presumably, this was deleted when a relist happened. // Don't provide a second report of the same deletion. return nil } } return f.queueActionLocked(Deleted, obj) } 3.4 DeltaFIFO.Replace
DeltaFIFO的Replace操作，主要逻辑：
（1）加锁；
（2）遍历list，计算对象的key，循环调用f.queueActionLocked，操作DeltaFIFO中的queue与Deltas，根据对象key构造Sync类型的新Delta追加到相应的Deltas中；
（3）对比DeltaFIFO中的items与Replace方法的list，如果DeltaFIFO中的items有，但传进来Replace方法的list中没有某个key，则调用f.queueActionLocked，操作DeltaFIFO中的queue与Deltas，根据对象key构造Deleted类型的新Delta追加到相应的Deltas中（避免重复，使用DeletedFinalStateUnknown包装对象）；
（4）释放锁；
// staging/src/k8s.io/client-go/tools/cache/delta_fifo.go func (f *DeltaFIFO) Replace(list []interface{}, resourceVersion string) error { //（1）加锁 f.lock.Lock() //（4）释放锁 defer f.lock.Unlock() keys := make(sets.String, len(list)) //（2）遍历list，计算对象的key，循环调用f.queueActionLocked，操作DeltaFIFO中的queue与Deltas，根据对象key构造Sync类型的新Delta追加到相应的Deltas中 for _, item := range list { key, err := f.KeyOf(item) if err != nil { return KeyError{item, err} } keys.Insert(key) if err := f.queueActionLocked(Sync, item); err != nil { return fmt.Errorf("couldn't enqueue object: %v", err) } } // informer的用法中，f.knownObjects不为nil if f.knownObjects == nil { // Do deletion detection against our own list. queuedDeletions := 0 for k, oldItem := range f.items { if keys.Has(k) { continue } var deletedObj interface{} if n := oldItem.Newest(); n != nil { deletedObj = n.Object } queuedDeletions++ if err := f.queueActionLocked(Deleted, DeletedFinalStateUnknown{k, deletedObj}); err != nil { return err } } if !f.populated { f.populated = true // While there shouldn't be any queued deletions in the initial // population of the queue, it's better to be on the safe side. f.initialPopulationCount = len(list) + queuedDeletions } return nil } //（3）找出DeltaFIFO中的items有，但传进来Replace方法的list中没有的key，调用f.queueActionLocked，操作DeltaFIFO中的queue与Deltas，根据对象key构造Deleted类型的新Delta追加到相应的Deltas中（避免重复，使用DeletedFinalStateUnknown包装对象） // Detect deletions not already in the queue. knownKeys := f.knownObjects.ListKeys() queuedDeletions := 0 for _, k := range knownKeys { if keys.Has(k) { continue } deletedObj, exists, err := f.knownObjects.GetByKey(k) if err != nil { deletedObj = nil klog.Errorf("Unexpected error %v during lookup of key %v, placing DeleteFinalStateUnknown marker without object", err, k) } else if !exists { deletedObj = nil klog.Infof("Key %v does not exist in known objects store, placing DeleteFinalStateUnknown marker without object", k) } queuedDeletions++ if err := f.queueActionLocked(Deleted, DeletedFinalStateUnknown{k, deletedObj}); err != nil { return err } } // 第一次调用Replace方法后，populated值为true if !f.populated { f.populated = true // initialPopulationCount代表第一次调用Replace方法加入DeltaFIFO中的items数量 f.initialPopulationCount = len(list) + queuedDeletions } return nil } 3.5 DeltaFIFO.Pop
DeltaFIFO的Pop操作，queue为空时会阻塞，直至非空，主要逻辑：
（1）加锁；
（2）循环判断queue的长度是否为0，为0则阻塞住，调用f.cond.Wait()，等待通知（与queueActionLocked方法中的f.cond.Broadcast()相对应，即queue中有对象key则发起通知）；
（3）取出queue的队头对象key；
（4）更新queue，把queue中所有的对象key前移，相当于把第一个对象key给pop出去；
（5）initialPopulationCount变量减1，当减到0时则说明initialPopulationCount代表第一次调用Replace方法加入DeltaFIFO中的对象key已经被pop完成；
（6）根据对象key从items中获取Deltas；
（7）把Deltas从items中删除；
（8）调用PopProcessFunc处理获取到的Deltas；
（9）释放锁。
// staging/src/k8s.io/client-go/tools/cache/delta_fifo.go func (f *DeltaFIFO) Pop(process PopProcessFunc) (interface{}, error) { //（1）加锁 f.lock.Lock() //（9）释放锁 defer f.lock.Unlock() //（2）循环判断queue的长度是否为0，为0则阻塞住，调用f.cond.Wait()，等待通知（与queueActionLocked方法中的f.cond.Broadcast()相对应，即queue中有对象key则发起通知） for { for len(f.queue) == 0 { // When the queue is empty, invocation of Pop() is blocked until new item is enqueued. // When Close() is called, the f.closed is set and the condition is broadcasted. // Which causes this loop to continue and return from the Pop(). if f.IsClosed() { return nil, ErrFIFOClosed } f.cond.Wait() } //（3）取出queue的队头对象key id := f.queue[0] //（4）更新queue，把queue中所有的对象key前移，相当于把第一个对象key给pop出去 f.queue = f.queue[1:] //（5）initialPopulationCount变量减1，当减到0时则说明initialPopulationCount代表第一次调用Replace方法加入DeltaFIFO中的对象key已经被pop完成 if f.initialPopulationCount > 0 { f.initialPopulationCount-- } //（6）根据对象key从items中获取对象 item, ok := f.items[id] if !ok { // Item may have been deleted subsequently. continue } //（7）把对象从items中删除 delete(f.items, id) //（8）调用PopProcessFunc处理pop出来的对象 err := process(item) if e, ok := err.(ErrRequeue); ok { f.addIfNotPresent(id, item) err = e.Err } // Don't need to copyDeltas here, because we're transferring // ownership to the caller. return item, err } } 3.6 DeltaFIFO.HasSynced
HasSynced从字面意思上看代表是否同步完成，是否同步完成其实是指第一次从kube-apiserver中获取到的全量的对象是否全部从DeltaFIFO中pop完成，全部pop完成，说明list回来的对象已经全部同步到了Indexer缓存中去了。
方法是否返回true是根据populated和initialPopulationCount两个变量来判断的，当且仅当populated为true且initialPopulationCount 为0的时候方法返回true，否则返回false。
populated属性值在第一次调用DeltaFIFO的Replace方法中就已经将其值设置为true。
而initialPopulationCount的值在第一次调用DeltaFIFO的Replace方法中设置值为加入到items中的Deltas的数量，然后每pop一个Deltas，则initialPopulationCount的值减1，pop完成时值则为0。
// staging/src/k8s.io/client-go/tools/cache/delta_fifo.go func (f *DeltaFIFO) HasSynced() bool { f.lock.Lock() defer f.lock.Unlock() return f.populated && f.initialPopulationCount == 0 } 在前面做informer的初始化与启动分析时也提到过，DeltaFIFO.HasSynced方法的调用链如下：
sharedIndexInformer.WaitForCacheSync --> cache.WaitForCacheSync --> sharedIndexInformer.controller.HasSynced --> controller.config.Queue.HasSynced --> DeltaFIFO.HasSynced
至此DeltaFIFO的分析就结束了，最后来总结一下。
总结
DeltaFIFO核心处理方法
Reflector调用的r.store.Replace、r.store.Add、r.store.Update、r.store.Delete方法其实就是DeltaFIFO的Replace、Add、Update、Delete方法。
（1）DeltaFIFO.Replace：构造Sync类型的Delta加入DeltaFIFO中，此外还会对比DeltaFIFO中的items与Replace方法的list，如果DeltaFIFO中的items有，但传进来Replace方法的list中没有某个key，则构造Deleted类型的Delta加入DeltaFIFO中；
（2）DeltaFIFO.Add：构建Added类型的Delta加入DeltaFIFO中；
（3）DeltaFIFO.Update：构建Updated类型的Delta加入DeltaFIFO中；
（4）DeltaFIFO.Delete：构建Deleted类型的Delta加入DeltaFIFO中；
（5）DeltaFIFO.Pop：从DeltaFIFO的queue中pop出队头key，从map中取出key对应的Deltas返回，并把该key:Deltas从map中移除；
（6）DeltaFIFO.HasSynced：返回true代表同步完成，是否同步完成指第一次从kube-apiserver中获取到的全量的对象是否全部从DeltaFIFO中pop完成，全部pop完成，说明list回来的对象已经全部同步到了Indexer缓存中去了；
informer架构中的DeltaFIFO
在对informer中的DeltaFIFO分析完之后，接下来将分析informer中的Controller与Processor。
