浅析kubernetes中client-go Informer

之前了解了client-go中的架构设计，也就是 tools/cache 下面的一些概念，那么下面将对informer进行分析

Controller

在client-go informer架构中存在一个 controller ，这个不是 Kubernetes 中的Controller组件；而是在 tools/cache 中的一个概念，controller 位于 informer 之下，Reflector 之上。code

Config

从严格意义上来讲，controller 是作为一个 sharedInformer 使用，通过接受一个 Config ，而 Reflector 则作为 controller 的 slot。Config 则包含了这个 controller 里所有的设置。

type Config struct {

	Queue // DeltaFIFO

	ListerWatcher // 用于list watch的

	Process ProcessFunc // 定义如何从DeltaFIFO中弹出数据后处理的操作

	ObjectType runtime.Object // Controller处理的对象数据，实际上就是kubernetes中的资源

	FullResyncPeriod time.Duration // 全量同步的周期

	ShouldResync ShouldResyncFunc // Reflector通过该标记来确定是否应该重新同步

	RetryOnError bool

}

controller

然后 controller 又为 reflertor 的上层

type controller struct {

	config         Config

	reflector      *Reflector

	reflectorMutex sync.RWMutex

	clock          clock.Clock

}

type Controller interface {

	// controller 主要做两件事，

    // 1. 构建并运行 Reflector,将listerwacther中的泵压到queue（Delta fifo）中

    // 2. Queue用Pop()弹出数据，具体的操作是Process

    // 直到 stopCh 不阻塞，这两个协程将退出

	Run(stopCh <-chan struct{})

	HasSynced() bool // 这个实际上是从store中继承的，标记这个controller已经

	LastSyncResourceVersion() string

}

controller 中的方法，仅有一个 Run() 和 New()；这意味着，controller 只是一个抽象的概念，作为 Reflector, Delta FIFO 整合的工作流

而 controller 则是 SharedInformer 了。

Queue

这里的 queue 可以理解为是一个具有 Pop() 功能的 Indexer ;而 Pop() 的功能则是 controller 中的一部分；也就是说 queue 是一个扩展的 Store ， Store 是不具备弹出功能的。

type Queue interface {

	Store

	// Pop会阻塞等待，直到有内容弹出，删除对应的值并处理计数器

	Pop(PopProcessFunc) (interface{}, error)

	// AddIfNotPresent puts the given accumulator into the Queue (in

	// association with the accumulator's key) if and only if that key

	// is not already associated with a non-empty accumulator.

	AddIfNotPresent(interface{}) error

	// HasSynced returns true if the first batch of keys have all been

	// popped.  The first batch of keys are those of the first Replace

	// operation if that happened before any Add, Update, or Delete;

	// otherwise the first batch is empty.

	HasSynced() bool

	Close() // 关闭queue

}

而弹出的操作是通过 controller 中的 processLoop() 进行的，最终走到Delta FIFO中进行处理。

通过忙等待去读取要弹出的数据，然后在弹出前通过PopProcessFunc 进行处理

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)

			}

		}

	}

}

DeltaFIFO.Pop()

func (f *DeltaFIFO) Pop(process PopProcessFunc) (interface{}, error) {

	f.lock.Lock()

	defer f.lock.Unlock()

	for {

		for len(f.queue) == 0 {

			// When the queue is empty, invocation of Pop() is blocked until new item is enqueued.

			// When Close() is called, the f.closed is set and the condition is broadcasted.

			// Which causes this loop to continue and return from the Pop().

			if f.IsClosed() {

				return nil, ErrFIFOClosed

			}

			f.cond.Wait()

		}

		id := f.queue[0]

		f.queue = f.queue[1:]

		if f.initialPopulationCount > 0 {

			f.initialPopulationCount--

		}

		item, ok := f.items[id]

		if !ok {

			// Item may have been deleted subsequently.

			continue

		}

		delete(f.items, id)

		err := process(item) // 进行处理

		if e, ok := err.(ErrRequeue); ok {

			f.addIfNotPresent(id, item) // 如果失败，再重新加入到队列中

			err = e.Err

		}

		// Don't need to copyDeltas here, because we're transferring

		// ownership to the caller.

		return item, err

	}

}

Informer

通过对 Reflector, Store, Queue, ListerWatcher、ProcessFunc, 等的概念，发现由 controller 所包装的起的功能并不能完成通过对API的动作监听，并通过动作来处理本地缓存的一个能力；这个情况下诞生了 informer 严格意义上来讲是 sharedInformer

func newInformer(

	lw ListerWatcher,

	objType runtime.Object,

	resyncPeriod time.Duration,

	h ResourceEventHandler,

	clientState Store,

) Controller {

	// This will hold incoming changes. Note how we pass clientState in as a

	// KeyLister, that way resync operations will result in the correct set

	// of update/delete deltas.

	fifo := NewDeltaFIFOWithOptions(DeltaFIFOOptions{

		KnownObjects:          clientState,

		EmitDeltaTypeReplaced: true,

	})

	cfg := &Config{

		Queue:            fifo,

		ListerWatcher:    lw,

		ObjectType:       objType,

		FullResyncPeriod: resyncPeriod,

		RetryOnError:     false,

		Process: func(obj interface{}) error {

			// from oldest to newest

			for _, d := range obj.(Deltas) {

				switch d.Type {

				case Sync, Replaced, Added, Updated:

					if old, exists, err := clientState.Get(d.Object); err == nil && exists {

						if err := clientState.Update(d.Object); err != nil {

							return err

						}

						h.OnUpdate(old, d.Object)

					} else {

						if err := clientState.Add(d.Object); err != nil {

							return err

						}

						h.OnAdd(d.Object)

					}

				case Deleted:

					if err := clientState.Delete(d.Object); err != nil {

						return err

					}

					h.OnDelete(d.Object)

				}

			}

			return nil

		},

	}

	return New(cfg)

}

newInformer是位于 tools/cache/controller.go 下，可以看出，这里面并没有informer的概念，这里通过注释可以看到，newInformer实际上是一个提供了存储和事件通知的informer。他关联的 queue 则是 Delta FIFO，并包含了 ProcessFunc, Store 等 controller的概念。最终对外的方法为 NewInformer()

func NewInformer(

	lw ListerWatcher,

	objType runtime.Object,

	resyncPeriod time.Duration,

	h ResourceEventHandler,

) (Store, Controller) {

	// This will hold the client state, as we know it.

	clientState := NewStore(DeletionHandlingMetaNamespaceKeyFunc)

	return clientState, newInformer(lw, objType, resyncPeriod, h, clientState)

}

type ResourceEventHandler interface {

	OnAdd(obj interface{})

	OnUpdate(oldObj, newObj interface{})

	OnDelete(obj interface{})

}

可以看到 NewInformer() 就是一个带有 Store功能的controller，通过这些可以假定出，Informer 就是controller ，将queue中相关操作分发给不同事件处理的功能

SharedIndexInformer

shareInformer 为客户端提供了与apiserver一致的数据对象本地缓存，并支持多事件处理程序的informer，而 shareIndexInformer 则是对shareInformer 的扩展

type SharedInformer interface {

	// AddEventHandler adds an event handler to the shared informer using the shared informer's resync

	// period.  Events to a single handler are delivered sequentially, but there is no coordination

	// between different handlers.

	AddEventHandler(handler ResourceEventHandler)

	// AddEventHandlerWithResyncPeriod adds an event handler to the

	// shared informer with the requested resync period; zero means

	// this handler does not care about resyncs.  The resync operation

	// consists of delivering to the handler an update notification

	// for every object in the informer's local cache; it does not add

	// any interactions with the authoritative storage.  Some

	// informers do no resyncs at all, not even for handlers added

	// with a non-zero resyncPeriod.  For an informer that does

	// resyncs, and for each handler that requests resyncs, that

	// informer develops a nominal resync period that is no shorter

	// than the requested period but may be longer.  The actual time

	// between any two resyncs may be longer than the nominal period

	// because the implementation takes time to do work and there may

	// be competing load and scheduling noise.

	AddEventHandlerWithResyncPeriod(handler ResourceEventHandler, resyncPeriod time.Duration)

	// GetStore returns the informer's local cache as a Store.

	GetStore() Store

	// GetController is deprecated, it does nothing useful

	GetController() Controller

	// Run starts and runs the shared informer, returning after it stops.

	// The informer will be stopped when stopCh is closed.

	Run(stopCh <-chan struct{})

	// HasSynced returns true if the shared informer's store has been

	// informed by at least one full LIST of the authoritative state

	// of the informer's object collection.  This is unrelated to "resync".

	HasSynced() bool

	// LastSyncResourceVersion is the resource version observed when last synced with the underlying

	// store. The value returned is not synchronized with access to the underlying store and is not

	// thread-safe.

	LastSyncResourceVersion() string

}

SharedIndexInformer 是对SharedInformer的实现，可以从结构中看出，SharedIndexInformer 大致具有如下功能：

索引本地缓存
controller，通过list watch拉取API并推入 Deltal FIFO
事件的处理

type sharedIndexInformer struct {

	indexer    Indexer // 具有索引的本地缓存

	controller Controller // controller

	processor             *sharedProcessor // 事件处理函数集合

	cacheMutationDetector MutationDetector

	listerWatcher ListerWatcher

	objectType runtime.Object

	resyncCheckPeriod time.Duration

	defaultEventHandlerResyncPeriod time.Duration

	clock clock.Clock

	started, stopped bool

	startedLock      sync.Mutex

	blockDeltas sync.Mutex

}

而在 tools/cache/share_informer.go 可以看到 shareIndexInformer 的运行过程

func (s *sharedIndexInformer) Run(stopCh <-chan struct{}) {

	defer utilruntime.HandleCrash()

	fifo := NewDeltaFIFOWithOptions(DeltaFIFOOptions{

		KnownObjects:          s.indexer,

		EmitDeltaTypeReplaced: true,

	})

	cfg := &Config{

		Queue:            fifo,

		ListerWatcher:    s.listerWatcher,

		ObjectType:       s.objectType,

		FullResyncPeriod: s.resyncCheckPeriod,

		RetryOnError:     false,

		ShouldResync:     s.processor.shouldResync,

		Process: s.HandleDeltas, // process 弹出时操作的流程

	}

	func() {

		s.startedLock.Lock()

		defer s.startedLock.Unlock()

		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

	}()

    s.controller.Run(stopCh) // 启动controller，controller会启动Reflector和fifo的Pop()

}

而在操作Delta FIFO中可以看到，做具体操作时，会将动作分发至对应的事件处理函数中，这个是informer初始化时对事件操作的函数

func (s *sharedIndexInformer) HandleDeltas(obj interface{}) error {

	s.blockDeltas.Lock()

	defer s.blockDeltas.Unlock()

	for _, d := range obj.(Deltas) {

		switch d.Type {

		case Sync, Replaced, Added, Updated:

			s.cacheMutationDetector.AddObject(d.Object)

			if old, exists, err := s.indexer.Get(d.Object); err == nil && exists {

				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()

						}

					}

				}

                // 事件的分发

				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

}

事件处理函数 processor

启动informer时也会启动注册进来的事件处理函数；processor 就是这个事件处理函数。

run() 函数会启动两个 listener，j监听事件处理业务函数 listener.run 和事件的处理

wg.StartWithChannel(processorStopCh, s.processor.run)

func (p *sharedProcessor) run(stopCh <-chan struct{}) {

	func() {

		p.listenersLock.RLock()

		defer p.listenersLock.RUnlock()

		for _, listener := range p.listeners {

			p.wg.Start(listener.run)

			p.wg.Start(listener.pop)

		}

		p.listenersStarted = true

	}()

	<-stopCh

	p.listenersLock.RLock()

	defer p.listenersLock.RUnlock()

	for _, listener := range p.listeners {

		close(listener.addCh) // Tell .pop() to stop. .pop() will tell .run() to stop

	}

	p.wg.Wait() // Wait for all .pop() and .run() to stop

}

可以看出，就是拿到的事件，根据注册的到informer的事件函数进行处理

func (p *processorListener) run() {

	stopCh := make(chan struct{})

	wait.Until(func() {

		for next := range p.nextCh { // 消费事件

			switch notification := next.(type) {

			case updateNotification:

				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)

}

informer中的事件的设计

了解了informer如何处理事件，就需要学习下，informer的事件系统设计 prossorListener

事件的添加

当在handleDelta时，会分发具体的事件

// 事件的分发

s.processor.distribute(updateNotification{oldObj: old, newObj: d.Object}, isSync)

此时，事件泵 Pop() 会根据接收到的事件进行处理

// run() 时会启动一个事件泵

p.wg.Start(listener.pop)

func (p *processorListener) pop() {

	defer utilruntime.HandleCrash()

	defer close(p.nextCh) 

	var nextCh chan<- interface{}

	var notification interface{}

	for {

		select {

        case nextCh <- notification: // 这里实际上是一个阻塞的等待

            // 单向channel 可能不会走到这步骤

			var ok bool

            // deltahandle 中 distribute 会将事件添加到addCh待处理事件中

            // 处理完事件会再次拿到一个事件

			notification, ok = p.pendingNotifications.ReadOne()

			if !ok { // Nothing to pop

				nextCh = nil // Disable this select case

			}

        // 处理 分发过来的事件 addCh

		case notificationToAdd, ok := <-p.addCh: // distribute分发的事件

			if !ok {

				return

			}

            // 这里代表第一次，没有任何事件时，或者上面步骤完成读取

			if notification == nil { // 就会走这里

				notification = notificationToAdd

				nextCh = p.nextCh

			} else {

                // notification否则代表没有处理完，将数据再次添加到待处理中

				p.pendingNotifications.WriteOne(notificationToAdd)

			}

		}

	}

}

该消息事件的流程图为

通过一个简单实例来学习client-go中的消息通知机制

package main

import (

	"fmt"

	"time"

	"k8s.io/utils/buffer"

)

var nextCh1 = make(chan interface{})

var addCh = make(chan interface{})

var stopper = make(chan struct{})

var notification interface{}

var pendding = *buffer.NewRingGrowing(2)

func main() {

	// pop

	go func() {

		var nextCh chan<- interface{}

		var notification interface{}

		//var n int

		for {

			fmt.Println("busy wait")

			fmt.Println("entry select", notification)

			select {

			// 初始时，一个未初始化的channel，nil，形成一个阻塞（单channel下是死锁）

			case nextCh <- notification:

				fmt.Println("entry nextCh", notification)

				var ok bool

				// 读不到数据代表已处理完，置空锁

				notification, ok = pendding.ReadOne()

				if !ok {

					fmt.Println("unactive nextch")

					nextCh = nil

				}

			// 事件的分发，监听，初始时也是一个阻塞

			case notificationToAdd, ok := <-addCh:

				fmt.Println(notificationToAdd, notification)

				if !ok {

					return

				}

				// 线程安全

				// 当消息为空时，没有被处理

				// 锁为空，就分发数据

				if notification == nil {

					fmt.Println("frist notification nil")

					notification = notificationToAdd

					nextCh = nextCh1 // 这步骤等于初始化了局部的nextCh，会触发上面的流程

				} else {

					// 在第三次时，会走到这里，数据进入环

					fmt.Println("into ring", notificationToAdd)

					pendding.WriteOne(notificationToAdd)

				}

			}

		}

	}()

	// producer

	go func() {

		i := 0

		for {

			i++

			if i%5 == 0 {

				addCh <- fmt.Sprintf("thread 2 inner -- %d", i)

				time.Sleep(time.Millisecond * 9000)

			} else {

				addCh <- fmt.Sprintf("thread 2 outer -- %d", i)

				time.Sleep(time.Millisecond * 500)

			}

		}

	}()

	// subsriber

	go func() {

		for {

			for next := range nextCh1 {

				time.Sleep(time.Millisecond * 300)

				fmt.Println("consumer", next)

			}

		}

	}()

	<-stopper

}

总结，这里的机制类似于线程安全，进入临界区的一些算法，临界区就是 nextCh，notification 就是保证了至少有一个进程可以进入临界区（要么分发事件，要么生产事件）；nextCh 和 nextCh1 一个是局部管道一个是全局的，管道未初始化代表了死锁（阻塞）；当有消息要处理时，会将局部管道 nextCh 赋值给全局 nextCh1 此时相当于解除了分发的步骤（对管道赋值，触发分发操作）；ringbuffer 实际上是提供了一个对 notification 加锁的操作，在没有处理的消息时，需要保障 notification 为空，同时也关闭了流程 nextCh 的写入。这里主要是考虑对golang中channel的用法

巴特西