您的位置:首页 > 文旅 > 旅游 > [k8s源码]8.deltaFIFO

[k8s源码]8.deltaFIFO

2024/10/5 1:18:52 来源:https://blog.csdn.net/weixin_45396500/article/details/140632377  浏览:    关键词:[k8s源码]8.deltaFIFO

deltaFIFO 

DeltaFIFO: 这是一个特殊类型的队列,它结合了FIFO(先进先出)队列的特性和增量(Delta)处理的能力。DeltaFIFO 中是按顺序存储的,但它们不必严格按照发生的顺序逐个处理。这种设计提供了处理的灵活性和优化的机会,允许控制器根据实际需求选择最有效的处理策略。这是 DeltaFIFO 设计的一个重要特性,使其能够高效地处理复杂的资源变化场景。

 Delta 实际上是一个结构体(struct),它在 Kubernetes 的 client-go 库中定义。

type Delta struct {Type   DeltaTypeObject 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:// 首先更新 Indexerif err := c.indexer.Add(delta.Object); err != nil {return err}// 然后调用事件处理函数c.handleAddition(delta.Object)case cache.Updated:// 首先更新 Indexerif err := c.indexer.Update(delta.Object); err != nil {return err}// 然后调用事件处理函数c.handleUpdate(delta.Object)case cache.Deleted:// 首先更新 Indexerif 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 deltashandler ResourceEventHandler,clientState Store,deltas Deltas,isInInitialList bool,
) error {// from oldest to newestfor _, d := range deltas {obj := d.Objectswitch 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参数设定。

版权声明:

本网仅为发布的内容提供存储空间,不对发表、转载的内容提供任何形式的保证。凡本网注明“来源:XXX网络”的作品,均转载自其它媒体,著作权归作者所有,商业转载请联系作者获得授权,非商业转载请注明出处。

我们尊重并感谢每一位作者,均已注明文章来源和作者。如因作品内容、版权或其它问题,请及时与我们联系,联系邮箱:809451989@qq.com,投稿邮箱:809451989@qq.com