DeltaFIFO¶
在介绍 Informer 之前,我们还得先介绍一个 DeltaFIFO
的概念,从下图可以看出来 DeltaFIFO
也包含在 Informer 中。
client-go 官方架构图:
DeltaFIFO 是一个生产者-消费者的队列,生产者是 Reflector,消费者是 Pop 函数,从架构图上可以看出 DeltaFIFO 的数据来源为 Reflector,通过 Pop 操作消费数据,消费的数据一方面存储到 Indexer 中,另一方面可以通过 Informer 的 handler 进行处理,Informer 的 handler 处理的数据需要与存储在 Indexer 中的数据匹配。需要注意的是,Pop 的单位是一个 Deltas,而不是 Delta。
DetlaFIFO 同时实现了 Queue 和 Store 接口,使用 Deltas 保存了对象状态的变更(Add/Delete/Update)信息(如 Pod 的删除添加邓),Deltas 缓存了针对相同对象的多个状态变更信息,如 Pod 的 Deltas[0]可能更新了标签,Deltas[1]可能删除了该 Pod。最老的状态变更信息为 Oldest(),最新的状态变更信息为 Newest(),使用中,获取 DeltaFIFO 中对象的 key 以及获取 DeltaFIFO 都以最新状态为准。
接口¶
Delta 我们可以简单理解为变化,在 client-go 中关于 Delta 的定义如下所示:
// client-go/tools/cache/delta_fifo.go
// DeltaType 是变化类型 (添加、删除等)
type DeltaType string
const (
Added DeltaType = "Added" // 增加
Updated DeltaType = "Updated" // 更新
Deleted DeltaType = "Deleted" // 删除
Sync DeltaType = "Sync" // 同步
)
// Delta 是 DeltaFIFO 存储的类型,他告诉你发生了什么变化以及变化后对象的状态。
type Delta struct {
Type DeltaType // Delta 类型
Object interface{} // 对象,Delta的粒度是一个对象
}
Delta 其实就是 Kubernetes 中对象的变化(增、删、改、同步),FIFO 是一个先入先出的队列,那么 DeltaFIFO 就是一个按序的(先入先出)Kubernetes 对象变化的队列。除了 Delta 之外还有几个比较重要的类型如下所示:
// client-go/tools/cache/delta_fifo.go
// KeyLister 和 KeyGetter 的组合
type KeyListerGetter interface {
KeyLister
KeyGetter
}
// KeyLister 是一个通用的接口类型,只定义了一个接口函数
type KeyLister interface {
ListKeys() []string // 返回所有的keys
}
type KeyGetter interface {
GetByKey(key string) (interface{}, bool, error) // 通过 key 获取对象
}
其实我仔细观察会发现前面 Indexer 中提到的 cache(client-go/tools/cache/store
)就实现了上面三个接口,上面三个接口基本上就是 kv 的标准接口,但凡是通过 kv 方式访问的对象(存储、队列、索引等)多半具备以上接口。肯定有人会问直接使用具体的类型不就完了么,定义这些有什么用?答案很简单,当你需要对 kv 的对象只读但是不关心具体实现时就用上了~
接下来再来认识一个类型:
// client-go/tools/cache/fifo.go
// Queue 才是 FIFO 的抽象,DeltaFIFO 是一种具体的实现,和 Store 相同,但也有一个Pop()方法。
type Queue interface {
Store // 实现了存储接口,FIFO也是一种存储
Pop(PopProcessFunc) (interface{}, error) // 弹出对象
AddIfNotPresent(interface{}) error // 对象如果不在队列中就添加
HasSynced() bool // 通过Replace()放入第一批对象到队列中并且已经被Pop()全部取走
Close() // 关闭队列
}
Queue
是在 Store 基础上扩展了 Pop 方法可以让对象有序的弹出,Indexer 是在 Store 基础上建立了索引,可以快速检索对象。
实现¶
上面是 Queue 的接口抽象定义,接下来我们来看下具体的实现 DeltaFIFO,对应的结构体定义如下所示:
// client-go/tools/cache/delta_fifo.go
type DeltaFIFO struct {
lock sync.RWMutex // 读写锁
cond sync.Cond // 给Pop()函数使用,在没有对象的时候可以阻塞,内部锁复用读写锁
items map[string]Deltas // 按照kv的方式存储对象,但是存储的是对象的Deltas数组
queue []string // 为先入先出实现的,存储的是对象的键
populated bool // 通过 Replace() 接口将第一批对象放入队列,或者第一次调用增、删、改接口时标记为true
initialPopulationCount int // 通过 Replace() 接口将第一批对象放入队列的对象数量
keyFunc KeyFunc // 对象键计算函数
knownObjects KeyListerGetter // 其实就是 Indexer
closed bool // 是否已经关闭的标记
closedLock sync.Mutex
}
其中比较难理解的是 knownObjects,它的类型是 KeyListerGetter,这个接口中的 ListKeys 和 GetByKey 方法也是 Store 接口中的方法,所以 knownObjects 能够被赋值为实现了 Store 的类型指针;同样地,由于 Indexer 继承了 Store 方法,所以 knonwObjects 能够被赋值为实现了 Indexer 的类型指针。knownObjects 实际使用时是 Indexer,对应图中得 localStore。DeltaFIFO 根据其保存得对象状态变更消息处理(增/删/改/同步)knownObjects 中相应得对象。同步(Sync)Deltas 中即将被删除的对象是没有一样的。
DeltaFIFO.knowObjects.GetByKey 就是执行的 store.go 中的 GetByKey 函数,用于获取 Indexer 中的对象键。initialPopulationCount
用于表示是否完成全量同步,在 Replace 函数中增加,在 Pop 函数中减小,当 initialPopulationCount 为 0 且 populated 为 true 时表示 Pop 了所有 Replace 添加到 DeltaFIFO 中的对象,populated 用于判断 DletaFIFO 是否为初始化状态(即没有处理过任何对象)。
DeltaFIFO 的计算对象键的函数略有不同:
// client-go/tools/cache/delta_fifo.go
// KeyOf 暴露 f 的 keyFunc,但是也要检查 Deltas 对象的 key
func (f *DeltaFIFO) KeyOf(obj interface{}) (string, error) {
// 先用 Deltas 做一次强制转换
if d, ok := obj.(Deltas); ok {
if len(d) == 0 {
return "", KeyError{obj, ErrZeroLengthDeltasObject}
}
// 只用最新版本的对象
obj = d.Newest().Object
}
if d, ok := obj.(DeletedFinalStateUnknown); ok {
return d.Key, nil
}
return f.keyFunc(obj)
}
DeltaFIFO 的计算对象键的方式为什么要先做一次 Deltas 的类型转换呢?原因很简单,那就是从 DeltaFIFO.Pop() 出去的对象很可能还要再添加进来(比如处理失败需要再放进来),此时添加的对象就是已经封装好的 Deltas。
既然 DeltaFIFO 是 Store 的一种实现,接下来我们来看下 DeltaFIFO 相应的实现:
// client-go/tools/cache/delta_fifo.go
// Add 添加元素到队列中
func (f *DeltaFIFO) Add(obj interface{}) error {
f.lock.Lock()
defer f.lock.Unlock()
f.populated = true // 队列第一次写入操作都要设置标记
return f.queueActionLocked(Added, obj)
}
// Update 和 Add 一样,只是更新 Delta 对象
func (f *DeltaFIFO) Update(obj interface{}) error {
f.lock.Lock()
defer f.lock.Unlock()
f.populated = true // 队列第一次写入操作都要设置标记
return f.queueActionLocked(Updated, obj)
}
// Delete 删除对象接口
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 {
// 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)
}
// 列举对象键接口
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
}
// 获取对象接口,比如说用 Service 对象获取 Pod 对象,因为他们的对象键是相同的
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
}
我们可以看到上面的实现大部分很简单,好几个方法的最终实现都是通过调用 queueActionLocked
函数实现的,如下所示:
// client-go/tools/cache/delta_fifo.go
// 从函数名称来看把“动作”放入队列中,这个动作就是 DeltaType,而且已经加锁了
func (f *DeltaFIFO) queueActionLocked(actionType DeltaType, obj interface{}) error {
// 计算对象键的函数
id, err := f.KeyOf(obj)
if err != nil {
return KeyError{obj, err}
}
// 如果是同步,并且对象未来会被删除,那么就直接返回,没必要记录这个动作了
// 肯定有人会问为什么Add/Delete/Update这些动作可以,因为同步对于已经删除的对象是没有意义的
// 已经删除的对象后续有可能添加、更新,因为同名的对象又被添加了,删除也是有可能
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]
if len(newDeltas) > 0 {
// 如果对象没有存在过,那就放入队列中,如果存在说明已经在 queue 中了,也就没必要再添加了
if !exists {
f.queue = append(f.queue, id)
}
// 更新 Deltas 数组,通知所有调用 Pop() 的人
f.items[id] = newDeltas
f.cond.Broadcast()
} else if exists {
// 直接把对象删除,这段代码我不知道什么条件会进来,因为dedupDeltas()肯定有返回结果的
// 后面会有dedupDeltas()详细说明
delete(f.items, id)
}
return nil
}
首先我们想想为什么每个对象一个 Deltas 而不是 Delta?对一个对象的多个操作,什么操作可以合并?
DeltaFIFO 生产者和消费者是异步的,如果同一个目标的频繁操作,前面操作还缓存在队列中的时候,那么队列就要缓冲对象的所有操作,那可以将多个操作合并么?
对于更新这种类型的操作在没有全量基础的情况下是没法合并的,同时我们还不知道具体是什么类型的对象,所以能合并的也就是有添加/删除,两个添加/删除操作其实可以视为一个。
合并的操作如下所示:
// client-go/tools/cache/delta_fifo.go
// 合并 Deltas
func dedupDeltas(deltas Deltas) Deltas {
// 小于2个delta,就没必要合并了
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
}
// 理论上返回最后一个比较好,但是对象已经不再系统监控范围,前一个删除状态是好的
if _, ok := b.Object.(DeletedFinalStateUnknown); ok {
return a
}
return b
}
因为系统对于删除的对象有 DeletedFinalStateUnknown
这个状态,所以会存在两次删除的情况,但是两次添加同一个对象由于 apiserver 可以保证对象的唯一性,所以处理中就没有考虑合并两次添加操作。
接下来我们来看看 Replace() 函数的实现,这个也是 Store 定义的接口:
// 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))
// 遍历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)
}
}
// 如果没有Indexer存储的话,自己存储的就是所有的老对象,目的是要看哪些老对象不在全量集合中,那么就是删除的对象了
if f.knownObjects == nil {
// 遍历所有的元素
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++
// 队列中存储对象的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) + queuedDeletions
}
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
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++
// 把对象删除的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
}
从 Replace() 的实现来看,主要用于实现对象的全量更新。这个可以理解为 DeltaFIFO 在必要的时候做一次全量更新,这个时刻可以是定期的,也可以是事件触发的。由于 DeltaFIFO 对外输出的就是所有目标的增量变化,所以每次全量更新都要判断对象是否已经删除,因为在全量更新前可能没有收到目标删除的请求。这一点与 cache 不同,cache 的 Replace() 相当于重建,因为cache 就是对象全量的一种内存映射,所以 Replace() 就等于重建。
那我来问题一个非常有水平的问题,为什么 knownObjects 为 nil 时需要对比队列和对象全量来判断对象是否删除,而knownObjects 不为空的时候就不需要了?如果读者想判断自己是否已经全部理解可以不看下面自己想想。
我们前面说过,knownObjects 就是 Indexer(具体实现是 cache),而开篇的那副图已经非常明确的描述了二者以及使用之间的关系。也就是说 knownObjects 有的对象就是使用者知道的所有对象,此时即便队列(DeltaFIFO)中有相应的对象,在更新的全量对象中又被删除了,那就没必要通知使用者对象删除了,这种情况可以假想为系统短时间添加并删除了对象,对使用者来说等同于没有这个对象。
=======
ListWatch 的 list 步骤会调用 Replace(client-go/tools/cache/delta_fifo.go)函数来对 DeltaFIFO 进行全量更新:
- Sync 所有 DeltaFIFO 中的对象,将输入对象全部加入到 DeltaFIFO
- 如果 knowObjects 为空,则删除 DeltaFIFO 中不存在于输入对象的对象,使 DeltaFIFO 中的有效对象(非 DeletedFinalStateUnknown)等同于输入对象
- 如果 knownObjects 非空,获取 knowObjects 中不存在于输入对象的对象,并在 DeltaFIFO 中删除这些对象
第2步好理解,knownObjects 为空,只需要更新 DeltaFIFO 即可。第3步中,当 knownObjects 非空时,需要以 knowObjects 为基准进行对象的删除,否则会造成 indexer 中的数据与 apiserver 的数据不一致,举个例子,比如 knownObjects 中的对象为 {obj1, obj2, obj3},而 DeltaFIFO中待处理的对象为 {obj2, obj3,obj4},如果仅按照2步骤进行处理,会导致 knownObjects 中残留 obj1,因此需要在 DeltaFIFO 中添加删除 obj1 的变更消息。从下面 ShareInformer 的图中可以看出,knownObjects(即Indexer)的数据只能通过 DeltaFIFO 变更。