最近加入云原生社区组织的k8s源码研习社,开始学习k8s底层源码,并整理成笔记。欢迎感兴趣的同学一起加入,共同学习进步。群里和社区里有各种大佬,随时可以帮你答疑解惑。github.com/cloudnative…
Informer在k8s的重要性就不再赘述,直接切入正题。
由于Informer这部分的源码比较复杂,调用链路也很长,后面的源码分析,都会围绕这一张图展开。
k8s中,组件之间通过http通讯,在不依赖任何中间件的情况下,需要保证消息的可靠性、实时性、顺序性等?k8s是如何做到的呢?--- 答案就是Informer。k8s的其他组件都是通过informer与api-server通讯的。
各个组件包括:
// 通过informer机制,实现k8s资源的监控
func informer() {
// 因为informer是一个持久运行的groutine,channel作用:进程退出前通知informer退出
stopChan := make(chan struct{})
defer close(stopChan)
// 创建连接k8s的client对象
clientSet, err := kubernetes.NewForConfig(config)
if err != nil {
log.Printf("init clientset error.")
return
}
// 第一步:创建sharedInformer对象,第二个参数为重新同步数据的间隔时间
sharedInformers := informers.NewSharedInformerFactory(clientSet, time.Minute)
// 第二步:每个资源都有informer对象,这里获取pod资源的informer对象
podInformer := sharedInformers.Core().V1().Pods().Informer()
// 第三步:添加自定义回调函数
podInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
// 添加资源的回调函数,返回的是接口类型,需要强制转换为真正的类型
AddFunc: func(obj interface{}) {
mObj := obj.(v1.Object)
log.Printf("New pod added: %s", mObj.GetName())
},
// 更新资源的回调函数
UpdateFunc: func(oldObj, newObj interface{}) {
oObj := oldObj.(v1.Object)
nObj := newObj.(v1.Object)
log.Printf("%s pod updated to %s", oObj.GetName(), nObj.GetName())
},
// 删除资源的回调函数
DeleteFunc: func(obj interface{}) {
mObj := obj.(v1.Object)
log.Printf("pod deleted from store: %s", mObj.GetName())
},
})
// 第四步:开始运行informer对象
podInformer.Run(stopChan)
}
复制代码
前面demo中,第一步便是创建SharedInformer对象,下面先介绍一下Informer和SharedInformer
type PodInformer interface {
Informer() cache.SharedIndexInformer
Lister() v1.PodLister
}
若同一个资源的Informer被实例化了多次,每个Informer使用一个Reflector,那么会运行过多相同的ListAndWatch,太多重复的序列化和反序列化操作会导致api-server负载过重
SharedInformer可以使同一类资源Informer共享一个Reflector。内部定义了一个map字段,用于存放所有Infromer的字段。
前面demo中第一步创建SharedInformer,sharedInformers := informers.NewSharedInformerFactory(clientSet, time.Minute)
,内部初始化了一个sharedInformerFactory对象,先看下sharedInformerFactory
源码位置:vendor/k8s.io/client-go/informer/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
// 按照类型存放共享的informer
informers map[reflect.Type]cache.SharedIndexInformer
// 这个字段用来追踪informers是否被启动了
// 可以保证Start()方法安全的重复调用多次(幂等性)
startedInformers map[reflect.Type]bool
}
k8s中的Controller-Manager组件,源码中的Run方法调用了SharedInformerFactory的Start方法
源码位置:cmd/kube-controller-manager/app/controllermanager.go
func Run(c *config.CompletedConfig, stopCh <-chan struct{}) error {
...
controllerContext.InformerFactory.Start(controllerContext.Stop)
...
}
复制代码
源码位置:k8s.io/client-go/informers/factory.go
func (f *sharedInformerFactory) Start(stopCh <-chan struct{}) {
...
// 遍历所有的informers
for informerType, informer := range f.informers {
if !f.startedInformers[informerType] {
// 每一种informer启动一个协程,运行Run方法
go informer.Run(stopCh)
f.startedInformers[informerType] = true
}
}
}
复制代码
前面demo中,创建好sharedInformer对象后,第二步是调用podInformer := sharedInformers.Core().V1().Pods().Informer()
,获取具体的informer实例,下面开始分析Informer方法
关键逻辑包括:
以pod为例,源码位置:client-go/informers/core/v1/pod.go
// 获取pod的informer,内部调用InformerFor,参数需要传入f.defaultInformer
func (f *podInformer) Informer() cache.SharedIndexInformer {
return f.factory.InformerFor(&corev1.Pod{}, f.defaultInformer)
}
func (f *podInformer) defaultInformer(client kubernetes.Interface, resyncPeriod time.Duration) cache.SharedIndexInformer {
// 最后一个参数,初始化indexers
return NewFilteredPodInformer(client, f.namespace, resyncPeriod, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}, f.tweakListOptions)
}
func NewFilteredPodInformer(client kubernetes.Interface, namespace string, resyncPeriod time.Duration, indexers cache.Indexers, tweakListOptions internalinterfaces.TweakListOptionsFunc) cache.SharedIndexInformer {
return cache.NewSharedIndexInformer(
// 注册List、Watch方法
&cache.ListWatch{
ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
if tweakListOptions != nil {
tweakListOptions(&options)
}
// List方法是该种资源对象的List方法(这里是pod)
return client.CoreV1().Pods(namespace).List(context.TODO(), options)
},
WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
if tweakListOptions != nil {
tweakListOptions(&options)
}
// Watch方法是该种资源对象的Watch方法(这里是pod)
return client.CoreV1().Pods(namespace).Watch(context.TODO(), options)
},
},
&corev1.Pod{},
resyncPeriod,
indexers,
)
}
func NewSharedIndexInformer(lw ListerWatcher, exampleObject runtime.Object, defaultEventHandlerResyncPeriod time.Duration, indexers Indexers) SharedIndexInformer {
realClock := &clock.RealClock{}
sharedIndexInformer := &sharedIndexInformer{
// 这里是processor的初始化
processor: &sharedProcessor{clock: realClock},
// 这里是Indexer的初始化,接口为Indexer,实现类为cache
indexer: NewIndexer(DeletionHandlingMetaNamespaceKeyFunc, indexers),
listerWatcher: lw,
objectType: exampleObject,
resyncCheckPeriod: defaultEventHandlerResyncPeriod,
defaultEventHandlerResyncPeriod: defaultEventHandlerResyncPeriod,
cacheMutationDetector: NewCacheMutationDetector(fmt.Sprintf("%T", exampleObject)),
clock: realClock,
}
return sharedIndexInformer
}
// Index接口,实现类是cache类
func NewIndexer(keyFunc KeyFunc, indexers Indexers) Indexer {
return &cache{
cacheStorage: NewThreadSafeStore(indexers, Indices{}),
keyFunc: keyFunc,
}
}
复制代码
得到Informer对象之后,第三步是给该infomer注册自定义回调函数,当k8s的资源发送变更时,可以实现自己的业务逻辑。 下面分析一下podInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{...})
的逻辑
// 开始注册事件处理函数
func (s *sharedIndexInformer) AddEventHandler(handler ResourceEventHandler) {
s.AddEventHandlerWithResyncPeriod(handler, s.defaultEventHandlerResyncPeriod)
}
func (s *sharedIndexInformer) AddEventHandlerWithResyncPeriod(handler ResourceEventHandler, resyncPeriod time.Duration) {
...
// 每一个监听者,都会注册为一个listner实例
// 每个listener中持有一个handler对象,后面事件发生时,框架会调用handler方法,也就走到用户注册的代码逻辑了
listener := newProcessListener(handler, resyncPeriod, determineResyncPeriod(resyncPeriod, s.resyncCheckPeriod), s.clock.Now(), initialBufferSize)
...
// 将listner添加到sharedProcessor中
s.processor.addListener(listener)
for _, item := range s.indexer.List() {
listener.add(addNotification{newObj: item})
}
}
// 将listner添加到sharedProcessor中
func (p *sharedProcessor) addListener(listener *processorListener) {
...
if p.listenersStarted {
// listener后台启动了两个协程,这两个协程很关键,后面会介绍
p.wg.Start(listener.run)
p.wg.Start(listener.pop)
}
}
复制代码
前面demo中,得到informer实例之后,最后一步就是调用Run方法,下面开始解析Run方法的逻辑, 核心逻辑包括:
源码位置:k8s.io/client-go/tools/cache/shared_informer.go
func (s *sharedIndexInformer) Run(stopCh <-chan struct{}) {
// 初始化DeltaFIFO
fifo := NewDeltaFIFOWithOptions(DeltaFIFOOptions{
KnownObjects: s.indexer,
EmitDeltaTypeReplaced: true,
})
// config初始化
// 这里重点关注ListerWatcher对象和Process对象,Process关联的是HandleDeltas函数
// HandleDeltas是消费增量信息(Delta对象)的核心方法
cfg := &Config{
Queue: fifo,
// ListAndWatch对象
ListerWatcher: s.listerWatcher,
ObjectType: s.objectType,
FullResyncPeriod: s.resyncCheckPeriod,
RetryOnError: false,
ShouldResync: s.processor.shouldResync,
// 注册回调函数 HandleDeltas,资源变更时,存到到本地Indexer
Process: s.HandleDeltas,
WatchErrorHandler: s.watchErrorHandler,
}
// 这里主要是controller的初始化
func() {
s.startedLock.Lock()
defer s.startedLock.Unlock()
// 初始化Controller对象
s.controller = New(cfg)
s.controller.(*controller).clock = s.clock
s.started = true
}()
// s.cacheMutationDetector.Run检查缓存对象是否存在
wg.StartWithChannel(processorStopCh, s.cacheMutationDetector.Run)
// 执行sharedProcessor.run方法
// 这个方法非常重要
wg.StartWithChannel(processorStopCh, s.processor.run)
...
// 调用Controller的Run方法
s.controller.Run(stopCh)
}
复制代码
func (p *sharedProcessor) run(stopCh <-chan struct{}) {
func() {
...
// sharedProcessor的所有Listner,每个后台启动两个协程
// 分别指向run和pop方法
for _, listener := range p.listeners {
p.wg.Start(listener.run)
p.wg.Start(listener.pop)
}
p.listenersStarted = true
}()
...
}
复制代码
processorListener代表一个消费者对象,该函数周期性执行,主要是从自己的nextCh通道中获取从api-server得到的增量信息,然后调用handler的相关方法,handler方法就是前面用户自定义传进来的方法。
这里我们只需要知道,run方法是消费者方法,负责消费事件。后面会介绍到谁是生产者,谁往processorLister的nextCh通道中放入增量信息(其实就是下面的pop方法)
func (p *processorListener) run() {
stopCh := make(chan struct{})
wait.Until(func() {
// 消费者方法,不断从通道中获取事件
for next := range p.nextCh {
switch notification := next.(type) {
case updateNotification:
// 调用handler的方法,
p.handler.OnUpdate(notification.oldObj, notification.newObj)
case addNotification:
p.handler.OnAdd(notification.newObj)
case deleteNotification:
p.handler.OnDelete(notification.oldObj)
default:
utilruntime.HandleError(fmt.Errorf("unrecognized notification: %T", next))
}
}
// the only way to get here is if the p.nextCh is empty and closed
close(stopCh)
}, 1*time.Second, stopCh)
}
// OnUpdate方法,内部就是调用demo中注册的UpdateFunc方法
// 其他方法类似
func (r ResourceEventHandlerFuncs) OnUpdate(oldObj, newObj interface{}) {
if r.UpdateFunc != nil {
r.UpdateFunc(oldObj, newObj)
}
}
复制代码
这个方法的实现非常复杂,但是总体的目的很简单:就是往nextCh中生产消息,然后前面介绍的run方法就可以消费到
这里主要用到了类似缓冲区的方法,并不是一次就获取一个事件
func (p *processorListener) pop() {
...
var nextCh chan<- interface{}
var notification interface{}
for {
select {
// 函数第一次进来,这个notification一定是空的,这个case会被阻塞住
// 当第二个case调用完成后,notification被赋值为notificationToAdd,才会进入到这里
case nextCh <- notification:
// Notification dispatched
var ok bool
notification, ok = p.pendingNotifications.ReadOne()
if !ok { // Nothing to pop
nextCh = nil // Disable this select case
}
// 第一次调用时,会进入这里,首先从addCh中获取数据(后面会介绍谁往addCh中放数据)
case notificationToAdd, ok := <-p.addCh:
if !ok {
return
}
if notification == nil { // No notification to pop (and pendingNotifications is empty)
notification = notificationToAdd
// channel是引用类型,将p.nextCh指向nextCh,对nextCh的操作就是操作p.nextCh
// 这里也解答了前面的run方法里面提到的疑问,谁是nextCh的生产者,往nextCh放入数据
nextCh = p.nextCh
} else { // There is already a notification waiting to be dispatched
p.pendingNotifications.WriteOne(notificationToAdd)
}
}
}
}
复制代码
pop函数中我们看到,主要是消费p.addCh,后面详细介绍p.addCh的生产者是谁。这里简单提一下,watch函数监测到api-server的事件变化时,触发HandlerDelta函数,这个函数会更新Indexer,同时调用distribute方法,将该事件通知给所有的listener,内部实现就是:往每个listener的addCh这个通道中放入数据。
Run方法内部关键逻辑包括:
源码位置:k8s.io/client-go/tools/cache/controller.go
func (c *controller) Run(stopCh <-chan struct{}) {
// 调用NewReflector初始化一个Reflector
// 必须传入ListWatcher数据接口对象
r := NewReflector(
c.config.ListerWatcher,
c.config.ObjectType,
c.config.Queue,
c.config.FullResyncPeriod,
)
...
// 调用Reflector的Run方法,启动监控,并处理监控事件
wg.StartWithChannel(stopCh, r.Run)
// processLoop负责从DeltaFIFO取出数据并消费
wait.Until(c.processLoop, time.Second, stopCh)
}
复制代码
这里先暂停源码分析,简单介绍一下Reflector,Reflector用于监控指定的k8s资源,并触发相应的变更事件。
Run中的核心函数是ListAndWatch,流程包括:
type Reflector struct {
// 名称
name string
expectedTypeName string
// 期望放入缓存store的资源类型
expectedType reflect.Type
// The GVK of the object we expect to place in the store if unstructured.
expectedGVK *schema.GroupVersionKind
// 存放同步监听到的资源,这里是DeltaFIFO类
store Store
// 用来执行List和Watch的对象
listerWatcher ListerWatcher
backoffManager wait.BackoffManager
// resync周期
resyncPeriod time.Duration
ShouldResync func() bool
clock clock.Clock
paginatedResult bool
// 最新一次看到的资源版本号
lastSyncResourceVersion string
isLastSyncResourceVersionUnavailable bool
lastSyncResourceVersionMutex sync.RWMutex
WatchListPageSize int64
watchErrorHandler WatchErrorHandler
}
复制代码
核心逻辑包括:
// Run函数
func (r *Reflector) Run(stopCh <-chan struct{}) {
...
wait.BackoffUntil(func() {
// 核心函数:ListAndWatch
if err := r.ListAndWatch(stopCh); err != nil {
r.watchErrorHandler(r, err)
}
}, r.backoffManager, true, stopCh)
...
}
复制代码
// ListAndWatch函数
func (r *Reflector) ListAndWatch(stopCh <-chan struct{}) error {
...
if err := func() error {
...
go func() {
...
pager := pager.New(pager.SimplePageFunc(func(opts metav1.ListOptions) (runtime.Object, error) {
// 调用List方法获取资源对象下所有的数据
return r.listerWatcher.List(opts)
}))
...
}()
...
// 获取资源版本号
resourceVersion = listMetaInterface.GetResourceVersion()
initTrace.Step("Resource version extracted")
// 将资源数据转换为资源对象列表
items, err := meta.ExtractList(list)
// 将资源信息存储到DeltaFIFO中,全量替换本地缓存
// 内部调用了replace方法
if err := r.syncWith(items, resourceVersion); err != nil {
return fmt.Errorf("unable to sync list result: %v", err)
}
// 设置最新的资源版本号
r.setLastSyncResourceVersion(resourceVersion)
return nil
}(); err != nil {
return err
}
go func() {
...
for {
...
// 同步资源
if r.ShouldResync == nil || r.ShouldResync() {
// 调用DeltaFIFO的Resync方法
if err := r.store.Resync(); err != nil {
...
}
}
resyncCh, cleanup = r.resyncChan()
}
}()
for {
...
// 监听资源
w, err := r.listerWatcher.Watch(options)
// 处理handler事件,用户注册的Add,Delete,Update函数
if err := r.watchHandler(start, w, &resourceVersion, resyncerrc, stopCh);
...
}
}
// syncWith函数
func (r *Reflector) syncWith(items []runtime.Object, resourceVersion string) error {
found := make([]interface{}, 0, len(items))
for _, item := range items {
found = append(found, item)
}
// 调用cache.Replace
return r.store.Replace(found, resourceVersion)
}
// cache.Replace
func (c *cache) Replace(list []interface{}, resourceVersion string) error {
items := make(map[string]interface{}, len(list))
for _, item := range list {
key, err := c.keyFunc(item)
if err != nil {
return KeyError{item, err}
}
items[key] = item
}
c.cacheStorage.Replace(items, resourceVersion)
return nil
}
复制代码
watchHandler函数,处理watch到的各种事件,所有的事件都存放在ResultChan中,包括:事件类型,资源对象
func (r *Reflector) watchHandler(start time.Time, w watch.Interface, resourceVersion *string, errc chan error, stopCh <-chan struct{}) error {
...
for {
select {
...
// 获取watch接口中的事件的channel
case event, ok := <-w.ResultChan():
...
switch event.Type {
// 处理Add函数
case watch.Added:
// store是DeltaFIFO类
err := r.store.Add(event.Object)
// 处理Modified函数
case watch.Modified:
err := r.store.Update(event.Object)
// 处理Deleted函数
case watch.Deleted:
err := r.store.Delete(event.Object)
}
*resourceVersion = newResourceVersion
// 设置资源版本
r.setLastSyncResourceVersion(newResourceVersion)
eventCount++
}
}
...
}
复制代码
后面的源码分析会用到DeltaFIFO,这里先介绍一下。
DeltaFIFO用来存储Watch API返回的各种事件,队列中会存在拥有不同操作类型的同一个资源对象 DeltaFIFO实现了Queue接口,Queue继承了Store接口
源码路径:vendor/k8s.io/client-go/tools/cache/delta_fifo.go
type DeltaFIFO struct {
// lock/cond protects access to 'items' and 'queue'.
lock sync.RWMutex
cond sync.Cond
// We depend on the property that items in the set are in
// the queue and vice versa, and that all Deltas in this
// map have at least one Delta.
// map结构存储:key是资源对象的key,value是对象的Deltas数组
items map[string]Deltas
// 存储资源对象的key
queue []string
// populated is true if the first batch of items inserted by Replace() has been populated
// or Delete/Add/Update was called first.
populated bool
// initialPopulationCount is the number of items inserted by the first call of Replace()
initialPopulationCount int
// keyFunc is used to make the key used for queued item
// insertion and retrieval, and should be deterministic.
keyFunc KeyFunc
// Index本地存储对象
knownObjects KeyListerGetter
closed bool
closedLock sync.Mutex
}
核心功能包括:
Reflector监听到资源变化后,将Add、Delete、Update等资源变更信息加入到DeltaFIFO。也就是队列的生产者,方法如下,内部都调用了
入口函数是 r.store.Add(event.Object),在前面的watchHandler介绍过
func (f *DeltaFIFO) Add(obj interface{}) error {
...
return f.queueActionLocked(Added, obj)
}
func (f *DeltaFIFO) Update(obj interface{}) error {
...
return f.queueActionLocked(Updated, obj)
}
func (f *DeltaFIFO) Delete(obj interface{}) error {
...
return f.queueActionLocked(Deleted, obj)
}
// 举例说明其中一个处理函数,其他的类似,内部都调用queueActionLocked
func (f *DeltaFIFO) Update(obj interface{}) error {
f.lock.Lock()
defer f.lock.Unlock()
f.populated = true
// 内部调用queueActionLocked
return f.queueActionLocked(Updated, obj)
}
复制代码
// 内部主要是封装Delta事件,并加入队列,供消费者消费(HandleDeltas函数)
func (f *DeltaFIFO) queueActionLocked(actionType DeltaType, obj interface{}) error {
// 根据资源对象,得到key,一般是 namespace/name 格式
id, err := f.KeyOf(obj)
// 将watch到的事件类型和资源对象,封装成Delta对象
newDeltas := append(f.items[id], Delta{actionType, obj})
// 去重操作
newDeltas = dedupDeltas(newDeltas)
// 将Delta对象加入到队列中
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)
}
return nil
}
复制代码
前面的源码分析,分析完了Reflector的Run方法,下一步便是controller的processLoop方法
func (c *controller) Run(stopCh <-chan struct{}) {
// 调用NewReflector初始化一个Reflector
// 必须传入ListWatcher数据接口对象
r := NewReflector(
c.config.ListerWatcher,
c.config.ObjectType,
c.config.Queue,
c.config.FullResyncPeriod,
)
...
// 调用Reflector的Run方法,启动监控,并处理监控事件
wg.StartWithChannel(stopCh, r.Run)
// processLoop负责从DeltaFIFO取出数据并消费
wait.Until(c.processLoop, time.Second, stopCh)
}
复制代码
func (c *controller) processLoop() {
for {
// 从DeltaFIFO队列中取出数据,并交给process处理
// process函数保存在config.Process中,也就是前面传入的 HandleDeltas
obj, err := c.config.Queue.Pop(PopProcessFunc(c.config.Process))
...
}
}
复制代码
DeltaFIFO的消费方法为Pop,该函数需要传入process函数,用于接收并处理对象的回调方法
func (f *DeltaFIFO) Pop(process PopProcessFunc) (interface{}, error) {
for {
for len(f.queue) == 0 {
// 当队列为空时,Pop函数阻塞住,知道新的数据入队列才唤醒
// 如果Close函数被调用,closed状态被设置,并且广播
if f.closed {
return nil, ErrFIFOClosed
}
f.cond.Wait()
}
// 走到这里,说明队列中有数据,取出数据
id := f.queue[0]
...
// 将数据交给上层回调函数处理
err := process(item)
// 出错则将数据重新放入队列
if e, ok := err.(ErrRequeue); ok {
f.addIfNotPresent(id, item)
err = e.Err
}
return item, err
}
}
复制代码
核心逻辑包括:
func (s *sharedIndexInformer) HandleDeltas(obj interface{}) error {
// 获取所有的Delta资源
for _, d := range obj.(Deltas) {
// 判断资源类型
switch d.Type {
// 如果是下列类型,将资源存储到Indexer
case Sync, Replaced, Added, Updated:
s.cacheMutationDetector.AddObject(d.Object)
if old, exists, err := s.indexer.Get(d.Object); err == nil && exists {
// indexer的实现类是前面介绍过的cache
if err := s.indexer.Update(d.Object); err != nil {
return err
}
isSync := false
switch {
case d.Type == Sync:
isSync = true
case d.Type == Replaced:
if accessor, err := meta.Accessor(d.Object); err == nil {
if oldAccessor, err := meta.Accessor(old); err == nil {
isSync = accessor.GetResourceVersion() == oldAccessor.GetResourceVersion()
}
}
}
// 将资源对象分发至 SharedInformer 的事件处理函数中
s.processor.distribute(updateNotification{oldObj: old, newObj: d.Object}, isSync)
} else {
if err := s.indexer.Add(d.Object); err != nil {
return err
}
s.processor.distribute(addNotification{newObj: d.Object}, false)
}
case Deleted:
if err := s.indexer.Delete(d.Object); err != nil {
return err
}
s.processor.distribute(deleteNotification{oldObj: d.Object}, false)
}
}
return nil
}
// 更新本地缓存cacheStorage
// 其实就是更新 threadSafeMap 这个数据结构,threadSafeMap的初始化在前面介绍过
func (c *cache) Update(obj interface{}) error {
key, err := c.keyFunc(obj)
if err != nil {
return KeyError{obj, err}
}
c.cacheStorage.Update(key, obj)
return nil
}
// distribute函数
func (p *sharedProcessor) distribute(obj interface{}, sync bool) {
...
if sync {
for _, listener := range p.syncingListeners {
// 核心方法
listener.add(obj)
}
} else {
for _, listener := range p.listeners {
listener.add(obj)
}
}
}
// listener.add方法,这里将事件添加到listener的addCh通道中
// 至此,也回答的前面的问题-- 谁往p.addCh中生产数据
func (p *processorListener) add(notification interface{}) {
// 不同更新类型的对象加入到channel中
// 供给processorListener的Run方法使用
p.addCh <- notification
}
复制代码
ListAndWatch方法中的三步:
RSync负责将Indexer本地存储的资源对象同步到DeltaFIFO中,并设置资源类型为Sync类型。在Reflector中定时执行
func (f *DeltaFIFO) Resync() error {
// 获取indexer本地存储对象
keys := f.knownObjects.ListKeys()
for _, k := range keys {
if err := f.syncKeyLocked(k); err != nil {
return err
}
}
return nil
}
复制代码
前面的分析中说到,资源的变更会保存到本地的Indexer,这里介绍一下。
源码位置:k8s.io/client-go/tools/cache/index.go
// 存储缓存数据
// type Empty struct{}
// type String map[string]Empty
// 这里的sets.String是用map模拟set,map中的value都是空结构体
type Index map[string]sets.String
// 索引器函数,接收资源对象,返回检索结果列表
type IndexFunc func(obj interface{}) ([]string, error)
// 存储索引器,key为索引器名称,value为索引器实现函数
type Indexers map[string]IndexFunc
// 存储缓存器,key为缓存器名称,value为缓存数据
type Indices map[string]Index
复制代码
Indexer基于ThreadSafeMap做了封装,先看一下ThreadSafeMap
源码位置:k8s.io/tools/cache/thread_safe_store.go
type threadSafeMap struct {
lock sync.RWMutex
// map结构存储资源数据
// map中的key是通过keyFunc函数计算得出,默认使用MetaNamespaceFunc函数
// 该函数根据资源对象计算出/格式的key
// value是Delta对象,包括Type和Object资源对象
items map[string]interface{}
// 存储索引器,key为索引器名称,value为索引器实现函数
indexers Indexers
// 存储缓存器,key为缓存器名称,value为缓存的资源对象数据
indices Indices
}
// 通过执行索引器函数得到索引结果
// 需要两个参数:索引器名称、需要检索的key
func (c *threadSafeMap) ByIndex(indexName, indexedValue string) ([]interface{}, error) {
// 查找指定的索引器函数
indexFunc := c.indexers[indexName]
if indexFunc == nil {
return nil, fmt.Errorf("Index with name %s does not exist", indexName)
}
// 查找指定的缓存器函数
index := c.indices[indexName]
// 从缓存数据中查找并返回数据
set := index[indexedValue]
list := make([]interface{}, 0, set.Len())
for key := range set {
list = append(list, c.items[key])
}
return list, nil
}
复制代码
Informer机制在k8s中占据重要的角色,它的源码也是非常的复杂。学习的过程中一定要配合文章开始的那个图,否则很容易就绕进去了。里面使用Queue和Channel来解耦各个组件。个人的一个心得是:围绕着一个核心思路,可能分析起来逻辑会清晰一点,那就是:谁往channel放了数据,谁又从channel取了数据。