Informer DeltaFIFO源码解析

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53

// 代码源自client-go/tools/cache/delta_fifo.go type DeltaFIFO struct { // lock/cond protects access to 'items' and 'queue'. // 读写锁，性能要比Mutex好 lock sync.RWMutex // 供pop函数使用，没有对象的时候可以阻塞 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. // 注意map值类型为Deltas，DeltaFIFO生产者和消费者是异步的，如果同一个目标的频繁操作，前面操作还缓存在队列中的时候，那么队列就要缓冲对象的所有操作 // queue里存的是map的keys，用map实现高速查询，用slice保证有序 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. // 这里其实就是Indexer 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 } // 代码源自client-go/tools/cache/delta_fifo.go type Delta struct { Type DeltaType Object interface{} // 对象，也就是k8s里的资源 } type DeltaType string // Delta的类型用字符串表达 const ( Added DeltaType = "Added" // 增加 Updated DeltaType = "Updated" // 更新 Deleted DeltaType = "Deleted" // 删除 Sync DeltaType = "Sync" // 同步 ) type Deltas []Delta // Delta数组

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25

// 代码源自client-go/tools/cache/fifo.go type Queue interface { // 存储接口，再Indexer分析过了 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在对列为空的时候会阻塞，就是通过上面提到的sync.Cond实现的 // 具体弹出的对象是通过传入的PopProcessFunc处理的，处理完返回错误后，可能会再入对列 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() }

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94

// 代码源自client-go/tools/cache/delta_fifo.go // 添加对象接口 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) 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 // 队列第一次写入操作都要设置标记 // 此处是需要注意的，knownObjects就是Indexer，里面存有已知全部的对象 if f.knownObjects == nil { // 在没有Indexer的条件下只能通过自己存储的对象查一下 if _, exists := f.items[id]; !exists { return nil } } else { // 自己和Indexer里面有任何一个有这个对象多算存在 _, exists, err := f.knownObjects.GetByKey(id) _, itemsExist := f.items[id] if err == nil && !exists && !itemsExist { return nil } } return f.queueActionLocked(Deleted, obj) } // 列举对象键接口 func (f *DeltaFIFO) ListKeys() []string { f.lock.RLock() defer f.lock.RUnlock() list := make([]string, 0, len(f.items)) for key := range f.items { list = append(list, key) } return list } // 列举对象接口 func (f *DeltaFIFO) List() []interface{} { f.lock.RLock() defer f.lock.RUnlock() return f.listLocked() } // 列举对象的具体实现 func (f *DeltaFIFO) listLocked() []interface{} { list := make([]interface{}, 0, len(f.items)) for _, item := range f.items { item = copyDeltas(item) list = append(list, item.Newest().Object) } return list } // 获取对象接口，这个有点意思，用对象获取对象 func (f *DeltaFIFO) Get(obj interface{}) (item interface{}, exists bool, err error) { key, err := f.KeyOf(obj) if err != nil { return nil, false, KeyError{obj, err} } return f.GetByKey(key) } // 通过对象键获取对象 func (f *DeltaFIFO) GetByKey(key string) (item interface{}, exists bool, err error) { f.lock.RLock() defer f.lock.RUnlock() d, exists := f.items[key] if exists { d = copyDeltas(d) } return d, exists, nil } // 判断是否关闭 func (f *DeltaFIFO) IsClosed() bool { f.closedLock.Lock() defer f.closedLock.Unlock() if f.closed { return true } return false }

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34

// 代码源自client-go/tools/cache/delta_fifo.go func (f *DeltaFIFO) queueActionLocked(actionType DeltaType, obj interface{}) error { // 前面提到的计算对象键的函数 id, err := f.KeyOf(obj) if err != nil { return KeyError{obj, err} } // 如果是同步，并且对象未来会被删除，那么就直接返回，没必要记录这个动作了 if actionType == Sync && f.willObjectBeDeletedLocked(id) { return nil } // 同一个对象的多次操作，所以要追加到Deltas数组中 newDeltas := append(f.items[id], Delta{actionType, obj}) // 合并操作，去掉冗余的delta newDeltas = dedupDeltas(newDeltas) // 判断对象是否已经存在 _, exists := f.items[id] // 合并后操作有可能变成没有Delta么？后面的代码分析来看应该不会，所以暂时不知道这个判断目的 if len(newDeltas) > 0 { // 如果对象没有存在过，那就放入队列中，如果存在说明已经在queue中了，也就没必要再添加了 if !exists { f.queue = append(f.queue, id) } // 更新Deltas数组，通知所有调用Pop()的人 f.items[id] = newDeltas // 还记得Pop在对列没数据时会阻塞吗，这里就会唤起GoRoutine继续执行 f.cond.Broadcast() } else if exists { // 直接把对象删除，这段代码我不知道什么条件会进来，因为dedupDeltas()肯定有返回结果的 // 后面会有dedupDeltas()详细说明 delete(f.items, id) } return nil }

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39

// 代码源自client-go/tools/cache/delta_fifo.go func dedupDeltas(deltas Deltas) Deltas { // 小于2个delta，那就是1个呗，没啥好合并的 n := len(deltas) if n < 2 { return deltas } // 取出最后两个 a := &deltas[n-1] b := &deltas[n-2] // 判断如果是重复的，那就删除这两个delta把合并后的追加到Deltas数组尾部 if out := isDup(a, b); out != nil { d := append(Deltas{}, deltas[:n-2]...) return append(d, *out) } return deltas } // 判断两个Delta是否是重复的 func isDup(a, b *Delta) *Delta { // 只有一个判断，只能判断是否为删除类操作，和我们上面的判断相同 // 这个函数的本意应该还可以判断多种类型的重复，当前来看只能有删除这一种能够合并 if out := isDeletionDup(a, b); out != nil { return out } return nil } // 判断是否为删除类的重复 func isDeletionDup(a, b *Delta) *Delta { // 二者都是删除那肯定有一个是重复的 if b.Type != Deleted || a.Type != Deleted { return nil } // 系统对于删除的对象有DeletedFinalStateUnknown这个状态，所以会存在两次删除的情况 if _, ok := b.Object.(DeletedFinalStateUnknown); ok { return a } return b }

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80

// 代码源自client-go/tools/cache/delta_fifo.go func (f *DeltaFIFO) Replace(list []interface{}, resourceVersion string) error { f.lock.Lock() defer f.lock.Unlock() keys := make(sets.String, len(list)) // 遍历所有的输入目标 for _, item := range list { // 计算目标键 key, err := f.KeyOf(item) if err != nil { return KeyError{item, err} } // 记录处理过的目标键，采用set存储，是为了后续快速查找 keys.Insert(key) // 因为输入是目标全量，所以每个目标相当于重新同步了一次 if err := f.queueActionLocked(Sync, item); err != nil { return fmt.Errorf("couldn't enqueue object: %v", err) } } // 如果没有存储的话，自己存储的就是所有的老对象，目的要看看那些老对象不在全量集合中，那么就是删除的对象了 if f.knownObjects == nil { // 遍历所有的元素 for k, oldItem := range f.items { // 这个目标在输入的对象中存在就可以忽略 if keys.Has(k) { continue } // 输入对象中没有，说明对象已经被删除了。 var deletedObj interface{} if n := oldItem.Newest(); n != nil { deletedObj = n.Object } // 终于看到哪里用到DeletedFinalStateUnknown了，队列中存储对象的Deltas数组中 // 可能已经存在Delete了，避免重复，采用DeletedFinalStateUnknown这种类型 if err := f.queueActionLocked(Deleted, DeletedFinalStateUnknown{k, deletedObj}); err != nil { return err } } // 如果populated还没有设置，说明是第一次并且还没有任何修改操作执行过 if !f.populated { f.populated = true f.initialPopulationCount = len(list) // 记录第一次通过来的对象数量 } return nil } // 下面处理的就是检测某些目标删除但是Delta没有在队列中 // 从存储中获取所有对象键 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 glog.Errorf("Unexpected error %v during lookup of key %v, placing DeleteFinalStateUnknown marker without object", err, k) } else if !exists { deletedObj = nil glog.Infof("Key %v does not exist in known objects store, placing DeleteFinalStateUnknown marker without object", k) } // 累积删除的对象数量 queuedDeletions++ // 把对象删除的Delta放入队列 if err := f.queueActionLocked(Deleted, DeletedFinalStateUnknown{k, deletedObj}); err != nil { return err } } // 和上面的代码差不多，只是计算initialPopulationCount值的时候增加了删除对象的数量 if !f.populated { f.populated = true f.initialPopulationCount = len(list) + queuedDeletions } return nil }

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41

// 代码源自client-go/tools/cache/delta_fifo.go func (f *DeltaFIFO) Pop(process PopProcessFunc) (interface{}, error) { f.lock.Lock() defer f.lock.Unlock() for { // 队列中有数据么？ for len(f.queue) == 0 { // 看来是先判断的是否有数据，后判断是否关闭，这个和chan像 if f.IsClosed() { return nil, FIFOClosedError } // 没数据那就等待把 f.cond.Wait() } // 取出第一个对象 id := f.queue[0] // 数组缩小，相当于把数组中的第一个元素弹出去了，这个不多解释哈 f.queue = f.queue[1:] // 取出对象，因为queue中存的是对象键 item, ok := f.items[id] // 同步对象计数减一，当减到0就说明外部已经全部同步完毕了 if f.initialPopulationCount > 0 { f.initialPopulationCount-- } // 对象不存在，这个是什么情况？貌似我们在合并对象的时候代码上有这个逻辑，估计永远不会执行 if !ok { continue } // 把对象删除 delete(f.items, id) // Pop()需要传入一个回调函数，用于处理对象 err := process(item) // 如果需要重新入队列，那就重新入队列 if e, ok := err.(ErrRequeue); ok { f.addIfNotPresent(id, item) err = e.Err } return item, err } }

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82

// 代码源自client-go/tools/cache/delta_fifo.go func (f *DeltaFIFO) HasSynced() bool { f.lock.Lock() defer f.lock.Unlock() // 这里就比较明白了，一次同步全量对象后，并且全部Pop()出去才能算是同步完成 // 其实这里所谓的同步就是全量内容已经进入Indexer，Indexer已经是系统中对象的全量快照了 return f.populated && f.initialPopulationCount == 0 } // 添加不存在的对象 func (f *DeltaFIFO) AddIfNotPresent(obj interface{}) error { // 这个要求放入的必须是Deltas数组，就是通过Pop（）弹出的对象 deltas, ok := obj.(Deltas) if !ok { return fmt.Errorf("object must be of type deltas, but got: %#v", obj) } // 多个Delta都是一个对象，所以用最新的就可以了 id, err := f.KeyOf(deltas.Newest().Object) if err != nil { return KeyError{obj, err} } f.lock.Lock() defer f.lock.Unlock() f.addIfNotPresent(id, deltas) return nil } // 这个是添加不存在对象的实现 func (f *DeltaFIFO) addIfNotPresent(id string, deltas Deltas) { f.populated = true // 这里判断的对象是否存在 if _, exists := f.items[id]; exists { return } // 放入队列中 f.queue = append(f.queue, id) f.items[id] = deltas f.cond.Broadcast() } // 重新同步，这个在cache实现是空的，这里面有具体实现，informer里讲到的重新同步也是从Indexer同步，而不是调用list接口同步 func (f *DeltaFIFO) Resync() error { f.lock.Lock() defer f.lock.Unlock() // 如果没有Indexer那么重新同步是没有意义的，因为连同步了哪些对象都不知道 if f.knownObjects == nil { return nil } // 列举Indexer里面所有的对象键 keys := f.knownObjects.ListKeys() // 遍历对象键，为每个对象产生一个同步的Delta for _, k := range keys { // 具体实现后面有介绍 if err := f.syncKeyLocked(k); err != nil { return err } } return nil } // 具体对象同步实现接口 func (f *DeltaFIFO) syncKeyLocked(key string) error { // 获取对象 obj, exists, err := f.knownObjects.GetByKey(key) if err != nil { glog.Errorf("Unexpected error %v during lookup of key %v, unable to queue object for sync", err, key) return nil } else if !exists { glog.Infof("Key %v does not exist in known objects store, unable to queue object for sync", key) return nil } // 计算对象的键值，有人会问对象键不是已经传入了么？那个是存在Indexer里面的对象键，可能与这里的计算方式不同 id, err := f.KeyOf(obj) if err != nil { return KeyError{obj, err} } // 对象已经在存在，说明后续会通知对象的新变化，所以再加更新也没意义 if len(f.items[id]) > 0 { return nil } // 添加对象同步的这个Delta if err := f.queueActionLocked(Sync, obj); err != nil { return fmt.Errorf("couldn't queue object: %v", err) } return nil }