Kubernetes 事件队列

￡神魔★判官ぃ 2022-04-24 16:38 240阅读 0赞

# Kubernetes 事件队列 #

*  Kubernetes 事件队列
    
     *  Queue 定义了队列接口
     *  FIFO 是先入先出的队列
        
         *  Add() 方法
         *  Update() 方法
         *  `Delete()` 方法
         *  Pop() 方法
         *  Replace() 方法
         *  HasSyncd() 方法
         *  Resync() 方法
     *  DeltaFIFO 是记录对象历史事件的队列
        
         *  DeltaFIFO 的生产者和消费者
         *  记录对象事件的 Delta、Deltas 和 DeletedFinalStateUnknown 类型
         *  Add() 和 Update() 方法
         *  queueActionLocked() 方法
         *  Delete() 方法
         *  Get/GetByKey/List/ListKeys() 方法
         *  Replace() 方法
         *  Resync() 方法
         *  Pop() 方法
         *  HasSyncd() 方法
         *  DeltaFIFO 和 knownObjects 对象缓存的同步

`Queue` 接口是在 `Store` 的基础上添加了 `Pop()` 方法。

`FIFO` 和 `DeltaFIFO` 类型(非接口)实现了 `Queue` 接口。

`DeltaFIFO` 是 Kubernetes 中非常重要的数据结构，用于保存对象的变化事件。

## Queue 定义了队列接口 ##

`Queue` 是在对象缓存的基础上，添加了 `Pop()` 方法，这样既能缓存对象、按照 Key 查找对象、也能按序（Add 的顺序）弹出对象：

// 来源于 k8s.io/client-go/tools/cache/fifo.go
    type Queue interface { 
        // 对象缓存接口
        Store
    
        // 弹出队列中的一个对象，如果队列为空，则一直阻塞。返回处理后的对象，以及处理结果。
        Pop(PopProcessFunc) (interface{ }, error)
    
        AddIfNotPresent(interface{ }) error
    
        // 当队列中第一批对象都弹出后，返回 true。
        HasSynced() bool
        Close()
    }

## FIFO 是先入先出的队列 ##

FIFO (struct 类型，非接口)实现了 Queue 接口，只缓存对象的**一个最新值**，例如，队列中对象 A 的值为 a1，在被弹出前，进行两次更新，值分别为 a2, a3，则只会弹出一次且值为 a3。

FIFO 适用的情况：

1.  你希望最多一个 worker 处理某个对象(对象在队列中是唯一的）；
2.  当 worker 处理该对象时，对象值是最新的；
3.  你不需要处理删除的对象(删除的对象不会被弹出)；

FIFO 类型定义如下：

// 来源于 k8s.io/client-go/tools/cache/fifo.go
    type FIFO struct { 
        lock sync.RWMutex
        cond sync.Cond
    
        // 对象缓存，用于快速查询。map key 为对象的 Key，该 Key 由 keyFunc 函数生成
        items map[string]interface{ }
    
        // 对象弹出(Pop)顺序队列，队列中各对象 Key 是**唯一**的
        queue []string
    
        // 首先调用 Delete/Add/Update 或 Replace() 添加的第一批对象都 Pop 后为 true
        populated bool
    
        // Replace() 添加的第一批对象的数目
        initialPopulationCount int
    
        // 根据对象生成它的标识 Key 的函数
        keyFunc KeyFunc
    
        closed     bool
        closedLock sync.Mutex
    }

函数 `NewFIFO()` 返回 FIFO 类型对象，传入的 `KeyFunc` (一般是 `DeletionHandlingMetaNamespaceKeyFunc`) 用于生成对象 Key：

// 来源于 k8s.io/client-go/tools/cache/fifo.go
    func NewFIFO(keyFunc KeyFunc) *FIFO { 
        f := &FIFO{ 
            // 初始化对象缓存
            items:   map[string]interface{ }{ },
            // 初始化对象 Key 队列
            queue:   []string{ },
            keyFunc: keyFunc,
        }
        f.cond.L = &f.lock
        return f
    }

### Add() 方法 ###

将对象更新到缓存(f.items)，如果缓存中没有该对象，则将它加到弹出队列（f.queue），这样可以保证只会弹出对象一次，且弹出的是最新值。

// 来源于 k8s.io/client-go/tools/cache/fifo.go
    func (f *FIFO) Add(obj interface{ }) error { 
        id, err := f.keyFunc(obj)
        if err != nil { 
            return KeyError{ obj, err}
        }
        f.lock.Lock()
        defer f.lock.Unlock()
        f.populated = true
    
        // 缓存中没有该对象，则将它的 key 加到队列 f.queue 中
        if _, exists := f.items[id]; !exists { 
            f.queue = append(f.queue, id)
        }
        // 更新对象缓存
        f.items[id] = obj
        f.cond.Broadcast()
        return nil
    }

什么情况下缓存中没有该对象呢？

1.  第一次向 FIFO Add/Update 该对象；
2.  或者调用 FIFO 的 Delete 方法删除该对象；
3.  或者，该对象被 Pop 处理了；
4.  或者，调用 Replace 方法，用新的一组对象替换当前缓存 f.items 和队列 f.queue；

### Update() 方法 ###

通过 `Add()` 方法实现：

func (f *FIFO) Update(obj interface{ }) error { 
        return f.Add(obj)
    }

### `Delete()` 方法 ###

从缓存中删除对象：

// 来源于 k8s.io/client-go/tools/cache/fifo.go
    func (f *FIFO) Delete(obj interface{ }) error { 
        id, err := f.keyFunc(obj)
        if err != nil { 
            return KeyError{ obj, err}
        }
        f.lock.Lock()
        defer f.lock.Unlock()
        f.populated = true
        // 从缓存中删除对象，注意 f.queue 中还可能有对象的 Key
        delete(f.items, id)
        return err
    }

注意：**没有**从弹出队列(f.queue) 中删除该对象 Key，后续弹出过程中会**忽略**这种已删除对象的 Key，继续弹出下一个对象。

### Pop() 方法 ###

从弹出队列（f.queue）弹出一个对象，并调用用户注册的回调函数进行处理，返回处理后的对象和出错信息。

如果队列为空，则一直阻塞。

处理函数执行失败时应该返回 `ErrRequeue` 类型的错误，这时该对象会被**重新加回** FIFO，后续可以再次被弹出处理。

`Pop()` 是在对 FIFO 加锁的情况下调用处理函数的，所以可以在多个 goroutine 中**并发调用** 该方法。

// 来源于 k8s.io/client-go/tools/cache/fifo.go
    func (f *FIFO) Pop(process PopProcessFunc) (interface{ }, error) { 
        f.lock.Lock()
        defer f.lock.Unlock()
        for { 
            for len(f.queue) == 0 { 
                if f.IsClosed() { 
                    return nil, FIFOClosedError
                }
                // 如果队列未关闭，但为空，则阻塞等待
                f.cond.Wait()
            }
            // 先从 queue 弹出对象 id
            id := f.queue[0]
            f.queue = f.queue[1:]
            if f.initialPopulationCount > 0 { 
                f.initialPopulationCount--
            }
            // 从缓存中获取对象
            item, ok := f.items[id]
            if !ok { 
                // 前面提到，当 Add/Update 对象，在 Pop 前又 Delete 了该对象，就会出现 queue 中有 id，而 items 中无对象的情况
                // 由于对象已经被删除，所以跳过，Pop 下一个对象
                continue
            }
            // 从缓存中删除对象
            delete(f.items, id)
            // 调用处理函数，该函数处于 f.lock 锁保护中，可以并发执行
            err := process(item)
            // 如果处理 item 失败，应该返回 ErrRequeue 类型错误，再将对象加回队列
            if e, ok := err.(ErrRequeue); ok { 
                f.addIfNotPresent(id, item)
                err = e.Err
            }
            return item, err
        }
    }

### Replace() 方法 ###

用传入的一组对象替换对象缓存 f.items 和弹出队列 f.queue：

// 来源于 k8s.io/client-go/tools/cache/fifo.go
    func (f *FIFO) Replace(list []interface{ }, resourceVersion string) error { 
        items := make(map[string]interface{ }, len(list))
        for _, item := range list { 
            key, err := f.keyFunc(item)
            if err != nil { 
                return KeyError{ item, err}
            }
            items[key] = item
        }
    
        f.lock.Lock()
        defer f.lock.Unlock()
    
        if !f.populated { 
            f.populated = true
            f.initialPopulationCount = len(items)
        }
    
        f.items = items
        f.queue = f.queue[:0]
        for id := range items { 
            f.queue = append(f.queue, id)
        }
        if len(f.queue) > 0 { 
            f.cond.Broadcast()
        }
        return nil
    }

### HasSyncd() 方法 ###

参考后文对 DeltaFIFO 的 `HasSyncd()` 方法分析。

### Resync() 方法 ###

将对象缓存 `f.items` 中的对象都更新到弹出队列 `f.queue` 中：

// 来源于 k8s.io/client-go/tools/cache/fifo.go
    func (f *FIFO) Resync() error { 
        f.lock.Lock()
        defer f.lock.Unlock()
    
        inQueue := sets.NewString()
        for _, id := range f.queue { 
            inQueue.Insert(id)
        }
        for id := range f.items { 
            if !inQueue.Has(id) { 
                f.queue = append(f.queue, id)
            }
        }
        if len(f.queue) > 0 { 
            f.cond.Broadcast()
        }
        return nil
    }

**FIXME!!!**：对象加入和弹出时都会同时更新 f.items 和 f.queue，按说是完全一致的，所以 `Resync()` 方法是多余的？

## DeltaFIFO 是记录对象历史事件的队列 ##

`DeltaFIFO` 与 `FIFO` 类型的区别：

1.  DeltaFIFO 缓存对象的事件列表，而 FIFO 缓存对象的最新值；
2.  FIFO 内部维护了一个对象缓存(f.items)，而 DeltaFIFO 需要和一个外部维护的、包含所有对象的缓存(knownObjects) 结合使用：
    
     *  Delete()：用 knownObjects 检查待删除的对象是否存在，如果不存在则直接返回，否则生成 Deleted 事件；
     *  Replace()：用传入的队列列表更新 DeltaFIFO，检查 knownObjects，为不在传入的对象列表中的对象生成 Deleted 事件；
     *  Resync()：将 knownObjects 中的对象同步到 DeltaFIFO 中，并生成 Sync 事件；
3.  Delete/Replace/Resync() 方法不会从 DeltaFIFO 删除/替换对象，而是生成对应的事件。DeltaFIFO 的消费者需要将他们从 knownObjects 删除（见后文）；
4.  DeltaFIFO 的 Pop/Get() 方法，返回的不是对象最新值，而是对象事件列表。

DeltaFIFO 适用的情况：

1.  你希望最多一个 worker 处理某个对象的事件(对象在队列中是唯一的，)；
2.  当 worker 处理该对象时，可以获得自上次弹出该对象以来的所有事件，如 Add/Updat/Delete（FIFO 只缓存和弹出对象的最新值）；
3.  你可以处理删除对象的事件（FIFO 不弹出被删除的对象）；
4.  你想周期处理所有的对象（Reflector 周期调用 Resync() 方法，将 knownObjects 中对象同步到 DeltaFIFO）；

DeltaFIFO 是一个生产者-消费者队列，生产者是 Reflector，消费者是 controller/sharedInformer/sharedIndexInformer。

函数 `NewDeltaFIFO()` 返回一个 `DeltaFIFO` 类型对象：

// 来源于 k8s.io/client-go/tools/cache/delta_fifo.go
    func NewDeltaFIFO(keyFunc KeyFunc, knownObjects KeyListerGetter) *DeltaFIFO { 
        f := &DeltaFIFO{ 
            // 对象事件缓存，Key 为对象 Key，Value 为该对象的事件列表类型 Deltas；
            items:        map[string]Deltas{ },
            // 对象弹出队列，缓存的是对象 Key，后续 Pop 方法按序弹出；
            queue:        []string{ },
            // 生成对象标识 Key 的函数，一般是预定义的 MetaNamespaceKeyFunc 函数；
            keyFunc:      keyFunc,
            // 关联的外部对象缓存
            knownObjects: knownObjects,
        }
        f.cond.L = &f.lock
        return f
    }

*  `knownObjects` 是外部的对象缓存，DelaFIFO **不对它进行更新**，只用它来查找对象。
 *  `keyFunc` 一般是预定义的 `MetaNamespaceKeyFunc` 函数，即提取对象的 Namespace/Name 作为标识 Key。

例如 NewIndexerInformer() 创建 knownObjects 和 DeltaFIFO 的过程如下：

// 来源于：k8s.io/client-go/tools/cache/controller.go
    ...
    clientState := NewIndexer(DeletionHandlingMetaNamespaceKeyFunc, indexers)
    ...
    fifo := NewDeltaFIFO(MetaNamespaceKeyFunc, clientState)

DeltaFIFO 的消费者根据从 DeltaFIFO 弹出的 Delta 对象对 knownObjects 缓存(上面的 clientState )进行更新，从而保证 FIFO 缓存和该缓存的一致性。

### DeltaFIFO 的生产者和消费者 ###

后续文章会介绍，各种 `Informer`（如 `Informer、IndexInformer、SharedInformer、SharedIndexInformer`）的初始化函数依次创建  
`knownObjects` 缓存、`DeltaFIO` 和 `controller`。`controller` 再将 `DeltaFIFO` 传给 Reflector，

**Reflector 的 `ListAndWatch()` 方法是 DeltaFIFO 的生产者**：

1.  List etcd 中(通过 kube-apiserver，下同)特定类型的所有对象，然后调用 DeltaFIFO 的 `Replace()` 方法，将它们完整同步到 DeltaFIFO；
2.  根据配置的 Resync 时间，**周期调用** DeltaFIFO 的 `Resync()` 方法（见后文），将 knownObjects 中的对象更新到 DeltaFIFO 中；
3.  阻塞式 Watch etcd 中特对类型的对象变化，根据事件的类型分别调用 DeltaFIFO 的 Add/Update/Delete（）方法，将对象更新到 DeltaFIFO；

Watch etcd 会**周期性的**超时(5min ~ 10min)，这时 `ListAndWatch()` 出错返回，Reflector 会等待一段时间再执行它，从而实现**周期的将 `etcd` 中特定类型的全部对象**同步到 `DeltaFIFO`。

**`controller` 是 `DeltaFIFO` 的消费者，它用 DeltaFIFO 弹出的对象更新 `knownObjects` 缓存**，然后调用注册的 OnUpdate/OnAdd/OnDelete 回调函数。  
详情参考 Reflector 和 controller 和 Informer 文档。

### 记录对象事件的 Delta、Deltas 和 DeletedFinalStateUnknown 类型 ###

DeltaFIFO 使用 Delta 类型记录对象的事件类型和发生**事件后**的对象值：

type Delta struct { 
        // DeltaType 可能是：Added、Deleted、Updated、Sync
    	Type   DeltaType
    	Object interface{ }
    }

DeltaFIFO Watch apiserver 过程中可能因网络等问题出现丢事件的情况，如果丢失了 Delete 事件，则后续 Reflector 重复执行 `ListAndWatch()` 方法从 apiserver 获取的对象集合 set1 会出现与 knownObjects 对象集合 set2 不一致的情况。  
为了保证两者一致，DeltaFIFO 的 `Replace()` 方法将位于 set1 但不在 set2 中的对象用 `DeletedFinalStateUnknown` 类型对象封装，再保存到 Delta Object 中。

type DeletedFinalStateUnknown struct { 
        // 对象的 Key
        Key string
        // knownObjects 缓存中的对象值
        Obj interface{ }
    }

`Replace()` 方法是**唯一**产生 `DeletedFinalStateUnknown` 类型对象的方法。

### Add() 和 Update() 方法 ###

// 来源于 k8s.io/client-go/tools/cache/delta_fifo.go
    func (f *DeltaFIFO) Add(obj interface{ }) error { 
        f.lock.Lock()
        defer f.lock.Unlock()
        f.populated = true
        // Added 类型事件；
        return f.queueActionLocked(Added, obj)
    }

`Update()` 方法和 `Add()` 方法类似，差别在于产生的是 `Updated` 类型 Delta 事件；

### queueActionLocked() 方法 ###

将对象的事件存入事件队列 f.items，如果事件队列中没有该对象则还将对象（Key）加入弹出队列（f.queue），另外它还做如下操作：

1.  如果事件类型为 Sync，且对象的事件列表中最后一个事件类型为 Deleted，则直接返回（没有必要再同步一个已删除的对象，一般是 Reflector 周期调用 Rsync() 方法产生的 Sync 事件）；
2.  合并连续重复的 Deleted 事件为一个；

func (f *DeltaFIFO) queueActionLocked(actionType DeltaType, obj interface{ }) error { 
        id, err := f.KeyOf(obj)
        if err != nil { 
            return KeyError{ obj, err}
        }
    
        // FIXME!!! 感觉这个逻辑不太对。产生 Sync 事件有两种情形：
        // 1. 周期的 Rsync(), 这时处理逻辑 OK；
        // 2. Reflector 执行 ListAndWatch() LIST etcd 获取特定类型的全部对象；
        if actionType == Sync && f.willObjectBeDeletedLocked(id) { 
            return nil
        }
        // 将对象保存到 Delta 中
        newDeltas := append(f.items[id], Delta{ actionType, obj})
        newDeltas = dedupDeltas(newDeltas)
    
        if len(newDeltas) > 0 { 
            if _, exists := f.items[id]; !exists { 
                f.queue = append(f.queue, id)
            }
            f.items[id] = newDeltas
            f.cond.Broadcast()
        } else { 
            delete(f.items, id)
        }
    }

### Delete() 方法 ###

如果 `f.knownObjects` 对象缓存和事件队列 `f.items` 中均没有待删除的对象，则**直接返回**，否则为对象生成 `Deleted` 事件(非`DeletedFinalStateUnknown` 类型)。

FIFO 的 Delete() 方法将对象从缓存中删除，而 DeltaFIFO 的 Delete() 方法**不将对象从事件缓存 f.items 和弹出队列 f.queue 删除**，而是后续弹出时，将它从 f.items、f.queue **和 f.knownObjects** 中删除。

### Get/GetByKey/List/ListKeys() 方法 ###

查询事件缓存 f.items，返回对象的事件列表 Deltas 或 Key 列表；

### Replace() 方法 ###

`Replace(list []interface{}, resourceVersion string)`

1.  为 list 中的每个对象生成 `Sync` 事件；
2.  遍历 f.knownObjects 中对象，对不在传入的 list 中的对象，用 `DeletedFinalStateUnknown` 类型对象封装，再保存到 Deleted 类型的 Delta.Object 中；

Reflector 的 `ListAndWatch()` 方法因 Watch 超时而周期调用 `Replace()` 方法，从而周期地将 etcd 中特定类型的所有对象同步到 DeltaFIFO 中。  
`controller` 用 DeltaFIFO 弹出的对象更新 `knownObjects` 缓存，详情参考 Reflector 和 controller 和 Informer 文档。

`Replace()` 方法是**唯一**产生 `DeletedFinalStateUnknown` 类型对象的方法。

### Resync() 方法 ###

遍历 knownObjects 中的对象：

1.  如果该对象位于事件缓存 f.items 中，则跳过；
2.  否则，生成 Sync 事件；

前文我们提到 DeltaFIFO 的使用场景之一是：**“你想周期处理所有的对象”**，而这是通过周期将 knownObjects 中的对象同步到 DeltaFIFO 来实现的。

Reflector 的 `ListAndWatch()` 方法周期执行 DeltaFIFO 的 Resync() 方法，将 knownObjects 中的对象同步到 DeltaFIFO（产生 Sync 事件），从而有机会再次调用注册的 `OnUpdate()` 处理函数。

**只有 Replace() 和 Rsync() 方法才产生 Sync 事件**。

### Pop() 方法 ###

Pop(process PopProcessFunc)

1.  如果弹出队列 f.queue 为空，则**阻塞等待**；
2.  每次弹出队列头部对象的事件列表（Deltas 类型），然后将该对象的事件列表从缓存(f.items)中删除;
3.  调用配置的回调函数 PopProcessFunc（传入事件列表 Deltas）；

如果函数 PopProcessFunc() 执行失败，应该调用 `AddIfNotPresent()` 方法将 Deltas 重新加回 DeltaFIFO，这样后续可以再次被弹出处理，防止丢事件。（controler 已实现该逻辑）

注意，Pop() 方法是在加锁的情况下调用 PopProcessFunc 的，所以在多个 goroutine 并发调用 Pop() 的情况下，它们实际是**串行**执行的。

### HasSyncd() 方法 ###

创建 DealtaFIFO 后，如果首先调用的是 `Replace()` 方法，则 `f.populated` 被设置为 `true`，`f.initialPopulationCount` 被设置为传入的对象数量。当这**第一批**对象都被弹出完毕时（包含弹出前被删除的对象），`HasSynced()` 方法返回 `true`：

// 来源于 k8s.io/client-go/tools/cache/fifo.go
    func (f *DeltaFIFO) HasSynced() bool { 
        f.lock.Lock()
        defer f.lock.Unlock()
        return f.populated && f.initialPopulationCount == 0
    }

另外，如果在调用 `Replace()` 方法前，**首先**调用了 `Add/Update/Delete/AddIfNotPresent()` 方法，则 `HasSynced()` 方法也会返回 `true`。

第一批对象弹出完毕后，后续不管是否再次调用 Replace()或其它方法，HasSynced() 方法**总是返回 true**

### DeltaFIFO 和 knownObjects 对象缓存的同步 ###

1.  Reflector 从 etcd List 出特定类型的所有对象，调用 DeltaFIFO 的 Replace() 方法为各对象生成 Sync 事件，此时 knownObjects 对象缓存为空；
2.  controller 从 DeltaFIFO 弹出对象事件列表 Deltas，遍历 Deltas，根据 Delta 的事件类型更新 knownObjects，从而实现 DeltaFIFO 和 knownObjects 缓存中的对象一致：
    
    controller 每次**启动**时，因 knownObjects 为空且事件类型为 Sync，所以会为同步来的所有对象：
    
    1.  调用 knownObjects 的 **Add() 方法**，将它们加入到对象缓存；
    2.  调用注册的 OnAdd() 回调函数。所以**第一次对象同步时， controller 也会调用用户注册的 OnAdd() 回调函数**。
    
        // 来源于：k8s.io/client-go/tools/cache/controller.go
        for _, d := range obj.(Deltas) { 
            switch d.Type { 
            // Replace() 方法生成的 Sync 事件涉及到的对象，
            case Sync, Added, Updated:
                // clientState 即为 knownObjects 对象缓存
                if old, exists, err := clientState.Get(d.Object); err == nil && exists { 
                    if err := clientState.Update(d.Object); err != nil { 
                        return err
                    }
                    h.OnUpdate(old, d.Object)
                } else { 
                    if err := clientState.Add(d.Object); err != nil { 
                        return err
                    }
                    h.OnAdd(d.Object)
                }
            case Deleted:
                // 先从缓存中删除，再调用处理函数
                if err := clientState.Delete(d.Object); err != nil { 
                    return err
                }
                h.OnDelete(d.Object)
            }
        }
3.  但是，Reflector 的 Watch 可能会出现**丢失事件**的情况（如 ListAndWatch 出错返回后，Reflector 会 Sleep 一段时间再执行它，期间 etcd 的对象变化事件丢失），这样再次 List 到的对象集合 set1 与 knownObjects 缓存中的对象集合 set2 不一致。如何解决这个问题呢？
    
    答案在于：List 到对象集合后，DeltaFIFO 调用的 `Replace()` 方法将位于 set1 但不在 set2 中的对象用 `DeletedFinalStateUnknown` 类型对象封装，再保存到 Delta.Object 中。而上面 handlerObject() 的参数即为 Delta.Object。
4.  ListAndWatch() 方法起一个 goroutine，周期调用 Resync() 方法，将 knownObjects 中的对象更新到 DeltaFIFO。为何要这么做呢？
    
    前文我们提到 DeltaFIFO 的使用场景之一是：**“你想周期处理所有的对象”**，但对象一旦从 DeltaFIFO 中弹出，如果没有产生新的 Watch 事件，就不会对它再调用注册的回调函数。Reflector 的 `ListAndWatch()` 方法会周期执行 DeltaFIFO 的 Resync() 方法，目的就是**为对象产生新的 Sync 事件**，从而有机会再次调用注册的 `OnUpdate()` 处理函数。因此 Resync 时，如果对象已经在 f.items，则后续由机会被弹出，所以不需要为它生成 Sync 事件。

后续文章会介绍，`controller` 一般是在 `Informer` 中使用的，`controller` 调用的 `OnUpdate()` 函数会调用用户创建 `Informer` 时传入的 `ResourceEventHandler` 中的 `OnUpdate()` 函数。所以用户的 `OnUpdate()` 函数可能会因 DeltaFIFO 的周期 Resync() 而被调用，它应该检查传入的 old、new 对象是否发生变化，未变化时直接返回：

// 来源于 https://github.com/kubernetes/sample-controller/blob/master/controller.go#L131
    deployInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{ 
        // 第一次从 etcd 同步的对象会生成 Added 事件，调用该函数
        // 后续 Added 事件就代表的确有**新创建**的资源对象；
        AddFunc: controller.handleDeploy,
        UpdateFunc: func(old, new interface{ }) { 
            // 由于 Reflector 周期调用 DeltaFIFO 的 Rsync() 方法，`controller` 会调用注册的 OnUpdate 回调函数，所以需要判断新旧对象是否相同，如果相同则不需处理
            // 也可以用
            newDepl := new.(*extensionsv1beta1.Deployment)
            oldDepl := old.(*extensionsv1beta1.Deployment)
            if newDepl.ResourceVersion == oldDepl.ResourceVersion { 
                // Periodic resync will send update events for all known Deployments.
                // Two different versions of the same Deployment will always have different RVs.
                return
            }
            controller.handleDeploy(new)
        },
        DeleteFunc: controller.handleDeploy,
    })

另外，前面的第 2 条提到过，`controller` 刚启动时，knownObjects 为空，会为从 etcd 同步来的特定类型的所有对象生成 Added 事件，进而调用上面注册的 AddFunc。