k8s client-go 源码分析 informer 源码分析 (4)-DeltaFIFO 源码分析

type Delta struct {    Type   DeltaType    Object interface{}}

type DeltaFIFO struct {    ...    items map[string]Deltas  queue []string    ...}
type Deltas []Delta

// staging/src/k8s.io/client-go/tools/cache/delta_fifo.gotype 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 []Delta

type Delta struct {  Type   DeltaType  Object interface{}}

// staging/src/k8s.io/client-go/tools/cache/delta_fifo.gotype DeltaType string
// Change type definitionconst (  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")

// staging/src/k8s.io/client-go/tools/cache/delta_fifo.gofunc 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}

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)

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}

func (f *DeltaFIFO) Add(obj interface{}) error {  f.lock.Lock()  defer f.lock.Unlock()  f.populated = true  return f.queueActionLocked(Added, obj)}

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}

func (f *DeltaFIFO) Update(obj interface{}) error {  f.lock.Lock()  defer f.lock.Unlock()  f.populated = true  return f.queueActionLocked(Updated, obj)}

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

// staging/src/k8s.io/client-go/tools/cache/delta_fifo.gofunc (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}

// staging/src/k8s.io/client-go/tools/cache/delta_fifo.gofunc (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  }}

// staging/src/k8s.io/client-go/tools/cache/delta_fifo.gofunc (f *DeltaFIFO) HasSynced() bool {  f.lock.Lock()  defer f.lock.Unlock()  return f.populated && f.initialPopulationCount == 0}