K8sKubernetes

K8s Informer原理解析

2021-08-18  本文已影响0人  王勇1024

背景

什么是 Informer 机制

Informer 是 client-go 中的核心工具包，已经被 kubernetes 中众多组件所使用。所谓 Informer，其实就是一个带有本地缓存和索引机制的、可以注册 EventHandler 的 client，本地缓存被称为 Store，索引被称为 Index。使用 informer 的目的是为了减轻 apiserver 数据交互的压力而抽象出来的一个 cache 层, 客户端对 apiserver 数据的 "读取" 和 "监听" 操作都通过本地 informer 进行。

Informer 的主要功能：

为什么需要 Informer 机制？

我们知道Kubernetes各个组件都是通过REST API跟API Server交互通信的，而如果每次每一个组件都直接跟API Server交互去读取/写入到后端的etcd的话，会对API Server以及etcd造成非常大的负担。 而Informer机制是为了保证各个组件之间通信的实时性、可靠性，并且减缓对API Server和etcd的负担。

Informer 需要满足哪些要求？

核心功能

Informer的工作流程

List & Watch

List所做的，就是向API Server发送一个http短链接请求，罗列所有目标资源的对象。而Watch所做的是实际的“监听”工作，通过http长链接的方式，其与API Server能够建立一个持久的监听关系，当目标资源发生了变化时，API Server会返回一个对应的事件，从而完成一次成功的监听，之后的事情便交给后面的handler来做。

这样一个List & Watch机制，带来了如下几个优势：

  1. 事件响应的实时性：通过Watch的调用，当API Server中的目标资源产生变化时，能够及时的收到事件的返回，从而保证了事件响应的实时性。而倘若是一个轮询的机制，其实时性将受限于轮询的时间间隔。

  2. 事件响应的可靠性：倘若仅调用Watch，则如果在某个时间点连接被断开，就可能导致事件被丢失。List的调用带来了查询资源期望状态的能力，客户端通过期望状态与实际状态的对比，可以纠正状态的不一致。二者结合保证了事件响应的可靠性。

  3. 高性能：倘若仅周期性地调用List，轮询地获取资源的期望状态并在与当前状态不一致时执行更新，自然也可以do the job。但是高频的轮询会大大增加API Server的负担，低频的轮询也会影响事件响应的实时性。Watch这一异步消息机制的结合，在保证了实时性的基础上也减少了API Server的负担，保证了高性能。

  4. 事件处理的顺序性：我们知道，每个资源对象都有一个递增的ResourceVersion，唯一地标识它当前的状态是“第几个版本”，每当这个资源内容发生变化时，对应产生的事件的ResourceVersion也会相应增加。在并发场景下，K8s可能获得同一资源的多个事件，由于K8s只关心资源的最终状态，因此只需要确保执行事件的ResourceVersion是最新的，即可确保事件处理的顺序性。

ResourceVersion

Kubernetes 请求并发控制与数据一致性（含ResourceVersion、Update、Patch简析）

Kubernetes-resourceVersion机制分析

秘诀就是 <u>Chunked transfer encoding(分块传输编码)</u>，它首次出现在HTTP/1.1。正如维基百科所说：

HTTP 分块传输编码允许服务器为动态生成的内容维持 HTTP 持久链接。通常，持久链接需要服务器在开始发送消息体前发送Content-Length消息头字段，但是对于动态生成的内容来说，在内容创建完之前是不可知的。使用分块传输编码，数据分解成一系列数据块，并以一个或多个块发送，这样服务器可以发送数据而不需要预先知道发送内容的总大小。

当客户端调用 watch API 时，apiserver 在response 的 HTTP Header 中设置 Transfer-Encoding的值为chunked，表示采用分块传输编码，客户端收到该信息后，便和服务端该链接，并等待下一个数据块，即资源的事件信息。例如：

$ curl -i http://{kube-api-server-ip}:8080/api/v1/watch/pods?watch=yesHTTP/1.1 200 OKContent-Type: application/json
Transfer-Encoding: chunked
Date: Thu, 02 Jan 2019 20:22:59 GMTTransfer-Encoding: chunked{"type":"ADDED", "object":{"kind":"Pod","apiVersion":"v1",...}}{"type":"ADDED", "object":{"kind":"Pod","apiVersion":"v1",...}}{"type":"MODIFIED", "object":{"kind":"Pod","apiVersion":"v1",...}}

Informer 能保证通过list+watch不会丢失事件，如果网络抖动重新恢复后，watch会带着之前的resourceVersion号重连，resourceVersion是单调递增的， API Server 收到该请求后会将所有大于该resourceVersion的变更同步过来。

二级缓存

二级缓存属于 Informer 的底层缓存机制，这两级缓存分别是 DeltaFIFO 和 LocalStore。这两级缓存的用途各不相同。DeltaFIFO 用来存储 Watch API 返回的各种事件 ，LocalStore 只会被 Lister 的 List/Get 方法访问 。

如果K8s每次想查看资源对象的状态，都要经历一遍List调用，显然对 API Server 也是一个不小的负担，对此，一个容易想到的方法是使用一个cache作保存，需要获取资源状态时直接调cache，当事件来临时除了响应事件外，也对cache进行刷新。

虽然 Informer 和 Kubernetes 之间没有 resync 机制，但 Informer 内部的这两级缓存之间存在 resync 机制。

Resync

Resync 机制会将 Indexer 的本地缓存重新同步到 DeltaFIFO 队列中。一般我们会设置一个时间周期，让 Indexer 周期性地将缓存同步到队列中。直接 list/watch API Server 就已经能拿到集群中资源对象变化的 event 了，这里引入 Resync 的作用是什么呢？去掉会有什么影响呢？

自定义事件处理

ResourceEventHandler 用于处理对象的变更事件，用户可以通过实现 ResourceEventHandler 接口，并调用 sharedIndexInformer.AddEventHandler() 或 sharedIndexInformer.AddEventHandlerWithResyncPeriod() 方法注册到 sharedProcessor 中。这样，当数据发送变化时，就会回调 ResourceEventHandler 中对应的 OnAdd/OnUpdate/OnDelete 方法来实现用户自定义的处理逻辑

核心对象

Informer相关

client-go 中提供了几种不同的 Informer：

这里带有 Indexer 和不带 Indexer 的大家好理解写，从字面意思来看，就是一个是带有 Indexer 功能一个不带有 Indexer 功能的 Informer。而这里的 Shared 的 Informer 引入，其实是因为随着 K8S 中，相同资源的监听者在不断地增加，从而导致很多调用者通过 Watch API 对 API Server 建立一个长连接去监听事件的变化，这将严重增加了 API Server 的工作负载，及资源的浪费。比如在 kube-controller-manager 组件中，有很多控制管理都需要监听 Pod 资源的变化，如果都独立的调用 Informer 去维护一个对 APIServer 的长连接，这将导致 kube-controller-manager 中资源的浪费及增加了 APIServer 的负载，而不同控制管理者通过创建 Shared 的 Informer 则实现了这些控制管理者使用同一个 Watch 去和 APIServer 建立长连接，并在收到事件后，分发给下游的调用者。

SharedInformer

我们平时说的 Informer 其实就是 SharedInformer，它是可以共享使用的。如果同一个资源的 Informer 被实例化多次，那么就会运行多个 ListAndWatch 操作，这会加大 APIServer 的压力。而 SharedInformer 通过一个 map 来让同一类资源的 Informer 实现共享一个 Refelctor，这样就不会出现上面这个问题了。

Informer通过Local Store缓存目标资源对象，且仅为自己所用。但是在K8s中，一个Controller可以关心不止一种资源，使得多个Controller所关心的资源彼此会存在交集。如果几个Controller都用自己的Informer来缓存同一个目标资源，显然会导致不小的空间开销，因此K8s引入了SharedInformer来解决这个问题。

SharedInformer拥有为多个Controller提供一个共享cache的能力，从而避免资源缓存的重复、减小空间开销。除此之外，一个SharedInformer对一种资源只建立一个与API Server的Watch监听，且能够将监听得到的事件分发给下游所有感兴趣的Controller，这也显著地减少了API Server的负载压力。实际上，K8s中广泛使用的都是SharedInformer，Informer则出场甚少。

SharedIndexInformer

SharedIndexInformer 扩展了 SharedInformer 接口，提供了构建索引的能力。

type SharedIndexInformer interface {
    SharedInformer
    // AddIndexers add indexers to the informer before it starts.
    AddIndexers(indexers Indexers) error
    GetIndexer() Indexer
}

SharedIndexInformerFactory

使用sharedInformerFactory可以统一管理控制器中需要的各资源对象的informer实例，避免同一个资源创建多个实例

默认的 Informer 实现

Informer 机制为 K8s 的各种对象提供了默认的 Informer 实现，可以通过以下方式快速创建一个 Informer 对象，并交由 SharedIndexInformerFactory 统一管理。

SharedInformerFactory.Core().V1().Nodes()
SharedInformerFactory.Core().V1().Pods()
SharedInformerFactory.Apps().V1().Deployments()
SharedInformerFactory.Core().V1().Secrets()
SharedInformerFactory.Batch().V1beta1().CronJobs()
……

List-Watch相关

Reflector

Reflector用来watch特定的k8s API资源。具体的实现是通过ListAndWatch的方法，watch可以是k8s内建的资源或者是自定义的资源。当reflector通过watch API接收到有关新资源实例存在的通知时，它使用相应的列表API获取新创建的对象，并将其放入watchHandler函数内的Delta Fifo队列中。

ListerWatcher

ListerWatcher 是 Informer 机制中的核心对象之一，其功能是通过 List() 方法从 API Server 中获取某一类型的全量数据，再通过 Watch() 方法监听 API Server 中数据的增量更新。

ListerWatcher 继承自 Lister 和 Watcher 接口，从而使其既能获取全量数据，又能监听增量数据更新。

type ListerWatcher interface {
  Lister
  Watcher
}

Lister

Lister 接口用于完成全量数据的初始化。

type Lister interface {
  // List 方法返回一个对象列表，可以通过 options 参数的 FieldSelector 指定要提取的字段，ResourceVersion 指定读取数据的起始位置
  List(options metav1.ListOptions) (runtime.Object, error)
}

Watcher

Watcher 接口用于监听数据的增量更新。

type Watcher interface {
  // 监听数据更新，需要指定 ResourceVersion
  Watch(options metav1.ListOptions) (watch.Interface, error)
}

事件队列相关

Store

Store是一个通用的对象存储接口，其中定义了一系列与对象增删改查相关的方法。Store 要求对象有唯一键，键的计算方式由接口实现类中关联的 KeyFunc 决定的。

Queue

从 Queue 接口的定义可以看出，它继承自 Store 接口，所以其具备基本的数据存取能力。同时，它又具备从队列头部取出数据并调用 PopProcessFunc 处理头部数据并返回处理结果的能力。

DeltaFIFO

DeltaFIFO 是一个生产者-消费者的队列，生产者是 Reflector，消费者是 Pop 函数，从架构图上可以看出 DeltaFIFO 的数据来源为 Reflector，通过 Pop 操作消费数据，消费的数据一方面存储到 Indexer 中，另一方面可以通过 Informer 的 handler 进行处理，Informer 的 handler 处理的数据需要与存储在 Indexer 中的数据匹配。需要注意的是，Pop 的单位是一个 Deltas，而不是 Delta。

Delta

Delta 是 DeltaFIFO 存储的类型，它记录了对象发生了什么变化以及变化后对象的状态。如果变更是删除，它会记录对象删除之前的最终状态。

type Delta struct {
    Type   DeltaType
    Object interface{}
}

Deltas

Deltas 保存了对象状态的变更（Add/Delete/Update）信息（如 Pod 的删除添加等），Deltas 缓存了针对相同对象的多个状态变更信息，如 Pod 的 Deltas[0]可能更新了标签，Deltas[1]可能删除了该 Pod。最老的状态变更信息为 Oldest()，最新的状态变更信息为 Newest()，使用中，获取 DeltaFIFO 中对象的 key 以及获取 DeltaFIFO 都以最新状态为准。

最旧的 delta 在索引0位置，最新的 delta 在最后一个索引位置。

type Deltas []Delta

DeltaType

// DeltaType 是变化类型 (添加、删除等)
type DeltaType string

const (
    Added   DeltaType = "Added"  // 增加
    Updated DeltaType = "Updated"  // 更新
    Deleted DeltaType = "Deleted"  // 删除
    Sync DeltaType = "Sync"  // 同步
)

DeltaFIFO 中数据存储形式

事件处理相关

Controller

Processor

ResourceEventHandler

当经过List & Watch得到事件时，接下来的实际响应工作就交由ResourceEventHandler来进行，这个Interface定义如下：

type ResourceEventHandler interface {
    // 添加对象回调函数
    OnAdd(obj interface{})
    // 更新对象回调函数
    OnUpdate(oldObj, newObj interface{})
    // 删除对象回调函数
    OnDelete(obj interface{})
}

当事件到来时，Informer根据事件的类型（添加/更新/删除资源对象）进行判断，将事件分发给绑定的EventHandler，即分别调用对应的handle方法（OnAdd/OnUpdate/OnDelete），最后EventHandler将事件发送给Workqueue。

缓存相关

Indexer

Indexer在Store基础上扩展了索引能力，就好比给数据库添加的索引，以便查询更快，那么肯定需要有个结构来保存索引。典型的索引用例是基于对象标签创建索引。 Indexer可以根据多个索引函数维护索引。Indexer使用线程安全的数据存储来存储对象及其键。 在Store中定义了一个名为MetaNamespaceKeyFunc 的默认函数，该函数生成对象的键作为该对象的<namespace> / <name>组合。

Reflector 通过 ListAndWatch 把数据传入 DeltaFIFO 后，经过 DeltaFIFO 的 Pop 函数将资源对象存入到了本地的一个存储 Indexer 中，而这个底层真正的存储其实就是上面的 ThreadSafeStore。

ThreadSafeStore

使用示例

    var kubeconfig *string
    if home := homedir.HomeDir(); home != "" {
        kubeconfig = flag.String("kubeconfig", filepath.Join(home, ".kube", "config"), "(optional) absolute path to the kubeconfig file")
    } else {
        kubeconfig = flag.String("kubeconfig", "", "absolute path to the kubeconfig file")
    }
    flag.Parse()

    config, err := clientcmd.BuildConfigFromFlags("", *kubeconfig)
    if err != nil {
        panic(err)
    }

    // 初始化 client
    clientset, err := kubernetes.NewForConfig(config)
    if err != nil {
        log.Panic(err.Error())
    }

    stopper := make(chan struct{})
    defer close(stopper)

    // 初始化 informer
    factory := informers.NewSharedInformerFactory(clientset, 0)
    nodeInformer := factory.Core().V1().Nodes()
    informer := nodeInformer.Informer()
    defer runtime.HandleCrash()

    // 启动 informer，list & watch
    go factory.Start(stopper)

    // 从 apiserver 同步资源，必不可少
    if !cache.WaitForCacheSync(stopper, informer.HasSynced) {
        runtime.HandleError(fmt.Errorf("Timed out waiting for caches to sync"))
        return
    }

    // 使用自定义 handler
    informer.AddEventHandler(cache.ResourceEventHandlerFuncs{
        AddFunc:    onAdd,
        UpdateFunc: func(interface{}, interface{}) { fmt.Println("update not implemented") }, // 此处省略 workqueue 的使用
        DeleteFunc: func(interface{}) { fmt.Println("delete not implemented") },
    })

实现原理

构建过程

启动过程

List 过程

Watch过程

变更分发过程

变更处理过程（待补充）

完整流程

使用中的一些问题

Informer如何保证数据一致性？

list 和 watch 一起保证了消息的可靠性，避免因消息丢失而造成状态不一致场景。具体而言，list API可以查询当前的资源及其对应的状态(即期望的状态)，客户端通过拿期望的状态和实际的状态进行对比，纠正状态不一致的资源。Watch API 和 apiserver保持一个长链接，接收资源的状态变更事件并做相应处理。如果仅调用 watch API，若某个时间点连接中断，就有可能导致消息丢失，所以需要通过list API解决消息丢失的问题。从另一个角度出发，我们可以认为list API获取全量数据，watch API获取增量数据。虽然仅仅通过轮询 list API，也能达到同步资源状态的效果，但是存在开销大，实时性不足的问题。

Informer中为什么要引入Resync机制？

因为在处理 SharedInformer 事件回调时，可能存在处理失败的情况，定时的 Resync 让这些处理失败的事件有了重新 onUpdate 处理的机会。

Resync 机制的引入，定时将 Indexer 缓存事件重新同步到 Delta FIFO 队列中，在处理 SharedInformer 事件回调时，让处理失败的事件得到重新处理。并且通过入队前判断 FIFO 队列中是否已经有了更新版本的 event，来决定是否丢弃 Indexer 缓存不进行 Resync 入队。在处理 Delta FIFO 队列中的 Resync 的事件数据时，触发 onUpdate 回调来让事件重新处理。

如何解决内存溢出问题？

参考资料

理解 K8S 的设计精髓之 List-Watch机制和Informer模块

深入了解 Kubernetes Informer

深入k8s：Informer使用及其源码分析

Kubernetes Informer 与 Lister 详解

k8s client-go informer中的processorlistener数据消费，缓存的分析

Informer 中为什么需要引入 Resync 机制？

informer之controller和shared_informer

Kubernetes 是怎么实现定时任务的

client-go 之 Informer 篇

上一篇下一篇

猜你喜欢

热点阅读