- (三)Kubernetes 源码剖析之学习Informer机制
- 如何高效掌控K8s资源变化？K8s Informer实现机制浅析
- 25 | 深入解析声明式API（二）：编写自定义控制器
- k8s client-go informer中的processorlistener数据消费，缓存的分析
- client-go系列之5—Informer
- Kubernetes Informer 与 Lister 详解
- 戚风蛋糕制作模板表单 —— CRD，Pod 的定义（API 的类型，资源的类型），也可以成为编程语言中的 类型（如 struct 的定义）
- 基于模板用户提交的个性化表单 —— 用户提交的 yaml —— 在 k8s 形成 API 对象 （对 struct 的填充）
- 【API 对象】包含两部分内容：
- 【key】为【API 对象名】，一般表示为【namespace/name】
- 【value】为 【API 对象内容】（提交的 yaml内容 ，如 pod 的各个字段的值等），一般称为【object】
- 最后制作出的蛋糕 —— 资源对象 —— 可以理解为 实例 （最后形成的 struct 对象）
-
Informer：
-
**Reflector：**一个 【Reflector】 监听【一种 API 对象】 的变化【Delta】（可能是【新增、更新、创建】该类型的【 API 对象】）
- 【Delta】可以认为是 【 一个 API 对象】+ 【该对象的事件类型（新增、更新、创建）】
-
**DeltaFIFO：**将此【Delta】存入【DeltaFIFO队列中】，主要是为了防止【Delta产生过快、导致遗失或处理不过来，而设置的缓冲区】
-
【DeltaFIFO队列】弹出【Delta】
-
根据【Delta】的【事件类型（新增、更新、创建）】调用不同的处理函数【AddFunc、UpdateFunc、DeleteFunc】，之后将此【Delta的 API 对象的名称 namespace/name】 存入到 【workqueue】中
-
与此同时，该【Delta】也会触发【索引Indexer】和【缓存 Local Store】的同步
- 根据【Delta的 API 对象的名称 namespace/name】构建【索引Indexer】，便于通过【某一特征（如ns）】快速找到【对应的所有API对象（如该ns下的所有Pod）】（可理解为 kubectl get pod -n 某个ns）
- 根据【Delta的 API 对象的名称 namespace/name 和 object 】构建或更新【缓存 Local Store】，【新增、更新事件】就【在缓存中新增】，【删除事件】就【在缓存中删除】，从而保障【缓存永远是最新的】
- 备注：
- 这里 Indexer「索引」 和 Local Store 「缓存」 是分开表示的
- 在源码级别，基本上是一起实现的，一个 Indexer 接口涵盖，一个结构体 cache 实现
-
-
-
用户自定义的 Control Loop：
- 从【workqueue】中取出【该处理的API对象的名称】，通过【Informer的Lister方法】根据【该处理的API对象的名称】获取对应的【object（即期望状态）】
- 【自行实现逻辑】从【真实环境中】获取【真实状态】
- 最后【调谐】，将【环境中资源】实现从【真实状态】转变为【目标状态】
- 上面已经将大体逻辑了解清楚了，下面的框架图辅助【理解源码架构】
- Controller 不是上面的 Control Loop，这里指的是【 Informer 的内部管家】，这里主要是协调 Reflector
- Reflector 通过 【ListAndWatch】机制监听【API对象的变化，及事件类型】 形成 【Delta】放入【DeltaFIFO 队列】
- 之后从【DeltaFIFO 队列】pop弹出【Delta】，之后交给【HandleDeltas函数处理】
- 【HandleDeltas函数】首先会通知【所有的订阅者Listener（也就是多个Informer实例）】，同时会更新【索引和缓存（这里代码里统称为Indexer接口）】
- 上面已经将大体逻辑了解清楚了，下面的框架图辅助【理解源码架构】
- Controller 不是上面的 Control Loop，这里指的是【 Informer 的内部管家】，这里主要是协调 Reflector
- Reflector 通过 【ListAndWatch】机制监听【API对象的变化，及事件类型】 形成 【Delta】放入【DeltaFIFO 队列】
- 之后从【DeltaFIFO 队列】pop弹出【Delta】，之后交给【HandleDeltas函数处理（也就是Process函数，这是个变量）】
- 【HandleDeltas函数】首先会通知【所有的订阅者Listener（也就是多个Informer实例）】，同时会更新【索引和缓存（这里代码里统称为Indexer接口）】
- 处理过程中通过【channel addCh 和 nextCh】进行【Delta传输】，为了避免【数据丢失、和压力过大】，采用【环形缓存pendingNotifications】进行缓存，之后【nextCh】取出，进行传递
Informer也被称为Shared Informer，它是可以共享使用的。在用client-go编写代码程序时，若同一资源的Informer被实例化了多次，每个Informer使用一个Reflector，那么会运行过多相同的ListAndWatch，太多重复的序列化和反序列化操作会导致Kubernetes API Server负载过重。Shared Informer可以使同一类资源Informer共享一个Reflector，这样可以节约很多资源。通过map数据结构实现共享的Informer机制。SharedInformer定义了一个map数据结构，用于存放所有Informer的字段
简言之，ShareInformer 指的是（一类资源共享同一个 Reflector 、Indexer 和 Local Store）
- 对某一类资源（如Pod）只用一个 Reflector 去监听（ListAndWatch）资源的变化
- 同时，将资源存储在同一个缓存中（Indexer 和 Local Store），缓存中的消息永远是最新的
基于 ShareInformer 创建的 Informer 指的是什么吗？
- 使用 ShareInformer 共享同一个 Reflector 、Indexer 和 Local Store
- 不同的只是 事件筛选逻辑（AddFunc、UpdateFunc、DeleteFunc)
SharedInformer的意思是指：对于属于同一类型的资源来说，他们将会共享同一个Informer和Reflector，其实现代码如下：
// 代码位置: client-go/informers/factory.go
type sharedInformerFactory struct {
client kubernetes.Interface
namespace string
tweakListOptions internalinterfaces.TweakListOptionsFunc
lock sync.Mutex
defaultResync time.Duration
customResync map[reflect.Type]time.Duration
informers map[reflect.Type]cache.SharedIndexInformer // sharedInformer的数据都存放在这个map中。
// startedInformers is used for tracking which informers have been started.
// This allows Start() to be called multiple times safely.
startedInformers map[reflect.Type]bool
}
2. Lister 和 Listener 都是什么
Lister —— 用户级使用函数
Lister 是一个方法，每个 Informer 实例都可以调用，用于从 Informer 的缓存（Local Store）中获取 资源对象（Object）
// 代码位置：clent-go/informers/core/v1/pod.go
// PodInformer 提供了与pod相关的shared informer及lister进行交互的途径
// 每个k8s resource都会有这样一个Informer，且Informer中都有以下两个方法：Informer(), Lister()
type PodInformer interface {
Informer() cache.SharedIndexInformer
Lister() v1.PodLister
}
type podInformer struct {
factory internalinterfaces.SharedInformerFactory
tweakListOptions internalinterfaces.TweakListOptionsFunc
namespace string
}
Listener —— 源码级框架函数
Listener 是代码中定义的一个名称（在代码中完整名称为 processorListener），控制器编程是并不会使用到此名称
从整体逻辑图中，可以看出其用来处理从 【DeltaFIFO 弹出的 Delta】，之后通过【本身的channel addCh】
传给【下个channel nextCh】（若【channel nextCh】处理不过来，先在【pendingNotifications环形缓存区进行暂存】，之后再取出）
【channel nextCh】会传给【handler的回调函数进行处理】
【handler的回调函数处理后】便进入了【workqueue】等待【用户定义的 Controller】进行下一步处理
我们最熟悉就是【processorListener】中的【handler的回调函数进行处理】，看下面代码便能明白，就是我们注册的【AddFunc、UpdateFunc、DeleteFunc 函数】
// k8s.io/client-go/tools/cache/shared_informer.go
type processorListener struct {
// nextCh：数据从此通道中读出并调用handler函数处理，非缓冲通道
nextCh chan interface{}
// addCh：FIFO中POP出的数据通过distribute函数放入此通道中，非缓冲通道
addCh chan interface{}
// 此处即为 Informer 的处理逻辑 （AddFunc、DeleteFunc、UpdateFunc）
handler ResourceEventHandler
// pendingNotifications：addCh中读出数据放入nextCh中，如果阻塞，则放入此缓冲区域（k8s自定义对象）
// 相当于缓存区
// pendingNotifications is an unbounded ring buffer that holds all notifications not yet distributed.
// There is one per listener, but a failing/stalled listener will have infinite pendingNotifications
// added until we OOM.
// TODO: This is no worse than before, since reflectors were backed by unbounded DeltaFIFOs, but
// we should try to do something better.
pendingNotifications buffer.RingGrowing
// requestedResyncPeriod is how frequently the listener wants a full resync from the shared informer
requestedResyncPeriod time.Duration
// resyncPeriod is how frequently the listener wants a full resync from the shared informer. This
// value may differ from requestedResyncPeriod if the shared informer adjusts it to align with the
// informer's overall resync check period.
resyncPeriod time.Duration
// nextResync is the earliest time the listener should get a full resync
nextResync time.Time
// resyncLock guards access to resyncPeriod and nextResync
resyncLock sync.Mutex
}
// k8s.io/client-go/tools/cache/controller.go
type ResourceEventHandler interface {
OnAdd(obj interface{})
OnUpdate(oldObj, newObj interface{})
OnDelete(obj interface{})
}
// ResourceEventHandlerFuncs is an adaptor to let you easily specify as many or
// as few of the notification functions as you want while still implementing
// ResourceEventHandler.
type ResourceEventHandlerFuncs struct {
AddFunc func(obj interface{})
UpdateFunc func(oldObj, newObj interface{})
DeleteFunc func(obj interface{})
}
// k8s.io/utils/buffer/ring_growing.go
// RingGrowing is a growing ring buffer.
// Not thread safe.
type RingGrowing struct {
data []interface{}
n int // Size of Data
beg int // First available element
readable int // Number of data items available
}
3. Indexer、Local Store 和 cache 是什么关系
4. Infomer 中的 controller 和用户定义的 Controller Loop有什么区别？口语化
- 普通常说的【Indexer】指的是【Informer中的索引模块】，【Local Store】指的是【Informer中的缓存模块】
代码框架级
- 【源码中存在一个 Indexer 接口】，其定义的方法包含着两类行为【对「索引」的操作、对「缓存」的操作】，感觉就是一体的，【Indexer 接口】的实现对应【 cache 结构体（注意小写为私有结构体）】，【索引和缓存的存储结构】对应着【 cache 下面的 threadSafeMap 结构体】
Infomer 中的 controller —— 源码的架构
相当于 Informer 中的总管家，协调着各个组件的关系
利用 Informer 传入的信息（DeltaFIFO、ListAndWatch 函数、handler回调函数等），将其他们形成 Reflector 组件、DeltaFIFO 组件、handler 等组件，组合成一个【执行框架】，处理事件的同时，还要保障更新【索引】和【缓存】
简言之，就是 Informer 运行之根本（称之为运行逻辑结构）
用户定义的 Controller Loop —— 用户逻辑
- 从 【workqueue】中取出事件，获取到【期望状态】，同时从现有环境中获取【现有状态】，之后进行【一系列操作】，使得【现有状态】达到【目标状态】
// k8s.io/client-go/tools/cache/shared_informer.go
func (s *sharedIndexInformer) Run(stopCh <-chan struct{}) {
defer utilruntime.HandleCrash()
// DeltaFIFO
fifo := NewDeltaFIFO(MetaNamespaceKeyFunc, s.indexer)
// 注意此处的 Config
cfg := &Config{
Queue: fifo, // DeltaFIFO
ListerWatcher: s.listerWatcher, // ListAndWatch 函数
ObjectType: s.objectType,
FullResyncPeriod: s.resyncCheckPeriod,
RetryOnError: false,
ShouldResync: s.processor.shouldResync,
// 重点!!!
// 处理 Deltas 的函数，也就是 handler（调用注册的 AddFunc、UpdateFunc、DeleteFunc）
// 同时负责同步 索引Indexer 和 缓存 Local Store
Process: s.HandleDeltas,
}
// 使用 Config 创建 共享Informer 内部的 controller
func() {
s.startedLock.Lock()
defer s.startedLock.Unlock()
// 创建 共享Informer 内部的 controller
s.controller = New(cfg)
s.controller.(*controller).clock = s.clock
s.started = true
}()
// Separate stop channel because Processor should be stopped strictly after controller
processorStopCh := make(chan struct{})
var wg wait.Group
defer wg.Wait() // Wait for Processor to stop
defer close(processorStopCh) // Tell Processor to stop
wg.StartWithChannel(processorStopCh, s.cacheMutationDetector.Run)
wg.StartWithChannel(processorStopCh, s.processor.run)
defer func() {
s.startedLock.Lock()
defer s.startedLock.Unlock()
s.stopped = true // Don't want any new listeners
}()
// Run 的时候会创建 Reflector
s.controller.Run(stopCh)
}
// k8s.io/client-go/tools/cache/controller.go
// 接口定义又哪些行为
// Controller is a generic controller framework.
type Controller interface {
Run(stopCh <-chan struct{})
HasSynced() bool
LastSyncResourceVersion() string
}
// controller 的具体实现
// Controller is a generic controller framework.
type controller struct {
config Config // 包含着 ListAndWatch 函数，DeltaFIFO
reflector *Reflector // Reflector ， 用于 ListAndWatch
reflectorMutex sync.RWMutex
clock clock.Clock
}
// Run 的时候会创建 Reflector
func (c *controller) Run(stopCh <-chan struct{}) {
defer utilruntime.HandleCrash()
go func() {
<-stopCh
c.config.Queue.Close()
}()
// Reflector 的构建依赖于 Config
r := NewReflector(
c.config.ListerWatcher, // ListAndWatch 函数
c.config.ObjectType,
c.config.Queue, // Delta FIFO
c.config.FullResyncPeriod,
)
r.ShouldResync = c.config.ShouldResync
r.clock = c.clock
c.reflectorMutex.Lock()
c.reflector = r // Reflector
c.reflectorMutex.Unlock()
var wg wait.Group
defer wg.Wait()
wg.StartWithChannel(stopCh, r.Run)
// processLoop 就是 HandleDeltas 函数
// 1. 用于更新索引Indexer和缓存 Loacl Store
// 2. 用于 AddEventHandler 注册的回调函数（AddFunc、UpdateFunc、DeleteFunc)的过滤及处理，然后放入 workqueue，供用户自定义的 Controller （这里指的是用户的业务逻辑）消费使用
wait.Until(c.processLoop, time.Second, stopCh)
}
// 启动 processLoop 不断从 DeltaFIFO Pop 进行消费
// c.config.Process 就是 HandleDeltas 函数，在 config 初始化可以看到
func (c *controller) processLoop() {
for {
obj, err := c.config.Queue.Pop(PopProcessFunc(c.config.Process))
if err != nil {
if err == ErrFIFOClosed {
return
}
if c.config.RetryOnError {
// This is the safe way to re-enqueue.
c.config.Queue.AddIfNotPresent(obj)
}
}
}
}
5. 一个 ShareInformer 对应多个 Informer 实例（包含着不同的 handler），如何处理？
【多个 Informer 实例】可理解为【1个共享Informer】+【多个具有不同事件处理逻辑 handler 的监听器 processorListener】
当 Informer 关注的资源变化时，会将【此资源的变化通知】告知给所有的【Informer实例】，也就是调用【processorListener的handler 】进行处理
// staging/src/k8s.io/client-go/tools/cache/shared_informer.go
func (p *sharedProcessor) distribute(obj interface{}, sync bool) {
p.listenersLock.RLock()
defer p.listenersLock.RUnlock()
if sync {
for _, listener := range p.syncingListeners {
listener.add(obj)
}
} else {
for _, listener := range p.listeners {
listener.add(obj)
}
}
}
