deltaFIFO
DeltaFIFO: 这是一个特殊类型的队列，它结合了FIFO（先进先出）队列的特性和增量（Delta）处理的能力。DeltaFIFO 中是按顺序存储的，但它们不必严格按照发生的顺序逐个处理。这种设计提供了处理的灵活性和优化的机会，允许控制器根据实际需求选择最有效的处理策略。这是 DeltaFIFO 设计的一个重要特性，使其能够高效地处理复杂的资源变化场景。
Delta 实际上是一个结构体（struct）,它在 Kubernetes 的 client-go 库中定义。
type Delta struct {
Type DeltaType
Object interface{}
}interface{} 允许 Delta 结构体存储任何类型的 Kubernetes 对象，不仅仅是 Pod。这使得 Delta 结构体可以用于所有类型的 Kubernetes 资源。
type DeltaFIFO struct {
// 用于存储对象键的队列
queue []string
// 存储每个键对应的 Delta 列表
items map[string][]Delta
// 其他字段，如锁、条件变量等
}
DeltaType 的可能值：
const (
Added DeltaType = "Added"
Updated DeltaType = "Updated"
Deleted DeltaType = "Deleted"
// 可能还有其他类型，如 Sync
)
可以看到图中一个key对应一个delta值，当需要使用存储的对象时，通常需要进行类型断言：
if pod, ok := delta.Object.(*v1.Pod); ok {
// 使用 pod 对象
}// 存储 Pod
podDelta := Delta{
Type: Added,
Object: &v1.Pod{Metadata: metav1.ObjectMeta{Name: "mypod"}},
}
// 存储 Service
serviceDelta := Delta{
Type: Updated,
Object: &v1.Service{Metadata: metav1.ObjectMeta{Name: "myservice"}},
}controller消费
Controller 通过调用 DeltaFIFO 的 Pop 方法来消费队列中的项目。这个方法通常在控制器的主循环中被调用。
func (c *Controller) processNextItem() bool {
obj, shutdown := c.queue.Get()
if shutdown {
return false
}
defer c.queue.Done(obj)
err := func(obj interface{}) error {
deltas, ok := obj.(cache.Deltas)
if !ok {
return fmt.Errorf("expected cache.Deltas, got %v", obj)
}
for _, delta := range deltas {
switch delta.Type {
case cache.Added:
// 首先更新 Indexer
if err := c.indexer.Add(delta.Object); err != nil {
return err
}
// 然后调用事件处理函数
c.handleAddition(delta.Object)
case cache.Updated:
// 首先更新 Indexer
if err := c.indexer.Update(delta.Object); err != nil {
return err
}
// 然后调用事件处理函数
c.handleUpdate(delta.Object)
case cache.Deleted:
// 首先更新 Indexer
if err := c.indexer.Delete(delta.Object); err != nil {
return err
}
// 然后调用事件处理函数
c.handleDeletion(delta.Object)
}
}
return nil
}(obj)
if err != nil {
utilruntime.HandleError(err)
c.queue.AddRateLimited(obj)
return true
}
c.queue.Forget(obj)
return true
}初始化阶段：
创建 Informer（通常是 SharedInformer）
创建 Controller
将 Controller 的事件处理函数注册到 Informer
数据流动：
a. API 服务器 -> Reflector：
Reflector 通过 client-go API 监听 Kubernetes API 服务器
获取资源对象（如 Pod）的变化
b. Reflector -> DeltaFIFO：
Reflector 将这些变化（Delta）放入 DeltaFIFO 队列
c. DeltaFIFO -> Controller：
Controller 的 processLoop 方法从 DeltaFIFO 队列中取出数据
使用 Pop 方法，该方法包含一个 process 回调函数
d. Controller -> Indexer：
process 回调函数处理 Delta
更新 Indexer（本地缓存）
e. Controller -> 事件处理：
调用相应的事件处理函数（如 OnAdd, OnUpdate, OnDelete）
func (s *sharedIndexInformer) HandleDeltas(obj interface{}, isInInitialList bool) error {
s.blockDeltas.Lock()
defer s.blockDeltas.Unlock()
if deltas, ok := obj.(Deltas); ok {
return processDeltas(s, s.indexer, deltas, isInInitialList)
}
return errors.New("object given as Process argument is not Deltas")
}
// Multiplexes updates in the form of a list of Deltas into a Store, and informs
// a given handler of events OnUpdate, OnAdd, OnDelete
func processDeltas(
// Object which receives event notifications from the given deltas
handler ResourceEventHandler,
clientState Store,
deltas Deltas,
isInInitialList bool,
) error {
// from oldest to newest
for _, d := range deltas {
obj := d.Object
switch d.Type {
case Sync, Replaced, Added, Updated:
if old, exists, err := clientState.Get(obj); err == nil && exists {
if err := clientState.Update(obj); err != nil {
return err
}
handler.OnUpdate(old, obj)
} else {
if err := clientState.Add(obj); err != nil {
return err
}
handler.OnAdd(obj, isInInitialList)
}
case Deleted:
if err := clientState.Delete(obj); err != nil {
return err
}
handler.OnDelete(obj)
}
}
return nil
}Resync机制会将Indexer本地存储中的资源对象同步到DeltaFIFO中，并将这些资源对象设置为Sync的操作类型。Resync函数在Reflector中定时执行，它的执行周期由NewReflector函数传入的resyncPeriod参数设定。
