informer中的DeltaFIFO机制的实现分析与源码解读

DeltaFIFO作为informer中重要组件，本文从源码层面了解是如何DelatFIFO是实现的。

DeltaFIFO的定义

找到delta_fifo.go的源码，位于client-go/tools/cache/delta_fifo.go

代码结构大致如下:

store定义了一个通用的存储接口

type Store interface {
	Add(obj interface{}) error
	Update(obj interface{}) error
	Delete(obj interface{}) error
	List() []interface{}
	ListKeys() []string
	Get(obj interface{}) (item interface{}, exists bool, err error)
	GetByKey(key string) (item interface{}, exists bool, err error)

	// Replace will delete the contents of the store, using instead the
	// given list. Store takes ownership of the list, you should not reference
	// it after calling this function.
	Replace([]interface{}, string) error
	Resync() error
}

Queue接口继承了store，但添加了Pop()重要方法，实现了队列的能力

// Queue is exactly like a Store, but has a Pop() method too.
type Queue interface {
	Store

	// Pop blocks until it has something to process.
	// It returns the object that was process and the result of processing.
	// The PopProcessFunc may return an ErrRequeue{...} to indicate the item
	// should be requeued before releasing the lock on the queue.
	Pop(PopProcessFunc) (interface{}, error)

	// AddIfNotPresent adds a value previously
	// returned by Pop back into the queue as long
	// as nothing else (presumably more recent)
	// has since been added.
	AddIfNotPresent(interface{}) error

	// HasSynced returns true if the first batch of items has been popped
	HasSynced() bool

	// Close queue
	Close()
}

FIFO类型实现了Queue接口

type FIFO struct {
	lock sync.RWMutex
	cond sync.Cond
	// We depend on the property that items in the set are in the queue and vice versa.
	items map[string]interface{}
	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

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

var (
	_ = Queue(&FIFO{}) // FIFO is a Queue
)

DeltaFIFO类型也实现了Queue接口，与FIFO主要的区别是有两种特殊的方式：replaced和sync。replaced一般发生在资源版本更新时，而sync由resync定时发起。

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
}

var (
	_ = Queue(&DeltaFIFO{}) // DeltaFIFO is a Queue
)

DeltaFIFO的构造函数

// NewDeltaFIFOWithOptions returns a Queue which can be used to process changes to
// items. See also the comment on DeltaFIFO.
func NewDeltaFIFO(keyFunc KeyFunc, knownObjects KeyListerGetter) *DeltaFIFO {
	return NewDeltaFIFOWithOptions(DeltaFIFOOptions{
		KeyFunction:  keyFunc,
		KnownObjects: knownObjects,
	})
}

func NewDeltaFIFOWithOptions(opts DeltaFIFOOptions) *DeltaFIFO {
	if opts.KeyFunction == nil {
		opts.KeyFunction = MetaNamespaceKeyFunc // 如果不指定keyFunc，默认就是MetaNamespaceKeyFunc
	}

	f := &DeltaFIFO{
		items:        map[string]Deltas{}, 	// 存放[]Delta的数组,
		queue:        []string{},						// 存储obj的key，key通常是ns/name格式的字符串
		keyFunc:      opts.KeyFunction,			// 由obj生成key的函数
		knownObjects: opts.KnownObjects,

		emitDeltaTypeReplaced: opts.EmitDeltaTypeReplaced,
		transformer:           opts.Transformer,
	}
	f.cond.L = &f.lock
	return f
}

var (
	_ = Queue(&DeltaFIFO{}) // DeltaFIFO is a Queue
)

Delta的定义

根据Delta数据结构的定义，delta包含了一个资源对象的变更类型及变更的内容。这里的Object不一定是完整的资源数据，大部分场景下只会有变更的部分信息。

// Delta is a member of Deltas (a list of Delta objects) which
// in its turn is the type stored by a DeltaFIFO. It tells you what
// change happened, and the object's state after* that change.
//
// [*] Unless the change is a deletion, and then you'll get the final
// state of the object before it was deleted.
type Delta struct {
	Type   DeltaType		// 表示对obj的操作类型"Added/Updated/Deleted/Replaced/Sync"
	Object interface{}	// 表示某个资源对象,比如命名为"one"的pod
}


// Deltas is a list of one or more 'Delta's to an individual object.
// The oldest delta is at index 0, the newest delta is the last one.
type Deltas []Delta


// DeltaType is the type of a change (addition, deletion, etc)
type DeltaType string


// Change type definition
const (
	Added   DeltaType = "Added"
	Updated DeltaType = "Updated"
	Deleted DeltaType = "Deleted"
	// Replaced is emitted when we encountered watch errors and had to do a
	// relist. We don't know if the replaced object has changed.
	//
	// NOTE: Previous versions of DeltaFIFO would use Sync for Replace events
	// as well. Hence, Replaced is only emitted when the option
	// EmitDeltaTypeReplaced is true.
	Replaced DeltaType = "Replaced"
	// Sync is for synthetic events during a periodic resync.
	Sync DeltaType = "Sync"
)

Deletas的ADD()入队分析

watch机制监控到事件后，会把事件入队操作。

// Add inserts an item, and puts it in the queue. The item is only enqueued
// if it doesn't already exist in the set.
func (f *DeltaFIFO) Add(obj interface{}) error {
	f.lock.Lock()
	defer f.lock.Unlock()
	f.populated = true
  return f.queueActionLocked(Added, obj)	// 实际调用的是函数queueActionLocked()
}

queueActionLocked的逻辑主要包括从obj生产key(代码中是id),再有actionType和Obj构建一个新的Delta, 再把Delta加入Deltas切片中，之后，把Deltas放入items哈希表，key放入Queue队列中去。要注意Delta加入Deltas时需要进行出重。

// queueActionLocked appends to the delta list for the object.
// Caller must lock first.
func (f *DeltaFIFO) queueActionLocked(actionType DeltaType, obj interface{}) error {
	id, err := f.KeyOf(obj)									// 先用keyFunc，通过obj获取到对应的key
	if err != nil {
		return KeyError{obj, err}
	}

  newDeltas := append(f.items[id], Delta{actionType, obj})	// 用actionType和Obj构建一个新的Delta,再把Delta追加到(f.items[id]返回的)Deltas切片
	newDeltas = dedupDeltas(newDeltas)												// 对新的Deltas切去去重

	if len(newDeltas) > 0 {																		// 如果newDeltas切片中存在Delta
		if _, exists := f.items[id]; !exists {									
			f.queue = append(f.queue, id)													// 将key放到queue中
		}
		f.items[id] = newDeltas																	// 把新的Deltes切片，放到items哈希表
		f.cond.Broadcast()
	}
	return nil
}

Delta去重

Delta进行Add()操作时，会对加入的delta进行去重。去重逻辑目前只针对两个delete类型的delta有效：当delta数组中倒数第一个和第二个delta都是delete类型时，将会去掉其中一个。

// re-listing and watching can deliver the same update multiple times in any
// order. This will combine the most recent two deltas if they are the same.
func dedupDeltas(deltas Deltas) Deltas {
	n := len(deltas)
	if n < 2 {
		return deltas
	}
	a := &deltas[n-1]
	b := &deltas[n-2]
	if out := isDup(a, b); out != nil {
		d := append(Deltas{}, deltas[:n-2]...)
		return append(d, *out)
	}
	return deltas
}

// If a & b represent the same event, returns the delta that ought to be kept.
// Otherwise, returns nil.
// TODO: is there anything other than deletions that need deduping?
func isDup(a, b *Delta) *Delta {
	if out := isDeletionDup(a, b); out != nil {
		return out
	}
	// TODO: Detect other duplicate situations? Are there any?
	return nil
}

// keep the one with the most information if both are deletions.
func isDeletionDup(a, b *Delta) *Delta {
	if b.Type != Deleted || a.Type != Deleted {					// 仅处理a,b都是"Deleted"类型的事件;
		return nil
	}
	// Do more sophisticated checks, or is this sufficient?
	if _, ok := b.Object.(DeletedFinalStateUnknown); ok { // 如果a,b都是"Deleted"，就只返回一个Delta
		return a
	}
	return b
}

Deltas的pop出队

deltaFIFO出队的操作和普通的队列出队类似，从队头取出一个资源对象key，并删除items中key对应的deltas数组。

pop出队时，会调用传参PopProcessFunc对出队元素进行处理。

// Pop blocks until an item is added to the queue, and then returns it.  If
// multiple items are ready, they are returned in the order in which they were
// added/updated. The item is removed from the queue (and the store) before it
// is returned, so if you don't successfully process it, you need to add it back
// with AddIfNotPresent().
// process function is called under lock, so it is safe update data structures
// in it that need to be in sync with the queue (e.g. knownKeys). The PopProcessFunc
// may return an instance of ErrRequeue with a nested error to indicate the current
// item should be requeued (equivalent to calling AddIfNotPresent under the lock).
//
// Pop returns a 'Deltas', which has a complete list of all the things
// that happened to the object (deltas) while it was sitting in the queue.
func (f *DeltaFIFO) Pop(process PopProcessFunc) (interface{}, error) {
	f.lock.Lock()
	defer f.lock.Unlock()
	for {
    // 如果队列为空，就f.cond.Wait阻塞等待
		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()					
		}
    // 从f.queue中去重第一个元素
		id := f.queue[0]
		f.queue = f.queue[1:]
    
		if f.initialPopulationCount > 0 {
			f.initialPopulationCount--
		}
    // 从items哈希表中根据id，取出Deltas
		item, ok := f.items[id]
    // 如果itmes哈希表中差不到id对应的Deltas，就结束进入下次循环
		if !ok {
			// Item may have been deleted subsequently.
			continue
		}
    // 从items哈希表中删除id对应的Deltas
		delete(f.items, id)
    
    // process()函数来处理从items哈希表中取出的Deltas
		err := process(item)
    
    // 如果出现错误，就把id加回queue,同时把Deltas加回items
		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
	}

接着我们看看process()函数具体是什么。

如果对informer启动比较熟悉的话，可以知道在创建informer时，newInformer()函数需要指定ProcessFunc。这个处理函数包括数据同步到存储，以及调用注册的用户函数两个操作。

func newInformer(
	lw ListerWatcher,
	objType runtime.Object,
	resyncPeriod time.Duration,
	h ResourceEventHandler,
	clientState Store,
) Controller {
	// This will hold incoming changes. Note how we pass clientState in as a
	// KeyLister, that way resync operations will result in the correct set
	// of update/delete deltas.
	fifo := NewDeltaFIFOWithOptions(DeltaFIFOOptions{
		KnownObjects:          clientState,
		EmitDeltaTypeReplaced: true,
	})

	cfg := &Config{
		Queue:            fifo,
		ListerWatcher:    lw,
		ObjectType:       objType,
		FullResyncPeriod: resyncPeriod,
		RetryOnError:     false,
		// 指定处理从deltaFIFO队列pop处理的数据的处理函数ProcessFunc
		Process: func(obj interface{}) error {
			// from oldest to newest
			for _, d := range obj.(Deltas) {
				switch d.Type {
				case Sync, Replaced, Added, Updated:
				   // 同步存储数据，clientState是一个store
					if old, exists, err := clientState.Get(d.Object); err == nil && exists {
						if err := clientState.Update(d.Object); err != nil {
							return err
						}
						// 回调用户定义的hander函数
						h.OnUpdate(old, d.Object)
					} else {
					   // 同步存储数据
						if err := clientState.Add(d.Object); err != nil {
							return err
						}
						// 回调用户定义的hander函数
						h.OnAdd(d.Object)
					}
				case Deleted:
				   // 同步存储数据
					if err := clientState.Delete(d.Object); err != nil {
						return err
					}
					// 回调用户定义的hander函数
					h.OnDelete(d.Object)
				}
			}
			return nil
		},
	}
	return New(cfg)
}

进一步探究一下，informer启动run()后，会调用controller.Run()，最后c.processLoop会循环处理pop出队处理，流程大致如下：

informer.run(stopCh) —> s.controller.Run(stopCh)—>c.processLoop—>c.config.Queue.Pop(PopProcessFunc(c.config.Process))

源码位于：vender/k8s.io/client-go/tools/cache/controller.go

func (c *controller) Run(stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
	go func() {
		<-stopCh
		c.config.Queue.Close()
	}()
	// 构建一个Reflector
	r := NewReflector(
		c.config.ListerWatcher,
		c.config.ObjectType,
		c.config.Queue,
		c.config.FullResyncPeriod,
	)
	r.ShouldResync = c.config.ShouldResync
	r.clock = c.clock

	c.reflectorMutex.Lock()
	c.reflector = r
	c.reflectorMutex.Unlock()

	var wg wait.Group

	wg.StartWithChannel(stopCh, r.Run)

  // 调用c.processLoop函数
	wait.Until(c.processLoop, time.Second, stopCh)
	wg.Wait()
}

func (c *controller) processLoop() {
	for {
	  // 循环的Pop出队，把出队的事件交给PopProcessFunc函数处理
		obj, err := c.config.Queue.Pop(PopProcessFunc(c.config.Process))
		if err != nil {
			if err == ErrFIFOClosed {
				return
			}
			if c.config.RetryOnError {
				// This is the safe way to re-enqueue.
				c.config.Queue.AddIfNotPresent(obj)
			}
		}
	}
}

DeltaFIFO出队入队的流程图