做网站的电脑,六盘水遵义网站建设怎么做,无锡网站开发,室内设计者联盟网站informer中的DeltaFIFO机制的实现分析与源码解读
DeltaFIFO作为informer中重要组件#xff0c;本文从源码层面了解是如何DelatFIFO是实现的。
DeltaFIFO的定义
找到delta_fifo.go的源码#xff0c;位于client-go/tools/cache/delta_fifo.go
代码结构大致如下:
store定义…informer中的DeltaFIFO机制的实现分析与源码解读
DeltaFIFO作为informer中重要组件本文从源码层面了解是如何DelatFIFO是实现的。
DeltaFIFO的定义
找到delta_fifo.go的源码位于client-go/tools/cache/delta_fifo.go
代码结构大致如下:
store定义了一个通用的存储接口
type Store interface {Add(obj interface{}) errorUpdate(obj interface{}) errorDelete(obj interface{}) errorList() []interface{}ListKeys() []stringGet(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) errorResync() 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 poppedHasSynced() bool// Close queueClose()
}
FIFO类型实现了Queue接口
type FIFO struct {lock sync.RWMutexcond 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 boolclosedLock 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.RWMutexcond 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]Deltasqueue []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 boolclosedLock 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的keykey通常是ns/name格式的字符串keyFunc: opts.KeyFunction, // 由obj生成key的函数knownObjects: opts.KnownObjects,emitDeltaTypeReplaced: opts.EmitDeltaTypeReplaced,transformer: opts.Transformer,}f.cond.L f.lockreturn 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 objects state after* that change.
//
// [*] Unless the change is a deletion, and then youll get the final
// state of the object before it was deleted.
type Delta struct {Type DeltaType // 表示对obj的操作类型Added/Updated/Deleted/Replaced/SyncObject interface{} // 表示某个资源对象,比如命名为one的pod
}// Deltas is a list of one or more Deltas 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 AddedUpdated DeltaType UpdatedDeleted DeltaType Deleted// Replaced is emitted when we encountered watch errors and had to do a// relist. We dont 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 doesnt already exist in the set.
func (f *DeltaFIFO) Add(obj interface{}) error {f.lock.Lock()defer f.lock.Unlock()f.populated truereturn 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获取到对应的keyif 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切片中存在Deltaif _, 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就只返回一个Deltareturn 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 dont 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取出Deltasitem, ok : f.items[id]// 如果itmes哈希表中差不到id对应的Deltas就结束进入下次循环if !ok {// Item may have been deleted subsequently.continue}// 从items哈希表中删除id对应的Deltasdelete(f.items, id)// process()函数来处理从items哈希表中取出的Deltaserr : process(item)// 如果出现错误就把id加回queue,同时把Deltas加回itemsif e, ok : err.(ErrRequeue); ok {f.addIfNotPresent(id, item)err e.Err}// Dont need to copyDeltas here, because were 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处理的数据的处理函数ProcessFuncProcess: func(obj interface{}) error {// from oldest to newestfor _, d : range obj.(Deltas) {switch d.Type {case Sync, Replaced, Added, Updated:// 同步存储数据clientState是一个storeif 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() {-stopChc.config.Queue.Close()}()// 构建一个Reflectorr : NewReflector(c.config.ListerWatcher,c.config.ObjectType,c.config.Queue,c.config.FullResyncPeriod,)r.ShouldResync c.config.ShouldResyncr.clock c.clockc.reflectorMutex.Lock()c.reflector rc.reflectorMutex.Unlock()var wg wait.Groupwg.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出队入队的流程图
